Emit ArtifactWritten event when artifacts are written

.changes/unreleased/Features-20241029-181728.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Emit debug logging event whenever artifacts are written
+time: 2024-10-29T18:17:28.321188-05:00
+custom:
+  Author: QMalcolm
+  Issue: 10937

core/dbt/contracts/graph/manifest.py

@@ -67,7 +67,7 @@
 )
 from dbt.contracts.graph.unparsed import SourcePatch, UnparsedVersion
 from dbt.contracts.util import SourceKey
-from dbt.events.types import UnpinnedRefNewVersionAvailable
+from dbt.events.types import ArtifactWritten, UnpinnedRefNewVersionAvailable
 from dbt.exceptions import (
     AmbiguousResourceNameRefError,
     CompilationError,
@@ -1219,7 +1219,9 @@ def writable_manifest(self) -> "WritableManifest":
         )
 
     def write(self, path):
-        self.writable_manifest().write(path)
+        writable = self.writable_manifest()
+        writable.write(path)
+        fire_event(ArtifactWritten(artifact_type=writable.__class__.__name__, artifact_path=path))
 
     # Called in dbt.compilation.Linker.write_graph and
     # dbt.graph.queue.get and ._include_in_cost

core/dbt/contracts/graph/semantic_manifest.py

@@ -7,7 +7,7 @@
 )
 from dbt.contracts.graph.manifest import Manifest
 from dbt.contracts.graph.nodes import ModelNode
-from dbt.events.types import SemanticValidationFailure
+from dbt.events.types import ArtifactWritten, SemanticValidationFailure
 from dbt.exceptions import ParsingError
 from dbt_common.clients.system import write_file
 from dbt_common.events.base_types import EventLevel
@@ -71,6 +71,7 @@ def write_json_to_file(self, file_path: str):
         semantic_manifest = self._get_pydantic_semantic_manifest()
         json = semantic_manifest.json()
         write_file(file_path, json)
+        fire_event(ArtifactWritten(artifact_type=self.__class__.__name__, artifact_path=file_path))
 
     def _get_pydantic_semantic_manifest(self) -> PydanticSemanticManifest:
         pydantic_time_spines: List[PydanticTimeSpine] = []

core/dbt/contracts/sql.py

@@ -7,7 +7,9 @@
 from dbt.artifacts.schemas.results import ExecutionResult, TimingInfo
 from dbt.artifacts.schemas.run import RunExecutionResult, RunResult, RunResultsArtifact
 from dbt.contracts.graph.nodes import ResultNode
+from dbt.events.types import ArtifactWritten
 from dbt_common.dataclass_schema import dbtClassMixin
+from dbt_common.events.functions import fire_event
 
 TaskTags = Optional[Dict[str, Any]]
 TaskID = uuid.UUID
@@ -49,6 +51,7 @@
             args=self.args,
         )
         writable.write(path)
