yujunjun/OpenMetadata
1
0
Fork 0
mirror of https://github.com/open-metadata/OpenMetadata.git synced 2025-08-07 08:38:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix #7904: Filter out old tasks which doesn't have dag runs while publishing status (#7905)

Browse Source 
* Fix #7904: Filter out old tasks which doesn't have dag runs while publishing status

* Fix #7904: Filter out old tasks which doesn't have dag runs while publishing status
This commit is contained in:
Sriharsha Chintalapani 2022-10-03 21:34:55 -07:00 committed by GitHub
parent 566602373f
commit 8efc0370e5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 31 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

60
ingestion/src/metadata/ingestion/source/pipeline/airflow.py
View File

@ -207,37 +207,40 @@ class AirflowSource(PipelineServiceSource):
dag_run_list = self.get_pipeline_status(pipeline_details.dag_id) dag_run_list = self.get_pipeline_status(pipeline_details.dag_id)
for dag_run in dag_run_list: for dag_run in dag_run_list:
tasks = self.get_task_instances( if (
dag_id=dag_run.dag_id, dag_run.run_id
run_id=dag_run.run_id, ): # Airflow dags can have old task which are turned off/commented out in code
execution_date=dag_run.execution_date, # Used for Airflow 2.1.4 query fallback tasks = self.get_task_instances(
) dag_id=dag_run.dag_id,
run_id=dag_run.run_id,
execution_date=dag_run.execution_date, # Used for Airflow 2.1.4 query fallback
)
task_statuses = [ task_statuses = [
TaskStatus( TaskStatus(
name=task.task_id, name=task.task_id,
executionStatus=STATUS_MAP.get(
task.state, StatusType.Pending.value
),
startTime=datetime_to_ts(task.start_date),
endTime=datetime_to_ts(
task.end_date
), # Might be None for running tasks
) # Log link might not be present in all Airflow versions
for task in tasks
]
pipeline_status = PipelineStatus(
taskStatus=task_statuses,
executionStatus=STATUS_MAP.get( executionStatus=STATUS_MAP.get(
task.state, StatusType.Pending.value dag_run.state, StatusType.Pending.value
), ),
startTime=datetime_to_ts(task.start_date), timestamp=dag_run.execution_date.timestamp(),
endTime=datetime_to_ts( )
task.end_date yield OMetaPipelineStatus(
), # Might be None for running tasks pipeline_fqn=self.context.pipeline.fullyQualifiedName.__root__,
) # Log link might not be present in all Airflow versions pipeline_status=pipeline_status,
for task in tasks )
]
pipeline_status = PipelineStatus(
taskStatus=task_statuses,
executionStatus=STATUS_MAP.get(
dag_run.state, StatusType.Pending.value
),
timestamp=dag_run.execution_date.timestamp(),
)
yield OMetaPipelineStatus(
pipeline_fqn=self.context.pipeline.fullyQualifiedName.__root__,
pipeline_status=pipeline_status,
)
except Exception as exc: except Exception as exc:
logger.debug(traceback.format_exc()) logger.debug(traceback.format_exc())
logger.warning( logger.warning(
@ -264,7 +267,6 @@ class AirflowSource(PipelineServiceSource):
json_data_column, json_data_column,
SerializedDagModel.fileloc, SerializedDagModel.fileloc,
).all(): ).all():
yield OMSerializedDagDetails( yield OMSerializedDagDetails(
dag_id=serialized_dag[0], dag_id=serialized_dag[0],
data=serialized_dag[1], data=serialized_dag[1],