mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-30 14:56:19 +00:00
Vladyslav SimonenkoandGitHub
681e5107c7
## What changed? Changes activity metrics ### Deprecated Metrics - **activity_e2e_latency** → Deprecated, as it measures individual attempts rather than true end-to-end latency, and replaced with clearer **activity_start_to_close_latency** ### New & Replacement Metrics - **activity_start_to_close_latency**: Measures the latency from activity start to close (per attempt). - **activity_schedule_to_close_latency**: Measures true end-to-end duration, including retries and backoff. - **activity_success**: Counts the number of succeeded activities. Aligned with the existing **workflow_success** counter. - **activity_fail**: Counts final failures for activities. Similar to **workflow_failed,** although it doesn’t include retries. - **activity_timeout**: Incremented on the final activity timeout (including ScheduleToStartTimeout), tagged by `timeout_type`. Aligned with the existing **workflow_timeout** counter. - **activity_task_fail**: Counts failures for activities including retries. Note that we don’t need to capture the number of retries, as this metric represents this number well. - **activity_task_timeout**: Incremented on the activity attempt timeout (including ScheduleToStartTimeout), tagged by `timeout_type`. - **activity_cancel:** Incremented when an activity is cancelled **** ## Why? ActivityE2ELatency is inaccurate, since it measures only individual attempts. The other metrics are not recorded yet. ## How did you test it? - [X] built - [X] run locally and tested manually - [X] covered by existing tests - [X] added new unit test(s) - [ ] added new functional test(s) Ran the server locally and checked the metrics in Prometheus: Test 1: Activity with 5 attempts (1s each, 1s gap, all time out) - activity_task_timeout = 5 - activity_timeout = 1 - activity_success = 0 - activity_fail = 0 (counted as timeout instead) - activity_task_fail = 0 (counted as timeout instead) - activity_schedule_to_close_latency_sum = 9s - activity_end_to_end_latency_sum = 1s (deprecated) - activity_start_to_close_latency_sum = 1s Test 2: Basic helloworld activity - activity_success = 1 Test 3: Activity with 5 attempts (1s each, 1s gap, all fail except for the last one) - activity_task_fail = 4 - activity_success = 1
131 lines
4.1 KiB
Go
131 lines
4.1 KiB
Go
package respondactivitytaskcanceled
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
|
|
"go.temporal.io/server/api/historyservice/v1"
|
|
"go.temporal.io/server/common"
|
|
"go.temporal.io/server/common/definition"
|
|
"go.temporal.io/server/common/metrics"
|
|
"go.temporal.io/server/common/namespace"
|
|
"go.temporal.io/server/common/tasktoken"
|
|
"go.temporal.io/server/service/history/api"
|
|
"go.temporal.io/server/service/history/consts"
|
|
historyi "go.temporal.io/server/service/history/interfaces"
|
|
"go.temporal.io/server/service/history/workflow"
|
|
)
|
|
|
|
func Invoke(
|
|
ctx context.Context,
|
|
req *historyservice.RespondActivityTaskCanceledRequest,
|
|
shard historyi.ShardContext,
|
|
workflowConsistencyChecker api.WorkflowConsistencyChecker,
|
|
) (resp *historyservice.RespondActivityTaskCanceledResponse, retError error) {
|
|
namespaceEntry, err := api.GetActiveNamespace(shard, namespace.ID(req.GetNamespaceId()))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
namespace := namespaceEntry.Name()
|
|
|
|
request := req.CancelRequest
|
|
tokenSerializer := tasktoken.NewSerializer()
|
|
token, err0 := tokenSerializer.Deserialize(request.TaskToken)
|
|
if err0 != nil {
|
|
return nil, consts.ErrDeserializingToken
|
|
}
|
|
if err := api.SetActivityTaskRunID(ctx, token, workflowConsistencyChecker); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
var attemptStartedTime time.Time
|
|
var firstScheduledTime time.Time
|
|
var taskQueue string
|
|
var workflowTypeName string
|
|
err = api.GetAndUpdateWorkflowWithNew(
|
|
ctx,
|
|
token.Clock,
|
|
definition.NewWorkflowKey(
|
|
token.NamespaceId,
|
|
token.WorkflowId,
|
|
token.RunId,
|
|
),
|
|
func(workflowLease api.WorkflowLease) (*api.UpdateWorkflowAction, error) {
|
|
mutableState := workflowLease.GetMutableState()
|
|
workflowTypeName = mutableState.GetWorkflowType().GetName()
|
|
if !mutableState.IsWorkflowExecutionRunning() {
|
|
return nil, consts.ErrWorkflowCompleted
|
|
}
|
|
|
|
scheduledEventID := token.GetScheduledEventId()
|
|
if scheduledEventID == common.EmptyEventID { // client call CompleteActivityById, so get scheduledEventID by activityID
|
|
scheduledEventID, err0 = api.GetActivityScheduledEventID(token.GetActivityId(), mutableState)
|
|
if err0 != nil {
|
|
return nil, err0
|
|
}
|
|
}
|
|
ai, isRunning := mutableState.GetActivityInfo(scheduledEventID)
|
|
|
|
// First check to see if cache needs to be refreshed as we could potentially have stale workflow execution in
|
|
// some extreme cassandra failure cases.
|
|
if !isRunning && scheduledEventID >= mutableState.GetNextEventID() {
|
|
metrics.StaleMutableStateCounter.With(shard.GetMetricsHandler()).Record(
|
|
1,
|
|
metrics.OperationTag(metrics.HistoryRespondActivityTaskCanceledScope))
|
|
return nil, consts.ErrStaleState
|
|
}
|
|
|
|
if !isRunning ||
|
|
ai.StartedEventId == common.EmptyEventID ||
|
|
(token.GetScheduledEventId() != common.EmptyEventID && token.Attempt != ai.Attempt) ||
|
|
(token.GetVersion() != common.EmptyVersion && token.Version != ai.Version) {
|
|
return nil, consts.ErrActivityTaskNotFound
|
|
}
|
|
|
|
// sanity check if activity is requested to be cancelled
|
|
if !ai.CancelRequested {
|
|
return nil, consts.ErrActivityTaskNotCancelRequested
|
|
}
|
|
|
|
if _, err := mutableState.AddActivityTaskCanceledEvent(
|
|
scheduledEventID,
|
|
ai.StartedEventId,
|
|
ai.CancelRequestId,
|
|
request.Details,
|
|
request.Identity); err != nil {
|
|
// Unable to add ActivityTaskCanceled event to history
|
|
return nil, err
|
|
}
|
|
|
|
attemptStartedTime = ai.StartedTime.AsTime()
|
|
firstScheduledTime = ai.FirstScheduledTime.AsTime()
|
|
taskQueue = ai.TaskQueue
|
|
return &api.UpdateWorkflowAction{
|
|
Noop: false,
|
|
CreateWorkflowTask: true,
|
|
}, nil
|
|
},
|
|
nil,
|
|
shard,
|
|
workflowConsistencyChecker,
|
|
)
|
|
|
|
if err == nil {
|
|
workflow.RecordActivityCompletionMetrics(
|
|
shard,
|
|
namespace,
|
|
taskQueue,
|
|
workflow.ActivityCompletionMetrics{
|
|
Status: workflow.ActivityStatusCanceled,
|
|
AttemptStartedTime: attemptStartedTime,
|
|
FirstScheduledTime: firstScheduledTime,
|
|
Closed: true,
|
|
},
|
|
metrics.OperationTag(metrics.HistoryRespondActivityTaskCanceledScope),
|
|
metrics.WorkflowTypeTag(workflowTypeName),
|
|
metrics.ActivityTypeTag(token.ActivityType),
|
|
)
|
|
}
|
|
return &historyservice.RespondActivityTaskCanceledResponse{}, err
|
|
}
|