[Wf-Diagnostics] Improve identification of activity timeouts (#6232)

uber · Aug 16, 2024 · e23c5a6 · e23c5a6
1 parent beeac59
commit e23c5a6
Show file tree

Hide file tree

Showing 4 changed files with 79 additions and 17 deletions.
diff --git a/service/worker/diagnostics/invariants/timeout.go b/service/worker/diagnostics/invariants/timeout.go
@@ -59,14 +59,14 @@ func (t *timeout) Check(context.Context) ([]InvariantCheckResult, error) {
  })
  }
  if event.ActivityTaskTimedOutEventAttributes != nil {
- timeoutLimit, err := getActivityTaskConfiguredTimeout(event, events)
+ metadata, err := getActivityTaskMetadata(event, events)
  if err != nil {
  return nil, err
  }
  result = append(result, InvariantCheckResult{
  InvariantType: TimeoutTypeActivity.String(),
  Reason: event.GetActivityTaskTimedOutEventAttributes().GetTimeoutType().String(),
- Metadata: timeoutLimitInBytes(timeoutLimit),
+ Metadata: marshalData(metadata),
  })
  }
  if event.DecisionTaskTimedOutEventAttributes != nil {

diff --git a/service/worker/diagnostics/invariants/timeout_test.go b/service/worker/diagnostics/invariants/timeout_test.go
@@ -39,6 +39,7 @@ const (
  taskTimeoutSecond = int32(50)
  testTimeStamp = int64(2547596872371000000)
  timeUnit = time.Second
+ testTasklist = "test-tasklist"
 )
 
 func Test__Check(t *testing.T) {
@@ -61,10 +62,27 @@ func Test__Check(t *testing.T) {
  RunID: "abc",
  },
  }
+ activityTimeoutData := ActivityTimeoutMetadata{
+ TimeoutType: types.TimeoutTypeScheduleToStart.Ptr(),
+ ConfiguredTimeout: 50 * time.Second,
+ TimeElapsed: 50 * time.Second,
+ RetryPolicy: nil,
+ HeartBeatTimeout: 0,
+ Tasklist: &types.TaskList{
+ Name: testTasklist,
+ Kind: nil,
+ },
+ }
  workflowTimeoutDataInBytes, err := json.Marshal(workflowTimeoutData)
  require.NoError(t, err)
  childWfTimeoutDataInBytes, err := json.Marshal(childWfTimeoutData)
  require.NoError(t, err)
+ activityTimeoutDataInBytes, err := json.Marshal(activityTimeoutData)
+ require.NoError(t, err)
+ activityTimeoutData.TimeoutType = types.TimeoutTypeHeartbeat.Ptr()
+ activityTimeoutData.HeartBeatTimeout = 50 * time.Second
+ activityHeartBeatTimeoutDataInBytes, err := json.Marshal(activityTimeoutData)
+ require.NoError(t, err)
  taskTimeoutSecondInBytes, err := json.Marshal(taskTimeoutSecond)
  require.NoError(t, err)
  testCases := []struct {
@@ -104,12 +122,12 @@ func Test__Check(t *testing.T) {
  {
  InvariantType: TimeoutTypeActivity.String(),
  Reason: "SCHEDULE_TO_START",
- Metadata: taskTimeoutSecondInBytes,
+ Metadata: activityTimeoutDataInBytes,
  },
  {
  InvariantType: TimeoutTypeActivity.String(),
  Reason: "HEARTBEAT",
- Metadata: taskTimeoutSecondInBytes,
+ Metadata: activityHeartBeatTimeoutDataInBytes,
  },
  },
  err: nil,
@@ -205,28 +223,47 @@ func activityTimeoutHistory() *types.GetWorkflowExecutionHistoryResponse {
  History: &types.History{
  Events: []*types.HistoryEvent{
  {
- ID: 5,
+ ID: 1,
+ Timestamp: common.Int64Ptr(testTimeStamp),
  ActivityTaskScheduledEventAttributes: &types.ActivityTaskScheduledEventAttributes{
  ScheduleToStartTimeoutSeconds: common.Int32Ptr(taskTimeoutSecond),
+ TaskList: &types.TaskList{
+ Name: testTasklist,
+ Kind: nil,
+ },
  },
  },
  {
+ ID: 2,
+ Timestamp: common.Int64Ptr(testTimeStamp + int64(taskTimeoutSecond)*timeUnit.Nanoseconds()),
  ActivityTaskTimedOutEventAttributes: &types.ActivityTaskTimedOutEventAttributes{
- ScheduledEventID: 5,
- StartedEventID: 6,
+ ScheduledEventID: 1,
  TimeoutType: types.TimeoutTypeScheduleToStart.Ptr(),
  },
  },
  {
- ID: 21,
+ ID: 3,
  ActivityTaskScheduledEventAttributes: &types.ActivityTaskScheduledEventAttributes{
  HeartbeatTimeoutSeconds: common.Int32Ptr(taskTimeoutSecond),
+ TaskList: &types.TaskList{
+ Name: testTasklist,
+ Kind: nil,
+ },
  },
  },
  {
- ActivityTaskTimedOutEventAttributes: &types.ActivityTaskTimedOutEventAttributes{
+ ID: 4,
+ Timestamp: common.Int64Ptr(testTimeStamp),
+ ActivityTaskStartedEventAttributes: &types.ActivityTaskStartedEventAttributes{
  ScheduledEventID: 21,
- StartedEventID: 22,
+ },
+ },
+ {
+ ID: 5,
+ Timestamp: common.Int64Ptr(testTimeStamp + int64(taskTimeoutSecond)*timeUnit.Nanoseconds()),
+ ActivityTaskTimedOutEventAttributes: &types.ActivityTaskTimedOutEventAttributes{
+ ScheduledEventID: 3,
+ StartedEventID: 4,
  TimeoutType: types.TimeoutTypeHeartbeat.Ptr(),
  },
  },

diff --git a/service/worker/diagnostics/invariants/timeout_utils.go b/service/worker/diagnostics/invariants/timeout_utils.go
@@ -56,27 +56,43 @@ func getWorkflowExecutionConfiguredTimeout(events []*types.HistoryEvent) int32 {
  return 0
 }
 
-func getActivityTaskConfiguredTimeout(e *types.HistoryEvent, events []*types.HistoryEvent) (int32, error) {
+func getActivityTaskMetadata(e *types.HistoryEvent, events []*types.HistoryEvent) (ActivityTimeoutMetadata, error) {
  eventScheduledID := e.GetActivityTaskTimedOutEventAttributes().GetScheduledEventID()
+ eventstartedID := e.GetActivityTaskTimedOutEventAttributes().StartedEventID
  timeoutType := e.GetActivityTaskTimedOutEventAttributes().GetTimeoutType()
+ var configuredTimeout int32
+ var timeElapsed time.Duration
  for _, event := range events {
  if event.ID == eventScheduledID {
  attr := event.GetActivityTaskScheduledEventAttributes()
  switch timeoutType {
  case types.TimeoutTypeHeartbeat:
- return attr.GetHeartbeatTimeoutSeconds(), nil
+ configuredTimeout = attr.GetHeartbeatTimeoutSeconds()
+ timeElapsed = getExecutionTime(eventstartedID, e.ID, events)
  case types.TimeoutTypeScheduleToClose:
- return attr.GetScheduleToCloseTimeoutSeconds(), nil
+ configuredTimeout = attr.GetScheduleToCloseTimeoutSeconds()
+ timeElapsed = getExecutionTime(eventScheduledID, e.ID, events)
  case types.TimeoutTypeScheduleToStart:
- return attr.GetScheduleToStartTimeoutSeconds(), nil
+ configuredTimeout = attr.GetScheduleToStartTimeoutSeconds()
+ timeElapsed = getExecutionTime(eventScheduledID, e.ID, events)
  case types.TimeoutTypeStartToClose:
- return attr.GetStartToCloseTimeoutSeconds(), nil
+ configuredTimeout = attr.GetStartToCloseTimeoutSeconds()
+ timeElapsed = getExecutionTime(eventstartedID, e.ID, events)
  default:
- return 0, fmt.Errorf("unknown timeout type")
+ return ActivityTimeoutMetadata{}, fmt.Errorf("unknown timeout type")
  }
+ return ActivityTimeoutMetadata{
+ TimeoutType: timeoutType.Ptr(),
+ ConfiguredTimeout: time.Duration(configuredTimeout) * time.Second,
+ TimeElapsed: timeElapsed,
+ RetryPolicy: attr.RetryPolicy,
+ HeartBeatTimeout: time.Duration(attr.GetHeartbeatTimeoutSeconds()) * time.Second,
+ Tasklist: attr.TaskList,
+ }, nil
  }
+
  }
- return 0, fmt.Errorf("activity scheduled event not found")
+ return ActivityTimeoutMetadata{}, fmt.Errorf("activity scheduled event not found")
 }
 
 func getDecisionTaskConfiguredTimeout(eventScheduledID int64, events []*types.HistoryEvent) int32 {

diff --git a/service/worker/diagnostics/invariants/types.go b/service/worker/diagnostics/invariants/types.go
@@ -52,3 +52,12 @@ type ChildWfTimeoutMetadata struct {
  ConfiguredTimeout time.Duration
  Execution *types.WorkflowExecution
 }
+
+type ActivityTimeoutMetadata struct {
+ TimeoutType *types.TimeoutType
+ ConfiguredTimeout time.Duration
+ TimeElapsed time.Duration
+ RetryPolicy *types.RetryPolicy
+ HeartBeatTimeout time.Duration
+ Tasklist *types.TaskList
+}