Experiment specific stable

.gitignore

@@ -6,3 +6,4 @@ orchestrator/__pycache__
 services/kernel/__pycache__
 ext_depen/__pycache__
 .DS_Store
+.idea

README.md

@@ -118,3 +118,4 @@ Below we list the initial developers of the testbed. Maintainer names are omitte
 | Vinod Khandkar     |v.khandkar@surrey.ac.uk             |
 | Abdullahi Abubakar |a.abubakar@surrey.ac.uk             |
 | Roger Zhang        |yz02055@surrey.ac.uk             |
+| Saeed Fadaei       |s.fadaei@surrey.ac.uk

common/client.py

@@ -213,10 +213,17 @@ def schedule_job(self, jobid, nodeid, type_name,
         """        
         log.info("[schedule_job] jobid=%s nodeid=%s type=%s params" 
                     "\{mode=%s deploy=%s execute=%s finish=%s\} schedule=%s"
-                    "start_date=%s end_date=%s length=%d overhead=%s server=%s trigger=%s"
+                    "start_date=%s end_date=%s length=%d overhead=%s server=%s trigger=%s experiment_config=%s"
                     % (jobid, nodeid, type_name, params_mode, params_deploy, 
                         params_execute, params_finish, schedule, 
-                        start_date, end_date, length, overhead, server, trigger))
+                        start_date, end_date, length, overhead, server, trigger, experiment_config))
+
+
+        print("experiment config is ***", experiment_config)
+
+
+        log.info("experiment configs are *** %s", experiment_config)
+
 
         for attempt in self._retry():
             with attempt:
@@ -259,6 +266,7 @@ def schedule_job(self, jobid, nodeid, type_name,
                 #                 container, exp_args_file, remote_path)                    
 
                 _type = pb2.job_type.Value(type_name.upper())
+                log.info("Scheduling job on nodeis : %s" %(nodeid))
                 message = pb2.message_schedule_job(
                     id = jobid,
                     nodeid = nodeid,
@@ -369,12 +377,15 @@ def update_run(self, runid, jobid, nodeid, userid, start_time,
                 return self.grpc_stub.update_run(message, timeout=self.timeout)
 
 
-    def get_runs(self, runid=None, jobid=None, nodeid=None, time_range=None, limit=None):
+    def get_runs(self, userid=None, runid=None, jobid=None, nodeid=None, time_range=None, limit=None):
         for attempt in self._retry():
             with attempt:
                 log.info('sending request to fetch runs')
                 message = pb2.message_get_runs()
 
+                if userid:
+                    message.userid = userid
+
                 if runid:
                     message.runid = runid
 

docker-compose-node.yaml

@@ -19,6 +19,7 @@ services:
       - /var/run/docker.sock:/var/run/docker.sock
       - /datadrive1/leoscope/jobs:/artifacts
       - ./executor-config.yaml:/executor-config.yaml
+      - ./experiment_configs:/leotest/experiment_configs
 
     # network_mode: "host"
     networks:

docker/node/Dockerfile

@@ -1,4 +1,4 @@
-FROM python:3.7-bullseye
+FROM python:3.10-bullseye
 
 RUN apt-get update
 RUN apt-get install -y git
@@ -18,6 +18,8 @@ RUN pip install requests
 COPY node /leotest/node/
 COPY common /leotest/common/
 COPY certs /leotest/certs/
+COPY ext_depen /leotest/ext_depen/
+COPY experiment_configs /leotest/experiment_configs/
 
 ENTRYPOINT [ "python", "-m", "node"]
-# ENTRYPOINT ["/bin/bash"]
+#ENTRYPOINT ["/bin/bash"]

docker/orchestrator/Dockerfile

@@ -1,4 +1,4 @@
-FROM python:3.7-bullseye
+FROM python:3.10-bullseye
 
 RUN apt-get update
 RUN apt-get install -y git
@@ -14,4 +14,4 @@ COPY common /leotest/common/
 COPY certs /leotest/certs/
 
 ENTRYPOINT [ "python", "-m", "orchestrator"]
-# ENTRYPOINT ["/bin/bash"]
+# ENTRYPOINT ["/bin/bash"]

docker/services/kernel/Dockerfile

@@ -1,4 +1,4 @@
-FROM python:3.7-bullseye
+FROM python:3.10-bullseye
 
 RUN apt-get update
 RUN apt-get install -y git
@@ -13,4 +13,4 @@ COPY common /leotest/common/
 COPY certs /leotest/certs/
 
 ENTRYPOINT [ "python", "-m", "services.kernel"]
-# ENTRYPOINT ["/bin/bash"]
+# ENTRYPOINT ["/bin/bash"]

docker/starlink-grpc-tools/Dockerfile

@@ -1,4 +1,4 @@
-FROM python:3.7-bullseye
+FROM python:3.10-bullseye
 
 RUN apt-get update
 RUN apt-get install -y git
@@ -13,4 +13,4 @@ RUN pip install redis
 COPY docker/starlink-grpc-tools/check_grpc.py check_grpc.py  
 
 ENTRYPOINT [ "python", "check_grpc.py", "dish_grpc_mqtt.py", "--port", "1883", "-t", "2", "status", "-v"]
-# ENTRYPOINT ["/bin/bash"]
+# ENTRYPOINT ["/bin/bash"]

node/executor.py

@@ -379,15 +379,16 @@ def _execute_job_loop(self):
 
         network_mode = None
         network = None
-        # if (not self.params['server_mode']) and 'network' in self.params["executor"]["docker"]["execute"]: 
-        if 'network' in self.params["executor"]["docker"]["execute"]: 
+
+        if 'network' in self.params["executor"]["docker"]["execute"]:
             network = self.params["executor"]["docker"]["execute"]["network"]
         else:
             network_mode = 'host'
 
-        ports = None 
-        if network != 'host' and 'ports' in self.params["experiment"]["docker"]["execute"]:
-            ports = self.params["experiment"]["docker"]["execute"]["ports"]
+        # ports = None
+        # self.log.info("ports params are *** %s" % self.params["experiment"]["docker"]["execute"])
+        # if network != 'host' and 'ports' in self.params["experiment"]["docker"]["execute"]:
+        #         ports = self.params["experiment"]["docker"]["execute"]["ports"]
 
         image = self.params["experiment"]["docker"]["image"]
 
@@ -404,8 +405,7 @@ def _execute_job_loop(self):
         else:
             environment.append("LEOTEST_SERVERIP=None")
             environment.append("LEOTEST_SERVER_NODEID=None")
-        # container_name = self.params["experiment"]["docker"]["execute"]["name"]
-        # container_name += "_" + self.params["jobid"]
+
         # volume
         shared = {src_path: {'bind': dst_path, 'mode': 'rw'}}
         print(shared)
@@ -425,32 +425,28 @@ def _execute_job_loop(self):
         }
 
         self.log.info('container network: %s' % network)
-        self.log.info('container ports: %s' % str(ports))
+        # self.log.info('container ports: %s' % str(ports))
         self.log.info('--- log stream started for %s ---' % (self.container_name))
 
         try:
             output = self.client.containers.run(
-                        image=image,
-                        user=os.getuid(), # this helps avoid files being generated with root as owner
-                        #command = '/bin/bash',
-                        detach = False,    # If kept to false, will block the script. If script is killed, container exits 
-                        name = self.container_name,
-                        network_mode = network_mode,
-                        privileged = False,
-                        # cap_add = ["CAP_NET_ADMIN"],
-                        network = network,
-                        ports = ports,
-                        stdin_open = True, # have to keep STDIN open to avoid container exit
-                        tty = False,           
-                        # allocate pseudo tty to avoid container exit 
-                        # (however, this causes character-output instead of line-output when stream=True)
-                        # https://github.com/docker/docker-py/issues/2913
-                        volumes = shared,
-                        stdout=True, 
-                        stderr=True,
-                        stream=True,
-                        environment=environment,
-                        labels=labels
+                image=image,
+                user=os.getuid(),
+                detach=False,
+                name=self.container_name,
+                network_mode=network_mode,
+                privileged=False,
+                network=network,
+                stdin_open=True,
+                tty=False,
+                mem_limit="512m",
+                volumes=shared,
+                # ports=ports,
+                stdout=True,
+                stderr=True,
+                stream=True,
+                environment=environment,
+                labels=labels
             )
         except Exception as e:
             self.log.info(str(e))
@@ -568,20 +564,47 @@ def main():
     client = LeotestClient(grpc_hostname=args.grpc_hostname, 
                             grpc_port=args.grpc_port,
                             userid=nodeid, access_token=access_token)
-    res = client.get_config()
 
+    res = client.get_config()
     config = MessageToDict(res)['config']
-    print(config)
     connection_string = config['datastore']['blob']['connectionString']
     container = config['datastore']['blob']['container']
-
-    #possible change here as well for exp specific cloud 
-    # res2 = client.get_job_by_id(jobid)
-    # exp_args_from_orch = MessageToDict(res2)
-    # exp_config = exp_args_from_orch['config']
-    # #possibly write to yaml and safe load to act like dictionary
-    # connection_string = exp_config["cloud_config"]["connection_string"]
-    # container = exp_config["cloud_config"]["container"] 
+
+    result_dict = {}
+    stack = [result_dict]
+    jobs = client.get_jobs_by_nodeid(nodeid)
+
+    if len(jobs.jobs) > 0:
+        for job in jobs.jobs:
+            fetched_job = client.get_job_by_id(job.id)
+            serialized_job = MessageToDict(fetched_job)
+            data = serialized_job["config"]
+            lines = data.split('\n')
+            job_id = fetched_job.id
+
+            if job_id:
+                for line in lines:
+                    if not line.strip().startswith('#') and line.strip() != '':
+                        indentation = len(line) - len(line.lstrip())
+                        while len(stack) > indentation + 1:
+                            stack.pop()
+                        key, value = map(str.strip, line.split(':', 1))
+                        try:
+                            value = json.loads(value)
+                        except json.JSONDecodeError:
+                            pass
+                        stack[-1][key] = value
+                        if isinstance(value, dict):
+                            stack.append(value)
+        connection_string = result_dict.get("connection_string").strip('"')
+        container = result_dict.get("container").strip('"')
+
+    elif (connection_string is None) or (container is None) or (len(jobs.jobs) < 1):
+        res = client.get_config()
+        config = MessageToDict(res)['config']
+
+        connection_string = config['datastore']['blob']['connectionString']
+        container = config['datastore']['blob']['container']
 
 
     server_ip = None 

node/scheduler.py

@@ -1,4 +1,4 @@
-
+import json
 import os
 import time
 import redis 
@@ -614,6 +614,37 @@ def _scheduler_execute(nodeid, client, cron_scheduler, atq_scheduler, task_sched
     if 'jobs' in res:
         logmsg = 'syncing jobs: '
         for job in res['jobs']:
+            # Saving experiment config in file:
+            log.info(f"reading job config for job id {job['id']}!")
+            retrieve_job = client.get_job_by_id(job["id"])
+            data = retrieve_job.config
+            lines = data.split('\n')
+            result_dict = {}
+            stack = [result_dict]
+
+            for line in lines:
+                if not line.strip().startswith('#') and line.strip() != '':
+                    indentation = len(line) - len(line.lstrip())
+                    while len(stack) > indentation + 1:
+                        stack.pop()
+                    key, value = map(str.strip, line.split(':', 1))
+                    try:
+                        value = json.loads(value)
+                    except json.JSONDecodeError:
+                        pass
+                    stack[-1][key] = value
+                    if isinstance(value, dict):
+                        stack.append(value)
+
+            log.info(f"finished reading job config for job id {job['id']}!")
+            log.info(f"experiment config result for job id {job['id']} is: {result_dict}")
+
+            file_name = f"/leotest/experiment_configs/experiment_config_{job['id']}.json"
+
+            # Save data to the file inside the container
+            with open(file_name, "w") as file:
+                json.dump(result_dict, file)
+
             if nodeid != job['nodeid']:
                 continue
             server = job['server'] if 'server' in job else None

node/trigger.py

@@ -57,7 +57,7 @@ def __init__(self, triggerobj, dish_id, fields, hostname='mqtt'):
         self.port = 1883
         self.id = dish_id
         self.fields = fields
-        self.client = mqtt.Client("leotest_grpc")
+        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION1, "leotest_grpc")
         self.client.on_connect = self.on_connect  
         self.client.on_message = self.on_message 
         self.triggerobj = triggerobj 
@@ -387,7 +387,7 @@ def __init__(self, fields=FIELDS, history=5, mqtt_hostname='mqtt', mqtt_port=188
                 self.env["%s_%d" % (field, i)] = -1
 
         # init queue 
-        self.mqtt_client = mqtt.Client("leotest_trigger")
+        self.mqtt_client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION1, "leotest_trigger")
         self.mqtt_client.on_connect = self.on_queue_connect  
         self.mqtt_client.on_message = self.on_queue_message
 

orchestrator/orchestrator.py

@@ -617,7 +617,7 @@ def schedule_job(self, request, context):
                                             server=server,
                                             trigger=trigger,
                                             config=config)
-
+                    print("Orchestrator : Job added %s" %(id)) 
                     state, msg = self.db.add_job(job)
 
         result = {
@@ -689,7 +689,7 @@ def reschedule_job_nearest(self, request, context):
                     if next_slot:
                         log.info('[reschedule_job_nearest] found next slot: %s' 
                                                                 % str(next_slot))
-                        end_time = next_slot
+                        end_time = job.end_date
                         self.db.update_job_date(jobid, next_slot, end_time)
                         state = True
                         msg = "updated job schedule successfully; next_slot=%s" % str(next_slot)

requirements.txt

@@ -15,3 +15,4 @@ redis==4.6.0
 requests==2.28.1
 skyfield==1.46
 tenacity==8.2.2
+ipython==8.18.1