使用装饰器使用 OpenTelemetry 跟踪来检测 Python 代码
Python 装饰器有助于保持 OpenTelemetry 跟踪检测的 DRY
上周，我写了一篇关于使用装饰器设计模式来帮助减少设置追踪所需样板代码的文章。我的示例代码使用的是 .NET，需要进行大量繁琐的工作，使用 DispatchProxy 类来拦截方法调用并注入追踪逻辑。然而，Python 通过内置的函数装饰器支持，极大地简化了我们的工作。
为什么要使用追踪装饰器？
OpenTelemetry 追踪功能非常强大！它允许在代码中定义称为“Span”的特定片段，并跟踪它们在运行时执行情况和依赖关系。但是，为了使这些信息可用，需要在代码中进行大量规范化的仪器化。基本上，对于每个类/模块/函数，您都需要添加类似这样的代码
tracer = trace.get_tracer(__name__)
class UserService:
def all(self, id):
with tracer.start_as_current_span("retrieve users") as span:
users = self.user_store.get_users(id)
span.set_attribute('some attribute', 'some_value')
...
async def validate(self, user_name):
with tracer.start_as_current_span("validate users") as span:
span.set_attribute('some attribute', 'some_value')
try:
await self.user_store.get_user(user_name)
...
async def test(self, user_name):
with tracer.start_as_current_span("test user") as span:
span.set_attribute('some attribute', 'some_value')
try:
await self.user_store.get_user(user_name)
...
除了重复性以及需要在所有地方强制添加 tracer.start_as_current_span 调用之外，为 Span 制定正确的命名约定，并确保它们具有唯一的名称和粒度级别，对于大型代码库来说可能非常具有挑战性。
如果我们能够编写如下代码，而不是上述代码，那将是极好的
@instrument({"some attribute", "some_value"})
class UserService:
def all(self, id):
with tracer.start_as_current_span("retrieve users"):
users = self.user_store.get_users(id)
...
@instrument(span_name="custom_name")
async def validate(self, user_name):
try:
await self.user_store.get_user(user_name)
...
async def test(self, user_name):
try:
await self.user_store.get_user(user_name)
...
这样，整个类就可以自动进行仪器化，并注入 Span 属性。可以通过约定（例如，函数名称）来分配名称，并且可以轻松地更改该约定，如果需要的话。
本文的全部源代码可在此仓库中找到。它也作为一个pypi 包提供，其中包含装饰器实现，如果您只想立即使用它的话。
使用 Python 装饰器实现基本追踪装饰器
Python 装饰器非常巧妙。如果您正在寻找更全面的使用文档，我推荐Geir Arne Hjelle的文章Python 装饰器入门指南，它非常出色地涵盖了这个主题。
为了开始，我们将创建一个简单的装饰器，可以添加到函数上以自动对其进行仪器化。这将消除添加追踪样板代码的需要，并负责 span 的默认命名约定（我们稍后会对其进行泛化）。Python 中的实现相当直接，它包含返回将在解释器中作为函数装饰器应用的包装器函数。这是代码
def instrument(_func=None, *, span_name: str = "", record_exception: bool = True,
attributes: Dict[str, str] = None, existing_tracer: Tracer = None):
def span_decorator(func):
tracer = existing_tracer or trace.get_tracer(func.__module__)
def _set_attributes(span, attributes_dict):
if attributes_dict:
for att in attributes_dict:
span.set_attribute(att, attributes_dict[att])
@wraps(func)
def wrap_with_span(*args, **kwargs):
name = span_name or TracingDecoratorOptions.naming_scheme(func)
with tracer.start_as_current_span(name, record_exception=record_exception) as span:
_set_attributes(span, attributes)
return func(*args, **kwargs)
return wrap_with_span
if _func is None:
return span_decorator
else:
return span_decorator(_func)
基本上，在这个阶段，我们所做的就是返回包装器函数 wrap_with_span，该函数将被解释器作为函数装饰器应用。我们使用了 functools.wraps() 装饰器（第 12 行），它确保包装器函数将具有与原始函数相同的名称和元数据，从而使装饰器对函数调用者透明。
wrap_with_span 函数将自动创建和命名 span，并设置其属性。默认情况下，我们根据函数名称命名 span，但我们可以围绕它进行扩展。例如，这个 static 类可以允许开发人员用其他实现替换默认的命名约定
class TracingDecoratorOptions:
class NamingSchemes:
@staticmethod
def function_qualified_name(func: Callable):
return func.__qualname__
default_scheme = function_qualified_name
naming_scheme: Callable[[Callable], str] = NamingSchemes.default_scheme
default_attributes: Dict[str, str] = {}
@staticmethod
def set_naming_scheme(naming_scheme: Callable[[Callable], str]):
TracingDecoratorOptions.naming_scheme = naming_scheme
我们还提供了一个 span_name 参数，开发人员可以使用它来设置自定义的 span 名称，如果需要的话。
测试代码
让我们添加一个测试来检查我们新装饰器是否正常工作
@classmethod
def setup_class(cls):
resource = Resource.create(attributes={SERVICE_NAME: "test"})
provider = TracerProvider(resource=resource)
trace.set_tracer_provider(provider)
@instrument
def test_decorated_function_gets_instrumented_automatically_with_span():
assert trace.get_current_span().is_recording() is True
我们设置了 OTEL，以便追踪操作会产生效果，然后在测试方法内部进行验证，该方法已应用了新装饰器，其中有一个活动的追踪 span。
成功！TracingDecorator 的第一个迭代已完成。
仪器化整个类
逐个函数地添加装饰器也可能变得有些繁琐和重复。为了解决这个问题，我们可以修改装饰器，使其遍历类中的每个函数并对其进行装饰，暂时忽略 private 函数
def instrument(_func_or_class=None, *, span_name: str = "", record_exception: bool = True,
attributes: Dict[str, str] = None, existing_tracer: Tracer = None, ignore=False):
def decorate_class(cls):
for name, method in inspect.getmembers(cls, inspect.isfunction):
# Ignore private functions, TODO: maybe make this a setting?
if not name.startswith('_'):
setattr(cls, name, instrument(record_exception=record_exception,
attributes=attributes,
existing_tracer=existing_tracer)(method))
return cls
# Check if this is a span or class decorator
if inspect.isclass(_func_or_class):
return decorate_class(_func_or_class)
def span_decorator(func_or_class):
if inspect.isclass(func_or_class):
return decorate_class(func_or_class)
...
if _func_or_class is None:
return span_decorator
else:
return span_decorator(_func_or_class)
在上面的代码中，我们选择不为类创建单独的新装饰器，而是重载用于函数的相同装饰器。为了实现这一点，我们添加了一个检查，测试传递的参数是函数还是类（第 14、19 行），并相应地应用正确的逻辑。
对于类，我们遍历类函数并注入装饰器，返回未更改的类对象。
对于函数，我们遵循与之前相同的逻辑来应用包装器。
有了这段代码，我们现在可以重写原始代码来利用新装饰器
@instrument({"some attribute", "some_value"})
class UserService:
def all(self, id):
with tracer.start_as_current_span("retrieve users"):
users = self.user_store.get_users(id)
...
@instrument(span_name="custom_name")
async def validate(self, user_name):
try:
await self.user_store.get_user(user_name)
...
async def test(self, user_name):
try:
await self.user_store.get_user(user_name)
*.基本*有效
请注意，我们仍然有一个小问题。上面示例中的 validate 函数将同时应用两个装饰器。结果是会创建两个 span，而不是一个，这并非我们期望的行为。
为了处理这种情况，装饰器代码必须具有某种“记忆”，即一个函数是否已经被装饰。有几种方法可以做到这一点，在我们的实现中，我们选择将该信息保存在函数元数据中。因此，在装饰函数之前，我们会检查它是否已经被装饰
def span_decorator(func_or_class):
if inspect.isclass(func_or_class):
return decorate_class(func_or_class)
# Check if already decorated (happens if both class and function
# decorated). If so, we keep the function decorator settings only
undecorated_func = getattr(func_or_class, '__tracing_unwrapped__', None)
if undecorated_func:
# We have already decorated this function, override
return func_or_class
setattr(func_or_class, '__tracing_unwrapped__', func_or_class)
如果函数已经被装饰，我们原样返回它。
您还会添加什么？
您可以在Digma OpenTelemetry 仓库中找到完整的源代码。如果这对您的项目有用，请告诉我！另外，如果您觉得有任何功能会很有用，请随时与我联系，或在 GitHub 上打开一个 issue 或 PR。
历史
- 2023 年 7 月 4 日：初始版本