diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 69353c7..2574cda 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -37,6 +37,7 @@ jobs: # /healthz flip. No Kubernetes needed. - run: bash smoke_test.sh - run: bash smoke_probe_test.sh + - run: bash smoke_engine_test.sh - run: docker build -t hang-watcher:ci . gate: diff --git a/CHANGELOG.md b/CHANGELOG.md index 69f69d0..a8e0d35 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,23 @@ of; the commit history for them was carried over, but the release tags were not. ## [Unreleased] ### Added +- **`GET /engine`: which engine instance is running and when it started**, for + clients that share state with the engine and must notice a restart -- a + prefill/decode router, for one. The kubelet restarts the engine container in + place, keeping the pod IP, so watching pods or endpoints shows nothing. + Returns `{"engine_id", "started_at", "source", "restarts", + "watcher_started_at"}` on the `/healthz` port (`restarts` counts the engine + restarts this sidecar has seen since it started); + `503` before the engine has been identified. The identity is taken from + vLLM's `process_start_time_seconds` (exact start time), sglang's `startup_*` + gauges (a fingerprint that survives sidecar restarts; `started_at` is then + when the sidecar first saw the instance), or, with neither, an id minted by + the sidecar that does not survive its restart. +- A restart detected this way puts the watcher back into its startup state: a + fresh baseline and warm-up grace period from the new instance's first + scrape. Counter regression, the only signal before, misses an engine that was + idle (counters stay 0) or whose new process outran the old count within one + poll. - CI on every pull request: `gofmt`, `go vet`, `go test`, both smoke tests and a docker build, plus a gate for the license text and committed credentials. - Dependabot for the GitHub Actions, which are pinned to commit SHAs. diff --git a/README.md b/README.md index 6568f66..be7a766 100644 --- a/README.md +++ b/README.md @@ -200,6 +200,69 @@ Two things matter: One sidecar and one probe cover both; LWS only adds the switch that turns a single-pod restart into a group recreation. +## Engine identity: `GET /engine` + +A client that keeps state tied to one engine process -- a prefill/decode router +holding KV-transfer sessions with its workers, for example -- needs to know when +that process was replaced. Watching the pod does not tell it: the kubelet +restarts the engine container in place, and the pod, its IP and its endpoints +all stay the same. The sidecar already scrapes the engine every poll, so it +reports which instance it is looking at, on the same port as `/healthz`: + +```console +$ curl -s http://:9090/engine +{"engine_id":"pst-1791943641.29","started_at":"2026-10-14T02:07:21.29Z","source":"process_start_time", + "restarts":1,"watcher_started_at":"2026-10-14T01:52:10.48Z"} +``` + +Before the engine has been identified it answers `503`, so "not known yet" can +never be mistaken for an id. + +- `engine_id` changes exactly when the engine process is replaced. **Compare + this**, not `started_at`. +- `started_at` is the engine's start time (UTC), as well as it can be known. +- `restarts` is how many times this sidecar has seen the engine replaced since + `watcher_started_at`, the sidecar's own start. It starts again from 0 when the + sidecar restarts, so it reads well for a person; a client deciding whether + to act should compare `engine_id`. +- `source` says where `engine_id` and `started_at` came from, and so how far + each can be trusted: + +| `source` | taken from | `started_at` | `engine_id` survives a sidecar restart | +|---|---|---|---| +| `process_start_time` | `process_start_time_seconds` (vLLM) | the engine's own start time | yes | +| `startup_fingerprint` | sglang's `startup_time_seconds` / `startup_cuda_graph_time_seconds` (sglang >= 0.5.19) | when this sidecar first saw the instance | yes | +| `observed` | neither available; minted by the sidecar, renewed when token counters go backwards | when this sidecar first saw the instance | **no** | + +sglang publishes no start timestamp, only how long each startup phase took. +Those durations are set once per process and measured to the nanosecond, so two +processes practically never report the same set, and together they serve as a +fingerprint -- but they cannot say *when* the process started. A scrape need +not carry all of them (some phases are reported late, and a body cut off at the +read limit loses the tail), so only a series seen before with a different value +counts as a new process; series joining or missing do not. + +With `observed`, a restarted sidecar reports a new id for the same engine, and +an engine restarted while idle (counters at 0 before and after) goes unnoticed; +treat it as advisory. The sidecar logs a warning when it falls back to it. + +**vLLM with `--api-server-count` > 1 ends up on `observed`.** vLLM then runs +its metrics in multiprocess mode (it sets `PROMETHEUS_MULTIPROC_DIR` itself, +and so does any deployment that sets it by hand), and that registry has no +process collector, so `process_start_time_seconds` is not exported. + +Why not read the engine process the way `ps` does: containers in a pod have +separate pid namespaces, so the sidecar's `/proc` holds only its own processes. +`shareProcessNamespace` would change that, but it makes the pause container +pid 1, and engine charts commonly rely on the engine being pid 1 of its +container -- a preStop hook that signals pid 1 and then kills every other +process it can see would take down the whole pod. + +A restart detected this way puts the watcher back into its startup state (a +fresh baseline and warm-up grace period from the new instance's first scrape). +Counter regression, the only signal before, misses an engine that was idle or +whose new process outran the old count within one poll. + ## Build and test ```bash @@ -208,7 +271,7 @@ CGO_ENABLED=0 go build -o hang-watcher . docker build -t hang-watcher:dev . ``` -`smoke_test.sh` and `smoke_probe_test.sh` run the real binary against a fake +`smoke_test.sh`, `smoke_probe_test.sh` and `smoke_engine_test.sh` run the real binary against a fake engine and assert end-to-end behaviour, including the case a unit test cannot reach: that a *single* probe failure does not flip `/healthz` to 503. diff --git a/engine.go b/engine.go new file mode 100644 index 0000000..c0b5a14 --- /dev/null +++ b/engine.go @@ -0,0 +1,236 @@ +// Engine instance identity: which engine process is the sidecar looking at, +// and when did it start. Served on GET /engine so that something sharing state +// with the engine -- a prefill/decode router, for one -- can tell that the +// engine behind it was replaced and act on it. +// +// Why the sidecar and not the engine: the kubelet restarts the engine +// container in place, keeping the pod and its IP, so a client that only +// watches pod events or endpoints sees nothing. And the engines do not agree +// on how to say it: vLLM exposes process_start_time_seconds, sglang does not. +// One endpoint with one meaning saves every consumer from learning both. +// +// Why /metrics and not the process itself: containers in a pod have separate +// pid namespaces, so the sidecar cannot see the engine's processes (its /proc +// holds only its own). shareProcessNamespace would change that, but it makes +// the pause container pid 1, and engine charts commonly rely on the engine +// being pid 1 of its container (a preStop that signals pid 1 and then kills +// every other process it can see would take down the whole pod). +// +// The identity comes from the engine's own /metrics, strongest first: +// +// 1. process_start_time_seconds (vLLM, and anything using the default +// Prometheus process collector) -- an exact start time, so the id is +// derived from it and started_at is exact. Not there when vLLM runs its +// metrics in multiprocess mode (PROMETHEUS_MULTIPROC_DIR, which it sets +// itself for --api-server-count > 1): that registry has no process +// collector, and the identity falls through to observed. +// 2. sglang's startup_* gauges (startup_time_seconds, +// startup_cuda_graph_time_seconds, sglang >= 0.5.19). They are durations, +// not timestamps, but they are set once per process and measured to the +// nanosecond, so two processes practically never report the same set. +// That makes them a fingerprint: the id is stable across sidecar restarts, +// but started_at is only the time this sidecar first saw the instance. +// 3. Neither present: the id is minted by the sidecar on first sight and +// re-minted when the token counters go backwards. This one does NOT +// survive a sidecar restart -- a restarted sidecar reports a new id for +// the same engine -- so source tells consumers how far to trust it. +// +// Consumers should compare engine_id, not started_at: for sources 2 and 3 +// started_at is an observation time and moves when the sidecar restarts. +package main + +import ( + "encoding/json" + "hash/fnv" + "math" + "net/http" + "sort" + "strconv" + "time" +) + +const processStartMetric = "process_start_time_seconds" + +// startupMetrics: sglang gauges that are written once, at startup. +var startupMetrics = map[string]bool{ + "sglang:startup_time_seconds": true, + "sglang:startup_cuda_graph_time_seconds": true, +} + +const ( + idProcessStart = "process_start_time" + idStartup = "startup_fingerprint" + idObserved = "observed" +) + +type engineState struct { + id string + idSource string + startedAt time.Time + restarts int // restarts seen by this sidecar since it started + + // The startup fingerprint is kept as series -> value rather than as a hash, + // because a scrape need not carry every series: during startup some phases + // have not been reported yet, and a /metrics body cut off at the read limit + // loses whatever came last. So only a series seen before with a DIFFERENT + // value is a new process; series joining or missing are not. + startup map[string]float64 +} + +// noteEngine folds one successful scrape into the identity and reports whether +// the engine was replaced since the last one. The first identity ever seen is +// not a restart, and neither is moving to a stronger source (an sglang engine +// whose startup gauges only show up after its first scrape). +func (w *watcher) noteEngine(now time.Time, m metricsSnap, metricsOK bool) bool { + w.mu.Lock() + defer w.mu.Unlock() + e := &w.eng + if !metricsOK { + return false + } + switch { + case m.startTime > 0: + id := "pst-" + strconv.FormatFloat(m.startTime, 'f', -1, 64) + if e.idSource == idProcessStart && e.id == id { + return false + } + restarted := e.idSource == idProcessStart + e.id, e.idSource, e.startup = id, idProcessStart, nil + // to the millisecond: the float carries no more, and finer digits are noise + e.startedAt = time.UnixMilli(int64(math.Round(m.startTime * 1000))).UTC() + if restarted { + e.restarts++ + } + return restarted + case len(m.startup) > 0: + if e.idSource == idStartup && sameProcess(e.startup, m.startup) { + for k, v := range m.startup { + e.startup[k] = v // phases reported late join the set; the id stays put + } + return false + } + restarted := e.idSource == idStartup + e.id, e.idSource, e.startup = "sfp-"+fingerprint(m.startup), idStartup, map[string]float64{} + for k, v := range m.startup { + e.startup[k] = v + } + if restarted || e.startedAt.IsZero() { + e.startedAt = now + } + if restarted { + e.restarts++ + } + return restarted + default: + if e.id == "" { + e.id, e.idSource, e.startedAt = observedID(now, 0), idObserved, now + } + return false + } +} + +// noteRegress handles token counters going backwards. That is only the +// identity signal when the engine offers nothing better; with a real identity +// noteEngine has already decided, and a regression without an identity change +// is not a second restart. +func (w *watcher) noteRegress(now time.Time) { + w.mu.Lock() + defer w.mu.Unlock() + e := &w.eng + if e.idSource != idObserved { + return + } + e.restarts++ + e.id, e.startedAt = observedID(now, e.restarts), now +} + +// sameProcess: no series present in both has changed value. A series missing +// from cur says nothing (not reported yet, or cut off) and is not a restart. +func sameProcess(old, cur map[string]float64) bool { + for k, v := range cur { + if ov, ok := old[k]; ok && ov != v { + return false + } + } + return true +} + +// fingerprint hashes the startup series. Computed when the identity is first +// taken, so a series joining later does not change the id. +func fingerprint(m map[string]float64) string { + keys := make([]string, 0, len(m)) + for k := range m { + keys = append(keys, k) + } + sort.Strings(keys) + h := fnv.New64a() + for _, k := range keys { + _, _ = h.Write([]byte(k + "=" + strconv.FormatFloat(m[k], 'g', -1, 64) + "\n")) + } + return strconv.FormatUint(h.Sum64(), 16) +} + +func observedID(now time.Time, n int) string { + return "obs-" + strconv.FormatInt(now.UnixNano(), 36) + "-" + strconv.Itoa(n) +} + +// engineInfo is the /engine response body. source says how far started_at can +// be trusted: process_start_time is the engine's real start; +// startup_fingerprint and observed are fallbacks, where started_at is when this +// sidecar first saw the instance. +// +// restarts counts engine restarts this sidecar has seen since +// watcher_started_at. It starts from 0 whenever the sidecar does, so it is a +// convenience for people reading the endpoint; a client deciding whether the +// engine was replaced should still compare engine_id. +type engineInfo struct { + EngineID string `json:"engine_id"` + StartedAt string `json:"started_at"` + Source string `json:"source"` + Restarts int `json:"restarts"` + WatcherStartedAt string `json:"watcher_started_at"` +} + +func (w *watcher) Engine() (engineInfo, bool) { + w.mu.RLock() + defer w.mu.RUnlock() + e := w.eng + return engineInfo{ + EngineID: e.id, StartedAt: rfc(e.startedAt), Source: e.idSource, + Restarts: e.restarts, WatcherStartedAt: rfc(w.bornAt), + }, e.id != "" +} + +// engineRestarts: restarts seen by this sidecar since it started. +func (w *watcher) engineRestarts() int { + w.mu.RLock() + defer w.mu.RUnlock() + return w.eng.restarts +} + +func rfc(t time.Time) string { + if t.IsZero() { + return "" + } + return t.UTC().Format(time.RFC3339Nano) +} + +// engineHandler serves GET /engine: 200 with the identity once one is known, +// 503 before the first sighting, so "not known yet" is never read as an id. +func engineHandler(w *watcher) http.HandlerFunc { + return func(rw http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet && r.Method != http.MethodHead { + rw.Header().Set("Allow", "GET, HEAD") + http.Error(rw, "method not allowed", http.StatusMethodNotAllowed) + return + } + info, known := w.Engine() + rw.Header().Set("Cache-Control", "no-store") + if !known { + http.Error(rw, "engine not identified yet", http.StatusServiceUnavailable) + return + } + rw.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(rw).Encode(info) + } +} diff --git a/engine_test.go b/engine_test.go new file mode 100644 index 0000000..252467e --- /dev/null +++ b/engine_test.go @@ -0,0 +1,282 @@ +package main + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +// Metric lines as a real vLLM (v0.29 nightly) and sglang (0.5.19) print them. +const ( + vllmText = `# HELP process_start_time_seconds Start time of the process since unix epoch in seconds. +process_start_time_seconds 1.79194364129e+09 +vllm:generation_tokens_total{model_name="m"} 100 +vllm:num_requests_running{model_name="m"} 0 +` + vllmRestartedText = `process_start_time_seconds 1.79195001337e+09 +vllm:generation_tokens_total{model_name="m"} 5000 +vllm:num_requests_running{model_name="m"} 1 +` + sglangPartial = `sglang:startup_time_seconds{engine_type="unified",model_name="qwen",phase="load_weight"} 6.029548802413046 +sglang:realtime_tokens_total{mode="decode"} 0 +sglang:num_running_reqs 0 +` + sglangFull = `sglang:startup_time_seconds{engine_type="unified",model_name="qwen",phase="load_weight"} 6.029548802413046 +sglang:startup_time_seconds{engine_type="unified",model_name="qwen",phase="scheduler_e2e"} 109.03901622537524 +sglang:startup_cuda_graph_time_seconds{engine_type="unified",model_name="qwen",phase="decode"} 17.055109737440944 +sglang:realtime_tokens_total{mode="decode"} 0 +sglang:num_running_reqs 0 +` + sglangRestarted = `sglang:startup_time_seconds{engine_type="unified",model_name="qwen",phase="load_weight"} 5.98812301 +sglang:startup_time_seconds{engine_type="unified",model_name="qwen",phase="scheduler_e2e"} 108.71100912 +sglang:startup_cuda_graph_time_seconds{engine_type="unified",model_name="qwen",phase="decode"} 17.30122841 +sglang:realtime_tokens_total{mode="decode"} 0 +sglang:num_running_reqs 0 +` + bareText = "sglang:generation_tokens_total 1000\nsglang:num_running_reqs 0\n" +) + +// view: the /engine body plus what the tests also assert on -- whether an id is +// known, and the restarts this sidecar counted. +type engView struct { + EngineID, StartedAt, IDSource string + Known, Exact bool + RestartsObserved int +} + +func view(w *watcher) engView { + info, known := w.Engine() + return engView{info.EngineID, info.StartedAt, info.Source, known, + info.Source == idProcessStart, w.engineRestarts()} +} + +func scrape(w *watcher, at time.Time, text string) { + w.stepNoWarmup(at, parseMetrics(text), nil, 30*time.Second, func() bool { return true }) +} + +func TestEngineVLLMStartTimeIsExact(t *testing.T) { + w := newWatcher() + t0 := time.Now() + scrape(w, t0, vllmText) + e := view(w) + if !e.Known || e.IDSource != idProcessStart || !e.Exact { + t.Fatalf("want an exact process_start_time identity, got %+v", e) + } + if e.StartedAt != "2026-10-14T02:07:21.29Z" { + t.Errorf("started_at = %s, want the engine's own start time 2026-10-14T02:07:21.29Z", e.StartedAt) + } + id := e.EngineID + scrape(w, t0.Add(5*time.Second), vllmText) + if e := view(w); e.EngineID != id || e.RestartsObserved != 0 { + t.Fatalf("same process must keep its id: %+v", e) + } +} + +// The restart that counter regression cannot see: the new process has already +// produced more tokens than the old one had in total by the next poll. The +// control shows that without a start time this goes unnoticed, so the test +// would fail if the restart were detected by the counters instead. +func TestEngineRestartWithoutCounterRegression(t *testing.T) { + t0 := time.Now() + + ctrl := newWatcher() + scrape(ctrl, t0, "vllm:generation_tokens_total 100\n") + scrape(ctrl, t0.Add(5*time.Second), "vllm:generation_tokens_total 5000\n") + if e := view(ctrl); e.RestartsObserved != 0 { + t.Fatalf("control: without an identity, counters alone cannot see this restart; got %+v", e) + } + + w := newWatcher() + scrape(w, t0, vllmText) + before := view(w).EngineID + scrape(w, t0.Add(5*time.Second), vllmRestartedText) + e := view(w) + if e.RestartsObserved != 1 || e.EngineID == before { + t.Fatalf("a new process_start_time is a restart: %+v", e) + } + if info, _ := w.Engine(); info.Restarts != 1 { + t.Errorf("restarts on /engine = %d, want 1", info.Restarts) + } + if e.StartedAt != "2026-10-14T03:53:33.37Z" { + t.Errorf("started_at should follow the new process, got %s", e.StartedAt) + } + if _, state, _ := w.Hung(); state != "baseline" { + t.Errorf("state = %q, want baseline (a restart starts over from a fresh baseline)", state) + } + if !w.readyAt.Equal(t0.Add(5 * time.Second)) { + t.Errorf("warm-up must re-arm at the restart, readyAt=%v", w.readyAt) + } +} + +// An idle engine never moves its counters, so a restart leaves them at 0 and +// only the identity notices. +func TestEngineIdleRestart(t *testing.T) { + w := newWatcher() + t0 := time.Now() + scrape(w, t0, sglangFull) + scrape(w, t0.Add(5*time.Second), sglangRestarted) + if e := view(w); e.RestartsObserved != 1 { + t.Fatalf("idle sglang restart must be detected from the startup gauges: %+v", e) + } +} + +// Startup gauges arriving over several scrapes are the same process; a value +// changing is a new one. +func TestEngineSglangFingerprintGrowsWithoutRestart(t *testing.T) { + w := newWatcher() + t0 := time.Now() + scrape(w, t0, sglangPartial) + e1 := view(w) + if e1.IDSource != idStartup || e1.Exact { + t.Fatalf("want a startup fingerprint with an observed start time: %+v", e1) + } + scrape(w, t0.Add(5*time.Second), sglangFull) + if e := view(w); e.EngineID != e1.EngineID || e.RestartsObserved != 0 { + t.Fatalf("more phases appearing is not a restart: before %+v after %+v", e1, e) + } + scrape(w, t0.Add(10*time.Second), sglangRestarted) + if e := view(w); e.EngineID == e1.EngineID || e.RestartsObserved != 1 { + t.Fatalf("a changed startup value is a restart: %+v", e) + } +} + +// What makes the fingerprint worth having over an observed id: a sidecar that +// restarts and sees the same engine reports the same id. +func TestEngineIDSurvivesSidecarRestart(t *testing.T) { + t0 := time.Now() + for _, text := range []string{vllmText, sglangFull} { + a, b := newWatcher(), newWatcher() + scrape(a, t0, text) + scrape(b, t0.Add(time.Hour), text) + if view(a).EngineID != view(b).EngineID { + t.Errorf("same engine, two sidecars, different ids: %s vs %s", view(a).EngineID, view(b).EngineID) + } + } + // ...and an observed id does not, which is why id_source is reported. + a, b := newWatcher(), newWatcher() + scrape(a, t0, bareText) + scrape(b, t0.Add(time.Hour), bareText) + if view(a).EngineID == view(b).EngineID { + t.Errorf("observed ids are per sidecar; equal ids would mean the test is not exercising that path") + } +} + +func TestEngineObservedFallsBackToRegression(t *testing.T) { + w := newWatcher() + t0 := time.Now() + scrape(w, t0, bareText) + e1 := view(w) + if e1.IDSource != idObserved { + t.Fatalf("no identity metrics: want observed, got %+v", e1) + } + scrape(w, t0.Add(5*time.Second), "sglang:generation_tokens_total 3\nsglang:num_running_reqs 0\n") + if e := view(w); e.RestartsObserved != 1 || e.EngineID == e1.EngineID { + t.Fatalf("counters going backwards must mint a new observed id: %+v", e) + } +} + +// With a real identity a regression is not counted a second time. +func TestEngineRegressionNotDoubleCounted(t *testing.T) { + w := newWatcher() + t0 := time.Now() + scrape(w, t0, vllmRestartedText) // tp 5000 + scrape(w, t0.Add(5*time.Second), vllmText) + if e := view(w); e.RestartsObserved != 1 { + t.Fatalf("one restart, got %d", e.RestartsObserved) + } +} + +func TestEngineUnreachableKeepsID(t *testing.T) { + w := newWatcher() + t0 := time.Now() + scrape(w, t0, vllmText) + id := view(w).EngineID + w.stepNoWarmup(t0.Add(5*time.Second), metricsSnap{}, errTest, 30*time.Second, nil) + if e := view(w); e.EngineID != id || e.RestartsObserved != 0 { + t.Fatalf("an outage is not a new id: %+v", e) + } +} + +var errTest = &httpErr{500} + +func TestEngineHandler(t *testing.T) { + w := newWatcher() + h := engineHandler(w) + + rec := httptest.NewRecorder() + h(rec, httptest.NewRequest(http.MethodGet, "/engine", nil)) + if rec.Code != http.StatusServiceUnavailable { + t.Fatalf("before the first scrape: want 503, got %d %s", rec.Code, rec.Body.String()) + } + + scrape(w, time.Now(), vllmText) + rec = httptest.NewRecorder() + h(rec, httptest.NewRequest(http.MethodGet, "/engine", nil)) + var got map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatalf("body is not JSON: %v: %s", err, rec.Body.String()) + } + want := map[string]any{"engine_id": "pst-1791943641.29", "started_at": "2026-10-14T02:07:21.29Z", "source": "process_start_time", + "restarts": float64(0), "watcher_started_at": rfc(w.bornAt)} + if rec.Code != http.StatusOK || len(got) != len(want) { + t.Fatalf("want 200 with exactly %v, got %d %v", want, rec.Code, got) + } + for k, v := range want { + if got[k] != v { + t.Errorf("%s = %v, want %v", k, got[k], v) + } + } + + rec = httptest.NewRecorder() + h(rec, httptest.NewRequest(http.MethodPost, "/engine", nil)) + if rec.Code != http.StatusMethodNotAllowed { + t.Errorf("POST: want 405, got %d", rec.Code) + } +} + +// A scrape that lacks some startup series -- reported late, or a body cut off +// at the read limit -- is not a restart. Control: the same scrape with one value +// changed is, so the test cannot pass against an implementation that never +// detects sglang restarts. +func TestEngineSglangMissingSeriesIsNotRestart(t *testing.T) { + t0 := time.Now() + truncated := `sglang:startup_time_seconds{engine_type="unified",model_name="qwen",phase="load_weight"} 6.029548802413046 +sglang:realtime_tokens_total{mode="decode"} 0 +` + w := newWatcher() + scrape(w, t0, sglangFull) + id := view(w).EngineID + scrape(w, t0.Add(5*time.Second), truncated) + scrape(w, t0.Add(10*time.Second), sglangFull) // and back: still the same process + if e := view(w); e.EngineID != id || e.RestartsObserved != 0 { + t.Fatalf("missing series must not read as a restart: %+v", e) + } + + changed := `sglang:startup_time_seconds{engine_type="unified",model_name="qwen",phase="load_weight"} 5.98812301 +sglang:realtime_tokens_total{mode="decode"} 0 +` + ctrl := newWatcher() + scrape(ctrl, t0, sglangFull) + scrape(ctrl, t0.Add(5*time.Second), changed) + if e := view(ctrl); e.RestartsObserved != 1 { + t.Fatalf("control: a changed value in a partial scrape is a restart, got %+v", e) + } +} + +// Late phases join the remembered set, so a later change in one of them is +// still caught. +func TestEngineSglangLatePhaseChangeIsRestart(t *testing.T) { + t0 := time.Now() + w := newWatcher() + scrape(w, t0, sglangPartial) // only load_weight + scrape(w, t0.Add(5*time.Second), sglangFull) + late := `sglang:startup_time_seconds{engine_type="unified",model_name="qwen",phase="load_weight"} 6.029548802413046 +sglang:startup_time_seconds{engine_type="unified",model_name="qwen",phase="scheduler_e2e"} 99.5 +` + scrape(w, t0.Add(10*time.Second), late) + if e := view(w); e.RestartsObserved != 1 { + t.Fatalf("a phase first seen late and then changed is a restart: %+v", e) + } +} diff --git a/main.go b/main.go index 7458a66..ae7c573 100644 --- a/main.go +++ b/main.go @@ -211,9 +211,12 @@ func main() { } _, _ = io.WriteString(rw, "ok: "+reason) }) + // /engine: which engine instance this is and when it started, for clients + // that must notice a restart (see engine.go). + mux.HandleFunc("/engine", engineHandler(w)) srv := &http.Server{Addr: listen, Handler: mux, ReadHeaderTimeout: 5 * time.Second} go func() { - log.Printf("HTTP on %s (/healthz);engine=%s;config=%s", listen, engineURL, configFile) + log.Printf("HTTP on %s (/healthz, /engine);engine=%s;config=%s", listen, engineURL, configFile) if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { log.Fatalf("http: %v", err) } @@ -224,6 +227,7 @@ func main() { client := &http.Client{} var lastState string + var lastSource string // identity source last logged, see the /engine block below hot := loadHot(configFile, defaults) var lastMtime time.Time if fi, err := os.Stat(configFile); err == nil { @@ -299,9 +303,26 @@ func main() { if err == nil { snap = parseMetrics(text) } + restartsBefore := w.engineRestarts() w.step(time.Now(), snap, err, time.Duration(hot.StallSec)*time.Second, time.Duration(hot.WarmupSec)*time.Second, makeActiveProbe(client, engineURL, hot)) + if e, known := w.Engine(); known && e.Source != lastSource { + // Logged on change only. observed is the one worth a warning: its id + // changes when this sidecar restarts and misses an idle engine's + // restart, so a client relying on /engine should know. + lastSource = e.Source + if e.Source == idObserved { + log.Printf("警告:/engine 身份来源=observed(引擎 /metrics 里既无 process_start_time_seconds 也无 sglang startup_* 指标," + + "如 vLLM --api-server-count>1 的多进程 metrics);engine_id 在 hang-watcher 重启后会变,空闲引擎重启也看不出来") + } else { + log.Printf("/engine 身份来源=%s engine_id=%s", e.Source, e.EngineID) + } + } + if w.engineRestarts() != restartsBefore { + e, _ := w.Engine() + log.Printf("引擎重启:engine_id=%s started_at=%s(%s)", e.EngineID, e.StartedAt, e.Source) + } if hung, state, reason := w.Hung(); state != lastState { // 粗状态变化才打日志(reason 内数字每轮变,按 state 去重免刷屏) lastState = state if hung { diff --git a/smoke_engine_test.sh b/smoke_engine_test.sh new file mode 100644 index 0000000..5531163 --- /dev/null +++ b/smoke_engine_test.sh @@ -0,0 +1,94 @@ +#!/usr/bin/env bash +# End-to-end check of GET /engine against a fake engine (no Kubernetes): +# 1. before the first scrape -> 503 +# 2. engine up -> 200, {engine_id, started_at, source} +# 3. engine restarted -> new id +# 4. sidecar restarted, same engine -> same id (what consumers rely on) +# and, with no identity metrics at all, source=observed plus a warning in the log +# Run for both identity sources: vLLM process_start_time and sglang startup gauges. +set -uo pipefail +HERE="$(cd "$(dirname "$0")" && pwd)"; WORK="$(mktemp -d)" +EPORT=${EPORT:-18056}; HPORT=${HPORT:-19091} +# Only non-empty pids: the helpers below clear them, and `kill 0` would signal +# this script's own process group. +trap 'for p in ${FAKE:-} ${SC:-}; do kill "$p" 2>/dev/null; done; rm -rf "$WORK"' EXIT + +echo "== build =="; ( cd "$HERE" && CGO_ENABLED=0 go build -o "$WORK/hang-watcher" . ) || { echo BUILD_FAIL; exit 1; } + +# The fake engine serves $WORK/metrics.txt, re-read on every request, so the +# test "restarts" it by rewriting the file. +cat > "$WORK/fake.py" <<'PY' +import sys +from http.server import BaseHTTPRequestHandler, HTTPServer +class H(BaseHTTPRequestHandler): + def do_GET(self): + body = open(sys.argv[2], "rb").read() + self.send_response(200); self.end_headers(); self.wfile.write(body) + def log_message(self, *a): pass +HTTPServer(("127.0.0.1", int(sys.argv[1])), H).serve_forever() +PY +echo '{"poll_interval_sec":1,"stall_sec":30,"metrics_timeout_sec":3,"active_probe_enabled":false}' > "$WORK/config.json" + +start_sidecar() { + WARMUP_SEC=0 ENGINE_URL="http://127.0.0.1:$EPORT" LISTEN="127.0.0.1:$HPORT" CONFIG_FILE="$WORK/config.json" \ + "$WORK/hang-watcher" >> "$WORK/sc.log" 2>&1 & + SC=$! + for _ in $(seq 1 30); do curl -s -o /dev/null "http://127.0.0.1:$HPORT/engine" && return; sleep 0.2; done +} +stop_sidecar() { kill $SC 2>/dev/null; wait $SC 2>/dev/null; SC=; } +field() { python3 -c 'import json,sys; print(json.loads(sys.argv[1]).get(sys.argv[2]))' "$1" "$2"; } + +fail=0 +check() { # $1 description, $2 actual, $3 expected + if [ "$2" = "$3" ]; then echo " ok $1 = $2"; else echo " FAIL $1 = $2, want $3"; fail=1; fi +} + +run_case() { # $1 name, $2 first metrics, $3 restarted metrics, $4 expected source + echo "== $1 ==" + rm -f "$WORK/metrics.txt" + # 1. sidecar up, engine down + start_sidecar + code=$(curl -s -o /dev/null -w "%{http_code}" "http://127.0.0.1:$HPORT/engine") + check "before first scrape: status" "$code" 503 + # 2. engine up + printf '%b' "$2" > "$WORK/metrics.txt" + python3 "$WORK/fake.py" "$EPORT" "$WORK/metrics.txt" > /dev/null 2>&1 & + FAKE=$! + sleep 2.5 + b=$(curl -s "http://127.0.0.1:$HPORT/engine"); id1=$(field "$b" engine_id) + check "engine up: source" "$(field "$b" source)" "$4" + check "engine up: fields" "$(python3 -c 'import json,sys; print(",".join(sorted(json.loads(sys.argv[1]))))' "$b")" "engine_id,restarts,source,started_at,watcher_started_at" + # 3. engine restarted + printf '%b' "$3" > "$WORK/metrics.txt" + sleep 2.5 + b=$(curl -s "http://127.0.0.1:$HPORT/engine"); id2=$(field "$b" engine_id) + check "engine restarted: id changed" "$([ "$id1" != "$id2" ] && echo yes || echo no)" yes + check "engine restarted: restarts" "$(field "$b" restarts)" 1 + # 4. sidecar restarted, engine untouched + stop_sidecar; start_sidecar; sleep 2.5 + b=$(curl -s "http://127.0.0.1:$HPORT/engine") + check "sidecar restarted: same id" "$(field "$b" engine_id)" "$id2" + check "sidecar restarted: restarts counts from 0 again" "$(field "$b" restarts)" 0 + stop_sidecar; kill $FAKE 2>/dev/null; wait $FAKE 2>/dev/null; FAKE= +} + +run_case vllm \ + 'process_start_time_seconds 1.79194364129e+09\nvllm:generation_tokens_total 100\nvllm:num_requests_running 0\n' \ + 'process_start_time_seconds 1.79195001337e+09\nvllm:generation_tokens_total 5000\nvllm:num_requests_running 0\n' \ + process_start_time +run_case sglang \ + 'sglang:startup_time_seconds{phase="scheduler_e2e"} 109.03901622537524\nsglang:realtime_tokens_total 0\nsglang:num_running_reqs 0\n' \ + 'sglang:startup_time_seconds{phase="scheduler_e2e"} 108.71100912\nsglang:realtime_tokens_total 0\nsglang:num_running_reqs 0\n' \ + startup_fingerprint + +echo "== no identity metrics (observed) ==" +printf 'sglang:generation_tokens_total 1000\nsglang:num_running_reqs 0\n' > "$WORK/metrics.txt" +python3 "$WORK/fake.py" "$EPORT" "$WORK/metrics.txt" > /dev/null 2>&1 & +FAKE=$! +: > "$WORK/sc.log"; start_sidecar; sleep 2.5 +check "source" "$(field "$(curl -s "http://127.0.0.1:$HPORT/engine")" source)" observed +check "fallback warned in the log" "$(grep -c '警告:/engine 身份来源=observed' "$WORK/sc.log")" 1 +stop_sidecar; kill $FAKE 2>/dev/null; wait $FAKE 2>/dev/null; FAKE= + +if [ $fail = 0 ]; then echo "✅ 通过"; else echo "❌ 失败"; tail -20 "$WORK/sc.log"; fi +exit $fail diff --git a/watcher.go b/watcher.go index 184bdcc..7e67830 100644 --- a/watcher.go +++ b/watcher.go @@ -85,6 +85,12 @@ type metricsSnap struct { running float64 // 在途请求数 kv float64 // KV 水位 gauge(kvGaugeMetrics 求和),prefill 期间的进度信号 haveKV bool // 是否解析到任一 KV gauge(sglang 侧没有 → false,该腿自动不生效) + + // Engine instance identity, see engine.go. startTime is the unix time from + // process_start_time_seconds (vLLM exposes it; sglang does not), 0 if absent. + // startup holds sglang's startup_* gauges keyed by full series, nil if absent. + startTime float64 + startup map[string]float64 } // parseMetrics:解析 Prometheus 文本 → token_progress + running。 @@ -110,6 +116,15 @@ func parseMetrics(text string) metricsSnap { if err != nil { continue } + if name == processStartMetric { + s.startTime = val + } + if startupMetrics[name] { + if s.startup == nil { + s.startup = map[string]float64{} + } + s.startup[fullKey] = val + } if progressMetrics[name] { s.tp += val s.haveTP = true @@ -171,6 +186,10 @@ type watcher struct { // 每轮停滞都探,成功的探测会推进 progress 把停滞计时按回零,所以「停滞累计到 stall」 // 本身已经蕴含「这段时间里每次探测都失败」,不需要再拿这个计数器当门槛。 probeFails int + + // Which engine instance is being watched, served on /engine. See engine.go. + eng engineState + bornAt time.Time // when this sidecar started, reported on /engine } // setRecheck:注册「再取一次 token_progress」的回调。主动探测失败后用它复核 —— @@ -202,7 +221,7 @@ func (w *watcher) noteLogHits(now time.Time, hits int, err error) { } func newWatcher() *watcher { - return &watcher{state: "startup", reason: "启动中(未取到首个 metrics)"} + return &watcher{state: "startup", reason: "启动中(未取到首个 metrics)", bornAt: time.Now()} } func (w *watcher) set(hung bool, state, reason string) { @@ -263,6 +282,15 @@ func (w *watcher) Hung() (bool, string, string) { // running>0 停滞的误杀);打不通才坐实 hang。其它分支(空闲/不可达/出词)不触发,不额外加载。 func (w *watcher) step(now time.Time, m metricsSnap, fetchErr error, stall, warmup time.Duration, probe func() bool) { w.warmup = warmup + if w.noteEngine(now, m, fetchErr == nil) { + // The engine was replaced. Start over as if the sidecar had just come + // up: a fresh baseline, and the warm-up grace period from this scrape. + // Counter regression below covers only some restarts: it misses an + // engine that was idle (counters 0 before and after) or whose new + // process outran the old count within one poll. + w.started, w.haveBaseline, w.firstFail, w.probeFails = false, false, time.Time{}, 0 + w.set(false, "restart", "引擎实例变了("+w.eng.idSource+"),重置基线与 warmup") + } if fetchErr != nil { if !w.started { w.set(false, "startup", "启动中:/metrics 暂不可达(startupProbe 兜)") @@ -314,6 +342,7 @@ func (w *watcher) step(now time.Time, m metricsSnap, fetchErr error, stall, warm // and never be declared hung. A counter reset is hard evidence of a new // process; a flap is not. w.set(false, "regress", "计数器倒退(引擎重启过),重置基线与 warmup") + w.noteRegress(now) w.lastTP, w.lastGrow, w.probeFails = m.tp, now, 0 w.lastKV, w.haveKVBase = m.kv, m.haveKV w.readyAt = now