-
Notifications
You must be signed in to change notification settings - Fork 259
Add lifecycle endpoint to clear metrics #714
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: master
Are you sure you want to change the base?
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,70 @@ | ||
| // Copyright 2026 The Prometheus Authors | ||
| // Licensed under the Apache License, Version 2.0 (the "License"); | ||
| // you may not use this file except in compliance with the License. | ||
| // You may obtain a copy of the License at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, software | ||
| // distributed under the License is distributed on an "AS IS" BASIS, | ||
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| // See the License for the specific language governing permissions and | ||
| // limitations under the License. | ||
|
|
||
| package main | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "net/http" | ||
| "net/http/httptest" | ||
| "strings" | ||
| "testing" | ||
|
|
||
| "github.com/prometheus/common/promslog" | ||
| ) | ||
|
|
||
| type fakeMetricsClearer struct { | ||
| cleared int | ||
| called int | ||
| err error | ||
| } | ||
|
|
||
| func (f *fakeMetricsClearer) ClearMetrics(_ context.Context) (int, error) { | ||
| f.called++ | ||
| return f.cleared, f.err | ||
| } | ||
|
|
||
| func TestClearMetricsHandler(t *testing.T) { | ||
| clearer := &fakeMetricsClearer{cleared: 3} | ||
| handler := clearMetricsHandler(clearer, promslog.NewNopLogger()) | ||
|
|
||
| request := httptest.NewRequest(http.MethodPost, "/-/clear", nil) | ||
| response := httptest.NewRecorder() | ||
|
|
||
| handler.ServeHTTP(response, request) | ||
|
|
||
| if clearer.called != 1 { | ||
| t.Fatalf("expected clearer to be called once, got %d", clearer.called) | ||
| } | ||
| if body := response.Body.String(); !strings.Contains(body, "Cleared 3 metric series") { | ||
| t.Fatalf("unexpected response body: %q", body) | ||
| } | ||
| } | ||
|
|
||
| func TestClearMetricsHandlerError(t *testing.T) { | ||
| clearer := &fakeMetricsClearer{err: errors.New("clear failed")} | ||
| handler := clearMetricsHandler(clearer, promslog.NewNopLogger()) | ||
|
|
||
| request := httptest.NewRequest(http.MethodPost, "/-/clear", nil) | ||
| response := httptest.NewRecorder() | ||
|
|
||
| handler.ServeHTTP(response, request) | ||
|
|
||
| if response.Code != http.StatusServiceUnavailable { | ||
| t.Fatalf("expected status %d, got %d", http.StatusServiceUnavailable, response.Code) | ||
| } | ||
| if body := response.Body.String(); !strings.Contains(body, "clear failed") { | ||
| t.Fatalf("unexpected response body: %q", body) | ||
| } | ||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -14,8 +14,11 @@ | |
| package exporter | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "log/slog" | ||
| "os" | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/prometheus/client_golang/prometheus" | ||
|
|
@@ -39,6 +42,26 @@ type Registry interface { | |
| RemoveStaleMetrics() | ||
| } | ||
|
|
||
| var ( | ||
| // ErrClearMetricsUnsupported indicates that the configured registry cannot clear metrics. | ||
| ErrClearMetricsUnsupported = errors.New("registry does not support clearing metrics") | ||
| // ErrExporterNotRunning indicates that the exporter event loop is not running. | ||
| ErrExporterNotRunning = errors.New("exporter listener is not running") | ||
| ) | ||
|
|
||
| type clearableRegistry interface { | ||
| ClearMetrics() int | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Adding Suggestion: Consider keeping the exported
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Done. I restored the exported Clearing is now handled through an internal optional |
||
| } | ||
|
|
||
| type clearMetricsRequest struct { | ||
| result chan clearMetricsResult | ||
| } | ||
|
|
||
| type clearMetricsResult struct { | ||
| cleared int | ||
| err error | ||
| } | ||
|
|
||
| type Exporter struct { | ||
| Mapper *mapper.MetricMapper | ||
| Registry Registry | ||
|
|
@@ -49,21 +72,40 @@ type Exporter struct { | |
| EventStats *prometheus.CounterVec | ||
| ConflictingEventStats *prometheus.CounterVec | ||
| MetricsCount *prometheus.GaugeVec | ||
| clearMetrics chan clearMetricsRequest | ||
| listenStateMtx sync.RWMutex | ||
| listening bool | ||
| listenDone chan struct{} | ||
| } | ||
|
|
||
| // Listen handles all events sent to the given channel sequentially. It | ||
| // terminates when the channel is closed. | ||
| func (b *Exporter) Listen(e <-chan event.Events) { | ||
| removeStaleMetricsTicker := clock.NewTicker(time.Second) | ||
| defer removeStaleMetricsTicker.Stop() | ||
|
|
||
| b.listenStateMtx.Lock() | ||
| b.listening = true | ||
| b.listenDone = make(chan struct{}) | ||
| done := b.listenDone | ||
| b.listenStateMtx.Unlock() | ||
|
|
||
| defer func() { | ||
| b.listenStateMtx.Lock() | ||
| b.listening = false | ||
| close(done) | ||
| b.listenStateMtx.Unlock() | ||
| }() | ||
|
|
||
| for { | ||
| select { | ||
| case <-removeStaleMetricsTicker.C: | ||
| b.Registry.RemoveStaleMetrics() | ||
| case request := <-b.clearMetrics: | ||
| request.result <- b.clearRegistryMetrics() | ||
| case events, ok := <-e: | ||
| if !ok { | ||
| b.Logger.Debug("Channel is closed. Break out of Exporter.Listener.") | ||
| removeStaleMetricsTicker.Stop() | ||
| return | ||
| } | ||
| for _, event := range events { | ||
|
|
@@ -73,6 +115,51 @@ func (b *Exporter) Listen(e <-chan event.Events) { | |
| } | ||
| } | ||
|
|
||
| // ClearMetrics clears all dynamically registered StatsD time series. | ||
| func (b *Exporter) ClearMetrics(ctx context.Context) (int, error) { | ||
| if ctx == nil { | ||
| ctx = context.Background() | ||
| } | ||
|
|
||
| done, ok := b.listenState() | ||
| if !ok { | ||
| return 0, ErrExporterNotRunning | ||
| } | ||
|
|
||
| request := clearMetricsRequest{ | ||
| result: make(chan clearMetricsResult, 1), | ||
| } | ||
|
|
||
| select { | ||
| case b.clearMetrics <- request: | ||
| case <-done: | ||
| return 0, ErrExporterNotRunning | ||
| case <-ctx.Done(): | ||
| return 0, ctx.Err() | ||
| } | ||
|
|
||
| select { | ||
| case result := <-request.result: | ||
| return result.cleared, result.err | ||
| case <-ctx.Done(): | ||
| return 0, ctx.Err() | ||
| } | ||
| } | ||
|
|
||
| func (b *Exporter) listenState() (<-chan struct{}, bool) { | ||
| b.listenStateMtx.RLock() | ||
| defer b.listenStateMtx.RUnlock() | ||
| return b.listenDone, b.listening | ||
| } | ||
|
|
||
| func (b *Exporter) clearRegistryMetrics() clearMetricsResult { | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I wonder if we should instrument this, for large metric sets, we could block for a while and cause the whole exporter to get "stuck" here. Maybe lets add at least a doc block saying that this function may cause "hangs" when working with larger sets. |
||
| clearable, ok := b.Registry.(clearableRegistry) | ||
| if !ok { | ||
| return clearMetricsResult{err: ErrClearMetricsUnsupported} | ||
| } | ||
| return clearMetricsResult{cleared: clearable.ClearMetrics()} | ||
| } | ||
|
|
||
| // handleEvent processes a single Event according to the configured mapping. | ||
| func (b *Exporter) handleEvent(thisEvent event.Event) { | ||
| mapping, labels, present := b.Mapper.GetMapping(thisEvent.MetricName(), thisEvent.MetricType()) | ||
|
|
@@ -207,5 +294,6 @@ func NewExporter(reg prometheus.Registerer, mapper *mapper.MetricMapper, logger | |
| EventStats: eventStats, | ||
| ConflictingEventStats: conflictingEventStats, | ||
| MetricsCount: metricsCount, | ||
| clearMetrics: make(chan clearMetricsRequest), | ||
| } | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Lets return 405 Method Not Allowed when the method is not POST|PUT.
Allow: PUT, POST, and test both supported methods and at least one unsupported method.