feat: add POST_RUN outcome telemetry

CHANGELOG.md

@@ -1,5 +1,10 @@
 ## dbt-databricks next
 
+### Features
+
+- Add opt-in POST_PARSE invocation telemetry for eligible commands via `connection_parameters.enable_dbt_telemetry` ([#1647](https://github.com/databricks/dbt-databricks/pull/1647))
+- Add POST_RUN outcome telemetry to the opt-in invocation telemetry path ([#1648](https://github.com/databricks/dbt-databricks/pull/1648))
+
 ### Fixes
 
 - Skip unnecessary Unity Catalog constraint metadata queries for incremental models without enforced contracts ([#1643](https://github.com/databricks/dbt-databricks/pull/1643) resolves [#1641](https://github.com/databricks/dbt-databricks/issues/1641))

dbt/adapters/databricks/connections.py

@@ -55,6 +55,7 @@
 from dbt.adapters.databricks.logging import logger
 from dbt.adapters.databricks.python_models.run_tracking import PythonRunTracker
 from dbt.adapters.databricks.spog.decision import check_spog_preconditions
+from dbt.adapters.databricks.telemetry import hooks as telemetry_hooks
 from dbt.adapters.databricks.utils import QueryTagsUtils, is_cluster_http_path, redact_credentials
 
 if TYPE_CHECKING:
@@ -482,7 +483,9 @@ def open(cls, connection: Connection) -> Connection:
         creds: DatabricksCredentials = connection.credentials
         timeout = creds.connect_timeout
 
-        cls.credentials_manager = creds.authenticate()
+        # Avoid a manager overwritten by another concurrent open.
+        credentials_manager = creds.authenticate()
+        cls.credentials_manager = credentials_manager
 
         # SPOG decision matrix: collect every http_path in play (default +
         # per-compute) and validate them against the host's discovery probe.
@@ -502,7 +505,7 @@ def open(cls, connection: Connection) -> Connection:
             merged_query_tags = QueryConfigUtils.get_merged_query_tags(query_header_context, creds)
 
         conn_args = SqlUtils.prepare_connection_arguments(
-            creds, cls.credentials_manager, databricks_connection.http_path, merged_query_tags
+            creds, credentials_manager, databricks_connection.http_path, merged_query_tags
         )
 
         def connect() -> DatabricksHandle:
@@ -518,6 +521,10 @@ def connect() -> DatabricksHandle:
                     databricks_connection.capabilities = cls._get_capabilities_for_http_path(
                         databricks_connection.http_path
                     )
+
+                    telemetry_hooks.on_connection_open(
+                        creds, credentials_manager, databricks_connection.http_path
+                    )
                     return conn
                 else:
                     raise DbtDatabaseError("Failed to create connection")

dbt/adapters/databricks/handle.py

@@ -394,6 +394,8 @@ def prepare_connection_arguments(
 
         connection_parameters = creds.connection_parameters.copy()  # type: ignore[union-attr]
 
+        connection_parameters.pop("enable_dbt_telemetry", None)
+
         http_headers: list[tuple[str, str]] = list(
             creds.get_all_http_headers(connection_parameters.pop("http_headers", {})).items()
         )

dbt/adapters/databricks/impl.py

@@ -102,6 +102,7 @@
     sdk_supports_workspace_id,
 )
 from dbt.adapters.databricks.spog.extract import extract_workspace_id
+from dbt.adapters.databricks.telemetry import hooks as telemetry_hooks
 from dbt.adapters.databricks.utils import (
     get_first_row,
     handle_missing_objects,
@@ -296,6 +297,8 @@ def __init__(self, config: Any, mp_context: SpawnContext) -> None:
             self.get_behavior_flag_no_warn(USE_MANAGED_ICEBERG["name"])
         )
 
+        telemetry_hooks.on_adapter_init(self)
+
         # Warehouses always meet capability cutoffs at parse time; clusters keep the
         # conservative False until a real connection is available.
         # `_parse_replacements_` is injected by AdapterMeta, so mypy can't resolve it here.
@@ -903,6 +906,14 @@ def get_behavior_flag_no_warn(self, behavior_flag_name: str) -> bool:
         behavior_flag = getattr(self.behavior, behavior_flag_name)
         return behavior_flag.no_warn
 
+    def set_macro_resolver(self, macro_resolver: Any) -> None:
+        super().set_macro_resolver(macro_resolver)
+        telemetry_hooks.on_post_parse(self, macro_resolver)
+
+    def cleanup_connections(self) -> None:
+        telemetry_hooks.on_run_end(self)
+        super().cleanup_connections()
+
     @available.parse(lambda *a, **k: (None, None))
     @record_function(
         DatabricksAdapterAddQueryRecord,

dbt/adapters/databricks/telemetry/__init__.py

@@ -0,0 +1,3 @@
+from dbt.adapters.databricks.telemetry.config import is_enabled
+
+__all__ = ["is_enabled"]