observability integration (#369)

kat-statsig · web-flow · commit 59edbd8ef4bc · 2024-11-11T11:25:07.000-08:00
- allow user to plug in observability client (statsd/opentelemetry) - logger is now a telemetry logger sitting on top of obsv client + output logger. - https://www.notion.so/statsig/SDK-Observability-5c48660fa03c432f9d15e18b3d5c3600
diff --git a/statsig/globals.py b/statsig/globals.py
@@ -1,19 +1,21 @@
 import os
 
-from .output_logger import OutputLogger, LogLevel
+from .statsig_options import StatsigOptions
+from .statsig_telemetry_logger import StatsigTelemetryLogger
 
 STATSIG_BATCHING_INTERVAL_SECONDS = 60.0
 STATSIG_LOGGING_INTERVAL_SECONDS = 1.0
 
-logger = OutputLogger('statsig.sdk')
+logger = StatsigTelemetryLogger()
 
 os.environ["GRPC_VERBOSITY"] = "NONE"
 
 
-def set_logger(output_logger):
-    global logger
-    logger = output_logger
-
-
-def set_log_level(log_level: LogLevel):
-    logger.set_log_level(log_level)
+def init_logger(options: StatsigOptions):
+    if options.custom_logger is not None:
+        logger.set_logger(options.custom_logger)
+    elif options.output_logger_level is not None:
+        logger.set_log_level(options.output_logger_level)
+    if options.observability_client is not None:
+        logger.set_ob_client(options.observability_client)
+    logger.init()
diff --git a/statsig/grpc_websocket_worker.py b/statsig/grpc_websocket_worker.py
@@ -206,7 +206,7 @@ def get_dcs(
             log_on_exception: Optional[bool] = False,
             init_timeout: Optional[int] = None,
     ):
-        self.context.dcs_api = self.proxy_config.proxy_address
+        self.context.source_api = self.proxy_config.proxy_address
         self._diagnostics.add_marker(
             Marker().download_config_specs().network_request().start()
         )
diff --git a/statsig/grpc_worker.py b/statsig/grpc_worker.py
@@ -30,7 +30,7 @@ def get_dcs(self, on_complete: Callable, since_time: int = 0, log_on_exception:
                 init_timeout: Optional[int] = None):
         request = ConfigSpecRequest(sdkKey=self.sdk_key, sinceTime=since_time)
         try:
-            self.context.dcs_api = self.proxy_config.proxy_address
+            self.context.source_api = self.proxy_config.proxy_address
             response = self.stub.getConfigSpec(request)
             on_complete(DataSource.NETWORK, response.spec, None)
         except Exception as e:
diff --git a/statsig/http_worker.py b/statsig/http_worker.py
@@ -51,7 +51,7 @@ def get_dcs(self, on_complete: Callable, since_time=0, log_on_exception=False, i
             url=f"{self.__api_for_download_config_specs}download_config_specs/{self.__sdk_key}.json?sinceTime={since_time}",
             headers=None, init_timeout=init_timeout, log_on_exception=log_on_exception,
             tag="download_config_specs")
-        self._context.dcs_api = self.__api_for_download_config_specs
+        self._context.source_api = self.__api_for_download_config_specs
         if response is not None and self._is_success_code(response.status_code):
             on_complete(DataSource.NETWORK, response.json() or {}, None)
             return
@@ -62,7 +62,7 @@ def get_dcs_fallback(self, on_complete: Callable, since_time=0, log_on_exception
             url=f"{STATSIG_CDN}download_config_specs/{self.__sdk_key}.json?sinceTime={since_time}",
             headers=None, init_timeout=init_timeout, log_on_exception=log_on_exception,
             tag="download_config_specs")
-        self._context.dcs_api = STATSIG_CDN
+        self._context.source_api = STATSIG_CDN
         if response is not None and self._is_success_code(response.status_code):
             on_complete(DataSource.STATSIG_NETWORK, response.json() or {}, None)
             return
diff --git a/statsig/initialize_details.py b/statsig/initialize_details.py
@@ -0,0 +1,21 @@
+from typing import Optional
+
+
+class InitializeDetails:
+    duration: int
+    source: str
+    init_success: bool
+    store_populated: bool
+    error: Optional[Exception]
+    timed_out: bool
+    init_source_api: Optional[str]
+
+    def __init__(self, duration: int, source: str, init_success: bool, store_populated: bool,
+                 error: Optional[Exception], init_source_api: Optional[str] = None, timed_out: bool = False):
+        self.duration = duration
+        self.source = source
+        self.init_success = init_success
+        self.error = error
+        self.store_populated = store_populated
+        self.init_source_api = init_source_api
+        self.timed_out = timed_out
diff --git a/statsig/interface_observability_client.py b/statsig/interface_observability_client.py
@@ -0,0 +1,57 @@
+from typing import Dict, Optional, Any
+
+
+# pylint: disable=unused-argument
+class ObservabilityClient:
+    """
+    An interface for observability clients that allows users to plug in their
+    own observability integration for metrics collection and monitoring.
+    """
+
+    def init(self, *args, **kwargs):
+        """
+        Initializes the observability client with necessary configuration.
+
+        :param args: Positional arguments for initialization.
+        :param kwargs: Keyword arguments for initialization.
+        """
+
+    def increment(self, metric_name: str, value: int = 1, tags: Optional[Dict[str, Any]] = None) -> None:
+        """
+        Increment a counter metric.
+
+        :param metric_name: The name of the metric to increment.
+        :param value: The value by which the counter should be incremented (default is 1).
+        :param tags: Optional dictionary of tags for metric dimensions.
+        """
+
+    def gauge(self, metric_name: str, value: float, tags: Optional[Dict[str, Any]] = None) -> None:
+        """
+        Set a gauge metric.
+
+        :param metric_name: The name of the metric to set.
+        :param value: The value to set the gauge to.
+        :param tags: Optional dictionary of tags for metric dimensions.
+        """
+
+    def distribution(self, metric_name: str, value: float, tags: Optional[Dict[str, Any]] = None) -> None:
+        """
+        Record a distribution metric for tracking statistical data.
+
+        :param metric_name: The name of the metric to record.
+        :param value: The recorded value for the distribution metric.
+        :param tags: Optional dictionary of tags that represent dimensions to associate with the metric.
+        """
+
+    def should_enable_high_cardinality_for_this_tag(self, tag: str) -> bool:
+        """
+        Determine if a high cardinality tag should be logged
+
+        :param tag: The tag to check for high cardinality enabled.
+        """
+        return False
+
+    def shutdown(self) -> None:
+        """
+        Shutdown the observability client.
+        """
diff --git a/statsig/spec_store.py b/statsig/spec_store.py
@@ -58,6 +58,7 @@ def __init__(
 
         self._id_lists: Dict[str, dict] = {}
         self.unsupported_configs: Set[str] = set()
+        self.context = context
 
         self.spec_updater = SpecUpdater(
             network,
@@ -152,8 +153,13 @@ def _initialize_specs(self):
 
     def _process_specs(self, specs_json, source: DataSource) -> Tuple[bool, bool]:  # has update, parse success
         self._log_process("Processing specs...")
+        prev_lcut = self.spec_updater.last_update_time
         if specs_json.get("has_updates", False) is False:
-            globals.logger.debug("Received update: %s", "No Update")
+            globals.logger.log_config_sync_update(self.spec_updater.initialized, False,
+                                                  self.spec_updater.last_update_time,
+                                                  prev_lcut,
+                                                  self.context.source,
+                                                  self.context.source_api)
             return False, True
         if not self.spec_updater.is_specs_json_valid(specs_json):
             self._log_process("Failed to process specs")
@@ -236,9 +242,14 @@ def parse_target_value_map_from_spec(spec, parsed):
         self._experiment_to_layer = new_experiment_to_layer
         self.spec_updater.last_update_time = specs_json.get("time", 0)
         self.init_source = source
+        self.context.source = source
         self._default_environment = specs_json.get("default_environment", None)
-        globals.logger.debug("Received update: %s", self.spec_updater.last_update_time)
-
+        if self.spec_updater.last_update_time > prev_lcut:
+            globals.logger.log_config_sync_update(self.spec_updater.initialized, True,
+                                                  self.spec_updater.last_update_time,
+                                                  prev_lcut,
+                                                  self.context.source,
+                                                  self.context.source_api)
         flags = specs_json.get("sdk_flags", {})
         _SDK_Configs.set_flags(flags)
         configs = specs_json.get("sdk_configs", {})
diff --git a/statsig/spec_updater.py b/statsig/spec_updater.py
@@ -384,23 +384,23 @@ def sync_config_spec():
                     globals.logger.log_process(
                         "Config Sync",
                         f"Syncing config values with {strategy.value}"
-                        + (f"[{self.context.dcs_api}]" if self.context.dcs_api else "")
+                        + (f"[{self.context.source_api}]" if self.context.source_api else "")
                         + " successful"
                     )
                     break
                 if i < len(self._config_sync_strategies) - 1:
                     globals.logger.log_process(
                         "Config Sync",
                         f"Syncing config values failed with {strategy.value}"
-                        + (f"[{self.context.dcs_api}]" if self.context.dcs_api else "")
+                        + (f"[{self.context.source_api}]" if self.context.source_api else "")
                         + ", falling back to next available configured config sync method"
                     )
 
                 else:
                     globals.logger.log_process(
                         "Config Sync",
                         f"Syncing config values failed with {strategy.value}"
-                        + (f"[{self.context.dcs_api}]" if self.context.dcs_api else "")
+                        + (f"[{self.context.source_api}]" if self.context.source_api else "")
                         + f". No more strategies left. The next sync will be in {interval} seconds."
                     )
 
diff --git a/statsig/statsig.py b/statsig/statsig.py
@@ -1,12 +1,13 @@
 from typing import Optional
 
-from . import globals, FeatureGate
+from . import FeatureGate
 from .client_initialize_formatter import ClientInitializeResponse
 from .dynamic_config import DynamicConfig
+from .initialize_details import InitializeDetails
 from .layer import Layer
 from .statsig_event import StatsigEvent
 from .statsig_options import StatsigOptions
-from .statsig_server import StatsigServer, InitializeDetails
+from .statsig_server import StatsigServer
 from .statsig_user import StatsigUser
 from .utils import HashingAlgorithm
 
@@ -21,14 +22,6 @@ def initialize(secret_key: str, options: Optional[StatsigOptions] = None) -> Ini
     :param options: The StatsigOptions object used to configure the SDK
     :return: The initialization details
     """
-    if options is None:
-        options = StatsigOptions()
-
-    if options.custom_logger is not None:
-        globals.set_logger(options.custom_logger)
-    elif options.output_logger_level is not None:
-        globals.set_log_level(options.output_logger_level)
-
     return __instance.initialize(secret_key, options)
 
 
diff --git a/statsig/statsig_context.py b/statsig/statsig_context.py
@@ -10,7 +10,7 @@ class InitContext:
     error: Optional[Exception]
     source: DataSource
     store_populated: bool
-    dcs_api: Optional[str]
+    source_api: Optional[str]
     timed_out: bool
 
     def __init__(self):
@@ -19,5 +19,5 @@ def __init__(self):
         self.error = None
         self.source = DataSource.UNINITIALIZED
         self.store_populated = False
-        self.dcs_api = None
+        self.source_api = None
         self.timed_out = False
diff --git a/statsig/statsig_options.py b/statsig/statsig_options.py
@@ -6,6 +6,7 @@
 from .feature_gate import FeatureGate
 from .interface_data_store import IDataStore
 from .interface_network import NetworkProtocol, NetworkEndpoint
+from .interface_observability_client import ObservabilityClient
 from .layer import Layer
 from .output_logger import OutputLogger, LogLevel
 from .statsig_environment_tier import StatsigEnvironmentTier
@@ -108,6 +109,7 @@ def __init__(
             config_sync_sources: Optional[List[DataSource]] = None,
             output_logger_level: Optional[LogLevel] = LogLevel.WARNING,
             overall_init_timeout: Optional[float] = None,
+            observability_client: Optional[ObservabilityClient] = None,
     ):
         self.data_store = data_store
         self._environment: Union[None, dict] = None
@@ -154,6 +156,7 @@ def __init__(
         self.config_sync_sources = config_sync_sources
         self.output_logger_level = output_logger_level
         self.overall_init_timeout = overall_init_timeout
+        self.observability_client = observability_client
         self._logging_copy: Dict[str, Any] = {}
         self._set_logging_copy()
         self._attributes_changed = False
@@ -222,5 +225,7 @@ def _set_logging_copy(self):
             logging_copy["retry_queue_size"] = self.retry_queue_size
         if self.overall_init_timeout is not None:
             logging_copy["overall_init_timeout"] = self.overall_init_timeout
+        if self.observability_client is not None:
+            logging_copy["observability_client"] = "SET"
         self._logging_copy = logging_copy
         self._attributes_changed = False
diff --git a/statsig/statsig_server.py b/statsig/statsig_server.py
diff --git a/statsig/statsig_telemetry_logger.py b/statsig/statsig_telemetry_logger.py
diff --git a/tests/test_api_overrides.py b/tests/test_api_overrides.py
diff --git a/tests/test_telemetry_logger.py b/tests/test_telemetry_logger.py

Original file line number	Diff line number	Diff line change
`@@ -206,7 +206,7 @@ def get_dcs(`
`206`	`206`	`log_on_exception: Optional[bool] = False,`
`207`	`207`	`init_timeout: Optional[int] = None,`
`208`	`208`	`):`
`209`		`- self.context.dcs_api = self.proxy_config.proxy_address`
	`209`	`+ self.context.source_api = self.proxy_config.proxy_address`
`210`	`210`	`self._diagnostics.add_marker(`
`211`	`211`	`Marker().download_config_specs().network_request().start()`
`212`	`212`	`)`