Export dag_id in process_metrics for Global Uniqueness and DAG Params to DB Metadata (GoogleCloudPlatform#922)

* Export dag_id in process_metrics to ensure global uniqueness.

Export dag_id in process_metrics in addition to run_id. The run_id is only unique within the current DAG run, so dag_id is required to maintain global uniqueness for metrics.

* Export params to database as metadata too

Author: ortibazar

xlml/utils/metric.py

@@ -624,6 +624,19 @@ def add_airflow_metadata(
             metadata_value=airflow_dag_run_link,
         )
     )
+    airflow_meta.append(
+        bigquery.MetadataHistoryRow(
+            job_uuid=uuid, metadata_key="dag_id", metadata_value=dag_id
+        )
+    )
+    for key, value in context.get("params", {}).items():
+      airflow_meta.append(
+          bigquery.MetadataHistoryRow(
+              job_uuid=uuid,
+              metadata_key=f"param:{key}",
+              metadata_value=str(value),
+          )
+      )
 
     metadata[index].extend(airflow_meta)
   return metadata