Async workflow request consumer manager in worker

common/dynamicconfig/constants.go

@@ -1825,6 +1825,12 @@ const (
 	// Default value: false
 	// Allowed filters: N/A
 	EnableESAnalyzer
+	// EnableAsyncWorkflowConsumption decides whether to enable system workers for processing async workflows
+	// KeyName: system.enableAsyncWorkflowConsumption
+	// Value type: Bool
+	// Default value: false
+	// Allowed filters: N/A
+	EnableAsyncWorkflowConsumption
 
 	// EnableStickyQuery indicates if sticky query should be enabled per domain
 	// KeyName: system.enableStickyQuery
@@ -4063,6 +4069,11 @@ var BoolKeys = map[BoolKey]DynamicBool{
 		Description:  "EnableESAnalyzer decides whether to enable system workers for processing ElasticSearch Analyzer",
 		DefaultValue: false,
 	},
+	EnableAsyncWorkflowConsumption: DynamicBool{
+		KeyName:      "worker.enableAsyncWorkflowConsumption",
+		Description:  "EnableAsyncWorkflowConsumption decides whether to enable async workflows",
+		DefaultValue: false,
+	},
 	EnableStickyQuery: DynamicBool{
 		KeyName:      "system.enableStickyQuery",
 		Filters:      []Filter{DomainName},

common/log/tag/values.go

@@ -133,6 +133,7 @@ var (
 	ComponentShardScanner               = component("shardscanner-scanner")
 	ComponentShardFixer                 = component("shardscanner-fixer")
 	ComponentPinotVisibilityManager     = component("pinot-visibility-manager")
+	ComponentAsyncWFConsumptionManager  = component("async-wf-consumption-manager")
 )
 
 // Pre-defined values for TagSysLifecycle

common/messaging/noop_consumer.go

@@ -0,0 +1,40 @@
+// 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 messaging
+
+type noopConsumer struct{}
+
+// NewNoopProducer returns a no-op message consumer
+func NewNoopConsumer() Consumer {
+	return &noopConsumer{}
+}
+
+func (c *noopConsumer) Start() error {
+	return nil
+}
+
+func (c *noopConsumer) Stop() {}
+
+func (c *noopConsumer) Messages() <-chan Message {
+	return nil
+}

common/metrics/defs.go

@@ -1341,6 +1341,8 @@ const (
 	CheckDataCorruptionWorkflowScope
 	// ESAnalyzerScope is scope used by ElasticSearch Analyzer (esanalyzer) workflow
 	ESAnalyzerScope
+	// AsyncWorkflowConsumerScope is scope used by async workflow consumer
+	AsyncWorkflowConsumerScope
 
 	NumWorkerScopes
 )
@@ -2565,6 +2567,7 @@ const (
 	ESAnalyzerNumStuckWorkflowsRefreshed
 	ESAnalyzerNumStuckWorkflowsFailedToRefresh
 	ESAnalyzerNumLongRunningWorkflows
+	AsyncWorkflowConsumerCount
 
 	NumWorkerMetrics
 )
@@ -3183,6 +3186,7 @@ var MetricDefs = map[ServiceIdx]map[int]metricDefinition{
 		ESAnalyzerNumStuckWorkflowsRefreshed:          {metricName: "es_analyzer_num_stuck_workflows_refreshed", metricType: Counter},
 		ESAnalyzerNumStuckWorkflowsFailedToRefresh:    {metricName: "es_analyzer_num_stuck_workflows_failed_to_refresh", metricType: Counter},
 		ESAnalyzerNumLongRunningWorkflows:             {metricName: "es_analyzer_num_long_running_workflows", metricType: Counter},
+		AsyncWorkflowConsumerCount:                    {metricName: "async_workflow_consumer_count", metricType: Gauge},
 	},
 }
 

service/frontend/api/producer_manager.go

@@ -37,6 +37,8 @@ import (
 )
 
 type (
+	// ProducerManager is used to create a producer for a domain.
+	// Producer is used for Async APIs such as StartWorkflowExecutionAsync
 	ProducerManager interface {
 		GetProducerByDomain(domain string) (messaging.Producer, error)
 	}

service/worker/asyncworkflow/async_workflow_consumer_manager.go

@@ -0,0 +1,213 @@
+// 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 asyncworkflow
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/uber/cadence/common"
+	"github.com/uber/cadence/common/asyncworkflow/queue"
+	"github.com/uber/cadence/common/asyncworkflow/queue/provider"
+	"github.com/uber/cadence/common/cache"
+	"github.com/uber/cadence/common/clock"
+	"github.com/uber/cadence/common/log"
+	"github.com/uber/cadence/common/log/tag"
+	"github.com/uber/cadence/common/messaging"
+	"github.com/uber/cadence/common/metrics"
+	"github.com/uber/cadence/common/types"
+)
+
+const (
+	defaultRefreshInterval = 5 * time.Minute
+	defaultShutdownTimeout = 5 * time.Second
+)
+
+type ConsumerManagerOptions func(*ConsumerManager)
+
+func WithTimeSource(timeSrc clock.TimeSource) ConsumerManagerOptions {
+	return func(c *ConsumerManager) {
+		c.timeSrc = timeSrc
+	}
+}
+
+func NewConsumerManager(
+	logger log.Logger,
+	metricsClient metrics.Client,
+	domainCache cache.DomainCache,
+	queueProvider queue.Provider,
+	options ...ConsumerManagerOptions,
+) *ConsumerManager {
+	ctx, cancel := context.WithCancel(context.Background())
+	cm := &ConsumerManager{
+		logger:          logger.WithTags(tag.ComponentAsyncWFConsumptionManager),
+		metricsClient:   metricsClient,
+		domainCache:     domainCache,
+		queueProvider:   queueProvider,
+		refreshInterval: defaultRefreshInterval,
+		shutdownTimeout: defaultShutdownTimeout,
+		ctx:             ctx,
+		cancelFn:        cancel,
+		activeConsumers: make(map[string]messaging.Consumer),
+		timeSrc:         clock.NewRealTimeSource(),
+	}
+
+	for _, opt := range options {
+		opt(cm)
+	}
+	return cm
+}
+
+type ConsumerManager struct {
+	logger          log.Logger
+	metricsClient   metrics.Client
+	timeSrc         clock.TimeSource
+	domainCache     cache.DomainCache
+	queueProvider   queue.Provider
+	refreshInterval time.Duration
+	shutdownTimeout time.Duration
+	ctx             context.Context
+	cancelFn        context.CancelFunc
+	wg              sync.WaitGroup
+	activeConsumers map[string]messaging.Consumer
+}
+
+func (c *ConsumerManager) Start() {
+	c.logger.Info("Starting ConsumerManager")
+	c.wg.Add(1)
+	go c.run()
+}
+
+func (c *ConsumerManager) Stop() {
+	c.logger.Info("Stopping ConsumerManager")
+	c.cancelFn()
+	c.wg.Wait()
+	if !common.AwaitWaitGroup(&c.wg, c.shutdownTimeout) {
+		c.logger.Warn("ConsumerManager timed out on shutdown", tag.Dynamic("timeout", c.shutdownTimeout))
+		return
+	}
+
+	for qID, consumer := range c.activeConsumers {
+		consumer.Stop()
+		c.logger.Info("Stopped consumer", tag.Dynamic("queue-id", qID))
+	}
+
+	c.logger.Info("Stopped ConsumerManager")
+}
+
+func (c *ConsumerManager) run() {
+	defer c.wg.Done()
+
+	timer := c.timeSrc.NewTimer(c.refreshInterval)
+	defer timer.Stop()
+	c.logger.Info("ConsumerManager background loop started", tag.Dynamic("refresh-interval", c.refreshInterval))
+
+	c.refreshConsumers()
+
+	for {
+		select {
+		case <-timer.Chan():
+			c.refreshConsumers()
+		case <-c.ctx.Done():
+			c.logger.Info("ConsumerManager background loop stopped because context is done")
+			return
+		}
+	}
+}
+
+func (c *ConsumerManager) refreshConsumers() {
+	domains := c.domainCache.GetAllDomain()
+	c.logger.Info("Refreshing consumers", tag.Dynamic("domain-count", len(domains)), tag.Dynamic("consumer-count", len(c.activeConsumers)))
+	refCounts := make(map[string]int, len(c.activeConsumers))
+
+	for _, domain := range domains {
+		select {
+		default:
+		case <-c.ctx.Done():
+			c.logger.Info("refreshConsumers is terminating because context is done")
+			return
+		}
+
+		// domain config is not set or async workflow config is not set
+		if domain.GetConfig() == nil || domain.GetConfig().AsyncWorkflowConfig == (types.AsyncWorkflowConfiguration{}) {
+			continue
+		}
+
+		cfg := domain.GetConfig().AsyncWorkflowConfig
+		queue, err := c.getQueue(cfg)
+		if err != nil {
+			c.logger.Error("Failed to get queue", tag.Error(err), tag.WorkflowDomainName(domain.GetInfo().Name))
+			continue
+		}
+
+		if !cfg.Enabled {
+			// lready running active consumers for such queues will be stopped in the next loop
+			continue
+		}
+
+		// async workflow config is enabled. check if consumer is already running
+		if c.activeConsumers[queue.ID()] != nil {
+			c.logger.Debug("Consumer already running", tag.WorkflowDomainName(domain.GetInfo().Name), tag.Dynamic("queue-id", queue.ID()))
+			refCounts[queue.ID()]++
+			continue
+		}
+
+		c.logger.Info("Starting consumer", tag.WorkflowDomainName(domain.GetInfo().Name), tag.Dynamic("queue-id", queue.ID()))
+		consumer, err := queue.CreateConsumer(&provider.Params{
+			Logger:        c.logger,
+			MetricsClient: c.metricsClient,
+		})
+		if err != nil {
+			c.logger.Error("Failed to create consumer", tag.Error(err), tag.WorkflowDomainName(domain.GetInfo().Name), tag.Dynamic("queue-id", queue.ID()))
+			continue
+		}
+
+		c.activeConsumers[queue.ID()] = consumer
+		refCounts[queue.ID()]++
+		c.logger.Info("Created consumer", tag.WorkflowDomainName(domain.GetInfo().Name), tag.Dynamic("queue-id", queue.ID()))
+	}
+
+	// stop consumers that are not needed
+	for qID, consumer := range c.activeConsumers {
+		if refCounts[qID] > 0 {
+			continue
+		}
+
+		c.logger.Info("Stopping consumer because it's not needed", tag.Dynamic("queue-id", qID))
+		consumer.Stop()
+		delete(c.activeConsumers, qID)
+		c.logger.Info("Stopped consumer", tag.Dynamic("queue-id", qID))
+	}
+
+	c.logger.Info("Refreshed consumers", tag.Dynamic("consumer-count", len(c.activeConsumers)))
+	c.metricsClient.Scope(metrics.AsyncWorkflowConsumerScope).UpdateGauge(metrics.AsyncWorkflowConsumerCount, float64(len(c.activeConsumers)))
+}
+
+func (c *ConsumerManager) getQueue(cfg types.AsyncWorkflowConfiguration) (provider.Queue, error) {
+	if cfg.PredefinedQueueName != "" {
+		return c.queueProvider.GetPredefinedQueue(cfg.PredefinedQueueName)
+	}
+
+	return c.queueProvider.GetQueue(cfg.QueueType, cfg.QueueConfig)
+}