Python 跟踪 SDK

Instana Python 追踪 SDK

Instana Python 模块提供基于 OpenTelemetry API 的应用程序接口，用于跟踪应用程序的任意部分。

您可以按以下方式使用 API 设置一段代码以进行跟踪：

from instana.singletons import tracer


try:
    span = tracer.start_span("span_name")
    # The code to be instrumented
    id = user.find_by_name("john.smith")
    span.set_attribute("user_id", id)
except Exception as e:
    span.record_exception(e)
finally:
    if span and span.is_recording():
        span.end()

另外，也可以使用 with-as 语句来使用上下文管理器，它可以自动捕获并记录任何异常，如下所示：

from instana.singletons import tracer


with tracer.start_as_current_span("span_name") as span:
    # The code to be instrumented.
    id = user.find_by_name("john.smith")
    span.set_attribute("user_id", id)

异步跟踪

您要跟踪的某些操作可能是异步的，这意味着它们会立即返回，但会继续与主指令序列分开运行。 要跟踪这些操作，例如使用 asyncio 库，可以使用以下跟踪方法：

import asyncio
from instana.singletons import tracer


async def do_work(parent_span):
    parent_context = parent_span.get_span_context() if parent_span else None
    with tracer.start_as_current_span(
        "launch_async_work",
        span_context=parent_context
    ):
        print("Work stared")
        await asyncio.sleep(1)
        print("Work finished!")

with tracer.start_as_current_span("launch_uvloop") as sync_span:
    asyncio.run(do_work(sync_span))

跟踪分叉进程中的跨度

如果要跟踪分叉进程中的跨度，则在 span.end() 后添加 time.sleep() 。 请参阅以下示例：

from instana.singletons import tracer


 def forked_process(num):
    print(f"in forked process {num}")
    try:
        span = tracer.start_span(f"forked process {num}")
        print(f"sleep 10 for forked process {num}")
        time.sleep(10)
    except Exception as e:
        span.record_exception(e)
    finally:
        if span and span.is_recording():
            span.end()

    logger.warning(f"sleep 2 for forked process {num}")
    time.sleep(2)   ##  < -- with a sleep after span.finish(), the span can be collected

将上下文带入新线程

跟踪是线程的本地功能。 如果您创建了一个新线程，则上下文必须被传输到该新线程，然后再被拾取。 您可以按如下方式使用代码，将上下文带入新线程：

from instana.singletons import tracer
from threading import Thread


def child_thread_function(parent_span):
    parent_context = parent_span.get_span_context() if parent_span else None
    with tracer.start_as_current_span(
        "child_thread_span",
        span_context=parent_context
    ) as child_span:
        print("Thread offloaded work goes here")

with tracer.start_as_current_span("parent_thread_span") as parent_span:
    thread = Thread(
        target = child_thread_function,
        args = (parent_span,)
    )
    thread.start()
    thread.join()

跟踪计划稍后运行的作业

您可以按以下方式对排队等待稍后运行的作业进行设置：

# Python 3.8
import asyncio
import datetime
import uvloop
import aiohttp

from instana.singletons import tracer
from opentelemetry.trace import SpanKind


uvloop.install()


async def launch_async_calls(parent_span):
    parent_context = parent_span.get_span_context() if parent_span else None
    with tracer.start_as_current_span(
        "launch_async_calls",
        span_context=parent_context
    ):
        async with aiohttp.ClientSession() as session:
            async with session.get("https://wikipedia.org") as resp:
               print(resp.status)
               print(await resp.text())


async def run_at(dt, coro):
    await asyncio.sleep((dt - datetime.datetime.now()).total_seconds())
    return await coro


with tracer.start_as_current_span("launch_uvloop") as sync_span:
    sync_span.set_attribute("span.kind", SpanKind.SERVER)
    asyncio.run(
        run_at(
            datetime.datetime.now() + datetime.timedelta(seconds=5),
            launch_async_calls(sync_scope.span)
        )
    )

为 Instana 跨度添加自定义标记

要为 Instana span 添加自定义属性，请使用 set_attribute() 方法执行以下代码：

from instana.singletons import tracer


with tracer.start_as_current_span("span_name") as span:
    span.set_attribute("custom_attribute", "custom_value")

相关话题

• Python 概述
• Python 配置
• Python OpenTelemetry
• Python OpenTracing （弃用）
• Python 故障诊断
• Python 支持的组件和版本
• Python Github 存储库