Skip to content
New issue

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

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

Already on GitHub? Sign in to your account

Fix DagProcessorJob integration for standalone dag-processor #30278

Merged
merged 1 commit into from
Mar 24, 2023
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion airflow/cli/cli_config.py
Original file line number Diff line number Diff line change
@@ -1014,7 +1014,7 @@ def string_lower_type(val):
# jobs check
ARG_JOB_TYPE_FILTER = Arg(
("--job-type",),
choices=("BackfillJob", "LocalTaskJob", "SchedulerJob", "TriggererJob"),
choices=("BackfillJob", "LocalTaskJob", "SchedulerJob", "TriggererJob", "DagProcessorJob"),
action="store",
help="The type of job(s) that will be checked.",
)
5 changes: 5 additions & 0 deletions airflow/dag_processing/manager.py
Original file line number Diff line number Diff line change
@@ -46,6 +46,7 @@
from airflow.callbacks.callback_requests import CallbackRequest, SlaCallbackRequest
from airflow.configuration import conf
from airflow.dag_processing.processor import DagFileProcessorProcess
from airflow.jobs.base_job import BaseJob
from airflow.models import errors
from airflow.models.dag import DagModel
from airflow.models.dagwarning import DagWarning
@@ -380,6 +381,7 @@ def __init__(
pickle_dags: bool,
signal_conn: MultiprocessingConnection | None = None,
async_mode: bool = True,
job: BaseJob | None = None,
):
super().__init__()
# known files; this will be updated every `dag_dir_list_interval` and stuff added/removed accordingly
@@ -393,6 +395,7 @@ def __init__(
self._async_mode = async_mode
self._parsing_start_time: int | None = None
self._dag_directory = dag_directory
self._job = job

# Set the signal conn in to non-blocking mode, so that attempting to
# send when the buffer is full errors, rather than hangs for-ever
@@ -573,6 +576,8 @@ def _run_parsing_loop(self):
while True:
loop_start_time = time.monotonic()
ready = multiprocessing.connection.wait(self.waitables.keys(), timeout=poll_time)
if self._job:
self._job.heartbeat()
if self._direct_scheduler_conn is not None and self._direct_scheduler_conn in ready:
agent_signal = self._direct_scheduler_conn.recv()

1 change: 1 addition & 0 deletions airflow/jobs/dag_processor_job.py
Original file line number Diff line number Diff line change
@@ -54,6 +54,7 @@ def __init__(
processor_timeout=processor_timeout,
dag_ids=dag_ids,
pickle_dags=pickle_dags,
job=self,
)
super().__init__(*args, **kwargs)

1 change: 1 addition & 0 deletions airflow/models/__init__.py
Original file line number Diff line number Diff line change
@@ -61,6 +61,7 @@ def import_all_models():

import airflow.jobs.backfill_job
import airflow.jobs.base_job
import airflow.jobs.dag_processor_job
import airflow.jobs.local_task_job
import airflow.jobs.scheduler_job
import airflow.jobs.triggerer_job