feat: add resource retrieval telemetry tracking with new record type and helper function

Author: msanatan

MCPForUnity/UnityMcpServer~/src/telemetry.py

@@ -20,7 +20,7 @@
 import sys
 import threading
 import time
-from typing import Optional, Dict, Any
+from typing import Any
 from urllib.parse import urlparse
 import uuid
 
@@ -55,6 +55,7 @@ class RecordType(str, Enum):
     USAGE = "usage"
     LATENCY = "latency"
     FAILURE = "failure"
+    RESOURCE_RETRIEVAL = "resource_retrieval"
     TOOL_EXECUTION = "tool_execution"
     UNITY_CONNECTION = "unity_connection"
     CLIENT_CONNECTION = "client_connection"
@@ -78,8 +79,8 @@ class TelemetryRecord:
     timestamp: float
     customer_uuid: str
     session_id: str
-    data: Dict[str, Any]
-    milestone: Optional[MilestoneType] = None
+    data: dict[str, Any]
+    milestone: MilestoneType | None = None
 
 
 class TelemetryConfig:
@@ -208,8 +209,8 @@ class TelemetryCollector:
 
     def __init__(self):
         self.config = TelemetryConfig()
-        self._customer_uuid: Optional[str] = None
-        self._milestones: Dict[str, Dict[str, Any]] = {}
+        self._customer_uuid: str | None = None
+        self._milestones: dict[str, dict[str, Any]] = {}
         self._lock: threading.Lock = threading.Lock()
         # Bounded queue with single background worker (records only; no context propagation)
         self._queue: "queue.Queue[TelemetryRecord]" = queue.Queue(maxsize=1000)
@@ -262,7 +263,7 @@ def _save_milestones(self):
         except OSError as e:
             logger.warning(f"Failed to save milestones: {e}", exc_info=True)
 
-    def record_milestone(self, milestone: MilestoneType, data: Optional[Dict[str, Any]] = None) -> bool:
+    def record_milestone(self, milestone: MilestoneType, data: dict[str, Any] | None = None) -> bool:
         """Record a milestone event, returns True if this is the first occurrence"""
         if not self.config.enabled:
             return False
@@ -288,8 +289,8 @@ def record_milestone(self, milestone: MilestoneType, data: Optional[Dict[str, An
 
     def record(self,
                record_type: RecordType,
-               data: Dict[str, Any],
-               milestone: Optional[MilestoneType] = None):
+               data: dict[str, Any],
+               milestone: MilestoneType | None = None):
         """Record a telemetry event (async, non-blocking)"""
         if not self.config.enabled:
             return
@@ -393,7 +394,7 @@ def _send_telemetry(self, record: TelemetryRecord):
 
 
 # Global telemetry instance
-_telemetry_collector: Optional[TelemetryCollector] = None
+_telemetry_collector: TelemetryCollector | None = None
 
 
 def get_telemetry() -> TelemetryCollector:
@@ -405,18 +406,18 @@ def get_telemetry() -> TelemetryCollector:
 
 
 def record_telemetry(record_type: RecordType,
-                     data: Dict[str, Any],
-                     milestone: Optional[MilestoneType] = None):
+                     data: dict[str, Any],
+                     milestone: MilestoneType | None = None):
     """Convenience function to record telemetry"""
     get_telemetry().record(record_type, data, milestone)
 
 
-def record_milestone(milestone: MilestoneType, data: Optional[Dict[str, Any]] = None) -> bool:
+def record_milestone(milestone: MilestoneType, data: dict[str, Any] | None = None) -> bool:
     """Convenience function to record a milestone"""
     return get_telemetry().record_milestone(milestone, data)
 
 
-def record_tool_usage(tool_name: str, success: bool, duration_ms: float, error: Optional[str] = None, sub_action: Optional[str] = None):
+def record_tool_usage(tool_name: str, success: bool, duration_ms: float, error: str | None = None, sub_action: str | None = None):
     """Record tool usage telemetry
 
     Args:
@@ -445,7 +446,28 @@ def record_tool_usage(tool_name: str, success: bool, duration_ms: float, error:
     record_telemetry(RecordType.TOOL_EXECUTION, data)
 
 
-def record_latency(operation: str, duration_ms: float, metadata: Optional[Dict[str, Any]] = None):
+def record_resource_usage(resource_name: str, success: bool, duration_ms: float, error: str | None = None):
+    """Record resource usage telemetry
+
+    Args:
+        resource_name: Name of the resource invoked (e.g., 'get_tests').
+        success: Whether the resource completed successfully.
+        duration_ms: Execution duration in milliseconds.
+        error: Optional error message (truncated if present).
+    """
+    data = {
+        "resource_name": resource_name,
+        "success": success,
+        "duration_ms": round(duration_ms, 2)
+    }
+
+    if error:
+        data["error"] = str(error)[:200]  # Limit error message length
+
+    record_telemetry(RecordType.RESOURCE_RETRIEVAL, data)
+
+
+def record_latency(operation: str, duration_ms: float, metadata: dict[str, Any] | None = None):
     """Record latency telemetry"""
     data = {
         "operation": operation,
@@ -458,7 +480,7 @@ def record_latency(operation: str, duration_ms: float, metadata: Optional[Dict[s
     record_telemetry(RecordType.LATENCY, data)
 
 
-def record_failure(component: str, error: str, metadata: Optional[Dict[str, Any]] = None):
+def record_failure(component: str, error: str, metadata: dict[str, Any] | None = None):
     """Record failure telemetry"""
     data = {
         "component": component,

MCPForUnity/UnityMcpServer~/src/telemetry_decorator.py

@@ -8,7 +8,7 @@
 import time
 from typing import Callable, Any
 
-from telemetry import record_tool_usage, record_milestone, MilestoneType
+from telemetry import record_resource_usage, record_tool_usage, record_milestone, MilestoneType
 
 _log = logging.getLogger("unity-mcp-telemetry")
 _decorator_log_count = 0