Github/temporal
1
0
Fork 0
mirror of https://github.com/temporalio/temporal.git synced 2026-08-30 18:41:49 -07:00
Code Issues Packages Projects Releases Wiki Activity

Tag Nexus completion request logs (#11684)

Browse Source 
## What changed

The completion (callback) handler's request-scoped logger carried only
the namespace, even though a richer one was built just above it.

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Stephan Behnke
2026-08-20 20:34:31 -07:00
committed by GitHub
parent 24434b11e9
commit 2f3cbadeec
1 changed files with 23 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
service/frontend/nexus_completion_http_handler.go
View File

@@ -40,7 +40,7 @@ import (
)
const nexusCompletionAPIName = configs.CompleteNexusOperation
const nexusCompletionMethodNameForMetrics = "CompleteNexusOperation"
const nexusCompletionMethodName = "CompleteNexusOperation"
type nexusCompletionHandler struct {
ClusterMetadata cluster.Metadata
@@ -157,18 +157,21 @@ func (h *nexusCompletionHandler) CompleteOperation(ctx context.Context, r *nexus
}
logger := log.With(
h.Logger,
tag.Operation(nexusCompletionMethodName),
tag.WorkflowNamespace(ns.Name().String()),
tag.WorkflowNamespaceID(targetNamespaceID),
tag.WorkflowID(targetBusinessID),
tag.WorkflowRunID(targetRunID),
tag.RequestID(completion.GetRequestId()),
)
rCtx := &requestContext{
nexusCompletionHandler: h,
namespace: ns,
businessID: targetBusinessID,
logger: log.With(h.Logger, tag.WorkflowNamespace(ns.Name().String())),
logger: logger,
metricsHandler: h.MetricsHandler.WithTags(metrics.NamespaceTag(ns.Name().String())),
metricsHandlerForInterceptors: h.MetricsHandler.WithTags(
metrics.OperationTag(nexusCompletionMethodNameForMetrics),
metrics.OperationTag(nexusCompletionMethodName),
metrics.NamespaceTag(ns.Name().String()),
),
requestStartTime: startTime,
@@ -183,16 +186,14 @@ func (h *nexusCompletionHandler) CompleteOperation(ctx context.Context, r *nexus
nsNameEscaped := commonnexus.RouteCompletionCallback.Deserialize(mux.Vars(r.HTTPRequest))
nsName, err := url.PathUnescape(nsNameEscaped)
if err != nil {
h.Logger.Error("failed to extract namespace from request", tag.Error(err))
logger.Error("failed to extract namespace from request", tag.Error(err))
h.preProcessErrorsCounter.Record(1)
return nexus.NewHandlerErrorf(nexus.HandlerErrorTypeBadRequest, "invalid URL")
}
if nsName != ns.Name().String() {
logger.Error(
"namespace ID in token doesn't match the token",
tag.WorkflowNamespaceID(ns.ID().String()),
tag.Error(err),
tag.String("completion-namespace-id", targetNamespaceID),
"namespace in callback URL doesn't match the completion token",
tag.String("url-namespace", nsName),
)
return nexus.NewHandlerErrorf(nexus.HandlerErrorTypeBadRequest, "invalid callback token")
}
@@ -210,7 +211,7 @@ func (h *nexusCompletionHandler) CompleteOperation(ctx context.Context, r *nexus
return nexus.NewHandlerErrorf(nexus.HandlerErrorTypeBadRequest, "operation token length exceeds allowed limit (%d/%d)", len(r.OperationToken), tokenLimit)
}
links := commonnexus.ConvertNexusLinksToProtoLinks(r.Links, h.Logger)
links := commonnexus.ConvertNexusLinksToProtoLinks(r.Links, logger)
var successPayload *commonpb.Payload
switch r.State { // nolint:exhaustive
@@ -223,7 +224,7 @@ func (h *nexusCompletionHandler) CompleteOperation(ctx context.Context, r *nexus
return nexus.NewHandlerErrorf(nexus.HandlerErrorTypeBadRequest, "invalid result content")
}
if result.Size() > h.Config.BlobSizeLimitError(ns.Name().String()) {
logger.Error("payload size exceeds error limit for Nexus CompleteOperation request", tag.WorkflowNamespace(ns.Name().String()))
logger.Error("payload size exceeds error limit for Nexus CompleteOperation request")
return nexus.NewHandlerErrorf(nexus.HandlerErrorTypeBadRequest, "result exceeds size limit")
}
successPayload = result
@@ -403,26 +404,26 @@ func (h *nexusCompletionHandler) completeChasmOperation(
func (h *nexusCompletionHandler) forwardCompleteOperation(ctx context.Context, r *nexusrpc.CompletionRequest, rCtx *requestContext) error {
targetCluster := rCtx.namespace.ActiveClusterName(namespace.RoutingKey{ID: rCtx.businessID})
logger := log.With(
rCtx.logger,
tag.SourceCluster(h.ClusterMetadata.GetCurrentClusterName()),
tag.TargetCluster(targetCluster),
)
client, err := h.ForwardingClients.Get(targetCluster)
if err != nil {
h.Logger.Error("unable to get HTTP client for forward request", tag.Operation(nexusCompletionAPIName), tag.WorkflowNamespace(rCtx.namespace.Name().String()), tag.Error(err), tag.SourceCluster(h.ClusterMetadata.GetCurrentClusterName()), tag.TargetCluster(targetCluster))
logger.Error("unable to get HTTP client for forward request", tag.Error(err))
return nexus.NewHandlerErrorf(nexus.HandlerErrorTypeInternal, "internal error")
}
forwardURL, err := url.JoinPath(client.BaseURL(), commonnexus.RouteCompletionCallback.Path(rCtx.namespace.Name().String()))
if err != nil {
h.Logger.Error("failed to construct forwarding request URL", tag.Operation(nexusCompletionAPIName), tag.WorkflowNamespace(rCtx.namespace.Name().String()), tag.Error(err), tag.TargetCluster(targetCluster))
logger.Error("failed to construct forwarding request URL", tag.Error(err))
return nexus.NewHandlerErrorf(nexus.HandlerErrorTypeInternal, "internal error")
}
if h.HTTPTraceProvider != nil {
traceLogger := log.With(h.Logger,
tag.Operation(nexusCompletionAPIName),
tag.WorkflowNamespace(rCtx.namespace.Name().String()),
tag.AttemptStart(time.Now().UTC()),
tag.SourceCluster(h.ClusterMetadata.GetCurrentClusterName()),
tag.TargetCluster(targetCluster),
)
traceLogger := log.With(logger, tag.AttemptStart(time.Now().UTC()))
if trace := h.HTTPTraceProvider.NewForwardingTrace(traceLogger); trace != nil {
ctx = httptrace.WithClientTrace(ctx, trace)
}
@@ -509,7 +510,7 @@ func (c *requestContext) augmentContext(ctx context.Context, header http.Header)
ctx = interceptor.PopulateCallerInfo(
ctx,
func() string { return c.namespace.Name().String() },
func() string { return nexusCompletionMethodNameForMetrics },
func() string { return nexusCompletionMethodName },
)
if userAgent := header.Get(headerUserAgent); userAgent != "" {
// Preserve original strict behavior: only process if exactly one delimiter present.
@@ -624,7 +625,7 @@ func (c *requestContext) interceptRequest(ctx context.Context, request *nexusrpc
if c.namespace.ActiveClusterName(namespace.RoutingKey{ID: c.businessID}) != c.ClusterMetadata.GetCurrentClusterName() {
if c.shouldForwardRequest(ctx, request.HTTPRequest.Header, c.businessID) {
c.forwarded = true
handler, forwardStartTime := c.RedirectionInterceptor.BeforeCall(nexusCompletionMethodNameForMetrics)
handler, forwardStartTime := c.RedirectionInterceptor.BeforeCall(nexusCompletionMethodName)
c.cleanupFunctions = append(c.cleanupFunctions, func(retErr error) {
c.RedirectionInterceptor.AfterCall(handler, forwardStartTime, c.namespace.ActiveClusterName(namespace.RoutingKey{ID: c.businessID}), c.namespace.Name().String(), retErr)
})
@@ -641,7 +642,7 @@ func (c *requestContext) interceptRequest(ctx context.Context, request *nexusrpc
request,
"",
c.metricsHandlerForInterceptors,
[]tag.Tag{tag.Operation(nexusCompletionMethodNameForMetrics), tag.WorkflowNamespace(c.namespace.Name().String())},
[]tag.Tag{tag.Operation(nexusCompletionMethodName), tag.WorkflowNamespace(c.namespace.Name().String())},
retErr,
c.namespace.Name(),
)