feat(oauth): use oauth to unify service-to-service authentication.

Author: duanhongyi

.gitignore

@@ -2,3 +2,5 @@
 *.swp
 *.swo
 .DS_Store
+.sisyphus
+__pycache__

charts/grafana/templates/grafana-cronjob-daily.yaml

@@ -0,0 +1,45 @@
+apiVersion: batch/v1
+kind: CronJob
+metadata:
+  name: grafana-oauth2-token-refresher
+  labels:
+    app: grafana
+    component: token-refresher
+    heritage: drycc
+spec:
+  # Run daily at 2 AM (avoid peak business hours)
+  schedule: "0 2 * * *"
+  timeZone: "Asia/Shanghai"
+  
+  # Prevent concurrent executions (ensure a single instance)
+  concurrencyPolicy: Forbid
+  
+  # Keep the latest 3 successful and failed job records
+  successfulJobsHistoryLimit: 3
+  failedJobsHistoryLimit: 3
+  
+  jobTemplate:
+    spec:
+      # Retry the job up to 3 times
+      backoffLimit: 3
+      
+      template:
+        metadata:
+          labels:
+            app: grafana
+            component: token-refresher
+        spec:
+          restartPolicy: OnFailure
+          containers:
+          - name: token-refresher
+            image: {{ .Values.imageRegistry }}/{{ .Values.imageOrg }}/grafana:{{ .Values.imageTag }}
+            imagePullPolicy: {{ .Values.imagePullPolicy }}
+            command: ["/usr/bin/env", "python3", "/usr/share/grafana/oauth2/token.py"]
+            {{- include "grafana.envs" . | indent 12 }}
+            resources:
+              requests:
+                memory: "64Mi"
+                cpu: "100m"
+              limits:
+                memory: "128Mi"
+                cpu: "200m"

rootfs/usr/share/grafana/oauth2/datasources/prometheus.json

@@ -2,16 +2,12 @@
     "name": "Prometheus on Drycc",
     "type": "prometheus",
     "uid": "prometheus_on_drycc",
-    "url": "${controller_url}/v2/prometheus/${workspace}",
+    "url": "http://localhost:4000/proxy/prometheus/${workspace}",
     "access": "proxy",
     "isDefault": true,
     "basicAuth": false,
     "jsonData": {
-        "httpHeaderName1": "Authorization",
         "httpMethod": "POST",
         "timeInterval": "${time_interval}"
-    },
-    "secureJsonData": {
-        "httpHeaderValue1": "Token ${token}"
     }
-}
+}

rootfs/usr/share/grafana/oauth2/datasources/quickwit.json

@@ -2,15 +2,11 @@
     "name": "Application Logs",
     "type": "quickwit-quickwit-datasource",
     "uid": "application_logs",
-    "url": "${controller_url}/v2/quickwit/${workspace}",
+    "url": "http://localhost:4000/proxy/quickwit/${workspace}",
     "access": "proxy",
     "basicAuth": false,
     "jsonData": {
-        "httpHeaderName1": "Authorization",
         "index": "logs-*",
         "logMessageField": "log"
-    },
-    "secureJsonData": {
-        "httpHeaderValue1": "Token ${token}"
     }
 }

rootfs/usr/share/grafana/oauth2/hook/grafana.py

@@ -1,17 +1,16 @@
 import logging
-import os
 import time
 import json
 import httpx
+from pathlib import Path
 from string import Template
 from psycopg import AsyncConnection
+from ..settings import settings
 
 logger = logging.getLogger(__name__)
 
 DEFAULT_HEADERS = {"Content-Type": "application/json"}
