Misc cleanup of ProcessingExporter

mpenn · mpenn · commit 256f8de4bc83 · 2025-09-04T22:01:09.000-04:00
Signed-off-by: Matthew Penn &lt;mpenn@nvidia.com&gt;
diff --git a/src/nat/observability/exporter/processing_exporter.py b/src/nat/observability/exporter/processing_exporter.py
@@ -220,7 +220,7 @@ def get_processor_by_name(self, name: str) -> Processor | None:
         if name in self._processor_names:
             position = self._processor_names[name]
             return self._processors[position]
-        logger.warning("Processor '%s' not found in pipeline", name)
+        logger.debug("Processor '%s' not found in pipeline", name)
         return None
 
     def _check_pipeline_locked(self) -> None:
@@ -343,9 +343,8 @@ def _check_processor_compatibility(self,
                 target_type)
 
     async def _pre_start(self) -> None:
-        # Lock the pipeline to prevent further modifications
-        self._pipeline_locked = True
 
+        # Validate that the pipeline is compatible with the exporter
         if len(self._processors) > 0:
             first_processor = self._processors[0]
             last_processor = self._processors[-1]
@@ -382,14 +381,17 @@ async def _pre_start(self) -> None:
                     self.output_type,
                     e)
 
+        # Lock the pipeline to prevent further modifications
+        self._pipeline_locked = True
+
     async def _process_pipeline(self, item: PipelineInputT) -> PipelineOutputT | None:
         """Process item through all registered processors.
 
         Args:
             item (PipelineInputT): The item to process (starts as PipelineInputT, can transform to PipelineOutputT)
 
         Returns:
-            PipelineOutputT: The processed item after running through all processors
+            PipelineOutputT | None: The processed item after running through all processors
         """
         return await self._process_through_processors(self._processors, item)  # type: ignore
 
@@ -529,7 +531,7 @@ async def export_processed(self, item: PipelineOutputT | list[PipelineOutputT])
         """
         pass
 
-    def _create_export_task(self, coro: Coroutine):
+    def _create_export_task(self, coro: Coroutine) -> None:
         """Create task with minimal overhead but proper tracking.
 
         Args:
@@ -549,7 +551,7 @@ def _create_export_task(self, coro: Coroutine):
             raise
 
     @override
-    async def _cleanup(self):
+    async def _cleanup(self) -> None:
         """Enhanced cleanup that shuts down all shutdown-aware processors.
 
         Each processor is responsible for its own cleanup, including routing
diff --git a/tests/nat/observability/exporter/test_processing_exporter.py b/tests/nat/observability/exporter/test_processing_exporter.py
@@ -394,7 +394,7 @@ def test_get_processor_by_name_exists(self, processing_exporter):
 
     def test_get_processor_by_name_not_exists(self, processing_exporter, caplog):
         """Test getting processor by name when it doesn't exist."""
-        with caplog.at_level(logging.WARNING):
+        with caplog.at_level(logging.DEBUG):
             retrieved = processing_exporter.get_processor_by_name("nonexistent")
 
         assert retrieved is None
