From 9c6c23d677d0e51347757834e59560bdca486408 Mon Sep 17 00:00:00 2001 From: scttfrdmn <3011922+scttfrdmn@users.noreply.github.com> Date: Tue, 11 Aug 2026 11:29:22 -0700 Subject: [PATCH] feat(agent): completion webhook + always-on heartbeat tag (#497) Closes the caller-facing gap where waiting for a launched instance's workload meant polling an artifact against a pre-guessed wall-clock deadline (the flaw behind a real calque incident). --completion-webhook-url mirrors the existing --spot-webhook-url (#228) fire-once POST pattern, fired from checkCompletion before the grace-period sleep. Independently, spawn:last-heartbeat is now stamped on every monitor tick (throttled to 1/min) unconditionally, giving a poller a liveness signal that needs no guessed timeout at all. --- CHANGELOG.md | 19 +++++ cmd/launch_config.go | 5 ++ cmd/launch_flags.go | 44 +++++----- docs-gen/launch.md | 5 +- pkg/agent/agent.go | 149 +++++++++++++++++++++++++++------ pkg/agent/agent_test.go | 33 +++++++- pkg/agent/spot_webhook_test.go | 120 ++++++++++++++++++++++++++ pkg/aws/client.go | 7 ++ pkg/aws/tags.go | 12 +++ pkg/aws/tags_test.go | 28 +++++++ pkg/provider/ec2.go | 2 + pkg/provider/provider.go | 9 ++ 12 files changed, 381 insertions(+), 52 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f678795..39ba322 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,25 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added +- **`--completion-webhook-url`** (launch) and **`spawn:last-heartbeat`** EC2 tag + (#497): closes the caller-facing gap where waiting for a launched instance's + workload to finish meant polling an artifact against a pre-guessed + wall-clock deadline (the exact flaw behind a real calque incident — a run + still legitimately executing at 40 minutes had no way to be distinguished + from "stuck," other than the caller's own guess). `--completion-webhook-url` + makes spored POST a fire-once, best-effort notice (mirroring the existing + `--spot-webhook-url` from #228) when the on-instance completion sentinel + (`--completion-file`) is detected — before the grace-period sleep and + lifecycle action — so a caller can register its own webhook/queue target + instead of reinventing an S3-polling loop unaware of spored's own sentinel. + Shares `--webhook-correlation`/`--webhook-timeout` with the spot webhook. + Independently, `spawn:last-heartbeat` is now stamped with the current time + on every monitor tick (throttled to once/minute) regardless of + configuration — an always-on liveness signal a poller can check to tell + "still alive and ticking" from "hung" or "gone," without needing to guess a + timeout at all. + ### Added - **`spawn resume --max-concurrent-auto`** — the same quota-derived concurrency ceiling `spawn launch` gained in #492 (v0.99.0), now available diff --git a/cmd/launch_config.go b/cmd/launch_config.go index 13f0038..f11b892 100644 --- a/cmd/launch_config.go +++ b/cmd/launch_config.go @@ -236,6 +236,11 @@ func buildLaunchConfig(truffleInput *input.TruffleInput) (*aws.LaunchConfig, err config.WebhookCorrelation = webhookCorrelation config.WebhookTimeout = webhookTimeout } + if completionWebhookURL != "" { + config.CompletionWebhookURL = completionWebhookURL + config.WebhookCorrelation = webhookCorrelation + config.WebhookTimeout = webhookTimeout + } if onComplete != "" { config.OnComplete = onComplete } diff --git a/cmd/launch_flags.go b/cmd/launch_flags.go index 2f25b4e..8c0d52d 100644 --- a/cmd/launch_flags.go +++ b/cmd/launch_flags.go @@ -27,25 +27,26 @@ var ( keyPair string // Behavior - spot bool - spotMaxPrice string - useReservation bool - reservationID string - capacityBlock bool - hibernate bool - ttl string - idleTimeout string - hibernateOnIdle bool - onIdle string - preStop string - preStopTimeout string - spotWebhookURL string - webhookCorrelation string - webhookTimeout string - onComplete string - completionFile string - completionDelay string - sessionTimeout string + spot bool + spotMaxPrice string + useReservation bool + reservationID string + capacityBlock bool + hibernate bool + ttl string + idleTimeout string + hibernateOnIdle bool + onIdle string + preStop string + preStopTimeout string + spotWebhookURL string + completionWebhookURL string + webhookCorrelation string + webhookTimeout string + onComplete string + completionFile string + completionDelay string + sessionTimeout string // Meta name string @@ -228,8 +229,9 @@ func init() { launchCmd.Flags().StringVar(&preStop, "pre-stop", "", "Shell command to run on the instance before any lifecycle-triggered stop/terminate (e.g., \"aws s3 sync /results s3://bucket/\")") launchCmd.Flags().StringVar(&preStopTimeout, "pre-stop-timeout", "", "Max time to wait for --pre-stop command (default: 5m, spot: 90s)") launchCmd.Flags().StringVar(&spotWebhookURL, "spot-webhook-url", "", "On spot interruption, spored POSTs a fire-once, best-effort notice to this URL within the ~2-min window (off-node consumers; empty = disabled)") - launchCmd.Flags().StringVar(&webhookCorrelation, "webhook-correlation", "", "Opaque blob echoed verbatim in the spot-webhook payload so a consumer can correlate the event to its own record (never parsed by spawn)") - launchCmd.Flags().StringVar(&webhookTimeout, "webhook-timeout", "", "Hard cap on the spot-webhook POST so it can't eat the reclamation window (default: 2s)") + launchCmd.Flags().StringVar(&completionWebhookURL, "completion-webhook-url", "", "On workload completion (--completion-file detected), spored POSTs a fire-once, best-effort notice to this URL (spawn#497) — lets a caller wait on its own webhook/queue instead of polling an artifact against a pre-guessed deadline; empty = disabled") + launchCmd.Flags().StringVar(&webhookCorrelation, "webhook-correlation", "", "Opaque blob echoed verbatim in the spot-webhook/completion-webhook payload so a consumer can correlate the event to its own record (never parsed by spawn)") + launchCmd.Flags().StringVar(&webhookTimeout, "webhook-timeout", "", "Hard cap on the spot-webhook/completion-webhook POST so it can't eat the reclamation window or delay the completion action (default: 2s)") launchCmd.Flags().StringVar(&onComplete, "on-complete", "", "Action when workload signals completion: terminate, stop, hibernate. Use 'terminate' for batch/headless workloads — 'stop' leaves EBS (and any attached EIP) billing indefinitely, which is easy to forget in accounts without a hosted reaper") launchCmd.Flags().StringVar(&completionFile, "completion-file", "/tmp/SPAWN_COMPLETE", "File to watch for completion signal") launchCmd.Flags().StringVar(&completionDelay, "completion-delay", "30s", "Grace period after completion signal") diff --git a/docs-gen/launch.md b/docs-gen/launch.md index bbdb3d3..354a339 100644 --- a/docs-gen/launch.md +++ b/docs-gen/launch.md @@ -41,6 +41,7 @@ spawn launch [flags] | `--command` | | string | | Command to run on all instances (executed after spored setup) | | `--completion-delay` | | string | `30s` | Grace period after completion signal | | `--completion-file` | | string | `/tmp/SPAWN_COMPLETE` | File to watch for completion signal | +| `--completion-webhook-url` | | string | | On workload completion (--completion-file detected), spored POSTs a fire-once, best-effort notice to this URL (spawn#497) — lets a caller wait on its own webhook/queue instead of polling an artifact against a pre-guessed deadline; empty = disabled | | `--compliance-strict` | | bool | | Strict mode: fail on warnings (default: show warnings only) | | `--config` | | string | | Launch config YAML file (supports plugins: list) | | `--cost-limit` | | float64 | | Terminate/stop when compute spend reaches this amount in USD (compute cost only; 0 = disabled) | @@ -143,7 +144,7 @@ spawn launch [flags] | `--wait-for-ssh` | | bool | `true` | Wait until SSH is ready | | `--wait-timeout` | | string | | Timeout for --wait (e.g., 2h, 30m, 0=no timeout) | | `--wait` | | bool | | Wait for sweep/launch to complete (requires --detach) | -| `--webhook-correlation` | | string | | Opaque blob echoed verbatim in the spot-webhook payload so a consumer can correlate the event to its own record (never parsed by spawn) | -| `--webhook-timeout` | | string | | Hard cap on the spot-webhook POST so it can't eat the reclamation window (default: 2s) | +| `--webhook-correlation` | | string | | Opaque blob echoed verbatim in the spot-webhook/completion-webhook payload so a consumer can correlate the event to its own record (never parsed by spawn) | +| `--webhook-timeout` | | string | | Hard cap on the spot-webhook/completion-webhook POST so it can't eat the reclamation window or delay the completion action (default: 2s) | | `--yes` | `-y` | bool | | Auto-approve cost estimate (skip confirmation) | diff --git a/pkg/agent/agent.go b/pkg/agent/agent.go index 3ab5961..4b2e556 100644 --- a/pkg/agent/agent.go +++ b/pkg/agent/agent.go @@ -40,25 +40,26 @@ type Agent struct { // while other goroutines (FSx mount, spot monitor) read it concurrently, so // access goes through cfg()/setConfig() under configMu (#175). Don't read the // field directly from code that can run off the monitor goroutine. - config *provider.Config - configMu sync.RWMutex - dnsClient *dns.Client - dnsDomain string // DNS domain (e.g. "spore.host" or "prismcloud.host") - registry *registry.PeerRegistry - pluginRuntime *pluginruntime.Runtime - notifier *Notifier // Slack lifecycle notifications (nil if not configured) - startTime time.Time - lastActivityTime time.Time - preStopDone bool // guards against running pre-stop hook more than once - spotWebhookFired bool // fire-once guard for the spot-interruption webhook (#228); the spot monitor re-enters every 5s - prevCPUIdle int64 // /proc/stat idle jiffies at last getCPUUsage call - prevCPUTotal int64 // /proc/stat total jiffies at last getCPUUsage call - lastSessionTagWrite time.Time // throttle spawn:logged-in-count tag writes - lastComputeTagWrite time.Time // throttle spawn:compute-seconds tag writes - computeSecondsBase int64 // compute-seconds already accumulated before this spored start - prevNetRx int64 // /proc/net/dev RX bytes at last getNetworkBytes call - prevNetTx int64 // /proc/net/dev TX bytes at last getNetworkBytes call - idleWarned bool // send idle_warning notification only once + config *provider.Config + configMu sync.RWMutex + dnsClient *dns.Client + dnsDomain string // DNS domain (e.g. "spore.host" or "prismcloud.host") + registry *registry.PeerRegistry + pluginRuntime *pluginruntime.Runtime + notifier *Notifier // Slack lifecycle notifications (nil if not configured) + startTime time.Time + lastActivityTime time.Time + preStopDone bool // guards against running pre-stop hook more than once + spotWebhookFired bool // fire-once guard for the spot-interruption webhook (#228); the spot monitor re-enters every 5s + prevCPUIdle int64 // /proc/stat idle jiffies at last getCPUUsage call + prevCPUTotal int64 // /proc/stat total jiffies at last getCPUUsage call + lastSessionTagWrite time.Time // throttle spawn:logged-in-count tag writes + lastComputeTagWrite time.Time // throttle spawn:compute-seconds tag writes + lastHeartbeatTagWrite time.Time // throttle spawn:last-heartbeat tag writes (#497) + computeSecondsBase int64 // compute-seconds already accumulated before this spored start + prevNetRx int64 // /proc/net/dev RX bytes at last getNetworkBytes call + prevNetTx int64 // /proc/net/dev TX bytes at last getNetworkBytes call + idleWarned bool // send idle_warning notification only once // DCV auth token verifier (embedded HTTP server for seamless browser auth) dcvTokens map[string]string // token → username @@ -404,6 +405,14 @@ func (a *Agent) checkAndAct(ctx context.Context) { // 0b. Keep spawn:compute-seconds tag current (throttled to 5/min). a.writeComputeSecondsTag(ctx) + // 0c. Keep spawn:last-heartbeat current (throttled to 1/min, matching the + // production monitor interval — #497): an always-on liveness signal a + // caller can poll to tell "still alive and ticking" from "hung" (spored + // froze) or "gone" (terminated), independent of whatever completion + // artifact the workload itself writes. No opt-in flag, unlike the webhooks + // above — it costs nothing to a caller who never reads it. + a.writeHeartbeatTag(ctx) + // 1. Check for completion signal (HIGH PRIORITY) if a.config.OnComplete != "" { if a.checkCompletion(ctx) { @@ -616,6 +625,31 @@ func (a *Agent) writeSessionCountTag(ctx context.Context, count int) { }) } +// writeHeartbeatTag stamps spawn:last-heartbeat with the current time, throttled +// to once per minute (#497): an always-on liveness signal a caller can poll to +// distinguish "still alive and ticking" from "hung" (spored crashed/froze) or +// "gone" (instance terminated, tag no longer resolvable) — independent of +// whatever completion artifact the workload itself writes. Follows the +// writeSessionCountTag/writeComputeSecondsTag throttle pattern so tests can +// push lastHeartbeatTagWrite into the future to skip the real EC2 call. +func (a *Agent) writeHeartbeatTag(ctx context.Context) { + if time.Since(a.lastHeartbeatTagWrite) < time.Minute { + return + } + a.lastHeartbeatTagWrite = time.Now() + cfg, err := awsconfig.LoadDefaultConfig(ctx, awsconfig.WithRegion(a.identity.Region)) + if err != nil { + return + } + client := ec2.NewFromConfig(cfg) + _, _ = client.CreateTags(ctx, &ec2.CreateTagsInput{ + Resources: []string{a.identity.InstanceID}, + Tags: []ec2types.Tag{ + {Key: aws.String("spawn:last-heartbeat"), Value: aws.String(time.Now().UTC().Format(time.RFC3339))}, + }, + }) +} + // writeComputeSecondsTag persists the total compute seconds (base + current uptime) to an EC2 tag. // Throttle: every 1 minute for the first 10 minutes (fast feedback on fresh instances), // then every 5 minutes thereafter. @@ -1282,18 +1316,27 @@ func (a *Agent) emitSpotInterruptionWebhook(info *provider.InterruptionInfo) { EmittedAt: time.Now().UTC().Format(time.RFC3339), } + postWebhook("Spot webhook", cfg.SpotWebhookURL, timeout, payload) +} + +// postWebhook marshals payload and POSTs it to url, best-effort, time-boxed by +// timeout. Any failure — marshal, build, timeout, DNS, non-2xx — is logged at +// most and dropped; the caller (spot/completion webhook) must never block the +// lifecycle action it precedes on a slow or dead endpoint. label prefixes log +// lines so spored.log can distinguish which webhook fired. +func postWebhook(label, url string, timeout time.Duration, payload any) { body, err := json.Marshal(payload) if err != nil { - log.Printf("Spot webhook: marshal failed, dropping: %v", err) + log.Printf("%s: marshal failed, dropping: %v", label, err) return } ctx, cancel := context.WithTimeout(context.Background(), timeout) defer cancel() - req, err := http.NewRequestWithContext(ctx, "POST", cfg.SpotWebhookURL, bytes.NewReader(body)) + req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body)) if err != nil { - log.Printf("Spot webhook: request build failed, dropping: %v", err) + log.Printf("%s: request build failed, dropping: %v", label, err) return } req.Header.Set("Content-Type", "application/json") @@ -1301,15 +1344,65 @@ func (a *Agent) emitSpotInterruptionWebhook(info *provider.InterruptionInfo) { client := &http.Client{Timeout: timeout} resp, err := client.Do(req) if err != nil { - log.Printf("Spot webhook: POST to %s failed, dropping (best-effort): %v", cfg.SpotWebhookURL, err) + log.Printf("%s: POST to %s failed, dropping (best-effort): %v", label, url, err) return } defer resp.Body.Close() if resp.StatusCode >= 400 { - log.Printf("Spot webhook: endpoint %s returned %d, dropping (best-effort)", cfg.SpotWebhookURL, resp.StatusCode) + log.Printf("%s: endpoint %s returned %d, dropping (best-effort)", label, url, resp.StatusCode) return } - log.Printf("Spot webhook: notice POSTed to %s (action=%s)", cfg.SpotWebhookURL, info.Action) + log.Printf("%s: notice POSTed to %s", label, url) +} + +// completionWebhookPayload is the fixed, stable on-node fact-struct spored +// POSTs when the completion sentinel fires (#497) — the caller-facing signal +// that was previously only reachable by parsing spored's Slack notification or +// by the caller reinventing its own artifact-polling loop (as calque's +// WaitForSummary did). Mirrors spotWebhookPayload's shape: Correlation is the +// only caller-supplied field, echoed verbatim and never parsed. +type completionWebhookPayload struct { + Event string `json:"event"` // always "completion" + InstanceID string `json:"instance_id"` // + Region string `json:"region"` // + NameTag string `json:"name_tag,omitempty"` // spawn:name + OnComplete string `json:"on_complete"` // the configured action: terminate/stop/hibernate/exit + ComputeSeconds int64 `json:"compute_seconds"` // accumulated compute time + LastActivityTime string `json:"last_activity_time"` // RFC3339 + Correlation string `json:"correlation,omitempty"` // opaque caller blob, verbatim + EmittedAt string `json:"emitted_at"` // RFC3339, when spored sent this +} + +// emitCompletionWebhook POSTs the fixed payload to the launch-configured URL +// exactly once, best-effort, time-boxed by WebhookTimeout (default 2s) — the +// same fire-and-forget discipline as emitSpotInterruptionWebhook (#228), +// applied to the completion sentinel instead of a spot notice (#497). Called +// from checkCompletion BEFORE the grace-period sleep and lifecycle action, so +// a caller learns of completion as early as spored itself does. +func (a *Agent) emitCompletionWebhook(ctx context.Context) { + cfg := a.cfg() + if cfg == nil || cfg.CompletionWebhookURL == "" { + return // opt-in; empty URL = today's behavior + } + + timeout := cfg.WebhookTimeout + if timeout <= 0 { + timeout = 2 * time.Second + } + + payload := completionWebhookPayload{ + Event: "completion", + InstanceID: a.identity.InstanceID, + Region: a.identity.Region, + NameTag: a.identity.Name, + OnComplete: cfg.OnComplete, + ComputeSeconds: a.TotalComputeSeconds(), + LastActivityTime: a.lastActivityTime.UTC().Format(time.RFC3339), + Correlation: cfg.WebhookCorrelation, + EmittedAt: time.Now().UTC().Format(time.RFC3339), + } + + postWebhook("Completion webhook", cfg.CompletionWebhookURL, timeout, payload) } func (a *Agent) sendSpotInterruptionNotification(action, interruptTime string) { @@ -1358,6 +1451,12 @@ func (a *Agent) checkCompletion(ctx context.Context) bool { // Notify via Slack before the grace period a.notifier.Notify(ctx, "completion", "") + // Fire the optional off-node completion webhook (#497), same + // fire-and-forget discipline as the spot-interruption webhook (#228) — + // so a caller waiting on ITS OWN target learns of completion as early + // as spored itself does, before the grace-period sleep below. + a.emitCompletionWebhook(ctx) + // Warn users with grace period delay := a.config.CompletionDelay a.warnUsers(i18n.Tf("spawn.agent.workload_complete", map[string]interface{}{ diff --git a/pkg/agent/agent_test.go b/pkg/agent/agent_test.go index 8b3be70..1f9cb82 100644 --- a/pkg/agent/agent_test.go +++ b/pkg/agent/agent_test.go @@ -541,8 +541,9 @@ func TestMonitor_SpotDetectionDoesNotGateTicker(t *testing.T) { monitorInterval: 10 * time.Millisecond, // fast ticker for the test // Push tag-write throttles into the future so checkAndAct skips its // (real-AWS) CreateTags calls and the test stays hermetic. - lastSessionTagWrite: time.Now().Add(time.Hour), - lastComputeTagWrite: time.Now().Add(time.Hour), + lastSessionTagWrite: time.Now().Add(time.Hour), + lastComputeTagWrite: time.Now().Add(time.Hour), + lastHeartbeatTagWrite: time.Now().Add(time.Hour), } ctx, cancel := context.WithCancel(context.Background()) @@ -603,8 +604,9 @@ func TestCheckAndAct_ExpiredTTL_AlwaysTerminates(t *testing.T) { startTime: time.Now().Add(-2 * time.Hour), lastActivityTime: time.Now(), // Keep checkAndAct hermetic: skip the (real-AWS) tag-write calls. - lastSessionTagWrite: time.Now().Add(time.Hour), - lastComputeTagWrite: time.Now().Add(time.Hour), + lastSessionTagWrite: time.Now().Add(time.Hour), + lastComputeTagWrite: time.Now().Add(time.Hour), + lastHeartbeatTagWrite: time.Now().Add(time.Hour), } a.checkAndAct(context.Background()) @@ -620,6 +622,29 @@ func TestCheckAndAct_ExpiredTTL_AlwaysTerminates(t *testing.T) { } } +// TestWriteHeartbeatTag_Throttled verifies the spawn#497 throttle guard: a +// recent write must return immediately (no EC2 call attempted) rather than +// re-firing every tick. This mirrors the writeSessionCountTag/ +// writeComputeSecondsTag throttle pattern. A due write can't be asserted here +// without a real AWS call (same as its siblings, which have no direct test) — +// this only pins the "skip when recent" half, which is what keeps +// checkAndAct-driving tests hermetic. +func TestWriteHeartbeatTag_Throttled(t *testing.T) { + a := newTestAgent(t, nil) + a.lastHeartbeatTagWrite = time.Now() + + a.writeHeartbeatTag(context.Background()) + + if a.lastHeartbeatTagWrite.Equal(time.Time{}) { + t.Error("lastHeartbeatTagWrite unexpectedly reset to zero") + } + // Since it returned before reassigning lastHeartbeatTagWrite, the value must + // still be very recent (the "time.Now()" set right above), not overwritten. + if time.Since(a.lastHeartbeatTagWrite) > time.Second { + t.Error("writeHeartbeatTag did not return early on a recent write — throttle not honored") + } +} + func TestTailBuffer_RetainsLastBytes(t *testing.T) { tb := newTailBuffer(10) // Write more than the cap across multiple writes (mimics stdout+stderr teeing). diff --git a/pkg/agent/spot_webhook_test.go b/pkg/agent/spot_webhook_test.go index 9ae7099..7485749 100644 --- a/pkg/agent/spot_webhook_test.go +++ b/pkg/agent/spot_webhook_test.go @@ -1,10 +1,12 @@ package agent import ( + "context" "encoding/json" "io" "net/http" "net/http/httptest" + "os" "sync/atomic" "testing" "time" @@ -144,3 +146,121 @@ func TestSpotWebhook_FiresOnce(t *testing.T) { t.Errorf("webhook POSTed %d times across 3 entries; want exactly 1 (fire-once)", got) } } + +// TestEmitCompletionWebhook_PayloadAndEcho is the spawn#497 counterpart to +// TestEmitSpotInterruptionWebhook_PayloadAndEcho: with a configured URL spored +// POSTs the fixed completion fact-struct, echoing WebhookCorrelation verbatim +// and carrying the configured OnComplete action. +func TestEmitCompletionWebhook_PayloadAndEcho(t *testing.T) { + var gotBody []byte + var gotContentType string + done := make(chan struct{}) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotContentType = r.Header.Get("Content-Type") + gotBody, _ = io.ReadAll(r.Body) + close(done) + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + a := newTestAgent(t, &provider.Config{ + CompletionWebhookURL: srv.URL, + WebhookCorrelation: "calque-run-42", + WebhookTimeout: 2 * time.Second, + OnComplete: "terminate", + }) + a.identity.Name = "vllm-node-0" + + a.emitCompletionWebhook(context.Background()) + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("completion webhook endpoint never received the POST") + } + + if gotContentType != "application/json" { + t.Errorf("Content-Type = %q, want application/json", gotContentType) + } + var p map[string]any + if err := json.Unmarshal(gotBody, &p); err != nil { + t.Fatalf("payload not valid JSON: %v\n%s", err, gotBody) + } + if p["event"] != "completion" { + t.Errorf("event = %v, want completion", p["event"]) + } + if p["on_complete"] != "terminate" { + t.Errorf("on_complete = %v, want terminate", p["on_complete"]) + } + if p["correlation"] != "calque-run-42" { + t.Errorf("correlation = %v, want the verbatim echo", p["correlation"]) + } + if p["instance_id"] != "i-test123" { + t.Errorf("instance_id = %v", p["instance_id"]) + } + if p["name_tag"] != "vllm-node-0" { + t.Errorf("name_tag = %v", p["name_tag"]) + } +} + +// TestEmitCompletionWebhook_DisabledWhenNoURL verifies opt-in: an empty URL +// means today's behavior — nothing is sent. +func TestEmitCompletionWebhook_DisabledWhenNoURL(t *testing.T) { + var hits int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + atomic.AddInt32(&hits, 1) + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + a := newTestAgent(t, &provider.Config{}) // no CompletionWebhookURL + a.emitCompletionWebhook(context.Background()) + + if got := atomic.LoadInt32(&hits); got != 0 { + t.Errorf("completion webhook fired %d times with no URL configured; want 0", got) + } +} + +// TestEmitCompletionWebhook_BestEffortDrop verifies a dead/erroring endpoint +// never panics or blocks the caller — failure is silently dropped, mirroring +// the spot webhook's best-effort discipline. +func TestEmitCompletionWebhook_BestEffortDrop(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusInternalServerError) // 5xx → dropped + })) + defer srv.Close() + + a := newTestAgent(t, &provider.Config{CompletionWebhookURL: srv.URL, WebhookTimeout: time.Second}) + // Must simply return; a panic or hang fails the test. + a.emitCompletionWebhook(context.Background()) +} + +// TestCheckCompletion_FiresCompletionWebhook verifies checkCompletion actually +// wires emitCompletionWebhook into the completion path (not just that the +// function works in isolation) — the end-to-end guard for spawn#497. +func TestCheckCompletion_FiresCompletionWebhook(t *testing.T) { + var hits int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + atomic.AddInt32(&hits, 1) + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + f := t.TempDir() + "/SPAWN_COMPLETE" + if err := os.WriteFile(f, []byte{}, 0644); err != nil { + t.Fatalf("cannot create completion file: %v", err) + } + + a := newTestAgent(t, &provider.Config{ + OnComplete: "noop_test_action", // avoids the terminate/stop sleep + CompletionFile: f, + CompletionDelay: 0, + CompletionWebhookURL: srv.URL, + WebhookTimeout: 2 * time.Second, + }) + _ = a.checkCompletion(context.Background()) + + if got := atomic.LoadInt32(&hits); got != 1 { + t.Errorf("completion webhook POSTed %d times via checkCompletion; want exactly 1", got) + } +} diff --git a/pkg/aws/client.go b/pkg/aws/client.go index a3f224c..f30364d 100644 --- a/pkg/aws/client.go +++ b/pkg/aws/client.go @@ -183,6 +183,13 @@ type LaunchConfig struct { WebhookCorrelation string // opaque caller blob, echoed verbatim in the payload, never parsed WebhookTimeout string // hard cap on the POST (default 2s) so it can't eat the window + // Completion webhook (#497): an optional, fire-once best-effort POST spored + // emits when the on-instance completion sentinel (CompletionFile) is + // detected, so a caller can wait on ITS OWN webhook/queue instead of + // polling an artifact against a pre-guessed wall-clock deadline. Shares + // WebhookCorrelation/WebhookTimeout above with the spot webhook. + CompletionWebhookURL string // POST target; empty disables the webhook + // Completion signal settings OnComplete string // Action: terminate, stop, hibernate CompletionFile string // File path to watch (default: /tmp/SPAWN_COMPLETE) diff --git a/pkg/aws/tags.go b/pkg/aws/tags.go index b819899..c2d2ae6 100644 --- a/pkg/aws/tags.go +++ b/pkg/aws/tags.go @@ -208,6 +208,18 @@ func buildTags(config LaunchConfig, accountID, userARN, accountNameSlug string) } } + // Completion webhook (#497): only tagged when a URL is set (opt-in). Shares + // spawn:webhook-correlation/spawn:webhook-timeout with the spot webhook above. + if config.CompletionWebhookURL != "" { + tags = append(tags, types.Tag{Key: aws.String("spawn:completion-webhook-url"), Value: aws.String(config.CompletionWebhookURL)}) + if config.WebhookCorrelation != "" { + tags = append(tags, types.Tag{Key: aws.String("spawn:webhook-correlation"), Value: aws.String(config.WebhookCorrelation)}) + } + if config.WebhookTimeout != "" { + tags = append(tags, types.Tag{Key: aws.String("spawn:webhook-timeout"), Value: aws.String(config.WebhookTimeout)}) + } + } + // Record the instance's primary user so spored can run the pre-stop hook as // that user rather than root (#63). Tagged whenever known. if config.Username != "" { diff --git a/pkg/aws/tags_test.go b/pkg/aws/tags_test.go index 5292c91..becb6b6 100644 --- a/pkg/aws/tags_test.go +++ b/pkg/aws/tags_test.go @@ -119,6 +119,34 @@ func TestBuildTags_SpotWebhook(t *testing.T) { } } +// TestBuildTags_CompletionWebhook is the spawn#497 counterpart to +// TestBuildTags_SpotWebhook: the completion-webhook URL is tagged only when +// set (opt-in), sharing the same webhook-correlation/webhook-timeout tags. +func TestBuildTags_CompletionWebhook(t *testing.T) { + withURL := buildTags(LaunchConfig{ + Name: "t", + CompletionWebhookURL: "https://example.test/completion-hook", + WebhookCorrelation: "opaque-blob-42", + WebhookTimeout: "3s", + }, "123456789012", "arn:aws:iam::123456789012:user/test", "") + + if got := findTagValue(withURL, "spawn:completion-webhook-url"); got != "https://example.test/completion-hook" { + t.Errorf("spawn:completion-webhook-url = %q, want the URL", got) + } + if got := findTagValue(withURL, "spawn:webhook-correlation"); got != "opaque-blob-42" { + t.Errorf("spawn:webhook-correlation = %q, want the verbatim blob", got) + } + if got := findTagValue(withURL, "spawn:webhook-timeout"); got != "3s" { + t.Errorf("spawn:webhook-timeout = %q, want 3s", got) + } + + // No URL → the completion-webhook tag is not written (opt-in). + without := buildTags(LaunchConfig{Name: "t"}, "123456789012", "arn:aws:iam::123456789012:user/test", "") + if got := findTagValue(without, "spawn:completion-webhook-url"); got != "" { + t.Errorf("spawn:completion-webhook-url = %q, want empty when no URL is set", got) + } +} + func TestBuildTags_FSxMountPointDefault(t *testing.T) { config := LaunchConfig{ Name: "test-instance", diff --git a/pkg/provider/ec2.go b/pkg/provider/ec2.go index 1182fc2..1561965 100644 --- a/pkg/provider/ec2.go +++ b/pkg/provider/ec2.go @@ -546,6 +546,8 @@ func loadConfigFromEC2Tags(ctx context.Context, client *ec2.Client, instanceID s } case tagprefix.Tag("spot-webhook-url"): config.SpotWebhookURL = *tag.Value + case tagprefix.Tag("completion-webhook-url"): + config.CompletionWebhookURL = *tag.Value case tagprefix.Tag("webhook-correlation"): config.WebhookCorrelation = *tag.Value case tagprefix.Tag("webhook-timeout"): diff --git a/pkg/provider/provider.go b/pkg/provider/provider.go index 7ebb4ad..23bd597 100644 --- a/pkg/provider/provider.go +++ b/pkg/provider/provider.go @@ -51,6 +51,15 @@ type Config struct { WebhookCorrelation string // opaque caller blob, echoed verbatim, never parsed WebhookTimeout time.Duration // hard cap on the POST; zero = default (2s) + // Completion webhook (#497): a fire-once, best-effort POST spored emits when + // the on-instance completion sentinel (CompletionFile) is detected, so a + // caller (e.g. calque) can block on ITS OWN webhook/queue instead of polling + // an S3 artifact against a pre-guessed wall-clock deadline. Shares + // WebhookCorrelation/WebhookTimeout with the spot webhook (both are + // fire-once best-effort POSTs of an on-node fact-struct; a caller wanting + // different correlation/timeout per event type can use two launches). + CompletionWebhookURL string // POST target; empty = disabled (today's behavior) + // Ephemeral FSx (#194): when an FSx is created asynchronously alongside the // instance, the launch path tags spawn:fsx-pending= + spawn:fsx-mount-point. // spored polls until the filesystem is AVAILABLE, mounts it, then flips the