metricstarttimeprocessor: Copy true reset strategy from prometheus receiver (#37855)

#### Description

Copy the "metric adjuster" from the prometheus receiver to a new
"true_reset_point" strategy in the metricstarttimeprocessor

#### Link to tracking issue
Part of
#37186

#### Testing

Copied unit tests

#### Documentation

Updated the README.

#### Notes for reviewers

* I would recommend reviewing commit-by-commit. The first commit is
copied verbatim, with minimal changes other than splitting into multiple
files.
* I changed the function signature of `AdjustMetrics` to match the
processMetrics function so that it doesn't need an additional wrapper
layer.
* I removed the MetricsAdjuster interface, since it isn't needed (we
implement the processor function for metrics now).
* I had to remove the validation of job + instance being present because
it didn't pass generated tests. Regardless, we should not rely on those
in a generic processor.

Author: dashpole

.chloggen/metricstarttime-true-reset.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: metricstarttimeprocessor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: add true_reset_point strategy for setting the start timestamp of cumulative points.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [37186]
+
+# (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: The implementation is copied from the Prometheus receiver.
+
+# 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, api]

processor/metricstarttimeprocessor/README.md

@@ -8,3 +8,45 @@
 
 [development]: https://github.com/open-telemetry/opentelemetry-collector/blob/main/docs/component-stability.md#development
 <!-- end autogenerated section -->
+
+## Description
+
+The metric start time processor (`metricstarttime`) is used to set the start
+time for metric points with a cumulative aggregation temporality. It is
+commonly used with the `prometheus` receiver, which usually produces metric
+points without a start time.
+
+## Configuration
+
+Configuration allows configuring the strategy used to set the start time.
+
+```yaml
+processors:
+    # processor name: metricstarttime
+    metricstarttime:
+
+        # specify the strategy to use for setting the start time
+        strategy: true_reset_point
+```
+
+### Strategy: True Reset Point
+
+The `true_reset_point` strategy handles missing start times for cumulative
+points by producing a stream of points that starts with a
+[True Reset Point](https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/data-model.md#cumulative-streams-inserting-true-reset-points).
+The true reset point has its start time set to its end timestamp. It is meant
+to indicate the absolute value of the cumulative point when the collector first
+observed it. Subsequent points re-use the start timestamp of the initial True
+Reset point.
+
+Pros:
+
+* The absolute value of the cumulative metric is preserved.
+* It is possible to calculate the correct rate between any two points since the timestamps and values are not modified.
+
+Cons:
+
+* This strategy is **stateful** because the initial True Reset point is necessary to properly calculate rates on subsequent points.
+* The True Reset point doesn't make sense semantically. It has a zero duration, but non-zero values.
+* Many backends reject points with equal start and end timestamps.
+    * If the True Reset point is rejected, the next point will appear to have a very large rate.

processor/metricstarttimeprocessor/config.go

@@ -4,15 +4,36 @@
 package metricstarttimeprocessor // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/metricstarttimeprocessor"
 
 import (
+	"fmt"
+	"time"
+
 	"go.opentelemetry.io/collector/component"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/metricstarttimeprocessor/internal/truereset"
 )
 
 // Config holds configuration of the metric start time processor.
-type Config struct{}
+type Config struct {
+	Strategy   string        `mapstructure:"strategy"`
+	GCInterval time.Duration `mapstructure:"gc_interval"`
+}
 
 var _ component.Config = (*Config)(nil)
 
+func createDefaultConfig() component.Config {
+	return &Config{
+		Strategy:   truereset.Type,
+		GCInterval: 10 * time.Minute,
+	}
+}
+
 // Validate checks the configuration is valid
 func (cfg *Config) Validate() error {
+	if cfg.Strategy != truereset.Type {
+		return fmt.Errorf("%v is not a valid strategy", cfg.Strategy)
+	}
+	if cfg.GCInterval <= 0 {
+		return fmt.Errorf("gc_interval must be positive")
+	}
 	return nil
 }

processor/metricstarttimeprocessor/factory.go

@@ -9,8 +9,10 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/processor"
+	"go.opentelemetry.io/collector/processor/processorhelper"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/metricstarttimeprocessor/internal/metadata"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/metricstarttimeprocessor/internal/truereset"
 )
 
 // NewFactory creates a new metric start time processor factory.
@@ -21,10 +23,6 @@ func NewFactory() processor.Factory {
 		processor.WithMetrics(createMetricsProcessor, metadata.MetricsStability))
 }
 
-func createDefaultConfig() component.Config {
-	return &Config{}
-}
-
 // createMetricsProcessor creates a metrics processor based on provided config.
 func createMetricsProcessor(
 	ctx context.Context,
@@ -33,5 +31,14 @@ func createMetricsProcessor(
 	nextConsumer consumer.Metrics,
 ) (processor.Metrics, error) {
 	rCfg := cfg.(*Config)
-	return newMetricStartTimeProcessor(ctx, set, nextConsumer, rCfg)
+
+	adjuster := truereset.NewAdjuster(set.TelemetrySettings, rCfg.GCInterval)
+
+	return processorhelper.NewMetrics(
+		ctx,
+		set,
+		cfg,
+		nextConsumer,
+		adjuster.AdjustMetrics,
+		processorhelper.WithCapabilities(consumer.Capabilities{MutatesData: true}))
 }

processor/metricstarttimeprocessor/go.mod

@@ -3,6 +3,7 @@ module github.com/open-telemetry/opentelemetry-collector-contrib/processor/metri
 go 1.23.0
 
 require (
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil v0.120.1
 	github.com/stretchr/testify v1.10.0
 	go.opentelemetry.io/collector/component v0.120.1-0.20250226024140-8099e51f9a77
 	go.opentelemetry.io/collector/component/componenttest v0.120.1-0.20250226024140-8099e51f9a77
@@ -12,10 +13,13 @@ require (
 	go.opentelemetry.io/collector/pdata v1.26.1-0.20250226024140-8099e51f9a77
 	go.opentelemetry.io/collector/processor v0.120.1-0.20250226024140-8099e51f9a77
 	go.opentelemetry.io/collector/processor/processortest v0.120.1-0.20250226024140-8099e51f9a77
+	go.opentelemetry.io/collector/semconv v0.120.1-0.20250226024140-8099e51f9a77
 	go.uber.org/goleak v1.3.0
+	go.uber.org/zap v1.27.0
 )
 
 require (
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
@@ -44,7 +48,6 @@ require (
 	go.opentelemetry.io/otel/sdk/metric v1.34.0 // indirect
 	go.opentelemetry.io/otel/trace v1.34.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
-	go.uber.org/zap v1.27.0 // indirect
 	golang.org/x/net v0.33.0 // indirect
 	golang.org/x/sys v0.29.0 // indirect
 	golang.org/x/text v0.21.0 // indirect
@@ -59,3 +62,5 @@ replace go.opentelemetry.io/collector/extension/extensionauth => go.opentelemetr
 replace go.opentelemetry.io/collector/extension/extensionauth/extensionauthtest => go.opentelemetry.io/collector/extension/extensionauth/extensionauthtest v0.0.0-20250226024140-8099e51f9a77
 
 replace go.opentelemetry.io/collector/service/hostcapabilities => go.opentelemetry.io/collector/service/hostcapabilities v0.0.0-20250226024140-8099e51f9a77
+
+replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil => ../../pkg/pdatautil