feat(controller):add drycc resource cmd

Author: lijianguo

charts/controller/templates/controller-clusterrole.yaml

@@ -67,9 +67,9 @@ rules:
   verbs: ["get", "list", "watch", "create", "delete", "patch", "update"]
 - apiGroups: ["servicecatalog.k8s.io"]
   resources: ["serviceinstances"]
-  verbs: ["get", "list", "watch", "create", "delete", "update"]
+  verbs: ["get", "list", "watch", "create", "delete", "patch", "update"]
 - apiGroups: ["servicecatalog.k8s.io"]
-  resources: ["serviceinstances"]
-  verbs: ["get", "list", "watch", "create", "delete", "update"]
+  resources: ["servicebindings"]
+  verbs: ["get", "list", "watch", "create", "delete", "patch", "update"]
 {{- end -}}
 {{- end -}}

rootfs/api/models/resource.py

@@ -71,7 +71,7 @@ def attach(self, *args, **kwargs):
 
     @transaction.atomic
     def delete(self, *args, **kwargs):
-        if self.binding and self.binding == "Ready":
+        if self.binding == "Ready":
             raise DryccException("the plan is still binding")
         # Deatch ServiceInstance, updates k8s
         self.detach(*args, **kwargs)
@@ -84,18 +84,18 @@ def detach(self, *args, **kwargs):
             self._scheduler.servicecatalog.get_instance(self.app.id, self.name)
             self._scheduler.servicecatalog.delete_instance(self.app.id, self.name)
         except KubeException as e:
-            raise ServiceUnavailable("Could not delete volume {} for application {}".format(name, self.app_id)) from e  # noqa
+            raise ServiceUnavailable("Could not delete resource {} for application {}".format(self.name, self.app_id)) from e  # noqa
 
     def log(self, message, level=logging.INFO):
         """Logs a message in the context of this service.
 
         This prefixes log messages with an application "tag" that the customized
         drycc-logspout will be on the lookout for.  When it's seen, the message-- usually
         an application event of some sort like releasing or scaling, will be considered
-        as "belonging" to the application instead of the controller and will be handled
+        as "belonging" to the application instead of the contoller and will be handled
         accordingly.
         """
-        logger.log(level, "[{}]: {}".format(self.id, message))
+        logger.log(level, "[{}]: {}".format(self.uuid, message))
 
     def bind(self, *args, **kwargs):
         if self.status != "Ready":
@@ -124,12 +124,14 @@ def bind(self, *args, **kwargs):
                 raise ServiceUnavailable(msg) from e
 
     def unbind(self, *args, **kwargs):
-        if self.binding != "Ready":
+        if not self.binding:
             raise DryccException("the resource is not binding")
         try:
             # We raise an exception when a resource doesn't exist
             self._scheduler.servicecatalog.get_binding(self.app.id, self.name)
             self._scheduler.servicecatalog.delete_binding(self.app.id, self.name)
+            self.binding = None
+            self.save()
         except KubeException as e:
             raise ServiceUnavailable("Could not unbind resource {} for application {}".format(self.name, self.app_id)) from e  # noqa
 
@@ -138,21 +140,22 @@ def attach_update(self, *args, **kwargs):
             data = self._scheduler.servicecatalog.get_instance(
                 self.app.id, self.name).json()
         except KubeException as e:
-            self.log("certificate {} does not exist".format(self.app.id),
-                     level=logging.INFO)
-            data = None
-            logger.info(e)
+            self.DryccException("resource {} does not exist".format(self.name))
         try:
             version = data["metadata"]["resourceVersion"]
             instance = self.plan.split(":")
             kwargs = {
                 "instance_class": instance[0],
                 "instance_plan": ":".join(instance[1:]),
                 "parameters": self.options,
+                "external_id": data["spec"]["externalID"]
             }
             self._scheduler.servicecatalog.put_instance(
                 self.app.id, self.name, version, **kwargs
             )
