Datadog无代码修改收集追踪数据的原理及Python实现问询

一、核心技术原理 #

ddtrace-run能自动收集Flask等框架的请求数据，核心依赖以下三种技术：

1. Monkey Patching（猴子补丁） #

这是ddtrace最常用的手段。Python作为动态语言，允许在运行时替换已加载模块的函数、方法甚至类。比如针对Flask框架，ddtrace会找到flask.app.Flask.add_url_rule或者视图函数的底层实现，将其替换为自己包装后的函数——在原业务逻辑执行前后，插入收集请求路径、开始时间、结束时间、响应状态码的逻辑。

这种方式无需修改业务代码，只要在程序启动前完成补丁注入，就能自动生效。

2. Bytecode Instrumentation（字节码插桩） #

当猴子补丁无法覆盖某些场景（比如内置函数、第三方库的内部私有方法）时，ddtrace会直接操作Python编译后的字节码。Python代码运行前会被编译成字节码（可通过dis模块查看），字节码插桩就是在这些字节码指令中插入额外的埋点逻辑，比如函数进入时记录时间、退出时上报数据。

这种方式更底层，不需要依赖目标模块的代码结构，但实现复杂度更高。

3. Aspect-Oriented Programming（AOP，面向切面编程） #

上面两种技术都是AOP思想的落地实现。AOP的核心是将**非业务核心逻辑（比如埋点、日志、权限校验）**从业务代码中剥离，作为独立的“切面”，在指定的“切点”（比如Flask请求开始/结束、函数调用前后）自动执行。ddtrace的埋点逻辑就是一个典型的切面，通过前两种技术把切面逻辑注入到业务流程中。

二、Python中实现类似自动埋点功能 #

1. 基于猴子补丁实现Flask埋点 #

以下是简化版实现，模拟ddtrace收集请求数据的逻辑：

# trace_flask.py
import time
from flask import Flask, request

def trace_decorator(view_func):
    def wrapped(*args, **kwargs):
        # 请求开始前收集基础信息
        start_time = time.time()
        path = request.path
        method = request.method
        
        # 执行原视图函数
        response = view_func(*args, **kwargs)
        
        # 请求结束后计算延迟并收集状态码
        latency = time.time() - start_time
        status_code = response.status_code
        
        # 此处可替换为上报到监控系统的逻辑
        print(f"Request [{method} {path}] - Status: {status_code}, Latency: {latency:.4f}s")
        
        return response
    return wrapped

def patch_flask():
    # 猴子补丁：替换Flask的add_url_rule方法，给所有视图函数加埋点
    original_add_url_rule = Flask.add_url_rule
    def patched_add_url_rule(self, rule, endpoint=None, view_func=None, **options):
        if view_func is not None:
            view_func = trace_decorator(view_func)
        original_add_url_rule(self, rule, endpoint, view_func, **options)
    Flask.add_url_rule = patched_add_url_rule

启动Flask应用前加载补丁：

python -c "import trace_flask; trace_flask.patch_flask(); import app; app.app.run()"

2. 基于字节码插桩实现函数埋点 #

用bytecode库（需先安装pip install bytecode）实现对任意函数的自动计时：

# bytecode_trace.py
import time
from bytecode import Bytecode, Instr

def add_trace_to_function(func):
    bc = Bytecode.from_code(func.__code__)
    
    # 在函数开头插入计时逻辑
    bc.insert(0, Instr("LOAD_GLOBAL", name="time"))
    bc.insert(1, Instr("LOAD_METHOD", name="time"))
    bc.insert(2, Instr("CALL_METHOD", argc=0))
    bc.insert(3, Instr("STORE_FAST", name="_start_time"))
    
    # 在函数结尾插入计算延迟并打印的逻辑
    bc.append(Instr("LOAD_GLOBAL", name="time"))
    bc.append(Instr("LOAD_METHOD", name="time"))
    bc.append(Instr("CALL_METHOD", argc=0))
    bc.append(Instr("LOAD_FAST", name="_start_time"))
    bc.append(Instr("SUBTRACT"))
    bc.append(Instr("STORE_FAST", name="_latency"))
    bc.append(Instr("LOAD_GLOBAL", name="print"))
    bc.append(Instr("LOAD_CONST", value=f"Function {func.__name__} latency:"))
    bc.append(Instr("LOAD_FAST", name="_latency"))
    bc.append(Instr("CALL_FUNCTION", argc=2))
    bc.append(Instr("POP_TOP"))
    
    # 更新函数的字节码
    func.__code__ = bc.to_code()
    return func

# 测试用例
@add_trace_to_function
def test_func():
    time.sleep(0.5)

test_func()

3. 封装成自定义启动脚本 #

参考ddtrace-run的实现，写一个自定义启动脚本my_trace_run.py，支持用户用my_trace_run python app.py启动应用：

# my_trace_run.py
import sys
import trace_flask

# 先加载猴子补丁
trace_flask.patch_flask()

# 执行目标脚本
if len(sys.argv) < 2:
    print("Usage: my_trace_run <script.py>")
    sys.exit(1)

script_path = sys.argv[1]
sys.argv = sys.argv[1:]  # 把脚本参数传给目标脚本

with open(script_path) as f:
    code = compile(f.read(), script_path, "exec")
    exec(code)

使用方式：

python my_trace_run.py app.py

内容的提问来源于stack exchange，提问作者Curious Learner