[ratelimit] Add concurrent_requests and request_duration telemetry metrics (#588)

This PR is to add two more telemetry metrics for rate limit processor:
otelcol_ratelimit.concurrent_requests: records the number of requests that are being processed by the rate limiter at any given moment.
otelcol_ratelimit.request_duration: Measures the duration of time taken to process each rate limiting request.

Author: kaiyan-sheng

processor/ratelimitprocessor/documentation.md

@@ -6,6 +6,22 @@
 
 The following telemetry is emitted by this component.
 
+### otelcol_ratelimit.concurrent_requests
+
+Number of in-flight requests at any given time
+
+| Unit | Metric Type | Value Type |
+| ---- | ----------- | ---------- |
+| {requests} | Gauge | Int |
+
+### otelcol_ratelimit.request_duration
+
+Time(in seconds) taken to process a rate limit request
+
+| Unit | Metric Type | Value Type |
+| ---- | ----------- | ---------- |
+| {seconds} | Histogram | Double |
+
 ### otelcol_ratelimit.requests
 
 Number of rate-limiting requests

processor/ratelimitprocessor/factory.go

@@ -75,13 +75,16 @@ func createLogsProcessor(
 	if err != nil {
 		return nil, err
 	}
+
+	var inflight int64
 	return NewLogsRateLimiterProcessor(
 		rateLimiter,
 		set.TelemetrySettings,
 		config.Strategy,
 		func(ctx context.Context, ld plog.Logs) error {
 			return nextConsumer.ConsumeLogs(ctx, ld)
 		},
+		&inflight,
 	)
 }
 
@@ -96,13 +99,15 @@ func createMetricsProcessor(
 	if err != nil {
 		return nil, err
 	}
+	var inflight int64
 	return NewMetricsRateLimiterProcessor(
 		rateLimiter,
 		set.TelemetrySettings,
 		config.Strategy,
 		func(ctx context.Context, md pmetric.Metrics) error {
 			return nextConsumer.ConsumeMetrics(ctx, md)
 		},
+		&inflight,
 	)
 }
 
@@ -117,13 +122,15 @@ func createTracesProcessor(
 	if err != nil {
 		return nil, err
 	}
+	var inflight int64
 	return NewTracesRateLimiterProcessor(
 		rateLimiter,
 		set.TelemetrySettings,
 		config.Strategy,
 		func(ctx context.Context, td ptrace.Traces) error {
 			return nextConsumer.ConsumeTraces(ctx, td)
 		},
+		&inflight,
 	)
 }
 
@@ -138,12 +145,14 @@ func createProfilesProcessor(
 	if err != nil {
 		return nil, err
 	}
+	var inflight int64
 	return NewProfilesRateLimiterProcessor(
 		rateLimiter,
 		set.TelemetrySettings,
 		config.Strategy,
 		func(ctx context.Context, td pprofile.Profiles) error {
 			return nextConsumer.ConsumeProfiles(ctx, td)
 		},
+		&inflight,
 	)
 }

processor/ratelimitprocessor/internal/metadata/generated_telemetry.go

@@ -21,7 +21,21 @@ telemetry:
         value_type: int
         monotonic: true
       attributes: ["decision", "reason"]
-
+    ratelimit.request_duration:
+      enabled: true
+      description: Time(in seconds) taken to process a rate limit request
+      unit: "{seconds}"
+      histogram:
+        value_type: double
+        monotonic: true
+        bucket_boundaries: [ 0.0001, 0.0005, 0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.5, 1.0, 5.0, 10.0 ]
+    ratelimit.concurrent_requests:
+      enabled: true
+      description: Number of in-flight requests at any given time
+      unit: "{requests}"
+      gauge:
+        value_type: int
+        monotonic: true
 attributes:
   decision:
     description: rate limit decision

processor/ratelimitprocessor/internal/metadatatest/generated_telemetrytest.go

@@ -21,9 +21,9 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sync/atomic"
+	"time"
 
-	"github.com/elastic/opentelemetry-collector-components/processor/ratelimitprocessor/internal/metadata"
-	"github.com/elastic/opentelemetry-collector-components/processor/ratelimitprocessor/internal/telemetry"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/pdata/plog"
@@ -34,13 +34,16 @@ import (
 	"go.opentelemetry.io/otel/metric"
 
 	"github.com/elastic/opentelemetry-collector-components/internal/sharedcomponent"
+	"github.com/elastic/opentelemetry-collector-components/processor/ratelimitprocessor/internal/metadata"
+	"github.com/elastic/opentelemetry-collector-components/processor/ratelimitprocessor/internal/telemetry"
 )
 
 type rateLimiterProcessor struct {
 	component.Component
 	rl               RateLimiter
 	metadataKeys     []string
 	telemetryBuilder *metadata.TelemetryBuilder
+	inflight         *int64
 }
 
 type LogsRateLimiterProcessor struct {
@@ -72,6 +75,7 @@ func NewLogsRateLimiterProcessor(
 	telemetrySettings component.TelemetrySettings,
 	strategy Strategy,
 	next func(ctx context.Context, logs plog.Logs) error,
+	inflight *int64,
 ) (*LogsRateLimiterProcessor, error) {
 	telemetryBuilder, err := metadata.NewTelemetryBuilder(telemetrySettings)
 	if err != nil {
@@ -83,6 +87,7 @@ func NewLogsRateLimiterProcessor(
 			Component:        rateLimiter,
 			rl:               rateLimiter.Unwrap(),
 			telemetryBuilder: telemetryBuilder,
+			inflight:         inflight,
 		},
 		count: getLogsCountFunc(strategy),
 		next:  next,
@@ -94,6 +99,7 @@ func NewMetricsRateLimiterProcessor(
 	telemetrySettings component.TelemetrySettings,
 	strategy Strategy,
 	next func(ctx context.Context, metrics pmetric.Metrics) error,
+	inflight *int64, // used to calculate concurrent requests
 ) (*MetricsRateLimiterProcessor, error) {
 	telemetryBuilder, err := metadata.NewTelemetryBuilder(telemetrySettings)
 	if err != nil {
@@ -105,6 +111,7 @@ func NewMetricsRateLimiterProcessor(
 			Component:        rateLimiter,
 			rl:               rateLimiter.Unwrap(),
 			telemetryBuilder: telemetryBuilder,
+			inflight:         inflight,
 		},
 		count: getMetricsCountFunc(strategy),
 		next:  next,
@@ -116,6 +123,7 @@ func NewTracesRateLimiterProcessor(
 	telemetrySettings component.TelemetrySettings,
 	strategy Strategy,
 	next func(ctx context.Context, traces ptrace.Traces) error,
+	inflight *int64,
 ) (*TracesRateLimiterProcessor, error) {
 	telemetryBuilder, err := metadata.NewTelemetryBuilder(telemetrySettings)
 	if err != nil {
@@ -127,6 +135,7 @@ func NewTracesRateLimiterProcessor(
 			Component:        rateLimiter,
 			rl:               rateLimiter.Unwrap(),
 			telemetryBuilder: telemetryBuilder,
+			inflight:         inflight,
 		},
 		count: getTracesCountFunc(strategy),
 		next:  next,
@@ -138,6 +147,7 @@ func NewProfilesRateLimiterProcessor(
 	telemetrySettings component.TelemetrySettings,
 	strategy Strategy,
 	next func(ctx context.Context, profiles pprofile.Profiles) error,
+	inflight *int64,
 ) (*ProfilesRateLimiterProcessor, error) {
 	telemetryBuilder, err := metadata.NewTelemetryBuilder(telemetrySettings)
 	if err != nil {
@@ -149,6 +159,7 @@ func NewProfilesRateLimiterProcessor(
 			Component:        rateLimiter,
 			rl:               rateLimiter.Unwrap(),
 			telemetryBuilder: telemetryBuilder,
+			inflight:         inflight,
 		},
 		count: getProfilesCountFunc(strategy),
 		next:  next,
@@ -200,7 +211,16 @@ func rateLimit(
 	rateLimit func(ctx context.Context, n int) error,
 	metadataKeys []string,
 	telemetryBuilder *metadata.TelemetryBuilder,
+	inflight *int64,
 ) error {
+	current := atomic.AddInt64(inflight, 1)
+	telemetryBuilder.RatelimitConcurrentRequests.Record(ctx, current)
+
+	defer func(start time.Time) {
+		atomic.AddInt64(inflight, -1)
+		telemetryBuilder.RatelimitRequestDuration.Record(ctx, time.Since(start).Seconds())
+	}(time.Now())
+
 	err := rateLimit(ctx, hits)
 
 	attrs := getTelemetryAttrs(ctx, metadataKeys, err)
@@ -218,6 +238,7 @@ func (r *LogsRateLimiterProcessor) ConsumeLogs(ctx context.Context, ld plog.Logs
 		r.rl.RateLimit,
 		r.metadataKeys,
 		r.telemetryBuilder,
+		r.inflight,
 	); err != nil {
 		return err
 	}
@@ -234,6 +255,7 @@ func (r *MetricsRateLimiterProcessor) ConsumeMetrics(ctx context.Context, md pme
 		r.rl.RateLimit,
 		r.metadataKeys,
 		r.telemetryBuilder,
+		r.inflight,
 	); err != nil {
 		return err
 	}
@@ -250,6 +272,7 @@ func (r *TracesRateLimiterProcessor) ConsumeTraces(ctx context.Context, td ptrac
 		r.rl.RateLimit,
 		r.metadataKeys,
 		r.telemetryBuilder,
+		r.inflight,
 	); err != nil {
 		return err
 	}
@@ -266,6 +289,7 @@ func (r *ProfilesRateLimiterProcessor) ConsumeProfiles(ctx context.Context, pd p
 		r.rl.RateLimit,
 		r.metadataKeys,
 		r.telemetryBuilder,
+		r.inflight,
 	); err != nil {
 		return err
 	}