Tracking cache invalidation for different types of workflow depends on their runtime length (#1277)

inject workflow execution timeout into getOrCreateWorkflowContext func and emit a new tag based on that timeout

Author: timl3136

internal/internal_logging_tags.go

@@ -21,25 +21,26 @@
 package internal
 
 const (
-	tagActivityID        = "ActivityID"
-	tagActivityType      = "ActivityType"
-	tagDomain            = "Domain"
-	tagEventID           = "EventID"
-	tagEventType         = "EventType"
-	tagRunID             = "RunID"
-	tagTaskList          = "TaskList"
-	tagTimerID           = "TimerID"
-	tagWorkflowID        = "WorkflowID"
-	tagWorkflowType      = "WorkflowType"
-	tagWorkerID          = "WorkerID"
-	tagWorkerType        = "WorkerType"
-	tagSideEffectID      = "SideEffectID"
-	tagChildWorkflowID   = "ChildWorkflowID"
-	tagLocalActivityType = "LocalActivityType"
-	tagLocalActivityID   = "LocalActivityID"
-	tagQueryType         = "QueryType"
-	tagVisibilityQuery   = "VisibilityQuery"
-	tagPanicError        = "PanicError"
-	tagPanicStack        = "PanicStack"
-	causeTag             = "pollerrorcause"
+	tagActivityID            = "ActivityID"
+	tagActivityType          = "ActivityType"
+	tagDomain                = "Domain"
+	tagEventID               = "EventID"
+	tagEventType             = "EventType"
+	tagRunID                 = "RunID"
+	tagTaskList              = "TaskList"
+	tagTimerID               = "TimerID"
+	tagWorkflowID            = "WorkflowID"
+	tagWorkflowType          = "WorkflowType"
+	tagWorkerID              = "WorkerID"
+	tagWorkerType            = "WorkerType"
+	tagSideEffectID          = "SideEffectID"
+	tagChildWorkflowID       = "ChildWorkflowID"
+	tagLocalActivityType     = "LocalActivityType"
+	tagLocalActivityID       = "LocalActivityID"
+	tagQueryType             = "QueryType"
+	tagVisibilityQuery       = "VisibilityQuery"
+	tagPanicError            = "PanicError"
+	tagPanicStack            = "PanicStack"
+	causeTag                 = "pollerrorcause"
+	tagworkflowruntimelength = "workflowruntimelength"
 )

internal/internal_task_handlers.go

@@ -51,6 +51,12 @@ const (
 	defaultStickyCacheSize = 10000
 
 	noRetryBackoff = time.Duration(-1)
+
+	defaultInstantLivedWorkflowTimeoutUpperLimitInSec = 1
+
+	defaultShortLivedWorkflowTimeoutUpperLimitInSec = 1 * 1800
+
+	defaultMediumLivedWorkflowTimeoutUpperLimitInSec = 8 * 3600
 )
 
 type (
@@ -677,6 +683,9 @@ func (wth *workflowTaskHandlerImpl) getOrCreateWorkflowContext(
 
 	if workflowContext != nil {
 		workflowContext.Lock()
+		// add new tag on metrics scope with workflow runtime length category
+		executionRuntimeType := workflowCategorizedByTimeout(workflowContext.workflowInfo.ExecutionStartToCloseTimeoutSeconds)
+		metricsScope = metricsScope.Tagged(map[string]string{tagworkflowruntimelength: executionRuntimeType})
 		if task.Query != nil && !isFullHistory {
 			// query task and we have a valid cached state
 			metricsScope.Counter(metrics.StickyCacheHit).Inc(1)
@@ -1825,3 +1834,15 @@ func traceLog(fn func()) {
 		fn()
 	}
 }
+
+func workflowCategorizedByTimeout(executionTimeout int32) string {
+	if executionTimeout <= defaultInstantLivedWorkflowTimeoutUpperLimitInSec {
+		return "instant"
+	} else if executionTimeout <= defaultShortLivedWorkflowTimeoutUpperLimitInSec {
+		return "short"
+	} else if executionTimeout <= defaultMediumLivedWorkflowTimeoutUpperLimitInSec {
+		return "intermediate"
+	} else {
+		return "long"
+	}
+}