You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2021/10/28 15:50:31 UTC
[GitHub] [airflow] WesternStar opened a new issue #19286: metric: Time after triggering
WesternStar opened a new issue #19286:
URL: https://github.com/apache/airflow/issues/19286
### Description
Can we have a metric that indicates how long after a dag is triggered its scheduled? I notice that first task scheduling delay doesn't include external triggers but it might be nice to have how long after a dag is triggered it ran to get an idea of things like scheduler failures, full queues etc.
### Use case/motivation
We currently have a critical pipeline and a set of user focused dags that are triggered from the critical pipeline. We would like to have an idea of how fast the scheduler picks up and runs those dags. This is to get a better idea of scheduler responsiveness and also figure out if we need to increase the size of our environment, split environments
### Related issues
_No response_
### Are you willing to submit a PR?
- [X] Yes I am willing to submit a PR!
### Code of Conduct
- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ShakaibKhan removed a comment on issue #19286: metric: Time after triggering
Posted by GitBox <gi...@apache.org>.
ShakaibKhan removed a comment on issue #19286:
URL: https://github.com/apache/airflow/issues/19286#issuecomment-960430074
If everyone is okay with it could I try taking on this issue?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] EricGao888 commented on issue #19286: metric: Time after triggering
Posted by GitBox <gi...@apache.org>.
EricGao888 commented on issue #19286:
URL: https://github.com/apache/airflow/issues/19286#issuecomment-976229723
>
Thanks for the detailed suggestions!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] EricGao888 edited a comment on issue #19286: metric: Time after triggering
Posted by GitBox <gi...@apache.org>.
EricGao888 edited a comment on issue #19286:
URL: https://github.com/apache/airflow/issues/19286#issuecomment-986380136
> BTW, for backwards compatibility reasons, both the `DagRun` column and `create_dagrun` argument need to be optional (None by default), and if it’s None, we’ll fall back to `data_interval_end` like we currently use (for lack of a better value).
After I added the `run_after` field to `dagrun` and changed the related logic to calculate the scheduling delay metric, I got blocked by the following error when initializing db:
```
(airflow_dev) ➜ ~ airflow standalone
standalone | Starting Airflow Standalone
standalone | Checking database is initialized
INFO [alembic.runtime.migration] Context impl SQLiteImpl.
INFO [alembic.runtime.migration] Will assume non-transactional DDL.
WARNI [airflow.models.crypto] empty cryptography key - values will not be stored encrypted.
Traceback (most recent call last):
File "/opt/anaconda3/envs/airflow_dev/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1277, in _execute_context
cursor, statement, parameters, context
File "/opt/anaconda3/envs/airflow_dev/lib/python3.6/site-packages/sqlalchemy/engine/default.py", line 608, in do_execute
cursor.execute(statement, parameters)
sqlite3.OperationalError: no such column: dag_run.run_after
```
It seems somehow the new field was not mapped to a column when creating the table. May I ask which part of the code should I look into to fix this? Any help will be appreciated!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr edited a comment on issue #19286: metric: Time after triggering
Posted by GitBox <gi...@apache.org>.
uranusjr edited a comment on issue #19286:
URL: https://github.com/apache/airflow/issues/19286#issuecomment-975147311
All `DagRun` instances are populated from `DagRunInfo`. Actually, `DagRunInfo` is already used to populate some fields on `DagModel` that are used to schedule the next DAG run:
https://github.com/apache/airflow/blob/1b89e682a1a04adf2cf5ebb37a453e245a6e7191/airflow/models/dag.py#L2920-L2928
So what we need to do is to pass that information on when the scheduler decides to actually create the next DAG run.
A DAG run is created with `DAG.create_dagrun()`, so you need to add an argument on that function, and change all `create_dagrun()` calls to use that argument. This will allow `DagRun.run_after` to be populated by `DagRunInfo.run_after` and thus in sync.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] EricGao888 commented on issue #19286: metric: Time after triggering
Posted by GitBox <gi...@apache.org>.
EricGao888 commented on issue #19286:
URL: https://github.com/apache/airflow/issues/19286#issuecomment-962999214
` def _update_state(dag: DAG, dag_run: DagRun):
dag_run.state = State.RUNNING
dag_run.start_date = timezone.utcnow()
if dag.timetable.periodic:
# TODO: Logically, this should be DagRunInfo.run_after, but the
# information is not stored on a DagRun, only before the actual
# execution on DagModel.next_dagrun_create_after. We should add
# a field on DagRun for this instead of relying on the run
# always happening immediately after the data interval.
expected_start_date = dag.get_run_data_interval(dag_run).end
schedule_delay = dag_run.start_date - expected_start_date
Stats.timing(f'dagrun.schedule_delay.{dag.dag_id}', schedule_delay)`
` if not self.dag.timetable.periodic:
# We can't emit this metric if there is no following schedule to calculate from!
return
ordered_tis_by_start_date = [ti for ti in finished_tis if ti.start_date]
ordered_tis_by_start_date.sort(key=lambda ti: ti.start_date, reverse=False)
first_start_date = ordered_tis_by_start_date[0].start_date
if first_start_date:
# TODO: Logically, this should be DagRunInfo.run_after, but the
# information is not stored on a DagRun, only before the actual
# execution on DagModel.next_dagrun_create_after. We should add
# a field on DagRun for this instead of relying on the run
# always happening immediately after the data interval.
data_interval_end = dag.get_run_data_interval(self).end
true_delay = first_start_date - data_interval_end
if true_delay.total_seconds() > 0:
Stats.timing(f'dagrun.{dag.dag_id}.first_task_scheduling_delay', true_delay)`
I think the above two code fragments copied from scheduler_job.py and dagrun.py calculate the metics for dagrun.schedule_delay.<dag_id> and dagrun.<dag_id>.first_task_scheduling_delay respectively. It seems if `dag.timetable.periodic` is false, these two metrics will not be calculated and we cannot achieve the delay for these kinds of dags.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] EricGao888 edited a comment on issue #19286: metric: Time after triggering
Posted by GitBox <gi...@apache.org>.
EricGao888 edited a comment on issue #19286:
URL: https://github.com/apache/airflow/issues/19286#issuecomment-975142729
> The TODO comment hints what a proper solution would be. We should add a new field on `DagRun` to store `DagRunInfo.run_after` (i.e. the earliest time the DagRun _can_ happen), and calculate that against `DagRun.start_date`. That value is always available and would enable us to remove the block below.
```python
@classmethod
def exact(cls, at: DateTime) -> "DagRunInfo":
"""Represent a run on an exact time."""
return cls(run_after=at, data_interval=DataInterval.exact(at))
@classmethod
def interval(cls, start: DateTime, end: DateTime) -> "DagRunInfo":
"""Represent a run on a continuous schedule.
In such a schedule, each data interval starts right after the previous
one ends, and each run is scheduled right after the interval ends. This
applies to all schedules prior to AIP-39 except ``@once`` and ``None``.
"""
return cls(run_after=end, data_interval=DataInterval(start, end))
```
It seems that these two methods above in `base.py` are called to assign value to `DagRunInfo.run_after`. If a new field is added in `DagRun` as the suggestion indicated above, may I ask if there is a good way to make sure that the new field is consistent with `DagRunInfo.run_after`? Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] boring-cyborg[bot] commented on issue #19286: metric: Time after triggering
Posted by GitBox <gi...@apache.org>.
boring-cyborg[bot] commented on issue #19286:
URL: https://github.com/apache/airflow/issues/19286#issuecomment-953976220
Thanks for opening your first issue here! Be sure to follow the issue template!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] EricGao888 removed a comment on issue #19286: metric: Time after triggering
Posted by GitBox <gi...@apache.org>.
EricGao888 removed a comment on issue #19286:
URL: https://github.com/apache/airflow/issues/19286#issuecomment-976229723
>
Thanks for the detailed suggestions!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] EricGao888 edited a comment on issue #19286: metric: Time after triggering
Posted by GitBox <gi...@apache.org>.
EricGao888 edited a comment on issue #19286:
URL: https://github.com/apache/airflow/issues/19286#issuecomment-962999214
`
# dagrun.schedule_delay
def _update_state(dag: DAG, dag_run: DagRun):
dag_run.state = State.RUNNING
dag_run.start_date = timezone.utcnow()
if dag.timetable.periodic:
# TODO: Logically, this should be DagRunInfo.run_after, but the
# information is not stored on a DagRun, only before the actual
# execution on DagModel.next_dagrun_create_after. We should add
# a field on DagRun for this instead of relying on the run
# always happening immediately after the data interval.
expected_start_date = dag.get_run_data_interval(dag_run).end
schedule_delay = dag_run.start_date - expected_start_date
Stats.timing(f'dagrun.schedule_delay.{dag.dag_id}', schedule_delay)
# first_task_scheduling_delay
if not self.dag.timetable.periodic:
# We can't emit this metric if there is no following schedule to calculate from!
return
ordered_tis_by_start_date = [ti for ti in finished_tis if ti.start_date]
ordered_tis_by_start_date.sort(key=lambda ti: ti.start_date, reverse=False)
first_start_date = ordered_tis_by_start_date[0].start_date
if first_start_date:
# TODO: Logically, this should be DagRunInfo.run_after, but the
# information is not stored on a DagRun, only before the actual
# execution on DagModel.next_dagrun_create_after. We should add
# a field on DagRun for this instead of relying on the run
# always happening immediately after the data interval.
data_interval_end = dag.get_run_data_interval(self).end
true_delay = first_start_date - data_interval_end
if true_delay.total_seconds() > 0:
Stats.timing(f'dagrun.{dag.dag_id}.first_task_scheduling_delay', true_delay)
`
I think the above two code fragments copied from scheduler_job.py and dagrun.py calculate the metics for dagrun.schedule_delay.<dag_id> and dagrun.<dag_id>.first_task_scheduling_delay respectively. It seems if `dag.timetable.periodic` is false, these two metrics will not be calculated and we cannot achieve the delay for these kinds of dags.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] EricGao888 edited a comment on issue #19286: metric: Time after triggering
Posted by GitBox <gi...@apache.org>.
EricGao888 edited a comment on issue #19286:
URL: https://github.com/apache/airflow/issues/19286#issuecomment-962999214
` def _update_state(dag: DAG, dag_run: DagRun):
dag_run.state = State.RUNNING
dag_run.start_date = timezone.utcnow()
if dag.timetable.periodic:
# TODO: Logically, this should be DagRunInfo.run_after, but the
# information is not stored on a DagRun, only before the actual
# execution on DagModel.next_dagrun_create_after. We should add
# a field on DagRun for this instead of relying on the run
# always happening immediately after the data interval.
expected_start_date = dag.get_run_data_interval(dag_run).end
schedule_delay = dag_run.start_date - expected_start_date
Stats.timing(f'dagrun.schedule_delay.{dag.dag_id}', schedule_delay)`
----------------------------------------------------------------------------------------
` if not self.dag.timetable.periodic:
# We can't emit this metric if there is no following schedule to calculate from!
return
ordered_tis_by_start_date = [ti for ti in finished_tis if ti.start_date]
ordered_tis_by_start_date.sort(key=lambda ti: ti.start_date, reverse=False)
first_start_date = ordered_tis_by_start_date[0].start_date
if first_start_date:
# TODO: Logically, this should be DagRunInfo.run_after, but the
# information is not stored on a DagRun, only before the actual
# execution on DagModel.next_dagrun_create_after. We should add
# a field on DagRun for this instead of relying on the run
# always happening immediately after the data interval.
data_interval_end = dag.get_run_data_interval(self).end
true_delay = first_start_date - data_interval_end
if true_delay.total_seconds() > 0:
Stats.timing(f'dagrun.{dag.dag_id}.first_task_scheduling_delay', true_delay)`
I think the above two code fragments copied from scheduler_job.py and dagrun.py calculate the metics for dagrun.schedule_delay.<dag_id> and dagrun.<dag_id>.first_task_scheduling_delay respectively. It seems if `dag.timetable.periodic` is false, these two metrics will not be calculated and we cannot achieve the delay for these kinds of dags.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ShakaibKhan commented on issue #19286: metric: Time after triggering
Posted by GitBox <gi...@apache.org>.
ShakaibKhan commented on issue #19286:
URL: https://github.com/apache/airflow/issues/19286#issuecomment-960430074
If everyone is okay with it could I try taking on this issue?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] EricGao888 edited a comment on issue #19286: metric: Time after triggering
Posted by GitBox <gi...@apache.org>.
EricGao888 edited a comment on issue #19286:
URL: https://github.com/apache/airflow/issues/19286#issuecomment-962999214
` def _update_state(dag: DAG, dag_run: DagRun):
dag_run.state = State.RUNNING
dag_run.start_date = timezone.utcnow()
if dag.timetable.periodic:
# TODO: Logically, this should be DagRunInfo.run_after, but the
# information is not stored on a DagRun, only before the actual
# execution on DagModel.next_dagrun_create_after. We should add
# a field on DagRun for this instead of relying on the run
# always happening immediately after the data interval.
expected_start_date = dag.get_run_data_interval(dag_run).end
schedule_delay = dag_run.start_date - expected_start_date
Stats.timing(f'dagrun.schedule_delay.{dag.dag_id}', schedule_delay)
`
` if not self.dag.timetable.periodic:
# We can't emit this metric if there is no following schedule to calculate from!
return
ordered_tis_by_start_date = [ti for ti in finished_tis if ti.start_date]
ordered_tis_by_start_date.sort(key=lambda ti: ti.start_date, reverse=False)
first_start_date = ordered_tis_by_start_date[0].start_date
if first_start_date:
# TODO: Logically, this should be DagRunInfo.run_after, but the
# information is not stored on a DagRun, only before the actual
# execution on DagModel.next_dagrun_create_after. We should add
# a field on DagRun for this instead of relying on the run
# always happening immediately after the data interval.
data_interval_end = dag.get_run_data_interval(self).end
true_delay = first_start_date - data_interval_end
if true_delay.total_seconds() > 0:
Stats.timing(f'dagrun.{dag.dag_id}.first_task_scheduling_delay', true_delay)
`
I think the above two code fragments copied from scheduler_job.py and dagrun.py calculate the metics for dagrun.schedule_delay.<dag_id> and dagrun.<dag_id>.first_task_scheduling_delay respectively. It seems if `dag.timetable.periodic` is false, these two metrics will not be calculated and we cannot achieve the delay for these kinds of dags.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] EricGao888 commented on issue #19286: metric: Time after triggering
Posted by GitBox <gi...@apache.org>.
EricGao888 commented on issue #19286:
URL: https://github.com/apache/airflow/issues/19286#issuecomment-975142729
> The TODO comment hints what a proper solution would be. We should add a new field on `DagRun` to store `DagRunInfo.run_after` (i.e. the earliest time the DagRun _can_ happen), and calculate that against `DagRun.start_date`. That value is always available and would enable us to remove the block below.
` @classmethod
def exact(cls, at: DateTime) -> "DagRunInfo":
"""Represent a run on an exact time."""
return cls(run_after=at, data_interval=DataInterval.exact(at))
@classmethod
def interval(cls, start: DateTime, end: DateTime) -> "DagRunInfo":
"""Represent a run on a continuous schedule.
In such a schedule, each data interval starts right after the previous
one ends, and each run is scheduled right after the interval ends. This
applies to all schedules prior to AIP-39 except ``@once`` and ``None``.
"""
return cls(run_after=end, data_interval=DataInterval(start, end))`
It seems that these two methods above in `base.py` are called to assign value to `DagRunInfo.run_after`. If a new field is added in `DagRun` as the suggestion indicated above, may I ask if there is a good way to make sure that the new field is consistent with `DagRunInfo.run_after`? Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on issue #19286: metric: Time after triggering
Posted by GitBox <gi...@apache.org>.
uranusjr commented on issue #19286:
URL: https://github.com/apache/airflow/issues/19286#issuecomment-975147311
All `DagRun` instances are populated from `DagRunInfo`. Actually, `DagRunInfo` is already used to populate some fields on `DagModel` that are used to schedule the next DAG run:
https://github.com/apache/airflow/blob/main/airflow/models/dag.py#L2920-L2928
So what we need to do is to pass that information on when the scheduler decides to actually create the next DAG run.
A DAG run is created with `DAG.create_dagrun()`, so you need to add an argument on that function, and change all `create_dagrun()` calls to use that argument. This will allow `DagRun.run_after` to be populated by `DagRunInfo.run_after` and thus in sync.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on issue #19286: metric: Time after triggering
Posted by GitBox <gi...@apache.org>.
uranusjr commented on issue #19286:
URL: https://github.com/apache/airflow/issues/19286#issuecomment-976292739
BTW, for backwards compatibility reasons, both the `DagRun` column and `create_dagrun` argument need to be optional (None by default), and if it’s None, we’ll fall back to `data_interval_end` like we currently use (for lack of a better value).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] EricGao888 commented on issue #19286: metric: Time after triggering
Posted by GitBox <gi...@apache.org>.
EricGao888 commented on issue #19286:
URL: https://github.com/apache/airflow/issues/19286#issuecomment-976229933
> All `DagRun` instances are populated from `DagRunInfo`. Actually, `DagRunInfo` is already used to populate some fields on `DagModel` that are used to schedule the next DAG run:
>
> https://github.com/apache/airflow/blob/1b89e682a1a04adf2cf5ebb37a453e245a6e7191/airflow/models/dag.py#L2920-L2928
>
> So what we need to do is to pass that information on when the scheduler decides to actually create the next DAG run.
>
> A DAG run is created with `DAG.create_dagrun()`, so you need to add an argument on that function, and change all `create_dagrun()` calls to use that argument. This will allow `DagRun.run_after` to be populated by `DagRunInfo.run_after` and thus in sync.
Thanks for the detailed suggestions!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on issue #19286: metric: Time after triggering
Posted by GitBox <gi...@apache.org>.
uranusjr commented on issue #19286:
URL: https://github.com/apache/airflow/issues/19286#issuecomment-968885994
The TODO comment hints what a proper solution would be. We should add a new field on `DagRun` to store `DagRunInfo.run_after` (i.e. the earliest time the DagRun _can_ happen), and calculate that against `DagRun.start_date`. That value is always available and would enable us to remove the block below.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] EricGao888 edited a comment on issue #19286: metric: Time after triggering
Posted by GitBox <gi...@apache.org>.
EricGao888 edited a comment on issue #19286:
URL: https://github.com/apache/airflow/issues/19286#issuecomment-962999214
```
# dagrun.schedule_delay
def _update_state(dag: DAG, dag_run: DagRun):
dag_run.state = State.RUNNING
dag_run.start_date = timezone.utcnow()
if dag.timetable.periodic:
# TODO: Logically, this should be DagRunInfo.run_after, but the
# information is not stored on a DagRun, only before the actual
# execution on DagModel.next_dagrun_create_after. We should add
# a field on DagRun for this instead of relying on the run
# always happening immediately after the data interval.
expected_start_date = dag.get_run_data_interval(dag_run).end
schedule_delay = dag_run.start_date - expected_start_date
Stats.timing(f'dagrun.schedule_delay.{dag.dag_id}', schedule_delay)
# first_task_scheduling_delay
if not self.dag.timetable.periodic:
# We can't emit this metric if there is no following schedule to calculate from!
return
ordered_tis_by_start_date = [ti for ti in finished_tis if ti.start_date]
ordered_tis_by_start_date.sort(key=lambda ti: ti.start_date, reverse=False)
first_start_date = ordered_tis_by_start_date[0].start_date
if first_start_date:
# TODO: Logically, this should be DagRunInfo.run_after, but the
# information is not stored on a DagRun, only before the actual
# execution on DagModel.next_dagrun_create_after. We should add
# a field on DagRun for this instead of relying on the run
# always happening immediately after the data interval.
data_interval_end = dag.get_run_data_interval(self).end
true_delay = first_start_date - data_interval_end
if true_delay.total_seconds() > 0:
Stats.timing(f'dagrun.{dag.dag_id}.first_task_scheduling_delay', true_delay)
```
I think the above two code fragments copied from scheduler_job.py and dagrun.py calculate the metics for dagrun.schedule_delay.<dag_id> and dagrun.<dag_id>.first_task_scheduling_delay respectively. It seems if `dag.timetable.periodic` is false, these two metrics will not be calculated and we cannot achieve the delay for these kinds of dags.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on issue #19286: metric: Time after triggering
Posted by GitBox <gi...@apache.org>.
uranusjr commented on issue #19286:
URL: https://github.com/apache/airflow/issues/19286#issuecomment-954355936
This makes perfect sense to me. Feel free to submit a PR!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] EricGao888 edited a comment on issue #19286: metric: Time after triggering
Posted by GitBox <gi...@apache.org>.
EricGao888 edited a comment on issue #19286:
URL: https://github.com/apache/airflow/issues/19286#issuecomment-962999214
`
# dagrun.schedule_delay
def _update_state(dag: DAG, dag_run: DagRun):
dag_run.state = State.RUNNING
dag_run.start_date = timezone.utcnow()
if dag.timetable.periodic:
# TODO: Logically, this should be DagRunInfo.run_after, but the
# information is not stored on a DagRun, only before the actual
# execution on DagModel.next_dagrun_create_after. We should add
# a field on DagRun for this instead of relying on the run
# always happening immediately after the data interval.
expected_start_date = dag.get_run_data_interval(dag_run).end
schedule_delay = dag_run.start_date - expected_start_date
Stats.timing(f'dagrun.schedule_delay.{dag.dag_id}', schedule_delay)
# first_task_scheduling_delay
if not self.dag.timetable.periodic:
# We can't emit this metric if there is no following schedule to calculate from!
return
ordered_tis_by_start_date = [ti for ti in finished_tis if ti.start_date]
ordered_tis_by_start_date.sort(key=lambda ti: ti.start_date, reverse=False)
first_start_date = ordered_tis_by_start_date[0].start_date
if first_start_date:
# TODO: Logically, this should be DagRunInfo.run_after, but the
# information is not stored on a DagRun, only before the actual
# execution on DagModel.next_dagrun_create_after. We should add
# a field on DagRun for this instead of relying on the run
# always happening immediately after the data interval.
data_interval_end = dag.get_run_data_interval(self).end
true_delay = first_start_date - data_interval_end
if true_delay.total_seconds() > 0:
Stats.timing(f'dagrun.{dag.dag_id}.first_task_scheduling_delay', true_delay)
`
I think the above two code fragments copied from scheduler_job.py and dagrun.py calculate the metics for dagrun.schedule_delay.<dag_id> and dagrun.<dag_id>.first_task_scheduling_delay respectively. It seems if `dag.timetable.periodic` is false, these two metrics will not be calculated and we cannot achieve the delay for these kinds of dags.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] EricGao888 commented on issue #19286: metric: Time after triggering
Posted by GitBox <gi...@apache.org>.
EricGao888 commented on issue #19286:
URL: https://github.com/apache/airflow/issues/19286#issuecomment-986380136
> BTW, for backwards compatibility reasons, both the `DagRun` column and `create_dagrun` argument need to be optional (None by default), and if it’s None, we’ll fall back to `data_interval_end` like we currently use (for lack of a better value).
After I added the `run_after` field to `dagrun` and changed the related logic to calculate the scheduling delay metric, I got blocked by the following error when initializing db:
`(airflow_dev) ➜ ~ airflow standalone
standalone | Starting Airflow Standalone
standalone | Checking database is initialized
INFO [alembic.runtime.migration] Context impl SQLiteImpl.
INFO [alembic.runtime.migration] Will assume non-transactional DDL.
WARNI [airflow.models.crypto] empty cryptography key - values will not be stored encrypted.
Traceback (most recent call last):
File "/opt/anaconda3/envs/airflow_dev/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1277, in _execute_context
cursor, statement, parameters, context
File "/opt/anaconda3/envs/airflow_dev/lib/python3.6/site-packages/sqlalchemy/engine/default.py", line 608, in do_execute
cursor.execute(statement, parameters)
sqlite3.OperationalError: no such column: dag_run.run_after`
It seems somehow the new field was not mapped to a column when creating the table. May I ask which part of the code should I look into to fix this? Any help will be appreciated!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on issue #19286: metric: Time after triggering
Posted by GitBox <gi...@apache.org>.
uranusjr commented on issue #19286:
URL: https://github.com/apache/airflow/issues/19286#issuecomment-986507656
You need to create a migration file. See https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#metadata-database-updates
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org