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.

Sign up for GitHub

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

Jump to bottom

Set parallelism log messages to warning level for better visiblity #39298

Open
wants to merge 2 commits into
base: main
Choose a base branch
from
+4 −1
Open

Set parallelism log messages to warning level for better visiblity #39298

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
184f9b9
Set log messages related to parallelism to warning level for better v…
BasPH Apr 28, 2024
14d8f94
Set a few logging statements back to info
BasPH May 2, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
5 changes: 4 additions & 1 deletion airflow/executors/base_executor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -221,7 +221,10 @@ def heartbeat(self) -> None:

self.log.debug("%s running task instances", num_running_tasks)
self.log.debug("%s in queue", num_queued_tasks)
self.log.debug("%s open slots", open_slots)
if open_slots == 0:
self.log.warning("Executor parallelism limit reached. 0 open slots.")
BasPH marked this conversation as resolved.
Show resolved Hide resolved
else:
self.log.debug("%s open slots", open_slots)

Stats.gauge(
"executor.open_slots", value=open_slots, tags={"status": "open", "name": self.__class__.__name__}
Expand Down
10 changes: 5 additions & 5 deletions airflow/jobs/scheduler_job_runner.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -434,7 +434,7 @@ def _executable_task_instances_to_queued(self, max_tis: int, session: Session) -
open_slots = pool_stats["open"]

if open_slots <= 0:
self.log.info(
self.log.warning(
"Not scheduling since there are %s open slots in pool %s", open_slots, pool_name
)
# Can't schedule any more since there are no more open slots.
Expand Down Expand Up @@ -486,7 +486,7 @@ def _executable_task_instances_to_queued(self, max_tis: int, session: Session) -
max_active_tasks_per_dag_limit,
)
if current_active_tasks_per_dag >= max_active_tasks_per_dag_limit:
self.log.info(
self.log.warning(
"Not executing %s since the number of tasks running or queued "
"from DAG %s is >= to the DAG's max_active_tasks limit of %s",
task_instance,
Expand Down Expand Up @@ -527,7 +527,7 @@ def _executable_task_instances_to_queued(self, max_tis: int, session: Session) -
]

if current_task_concurrency >= task_concurrency_limit:
self.log.info(
self.log.warning(
"Not executing %s since the task concurrency for"
" this task has been reached.",
task_instance,
Expand All @@ -547,7 +547,7 @@ def _executable_task_instances_to_queued(self, max_tis: int, session: Session) -
]

if current_task_dagrun_concurrency >= task_dagrun_concurrency_limit:
self.log.info(
self.log.warning(
"Not executing %s since the task concurrency per DAG run for"
" this task has been reached.",
task_instance,
Expand Down Expand Up @@ -1330,7 +1330,7 @@ def _should_update_dag_next_dagruns(
total_active_runs = dag.get_num_active_runs(only_running=False, session=session)

if total_active_runs and total_active_runs >= dag.max_active_runs:
self.log.info(
self.log.warning(
"DAG %s is at (or above) max_active_runs (%d of %d), not creating any more runs",
dag_model.dag_id,
total_active_runs,
Expand Down