-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge branch 'ash2k/limiter' into 'master'
Rate limiter instrumentation Closes #154 See merge request https://gitlab.com/gitlab-org/cluster-integration/gitlab-agent/-/merge_requests/1149 Merged-by: Mikhail Mazurskiy <[email protected]> Approved-by: Taka Nishida <[email protected]>
- Loading branch information
Showing
6 changed files
with
245 additions
and
20 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,40 @@ | ||
| package metric | ||
|
|
||
| import ( | ||
| "context" | ||
|
|
||
| otelmetric "go.opentelemetry.io/otel/metric" | ||
| "go.opentelemetry.io/otel/trace" | ||
| ) | ||
|
|
||
| var ( | ||
| _ AllowLimiter = (*AllowLimiterInstrumentation)(nil) | ||
| ) | ||
|
|
||
| type AllowLimiter interface { | ||
| Allow(context.Context) bool | ||
| } | ||
|
|
||
| type AllowLimiterInstrumentation struct { | ||
| wrapper *LimiterWrapper | ||
| delegate AllowLimiter | ||
| } | ||
|
|
||
| func NewAllowLimiterInstrumentation(limiterName string, limit float64, limitUnit string, tr trace.Tracer, m otelmetric.Meter, delegate AllowLimiter) (*AllowLimiterInstrumentation, error) { | ||
| w, err := NewLimiterWrapper(limiterName, limit, limitUnit, m, tr) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return &AllowLimiterInstrumentation{ | ||
| wrapper: w, | ||
| delegate: delegate, | ||
| }, nil | ||
| } | ||
|
|
||
| func (i *AllowLimiterInstrumentation) Allow(ctx context.Context) (allowed bool) { | ||
| ctx, done := i.wrapper.Start(ctx) | ||
| defer func() { // to handle panics | ||
| done(allowed) | ||
| }() | ||
| return i.delegate.Allow(ctx) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,80 @@ | ||
| package metric | ||
|
|
||
| import ( | ||
| "context" | ||
| "time" | ||
|
|
||
| "go.opentelemetry.io/otel/attribute" | ||
| "go.opentelemetry.io/otel/codes" | ||
| otelmetric "go.opentelemetry.io/otel/metric" | ||
| "go.opentelemetry.io/otel/trace" | ||
| ) | ||
|
|
||
| const ( | ||
| rateLimiterBlockDurationName = "limiter_block_duration" | ||
| rateLimiterLimitName = "limiter_limit" | ||
| allowedAttr attribute.Key = "allowed" | ||
| limiterNameAttr attribute.Key = "limiter_name" | ||
| limiterLimitUnitAttr attribute.Key = "unit" | ||
| ) | ||
|
|
||
| type LimiterWrapper struct { | ||
| limiterName string | ||
| tr trace.Tracer | ||
| hist otelmetric.Float64Histogram | ||
| } | ||
|
|
||
| func NewLimiterWrapper(limiterName string, limit float64, limitUnit string, m otelmetric.Meter, tr trace.Tracer) (*LimiterWrapper, error) { | ||
| limitAttrs := otelmetric.WithAttributeSet(attribute.NewSet( | ||
| limiterNameAttr.String(limiterName), | ||
| limiterLimitUnitAttr.String(limitUnit), | ||
| )) // allocate once | ||
| _, err := m.Float64ObservableGauge( | ||
| rateLimiterLimitName, | ||
| otelmetric.WithDescription("Limit for the rate limiter"), | ||
| otelmetric.WithUnit(limitUnit), | ||
| otelmetric.WithFloat64Callback(func(ctx context.Context, observer otelmetric.Float64Observer) error { | ||
| observer.Observe(limit, limitAttrs) | ||
| return nil | ||
| }), | ||
| ) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| hist, err := m.Float64Histogram( | ||
| rateLimiterBlockDurationName, | ||
| // TODO switch to "seconds" once API to set histogram's buckets is available | ||
| // See https://github.com/open-telemetry/opentelemetry-go/issues/4094 | ||
| otelmetric.WithUnit("ms"), | ||
| otelmetric.WithDescription("Duration the rate limiter blocked for deciding to allow/block the call"), | ||
| ) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return &LimiterWrapper{ | ||
| limiterName: limiterName, | ||
| tr: tr, | ||
| hist: hist, | ||
| }, nil | ||
| } | ||
|
|
||
| func (w *LimiterWrapper) Start(ctx context.Context) (context.Context, func(allowed bool)) { | ||
| start := time.Now() | ||
| ctx, span := w.tr.Start(ctx, "limiter", trace.WithSpanKind(trace.SpanKindInternal)) | ||
| return ctx, func(allowed bool) { | ||
| // TODO switch to "seconds" once API to set histogram's buckets is available | ||
| // See https://github.com/open-telemetry/opentelemetry-go/issues/4094 | ||
| duration := float64(time.Since(start)) / float64(time.Millisecond) | ||
| // Pass background context because we always want to record the duration. | ||
| w.hist.Record(context.Background(), duration, otelmetric.WithAttributeSet(attribute.NewSet( | ||
| allowedAttr.Bool(allowed), | ||
| limiterNameAttr.String(w.limiterName), | ||
| ))) | ||
| if allowed { | ||
| span.SetStatus(codes.Ok, "") | ||
| } else { | ||
| span.SetStatus(codes.Error, "") | ||
| } | ||
| span.End() | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,43 @@ | ||
| package metric | ||
|
|
||
| import ( | ||
| "context" | ||
|
|
||
| otelmetric "go.opentelemetry.io/otel/metric" | ||
| "go.opentelemetry.io/otel/trace" | ||
| ) | ||
|
|
||
| var ( | ||
| _ WaitLimiter = (*WaitLimiterInstrumentation)(nil) | ||
| ) | ||
|
|
||
| type WaitLimiter interface { | ||
| Wait(context.Context) error | ||
| } | ||
|
|
||
| type WaitLimiterInstrumentation struct { | ||
| wrapper *LimiterWrapper | ||
| delegate WaitLimiter | ||
| } | ||
|
|
||
| func NewWaitLimiterInstrumentation(limiterName string, limit float64, limitUnit string, tr trace.Tracer, m otelmetric.Meter, delegate WaitLimiter) (*WaitLimiterInstrumentation, error) { | ||
| w, err := NewLimiterWrapper(limiterName, limit, limitUnit, m, tr) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return &WaitLimiterInstrumentation{ | ||
| wrapper: w, | ||
| delegate: delegate, | ||
| }, nil | ||
| } | ||
|
|
||
| func (i *WaitLimiterInstrumentation) Wait(ctx context.Context) error { | ||
| ctx, done := i.wrapper.Start(ctx) | ||
| allowed := false | ||
| defer func() { // to handle panics | ||
| done(allowed) | ||
| }() | ||
| err := i.delegate.Wait(ctx) | ||
| allowed = err == nil | ||
| return err | ||
| } |