active-active prototyping

Author: taylanisikdemir

common/activecluster/manager.go

@@ -0,0 +1,112 @@
+// 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"
+
+	"github.com/uber/cadence/common"
+	"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"
+)
+
+//go:generate mockgen -package $GOPACKAGE -destination manager_mock.go -self_package github.com/uber/cadence/common/activecluster github.com/uber/cadence/common/activecluster Manager
+
+type Manager interface {
+	common.Daemon
+
+	// ActiveCluster returns active cluster name of domains.
+	//  1. If domain is local, return current cluster name.
+	//  2. If domain is active-passive global domain, return it's domain entry's ActiveClusterName.
+	//  3. If domain is active-active global domain, check workflow's activeness metadata and return corresponding active cluster name.
+	ActiveCluster(ctx context.Context, domainID, wfID, rID string) (string, error)
+
+	// FailoverVersion returns failover version of domains.
+	//  1. If domain is local, return current it's domain entry's failover version.
+	//  2. If domain is active-passive global domain, return it's domain entry's failover version.
+	//  3. If domain is active-active domain, check workflow's activeness metadata and lookup failover version from EntityActiveRegion lookup table.
+	FailoverVersion(ctx context.Context, domainID, wfID, rID string) (int64, error)
+}
+
+type DomainIDToDomainFn func(id string) (*cache.DomainCacheEntry, error)
+
+type manager struct {
+	domainIDToDomainFn DomainIDToDomainFn
+	clusterMetadata    cluster.Metadata
+	metricsCl          metrics.Client
+	logger             log.Logger
+}
+
+func NewManager(
+	domainIDToDomainFn DomainIDToDomainFn,
+	clusterMetadata cluster.Metadata,
+	metricsCl metrics.Client,
+	logger log.Logger,
+) Manager {
+	return &manager{
+		domainIDToDomainFn: domainIDToDomainFn,
+		clusterMetadata:    clusterMetadata,
+		metricsCl:          metricsCl,
+		logger:             logger.WithTags(tag.ComponentActiveRegionManager),
+	}
+}
+
+func (m *manager) ActiveCluster(ctx context.Context, domainID, wfID, rID string) (string, error) {
+	d, err := m.domainIDToDomainFn(domainID)
+	if err != nil {
+		return "", err
+	}
+
+	if !d.GetReplicationConfig().IsActiveActive() {
+		// Not an active-active domain. return ActiveClusterName from domain entry
+		return d.GetReplicationConfig().ActiveClusterName, nil
+	}
+
+	// TODO: 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
+	if wfID == "wf1" {
+		return "cluster0", nil
+	}
+	if wfID == "wf2" {
+		return "cluster1", nil
+	}
+
+	return d.GetReplicationConfig().ActiveClusterName, nil
+}
+
+func (m *manager) FailoverVersion(ctx context.Context, domainID, wfID, rID string) (int64, error) {
+	// TODO: implement this
+	return 0, nil
+}
+
+func (m *manager) Start() {
+}
+
+func (m *manager) Stop() {
+}

common/activecluster/manager_mock.go

