Skip to content

Commit 90c8579

Browse files
juliusgeojuliusgeo
authored
fix: cross-batch terminal events to OLAP with retried tasks would cause incorrect Prom metrics (#5176)
1 parent b4cb120 commit 90c8579

9 files changed

Lines changed: 183 additions & 43 deletions

File tree

‎internal/services/controllers/olap/controller.go‎

Lines changed: 19 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -1004,6 +1004,7 @@ func (tc *OLAPControllerImpl) handleCreateMonitoringEvent(ctx context.Context, t
10041004
// event per DAG, and the query decides from there whether it actually applies.
10051005
orchestratorUpdates := make([]v1.OrchestratorDAGStatusUpdateOpt, 0)
10061006
operatorRunIds := make(map[uuid.UUID]struct{})
1007+
retriedEventIds := make(map[uuid.UUID]struct{})
10071008

10081009
for ix, msg := range msgs {
10091010
taskMeta := taskIdsToMetas[msg.TaskId]
@@ -1021,17 +1022,19 @@ func (tc *OLAPControllerImpl) handleCreateMonitoringEvent(ctx context.Context, t
10211022
readableStatus := olapEventTypeToReadableStatus(msg.EventType)
10221023

10231024
if taskMeta.IsDagOrchestrator {
1024-
orchestratorUpdates = append(orchestratorUpdates, v1.OrchestratorDAGStatusUpdateOpt{
1025-
DagId: msg.TaskId,
1026-
DagInsertedAt: taskMeta.InsertedAt,
1027-
ReadableStatus: readableStatus,
1028-
RetryCount: msg.RetryCount,
1029-
ExternalId: taskMeta.WorkflowRunID,
1030-
DisplayName: taskMeta.DisplayName,
1031-
WorkflowId: taskMeta.WorkflowID,
1032-
WorkflowVersionId: taskMeta.WorkflowVersionID,
1033-
AdditionalMetadata: taskMeta.AdditionalMetadata,
1034-
})
1025+
if !msg.WillRetry {
1026+
orchestratorUpdates = append(orchestratorUpdates, v1.OrchestratorDAGStatusUpdateOpt{
1027+
DagId: msg.TaskId,
1028+
DagInsertedAt: taskMeta.InsertedAt,
1029+
ReadableStatus: readableStatus,
1030+
RetryCount: msg.RetryCount,
1031+
ExternalId: taskMeta.WorkflowRunID,
1032+
DisplayName: taskMeta.DisplayName,
1033+
WorkflowId: taskMeta.WorkflowID,
1034+
WorkflowVersionId: taskMeta.WorkflowVersionID,
1035+
AdditionalMetadata: taskMeta.AdditionalMetadata,
1036+
})
1037+
}
10351038

10361039
if msg.StatusOnly {
10371040
continue
@@ -1055,6 +1058,10 @@ func (tc *OLAPControllerImpl) handleCreateMonitoringEvent(ctx context.Context, t
10551058
eventExternalIdToWorkflowRunId[externalId] = taskMeta.WorkflowRunID
10561059
externalIdToMsg[externalId] = msg
10571060

1061+
if msg.WillRetry {
1062+
retriedEventIds[externalId] = struct{}{}
1063+
}
1064+
10581065
if taskMeta.WasTriggeredByDagOrchestrator {
10591066
operatorRunIds[taskMeta.WorkflowRunID] = struct{}{}
10601067
}
@@ -1144,7 +1151,7 @@ func (tc *OLAPControllerImpl) handleCreateMonitoringEvent(ctx context.Context, t
11441151

11451152
attempts++
11461153

1147-
result, workflowRunIdsOfLocksNotAcquired, err := tc.repo.OLAP().CreateTaskEvents(ctx, tenantId, opts, eventExternalIdToWorkflowRunId, orchestratorUpdates, operatorRunIds)
1154+
result, workflowRunIdsOfLocksNotAcquired, err := tc.repo.OLAP().CreateTaskEvents(ctx, tenantId, opts, eventExternalIdToWorkflowRunId, retriedEventIds, orchestratorUpdates, operatorRunIds)
11481155

11491156
if err != nil {
11501157
return fmt.Errorf("failed to create task events: %w", err)

‎internal/services/controllers/olap/controller_monitoring_events_test.go‎

Lines changed: 30 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -38,11 +38,13 @@ func (f *fakeMonitoringTasks) ListTaskMetas(_ context.Context, _ uuid.UUID, _ []
3838
type fakeMonitoringOLAP struct {
3939
v1.OLAPRepository
4040
events int
41+
retriedEvents int
4142
orchestratorUpdates int
4243
}
4344

44-
func (f *fakeMonitoringOLAP) CreateTaskEvents(_ context.Context, _ uuid.UUID, events []sqlcv1.CreateTaskEventsOLAPParams, _ map[uuid.UUID]uuid.UUID, orchestratorUpdates []v1.OrchestratorDAGStatusUpdateOpt, _ map[uuid.UUID]struct{}) (*v1.StatusUpdateResult, map[uuid.UUID]struct{}, error) {
45+
func (f *fakeMonitoringOLAP) CreateTaskEvents(_ context.Context, _ uuid.UUID, events []sqlcv1.CreateTaskEventsOLAPParams, _ map[uuid.UUID]uuid.UUID, retriedEventIds map[uuid.UUID]struct{}, orchestratorUpdates []v1.OrchestratorDAGStatusUpdateOpt, _ map[uuid.UUID]struct{}) (*v1.StatusUpdateResult, map[uuid.UUID]struct{}, error) {
4546
f.events += len(events)
47+
f.retriedEvents += len(retriedEventIds)
4648
f.orchestratorUpdates += len(orchestratorUpdates)
4749
return nil, nil, nil
4850
}
@@ -64,10 +66,19 @@ func TestHandleCreateMonitoringEvent_OrchestratorStatusOnly(t *testing.T) {
6466
}
6567
}
6668

69+
failed := func(willRetry bool) tasktypes.CreateMonitoringEventPayload {
70+
return tasktypes.CreateMonitoringEventPayload{
71+
TaskId: orchestrator.ID,
72+
EventType: sqlcv1.V1EventTypeOlapFAILED,
73+
EventTimestamp: time.Now(),
74+
WillRetry: willRetry,
75+
}
76+
}
77+
6778
tests := []struct {
68-
name string
69-
msgs []tasktypes.CreateMonitoringEventPayload
70-
wantEvents, wantUpdates int
79+
name string
80+
msgs []tasktypes.CreateMonitoringEventPayload
81+
wantEvents, wantRetried, wantUpdates int
7182
}{
7283
{
7384
name: "regular and status-only terminal events write one task event",
@@ -81,6 +92,20 @@ func TestHandleCreateMonitoringEvent_OrchestratorStatusOnly(t *testing.T) {
8192
wantEvents: 0,
8293
wantUpdates: 1,
8394
},
95+
{
96+
name: "retried failure writes its event without a terminal update",
97+
msgs: []tasktypes.CreateMonitoringEventPayload{failed(true)},
98+
wantEvents: 1,
99+
wantRetried: 1,
100+
wantUpdates: 0,
101+
},
102+
{
103+
name: "final failure updates the DAG",
104+
msgs: []tasktypes.CreateMonitoringEventPayload{failed(false)},
105+
wantEvents: 1,
106+
wantRetried: 0,
107+
wantUpdates: 1,
108+
},
84109
}
85110

86111
for _, tt := range tests {
@@ -101,6 +126,7 @@ func TestHandleCreateMonitoringEvent_OrchestratorStatusOnly(t *testing.T) {
101126

102127
require.NoError(t, tc.handleCreateMonitoringEvent(context.Background(), uuid.New(), payloads))
103128
assert.Equal(t, tt.wantEvents, olap.events)
129+
assert.Equal(t, tt.wantRetried, olap.retriedEvents)
104130
assert.Equal(t, tt.wantUpdates, olap.orchestratorUpdates)
105131
})
106132
}

‎internal/services/controllers/task/controller.go‎

Lines changed: 34 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -764,8 +764,40 @@ func (tc *TasksControllerImpl) handleTaskFailed(ctx context.Context, tenantId uu
764764
if msg.ErrorMsg != "" {
765765
idsToErrorMsg[msg.TaskId] = msg.ErrorMsg
766766
}
767+
}
768+
769+
res, err := tc.repov1.Tasks().FailTasks(ctx, tenantId, opts)
770+
771+
if err != nil {
772+
err = fmt.Errorf("could not fail tasks: %w", err)
773+
span.RecordError(err)
774+
span.SetStatus(codes.Error, "could not fail tasks")
775+
return err
776+
}
777+
778+
retried := make(map[int64]struct{}, len(res.RetriedTasks))
779+
for _, r := range res.RetriedTasks {
780+
retried[r.Id] = struct{}{}
781+
}
782+
783+
type attempt struct {
784+
id int64
785+
retryCount int32
786+
}
787+
788+
// retried and stale reports are no longer the task's current retry, so only final failures land here
789+
failed := make(map[attempt]struct{}, len(res.ReleasedTasks))
790+
for _, rt := range res.ReleasedTasks {
791+
if rt != nil && rt.IsCurrentRetry {
792+
failed[attempt{id: rt.ID, retryCount: rt.RetryCount}] = struct{}{}
793+
}
794+
}
795+
796+
// published after FailTasks so the event can say whether it ended this attempt for good
797+
for _, msg := range msgs {
798+
_, final := failed[attempt{id: msg.TaskId, retryCount: msg.RetryCount}]
799+
willRetry := !final
767800

768-
// send failed tasks to the olap repository
769801
olapMsg, err := tasktypes.MonitoringEventMessageFromInternal(
770802
tenantId,
771803
tasktypes.CreateMonitoringEventPayload{
@@ -774,6 +806,7 @@ func (tc *TasksControllerImpl) handleTaskFailed(ctx context.Context, tenantId uu
774806
EventType: sqlcv1.V1EventTypeOlapFAILED,
775807
EventTimestamp: time.Now().UTC(),
776808
EventPayload: msg.ErrorMsg,
809+
WillRetry: willRetry,
777810
},
778811
)
779812

@@ -796,15 +829,6 @@ func (tc *TasksControllerImpl) handleTaskFailed(ctx context.Context, tenantId uu
796829
}
797830
}
798831

799-
res, err := tc.repov1.Tasks().FailTasks(ctx, tenantId, opts)
800-
801-
if err != nil {
802-
err = fmt.Errorf("could not fail tasks: %w", err)
803-
span.RecordError(err)
804-
span.SetStatus(codes.Error, "could not fail tasks")
805-
return err
806-
}
807-
808832
err = tc.processFailTasksResponse(ctx, tenantId, res)
809833

810834
if err != nil {
@@ -816,11 +840,6 @@ func (tc *TasksControllerImpl) handleTaskFailed(ctx context.Context, tenantId uu
816840

817841
// a retried task is still running, so only orchestrators that reached terminal FAILED get the
818842
// reliable OLAP publish
819-
retried := make(map[int64]struct{}, len(res.RetriedTasks))
820-
for _, r := range res.RetriedTasks {
821-
retried[r.Id] = struct{}{}
822-
}
823-
824843
if err := tc.emitOrchestratorTerminalEvents(
825844
ctx, tenantId, res.ReleasedTasks, idsToErrorMsg, retried, sqlcv1.V1EventTypeOlapFAILED,
826845
); err != nil {

‎internal/services/controllers/task/process_timeouts.go‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,11 @@ func (tc *TasksControllerImpl) processTaskTimeouts(ctx context.Context, tenantId
3636
return false, fmt.Errorf("could not process fail tasks response: %w", err)
3737
}
3838

39+
retried := make(map[int64]struct{}, len(res.RetriedTasks))
40+
for _, r := range res.RetriedTasks {
41+
retried[r.Id] = struct{}{}
42+
}
43+
3944
cancellationSignals := make([]tasktypes.SignalTaskCancelledPayload, 0, len(res.TimeoutTasks))
4045

4146
for _, task := range res.TimeoutTasks {
@@ -51,6 +56,8 @@ func (tc *TasksControllerImpl) processTaskTimeouts(ctx context.Context, tenantId
5156
WorkerId: workerId,
5257
})
5358

59+
_, willRetry := retried[task.ID]
60+
5461
// send failed tasks to the olap repository
5562
olapMsg, err := tasktypes.MonitoringEventMessageFromInternal(
5663
tenantIdUUID,
@@ -60,6 +67,7 @@ func (tc *TasksControllerImpl) processTaskTimeouts(ctx context.Context, tenantId
6067
EventType: sqlcv1.V1EventTypeOlapTIMEDOUT,
6168
EventTimestamp: time.Now(),
6269
EventMessage: fmt.Sprintf("Task exceeded timeout of %s", task.StepTimeout),
70+
WillRetry: willRetry,
6371
},
6472
)
6573

‎internal/services/shared/tasktypes/v1/olap.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -106,6 +106,9 @@ type CreateMonitoringEventPayload struct {
106106
// event, for publishes that duplicate an event already sent by the regular reporting path.
107107
// Ignored for other tasks.
108108
StatusOnly bool `json:"status_only,omitempty"`
109+
110+
// WillRetry records the failure or timeout event but leaves the task and its run non-terminal.
111+
WillRetry bool `json:"will_retry,omitempty"`
109112
}
110113

111114
func MonitoringEventMessageFromActionEvent(tenantId uuid.UUID, taskId int64, retryCount int32, durableInvocationCount int32, request *contracts.StepActionEvent) (*msgqueue.Message, error) {

‎pkg/repository/olap.go‎

Lines changed: 9 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -288,7 +288,7 @@ type OLAPRepository interface {
288288
ListWorkflowRunDisplayNames(ctx context.Context, tenantId uuid.UUID, externalIds []uuid.UUID) ([]*sqlcv1.ListWorkflowRunDisplayNamesRow, error)
289289
ReadTaskRunMetrics(ctx context.Context, tenantId uuid.UUID, opts ReadTaskRunMetricsOpts) ([]TaskRunMetric, error)
290290
CreateTasks(ctx context.Context, tenantId uuid.UUID, tasks []*V1TaskWithPayload) (*StatusUpdateResult, map[uuid.UUID]struct{}, error)
291-
CreateTaskEvents(ctx context.Context, tenantId uuid.UUID, events []sqlcv1.CreateTaskEventsOLAPParams, eventExternalIdToWorkflowRunId map[uuid.UUID]uuid.UUID, orchestratorUpdates []OrchestratorDAGStatusUpdateOpt, operatorRunIds map[uuid.UUID]struct{}) (*StatusUpdateResult, map[uuid.UUID]struct{}, error)
291+
CreateTaskEvents(ctx context.Context, tenantId uuid.UUID, events []sqlcv1.CreateTaskEventsOLAPParams, eventExternalIdToWorkflowRunId map[uuid.UUID]uuid.UUID, retriedEventIds map[uuid.UUID]struct{}, orchestratorUpdates []OrchestratorDAGStatusUpdateOpt, operatorRunIds map[uuid.UUID]struct{}) (*StatusUpdateResult, map[uuid.UUID]struct{}, error)
292292
CreateDAGs(ctx context.Context, tenantId uuid.UUID, dags []*DAGWithData) (map[uuid.UUID]struct{}, error)
293293
GetTaskPointMetrics(ctx context.Context, tenantId uuid.UUID, startTimestamp *time.Time, endTimestamp *time.Time, bucketInterval time.Duration) ([]*sqlcv1.GetTaskPointMetricsRow, error)
294294
UpdateTaskStatuses(ctx context.Context, tenantIds []uuid.UUID) (bool, []UpdateTaskStatusRow, error)
@@ -2020,7 +2020,7 @@ func (r *OLAPRepositoryImpl) tryAcquireAdvisoryLocksForWorkflowRuns(ctx context.
20202020
return locksNotAcquired, nil
20212021
}
20222022

2023-
func (r *OLAPRepositoryImpl) writeTaskEventBatch(ctx context.Context, tenantId uuid.UUID, events []sqlcv1.CreateTaskEventsOLAPParams, eventExternalIdToWorkflowRunId map[uuid.UUID]uuid.UUID, orchestratorUpdates []OrchestratorDAGStatusUpdateOpt, operatorRunIds map[uuid.UUID]struct{}) (*StatusUpdateResult, map[uuid.UUID]struct{}, error) {
2023+
func (r *OLAPRepositoryImpl) writeTaskEventBatch(ctx context.Context, tenantId uuid.UUID, events []sqlcv1.CreateTaskEventsOLAPParams, eventExternalIdToWorkflowRunId map[uuid.UUID]uuid.UUID, retriedEventIds map[uuid.UUID]struct{}, orchestratorUpdates []OrchestratorDAGStatusUpdateOpt, operatorRunIds map[uuid.UUID]struct{}) (*StatusUpdateResult, map[uuid.UUID]struct{}, error) {
20242024
tx, commit, rollback, err := sqlchelpers.PrepareTx(ctx, r.pool, r.l)
20252025
if err != nil {
20262026
return nil, nil, err
@@ -2069,7 +2069,10 @@ func (r *OLAPRepositoryImpl) writeTaskEventBatch(ctx context.Context, tenantId u
20692069
eventsToWrite = append(eventsToWrite, event)
20702070
}
20712071

2072-
eventsForStatusUpdate = append(eventsForStatusUpdate, event)
2072+
// a retried attempt's failure must not make the task, or its DAG, terminal
2073+
if _, retried := retriedEventIds[event.ExternalID]; !retried {
2074+
eventsForStatusUpdate = append(eventsForStatusUpdate, event)
2075+
}
20732076

20742077
dummyInsertedAt := time.Now().Add(time.Duration(rand.Intn(2*300+1)-300) * time.Millisecond) // #nosec G404 -- non-cryptographic jitter, not security-sensitive
20752078

@@ -2080,7 +2083,7 @@ func (r *OLAPRepositoryImpl) writeTaskEventBatch(ctx context.Context, tenantId u
20802083
})
20812084
}
20822085

2083-
if len(eventsForStatusUpdate) == 0 && len(orchestratorUpdates) == 0 {
2086+
if len(eventsToWrite) == 0 && len(eventsForStatusUpdate) == 0 && len(orchestratorUpdates) == 0 {
20842087
return nil, workflowRunIdsOfLocksNotAcquired, nil
20852088
}
20862089

@@ -2664,8 +2667,8 @@ func (r *OLAPRepositoryImpl) writeDAGBatch(ctx context.Context, tenantId uuid.UU
26642667
return workflowRunIdsOfLocksNotAcquired, nil
26652668
}
26662669

2667-
func (r *OLAPRepositoryImpl) CreateTaskEvents(ctx context.Context, tenantId uuid.UUID, events []sqlcv1.CreateTaskEventsOLAPParams, eventExternalIdToWorkflowRunId map[uuid.UUID]uuid.UUID, orchestratorUpdates []OrchestratorDAGStatusUpdateOpt, operatorRunIds map[uuid.UUID]struct{}) (*StatusUpdateResult, map[uuid.UUID]struct{}, error) {
2668-
return r.writeTaskEventBatch(ctx, tenantId, events, eventExternalIdToWorkflowRunId, orchestratorUpdates, operatorRunIds)
2670+
func (r *OLAPRepositoryImpl) CreateTaskEvents(ctx context.Context, tenantId uuid.UUID, events []sqlcv1.CreateTaskEventsOLAPParams, eventExternalIdToWorkflowRunId map[uuid.UUID]uuid.UUID, retriedEventIds map[uuid.UUID]struct{}, orchestratorUpdates []OrchestratorDAGStatusUpdateOpt, operatorRunIds map[uuid.UUID]struct{}) (*StatusUpdateResult, map[uuid.UUID]struct{}, error) {
2671+
return r.writeTaskEventBatch(ctx, tenantId, events, eventExternalIdToWorkflowRunId, retriedEventIds, orchestratorUpdates, operatorRunIds)
26692672
}
26702673

26712674
func (r *OLAPRepositoryImpl) CreateTasks(ctx context.Context, tenantId uuid.UUID, tasks []*V1TaskWithPayload) (*StatusUpdateResult, map[uuid.UUID]struct{}, error) {

‎pkg/repository/olap_list_workflow_runs_test.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -115,7 +115,7 @@ func seedDAGWithFinishedChildren(t *testing.T, ctx context.Context, repo *OLAPRe
115115
}
116116
f.lastFinishedEventId = event.ExternalID
117117

118-
_, locksNotAcquired, err = repo.CreateTaskEvents(ctx, f.tenantId, []sqlcv1.CreateTaskEventsOLAPParams{event}, map[uuid.UUID]uuid.UUID{event.ExternalID: dagExternalId}, nil, nil)
118+
_, locksNotAcquired, err = repo.CreateTaskEvents(ctx, f.tenantId, []sqlcv1.CreateTaskEventsOLAPParams{event}, map[uuid.UUID]uuid.UUID{event.ExternalID: dagExternalId}, nil, nil, nil)
119119
require.NoError(t, err)
120120
require.Empty(t, locksNotAcquired)
121121
}

‎pkg/repository/olap_operator_dag_test.go‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -96,7 +96,7 @@ func (f operatorDagFixture) applyOrchestratorMonitoringEvents(t *testing.T, ctx
9696
eventExternalIdToWorkflowRunId[e.ExternalID] = f.dagExternalId
9797
}
9898

99-
_, locksNotAcquired, err := repo.CreateTaskEvents(ctx, f.tenantId, events, eventExternalIdToWorkflowRunId, nil, f.operatorRunIds())
99+
_, locksNotAcquired, err := repo.CreateTaskEvents(ctx, f.tenantId, events, eventExternalIdToWorkflowRunId, nil, nil, f.operatorRunIds())
100100
require.NoError(t, err)
101101
require.Empty(t, locksNotAcquired)
102102
}
@@ -163,7 +163,7 @@ func (f operatorDagFixture) applyChildEvents(t *testing.T, ctx context.Context,
163163
eventExternalIdToWorkflowRunId[e.ExternalID] = f.dagExternalId
164164
}
165165

166-
_, locksNotAcquired, err := repo.CreateTaskEvents(ctx, f.tenantId, events, eventExternalIdToWorkflowRunId, nil, f.operatorRunIds())
166+
_, locksNotAcquired, err := repo.CreateTaskEvents(ctx, f.tenantId, events, eventExternalIdToWorkflowRunId, nil, nil, f.operatorRunIds())
167167
require.NoError(t, err)
168168
require.Empty(t, locksNotAcquired)
169169
}
@@ -189,7 +189,7 @@ func (f operatorDagFixture) orchestratorUpdate(status sqlcv1.V1ReadableStatusOla
189189
func (f operatorDagFixture) applyOrchestratorEvents(t *testing.T, ctx context.Context, repo *OLAPRepositoryImpl, updates ...OrchestratorDAGStatusUpdateOpt) *StatusUpdateResult {
190190
t.Helper()
191191

192-
result, locksNotAcquired, err := repo.CreateTaskEvents(ctx, f.tenantId, nil, nil, updates, f.operatorRunIds())
192+
result, locksNotAcquired, err := repo.CreateTaskEvents(ctx, f.tenantId, nil, nil, nil, updates, f.operatorRunIds())
193193
require.NoError(t, err)
194194
require.Empty(t, locksNotAcquired)
195195

0 commit comments

Comments
 (0)