Add/port metrics to the new Execution interface

Find a way to emit metrics from the new supervisor process.

Some examples of what needs porting over:

airflow/airflow/jobs/local_task_job_runner.py

Line 224 in 84907f1

Stats.incr("local_task_job_prolonged_heartbeat_failure", 1, 1)

airflow/airflow/jobs/local_task_job_runner.py

Lines 347 to 361 in 84907f1

    
           def _log_return_code_metric(self, return_code: int): 
        
               Stats.incr( 
        
                   "local_task_job.task_exit." 
        
                   f"{self.job.id}.{self.task_instance.dag_id}.{self.task_instance.task_id}.{return_code}" 
        
               ) 
        
               # Same metric with tagging 
        
               Stats.incr( 
        
                   "local_task_job.task_exit", 
        
                   tags={ 
        
                       "job_id": self.job.id, 
        
                       "dag_id": self.task_instance.dag_id, 
        
                       "task_id": self.task_instance.task_id, 
        
                       "return_code": return_code, 
        
                   }, 
        
               )

@kaxil

@kaxil is this still open? If so, can I work on this?

Yes still open, and assigned to you

Thank you

not sure if ti failure metrics are out of scope for this issue? I didn't see it being handled in the associated PR. Otherwise I put up #59072

@kaxil

@kaxil , is the idea to centralise all previously task process emitted metrics (in airflow 2) to supervisor? #59072 follows airflow 2 very closely in that it's emitted from task runner rather than supervisor.

Will TaskInstanceStarted and TaskInstanceFinished be covered in this issue?

@michaelfu1029

All the related issues are closed, I think we emit all the metrics we did with Airflow 2 now.

#53722 was the PR that ported over the infra and we have had incremental PRs adding metrics like: #55322, #59731. The intent of this task is complete. However, if there are more issues to be created, follow it on with PRs as you like.

@michaelfu1029 for example ^

Navigation Menu

Repository navigation

Add/port metrics to the new Execution interface #45511

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	def _log_return_code_metric(self, return_code: int):
	Stats.incr(
	"local_task_job.task_exit."
	f"{self.job.id}.{self.task_instance.dag_id}.{self.task_instance.task_id}.{return_code}"
	)
	# Same metric with tagging
	Stats.incr(
	"local_task_job.task_exit",
	tags={
	"job_id": self.job.id,
	"dag_id": self.task_instance.dag_id,
	"task_id": self.task_instance.task_id,
	"return_code": return_code,
	},
	)

Add/port metrics to the new Execution interface #45511

Description

Activity

kaxil commented on Aug 13, 2025

KoviAnusha commented on Oct 22, 2025

kaxil commented on Oct 22, 2025

KoviAnusha commented on Oct 24, 2025

antonlin1 commented on Dec 4, 2025

antonlin1 commented on Dec 5, 2025

michaelfu1029 commented on Jan 12, 2026

amoghrajesh commented on Feb 12, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions