Make client timeouts configurable and increase frontend timeout (cade…

…nce-workflow#1258)
zshni · Nov 16, 2018 · eecdbba · eecdbba
1 parent 948ee2a
commit eecdbba
Show file tree

Hide file tree

Showing 5 changed files with 80 additions and 27 deletions.
diff --git a/client/clientfactory.go b/client/clientfactory.go
@@ -27,13 +27,18 @@ import (
 	"github.com/uber/cadence/common"
 	"github.com/uber/cadence/common/membership"
 	"github.com/uber/cadence/common/metrics"
+	"time"
 )
 
 // Factory can be used to create RPC clients for cadence services
 type Factory interface {
 	NewHistoryClient() (history.Client, error)
 	NewMatchingClient() (matching.Client, error)
 	NewFrontendClient() (frontend.Client, error)
+
+	NewHistoryClientWithTimeout(timeout time.Duration) (history.Client, error)
+	NewMatchingClientWithTimeout(timeout time.Duration, longPollTimeout time.Duration) (matching.Client, error)
+	NewFrontendClientWithTimeout(timeout time.Duration) (frontend.Client, error)
 }
 
 type rpcClientFactory struct {
@@ -55,7 +60,19 @@ func NewRPCClientFactory(df common.RPCFactory,
 }
 
 func (cf *rpcClientFactory) NewHistoryClient() (history.Client, error) {
-	client, err := history.NewClient(cf.df, cf.monitor, cf.numberOfHistoryShards)
+	return cf.NewHistoryClientWithTimeout(history.DefaultTimeout)
+}
+
+func (cf *rpcClientFactory) NewMatchingClient() (matching.Client, error) {
+	return cf.NewMatchingClientWithTimeout(matching.DefaultTimeout, matching.DefaultLongPollTimeout)
+}
+
+func (cf *rpcClientFactory) NewFrontendClient() (frontend.Client, error) {
+	return cf.NewFrontendClientWithTimeout(frontend.DefaultTimeout)
+}
+
+func (cf *rpcClientFactory) NewHistoryClientWithTimeout(timeout time.Duration) (history.Client, error) {
+	client, err := history.NewClient(cf.df, cf.monitor, cf.numberOfHistoryShards, timeout)
 	if err != nil {
 		return nil, err
 	}
@@ -65,8 +82,12 @@ func (cf *rpcClientFactory) NewHistoryClient() (history.Client, error) {
 	return client, nil
 }
 
-func (cf *rpcClientFactory) NewMatchingClient() (matching.Client, error) {
-	client, err := matching.NewClient(cf.df, cf.monitor)
+func (cf *rpcClientFactory) NewMatchingClientWithTimeout(
+	timeout time.Duration,
+	longPollTimeout time.Duration,
+) (matching.Client, error) {
+
+	client, err := matching.NewClient(cf.df, cf.monitor, timeout, longPollTimeout)
 	if err != nil {
 		return nil, err
 	}
@@ -76,8 +97,8 @@ func (cf *rpcClientFactory) NewMatchingClient() (matching.Client, error) {
 	return client, nil
 }
 
-func (cf *rpcClientFactory) NewFrontendClient() (frontend.Client, error) {
-	client, err := frontend.NewClient(cf.df, cf.monitor)
+func (cf *rpcClientFactory) NewFrontendClientWithTimeout(timeout time.Duration) (frontend.Client, error) {
+	client, err := frontend.NewClient(cf.df, cf.monitor, timeout)
 	if err != nil {
 		return nil, err
 	}

diff --git a/client/frontend/client.go b/client/frontend/client.go
@@ -36,15 +36,21 @@ import (
 
 var _ Client = (*clientImpl)(nil)
 
+const (
+	// DefaultTimeout is the default timeout used to make calls
+	DefaultTimeout = time.Minute * 3
+)
+
 type clientImpl struct {
 	resolver        membership.ServiceResolver
 	thriftCacheLock sync.RWMutex
 	thriftCache     map[string]workflowserviceclient.Interface
 	rpcFactory      common.RPCFactory
+	timeout         time.Duration
 }
 
 // NewClient creates a new frontend service TChannel client
-func NewClient(d common.RPCFactory, monitor membership.Monitor) (Client, error) {
+func NewClient(d common.RPCFactory, monitor membership.Monitor, timeout time.Duration) (Client, error) {
 	sResolver, err := monitor.GetResolver(common.FrontendServiceName)
 	if err != nil {
 		return nil, err
@@ -54,6 +60,7 @@ func NewClient(d common.RPCFactory, monitor membership.Monitor) (Client, error)
 		rpcFactory:  d,
 		resolver:    sResolver,
 		thriftCache: make(map[string]workflowserviceclient.Interface),
+		timeout:     timeout,
 	}
 	return client, nil
 }
@@ -539,12 +546,10 @@ func (c *clientImpl) UpdateDomain(
 }
 
 func (c *clientImpl) createContext(parent context.Context) (context.Context, context.CancelFunc) {
-	// TODO: make timeout configurable
-	timeout := time.Minute * 1
 	if parent == nil {
-		return context.WithTimeout(context.Background(), timeout)
+		return context.WithTimeout(context.Background(), c.timeout)
 	}
-	return context.WithTimeout(parent, timeout)
+	return context.WithTimeout(parent, c.timeout)
 }
 
 func (c *clientImpl) getRandomHost() (workflowserviceclient.Interface, error) {

diff --git a/client/history/client.go b/client/history/client.go
@@ -35,6 +35,11 @@ import (
 
 var _ Client = (*clientImpl)(nil)
 
+const (
+	// DefaultTimeout is the default timeout used to make calls
+	DefaultTimeout = time.Second * 30
+)
+
 type clientImpl struct {
 	resolver        membership.ServiceResolver
 	tokenSerializer common.TaskTokenSerializer
@@ -43,10 +48,11 @@ type clientImpl struct {
 	thriftCacheLock sync.RWMutex
 	thriftCache     map[string]historyserviceclient.Interface
 	rpcFactory      common.RPCFactory
+	timeout         time.Duration
 }
 
 // NewClient creates a new history service TChannel client
-func NewClient(d common.RPCFactory, monitor membership.Monitor, numberOfShards int) (Client, error) {
+func NewClient(d common.RPCFactory, monitor membership.Monitor, numberOfShards int, timeout time.Duration) (Client, error) {
 	sResolver, err := monitor.GetResolver(common.HistoryServiceName)
 	if err != nil {
 		return nil, err
@@ -58,6 +64,7 @@ func NewClient(d common.RPCFactory, monitor membership.Monitor, numberOfShards i
 		tokenSerializer: common.NewJSONTaskTokenSerializer(),
 		numberOfShards:  numberOfShards,
 		thriftCache:     make(map[string]historyserviceclient.Interface),
+		timeout:         timeout,
 	}
 	return client, nil
 }
@@ -614,12 +621,10 @@ func (c *clientImpl) getHostForRequest(workflowID string) (historyserviceclient.
 }
 
 func (c *clientImpl) createContext(parent context.Context) (context.Context, context.CancelFunc) {
-	// TODO: make timeout configurable
-	timeout := time.Second * 30
 	if parent == nil {
-		return context.WithTimeout(context.Background(), timeout)
+		return context.WithTimeout(context.Background(), c.timeout)
 	}
-	return context.WithTimeout(parent, timeout)
+	return context.WithTimeout(parent, c.timeout)
 }
 
 func (c *clientImpl) getThriftClient(hostPort string) historyserviceclient.Interface {

diff --git a/client/matching/client.go b/client/matching/client.go
@@ -35,24 +35,40 @@ import (
 
 var _ Client = (*clientImpl)(nil)
 
+const (
+	// DefaultTimeout is the default timeout used to make calls
+	DefaultTimeout = time.Minute
+	// DefaultLongPollTimeout is the long poll default timeout used to make calls
+	DefaultLongPollTimeout = time.Minute * 2
+)
+
 type clientImpl struct {
 	resolver        membership.ServiceResolver
 	thriftCacheLock sync.RWMutex
 	thriftCache     map[string]matchingserviceclient.Interface
 	rpcFactory      common.RPCFactory
+	timeout         time.Duration
+	longPollTimeout time.Duration
 }
 
 // NewClient creates a new history service TChannel client
-func NewClient(d common.RPCFactory, monitor membership.Monitor) (Client, error) {
+func NewClient(
+	d common.RPCFactory,
+	monitor membership.Monitor,
+	timeout time.Duration,
+	longPollTimeout time.Duration,
+) (Client, error) {
 	sResolver, err := monitor.GetResolver(common.MatchingServiceName)
 	if err != nil {
 		return nil, err
 	}
 
 	client := &clientImpl{
-		rpcFactory:  d,
-		resolver:    sResolver,
-		thriftCache: make(map[string]matchingserviceclient.Interface),
+		rpcFactory:      d,
+		resolver:        sResolver,
+		thriftCache:     make(map[string]matchingserviceclient.Interface),
+		timeout:         timeout,
+		longPollTimeout: longPollTimeout,
 	}
 	return client, nil
 }
@@ -166,21 +182,17 @@ func (c *clientImpl) getHostForRequest(key string) (matchingserviceclient.Interf
 }
 
 func (c *clientImpl) createContext(parent context.Context) (context.Context, context.CancelFunc) {
-	// TODO: make timeout configurable
-	timeout := time.Minute * 1
 	if parent == nil {
-		return context.WithTimeout(context.Background(), timeout)
+		return context.WithTimeout(context.Background(), c.timeout)
 	}
-	return context.WithTimeout(parent, timeout)
+	return context.WithTimeout(parent, c.timeout)
 }
 
 func (c *clientImpl) createLongPollContext(parent context.Context) (context.Context, context.CancelFunc) {
-	// TODO: make timeout configurable
-	timeout := time.Minute * 2
 	if parent == nil {
-		return context.WithTimeout(context.Background(), timeout)
+		return context.WithTimeout(context.Background(), c.longPollTimeout)
 	}
-	return context.WithTimeout(parent, timeout)
+	return context.WithTimeout(parent, c.longPollTimeout)
 }
 
 func (c *clientImpl) getThriftClient(hostPort string) matchingserviceclient.Interface {

diff --git a/service/worker/service.go b/service/worker/service.go
@@ -28,6 +28,7 @@ import (
 	persistencefactory "github.com/uber/cadence/common/persistence/persistence-factory"
 	"github.com/uber/cadence/common/service"
 	"github.com/uber/cadence/common/service/dynamicconfig"
+	"go.uber.org/cadence/worker"
 )
 
 const (
@@ -93,9 +94,18 @@ func (s *Service) Start() {
 	if s.params.ClusterMetadata.IsGlobalDomainEnabled() {
 		s.startReplicator(params, base, log)
 	}
+
+	frontendClient := s.getFrontendClient(base, log)
+	w := worker.New(frontendClient, SystemWorkflowDomain, SystemTaskList, worker.Options{})
+	if err := w.Start(); err != nil {
+		w.Stop()
+		log.Fatalf("failed to start worker: %v", err)
+	}
+
 	log.Infof("%v started", common.WorkerServiceName)
 
 	<-s.stopC
+	w.Stop()
 	base.Stop()
 }