chore(volumes): add add filer delete

Author: duanhongyi

rootfs/api/filer.py

@@ -0,0 +1,81 @@
+import uuid
+import requests
+from django.conf import settings
+from django.core.cache import cache
+from requests.auth import HTTPBasicAuth
+
+from .utils import random_string, get_session, CacheLock
+
+
+class FilerClient(object):
+
+    def __init__(self, app_id, volume, scheduler):
+        self.bind = ":9000"
+        self.path = "/data"
+        self.app_id = app_id
+        self.volume = volume
+        self.scheduler = scheduler
+
+    @property
+    def server(self):
+        lock_key = f"filer:lock:{self.app_id}:{self.volume.name}"
+        lock = CacheLock(lock_key)
+        try:
+            lock.acquire()
+            _server = cache.get(self.cache_key, None)
+            if not _server or not self.health(_server):
+                _server = self.get_server()
+                cache.set(self.cache_key, _server, timeout=settings.DRYCC_FILER_DURATION)
+        finally:
+            lock.release()
+        return _server
+
+    @property
+    def cache_key(self):
+        return f"filer:{self.app_id}:{self.volume.name}"
+
+    def get_server(self):
+        pod_name = f"drycc-filer-{uuid.uuid4().hex}"
+        k8s_volume = {"name": self.volume.name}
+        if self.volume.type == "csi":
+            k8s_volume.update({"persistentVolumeClaim": {"claimName": self.volume.name}})
+        else:
+            k8s_volume.update(self.volume.parameters)
+        username, password = random_string(32), random_string(32)
+        self.scheduler.pod.create(self.app_id, pod_name, settings.DRYCC_FILER_IMAGE, **{
+            "args": [
+                "filer",
+                "--bind", self.bind, "--path", self.path,
+                "--duration", f"{settings.DRYCC_FILER_DURATION}",
+                "--waittime", f"{settings.DRYCC_FILER_WAITTIME}",
+                "--username", f"{username}", "--password", f"{password}",
+            ],
+            "labels": {"app": self.app_id, "pod": pod_name, "volume": self.volume.name},
+            "app_type": "filer", "replicas": 1, "deploy_timeout": 120, "volumes": [k8s_volume],
+            "volume_mounts": [{"mountPath": self.path, "name": self.volume.name}],
+            "restart_policy": "Never", "image_pull_policy": "Always",
+        })
+        address = self.scheduler.pod.get(self.app_id, pod_name).json()["status"]["podIP"]
+        return {"address": address, "username": username, "password": password}
+
+    def health(self, server):
+        try:
+            return self.request(
+                "OPTIONS", server, timeout=2).headers.get('server') == 'drycc-filer'
+        except requests.exceptions.Timeout:
+            return False
+
+    def request(self, method, server, path="/", **kwargs):
+        cache.touch(self.cache_key, timeout=settings.DRYCC_FILER_DURATION)
+        url = f"http://{server["address"]}:{self.bind.split(":")[1]}/{path}"
+        kwargs["auth"] = HTTPBasicAuth(server["username"], server["password"])
+        return get_session().request(method, url, **kwargs)
+
+    def get(self, path, **kwargs):
+        return self.request("GET", self.server, path, **kwargs)
+
+    def post(self, path, **kwargs):
+        return self.request("POST", self.server, path, **kwargs)
+
+    def delete(self, path, **kwargs):
+        return self.request("DELETE", self.server, path, **kwargs)

rootfs/api/monitor.py

@@ -1,4 +1,5 @@
 import time
+import aiohttp
 import requests
 from typing import Iterator, Dict
 from contextlib import closing
