diff --git a/logtail/handler.py b/logtail/handler.py index 3c83c56..2849b6e 100644 --- a/logtail/handler.py +++ b/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(): diff --git a/logtail/uploader.py b/logtail/uploader.py index 9d8dce6..da28c6b 100644 --- a/logtail/uploader.py +++ b/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,11 +15,17 @@ 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) @@ -25,3 +33,15 @@ def __call__(self, frame): 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 diff --git a/tests/test_handler.py b/tests/test_handler.py index 32e8476..603e604 100644 --- a/tests/test_handler.py +++ b/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) diff --git a/tests/test_uploader.py b/tests/test_uploader.py index d2027a9..e1f680a 100644 --- a/tests/test_uploader.py +++ b/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)