diff --git a/pkg/distributor/distributor.go b/pkg/distributor/distributor.go index 351187ba83..a3aa75b5b8 100644 --- a/pkg/distributor/distributor.go +++ b/pkg/distributor/distributor.go @@ -874,7 +874,13 @@ func (d *Distributor) prepareSeriesKeys(ctx context.Context, req *cortexpb.Write d.validateMetrics.DiscardedSamples.WithLabelValues( validation.DroppedByRelabelConfiguration, userID, - ).Add(float64(len(ts.Samples))) + ).Add(float64(len(ts.Samples) + len(ts.Histograms))) + + // all labels are gone, exemplars will be discarded + d.validateMetrics.DiscardedExemplars.WithLabelValues( + validation.DroppedByRelabelConfiguration, + userID, + ).Add(float64(len(ts.Exemplars))) continue } ts.Labels = cortexpb.FromLabelsToLabelAdapters(l) @@ -892,11 +898,15 @@ func (d *Distributor) prepareSeriesKeys(ctx context.Context, req *cortexpb.Write } if len(ts.Labels) == 0 { - d.validateMetrics.DiscardedExemplars.WithLabelValues( + d.validateMetrics.DiscardedSamples.WithLabelValues( validation.DroppedByUserConfigurationOverride, userID, - ).Add(float64(len(ts.Samples))) + ).Add(float64(len(ts.Samples) + len(ts.Histograms))) + d.validateMetrics.DiscardedExemplars.WithLabelValues( + validation.DroppedByUserConfigurationOverride, + userID, + ).Add(float64(len(ts.Exemplars))) continue } diff --git a/pkg/distributor/distributor_test.go b/pkg/distributor/distributor_test.go index e5eda834ca..28cf8b1dc6 100644 --- a/pkg/distributor/distributor_test.go +++ b/pkg/distributor/distributor_test.go @@ -1490,6 +1490,7 @@ func TestDistributor_Push_LabelRemoval(t *testing.T) { expectedSeries labels.Labels removeReplica bool removeLabels []string + exemplars []cortexpb.Exemplar } cases := []testcase{ @@ -1536,6 +1537,20 @@ func TestDistributor_Push_LabelRemoval(t *testing.T) { {Name: "cluster", Value: "one"}, }, }, + // No labels left. + { + removeReplica: true, + removeLabels: []string{"cluster"}, + inputSeries: labels.Labels{ + {Name: "cluster", Value: "one"}, + {Name: "__replica__", Value: "two"}, + }, + expectedSeries: labels.Labels{}, + exemplars: []cortexpb.Exemplar{ + {Labels: cortexpb.FromLabelsToLabelAdapters(labels.FromStrings("test", "a")), Value: 1, TimestampMs: 0}, + {Labels: cortexpb.FromLabelsToLabelAdapters(labels.FromStrings("test", "b")), Value: 1, TimestampMs: 0}, + }, + }, } for _, tc := range cases { @@ -1546,6 +1561,15 @@ func TestDistributor_Push_LabelRemoval(t *testing.T) { limits.DropLabels = tc.removeLabels limits.AcceptHASamples = tc.removeReplica + expectedDiscardedSamples := 0 + expectedDiscardedExemplars := 0 + if tc.expectedSeries.Len() == 0 { + expectedDiscardedSamples = 1 + expectedDiscardedExemplars = len(tc.exemplars) + // Allow series with no labels to ingest + limits.EnforceMetricName = false + } + ds, ingesters, _, _ := prepare(t, prepConfig{ numIngesters: 2, happyIngesters: 2, @@ -1556,14 +1580,24 @@ func TestDistributor_Push_LabelRemoval(t *testing.T) { // Push the series to the distributor req := mockWriteRequest([]labels.Labels{tc.inputSeries}, 1, 1, histogram) + req.Timeseries[0].Exemplars = tc.exemplars _, err = ds[0].Push(ctx, req) require.NoError(t, err) + actualDiscardedSamples := testutil.ToFloat64(ds[0].validateMetrics.DiscardedSamples.WithLabelValues(validation.DroppedByUserConfigurationOverride, "user")) + actualDiscardedExemplars := testutil.ToFloat64(ds[0].validateMetrics.DiscardedExemplars.WithLabelValues(validation.DroppedByUserConfigurationOverride, "user")) + require.Equal(t, float64(expectedDiscardedSamples), actualDiscardedSamples) + require.Equal(t, float64(expectedDiscardedExemplars), actualDiscardedExemplars) + // Since each test pushes only 1 series, we do expect the ingester // to have received exactly 1 series for i := range ingesters { timeseries := ingesters[i].series() - assert.Equal(t, 1, len(timeseries)) + expectedSeries := 1 + if tc.expectedSeries.Len() == 0 { + expectedSeries = 0 + } + assert.Equal(t, expectedSeries, len(timeseries)) for _, v := range timeseries { assert.Equal(t, tc.expectedSeries, cortexpb.FromLabelAdaptersToLabels(v.Labels)) } @@ -3777,39 +3811,43 @@ func TestDistributor_Push_RelabelDropWillExportMetricOfDroppedSamples(t *testing flagext.DefaultValues(&limits) limits.MetricRelabelConfigs = metricRelabelConfigs - ds, ingesters, regs, _ := prepare(t, prepConfig{ - numIngesters: 2, - happyIngesters: 2, - numDistributors: 1, - shardByAllLabels: true, - limits: &limits, - }) - - // Push the series to the distributor - req := mockWriteRequest(inputSeries, 1, 1, false) - ctx := user.InjectOrgID(context.Background(), "userDistributorPushRelabelDropWillExportMetricOfDroppedSamples") - _, err = ds[0].Push(ctx, req) - require.NoError(t, err) + for _, histogramEnabled := range []bool{false, true} { + ds, ingesters, _, _ := prepare(t, prepConfig{ + numIngesters: 2, + happyIngesters: 2, + numDistributors: 1, + shardByAllLabels: true, + limits: &limits, + }) - // Since each test pushes only 1 series, we do expect the ingester - // to have received exactly 1 series - for i := range ingesters { - timeseries := ingesters[i].series() - assert.Equal(t, 1, len(timeseries)) - } + // Push the series to the distributor + id := "user" + req := mockWriteRequest(inputSeries, 1, 1, histogramEnabled) + req.Timeseries[0].Exemplars = []cortexpb.Exemplar{ + {Labels: cortexpb.FromLabelsToLabelAdapters(labels.FromStrings("test", "a")), Value: 1, TimestampMs: 0}, + {Labels: cortexpb.FromLabelsToLabelAdapters(labels.FromStrings("test", "b")), Value: 1, TimestampMs: 0}, + } + ctx := user.InjectOrgID(context.Background(), id) + _, err = ds[0].Push(ctx, req) + require.NoError(t, err) - metrics := []string{"cortex_distributor_received_samples_total", "cortex_discarded_samples_total"} + for i := range ingesters { + timeseries := ingesters[i].series() + assert.Equal(t, 1, len(timeseries)) + } - expectedMetrics := ` - # HELP cortex_discarded_samples_total The total number of samples that were discarded. - # TYPE cortex_discarded_samples_total counter - cortex_discarded_samples_total{reason="relabel_configuration",user="userDistributorPushRelabelDropWillExportMetricOfDroppedSamples"} 1 - # HELP cortex_distributor_received_samples_total The total number of received samples, excluding rejected and deduped samples. - # TYPE cortex_distributor_received_samples_total counter - cortex_distributor_received_samples_total{type="float",user="userDistributorPushRelabelDropWillExportMetricOfDroppedSamples"} 1 - cortex_distributor_received_samples_total{type="histogram",user="userDistributorPushRelabelDropWillExportMetricOfDroppedSamples"} 0 - ` - require.NoError(t, testutil.GatherAndCompare(regs[0], strings.NewReader(expectedMetrics), metrics...)) + require.Equal(t, testutil.ToFloat64(ds[0].validateMetrics.DiscardedSamples.WithLabelValues(validation.DroppedByRelabelConfiguration, id)), float64(1)) + require.Equal(t, testutil.ToFloat64(ds[0].validateMetrics.DiscardedExemplars.WithLabelValues(validation.DroppedByRelabelConfiguration, id)), float64(2)) + receivedFloatSamples := testutil.ToFloat64(ds[0].receivedSamples.WithLabelValues(id, "float")) + receivedHistogramSamples := testutil.ToFloat64(ds[0].receivedSamples.WithLabelValues(id, "histogram")) + if histogramEnabled { + require.Equal(t, receivedFloatSamples, float64(0)) + require.Equal(t, receivedHistogramSamples, float64(1)) + } else { + require.Equal(t, receivedFloatSamples, float64(1)) + require.Equal(t, receivedHistogramSamples, float64(0)) + } + } } func countMockIngestersCalls(ingesters []*mockIngester, name string) int {