feat(helmbroker): add instance lock

Author: duanhongyi

charts/helmbroker/templates/helmbroker-clusterrole.yaml

@@ -8,7 +8,7 @@ metadata:
 roleRef:
   apiGroup: rbac.authorization.k8s.io
   kind: ClusterRole
-  name: drycc:drycc-helmbroker
+  name: cluster-admin
 subjects:
 - kind: ServiceAccount
   name: drycc-helmbroker

charts/helmbroker/templates/helmbroker-clusterrolebinding.yaml

@@ -1,4 +1,5 @@
 import os
+import time
 import shutil
 from typing import Union, List, Optional
 
@@ -12,9 +13,10 @@
     UpdateDetails, UpdateServiceSpec, DeprovisionDetails, \
     DeprovisionServiceSpec, LastOperation, OperationState
 
-from .meta import load_instance_meta, load_binding_meta
+from .meta import load_instance_meta, load_binding_meta, dump_instance_meta
 from .utils import get_instance_path, get_chart_path, get_plan_path, \
-    get_addon_path, get_addon_updateable, get_addon_bindable
+    get_addon_path, get_addon_updateable, get_addon_bindable, InstanceLock, \
+    get_instance_file
 from .tasks import provision, bind, deprovision, update
 from helmbroker.meta import load_addons_meta
 
