Add a jobs_to_es processor

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

Author: s0undt3ch

examples/setup.cfg

@@ -49,6 +49,7 @@ saf.process =
   notebook_output = saltext.safexamples.process.notebook_output
   numpy_save_keys = saltext.safexamples.process.numpy_save_keys
   beacons_to_es = saltext.safexamples.process.beacons_to_es
+  jobs_to_es = saltext.safexamples.process.jobs_to_es
 
 [bdist_wheel]
 # Use this option if your package is pure-python

examples/src/saltext/safexamples/process/beacons_to_es.py

@@ -6,7 +6,7 @@
 from __future__ import annotations
 
 import logging
-import uuid
+import pprint
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import AsyncIterator
@@ -45,7 +45,6 @@ async def process(
     """
     data = cast(dict[str, Any], event.data).copy()
     data["role"] = ctx.info.salt.role
-    if TYPE_CHECKING:
-        assert event.timestamp
-    data["@timestamp"] = event.timestamp.isoformat()
-    yield ElasticSearchEvent.construct(index=event.beacon, id=str(uuid.uuid4()), data=data)
+    evt = ElasticSearchEvent.construct(index=event.beacon, data=data)
+    log.debug("ElasticSearchEvent: %s", pprint.pformat(evt.dict()))
+    yield evt

examples/src/saltext/safexamples/process/jobs_to_es.py

@@ -0,0 +1,55 @@
+# Copyright 2021-2023 VMware, Inc.
+# SPDX-License-Identifier: Apache-2.0
+"""
+Convert beacon event to an elastic search index event.
+"""
+from __future__ import annotations
+
+import json
+import logging
+import pprint
+from typing import TYPE_CHECKING
+from typing import AsyncIterator
+from typing import Type
+
+from saf.forward.elasticsearch import ElasticSearchEvent
+from saf.models import PipelineRunContext
+from saf.models import ProcessConfigBase
+
+if TYPE_CHECKING:
+    from saf.process.job_aggregate import JobAggregateCollectedEvent
+
+log = logging.getLogger(__name__)
+
+
+class SaltJobsToESConfig(ProcessConfigBase):
+    """
+    Processor configuration.
+    """
+
+
+def get_config_schema() -> Type[SaltJobsToESConfig]:
+    """
+    Get the test collect plugin configuration schema.
+    """
+    return SaltJobsToESConfig
+
+
+async def process(
+    *,
+    ctx: PipelineRunContext[SaltJobsToESConfig],  # noqa: ARG001
+    event: JobAggregateCollectedEvent,
+) -> AsyncIterator[ElasticSearchEvent]:
+    """
+    Method called to collect events, in this case, generate.
+    """
+    data = event.dict()
+    data.pop("data", None)
+    data.update(event.data)
+    # Have the return field always be a JSON string
+    if "return" in data:
+        data["return"] = json.dumps(data["return"])
+    data["@timestamp"] = event.start_time
+    evt = ElasticSearchEvent.construct(index="salt_jobs", data=data)
+    log.debug("ElasticSearchEvent: %s", pprint.pformat(evt.dict()))
+    yield evt