adding more client metrics (#207)

* adding more client metrics

* make tagScope take tagName, tagValue

Author: yiminc

client.go

@@ -201,9 +201,7 @@ func NewClient(service m.TChanWorkflowService, domain string, options *ClientOpt
 	if options != nil {
 		metricScope = options.MetricsScope
 	}
-	if metricScope == nil {
-		metricScope = tally.NoopScope
-	}
+	metricScope = tagScope(metricScope, tagDomain, domain)
 	return &workflowClient{
 		workflowService: metrics.NewWorkflowServiceWrapper(service, metricScope),
 		domain:          domain,
@@ -224,9 +222,7 @@ func NewDomainClient(service m.TChanWorkflowService, options *ClientOptions) Dom
 	if options != nil {
 		metricScope = options.MetricsScope
 	}
-	if metricScope == nil {
-		metricScope = tally.NoopScope
-	}
+	metricScope = tagScope(metricScope, tagDomain, "domain-client")
 	return &domainClient{
 		workflowService: metrics.NewWorkflowServiceWrapper(service, metricScope),
 		metricsScope:    metricScope,

common/metrics/constants.go

@@ -64,6 +64,9 @@ const (
 
 	UnhandledSignalsCounter = "unhandled-signals"
 
+	WorkerStartCounter = "worker-start"
+	PollerStartCounter = "poller-start"
+
 	CadenceRequest        = "cadence-request"
 	CadenceError          = "cadence-error"
 	CadenceLatency        = "cadence-latency"

internal_worker.go

@@ -233,7 +233,9 @@ func newWorkflowTaskWorkerInternal(
 		taskWorker:                 poller,
 		identity:                   params.Identity,
 		workerType:                 "DecisionWorker"},
-		params.Logger)
+		params.Logger,
+		params.MetricsScope,
+	)
 
 	return &workflowWorker{
 		executionParameters: params,
@@ -313,6 +315,7 @@ func newActivityTaskWorker(
 			workerType:                 "ActivityWorker",
 		},
 		workerParams.Logger,
+		workerParams.MetricsScope,
 	)
 
 	return &activityWorker{
@@ -1026,10 +1029,7 @@ func newAggregatedWorker(
 	}
 
 	ensureRequiredParams(&workerParams)
-	tags := map[string]string{
-		tagDomain: domain,
-	}
-	workerParams.MetricsScope = workerParams.MetricsScope.Tagged(tags)
+	workerParams.MetricsScope = tagScope(workerParams.MetricsScope, tagDomain, domain)
 	workerParams.Logger = workerParams.Logger.With(
 		zapcore.Field{Key: tagDomain, Type: zapcore.StringType, String: domain},
 		zapcore.Field{Key: tagTaskList, Type: zapcore.StringType, String: taskList},
@@ -1083,6 +1083,14 @@ func newAggregatedWorker(
 	}
 }
 
+func tagScope(metricsScope tally.Scope, tagName, tagValue string) tally.Scope {
+	if metricsScope == nil {
+		metricsScope = tally.NoopScope
+	}
+	tagsMap := map[string]string{tagName: tagValue}
+	return metricsScope.Tagged(tagsMap)
+}
+
 func processTestTags(wOptions *WorkerOptions, ep *workerExecutionParameters) {
 	testTags := getTestTags(wOptions.BackgroundActivityContext)
 	if testTags != nil {

internal_worker_base.go

@@ -29,6 +29,7 @@ import (
 
 	"github.com/uber-go/tally"
 	"go.uber.org/cadence/common/backoff"
+	"go.uber.org/cadence/common/metrics"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 	"golang.org/x/time/rate"
@@ -102,6 +103,7 @@ type (
 		limiterContextCancel func()
 		retrier              *backoff.ConcurrentRetrier // Service errors back off retrier
 		logger               *zap.Logger
+		metricsScope         tally.Scope
 
 		pollerRequestCh chan struct{}
 		taskQueueCh     chan interface{}
@@ -115,16 +117,16 @@ func createPollRetryPolicy() backoff.RetryPolicy {
 	return policy
 }
 
-func newBaseWorker(options baseWorkerOptions, logger *zap.Logger) *baseWorker {
+func newBaseWorker(options baseWorkerOptions, logger *zap.Logger, metricsScope tally.Scope) *baseWorker {
 	ctx, cancel := context.WithCancel(context.Background())
 	return &baseWorker{
-		options:     options,
-		shutdownCh:  make(chan struct{}),
-		pollLimiter: rate.NewLimiter(rate.Every(time.Millisecond*100), 100),
-		taskLimiter: rate.NewLimiter(rate.Every(options.maxTaskRateRefreshDuration), options.maxTaskRate),
-		retrier:     backoff.NewConcurrentRetrier(pollOperationRetryPolicy),
-		logger:      logger.With(zapcore.Field{Key: tagWorkerType, Type: zapcore.StringType, String: options.workerType}),
-
+		options:         options,
+		shutdownCh:      make(chan struct{}),
+		pollLimiter:     rate.NewLimiter(rate.Every(time.Millisecond*100), 100),
+		taskLimiter:     rate.NewLimiter(rate.Every(options.maxTaskRateRefreshDuration), options.maxTaskRate),
+		retrier:         backoff.NewConcurrentRetrier(pollOperationRetryPolicy),
+		logger:          logger.With(zapcore.Field{Key: tagWorkerType, Type: zapcore.StringType, String: options.workerType}),
+		metricsScope:    tagScope(metricsScope, tagWorkerType, options.workerType),
 		pollerRequestCh: make(chan struct{}, options.maxConcurrentTask),
 		taskQueueCh:     make(chan interface{}), // no buffer, so poller only able to poll new task after previous is dispatched.
 
@@ -139,6 +141,8 @@ func (bw *baseWorker) Start() {
 		return
 	}
 
+	bw.metricsScope.Counter(metrics.WorkerStartCounter).Inc(1)
+
 	for i := 0; i < bw.options.pollerCount; i++ {
 		bw.shutdownWG.Add(1)
 		go bw.runPoller()
@@ -168,6 +172,7 @@ func (bw *baseWorker) isShutdown() bool {
 
 func (bw *baseWorker) runPoller() {
 	defer bw.shutdownWG.Done()
+	bw.metricsScope.Counter(metrics.PollerStartCounter).Inc(1)
 
 	for {
 		select {