feat: Add telemetry

fabriziodemaria · fabriziodemaria · commit ea6ea8761cce · 2025-05-05T17:33:22.000+02:00
diff --git a/confidence/confidence.py b/confidence/confidence.py
@@ -19,6 +19,7 @@
 import requests
 import httpx
 from typing_extensions import TypeGuard
+import time
 
 from confidence import __version__
 from confidence.errors import (
@@ -30,6 +31,7 @@
 )
 from .flag_types import FlagResolutionDetails, Reason, ErrorCode
 from .names import FlagName, VariantName
+from .telemetry import Telemetry, ProtoTraceId, ProtoStatus
 
 EU_RESOLVE_API_ENDPOINT = "https://resolver.eu.confidence.dev"
 US_RESOLVE_API_ENDPOINT = "https://resolver.us.confidence.dev"
@@ -111,6 +113,16 @@ def __init__(
         self.async_client = async_client
         self._setup_logger(logger)
         self._custom_resolve_base_url = custom_resolve_base_url
+        self._telemetry = Telemetry(__version__)
+
+    def _get_resolve_headers(self) -> Dict[str, str]:
+        telemetry_header = self._telemetry.get_monitoring_header()    
+        headers = {
+            "Content-Type": "application/json",
+            "Accept": "application/json",
+            "X-CONFIDENCE-TELEMETRY": telemetry_header
+        }
+        return headers
 
     def resolve_boolean_details(
         self, flag_key: str, default_value: bool
@@ -367,7 +379,6 @@ def _send_event_internal(self, event_name: str, data: Dict[str, FieldType]) -> N
             )
             if response.status_code == 200:
                 json = response.json()
-
                 json_errors = json.get("errors")
                 if json_errors:
                     self.logger.warning("events emitted with errors:")
@@ -407,37 +418,55 @@ def _handle_resolve_response(
     def _resolve(
         self, flag_name: FlagName, context: Dict[str, FieldType]
     ) -> ResolveResult:
-        request_body = {
-            "clientSecret": self._client_secret,
-            "evaluationContext": context,
-            "apply": self._apply_on_resolve,
-            "flags": [str(flag_name)],
-            "sdk": {"id": "SDK_ID_PYTHON_CONFIDENCE", "version": __version__},
-        }
-        base_url = self._api_endpoint
-        if self._custom_resolve_base_url is not None:
-            base_url = self._custom_resolve_base_url
-
-        resolve_url = f"{base_url}/v1/flags:resolve"
-        timeout_sec = None if self._timeout_ms is None else self._timeout_ms / 1000.0
+        start_time = time.perf_counter()
         try:
-            response = requests.post(
-                resolve_url, json=request_body, timeout=timeout_sec
-            )
-            return self._handle_resolve_response(response, flag_name)
-        except requests.exceptions.Timeout:
-            self.logger.warning(
-                f"Request timed out after {timeout_sec}s"
-                f" when resolving flag {flag_name}"
-            )
-            raise TimeoutError()
-        except requests.exceptions.RequestException as e:
-            self.logger.warning(f"Error resolving flag {flag_name}: {str(e)}")
-            raise GeneralError(str(e))
+            request_body = {
+                "clientSecret": self._client_secret,
+                "evaluationContext": context,
+                "apply": self._apply_on_resolve,
+                "flags": [str(flag_name)],
+                "sdk": {"id": "SDK_ID_PYTHON_CONFIDENCE", "version": __version__},
+            }
+            base_url = self._api_endpoint
+            if self._custom_resolve_base_url is not None:
+                base_url = self._custom_resolve_base_url
+
+            resolve_url = f"{base_url}/v1/flags:resolve"
+            timeout_sec = None if self._timeout_ms is None else self._timeout_ms / 1000.0
+            
+            try:
+                response = requests.post(
+                    resolve_url, 
+                    json=request_body, 
+                    headers=self._get_resolve_headers(),
+                    timeout=timeout_sec
+                )
+                
+                result = self._handle_resolve_response(response, flag_name)
+                duration_ms = int((time.perf_counter() - start_time) * 1000)
+                self._telemetry.add_trace(ProtoTraceId.PROTO_TRACE_ID_RESOLVE_LATENCY, duration_ms, ProtoStatus.PROTO_STATUS_SUCCESS)
+                return result
+            except requests.exceptions.Timeout:
+                duration_ms = int((time.perf_counter() - start_time) * 1000)
+                self._telemetry.add_trace(ProtoTraceId.PROTO_TRACE_ID_RESOLVE_LATENCY, duration_ms, ProtoStatus.PROTO_STATUS_TIMEOUT)
+                self.logger.warning(
+                    f"Request timed out after {timeout_sec}s"
+                    f" when resolving flag {flag_name}"
+                )
+                raise TimeoutError()
+            except requests.exceptions.RequestException as e:
+                duration_ms = int((time.perf_counter() - start_time) * 1000)
+                self._telemetry.add_trace(ProtoTraceId.PROTO_TRACE_ID_RESOLVE_LATENCY, duration_ms, ProtoStatus.PROTO_STATUS_ERROR)
+                self.logger.warning(f"Error resolving flag {flag_name}: {str(e)}")
+                raise GeneralError(str(e))
+        except Exception as e:
+            # Just re-raise any other exceptions without adding another trace
+            raise e
 
     async def _resolve_async(
         self, flag_name: FlagName, context: Dict[str, FieldType]
     ) -> ResolveResult:
+        start_time = time.perf_counter()
         request_body = {
             "clientSecret": self._client_secret,
             "evaluationContext": context,
@@ -453,16 +482,26 @@ async def _resolve_async(
         timeout_sec = None if self._timeout_ms is None else self._timeout_ms / 1000.0
         try:
             response = await self.async_client.post(
-                resolve_url, json=request_body, timeout=timeout_sec
+                resolve_url, 
+                json=request_body, 
+                headers=self._get_resolve_headers(),
+                timeout=timeout_sec
             )
-            return self._handle_resolve_response(response, flag_name)
+            result = self._handle_resolve_response(response, flag_name)
+            duration_ms = int((time.perf_counter() - start_time) * 1000)
+            self._telemetry.add_trace(ProtoTraceId.PROTO_TRACE_ID_RESOLVE_LATENCY, duration_ms, ProtoStatus.PROTO_STATUS_SUCCESS)
+            return result
         except httpx.TimeoutException:
+            duration_ms = int((time.perf_counter() - start_time) * 1000)
+            self._telemetry.add_trace(ProtoTraceId.PROTO_TRACE_ID_RESOLVE_LATENCY, duration_ms, ProtoStatus.PROTO_STATUS_TIMEOUT)
             self.logger.warning(
                 f"Request timed out after {timeout_sec}s"
                 f" when resolving flag {flag_name}"
             )
             raise TimeoutError()
         except httpx.HTTPError as e:
+            duration_ms = int((time.perf_counter() - start_time) * 1000)
+            self._telemetry.add_trace(ProtoTraceId.PROTO_TRACE_ID_RESOLVE_LATENCY, duration_ms, ProtoStatus.PROTO_STATUS_ERROR)
             self.logger.warning(f"Error resolving flag {flag_name}: {str(e)}")
             raise GeneralError(str(e))
 
diff --git a/confidence/telemetry.proto b/confidence/telemetry.proto
@@ -0,0 +1,66 @@
+syntax = "proto3";
+
+package confidence.telemetry.v1;
+
+option go_package = "github.com/spotify/confidence-sdk-go/pkg/confidence";
+
+enum ProtoPlatform {
+  PROTO_PLATFORM_UNSPECIFIED = 0;
+  PROTO_PLATFORM_JS_WEB = 4;
+  PROTO_PLATFORM_JS_SERVER = 5;
+  PROTO_PLATFORM_PYTHON = 6;
+  PROTO_PLATFORM_GO = 7;
+}
+
+message ProtoMonitoring {
+  repeated ProtoLibraryTraces library_traces = 1;
+  ProtoPlatform platform = 2;
+}
+
+message ProtoLibraryTraces {
+  ProtoLibrary library = 1;
+  string library_version = 2;
+  repeated ProtoTrace traces = 3;
+
+  message ProtoTrace {
+    ProtoTraceId id = 1;
+
+    // DEPRECATED
+    optional uint64 millisecond_duration = 2;
+
+    oneof trace {
+      ProtoRequestTrace request_trace = 3;
+      ProtoCountTrace count_trace = 4;
+    }
+
+    message ProtoCountTrace {}
+
+    message ProtoRequestTrace {
+      uint64 millisecond_duration = 1;
+      ProtoStatus status = 2;
+
+      enum ProtoStatus {
+        PROTO_STATUS_UNSPECIFIED = 0;
+        PROTO_STATUS_SUCCESS = 1;
+        PROTO_STATUS_ERROR = 2;
+        PROTO_STATUS_TIMEOUT = 3;
+        PROTO_STATUS_CACHED = 4;
+      }
+    }
+  }
+
+  enum ProtoLibrary {
+    PROTO_LIBRARY_UNSPECIFIED = 0;
+    PROTO_LIBRARY_CONFIDENCE = 1;
+    PROTO_LIBRARY_OPEN_FEATURE = 2;
+    PROTO_LIBRARY_REACT = 3;
+  }
+
+  enum ProtoTraceId {
+    PROTO_TRACE_ID_UNSPECIFIED = 0;
+    PROTO_TRACE_ID_RESOLVE_LATENCY = 1;
+    PROTO_TRACE_ID_STALE_FLAG = 2;
+    PROTO_TRACE_ID_FLAG_TYPE_MISMATCH = 3;
+    PROTO_TRACE_ID_WITH_CONTEXT = 4;
+  }
+} 
diff --git a/confidence/telemetry.py b/confidence/telemetry.py
@@ -0,0 +1,61 @@
+import time
+from typing import List
+import base64
+from queue import Queue
+from confidence.telemetry_pb2 import (
+    ProtoMonitoring,
+    ProtoLibraryTraces,
+    ProtoPlatform,
+)
+
+# Get the nested classes from ProtoLibraryTraces
+ProtoTrace = ProtoLibraryTraces.ProtoTrace
+ProtoLibrary = ProtoLibraryTraces.ProtoLibrary
+ProtoTraceId = ProtoLibraryTraces.ProtoTraceId
+ProtoStatus = ProtoLibraryTraces.ProtoTrace.ProtoRequestTrace.ProtoStatus
+
+class Telemetry:
+    _instance = None
+    _initialized = False
+
+    def __new__(cls, version: str):
+        if cls._instance is None:
+            cls._instance = super(Telemetry, cls).__new__(cls)
+        return cls._instance
+
+    def __init__(self, version: str):
+        if not self._initialized:
+            self.version = version
+            self._traces_queue = Queue()
+            self._initialized = True
+
+    def add_trace(self, trace_id: ProtoTraceId, duration_ms: int, status: ProtoStatus) -> None:
+        trace = ProtoTrace()
+        trace.id = trace_id
+        trace.millisecond_duration = duration_ms
+        request_trace = ProtoTrace.ProtoRequestTrace()
+        request_trace.status = status
+        trace.request_trace.CopyFrom(request_trace)
+        self._traces_queue.put(trace)
+
+    def get_monitoring_header(self) -> str:
+        # Get all current traces atomically
+        current_traces = []
+        while not self._traces_queue.empty():
+            try:
+                current_traces.append(self._traces_queue.get_nowait())
+            except:
+                break
+            
+        # Create monitoring data with the captured traces
+        monitoring = ProtoMonitoring()
+        library_traces = monitoring.library_traces.add()
+        library_traces.library = ProtoLibrary.PROTO_LIBRARY_CONFIDENCE
+        library_traces.library_version = self.version
+        library_traces.traces.extend(current_traces)
+        monitoring.platform = ProtoPlatform.PROTO_PLATFORM_PYTHON
+        
+        # Serialize to protobuf and base64 encode
+        serialized = monitoring.SerializeToString()
+        encoded = base64.b64encode(serialized).decode()
+        return encoded
diff --git a/confidence/telemetry_pb2.py b/confidence/telemetry_pb2.py
diff --git a/generate_proto.py b/generate_proto.py
@@ -0,0 +1,40 @@
+#!/usr/bin/env python3
+import os
+import subprocess
+import sys
+
+def generate_proto():
+    proto_file = "confidence/telemetry.proto"
+    output_dir = "confidence"
+    
+    # Check if protoc is installed
+    try:
+        version = subprocess.check_output(["protoc", "--version"]).decode().strip()
+        print(f"Found protoc version: {version}")
+    except FileNotFoundError:
+        print("Error: protoc compiler not found. Please install it first.")
+        print("You can install it via:")
+        print("  - macOS: brew install protobuf")
+        print("  - Linux: apt-get install protobuf-compiler")
+        print("  - Windows: Download from https://github.com/protocolbuffers/protobuf/releases")
+        sys.exit(1)
+
+    # Generate Python code
+    cmd = [
+        "protoc",
+        f"--python_out={output_dir}",
+        f"--proto_path={os.path.dirname(proto_file)}",
+        proto_file
+    ]
+    
+    print(f"Generating Python code from {proto_file}...")
+    try:
+        subprocess.check_call(cmd)
+        output_file = os.path.join(output_dir, os.path.basename(os.path.splitext(proto_file)[0]) + "_pb2.py")
+        print(f"Successfully generated {output_file}")
+    except subprocess.CalledProcessError as e:
+        print(f"Error generating proto code: {e}")
+        sys.exit(1)
+
+if __name__ == "__main__":
+    generate_proto()
diff --git a/tests/test_telemetry.py b/tests/test_telemetry.py
