Populate resource_id fields for request header routing #2226
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -1961,6 +1961,7 @@ func (wth *workflowTaskHandlerImpl) completeWorkflow( | |
| BinaryChecksum: wth.workerBuildID, | ||
| QueryResults: queryResults, | ||
| Namespace: wth.namespace, | ||
| ResourceId: fmt.Sprintf("workflow:%s", task.WorkflowExecution.WorkflowId), | ||
| MeteringMetadata: &commonpb.MeteringMetadata{NonfirstLocalActivityExecutionAttempts: nonfirstLAAttempts}, | ||
| SdkMetadata: &sdk.WorkflowTaskCompletedMetadata{ | ||
| LangUsedFlags: langUsedFlags, | ||
|
|
@@ -2300,7 +2301,8 @@ func (ath *activityTaskHandlerImpl) Execute(taskQueue string, t *workflowservice | |
| metricsHandler.Counter(metrics.UnregisteredActivityInvocationCounter).Inc(1) | ||
| return convertActivityResultToRespondRequest(ath.identity, t.TaskToken, nil, | ||
| NewActivityNotRegisteredError(activityType, ath.getRegisteredActivityNames()), | ||
| ath.dataConverter, ath.failureConverter, ath.namespace, false, ath.versionStamp, ath.deployment, ath.workerDeploymentOptions, | ||
| t.WorkflowExecution.GetWorkflowId(), t.ActivityId), nil | ||
| } | ||
|
|
||
| // panic handler | ||
|
|
@@ -2318,7 +2320,8 @@ func (ath *activityTaskHandlerImpl) Execute(taskQueue string, t *workflowservice | |
| metricsHandler.Counter(metrics.ActivityTaskErrorCounter).Inc(1) | ||
| panicErr := newPanicError(p, st) | ||
| result = convertActivityResultToRespondRequest(ath.identity, t.TaskToken, nil, panicErr, | ||
| ath.dataConverter, ath.failureConverter, ath.namespace, false, ath.versionStamp, ath.deployment, ath.workerDeploymentOptions, | ||
| t.WorkflowExecution.GetWorkflowId(), t.ActivityId) | ||
| } | ||
| }() | ||
|
|
||
|
|
@@ -2365,7 +2368,8 @@ func (ath *activityTaskHandlerImpl) Execute(taskQueue string, t *workflowservice | |
| ) | ||
| } | ||
| return convertActivityResultToRespondRequest(ath.identity, t.TaskToken, output, err, | ||
| ath.dataConverter, ath.failureConverter, ath.namespace, isActivityCanceled, ath.versionStamp, ath.deployment, ath.workerDeploymentOptions, | ||
| t.WorkflowExecution.GetWorkflowId(), t.ActivityId), nil | ||
| } | ||
|
|
||
| func (ath *activityTaskHandlerImpl) getActivity(name string) activity { | ||
|
|
@@ -2425,15 +2429,24 @@ func createNewCommandWithMetadata(commandType enumspb.CommandType, metadata *sdk | |
| } | ||
| } | ||
|
|
||
| func getActivityResourceIdFromCtx(ctx context.Context) string { | ||
| env := getActivityEnvironmentFromCtx(ctx) | ||
| if env == nil { | ||
| return "" | ||
|
Contributor
There was a problem hiding this comment. do we want test coverage for when env == nil? |
||
| } | ||
| return getActivityResourceId(env.workflowExecution.ID, env.activityID) | ||
| } | ||
|
|
||
| func recordActivityHeartbeat(ctx context.Context, service workflowservice.WorkflowServiceClient, metricsHandler metrics.Handler, | ||
| identity string, taskToken []byte, details *commonpb.Payloads, | ||
| ) error { | ||
| namespace := getNamespaceFromActivityCtx(ctx) | ||
| request := &workflowservice.RecordActivityTaskHeartbeatRequest{ | ||
| TaskToken: taskToken, | ||
| Details: details, | ||
| Identity: identity, | ||
| Namespace: namespace, | ||
| ResourceId: getActivityResourceIdFromCtx(ctx), | ||
| } | ||
|
Comment on lines
2461
to
2471
There was a problem hiding this comment. 🔴 Extended reasoning...Bug:
|
||
|
|
||
| var heartbeatResponse *workflowservice.RecordActivityTaskHeartbeatResponse | ||
|
|
@@ -2465,6 +2478,7 @@ func recordActivityHeartbeatByID(ctx context.Context, service workflowservice.Wo | |
| ActivityId: activityID, | ||
| Details: details, | ||
| Identity: identity, | ||
| ResourceId: getActivityResourceId(workflowID, activityID), | ||
| } | ||
|
|
||
| var heartbeatResponse *workflowservice.RecordActivityTaskHeartbeatByIdResponse | ||
|
|
||
Uh oh!
There was an error while loading. Please reload this page.