Add task queue to worker metrics (#785)

* Add task queue to worker metrics

Fixes #784

* Move task queue tag to a higher level

internal/common/metrics/tags.go

@@ -67,6 +67,13 @@ func LocalActivityTags(workflowType, activityType string) map[string]string {
 	}
 }
 
+// TaskQueueTags returns a set of tags for a task queue.
+func TaskQueueTags(taskQueue string) map[string]string {
+	return map[string]string{
+		TaskQueueTagName: taskQueue,
+	}
+}
+
 // WorkerTags returns a set of tags for workers.
 func WorkerTags(workerType string) map[string]string {
 	return map[string]string{

internal/internal_worker.go

@@ -1303,7 +1303,7 @@ func NewAggregatedWorker(client *WorkflowClient, taskQueue string, options Worke
 		ConcurrentWorkflowTaskExecutionSize:   options.MaxConcurrentWorkflowTaskExecutionSize,
 		MaxConcurrentWorkflowTaskQueuePollers: options.MaxConcurrentWorkflowTaskPollers,
 		Identity:                              client.identity,
-		MetricsHandler:                        client.metricsHandler,
+		MetricsHandler:                        client.metricsHandler.WithTags(metrics.TaskQueueTags(taskQueue)),
 		Logger:                                client.logger,
 		EnableLoggingInReplay:                 options.EnableLoggingInReplay,
 		UserContext:                           backgroundActivityContext,

test/integration_test.go

@@ -1768,7 +1768,11 @@ func (ts *IntegrationTestSuite) TestSlotsAvailableCounter() {
 		// Try for two seconds
 		var lastCount float64
 		for start := time.Now(); time.Since(start) <= 2*time.Second; {
-			lastCount = ts.metricGauge(metrics.WorkerTaskSlotsAvailable, "worker_type", "ActivityWorker")
+			lastCount = ts.metricGauge(
+				metrics.WorkerTaskSlotsAvailable,
+				"worker_type", "ActivityWorker",
+				"task_queue", ts.taskQueueName,
+			)
 			if lastCount == expected {
 				return
 			}