From e63d2ff1ea83e0de10baf150be8188b7ac1fed84 Mon Sep 17 00:00:00 2001 From: Giorgos Alexakis <58263228+Looper2074@users.noreply.github.com> Date: Sat, 3 Sep 2022 16:09:11 +0300 Subject: [PATCH 01/22] Send script with HTTP Post first --- src/mist/api/machines/methods.py | 49 +++++++++++++++++++++++++++++++- src/mist/api/scripts/base.py | 21 ++++++++++---- 2 files changed, 64 insertions(+), 6 deletions(-) diff --git a/src/mist/api/machines/methods.py b/src/mist/api/machines/methods.py index 4e1bed2c3..80ef40570 100644 --- a/src/mist/api/machines/methods.py +++ b/src/mist/api/machines/methods.py @@ -596,7 +596,7 @@ def create_machine(auth_context, cloud_id, key_id, machine_name, location_id, cloud.ctl.compute.list_machines() except Exception as e: if i > 8: - raise(e) + raise (e) else: continue @@ -2615,7 +2615,54 @@ def find_best_ssh_params(machine, auth_context=None): raise MachineUnauthorizedError +def prepare_ssh_dict(auth_context, machine, + command): + key_association_id, hostname, user, port = find_best_ssh_params( + machine, auth_context=auth_context) + association = KeyMachineAssociation.objects.get(id=key_association_id) + key = association.key + key_path = key.private.secret.name + expiry = int(datetime.now().timestamp()) + 100 + org = machine.owner + vault_token = org.vault_token if org.vault_token is not None else \ + config.VAULT_TOKEN + vault_secret_engine_path = machine.owner.vault_secret_engine_path + vault_addr = org.vault_address if org.vault_address is not None else \ + config.VAULT_ADDR + msg_to_encrypt = '%s,%s,%s,%s' % ( + vault_token, + vault_addr, + vault_secret_engine_path, + key_path) + from mist.api.helpers import encrypt + # ENCRYPTION KEY AND HMAC KEY SHOULD BE DIFFERENT! + encrypted_msg = encrypt(msg_to_encrypt, segment_size=128) + command_encoded = base64.urlsafe_b64encode( + command.encode()).decode() + msg = '%s,%s,%s,%s,%s,%s' % ( + user, + hostname, + port, + expiry, + command_encoded, + encrypted_msg) + mac = hmac.new( + config.SIGN_KEY.encode(), + msg=msg.encode(), + digestmod=hashlib.sha256).hexdigest() + ssh_dict = { + "user": user, + "hostname": hostname, + "port": port, + "expiry": expiry, + "command_encoded": command_encoded, + "encrypted_msg": encrypted_msg, + "mac": mac, + } + return ssh_dict, key.name # SEC + + def prepare_ssh_uri(auth_context, machine, command=config.DEFAULT_EXEC_TERMINAL, job_id=None): diff --git a/src/mist/api/scripts/base.py b/src/mist/api/scripts/base.py index 5ec0337a8..456ae02e2 100644 --- a/src/mist/api/scripts/base.py +++ b/src/mist/api/scripts/base.py @@ -5,6 +5,7 @@ import urllib.request import urllib.parse import urllib.error +from mist.api.machines.methods import prepare_ssh_dict import mongoengine as me @@ -268,11 +269,21 @@ def run(self, auth_context, machine, host=None, port=None, username=None, f' return "$retval";' '} && fetchrun' ) - ssh_user, key_name, ws_uri = prepare_ssh_uri( - auth_context=auth_context, machine=machine, job_id=job_id, + ssh_dict, key_name = prepare_ssh_dict( + auth_context=auth_context, machine=machine, command=command) - exit_code, stdout = websocket_for_scripts( - ws_uri).wait_command_to_finish() + sendScriptURI = '%s/sshJob/sendScript/%s/' % ( + config.PORTAL_URI, + job_id + ) + resp = requests.post(sendScriptURI, json=ssh_dict) + if resp.status_code == 200: + ws_uri = '%s/ssh/runScript/%s/' % ( + config.PORTAL_URI.replace('http', 'ws'), + job_id + ) + exit_code, stdout = websocket_for_scripts( + ws_uri).wait_command_to_finish() return { 'command': command, @@ -280,7 +291,7 @@ def run(self, auth_context, machine, host=None, port=None, username=None, 'stdout': re.sub(r"(\n)\1+", r"\1", stdout.replace( '\r\n', '\n').replace('\r', '\n')), 'key_name': key_name, - 'ssh_user': ssh_user + 'ssh_user': ssh_dict["user"], } def _preparse_file(self): From ccddf326e4558a0d3b35ea6c095a86ed0656a7e8 Mon Sep 17 00:00:00 2001 From: Giorgos Alexakis <58263228+Looper2074@users.noreply.github.com> Date: Sat, 3 Sep 2022 17:19:11 +0300 Subject: [PATCH 02/22] initiliaze exit_code --- src/mist/api/scripts/base.py | 1 + 1 file changed, 1 insertion(+) diff --git a/src/mist/api/scripts/base.py b/src/mist/api/scripts/base.py index 456ae02e2..da1d42970 100644 --- a/src/mist/api/scripts/base.py +++ b/src/mist/api/scripts/base.py @@ -277,6 +277,7 @@ def run(self, auth_context, machine, host=None, port=None, username=None, job_id ) resp = requests.post(sendScriptURI, json=ssh_dict) + exit_code = 1 if resp.status_code == 200: ws_uri = '%s/ssh/runScript/%s/' % ( config.PORTAL_URI.replace('http', 'ws'), From 83f1fa02085460597f834bb4b94451ba97bb3795 Mon Sep 17 00:00:00 2001 From: Giorgos Alexakis <58263228+Looper2074@users.noreply.github.com> Date: Sat, 3 Sep 2022 19:10:08 +0300 Subject: [PATCH 03/22] Initiliaze stdout --- src/mist/api/scripts/base.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/mist/api/scripts/base.py b/src/mist/api/scripts/base.py index da1d42970..b44217647 100644 --- a/src/mist/api/scripts/base.py +++ b/src/mist/api/scripts/base.py @@ -277,7 +277,7 @@ def run(self, auth_context, machine, host=None, port=None, username=None, job_id ) resp = requests.post(sendScriptURI, json=ssh_dict) - exit_code = 1 + exit_code, stdout = 1, "" if resp.status_code == 200: ws_uri = '%s/ssh/runScript/%s/' % ( config.PORTAL_URI.replace('http', 'ws'), From 0744c3d1f73074801181fd0db9dbc8b76c635a33 Mon Sep 17 00:00:00 2001 From: Giorgos Alexakis <58263228+Looper2074@users.noreply.github.com> Date: Sat, 3 Sep 2022 19:13:47 +0300 Subject: [PATCH 04/22] Fix circular dependency --- src/mist/api/scripts/base.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/mist/api/scripts/base.py b/src/mist/api/scripts/base.py index b44217647..8aaa14d9b 100644 --- a/src/mist/api/scripts/base.py +++ b/src/mist/api/scripts/base.py @@ -5,7 +5,6 @@ import urllib.request import urllib.parse import urllib.error -from mist.api.machines.methods import prepare_ssh_dict import mongoengine as me @@ -233,7 +232,7 @@ def run(self, auth_context, machine, host=None, port=None, username=None, password=None, su=False, key_id=None, params=None, job_id=None, env='', owner=None): from mist.api.users.models import Organization - from mist.api.machines.methods import prepare_ssh_uri + from mist.api.machines.methods import prepare_ssh_dict import re if auth_context: owner = auth_context.owner From 3022b3928fa90a7b724ea033aada94f7651a58b2 Mon Sep 17 00:00:00 2001 From: George Alexakis Date: Sun, 4 Sep 2022 09:49:29 +0300 Subject: [PATCH 05/22] Remove websocket on_open() --- src/mist/api/helpers.py | 1 - 1 file changed, 1 deletion(-) diff --git a/src/mist/api/helpers.py b/src/mist/api/helpers.py index d8da4b97e..8377c869a 100644 --- a/src/mist/api/helpers.py +++ b/src/mist/api/helpers.py @@ -2021,7 +2021,6 @@ def __init__(self, uri): on_error=self.on_error, on_close=self.on_close) self.ws = ws - self.ws.on_open = self.on_open self.buffer = "" def on_message(self, message): From 5b74fedd8f5fa0ebe846a046d1b61a59173aff46 Mon Sep 17 00:00:00 2001 From: George Alexakis Date: Sun, 4 Sep 2022 11:10:36 +0300 Subject: [PATCH 06/22] Fix ssh_dict --- src/mist/api/machines/methods.py | 4 ++-- src/mist/api/scripts/base.py | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/src/mist/api/machines/methods.py b/src/mist/api/machines/methods.py index 80ef40570..9bd82fa68 100644 --- a/src/mist/api/machines/methods.py +++ b/src/mist/api/machines/methods.py @@ -2653,8 +2653,8 @@ def prepare_ssh_dict(auth_context, machine, ssh_dict = { "user": user, "hostname": hostname, - "port": port, - "expiry": expiry, + "port": str(port), + "expiry": str(expiry), "command_encoded": command_encoded, "encrypted_msg": encrypted_msg, "mac": mac, diff --git a/src/mist/api/scripts/base.py b/src/mist/api/scripts/base.py index 8aaa14d9b..7bedc99d4 100644 --- a/src/mist/api/scripts/base.py +++ b/src/mist/api/scripts/base.py @@ -271,14 +271,14 @@ def run(self, auth_context, machine, host=None, port=None, username=None, ssh_dict, key_name = prepare_ssh_dict( auth_context=auth_context, machine=machine, command=command) - sendScriptURI = '%s/sshJob/sendScript/%s/' % ( + sendScriptURI = '%s/sshJob/sendScript/%s' % ( config.PORTAL_URI, job_id ) resp = requests.post(sendScriptURI, json=ssh_dict) exit_code, stdout = 1, "" if resp.status_code == 200: - ws_uri = '%s/ssh/runScript/%s/' % ( + ws_uri = '%s/ssh/runScript/%s' % ( config.PORTAL_URI.replace('http', 'ws'), job_id ) From d350b48de28099e9d5c6729cdeb0524c9a994f74 Mon Sep 17 00:00:00 2001 From: George Alexakis Date: Sun, 4 Sep 2022 12:33:12 +0300 Subject: [PATCH 07/22] Remove unused import --- src/mist/api/helpers.py | 1 - 1 file changed, 1 deletion(-) diff --git a/src/mist/api/helpers.py b/src/mist/api/helpers.py index 8377c869a..05ebd151a 100644 --- a/src/mist/api/helpers.py +++ b/src/mist/api/helpers.py @@ -70,7 +70,6 @@ import secrets import operator import websocket -import _thread # Python 2 and 3 support from future.utils import string_types From 0d7347fe42aee9ab0ccd4ce197368dbf535bd43f Mon Sep 17 00:00:00 2001 From: George Alexakis Date: Mon, 24 Oct 2022 19:51:54 +0300 Subject: [PATCH 08/22] Remove websocket --- requirements.txt | 1 + src/mist/api/scripts/base.py | 13 +++++-------- 2 files changed, 6 insertions(+), 8 deletions(-) diff --git a/requirements.txt b/requirements.txt index 60e24d6fa..8db547639 100644 --- a/requirements.txt +++ b/requirements.txt @@ -58,3 +58,4 @@ troposphere #tornado_profile uwsgidecorators websocket-client +rstream diff --git a/src/mist/api/scripts/base.py b/src/mist/api/scripts/base.py index 7bedc99d4..1c9a2bbcc 100644 --- a/src/mist/api/scripts/base.py +++ b/src/mist/api/scripts/base.py @@ -271,20 +271,17 @@ def run(self, auth_context, machine, host=None, port=None, username=None, ssh_dict, key_name = prepare_ssh_dict( auth_context=auth_context, machine=machine, command=command) - sendScriptURI = '%s/sshJob/sendScript/%s' % ( + sendScriptURI = '%s/ssh/jobs/%s' % ( config.PORTAL_URI, job_id ) resp = requests.post(sendScriptURI, json=ssh_dict) exit_code, stdout = 1, "" if resp.status_code == 200: - ws_uri = '%s/ssh/runScript/%s' % ( - config.PORTAL_URI.replace('http', 'ws'), - job_id - ) - exit_code, stdout = websocket_for_scripts( - ws_uri).wait_command_to_finish() - + # start reading from rabbitmq-stream + # exit_code, stdout = websocket_for_scripts( + # ws_uri).wait_command_to_finish() + log.info("reading logs from rabbitmq-stream of job_id:%s", job_id) return { 'command': command, 'exit_code': exit_code, From 1d82121afee3e451f18206f84bbad34e06cf0f4d Mon Sep 17 00:00:00 2001 From: George Alexakis Date: Tue, 25 Oct 2022 16:58:36 +0300 Subject: [PATCH 09/22] Get stdout from rabbitmq stream --- src/mist/api/helpers.py | 36 ++++++++++++++++++++++++++++++++++++ src/mist/api/scripts/base.py | 8 ++++++-- 2 files changed, 42 insertions(+), 2 deletions(-) diff --git a/src/mist/api/helpers.py b/src/mist/api/helpers.py index 05ebd151a..66b2417e4 100644 --- a/src/mist/api/helpers.py +++ b/src/mist/api/helpers.py @@ -11,6 +11,9 @@ """ +import asyncio +import signal +from rstream import Consumer, amqp_decoder, AMQPMessage from functools import reduce from mist.api import config from mist.api.exceptions import WorkflowExecutionError, BadRequestError @@ -2064,3 +2067,36 @@ def extract_selector_type(**kwargs): if error_count == len(kwargs.get('selectors', [])): raise BadRequestError('selector_type') return selector_type + + +class RabbitMQStreamConsumer: + def __init__(self, job_id): + self.stream_name = job_id + self.buffer = "" + self.exit_code = 1 + + def on_message(self, msg: AMQPMessage): + message = msg.Body() + if message.startswith('retval:'): + self.exit_code = message.replace('retval:', '', 1) + else: + self.buffer = self.buffer + message + + async def consume(self): + consumer = Consumer( + host=os.getenv("RABBITMQ_HOST"), + port=5552, + vhost='/', + username=os.getenv("RABBITMQ_USERNAME"), + password=os.getenv("RABBITMQ_PASSWORD"), + ) + + loop = asyncio.get_event_loop() + loop.add_signal_handler( + signal.SIGINT, lambda: asyncio.create_task(consumer.close())) + + await consumer.start() + await consumer.subscribe(self.stream_name, self.on_message, + decoder=amqp_decoder) + await consumer.run() + return self.exit_code, self.buffer diff --git a/src/mist/api/scripts/base.py b/src/mist/api/scripts/base.py index 1c9a2bbcc..a1f560b93 100644 --- a/src/mist/api/scripts/base.py +++ b/src/mist/api/scripts/base.py @@ -10,12 +10,13 @@ from mist.api.exceptions import BadRequestError -from mist.api.helpers import trigger_session_update, mac_sign -from mist.api.helpers import websocket_for_scripts +from mist.api.helpers import trigger_session_update, mac_sign, RabbitMQStreamConsumer from mist.api.exceptions import ScriptNameExistsError +import asyncio from mist.api import config + log = logging.getLogger(__name__) @@ -282,6 +283,9 @@ def run(self, auth_context, machine, host=None, port=None, username=None, # exit_code, stdout = websocket_for_scripts( # ws_uri).wait_command_to_finish() log.info("reading logs from rabbitmq-stream of job_id:%s", job_id) + c = RabbitMQStreamConsumer(job_id) + exit_code, stdout = asyncio.run(c.consume()) + return { 'command': command, 'exit_code': exit_code, From 058c1acb93e9697c810eba809eea402aeebcdefe Mon Sep 17 00:00:00 2001 From: George Alexakis Date: Tue, 25 Oct 2022 18:44:28 +0300 Subject: [PATCH 10/22] Update python version --- Dockerfile | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Dockerfile b/Dockerfile index 804a4a57d..8b97b40fb 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,4 +1,4 @@ -FROM python:3.7-slim-bullseye +FROM python:3.9-slim-buster # Install libvirt which requires system dependencies. RUN apt update && \ @@ -63,4 +63,4 @@ ENV JS_BUILD=1 \ RUN echo "{\"sha\":\"$VERSION_SHA\",\"name\":\"$VERSION_NAME\",\"repo\":\"$VERSION_REPO\",\"modified\":false}" \ - > /mist-version.json + > /mist-version.json From da6c767ac42fe6e89efcbf6bdf77f4764289ad10 Mon Sep 17 00:00:00 2001 From: Dimitris Moraitis Date: Sun, 20 Nov 2022 23:48:27 +0200 Subject: [PATCH 11/22] Fix flake8 warnings --- .../0019-cleanup-keys-libvirt-locations.py | 2 +- migrations/0041-remove-control-plane-cost.py | 2 +- .../api/clouds/controllers/compute/base.py | 21 ++++++------ .../clouds/controllers/compute/controllers.py | 32 +++++++++---------- .../api/clouds/controllers/container/base.py | 8 ++--- src/mist/api/clouds/controllers/main/base.py | 3 +- src/mist/api/config.py | 4 +-- src/mist/api/helpers.py | 2 ++ src/mist/api/keys/base.py | 12 ++++--- src/mist/api/portal/methods.py | 17 +++++----- src/mist/api/portal/tasks.py | 5 +-- src/mist/api/scripts/base.py | 3 +- src/mist/api/users/models.py | 2 +- 13 files changed, 61 insertions(+), 52 deletions(-) diff --git a/migrations/0019-cleanup-keys-libvirt-locations.py b/migrations/0019-cleanup-keys-libvirt-locations.py index e3e2e9a07..9903a4bd2 100644 --- a/migrations/0019-cleanup-keys-libvirt-locations.py +++ b/migrations/0019-cleanup-keys-libvirt-locations.py @@ -21,7 +21,7 @@ def cleanup_libvirt_cloud_locations(): """ from mist.api.models import Machine from mist.api.clouds.models import CloudLocation, LibvirtCloud - libvirt_cloud_ids = [l.id for l in LibvirtCloud.objects( + libvirt_cloud_ids = [loc.id for loc in LibvirtCloud.objects( deleted=None).only('id')] for loc in CloudLocation.objects(cloud__in=libvirt_cloud_ids): diff --git a/migrations/0041-remove-control-plane-cost.py b/migrations/0041-remove-control-plane-cost.py index a75cfbad2..24e0e6c50 100644 --- a/migrations/0041-remove-control-plane-cost.py +++ b/migrations/0041-remove-control-plane-cost.py @@ -23,7 +23,7 @@ def remove_control_plane_costs(): '$unset': {"cost.control_plane_monthly": 1} }) print(f'{res.modified_count} machines were modified.') - print(f'Removing control_plane_monthly from clusters ...') + print('Removing control_plane_monthly from clusters ...') res = db_clusters.update_many({}, { '$unset': {"cost.control_plane_monthly": 1} }) diff --git a/src/mist/api/clouds/controllers/compute/base.py b/src/mist/api/clouds/controllers/compute/base.py index 44b2b62fe..0b9800693 100644 --- a/src/mist/api/clouds/controllers/compute/base.py +++ b/src/mist/api/clouds/controllers/compute/base.py @@ -1466,8 +1466,8 @@ def list_locations(self, persist=True): task_key = 'cloud:list_locations:%s' % self.cloud.id task = PeriodicTaskInfo.get_or_add(task_key) with task.task_runner(persist=persist): - cached_locations = {'%s' % l.id: l.as_dict() - for l in self.list_cached_locations()} + cached_locations = {'%s' % loc.id: loc.as_dict() + for loc in self.list_cached_locations()} locations = self._list_locations() @@ -1475,9 +1475,10 @@ def list_locations(self, persist=True): if location.id not in cached_locations.keys()] if amqp_owner_listening(self.cloud.owner.id): - locations_dict = [l.as_dict() for l in locations] + locations_dict = [loc.as_dict() for loc in locations] if cached_locations and locations_dict: - new_locations = {'%s' % l['id']: l for l in locations_dict} + new_locations = { + '%s' % loc['id']: loc for loc in locations_dict} # Pop extra to prevent weird patches for loc in cached_locations: cached_locations[loc].pop('extra') @@ -1606,19 +1607,19 @@ def _list_locations(self): # update missing_since for locations not returned by libcloud CloudLocation.objects(cloud=self.cloud, missing_since=None, - external_id__nin=[l.external_id - for l in locations]).update( + external_id__nin=[loc.external_id + for loc in locations]).update( missing_since=now) # update locations for locations seen for the first time CloudLocation.objects(cloud=self.cloud, first_seen=None, - external_id__in=[l.external_id - for l in locations]).update( + external_id__in=[loc.external_id + for loc in locations]).update( first_seen=now) # update last_seen, unset missing_since for locations we just saw CloudLocation.objects(cloud=self.cloud, - external_id__in=[l.external_id - for l in locations]).update( + external_id__in=[loc.external_id + for loc in locations]).update( last_seen=now, missing_since=None) return locations diff --git a/src/mist/api/clouds/controllers/compute/controllers.py b/src/mist/api/clouds/controllers/compute/controllers.py index c385ce4cf..2c989107d 100644 --- a/src/mist/api/clouds/controllers/compute/controllers.py +++ b/src/mist/api/clouds/controllers/compute/controllers.py @@ -298,7 +298,7 @@ def _list_images__fetch_images(self, search=None): if 'UnauthorizedOperation' in str(e.message): images = [] else: - raise() + raise for image in images: if image.id in default_images: image.name = default_images[image.id] @@ -308,7 +308,7 @@ def _list_images__fetch_images(self, search=None): if 'UnauthorizedOperation' in str(e.message): pass else: - raise() + raise else: # search on EC2. search = search.lstrip() @@ -335,7 +335,7 @@ def _list_images__fetch_images(self, search=None): if 'UnauthorizedOperation' in str(e.message): break else: - raise() + raise else: if images: break @@ -5565,14 +5565,14 @@ def _list_locations__fetch_locations(self): if locations: hypervisors = self.connection.connection.request( "/settings/hypervisor_zones.json") - for l in locations: + for loc in locations: for hypervisor in hypervisors.object: h = hypervisor.get("hypervisor_group") - if str(h.get("location_group_id")) == l.id: + if str(h.get("location_group_id")) == loc.id: # get max_memory/max_cpu - l.extra["max_memory"] = h.get("max_host_free_memory") - l.extra["max_cpu"] = h.get("max_host_cpu") - l.extra["hypervisor_group_id"] = h.get("id") + loc.extra["max_memory"] = h.get("max_host_free_memory") + loc.extra["max_cpu"] = h.get("max_host_cpu") + loc.extra["hypervisor_group_id"] = h.get("id") break try: @@ -5583,13 +5583,13 @@ def _list_locations__fetch_locations(self): except: pass - for l in locations: + for loc in locations: # get data store zones, and match with locations # through location_group_id # then calculate max_disk_size per data store, # by matching data store zones and data stores try: - store_zones = [dsg for dsg in data_store_zones if l.id is + store_zones = [dsg for dsg in data_store_zones if loc.id is str(dsg['data_store_group'] ['location_group_id'])] for store_zone in store_zones: @@ -5597,7 +5597,7 @@ def _list_locations__fetch_locations(self): store['data_store']['data_store_group_id'] is store_zone['data_store_group']['id']] for store in stores: - l.extra['max_disk_size'] = store['data_store'] + loc.extra['max_disk_size'] = store['data_store'] ['data_store_size'] - store['data_store']['usage'] except: pass @@ -5608,16 +5608,16 @@ def _list_locations__fetch_locations(self): except: pass - for l in locations: + for loc in locations: # match locations with network ids (through location_group_id) - l.extra['networks'] = [] + loc.extra['networks'] = [] try: for network in networks: net = network["network_group"] - if str(net["location_group_id"]) == l.id: - l.extra['networks'].append({'name': net['label'], - 'id': net['id']}) + if str(net["location_group_id"]) == loc.id: + loc.extra['networks'].append({'name': net['label'], + 'id': net['id']}) except: pass diff --git a/src/mist/api/clouds/controllers/container/base.py b/src/mist/api/clouds/controllers/container/base.py index d5265caf1..be0ea13ca 100644 --- a/src/mist/api/clouds/controllers/container/base.py +++ b/src/mist/api/clouds/controllers/container/base.py @@ -489,10 +489,10 @@ def _update_from_libcloud_cluster(self, self._list_clusters__cost_nodes(cluster, libcloud_cluster) cph = nodes_cph + control_plane_cph cpm = nodes_cpm + control_plane_cpm - if(cluster.total_cost.hourly != round(cph, 2) or - cluster.total_cost.monthly != round(cpm, 2) or - cluster.cost.hourly != round(control_plane_cph, 2) or - cluster.cost.monthly != round(control_plane_cpm, 2)): + if cluster.total_cost.hourly != round(cph, 2) or \ + cluster.total_cost.monthly != round(cpm, 2) or \ + cluster.cost.hourly != round(control_plane_cph, 2) or \ + cluster.cost.monthly != round(control_plane_cpm, 2): cluster.total_cost.hourly = round(cph, 2) cluster.total_cost.monthly = round(cpm, 2) cluster.cost.hourly = round(control_plane_cph, 2) diff --git a/src/mist/api/clouds/controllers/main/base.py b/src/mist/api/clouds/controllers/main/base.py index e6d9fe039..724c5d9cb 100644 --- a/src/mist/api/clouds/controllers/main/base.py +++ b/src/mist/api/clouds/controllers/main/base.py @@ -309,7 +309,8 @@ def update(self, user=None, fail_on_error=True, if secret: # value will be obtained from vault data = secret.data if _key not in data.keys(): - raise BadRequestError('The key specified (%s) does not exist in \ + raise BadRequestError( + 'The key specified (%s) does not exist in \ secret `%s`' % (_key, secret.name)) if key in self.cloud._private_fields: diff --git a/src/mist/api/config.py b/src/mist/api/config.py index b0a6db473..824499c6a 100755 --- a/src/mist/api/config.py +++ b/src/mist/api/config.py @@ -1287,8 +1287,8 @@ def dirname(path, num=1): GRAPHITE_URI = "http://graphite" VICTORIAMETRICS_URI = "http://vmselect:8481/select//prometheus" -VICTORIAMETRICS_WRITE_URI = (f"http://vminsert:8480/insert//" - f"prometheus") +VICTORIAMETRICS_WRITE_URI = ("http://vminsert:8480/insert//" + "prometheus") GRAPHITE_TO_VICTORIAMETRICS_METRICS_MAP = {} diff --git a/src/mist/api/helpers.py b/src/mist/api/helpers.py index 66b2417e4..05536721f 100644 --- a/src/mist/api/helpers.py +++ b/src/mist/api/helpers.py @@ -2040,6 +2040,8 @@ def on_error(self, error): log.error("Got Websocket error: %s" % error) def on_open(self): + import _thread + def run(*args): self.ws.wait_command_to_finish() _thread.start_new_thread(run, ()) diff --git a/src/mist/api/keys/base.py b/src/mist/api/keys/base.py index a4b181e57..f9b9b26e2 100644 --- a/src/mist/api/keys/base.py +++ b/src/mist/api/keys/base.py @@ -65,7 +65,8 @@ def add(self, user=None, fail_on_invalid_params=True, **kwargs): if secret: data = secret.data if _key not in data.keys(): - raise BadRequestError('The key specified (%s) does not exist in \ + raise BadRequestError( + 'The key specified (%s) does not exist in \ secret `%s`' % (_key, secret.name)) secret_value = SecretValue(secret=secret, @@ -82,10 +83,11 @@ def add(self, user=None, fail_on_invalid_params=True, **kwargs): if user: secret.assign_to(user) except me.NotUniqueError: - raise KeyExistsError("The path `%s%s` exists on Vault. \ - Try changing the name of the key" % - (config.VAULT_KEYS_PATH, - self.key.name)) + raise KeyExistsError( + "The path `%s%s` exists on Vault. \ + Try changing the name of the key" % ( + config.VAULT_KEYS_PATH, + self.key.name)) try: secret.create_or_update({key: value}) except Exception as exc: diff --git a/src/mist/api/portal/methods.py b/src/mist/api/portal/methods.py index 81c1c27d2..0e23499de 100644 --- a/src/mist/api/portal/methods.py +++ b/src/mist/api/portal/methods.py @@ -43,14 +43,15 @@ def should_task_exist_for_cloud(task, cloud): """ Return whether a given cloud should have the specified scheduled task. """ - if ((task == "list_zones" and cloud.dns_enabled is False) or - (task == "list_buckets" and cloud.object_storage_enabled is False) or - (task == "list_clusters" and cloud.container_enabled is False) or # noqa: E501 - (task == "list_networks" and getattr(cloud.ctl, "network", None) is None) or # noqa: E501 - (task == "list_volumes" and getattr(cloud.ctl, "storage", None) is None) or # noqa: E501 - (task == "list_sizes" and cloud._cls == "Cloud.LibvirtCloud") or - (task != "list machines" and cloud._cls == "Cloud.OtherCloud") - ): + if (task == "list_zones" and cloud.dns_enabled is False) or \ + (task == "list_buckets" and cloud.object_storage_enabled is False) or \ + (task == "list_clusters" and cloud.container_enabled is False) or \ + (task == "list_networks" and getattr( + cloud.ctl, "network", None) is None) or \ + (task == "list_volumes" and getattr( + cloud.ctl, "storage", None) is None) or \ + (task == "list_sizes" and cloud._cls == "Cloud.LibvirtCloud") or \ + (task != "list machines" and cloud._cls == "Cloud.OtherCloud"): return False return True diff --git a/src/mist/api/portal/tasks.py b/src/mist/api/portal/tasks.py index 0907aa8d7..6ba92eb63 100644 --- a/src/mist/api/portal/tasks.py +++ b/src/mist/api/portal/tasks.py @@ -367,8 +367,9 @@ def restore_backup(backup, portal=None, until=False, databases=[ ) result = subprocess.check_output(cmd, shell=True) available_backups = [ - int(l.strip().split('/victoria/')[1].rstrip('/')) - for l in result.decode().split('\n') if '/victoria/' in l] + int(line.strip().split('/victoria/')[1].rstrip('/')) + for line in result.decode().split('\n') + if '/victoria/' in line] available_backups.sort(reverse=True) for b in available_backups: if b < int(backup) and b >= int(until or 0): diff --git a/src/mist/api/scripts/base.py b/src/mist/api/scripts/base.py index a1f560b93..2a5fe713d 100644 --- a/src/mist/api/scripts/base.py +++ b/src/mist/api/scripts/base.py @@ -10,7 +10,8 @@ from mist.api.exceptions import BadRequestError -from mist.api.helpers import trigger_session_update, mac_sign, RabbitMQStreamConsumer +from mist.api.helpers import trigger_session_update, mac_sign +from mist.api.helpers import RabbitMQStreamConsumer from mist.api.exceptions import ScriptNameExistsError import asyncio diff --git a/src/mist/api/users/models.py b/src/mist/api/users/models.py index ba0443225..d113c6d3f 100644 --- a/src/mist/api/users/models.py +++ b/src/mist/api/users/models.py @@ -443,7 +443,7 @@ def as_dict_v2(self, deref='auto', only=''): } ret = prepare_dereferenced_dict(standard_fields, deref_map, self, deref, only) - if(ret.get('policy')): + if ret.get('policy'): ret['policy'] = ret['policy'].__str__() ret['members_count'] = len(ret.get('members', [])) return ret From 852d89dd3f22d6176267c7aa5999303ecd56aef8 Mon Sep 17 00:00:00 2001 From: Dimitris Moraitis Date: Mon, 21 Nov 2022 12:50:56 +0200 Subject: [PATCH 12/22] Upgrade Debian version in base image --- Dockerfile | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Dockerfile b/Dockerfile index 8b97b40fb..f1affa8df 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,4 +1,4 @@ -FROM python:3.9-slim-buster +FROM python:3.9-slim-bullseye # Install libvirt which requires system dependencies. RUN apt update && \ From ad63ed1a457027c00deab7c9e0c4daf13a6d9634 Mon Sep 17 00:00:00 2001 From: Dimitris Moraitis Date: Wed, 23 Nov 2022 00:11:49 +0200 Subject: [PATCH 13/22] Guard against exceptions when looking up amqp listener --- src/mist/api/clouds/controllers/compute/base.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/mist/api/clouds/controllers/compute/base.py b/src/mist/api/clouds/controllers/compute/base.py index 0b9800693..0ebe6e81e 100644 --- a/src/mist/api/clouds/controllers/compute/base.py +++ b/src/mist/api/clouds/controllers/compute/base.py @@ -1474,7 +1474,12 @@ def list_locations(self, persist=True): new_location_objects = [location for location in locations if location.id not in cached_locations.keys()] - if amqp_owner_listening(self.cloud.owner.id): + try: + owner_listening = amqp_owner_listening(self.cloud.owner.id) + except Exception as e: + log.error('Exception raised during amqp owner lookup', repr(e)) + owner_listening = False + if owner_listening: locations_dict = [loc.as_dict() for loc in locations] if cached_locations and locations_dict: new_locations = { From 731668e1063675420e49800389fb5573247127d4 Mon Sep 17 00:00:00 2001 From: Dimitris Moraitis Date: Wed, 23 Nov 2022 00:13:43 +0200 Subject: [PATCH 14/22] Respond with http code 423 on concurrency locking exception --- src/mist/api/concurrency/models.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/mist/api/concurrency/models.py b/src/mist/api/concurrency/models.py index 1d44b704a..844712d5f 100644 --- a/src/mist/api/concurrency/models.py +++ b/src/mist/api/concurrency/models.py @@ -6,6 +6,7 @@ import mongoengine as me +from mist.api.exceptions import MistError log = logging.getLogger(__name__) @@ -18,8 +19,9 @@ class PeriodicTaskTooRecentLastRun(Exception): pass -class PeriodicTaskLockTakenError(Exception): - pass +class PeriodicTaskLockTakenError(MistError): + msg = "Periodic task lock taken" + http_code = 423 class PeriodicTaskInfo(me.Document): From a2a4e403cefcfce76225059c8ff6d0e4b1b7ddaf Mon Sep 17 00:00:00 2001 From: Dimitris Moraitis Date: Wed, 23 Nov 2022 12:36:47 +0200 Subject: [PATCH 15/22] Update v2 submodule --- v2 | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/v2 b/v2 index da3954cef..9937d42d3 160000 --- a/v2 +++ b/v2 @@ -1 +1 @@ -Subproject commit da3954cef9a80de617ccfad4234f34d0656be38c +Subproject commit 9937d42d363837f6962e77fb6860145edc100bc9 From ed6367a1347cd7dcc3127a2335058883931eb310 Mon Sep 17 00:00:00 2001 From: Dimitris Moraitis Date: Wed, 23 Nov 2022 16:55:23 +0200 Subject: [PATCH 16/22] Update python version --- Dockerfile | 4 ++-- lc | 2 +- requirements-frozen.txt | 18 ++++++++--------- requirements.txt | 3 +-- src/mist/api/amqp_tornado.py | 35 ++++++++++++++++---------------- src/mist/api/helpers.py | 20 +----------------- src/mist/api/logs/helpers.py | 39 ++++-------------------------------- src/mist/api/logs/methods.py | 27 ++++++++++++------------- src/mist/api/sock.py | 23 +++++++++++++-------- 9 files changed, 63 insertions(+), 108 deletions(-) diff --git a/Dockerfile b/Dockerfile index f1affa8df..a35c0e9aa 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,4 +1,4 @@ -FROM python:3.9-slim-bullseye +FROM python:3.11-slim-bullseye # Install libvirt which requires system dependencies. RUN apt update && \ @@ -20,7 +20,7 @@ RUN ln -s /influxdb-1.8.4-1/influxd /usr/local/bin/influxd && \ RUN pip install --no-cache-dir --upgrade pip && \ pip install --no-cache-dir --upgrade setuptools && \ - pip install libvirt-python==8.8.0 uwsgi==2.0.20 && \ + pip install libvirt-python==8.8.0 uwsgi==2.0.21 && \ pip install --no-cache-dir ipython ipdb flake8 pytest pytest-cov # Remove `-frozen` to build without strictly pinned dependencies. diff --git a/lc b/lc index d94bec4f5..5be25aaac 160000 --- a/lc +++ b/lc @@ -1 +1 @@ -Subproject commit d94bec4f5ba9ab603051eccbaee5d8fa4deeb058 +Subproject commit 5be25aaac8cf3091849947661266a87eed2d8c4d diff --git a/requirements-frozen.txt b/requirements-frozen.txt index e862f5800..5cd3d6edb 100644 --- a/requirements-frozen.txt +++ b/requirements-frozen.txt @@ -3,7 +3,7 @@ ## direct dependecies and their dependencies and so on. This ensures that ## builds wont start failing just because of a new release of some dependency. -amqp==2.6.1 # 5.1.1 +amqp==5.1.1 apscheduler==3.9.1 asn1crypto==1.5.1 atomicwrites==1.4.1 @@ -22,8 +22,7 @@ cryptography==38.0.1 dateparser==1.1.1 decorator==4.4.2 dnspython==1.16.0 -elasticsearch==6.8.0 -elasticsearch-tornado==2.0.9 +elasticsearch[async]==7.10.1 flake8==5.0.4 future==0.18.2 funcsigs==1.0.2 @@ -40,7 +39,7 @@ Jinja2==2.11.3 jsonpatch==1.32 jsonpickle==2.2.0 jsonpointer==2.3 -kombu==4.6.11 +kombu==5.2.4 Logbook==1.5.3 lxml==4.9.1 Mako==1.2.3 @@ -63,7 +62,7 @@ pathlib2==2.3.7.post1 pbr==5.10.0 pexpect==4.8.0 pickleshare==0.7.5 -pika==0.12.0 # 1.3.0 +pika==1.3.1 pingparsing==1.4.0 pluggy==0.13.1 pretty==0.1 @@ -99,11 +98,10 @@ sentry-sdk==1.9.10 simplegeneric==0.8.1 singledispatch==3.7.0 six==1.16.0 -sockjs-tornado==1.0.6 +sockjs-tornado==1.0.7 subprocrunner==1.6.0 -tornado==5.1.1 +tornado==6.2 troposphere==3.2.2 -#tornado-profile==1.2.0 traitlets==5.4.0 translationstring==1.4 typepy==1.3.0 @@ -111,8 +109,8 @@ typing_extensions==4.4.0 urllib3==1.26.12 uwsgidecorators==1.1.0 venusian==1.2.0 -vine==1.3.0 +vine==5.0.0 wcwidth==0.2.5 WebOb==1.8.7 websocket-client==1.4.1 -yappi==1.3.6 +yappi==1.4.0 diff --git a/requirements.txt b/requirements.txt index 8db547639..41a190bed 100644 --- a/requirements.txt +++ b/requirements.txt @@ -6,7 +6,7 @@ ## ensures that the build won't break because of a new release of some ## dependency. -amqp<3.0 +amqp apscheduler beautifulsoup4 boto3 @@ -14,7 +14,6 @@ dnspython dateparser dramatiq elasticsearch -elasticsearch_tornado flake8 future gevent diff --git a/src/mist/api/amqp_tornado.py b/src/mist/api/amqp_tornado.py index 16cd7d809..143dd94bb 100644 --- a/src/mist/api/amqp_tornado.py +++ b/src/mist/api/amqp_tornado.py @@ -9,8 +9,7 @@ import logging import pika -from pika import adapters - +import pika.adapters.tornado_connection log = logging.getLogger(__name__) @@ -61,8 +60,8 @@ def connect(self): """ log.info('Connecting to %s', self.amqp_url) - return adapters.TornadoConnection(pika.URLParameters(self.amqp_url), - self.on_connection_open) + return pika.adapters.tornado_connection.TornadoConnection( + pika.URLParameters(self.amqp_url), self.on_connection_open) def close_connection(self): """This method closes the connection to RabbitMQ.""" @@ -77,7 +76,7 @@ def add_on_connection_close_callback(self): log.debug('Adding connection close callback') self._connection.add_on_close_callback(self.on_connection_closed) - def on_connection_closed(self, connection, reply_code, reply_text): + def on_connection_closed(self, connection, reply_code, reply_text=''): """This method is invoked by pika when the connection to RabbitMQ is closed unexpectedly. Since it is unexpected, we will reconnect to RabbitMQ if it disconnects. @@ -125,7 +124,7 @@ def add_on_channel_close_callback(self): log.debug('Adding channel close callback') self._channel.add_on_close_callback(self.on_channel_closed) - def on_channel_closed(self, channel, reply_code, reply_text): + def on_channel_closed(self, channel, reply_code, reply_text=''): """Invoked by pika when RabbitMQ unexpectedly closes the channel. Channels are usually closed if you attempt to do something that violates the protocol, such as re-declare an exchange or queue with @@ -164,10 +163,10 @@ def setup_exchange(self, exchange_name): """ log.debug('Declaring exchange %s', exchange_name) - self._channel.exchange_declare(self.on_exchange_declareok, - exchange_name, + self._channel.exchange_declare(exchange_name, self.exchange_type, - **self.exchange_kwargs) + callback=self.on_exchange_declareok, + arguments=self.exchange_kwargs) def on_exchange_declareok(self, unused_frame): """Invoked by pika when RabbitMQ has finished the Exchange.Declare RPC @@ -189,8 +188,9 @@ def setup_queue(self, queue_name): """ log.debug('Declaring queue %s', queue_name) - self._channel.queue_declare(self.on_queue_declareok, queue_name, - **self.queue_kwargs) + self._channel.queue_declare(queue_name, + callback=self.on_queue_declareok, + arguments=self.queue_kwargs) def on_queue_declareok(self, method_frame): """Method invoked by pika when the Queue.Declare RPC call made in @@ -204,8 +204,8 @@ def on_queue_declareok(self, method_frame): """ log.debug('Binding %s to %s with %s', self.exchange, self.queue, self.routing_key) - self._channel.queue_bind(self.on_bindok, self.queue, - self.exchange, self.routing_key) + self._channel.queue_bind(self.queue, self.exchange, self.routing_key, + callback=self.on_bindok) def add_on_cancel_callback(self): """Add a callback that will be invoked if RabbitMQ cancels the consumer @@ -276,7 +276,8 @@ def stop_consuming(self): """ if self._channel: log.info('Sending a Basic.Cancel RPC command to RabbitMQ') - self._channel.basic_cancel(self.on_cancelok, self._consumer_tag) + self._channel.basic_cancel( + consumer_tag=self._consumer_tag, callback=self.on_cancelok) def start_consuming(self): """This method sets up the consumer by first calling @@ -290,9 +291,9 @@ def start_consuming(self): """ log.debug('Issuing consumer related RPC commands') self.add_on_cancel_callback() - self._consumer_tag = self._channel.basic_consume(self.on_message, - self.queue, - no_ack=not self.ack) + self._consumer_tag = self._channel.basic_consume(self.queue, + self.on_message, + auto_ack=not self.ack) def on_bindok(self, unused_frame): """Invoked by pika when the Queue.Bind method has completed. At this diff --git a/src/mist/api/helpers.py b/src/mist/api/helpers.py index 05536721f..2b8d550d9 100644 --- a/src/mist/api/helpers.py +++ b/src/mist/api/helpers.py @@ -26,7 +26,6 @@ from libcloud.container.providers import get_driver as get_container_driver from libcloud.container.drivers.docker import DockerException from libcloud.container.base import ContainerImage -from elasticsearch_tornado import EsClient from elasticsearch import Elasticsearch from distutils.version import LooseVersion from amqp.exceptions import NotFound as AmqpNotFound @@ -1205,24 +1204,6 @@ def view_config(*args, **kwargs): **kwargs) -class AsyncElasticsearch(EsClient): - """Tornado-compatible Elasticsearch client.""" - - def mk_req(self, url, **kwargs): - """Update kwargs with authentication credentials.""" - kwargs.update({ - 'auth_username': config.ELASTICSEARCH['elastic_username'], - 'auth_password': config.ELASTICSEARCH['elastic_password'], - 'validate_cert': config.ELASTICSEARCH['elastic_verify_certs'], - 'ca_certs': None, - - }) - for param in ('connect_timeout', 'request_timeout'): - if param not in kwargs: - kwargs[param] = 30.0 # Increase default timeout by 10 sec. - return super(AsyncElasticsearch, self).mk_req(url, **kwargs) - - def es_client(asynchronous=False): """Returns an initialized Elasticsearch client.""" if not asynchronous: @@ -1236,6 +1217,7 @@ def es_client(asynchronous=False): ) else: method = 'https' if config.ELASTICSEARCH['elastic_use_ssl'] else 'http' + from elasticsearch import AsyncElasticsearch return AsyncElasticsearch( config.ELASTICSEARCH['elastic_host'], port=config.ELASTICSEARCH['elastic_port'], method=method, diff --git a/src/mist/api/logs/helpers.py b/src/mist/api/logs/helpers.py index d12601c98..798fd45a7 100644 --- a/src/mist/api/logs/helpers.py +++ b/src/mist/api/logs/helpers.py @@ -3,11 +3,6 @@ from mist.api.helpers import es_client as es -from mist.api.exceptions import NotFoundError -from mist.api.exceptions import RateLimitError -from mist.api.exceptions import BadRequestError -from mist.api.exceptions import ServiceUnavailableError - from mist.api.logs.constants import TYPES @@ -16,7 +11,7 @@ def _filtered_query(owner_id, close=None, error=None, range=None, type=None, - callback=None, tornado_async=False, **kwargs): + callback=None, es_async=False, **kwargs): """Filter Elasticsearch documents. Executes a filtering aggregation on Elasticsearch documents in order to @@ -94,37 +89,11 @@ def _filtered_query(owner_id, close=None, error=None, range=None, type=None, {"term": {key: value}} ) # Perform Elasticsearch request. - if not tornado_async: + if not es_async: result = es().search(index=index, doc_type=TYPES.get(type), body=query) if callback: return callback(result) return result else: - es(tornado_async).search(index=index, doc_type=TYPES.get(type), - body=json.dumps(query), callback=callback) - - -def _on_response_callback(response, tornado_async=False): - """HTTP Response-handling callback. - - This method is meant to return HTTP Response objects generated either in a - Tornado or synchronous execution context. - - Arguments: - - response: HTTP Response object. - - tornado_async: Denotes if a Tornado-safe HTTP request was issued. - - """ - if tornado_async: - if response.code != 200: - log.error('Error on Elasticsearch query in tornado_async mode. ' - 'Got %d status code: %s', response.code, response.body) - if response.code == 400: - raise BadRequestError() - if response.code == 404: - raise NotFoundError() - if response.code == 429: - raise RateLimitError() - raise ServiceUnavailableError() - response = json.loads(response.body) - return response + es(es_async).search(index=index, doc_type=TYPES.get(type), + body=json.dumps(query), callback=callback) diff --git a/src/mist/api/logs/methods.py b/src/mist/api/logs/methods.py index 2be8be918..60f21ca30 100644 --- a/src/mist/api/logs/methods.py +++ b/src/mist/api/logs/methods.py @@ -16,7 +16,6 @@ from mist.api.users.models import User from mist.api.logs.helpers import _filtered_query -from mist.api.logs.helpers import _on_response_callback from mist.api.logs.constants import FIELDS, JOBS from mist.api.logs.constants import EXCLUDED_BUCKETS, TYPES @@ -398,7 +397,7 @@ def get_events(auth_context, owner_id='', user_id='', event_type='', action='', def get_stories(story_type='', owner_id='', user_id='', sort_order=-1, limit=0, error=None, range=None, pending=None, expand=False, - tornado_callback=None, tornado_async=False, **kwargs): + callback=None, es_async=False, **kwargs): """Fetch stories. Query Elasticsearch for story documents based on the provided arguments. @@ -431,7 +430,7 @@ def get_stories(story_type='', owner_id='', user_id='', sort_order=-1, limit=0, includes += list(FIELDS) + ["action", "extra"] else: includes = [] - assert not tornado_async + assert not es_async if story_type: assert story_type in TYPES @@ -513,28 +512,28 @@ def get_stories(story_type='', owner_id='', user_id='', sort_order=-1, limit=0, # Process returned stories. def _on_stories_callback(response): - result = _on_response_callback(response, tornado_async) return process_stories( - buckets=result["aggregations"]["stories"]["buckets"], - callback=tornado_callback, type=story_type + buckets=response["aggregations"]["stories"]["buckets"], + callback=callback, type=story_type ) # Fetch stories. Invoke callback to process and return results. def _on_request_callback(query): - if not tornado_async: + if es_async is False: result = es().search(index=index, doc_type=TYPES.get(story_type), body=query) return _on_stories_callback(result) else: - es(tornado_async).search(index=index, - body=json.dumps(query), - doc_type=TYPES.get(story_type), - callback=_on_stories_callback) + async def search_async(query): + result = await es_async.search(index=index, + body=json.dumps(query), + doc_type=TYPES.get(story_type)) + return _on_stories_callback(result) + return search_async(query) # Process aggregation results in order to be applied as filters. def _on_filters_callback(response): - results = _on_response_callback(response, tornado_async) - filters = results["aggregations"]["main_bucket"]["buckets"] + filters = response["aggregations"]["main_bucket"]["buckets"] process_filters(query, filters, pending, error) return _on_request_callback(query) @@ -546,7 +545,7 @@ def _on_filters_callback(response): return _filtered_query(owner_id, close=pending, error=error, range=range, type=story_type, callback=_on_filters_callback, - tornado_async=tornado_async, **kwargs) + es_async=es_async, **kwargs) else: return _on_request_callback(query) diff --git a/src/mist/api/sock.py b/src/mist/api/sock.py index e0de04785..56c0629ba 100644 --- a/src/mist/api/sock.py +++ b/src/mist/api/sock.py @@ -20,6 +20,7 @@ import tornado.httpclient from sockjs.tornado import SockJSConnection, SockJSRouter +from mist.api.helpers import es_client from mist.api.sockjs_mux import MultiplexConnection from mist.api.logs.methods import log_event @@ -128,6 +129,7 @@ def get_dict(self): 'session_id': self.session_id, } + @tornado.gen.coroutine def internal_request(self, path, params=None, callback=None): if path.startswith('/'): path = path[1:] @@ -150,14 +152,14 @@ def response_callback(resp): headers = {'Authorization': 'internal %s %s' % ( Portal.get_singleton().internal_api_key, self.cookie_session_id)} - - tornado.httpclient.AsyncHTTPClient( - force_instance=True, max_clients=100).fetch( + client = tornado.httpclient.AsyncHTTPClient( + force_instance=True, max_clients=100) + response = yield client.fetch( '%s/%s' % (config.INTERNAL_API_URL, path), headers=headers, - callback=response_callback, connect_timeout=600, request_timeout=600, ) + response_callback(response) def __repr__(self): conn_dict = self.get_dict() @@ -670,6 +672,7 @@ def on_open(self, conn_info): self.enabled = True self.consumer = None self.enforce_logs_for = self.auth_context.org.id + self.es_client = es_client(asynchronous=True) def on_ready(self): """Initiate the RabbitMQ Consumer.""" @@ -699,6 +702,7 @@ def emit_event(self, event): self.send('event', self.parse_log(event)) self.patch_stories(event) + @tornado.gen.coroutine def send_stories(self, stype): """Send stories of the specified type.""" @@ -723,10 +727,11 @@ def callback(stories): } if self.enforce_logs_for is not None: kwargs['owner_id'] = self.enforce_logs_for - get_stories(tornado_async=True, - tornado_callback=callback, - limit=100, - **kwargs) + + yield get_stories(es_async=self.es_client, + callback=callback, + limit=100, + **kwargs) def patch_stories(self, event): """Send a stories patch. @@ -771,8 +776,10 @@ def filter_log(self, event): return filter_log_event(self.auth_context, event) return event + @tornado.gen.coroutine def on_close(self, stale=False): """Stop the Consumer and close the WebSocket.""" + yield self.es_client.close() if self.consumer is not None: try: self.consumer.stop() From 45f3499c095bcdf96798fb810464a0714d89f1c2 Mon Sep 17 00:00:00 2001 From: Dimitris Moraitis Date: Wed, 23 Nov 2022 19:16:47 +0200 Subject: [PATCH 17/22] Address syntax warnings --- src/mist/api/machines/views.py | 2 +- src/mist/api/metering/methods.py | 2 +- src/mist/api/monitoring/influxdb/handlers.py | 2 +- src/mist/api/rules/base.py | 4 ++-- src/mist/api/rules/models/conditions.py | 4 ++-- src/mist/api/rules/models/main.py | 8 ++++---- src/mist/api/shell.py | 6 +++--- src/mist/api/users/models.py | 2 +- src/mist/api/when/models.py | 4 ++-- 9 files changed, 17 insertions(+), 17 deletions(-) diff --git a/src/mist/api/machines/views.py b/src/mist/api/machines/views.py index fbaf36ef8..3aaf627a1 100644 --- a/src/mist/api/machines/views.py +++ b/src/mist/api/machines/views.py @@ -417,7 +417,7 @@ def create_machine(request): if not isinstance(mtags, dict): if not isinstance(mtags, list): raise ValueError() - if not all((isinstance(t, dict) and len(t) is 1 for t in mtags)): + if not all((isinstance(t, dict) and len(t) == 1 for t in mtags)): raise ValueError() mtags = {key: val for item in mtags for key, val in list(item.items())} diff --git a/src/mist/api/metering/methods.py b/src/mist/api/metering/methods.py index 04420e67e..e4f71da9c 100644 --- a/src/mist/api/metering/methods.py +++ b/src/mist/api/metering/methods.py @@ -109,7 +109,7 @@ def _parse_checks_or_datapoints_series(results, field, owner_id=''): for start_iso, result in results: for series in result: values = series.get('values', []) - assert len(values) is 1, 'Expected a single value. Got %s' % values + assert len(values) == 1, 'Expected a single value. Got %s' % values value = values[0][-1] value = int(round(value)) if value else None owner = series.get('tags', {}).get('owner', owner_id) diff --git a/src/mist/api/monitoring/influxdb/handlers.py b/src/mist/api/monitoring/influxdb/handlers.py index c34db036b..29b416449 100644 --- a/src/mist/api/monitoring/influxdb/handlers.py +++ b/src/mist/api/monitoring/influxdb/handlers.py @@ -213,7 +213,7 @@ def parse_path(self, metric): if len(fields) > 1: for tag in fields[:-1]: tag = tag.split('=') - if len(tag) is not 2: + if len(tag) != 2: log.error('%s got unexpected tag: %s', self.__class__.__name__, tag) continue diff --git a/src/mist/api/rules/base.py b/src/mist/api/rules/base.py index be2a4b271..34baa33b9 100644 --- a/src/mist/api/rules/base.py +++ b/src/mist/api/rules/base.py @@ -354,7 +354,7 @@ def includes_only(self, resource): return False # The rule contains multiple selectors. - if len(self.rule.selectors) is not 1: + if len(self.rule.selectors) != 1: return False # The rule does not refer to resources by their UUID. @@ -362,7 +362,7 @@ def includes_only(self, resource): return False # The rule refers to multiple resources. - if len(self.rule.selectors[0].ids) is not 1: + if len(self.rule.selectors[0].ids) != 1: return False # The rule's single resource does not match `resource`. diff --git a/src/mist/api/rules/models/conditions.py b/src/mist/api/rules/models/conditions.py index 7f7194bc1..2b417e82c 100644 --- a/src/mist/api/rules/models/conditions.py +++ b/src/mist/api/rules/models/conditions.py @@ -186,8 +186,8 @@ def as_dict(self): return {'offset': self.offset, 'period': self.period} def __str__(self): - if self.offset is 0: + if self.offset == 0: return 'Trigger offset is 0' - if self.offset is 1: + if self.offset == 1: return 'Trigger offset of 1 %s' % self.period_singular return 'Trigger offset %s %s' % (self.offset, self.period) diff --git a/src/mist/api/rules/models/main.py b/src/mist/api/rules/models/main.py index 4e2f35ef0..1c2757f74 100644 --- a/src/mist/api/rules/models/main.py +++ b/src/mist/api/rules/models/main.py @@ -351,22 +351,22 @@ def as_dict(self): @property def metric(self): - assert len(self.queries) is 1 + assert len(self.queries) == 1 return self.queries[0].target @property def operator(self): - assert len(self.queries) is 1 + assert len(self.queries) == 1 return self.queries[0].operator @property def value(self): - assert len(self.queries) is 1 + assert len(self.queries) == 1 return self.queries[0].threshold @property def aggregate(self): - assert len(self.queries) is 1 + assert len(self.queries) == 1 return self.queries[0].aggregation @property diff --git a/src/mist/api/shell.py b/src/mist/api/shell.py index 236e6da5b..2b0d45ead 100644 --- a/src/mist/api/shell.py +++ b/src/mist/api/shell.py @@ -398,7 +398,7 @@ def disconnect(self, **kwargs): pass def _wrap_command(self, cmd): - if cmd[-1] is not "\n": + if cmd[-1] != "\n": cmd = cmd + "\n" return cmd @@ -611,7 +611,7 @@ def set_ws_data(self, uuid, secret, control): self._control = control def _wrap_command(self, cmd): - if cmd[-1] is not "\r": + if cmd[-1] != "\r": cmd = cmd + "\r" return cmd @@ -741,7 +741,7 @@ def disconnect(self, **kwargs): pass def _wrap_command(self, cmd): - if cmd[-1] is not "\n": + if cmd[-1] != "\n": cmd = cmd + "\n" return cmd diff --git a/src/mist/api/users/models.py b/src/mist/api/users/models.py index d113c6d3f..acb65838c 100644 --- a/src/mist/api/users/models.py +++ b/src/mist/api/users/models.py @@ -714,7 +714,7 @@ def clean(self): elif team.name == 'Owners': raise me.ValidationError( 'RBAC Mappings are not intended for Team Owners') - elif len(mappings) is not 2: + elif len(mappings) != 2: raise me.ValidationError( 'RBAC Mappings have not been properly initialized for ' 'Team %s' % team) diff --git a/src/mist/api/when/models.py b/src/mist/api/when/models.py index 427b1772b..70e9d85f9 100644 --- a/src/mist/api/when/models.py +++ b/src/mist/api/when/models.py @@ -172,8 +172,8 @@ def as_dict(self): return {'offset': self.offset, 'period': self.period} def __str__(self): - if self.offset is 0: + if self.offset == 0: return 'Trigger offset is 0' - if self.offset is 1: + if self.offset == 1: return 'Trigger offset of 1 %s' % self.period_singular return 'Trigger offset %s %s' % (self.offset, self.period) From 71ba6df3cff41454635f1b0dae2373cb7bc4ea44 Mon Sep 17 00:00:00 2001 From: Dimitris Moraitis Date: Fri, 25 Nov 2022 17:40:04 +0200 Subject: [PATCH 18/22] Populate locations in parallel --- .../api/clouds/controllers/compute/base.py | 154 +++++++++++------- 1 file changed, 92 insertions(+), 62 deletions(-) diff --git a/src/mist/api/clouds/controllers/compute/base.py b/src/mist/api/clouds/controllers/compute/base.py index 0ebe6e81e..3a418705c 100644 --- a/src/mist/api/clouds/controllers/compute/base.py +++ b/src/mist/api/clouds/controllers/compute/base.py @@ -1545,69 +1545,19 @@ def _list_locations(self): len(fetched_locations), self.cloud) locations = [] + try: + loop = asyncio.get_event_loop() + if loop.is_closed(): + raise RuntimeError('loop is closed') + except RuntimeError: + asyncio.set_event_loop(asyncio.new_event_loop()) + loop = asyncio.get_event_loop() + locations = loop.run_until_complete( + self._list_locations_populate_all_locations( + fetched_locations, loop + ) + ) - for loc in fetched_locations: - try: - _location = CloudLocation.objects.get(cloud=self.cloud, - external_id=loc.id) - except CloudLocation.DoesNotExist: - _location = CloudLocation(cloud=self.cloud, - owner=self.cloud.owner, - external_id=loc.id) - try: - _location.country = loc.country - except AttributeError: - _location.country = None - _location.name = loc.name - _location.extra = copy.deepcopy(loc.extra) - _location.missing_since = None - _location.parent = self._list_locations__get_parent(_location, loc) - _location.location_type = self._list_locations__get_type( - _location, loc) - _location.images_location = self._list_locations__get_images_location(loc) # noqa: E501 - try: - created = self._list_locations__location_creation_date(loc) - if created: - created = get_datetime(created) - if _location.created != created: - _location.created = created - except Exception as exc: - log.exception("Error finding creation date for %s in %s.\n%r", - self.cloud, _location, exc) - try: - capabilities = self._list_locations__get_capabilities(loc) - except Exception as exc: - log.error( - "Failed to get location capabilities for cloud: %s", - self.cloud.id) - else: - _location.capabilities = capabilities - - try: - available_sizes = self._list_locations__get_available_sizes(loc) # noqa - except Exception as exc: - log.error('Error adding location-size constraint: %s' - % repr(exc)) - else: - if available_sizes: - _location.available_sizes = available_sizes - - try: - available_images = self._list_locations__get_available_images(loc) # noqa - except Exception as exc: - log.error('Error adding location-image constraint: %s' - % repr(exc)) - else: - if available_images: - _location.available_images = available_images - - try: - _location.save() - except me.ValidationError as exc: - log.error("Error adding %s: %s", loc.name, exc.to_dict()) - raise BadRequestError({"msg": str(exc), - "errors": exc.to_dict()}) - locations.append(_location) now = datetime.datetime.utcnow() # update missing_since for locations not returned by libcloud CloudLocation.objects(cloud=self.cloud, @@ -1629,6 +1579,86 @@ def _list_locations(self): missing_since=None) return locations + async def _list_locations_populate_all_locations(self, locations, loop): + result = [ + loop.run_in_executor( + None, + self._list_locations__populate_location, libcloud_location + ) for libcloud_location in locations + ] + return await asyncio.gather(*result) + + def _list_locations__populate_location(self, libcloud_location): + from mist.api.clouds.models import CloudLocation + try: + _location = CloudLocation.objects.get( + cloud=self.cloud, external_id=libcloud_location.id) + except CloudLocation.DoesNotExist: + _location = CloudLocation( + cloud=self.cloud, owner=self.cloud.owner, + external_id=libcloud_location.id) + try: + _location.country = libcloud_location.country + except AttributeError: + _location.country = None + _location.name = libcloud_location.name + _location.extra = copy.deepcopy(libcloud_location.extra) + _location.missing_since = None + _location.parent = self._list_locations__get_parent( + _location, libcloud_location) + _location.location_type = self._list_locations__get_type( + _location, libcloud_location) + _location.images_location = self._list_locations__get_images_location( + libcloud_location) + try: + created = self._list_locations__location_creation_date( + libcloud_location) + if created: + created = get_datetime(created) + if _location.created != created: + _location.created = created + except Exception as exc: + log.exception("Error finding creation date for %s in %s.\n%r", + self.cloud, _location, exc) + try: + capabilities = self._list_locations__get_capabilities( + libcloud_location) + except Exception as exc: + log.error( + "Failed to get location capabilities for cloud: %s", + self.cloud.id) + else: + _location.capabilities = capabilities + + try: + available_sizes = self._list_locations__get_available_sizes( + libcloud_location) + except Exception as exc: + log.error('Error adding location-size constraint: %s' + % repr(exc)) + else: + if available_sizes: + _location.available_sizes = available_sizes + + try: + available_images = self._list_locations__get_available_images( + libcloud_location) + except Exception as exc: + log.error('Error adding location-image constraint: %s' + % repr(exc)) + else: + if available_images: + _location.available_images = available_images + + try: + _location.save() + except me.ValidationError as exc: + log.error( + "Error adding %s: %s", libcloud_location.name, exc.to_dict()) + raise BadRequestError({"msg": str(exc), + "errors": exc.to_dict()}) + return _location + def _list_locations__fetch_locations(self): """Fetch location listing in a libcloud compatible format From 1e2d6f0b041c09d28c2e7e11c58a75b6c32c117b Mon Sep 17 00:00:00 2001 From: Dimitris Moraitis Date: Thu, 1 Dec 2022 18:55:41 +0200 Subject: [PATCH 19/22] Fix fetching of output when running scripts --- requirements-frozen.txt | 1 + requirements.txt | 5 +- src/mist/api/helpers.py | 78 ++++++++++-------------------- src/mist/api/hub/tornado_client.py | 2 +- src/mist/api/methods.py | 6 ++- src/mist/api/scripts/base.py | 24 ++++++--- src/mist/api/scripts/views.py | 30 +++++++----- src/mist/api/tasks.py | 17 +++---- 8 files changed, 78 insertions(+), 85 deletions(-) diff --git a/requirements-frozen.txt b/requirements-frozen.txt index 5cd3d6edb..e3f8d56e0 100644 --- a/requirements-frozen.txt +++ b/requirements-frozen.txt @@ -5,6 +5,7 @@ amqp==5.1.1 apscheduler==3.9.1 +asgiref==3.5.2 asn1crypto==1.5.1 atomicwrites==1.4.1 attrs==22.1.0 diff --git a/requirements.txt b/requirements.txt index 41a190bed..b73a80671 100644 --- a/requirements.txt +++ b/requirements.txt @@ -8,6 +8,7 @@ amqp apscheduler +asgiref beautifulsoup4 boto3 dnspython @@ -46,6 +47,7 @@ pytest python3-openid pyvmomi requests +rstream s3cmd scp sendgrid-python @@ -56,5 +58,4 @@ tornado troposphere #tornado_profile uwsgidecorators -websocket-client -rstream +websocket-client \ No newline at end of file diff --git a/src/mist/api/helpers.py b/src/mist/api/helpers.py index 2b8d550d9..e0a3c5645 100644 --- a/src/mist/api/helpers.py +++ b/src/mist/api/helpers.py @@ -12,8 +12,9 @@ """ import asyncio -import signal +from asgiref.sync import async_to_sync from rstream import Consumer, amqp_decoder, AMQPMessage +from rstream.exceptions import StreamDoesNotExist from functools import reduce from mist.api import config from mist.api.exceptions import WorkflowExecutionError, BadRequestError @@ -71,7 +72,6 @@ import codecs import secrets import operator -import websocket # Python 2 and 3 support from future.utils import string_types @@ -1996,45 +1996,6 @@ def create_helm_command(repo_url, release_name, chart_name, host, port, token, return helm_install_command -class websocket_for_scripts(object): - - def __init__(self, uri): - self.uri = uri - ws = websocket.WebSocketApp(self.uri, - on_message=self.on_message, - on_error=self.on_error, - on_close=self.on_close) - self.ws = ws - self.buffer = "" - - def on_message(self, message): - message = message.decode('utf-8') - if message.startswith('retval:'): - self.retval = message.replace('retval:', '', 1) - else: - self.buffer = self.buffer + message - - def on_close(self): - self.ws.close() - - def on_error(self, error): - self.ws.close() - log.error("Got Websocket error: %s" % error) - - def on_open(self): - import _thread - - def run(*args): - self.ws.wait_command_to_finish() - _thread.start_new_thread(run, ()) - - def wait_command_to_finish(self): - self.ws.run_forever(ping_interval=9, ping_timeout=8) - self.retval = 0 - output = self.buffer.split("\n")[0:-1] - return self.retval, "\n".join(output) - - def extract_selector_type(**kwargs): error_count = 0 for selector in kwargs.get('selectors', []): @@ -2060,27 +2021,38 @@ def __init__(self, job_id): self.exit_code = 1 def on_message(self, msg: AMQPMessage): - message = msg.Body() + message = next(msg.data).decode('utf-8') if message.startswith('retval:'): - self.exit_code = message.replace('retval:', '', 1) + self.exit_code = int(message.replace('retval:', '', 1)) + import asyncio + asyncio.create_task(self.consumer.close()) else: self.buffer = self.buffer + message + @async_to_sync async def consume(self): - consumer = Consumer( - host=os.getenv("RABBITMQ_HOST"), + self.consumer = Consumer( + host=os.getenv("RABBITMQ_HOST", 'rabbitmq'), port=5552, vhost='/', - username=os.getenv("RABBITMQ_USERNAME"), - password=os.getenv("RABBITMQ_PASSWORD"), + username=os.getenv("RABBITMQ_USERNAME", 'guest'), + password=os.getenv("RABBITMQ_PASSWORD", 'guest'), ) loop = asyncio.get_event_loop() - loop.add_signal_handler( - signal.SIGINT, lambda: asyncio.create_task(consumer.close())) + await self.consumer.start() + sleep_time = 0 + SLEEP_TIMEOUT = 30 + SLEEP_INTERVAL = 3 + while sleep_time < SLEEP_TIMEOUT: - await consumer.start() - await consumer.subscribe(self.stream_name, self.on_message, - decoder=amqp_decoder) - await consumer.run() + try: + await self.consumer.subscribe( + self.stream_name, self.on_message, decoder=amqp_decoder) + break + except StreamDoesNotExist: + sleep(SLEEP_INTERVAL) + sleep_time += SLEEP_INTERVAL + + await self.consumer.run() return self.exit_code, self.buffer diff --git a/src/mist/api/hub/tornado_client.py b/src/mist/api/hub/tornado_client.py index eacfbb4c4..56110fb79 100644 --- a/src/mist/api/hub/tornado_client.py +++ b/src/mist/api/hub/tornado_client.py @@ -98,10 +98,10 @@ def on_message(self, unused_channel, basic_deliver, properties, body): log.debug("%s: Will start listening for routing_key 'from_%s.#'.", self.lbl, self.worker_id) self._channel.queue_bind( - self.ready_callback, self.queue, self.exchange, 'from_%s.#' % self.worker_id, + callback=self.ready_callback, ) return diff --git a/src/mist/api/methods.py b/src/mist/api/methods.py index 3bc582293..2ae521faf 100644 --- a/src/mist/api/methods.py +++ b/src/mist/api/methods.py @@ -372,8 +372,10 @@ def find_public_ips(ips): def notify_admin(title, message="", team="all"): """ This will only work on a multi-user setup configured to send emails """ from mist.api.helpers import send_email - send_email(title, message, - config.NOTIFICATION_EMAIL.get(team, config.NOTIFICATION_EMAIL)) + email = config.NOTIFICATION_EMAIL.get(team, config.NOTIFICATION_EMAIL) + if email: + send_email(title, message, + email) def notify_user(owner, title, message="", email_notify=True, **kwargs): diff --git a/src/mist/api/scripts/base.py b/src/mist/api/scripts/base.py index 2a5fe713d..1d558fd21 100644 --- a/src/mist/api/scripts/base.py +++ b/src/mist/api/scripts/base.py @@ -8,12 +8,12 @@ import mongoengine as me +from time import time from mist.api.exceptions import BadRequestError from mist.api.helpers import trigger_session_update, mac_sign from mist.api.helpers import RabbitMQStreamConsumer from mist.api.exceptions import ScriptNameExistsError -import asyncio from mist.api import config @@ -232,7 +232,7 @@ def generate_signed_url_v2(self): def run(self, auth_context, machine, host=None, port=None, username=None, password=None, su=False, key_id=None, params=None, job_id=None, - env='', owner=None): + env='', owner=None, ret=None, action_prefix=None): from mist.api.users.models import Organization from mist.api.machines.methods import prepare_ssh_dict import re @@ -270,6 +270,7 @@ def run(self, auth_context, machine, host=None, port=None, username=None, f' return "$retval";' '} && fetchrun' ) + log.info('Preparing ssh dict') ssh_dict, key_name = prepare_ssh_dict( auth_context=auth_context, machine=machine, command=command) @@ -277,16 +278,25 @@ def run(self, auth_context, machine, host=None, port=None, username=None, config.PORTAL_URI, job_id ) + log.info('Sending request to sheller:: %s' % sendScriptURI) + log.info(ssh_dict) + start = time() resp = requests.post(sendScriptURI, json=ssh_dict) + log.info('Sheller returned %s in %d' % ( + resp.status_code, time() - start)) exit_code, stdout = 1, "" if resp.status_code == 200: + from mist.api.logs.methods import log_event + log_event( + event_type='job', + action=action_prefix + 'script_started', + **ret + ) + log.info('Script started: %s' % ret) # start reading from rabbitmq-stream - # exit_code, stdout = websocket_for_scripts( - # ws_uri).wait_command_to_finish() - log.info("reading logs from rabbitmq-stream of job_id:%s", job_id) c = RabbitMQStreamConsumer(job_id) - exit_code, stdout = asyncio.run(c.consume()) - + log.info("reading logs from rabbitmq-stream of job_id:%s" % job_id) + exit_code, stdout = c.consume() return { 'command': command, 'exit_code': exit_code, diff --git a/src/mist/api/scripts/views.py b/src/mist/api/scripts/views.py index 1610b4ae1..9a5a8bd7e 100644 --- a/src/mist/api/scripts/views.py +++ b/src/mist/api/scripts/views.py @@ -402,6 +402,7 @@ def run_script(request): su = params.get('su', False) env = params.get('env') job_id = params.get('job', params.get('job_id', None)) + run_async = params.get('async', True) if not job_id: job = 'run_script' job_id = uuid.uuid4().hex @@ -453,17 +454,24 @@ def run_script(request): except me.DoesNotExist: raise NotFoundError('Script id not found') job_id = job_id or uuid.uuid4().hex - tasks.run_script.send_with_options( - args=(auth_context.serialize(), script.id, machine.id), - kwargs={ - "params": script_params, - "env": env, - "su": su, - "job_id": job_id, - "job": job - }, - delay=1_000 - ) + if run_async: + tasks.run_script.send_with_options( + args=(auth_context.serialize(), script.id, machine.id), + kwargs={ + "params": script_params, + "env": env, + "su": su, + "job_id": job_id, + "job": job + }, + delay=1_000 + ) + else: + return tasks.run_script( + auth_context.serialize(), + script.id, machine.id, + params=script_params, env=env, + su=su, job=job, job_id=job_id) return {'job_id': job_id, 'job': job} diff --git a/src/mist/api/tasks.py b/src/mist/api/tasks.py index 203344331..f3f047b94 100644 --- a/src/mist/api/tasks.py +++ b/src/mist/api/tasks.py @@ -1148,6 +1148,7 @@ def group_run_script(auth_context_serialized, script_id, name, machine_ids, @dramatiq.actor(queue_name='dramatiq_schedules', time_limit=3_600_000, store_results=True, + max_retries=0, throws=(me.DoesNotExist,)) def run_script(auth_context_serialized, script_id, machine_id, params='', host='', key_id='', username='', password='', port=22, @@ -1184,12 +1185,9 @@ def run_script(auth_context_serialized, script_id, machine_id, params='', 'port': port, 'command': '', 'stdout': '', - 'exit_code': '', - 'wrapper_stdout': '', - 'extra_output': '', + 'exit_code': -255, 'error': False, } - started_at = time() machine_name = '' cloud_id = '' @@ -1212,18 +1210,19 @@ def run_script(auth_context_serialized, script_id, machine_id, params='', if not host: raise MistError("No host provided and none could be discovered.") - + started_at = time() result = script.ctl.run( auth_context, machine, host=host, port=port, username=username, password=password, su=su, key_id=key_id, params=params, - job_id=job_id, env=env, owner=owner + job_id=job_id, env=env, owner=owner, ret=ret, + action_prefix=action_prefix ) ret.update(result) - except Exception as exc: + log.info("Script result: %s" % result) + except TypeError as exc: ret['error'] = str(exc) - log_event(event_type='job', action=action_prefix + 'script_started', **ret) - log.info('Script started: %s', ret) + log.error("Script error: %s" % exc) if not ret['error']: if ret['exit_code'] > 0: ret['error'] = 'Script exited with code %s' % ret['exit_code'] From bc4c0b014fb093b498629046247da0a52a5685ad Mon Sep 17 00:00:00 2001 From: Dimitris Moraitis Date: Wed, 7 Dec 2022 01:22:32 +0200 Subject: [PATCH 20/22] Fix create-machine with schedules in api-v2 --- .../api/clouds/controllers/compute/base.py | 121 ++++++++---------- src/mist/api/helpers.py | 2 +- src/mist/api/scheduler.py | 5 +- src/mist/api/schedules/base.py | 7 +- src/mist/api/schedules/models.py | 14 +- v2 | 2 +- 6 files changed, 68 insertions(+), 83 deletions(-) diff --git a/src/mist/api/clouds/controllers/compute/base.py b/src/mist/api/clouds/controllers/compute/base.py index 3a418705c..6983b7141 100644 --- a/src/mist/api/clouds/controllers/compute/base.py +++ b/src/mist/api/clouds/controllers/compute/base.py @@ -3339,36 +3339,6 @@ def _generate_plan__parse_schedules(self, auth_context, """Parse & validate machine's schedules list from the create machine request. - Schedule attributes: - `schedule_type`: 'one_off', 'interval', 'crontab' - `action`: 'start' 'stop', 'reboot', 'destroy' - `script`: dictionary containing: - `script`: id or name of the script to run - `params`: optional script parameters - - one_off schedule_type parameters: - `datetime`: when schedule should run, - e.g '2020-12-15T22:00:00Z' - crontab schedule_type parameters: - `minute`: e.g '*/10', - `hour` - `day_of_month` - `month_of_year` - `day_of_week` - - interval schedule_type parameters: - `every`: int , - `period`: minutes,hours,days - - `expires`: date when schedule should expire, - e.g '2020-12-15T22:00:00Z' - - `start_after`: date when schedule should start running, - e.g '2020-12-15T22:00:00Z' - - `max_run_count`: max number of times to run - description: - Parameters: auth_context(AuthContext): The AuthContext object of the user making the request. @@ -3381,48 +3351,57 @@ def _generate_plan__parse_schedules(self, auth_context, return None ret_schedules = [] for schedule in schedules: - schedule_type = schedule.get('schedule_type') + when = schedule.get('when') + schedule_type = when.get('schedule_type') if schedule_type not in ['crontab', 'interval', 'one_off']: raise BadRequestError('schedule type must be one of ' 'these (crontab, interval, one_off)]') ret_schedule = { - 'schedule_type': schedule_type, 'description': schedule.get('description', ''), - 'task_enabled': True, + 'when': when, + 'task_enabled': schedule.get('task_enabled', True), + 'actions': [] } - action = schedule.get('action') - script = schedule.get('script') - if action is None and script is None: - raise BadRequestError('Schedule action or script not defined') - if action and script: - raise BadRequestError( - 'One of action or script should be defined') - if action: - if action not in ['reboot', 'destroy', 'start', 'stop']: + actions = schedule.get('actions', []) + for action in actions: + action_type = action.get('action_type') + if action_type is None: + raise BadRequestError('Schedule action not defined') + if action_type not in [ + 'reboot', 'destroy', 'start', 'stop', 'delete', 'webhook', + 'notify', 'undefine', 'resize', 'run_script']: raise BadRequestError('Action is not correct') - ret_schedule['action'] = action - else: - from mist.api.methods import list_resources - script_search = script.get('script') - if not script_search: - raise BadRequestError('script parameter is required') - try: - [script_obj], _ = list_resources(auth_context, 'script', - search=script_search, - limit=1) - except ValueError: - raise NotFoundError('Schedule script does not exist') - auth_context.check_perm('script', 'run', script_obj.id) - ret_schedule['script_id'] = script_obj.id - ret_schedule['script_name'] = script_obj.name - ret_schedule['params'] = script.get('params') + ret_action = { + 'action_type': action_type + } + if action_type == 'run_script': + script_type = action.get('script_type') + if script_type == 'existing': + from mist.api.methods import list_resources + script_search = action.get('script') + if not script_search: + raise BadRequestError( + 'script parameter is required') + try: + [script_obj], _ = list_resources( + auth_context, 'script', search=script_search, + limit=1) + except ValueError: + raise NotFoundError( + 'Schedule script does not exist') + auth_context.check_perm('script', 'run', script_obj.id) + ret_action['script'] = script_obj.id + ret_action['script_name'] = script_obj.name + ret_action['params'] = action.get('params') + ret_schedule['actions'].append(ret_action) + if schedule_type == 'one_off': # convert schedule_entry from ISO format # to '%Y-%m-%d %H:%M:%S' try: - ret_schedule['schedule_entry'] = datetime.datetime.strptime( # noqa - schedule['datetime'], '%Y-%m-%dT%H:%M:%SZ' + ret_schedule['when']['datetime'] = datetime.datetime.strptime( # noqa + when['datetime'], '%Y-%m-%dT%H:%M:%SZ' ).strftime("%Y-%m-%d %H:%M:%S") except KeyError: raise BadRequestError( @@ -3433,21 +3412,24 @@ def _generate_plan__parse_schedules(self, auth_context, ' format %Y-%m-%dT%H:%M:%SZ') elif schedule_type == 'interval': try: - ret_schedule['schedule_entry'] = { - 'every': schedule['every'], - 'period': schedule['period'] + ret_schedule['when'] = { + 'schedule_type': 'interval', + 'every': when['every'], + 'period': when['period'], + 'max_run_count': when.get('max_run_count') } except KeyError: raise BadRequestError( 'interval schedule parameter missing') elif schedule_type == 'crontab': try: - ret_schedule['schedule_entry'] = { - 'minute': schedule['minute'], - 'hour': schedule['hour'], - 'day_of_month': schedule['day_of_month'], - 'month_of_year': schedule['month_of_year'], - 'day_of_week': schedule['day_of_week'] + ret_schedule['when'] = { + 'schedule_type': 'crontab', + 'minute': when['minute'], + 'hour': when['hour'], + 'day_of_month': when['day_of_month'], + 'month_of_year': when['month_of_year'], + 'day_of_week': when['day_of_week'] } except KeyError: raise BadRequestError( @@ -3466,6 +3448,7 @@ def _generate_plan__parse_schedules(self, auth_context, 'not match format ' '%Y-%m-%dT%H:%M:%SZ' ) + if schedule.get('expires'): # convert `expires` from ISO format # to '%Y-%m-%d %H:%M:%S' diff --git a/src/mist/api/helpers.py b/src/mist/api/helpers.py index e0a3c5645..6379d6373 100644 --- a/src/mist/api/helpers.py +++ b/src/mist/api/helpers.py @@ -644,7 +644,7 @@ def random_string(length=5, punc=False): When punc=True, the string will also contain punctuation apart from letters and digits """ - _chars = string.letters + string.digits + _chars = string.ascii_letters + string.digits _chars += string.punctuation if punc else '' return ''.join(random.choice(_chars) for _ in range(length)) diff --git a/src/mist/api/scheduler.py b/src/mist/api/scheduler.py index a505a1fa3..733dabd18 100644 --- a/src/mist/api/scheduler.py +++ b/src/mist/api/scheduler.py @@ -27,6 +27,7 @@ def schedule_to_actor(schedule): + task_path = None if isinstance(schedule, PollingSchedule) or isinstance(schedule, Rule): task_path = schedule.task.split('.') else: @@ -75,7 +76,7 @@ def add_job(scheduler, schedule, actor, first_run=False): if schedule_action._cls == 'ScriptAction': job['args'] = ( None, - schedule_action.script, + schedule_action.script.id, schedule.name, [r.id for r in schedule.get_resources()], schedule_action.params, @@ -157,7 +158,7 @@ def update_job(scheduler, schedule, actor, existing): if schedule_action._cls == 'ScriptAction': new_args = ( None, - schedule_action.script, + schedule_action.script.id, schedule.name, [r.id for r in schedule.get_resources()], schedule_action.params, diff --git a/src/mist/api/schedules/base.py b/src/mist/api/schedules/base.py index ecf2987b1..2edc9c7d2 100644 --- a/src/mist/api/schedules/base.py +++ b/src/mist/api/schedules/base.py @@ -6,7 +6,6 @@ """ import logging import datetime -import ast import mongoengine as me from mist.api.scripts.models import Script @@ -142,7 +141,6 @@ def update(self, **kwargs): script_type = kwargs.get('actions')[0].get('script_type') if script_type == 'existing': script = kwargs.get('actions')[0].get('script') - script = ast.literal_eval(script) script_id = script['script'] if script_id: try: @@ -238,9 +236,8 @@ def update(self, **kwargs): # TODO Make possible to have notification actions on schedules raise NotImplementedError() elif action == 'run_script': - script = ast.literal_eval(actions[0]['script']) - script_id = script['script'] - params = script['params'] + script_id = actions[0]['script'] + params = actions[0]['params'] if script_id: self.schedule.actions[0] = acts.ScriptAction( script=script_id, params=params) diff --git a/src/mist/api/schedules/models.py b/src/mist/api/schedules/models.py index f6fa69e91..c468747c6 100644 --- a/src/mist/api/schedules/models.py +++ b/src/mist/api/schedules/models.py @@ -336,10 +336,7 @@ def as_dict(self): selectors = [selector.as_dict() for selector in self.selectors] - if self.actions[0].__class__.__name__ == 'ScriptAction': - action = 'run script' - else: - action = self.actions[0].action + action = self.actions[0] sdict = { 'id': self.id, @@ -348,7 +345,6 @@ def as_dict(self): 'schedule': str(self.when), 'schedule_type': self.when.type, 'schedule_entry': self.when.as_dict(), - 'task_type': action, 'expires': str(self.expires or ''), 'start_after': str(self.start_after or ''), 'task_enabled': self.task_enabled, @@ -361,6 +357,14 @@ def as_dict(self): 'owned_by': self.owned_by.id if self.owned_by else '', 'created_by': self.created_by.id if self.created_by else '', } + task_type = {} + if action.__class__.__name__ == 'ScriptAction': + task_type['action'] = 'run script' + task_type['script_id'] = action.script + task_type['params'] = action.params + else: + task_type['action'] = self.actions[0].atype + sdict['task_type'] = task_type return sdict diff --git a/v2 b/v2 index 9937d42d3..3fbe1b3f9 160000 --- a/v2 +++ b/v2 @@ -1 +1 @@ -Subproject commit 9937d42d363837f6962e77fb6860145edc100bc9 +Subproject commit 3fbe1b3f9b525a15e1c49ec3ba653bf40803e647 From 3ee1a2123610c6b64e97c7b6707f573c8e72963c Mon Sep 17 00:00:00 2001 From: Dimitris Moraitis Date: Wed, 7 Dec 2022 01:41:52 +0200 Subject: [PATCH 21/22] Update v2 submodule --- v2 | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/v2 b/v2 index 3fbe1b3f9..6673b8a73 160000 --- a/v2 +++ b/v2 @@ -1 +1 @@ -Subproject commit 3fbe1b3f9b525a15e1c49ec3ba653bf40803e647 +Subproject commit 6673b8a7346f48ba64daa149a2e9c9182e991a0d From efea91a9753df0c0db59ac74cabf5a40863740b6 Mon Sep 17 00:00:00 2001 From: Dimitris Moraitis Date: Wed, 7 Dec 2022 14:55:01 +0200 Subject: [PATCH 22/22] Update ipython, hide `run_immediately` in schedule dict --- requirements-frozen.txt | 2 +- src/mist/api/schedules/models.py | 1 - 2 files changed, 1 insertion(+), 2 deletions(-) diff --git a/requirements-frozen.txt b/requirements-frozen.txt index e3f8d56e0..47efdb8ec 100644 --- a/requirements-frozen.txt +++ b/requirements-frozen.txt @@ -32,7 +32,7 @@ greenlet==1.1.3.post0 idna==2.10 ipaddress==1.0.23 ipdb==0.13.9 -ipython==7.34.0 +ipython==8.7.0 ipython-genutils==0.2.0 iso8601==0.1.16 jedi==0.18.1 diff --git a/src/mist/api/schedules/models.py b/src/mist/api/schedules/models.py index c468747c6..bb36d2721 100644 --- a/src/mist/api/schedules/models.py +++ b/src/mist/api/schedules/models.py @@ -349,7 +349,6 @@ def as_dict(self): 'start_after': str(self.start_after or ''), 'task_enabled': self.task_enabled, 'active': self.enabled, - 'run_immediately': self.run_immediately or '', 'last_run_at': last_run, 'total_run_count': self.total_run_count, 'max_run_count': self.max_run_count,