Skip to content

Tag HSM outbound Nexus call failure logs #11662

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
stephanos wants to merge 8 commits into
base: main
Choose a base branch
from
+77 −38
Open

Tag HSM outbound Nexus call failure logs #11662

Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
9ccc060
Tag HSM outbound Nexus call failure logs
stephanos Aug 19, 2026
fcfc6fb
Trim outbound call log helper comments
stephanos Aug 19, 2026
ee35c79
Shorten outbound call log helper comments
stephanos Aug 19, 2026
350caa9
Drop outboundCallLogTags comment
stephanos Aug 19, 2026
9fd064d
Add missing target namespace ID tag
stephanos Aug 19, 2026
b7daae0
Return a bound logger instead of a tag slice
stephanos Aug 19, 2026
a8a34e1
Mirror CHASM's invocationTraceContext in the HSM executor
stephanos Aug 19, 2026
49f8854
Move logging helpers after the outcome-tag helpers
stephanos Aug 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
115 changes: 77 additions & 38 deletions components/nexusoperations/executors.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -270,6 +270,19 @@ func (e taskExecutor) executeInvocationTask(ctx context.Context, env hsm.Environ
Links: []nexus.Link{args.nexusLink},
}

traceCtx := invocationTraceContext{

@stephanos stephanos Aug 20, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This entire change mimics CHASM's

temporal/chasm/lib/nexusoperation/invocation.go

Line 44 in 612823d

type invocationTraceContext struct {

operationTag: "StartOperation",
namespaceName: ns.Name().String(),
targetNamespaceID: endpoint.GetEndpoint().GetSpec().GetTarget().GetWorker().GetNamespaceId(),
requestID: args.requestID,
operation: args.operation,
endpointName: args.endpointName,
workflowID: ref.WorkflowKey.WorkflowID,
runID: ref.WorkflowKey.RunID,
attemptStart: time.Now().UTC(),
attempt: task.Attempt,
}

var result *nexusrpc.ClientStartOperationResponse[*commonpb.Payload]
var callErr error
var startTime time.Time
Expand All @@ -291,18 +304,8 @@ func (e taskExecutor) executeInvocationTask(ctx context.Context, env hsm.Environ
}

if e.HTTPTraceProvider != nil {
traceLogger := log.With(e.Logger,
tag.Operation("StartOperation"),
tag.WorkflowNamespace(ns.Name().String()),
tag.RequestID(args.requestID),
tag.NexusOperation(args.operation),
tag.Endpoint(args.endpointName),
tag.WorkflowID(ref.WorkflowKey.WorkflowID),
tag.WorkflowRunID(ref.WorkflowKey.RunID),
tag.AttemptStart(time.Now().UTC()),
tag.Attempt(task.Attempt),
)
if trace := e.HTTPTraceProvider.NewTrace(task.Attempt, traceLogger); trace != nil {
traceLogger := log.With(e.Logger, traceCtx.tags()...)
if trace := e.HTTPTraceProvider.NewTrace(traceCtx.attempt, traceLogger); trace != nil {
callCtx = httptrace.WithClientTrace(callCtx, trace)
}
}
Expand Down Expand Up @@ -333,13 +336,7 @@ func (e taskExecutor) executeInvocationTask(ctx context.Context, env hsm.Environ
chasmnexus.OutboundRequestCounter.With(e.MetricsHandler).Record(1, namespaceTag, destTag, methodTag, outcomeTag, failureSourceTag)
chasmnexus.OutboundRequestLatency.With(e.MetricsHandler).Record(time.Since(startTime), namespaceTag, destTag, methodTag, outcomeTag, failureSourceTag)

if callErr != nil {
if failureSource == commonnexus.FailureSourceWorker || errors.As(callErr, new(*operationTimeoutBelowMinError)) {
e.Logger.Debug("Nexus StartOperation request failed", tag.Error(callErr))
} else {
e.Logger.Error("Nexus StartOperation request failed", tag.Error(callErr))
}
}
e.logCallFailure(traceCtx, callErr, failureSource)

err = e.saveResult(ctx, env, ref, result, callErr)

Expand Down Expand Up @@ -734,6 +731,19 @@ func (e taskExecutor) executeCancelationTask(ctx context.Context, env hsm.Enviro
// Set this value on the parent context so that our custom HTTP caller can mutate it since we cannot access response headers directly.
callCtx = context.WithValue(callCtx, commonnexus.FailureSourceContextKey, &atomic.Value{})

traceCtx := invocationTraceContext{
operationTag: "CancelOperation",
namespaceName: ns.Name().String(),
targetNamespaceID: endpoint.GetEndpoint().GetSpec().GetTarget().GetWorker().GetNamespaceId(),
requestID: args.requestID,
operation: args.operation,
endpointName: args.endpointName,
workflowID: ref.WorkflowKey.WorkflowID,
runID: ref.WorkflowKey.RunID,
attemptStart: time.Now().UTC(),
attempt: task.Attempt,
}

var callErr error
var startTime time.Time
if callTimeout < e.Config.MinRequestTimeout(ns.Name().String()) {
Expand All @@ -758,18 +768,8 @@ func (e taskExecutor) executeCancelationTask(ctx context.Context, env hsm.Enviro
}

if e.HTTPTraceProvider != nil {
traceLogger := log.With(e.Logger,
tag.Operation("CancelOperation"),
tag.WorkflowNamespace(ns.Name().String()),
tag.RequestID(args.requestID),
tag.NexusOperation(args.operation),
tag.Endpoint(args.endpointName),
tag.WorkflowID(ref.WorkflowKey.WorkflowID),
tag.WorkflowRunID(ref.WorkflowKey.RunID),
tag.AttemptStart(time.Now().UTC()),
tag.Attempt(task.Attempt),
)
if trace := e.HTTPTraceProvider.NewTrace(task.Attempt, traceLogger); trace != nil {
traceLogger := log.With(e.Logger, traceCtx.tags()...)
if trace := e.HTTPTraceProvider.NewTrace(traceCtx.attempt, traceLogger); trace != nil {
callCtx = httptrace.WithClientTrace(callCtx, trace)
}
}
Expand All @@ -791,13 +791,7 @@ func (e taskExecutor) executeCancelationTask(ctx context.Context, env hsm.Enviro
chasmnexus.OutboundRequestCounter.With(e.MetricsHandler).Record(1, namespaceTag, destTag, methodTag, statusCodeTag, failureSourceTag)
chasmnexus.OutboundRequestLatency.With(e.MetricsHandler).Record(time.Since(startTime), namespaceTag, destTag, methodTag, statusCodeTag, failureSourceTag)

if callErr != nil {
if failureSource == commonnexus.FailureSourceWorker || errors.As(callErr, new(*operationTimeoutBelowMinError)) {
e.Logger.Debug("Nexus CancelOperation request failed", tag.Error(callErr))
} else {
e.Logger.Error("Nexus CancelOperation request failed", tag.Error(callErr))
}
}
e.logCallFailure(traceCtx, callErr, failureSource)

err = e.saveCancelationResult(ctx, env, ref, callErr, args.scheduledEventID)

Expand Down Expand Up @@ -1025,6 +1019,51 @@ func cancelCallOutcomeTag(callCtx context.Context, callErr error) string {
return "successful"
}

// invocationTraceContext captures per-call contextual information used for HTTP tracing and failure logging.
type invocationTraceContext struct {
operationTag string // "StartOperation" or "CancelOperation"
namespaceName string // source (caller) namespace
targetNamespaceID string
requestID string
operation string
endpointName string
workflowID string
runID string
attemptStart time.Time
attempt int32
}

// tags returns the structured log tags describing the call.
func (c invocationTraceContext) tags() []tag.Tag {
return []tag.Tag{
tag.Operation(c.operationTag),
tag.WorkflowNamespace(c.namespaceName),
tag.NexusEndpointTargetNamespaceID(c.targetNamespaceID),

@stephanos stephanos Aug 20, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is a net new tag.

tag.RequestID(c.requestID),
tag.NexusOperation(c.operation),
tag.Endpoint(c.endpointName),
tag.WorkflowID(c.workflowID),
tag.WorkflowRunID(c.runID),
tag.AttemptStart(c.attemptStart),
tag.Attempt(c.attempt),
}
}
Comment thread
stephanos marked this conversation as resolved.

// logCallFailure logs a failed outbound Nexus call.
func (e taskExecutor) logCallFailure(traceCtx invocationTraceContext, callErr error, failureSource string) {
if callErr == nil {
return
}
tags := append(traceCtx.tags(), tag.Error(callErr))
msg := fmt.Sprintf("Nexus %s request failed", traceCtx.operationTag)
_, isTimeoutBelowMin := errors.AsType[*operationTimeoutBelowMinError](callErr)
if failureSource == commonnexus.FailureSourceWorker || isTimeoutBelowMin {
e.Logger.Debug(msg, tags...)
} else {
e.Logger.Error(msg, tags...)

@stephanos stephanos Aug 20, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is the main fix

}
}

func isDestinationDown(err error) bool {
var serviceErr serviceerror.ServiceError
// For the system endpoint, we don't even consider the destination down since it's internal.
Expand Down
Loading