@@ -0,0 +1,382 @@
package stats
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"log"
"net/http"
"strings"
"sync"
"time"
)
// GrafanaProvider реализует StatsProvider через Grafana HTTP API.
//
// Для каждого namespace создаётся изолированная Grafana Organization:
// - PostgreSQL datasource (тот же DSN, uid="fission-user-pg")
// - Dashboard с
// - Public Dashboard (без логина) → accessToken
//
// Потокобезопасен: sync.RWMutex + per-namespace singleflight.
type GrafanaProvider struct {
internalURL string // http://grafana.grafana.svc.cluster.local:3000
publicURL string // https://fission.kube5s.ru/grafana
adminUser string
adminPass string
http * http . Client
mu sync . RWMutex
tokens map [ string ] string // namespace → publicDashboardAccessToken
orgIDs map [ string ] int64 // namespace → grafana orgId
}
// NewGrafanaProvider создаёт GrafanaProvider.
func NewGrafanaProvider ( internalURL , publicURL , adminUser , adminPass string ) * GrafanaProvider {
return & GrafanaProvider {
internalURL : strings . TrimRight ( internalURL , "/" ) ,
publicURL : strings . TrimRight ( publicURL , "/" ) ,
adminUser : adminUser ,
adminPass : adminPass ,
http : & http . Client { Timeout : 20 * time . Second } ,
tokens : make ( map [ string ] string ) ,
orgIDs : make ( map [ string ] int64 ) ,
}
}
// EnsureOrgForNamespace идемпотентно создаёт Grafana Org + datasource + dashboard + public link.
func ( g * GrafanaProvider ) EnsureOrgForNamespace ( ctx context . Context , namespace , email string ) error {
// Быстрый путь: уже провизировано в этом процессе
g . mu . RLock ( )
_ , cached := g . tokens [ namespace ]
g . mu . RUnlock ( )
if cached {
return nil
}
// Шаг 1: получить или создать Org
orgID , err := g . getOrCreateOrg ( ctx , namespace )
if err != nil {
return fmt . Errorf ( "getOrCreateOrg(%s): %w" , namespace , err )
}
// Шаг 2: создать datasource в этой Org (идемпотентно)
if err := g . ensureDatasource ( ctx , orgID ) ; err != nil {
log . Printf ( "stats: ensureDatasource org=%d ns=%s: %v" , orgID , namespace , err )
// не фатально — dashboard может не работать но org создана
}
// Шаг 3: создать dashboard с
dashUID , err := g . ensureDashboard ( ctx , orgID , namespace )
if err != nil {
return fmt . Errorf ( "ensureDashboard org=%d ns=%s: %w" , orgID , namespace , err )
}
// Шаг 4: получить или создать public dashboard → accessToken
token , err := g . ensurePublicDashboard ( ctx , orgID , dashUID )
if err != nil {
return fmt . Errorf ( "ensurePublicDashboard org=%d dash=%s: %w" , orgID , dashUID , err )
}
// Кэшируем
g . mu . Lock ( )
g . tokens [ namespace ] = token
g . orgIDs [ namespace ] = orgID
g . mu . Unlock ( )
log . Printf ( "stats: org provisioned ns=%s orgId=%d publicToken=%s..." , namespace , orgID , token [ : 8 ] )
return nil
}
// DashboardURL возвращает публичный URL или "" если ещё не провизировано.
func ( g * GrafanaProvider ) DashboardURL ( ctx context . Context , namespace string ) string {
// Сначала пробуем из кэша
g . mu . RLock ( )
token , ok := g . tokens [ namespace ]
g . mu . RUnlock ( )
if ok && token != "" {
return g . publicURL + "/public-dashboards/" + token
}
// Кэш промах (после перезапуска сервера) — провизируем заново
if err := g . EnsureOrgForNamespace ( ctx , namespace , "" ) ; err != nil {
log . Printf ( "stats: DashboardURL re-provision ns=%s: %v" , namespace , err )
return ""
}
g . mu . RLock ( )
token = g . tokens [ namespace ]
g . mu . RUnlock ( )
if token == "" {
return ""
}
return g . publicURL + "/public-dashboards/" + token
}
// --- Grafana API helpers ---
// getOrCreateOrg возвращает orgId существующей или создаёт новую Org.
func ( g * GrafanaProvider ) getOrCreateOrg ( ctx context . Context , namespace string ) ( int64 , error ) {
// Проверяем кэш orgIDs
g . mu . RLock ( )
if id , ok := g . orgIDs [ namespace ] ; ok {
g . mu . RUnlock ( )
return id , nil
}
g . mu . RUnlock ( )
// GET /api/orgs/name/{namespace}
resp , body , err := g . grafanaRequest ( ctx , http . MethodGet , "/api/orgs/name/" + namespace , 0 , nil )
if err != nil {
return 0 , err
}
if resp . StatusCode == http . StatusOK {
var org struct {
ID int64 ` json:"id" `
}
if err := json . Unmarshal ( body , & org ) ; err != nil {
return 0 , fmt . Errorf ( "parse org: %w" , err )
}
return org . ID , nil
}
// Org не найдена — создаём
payload := map [ string ] string { "name" : namespace }
resp , body , err = g . grafanaRequest ( ctx , http . MethodPost , "/api/orgs" , 0 , payload )
if err != nil {
return 0 , err
}
if resp . StatusCode != http . StatusOK && resp . StatusCode != http . StatusCreated {
return 0 , fmt . Errorf ( "create org status=%d body=%s" , resp . StatusCode , string ( body ) )
}
var created struct {
OrgID int64 ` json:"orgId" `
}
if err := json . Unmarshal ( body , & created ) ; err != nil {
return 0 , fmt . Errorf ( "parse create org: %w" , err )
}
return created . OrgID , nil
}
// ensureDatasource создаёт PostgreSQL datasource в org (uid="fission-user-pg").
// Идемпотентен: 409 Conflict считается успехом.
func ( g * GrafanaProvider ) ensureDatasource ( ctx context . Context , orgID int64 ) error {
// Проверяем есть ли уже datasource в этой org
resp , _ , err := g . grafanaRequest ( ctx , http . MethodGet , "/api/datasources/uid/fission-user-pg" , orgID , nil )
if err != nil {
return err
}
if resp . StatusCode == http . StatusOK {
return nil // уже есть
}
// Получаем DSN из уже существующего datasource в Org 1 (uid=fission-pg)
_ , body , err := g . grafanaRequest ( ctx , http . MethodGet , "/api/datasources/uid/fission-pg" , 1 , nil )
if err != nil {
return fmt . Errorf ( "get main datasource: %w" , err )
}
var ds struct {
URL string ` json:"url" `
JSONData json . RawMessage ` json:"jsonData" `
SecureJSONData struct {
Password string ` json:"password" `
} ` json:"secureJsonData" `
}
if err := json . Unmarshal ( body , & ds ) ; err != nil {
return fmt . Errorf ( "parse main datasource: %w" , err )
}
// Создаём копию datasource в новой Org
payload := map [ string ] any {
"name" : "fission-pg" ,
"type" : "postgres" ,
"uid" : "fission-user-pg" ,
"url" : ds . URL ,
"access" : "proxy" ,
"jsonData" : map [ string ] any {
"sslmode" : "disable" ,
"postgresVersion" : 1700 ,
"timescaledb" : false ,
} ,
"secureJsonData" : ds . SecureJSONData ,
}
resp , body , err = g . grafanaRequest ( ctx , http . MethodPost , "/api/datasources" , orgID , payload )
if err != nil {
return err
}
if resp . StatusCode == http . StatusConflict {
return nil // уже существует
}
if resp . StatusCode != http . StatusOK && resp . StatusCode != http . StatusCreated {
return fmt . Errorf ( "create datasource status=%d body=%s" , resp . StatusCode , string ( body ) )
}
return nil
}
// ensureDashboard создаёт/обновляет user-дашборд в org.
// Возвращает uid дашборда.
func ( g * GrafanaProvider ) ensureDashboard ( ctx context . Context , orgID int64 , namespace string ) ( string , error ) {
const dashUID = "fission-user-overview"
// Проверяем существование
resp , _ , err := g . grafanaRequest ( ctx , http . MethodGet , "/api/dashboards/uid/" + dashUID , orgID , nil )
if err != nil {
return "" , err
}
if resp . StatusCode == http . StatusOK {
return dashUID , nil // уже есть
}
// Импортируем dashboard JSON с
dashJSON := userDashboardJSON ( namespace )
payload := map [ string ] any {
"dashboard" : json . RawMessage ( dashJSON ) ,
"overwrite" : true ,
"folderId" : 0 ,
}
resp , body , err := g . grafanaRequest ( ctx , http . MethodPost , "/api/dashboards/db" , orgID , payload )
if err != nil {
return "" , err
}
if resp . StatusCode != http . StatusOK {
return "" , fmt . Errorf ( "import dashboard status=%d body=%s" , resp . StatusCode , string ( body ) )
}
return dashUID , nil
}
// ensurePublicDashboard создаёт public dashboard и возвращает accessToken.
// Идемпотентен: если уже существует — возвращает существующий token.
func ( g * GrafanaProvider ) ensurePublicDashboard ( ctx context . Context , orgID int64 , dashUID string ) ( string , error ) {
path := "/api/dashboards/uid/" + dashUID + "/public-dashboards"
// Проверяем существование
resp , body , err := g . grafanaRequest ( ctx , http . MethodGet , path , orgID , nil )
if err != nil {
return "" , err
}
if resp . StatusCode == http . StatusOK {
var pd struct {
AccessToken string ` json:"accessToken" `
}
if err := json . Unmarshal ( body , & pd ) ; err != nil {
return "" , fmt . Errorf ( "parse public dashboard: %w" , err )
}
if pd . AccessToken != "" {
return pd . AccessToken , nil
}
}
// Создаём
payload := map [ string ] any {
"isEnabled" : true ,
"annotationsEnabled" : false ,
"timeSelectionEnabled" : true ,
}
resp , body , err = g . grafanaRequest ( ctx , http . MethodPost , path , orgID , payload )
if err != nil {
return "" , err
}
if resp . StatusCode != http . StatusOK && resp . StatusCode != http . StatusCreated {
return "" , fmt . Errorf ( "create public dashboard status=%d body=%s" , resp . StatusCode , string ( body ) )
}
var pd struct {
AccessToken string ` json:"accessToken" `
}
if err := json . Unmarshal ( body , & pd ) ; err != nil {
return "" , fmt . Errorf ( "parse created public dashboard: %w" , err )
}
if pd . AccessToken == "" {
return "" , fmt . Errorf ( "empty accessToken in response: %s" , string ( body ) )
}
return pd . AccessToken , nil
}
// grafanaRequest выполняет HTTP запрос к Grafana API.
// orgID > 0 → устанавливает X-Grafana-Org-Id заголовок (thread-safe, без смены контекста).
// orgID == 0 → без заголовка (используется Org 1 admin по умолчанию).
func ( g * GrafanaProvider ) grafanaRequest ( ctx context . Context , method , path string , orgID int64 , payload any ) ( * http . Response , [ ] byte , error ) {
var bodyReader io . Reader
if payload != nil {
data , err := json . Marshal ( payload )
if err != nil {
return nil , nil , fmt . Errorf ( "marshal payload: %w" , err )
}
bodyReader = bytes . NewReader ( data )
}
req , err := http . NewRequestWithContext ( ctx , method , g . internalURL + path , bodyReader )
if err != nil {
return nil , nil , fmt . Errorf ( "new request: %w" , err )
}
req . SetBasicAuth ( g . adminUser , g . adminPass )
if payload != nil {
req . Header . Set ( "Content-Type" , "application/json" )
}
if orgID > 0 {
req . Header . Set ( "X-Grafana-Org-Id" , fmt . Sprintf ( "%d" , orgID ) )
}
resp , err := g . http . Do ( req )
if err != nil {
return nil , nil , fmt . Errorf ( "do request %s %s: %w" , method , path , err )
}
defer resp . Body . Close ( )
body , err := io . ReadAll ( resp . Body )
if err != nil {
return resp , nil , fmt . Errorf ( "read body: %w" , err )
}
return resp , body , nil
}
// userDashboardJSON генерирует JSON дашборда для конкретного namespace.
// Namespace вшит прямо в SQL запросы — без template variables.
// Dashboard uid="fission-user-overview" (per-org, без конфликтов между org).
func userDashboardJSON ( namespace string ) string {
// Безопасное экранирование namespace для SQL (namespace это sha256 hex — только [a-z0-9-])
ns := strings . ReplaceAll ( namespace , "'" , "''" )
return fmt . Sprintf ( ` {
"title": "Мои функции — %s",
"uid": "fission-user-overview",
"tags": ["fission", "user"],
"timezone": "browser",
"refresh": "1m",
"time": { "from": "now-24h", "to": "now"},
"panels": [
{
"id": 1, "title": "Вызовы в час", "type": "timeseries",
"gridPos": { "x": 0, "y": 0, "w": 16, "h": 8},
"datasource": { "type": "postgres", "uid": "fission-user-pg"},
"targets": [ { "rawSql": "SELECT date_trunc('hour', started_at) AS time, count(*) AS value, function_name FROM invocations WHERE namespace = '%s' AND started_at BETWEEN $__timeFrom() AND $__timeTo() GROUP BY 1, function_name ORDER BY 1", "format": "time_series", "refId": "A"}]
},
{
"id": 2, "title": "Успех vs Ошибки", "type": "piechart",
"gridPos": { "x": 16, "y": 0, "w": 8, "h": 8},
"datasource": { "type": "postgres", "uid": "fission-user-pg"},
"targets": [ { "rawSql": "SELECT CASE WHEN status_code >= 200 AND status_code < 300 THEN 'success' WHEN status_code = 0 THEN 'event' ELSE 'error' END AS metric, count(*) AS value FROM invocations WHERE namespace = '%s' AND started_at BETWEEN $__timeFrom() AND $__timeTo() GROUP BY 1", "format": "table", "refId": "A"}]
},
{
"id": 3, "title": "Топ функций", "type": "bargauge",
"gridPos": { "x": 0, "y": 8, "w": 12, "h": 8},
"datasource": { "type": "postgres", "uid": "fission-user-pg"},
"targets": [ { "rawSql": "SELECT function_name AS metric, count(*) AS value FROM invocations WHERE namespace = '%s' AND started_at BETWEEN $__timeFrom() AND $__timeTo() GROUP BY 1 ORDER BY 2 DESC LIMIT 10", "format": "table", "refId": "A"}]
},
{
"id": 4, "title": "Средняя латентность (ms)", "type": "timeseries",
"gridPos": { "x": 12, "y": 8, "w": 12, "h": 8},
"datasource": { "type": "postgres", "uid": "fission-user-pg"},
"targets": [ { "rawSql": "SELECT date_trunc('hour', started_at) AS time, round(avg(duration_ms)) AS avg_ms FROM invocations WHERE namespace = '%s' AND started_at BETWEEN $__timeFrom() AND $__timeTo() AND trigger_type != 'event' GROUP BY 1 ORDER BY 1", "format": "time_series", "refId": "A"}]
},
{
"id": 5, "title": "Последние события", "type": "table",
"gridPos": { "x": 0, "y": 16, "w": 24, "h": 8},
"datasource": { "type": "postgres", "uid": "fission-user-pg"},
"targets": [ { "rawSql": "SELECT started_at AS time, function_name, trigger_type, event_type, status_code, duration_ms, error_msg FROM invocations WHERE namespace = '%s' ORDER BY started_at DESC LIMIT 50", "format": "table", "refId": "A"}]
}
],
"schemaVersion": 39
} ` , namespace , ns , ns , ns , ns , ns )
}
“Naeel”