ratelimit: add decision attributes to request size (#876)

Adds decision and reason attributes to the request size metric to alllow
breaking down the request size metric by decision and reason.

Related to #874

Signed-off-by: Marc Lopez Rubio <[email protected]>

Author: marclop

processor/ratelimitprocessor/processor.go

@@ -220,25 +220,34 @@ func rateLimit(ctx context.Context,
 	hits int,
 	rateLimit func(ctx context.Context, n int) error,
 	metadataKeys []string,
-	telemetryBuilder *metadata.TelemetryBuilder,
+	tb *metadata.TelemetryBuilder,
 	logger *zap.Logger,
 	inflight *int64,
 ) error {
 	current := atomic.AddInt64(inflight, 1)
 	attrsCommon := getAttrsFromContext(ctx, metadataKeys)
-	telemetryBuilder.RatelimitConcurrentRequests.Record(ctx, current, metric.WithAttributes(attrsCommon...))
+	attrsSet := attribute.NewSet(attrsCommon...)
+	tb.RatelimitConcurrentRequests.Record(ctx, current,
+		metric.WithAttributeSet(attrsSet),
+	)
 
 	defer func(start time.Time) {
 		atomic.AddInt64(inflight, -1)
-		telemetryBuilder.RatelimitRequestDuration.Record(ctx, time.Since(start).Seconds(), metric.WithAttributes(attrsCommon...))
+		tb.RatelimitRequestDuration.Record(ctx, time.Since(start).Seconds(),
+			metric.WithAttributeSet(attrsSet),
+		)
 	}(time.Now())
 
 	err := rateLimit(ctx, hits)
+	attrRequests := getTelemetryAttrs(attrsCommon, err)
+	attrRequestsSet := attribute.NewSet(attrRequests...)
+	tb.RatelimitRequestSize.Record(ctx, int64(hits),
+		metric.WithAttributeSet(attrRequestsSet),
+	)
 	if err != nil {
 		// enhance error logging with metadata keys
-		fields := []zap.Field{
-			zap.Int("hits", hits),
-		}
+		fields := make([]zap.Field, 0, len(attrsCommon)+1)
+		fields = append(fields, zap.Int("hits", hits))
 		for _, kv := range attrsCommon {
 			switch kv.Value.Type() {
 			case attribute.STRINGSLICE:
@@ -250,26 +259,14 @@ func rateLimit(ctx context.Context,
 		logger.Error("request is over the limits defined by the rate limiter", append(fields, zap.Error(err))...)
 	}
 
-	attrRequests := getTelemetryAttrs(attrsCommon, err)
-	telemetryBuilder.RatelimitRequests.Add(ctx, 1, metric.WithAttributes(attrRequests...))
-
+	tb.RatelimitRequests.Add(ctx, 1, metric.WithAttributeSet(attrRequestsSet))
 	return err
 }
 
-func recordRequestSize(ctx context.Context, tb *metadata.TelemetryBuilder, strategy Strategy, hits int, metadataKeys []string) {
-	if tb != nil && strategy == StrategyRateLimitBytes {
-		attrsCommon := getAttrsFromContext(ctx, metadataKeys)
-		tb.RatelimitRequestSize.Record(ctx, int64(hits), metric.WithAttributes(attrsCommon...))
-	}
-}
-
 func (r *LogsRateLimiterProcessor) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
-	hits := r.count(ld)
-	recordRequestSize(ctx, r.telemetryBuilder, r.strategy, hits, r.metadataKeys)
-
 	if err := rateLimit(
 		ctx,
-		hits,
+		r.count(ld),
 		r.rl.RateLimit,
 		r.metadataKeys,
 		r.telemetryBuilder,
@@ -278,17 +275,13 @@ func (r *LogsRateLimiterProcessor) ConsumeLogs(ctx context.Context, ld plog.Logs
 	); err != nil {
 		return err
 	}
-
 	return r.next(ctx, ld)
 }
 
 func (r *MetricsRateLimiterProcessor) ConsumeMetrics(ctx context.Context, md pmetric.Metrics) error {
-	hits := r.count(md)
-	recordRequestSize(ctx, r.telemetryBuilder, r.strategy, hits, r.metadataKeys)
-
 	if err := rateLimit(
 		ctx,
-		hits,
+		r.count(md),
 		r.rl.RateLimit,
 		r.metadataKeys,
 		r.telemetryBuilder,
@@ -297,17 +290,13 @@ func (r *MetricsRateLimiterProcessor) ConsumeMetrics(ctx context.Context, md pme
 	); err != nil {
 		return err
 	}
-
 	return r.next(ctx, md)
 }
 
 func (r *TracesRateLimiterProcessor) ConsumeTraces(ctx context.Context, td ptrace.Traces) error {
-	hits := r.count(td)
-	recordRequestSize(ctx, r.telemetryBuilder, r.strategy, hits, r.metadataKeys)
-
 	if err := rateLimit(
 		ctx,
-		hits,
+		r.count(td),
 		r.rl.RateLimit,
 		r.metadataKeys,
 		r.telemetryBuilder,
@@ -316,17 +305,13 @@ func (r *TracesRateLimiterProcessor) ConsumeTraces(ctx context.Context, td ptrac
 	); err != nil {
 		return err
 	}
-
 	return r.next(ctx, td)
 }
 
 func (r *ProfilesRateLimiterProcessor) ConsumeProfiles(ctx context.Context, pd pprofile.Profiles) error {
-	hits := r.count(pd)
-	recordRequestSize(ctx, r.telemetryBuilder, r.strategy, hits, r.metadataKeys)
-
 	if err := rateLimit(
 		ctx,
-		hits,
+		r.count(pd),
 		r.rl.RateLimit,
 		r.metadataKeys,
 		r.telemetryBuilder,
@@ -335,14 +320,13 @@ func (r *ProfilesRateLimiterProcessor) ConsumeProfiles(ctx context.Context, pd p
 	); err != nil {
 		return err
 	}
-
 	return r.next(ctx, pd)
 }
 
 func getLogsCountFunc(strategy Strategy) func(ld plog.Logs) int {
 	switch strategy {
 	case StrategyRateLimitRequests:
-		return func(ld plog.Logs) int {
+		return func(plog.Logs) int {
 			return 1
 		}
 	case StrategyRateLimitRecords:
@@ -355,13 +339,13 @@ func getLogsCountFunc(strategy Strategy) func(ld plog.Logs) int {
 			return pm.LogsSize(ld)
 		}
 	}
-	return nil
+	return nil // cannot happen, prevented by config.Validate()
 }
 
 func getMetricsCountFunc(strategy Strategy) func(md pmetric.Metrics) int {
 	switch strategy {
 	case StrategyRateLimitRequests:
-		return func(md pmetric.Metrics) int {
+		return func(pmetric.Metrics) int {
 			return 1
 		}
 	case StrategyRateLimitRecords:
@@ -374,14 +358,13 @@ func getMetricsCountFunc(strategy Strategy) func(md pmetric.Metrics) int {
 			return pm.MetricsSize(md)
 		}
 	}
-	// cannot happen, prevented by config.Validate()
-	return nil
+	return nil // cannot happen, prevented by config.Validate()
 }
 
 func getTracesCountFunc(strategy Strategy) func(td ptrace.Traces) int {
 	switch strategy {
 	case StrategyRateLimitRequests:
-		return func(td ptrace.Traces) int {
+		return func(ptrace.Traces) int {
 			return 1
 		}
 	case StrategyRateLimitRecords:
@@ -394,14 +377,13 @@ func getTracesCountFunc(strategy Strategy) func(td ptrace.Traces) int {
 			return pm.TracesSize(td)
 		}
 	}
-	// cannot happen, prevented by config.Validate()
-	return nil
+	return nil // cannot happen, prevented by config.Validate()
 }
 
 func getProfilesCountFunc(strategy Strategy) func(pd pprofile.Profiles) int {
 	switch strategy {
 	case StrategyRateLimitRequests:
-		return func(pd pprofile.Profiles) int {
+		return func(pprofile.Profiles) int {
 			return 1
 		}
 	case StrategyRateLimitRecords:
@@ -414,6 +396,5 @@ func getProfilesCountFunc(strategy Strategy) func(pd pprofile.Profiles) int {
 			return pm.ProfilesSize(pd)
 		}
 	}
-	// cannot happen, prevented by config.Validate()
-	return nil
+	return nil // cannot happen, prevented by config.Validate()
 }