Add exemplar metrics for distributor received/in and validation discarded per reason

mdisibio · mdisibio · commit 9f0b9d8b51ac · 2021-04-23T15:24:13.000-04:00
Signed-off-by: Martin Disibio &lt;mdisibio@gmail.com&gt;
diff --git a/pkg/distributor/distributor.go b/pkg/distributor/distributor.go
@@ -93,8 +93,10 @@ type Distributor struct {
 	// Metrics
 	queryDuration                    *instrument.HistogramCollector
 	receivedSamples                  *prometheus.CounterVec
+	receivedExemplars                *prometheus.CounterVec
 	receivedMetadata                 *prometheus.CounterVec
 	incomingSamples                  *prometheus.CounterVec
+	incomingExemplars                *prometheus.CounterVec
 	incomingMetadata                 *prometheus.CounterVec
 	nonHASamples                     *prometheus.CounterVec
 	dedupedSamples                   *prometheus.CounterVec
@@ -241,6 +243,11 @@ func New(cfg Config, clientConfig ingester_client.Config, limits *validation.Ove
 			Name:      "distributor_received_samples_total",
 			Help:      "The total number of received samples, excluding rejected and deduped samples.",
 		}, []string{"user"}),
+		receivedExemplars: promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
+			Namespace: "cortex",
+			Name:      "distributor_received_exemplars_total",
+			Help:      "The total number of received exemplars, excluding rejected and deduped exemplars.",
+		}, []string{"user"}),
 		receivedMetadata: promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
 			Namespace: "cortex",
 			Name:      "distributor_received_metadata_total",
@@ -251,6 +258,11 @@ func New(cfg Config, clientConfig ingester_client.Config, limits *validation.Ove
 			Name:      "distributor_samples_in_total",
 			Help:      "The total number of samples that have come in to the distributor, including rejected or deduped samples.",
 		}, []string{"user"}),
+		incomingExemplars: promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
+			Namespace: "cortex",
+			Name:      "distributor_exemplars_in_total",
+			Help:      "The total number of exemplars that have come in to the distributor, including rejected or deduped exemplars.",
+		}, []string{"user"}),
 		incomingMetadata: promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
 			Namespace: "cortex",
 			Name:      "distributor_metadata_in_total",
