error handling update (#389)

- added sdk_error_callback option to listen for unexpected exceptions
caught
- new metrics *sdk_exceptions_count* + *events_successfully_sent*
- non retryable events will now be considered a *log_event_failed*
exception
- stricter typing on statsig user to only allow JSON serializable type
- copy more options into logging copy
- unit tests for new telemetry logger functionalities

Author: kat-statsig

statsig/globals.py

@@ -18,4 +18,6 @@ def init_logger(options: StatsigOptions):
         logger.set_log_level(options.output_logger_level)
     if options.observability_client is not None:
         logger.set_ob_client(options.observability_client)
+    if options.sdk_error_callback is not None:
+        logger.set_sdk_error_callback(options.sdk_error_callback)
     logger.init()

statsig/http_worker.py

@@ -2,19 +2,18 @@
 import json
 import time
 from concurrent.futures.thread import ThreadPoolExecutor
-from dataclasses import dataclass
 from decimal import Decimal
 from io import BytesIO
-from typing import Callable, Tuple, Optional, Any, Dict, Union
+from typing import Callable, Tuple, Optional, Any, Dict
 
 import ijson
 import requests
-from requests.structures import CaseInsensitiveDict
 
 from . import globals
 from .diagnostics import Diagnostics, Marker
 from .evaluation_details import DataSource
 from .interface_network import IStatsigNetworkWorker, NetworkProtocol, NetworkEndpoint
+from .request_result import RequestResult
 from .sdk_configs import _SDK_Configs
 from .statsig_context import InitContext
 from .statsig_error_boundary import _StatsigErrorBoundary
@@ -23,16 +22,6 @@
 REQUEST_TIMEOUT = 20
 
 
-@dataclass
-class RequestResult:
-    data: Optional[Dict[str, Any]]
-    success: bool
-    status_code: Optional[int]
-    text: Optional[str] = None
-    headers: Optional[Union[CaseInsensitiveDict, Dict[str, str]]] = None
-    error: Optional[Exception] = None
-
-
 class HttpWorker(IStatsigNetworkWorker):
     _raise_on_error = False
     __RETRY_CODES = [408, 500, 502, 503, 504, 522, 524, 599]
@@ -114,7 +103,7 @@ def get_id_list(self, on_complete, url, headers, log_on_exception=False):
             return on_complete(resp)
         return on_complete(None)
 
-    def log_events(self, payload, headers=None, log_on_exception=False, retry=0):
+    def log_events(self, payload, headers=None, log_on_exception=False, retry=0) -> RequestResult:
         disable_compression = _SDK_Configs.on("stop_log_event_compression")
         additional_headers = {
             'STATSIG-RETRY': str(retry),
@@ -127,9 +116,9 @@ def log_events(self, payload, headers=None, log_on_exception=False, retry=0):
             headers=additional_headers,
             payload=payload, log_on_exception=log_on_exception, init_timeout=None, zipped=not disable_compression,
             tag="log_event")
-        if response is None or response.status_code in self.__RETRY_CODES:
-            return payload
-        return None
+        if response.status_code in self.__RETRY_CODES:
+            response.retryable = True
+        return response
 
     def shutdown(self) -> None:
         self._executor.shutdown(wait=False)

statsig/request_result.py

@@ -0,0 +1,15 @@
+from dataclasses import dataclass
+from typing import Optional, Any, Dict, Union
+
+from requests.structures import CaseInsensitiveDict
+
+
+@dataclass
+class RequestResult:
+    data: Optional[Dict[str, Any]]
+    success: bool
+    status_code: Optional[int]
+    text: Optional[str] = None
+    headers: Optional[Union[CaseInsensitiveDict, Dict[str, str]]] = None
+    error: Optional[Exception] = None
+    retryable: bool = False

statsig/statsig_error_boundary.py

@@ -106,6 +106,7 @@ def log_exception(
                 return
             self._seen.add(name)
 
+            globals.logger.log_sdk_exception(name, exception)
             self._executor.submit(
                 self._post_exception,
                 name,

statsig/statsig_logger_worker.py

@@ -118,24 +118,30 @@ def _send_and_reset_dropped_events_count(self):
     def _flush_to_server(self, batched_events: BatchEventLogs):
         if self._local_mode:
             return
-        res = self._net.log_events(batched_events.payload, retry=batched_events.retries,
-                                   log_on_exception=True, headers=batched_events.headers)
-        if res is not None:
-            if batched_events.retries >= 10:
-                message = (
-                    f"Failed to post {batched_events.event_count} logs after 10 retries, dropping the request"
-                )
-                self._error_boundary.log_exception(
-                    "statsig::log_event_failed",
-                    Exception(message),
-                    {"eventCount": batched_events.event_count, "error": message},
-                    bypass_dedupe=True
-                )
-                globals.logger.warning(message)
-                return
+        result = self._net.log_events(batched_events.payload, retry=batched_events.retries,
+                                      log_on_exception=True, headers=batched_events.headers)
 
-            self._failure_backoff()
+        if result.success:
+            globals.logger.increment("events_successfully_sent_count", batched_events.event_count)
+            self._success_backoff()
+            return
 
+        if batched_events.retries >= 10 or not result.retryable:
+            message = (
+                f"Failed to post {batched_events.event_count} logs. The request was either not retryable or failed after 10 retries, and has been dropped."
+            )
+            self._error_boundary.log_exception(
+                "statsig::log_event_failed",
+                Exception(message),
+                {"eventCount": batched_events.event_count, "error": message},
+                bypass_dedupe=True
+            )
+            globals.logger.warning(message)
+            return
+
+        self._failure_backoff()
+
+        if result.retryable:
             self.event_batch_processor.add_to_batched_events_queue(
                 BatchEventLogs(
                     batched_events.payload,
@@ -144,8 +150,6 @@ def _flush_to_server(self, batched_events: BatchEventLogs):
                     batched_events.retries + 1,
                 )
             )
-        else:
-            self._success_backoff()
 
     def _get_curr_interval(self):
         with self.lock:

statsig/statsig_network.py

@@ -4,7 +4,7 @@
 
 from . import globals
 from .diagnostics import Diagnostics
-from .http_worker import HttpWorker
+from .http_worker import HttpWorker, RequestResult
 from .interface_network import (
     IStreamingFallback,
     IStreamingListeners,
@@ -221,10 +221,10 @@ def get_id_list(self, on_complete: Any, url, headers, log_on_exception=False):
             return
         self.http_worker.get_id_list(on_complete, url, headers, log_on_exception)
 
-    def log_events(self, payload, headers=None, log_on_exception=False, retry=0):
+    def log_events(self, payload, headers=None, log_on_exception=False, retry=0) -> RequestResult:
         if self.options.local_mode:
             globals.logger.warning("Local mode is enabled. Not logging events.")
-            return None
+            return RequestResult(data=None, status_code=None, success=False, error=None)
         return self.log_event_worker.log_events(
             payload, headers=headers, log_on_exception=log_on_exception, retry=retry
         )

statsig/statsig_options.py

@@ -110,6 +110,7 @@ def __init__(
             output_logger_level: Optional[LogLevel] = LogLevel.WARNING,
             overall_init_timeout: Optional[float] = None,
             observability_client: Optional[ObservabilityClient] = None,
+            sdk_error_callback: Optional[Callable[[str, Exception], None]] = None,
     ):
         self.data_store = data_store
         self._environment: Union[None, dict] = None
@@ -157,6 +158,7 @@ def __init__(
         self.output_logger_level = output_logger_level
         self.overall_init_timeout = overall_init_timeout
         self.observability_client = observability_client
+        self.sdk_error_callback = sdk_error_callback
         self._logging_copy: Dict[str, Any] = {}
         self._set_logging_copy()
         self._attributes_changed = False
@@ -203,7 +205,7 @@ def _set_logging_copy(self):
             logging_copy["init_timeout"] = self.init_timeout
         if self.timeout:
             logging_copy["timeout"] = self.timeout
-        if (self.rulesets_sync_interval) != DEFAULT_RULESET_SYNC_INTERVAL:
+        if self.rulesets_sync_interval != DEFAULT_RULESET_SYNC_INTERVAL:
             logging_copy["rulesets_sync_interval"] = self.rulesets_sync_interval
         if self.idlists_sync_interval != DEFAULT_IDLIST_SYNC_INTERVAL:
             logging_copy["idlists_sync_interval"] = self.idlists_sync_interval
@@ -219,6 +221,8 @@ def _set_logging_copy(self):
             logging_copy["logging_interval"] = self.logging_interval
         if self.disable_diagnostics:
             logging_copy["disable_diagnostics"] = self.disable_diagnostics
+        if self.disable_all_logging:
+            logging_copy["disable_all_logging"] = self.disable_all_logging
         if self.event_queue_size != DEFAULT_EVENT_QUEUE_SIZE:
             logging_copy["event_queue_size"] = self.event_queue_size
         if self.retry_queue_size != DEFAULT_RETRY_QUEUE_SIZE:
@@ -227,5 +231,15 @@ def _set_logging_copy(self):
             logging_copy["overall_init_timeout"] = self.overall_init_timeout
         if self.observability_client is not None:
             logging_copy["observability_client"] = "SET"
+        if self.fallback_to_statsig_api:
+            logging_copy["fallback_to_statsig_api"] = self.fallback_to_statsig_api
+        if self.out_of_sync_threshold_in_s:
+            logging_copy["out_of_sync_threshold_in_s"] = self.out_of_sync_threshold_in_s
+        if self.initialize_sources:
+            logging_copy["initialize_sources"] = "SET"
+        if self.config_sync_sources:
+            logging_copy["config_sync_sources"] = "SET"
+        if self.sdk_error_callback:
+            logging_copy["sdk_error_callback"] = "SET"
         self._logging_copy = logging_copy
         self._attributes_changed = False

statsig/statsig_telemetry_logger.py

@@ -1,6 +1,6 @@
 import functools
 import time
-from typing import Optional, Dict, Any
+from typing import Optional, Dict, Any, Callable
 
 from .initialize_details import InitializeDetails
 from .interface_observability_client import ObservabilityClient
@@ -49,17 +49,22 @@ def __init_subclass__(cls, **kwargs):
 
 
 class StatsigTelemetryLogger(AutoTryCatch):
-    def __init__(self, logger=None, ob_client: Optional[ObservabilityClient] = None):
+    def __init__(self, logger=None, ob_client: Optional[ObservabilityClient] = None,
+                 sdk_error_callback: Optional[Callable[[str, Exception], None]] = None):
         self.high_cardinality_tags = {"lcut", "prev_lcut"}
         self.logger = logger or OutputLogger(TELEMETRY_PREFIX)
         self.ob_client = ob_client or NoopObservabilityClient()
+        self.sdk_error_callback = sdk_error_callback
 
     def set_logger(self, output_logger):
         self.logger = output_logger
 
     def set_ob_client(self, ob_client):
         self.ob_client = ob_client
 
+    def set_sdk_error_callback(self, sdk_error_callback):
+        self.sdk_error_callback = sdk_error_callback
+
     def init(self):
         self.ob_client.init()
 
@@ -84,6 +89,17 @@ def exception(self, msg, *args, **kwargs):
     def log_process(self, process, msg):
         self.logger.log_process(process, msg)
 
+    def increment(self, metric_name: str, value: int = 1, tags: Optional[Dict[str, Any]] = None):
+        self.ob_client.increment(f'{TELEMETRY_PREFIX}.{metric_name}', value,
+                                 self.filter_high_cardinality_tags(tags or {}))
+
+    def gauge(self, metric_name: str, value: float, tags: Optional[Dict[str, Any]] = None):
+        self.ob_client.gauge(f'{TELEMETRY_PREFIX}.{metric_name}', value, self.filter_high_cardinality_tags(tags or {}))
+
+    def distribution(self, metric_name: str, value: float, tags: Optional[Dict[str, Any]] = None):
+        self.ob_client.distribution(f'{TELEMETRY_PREFIX}.{metric_name}', value,
+                                    self.filter_high_cardinality_tags(tags or {}))
+
     def log_post_init(self, options: StatsigOptions, init_details: InitializeDetails):
         if options.local_mode:
             if init_details.init_success:
@@ -93,11 +109,11 @@ def log_post_init(self, options: StatsigOptions, init_details: InitializeDetails
                 self.logger.error("Statsig SDK instance failed to initialize in local mode.")
             return
 
-        self.ob_client.distribution(f"{TELEMETRY_PREFIX}.initialization", init_details.duration,
-                                    self.filter_high_cardinality_tags({"source": init_details.source,
-                                                                       "store_populated": init_details.store_populated,
-                                                                       "init_success": init_details.init_success,
-                                                                       "init_source_api": init_details.init_source_api}))
+        self.distribution("initialization", init_details.duration,
+                          self.filter_high_cardinality_tags({"source": init_details.source,
+                                                             "store_populated": init_details.store_populated,
+                                                             "init_success": init_details.init_success,
+                                                             "init_source_api": init_details.init_source_api}))
 
         if init_details.init_success:
             if init_details.store_populated:
@@ -120,20 +136,26 @@ def log_config_sync_update(self, initialized: bool, has_update: bool, lcut: int,
             return  # do not log for initialize
         if not has_update:
             self.log_process("Config Sync", "No update")
-            self.ob_client.increment(f"{TELEMETRY_PREFIX}.config_no_update", 1, {"source": source, "source_api": api})
+            self.increment("config_no_update", 1, {"source": source, "source_api": api})
             return
 
         lcut_diff = abs(lcut - int(time.time() * 1000))
         if lcut_diff > 0:
-            self.ob_client.distribution(f"{TELEMETRY_PREFIX}.config_propagation_diff", lcut_diff,
-                                        self.filter_high_cardinality_tags({
-                                            "source": source,
-                                            "source_api": api,
-                                            "lcut": lcut,
-                                            "prev_lcut": prev_lcut
-                                        }))
+            self.distribution("config_propagation_diff", lcut_diff,
+                              self.filter_high_cardinality_tags({
+                                  "source": source,
+                                  "source_api": api,
+                                  "lcut": lcut,
+                                  "prev_lcut": prev_lcut
+                              }))
         self.log_process("Config Sync", f"Received updated configs from {lcut}")
 
+    def log_sdk_exception(self, tag: str, exception: Exception):
+        if self.sdk_error_callback is not None:
+            self.sdk_error_callback(tag, exception)
+
+        self.increment("sdk_exceptions_count")
+
     def filter_high_cardinality_tags(self, tags: Dict[str, Any]) -> Dict[str, Any]:
         return {tag: value for tag, value in tags.items()
                 if tag not in self.high_cardinality_tags or self.ob_client.should_enable_high_cardinality_for_this_tag(

statsig/statsig_user.py

@@ -1,10 +1,13 @@
 from dataclasses import dataclass
-from typing import Optional
+from typing import Optional, Union, List, Dict
 
 from .statsig_environment_tier import StatsigEnvironmentTier
 from .statsig_errors import StatsigValueError
 from .utils import str_or_none, to_raw_dict_or_none, djb2_hash_for_dict
 
+JSONPrimitive = Union[str, int, float, bool, None]
+JSONValue = Union[JSONPrimitive, List['JSONValue'], Dict[str, 'JSONValue']]
+
 
 @dataclass
 class StatsigUser:
@@ -21,10 +24,10 @@ class StatsigUser:
     country: Optional[str] = None
     locale: Optional[str] = None
     app_version: Optional[str] = None
-    custom: Optional[dict] = None
-    private_attributes: Optional[dict] = None
-    custom_ids: Optional[dict] = None
-    _statsig_environment: Optional[dict] = None
+    custom: Optional[Dict[str, JSONValue]] = None
+    private_attributes: Optional[Dict[str, JSONValue]] = None
+    custom_ids: Optional[Dict[str, str]] = None
+    _statsig_environment: Optional[Dict[str, JSONValue]] = None
 
     def __post_init__(self):
         # ensure there is a user id or at least a custom ID, empty dict