fix(observability): close each telemetry exporter when the builder exits, draining what its per-run copies shared

Signed-off-by: Somay-kousis <somaykaush@gmail.com>

Author: Somay-kousis

packages/nvidia_nat_core/src/nat/builder/workflow_builder.py

@@ -1402,6 +1402,9 @@ async def add_telemetry_exporter(self, name: str, config: TelemetryExporterBaseC
 
             # Only protect the shared state modifications (serialized)
             exporter = await self._get_exit_stack().enter_async_context(exporter_context_manager)
+            if isinstance(exporter, BaseExporter):
+                # runs use per-run copies of the exporter, so this instance may never be started or stopped
+                self._get_exit_stack().push_async_callback(exporter.close)
 
         self._telemetry_exporters[name] = ConfiguredTelemetryExporter(config=config, instance=exporter)
 

packages/nvidia_nat_core/src/nat/observability/exporter/base_exporter.py

@@ -377,6 +377,17 @@ async def wait_for_tasks(self, timeout: float = 5.0):
         except Exception as e:
             logger.exception("%s: Error while waiting for tasks: %s", self.name, e)
 
+    async def close(self):
+        """Release what this exporter shares with its per-run copies, whether or not it was started itself.
+
+        Per-run copies share the processors of this instance, so this is where a batcher's remaining items are
+        exported for the last time. Safe to call more than once.
+        """
+        if self._running:
+            await self.stop()
+        else:
+            await self._cleanup()
+
     @override
     async def stop(self):
         """Stop the exporter immediately without waiting for background tasks.