tutorial

How to Monitor Asynq with Vigilmon

Asynq brings Redis-backed background job processing to Go — but silent worker panics, Redis disconnections, and queue backlog growth can halt your task pipeline without a single alert. Learn how to monitor queue health, worker liveness, failure rates, and Redis connectivity with Vigilmon.

Asynq is a simple, reliable, and efficient distributed task queue library for Go, backed by Redis. It supports priority queues, scheduled tasks, task deduplication, rate limiting, and automatic retry — making it the go-to background job solution for Go services that need reliable async processing without pulling in a full message broker.

But Asynq failures can be invisible. A Go worker that panics and exits, a Redis server that times out, or a queue whose retry limit is quietly exhausted leaves jobs unprocessed while the scheduler marks them as enqueued. Vigilmon gives you external visibility through HTTP monitors for Asynq's built-in inspector API and heartbeat monitors for worker goroutines.

Why Asynq Needs External Monitoring

Asynq ships with asynqmon — a web UI and CLI for queue inspection. But external monitoring with Vigilmon adds:

  • Proactive alerting when worker processes exit or Redis connections drop
  • Queue depth threshold checks before pending tasks exceed processing capacity
  • Retry and archive monitoring to catch job classes that always fail
  • Active task count checks to detect worker goroutine exhaustion
  • Heartbeat monitoring for worker processes that must prove they are dequeuing

Step 1: Use the Asynq Inspector API

Asynq provides an Inspector type for programmatic queue introspection. Build a health endpoint around it:

// health/asynq.go
package health

import (
	"encoding/json"
	"fmt"
	"net/http"
	"os"
	"strconv"
	"strings"

	"github.com/hibiken/asynq"
)

var (
	redisAddr         = getEnv("REDIS_ADDR", "localhost:6379")
	redisPassword     = getEnv("REDIS_PASSWORD", "")
	pendingThreshold  = getEnvInt("ASYNQ_PENDING_THRESHOLD", 500)
	archivedThreshold = getEnvInt("ASYNQ_ARCHIVED_THRESHOLD", 50)
	queueNames        = strings.Split(getEnv("ASYNQ_QUEUES", "default"), ",")
)

type QueueStats struct {
	Queue    string `json:"queue"`
	Pending  int    `json:"pending"`
	Active   int    `json:"active"`
	Retry    int    `json:"retry"`
	Archived int    `json:"archived"`
	Paused   bool   `json:"paused"`
}

type HealthResponse struct {
	Status string       `json:"status"`
	Issues []string     `json:"issues,omitempty"`
	Queues []QueueStats `json:"queues,omitempty"`
	Reason string       `json:"reason,omitempty"`
}

func AsynqHealthHandler(w http.ResponseWriter, r *http.Request) {
	inspector := asynq.NewInspector(asynq.RedisClientOpt{
		Addr:     redisAddr,
		Password: redisPassword,
	})
	defer inspector.Close()

	var issues []string
	var allStats []QueueStats

	for _, q := range queueNames {
		info, err := inspector.GetQueueInfo(q)
		if err != nil {
			issues = append(issues, fmt.Sprintf("queue_%s_unreachable", q))
			continue
		}

		stats := QueueStats{
			Queue:    q,
			Pending:  info.Pending,
			Active:   info.Active,
			Retry:    info.Retry,
			Archived: info.Archived,
			Paused:   info.Paused,
		}
		allStats = append(allStats, stats)

		if info.Paused {
			issues = append(issues, fmt.Sprintf("queue_%s_paused", q))
		}
		if info.Pending > pendingThreshold {
			issues = append(issues, fmt.Sprintf("queue_%s_pending_%d", q, info.Pending))
		}
		if info.Archived > archivedThreshold {
			issues = append(issues, fmt.Sprintf("queue_%s_archived_%d", q, info.Archived))
		}
		if info.Active == 0 && info.Pending > 0 {
			issues = append(issues, fmt.Sprintf("queue_%s_no_active_workers", q))
		}
	}

	resp := HealthResponse{Queues: allStats}
	status := http.StatusOK
	if len(issues) > 0 {
		resp.Status = "degraded"
		resp.Issues = issues
		status = http.StatusServiceUnavailable
	} else {
		resp.Status = "ok"
	}

	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(status)
	json.NewEncoder(w).Encode(resp)
}

func getEnv(key, fallback string) string {
	if v := os.Getenv(key); v != "" {
		return v
	}
	return fallback
}

func getEnvInt(key string, fallback int) int {
	if v := os.Getenv(key); v != "" {
		if n, err := strconv.Atoi(v); err == nil {
			return n
		}
	}
	return fallback
}

Wire it into your HTTP server:

// main.go
package main

import (
	"log"
	"net/http"

	"myapp/health"
)

func main() {
	http.HandleFunc("/health/asynq", health.AsynqHealthHandler)
	log.Fatal(http.ListenAndServe(":3008", nil))
}

Step 2: Add Heartbeat Pings to Workers

Wire Vigilmon heartbeat pings directly into your Asynq task handlers:

// workers/worker.go
package workers

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"os"
	"sync/atomic"
	"time"

	"github.com/hibiken/asynq"
)

