diff --git a/cmd/spinifex/cmd/admin_ochre_endpoint.go b/cmd/spinifex/cmd/admin_ochre_endpoint.go index ed98ee8ca..038255eaa 100644 --- a/cmd/spinifex/cmd/admin_ochre_endpoint.go +++ b/cmd/spinifex/cmd/admin_ochre_endpoint.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "os" + "strconv" "time" handlers_bedrock "github.com/mulgadc/spinifex/spinifex/handlers/bedrock" @@ -19,8 +20,11 @@ var ochreEndpointCmd = &cobra.Command{ Long: `Operator surface over the daemon's serving-endpoint lifecycle: request an endpoint for a staged model, inspect its state, and tear it down. -The gateway does not yet request endpoints itself, so until it does this is -the only way to start a serving VM.`, +The gateway requests endpoints on its own: an invoke for a model with no +endpoint returns ModelNotReadyException and launches one in the background, +and the daemon reclaims an endpoint that has been idle past its TTL. These +commands drive the same lifecycle by hand, for staging a model ahead of first +use or taking its GPU back early.`, } var ochreEndpointEnsureCmd = &cobra.Command{ @@ -164,6 +168,7 @@ func formatEndpointRecord(rec handlers_bedrock.EndpointRecord) string { rows = append(rows, [2]string{"Startup", rec.ReadyAt.Sub(rec.CreatedAt).Round(time.Second).String()}) } } + rows = append(rows, reclaimRows(rec)...) out := "" for _, row := range rows { @@ -172,6 +177,33 @@ func formatEndpointRecord(rec handlers_bedrock.EndpointRecord) string { return out } +// reclaimRows renders what the daemon's idle sweep last observed, so an +// operator asking why an endpoint was or was not reclaimed can see the inputs +// rather than infer them. Only meaningful for a READY endpoint: no other state +// is swept. +// +// "Idle for" is measured from the record's LastActive, which falls back to +// ReadyAt, so an endpoint that has been quiet since launch reads as idle since +// launch rather than since the zero time. +func reclaimRows(rec handlers_bedrock.EndpointRecord) [][2]string { + if rec.State != handlers_bedrock.StateReady { + return nil + } + rows := [][2]string{{"In flight", strconv.Itoa(rec.InFlight)}} + if since := rec.LastActive(); !since.IsZero() { + rows = append(rows, + [2]string{"Last active", since.Format(time.RFC3339)}, + [2]string{"Idle for", time.Since(since).Round(time.Second).String()}) + } + if rec.Pinned { + rows = append(rows, [2]string{"Pinned", "yes (never reclaimed or evicted)"}) + } + if rec.ScrapeFailures > 0 { + rows = append(rows, [2]string{"Scrape failures", strconv.Itoa(rec.ScrapeFailures)}) + } + return rows +} + // listEndpointsOutput renders 'ochre endpoint list'. Split from its Run // function so it is testable against a fake service with no NATS connection. func listEndpointsOutput(ctx context.Context, svc handlers_bedrock.EndpointService) (string, error) { diff --git a/cmd/spinifex/cmd/admin_ochre_endpoint_test.go b/cmd/spinifex/cmd/admin_ochre_endpoint_test.go index 5eac15551..3efd70cb1 100644 --- a/cmd/spinifex/cmd/admin_ochre_endpoint_test.go +++ b/cmd/spinifex/cmd/admin_ochre_endpoint_test.go @@ -365,3 +365,61 @@ func TestRunOchreEndpointDelete_ErrorExits1(t *testing.T) { code := withOchreExitCapture(t, func() { runOchreEndpointDelete(&cmd, nil) }) require.Equal(t, 1, code) } + +// The reclaim inputs are what an operator needs to answer "why was this +// endpoint reaped" or "why was it not", so a READY record must show them. +func TestFormatEndpointRecord_ShowsReclaimInputs(t *testing.T) { + ready := time.Now().UTC().Add(-30 * time.Minute) + out := formatEndpointRecord(handlers_bedrock.EndpointRecord{ + ModelID: testModelID, + State: handlers_bedrock.StateReady, + ReadyAt: ready, + LastActiveAt: ready.Add(20 * time.Minute), + InFlight: 3, + }) + + require.Contains(t, out, "In flight:") + require.Contains(t, out, "3") + require.Contains(t, out, "Last active:") + require.Contains(t, out, "Idle for:") +} + +// An endpoint quiet since launch is idle since launch, not since the zero +// time: the fallback is what keeps the reported figure meaningful. +func TestFormatEndpointRecord_NeverActiveFallsBackToReadyAt(t *testing.T) { + out := formatEndpointRecord(handlers_bedrock.EndpointRecord{ + ModelID: testModelID, + State: handlers_bedrock.StateReady, + ReadyAt: time.Now().UTC().Add(-90 * time.Second), + }) + + require.Contains(t, out, "Idle for:") + require.NotContains(t, out, "0001-01-01", "an unsampled record must not report the zero time") + require.NotContains(t, out, "Scrape failures", "a healthy endpoint must not carry a failure row") +} + +func TestFormatEndpointRecord_SurfacesPinnedAndScrapeFailures(t *testing.T) { + out := formatEndpointRecord(handlers_bedrock.EndpointRecord{ + ModelID: testModelID, + State: handlers_bedrock.StateReady, + ReadyAt: time.Now().UTC().Add(-time.Hour), + Pinned: true, + ScrapeFailures: 2, + }) + + require.Contains(t, out, "Pinned:") + require.Contains(t, out, "Scrape failures:") + require.Contains(t, out, "2") +} + +// Only READY endpoints are swept, so reclaim rows on any other state would be +// reporting a decision nothing is making. +func TestFormatEndpointRecord_NoReclaimRowsWhenNotReady(t *testing.T) { + out := formatEndpointRecord(handlers_bedrock.EndpointRecord{ + ModelID: testModelID, + State: handlers_bedrock.StateStarting, + }) + + require.NotContains(t, out, "In flight") + require.NotContains(t, out, "Idle for") +} diff --git a/spinifex/handlers/bedrock/capacity.go b/spinifex/handlers/bedrock/capacity.go index 6f8fdf832..8a8c1e41d 100644 --- a/spinifex/handlers/bedrock/capacity.go +++ b/spinifex/handlers/bedrock/capacity.go @@ -81,7 +81,7 @@ func selectEvictable(recs []EndpointRecord, nodeID string, minResidency time.Dur rec.NodeID != nodeID, now.Sub(rec.ReadyAt) < minResidency: continue } - if !found || lastActive(rec).Before(lastActive(victim)) { + if !found || rec.LastActive().Before(victim.LastActive()) { victim, found = rec, true } } diff --git a/spinifex/handlers/bedrock/reaper.go b/spinifex/handlers/bedrock/reaper.go index 126b3a811..076c8ab49 100644 --- a/spinifex/handlers/bedrock/reaper.go +++ b/spinifex/handlers/bedrock/reaper.go @@ -237,9 +237,12 @@ func (r *Reaper) sweepEndpoint(ctx context.Context, kv jetstream.KeyValue, rec E } if r.shouldReap(updated, now) { + // Seconds as a plain number, not a time.Duration: slog renders a Duration + // as int64 nanoseconds in JSON, which no reader parses at a glance and + // which reaches the metrics sink in a unit nothing charts. slog.InfoContext(ctx, "bedrock reaper: reclaiming an idle endpoint", "model", rec.ModelID, "instanceId", rec.InstanceID, - "idleFor", now.Sub(lastActive(updated)).Round(time.Second)) + "idleSeconds", int64(now.Sub(updated.LastActive()).Seconds())) if _, err := r.svc.Delete(ctx, &DeleteEndpointInput{ModelID: rec.ModelID}, utils.GlobalAccountID); err != nil { return fmt.Errorf("reap idle endpoint: %w", err) } @@ -258,7 +261,7 @@ func (r *Reaper) shouldReap(rec EndpointRecord, now time.Time) bool { return false } ttl := r.idleTTL() - return now.Sub(lastActive(rec)) >= ttl && now.Sub(rec.ReadyAt) >= ttl + return now.Sub(rec.LastActive()) >= ttl && now.Sub(rec.ReadyAt) >= ttl } // recordScrapeFailure counts an unknown sample and escalates a persistently @@ -317,13 +320,3 @@ func (r *Reaper) persistObservation(ctx context.Context, kv jetstream.KeyValue, } return nil } - -// lastActive is rec's LRU ordering key: the last sweep that saw it serving, -// falling back to when it became READY. The fallback covers a record written -// before its first sweep, which is a fresh endpoint, not a stale one. -func lastActive(rec EndpointRecord) time.Time { - if !rec.LastActiveAt.IsZero() { - return rec.LastActiveAt - } - return rec.ReadyAt -} diff --git a/spinifex/handlers/bedrock/reaper_test.go b/spinifex/handlers/bedrock/reaper_test.go index ed1d564d8..ae90a96e9 100644 --- a/spinifex/handlers/bedrock/reaper_test.go +++ b/spinifex/handlers/bedrock/reaper_test.go @@ -350,3 +350,14 @@ func TestReaper_NonLeaderSweepsNothing(t *testing.T) { assert.True(t, found, "a non-leader must not reap") assert.Zero(t, f.transport.scrapeCount()) } + +// LastActive is the one definition both the idle clock and the LRU eviction +// order read, so they cannot drift apart. +func TestEndpointRecord_LastActive(t *testing.T) { + ready := time.Now().UTC().Add(-time.Hour) + active := ready.Add(30 * time.Minute) + + assert.Equal(t, active, EndpointRecord{ReadyAt: ready, LastActiveAt: active}.LastActive()) + assert.Equal(t, ready, EndpointRecord{ReadyAt: ready}.LastActive(), + "an endpoint quiet since launch is idle since launch, not since the zero time") +} diff --git a/spinifex/handlers/bedrock/service.go b/spinifex/handlers/bedrock/service.go index 7381f7696..3fdd97f8f 100644 --- a/spinifex/handlers/bedrock/service.go +++ b/spinifex/handlers/bedrock/service.go @@ -257,7 +257,7 @@ func (s *Service) evictForCapacity(ctx context.Context, kv jetstream.KeyValue, w slog.InfoContext(ctx, "bedrock: evicting an idle endpoint to make room", "evicting", victim.ModelID, "instanceId", victim.InstanceID, "for", wantModelID, - "idleSince", lastActive(victim)) + "idleSince", victim.LastActive()) if _, err := s.Delete(ctx, &DeleteEndpointInput{ModelID: victim.ModelID}, utils.GlobalAccountID); err != nil { slog.ErrorContext(ctx, "bedrock: eviction failed; the pending launch keeps its capacity refusal", "evicting", victim.ModelID, "for", wantModelID, "err", err) diff --git a/spinifex/handlers/bedrock/state.go b/spinifex/handlers/bedrock/state.go index b3061deb8..b44a8fbfd 100644 --- a/spinifex/handlers/bedrock/state.go +++ b/spinifex/handlers/bedrock/state.go @@ -97,3 +97,15 @@ type EndpointRecord struct { // checks Generation before deciding whether a launch is still needed. Generation uint64 `json:"generation"` } + +// LastActive is the instant this endpoint was last known to be doing work: the +// last sweep that saw it serving, or when it became READY if no sweep ever +// has. Both the idle clock and the LRU eviction order are measured from here, +// so they cannot drift apart. The ReadyAt fallback is what stops an endpoint +// that has been quiet since launch reading as idle since the zero time. +func (r EndpointRecord) LastActive() time.Time { + if !r.LastActiveAt.IsZero() { + return r.LastActiveAt + } + return r.ReadyAt +}