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/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 diff --git a/pkg/metrics/worker_middleware.go b/pkg/metrics/worker_middleware.go index 517cf6dc..a9267884 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,6 +69,12 @@ func WorkerMiddleware() asynq.MiddlewareFunc { err := next.ProcessTask(ctx, t) 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 neither processed nor failed yet. + var rlErr *dom.ErrRateLimitExceeded + if errors.As(err, &rlErr) { + return err + } if err != nil { failedCounter.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..b6bfd104 --- /dev/null +++ b/pkg/metrics/worker_middleware_test.go @@ -0,0 +1,46 @@ +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()) + + 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(2, processed()) +}