Support client-side specification of consistency for DescribeWorkflowExecution and GetWorkflowExecutionHistory (#6789)

* feat(frontend): Support client requested consistency levels

Some clients require strong consistency for reads on the DescribeWorkflowExecution and GetWorkflowExecutionHistory rpcs, and are willing to accept the latency trade-off.

This adds internal support for endpoints to define a consistency level.

* Add tests and remove QueryWorkflowStrongConsistency from configuration

* make pr

Author: c-warren

common/types/shared.go

@@ -1808,8 +1808,9 @@ func (v *DescribeTaskListResponse) GetTaskListStatus() (o *TaskListStatus) {
 
 // DescribeWorkflowExecutionRequest is an internal type (TBD...)
 type DescribeWorkflowExecutionRequest struct {
-	Domain    string             `json:"domain,omitempty"`
-	Execution *WorkflowExecution `json:"execution,omitempty"`
+	Domain                string                 `json:"domain,omitempty"`
+	Execution             *WorkflowExecution     `json:"execution,omitempty"`
+	QueryConsistencyLevel *QueryConsistencyLevel `json:"queryConsistencyLevel,omitempty"`
 }
 
 // GetDomain is an internal getter (TBD...)
@@ -1828,6 +1829,14 @@ func (v *DescribeWorkflowExecutionRequest) GetExecution() (o *WorkflowExecution)
 	return
 }
 
+// GetQueryConsistencyLevel is an internal getter (TBD...)
+func (v *DescribeWorkflowExecutionRequest) GetQueryConsistencyLevel() (o QueryConsistencyLevel) {
+	if v != nil && v.QueryConsistencyLevel != nil {
+		return *v.QueryConsistencyLevel
+	}
+	return
+}
+
 // DescribeWorkflowExecutionResponse is an internal type (TBD...)
 type DescribeWorkflowExecutionResponse struct {
 	ExecutionConfiguration *WorkflowExecutionConfiguration `json:"executionConfiguration,omitempty"`
@@ -2599,6 +2608,7 @@ type GetWorkflowExecutionHistoryRequest struct {
 	WaitForNewEvent        bool                    `json:"waitForNewEvent,omitempty"`
 	HistoryEventFilterType *HistoryEventFilterType `json:"HistoryEventFilterType,omitempty"`
 	SkipArchival           bool                    `json:"skipArchival,omitempty"`
+	QueryConsistencyLevel  *QueryConsistencyLevel  `json:"queryConsistencyLevel,omitempty"`
 }
 
 // GetDomain is an internal getter (TBD...)
@@ -2657,6 +2667,14 @@ func (v *GetWorkflowExecutionHistoryRequest) GetSkipArchival() (o bool) {
 	return
 }
 
+// GetQueryConsistencyLevel is an internal getter (TBD...)
+func (v *GetWorkflowExecutionHistoryRequest) GetQueryConsistencyLevel() (o QueryConsistencyLevel) {
+	if v != nil && v.QueryConsistencyLevel != nil {
+		return *v.QueryConsistencyLevel
+	}
+	return
+}
+
 // GetWorkflowExecutionHistoryResponse is an internal type (TBD...)
 type GetWorkflowExecutionHistoryResponse struct {
 	History       *History    `json:"history,omitempty"`

service/frontend/templates/clusterredirection.tmpl

@@ -12,10 +12,10 @@ import (
 	frontendcfg "github.com/uber/cadence/service/frontend/config"
 )
 
-{{$nonFowradingAPIs := list "Health" "DeprecateDomain" "DescribeDomain" "ListDomains" "RegisterDomain" "UpdateDomain" "GetSearchAttributes" "GetClusterInfo" "DiagnoseWorkflowExecution"}}
+{{$nonForwardingAPIs := list "Health" "DeprecateDomain" "DescribeDomain" "ListDomains" "RegisterDomain" "UpdateDomain" "GetSearchAttributes" "GetClusterInfo" "DiagnoseWorkflowExecution"}}
 {{$domainIDAPIs := list "RecordActivityTaskHeartbeat" "RespondActivityTaskCanceled" "RespondActivityTaskCompleted" "RespondActivityTaskFailed" "RespondDecisionTaskCompleted" "RespondDecisionTaskFailed" "RespondQueryTaskCompleted"}}
 {{$queryTaskTokenAPIs := list "RespondQueryTaskCompleted"}}
-{{$specialCaseAPIs := list "QueryWorkflow"}}
+{{$readAPIsWithStrongConsistency := list "QueryWorkflow" "DescribeWorkflowExecution" "GetWorkflowExecutionHistory"}}
 
 type (
 	// ClusterRedirectionHandlerImpl is simple wrapper over frontend service, doing redirection based on policy for global domains not being active in current cluster
@@ -55,13 +55,24 @@ func NewAPIHandler(
 }
 
 {{range $method := .Interface.Methods}}
-{{- if not (has $method.Name $specialCaseAPIs)}}
 func (handler *clusterRedirectionHandler) {{$method.Declaration}} {
-	{{- if has $method.Name $nonFowradingAPIs}}
+	{{- if has $method.Name $nonForwardingAPIs}}
 	return handler.frontendHandler.{{$method.Call}}
 	{{- else}}
-	var apiName = "{{$method.Name}}"
-	var cluster string
+	var (
+		apiName = "{{$method.Name}}"
+		cluster string
+		requestedConsistencyLevel types.QueryConsistencyLevel = types.QueryConsistencyLevelEventual
+	)
+
+	{{- if has $method.Name $readAPIsWithStrongConsistency}}
+	// Only autoforward strong consistent queries, this is done for two reasons:
+	// 1. Query is meant to be fast, autoforwarding all queries will increase latency.
+	// 2. If eventual consistency was requested then the results from running out of local dc will be fine.
+	if {{(index $method.Params 1).Name}}.GetQueryConsistencyLevel() == types.QueryConsistencyLevelStrong {
+		requestedConsistencyLevel = types.QueryConsistencyLevelStrong
+	}
+	{{- end}}
 
     {{$policyMethod := "WithDomainNameRedirect"}}
     {{$domain := printf "%s.GetDomain()" (index $method.Params 1).Name}}
@@ -94,7 +105,7 @@ func (handler *clusterRedirectionHandler) {{$method.Declaration}} {
 	}
 	{{- end}}
 
-	err = handler.redirectionPolicy.{{$policyMethod}}(ctx, {{$domain}}, apiName, func(targetDC string) error {
+	err = handler.redirectionPolicy.{{$policyMethod}}(ctx, {{$domain}}, apiName, requestedConsistencyLevel, func(targetDC string) error {
 		cluster = targetDC
 		switch {
 		case targetDC == handler.currentClusterName:
@@ -110,38 +121,3 @@ func (handler *clusterRedirectionHandler) {{$method.Declaration}} {
 	{{- end}}
 }
 {{end}}
-{{end}}
-
-func (handler *clusterRedirectionHandler) QueryWorkflow(
-	ctx context.Context,
-	request *types.QueryWorkflowRequest,
-) (resp *types.QueryWorkflowResponse, retError error) {
-	var apiName = "QueryWorkflow"
-	var err error
-	var cluster string
-
-	// Only autoforward strong consistent queries, this is done for two reasons:
-	// 1. Query is meant to be fast, autoforwarding all queries will increase latency.
-	// 2. If eventual consistency was requested then the results from running out of local dc will be fine.
-	if request.GetQueryConsistencyLevel() == types.QueryConsistencyLevelStrong {
-		apiName = "QueryWorkflowStrongConsistency"
-	}
-	scope, startTime := handler.beforeCall(metrics.DCRedirectionQueryWorkflowScope)
-	defer func() {
-		handler.afterCall(recover(), scope, startTime, request.GetDomain(), "", cluster, &retError)
-	}()
-
-	err = handler.redirectionPolicy.WithDomainNameRedirect(ctx, request.GetDomain(), apiName, func(targetDC string) error {
-		cluster = targetDC
-		switch {
-		case targetDC == handler.currentClusterName:
-			resp, err = handler.frontendHandler.QueryWorkflow(ctx, request)
-		default:
-			remoteClient := handler.GetRemoteFrontendClient(targetDC)
-			resp, err = remoteClient.QueryWorkflow(ctx, request, handler.callOptions...)
-		}
-		return err
-	})
-
-	return resp, err
-}