adding tags to config telemetry to capture uwsgi worker

Author: chillaq

splitio/api/telemetry.py

@@ -1,6 +1,5 @@
 """Impressions API module."""
 import logging
-import time
 
 from splitio.api import APIException
 from splitio.api.client import HttpClientException

splitio/client/factory.py

@@ -1,6 +1,7 @@
 """A module for Split.io Factories."""
 import logging
 import threading
+import sys
 from collections import Counter
 
 from enum import Enum
@@ -96,6 +97,7 @@ def __init__(  # pylint: disable=too-many-arguments
             sync_manager=None,
             sdk_ready_flag=None,
             telemetry_producer=None,
+            telemetry_init_producer=None,
             telemetry_submitter=None,
             preforked_initialization=False,
     ):
@@ -124,10 +126,8 @@ def __init__(  # pylint: disable=too-many-arguments
         self._sdk_internal_ready_flag = sdk_ready_flag
         self._recorder = recorder
         self._preforked_initialization = preforked_initialization
-        self._telemetry_evaluation_producer = None
-        self._telemetry_init_producer = None
         self._telemetry_evaluation_producer = telemetry_producer.get_telemetry_evaluation_producer()
-        self._telemetry_init_producer = telemetry_producer.get_telemetry_init_producer()
+        self._telemetry_init_producer = telemetry_init_producer
         self._telemetry_submitter = telemetry_submitter
         self._ready_time = get_current_epoch_time_ms()
         self._start_status_updater()
@@ -331,7 +331,7 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
     telemetry_consumer = TelemetryStorageConsumer(telemetry_storage)
     telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
     telemetry_evaluation_producer = telemetry_producer.get_telemetry_evaluation_producer()
-
+    telemetry_init_producer = telemetry_producer.get_telemetry_init_producer()
 
     http_client = HttpClient(
         sdk_url=sdk_url,
@@ -418,21 +418,25 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
         telemetry_evaluation_producer
     )
 
+    telemetry_init_producer.record_config(cfg, extra_cfg)
+    if int(_get_uwsgi_worker_id()) > -1:
+        telemetry_init_producer.add_config_tag("initilization:uwsgi")
+        telemetry_init_producer.add_config_tag("uwsgi_worker:#" + _get_uwsgi_worker_id())
+
     if preforked_initialization:
         synchronizer.sync_all(max_retry_attempts=_MAX_RETRY_SYNC_ALL)
         synchronizer._split_synchronizers._segment_sync.shutdown()
+
         return SplitFactory(api_key, storages, cfg['labelsEnabled'],
-                            recorder, manager, None, telemetry_producer, apis['telemetry'], preforked_initialization=preforked_initialization)
+                            recorder, manager, None, telemetry_producer, telemetry_init_producer, telemetry_submitter, preforked_initialization=preforked_initialization)
 
     initialization_thread = threading.Thread(target=manager.start, name="SDKInitializer")
     initialization_thread.setDaemon(True)
     initialization_thread.start()
 
-    telemetry_producer.get_telemetry_init_producer().record_config(cfg, extra_cfg)
-
     return SplitFactory(api_key, storages, cfg['labelsEnabled'],
                         recorder, manager, sdk_ready_flag,
-                        telemetry_producer,
+                        telemetry_producer, telemetry_init_producer,
                         telemetry_submitter)
 
 def _build_redis_factory(api_key, cfg):
@@ -450,6 +454,7 @@ def _build_redis_factory(api_key, cfg):
     }
     telemetry_producer = TelemetryStorageProducer(storages['telemetry'])
     telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
+    telemetry_init_producer = telemetry_producer.get_telemetry_init_producer()
     telemetry_submitter = RedisTelemetrySubmitter(storages['telemetry'])
 
     data_sampling = cfg.get('dataSampling', DEFAULT_DATA_SAMPLING)
@@ -497,7 +502,10 @@ def _build_redis_factory(api_key, cfg):
     initialization_thread.setDaemon(True)
     initialization_thread.start()
 
-    telemetry_producer.get_telemetry_init_producer().record_config(cfg, {})
+    telemetry_init_producer.record_config(cfg, {})
+    if int(_get_uwsgi_worker_id()) > -1:
+        telemetry_init_producer.add_config_tag("initilization:uwsgi")
+        telemetry_init_producer.add_config_tag("uwsgi_worker:#" + _get_uwsgi_worker_id())
 
     split_factory = SplitFactory(
         api_key,
@@ -507,6 +515,7 @@ def _build_redis_factory(api_key, cfg):
         manager,
         sdk_ready_flag=None,
         telemetry_producer=telemetry_producer,
+        telemetry_init_producer=telemetry_init_producer
     )
     redundant_factory_count, active_factory_count = _get_active_and_redundant_count()
     storages['telemetry'].record_active_and_redundant_factories(active_factory_count, redundant_factory_count)
@@ -560,6 +569,7 @@ def _build_localhost_factory(cfg):
         manager,
         ready_event,
         telemetry_producer=telemetry_producer,
+        telemetry_init_producer=telemetry_producer.get_telemetry_init_producer(),
         telemetry_submitter=LocalhostTelemetrySubmitter(),
     )
 
@@ -613,3 +623,12 @@ def _get_active_and_redundant_count():
         active_factory_count += _INSTANTIATED_FACTORIES[item]
     _INSTANTIATED_FACTORIES_LOCK.release()
     return redundant_factory_count, active_factory_count
+
+def _get_uwsgi_worker_id():
+    try:
+        import uwsgi
+        _LOGGER.debug("uwsgi lib detected")
+        return str(uwsgi.worker_id())
+    except ModuleNotFoundError:
+        pass
+        return "-1"

