SDK de traçage Python

Le module Instana Python fournit une API basée sur OpenTelemetry API permettant de tracer n'importe quelle partie de votre application.

Vous pouvez instrumenter une section de code à des fins de traçage à l'aide d'une balise ` API ` comme suit :

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()
 

Vous pouvez également utiliser le gestionnaire de contexte avec l'instruction with-as , qui capture et consigne automatiquement toutes les exceptions émises, comme suit:

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)
 

Certaines opérations que vous souhaitez suivre peuvent être asynchrones, ce qui signifie qu'elles renvoient un résultat immédiatement mais continuent à s'exécuter indépendamment de la séquence principale d'instructions. Pour tracer ces opérations, par exemple avec la bibliothèque asyncio, vous pouvez utiliser les méthodes de traçage suivantes :

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))
 

Si vous souhaitez suivre les intervalles de temps dans un processus dérivé, ajoutez un time.sleep() après span.end(). Voir l'exemple suivant :

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
 

Le traçage est local pour une unité d'exécution. Si vous créez une nouvelle unité d'exécution, le contexte doit être transporté vers cette nouvelle unité d'exécution, puis récupéré. Vous pouvez instrumenter le code pour transmettre le contexte à de nouvelles unités d'exécution comme suit:

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()
 

Vous pouvez instrumenter les travaux qui sont mis en file d'attente pour une exécution ultérieure comme suit:

# 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)
        )
    )
 

Pour ajouter des attributs personnalisés à un élément `span` d' Instana, intégrez le code suivant à l'aide de la set_attribute() méthode :

from instana.singletons import tracer


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

Pour définir explicitement le type d'un élément Instana, définissez l'attribut personnalisé span.kind comme suit :

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


with tracer.start_as_current_span("span_name") as span:
    span.set_attribute("span.kind", <value>)

Remplacez <value> par l'une des valeurs autorisées suivantes :

• Python aperçu
• Configuration Python
• Python OpenTelemetry
• Python OpenTracing (obsolète)
• Traitement des incidents liés à Python
• Python composants et versions pris en charge
• RéférentielPython GitHub