WIP

Author: kaya-david

benchmark.py

@@ -1,5 +1,4 @@
 # pylint: disable=C0103
-
 """
 Benchmark runner for logprep (logprep-ng and non-ng).
 
@@ -286,6 +285,33 @@ def opensearch_count_processed(opensearch_url: str, processed_index: str) -> int
     return int(resp.json()["count"])
 
 
+def opensearch_debug_snapshot(opensearch_url: str) -> None:
+    """
+    Print a small OpenSearch state snapshot for debugging.
+    Never raises (best-effort).
+    """
+    try:
+        r = requests.get(f"{opensearch_url}/_cat/indices?v", timeout=10)
+        print("\n--- _cat/indices ---")
+        print(r.text)
+    except Exception as e:
+        print(f"\n--- _cat/indices (failed) ---\n{e}")
+
+    try:
+        r = requests.get(f"{opensearch_url}/_cat/count?v", timeout=10)
+        print("\n--- _cat/count ---")
+        print(r.text)
+    except Exception as e:
+        print(f"\n--- _cat/count (failed) ---\n{e}")
+
+    try:
+        r = requests.get(f"{opensearch_url}/_cat/aliases?v", timeout=10)
+        print("\n--- _cat/aliases ---")
+        print(r.text)
+    except Exception as e:
+        print(f"\n--- _cat/aliases (failed) ---\n{e}")
+
+
 def reset_prometheus_dir(path: str) -> None:
     """
     Recreate PROMETHEUS_MULTIPROC_DIR.
