Skip to content

symfonic.platform.telemetry_sink

telemetry_sink

Run-scoped observability binding for the kernel event fan-out.

TelemetryEventSink 

TelemetryEventSink(collector: Any, sinks: tuple[Any, ...], tenant: str)

Deliver events and bind safe attribution before the first one.

Source code in src/symfonic/platform/telemetry_sink.py 
16
17
18
19
def __init__(self, collector: Any, sinks: tuple[Any, ...], tenant: str) -> None:
    self._collector = collector
    self._sinks = sinks
    self._tenant = tenant

bind_run 

bind_run(run_id: str, *, root_run_id: str, parent_run_id: str | None, session_id: str = '') -> None

Bind only durable identifiers and attribution, never turn content.

Source code in src/symfonic/platform/telemetry_sink.py 
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
def bind_run(
    self,
    run_id: str,
    *,
    root_run_id: str,
    parent_run_id: str | None,
    session_id: str = "",
) -> None:
    """Bind only durable identifiers and attribution, never turn content."""
    lineage = getattr(self._collector, "set_lineage", None)
    if callable(lineage):
        lineage(run_id, root_run_id=root_run_id, parent_run_id=parent_run_id)
    tenant = getattr(self._collector, "set_tenant", None)
    if self._tenant and callable(tenant):
        tenant(run_id, self._tenant)
    conversation = getattr(self._collector, "set_conversation", None)
    if callable(conversation):
        conversation(run_id, session_id or run_id)
    for sink in self._sinks:
        bind = getattr(sink, "bind_run", None)
        if callable(bind):
            try:
                bind(
                    run_id,
                    root_run_id=root_run_id,
                    parent_run_id=parent_run_id,
                    tenant_id=self._tenant,
                    session_id=session_id,
                )
            except TypeError:
                bind(run_id, tenant_id=self._tenant, session_id=session_id)