Log more tags while putting to replication task to dlq (uber#4754)

service/history/replication/task_processor.go

@@ -115,6 +115,7 @@ func NewTaskProcessor(
 	taskExecutor TaskExecutor,
 ) TaskProcessor {
 	shardID := shard.GetShardID()
+	sourceCluster := taskFetcher.GetSourceCluster()
 	firstRetryPolicy := backoff.NewExponentialRetryPolicy(config.ReplicationTaskProcessorErrorRetryWait(shardID))
 	firstRetryPolicy.SetMaximumAttempts(config.ReplicationTaskProcessorErrorRetryMaxAttempts(shardID))
 	secondRetryPolicy := backoff.NewExponentialRetryPolicy(config.ReplicationTaskProcessorErrorSecondRetryWait(shardID))
@@ -131,14 +132,14 @@ func NewTaskProcessor(
 	noTaskRetrier := backoff.NewRetrier(noTaskBackoffPolicy, backoff.SystemClock)
 	return &taskProcessorImpl{
 		currentCluster:         shard.GetClusterMetadata().GetCurrentClusterName(),
-		sourceCluster:          taskFetcher.GetSourceCluster(),
+		sourceCluster:          sourceCluster,
 		status:                 common.DaemonStatusInitialized,
 		shard:                  shard,
 		historyEngine:          historyEngine,
 		historySerializer:      persistence.NewPayloadSerializer(),
 		config:                 config,
 		metricsClient:          metricsClient,
-		logger:                 shard.GetLogger(),
+		logger:                 shard.GetLogger().WithTags(tag.SourceCluster(sourceCluster), tag.ShardID(shardID)),
 		taskExecutor:           taskExecutor,
 		hostRateLimiter:        taskFetcher.GetRateLimiter(),
 		shardRateLimiter:       quotas.NewDynamicRateLimiter(config.ReplicationTaskProcessorShardQPS.AsFloat64()),
@@ -426,17 +427,26 @@ func (p *taskProcessorImpl) processSingleTask(replicationTask *types.Replication
 		p.logger.Warn("Skip adding new messages to DLQ.", tag.Error(err))
 		return err
 	default:
-		p.logger.Error(
-			"Failed to apply replication task after retry. Putting task into DLQ.",
-			tag.TaskID(replicationTask.GetSourceTaskID()),
+		request, err := p.generateDLQRequest(replicationTask)
+		if err != nil {
+			p.logger.Error("Failed to generate DLQ replication task.", tag.Error(err))
+			// We cannot deserialize the task. Dropping it.
+			return nil
+		}
+		p.logger.Error("Failed to apply replication task after retry. Putting task into DLQ.",
+			tag.WorkflowDomainID(request.TaskInfo.GetDomainID()),
+			tag.WorkflowID(request.TaskInfo.GetWorkflowID()),
+			tag.WorkflowRunID(request.TaskInfo.GetRunID()),
+			tag.TaskID(request.TaskInfo.GetTaskID()),
+			tag.TaskType(request.TaskInfo.GetTaskType()),
 			tag.Error(err),
 		)
 		//TODO: uncomment this when the execution fixer workflow is ready
 		//if err = p.triggerDataInconsistencyScan(replicationTask); err != nil {
 		//	p.logger.Warn("Failed to trigger data scan", tag.Error(err))
 		//	p.metricsClient.IncCounter(metrics.ReplicationDLQStatsScope, metrics.ReplicationDLQValidationFailed)
 		//}
-		return p.putReplicationTaskToDLQ(replicationTask)
+		return p.putReplicationTaskToDLQ(request)
 	}
 }
 
@@ -466,21 +476,7 @@ func (p *taskProcessorImpl) processTaskOnce(replicationTask *types.ReplicationTa
 	return err
 }
 
-func (p *taskProcessorImpl) putReplicationTaskToDLQ(replicationTask *types.ReplicationTask) error {
-	request, err := p.generateDLQRequest(replicationTask)
-	if err != nil {
-		p.logger.Error("Failed to generate DLQ replication task.", tag.Error(err))
-		// We cannot deserialize the task. Dropping it.
-		return nil
-	}
-	p.logger.Info("Put history replication to DLQ",
-		tag.WorkflowDomainID(request.TaskInfo.GetDomainID()),
-		tag.WorkflowID(request.TaskInfo.GetWorkflowID()),
-		tag.WorkflowRunID(request.TaskInfo.GetRunID()),
-		tag.TaskID(request.TaskInfo.GetTaskID()),
-		tag.ShardID(p.shard.GetShardID()),
-	)
-
+func (p *taskProcessorImpl) putReplicationTaskToDLQ(request *persistence.PutReplicationTaskToDLQRequest) error {
 	p.metricsClient.Scope(
 		metrics.ReplicationDLQStatsScope,
 		metrics.TargetClusterTag(p.sourceCluster),

service/history/replication/task_processor_test.go

@@ -176,70 +176,35 @@ func (s *taskProcessorSuite) TestHandleSyncShardStatus() {
 }
 
 func (s *taskProcessorSuite) TestPutReplicationTaskToDLQ_SyncActivityReplicationTask() {
-	domainID := uuid.New()
-	workflowID := uuid.New()
-	runID := uuid.New()
-	task := &types.ReplicationTask{
-		TaskType: types.ReplicationTaskTypeSyncActivity.Ptr(),
-		SyncActivityTaskAttributes: &types.SyncActivityTaskAttributes{
-			DomainID:   domainID,
-			WorkflowID: workflowID,
-			RunID:      runID,
-		},
-	}
 	request := &persistence.PutReplicationTaskToDLQRequest{
 		SourceClusterName: "standby",
 		TaskInfo: &persistence.ReplicationTaskInfo{
-			DomainID:   domainID,
-			WorkflowID: workflowID,
-			RunID:      runID,
+			DomainID:   uuid.New(),
+			WorkflowID: uuid.New(),
+			RunID:      uuid.New(),
 			TaskType:   persistence.ReplicationTaskTypeSyncActivity,
 		},
 	}
 	s.executionManager.On("PutReplicationTaskToDLQ", mock.Anything, request).Return(nil)
-	err := s.taskProcessor.putReplicationTaskToDLQ(task)
+	err := s.taskProcessor.putReplicationTaskToDLQ(request)
 	s.NoError(err)
 }
 
 func (s *taskProcessorSuite) TestPutReplicationTaskToDLQ_HistoryV2ReplicationTask() {
-	domainID := uuid.New()
-	workflowID := uuid.New()
-	runID := uuid.New()
-	events := []*types.HistoryEvent{
-		{
-			EventID: 1,
-			Version: 1,
-		},
-	}
-	serializer := s.mockShard.GetPayloadSerializer()
-	data, err := serializer.SerializeBatchEvents(events, common.EncodingTypeThriftRW)
-	s.NoError(err)
-	task := &types.ReplicationTask{
-		TaskType: types.ReplicationTaskTypeHistoryV2.Ptr(),
-		HistoryTaskV2Attributes: &types.HistoryTaskV2Attributes{
-			DomainID:   domainID,
-			WorkflowID: workflowID,
-			RunID:      runID,
-			Events: &types.DataBlob{
-				EncodingType: types.EncodingTypeThriftRW.Ptr(),
-				Data:         data.Data,
-			},
-		},
-	}
 	request := &persistence.PutReplicationTaskToDLQRequest{
 		SourceClusterName: "standby",
 		TaskInfo: &persistence.ReplicationTaskInfo{
-			DomainID:     domainID,
-			WorkflowID:   workflowID,
-			RunID:        runID,
+			DomainID:     uuid.New(),
+			WorkflowID:   uuid.New(),
+			RunID:        uuid.New(),
 			TaskType:     persistence.ReplicationTaskTypeHistory,
 			FirstEventID: 1,
 			NextEventID:  2,
 			Version:      1,
 		},
 	}
 	s.executionManager.On("PutReplicationTaskToDLQ", mock.Anything, request).Return(nil)
-	err = s.taskProcessor.putReplicationTaskToDLQ(task)
+	err := s.taskProcessor.putReplicationTaskToDLQ(request)
 	s.NoError(err)
 }