Check on-going failover across clusters (uber#3206)

common/domain/errors.go

@@ -31,6 +31,7 @@ var (
 	errActiveClusterNotInClusters          = &workflow.BadRequestError{Message: "Active cluster is not contained in all clusters."}
 	errCannotDoDomainFailoverAndUpdate     = &workflow.BadRequestError{Message: "Cannot set active cluster to current cluster when other parameters are set."}
 	errCannotDoGracefulFailoverFromCluster = &workflow.BadRequestError{Message: "Cannot start the graceful failover from a to-be-passive cluster."}
+	errGracefulFailoverInActiveCluster     = &workflow.BadRequestError{Message: "Cannot start the graceful failover from an active cluster to an active cluster."}
 	errOngoingGracefulFailover             = &workflow.BadRequestError{Message: "Cannot start concurrent graceful failover."}
 	errInvalidGracefulFailover             = &workflow.BadRequestError{Message: "Cannot start graceful failover without updating active cluster or in local domain."}
 

common/domain/handler.go

@@ -363,6 +363,7 @@ func (d *HandlerImpl) UpdateDomain(
 	failoverNotificationVersion := getResponse.FailoverNotificationVersion
 	isGlobalDomain := getResponse.IsGlobalDomain
 	gracefulFailoverEndTime := getResponse.FailoverEndTime
+	currentActiveCluster := replicationConfig.ActiveClusterName
 
 	// whether history archival config changed
 	historyArchivalConfigChanged := false
@@ -442,6 +443,9 @@ func (d *HandlerImpl) UpdateDomain(
 		if replicationConfig.ActiveClusterName != d.clusterMetadata.GetCurrentClusterName() {
 			return nil, errCannotDoGracefulFailoverFromCluster
 		}
+		if replicationConfig.ActiveClusterName == currentActiveCluster {
+			return nil, errGracefulFailoverInActiveCluster
+		}
 		// cannot have concurrent failover
 		if gracefulFailoverEndTime != nil {
 			return nil, errOngoingGracefulFailover

common/domain/handler_test.go

@@ -436,6 +436,10 @@ func (s *domainHandlerCommonSuite) TestUpdateDomain_GracefulFailover_Success() {
 	}
 	err := s.handler.RegisterDomain(context.Background(), registerRequest)
 	s.NoError(err)
+	resp1, _ := s.metadataMgr.GetDomain(&persistence.GetDomainRequest{
+		Name: domain,
+	})
+	s.Equal(resp1.ReplicationConfig.ActiveClusterName, "standby")
 
 	updateRequest := &workflow.UpdateDomainRequest{
 		Name: common.StringPtr(domain),

service/frontend/workflowHandler.go

@@ -24,6 +24,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -37,6 +38,7 @@ import (
 	h "github.com/uber/cadence/.gen/go/history"
 	m "github.com/uber/cadence/.gen/go/matching"
 	gen "github.com/uber/cadence/.gen/go/shared"
+	"github.com/uber/cadence/client/frontend"
 	"github.com/uber/cadence/common"
 	"github.com/uber/cadence/common/archiver"
 	"github.com/uber/cadence/common/backoff"
@@ -370,6 +372,15 @@ func (wh *WorkflowHandler) UpdateDomain(
 		}
 	}
 
+	if isGraceFailoverRequest(updateRequest) {
+		if err := wh.checkOngoingFailover(
+			ctx,
+			updateRequest.Name,
+		); err != nil {
+			return nil, err
+		}
+	}
+
 	if updateRequest.GetName() == "" {
 		return nil, errDomainNotSet
 	}
@@ -3706,6 +3717,68 @@ func isFailoverRequest(updateRequest *gen.UpdateDomainRequest) bool {
 	return updateRequest.ReplicationConfiguration != nil && updateRequest.ReplicationConfiguration.ActiveClusterName != nil
 }
 
+func isGraceFailoverRequest(updateRequest *gen.UpdateDomainRequest) bool {
+	return updateRequest.IsSetFailoverTimeoutInSeconds()
+}
+
+func (wh *WorkflowHandler) checkOngoingFailover(
+	ctx context.Context,
+	domainName *string,
+) error {
+
+	clusterMetadata := wh.GetClusterMetadata()
+	respChan := make(chan *gen.DescribeDomainResponse, len(clusterMetadata.GetAllClusterInfo()))
+	wg := &sync.WaitGroup{}
+
+	describeDomain := func(
+		ctx context.Context,
+		client frontend.Client,
+		domainName *string,
+	) {
+		defer wg.Done()
+		resp, _ := client.DescribeDomain(
+			ctx,
+			&gen.DescribeDomainRequest{
+				Name: domainName,
+			},
+		)
+		respChan <- resp
+	}
+
+	for clusterName, cluster := range clusterMetadata.GetAllClusterInfo() {
+		if !cluster.Enabled {
+			continue
+		}
+		frontendClient := wh.GetRemoteFrontendClient(clusterName)
+		wg.Add(1)
+		go describeDomain(
+			ctx,
+			frontendClient,
+			domainName,
+		)
+	}
+	wg.Wait()
+	close(respChan)
+
+	var failoverVersion *int64
+	for resp := range respChan {
+		if resp == nil {
+			return &gen.InternalServiceError{
+				Message: "Failed to verify failover version from all clusters",
+			}
+		}
+		if failoverVersion == nil {
+			failoverVersion = resp.FailoverVersion
+		}
+		if failoverVersion != resp.FailoverVersion {
+			return &gen.BadRequestError{
+				Message: "Concurrent failover is not allow.",
+			}
+		}
+	}
+	return nil
+}
+
 func (wh *WorkflowHandler) historyArchived(ctx context.Context, request *gen.GetWorkflowExecutionHistoryRequest, domainID string) bool {
 	if request.GetExecution() == nil || request.GetExecution().GetRunId() == "" {
 		return false

tools/cli/domainCommands.go

@@ -172,9 +172,17 @@ func (d *domainCLIImpl) UpdateDomain(c *cli.Context) {
 		replicationConfig := &shared.DomainReplicationConfiguration{
 			ActiveClusterName: common.StringPtr(activeCluster),
 		}
+
+		var failoverTimeout *int32
+		if c.String(FlagFailoverType) == gracefulFailoverType {
+			timeout := int32(c.Int(FlagFailoverTimeout))
+			failoverTimeout = &timeout
+		}
+
 		updateRequest = &shared.UpdateDomainRequest{
 			Name:                     common.StringPtr(domainName),
 			ReplicationConfiguration: replicationConfig,
+			FailoverTimeoutInSeconds: failoverTimeout,
 		}
 	} else {
 		resp, err := d.describeDomain(ctx, &shared.DescribeDomainRequest{
@@ -247,12 +255,6 @@ func (d *domainCLIImpl) UpdateDomain(c *cli.Context) {
 			badBinaryToDelete = common.StringPtr(c.String(FlagRemoveBadBinary))
 		}
 
-		var failoverTimeout *int32
-		if c.String(FlagFailoverType) == gracefulFailoverType {
-			timeout := int32(c.Int(FlagFailoverTimeout))
-			failoverTimeout = &timeout
-		}
-
 		updateInfo := &shared.UpdateDomainInfo{
 			Description: common.StringPtr(description),
 			OwnerEmail:  common.StringPtr(ownerEmail),
@@ -276,7 +278,6 @@ func (d *domainCLIImpl) UpdateDomain(c *cli.Context) {
 			Configuration:            updateConfig,
 			ReplicationConfiguration: replicationConfig,
 			DeleteBadBinary:          badBinaryToDelete,
-			FailoverTimeoutInSeconds: failoverTimeout,
 		}
 	}