diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index 370c1ee63..000c30005 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -578,7 +578,7 @@ func (d *Daemon) Start() error { // Create services before loading/launching instances, since LaunchInstance depends on them store := objectstore.NewS3ObjectStoreFromConfig(admin.DialTarget(d.config.Predastore.Host), d.config.Predastore.Region, d.config.Predastore.AccessKey, d.config.Predastore.SecretKey) - d.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(d.config, d.resourceMgr.instanceTypes, d.natsConn, &d.Instances, store) + d.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(d.config, d.resourceMgr.instanceTypes, d.natsConn, &d.Instances, store, d.resourceMgr, d.jsManager) d.keyService = handlers_ec2_key.NewKeyServiceImpl(d.config) d.imageService = handlers_ec2_image.NewImageServiceImpl(d.config, d.natsConn) diff --git a/spinifex/daemon/daemon_handlers_instance.go b/spinifex/daemon/daemon_handlers_instance.go index 2c2ff41c1..1555459e0 100644 --- a/spinifex/daemon/daemon_handlers_instance.go +++ b/spinifex/daemon/daemon_handlers_instance.go @@ -11,7 +11,6 @@ import ( "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/filterutil" handlers_ec2_placementgroup "github.com/mulgadc/spinifex/spinifex/handlers/ec2/placementgroup" "github.com/mulgadc/spinifex/spinifex/qmp" "github.com/mulgadc/spinifex/spinifex/types" @@ -633,265 +632,14 @@ func (d *Daemon) handleStopOrTerminateInstance(msg *nats.Msg, command types.EC2I }(instance, command.Attributes) } -// describeInstancesValidFilters defines the set of filter names accepted by DescribeInstances. -var describeInstancesValidFilters = map[string]bool{ - "instance-state-name": true, - "instance-id": true, - "instance-type": true, - "vpc-id": true, - "subnet-id": true, - "tag-key": true, - "tag-value": true, -} - -// instanceMatchesFilters checks whether a VM + its built ec2.Instance copy satisfy all parsed filters. -func instanceMatchesFilters(inst *vm.VM, ic *ec2.Instance, filters map[string][]string) bool { - for name, values := range filters { - if strings.HasPrefix(name, "tag:") { - // tag:Key filters are handled after all field filters. - continue - } - - var field string - switch name { - case "instance-state-name": - if ic.State != nil && ic.State.Name != nil { - field = *ic.State.Name - } - case "instance-id": - field = inst.ID - case "instance-type": - field = inst.InstanceType - case "vpc-id": - if ic.VpcId != nil { - field = *ic.VpcId - } - case "subnet-id": - if ic.SubnetId != nil { - field = *ic.SubnetId - } - case "tag-key": - if !matchTagKey(ic.Tags, values) { - return false - } - continue - case "tag-value": - if !matchTagValue(ic.Tags, values) { - return false - } - continue - default: - // Filter name passed ParseFilters but has no case — treat as non-match - // to avoid silently ignoring it. - return false - } - - if !filterutil.MatchesAny(values, field) { - return false - } - } - - // Check tag:Key filters via the instance's Tag slice. - tags := filterutil.EC2TagsToMap(ic.Tags) - return filterutil.MatchesTags(filters, tags) -} - -// matchTagKey returns true if any tag key on the resource matches any of the filter values. -func matchTagKey(tags []*ec2.Tag, values []string) bool { - for _, t := range tags { - if t.Key != nil && filterutil.MatchesAny(values, *t.Key) { - return true - } - } - return false -} - -// matchTagValue returns true if any tag value on the resource matches any of the filter values. -func matchTagValue(tags []*ec2.Tag, values []string) bool { - for _, t := range tags { - if t.Value != nil && filterutil.MatchesAny(values, *t.Value) { - return true - } - } - return false -} - -// handleEC2DescribeInstances processes incoming EC2 DescribeInstances requests -// This handler responds with instances running on this node owned by the caller's account +// handleEC2DescribeInstances responds with instances running on this node visible to the caller. func (d *Daemon) handleEC2DescribeInstances(msg *nats.Msg) { - slog.Debug("Received message", "subject", msg.Subject, "data", string(msg.Data)) - - // Extract account ID from NATS header for scoping - accountID := utils.AccountIDFromMsg(msg) - - // Initialize describeInstancesInput before unmarshaling into it - describeInstancesInput := &ec2.DescribeInstancesInput{} - errResp := utils.UnmarshalJsonPayload(describeInstancesInput, msg.Data) - - if errResp != nil { - if err := msg.Respond(errResp); err != nil { - slog.Error("Failed to respond to NATS request", "err", err) - } - slog.Error("Request does not match DescribeInstancesInput") - return - } - - slog.Info("Processing DescribeInstances request from this node", "accountID", accountID) - - d.Instances.Mu.Lock() - defer d.Instances.Mu.Unlock() - - // Validate and filter instances if specific instance IDs were requested - instanceIDFilter := make(map[string]bool) - if len(describeInstancesInput.InstanceIds) > 0 { - for _, id := range describeInstancesInput.InstanceIds { - if id != nil && *id != "" { - if !strings.HasPrefix(*id, "i-") { - respondWithError(msg, awserrors.ErrorInvalidInstanceIDMalformed) - return - } - instanceIDFilter[*id] = true - } - } - } - - // Parse filters (returns error for unknown filter names) - parsedFilters, err := filterutil.ParseFilters(describeInstancesInput.Filters, describeInstancesValidFilters) - if err != nil { - slog.Warn("DescribeInstances: invalid filter", "err", err) - respondWithError(msg, awserrors.ErrorInvalidParameterValue) - return - } - - // Group instances by reservation ID (AWS returns instances grouped by reservation) - reservationMap := make(map[string]*ec2.Reservation) - - // Iterate through all instances on this node - for _, instance := range d.Instances.VMS { - // Skip instances not owned by the caller's account. - // Pre-Phase4 instances (empty AccountID) are only visible to root. - if !isInstanceVisible(accountID, instance.AccountID) { - continue - } - - // Skip if filtering by instance IDs and this instance is not in the filter - if len(instanceIDFilter) > 0 && !instanceIDFilter[instance.ID] { - continue - } - - // Use stored reservation metadata if available - if instance.Reservation != nil && instance.Instance != nil { - resID := "" - if instance.Reservation.ReservationId != nil { - resID = *instance.Reservation.ReservationId - } - - // Create reservation entry if it doesn't exist - if _, exists := reservationMap[resID]; !exists { - reservation := &ec2.Reservation{} - reservation.SetReservationId(resID) - if instance.Reservation.OwnerId != nil { - reservation.SetOwnerId(*instance.Reservation.OwnerId) - } - reservation.Instances = []*ec2.Instance{} - reservationMap[resID] = reservation - } - - // Update the instance state to current state - instanceCopy := *instance.Instance - instanceCopy.State = &ec2.InstanceState{} - - // Populate PublicIpAddress from VM if stored - if instance.PublicIP != "" && instanceCopy.PublicIpAddress == nil { - instanceCopy.PublicIpAddress = aws.String(instance.PublicIP) - } - - // Map internal status to EC2 state codes using the centralized mapping - if info, ok := vm.EC2StateCodes[instance.Status]; ok { - instanceCopy.State.SetCode(info.Code) - instanceCopy.State.SetName(info.Name) - } else { - slog.Warn("Instance has unmapped status, reporting as pending", - "instanceId", instance.ID, "status", string(instance.Status)) - instanceCopy.State.SetCode(0) - instanceCopy.State.SetName("pending") - } - - // Populate Placement if instance belongs to a placement group - if instance.PlacementGroupName != "" { - instanceCopy.Placement = &ec2.Placement{ - GroupName: aws.String(instance.PlacementGroupName), - AvailabilityZone: aws.String(d.config.AZ), - } - } - - // Apply filters against the fully-built instance copy - if len(parsedFilters) > 0 && !instanceMatchesFilters(instance, &instanceCopy, parsedFilters) { - continue - } - - // Add instance to its reservation - reservationMap[resID].Instances = append(reservationMap[resID].Instances, &instanceCopy) - } - } - - // Convert map to slice - reservations := make([]*ec2.Reservation, 0, len(reservationMap)) - for _, reservation := range reservationMap { - reservations = append(reservations, reservation) - } - - // Create the response - output := &ec2.DescribeInstancesOutput{ - Reservations: reservations, - } - - respondWithJSON(msg, output) - slog.Info("handleEC2DescribeInstances completed", "count", len(reservations)) + handleNATSRequest(msg, d.instanceService.DescribeInstances) } -// handleEC2DescribeInstanceTypes processes incoming EC2 DescribeInstanceTypes requests -// This handler responds with instance types that can currently be provisioned on this node -// based on available resources (CPU and memory not already allocated to running instances) +// handleEC2DescribeInstanceTypes responds with instance types provisionable on this node. func (d *Daemon) handleEC2DescribeInstanceTypes(msg *nats.Msg) { - slog.Debug("Received message", "subject", msg.Subject) - - // Initialize input - describeInput := &ec2.DescribeInstanceTypesInput{} - errResp := utils.UnmarshalJsonPayload(describeInput, msg.Data) - if errResp != nil { - if err := msg.Respond(errResp); err != nil { - slog.Error("Failed to respond to NATS request", "err", err) - } - slog.Error("Request does not match DescribeInstanceTypesInput") - return - } - - slog.Info("Processing DescribeInstanceTypes request from this node") - - // Check if "capacity" filter is set to "true" - showCapacity := false - for _, f := range describeInput.Filters { - if f.Name != nil && *f.Name == "capacity" { - for _, v := range f.Values { - if v != nil && *v == "true" { - showCapacity = true - break - } - } - } - } - - // Get instance types based on capacity and the showCapacity flag - filteredTypes := d.resourceMgr.GetAvailableInstanceTypeInfos(showCapacity) - - // Create the response - output := &ec2.DescribeInstanceTypesOutput{ - InstanceTypes: filteredTypes, - } - - respondWithJSON(msg, output) - slog.Info("handleEC2DescribeInstanceTypes completed", "count", len(filteredTypes)) + handleNATSRequest(msg, d.instanceService.DescribeInstanceTypes) } // startStoppedInstanceRequest is the payload for ec2.start topic @@ -1153,109 +901,12 @@ func (d *Daemon) handleEC2TerminateStoppedInstance(msg *nats.Msg) { // handleEC2DescribeStoppedInstances returns stopped instances from shared KV. func (d *Daemon) handleEC2DescribeStoppedInstances(msg *nats.Msg) { - d.describeInstancesFromKV(msg, d.jsManager.ListStoppedInstances, 80, "stopped", "handleEC2DescribeStoppedInstances") + handleNATSRequest(msg, d.instanceService.DescribeStoppedInstances) } // handleEC2DescribeTerminatedInstances returns terminated instances from the terminated KV bucket. func (d *Daemon) handleEC2DescribeTerminatedInstances(msg *nats.Msg) { - d.describeInstancesFromKV(msg, d.jsManager.ListTerminatedInstances, 48, "terminated", "handleEC2DescribeTerminatedInstances") -} - -// describeInstancesFromKV is a shared helper for DescribeStopped/TerminatedInstances handlers. -// It lists instances from a KV bucket, filters by account/instance ID, and responds with reservations. -func (d *Daemon) describeInstancesFromKV(msg *nats.Msg, listFn func() ([]*vm.VM, error), fallbackCode int64, fallbackName, handlerName string) { - if d.jsManager == nil { - respondWithError(msg, awserrors.ErrorServerInternal) - return - } - - accountID := utils.AccountIDFromMsg(msg) - - describeInput := &ec2.DescribeInstancesInput{} - if len(msg.Data) > 0 { - if errResp := utils.UnmarshalJsonPayload(describeInput, msg.Data); errResp != nil { - if err := msg.Respond(errResp); err != nil { - slog.Error("Failed to respond to NATS request", "err", err) - } - return - } - } - - instanceIDFilter := make(map[string]bool) - for _, id := range describeInput.InstanceIds { - if id != nil { - instanceIDFilter[*id] = true - } - } - - parsedFilters, filterErr := filterutil.ParseFilters(describeInput.Filters, describeInstancesValidFilters) - if filterErr != nil { - slog.Warn(handlerName+": invalid filter", "err", filterErr) - respondWithError(msg, awserrors.ErrorInvalidParameterValue) - return - } - - instances, err := listFn() - if err != nil { - slog.Error(handlerName+": failed to list instances", "err", err) - respondWithError(msg, awserrors.ErrorServerInternal) - return - } - - reservationMap := make(map[string]*ec2.Reservation) - - for _, instance := range instances { - if !isInstanceVisible(accountID, instance.AccountID) { - continue - } - if len(instanceIDFilter) > 0 && !instanceIDFilter[instance.ID] { - continue - } - if instance.Reservation == nil || instance.Instance == nil { - slog.Warn(handlerName+": skipping instance with nil Reservation/Instance (data integrity issue)", - "instanceId", instance.ID) - continue - } - - resID := "" - if instance.Reservation.ReservationId != nil { - resID = *instance.Reservation.ReservationId - } - - if _, exists := reservationMap[resID]; !exists { - reservation := &ec2.Reservation{} - reservation.SetReservationId(resID) - if instance.Reservation.OwnerId != nil { - reservation.SetOwnerId(*instance.Reservation.OwnerId) - } - reservation.Instances = []*ec2.Instance{} - reservationMap[resID] = reservation - } - - instanceCopy := *instance.Instance - instanceCopy.State = &ec2.InstanceState{} - if info, ok := vm.EC2StateCodes[instance.Status]; ok { - instanceCopy.State.SetCode(info.Code) - instanceCopy.State.SetName(info.Name) - } else { - instanceCopy.State.SetCode(fallbackCode) - instanceCopy.State.SetName(fallbackName) - } - - if len(parsedFilters) > 0 && !instanceMatchesFilters(instance, &instanceCopy, parsedFilters) { - continue - } - - reservationMap[resID].Instances = append(reservationMap[resID].Instances, &instanceCopy) - } - - reservations := make([]*ec2.Reservation, 0, len(reservationMap)) - for _, reservation := range reservationMap { - reservations = append(reservations, reservation) - } - - respondWithJSON(msg, &ec2.DescribeInstancesOutput{Reservations: reservations}) - slog.Info(handlerName+" completed", "count", len(reservations)) + handleNATSRequest(msg, d.instanceService.DescribeTerminatedInstances) } // handleEC2ModifyInstanceAttribute modifies attributes of a stopped instance in shared KV. @@ -1365,120 +1016,8 @@ func (d *Daemon) handleEC2ModifyInstanceAttribute(msg *nats.Msg) { } // handleEC2DescribeInstanceAttribute returns a single requested attribute for an instance. -// It checks running instances first (in-memory), then falls back to stopped instances in KV. func (d *Daemon) handleEC2DescribeInstanceAttribute(msg *nats.Msg) { - var input ec2.DescribeInstanceAttributeInput - if err := json.Unmarshal(msg.Data, &input); err != nil { - slog.Error("handleEC2DescribeInstanceAttribute: failed to unmarshal request", "err", err) - respondWithError(msg, awserrors.ErrorServerInternal) - return - } - - if input.InstanceId == nil || *input.InstanceId == "" { - slog.Error("handleEC2DescribeInstanceAttribute: missing instance_id") - respondWithError(msg, awserrors.ErrorMissingParameter) - return - } - if input.Attribute == nil || *input.Attribute == "" { - slog.Error("handleEC2DescribeInstanceAttribute: missing attribute") - respondWithError(msg, awserrors.ErrorMissingParameter) - return - } - - instanceID := *input.InstanceId - attribute := *input.Attribute - accountID := utils.AccountIDFromMsg(msg) - - // Look up instance: running first, then stopped KV. - var instance *vm.VM - - d.Instances.Mu.Lock() - if running, ok := d.Instances.VMS[instanceID]; ok { - instance = running - } - d.Instances.Mu.Unlock() - - if instance == nil { - if d.jsManager == nil { - slog.Error("handleEC2DescribeInstanceAttribute: JetStream not available") - respondWithError(msg, awserrors.ErrorServerInternal) - return - } - stopped, err := d.jsManager.LoadStoppedInstance(instanceID) - if err != nil { - slog.Error("handleEC2DescribeInstanceAttribute: failed to load stopped instance", - "instanceId", instanceID, "err", err) - respondWithError(msg, awserrors.ErrorServerInternal) - return - } - instance = stopped - } - - if instance == nil { - slog.Warn("handleEC2DescribeInstanceAttribute: instance not found", - "instanceId", instanceID) - respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound) - return - } - - if !checkInstanceOwnership(msg, instanceID, instance.AccountID) { - return - } - - output := &ec2.DescribeInstanceAttributeOutput{ - InstanceId: &instanceID, - } - - switch attribute { - case ec2.InstanceAttributeNameInstanceType: - val := instance.InstanceType - output.InstanceType = &ec2.AttributeValue{Value: &val} - - case ec2.InstanceAttributeNameUserData: - val := instance.UserData - output.UserData = &ec2.AttributeValue{Value: &val} - - case ec2.InstanceAttributeNameDisableApiTermination: - val := false - output.DisableApiTermination = &ec2.AttributeBooleanValue{Value: &val} - - case ec2.InstanceAttributeNameDisableApiStop: - val := false - output.DisableApiStop = &ec2.AttributeBooleanValue{Value: &val} - - case ec2.InstanceAttributeNameInstanceInitiatedShutdownBehavior: - val := ec2.ShutdownBehaviorStop - output.InstanceInitiatedShutdownBehavior = &ec2.AttributeValue{Value: &val} - - case ec2.InstanceAttributeNameEbsOptimized: - val := false - output.EbsOptimized = &ec2.AttributeBooleanValue{Value: &val} - - case ec2.InstanceAttributeNameEnaSupport: - val := true - output.EnaSupport = &ec2.AttributeBooleanValue{Value: &val} - - case ec2.InstanceAttributeNameSourceDestCheck: - val := true - output.SourceDestCheck = &ec2.AttributeBooleanValue{Value: &val} - - case ec2.InstanceAttributeNameGroupSet: - if instance.Instance != nil && len(instance.Instance.SecurityGroups) > 0 { - output.Groups = instance.Instance.SecurityGroups - } else { - output.Groups = []*ec2.GroupIdentifier{} - } - - default: - slog.Warn("handleEC2DescribeInstanceAttribute: unsupported attribute", - "instanceId", instanceID, "attribute", attribute) - respondWithError(msg, awserrors.ErrorInvalidParameterValue) - return - } - - slog.Info("handleEC2DescribeInstanceAttribute: completed", - "instanceId", instanceID, "attribute", attribute, "accountID", accountID) - respondWithJSON(msg, output) + handleNATSRequest(msg, d.instanceService.DescribeInstanceAttribute) } // publishNATEvent sends a NAT lifecycle event (vpc.add-nat or vpc.delete-nat) to NATS. diff --git a/spinifex/daemon/daemon_handlers_test.go b/spinifex/daemon/daemon_handlers_test.go index 6d3f64807..66aebd4e8 100644 --- a/spinifex/daemon/daemon_handlers_test.go +++ b/spinifex/daemon/daemon_handlers_test.go @@ -87,6 +87,14 @@ func createFullTestDaemonWithJetStream(t *testing.T, natsURL string) *Daemon { err = daemon.jsManager.InitTerminatedInstanceBucket() require.NoError(t, err) + // Re-bind the instance service so describe-stopped/terminated handlers see + // the KV that was just initialised. + daemon.instanceService = handlers_ec2_instance.NewInstanceServiceImpl( + daemon.config, daemon.resourceMgr.instanceTypes, daemon.natsConn, + &daemon.Instances, objectstore.NewMemoryObjectStore(), + daemon.resourceMgr, daemon.jsManager, + ) + return daemon } @@ -512,6 +520,7 @@ func TestHandleEC2RunInstances_ServiceErrorPropagated(t *testing.T) { daemon.instanceService = handlers_ec2_instance.NewInstanceServiceImpl( daemon.config, emptyTypes, daemon.natsConn, &daemon.Instances, objectstore.NewMemoryObjectStore(), + daemon.resourceMgr, nil, ) sub, err := daemon.natsConn.QueueSubscribe("ec2.RunInstances", "spinifex-workers", daemon.handleEC2RunInstances) @@ -2306,7 +2315,9 @@ func TestHandleEC2DescribeInstanceAttribute_InvalidJSON(t *testing.T) { var errResp map[string]any err = json.Unmarshal(reply.Data, &errResp) require.NoError(t, err) - assert.Equal(t, "ServerInternal", errResp["Code"]) + // handleNATSRequest reports unmarshal failures as ValidationError, matching + // the rest of the Pattern A handlers. + assert.Equal(t, "ValidationError", errResp["Code"]) } // --- Delegate handler round-trip tests (table-driven) --- diff --git a/spinifex/daemon/daemon_test.go b/spinifex/daemon/daemon_test.go index ccef2d1cf..8ae4364ae 100644 --- a/spinifex/daemon/daemon_test.go +++ b/spinifex/daemon/daemon_test.go @@ -82,7 +82,10 @@ func createTestDaemon(t *testing.T, natsURL string) *Daemon { daemon.detachDelay = 0 // Skip sleep in tests // Initialize services (needed for handler tests) - daemon.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(cfg, daemon.resourceMgr.instanceTypes, nc, &daemon.Instances, objectstore.NewMemoryObjectStore()) + // jsManager is nil here; pass a nil literal to keep the StoppedInstanceStore + // interface itself nil (rather than a typed-nil pointer) so the service can + // short-circuit cleanly when no KV is available. + daemon.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(cfg, daemon.resourceMgr.instanceTypes, nc, &daemon.Instances, objectstore.NewMemoryObjectStore(), daemon.resourceMgr, nil) daemon.volumeService = handlers_ec2_volume.NewVolumeServiceImplWithStore(cfg, objectstore.NewMemoryObjectStore(), nc) t.Cleanup(func() { diff --git a/spinifex/daemon/instance_filter_test.go b/spinifex/handlers/ec2/instance/instance_filter_test.go similarity index 97% rename from spinifex/daemon/instance_filter_test.go rename to spinifex/handlers/ec2/instance/instance_filter_test.go index 0c630e05a..34b84425d 100644 --- a/spinifex/daemon/instance_filter_test.go +++ b/spinifex/handlers/ec2/instance/instance_filter_test.go @@ -1,4 +1,4 @@ -package daemon +package handlers_ec2_instance import ( "testing" @@ -68,14 +68,12 @@ func TestInstanceMatchesFilters_MultipleFilters_ANDLogic(t *testing.T) { VpcId: aws.String("vpc-111"), State: &ec2.InstanceState{Name: aws.String("running")}, } - // Both must match filters := map[string][]string{ "instance-state-name": {"running"}, "vpc-id": {"vpc-111"}, } assert.True(t, instanceMatchesFilters(inst, ic, filters)) - // One doesn't match filters["vpc-id"] = []string{"vpc-999"} assert.False(t, instanceMatchesFilters(inst, ic, filters)) } @@ -189,7 +187,6 @@ func TestInstanceMatchesFilters_EmptyResults(t *testing.T) { InstanceType: aws.String("t3.micro"), State: &ec2.InstanceState{Name: aws.String("running")}, } - // Filter that matches nothing filters := map[string][]string{"instance-type": {"c5.4xlarge"}} assert.False(t, instanceMatchesFilters(inst, ic, filters)) } @@ -198,7 +195,7 @@ func TestParseFilters_UnknownFilterName(t *testing.T) { filters := []*ec2.Filter{ {Name: aws.String("bogus-filter"), Values: []*string{aws.String("val")}}, } - _, err := filterutil.ParseFilters(filters, describeInstancesValidFilters) + _, err := filterutil.ParseFilters(filters, DescribeInstancesValidFilters) require.Error(t, err) assert.Contains(t, err.Error(), "InvalidParameterValue") } diff --git a/spinifex/handlers/ec2/instance/service.go b/spinifex/handlers/ec2/instance/service.go index 5af6caaf7..0c45e53e7 100644 --- a/spinifex/handlers/ec2/instance/service.go +++ b/spinifex/handlers/ec2/instance/service.go @@ -1,8 +1,30 @@ package handlers_ec2_instance -import "github.com/aws/aws-sdk-go/service/ec2" +import ( + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/vm" +) // InstanceService defines the interface for EC2 instance operations business logic type InstanceService interface { RunInstances(input *ec2.RunInstancesInput, accountID string) (*ec2.Reservation, error) + DescribeInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) + DescribeInstanceTypes(input *ec2.DescribeInstanceTypesInput, accountID string) (*ec2.DescribeInstanceTypesOutput, error) + DescribeInstanceAttribute(input *ec2.DescribeInstanceAttributeInput, accountID string) (*ec2.DescribeInstanceAttributeOutput, error) + DescribeStoppedInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) + DescribeTerminatedInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) +} + +// ResourceCapacityProvider exposes per-node instance-type availability used by +// DescribeInstanceTypes. Implemented by daemon.ResourceManager. +type ResourceCapacityProvider interface { + GetAvailableInstanceTypeInfos(showCapacity bool) []*ec2.InstanceTypeInfo +} + +// StoppedInstanceStore provides KV-backed read access to stopped/terminated +// instances. Implemented by daemon.JetStreamManager. +type StoppedInstanceStore interface { + LoadStoppedInstance(instanceID string) (*vm.VM, error) + ListStoppedInstances() ([]*vm.VM, error) + ListTerminatedInstances() ([]*vm.VM, error) } diff --git a/spinifex/handlers/ec2/instance/service_impl.go b/spinifex/handlers/ec2/instance/service_impl.go index 4c7c097ce..6d5fb1170 100644 --- a/spinifex/handlers/ec2/instance/service_impl.go +++ b/spinifex/handlers/ec2/instance/service_impl.go @@ -19,6 +19,7 @@ import ( "github.com/kdomanski/iso9660" "github.com/mulgadc/spinifex/spinifex/awserrors" "github.com/mulgadc/spinifex/spinifex/config" + "github.com/mulgadc/spinifex/spinifex/filterutil" "github.com/mulgadc/spinifex/spinifex/objectstore" spxtypes "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/utils" @@ -233,16 +234,28 @@ type InstanceServiceImpl struct { natsConn *nats.Conn instances *vm.Instances objectStore objectstore.ObjectStore + resourceMgr ResourceCapacityProvider + stoppedStore StoppedInstanceStore } // NewInstanceServiceImpl creates a new instance service implementation for daemon use -func NewInstanceServiceImpl(cfg *config.Config, instanceTypes map[string]*ec2.InstanceTypeInfo, nc *nats.Conn, instances *vm.Instances, store objectstore.ObjectStore) *InstanceServiceImpl { +func NewInstanceServiceImpl( + cfg *config.Config, + instanceTypes map[string]*ec2.InstanceTypeInfo, + nc *nats.Conn, + instances *vm.Instances, + store objectstore.ObjectStore, + resourceMgr ResourceCapacityProvider, + stoppedStore StoppedInstanceStore, +) *InstanceServiceImpl { return &InstanceServiceImpl{ config: cfg, instanceTypes: instanceTypes, natsConn: nc, instances: instances, objectStore: store, + resourceMgr: resourceMgr, + stoppedStore: stoppedStore, } } @@ -844,3 +857,401 @@ func generateHostname(instanceID string) string { } return "spinifex-vm-unknown" } + +// DescribeInstancesValidFilters lists the filter names accepted by DescribeInstances +// (and the stopped/terminated variants, which share the same filter shape). +var DescribeInstancesValidFilters = map[string]bool{ + "instance-state-name": true, + "instance-id": true, + "instance-type": true, + "vpc-id": true, + "subnet-id": true, + "tag-key": true, + "tag-value": true, +} + +// IsInstanceVisible reports whether the caller can see this instance. +// Pre-Phase4 instances (empty AccountID) are only visible to root (GlobalAccountID). +func IsInstanceVisible(callerAccountID, ownerAccountID string) bool { + if ownerAccountID == "" { + return callerAccountID == utils.GlobalAccountID + } + return callerAccountID == ownerAccountID +} + +// instanceMatchesFilters checks whether a VM + its built ec2.Instance copy satisfy all parsed filters. +func instanceMatchesFilters(inst *vm.VM, ic *ec2.Instance, filters map[string][]string) bool { + for name, values := range filters { + if strings.HasPrefix(name, "tag:") { + continue + } + + var field string + switch name { + case "instance-state-name": + if ic.State != nil && ic.State.Name != nil { + field = *ic.State.Name + } + case "instance-id": + field = inst.ID + case "instance-type": + field = inst.InstanceType + case "vpc-id": + if ic.VpcId != nil { + field = *ic.VpcId + } + case "subnet-id": + if ic.SubnetId != nil { + field = *ic.SubnetId + } + case "tag-key": + if !matchTagKey(ic.Tags, values) { + return false + } + continue + case "tag-value": + if !matchTagValue(ic.Tags, values) { + return false + } + continue + default: + return false + } + + if !filterutil.MatchesAny(values, field) { + return false + } + } + + tags := filterutil.EC2TagsToMap(ic.Tags) + return filterutil.MatchesTags(filters, tags) +} + +func matchTagKey(tags []*ec2.Tag, values []string) bool { + for _, t := range tags { + if t.Key != nil && filterutil.MatchesAny(values, *t.Key) { + return true + } + } + return false +} + +func matchTagValue(tags []*ec2.Tag, values []string) bool { + for _, t := range tags { + if t.Value != nil && filterutil.MatchesAny(values, *t.Value) { + return true + } + } + return false +} + +// DescribeInstances returns instances on this node visible to the caller's account. +func (s *InstanceServiceImpl) DescribeInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) { + slog.Info("Processing DescribeInstances request from this node", "accountID", accountID) + + s.instances.Mu.Lock() + defer s.instances.Mu.Unlock() + + instanceIDFilter := make(map[string]bool) + for _, id := range input.InstanceIds { + if id != nil && *id != "" { + if !strings.HasPrefix(*id, "i-") { + return nil, errors.New(awserrors.ErrorInvalidInstanceIDMalformed) + } + instanceIDFilter[*id] = true + } + } + + parsedFilters, err := filterutil.ParseFilters(input.Filters, DescribeInstancesValidFilters) + if err != nil { + slog.Warn("DescribeInstances: invalid filter", "err", err) + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + + reservationMap := make(map[string]*ec2.Reservation) + + for _, instance := range s.instances.VMS { + if !IsInstanceVisible(accountID, instance.AccountID) { + continue + } + if len(instanceIDFilter) > 0 && !instanceIDFilter[instance.ID] { + continue + } + if instance.Reservation == nil || instance.Instance == nil { + continue + } + + resID := "" + if instance.Reservation.ReservationId != nil { + resID = *instance.Reservation.ReservationId + } + + if _, exists := reservationMap[resID]; !exists { + reservation := &ec2.Reservation{} + reservation.SetReservationId(resID) + if instance.Reservation.OwnerId != nil { + reservation.SetOwnerId(*instance.Reservation.OwnerId) + } + reservation.Instances = []*ec2.Instance{} + reservationMap[resID] = reservation + } + + instanceCopy := *instance.Instance + instanceCopy.State = &ec2.InstanceState{} + + if instance.PublicIP != "" && instanceCopy.PublicIpAddress == nil { + instanceCopy.PublicIpAddress = aws.String(instance.PublicIP) + } + + if info, ok := vm.EC2StateCodes[instance.Status]; ok { + instanceCopy.State.SetCode(info.Code) + instanceCopy.State.SetName(info.Name) + } else { + slog.Warn("Instance has unmapped status, reporting as pending", + "instanceId", instance.ID, "status", string(instance.Status)) + instanceCopy.State.SetCode(0) + instanceCopy.State.SetName("pending") + } + + if instance.PlacementGroupName != "" { + instanceCopy.Placement = &ec2.Placement{ + GroupName: aws.String(instance.PlacementGroupName), + AvailabilityZone: aws.String(s.config.AZ), + } + } + + if len(parsedFilters) > 0 && !instanceMatchesFilters(instance, &instanceCopy, parsedFilters) { + continue + } + + reservationMap[resID].Instances = append(reservationMap[resID].Instances, &instanceCopy) + } + + reservations := make([]*ec2.Reservation, 0, len(reservationMap)) + for _, reservation := range reservationMap { + reservations = append(reservations, reservation) + } + + slog.Info("DescribeInstances completed", "count", len(reservations)) + return &ec2.DescribeInstancesOutput{Reservations: reservations}, nil +} + +// DescribeInstanceTypes returns instance types provisionable on this node. +// The "capacity" filter (when "true") expands each type to one entry per +// available slot so callers can report cluster-wide capacity by aggregating. +func (s *InstanceServiceImpl) DescribeInstanceTypes(input *ec2.DescribeInstanceTypesInput, _ string) (*ec2.DescribeInstanceTypesOutput, error) { + slog.Info("Processing DescribeInstanceTypes request from this node") + + if s.resourceMgr == nil { + return nil, errors.New(awserrors.ErrorServerInternal) + } + + showCapacity := false + for _, f := range input.Filters { + if f.Name != nil && *f.Name == "capacity" { + for _, v := range f.Values { + if v != nil && *v == "true" { + showCapacity = true + break + } + } + } + } + + filteredTypes := s.resourceMgr.GetAvailableInstanceTypeInfos(showCapacity) + slog.Info("DescribeInstanceTypes completed", "count", len(filteredTypes)) + return &ec2.DescribeInstanceTypesOutput{InstanceTypes: filteredTypes}, nil +} + +// DescribeStoppedInstances returns stopped instances from shared KV. +func (s *InstanceServiceImpl) DescribeStoppedInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) { + if s.stoppedStore == nil { + return nil, errors.New(awserrors.ErrorServerInternal) + } + return s.describeInstancesFromKV(input, accountID, s.stoppedStore.ListStoppedInstances, 80, "stopped", "DescribeStoppedInstances") +} + +// DescribeTerminatedInstances returns terminated instances from the terminated KV bucket. +func (s *InstanceServiceImpl) DescribeTerminatedInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) { + if s.stoppedStore == nil { + return nil, errors.New(awserrors.ErrorServerInternal) + } + return s.describeInstancesFromKV(input, accountID, s.stoppedStore.ListTerminatedInstances, 48, "terminated", "DescribeTerminatedInstances") +} + +func (s *InstanceServiceImpl) describeInstancesFromKV(input *ec2.DescribeInstancesInput, accountID string, listFn func() ([]*vm.VM, error), fallbackCode int64, fallbackName, opName string) (*ec2.DescribeInstancesOutput, error) { + instanceIDFilter := make(map[string]bool) + for _, id := range input.InstanceIds { + if id != nil { + instanceIDFilter[*id] = true + } + } + + parsedFilters, filterErr := filterutil.ParseFilters(input.Filters, DescribeInstancesValidFilters) + if filterErr != nil { + slog.Warn(opName+": invalid filter", "err", filterErr) + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + + instances, err := listFn() + if err != nil { + slog.Error(opName+": failed to list instances", "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + + reservationMap := make(map[string]*ec2.Reservation) + + for _, instance := range instances { + if !IsInstanceVisible(accountID, instance.AccountID) { + continue + } + if len(instanceIDFilter) > 0 && !instanceIDFilter[instance.ID] { + continue + } + if instance.Reservation == nil || instance.Instance == nil { + slog.Warn(opName+": skipping instance with nil Reservation/Instance (data integrity issue)", + "instanceId", instance.ID) + continue + } + + resID := "" + if instance.Reservation.ReservationId != nil { + resID = *instance.Reservation.ReservationId + } + + if _, exists := reservationMap[resID]; !exists { + reservation := &ec2.Reservation{} + reservation.SetReservationId(resID) + if instance.Reservation.OwnerId != nil { + reservation.SetOwnerId(*instance.Reservation.OwnerId) + } + reservation.Instances = []*ec2.Instance{} + reservationMap[resID] = reservation + } + + instanceCopy := *instance.Instance + instanceCopy.State = &ec2.InstanceState{} + if info, ok := vm.EC2StateCodes[instance.Status]; ok { + instanceCopy.State.SetCode(info.Code) + instanceCopy.State.SetName(info.Name) + } else { + instanceCopy.State.SetCode(fallbackCode) + instanceCopy.State.SetName(fallbackName) + } + + if len(parsedFilters) > 0 && !instanceMatchesFilters(instance, &instanceCopy, parsedFilters) { + continue + } + + reservationMap[resID].Instances = append(reservationMap[resID].Instances, &instanceCopy) + } + + reservations := make([]*ec2.Reservation, 0, len(reservationMap)) + for _, reservation := range reservationMap { + reservations = append(reservations, reservation) + } + + slog.Info(opName+" completed", "count", len(reservations)) + return &ec2.DescribeInstancesOutput{Reservations: reservations}, nil +} + +// DescribeInstanceAttribute returns a single requested attribute for an instance. +// Checks running instances first (in-memory), then falls back to stopped instances in KV. +func (s *InstanceServiceImpl) DescribeInstanceAttribute(input *ec2.DescribeInstanceAttributeInput, accountID string) (*ec2.DescribeInstanceAttributeOutput, error) { + if input.InstanceId == nil || *input.InstanceId == "" { + return nil, errors.New(awserrors.ErrorMissingParameter) + } + if input.Attribute == nil || *input.Attribute == "" { + return nil, errors.New(awserrors.ErrorMissingParameter) + } + + instanceID := *input.InstanceId + attribute := *input.Attribute + + var instance *vm.VM + s.instances.Mu.Lock() + if running, ok := s.instances.VMS[instanceID]; ok { + instance = running + } + s.instances.Mu.Unlock() + + if instance == nil { + if s.stoppedStore == nil { + slog.Error("DescribeInstanceAttribute: stopped store not available") + return nil, errors.New(awserrors.ErrorServerInternal) + } + stopped, err := s.stoppedStore.LoadStoppedInstance(instanceID) + if err != nil { + slog.Error("DescribeInstanceAttribute: failed to load stopped instance", + "instanceId", instanceID, "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + instance = stopped + } + + if instance == nil { + slog.Warn("DescribeInstanceAttribute: instance not found", + "instanceId", instanceID) + return nil, errors.New(awserrors.ErrorInvalidInstanceIDNotFound) + } + + if !IsInstanceVisible(accountID, instance.AccountID) { + slog.Warn("DescribeInstanceAttribute: instance not visible", + "instanceId", instanceID, "callerAccount", accountID, "ownerAccount", instance.AccountID) + return nil, errors.New(awserrors.ErrorInvalidInstanceIDNotFound) + } + + output := &ec2.DescribeInstanceAttributeOutput{ + InstanceId: &instanceID, + } + + switch attribute { + case ec2.InstanceAttributeNameInstanceType: + val := instance.InstanceType + output.InstanceType = &ec2.AttributeValue{Value: &val} + + case ec2.InstanceAttributeNameUserData: + val := instance.UserData + output.UserData = &ec2.AttributeValue{Value: &val} + + case ec2.InstanceAttributeNameDisableApiTermination: + val := false + output.DisableApiTermination = &ec2.AttributeBooleanValue{Value: &val} + + case ec2.InstanceAttributeNameDisableApiStop: + val := false + output.DisableApiStop = &ec2.AttributeBooleanValue{Value: &val} + + case ec2.InstanceAttributeNameInstanceInitiatedShutdownBehavior: + val := ec2.ShutdownBehaviorStop + output.InstanceInitiatedShutdownBehavior = &ec2.AttributeValue{Value: &val} + + case ec2.InstanceAttributeNameEbsOptimized: + val := false + output.EbsOptimized = &ec2.AttributeBooleanValue{Value: &val} + + case ec2.InstanceAttributeNameEnaSupport: + val := true + output.EnaSupport = &ec2.AttributeBooleanValue{Value: &val} + + case ec2.InstanceAttributeNameSourceDestCheck: + val := true + output.SourceDestCheck = &ec2.AttributeBooleanValue{Value: &val} + + case ec2.InstanceAttributeNameGroupSet: + if instance.Instance != nil && len(instance.Instance.SecurityGroups) > 0 { + output.Groups = instance.Instance.SecurityGroups + } else { + output.Groups = []*ec2.GroupIdentifier{} + } + + default: + slog.Warn("DescribeInstanceAttribute: unsupported attribute", + "instanceId", instanceID, "attribute", attribute) + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + + slog.Info("DescribeInstanceAttribute: completed", + "instanceId", instanceID, "attribute", attribute, "accountID", accountID) + return output, nil +} diff --git a/spinifex/handlers/ec2/instance/service_impl_test.go b/spinifex/handlers/ec2/instance/service_impl_test.go index b92d86789..798397d29 100644 --- a/spinifex/handlers/ec2/instance/service_impl_test.go +++ b/spinifex/handlers/ec2/instance/service_impl_test.go @@ -26,7 +26,7 @@ func TestNewInstanceServiceImpl(t *testing.T) { store := objectstore.NewMemoryObjectStore() instances := &vm.Instances{VMS: make(map[string]*vm.VM)} - svc := NewInstanceServiceImpl(cfg, instanceTypes, nil, instances, store) + svc := NewInstanceServiceImpl(cfg, instanceTypes, nil, instances, store, nil, nil) require.NotNil(t, svc) assert.Equal(t, cfg, svc.config) diff --git a/spinifex/handlers/ec2/instance/service_mock.go b/spinifex/handlers/ec2/instance/service_mock.go index 93fd5a0b5..071e13b3d 100644 --- a/spinifex/handlers/ec2/instance/service_mock.go +++ b/spinifex/handlers/ec2/instance/service_mock.go @@ -33,3 +33,23 @@ func (s *MockInstanceService) RunInstances(input *ec2.RunInstancesInput, account return reservation, nil } + +func (s *MockInstanceService) DescribeInstances(_ *ec2.DescribeInstancesInput, _ string) (*ec2.DescribeInstancesOutput, error) { + return &ec2.DescribeInstancesOutput{}, nil +} + +func (s *MockInstanceService) DescribeInstanceTypes(_ *ec2.DescribeInstanceTypesInput, _ string) (*ec2.DescribeInstanceTypesOutput, error) { + return &ec2.DescribeInstanceTypesOutput{}, nil +} + +func (s *MockInstanceService) DescribeInstanceAttribute(input *ec2.DescribeInstanceAttributeInput, _ string) (*ec2.DescribeInstanceAttributeOutput, error) { + return &ec2.DescribeInstanceAttributeOutput{InstanceId: input.InstanceId}, nil +} + +func (s *MockInstanceService) DescribeStoppedInstances(_ *ec2.DescribeInstancesInput, _ string) (*ec2.DescribeInstancesOutput, error) { + return &ec2.DescribeInstancesOutput{}, nil +} + +func (s *MockInstanceService) DescribeTerminatedInstances(_ *ec2.DescribeInstancesInput, _ string) (*ec2.DescribeInstancesOutput, error) { + return &ec2.DescribeInstancesOutput{}, nil +} diff --git a/spinifex/handlers/ec2/instance/service_nats.go b/spinifex/handlers/ec2/instance/service_nats.go index 6fc51b72c..6d611cfbd 100644 --- a/spinifex/handlers/ec2/instance/service_nats.go +++ b/spinifex/handlers/ec2/instance/service_nats.go @@ -10,7 +10,14 @@ import ( "github.com/nats-io/nats.go" ) -// NATSInstanceService handles instance operations via NATS messaging +// NATSInstanceService handles instance operations via NATS messaging. +// +// Note: DescribeInstances and DescribeInstanceTypes subscriptions are +// fan-out (no queue group); a single NATSRequest only captures one +// responder. Cluster-wide aggregation is the gateway's job (see +// spinifex/gateway/ec2/instance). The methods below are kept for symmetry +// with the InstanceService interface and intentionally fail rather than +// silently returning a one-node view. type NATSInstanceService struct { natsConn *nats.Conn } @@ -27,3 +34,27 @@ func (s *NATSInstanceService) RunInstances(input *ec2.RunInstancesInput, account topic := fmt.Sprintf("ec2.RunInstances.%s", aws.StringValue(input.InstanceType)) return utils.NATSRequest[ec2.Reservation](s.natsConn, topic, input, 5*time.Minute, accountID) } + +// DescribeInstances is a fan-out operation; use the gateway's scatter-gather +// helper instead of this single-node client. +func (s *NATSInstanceService) DescribeInstances(_ *ec2.DescribeInstancesInput, _ string) (*ec2.DescribeInstancesOutput, error) { + return nil, fmt.Errorf("DescribeInstances is fan-out; use gateway scatter-gather, not NATSInstanceService") +} + +// DescribeInstanceTypes is a fan-out operation; use the gateway's scatter-gather +// helper instead of this single-node client. +func (s *NATSInstanceService) DescribeInstanceTypes(_ *ec2.DescribeInstanceTypesInput, _ string) (*ec2.DescribeInstanceTypesOutput, error) { + return nil, fmt.Errorf("DescribeInstanceTypes is fan-out; use gateway scatter-gather, not NATSInstanceService") +} + +func (s *NATSInstanceService) DescribeInstanceAttribute(input *ec2.DescribeInstanceAttributeInput, accountID string) (*ec2.DescribeInstanceAttributeOutput, error) { + return utils.NATSRequest[ec2.DescribeInstanceAttributeOutput](s.natsConn, "ec2.DescribeInstanceAttribute", input, 30*time.Second, accountID) +} + +func (s *NATSInstanceService) DescribeStoppedInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) { + return utils.NATSRequest[ec2.DescribeInstancesOutput](s.natsConn, "ec2.DescribeStoppedInstances", input, 30*time.Second, accountID) +} + +func (s *NATSInstanceService) DescribeTerminatedInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) { + return utils.NATSRequest[ec2.DescribeInstancesOutput](s.natsConn, "ec2.DescribeTerminatedInstances", input, 30*time.Second, accountID) +}