Monitoraggio e Allarmi

Si consiglia di utilizzare strumenti di monitoraggio come Prometheus per monitorare i processi e le code dei lavori nell’ambiente di produzione.

Metriche della Coda

Se si utilizza Web UI, è possibile abilitare l’integrazione con Prometheus fornendo due parametri:

• --enable-metrics-exporter: Abilita la raccolta delle metriche della coda e le esporta all’endpoint /metrics.
• --prometheus-addr: Abilita la visualizzazione delle metriche della coda all’interno dell’interfaccia utente Web.

La pagina delle metriche della coda ha questo aspetto:

Se non si utilizza l’interfaccia utente Web, Asynq viene fornito con un file binario che è possibile eseguire per esportare le metriche della coda. Include anche un package x/metrics per la raccolta delle metriche della coda.

Metriche dei Processi dei Lavori

L’interfaccia Handler di Asynq e ServeMux possono essere strumentate con metriche per l’osservabilità.

Ecco un esempio di esportazione delle metriche dei processi dei lavori utilizzando Prometheus. Possiamo strumentare il nostro codice all’interno dell’applicazione per tenere traccia di metriche specifiche dell’applicazione, nonché di metriche predefinite (come memoria e CPU) tracciate da Prometheus.

Ecco un elenco di metriche specifiche dell’applicazione tracciate nel codice di esempio:

• Il numero totale di compiti elaborati dal processo del lavoro (inclusi compiti riusciti e falliti).
• Il numero di compiti falliti elaborati dal processo del lavoro.
• Il numero attuale di compiti in elaborazione dal processo del lavoro.

package main

import (
    "context"
    "log"
    "net/http"
    "os"
    "os/signal"
    "runtime"

    "github.com/hibiken/asynq"
    "github.com/hibiken/asynq/examples/tasks"
    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/promauto"
    "github.com/prometheus/client_golang/prometheus/promhttp"
    "golang.org/x/sys/unix"
)

// Variabili delle metriche.
var (
    processedCounter = promauto.NewCounterVec(
        prometheus.CounterOpts{
            Name: "processed_tasks_total",
            Help: "Numero totale di compiti elaborati",
        },
        []string{"task_type"},
    )

    failedCounter = promauto.NewCounterVec(
        prometheus.CounterOpts{
            Name: "failed_tasks_total",
            Help: "Numero totale di compiti falliti elaborati",
        },
        []string{"task_type"},
    )

    inProgressGauge = promauto.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: "in_progress_tasks",
            Help: "Numero attuale di compiti in elaborazione",
        },
        []string{"task_type"},
    )
)

func metricsMiddleware(next asynq.Handler) asynq.Handler {
    return asynq.HandlerFunc(func(ctx context.Context, t *asynq.Task) error {
        inProgressGauge.WithLabelValues(t.Type()).Inc()
        err := next.ProcessTask(ctx, t)
        inProgressGauge.WithLabelValues(t.Type()).Dec()
        if err != nil {
            failedCounter.WithLabelValues(t.Type()).Inc()
        }
        processedCounter.WithLabelValues(t.Type()).Inc()
        return err
    })
}

func main() {
    httpServeMux := http.NewServeMux()
    httpServeMux.Handle("/metrics", promhttp.Handler())
    metricsSrv := &http.Server{
        Addr:    ":2112",
        Handler: httpServeMux,
    }
    done := make(chan struct{})

    // Avvia il server delle metriche.
    go func() {
        err := metricsSrv.ListenAndServe()
        if err != nil && err != http.ErrServerClosed {
            log.Printf("Errore: il server delle metriche ha generato un errore: %v", err)
        }
        close(done)
    }()

    srv := asynq.NewServer(
        asynq.RedisClientOpt{Addr: ":6379"},
        asynq.Config{Concurrency: 20},
    )

    mux := asynq.NewServeMux()
    mux.Use(metricsMiddleware)
    mux.HandleFunc(tasks.TypeEmail, tasks.HandleEmailTask)

    // Avvia il server dei worker.
    if err := srv.Start(mux); err != nil {
        log.Fatalf("Impossibile avviare il server dei worker: %v", err)
    }

    // Attendi i segnali di terminazione.
    sigs := make(chan os.Signal, 1)
    signal.Notify(sigs, unix.SIGTERM, unix.SIGINT)
}