-DRYCC_CONTROLLER_URL = os.environ.get('DRYCC_CONTROLLER_URL')
-DRYCC_GRAFANA_REFRESH = os.environ.get('DRYCC_GRAFANA_REFRESH', '60s')
-DRYCC_GRAFANA_DASHBOARD = os.path.join(os.path.dirname(os.path.abspath(__file__)), "../")
+DRYCC_GRAFANA_DASHBOARD = Path(__file__).resolve().parent.parent
 
 # Drycc Workspace role to Grafana role mapping
 DRYCC_WORKSPACE_ROLE_MAPPING = {"admin": "Editor", "member": "Editor", "viewer": "Viewer"}
@@ -107,15 +106,15 @@ async def sync_alerting(context: dict, token: dict, userinfo: dict):
     The alerts field only controls notification channels (handled in sync_default).
     """
     workspace_orgs = context.get("workspace_orgs", {})
-    alerting_path = os.path.join(os.path.dirname(__file__), "..", "alerting")
+    alerting_path = Path(__file__).resolve().parent.parent / "alerting"
 
     for ws_name, ws_info in workspace_orgs.items():
         org_id = ws_info["org_id"]
         ctx = {**context, "org_id": org_id}
 
         async with httpx.AsyncClient() as client:
-            for filename in os.listdir(alerting_path):
-                with open(os.path.join(alerting_path, filename)) as f:
+            for filepath in alerting_path.glob("*.json"):
+                with filepath.open() as f:
                     rule = json.load(f)
                 # Use PUT for idempotent upsert (POST would create duplicates)
                 resp = await client.put(
@@ -135,23 +134,20 @@ async def sync_alerting(context: dict, token: dict, userinfo: dict):
 async def sync_datasources(context: dict, token: dict, userinfo: dict):
     """Create datasources for each workspace org with workspace-specific URLs."""
     workspace_orgs = context.get("workspace_orgs", {})
-    datasources_path = os.path.join(os.path.dirname(__file__), "..", "datasources")
-    drycc_token = context.get("drycc_token")
+    datasources_path = Path(__file__).resolve().parent.parent / "datasources"
 
     for ws_name, ws_info in workspace_orgs.items():
         org_id = ws_info["org_id"]
         ctx = {**context, "org_id": org_id}
         headers = _api_headers(ctx, userinfo)
 
         async with httpx.AsyncClient() as client:
-            for filename in os.listdir(datasources_path):
-                with open(os.path.join(datasources_path, filename)) as f:
+            for filepath in datasources_path.glob("*.json"):
+                with filepath.open() as f:
                     template = Template(f.read())
                     datasource = json.loads(template.substitute(
-                        controller_url=DRYCC_CONTROLLER_URL,
-                        workspace=ws_name,
-                        time_interval=DRYCC_GRAFANA_REFRESH,
-                        token=drycc_token
+                        controller_url=settings.drycc_controller_url,
+                        time_interval=settings.drycc_grafana_refresh
                     ))
                     resp = await client.get(
                         _api_url(f"/api/datasources/name/{datasource['name']}"), headers=headers)
@@ -174,17 +170,17 @@ async def sync_datasources(context: dict, token: dict, userinfo: dict):
 async def sync_dashboards(context: dict, token: dict, userinfo: dict):
     """Create dashboards for each workspace org."""
     workspace_orgs = context.get("workspace_orgs", {})
-    dashboards_path = os.path.join(os.path.dirname(__file__), "..", "dashboards")
+    dashboards_path = Path(__file__).resolve().parent.parent / "dashboards"
 
     for ws_name, ws_info in workspace_orgs.items():
         org_id = ws_info["org_id"]
         ctx = {**context, "org_id": org_id}
 
         async with httpx.AsyncClient() as client:
-            for filename in os.listdir(dashboards_path):
-                with open(os.path.join(dashboards_path, filename)) as f:
+            for filepath in dashboards_path.glob("*.json"):
+                with filepath.open() as f:
                     dashboard = json.load(f)
-                    dashboard.update({"id": None, "refresh": DRYCC_GRAFANA_REFRESH})
+                    dashboard.update({"id": None, "refresh": settings.drycc_grafana_refresh})
                     await client.post(
                         _api_url("/api/dashboards/db"),
                         headers=_api_headers(ctx, userinfo),
@@ -202,12 +198,12 @@ async def sync_dashboards(context: dict, token: dict, userinfo: dict):
 def _api_url(url_path, is_admin=False):
     if is_admin:
         return "http://{}:{}@localhost:{}{}".format(
-            os.environ.get('GF_SECURITY_ADMIN_USER'),
-            os.environ.get('GF_SECURITY_ADMIN_PASSWORD'),
-            os.environ.get('GF_SERVER_HTTP_PORT', 3000),
-            url_path,
+            settings.gf_security_admin_user,
+            settings.gf_security_admin_password,
+            settings.gf_server_http_port,
+            url_path
         )
-    return "http://localhost:{}{}".format(os.environ.get('GF_SERVER_HTTP_PORT', 3000), url_path)
+    return "http://localhost:{}{}".format(settings.gf_server_http_port, url_path)
 
 
 def _api_headers(context: dict, userinfo):
@@ -233,7 +229,7 @@ async def _get_workspaces(drycc_token: str) -> list:
     headers = {"Authorization": f"Token {drycc_token}"}
     async with httpx.AsyncClient() as client:
         resp = await client.get(
-            f"{DRYCC_CONTROLLER_URL}/v2/workspaces", headers=headers)
+            f"{settings.drycc_controller_url}/v2/workspaces", headers=headers)
         resp.raise_for_status()
         return resp.json().get("results", [])
 
@@ -243,7 +239,7 @@ async def _get_workspace_members(workspace_name: str, drycc_token: str) -> list:
     headers = {"Authorization": f"Token {drycc_token}"}
     async with httpx.AsyncClient() as client:
         resp = await client.get(
-            f"{DRYCC_CONTROLLER_URL}/v2/workspaces/{workspace_name}/members",
+            f"{settings.drycc_controller_url}/v2/workspaces/{workspace_name}/members",
             headers=headers)
         resp.raise_for_status()
         return resp.json().get("results", [])
@@ -420,7 +416,7 @@ def _build_alertmanager_config(alert_addresses: str) -> str:
 
 async def _upsert_alert_configuration(org_id: int, config: str):
     """Insert or update alert configuration for an org using parameterized query."""
-    async with await AsyncConnection.connect(os.environ.get("GF_DATABASE_URL")) as conn:
+    async with await AsyncConnection.connect(settings.gf_database_url) as conn:
         async with conn.cursor() as cursor:
             await cursor.execute(
                 """

