Python 追蹤 SDK
使用 Instana 進行追蹤是自動的,但如果您想要對自訂程式碼、特定應用程式區域或部分內部元件有更多可見性,則可以使用 Instana 的 Python 追蹤 SDK。
Instana Python 追蹤 SDK
Instana Python 模組提供 API 來追蹤應用程式的任何任意部分。
您可以使用 API 來檢測用於追蹤的程式碼區段,如下所示:
from instana.singletons import tracer
try:
span = tracer.start_span("mywork")
# The code to be instrumented
id = user.find_by_name('john.smith')
span.set_tag('user_id', id)
except Exception as e:
span.log_exception(e)
finally:
span.finish()
或者,您可以將環境定義管理程式與 with-as 陳述式搭配使用,這會自動擷取並記載所引發的任何異常狀況,如下所示:
from instana.singletons import tracer
with tracer.start_active_span("mywork") as scope:
# The code to be instrumented
id = user.find_by_name('john.smith')
scope.span.set_tag('user_id', id)
非同步追蹤
您要追蹤的部分作業可能非同步,這表示它們會立即傳回,但仍與主要指令序列分開繼續運作。 若要追蹤這些作業 (例如,使用 asyncio) ,您可以使用 async_tracer-related 追蹤方法,如下所示:
import asyncio
from instana.singletons import tracer, async_tracer
async def do_work(parent_span):
with async_tracer.start_active_span('launch_async_work', child_of=parent_span):
print('Work stared')
await asyncio.sleep(1)
print('Work finished!')
with tracer.start_active_span('launch_uvloop') as sync_scope:
asyncio.run(do_work(sync_scope.span))
將環境定義帶入新的執行緒
追蹤是執行緒的本端。 如果您建立新的執行緒,則必須將環境定義帶入該新執行緒,然後再挑選。 您可以檢測程式碼,以將環境定義帶至新的執行緒,如下所示:
def child_thread_function(parent_span):
tracer.start_active_span('child_thread_span', child_of=parent_span) as child_scope:
print('Thread offloaded work goes here')
tracer.start_active_span('parent_thread_span') as parent_scope:
thread = Thread(target = threaded_function, args = (parent_scope.span, ))
thread.start()
thread.join()
追蹤排定稍後執行的工作
您可以檢測已排入佇列等待稍後執行的工作,如下所示:
# Python 3.8
import asyncio
import datetime
import uvloop
import aiohttp
from instana.singletons import tracer, async_tracer
uvloop.install()
async def launch_async_calls(parent_span):
with async_tracer.start_active_span('launch_async_calls', child_of=parent_span):
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_active_span('launch_uvloop') as sync_scope:
sync_scope.span.set_tag('span.kind', 'entry')
asyncio.run(
run_at(datetime.datetime.now() + datetime.timedelta(seconds=5),
launch_async_calls(sync_scope.span)))