splitio/engine/telemetry.py

@@ -51,6 +51,10 @@ def record_not_ready_usage(self):
     def record_active_and_redundant_factories(self, active_factory_count, redundant_factory_count):
         self._telemetry_storage.record_active_and_redundant_factories(active_factory_count, redundant_factory_count)
 
+    def add_config_tag(self, tag):
+        """Record tag string."""
+        self._telemetry_storage.add_config_tag(tag)
+
 class TelemetryEvaluationProducer(object):
     """Telemetry evaluation producer class."""
 
@@ -150,12 +154,19 @@ def get_not_ready_usage(self):
         return self._telemetry_storage.get_not_ready_usage()
 
     def get_config_stats(self):
-        """Get none-ready usage."""
-        return self._telemetry_storage.get_config_stats()
+        """Get config stats."""
+        config_stats = self._telemetry_storage.get_config_stats()
+        config_stats.update({'t': self.pop_config_tags()})
+        return config_stats
 
     def get_config_stats_to_json(self):
+        """Get config stats in json."""
         return json.dumps(self._telemetry_storage.get_config_stats())
 
+    def pop_config_tags(self):
+        """Get and reset tags."""
+        return self._telemetry_storage.pop_config_tags()
+
 class TelemetryEvaluationConsumer(object):
     """Telemetry evaluation consumer class."""
 
@@ -215,7 +226,7 @@ def get_last_synchronization(self):
         return self._telemetry_storage.get_last_synchronization()['lastSynchronizations']
 
     def pop_tags(self):
-        """Get and reset http errors."""
+        """Get and reset tags."""
         return self._telemetry_storage.pop_tags()
 
     def pop_http_errors(self):

splitio/storage/inmemmory.py

@@ -469,6 +469,7 @@ def __init__(self):
         """Constructor"""
         self._lock = threading.RLock()
         self._reset_tags()
+        self._reset_config_tags()
         self._method_exceptions = MethodExceptions()
         self._last_synchronization = LastSynchronization()
         self._counters = TelemetryCounters()
@@ -482,6 +483,10 @@ def _reset_tags(self):
         with self._lock:
             self._tags = []
 
+    def _reset_config_tags(self):
+        with self._lock:
+            self._config_tags = []
+
     def record_config(self, config, extra_config):
         """Record configurations."""
         self._tel_config.record_config(config, extra_config)
@@ -500,6 +505,12 @@ def add_tag(self, tag):
             if len(self._tags) < MAX_TAGS:
                 self._tags.append(tag)
 
+    def add_config_tag(self, tag):
+        """Record tag string."""
+        with self._lock:
+            if len(self._config_tags) < MAX_TAGS:
+                self._config_tags.append(tag)
+
     def record_bur_time_out(self):
         """Record block until ready timeout."""
         self._tel_config.record_bur_time_out()
@@ -575,6 +586,13 @@ def pop_tags(self):
             self._reset_tags()
             return tags
 
+    def pop_config_tags(self):
+        """Get and reset tags."""
+        with self._lock:
+            tags = self._config_tags
+            self._reset_config_tags()
+            return tags
+
     def pop_latencies(self):
         """Get and reset eval latencies."""
         return self._method_latencies.pop_all()

splitio/storage/redis.py

@@ -1,6 +1,7 @@
 """Redis storage module."""
 import json
 import logging
+import threading
 
 from splitio.models.impressions import Impression
 from splitio.models import splits, segments
@@ -12,7 +13,7 @@
 
 
 _LOGGER = logging.getLogger(__name__)
-
+MAX_TAGS = 10
 
 class RedisSplitStorage(SplitStorage):
     """Redis-based storage for splits."""
@@ -594,13 +595,25 @@ def __init__(self, redis_client, sdk_metadata):
         :param sdk_metadata: SDK & Machine information.
         :type sdk_metadata: splitio.client.util.SdkMetadata
         """
+        self._lock = threading.RLock()
+        self._reset_config_tags()
         self._redis_client = redis_client
         self._sdk_metadata = sdk_metadata
         self._method_latencies = MethodLatencies()
         self._method_exceptions = MethodExceptions()
         self._tel_config = TelemetryConfig()
         self._make_pipe = redis_client.pipeline
 
+    def _reset_config_tags(self):
+        with self._lock:
+            self._config_tags = []
+
+    def add_config_tag(self, tag):
+        """Record tag string."""
+        with self._lock:
+            if len(self._config_tags) < MAX_TAGS:
+                self._config_tags.append(tag)
+
     def record_config(self, config, extra_config):
         """
         initilize telemetry objects
@@ -610,6 +623,13 @@ def record_config(self, config, extra_config):
         """
         self._tel_config.record_config(config, extra_config)
 
+    def pop_config_tags(self):
+        """Get and reset tags."""
+        with self._lock:
+            tags = self._config_tags
+            self._reset_config_tags()
+            return tags
+
     def push_config_stats(self):
         """push config stats to redis."""
         self._redis_client.hset(self._TELEMETRY_CONFIG_KEY, self._sdk_metadata.sdk_version + '/' + self._sdk_metadata.instance_name + '/' + self._sdk_metadata.instance_ip, str(self._format_config_stats()))
@@ -621,7 +641,8 @@ def _format_config_stats(self):
             'aF': config_stats['aF'],
             'rF': config_stats['rF'],
             'sT': config_stats['sT'],
-            'oM': config_stats['oM']
+            'oM': config_stats['oM'],
+            't': self.pop_config_tags()
         })
 
     def record_active_and_redundant_factories(self, active_factory_count, redundant_factory_count):