rootfs/usr/share/grafana/oauth2/main.py

@@ -1,13 +1,17 @@
 import random
 import string
 import argparse
+import asyncio
+import httpx
 from contextlib import asynccontextmanager
-from fastapi import FastAPI, Request
-from fastapi.responses import JSONResponse, RedirectResponse
+from fastapi import FastAPI, Request, HTTPException
+from fastapi.responses import JSONResponse, RedirectResponse, StreamingResponse
 from authlib.integrations.starlette_client import OAuth
 from starlette.middleware.sessions import SessionMiddleware
 
 from hook import startup_hooks, login_hooks, destroy_hooks
+from settings import settings
+from token import get_token
 
 
 def randstr(k=32):
@@ -25,11 +29,23 @@ def randstr(k=32):
 args = parser.parse_args()
 
 
+http_client = None
+
 @asynccontextmanager
 async def lifespan(app: FastAPI):
+    global http_client
+    http_client = httpx.AsyncClient(timeout=30.0)
+    
+    # Ensure token is initialized at startup
+    try:
+        await get_token()
+    except Exception as e:
+        raise RuntimeError(f"Failed to initialize token: {e}")
+    
     for startup_hook in startup_hooks:
         await startup_hook()
     yield
+    await http_client.aclose()
     for destroy_hook in destroy_hooks:
         await destroy_hook()
 
@@ -48,6 +64,8 @@ async def lifespan(app: FastAPI):
 )
 
 
