Observability¶

TaskRunner exposes several layers of signal: structured logs, the Stats struct, the per-instance counters, the LongQueueHook periodic callback, and the Redis backend's own queue metrics. This page shows how to wire them into a production monitoring stack.

Signal layers¶

Layer	Source	What it tells you
Logs	`logrus` (structured JSON)	Lifecycle events: leader changes, worker tuning decisions, panics.
Error channel	`tr.ErrorChannel()`	Per-error infrastructure failures (ack, fetch, election).
Stats	`tr.GetTimingStatistics()`	On-demand snapshot: per-task avg, predicted wait, TPS.
LongQueueHook	`TaskRunnerConfig.LongQueueHook`	Periodic `Stats` callback on the aggregation cadence.
Counters	`inFlight`, `processed`, `fails`	Internal atomic counters (exposed via fields, not methods).
Redis metrics	`queue.GetMetrics(ctx)`	Stream length, PEL, claim-delay distribution, consumer idle.

1. Structured logging¶

TaskRunner logs via logrus, imported as log. To get JSON output (for any log aggregator), configure logrus once in your main:

import log "github.com/sirupsen/logrus"

func main() {
    log.SetFormatter(&log.JSONFormatter{
        DisableTimestamp: false,
        TimestampFormat:  time.RFC3339Nano,
    })
    log.SetLevel(log.InfoLevel)
    // ...
}

Key log lines to watch:

Event	Level	Fields
Worker pool tuned	`info`	`old_total`, `new_total`, `old_predicted_wait_ms`, `reason`
Fetcher panic	`error`	`fetcher_id`, `cause` (+ stack)
Worker panic	`error`	task name, error
Consumer cleanup	--	per-consumer deletions are logged on error only

2. Drain the error channel¶

go func() {
    for err := range tr.ErrorChannel() {
        log.WithError(err).Error("taskrunner.runtime")

        // Feed a counter -- alert on sustained error rates.
        if metricsErr := statsdClient.Increment("taskrunner.errors"); metricsErr != nil {
            log.WithError(metricsErr).Warn("failed to emit metric")
        }
    }
}()

Tag by error type using errors.As:

var execErr runner.TaskExecutionError
if errors.As(err, &execErr) {
    statsdClient.Increment("taskrunner.task_error",
        fmt.Sprintf("task:%s", execErr.GetTaskName()))
    continue
}
if errors.Is(err, contracts.ErrNoNewMessage) {
    continue // benign -- skip
}
statsdClient.Increment("taskrunner.infra_error")

3. Periodic stats via LongQueueHook¶

LongQueueHook fires on every aggregation tick (every LongQueueThreshold/2). It's the natural place to emit gauges:

tr := runner.NewTaskRunner(runner.TaskRunnerConfig{
    LongQueueThreshold: 15 * time.Second,
    LongQueueHook: func(s runner.Stats) {
        m := statsdClient
        m.Gauge("taskrunner.predicted_wait_ms", s.PredictedWaitTime, nil, 1)
        m.Gauge("taskrunner.tps_estimated", s.TPS, nil, 1)
        m.Gauge("taskrunner.avg_exec_ms", float64(s.AvgTiming.Milliseconds()), nil, 1)
        m.Gauge("taskrunner.task_kinds", float64(len(s.PerTaskTiming)), nil, 1)

        for task, avgMs := range s.PerTaskTiming {
            m.Gauge("taskrunner.task_avg_exec_ms", float64(avgMs),
                []string{fmt.Sprintf("task:%s", task)}, 1)
        }
    },
}, rdb, queue)

The hook fires on leaders and followers

Every instance runs the timing aggregator loop, so the hook fires on all of them. To avoid double-counting gauges, either emit from all instances (and average in your dashboard) or gate on tr.IsLeader() if you want a single emitter.

4. On-demand stats¶

Expose a lightweight HTTP endpoint:

http.HandleFunc("/internal/taskrunner/stats", func(w http.ResponseWriter, r *http.Request) {
    stats, err := tr.GetTimingStatistics()
    if err != nil {
        http.Error(w, err.Error(), http.StatusInternalServerError)
        return
    }
    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(stats)
})

GetTimingStatistics issues one HGET per registered task plus an XLEN, so keep the task count bounded (hundreds is fine; tens of thousands is not).

5. Redis backend metrics¶

The redisstream backend tracks per-message claim delays in a Redis ring buffer (capacity 1000) and exposes a richer snapshot:

import "github.com/soroosh-tanzadeh/taskrunner/redisstream"

q := redisstream.NewRedisStreamMessageQueueWithOptions(rdb, /*...*/)

m, err := q.GetMetrics(ctx)
// m is map[string]any with:
//   queue_size                  -> int64 (XLEN)
//   message_claim_delay_avg     -> float64 (ms), or -1 if <10 samples
//   message_claim_delay_std     -> float64
//   message_claim_delay_max     -> float64
//   message_claim_delay_last    -> float64
//   group_<name>                -> {pending_entries_count, lag}
//   (consumer info nested under group)

Poll this every 15-30s for a dashboard. message_claim_delay_* tells you how long messages wait before being claimed -- a rising average signals that workers can't keep up.

6. What to alert on¶

Metric	Source	Alert when
`PredictedWaitTime`	Stats	Sustained above `DesiredWaitTime + tolerance` for >2 ticks (tuner can't keep up -- raise `MaxWorkers` or add instances).
`message_claim_delay_avg`	`queue.GetMetrics`	Trending up over 5 min (workers saturated or stuck).
Error channel rate	`tr.ErrorChannel()`	Sustained non-zero rate (infra failure -- check Redis health).
`TaskExecutionError` rate	error channel (`errors.As`)	Per-task error spike (downstream service degraded).
Leadership churn	logs (`onDemote`/`onPromote`)	>2 changes/min (Redis instability or network partition).
Worker pool at `MaxWorkers`	logs (`worker pool tuned`)	Sustained (capacity ceiling -- raise `MaxWorkers`).
Pending entries (PEL)	`queue.GetMetrics`	Growing over time (stuck messages, ack failures).

7. Tracing context¶

TaskRunner deliberately uses context.Background() inside the worker hot path for the task action (see Architecture for why). If you need distributed tracing, propagate a trace ID inside the payload itself and start a span inside your Action:

Action: func(ctx context.Context, payload any) error {
    p := payload.(MyPayload)
    ctx, span := tracer.Start(ctx, "task."+p.TaskName,
        trace.WithAttributes(attribute.String("task.id", p.ID)))
    defer span.End()
    return doWork(ctx, p)
},

Don't try to thread the Start(ctx) cancel context through to the action -- it would cause in-flight tasks to abort on shutdown, breaking the visibility-timeout guarantee.