feat(controller): add a multi-tenant proxy for prometheus api

Author: duanhongyi

charts/controller/templates/_helpers.tpl

@@ -112,15 +112,15 @@ env:
   valueFrom:
     fieldRef:
       fieldPath: metadata.namespace
-{{- if (.Values.prometheusUrl) }}
-- name: "DRYCC_PROMETHEUS_URL"
+{{- if (.Values.victoriametricsUrl) }}
+- name: "DRYCC_VICTORIAMETRICS_URL"
   valueFrom:
     secretKeyRef:
       name: controller-creds
-      key: prometheus-url
-{{- else if .Values.prometheus.enabled }}
-- name: "DRYCC_PROMETHEUS_URL"
-  value: "http://drycc-victoriametrics-vmselect.{{$.Release.Namespace}}.svc.{{$.Values.global.clusterDomain}}:8481/select/0/prometheus"
+      key: victoriametrics-url
+{{- else if .Values.victoriametrics.enabled }}
+- name: "DRYCC_VICTORIAMETRICS_URL"
+  value: "http://drycc-victoriametrics-vmselect.{{$.Release.Namespace}}.svc.{{$.Values.global.clusterDomain}}:8481"
 {{- end }}
 {{- if .Values.passport.enabled }}
 - name: "DRYCC_PASSPORT_URL"

charts/controller/templates/controller-secret-creds.yaml

@@ -14,8 +14,8 @@ data:
   {{- if (.Values.databaseReplicaUrl) }}
   database-replica-url: {{ .Values.databaseReplicaUrl | b64enc }}
   {{- end }}
-  {{- if (.Values.prometheusUrl) }}
-  prometheus-url: {{ .Values.prometheusUrl | b64enc }}
+  {{- if (.Values.victoriametricsUrl) }}
+  victoriametrics-url: {{ .Values.victoriametricsUrl | b64enc }}
   {{- end }}
   {{- if (.Values.passportUrl) }}
   passport-url: {{ .Values.passportUrl | b64enc }}

charts/controller/values.yaml

@@ -63,8 +63,8 @@ databaseReplicaUrl: ""
 passportUrl: ""
 passportKey: ""
 passportSecret: ""
-# prometheusUrl is will no longer use the built-in prometheus component
-prometheusUrl: ""
+# victoriametricsUrl is will no longer use the built-in victoriametrics component
+victoriametricsUrl: ""
 # Workflow-manager Configuration Options
 workflowManagerUrl: ""
 workflowManagerAccessKey: ""
@@ -169,7 +169,7 @@ registry:
 passport:
   enabled: true
 
-prometheus:
+victoriametrics:
   enabled: true
 
 global:

rootfs/api/authentication.py

@@ -45,8 +45,11 @@ def authenticate(self, request):
         if token_type is None or token is None:
             return None
         if token_type == 'bearer':  # drycc oauth access token
-            from api.backend import OauthCacheManager
-            return OauthCacheManager().get_user(token), token
+            try:
+                from api.backend import OauthCacheManager
+                return OauthCacheManager().get_user(token), token
+            except exceptions.AuthenticationFailed:
+                return None, None
         # drycc token
         user = cache.get(token, None)
         if not user:
@@ -62,11 +65,14 @@ def authenticate_credentials(self, key):
         if not token.owner.is_active:
             raise exceptions.AuthenticationFailed(gettext_lazy('User inactive or deleted.'))
         if token.expires():
-            from api.backend import OauthCacheManager
-            token.refresh_token()
-            user = OauthCacheManager().get_user(token.oauth['access_token'])
-            cache.set(key, user, timeout=token.oauth['expires_in'])
-            return user, token.key
+            try:
+                from api.backend import OauthCacheManager
+                user = OauthCacheManager().get_user(token.oauth['access_token'])
+                cache.set(key, user, timeout=token.oauth['expires_in'])
+                token.refresh_token()
+                return user, token.key
+            except exceptions.AuthenticationFailed:
+                return None, None
         return (token.owner, token.key)
 
     def authenticate_header(self, request):

