406 lines
15 KiB
Go
406 lines
15 KiB
Go
package store
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"embed"
|
|
"fmt"
|
|
"log"
|
|
"regexp"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/stdlib"
|
|
)
|
|
|
|
// ── Verbindungen ─────────────────────────────────────────────────────────────
|
|
//
|
|
// PostgreSQL erlaubt nur eine feste Zahl gleichzeitiger Verbindungen
|
|
// (max_connections, Standard 100). Würde Munin mehr öffnen, lehnte PostgreSQL
|
|
// neue Verbindungen ab und Anfragen schlügen fehl. Deshalb gelten hier feste,
|
|
// kleine Pools, und Munin rechnet vor dem Start und vor jedem neuen Workspace
|
|
// nach, ob sie reichen (CheckCapacity). Hintergrund und Rechnung:
|
|
// AUSBAUSTUFEN.md 4.10, „Verbindungen fest begrenzt“.
|
|
//
|
|
// Kleine Pools reichen, weil eine Verbindung nur für die Dauer einer Abfrage
|
|
// belegt ist — Millisekunden. Während ein Agent nachdenkt, hält Munin keine.
|
|
// Ist ein Pool voll, wartet die Anfrage in Go auf eine freie Verbindung (bis
|
|
// zum Ende ihres Kontexts), statt in PostgreSQL zu scheitern.
|
|
const (
|
|
// pgReadConns je Workspace (und für die Kontroll-Datenbank).
|
|
pgReadConns = 2
|
|
|
|
// pgWriteConns je Workspace. Vorerst 1, also dasselbe Verhalten
|
|
// wie unter SQLite: Schreibvorgänge eines Workspace laufen nacheinander.
|
|
// Mehr erst, wenn die schreibenden Methoden gegen gleichzeitiges Schreiben
|
|
// abgesichert sind (AUSBAUSTUFEN.md 4.10, Schritt 4) — sonst kann z.B.
|
|
// PutEntry unter PostgreSQL Änderungen verlieren.
|
|
pgWriteConns = 1
|
|
|
|
// pgConnsPerWorkspace ist, was ein geöffneter Workspace höchstens
|
|
// belegt.
|
|
pgConnsPerWorkspace = pgReadConns + pgWriteConns
|
|
|
|
// pgFixedConns: die Kontroll-Datenbank (wie ein Workspace) plus
|
|
// eine Verwaltungsverbindung zum Anlegen und Löschen von Schemas.
|
|
pgFixedConns = pgConnsPerWorkspace + 1
|
|
|
|
// pgReserve bleibt frei — für psql, Backups (pg_dump) und die
|
|
// Verbindungen, die PostgreSQL selbst für Superuser zurückhält.
|
|
pgReserve = 10
|
|
|
|
// pgIdleTime: Nach so langer Ruhe schließt ein Pool seine Verbindungen.
|
|
// Ein Workspace, den gerade niemand benutzt, belegt dann keine.
|
|
pgIdleTime = 2 * time.Minute
|
|
|
|
// Beim Start ist PostgreSQL vielleicht noch nicht bereit (der Container
|
|
// fährt gerade hoch). Munin versucht es so oft, mit dieser Pause dazwischen.
|
|
pgStartAttempts = 15
|
|
pgStartPause = 2 * time.Second
|
|
)
|
|
|
|
// ── Schemanamen ──────────────────────────────────────────────────────────────
|
|
|
|
// Alles von Munin liegt in Schemas mit dem Präfix "munin_" — auf einen Blick
|
|
// erkennbar (\dn munin_*), auch wenn in derselben Datenbank andere
|
|
// Anwendungen liegen. Begründung: AUSBAUSTUFEN.md 4.10, „Schemanamen aus der
|
|
// Workspace-ID“.
|
|
const (
|
|
ControlSchema = "munin_control"
|
|
workspaceSchemaPrefix = "munin_ws_"
|
|
)
|
|
|
|
// validWorkspaceID passt auf IDs, wie NewID sie erzeugt (Kleinbuchstaben,
|
|
// Ziffern, Bindestrich). Alles andere wird abgelehnt, bevor es in einen
|
|
// Schemanamen wandert.
|
|
var validWorkspaceID = regexp.MustCompile(`^[a-z0-9-]+$`)
|
|
|
|
// WorkspaceSchema liefert den Schemanamen eines Workspace, z.B.
|
|
// "munin_ws_mue9bh3x_1c0e1120".
|
|
//
|
|
// Die ID statt des Namens, weil sich Namen ändern können und freier Text sind.
|
|
// "-" wird zu "_", weil ein Bindestrich in PostgreSQL überall Anführungszeichen
|
|
// erzwingen würde. Den lesbaren Namen trägt ein Kommentar am Schema (siehe
|
|
// afterRename).
|
|
func WorkspaceSchema(workspaceID string) (string, error) {
|
|
if !validWorkspaceID.MatchString(workspaceID) {
|
|
return "", fmt.Errorf("Workspace-ID %q ist als Schemaname nicht geeignet", workspaceID)
|
|
}
|
|
name := workspaceSchemaPrefix + strings.ReplaceAll(workspaceID, "-", "_")
|
|
|
|
// PostgreSQL kürzt längere Namen stillschweigend auf 63 Byte. Zwei lange
|
|
// IDs könnten dann im selben Schema landen — lieber gleich ablehnen.
|
|
if len(name) > 63 {
|
|
return "", fmt.Errorf("Workspace-ID %q ergibt einen zu langen Schemanamen", workspaceID)
|
|
}
|
|
return name, nil
|
|
}
|
|
|
|
// ── Speicher ─────────────────────────────────────────────────────────────────
|
|
|
|
// PostgresBackend legt die Kontroll-Datenbank und jeden Workspace als eigenes
|
|
// Schema in einer PostgreSQL-Datenbank ab.
|
|
//
|
|
// Jedes Schema bekommt eigene Pools, deren Verbindungen mit
|
|
// search_path=<schema> starten. Dadurch laufen dieselben Abfragen wie unter
|
|
// SQLite unverändert — "SELECT … FROM entries" trifft automatisch die Tabelle
|
|
// im richtigen Schema, und ein Workspace kann die Tabellen eines anderen gar
|
|
// nicht erst sehen.
|
|
type PostgresBackend struct {
|
|
base *pgx.ConnConfig // aus DATABASE_URL; je Schema kopiert und ergänzt
|
|
admin *sql.DB // eine Verbindung ohne search_path, für CREATE/DROP SCHEMA
|
|
|
|
// controlSchema ist im Betrieb immer ControlSchema. Nur Tests setzen ein
|
|
// eigenes, damit sie die Kontroll-Datenbank eines Entwicklers nicht
|
|
// anfassen (siehe postgres_test.go).
|
|
controlSchema string
|
|
|
|
// hints sammelt beim Start, was nicht zum Abbruch reicht, aber auf die
|
|
// Systemseite gehört (siehe Hinweise).
|
|
hints []string
|
|
}
|
|
|
|
// NewPostgres verbindet sich mit der Datenbank aus url, wartet nötigenfalls,
|
|
// bis sie bereit ist, und legt das Schema der Kontroll-Datenbank an.
|
|
func NewPostgres(ctx context.Context, url string) (*PostgresBackend, error) {
|
|
base, err := pgx.ParseConfig(url)
|
|
if err != nil {
|
|
// pgx ersetzt das Passwort in der Fehlermeldung durch "xxxxx".
|
|
return nil, fmt.Errorf("DATABASE_URL ist ungültig: %w", err)
|
|
}
|
|
|
|
admin := stdlib.OpenDB(*base)
|
|
admin.SetMaxOpenConns(1)
|
|
admin.SetConnMaxIdleTime(pgIdleTime)
|
|
|
|
p := &PostgresBackend{base: base, admin: admin, controlSchema: ControlSchema}
|
|
|
|
if err := p.waitForDatabase(ctx); err != nil {
|
|
admin.Close()
|
|
return nil, err
|
|
}
|
|
|
|
p.checkCollation(ctx)
|
|
|
|
if err := p.createSchema(ctx, p.controlSchema); err != nil {
|
|
admin.Close()
|
|
return nil, err
|
|
}
|
|
return p, nil
|
|
}
|
|
|
|
// waitForDatabase versucht mehrmals, die Datenbank zu erreichen.
|
|
//
|
|
// Mit SQLite ist die Datei immer da. PostgreSQL läuft dagegen als eigener
|
|
// Dienst und ist nach einem gemeinsamen Start (docker compose up) oft ein paar
|
|
// Sekunden später bereit als Munin. Ohne Wiederholung bräche Munin dann sofort
|
|
// ab.
|
|
func (p *PostgresBackend) waitForDatabase(ctx context.Context) error {
|
|
var lastErr error
|
|
for attempt := 1; attempt <= pgStartAttempts; attempt++ {
|
|
pingCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
|
|
lastErr = p.admin.PingContext(pingCtx)
|
|
cancel()
|
|
if lastErr == nil {
|
|
return nil
|
|
}
|
|
|
|
if attempt < pgStartAttempts {
|
|
log.Printf("PostgreSQL ist noch nicht erreichbar (Versuch %d von %d): %v", attempt, pgStartAttempts, lastErr)
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case <-time.After(pgStartPause):
|
|
}
|
|
}
|
|
}
|
|
return fmt.Errorf("PostgreSQL ist nicht erreichbar (%s): %w", p.Description(), lastErr)
|
|
}
|
|
|
|
// checkCollation warnt, wenn die Datenbank nicht mit der Sortierung "C"
|
|
// angelegt wurde.
|
|
//
|
|
// SQLite sortiert Text Byte für Byte. PostgreSQL sortiert nach der
|
|
// Spracheinstellung der Datenbank, und bei en_US/de_DE werden Bindestriche
|
|
// zunächst übergangen — "a-c" stünde dann vor "ab". Listen wären anders
|
|
// sortiert als unter SQLite. Kein Grund, den Start zu verweigern, aber einer
|
|
// für einen deutlichen Hinweis.
|
|
func (p *PostgresBackend) checkCollation(ctx context.Context) {
|
|
var sortOrder string
|
|
err := p.admin.QueryRowContext(ctx,
|
|
`SELECT datcollate FROM pg_database WHERE datname = current_database()`,
|
|
).Scan(&sortOrder)
|
|
if err != nil {
|
|
log.Printf("Hinweis: Sortierung der Datenbank konnte nicht geprüft werden: %v", err)
|
|
return
|
|
}
|
|
if sortOrder != "C" && sortOrder != "POSIX" {
|
|
hint := fmt.Sprintf("Die Datenbank sortiert nach %q statt \"C\". Listen sind dann anders sortiert als unter SQLite. "+
|
|
"Abhilfe: Datenbank mit --locale=C anlegen (siehe AUSBAUSTUFEN.md 4.10).", sortOrder)
|
|
log.Printf("Warnung: %s", hint)
|
|
p.hints = append(p.hints, hint)
|
|
}
|
|
}
|
|
|
|
// Hints nennt, was beim Start auffiel, etwa eine falsche Sortierung. Die
|
|
// Liste entsteht einmal in NewPostgres: Die Sortierung einer Datenbank lässt
|
|
// sich nachträglich nicht ändern, ein erneutes Nachfragen brächte nichts.
|
|
func (p *PostgresBackend) Hints() []string { return p.hints }
|
|
|
|
// OpenControl öffnet die Kontroll-Datenbank im Schema munin_control.
|
|
//
|
|
// Der ControlStore bekommt den Speicher als Haken mit: Er prüft vor jedem neuen
|
|
// Workspace die Verbindungen und legt danach dessen Schema samt lesbarem Namen
|
|
// an (siehe beforeCreate, afterRename).
|
|
func (p *PostgresBackend) OpenControl() (*ControlStore, error) {
|
|
read, write, err := p.openPools(p.controlSchema, controlMigrationsFS, "migrations_control")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &ControlStore{read: read, write: write, hooks: p}, nil
|
|
}
|
|
|
|
// OpenWorkspace öffnet das Schema eines Workspace und legt es an, falls es
|
|
// noch fehlt.
|
|
func (p *PostgresBackend) OpenWorkspace(workspaceID string) (*Store, error) {
|
|
schema, err := WorkspaceSchema(workspaceID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if err := p.createSchema(context.Background(), schema); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
read, write, err := p.openPools(schema, migrationsFS, "migrations")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &Store{read: read, write: write}, nil
|
|
}
|
|
|
|
// DeleteWorkspace löscht das Schema eines Workspace samt aller Tabellen und
|
|
// Daten darin.
|
|
func (p *PostgresBackend) DeleteWorkspace(workspaceID string) error {
|
|
schema, err := WorkspaceSchema(workspaceID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if _, err := p.admin.ExecContext(context.Background(),
|
|
`DROP SCHEMA IF EXISTS `+pgx.Identifier{schema}.Sanitize()+` CASCADE`,
|
|
); err != nil {
|
|
return fmt.Errorf("Schema %s konnte nicht gelöscht werden: %w", schema, err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Capacity liest max_connections und nennt die Größen, mit denen
|
|
// CheckCapacity rechnet.
|
|
func (p *PostgresBackend) Capacity(ctx context.Context) (Capacity, error) {
|
|
allowed, err := p.maxConnections(ctx)
|
|
if err != nil {
|
|
return Capacity{}, err
|
|
}
|
|
return Capacity{
|
|
Limited: true, Allowed: allowed, Reserve: pgReserve,
|
|
PerWorkspace: pgConnsPerWorkspace, Fixed: pgFixedConns,
|
|
}, nil
|
|
}
|
|
|
|
// maxConnections liest max_connections des Servers.
|
|
func (p *PostgresBackend) maxConnections(ctx context.Context) (int, error) {
|
|
var text string
|
|
if err := p.admin.QueryRowContext(ctx, `SHOW max_connections`).Scan(&text); err != nil {
|
|
return 0, fmt.Errorf("max_connections konnte nicht gelesen werden: %w", err)
|
|
}
|
|
allowed, err := strconv.Atoi(text)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("max_connections hat einen unerwarteten Wert %q", text)
|
|
}
|
|
return allowed, nil
|
|
}
|
|
|
|
// CheckCapacity rechnet nach, ob count Workspaces mit ihren Pools unter
|
|
// max_connections bleiben (abzüglich Reserve).
|
|
//
|
|
// max_connections wird jedes Mal frisch gelesen: Wer es nach einer Meldung
|
|
// erhöht und PostgreSQL neu startet, soll Munin dafür nicht auch neu starten
|
|
// müssen.
|
|
func (p *PostgresBackend) CheckCapacity(ctx context.Context, count int) error {
|
|
allowed, err := p.maxConnections(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
demand := count*pgConnsPerWorkspace + pgFixedConns
|
|
if demand > allowed-pgReserve {
|
|
return fmt.Errorf("zu wenige PostgreSQL-Verbindungen: %d Workspaces brauchen bis zu %d Verbindungen "+
|
|
"(%d je Workspace + %d fest), max_connections ist %d (davon %d Reserve). "+
|
|
"max_connections auf mindestens %d erhöhen",
|
|
count, demand, pgConnsPerWorkspace, pgFixedConns, allowed, pgReserve, demand+pgReserve)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Location nennt das Schema des Workspace, z.B. "munin_ws_mue9bh3x_1c0e1120".
|
|
// Eine ID, die keinen gültigen Schemanamen ergibt, kann es hier nicht geben
|
|
// (OpenWorkspace hätte sie abgelehnt); falls doch, steht der Grund da.
|
|
func (p *PostgresBackend) Location(workspaceID string) string {
|
|
schema, err := WorkspaceSchema(workspaceID)
|
|
if err != nil {
|
|
return err.Error()
|
|
}
|
|
return schema
|
|
}
|
|
|
|
// Description nennt Server und Datenbank, nie das Passwort.
|
|
func (p *PostgresBackend) Description() string {
|
|
return fmt.Sprintf("PostgreSQL — %s:%d/%s (Schemas %s, %s…)",
|
|
p.base.Host, p.base.Port, p.base.Database, ControlSchema, workspaceSchemaPrefix)
|
|
}
|
|
|
|
// Close schließt die Verwaltungsverbindung.
|
|
func (p *PostgresBackend) Close() error { return p.admin.Close() }
|
|
|
|
// ── Haken für den ControlStore ───────────────────────────────────────────────
|
|
|
|
func (p *PostgresBackend) beforeCreate(ctx context.Context, count int) error {
|
|
return p.CheckCapacity(ctx, count)
|
|
}
|
|
|
|
// afterRename legt das Schema des Workspace an (falls noch nicht da) und
|
|
// schreibt den lesbaren Namen als Kommentar daran. In psql zeigt \dn+ ihn
|
|
// neben dem Schemanamen.
|
|
func (p *PostgresBackend) afterRename(ctx context.Context, workspaceID, name string) error {
|
|
schema, err := WorkspaceSchema(workspaceID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := p.createSchema(ctx, schema); err != nil {
|
|
return err
|
|
}
|
|
|
|
// COMMENT ON nimmt keine Platzhalter ($1) an, der Name muss also als
|
|
// Text in die Anweisung. textLiteral sorgt dafür, dass er dort nichts
|
|
// anderes sein kann als Text.
|
|
if _, err := p.admin.ExecContext(ctx,
|
|
`COMMENT ON SCHEMA `+pgx.Identifier{schema}.Sanitize()+` IS `+textLiteral(name),
|
|
); err != nil {
|
|
return fmt.Errorf("Name konnte nicht am Schema %s vermerkt werden: %w", schema, err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// ── Hilfsfunktionen ──────────────────────────────────────────────────────────
|
|
|
|
func (p *PostgresBackend) createSchema(ctx context.Context, schema string) error {
|
|
if _, err := p.admin.ExecContext(ctx,
|
|
`CREATE SCHEMA IF NOT EXISTS `+pgx.Identifier{schema}.Sanitize(),
|
|
); err != nil {
|
|
return fmt.Errorf("Schema %s konnte nicht angelegt werden: %w", schema, err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// openPools öffnet Lese- und Schreibpool für ein Schema und bringt dessen
|
|
// Tabellen auf den aktuellen Stand — das Gegenstück zu openSQLite.
|
|
func (p *PostgresBackend) openPools(schema string, fsys embed.FS, migrationsDir string) (read, write pool, err error) {
|
|
d := postgresDialect{}
|
|
|
|
// Jede Verbindung dieser Pools startet mit search_path=<schema>. Kopie,
|
|
// damit die Grundeinstellung für die anderen Schemas unverändert bleibt.
|
|
cfg := p.base.Copy()
|
|
cfg.RuntimeParams["search_path"] = schema
|
|
|
|
open := func(maxConns int) *sql.DB {
|
|
db := stdlib.OpenDB(*cfg)
|
|
db.SetMaxOpenConns(maxConns)
|
|
db.SetMaxIdleConns(maxConns)
|
|
db.SetConnMaxIdleTime(pgIdleTime)
|
|
return db
|
|
}
|
|
|
|
writeDB := open(pgWriteConns)
|
|
write = pool{db: writeDB, d: d}
|
|
if err := runMigrations(write, fsys, migrationsDir); err != nil {
|
|
writeDB.Close()
|
|
return pool{}, pool{}, fmt.Errorf("Schema %s: %w", schema, err)
|
|
}
|
|
|
|
read = pool{db: open(pgReadConns), d: d}
|
|
return read, write, nil
|
|
}
|
|
|
|
// textLiteral schreibt s als PostgreSQL-Textkonstante, z.B. E'Vertrieb & Co'.
|
|
//
|
|
// Die Form E'…' wertet Backslashes immer gleich aus, unabhängig von der
|
|
// Einstellung standard_conforming_strings. Darin werden Backslash und
|
|
// Hochkomma verdoppelt; ein Nullbyte kann PostgreSQL in Text ohnehin nicht
|
|
// speichern und wird entfernt.
|
|
func textLiteral(s string) string {
|
|
s = strings.ReplaceAll(s, "\x00", "")
|
|
s = strings.ReplaceAll(s, `\`, `\\`)
|
|
s = strings.ReplaceAll(s, `'`, `''`)
|
|
return "E'" + s + "'"
|
|
}
|