Skip to content

Fix sla checks being skipped when the preceding task instance misses the sla #26968

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
potiuk merged 1 commit into from
Dec 14, 2022
Merged

Fix sla checks being skipped when the preceding task instance misses the sla #26968

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion airflow/dag_processing/processor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -425,7 +425,7 @@ def manage_slas(self, dag: DAG, session: Session = None) -> None:
if next_info is None:
break
if (ti.dag_id, ti.task_id, next_info.logical_date) in recorded_slas_query:
break
continue
Copy link
Contributor

@ephraimbuddy ephraimbuddy Nov 3, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This seems like a change in behaviour. It seems like it's going to record too many SLAs this way.
cc: @ashb

Copy link
Contributor Author

@doiken doiken Nov 4, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This seems like a change in behaviour

I would like not to change but fix the behaviour changed by #19553.

I think the current sla check behaviour is split as follows

  • No SlaMiss record: all task instances missing sla are recorded
  • SlaMiss record already exists: no task instance is recorded by break

It seems like it's going to record too many SLAs this way.

Only tasks satisfying next_info.logical_date + task.sla < ts are recorded.

If we want to break here,
I think we need break after first SlaMiss is appended as well to keep behaviour consistent.

...
sla_miss = SlaMiss(
    task_id=ti.task_id,
    dag_id=ti.dag_id,
    execution_date=next_info.logical_date,
    timestamp=ts,
)
sla_misses.append(sla_miss)
Stats.incr('sla_missed')
break

if next_info.logical_date + task.sla < ts:

sla_miss = SlaMiss(
Expand Down
43 changes: 43 additions & 0 deletions tests/dag_processing/test_processor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -249,6 +249,49 @@ def test_dag_file_processor_sla_miss_doesnot_raise_integrity_error(self, mock_st
# ti is successful thereby trying to insert a duplicate record.
dag_file_processor.manage_slas(dag=dag, session=session)

@mock.patch("airflow.dag_processing.processor.Stats.incr")
def test_dag_file_processor_sla_miss_continue_checking_the_task_instances_after_recording_missing_sla(
self, mock_stats_incr, dag_maker
):
"""
Test that the dag file processor continue checking subsequent task instances
even if the preceding task instance misses the sla ahead
"""
session = settings.Session()

# Create a dag with a start of 3 days ago and sla of 1 day,
# so we have 2 missing slas
now = timezone.utcnow()
test_start_date = now - datetime.timedelta(days=3)
with dag_maker(
dag_id="test_sla_miss",
default_args={"start_date": test_start_date, "sla": datetime.timedelta(days=1)},
) as dag:
task = EmptyOperator(task_id="dummy")

dag_maker.create_dagrun(execution_date=test_start_date, state=State.SUCCESS)

session.merge(TaskInstance(task=task, execution_date=test_start_date, state="success"))
session.merge(
SlaMiss(task_id=task.task_id, dag_id=dag.dag_id, execution_date=now - datetime.timedelta(days=2))
)
session.flush()

dag_file_processor = DagFileProcessor(
dag_ids=[], dag_directory=TEST_DAGS_FOLDER, log=mock.MagicMock()
)
dag_file_processor.manage_slas(dag=dag, session=session)
sla_miss_count = (
session.query(SlaMiss)
.filter(
SlaMiss.dag_id == dag.dag_id,
SlaMiss.task_id == task.task_id,
)
.count()
)
assert sla_miss_count == 2
mock_stats_incr.assert_called_with("sla_missed")

@mock.patch("airflow.dag_processing.processor.Stats.incr")
def test_dag_file_processor_sla_miss_callback_exception(self, mock_stats_incr, create_dummy_dag):
"""
Expand Down