[WIP] feat: add telemetry for cli

src/uipath/_cli/_telemetry.py

@@ -0,0 +1,195 @@
+import logging
+import os
+import time
+from functools import wraps
+from typing import Any, Callable, Dict, Optional
+
+from uipath.telemetry._track import flush_events, is_telemetry_enabled, track_event
+
+logger = logging.getLogger(__name__)
+
+# Telemetry event name templates for Application Insights
+CLI_COMMAND_STARTED = "Cli.{command}.Start.URT"
+CLI_COMMAND_COMPLETED = "Cli.{command}.End.URT"
+CLI_COMMAND_FAILED = "Cli.{command}.Failed.URT"
+
+
+class CliTelemetryTracker:
+    """Tracks CLI command execution and sends telemetry to Application Insights.
+
+    This class handles tracking of CLI command lifecycle events:
+    - Command start events
+    - Command completion events (success)
+    - Command failure events (with error details)
+    """
+
+    def __init__(self) -> None:
+        self._start_times: Dict[str, float] = {}
+
+    @staticmethod
+    def _get_event_name(command: str, status: str) -> str:
+        return f"Cli.{command.capitalize()}.{status}.URT"
+
+    def _enrich_properties(self, properties: Dict[str, Any]) -> None:
+        """Enrich properties with common context information.
+
+        Args:
+            properties: The properties dictionary to enrich.
+        """
+        # Add UiPath context
+        project_id = os.getenv("UIPATH_PROJECT_ID")
+        if project_id:
+            properties["ProjectId"] = project_id
+
+        org_id = os.getenv("UIPATH_CLOUD_ORGANIZATION_ID")
+        if org_id:
+            properties["CloudOrganizationId"] = org_id
+
+        user_id = os.getenv("UIPATH_CLOUD_USER_ID")
+        if user_id:
+            properties["CloudUserId"] = user_id
+
+        tenant_id = os.getenv("UIPATH_TENANT_ID")
+        if tenant_id:
+            properties["TenantId"] = tenant_id
+
+        # Add source identifier
+        properties["Source"] = "uipath-python-cli"
+        properties["ApplicationName"] = "UiPath.Cli"
+
+    def track_command_start(self, command: str) -> None:
+        try:
+            self._start_times[command] = time.time()
+
+            properties: Dict[str, Any] = {"Command": command}
+            self._enrich_properties(properties)
+
+            track_event(self._get_event_name(command, "Start"), properties)
+            logger.debug(f"Tracked CLI command started: {command}")
+
+        except Exception as e:
+            logger.debug(f"Error tracking CLI command start: {e}")
+
+    def track_command_end(
+        self,
+        command: str,
+        duration_ms: Optional[int] = None,
+    ) -> None:
+        try:
+            if duration_ms is None:
+                start_time = self._start_times.pop(command, None)
+                if start_time:
+                    duration_ms = int((time.time() - start_time) * 1000)
+
+            properties: Dict[str, Any] = {
+                "Command": command,
+                "Success": True,
+            }
+
+            if duration_ms is not None:
+                properties["DurationMs"] = duration_ms
+
+            self._enrich_properties(properties)
+
+            track_event(self._get_event_name(command, "End"), properties)
+            logger.debug(f"Tracked CLI command completed: {command}")
+
+        except Exception as e:
+            logger.debug(f"Error tracking CLI command end: {e}")
+
+    def track_command_failed(
+        self,
+        command: str,
+        duration_ms: Optional[int] = None,
+        exception: Optional[Exception] = None,
+    ) -> None:
+        try:
+            if duration_ms is None:
+                start_time = self._start_times.pop(command, None)
+                if start_time:
+                    duration_ms = int((time.time() - start_time) * 1000)
+
+            properties: Dict[str, Any] = {
+                "Command": command,
+                "Success": False,
+            }
+
+            if duration_ms is not None:
+                properties["DurationMs"] = duration_ms
+
+            if exception is not None:
+                properties["ErrorType"] = type(exception).__name__
+                properties["ErrorMessage"] = str(exception)[:500]
+
+            self._enrich_properties(properties)
+
+            track_event(self._get_event_name(command, "Failed"), properties)
+            logger.debug(f"Tracked CLI command failed: {command}")
+
+        except Exception as e:
+            logger.debug(f"Error tracking CLI command failed: {e}")
+
+    def flush(self) -> None:
+        """Flush any pending telemetry events."""
+        try:
+            flush_events()
+        except Exception as e:
+            logger.debug(f"Error flushing CLI telemetry events: {e}")
+
+
+def track_cli_command(command: str) -> Callable[..., Any]:
+    """Decorator to track CLI command execution.
+
+    Tracks the following events to Application Insights:
+    - Cli.<Command>.Start.URT - when command begins
+    - Cli.<Command>.End.URT - on successful completion
+    - Cli.<Command>.Failed.URT - on exception
+
+    Properties tracked include:
+    - Command: The command name
+    - Success: Whether the command succeeded
+    - DurationMs: Execution time in milliseconds
+    - ErrorType: Exception type name (on failure)
+    - ErrorMessage: Exception message (on failure, truncated to 500 chars)
+    - ProjectId, CloudOrganizationId, etc. (if available)
+
+    Telemetry failures are silently ignored to ensure CLI execution
+    is never blocked by telemetry issues.
+
+    Args:
+        command: The CLI command name (e.g., "pack", "publish", "run").
+
+    Returns:
+        A decorator function that wraps the CLI command.
+
+    Example:
+        @click.command()
+        @track_cli_command("pack")
+        def pack(root, nolock):
+            ...
+    """
+
+    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
+        @wraps(func)
+        def wrapper(*args: Any, **kwargs: Any) -> Any:
+            if not is_telemetry_enabled():
+                return func(*args, **kwargs)
+
+            tracker = CliTelemetryTracker()
+            tracker.track_command_start(command)
+
+            try:
+                result = func(*args, **kwargs)
+                tracker.track_command_end(command)
+                return result
+
+            except Exception as e:
+                tracker.track_command_failed(command, exception=e)
+                raise
+
+            finally:
+                tracker.flush()
+
+        return wrapper
+
+    return decorator

