Add metrics to measture the time a task waiting in history queue

common/metrics/defs.go

@@ -2234,6 +2234,7 @@ const (
  TaskBatchCompleteFailure
  TaskProcessingLatency
  TaskQueueLatency
+ ScheduleToStartHistoryQueueLatencyPerTaskList
 
  TaskRequestsPerDomain
  TaskLatencyPerDomain
@@ -2901,6 +2902,7 @@ var MetricDefs = map[ServiceIdx]map[int]metricDefinition{
  TaskLimitExceededCounter: {metricName: "task_errors_limit_exceeded_counter", metricType: Counter},
  TaskProcessingLatency: {metricName: "task_latency_processing", metricType: Timer},
  TaskQueueLatency: {metricName: "task_latency_queue", metricType: Timer},
+ ScheduleToStartHistoryQueueLatencyPerTaskList: {metricName: "schedule_to_start_history_queue_latency_per_tl", metricType: Timer},
 
  // per domain task metrics
 

common/util.go

@@ -108,6 +108,7 @@ var (
  ErrContextTimeoutNotSet = &types.BadRequestError{Message: "Context timeout is not set."}
  // ErrDecisionResultCountTooLarge error for decision result count exceeds limit
  ErrDecisionResultCountTooLarge = &types.BadRequestError{Message: "Decision result count exceeds limit."}
+ stickyTaskListMetricTag = metrics.TaskListTag("__sticky__")
 )
 
 // AwaitWaitGroup calls Wait on the given wait
@@ -1038,3 +1039,25 @@ func IntersectionStringSlice(a, b []string) []string {
  }
  return result
 }
+
+// NewPerTaskListScope creates a tasklist metrics scope
+func NewPerTaskListScope(
+ domainName string,
+ taskListName string,
+ taskListKind types.TaskListKind,
+ client metrics.Client,
+ scopeIdx int,
+) metrics.Scope {
+ domainTag := metrics.DomainUnknownTag()
+ taskListTag := metrics.TaskListUnknownTag()
+ if domainName != "" {
+ domainTag = metrics.DomainTag(domainName)
+ }
+ if taskListName != "" && taskListKind != types.TaskListKindSticky {
+ taskListTag = metrics.TaskListTag(taskListName)
+ }
+ if taskListKind == types.TaskListKindSticky {
+ taskListTag = stickyTaskListMetricTag
+ }
+ return client.Scope(scopeIdx, domainTag, taskListTag)
+}

common/util_test.go

@@ -1633,3 +1633,7 @@ func TestSecondsToDuration(t *testing.T) {
  })
  }
 }
+
+func TestNewPerTaskListScope(t *testing.T) {
+ assert.NotNil(t, NewPerTaskListScope("test-domain", "test-tasklist", types.TaskListKindNormal, NewNoopMetricsClient(), 0))
+}

service/history/task/task.go

