Fix replication task execution latency (#3671)

* Fix replication task execution latency * Fix emit latency
temporalio · Nov 28, 2022 · 2105645 · 2105645
1 parent 82cbe64
commit 2105645
Show file tree

Hide file tree

Showing 15 changed files with 73 additions and 43 deletions.
diff --git a/common/authorization/interceptor.go b/common/authorization/interceptor.go
@@ -159,7 +159,9 @@ func (a *interceptor) authorize(
 	callTarget *CallTarget,
 	metricsHandler metrics.MetricsHandler) (Result, error) {
 	startTime := time.Now().UTC()
-	defer metricsHandler.Timer(metrics.ServiceAuthorizationLatency.GetMetricName()).Record(time.Since(startTime))
+	defer func() {
+		metricsHandler.Timer(metrics.ServiceAuthorizationLatency.GetMetricName()).Record(time.Since(startTime))
+	}()
 	return a.authorizer.Authorize(ctx, claims, callTarget)
 }
 

diff --git a/common/namespace/registry.go b/common/namespace/registry.go
@@ -617,7 +617,9 @@ func (r *registry) triggerNamespaceChangePrepareCallback(
 	prepareCallbacks []PrepareCallbackFn,
 ) {
 	startTime := time.Now().UTC()
-	defer r.metricsHandler.Timer(metrics.NamespaceCachePrepareCallbacksLatency.GetMetricName()).Record(time.Since(startTime))
+	defer func() {
+		r.metricsHandler.Timer(metrics.NamespaceCachePrepareCallbacksLatency.GetMetricName()).Record(time.Since(startTime))
+	}()
 
 	for _, prepareCallback := range prepareCallbacks {
 		prepareCallback()
@@ -630,7 +632,9 @@ func (r *registry) triggerNamespaceChangeCallback(
 	newNamespaces []*Namespace,
 ) {
 	startTime := time.Now().UTC()
-	defer r.metricsHandler.Timer(metrics.NamespaceCacheCallbacksLatency.GetMetricName()).Record(time.Since(startTime))
+	defer func() {
+		r.metricsHandler.Timer(metrics.NamespaceCacheCallbacksLatency.GetMetricName()).Record(time.Since(startTime))
+	}()
 
 	for _, callback := range callbacks {
 		callback(oldNamespaces, newNamespaces)

diff --git a/service/frontend/adminHandler.go b/service/frontend/adminHandler.go
@@ -653,7 +653,9 @@ func (adh *AdminHandler) GetWorkflowExecutionRawHistoryV2(ctx context.Context, r
 	defer log.CapturePanic(adh.logger, &retError)
 
 	taggedMetricsHandler, startTime := adh.startRequestProfile(metrics.AdminGetWorkflowExecutionRawHistoryV2Scope)
-	defer taggedMetricsHandler.Timer(metrics.ServiceLatency.GetMetricName()).Record(time.Since(startTime))
+	defer func() {
+		taggedMetricsHandler.Timer(metrics.ServiceLatency.GetMetricName()).Record(time.Since(startTime))
+	}()
 
 	if err := adh.validateGetWorkflowExecutionRawHistoryV2Request(
 		request,

diff --git a/service/frontend/operator_handler.go b/service/frontend/operator_handler.go
@@ -154,7 +154,7 @@ func (h *OperatorHandlerImpl) AddSearchAttributes(ctx context.Context, request *
 	defer log.CapturePanic(h.logger, &retError)
 
 	scope, startTime := h.startRequestProfile(metrics.OperatorAddSearchAttributesScope)
-	defer scope.Timer(metrics.ServiceLatency.GetMetricName()).Record(time.Since(startTime))
+	defer func() { scope.Timer(metrics.ServiceLatency.GetMetricName()).Record(time.Since(startTime)) }()
 
 	// validate request
 	if request == nil {
@@ -294,7 +294,7 @@ func (h *OperatorHandlerImpl) DeleteNamespace(ctx context.Context, request *oper
 	defer log.CapturePanic(h.logger, &retError)
 
 	scope, startTime := h.startRequestProfile(metrics.OperatorDeleteNamespaceScope)
-	defer scope.Timer(metrics.ServiceLatency.GetMetricName()).Record(time.Since(startTime))
+	defer func() { scope.Timer(metrics.ServiceLatency.GetMetricName()).Record(time.Since(startTime)) }()
 
 	// validate request
 	if request == nil {
@@ -353,7 +353,7 @@ func (h *OperatorHandlerImpl) AddOrUpdateRemoteCluster(
 ) (_ *operatorservice.AddOrUpdateRemoteClusterResponse, retError error) {
 	defer log.CapturePanic(h.logger, &retError)
 	scope, startTime := h.startRequestProfile(metrics.OperatorAddOrUpdateRemoteClusterScope)
-	defer scope.Timer(metrics.ServiceLatency.GetMetricName()).Record(time.Since(startTime))
+	defer func() { scope.Timer(metrics.ServiceLatency.GetMetricName()).Record(time.Since(startTime)) }()
 
 	adminClient := h.clientFactory.NewRemoteAdminClientWithTimeout(
 		request.GetFrontendAddress(),
@@ -423,7 +423,7 @@ func (h *OperatorHandlerImpl) RemoveRemoteCluster(
 ) (_ *operatorservice.RemoveRemoteClusterResponse, retError error) {
 	defer log.CapturePanic(h.logger, &retError)
 	scope, startTime := h.startRequestProfile(metrics.OperatorRemoveRemoteClusterScope)
-	defer scope.Timer(metrics.ServiceLatency.GetMetricName()).Record(time.Since(startTime))
+	defer func() { scope.Timer(metrics.ServiceLatency.GetMetricName()).Record(time.Since(startTime)) }()
 
 	if err := h.clusterMetadataManager.DeleteClusterMetadata(
 		ctx,
@@ -441,7 +441,7 @@ func (h *OperatorHandlerImpl) ListClusters(
 ) (_ *operatorservice.ListClustersResponse, retError error) {
 	defer log.CapturePanic(h.logger, &retError)
 	scope, startTime := h.startRequestProfile(metrics.OperatorListClustersScope)
-	defer scope.Timer(metrics.ServiceLatency.GetMetricName()).Record(time.Since(startTime))
+	defer func() { scope.Timer(metrics.ServiceLatency.GetMetricName()).Record(time.Since(startTime)) }()
 
 	if request == nil {
 		return nil, errRequestNotSet

diff --git a/service/frontend/versionChecker.go b/service/frontend/versionChecker.go
@@ -111,7 +111,9 @@ func (vc *VersionChecker) performVersionCheck(
 	ctx context.Context,
 ) {
 	startTime := time.Now().UTC()
-	defer vc.metricsHandler.Timer(metrics.VersionCheckLatency.GetMetricName()).Record(time.Since(startTime))
+	defer func() {
+		vc.metricsHandler.Timer(metrics.VersionCheckLatency.GetMetricName()).Record(time.Since(startTime))
+	}()
 	metadata, err := vc.clusterMetadataManager.GetCurrentClusterMetadata(ctx)
 	if err != nil {
 		vc.metricsHandler.Counter(metrics.VersionCheckFailedCount.GetMetricName()).Record(1)

diff --git a/service/history/api/queryworkflow/api.go b/service/history/api/queryworkflow/api.go
@@ -154,7 +154,7 @@ func Invoke(
 	// If we get here it means query could not be dispatched through matching directly, so it must block
 	// until either an result has been obtained on a workflow task response or until it is safe to dispatch directly through matching.
 	startTime := time.Now().UTC()
-	defer scope.Timer(metrics.WorkflowTaskQueryLatency.GetMetricName()).Record(time.Since(startTime))
+	defer func() { scope.Timer(metrics.WorkflowTaskQueryLatency.GetMetricName()).Record(time.Since(startTime)) }()
 
 	queryReg := mutableState.GetQueryRegistry()
 	if len(queryReg.GetBufferedIDs()) >= shard.GetConfig().MaxBufferedQueryCount() {
@@ -229,7 +229,7 @@ func queryDirectlyThroughMatching(
 ) (*historyservice.QueryWorkflowResponse, error) {
 
 	startTime := time.Now().UTC()
-	defer scope.Timer(metrics.DirectQueryDispatchLatency.GetMetricName()).Record(time.Since(startTime))
+	defer func() { scope.Timer(metrics.DirectQueryDispatchLatency.GetMetricName()).Record(time.Since(startTime)) }()
 
 	if msResp.GetIsStickyTaskQueueEnabled() &&
 		len(msResp.GetStickyTaskQueue().GetName()) != 0 &&

diff --git a/service/history/events/cache.go b/service/history/events/cache.go
@@ -114,7 +114,7 @@ func (e *CacheImpl) GetEvent(ctx context.Context, key EventKey, firstEventID int
 	handler := e.metricsHandler.WithTags(metrics.OperationTag(metrics.EventsCacheGetEventScope))
 	handler.Counter(metrics.CacheRequests.GetMetricName()).Record(1)
 	startTime := time.Now().UTC()
-	defer handler.Timer(metrics.CacheLatency.GetMetricName()).Record(time.Since(startTime))
+	defer func() { handler.Timer(metrics.CacheLatency.GetMetricName()).Record(time.Since(startTime)) }()
 
 	validKey := e.validateKey(key)
 
@@ -150,7 +150,7 @@ func (e *CacheImpl) PutEvent(key EventKey, event *historypb.HistoryEvent) {
 	handler := e.metricsHandler.WithTags(metrics.OperationTag(metrics.EventsCachePutEventScope))
 	handler.Counter(metrics.CacheRequests.GetMetricName()).Record(1)
 	startTime := time.Now().UTC()
-	defer handler.Timer(metrics.CacheLatency.GetMetricName()).Record(time.Since(startTime))
+	defer func() { handler.Timer(metrics.CacheLatency.GetMetricName()).Record(time.Since(startTime)) }()
 
 	if !e.validateKey(key) {
 		return
@@ -162,7 +162,7 @@ func (e *CacheImpl) DeleteEvent(key EventKey) {
 	handler := e.metricsHandler.WithTags(metrics.OperationTag(metrics.EventsCacheDeleteEventScope))
 	handler.Counter(metrics.CacheRequests.GetMetricName()).Record(1)
 	startTime := time.Now().UTC()
-	defer handler.Timer(metrics.CacheLatency.GetMetricName()).Record(time.Since(startTime))
+	defer func() { handler.Timer(metrics.CacheLatency.GetMetricName()).Record(time.Since(startTime)) }()
 
 	e.validateKey(key) // just for log message, delete anyway
 	e.Delete(key)
@@ -178,7 +178,7 @@ func (e *CacheImpl) getHistoryEventFromStore(
 	handler := e.metricsHandler.WithTags(metrics.OperationTag(metrics.EventsCacheGetFromStoreScope))
 	handler.Counter(metrics.CacheRequests.GetMetricName()).Record(1)
 	startTime := time.Now().UTC()
-	defer handler.Timer(metrics.CacheLatency.GetMetricName()).Record(time.Since(startTime))
+	defer func() { handler.Timer(metrics.CacheLatency.GetMetricName()).Record(time.Since(startTime)) }()
 
 	response, err := e.eventsMgr.ReadHistoryBranch(ctx, &persistence.ReadHistoryBranchRequest{
 		BranchToken:   branchToken,

diff --git a/service/history/events/notifier.go b/service/history/events/notifier.go
@@ -198,7 +198,9 @@ func (notifier *NotifierImpl) dispatchHistoryEventNotification(event *Notificati
 	identifier := event.ID
 
 	startTime := time.Now().UTC()
-	defer notifier.metricsHandler.Timer(metrics.HistoryEventNotificationFanoutLatency.GetMetricName()).Record(time.Since(startTime))
+	defer func() {
+		notifier.metricsHandler.Timer(metrics.HistoryEventNotificationFanoutLatency.GetMetricName()).Record(time.Since(startTime))
+	}()
 	_, _, _ = notifier.eventsPubsubs.GetAndDo(identifier, func(key interface{}, value interface{}) error {
 		subscribers := value.(map[string]chan *Notification)
 

diff --git a/service/history/replication/task_executor.go b/service/history/replication/task_executor.go
@@ -146,10 +146,12 @@ func (e *taskExecutorImpl) handleActivityTask(
 	}
 
 	startTime := time.Now().UTC()
-	defer e.metricsHandler.Timer(metrics.ServiceLatency.GetMetricName()).Record(
-		time.Since(startTime),
-		metrics.OperationTag(metrics.SyncActivityTaskScope),
-	)
+	defer func() {
+		e.metricsHandler.Timer(metrics.ServiceLatency.GetMetricName()).Record(
+			time.Since(startTime),
+			metrics.OperationTag(metrics.SyncActivityTaskScope),
+		)
+	}()
 
 	request := &historyservice.SyncActivityRequest{
 		NamespaceId:        attr.NamespaceId,
@@ -181,10 +183,12 @@ func (e *taskExecutorImpl) handleActivityTask(
 			metrics.OperationTag(metrics.HistoryRereplicationByActivityReplicationScope),
 		)
 		startTime := time.Now().UTC()
-		defer e.metricsHandler.Timer(metrics.ClientLatency.GetMetricName()).Record(
-			time.Since(startTime),
-			metrics.OperationTag(metrics.HistoryRereplicationByActivityReplicationScope),
-		)
+		defer func() {
+			e.metricsHandler.Timer(metrics.ClientLatency.GetMetricName()).Record(
+				time.Since(startTime),
+				metrics.OperationTag(metrics.HistoryRereplicationByActivityReplicationScope),
+			)
+		}()
 
 		resendErr := e.nDCHistoryResender.SendSingleWorkflowHistory(
 			ctx,
@@ -227,10 +231,12 @@ func (e *taskExecutorImpl) handleHistoryReplicationTask(
 	}
 
 	startTime := time.Now().UTC()
-	defer e.metricsHandler.Timer(metrics.ServiceLatency.GetMetricName()).Record(
-		time.Since(startTime),
-		metrics.OperationTag(metrics.HistoryReplicationTaskScope),
-	)
+	defer func() {
+		e.metricsHandler.Timer(metrics.ServiceLatency.GetMetricName()).Record(
+			time.Since(startTime),
+			metrics.OperationTag(metrics.HistoryReplicationTaskScope),
+		)
+	}()
 
 	request := &historyservice.ReplicateEventsV2Request{
 		NamespaceId: attr.NamespaceId,
@@ -257,10 +263,12 @@ func (e *taskExecutorImpl) handleHistoryReplicationTask(
 			metrics.OperationTag(metrics.HistoryRereplicationByHistoryReplicationScope),
 		)
 		startTime := time.Now().UTC()
-		defer e.metricsHandler.Timer(metrics.ClientLatency.GetMetricName()).Record(
-			time.Since(startTime),
-			metrics.OperationTag(metrics.HistoryRereplicationByHistoryReplicationScope),
-		)
+		defer func() {
+			e.metricsHandler.Timer(metrics.ClientLatency.GetMetricName()).Record(
+				time.Since(startTime),
+				metrics.OperationTag(metrics.HistoryRereplicationByHistoryReplicationScope),
+			)
+		}()
 
 		resendErr := e.nDCHistoryResender.SendSingleWorkflowHistory(
 			ctx,

diff --git a/service/history/shard/context_impl.go b/service/history/shard/context_impl.go
@@ -1538,7 +1538,7 @@ func (s *ContextImpl) wLock() {
 	handler := s.metricsHandler.WithTags(metrics.OperationTag(metrics.ShardInfoScope))
 	handler.Counter(metrics.LockRequests.GetMetricName()).Record(1)
 	startTime := time.Now().UTC()
-	defer handler.Timer(metrics.LockLatency.GetMetricName()).Record(time.Since(startTime))
+	defer func() { handler.Timer(metrics.LockLatency.GetMetricName()).Record(time.Since(startTime)) }()
 
 	s.rwLock.Lock()
 }
@@ -1547,7 +1547,7 @@ func (s *ContextImpl) rLock() {
 	handler := s.metricsHandler.WithTags(metrics.OperationTag(metrics.ShardInfoScope))
 	handler.Counter(metrics.LockRequests.GetMetricName()).Record(1)
 	startTime := time.Now().UTC()
-	defer handler.Timer(metrics.LockLatency.GetMetricName()).Record(time.Since(startTime))
+	defer func() { handler.Timer(metrics.LockLatency.GetMetricName()).Record(time.Since(startTime)) }()
 
 	s.rwLock.RLock()
 }

diff --git a/service/history/shard/controller_impl.go b/service/history/shard/controller_impl.go
@@ -188,14 +188,18 @@ func (c *ControllerImpl) GetShardByID(
 	shardID int32,
 ) (Context, error) {
 	startTime := time.Now().UTC()
-	defer c.taggedMetricsHandler.Timer(metrics.GetEngineForShardLatency.GetMetricName()).Record(time.Since(startTime))
+	defer func() {
+		c.taggedMetricsHandler.Timer(metrics.GetEngineForShardLatency.GetMetricName()).Record(time.Since(startTime))
+	}()
 
 	return c.getOrCreateShardContext(shardID)
 }
 
 func (c *ControllerImpl) CloseShardByID(shardID int32) {
 	startTime := time.Now().UTC()
-	defer c.taggedMetricsHandler.Timer(metrics.RemoveEngineForShardLatency.GetMetricName()).Record(time.Since(startTime))
+	defer func() {
+		c.taggedMetricsHandler.Timer(metrics.RemoveEngineForShardLatency.GetMetricName()).Record(time.Since(startTime))
+	}()
 
 	shard, newNumShards := c.removeShard(shardID, nil)
 	// Stop the current shard, if it exists.
@@ -209,7 +213,9 @@ func (c *ControllerImpl) CloseShardByID(shardID int32) {
 
 func (c *ControllerImpl) shardClosedCallback(shard *ContextImpl) {
 	startTime := time.Now().UTC()
-	defer c.taggedMetricsHandler.Timer(metrics.RemoveEngineForShardLatency.GetMetricName()).Record(time.Since(startTime))
+	defer func() {
+		c.taggedMetricsHandler.Timer(metrics.RemoveEngineForShardLatency.GetMetricName()).Record(time.Since(startTime))
+	}()
 
 	c.taggedMetricsHandler.Counter(metrics.ShardContextClosedCounter.GetMetricName()).Record(1)
 	_, newNumShards := c.removeShard(shard.shardID, shard)
@@ -350,7 +356,9 @@ func (c *ControllerImpl) shardManagementPump() {
 func (c *ControllerImpl) acquireShards() {
 	c.taggedMetricsHandler.Counter(metrics.AcquireShardsCounter.GetMetricName()).Record(1)
 	startTime := time.Now().UTC()
-	defer c.taggedMetricsHandler.Timer(metrics.AcquireShardsLatency.GetMetricName()).Record(time.Since(startTime))
+	defer func() {
+		c.taggedMetricsHandler.Timer(metrics.AcquireShardsLatency.GetMetricName()).Record(time.Since(startTime))
+	}()
 
 	tryAcquire := func(shardID int32) {
 		info, err := c.historyServiceResolver.Lookup(convert.Int32ToString(shardID))

diff --git a/service/history/timerQueueStandbyTaskExecutor.go b/service/history/timerQueueStandbyTaskExecutor.go
@@ -510,7 +510,7 @@ func (t *timerQueueStandbyTaskExecutor) fetchHistoryFromRemote(
 	scope := t.metricHandler.WithTags(metrics.OperationTag(metrics.HistoryRereplicationByTimerTaskScope))
 	scope.Counter(metrics.ClientRequests.GetMetricName()).Record(1)
 	startTime := time.Now()
-	defer scope.Timer(metrics.ClientLatency.GetMetricName()).Record(time.Since(startTime))
+	defer func() { scope.Timer(metrics.ClientLatency.GetMetricName()).Record(time.Since(startTime)) }()
 
 	adminClient, err := t.shard.GetRemoteAdminClient(remoteClusterName)
 	if err != nil {

diff --git a/service/history/transferQueueStandbyTaskExecutor.go b/service/history/transferQueueStandbyTaskExecutor.go
@@ -625,7 +625,7 @@ func (t *transferQueueStandbyTaskExecutor) fetchHistoryFromRemote(
 	scope := t.metricHandler.WithTags(metrics.OperationTag(metrics.HistoryRereplicationByTransferTaskScope))
 	scope.Counter(metrics.ClientRequests.GetMetricName()).Record(1)
 	startTime := time.Now().UTC()
-	defer scope.Timer(metrics.ClientLatency.GetMetricName()).Record(time.Since(startTime))
+	defer func() { scope.Timer(metrics.ClientLatency.GetMetricName()).Record(time.Since(startTime)) }()
 
 	adminClient, err := t.shard.GetRemoteAdminClient(remoteClusterName)
 	if err != nil {

diff --git a/service/history/workflow/cache.go b/service/history/workflow/cache.go
@@ -107,7 +107,7 @@ func (c *CacheImpl) GetOrCreateWorkflowExecution(
 	handler := c.metricsHandler.WithTags(metrics.OperationTag(metrics.HistoryCacheGetOrCreateScope))
 	handler.Counter(metrics.CacheRequests.GetMetricName()).Record(1)
 	start := time.Now()
-	defer handler.Timer(metrics.CacheLatency.GetMetricName()).Record(time.Since(start))
+	defer func() { handler.Timer(metrics.CacheLatency.GetMetricName()).Record(time.Since(start)) }()
 
 	weCtx, weReleaseFunc, err := c.getOrCreateWorkflowExecutionInternal(
 		ctx,

diff --git a/service/worker/replicator/namespace_replication_message_processor.go b/service/worker/replicator/namespace_replication_message_processor.go
@@ -213,7 +213,9 @@ func (p *namespaceReplicationMessageProcessor) handleNamespaceReplicationTask(
 ) error {
 	p.metricsHandler.Counter(metrics.ReplicatorMessages.GetMetricName()).Record(1)
 	startTime := time.Now().UTC()
-	defer p.metricsHandler.Timer(metrics.ReplicatorLatency.GetMetricName()).Record(time.Since(startTime))
+	defer func() {
+		p.metricsHandler.Timer(metrics.ReplicatorLatency.GetMetricName()).Record(time.Since(startTime))
+	}()
 
 	return p.taskExecutor.Execute(ctx, task.GetNamespaceTaskAttributes())
 }