[chore]: [deltatocumulative]: remove nested implementation

processor/deltatocumulativeprocessor/benchmark_test.go

@@ -0,0 +1,182 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package deltatocumulativeprocessor
+
+import (
+	"context"
+	"math/rand/v2"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor/internal/data/expo"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor/internal/data/expo/expotest"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor/internal/data/histo"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor/internal/testing/sdktest"
+)
+
+var out *consumertest.MetricsSink
+
+func BenchmarkProcessor(gb *testing.B) {
+	const (
+		metrics = 5
+		streams = 10
+	)
+
+	type Case struct {
+		name string
+		fill func(m pmetric.Metric)
+		next func(m pmetric.Metric)
+	}
+
+	run := func(b *testing.B, proc consumer.Metrics, cs Case) {
+		md := pmetric.NewMetrics()
+		ms := md.ResourceMetrics().AppendEmpty().ScopeMetrics().AppendEmpty().Metrics()
+		for i := range metrics {
+			m := ms.AppendEmpty()
+			m.SetName(strconv.Itoa(i))
+			cs.fill(m)
+		}
+
+		b.ReportAllocs()
+		b.ResetTimer()
+		b.StopTimer()
+
+		ctx := context.Background()
+		for range b.N {
+			for i := range ms.Len() {
+				cs.next(ms.At(i))
+			}
+			req := pmetric.NewMetrics()
+			md.CopyTo(req)
+
+			b.StartTimer()
+			err := proc.ConsumeMetrics(ctx, req)
+			b.StopTimer()
+			require.NoError(b, err)
+		}
+	}
+
+	now := time.Now()
+	start := pcommon.NewTimestampFromTime(now)
+	ts := pcommon.NewTimestampFromTime(now.Add(time.Minute))
+
+	cases := []Case{{
+		name: "sums",
+		fill: func(m pmetric.Metric) {
+			sum := m.SetEmptySum()
+			sum.SetAggregationTemporality(pmetric.AggregationTemporalityDelta)
+			for i := range streams {
+				dp := sum.DataPoints().AppendEmpty()
+				dp.SetIntValue(int64(rand.IntN(10)))
+				dp.Attributes().PutStr("idx", strconv.Itoa(i))
+				dp.SetStartTimestamp(start)
+				dp.SetTimestamp(ts)
+			}
+		},
+		next: next(pmetric.Metric.Sum),
+	}, {
+		name: "histogram",
+		fill: func(m pmetric.Metric) {
+			hist := m.SetEmptyHistogram()
+			hist.SetAggregationTemporality(pmetric.AggregationTemporalityDelta)
+			for i := range streams {
+				dp := hist.DataPoints().AppendEmpty()
+				histo.DefaultBounds.Observe(
+					float64(rand.IntN(1000)),
+					float64(rand.IntN(1000)),
+					float64(rand.IntN(1000)),
+					float64(rand.IntN(1000)),
+				).CopyTo(dp)
+
+				dp.SetStartTimestamp(start)
+				dp.SetTimestamp(ts)
+				dp.Attributes().PutStr("idx", strconv.Itoa(i))
+			}
+		},
+		next: next(pmetric.Metric.Histogram),
+	}, {
+		name: "exponential",
+		fill: func(m pmetric.Metric) {
+			ex := m.SetEmptyExponentialHistogram()
+			ex.SetAggregationTemporality(pmetric.AggregationTemporalityDelta)
+			for i := range streams {
+				dp := ex.DataPoints().AppendEmpty()
+				o := expotest.Observe(expo.Scale(2),
+					float64(rand.IntN(31)+1),
+					float64(rand.IntN(31)+1),
+					float64(rand.IntN(31)+1),
+					float64(rand.IntN(31)+1),
+				)
+				o.CopyTo(dp.Positive())
+				o.CopyTo(dp.Negative())
+
+				dp.SetStartTimestamp(start)
+				dp.SetTimestamp(ts)
+				dp.Attributes().PutStr("idx", strconv.Itoa(i))
+			}
+		},
+		next: next(pmetric.Metric.ExponentialHistogram),
+	}}
+
+	tel := func(n int) sdktest.Spec {
+		total := int64(n * metrics * streams)
+		tracked := int64(metrics * streams)
+		return sdktest.Expect(map[string]sdktest.Metric{
+			"otelcol_deltatocumulative.datapoints.linear": {
+				Type:      sdktest.TypeSum,
+				Numbers:   []sdktest.Number{{Int: &total}},
+				Monotonic: true,
+			},
+			"otelcol_deltatocumulative.streams.tracked.linear": {
+				Type:    sdktest.TypeSum,
+				Numbers: []sdktest.Number{{Int: &tracked}},
+			},
+		})
+	}
+
+	for _, cs := range cases {
+		gb.Run(cs.name, func(b *testing.B) {
+			st := setup(b, nil)
+			out = st.sink
+			run(b, st.proc, cs)
+
+			// verify all dps are processed without error
+			b.StopTimer()
+			if err := sdktest.Test(tel(b.N), st.tel.reader); err != nil {
+				b.Fatal(err)
+			}
+		})
+	}
+}
+
+func next[
+	T interface{ DataPoints() Ps },
+	Ps interface {
+		At(int) P
+		Len() int
+	},
+	P interface {
+		Timestamp() pcommon.Timestamp
+		SetStartTimestamp(pcommon.Timestamp)
+		SetTimestamp(pcommon.Timestamp)
+	},
+](sel func(pmetric.Metric) T) func(m pmetric.Metric) {
+	return func(m pmetric.Metric) {
+		dps := sel(m).DataPoints()
+		for i := range dps.Len() {
+			dp := dps.At(i)
+			dp.SetStartTimestamp(dp.Timestamp())
+			dp.SetTimestamp(pcommon.NewTimestampFromTime(
+				dp.Timestamp().AsTime().Add(time.Minute),
+			))
+		}
+	}
+}

