update debug logs

Author: mhoff

logprep/ng/sender.py

@@ -34,6 +34,7 @@ def __init__(
         self._error_output = error_output
 
     async def process(self, batch: list[LogEvent]) -> list[LogEvent]:
+        logger.debug("Receiving event from worker: %d", len(batch))
         await self._send_and_flush_processed_events(batch_events=batch)
         if self._error_output:
             await self._send_and_flush_failed_events(batch_events=batch)
@@ -87,6 +88,7 @@ async def _send_and_flush_processed_events(self, batch_events: list[LogEvent]) -
 
         # flush once per output after sending
         try:
+            logger.debug("Flushing all outputs after sending %d events", len(batch_events))
             results = await asyncio.gather(
                 *(output.flush() for output in self._outputs.values()),
                 return_exceptions=True,

logprep/ng/util/worker/worker.py

@@ -88,10 +88,8 @@ async def _flush_after_interval(self) -> None:
         if the batch has not already been drained by the size trigger.
         """
         try:
-            logger.debug("timer sleeping")
             await asyncio.sleep(self._batch_interval_s)
         except asyncio.CancelledError:
-            logger.debug("timer caught cancelled error")
             return
 
         batch: list[Input] | None = None
@@ -102,6 +100,7 @@ async def _flush_after_interval(self) -> None:
                 self._flush_timer = None
 
         if batch:
+            logger.debug("Flushing messages based on timer")
             await self._flush_batch(batch)
 
     def _drain_locked(self) -> list[Input]:
@@ -136,6 +135,9 @@ async def add(self, item: Input) -> None:
                 batch_to_flush = self._drain_locked()
 
         if batch_to_flush:
+            logger.debug("Flushing messages based on backlog size")
+            logger.debug("Remaining items in _batch_buffer: %d", len(self._batch_buffer))
+            logger.debug("Batch size to flush after: %d", self._batch_size)
             await self._flush_batch(batch_to_flush)
 
     async def flush(self) -> None:
@@ -150,6 +152,7 @@ async def flush(self) -> None:
             if self._batch_buffer:
                 batch_to_flush = self._drain_locked()
         if batch_to_flush:
+            logger.debug("Flushing messages based on manual trigger")
             await self._flush_batch(batch_to_flush)
 
     async def _process_batch(self, batch: list[Input]) -> list[Output]: