Merge pull request #10 from 5genesis/rest_task

v3.0.1

.gitignore

@@ -101,7 +101,9 @@ ENV/
 Persistence/
 config.yml
 facility.yml
+evolved5g.yml
 *.log.*
+.flaskenv
 TestCases/
 UEs/
 Resources/

CHANGELOG.md

@@ -1,3 +1,9 @@
+**05/11/2021** [Version 3.0.1]
+
+ - Implement RestApi, JenkinsJob, JenkinsStatus tasks
+ - Add separate EVOLVED-5G configuration file
+ - Allow checking the status of finished experiments
+
 **11/10/2021** [Version 3.0.0]
 
  - Initial EVOLVED-5G release

Executor/Tasks/Evolved5g/__init__.py

@@ -0,0 +1 @@
+from .jenkins_api import JenkinsJob, JenkinsStatus

Executor/Tasks/Evolved5g/jenkins_api.py

@@ -0,0 +1,81 @@
+from Task import Task
+from Settings import EvolvedConfig
+from Interfaces import Evolved5gJenkinsApi
+from Helper import Level
+
+
+class JenkinsBase(Task):
+    def __init__(self, name, parent, params, logMethod):
+        super().__init__(name, parent, params, logMethod, None)
+        self.config = EvolvedConfig().JenkinsApi
+        self.client = None
+
+    def Run(self):
+        try:
+            self.client = self.getApiClient()
+        except Exception as e:
+            self.Log(Level.Error, f"Unable to create Jenkins API client: {e}")
+            self.client = None
+
+    def getApiClient(self) -> Evolved5gJenkinsApi:
+        if not self.config.Enabled:
+            raise RuntimeError(f"Trying to run {self.name} Task while Jenkins API is not enabled")
+
+        return Evolved5gJenkinsApi(self.config.Host, self.config.Port,
+                                   self.config.User, self.config.Password)
+
+
+class JenkinsJob(JenkinsBase):
+    def __init__(self, logMethod, parent, params):
+        super().__init__("Jenkins Job", parent, params, logMethod)
+        self.paramRules = {
+            'Instance': (None, True),
+            'Job': (None, True),
+            'GitUrl': (None, True),
+            'GitBranch': (None, True),
+            'Version': ('1.0', False),
+            'PublishKey': ('JenkinsJobId', False),
+        }
+
+    def Run(self):
+        super().Run()
+        if self.client is None: return
+
+        instance = self.params["Instance"]
+        job = self.params["Job"]
+        url = self.params["GitUrl"]
+        branch = self.params["GitBranch"]
+        version = self.params["Version"]
+
+        self.Log(Level.DEBUG,
+                 f"Trying to trigger job '{job}' on instance '{instance}' ({url}|{branch}|{version})")
+
+        try:
+            jobId = self.client.TriggerJob(instance, job, url, branch, version)
+            self.Log(Level.INFO, f"Triggered '{job}'. Received Job Id: {jobId}")
+            self.Publish(self.params["PublishKey"], jobId)
+        except Exception as e:
+            self.Log(Level.ERROR, f"Unable to trigger job: {e}")
+
+
+class JenkinsStatus(JenkinsBase):
+    def __init__(self, logMethod, parent, params):
+        super().__init__("Jenkins Status", parent, params, logMethod)
+        self.paramRules = {
+            'JobId': (None, True),
+            'PublishKey': ('JenkinsJobStatus', False),
+        }
+
+    def Run(self):
+        super().Run()
+        if self.client is None: return
+
+        jobId = self.params['JobId']
+
+        try:
+            status, message = self.client.CheckJob(jobId)
+            message = message if message is not None else "<No details>"
+            self.Log(Level.INFO, f"Status of job '{jobId}': {status} ('{message}')")
+            self.Publish(self.params["PublishKey"], status)
+        except Exception as e:
+            self.Log(Level.ERROR, f"Unable to check job '{jobId}' status: {e}")

Executor/Tasks/PostRun/farewell.py

@@ -1,5 +1,6 @@
 from Task import Task
-from Helper import Level, Config
+from Helper import Level
+from Settings import Config
 from time import sleep
 from Interfaces import RemoteApi
 

Executor/Tasks/PreRun/coordinate.py

@@ -1,5 +1,6 @@
 from Task import Task
-from Helper import Level, Config
+from Helper import Level
+from Settings import Config
 from time import sleep
 from Interfaces import RemoteApi
 

Executor/Tasks/PreRun/instantiate.py

@@ -1,5 +1,6 @@
 from Task import Task
-from Helper import Level, Config
+from Helper import Level
+from Settings import Config
 from Data import NsInfo
 from typing import List, Dict
 from Interfaces import Management

Executor/Tasks/Remote/base_remote_task.py

@@ -1,5 +1,5 @@
 from Task import Task
-from Helper import Config
+from Settings import Config
 
 
 TIMEOUT = Config().EastWest.Timeout

