admin
5f5d31f86d
masterplan-lock / check (push) Successful in 5s
CI (SIAX Cloud) / sonar (pull_request) Skipped
masterplan-lock / check (pull_request) Successful in 4s
CI (SIAX Cloud) / security (pull_request) Failing after 14s
Security Scan (SIAX Cloud) / security (pull_request) Failing after 17s
CI (SIAX Cloud) / integration (pull_request) Successful in 58s
CI (SIAX Cloud) / quality (pull_request) Canceled after 1m59s
465 lines
19 KiB
Python
465 lines
19 KiB
Python
"""
|
|
0pt analytics HTTP surface (BI-P068-compatible).
|
|
|
|
This is the missing read plane between API0's `/analytics/*` contract and the
|
|
`opt_analytics` ClickHouse database. It does not invent a second analytics
|
|
model:
|
|
|
|
* datasets/metrics are the distinct ids actually present in
|
|
`opt_analytics_event` / `opt_metric_daily` / `opt_dataset_projection`;
|
|
* query/estimate/explain are computed against the tenant-forced ClickHouse
|
|
rows, never against a local cache;
|
|
* dashboards/reports are honest empty lists until the 0pt BI metadata lane
|
|
writes them;
|
|
* every response is either real ClickHouse data or an explicit error.
|
|
|
|
IDENTITY. API0 verifies the caller and forwards the caller's own Keycloak
|
|
token plus `X-Tenant-Id`. This service verifies the same token against the
|
|
portal realm's JWKS and cross-checks `tenant_id` against `X-Tenant-Id`. A
|
|
service-to-service `X-Internal-Token` is also accepted for the ingestion
|
|
worker, but never for a request that reads tenant data without an explicit,
|
|
already-verified `X-Tenant-Id`.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import time
|
|
from functools import lru_cache
|
|
from typing import Any, Iterable, Literal
|
|
|
|
import httpx
|
|
import jwt
|
|
from fastapi import FastAPI, Header, HTTPException, Query, Request
|
|
from fastapi.responses import JSONResponse
|
|
from pydantic import BaseModel, Field
|
|
|
|
CLICKHOUSE_URL = os.getenv("CLICKHOUSE_URL", "").rstrip("/")
|
|
CLICKHOUSE_USER = os.getenv("CLICKHOUSE_USER", "default")
|
|
CLICKHOUSE_PASSWORD = os.getenv("CLICKHOUSE_PASSWORD", "")
|
|
CLICKHOUSE_DATABASE = os.getenv("CLICKHOUSE_DATABASE", "opt_analytics")
|
|
KEYCLOAK_ISSUER = os.getenv("KEYCLOAK_ISSUER", "").rstrip("/")
|
|
KEYCLOAK_AUDIENCE = os.getenv("KEYCLOAK_AUDIENCE", "account")
|
|
KEYCLOAK_AUTHORIZED_PARTY = os.getenv("KEYCLOAK_AUTHORIZED_PARTY", "siax-portal")
|
|
INTERNAL_TOKEN = os.getenv("INTERNAL_TOKEN", "")
|
|
ALLOWED_DIMENSIONS = {"event_type", "app_id", "capability_id", "risk_level", "source_service"}
|
|
|
|
app = FastAPI(title="0pt analytics", version="0.1")
|
|
|
|
|
|
class AnalyticsError(Exception):
|
|
def __init__(self, code: str, message: str, status: int = 400):
|
|
self.code = code
|
|
self.message = message
|
|
self.status = status
|
|
super().__init__(message)
|
|
|
|
|
|
@app.exception_handler(AnalyticsError)
|
|
async def analytics_error_handler(_request: Request, exc: AnalyticsError) -> JSONResponse:
|
|
return JSONResponse(status_code=exc.status, content={"error": {"code": exc.code, "message": exc.message}})
|
|
|
|
|
|
# ── identity ────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def _jwks_client() -> jwt.PyJWKClient:
|
|
if not KEYCLOAK_ISSUER:
|
|
raise AnalyticsError("ANALYTICS_NOT_CONFIGURED", "KEYCLOAK_ISSUER is not set", 503)
|
|
return jwt.PyJWKClient(f"{KEYCLOAK_ISSUER}/protocol/openid-connect/certs", cache_keys=True, lifespan=3600)
|
|
|
|
|
|
def _bearer(request: Request) -> str | None:
|
|
raw = request.headers.get("authorization", "")
|
|
if not raw.lower().startswith("bearer "):
|
|
return None
|
|
return raw[7:].strip() or None
|
|
|
|
|
|
def verified_identity(request: Request, x_tenant_id: str | None) -> str:
|
|
"""Return the tenant the request is allowed to read.
|
|
|
|
An `X-Internal-Token` may only be used by the ingestion worker and never
|
|
substitutes for a tenant claim. For user/API0 calls the JWT is the source
|
|
of truth; a mismatching X-Tenant-Id is a hard 403, not a preference.
|
|
"""
|
|
if INTERNAL_TOKEN and request.headers.get("x-internal-token") == INTERNAL_TOKEN:
|
|
if not x_tenant_id:
|
|
raise AnalyticsError("TENANT_REQUIRED", "X-Tenant-Id is required for internal calls", 400)
|
|
return x_tenant_id
|
|
|
|
token = _bearer(request)
|
|
if not token:
|
|
raise AnalyticsError("UNAUTHENTICATED", "A bearer token is required", 401)
|
|
try:
|
|
signing_key = _jwks_client().get_signing_key_from_jwt(token)
|
|
claims = jwt.decode(
|
|
token,
|
|
signing_key.key,
|
|
algorithms=["RS256", "RS384", "RS512"],
|
|
audience=KEYCLOAK_AUDIENCE,
|
|
issuer=KEYCLOAK_ISSUER,
|
|
options={"require": ["exp", "iss", "sub"]},
|
|
)
|
|
except jwt.PyJWTError as exc:
|
|
raise AnalyticsError("UNAUTHENTICATED", f"Invalid bearer token: {exc}", 401) from exc
|
|
|
|
if KEYCLOAK_AUTHORIZED_PARTY and claims.get("azp") not in (None, KEYCLOAK_AUTHORIZED_PARTY):
|
|
raise AnalyticsError("FORBIDDEN", "Token was issued for a different client", 403)
|
|
tenant = claims.get("tenant_id")
|
|
if not isinstance(tenant, str) or not tenant.strip():
|
|
raise AnalyticsError("NO_TENANT", "Token is not bound to a tenant", 403)
|
|
if x_tenant_id and x_tenant_id != tenant:
|
|
raise AnalyticsError("TENANT_MISMATCH", "X-Tenant-Id does not match the verified tenant", 403)
|
|
return tenant
|
|
|
|
|
|
# ── clickhouse ──────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _ch_client() -> httpx.Client:
|
|
if not CLICKHOUSE_URL:
|
|
raise AnalyticsError("ANALYTICS_NOT_CONFIGURED", "CLICKHOUSE_URL is not set", 503)
|
|
auth = (CLICKHOUSE_USER, CLICKHOUSE_PASSWORD) if CLICKHOUSE_PASSWORD else None
|
|
return httpx.Client(base_url=CLICKHOUSE_URL, auth=auth, timeout=20.0)
|
|
|
|
|
|
def ch_json(sql: str, params: dict[str, str] | None = None) -> dict[str, Any]:
|
|
params = params or {}
|
|
query_params = {
|
|
"database": CLICKHOUSE_DATABASE,
|
|
"output_format_json_quote_64bit_integers": "0",
|
|
**{f"param_{key}": value for key, value in params.items()},
|
|
}
|
|
try:
|
|
with _ch_client() as client:
|
|
response = client.post("/", params=query_params, content=sql)
|
|
except httpx.HTTPError as exc:
|
|
raise AnalyticsError("CLICKHOUSE_UNREACHABLE", str(exc), 503) from exc
|
|
if response.status_code != 200:
|
|
raise AnalyticsError("CLICKHOUSE_REFUSED", response.text[:500], 502)
|
|
try:
|
|
return response.json()
|
|
except ValueError as exc:
|
|
raise AnalyticsError("CLICKHOUSE_BAD_RESPONSE", "ClickHouse did not return JSON", 502) from exc
|
|
|
|
|
|
def _literal(value: str) -> str:
|
|
# ClickHouse query parameters are preferable; the only values interpolated
|
|
# here are identifiers from ALLOWED_DIMENSIONS, never caller text.
|
|
if value not in ALLOWED_DIMENSIONS:
|
|
raise AnalyticsError("INVALID_DIMENSION", f"Unsupported dimension: {value}")
|
|
return value
|
|
|
|
|
|
def rows(sql: str, params: dict[str, str] | None = None) -> list[dict[str, Any]]:
|
|
return ch_json(sql, params).get("data", [])
|
|
|
|
|
|
def scalar(sql: str, params: dict[str, str] | None = None) -> Any:
|
|
data = rows(sql, params)
|
|
if not data:
|
|
return None
|
|
return next(iter(data[0].values()))
|
|
|
|
|
|
def build_query(
|
|
*,
|
|
tenant: str,
|
|
dataset: str | None,
|
|
metrics: Iterable[str],
|
|
dimensions: Iterable[str],
|
|
filters: dict[str, str] | None,
|
|
from_time: str | None,
|
|
to_time: str | None,
|
|
limit: int,
|
|
) -> tuple[str, dict[str, str]]:
|
|
dims = [_literal(d) for d in dimensions]
|
|
metric_list = [m for m in metrics if m.strip()]
|
|
if not metric_list:
|
|
raise AnalyticsError("INVALID_REQUEST", "At least one metric is required")
|
|
where = ["tenant_id = {tenant:String}"]
|
|
params: dict[str, str] = {"tenant": tenant, "limit": str(max(1, min(limit, 1000)))}
|
|
if dataset:
|
|
where.append("dataset_id = {dataset:String}")
|
|
params["dataset"] = dataset
|
|
metric_params = []
|
|
for i, metric in enumerate(metric_list):
|
|
key = f"metric{i}"
|
|
metric_params.append(f"{{{key}:String}}")
|
|
params[key] = metric
|
|
where.append(f"metric IN ({', '.join(metric_params)})")
|
|
for i, (key, value) in enumerate((filters or {}).items()):
|
|
dim = _literal(key)
|
|
p = f"filter{i}"
|
|
where.append(f"dim_values[indexOf(dim_names, '{dim}')] = {{{p}:String}}")
|
|
params[p] = value
|
|
if from_time:
|
|
where.append("event_time >= {from:String}")
|
|
params["from"] = from_time
|
|
if to_time:
|
|
where.append("event_time <= {to:String}")
|
|
params["to"] = to_time
|
|
|
|
select_dims = []
|
|
for i, dim in enumerate(dims):
|
|
select_dims.append(f"dim_values[indexOf(dim_names, '{dim}')] AS d{i}")
|
|
select = ", ".join(select_dims + ["count() AS value"])
|
|
group = f" GROUP BY {', '.join(f'd{i}' for i in range(len(dims)))}" if dims else ""
|
|
sql = (
|
|
f"SELECT {select} FROM {CLICKHOUSE_DATABASE}.opt_analytics_event "
|
|
f"WHERE {' AND '.join(where)}{group} ORDER BY value DESC LIMIT {{limit:UInt32}} FORMAT JSON"
|
|
)
|
|
return sql, params
|
|
|
|
|
|
# ── models ─────────────────────────────────────────────────────────────────
|
|
|
|
|
|
class QueryRequest(BaseModel):
|
|
dataset: str | None = None
|
|
metrics: list[str] = Field(default_factory=list)
|
|
dimensions: list[str] = Field(default_factory=list)
|
|
filters: dict[str, str] | None = None
|
|
timeRange: dict[str, str] | None = None
|
|
limit: int = 100
|
|
|
|
|
|
# ── surface ────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@app.get("/health")
|
|
def health() -> dict[str, Any]:
|
|
try:
|
|
ch_json("SELECT 1")
|
|
return {"status": "ok", "providers": {"clickhouse": {"status": "ok"}}}
|
|
except AnalyticsError as exc:
|
|
return {"status": "error", "providers": {"clickhouse": {"status": "error", "detail": exc.message}}}
|
|
|
|
|
|
@app.get("/api/0pt/analytics/health")
|
|
def analytics_health() -> dict[str, Any]:
|
|
return health()
|
|
|
|
|
|
@app.get("/api/0pt/analytics/datasets")
|
|
def list_datasets(request: Request, x_tenant_id: str | None = Header(default=None)) -> list[dict[str, Any]]:
|
|
tenant = verified_identity(request, x_tenant_id)
|
|
rows_ = rows(
|
|
f"SELECT dataset_id, count() AS events FROM {CLICKHOUSE_DATABASE}.opt_analytics_event "
|
|
"WHERE tenant_id = {tenant:String} GROUP BY dataset_id ORDER BY dataset_id FORMAT JSON",
|
|
{"tenant": tenant},
|
|
)
|
|
if not rows_:
|
|
rows_ = rows(
|
|
f"SELECT dataset_id, count() AS rows FROM {CLICKHOUSE_DATABASE}.opt_dataset_projection "
|
|
"WHERE tenant_id = {tenant:String} GROUP BY dataset_id ORDER BY dataset_id FORMAT JSON",
|
|
{"tenant": tenant},
|
|
)
|
|
return [
|
|
{
|
|
"id": r["dataset_id"],
|
|
"name": r["dataset_id"],
|
|
"tier": "SERVING",
|
|
"owner": "0pt",
|
|
"source": "0pt analytics",
|
|
"tenant_id": tenant,
|
|
**({"events": r.get("events")} if "events" in r else {}),
|
|
**({"rows": r.get("rows")} if "rows" in r else {}),
|
|
}
|
|
for r in rows_
|
|
]
|
|
|
|
|
|
@app.get("/api/0pt/analytics/datasets/{dataset_id}")
|
|
def get_dataset(dataset_id: str, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
|
|
tenant = verified_identity(request, x_tenant_id)
|
|
found = [d for d in list_datasets(request, x_tenant_id) if d["id"] == dataset_id]
|
|
if not found:
|
|
raise AnalyticsError("NOT_FOUND", f"dataset {dataset_id} does not exist for this tenant", 404)
|
|
return found[0]
|
|
|
|
|
|
@app.get("/api/0pt/analytics/metrics")
|
|
def list_metrics(request: Request, x_tenant_id: str | None = Header(default=None)) -> list[dict[str, Any]]:
|
|
tenant = verified_identity(request, x_tenant_id)
|
|
rows_ = rows(
|
|
f"SELECT metric, count() AS events FROM {CLICKHOUSE_DATABASE}.opt_analytics_event "
|
|
"WHERE tenant_id = {tenant:String} GROUP BY metric ORDER BY metric FORMAT JSON",
|
|
{"tenant": tenant},
|
|
)
|
|
if not rows_:
|
|
rows_ = rows(
|
|
f"SELECT metric, sum(count_state) AS events FROM {CLICKHOUSE_DATABASE}.opt_metric_daily "
|
|
"WHERE tenant_id = {tenant:String} GROUP BY metric ORDER BY metric FORMAT JSON",
|
|
{"tenant": tenant},
|
|
)
|
|
dims = rows(
|
|
f"SELECT DISTINCT arrayJoin(dim_names) AS name FROM {CLICKHOUSE_DATABASE}.opt_analytics_event "
|
|
"WHERE tenant_id = {tenant:String} FORMAT JSON",
|
|
{"tenant": tenant},
|
|
)
|
|
dimension_defs = [{"name": d["name"], "type": "string"} for d in dims]
|
|
return [
|
|
{
|
|
"id": r["metric"],
|
|
"name": r["metric"],
|
|
"description": f"0pt analytics metric {r['metric']}",
|
|
"datasets": ["aud0.events"],
|
|
"dimensions": dimension_defs,
|
|
"tenant_id": tenant,
|
|
**({"events": r.get("events")} if "events" in r else {}),
|
|
}
|
|
for r in rows_
|
|
]
|
|
|
|
|
|
@app.get("/api/0pt/analytics/metrics/{metric_id}")
|
|
def get_metric(metric_id: str, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
|
|
tenant = verified_identity(request, x_tenant_id)
|
|
found = [m for m in list_metrics(request, x_tenant_id) if m["id"] == metric_id]
|
|
if not found:
|
|
raise AnalyticsError("NOT_FOUND", f"metric {metric_id} does not exist for this tenant", 404)
|
|
return found[0]
|
|
|
|
|
|
def _run_query(request: Request, x_tenant_id: str | None, body: QueryRequest, metric: str | None = None) -> dict[str, Any]:
|
|
tenant = verified_identity(request, x_tenant_id)
|
|
metrics = [metric] if metric else body.metrics
|
|
dataset = body.dataset or "aud0.events"
|
|
sql, params = build_query(
|
|
tenant=tenant,
|
|
dataset=dataset,
|
|
metrics=metrics,
|
|
dimensions=body.dimensions,
|
|
filters=body.filters,
|
|
from_time=(body.timeRange or {}).get("from"),
|
|
to_time=(body.timeRange or {}).get("to"),
|
|
limit=body.limit,
|
|
)
|
|
started = time.time()
|
|
data = ch_json(sql, params)
|
|
result_rows = data.get("data", [])
|
|
duration_ms = int((time.time() - started) * 1000)
|
|
columns = []
|
|
if result_rows:
|
|
columns = [{"name": key, "type": "string" if key.startswith("d") else "number"} for key in result_rows[0].keys()]
|
|
return {
|
|
"query_id": f"ch-{int(time.time() * 1000)}",
|
|
"status": "succeeded",
|
|
"engine": "clickhouse",
|
|
"submitted_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
|
"completed_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
|
"duration": duration_ms,
|
|
"rows": len(result_rows),
|
|
"tenant_id": tenant,
|
|
"result": {"columns": columns, "rows": result_rows, "truncated": len(result_rows) >= body.limit},
|
|
}
|
|
|
|
|
|
@app.post("/api/0pt/analytics/query")
|
|
def execute_query(body: QueryRequest, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
|
|
return _run_query(request, x_tenant_id, body)
|
|
|
|
|
|
@app.post("/api/0pt/analytics/metrics/{metric_id}/query")
|
|
def execute_metric(metric_id: str, body: QueryRequest, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
|
|
return _run_query(request, x_tenant_id, body, metric=metric_id)
|
|
|
|
|
|
@app.post("/api/0pt/analytics/query/estimate")
|
|
def estimate_query(body: QueryRequest, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
|
|
tenant = verified_identity(request, x_tenant_id)
|
|
sql, params = build_query(
|
|
tenant=tenant,
|
|
dataset=body.dataset or "aud0.events",
|
|
metrics=body.metrics or ["event_count"],
|
|
dimensions=body.dimensions,
|
|
filters=body.filters,
|
|
from_time=(body.timeRange or {}).get("from"),
|
|
to_time=(body.timeRange or {}).get("to"),
|
|
limit=body.limit,
|
|
)
|
|
count = scalar(
|
|
f"SELECT count() FROM {CLICKHOUSE_DATABASE}.opt_analytics_event WHERE tenant_id = {{tenant:String}} FORMAT JSON",
|
|
{"tenant": tenant},
|
|
)
|
|
return {"estimated_rows_to_read": count or 0, "within_budget": True, "rows": count or 0}
|
|
|
|
|
|
@app.post("/api/0pt/analytics/query/explain")
|
|
def explain_query(body: QueryRequest, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
|
|
tenant = verified_identity(request, x_tenant_id)
|
|
sql, _params = build_query(
|
|
tenant=tenant,
|
|
dataset=body.dataset or "aud0.events",
|
|
metrics=body.metrics or ["event_count"],
|
|
dimensions=body.dimensions,
|
|
filters=body.filters,
|
|
from_time=(body.timeRange or {}).get("from"),
|
|
to_time=(body.timeRange or {}).get("to"),
|
|
limit=body.limit,
|
|
)
|
|
return {"plan": sql, "forced_tenant_predicate": "tenant_id = {tenant:String}", "table": "opt_analytics_event"}
|
|
|
|
|
|
@app.get("/api/0pt/analytics/metrics/{metric_id}/explain")
|
|
def explain_metric(metric_id: str, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
|
|
return explain_query(QueryRequest(metrics=[metric_id]), request, x_tenant_id)
|
|
|
|
|
|
@app.get("/api/0pt/analytics/dashboards")
|
|
def list_dashboards(request: Request, x_tenant_id: str | None = Header(default=None)) -> list[dict[str, Any]]:
|
|
verified_identity(request, x_tenant_id)
|
|
return []
|
|
|
|
|
|
@app.get("/api/0pt/analytics/dashboards/{dashboard_id}")
|
|
def get_dashboard(dashboard_id: str, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
|
|
verified_identity(request, x_tenant_id)
|
|
raise AnalyticsError("NOT_FOUND", f"dashboard {dashboard_id} does not exist", 404)
|
|
|
|
|
|
@app.get("/api/0pt/analytics/reports")
|
|
def list_reports(request: Request, x_tenant_id: str | None = Header(default=None)) -> list[dict[str, Any]]:
|
|
verified_identity(request, x_tenant_id)
|
|
return []
|
|
|
|
|
|
@app.get("/api/0pt/analytics/reports/{report_id}")
|
|
def get_report(report_id: str, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
|
|
verified_identity(request, x_tenant_id)
|
|
raise AnalyticsError("NOT_FOUND", f"report {report_id} does not exist", 404)
|
|
|
|
|
|
@app.get("/api/0pt/analytics/search")
|
|
def search_catalog(
|
|
request: Request,
|
|
q: str = Query(default=""),
|
|
types: str = Query(default=""),
|
|
limit: int = Query(default=20, ge=1, le=100),
|
|
x_tenant_id: str | None = Header(default=None),
|
|
) -> dict[str, Any]:
|
|
tenant = verified_identity(request, x_tenant_id)
|
|
wanted = {t.strip() for t in types.split(",") if t.strip()}
|
|
hits: list[dict[str, Any]] = []
|
|
for dataset in list_datasets(request, x_tenant_id):
|
|
if (not wanted or "dataset" in wanted) and (not q or q.lower() in dataset["id"].lower()):
|
|
hits.append({**dataset, "type": "dataset", "canonical": False, "canonicalOperation": "getDataset"})
|
|
for metric in list_metrics(request, x_tenant_id):
|
|
if (not wanted or "metric" in wanted) and (not q or q.lower() in metric["id"].lower()):
|
|
hits.append({**metric, "type": "metric", "canonical": False, "canonicalOperation": "getMetric"})
|
|
return {"query": q, "derived": True, "hits": hits[:limit], "tenant_id": tenant}
|
|
|
|
|
|
@app.get("/api/0pt/analytics/query/{query_id}")
|
|
def get_query(query_id: str, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
|
|
verified_identity(request, x_tenant_id)
|
|
raise AnalyticsError("NOT_FOUND", f"query {query_id} is not persisted", 404)
|
|
|
|
|
|
@app.post("/api/0pt/analytics/query/{query_id}/cancel")
|
|
def cancel_query(query_id: str, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
|
|
verified_identity(request, x_tenant_id)
|
|
raise AnalyticsError("NOT_FOUND", f"query {query_id} is not running", 404)
|