()
| 55 | } |
| 56 | |
| 57 | func main() { |
| 58 | cfg := parseFlags() |
| 59 | |
| 60 | redisDSN, skipTLSVerify, err := resolveRedisSettings(cfg) |
| 61 | if err != nil { |
| 62 | exitf("failed to resolve redis settings: %v", err) |
| 63 | } |
| 64 | |
| 65 | redisOpt, err := redis_db.ParseRedisURL(redisDSN, skipTLSVerify) |
| 66 | if err != nil { |
| 67 | exitf("failed to parse redis config: %v", err) |
| 68 | } |
| 69 | |
| 70 | inspector := asynq.NewInspector(asynq.RedisClientOpt{ |
| 71 | Addr: redisOpt.Addr, |
| 72 | Password: redisOpt.Password, |
| 73 | DB: redisOpt.DB, |
| 74 | TLSConfig: redisOpt.TLSConfig, |
| 75 | }) |
| 76 | defer func() { _ = inspector.Close() }() |
| 77 | signalCh := make(chan os.Signal, 1) |
| 78 | signal.Notify(signalCh, os.Interrupt, syscall.SIGTERM) |
| 79 | defer signal.Stop(signalCh) |
| 80 | ticker := time.NewTicker(cfg.interval) |
| 81 | defer ticker.Stop() |
| 82 | |
| 83 | startDeadline := time.Now().Add(cfg.timeout) |
| 84 | var started bool |
| 85 | var startSnapshot snapshot |
| 86 | var samples []snapshot |
| 87 | var latestSnapshot snapshot |
| 88 | var haveLatestSnapshot bool |
| 89 | interrupted := false |
| 90 | |
| 91 | for { |
| 92 | snap, err := takeSnapshot(inspector, cfg) |
| 93 | if err != nil { |
| 94 | exitf("failed to inspect queues: %v", err) |
| 95 | } |
| 96 | latestSnapshot = snap |
| 97 | haveLatestSnapshot = true |
| 98 | |
| 99 | if !started { |
| 100 | if cfg.wait && !hasWork(snap) { |
| 101 | if time.Now().After(startDeadline) { |
| 102 | exitf("timed out waiting for queue backlog") |
| 103 | } |
| 104 | if waitForNextTickOrInterrupt(ticker, signalCh) { |
| 105 | interrupted = true |
| 106 | break |
| 107 | } |
| 108 | continue |
| 109 | } |
| 110 | |
| 111 | started = true |
| 112 | startSnapshot = snap |
| 113 | samples = append(samples, snap) |
| 114 | if err := writeCurrentSummary(cfg, startSnapshot, snap, samples); err != nil { |
nothing calls this directly
no test coverage detected