Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

## master / unreleased

* [CHANGE] Distributor/Ingester: Add `type` label to `cortex_discarded_samples_total` metric to distinguish between float and native histogram samples. #7898
* [FEATURE] Ruler: Add experimental support for federated rule groups. A rule group listing tenants in its `source_tenants` field is evaluated against those tenants while the resulting series and alerts are written to the tenant owning the rule group. Enabled with `-ruler.enable-federated-rules` (requires `-tenant-federation.enabled`), and restricted to selected tenants with `-ruler.allowed-federated-tenants` and `-ruler.disallowed-federated-tenants`. #7828
* [FEATURE] Distributor: Add experimental OTLP metrics ingestion over gRPC (`opentelemetry.proto.collector.metrics.v1.MetricsService/Export`) on the gRPC server port. Enable it with `-distributor.otlp.grpc-enabled`. The tenant is read from the `X-Scope-OrgID` gRPC metadata, and the request size is limited by `-server.grpc-max-recv-msg-size-bytes`. #7873
* [ENHANCEMENT] Ruler: Add rule information to query parameters sent to the Query Frontend when `-ruler.frontend-address` is configured, so the Query Frontend query stats logs (`-frontend.enabled-ruler-query-stats`) can identify the evaluated rule. In this mode, the Ruler query stats log (`-ruler.query-stats-enabled`) omits the fetched statistics fields, which are always zero. #6539
Expand Down
73 changes: 56 additions & 17 deletions pkg/distributor/distributor.go
Original file line number Diff line number Diff line change
Expand Up @@ -689,7 +689,7 @@ func (d *Distributor) checkSample(ctx context.Context, userID, cluster, replica
func (d *Distributor) validateSeries(ts cortexpb.PreallocTimeseries, userID string, skipLabelNameValidation bool, limits *validation.Limits) (cortexpb.PreallocTimeseries, validation.ValidationError) {
d.labelsHistogram.Observe(float64(len(ts.Labels)))

if err := validation.ValidateLabels(d.validateMetrics, limits, userID, ts.Labels, skipLabelNameValidation, d.cfg.NameValidationScheme); err != nil {
if err := validation.ValidateLabels(d.validateMetrics, limits, userID, ts.Labels, skipLabelNameValidation, d.cfg.NameValidationScheme, len(ts.Samples), len(ts.Histograms)); err != nil {
return emptyPreallocSeries, err
}

Expand All @@ -698,7 +698,7 @@ func (d *Distributor) validateSeries(ts cortexpb.PreallocTimeseries, userID stri
// Only alloc when data present
samples = make([]cortexpb.Sample, 0, len(ts.Samples))
for _, s := range ts.Samples {
if err := validation.ValidateSampleTimestamp(d.validateMetrics, limits, userID, ts.Labels, s.TimestampMs); err != nil {
if err := validation.ValidateSampleTimestamp(d.validateMetrics, limits, userID, ts.Labels, s.TimestampMs, validation.SampleTypeFloat); err != nil {
return emptyPreallocSeries, err
}
samples = append(samples, s)
Expand Down Expand Up @@ -727,7 +727,7 @@ func (d *Distributor) validateSeries(ts cortexpb.PreallocTimeseries, userID stri
histograms = make([]cortexpb.WrappedHistogram, 0, len(ts.Histograms))
receivedBucketsObserver := d.receivedHistogramBuckets.WithLabelValues(userID)
for i, h := range ts.Histograms {
if err := validation.ValidateSampleTimestamp(d.validateMetrics, limits, userID, ts.Labels, h.TimestampMs); err != nil {
if err := validation.ValidateSampleTimestamp(d.validateMetrics, limits, userID, ts.Labels, h.TimestampMs, validation.SampleTypeHistogram); err != nil {
return emptyPreallocSeries, err
}
receivedBucketsObserver.Observe(float64(h.BucketCount()))
Expand Down Expand Up @@ -833,7 +833,12 @@ func (d *Distributor) Push(ctx context.Context, req *cortexpb.WriteRequest) (*co
}

if errors.Is(err, ha.TooManyReplicaGroupsError{}) {
d.validateMetrics.DiscardedSamples.WithLabelValues(validation.TooManyHAClusters, userID).Add(float64(numFloatSamples + numHistogramSamples))
if numFloatSamples > 0 {
d.validateMetrics.DiscardedSamples.WithLabelValues(validation.TooManyHAClusters, userID, validation.SampleTypeFloat).Add(float64(numFloatSamples))
}
if numHistogramSamples > 0 {
d.validateMetrics.DiscardedSamples.WithLabelValues(validation.TooManyHAClusters, userID, validation.SampleTypeHistogram).Add(float64(numHistogramSamples))
}
return nil, httpgrpc.Errorf(http.StatusBadRequest, "%s", err.Error())
}
return nil, err
Expand Down Expand Up @@ -864,7 +869,12 @@ func (d *Distributor) Push(ctx context.Context, req *cortexpb.WriteRequest) (*co
totalSamples := validatedFloatSamples + validatedHistogramSamples
totalN := totalSamples + validatedExemplars + len(validatedMetadata)
if !d.ingestionRateLimiter.AllowN(now, userID, totalN) {
d.validateMetrics.DiscardedSamples.WithLabelValues(validation.RateLimited, userID).Add(float64(totalSamples))
if validatedFloatSamples > 0 {
d.validateMetrics.DiscardedSamples.WithLabelValues(validation.RateLimited, userID, validation.SampleTypeFloat).Add(float64(validatedFloatSamples))
}
if validatedHistogramSamples > 0 {
d.validateMetrics.DiscardedSamples.WithLabelValues(validation.RateLimited, userID, validation.SampleTypeHistogram).Add(float64(validatedHistogramSamples))
}
d.validateMetrics.DiscardedExemplars.WithLabelValues(validation.RateLimited, userID).Add(float64(validatedExemplars))
d.validateMetrics.DiscardedMetadata.WithLabelValues(validation.RateLimited, userID).Add(float64(len(validatedMetadata)))
// Return a 429 here to tell the client it is going too fast.
Expand All @@ -879,7 +889,7 @@ func (d *Distributor) Push(ctx context.Context, req *cortexpb.WriteRequest) (*co
var nativeHistogramErr error

if !d.nativeHistogramIngestionRateLimiter.AllowN(now, userID, validatedHistogramSamples) {
d.validateMetrics.DiscardedSamples.WithLabelValues(validation.NativeHistogramRateLimited, userID).Add(float64(validatedHistogramSamples))
d.validateMetrics.DiscardedSamples.WithLabelValues(validation.NativeHistogramRateLimited, userID, validation.SampleTypeHistogram).Add(float64(validatedHistogramSamples))
nativeHistogramErr = httpgrpc.Errorf(http.StatusTooManyRequests, "native histogram ingestion rate limit (%v) exceeded while adding %d native histogram samples", d.nativeHistogramIngestionRateLimiter.Limit(now, userID), validatedHistogramSamples)
validatedHistogramSamples = 0
} else {
Expand Down Expand Up @@ -1121,7 +1131,12 @@ func (d *Distributor) prepareSeriesKeys(ctx context.Context, req *cortexpb.Write
d.dedupedSamples.WithLabelValues(userID, cluster).Add(float64(len(ts.Samples) + len(ts.Histograms)))
}
if errors.Is(err, ha.TooManyReplicaGroupsError{}) {
d.validateMetrics.DiscardedSamples.WithLabelValues(validation.TooManyHAClusters, userID).Add(float64(len(ts.Samples) + len(ts.Histograms)))
if len(ts.Samples) > 0 {
d.validateMetrics.DiscardedSamples.WithLabelValues(validation.TooManyHAClusters, userID, validation.SampleTypeFloat).Add(float64(len(ts.Samples)))
}
if len(ts.Histograms) > 0 {
d.validateMetrics.DiscardedSamples.WithLabelValues(validation.TooManyHAClusters, userID, validation.SampleTypeHistogram).Add(float64(len(ts.Histograms)))
}
}

continue
Expand All @@ -1145,10 +1160,20 @@ func (d *Distributor) prepareSeriesKeys(ctx context.Context, req *cortexpb.Write
l, _ := relabel.Process(cortexpb.FromLabelAdaptersToLabels(ts.Labels), mrc...)
if l.Len() == 0 {
// all labels are gone, samples will be discarded
d.validateMetrics.DiscardedSamples.WithLabelValues(
validation.DroppedByRelabelConfiguration,
userID,
).Add(float64(len(ts.Samples) + len(ts.Histograms)))
if len(ts.Samples) > 0 {
d.validateMetrics.DiscardedSamples.WithLabelValues(
validation.DroppedByRelabelConfiguration,
userID,
validation.SampleTypeFloat,
).Add(float64(len(ts.Samples)))
}
if len(ts.Histograms) > 0 {
d.validateMetrics.DiscardedSamples.WithLabelValues(
validation.DroppedByRelabelConfiguration,
userID,
validation.SampleTypeHistogram,
).Add(float64(len(ts.Histograms)))
}

// all labels are gone, exemplars will be discarded
d.validateMetrics.DiscardedExemplars.WithLabelValues(
Expand All @@ -1173,12 +1198,16 @@ func (d *Distributor) prepareSeriesKeys(ctx context.Context, req *cortexpb.Write

// Reject series with missing or empty metric name before removeEmptyLabels (which would strip __name__="").
if validationErr, reason := validation.ValidateMetricName(limits, ts.Labels, d.cfg.NameValidationScheme); reason != "" {
samplesCount := float64(len(ts.Samples) + len(ts.Histograms))
exemplarsCount := float64(len(ts.Exemplars))
if firstPartialErr == nil {
firstPartialErr = httpgrpc.Errorf(http.StatusBadRequest, "%s", validationErr.Error())
}
d.validateMetrics.DiscardedSamples.WithLabelValues(reason, userID).Add(samplesCount)
if len(ts.Samples) > 0 {
d.validateMetrics.DiscardedSamples.WithLabelValues(reason, userID, validation.SampleTypeFloat).Add(float64(len(ts.Samples)))
}
if len(ts.Histograms) > 0 {
d.validateMetrics.DiscardedSamples.WithLabelValues(reason, userID, validation.SampleTypeHistogram).Add(float64(len(ts.Histograms)))
}
d.validateMetrics.DiscardedExemplars.WithLabelValues(reason, userID).Add(exemplarsCount)
continue
}
Expand All @@ -1187,10 +1216,20 @@ func (d *Distributor) prepareSeriesKeys(ctx context.Context, req *cortexpb.Write
removeEmptyLabels(&ts.Labels)

if len(ts.Labels) == 0 {
d.validateMetrics.DiscardedSamples.WithLabelValues(
validation.DroppedByUserConfigurationOverride,
userID,
).Add(float64(len(ts.Samples) + len(ts.Histograms)))
if len(ts.Samples) > 0 {
d.validateMetrics.DiscardedSamples.WithLabelValues(
validation.DroppedByUserConfigurationOverride,
userID,
validation.SampleTypeFloat,
).Add(float64(len(ts.Samples)))
}
if len(ts.Histograms) > 0 {
d.validateMetrics.DiscardedSamples.WithLabelValues(
validation.DroppedByUserConfigurationOverride,
userID,
validation.SampleTypeHistogram,
).Add(float64(len(ts.Histograms)))
}

d.validateMetrics.DiscardedExemplars.WithLabelValues(
validation.DroppedByUserConfigurationOverride,
Expand Down
14 changes: 11 additions & 3 deletions pkg/distributor/distributor_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -981,7 +981,7 @@ func TestDistributor_PushIngestionRateLimiter_Histograms(t *testing.T) {
assert.Nil(t, response)
}
}
assert.Equal(t, float64(push.expectedNHDiscardedSampleMetricValue), testutil.ToFloat64(distributors[0].validateMetrics.DiscardedSamples.WithLabelValues(validation.NativeHistogramRateLimited, "user")))
assert.Equal(t, float64(push.expectedNHDiscardedSampleMetricValue), testutil.ToFloat64(distributors[0].validateMetrics.DiscardedSamples.WithLabelValues(validation.NativeHistogramRateLimited, "user", validation.SampleTypeHistogram)))
}
})
}
Expand Down Expand Up @@ -2142,7 +2142,11 @@ func TestDistributor_Push_LabelRemoval(t *testing.T) {
_, err = ds[0].Push(ctx, req)
require.NoError(t, err)

actualDiscardedSamples := testutil.ToFloat64(ds[0].validateMetrics.DiscardedSamples.WithLabelValues(validation.DroppedByUserConfigurationOverride, "user"))
sampleType := validation.SampleTypeFloat
if histogram {
sampleType = validation.SampleTypeHistogram
}
actualDiscardedSamples := testutil.ToFloat64(ds[0].validateMetrics.DiscardedSamples.WithLabelValues(validation.DroppedByUserConfigurationOverride, "user", sampleType))
actualDiscardedExemplars := testutil.ToFloat64(ds[0].validateMetrics.DiscardedExemplars.WithLabelValues(validation.DroppedByUserConfigurationOverride, "user"))
require.Equal(t, float64(expectedDiscardedSamples), actualDiscardedSamples)
require.Equal(t, float64(expectedDiscardedExemplars), actualDiscardedExemplars)
Expand Down Expand Up @@ -4613,7 +4617,11 @@ func TestDistributor_Push_RelabelDropWillExportMetricOfDroppedSamples(t *testing
assert.Equal(t, 1, len(timeseries))
}

require.Equal(t, testutil.ToFloat64(ds[0].validateMetrics.DiscardedSamples.WithLabelValues(validation.DroppedByRelabelConfiguration, id)), float64(1))
sampleType := validation.SampleTypeFloat
if histogramEnabled {
sampleType = validation.SampleTypeHistogram
}
require.Equal(t, testutil.ToFloat64(ds[0].validateMetrics.DiscardedSamples.WithLabelValues(validation.DroppedByRelabelConfiguration, id, sampleType)), 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"))
Expand Down
Loading