Merge pull request #504 from splitio/async-sse-timeouts

added SSE total and socket read timeouts

Author: chillaq

splitio/client/client.py

@@ -229,7 +229,7 @@ def get_treatment(self, key, feature_flag_name, attributes=None):
             return treatment
 
         except:
-            # TODO: maybe log here?
+            _LOGGER.error('get_treatment failed')
             return CONTROL
 
     def get_treatment_with_config(self, key, feature_flag_name, attributes=None):
@@ -252,7 +252,7 @@ def get_treatment_with_config(self, key, feature_flag_name, attributes=None):
             return self._get_treatment(MethodExceptionsAndLatencies.TREATMENT_WITH_CONFIG, key, feature_flag_name, attributes)
 
         except Exception:
-            # TODO: maybe log here?
+            _LOGGER.error('get_treatment_with_config failed')
             return CONTROL, None
 
     def _get_treatment(self, method, key, feature, attributes=None):
@@ -289,7 +289,7 @@ def _get_treatment(self, method, key, feature, attributes=None):
                 ctx = self._context_factory.context_for(key, [feature])
                 input_validator.validate_feature_flag_names({feature: ctx.flags.get(feature)}, 'get_' + method.value)
                 result = self._evaluator.eval_with_context(key, bucketing, feature, attributes, ctx)
-            except Exception as e: # toto narrow this
+            except RuntimeError as e:
                 _LOGGER.error('Error getting treatment for feature flag')
                 _LOGGER.debug('Error: ', exc_info=True)
                 self._telemetry_evaluation_producer.record_exception(method)
@@ -565,7 +565,7 @@ def _get_treatments(self, key, features, method, attributes=None):
                 ctx = self._context_factory.context_for(key, features)
                 input_validator.validate_feature_flag_names({feature: ctx.flags.get(feature) for feature in features}, 'get_' + method.value)
                 results = self._evaluator.eval_many_with_context(key, bucketing, features, attributes, ctx)
-            except Exception as e: # toto narrow this
+            except RuntimeError as e:
                 _LOGGER.error('Error getting treatment for feature flag')
                 _LOGGER.debug('Error: ', exc_info=True)
                 self._telemetry_evaluation_producer.record_exception(method)
@@ -696,7 +696,7 @@ async def get_treatment(self, key, feature_flag_name, attributes=None):
             return treatment
 
         except:
-            # TODO: maybe log here?
+            _LOGGER.error('get_treatment failed')
             return CONTROL
 
     async def get_treatment_with_config(self, key, feature_flag_name, attributes=None):