@@ -375,8 +387,10 @@ func (d *Distributor) cleanupInactiveUser(userID string) {
 	d.HATracker.cleanupHATrackerMetricsForUser(userID)
 
 	d.receivedSamples.DeleteLabelValues(userID)
+	d.receivedExemplars.DeleteLabelValues(userID)
 	d.receivedMetadata.DeleteLabelValues(userID)
 	d.incomingSamples.DeleteLabelValues(userID)
+	d.incomingExemplars.DeleteLabelValues(userID)
 	d.incomingMetadata.DeleteLabelValues(userID)
 	d.nonHASamples.DeleteLabelValues(userID)
 	d.latestSeenSampleTimestampPerUser.DeleteLabelValues(userID)
@@ -543,11 +557,14 @@ func (d *Distributor) Push(ctx context.Context, req *cortexpb.WriteRequest) (*co
 	removeReplica := false
 
 	numSamples := 0
+	numExemplars := 0
 	for _, ts := range req.Timeseries {
 		numSamples += len(ts.Samples)
+		numExemplars += len(ts.Exemplars)
 	}
 	// Count the total samples in, prior to validation or deduplication, for comparison with other metrics.
 	d.incomingSamples.WithLabelValues(userID).Add(float64(numSamples))
+	d.incomingExemplars.WithLabelValues(userID).Add(float64(numExemplars))
 	// Count the total number of metadata in.
 	d.incomingMetadata.WithLabelValues(userID).Add(float64(len(req.Metadata)))
 
@@ -675,6 +692,7 @@ func (d *Distributor) Push(ctx context.Context, req *cortexpb.WriteRequest) (*co
 	}
 
 	d.receivedSamples.WithLabelValues(userID).Add(float64(validatedSamples))
+	d.receivedExemplars.WithLabelValues(userID).Add((float64(validatedExemplars)))
 	d.receivedMetadata.WithLabelValues(userID).Add(float64(len(validatedMetadata)))
 
 	if len(seriesKeys) == 0 && len(metadataKeys) == 0 {
@@ -692,6 +710,7 @@ func (d *Distributor) Push(ctx context.Context, req *cortexpb.WriteRequest) (*co
 		// Return a 4xx here to have the client discard the data and not retry. If a client
 		// is sending too much data consistently we will unlikely ever catch up otherwise.
 		validation.DiscardedSamples.WithLabelValues(validation.RateLimited, userID).Add(float64(validatedSamples))
+		validation.DiscardedExemplars.WithLabelValues(validation.RateLimited, userID).Add(float64(validatedExemplars))
 		validation.DiscardedMetadata.WithLabelValues(validation.RateLimited, userID).Add(float64(len(validatedMetadata)))
 		return nil, httpgrpc.Errorf(http.StatusTooManyRequests, "ingestion rate limit (%v) exceeded while adding %d samples and %d metadata", d.ingestionRateLimiter.Limit(now, userID), validatedSamples, len(validatedMetadata))
 	}
diff --git a/pkg/distributor/distributor_test.go b/pkg/distributor/distributor_test.go
@@ -273,19 +273,24 @@ func TestDistributor_MetricsCleanup(t *testing.T) {
 
 	metrics := []string{
 		"cortex_distributor_received_samples_total",
+		"cortex_distributor_received_exemplars_total",
 		"cortex_distributor_received_metadata_total",
 		"cortex_distributor_deduped_samples_total",
 		"cortex_distributor_samples_in_total",
+		"cortex_distributor_exemplars_in_total",
 		"cortex_distributor_metadata_in_total",
 		"cortex_distributor_non_ha_samples_received_total",
 		"cortex_distributor_latest_seen_sample_timestamp_seconds",
 	}
 
 	d.receivedSamples.WithLabelValues("userA").Add(5)
 	d.receivedSamples.WithLabelValues("userB").Add(10)
+	d.receivedExemplars.WithLabelValues("userA").Add(5)
+	d.receivedExemplars.WithLabelValues("userB").Add(10)
 	d.receivedMetadata.WithLabelValues("userA").Add(5)
 	d.receivedMetadata.WithLabelValues("userB").Add(10)
 	d.incomingSamples.WithLabelValues("userA").Add(5)
+	d.incomingExemplars.WithLabelValues("userA").Add(5)
 	d.incomingMetadata.WithLabelValues("userA").Add(5)
 	d.nonHASamples.WithLabelValues("userA").Add(5)
 	d.dedupedSamples.WithLabelValues("userA", "cluster1").Inc() // We cannot clean this metric
@@ -318,10 +323,19 @@ func TestDistributor_MetricsCleanup(t *testing.T) {
 		cortex_distributor_received_samples_total{user="userA"} 5
 		cortex_distributor_received_samples_total{user="userB"} 10
 
+		# HELP cortex_distributor_received_exemplars_total The total number of received exemplars, excluding rejected and deduped exemplars.
+		# TYPE cortex_distributor_received_exemplars_total counter
+		cortex_distributor_received_exemplars_total{user="userA"} 5
+		cortex_distributor_received_exemplars_total{user="userB"} 10
+
 		# HELP cortex_distributor_samples_in_total The total number of samples that have come in to the distributor, including rejected or deduped samples.
 		# TYPE cortex_distributor_samples_in_total counter
 		cortex_distributor_samples_in_total{user="userA"} 5
-`), metrics...))
+
+		# HELP cortex_distributor_exemplars_in_total The total number of exemplars that have come in to the distributor, including rejected or deduped exemplars.
+		# TYPE cortex_distributor_exemplars_in_total counter
+		cortex_distributor_exemplars_in_total{user="userA"} 5
+		`), metrics...))
 
 	d.cleanupInactiveUser("userA")
 
@@ -346,9 +360,16 @@ func TestDistributor_MetricsCleanup(t *testing.T) {
 		# TYPE cortex_distributor_received_samples_total counter
 		cortex_distributor_received_samples_total{user="userB"} 10
 
+		# HELP cortex_distributor_received_exemplars_total The total number of received exemplars, excluding rejected and deduped exemplars.
+		# TYPE cortex_distributor_received_exemplars_total counter
+		cortex_distributor_received_exemplars_total{user="userB"} 10
+
 		# HELP cortex_distributor_samples_in_total The total number of samples that have come in to the distributor, including rejected or deduped samples.
 		# TYPE cortex_distributor_samples_in_total counter
-`), metrics...))
+
+		# HELP cortex_distributor_exemplars_in_total The total number of exemplars that have come in to the distributor, including rejected or deduped exemplars.
+		# TYPE cortex_distributor_exemplars_in_total counter
+		`), metrics...))
 }
 
 func TestDistributor_PushIngestionRateLimiter(t *testing.T) {
diff --git a/pkg/util/validation/validate.go b/pkg/util/validation/validate.go
@@ -45,6 +45,11 @@ const (
 	labelsNotSorted         = "labels_not_sorted"
 	labelValueTooLong       = "label_value_too_long"
 
+	// Exemplar-specific validation reasons
+	exemplarLabelsMissing    = "exemplar_labels_missing"
+	exemplarLabelsTooLong    = "exemplar_labels_too_long"
+	exemplarTimestampInvalid = "exemplar_timestamp_invalid"
+
 	// RateLimited is one of the values for the reason to discard samples.
 	// Declared here to avoid duplication in ingester and distributor.
 	RateLimited = "rate_limited"
@@ -66,6 +71,15 @@ var DiscardedSamples = prometheus.NewCounterVec(
 	[]string{discardReasonLabel, "user"},
 )
 
+// DiscardedExemplars is a metric of the number of discarded exemplars, by reason.
+var DiscardedExemplars = prometheus.NewCounterVec(
+	prometheus.CounterOpts{
+		Name: "cortex_discarded_exemplars_total",
+		Help: "The total number of exemplars that were discarded.",
+	},
+	[]string{discardReasonLabel, "user"},
+)
+
 // DiscardedMetadata is a metric of the number of discarded metadata, by reason.
 var DiscardedMetadata = prometheus.NewCounterVec(
 	prometheus.CounterOpts{
@@ -77,6 +91,7 @@ var DiscardedMetadata = prometheus.NewCounterVec(
 
 func init() {
 	prometheus.MustRegister(DiscardedSamples)
+	prometheus.MustRegister(DiscardedExemplars)
 	prometheus.MustRegister(DiscardedMetadata)
 }
 
@@ -107,11 +122,13 @@ func ValidateSample(cfg SampleValidationConfig, userID string, ls []cortexpb.Lab
 
 func ValidateExemplar(userID string, ls []cortexpb.LabelAdapter, e cortexpb.Exemplar) ValidationError {
 	if len(e.Labels) <= 0 {
+		DiscardedExemplars.WithLabelValues(exemplarLabelsMissing, userID).Inc()
 		return fmt.Errorf(`exemplar missing labels: series: %s`,
 			cortexpb.FromLabelAdaptersToLabels(ls).String())
 	}
 
 	if e.TimestampMs == 0 {
+		DiscardedExemplars.WithLabelValues(exemplarTimestampInvalid, userID).Inc()
 		return fmt.Errorf(`exemplar missing timestamp: series: %s labels: %s`,
 			cortexpb.FromLabelAdaptersToLabels(ls).String(),
 			cortexpb.FromLabelAdaptersToLabels(e.Labels).String())
@@ -124,6 +141,7 @@ func ValidateExemplar(userID string, ls []cortexpb.LabelAdapter, e cortexpb.Exem
 	}
 
 	if labelSetLen > ExemplarMaxLabelSetLength {
+		DiscardedExemplars.WithLabelValues(exemplarLabelsTooLong, userID).Inc()
 		return fmt.Errorf(`exemplar combined labelset too long: series: %s labels: %s`,
 			cortexpb.FromLabelAdaptersToLabels(ls).String(),
 			cortexpb.FromLabelAdaptersToLabels(e.Labels).String())
@@ -235,6 +253,9 @@ func DeletePerUserValidationMetrics(userID string, log log.Logger) {
 	if err := util.DeleteMatchingLabels(DiscardedSamples, filter); err != nil {
 		level.Warn(log).Log("msg", "failed to remove cortex_discarded_samples_total metric for user", "user", userID, "err", err)
 	}
+	if err := util.DeleteMatchingLabels(DiscardedExemplars, filter); err != nil {
+		level.Warn(log).Log("msg", "failed to remove cortex_discarded_exemplars_total metric for user", "user", userID, "err", err)
+	}
 	if err := util.DeleteMatchingLabels(DiscardedMetadata, filter); err != nil {
 		level.Warn(log).Log("msg", "failed to remove cortex_discarded_metadata_total metric for user", "user", userID, "err", err)
 	}
diff --git a/pkg/util/validation/validate_test.go b/pkg/util/validation/validate_test.go
@@ -148,6 +148,51 @@ func TestValidateLabels(t *testing.T) {
 	`), "cortex_discarded_samples_total"))
 }
 
+func TestValidateExemplars(t *testing.T) {
+	userID := "testUser"
+
+	invalidExemplars := []cortexpb.Exemplar{
+		{
+			// Missing labels
+			Labels: nil,
+		},
+		{
+			// Invalid timestamp
+			Labels: []cortexpb.LabelAdapter{{Name: "foo", Value: "bar"}},
+		},
+		{
+			// Combined lableset too long
+			Labels:      []cortexpb.LabelAdapter{{Name: "foo", Value: strings.Repeat("0", 126)}},
+			TimestampMs: 1000,
+		},
+	}
+
+	for _, ie := range invalidExemplars {
+		err := ValidateExemplar(userID, []cortexpb.LabelAdapter{}, ie)
+		assert.NotNil(t, err)
+	}
+
+	DiscardedExemplars.WithLabelValues("random reason", "different user").Inc()
+
+	require.NoError(t, testutil.GatherAndCompare(prometheus.DefaultGatherer, strings.NewReader(`
+			# HELP cortex_discarded_exemplars_total The total number of exemplars that were discarded.
+			# TYPE cortex_discarded_exemplars_total counter
+			cortex_discarded_exemplars_total{reason="exemplar_labels_missing",user="testUser"} 1
+			cortex_discarded_exemplars_total{reason="exemplar_labels_too_long",user="testUser"} 1
+			cortex_discarded_exemplars_total{reason="exemplar_timestamp_invalid",user="testUser"} 1
+
+			cortex_discarded_exemplars_total{reason="random reason",user="different user"} 1
+		`), "cortex_discarded_exemplars_total"))
+
+	// Delete test user and verify only different remaining
+	DeletePerUserValidationMetrics(userID, util_log.Logger)
+	require.NoError(t, testutil.GatherAndCompare(prometheus.DefaultGatherer, strings.NewReader(`
+			# HELP cortex_discarded_exemplars_total The total number of exemplars that were discarded.
+			# TYPE cortex_discarded_exemplars_total counter
+			cortex_discarded_exemplars_total{reason="random reason",user="different user"} 1
+	`), "cortex_discarded_exemplars_total"))
+}
+
 func TestValidateMetadata(t *testing.T) {
 	userID := "testUser"
 	var cfg validateMetadataCfg
