From 235403e309949eb4f8b2a8cb09f26d1e4f75c980 Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Tue, 18 Aug 2026 13:30:25 +1000 Subject: [PATCH] fix: join the caller's trace on the EBS hop and name the account (#823) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ebsRequestWithTrace started its span from context.Background(), so every ebs.mount, ebs.unmount and ebs.delete rooted a trace of its own. Block storage work was therefore unreachable from the request that caused it: viperblockd recorded 10,209 transactions in 24h and not one could be tied to a caller or an account. VolumeMounter now takes the caller's context. Launch and hot attach/detach have one to give. Shutdown, crash recovery and teardown genuinely do not — they answer to no request — so they pass a background context, which is an honest root rather than a missing one. The owning account rides both the span and the NATS header, because viperblockd's consumer span reads that header to attribute itself. It comes off the instance where there is one, and is passed explicitly for a lone volume request, which carries no instance to read it from. An absent account is left off rather than blanked: recovery and teardown act for nobody, and crediting that work to whoever happens to be admin would read as tenant activity. --- spinifex/daemon/daemon_test.go | 16 +-- spinifex/daemon/vm_adapters.go | 58 ++++++---- .../daemon/vm_adapters_invariants_test.go | 2 +- spinifex/daemon/vm_adapters_test.go | 8 +- spinifex/daemon/vm_adapters_trace_test.go | 104 +++++++++++++++++- .../ec2/instance/service_impl_test.go | 8 +- spinifex/vm/crash_recovery.go | 2 +- spinifex/vm/lifecycle.go | 4 +- spinifex/vm/manager_deps_test.go | 9 +- spinifex/vm/restore_test.go | 8 +- spinifex/vm/shutdown.go | 2 +- spinifex/vm/volume_mounter.go | 22 +++- spinifex/vm/volumes.go | 20 ++-- 13 files changed, 194 insertions(+), 69 deletions(-) diff --git a/spinifex/daemon/daemon_test.go b/spinifex/daemon/daemon_test.go index e0ddfbd25..3c553d0ff 100644 --- a/spinifex/daemon/daemon_test.go +++ b/spinifex/daemon/daemon_test.go @@ -3310,7 +3310,7 @@ func TestVolumeMounterAdapter_UnmountOne_Success(t *testing.T) { require.NoError(t, err) defer sub.Unsubscribe() - adapter.UnmountOne(types.EBSRequest{ + adapter.UnmountOne(t.Context(), "", types.EBSRequest{ Name: "vol-rollback-test", DeviceName: "/dev/sdf", }) @@ -3340,7 +3340,7 @@ func TestVolumeMounterAdapter_UnmountOne_UnmountError(t *testing.T) { require.NoError(t, err) defer sub.Unsubscribe() - require.Error(t, adapter.UnmountOne(types.EBSRequest{Name: "vol-rollback-err"}), + require.Error(t, adapter.UnmountOne(t.Context(), "", types.EBSRequest{Name: "vol-rollback-err"}), "an ebs.unmount error response must propagate so DetachVolume can keep the volume attached") } @@ -3360,7 +3360,7 @@ func TestVolumeMounterAdapter_UnmountOne_StillMounted(t *testing.T) { require.NoError(t, err) defer sub.Unsubscribe() - require.Error(t, adapter.UnmountOne(types.EBSRequest{Name: "vol-still-mounted"}), + require.Error(t, adapter.UnmountOne(t.Context(), "", types.EBSRequest{Name: "vol-still-mounted"}), "a still-mounted response must propagate as an error") } @@ -3374,7 +3374,7 @@ func TestVolumeMounterAdapter_UnmountOne_RequestFailure(t *testing.T) { nc.Close() adapter := newVolumeMounterAdapter(nc, "node-1", nil) - require.Error(t, adapter.UnmountOne(types.EBSRequest{Name: "vol-timeout"}), + require.Error(t, adapter.UnmountOne(t.Context(), "", types.EBSRequest{Name: "vol-timeout"}), "a failed ebs.unmount request must propagate as an error") } @@ -3429,7 +3429,7 @@ func TestVolumeMounterAdapter_Unmount_SealFailureSkipsAvailable(t *testing.T) { Requests: []types.EBSRequest{{Name: "vol-fail"}, {Name: "vol-ok"}}, }, } - require.NoError(t, adapter.Unmount(inst), + require.NoError(t, adapter.Unmount(t.Context(), inst), "bulk Unmount must tolerate a per-volume seal failure so terminate stays idempotent") calls := volState.snapshot() @@ -3472,7 +3472,7 @@ func TestVolumeMounterAdapter_Unmount_NotFoundFlipsToAvailable(t *testing.T) { }, }, } - require.NoError(t, adapter.Unmount(inst)) + require.NoError(t, adapter.Unmount(t.Context(), inst)) calls := volState.snapshot() assert.Contains(t, calls, "vol-data-retry:available", @@ -3582,7 +3582,7 @@ func TestVolumeMounterAdapter_Mount_PartialFailureRollback(t *testing.T) { }, } - err = adapter.Mount(instance) + err = adapter.Mount(t.Context(), instance) require.Error(t, err, "Mount should propagate the vol-2 failure") assert.Contains(t, err.Error(), tt.wantErrSub) @@ -3640,7 +3640,7 @@ func TestVolumeMounterAdapter_Mount_RollbackFailurePropagates(t *testing.T) { }, } - err = adapter.Mount(instance) + err = adapter.Mount(t.Context(), instance) require.Error(t, err) assert.Contains(t, err.Error(), "primary mount failure", "original mount error must be preserved") diff --git a/spinifex/daemon/vm_adapters.go b/spinifex/daemon/vm_adapters.go index 73b36d32a..ae0b0b6ee 100644 --- a/spinifex/daemon/vm_adapters.go +++ b/spinifex/daemon/vm_adapters.go @@ -112,19 +112,26 @@ func (a *volumeMounterAdapter) topic(action string) string { return fmt.Sprintf("ebs.%s.%s", a.node, action) } -// ebsRequestWithTrace sends an ebs.* NATS request, opening a client span and -// injecting it into the message headers so viperblockd's consumer span +// ebsRequestWithTrace sends an ebs.* NATS request, opening a client span under +// ctx and injecting it into the message headers so viperblockd's consumer span // (utils.StartConsumerSpan) joins this trace instead of rooting a new one. // -// VolumeMounter carries no caller context, so each call opens its own trace -// here rather than threading one through the interface. -func ebsRequestWithTrace(nc *nats.Conn, subject string, data []byte, timeout time.Duration) (msg *nats.Msg, err error) { - ctx, span := otel.Tracer(daemonTracerName).Start(context.Background(), "NATS "+subject, +// accountID names the owner on the span and on the header viperblockd reads, +// so block-storage work is attributable to a tenant. Empty is left off: a +// sweep or a recovery belongs to nobody, and crediting it to whoever happens +// to be admin would be worse than leaving it blank. +func ebsRequestWithTrace(ctx context.Context, nc *nats.Conn, accountID, subject string, data []byte, timeout time.Duration) (msg *nats.Msg, err error) { + attrs := []attribute.KeyValue{ + attribute.String("messaging.system", "nats"), + attribute.String("messaging.destination.name", subject), + } + if accountID != "" { + attrs = append(attrs, attribute.String(utils.AttrAccountID, accountID)) + } + + ctx, span := otel.Tracer(daemonTracerName).Start(ctx, "NATS "+subject, trace.WithSpanKind(trace.SpanKindClient), - trace.WithAttributes( - attribute.String("messaging.system", "nats"), - attribute.String("messaging.destination.name", subject), - )) + trace.WithAttributes(attrs...)) defer func() { if err != nil { span.RecordError(err) @@ -136,12 +143,15 @@ func ebsRequestWithTrace(nc *nats.Conn, subject string, data []byte, timeout tim reqMsg := nats.NewMsg(subject) reqMsg.Data = data utils.InjectTraceContext(ctx, reqMsg.Header) + if accountID != "" { + reqMsg.Header.Set(utils.AccountIDHeader, accountID) + } msg, err = nc.RequestMsg(reqMsg, timeout) return msg, err } -func (a *volumeMounterAdapter) Mount(instance *vm.VM) error { +func (a *volumeMounterAdapter) Mount(ctx context.Context, instance *vm.VM) error { instance.EBSRequests.Mu.Lock() defer instance.EBSRequests.Mu.Unlock() @@ -150,7 +160,7 @@ func (a *volumeMounterAdapter) Mount(instance *vm.VM) error { var rbErrs []error for _, idx := range mounted { req := instance.EBSRequests.Requests[idx] - if err := a.unmountOne(req); err != nil { + if err := a.unmountOne(ctx, instance.AccountID, req); err != nil { slog.Error("Mount rollback: unmount failed", "volume", req.Name, "err", err) rbErrs = append(rbErrs, fmt.Errorf("unmount %s: %w", req.Name, err)) @@ -169,7 +179,7 @@ func (a *volumeMounterAdapter) Mount(instance *vm.VM) error { return rollback(err) } - reply, err := ebsRequestWithTrace(a.nc, a.topic("mount"), ebsMountRequest, 30*time.Second) + reply, err := ebsRequestWithTrace(ctx, a.nc, instance.AccountID, a.topic("mount"), ebsMountRequest, 30*time.Second) slog.Info("Mounting volume", "Vol", v.Name, "NBDURI", v.NBDURI) @@ -205,7 +215,7 @@ func (a *volumeMounterAdapter) Mount(instance *vm.VM) error { return nil } -func (a *volumeMounterAdapter) Unmount(instance *vm.VM) error { +func (a *volumeMounterAdapter) Unmount(ctx context.Context, instance *vm.VM) error { instance.EBSRequests.Mu.Lock() defer instance.EBSRequests.Mu.Unlock() @@ -222,7 +232,7 @@ func (a *volumeMounterAdapter) Unmount(instance *vm.VM) error { // local WAL would find no checkpoint (bad superblock). On terminate the // volume is deleted regardless; on stop it stays attached/retryable. sealed := true - msg, err := ebsRequestWithTrace(a.nc, a.topic("unmount"), ebsUnMountRequest, unmountSealTimeout) + msg, err := ebsRequestWithTrace(ctx, a.nc, instance.AccountID, a.topic("unmount"), ebsUnMountRequest, unmountSealTimeout) if err != nil { slog.Error("Failed to unmount volume", "name", ebsRequest.Name, "instance", instance.ID, "err", err) @@ -253,13 +263,13 @@ func (a *volumeMounterAdapter) Unmount(instance *vm.VM) error { // MountOne sends ebs.mount for a single request and writes the resolved // NBDURI back into req.NBDURI. Used by hot-attach (Manager.AttachVolume). -func (a *volumeMounterAdapter) MountOne(req *types.EBSRequest) error { +func (a *volumeMounterAdapter) MountOne(ctx context.Context, accountID string, req *types.EBSRequest) error { payload, err := json.Marshal(req) if err != nil { return fmt.Errorf("marshal ebs.mount request: %w", err) } - reply, err := ebsRequestWithTrace(a.nc, a.topic("mount"), payload, 30*time.Second) + reply, err := ebsRequestWithTrace(ctx, a.nc, accountID, a.topic("mount"), payload, 30*time.Second) if err != nil { return fmt.Errorf("ebs.mount NATS request: %w", err) } @@ -282,8 +292,8 @@ func (a *volumeMounterAdapter) MountOne(req *types.EBSRequest) error { // UnmountOne sends ebs.unmount and returns any error. The handler seals the // volume's block map to predastore, so the caller decides whether a failure // blocks the volume's available transition. -func (a *volumeMounterAdapter) UnmountOne(req types.EBSRequest) error { - if err := a.unmountOne(req); err != nil { +func (a *volumeMounterAdapter) UnmountOne(ctx context.Context, accountID string, req types.EBSRequest) error { + if err := a.unmountOne(ctx, accountID, req); err != nil { slog.Error("UnmountOne failed", "volume", req.Name, "err", err) return err } @@ -292,12 +302,12 @@ func (a *volumeMounterAdapter) UnmountOne(req types.EBSRequest) error { } // unmountOne sends ebs.unmount and returns any error. -func (a *volumeMounterAdapter) unmountOne(req types.EBSRequest) error { +func (a *volumeMounterAdapter) unmountOne(ctx context.Context, accountID string, req types.EBSRequest) error { payload, err := json.Marshal(req) if err != nil { return fmt.Errorf("marshal unmount request: %w", err) } - msg, err := ebsRequestWithTrace(a.nc, a.topic("unmount"), payload, unmountSealTimeout) + msg, err := ebsRequestWithTrace(ctx, a.nc, accountID, a.topic("unmount"), payload, unmountSealTimeout) if err != nil { return fmt.Errorf("ebs.unmount NATS request: %w", err) } @@ -670,7 +680,11 @@ func (a *instanceCleanerAdapter) DeleteVolumes(instance *vm.VM) error { firstErr = cmp.Or(firstErr, err) continue } - deleteMsg, err := ebsRequestWithTrace(a.d.natsConn, "ebs.delete", ebsDeleteData, 30*time.Second) + // Teardown runs from the reaper and from terminate cleanup, neither + // of which has a caller to inherit a trace from. The account still + // comes off the instance, so the work stays attributable. + deleteMsg, err := ebsRequestWithTrace(context.Background(), a.d.natsConn, instance.AccountID, + "ebs.delete", ebsDeleteData, 30*time.Second) if err != nil { slog.Warn("Failed to send ebs.delete for internal volume", "name", ebsRequest.Name, "id", instance.ID, "err", err) diff --git a/spinifex/daemon/vm_adapters_invariants_test.go b/spinifex/daemon/vm_adapters_invariants_test.go index 4cf123818..9f3df46a4 100644 --- a/spinifex/daemon/vm_adapters_invariants_test.go +++ b/spinifex/daemon/vm_adapters_invariants_test.go @@ -46,7 +46,7 @@ func TestUnmountNeverReleasesBootVolume(t *testing.T) { }, }, } - require.NoError(t, adapter.Unmount(inst)) + require.NoError(t, adapter.Unmount(t.Context(), inst)) calls := volState.snapshot() assert.NotContains(t, calls, "vol-bios-root:available", diff --git a/spinifex/daemon/vm_adapters_test.go b/spinifex/daemon/vm_adapters_test.go index e3cdf2627..fa3042e3e 100644 --- a/spinifex/daemon/vm_adapters_test.go +++ b/spinifex/daemon/vm_adapters_test.go @@ -324,7 +324,7 @@ func TestVolumeMounterAdapter_MountOne(t *testing.T) { NBDURI: tt.initialURI, } - err := adapter.MountOne(req) + err := adapter.MountOne(t.Context(), "", req) if tt.wantErr { require.Error(t, err) if tt.wantErrSub != "" { @@ -593,7 +593,7 @@ func TestVolumeMounterAdapter_Mount_WrapsErrMountRetryable(t *testing.T) { instance := &vm.VM{ID: "i-mount-retryable"} instance.EBSRequests.Requests = []types.EBSRequest{{Name: "vol-retryable"}} - err = adapter.Mount(instance) + err = adapter.Mount(t.Context(), instance) require.Error(t, err) assert.ErrorIs(t, err, vm.ErrMountRetryable, "a Retryable mount response must wrap vm.ErrMountRetryable so relaunchAll can retry it") @@ -614,7 +614,7 @@ func TestVolumeMounterAdapter_Mount_PermanentErrorNotWrapped(t *testing.T) { instance := &vm.VM{ID: "i-mount-permanent"} instance.EBSRequests.Requests = []types.EBSRequest{{Name: "vol-permanent"}} - err = adapter.Mount(instance) + err = adapter.Mount(t.Context(), instance) require.Error(t, err) assert.NotErrorIs(t, err, vm.ErrMountRetryable, "a non-Retryable mount response must not be mistaken for a transient failure") @@ -629,7 +629,7 @@ func TestVolumeMounterAdapter_Mount_NoResponderIsRetryable(t *testing.T) { instance := &vm.VM{ID: "i-mount-noresponder"} instance.EBSRequests.Requests = []types.EBSRequest{{Name: "vol-noresponder"}} - err := adapter.Mount(instance) + err := adapter.Mount(t.Context(), instance) require.Error(t, err) assert.ErrorIs(t, err, vm.ErrMountRetryable, "a mount request with no responder must wrap vm.ErrMountRetryable so relaunchAll can retry it") diff --git a/spinifex/daemon/vm_adapters_trace_test.go b/spinifex/daemon/vm_adapters_trace_test.go index bbd1512eb..5f5353056 100644 --- a/spinifex/daemon/vm_adapters_trace_test.go +++ b/spinifex/daemon/vm_adapters_trace_test.go @@ -6,6 +6,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/mulgadc/spinifex/spinifex/vm" "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -56,7 +57,7 @@ func TestEbsRequestWithTrace_HeaderCarriesTraceparent(t *testing.T) { adapter := newVolumeMounterAdapter(daemon.natsConn, daemon.node, nil) req := &types.EBSRequest{Name: "vol-traced", DeviceName: "/dev/sdf"} - require.NoError(t, adapter.MountOne(req)) + require.NoError(t, adapter.MountOne(t.Context(), "", req)) hdr := <-headerCh assert.NotEmpty(t, hdr.Get("traceparent"), "outbound ebs.mount request must carry a traceparent header") @@ -87,7 +88,7 @@ func TestEbsRequestWithTrace_ProducerConsumerLinked(t *testing.T) { defer sub.Unsubscribe() adapter := newVolumeMounterAdapter(daemon.natsConn, daemon.node, nil) - require.NoError(t, adapter.UnmountOne(types.EBSRequest{Name: "vol-linked"})) + require.NoError(t, adapter.UnmountOne(t.Context(), "", types.EBSRequest{Name: "vol-linked"})) spans := sr.Ended() require.Len(t, spans, 2, "expected one producer span and one consumer span") @@ -112,3 +113,102 @@ func TestEbsRequestWithTrace_ProducerConsumerLinked(t *testing.T) { assert.Equal(t, producer.SpanContext().SpanID(), consumer.Parent().SpanID(), "consumer span's parent must be the producer span") } + +// mountOnceOn answers a single ebs.mount on subject and hands back the headers +// the adapter sent, so a test can assert on the wire rather than the span only. +func mountOnceOn(t *testing.T, nc *nats.Conn, subject string) <-chan nats.Header { + t.Helper() + headerCh := make(chan nats.Header, 1) + sub, err := nc.Subscribe(subject, func(msg *nats.Msg) { + headerCh <- msg.Header + data, marshalErr := json.Marshal(types.EBSMountResponse{URI: "nbd://vol"}) + require.NoError(t, marshalErr) + require.NoError(t, msg.Respond(data)) + }) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + return headerCh +} + +// The ebs hop used to root its own trace from context.Background(), so block +// storage work appeared as an orphan trace no request could be followed into. +// Given a caller's context it must extend that trace instead. +func TestEbsRequestWithTrace_JoinsTheCallersTrace(t *testing.T) { + sr := withRecordedSpans(t) + daemon := createTestDaemon(t, sharedNATSURL) + mountOnceOn(t, daemon.natsConn, "ebs.node-1.mount") + + adapter := newVolumeMounterAdapter(daemon.natsConn, daemon.node, nil) + ctx, caller := otel.Tracer("test").Start(t.Context(), "caller") + require.NoError(t, adapter.MountOne(ctx, "000000000042", &types.EBSRequest{Name: "vol-ctx"})) + caller.End() + + var producer sdktrace.ReadOnlySpan + for _, s := range sr.Ended() { + if s.SpanKind() == trace.SpanKindClient { + producer = s + } + } + require.NotNil(t, producer, "producer span must be recorded") + + assert.Equal(t, caller.SpanContext().TraceID(), producer.SpanContext().TraceID(), + "the ebs span must join the caller's trace rather than root one of its own") + assert.Equal(t, caller.SpanContext().SpanID(), producer.Parent().SpanID(), + "the ebs span's parent must be the caller") +} + +// One cluster serves many accounts and a volume belongs to exactly one of +// them. The account rides both the span and the header, because viperblockd +// reads the header to attribute its own consumer span. +func TestEbsRequestWithTrace_CarriesTheAccount(t *testing.T) { + sr := withRecordedSpans(t) + daemon := createTestDaemon(t, sharedNATSURL) + headerCh := mountOnceOn(t, daemon.natsConn, "ebs.node-1.mount") + + adapter := newVolumeMounterAdapter(daemon.natsConn, daemon.node, nil) + instance := &vm.VM{ID: "i-acct", AccountID: "000000000042"} + instance.EBSRequests.Requests = []types.EBSRequest{{Name: "vol-acct"}} + require.NoError(t, adapter.Mount(t.Context(), instance)) + + assert.Equal(t, "000000000042", (<-headerCh).Get(utils.AccountIDHeader), + "viperblockd reads the account off the header, so it must be set") + + var producer sdktrace.ReadOnlySpan + for _, s := range sr.Ended() { + if s.SpanKind() == trace.SpanKindClient { + producer = s + } + } + require.NotNil(t, producer) + assert.Equal(t, "000000000042", spanAccountOf(producer), + "the ebs span must name the account it acted for") +} + +// Recovery and teardown act for no caller. Crediting that work to an account +// would be worse than leaving it unattributed. +func TestEbsRequestWithTrace_OmitsAnAbsentAccount(t *testing.T) { + sr := withRecordedSpans(t) + daemon := createTestDaemon(t, sharedNATSURL) + headerCh := mountOnceOn(t, daemon.natsConn, "ebs.node-1.mount") + + adapter := newVolumeMounterAdapter(daemon.natsConn, daemon.node, nil) + require.NoError(t, adapter.MountOne(t.Context(), "", &types.EBSRequest{Name: "vol-none"})) + + assert.Empty(t, (<-headerCh).Get(utils.AccountIDHeader), + "an unattributed request must not carry a blank account header") + for _, s := range sr.Ended() { + if s.SpanKind() == trace.SpanKindClient { + assert.Empty(t, spanAccountOf(s), "an unattributed span must carry no account") + } + } +} + +// spanAccountOf returns the account attribute of span, or "". +func spanAccountOf(span sdktrace.ReadOnlySpan) string { + for _, attr := range span.Attributes() { + if string(attr.Key) == utils.AttrAccountID { + return attr.Value.AsString() + } + } + return "" +} diff --git a/spinifex/handlers/ec2/instance/service_impl_test.go b/spinifex/handlers/ec2/instance/service_impl_test.go index cf8aa2f88..a2f2ffa4f 100644 --- a/spinifex/handlers/ec2/instance/service_impl_test.go +++ b/spinifex/handlers/ec2/instance/service_impl_test.go @@ -2319,10 +2319,10 @@ func TestStartStoppedInstance_ClaimConflict(t *testing.T) { // fails fast and deterministically without touching a real qemu process. type raceVolumeMounter struct{} -func (raceVolumeMounter) Mount(*vm.VM) error { return nil } -func (raceVolumeMounter) Unmount(*vm.VM) error { return nil } -func (raceVolumeMounter) MountOne(*spxtypes.EBSRequest) error { return nil } -func (raceVolumeMounter) UnmountOne(spxtypes.EBSRequest) error { return nil } +func (raceVolumeMounter) Mount(context.Context, *vm.VM) error { return nil } +func (raceVolumeMounter) Unmount(context.Context, *vm.VM) error { return nil } +func (raceVolumeMounter) MountOne(context.Context, string, *spxtypes.EBSRequest) error { return nil } +func (raceVolumeMounter) UnmountOne(context.Context, string, spxtypes.EBSRequest) error { return nil } // TestStartStoppedInstance_ConcurrentClaimRace is the regression test for // the double-start bug this claim closes: two nodes (or a forwarded call diff --git a/spinifex/vm/crash_recovery.go b/spinifex/vm/crash_recovery.go index 7866c23ec..1ff07f09a 100644 --- a/spinifex/vm/crash_recovery.go +++ b/spinifex/vm/crash_recovery.go @@ -133,7 +133,7 @@ func (m *Manager) HandleCrash(instance *VM, waitErr error) { removeTelemetryArtifacts(instance) if m.deps.VolumeMounter != nil { - if err := m.deps.VolumeMounter.Unmount(instance); err != nil { + if err := m.deps.VolumeMounter.Unmount(context.Background(), instance); err != nil { slog.Error("Volume unmount failed during crash handling", "instance", instance.ID, "err", err) } diff --git a/spinifex/vm/lifecycle.go b/spinifex/vm/lifecycle.go index 34669ad5f..c7e29d3bb 100644 --- a/spinifex/vm/lifecycle.go +++ b/spinifex/vm/lifecycle.go @@ -152,7 +152,7 @@ func (m *Manager) launch(ctx context.Context, instance *VM) (err error) { } _, mountSpan := otel.Tracer(vmTracerName).Start(ctx, "vm.launch.mount_volumes") - mountErr := m.deps.VolumeMounter.Mount(instance) + mountErr := m.deps.VolumeMounter.Mount(ctx, instance) endSpanWithError(mountSpan, mountErr) if mountErr != nil { slog.ErrorContext(ctx, "Failed to mount volumes", "err", mountErr) @@ -169,7 +169,7 @@ func (m *Manager) launch(ctx context.Context, instance *VM) (err error) { // fails, unmounting seals the untouched backend and keeps the launch from // exposing a writer that snapshots would classify as available. if stateErr := m.markAttachedVolumesInUse(instance); stateErr != nil { - if unmountErr := m.deps.VolumeMounter.Unmount(instance); unmountErr != nil { + if unmountErr := m.deps.VolumeMounter.Unmount(ctx, instance); unmountErr != nil { return errors.Join( fmt.Errorf("persist attached volume state: %w", stateErr), fmt.Errorf("rollback mounted volumes: %w", unmountErr), diff --git a/spinifex/vm/manager_deps_test.go b/spinifex/vm/manager_deps_test.go index 98d90e70d..8c466d9f7 100644 --- a/spinifex/vm/manager_deps_test.go +++ b/spinifex/vm/manager_deps_test.go @@ -1,6 +1,7 @@ package vm import ( + "context" "errors" "maps" "sync" @@ -153,7 +154,7 @@ type fakeVolumeMounter struct { onMount func(*VM) } -func (f *fakeVolumeMounter) Mount(v *VM) error { +func (f *fakeVolumeMounter) Mount(_ context.Context, v *VM) error { f.mu.Lock() f.mounted = append(f.mounted, v.ID) err := f.mountErr @@ -165,14 +166,14 @@ func (f *fakeVolumeMounter) Mount(v *VM) error { return err } -func (f *fakeVolumeMounter) Unmount(v *VM) error { +func (f *fakeVolumeMounter) Unmount(_ context.Context, v *VM) error { f.mu.Lock() f.unmounted = append(f.unmounted, v.ID) f.mu.Unlock() return nil } -func (f *fakeVolumeMounter) MountOne(req *types.EBSRequest) error { +func (f *fakeVolumeMounter) MountOne(_ context.Context, _ string, req *types.EBSRequest) error { f.mu.Lock() f.mountedOne = append(f.mountedOne, req.Name) mountOneErr := f.mountOneErr @@ -187,7 +188,7 @@ func (f *fakeVolumeMounter) MountOne(req *types.EBSRequest) error { return nil } -func (f *fakeVolumeMounter) UnmountOne(req types.EBSRequest) error { +func (f *fakeVolumeMounter) UnmountOne(_ context.Context, _ string, req types.EBSRequest) error { f.mu.Lock() f.unmountedOne = append(f.unmountedOne, req.Name) err := f.unmountOneErr diff --git a/spinifex/vm/restore_test.go b/spinifex/vm/restore_test.go index 26c604e70..ec8327511 100644 --- a/spinifex/vm/restore_test.go +++ b/spinifex/vm/restore_test.go @@ -556,7 +556,7 @@ type recoveryMounter struct { behavior map[string]func(*VM) error } -func (f *recoveryMounter) Mount(v *VM) error { +func (f *recoveryMounter) Mount(_ context.Context, v *VM) error { f.mu.Lock() f.mounted = append(f.mounted, v.ID) fn := f.behavior[v.ID] @@ -567,9 +567,9 @@ func (f *recoveryMounter) Mount(v *VM) error { return nil } -func (f *recoveryMounter) Unmount(*VM) error { return nil } -func (f *recoveryMounter) MountOne(*types.EBSRequest) error { return nil } -func (f *recoveryMounter) UnmountOne(types.EBSRequest) error { return nil } +func (f *recoveryMounter) Unmount(context.Context, *VM) error { return nil } +func (f *recoveryMounter) MountOne(context.Context, string, *types.EBSRequest) error { return nil } +func (f *recoveryMounter) UnmountOne(context.Context, string, types.EBSRequest) error { return nil } var _ VolumeMounter = (*recoveryMounter)(nil) diff --git a/spinifex/vm/shutdown.go b/spinifex/vm/shutdown.go index 21123aa7e..ccc4d4700 100644 --- a/spinifex/vm/shutdown.go +++ b/spinifex/vm/shutdown.go @@ -484,7 +484,7 @@ func (m *Manager) shutdownAndUnmount(instance *VM) { } if m.deps.VolumeMounter != nil { - if err := m.deps.VolumeMounter.Unmount(instance); err != nil { + if err := m.deps.VolumeMounter.Unmount(context.Background(), instance); err != nil { slog.Error("Volume unmount failed", "id", instance.ID, "err", err) } } diff --git a/spinifex/vm/volume_mounter.go b/spinifex/vm/volume_mounter.go index e631d56ef..e020bcc86 100644 --- a/spinifex/vm/volume_mounter.go +++ b/spinifex/vm/volume_mounter.go @@ -1,23 +1,33 @@ package vm -import "github.com/mulgadc/spinifex/spinifex/types" +import ( + "context" + + "github.com/mulgadc/spinifex/spinifex/types" +) // VolumeMounter mounts and unmounts the EBS volumes attached to a VM. The // real implementation routes ebs.mount / ebs.unmount NATS requests; the // abstraction keeps NATS out of the manager. +// +// Every method takes the caller's context so the ebs.* span joins the trace +// that asked for the work rather than rooting one of its own. Shutdown and +// crash recovery have no caller and pass a background context, which is an +// honest answer rather than a missing one. type VolumeMounter interface { // Mount mounts every attached volume in v.EBSRequests.Requests, recording // the resolved NBDURI back onto each request entry. - Mount(v *VM) error + Mount(ctx context.Context, v *VM) error // Unmount sends ebs.unmount for each attached volume. Errors are logged // per volume and aggregated; partial failure is tolerated. - Unmount(v *VM) error + Unmount(ctx context.Context, v *VM) error // MountOne sends ebs.mount for a single request and writes the resolved - // NBDURI back into req.NBDURI on success. Used by hot-attach. - MountOne(req *types.EBSRequest) error + // NBDURI back into req.NBDURI on success. Used by hot-attach. accountID + // names the owner, which a lone request carries no instance to supply. + MountOne(ctx context.Context, accountID string, req *types.EBSRequest) error // UnmountOne sends ebs.unmount for a single request and returns any error. // ebs.unmount drives the synchronous block-map seal to predastore, so hot // detach gates the volume's available transition on the returned error. - UnmountOne(req types.EBSRequest) error + UnmountOne(ctx context.Context, accountID string, req types.EBSRequest) error } diff --git a/spinifex/vm/volumes.go b/spinifex/vm/volumes.go index 2c36fcb0a..b8a6c0816 100644 --- a/spinifex/vm/volumes.go +++ b/spinifex/vm/volumes.go @@ -33,11 +33,11 @@ const detachAggregateTimeout = 3 * time.Minute // rollbackUnmount unmounts a volume while unwinding a failed AttachVolume. // Callers keep the volume non-available when the seal fails so a later attach // cannot discard local state that may still be owned by the backend. -func (m *Manager) rollbackUnmount(req types.EBSRequest) error { +func (m *Manager) rollbackUnmount(ctx context.Context, accountID string, req types.EBSRequest) error { if m.deps.VolumeMounter == nil { return nil } - if err := m.deps.VolumeMounter.UnmountOne(req); err != nil { + if err := m.deps.VolumeMounter.UnmountOne(ctx, accountID, req); err != nil { slog.Warn("AttachVolume: rollback unmount failed", "volume", req.Name, "err", err) return err } @@ -70,7 +70,7 @@ func (m *Manager) rollbackHotAttach(ctx context.Context, instance *VM, req types } m.delIothreadBestEffort(ctx, instance, iothreadID, req.Name) - return m.rollbackUnmount(req) == nil + return m.rollbackUnmount(ctx, instance.AccountID, req) == nil } // AttachVolume hot-plugs a volume via the pipeline mount → blockdev-add → @@ -110,12 +110,12 @@ func (m *Manager) AttachVolume(ctx context.Context, id, volumeID, device string) if m.deps.VolumeMounter == nil { return "", fmt.Errorf("VolumeMounter not wired") } - if err := m.deps.VolumeMounter.MountOne(&ebsRequest); err != nil { + if err := m.deps.VolumeMounter.MountOne(ctx, instance.AccountID, &ebsRequest); err != nil { slog.ErrorContext(ctx, "AttachVolume: ebs.mount failed", "volumeId", volumeID, "err", err) // Empty-URI response leaves backend NBD state ambiguous; unmount // defensively to avoid orphaning a half-started mount. if errors.Is(err, ErrMountAmbiguous) { - _ = m.rollbackUnmount(ebsRequest) + _ = m.rollbackUnmount(ctx, instance.AccountID, ebsRequest) } return "", fmt.Errorf("mount volume %s: %w", volumeID, err) } @@ -123,7 +123,7 @@ func (m *Manager) AttachVolume(ctx context.Context, id, volumeID, device string) serverType, socketPath, nbdHost, nbdPort, err := utils.ParseNBDURI(ebsRequest.NBDURI) if err != nil { slog.ErrorContext(ctx, "AttachVolume: failed to parse NBDURI", "uri", ebsRequest.NBDURI, "err", err) - _ = m.rollbackUnmount(ebsRequest) + _ = m.rollbackUnmount(ctx, instance.AccountID, ebsRequest) return "", fmt.Errorf("parse NBDURI: %w", err) } serverArg := NBDServerOpts{Type: serverType, Path: socketPath, Host: nbdHost, Port: nbdPort}.QMPArg() @@ -146,7 +146,7 @@ func (m *Manager) AttachVolume(ctx context.Context, id, volumeID, device string) instance.EBSRequests.Mu.Unlock() if hotplugPort == 0 { slog.ErrorContext(ctx, "AttachVolume: EBS hot-plug port pool exhausted", "volumeId", volumeID) - _ = m.rollbackUnmount(ebsRequest) + _ = m.rollbackUnmount(ctx, instance.AccountID, ebsRequest) return "", ErrAttachmentLimitExceeded } ebsRequest.HotplugPort = hotplugPort @@ -159,7 +159,7 @@ func (m *Manager) AttachVolume(ctx context.Context, id, volumeID, device string) }, }, instance.ID); err != nil { slog.ErrorContext(ctx, "AttachVolume: QMP object-add iothread failed", "volumeId", volumeID, "err", err) - _ = m.rollbackUnmount(ebsRequest) + _ = m.rollbackUnmount(ctx, instance.AccountID, ebsRequest) return "", fmt.Errorf("QMP object-add iothread: %w", err) } @@ -175,7 +175,7 @@ func (m *Manager) AttachVolume(ctx context.Context, id, volumeID, device string) }, instance.ID); err != nil { slog.ErrorContext(ctx, "AttachVolume: QMP blockdev-add failed", "volumeId", volumeID, "err", err) m.delIothreadBestEffort(ctx, instance, iothreadID, volumeID) - _ = m.rollbackUnmount(ebsRequest) + _ = m.rollbackUnmount(ctx, instance.AccountID, ebsRequest) return "", fmt.Errorf("QMP blockdev-add: %w", err) } @@ -409,7 +409,7 @@ func (m *Manager) DetachVolume(ctx context.Context, id, volumeID, device string, // and return the error; an AWS-CLI retry re-drives the seal and same-node // reattach still works meanwhile. if m.deps.VolumeMounter != nil { - if err := m.deps.VolumeMounter.UnmountOne(ebsReq); err != nil { + if err := m.deps.VolumeMounter.UnmountOne(ctx, instance.AccountID, ebsReq); err != nil { slog.ErrorContext(ctx, "DetachVolume: ebs.unmount seal failed, leaving volume attached", "volumeId", volumeID, "err", err) return "", fmt.Errorf("ebs.unmount seal: %w", err)