From fa09228a36fa3dc82a95cfa9cd978bb32d4fde98 Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Wed, 27 May 2026 22:54:46 +1000 Subject: [PATCH] =?UTF-8?q?feat:=20ENI=20hot-plug=20Sprint=203b=20?= =?UTF-8?q?=E2=80=94=20live=20QMP=20pipeline=20+=20AttachENI=20wiring=20(s?= =?UTF-8?q?tage=202)=20(#258)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat: ENI hot-plug Sprint 3a scaffolding Lays the Sprint 3a Tier 0 groundwork from docs/development/feature/enihotplug-v1.md so the live QMP pipeline in 3b can plug in without further plumbing changes. - types.ENIRequests: per-VM PCIe slot allocator state (free-list + eniID→slot reverse map), mirrors EBSRequests pattern. - instancetypes.MaxENIsForType / HotPlugENISlotsForType: instance-type → ENI cap lookup with explicit-type-then-family-then-default precedence. - vm.DeviceController interface + QMPDeviceController (real) + StubDeviceController (in-memory): unit-testable QMP device_add / device_del / netdev_add / netdev_del surface. QueryPCI body lands in Sprint 3b alongside the live response parser. - buildBaseVMConfig now accepts the instance type and emits two distinct PCIe root-port pools at boot: hotplug-ebs{1..11} (fixed, /dev/sd[f-p]) and hotplug-eni{1..N} (per-type, N = MaxENIs - 1). - volumes.go bus id renamed hotplug%d → hotplug-ebs%d; query-block QDev regex narrowed to hotplug-ebs(\d+) so future virtio-net devices on hotplug-eni ports cannot collide with the EBS ordering. - restore + ResetNodeLocalState reset ENIRequests.Mu alongside the EBS mutex so restored VMs start with a fresh allocator. Closes mulga-cs-enihotplug-3a. * feat: ENI hot-plug Sprint 3b — live QMP pipeline + AttachENI wiring Lands the Sprint 3b Tier 1 deliverables from docs/development/feature/enihotplug-v1.md so AttachNetworkInterface / DetachNetworkInterface AWS APIs drive the host-side QMP pipeline against a running VM. - vm/qmp_device.go: real QueryPCI parser. Walks the per-bus query-pci response and recurses into pci_bridge.devices so virtio-net devices attached under hotplug-eni{N} root ports surface in the flat result. - vm/eni_hotplug.go (new): HotPlugENI / HotUnplugENI on Manager. 4-step pipeline (tap → ovs-add-port → netdev_add → device_add + query-pci poll) with full rollback on step-N failure. Tap + OVS calls are stubbed in Sprint 3b; Sprint 3c drops in real NetworkPlumber / ovs-vsctl invocations against the same call-site interface. - vm/eni_hotplug_test.go (new): unit tests with StubDeviceController — success path, rollback on each step, AttachmentLimitExceeded, idempotent re-attach, non-running rejection, force-detach semantics, concurrent serialization, query-pci timeout. - types/ec2.go: EC2InstanceCommand gains AttachENI / DetachENI attrs + AttachENIData / DetachENIData payloads, mirroring the AttachVolume command surface. - handlers/ec2/vpc/eni.go: ENIRecord gains AttachmentStatus, HotPlugSlot, LastAttachError, DetachInFlight, DetachForce fields per plan §State machine. New helpers: GetENIRecord, UpdateENI (CAS patcher), FindENIByAttachment (used by DetachNetworkInterface lookup by attachmentId). - gateway/ec2/vpc/{Attach,Detach}NetworkInterface.go (new): AWS SDK request decoders that route via the per-instance ec2.cmd.{id} channel; DetachNetworkInterface resolves attachmentId → instanceId via DescribeNetworkInterfaces first. - gateway/ec2.go: registers the two new EC2 actions in the dispatcher. - daemon/daemon_handlers.go: handleEC2Events dispatches the two new attribute flags to per-instance handlers. - daemon/daemon_handlers_eni.go (new): composes vpcService.AttachENI (KV) with vmMgr.HotPlugENI (QMP) and publishes vpc.eni-hotplug.{attached,detached}.{instanceID} on success. KV state machine: available → attaching → attached → detaching → available. On QMP failure: KV rolls back to available with LastAttachError persisted for DescribeNetworkInterfaces visibility. - vm/errors.go: ErrENINotAttached, ErrENIPipelineTimeout, ErrQMPUnavailable. The Sprint 3c bead lands the real tap + ovs-vsctl wiring on the same pipeline; 3d adds the reconciler that converges the attaching / detaching KV states left behind by a daemon crash mid-pipeline. Closes mulga-cs-enihotplug-3b. * Update ec2.go * Update ec2.go --- spinifex/daemon/daemon_handlers.go | 4 + spinifex/daemon/daemon_handlers_eni.go | 219 +++++++++++++ spinifex/gateway/ec2.go | 6 + .../gateway/ec2/vpc/AttachNetworkInterface.go | 93 ++++++ .../gateway/ec2/vpc/DetachNetworkInterface.go | 130 ++++++++ spinifex/handlers/ec2/vpc/eni.go | 91 +++++- spinifex/types/ec2.go | 22 +- spinifex/vm/eni_hotplug.go | 290 +++++++++++++++++ spinifex/vm/eni_hotplug_test.go | 301 ++++++++++++++++++ spinifex/vm/errors.go | 14 + spinifex/vm/qmp_device.go | 55 +++- spinifex/vm/qmp_device_test.go | 40 ++- 12 files changed, 1253 insertions(+), 12 deletions(-) create mode 100644 spinifex/daemon/daemon_handlers_eni.go create mode 100644 spinifex/gateway/ec2/vpc/AttachNetworkInterface.go create mode 100644 spinifex/gateway/ec2/vpc/DetachNetworkInterface.go create mode 100644 spinifex/vm/eni_hotplug.go create mode 100644 spinifex/vm/eni_hotplug_test.go diff --git a/spinifex/daemon/daemon_handlers.go b/spinifex/daemon/daemon_handlers.go index c84bd7451..73b3d9255 100644 --- a/spinifex/daemon/daemon_handlers.go +++ b/spinifex/daemon/daemon_handlers.go @@ -86,6 +86,10 @@ func (d *Daemon) handleEC2Events(msg *nats.Msg) { d.handleAttachVolume(msg, command, instance) case command.Attributes.DetachVolume: d.handleDetachVolume(msg, command, instance) + case command.Attributes.AttachENI: + d.handleAttachNetworkInterface(msg, command, instance) + case command.Attributes.DetachENI: + d.handleDetachNetworkInterface(msg, command, instance) case command.Attributes.AssociateIamInstanceProfile: d.handleAssociateIamInstanceProfile(msg, command, instance) case command.Attributes.StartInstance: diff --git a/spinifex/daemon/daemon_handlers_eni.go b/spinifex/daemon/daemon_handlers_eni.go new file mode 100644 index 000000000..ee37eccb7 --- /dev/null +++ b/spinifex/daemon/daemon_handlers_eni.go @@ -0,0 +1,219 @@ +package daemon + +import ( + "encoding/json" + "errors" + "log/slog" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/awserrors" + handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc" + "github.com/mulgadc/spinifex/spinifex/types" + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/mulgadc/spinifex/spinifex/vm" + "github.com/nats-io/nats.go" +) + +// handleAttachNetworkInterface composes the existing KV-side AttachENI +// with the live QMP pipeline vm.Manager.HotPlugENI. The KV update happens +// first so a daemon crash before the QMP pipeline leaves the ENI record +// in attaching state for the Sprint 3d reconciler to converge. On QMP +// failure the KV record is rolled back to available with the error +// reason persisted in LastAttachError. +func (d *Daemon) handleAttachNetworkInterface(msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) { + slog.Info("Attaching ENI to instance", "instanceId", command.ID) + + if command.AttachENIData == nil || command.AttachENIData.NetworkInterfaceID == "" { + respondWithError(msg, awserrors.ErrorInvalidParameterValue) + return + } + + eniID := command.AttachENIData.NetworkInterfaceID + deviceIndex := command.AttachENIData.DeviceIndex + accountID := utils.AccountIDFromMsg(msg) + + if status := d.vmMgr.Status(instance); status != vm.StateRunning { + respondWithError(msg, awserrors.ErrorIncorrectInstanceState) + return + } + + record, err := d.vpcService.GetENIRecord(accountID, eniID) + if err != nil { + respondWithError(msg, errorCodeFor(err)) + return + } + if record.Status == "in-use" && record.InstanceId != instance.ID { + respondWithError(msg, awserrors.ErrorInvalidNetworkInterfaceInUse) + return + } + + attachmentID, err := d.vpcService.AttachENI(accountID, eniID, instance.ID, deviceIndex) + if err != nil { + respondWithError(msg, errorCodeFor(err)) + return + } + if err := d.vpcService.UpdateENI(accountID, eniID, func(r *handlers_ec2_vpc.ENIRecord) { + r.AttachmentStatus = "attaching" + r.LastAttachError = "" + }); err != nil { + slog.Warn("AttachNetworkInterface: failed to mark attaching state", + "eniId", eniID, "err", err) + } + + res, hotPlugErr := d.vmMgr.HotPlugENI(instance, eniID, record.MacAddress) + if hotPlugErr != nil { + slog.Error("AttachNetworkInterface: hot-plug pipeline failed, rolling back KV", + "eniId", eniID, "instanceId", instance.ID, "err", hotPlugErr) + if rollbackErr := d.vpcService.DetachENI(accountID, eniID); rollbackErr != nil { + slog.Error("AttachNetworkInterface: KV rollback failed", + "eniId", eniID, "err", rollbackErr) + } + _ = d.vpcService.UpdateENI(accountID, eniID, func(r *handlers_ec2_vpc.ENIRecord) { + r.AttachmentStatus = "" + r.HotPlugSlot = 0 + r.LastAttachError = hotPlugErr.Error() + }) + respondWithError(msg, eniHotplugErrorCode(hotPlugErr)) + return + } + + if err := d.vpcService.UpdateENI(accountID, eniID, func(r *handlers_ec2_vpc.ENIRecord) { + r.AttachmentStatus = "attached" + r.HotPlugSlot = res.Slot + }); err != nil { + slog.Warn("AttachNetworkInterface: failed to mark attached state", + "eniId", eniID, "err", err) + } + + publishENIHotplugEvent(d.natsConn, "vpc.eni-hotplug.attached", instance.ID, map[string]any{ + "eniId": eniID, + "mac": record.MacAddress, + "attachmentId": attachmentID, + "hotPlugSlot": res.Slot, + "deviceIndex": deviceIndex, + }) + + respondWithJSON(msg, ec2.AttachNetworkInterfaceOutput{ + AttachmentId: aws.String(attachmentID), + }) +} + +// handleDetachNetworkInterface reverses the attach pipeline. The KV +// record is marked detaching first so a daemon crash mid-flight leaves +// state for the reconciler; on QMP success the record returns to +// available. +func (d *Daemon) handleDetachNetworkInterface(msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) { + slog.Info("Detaching ENI from instance", "instanceId", command.ID) + + if command.DetachENIData == nil || command.DetachENIData.AttachmentID == "" { + respondWithError(msg, awserrors.ErrorInvalidParameterValue) + return + } + + attachmentID := command.DetachENIData.AttachmentID + force := command.DetachENIData.Force + accountID := utils.AccountIDFromMsg(msg) + + record, err := d.vpcService.FindENIByAttachment(accountID, attachmentID) + if err != nil { + respondWithError(msg, errorCodeFor(err)) + return + } + if record.InstanceId != instance.ID { + respondWithError(msg, awserrors.ErrorInvalidAttachmentIDNotFound) + return + } + + if status := d.vmMgr.Status(instance); status != vm.StateRunning { + respondWithError(msg, awserrors.ErrorIncorrectInstanceState) + return + } + + if err := d.vpcService.UpdateENI(accountID, record.NetworkInterfaceId, func(r *handlers_ec2_vpc.ENIRecord) { + r.AttachmentStatus = "detaching" + r.DetachInFlight = true + r.DetachForce = force + }); err != nil { + slog.Warn("DetachNetworkInterface: failed to mark detaching state", + "eniId", record.NetworkInterfaceId, "err", err) + } + + if err := d.vmMgr.HotUnplugENI(instance, record.NetworkInterfaceId, force); err != nil { + slog.Error("DetachNetworkInterface: hot-unplug pipeline failed", + "eniId", record.NetworkInterfaceId, "instanceId", instance.ID, "err", err) + _ = d.vpcService.UpdateENI(accountID, record.NetworkInterfaceId, func(r *handlers_ec2_vpc.ENIRecord) { + r.DetachInFlight = false + }) + respondWithError(msg, eniHotplugErrorCode(err)) + return + } + + if err := d.vpcService.DetachENI(accountID, record.NetworkInterfaceId); err != nil { + slog.Warn("DetachNetworkInterface: KV detach failed after QMP success", + "eniId", record.NetworkInterfaceId, "err", err) + } + _ = d.vpcService.UpdateENI(accountID, record.NetworkInterfaceId, func(r *handlers_ec2_vpc.ENIRecord) { + r.AttachmentStatus = "" + r.HotPlugSlot = 0 + r.DetachInFlight = false + r.DetachForce = false + }) + + publishENIHotplugEvent(d.natsConn, "vpc.eni-hotplug.detached", instance.ID, map[string]any{ + "eniId": record.NetworkInterfaceId, + "mac": record.MacAddress, + "attachmentId": attachmentID, + }) + + respondWithJSON(msg, ec2.DetachNetworkInterfaceOutput{}) +} + +// eniHotplugErrorCode maps a vm.Manager hot-plug error to the AWS API +// error code that the AWS SDK surfaces. +func eniHotplugErrorCode(err error) string { + switch { + case errors.Is(err, vm.ErrInstanceNotFound): + return awserrors.ErrorInvalidInstanceIDNotFound + case errors.Is(err, vm.ErrInvalidTransition): + return awserrors.ErrorIncorrectInstanceState + case errors.Is(err, vm.ErrAttachmentLimitExceeded): + return awserrors.ErrorAttachmentLimitExceeded + case errors.Is(err, vm.ErrENINotAttached): + return awserrors.ErrorInvalidAttachmentIDNotFound + case errors.Is(err, vm.ErrQMPUnavailable), + errors.Is(err, vm.ErrENIPipelineTimeout): + return awserrors.ErrorServerInternal + default: + return awserrors.ErrorServerInternal + } +} + +// errorCodeFor passes through errors whose message is already a valid AWS +// code (the convention VPCServiceImpl uses for its sentinel errors). +func errorCodeFor(err error) string { + if err == nil { + return "" + } + return awserrors.ValidErrorCode(err.Error()) +} + +// publishENIHotplugEvent emits the per-instance NATS event on best-effort +// basis — observers (Phase 4 ecs-agent, future vpcd subscribers) read it +// for fanout. A publish failure does not roll back the pipeline. +func publishENIHotplugEvent(nc *nats.Conn, subject, instanceID string, payload map[string]any) { + if nc == nil { + return + } + fullSubject := subject + "." + instanceID + data, err := json.Marshal(payload) + if err != nil { + slog.Warn("ENI hot-plug event marshal failed", + "subject", fullSubject, "err", err) + return + } + if err := nc.Publish(fullSubject, data); err != nil { + slog.Warn("ENI hot-plug event publish failed", + "subject", fullSubject, "err", err) + } +} diff --git a/spinifex/gateway/ec2.go b/spinifex/gateway/ec2.go index f729a10bd..fb5194333 100644 --- a/spinifex/gateway/ec2.go +++ b/spinifex/gateway/ec2.go @@ -362,6 +362,12 @@ var ec2Actions = map[string]EC2Handler{ "ModifyNetworkInterfaceAttribute": ec2Handler(func(input *ec2.ModifyNetworkInterfaceAttributeInput, gw *GatewayConfig, accountID string) (any, error) { return gateway_ec2_vpc.ModifyNetworkInterfaceAttribute(input, gw.NATSConn, accountID) }), + "AttachNetworkInterface": ec2Handler(func(input *ec2.AttachNetworkInterfaceInput, gw *GatewayConfig, accountID string) (any, error) { + return gateway_ec2_vpc.AttachNetworkInterface(input, gw.NATSConn, accountID) + }), + "DetachNetworkInterface": ec2Handler(func(input *ec2.DetachNetworkInterfaceInput, gw *GatewayConfig, accountID string) (any, error) { + return gateway_ec2_vpc.DetachNetworkInterface(input, gw.NATSConn, accountID) + }), "CreateSecurityGroup": ec2Handler(func(input *ec2.CreateSecurityGroupInput, gw *GatewayConfig, accountID string) (any, error) { return gateway_ec2_vpc.CreateSecurityGroup(input, gw.NATSConn, accountID) }), diff --git a/spinifex/gateway/ec2/vpc/AttachNetworkInterface.go b/spinifex/gateway/ec2/vpc/AttachNetworkInterface.go new file mode 100644 index 000000000..33fd76714 --- /dev/null +++ b/spinifex/gateway/ec2/vpc/AttachNetworkInterface.go @@ -0,0 +1,93 @@ +package gateway_ec2_vpc + +import ( + "encoding/json" + "errors" + "fmt" + "log/slog" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/awserrors" + "github.com/mulgadc/spinifex/spinifex/types" + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/nats-io/nats.go" +) + +// ValidateAttachNetworkInterfaceInput validates the input parameters. +// NetworkInterfaceId, InstanceId, and DeviceIndex are required. +func ValidateAttachNetworkInterfaceInput(input *ec2.AttachNetworkInterfaceInput) error { + if input == nil { + return errors.New(awserrors.ErrorInvalidParameterValue) + } + if input.NetworkInterfaceId == nil || *input.NetworkInterfaceId == "" { + return errors.New(awserrors.ErrorMissingParameter) + } + if input.InstanceId == nil || *input.InstanceId == "" { + return errors.New(awserrors.ErrorMissingParameter) + } + if input.DeviceIndex == nil { + return errors.New(awserrors.ErrorMissingParameter) + } + return nil +} + +// AttachNetworkInterface routes the AWS EC2 AttachNetworkInterface call +// to the daemon that owns the target instance via the per-instance +// ec2.cmd.{instanceID} channel. The daemon runs both the KV-side +// AttachENI and the live QMP hot-plug pipeline. +func AttachNetworkInterface(input *ec2.AttachNetworkInterfaceInput, natsConn *nats.Conn, accountID string) (ec2.AttachNetworkInterfaceOutput, error) { + var output ec2.AttachNetworkInterfaceOutput + + if err := ValidateAttachNetworkInterfaceInput(input); err != nil { + return output, err + } + + instanceID := *input.InstanceId + eniID := *input.NetworkInterfaceId + deviceIndex := *input.DeviceIndex + + command := types.EC2InstanceCommand{ + ID: instanceID, + Attributes: types.EC2CommandAttributes{AttachENI: true}, + AttachENIData: &types.AttachENIData{ + NetworkInterfaceID: eniID, + DeviceIndex: deviceIndex, + }, + } + + jsonData, err := json.Marshal(command) + if err != nil { + slog.Error("AttachNetworkInterface: marshal command failed", "err", err) + return output, errors.New(awserrors.ErrorServerInternal) + } + + reqMsg := nats.NewMsg(fmt.Sprintf("ec2.cmd.%s", instanceID)) + reqMsg.Data = jsonData + reqMsg.Header.Set(utils.AccountIDHeader, accountID) + msg, err := natsConn.RequestMsg(reqMsg, 30*time.Second) + if err != nil { + slog.Error("AttachNetworkInterface: NATS request failed", + "instanceId", instanceID, "eniId", eniID, "err", err) + if errors.Is(err, nats.ErrNoResponders) { + return output, errors.New(awserrors.ErrorInvalidInstanceIDNotFound) + } + return output, errors.New(awserrors.ErrorServerInternal) + } + + responseError, err := utils.ValidateErrorPayload(msg.Data) + if err != nil { + return output, errors.New(*responseError.Code) + } + + if err := json.Unmarshal(msg.Data, &output); err != nil { + slog.Error("AttachNetworkInterface: unmarshal response failed", "err", err) + return output, errors.New(awserrors.ErrorServerInternal) + } + + slog.Info("AttachNetworkInterface completed", + "instanceId", instanceID, "eniId", eniID, + "attachmentId", aws.StringValue(output.AttachmentId)) + return output, nil +} diff --git a/spinifex/gateway/ec2/vpc/DetachNetworkInterface.go b/spinifex/gateway/ec2/vpc/DetachNetworkInterface.go new file mode 100644 index 000000000..cde067897 --- /dev/null +++ b/spinifex/gateway/ec2/vpc/DetachNetworkInterface.go @@ -0,0 +1,130 @@ +package gateway_ec2_vpc + +import ( + "encoding/json" + "errors" + "fmt" + "log/slog" + "time" + + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/awserrors" + "github.com/mulgadc/spinifex/spinifex/types" + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/nats-io/nats.go" +) + +// ValidateDetachNetworkInterfaceInput validates the input parameters. +// AttachmentId is required; Force defaults to false. +func ValidateDetachNetworkInterfaceInput(input *ec2.DetachNetworkInterfaceInput) error { + if input == nil { + return errors.New(awserrors.ErrorInvalidParameterValue) + } + if input.AttachmentId == nil || *input.AttachmentId == "" { + return errors.New(awserrors.ErrorMissingParameter) + } + return nil +} + +// DetachNetworkInterface looks up the owning instance for the supplied +// attachment id and dispatches the detach command to that instance's +// daemon. The AWS API itself does not require the InstanceId, so the +// gateway resolves it via DescribeNetworkInterfaces over NATS first. +func DetachNetworkInterface(input *ec2.DetachNetworkInterfaceInput, natsConn *nats.Conn, accountID string) (ec2.DetachNetworkInterfaceOutput, error) { + var output ec2.DetachNetworkInterfaceOutput + + if err := ValidateDetachNetworkInterfaceInput(input); err != nil { + return output, err + } + + attachmentID := *input.AttachmentId + force := false + if input.Force != nil { + force = *input.Force + } + + instanceID, err := resolveAttachmentInstance(natsConn, accountID, attachmentID) + if err != nil { + return output, err + } + + command := types.EC2InstanceCommand{ + ID: instanceID, + Attributes: types.EC2CommandAttributes{DetachENI: true}, + DetachENIData: &types.DetachENIData{ + AttachmentID: attachmentID, + Force: force, + }, + } + + jsonData, err := json.Marshal(command) + if err != nil { + slog.Error("DetachNetworkInterface: marshal command failed", "err", err) + return output, errors.New(awserrors.ErrorServerInternal) + } + + reqMsg := nats.NewMsg(fmt.Sprintf("ec2.cmd.%s", instanceID)) + reqMsg.Data = jsonData + reqMsg.Header.Set(utils.AccountIDHeader, accountID) + msg, err := natsConn.RequestMsg(reqMsg, 30*time.Second) + if err != nil { + slog.Error("DetachNetworkInterface: NATS request failed", + "instanceId", instanceID, "attachmentId", attachmentID, "err", err) + if errors.Is(err, nats.ErrNoResponders) { + return output, errors.New(awserrors.ErrorInvalidInstanceIDNotFound) + } + return output, errors.New(awserrors.ErrorServerInternal) + } + + responseError, err := utils.ValidateErrorPayload(msg.Data) + if err != nil { + return output, errors.New(*responseError.Code) + } + + if err := json.Unmarshal(msg.Data, &output); err != nil { + slog.Error("DetachNetworkInterface: unmarshal response failed", "err", err) + return output, errors.New(awserrors.ErrorServerInternal) + } + + slog.Info("DetachNetworkInterface completed", + "instanceId", instanceID, "attachmentId", attachmentID) + return output, nil +} + +// resolveAttachmentInstance queries DescribeNetworkInterfaces (which fans +// out to all daemons) to find which ENI carries attachmentID and which +// instance owns it. Returns InvalidAttachmentID.NotFound when no match. +func resolveAttachmentInstance(natsConn *nats.Conn, accountID, attachmentID string) (string, error) { + input := &ec2.DescribeNetworkInterfacesInput{} + reqData, err := json.Marshal(input) + if err != nil { + return "", errors.New(awserrors.ErrorServerInternal) + } + reqMsg := nats.NewMsg("ec2.DescribeNetworkInterfaces") + reqMsg.Data = reqData + reqMsg.Header.Set(utils.AccountIDHeader, accountID) + msg, err := natsConn.RequestMsg(reqMsg, 10*time.Second) + if err != nil { + return "", errors.New(awserrors.ErrorServerInternal) + } + if _, err := utils.ValidateErrorPayload(msg.Data); err != nil { + return "", errors.New(awserrors.ErrorInvalidAttachmentIDNotFound) + } + var desc ec2.DescribeNetworkInterfacesOutput + if err := json.Unmarshal(msg.Data, &desc); err != nil { + return "", errors.New(awserrors.ErrorServerInternal) + } + for _, ni := range desc.NetworkInterfaces { + if ni.Attachment == nil || ni.Attachment.AttachmentId == nil { + continue + } + if *ni.Attachment.AttachmentId != attachmentID { + continue + } + if ni.Attachment.InstanceId == nil || *ni.Attachment.InstanceId == "" { + return "", errors.New(awserrors.ErrorInvalidAttachmentIDNotFound) + } + return *ni.Attachment.InstanceId, nil + } + return "", errors.New(awserrors.ErrorInvalidAttachmentIDNotFound) +} diff --git a/spinifex/handlers/ec2/vpc/eni.go b/spinifex/handlers/ec2/vpc/eni.go index 66b1969e4..a90494b10 100644 --- a/spinifex/handlers/ec2/vpc/eni.go +++ b/spinifex/handlers/ec2/vpc/eni.go @@ -31,7 +31,7 @@ type ENIRecord struct { PrivateIpAddress string `json:"private_ip_address"` MacAddress string `json:"mac_address"` Description string `json:"description"` - Status string `json:"status"` // available, in-use, attaching, detaching + Status string `json:"status"` // available, in-use AttachmentId string `json:"attachment_id,omitempty"` InstanceId string `json:"instance_id,omitempty"` DeviceIndex int64 `json:"device_index"` @@ -40,6 +40,26 @@ type ENIRecord struct { SecurityGroupIds []string `json:"security_group_ids,omitempty"` Tags map[string]string `json:"tags"` CreatedAt time.Time `json:"created_at"` + + // AttachmentStatus carries the hot-plug transition state independent of + // Status. AWS-parity field: "" (not transitioning), "attaching", + // "attached", "detaching", "detached". Empty on records with a non-empty + // AttachmentId pre-dating Sprint 3b is interpreted as "attached". + AttachmentStatus string `json:"attachment_status,omitempty"` + // HotPlugSlot is the PCIe root-port slot index (1-based) into which the + // ENI was hot-plugged. Zero means the ENI is not on the hot-plug pool + // (boot-time ENI or KV-only attach on a stopped instance). + HotPlugSlot int `json:"hot_plug_slot,omitempty"` + // LastAttachError captures the most recent attach-pipeline failure + // reason so DescribeNetworkInterfaces can surface it. Cleared on the + // next successful attach. + LastAttachError string `json:"last_attach_error,omitempty"` + // DetachInFlight signals that a detach pipeline is mid-flight; persists + // across daemon restart so the reconciler picks up where it left off. + DetachInFlight bool `json:"detach_in_flight,omitempty"` + // DetachForce records the force flag from the in-flight detach call so + // the reconciler can replay with the original semantics. + DetachForce bool `json:"detach_force,omitempty"` } // CreateNetworkInterface creates a new ENI in the specified subnet @@ -497,6 +517,75 @@ func (s *VPCServiceImpl) DetachENI(accountID, eniId string) error { return nil } +// GetENIRecord reads the ENIRecord for eniId scoped to accountID. Used by +// hot-plug handlers that need the record's MAC + AttachmentId before +// running the QMP pipeline. +func (s *VPCServiceImpl) GetENIRecord(accountID, eniId string) (ENIRecord, error) { + key := utils.AccountKey(accountID, eniId) + entry, err := s.eniKV.Get(key) + if err != nil { + return ENIRecord{}, errors.New(awserrors.ErrorInvalidNetworkInterfaceIDNotFound) + } + var record ENIRecord + if err := json.Unmarshal(entry.Value(), &record); err != nil { + return ENIRecord{}, errors.New(awserrors.ErrorServerInternal) + } + return record, nil +} + +// UpdateENI applies fn to the ENIRecord identified by eniId and persists +// the result under the same revision token (compare-and-swap). fn must +// not block on external I/O; the KV update slot is contended. +func (s *VPCServiceImpl) UpdateENI(accountID, eniId string, fn func(*ENIRecord)) error { + key := utils.AccountKey(accountID, eniId) + entry, err := s.eniKV.Get(key) + if err != nil { + return errors.New(awserrors.ErrorInvalidNetworkInterfaceIDNotFound) + } + var record ENIRecord + if err := json.Unmarshal(entry.Value(), &record); err != nil { + return errors.New(awserrors.ErrorServerInternal) + } + fn(&record) + data, err := json.Marshal(record) + if err != nil { + return fmt.Errorf("marshal ENI record: %w", err) + } + if _, err := s.eniKV.Update(key, data, entry.Revision()); err != nil { + return errors.New(awserrors.ErrorServerInternal) + } + return nil +} + +// FindENIByAttachment scans the per-account ENI bucket for a record whose +// AttachmentId matches attachmentId. Returns NotFound semantics when no +// match exists. Used by DetachNetworkInterface, whose AWS input identifies +// the attachment by id rather than by ENI id. +func (s *VPCServiceImpl) FindENIByAttachment(accountID, attachmentId string) (ENIRecord, error) { + keys, err := s.eniKV.Keys() + if err != nil && !errors.Is(err, nats.ErrNoKeysFound) { + return ENIRecord{}, errors.New(awserrors.ErrorServerInternal) + } + prefix := accountID + "." + for _, key := range keys { + if key == utils.VersionKey || !strings.HasPrefix(key, prefix) { + continue + } + entry, err := s.eniKV.Get(key) + if err != nil { + continue + } + var record ENIRecord + if err := json.Unmarshal(entry.Value(), &record); err != nil { + continue + } + if record.AttachmentId == attachmentId { + return record, nil + } + } + return ENIRecord{}, errors.New(awserrors.ErrorInvalidAttachmentIDNotFound) +} + // UpdateENIPublicIP updates the PublicIpAddress and PublicIpPool on an ENI record. func (s *VPCServiceImpl) UpdateENIPublicIP(accountID, eniId, publicIP, poolName string) error { key := utils.AccountKey(accountID, eniId) diff --git a/spinifex/types/ec2.go b/spinifex/types/ec2.go index bb9b7f0f7..8c914d938 100644 --- a/spinifex/types/ec2.go +++ b/spinifex/types/ec2.go @@ -1,13 +1,16 @@ package types // EC2InstanceCommand is the NATS wire format for EC2 instance commands -// (stop, terminate, start, attach-volume, detach-volume). -// It replaces direct use of qmp.Command on the gateway→daemon boundary. +// (stop, terminate, start, attach-volume, detach-volume, attach-eni, +// detach-eni). It replaces direct use of qmp.Command on the gateway→daemon +// boundary. type EC2InstanceCommand struct { ID string `json:"id"` Attributes EC2CommandAttributes `json:"attributes"` AttachVolumeData *AttachVolumeData `json:"attach_volume_data,omitempty"` DetachVolumeData *DetachVolumeData `json:"detach_volume_data,omitempty"` + AttachENIData *AttachENIData `json:"attach_eni_data,omitempty"` + DetachENIData *DetachENIData `json:"detach_eni_data,omitempty"` IamProfileAssociationData *IamProfileAssociationData `json:"iam_profile_association_data,omitempty"` } @@ -19,6 +22,8 @@ type EC2CommandAttributes struct { AttachVolume bool `json:"attach_volume"` DetachVolume bool `json:"detach_volume"` RebootInstance bool `json:"reboot_instance"` + AttachENI bool `json:"attach_eni"` + DetachENI bool `json:"detach_eni"` AssociateIamInstanceProfile bool `json:"associate_iam_instance_profile,omitempty"` } @@ -35,6 +40,19 @@ type DetachVolumeData struct { Force bool `json:"force,omitempty"` } +// AttachENIData carries parameters for an attach-network-interface command. +type AttachENIData struct { + NetworkInterfaceID string `json:"network_interface_id"` + DeviceIndex int64 `json:"device_index"` +} + +// DetachENIData carries parameters for a detach-network-interface command. +// AttachmentID is the AWS attachment ID returned by AttachNetworkInterface. +type DetachENIData struct { + AttachmentID string `json:"attachment_id"` + Force bool `json:"force,omitempty"` +} + // IamProfileAssociationData carries parameters for an associate-iam-instance-profile // command. The gateway has already resolved the profile reference to a canonical // ARN and enforced iam:PassRole — the daemon only needs the ARN to persist on diff --git a/spinifex/vm/eni_hotplug.go b/spinifex/vm/eni_hotplug.go new file mode 100644 index 000000000..c46909c2a --- /dev/null +++ b/spinifex/vm/eni_hotplug.go @@ -0,0 +1,290 @@ +package vm + +import ( + "fmt" + "log/slog" + "time" +) + +// eniPipelineSettings governs the query-pci polling cadence of the +// HotPlugENI / HotUnplugENI pipelines. Overridable for tests via the +// package-level seam below. +type eniPipelineSettings struct { + AttachPollInterval time.Duration + AttachPollMax int + DetachPollInterval time.Duration + DetachPollMaxSoft int // !force timeout = interval * max + DetachPollMaxForce int // force=true short-circuits after this many polls +} + +var eniPipeline = eniPipelineSettings{ + AttachPollInterval: 200 * time.Millisecond, + AttachPollMax: 25, // 5s total + DetachPollInterval: 200 * time.Millisecond, + DetachPollMaxSoft: 25, // 5s for graceful detach + DetachPollMaxForce: 5, // 1s for force=true +} + +// newDeviceController is the production factory bound to the live +// QMPClient on the VM. Tests override this seam to inject a +// StubDeviceController without touching the VM's QMP socket. +var newDeviceController = func(v *VM) DeviceController { + return NewQMPDeviceController(v.QMPClient, v.ID) +} + +// HotPlugENIResult carries the slot index assigned to a successful +// hot-plug. Returned to the daemon handler so the AttachmentId KV record +// can be annotated with HotPlugSlot before the NATS success event fires. +type HotPlugENIResult struct { + Slot int +} + +// HotPlugENI runs the 4-step attach pipeline against the live VM: +// +// 1. tap device creation (placeholder — wired in Sprint 3c) +// 2. ovs-vsctl add-port (placeholder — wired in Sprint 3c) +// 3. QMP netdev_add (real) +// 4. QMP device_add + poll (real) +// +// On step-N failure, steps 1..N-1 are rolled back in reverse order and +// the slot is returned to ENIRequests.AvailableSlots. The instance must +// be running and have a live QMPClient. +// +// Caller (daemon handler) owns the KV side-effects: marking +// AttachmentStatus=attached and publishing the +// vpc.eni-hotplug.attached.{instanceID} event happens after this returns +// nil. +func (m *Manager) HotPlugENI(instance *VM, eniID, mac string) (HotPlugENIResult, error) { + if instance == nil { + return HotPlugENIResult{}, ErrInstanceNotFound + } + if status := m.Status(instance); status != StateRunning { + return HotPlugENIResult{}, fmt.Errorf("%w: cannot hot-plug ENI on instance %s in state %s", + ErrInvalidTransition, instance.ID, status) + } + if instance.QMPClient == nil { + return HotPlugENIResult{}, fmt.Errorf("%w: instance %s", ErrQMPUnavailable, instance.ID) + } + + instance.ENIRequests.Mu.Lock() + defer instance.ENIRequests.Mu.Unlock() + + if existing, ok := instance.ENIRequests.AttachedByENIID[eniID]; ok { + // Idempotent re-attach against the same instance is a noop — + // returning the existing slot lets the handler treat the call as + // a successful state confirmation. + return HotPlugENIResult{Slot: existing}, nil + } + + if len(instance.ENIRequests.AvailableSlots) == 0 { + return HotPlugENIResult{}, ErrAttachmentLimitExceeded + } + slot := instance.ENIRequests.AvailableSlots[0] + instance.ENIRequests.AvailableSlots = instance.ENIRequests.AvailableSlots[1:] + if instance.ENIRequests.AttachedByENIID == nil { + instance.ENIRequests.AttachedByENIID = make(map[string]int) + } + instance.ENIRequests.AttachedByENIID[eniID] = slot + + dc := newDeviceController(instance) + netdevID := fmt.Sprintf("hostnet-eni-%d", slot) + deviceID := fmt.Sprintf("net-eni-%d", slot) + tapName := TapDeviceName(eniID) + busID := fmt.Sprintf("hotplug-eni%d", slot) + + // Step 1: tap device. Sprint 3c replaces this stub with a real call + // to NetworkPlumber (ip tuntap add). + if err := stubTapAdd(instance.ID, eniID, tapName); err != nil { + m.releaseSlotLocked(instance, eniID, slot) + return HotPlugENIResult{}, fmt.Errorf("tap add (stub): %w", err) + } + + // Step 2: OVS port. Sprint 3c replaces this stub with the real + // ovs-vsctl add-port invocation against br-int. + if err := stubOVSAddPort(instance.ID, eniID, tapName, mac); err != nil { + _ = stubTapDel(instance.ID, eniID, tapName) + m.releaseSlotLocked(instance, eniID, slot) + return HotPlugENIResult{}, fmt.Errorf("ovs add-port (stub): %w", err) + } + + // Step 3: QMP netdev_add. + if err := dc.NetdevAdd(map[string]any{ + "type": "tap", + "id": netdevID, + "ifname": tapName, + "script": "no", + "downscript": "no", + }); err != nil { + _ = stubOVSDelPort(instance.ID, eniID, tapName) + _ = stubTapDel(instance.ID, eniID, tapName) + m.releaseSlotLocked(instance, eniID, slot) + return HotPlugENIResult{}, fmt.Errorf("QMP netdev_add: %w", err) + } + + // Step 4: QMP device_add then poll query-pci until materialized. + if err := dc.DeviceAdd(map[string]any{ + "driver": "virtio-net-pci", + "id": deviceID, + "bus": busID, + "netdev": netdevID, + "mac": mac, + }); err != nil { + _ = dc.NetdevDel(netdevID) + _ = stubOVSDelPort(instance.ID, eniID, tapName) + _ = stubTapDel(instance.ID, eniID, tapName) + m.releaseSlotLocked(instance, eniID, slot) + return HotPlugENIResult{}, fmt.Errorf("QMP device_add: %w", err) + } + + if err := waitForPCIDevice(dc, deviceID, true); err != nil { + _ = dc.DeviceDel(deviceID) + _ = dc.NetdevDel(netdevID) + _ = stubOVSDelPort(instance.ID, eniID, tapName) + _ = stubTapDel(instance.ID, eniID, tapName) + m.releaseSlotLocked(instance, eniID, slot) + return HotPlugENIResult{}, fmt.Errorf("query-pci wait (attach): %w", err) + } + + slog.Info("ENI hot-plugged", + "instanceId", instance.ID, "eniId", eniID, "slot", slot, + "deviceId", deviceID, "netdevId", netdevID, "tap", tapName) + return HotPlugENIResult{Slot: slot}, nil +} + +// HotUnplugENI runs the reverse 4-step detach pipeline. force=true +// shortens the post-device_del poll budget so an unresponsive guest does +// not block the call — at the cost of leaving the guest with a dangling +// virtio-net device until reboot if the kernel never releases it. +func (m *Manager) HotUnplugENI(instance *VM, eniID string, force bool) error { + if instance == nil { + return ErrInstanceNotFound + } + if status := m.Status(instance); status != StateRunning { + return fmt.Errorf("%w: cannot hot-unplug ENI from instance %s in state %s", + ErrInvalidTransition, instance.ID, status) + } + if instance.QMPClient == nil { + return fmt.Errorf("%w: instance %s", ErrQMPUnavailable, instance.ID) + } + + instance.ENIRequests.Mu.Lock() + defer instance.ENIRequests.Mu.Unlock() + + slot, ok := instance.ENIRequests.AttachedByENIID[eniID] + if !ok { + return fmt.Errorf("%w: %s", ErrENINotAttached, eniID) + } + + dc := newDeviceController(instance) + netdevID := fmt.Sprintf("hostnet-eni-%d", slot) + deviceID := fmt.Sprintf("net-eni-%d", slot) + tapName := TapDeviceName(eniID) + + // Step 1: device_del (guest driver release request). + if err := dc.DeviceDel(deviceID); err != nil { + if !force { + return fmt.Errorf("QMP device_del: %w", err) + } + slog.Warn("HotUnplugENI: device_del failed (force=true, continuing)", + "instanceId", instance.ID, "eniId", eniID, "err", err) + } + + // Step 2: poll query-pci for absence. force=true uses a shorter budget. + if err := waitForPCIDevice(dc, deviceID, false); err != nil { + if !force { + return fmt.Errorf("query-pci wait (detach): %w", err) + } + slog.Warn("HotUnplugENI: device removal not observed (force=true, continuing)", + "instanceId", instance.ID, "eniId", eniID, "err", err) + } + + // Step 3: QMP netdev_del. + if err := dc.NetdevDel(netdevID); err != nil { + slog.Warn("HotUnplugENI: netdev_del failed (continuing detach)", + "instanceId", instance.ID, "eniId", eniID, "err", err) + } + + // Step 4: OVS port + tap (stubs, real wiring in Sprint 3c). + _ = stubOVSDelPort(instance.ID, eniID, tapName) + _ = stubTapDel(instance.ID, eniID, tapName) + + delete(instance.ENIRequests.AttachedByENIID, eniID) + instance.ENIRequests.AvailableSlots = append(instance.ENIRequests.AvailableSlots, slot) + + slog.Info("ENI hot-unplugged", + "instanceId", instance.ID, "eniId", eniID, "slot", slot, "force", force) + return nil +} + +// releaseSlotLocked returns slot to the free-list and clears the +// AttachedByENIID entry. Caller must hold instance.ENIRequests.Mu. +func (m *Manager) releaseSlotLocked(instance *VM, eniID string, slot int) { + delete(instance.ENIRequests.AttachedByENIID, eniID) + instance.ENIRequests.AvailableSlots = append(instance.ENIRequests.AvailableSlots, slot) +} + +// waitForPCIDevice polls QueryPCI until presence (wantPresent=true) or +// absence (wantPresent=false) of deviceID is observed. Attach and detach +// use distinct poll budgets via the package-level eniPipeline settings. +func waitForPCIDevice(dc DeviceController, deviceID string, wantPresent bool) error { + interval := eniPipeline.AttachPollInterval + maxAttempts := eniPipeline.AttachPollMax + if !wantPresent { + interval = eniPipeline.DetachPollInterval + maxAttempts = eniPipeline.DetachPollMaxSoft + } + for attempt := 1; attempt <= maxAttempts; attempt++ { + devs, err := dc.QueryPCI() + if err != nil { + return fmt.Errorf("query-pci attempt %d: %w", attempt, err) + } + found := false + for _, d := range devs { + if d.QDevID == deviceID { + found = true + break + } + } + if found == wantPresent { + return nil + } + if attempt < maxAttempts { + eniPipelineSleep(interval) + } + } + return fmt.Errorf("%w: device %s want_present=%v", ErrENIPipelineTimeout, deviceID, wantPresent) +} + +// eniPipelineSleep is the sleep seam used by waitForPCIDevice. Tests +// override it to drive poll cadence without burning wall-clock time. +var eniPipelineSleep = time.Sleep + +// stubTapAdd / stubTapDel / stubOVSAddPort / stubOVSDelPort emit a debug +// log and return nil. Sprint 3c replaces these with real +// NetworkPlumber.SetupTap / TeardownTap + ovs-vsctl invocations. The +// pipeline structure and rollback semantics are stable in 3b so 3c is a +// drop-in replacement. + +func stubTapAdd(instanceID, eniID, tapName string) error { + slog.Debug("ENI hot-plug step 1 (tap add) — stubbed in Sprint 3b", + "instanceId", instanceID, "eniId", eniID, "tap", tapName) + return nil +} + +func stubTapDel(instanceID, eniID, tapName string) error { + slog.Debug("ENI hot-plug rollback (tap del) — stubbed in Sprint 3b", + "instanceId", instanceID, "eniId", eniID, "tap", tapName) + return nil +} + +func stubOVSAddPort(instanceID, eniID, tapName, mac string) error { + slog.Debug("ENI hot-plug step 2 (ovs add-port) — stubbed in Sprint 3b", + "instanceId", instanceID, "eniId", eniID, "tap", tapName, "mac", mac) + return nil +} + +func stubOVSDelPort(instanceID, eniID, tapName string) error { + slog.Debug("ENI hot-plug rollback (ovs del-port) — stubbed in Sprint 3b", + "instanceId", instanceID, "eniId", eniID, "tap", tapName) + return nil +} diff --git a/spinifex/vm/eni_hotplug_test.go b/spinifex/vm/eni_hotplug_test.go new file mode 100644 index 000000000..981d35937 --- /dev/null +++ b/spinifex/vm/eni_hotplug_test.go @@ -0,0 +1,301 @@ +package vm + +import ( + "errors" + "fmt" + "slices" + "strings" + "sync" + "testing" + "time" + + "github.com/mulgadc/spinifex/spinifex/qmp" + "github.com/mulgadc/spinifex/spinifex/types" +) + +// newHotPlugTestVM returns a VM in StateRunning with a non-nil QMPClient +// and a pre-populated ENIRequests allocator. The stub controller is +// substituted via the newDeviceController seam for the duration of the +// test (cleanup restores the production factory). +func newHotPlugTestVM(t *testing.T, slots int) (*Manager, *VM, *StubDeviceController) { + t.Helper() + mgr := NewManager() + stub := NewStubDeviceController() + available := make([]int, 0, slots) + for i := 1; i <= slots; i++ { + available = append(available, i) + } + v := &VM{ + ID: "i-test", + Status: StateRunning, + QMPClient: &qmp.QMPClient{}, + ENIRequests: types.ENIRequests{ + AvailableSlots: available, + AttachedByENIID: make(map[string]int), + }, + } + mgr.Insert(v) + + origFactory := newDeviceController + newDeviceController = func(*VM) DeviceController { return stub } + origSleep := eniPipelineSleep + eniPipelineSleep = func(time.Duration) {} + t.Cleanup(func() { + newDeviceController = origFactory + eniPipelineSleep = origSleep + }) + return mgr, v, stub +} + +func TestHotPlugENI_SuccessPath(t *testing.T) { + mgr, v, stub := newHotPlugTestVM(t, 2) + + res, err := mgr.HotPlugENI(v, "eni-aaaa", "02:00:00:00:00:01") + if err != nil { + t.Fatalf("HotPlugENI: %v", err) + } + if res.Slot != 1 { + t.Errorf("slot = %d, want 1", res.Slot) + } + if !stub.HasDevice("net-eni-1") { + t.Errorf("device net-eni-1 not attached in stub") + } + if !stub.HasNetdev("hostnet-eni-1") { + t.Errorf("netdev hostnet-eni-1 not attached in stub") + } + if got, ok := v.ENIRequests.AttachedByENIID["eni-aaaa"]; !ok || got != 1 { + t.Errorf("AttachedByENIID[eni-aaaa] = %v ok=%v, want 1 true", got, ok) + } + if len(v.ENIRequests.AvailableSlots) != 1 || v.ENIRequests.AvailableSlots[0] != 2 { + t.Errorf("AvailableSlots = %v, want [2]", v.ENIRequests.AvailableSlots) + } +} + +func TestHotPlugENI_CallOrder(t *testing.T) { + mgr, v, stub := newHotPlugTestVM(t, 1) + + if _, err := mgr.HotPlugENI(v, "eni-aaaa", "02:00:00:00:00:01"); err != nil { + t.Fatalf("HotPlugENI: %v", err) + } + calls := stub.Calls() + want := []string{"netdev_add", "device_add", "query-pci"} + if len(calls) < len(want) { + t.Fatalf("call count = %d, want >= %d (%v)", len(calls), len(want), calls) + } + for i, w := range want { + if calls[i].Execute != w { + t.Errorf("call[%d] = %s, want %s", i, calls[i].Execute, w) + } + } +} + +func TestHotPlugENI_DeviceAddFailureRollsBack(t *testing.T) { + mgr, v, stub := newHotPlugTestVM(t, 2) + wantErr := errors.New("device_add boom") + stub.SetFailNext("device_add", wantErr) + + _, err := mgr.HotPlugENI(v, "eni-aaaa", "02:00:00:00:00:01") + if err == nil || !strings.Contains(err.Error(), "device_add") { + t.Fatalf("HotPlugENI err = %v, want device_add failure", err) + } + if stub.HasDevice("net-eni-1") { + t.Errorf("device net-eni-1 left attached after rollback") + } + if stub.HasNetdev("hostnet-eni-1") { + t.Errorf("netdev hostnet-eni-1 left attached after rollback") + } + if _, ok := v.ENIRequests.AttachedByENIID["eni-aaaa"]; ok { + t.Errorf("AttachedByENIID still tracks eni-aaaa after rollback") + } + if !containsSlot(v.ENIRequests.AvailableSlots, 1) { + t.Errorf("slot 1 not returned to free-list: %v", v.ENIRequests.AvailableSlots) + } +} + +func TestHotPlugENI_NetdevAddFailureRollsBack(t *testing.T) { + mgr, v, stub := newHotPlugTestVM(t, 1) + stub.SetFailNext("netdev_add", errors.New("netdev boom")) + + _, err := mgr.HotPlugENI(v, "eni-aaaa", "02:00:00:00:00:01") + if err == nil { + t.Fatalf("HotPlugENI: want error, got nil") + } + if stub.HasNetdev("hostnet-eni-1") { + t.Errorf("netdev should not exist after netdev_add failure") + } + if !containsSlot(v.ENIRequests.AvailableSlots, 1) { + t.Errorf("slot 1 not returned to free-list") + } +} + +func TestHotPlugENI_AttachmentLimitExceeded(t *testing.T) { + mgr, v, _ := newHotPlugTestVM(t, 0) + _, err := mgr.HotPlugENI(v, "eni-aaaa", "02:00:00:00:00:01") + if !errors.Is(err, ErrAttachmentLimitExceeded) { + t.Fatalf("err = %v, want ErrAttachmentLimitExceeded", err) + } +} + +func TestHotPlugENI_IdempotentReattach(t *testing.T) { + mgr, v, stub := newHotPlugTestVM(t, 2) + + res1, err := mgr.HotPlugENI(v, "eni-aaaa", "02:00:00:00:00:01") + if err != nil { + t.Fatalf("first attach: %v", err) + } + callsBefore := len(stub.Calls()) + res2, err := mgr.HotPlugENI(v, "eni-aaaa", "02:00:00:00:00:01") + if err != nil { + t.Fatalf("idempotent reattach: %v", err) + } + if res1.Slot != res2.Slot { + t.Errorf("idempotent slot drift: %d vs %d", res1.Slot, res2.Slot) + } + if got := len(stub.Calls()); got != callsBefore { + t.Errorf("second attach issued %d new calls, want 0", got-callsBefore) + } +} + +func TestHotPlugENI_RejectsNonRunning(t *testing.T) { + mgr, v, _ := newHotPlugTestVM(t, 1) + v.Status = StateStopped + _, err := mgr.HotPlugENI(v, "eni-aaaa", "02:00:00:00:00:01") + if !errors.Is(err, ErrInvalidTransition) { + t.Fatalf("err = %v, want ErrInvalidTransition", err) + } +} + +func TestHotPlugENI_RejectsNilQMP(t *testing.T) { + mgr, v, _ := newHotPlugTestVM(t, 1) + v.QMPClient = nil + _, err := mgr.HotPlugENI(v, "eni-aaaa", "02:00:00:00:00:01") + if !errors.Is(err, ErrQMPUnavailable) { + t.Fatalf("err = %v, want ErrQMPUnavailable", err) + } +} + +func TestHotUnplugENI_SuccessFreesSlot(t *testing.T) { + mgr, v, stub := newHotPlugTestVM(t, 2) + if _, err := mgr.HotPlugENI(v, "eni-aaaa", "02:00:00:00:00:01"); err != nil { + t.Fatalf("attach: %v", err) + } + if err := mgr.HotUnplugENI(v, "eni-aaaa", false); err != nil { + t.Fatalf("detach: %v", err) + } + if stub.HasDevice("net-eni-1") { + t.Errorf("device should be gone after detach") + } + if stub.HasNetdev("hostnet-eni-1") { + t.Errorf("netdev should be gone after detach") + } + if _, ok := v.ENIRequests.AttachedByENIID["eni-aaaa"]; ok { + t.Errorf("AttachedByENIID still tracks eni-aaaa") + } + if !containsSlot(v.ENIRequests.AvailableSlots, 1) { + t.Errorf("slot 1 not returned to free-list: %v", v.ENIRequests.AvailableSlots) + } +} + +func TestHotUnplugENI_NotAttached(t *testing.T) { + mgr, v, _ := newHotPlugTestVM(t, 1) + err := mgr.HotUnplugENI(v, "eni-missing", false) + if !errors.Is(err, ErrENINotAttached) { + t.Fatalf("err = %v, want ErrENINotAttached", err) + } +} + +func TestHotUnplugENI_DeviceDelFailureWithoutForce(t *testing.T) { + mgr, v, stub := newHotPlugTestVM(t, 1) + if _, err := mgr.HotPlugENI(v, "eni-aaaa", "02:00:00:00:00:01"); err != nil { + t.Fatalf("attach: %v", err) + } + stub.SetFailNext("device_del", errors.New("device_del boom")) + + err := mgr.HotUnplugENI(v, "eni-aaaa", false) + if err == nil { + t.Fatalf("want error, got nil") + } + if !stub.HasDevice("net-eni-1") { + t.Errorf("device should still be present after device_del failure (no force)") + } +} + +func TestHotUnplugENI_ForceContinuesPastFailure(t *testing.T) { + mgr, v, stub := newHotPlugTestVM(t, 1) + if _, err := mgr.HotPlugENI(v, "eni-aaaa", "02:00:00:00:00:01"); err != nil { + t.Fatalf("attach: %v", err) + } + stub.SetFailNext("device_del", errors.New("device_del boom")) + + // query-pci poll will see device still present until we manually nuke it. + // Pre-clear so the wait-for-absence loop exits quickly under force. + stub.mu.Lock() + delete(stub.devices, "net-eni-1") + stub.mu.Unlock() + + if err := mgr.HotUnplugENI(v, "eni-aaaa", true); err != nil { + t.Fatalf("force detach: %v", err) + } + if _, ok := v.ENIRequests.AttachedByENIID["eni-aaaa"]; ok { + t.Errorf("AttachedByENIID still tracks eni-aaaa after force detach") + } +} + +func TestWaitForPCIDevice_TimeoutOnAttach(t *testing.T) { + stub := NewStubDeviceController() + origSleep := eniPipelineSleep + origMax := eniPipeline.AttachPollMax + eniPipelineSleep = func(time.Duration) {} + eniPipeline.AttachPollMax = 3 + t.Cleanup(func() { + eniPipelineSleep = origSleep + eniPipeline.AttachPollMax = origMax + }) + + err := waitForPCIDevice(stub, "net-eni-99", true) + if !errors.Is(err, ErrENIPipelineTimeout) { + t.Fatalf("err = %v, want ErrENIPipelineTimeout", err) + } +} + +func TestHotPlugENI_ConcurrentAttachesSerialize(t *testing.T) { + mgr, v, _ := newHotPlugTestVM(t, 3) + + var wg sync.WaitGroup + results := make(chan int, 3) + errs := make(chan error, 3) + for i := range 3 { + wg.Add(1) + eniID := fmt.Sprintf("eni-%04d", i) + mac := fmt.Sprintf("02:00:00:00:00:%02d", i+1) + go func() { + defer wg.Done() + res, err := mgr.HotPlugENI(v, eniID, mac) + if err != nil { + errs <- err + return + } + results <- res.Slot + }() + } + wg.Wait() + close(results) + close(errs) + for e := range errs { + t.Errorf("concurrent attach failed: %v", e) + } + seen := map[int]bool{} + for s := range results { + if seen[s] { + t.Errorf("slot %d returned twice", s) + } + seen[s] = true + } + if len(seen) != 3 { + t.Errorf("got %d distinct slots, want 3", len(seen)) + } +} + +func containsSlot(slots []int, s int) bool { + return slices.Contains(slots, s) +} diff --git a/spinifex/vm/errors.go b/spinifex/vm/errors.go index 55e886894..6eb5de11d 100644 --- a/spinifex/vm/errors.go +++ b/spinifex/vm/errors.go @@ -32,3 +32,17 @@ var ErrVolumeDeviceMismatch = errors.New("volume device name mismatch") // or may not have started serving NBD; AttachVolume must invoke UnmountOne // defensively so a half-started backend mount is not orphaned. var ErrMountAmbiguous = errors.New("ebs.mount succeeded with empty URI") + +// ErrENINotAttached is returned by HotUnplugENI when the supplied ENI id +// is not present in the instance's ENIRequests.AttachedByENIID map. +var ErrENINotAttached = errors.New("ENI not attached to instance") + +// ErrENIPipelineTimeout is returned by HotPlugENI / HotUnplugENI when the +// query-pci poll did not observe the expected materialization (attach) or +// removal (detach) within the configured budget. +var ErrENIPipelineTimeout = errors.New("ENI hot-plug pipeline timed out") + +// ErrQMPUnavailable is returned by hot-plug entry points when the +// instance has no live QMPClient (daemon restart mid-launch, terminated +// VM, etc). +var ErrQMPUnavailable = errors.New("QMP unavailable for instance") diff --git a/spinifex/vm/qmp_device.go b/spinifex/vm/qmp_device.go index 3804d17b1..ca31c8d94 100644 --- a/spinifex/vm/qmp_device.go +++ b/spinifex/vm/qmp_device.go @@ -1,6 +1,7 @@ package vm import ( + "encoding/json" "fmt" "maps" "sync" @@ -76,11 +77,57 @@ func (c *QMPDeviceController) NetdevDel(netdevID string) error { return err } -// QueryPCI is unwired in Sprint 3a; the live query-pci response parser -// lands alongside the hot-plug pipeline in Sprint 3b. Callers (the stub -// backend below + future eni_hotplug.go) get the contract today. +// QueryPCI issues query-pci and flattens the nested QEMU response into a +// list of PCIDevice entries. The response is a per-bus list whose devices +// may themselves carry a "pci_bridge" subfield containing a nested +// "devices" list (every pcie-root-port surfaces this way), so the parse +// recurses through bridges. Only devices with a non-empty qdev_id are +// returned — the hot-plug pipeline only needs to detect materialization +// of a known id. func (c *QMPDeviceController) QueryPCI() ([]PCIDevice, error) { - return nil, fmt.Errorf("QueryPCI: live query-pci wiring lands in Sprint 3b (enihotplug)") + resp, err := sendQMPCommand(c.client, qmp.QMPCommand{Execute: "query-pci"}, c.instanceID) + if err != nil { + return nil, fmt.Errorf("query-pci: %w", err) + } + var buses []pciBusInfo + if err := json.Unmarshal(resp.Return, &buses); err != nil { + return nil, fmt.Errorf("parse query-pci response: %w", err) + } + var out []PCIDevice + for _, bus := range buses { + out = appendPCIDevices(out, bus.Devices) + } + return out, nil +} + +// pciBusInfo and pciDeviceInfo mirror only the fields the hot-plug +// pipeline reads from query-pci. Full QEMU output is much richer +// (class_info, regions, irq, …) but we ignore everything else to keep the +// parser stable against QMP version drift. +type pciBusInfo struct { + Bus int `json:"bus"` + Devices []pciDeviceInfo `json:"devices"` +} + +type pciDeviceInfo struct { + Bus int `json:"bus"` + Slot int `json:"slot"` + QDevID string `json:"qdev_id"` + PCIBridge *struct { + Devices []pciDeviceInfo `json:"devices"` + } `json:"pci_bridge,omitempty"` +} + +func appendPCIDevices(dst []PCIDevice, devs []pciDeviceInfo) []PCIDevice { + for _, d := range devs { + if d.QDevID != "" { + dst = append(dst, PCIDevice{Bus: d.Bus, Slot: d.Slot, QDevID: d.QDevID}) + } + if d.PCIBridge != nil { + dst = appendPCIDevices(dst, d.PCIBridge.Devices) + } + } + return dst } // StubDeviceController is an in-memory DeviceController for unit tests. It diff --git a/spinifex/vm/qmp_device_test.go b/spinifex/vm/qmp_device_test.go index cb48d5d40..416d54af4 100644 --- a/spinifex/vm/qmp_device_test.go +++ b/spinifex/vm/qmp_device_test.go @@ -90,11 +90,41 @@ func TestStubDeviceController_CallsSnapshotIsCopy(t *testing.T) { assert.Equal(t, "device_add", again[0].Execute, "Calls() must return independent copies") } -func TestQMPDeviceController_QueryPCIUnwiredUntil3b(t *testing.T) { - c := NewQMPDeviceController(&qmp.QMPClient{}, "i-test") - _, err := c.QueryPCI() - require.Error(t, err) - assert.Contains(t, err.Error(), "Sprint 3b") +func TestQMPDeviceController_QueryPCIFlattensBridges(t *testing.T) { + // query-pci returns one bus whose root devices include pcie-root-ports + // (each itself a PCI-PCI bridge). The hot-plugged virtio-net device + // sits under the bridge. The parser must recurse and flatten. + client, cancel := newMockQMPClient(t, func(cmd qmp.QMPCommand) map[string]any { + require.Equal(t, "query-pci", cmd.Execute) + return map[string]any{ + "return": []any{ + map[string]any{ + "bus": 0, + "devices": []any{ + map[string]any{"bus": 0, "slot": 1, "qdev_id": "primary"}, + map[string]any{ + "bus": 0, "slot": 2, "qdev_id": "hotplug-eni1", + "pci_bridge": map[string]any{ + "devices": []any{ + map[string]any{"bus": 5, "slot": 0, "qdev_id": "net-eni-1"}, + }, + }, + }, + }, + }, + }, + } + }) + defer cancel() + + c := NewQMPDeviceController(client, "i-test") + pci, err := c.QueryPCI() + require.NoError(t, err) + ids := make([]string, 0, len(pci)) + for _, d := range pci { + ids = append(ids, d.QDevID) + } + assert.ElementsMatch(t, []string{"primary", "hotplug-eni1", "net-eni-1"}, ids) } func TestQMPDeviceController_DispatchesExpectedCommands(t *testing.T) {