Executor/Tasks/Run/__init__.py

@@ -10,3 +10,4 @@
 from .csvToInflux import CsvToInflux
 from .add_milestone import AddMilestone
 from .publish_from_source import PublishFromPreviousTaskLog, PublishFromFile
+from .rest_api import RestApi

Executor/Tasks/Run/add_milestone.py

@@ -5,11 +5,9 @@
 class AddMilestone(Task):
     def __init__(self, logMethod, parent, params):
         super().__init__("Add Milestone", parent, params, logMethod, None)
+        self.paramRules = {'Milestone': (None, True)}
 
     def Run(self):
-        try:
-            milestone = self.params['Milestone']
-            self.Log(Level.INFO, f"Adding milestone '{milestone}' to experiment.")
-            self.parent.AddMilestone(milestone)
-        except KeyError:
-            self.Log(Level.ERROR, "'Milestone' value not set")
+        milestone = self.params['Milestone']
+        self.Log(Level.INFO, f"Adding milestone '{milestone}' to experiment.")
+        self.parent.AddMilestone(milestone)

Executor/Tasks/Run/cli_execute.py

@@ -5,10 +5,11 @@
 class CliExecute(Task):
     def __init__(self, logMethod, parent, params):
         super().__init__("CLI Execute", parent, params, logMethod, None)
+        self.paramRules = {
+            'Parameters': (None, True),
+            'CWD': (None, True)
+        }
 
     def Run(self):
-        parameters = self.params['Parameters']
-        cwd = self.params['CWD']
-
-        cli = Cli(parameters, cwd, self.logMethod)
+        cli = Cli(self.params['Parameters'], self.params['CWD'], self.logMethod)
         cli.Execute()

Executor/Tasks/Run/compress_files.py

@@ -6,13 +6,19 @@
 class CompressFiles(Task):
     def __init__(self, logMethod, parent, params):
         super().__init__("Compress Files", parent, params, logMethod, None)
+        self.paramRules = {
+            'Files': ([], False),
+            'Folders': ([], False),
+            'Output': (None, True)
+        }
 
     def Run(self):
         from Helper import Compress, IO
 
-        files = [abspath(f) for f in self.params.get("Files", [])]
-        folders = [abspath(f) for f in self.params.get("Folders", [])]
-        output = self.params.get("Output", "")
+        files = [abspath(f) for f in self.params["Files"]]
+        folders = [abspath(f) for f in self.params["Folders"]]
+        output = self.params["Output"] .get("Output", "")
+
         self.Log(Level.INFO, f"Compressing files to output: {output}")
 
         for folder in folders:

Executor/Tasks/Run/csvToInflux.py

@@ -6,26 +6,22 @@
 class CsvToInflux(Task):
     def __init__(self, logMethod, parent, params):
         super().__init__("Csv To Influx", parent, params, logMethod, None)
+        self.paramRules = {
+            'ExecutionId': (None, True),
+            'CSV': (None, True),
+            'Measurement': (None, True),
+            'Delimiter': (',', False),
+            'Timestamp': ('Timestamp', False),
+            'Convert': (True, False)
+        }
 
     def Run(self):
-        try: executionId = self.params['ExecutionId']
-        except KeyError:
-            self.Log(Level.ERROR, "ExecutionId value not defined, please review the Task configuration.")
-            return
-
-        try: csvFile = self.params["CSV"]
-        except KeyError:
-            self.Log(Level.ERROR, "CSV file not defined, please review the Task configuration.")
-            return
-
-        try: measurement = self.params["Measurement"]
-        except KeyError:
-            self.Log(Level.ERROR, "Measurement not defined, please review the Task configuration.")
-            return
-
-        delimiter = self.params.get("Delimiter", ',')
-        timestamp = self.params.get("Timestamp", "Timestamp")
-        tryConvert = self.params.get("Convert", True)
+        executionId = self.params['ExecutionId']
+        csvFile = self.params["CSV"]
+        measurement = self.params["Measurement"]
+        delimiter = self.params["Delimiter"]
+        timestamp = self.params["Timestamp"]
+        tryConvert = self.params["Convert"]
 
         try:
             from Helper import InfluxDb, InfluxPayload  # Delayed to avoid cyclic imports

Executor/Tasks/Run/delay.py

@@ -6,9 +6,10 @@
 class Delay(Task):
     def __init__(self, logMethod, parent, params):
         super().__init__("Delay", parent, params, logMethod, None)
+        self.paramRules = {'Time': (60, False)}
 
     def Run(self):
-        value = self.params.get('Time', 60)
+        value = self.params['Time']
         try:
             time = int(value)
             if time < 0:

Executor/Tasks/Run/message.py

@@ -1,11 +1,14 @@
 from Task import Task
 from Helper import Level
-from time import sleep
 
 
 class Message(Task):
     def __init__(self, logMethod, parent, params):
         super().__init__("Message", parent, params, logMethod, None)
