experiment manager 重构

open-hackathon-server/requirements.txt

@@ -1,3 +1,4 @@
+amqp==2.5.2
 aniso8601==8.0.0
 APScheduler==3.0.3
 attrs==19.3.0
@@ -15,7 +16,9 @@ azure-servicebus==0.20.1
 azure-servicemanagement-legacy==0.20.1
 azure-storage==0.20.2
 Beaker==1.11.0
+billiard==3.6.3.0
 cachetools==4.0.0
+celery==4.4.2
 certifi==2019.11.28
 chardet==3.0.4
 click==7.1.1
@@ -30,6 +33,7 @@ idna==2.9
 importlib-metadata==1.5.0
 itsdangerous==1.1.0
 Jinja2==2.11.1
+kombu==4.6.8
 kubernetes==9.0.0
 lxml==4.3.0
 mailthon==0.1.1
@@ -62,6 +66,7 @@ subprocess32==3.5.4
 tzlocal==2.0.0
 urllib3==1.25.8
 validictory==1.0.0
+vine==1.3.0
 wcwidth==0.1.8
 websocket-client==0.57.0
 Werkzeug==0.15.3

open-hackathon-server/src/hackathon/__init__.py

@@ -179,9 +179,8 @@ def init_db():
 
 
 def init_expr_components():
-    from .expr import ExprManager, K8SExprStarter
+    from .expr import ExprManager
     factory.provide("expr_manager", ExprManager)
-    factory.provide("k8s_service", K8SExprStarter)
 
 
 def init_voice_verify():

open-hackathon-server/src/hackathon/cloud_providers/__init__.py

@@ -0,0 +1,114 @@
+import time
+import random
+import string
+import logging
+
+from hackathon.hmongo.models import K8sEnvironment, VirtualEnvironment
+from hackathon.constants import VE_PROVIDER, EStatus, VEStatus, VERemoteProvider
+
+__all__ = ["Provider", "ProviderError"]
+
+LOG = logging.getLogger(__name__)
+
+_provider_classes = {}
+
+
+class ProviderError(Exception):
+    pass
+
+
+class Provider:
+    def __init__(self, provider_type, provider_cfg):
+        if provider_type not in _provider_classes:
+            raise ProviderError("{} provider not found", provider_type)
+        self.provider_type = provider_type
+        self.p_class = _provider_classes[provider_type]
+        self.p = self.p_class(provider_cfg)
+
+        try:
+            self.p.connect()
+        except Exception as e:
+            raise ProviderError("connect Provider Platform error: {}".format(e))
+
+    def create_expr(self, experiment, template_content):
+        LOG.info("creating experiment {}".format(experiment.id))
+        start_at = time.time()
+        experiment.status = EStatus.STARTING
+        experiment.save()
+
+        try:
+            new_env = self.p.create_instance(self._get_ve_config(experiment, template_content))
+            experiment.virtual_environments[0] = new_env
+            experiment.save()
+
+            self.wait_for_ready()
+        except Exception as e:
+            LOG.error("wait for start error: {}".format(e))
+            experiment.status = EStatus.FAILED
+        else:
+            experiment.virtual_environments[0].status = VEStatus.RUNNING
+            experiment.status = EStatus.RUNNING
+        experiment.save()
+        LOG.info("create expr spend time: {}".format(time.time() - start_at))
+
+    def start_expr(self, experiment):
+        # TODO
+        pass
+
+    def pause_expr(self, experiment):
+        # TODO
+        pass
+
+    def delete_expr(self, experiment):
+        LOG.info("deleting experiment {}".format(experiment.id))
+
+        try:
+            self.p.delete_instance(experiment.virtual_environments[0])
+        except Exception as e:
+            LOG.error("delete expr error: {}".format(e))
+            experiment.status = EStatus.FAILED
+        else:
+            experiment.status = EStatus.STOPPED
+        experiment.save()
+        LOG.info("experiment {} deleted".format(experiment.id))
+
+    def wait_for_ready(self):
+        self.p.wait_instance_ready()
+
+    def get_remote_config(self):
+        pass
+
+    def _get_ve_config(self, experiment, template_content):
+        if not experiment.virtual_environments or len(experiment.virtual_environments) == 0:
+            experiment.virtual_environments = [self._init_ve_config(experiment, template_content)]
+            experiment.save()
+
+        return experiment.virtual_environments[0]
+
+    def _init_ve_config(self, experiment, template_content):
+        if self.provider_type == VE_PROVIDER.K8S:
+            hackathon = experiment.hackathon
+            unit = template_content.unit
+            env_name = "{}-{}".format(hackathon.name, "".join(random.sample(string.ascii_lowercase, 6)))
+            yaml_content = unit.gen_k8s_yaml(env_name)
+            k8s_env = K8sEnvironment.load_from_yaml(env_name, yaml_content)
+            return VirtualEnvironment(
+                provider=VE_PROVIDER.K8S,
+                name=env_name,
+                status=VEStatus.INIT,
+                remote_provider=VERemoteProvider.Guacamole,
+                k8s_resource=k8s_env,
+            )
+        raise ProviderError("not found {} instance config".format(self.provider_type))
+
+
+def register_provider(provider_type, provider_class):
+    _provider_classes[provider_type] = provider_class
+
+
+def _register():
+    from .k8s import K8sProvider
+    register_provider(VE_PROVIDER.K8S, K8sProvider)
+
+
+_register()