Merge branch 'master' into rate-limiter-metrics

pdoerner · May 22, 2023 · 9543764 · 9543764
2 parents c037475 + 981d22b
commit 9543764
Show file tree

Hide file tree

Showing 41 changed files with 1,436 additions and 640 deletions.
diff --git a/api/historyservice/v1/request_response.pb.go b/api/historyservice/v1/request_response.pb.go
diff --git a/api/persistence/v1/executions.pb.go b/api/persistence/v1/executions.pb.go
diff --git a/common/config/config.go b/common/config/config.go
@@ -426,7 +426,6 @@ type (
 	// DCRedirectionPolicy contains the frontend datacenter redirection policy
 	DCRedirectionPolicy struct {
 		Policy string `yaml:"policy"`
-		ToDC   string `yaml:"toDC"`
 	}
 
 	// Archival contains the config for archival

diff --git a/common/rpc/interceptor/telemetry.go b/common/rpc/interceptor/telemetry.go
@@ -296,6 +296,7 @@ func (ti *TelemetryInterceptor) handleError(
 	// we emit service_error_with_type metrics, no need to emit specific metric for these known error types.
 	case *serviceerror.AlreadyExists,
 		*serviceerror.CancellationAlreadyRequested,
+		*serviceerror.FailedPrecondition,
 		*serviceerror.NamespaceInvalidState,
 		*serviceerror.NamespaceNotActive,
 		*serviceerror.NamespaceNotFound,

diff --git a/config/development-cass-archival.yaml b/config/development-cass-archival.yaml
@@ -68,7 +68,6 @@ clusterMetadata:
 
 dcRedirectionPolicy:
   policy: "noop"
-  toDC: ""
 
 archival:
   history:

diff --git a/config/development-cass-es.yaml b/config/development-cass-es.yaml
@@ -75,7 +75,6 @@ clusterMetadata:
 
 dcRedirectionPolicy:
   policy: "noop"
-  toDC: ""
 
 archival:
   history:

diff --git a/config/development-cass-s3.yaml b/config/development-cass-s3.yaml
@@ -66,7 +66,6 @@ clusterMetadata:
 
 dcRedirectionPolicy:
   policy: "noop"
-  toDC: ""
 
 archival:
   history:

diff --git a/config/development-cass.yaml b/config/development-cass.yaml
@@ -92,7 +92,6 @@ clusterMetadata:
 
 dcRedirectionPolicy:
   policy: "noop"
-  toDC: ""
 
 archival:
   history:

diff --git a/config/development-cluster-a.yaml b/config/development-cluster-a.yaml
@@ -87,7 +87,6 @@ clusterMetadata:
 
 dcRedirectionPolicy:
   policy: "selected-apis-forwarding"
-  toDC: ""
 
 archival:
   history:

diff --git a/config/development-cluster-b.yaml b/config/development-cluster-b.yaml
@@ -92,7 +92,6 @@ clusterMetadata:
 
 dcRedirectionPolicy:
   policy: "selected-apis-forwarding"
-  toDC: ""
 
 archival:
   history:

diff --git a/config/development-cluster-c.yaml b/config/development-cluster-c.yaml
@@ -92,7 +92,6 @@ clusterMetadata:
 
 dcRedirectionPolicy:
   policy: "selected-apis-forwarding"
-  toDC: ""
 
 archival:
   history:

diff --git a/config/development-mysql-es.yaml b/config/development-mysql-es.yaml
@@ -81,7 +81,6 @@ clusterMetadata:
 
 dcRedirectionPolicy:
   policy: "noop"
-  toDC: ""
 
 archival:
   history:

diff --git a/config/development-mysql.yaml b/config/development-mysql.yaml
@@ -83,7 +83,6 @@ clusterMetadata:
 
 dcRedirectionPolicy:
   policy: "noop"
-  toDC: ""
 
 archival:
   history:

diff --git a/config/development-mysql8.yaml b/config/development-mysql8.yaml
@@ -83,7 +83,6 @@ clusterMetadata:
 
 dcRedirectionPolicy:
   policy: "noop"
-  toDC: ""
 
 archival:
   history:

diff --git a/config/development-postgres-es.yaml b/config/development-postgres-es.yaml
@@ -83,7 +83,6 @@ clusterMetadata:
 
 dcRedirectionPolicy:
   policy: "noop"
-  toDC: ""
 
 archival:
   history:

diff --git a/config/development-postgres.yaml b/config/development-postgres.yaml
@@ -83,7 +83,6 @@ clusterMetadata:
 
 dcRedirectionPolicy:
   policy: "noop"
-  toDC: ""
 
 archival:
   history:

diff --git a/config/development-postgres12.yaml b/config/development-postgres12.yaml
@@ -83,7 +83,6 @@ clusterMetadata:
 
 dcRedirectionPolicy:
   policy: "noop"
-  toDC: ""
 
 archival:
   history:

diff --git a/config/development-sqlite-file.yaml b/config/development-sqlite-file.yaml
@@ -107,7 +107,6 @@ clusterMetadata:
 
 dcRedirectionPolicy:
   policy: "noop"
-  toDC: ""
 
 archival:
   history:

diff --git a/config/development-sqlite.yaml b/config/development-sqlite.yaml
@@ -103,7 +103,6 @@ clusterMetadata:
 
 dcRedirectionPolicy:
   policy: "noop"
-  toDC: ""
 
 archival:
   history:

diff --git a/docker/config_template.yaml b/docker/config_template.yaml
@@ -345,7 +345,6 @@ clusterMetadata:
 
 dcRedirectionPolicy:
     policy: "noop"
-    toDC: ""
 
 archival:
   history:

diff --git a/proto/internal/temporal/server/api/historyservice/v1/request_response.proto b/proto/internal/temporal/server/api/historyservice/v1/request_response.proto
@@ -574,6 +574,7 @@ message GenerateLastHistoryReplicationTasksRequest {
 }
 
 message GenerateLastHistoryReplicationTasksResponse {
+    int64 state_transition_count = 1;
 }
 
 message GetReplicationStatusRequest {

diff --git a/proto/internal/temporal/server/api/persistence/v1/executions.proto b/proto/internal/temporal/server/api/persistence/v1/executions.proto
@@ -128,6 +128,7 @@ message WorkflowExecutionInfo {
     int64 user_timer_count = 73;
     int64 request_cancel_external_count = 74;
     int64 signal_external_count = 75;
+    int64 update_count = 77;
     reserved 47;
     reserved 48;
     reserved 49;

diff --git a/service/history/api/queryworkflow/api.go b/service/history/api/queryworkflow/api.go
@@ -33,6 +33,8 @@ import (
 	"go.temporal.io/api/serviceerror"
 	"go.temporal.io/api/workflowservice/v1"
 
+	"go.temporal.io/server/common/log/tag"
+
 	"go.temporal.io/server/api/historyservice/v1"
 	"go.temporal.io/server/api/matchingservice/v1"
 	"go.temporal.io/server/common"
@@ -48,6 +50,9 @@ import (
 	"go.temporal.io/server/service/history/workflow"
 )
 
+// Fail query fast if workflow task keeps failing (attempt >= 3).
+const failQueryWorkflowTaskAttemptCount = 3
+
 func Invoke(
 	ctx context.Context,
 	request *historyservice.QueryWorkflowRequest,
@@ -117,10 +122,15 @@ func Invoke(
 		return nil, consts.ErrWorkflowTaskNotScheduled
 	}
 
-	if mutableState.IsTransientWorkflowTask() {
+	if mutableState.GetExecutionInfo().WorkflowTaskAttempt >= failQueryWorkflowTaskAttemptCount {
 		// while workflow task is failing, the query to that workflow will also fail. Failing fast here to prevent wasting
 		// resources to load history for a query that will fail.
-		return nil, serviceerror.NewFailedPrecondition("Cannot query workflow due to Workflow Task in failed state.")
+		shard.GetLogger().Info("Fail query fast due to WorkflowTask in failed state.",
+			tag.WorkflowNamespace(request.Request.Namespace),
+			tag.WorkflowNamespaceID(workflowKey.NamespaceID),
+			tag.WorkflowID(workflowKey.WorkflowID),
+			tag.WorkflowRunID(workflowKey.RunID))
+		return nil, serviceerror.NewWorkflowNotReady("Unable to query workflow due to Workflow Task in failed state.")
 	}
 
 	// There are two ways in which queries get dispatched to workflow worker. First, queries can be dispatched on workflow tasks.

diff --git a/service/history/api/replication/generate_task.go b/service/history/api/replication/generate_task.go
@@ -65,7 +65,8 @@ func GenerateTask(
 	}
 	defer func() { wfContext.GetReleaseFn()(retError) }()
 
-	task, err := wfContext.GetMutableState().GenerateMigrationTasks()
+	mutableState := wfContext.GetMutableState()
+	task, stateTransitionCount, err := mutableState.GenerateMigrationTasks()
 	if err != nil {
 		return nil, err
 	}
@@ -83,5 +84,7 @@ func GenerateTask(
 	if err != nil {
 		return nil, err
 	}
-	return &historyservice.GenerateLastHistoryReplicationTasksResponse{}, nil
+	return &historyservice.GenerateLastHistoryReplicationTasksResponse{
+		StateTransitionCount: stateTransitionCount,
+	}, nil
 }
diff --git a/service/history/api/updateworkflow/api.go b/service/history/api/updateworkflow/api.go
@@ -42,6 +42,7 @@ import (
 	"go.temporal.io/server/api/matchingservice/v1"
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/definition"
+	"go.temporal.io/server/common/log/tag"
 	"go.temporal.io/server/common/namespace"
 	serviceerrors "go.temporal.io/server/common/serviceerror"
 	"go.temporal.io/server/internal/effect"
@@ -52,6 +53,11 @@ import (
 	"go.temporal.io/server/service/history/workflow/update"
 )
 
+const (
+	// Fail update fast if workflow task keeps failing (attempt >= 3).
+	failUpdateWorkflowTaskAttemptCount = 3
+)
+
 func Invoke(
 	ctx context.Context,
 	req *historyservice.UpdateWorkflowExecutionRequest,
@@ -127,6 +133,19 @@ func Invoke(
 			return consts.ErrWorkflowExecutionNotFound
 		}
 
+		if ms.GetExecutionInfo().WorkflowTaskAttempt >= failUpdateWorkflowTaskAttemptCount {
+			// If workflow task is constantly failing, the update to that workflow will also fail.
+			// Additionally, workflow update can't "fix" workflow state because updates (delivered with messages)
+			// are applied after events.
+			// Failing API call fast here to prevent wasting resources for an update that will fail.
+			shardCtx.GetLogger().Info("Fail update fast due to WorkflowTask in failed state.",
+				tag.WorkflowNamespace(req.Request.Namespace),
+				tag.WorkflowNamespaceID(wfKey.NamespaceID),
+				tag.WorkflowID(wfKey.WorkflowID),
+				tag.WorkflowRunID(wfKey.RunID))
+			return serviceerror.NewWorkflowNotReady("Unable to perform workflow execution update due to Workflow Task in failed state.")
+		}
+
 		updateID := req.GetRequest().GetRequest().GetMeta().GetUpdateId()
 		updateReg := weCtx.GetUpdateRegistry(ctx)
 		var alreadyExisted bool

diff --git a/service/history/configs/config.go b/service/history/configs/config.go
@@ -239,7 +239,6 @@ type Config struct {
 	ReplicationEnableDLQMetrics                          dynamicconfig.BoolPropertyFn
 
 	ReplicationStreamSyncStatusDuration      dynamicconfig.DurationPropertyFn
-	ReplicationStreamMinReconnectDuration    dynamicconfig.DurationPropertyFn
 	ReplicationProcessorSchedulerQueueSize   dynamicconfig.IntPropertyFn
 	ReplicationProcessorSchedulerWorkerCount dynamicconfig.IntPropertyFn
 
@@ -419,7 +418,6 @@ func NewConfig(
 		ReplicationEnableDLQMetrics:                           dc.GetBoolProperty(dynamicconfig.ReplicationEnableDLQMetrics, true),
 
 		ReplicationStreamSyncStatusDuration:      dc.GetDurationProperty(dynamicconfig.ReplicationStreamSyncStatusDuration, 1*time.Second),
-		ReplicationStreamMinReconnectDuration:    dc.GetDurationProperty(dynamicconfig.ReplicationStreamMinReconnectDuration, 4*time.Second),
 		ReplicationProcessorSchedulerQueueSize:   dc.GetIntProperty(dynamicconfig.ReplicationProcessorSchedulerQueueSize, 128),
 		ReplicationProcessorSchedulerWorkerCount: dc.GetIntProperty(dynamicconfig.ReplicationProcessorSchedulerWorkerCount, 512),
 

diff --git a/service/history/replication/bi_direction_stream.go b/service/history/replication/bi_direction_stream.go
@@ -63,6 +63,7 @@ type (
 		Send(Req) error
 		Recv() (<-chan StreamResp[Resp], error)
 		Close()
+		IsValid() bool
 	}
 	StreamResp[Resp any] struct {
 		Resp Resp
@@ -135,6 +136,12 @@ func (s *BiDirectionStreamImpl[Req, Resp]) Close() {
 	s.closeLocked()
 }
 
+func (s *BiDirectionStreamImpl[Req, Resp]) IsValid() bool {
+	s.Lock()
+	defer s.Unlock()
+	return s.status != streamStatusClosed
+}
+
 func (s *BiDirectionStreamImpl[Req, Resp]) closeLocked() {
 	if s.status == streamStatusClosed {
 		return

diff --git a/service/history/replication/bi_direction_stream_test.go b/service/history/replication/bi_direction_stream_test.go
@@ -117,18 +117,21 @@ func (s *biDirectionStreamSuite) TestLazyInit() {
 	s.biDirectionStream.Unlock()
 	s.NoError(err)
 	s.Equal(s.streamClient, s.biDirectionStream.streamingClient)
+	s.True(s.biDirectionStream.IsValid())
 
 	s.biDirectionStream.Lock()
 	err = s.biDirectionStream.lazyInitLocked()
 	s.biDirectionStream.Unlock()
 	s.NoError(err)
 	s.Equal(s.streamClient, s.biDirectionStream.streamingClient)
+	s.True(s.biDirectionStream.IsValid())
 
 	s.biDirectionStream.Close()
 	s.biDirectionStream.Lock()
 	err = s.biDirectionStream.lazyInitLocked()
 	s.biDirectionStream.Unlock()
 	s.Error(err)
+	s.False(s.biDirectionStream.IsValid())
 }
 
 func (s *biDirectionStreamSuite) TestSend() {
@@ -140,9 +143,7 @@ func (s *biDirectionStreamSuite) TestSend() {
 		s.NoError(err)
 	}
 	s.Equal(reqs, s.streamClient.requests)
-	s.biDirectionStream.Lock()
-	defer s.biDirectionStream.Unlock()
-	s.Equal(streamStatusOpen, s.biDirectionStream.status)
+	s.True(s.biDirectionStream.IsValid())
 }
 
 func (s *biDirectionStreamSuite) TestSend_Err() {
@@ -152,9 +153,7 @@ func (s *biDirectionStreamSuite) TestSend_Err() {
 
 	err := s.biDirectionStream.Send(rand.Int())
 	s.Error(err)
-	s.biDirectionStream.Lock()
-	defer s.biDirectionStream.Unlock()
-	s.Equal(streamStatusClosed, s.biDirectionStream.status)
+	s.False(s.biDirectionStream.IsValid())
 }
 
 func (s *biDirectionStreamSuite) TestRecv() {
@@ -168,9 +167,7 @@ func (s *biDirectionStreamSuite) TestRecv() {
 		resps = append(resps, streamResp.Resp)
 	}
 	s.Equal(s.streamClient.responses, resps)
-	s.biDirectionStream.Lock()
-	defer s.biDirectionStream.Unlock()
-	s.Equal(streamStatusClosed, s.biDirectionStream.status)
+	s.False(s.biDirectionStream.IsValid())
 }
 
 func (s *biDirectionStreamSuite) TestRecv_Err() {
@@ -183,9 +180,8 @@ func (s *biDirectionStreamSuite) TestRecv_Err() {
 	s.Error(streamResp.Err)
 	_, ok := <-streamRespChan
 	s.False(ok)
-	s.biDirectionStream.Lock()
-	defer s.biDirectionStream.Unlock()
-	s.Equal(streamStatusClosed, s.biDirectionStream.status)
+	s.False(s.biDirectionStream.IsValid())
+
 }
 
 func (p *mockStreamClientProvider) Get(
-Original file line number
+Diff line change
@@ Expand Up / @@ -574,6 +574,7 @@ message GenerateLastHistoryReplicationTasksRequest { @@
     }
     message GenerateLastHistoryReplicationTasksResponse {
+        int64 state_transition_count = 1;
     }
     message GetReplicationStatusRequest {
@@ Expand Down @@