abstract the metadata function and use WithErrorReason

Signed-off-by: Andreas Gkizas <andreas.gkizas@elastic.co>

Author: gizas

processor/ratelimitprocessor/gubernator.go

@@ -97,8 +97,8 @@ func (r *gubernatorRateLimiter) Shutdown(ctx context.Context) error {
 }
 
 func (r *gubernatorRateLimiter) RateLimit(ctx context.Context, hits int) error {
-	projectID := getProjectIDFromMetadata(ctx)
 	uniqueKey := getUniqueKey(ctx, r.cfg.MetadataKeys)
+
 	createdAt := time.Now().UnixMilli()
 	getRateLimitsResp, err := r.client.GetRateLimits(ctx, &gubernator.GetRateLimitsReq{
 		Requests: []*gubernator.RateLimitReq{{
@@ -115,32 +115,35 @@ func (r *gubernatorRateLimiter) RateLimit(ctx context.Context, hits int) error {
 	})
 	if err != nil {
 		r.set.Logger.Error("error executing gubernator rate limit request", zap.Error(err))
-		r.telemetryBuilder.RatelimitRequests.Add(ctx, 1, metric.WithAttributeSet(attribute.NewSet(
-			telemetry.WithProjectID(projectID),
-			telemetry.WithLimitReason("request_error"),
+		attrs := []attribute.KeyValue{
+			telemetry.WithErrorReason(telemetry.BadRequest),
 			telemetry.WithDecision("rejected"),
-		)))
+		}
+		attrs = attrsFromMetadata(ctx, r.cfg.MetadataKeys, attrs)
+		r.telemetryBuilder.RatelimitRequests.Add(ctx, 1, metric.WithAttributeSet(attribute.NewSet(attrs...)))
 		return errRateLimitInternalError
 	}
 
 	// Inside the gRPC response, we should have a single-item list of responses.
 	responses := getRateLimitsResp.GetResponses()
 	if n := len(responses); n != 1 {
-		r.telemetryBuilder.RatelimitRequests.Add(ctx, 1, metric.WithAttributeSet(attribute.NewSet(
-			telemetry.WithProjectID(projectID),
-			telemetry.WithLimitReason("request_error"),
+		attrs := []attribute.KeyValue{
+			telemetry.WithErrorReason(telemetry.Invalid),
 			telemetry.WithDecision("accepted"),
-		)))
+		}
+		attrs = attrsFromMetadata(ctx, r.cfg.MetadataKeys, attrs)
+		r.telemetryBuilder.RatelimitRequests.Add(ctx, 1, metric.WithAttributeSet(attribute.NewSet(attrs...)))
 		return fmt.Errorf("expected 1 response from gubernator, got %d", n)
 	}
 	resp := responses[0]
 	if resp.GetError() != "" {
 		r.set.Logger.Error("failed to get response from gubernator", zap.Error(errors.New(resp.GetError())))
-		r.telemetryBuilder.RatelimitRequests.Add(ctx, 1, metric.WithAttributeSet(attribute.NewSet(
-			telemetry.WithProjectID(projectID),
-			telemetry.WithLimitReason("limit_error"),
+		attrs := []attribute.KeyValue{
+			telemetry.WithErrorReason(telemetry.ServerError),
 			telemetry.WithDecision("rejected"),
-		)))
+		}
+		attrs = attrsFromMetadata(ctx, r.cfg.MetadataKeys, attrs)
+		r.telemetryBuilder.RatelimitRequests.Add(ctx, 1, metric.WithAttributeSet(attribute.NewSet(attrs...)))
 		return errRateLimitInternalError
 	}
 
@@ -154,28 +157,36 @@ func (r *gubernatorRateLimiter) RateLimit(ctx context.Context, hits int) error {
 				zap.String("processor_id", r.set.ID.String()),
 				zap.Strings("metadata_keys", r.cfg.MetadataKeys),
 			)
-			r.telemetryBuilder.RatelimitRequests.Add(ctx, 1, metric.WithAttributeSet(attribute.NewSet(
-				telemetry.WithProjectID(projectID),
-				telemetry.WithLimitReason("throttled"),
-				telemetry.WithDecision("accepted"),
-			)))
+			attrs := []attribute.KeyValue{
+				telemetry.WithErrorReason(telemetry.StatusOverLimit),
+				telemetry.WithDecision("rejected"),
+			}
+			attrs = attrsFromMetadata(ctx, r.cfg.MetadataKeys, attrs)
+			r.telemetryBuilder.RatelimitRequests.Add(ctx, 1, metric.WithAttributeSet(attribute.NewSet(attrs...)))
 			return errTooManyRequests
 		case ThrottleBehaviorDelay:
 			delay := time.Duration(resp.GetResetTime()-createdAt) * time.Millisecond
 			timer := time.NewTimer(delay)
 			defer timer.Stop()
 			select {
 			case <-ctx.Done():
+				attrs := []attribute.KeyValue{
+					telemetry.WithErrorReason(telemetry.StatusOverLimit),
+					telemetry.WithDecision("accepted"),
+				}
+				attrs = attrsFromMetadata(ctx, r.cfg.MetadataKeys, attrs)
+				r.telemetryBuilder.RatelimitRequests.Add(ctx, 1, metric.WithAttributeSet(attribute.NewSet(attrs...)))
 				return ctx.Err()
 			case <-timer.C:
 			}
 		}
 	}
 
-	r.telemetryBuilder.RatelimitRequests.Add(ctx, 1, metric.WithAttributeSet(attribute.NewSet(
-		telemetry.WithProjectID(projectID),
-		telemetry.WithLimitReason("under_limit"),
+	attrs := []attribute.KeyValue{
+		telemetry.WithErrorReason(telemetry.StatusUnderLimit),
 		telemetry.WithDecision("accepted"),
-	)))
+	}
+	attrs = attrsFromMetadata(ctx, r.cfg.MetadataKeys, attrs)
+	r.telemetryBuilder.RatelimitRequests.Add(ctx, 1, metric.WithAttributeSet(attribute.NewSet(attrs...)))
 	return nil
 }

processor/ratelimitprocessor/gubernator_test.go

@@ -211,7 +211,12 @@ func TestGubernatorRateLimiter_RateLimit_Meterprovider(t *testing.T) {
 	mp := metric.NewMeterProvider(metric.WithReader(reader))
 	otel.SetMeterProvider(mp)
 
-	cfg := createDefaultConfig().(*Config)
+	cfg := &Config{
+		Rate:             1,
+		Burst:            2,
+		MetadataKeys:     []string{"x-elastic-project-id"},
+		ThrottleBehavior: ThrottleBehaviorError,
+	}
 	server, rl := newTestGubernatorRateLimiter(t, cfg, mp)
 
 	//We provide the x-elastic-project-id in oder to set the ProjectID

processor/ratelimitprocessor/internal/telemetry/attributes.go

@@ -31,12 +31,13 @@ const (
 	outcomeKey     = "outcome"
 	errorReasonKey = "error_reason"
 	descisionKey   = "ratelimit_decision"
-	limitReasonKey = "reason"
 
-	TooLarge   ErrorReason = "too_large"
-	BadRequest ErrorReason = "bad_request"
-	Invalid    ErrorReason = "invalid"
-	TooMany    ErrorReason = "too_many"
+	TooLarge         ErrorReason = "too_large"
+	BadRequest       ErrorReason = "bad_request"
+	Invalid          ErrorReason = "invalid"
+	TooMany          ErrorReason = "too_many"
+	StatusUnderLimit ErrorReason = "underl_limit"
+	StatusOverLimit  ErrorReason = "throttled"
 
 	SignalTrace  = "trace"
 	SignalMetric = "metric"
@@ -56,11 +57,6 @@ func WithDecision(decision string) attribute.KeyValue {
 	return attribute.String(descisionKey, decision)
 }
 
-// WithLimitReason returns limitReason attribute with key.
-func WithLimitReason(limitReason string) attribute.KeyValue {
-	return attribute.String(limitReasonKey, limitReason)
-}
-
 // WithProtocol returns a protocol attribute with key.
 func WithProtocol(protocol string) attribute.KeyValue {
 	return attribute.String(protocolKey, protocol)

processor/ratelimitprocessor/ratelimiter.go

@@ -23,6 +23,7 @@ import (
 	"strings"
 
 	"go.opentelemetry.io/collector/client"
+	"go.opentelemetry.io/otel/attribute"
 )
 
 var (
@@ -56,13 +57,17 @@ const (
 	ProjectID = "x-elastic-project-id"
 )
 
-func getProjectIDFromMetadata(ctx context.Context) string {
+func attrsFromMetadata(ctx context.Context, metadataKeys []string, attrs []attribute.KeyValue) []attribute.KeyValue {
 	clientInfo := client.FromContext(ctx)
-	values := clientInfo.Metadata.Get(ProjectID)
-	if len(values) > 0 {
-		return values[0]
+
+	for _, key := range metadataKeys {
+		values := clientInfo.Metadata.Get(key)
+		if len(values) > 0 {
+			attrs = append(attrs, attribute.String(key, strings.Join(values, ",")))
+		}
 	}
-	return ""
+
+	return attrs
 }
 
 func getUniqueKey(ctx context.Context, metadataKeys []string) string {