DTT1 - Iteration 2 - QA Workflow Engine - Implementation

.gitignore

@@ -69,3 +69,6 @@ ansible.log
 
 # Ssh keys
 key
+
+#Python
+__pycache__/

deployability/launchers/workflow_engine.py

@@ -5,24 +5,33 @@
 import os
 import sys
 import argparse
+import logging
+import colorlog
 project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))
 sys.path.append(project_root)
 from modules.workflow_engine.workflow_processor import WorkflowProcessor
 from modules import SchemaValidator
 
 
+
 def parse_arguments() -> argparse.Namespace:
     """Parse command line arguments."""
     parser = argparse.ArgumentParser(description='Execute tasks in a workflow.')
     parser.add_argument('workflow_file', type=str, help='Path to the workflow file (YAML format).')
     parser.add_argument('schema_file', type=str, default="./schema.json", help='Path to the schema definition file.')
     parser.add_argument('--threads', type=int, default=1, help='Number of threads to use for parallel execution.')
     parser.add_argument('--dry-run', action='store_true', help='Display the plan without executing tasks.')
-    parser.add_argument('--log-format', choices=['plain', 'json'], default='plain', help='Log format (plain or json).')
     parser.add_argument('--log-level', choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'], default='INFO',
                         help='Log level.')
     return parser.parse_args()
 
+def setup_logger(log_level: str) -> None:
+    """Setup logger."""
+    logger = logging.getLogger()
+    console_handler = colorlog.StreamHandler()
+    console_handler.setFormatter(colorlog.ColoredFormatter("%(log_color)s[%(asctime)s] [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S"))
+    logger.addHandler(console_handler)
+    logger.setLevel(log_level)
 
 def main() -> None:
     """Main entry point."""
@@ -31,10 +40,9 @@ def main() -> None:
     validator = SchemaValidator(args.schema_file, args.workflow_file)
     validator.preprocess_data()
     validator.validateSchema()
-
+    setup_logger(args.log_level)
     processor = WorkflowProcessor(args.workflow_file, args.dry_run, args.threads)
-    processor.logger = processor.setup_logger(log_format=args.log_format, log_level=args.log_level)
-    processor.main()
+    processor.run()
 
 
 if __name__ == "__main__":

deployability/modules/generic/schemaValidator.py

@@ -41,7 +41,6 @@ def preprocess_data(self):
     def validateSchema(self):
         try:
             jsonschema.validate(self.yamlData, self.schemaData)
-            print("YAML is valid!")
         except jsonschema.exceptions.ValidationError as e:
             print(f"Validation error: {e}")
         except Exception as e:

deployability/modules/workflow_engine/examples/dtt1-agents.yaml

@@ -35,78 +35,85 @@ variables:
 
 tasks:
   # Generic agent test task
-  - task: "run-agent-tests-{agent}"
+  - task: "test-agent-{agent}"
     description: "Run tests for the {agent} agent."
     do:
       this: process
       with:
         path: /bin/echo
         args:
+          - -n
           - "Running tests for {agent}"
     depends-on:
-      - "provision-{agent}"
-      - "provision-manager"
+      - "provision-agent-{agent}"
     foreach:
       - variable: agents-os
         as: agent
 
   # Unique manager provision task
-  - task: "provision-manager"
+  - task: "provision-manager-{manager-os}"
     description: "Provision the manager."
     do:
       this: process
       with:
         path: /bin/echo
         args:
+          - -n
           - "Running provision for manager"
     depends-on:
-      - "allocate-manager"
+      - "allocate-manager-{manager-os}"
 
   # Unique manager allocate task
-  - task: "allocate-manager"
+  - task: "allocate-manager-{manager-os}"
     description: "Allocate resources for the manager."
     do:
       this: process
       with:
         path: /bin/echo
         args:
+          - -n
           - "Running allocate for manager"
     cleanup:
       this: process
       with:
         path: /bin/echo
         args:
+          - -n
           - "Running cleanup for manager"
 
   # Generic agent provision task
-  - task: "provision-{agent}"
+  - task: "provision-agent-{agent}"
     description: "Provision resources for the {agent} agent."
     do:
       this: process
       with:
         path: /bin/echo
         args:
+          - -n
           - "Running provision for {agent}"
     depends-on:
-      - "allocate-{agent}"
+      - "allocate-agent-{agent}"
+      - "provision-manager-{manager-os}"
     foreach:
       - variable: agents-os
         as: agent
 
   # Generic agent allocate task
-  - task: "allocate-{agent}"
+  - task: "allocate-agent-{agent}"
     description: "Allocate resources for the {agent} agent."
     do:
       this: process
       with:
         path: /bin/echo
         args:
+          - -n
           - "Running allocate for {agent}"
     cleanup:
       this: process
       with:
         path: /bin/echo
         args:
+          - -n
           - "Running cleanup for allocate for {agent}"
     foreach:
       - variable: agents-os

deployability/modules/workflow_engine/task.py

@@ -8,6 +8,7 @@
 import random
 import time
 
+logger = (lambda: logging.getLogger())()
 
 class Task(ABC):
     """Abstract base class for tasks."""
@@ -21,7 +22,7 @@ def execute(self) -> None:
 class ProcessTask(Task):
     """Task for executing a process."""
 
-    def __init__(self, task_name: str, task_parameters: dict, logger: logging.Logger):
+    def __init__(self, task_name: str, task_parameters: dict):
         """
         Initialize ProcessTask.
 
@@ -44,6 +45,14 @@ def format_key_value(task_arg):
 
         task_args = [str(task_arg) if isinstance(task_arg, str) else format_key_value(task_arg) for task_arg in self.task_parameters['args']]
 
+<<<<<<< HEAD:poc-tests/modules/workflow_engine/task.py
+        result = subprocess.run(
+            [self.task_parameters['path']] + task_args,
+            check=True,
+            capture_output=True,
+            text=True,
+        )
+=======
         try:
             self.logger.info("ejecutando task ")
             self.logger.info(f"{[self.task_parameters['path']]+ task_args}")
@@ -53,41 +62,39 @@ def format_key_value(task_arg):
                 capture_output=True,
                 text=True,
             )
+>>>>>>> enhancement/4751-dtt1-iteration-2-poc:deployability/modules/workflow_engine/task.py
 
-            self.logger.info("Output:\n%s", result.stdout, extra={'tag': self.task_name})
-
-            if result.returncode != 0:
-                raise subprocess.CalledProcessError(returncode=result.returncode, cmd=result.args, output=result.stdout)
-
-        except Exception as e:
-            self.logger.error("Task failed with error: %s", e, extra={'tag': self.task_name})
-            # Handle the exception or re-raise if necessary
-            raise
+        if result.returncode != 0:
+            raise subprocess.CalledProcessError(returncode=result.returncode, cmd=result.args, output=result.stdout)
 
 
 class DummyTask(Task):
-    def __init__(self, task_name, task_parameters, logger: logging.Logger):
+    def __init__(self, task_name, task_parameters):
         self.task_name = task_name
         self.task_parameters = task_parameters
-        self.logger = logger
 
     def execute(self):
         message = self.task_parameters.get('message', 'No message provided')
-        self.logger.info("%s: %s", message, self.task_name, extra={'tag': self.task_name})
+        logger.info("%s: %s", message, self.task_name, extra={'tag': self.task_name})
 
 
 class DummyRandomTask(Task):
-    def __init__(self, task_name, task_parameters, logger: logging.Logger):
+    def __init__(self, task_name, task_parameters):
         self.task_name = task_name
         self.task_parameters = task_parameters
-        self.logger = logger
 
     def execute(self):
         time_interval = self.task_parameters.get('time-seconds', [1, 5])
         sleep_time = random.uniform(time_interval[0], time_interval[1])
 
         message = self.task_parameters.get('message', 'No message provided')
-        self.logger.info("%s: %s (Sleeping for %.2f seconds)", message, self.task_name, sleep_time, extra={'tag': self.task_name})
+        logger.info("%s: %s (Sleeping for %.2f seconds)", message, self.task_name, sleep_time, extra={'tag': self.task_name})
 
         time.sleep(sleep_time)
 
+
+TASKS_HANDLERS = {
+    'process': ProcessTask,
+    'dummy': DummyTask,
+    'dummy-random': DummyRandomTask,
+}