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/09/14 12:10:08 UTC
[GitHub] [airflow] Jorricks opened a new pull request #17207: Fix external_executor_id not being set for manually run jobs.
Jorricks opened a new pull request #17207:
URL: https://github.com/apache/airflow/pull/17207
**Summary of the problem:**
Currently the celery `task_id` is being stored only after a scheduler launched a task in its executor.
Then the celery `task_id` is being put on the `event_buffer` and the scheduler periodically reads out the `event_buffer` and stores the `external_executor_id`.
Because manually trigger tasks enter the adoption flow -- as their executor instances are only there for the launching of that one specific tasks -- and the `external_executor_id` is not set, they won't get adopted. Instead they get killed. Meaning, any manually triggered task that doesn't have an `external_executor_id` from a previous scheduled run before being launched, might get killed if the adoption routines kicks in while the task is still running.
**Solution:**
I could imagine two solution:
1. After every manually triggered task, we read the event_buffer and store that in the task instances.
2. Every task that is triggered for Celery Executors automatically stores its `external_executor_id` at the start up of the task.
I implemented both but found the second version nicer.
I am looking for some feedback so please provide me with any you can think of :)
**Opened issues that are related:**
related: https://github.com/apache/airflow/issues/16023
--
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] Jorricks edited a comment on pull request #17207: WIP: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks edited a comment on pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#issuecomment-901830794
> I feel we should not make this change but dig more to reliably reproduce the linked issue.
> On my first try, I was able to see `Scheduler marked as failed` but on several tries now, I can't reproduce it. I ended up having task stuck in running which is related to another issue(Trying to find how I got to that)
First of all, thanks for taking a look at the PR!
There are a couple things that need to hold for you to be able to reproduce it:
- Before starting the task, the task should not have run before or at least the `external_executor_id` must be None.
- The task must still be running when the adoption flow of one of the schedulers kicks in.
I am able to reliable reproduce this issue when these two things are held.
However, I am not really sure what you mean with dig deeper. Unless the whole executor part is rewritten, this is the best I could come up with. Could you please be a bit more explicit?
--
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] ephraimbuddy commented on a change in pull request #17207: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on a change in pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#discussion_r707321847
##########
File path: tests/executors/test_celery_executor.py
##########
@@ -71,6 +73,7 @@ def _prepare_app(broker_url=None, execute=None):
test_config = dict(celery_executor.celery_configuration)
test_config.update({'broker_url': broker_url})
test_app = Celery(broker_url, config_source=test_config)
+ mock.patch(test_app)
Review comment:
Can you explain why we have this here?
--
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] ephraimbuddy commented on pull request #17207: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#issuecomment-919898183
cc: @ashb @kaxil
--
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] ashb commented on pull request #17207: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
ashb commented on pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#issuecomment-919962266
Should be fixed on main now, so re-triggering build :)
--
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] Jorricks commented on a change in pull request #17207: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks commented on a change in pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#discussion_r707032501
##########
File path: tests/executors/test_celery_executor.py
##########
@@ -288,6 +291,9 @@ def test_command_validation(self, command, expected_exception):
with mock.patch(
'airflow.executors.celery_executor._execute_in_subprocess'
) as mock_subproc, mock.patch('airflow.executors.celery_executor._execute_in_fork') as mock_fork:
+ test_app = MagicMock()
+ test_app.current_task.request.id.side_effect = "abcdef-124215-abcdef"
+ mock.patch('airflow.executors.celery_executor.app', test_app)
Review comment:
I don't really like having two mock.patches on the same line. I choose not to use three because it becomes more unreadable the more you add. I think it's better the way it is already.
--
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] ephraimbuddy commented on pull request #17207: WIP: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#issuecomment-901928314
> > I feel we should not make this change but dig more to reliably reproduce the linked issue.
> > On my first try, I was able to see `Scheduler marked as failed` but on several tries now, I can't reproduce it. I ended up having task stuck in running which is related to another issue(Trying to find how I got to that)
>
> First of all, thanks for taking a look at the PR!
>
> There are a couple things that need to hold for you to be able to reproduce it:
>
> * Before starting the task, the task should not have run before or at least the `external_executor_id` must be None.
> * The task must still be running when the adoption flow of one of the schedulers kicks in.
>
> I am able to reliable reproduce this issue when these two things are held.
>
> However, I am not really sure what you mean with dig deeper. Unless the whole executor part is rewritten, this is the best I could come up with. Could you please be a bit more explicit?
I did reproduce it initially but can't reproduce it again. Have done `airflow db reset` in `breeze` but can't reproduce it. Well, I do not fully understand this area, let's hear from others
--
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] Jorricks commented on pull request #17207: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks commented on pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#issuecomment-919083153
Can someone please re-trigger the pipeline :)?
--
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] Jorricks commented on a change in pull request #17207: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks commented on a change in pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#discussion_r707032501
##########
File path: tests/executors/test_celery_executor.py
##########
@@ -288,6 +291,9 @@ def test_command_validation(self, command, expected_exception):
with mock.patch(
'airflow.executors.celery_executor._execute_in_subprocess'
) as mock_subproc, mock.patch('airflow.executors.celery_executor._execute_in_fork') as mock_fork:
+ test_app = MagicMock()
+ test_app.current_task.request.id.side_effect = "abcdef-124215-abcdef"
+ mock.patch('airflow.executors.celery_executor.app', test_app)
Review comment:
I don't really like having two mock.patches on the same line. I choose not to use three because it becomes more unreadable the more you add. I think it's better the way it is already.
However, I don't really have a strong opinion on it, so if you would like me to change it, let me know.
--
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] ephraimbuddy commented on a change in pull request #17207: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on a change in pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#discussion_r706633517
##########
File path: tests/executors/test_celery_executor.py
##########
@@ -288,6 +291,9 @@ def test_command_validation(self, command, expected_exception):
with mock.patch(
'airflow.executors.celery_executor._execute_in_subprocess'
) as mock_subproc, mock.patch('airflow.executors.celery_executor._execute_in_fork') as mock_fork:
+ test_app = MagicMock()
+ test_app.current_task.request.id.side_effect = "abcdef-124215-abcdef"
+ mock.patch('airflow.executors.celery_executor.app', test_app)
Review comment:
I think we can approach this like this:
```python
with mock.patch(
'airflow.executors.celery_executor._execute_in_subprocess'
) as mock_subproc, mock.patch('airflow.executors.celery_executor._execute_in_fork') as mock_fork,
mock.patch('airflow.executors.celery_executor.app') as test_app:
test_app.current_task.request.id.side_effect = "abcdef-124215-abcdef"
```
WDYT?
##########
File path: tests/executors/test_celery_executor.py
##########
@@ -71,6 +73,7 @@ def _prepare_app(broker_url=None, execute=None):
test_config = dict(celery_executor.celery_configuration)
test_config.update({'broker_url': broker_url})
test_app = Celery(broker_url, config_source=test_config)
+ mock.patch(test_app)
Review comment:
```suggestion
```
Do we need it here?
##########
File path: tests/cli/commands/test_task_command.py
##########
@@ -106,7 +106,41 @@ def test_test_with_existing_dag_run(self):
task_command.task_test(args)
# Check that prints, and log messages, are shown
- assert f"Marking task as SUCCESS. dag_id={self.dag_id}, task_id={task_id}" in stdout.getvalue()
+ assert f"Marking task as SUCCESS. dag_id={dag_id}, task_id={task_id}" in stdout.getvalue()
Review comment:
```suggestion
assert f"Marking task as SUCCESS. dag_id={self.dag_id}, task_id={task_id}" in stdout.getvalue()
```
Any reason for the change?
--
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] Jorricks commented on pull request #17207: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks commented on pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#issuecomment-919131029
AFAIK, the failures in the tests are not related to this 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] ephraimbuddy commented on a change in pull request #17207: WIP: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on a change in pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#discussion_r693832397
##########
File path: airflow/executors/celery_executor.py
##########
@@ -125,8 +129,10 @@ def _execute_in_fork(command_to_exec: CommandType) -> None:
os._exit(ret)
-def _execute_in_subprocess(command_to_exec: CommandType) -> None:
+def _execute_in_subprocess(command_to_exec: CommandType, celery_task_id: Optional[str] = None) -> None:
env = os.environ.copy()
+ if celery_task_id:
+ env["celery_task_id"] = celery_task_id
Review comment:
Looks like we should do `env['external_executor_id'] = celery_task` here, right 🤔?
--
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] ephraimbuddy removed a comment on pull request #17207: WIP: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
ephraimbuddy removed a comment on pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#issuecomment-902985223
Now, I reproduced this consistently with LocalExecutor in breeze by running the example DAG below.
```
from airflow import DAG
import os
from datetime import datetime, timedelta
from airflow.operators.python import PythonOperator
import time
dag = DAG(os.path.basename(__file__).replace('.py', ''),
start_date=datetime(2021, 5, 11),
schedule_interval=timedelta(days=1))
def sleep_tester(time_out, retries):
for i in range(retries):
print(f'hi there, try {i}, going to sleep for {time_out}')
time.sleep(time_out)
print("Aaah, good times, see ya soon")
sleeping = PythonOperator(task_id="sleep_well",
python_callable=sleep_tester,
op_kwargs={'time_out': 15, 'retries': 50},
dag=dag)
```
The first 16 tasks that were queued ran successfully but when the next 16 was to be run, all the 16 tasks failed.
This issue looks related to task concurrency.
Try if you can reproduce it too with LocalExecutor. Run the dag as is and allow it to run without stopping it. It will take time but just allow it to run
--
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] Jorricks commented on a change in pull request #17207: WIP: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks commented on a change in pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#discussion_r694076466
##########
File path: airflow/executors/celery_executor.py
##########
@@ -125,8 +129,10 @@ def _execute_in_fork(command_to_exec: CommandType) -> None:
os._exit(ret)
-def _execute_in_subprocess(command_to_exec: CommandType) -> None:
+def _execute_in_subprocess(command_to_exec: CommandType, celery_task_id: Optional[str] = None) -> None:
env = os.environ.copy()
+ if celery_task_id:
+ env["celery_task_id"] = celery_task_id
Review comment:
You are absolutely right 🥇
I will add tests ones I find the time this week.
--
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] Jorricks commented on a change in pull request #17207: WIP: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks commented on a change in pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#discussion_r693824593
##########
File path: airflow/executors/celery_executor.py
##########
@@ -125,8 +129,10 @@ def _execute_in_fork(command_to_exec: CommandType) -> None:
os._exit(ret)
-def _execute_in_subprocess(command_to_exec: CommandType) -> None:
+def _execute_in_subprocess(command_to_exec: CommandType, celery_task_id: Optional[str] = None) -> None:
env = os.environ.copy()
+ if celery_task_id:
+ env["celery_task_id"] = celery_task_id
Review comment:
@ephraimbuddy this is the missing environment variable being set :)
--
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] Jorricks commented on a change in pull request #17207: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks commented on a change in pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#discussion_r707031769
##########
File path: tests/cli/commands/test_task_command.py
##########
@@ -106,7 +106,41 @@ def test_test_with_existing_dag_run(self):
task_command.task_test(args)
# Check that prints, and log messages, are shown
- assert f"Marking task as SUCCESS. dag_id={self.dag_id}, task_id={task_id}" in stdout.getvalue()
+ assert f"Marking task as SUCCESS. dag_id={dag_id}, task_id={task_id}" in stdout.getvalue()
Review comment:
I am pretty sure that was automatically done by me re-basing. I definitely didn't do that by hand at least :p
--
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] ephraimbuddy commented on a change in pull request #17207: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on a change in pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#discussion_r707038035
##########
File path: tests/executors/test_celery_executor.py
##########
@@ -288,6 +291,9 @@ def test_command_validation(self, command, expected_exception):
with mock.patch(
'airflow.executors.celery_executor._execute_in_subprocess'
) as mock_subproc, mock.patch('airflow.executors.celery_executor._execute_in_fork') as mock_fork:
+ test_app = MagicMock()
+ test_app.current_task.request.id.side_effect = "abcdef-124215-abcdef"
+ mock.patch('airflow.executors.celery_executor.app', test_app)
Review comment:
I don't have strong opinion here too
--
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] Jorricks edited a comment on pull request #17207: WIP: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks edited a comment on pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#issuecomment-901830794
> I feel we should not make this change but dig more to reliably reproduce the linked issue.
> On my first try, I was able to see `Scheduler marked as failed` but on several tries now, I can't reproduce it. I ended up having task stuck in running which is related to another issue(Trying to find how I got to that)
There are a couple things that need to hold for you to be able to reproduce it:
- Before starting the task, the task should not have run before or at least the `external_executor_id` must be None.
- The task must still be running when the adoption flow of one of the schedulers kicks in.
I am able to reliable reproduce this issue when these two things are held.
However, I am not really sure what you mean with dig deeper. Unless the whole executor part is rewritten, this is the best I could come up with.
--
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] Jorricks commented on a change in pull request #17207: WIP: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks commented on a change in pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#discussion_r677148619
##########
File path: airflow/cli/commands/task_command.py
##########
@@ -174,6 +175,14 @@ def _run_raw_task(args, ti: TaskInstance) -> None:
)
+def _extract_external_executor_id(args) -> Optional[str]:
+ if "external_executor_id" in args:
+ return args.external_executor_id
+ elif "external_executor_id" in os.environ:
+ return args.external_executor_id
Review comment:
Good catch! Changed
##########
File path: airflow/cli/commands/task_command.py
##########
@@ -174,6 +175,14 @@ def _run_raw_task(args, ti: TaskInstance) -> None:
)
+def _extract_external_executor_id(args) -> Optional[str]:
+ if "external_executor_id" in args:
+ return args.external_executor_id
+ elif "external_executor_id" in os.environ:
+ return args.external_executor_id
Review comment:
Good catch! Will change that.
--
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] Jorricks commented on a change in pull request #17207: WIP: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks commented on a change in pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#discussion_r679388196
##########
File path: airflow/cli/commands/task_command.py
##########
@@ -174,6 +175,14 @@ def _run_raw_task(args, ti: TaskInstance) -> None:
)
+def _extract_external_executor_id(args) -> Optional[str]:
+ if "external_executor_id" in args:
+ return args.external_executor_id
+ elif "external_executor_id" in os.environ:
+ return args.external_executor_id
Review comment:
Fixed the reference.
--
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] Jorricks commented on pull request #17207: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks commented on pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#issuecomment-919455634
Rebased on latest main again to see if that makes it better.
--
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] Jorricks commented on pull request #17207: WIP: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks commented on pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#issuecomment-901830794
> I feel we should not make this change but dig more to reliably reproduce the linked issue.
> On my first try, I was able to see `Scheduler marked as failed` but on several tries now, I can't reproduce it. I ended up having task stuck in running which is related to another issue(Trying to find how I got to that)
There are a couple things that need to hold for you to be able to reproduce it:
- Before starting the task, the task should not have run before or at least the `external_executor_id` must be None.
- The task must still be running when the adoption flow of one of the schedulers kicks in.
I am able to reliable reproduce this issue when these two things are held.
--
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] Jorricks commented on a change in pull request #17207: WIP: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks commented on a change in pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#discussion_r694266238
##########
File path: airflow/executors/celery_executor.py
##########
@@ -125,8 +129,10 @@ def _execute_in_fork(command_to_exec: CommandType) -> None:
os._exit(ret)
-def _execute_in_subprocess(command_to_exec: CommandType) -> None:
+def _execute_in_subprocess(command_to_exec: CommandType, celery_task_id: Optional[str] = None) -> None:
env = os.environ.copy()
+ if celery_task_id:
+ env["celery_task_id"] = celery_task_id
Review comment:
This should now be fixed :)
--
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] ephraimbuddy closed pull request #17207: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
ephraimbuddy closed pull request #17207:
URL: https://github.com/apache/airflow/pull/17207
--
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] Jorricks commented on pull request #17207: WIP: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks commented on pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#issuecomment-903019652
@ephraimbuddy please make sure you are using a form of a Celery Executor, as this ticket is only for celery setups.
--
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] Jorricks edited a comment on pull request #17207: WIP: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks edited a comment on pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#issuecomment-901830794
> I feel we should not make this change but dig more to reliably reproduce the linked issue.
> On my first try, I was able to see `Scheduler marked as failed` but on several tries now, I can't reproduce it. I ended up having task stuck in running which is related to another issue(Trying to find how I got to that)
First of all, thanks for taking a look at the PR!
There are a couple things that need to hold for you to be able to reproduce it:
- Before starting the task, the task should not have run before or at least the `external_executor_id` must be None.
- The task must still be running when the adoption flow of one of the schedulers kicks in.
I am able to reliable reproduce this issue when these two things are held.
However, I am not really sure what you mean with dig deeper. Unless the whole executor part is rewritten, this is the best I could come up with.
--
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] ephraimbuddy commented on pull request #17207: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#issuecomment-918179105
> So @ephraimbuddy, do you think it's ready to be merged now or?
Looks like you have test failure
--
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] ephraimbuddy commented on pull request #17207: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#issuecomment-915004710
You have some conflicts :)
--
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] ephraimbuddy commented on a change in pull request #17207: WIP: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on a change in pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#discussion_r693819227
##########
File path: airflow/cli/commands/task_command.py
##########
@@ -174,6 +175,12 @@ def _run_raw_task(args, ti: TaskInstance) -> None:
)
+def _extract_external_executor_id(args) -> Optional[str]:
+ if "external_executor_id" in args:
+ return args.external_executor_id
+ return os.environ.get("external_executor_id", None)
Review comment:
Looks like we didn't set `external_executor_id` in the environment, am I missing something?
--
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] Jorricks commented on pull request #17207: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks commented on pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#issuecomment-917430008
> You have some conflicts :)
Fixed.
--
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] Jorricks commented on pull request #17207: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks commented on pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#issuecomment-919732836
Yes the remaining failure is not related to this 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] github-actions[bot] commented on pull request #17207: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#issuecomment-919897261
The PR most likely needs to run full matrix of tests because it modifies parts of the core of Airflow. However, committers might decide to merge it quickly and take the risk. If they don't merge it quickly - please rebase it to the latest main at your convenience, or amend the last commit of the PR, and push it with --force-with-lease.
--
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] Jorricks commented on pull request #17207: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks commented on pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#issuecomment-918129975
So @ephraimbuddy, do you think it's ready to be merged now or?
--
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] ashb closed pull request #17207: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
ashb closed pull request #17207:
URL: https://github.com/apache/airflow/pull/17207
--
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] ephraimbuddy commented on pull request #17207: WIP: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#issuecomment-902985223
Now, I reproduced this consistently with LocalExecutor in breeze by running the example DAG below.
```
from airflow import DAG
import os
from datetime import datetime, timedelta
from airflow.operators.python import PythonOperator
import time
dag = DAG(os.path.basename(__file__).replace('.py', ''),
start_date=datetime(2021, 5, 11),
schedule_interval=timedelta(days=1))
def sleep_tester(time_out, retries):
for i in range(retries):
print(f'hi there, try {i}, going to sleep for {time_out}')
time.sleep(time_out)
print("Aaah, good times, see ya soon")
sleeping = PythonOperator(task_id="sleep_well",
python_callable=sleep_tester,
op_kwargs={'time_out': 15, 'retries': 50},
dag=dag)
```
The first 16 tasks that were queued ran successfully but when the next 16 was to be run, all the 16 tasks failed.
This issue looks related to task concurrency.
Try if you can reproduce it too with LocalExecutor. Run the dag as is and allow it to run without stopping it. It will take time but just allow it to run
--
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] Jorricks commented on pull request #17207: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks commented on pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#issuecomment-904437724
CI/CD failed on transient errors, not related to this 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] Jorricks commented on a change in pull request #17207: WIP: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks commented on a change in pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#discussion_r677264366
##########
File path: airflow/cli/commands/task_command.py
##########
@@ -174,6 +175,14 @@ def _run_raw_task(args, ti: TaskInstance) -> None:
)
+def _extract_external_executor_id(args) -> Optional[str]:
+ if "external_executor_id" in args:
+ return args.external_executor_id
+ elif "external_executor_id" in os.environ:
+ return args.external_executor_id
Review comment:
What did you think of the overal implementation @mik-laj ?
--
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] Jorricks commented on a change in pull request #17207: WIP: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks commented on a change in pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#discussion_r692015638
##########
File path: airflow/cli/commands/task_command.py
##########
@@ -174,6 +175,14 @@ def _run_raw_task(args, ti: TaskInstance) -> None:
)
+def _extract_external_executor_id(args) -> Optional[str]:
+ if "external_executor_id" in args:
+ return args.external_executor_id
+ elif "external_executor_id" in os.environ:
+ return os.environ.get("external_executor_id")
+ return None
Review comment:
Good suggestion! Missed that one.
--
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] Jorricks commented on pull request #17207: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks commented on pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#issuecomment-919129555
> You can actually retrigger it by closing and opening the PR
Didn't know I could re-open it myself. Thank you!
--
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] Jorricks commented on pull request #17207: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks commented on pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#issuecomment-906966271
> > I tried to add some tests on the CLI task part. That is pretty much done.
> > However, I had quite some trouble wrapping my head around a decent test approach on the celery_executor part.
> > There is currently not really a test for any of the functions I modified which makes me wonder if I should add them.
> > If so, do you have any remarks on how I could best do that?
>
> For the methods modified in CeleryExecutor, you can mock them and assert they were called with a celeryID.
> The app can be mocked to return an ID. See
>
> https://github.com/apache/airflow/blob/3d96ad62f91e662b62481441d8eec2994651e122/tests/executors/test_celery_executor.py#L75
>
>
> You can provide appID there.
I will do this after I am back from my vacation in about two weeks 👍.
--
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] Jorricks commented on a change in pull request #17207: WIP: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks commented on a change in pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#discussion_r692015638
##########
File path: airflow/cli/commands/task_command.py
##########
@@ -174,6 +175,14 @@ def _run_raw_task(args, ti: TaskInstance) -> None:
)
+def _extract_external_executor_id(args) -> Optional[str]:
+ if "external_executor_id" in args:
+ return args.external_executor_id
+ elif "external_executor_id" in os.environ:
+ return os.environ.get("external_executor_id")
+ return None
Review comment:
Good suggestion though.
--
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] Jorricks commented on pull request #17207: WIP: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks commented on pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#issuecomment-904077786
I tried to add some tests on the CLI task part. That is pretty much done.
However, I had quite some trouble wrapping my head around a decent test approach on the `celery_executor` part.
There is currently not really a test for any of the functions I modified which makes me wonder if I should add them.
If so, do you have any remarks on how I could best do that?
--
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] Jorricks commented on a change in pull request #17207: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks commented on a change in pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#discussion_r703439552
##########
File path: airflow/executors/celery_executor.py
##########
@@ -81,14 +81,16 @@ def execute_command(command_to_exec: CommandType) -> None:
"""Executes command."""
BaseExecutor.validate_command(command_to_exec)
log.info("Executing command in Celery: %s", command_to_exec)
+ celery_task_id = str(app.current_task.request.id) if app.current_task.request.id else None
+ log.info(f"Found the following celery_task_id: {celery_task_id}")
Review comment:
Changed :)
--
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] mik-laj commented on a change in pull request #17207: WIP: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#discussion_r677127098
##########
File path: airflow/cli/commands/task_command.py
##########
@@ -174,6 +175,14 @@ def _run_raw_task(args, ti: TaskInstance) -> None:
)
+def _extract_external_executor_id(args) -> Optional[str]:
+ if "external_executor_id" in args:
+ return args.external_executor_id
+ elif "external_executor_id" in os.environ:
+ return args.external_executor_id
Review comment:
Should we read env variable here?
--
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] Jorricks edited a comment on pull request #17207: WIP: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks edited a comment on pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#issuecomment-886190622
The other idea:
> After every manually triggered task, we read the event_buffer and store that in the task instances.
Would look something like adding this to the `def run()` of `class TaskInstanceModelView`:
```python3
event_buffer = executor.get_event_buffer()
for ti_key, (state, external_executor_id) in event_buffer.items():
if ti.key == ti_key and state == State.QUEUED:
session.merge(ti)
ti.external_executor_id = external_executor_id
session.commit()
updated = True
```
--
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] Jorricks commented on pull request #17207: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks commented on pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#issuecomment-914236678
> > I tried to add some tests on the CLI task part. That is pretty much done.
> > However, I had quite some trouble wrapping my head around a decent test approach on the celery_executor part.
> > There is currently not really a test for any of the functions I modified which makes me wonder if I should add them.
> > If so, do you have any remarks on how I could best do that?
>
> For the methods modified in CeleryExecutor, you can mock them and assert they were called with a celeryID.
> The app can be mocked to return an ID. See
>
> https://github.com/apache/airflow/blob/3d96ad62f91e662b62481441d8eec2994651e122/tests/executors/test_celery_executor.py#L75
>
>
> You can provide appID there.
I modified the existing test to add this behaviour.
--
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] Jorricks commented on pull request #17207: WIP: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks commented on pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#issuecomment-890525699
Rebased on latest main.
--
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] Jorricks commented on pull request #17207: WIP: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks commented on pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#issuecomment-886190622
The other idea:
> After every manually triggered task, we read the event_buffer and store that in the task instances.
Would look something like adding this to the `def run()` of `class TaskInstanceModelView`:
```python3
event_buffer = executor.get_event_buffer()
for ti_key, (state, external_executor_id) in event_buffer.items():
if state == State.QUEUED:
session.merge(ti)
ti.external_executor_id = external_executor_id
session.commit()
updated = True
```
--
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] Jorricks commented on pull request #17207: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks commented on pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#issuecomment-918514365
Tests should be fixed now 👍
--
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] ephraimbuddy commented on pull request #17207: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#issuecomment-919089282
You can actually retrigger it by closing and opening the 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] Jorricks commented on a change in pull request #17207: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
Jorricks commented on a change in pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#discussion_r707337767
##########
File path: tests/executors/test_celery_executor.py
##########
@@ -71,6 +73,7 @@ def _prepare_app(broker_url=None, execute=None):
test_config = dict(celery_executor.celery_configuration)
test_config.update({'broker_url': broker_url})
test_app = Celery(broker_url, config_source=test_config)
+ mock.patch(test_app)
Review comment:
My fault. Should have spotted that.
--
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] ephraimbuddy commented on a change in pull request #17207: WIP: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on a change in pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#discussion_r691914862
##########
File path: airflow/cli/commands/task_command.py
##########
@@ -174,6 +175,14 @@ def _run_raw_task(args, ti: TaskInstance) -> None:
)
+def _extract_external_executor_id(args) -> Optional[str]:
+ if "external_executor_id" in args:
+ return args.external_executor_id
+ elif "external_executor_id" in os.environ:
+ return os.environ.get("external_executor_id")
+ return None
Review comment:
```suggestion
return os.environ.get("external_executor_id", None)
```
--
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] ashb merged pull request #17207: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
ashb merged pull request #17207:
URL: https://github.com/apache/airflow/pull/17207
--
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] ephraimbuddy commented on a change in pull request #17207: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on a change in pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#discussion_r694764613
##########
File path: airflow/executors/celery_executor.py
##########
@@ -81,14 +81,16 @@ def execute_command(command_to_exec: CommandType) -> None:
"""Executes command."""
BaseExecutor.validate_command(command_to_exec)
log.info("Executing command in Celery: %s", command_to_exec)
+ celery_task_id = str(app.current_task.request.id) if app.current_task.request.id else None
+ log.info(f"Found the following celery_task_id: {celery_task_id}")
Review comment:
```suggestion
celery_task_id = app.current_task.request.id
log.info(f"Celery Task ID: {celery_task_id}")
```
--
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] ephraimbuddy commented on a change in pull request #17207: WIP: Fix external_executor_id not being set for manually run jobs.
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on a change in pull request #17207:
URL: https://github.com/apache/airflow/pull/17207#discussion_r693832397
##########
File path: airflow/executors/celery_executor.py
##########
@@ -125,8 +129,10 @@ def _execute_in_fork(command_to_exec: CommandType) -> None:
os._exit(ret)
-def _execute_in_subprocess(command_to_exec: CommandType) -> None:
+def _execute_in_subprocess(command_to_exec: CommandType, celery_task_id: Optional[str] = None) -> None:
env = os.environ.copy()
+ if celery_task_id:
+ env["celery_task_id"] = celery_task_id
Review comment:
Looks like we should do `env['external_executor_id'] = celery_task_id` here, right 🤔?
--
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