From 8ff6525b193398fab4c21fc7c33e1d1b670f350c Mon Sep 17 00:00:00 2001 From: Denis Ivanov Date: Tue, 27 Oct 2026 10:45:08 +0300 Subject: [PATCH 1/3] metrics: do not count postponed tasks as failed A task that hits a storage rate limit or cannot obtain the object lock returns dom.ErrRateLimitExceeded and is retried after a short delay; the worker error handler logs it at debug level, not as a failure. The worker middleware still counted it in worker_failed_tasks_total. CompleteMultipartUpload (and PutObject with ACL or tagging headers) enqueues object sync, object ACL sync and object tags sync tasks of the same object at once, and all of them take the same object lock, so some of them wait and are counted as failed. A burst of uploads (e.g. a database backup) produces hundreds of "failed" tasks per 15 minutes with no errors in the logs, which makes the metric useless for alerting. Signed-off-by: Denis Ivanov --- pkg/metrics/worker_middleware.go | 8 ++++- pkg/metrics/worker_middleware_test.go | 45 +++++++++++++++++++++++++++ 2 files changed, 52 insertions(+), 1 deletion(-) create mode 100644 pkg/metrics/worker_middleware_test.go diff --git a/pkg/metrics/worker_middleware.go b/pkg/metrics/worker_middleware.go index 517cf6dc..bf2933ea 100644 --- a/pkg/metrics/worker_middleware.go +++ b/pkg/metrics/worker_middleware.go @@ -18,10 +18,13 @@ package metrics import ( "context" + "errors" "github.com/hibiken/asynq" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promauto" + + "github.com/clyso/chorus/pkg/dom" ) var ( @@ -66,7 +69,10 @@ func WorkerMiddleware() asynq.MiddlewareFunc { err := next.ProcessTask(ctx, t) timer.ObserveDuration() inProgressGauge.WithLabelValues(queue, t.Type()).Dec() - if err != nil { + // A rate limit or a busy object lock postpones the task (dom.ErrRateLimitExceeded, + // logged at debug level by the worker error handler) — it is not a failure. + var rlErr *dom.ErrRateLimitExceeded + if err != nil && !errors.As(err, &rlErr) { failedCounter.WithLabelValues(queue, t.Type()).Inc() } processedCounter.WithLabelValues(queue, t.Type()).Inc() diff --git a/pkg/metrics/worker_middleware_test.go b/pkg/metrics/worker_middleware_test.go new file mode 100644 index 00000000..1e2d62cc --- /dev/null +++ b/pkg/metrics/worker_middleware_test.go @@ -0,0 +1,45 @@ +package metrics + +import ( + "context" + "errors" + "fmt" + "testing" + "time" + + "github.com/hibiken/asynq" + "github.com/prometheus/client_golang/prometheus" + dto "github.com/prometheus/client_model/go" + "github.com/stretchr/testify/require" + + "github.com/clyso/chorus/pkg/dom" +) + +func TestWorkerMiddleware_FailedTasks(t *testing.T) { + r := require.New(t) + taskType := "test:failed-tasks" + value := func(c *prometheus.CounterVec) float64 { + m := &dto.Metric{} + r.NoError(c.WithLabelValues("", taskType).Write(m)) + return m.GetCounter().GetValue() + } + failed := func() float64 { return value(failedCounter) } + processed := func() float64 { return value(processedCounter) } + run := func(err error) error { + h := WorkerMiddleware()(asynq.HandlerFunc(func(context.Context, *asynq.Task) error { return err })) + return h.ProcessTask(context.Background(), asynq.NewTask(taskType, nil)) + } + + r.NoError(run(nil)) + r.EqualValues(0, failed()) + + r.Error(run(errors.New("copy failed"))) + r.EqualValues(1, failed()) + + // postponed by a rate limit or a busy lock: not a failure + postponed := fmt.Errorf("%w: lock not obtained", &dom.ErrRateLimitExceeded{RetryIn: time.Second}) + r.ErrorIs(run(postponed), postponed) + r.EqualValues(1, failed()) + + r.EqualValues(3, processed()) +} From b6d4914537c068149420155ae87a937f3702e532 Mon Sep 17 00:00:00 2001 From: Denis Ivanov Date: Tue, 27 Oct 2026 15:41:59 +0300 Subject: [PATCH 2/3] metrics: do not count postponed tasks as processed A postponed task returns to the queue and is processed later, so counting each postponed attempt inflated worker_processed_tasks_total as well. The middleware now skips both counters for dom.ErrRateLimitExceeded; the metric descriptions in the chart README say so. Signed-off-by: Denis Ivanov --- deploy/chorus/README.md | 4 ++-- pkg/metrics/worker_middleware.go | 7 +++++-- pkg/metrics/worker_middleware_test.go | 7 ++++--- 3 files changed, 11 insertions(+), 7 deletions(-) diff --git a/deploy/chorus/README.md b/deploy/chorus/README.md index 739db1b9..81766c0f 100644 --- a/deploy/chorus/README.md +++ b/deploy/chorus/README.md @@ -120,8 +120,8 @@ Worker: | Metric | Type | Labels | Description | |--------|------|--------|-------------| -| `worker_processed_tasks_total` | counter | `queue`, `task_type` | Tasks processed (both successful and failed) | -| `worker_failed_tasks_total` | counter | `queue`, `task_type` | Tasks whose processing returned an error | +| `worker_processed_tasks_total` | counter | `queue`, `task_type` | Tasks processed (both successful and failed); tasks postponed by a rate limit or a busy object lock are not counted | +| `worker_failed_tasks_total` | counter | `queue`, `task_type` | Tasks whose processing returned an error, except tasks postponed by a rate limit or a busy object lock | | `worker_in_progress_tasks` | gauge | `queue`, `task_type` | Tasks currently being processed | | `worker_task_duration_seconds` | histogram | `queue`, `task_type` | Task processing time (exponential buckets, 0.1s–600s) | | `copy_in_progress_obj_bytes` | gauge | `flow`, `user`, `bucket` | Total size of objects currently being copied | diff --git a/pkg/metrics/worker_middleware.go b/pkg/metrics/worker_middleware.go index bf2933ea..a9267884 100644 --- a/pkg/metrics/worker_middleware.go +++ b/pkg/metrics/worker_middleware.go @@ -70,9 +70,12 @@ func WorkerMiddleware() asynq.MiddlewareFunc { timer.ObserveDuration() inProgressGauge.WithLabelValues(queue, t.Type()).Dec() // A rate limit or a busy object lock postpones the task (dom.ErrRateLimitExceeded, - // logged at debug level by the worker error handler) — it is not a failure. + // logged at debug level by the worker error handler): it is neither processed nor failed yet. var rlErr *dom.ErrRateLimitExceeded - if err != nil && !errors.As(err, &rlErr) { + if errors.As(err, &rlErr) { + return err + } + if err != nil { failedCounter.WithLabelValues(queue, t.Type()).Inc() } processedCounter.WithLabelValues(queue, t.Type()).Inc() diff --git a/pkg/metrics/worker_middleware_test.go b/pkg/metrics/worker_middleware_test.go index 1e2d62cc..b6bfd104 100644 --- a/pkg/metrics/worker_middleware_test.go +++ b/pkg/metrics/worker_middleware_test.go @@ -36,10 +36,11 @@ func TestWorkerMiddleware_FailedTasks(t *testing.T) { r.Error(run(errors.New("copy failed"))) r.EqualValues(1, failed()) - // postponed by a rate limit or a busy lock: not a failure + r.EqualValues(2, processed()) + + // postponed by a rate limit or a busy lock: neither processed nor failed, it will be retried postponed := fmt.Errorf("%w: lock not obtained", &dom.ErrRateLimitExceeded{RetryIn: time.Second}) r.ErrorIs(run(postponed), postponed) r.EqualValues(1, failed()) - - r.EqualValues(3, processed()) + r.EqualValues(2, processed()) } From a4db7c72c9d4d21eb6059b6ec395e59bc5990946 Mon Sep 17 00:00:00 2001 From: Denis Ivanov Date: Wed, 28 Oct 2026 08:31:19 +0300 Subject: [PATCH 3/3] go.mod: prometheus/client_model is a direct dependency The worker middleware test reads counter values with client_model, so go mod tidy moves it from indirect to direct requirements. Signed-off-by: Denis Ivanov --- go.mod | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/go.mod b/go.mod index 67f62914..bd2a2fa9 100644 --- a/go.mod +++ b/go.mod @@ -52,6 +52,7 @@ require ( github.com/moby/moby/api v1.55.0 github.com/onsi/ginkgo/v2 v2.32.1 github.com/onsi/gomega v1.42.1 + github.com/prometheus/client_model v0.6.2 github.com/testcontainers/testcontainers-go v0.44.0 golang.org/x/crypto v0.57.0 ) @@ -283,7 +284,6 @@ require ( github.com/philhofer/fwd v1.2.0 // indirect github.com/pkg/errors v0.9.1 // indirect github.com/power-devops/perfstat v0.0.0-20260805114148-88456608a4f6 // indirect - github.com/prometheus/client_model v0.6.2 // indirect github.com/prometheus/common v0.70.1 // indirect github.com/prometheus/procfs v0.21.1 // indirect github.com/prometheus/prom2json v1.5.0 // indirect