@@ -97,15 +97,16 @@ type (
 	}
 
 	DefaultDomainCache struct {
-		status        int32
-		shutdownChan  chan struct{}
-		clusterGroup  string
-		cacheNameToID *atomic.Value
-		cacheByID     *atomic.Value
-		domainManager persistence.DomainManager
-		timeSource    clock.TimeSource
-		scope         metrics.Scope
-		logger        log.Logger
+		status          int32
+		shutdownChan    chan struct{}
+		clusterGroup    string
+		clusterMetadata cluster.Metadata
+		cacheNameToID   *atomic.Value
+		cacheByID       *atomic.Value
+		domainManager   persistence.DomainManager
+		timeSource      clock.TimeSource
+		scope           metrics.Scope
+		logger          log.Logger
 
 		// refresh lock is used to guarantee at most one
 		// coroutine is doing domain refreshment
@@ -126,11 +127,12 @@ type (
 
 	// DomainCacheEntry contains the info and config for a domain
 	DomainCacheEntry struct {
-		mu                          sync.RWMutex
-		info                        *persistence.DomainInfo
-		config                      *persistence.DomainConfig
-		replicationConfig           *persistence.DomainReplicationConfig
-		configVersion               int64
+		mu                sync.RWMutex
+		info              *persistence.DomainInfo
+		config            *persistence.DomainConfig
+		replicationConfig *persistence.DomainReplicationConfig
+		configVersion     int64
+		// failoverVersion is the failover version of domain's active cluster
 		failoverVersion             int64
 		isGlobalDomain              bool
 		failoverNotificationVersion int64
@@ -169,6 +171,7 @@ func NewDomainCache(
 		status:           domainCacheInitialized,
 		shutdownChan:     make(chan struct{}),
 		clusterGroup:     getClusterGroupIdentifier(metadata),
+		clusterMetadata:  metadata,
 		cacheNameToID:    &atomic.Value{},
 		cacheByID:        &atomic.Value{},
 		domainManager:    domainManager,
@@ -459,6 +462,8 @@ func (c *DefaultDomainCache) refreshDomainsLocked() error {
 	for continuePage {
 		ctx, cancel := context.WithTimeout(context.Background(), domainCachePersistenceTimeout)
 		request.NextPageToken = token
+		// TODO: update DB layer to support ActiveClusterNames
+		// Also think about how to support failover overrides and rebalance.
 		response, err := c.domainManager.ListDomains(ctx, request)
 		cancel()
 		if err != nil {
@@ -496,6 +501,7 @@ UpdateLoop:
 			c.logger.Info("Domain notification is not less than than metadata notification version", tag.WorkflowDomainName(domain.GetInfo().Name))
 			break UpdateLoop
 		}
+
 		triggerCallback, nextEntry, err := c.updateIDToDomainCache(newCacheByID, domain.info.ID, domain)
 		if err != nil {
 			return err
@@ -506,6 +512,7 @@ UpdateLoop:
 			metrics.DomainTypeTag(nextEntry.isGlobalDomain),
 			metrics.ClusterGroupTag(c.clusterGroup),
 			metrics.ActiveClusterTag(nextEntry.replicationConfig.ActiveClusterName),
+			metrics.IsActiveActiveDomainTag(nextEntry.replicationConfig.IsActiveActive()),
 		).UpdateGauge(metrics.ActiveClusterGauge, 1)
 
 		c.updateNameToIDCache(newCacheNameToID, nextEntry.info.Name, nextEntry.info.ID)
@@ -625,6 +632,11 @@ func (c *DefaultDomainCache) getDomainByID(
 ) (*DomainCacheEntry, error) {
 
 	var result *DomainCacheEntry
+	defer func() {
+		if result != nil {
+			c.logger.Debugf("GetDomainByID returning domain %s, failoverVersion: %d", result.info.Name, result.failoverVersion)
+		}
+	}()
 	entry, cacheHit := c.cacheByID.Load().(Cache).Get(id).(*DomainCacheEntry)
 	if cacheHit {
 		entry.mu.RLock()
@@ -695,7 +707,7 @@ func (c *DefaultDomainCache) buildEntryFromRecord(
 
 	// this is a shallow copy, but since the record is generated by persistence
 	// and only accessible here, it would be fine
-	return &DomainCacheEntry{
+	entry := &DomainCacheEntry{
 		info:                        record.Info,
 		config:                      record.Config,
 		replicationConfig:           record.ReplicationConfig,
@@ -708,6 +720,8 @@ func (c *DefaultDomainCache) buildEntryFromRecord(
 		notificationVersion:         record.NotificationVersion,
 		initialized:                 true,
 	}
+
+	return entry
 }
 
 func copyResetBinary(bins types.BadBinaries) types.BadBinaries {
@@ -749,7 +763,12 @@ func (entry *DomainCacheEntry) duplicate() *DomainCacheEntry {
 		ActiveClusterName: entry.replicationConfig.ActiveClusterName,
 	}
 	for _, clusterCfg := range entry.replicationConfig.Clusters {
-		result.replicationConfig.Clusters = append(result.replicationConfig.Clusters, &*clusterCfg)
+		c := *clusterCfg
+		result.replicationConfig.Clusters = append(result.replicationConfig.Clusters, &c)
+	}
+	for _, clusterCfg := range entry.replicationConfig.ActiveClusters {
+		c := *clusterCfg
+		result.replicationConfig.ActiveClusters = append(result.replicationConfig.ActiveClusters, &c)
 	}
 	result.configVersion = entry.configVersion
 	result.failoverVersion = entry.failoverVersion
@@ -821,12 +840,24 @@ func (entry *DomainCacheEntry) IsActiveIn(currentCluster string) (bool, error) {
 	}
 
 	domainName := entry.GetInfo().Name
-	activeCluster := entry.GetReplicationConfig().ActiveClusterName
-
 	if entry.IsDomainPendingActive() {
 		return false, errors.NewDomainPendingActiveError(domainName, currentCluster)
 	}
 
+	if len(entry.GetReplicationConfig().ActiveClusters) > 0 {
+		// TODO: optimize this loop by using a map
+		var activeClusters []string
+		for _, cl := range entry.GetReplicationConfig().ActiveClusters {
+			if cl.ClusterName == currentCluster {
+				return true, nil
+			}
+			activeClusters = append(activeClusters, cl.ClusterName)
+		}
+
+		return false, errors.NewDomainNotActiveError(domainName, currentCluster, activeClusters...)
+	}
+
+	activeCluster := entry.GetReplicationConfig().ActiveClusterName
 	if currentCluster != activeCluster {
 		return false, errors.NewDomainNotActiveError(domainName, currentCluster, activeCluster)
 	}

common/cache/domainCache.go

@@ -1179,6 +1179,9 @@ func (d *handlerImpl) updateReplicationConfig(
 		activeClusterUpdated = true
 		config.ActiveClusterName = *updateRequest.ActiveClusterName
 	}
+
+	// TODO: handle active-active case here which would be updateRequest.ActiveClusters != nil.
+
 	return config, clusterUpdated, activeClusterUpdated, nil
 }