+        self.paramRules = {
+            'Severity': ('INFO', False),
+            'Message': (None, True)
+        }
 
     def Run(self):
         level = Level[self.params['Severity']]

Executor/Tasks/Run/publish_from_source.py

@@ -7,26 +7,26 @@
 class PublishFromSource(Task):
     def __init__(self, name, parent, params, logMethod):
         super().__init__(name, parent, params, logMethod, None)
+        self.paramRules = {
+            'Pattern': (None, True),
+            'Keys': (None, True),
+            'Path': (None, False)  # Mandatory only for PublishFromFile, handled below
+        }
 
     def Run(self):
         self.Log(Level.INFO, f'Running task {self.name} with params: {self.params}')
 
-        filePath = self.params.get("Path", None)
-        pattern = self.params.get("Pattern", None)
-        keys = self.params.get("Keys", None)
+        filePath = self.params["Path"]
+        pattern = self.params["Pattern"]
+        keys = self.params["Keys"]
 
-        if pattern is None:
-            self.raiseConfigError("Pattern")
+        self.Log(Level.DEBUG, f"Looking for pattern: '{pattern}'; Assigning groups as:")
 
-        if keys is None:
-            self.raiseConfigError("Keys")
-        else:
-            self.Log(Level.DEBUG, f"Looking for pattern: '{pattern}'; Assigning groups as:")
-            try:
-                for index, key in keys:
-                    self.Log(Level.DEBUG, f"  {index}: {key}")
-            except Exception as e:
-                raise RuntimeError(f"Invalid 'Keys' definition: {e}")
+        try:
+            for index, key in keys:
+                self.Log(Level.DEBUG, f"  {index}: {key}")
+        except Exception as e:
+            raise RuntimeError(f"Invalid 'Keys' definition: {e}")
 
         regex = re.compile(pattern)
 

Executor/Tasks/Run/rest_api.py

@@ -0,0 +1,78 @@
+from Task import Task
+from Helper import Level
+from REST import RestClient, Payload
+
+
+class RestApi(Task):
+    def __init__(self, logMethod, parent, params):
+        super().__init__("REST API", parent, params, logMethod, None)
+        self.paramRules = {
+            'Host': (None, True),
+            'Port': (None, True),
+            'Endpoint': (None, True),
+            'Https': (False, False),
+            'Insecure': (False, False),
+            'Method': ('GET', False),
+            'Payload': ('{}', False),
+            'PayloadMode': (None, False),
+            'Responses': (None, False),
+            'Timeout': (10, False),
+            'Headers': (None, False)
+        }
+
+    def Run(self):
+        client = RestClient(self.params['Host'], self.params['Port'], "",
+                            self.params['Https'], self.params['Insecure'])
+
+        endpoint = self.params['Endpoint']
+        method = str(self.params['Method']).upper()
+        payload = self.params['Payload']
+        payloadMode = self.params['PayloadMode']
+        timeout = self.params['Timeout']
+        headers = self.params['Headers']
+        statusCodes = self.params['Responses']
+
+        if statusCodes is not None:
+            if not isinstance(statusCodes, (tuple, list)):
+                statusCodes = [statusCodes]
+
+            if "Success" in statusCodes:
+                statusCodes.remove("Success")
+                statusCodes.extend([*range(200, 300)])
+
+        self.Log(Level.INFO, f"Sending '{method}' request to '{client.api_url}', endpoint '{endpoint}'.")
+        self.Log(Level.DEBUG, f"Timeout: {timeout}; Extra Headers: {headers}")
+        self.Log(Level.DEBUG, f"Payload: {payload}")
+        if statusCodes is not None:
+            self.Log(Level.DEBUG, f"Accepted status codes: {statusCodes}")
+
+        match method:
+            case "GET":
+                response = client.HttpGet(endpoint, extra_headers=headers, timeout=timeout)
+            case "POST":
+                payloadMode = None if payloadMode is None else Payload[payloadMode]
+                response = client.HttpPost(endpoint, extra_headers=headers,
+                                           body=payload, payload=payloadMode, timeout=timeout)
+            case "PATCH":
+                response = client.HttpPatch(endpoint, extra_headers=headers, body=payload, timeout=timeout)
+            case "DELETE":
+                response = client.HttpDelete(endpoint, extra_headers=headers, timeout=timeout)
+            case _:
+                self.Log(Level.ERROR, f"Unsupported method '{method}'")
+                return
+
+        status = client.ResponseStatusCode(response)
+        try:
+            data = client.ResponseToJson(response)
+        except RuntimeError:
+            try:
+                data = client.ResponseToRaw(response)
+            except RuntimeError:
+                data = None
+
+        self.Log(Level.INFO, f"Status '{status}'; Response: '{data}'")
+        if statusCodes is not None:
+            if status not in statusCodes:
+                message = f"Unexpected status code received: {status}"
+                self.Log(Level.ERROR, message)
+                raise RuntimeError(message)