+# ── OAuth2 routes (unchanged) ────────────────────────────────────
+
 @app.get("/oauth2/healthz")
 async def healthz():
     return {"status": "ok"}
@@ -90,6 +108,45 @@ async def oauth2_userinfo(request: Request):
     return JSONResponse(content=userinfo, headers=headers)
 
 
+# ── Proxy routes (zero-overhead version) ───────────────────────────
+
+async def _proxy_request(request: Request, url: str):
+    """Minimal proxy: get the token directly from token.py and use it."""
+    try:
+        token = await get_token()
+    except RuntimeError as e:
+        raise HTTPException(status_code=503, detail=str(e))
+
+    headers = dict(request.headers)
+    headers.pop("host", None)
+    headers["Authorization"] = f"Bearer {token}"
+
+    resp = await http_client.request(
+        method=request.method,
+        url=url,
+        headers=headers,
+        params=request.query_params,
+        content=await request.body(),
+    )
+    return StreamingResponse(
+        resp.aiter_bytes(),
+        status_code=resp.status_code,
+        headers=dict(resp.headers),
+    )
+
+
+@app.api_route("/proxy/prometheus/{workspace}/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])
+async def proxy_prometheus(workspace: str, path: str, request: Request):
+    url = f"{settings.drycc_controller_url}/v2/prometheus/{workspace}/{path}"
+    return await _proxy_request(request, url)
+
+
+@app.api_route("/proxy/quickwit/{workspace}/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])
+async def proxy_quickwit(workspace: str, path: str, request: Request):
+    url = f"{settings.drycc_controller_url}/v2/quickwit/{workspace}/{path}"
+    return await _proxy_request(request, url)
+
+
 if __name__ == "__main__":
     import uvicorn
     uvicorn.run(app, host=args.bind, port=int(args.port))

rootfs/usr/share/grafana/oauth2/requirements.txt

@@ -7,3 +7,4 @@ psycopg[binary]>=3.2.9
 python-jose>=3.3.0
 python-multipart>=0.0.5
 itsdangerous>=2.2.0
+pydantic_settings>=2.14.1

rootfs/usr/share/grafana/oauth2/settings.py

@@ -0,0 +1,32 @@
+from pydantic import Field, computed_field, RedisDsn, HttpUrl
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(
+        env_prefix="",
+        extra="ignore",
+        case_sensitive=True,
+    )
+
+    # Drycc configuration
+    drycc_valkey_url: RedisDsn = Field(default="redis://localhost:6379/0", validation_alias="DRYCC_VALKEY_URL")
+    drycc_passport_url: HttpUrl = Field(default="http://passport.drycc.cc", validation_alias="DRYCC_PASSPORT_URL")
+    drycc_passport_key: str = Field(default="", validation_alias="DRYCC_PASSPORT_KEY")
+    drycc_passport_secret: str = Field(default="", validation_alias="DRYCC_PASSPORT_SECRET")
+    drycc_controller_url: HttpUrl = Field(default="http://controller.drycc.cc", validation_alias="DRYCC_CONTROLLER_URL")
+    drycc_grafana_refresh: str = Field(default="60s", validation_alias="DRYCC_GRAFANA_REFRESH")
+
+    # Grafana configuration
+    gf_security_admin_user: str = Field(default="admin", validation_alias="GF_SECURITY_ADMIN_USER")
+    gf_security_admin_password: str = Field(default="admin", validation_alias="GF_SECURITY_ADMIN_PASSWORD")
+    gf_server_http_port: int = Field(default=3000, validation_alias="GF_SERVER_HTTP_PORT")
+    gf_database_url: str = Field(default="", validation_alias="GF_DATABASE_URL")
+
+    @computed_field
+    @property
+    def passport_token_url(self) -> str:
+        return f"{str(self.drycc_passport_url).rstrip('/')}/oauth/token/"
+
+
+settings = Settings()