feat/1681 collects load job metrics and adds remote uri (#1708)

* collects basic load job metrics in LoadJob

* adds remote uri to filesystem copy jobs metrics

* adds job id to load package info

* adds table name to job metrics

* skips run step when serializing trace

* adds trace shape test with trace schema

* tests job file name too long

* docs running pipelines with the same name for different envs

* extracts step metrics in common, renames followup jobs

* fixes tests

* fixes tests

* tests delta filesystem for remote_uri

* adds exec_info to trace contract test

* tests remote_uri for filesystem copy

* fixes platform test

Author: rudolfix

dlt/common/data_writers/__init__.py

@@ -1,6 +1,5 @@
 from dlt.common.data_writers.writers import (
     DataWriter,
-    DataWriterMetrics,
     TDataItemFormat,
     FileWriterSpec,
     create_import_spec,
@@ -22,7 +21,6 @@
     "resolve_best_writer_spec",
     "get_best_writer_spec",
     "is_native_writer",
-    "DataWriterMetrics",
     "TDataItemFormat",
     "BufferedDataWriter",
     "new_file_id",

dlt/common/data_writers/buffered.py

@@ -3,14 +3,15 @@
 import contextlib
 from typing import ClassVar, Iterator, List, IO, Any, Optional, Type, Generic
 
+from dlt.common.metrics import DataWriterMetrics
 from dlt.common.typing import TDataItem, TDataItems
 from dlt.common.data_writers.exceptions import (
     BufferedDataWriterClosed,
     DestinationCapabilitiesRequired,
     FileImportNotFound,
     InvalidFileNameTemplateException,
 )
-from dlt.common.data_writers.writers import TWriter, DataWriter, DataWriterMetrics, FileWriterSpec
+from dlt.common.data_writers.writers import TWriter, DataWriter, FileWriterSpec
 from dlt.common.schema.typing import TTableSchemaColumns
 from dlt.common.configuration import with_config, known_sections, configspec
 from dlt.common.configuration.specs import BaseConfiguration

dlt/common/data_writers/writers.py

@@ -34,6 +34,7 @@
     TLoaderFileFormat,
     ALL_SUPPORTED_FILE_FORMATS,
 )
+from dlt.common.metrics import DataWriterMetrics
 from dlt.common.schema.typing import TTableSchemaColumns
 from dlt.common.typing import StrAny
 
@@ -59,25 +60,6 @@ class FileWriterSpec(NamedTuple):
     supports_compression: bool = False
 
 
-class DataWriterMetrics(NamedTuple):
-    file_path: str
-    items_count: int
-    file_size: int
-    created: float
-    last_modified: float
-
-    def __add__(self, other: Tuple[object, ...], /) -> Tuple[object, ...]:
-        if isinstance(other, DataWriterMetrics):
-            return DataWriterMetrics(
-                "",  # path is not known
-                self.items_count + other.items_count,
-                self.file_size + other.file_size,
-                min(self.created, other.created),
-                max(self.last_modified, other.last_modified),
-            )
-        return NotImplemented
-
-
 EMPTY_DATA_WRITER_METRICS = DataWriterMetrics("", 0, 0, 2**32, 0.0)
 
 

dlt/common/destination/reference.py

@@ -24,10 +24,11 @@
 from copy import deepcopy
 import inspect
 
-from dlt.common import logger
+from dlt.common import logger, pendulum
 from dlt.common.configuration.specs.base_configuration import extract_inner_hint
 from dlt.common.destination.utils import verify_schema_capabilities
 from dlt.common.exceptions import TerminalValueError
+from dlt.common.metrics import LoadJobMetrics
 from dlt.common.normalizers.naming import NamingConvention
 from dlt.common.schema import Schema, TTableSchema, TSchemaTables
 from dlt.common.schema.utils import (
@@ -284,6 +285,8 @@ def __init__(self, file_path: str) -> None:
         # NOTE: we only accept a full filepath in the constructor
         assert self._file_name != self._file_path
         self._parsed_file_name = ParsedLoadJobFileName.parse(self._file_name)
+        self._started_at: pendulum.DateTime = None
+        self._finished_at: pendulum.DateTime = None
 
     def job_id(self) -> str:
         """The job id that is derived from the file name and does not changes during job lifecycle"""
@@ -306,6 +309,18 @@ def exception(self) -> str:
         """The exception associated with failed or retry states"""
         pass
 
+    def metrics(self) -> Optional[LoadJobMetrics]:
+        """Returns job execution metrics"""
+        return LoadJobMetrics(
+            self._parsed_file_name.job_id(),
+            self._file_path,
+            self._parsed_file_name.table_name,
+            self._started_at,
+            self._finished_at,
+            self.state(),
+            None,
+        )
+
 
 class RunnableLoadJob(LoadJob, ABC):
     """Represents a runnable job that loads a single file
@@ -361,6 +376,7 @@ def run_managed(
         # filepath is now moved to running
         try:
             self._state = "running"
+            self._started_at = pendulum.now()
             self._job_client.prepare_load_job_execution(self)
             self.run()
             self._state = "completed"
@@ -371,6 +387,7 @@ def run_managed(
             self._state = "retry"
             self._exception = e
         finally:
+            self._finished_at = pendulum.now()
             # sanity check
             assert self._state in ("completed", "retry", "failed")
 
@@ -391,7 +408,7 @@ def exception(self) -> str:
         return str(self._exception)
 
 
-class FollowupJob:
+class FollowupJobRequest:
     """Base class for follow up jobs that should be created"""
 
     @abstractmethod
@@ -403,8 +420,8 @@ def new_file_path(self) -> str:
 class HasFollowupJobs:
     """Adds a trait that allows to create single or table chain followup jobs"""
 
-    def create_followup_jobs(self, final_state: TLoadJobState) -> List[FollowupJob]:
-        """Return list of new jobs. `final_state` is state to which this job transits"""
+    def create_followup_jobs(self, final_state: TLoadJobState) -> List[FollowupJobRequest]:
+        """Return list of jobs requests for jobs that should be created. `final_state` is state to which this job transits"""
         return []
 
 
@@ -479,7 +496,7 @@ def create_table_chain_completed_followup_jobs(
         self,
         table_chain: Sequence[TTableSchema],
         completed_table_chain_jobs: Optional[Sequence[LoadJobInfo]] = None,
-    ) -> List[FollowupJob]:
+    ) -> List[FollowupJobRequest]:
         """Creates a list of followup jobs that should be executed after a table chain is completed"""
         return []
 

dlt/common/metrics.py

@@ -0,0 +1,71 @@
+import datetime  # noqa: I251
+from typing import Any, Dict, List, NamedTuple, Optional, Tuple, TypedDict  # noqa: 251
+
+
+class DataWriterMetrics(NamedTuple):
+    file_path: str
+    items_count: int
+    file_size: int
+    created: float
+    last_modified: float
+
+    def __add__(self, other: Tuple[object, ...], /) -> Tuple[object, ...]:
+        if isinstance(other, DataWriterMetrics):
+            return DataWriterMetrics(
+                self.file_path if self.file_path == other.file_path else "",
+                # self.table_name if self.table_name == other.table_name else "",
+                self.items_count + other.items_count,
+                self.file_size + other.file_size,
+                min(self.created, other.created),
+                max(self.last_modified, other.last_modified),
+            )
+        return NotImplemented
+
+
+class StepMetrics(TypedDict):
+    """Metrics for particular package processed in particular pipeline step"""
+
+    started_at: datetime.datetime
+    """Start of package processing"""
+    finished_at: datetime.datetime
+    """End of package processing"""
+
+
+class ExtractDataInfo(TypedDict):
+    name: str
+    data_type: str
+
+
+class ExtractMetrics(StepMetrics):
+    schema_name: str
+    job_metrics: Dict[str, DataWriterMetrics]
+    """Metrics collected per job id during writing of job file"""
+    table_metrics: Dict[str, DataWriterMetrics]
+    """Job metrics aggregated by table"""
+    resource_metrics: Dict[str, DataWriterMetrics]
+    """Job metrics aggregated by resource"""
+    dag: List[Tuple[str, str]]
+    """A resource dag where elements of the list are graph edges"""
+    hints: Dict[str, Dict[str, Any]]
+    """Hints passed to the resources"""
+
+
+class NormalizeMetrics(StepMetrics):
+    job_metrics: Dict[str, DataWriterMetrics]
+    """Metrics collected per job id during writing of job file"""
+    table_metrics: Dict[str, DataWriterMetrics]
+    """Job metrics aggregated by table"""
+
+
+class LoadJobMetrics(NamedTuple):
+    job_id: str
+    file_path: str
+    table_name: str
+    started_at: datetime.datetime
+    finished_at: datetime.datetime
+    state: Optional[str]
+    remote_uri: Optional[str]
+
+
+class LoadMetrics(StepMetrics):
+    job_metrics: Dict[str, LoadJobMetrics]