feat: add wiring for telemetry v1

CHANGELOG.md

@@ -1,5 +1,9 @@
 ## dbt-databricks next
 
+### Features
+
+- Add opt-in invocation telemetry for eligible commands via `connection_parameters.enable_dbt_telemetry` ([#1620](https://github.com/databricks/dbt-databricks/pull/1620))
+
 ### Fixes
 
 - Recreate materialized views when query schema drifts, honoring `on_configuration_change` ([#1621](https://github.com/databricks/dbt-databricks/pull/1621) resolves [#1359](https://github.com/databricks/dbt-databricks/issues/1359))

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,8 @@ def connect() -> DatabricksHandle:
                     databricks_connection.capabilities = cls._get_capabilities_for_http_path(
                         databricks_connection.http_path
                     )
+
+                    telemetry_hooks.on_connection_open(creds, credentials_manager)
                     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

@@ -101,6 +101,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,
@@ -295,6 +296,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.
@@ -902,6 +905,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"]