fix updater, deployer, waiter to support new architecture

Author: OriHoch

cwm_worker_operator/config.py

@@ -150,3 +150,7 @@
 THROTTLER_THROTTLE_MAX_REQUESTS = int(os.environ.get('THROTTLER_THROTTLE_MAX_REQUESTS') or '10000000')
 # period of time during which the maximum number of requests should be reached to throttle
 THROTTLER_CHECK_TTL_SECONDS = int(os.environ.get('THROTTLER_CHECK_TTL_SECONDS') or '60')
+
+MINIO_TENANT_ENDPOINT = os.environ.get('MINIO_TENANT_ENDPOINT')
+MINIO_TENANT_ADMIN_USER = os.environ.get('MINIO_TENANT_ADMIN_USER')
+MINIO_TENANT_ADMIN_PASSWORD = os.environ.get('MINIO_TENANT_ADMIN_PASSWORD')

cwm_worker_operator/deployments_manager.py

@@ -12,6 +12,9 @@
 
 import requests
 from ruamel import yaml
+from minio.minioadmin import MinioAdmin
+from minio.credentials.providers import StaticProvider
+from minio import Minio
 
 from cwm_worker_operator import logs
 from cwm_worker_operator import config
@@ -48,6 +51,17 @@ def parse_datetime_from_kubelet_log_line(line):
     return datetime.datetime.strptime('{}{} {}+00:00'.format(datetime.datetime.now().year, datepart, timepart), '%YI%m%d %H:%M:%S.%f%z')
 
 
+def get_minio_admin():
+    return MinioAdmin(
+        config.MINIO_TENANT_ENDPOINT,
+        StaticProvider(config.MINIO_TENANT_ADMIN_USER, config.MINIO_TENANT_ADMIN_PASSWORD)
+    )
+
+
+def get_minio():
+    return Minio(config.MINIO_TENANT_ENDPOINT, config.MINIO_TENANT_ADMIN_USER, config.MINIO_TENANT_ADMIN_PASSWORD)
+
+
 class NodeCleanupPod:
 
     def __init__(self, namespace_name, pod_name, node_name):
@@ -151,22 +165,65 @@ def init_cache(self):
                 print("Failed to initialize chart cache for version {}".format(version))
 
     def init(self, deployment_config):
-        cwm_worker_deployment.deployment.init(deployment_config)
+        pass
+        # cwm_worker_deployment.deployment.init(deployment_config)
 
     def deploy_external_service(self, deployment_config):
-        cwm_worker_deployment.deployment.deploy_external_service(deployment_config)
+        pass
+        # cwm_worker_deployment.deployment.deploy_external_service(deployment_config)
 
     def deploy_extra_objects(self, deployment_config, extra_objects):
-        cwm_worker_deployment.deployment.deploy_extra_objects(deployment_config, extra_objects)
+        pass
+        # cwm_worker_deployment.deployment.deploy_extra_objects(deployment_config, extra_objects)
 
     def deploy_preprocess_specs(self, specs):
-        return cwm_worker_deployment.deployment.deploy_preprocess_specs(specs)
-
-    def deploy(self, deployment_config, **kwargs):
-        return cwm_worker_deployment.deployment.deploy(deployment_config, **kwargs)
+        res = {}
+        for key, spec in specs.items():
+            # currently we don't have any preprocess spec, but we should set True to prevent re-preprocessing
+            res[key] = True
+        return res
+
+    def deploy(self, deployment_config, dry_run=False, **kwargs):
+        # return cwm_worker_deployment.deployment.deploy(deployment_config, **kwargs)
+        username = deployment_config['cwm-worker-deployment']['namespace']
+        password = deployment_config['minio']['access_key']
+        if dry_run:
+            return f"username: {username}"
+        else:
+            minio_admin = get_minio_admin()
+            minio_admin.user_add(username, password)
+            with tempfile.NamedTemporaryFile('w') as f:
+                json.dump({
+                    "Version": "2012-10-17",
+                    "Statement": [
+                        {
+                            "Action": ["s3:*"],
+                            "Effect": "Allow",
+                            "Resource": [f"arn:aws:s3:::{username}",f"arn:aws:s3:::{username}/*"]
+                        },
+                        {
+                            "Action": ["s3:DeleteBucket"],
+                            "Effect": "Deny",
+                            "Resource": [f"arn:aws:s3:::{username}"]
+                        },
+                        {
+                            "Action": ["s3:ListAllMyBuckets"],
+                            "Effect": "Allow",
+                            "Resource": ["arn:aws:s3:::*"]
+                        }
+                    ]
+                }, f)
+                f.seek(0)
+                minio_admin.policy_add(username, f.name)
+            minio_admin.policy_set(username, username)
+            minio = get_minio()
+            if not minio.bucket_exists(username):
+                minio.make_bucket(username)
+            return f"OK, username/bucket: {username}"
 
     def is_ready(self, namespace_name, deployment_type, minimal_check=False):
