[processor/deltatocumulative]: observe accumulation metrics

.chloggen/deltatocumulative-metrics.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: "enhancement"
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: deltatocumulativeprocessor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: self-instrumentation to observe key metrics of the stream accumulation
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [30705]
+
+# (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:
+
+# 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/README.md

@@ -29,4 +29,21 @@ processors:
         # upper limit of streams to track. new streams exceeding this limit
         # will be dropped
         [ max_streams: <int> | default = 0 (off) ]
+
 ```
+
+There is no further configuration required. All delta samples are converted to cumulative.
+
+## Troubleshooting
+
+The following metrics are recorded when [telemetry is
+enabled](https://opentelemetry.io/docs/collector/configuration/#telemetry):
+
+| Name                                     | Description                                                                           | Unit          |
+|------------------------------------------|---------------------------------------------------------------------------------------|---------------|
+| `deltatocumulative.streams.tracked`      | Number of streams currently tracked by the aggregation state                          | `{stream}`    |
+| `deltatocumulative.streams.limit`        | Upper limit of tracked streams                                                        | `{stream}`    |
+| `deltatocumulative.streams.evicted`      | Number of streams removed from tracking to ingest newer streams                       | `{stream}`    |
+| `deltatocumulative.datapoints.processed` | Total number of datapoints processed, whether successful or not                       | `{datapoint}` |
+| `deltatocumulative.datapoints.dropped`   | Faulty datapoints that were dropped due to the reason given in the `reason` attribute | `{datapoint}` |
+| `deltatocumulative.gaps.length`          | Total length of all gaps in the streams, which occur e.g. due to lost in transit      | `second`      |

processor/deltatocumulativeprocessor/config.go

@@ -13,8 +13,8 @@ import (
 var _ component.ConfigValidator = (*Config)(nil)
 
 type Config struct {
-	MaxStale   time.Duration `json:"max_stale"`
-	MaxStreams int           `json:"max_streams"`
+	MaxStale   time.Duration `mapstructure:"max_stale"`
+	MaxStreams int           `mapstructure:"max_streams"`
 }
 
 func (c *Config) Validate() error {

processor/deltatocumulativeprocessor/factory.go

@@ -28,5 +28,6 @@ func createMetricsProcessor(_ context.Context, set processor.CreateSettings, cfg
 		return nil, fmt.Errorf("configuration parsing error")
 	}
 
-	return newProcessor(pcfg, set.Logger, next), nil
+	meter := metadata.Meter(set.TelemetrySettings)
+	return newProcessor(pcfg, set.Logger, meter, next), nil
 }

processor/deltatocumulativeprocessor/go.mod

@@ -10,6 +10,7 @@ require (
 	go.opentelemetry.io/collector/consumer v0.96.1-0.20240306115632-b2693620eff6
 	go.opentelemetry.io/collector/pdata v1.3.1-0.20240306115632-b2693620eff6
 	go.opentelemetry.io/collector/processor v0.96.1-0.20240306115632-b2693620eff6
+	go.opentelemetry.io/otel v1.24.0
 	go.opentelemetry.io/otel/metric v1.24.0
 	go.opentelemetry.io/otel/trace v1.24.0
 	go.uber.org/zap v1.27.0
@@ -40,7 +41,6 @@ require (
 	github.com/prometheus/procfs v0.12.0 // indirect
 	go.opentelemetry.io/collector v0.96.1-0.20240306115632-b2693620eff6 // indirect
 	go.opentelemetry.io/collector/config/configtelemetry v0.96.1-0.20240306115632-b2693620eff6 // indirect
-	go.opentelemetry.io/otel v1.24.0 // indirect
 	go.opentelemetry.io/otel/exporters/prometheus v0.46.0 // indirect
 	go.opentelemetry.io/otel/sdk v1.24.0 // indirect
 	go.opentelemetry.io/otel/sdk/metric v1.24.0 // indirect

processor/deltatocumulativeprocessor/internal/delta/delta.go

@@ -44,8 +44,17 @@ func (a Accumulator[D]) Store(id streams.Ident, dp D) error {
 		return ErrOutOfOrder{Last: aggr.Timestamp(), Sample: dp.Timestamp()}
 	}
 
+	// detect gaps
+	var gap error
+	if dp.StartTimestamp() > aggr.Timestamp() {
+		gap = ErrGap{From: aggr.Timestamp(), To: dp.StartTimestamp()}
+	}
+
 	res := aggr.Add(dp)
-	return a.Map.Store(id, res)
+	if err := a.Map.Store(id, res); err != nil {
+		return err
+	}
+	return gap
 }
 
 type ErrOlderStart struct {
@@ -65,3 +74,11 @@ type ErrOutOfOrder struct {
 func (e ErrOutOfOrder) Error() string {
 	return fmt.Sprintf("out of order: dropped sample from time=%s, because series is already at time=%s", e.Sample, e.Last)
 }
+
+type ErrGap struct {
+	From, To pcommon.Timestamp
+}
+
+func (e ErrGap) Error() string {
+	return fmt.Sprintf("gap in stream from %s to %s", e.From, e.To)
+}

processor/deltatocumulativeprocessor/internal/maybe/option.go

@@ -0,0 +1,22 @@
+package maybe
+
+// Ptr is a pointer that points to something that is not guaranteed to exist.
+// Any use must "Try()" accessing the underlying value, enforcing checking the
+// ok return value too.
+// This provides a clear distinction between "not set" and "set to nil"
+type Ptr[T any] struct {
+	to *T
+	ok bool
+}
+
+func None[T any]() Ptr[T] {
+	return Ptr[T]{to: nil, ok: false}
+}
+
+func Some[T any](ptr *T) Ptr[T] {
+	return Ptr[T]{to: ptr, ok: true}
+}
+
+func (ptr Ptr[T]) Try() (_ *T, ok bool) {
+	return ptr.to, ptr.ok
+}

processor/deltatocumulativeprocessor/internal/streams/limit.go

@@ -57,7 +57,3 @@ type ErrEvicted struct {
 func (e ErrEvicted) Error() string {
 	return fmt.Sprintf("%s. evicted stream %s", e.ErrLimit, e.id)
 }
-
-func (e ErrEvicted) Unwrap() error {
-	return e.ErrLimit
-}

processor/deltatocumulativeprocessor/internal/telemetry/metrics.go

@@ -0,0 +1,177 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package telemetry // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor/internal/telemetry"
+
+import (
+	"context"
+	"errors"
+
+	"go.opentelemetry.io/collector/processor/processorhelper"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor/internal/delta"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor/internal/metadata"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor/internal/streams"
+)
+
+type Streams struct {
+	tracked metric.Int64UpDownCounter
+	limit   metric.Int64ObservableGauge
+	evicted metric.Int64Counter
+}
+
+type Datapoints struct {
+	total   metric.Int64Counter
+	dropped metric.Int64Counter
+}
+
+type Metrics struct {
+	streams Streams
+	dps     Datapoints
+
+	gaps metric.Int64Counter
+}
+
+func metrics(meter metric.Meter) Metrics {
+	var (
+		count  = use(meter.Int64Counter)
+		updown = use(meter.Int64UpDownCounter)
+		gauge  = use(meter.Int64ObservableGauge)
+	)
+
+	return Metrics{
+		streams: Streams{
+			tracked: updown("streams.tracked",
+				metric.WithDescription("number of streams tracked"),
+				metric.WithUnit("{stream}"),
+			),
+			limit: gauge("streams.limit",
+				metric.WithDescription("upper limit of tracked streams"),
+				metric.WithUnit("{stream}"),
+			),
+			evicted: count("streams.evicted",
+				metric.WithDescription("number of streams evicted"),
+				metric.WithUnit("{stream}"),
+			),
+		},
+		dps: Datapoints{
+			total: count("datapoints.processed",
+				metric.WithDescription("number of datapoints processed"),
+				metric.WithUnit("{datapoint}"),
+			),
+			dropped: count("datapoints.dropped",
+				metric.WithDescription("number of dropped datapoints due to given 'reason'"),
+				metric.WithUnit("{datapoint}"),
+			),
+		},
+		gaps: count("gaps.length",
+			metric.WithDescription("total duration where data was expected but not received"),
+			metric.WithUnit("s"),
+		),
+	}
+}
+
+func (m Metrics) WithLimit(meter metric.Meter, max int64) {
+	then := metric.Callback(func(_ context.Context, o metric.Observer) error {
+		o.ObserveInt64(m.streams.limit, max)
+		return nil
+	})
+	_, err := meter.RegisterCallback(then, m.streams.limit)
+	if err != nil {
+		panic(err)
+	}
+}
+
+func Observe[T any](items streams.Map[T], meter metric.Meter) *Map[T] {
+	return &Map[T]{
+		Map:     items,
+		Metrics: metrics(meter),
+	}
+}
+
+var _ streams.Map[any] = (*Map[any])(nil)
+
+type Map[T any] struct {
+	streams.Map[T]
+	Metrics
+}
+
+func (m *Map[T]) Store(id streams.Ident, v T) error {
+	inc(m.dps.total)
+	_, old := m.Load(id)
+
+	var (
+		olderStart delta.ErrOlderStart
+		outOfOrder delta.ErrOutOfOrder
+		gap        delta.ErrGap
+		limit      streams.ErrLimit
+		evict      streams.ErrEvicted
+	)
+
+	err := m.Map.Store(id, v)
+	switch {
+	case err == nil:
+		// all good
+	case errors.As(err, &olderStart):
+		// non fatal. record but ignore
+		inc(m.dps.dropped, reason("older-start"))
+		err = nil
+	case errors.As(err, &outOfOrder):
+		// non fatal. record but ignore
+		inc(m.dps.dropped, reason("out-of-order"))
+		err = nil
+	case errors.As(err, &evict):
+		inc(m.streams.evicted)
+		err = nil
+	case errors.As(err, &limit):
+		inc(m.dps.dropped, reason("stream-limit"))
+		err = nil
+	case errors.As(err, &gap):
+		// a gap occurred. record its length, but ignore
+		from := gap.From.AsTime()
+		to := gap.To.AsTime()
+		lost := to.Sub(from).Seconds()
+		m.gaps.Add(context.TODO(), int64(lost))
+		err = nil
+	}
+
+	// not dropped and not seen before => new stream
+	if err == nil && !old {
+		inc(m.streams.tracked)
+	}
+	return err
+}
+
+func (m *Map[T]) Delete(id streams.Ident) {
+	dec(m.streams.tracked)
+	m.Map.Delete(id)
+}
+
+type addable[Opts any] interface {
+	Add(context.Context, int64, ...Opts)
+}
+
+func inc[A addable[O], O any](a A, opts ...O) {
+	a.Add(context.Background(), 1, opts...)
+}
+
+func dec[A addable[O], O any](a A, opts ...O) {
+	a.Add(context.Background(), -1, opts...)
+}
+
+func reason(reason string) metric.AddOption {
+	return metric.WithAttributes(attribute.String("reason", reason))
+}
+
+func use[F func(string, ...O) (M, error), M any, O any](f F) func(string, ...O) M {
+	return func(name string, opts ...O) M {
+		name = processorhelper.BuildCustomMetricName(metadata.Type.String(), name)
+		m, err := f(name, opts...)
+		if err != nil {
+			panic(err)
+		}
+		return m
+	}
+}