@@ -346,6 +346,7 @@ func (t *taskImpl) Ack() {
  t.scope.RecordTimer(metrics.TaskAttemptTimerPerDomain, time.Duration(t.attempt))
  t.scope.RecordTimer(metrics.TaskLatencyPerDomain, time.Since(t.submitTime))
  t.scope.RecordTimer(metrics.TaskQueueLatencyPerDomain, time.Since(t.GetVisibilityTimestamp()))
+
  }
 
  if t.eventLogger != nil && t.shouldProcessTask && t.attempt != 0 {

service/history/task/transfer_active_task_executor.go

@@ -189,6 +189,7 @@ func (t *transferActiveTaskExecutor) processActivityTask(
  return nil
  }
 
+ domainName := mutableState.GetDomainEntry().GetInfo().Name
  ai, ok := mutableState.GetActivityInfo(task.ScheduleID)
  if !ok {
  t.logger.Debug("Potentially duplicate ", tag.TaskID(task.TaskID), tag.WorkflowScheduleID(task.ScheduleID), tag.TaskType(persistence.TransferTaskTypeActivityTask))
@@ -209,7 +210,12 @@ func (t *transferActiveTaskExecutor) processActivityTask(
  return errWorkflowRateLimited
  }
 
- return t.pushActivity(ctx, task, timeout, mutableState.GetExecutionInfo().PartitionConfig)
+ err = t.pushActivity(ctx, task, timeout, mutableState.GetExecutionInfo().PartitionConfig)
+ if err == nil {
+ scope := common.NewPerTaskListScope(domainName, task.TaskList, types.TaskListKindNormal, t.metricsClient, metrics.TransferActiveTaskActivityScope)
+ scope.RecordTimer(metrics.ScheduleToStartHistoryQueueLatencyPerTaskList, time.Since(task.GetVisibilityTimestamp()))
+ }
+ return err
 }
 
 func (t *transferActiveTaskExecutor) processDecisionTask(
@@ -248,6 +254,7 @@ func (t *transferActiveTaskExecutor) processDecisionTask(
  return err
  }
 
+ domainName := mutableState.GetDomainEntry().GetInfo().Name
  executionInfo := mutableState.GetExecutionInfo()
  workflowTimeout := executionInfo.WorkflowTimeout
  decisionTimeout := common.MinInt32(workflowTimeout, common.MaxTaskTimeout)
@@ -298,6 +305,14 @@ func (t *transferActiveTaskExecutor) processDecisionTask(
  // will re-create a new non-sticky task and reset sticky.
  err = t.pushDecision(ctx, task, taskList, decisionTimeout, mutableState.GetExecutionInfo().PartitionConfig)
  }
+ if err == nil {
+ tlKind := types.TaskListKindNormal
+ if taskList.Kind != nil {
+ tlKind = *taskList.Kind
+ }
+ scope := common.NewPerTaskListScope(domainName, taskList.Name, tlKind, t.metricsClient, metrics.TransferActiveTaskDecisionScope)
+ scope.RecordTimer(metrics.ScheduleToStartHistoryQueueLatencyPerTaskList, time.Since(task.GetVisibilityTimestamp()))
+ }
  return err
 }
 

service/matching/handler/context.go

@@ -24,11 +24,11 @@ import (
  "context"
  "sync"
 
+ "github.com/uber/cadence/common"
  "github.com/uber/cadence/common/log"
  "github.com/uber/cadence/common/log/tag"
  "github.com/uber/cadence/common/metrics"
  "github.com/uber/cadence/common/types"
- "github.com/uber/cadence/service/matching/tasklist"
 )
 
 type handlerContext struct {
@@ -47,7 +47,7 @@ func newHandlerContext(
 ) *handlerContext {
  return &handlerContext{
  Context: ctx,
- scope: tasklist.NewPerTaskListScope(domainName, taskList.GetName(), taskList.GetKind(), metricsClient, metricsScope).Tagged(metrics.GetContextTags(ctx)...),
+ scope: common.NewPerTaskListScope(domainName, taskList.GetName(), taskList.GetKind(), metricsClient, metricsScope).Tagged(metrics.GetContextTags(ctx)...),
  logger: logger.WithTags(tag.WorkflowDomainName(domainName), tag.WorkflowTaskListName(taskList.GetName())),
  }
 }

service/matching/tasklist/task_list_manager.go

@@ -63,7 +63,6 @@ var (
  persistenceOperationRetryPolicy = common.CreatePersistenceRetryPolicy()
  taskListActivityTypeTag = metrics.TaskListTypeTag("activity")
  taskListDecisionTypeTag = metrics.TaskListTypeTag("decision")
- stickyTaskListMetricTag = metrics.TaskListTag("__sticky__")
 )
 
 type (
@@ -154,7 +153,7 @@ func NewManager(
  taskListKind = &normalTaskListKind
  }
 
- scope := NewPerTaskListScope(domainName, taskList.GetName(), *taskListKind, metricsClient, metrics.MatchingTaskListMgrScope).
+ scope := common.NewPerTaskListScope(domainName, taskList.GetName(), *taskListKind, metricsClient, metrics.MatchingTaskListMgrScope).
  Tagged(getTaskListTypeTag(taskList.GetType()))
  db := newTaskListDB(taskManager, taskList.GetDomainID(), domainName, taskList.GetName(), taskList.GetType(), int(*taskListKind), logger)
 
@@ -730,27 +729,6 @@ func newTaskListConfig(id *Identifier, cfg *config.Config, domainName string) *c
  }
 }
 
-func NewPerTaskListScope(
- domainName string,
- taskListName string,
- taskListKind types.TaskListKind,
- client metrics.Client,
- scopeIdx int,
-) metrics.Scope {
- domainTag := metrics.DomainUnknownTag()
- taskListTag := metrics.TaskListUnknownTag()
- if domainName != "" {
- domainTag = metrics.DomainTag(domainName)
- }
- if taskListName != "" && taskListKind != types.TaskListKindSticky {
- taskListTag = metrics.TaskListTag(taskListName)
- }
- if taskListKind == types.TaskListKindSticky {
- taskListTag = stickyTaskListMetricTag
- }
- return client.Scope(scopeIdx, domainTag, taskListTag)
-}
-
 func IdentityFromContext(ctx context.Context) string {
  val, ok := ctx.Value(identityCtxKey{}).(string)
  if !ok {