Dev opensearch output stores event object (#859)

* add opensearch output to ng connectors
* implement new error handling to bulk

Author: ekneg54

.vscode/settings.json

@@ -48,6 +48,7 @@
         "concatenator",
         "depseudonymization",
         "encrypter",
+        "opensearchpy",
         "PKCS",
         "pseudonymization",
         "pseudonymize",

CHANGELOG.md

@@ -6,6 +6,15 @@
 ### Features
 
 ### Improvements
+
+* make `processors` handle Event class based objects
+* add an EventBacklog class hierarchies 
+* implement an iterator interface to Input connectors
+* make simple connectors handle Event class based objects
+* make `opensearch_output` handle Event class based objects
+* deprecate `s3_output` as it does not fit into new architecture
+* deprecate `http_output` as it does not fit into new architecture
+
 ### Bugfix
 
 ## 17.0.1

doc/source/development/notebooks/new_architecture_examples/4_further_event_classes.ipynb

@@ -3,26 +3,45 @@
   {
    "cell_type": "markdown",
    "metadata": {},
-   "source": "# Specialized Event Classes: ErrorEvent, PseudonymEvent, SreEvent\n"
+   "source": [
+    "# Specialized Event Classes: ErrorEvent, PseudonymEvent, SreEvent\n"
+   ]
   },
   {
-   "metadata": {},
    "cell_type": "markdown",
-   "source": "Overview of specialized Event Classes with further data."
+   "metadata": {},
+   "source": [
+    "Overview of specialized Event Classes with further data."
+   ]
   },
   {
-   "metadata": {},
    "cell_type": "markdown",
-   "source": "**ErrorEvent**"
+   "metadata": {},
+   "source": [
+    "**ErrorEvent**"
+   ]
   },
   {
+   "cell_type": "code",
+   "execution_count": 1,
    "metadata": {
     "ExecuteTime": {
      "end_time": "2025-07-04T08:29:47.373152Z",
      "start_time": "2025-07-04T08:29:47.353933Z"
     }
    },
-   "cell_type": "code",
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "received\n",
+      "processing\n",
+      "failed\n",
+      "ErrorEvent(data={'@timestamp': '2025-07-28T06:33:20.184077+00:00', 'reason': \"Field 'data' has to be a string!\", 'original': b'data: 1', 'event': b\"{'data': 1}\"}, state=receiving)\n"
+     ]
+    }
+   ],
    "source": [
     "from logprep.ng.event.error_event import ErrorEvent\n",
     "from logprep.ng.event.log_event import LogEvent\n",
@@ -37,38 +56,43 @@
     "error_event = ErrorEvent(log_event=log_event, reason=TypeError(\"Field 'data' has to be a string!\"))\n",
     "\n",
     "print(error_event)"
-   ],
-   "outputs": [
-    {
-     "name": "stdout",
-     "output_type": "stream",
-     "text": [
-      "received\n",
-      "processing\n",
-      "failed\n",
-      "ErrorEvent(data={'@timestamp': '2025-07-04T08:29:47.370650+00:00', 'reason': \"Field 'data' has to be a string!\", 'original': b'data: 1', 'event': b\"{'data': 1}\"}, state=receiving)\n"
-     ]
-    }
-   ],
-   "execution_count": 1
+   ]
   },
   {
-   "metadata": {},
    "cell_type": "markdown",
-   "source": "**SreEvent** and **PseudonymEvent** don't have any additional data; they only serve to distinguish between the event types."
+   "metadata": {},
+   "source": [
+    "**SreEvent** and **PseudonymEvent** don't have any additional data; they only serve to distinguish between the event types."
+   ]
   },
   {
+   "cell_type": "code",
+   "execution_count": null,
    "metadata": {
     "ExecuteTime": {
      "end_time": "2025-07-04T08:32:18.392509Z",
      "start_time": "2025-07-04T08:32:18.379941Z"
     }
    },
-   "cell_type": "code",
+   "outputs": [
+    {
+     "ename": "TypeError",
+     "evalue": "ExtraDataEvent.__init__() missing 1 required keyword-only argument: 'outputs'",
+     "output_type": "error",
+     "traceback": [
+      "\u001b[31m---------------------------------------------------------------------------\u001b[39m",
+      "\u001b[31mTypeError\u001b[39m                                 Traceback (most recent call last)",
+      "\u001b[36mCell\u001b[39m\u001b[36m \u001b[39m\u001b[32mIn[2]\u001b[39m\u001b[32m, line 3\u001b[39m\n\u001b[32m      1\u001b[39m \u001b[38;5;28;01mfrom\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[34;01mlogprep\u001b[39;00m\u001b[34;01m.\u001b[39;00m\u001b[34;01mng\u001b[39;00m\u001b[34;01m.\u001b[39;00m\u001b[34;01mevent\u001b[39;00m\u001b[34;01m.\u001b[39;00m\u001b[34;01msre_event\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[38;5;28;01mimport\u001b[39;00m SreEvent\n\u001b[32m----> \u001b[39m\u001b[32m3\u001b[39m sre_event = \u001b[43mSreEvent\u001b[49m\u001b[43m(\u001b[49m\u001b[43mdata\u001b[49m\u001b[43m=\u001b[49m\u001b[43m{\u001b[49m\u001b[33;43m\"\u001b[39;49m\u001b[33;43mdata\u001b[39;49m\u001b[33;43m\"\u001b[39;49m\u001b[43m:\u001b[49m\u001b[43m \u001b[49m\u001b[32;43m1\u001b[39;49m\u001b[43m}\u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m      5\u001b[39m \u001b[38;5;66;03m# Simulate log_event states\u001b[39;00m\n\u001b[32m      6\u001b[39m \u001b[38;5;28mprint\u001b[39m(sre_event.state.next_state())  \u001b[38;5;66;03m# RECEIVED\u001b[39;00m\n",
+      "\u001b[31mTypeError\u001b[39m: ExtraDataEvent.__init__() missing 1 required keyword-only argument: 'outputs'"
+     ]
+    }
+   ],
    "source": [
     "from logprep.ng.event.sre_event import SreEvent\n",
     "\n",
-    "sre_event = SreEvent(data={\"data\": 1})\n",
+    "outputs = ({\"output\": \"output1\"}, {\"output\": \"output2\"})\n",
+    "\n",
+    "sre_event = SreEvent(data={\"data\": 1}, outputs=outputs)\n",
     "\n",
     "# Simulate log_event states\n",
     "print(sre_event.state.next_state())  # RECEIVED\n",
@@ -77,27 +101,12 @@
     "print(sre_event.state.next_state()) # STORED_IN_OUTPUT\n",
     "print(sre_event.state.next_state(success=True))  # successfully DELIVERED\n",
     "print(sre_event.state.next_state()) # ACKED"
-   ],
-   "outputs": [
-    {
-     "name": "stdout",
-     "output_type": "stream",
-     "text": [
-      "received\n",
-      "processing\n",
-      "processed\n",
-      "stored_in_output\n",
-      "delivered\n",
-      "acked\n"
-     ]
-    }
-   ],
-   "execution_count": 2
+   ]
   }
  ],
  "metadata": {
   "kernelspec": {
-   "display_name": "Python 3.11.0 ('.venv': venv)",
+   "display_name": ".venv",
    "language": "python",
    "name": "python3"
   },
@@ -111,14 +120,9 @@
    "name": "python",
    "nbconvert_exporter": "python",
    "pygments_lexer": "ipython3",
-   "version": "3.11.0"
+   "version": "3.12.3"
   },
-  "orig_nbformat": 4,
-  "vscode": {
-   "interpreter": {
-    "hash": "586280540a85d3e21edc698fe7b86af2848b9b02644e6c22463da25c40a3f1be"
-   }
-  }
+  "orig_nbformat": 4
  },
  "nbformat": 4,
  "nbformat_minor": 2