src/uipath/_cli/cli_init.py

@@ -27,6 +27,7 @@
 
 from .._utils.constants import ENV_TELEMETRY_ENABLED
 from ..telemetry._constants import _PROJECT_KEY, _TELEMETRY_CONFIG_FILE
+from ._telemetry import track_cli_command
 from ._utils._console import ConsoleLogger
 from .middlewares import Middlewares
 from .models.runtime_schema import Bindings
@@ -252,6 +253,7 @@ def _add_graph_to_chart(chart: Chart | Subgraph, graph: UiPathRuntimeGraph) -> N
     default=False,
     help="Won't override existing .agent files and AGENTS.md file.",
 )
+@track_cli_command("init")
 def init(no_agents_md_override: bool) -> None:
     """Initialize the project."""
     with console.spinner("Initializing UiPath project ..."):

src/uipath/_cli/cli_invoke.py

@@ -7,6 +7,7 @@
 import httpx
 
 from .._utils._ssl_context import get_httpx_client_kwargs
+from ._telemetry import track_cli_command
 from ._utils._common import get_env_vars
 from ._utils._console import ConsoleLogger
 from ._utils._folders import get_personal_workspace_info_async
@@ -43,6 +44,7 @@ def _read_project_details() -> tuple[str, str]:
     type=click.Path(exists=True),
     help="File path for the .json input",
 )
+@track_cli_command("invoke")
 def invoke(entrypoint: str | None, input: str | None, file: str | None) -> None:
     """Invoke an agent published in my workspace."""
     if file:

src/uipath/_cli/cli_new.py

@@ -4,6 +4,7 @@
 
 import click
 
+from ._telemetry import track_cli_command
 from ._utils._console import ConsoleLogger
 from .middlewares import Middlewares
 
@@ -46,6 +47,7 @@ def generate_uipath_json(target_directory):
 
 @click.command()
 @click.argument("name", type=str, default="")
+@track_cli_command("new")
 def new(name: str):
     """Generate a quick-start project."""
     directory = os.getcwd()

src/uipath/_cli/cli_pack.py

@@ -14,6 +14,7 @@
 from uipath.platform.common import UiPathConfig
 
 from ..telemetry._constants import _PROJECT_KEY, _TELEMETRY_CONFIG_FILE
+from ._telemetry import track_cli_command
 from ._utils._console import ConsoleLogger
 from ._utils._project_files import (
     ensure_config_file,
@@ -336,6 +337,7 @@ def display_project_info(config):
     is_flag=True,
     help="Skip running uv lock and exclude uv.lock from the package",
 )
+@track_cli_command("pack")
 def pack(root, nolock):
     """Pack the project."""
     version = get_project_version(root)

src/uipath/_cli/cli_publish.py

@@ -6,6 +6,7 @@
 import httpx
 
 from .._utils._ssl_context import get_httpx_client_kwargs
+from ._telemetry import track_cli_command
 from ._utils._common import get_env_vars
 from ._utils._console import ConsoleLogger
 from ._utils._folders import get_personal_workspace_info_async
@@ -118,6 +119,7 @@ def find_feed_by_folder_name(
     type=str,
     help="Folder name to publish to (skips interactive selection)",
 )
+@track_cli_command("publish")
 def publish(feed, folder):
     """Publish the package."""
     [base_url, token] = get_env_vars()

src/uipath/_cli/cli_run.py

@@ -23,6 +23,7 @@
 from uipath._utils._bindings import ResourceOverwritesContext
 from uipath.tracing import JsonLinesFileExporter, LlmOpsHttpExporter
 
+from ._telemetry import track_cli_command
 from ._utils._console import ConsoleLogger
 from .middlewares import Middlewares
 
@@ -80,6 +81,7 @@
     is_flag=True,
     help="Keep the temporary state file even when not resuming and no job id is provided",
 )
+@track_cli_command("run")
 def run(
     entrypoint: str | None,
     input: str | None,