shardController: revisit shard shutdown order, add latency metrics (#314)

Author: venkat1109

common/metrics/defs.go

@@ -488,6 +488,9 @@ const (
 	ShardItemRemovedCounter
 	MembershipChangedCounter
 	NumShardsGauge
+	GetEngineForShardErrorCounter
+	GetEngineForShardLatency
+	RemoveEngineForShardLatency
 )
 
 // Matching metrics enum
@@ -558,6 +561,9 @@ var MetricDefs = map[ServiceIdx]map[int]metricDefinition{
 		ShardItemRemovedCounter:                   {metricName: "sharditem-removed-count", metricType: Counter},
 		MembershipChangedCounter:                  {metricName: "membership-changed-count", metricType: Counter},
 		NumShardsGauge:                            {metricName: "numshards-gauge", metricType: Gauge},
+		GetEngineForShardErrorCounter:             {metricName: "get-engine-for-shard-errors", metricType: Counter},
+		GetEngineForShardLatency:                  {metricName: "get-engine-for-shard-latency", metricType: Timer},
+		RemoveEngineForShardLatency:               {metricName: "remove-engine-for-shard-latency", metricType: Timer},
 	},
 	Matching: {
 		PollSuccessCounter:          {metricName: "poll.success"},

glide.lock

@@ -20,9 +20,6 @@ import:
 - package: github.com/dgryski/go-farm
 - package: github.com/emirpasic/gods
 - package: github.com/davecgh/go-spew
-- package: github.com/uber-go/timer
-  subpackages:
-  - twheel
 - package: github.com/urfave/cli
 - package: gopkg.in/yaml.v2
 - package: gopkg.in/validator.v2

glide.yaml

@@ -167,15 +167,18 @@ func (c *shardController) GetEngine(workflowID string) (Engine, error) {
 }
 
 func (c *shardController) getEngineForShard(shardID int) (Engine, error) {
+	sw := c.metricsClient.StartTimer(metrics.HistoryShardControllerScope, metrics.GetEngineForShardLatency)
+	defer sw.Stop()
 	item, err := c.getOrCreateHistoryShardItem(shardID)
 	if err != nil {
 		return nil, err
 	}
-
 	return item.getOrCreateEngine(c.shardClosedCh)
 }
 
 func (c *shardController) removeEngineForShard(shardID int) {
+	sw := c.metricsClient.StartTimer(metrics.HistoryShardControllerScope, metrics.RemoveEngineForShardLatency)
+	defer sw.Stop()
 	item, _ := c.removeHistoryShardItem(shardID)
 	if item != nil {
 		item.stopEngine()
@@ -297,6 +300,7 @@ AcquireLoop:
 		if info.Identity() == c.host.Identity() {
 			_, err1 := c.getEngineForShard(shardID)
 			if err1 != nil {
+				c.metricsClient.IncCounter(metrics.HistoryShardControllerScope, metrics.GetEngineForShardErrorCounter)
 				logging.LogOperationFailedEvent(c.logger, fmt.Sprintf("Unable to create history shard engine: %v", shardID),
 					err1)
 				continue AcquireLoop
@@ -381,16 +385,19 @@ func (i *historyShardsItem) stopEngine() {
 	i.Lock()
 	defer i.Unlock()
 
-	if i.executionMgr != nil {
-		i.executionMgr.Close()
-	}
-
 	if i.engine != nil {
 		logging.LogShardEngineStoppingEvent(i.logger, i.host.Identity(), i.shardID)
 		i.engine.Stop()
 		i.engine = nil
 		logging.LogShardEngineStoppedEvent(i.logger, i.host.Identity(), i.shardID)
 	}
+
+	// Shutting down executionMgr will close all connections
+	// to cassandra for this engine. So, make sure to
+	// close executionMgr only after stopping the engine
+	if i.executionMgr != nil {
+		i.executionMgr.Close()
+	}
 }
 
 func isShardOwnershiptLostError(err error) bool {