var (
	vigilmonURL   = os.Getenv("VIGILMON_HEARTBEAT_URL")
	jobsProcessed atomic.Int64
	heartbeatEvery int64 = 10
)

type EmailPayload struct {
	To      string `json:"to"`
	Subject string `json:"subject"`
}

func HandleEmailTask(ctx context.Context, t *asynq.Task) error {
	var p EmailPayload
	if err := json.Unmarshal(t.Payload(), &p); err != nil {
		return fmt.Errorf("unmarshal: %w", asynq.SkipRetry)
	}

	if err := sendEmail(p); err != nil {
		return err
	}

	n := jobsProcessed.Add(1)
	if n%heartbeatEvery == 0 {
		pingVigilmon()
	}
	return nil
}

func sendEmail(p EmailPayload) error {
	// email sending logic
	return nil
}

func pingVigilmon() {
	if vigilmonURL == "" {
		return
	}
	go func() {
		client := &http.Client{Timeout: 5 * time.Second}
		if _, err := client.Get(vigilmonURL); err != nil {
			log.Printf("vigilmon heartbeat failed: %v", err)
		}
	}()
}

For low-traffic workers or to ping on a timer:

// Start a background goroutine that pings on a schedule
func StartHeartbeatTicker(ctx context.Context) {
	if vigilmonURL == "" {
		return
	}
	ticker := time.NewTicker(60 * time.Second)
	go func() {
		for {
			select {
			case <-ticker.C:
				pingVigilmon()
			case <-ctx.Done():
				ticker.Stop()
				return
			}
		}
	}()
}

Start the ticker from main before the Asynq server starts.

Step 3: Wire Up the Asynq Server with Graceful Shutdown

// main.go
package main

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

	"github.com/hibiken/asynq"
	"myapp/health"
	"myapp/workers"
)

func main() {
	redisOpt := asynq.RedisClientOpt{
		Addr:     os.Getenv("REDIS_ADDR"),
		Password: os.Getenv("REDIS_PASSWORD"),
	}

	srv := asynq.NewServer(redisOpt, asynq.Config{
		Concurrency: 20,
		Queues: map[string]int{
			"critical": 6,
			"default":  3,
			"low":      1,
		},
	})

	mux := asynq.NewServeMux()
	mux.HandleFunc("email:send", workers.HandleEmailTask)

	// Start health HTTP server
	go func() {
		http.HandleFunc("/health/asynq", health.AsynqHealthHandler)
		log.Fatal(http.ListenAndServe(":3008", nil))
	}()

	// Start heartbeat ticker
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	workers.StartHeartbeatTicker(ctx)

	// Handle signals
	sig := make(chan os.Signal, 1)
	signal.Notify(sig, syscall.SIGTERM, syscall.SIGINT)
	go func() {
		<-sig
		cancel()
		srv.Shutdown()
	}()

	if err := srv.Run(mux); err != nil {
		log.Fatal(err)
	}
}

Step 4: Configure Vigilmon HTTP Monitors

  1. Log in to vigilmon.online and go to Monitors → New Monitor
  2. Choose HTTP / HTTPS
  3. Set the URL to your Asynq health endpoint: https://your-app.example.com/health/asynq
  4. Set the check interval to 1 minute
  5. Under Expected response, configure:
    • Status code: 200
    • Response body contains: "status":"ok"
    • Response time threshold: 2000ms
  6. Under Alert channels, assign your Slack or PagerDuty integration
  7. Save the monitor

| Monitor URL | Purpose | Interval | |---|---|---| | /health/asynq | Queue depth, archived count, paused state, worker presence | 1 min |

Step 5: Configure Heartbeat Monitors

  1. In Vigilmon, go to Monitors → New Monitor → Heartbeat
  2. Set the name: asynq-worker
  3. Set the expected interval: 2 minutes
  4. Set the grace period: 5 minutes
  5. Save — copy the heartbeat URL: https://vigilmon.online/heartbeat/abc123xyz
  6. Set VIGILMON_HEARTBEAT_URL in your worker environment

| Worker Service | Heartbeat Interval | Grace Period | |---|---|---| | email-worker | 2 min | 5 min | | report-worker | 5 min | 10 min | | critical-worker | 1 min | 3 min |

Step 6: Alert Routing

| Monitor | Alert Channel | Priority | |---|---|---| | Queue health /health/asynq | Slack + PagerDuty | P1 | | Worker heartbeats | Slack + PagerDuty | P1 |

Key failure modes to watch:

  • Archived task growth: tasks that exhaust all retries land in the archive — sustained growth means a bug in your handler is silently discarding work
  • Paused queues: Asynq queues can be paused via CLI or API; accidental pauses are a common silent outage
  • Zero active workers with pending tasks: the worker process has exited without restarting

Summary

Asynq's Redis-backed simplicity makes it easy to deploy — but that simplicity also means failures can be quiet. Vigilmon catches them before users notice:

| Monitor Type | What It Covers | |---|---| | HTTP monitor on /health/asynq | Queue depth, archive accumulation, pause state | | Heartbeat monitor | Worker goroutine liveness and active dequeuing |

Get started free at vigilmon.online — your first Asynq monitor is running in under two minutes.

Monitor your app with Vigilmon

Free plan — 5 monitors, no credit card required. Up and running in 60 seconds.

Start free →
← Back to all articles