mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
1e31c76ed1
## What changed? Cherry-pick versioning PRs - #9168 - Cache for system protection - #9262 - Cache for system protection - #9239 - Critical PR to enable sending `TargetVersionChanged` flag for Upgrade-on-CaN feature - #9147 - Tracks version drainage properly when version receives workflows via `VersioningOverride`. Needed for automated worker controllers to correctly scale versioned workers that received workflows via `VersioningOverride`. - #9300 - Needed for `approximate_backlog_count` metric to track Current and Ramping version tasks correctly - #9316 - Needed for `approximate_backlog_count` metric to track Current and Ramping version tasks correctly - #8957 - Contains minor metric improvement. Included because it adds a test harness that is used in the two metrics PRs above - #9250 - Bug fix of task rescheduling edge case during AutoUpgrade Transition ## Why? For OSS v1.30.2 ## How did you test it? - [x] built - [ ] run locally and tested manually - [x] covered by existing tests - [ ] added new unit test(s) - [ ] added new functional test(s) --- > [!NOTE] > **Medium Risk** > Touches history/matching worker-versioning paths, adding new caches and changing workflow task/start handling and backlog metric emission; incorrect caching or signaling could affect dispatch/upgrade behavior and observability. > > **Overview** > Adds new worker-versioning protections and upgrade signaling: workflow task started events now persist a `workflow_task_target_worker_deployment_version_changed` flag (and emit a new `workflow_target_version_changed_count` metric) under a new `EnableSendTargetVersionChanged` dynamic config. > > Introduces two new caches with metrics and dynamic config knobs: a `RoutingInfoCache` to avoid repeated `GetTaskQueueUserData` lookups during activity start/transition logic, and a `ReactivationSignalCache` plus `EnableVersionReactivationSignals` to dedupe and asynchronously send “reactivation” signals when workflows are pinned (via start/signal-with-start/reset/update-options) to potentially drained/inactive worker versions. > > Extends matching backlog metrics to support version-attributed reporting by adding `BacklogMetricsEmitInterval` and switching queue DB emission to *physical* backlog gauges (`physical_approximate_backlog_*`) when attribution is enabled, while keeping legacy gauges when disabled. > > Adds frontend scaffolding for a new visibility RPC `CountSchedules` (client plumbing, interception/metadata/quota wiring) but leaves the frontend handler unimplemented, and bumps `go.temporal.io/api` to `v1.62.2`. > > <sup>Written by [Cursor Bugbot]() for commit cb8ae14c86666654c51dd66ae91cb225722e1ba7. This will update automatically on new commits. Configure [here]().</sup> --------- Co-authored-by: Shahab Tajik <[email protected]> Co-authored-by: Shivam <[email protected]>
233 lines
17 KiB
Go
233 lines
17 KiB
Go
package api
|
|
|
|
import "strings"
|
|
|
|
type (
|
|
// Describes the scope of a method (whole cluster or inividual namespace).
|
|
Scope int32
|
|
|
|
// Describes what level of access is needed for a method. Note that this field is
|
|
// completely advisory. Any authorizer implementation may implement whatever logic it
|
|
// chooses, including ignoring this field. It is used by the "default" authorizer to check
|
|
// against roles in claims.
|
|
Access int32
|
|
|
|
// Describes if the method supports long-polled requests.
|
|
Polling int32
|
|
|
|
MethodMetadata struct {
|
|
// Describes the scope of a method (whole cluster or inividual namespace).
|
|
Scope Scope
|
|
// Describes what level of access is needed for a method (advisory).
|
|
Access Access
|
|
// Describes if long polling is supported by the method.
|
|
Polling Polling
|
|
}
|
|
)
|
|
|
|
const (
|
|
// Represents a missing Scope value.
|
|
ScopeUnknown Scope = iota
|
|
// Method affects a single namespace. The request message must contain a string field named "Namespace".
|
|
ScopeNamespace
|
|
// Method affects the whole cluster. The request message must _not_ contain any field named "Namespace".
|
|
ScopeCluster
|
|
)
|
|
|
|
const (
|
|
// Represents a missing Access value.
|
|
AccessUnknown Access = iota
|
|
// Method is read-only and should be accessible to readers.
|
|
AccessReadOnly
|
|
// Method is a normal write method.
|
|
AccessWrite
|
|
// Method is an administrative operation.
|
|
AccessAdmin
|
|
)
|
|
|
|
const (
|
|
// Represents a missing Polling value.
|
|
PollingUnknown Polling = iota
|
|
// Method isn't capable of long-polling.
|
|
PollingNone
|
|
// Method can optionally return long-polled responses.
|
|
PollingCapable
|
|
// Method responses are always long-polled.
|
|
PollingAlways
|
|
)
|
|
|
|
const (
|
|
WorkflowServicePrefix = "/temporal.api.workflowservice.v1.WorkflowService/"
|
|
OperatorServicePrefix = "/temporal.api.operatorservice.v1.OperatorService/"
|
|
HistoryServicePrefix = "/temporal.server.api.historyservice.v1.HistoryService/"
|
|
AdminServicePrefix = "/temporal.server.api.adminservice.v1.AdminService/"
|
|
MatchingServicePrefix = "/temporal.server.api.matchingservice.v1.MatchingService/"
|
|
// Technically not a gRPC service, but still using this format for metadata.
|
|
NexusServicePrefix = "/temporal.api.nexusservice.v1.NexusService/"
|
|
)
|
|
|
|
var (
|
|
workflowServiceMetadata = map[string]MethodMetadata{
|
|
"RegisterNamespace": {Scope: ScopeNamespace, Access: AccessAdmin, Polling: PollingNone},
|
|
"DescribeNamespace": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"ListNamespaces": {Scope: ScopeCluster, Access: AccessReadOnly, Polling: PollingNone},
|
|
"UpdateNamespace": {Scope: ScopeNamespace, Access: AccessAdmin, Polling: PollingNone},
|
|
"DeprecateNamespace": {Scope: ScopeNamespace, Access: AccessAdmin, Polling: PollingNone},
|
|
"StartWorkflowExecution": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"GetWorkflowExecutionHistory": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingCapable},
|
|
"GetWorkflowExecutionHistoryReverse": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"PollWorkflowTaskQueue": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingAlways},
|
|
"RespondWorkflowTaskCompleted": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"RespondWorkflowTaskFailed": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"PollActivityTaskQueue": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingAlways},
|
|
"RecordActivityTaskHeartbeat": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"RecordActivityTaskHeartbeatById": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"RespondActivityTaskCompleted": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"RespondActivityTaskCompletedById": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"RespondActivityTaskFailed": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"RespondActivityTaskFailedById": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"RespondActivityTaskCanceled": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"RespondActivityTaskCanceledById": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"CountActivityExecutions": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"DeleteActivityExecution": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"DescribeActivityExecution": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingCapable},
|
|
"PollActivityExecution": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingAlways},
|
|
"ListActivityExecutions": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"RequestCancelActivityExecution": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"StartActivityExecution": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"TerminateActivityExecution": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"PollNexusTaskQueue": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingAlways},
|
|
"RespondNexusTaskCompleted": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"RespondNexusTaskFailed": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"RequestCancelWorkflowExecution": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"SignalWorkflowExecution": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"SignalWithStartWorkflowExecution": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"ResetWorkflowExecution": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"TerminateWorkflowExecution": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"DeleteWorkflowExecution": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"ListOpenWorkflowExecutions": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"ListClosedWorkflowExecutions": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"ListWorkflowExecutions": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"ListArchivedWorkflowExecutions": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"ScanWorkflowExecutions": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"CountWorkflowExecutions": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"GetSearchAttributes": {Scope: ScopeCluster, Access: AccessReadOnly, Polling: PollingNone},
|
|
"RespondQueryTaskCompleted": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"ResetStickyTaskQueue": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"ShutdownWorker": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"ExecuteMultiOperation": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"QueryWorkflow": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"DescribeWorkflowExecution": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"DescribeTaskQueue": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"GetClusterInfo": {Scope: ScopeCluster, Access: AccessReadOnly, Polling: PollingNone},
|
|
"GetSystemInfo": {Scope: ScopeCluster, Access: AccessReadOnly, Polling: PollingNone},
|
|
"ListTaskQueuePartitions": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"CreateSchedule": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"DescribeSchedule": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"UpdateSchedule": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"PatchSchedule": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"ListScheduleMatchingTimes": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"DeleteSchedule": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"ListSchedules": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"CountSchedules": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"UpdateWorkerBuildIdCompatibility": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"GetWorkerBuildIdCompatibility": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"UpdateWorkerVersioningRules": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"GetWorkerVersioningRules": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"GetWorkerTaskReachability": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"UpdateWorkflowExecution": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"PollWorkflowExecutionUpdate": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingAlways},
|
|
"StartBatchOperation": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"StopBatchOperation": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"DescribeBatchOperation": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"ListBatchOperations": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"UpdateActivityOptions": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"PauseActivity": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"UnpauseActivity": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"ResetActivity": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"UpdateWorkflowExecutionOptions": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"DescribeDeployment": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone}, // [cleanup-wv-pre-release]
|
|
"ListDeployments": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone}, // [cleanup-wv-pre-release]
|
|
"GetDeploymentReachability": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone}, // [cleanup-wv-pre-release]
|
|
"GetCurrentDeployment": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone}, // [cleanup-wv-pre-release]
|
|
"SetCurrentDeployment": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone}, // [cleanup-wv-pre-release]
|
|
"DescribeWorkerDeploymentVersion": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"DescribeWorkerDeployment": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"SetWorkerDeploymentCurrentVersion": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"SetWorkerDeploymentRampingVersion": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"SetWorkerDeploymentManager": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"DeleteWorkerDeployment": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"DeleteWorkerDeploymentVersion": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"UpdateWorkerDeploymentVersionMetadata": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"ListWorkerDeployments": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"CreateWorkflowRule": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"DescribeWorkflowRule": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"DeleteWorkflowRule": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"ListWorkflowRules": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"TriggerWorkflowRule": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"RecordWorkerHeartbeat": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"ListWorkers": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"DescribeWorker": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"UpdateTaskQueueConfig": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"FetchWorkerConfig": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"UpdateWorkerConfig": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"PauseWorkflowExecution": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"UnpauseWorkflowExecution": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
}
|
|
operatorServiceMetadata = map[string]MethodMetadata{
|
|
"AddSearchAttributes": {Scope: ScopeNamespace, Access: AccessAdmin, Polling: PollingNone},
|
|
"RemoveSearchAttributes": {Scope: ScopeNamespace, Access: AccessAdmin, Polling: PollingNone},
|
|
"ListSearchAttributes": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone},
|
|
"DeleteNamespace": {Scope: ScopeNamespace, Access: AccessAdmin, Polling: PollingNone},
|
|
"AddOrUpdateRemoteCluster": {Scope: ScopeCluster, Access: AccessAdmin, Polling: PollingNone},
|
|
"RemoveRemoteCluster": {Scope: ScopeCluster, Access: AccessAdmin, Polling: PollingNone},
|
|
"ListClusters": {Scope: ScopeCluster, Access: AccessAdmin, Polling: PollingNone},
|
|
"CreateNexusEndpoint": {Scope: ScopeCluster, Access: AccessAdmin, Polling: PollingNone},
|
|
"UpdateNexusEndpoint": {Scope: ScopeCluster, Access: AccessAdmin, Polling: PollingNone},
|
|
"DeleteNexusEndpoint": {Scope: ScopeCluster, Access: AccessAdmin, Polling: PollingNone},
|
|
"GetNexusEndpoint": {Scope: ScopeCluster, Access: AccessAdmin, Polling: PollingNone},
|
|
"ListNexusEndpoints": {Scope: ScopeCluster, Access: AccessAdmin, Polling: PollingNone},
|
|
}
|
|
nexusServiceMetadata = map[string]MethodMetadata{
|
|
"DispatchNexusTask": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"DispatchByNamespaceAndTaskQueue": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"DispatchByEndpoint": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"CompleteNexusOperation": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
"CompleteNexusOperationChasm": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone},
|
|
}
|
|
)
|
|
|
|
// GetMethodMetadata gets metadata for a given API method in one of the services exported by
|
|
// frontend (WorkflowService, OperatorService, AdminService).
|
|
func GetMethodMetadata(fullApiName string) MethodMetadata {
|
|
switch {
|
|
case strings.HasPrefix(fullApiName, WorkflowServicePrefix):
|
|
return workflowServiceMetadata[MethodName(fullApiName)]
|
|
case strings.HasPrefix(fullApiName, OperatorServicePrefix):
|
|
return operatorServiceMetadata[MethodName(fullApiName)]
|
|
case strings.HasPrefix(fullApiName, NexusServicePrefix):
|
|
return nexusServiceMetadata[MethodName(fullApiName)]
|
|
case strings.HasPrefix(fullApiName, AdminServicePrefix):
|
|
return MethodMetadata{Scope: ScopeCluster, Access: AccessAdmin}
|
|
default:
|
|
return MethodMetadata{Scope: ScopeUnknown, Access: AccessUnknown}
|
|
}
|
|
}
|
|
|
|
// MethodName returns just the method name from a fully qualified name.
|
|
func MethodName(fullApiName string) string {
|
|
index := strings.LastIndex(fullApiName, "/")
|
|
if index > -1 {
|
|
return fullApiName[index+1:]
|
|
}
|
|
return fullApiName
|
|
}
|
|
|
|
func ServiceName(fullApiName string) string {
|
|
index := strings.LastIndex(fullApiName, "/")
|
|
if index > -1 {
|
|
return fullApiName[:index+1]
|
|
}
|
|
return ""
|
|
}
|