T-17170 Resolve proxy settings once so forked RQ work horses on macOS stop crashing at flush

logtail/handler.py

@@ -100,7 +100,7 @@ def _reset_after_fork(self):
         # whose socket it shares with the parent, so it starts over with fresh ones.
         self.pipe = queue.Queue(maxsize=self.buffer_capacity)
         self.flush_thread = None
-        self.uploader = Uploader(self.source_token, self.host, self.uploader.timeout)
+        self.uploader.reset()
 
 
 def _reset_handlers_after_fork():

logtail/uploader.py

@@ -1,7 +1,9 @@
 # coding: utf-8
 from __future__ import print_function, unicode_literals
+import os
 import msgpack
 import requests
+import requests.utils
 
 class Fake500(object):
     def __init__(self, exception):
@@ -13,15 +15,33 @@ def __init__(self, source_token, host, timeout):
         self.source_token = source_token
         self.host = host
         self.timeout = timeout
-        self.session = requests.Session()
+        # requests would otherwise ask the environment, and on macOS SystemConfiguration, for
+        # proxy settings on every request. Resolving them once here keeps the flush thread out
+        # of CoreFoundation, which refuses to run in a child forked from a multithreaded
+        # process, such as an RQ work horse. The CA bundle variables follow the same rule.
+        self.proxies = requests.utils.get_environ_proxies(host)
+        self.verify = os.environ.get('REQUESTS_CA_BUNDLE') or os.environ.get('CURL_CA_BUNDLE') or True
         self.headers = {
             'Authorization': 'Bearer %s' % source_token,
             'Content-Type': 'application/msgpack',
         }
+        self.session = self._new_session()
 
     def __call__(self, frame):
         data = msgpack.packb(frame, use_bin_type=True)
         try:
             return self.session.post(self.host, data=data, headers=self.headers, timeout=self.timeout)
         except requests.RequestException as e:
             return Fake500(e)
+
+    def reset(self):
+        # A forked child shares the parent's pooled socket, so it needs a session of its own,
+        # built from the settings resolved above rather than looked up again.
+        self.session = self._new_session()
+
+    def _new_session(self):
+        session = requests.Session()
+        session.trust_env = False
+        session.proxies = self.proxies
+        session.verify = self.verify
+        return session

tests/test_handler.py

@@ -285,11 +285,15 @@ def test_forked_child_starts_with_a_fresh_queue_thread_and_session(self, MockWor
         logger.handlers = []
         logger.addHandler(handler)
         logger.critical('queued in the parent')
-        parent_session = handler.uploader.session
+        uploader = handler.uploader
+        parent_session = uploader.session
 
         pid = os.fork()
         if pid == 0:
-            fresh = handler.pipe.empty() and handler.flush_thread is None and handler.uploader.session is not parent_session
+            # The child gets its own session, but never resolves proxy settings again: on macOS
+            # that means SystemConfiguration, which refuses to run in a forked child.
+            fresh = (handler.pipe.empty() and handler.flush_thread is None and handler.uploader is uploader
+                     and uploader.session is not parent_session and uploader.session.trust_env is False)
             os._exit(0 if fresh else 1)
         _, status = os.waitpid(pid, 0)
 

tests/test_uploader.py

@@ -34,3 +34,33 @@ def mock_post(endpoint, data=None, headers=None, timeout=None):
         u(self.frame)
 
         self.assertTrue(post.called)
+
+    @patch('logtail.uploader.requests.utils.get_environ_proxies', return_value={'https': 'http://proxy.internal:3128'})
+    def test_resolves_proxy_settings_once_instead_of_on_every_request(self, get_environ_proxies):
+        u = Uploader(self.source_token, self.host, self.timeout)
+
+        with patch('logtail.uploader.requests.Session.post'):
+            u(self.frame)
+            u(self.frame)
+
+        get_environ_proxies.assert_called_once_with(self.host)
+        self.assertEqual(u.session.proxies, {'https': 'http://proxy.internal:3128'})
+        self.assertFalse(u.session.trust_env)
+
+    @patch.dict('os.environ', {'REQUESTS_CA_BUNDLE': '/etc/ssl/corporate-ca.pem'})
+    def test_uses_the_ca_bundle_from_the_environment(self):
+        u = Uploader(self.source_token, self.host, self.timeout)
+
+        self.assertEqual(u.session.verify, '/etc/ssl/corporate-ca.pem')
+
+    @patch('logtail.uploader.requests.utils.get_environ_proxies', return_value={'https': 'http://proxy.internal:3128'})
+    def test_reset_replaces_the_session_but_keeps_the_resolved_settings(self, get_environ_proxies):
+        u = Uploader(self.source_token, self.host, self.timeout)
+        session = u.session
+
+        u.reset()
+
+        self.assertIsNot(u.session, session)
+        self.assertEqual(u.session.proxies, {'https': 'http://proxy.internal:3128'})
+        self.assertFalse(u.session.trust_env)
+        get_environ_proxies.assert_called_once_with(self.host)