@@ -43,10 +45,29 @@ def provision(self,
         if not async_allowed:
             raise ErrAsyncRequired()
         os.makedirs(instance_path, exist_ok=True)
-        chart_path, plan_path = get_chart_path(instance_id), get_plan_path(instance_id)  # noqa
-        addon_chart_path, addon_plan_path = get_addon_path(details.service_id, details.plan_id)  # noqa
+        chart_path, plan_path = (
+            get_chart_path(instance_id), get_plan_path(instance_id))
+        addon_chart_path, addon_plan_path = (
+            get_addon_path(details.service_id, details.plan_id))
         shutil.copytree(addon_chart_path, chart_path)
         shutil.copytree(addon_plan_path, plan_path)
+        data = {
+            "id": instance_id,
+            "details": {
+                "service_id": details.service_id,
+                "plan_id": details.plan_id,
+                "context": details.context,
+                "parameters": details.parameters,
+            },
+            "last_operation": {
+                "state": OperationState.IN_PROGRESS.value,
+                "description": (
+                    "provision %s in progress at %s" % (
+                        instance_id, time.time()))
+            }
+        }
+        with InstanceLock(instance_id):
+            dump_instance_meta(instance_id, data)
         provision.delay(instance_id, details)
         return ProvisionedServiceSpec(state=ProvisionState.IS_ASYNC)
 
@@ -69,15 +90,18 @@ def bind(self,
              ) -> Binding:
         is_addon_bindable = get_addon_bindable(details.service_id)
         if not is_addon_bindable:
-            raise ErrBadRequest(msg="Instance %s does not bindable" % instance_id)  # noqa
+            raise ErrBadRequest(
+                msg="Instance %s does not bindable" % instance_id)
         instance_meta = load_instance_meta(instance_id)
         if not (instance_meta and
                 instance_meta['last_operation']['state'] == 'succeeded'):
-            raise ErrBadRequest(msg="This instance %s is not ready" % instance_id)  # noqa
+            raise ErrBadRequest(
+                msg="This instance %s is not ready" % instance_id)
         instance_path = get_instance_path(instance_id)
         if os.path.exists(f'{instance_path}/bind.json'):
             raise ErrBindingAlreadyExists()
-        chart_path, plan_path = get_chart_path(instance_id), get_plan_path(instance_id)  # noqa
+        chart_path, plan_path = (
+            get_chart_path(instance_id), get_plan_path(instance_id))
         shutil.copy(f'{plan_path}/bind.yaml', f'{chart_path}/templates')
         bind(instance_id, binding_id, details, async_allowed, **kwargs)
         data = load_binding_meta(instance_id)
@@ -111,14 +135,16 @@ def update(self,
             raise ErrBadRequest(msg="Instance %s does not exist" % instance_id)
         is_plan_updateable = get_addon_updateable(details.service_id)
         if not is_plan_updateable:
-            raise ErrBadRequest(msg="Instance %s does not updateable" % instance_id)  # noqa
+            raise ErrBadRequest(
+                msg="Instance %s does not updateable" % instance_id)
         if not async_allowed:
             raise ErrAsyncRequired()
         if details.plan_id is not None:
             plan_path = get_plan_path(instance_id)
             # delete the pre plan
             shutil.rmtree(plan_path, ignore_errors=True)
-            _, addon_plan_path = get_addon_path(details.service_id, details.plan_id)  # noqa
+            _, addon_plan_path = get_addon_path(
+                details.service_id, details.plan_id)
             # add the new plan
             shutil.copytree(addon_plan_path, plan_path)
         update.delay(instance_id, details)
@@ -130,11 +156,14 @@ def deprovision(self,
                     async_allowed: bool,
                     **kwargs) -> DeprovisionServiceSpec:
         instance_path = get_instance_path(instance_id)
-        if not os.path.exists(instance_path):
+        if os.path.exists(instance_path):
+            if not os.path.exists(get_instance_file(instance_id)):
+                return DeprovisionServiceSpec(
+                    is_async=False, operation=OperationState.SUCCEEDED)
+        else:
             raise ErrInstanceDoesNotExist()
         if not async_allowed:
             raise ErrAsyncRequired()
-
         deprovision.delay(instance_id)
         return DeprovisionServiceSpec(is_async=True)
 

rootfs/helmbroker/broker.py

@@ -8,13 +8,14 @@
 from .config import INSTANCES_PATH
 from .meta import load_instance_meta
 from .tasks import deprovision
+from .utils import get_instance_file
 
 logger = logging.getLogger(__name__)
 
 
 def clean_instance():
     for instance_id in os.listdir(INSTANCES_PATH):
-        if os.path.exists(os.path.join(INSTANCES_PATH, instance_id, "instance.json")):  # noqa
+        if os.path.exists(get_instance_file(instance_id)):  # noqa
             data = load_instance_meta(instance_id)
             interval = time.time() - data["last_modified_time"]
             if interval > 3600 * 24 and data["last_operation"]["state"] != OperationState.SUCCEEDED:  # noqa

rootfs/helmbroker/cleaner.py

@@ -2,7 +2,8 @@
 import json
 import time
 from jsonschema import validate
-from .config import INSTANCES_PATH, ADDONS_PATH
+from .utils import get_instance_path, get_instance_file
+from .config import ADDONS_PATH
 
 INSTANCE_META_SCHEMA = {
     "type": "object",
@@ -35,7 +36,7 @@
 
 
 def load_instance_meta(instance_id):
-    file = os.path.join(INSTANCES_PATH, instance_id, "instance.json")
+    file = get_instance_file(instance_id)
     with open(file) as f:
         data = json.load(f)
         validate(instance=data, schema=INSTANCE_META_SCHEMA)
@@ -44,7 +45,7 @@ def load_instance_meta(instance_id):
 
 def dump_instance_meta(instance_id, data):
     data["last_modified_time "] = time.time()
-    file = os.path.join(INSTANCES_PATH, instance_id, "instance.json")
+    file = get_instance_file(instance_id)
     validate(instance=data, schema=INSTANCE_META_SCHEMA)
     with open(file, "w") as f:
         f.write(json.dumps(data, sort_keys=True, indent=2))
@@ -70,7 +71,7 @@ def dump_instance_meta(instance_id, data):
 
 
 def load_binding_meta(instance_id):
-    file = os.path.join(INSTANCES_PATH, instance_id, "binding.json")
+    file = os.path.join(get_instance_path(instance_id), "binding.json")
     with open(file, 'r') as f:
         data = json.loads(f.read())
         validate(instance=data, schema=INSTANCE_META_SCHEMA)
@@ -79,7 +80,7 @@ def load_binding_meta(instance_id):
 
 def dump_binding_meta(instance_id, data):
     data["last_modified_time "] = time.time()
-    file = os.path.join(INSTANCES_PATH, instance_id, "binding.json")
+    file = os.path.join(get_instance_path(instance_id), "binding.json")
     validate(instance=data, schema=INSTANCE_META_SCHEMA)
     with open(file, "w") as f:
         f.write(json.dumps(data, sort_keys=True, indent=2))

rootfs/helmbroker/meta.py

@@ -1,57 +1,50 @@
 import os
+import shutil
 import time
 import yaml
 
 from openbrokerapi.service_broker import ProvisionDetails, OperationState, \
     UpdateDetails, BindDetails
 
 from .celery import app
-from .utils import command, get_plan_path, get_chart_path, get_cred_value
+from .utils import command, get_plan_path, get_chart_path, get_cred_value, \
+    InstanceLock, get_instance_file
 from .meta import dump_instance_meta, dump_binding_meta, load_instance_meta
 
 
 @app.task(serializer='pickle')
 def provision(instance_id: str, details: ProvisionDetails):
-    data = {
-        "id": instance_id,
-        "details": {
-            "service_id": details.service_id,
-            "plan_id": details.plan_id,
-            "context": details.context,
-            "parameters": details.parameters,
-        },
-        "last_operation": {
-            "state": OperationState.IN_PROGRESS.value,
-            "description": "provision %s in progress at %s" % (instance_id, time.time())  # noqa
-        }
-    }
-    dump_instance_meta(instance_id, data)
-    chart_path = get_chart_path(instance_id)
-    values_file = os.path.join(get_plan_path(instance_id), "values.yaml")
-    args = [
-        "install",
-        details.context["instance_name"],
-        chart_path,
-        "--namespace",
-        details.context["namespace"],
-        "--create-namespace",
-        "--wait",
-        "--timeout",
-        "10m0s",
-        "-f",
-        values_file,
-        "--set",
-        f"fullnameOverride={details.context['instance_name']}"
-    ]
+    with InstanceLock(instance_id):
+        chart_path = get_chart_path(instance_id)
+        values_file = os.path.join(get_plan_path(instance_id), "values.yaml")
+        args = [
+            "install",
+            details.context["instance_name"],
+            chart_path,
+            "--namespace",
+            details.context["namespace"],
+            "--create-namespace",
+            "--wait",
+            "--timeout",
+            "10m0s",
+            "-f",
+            values_file,
+            "--set",
+            f"fullnameOverride={details.context['instance_name']}"
+        ]
 
-    status, output = command("helm", *args)
-    if status != 0:
-        data["last_operation"]["state"] = OperationState.FAILED.value
-        data["last_operation"]["description"] = "provision error:\n%s" % output
-    else:
-        data["last_operation"]["state"] = OperationState.SUCCEEDED.value
-        data["last_operation"]["description"] = "provision succeeded at %s" % time.time()  # noqa
-    dump_instance_meta(instance_id, data)
+        status, output = command("helm", *args)
+        data = load_instance_meta(instance_id)
+        if status != 0:
+            data["last_operation"]["state"] = OperationState.FAILED.value
+            data["last_operation"]["description"] = (
+                "provision error:\n%s" % output)
+        else:
+            data["last_operation"]["state"] = OperationState.SUCCEEDED.value
+            data["last_operation"]["description"] = (
+                "provision succeeded at %s" % time.time())
+        print(data)
+        dump_instance_meta(instance_id, data)
 
 
 @app.task(serializer='pickle')
@@ -158,21 +151,29 @@ def bind(instance_id: str,
 
 @app.task()
 def deprovision(instance_id: str):
-    data = load_instance_meta(instance_id)
-    data["last_operation"]["state"] = OperationState.IN_PROGRESS.value
-    data["last_operation"]["description"] = "deprovision %s in progress at %s" % (instance_id, time.time())  # noqa
-    dump_instance_meta(instance_id, data)
-    status, output = command(
-        "helm",
-        "uninstall",
-        data["details"]["context"]["instance_name"],
-        "--namespace",
-        data["details"]["context"]["namespace"],
-    )
-    if status != 0:
-        data["last_operation"]["state"] = OperationState.FAILED.value
-        data["last_operation"]["description"] = "deprovision error:\n%s" % output  # noqa
-    else:
-        data["last_operation"]["state"] = OperationState.SUCCEEDED.value
-        data["last_operation"]["description"] = "deprovision succeeded at %s" % time.time()  # noqa
-    dump_instance_meta(instance_id, data)
+    with InstanceLock(instance_id):
+        data = load_instance_meta(instance_id)
+        data["last_operation"]["state"] = OperationState.IN_PROGRESS.value
+        data["last_operation"]["description"] = "deprovision %s in progress at %s" % (instance_id, time.time())  # noqa
+        dump_instance_meta(instance_id, data)
+        status, output = command(
+            "helm",
+            "uninstall",
+            data["details"]["context"]["instance_name"],
+            "--namespace",
+            data["details"]["context"]["namespace"],
+        )
+        if status != 0:
+            data["last_operation"]["state"] = OperationState.FAILED.value
+            data["last_operation"]["description"] = (
+                "deprovision error:\n%s" % output)
+            shutil.copy(get_instance_file(instance_id), "%s.%s" % (
+                get_instance_file(instance_id),
+                time.time()
+            ))
+        else:
+            data["last_operation"]["state"] = OperationState.SUCCEEDED.value
+            data["last_operation"]["description"] = (
+                "deprovision succeeded at %s" % time.time())
+        os.remove(get_instance_file(instance_id))
+        dump_instance_meta(instance_id, data)