如何装饰 Python 进程以捕获 HTTP 请求和响应？

Question

我目前正在研究一个非常大的 Python 代码库，它有很多副作用和意外行为，我想通过查看它在整个执行过程中发出的所有出站 HTTP 请求来了解它在做什么，在调用堆栈中的任何位置。是否有任何实用程序或集成路径允许我自动分析由 Python 编写的代码进行的完整网络调用集？

具体来说，与单独的外部工具相比，我希望能够从分析模块或相邻模块中以编程方式与捕获的 HTTP 请求和响应进行交互；例如：

• 使用该代码库中现有的logging handlers 记录请求/响应
• 将请求发布到事件代理，例如Kafka
• 解析成pandas dataframe进行分析
• 集成到现有的unittest 或pytest 套件中

我查看了不同可观察性工具的产品。例如，Sentry 似乎自动与 Python 的 httplib 集成，为每个请求创建一个“面包屑”；然而，Sentry 仅在抛出异常时记录此信息，并且其默认行为仅发布到其 Web UI。 New Relic 还提供了查看“外部服务”调用的能力，作为其应用程序性能监控产品的一部分，同样是通过它自己的仪表板。然而，在这两种情况下，它们都缺少官方支持的 Python 处理程序，该处理程序允许上述任务发生在生成出站网络请求的进程中。

Answer 1

我查看了Sentry's Python SDK source code，了解它们如何与http.client 集成，并以一种通用的方式调整了他们的方法以满足我的需求。

这是我编写的代码，用于装饰http.client.HTTPConnection 对象以利用请求、请求正文和响应对象。这个特定示例将我要收集的数据附加到分析模块下的全局列表中，并将相同的数据记录到标准输出中。您可以轻松替换任何您想要的定制功能来代替对list.append 和logger.info 的调用：

import logging
import sys
from http.client import HTTPConnection

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler(stream=sys.stdout)
formatter = logging.Formatter(fmt="%(name)s %(funcName)s %(levelname)s: %(message)s")
handler.setFormatter(formatter)
logger.addHandler(handler)

put_request_content = []
get_response_content = []
request_bodies = []


def decorate_HTTPConnection():
    """Taken loosely from https://github.com/getsentry/sentry-python/blob/master/sentry_sdk/integrations/stdlib.py"""

    global put_request_content, get_response_content, request_bodies

    real_putrequest = HTTPConnection.putrequest
    real_getresponse = HTTPConnection.getresponse
    real__send_output = HTTPConnection._send_output

    def new_putrequest(self, method, url, skip_host=False, skip_accept_encoding=False):
        logger.info(f'{method}: {url}')
        put_request_content.append((method, url))

        real_putrequest(self, method, url, skip_host=skip_host, skip_accept_encoding=skip_accept_encoding)

    def new_getresponse(self):
        returned_response = real_getresponse(self)

        logger.info(returned_response)
        get_response_content.append(returned_response)

        return returned_response

    def new__send_output(self, message_body=None, encode_chunked=False):
        logger.info(f'Message body: {message_body}')
        request_bodies.append(message_body)

        real__send_output(self, message_body=message_body, encode_chunked=encode_chunked)

    HTTPConnection.putrequest = new_putrequest
    HTTPConnection.getresponse = new_getresponse
    HTTPConnection._send_output = new__send_output


decorate_HTTPConnection()

这是我用来测试其行为的一个非常简单的脚本：

import logging
import sys
import requests

from http_profiler.connection_decorator import put_request_content, get_response_content, request_bodies

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler(stream=sys.stdout)
formatter = logging.Formatter(fmt="%(name)s %(funcName)s %(levelname)s: %(message)s")
handler.setFormatter(formatter)
logger.addHandler(handler)


def test_profile_http_get_via_requests_library(url):
    prev_len_put_request_content = len(put_request_content)
    prev_len_get_repsonse_content = len(get_response_content)
    prev_len_request_bodies = len(request_bodies)

    logger.info(f"Starting the test: GET {url}")
    resp = requests.get(url=url)

    assert resp is not None
    assert len(put_request_content) - prev_len_put_request_content == 1
    assert len(get_response_content) - prev_len_get_repsonse_content == 1
    assert len(request_bodies) - prev_len_request_bodies == 1


def test_profile_http_post_via_requests_library(url, data=None):
    if data is None:
        data = {"message": "Hello world!"}

    prev_len_put_request_content = len(put_request_content)
    prev_len_get_repsonse_content = len(get_response_content)
    prev_len_request_bodies = len(request_bodies)

    logger.info(f"Starting the test: POST {url} with {data}")
    resp = requests.post(url=url, data=data)

    assert resp is not None
    assert len(put_request_content) - prev_len_put_request_content == 1
    assert len(get_response_content) - prev_len_get_repsonse_content == 1
    assert len(request_bodies) - prev_len_request_bodies == 1


if __name__ == "__main__":
    test_profile_http_get_via_requests_library("https://example.com")
    test_profile_http_post_via_requests_library("https://example.com")
    logger.info(f'Requests: {put_request_content}')
    logger.info(f'Request bodies: {request_bodies}')
    logger.info(f'Responses: {[f"{response.status} {response.reason}" for response in get_response_content]}')

这是测试脚本的输出：

__main__ test_profile_http_get_via_requests_library INFO: Starting the test: GET https://example.com
http_profiler.connection_decorator new_putrequest INFO: GET: /
http_profiler.connection_decorator new__send_output INFO: Message body: None
http_profiler.connection_decorator new_getresponse INFO: <http.client.HTTPResponse object at 0x7ff40aa5df10>
__main__ test_profile_http_post_via_requests_library INFO: Starting the test: POST https://example.com with {'message': 'Hello world!'}
http_profiler.connection_decorator new_putrequest INFO: POST: /
http_profiler.connection_decorator new__send_output INFO: Message body: b'message=Hello+world%21'
http_profiler.connection_decorator new_getresponse INFO: <http.client.HTTPResponse object at 0x7ff40aa5deb0>
__main__ <module> INFO: Requests: [('GET', '/'), ('POST', '/')]
__main__ <module> INFO: Request bodies: [None, b'message=Hello+world%21']
__main__ <module> INFO: Responses: ['200 OK', '200 OK']