You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "AmFlint (via GitHub)" <gi...@apache.org> on 2023/05/10 11:08:17 UTC
[GitHub] [airflow] AmFlint opened a new pull request, #31182: fix dag processor job_runner, add missing job_type
AmFlint opened a new pull request, #31182:
URL: https://github.com/apache/airflow/pull/31182
## Description
We're running custom probes on the dag processor component.
Since the DagProcessor component is represented as a `job_runner`, we're running SQL queries to check the latest_heartbeat for the associated job object.
To speed up queries in big deployments, we're using a the index on `job_type`:
```sql
SELECT latest_heartbeat FROM job
WHERE
job_type = 'DagProcessorJob'
AND
hostname = '%s';
```
(hostname being the IP address of the k8s pod).
The problem is that there's a bug currently, and the job_type is always NULL for dag processor job_runners.
I added the missing line of code that should fix 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] ashb commented on a diff in pull request #31182: fix dag processor job_runner, add missing job_type
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #31182:
URL: https://github.com/apache/airflow/pull/31182#discussion_r1189755421
##########
airflow/jobs/dag_processor_job_runner.py:
##########
@@ -53,6 +53,7 @@ def __init__(
f"The job is already assigned a different job_type: {job.job_type}."
f"This is a bug and should be reported."
)
+ self.job.job_type = self.job_type
Review Comment:
Should this be done in the case class so no other job subclasses forget 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
[GitHub] [airflow] AmFlint commented on a diff in pull request #31182: fix dag processor job_runner, add missing job_type
Posted by "AmFlint (via GitHub)" <gi...@apache.org>.
AmFlint commented on code in PR #31182:
URL: https://github.com/apache/airflow/pull/31182#discussion_r1189879660
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -252,14 +252,12 @@ def __init__(
job: Job | JobPydantic,
capacity=None,
):
- super().__init__()
+ super().__init__(job)
if job.job_type and job.job_type != self.job_type:
Review Comment:
Sorry forgot about that I wasn't fully focused, was fighting to get Airflow running locally on ARM macbook for the first time it's not really easy haha
--
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] potiuk commented on pull request #31182: Make sure that DAG processor job row has filed value in `job_type` column
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #31182:
URL: https://github.com/apache/airflow/pull/31182#issuecomment-1543641376
FYI: This is what I was talking about -> result if we add the typing now.
It's ol how it is now, I won't complai, and I will solve it later.
But just FYI why it has been separated - previously Scheduler and BackfillJobRunner had `Job` only while the others could have either `Job` or `JobPydantic`. By bring it as a common field in Base Runner, we have no easy way to distinguish tose two cases.
```
airflow/jobs/backfill_job_runner.py:258: error: Item "JobPydantic" of
"Union[Job, JobPydantic]" has no attribute "executor" [union-attr]
executor = self.job.executor
^
airflow/jobs/backfill_job_runner.py:540: error: Argument 1 to
"import_executor_cls" of "ExecutorLoader" has incompatible type
"Union[Any, str, None]"; expected "str" [arg-type]
self.job.executor_class,
^
airflow/jobs/backfill_job_runner.py:898: error: Item "JobPydantic" of
"Union[Job, JobPydantic]" has no attribute "executor" [union-attr]
executor = self.job.executor
^
airflow/jobs/backfill_job_runner.py:964: error: Item "JobPydantic" of
"Union[Job, JobPydantic]" has no attribute "executor" [union-attr]
queued_tis = self.job.executor.queued_tasks
^
airflow/jobs/backfill_job_runner.py:966: error: Item "JobPydantic" of
"Union[Job, JobPydantic]" has no attribute "executor" [union-attr]
running_tis = self.job.executor.running
^
airflow/jobs/backfill_job_runner.py:1005: error: Argument 4 to
"reduce_in_chunks" has incompatible type "Optional[int]"; expected "int"
[arg-type]
...elpers.reduce_in_chunks(query, tis_to_reset, [], self.job.max_tis_per_...
^
airflow/jobs/scheduler_job_runner.py:264: error: Item "JobPydantic" of
"Union[Job, JobPydantic]" has no attribute "executor" [union-attr]
self.job.executor.debug_dump()
^
airflow/jobs/scheduler_job_runner.py:280: error: Item "JobPydantic" of
"Union[Job, JobPydantic]" has no attribute "is_alive" [union-attr]
return self.job.is_alive(grace_multiplier=grace_multiplier...
^
airflow/jobs/scheduler_job_runner.py:650: error: Item "JobPydantic" of
"Union[Job, JobPydantic]" has no attribute "executor" [union-attr]
self.job.executor.queue_command(
^
airflow/jobs/scheduler_job_runner.py:677: error: Item "JobPydantic" of
"Union[Job, JobPydantic]" has no attribute "executor" [union-attr]
max_tis = self.job.executor.slots_available
^
airflow/jobs/scheduler_job_runner.py:679: error: Item "JobPydantic" of
"Union[Job, JobPydantic]" has no attribute "executor" [union-attr]
... max_tis = min(self.job.max_tis_per_query, self.job.executor.slo...
^
airflow/jobs/scheduler_job_runner.py:690: error: Item "JobPydantic" of
"Union[Job, JobPydantic]" has no attribute "executor" [union-attr]
event_buffer = self.job.executor.get_event_buffer()
^
airflow/jobs/scheduler_job_runner.py:772: error: Item "JobPydantic" of
"Union[Job, JobPydantic]" has no attribute "executor" [union-attr]
or self.job.executor.has_task(ti) # This scheduler ha...
^
airflow/jobs/scheduler_job_runner.py:802: error: Item "JobPydantic" of
"Union[Job, JobPydantic]" has no attribute "executor" [union-attr]
self.job.executor.send_callback(request)
^
airflow/jobs/scheduler_job_runner.py:837: error: Item "JobPydantic" of
"Union[Job, JobPydantic]" has no attribute "executor" [union-attr]
self.job.executor.job_id = self.job.id
^
airflow/jobs/scheduler_job_runner.py:840: error: Item "JobPydantic" of
"Union[Job, JobPydantic]" has no attribute "executor" [union-attr]
self.job.executor.callback_sink = PipeCallbackSink(
^
airflow/jobs/scheduler_job_runner.py:847: error: Item "JobPydantic" of
"Union[Job, JobPydantic]" has no attribute "executor" [union-attr]
self.job.executor.callback_sink = DatabaseCallbackSink...
^
airflow/jobs/scheduler_job_runner.py:849: error: Item "JobPydantic" of
"Union[Job, JobPydantic]" has no attribute "executor" [union-attr]
self.job.executor.start()
^
airflow/jobs/scheduler_job_runner.py:879: error: Item "JobPydantic" of
"Union[Job, JobPydantic]" has no attribute "executor" [union-attr]
self.job.executor.end()
^
airflow/jobs/scheduler_job_runner.py:994: error: Item "JobPydantic" of
"Union[Job, JobPydantic]" has no attribute "executor" [union-attr]
self.job.executor.heartbeat()
^
airflow/jobs/scheduler_job_runner.py:1091: error: Item "JobPydantic" of
"Union[Job, JobPydantic]" has no attribute "executor" [union-attr]
if self.job.executor.slots_available <= 0:
^
airflow/jobs/scheduler_job_runner.py:1477: error: Item "JobPydantic" of
"Union[Job, JobPydantic]" has no attribute "executor" [union-attr]
self.job.executor.send_callback(callback)
^
airflow/jobs/scheduler_job_runner.py:1504: error: Item "JobPydantic" of
"Union[Job, JobPydantic]" has no attribute "executor" [union-attr]
self.job.executor.send_callback(request)
^
airflow/jobs/scheduler_job_runner.py:1529: error: Item "JobPydantic" of
"Union[Job, JobPydantic]" has no attribute "executor" [union-attr]
tis_for_warning_message = self.job.executor.cleanup_stuck_...
^
airflow/jobs/scheduler_job_runner.py:1612: error: Item "JobPydantic" of
"Union[Job, JobPydantic]" has no attribute "executor" [union-attr]
to_reset = self.job.executor.try_adopt_task_instan...
^
airflow/jobs/scheduler_job_runner.py:1710: error: Item "JobPydantic" of
"Union[Job, JobPydantic]" has no attribute "executor" [union-attr]
self.job.executor.send_callback(request)
^
```
--
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] potiuk commented on pull request #31182: Make sure that DAG processor job row has filed value in `job_type` column
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #31182:
URL: https://github.com/apache/airflow/pull/31182#issuecomment-1545265975
Fix in https://github.com/apache/airflow/pull/31240 (cc: @AmFlint )
--
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 a diff in pull request #31182: fix dag processor job_runner, add missing job_type
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #31182:
URL: https://github.com/apache/airflow/pull/31182#discussion_r1189873576
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -252,14 +252,12 @@ def __init__(
job: Job | JobPydantic,
capacity=None,
):
- super().__init__()
+ super().__init__(job)
if job.job_type and job.job_type != self.job_type:
Review Comment:
Move this check in to the base class 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] boring-cyborg[bot] commented on pull request #31182: Make sure that DAG processor job row has filed value in `job_type` column
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #31182:
URL: https://github.com/apache/airflow/pull/31182#issuecomment-1542425802
Awesome work, congrats on your first merged pull request! You are invited to check our [Issue Tracker](https://github.com/apache/airflow/issues) for additional contributions.
--
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 diff in pull request #31182: fix dag processor job_runner, add missing job_type
Posted by "ephraimbuddy (via GitHub)" <gi...@apache.org>.
ephraimbuddy commented on code in PR #31182:
URL: https://github.com/apache/airflow/pull/31182#discussion_r1189756758
##########
airflow/jobs/dag_processor_job_runner.py:
##########
@@ -53,6 +53,7 @@ def __init__(
f"The job is already assigned a different job_type: {job.job_type}."
f"This is a bug and should be reported."
)
+ self.job.job_type = self.job_type
Review Comment:
Yeah. 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.
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 merged pull request #31182: Make sure that DAG processor job row has filed value in `job_type` column
Posted by "ephraimbuddy (via GitHub)" <gi...@apache.org>.
ephraimbuddy merged PR #31182:
URL: https://github.com/apache/airflow/pull/31182
--
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 a diff in pull request #31182: fix dag processor job_runner, add missing job_type
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #31182:
URL: https://github.com/apache/airflow/pull/31182#discussion_r1189755421
##########
airflow/jobs/dag_processor_job_runner.py:
##########
@@ -53,6 +53,7 @@ def __init__(
f"The job is already assigned a different job_type: {job.job_type}."
f"This is a bug and should be reported."
)
+ self.job.job_type = self.job_type
Review Comment:
Should this be done in the base class so no other job subclasses forget 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
[GitHub] [airflow] boring-cyborg[bot] commented on pull request #31182: fix dag processor job_runner, add missing job_type
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #31182:
URL: https://github.com/apache/airflow/pull/31182#issuecomment-1541964261
Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst)
Here are some useful points:
- Pay attention to the quality of your code (ruff, mypy and type annotations). Our [pre-commits]( https://github.com/apache/airflow/blob/main/STATIC_CODE_CHECKS.rst#prerequisites-for-pre-commit-hooks) will help you with that.
- In case of a new feature add useful documentation (in docstrings or in `docs/` directory). Adding a new operator? Check this short [guide](https://github.com/apache/airflow/blob/main/docs/apache-airflow/howto/custom-operator.rst) Consider adding an example DAG that shows how users should use it.
- Consider using [Breeze environment](https://github.com/apache/airflow/blob/main/BREEZE.rst) for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
- Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
- Please follow [ASF Code of Conduct](https://www.apache.org/foundation/policies/conduct) for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
- Be sure to read the [Airflow Coding style]( https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#coding-style-and-best-practices).
Apache Airflow is a community-driven project and together we are making it better 🚀.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack
--
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 #31182: Make sure that DAG processor job row has filed value in `job_type` column
Posted by "ephraimbuddy (via GitHub)" <gi...@apache.org>.
ephraimbuddy commented on PR #31182:
URL: https://github.com/apache/airflow/pull/31182#issuecomment-1542267869
@AmFlint , a test needs fixing: https://github.com/apache/airflow/actions/runs/4937323172/jobs/8826655492?pr=31182#step:6:287
--
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] potiuk commented on a diff in pull request #31182: Make sure that DAG processor job row has filed value in `job_type` column
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #31182:
URL: https://github.com/apache/airflow/pull/31182#discussion_r1189981203
##########
airflow/jobs/dag_processor_job_runner.py:
##########
@@ -53,6 +53,7 @@ def __init__(
f"The job is already assigned a different job_type: {job.job_type}."
f"This is a bug and should be reported."
)
+ self.job.job_type = self.job_type
Review Comment:
There will be additional problems if you try to extract it to common code especially if you want to add typing to job because of circular dependencies it will create - I think the original solution wil be better (at least for now). We can refactor it in the future when we complete AIP-44. It's solvable but I would focus on solvin this the simplest possible way now ie ^^
--
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] potiuk commented on a diff in pull request #31182: Make sure that DAG processor job row has filed value in `job_type` column
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #31182:
URL: https://github.com/apache/airflow/pull/31182#discussion_r1189981203
##########
airflow/jobs/dag_processor_job_runner.py:
##########
@@ -53,6 +53,7 @@ def __init__(
f"The job is already assigned a different job_type: {job.job_type}."
f"This is a bug and should be reported."
)
+ self.job.job_type = self.job_type
Review Comment:
There will be additional problems if you try to extract it to common code especially if you want to add typing to job because - I think the original solution wil be better (at least for now). We can refactor it in the future when we complete AIP-44. It's solvable but I would focus on solvin this the simplest possible way now ie ^^
--
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