Cherrypick #5062 to 0.31.x: Use staleness markers generated by promet…

…heus, rather than making our own (#3990)

* Revert "receiver/prometheus: glue and complete staleness marking for disappearing metrics (#3423)"

This reverts commit 8b79380.

* Revert "receiver/prometheus: add store to track stale metrics (#3414)"

This reverts commit cdc1634.

* stop dropping staleness markers from prometheus, and fix tests

* add staleness end to end test from #3423

* fix import grouping

receiver/prometheusreceiver/internal/metricsbuilder.go

@@ -26,7 +26,6 @@ import (
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/pkg/labels"
 	"github.com/prometheus/prometheus/pkg/textparse"
-	"github.com/prometheus/prometheus/pkg/value"
 	"go.uber.org/zap"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
@@ -59,13 +58,12 @@ type metricBuilder struct {
 	startTime            float64
 	logger               *zap.Logger
 	currentMf            MetricFamily
-	stalenessStore       *stalenessStore
 }
 
 // newMetricBuilder creates a MetricBuilder which is allowed to feed all the datapoints from a single prometheus
 // scraped page by calling its AddDataPoint function, and turn them into an opencensus data.MetricsData object
 // by calling its Build function
-func newMetricBuilder(mc MetadataCache, useStartTimeMetric bool, startTimeMetricRegex string, logger *zap.Logger, stalenessStore *stalenessStore) *metricBuilder {
+func newMetricBuilder(mc MetadataCache, useStartTimeMetric bool, startTimeMetricRegex string, logger *zap.Logger) *metricBuilder {
 	var regex *regexp.Regexp
 	if startTimeMetricRegex != "" {
 		regex, _ = regexp.Compile(startTimeMetricRegex)
@@ -78,7 +76,6 @@ func newMetricBuilder(mc MetadataCache, useStartTimeMetric bool, startTimeMetric
 		droppedTimeseries:    0,
 		useStartTimeMetric:   useStartTimeMetric,
 		startTimeMetricRegex: regex,
-		stalenessStore:       stalenessStore,
 	}
 }
 
@@ -91,7 +88,7 @@ func (b *metricBuilder) matchStartTimeMetric(metricName string) bool {
 }
 
 // AddDataPoint is for feeding prometheus data complexValue in its processing order
-func (b *metricBuilder) AddDataPoint(ls labels.Labels, t int64, v float64) (rerr error) {
+func (b *metricBuilder) AddDataPoint(ls labels.Labels, t int64, v float64) error {
 	// Any datapoint with duplicate labels MUST be rejected per:
 	// * https://github.com/open-telemetry/wg-prometheus/issues/44
 	// * https://github.com/open-telemetry/opentelemetry-collector/issues/3407
@@ -109,14 +106,6 @@ func (b *metricBuilder) AddDataPoint(ls labels.Labels, t int64, v float64) (rerr
 		return fmt.Errorf("invalid sample: non-unique label names: %q", dupLabels)
 	}
 
-	defer func() {
-		// Only mark this data point as in the current scrape
-		// iff it isn't a stale metric.
-		if rerr == nil && !value.IsStaleNaN(v) {
-			b.stalenessStore.markAsCurrentlySeen(ls, t)
-		}
-	}()
-
 	metricName := ls.Get(model.MetricNameLabel)
 	switch {
 	case metricName == "":

receiver/prometheusreceiver/internal/metricsbuilder_test.go

@@ -99,7 +99,7 @@ func runBuilderTests(t *testing.T, tests []buildTestData) {
 			mc := newMockMetadataCache(testMetadata)
 			st := startTs
 			for i, page := range tt.inputs {
-				b := newMetricBuilder(mc, true, "", testLogger, dummyStalenessStore())
+				b := newMetricBuilder(mc, true, "", testLogger)
 				b.startTime = defaultBuilderStartTime // set to a non-zero value
 				for _, pt := range page.pts {
 					// set ts for testing
@@ -123,7 +123,7 @@ func runBuilderStartTimeTests(t *testing.T, tests []buildTestData,
 			st := startTs
 			for _, page := range tt.inputs {
 				b := newMetricBuilder(mc, true, startTimeMetricRegex,
-					testLogger, dummyStalenessStore())
+					testLogger)
 				b.startTime = defaultBuilderStartTime // set to a non-zero value
 				for _, pt := range page.pts {
 					// set ts for testing
@@ -1201,7 +1201,7 @@ func Test_metricBuilder_summary(t *testing.T) {
 func Test_metricBuilder_baddata(t *testing.T) {
 	t.Run("empty-metric-name", func(t *testing.T) {
 		mc := newMockMetadataCache(testMetadata)
-		b := newMetricBuilder(mc, true, "", testLogger, dummyStalenessStore())
+		b := newMetricBuilder(mc, true, "", testLogger)
 		b.startTime = 1.0 // set to a non-zero value
 		if err := b.AddDataPoint(labels.FromStrings("a", "b"), startTs, 123); err != errMetricNameNotFound {
 			t.Error("expecting errMetricNameNotFound error, but get nil")
@@ -1215,7 +1215,7 @@ func Test_metricBuilder_baddata(t *testing.T) {
 
 	t.Run("histogram-datapoint-no-bucket-label", func(t *testing.T) {
 		mc := newMockMetadataCache(testMetadata)
-		b := newMetricBuilder(mc, true, "", testLogger, dummyStalenessStore())
+		b := newMetricBuilder(mc, true, "", testLogger)
 		b.startTime = 1.0 // set to a non-zero value
 		if err := b.AddDataPoint(createLabels("hist_test", "k", "v"), startTs, 123); err != errEmptyBoundaryLabel {
 			t.Error("expecting errEmptyBoundaryLabel error, but get nil")
@@ -1224,7 +1224,7 @@ func Test_metricBuilder_baddata(t *testing.T) {
 
 	t.Run("summary-datapoint-no-quantile-label", func(t *testing.T) {
 		mc := newMockMetadataCache(testMetadata)
-		b := newMetricBuilder(mc, true, "", testLogger, dummyStalenessStore())
+		b := newMetricBuilder(mc, true, "", testLogger)
 		b.startTime = 1.0 // set to a non-zero value
 		if err := b.AddDataPoint(createLabels("summary_test", "k", "v"), startTs, 123); err != errEmptyBoundaryLabel {
 			t.Error("expecting errEmptyBoundaryLabel error, but get nil")
@@ -1452,7 +1452,7 @@ func Test_heuristicalMetricAndKnownUnits(t *testing.T) {
 // Ensure that we reject duplicate label keys. See https://github.com/open-telemetry/wg-prometheus/issues/44.
 func TestMetricBuilderDuplicateLabelKeysAreRejected(t *testing.T) {
 	mc := newMockMetadataCache(testMetadata)
-	mb := newMetricBuilder(mc, true, "", testLogger, dummyStalenessStore())
+	mb := newMetricBuilder(mc, true, "", testLogger)
 
 	dupLabels := labels.Labels{
 		{Name: "__name__", Value: "test"},

receiver/prometheusreceiver/internal/ocastore.go

@@ -51,8 +51,7 @@ type OcaStore struct {
 	receiverID           config.ComponentID
 	externalLabels       labels.Labels
 
-	logger         *zap.Logger
-	stalenessStore *stalenessStore
+	logger *zap.Logger
 }
 
 // NewOcaStore returns an ocaStore instance, which can be acted as prometheus' scrape.Appendable
@@ -75,7 +74,6 @@ func NewOcaStore(
 		startTimeMetricRegex: startTimeMetricRegex,
 		receiverID:           receiverID,
 		externalLabels:       externalLabels,
-		stalenessStore:       newStalenessStore(),
 	}
 }
 
@@ -90,9 +88,6 @@ func (o *OcaStore) SetScrapeManager(scrapeManager *scrape.Manager) {
 func (o *OcaStore) Appender(context.Context) storage.Appender {
 	state := atomic.LoadInt32(&o.running)
 	if state == runningStateReady {
-		// Firstly prepare the stalenessStore for a new scrape cyle.
-		o.stalenessStore.refresh()
-
 		return newTransaction(
 			o.ctx,
 			o.jobsMap,
@@ -103,7 +98,6 @@ func (o *OcaStore) Appender(context.Context) storage.Appender {
 			o.sink,
 			o.externalLabels,
 			o.logger,
-			o.stalenessStore,
 		)
 	} else if state == runningStateInit {
 		panic("ScrapeManager is not set")

receiver/prometheusreceiver/internal/staleness_end_to_end_test.go

@@ -121,14 +121,12 @@ receivers:
           scrape_interval: 2ms
           static_configs:
             - targets: [%q]
-
 processors:
   batch:
 exporters:
   prometheusremotewrite:
     endpoint: %q
     insecure: true
-
 service:
   pipelines:
     metrics: