From c82bc3227af3d87d1c98ee803077f334a645744a Mon Sep 17 00:00:00 2001 From: iamsaeedfadaei Date: Fri, 24 May 2024 11:28:39 +0100 Subject: [PATCH 1/6] add changes to exp specific on orchestrator --- common/client.py | 17 ++++++++++++++--- common/leotest_pb2.py | 2 +- orchestrator/orchestrator.py | 4 ++-- 3 files changed, 17 insertions(+), 6 deletions(-) diff --git a/common/client.py b/common/client.py index 000cd84..11733ec 100644 --- a/common/client.py +++ b/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 diff --git a/common/leotest_pb2.py b/common/leotest_pb2.py index 452a145..50cfc4a 100644 --- a/common/leotest_pb2.py +++ b/common/leotest_pb2.py @@ -1,6 +1,7 @@ # -*- coding: utf-8 -*- # Generated by the protocol buffer compiler. DO NOT EDIT! # source: common/leotest.proto +# Protobuf Python Version: 4.25.0 """Generated protocol buffer code.""" from google.protobuf import descriptor as _descriptor from google.protobuf import descriptor_pool as _descriptor_pool @@ -19,7 +20,6 @@ _builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) _builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'common.leotest_pb2', _globals) if _descriptor._USE_C_DESCRIPTORS == False: - DESCRIPTOR._options = None _globals['_RESPONSE_STATE']._serialized_start=6728 _globals['_RESPONSE_STATE']._serialized_end=6769 diff --git a/orchestrator/orchestrator.py b/orchestrator/orchestrator.py index c4ce4bb..8127c27 100644 --- a/orchestrator/orchestrator.py +++ b/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) From 7d0f29988e132a869c1c68e5e8f8c00d3d190631 Mon Sep 17 00:00:00 2001 From: Saeed Fadaei Date: Fri, 24 May 2024 11:59:53 +0100 Subject: [PATCH 2/6] add sanity check --- sanity_test.sh | 69 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 69 insertions(+) create mode 100755 sanity_test.sh diff --git a/sanity_test.sh b/sanity_test.sh new file mode 100755 index 0000000..0334364 --- /dev/null +++ b/sanity_test.sh @@ -0,0 +1,69 @@ +#!/bin/bash + +# Assuming the file containing the filename is named "filename_file.txt" +config_file=$1 +if [ -n $config_file ]; then + echo "Using default config file " + config_file="sanity_test.conf" +fi + +config_param=`cat $config_file` +#echo "$config_param" + +userid=$(echo $config_param | grep -oP 'userid=\K[^ ]+') +access_token=$(echo $config_param | grep -oP 'access-token=\K[^ ]+') + +if [ -z "$userid" ]; then + echo "No userid provided using default user" + userid="test_user" + access_token="test_user_access_token" +else + access_token=$(echo $config_param | grep -oP 'access-token=\K[^ ]+') + if [ -z "$access_token" ]; then + echo "No access_token provided terminating the test" + echo "userid found = $userid" + exit + fi +fi + +#echo "userid found = $userid" +#echo "access_token found = $access_token" + +jobid=$(echo $config_param | grep -oP 'jobid=\K[^ ]+') +#echo "jobid found = $jobid" +nodeid=$(echo $config_param | grep -oP 'nodeid=\K[^ ]+') +#echo "nodeid found = $nodeid" +length=$(echo $config_param | grep -oP 'length=\K[^ ]+') +#echo "length found = $length" +exp_config=$(echo $config_param | grep -oP 'exp-config=\K[^ ]+') +#echo "exp_config found = $exp_config" + +curr_date="$(date +'%Y-%m-%d')" +curr_time=`date -d'5 min' +"%H:%M:%S.%3N"` +start_date="${curr_date}T${curr_time}" +#echo "Current date = $start_date" +end_date="$(date -d'3 months' +'%Y-%m-%d')" +#echo "end_date = $end_date" + +command=`python3 -m cli --userid=$userid --access-token=$access_token--grpc-host=leoscope.surrey.ac.uk job --action=schedule --jobid=$jobid --nodeid=$nodeid --type=atq --start-date=$start_date --end-date=$end_date --length=$length --overhead --exp-config=$exp_config` +echo $command + +echo "Waiting for results" +sleep 300 + +# Read results +# Define the MongoDB container name and database +container_name="global-testbed-datastore-1" +new_database="leotest" +# Check if the container is running +if docker inspect -f '{{.State.Running}}' "$container_name" &> /dev/null; then + # Run MongoDB commands inside the container + query="{\"jobid\": "\"$jobid"\"}" + docker exec -it "$container_name" mongoexport --db=leotest -o db_output.txt --pretty --collection=runs -q='{"jobid": $query}' --limit=1 --quiet + status=`docker exec -it "$container_name" bash -c "cat db_output.txt | grep -oP 'status_message\":\K[^?]+'"` + echo "Result : " $status +else + echo "Error: MongoDB container $container_name is not running." +fi + + From 5075b4c4c7a8c0e6af459d8984b6bbb2526dee1b Mon Sep 17 00:00:00 2001 From: Saeed Fadaei Date: Sat, 3 Feb 2024 23:09:43 +0000 Subject: [PATCH 3/6] fixing the experiment specific branch for accepting experiment config --- .gitignore | 1 + docker-compose-node.yaml | 1 + docker/node/Dockerfile | 4 +- node/executor.py | 101 ++++++++++++++++++++++++--------------- node/scheduler.py | 33 ++++++++++++- requirements.txt | 1 + 6 files changed, 100 insertions(+), 41 deletions(-) diff --git a/.gitignore b/.gitignore index 096ddbe..17ba594 100644 --- a/.gitignore +++ b/.gitignore @@ -6,3 +6,4 @@ orchestrator/__pycache__ services/kernel/__pycache__ ext_depen/__pycache__ .DS_Store +.idea diff --git a/docker-compose-node.yaml b/docker-compose-node.yaml index 6c14061..8ed04b8 100644 --- a/docker-compose-node.yaml +++ b/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: diff --git a/docker/node/Dockerfile b/docker/node/Dockerfile index 7b3e558..1101514 100644 --- a/docker/node/Dockerfile +++ b/docker/node/Dockerfile @@ -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"] \ No newline at end of file +#ENTRYPOINT ["/bin/bash"] \ No newline at end of file diff --git a/node/executor.py b/node/executor.py index b06741f..4688e99 100644 --- a/node/executor.py +++ b/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 diff --git a/node/scheduler.py b/node/scheduler.py index 665a958..c216004 100644 --- a/node/scheduler.py +++ b/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 diff --git a/requirements.txt b/requirements.txt index 21ce639..8a588b7 100644 --- a/requirements.txt +++ b/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 \ No newline at end of file From f8188a90840c05a9cfc2513856ceb03dd2501d9b Mon Sep 17 00:00:00 2001 From: Saeed Fadaei Date: Sat, 3 Feb 2024 23:28:02 +0000 Subject: [PATCH 4/6] adding my email to developers --- README.md | 1 + 1 file changed, 1 insertion(+) diff --git a/README.md b/README.md index 61cf910..3f12262 100644 --- a/README.md +++ b/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 From 89e6cee304ff0c7ac20d6463898932797198e31b Mon Sep 17 00:00:00 2001 From: Saeed Fadaei Date: Wed, 14 Feb 2024 11:50:21 +0000 Subject: [PATCH 5/6] change mqtt client to support new version --- node/trigger.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/node/trigger.py b/node/trigger.py index 20b953d..217db84 100644 --- a/node/trigger.py +++ b/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 From c5b1b7afb9bca2083817fadcfadfeaa033b13a79 Mon Sep 17 00:00:00 2001 From: Saeed Fadaei Date: Thu, 19 Jun 2025 09:51:17 +0100 Subject: [PATCH 6/6] fix python version --- docker/node/Dockerfile | 4 ++-- docker/orchestrator/Dockerfile | 4 ++-- docker/services/kernel/Dockerfile | 4 ++-- docker/starlink-grpc-tools/Dockerfile | 4 ++-- 4 files changed, 8 insertions(+), 8 deletions(-) diff --git a/docker/node/Dockerfile b/docker/node/Dockerfile index 1101514..03ab9cc 100644 --- a/docker/node/Dockerfile +++ b/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 @@ -22,4 +22,4 @@ COPY ext_depen /leotest/ext_depen/ COPY experiment_configs /leotest/experiment_configs/ ENTRYPOINT [ "python", "-m", "node"] -#ENTRYPOINT ["/bin/bash"] \ No newline at end of file +#ENTRYPOINT ["/bin/bash"] diff --git a/docker/orchestrator/Dockerfile b/docker/orchestrator/Dockerfile index e7fb165..7289a84 100644 --- a/docker/orchestrator/Dockerfile +++ b/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"] \ No newline at end of file +# ENTRYPOINT ["/bin/bash"] diff --git a/docker/services/kernel/Dockerfile b/docker/services/kernel/Dockerfile index 8c65d87..12aab34 100644 --- a/docker/services/kernel/Dockerfile +++ b/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"] \ No newline at end of file +# ENTRYPOINT ["/bin/bash"] diff --git a/docker/starlink-grpc-tools/Dockerfile b/docker/starlink-grpc-tools/Dockerfile index 60c412c..fa4b06c 100644 --- a/docker/starlink-grpc-tools/Dockerfile +++ b/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"] \ No newline at end of file +# ENTRYPOINT ["/bin/bash"]