Add attempt number to activity logger (#1451)

* Add attempt # to structured Activity logging

* nil checks

* Updated tests and changed executeLocalActivityTask

* Use getter for PollForActivityTaskResponse attempt

---------

Co-authored-by: Evan Sandoval <[email protected]>

Author: evansandoval

internal/activity.go

@@ -356,6 +356,7 @@ func WithActivityTask(
 		zapcore.Field{Key: tagWorkflowType, Type: zapcore.StringType, String: *task.WorkflowType.Name},
 		zapcore.Field{Key: tagWorkflowID, Type: zapcore.StringType, String: *task.WorkflowExecution.WorkflowId},
 		zapcore.Field{Key: tagRunID, Type: zapcore.StringType, String: *task.WorkflowExecution.RunId},
+		zapcore.Field{Key: tagAttempt, Type: zapcore.Int64Type, Integer: int64(task.GetAttempt())},
 	)
 
 	return context.WithValue(ctx, activityEnvContextKey, &activityEnvironment{

internal/activity_task_handler_test.go

@@ -87,6 +87,7 @@ func TestActivityTaskHandler_Execute_deadline(t *testing.T) {
 				ScheduleToCloseTimeoutSeconds:   common.Int32Ptr(d.ScheduleDuration),
 				StartedTimestamp:                common.Int64Ptr(d.StartTS.UnixNano()),
 				StartToCloseTimeoutSeconds:      common.Int32Ptr(d.StartDuration),
+				Attempt:                         common.Int32Ptr(0),
 				WorkflowType: &s.WorkflowType{
 					Name: common.StringPtr("wType"),
 				},
@@ -139,6 +140,7 @@ func TestActivityTaskHandler_Execute_worker_stop(t *testing.T) {
 		ScheduleToCloseTimeoutSeconds:   common.Int32Ptr(1),
 		StartedTimestamp:                common.Int64Ptr(now.UnixNano()),
 		StartToCloseTimeoutSeconds:      common.Int32Ptr(1),
+		Attempt:                         common.Int32Ptr(0),
 		WorkflowType: &s.WorkflowType{
 			Name: common.StringPtr("wType"),
 		},
@@ -193,6 +195,7 @@ func TestActivityTaskHandler_Execute_with_propagators(t *testing.T) {
 		ScheduleToCloseTimeoutSeconds:   common.Int32Ptr(1),
 		StartedTimestamp:                common.Int64Ptr(now.UnixNano()),
 		StartToCloseTimeoutSeconds:      common.Int32Ptr(1),
+		Attempt:                         common.Int32Ptr(0),
 		WorkflowType: &s.WorkflowType{
 			Name: common.StringPtr("wType"),
 		},
@@ -245,6 +248,7 @@ func TestActivityTaskHandler_Execute_with_propagator_failure(t *testing.T) {
 		ScheduleToCloseTimeoutSeconds:   common.Int32Ptr(1),
 		StartedTimestamp:                common.Int64Ptr(now.UnixNano()),
 		StartToCloseTimeoutSeconds:      common.Int32Ptr(1),
+		Attempt:                         common.Int32Ptr(0),
 		WorkflowType: &s.WorkflowType{
 			Name: common.StringPtr("wType"),
 		},
@@ -301,6 +305,7 @@ func TestActivityTaskHandler_Execute_with_auto_heartbeat(t *testing.T) {
 		StartedTimestamp:                common.Int64Ptr(now.UnixNano()),
 		StartToCloseTimeoutSeconds:      common.Int32Ptr(1),
 		HeartbeatTimeoutSeconds:         common.Int32Ptr(1),
+		Attempt:                         common.Int32Ptr(0),
 		WorkflowType: &s.WorkflowType{
 			Name: common.StringPtr("wType"),
 		},

internal/internal_logging_tags.go

@@ -26,6 +26,7 @@ const (
 	tagDomain                      = "Domain"
 	tagEventID                     = "EventID"
 	tagEventType                   = "EventType"
+	tagAttempt                     = "Attempt"
 	tagRunID                       = "RunID"
 	tagTaskList                    = "TaskList"
 	tagTimerID                     = "TimerID"

internal/internal_task_pollers.go

@@ -611,7 +611,9 @@ func (lath *localActivityTaskHandler) executeLocalActivityTask(task *localActivi
 		zap.String(tagLocalActivityType, activityType),
 		zap.String(tagWorkflowType, workflowType),
 		zap.String(tagWorkflowID, task.params.WorkflowInfo.WorkflowExecution.ID),
-		zap.String(tagRunID, task.params.WorkflowInfo.WorkflowExecution.RunID))
+		zap.String(tagRunID, task.params.WorkflowInfo.WorkflowExecution.RunID),
+		zap.Int64(tagAttempt, int64(task.attempt)),
+	)
 
 	metricsScope.Counter(metrics.LocalActivityTotalCounter).Inc(1)
 

internal/internal_workers_test.go

@@ -156,6 +156,7 @@ func (s *WorkersTestSuite) TestActivityWorkerStop() {
 		ScheduleToCloseTimeoutSeconds:   common.Int32Ptr(1),
 		StartedTimestamp:                common.Int64Ptr(time.Now().UnixNano()),
 		StartToCloseTimeoutSeconds:      common.Int32Ptr(1),
+		Attempt:                         common.Int32Ptr(0),
 		WorkflowType: &m.WorkflowType{
 			Name: common.StringPtr("wType"),
 		},

internal/internal_workflow_testsuite.go

@@ -1699,6 +1699,7 @@ func newTestActivityTask(workflowID, runID, activityID, workflowTypeName, domain
 		StartedTimestamp:                common.Int64Ptr(time.Now().UnixNano()),
 		StartToCloseTimeoutSeconds:      common.Int32Ptr(params.StartToCloseTimeoutSeconds),
 		HeartbeatTimeoutSeconds:         common.Int32Ptr(params.HeartbeatTimeoutSeconds),
+		Attempt:                         common.Int32Ptr(0),
 		WorkflowType: &shared.WorkflowType{
 			Name: common.StringPtr(workflowTypeName),
 		},