@@ -596,6 +622,9 @@ def benchmark_run(
 
         time.sleep(sleep_after_logprep_start_s)
 
+        print("\n=== OpenSearch snapshot (before measurement) ===")
+        opensearch_debug_snapshot(opensearch_url)
+
         baseline = opensearch_count_processed(opensearch_url, processed_index)
         startup_s = time.time() - t_startup
 
@@ -610,27 +639,10 @@ def benchmark_run(
         # ensure near-real-time writes are visible to _count before measuring
         opensearch_refresh(opensearch_url, processed_index)
 
-        after = opensearch_count_processed(opensearch_url, processed_index)
-
-        def opensearch_debug_snapshot(opensearch_url: str) -> None:
-            # welche Indizes existieren überhaupt?
-            r = requests.get(f"{opensearch_url}/_cat/indices?v", timeout=10)
-            print("\n--- _cat/indices ---")
-            print(r.text)
-
-            # wie viele docs pro index? (sehr schnell, super aufschlussreich)
-            r = requests.get(f"{opensearch_url}/_cat/count?v", timeout=10)
-            print("\n--- _cat/count ---")
-            print(r.text)
-
-            # optional: aliases / data streams
-            r = requests.get(f"{opensearch_url}/_cat/aliases?v", timeout=10)
-            print("\n--- _cat/aliases ---")
-            print(r.text)
-
-        # im benchmark_run nach dem kill + refresh:
+        print("\n=== OpenSearch snapshot (after run / after refresh) ===")
         opensearch_debug_snapshot(opensearch_url)
 
+        after = opensearch_count_processed(opensearch_url, processed_index)
         processed = max(0, after - baseline)
 
         return RunResult(

logprep/ng/abc/input.py

@@ -317,7 +317,7 @@ def acknowledge(self) -> None:
         self.event_backlog.unregister(state_type=EventStateType.ACKED)
 
         for event in self.event_backlog.get(state_type=EventStateType.DELIVERED):
-            event.state.next_state()
+            event.state.current_state = EventStateType.ACKED
 
     @property
     def _add_hmac(self) -> bool:
@@ -488,7 +488,7 @@ async def get_next(self, timeout: float) -> LogEvent | None:
         )
 
         self.event_backlog.register(events=[log_event])
-        log_event.state.next_state()
+        log_event.state.current_state = EventStateType.RECEIVED
 
         return log_event
 

logprep/ng/connector/confluent_kafka/output.py

@@ -36,6 +36,7 @@
 from logprep.metrics.metrics import GaugeMetric, Metric
 from logprep.ng.abc.event import Event
 from logprep.ng.abc.output import FatalOutputError, Output
+from logprep.ng.event.event_state import EventStateType
 from logprep.util.validators import keys_in_validator
 
 DEFAULTS = {
@@ -281,7 +282,7 @@ def describe(self) -> str:
             f"{self.config.kafka_config.get('bootstrap.servers')}"
         )
 
-    def store(self, event: Event) -> None:
+    def store(self, event: Event) -> None:  # type: ignore  # TODO: fix mypy issue
         """Store a document in the producer topic.
 
         Parameters
@@ -303,7 +304,8 @@ def store_custom(self, event: Event, target: str) -> None:
         target : str
             Topic to store event data in.
         """
-        event.state.next_state()
+        event.state.current_state = EventStateType.STORING_IN_OUTPUT
+
         document = event.data
         self.metrics.number_of_processed_events += 1
         try:
@@ -358,12 +360,12 @@ def setup(self) -> None:
     def on_delivery(self, event: Event, err: KafkaException, msg: Message) -> None:
         """Callback for delivery reports."""
         if err is not None:
-            event.state.next_state(success=False)
+            event.state.current_state = EventStateType.FAILED
             event.errors.append(err)
             logger.error("Message delivery failed: %s", err)
             self.metrics.number_of_errors += 1
             return
-        event.state.next_state(success=True)
+        event.state.current_state = EventStateType.DELIVERED
         logger.debug(
             "Message delivered to '%s' partition %s, offset %s",
             msg.topic(),

logprep/ng/connector/opensearch/output.py

@@ -51,6 +51,7 @@
 from logprep.metrics.metrics import Metric
 from logprep.ng.abc.event import Event
 from logprep.ng.abc.output import Output
+from logprep.ng.event.event_state import EventStateType
 
 logger = logging.getLogger("OpenSearchOutput")
 
@@ -281,7 +282,7 @@ async def store_custom(self, event: Event, target: str) -> None:
         target : str
             Index to store the document in.
         """
-        event.state.next_state()
+        event.state.current_state = EventStateType.STORING_IN_OUTPUT
         document = event.data
         document["_index"] = target
         document["_op_type"] = document.get("_op_type", self.config.default_op_type)
@@ -396,7 +397,7 @@ async def _bulk(self, client: AsyncOpenSearch, events: list["Event"]) -> None:
             except OpenSearchException as e:
                 # whole bulk request failed → mark all events failed
                 for ev in batch:
-                    ev.state.next_state(success=False)
+                    ev.state.current_state = EventStateType.FAILED
                     ev.errors.append(BulkError("Bulk request failed", exception=str(e)))
                 continue
 
@@ -417,7 +418,7 @@ async def _bulk(self, client: AsyncOpenSearch, events: list["Event"]) -> None:
 
                 ok = isinstance(status, int) and 200 <= status < 300 and not error_obj
                 if ok:
-                    ev.state.next_state(success=True)
+                    ev.state.current_state = EventStateType.STORED_IN_OUTPUT
                     continue
 
                 # normalize error into your BulkError shape
@@ -427,7 +428,7 @@ async def _bulk(self, client: AsyncOpenSearch, events: list["Event"]) -> None:
                 else:
                     message = str(error_obj) if error_obj else "Failed to index document"
 
-                ev.state.next_state(success=False)
+                ev.state.current_state = EventStateType.FAILED
                 ev.errors.append(
                     BulkError(
                         message,

logprep/ng/event/event_state.py

@@ -19,16 +19,26 @@ class EventStateType(StrEnum):
     PROCESSED = "processed"
     """The event has been processed by all pipeline processors."""
 
+    STORING_IN_OUTPUT = "storing_in_output"
+    """The event is storing in the output connector."""
+
     STORED_IN_OUTPUT = "stored_in_output"
     """The event was successfully stored in the output connector."""
 
     FAILED = "failed"
     """The event failed during processing or output storage."""
 
+    STORING_IN_ERROR = "storing_in_error"
+    """The event is storing in the error output (e.g. error queue or
+    fallback output)."""
+
     STORED_IN_ERROR = "stored_in_error"
     """The event was stored in the error output (e.g. error queue or
     fallback output)."""
 
+    DELIVERING = "delivering"
+    """The event is delivering to the target system or final destination."""
+
     DELIVERED = "delivered"
     """The event was delivered to the target system or final destination."""
 

logprep/ng/pipeline.py

@@ -3,6 +3,7 @@
 import logging
 
 from logprep.ng.abc.processor import Processor
+from logprep.ng.event.event_state import EventStateType
 from logprep.ng.event.log_event import LogEvent
 
 logger = logging.getLogger("Pipeline")
@@ -12,15 +13,15 @@ def _process_event(event: LogEvent | None, processors: list[Processor]) -> LogEv
     """process all processors for one event"""
     if event is None or not event.data:
         raise ValueError("no event given")
-    event.state.next_state()
+    event.state.current_state = EventStateType.PROCESSING
     for processor in processors:
         if not event.data:
             break
         processor.process(event)
     if not event.errors:
-        event.state.next_state(success=True)
+        event.state.current_state = EventStateType.PROCESSED
     else:
-        event.state.next_state(success=False)
+        event.state.current_state = EventStateType.FAILED
         logger.error("event failed: %s with errors: %s", event, event.errors)
     return event
 

logprep/ng/sender.py

@@ -68,11 +68,34 @@ async def _send_and_flush_processed_events(self, batch_events: list[LogEvent]) -
         if not processed:
             return
 
-        # send in parallel (minimal change vs. serial list comprehension)
-        await asyncio.gather(*(self._send_processed(event) for event in processed))
+        # send in parallel
+        try:
+            results = await asyncio.gather(
+                *(self._send_processed(event) for event in processed),
+                return_exceptions=True,
+            )
+            for r in results:
+                if isinstance(r, Exception):
+                    logger.exception("Error while sending processed event", exc_info=r)
+
+        finally:
+            for output in self._outputs.values():
+                try:
+                    await output.flush()
+                except Exception as e:
+                    logger.exception("Error while flushing output %s", output.name, exc_info=e)
 
         # flush once per output after sending
-        await asyncio.gather(*(output.flush() for output in self._outputs.values()))
+        try:
+            results = await asyncio.gather(
+                *(output.flush() for output in self._outputs.values()),
+                return_exceptions=True,
+            )
+            for r in results:
+                if isinstance(r, Exception):
+                    logger.exception("Error during final output flush", exc_info=r)
+        except Exception as e:
+            logger.exception("Unexpected error during final output flush", exc_info=e)
 
     async def _send_extra_data(self, event: LogEvent) -> None:
         extra_data_events = typing.cast(list[ExtraDataEvent], event.extra_data)