feat: Add postfork method to re-initialize after forking (#329)

Author: keelerm84

contract-tests/service.py

@@ -74,6 +74,7 @@ def status():
             'event-sampling',
             'polling-gzip',
             'inline-context-all',
+            'instance-id',
             'anonymous-redaction',
             'evaluation-hooks',
             'omit-anonymous-contexts',

ldclient/client.py

@@ -7,6 +7,7 @@
 import threading
 import traceback
 from typing import Any, Callable, Dict, List, Mapping, Optional, Tuple
+from uuid import uuid4
 
 from ldclient.config import Config
 from ldclient.context import Context
@@ -188,15 +189,43 @@ def __init__(self, config: Config, start_wait: float = 5):
         check_uwsgi()
 
         self._config = config
+        self._config._instance_id = str(uuid4())
         self._config._validate()
 
-        self.__hooks_lock = ReadWriteLock()
-        self.__hooks = config.hooks  # type: List[Hook]
-
         self._event_processor = None
         self._event_factory_default = EventFactory(False)
         self._event_factory_with_reasons = EventFactory(True)
 
+        self.__start_up(start_wait)
+
+    def postfork(self, start_wait: float = 5):
+        """
+        Re-initializes an existing client after a process fork.
+
+        The SDK relies on multiple background threads to operate correctly.
+        When a process forks, `these threads are not available to the child
+        <https://pythondev.readthedocs.io/fork.html#reinitialize-all-locks-after-fork>`.
+
+        As a result, the SDK will not function correctly in the child process
+        until it is re-initialized.
+
+        This method is effectively equivalent to instantiating a new client.
+        Future iterations of the SDK will provide increasingly efficient
+        re-initializing improvements.
+
+        Note that any configuration provided to the SDK will need to survive
+        the forking process independently. For this reason, it is recommended
+        that any listener or hook integrations be added postfork unless you are
+        certain it can survive the forking process.
+
+        :param start_wait: the number of seconds to wait for a successful connection to LaunchDarkly
+        """
+        self.__start_up(start_wait)
+
+    def __start_up(self, start_wait: float):
+        self.__hooks_lock = ReadWriteLock()
+        self.__hooks = self._config.hooks  # type: List[Hook]
+
         data_store_listeners = Listeners()
         store_sink = DataStoreUpdateSinkImpl(data_store_listeners)
         store = _FeatureStoreClientWrapper(self._config.feature_store, store_sink)

ldclient/config.py

@@ -289,6 +289,7 @@ def __init__(
         self.__omit_anonymous_contexts = omit_anonymous_contexts
         self.__payload_filter_key = payload_filter_key
         self._data_source_update_sink: Optional[DataSourceUpdateSink] = None
+        self._instance_id: Optional[str] = None
 
     def copy_with_new_sdk_key(self, new_sdk_key: str) -> 'Config':
         """Returns a new ``Config`` instance that is the same as this one, except for having a different SDK key.

ldclient/impl/http.py

@@ -25,6 +25,9 @@ def _application_header_value(application: dict) -> str:
 def _base_headers(config):
     headers = {'Authorization': config.sdk_key or '', 'User-Agent': 'PythonClient/' + VERSION}
 
+    if config._instance_id is not None:
+        headers['X-LaunchDarkly-Instance-Id'] = config._instance_id
+
     app_value = _application_header_value(config.application)
     if app_value:
         headers['X-LaunchDarkly-Tags'] = app_value

ldclient/testing/impl/datasource/test_feature_requester.py

@@ -37,6 +37,21 @@ def test_get_all_data_sends_headers():
         assert req.headers['Accept-Encoding'] == 'gzip'
         assert req.headers.get('X-LaunchDarkly-Wrapper') is None
         assert req.headers.get('X-LaunchDarkly-Tags') is None
+        assert req.headers.get('X-LaunchDarkly-Instance-Id') is None
+
+
+def test_sets_instance_id_header():
+    with start_server() as server:
+        config = Config(sdk_key='sdk-key', base_uri=server.uri)
+        config._instance_id = 'my-instance-id'
+        fr = FeatureRequesterImpl(config)
+
+        resp_data = {'flags': {}, 'segments': {}}
+        server.for_path('/sdk/latest-all', JsonResponse(resp_data))
+
+        fr.get_all_data()
+        req = server.require_request()
+        assert req.headers.get('X-LaunchDarkly-Instance-Id') == 'my-instance-id'
 
 
 def test_get_all_data_sends_wrapper_header():

ldclient/testing/impl/datasource/test_streaming.py

@@ -50,9 +50,26 @@ def test_request_properties():
                 assert req.headers.get('Authorization') == 'sdk-key'
                 assert req.headers.get('User-Agent') == 'PythonClient/' + VERSION
                 assert req.headers.get('X-LaunchDarkly-Wrapper') is None
+                assert req.headers.get('X-LaunchDarkly-Instance-Id') is None
                 assert req.headers.get('X-LaunchDarkly-Tags') is None
 
 
+def test_sends_instance_id():
+    store = InMemoryFeatureStore()
+    ready = Event()
+
+    with start_server() as server:
+        with stream_content(make_put_event()) as stream:
+            config = Config(sdk_key='sdk-key', stream_uri=server.uri, wrapper_name='Flask', wrapper_version='0.1.0')
+            config._instance_id = 'my-instance-id'
+            server.for_path('/all', stream)
+
+            with StreamingUpdateProcessor(config, store, ready, None) as sp:
+                sp.start()
+                req = server.await_request()
+                assert req.headers.get('X-LaunchDarkly-Instance-Id') == 'my-instance-id'
+
+
 def test_sends_wrapper_header():
     store = InMemoryFeatureStore()
     ready = Event()

ldclient/testing/impl/events/test_event_processor.py

@@ -60,7 +60,14 @@ def __init__(self, **kwargs):
             kwargs['diagnostic_opt_out'] = True
         if 'sdk_key' not in kwargs:
             kwargs['sdk_key'] = 'SDK_KEY'
+
+        instance_id = None
+        if 'instance_id' in kwargs:
+            instance_id = kwargs['instance_id']
+            del kwargs['instance_id']
+
         config = Config(**kwargs)
+        config._instance_id = instance_id
         diagnostic_accumulator = _DiagnosticAccumulator(create_diagnostic_id(config))
         DefaultEventProcessor.__init__(self, config, mock_http, diagnostic_accumulator=diagnostic_accumulator)
 
@@ -572,6 +579,24 @@ def test_wrapper_header_sent_when_set():
         assert mock_http.request_headers.get('X-LaunchDarkly-Wrapper') == "Flask/0.0.1"
 
 
+def test_instance_id_header_not_sent_when_not_set():
+    with DefaultTestProcessor() as ep:
+        ep.send_event(EventInputIdentify(timestamp, context))
+        ep.flush()
+        ep._wait_until_inactive()
+
+        assert mock_http.request_headers.get('X-LaunchDarkly-Wrapper') is None
+
+
+def test_instance_id_header_sent_when_set():
+    with DefaultTestProcessor(instance_id="my-instance-id") as ep:
+        ep.send_event(EventInputIdentify(timestamp, context))
+        ep.flush()
+        ep._wait_until_inactive()
+
+        assert mock_http.request_headers.get('X-LaunchDarkly-Instance-Id') == "my-instance-id"
+
+
 def test_wrapper_header_sent_without_version():
     with DefaultTestProcessor(wrapper_name="Flask") as ep:
         ep.send_event(EventInputIdentify(timestamp, context))

ldclient/testing/test_ldclient_end_to_end.py

@@ -15,6 +15,22 @@
 always_true_flag = {'key': 'flagkey', 'version': 1, 'on': False, 'offVariation': 1, 'variations': [False, True]}
 
 
+def test_config_ignores_initial_instance_id():
+    with start_server() as stream_server:
+        with stream_content(make_put_event([always_true_flag])) as stream_handler:
+            stream_server.for_path('/all', stream_handler)
+            config = Config(sdk_key=sdk_key, stream_uri=stream_server.uri, send_events=False)
+            config._instance_id = "Hey, I'm not supposed to modify this"
+
+            with LDClient(config=config) as client:
+                assert client.is_initialized()
+                assert client.variation(always_true_flag['key'], user, False) is True
+
+                r = stream_server.await_request()
+                assert r.headers['X-LaunchDarkly-Instance-Id'] == config._instance_id
+                assert r.headers['X-LaunchDarkly-Instance-Id'] != "Hey, I'm not supposed to modify this"
+
+
 def test_client_starts_in_streaming_mode():
     with start_server() as stream_server:
         with stream_content(make_put_event([always_true_flag])) as stream_handler: