[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/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.97.1-0.20240409140257-792fac1b62d4
  go.opentelemetry.io/collector/pdata v1.4.1-0.20240409140257-792fac1b62d4
  go.opentelemetry.io/collector/processor v0.97.1-0.20240409140257-792fac1b62d4
+ 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
@@ -39,9 +40,9 @@ require (
  github.com/prometheus/client_model v0.6.1 // indirect
  github.com/prometheus/common v0.48.0 // indirect
  github.com/prometheus/procfs v0.12.0 // indirect
+ go.opentelemetry.io/collector v0.97.1-0.20240409140257-792fac1b62d4 // indirect
  go.opentelemetry.io/collector/config/configtelemetry v0.97.1-0.20240409140257-792fac1b62d4 // indirect
  go.opentelemetry.io/collector/pdata/testdata v0.0.0-20240408153657-fc289290613a // 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. samples were likely lost in transit", e.From, e.To)
+}

processor/deltatocumulativeprocessor/internal/maybe/ptr.go

@@ -0,0 +1,52 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+// maybe provides utilities for representing data may or may not exist at
+// runtime in a safe way.
+//
+// A typical approach to this are pointers, but they suffer from two issues:
+// - Unsafety: permitting nil pointers must require careful checking on each use,
+// which is easily forgotten
+// - Blindness: nil itself does cannot differentiate between "set to nil" and
+// "not set all", leading to unexepcted edge cases
+//
+// The [Ptr] type of this package provides a safe alternative with a clear
+// distinction between "not set" and "set to nil".
+package maybe // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor/internal/maybe"
+
+// Ptr references some value of type T that is not guaranteed to exist.
+// Callers must use [Ptr.Try] to access the underlying value, checking the
+// ok return value too.
+// This provides a clear distinction between "not set" and "set to nil".
+//
+// Use [Some] and [None] to create Ptrs.
+type Ptr[T any] struct {
+ to *T
+ ok bool
+}
+
+// None returns a Ptr that represents "not-set".
+// This is equal to a zero-value Ptr.
+func None[T any]() Ptr[T] {
+ return Ptr[T]{to: nil, ok: false}
+}
+
+// Some returns a pointer to the passed T.
+//
+// The ptr argument may be nil, in which case this represents "explicitly set to
+// nil".
+func Some[T any](ptr *T) Ptr[T] {
+ return Ptr[T]{to: ptr, ok: true}
+}
+
+// Try attempts to de-reference the Ptr, giving one of three results:
+//
+// - nil, false: not-set
+// - nil, true: explicitly set to nil
+// - non-nil, true: set to some value
+//
+// This provides extra safety over bare pointers, because callers are forced by
+// the compiler to either check or explicitly ignore the ok value.
+func (ptr Ptr[T]) Try() (_ *T, ok bool) {
+ return ptr.to, ptr.ok
+}

processor/deltatocumulativeprocessor/internal/maybe/ptr_test.go

@@ -0,0 +1,64 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package maybe_test
+
+import (
+ "fmt"
+ "testing"
+
+ "github.com/stretchr/testify/require"
+
+ "github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor/internal/maybe"
+)
+
+func TestMaybe(t *testing.T) {
+ t.Run("zero-not-ok", func(t *testing.T) {
+ var ptr maybe.Ptr[int]
+ _, ok := ptr.Try()
+ require.False(t, ok)
+ })
+ t.Run("none-not-ok", func(t *testing.T) {
+ ptr := maybe.None[int]()
+ _, ok := ptr.Try()
+ require.False(t, ok)
+ })
+ t.Run("explicit-nil", func(t *testing.T) {
+ ptr := maybe.Some[int](nil)
+ v, ok := ptr.Try()
+ require.Nil(t, v)
+ require.True(t, ok)
+ })
+ t.Run("value", func(t *testing.T) {
+ num := 42
+ ptr := maybe.Some(&num)
+ v, ok := ptr.Try()
+ require.True(t, ok)
+ require.Equal(t, num, *v)
+ })
+}
+
+func ExamplePtr() {
+ var unset maybe.Ptr[int] // = maybe.None()
+ if v, ok := unset.Try(); ok {
+ fmt.Println("unset:", v)
+ } else {
+ fmt.Println("unset: !ok")
+ }
+
+ var xnil maybe.Ptr[int] = maybe.Some[int](nil)
+ if v, ok := xnil.Try(); ok {
+ fmt.Println("explicit nil:", v)
+ }
+
+ num := 42
+ var set maybe.Ptr[int] = maybe.Some(&num)
+ if v, ok := set.Try(); ok {
+ fmt.Println("set:", *v)
+ }
+
+ // Output:
+ // unset: !ok
+ // explicit nil: <nil>
+ // set: 42
+}

processor/deltatocumulativeprocessor/internal/streams/limit.go

@@ -23,7 +23,9 @@ type LimitMap[T any] struct {
 }
 
 func (m LimitMap[T]) Store(id identity.Stream, v T) error {
- if m.Map.Len() < m.Max {
+ _, ok := m.Map.Load(id)
+ avail := m.Map.Len() < m.Max
+ if ok || avail {
  return m.Map.Store(id, v)
  }
 
@@ -33,7 +35,7 @@ func (m LimitMap[T]) Store(id identity.Stream, v T) error {
  if err := m.Map.Store(id, v); err != nil {
  return err
  }
- return ErrEvicted{ErrLimit: errl, id: gone}
+ return ErrEvicted{ErrLimit: errl, Ident: gone}
  }
  return errl
 }
@@ -51,13 +53,9 @@ func AtLimit(err error) bool {
 
 type ErrEvicted struct {
  ErrLimit
- id Ident
+ Ident Ident
 }
 
 func (e ErrEvicted) Error() string {
- return fmt.Sprintf("%s. evicted stream %s", e.ErrLimit, e.id)
-}
-
-func (e ErrEvicted) Unwrap() error {
- return e.ErrLimit
+ return fmt.Sprintf("%s. evicted stream %s", e.ErrLimit, e.Ident)
 }

processor/deltatocumulativeprocessor/internal/streams/limit_test.go

@@ -22,11 +22,13 @@ func TestLimit(t *testing.T) {
  lim := streams.Limit(items, 10)
 
  ids := make([]identity.Stream, 10)
+ dps := make([]data.Number, 10)
 
  // write until limit must work
  for i := 0; i < 10; i++ {
  id, dp := sum.Stream()
  ids[i] = id
+ dps[i] = dp
  err := lim.Store(id, dp)
  require.NoError(t, err)
  }
@@ -40,6 +42,12 @@ func TestLimit(t *testing.T) {
  require.True(t, streams.AtLimit(err))
  }
 
+ // write to existing must work
+ {
+ err := lim.Store(ids[3], dps[3])
+ require.NoError(t, err)
+ }
+
  // after removing one, must be accepted again
  {
  lim.Delete(ids[0])