diff --git a/spinifex/awserrors/awserrors.go b/spinifex/awserrors/awserrors.go index 6f62e01cd..82f100bdc 100644 --- a/spinifex/awserrors/awserrors.go +++ b/spinifex/awserrors/awserrors.go @@ -3,6 +3,7 @@ package awserrors import ( "errors" "fmt" + "net/http" "strings" ) @@ -698,6 +699,16 @@ func ValidErrorCodeFromError(err error) string { return ErrorServerInternal } +// HTTPStatusForError returns the HTTP status err's AWS code maps to, or 500 +// for an error carrying no recognised code. It lets a caller off the HTTP path +// classify a failure the same way the gateway would have. +func HTTPStatusForError(err error) int { + if status := ErrorLookup[ValidErrorCodeFromError(err)].HTTPCode; status != 0 { + return status + } + return http.StatusInternalServerError +} + // HasErrorCode reports whether s is exactly a registered AWS error code. // Use it to decide whether an error string may be surfaced to clients verbatim // (mapped to its real HTTP status) rather than wrapped/sanitized to 500. diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index 8161eacd7..786980ff6 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -766,6 +766,11 @@ const ( outcomeError = "error" outcomeSkipped = "skipped" + // outcomeClientError is a caller mistake — an unknown id, a bad parameter. + // Separate from outcomeError so a daemon error rate measures the daemon + // rather than the callers reaching it. + outcomeClientError = "client_error" + // outcomeDeferred tells natsMetricsHandler the handler records its own // point. Handlers that answer from a goroutine must use it: timing the // wrapper would measure the dispatch and call every launch an instant @@ -782,6 +787,40 @@ func outcomeFor(replied bool) string { return outcomeError } +// outcomeForError classifies a handler failure by the status its AWS code maps +// to, through the same function the HTTP middleware uses. Anything carrying no +// recognised code lands on 500, so an unclassifiable failure counts against the +// daemon rather than being written off as the caller's fault. +func outcomeForError(err error) string { + return otelsetup.OutcomeForStatus(awserrors.HTTPStatusForError(err)) +} + +// outcomeForCode classifies by an AWS error code a handler has already chosen, +// for the paths that answer with a code rather than an error value. +func outcomeForCode(errCode string) string { + return outcomeForError(errors.New(errCode)) +} + +// ec2CmdAction names the metric action for a per-instance command. The subject +// carries the instance id, so the action is built from the command: an instance +// id in a metric dimension would make one series per instance. +func ec2CmdAction(command string) string { + return "ec2.cmd." + command +} + +// logHandlerError reports a handler failure at a level matching its +// classification: ERROR means the daemon failed. A client error is the caller's +// and is logged at WARN, which is what stops an expected fan-out miss writing an +// ERROR line on every node that was never going to answer. +func logHandlerError(ctx context.Context, msg string, subject string, err error) { + code := awserrors.ValidErrorCodeFromError(err) + if outcomeForError(err) == outcomeClientError { + slog.WarnContext(ctx, msg, "subject", subject, "code", code, "err", err) + return + } + slog.ErrorContext(ctx, msg, "subject", subject, "code", code, "err", err) +} + // natsHandler is a NATS handler that reports how it answered. The type exists // so the outcome is observable at the metrics chokepoint: a plain MsgHandler // writes its own reply and returns nothing, which is why daemon request points diff --git a/spinifex/daemon/daemon_handlers.go b/spinifex/daemon/daemon_handlers.go index b7fca9f6d..2b205b931 100644 --- a/spinifex/daemon/daemon_handlers.go +++ b/spinifex/daemon/daemon_handlers.go @@ -18,6 +18,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/formation" "github.com/mulgadc/spinifex/spinifex/gpu" "github.com/mulgadc/spinifex/spinifex/network/host" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/mulgadc/spinifex/spinifex/vm" @@ -65,6 +66,20 @@ func respondWithServiceError(msg *nats.Msg, err error) { } } +// respondErrorOutcome answers with errCode and reports the outcome that code +// implies, so a handler classifies its failure by the code it already chose +// rather than by a second judgement that could disagree with it. +func respondErrorOutcome(msg *nats.Msg, errCode string) string { + respondWithError(msg, errCode) + return outcomeForCode(errCode) +} + +// respondServiceErrorOutcome is respondErrorOutcome for an error value. +func respondServiceErrorOutcome(msg *nats.Msg, err error) string { + respondWithServiceError(msg, err) + return outcomeForError(err) +} + // respondWithJSON marshals data to JSON and sends it as a NATS response. // On marshal failure it responds with an internal server error. func respondWithJSON(msg *nats.Msg, data any) { @@ -98,17 +113,19 @@ func handleNATSRequest[I any, O any](serviceFn func(context.Context, *I, string) if err := msg.Respond(errResp); err != nil { slog.ErrorContext(ctx, "Failed to respond to NATS request", "err", err) } - return outcomeError + // A payload the daemon cannot parse is the caller's mistake, not a + // fault of its own. + return outcomeClientError } output, err := serviceFn(ctx, input, accountID) if err != nil { // The error was otherwise only recorded on the OTel span, invisible - // without a trace backend. Log at Error so handler failures show up - // in the journal too. - slog.ErrorContext(ctx, "handleNATSRequest: service call failed", "subject", msg.Subject, "err", err) + // without a trace backend, so it is logged here too — at a level + // that says whether the daemon or its caller was at fault. + logHandlerError(ctx, "handleNATSRequest: service call failed", msg.Subject, err) utils.MarkSpanError(span, err) respondWithServiceError(msg, err) - return outcomeError + return outcomeForError(err) } respondWithJSON(msg, output) return outcomeSuccess @@ -132,13 +149,14 @@ func handleNATSRequestWithPrincipal[I any, O any](serviceFn func(context.Context if err := msg.Respond(errResp); err != nil { slog.ErrorContext(ctx, "Failed to respond to NATS request", "err", err) } - return outcomeError + return outcomeClientError } output, err := serviceFn(ctx, input, accountID, principalARN) if err != nil { + logHandlerError(ctx, "handleNATSRequestWithPrincipal: service call failed", msg.Subject, err) utils.MarkSpanError(span, err) respondWithServiceError(msg, err) - return outcomeError + return outcomeForError(err) } respondWithJSON(msg, output) return outcomeSuccess @@ -146,7 +164,22 @@ func handleNATSRequestWithPrincipal[I any, O any](serviceFn func(context.Context } // handleEC2Events processes incoming EC2 instance events (start, stop, terminate, attach-volume). +// It reports its own action and outcome rather than returning them, because one +// subject carries a dozen different commands and the wrapper cannot name which +// one ran. func (d *Daemon) handleEC2Events(msg *nats.Msg) { + start := time.Now() + action, outcome := d.dispatchEC2Command(msg) + if outcome == outcomeDeferred { + return + } + otelsetup.RecordRequest(context.Background(), ec2CmdAction(action), outcome, time.Since(start)) +} + +// dispatchEC2Command runs one per-instance command and returns the command's +// name and how it answered. A command that reaches no case is named rather than +// dropped, so an attribute nothing handles is visible in the metric. +func (d *Daemon) dispatchEC2Command(msg *nats.Msg) (string, string) { ctx, span := utils.StartConsumerSpan(msg) defer span.End() @@ -156,69 +189,78 @@ func (d *Daemon) handleEC2Events(msg *nats.Msg) { slog.ErrorContext(ctx, "Error unmarshaling EC2 instance command", "err", err) utils.MarkSpanError(span, err) respondWithError(msg, awserrors.ErrorServerInternal) - return + return "unknown", outcomeError } slog.DebugContext(ctx, "Received message", "subject", msg.Subject, "data", string(msg.Data)) + name := ec2CommandName(command) + instance, ok := d.vmMgr.Get(command.ID) if !ok { slog.WarnContext(ctx, "Instance is not running on this node", "id", command.ID) - respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound) - return + return name, respondErrorOutcome(msg, awserrors.ErrorInvalidInstanceIDNotFound) } // Verify the caller owns this instance if !checkInstanceOwnership(msg, command.ID, instance.AccountID) { - return + return name, outcomeClientError } switch { case command.Attributes.AttachVolume: - d.handleAttachVolume(ctx, msg, command, instance) + return name, d.handleAttachVolume(ctx, msg, command, instance) case command.Attributes.DetachVolume: - d.handleDetachVolume(ctx, msg, command, instance) + return name, d.handleDetachVolume(ctx, msg, command, instance) case command.Attributes.DrainVolume: // A drain flushes the volume's whole dirty set to S3, so it is bounded by // the dirty set rather than by a unit of work. nats.go delivers a // subscription's messages serially, so running it inline would stall // every other command for this instance — stop, terminate, hot-plug — // for the duration. It touches no VM state, so it is safe off-thread. - go d.handleDrainVolume(ctx, msg, command, instance) + // + // It therefore records its own point: timing the dispatch would call + // every drain an instant success. + started := time.Now() + go func() { + outcome := d.handleDrainVolume(ctx, msg, command, instance) + otelsetup.RecordRequest(context.Background(), ec2CmdAction(name), outcome, time.Since(started)) + }() + return name, outcomeDeferred case command.Attributes.AttachENI: - d.handleAttachNetworkInterface(ctx, msg, command, instance) + return name, d.handleAttachNetworkInterface(ctx, msg, command, instance) case command.Attributes.DetachENI: - d.handleDetachNetworkInterface(ctx, msg, command, instance) + return name, d.handleDetachNetworkInterface(ctx, msg, command, instance) case command.Attributes.AssociateIamInstanceProfile: - d.handleAssociateIamInstanceProfile(ctx, msg, command, instance) + return name, d.handleAssociateIamInstanceProfile(ctx, msg, command, instance) case command.Attributes.SetSpotLineage: - d.handleSetSpotLineage(ctx, msg, command) + return name, d.handleSetSpotLineage(ctx, msg, command) case command.Attributes.SetInstanceTags, command.Attributes.RemoveInstanceTags: - d.handleSetInstanceTags(ctx, msg, command, instance) + return name, d.handleSetInstanceTags(ctx, msg, command, instance) case command.Attributes.StartInstance: opCtx, opSpan := startOpSpan(ctx, "ec2.StartInstance", instance.ID) err := d.instanceService.StartInstance(opCtx, instance, command) endOpSpan(opSpan, err) if err != nil { utils.MarkSpanError(span, err) - respondWithServiceError(msg, err) - return + return name, respondServiceErrorOutcome(msg, err) } if err := msg.Respond(fmt.Appendf(nil, `{"status":"running","instanceId":"%s"}`, instance.ID)); err != nil { slog.ErrorContext(ctx, "Failed to respond to NATS request", "err", err) } + return name, outcomeSuccess case command.Attributes.RebootInstance: opCtx, opSpan := startOpSpan(ctx, "ec2.RebootInstance", instance.ID) err := d.instanceService.RebootInstance(opCtx, instance, command) endOpSpan(opSpan, err) if err != nil { utils.MarkSpanError(span, err) - respondWithServiceError(msg, err) - return + return name, respondServiceErrorOutcome(msg, err) } if err := msg.Respond([]byte(`{}`)); err != nil { slog.ErrorContext(ctx, "Failed to respond to NATS request", "err", err) } + return name, outcomeSuccess case command.Attributes.StopInstance, command.Attributes.TerminateInstance: opName := "ec2.StopInstance" if command.Attributes.TerminateInstance { @@ -229,15 +271,50 @@ func (d *Daemon) handleEC2Events(msg *nats.Msg) { endOpSpan(opSpan, err) if err != nil { utils.MarkSpanError(span, err) - respondWithServiceError(msg, err) - return + return name, respondServiceErrorOutcome(msg, err) } if err := msg.Respond([]byte(`{}`)); err != nil { slog.ErrorContext(ctx, "Failed to respond to NATS request", "err", err) } + return name, outcomeSuccess default: slog.WarnContext(ctx, "Unhandled EC2 instance command", "id", command.ID, "attributes", command.Attributes) - respondWithError(msg, awserrors.ErrorServerInternal) + return name, respondErrorOutcome(msg, awserrors.ErrorServerInternal) + } +} + +// ec2CommandName names the command an EC2InstanceCommand carries, for the +// metric action. Order matches the dispatch switch. +func ec2CommandName(command types.EC2InstanceCommand) string { + switch { + case command.Attributes.AttachVolume: + return "AttachVolume" + case command.Attributes.DetachVolume: + return "DetachVolume" + case command.Attributes.DrainVolume: + return "DrainVolume" + case command.Attributes.AttachENI: + return "AttachNetworkInterface" + case command.Attributes.DetachENI: + return "DetachNetworkInterface" + case command.Attributes.AssociateIamInstanceProfile: + return "AssociateIamInstanceProfile" + case command.Attributes.SetSpotLineage: + return "SetSpotLineage" + case command.Attributes.SetInstanceTags: + return "SetInstanceTags" + case command.Attributes.RemoveInstanceTags: + return "RemoveInstanceTags" + case command.Attributes.StartInstance: + return "StartInstance" + case command.Attributes.RebootInstance: + return "RebootInstance" + case command.Attributes.TerminateInstance: + return "TerminateInstance" + case command.Attributes.StopInstance: + return "StopInstance" + default: + return "unknown" } } diff --git a/spinifex/daemon/daemon_handlers_eni.go b/spinifex/daemon/daemon_handlers_eni.go index 5a5274239..d95f682b4 100644 --- a/spinifex/daemon/daemon_handlers_eni.go +++ b/spinifex/daemon/daemon_handlers_eni.go @@ -20,12 +20,11 @@ import ( // handleAttachNetworkInterface updates KV first (crash-safe attaching state), // then runs the QMP hot-plug pipeline. On QMP failure the KV record rolls // back to available with the error persisted in LastAttachError. -func (d *Daemon) handleAttachNetworkInterface(ctx context.Context, msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) { +func (d *Daemon) handleAttachNetworkInterface(ctx context.Context, msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) string { slog.InfoContext(ctx, "Attaching ENI to instance", "instanceId", command.ID) if command.AttachENIData == nil || command.AttachENIData.NetworkInterfaceID == "" { - respondWithError(msg, awserrors.ErrorInvalidParameterValue) - return + return respondErrorOutcome(msg, awserrors.ErrorInvalidParameterValue) } eniID := command.AttachENIData.NetworkInterfaceID @@ -33,24 +32,20 @@ func (d *Daemon) handleAttachNetworkInterface(ctx context.Context, msg *nats.Msg accountID := utils.AccountIDFromMsg(msg) if status := d.vmMgr.Status(instance); status != vm.StateRunning { - respondWithError(msg, awserrors.ErrorIncorrectInstanceState) - return + return respondErrorOutcome(msg, awserrors.ErrorIncorrectInstanceState) } record, err := d.vpcService.GetENIRecord(accountID, eniID) if err != nil { - respondWithServiceError(msg, err) - return + return respondServiceErrorOutcome(msg, err) } if record.Status == "in-use" && record.InstanceId != instance.ID { - respondWithError(msg, awserrors.ErrorInvalidNetworkInterfaceInUse) - return + return respondErrorOutcome(msg, awserrors.ErrorInvalidNetworkInterfaceInUse) } attachmentID, err := d.vpcService.AttachENI(accountID, eniID, instance.ID, deviceIndex) if err != nil { - respondWithServiceError(msg, err) - return + return respondServiceErrorOutcome(msg, err) } if err := d.vpcService.UpdateENI(accountID, eniID, func(r *handlers_ec2_vpc.ENIRecord) { r.AttachmentStatus = "attaching" @@ -74,8 +69,7 @@ func (d *Daemon) handleAttachNetworkInterface(ctx context.Context, msg *nats.Msg r.HotPlugSlot = 0 r.LastAttachError = hotPlugErr.Error() }) - respondWithError(msg, eniHotplugErrorCode(hotPlugErr)) - return + return respondErrorOutcome(msg, eniHotplugErrorCode(hotPlugErr)) } if err := d.vpcService.UpdateENI(accountID, eniID, func(r *handlers_ec2_vpc.ENIRecord) { @@ -98,17 +92,17 @@ func (d *Daemon) handleAttachNetworkInterface(ctx context.Context, msg *nats.Msg respondWithJSON(msg, ec2.AttachNetworkInterfaceOutput{ AttachmentId: aws.String(attachmentID), }) + return outcomeSuccess } // handleDetachNetworkInterface marks the KV record as detaching first // (crash-safe), runs the QMP hot-unplug pipeline, then returns the // record to available on success. -func (d *Daemon) handleDetachNetworkInterface(ctx context.Context, msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) { +func (d *Daemon) handleDetachNetworkInterface(ctx context.Context, msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) string { slog.InfoContext(ctx, "Detaching ENI from instance", "instanceId", command.ID) if command.DetachENIData == nil || command.DetachENIData.AttachmentID == "" { - respondWithError(msg, awserrors.ErrorInvalidParameterValue) - return + return respondErrorOutcome(msg, awserrors.ErrorInvalidParameterValue) } attachmentID := command.DetachENIData.AttachmentID @@ -117,17 +111,14 @@ func (d *Daemon) handleDetachNetworkInterface(ctx context.Context, msg *nats.Msg record, err := d.vpcService.FindENIByAttachment(accountID, attachmentID) if err != nil { - respondWithServiceError(msg, err) - return + return respondServiceErrorOutcome(msg, err) } if record.InstanceId != instance.ID { - respondWithError(msg, awserrors.ErrorInvalidAttachmentIDNotFound) - return + return respondErrorOutcome(msg, awserrors.ErrorInvalidAttachmentIDNotFound) } if status := d.vmMgr.Status(instance); status != vm.StateRunning { - respondWithError(msg, awserrors.ErrorIncorrectInstanceState) - return + return respondErrorOutcome(msg, awserrors.ErrorIncorrectInstanceState) } if err := d.vpcService.UpdateENI(accountID, record.NetworkInterfaceId, func(r *handlers_ec2_vpc.ENIRecord) { @@ -146,8 +137,7 @@ func (d *Daemon) handleDetachNetworkInterface(ctx context.Context, msg *nats.Msg _ = d.vpcService.UpdateENI(accountID, record.NetworkInterfaceId, func(r *handlers_ec2_vpc.ENIRecord) { r.DetachInFlight = false }) - respondWithError(msg, eniHotplugErrorCode(err)) - return + return respondErrorOutcome(msg, eniHotplugErrorCode(err)) } if err := d.vpcService.DetachENI(ctx, accountID, record.NetworkInterfaceId); err != nil { @@ -169,6 +159,7 @@ func (d *Daemon) handleDetachNetworkInterface(ctx context.Context, msg *nats.Msg }) respondWithJSON(msg, ec2.DetachNetworkInterfaceOutput{}) + return outcomeSuccess } // eniHotplugErrorCode maps a vm.Manager hot-plug error to the AWS API diff --git a/spinifex/daemon/daemon_handlers_iam_profile.go b/spinifex/daemon/daemon_handlers_iam_profile.go index 126fda883..1f4e34143 100644 --- a/spinifex/daemon/daemon_handlers_iam_profile.go +++ b/spinifex/daemon/daemon_handlers_iam_profile.go @@ -11,11 +11,11 @@ import ( // handleAssociateIamInstanceProfile services the per-instance Associate path. // Gateway pre-validates the profile reference and iam:PassRole; ownership is // checked by checkInstanceOwnership before dispatch. -func (d *Daemon) handleAssociateIamInstanceProfile(ctx context.Context, msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) { +func (d *Daemon) handleAssociateIamInstanceProfile(ctx context.Context, msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) string { result, err := d.instanceService.AssociateIamInstanceProfile(ctx, instance, command) if err != nil { - respondWithServiceError(msg, err) - return + return respondServiceErrorOutcome(msg, err) } respondWithJSON(msg, result) + return outcomeSuccess } diff --git a/spinifex/daemon/daemon_handlers_instance.go b/spinifex/daemon/daemon_handlers_instance.go index 62b593c66..ce999006d 100644 --- a/spinifex/daemon/daemon_handlers_instance.go +++ b/spinifex/daemon/daemon_handlers_instance.go @@ -31,12 +31,11 @@ var startStoppedForwardTimeout = 30 * time.Second // instance: central store first, then the record under the manager lock, so a // failed S3 write leaves both stores untouched, matching the stopped path. // Ownership is checked by checkInstanceOwnership before dispatch. -func (d *Daemon) handleSetInstanceTags(ctx context.Context, msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) { +func (d *Daemon) handleSetInstanceTags(ctx context.Context, msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) string { remove := command.Attributes.RemoveInstanceTags data := command.InstanceTagsData if data == nil || (!remove && len(data.Tags) == 0) { - respondWithError(msg, awserrors.ErrorMissingParameter) - return + return respondErrorOutcome(msg, awserrors.ErrorMissingParameter) } var newTags []*ec2.Tag @@ -49,16 +48,14 @@ func (d *Daemon) handleSetInstanceTags(ctx context.Context, msg *nats.Msg, comma newTags = handlers_ec2_instance.ApplyInstanceTagMutation(v.Instance.Tags, data, remove) }) if missingRecord { - respondWithError(msg, awserrors.ErrorServerInternal) - return + return respondErrorOutcome(msg, awserrors.ErrorServerInternal) } accountID := utils.AccountIDFromMsg(msg) if err := d.tagsService.PutResourceTags(ctx, accountID, instance.ID, handlers_ec2_instance.TagsToMap(newTags)); err != nil { slog.ErrorContext(ctx, "SetInstanceTags: central tag store write failed", "instanceId", instance.ID, "err", err) - respondWithError(msg, awserrors.ErrorServerInternal) - return + return respondErrorOutcome(msg, awserrors.ErrorServerInternal) } found, err := d.vmMgr.UpdateAndPersist(instance.ID, func(v *vm.VM) bool { @@ -69,17 +66,16 @@ func (d *Daemon) handleSetInstanceTags(ctx context.Context, msg *nats.Msg, comma return true }) if err != nil { - respondWithServiceError(msg, err) - return + return respondServiceErrorOutcome(msg, err) } if !found { - respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound) - return + return respondErrorOutcome(msg, awserrors.ErrorInvalidInstanceIDNotFound) } 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 diff --git a/spinifex/daemon/daemon_handlers_spotinstance.go b/spinifex/daemon/daemon_handlers_spotinstance.go index e23366e20..b98c8ea8b 100644 --- a/spinifex/daemon/daemon_handlers_spotinstance.go +++ b/spinifex/daemon/daemon_handlers_spotinstance.go @@ -15,10 +15,9 @@ import ( // SpotInstanceRequestId) onto a launched VM. Dispatched from handleEC2Events, // which already verified ownership; the SIR id is the only datum on the wire as // the lifecycle is always spot for this command. -func (d *Daemon) handleSetSpotLineage(ctx context.Context, msg *nats.Msg, command types.EC2InstanceCommand) { +func (d *Daemon) handleSetSpotLineage(ctx context.Context, msg *nats.Msg, command types.EC2InstanceCommand) string { if command.SpotLineageData == nil { - respondWithError(msg, awserrors.ErrorMissingParameter) - return + return respondErrorOutcome(msg, awserrors.ErrorMissingParameter) } found, err := d.vmMgr.UpdateAndPersist(command.ID, func(v *vm.VM) bool { @@ -27,15 +26,14 @@ func (d *Daemon) handleSetSpotLineage(ctx context.Context, msg *nats.Msg, comman return true }) if err != nil { - respondWithServiceError(msg, err) - return + return respondServiceErrorOutcome(msg, err) } if !found { - respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound) - return + return respondErrorOutcome(msg, awserrors.ErrorInvalidInstanceIDNotFound) } if err := msg.Respond([]byte(`{}`)); err != nil { slog.ErrorContext(ctx, "Failed to respond to NATS request", "err", err) } + return outcomeSuccess } diff --git a/spinifex/daemon/daemon_handlers_volume.go b/spinifex/daemon/daemon_handlers_volume.go index 2869f28ea..060e03279 100644 --- a/spinifex/daemon/daemon_handlers_volume.go +++ b/spinifex/daemon/daemon_handlers_volume.go @@ -21,13 +21,12 @@ import ( // QMP/state-machine pipeline to vm.Manager.AttachVolume. The manager owns // every QMP and persistence side-effect; the daemon only emits the AWS API // response. -func (d *Daemon) handleAttachVolume(ctx context.Context, msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) { +func (d *Daemon) handleAttachVolume(ctx context.Context, msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) string { slog.InfoContext(ctx, "Attaching volume to instance", "instanceId", command.ID) if command.AttachVolumeData == nil || command.AttachVolumeData.VolumeID == "" { slog.ErrorContext(ctx, "AttachVolume: missing attach volume data") - respondWithError(msg, awserrors.ErrorInvalidParameterValue) - return + return respondErrorOutcome(msg, awserrors.ErrorInvalidParameterValue) } volumeID := command.AttachVolumeData.VolumeID @@ -38,15 +37,13 @@ func (d *Daemon) handleAttachVolume(ctx context.Context, msg *nats.Msg, command if status != vm.StateRunning { slog.ErrorContext(ctx, "AttachVolume: instance not running", "instanceId", command.ID, "status", status) - respondWithError(msg, awserrors.ErrorIncorrectInstanceState) - return + return respondErrorOutcome(msg, awserrors.ErrorIncorrectInstanceState) } volMeta, err := d.volumeService.GetVolumeMetadata(volumeID) if err != nil { slog.ErrorContext(ctx, "AttachVolume: failed to get volume metadata", "volumeId", volumeID, "err", err) - respondWithError(msg, awserrors.ErrorInvalidVolumeNotFound) - return + return respondErrorOutcome(msg, awserrors.ErrorInvalidVolumeNotFound) } callerAccountID := utils.AccountIDFromMsg(msg) @@ -55,8 +52,7 @@ func (d *Daemon) handleAttachVolume(ctx context.Context, msg *nats.Msg, command "volumeId", volumeID, "callerAccount", callerAccountID, "ownerAccount", volMeta.TenantID) - respondWithError(msg, awserrors.ErrorInvalidVolumeNotFound) - return + return respondErrorOutcome(msg, awserrors.ErrorInvalidVolumeNotFound) } if volMeta.State != "available" { @@ -69,8 +65,7 @@ func (d *Daemon) handleAttachVolume(ctx context.Context, msg *nats.Msg, command "attachedDevice", volMeta.DeviceName) // AWS returns VolumeInUse (not InvalidParameterValue) when a // re-attach targets a device other than the one already in use. - respondWithError(msg, awserrors.ErrorVolumeInUse) - return + return respondErrorOutcome(msg, awserrors.ErrorVolumeInUse) } // Volume is already attached to this instance (e.g. a CSI @@ -79,13 +74,12 @@ func (d *Daemon) handleAttachVolume(ctx context.Context, msg *nats.Msg, command slog.InfoContext(ctx, "AttachVolume: volume already attached to requesting instance, returning idempotent success", "volumeId", volumeID, "instanceId", command.ID, "device", volMeta.DeviceName) d.respondWithVolumeAttachment(msg, volumeID, command.ID, volMeta.DeviceName, "attached") - return + return outcomeSuccess } slog.ErrorContext(ctx, "AttachVolume: volume not available", "volumeId", volumeID, "state", volMeta.State) - respondWithError(msg, awserrors.ErrorVolumeInUse) - return + return respondErrorOutcome(msg, awserrors.ErrorVolumeInUse) } if volMeta.AvailabilityZone != "" && d.config.AZ != "" && @@ -94,14 +88,12 @@ func (d *Daemon) handleAttachVolume(ctx context.Context, msg *nats.Msg, command "volumeId", volumeID, "volumeAZ", volMeta.AvailabilityZone, "instanceAZ", d.config.AZ) - respondWithError(msg, awserrors.ErrorInvalidVolumeZoneMismatch) - return + return respondErrorOutcome(msg, awserrors.ErrorInvalidVolumeZoneMismatch) } device, err := d.vmMgr.AttachVolume(ctx, instance.ID, volumeID, command.AttachVolumeData.Device) if err != nil { - respondWithError(msg, attachDetachErrorCode(err)) - return + return respondErrorOutcome(msg, attachDetachErrorCode(err)) } // AttachVolume returns the API-form device name (/dev/sd[f-p]), not @@ -112,17 +104,17 @@ func (d *Daemon) handleAttachVolume(ctx context.Context, msg *nats.Msg, command // This diverges intentionally from BlockDeviceMappings, which retain // the guest path for in-guest device discovery. d.respondWithVolumeAttachment(msg, volumeID, command.ID, device, "attached") + return outcomeSuccess } // handleDetachVolume dispatches the QMP/state-machine pipeline to // vm.Manager.DetachVolume and emits the AWS API response. -func (d *Daemon) handleDetachVolume(ctx context.Context, msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) { +func (d *Daemon) handleDetachVolume(ctx context.Context, msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) string { slog.InfoContext(ctx, "Detaching volume from instance", "instanceId", command.ID) if command.DetachVolumeData == nil || command.DetachVolumeData.VolumeID == "" { slog.ErrorContext(ctx, "DetachVolume: missing detach volume data") - respondWithError(msg, awserrors.ErrorInvalidParameterValue) - return + return respondErrorOutcome(msg, awserrors.ErrorInvalidParameterValue) } deviceName, err := d.vmMgr.DetachVolume( @@ -133,11 +125,11 @@ func (d *Daemon) handleDetachVolume(ctx context.Context, msg *nats.Msg, command command.DetachVolumeData.Force, ) if err != nil { - respondWithError(msg, attachDetachErrorCode(err)) - return + return respondErrorOutcome(msg, attachDetachErrorCode(err)) } d.respondWithVolumeAttachment(msg, command.DetachVolumeData.VolumeID, command.ID, deviceName, "detaching") + return outcomeSuccess } // handleDrainVolume flushes the volume's in-flight writes to S3 by dialing the @@ -147,15 +139,14 @@ func (d *Daemon) handleDetachVolume(ctx context.Context, msg *nats.Msg, command // // Runs on its own goroutine (see handleEC2Events) so a long flush cannot hold // the instance's command subscription. -func (d *Daemon) handleDrainVolume(ctx context.Context, msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) { +func (d *Daemon) handleDrainVolume(ctx context.Context, msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) string { ctx, span := startOpSpan(ctx, "ec2.DrainVolume", command.ID) var err error defer func() { endOpSpan(span, err) }() if command.DrainVolumeData == nil || command.DrainVolumeData.VolumeID == "" { slog.ErrorContext(ctx, "DrainVolume: missing drain volume data", "instanceId", command.ID) - respondWithError(msg, awserrors.ErrorInvalidParameterValue) - return + return respondErrorOutcome(msg, awserrors.ErrorInvalidParameterValue) } volumeID := command.DrainVolumeData.VolumeID @@ -168,7 +159,7 @@ func (d *Daemon) handleDrainVolume(ctx context.Context, msg *nats.Msg, command t slog.InfoContext(ctx, "DrainVolume: instance teardown is complete, nothing to drain", "volumeId", volumeID, "instanceId", command.ID, "status", status) respondWithJSON(msg, types.DrainVolumeResponse{VolumeID: volumeID, Status: types.DrainVolumeStatusNotRunning}) - return + return outcomeSuccess } slog.InfoContext(ctx, "Draining volume before snapshot", "volumeId", volumeID, "instanceId", command.ID) @@ -178,11 +169,11 @@ func (d *Daemon) handleDrainVolume(ctx context.Context, msg *nats.Msg, command t // read a stale checkpoint. if err = handlers_ec2_snapshot.DrainVolumeSocket(d.config.DataDir, volumeID); err != nil { slog.ErrorContext(ctx, "DrainVolume: drain failed", "volumeId", volumeID, "instanceId", command.ID, "err", err) - respondWithError(msg, awserrors.ErrorServerInternal) - return + return respondErrorOutcome(msg, awserrors.ErrorServerInternal) } respondWithJSON(msg, types.DrainVolumeResponse{VolumeID: volumeID, Status: types.DrainVolumeStatusDrained}) + return outcomeSuccess } // attachDetachErrorCode maps a vm.Manager error returned by AttachVolume diff --git a/spinifex/daemon/nats_classification_test.go b/spinifex/daemon/nats_classification_test.go new file mode 100644 index 000000000..3b1ae89b2 --- /dev/null +++ b/spinifex/daemon/nats_classification_test.go @@ -0,0 +1,204 @@ +package daemon + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "log/slog" + "net/http" + "testing" + "time" + + "github.com/mulgadc/spinifex/spinifex/awserrors" + "github.com/mulgadc/spinifex/spinifex/otelsetup" + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/mulgadc/spinifex/spinifex/types" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// A daemon error rate is only useful if it counts daemon faults. An unknown +// instance id or a volume already in use is the caller's doing and must not +// land in the same series as a server fault. +func TestOutcomeForErrorSeparatesCallerFromDaemon(t *testing.T) { + tests := []struct { + name string + err error + want string + }{ + {"unknown instance", errors.New(awserrors.ErrorInvalidInstanceIDNotFound), outcomeClientError}, + {"volume in use", errors.New(awserrors.ErrorVolumeInUse), outcomeClientError}, + {"bad parameter", errors.New(awserrors.ErrorInvalidParameterValue), outcomeClientError}, + {"server fault", errors.New(awserrors.ErrorServerInternal), outcomeError}, + {"no recognised code", errors.New("the disk fell over"), outcomeError}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.want, outcomeForError(tc.err)) + }) + } +} + +// The NATS and HTTP paths must give the same word to the same failure, or a +// dashboard that spans both counts one service's client errors as faults. +func TestOutcomeForErrorAgreesWithTheHTTPMiddleware(t *testing.T) { + for _, code := range []string{ + awserrors.ErrorInvalidInstanceIDNotFound, + awserrors.ErrorVolumeInUse, + awserrors.ErrorServerInternal, + } { + status := awserrors.HTTPStatusForError(errors.New(code)) + assert.Equal(t, otelsetup.OutcomeForStatus(status), outcomeForError(errors.New(code)), + "%s (%d) must classify the same on both paths", code, status) + } +} + +// An error carrying no AWS code counts against the daemon. Defaulting the other +// way would let an unclassifiable failure disappear into the caller's column. +func TestHTTPStatusForErrorDefaultsToServerFault(t *testing.T) { + assert.Equal(t, http.StatusInternalServerError, + awserrors.HTTPStatusForError(errors.New("no code in here"))) +} + +// levelsLogged runs fn with a capturing logger and returns the levels it wrote. +func levelsLogged(t *testing.T, fn func()) []slog.Level { + t.Helper() + var buf bytes.Buffer + prev := slog.Default() + slog.SetDefault(slog.New(slog.NewJSONHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug}))) + t.Cleanup(func() { slog.SetDefault(prev) }) + + fn() + + var levels []slog.Level + for line := range bytes.SplitSeq(bytes.TrimSpace(buf.Bytes()), []byte("\n")) { + if len(line) == 0 { + continue + } + var rec struct { + Level string `json:"level"` + } + require.NoError(t, json.Unmarshal(line, &rec)) + var lvl slog.Level + require.NoError(t, lvl.UnmarshalText([]byte(rec.Level))) + levels = append(levels, lvl) + } + return levels +} + +// DescribeInstanceAttribute fans out and only the node holding the instance can +// answer; every other node replies InvalidInstanceID.NotFound by design and the +// aggregator drops those. Logging them at ERROR wrote three meaningless lines +// per IMDS fetch on a four-node cluster. +func TestFanOutMissIsNotLoggedAsAnError(t *testing.T) { + levels := levelsLogged(t, func() { + logHandlerError(context.Background(), "service call failed", + "ec2.DescribeInstanceAttribute", errors.New(awserrors.ErrorInvalidInstanceIDNotFound)) + }) + require.Len(t, levels, 1) + assert.Equal(t, slog.LevelWarn, levels[0], + "an expected fan-out miss must not be an ERROR line on the nodes that never held the instance") +} + +// The demotion above must not swallow a real fault on the same subject. +func TestServerFaultIsStillLoggedAsAnError(t *testing.T) { + levels := levelsLogged(t, func() { + logHandlerError(context.Background(), "service call failed", + "ec2.DescribeInstanceAttribute", errors.New(awserrors.ErrorServerInternal)) + }) + require.Len(t, levels, 1) + assert.Equal(t, slog.LevelError, levels[0], "a daemon fault must stay at ERROR") +} + +// commandFor builds a per-instance command with one attribute set. +func commandFor(id string, set func(*types.EC2CommandAttributes)) types.EC2InstanceCommand { + command := types.EC2InstanceCommand{ID: id} + set(&command.Attributes) + return command +} + +// The ec2.cmd subject carries the instance id, so the metric action has to come +// from the command. Every command must name itself, or the lifecycle work most +// likely to fail lands in one anonymous bucket. +func TestEC2CommandNameCoversEveryCommand(t *testing.T) { + tests := []struct { + want string + set func(*types.EC2CommandAttributes) + }{ + {"AttachVolume", func(a *types.EC2CommandAttributes) { a.AttachVolume = true }}, + {"DetachVolume", func(a *types.EC2CommandAttributes) { a.DetachVolume = true }}, + {"DrainVolume", func(a *types.EC2CommandAttributes) { a.DrainVolume = true }}, + {"AttachNetworkInterface", func(a *types.EC2CommandAttributes) { a.AttachENI = true }}, + {"DetachNetworkInterface", func(a *types.EC2CommandAttributes) { a.DetachENI = true }}, + {"AssociateIamInstanceProfile", func(a *types.EC2CommandAttributes) { a.AssociateIamInstanceProfile = true }}, + {"SetSpotLineage", func(a *types.EC2CommandAttributes) { a.SetSpotLineage = true }}, + {"SetInstanceTags", func(a *types.EC2CommandAttributes) { a.SetInstanceTags = true }}, + {"RemoveInstanceTags", func(a *types.EC2CommandAttributes) { a.RemoveInstanceTags = 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 }}, + } + for _, tc := range tests { + t.Run(tc.want, func(t *testing.T) { + assert.Equal(t, tc.want, ec2CommandName(commandFor("i-123", tc.set))) + }) + } + + assert.Equal(t, "unknown", ec2CommandName(types.EC2InstanceCommand{ID: "i-123"}), + "a command nothing handles must still be named, not dropped") +} + +// The per-instance command path emitted no request metrics at all, so a failed +// stop was invisible in Kibana. It must now record under an action naming the +// command — and never under one naming the instance, which would be one series +// per instance. +func TestEC2CommandRecordsUnderTheCommandNotTheInstance(t *testing.T) { + installOutcomeReader(t) + daemon := createTestDaemon(t, sharedNATSURL) + + const instanceID = "i-not-on-this-node" + subject := "ec2.cmd." + instanceID + sub, err := daemon.natsConn.Subscribe(subject, daemon.handleEC2Events) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + + data, err := json.Marshal(commandFor(instanceID, + func(a *types.EC2CommandAttributes) { a.StopInstance = true })) + require.NoError(t, err) + _, err = daemon.natsConn.Request(subject, data, 5*time.Second) + require.NoError(t, err) + + assert.Contains(t, requestOutcomesFor(t, ec2CmdAction("StopInstance")), outcomeClientError, + "a stop naming an instance this node does not hold is the caller's error, and must be counted") + assert.Empty(t, requestOutcomesFor(t, subject), + "the instance id must not reach the metric dimension") +} + +// A drain runs off-thread so the command subscription is not held for the +// length of the flush. Its point therefore has to be recorded by the goroutine +// once the work finishes: recording at dispatch would call every drain an +// instant success. +func TestDrainVolumeRecordsItsPointAfterTheWork(t *testing.T) { + installOutcomeReader(t) + const instanceID, volumeID = "i-drain-metric", "vol-drain-metric" + daemon := drainTestDaemon(t, instanceID) + testutil.StartDrainSocket(t, daemon.config.DataDir, volumeID, "OK\n") + subscribeEC2Commands(t, daemon, instanceID) + + before := len(requestOutcomesFor(t, ec2CmdAction("DrainVolume"))) + reply := drainRequest(t, daemon, instanceID, volumeID) + + var ack types.DrainVolumeResponse + require.NoError(t, json.Unmarshal(reply.Data, &ack)) + require.Equal(t, types.DrainVolumeStatusDrained, ack.Status) + + // The goroutine records after it replies, so poll rather than assume the + // point has landed by the time the reply arrives. + require.Eventually(t, func() bool { + return len(requestOutcomesFor(t, ec2CmdAction("DrainVolume"))) > before + }, 5*time.Second, 20*time.Millisecond, "the drain goroutine must record its own point") + + assert.Contains(t, requestOutcomesFor(t, ec2CmdAction("DrainVolume")), outcomeSuccess) +} diff --git a/spinifex/daemon/nats_outcome_test.go b/spinifex/daemon/nats_outcome_test.go index 3fffb8b0b..ceac2ac5b 100644 --- a/spinifex/daemon/nats_outcome_test.go +++ b/spinifex/daemon/nats_outcome_test.go @@ -105,8 +105,9 @@ func TestNATSMetricsHandlerSkipsDeferredOutcome(t *testing.T) { } // End to end over the real transport: the generic request constructor must -// report error for a failing service call and success for a passing one, so a -// failed NATS request is distinguishable in Kibana without reading logs. +// report the failure's class for a failing service call and success for a +// passing one, so a failed NATS request is distinguishable in Kibana without +// reading logs, and a caller mistake does not read as a daemon fault. func TestHandleNATSRequestReportsOutcome(t *testing.T) { tests := []struct { name string @@ -114,7 +115,9 @@ func TestHandleNATSRequestReportsOutcome(t *testing.T) { svcErr error want string }{ - {"service error", "test.outcome.err", errors.New(awserrors.ErrorVolumeInUse), outcomeError}, + {"caller error", "test.outcome.err", errors.New(awserrors.ErrorVolumeInUse), outcomeClientError}, + {"server fault", "test.outcome.fault", errors.New(awserrors.ErrorServerInternal), outcomeError}, + {"unrecognised failure", "test.outcome.opaque", errors.New("disk fell over"), outcomeError}, {"service success", "test.outcome.ok", nil, outcomeSuccess}, } @@ -147,8 +150,9 @@ func TestHandleNATSRequestReportsOutcome(t *testing.T) { } } -// A malformed payload never reaches the service, and that rejection is an -// error outcome too — otherwise a client sending garbage looks like success. +// A malformed payload never reaches the service. It is still not a success, +// but it is the caller's mistake, so it records as a client error rather than +// counting against the daemon. func TestHandleNATSRequestReportsOutcomeForBadPayload(t *testing.T) { installOutcomeReader(t) const subject = "test.outcome.malformed" @@ -169,7 +173,7 @@ func TestHandleNATSRequestReportsOutcomeForBadPayload(t *testing.T) { _, err = nc.Request(subject, []byte("{not json"), 5*time.Second) require.NoError(t, err) - assert.Equal(t, []string{outcomeError}, requestOutcomesFor(t, subject)) + assert.Equal(t, []string{outcomeClientError}, requestOutcomesFor(t, subject)) } // asMsgHandler drops the outcome a handler reports, for tests that subscribe a diff --git a/spinifex/otelsetup/http.go b/spinifex/otelsetup/http.go index 08afa3ac0..3ad773bf9 100644 --- a/spinifex/otelsetup/http.go +++ b/spinifex/otelsetup/http.go @@ -46,11 +46,12 @@ func (w *statusRecorder) Flush() { _ = http.NewResponseController(w.ResponseWriter).Flush() } -// outcomeForStatus classifies a response for the outcome dimension. 4xx is +// OutcomeForStatus classifies a response for the outcome dimension. 4xx is // client_error rather than error so a service's error rate stays server-fault // only, while client rejections (an IMDS 401, an unresolvable ENI's 404) stop -// being counted as successes. -func outcomeForStatus(status int) string { +// being counted as successes. Exported because the NATS path classifies against +// the same statuses and the two must agree. +func OutcomeForStatus(status int) string { switch { case status >= http.StatusInternalServerError: return "error" @@ -74,7 +75,7 @@ func HTTPMiddleware(serverName string) func(http.Handler) http.Handler { start := time.Now() rec := &statusRecorder{ResponseWriter: w, status: http.StatusOK} next.ServeHTTP(rec, r) - RecordRequest(r.Context(), r.Method+" /health", outcomeForStatus(rec.status), time.Since(start)) + RecordRequest(r.Context(), r.Method+" /health", OutcomeForStatus(rec.status), time.Since(start)) return } ctx := otel.GetTextMapPropagator().Extract(r.Context(), propagation.HeaderCarrier(r.Header)) @@ -97,7 +98,7 @@ func HTTPMiddleware(serverName string) func(http.Handler) http.Handler { if rec.status >= http.StatusInternalServerError { span.SetStatus(codes.Error, fmt.Sprintf("HTTP %d", rec.status)) } - RecordRequest(ctx, action.name, outcomeForStatus(rec.status), time.Since(start)) + RecordRequest(ctx, action.name, OutcomeForStatus(rec.status), time.Since(start)) }) } } diff --git a/spinifex/otelsetup/http_test.go b/spinifex/otelsetup/http_test.go index b0fcaf489..ff83bccdf 100644 --- a/spinifex/otelsetup/http_test.go +++ b/spinifex/otelsetup/http_test.go @@ -148,8 +148,8 @@ func TestOutcomeForStatusSeparatesClientErrors(t *testing.T) { } for _, tc := range tests { - if got := outcomeForStatus(tc.status); got != tc.want { - t.Errorf("outcomeForStatus(%d) = %q, want %q", tc.status, got, tc.want) + if got := OutcomeForStatus(tc.status); got != tc.want { + t.Errorf("OutcomeForStatus(%d) = %q, want %q", tc.status, got, tc.want) } } }