|
1 | 1 | from __future__ import annotations |
2 | 2 |
|
| 3 | +import asyncio |
| 4 | +import logging |
3 | 5 | from typing import TYPE_CHECKING |
4 | 6 | from threading import Lock |
5 | 7 |
|
@@ -78,3 +80,72 @@ def get_sync_tracing_processors(): |
78 | 80 |
|
79 | 81 | def get_async_tracing_processors(): |
80 | 82 | return GLOBAL_TRACING_PROCESSOR_MANAGER.get_async_processors() |
| 83 | + |
| 84 | + |
| 85 | +_logger = logging.getLogger(__name__) |
| 86 | + |
| 87 | +# Total wall-clock budget for draining every sync tracing processor. A pod's |
| 88 | +# terminationGracePeriodSeconds (30s by default) is shared with the OTel flush that |
| 89 | +# follows this, so the drain takes a small slice of it. |
| 90 | +SYNC_TRACING_SHUTDOWN_BUDGET_S = 5.0 |
| 91 | + |
| 92 | + |
| 93 | +async def shutdown_sync_tracing_processors( |
| 94 | + budget_s: float = SYNC_TRACING_SHUTDOWN_BUDGET_S, |
| 95 | +) -> None: |
| 96 | + """Drain the sync tracing processors' queues at shutdown. Never raises. |
| 97 | +
|
| 98 | + Nothing used to call this. The ACP lifespan drained ``shutdown_default_span_queue``, |
| 99 | + which is the ASYNC path only, so a sync agent dropped whatever business spans were |
| 100 | + still queued when the pod stopped. That matters beyond the lost spans: the business |
| 101 | + span is what an obs span's ``agentex.business_trace_id`` resolves to, so losing it |
| 102 | + breaks the pivot from Tempo back to the SGP store. |
| 103 | +
|
| 104 | + Off the event loop and on a deadline, both deliberately. |
| 105 | + ``SGPSyncTracingProcessor.shutdown`` calls ``flush_queue()``, a BLOCKING HTTP flush |
| 106 | + with retries. Calling it inline would stall the caller's loop, so a slow or |
| 107 | + unreachable collector could burn the whole termination grace period and stop the |
| 108 | + OTel flush that runs after it — trading a few business spans for all of the OTel |
| 109 | + ones. Each processor runs in a worker thread, and the budget is shared across all |
| 110 | + of them so one stalled export cannot starve the rest. |
| 111 | +
|
| 112 | + A timed-out flush leaks its thread until the process exits. Accepted: this runs |
| 113 | + only during shutdown, and the alternative is blocking on it. |
| 114 | +
|
| 115 | + Lives here rather than in the ACP server because both entry points need it — the |
| 116 | + ACP server AND the Temporal worker, which runs in its own process. |
| 117 | + """ |
| 118 | + try: |
| 119 | + processors = get_sync_tracing_processors() |
| 120 | + except Exception: # pragma: no cover - nothing to drain |
| 121 | + _logger.debug("sync tracing processors unavailable at shutdown", exc_info=True) |
| 122 | + return |
| 123 | + |
| 124 | + loop = asyncio.get_running_loop() |
| 125 | + deadline = loop.time() + budget_s |
| 126 | + |
| 127 | + for processor in processors: |
| 128 | + remaining = deadline - loop.time() |
| 129 | + if remaining <= 0: |
| 130 | + _logger.warning( |
| 131 | + "sync tracing shutdown budget of %.1fs exhausted; %s and any after it " |
| 132 | + "were not flushed and their business spans are lost", |
| 133 | + budget_s, |
| 134 | + type(processor).__name__, |
| 135 | + ) |
| 136 | + break |
| 137 | + try: |
| 138 | + await asyncio.wait_for(asyncio.to_thread(processor.shutdown), remaining) |
| 139 | + except (TimeoutError, asyncio.TimeoutError): |
| 140 | + _logger.warning( |
| 141 | + "%s did not flush within the remaining %.1fs; its business spans are " |
| 142 | + "lost, but shutdown continues", |
| 143 | + type(processor).__name__, |
| 144 | + remaining, |
| 145 | + ) |
| 146 | + except Exception: # noqa: PERF203 - one bad processor must not block the rest |
| 147 | + _logger.warning( |
| 148 | + "a sync tracing processor failed to flush on shutdown; " |
| 149 | + "some business spans may be lost", |
| 150 | + exc_info=True, |
| 151 | + ) |
0 commit comments