diff --git a/docs/COMMANDS.md b/docs/COMMANDS.md index 90a5eed6f..555558867 100644 --- a/docs/COMMANDS.md +++ b/docs/COMMANDS.md @@ -193,8 +193,8 @@ The surface is unreachable until an operator runs `spx admin principal create`; | `modify-instance-metadata-options` | `--instance-id`, `--http-put-response-hop-limit` (1–64), `--http-tokens` (`required`/`optional`), `--http-endpoint` (`enabled`), `--http-protocol-ipv6`/`--instance-metadata-tags` (`disabled`) — unmodelled values return `UnsupportedOperation` | `--dry-run` | **DONE** | | `describe-instance-credit-specifications` | `--instance-ids` | `--filters`, `--max-results`, `--dry-run` | **DONE** (stub — always returns `standard`) | | `describe-instance-status` | `--instance-ids`, `--include-all-instances`, `--filters` (availability-zone, instance-state-code, instance-state-name, tag:*) | `--max-results`, `--next-token`, `--dry-run`, event/instance-status/system-status filters | **DONE** (static health) | -| `monitor-instances` | — | `--instance-ids` | **NOT STARTED** | -| `unmonitor-instances` | — | `--instance-ids` | **NOT STARTED** | +| `monitor-instances` | `--instance-ids` | `--dry-run` | **DONE** (60s detailed telemetry; state is `enabled`/`disabled` only, never `pending`/`disabling`) | +| `unmonitor-instances` | `--instance-ids` | `--dry-run` | **DONE** (returns the instance to the 300s basic tier) | ### EC2 — Spot Instances diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index a988c8689..dd8515d37 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -970,6 +970,8 @@ func (d *Daemon) subscribeAll() error { {"ec2.RevokeSecurityGroupEgress", handleNATSRequest(d.vpcService.RevokeSecurityGroupEgress), "spinifex-workers"}, {"ec2.ModifyInstanceAttribute", handleNATSRequest(d.instanceService.ModifyInstanceAttribute), "spinifex-workers"}, {"ec2.ModifyInstanceMetadataOptions", handleNATSRequest(d.instanceService.ModifyInstanceMetadataOptions), "spinifex-workers"}, + {"ec2.MonitorInstances", handleNATSRequest(d.monitorInstances), "spinifex-workers"}, + {"ec2.UnmonitorInstances", handleNATSRequest(d.unmonitorInstances), "spinifex-workers"}, {"ec2.start", d.handleEC2StartStoppedInstance, "spinifex-workers"}, // ec2.start.{node} is the node-targeted variant: it always starts locally // and never re-forwards, so it goes straight to the service (no routing loop). diff --git a/spinifex/daemon/daemon_handlers.go b/spinifex/daemon/daemon_handlers.go index 2b205b931..148edfccb 100644 --- a/spinifex/daemon/daemon_handlers.go +++ b/spinifex/daemon/daemon_handlers.go @@ -237,6 +237,8 @@ func (d *Daemon) dispatchEC2Command(msg *nats.Msg) (string, string) { return name, d.handleSetSpotLineage(ctx, msg, command) case command.Attributes.SetInstanceTags, command.Attributes.RemoveInstanceTags: return name, d.handleSetInstanceTags(ctx, msg, command, instance) + case command.Attributes.SetInstanceMonitoring: + return name, d.handleSetInstanceMonitoring(ctx, msg, command, instance) case command.Attributes.StartInstance: opCtx, opSpan := startOpSpan(ctx, "ec2.StartInstance", instance.ID) err := d.instanceService.StartInstance(opCtx, instance, command) @@ -305,6 +307,8 @@ func ec2CommandName(command types.EC2InstanceCommand) string { return "SetInstanceTags" case command.Attributes.RemoveInstanceTags: return "RemoveInstanceTags" + case command.Attributes.SetInstanceMonitoring: + return "SetInstanceMonitoring" case command.Attributes.StartInstance: return "StartInstance" case command.Attributes.RebootInstance: diff --git a/spinifex/daemon/daemon_handlers_instance.go b/spinifex/daemon/daemon_handlers_instance.go index ce999006d..a4bd97bed 100644 --- a/spinifex/daemon/daemon_handlers_instance.go +++ b/spinifex/daemon/daemon_handlers_instance.go @@ -78,6 +78,43 @@ func (d *Daemon) handleSetInstanceTags(ctx context.Context, msg *nats.Msg, comma return outcomeSuccess } +// handleSetInstanceMonitoring flips a running instance's EC2 monitoring tier +// and rewrites its collector discovery file, so the poller resets its interval +// without a restart. Ownership is checked by checkInstanceOwnership before +// dispatch. +func (d *Daemon) handleSetInstanceMonitoring(ctx context.Context, msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) string { + data := command.InstanceMonitoringData + if data == nil { + return respondErrorOutcome(msg, awserrors.ErrorMissingParameter) + } + + found, err := d.vmMgr.UpdateAndPersist(instance.ID, func(v *vm.VM) bool { + if v.RunInstancesInput == nil { + return false + } + if v.RunInstancesInput.Monitoring == nil { + v.RunInstancesInput.Monitoring = &ec2.RunInstancesMonitoringEnabled{} + } + v.RunInstancesInput.Monitoring.Enabled = aws.Bool(data.Enabled) + return true + }) + if err != nil { + return respondServiceErrorOutcome(msg, err) + } + if !found { + return respondErrorOutcome(msg, awserrors.ErrorInvalidInstanceIDNotFound) + } + + // Off the manager lock: the refresh takes the VM's own ENI lock. + instance.RefreshTelemetryMeta() + slog.InfoContext(ctx, "SetInstanceMonitoring: tier applied", "instanceId", instance.ID, "enabled", data.Enabled) + + if err := msg.Respond([]byte(`{}`)); err != nil { + slog.ErrorContext(ctx, "Failed to respond to NATS request", "err", err) + } + return outcomeSuccess +} + // handleEC2RunInstances orchestrates the RunInstances flow across // InstanceService.PrepareRunInstances (validation + ENI creation), // daemon-local Insert/WriteState/per-instance subscribe, and diff --git a/spinifex/daemon/daemon_handlers_monitoring.go b/spinifex/daemon/daemon_handlers_monitoring.go new file mode 100644 index 000000000..a1b3f9efa --- /dev/null +++ b/spinifex/daemon/daemon_handlers_monitoring.go @@ -0,0 +1,120 @@ +package daemon + +import ( + "context" + "encoding/json" + "errors" + "log/slog" + "sync" + + "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" +) + +// monitorInstances moves instances to the 60s detailed telemetry tier. +func (d *Daemon) monitorInstances(ctx context.Context, input *ec2.MonitorInstancesInput, accountID string) (*ec2.MonitorInstancesOutput, error) { + if input == nil { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + monitorings, err := d.setInstanceMonitoring(ctx, input.InstanceIds, true, accountID) + if err != nil { + return nil, err + } + return &ec2.MonitorInstancesOutput{InstanceMonitorings: monitorings}, nil +} + +// unmonitorInstances returns instances to the 300s basic telemetry tier. +func (d *Daemon) unmonitorInstances(ctx context.Context, input *ec2.UnmonitorInstancesInput, accountID string) (*ec2.UnmonitorInstancesOutput, error) { + if input == nil { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + monitorings, err := d.setInstanceMonitoring(ctx, input.InstanceIds, false, accountID) + if err != nil { + return nil, err + } + return &ec2.UnmonitorInstancesOutput{InstanceMonitorings: monitorings}, nil +} + +// setInstanceMonitoring applies the tier to each instance through its owner +// concurrently, so a many-instance call costs one owner timeout rather than +// one per instance. The first failure in input order is returned. +func (d *Daemon) setInstanceMonitoring(ctx context.Context, instanceIDs []*string, enabled bool, accountID string) ([]*ec2.InstanceMonitoring, error) { + if len(instanceIDs) == 0 { + return nil, errors.New(awserrors.ErrorMissingParameter) + } + + errs := make([]error, len(instanceIDs)) + var wg sync.WaitGroup + for i, id := range instanceIDs { + if id == nil { + errs[i] = errors.New(awserrors.ErrorInvalidInstanceIDMalformed) + continue + } + wg.Go(func() { + errs[i] = d.setOneInstanceMonitoring(ctx, *id, enabled, accountID) + }) + } + wg.Wait() + for _, err := range errs { + if err != nil { + return nil, err + } + } + + // Every instance applied, so they all report the requested tier. Built + // after the error sweep so a partial failure returns no state at all. + state := ec2.MonitoringStateDisabled + if enabled { + state = ec2.MonitoringStateEnabled + } + monitorings := make([]*ec2.InstanceMonitoring, 0, len(instanceIDs)) + for _, id := range instanceIDs { + monitorings = append(monitorings, &ec2.InstanceMonitoring{ + InstanceId: aws.String(*id), + Monitoring: &ec2.Monitoring{State: aws.String(state)}, + }) + } + return monitorings, nil +} + +// setOneInstanceMonitoring asks the owning daemon over ec2.cmd. to apply +// the tier; only the owner subscribes there. No responders means the instance +// isn't running, so it falls back to the shared stopped store; a timeout +// (partitioned-but-subscribed owner) surfaces InvalidID.NotFound. +func (d *Daemon) setOneInstanceMonitoring(ctx context.Context, instanceID string, enabled bool, accountID string) error { + command := types.EC2InstanceCommand{ + ID: instanceID, + Attributes: types.EC2CommandAttributes{SetInstanceMonitoring: true}, + InstanceMonitoringData: &types.InstanceMonitoringData{Enabled: enabled}, + } + body, err := json.Marshal(command) + if err != nil { + slog.ErrorContext(ctx, "setInstanceMonitoring: failed to marshal command", "instanceId", instanceID, "err", err) + return errors.New(awserrors.ErrorServerInternal) + } + + reqMsg := nats.NewMsg("ec2.cmd." + instanceID) + reqMsg.Data = body + reqMsg.Header.Set(utils.AccountIDHeader, accountID) + utils.InjectTraceContext(ctx, reqMsg.Header) + + msg, err := d.natsConn.RequestMsg(reqMsg, instanceOwnerCommandTimeout) + switch { + case errors.Is(err, nats.ErrNoResponders): + return d.instanceService.SetStoppedInstanceMonitoring(ctx, instanceID, enabled, accountID) + case errors.Is(err, nats.ErrTimeout): + return errors.New(awserrors.ErrorInvalidInstanceIDNotFound) + case err != nil: + slog.ErrorContext(ctx, "setInstanceMonitoring: owner request failed", "instanceId", instanceID, "err", err) + return errors.New(awserrors.ErrorServerInternal) + } + + if responseError, parseErr := utils.ValidateErrorPayload(msg.Data); parseErr != nil { + return errors.New(*responseError.Code) + } + return nil +} diff --git a/spinifex/daemon/daemon_handlers_monitoring_test.go b/spinifex/daemon/daemon_handlers_monitoring_test.go new file mode 100644 index 000000000..5705870a6 --- /dev/null +++ b/spinifex/daemon/daemon_handlers_monitoring_test.go @@ -0,0 +1,187 @@ +//test:in-package — monitorInstances/unmonitorInstances are unexported, and the +//fixture reaches the daemon's own vmMgr and resourceMgr to seed a running +//instance and read the tier back off the record. + +package daemon + +import ( + "context" + "testing" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/awserrors" + handlers_ec2_instance "github.com/mulgadc/spinifex/spinifex/handlers/ec2/instance" + "github.com/mulgadc/spinifex/spinifex/objectstore" + "github.com/mulgadc/spinifex/spinifex/vm" + vmmock "github.com/mulgadc/spinifex/spinifex/vm/mock" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// monitoringTestDaemon returns a test daemon owning one running instance +// launched at the given tier, plus its stopped store. +func monitoringTestDaemon(t *testing.T, instanceID string, enabled bool) (*Daemon, *vmmock.StateStore) { + t.Helper() + d := createTestDaemon(t, sharedNATSURL) + + stopped := vmmock.New() + d.instanceService = handlers_ec2_instance.NewInstanceServiceImpl( + d.config, d.resourceMgr.instanceTypes, d.natsConn, + objectstore.NewMemoryObjectStore(), d.vmMgr, d.resourceMgr, stopped) + + d.vmMgr.Insert(&vm.VM{ + ID: instanceID, + Status: vm.StateRunning, + AccountID: testAccountID, + Instance: &ec2.Instance{InstanceId: aws.String(instanceID)}, + RunInstancesInput: &ec2.RunInstancesInput{ + Monitoring: &ec2.RunInstancesMonitoringEnabled{Enabled: aws.Bool(enabled)}, + }, + }) + return d, stopped +} + +func recordMonitoring(t *testing.T, d *Daemon, instanceID string) bool { + t.Helper() + got, ok := d.vmMgr.Get(instanceID) + require.True(t, ok) + require.NotNil(t, got.RunInstancesInput) + require.NotNil(t, got.RunInstancesInput.Monitoring) + return aws.BoolValue(got.RunInstancesInput.Monitoring.Enabled) +} + +func TestMonitorInstances_RoutedViaOwner(t *testing.T) { + const id = "i-mon-owner" + d, _ := monitoringTestDaemon(t, id, false) + subscribeOwner(t, d, id) + + out, err := d.monitorInstances(context.Background(), &ec2.MonitorInstancesInput{ + InstanceIds: []*string{aws.String(id)}, + }, testAccountID) + require.NoError(t, err) + + require.Len(t, out.InstanceMonitorings, 1) + assert.Equal(t, id, aws.StringValue(out.InstanceMonitorings[0].InstanceId)) + assert.Equal(t, ec2.MonitoringStateEnabled, aws.StringValue(out.InstanceMonitorings[0].Monitoring.State)) + assert.True(t, recordMonitoring(t, d, id)) +} + +func TestUnmonitorInstances_RoutedViaOwner(t *testing.T) { + const id = "i-mon-unowner" + d, _ := monitoringTestDaemon(t, id, true) + subscribeOwner(t, d, id) + + out, err := d.unmonitorInstances(context.Background(), &ec2.UnmonitorInstancesInput{ + InstanceIds: []*string{aws.String(id)}, + }, testAccountID) + require.NoError(t, err) + + require.Len(t, out.InstanceMonitorings, 1) + assert.Equal(t, ec2.MonitoringStateDisabled, aws.StringValue(out.InstanceMonitorings[0].Monitoring.State)) + assert.False(t, recordMonitoring(t, d, id)) +} + +// A launch that never asked for monitoring has no Monitoring block, so the +// running path has to create one rather than skip the instance. +func TestMonitorInstances_CreatesMissingBlock(t *testing.T) { + const id = "i-mon-noblock" + d, _ := monitoringTestDaemon(t, id, false) + d.vmMgr.Insert(&vm.VM{ + ID: id, + Status: vm.StateRunning, + AccountID: testAccountID, + Instance: &ec2.Instance{InstanceId: aws.String(id)}, + RunInstancesInput: &ec2.RunInstancesInput{}, + }) + subscribeOwner(t, d, id) + + _, err := d.monitorInstances(context.Background(), &ec2.MonitorInstancesInput{ + InstanceIds: []*string{aws.String(id)}, + }, testAccountID) + require.NoError(t, err) + assert.True(t, recordMonitoring(t, d, id)) +} + +// With no running owner the call falls back to the shared stopped store, so a +// stopped instance can still be moved between tiers. +func TestMonitorInstances_StoppedFallback(t *testing.T) { + const id = "i-mon-stopped" + d, stopped := monitoringTestDaemon(t, "i-mon-unrelated", false) + stopped.Stopped[id] = &vm.VM{ + ID: id, + Status: vm.StateStopped, + AccountID: testAccountID, + Instance: &ec2.Instance{InstanceId: aws.String(id)}, + RunInstancesInput: &ec2.RunInstancesInput{}, + } + + _, err := d.monitorInstances(context.Background(), &ec2.MonitorInstancesInput{ + InstanceIds: []*string{aws.String(id)}, + }, testAccountID) + require.NoError(t, err) + assert.True(t, aws.BoolValue(stopped.Stopped[id].RunInstancesInput.Monitoring.Enabled)) +} + +// An instance nobody owns and no stopped record covers is NotFound, not a +// silent success that leaves the caller believing the tier changed. +func TestMonitorInstances_NoOwnerNotFound(t *testing.T) { + d, _ := monitoringTestDaemon(t, "i-mon-unrelated2", false) + + _, err := d.monitorInstances(context.Background(), &ec2.MonitorInstancesInput{ + InstanceIds: []*string{aws.String("i-mon-gone1"), aws.String("i-mon-gone2")}, + }, testAccountID) + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error()) +} + +// The owner's cross-account rejection is relayed and the tier is untouched. +func TestMonitorInstances_OwnerErrorRelayed(t *testing.T) { + const id = "i-mon-crossacct" + const attacker = "999999999999" + d, _ := monitoringTestDaemon(t, id, false) + subscribeOwner(t, d, id) + + _, err := d.monitorInstances(context.Background(), &ec2.MonitorInstancesInput{ + InstanceIds: []*string{aws.String(id)}, + }, attacker) + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error()) + assert.False(t, recordMonitoring(t, d, id)) +} + +// One unreachable instance in a batch fails the whole call rather than +// reporting a tier for the instances that did apply. +func TestMonitorInstances_PartialFailureReportsNoState(t *testing.T) { + const id = "i-mon-partial" + d, _ := monitoringTestDaemon(t, id, false) + subscribeOwner(t, d, id) + + out, err := d.monitorInstances(context.Background(), &ec2.MonitorInstancesInput{ + InstanceIds: []*string{aws.String(id), aws.String("i-mon-partial-gone")}, + }, testAccountID) + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error()) + assert.Nil(t, out) +} + +func TestMonitorInstances_Validation(t *testing.T) { + d, _ := monitoringTestDaemon(t, "i-mon-valid", false) + + _, err := d.monitorInstances(context.Background(), nil, testAccountID) + assert.Equal(t, awserrors.ErrorInvalidParameterValue, err.Error()) + + _, err = d.monitorInstances(context.Background(), &ec2.MonitorInstancesInput{}, testAccountID) + assert.Equal(t, awserrors.ErrorMissingParameter, err.Error()) + + _, err = d.unmonitorInstances(context.Background(), nil, testAccountID) + assert.Equal(t, awserrors.ErrorInvalidParameterValue, err.Error()) + + _, err = d.unmonitorInstances(context.Background(), &ec2.UnmonitorInstancesInput{}, testAccountID) + assert.Equal(t, awserrors.ErrorMissingParameter, err.Error()) + + _, err = d.monitorInstances(context.Background(), &ec2.MonitorInstancesInput{ + InstanceIds: []*string{nil}, + }, testAccountID) + assert.Equal(t, awserrors.ErrorInvalidInstanceIDMalformed, err.Error()) +} diff --git a/spinifex/daemon/daemon_handlers_tag.go b/spinifex/daemon/daemon_handlers_tag.go index b7c22dc15..2f0434a38 100644 --- a/spinifex/daemon/daemon_handlers_tag.go +++ b/spinifex/daemon/daemon_handlers_tag.go @@ -16,10 +16,10 @@ import ( "github.com/nats-io/nats.go" ) -// instanceTagCommandTimeout bounds each ec2.cmd owner request. A stopped or +// instanceOwnerCommandTimeout bounds each ec2.cmd owner request. A stopped or // absent instance has no subscriber and fails fast with no responders, so // only a partitioned-but-subscribed owner pays the full timeout. -const instanceTagCommandTimeout = 5 * time.Second +const instanceOwnerCommandTimeout = 5 * time.Second // recordTagMirror is implemented by every service whose resource records // mirror the central tag store, so tag-filtered describes observe tags @@ -210,7 +210,7 @@ func (d *Daemon) tagInstance(ctx context.Context, instanceID string, data *types reqMsg.Data = body reqMsg.Header.Set(utils.AccountIDHeader, accountID) - msg, err := d.natsConn.RequestMsg(reqMsg, instanceTagCommandTimeout) + msg, err := d.natsConn.RequestMsg(reqMsg, instanceOwnerCommandTimeout) switch { case errors.Is(err, nats.ErrNoResponders): return d.instanceService.TagStoppedInstance(ctx, instanceID, data, remove, d.tagsService, accountID) diff --git a/spinifex/daemon/nats_classification_test.go b/spinifex/daemon/nats_classification_test.go index 3b1ae89b2..200b0e95c 100644 --- a/spinifex/daemon/nats_classification_test.go +++ b/spinifex/daemon/nats_classification_test.go @@ -7,6 +7,7 @@ import ( "errors" "log/slog" "net/http" + "reflect" "testing" "time" @@ -135,11 +136,17 @@ func TestEC2CommandNameCoversEveryCommand(t *testing.T) { {"SetSpotLineage", func(a *types.EC2CommandAttributes) { a.SetSpotLineage = true }}, {"SetInstanceTags", func(a *types.EC2CommandAttributes) { a.SetInstanceTags = true }}, {"RemoveInstanceTags", func(a *types.EC2CommandAttributes) { a.RemoveInstanceTags = true }}, + {"SetInstanceMonitoring", func(a *types.EC2CommandAttributes) { a.SetInstanceMonitoring = true }}, {"StartInstance", func(a *types.EC2CommandAttributes) { a.StartInstance = true }}, {"RebootInstance", func(a *types.EC2CommandAttributes) { a.RebootInstance = true }}, {"StopInstance", func(a *types.EC2CommandAttributes) { a.StopInstance = true }}, {"TerminateInstance", func(a *types.EC2CommandAttributes) { a.TerminateInstance = true }}, } + // The table is hand-written, so without this a new attribute would pass by + // simply not being listed — and its commands would report as "unknown". + assert.Len(t, tests, reflect.TypeFor[types.EC2CommandAttributes]().NumField(), + "every EC2CommandAttributes field needs a case here and in ec2CommandName") + for _, tc := range tests { t.Run(tc.want, func(t *testing.T) { assert.Equal(t, tc.want, ec2CommandName(commandFor("i-123", tc.set))) diff --git a/spinifex/gateway/ec2.go b/spinifex/gateway/ec2.go index 6c0f7db85..2f2189064 100644 --- a/spinifex/gateway/ec2.go +++ b/spinifex/gateway/ec2.go @@ -169,6 +169,12 @@ var ec2Actions = map[string]EC2Handler{ "RebootInstances": ec2Handler(func(ctx context.Context, input *ec2.RebootInstancesInput, gw *GatewayConfig, accountID string) (any, error) { return gateway_ec2_instance.RebootInstances(ctx, input, gw.NATSConn, accountID) }), + "MonitorInstances": ec2Handler(func(ctx context.Context, input *ec2.MonitorInstancesInput, gw *GatewayConfig, accountID string) (any, error) { + return gateway_ec2_instance.MonitorInstances(ctx, input, gw.NATSConn, accountID) + }), + "UnmonitorInstances": ec2Handler(func(ctx context.Context, input *ec2.UnmonitorInstancesInput, gw *GatewayConfig, accountID string) (any, error) { + return gateway_ec2_instance.UnmonitorInstances(ctx, input, gw.NATSConn, accountID) + }), "TerminateInstances": ec2Handler(func(ctx context.Context, input *ec2.TerminateInstancesInput, gw *GatewayConfig, accountID string) (any, error) { return gateway_ec2_instance.TerminateInstances(ctx, input, gw.NATSConn, accountID) }), diff --git a/spinifex/gateway/ec2/instance/MonitorInstances.go b/spinifex/gateway/ec2/instance/MonitorInstances.go new file mode 100644 index 000000000..3da297d46 --- /dev/null +++ b/spinifex/gateway/ec2/instance/MonitorInstances.go @@ -0,0 +1,56 @@ +package gateway_ec2_instance + +import ( + "context" + "errors" + "strings" + "time" + + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/awserrors" + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/nats-io/nats.go" +) + +// monitoringRequestTimeout bounds the round trip. The daemon fans out one +// owner request per instance concurrently and each of those is capped well +// below this, so the budget does not scale with the instance count. +const monitoringRequestTimeout = 30 * time.Second + +// validateMonitoringInstanceIDs rejects an empty or malformed id list before +// anything reaches NATS. +func validateMonitoringInstanceIDs(instanceIDs []*string) error { + if len(instanceIDs) == 0 { + return errors.New(awserrors.ErrorMissingParameter) + } + for _, id := range instanceIDs { + if id == nil || !strings.HasPrefix(*id, "i-") { + return errors.New(awserrors.ErrorInvalidInstanceIDMalformed) + } + } + return nil +} + +// MonitorInstances moves the given instances to the 60s detailed telemetry +// tier. The owning daemon applies it, so an instance running on any node or +// sitting stopped in shared KV is reachable. +func MonitorInstances(ctx context.Context, input *ec2.MonitorInstancesInput, natsConn *nats.Conn, accountID string) (*ec2.MonitorInstancesOutput, error) { + if input == nil { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + if err := validateMonitoringInstanceIDs(input.InstanceIds); err != nil { + return nil, err + } + return utils.NATSRequest[ec2.MonitorInstancesOutput](ctx, natsConn, "ec2.MonitorInstances", input, monitoringRequestTimeout, accountID) +} + +// UnmonitorInstances returns the given instances to the 300s basic tier. +func UnmonitorInstances(ctx context.Context, input *ec2.UnmonitorInstancesInput, natsConn *nats.Conn, accountID string) (*ec2.UnmonitorInstancesOutput, error) { + if input == nil { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + if err := validateMonitoringInstanceIDs(input.InstanceIds); err != nil { + return nil, err + } + return utils.NATSRequest[ec2.UnmonitorInstancesOutput](ctx, natsConn, "ec2.UnmonitorInstances", input, monitoringRequestTimeout, accountID) +} diff --git a/spinifex/gateway/ec2/instance/MonitorInstances_test.go b/spinifex/gateway/ec2/instance/MonitorInstances_test.go new file mode 100644 index 000000000..e707fcf4d --- /dev/null +++ b/spinifex/gateway/ec2/instance/MonitorInstances_test.go @@ -0,0 +1,119 @@ +//test:in-package — reuses startTestNATSServer, the package's unexported NATS +//fixture that every other gateway instance test is built on. + +package gateway_ec2_instance + +import ( + "context" + "encoding/json" + "testing" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/awserrors" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// serveMonitoring answers the daemon subject with the tier the request asked +// for, and records the subject it was reached on. +func serveMonitoring(t *testing.T, nc *nats.Conn, subject, state string, ids ...string) *string { + t.Helper() + seen := new(string) + monitorings := make([]*ec2.InstanceMonitoring, 0, len(ids)) + for _, id := range ids { + monitorings = append(monitorings, &ec2.InstanceMonitoring{ + InstanceId: aws.String(id), + Monitoring: &ec2.Monitoring{State: aws.String(state)}, + }) + } + _, err := nc.Subscribe(subject, func(msg *nats.Msg) { + *seen = msg.Subject + body, err := json.Marshal(&ec2.MonitorInstancesOutput{InstanceMonitorings: monitorings}) + require.NoError(t, err) + require.NoError(t, msg.Respond(body)) + }) + require.NoError(t, err) + return seen +} + +func TestMonitorInstances_Success(t *testing.T) { + _, nc := startTestNATSServer(t) + seen := serveMonitoring(t, nc, "ec2.MonitorInstances", ec2.MonitoringStateEnabled, "i-0123456789abcdef0") + + out, err := MonitorInstances(context.Background(), &ec2.MonitorInstancesInput{ + InstanceIds: []*string{aws.String("i-0123456789abcdef0")}, + }, nc, "123456789012") + + require.NoError(t, err) + require.Len(t, out.InstanceMonitorings, 1) + assert.Equal(t, ec2.MonitoringStateEnabled, aws.StringValue(out.InstanceMonitorings[0].Monitoring.State)) + assert.Equal(t, "ec2.MonitorInstances", *seen) +} + +func TestUnmonitorInstances_Success(t *testing.T) { + _, nc := startTestNATSServer(t) + seen := serveMonitoring(t, nc, "ec2.UnmonitorInstances", ec2.MonitoringStateDisabled, "i-0123456789abcdef0") + + out, err := UnmonitorInstances(context.Background(), &ec2.UnmonitorInstancesInput{ + InstanceIds: []*string{aws.String("i-0123456789abcdef0")}, + }, nc, "123456789012") + + require.NoError(t, err) + require.Len(t, out.InstanceMonitorings, 1) + assert.Equal(t, ec2.MonitoringStateDisabled, aws.StringValue(out.InstanceMonitorings[0].Monitoring.State)) + assert.Equal(t, "ec2.UnmonitorInstances", *seen) +} + +// The two actions must not share a subject, or disabling monitoring would +// enable it. +func TestMonitorInstances_DistinctSubjects(t *testing.T) { + _, nc := startTestNATSServer(t) + serveMonitoring(t, nc, "ec2.MonitorInstances", ec2.MonitoringStateEnabled, "i-0123456789abcdef0") + + _, err := UnmonitorInstances(context.Background(), &ec2.UnmonitorInstancesInput{ + InstanceIds: []*string{aws.String("i-0123456789abcdef0")}, + }, nc, "123456789012") + require.Error(t, err, "unmonitor must not be served by the monitor subscriber") +} + +func TestMonitorInstances_InvalidInput(t *testing.T) { + _, nc := startTestNATSServer(t) + + tests := []struct { + name string + ids []*string + want string + }{ + {"empty list", nil, awserrors.ErrorMissingParameter}, + {"nil id", []*string{nil}, awserrors.ErrorInvalidInstanceIDMalformed}, + {"missing i- prefix", []*string{aws.String("0123456789abcdef0")}, awserrors.ErrorInvalidInstanceIDMalformed}, + // One bad id in an otherwise valid list must reject the whole call + // rather than silently apply the tier to the good ones. + {"one bad id", []*string{aws.String("i-0123456789abcdef0"), aws.String("vol-1")}, awserrors.ErrorInvalidInstanceIDMalformed}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, err := MonitorInstances(context.Background(), &ec2.MonitorInstancesInput{InstanceIds: tt.ids}, nc, "123456789012") + require.Error(t, err) + assert.Equal(t, tt.want, err.Error()) + + _, err = UnmonitorInstances(context.Background(), &ec2.UnmonitorInstancesInput{InstanceIds: tt.ids}, nc, "123456789012") + require.Error(t, err) + assert.Equal(t, tt.want, err.Error()) + }) + } +} + +func TestMonitorInstances_NilInput(t *testing.T) { + _, nc := startTestNATSServer(t) + + _, err := MonitorInstances(context.Background(), nil, nc, "123456789012") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidParameterValue, err.Error()) + + _, err = UnmonitorInstances(context.Background(), nil, nc, "123456789012") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidParameterValue, err.Error()) +} diff --git a/spinifex/handlers/ec2/instance/instance_monitoring_test.go b/spinifex/handlers/ec2/instance/instance_monitoring_test.go new file mode 100644 index 000000000..1e3a91dc0 --- /dev/null +++ b/spinifex/handlers/ec2/instance/instance_monitoring_test.go @@ -0,0 +1,140 @@ +//test:in-package — constructs InstanceServiceImpl with only its unexported +//stoppedStore field set, so the stopped path is exercised without standing up +//the rest of the service. + +package handlers_ec2_instance + +import ( + "context" + "testing" + + "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/vm" + vmmock "github.com/mulgadc/spinifex/spinifex/vm/mock" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func stoppedMonitoredInstance(id, accountID string, enabled *bool) *vm.VM { + var monitoring *ec2.RunInstancesMonitoringEnabled + if enabled != nil { + monitoring = &ec2.RunInstancesMonitoringEnabled{Enabled: enabled} + } + return &vm.VM{ + ID: id, + AccountID: accountID, + Status: vm.StateStopped, + Instance: &ec2.Instance{InstanceId: aws.String(id)}, + RunInstancesInput: &ec2.RunInstancesInput{Monitoring: monitoring}, + } +} + +func storedMonitoringEnabled(t *testing.T, v *vm.VM) bool { + t.Helper() + require.NotNil(t, v) + require.NotNil(t, v.RunInstancesInput) + require.NotNil(t, v.RunInstancesInput.Monitoring) + return aws.BoolValue(v.RunInstancesInput.Monitoring.Enabled) +} + +func TestSetStoppedInstanceMonitoring_Enable(t *testing.T) { + stored := stoppedMonitoredInstance("i-123", "111122223333", aws.Bool(false)) + store := &vmmock.StateStore{Stopped: map[string]*vm.VM{"i-123": stored}} + svc := &InstanceServiceImpl{stoppedStore: store} + + require.NoError(t, svc.SetStoppedInstanceMonitoring(context.Background(), "i-123", true, "111122223333")) + assert.True(t, storedMonitoringEnabled(t, store.Stopped["i-123"])) + assert.Equal(t, 1, store.UpdateStoppedCalls) +} + +func TestSetStoppedInstanceMonitoring_Disable(t *testing.T) { + stored := stoppedMonitoredInstance("i-123", "111122223333", aws.Bool(true)) + store := &vmmock.StateStore{Stopped: map[string]*vm.VM{"i-123": stored}} + svc := &InstanceServiceImpl{stoppedStore: store} + + require.NoError(t, svc.SetStoppedInstanceMonitoring(context.Background(), "i-123", false, "111122223333")) + assert.False(t, storedMonitoringEnabled(t, store.Stopped["i-123"])) +} + +// A launch that never asked for monitoring has no Monitoring block at all, so +// enabling one has to create it rather than assume it is there. +func TestSetStoppedInstanceMonitoring_CreatesMissingBlock(t *testing.T) { + stored := stoppedMonitoredInstance("i-123", "111122223333", nil) + store := &vmmock.StateStore{Stopped: map[string]*vm.VM{"i-123": stored}} + svc := &InstanceServiceImpl{stoppedStore: store} + + require.NoError(t, svc.SetStoppedInstanceMonitoring(context.Background(), "i-123", true, "111122223333")) + assert.True(t, storedMonitoringEnabled(t, store.Stopped["i-123"])) +} + +func TestSetStoppedInstanceMonitoring_NotFound(t *testing.T) { + store := &vmmock.StateStore{Stopped: map[string]*vm.VM{}} + svc := &InstanceServiceImpl{stoppedStore: store} + + err := svc.SetStoppedInstanceMonitoring(context.Background(), "i-missing", true, "111122223333") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error()) + assert.Empty(t, store.WroteStopped) +} + +func TestSetStoppedInstanceMonitoring_CrossAccountRejected(t *testing.T) { + stored := stoppedMonitoredInstance("i-123", "999988887777", aws.Bool(false)) + store := &vmmock.StateStore{Stopped: map[string]*vm.VM{"i-123": stored}} + svc := &InstanceServiceImpl{stoppedStore: store} + + err := svc.SetStoppedInstanceMonitoring(context.Background(), "i-123", true, "111122223333") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error()) + assert.False(t, storedMonitoringEnabled(t, store.Stopped["i-123"])) + assert.Empty(t, store.WroteStopped) +} + +// A monitoring toggle racing a winning start-claim must not resurrect the +// stopped record the claim deleted between this call's Load and its CAS write. +func TestSetStoppedInstanceMonitoring_ConcurrentClaimDoesNotResurrect(t *testing.T) { + stored := stoppedMonitoredInstance("i-123", "111122223333", aws.Bool(false)) + store := &vmmock.StateStore{Stopped: map[string]*vm.VM{"i-123": stored}, ClaimAfterLoad: true} + svc := &InstanceServiceImpl{stoppedStore: store} + + err := svc.SetStoppedInstanceMonitoring(context.Background(), "i-123", true, "111122223333") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error()) + assert.Empty(t, store.Stopped, "the claimed record must not be resurrected") + assert.Equal(t, []string{"i-123"}, store.ClaimedStopped) +} + +func TestSetStoppedInstanceMonitoring_NoStore(t *testing.T) { + svc := &InstanceServiceImpl{} + err := svc.SetStoppedInstanceMonitoring(context.Background(), "i-123", true, "111122223333") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorServerInternal, err.Error()) +} + +func TestProjectInstance_MonitoringState(t *testing.T) { + tests := []struct { + name string + input *ec2.RunInstancesInput + want string + }{ + {"detailed", &ec2.RunInstancesInput{Monitoring: &ec2.RunInstancesMonitoringEnabled{Enabled: aws.Bool(true)}}, ec2.MonitoringStateEnabled}, + {"basic", &ec2.RunInstancesInput{Monitoring: &ec2.RunInstancesMonitoringEnabled{Enabled: aws.Bool(false)}}, ec2.MonitoringStateDisabled}, + {"no monitoring block", &ec2.RunInstancesInput{}, ec2.MonitoringStateDisabled}, + // Instances launched before the field existed carry no request at all. + {"no launch input", nil, ec2.MonitoringStateDisabled}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + v := &vm.VM{ + ID: "i-123", + Status: vm.StateRunning, + Instance: &ec2.Instance{InstanceId: aws.String("i-123")}, + RunInstancesInput: tt.input, + } + projected, _ := ProjectInstance(v, InstanceProjection{}) + require.NotNil(t, projected.Monitoring) + assert.Equal(t, tt.want, aws.StringValue(projected.Monitoring.State)) + }) + } +} diff --git a/spinifex/handlers/ec2/instance/projection.go b/spinifex/handlers/ec2/instance/projection.go index 90135780c..33502050c 100644 --- a/spinifex/handlers/ec2/instance/projection.go +++ b/spinifex/handlers/ec2/instance/projection.go @@ -113,6 +113,17 @@ func ProjectInstance(v *vm.VM, cfg InstanceProjection) (inst *ec2.Instance, stat } } + // Monitoring reads the same field the telemetry collector's period does, so + // a describe cannot disagree with the tier actually being collected. Only + // enabled/disabled: the collector converges within one discovery interval, + // leaving no pending/disabling window for a client to wait on. + monitoringState := ec2.MonitoringStateDisabled + if v.RunInstancesInput != nil && v.RunInstancesInput.Monitoring != nil && + aws.BoolValue(v.RunInstancesInput.Monitoring.Enabled) { + monitoringState = ec2.MonitoringStateEnabled + } + instanceCopy.Monitoring = &ec2.Monitoring{State: aws.String(monitoringState)} + // Spot lineage stamped by the post-launch write-back survives a stop, so // project it regardless of runtime state. Both empty for on-demand, so the // fields stay absent there. diff --git a/spinifex/handlers/ec2/instance/service_impl.go b/spinifex/handlers/ec2/instance/service_impl.go index e1a1f6c5f..e1fcf1891 100644 --- a/spinifex/handlers/ec2/instance/service_impl.go +++ b/spinifex/handlers/ec2/instance/service_impl.go @@ -419,6 +419,52 @@ func (s *InstanceServiceImpl) TagStoppedInstance(ctx context.Context, instanceID return nil } +// SetStoppedInstanceMonitoring applies a monitoring-tier change to a stopped +// instance's shared KV record. Nothing polls a stopped instance, so there is +// no discovery file to rewrite: the next start stamps one from this flag. +func (s *InstanceServiceImpl) SetStoppedInstanceMonitoring(ctx context.Context, instanceID string, enabled bool, accountID string) error { + if s.stoppedStore == nil { + slog.ErrorContext(ctx, "SetStoppedInstanceMonitoring: stopped store not available") + return errors.New(awserrors.ErrorServerInternal) + } + + instance, err := s.stoppedStore.LoadStoppedInstance(instanceID) + if err != nil { + slog.ErrorContext(ctx, "SetStoppedInstanceMonitoring: failed to load stopped instance", "instanceId", instanceID, "err", err) + return errors.New(awserrors.ErrorServerInternal) + } + if instance == nil { + return errors.New(awserrors.ErrorInvalidInstanceIDNotFound) + } + if !IsInstanceVisible(accountID, instance.AccountID) { + slog.WarnContext(ctx, "SetStoppedInstanceMonitoring: instance not visible", + "instanceId", instanceID, "callerAccount", accountID, "ownerAccount", instance.AccountID) + return errors.New(awserrors.ErrorInvalidInstanceIDNotFound) + } + + // CAS with createIfAbsent=false so a start-claim that deletes the record + // between the Load above and this write fails cleanly rather than + // resurrecting a stale stopped entry. + if _, err := s.stoppedStore.UpdateStoppedInstance(instanceID, func(v *vm.VM) { + if v.RunInstancesInput == nil { + return + } + if v.RunInstancesInput.Monitoring == nil { + v.RunInstancesInput.Monitoring = &ec2.RunInstancesMonitoringEnabled{} + } + v.RunInstancesInput.Monitoring.Enabled = aws.Bool(enabled) + }); err != nil { + if errors.Is(err, jetstream.ErrKeyNotFound) { + slog.WarnContext(ctx, "SetStoppedInstanceMonitoring: instance claimed concurrently, not resurrecting", "instanceId", instanceID) + return errors.New(awserrors.ErrorInvalidInstanceIDNotFound) + } + slog.ErrorContext(ctx, "SetStoppedInstanceMonitoring: failed to write stopped instance", "instanceId", instanceID, "err", err) + return errors.New(awserrors.ErrorServerInternal) + } + slog.InfoContext(ctx, "SetStoppedInstanceMonitoring: tier applied", "instanceId", instanceID, "enabled", enabled) + return nil +} + // deleteCentralInstanceTags removes a terminated instance's central tag store // entry so describe-tags stops reporting it, while the terminated record keeps // its tags until TTL. Best-effort: the terminate already succeeded, so a diff --git a/spinifex/services/qmpcollector/poller.go b/spinifex/services/qmpcollector/poller.go index 4a646b228..16bb5f35f 100644 --- a/spinifex/services/qmpcollector/poller.go +++ b/spinifex/services/qmpcollector/poller.go @@ -49,20 +49,35 @@ type poller struct { nc *nats.Conn cancel context.CancelFunc + // retier wakes run when the monitoring tier changes, so a period move is + // not queued behind a tick of the period being replaced. + retier chan struct{} + mu sync.Mutex meta types.GuestTelemetryMeta prev *sample } func newPoller(cfg *Config, nc *nats.Conn, meta types.GuestTelemetryMeta) *poller { - return &poller{cfg: cfg, nc: nc, meta: meta} + return &poller{cfg: cfg, nc: nc, meta: meta, retier: make(chan struct{}, 1)} } -// updateMeta refreshes taps/period after an ENI hot-plug rewrite. +// updateMeta refreshes taps/period after an ENI hot-plug or monitoring-tier +// rewrite, waking the run loop when the period moved. func (p *poller) updateMeta(meta types.GuestTelemetryMeta) { p.mu.Lock() + retier := meta.PeriodSeconds != p.meta.PeriodSeconds p.meta = meta p.mu.Unlock() + if !retier { + return + } + // Non-blocking: the loop re-reads the period itself, so one pending + // signal is enough and a slow poller must not stall the collector. + select { + case p.retier <- struct{}{}: + default: + } } func (p *poller) snapshotMeta() types.GuestTelemetryMeta { @@ -100,18 +115,33 @@ func (p *poller) run(ctx context.Context) { ticker := time.NewTicker(period) defer ticker.Stop() - for { - p.tick(ctx) - select { - case <-ctx.Done(): - return - case <-ticker.C: - } + + applyPeriod := func() { if next := p.period(); next != period { period = next ticker.Reset(period) } } + + for { + p.tick(ctx) + // Reset on the retier signal rather than waiting out the interval + // being replaced: at 300s, moving to the detailed tier would + // otherwise take five minutes to take effect. The next sample still + // lands a full new period away, so a published delta always spans + // the period it declares. + for ticked := false; !ticked; { + select { + case <-ctx.Done(): + return + case <-p.retier: + applyPeriod() + case <-ticker.C: + ticked = true + } + } + applyPeriod() + } } // tick takes a snapshot and publishes the delta against the previous one. diff --git a/spinifex/services/qmpcollector/poller_test.go b/spinifex/services/qmpcollector/poller_test.go index 479b29634..11952a662 100644 --- a/spinifex/services/qmpcollector/poller_test.go +++ b/spinifex/services/qmpcollector/poller_test.go @@ -194,6 +194,46 @@ func TestPollerPeriod(t *testing.T) { } } +// A tier change has to wake the run loop. Without the signal the reset waits +// out the interval being replaced, so moving a 300s instance to the detailed +// tier would take five minutes to take effect. +func TestUpdateMetaSignalsOnlyTierChanges(t *testing.T) { + p := newPoller(&Config{}, nil, types.GuestTelemetryMeta{InstanceID: "i-0aaa", PeriodSeconds: 300}) + + signalled := func() bool { + select { + case <-p.retier: + return true + default: + return false + } + } + + p.updateMeta(types.GuestTelemetryMeta{InstanceID: "i-0aaa", PeriodSeconds: 300, Taps: []string{"tap0"}}) + if signalled() { + t.Error("a taps-only rewrite must not signal a tier change") + } + + p.updateMeta(types.GuestTelemetryMeta{InstanceID: "i-0aaa", PeriodSeconds: 60}) + if !signalled() { + t.Fatal("moving to the detailed tier must signal") + } + if got := p.period(); got != 60*time.Second { + t.Errorf("period = %v, want 60s", got) + } + + // The signal is coalesced, so a poller that has not woken yet cannot + // accumulate a backlog of resets. + p.updateMeta(types.GuestTelemetryMeta{InstanceID: "i-0aaa", PeriodSeconds: 300}) + p.updateMeta(types.GuestTelemetryMeta{InstanceID: "i-0aaa", PeriodSeconds: 60}) + if !signalled() { + t.Fatal("a pending tier change must be visible") + } + if signalled() { + t.Error("repeated changes must coalesce into one pending signal") + } +} + func TestStaggerOffset(t *testing.T) { period := 60 * time.Second a := staggerOffset("i-0aaa", period) diff --git a/spinifex/types/ec2.go b/spinifex/types/ec2.go index da052c345..ba696f9f8 100644 --- a/spinifex/types/ec2.go +++ b/spinifex/types/ec2.go @@ -13,6 +13,7 @@ type EC2InstanceCommand struct { IamProfileAssociationData *IamProfileAssociationData `json:"iam_profile_association_data,omitempty"` SpotLineageData *SpotLineageData `json:"spot_lineage_data,omitempty"` InstanceTagsData *InstanceTagsData `json:"instance_tags_data,omitempty"` + InstanceMonitoringData *InstanceMonitoringData `json:"instance_monitoring_data,omitempty"` } // EC2CommandAttributes indicates which action the daemon should perform. @@ -30,6 +31,7 @@ type EC2CommandAttributes struct { SetSpotLineage bool `json:"set_spot_lineage,omitempty"` SetInstanceTags bool `json:"set_instance_tags,omitempty"` RemoveInstanceTags bool `json:"remove_instance_tags,omitempty"` + SetInstanceMonitoring bool `json:"set_instance_monitoring,omitempty"` } // AttachVolumeData carries parameters for an attach-volume command. @@ -100,6 +102,14 @@ type InstanceTagsData struct { TagKeys []string `json:"tag_keys,omitempty"` } +// InstanceMonitoringData carries the desired EC2 monitoring tier for a +// set-instance-monitoring command: enabled is 60s detailed, disabled is 300s +// basic. Carried as data rather than a second attribute so "disable" is an +// explicit value, not the absence of one. +type InstanceMonitoringData struct { + Enabled bool `json:"enabled"` +} + // SpotLineageData carries the SIR id stamped onto a spot-launched instance. // Lifecycle is always "spot" for this command, so only the SIR id travels. type SpotLineageData struct { diff --git a/spinifex/vm/telemetry_meta.go b/spinifex/vm/telemetry_meta.go index 4f45150ea..cf0ec64fc 100644 --- a/spinifex/vm/telemetry_meta.go +++ b/spinifex/vm/telemetry_meta.go @@ -80,6 +80,13 @@ func refreshTelemetryMeta(v *VM) { } } +// RefreshTelemetryMeta rewrites the discovery file so the collector converges +// on the VM's current monitoring tier and ENI set. Must not be called under +// the manager lock: telemetryTaps takes the VM's own ENI lock. +func (v *VM) RefreshTelemetryMeta() { + refreshTelemetryMeta(v) +} + // removeTelemetryArtifacts deletes the telemetry socket and discovery file so // the collector stops polling a gone VM. func removeTelemetryArtifacts(v *VM) { diff --git a/spinifex/vm/telemetry_meta_test.go b/spinifex/vm/telemetry_meta_test.go index 186b997de..e718815e5 100644 --- a/spinifex/vm/telemetry_meta_test.go +++ b/spinifex/vm/telemetry_meta_test.go @@ -90,3 +90,42 @@ func TestTelemetryMetaRoundtrip(t *testing.T) { t.Error("metadata must be removed") } } + +// A monitoring-tier change has to reach the discovery file, because that file +// is the only thing the collector reads to decide a poller's interval. +func TestRefreshTelemetryMetaCarriesTierChange(t *testing.T) { + t.Setenv("XDG_RUNTIME_DIR", t.TempDir()) + + v := &VM{ID: "i-0abc", AccountID: "123456789012", RunInstancesInput: &ec2.RunInstancesInput{}} + v.Config.TelemetryQMPSocket = utils.TelemetryMetaPath(v.ID) + ".sock" + + readPeriod := func() int { + t.Helper() + data, err := os.ReadFile(utils.TelemetryMetaPath(v.ID)) + if err != nil { + t.Fatalf("read back: %v", err) + } + var meta types.GuestTelemetryMeta + if err := json.Unmarshal(data, &meta); err != nil { + t.Fatalf("unmarshal: %v", err) + } + return meta.PeriodSeconds + } + + v.RefreshTelemetryMeta() + if got := readPeriod(); got != 300 { + t.Fatalf("basic tier period = %d, want 300", got) + } + + v.RunInstancesInput.Monitoring = &ec2.RunInstancesMonitoringEnabled{Enabled: aws.Bool(true)} + v.RefreshTelemetryMeta() + if got := readPeriod(); got != 60 { + t.Errorf("detailed tier period = %d, want 60", got) + } + + v.RunInstancesInput.Monitoring.Enabled = aws.Bool(false) + v.RefreshTelemetryMeta() + if got := readPeriod(); got != 300 { + t.Errorf("period after disable = %d, want 300", got) + } +}