rootfs/api/management/commands/measure_loadbalancers.py

@@ -48,7 +48,7 @@ def _measure_loadbalancers(self, app_map, timestamp):
         send_measurements.delay(loadbalancers)
 
     def handle(self, *args, **options):
-        if settings.WORKFLOW_MANAGER_URL and settings.DRYCC_PROMETHEUS_URL:
+        if settings.WORKFLOW_MANAGER_URL and settings.DRYCC_VICTORIAMETRICS_URL:
             timestamp = int(time.time())
             task_id = uuid.uuid4().hex
             logger.info(f"pushing {task_id} limits to workflow_manager when {timezone.now()}")

rootfs/api/monitor.py

@@ -58,7 +58,7 @@ async def query_prom(url, params) -> list[tuple[dict[str, str], int]]:
 async def last_metrics(namespace) -> AsyncGenerator[Iterator, str]:
     if not settings.DRYCC_METRICS_CONFIG:
         return
-    url = urljoin(settings.DRYCC_PROMETHEUS_URL, "/api/v1/query")
+    url = urljoin(settings.DRYCC_VICTORIAMETRICS_URL, "/select/0/prometheus/api/v1/query")
     promql = query_last_metrics_promql_tpl % (
       '|'.join(settings.DRYCC_METRICS_CONFIG.keys()),
       namespace,
@@ -76,29 +76,29 @@ async def last_metrics(namespace) -> AsyncGenerator[Iterator, str]:
 
 async def query_loadbalancer(namespaces: Iterator[str], start: int, stop: int
                              ) -> list[tuple[dict[str, str], int]]:
-    url = urljoin(settings.DRYCC_PROMETHEUS_URL, "/api/v1/query")
+    url = urljoin(settings.DRYCC_VICTORIAMETRICS_URL, "/select/0/prometheus/api/v1/query")
     promql = query_loadbalancer_promql_tpl % "|".join(namespaces)
     return await query_prom(url, {"query": promql, "start": start, "end": stop})
 
 
 async def query_network_receive_flow(namespaces: Iterator[str], start: int, stop: int
                                      ) -> list[tuple[dict[str, str], int]]:
-    url = urljoin(settings.DRYCC_PROMETHEUS_URL, "/api/v1/query")
+    url = urljoin(settings.DRYCC_VICTORIAMETRICS_URL, "/select/0/prometheus/api/v1/query")
     promql = query_network_receive_flow_promql_tpl % ("|".join(namespaces), f"{stop-start}s")
     return await query_prom(url, {"query": promql, "start": start, "end": stop})
 
 
 async def query_network_transmit_flow(namespaces: Iterator[str], start: int, stop: int
                                       ) -> list[tuple[dict[str, str], int]]:
-    url = urljoin(settings.DRYCC_PROMETHEUS_URL, "/api/v1/query")
+    url = urljoin(settings.DRYCC_VICTORIAMETRICS_URL, "/select/0/prometheus/api/v1/query")
     promql = query_network_transmit_flow_promql_tpl % ("|".join(namespaces), f"{stop-start}s")
     return await query_prom(url, {"query": promql, "start": start, "end": stop})
 
 
 async def query_cpu_usage(namespace: str, ptype: str, every: str,
                           start: int, stop: int, step: int,
                           ) -> list[tuple[dict[str, str], int]]:
-    url = urljoin(settings.DRYCC_PROMETHEUS_URL, "/api/v1/query_range")
+    url = urljoin(settings.DRYCC_VICTORIAMETRICS_URL, "/select/0/prometheus/api/v1/query_range")
     pod_prefix = "%s-%s" % (namespace, ptype)
     promql = query_cpu_usage_promql_tpl % (pod_prefix, namespace, every)
     return await query_prom(url, {"query": promql, "start": start, "end": stop, "step": step})
@@ -107,7 +107,7 @@ async def query_cpu_usage(namespace: str, ptype: str, every: str,
 async def query_memory_usage(namespace: str, ptype: str, every: str,
                              start: int, stop: int, step: int,
                              ) -> list[tuple[dict[str, str], int]]:
-    url = urljoin(settings.DRYCC_PROMETHEUS_URL, "/api/v1/query_range")
+    url = urljoin(settings.DRYCC_VICTORIAMETRICS_URL, "/select/0/prometheus/api/v1/query_range")
     pod_prefix = "%s-%s" % (namespace, ptype)
     promql = query_memory_usage_promql_tpl % (pod_prefix, namespace, every)
     return await query_prom(url, {"query": promql, "start": start, "end": stop, "step": step})
@@ -116,7 +116,7 @@ async def query_memory_usage(namespace: str, ptype: str, every: str,
 async def query_network_receive_usage(namespace: str, ptype: str, every: str,
                                       start: int, stop: int, step: int,
                                       ) -> list[tuple[dict[str, str], int]]:
-    url = urljoin(settings.DRYCC_PROMETHEUS_URL, "/api/v1/query_range")
+    url = urljoin(settings.DRYCC_VICTORIAMETRICS_URL, "/select/0/prometheus/api/v1/query_range")
     pod_prefix = "%s-%s" % (namespace, ptype)
     promql = query_network_receive_usage_promql_tpl % (pod_prefix, namespace, every)
     return await query_prom(url, {"query": promql, "start": start, "end": stop, "step": step})
@@ -125,7 +125,7 @@ async def query_network_receive_usage(namespace: str, ptype: str, every: str,
 async def query_network_transmit_usage(namespace: str, ptype: str, every: str,
                                        start: int, stop: int, step: int,
                                        ) -> list[tuple[dict[str, str], int]]:
-    url = urljoin(settings.DRYCC_PROMETHEUS_URL, "/api/v1/query_range")
+    url = urljoin(settings.DRYCC_VICTORIAMETRICS_URL, "/select/0/prometheus/api/v1/query_range")
     pod_prefix = "%s-%s" % (namespace, ptype)
     promql = query_network_transmit_usage_promql_tpl % (pod_prefix, namespace, every)
     return await query_prom(url, {"query": promql, "start": start, "end": stop, "step": step})

rootfs/api/settings/production.py

@@ -269,8 +269,8 @@
 
 K8S_API_VERIFY_TLS = os.environ.get('K8S_API_VERIFY_TLS', 'true').lower() == "true"
 
-# drycc prometheus url
-DRYCC_PROMETHEUS_URL = os.environ.get('DRYCC_PROMETHEUS_URL', '')
+# drycc victoriametrics url
+DRYCC_VICTORIAMETRICS_URL = os.environ.get('DRYCC_VICTORIAMETRICS_URL', '')
 
 # drycc metrics config file
 DRYCC_METRICS_CONFIG = {}

rootfs/api/urls.py

@@ -247,13 +247,11 @@
     re_path(
         r'^nodes/(?P<node>[a-zA-Z0-9-]+)/proxy/metrics(?:/(?P<metrics>[^/]+))?/?$',
         views.ProxyMetricsView.as_view()),
+    # prometheus
+    re_path(r'^prometheus/(?P<path>.+)/?$', views.PrometheusProxy.as_view()),
     # tokens
     re_path(r'^tokens/?$', views.TokenViewSet.as_view({'get': 'list'})),
     re_path(r"^tokens/(?P<pk>[-_\w]+)/?$", views.TokenViewSet.as_view({'delete': 'destroy'})),
-    # social login is placed at the end of the URL match
-    re_path(r'^login/(?P<backend>[^/]+){0}$'.format(extra), views.auth, name='begin'),
-    re_path(r'^complete/(?P<backend>[^/]+){0}$'.format(extra), views.complete, name='complete'),
-    re_path('', include('social_django.urls', namespace='social')),
 ]
 
 mutate_urlpatterns = [
@@ -263,8 +261,15 @@
     ),
 ]
 
+# social login is placed at the end of the URL match
+social_urlpatterns = [
+    re_path(r'^login/(?P<backend>[^/]+){0}$'.format(extra), views.auth, name='begin'),
+    re_path(r'^complete/(?P<backend>[^/]+){0}$'.format(extra), views.complete, name='complete'),
+    re_path('', include('social_django.urls', namespace='social')),
+]
+
 # If there is a mutating admission mutate configuration, use mutate url
 if settings.MUTATE_KEY:
     urlpatterns = mutate_urlpatterns
 else:
-    urlpatterns = app_urlpatterns
+    urlpatterns = app_urlpatterns + social_urlpatterns

rootfs/api/views.py

@@ -10,7 +10,9 @@
 import random
 import aiohttp
 import requests
+import warnings
 
+from urllib.parse import urljoin
 from asgiref.sync import async_to_sync
 from django.db.models import Q
 from django.core.cache import cache
@@ -38,7 +40,8 @@
 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 FileResponse, StreamingHttpResponse
+from django.http.response import FileResponse, JsonResponse, StreamingHttpResponse
+from channels.db import database_sync_to_async
 from social_django.utils import psa
 from social_django.views import _do_login
 from social_core.utils import setting_name
@@ -1217,6 +1220,8 @@ def wrap(app_id, ptype, every, start, stop, step):
     @method_decorator(cache_page(settings.DRYCC_METRICS_EXPIRY))
     @method_decorator(vary_on_headers("Authorization"))
     def status(self, request, **kwargs):
+        warnings.warn(
+            'this interface will be removed in the next version.', PendingDeprecationWarning)
         app_id = self._get_app().id
         data = serializers.MetricSerializer(data=self.request.query_params)
         if not data.is_valid():
@@ -1243,6 +1248,8 @@ def status(self, request, **kwargs):
     @method_decorator(cache_page(settings.DRYCC_METRICS_EXPIRY))
     @method_decorator(vary_on_headers("Authorization"))
     def metric(self, request, **kwargs):
+        warnings.warn(
+            'this interface will be removed in the next version.', PendingDeprecationWarning)
         app_id = self._get_app().id
         return StreamingHttpResponse(
             streaming_content=monitor.last_metrics(app_id)
@@ -1316,3 +1323,29 @@ async def stream_response():
                         yield sample
         content_type = f"text/plain; version={__version__}"
         return StreamingHttpResponse(stream_response(), content_type=content_type)
+
+
+@method_decorator(csrf_exempt, name='dispatch')
+class PrometheusProxy(View):
+    timeout = aiohttp.ClientTimeout(total=30, connect=10, sock_read=15)
+    authentication = authentication.DryccAuthentication()
+
+    async def proxy(self, request, path):
+        auth = await database_sync_to_async(self.authentication.authenticate)(request)
+        if not auth or len(auth) != 2 or not isinstance(auth[0], User):
+            return JsonResponse({'error': 'access denied'}, status=403)
+        if auth[0].is_superuser or auth[0].is_staff:
+            path = f"/select/0/prometheus/{path}"
+        else:
+            path = f"/select/{auth[0].id}/prometheus/{path}"
+        url = urljoin(settings.DRYCC_VICTORIAMETRICS_URL, path)
+        params = dict(request.GET) if request.method == "GET" else dict(request.POST)
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.get(url, params=params, timeout=self.timeout) as response:
+                    data, status = await response.json(), response.status
+        except aiohttp.ClientError as e:
+            data, status = {'error': f'victoriametrics connection failed: {str(e)}'}, 502
+        return JsonResponse(data, status=status)
+
+    get = post = proxy

rootfs/requirements.txt

@@ -1,7 +1,7 @@
 # Drycc controller requirements
 inotify==0.2.10
 backoff==2.2.1
-django==4.2.21
+django==4.2.22
 channels==4.2.2
 aiohttp==3.11.16
 django-cors-headers==4.7.0