feat: Do not enforce labels vs agg metric stream

pkg/distributor/distributor.go

@@ -548,13 +548,15 @@ func (d *Distributor) PushWithResolver(ctx context.Context, req *logproto.PushRe
 				continue
 			}
 
-			if missing, lbsMissing := d.missingEnforcedLabels(lbs, tenantID, policy); missing {
-				err := fmt.Errorf(validation.MissingEnforcedLabelsErrorMsg, strings.Join(lbsMissing, ","), tenantID, stream.Labels)
-				d.writeFailuresManager.Log(tenantID, err)
-				validationErrors.Add(err)
-				discardedBytes := util.EntriesTotalSize(stream.Entries)
-				d.validator.reportDiscardedDataWithTracker(ctx, validation.MissingEnforcedLabels, validationContext, lbs, retentionHours, policy, discardedBytes, len(stream.Entries))
-				continue
+			if !d.validator.IsAggregatedMetricStream(lbs) {
+				if missing, lbsMissing := d.missingEnforcedLabels(lbs, tenantID, policy); missing {
+					err := fmt.Errorf(validation.MissingEnforcedLabelsErrorMsg, strings.Join(lbsMissing, ","), tenantID, stream.Labels)
+					d.writeFailuresManager.Log(tenantID, err)
+					validationErrors.Add(err)
+					discardedBytes := util.EntriesTotalSize(stream.Entries)
+					d.validator.reportDiscardedDataWithTracker(ctx, validation.MissingEnforcedLabels, validationContext, lbs, retentionHours, policy, discardedBytes, len(stream.Entries))
+					continue
+				}
 			}
 
 			if block, statusCode, reason, err := d.validator.ShouldBlockIngestion(validationContext, now, policy); block {

pkg/distributor/distributor_test.go

@@ -507,6 +507,18 @@ func Test_PushWithEnforcedLabels(t *testing.T) {
 	// Metrics should remain unchanged
 	assert.Equal(t, float64(10000), testutil.ToFloat64(validation.DiscardedBytes))
 	assert.Equal(t, float64(100), testutil.ToFloat64(validation.DiscardedSamples))
+
+	// enforced labels are configured but the stream is an aggregated metric, so no errors.
+	limits.EnforcedLabels = []string{"app", "env"}
+	distributors, _ = prepare(t, 1, 3, limits, nil)
+
+	req = makeWriteRequestWithLabels(100, 100, []string{`{__aggregated_metric__="foo"}`}, false, false, false)
+	_, err = distributors[0].Push(ctx, req)
+	require.NoError(t, err)
+
+	// Metrics should remain unchanged
+	assert.Equal(t, float64(10000), testutil.ToFloat64(validation.DiscardedBytes))
+	assert.Equal(t, float64(100), testutil.ToFloat64(validation.DiscardedSamples))
 }
 
 func TestDistributorPushConcurrently(t *testing.T) {

pkg/distributor/validator.go

@@ -143,7 +143,11 @@ func (v Validator) ValidateEntry(ctx context.Context, vCtx validationContext, la
 	return nil
 }
 
-// Validate labels returns an error if the labels are invalid
+func (v Validator) IsAggregatedMetricStream(ls labels.Labels) bool {
+	return ls.Has(push.AggregatedMetricLabel)
+}
+
+// Validate labels returns an error if the labels are invalid and if the stream is an aggregated metric stream
 func (v Validator) ValidateLabels(vCtx validationContext, ls labels.Labels, stream logproto.Stream, retentionHours, policy string) error {
 	if len(ls) == 0 {
 		// TODO: is this one correct?
@@ -152,7 +156,7 @@ func (v Validator) ValidateLabels(vCtx validationContext, ls labels.Labels, stre
 	}
 
 	// Skip validation for aggregated metric streams, as we create those for internal use
-	if ls.Has(push.AggregatedMetricLabel) {
+	if v.IsAggregatedMetricStream(ls) {
 		return nil
 	}