Periodic upstream health checks (HEALTH_INTERVAL, default 30s); log down/recovered transitions
This commit is contained in:
@@ -61,7 +61,8 @@ override file values. Invalid values fail at startup.
|
||||
| `LOG_FILE` | _(empty)_ | Append logs to this file instead of stderr |
|
||||
| `UNLOAD_POLL_INTERVAL` | `500ms` | `/api/ps` poll interval while unloading |
|
||||
| `HISTORY_POLL_INTERVAL` | `1s` | `/history/<id>` poll interval while a job runs |
|
||||
| `PROBE_TIMEOUT` | `5s` | Startup probe of both upstreams |
|
||||
| `PROBE_TIMEOUT` | `5s` | Startup probe of both upstreams (also per-probe health check timeout) |
|
||||
| `HEALTH_INTERVAL` | `30s` | Periodic upstream probe; down/recovered changes are logged |
|
||||
| `FREE_TIMEOUT` | `30s` | `POST /free` call after an image job |
|
||||
| `WARM_TIMEOUT` | `2m` | Warm-model reload after an image job |
|
||||
| `SHUTDOWN_TIMEOUT` | `10s` | Graceful shutdown on SIGINT/SIGTERM |
|
||||
|
||||
@@ -147,7 +147,8 @@ override file values. A missing file is fine; a malformed one is fatal.
|
||||
| `LOG_FILE` | `` | append logs to this file instead of stderr (useful as a service) |
|
||||
| `UNLOAD_POLL_INTERVAL` | `500ms` | `/api/ps` poll interval while unloading |
|
||||
| `HISTORY_POLL_INTERVAL` | `1s` | `/history/<id>` poll interval while a job runs |
|
||||
| `PROBE_TIMEOUT` | `5s` | startup probe of both upstreams |
|
||||
| `PROBE_TIMEOUT` | `5s` | startup probe of both upstreams (also the per-probe health check timeout) |
|
||||
| `HEALTH_INTERVAL` | `30s` | periodic probe of enabled upstreams; status changes (down/recovered) are logged |
|
||||
| `FREE_TIMEOUT` | `30s` | `POST /free` call after an image job |
|
||||
| `WARM_TIMEOUT` | `2m` | warm-model reload after an image job |
|
||||
| `SHUTDOWN_TIMEOUT` | `10s` | graceful shutdown on SIGINT/SIGTERM |
|
||||
|
||||
@@ -428,6 +428,7 @@ func run(ctx context.Context, cfg config.Config, log *slog.Logger, logOut io.Wri
|
||||
"unload_poll_interval", cfg.UnloadPollInterval,
|
||||
"history_poll_interval", cfg.HistoryPollInterval,
|
||||
"probe_timeout", cfg.ProbeTimeout,
|
||||
"health_interval", cfg.HealthInterval,
|
||||
"free_timeout", cfg.FreeTimeout,
|
||||
"warm_timeout", cfg.WarmTimeout,
|
||||
"shutdown_timeout", cfg.ShutdownTimeout,
|
||||
@@ -491,18 +492,23 @@ func run(ctx context.Context, cfg config.Config, log *slog.Logger, logOut io.Wri
|
||||
}
|
||||
|
||||
// Probe the enabled upstreams once; failure is logged, not fatal.
|
||||
probeCtx, probeCancel := context.WithTimeout(ctx, cfg.ProbeTimeout)
|
||||
probes := map[string]func(context.Context) error{}
|
||||
if ollamaClient != nil {
|
||||
if err := ollamaClient.Probe(probeCtx); err != nil {
|
||||
log.Warn("ollama probe failed", "url", cfg.OllamaURL, "err", err)
|
||||
}
|
||||
probes["ollama"] = ollamaClient.Probe
|
||||
}
|
||||
if comfyClient != nil {
|
||||
if err := comfyClient.Probe(probeCtx); err != nil {
|
||||
log.Warn("comfy probe failed", "url", cfg.ComfyURL, "err", err)
|
||||
probes["comfy"] = comfyClient.Probe
|
||||
}
|
||||
probeCtx, probeCancel := context.WithTimeout(ctx, cfg.ProbeTimeout)
|
||||
for name, probe := range probes {
|
||||
if err := probe(probeCtx); err != nil {
|
||||
log.Warn(name+" probe failed", "err", err)
|
||||
}
|
||||
}
|
||||
probeCancel()
|
||||
if cfg.HealthInterval > 0 {
|
||||
go healthLoop(ctx, cfg.HealthInterval, cfg.ProbeTimeout, log, probes)
|
||||
}
|
||||
|
||||
// Bind the listeners up front so a port conflict fails fast and the
|
||||
// readiness notification below really means "accepting connections".
|
||||
@@ -560,6 +566,38 @@ func run(ctx context.Context, cfg config.Config, log *slog.Logger, logOut io.Wri
|
||||
return nil
|
||||
}
|
||||
|
||||
// healthLoop probes the enabled upstreams every interval and logs status
|
||||
// transitions — "is DOWN" when a previously healthy upstream stops
|
||||
// answering, "recovered" when it comes back. The first round only
|
||||
// establishes the baseline; the startup probe already reported that state.
|
||||
func healthLoop(ctx context.Context, interval, probeTimeout time.Duration, log *slog.Logger, probes map[string]func(context.Context) error) {
|
||||
ticker := time.NewTicker(interval)
|
||||
defer ticker.Stop()
|
||||
up := map[string]bool{}
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
}
|
||||
for name, probe := range probes {
|
||||
pctx, cancel := context.WithTimeout(ctx, probeTimeout)
|
||||
err := probe(pctx)
|
||||
cancel()
|
||||
was, seen := up[name]
|
||||
now := err == nil
|
||||
if seen && now != was {
|
||||
if now {
|
||||
log.Warn(name + " upstream recovered")
|
||||
} else {
|
||||
log.Warn(name+" upstream is DOWN", "err", err)
|
||||
}
|
||||
}
|
||||
up[name] = now
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// updateLoop checks for signed updates on startup and every UPDATE_INTERVAL.
|
||||
// In service mode a staged update is applied by exiting with exitCodeUpdate
|
||||
// once the GPU lock is idle; the service recovery configuration restarts the
|
||||
|
||||
@@ -26,6 +26,7 @@ type Config struct {
|
||||
UnloadPollInterval time.Duration
|
||||
HistoryPollInterval time.Duration
|
||||
ProbeTimeout time.Duration
|
||||
HealthInterval time.Duration
|
||||
FreeTimeout time.Duration
|
||||
WarmTimeout time.Duration
|
||||
ShutdownTimeout time.Duration
|
||||
@@ -70,6 +71,7 @@ func Defaults() Config {
|
||||
UnloadPollInterval: 500 * time.Millisecond,
|
||||
HistoryPollInterval: time.Second,
|
||||
ProbeTimeout: 5 * time.Second,
|
||||
HealthInterval: 30 * time.Second,
|
||||
FreeTimeout: 30 * time.Second,
|
||||
WarmTimeout: 2 * time.Minute,
|
||||
ShutdownTimeout: 10 * time.Second,
|
||||
@@ -166,6 +168,7 @@ func Load(getenv func(string) string) (Config, error) {
|
||||
{"UNLOAD_POLL_INTERVAL", &cfg.UnloadPollInterval},
|
||||
{"HISTORY_POLL_INTERVAL", &cfg.HistoryPollInterval},
|
||||
{"PROBE_TIMEOUT", &cfg.ProbeTimeout},
|
||||
{"HEALTH_INTERVAL", &cfg.HealthInterval},
|
||||
{"FREE_TIMEOUT", &cfg.FreeTimeout},
|
||||
{"WARM_TIMEOUT", &cfg.WarmTimeout},
|
||||
{"SHUTDOWN_TIMEOUT", &cfg.ShutdownTimeout},
|
||||
|
||||
@@ -39,7 +39,8 @@ func sampleEntries(logFile string) []sampleEntry {
|
||||
{"LOG_FILE", logFile, "Append logs to this file instead of stderr (a Windows service has no console)", logFile != ""},
|
||||
{"UNLOAD_POLL_INTERVAL", "500ms", "/api/ps poll interval while unloading", false},
|
||||
{"HISTORY_POLL_INTERVAL", "1s", "/history/<id> poll interval while a job runs", false},
|
||||
{"PROBE_TIMEOUT", "5s", "Startup probe timeout for the enabled upstreams", false},
|
||||
{"PROBE_TIMEOUT", "5s", "Probe timeout for the startup probe and the periodic health check", false},
|
||||
{"HEALTH_INTERVAL", "30s", "How often enabled upstreams are probed; status changes are logged", false},
|
||||
{"FREE_TIMEOUT", "30s", "Timeout for the POST /free call after an image job", false},
|
||||
{"WARM_TIMEOUT", "2m", "Timeout for the warm-model reload after an image job", false},
|
||||
{"SHUTDOWN_TIMEOUT", "10s", "Graceful shutdown timeout on SIGINT/SIGTERM", false},
|
||||
|
||||
Reference in New Issue
Block a user