[CDNC-3578] Worklfow start metric (cadence-workflow#5289)

* add workflow start metric

* fix lint

* remove logs and add workflow start scope

* fix lint second attempt

* emit metric in transferStandbyTaskExecutor

Author: ketsiambaku

common/metrics/defs.go

@@ -2292,6 +2292,7 @@ const (
 	HistoryFailoverCallbackCount
 	WorkflowVersionCount
 	WorkflowTypeCount
+	WorkflowStartedCount
 	LargeHistoryBlobCount
 	LargeHistoryEventCount
 	LargeHistorySizeCount
@@ -2899,6 +2900,7 @@ var MetricDefs = map[ServiceIdx]map[int]metricDefinition{
 		ReplicationTasksCount:                                        {metricName: "replication_tasks_count", metricType: Timer},
 		WorkflowVersionCount:                                         {metricName: "workflow_version_count", metricType: Gauge},
 		WorkflowTypeCount:                                            {metricName: "workflow_type_count", metricType: Gauge},
+		WorkflowStartedCount:                                         {metricName: "workflow_started_count", metricType: Counter},
 		LargeHistoryBlobCount:                                        {metricName: "large_history_blob_count", metricType: Counter},
 		LargeHistoryEventCount:                                       {metricName: "large_history_event_count", metricType: Counter},
 		LargeHistorySizeCount:                                        {metricName: "large_history_size_count", metricType: Counter},

service/history/task/transfer_active_task_executor.go

@@ -952,6 +952,9 @@ func (t *transferActiveTaskExecutor) processRecordWorkflowStartedOrUpsertHelper(
 	recordStart bool,
 ) (retError error) {
 
+	workflowStartedScope := t.metricsClient.Scope(metrics.TransferActiveTaskRecordWorkflowStartedScope,
+		metrics.DomainTag(task.DomainID))
+
 	wfContext, release, err := t.executionCache.GetOrCreateWorkflowExecutionWithTimeout(
 		task.DomainID,
 		getWorkflowExecution(task),
@@ -1011,6 +1014,7 @@ func (t *transferActiveTaskExecutor) processRecordWorkflowStartedOrUpsertHelper(
 	release(nil)
 
 	if recordStart {
+		workflowStartedScope.IncCounter(metrics.WorkflowStartedCount)
 		return t.recordWorkflowStarted(
 			ctx,
 			task.DomainID,

service/history/task/transfer_standby_task_executor.go

@@ -455,6 +455,9 @@ func (t *transferStandbyTaskExecutor) processRecordWorkflowStartedOrUpsertHelper
 	isRecordStart bool,
 ) error {
 
+	workflowStartedScope := t.metricsClient.Scope(metrics.TransferStandbyTaskRecordWorkflowStartedScope,
+		metrics.DomainTag(transferTask.DomainID))
+
 	// verify task version for RecordWorkflowStarted.
 	// upsert doesn't require verifyTask, because it is just a sync of mutableState.
 	if isRecordStart {
@@ -489,6 +492,7 @@ func (t *transferStandbyTaskExecutor) processRecordWorkflowStartedOrUpsertHelper
 	numClusters := (int16)(len(domainEntry.GetReplicationConfig().Clusters))
 
 	if isRecordStart {
+		workflowStartedScope.IncCounter(metrics.WorkflowStartedCount)
 		return t.recordWorkflowStarted(
 			ctx,
 			transferTask.DomainID,