diff --git a/spinifex/vm/record.go b/spinifex/vm/record.go new file mode 100644 index 000000000..9745f9b2b --- /dev/null +++ b/spinifex/vm/record.go @@ -0,0 +1,204 @@ +package vm + +import ( + "time" + + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/gpu" + "github.com/mulgadc/spinifex/spinifex/resource" + "github.com/mulgadc/spinifex/spinifex/types" +) + +// InstanceRecord is the persisted form of an instance under the per-resource +// key space. Nothing reads or writes it yet; it exists so the field-by-field +// boundary can be settled and tested before the keys move onto it. +type InstanceRecord = resource.Object[InstanceSpec, InstanceStatus] + +// InstanceSpec is what an operator asked for. It is written by the API layer +// and read by the node; the node never writes it. +type InstanceSpec struct { + InstanceType string `json:"instance_type,omitempty"` + + // Config is the machine the node builds QEMU from. It is spec because a + // launch is driven from it and a relaunch on another node replays it, but + // the launch path appends to Config.NetDevs and Config.Devices, so it is + // the one spec field a node currently writes. + Config Config `json:"config"` + + RunInstancesInput *ec2.RunInstancesInput `json:"run_instances_input,omitempty"` + + DesiredState DesiredState `json:"desired_state,omitempty"` + HostfwdPorts []int `json:"hostfwd_ports,omitempty"` + + ManagedBy string `json:"managed_by,omitempty"` + BootMode string `json:"boot_mode,omitempty"` + DirectBoot bool `json:"direct_boot,omitempty"` + IamInstanceProfileArn string `json:"iam_instance_profile_arn,omitempty"` + + PlacementGroupName string `json:"placement_group_name,omitempty"` + CapacityReservationId string `json:"capacity_reservation_id,omitempty"` + InstanceLifecycle string `json:"instance_lifecycle,omitempty"` + SpotInstanceRequestId string `json:"spot_instance_request_id,omitempty"` +} + +// InstanceStatus is what the node running the instance observed. It is written +// only by that node; the API layer never writes it. +type InstanceStatus struct { + Status InstanceState `json:"status"` + + // Instance and Reservation are AWS-shaped projections returned verbatim by + // DescribeInstances, each holding request fields and observed fields in one + // object. They are the one place the boundary does not hold: building them + // from spec and status on read is 143 call sites and its own change. + Instance *ec2.Instance `json:"instance,omitempty"` + Reservation *ec2.Reservation `json:"reservation,omitempty"` + + Health InstanceHealthState `json:"health"` + + // LastNode is the node that last ran this instance, and the field a claim + // CASes to take ownership. Ownership is carried here rather than in the key. + LastNode string `json:"last_node,omitempty"` + + MetadataServerAddress string `json:"metadata_server_address,omitempty"` + + ENIId string `json:"eni_id,omitempty"` + ENIMac string `json:"eni_mac,omitempty"` + ExtraENIs []ExtraENI `json:"extra_enis,omitempty"` + + PublicIP string `json:"public_ip,omitempty"` + PublicIPPool string `json:"public_ip_pool,omitempty"` + PublicIPAllocID string `json:"public_ip_alloc_id,omitempty"` + PublicIPAssocID string `json:"public_ip_assoc_id,omitempty"` + + DevMAC string `json:"dev_mac,omitempty"` + MgmtMAC string `json:"mgmt_mac,omitempty"` + MgmtIP string `json:"mgmt_ip,omitempty"` + + PlacementGroupNode string `json:"placement_group_node,omitempty"` + HostfwdPortMap map[int]int `json:"hostfwd_port_map,omitempty"` + + GPUAttachments []gpu.GPUAttachment `json:"gpu_attachments,omitempty"` + IamInstanceProfileAssociationId string `json:"iam_instance_profile_association_id,omitempty"` + + Teardown map[string]string `json:"teardown,omitempty"` + TerminatedAt time.Time `json:"terminated_at,omitzero"` + ShuttingDownAt time.Time `json:"shutting_down_at,omitzero"` + + // The request slices are carried without their mutexes: a lock is process + // state, not something a record describes. The wrappers are rebuilt with + // fresh locks on the way back. + EBSRequests []types.EBSRequest `json:"ebs_requests,omitempty"` + ENIAvailableSlots []int `json:"eni_available_slots,omitempty"` + ENIAttachedByID map[string]int `json:"eni_attached_by_id,omitempty"` +} + +// Record projects a VM onto the persisted record. QMPClient and the request +// mutexes do not cross: they are process state. Neither do the two legacy +// decode-only fields, which exist to fold records at the keys this record +// replaces and have no meaning at the new ones. +func (v *VM) Record() *InstanceRecord { + return &InstanceRecord{ + Metadata: resource.Metadata{ + Name: v.ID, + AccountID: v.AccountID, + DeletionTimestamp: v.DeletionTimestamp, + }, + Spec: InstanceSpec{ + InstanceType: v.InstanceType, + Config: v.Config, + RunInstancesInput: v.RunInstancesInput, + DesiredState: v.DesiredState, + HostfwdPorts: v.HostfwdPorts, + ManagedBy: v.ManagedBy, + BootMode: v.BootMode, + DirectBoot: v.DirectBoot, + IamInstanceProfileArn: v.IamInstanceProfileArn, + PlacementGroupName: v.PlacementGroupName, + CapacityReservationId: v.CapacityReservationId, + InstanceLifecycle: v.InstanceLifecycle, + SpotInstanceRequestId: v.SpotInstanceRequestId, + }, + Status: InstanceStatus{ + Status: v.Status, + Instance: v.Instance, + Reservation: v.Reservation, + Health: v.Health, + LastNode: v.LastNode, + MetadataServerAddress: v.MetadataServerAddress, + ENIId: v.ENIId, + ENIMac: v.ENIMac, + ExtraENIs: v.ExtraENIs, + PublicIP: v.PublicIP, + PublicIPPool: v.PublicIPPool, + PublicIPAllocID: v.PublicIPAllocID, + PublicIPAssocID: v.PublicIPAssocID, + DevMAC: v.DevMAC, + MgmtMAC: v.MgmtMAC, + MgmtIP: v.MgmtIP, + PlacementGroupNode: v.PlacementGroupNode, + HostfwdPortMap: v.HostfwdPortMap, + GPUAttachments: v.GPUAttachments, + IamInstanceProfileAssociationId: v.IamInstanceProfileAssociationId, + Teardown: v.Teardown, + TerminatedAt: v.TerminatedAt, + ShuttingDownAt: v.ShuttingDownAt, + EBSRequests: v.EBSRequests.Requests, + ENIAvailableSlots: v.ENIRequests.AvailableSlots, + ENIAttachedByID: v.ENIRequests.AttachedByENIID, + }, + } +} + +// VMFromRecord rebuilds the in-memory instance from a record. QMPClient is left +// nil for the caller to attach, matching what ResetNodeLocalState leaves behind +// after a cross-node move. A function rather than a method: InstanceRecord is an +// alias for an instantiated generic, which cannot carry one. +func VMFromRecord(r *InstanceRecord) *VM { + v := &VM{ + ID: r.Metadata.Name, + AccountID: r.Metadata.AccountID, + DeletionTimestamp: r.Metadata.DeletionTimestamp, + + InstanceType: r.Spec.InstanceType, + Config: r.Spec.Config, + RunInstancesInput: r.Spec.RunInstancesInput, + DesiredState: r.Spec.DesiredState, + HostfwdPorts: r.Spec.HostfwdPorts, + ManagedBy: r.Spec.ManagedBy, + BootMode: r.Spec.BootMode, + DirectBoot: r.Spec.DirectBoot, + IamInstanceProfileArn: r.Spec.IamInstanceProfileArn, + PlacementGroupName: r.Spec.PlacementGroupName, + CapacityReservationId: r.Spec.CapacityReservationId, + InstanceLifecycle: r.Spec.InstanceLifecycle, + SpotInstanceRequestId: r.Spec.SpotInstanceRequestId, + + Status: r.Status.Status, + Instance: r.Status.Instance, + Reservation: r.Status.Reservation, + Health: r.Status.Health, + LastNode: r.Status.LastNode, + MetadataServerAddress: r.Status.MetadataServerAddress, + ENIId: r.Status.ENIId, + ENIMac: r.Status.ENIMac, + ExtraENIs: r.Status.ExtraENIs, + PublicIP: r.Status.PublicIP, + PublicIPPool: r.Status.PublicIPPool, + PublicIPAllocID: r.Status.PublicIPAllocID, + PublicIPAssocID: r.Status.PublicIPAssocID, + DevMAC: r.Status.DevMAC, + MgmtMAC: r.Status.MgmtMAC, + MgmtIP: r.Status.MgmtIP, + PlacementGroupNode: r.Status.PlacementGroupNode, + HostfwdPortMap: r.Status.HostfwdPortMap, + GPUAttachments: r.Status.GPUAttachments, + IamInstanceProfileAssociationId: r.Status.IamInstanceProfileAssociationId, + Teardown: r.Status.Teardown, + TerminatedAt: r.Status.TerminatedAt, + ShuttingDownAt: r.Status.ShuttingDownAt, + } + v.EBSRequests.Requests = r.Status.EBSRequests + v.ENIRequests.AvailableSlots = r.Status.ENIAvailableSlots + v.ENIRequests.AttachedByENIID = r.Status.ENIAttachedByID + return v +} diff --git a/spinifex/vm/record_test.go b/spinifex/vm/record_test.go new file mode 100644 index 000000000..1544256a0 --- /dev/null +++ b/spinifex/vm/record_test.go @@ -0,0 +1,274 @@ +package vm_test + +import ( + "encoding/json" + "reflect" + "slices" + "testing" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/gpu" + "github.com/mulgadc/spinifex/spinifex/qmp" + "github.com/mulgadc/spinifex/spinifex/types" + "github.com/mulgadc/spinifex/spinifex/vm" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// metadataFields are carried by resource.Metadata rather than by either half. +var metadataFields = map[string]string{ + "ID": "Metadata.Name", + "AccountID": "Metadata.AccountID", + "DeletionTimestamp": "Metadata.DeletionTimestamp", +} + +// renamedFields are carried under a different name because the record drops the +// mutex each wrapper holds: a lock is process state, not a described fact. +var renamedFields = map[string]string{ + "EBSRequests": "Status.EBSRequests, minus its mutex", + "ENIRequests": "Status.ENIAvailableSlots + Status.ENIAttachedByID, minus its mutex", +} + +// notCarried are the fields that deliberately do not reach the record, each +// with the reason it does not. +var notCarried = map[string]string{ + "QMPClient": "a live connection, already json:\"-\"", + "LegacyAttributes": "decode-only, folds records at the keys this one replaces", + "LegacyExtraHostfwd": "decode-only, same", +} + +// Every exported field of VM must be assigned somewhere. A field added to VM +// and to neither half would otherwise be dropped silently on the way to the +// record, which is the failure the whole split is meant to make impossible. +func TestRecord_CoversEveryVMField(t *testing.T) { + specFields := exportedFields(reflect.TypeFor[vm.InstanceSpec]()) + statusFields := exportedFields(reflect.TypeFor[vm.InstanceStatus]()) + + for _, name := range exportedFields(reflect.TypeFor[vm.VM]()) { + t.Run(name, func(t *testing.T) { + switch { + case slices.Contains(specFields, name): + case slices.Contains(statusFields, name): + case metadataFields[name] != "": + case renamedFields[name] != "": + case notCarried[name] != "": + default: + t.Fatalf("VM.%s reaches neither InstanceSpec nor InstanceStatus.\n"+ + "Add it to the half that owns it — spec if the API layer writes it, status if the "+ + "node does — or to notCarried in this file with the reason it does not persist.", name) + } + }) + } +} + +// The converse: a field on either half that VM does not have is one nothing can +// populate, so the record would carry a value no reader could trust. +func TestRecord_HalvesHaveNoFieldsVMLacks(t *testing.T) { + vmFields := exportedFields(reflect.TypeFor[vm.VM]()) + renamed := []string{ + "EBSRequests", "ENIAvailableSlots", "ENIAttachedByID", + } + + for _, half := range []reflect.Type{reflect.TypeFor[vm.InstanceSpec](), reflect.TypeFor[vm.InstanceStatus]()} { + for _, name := range exportedFields(half) { + if slices.Contains(renamed, name) { + continue + } + assert.Contains(t, vmFields, name, + "%s.%s has no source on VM, so nothing can populate it", half.Name(), name) + } + } +} + +// No field on either half may hold a lock: the record is serialised, copied and +// compared, and a mutex that survives into it is a lock nobody owns. +func TestRecord_HalvesHoldNoLocks(t *testing.T) { + for _, half := range []reflect.Type{reflect.TypeFor[vm.InstanceSpec](), reflect.TypeFor[vm.InstanceStatus]()} { + for f := range half.Fields() { + assert.False(t, containsLock(f.Type, 0), + "%s.%s contains a sync lock; carry the payload without its wrapper", half.Name(), f.Name) + } + } +} + +func TestRecord_RoundTrip(t *testing.T) { + original := populatedVM() + + rebuilt := vm.VMFromRecord(original.Record()) + + assertVMFieldsEqual(t, original, rebuilt) +} + +// The record is persisted through kvstore, so the round trip has to survive +// JSON, not just the two conversions. +func TestRecord_RoundTripThroughJSON(t *testing.T) { + original := populatedVM() + + encoded, err := json.Marshal(original.Record()) + require.NoError(t, err) + + var decoded vm.InstanceRecord + require.NoError(t, json.Unmarshal(encoded, &decoded)) + + assertVMFieldsEqual(t, original, vm.VMFromRecord(&decoded)) +} + +// The mutexes are dropped on the way out, so the rebuilt instance must arrive +// with usable zero-value locks rather than ones somebody else left held. +func TestVMFromRecord_RequestWrappersAreUsable(t *testing.T) { + rebuilt := vm.VMFromRecord(populatedVM().Record()) + + assert.True(t, rebuilt.EBSRequests.Mu.TryLock(), "EBS request lock must arrive unheld") + assert.True(t, rebuilt.ENIRequests.Mu.TryLock(), "ENI request lock must arrive unheld") + + assert.Nil(t, rebuilt.QMPClient, "the caller attaches a live connection; the record cannot carry one") +} + +// An empty record must not decode into an instance carrying maps and slices +// that were never written. +func TestRecord_EmptyRoundTrips(t *testing.T) { + rebuilt := vm.VMFromRecord((&vm.VM{}).Record()) + + assert.Empty(t, rebuilt.ID) + assert.Nil(t, rebuilt.EBSRequests.Requests) + assert.Nil(t, rebuilt.ENIRequests.AttachedByENIID) + assert.Nil(t, rebuilt.HostfwdPortMap) + assert.Nil(t, rebuilt.DeletionTimestamp) +} + +// The round trip can only catch a dropped field if the fixture sets one, so a +// field added to VM and to the record but not to populatedVM would compare +// zero against zero and pass. Guarding the fixture is what stops that. +func TestPopulatedVM_LeavesNoCarriedFieldZero(t *testing.T) { + v := reflect.ValueOf(populatedVM()).Elem() + typ := reflect.TypeFor[vm.VM]() + + for i := range typ.NumField() { + f := typ.Field(i) + if !f.IsExported() || notCarried[f.Name] != "" { + continue + } + assert.False(t, v.Field(i).IsZero(), + "populatedVM leaves VM.%s at its zero value, so the round-trip tests would pass without carrying it", f.Name) + } +} + +// assertVMFieldsEqual compares every exported field except the ones the record +// deliberately drops, so a dropped field fails as itself rather than as a +// whole-struct diff. +func assertVMFieldsEqual(t *testing.T, want, got *vm.VM) { + t.Helper() + // Elem on the pointer rather than dereferencing: VM carries a mutex, so a + // copy of it is one govet rejects and nothing here needs. + wantV, gotV := reflect.ValueOf(want).Elem(), reflect.ValueOf(got).Elem() + typ := reflect.TypeFor[vm.VM]() + + for i := range typ.NumField() { + f := typ.Field(i) + if !f.IsExported() || notCarried[f.Name] != "" { + continue + } + // The wrappers hold a mutex, so compare their payloads instead. + switch f.Name { + case "EBSRequests": + assert.Equal(t, want.EBSRequests.Requests, got.EBSRequests.Requests) + continue + case "ENIRequests": + assert.Equal(t, want.ENIRequests.AvailableSlots, got.ENIRequests.AvailableSlots) + assert.Equal(t, want.ENIRequests.AttachedByENIID, got.ENIRequests.AttachedByENIID) + continue + } + assert.Equal(t, wantV.Field(i).Interface(), gotV.Field(i).Interface(), + "VM.%s did not survive the round trip", f.Name) + } +} + +// populatedVM sets every carried field to a distinctive value, so a field the +// conversion forgets shows up as a zero rather than matching by luck. +func populatedVM() *vm.VM { + deleted := time.Date(2026, 8, 28, 1, 0, 0, 0, time.UTC) + v := &vm.VM{ + ID: "i-round-trip", + AccountID: "111122223333", + DeletionTimestamp: &deleted, + + InstanceType: "m7i.large", + Config: vm.Config{InstanceType: "m7i.large", MachineType: "q35"}, + RunInstancesInput: &ec2.RunInstancesInput{ImageId: aws.String("ami-1")}, + DesiredState: vm.DesiredStopped, + HostfwdPorts: []int{8080, 8443}, + ManagedBy: "elbv2", + BootMode: "uefi", + DirectBoot: true, + IamInstanceProfileArn: "arn:aws:iam::111122223333:instance-profile/p", + PlacementGroupName: "pg-1", + CapacityReservationId: "cr-1", + InstanceLifecycle: "spot", + SpotInstanceRequestId: "sir-1", + + Status: vm.StateRunning, + Instance: &ec2.Instance{InstanceId: aws.String("i-round-trip")}, + Reservation: &ec2.Reservation{ReservationId: aws.String("r-1")}, + Health: vm.InstanceHealthState{CrashCount: 2, LastCrashReason: "oom"}, + LastNode: "node-1", + MetadataServerAddress: "127.0.0.1:12345", + ENIId: "eni-1", + ENIMac: "aa:bb:cc:dd:ee:01", + ExtraENIs: []vm.ExtraENI{{ENIID: "eni-2"}}, + PublicIP: "203.0.113.10", + PublicIPPool: "pool-1", + PublicIPAllocID: "eipalloc-1", + PublicIPAssocID: "eipassoc-1", + DevMAC: "02:00:00:00:00:01", + MgmtMAC: "02:a0:00:00:00:01", + MgmtIP: "10.255.0.5", + PlacementGroupNode: "node-2", + HostfwdPortMap: map[int]int{8080: 18080}, + GPUAttachments: []gpu.GPUAttachment{{PCIAddress: "0000:81:00.0"}}, + IamInstanceProfileAssociationId: "iip-assoc-1", + Teardown: map[string]string{"volume": "done"}, + TerminatedAt: time.Date(2026, 8, 28, 2, 0, 0, 0, time.UTC), + ShuttingDownAt: time.Date(2026, 8, 28, 3, 0, 0, 0, time.UTC), + + QMPClient: &qmp.QMPClient{}, + } + v.EBSRequests.Requests = []types.EBSRequest{{Name: "vol-1", Boot: true}} + v.ENIRequests.AvailableSlots = []int{1, 2} + v.ENIRequests.AttachedByENIID = map[string]int{"eni-1": 0} + return v +} + +func exportedFields(t reflect.Type) []string { + names := make([]string, 0, t.NumField()) + for f := range t.Fields() { + if f.IsExported() { + names = append(names, f.Name) + } + } + return names +} + +// containsLock reports whether t embeds a sync lock at any depth. Depth is +// bounded because ec2 SDK types are deep and none of them hold one. +func containsLock(t reflect.Type, depth int) bool { + if depth > 4 { + return false + } + for t.Kind() == reflect.Pointer || t.Kind() == reflect.Slice || t.Kind() == reflect.Array { + t = t.Elem() + } + if t.Kind() != reflect.Struct { + return false + } + if t.PkgPath() == "sync" { + return true + } + for field := range t.Fields() { + if containsLock(field.Type, depth+1) { + return true + } + } + return false +}