feat: [SNOW-2330065] collect dbt core args via info telemetry

src/snowflake/cli/_app/telemetry.py

@@ -23,7 +23,6 @@
 import click
 import typer
 from snowflake.cli import __about__
-from snowflake.cli._app.cli_app import INTERNAL_CLI_FLAGS
 from snowflake.cli._app.constants import PARAM_APPLICATION_NAME
 from snowflake.cli.api.cli_global_context import (
     _CliGlobalContextAccess,
@@ -75,6 +74,7 @@ class CLITelemetryField(Enum):
     ERROR_CAUSE = "error_cause"
     SQL_STATE = "sql_state"
     IS_CLI_EXCEPTION = "is_cli_exception"
+    EXTRA_INFO = "extra_info"
     # Project context
     PROJECT_DEFINITION_VERSION = "project_definition_version"
     MODE = "mode"
@@ -84,6 +84,7 @@ class TelemetryEvent(Enum):
     CMD_EXECUTION = "executing_command"
     CMD_EXECUTION_ERROR = "error_executing_command"
     CMD_EXECUTION_RESULT = "result_executing_command"
+    CMD_EXECUTION_INFO = "info_executing_command"
 
 
 TelemetryDict = Dict[Union[CLITelemetryField, TelemetryField], Any]
@@ -140,6 +141,8 @@ def _get_command_metrics() -> TelemetryDict:
 
 
 def _find_command_info() -> TelemetryDict:
+    from snowflake.cli._app.cli_app import INTERNAL_CLI_FLAGS
+
     ctx = click.get_current_context()
     command_path = ctx.command_path.split(" ")[1:]
 
@@ -303,6 +306,19 @@ def log_command_execution_error(exception: Exception, execution: ExecutionMetada
     )
 
 
+@ignore_exceptions()
+def log_command_info(custom_data: Dict[str, Any]):
+    """
+    Log custom telemetry data from any command.
+    """
+    _telemetry.send(
+        {
+            TelemetryField.KEY_TYPE: TelemetryEvent.CMD_EXECUTION_INFO.value,
+            CLITelemetryField.EXTRA_INFO: custom_data,
+        }
+    )
+
+
 @ignore_exceptions()
 def flush_telemetry():
     _telemetry.flush()

src/snowflake/cli/_plugins/dbt/commands.py

@@ -19,13 +19,15 @@
 
 import typer
 from click import types
+from snowflake.cli._app.telemetry import log_command_info
 from snowflake.cli._plugins.dbt.constants import (
     DBT_COMMANDS,
     OUTPUT_COLUMN_NAME,
     PROFILES_FILENAME,
     RESULT_COLUMN_NAME,
 )
 from snowflake.cli._plugins.dbt.manager import DBTManager
+from snowflake.cli._plugins.dbt.utils import _extract_dbt_args
 from snowflake.cli._plugins.object.command_aliases import add_object_command_aliases
 from snowflake.cli._plugins.object.commands import scope_option
 from snowflake.cli.api.commands.decorators import global_options_with_connection
@@ -185,6 +187,13 @@ def _dbt_execute(
         execute_args = (dbt_command, name, run_async, *dbt_cli_args)
         dbt_manager = DBTManager()
 
+        log_command_info(
+            {
+                "dbt_command": dbt_command,
+                "dbt_args": _extract_dbt_args(dbt_cli_args),
+            }
+        )
+
         if run_async is True:
             result = dbt_manager.execute(*execute_args)
             return MessageResult(

src/snowflake/cli/_plugins/dbt/constants.py

@@ -39,3 +39,5 @@
     "init",
     "source",
 ]
+
+KNOWN_SUBCOMMANDS = {"generate", "serve", "freshness"}

src/snowflake/cli/_plugins/dbt/utils.py

@@ -0,0 +1,31 @@
+# Copyright (c) 2025 Snowflake Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from snowflake.cli._plugins.dbt.constants import KNOWN_SUBCOMMANDS
+
+
+def _extract_dbt_args(args: list[str]) -> list[str]:
+    flags = set()
+
+    for arg in args:
+        if arg.startswith("-"):
+            if "=" in arg:
+                flag_name = arg.split("=", 1)[0]
+                flags.add(flag_name)
+            else:
+                flags.add(arg)
+        elif arg in KNOWN_SUBCOMMANDS:
+            flags.add(arg)
+
+    return sorted(list(flags))

tests/dbt/test_dbt_commands.py

@@ -490,3 +490,36 @@ def test_dbt_execute_no_rows_in_response(self, mock_connect, mock_cursor, runner
 
         assert result.exit_code == 1, result.output
         assert "No data returned from server" in result.output
+
+    @mock.patch("snowflake.cli._plugins.dbt.commands.log_command_info")
+    def test_dbt_execute_telemetry_data_masking(
+        self, mock_log_command_info, mock_connect, mock_cursor, runner
+    ):
+        cursor = mock_cursor(
+            rows=[(True, "command output")],
+            columns=[RESULT_COLUMN_NAME, OUTPUT_COLUMN_NAME],
+        )
+        mock_connect.mocked_ctx.cs = cursor
+
+        result = runner.invoke(
+            [
+                "dbt",
+                "execute",
+                "pipeline_name",
+                "test",
+                "generate",
+                "--select",
+                "my_sensitive_model",
+                "--vars",
+                "'{api_key: secret123}'",
+            ]
+        )
+
+        assert result.exit_code == 0, result.output
+
+        mock_log_command_info.assert_called_once()
+        call_args = mock_log_command_info.call_args[0][0]
+
+        assert call_args["dbt_command"] == "test"
+        actual_dbt_args = call_args["dbt_args"]
+        assert sorted(actual_dbt_args) == sorted(["generate", "--select", "--vars"])

tests/dbt/test_utils.py

@@ -0,0 +1,71 @@
+# Copyright (c) 2025 Snowflake Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import pytest
+from snowflake.cli._plugins.dbt.utils import _extract_dbt_args
+
+
+class TestDBTUtilsFunction:
+    @pytest.mark.parametrize(
+        "input_args,expected,sensitive_patterns",
+        [
+            pytest.param([], [], [], id="empty_args"),
+            pytest.param(
+                ["-f", "--debug"],
+                ["-f", "--debug"],
+                [],
+                id="safe_boolean_flags",
+            ),
+            pytest.param(
+                ["--select", "sensitive_model"],
+                ["--select"],
+                ["sensitive_model"],
+                id="model_names_masked",
+            ),
+            pytest.param(
+                ["--vars", "'{api_key: secret}'"],
+                ["--vars"],
+                ["secret", "api_key"],
+                id="variables_masked",
+            ),
+            pytest.param(
+                ["--format=JSON"],
+                ["--format"],
+                ["JSON"],
+                id="compound_args_handled",
+            ),
+            pytest.param(
+                ["generate", "--profiles-dir", "/secret/path"],
+                ["--profiles-dir", "generate"],
+                ["secret", "path"],
+                id="subcommand_with_sensitive_path",
+            ),
+            pytest.param(
+                ["--select", "pii.customers", "--vars", "'{password: abc123}'", "-f"],
+                ["-f", "--select", "--vars"],
+                ["pii", "customers", "abc123", "password"],
+                id="complex_sensitive_data_masked",
+            ),
+        ],
+    )
+    def test_extract_dbt_args(self, input_args, expected, sensitive_patterns):
+        result = _extract_dbt_args(list(input_args))
+
+        assert sorted(result) == sorted(expected)
+
+        result_str = str(result)
+        for pattern in sensitive_patterns:
+            assert (
+                pattern not in result_str
+            ), f"Sensitive '{pattern}' leaked in result: {result}"