Spaces:
Sleeping
Sleeping
File size: 2,683 Bytes
e00e744 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 |
from agents import TracingProcessor, Trace, Span
import sys
import os
from .logging_config import write_lineage_log
import secrets
import string
ALPHANUM = string.ascii_lowercase + string.digits
def log_trace_id(tag: str) -> str:
"""
Return a string of the form 'trace_<tag><random>',
where the total length after 'trace_' is 32 chars.
"""
tag += "0"
pad_len = 32 - len(tag)
random_suffix = ''.join(secrets.choice(ALPHANUM) for _ in range(pad_len))
return f"trace_{tag}{random_suffix}"
class LogTracer(TracingProcessor):
def get_name(self, trace_or_span: Trace | Span) -> str | None:
trace_id = trace_or_span.trace_id
name = trace_id.split("_")[1]
if '0' in name:
return name.split("0")[0]
else:
return None
def on_trace_start(self, trace) -> None:
name = self.get_name(trace)
if name:
write_lineage_log(name, "trace", f"Started: {trace.name}")
def on_trace_end(self, trace) -> None:
name = self.get_name(trace)
if name:
write_lineage_log(name, "trace", f"Ended: {trace.name}")
def on_span_start(self, span) -> None:
name = self.get_name(span)
type = span.span_data.type if span.span_data else "span"
if name:
message = "Started"
if span.span_data:
if span.span_data.type:
message += f" {span.span_data.type}"
if hasattr(span.span_data, "name") and span.span_data.name:
message += f" {span.span_data.name}"
if hasattr(span.span_data, "server") and span.span_data.server:
message += f" {span.span_data.server}"
if span.error:
message += f" {span.error}"
write_lineage_log(name, type, message)
def on_span_end(self, span) -> None:
name = self.get_name(span)
type = span.span_data.type if span.span_data else "span"
if name:
message = "Ended"
if span.span_data:
if span.span_data.type:
message += f" {span.span_data.type}"
if hasattr(span.span_data, "name") and span.span_data.name:
message += f" {span.span_data.name}"
if hasattr(span.span_data, "server") and span.span_data.server:
message += f" {span.span_data.server}"
if span.error:
message += f" {span.error}"
write_lineage_log(name, type, message)
def force_flush(self) -> None:
pass
def shutdown(self) -> None:
pass |