[processor/deltatocumulative]: timer-based expiry

.chloggen/deltatocumulative-timer.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: "deltatocumulativeprocessor"
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: timer-based expiry
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [31615]
+
+# (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: |
+  converts expiry to 1m timer, eliminating a race condition and failing test
+
+# 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/processor.go

@@ -7,13 +7,15 @@ import (
 	"context"
 	"errors"
 	"sync"
+	"time"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.opentelemetry.io/collector/processor"
 	"go.uber.org/zap"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/exp/metrics/staleness"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor/internal/data"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor/internal/delta"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor/internal/metrics"
@@ -29,8 +31,8 @@ type Processor struct {
 	ctx    context.Context
 	cancel context.CancelFunc
 
-	aggr streams.Aggregator[data.Number]
-	exp  *streams.Expiry[data.Number]
+	aggr  streams.Aggregator[data.Number]
+	stale *staleness.Staleness[data.Number]
 
 	mtx sync.Mutex
 }
@@ -49,32 +51,33 @@ func newProcessor(cfg *Config, log *zap.Logger, next consumer.Metrics) *Processo
 	dps = delta.New[data.Number]()
 
 	if cfg.MaxStale > 0 {
-		exp := streams.ExpireAfter(dps, cfg.MaxStale)
-		proc.exp = &exp
-		dps = &exp
+		stale := staleness.NewStaleness(cfg.MaxStale, dps)
+		proc.stale = stale
+		dps = stale
 	}
 
 	proc.aggr = streams.IntoAggregator(dps)
 	return &proc
 }
 
 func (p *Processor) Start(_ context.Context, _ component.Host) error {
-	if p.exp != nil {
-		go func() {
-			for {
+	if p.stale == nil {
+		return nil
+	}
+
+	go func() {
+		tick := time.NewTicker(time.Minute)
+		for {
+			select {
+			case <-p.ctx.Done():
+				return
+			case <-tick.C:
 				p.mtx.Lock()
-				next := p.exp.ExpireOldEntries()
+				p.stale.ExpireOldEntries()
 				p.mtx.Unlock()
-
-				select {
-				case <-next:
-				case <-p.ctx.Done():
-					return
-				}
 			}
-		}()
-	}
-
+		}
+	}()
 	return nil
 }