@@ -719,7 +719,7 @@ async def get_treatment_with_config(self, key, feature_flag_name, attributes=Non
             return await self._get_treatment(MethodExceptionsAndLatencies.TREATMENT_WITH_CONFIG, key, feature_flag_name, attributes)
 
         except Exception:
-            # TODO: maybe log here?
+            _LOGGER.error('get_treatment_with_config failed')
             return CONTROL, None
 
     async def _get_treatment(self, method, key, feature, attributes=None):

splitio/client/factory.py

@@ -101,6 +101,11 @@ class TimeoutException(Exception):
 class SplitFactoryBase(object):  # pylint: disable=too-many-instance-attributes
     """Split Factory/Container class."""
 
+    def __init__(self, sdk_key, storages):
+        self._sdk_key = sdk_key
+        self._storages = storages
+        self._status = None
+
     def _get_storage(self, name):
         """
         Return a reference to the specified storage.
@@ -162,8 +167,7 @@ def __init__(  # pylint: disable=too-many-arguments
             telemetry_producer=None,
             telemetry_init_producer=None,
             telemetry_submitter=None,
-            preforked_initialization=False,
-            manager_start_task=None
+            preforked_initialization=False
     ):
         """
         Class constructor.
@@ -183,8 +187,7 @@ def __init__(  # pylint: disable=too-many-arguments
         :param preforked_initialization: Whether should be instantiated as preforked or not.
         :type preforked_initialization: bool
         """
-        self._sdk_key = sdk_key
-        self._storages = storages
+        SplitFactoryBase.__init__(self, sdk_key, storages)
         self._labels_enabled = labels_enabled
         self._sync_manager = sync_manager
         self._recorder = recorder
@@ -328,12 +331,11 @@ def __init__(  # pylint: disable=too-many-arguments
             labels_enabled,
             recorder,
             sync_manager=None,
-            sdk_ready_flag=None,
             telemetry_producer=None,
             telemetry_init_producer=None,
             telemetry_submitter=None,
-            preforked_initialization=False,
-            manager_start_task=None
+            manager_start_task=None,
+            api_client=None
     ):
         """
         Class constructor.
@@ -353,12 +355,10 @@ def __init__(  # pylint: disable=too-many-arguments
         :param preforked_initialization: Whether should be instantiated as preforked or not.
         :type preforked_initialization: bool
         """
-        self._sdk_key = sdk_key
-        self._storages = storages
+        SplitFactoryBase.__init__(self, sdk_key, storages)
         self._labels_enabled = labels_enabled
         self._sync_manager = sync_manager
         self._recorder = recorder
-        self._preforked_initialization = preforked_initialization
         self._telemetry_evaluation_producer = telemetry_producer.get_telemetry_evaluation_producer()
         self._telemetry_init_producer = telemetry_init_producer
         self._telemetry_submitter = telemetry_submitter
@@ -367,16 +367,14 @@ def __init__(  # pylint: disable=too-many-arguments
         self._manager_start_task = manager_start_task
         self._status = Status.NOT_INITIALIZED
         self._sdk_ready_flag = asyncio.Event()
-        asyncio.get_running_loop().create_task(self._update_status_when_ready_async())
+        self._ready_task = asyncio.get_running_loop().create_task(self._update_status_when_ready_async())
+        self._api_client = api_client
 
     async def _update_status_when_ready_async(self):
         """Wait until the sdk is ready and update the status for async mode."""
-        if self._preforked_initialization:
-            self._status = Status.WAITING_FORK
-            return
-
         if self._manager_start_task is not None:
             await self._manager_start_task
+            self._manager_start_task = None
         await self._telemetry_init_producer.record_ready_time(get_current_epoch_time_ms() - self._ready_time)
         redundant_factory_count, active_factory_count = _get_active_and_redundant_count()
         await self._telemetry_init_producer.record_active_and_redundant_factories(active_factory_count, redundant_factory_count)
@@ -430,14 +428,22 @@ async def destroy(self, destroyed_event=None):
 
         try:
             _LOGGER.info('Factory destroy called, stopping tasks.')
+            if self._manager_start_task is not None and not self._manager_start_task.done():
+                self._manager_start_task.cancel()
+
             if self._sync_manager is not None:
                 await self._sync_manager.stop(True)
 
+                if not self._ready_task.done():
+                    self._ready_task.cancel()
+                    self._ready_task = None
+
                 if isinstance(self._storages['splits'], RedisSplitStorageAsync):
                     await self._get_storage('splits').redis.close()
 
                 if isinstance(self._sync_manager, ManagerAsync) and isinstance(self._telemetry_submitter, InMemoryTelemetrySubmitterAsync):
-                    await self._telemetry_submitter._telemetry_api._client.close_session()
+                    await self._api_client.close_session()
+
         except Exception as e:
             _LOGGER.error('Exception destroying factory.')
             _LOGGER.debug(str(e))
@@ -453,24 +459,6 @@ def client(self):
         """
         return ClientAsync(self, self._recorder, self._labels_enabled)
 
-
-    async def resume(self):
-        """
-        Function in charge of starting periodic/realtime synchronization after a fork.
-        """
-        if not self._waiting_fork():
-            _LOGGER.warning('Cannot call resume')
-            return
-        self._sync_manager.recreate()
-        self._sdk_ready_flag = asyncio.Event()
-        self._sdk_internal_ready_flag = self._sdk_ready_flag
-        self._sync_manager._ready_flag = self._sdk_ready_flag
-        await self._get_storage('impressions').clear()
-        await self._get_storage('events').clear()
-        self._preforked_initialization = False  # reset for status updater
-        asyncio.get_running_loop().create_task(self._update_status_when_ready_async())
-
-
 def _wrap_impression_listener(listener, metadata):
     """
     Wrap the impression listener if any.
@@ -718,8 +706,6 @@ async def _build_in_memory_factory_async(api_key, cfg, sdk_url=None, events_url=
 
     synchronizer = SynchronizerAsync(synchronizers, tasks)
 
-    preforked_initialization = cfg.get('preforkedInitialization', False)
-
     manager = ManagerAsync(synchronizer, apis['auth'], cfg['streamingEnabled'],
                       sdk_metadata, telemetry_runtime_producer, streaming_api_base_url, api_key[-4:])
 
@@ -739,19 +725,13 @@ async def _build_in_memory_factory_async(api_key, cfg, sdk_url=None, events_url=
 
     await telemetry_init_producer.record_config(cfg, extra_cfg, total_flag_sets, invalid_flag_sets)
 
-    if preforked_initialization:
-        await synchronizer.sync_all(max_retry_attempts=_MAX_RETRY_SYNC_ALL)
-        await synchronizer._split_synchronizers._segment_sync.shutdown()
-
-        return SplitFactoryAsync(api_key, storages, cfg['labelsEnabled'],
-                            recorder, manager, None, telemetry_producer, telemetry_init_producer, telemetry_submitter, preforked_initialization=preforked_initialization)
-
     manager_start_task = asyncio.get_running_loop().create_task(manager.start())
 
     return SplitFactoryAsync(api_key, storages, cfg['labelsEnabled'],
-                        recorder, manager, manager_start_task,
+                        recorder, manager,
                         telemetry_producer, telemetry_init_producer,
-                        telemetry_submitter, manager_start_task=manager_start_task)
+                        telemetry_submitter, manager_start_task=manager_start_task,
+                        api_client=http_client)
 
 def _build_redis_factory(api_key, cfg):
     """Build and return a split factory with redis-based storage."""

splitio/push/manager.py

@@ -349,10 +349,15 @@ async def stop(self, blocking=False):
 
         if self._token_task:
             self._token_task.cancel()
+            self._token_task = None
 
-        stop_task = asyncio.get_running_loop().create_task(self._stop_current_conn())
         if blocking:
-            await stop_task
+            await self._stop_current_conn()
+        else:
+            asyncio.get_running_loop().create_task(self._stop_current_conn())
+
+    async def close_sse_http_client(self):
+        await self._sse_client.close_sse_http_client()
 
     async def _event_handler(self, event):
         """
@@ -382,6 +387,7 @@ async def _token_refresh(self, current_token):
         :param current_token: token (parsed) JWT
         :type current_token: splitio.models.token.Token
         """
+        _LOGGER.debug("Next token refresh in " + str(self._get_time_period(current_token)) + " seconds")
         await asyncio.sleep(self._get_time_period(current_token))
         await self._stop_current_conn()
         self._running_task = asyncio.get_running_loop().create_task(self._trigger_connection_flow())
@@ -441,6 +447,7 @@ async def _trigger_connection_flow(self):
         finally:
             if self._token_task is not None:
                 self._token_task.cancel()
+                self._token_task = None
             self._running = False
             await self._processor.update_workers_status(False)
             self._done.set()
@@ -530,4 +537,5 @@ async def _stop_current_conn(self):
         await self._sse_client.stop()
         self._running_task.cancel()
         await self._running_task
+        self._running_task = None
         _LOGGER.debug("SplitSSE tasks are stopped")

splitio/push/splitsse.py

@@ -181,7 +181,7 @@ def __init__(self, sdk_metadata, client_key=None, base_url='https://streaming.sp
         self._base_url = base_url
         self.status = SplitSSEClient._Status.IDLE
         self._metadata = headers_from_metadata(sdk_metadata, client_key)
-        self._client = SSEClientAsync(timeout=self.KEEPALIVE_TIMEOUT)
+        self._client = SSEClientAsync(self.KEEPALIVE_TIMEOUT)
         self._event_source = None
         self._event_source_ended = asyncio.Event()
 
@@ -219,7 +219,7 @@ async def start(self, token):
             _LOGGER.debug('stack trace: ', exc_info=True)
         finally:
             self.status = SplitSSEClient._Status.IDLE
-            _LOGGER.debug('sse connection ended.')
+            _LOGGER.debug('Split sse connection ended.')
             self._event_source_ended.set()
 
     async def stop(self):
@@ -230,4 +230,13 @@ async def stop(self):
             return
 
         await self._client.shutdown()
-        await self._event_source_ended.wait()
+# catching exception to avoid task hanging
+        try:
+            await self._event_source_ended.wait()
+        except asyncio.CancelledError as e:
+            _LOGGER.error("Exception waiting for event source ended")
+            _LOGGER.debug('stack trace: ', exc_info=True)
+            pass
+
+    async def close_sse_http_client(self):
+        await self._client.close_session()