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

Metric scheduler.tasks.running is never updated from 0 #20898

Closed
1 of 2 tasks
easontm opened this issue Jan 17, 2022 · 7 comments
Closed
1 of 2 tasks

Metric scheduler.tasks.running is never updated from 0 #20898

easontm opened this issue Jan 17, 2022 · 7 comments
Assignees
Labels

Comments

@easontm
Copy link
Contributor

easontm commented Jan 17, 2022

Apache Airflow version

2.2.3 (latest released)

What happened

The metric scheduler.tasks.running is populated from the variable num_tasks_in_executor in scheduler_job.py. However, num_tasks_in_executor is initialized to 0 and then never updated again.

Function _executable_task_instances_to_queued:

num_tasks_in_executor = 0

Stats.gauge('scheduler.tasks.running', num_tasks_in_executor)

I considered just opening a PR modifying that value with the rest of these "counter" modifications:

executable_tis.append(task_instance)
open_slots -= task_instance.pool_slots
dag_max_active_tasks_map[dag_id] += 1
task_concurrency_map[(task_instance.dag_id, task_instance.task_id)] += 1

However, I'm not sure if this is the best course of action because we don't know if the tasks is successfully added until the the list of executable TIs is returned within _critical_section_execute_task_instances, and those TIs are attempted to be queued in _enqueue_task_instances_with_queued_state.

if self.max_tis_per_query == 0:
max_tis = self.executor.slots_available
else:
max_tis = min(self.max_tis_per_query, self.executor.slots_available)
queued_tis = self._executable_task_instances_to_queued(max_tis, session=session)
self._enqueue_task_instances_with_queued_state(queued_tis, session=session)

def queue_command(
self,
task_instance: TaskInstance,
command: CommandType,
priority: int = 1,
queue: Optional[str] = None,
):
"""Queues command to task"""
if task_instance.key not in self.queued_tasks and task_instance.key not in self.running:
self.log.info("Adding to queue: %s", command)
self.queued_tasks[task_instance.key] = (command, priority, queue, task_instance)
else:
self.log.error("could not queue task %s", task_instance.key)

What you expected to happen

No response

How to reproduce

No response

Operating System

Ubuntu 19.10

Versions of Apache Airflow Providers

No response

Deployment

Other Docker-based deployment

Deployment details

No response

Anything else

No response

Are you willing to submit PR?

  • Yes I am willing to submit a PR!

Code of Conduct

@easontm easontm added area:core kind:bug This is a clearly a bug labels Jan 17, 2022
@potiuk
Copy link
Member

potiuk commented Feb 1, 2022

Yeah. Confirmed it does look suspiciously 0-always metrics :), Would you maube like to attempt to fix that one @easontm ?

@potiuk
Copy link
Member

potiuk commented Feb 1, 2022

Opening PR and discussing there, might be the best way to get to the right solution as it will drag attention of those who are closer to the code.

@easontm
Copy link
Contributor Author

easontm commented Feb 2, 2022

I won't be able to do it right away because of other obligations, but I'll see what I can do in a few weeks!

@ihorlukianov
Copy link
Contributor

Hi @potiuk, @easontm, do you mind if I take over?

@potiuk
Copy link
Member

potiuk commented Sep 19, 2022

Feel free.

@shubham22
Copy link

I believe we could close this out as we removed that metrics #30374. You should be using executor running and executor queued task metrics. cc: @vincbeck

@vincbeck
Copy link
Contributor

Correct!

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
Projects
None yet
Development

No branches or pull requests

5 participants