[processor/deltatocumulative]: drop bad samples (#34979)

**Description:**
While writing a test that exercises the public api, I noticed we record
bad samples as metrics, but do not actually drop them.
This is another result of the highly layered architecture that makes
such things hard to understand. I'm working on removing that.

**Link to tracking Issue:**

**Testing:** public api tests were added

**Documentation:** not needed

.chloggen/deltatocumulative-apitest.yaml

@@ -0,0 +1,28 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: deltatocumulative
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: drop bad samples
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [34979]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext: |
+  removes bad (rejected) samples from output. previously identified and metric-tracked those as such, but didn't actually drop them.
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

processor/deltatocumulativeprocessor/internal/data/datatest/compare/compare.go

@@ -14,14 +14,14 @@ import (
 var Opts = []cmp.Option{
 	cmpopts.EquateApprox(0, 1e-9),
 	cmp.Exporter(func(ty reflect.Type) bool {
-		return strings.HasPrefix(ty.PkgPath(), "go.opentelemetry.io/collector/pdata")
+		return strings.HasPrefix(ty.PkgPath(), "go.opentelemetry.io/collector/pdata") || strings.HasPrefix(ty.PkgPath(), "github.com/open-telemetry/opentelemetry-collector-contrib")
 	}),
 }
 
-func Equal[T any](a, b T) bool {
-	return cmp.Equal(a, b, Opts...)
+func Equal[T any](a, b T, opts ...cmp.Option) bool {
+	return cmp.Equal(a, b, append(Opts, opts...)...)
 }
 
-func Diff[T any](a, b T) string {
-	return cmp.Diff(a, b, Opts...)
+func Diff[T any](a, b T, opts ...cmp.Option) string {
+	return cmp.Diff(a, b, append(Opts, opts...)...)
 }

processor/deltatocumulativeprocessor/internal/metrics/iter.go

@@ -0,0 +1,38 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package metrics // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor/internal/metrics"
+
+import (
+	"go.opentelemetry.io/collector/pdata/pmetric"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor/internal/putil/pslice"
+)
+
+func All(md pmetric.Metrics) func(func(Metric) bool) {
+	return func(yield func(Metric) bool) {
+		var ok bool
+		pslice.All(md.ResourceMetrics())(func(rm pmetric.ResourceMetrics) bool {
+			pslice.All(rm.ScopeMetrics())(func(sm pmetric.ScopeMetrics) bool {
+				pslice.All(sm.Metrics())(func(m pmetric.Metric) bool {
+					ok = yield(From(rm.Resource(), sm.Scope(), m))
+					return ok
+				})
+				return ok
+			})
+			return ok
+		})
+	}
+}
+
+func Filter(md pmetric.Metrics, keep func(Metric) bool) {
+	md.ResourceMetrics().RemoveIf(func(rm pmetric.ResourceMetrics) bool {
+		rm.ScopeMetrics().RemoveIf(func(sm pmetric.ScopeMetrics) bool {
+			sm.Metrics().RemoveIf(func(m pmetric.Metric) bool {
+				return !keep(From(rm.Resource(), sm.Scope(), m))
+			})
+			return sm.Metrics().Len() == 0
+		})
+		return rm.ScopeMetrics().Len() == 0
+	})
+}

processor/deltatocumulativeprocessor/internal/metrics/metrics.go

@@ -22,6 +22,14 @@ func (m *Metric) Ident() Ident {
 	return identity.OfResourceMetric(m.res, m.scope, m.Metric)
 }
 
+func (m *Metric) Resource() pcommon.Resource {
+	return m.res
+}
+
+func (m *Metric) Scope() pcommon.InstrumentationScope {
+	return m.scope
+}
+
 func From(res pcommon.Resource, scope pcommon.InstrumentationScope, metric pmetric.Metric) Metric {
 	return Metric{res: res, scope: scope, Metric: metric}
 }

processor/deltatocumulativeprocessor/internal/putil/pslice/pslice.go

@@ -19,3 +19,13 @@ func Equal[E comparable, S Slice[E]](a, b S) bool {
 	}
 	return true
 }
+
+func All[E any, S Slice[E]](slice S) func(func(E) bool) {
+	return func(yield func(E) bool) {
+		for i := 0; i < slice.Len(); i++ {
+			if !yield(slice.At(i)) {
+				break
+			}
+		}
+	}
+}

processor/deltatocumulativeprocessor/internal/streams/data.go

@@ -9,21 +9,16 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/exp/metrics/identity"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor/internal/data"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor/internal/metrics"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor/internal/putil/pslice"
 )
 
-// Samples returns an Iterator over each sample of all streams in the metric
-func Samples[D data.Point[D]](m metrics.Data[D]) Seq[D] {
-	mid := m.Ident()
-
-	return func(yield func(Ident, D) bool) bool {
-		for i := 0; i < m.Len(); i++ {
-			dp := m.At(i)
+func Datapoints[P data.Point[P], List metrics.Data[P]](dps List) func(func(identity.Stream, P) bool) {
+	return func(yield func(identity.Stream, P) bool) {
+		mid := dps.Ident()
+		pslice.All(dps)(func(dp P) bool {
 			id := identity.OfStream(mid, dp)
-			if !yield(id, dp) {
-				break
-			}
-		}
-		return false
+			return yield(id, dp)
+		})
 	}
 }
 

processor/deltatocumulativeprocessor/internal/streams/data_test.go

@@ -24,7 +24,7 @@ func BenchmarkSamples(b *testing.B) {
 		dps := generate(b.N)
 		b.ResetTimer()
 
-		streams.Samples(dps)(func(id streams.Ident, dp data.Number) bool {
+		streams.Datapoints(dps)(func(id streams.Ident, dp data.Number) bool {
 			rdp = dp
 			rid = id
 			return true

processor/deltatocumulativeprocessor/internal/streams/errors.go

@@ -19,3 +19,7 @@ type StreamErr struct {
 func (e StreamErr) Error() string {
 	return fmt.Sprintf("%s: %s", e.Ident, e.Err)
 }
+
+func (e StreamErr) Unwrap() error {
+	return e.Err
+}

processor/deltatocumulativeprocessor/internal/telemetry/faults_test.go

@@ -27,9 +27,13 @@ func TestFaults(t *testing.T) {
 	type Case struct {
 		Name string
 		Map  Map
-		Pre  func(Map, identity.Stream, data.Number) error
-		Bad  func(Map, identity.Stream, data.Number) error
-		Err  error
+		// data preparation, etc
+		Pre func(Map, identity.Stream, data.Number) error
+		// cause an error
+		Bad func(Map, identity.Stream, data.Number) error
+		// expected error that was caused
+		Err error
+		// expected return above error was converted into
 		Want error
 	}
 
@@ -49,7 +53,8 @@ func TestFaults(t *testing.T) {
 				dp.SetTimestamp(ts(40))
 				return dps.Store(id, dp)
 			},
-			Err: delta.ErrOlderStart{Start: ts(20), Sample: ts(10)},
+			Err:  delta.ErrOlderStart{Start: ts(20), Sample: ts(10)},
+			Want: streams.Drop,
 		},
 		{
 			Name: "out-of-order",
@@ -61,7 +66,8 @@ func TestFaults(t *testing.T) {
 				dp.SetTimestamp(ts(10))
 				return dps.Store(id, dp)
 			},
-			Err: delta.ErrOutOfOrder{Last: ts(20), Sample: ts(10)},
+			Err:  delta.ErrOutOfOrder{Last: ts(20), Sample: ts(10)},
+			Want: streams.Drop,
 		},
 		{
 			Name: "gap",
@@ -75,7 +81,8 @@ func TestFaults(t *testing.T) {
 				dp.SetTimestamp(ts(40))
 				return dps.Store(id, dp)
 			},
-			Err: delta.ErrGap{From: ts(20), To: ts(30)},
+			Err:  delta.ErrGap{From: ts(20), To: ts(30)},
+			Want: nil,
 		},
 		{
 			Name: "limit",
@@ -109,7 +116,8 @@ func TestFaults(t *testing.T) {
 				dp.SetTimestamp(ts(20))
 				return dps.Store(id, dp)
 			},
-			Err: streams.ErrEvicted{Ident: evid, ErrLimit: streams.ErrLimit(1)},
+			Err:  streams.ErrEvicted{Ident: evid, ErrLimit: streams.ErrLimit(1)},
+			Want: nil,
 		},
 	}
 
@@ -125,17 +133,17 @@ func TestFaults(t *testing.T) {
 			if dps == nil {
 				dps = delta.New[data.Number]()
 			}
-			onf := telemetry.ObserveNonFatal(dps, &tel.Metrics)
+			var realErr error
+			dps = errGrab[data.Number]{Map: dps, err: &realErr}
+			dps = telemetry.ObserveNonFatal(dps, &tel.Metrics)
 
 			if c.Pre != nil {
-				err := c.Pre(onf, id, dp.Clone())
+				err := c.Pre(dps, id, dp.Clone())
 				require.NoError(t, err)
 			}
 
 			err := c.Bad(dps, id, dp.Clone())
-			require.Equal(t, c.Err, err)
-
-			err = c.Bad(onf, id, dp.Clone())
+			require.Equal(t, c.Err, realErr)
 			require.Equal(t, c.Want, err)
 		})
 	}
@@ -154,3 +162,14 @@ func (e HeadEvictor[T]) Evict() (evicted identity.Stream, ok bool) {
 	})
 	return evicted, true
 }
+
+// errGrab stores any error that happens on Store() for later inspection
+type errGrab[T any] struct {
+	streams.Map[T]
+	err *error
+}
+
+func (e errGrab[T]) Store(id identity.Stream, dp T) error {
+	*e.err = e.Map.Store(id, dp)
+	return *e.err
+}

processor/deltatocumulativeprocessor/internal/telemetry/metrics.go

@@ -119,8 +119,10 @@ func (f Faults[T]) Store(id streams.Ident, v T) error {
 		return err
 	case errors.As(err, &olderStart):
 		inc(f.dps.dropped, reason("older-start"))
+		return streams.Drop
 	case errors.As(err, &outOfOrder):
 		inc(f.dps.dropped, reason("out-of-order"))
+		return streams.Drop
 	case errors.As(err, &limit):
 		inc(f.dps.dropped, reason("stream-limit"))
 		// no space to store stream, drop it instead of failing silently

processor/deltatocumulativeprocessor/processor.go

@@ -136,7 +136,8 @@ func (p *Processor) ConsumeMetrics(ctx context.Context, md pmetric.Metrics) erro
 	defer p.mtx.Unlock()
 
 	var errs error
-	metrics.Each(md, func(m metrics.Metric) {
+	metrics.Filter(md, func(m metrics.Metric) bool {
+		var n int
 		switch m.Type() {
 		case pmetric.MetricTypeSum:
 			sum := m.Sum()
@@ -145,25 +146,32 @@ func (p *Processor) ConsumeMetrics(ctx context.Context, md pmetric.Metrics) erro
 				errs = errors.Join(errs, err)
 				sum.SetAggregationTemporality(pmetric.AggregationTemporalityCumulative)
 			}
+			n = sum.DataPoints().Len()
 		case pmetric.MetricTypeHistogram:
 			hist := m.Histogram()
 			if hist.AggregationTemporality() == pmetric.AggregationTemporalityDelta {
 				err := streams.Apply(metrics.Histogram(m), p.hist.aggr.Aggregate)
 				errs = errors.Join(errs, err)
 				hist.SetAggregationTemporality(pmetric.AggregationTemporalityCumulative)
 			}
+			n = hist.DataPoints().Len()
 		case pmetric.MetricTypeExponentialHistogram:
 			expo := m.ExponentialHistogram()
 			if expo.AggregationTemporality() == pmetric.AggregationTemporalityDelta {
 				err := streams.Apply(metrics.ExpHistogram(m), p.expo.aggr.Aggregate)
 				errs = errors.Join(errs, err)
 				expo.SetAggregationTemporality(pmetric.AggregationTemporalityCumulative)
 			}
+			n = expo.DataPoints().Len()
 		}
+		return n > 0
 	})
 	if errs != nil {
 		return errs
 	}
 
+	if md.MetricCount() == 0 {
+		return nil
+	}
 	return p.next.ConsumeMetrics(ctx, md)
 }