feat: ENI hot-plug Sprint 3b — live QMP pipeline + AttachENI wiring (stage 2) (#258)

* 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
This commit is contained in:
Josh Siviour
2026-05-27 22:54:46 +10:00
committed by GitHub
parent a8e356b3d2
commit fa09228a36
12 changed files with 1253 additions and 12 deletions
+4
View File
@@ -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:
+219
View File
@@ -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)
}
}
+6
View File
@@ -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)
}),
@@ -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
}
@@ -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)
}
+90 -1
View File
@@ -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)
+20 -2
View File
@@ -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
+290
View File
@@ -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
}
+301
View File
@@ -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)
}
+14
View File
@@ -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")
+51 -4
View File
@@ -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
+35 -5
View File
@@ -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) {