@@ -114,29 +115,27 @@
 """
 
 
-def last_metrics(namespace):
+async def last_metrics(namespace):
     if not settings.DRYCC_METRICS_CONFIG:
-        return ()
+        return
     promql = query_last_metrics_promql_tpl % (
-        '|'.join(settings.DRYCC_METRICS_CONFIG.keys()),
-        namespace,
-        '1m'
-    )
+        '|'.join(settings.DRYCC_METRICS_CONFIG.keys()), namespace, '1m')
+    url = urljoin(settings.DRYCC_PROMETHEUS_URL, "/api/v1/query")
     params = {"query": promql, "start": int(time.time() - 60)}
-    response = requests.get(
-        urljoin(settings.DRYCC_PROMETHEUS_URL, "/api/v1/query"),
-        params=params
-    )
-    if response.status_code != 200:
-        return ()
-    return ('%s{%s} %s\n' % (
-        item['metric']['__name__'],
-        ','.join([
-            f'{key}="{value}"' for key, value in item['metric'].items()
-            if key in settings.DRYCC_METRICS_CONFIG[item['metric']['__name__']]
-          ]),
-        item['value'][1],
-    ) for item in response.json()['data']['result'])
+    async with aiohttp.ClientSession() as session:
+        async with session.get(url, params=params) as response:
+            if response.status != 200:
+                return
+            items = await response.json()
+            for item in items['data']['result']:
+                yield '%s{%s} %s\n' % (
+                    item['metric']['__name__'],
+                    ','.join([
+                        f'{key}="{value}"' for key, value in item['metric'].items()
+                        if key in settings.DRYCC_METRICS_CONFIG[item['metric']['__name__']]
+                    ]),
+                    item['value'][1]
+                )
 
 
 def query_loadbalancer(namespaces: Iterator[str],

rootfs/api/urls.py

@@ -110,10 +110,15 @@
     re_path(
         r"^apps/(?P<id>{})/volumes/(?P<name>[-_\w]+)/path/?$".format(settings.APP_URL_REGEX),
         views.AppVolumesViewSet.as_view({'patch': 'path'})),
+    # application filer
     re_path(
-        r"^apps/(?P<id>{})/volumes/(?P<name>[-_\w]+)/files(/?(?P<path>[\S]+)/?)?$".format(
+        r"^apps/(?P<id>{})/volumes/(?P<name>[-_\w]+)/client/?$".format(
             settings.APP_URL_REGEX),
-        views.AppVolumesViewSet.as_view({'get': 'client', 'post': 'client'})),
+        views.AppFilerClientViewSet.as_view({'get': 'list', 'post': 'create'})),
+    re_path(
+        r"^apps/(?P<id>{})/volumes/(?P<name>[-_\w]+)/client/(?P<path>[\S]+)$".format(
+            settings.APP_URL_REGEX),
+        views.AppFilerClientViewSet.as_view({'get': 'retrieve', 'delete': 'destroy'})),
     # application resources
     re_path(r"^resources/services/?$", views.AppResourcesViewSet.as_view({'get': 'services'})),
     re_path(

rootfs/api/utils.py

@@ -3,6 +3,8 @@
 """
 import os
 import re
+import uuid
+import time
 import base64
 import string
 import concurrent
@@ -16,13 +18,12 @@
 import jsonschema
 from copy import deepcopy
 from django.db import models
-from django.conf import settings
 from django.core.cache import cache
-from requests.auth import HTTPBasicAuth
+from asgiref.sync import sync_to_async
 from requests_toolbelt import user_agent
 from api import __version__ as drycc_version
 from rest_framework.exceptions import ValidationError
-from scheduler import KubeException
+
 
 logger = logging.getLogger(__name__)
 
@@ -200,76 +201,47 @@ def validate_json(value, schema, raise_exception=ValidationError):
     return value
 
 
-class VolumeClient(object):
+class CacheLock(object):
 
-    def __init__(self, app_id, volume, scheduler):
-        self.bind = ":9000"
-        self.path = "/data"
-        self.app_id = app_id
-        self.volume = volume
-        self.scheduler = scheduler
+    def __init__(self, key):
+        self.key = key
+        self.value = uuid.uuid4().hex
 
-    @property
-    def server(self):
-        _server = cache.get(self.cache_key, None)
-        if not _server or not self.health(_server):
-            self.cleanup()
-            k8s_volume = {"name": self.volume.name}
-            if self.volume.type == "csi":
-                k8s_volume.update({"persistentVolumeClaim": {"claimName": self.volume.name}})
-            else:
-                k8s_volume.update(self.volume.parameters)
-            username, password = random_string(32), random_string(32)
-            self.scheduler.pod.create(self.app_id, self.pod_name, settings.DRYCC_FILER_IMAGE, **{
-                "args": [
-                    "filer",
-                    "--bind", self.bind, "--path", self.path,
-                    "--duration", f"{settings.DRYCC_FILER_DURATION}",
-                    "--waittime", f"{settings.DRYCC_FILER_WAITTIME}",
-                    "--username", f"{username}", "--password", f"{password}",
-                ],
-                "app_type": "filer",
-                "replicas": 1, "deploy_timeout": 120, "volumes": [k8s_volume],
-                "volume_mounts": [{"mountPath": self.path, "name": self.volume.name}],
-                "image_pull_policy": "Always",
-            })
-            address = self.scheduler.pod.get(self.app_id, self.pod_name).json()["status"]["podIP"]
-            _server = {"address": address, "username": username, "password": password}
-            cache.set(self.cache_key, _server, timeout=settings.DRYCC_FILER_DURATION)
-        return _server
-
-    @property
-    def pod_name(self):
-        return f"drycc-filer-{self.volume.name}"
-
-    @property
-    def cache_key(self):
-        return f"filer:{self.pod_name}"
-
-    def health(self, server):
-        try:
-            return self.request(
-                "OPTIONS", server, timeout=2).headers.get('server') == 'drycc-filer'
-        except requests.exceptions.Timeout:
-            return False
+    def acquire(self, blocking=True, timeout=120):
+        value = None
+        for _ in range(timeout):
+            value = cache.get_or_set(self.key, self.value, timeout)
+            if blocking or value == self.value:
+                break
+            time.sleep(1)
+        return value == self.value
 
-    def cleanup(self):
-        try:
-            self.scheduler.pod.delete(self.app_id, self.pod_name)
-        except KubeException:
-            pass
+    def release(self):
+        value = cache.get(self.key, None)
+        if value == self.value:
+            return
+        cache.delete(self.key)
+
+
+class SyncIterToAsyncIter(object):
+
+    def __init__(self, iter_content):
+        self.iter_content = iter_content
+
+    def __aiter__(self):
+        return self
 
