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)