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/03/24 09:32:22 UTC
[GitHub] [airflow] koliankolin opened a new issue #14975: How to trigger dag manually in 2.0.1 version with pickled dags
koliankolin opened a new issue #14975:
URL: https://github.com/apache/airflow/issues/14975
**Apache Airflow version**: 2.0.1
**Kubernetes version (if you are using kubernetes)** (use `kubectl version`): No
**Environment**:
- **Cloud provider or hardware configuration**: Local Docker 7GB RAM, 2 CPU
- **OS** (e.g. from /etc/os-release): Debian GNU/Linux
- **Kernel** (e.g. `uname -a`): Linux 97bb49fdb749 4.19.121-linuxkit
- **Install tools**: pip
- **Others**:
**What happened**:
I tried to trigger dag manually from UI which was pickled in meta DB and it crashes.
**What you expected to happen**:
All tasks are running and executed successfully
**How to reproduce it**:
1) Pickle dag with DagPickle model. For example, if we have instance of created dag we execute dag.pickle()
2) Trigger Dag which was pickled from UI
Using `docker logs -f name_of_container` I see next logs:
```
[2021-03-19 09:34:54,735] {scheduler_job.py:975} INFO - Figuring out tasks to run in Pool(name=default_pool) with 128 open slots and 1 task instances ready to be queued
[2021-03-19 09:34:54,736] {scheduler_job.py:1002} INFO - DAG dummy has 0/16 running and queued tasks
[2021-03-19 09:34:54,737] {scheduler_job.py:1063} INFO - Setting the following tasks to queued state:
<TaskInstance: dummy.init 2021-03-19 09:34:53.681867+00:00 [scheduled]>
[2021-03-19 09:34:54,743] {scheduler_job.py:1105} INFO - Sending TaskInstanceKey(dag_id='dummy', task_id='init', execution_date=datetime.datetime(2021, 3, 19, 9, 34, 53, 681867, tzinfo=Timezone('UTC')), try_number=1) to executor with priority 7 and queue default
[2021-03-19 09:34:54,744] {base_executor.py:82} INFO - Adding to queue: ['airflow', 'tasks', 'run', 'dummy', 'init', '2021-03-19T09:34:53.681867+00:00', '--pickle', '141', '--local', '--pool', 'default_pool', '--subdir', '/usr/local/airflow/dags/dag_generator.py']
[2021-03-19 09:34:54,749] {local_executor.py:81} INFO - QueuedLocalWorker running ['airflow', 'tasks', 'run', 'dummy', 'init', '2021-03-19T09:34:53.681867+00:00', '--pickle', '141', '--local', '--pool', 'default_pool', '--subdir', '/usr/local/airflow/dags/dag_generator.py']
Loading pickle id: 141
Running <TaskInstance: dummy.init 2021-03-19T09:34:53.681867+00:00 [queued]> on host 49442bde5388
[2021-03-19 09:34:58,081] {scheduler_job.py:1206} INFO - Executor reports execution of dummy.init execution_date=2021-03-19 09:34:53.681867+00:00 exited with status success for try_number 1
[2021-03-19 09:34:58 +0000] [550] [INFO] Handling signal: ttin
[2021-03-19 09:34:58 +0000] [37648] [INFO] Booting worker with pid: 37648
[2021-03-19 09:35:05 +0000] [550] [INFO] Handling signal: ttou
[2021-03-19 09:35:05 +0000] [32497] [INFO] Worker exiting (pid: 32497)
[2021-03-19 09:35:06,926] {scheduler_job.py:941} INFO - 1 tasks up for execution:
<TaskInstance: dummy.init 2021-03-19 09:34:53.681867+00:00 [scheduled]>
[2021-03-19 09:35:06,929] {scheduler_job.py:975} INFO - Figuring out tasks to run in Pool(name=default_pool) with 128 open slots and 1 task instances ready to be queued
[2021-03-19 09:35:06,929] {scheduler_job.py:1002} INFO - DAG dummy has 0/16 running and queued tasks
[2021-03-19 09:35:06,929] {scheduler_job.py:1063} INFO - Setting the following tasks to queued state:
<TaskInstance: dummy.init 2021-03-19 09:34:53.681867+00:00 [scheduled]>
[2021-03-19 09:35:06,931] {scheduler_job.py:1105} INFO - Sending TaskInstanceKey(dag_id='dummy', task_id='init', execution_date=datetime.datetime(2021, 3, 19, 9, 34, 53, 681867, tzinfo=Timezone('UTC')), try_number=2) to executor with priority 7 and queue default
[2021-03-19 09:35:06,931] {base_executor.py:82} INFO - Adding to queue: ['airflow', 'tasks', 'run', 'dummy', 'init', '2021-03-19T09:34:53.681867+00:00', '--pickle', '141', '--local', '--pool', 'default_pool', '--subdir', '/usr/local/airflow/dags/dag_generator.py']
[2021-03-19 09:35:06,935] {local_executor.py:81} INFO - QueuedLocalWorker running ['airflow', 'tasks', 'run', 'dummy', 'init', '2021-03-19T09:34:53.681867+00:00', '--pickle', '141', '--local', '--pool', 'default_pool', '--subdir', '/usr/local/airflow/dags/dag_generator.py']
Loading pickle id: 141
Running <TaskInstance: dummy.init 2021-03-19T09:34:53.681867+00:00 [queued]> on host 49442bde5388
[2021-03-19 09:35:09,827] {scheduler_job.py:1206} INFO - Executor reports execution of dummy.init execution_date=2021-03-19 09:34:53.681867+00:00 exited with status success for try_number 2
[2021-03-19 09:35:18,884] {dagrun.py:430} ERROR - Marking run <DagRun dummy @ 2021-03-19 09:34:53.681867+00:00: manual__2021-03-19T09:34:53.681867+00:00, externally triggered: True> failed
```
And such kind of logs from failed task from UI:
```
*** Reading local file: /usr/local/airflow/logs/dummy/init/4-20210319T093453/2.log
>2021-03-19 09:35:09,215|INFO|taskinstance.py:851|Dependencies all met for <TaskInstance: dummy.init 2021-03-19T09:34:53.681867+00:00 [queued]>
>2021-03-19 09:35:09,231|INFO|taskinstance.py:851|Dependencies all met for <TaskInstance: dummy.init 2021-03-19T09:34:53.681867+00:00 [queued]>
>2021-03-19 09:35:09,231|INFO|taskinstance.py:1042|
--------------------------------------------------------------------------------
>2021-03-19 09:35:09,231|INFO|taskinstance.py:1043|Starting attempt 2 of 2
>2021-03-19 09:35:09,231|INFO|taskinstance.py:1044|
--------------------------------------------------------------------------------
>2021-03-19 09:35:09,247|INFO|taskinstance.py:1063|Executing <Task(PythonOperator): init> on 2021-03-19T09:34:53.681867+00:00
>2021-03-19 09:35:09,252|INFO|standard_task_runner.py:52|Started process 38115 to run task
>2021-03-19 09:35:09,257|INFO|standard_task_runner.py:76|Running: ['airflow', 'tasks', 'run', 'dummy', 'init', '2021-03-19T09:34:53.681867+00:00', '--pickle', '141', '--job-id', '25', '--pool', 'default_pool', '--raw', '--cfg-path', '/tmp/tmpo6temkwx', '--error-file', '/tmp/tmpv5y3gphw']
>2021-03-19 09:35:09,258|INFO|standard_task_runner.py:77|Job 25: Subtask init
>2021-03-19 09:35:09,284|INFO|local_task_job.py:146|Task exited with return code 1
```
**Anything else we need to know**:
**When I cleaned dag_pickle table in meta DB and clean last_pickle column in table dag and don't pickle dags at all everything is working well.**
To approve this statement show logs where flag --pickle wasn't used and everything worked successfully:
```
[2021-03-19 10:22:15,074] {scheduler_job.py:941} INFO - 1 tasks up for execution:
<TaskInstance: dummy.init 2021-03-19 10:21:11.696790+00:00 [scheduled]>
[2021-03-19 10:22:15,080] {scheduler_job.py:975} INFO - Figuring out tasks to run in Pool(name=default_pool) with 128 open slots and 1 task instances ready to be queued
[2021-03-19 10:22:15,082] {scheduler_job.py:1002} INFO - DAG dummy has 0/16 running and queued tasks
[2021-03-19 10:22:15,085] {scheduler_job.py:1063} INFO - Setting the following tasks to queued state:
<TaskInstance: dummy.init 2021-03-19 10:21:11.696790+00:00 [scheduled]>
[2021-03-19 10:22:15,089] {scheduler_job.py:1105} INFO - Sending TaskInstanceKey(dag_id='dummy', task_id='init', execution_date=datetime.datetime(2021, 3, 19, 10, 21, 11, 696790, tzinfo=Timezone('UTC')), try_number=1) to executor with priority 7 and queue default
[2021-03-19 10:22:15,090] {base_executor.py:82} INFO - Adding to queue: ['airflow', 'tasks', 'run', 'dummy', 'init', '2021-03-19T10:21:11.696790+00:00', '--local', '--pool', 'default_pool', '--subdir', '/usr/local/airflow/dags/dag_generator.py']
[2021-03-19 10:22:15,096] {local_executor.py:81} INFO - QueuedLocalWorker running ['airflow', 'tasks', 'run', 'dummy', 'init', '2021-03-19T10:21:11.696790+00:00', '--local', '--pool', 'default_pool', '--subdir', '/usr/local/airflow/dags/dag_generator.py']
[2021-03-19 10:22:15,320] {dagbag.py:448} INFO - Filling up the DagBag from /usr/local/airflow/dags/dag_generator.py
Running <TaskInstance: dummy.init 2021-03-19T10:21:11.696790+00:00 [queued]> on host 49442bde5388
[2021-03-19 10:22:18,384] {scheduler_job.py:941} INFO - 2 tasks up for execution:
<TaskInstance: dummy.df_now1 2021-03-19 10:21:11.696790+00:00 [scheduled]>
<TaskInstance: dummy.df_now2 2021-03-19 10:21:11.696790+00:00 [scheduled]>
[2021-03-19 10:22:18,386] {scheduler_job.py:975} INFO - Figuring out tasks to run in Pool(name=default_pool) with 128 open slots and 2 task instances ready to be queued
[2021-03-19 10:22:18,387] {scheduler_job.py:1002} INFO - DAG dummy has 0/16 running and queued tasks
[2021-03-19 10:22:18,387] {scheduler_job.py:1002} INFO - DAG dummy has 1/16 running and queued tasks
```
And also logs from UI about this task:
```
*** Reading local file: /usr/local/airflow/logs/dummy/init/5-20210319T102111/1.log
>2021-03-19 10:22:17,638|INFO|taskinstance.py:851|Dependencies all met for <TaskInstance: dummy.init 2021-03-19T10:21:11.696790+00:00 [queued]>
>2021-03-19 10:22:17,654|INFO|taskinstance.py:851|Dependencies all met for <TaskInstance: dummy.init 2021-03-19T10:21:11.696790+00:00 [queued]>
>2021-03-19 10:22:17,654|INFO|taskinstance.py:1042|
--------------------------------------------------------------------------------
>2021-03-19 10:22:17,654|INFO|taskinstance.py:1043|Starting attempt 1 of 2
>2021-03-19 10:22:17,654|INFO|taskinstance.py:1044|
--------------------------------------------------------------------------------
>2021-03-19 10:22:17,669|INFO|taskinstance.py:1063|Executing <Task(PythonOperator): init> on 2021-03-19T10:21:11.696790+00:00
>2021-03-19 10:22:17,674|INFO|standard_task_runner.py:52|Started process 6413 to run task
>2021-03-19 10:22:17,682|INFO|standard_task_runner.py:76|Running: ['airflow', 'tasks', 'run', 'dummy', 'init', '2021-03-19T10:21:11.696790+00:00', '--job-id', '26', '--pool', 'default_pool', '--raw', '--subdir', 'DAGS_FOLDER/dag_generator.py', '--cfg-path', '/tmp/tmpqkmkygyb', '--error-file', '/tmp/tmp2k6izpaa']
>2021-03-19 10:22:17,683|INFO|standard_task_runner.py:77|Job 26: Subtask init
>2021-03-19 10:22:17,733|INFO|logging_mixin.py:104|Running <TaskInstance: dummy.init 2021-03-19T10:21:11.696790+00:00 [running]> on host 49442bde5388
>2021-03-19 10:22:17,777|INFO|taskinstance.py:1257|Exporting the following env vars:
AIRFLOW_CTX_DAG_OWNER=airflow
AIRFLOW_CTX_DAG_ID=dummy
AIRFLOW_CTX_TASK_ID=init
AIRFLOW_CTX_EXECUTION_DATE=2021-03-19T10:21:11.696790+00:00
AIRFLOW_CTX_DAG_RUN_ID=manual__2021-03-19T10:21:11.696790+00:00
>2021-03-19 10:22:17,778|INFO|python.py:118|Done. Returned value was: None
>2021-03-19 10:22:17,787|INFO|taskinstance.py:1166|Marking task as SUCCESS. dag_id=dummy, task_id=init, execution_date=20210319T102111, start_date=20210319T102217, end_date=20210319T102217
>2021-03-19 10:22:17,818|INFO|taskinstance.py:1220|2 downstream tasks scheduled from follow-on schedule check
>2021-03-19 10:22:17,853|INFO|local_task_job.py:146|Task exited with return code 0
```
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] boring-cyborg[bot] commented on issue #14975: How to trigger dag manually in 2.0.1 version with pickled dags
Posted by GitBox <gi...@apache.org>.
boring-cyborg[bot] commented on issue #14975:
URL: https://github.com/apache/airflow/issues/14975#issuecomment-805645561
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] koliankolin commented on issue #14975: How to trigger dag manually in 2.0.1 version with pickled dags
Posted by GitBox <gi...@apache.org>.
koliankolin commented on issue #14975:
URL: https://github.com/apache/airflow/issues/14975#issuecomment-808853781
UPD: I found out reasons that cause this problem.
As I mentioned before bug is reproduced with filling pickle_id in dag table. For that purpose we use DagModel.
I should start with a fact that creating task_instance to launch happens differently. For example using DebugExecutor we import TaskInstance and create simple instance of it, using cli command (airflow tasks run --pickle 1) we also use logic as in DebugExecutor but when _scheduler_ is used logic is changed. In the file scheduler_job.py creating task_instance objects is placed in
method **airflow.jobs.scheduler_job.SchedulerJob._executable_task_instances_to_queued** and there is such kind of code:
```
query = (
session.query(TI)
.outerjoin(TI.dag_run)
.filter(or_(DR.run_id.is_(None), DR.run_type != DagRunType.BACKFILL_JOB))
.join(TI.dag_model)
.filter(not_(DM.is_paused))
.filter(TI.state == State.SCHEDULED)
.options(selectinload('dag_model'))
.limit(max_tis)
)
```
As you can see we can reproduce bug only using scheduler because only in this case we enrich our task_instance with DagModel. In different words only using _scheduler_ we add pickle_id to task_instance. Ok, keep this fact in mind and move on :)
Next we use method **airflow.models.taskinstance.TaskInstance.generate_command** to create list with command and previously kept fact we get such kind of list:
```
['airflow', 'tasks', 'run', 'dummy', 'init', '2021-03-26T11:30:20.470827+00:00', '--pickle', '11', '--local', '--pool', 'default_pool', '--subdir', '/usr/local/airflow/dags/dag_generator.py']
```
Ok, after forming this command list instance of **airflow.task.task_runner.standard_task_runner.StandardTaskRunner** is created and in our case method **airflow.task.task_runner.standard_task_runner.StandardTaskRunner._start_by_fork** is used to start task_instance. Next all magic is happened :)
In this method we feed argument parser with command list and go with this args to such kind of code:
```
parser = get_parser()
# [1:] - remove "airflow" from the start of the command
args = parser.parse_args(self._command[1:])
self.log.info('Running: %s', self._command)
self.log.info('Job %s: Subtask %s', self._task_instance.job_id, self._task_instance.task_id)
proc_title = "airflow task runner: {0.dag_id} {0.task_id} {0.execution_date}"
if hasattr(args, "job_id"):
proc_title += " {0.job_id}"
setproctitle(proc_title.format(args))
try:
args.func(args, dag=self.dag)
return_code = 0
except Exception: # pylint: disable=broad-except
return_code = 1
finally:
# Explicitly flush any pending exception to Sentry if enabled
Sentry.flush()
logging.shutdown()
os._exit(return_code) # pylint: disable=protected-access
```
Lines with try-except block are very interesting. **We can see that all kind of exceptions are excepted and nothing is written to logs.** I think that is not a good practice but I didn't fall into despair and added such kind of line in source code before return_code = 1:
```
self.log.error(e)
```
and triggered dag from UI. Several seconds after failing dag I saw such kind of logs in UI:
```
>2021-03-26 15:15:29,991|ERROR|standard_task_runner.py:88|You cannot use the --pickle option when using DAG.cli() method.
```
Hmmm... What a strange exception... And after some seeking I see that line of code:
```
if dag and args.pickle:
raise AirflowException("You cannot use the --pickle option when using DAG.cli() method.")
```
What does it mean? And this means that there is **logic bug** because enriching task_instance with pickle_id leads to this exception and this happens in that code lines of method **airflow.task.task_runner.standard_task_runner.StandardTaskRunner._start_by_fork**:
```
try:
args.func(args, dag=self.dag)
return_code = 0
```
exactly there --pickle flag and instance of dag is used together. I'm all.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] koliankolin commented on issue #14975: How to trigger dag manually in 2.0.1 version with pickled dags
Posted by GitBox <gi...@apache.org>.
koliankolin commented on issue #14975:
URL: https://github.com/apache/airflow/issues/14975#issuecomment-808884748
@turbaszek yes, of course I'll try to do my best to fix it as soon as possible
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on issue #14975: How to trigger dag manually in 2.0.1 version with pickled dags
Posted by GitBox <gi...@apache.org>.
turbaszek commented on issue #14975:
URL: https://github.com/apache/airflow/issues/14975#issuecomment-808879835
@koliankolin would you be interested in opening a PR with a fix?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal commented on issue #14975: How to trigger dag manually in 2.0.1 version with pickled dags
Posted by GitBox <gi...@apache.org>.
eladkal commented on issue #14975:
URL: https://github.com/apache/airflow/issues/14975#issuecomment-927268940
@koliankolin are you still working on it?
--
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