processor/deltatocumulativeprocessor/config.go

@@ -11,7 +11,7 @@ import (
 
 	"go.opentelemetry.io/collector/component"
 
-	telemetry "github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor/internal/lineartelemetry"
+	telemetry "github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor/internal/telemetry"
 )
 
 var _ component.ConfigValidator = (*Config)(nil)

processor/deltatocumulativeprocessor/factory.go

@@ -11,8 +11,8 @@ import (
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/processor"
 
-	ltel "github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor/internal/lineartelemetry"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor/internal/metadata"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor/internal/telemetry"
 )
 
 func NewFactory() processor.Factory {
@@ -29,13 +29,10 @@ func createMetricsProcessor(_ context.Context, set processor.Settings, cfg compo
 		return nil, fmt.Errorf("configuration parsing error")
 	}
 
-	ltel, err := ltel.New(set.TelemetrySettings)
+	tel, err := telemetry.New(set.TelemetrySettings)
 	if err != nil {
 		return nil, err
 	}
 
-	proc := newProcessor(pcfg, set.Logger, &ltel.TelemetryBuilder, next)
-	linear := newLinear(pcfg, ltel, proc)
-
-	return Chain{linear, proc}, nil
+	return newProcessor(pcfg, tel, next), nil
 }

processor/deltatocumulativeprocessor/internal/data/data.go

@@ -4,116 +4,23 @@
 package data // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor/internal/data"
 
 import (
-	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor/internal/data/expo"
 )
 
-var (
-	_ Point[Number]       = Number{}
-	_ Point[Histogram]    = Histogram{}
-	_ Point[ExpHistogram] = ExpHistogram{}
-	_ Point[Summary]      = Summary{}
-)
-
-type Point[Self any] interface {
-	StartTimestamp() pcommon.Timestamp
-	Timestamp() pcommon.Timestamp
-	Attributes() pcommon.Map
-
-	Clone() Self
-	CopyTo(Self)
-
-	Add(Self) Self
-}
-
-type Typed[Self any] interface {
-	Point[Self]
-	Number | Histogram | ExpHistogram | Summary
-}
-
 type Number struct {
 	pmetric.NumberDataPoint
 }
 
-func Zero[P Typed[P]]() P {
-	var point P
-	switch ty := any(&point).(type) {
-	case *Number:
-		ty.NumberDataPoint = pmetric.NewNumberDataPoint()
-	case *Histogram:
-		ty.HistogramDataPoint = pmetric.NewHistogramDataPoint()
-	case *ExpHistogram:
-		ty.DataPoint = pmetric.NewExponentialHistogramDataPoint()
-	}
-	return point
-}
-
-func (dp Number) Clone() Number {
-	clone := Number{NumberDataPoint: pmetric.NewNumberDataPoint()}
-	if dp.NumberDataPoint != (pmetric.NumberDataPoint{}) {
-		dp.CopyTo(clone)
-	}
-	return clone
-}
-
-func (dp Number) CopyTo(dst Number) {
-	dp.NumberDataPoint.CopyTo(dst.NumberDataPoint)
-}
-
 type Histogram struct {
 	pmetric.HistogramDataPoint
 }
 
-func (dp Histogram) Clone() Histogram {
-	clone := Histogram{HistogramDataPoint: pmetric.NewHistogramDataPoint()}
-	if dp.HistogramDataPoint != (pmetric.HistogramDataPoint{}) {
-		dp.CopyTo(clone)
-	}
-	return clone
-}
-
-func (dp Histogram) CopyTo(dst Histogram) {
-	dp.HistogramDataPoint.CopyTo(dst.HistogramDataPoint)
-}
-
 type ExpHistogram struct {
 	expo.DataPoint
 }
 
-func (dp ExpHistogram) Clone() ExpHistogram {
-	clone := ExpHistogram{DataPoint: pmetric.NewExponentialHistogramDataPoint()}
-	if dp.DataPoint != (expo.DataPoint{}) {
-		dp.CopyTo(clone)
-	}
-	return clone
-}
-
-func (dp ExpHistogram) CopyTo(dst ExpHistogram) {
-	dp.DataPoint.CopyTo(dst.DataPoint)
-}
-
-type mustPoint[D Point[D]] struct{ _ D }
-
-var (
-	_ = mustPoint[Number]{}
-	_ = mustPoint[Histogram]{}
-	_ = mustPoint[ExpHistogram]{}
-)
-
 type Summary struct {
 	pmetric.SummaryDataPoint
 }
-
-func (dp Summary) Clone() Summary {
-	clone := Summary{SummaryDataPoint: pmetric.NewSummaryDataPoint()}
-	if dp.SummaryDataPoint != (pmetric.SummaryDataPoint{}) {
-		dp.CopyTo(clone)
-	}
-	return clone
-}
-
-func (dp Summary) CopyTo(dst Summary) {
-	dp.SummaryDataPoint.CopyTo(dst.SummaryDataPoint)
-}