[processor/deltatocumulative]: drop samples of streams exceeding limit

.chloggen/deltatocumulative-drop-at-limit.yaml

@@ -0,0 +1,30 @@
+# 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: properly drop samples when at limit
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [33285]
+
+# (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:
+  fixes a segfault in the limiting behavior, where streams exceeding the limit still had their samples processed.
+  due to not being tracked, this led to a nil-pointer deref
+
+
+# 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/metrics/data.go

@@ -4,6 +4,8 @@
 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/data"
 )
 
@@ -28,6 +30,12 @@ func (s Sum) Ident() Ident {
 	return (*Metric)(&s).Ident()
 }
 
+func (s Sum) Filter(expr func(data.Number) bool) {
+	s.Sum().DataPoints().RemoveIf(func(dp pmetric.NumberDataPoint) bool {
+		return !expr(data.Number{NumberDataPoint: dp})
+	})
+}
+
 type Histogram Metric
 
 func (s Histogram) At(i int) data.Histogram {
@@ -43,6 +51,12 @@ func (s Histogram) Ident() Ident {
 	return (*Metric)(&s).Ident()
 }
 
+func (s Histogram) Filter(expr func(data.Histogram) bool) {
+	s.Histogram().DataPoints().RemoveIf(func(dp pmetric.HistogramDataPoint) bool {
+		return !expr(data.Histogram{HistogramDataPoint: dp})
+	})
+}
+
 type ExpHistogram Metric
 
 func (s ExpHistogram) At(i int) data.ExpHistogram {
@@ -57,3 +71,9 @@ func (s ExpHistogram) Len() int {
 func (s ExpHistogram) Ident() Ident {
 	return (*Metric)(&s).Ident()
 }
+
+func (s ExpHistogram) Filter(expr func(data.ExpHistogram) bool) {
+	s.ExponentialHistogram().DataPoints().RemoveIf(func(dp pmetric.ExponentialHistogramDataPoint) bool {
+		return !expr(data.ExpHistogram{DataPoint: dp})
+	})
+}

processor/deltatocumulativeprocessor/internal/streams/data.go

@@ -27,20 +27,34 @@ func Samples[D data.Point[D]](m metrics.Data[D]) Seq[D] {
 	}
 }
 
-// Aggregate each point and replace it by the result
-func Aggregate[D data.Point[D]](m metrics.Data[D], aggr Aggregator[D]) error {
+type filterable[D data.Point[D]] interface {
+	metrics.Data[D]
+	Filter(func(D) bool)
+}
+
+// Apply does dps[i] = fn(dps[i]) for each item in dps.
+// If fn returns [streams.Drop], the datapoint is removed from dps instead.
+// If fn returns another error, the datapoint is also removed and the error returned eventually
+func Apply[P data.Point[P], List filterable[P]](dps List, fn func(Ident, P) (P, error)) error {
 	var errs error
 
-	// for id, dp := range Samples(m)
-	Samples(m)(func(id Ident, dp D) bool {
-		next, err := aggr.Aggregate(id, dp)
+	mid := dps.Ident()
+	dps.Filter(func(dp P) bool {
+		id := identity.OfStream(mid, dp)
+		next, err := fn(id, dp)
 		if err != nil {
-			errs = errors.Join(errs, Error(id, err))
-			return true
+			if !errors.Is(err, Drop) {
+				errs = errors.Join(errs, err)
+			}
+			return false
 		}
+
 		next.CopyTo(dp)
 		return true
 	})
 
 	return errs
 }
+
+// Drop signals the current item (stream or datapoint) is to be dropped
+var Drop = errors.New("stream dropped") //nolint:revive // Drop is a good name for a signal, see fs.SkipAll

processor/deltatocumulativeprocessor/internal/streams/data_test.go

@@ -4,6 +4,7 @@
 package streams_test
 
 import (
+	"math/rand"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -23,7 +24,7 @@ func BenchmarkSamples(b *testing.B) {
 		dps := generate(b.N)
 		b.ResetTimer()
 
-		streams.Samples[data.Number](dps)(func(id streams.Ident, dp data.Number) bool {
+		streams.Samples(dps)(func(id streams.Ident, dp data.Number) bool {
 			rdp = dp
 			rid = id
 			return true
@@ -55,22 +56,6 @@ func BenchmarkSamples(b *testing.B) {
 	})
 }
 
-func TestSample(t *testing.T) {
-	const total = 1000
-	dps := generate(total)
-
-	// check that all samples are visited
-	seen := 0
-	streams.Samples[data.Number](dps)(func(id streams.Ident, dp data.Number) bool {
-		require.Equal(t, dps.id, id)
-		require.Equal(t, dps.dps[seen], dp)
-		seen++
-		return true
-	})
-
-	require.Equal(t, total, seen)
-}
-
 func TestAggregate(t *testing.T) {
 	const total = 1000
 	dps := generate(total)
@@ -82,7 +67,7 @@ func TestAggregate(t *testing.T) {
 		return dp, nil
 	})
 
-	err := streams.Aggregate(dps, inv)
+	err := streams.Apply(dps, inv.Aggregate)
 	require.NoError(t, err)
 
 	// check that all samples are inverted
@@ -91,15 +76,34 @@ func TestAggregate(t *testing.T) {
 	}
 }
 
-func generate(n int) Data {
+func TestDrop(t *testing.T) {
+	const total = 1000
+	dps := generate(total)
+
+	var want []data.Number
+	maybe := aggr(func(_ streams.Ident, dp data.Number) (data.Number, error) {
+		if rand.Intn(2) == 1 {
+			want = append(want, dp)
+			return dp, nil
+		}
+		return dp, streams.Drop
+	})
+
+	err := streams.Apply(dps, maybe.Aggregate)
+	require.NoError(t, err)
+
+	require.Equal(t, want, dps.dps)
+}
+
+func generate(n int) *Data {
 	id, ndp := random.Sum().Stream()
 	dps := Data{id: id, dps: make([]data.Number, n)}
 	for i := range dps.dps {
 		dp := ndp.Clone()
 		dp.SetIntValue(int64(i))
 		dps.dps[i] = dp
 	}
-	return dps
+	return &dps
 }
 
 type Data struct {
@@ -119,6 +123,16 @@ func (l Data) Ident() metrics.Ident {
 	return l.id.Metric()
 }
 
+func (l *Data) Filter(expr func(data.Number) bool) {
+	var next []data.Number
+	for _, dp := range l.dps {
+		if expr(dp) {
+			next = append(next, dp)
+		}
+	}
+	l.dps = next
+}
+
 type aggr func(streams.Ident, data.Number) (data.Number, error)
 
 func (a aggr) Aggregate(id streams.Ident, dp data.Number) (data.Number, error) {

processor/deltatocumulativeprocessor/internal/telemetry/faults_test.go

@@ -29,6 +29,7 @@ func TestFaults(t *testing.T) {
 		Pre  func(Map, identity.Stream, data.Number) error
 		Bad  func(Map, identity.Stream, data.Number) error
 		Err  error
+		Want error
 	}
 
 	sum := random.Sum()
@@ -87,7 +88,8 @@ func TestFaults(t *testing.T) {
 				dp.SetTimestamp(ts(20))
 				return dps.Store(id, dp)
 			},
-			Err: streams.ErrLimit(1),
+			Err:  streams.ErrLimit(1),
+			Want: streams.Drop, // we can't ignore being at limit, we need to drop the entire stream for this request
 		},
 		{
 			Name: "evict",
@@ -130,7 +132,7 @@ func TestFaults(t *testing.T) {
 			require.Equal(t, c.Err, err)
 
 			err = c.Bad(onf, id, dp.Clone())
-			require.NoError(t, err)
+			require.Equal(t, c.Want, err)
 		})
 	}
 }

processor/deltatocumulativeprocessor/internal/telemetry/metrics.go

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

processor/deltatocumulativeprocessor/processor.go

@@ -138,7 +138,7 @@ func (p *Processor) ConsumeMetrics(ctx context.Context, md pmetric.Metrics) erro
 		case pmetric.MetricTypeSum:
 			sum := m.Sum()
 			if sum.AggregationTemporality() == pmetric.AggregationTemporalityDelta {
-				err := streams.Aggregate(metrics.Sum(m), p.sums.aggr)
+				err := streams.Apply(metrics.Sum(m), p.sums.aggr.Aggregate)
 				errs = errors.Join(errs, err)
 				sum.SetAggregationTemporality(pmetric.AggregationTemporalityCumulative)
 			}
@@ -147,7 +147,7 @@ func (p *Processor) ConsumeMetrics(ctx context.Context, md pmetric.Metrics) erro
 		case pmetric.MetricTypeExponentialHistogram:
 			expo := m.ExponentialHistogram()
 			if expo.AggregationTemporality() == pmetric.AggregationTemporalityDelta {
-				err := streams.Aggregate(metrics.ExpHistogram(m), p.expo.aggr)
+				err := streams.Apply(metrics.ExpHistogram(m), p.expo.aggr.Aggregate)
 				errs = errors.Join(errs, err)
 				expo.SetAggregationTemporality(pmetric.AggregationTemporalityCumulative)
 			}