hanzoai/tasks
2
0
Fork 0
mirror of https://github.com/hanzoai/tasks.git synced 2026-10-04 02:04:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add request metrics for activity operator commands (#9661)

Browse Source 
## What changed?
Add instrumentation for the activity operator commands which previously
had no metrics. `activity_targeting_method` is `id` when targetting a
specific ID, else `type`. UpdateOptions also has a structured log line
for changed fields.

## Why?
Add visibility into the usage of the activity operator commands.

## How did you test it?
- [ ] built
- [ ] run locally and tested manually
- [ ] covered by existing tests
- [ ] added new unit test(s)
- [ ] added new functional test(s)

## Potential risks
Minimal, cardinality is low for these
This commit is contained in:
Sean Kane
2026-03-25 15:31:15 -06:00
committed by GitHub
parent 9c8ea108b3
commit b864c836f3
6 changed files with 99 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/metrics/metric_defs.go
+6 -2
View File
@@ -1109,8 +1109,12 @@ var (
ExecutionQueueSchedulerTaskLatency = NewTimerDef("execution_queue_scheduler_task_latency")
ExecutionQueueSchedulerQueueWaitTime = NewTimerDef("execution_queue_scheduler_queue_wait_time")
PausedActivitiesCounter = NewCounterDef("paused_activities")
ExternalPayloadUploadSize = NewBytesHistogramDef("external_payload_upload_size", WithDescription("The histogram of sizes in bytes of uploaded external payloads."))
PausedActivitiesCounter = NewCounterDef("paused_activities")
ActivityPauseRequests = NewCounterDef("activity_pause_requests")
ActivityUnpauseRequests = NewCounterDef("activity_unpause_requests")
ActivityResetRequests = NewCounterDef("activity_reset_requests")
ActivityUpdateOptionsRequests = NewCounterDef("activity_update_options_requests")
ExternalPayloadUploadSize = NewBytesHistogramDef("external_payload_upload_size", WithDescription("The histogram of sizes in bytes of uploaded external payloads."))
// Deadlock detector metrics
DDSuspectedDeadlocks = NewCounterDef("dd_suspected_deadlocks")
common/metrics/tags.go
+6
View File
@@ -58,6 +58,7 @@ const (
toUnversioned = "to_unversioned"
queryTypeTag = "query_type"
namespaceAllValue = "all"
activityTargetingMethod = "activity_targeting_method"
unknownValue = "_unknown_"
totalMetricSuffix = "_total"
tagExcludedValue = "_tag_excluded_"
@@ -205,6 +206,11 @@ func ActivityTypeTag(value string) Tag {
return Tag{Key: activityType, Value: value}
}
// ActivityTargetingMethodTag returns a tag indicating how the activity was targeted: "id" or "type".
func ActivityTargetingMethodTag(value string) Tag {
return Tag{Key: activityTargetingMethod, Value: value}
}
// CommandTypeTag returns a new command type tag.
func CommandTypeTag(value string) Tag {
if len(value) == 0 {
service/history/api/pauseactivity/api.go
+14
View File
@@ -7,6 +7,8 @@ import (
"go.temporal.io/server/api/historyservice/v1"
persistencespb "go.temporal.io/server/api/persistence/v1"
"go.temporal.io/server/common/definition"
"go.temporal.io/server/common/metrics"
"go.temporal.io/server/common/namespace"
"go.temporal.io/server/service/history/api"
"go.temporal.io/server/service/history/consts"
historyi "go.temporal.io/server/service/history/interfaces"
@@ -20,6 +22,7 @@ func Invoke(
shardContext historyi.ShardContext,
workflowConsistencyChecker api.WorkflowConsistencyChecker,
) (resp *historyservice.PauseActivityResponse, retError error) {
err := api.GetAndUpdateWorkflowWithNew(
ctx,
nil,
@@ -78,5 +81,16 @@ func Invoke(
return nil, err
}
targetingMethod := "type"
if _, ok := request.GetFrontendRequest().GetActivity().(*workflowservice.PauseActivityRequest_Id); ok {
targetingMethod = "id"
}
if ns, err := shardContext.GetNamespaceRegistry().GetNamespaceByID(namespace.ID(request.NamespaceId)); err == nil {
metrics.ActivityPauseRequests.With(shardContext.GetMetricsHandler().WithTags(
metrics.NamespaceTag(ns.Name().String()),
metrics.ActivityTargetingMethodTag(targetingMethod),
)).Record(1)
}
return &historyservice.PauseActivityResponse{}, nil
}
service/history/api/resetactivity/api.go
+14
View File
@@ -6,6 +6,8 @@ import (
"go.temporal.io/api/workflowservice/v1"
"go.temporal.io/server/api/historyservice/v1"
"go.temporal.io/server/common/definition"
"go.temporal.io/server/common/metrics"
"go.temporal.io/server/common/namespace"
"go.temporal.io/server/service/history/api"
"go.temporal.io/server/service/history/consts"
historyi "go.temporal.io/server/service/history/interfaces"
@@ -19,6 +21,7 @@ func Invoke(
workflowConsistencyChecker api.WorkflowConsistencyChecker,
) (resp *historyservice.ResetActivityResponse, retError error) {
request := req.GetFrontendRequest()
workflowKey := definition.NewWorkflowKey(
req.NamespaceId,
request.GetExecution().GetWorkflowId(),
@@ -72,5 +75,16 @@ func Invoke(
return nil, err
}
targetingMethod := "type"
if _, ok := req.GetFrontendRequest().GetActivity().(*workflowservice.ResetActivityRequest_Id); ok {
targetingMethod = "id"
}
if ns, err := shardContext.GetNamespaceRegistry().GetNamespaceByID(namespace.ID(req.NamespaceId)); err == nil {
metrics.ActivityResetRequests.With(shardContext.GetMetricsHandler().WithTags(
metrics.NamespaceTag(ns.Name().String()),
metrics.ActivityTargetingMethodTag(targetingMethod),
)).Record(1)
}
return &historyservice.ResetActivityResponse{}, nil
}
service/history/api/unpauseactivity/api.go
+24
View File
@@ -6,6 +6,9 @@ import (
"go.temporal.io/api/workflowservice/v1"
"go.temporal.io/server/api/historyservice/v1"
"go.temporal.io/server/common/definition"
"go.temporal.io/server/common/log/tag"
"go.temporal.io/server/common/metrics"
"go.temporal.io/server/common/namespace"
"go.temporal.io/server/service/history/api"
"go.temporal.io/server/service/history/consts"
historyi "go.temporal.io/server/service/history/interfaces"
@@ -49,6 +52,27 @@ func Invoke(
return nil, err
}
frontendReq := request.GetFrontendRequest()
targetingMethod := "type"
if _, ok := frontendReq.GetActivity().(*workflowservice.UnpauseActivityRequest_Id); ok {
targetingMethod = "id"
}
if ns, err := shardContext.GetNamespaceRegistry().GetNamespaceByID(namespace.ID(request.NamespaceId)); err == nil {
metrics.ActivityUnpauseRequests.With(shardContext.GetMetricsHandler().WithTags(
metrics.NamespaceTag(ns.Name().String()),
metrics.ActivityTargetingMethodTag(targetingMethod),
)).Record(1)
}
shardContext.GetLogger().Info("unpauseactivity: activity unpaused",
tag.WorkflowNamespaceID(request.GetNamespaceId()),
tag.WorkflowID(frontendReq.GetExecution().GetWorkflowId()),
tag.WorkflowRunID(frontendReq.GetExecution().GetRunId()),
tag.NewBoolTag("reset_attempts", frontendReq.GetResetAttempts()),
tag.NewBoolTag("reset_heartbeat", frontendReq.GetResetHeartbeat()),
tag.NewDurationTag("jitter", frontendReq.GetJitter().AsDuration()),
)
return response, err
}
service/history/api/updateactivityoptions/api.go
+35
View File
@@ -2,6 +2,7 @@ package updateactivityoptions
import (
"context"
"strings"
activitypb "go.temporal.io/api/activity/v1"
commandpb "go.temporal.io/api/command/v1"
@@ -15,6 +16,8 @@ import (
persistencespb "go.temporal.io/server/api/persistence/v1"
"go.temporal.io/server/common"
"go.temporal.io/server/common/definition"
"go.temporal.io/server/common/log/tag"
"go.temporal.io/server/common/metrics"
"go.temporal.io/server/common/namespace"
"go.temporal.io/server/common/util"
"go.temporal.io/server/service/history/api"
@@ -81,6 +84,38 @@ func Invoke(
return nil, err
}
targetingMethod := "type"
if _, ok := updateRequest.GetActivity().(*workflowservice.UpdateActivityOptionsRequest_Id); ok {
targetingMethod = "id"
}
if ns, err := shardContext.GetNamespaceRegistry().GetNamespaceByID(namespace.ID(request.NamespaceId)); err == nil {
metrics.ActivityUpdateOptionsRequests.With(shardContext.GetMetricsHandler().WithTags(
metrics.NamespaceTag(ns.Name().String()),
metrics.ActivityTargetingMethodTag(targetingMethod),
)).Record(1)
}
logger := shardContext.GetLogger()
if updateRequest.RestoreOriginal {
logger.Info("updateactivityoptions: activity options restored to original",
tag.WorkflowNamespaceID(request.GetNamespaceId()),
tag.WorkflowID(updateRequest.GetExecution().GetWorkflowId()),
tag.WorkflowRunID(updateRequest.GetExecution().GetRunId()),
)
} else if mask := updateRequest.GetUpdateMask(); mask != nil {
updatedFields := util.ParseFieldMask(mask)
fields := make([]string, 0, len(updatedFields))
for f := range updatedFields {
fields = append(fields, f)
}
logger.Info("updateactivityoptions: activity options updated",
tag.WorkflowNamespaceID(request.GetNamespaceId()),
tag.WorkflowID(updateRequest.GetExecution().GetWorkflowId()),
tag.WorkflowRunID(updateRequest.GetExecution().GetRunId()),
tag.NewStringTag("updated_fields", strings.Join(fields, ",")),
)
}
return response, err
}