DTT2 - Workflow engine - Implement observability plugin execution ID

[QU3B1M]

Epic: https://github.com/wazuh/wazuh-qa/issues/4837

---

Description

The objective of this issue is to investigate and analyze the implementation of an execution_id as identifier of the workflow execution in order to improve the tracking for the Observability module

Tasks

• [x] Investigate how to implement an unique ID for the execution.
• [x] Implement the ID generation in the engine.
• [x] Implement auto-injection of the ID in the workflow as global variable (to call this id inside the yaml without having it defined explicitly).
• [x] Update the influxdb plugin to use this execution ID as primary key.

[QU3B1M]

• Defined usage UUID as unique identifier for the worklow execution
• Applied required changes to configure this execution_id as global variable in the worklfow

[QU3B1M]

• Implement 'native_vars' dict in the workflow file to store all the engine's global/magic variables. deployability/modules/workflow_engine/workflow_processor.py

  class WorkflowFile:
      ...
      def __init__(self, workflow_file_path: Path | str, schema_path: Path | str = None, native_vars: dict = {}) -> None:
          ...
          self.__native_vars = native_vars
          ...
      def __process_workflow(self):
          ...
          workflow_vars = self.workflow_raw_data.get('variables', {})
          variables = {**workflow_vars, **self.__native_vars}
          for task in self.workflow_raw_data.get('tasks', []):
              task_collection.extend(self.__expand_task(task, variables))
          ...

• Update WorkflowProcessor to generate the execution_id and inject the native vars into the other classes (for now only to WorkflowFile) deployability/modules/workflow_engine/workflow_processor.py

  class WorkflowProcessor:
      ...
      def __init__(self, workflow_file: str, dry_run: bool, threads: int, log_level: str = 'INFO', schema_file: Path | str = None):
          ...
          # Initialize the intenal variables.
          self.__execution_id = self.__generate_execution_id()
          # Initialize the public variables.
          self.task_collection = WorkflowFile(workflow_file, schema_file, self.get_native_vars()).task_collection
          ...
      def get_native_vars(self) -> dict:
          return {'execution_id': self.__execution_id}
  
      def __generate_execution_id(self) -> str:
          return str(uuid.uuid4())
  

• Update InfluxDBReporter to use the execution_id if defined or generate a new one. deployability/plugins/influxdb_reporter/pytest_influxdb/reporter.py

  class InfluxDBReporter:
      ...
      def __init__(self, config: Config, config_file: str = None) -> None:
          ...
          self.execution_id: str = config.getoption('--execution-id') or str(uuid.uuid4())
          ...
  
      def __get_report_body(self, test_report: TestReport, datetime: str) -> dict:
          ...
          tags = {
              'execution_id': self.execution_id,
              'test': test_report.fspath,
              'markers': self.__get_pytest_marks(test_report.keywords),
              'when': test_report.when,
          }
          ...

[QU3B1M]

• The native/magic vars are also configured as environment vars to each execution, this way any module/script can access this values by checking the Env vars.

• A new task tag env was implemented in the workflow structure, this tag lets the user set environment variables for the execution of the task. Example:

    - task: "test-magic-vars"
      description: "Use a simple script to the new magic & env variables."
      do:
        this: process
        with:
          path: python3
          args:
            - /home/quebim/Wazuh/repos/wazuh-qa/deployability/test.py
          env:
            TEST: "Im testing"
  

[pro-akim]

Review Notes

LGTM