Error lineage - airflow_provider_openmetadata

[fredericocotrim]

Affected module

Ingestion Framework

Describe the bug

Error: 'int' object has no attribute 'timestamp'

2023-12-07, 19:18:24 UTC] {callback.py:105} ERROR - Traceback (most recent call last):
  File "/home/***/.local/lib/python3.8/site-packages/***_provider_openmetadata/lineage/callback.py", line 97, in success_callback
    add_status(
  File "/home/***/.local/lib/python3.8/site-packages/***_provider_openmetadata/lineage/status.py", line 116, in add_status
    timestamp=datetime_to_ts(execution_date),
  File "/home/***/.local/lib/python3.8/site-packages/metadata/utils/helpers.py", line 183, in datetime_to_ts
    return int(date.timestamp() * 1_000) if date else None
AttributeError: 'int' object has no attribute 'timestamp'

[2023-12-07, 19:18:24 UTC] {callback.py:106} ERROR - Lineage Callback exception 'int' object has no attribute 'timestamp'
[2023-12-07, 19:18:24 UTC] {local_task_job.py:212} INFO - Task exited with return code 0
[2023-12-07, 19:18:24 UTC] {taskinstance.py:2596} INFO - 0 downstream tasks scheduled from follow-on schedule check

Version:

• OS: linux
• Python version: 3.8
• OpenMetadata version: 1.2.2
• OpenMetadata Ingestion package version: openmetadata-ingestion==1.2.2

Additional context

In the add_status method of the file "/ingestion/src/airflow_provider_openmetadata/lineage/status.py" the execution_date is converted twice using the function datetime_to_ts:

    # Let this fail if we cannot properly extract & cast the start_date
    execution_date = datetime_to_ts(dag_run.execution_date)
    operator.log.info(f"Logging pipeline status for execution {execution_date}")

    updated_status = PipelineStatus(
        timestamp=datetime_to_ts(execution_date),
        executionStatus=get_dag_status(
            all_tasks=dag.task_ids,
            task_status=updated_task_status,
        ),
        taskStatus=updated_task_status,
    )

I resolved the error by replacing the line:

execution_date = datetime_to_ts(dag_run.execution_date)

with

execution_date = dag_run.execution_date

[pmbrull]

thank you for the detailed analysis. Appreciate it. Will take care of this