-        return cwm_worker_deployment.deployment.is_ready(namespace_name, deployment_type, minimal_check=minimal_check)
+        #return cwm_worker_deployment.deployment.is_ready(namespace_name, deployment_type, minimal_check=minimal_check)
+        return get_minio().bucket_exists(namespace_name)
 
     def get_health(self, namespace_name, deployment_type):
         return cwm_worker_deployment.deployment.get_health(namespace_name, deployment_type)
@@ -209,8 +266,8 @@ def delete(self, namespace_name, deployment_type, **kwargs):
         cwm_worker_deployment.deployment.delete(namespace_name, deployment_type, **kwargs)
 
     def iterate_all_releases(self):
-        for release in cwm_worker_deployment.helm.iterate_all_releases("minio"):
-            yield release
+        for user in json.loads(get_minio_admin().user_list()):
+            yield user
 
     def get_prometheus_metrics(self, namespace_name):
         metrics = {}

cwm_worker_operator/domains_config.py

@@ -252,8 +252,8 @@ def __init__(self, data, domains_config, request_hostname=None, is_data_from_cac
                         'payload': hostname['payload']
                     }
         self.primary_hostname = data.get('primary instance')
-        self.client_id = data.get("client_id")
-        self.secret = data.get("secret")
+        self.secret = self.client_id = data.get("client_id")
+        # self.secret = data.get("secret")
         self.cache_enabled = bool(data.get('cache'))
         try:
             self.cache_exclude_extensions = [ext.strip() for ext in data.get('cache-exclude', '').split('|') if ext.strip()]

cwm_worker_operator/updater.py

@@ -14,6 +14,7 @@
 from cwm_worker_operator.daemon import Daemon
 from cwm_worker_operator.domains_config import DomainsConfig
 from cwm_worker_operator.multiprocessor import Multiprocessor
+from cwm_worker_operator.deployments_manager import DeploymentsManager
 
 
 DATETIME_FORMAT = '%Y%m%dT%H%M%S%z'
