AN-362: log attribute ingestion

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "splight-lib"
-version = "5.24.13"
+version = "5.24.14"
 description = "Splight Python Library"
 authors = [
     {name = "Splight Dev",email = "dev@splight.com"}

splight_lib/models/_v3/datalake.py

@@ -9,6 +9,7 @@
 from splight_lib.client.datalake import DatalakeClientBuilder
 from splight_lib.client.datalake.common.abstract import AbstractDatalakeClient
 from splight_lib.client.datalake.v3.constants import StepName
+from splight_lib.logging._internal import LogTags, get_splight_logger
 from splight_lib.models._v3.asset import Asset
 from splight_lib.models._v3.attribute import Attribute
 from splight_lib.models._v3.exceptions import TraceAlreadyExistsError
@@ -21,6 +22,8 @@
 MAX_NUM_TRACES = 500
 T = TypeVar("T")
 
+logger = get_splight_logger()
+
 
 def hash(string: str) -> str:
     return sha256(string.encode("utf-8")).hexdigest()
@@ -166,10 +169,40 @@ class DataRecords(BaseModel):
     def apply(self) -> None:
         dl_client = get_datalake_client()
         dl_client.save(self.model_dump(mode="json"))
+        self._log_attribute_ingestion()
 
     async def async_apply(self) -> None:
         dl_client = get_datalake_client()
         await dl_client.async_save(self.model_dump(mode="json"))
+        self._log_attribute_ingestion()
+
+    def _log_attribute_ingestion(self) -> None:
+        try:
+            counters: dict[tuple[str, str], int] = {}
+            for record in self.records:
+                asset = record.get("asset")
+                attribute = record.get("attribute")
+                if asset is None or attribute is None:
+                    continue
+                if isinstance(asset, dict):
+                    asset = asset.get("id")
+                if isinstance(attribute, dict):
+                    attribute = attribute.get("id")
+                key = (asset, attribute)
+                counters[key] = counters.get(key, 0) + 1
+            for (asset, attribute), count in counters.items():
+                logger.info(
+                    "attribute_ingestion "
+                    f"asset={asset} "
+                    f"attribute={attribute} "
+                    f"count={count} "
+                    f"collection={self.collection}",
+                    tags=LogTags.DATALAKE,
+                )
+        except Exception:
+            logger.exception(
+                "Unable to log attribute ingestion", tags=LogTags.DATALAKE
+            )
 
 
 def chunk_list(