+            # self._scheduler.servicecatalog.patch_instance(
+            #     self.app.id, self.name, version, **kwargs
+            # )
             # create/patch/put  retrieve_task
             data = {
                 "task_id": uuid.uuid4().hex,
@@ -171,7 +174,7 @@ def retrieve(self, *args, **kwargs):
                 resp_i = self._scheduler.servicecatalog.get_instance(
                     self.app.id, self.name).json()
                 self.status = resp_i.get('status', {}).\
-                    get('lastConditionState', '').lower()
+                    get('lastConditionState')
                 update_flag = True
             except KubeException as e:
                 logger.info("retrieve instance info error: {}".format(e))
@@ -181,7 +184,7 @@ def retrieve(self, *args, **kwargs):
                 resp_b = self._scheduler.servicecatalog.get_binding(
                     self.app.id, self.name).json()
                 self.binding = resp_b.get('status', {}).\
-                    get('lastConditionState', '').lower()
+                    get('lastConditionState')
                 self.options = resp_b.get('spec', {}).get('parameters', {})
                 update_flag = True
                 secret_name = resp_b.get('spec', {}).get('secretName')
@@ -213,12 +216,8 @@ def detach_resource(self, *args, **kwargs):
                 logger.info("delete binding info error: {}".format(e))
             self.binding = None
 
-        if (self.status != "Ready") or (self.binding is None):
-            try:
-                self._scheduler.servicecatalog.delete_instance(
-                    self.app.id, self.name)
-            except KubeException as e:
-                logger.info("retrieve instance info error: {}".format(e))
+        if (self.status != "Ready") or (not self.binding):
+            self.delete()
 
 
 @task
@@ -234,7 +233,7 @@ def retrieve_task(data):
         if _:
             return True
         else:
-            t = time.time() - resource.created.timestamps()
+            t = time.time() - resource.created.timestamp()
             if t < 3600:
                 apply_async(retrieve_task, delay=30000, args=(data, ))
             elif t < 3600 * 12:

rootfs/api/serializers.py

@@ -659,6 +659,7 @@ class ResourceSerializer(serializers.ModelSerializer):
     owner = serializers.ReadOnlyField(source='owner.username')
     name = serializers.CharField(max_length=63, required=True)
     plan = serializers.CharField(max_length=128, required=True)
+    data = JSONFieldSerializer(required=False, binary=True)
     options = JSONFieldSerializer(required=False, binary=True)
 
     class Meta:
@@ -670,8 +671,7 @@ def update(self, instance, validated_data):
         if instance.plan.split(':')[0] != validated_data.get('plan', '').split(':')[0]:  # noqa
             raise DryccException("the resource cann't changed")
         instance.plan = validated_data.get('plan')
-        if validated_data.get('options'):
-            instance.options = validated_data.get('options')
+        instance.options = validated_data.get('options')
         instance.attach_update()
         instance.save()
         return instance

rootfs/api/settings/production.py

@@ -392,10 +392,10 @@
     'default': {
         'ENGINE': 'django.db.backends.postgresql',
         'NAME': os.environ.get(
-            'DRYCC_DATABASE_NAME', os.environ.get('DRYCC_DATABASE_USER', 'drycc')),
-        'USER': os.environ.get('DRYCC_DATABASE_USER', ''),
-        'PASSWORD': os.environ.get('DRYCC_DATABASE_PASSWORD', ''),
-        'HOST': os.environ.get('DRYCC_DATABASE_SERVICE_HOST', ''),
+            'DRYCC_DATABASE_NAME', os.environ.get('DRYCC_DATABASE_USER', 'drycc_controller')),
+        'USER': os.environ.get('DRYCC_DATABASE_USER', 'drycc_controller'),
+        'PASSWORD': os.environ.get('DRYCC_DATABASE_PASSWORD', '123456'),
+        'HOST': os.environ.get('DRYCC_DATABASE_SERVICE_HOST', '192.168.6.50'),
         'PORT': os.environ.get('DRYCC_DATABASE_SERVICE_PORT', 5432),
         # https://docs.djangoproject.com/en/1.11/ref/databases/#persistent-connections
         'CONN_MAX_AGE': 600,

rootfs/api/views.py

@@ -792,10 +792,14 @@ def binding(self, request, *args, **kwargs):
         bind_action = self.request.data.get('bind_action', '').lower()
         if bind_action == 'bind':
             resource.bind()
-            return HttpResponse(status=status.HTTP_202_ACCEPTED)
+            serializer = self.get_serializer(resource, many=False)
+            logger.info("resoruce bind response data: {}".format(serializer))
+            return Response(serializer.data)
         elif bind_action == 'unbind':
             resource.unbind()
-            return HttpResponse(status=status.HTTP_202_ACCEPTED)
+            serializer = self.get_serializer(resource, many=False)
+            logger.info("resoruce unbind response data: {}".format(serializer))
+            return Response(serializer.data)
         else:
             return Http404("unknown action")
 

rootfs/scheduler/__init__.py

@@ -221,6 +221,11 @@ def http_patch(self, path, data=None, **kwargs):
         """
         try:
             url = urljoin(self.url, path)
+            # accepted media types include:
+            # application/json-patch+json,
+            # application/merge-patch+json,
+            # application/apply-patch+yaml
+            # self.session.headers["Content-Type"] = "application/json-patch+json"
             response = self.session.patch(url, data=data, **kwargs)
         except requests.exceptions.ConnectionError as err:
             # reraise as KubeException, but log stacktrace.

rootfs/scheduler/resources/servicecatalog.py

@@ -1,6 +1,9 @@
+import logging
 from scheduler.resources import Resource
 from scheduler.exceptions import KubeHTTPException
 
+logger = logging.getLogger(__name__)
+
 
 class ServiceCatalog(Resource):
     api_version = 'servicecatalog.k8s.io/v1beta1'
@@ -22,14 +25,29 @@ def service_instance_manifest(namespace, name, version=None, **kwargs):
             "spec": {
                 "clusterServiceClassExternalName": kwargs.get('instance_class'),
                 "clusterServicePlanExternalName": kwargs.get('instance_plan'),
-                #{"param - 1": "value - 1","param - 2": "value - 2"}  # noqa
-                "parameters": kwargs.get('parameters'),
             }
         }
         if version:
             data["metadata"]["resourceVersion"] = version
+        if kwargs.get('parameters'):
+            data["spec"]["parameters"] = kwargs.get('parameters')
+        if kwargs.get('external_id'):
+            data["spec"]["externalID"] = kwargs.get('external_id')
         return data
 
+    # @staticmethod
+    # def service_instance_patch_manifest(namespace, name, version=None, **kwargs):
+    #     data = {
+    #         "spec": {
+    #             "clusterServicePlanExternalName": kwargs.get('instance_plan'),
+    #             "parameters": kwargs.get('parameters'),
+    #         }
+    #     }
+    #     if version:
+    #         data["metadata"] = {}
+    #         data["metadata"]["resourceVersion"] = version
+    #     return data
+
     def get_instance(self, namespace, name=None):
         """
         Fetch a single serviceinstance or a list of serviceinstances
@@ -52,6 +70,7 @@ def create_instance(self, namespace, name, **kwargs):
         """
         url = self.api('/namespaces/{}/serviceinstances', namespace)
         data = self.service_instance_manifest(namespace, name, **kwargs)
+        logging.info("create_instance_data:{}".format(data))
         response = self.http_post(url, json=data)
         if not response.status_code == 201:
             raise KubeHTTPException(
@@ -65,13 +84,24 @@ def put_instance(self, namespace, name, version, **kwargs):
         """
         url = self.api('/namespaces/{}/serviceinstances/{}', namespace, name)
         data = self.service_instance_manifest(namespace, name, version, **kwargs)
+        logger.info("put_instance data:{}".format(data))
         response = self.http_put(url, json=data)
         if not response.status_code == 200:
             raise KubeHTTPException(
                 response,
                 "update serviceinstances {}".format(namespace))
         return response
 
+    # def patch_instance(self, namespace, name, version, **kwargs):
+    #     data = self.service_instance_patch_manifest(
+    #         namespace, name, version, **kwargs)
+    #     url = self.api("/namespaces/{}/serviceinstances/{}", namespace, name)
+    #     logger.info("patch_instance data:{}".format(data))
+    #     response = self.http_patch(url, json=data)
+    #     if self.unhealthy(response.status_code):
+    #         raise KubeHTTPException(response, "patch serviceinstances {}".format(namespace))
+    #     return response
+
     def delete_instance(self, namespace, name):
         """
         Delete serviceinstances