feat: implement EC2 MonitorInstances and UnmonitorInstances (#853)

* feat: implement EC2 MonitorInstances and UnmonitorInstances

The telemetry collector already supports both EC2 monitoring tiers and
already switches a running poller between them when its discovery file
changes, but nothing could make that change: Monitoring.Enabled was only
ever written at launch, and both API actions returned InvalidAction.

Route through the owning daemon over ec2.cmd.<id> rather than the
spinifex-workers queue group, so an instance running on any node is
reachable; no responders falls back to the shared stopped store, where the
write is a CAS so a racing start-claim cannot resurrect a deleted record.

Also report Monitoring on DescribeInstances. It is derived in
ProjectInstance from the same field the collector's period reads, so a
describe cannot disagree with the tier actually being collected and
instances launched before this change report disabled without a backfill.

* fix: wake the telemetry poller when its monitoring tier changes

poller.run re-read its period inside the tick loop, so a reset waited out
the interval being replaced: moving a 300s instance to the detailed tier
took up to five minutes to have any effect, which reads as the API call
not working. Found verifying monitor-instances on env13.

updateMeta now signals a buffered retier channel when PeriodSeconds moves
and run selects on it alongside the ticker. The signal only resets the
ticker rather than forcing a tick, so a published delta still spans the
period its batch declares. The send is non-blocking and the channel
coalesces, so a slow poller cannot stall the reconcile loop.

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
 

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).

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:

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

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.<id> 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
+}

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())
+}

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)

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)))

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)
 	}),

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)
+}

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())
+}

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))
+		})
+	}
+}

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.

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

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.

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)

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 {

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) {

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)
+	}
+}