Pass config to cadence history service during initialization (#301)

Issue #286

Author: Tamer Eldeeb

cmd/server/server.go

@@ -110,7 +110,7 @@ func (s *server) startService() common.Daemon {
 	case frontendService:
 		daemon = frontend.NewService(&params, frontend.NewConfig())
 	case historyService:
-		daemon = history.NewService(&params)
+		daemon = history.NewService(&params, history.NewConfig(s.cfg.Cassandra.NumHistoryShards))
 	case matchingService:
 		daemon = matching.NewService(&params, matching.NewConfig())
 	}

host/onebox.go

@@ -180,8 +180,8 @@ func (c *cadenceImpl) startHistory(logger bark.Logger, shardMgr persistence.Shar
 		service := service.New(params)
 		var thriftServices []thrift.TChanServer
 		var handler *history.Handler
-		handler, thriftServices = history.NewHandler(service, shardMgr, metadataMgr, visibilityMgr, historyMgr, executionMgrFactory,
-			c.numberOfHistoryShards)
+		handler, thriftServices = history.NewHandler(service, history.NewConfig(c.numberOfHistoryShards), shardMgr, metadataMgr,
+			visibilityMgr, historyMgr, executionMgrFactory)
 		handler.Start(thriftServices)
 		c.historyHandlers = append(c.historyHandlers, handler)
 	}

service/history/handler.go

@@ -52,6 +52,7 @@ type Handler struct {
 	tokenSerializer       common.TaskTokenSerializer
 	startWG               sync.WaitGroup
 	metricsClient         metrics.Client
+	config                *Config
 	service.Service
 }
 
@@ -64,17 +65,17 @@ var (
 )
 
 // NewHandler creates a thrift handler for the history service
-func NewHandler(sVice service.Service, shardManager persistence.ShardManager, metadataMgr persistence.MetadataManager,
-	visibilityMgr persistence.VisibilityManager, historyMgr persistence.HistoryManager,
-	executionMgrFactory persistence.ExecutionManagerFactory, numberOfShards int) (*Handler, []thrift.TChanServer) {
+func NewHandler(sVice service.Service, config *Config, shardManager persistence.ShardManager,
+	metadataMgr persistence.MetadataManager, visibilityMgr persistence.VisibilityManager,
+	historyMgr persistence.HistoryManager, executionMgrFactory persistence.ExecutionManagerFactory) (*Handler, []thrift.TChanServer) {
 	handler := &Handler{
 		Service:             sVice,
+		config:              config,
 		shardManager:        shardManager,
 		metadataMgr:         metadataMgr,
 		historyMgr:          historyMgr,
 		visibilityMgr:       visibilityMgr,
 		executionMgrFactory: executionMgrFactory,
-		numberOfShards:      numberOfShards,
 		tokenSerializer:     common.NewJSONTaskTokenSerializer(),
 	}
 	// prevent us from trying to serve requests before shard controller is started and ready
@@ -102,8 +103,8 @@ func (h *Handler) Start(thriftService []thrift.TChanServer) error {
 		h.Service.GetLogger().Fatalf("Unable to get history service resolver.")
 	}
 	h.hServiceResolver = hServiceResolver
-	h.controller = newShardController(h.numberOfShards, h.GetHostInfo(), hServiceResolver, h.shardManager, h.historyMgr,
-		h.executionMgrFactory, h, h.GetLogger(), h.GetMetricsClient())
+	h.controller = newShardController(h.GetHostInfo(), hServiceResolver, h.shardManager, h.historyMgr,
+		h.executionMgrFactory, h, h.config, h.GetLogger(), h.GetMetricsClient())
 	h.controller.Start()
 	h.metricsClient = h.GetMetricsClient()
 	h.startWG.Done()

service/history/historyBuilder_test.go

@@ -58,7 +58,7 @@ func (s *historyBuilderSuite) SetupTest() {
 	// Have to define our overridden assertions in the test setup. If we did it earlier, s.T() will return nil
 	s.Assertions = require.New(s.T())
 	s.domainID = "history-builder-test-domain"
-	s.msBuilder = newMutableStateBuilder(s.logger)
+	s.msBuilder = newMutableStateBuilder(NewConfig(1), s.logger)
 	s.builder = newHistoryBuilder(s.msBuilder, s.logger)
 }
 

service/history/historyCache.go

@@ -21,8 +21,6 @@
 package history
 
 import (
-	"time"
-
 	workflow "github.com/uber/cadence/.gen/go/shared"
 	"github.com/uber/cadence/common/backoff"
 	"github.com/uber/cadence/common/cache"
@@ -33,12 +31,6 @@ import (
 	"github.com/uber/cadence/common"
 )
 
-const (
-	historyCacheInitialSize               = 256
-	historyCacheMaxSize                   = 1 * 1024
-	historyCacheTTL         time.Duration = time.Hour
-)
-
 type (
 	releaseWorkflowExecutionFunc func()
 
@@ -48,6 +40,7 @@ type (
 		executionManager persistence.ExecutionManager
 		disabled         bool
 		logger           bark.Logger
+		config           *Config
 	}
 )
 
@@ -57,19 +50,21 @@ var (
 	ErrTryLock = &workflow.InternalServiceError{Message: "Failed to acquire lock, backoff and retry"}
 )
 
-func newHistoryCache(maxSize int, shard ShardContext, logger bark.Logger) *historyCache {
+func newHistoryCache(shard ShardContext, logger bark.Logger) *historyCache {
 	opts := &cache.Options{}
-	opts.InitialCapacity = historyCacheInitialSize
-	opts.TTL = historyCacheTTL
+	config := shard.GetConfig()
+	opts.InitialCapacity = config.HistoryCacheInitialSize
+	opts.TTL = config.HistoryCacheTTL
 	opts.Pin = true
 
 	return &historyCache{
-		Cache:            cache.New(maxSize, opts),
+		Cache:            cache.New(config.HistoryCacheMaxSize, opts),
 		shard:            shard,
 		executionManager: shard.GetExecutionManager(),
 		logger: logger.WithFields(bark.Fields{
 			logging.TagWorkflowComponent: logging.TagValueHistoryCacheComponent,
 		}),
+		config: config,
 	}
 }
 

service/history/historyCache_test.go

@@ -66,18 +66,19 @@ func (s *historyCacheSuite) SetupTest() {
 		transferSequenceNumber:    1,
 		executionManager:          s.mockExecutionMgr,
 		shardManager:              &mocks.ShardManager{},
-		rangeSize:                 defaultRangeSize,
 		maxTransferSequenceNumber: 100000,
 		closeCh:                   make(chan int, 100),
+		config:                    NewConfig(1),
 		logger:                    s.logger,
 		metricsClient:             metrics.NewClient(tally.NoopScope, metrics.History),
 	}
-	s.cache = newHistoryCache(historyCacheMaxSize, s.mockShard, s.logger)
+	s.cache = newHistoryCache(s.mockShard, s.logger)
 }
 
 func (s *historyCacheSuite) TestHistoryCachePinning() {
+	s.mockShard.GetConfig().HistoryCacheMaxSize = 2
 	domain := "test_domain"
-	s.cache = newHistoryCache(2, s.mockShard, s.logger)
+	s.cache = newHistoryCache(s.mockShard, s.logger)
 	we := workflow.WorkflowExecution{
 		WorkflowId: common.StringPtr("wf-cache-test"),
 		RunId:      common.StringPtr(uuid.New()),

service/history/historyEngine.go

@@ -89,7 +89,7 @@ func NewEngineWithShardContext(shard ShardContext, metadataMgr persistence.Metad
 	logger := shard.GetLogger()
 	executionManager := shard.GetExecutionManager()
 	historyManager := shard.GetHistoryManager()
-	historyCache := newHistoryCache(historyCacheMaxSize, shard, logger)
+	historyCache := newHistoryCache(shard, logger)
 	domainCache := cache.NewDomainCache(metadataMgr, logger)
 	txProcessor := newTransferQueueProcessor(shard, visibilityMgr, matching, historyClient, historyCache, domainCache)
 	historyEngImpl := &historyEngineImpl{
@@ -166,7 +166,7 @@ func (e *historyEngineImpl) StartWorkflowExecution(startRequest *h.StartWorkflow
 
 	// Generate first decision task event.
 	taskList := request.GetTaskList().GetName()
-	msBuilder := newMutableStateBuilder(e.logger)
+	msBuilder := newMutableStateBuilder(e.shard.GetConfig(), e.logger)
 	startedEvent := msBuilder.AddWorkflowExecutionStartedEvent(domainID, workflowExecution, request)
 	if startedEvent == nil {
 		return nil, &workflow.InternalServiceError{Message: "Failed to add workflow execution started event."}
@@ -1513,7 +1513,7 @@ func (e *historyEngineImpl) getTimerBuilder(we *workflow.WorkflowExecution) *tim
 		logging.TagWorkflowExecutionID: we.GetWorkflowId(),
 		logging.TagWorkflowRunID:       we.GetRunId(),
 	})
-	return newTimerBuilder(lg, common.NewRealTimeSource())
+	return newTimerBuilder(e.shard.GetConfig(), lg, common.NewRealTimeSource())
 }
 
 func (s *shardContextWrapper) UpdateWorkflowExecution(request *persistence.UpdateWorkflowExecutionRequest) error {

service/history/historyEngine2_test.go

@@ -58,6 +58,7 @@ type (
 		mockShardManager   *mocks.ShardManager
 		shardClosedCh      chan int
 		eventSerializer    historyEventSerializer
+		config             *Config
 		logger             bark.Logger
 	}
 )
@@ -75,6 +76,7 @@ func (s *engine2Suite) SetupSuite() {
 	l := log.New()
 	l.Level = log.DebugLevel
 	s.logger = bark.NewLoggerFromLogrus(l)
+	s.config = NewConfig(1)
 }
 
 func (s *engine2Suite) TearDownSuite() {
@@ -101,14 +103,14 @@ func (s *engine2Suite) SetupTest() {
 		executionManager:          s.mockExecutionMgr,
 		historyMgr:                s.mockHistoryMgr,
 		shardManager:              s.mockShardManager,
-		rangeSize:                 defaultRangeSize,
 		maxTransferSequenceNumber: 100000,
 		closeCh:                   s.shardClosedCh,
+		config:                    s.config,
 		logger:                    s.logger,
 		metricsClient:             metrics.NewClient(tally.NoopScope, metrics.History),
 	}
 
-	historyCache := newHistoryCache(historyCacheMaxSize, mockShard, s.logger)
+	historyCache := newHistoryCache(mockShard, s.logger)
 	domainCache := cache.NewDomainCache(s.mockMetadataMgr, s.logger)
 	txProcessor := newTransferQueueProcessor(mockShard, s.mockVisibilityMgr, s.mockMatchingClient, s.mockHistoryClient, historyCache, domainCache)
 	h := &historyEngineImpl{
@@ -605,7 +607,7 @@ func (s *engine2Suite) TestRequestCancelWorkflowExecutionFail() {
 
 func (s *engine2Suite) createExecutionStartedState(we workflow.WorkflowExecution, tl, identity string,
 	startDecision bool) *mutableStateBuilder {
-	msBuilder := newMutableStateBuilder(s.logger)
+	msBuilder := newMutableStateBuilder(s.config, s.logger)
 	addWorkflowExecutionStartedEvent(msBuilder, we, "wType", tl, []byte("input"), 100, 200, identity)
 	scheduleEvent, _ := addDecisionTaskScheduledEvent(msBuilder)
 	if startDecision {
@@ -641,7 +643,7 @@ func (s *engine2Suite) TestRespondDecisionTaskCompletedRecordMarkerDecision() {
 	markerDetails := []byte("marker details")
 	markerName := "marker name"
 
-	msBuilder := newMutableStateBuilder(bark.NewLoggerFromLogrus(log.New()))
+	msBuilder := newMutableStateBuilder(s.config, bark.NewLoggerFromLogrus(log.New()))
 	addWorkflowExecutionStartedEvent(msBuilder, we, "wType", tl, []byte("input"), 100, 200, identity)
 	scheduleEvent, _ := addDecisionTaskScheduledEvent(msBuilder)
 	addDecisionTaskStartedEvent(msBuilder, scheduleEvent.GetEventId(), tl, identity)