@@ -49,14 +50,14 @@ def check_update_release(domains_config, updater_metrics, namespace_name, last_u
             domains_config.set_worker_force_delete(worker_id, allow_cancel=False)
             if updater_metrics:
                 updater_metrics.force_delete(worker_id, start_time)
-        elif instance_update == 'update':
-            msg = "domain force update (from cwm updates api)"
+        elif instance_update in ('update', 'create'):
+            msg = f"domain force {instance_update} (from cwm updates api)"
             logs.debug(msg, debug_verbosity=4, worker_id=worker_id, start_time=start_time)
             domains_config.set_worker_force_update(worker_id)
             if updater_metrics:
                 updater_metrics.not_deployed_force_update(worker_id, start_time)
         else:
-            hours_since_last_update = (common.now() - last_updated).total_seconds() / 60 / 60
+            hours_since_last_update = ((common.now() - last_updated).total_seconds() / 60 / 60)
             volume_config = domains_config.get_cwm_api_volume_config(worker_id=worker_id)
             disable_force_delete = volume_config.disable_force_delete
             disable_force_update = volume_config.disable_force_update
@@ -150,12 +151,14 @@ def get_instances_updates(domains_config: DomainsConfig, cwm_api_manager: CwmApi
 
 def update(namespace_name, last_updated, status, revision,
            worker_id, instance_update, start_time,
-           cwm_api_manager=None, domains_config=None, updater_metrics=None):
+           cwm_api_manager=None, domains_config=None, updater_metrics=None, deployments_manager=None):
     if not domains_config:
         domains_config = DomainsConfig()
     if not cwm_api_manager:
         cwm_api_manager = CwmApiManager()
-    last_updated = get_datetime_object(last_updated)
+    if not deployments_manager:
+        deployments_manager = DeploymentsManager()
+    last_updated = get_datetime_object(last_updated) if last_updated else None
     start_time = get_datetime_object(start_time)
     check_update_release(domains_config, updater_metrics, namespace_name, last_updated, status, revision,
                          instance_update, worker_id, start_time)
@@ -196,19 +199,27 @@ def run_single_iteration(domains_config, metrics, deployments_manager, cwm_api_m
     multiprocessor = UpdaterMultiprocessor(config.UPDATER_MAX_PARALLEL_DEPLOY_PROCESSES if is_async else 1)
     updater_metrics = metrics
     instances_updates = get_instances_updates(domains_config, cwm_api_manager)
-    all_releases = {release["namespace"]: release for release in deployments_manager.iterate_all_releases()}
-    for release in all_releases.values():
-        namespace_name = release["namespace"]
-        datestr, timestr, *_ = release["updated"].split(" ")
-        last_updated = common.strptime("{}T{}".format(datestr, timestr.split(".")[0]), "%Y-%m-%dT%H:%M:%S")
-        status = release["status"]
-        # app_version = release["app_version"]
-        revision = int(release["revision"])
-        start_time = common.now()
-        worker_id = common.get_worker_id_from_namespace_name(namespace_name)
-        instance_update = instances_updates.get(worker_id)
-        multiprocessor.process(domains_config, updater_metrics, namespace_name, last_updated, status, revision,
-                               worker_id, instance_update, start_time, cwm_api_manager)
+    updated_instances = set()
+    # all_releases = {release["namespace"]: release for release in deployments_manager.iterate_all_releases()}
+    # for release in all_releases.values():
+    #     updated_instances.add(namespace_name)
+    #     namespace_name = release["namespace"]
+    #     datestr, timestr, *_ = release["updated"].split(" ")
+    #     last_updated = common.strptime("{}T{}".format(datestr, timestr.split(".")[0]), "%Y-%m-%dT%H:%M:%S")
+    #     status = release["status"]
+    #     # app_version = release["app_version"]
+    #     revision = int(release["revision"])
+    #     start_time = common.now()
+    #     worker_id = common.get_worker_id_from_namespace_name(namespace_name)
+    #     instance_update = instances_updates.get(worker_id)
+    #     multiprocessor.process(domains_config, updater_metrics, namespace_name, last_updated, status, revision,
+    #                            worker_id, instance_update, start_time, cwm_api_manager)
+    for namespace_name, operation in instances_updates.items():
+        if operation == 'update' and namespace_name not in updated_instances:
+            start_time = common.now()
+            worker_id = common.get_worker_id_from_namespace_name(namespace_name)
+            multiprocessor.process(domains_config, updater_metrics, namespace_name, None, None, None,
+                                   worker_id, 'create', start_time, cwm_api_manager)
     multiprocessor.finalize()
 
 

cwm_worker_operator/waiter.py

@@ -17,21 +17,22 @@
 
 
 def _check_for_deployment_complete(domains_config, deployments_manager, waiter_metrics, start_time, log_kwargs, namespace_name, flow_manager, worker_id, volume_config):
-    has_hostnames_without_cert_but_with_challenge = False
-    check_hostname_challenge = None
-    for hostname in volume_config.hostnames:
-        if hostname not in volume_config.hostname_certs and hostname in volume_config.hostname_challenges:
-            has_hostnames_without_cert_but_with_challenge = True
-            check_hostname_challenge = {
-                'host': hostname,
-                **volume_config.hostname_challenges[hostname]
-            }
-            break
-    if deployments_manager.is_ready(namespace_name, "minio", minimal_check=has_hostnames_without_cert_but_with_challenge):
-        internal_hostname = deployments_manager.get_hostname(namespace_name, "minio")
+    # has_hostnames_without_cert_but_with_challenge = False
+    # check_hostname_challenge = None
+    # for hostname in volume_config.hostnames:
+    #     if hostname not in volume_config.hostname_certs and hostname in volume_config.hostname_challenges:
+    #         has_hostnames_without_cert_but_with_challenge = True
+    #         check_hostname_challenge = {
+    #             'host': hostname,
+    #             **volume_config.hostname_challenges[hostname]
+    #         }
+    #         break
+    if deployments_manager.is_ready(namespace_name, "minio"):  # , minimal_check=has_hostnames_without_cert_but_with_challenge):
+        # internal_hostname = deployments_manager.get_hostname(namespace_name, "minio")
+        internal_hostname = {'http': '--', 'https': '--'}
         ok = True
-        if config.WAITER_VERIFY_WORKER_ACCESS:
-            ok = deployments_manager.verify_worker_access(internal_hostname, log_kwargs, check_hostname_challenge=check_hostname_challenge)
+        # if config.WAITER_VERIFY_WORKER_ACCESS:
+        #     ok = deployments_manager.verify_worker_access(internal_hostname, log_kwargs, check_hostname_challenge=check_hostname_challenge)
         if ok:
             flow_manager.set_worker_available(worker_id, internal_hostname)
             if waiter_metrics:

requirements.txt

@@ -7,3 +7,4 @@ click==7.1.2
 uvicorn[standard]==0.27.0.post1
 fastapi==0.109.1
 gunicorn==21.2.0
+minio==7.2.3