[processor/deltatocumulative]: drop bad samples

.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)
 }