+        fire_event(ArtifactWritten(artifact_type=writable.__class__.__name__, artifact_path=path))
 
     @classmethod
     def from_local_result(

core/dbt/events/core_types.proto

@@ -1228,6 +1228,19 @@ message DepsScrubbedPackageNameMsg{
     DepsScrubbedPackageName data = 2;
 }
 
+// P - Artifacts
+
+// P001
+message ArtifactWritten {
+    string artifact_type = 1;
+    string artifact_path = 2;
+}
+
+message ArtifactWrittenMsg {
+    CoreEventInfo info = 1;
+    ArtifactWritten data = 2;
+}
+
 // Q - Node execution
 
 // Q001

core/dbt/events/types.py

@@ -1200,6 +1200,19 @@ def message(self) -> str:
         return f"Detected secret env var in {self.package_name}. dbt will write a scrubbed representation to the lock file. This will cause issues with subsequent 'dbt deps' using the lock file, requiring 'dbt deps --upgrade'"
 
 
+# =======================================================
+# P - Artifacts
+# =======================================================
+
+
+class ArtifactWritten(DebugLevel):
+    def code(self):
+        return "P001"
+
+    def message(self) -> str:
+        return f"Wrote artifact {self.artifact_type} to {self.artifact_path}"
+
+
 # =======================================================
 # Q - Node execution
 # =======================================================

core/dbt/parser/manifest.py

@@ -63,6 +63,7 @@
 )
 from dbt.contracts.graph.semantic_manifest import SemanticManifest
 from dbt.events.types import (
+    ArtifactWritten,
     DeprecatedModel,
     DeprecatedReference,
     InvalidDisabledTargetInTestNode,
@@ -2019,4 +2020,9 @@
         plugin_artifacts = pm.get_manifest_artifacts(manifest)
         for path, plugin_artifact in plugin_artifacts.items():
             plugin_artifact.write(path)
+            fire_event(
+                ArtifactWritten(
+                    artifact_type=plugin_artifact.__class__.__name__, artifact_path=path
+                )
+            )
     return manifest

core/dbt/task/docs/generate.py

@@ -33,6 +33,7 @@
 from dbt.constants import MANIFEST_FILE_NAME
 from dbt.contracts.graph.manifest import Manifest
 from dbt.contracts.graph.nodes import ResultNode
+from dbt.events.types import ArtifactWritten
 from dbt.exceptions import AmbiguousCatalogMatchError
 from dbt.graph import ResourceTypeSelector
 from dbt.graph.graph import UniqueId
@@ -309,6 +310,10 @@ def run(self) -> CatalogArtifact:
 
         catalog_path = os.path.join(self.config.project_target_path, CATALOG_FILENAME)
         results.write(catalog_path)
+        fire_event(
+            ArtifactWritten(artifact_type=results.__class__.__name__, artifact_path=catalog_path)
+        )
+
         if self.args.compile:
             write_manifest(self.manifest, self.config.project_target_path)
 

core/dbt/task/run_operation.py

@@ -11,6 +11,7 @@
 from dbt.contracts.files import FileHash
 from dbt.contracts.graph.nodes import HookNode
 from dbt.events.types import (
+    ArtifactWritten,
     LogDebugStackTrace,
     RunningOperationCaughtError,
     RunningOperationUncaughtError,
@@ -130,6 +131,11 @@ def run(self) -> RunResultsArtifact:
 
         if self.args.write_json:
             results.write(result_path)
+            fire_event(
+                ArtifactWritten(
+                    artifact_type=results.__class__.__name__, artifact_path=result_path
+                )
+            )
 
         return results
 

core/dbt/task/runnable.py

@@ -26,6 +26,7 @@
 from dbt.contracts.graph.nodes import ResultNode
 from dbt.contracts.state import PreviousState
 from dbt.events.types import (
+    ArtifactWritten,
     ConcurrencyLine,
     DefaultSelector,
     EndRunResult,
@@ -427,6 +428,12 @@
 
             if self.args.write_json and hasattr(run_result, "write"):
                 run_result.write(self.result_path())
+                fire_event(
+                    ArtifactWritten(
+                        artifact_type=run_result.__class__.__name__,
+                        artifact_path=self.result_path(),
+                    )
+                )
 
             self._cancel_connections(pool)
             print_run_end_messages(self.node_results, keyboard_interrupt=True)
@@ -591,6 +598,11 @@
             write_manifest(self.manifest, self.config.project_target_path)
             if hasattr(result, "write"):
                 result.write(self.result_path())
+                fire_event(
+                    ArtifactWritten(
+                        artifact_type=result.__class__.__name__, artifact_path=self.result_path()
+                    )
+                )
 
         self.task_end_messages(result.results)
         return result

tests/functional/artifacts/test_artifacts.py

@@ -8,6 +8,7 @@
 from dbt.artifacts.schemas.results import RunStatus
 from dbt.artifacts.schemas.run import RunResultsArtifact
 from dbt.contracts.graph.manifest import WritableManifest
+from dbt.events.types import ArtifactWritten
 from dbt.tests.util import (
     check_datetime_between,
     get_artifact,
@@ -24,6 +25,7 @@
     expected_run_results,
     expected_versions_run_results,
 )
+from tests.utils import EventCatcher
 
 models__schema_yml = """
 version: 2
@@ -617,8 +619,9 @@ def models(self):
 
     # Test generic "docs generate" command
     def test_run_and_generate(self, project, manifest_schema_path, run_results_schema_path):
+        catcher = EventCatcher(ArtifactWritten)
         start_time = datetime.utcnow()
-        results = run_dbt(["compile"])
+        results = run_dbt(args=["compile"], callbacks=[catcher.catch])
         assert len(results) == 7
         verify_manifest(
             project,
@@ -627,6 +630,38 @@ def test_run_and_generate(self, project, manifest_schema_path, run_results_schem
             manifest_schema_path,
         )
         verify_run_results(project, expected_run_results(), start_time, run_results_schema_path)
+        # manifest written twice, semantic manifest written twice, run results written once
+        assert len(catcher.caught_events) == 5
+        assert (
+            len(
+                [
+                    event
+                    for event in catcher.caught_events
+                    if event.data.artifact_type == "WritableManifest"
+                ]
+            )
+            > 0
+        )
+        assert (
+            len(
+                [
+                    event
+                    for event in catcher.caught_events
+                    if event.data.artifact_type == "SemanticManifest"
+                ]
+            )
+            > 0
+        )
+        assert (
+            len(
+                [
+                    event
+                    for event in catcher.caught_events
+                    if event.data.artifact_type == "RunExecutionResult"
+                ]
+            )
+            > 0
+        )
 
 
 class TestVerifyArtifactsReferences(BaseVerifyProject):

tests/unit/test_events.py

@@ -320,6 +320,8 @@ def test_event_codes(self):
     core_types.DepsScrubbedPackageName(package_name=""),
     core_types.DepsUnpinned(revision="", git=""),
     core_types.NoNodesForSelectionCriteria(spec_raw=""),
+    # P - Artifacts ======================
+    core_types.ArtifactWritten(artifact_type="manifest", artifact_path="path/to/artifact.json"),
     # Q - Node execution ======================
     core_types.RunningOperationCaughtError(exc=""),
     core_types.CompileComplete(),