add start_time_metric strategy to the metricstarttime processor (open-telemetry#41047)

#### Description

Add the `start_time_metric` strategy to the metricstarttime processor.
This is needed to reach feature parity with the prometheus receiver's
adjuster. It supports the same `start_time_metric_regex` configuration
parameter, and falls back to the collector's start time.

It does not yet support reset detection. That will be implemented in
open-telemetry#38381,
and I left a TODO in the code.

#### Link to tracking issue

Fixes
open-telemetry#38383

#### Testing

Added unit tests.

#### Documentation

Updated the README.md

@ridwanmsharif

Author: dashpole

.chloggen/metricstarttime-starttimemetric.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 the start_time_metric, which sets the start time based on another metric in the batch of metrics.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [38383]
+
+# (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, api]

processor/metricstarttimeprocessor/README.md

@@ -69,3 +69,34 @@ Cons:
 
 * The absolute value of counters is modified. This is generally not an issue, since counters are usually used to compute rates.
 * The initial point is dropped, which loses information.
+
+### Strategy: Start Time Metric
+
+The `start_time_metric` strategy handles missing start times by looking for the
+`process_start_time` metric, which is commonly supported by Prometheus exporters.
+If found, it uses the value of the `process_start_time` metric as the start time
+for all other cumulative points in the batch of metrics.
+
+Use the `start_time_metric_regex` configuration option to change the name of the
+metric used for the start time.
+
+If the start time metric is not found, it falls back to the time at which the
+collector started.
+
+This strategy should only be used in limited circumstances:
+
+* When your application has a metric with the start time in Unix nanoseconds,
+  such as `process_start_time`.
+* The metricstarttime processor is used _before_ any batching, so that the
+  batch of metrics all originate from a single application.
+* This strategy can be used when the collector is run as a sidecar to the
+  application, where the collector's start time is a good approximation of the
+  application's start time.
+
+Cons:
+
+* If the collector's start time is used as a fallback and the collector
+  restarts, it can produce rates that are incorrect and higher than expected.
+* The process' start time isn't the time at which individual instruments or
+  timeseries are initialized. It may result in lower rates if the first
+  observation is significantly later than the process' start time.

processor/metricstarttimeprocessor/config.go

@@ -6,10 +6,12 @@ package metricstarttimeprocessor // import "github.com/open-telemetry/openteleme
 import (
 	"errors"
 	"fmt"
+	"regexp"
 	"time"
 
 	"go.opentelemetry.io/collector/component"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/metricstarttimeprocessor/internal/starttimemetric"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/metricstarttimeprocessor/internal/subtractinitial"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/metricstarttimeprocessor/internal/truereset"
 )
@@ -18,6 +20,8 @@ import (
 type Config struct {
 	Strategy   string        `mapstructure:"strategy"`
 	GCInterval time.Duration `mapstructure:"gc_interval"`
+	// StartTimeMetricRegex only applies then the start_time_metric strategy is used
+	StartTimeMetricRegex string `mapstructure:"start_time_metric_regex"`
 }
 
 var _ component.Config = (*Config)(nil)
@@ -34,11 +38,20 @@ func (cfg *Config) Validate() error {
 	switch cfg.Strategy {
 	case truereset.Type:
 	case subtractinitial.Type:
+	case starttimemetric.Type:
 	default:
 		return fmt.Errorf("%q is not a valid strategy", cfg.Strategy)
 	}
 	if cfg.GCInterval <= 0 {
 		return errors.New("gc_interval must be positive")
 	}
+	if cfg.StartTimeMetricRegex != "" {
+		if _, err := regexp.Compile(cfg.StartTimeMetricRegex); err != nil {
+			return err
+		}
+		if cfg.Strategy != starttimemetric.Type {
+			return errors.New("start_time_metric_regex can only be used with the start_time_metric strategy")
+		}
+	}
 	return nil
 }

processor/metricstarttimeprocessor/config_test.go

@@ -15,6 +15,7 @@ import (
 	"go.opentelemetry.io/collector/confmap/xconfmap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/metricstarttimeprocessor/internal/metadata"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/metricstarttimeprocessor/internal/starttimemetric"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/metricstarttimeprocessor/internal/subtractinitial"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/metricstarttimeprocessor/internal/truereset"
 )
@@ -56,6 +57,29 @@ func TestLoadConfig(t *testing.T) {
 			id:           component.NewIDWithName(metadata.Type, "invalid_strategy"),
 			errorMessage: "\"bad\" is not a valid strategy",
 		},
+		{
+			id: component.NewIDWithName(metadata.Type, "true_reset_point"),
+			expected: &Config{
+				Strategy:   truereset.Type,
+				GCInterval: 10 * time.Minute,
+			},
+		},
+		{
+			id: component.NewIDWithName(metadata.Type, "start_time_metric"),
+			expected: &Config{
+				Strategy:             starttimemetric.Type,
+				GCInterval:           10 * time.Minute,
+				StartTimeMetricRegex: "^.+_process_start_time_seconds$",
+			},
+		},
+		{
+			id:           component.NewIDWithName(metadata.Type, "invalid_regex"),
+			errorMessage: "error parsing regexp: missing closing ): `((((`",
+		},
+		{
+			id:           component.NewIDWithName(metadata.Type, "regex_with_subtract_initial_point"),
+			errorMessage: "start_time_metric_regex can only be used with the start_time_metric strategy",
+		},
 	}
 
 	for _, tt := range tests {

processor/metricstarttimeprocessor/factory.go

@@ -5,13 +5,15 @@ package metricstarttimeprocessor // import "github.com/open-telemetry/openteleme
 
 import (
 	"context"
+	"regexp"
 
 	"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/starttimemetric"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/metricstarttimeprocessor/internal/subtractinitial"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/metricstarttimeprocessor/internal/truereset"
 )
@@ -42,6 +44,17 @@ func createMetricsProcessor(
 	case subtractinitial.Type:
 		adjuster := subtractinitial.NewAdjuster(set.TelemetrySettings, rCfg.GCInterval)
 		adjustMetrics = adjuster.AdjustMetrics
+	case starttimemetric.Type:
+		var startTimeMetricRegex *regexp.Regexp
+		var err error
+		if rCfg.StartTimeMetricRegex != "" {
+			startTimeMetricRegex, err = regexp.Compile(rCfg.StartTimeMetricRegex)
+			if err != nil {
+				return nil, err
+			}
+		}
+		adjuster := starttimemetric.NewAdjuster(set.TelemetrySettings, startTimeMetricRegex)
+		adjustMetrics = adjuster.AdjustMetrics
 	}
 
 	return processorhelper.NewMetrics(

processor/metricstarttimeprocessor/internal/starttimemetric/adjuster.go

@@ -0,0 +1,156 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package starttimemetric // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/metricstarttimeprocessor/internal/starttimemetric"
+
+import (
+	"context"
+	"errors"
+	"regexp"
+	"time"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.uber.org/zap"
+)
+
+const (
+	// Type is the value users can use to configure the start time metric adjuster.
+	Type                = "start_time_metric"
+	startTimeMetricName = "process_start_time_seconds"
+)
+
+var (
+	errNoStartTimeMetrics             = errors.New("start_time metric is missing")
+	errNoDataPointsStartTimeMetric    = errors.New("start time metric with no data points")
+	errUnsupportedTypeStartTimeMetric = errors.New("unsupported data type for start time metric")
+	// approximateCollectorStartTime is the approximate start time of the
+	// collector. Used as a fallback start time for metrics when the start time
+	// metric is not found. Set when the component is initialized.
+	approximateCollectorStartTime time.Time
+)
+
+func init() {
+	approximateCollectorStartTime = time.Now()
+}
+
+type Adjuster struct {
+	startTimeMetricRegex *regexp.Regexp
+	set                  component.TelemetrySettings
+}
+
+// NewAdjuster returns a new Adjuster which adjust metrics' start times based on the initial received points.
+func NewAdjuster(set component.TelemetrySettings, startTimeMetricRegex *regexp.Regexp) *Adjuster {
+	return &Adjuster{
+		set:                  set,
+		startTimeMetricRegex: startTimeMetricRegex,
+	}
+}
+
+// AdjustMetrics adjusts the start time of metrics based on a different metric in the batch.
+func (a *Adjuster) AdjustMetrics(_ context.Context, metrics pmetric.Metrics) (pmetric.Metrics, error) {
+	startTime, err := a.getStartTime(metrics)
+	if err != nil {
+		a.set.Logger.Debug("Couldn't get start time for metrics. Using fallback start time.", zap.Error(err), zap.Time("fallback_start_time", approximateCollectorStartTime))
+		startTime = float64(approximateCollectorStartTime.Unix())
+	}
+
+	startTimeTs := timestampFromFloat64(startTime)
+	for i := 0; i < metrics.ResourceMetrics().Len(); i++ {
+		rm := metrics.ResourceMetrics().At(i)
+		for j := 0; j < rm.ScopeMetrics().Len(); j++ {
+			ilm := rm.ScopeMetrics().At(j)
+			for k := 0; k < ilm.Metrics().Len(); k++ {
+				metric := ilm.Metrics().At(k)
+				switch metric.Type() {
+				case pmetric.MetricTypeGauge:
+					continue
+
+				case pmetric.MetricTypeSum:
+					dataPoints := metric.Sum().DataPoints()
+					for l := 0; l < dataPoints.Len(); l++ {
+						dp := dataPoints.At(l)
+						dp.SetStartTimestamp(startTimeTs)
+					}
+
+				case pmetric.MetricTypeSummary:
+					dataPoints := metric.Summary().DataPoints()
+					for l := 0; l < dataPoints.Len(); l++ {
+						dp := dataPoints.At(l)
+						dp.SetStartTimestamp(startTimeTs)
+					}
+
+				case pmetric.MetricTypeHistogram:
+					dataPoints := metric.Histogram().DataPoints()
+					for l := 0; l < dataPoints.Len(); l++ {
+						dp := dataPoints.At(l)
+						dp.SetStartTimestamp(startTimeTs)
+					}
+
+				case pmetric.MetricTypeExponentialHistogram:
+					dataPoints := metric.ExponentialHistogram().DataPoints()
+					for l := 0; l < dataPoints.Len(); l++ {
+						dp := dataPoints.At(l)
+						dp.SetStartTimestamp(startTimeTs)
+					}
+
+				case pmetric.MetricTypeEmpty:
+					fallthrough
+
+				default:
+					a.set.Logger.Warn("Unknown metric type", zap.String("type", metric.Type().String()))
+				}
+			}
+		}
+	}
+	// TODO: handle resets by factoring reset handling out of other strategies
+	return metrics, nil
+}
+
+func timestampFromFloat64(ts float64) pcommon.Timestamp {
+	secs := int64(ts)
+	nanos := int64((ts - float64(secs)) * 1e9)
+	return pcommon.Timestamp(secs*1e9 + nanos)
+}
+
+func (a *Adjuster) getStartTime(metrics pmetric.Metrics) (float64, error) {
+	for i := 0; i < metrics.ResourceMetrics().Len(); i++ {
+		rm := metrics.ResourceMetrics().At(i)
+		for j := 0; j < rm.ScopeMetrics().Len(); j++ {
+			ilm := rm.ScopeMetrics().At(j)
+			for k := 0; k < ilm.Metrics().Len(); k++ {
+				metric := ilm.Metrics().At(k)
+				if a.matchStartTimeMetric(metric.Name()) {
+					switch metric.Type() {
+					case pmetric.MetricTypeGauge:
+						if metric.Gauge().DataPoints().Len() == 0 {
+							return 0.0, errNoDataPointsStartTimeMetric
+						}
+						return metric.Gauge().DataPoints().At(0).DoubleValue(), nil
+
+					case pmetric.MetricTypeSum:
+						if metric.Sum().DataPoints().Len() == 0 {
+							return 0.0, errNoDataPointsStartTimeMetric
+						}
+						return metric.Sum().DataPoints().At(0).DoubleValue(), nil
+
+					case pmetric.MetricTypeEmpty, pmetric.MetricTypeHistogram, pmetric.MetricTypeExponentialHistogram, pmetric.MetricTypeSummary:
+						fallthrough
+					default:
+						return 0, errUnsupportedTypeStartTimeMetric
+					}
+				}
+			}
+		}
+	}
+	return 0.0, errNoStartTimeMetrics
+}
+
+func (a *Adjuster) matchStartTimeMetric(metricName string) bool {
+	if a.startTimeMetricRegex != nil {
+		return a.startTimeMetricRegex.MatchString(metricName)
+	}
+
+	return metricName == startTimeMetricName
+}