[airflow] 01/11: add detailed information to logging when a dag or a task finishes. (#19097)

[je...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

jedcunningham pushed a commit to branch v2-2-test
in repository https://gitbox.apache.org/repos/asf/airflow.git

commit 2f0bc8976d809bee996c3a6c123af33e8085748b
Author: Steve Zhang <86...@users.noreply.github.com>
AuthorDate: Fri Oct 22 08:35:26 2021 -0700

    add detailed information to logging when a dag or a task finishes. (#19097)
    
    * add detailed information to logging when a dag or a task finishes.
    
    * make logging of start_date/end_date ISO format to be consist
    
    * fix pre-commit
    
    * use only %s in new logging statements to gracefully handle when certain variables are None
    
    * fix precommit
    
    * use self._state instead of get_state().  add computation for dag run duration based on start_date and end_date
    
    * make linter happy with format
    
    * fix typo
    
    * put back missing reference to _state
    
    Co-authored-by: Daniel Imberman <da...@gmail.com>
    (cherry picked from commit 324c31c2d7ad756ce3814f74f0b6654d02f19426)
---
 airflow/jobs/scheduler_job.py | 25 +++++++++++++++++++++++++
 airflow/models/dagrun.py      | 25 +++++++++++++++++++++++++
 2 files changed, 50 insertions(+)

diff --git a/airflow/jobs/scheduler_job.py b/airflow/jobs/scheduler_job.py
index bf2410a..87dc158 100644
--- a/airflow/jobs/scheduler_job.py
+++ b/airflow/jobs/scheduler_job.py
@@ -533,6 +533,31 @@ class SchedulerJob(BaseJob):
                 self.log.info("Setting external_id for %s to %s", ti, info)
                 continue
 
+            msg = (
+                "TaskInstance Finished: dag_id=%s, task_id=%s, run_id=%s, "
+                "run_start_date=%s, run_end_date=%s, "
+                "run_duration=%s, state=%s, executor_state=%s, try_number=%s, max_tries=%s, job_id=%s, "
+                "pool=%s, queue=%s, priority_weight=%d, operator=%s"
+            )
+            self.log.info(
+                msg,
+                ti.dag_id,
+                ti.task_id,
+                ti.run_id,
+                ti.start_date,
+                ti.end_date,
+                ti.duration,
+                ti.state,
+                state,
+                try_number,
+                ti.max_tries,
+                ti.job_id,
+                ti.pool,
+                ti.queue,
+                ti.priority_weight,
+                ti.operator,
+            )
+
             if ti.try_number == buffer_key.try_number and ti.state == State.QUEUED:
                 Stats.incr('scheduler.tasks.killed_externally')
                 msg = (
diff --git a/airflow/models/dagrun.py b/airflow/models/dagrun.py
index 1d53265..e8eb685 100644
--- a/airflow/models/dagrun.py
+++ b/airflow/models/dagrun.py
@@ -542,6 +542,31 @@ class DagRun(Base, LoggingMixin):
         else:
             self.set_state(State.RUNNING)
 
+        if self._state == State.FAILED or self._state == State.SUCCESS:
+            msg = (
+                "DagRun Finished: dag_id=%s, execution_date=%s, run_id=%s, "
+                "run_start_date=%s, run_end_date=%s, run_duration=%s, "
+                "state=%s, external_trigger=%s, run_type=%s, "
+                "data_interval_start=%s, data_interval_end=%s, dag_hash=%s"
+            )
+            self.log.info(
+                msg,
+                self.dag_id,
+                self.execution_date,
+                self.run_id,
+                self.start_date,
+                self.end_date,
+                (self.end_date - self.start_date).total_seconds()
+                if self.start_date and self.end_date
+                else None,
+                self._state,
+                self.external_trigger,
+                self.run_type,
+                self.data_interval_start,
+                self.data_interval_end,
+                self.dag_hash,
+            )
+
         self._emit_true_scheduling_delay_stats_for_finished_state(finished_tasks)
         self._emit_duration_stats_for_finished_state()