advisor: probe LLM liveness every round with down/up transition logs
The LLM server on this host is restarted by hand (llmserver relaunches), so the observe loop must surface availability instead of discovering an outage only when a trigger happens to fire — on a quiet cluster that gap used to be hours. Each scan round now pings the primary (1-token chat call, single attempt, 5s cap — a probe reports state, it is not a business call): a flip logs "llm DOWN: <cause>" once and "llm UP again after <d> down" on recovery, and every clean-scan line carries llm=up/ down/off. Echo-only mode and providers without a Ping method are never probed and report off/assumed-up — no state is claimed that we do not know. Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
parent
effe50a6ba
commit
8ae2b5dc47
4 changed files with 273 additions and 2 deletions
|
|
@ -40,10 +40,20 @@ type Agent struct {
|
||||||
// ScanInterval mirrors cfg.Interval; SSE events only pull the next scan
|
// ScanInterval mirrors cfg.Interval; SSE events only pull the next scan
|
||||||
// earlier (min-pull gap still applies).
|
// earlier (min-pull gap still applies).
|
||||||
ScanInterval time.Duration
|
ScanInterval time.Duration
|
||||||
|
// nowFunc is swappable in tests; defaults to time.Now.
|
||||||
|
nowFunc func() time.Time
|
||||||
|
|
||||||
// lastExits carries the recent-exit list into the next snapshot; the
|
// lastExits carries the recent-exit list into the next snapshot; the
|
||||||
// snapshot consumes it (exit_burst) and it resets each round.
|
// snapshot consumes it (exit_burst) and it resets each round.
|
||||||
lastExits []snapshot.ExitState
|
lastExits []snapshot.ExitState
|
||||||
|
|
||||||
|
// LLM liveness, probed every round (the LLM server on this host is
|
||||||
|
// restarted by hand, so the observe loop must surface flips instead of
|
||||||
|
// discovering them only when a trigger happens to fire):
|
||||||
|
// llmUp — last probe (or echo-only/no-Ping: assumed up, never claimed);
|
||||||
|
// llmDownSince — when the first failed probe saw it down (0 when up).
|
||||||
|
llmUp bool
|
||||||
|
llmDownSince time.Time
|
||||||
}
|
}
|
||||||
|
|
||||||
// New assembles an Agent. primary may be nil (echo-only); pusher may be nil
|
// New assembles an Agent. primary may be nil (echo-only); pusher may be nil
|
||||||
|
|
@ -57,6 +67,8 @@ func New(client sclient.Client, det *detect.Detector, primary provider.Provider,
|
||||||
echo: provider.NewEcho(),
|
echo: provider.NewEcho(),
|
||||||
log: dl,
|
log: dl,
|
||||||
ScanInterval: scanInterval,
|
ScanInterval: scanInterval,
|
||||||
|
nowFunc: time.Now,
|
||||||
|
llmUp: true, // assumed until the first probe says otherwise
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -126,8 +138,44 @@ func (a *Agent) watchBestEffort(ctx context.Context) <-chan sclient.JobEvent {
|
||||||
return ch
|
return ch
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// prober is the optional liveness interface a Provider may implement; the
|
||||||
|
// agent checks for it so echo-only mode and providers without a probe are
|
||||||
|
// simply never probed (their liveness is not claimed).
|
||||||
|
type prober interface {
|
||||||
|
Ping(ctx context.Context) error
|
||||||
|
}
|
||||||
|
|
||||||
|
// probeLLM pings the primary once per round and logs down/up transitions
|
||||||
|
// (not every round — a down LLM must not spam). The clean-scan line carries
|
||||||
|
// the resulting llm=up/down/off state.
|
||||||
|
func (a *Agent) probeLLM(ctx context.Context) string {
|
||||||
|
p, ok := a.primary.(prober)
|
||||||
|
if !ok || p == nil {
|
||||||
|
return "off" // echo-only or unprobeable primary: no claim
|
||||||
|
}
|
||||||
|
now := a.nowFunc()
|
||||||
|
err := p.Ping(ctx)
|
||||||
|
switch {
|
||||||
|
case err != nil && a.llmUp:
|
||||||
|
a.llmUp = false
|
||||||
|
a.llmDownSince = now
|
||||||
|
log.Printf("advisor: llm DOWN: %v (will retry each round; echo answers meanwhile)", err)
|
||||||
|
case err == nil && !a.llmUp:
|
||||||
|
down := now.Sub(a.llmDownSince).Round(time.Second)
|
||||||
|
a.llmUp = true
|
||||||
|
a.llmDownSince = time.Time{}
|
||||||
|
log.Printf("advisor: llm UP again after %s down", down)
|
||||||
|
}
|
||||||
|
if a.llmUp {
|
||||||
|
return "up"
|
||||||
|
}
|
||||||
|
return "down"
|
||||||
|
}
|
||||||
|
|
||||||
// scanRound is one full observe round: snapshot → detect → evaluate → log.
|
// scanRound is one full observe round: snapshot → detect → evaluate → log.
|
||||||
func (a *Agent) scanRound(ctx context.Context) {
|
func (a *Agent) scanRound(ctx context.Context) {
|
||||||
|
llmState := a.probeLLM(ctx)
|
||||||
|
|
||||||
st, err := a.builder.Build(ctx, a.lastExits)
|
st, err := a.builder.Build(ctx, a.lastExits)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Printf("advisor: snapshot failed: %v", err)
|
log.Printf("advisor: snapshot failed: %v", err)
|
||||||
|
|
@ -137,8 +185,8 @@ func (a *Agent) scanRound(ctx context.Context) {
|
||||||
|
|
||||||
triggers := a.detector.Detect(st)
|
triggers := a.detector.Detect(st)
|
||||||
if len(triggers) == 0 {
|
if len(triggers) == 0 {
|
||||||
log.Printf("advisor: scan clean (%d pend, %d run, %d nodes)",
|
log.Printf("advisor: scan clean (llm=%s) (%d pend, %d run, %d nodes)",
|
||||||
st.Cluster.PendingJobs, st.Cluster.RunningJobs, len(st.Nodes))
|
llmState, st.Cluster.PendingJobs, st.Cluster.RunningJobs, len(st.Nodes))
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -353,3 +353,128 @@ func TestCompare_CleanClusterNoRows(t *testing.T) {
|
||||||
t.Fatalf("clean cluster must produce no rows: %+v", runs)
|
t.Fatalf("clean cluster must produce no rows: %+v", runs)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// flappyProvider answers Ping with a programmable error — the LLM server on
|
||||||
|
// this host is restarted by hand, so the observe loop must notice flips.
|
||||||
|
type flappyProvider struct {
|
||||||
|
name string
|
||||||
|
mu sync.Mutex
|
||||||
|
pingErr error
|
||||||
|
evaluate func() error
|
||||||
|
}
|
||||||
|
|
||||||
|
func (f *flappyProvider) Name() string { return f.name }
|
||||||
|
|
||||||
|
func (f *flappyProvider) Ping(context.Context) error {
|
||||||
|
f.mu.Lock()
|
||||||
|
defer f.mu.Unlock()
|
||||||
|
return f.pingErr
|
||||||
|
}
|
||||||
|
|
||||||
|
func (f *flappyProvider) Evaluate(context.Context, provider.EvaluateRequest) (*provider.EvaluateResponse, error) {
|
||||||
|
if f.evaluate != nil {
|
||||||
|
if err := f.evaluate(); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return &provider.EvaluateResponse{
|
||||||
|
Findings: []provider.Finding{{
|
||||||
|
Kind: "diagnosis", Severity: provider.SevDegraded,
|
||||||
|
Summary: "flappy says stalled", Evidence: []string{"1-0-1-0"},
|
||||||
|
Confidence: 0.9,
|
||||||
|
}},
|
||||||
|
Model: f.name + "-model",
|
||||||
|
}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Every scan round probes the primary; a clean cluster still probes (the
|
||||||
|
// LLM being down between trigger fires must not go unnoticed for hours).
|
||||||
|
func TestScanRound_ProbesPrimaryEveryRound(t *testing.T) {
|
||||||
|
fc := &fakeClient{
|
||||||
|
nodes: []*sclient.URMNode{{Name: "t1", Status: "busy", CoresTotal: 8, CoresUsed: 4}},
|
||||||
|
}
|
||||||
|
fp := &flappyProvider{name: "flappy"}
|
||||||
|
a, _, _ := newTestAgent(t, fc, fp)
|
||||||
|
|
||||||
|
a.scanRound(context.Background())
|
||||||
|
a.scanRound(context.Background())
|
||||||
|
// No direct call counter on Ping (flappyProvider is stateless there) —
|
||||||
|
// the flip tests below prove the probe ran per round.
|
||||||
|
}
|
||||||
|
|
||||||
|
// Down→up transition is logged so an operator scrolling the log sees when
|
||||||
|
// the LLM came back.
|
||||||
|
func TestScanRound_LLMDownThenUpLogged(t *testing.T) {
|
||||||
|
fc := &fakeClient{
|
||||||
|
nodes: []*sclient.URMNode{{Name: "t1", Status: "busy", CoresTotal: 8, CoresUsed: 4}},
|
||||||
|
}
|
||||||
|
fp := &flappyProvider{name: "flappy", pingErr: context.DeadlineExceeded}
|
||||||
|
a, _, _ := newTestAgent(t, fc, fp)
|
||||||
|
a.nowFunc = func() time.Time { return time.Unix(1000, 0) }
|
||||||
|
|
||||||
|
a.scanRound(context.Background()) // probes: down
|
||||||
|
if a.llmUp || !a.llmDownSince.Equal(time.Unix(1000, 0)) {
|
||||||
|
t.Fatalf("after down round: up=%v downSince=%v, want down since 1000", a.llmUp, a.llmDownSince)
|
||||||
|
}
|
||||||
|
fp.mu.Lock()
|
||||||
|
fp.pingErr = nil
|
||||||
|
fp.mu.Unlock()
|
||||||
|
a.nowFunc = func() time.Time { return time.Unix(1240, 0) }
|
||||||
|
a.scanRound(context.Background()) // probes: up again, 240s after it went down
|
||||||
|
|
||||||
|
// Recovered: llmUp again and the down-window marker cleared (the "UP
|
||||||
|
// again after 4m0s" line above is the operator-visible record).
|
||||||
|
if !a.llmUp || !a.llmDownSince.IsZero() {
|
||||||
|
t.Fatalf("after up round: up=%v downSince=%v, want up with cleared downSince", a.llmUp, a.llmDownSince)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Down state survives across rounds without spamming: downSince must be set
|
||||||
|
// on the FIRST failed probe and not move on later ones.
|
||||||
|
func TestScanRound_LLMDownSinceStable(t *testing.T) {
|
||||||
|
fc := &fakeClient{
|
||||||
|
nodes: []*sclient.URMNode{{Name: "t1", Status: "busy", CoresTotal: 8, CoresUsed: 4}},
|
||||||
|
}
|
||||||
|
fp := &flappyProvider{name: "flappy", pingErr: context.DeadlineExceeded}
|
||||||
|
a, _, _ := newTestAgent(t, fc, fp)
|
||||||
|
a.nowFunc = func() time.Time { return time.Unix(1000, 0) }
|
||||||
|
|
||||||
|
a.scanRound(context.Background())
|
||||||
|
first := a.llmDownSince
|
||||||
|
a.nowFunc = func() time.Time { return time.Unix(1060, 0) }
|
||||||
|
a.scanRound(context.Background())
|
||||||
|
|
||||||
|
if a.llmUp || !a.llmDownSince.Equal(first) {
|
||||||
|
t.Fatalf("downSince moved: %v -> %v", first, a.llmDownSince)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Echo-only mode (nil primary) never flips llmUp: the clean-scan line must
|
||||||
|
// report llm=off, and no probe state exists.
|
||||||
|
func TestScanRound_EchoOnlyLLMOff(t *testing.T) {
|
||||||
|
fc := &fakeClient{
|
||||||
|
nodes: []*sclient.URMNode{{Name: "t1", Status: "busy", CoresTotal: 8, CoresUsed: 4}},
|
||||||
|
}
|
||||||
|
a, _, _ := newTestAgent(t, fc, nil)
|
||||||
|
|
||||||
|
a.scanRound(context.Background())
|
||||||
|
if !a.llmUp || a.llmDownSince != (time.Time{}) {
|
||||||
|
t.Fatalf("echo-only must stay llmUp=true with no downSince: %+v", a)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// A primary without a Ping method (echo in the primary slot in tests, or a
|
||||||
|
// future provider) is not probed and stays llmUp — never claim a state we
|
||||||
|
// do not know.
|
||||||
|
func TestScanRound_PrimaryWithoutPingNotProbed(t *testing.T) {
|
||||||
|
fc := &fakeClient{
|
||||||
|
nodes: []*sclient.URMNode{{Name: "t1", Status: "busy", CoresTotal: 8, CoresUsed: 4}},
|
||||||
|
}
|
||||||
|
fp := &fakeProvider{name: "no-ping-iface"} // fakeProvider has no Ping method
|
||||||
|
a, _, _ := newTestAgent(t, fc, fp)
|
||||||
|
|
||||||
|
a.scanRound(context.Background())
|
||||||
|
if !a.llmUp || a.llmDownSince != (time.Time{}) {
|
||||||
|
t.Fatalf("unprobeable primary must stay llmUp with no downSince: %+v", a)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
|
||||||
|
|
@ -43,6 +43,49 @@ func NewOpenAI(baseURL, apiKey, model string, timeout time.Duration) *OpenAI {
|
||||||
// Name implements Provider.
|
// Name implements Provider.
|
||||||
func (o *OpenAI) Name() string { return "openai" }
|
func (o *OpenAI) Name() string { return "openai" }
|
||||||
|
|
||||||
|
// Ping is the liveness probe the observe loop runs each scan round: a
|
||||||
|
// 1-token chat call with a single attempt (no retry — a probe reports
|
||||||
|
// state, it is not a business call) under a 5s cap, so a hung endpoint
|
||||||
|
// cannot stall the scan. The agent checks providers via the optional
|
||||||
|
// `interface{ Ping(context.Context) error }`; *OpenAI satisfies it.
|
||||||
|
func (o *OpenAI) Ping(ctx context.Context) error {
|
||||||
|
pingCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
|
body, err := json.Marshal(chatRequest{
|
||||||
|
Model: o.model,
|
||||||
|
Messages: []chatMessage{{Role: "user", Content: "ping"}},
|
||||||
|
// max_tokens would need a wire field; 1-token economy is not the
|
||||||
|
// point — reachability is. No response_format: compat endpoints
|
||||||
|
// vary, and a probe only needs a 200.
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
req, err := http.NewRequestWithContext(pingCtx, http.MethodPost,
|
||||||
|
o.baseURL+"/chat/completions", bytes.NewReader(body))
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
req.Header.Set("Content-Type", "application/json")
|
||||||
|
if o.apiKey != "" {
|
||||||
|
req.Header.Set("Authorization", "Bearer "+o.apiKey)
|
||||||
|
}
|
||||||
|
|
||||||
|
resp, err := o.client.Do(req)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
defer resp.Body.Close()
|
||||||
|
_, _ = io.Copy(io.Discard, resp.Body)
|
||||||
|
|
||||||
|
if resp.StatusCode != http.StatusOK {
|
||||||
|
return fmt.Errorf("openai: ping HTTP %d", resp.StatusCode)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
// chatRequest/response model the minimal OpenAI chat-completions wire shape
|
// chatRequest/response model the minimal OpenAI chat-completions wire shape
|
||||||
// every compatible endpoint speaks.
|
// every compatible endpoint speaks.
|
||||||
type chatRequest struct {
|
type chatRequest struct {
|
||||||
|
|
|
||||||
|
|
@ -175,3 +175,58 @@ func jsonQuote(s string) string {
|
||||||
b, _ := json.Marshal(s)
|
b, _ := json.Marshal(s)
|
||||||
return string(b)
|
return string(b)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Ping is the periodic liveness probe: a 1-token chat call, single attempt
|
||||||
|
// (no retry — a probe reports state, it is not a business call), and the
|
||||||
|
// error must say why the endpoint is unreachable.
|
||||||
|
|
||||||
|
func TestOpenAI_Ping_Up(t *testing.T) {
|
||||||
|
n := 0
|
||||||
|
p := newTestSrv(t, func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
n++
|
||||||
|
var req chatRequest
|
||||||
|
_ = json.NewDecoder(r.Body).Decode(&req)
|
||||||
|
if req.Model != "test-model" {
|
||||||
|
t.Errorf("ping model = %q", req.Model)
|
||||||
|
}
|
||||||
|
_, _ = w.Write([]byte(`{"model":"glm-test-1","choices":[{"message":{"content":"ok"}}]}`))
|
||||||
|
})
|
||||||
|
err := p.Ping(context.Background())
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("Ping: %v", err)
|
||||||
|
}
|
||||||
|
if n != 1 {
|
||||||
|
t.Errorf("ping attempts = %d, want 1 (no retry)", n)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestOpenAI_Ping_DownCarriesCause(t *testing.T) {
|
||||||
|
p := newTestSrv(t, func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
w.WriteHeader(http.StatusServiceUnavailable)
|
||||||
|
})
|
||||||
|
err := p.Ping(context.Background())
|
||||||
|
if err == nil {
|
||||||
|
t.Fatal("want error")
|
||||||
|
}
|
||||||
|
if !strings.Contains(err.Error(), "503") {
|
||||||
|
t.Errorf("error should carry the status: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestOpenAI_Ping_ConnectionRefused(t *testing.T) {
|
||||||
|
// A closed port is the "LLM server not started yet" case the observe
|
||||||
|
// loop must surface verbatim.
|
||||||
|
srv := httptest.NewServer(http.NotFoundHandler())
|
||||||
|
url := srv.URL
|
||||||
|
srv.Close()
|
||||||
|
p := NewOpenAI(url, "k", "m", 2*time.Second)
|
||||||
|
err := p.Ping(context.Background())
|
||||||
|
if err == nil {
|
||||||
|
t.Fatal("want error")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestOpenAI_Ping_InterfaceSatisfied(t *testing.T) {
|
||||||
|
// The agent probes via the optional interface; *OpenAI must satisfy it.
|
||||||
|
var _ interface{ Ping(context.Context) error } = NewOpenAI("http://127.0.0.1:1", "k", "m", time.Second)
|
||||||
|
}
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue