Active-active domain support - Part 1/N

cmd/server/cadence/server.go

@@ -221,10 +221,7 @@ func (s *server) startService() common.Daemon {
 	params.GetIsolationGroups = getFromDynamicConfig(params, dc)
 
 	params.ClusterMetadata = cluster.NewMetadata(
-		clusterGroupMetadata.FailoverVersionIncrement,
-		clusterGroupMetadata.PrimaryClusterName,
-		clusterGroupMetadata.CurrentClusterName,
-		clusterGroupMetadata.ClusterGroup,
+		*clusterGroupMetadata,
 		dc.GetBoolPropertyFilteredByDomain(dynamicproperties.UseNewInitialFailoverVersion),
 		params.MetricsClient,
 		params.Logger,

common/activecluster/fake.go

@@ -0,0 +1,88 @@
+// The MIT License (MIT)
+
+// Copyright (c) 2017-2020 Uber Technologies Inc.
+
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+// SOFTWARE.
+
+package activecluster
+
+import (
+	"sync/atomic"
+	"time"
+)
+
+var (
+	wf1FailoverTime = 60 * time.Second
+
+	cluster0Result = &LookupResult{
+		Region:          "region0",
+		ClusterName:     "cluster0",
+		FailoverVersion: 1,
+	}
+
+	cluster1Result = &LookupResult{
+		Region:          "region1",
+		ClusterName:     "cluster1",
+		FailoverVersion: 2,
+	}
+)
+
+// TODO(active-active): Remove below fake implementation and implement properly
+// - lookup active region given <domain id, wf id, run id> from executions table RowType=ActiveCluster.
+// - cache this info
+// - add metrics for cache hit/miss
+// - return cluster name
+
+// Fake logic:
+//   - wf1 is active in cluster0 for first 60 seconds, then active in cluster1.
+//     Note: Simulation sleeps for 30s in the beginning and runs wf1 for 60s. So wf1 should start in cluster0 and complete in cluster1.
+//   - other workflows are always active in cluster1
+func (m *manager) fakeLookupWorkflow(wfID string) (*LookupResult, error) {
+	if wfID == "wf1" && (m.wf1StartTime.IsZero() || atomic.LoadInt32(&m.wf1FailedOver) == 0) {
+		if m.wf1StartTime.IsZero() {
+			m.logger.Debug("Initializing wf1 failover timer")
+			m.wf1StartTime = time.Now()
+			go m.fakeEntityMapChange()
+		}
+		m.logger.Debug("Returning cluster0 for wf1")
+		return cluster0Result, nil
+	}
+
+	if wfID == "wf1" {
+		m.logger.Debug("Returning cluster1 for wf1")
+	}
+
+	return cluster1Result, nil
+}
+
+func (m *manager) fakeEntityMapChange() {
+	// Based on the fake logic, wf1 will failover to cluster1 after 60 seconds.
+	t := time.NewTimer(wf1FailoverTime)
+	defer t.Stop()
+	for {
+		select {
+		case <-t.C:
+			m.logger.Debugf("Faking wf1 failover")
+			atomic.CompareAndSwapInt32(&m.wf1FailedOver, 0, 1)
+			m.notifyChangeCallbacks(ChangeTypeEntityMap)
+		case <-m.ctx.Done():
+			return
+		}
+	}
+}

common/activecluster/manager.go

@@ -0,0 +1,186 @@
+// The MIT License (MIT)
+
+// Copyright (c) 2017-2020 Uber Technologies Inc.
+
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+// SOFTWARE.
+
+package activecluster
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/uber/cadence/common/cache"
+	"github.com/uber/cadence/common/cluster"
+	"github.com/uber/cadence/common/log"
+	"github.com/uber/cadence/common/log/tag"
+	"github.com/uber/cadence/common/metrics"
+	"github.com/uber/cadence/common/types"
+)
+
+type DomainIDToDomainFn func(id string) (*cache.DomainCacheEntry, error)
+
+type manager struct {
+	domainIDToDomainFn DomainIDToDomainFn
+	clusterMetadata    cluster.Metadata
+	metricsCl          metrics.Client
+	logger             log.Logger
+	ctx                context.Context
+	cancel             context.CancelFunc
+
+	// TODO: fakes to be remove
+	wf1StartTime  time.Time
+	wf1FailedOver int32
+
+	changeCallbacksLock sync.RWMutex
+	changeCallbacks     map[int]func(ChangeType)
+}
+
+func NewManager(
+	domainIDToDomainFn DomainIDToDomainFn,
+	clusterMetadata cluster.Metadata,
+	metricsCl metrics.Client,
+	logger log.Logger,
+) Manager {
+	ctx, cancel := context.WithCancel(context.Background())
+	return &manager{
+		domainIDToDomainFn: domainIDToDomainFn,
+		clusterMetadata:    clusterMetadata,
+		metricsCl:          metricsCl,
+		logger:             logger.WithTags(tag.ComponentActiveClusterManager),
+		ctx:                ctx,
+		cancel:             cancel,
+		changeCallbacks:    make(map[int]func(ChangeType)),
+	}
+}
+
+func (m *manager) Start() {
+}
+
+func (m *manager) Stop() {
+	m.cancel()
+}
+
+func (m *manager) LookupExternalEntity(ctx context.Context, entityType, entityKey string) (*LookupResult, error) {
+	// TODO: implement this
+	return nil, errors.New("not implemented")
+}
+
+func (m *manager) LookupExternalEntityOfNewWorkflow(ctx context.Context, req *types.HistoryStartWorkflowExecutionRequest) (*LookupResult, error) {
+	d, err := m.domainIDToDomainFn(req.DomainUUID)
+	if err != nil {
+		return nil, err
+	}
+
+	if !d.GetReplicationConfig().IsActiveActive() {
+		// Not an active-active domain. return ActiveClusterName from domain entry
+		return &LookupResult{
+			ClusterName:     d.GetReplicationConfig().ActiveClusterName,
+			FailoverVersion: d.GetFailoverVersion(),
+		}, nil
+	}
+
+	wfID := req.StartRequest.WorkflowID
+	return m.fakeLookupWorkflow(wfID)
+}
+
+func (m *manager) LookupWorkflow(ctx context.Context, domainID, wfID, rID string) (*LookupResult, error) {
+	d, err := m.domainIDToDomainFn(domainID)
+	if err != nil {
+		return nil, err
+	}
+
+	if !d.GetReplicationConfig().IsActiveActive() {
+		// Not an active-active domain. return ActiveClusterName from domain entry
+		return &LookupResult{
+			ClusterName:     d.GetReplicationConfig().ActiveClusterName,
+			FailoverVersion: d.GetFailoverVersion(),
+		}, nil
+	}
+
+	return m.fakeLookupWorkflow(wfID)
+}
+
+func (m *manager) ClusterNameForFailoverVersion(failoverVersion int64, domainID string) (string, error) {
+	d, err := m.domainIDToDomainFn(domainID)
+	if err != nil {
+		return "", err
+	}
+
+	if !d.GetReplicationConfig().IsActiveActive() {
+		cluster, err := m.clusterMetadata.ClusterNameForFailoverVersion(failoverVersion)
+		if err != nil {
+			return "", err
+		}
+		return cluster, nil
+	}
+
+	// For active-active domains, the failover version might be mapped to a cluster or a region
+	// First check if it maps to a cluster
+	cluster, err := m.clusterMetadata.ClusterNameForFailoverVersion(failoverVersion)
+	if err == nil {
+		return cluster, nil
+	}
+
+	// Check if it maps to a region.
+	region, err := m.clusterMetadata.RegionForFailoverVersion(failoverVersion)
+	if err != nil {
+		return "", err
+	}
+
+	// Now we know the region, find the cluster in the domain's active cluster list which belongs to the region
+	cfg, ok := d.GetReplicationConfig().ActiveClusters.RegionToClusterMap[region]
+	if !ok {
+		return "", fmt.Errorf("could not find region %s in the domain's active cluster config", region)
+	}
+
+	enabledClusters := m.clusterMetadata.GetEnabledClusterInfo()
+	_, ok = enabledClusters[cfg.ActiveClusterName]
+	if !ok {
+		return "", fmt.Errorf("cluster %s is disabled", cfg.ActiveClusterName)
+	}
+
+	return cfg.ActiveClusterName, nil
+}
+
+func (m *manager) RegisterChangeCallback(shardID int, callback func(ChangeType)) {
+	m.changeCallbacksLock.Lock()
+	defer m.changeCallbacksLock.Unlock()
+
+	m.changeCallbacks[shardID] = callback
+}
+
+func (m *manager) UnregisterChangeCallback(shardID int) {
+	m.changeCallbacksLock.Lock()
+	defer m.changeCallbacksLock.Unlock()
+
+	delete(m.changeCallbacks, shardID)
+}
+
+func (m *manager) notifyChangeCallbacks(changeType ChangeType) {
+	m.changeCallbacksLock.RLock()
+	defer m.changeCallbacksLock.RUnlock()
+
+	for _, callback := range m.changeCallbacks {
+		callback(changeType)
+	}
+}