-    def request(self, method, server, path="/", **kwargs):
-        cache.touch(self.cache_key, timeout=settings.DRYCC_FILER_DURATION)
-        url = f"http://{server["address"]}:{self.bind.split(":")[1]}/{path}"
-        kwargs["auth"] = HTTPBasicAuth(server["username"], server["password"])
-        return get_session().request(method, url, **kwargs)
+    async def __anext__(self):
+        @sync_to_async
+        def async_next():
+            try:
+                return next(self.iter_content)
+            except StopIteration:
+                raise StopAsyncIteration
+        return await async_next()
 
-    def get(self, path, **kwargs):
-        return self.request("GET", self.server, path, **kwargs)
 
-    def post(self, path, **kwargs):
-        return self.request("POST", self.server, path, **kwargs)
+iter_to_aiter = SyncIterToAsyncIter
 
 
 if __name__ == "__main__":

rootfs/api/views.py

@@ -23,7 +23,6 @@
 from rest_framework.permissions import IsAuthenticated, AllowAny
 from rest_framework.response import Response
 from rest_framework.viewsets import GenericViewSet
-from rest_framework.decorators import parser_classes
 from rest_framework.parsers import MultiPartParser
 
 from api import monitor, models, permissions, serializers, viewsets, authentication
@@ -33,11 +32,11 @@
 from django.views.decorators.cache import never_cache
 from django.contrib.auth import REDIRECT_FIELD_NAME
 from django.views.decorators.csrf import csrf_exempt
-from django.http.response import StreamingHttpResponse
+from django.http.response import FileResponse, StreamingHttpResponse
 from social_django.utils import psa
 from social_django.views import _do_login
 from social_core.utils import setting_name
-from api import admissions, utils
+from api import admissions, utils, filer
 from api.backend import OauthCacheManager
 from api.apps_extra.social_core.actions import do_auth, do_complete
 
@@ -827,21 +826,6 @@ def get_object(self):
                                  app__id=self.kwargs['id'],
                                  name=self.kwargs['name'])
 
-    @method_decorator(parser_classes([MultiPartParser]))
-    def client(self, request, **kwargs):
-        path = self.kwargs.get('path', '')
-        volume = self.get_object()
-        client = utils.VolumeClient(volume.app.id, volume, volume.app.scheduler())
-        if request.method == "GET":
-            response = client.get(path, stream=True)
-            return StreamingHttpResponse(
-                status=response.status_code, streaming_content=response.iter_content(),
-                content_type=response.headers.get('Content-Type', 'application/octet-stream'))
-        elif request.method == "POST":
-            client.post(path, files=request.FILES)
-            return HttpResponse(status=status.HTTP_204_NO_CONTENT)
-        return HttpResponse(status=status.HTTP_405_METHOD_NOT_ALLOWED)
-
     def expand(self, request, **kwargs):
         volume = self.get_object()
         volume.expand(request.data['size'])
@@ -877,6 +861,46 @@ def path(self, request, *args, **kwargs):
         return Response(serializer.data)
 
 
+class AppFilerClientViewSet(BaseDryccViewSet):
+    """RESTful views for volumes apps with collaborators."""
+    model = models.volume.Volume
+    parser_classes = [MultiPartParser]
+
+    def get_client(self):
+        volume = get_object_or_404(
+            models.volume.Volume, app__id=self.kwargs['id'], name=self.kwargs['name'])
+        return filer.FilerClient(volume.app.id, volume, volume.app.scheduler())
+
+    def list(self, request, **kwargs):
+        path = request.query_params.get('path', '')
+        client = self.get_client()
+        results = client.get(path, params={"action": "list"}).json()
+        # fake out pagination for now
+        pagination = {'results': results, 'count': len(results)}
+        return Response(data=pagination)
+
+    def retrieve(self, request, **kwargs):
+        path = self.kwargs.get('path', '')
+        client = self.get_client()
+        response = client.get(path, stream=True, params={"action": "get"})
+        return FileResponse(
+            status=response.status_code,
+            streaming_content=utils.iter_to_aiter(response.iter_content()),
+        )
+
+    def create(self, request, **kwargs):
+        path = request.data.get('path', '')
+        client = self.get_client()
+        response = client.post(path, files=request.FILES)
+        return Response(data=response.content, status=response.status_code)
+
+    def destroy(self, request, **kwargs):
+        path = self.kwargs.get('path', '')
+        client = self.get_client()
+        response = client.delete(path)
+        return Response(data=response.content, status=response.status_code)
+
+
 class AppResourcesViewSet(AppResourceViewSet):
     """RESTful views for resources apps with collaborators."""
     model = models.resource.Resource
@@ -1193,8 +1217,6 @@ def status(self, request, **kwargs):
     @method_decorator(vary_on_headers("Authorization"))
     def metric(self, request, **kwargs):
         app_id = self._get_app().id
-        streaming_content = monitor.last_metrics(app_id)
         return StreamingHttpResponse(
-            content_type='application/json',
-            streaming_content=streaming_content
+            streaming_content=monitor.last_metrics(app_id)
         )