Allow providing the elastic search index in the configuration

Signed-off-by: Pedro Algarvio <[email protected]>

Author: s0undt3ch

src/saf/forward/elasticsearch.py

@@ -17,6 +17,7 @@
 from typing import List
 from typing import Optional
 from typing import Type
+from typing import cast
 
 from pydantic import Field
 
@@ -74,6 +75,7 @@ class ElasticSearchConfig(ForwardConfigBase):
     username: Optional[str] = None
     password: Optional[str] = None
     timeout: int = 10
+    index: Optional[str] = None
 
 
 def get_config_schema() -> Type[ElasticSearchConfig]:
@@ -86,7 +88,7 @@ def get_config_schema() -> Type[ElasticSearchConfig]:
 async def forward(
     *,
     ctx: PipelineRunContext[ElasticSearchConfig],
-    event: ElasticSearchEvent,
+    event: ElasticSearchEvent | CollectedEvent,
 ) -> None:
     """
     Method called to forward the event.
@@ -106,8 +108,31 @@ async def forward(
         connection_info = await client.info()
         log.warning("ES Connection Info:\n%s", pprint.pformat(connection_info))
     client = ctx.cache["client"]
-    data = event.data.copy()
+    if isinstance(event, ElasticSearchEvent):
+        es_id = event.id
+        es_index = event.index
+    else:
+        es_id = str(uuid.uuid4())
+        if ctx.config.index is None:
+            msg = (
+                "Event to forward is not an instance of ElasticSearchEvent and the "
+                "elasticsearch plugin index configuration is not defined."
+            )
+            raise RuntimeError(msg)
+        es_index = ctx.config.index
+    data = cast(Dict[str, Any], event.data).copy()
     if "@timestamp" not in data:
         data["@timestamp"] = event.timestamp
-    ret = await client.index(index=event.index, id=event.id, body=data)
-    log.warning("ES SEND:\n%s", pprint.pformat(ret))
+    log.debug(
+        "ElasticSearch Forward Details:\nindex: %s\nid: %s\ndata:\n%s",
+        es_index,
+        es_id,
+        pprint.pformat(data),
+    )
+    ret = await client.index(index=es_index, id=es_id, body=data)
+    log.debug(
+        "ElasticSearch Forward Response(index: %s; id: %s):\n%s",
+        es_index,
+        es_id,
+        pprint.pformat(ret),
+    )