From 2ca655cd394dd550e44c964879c5a4f73b8fb236 Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Thu, 25 Jun 2026 21:24:32 +1000 Subject: [PATCH] fix: EBS hot-plug PCIe port allocator (every attach past the first collided) (#421) * fix: serialize EBS hot-plug per instance to prevent PCIe port race Concurrent AttachVolume calls to one instance both allocated the lowest free hot-plug PCIe root port via nextHotplugEBSPort (which reads live QEMU state), then issued QMP device_add on the same hotplug-ebs bus. The second device_add failed with 'PCI: slot 0 function 0 already occupied by virtio-blk-pci', surfacing as nats:timeout -> ServerInternal and a CSI retry-storm ('retry quota exceeded'); volumes never attached. Add a per-instance attachMu and hold it across the whole AttachVolume and DetachVolume critical sections, mirroring the ENI hot-plug serialization. The mutex is non-persisted and shared via the Manager's stable *VM. * fix: allocate EBS hot-plug PCIe port from in-memory accounting AttachVolume derived the free PCIe hot-plug root port from a live query-block scan (nextHotplugEBSPort), extracting the bus with a hotplug-ebs(\d+) regex on each device's QDev. Real QEMU reports a hot-plugged blockdev by id (/machine/peripheral/vdisk-vol-xxx/ virtio-backend), with no hotplug-ebs component, so the regex never matched, the used set stayed empty, and the allocator always returned port 1. Only the first EBS volume per instance attached; every later attach collided on "PCI: slot 0 function 0 already occupied by virtio-blk-pci", rolling back the blockdev, killing NBDKit, and surfacing as nats: timeout / ServerInternal to the CSI driver, which then exhausted its retry quota. Allocate the port from authoritative in-memory accounting instead: record the chosen port as EBSRequest.HotplugPort (persisted, so it survives a daemon restart) and pick the lowest unclaimed port with freeHotplugEBSPort over the instance's recorded requests. Hold attachMu across allocate->device_add so concurrent CSI attaches cannot pick the same port. DetachVolume already removes the request, freeing the port. Drops the query-block round-trip from the attach path. --- spinifex/types/ebs.go | 4 ++++ spinifex/vm/device_map.go | 37 ++++++++++++++++------------------ spinifex/vm/device_map_test.go | 37 ++++++++++++++++++++++++++++++++++ spinifex/vm/vm.go | 7 +++++++ spinifex/vm/volumes.go | 32 ++++++++++++++++++++--------- spinifex/vm/volumes_test.go | 8 ++++---- 6 files changed, 91 insertions(+), 34 deletions(-) diff --git a/spinifex/types/ebs.go b/spinifex/types/ebs.go index dbe4587c0..fac045bf1 100644 --- a/spinifex/types/ebs.go +++ b/spinifex/types/ebs.go @@ -18,6 +18,10 @@ type EBSRequest struct { DeleteOnTermination bool `json:"DeleteOnTermination"` NBDURI string `json:"NBDURI"` // NBD URI - socket path (nbd:unix:/path.sock) or TCP (nbd://host:port) DeviceName string `json:"DeviceName"` // AWS API device name (e.g. /dev/sdf) for hot-plugged volumes + // HotplugPort is the PCIe hot-plug root port (hotplug-ebs{N}) a hot-plugged + // volume occupies. 0 for boot/non-hot-plugged volumes. Persisted so port + // accounting survives a daemon restart. + HotplugPort int `json:"HotplugPort,omitempty"` } // NBDTransport defines the transport type for NBD connections diff --git a/spinifex/vm/device_map.go b/spinifex/vm/device_map.go index b659878a5..d6d07877c 100644 --- a/spinifex/vm/device_map.go +++ b/spinifex/vm/device_map.go @@ -11,6 +11,7 @@ import ( "time" "github.com/mulgadc/spinifex/spinifex/qmp" + "github.com/mulgadc/spinifex/spinifex/types" ) // pciAddrRegexp extracts the device index from a boot-time QDev path. @@ -260,30 +261,26 @@ func nextAvailableDevice(instance *VM) string { return "" } -// nextHotplugEBSPort picks the lowest free EBS hot-plug PCIe port -// (1..EBSHotPlugSlotCount) from live QEMU state. The port is decoupled from the -// AWS device name: callers (e.g. the EBS CSI driver) supply arbitrary names like -// /dev/xvdaa that do not map onto the /dev/sd[f-p] slot range, so the physical -// port is allocated independently. Returns 0 when the pool is exhausted. -func nextHotplugEBSPort(qmpClient *qmp.QMPClient, instanceID string) (int, error) { - resp, err := sendQMPCommand(qmpClient, qmp.QMPCommand{Execute: "query-block"}, instanceID) - if err != nil { - return 0, fmt.Errorf("query-block failed: %w", err) - } - var devices []qmp.BlockDevice - if err := json.Unmarshal(resp.Return, &devices); err != nil { - return 0, fmt.Errorf("failed to parse query-block response: %w", err) - } - used := make(map[int]bool, len(devices)) - for _, dev := range devices { - if port := extractHotplugPort(dev.QDev); port > 0 { - used[port] = true +// freeHotplugEBSPort returns the lowest EBS hot-plug PCIe port +// (1..EBSHotPlugSlotCount) not already claimed by an attached volume, or 0 when +// the pool is exhausted. The port is decoupled from the AWS device name: +// callers (e.g. the EBS CSI driver) supply arbitrary names like /dev/xvdaa that +// do not map onto the /dev/sd[f-p] range, so the physical port is allocated +// independently. Accounting is in-memory and authoritative: QEMU reports block +// devices by id, not by bus, so a live query-block scan cannot tell which port +// is occupied. Callers must hold the EBSRequests lock; serialize the matching +// device_add under attachMu so allocation and use are atomic. +func freeHotplugEBSPort(reqs []types.EBSRequest) int { + used := make(map[int]bool, len(reqs)) + for _, r := range reqs { + if r.HotplugPort > 0 { + used[r.HotplugPort] = true } } for p := 1; p <= EBSHotPlugSlotCount; p++ { if !used[p] { - return p, nil + return p } } - return 0, nil + return 0 } diff --git a/spinifex/vm/device_map_test.go b/spinifex/vm/device_map_test.go index 54d366c5d..f206d5629 100644 --- a/spinifex/vm/device_map_test.go +++ b/spinifex/vm/device_map_test.go @@ -6,6 +6,7 @@ import ( "time" "github.com/mulgadc/spinifex/spinifex/qmp" + "github.com/mulgadc/spinifex/spinifex/types" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -326,3 +327,39 @@ func TestExtractHotplugPort(t *testing.T) { }) } } + +func TestFreeHotplugEBSPort(t *testing.T) { + mk := func(ports ...int) []types.EBSRequest { + reqs := make([]types.EBSRequest, 0, len(ports)) + for _, p := range ports { + reqs = append(reqs, types.EBSRequest{HotplugPort: p}) + } + return reqs + } + + tests := []struct { + name string + reqs []types.EBSRequest + want int + }{ + {name: "empty pool returns 1", reqs: nil, want: 1}, + {name: "first port taken returns 2", reqs: mk(1), want: 2}, + {name: "lowest gap is reused", reqs: mk(1, 3), want: 2}, + {name: "boot/zero ports ignored", reqs: mk(0, 0), want: 1}, + {name: "contiguous run returns next", reqs: mk(1, 2, 3), want: 4}, + {name: "out-of-order still finds gap", reqs: mk(3, 1), want: 2}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, freeHotplugEBSPort(tt.reqs)) + }) + } + + t.Run("exhausted pool returns 0", func(t *testing.T) { + full := make([]types.EBSRequest, 0, EBSHotPlugSlotCount) + for p := 1; p <= EBSHotPlugSlotCount; p++ { + full = append(full, types.EBSRequest{HotplugPort: p}) + } + assert.Equal(t, 0, freeHotplugEBSPort(full)) + }) +} diff --git a/spinifex/vm/vm.go b/spinifex/vm/vm.go index beef35ef3..b4ecd7aed 100644 --- a/spinifex/vm/vm.go +++ b/spinifex/vm/vm.go @@ -46,6 +46,13 @@ type VM struct { QMPClient *qmp.QMPClient `json:"-"` + // attachMu serializes EBS volume hot-plug (attach/detach) for this instance + // so PCIe hot-plug port allocation and the matching device_add are atomic. + // Two concurrent attaches would otherwise pick the same port and the second + // device_add fails with "slot 0 ... already occupied". Non-persisted; the + // Manager hands out a stable *VM, so the lock is shared across calls. + attachMu sync.Mutex `json:"-"` + // User attributes (user initiated stop/delete) Attributes types.EC2CommandAttributes `json:"attributes"` diff --git a/spinifex/vm/volumes.go b/spinifex/vm/volumes.go index 0fed4e1bd..458d857cb 100644 --- a/spinifex/vm/volumes.go +++ b/spinifex/vm/volumes.go @@ -65,6 +65,13 @@ func (m *Manager) AttachVolume(id, volumeID, device string) (AttachVolumeResult, ErrInvalidTransition, id, status) } + // Serialize hot-plug per instance so the port allocation below and the + // matching device_add are atomic; concurrent attaches would otherwise pick + // the same PCIe root port and the second device_add fails ("slot 0 already + // occupied"). Detach takes the same lock. + instance.attachMu.Lock() + defer instance.attachMu.Unlock() + if device == "" { m.UpdateState(id, func(v *VM) { device = nextAvailableDevice(v) }) if device == "" { @@ -108,21 +115,22 @@ func (m *Manager) AttachVolume(id, volumeID, device string) (AttachVolumeResult, deviceID := fmt.Sprintf("vdisk-%s", volumeID) iothreadID := fmt.Sprintf("ioth-%s", volumeID) - // Allocate the PCIe hot-plug port from live QEMU state before any QMP - // mutation, so an exhausted pool rolls back only the mount. virtio-blk-pci - // cannot land on pcie.0 (no hot-plug), so a free hotplug-ebs root port is - // required regardless of the AWS device name. - hotplugPort, err := nextHotplugEBSPort(instance.QMPClient, instance.ID) - if err != nil { - slog.Error("AttachVolume: failed to query hot-plug ports", "volumeId", volumeID, "err", err) - m.rollbackUnmount(ebsRequest) - return AttachVolumeResult{}, fmt.Errorf("query hot-plug ports: %w", err) - } + // Allocate the PCIe hot-plug port from in-memory accounting. QEMU reports + // block devices by id (/machine/peripheral//virtio-backend), not by bus, + // so a live query-block scan cannot tell which hotplug-ebs port is occupied + // and always returned the first one — every attach past the first collided + // on slot 0. The recorded HotplugPort per attached volume is authoritative; + // attachMu (held above) serializes allocation. virtio-blk-pci cannot land on + // pcie.0 (no hot-plug), so a free hotplug-ebs root port is always required. + instance.EBSRequests.Mu.Lock() + hotplugPort := freeHotplugEBSPort(instance.EBSRequests.Requests) + instance.EBSRequests.Mu.Unlock() if hotplugPort == 0 { slog.Error("AttachVolume: EBS hot-plug port pool exhausted", "volumeId", volumeID) m.rollbackUnmount(ebsRequest) return AttachVolumeResult{}, ErrAttachmentLimitExceeded } + ebsRequest.HotplugPort = hotplugPort if _, err := sendQMPCommand(instance.QMPClient, qmp.QMPCommand{ Execute: "object-add", @@ -282,6 +290,10 @@ func (m *Manager) DetachVolume(id, volumeID, device string, force bool) (string, ErrInvalidTransition, id, status) } + // Serialize against concurrent attach/detach on this instance's PCIe ports. + instance.attachMu.Lock() + defer instance.attachMu.Unlock() + instance.EBSRequests.Mu.Lock() var ebsReq types.EBSRequest found := false diff --git a/spinifex/vm/volumes_test.go b/spinifex/vm/volumes_test.go index 94ff3242f..114f8d042 100644 --- a/spinifex/vm/volumes_test.go +++ b/spinifex/vm/volumes_test.go @@ -616,7 +616,7 @@ func TestAttachVolume_ObjectAddFailure_TriggersUnmountOne(t *testing.T) { _, err := m.AttachVolume("i-1", "vol-1", "/dev/sdf") require.Error(t, err) assert.Contains(t, err.Error(), "object-add") - assert.Equal(t, []string{"query-block", "object-add"}, recorder.executes(), + assert.Equal(t, []string{"object-add"}, recorder.executes(), "object-add failure must short-circuit before blockdev-add") assert.Equal(t, []string{"vol-1"}, mounter.unmountedOne) } @@ -644,7 +644,7 @@ func TestAttachVolume_BlockdevAddFailure_TriggersUnmountOne(t *testing.T) { _, err := m.AttachVolume("i-1", "vol-1", "/dev/sdf") require.Error(t, err) assert.Contains(t, err.Error(), "blockdev-add") - assert.Equal(t, []string{"query-block", "object-add", "blockdev-add", "object-del"}, recorder.executes()) + assert.Equal(t, []string{"object-add", "blockdev-add", "object-del"}, recorder.executes()) assert.Equal(t, []string{"vol-1"}, mounter.unmountedOne) } @@ -671,7 +671,7 @@ func TestAttachVolume_DeviceAddFailure_BlockdevDelOK_Unmounts(t *testing.T) { _, err := m.AttachVolume("i-1", "vol-1", "/dev/sdf") require.Error(t, err) assert.Contains(t, err.Error(), "device_add") - assert.Equal(t, []string{"query-block", "object-add", "blockdev-add", "device_add", "blockdev-del", "object-del"}, recorder.executes()) + assert.Equal(t, []string{"object-add", "blockdev-add", "device_add", "blockdev-del", "object-del"}, recorder.executes()) assert.Equal(t, []string{"vol-1"}, mounter.unmountedOne, "successful blockdev-del rollback must be followed by UnmountOne") } @@ -704,7 +704,7 @@ func TestAttachVolume_DeviceAddFailure_BlockdevDelFails_SkipsUnmountOne(t *testi _, err := m.AttachVolume("i-1", "vol-1", "/dev/sdf") require.Error(t, err) assert.Contains(t, err.Error(), "device_add") - assert.Equal(t, []string{"query-block", "object-add", "blockdev-add", "device_add", "blockdev-del"}, recorder.executes()) + assert.Equal(t, []string{"object-add", "blockdev-add", "device_add", "blockdev-del"}, recorder.executes()) assert.Empty(t, mounter.unmountedOne, "failed blockdev-del must skip UnmountOne — unmounting under a live block node crashes QEMU") }