Compare commits
8
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 sysops
|
5fcae51aac | ||
|
sysops
|
369a40af10 | ||
|
sysops
|
06dbd52d4c | ||
|
sysops
|
da80643564 | ||
|
sysops
|
814a7fda0a | ||
|
sysopsandClaude Sonnet 5
|
6a03dcafd6 | ||
|
sysopsandClaude Sonnet 5
|
6f532d8350 | ||
|
sysopsandClaude Sonnet 5
|
4180a26c6e |
@@ -129,3 +129,34 @@ Keine Commits in dieser Session.
|
||||
- migrations/0001_tenant_registry.sql | 10 ++++++++++
|
||||
|
||||
---
|
||||
## 2026-08-28 23:55 – 23:59 (3m)
|
||||
**Beschreibung:** Claude Code Session
|
||||
**Projekt:** nexarch
|
||||
|
||||
### Commits
|
||||
- 369a40a OPS-05: alerting-bei-schwellwert-ueberschreitung (internal/alerting: regel-store, evaluator gegen ops-03-metriken, cfg-02-zustellung, drosselung je regel+zeitreihe)
|
||||
- 06dbd52 Merge branch 'feature/cfg-02-benachrichtigungs-dispatcher-core-service-fuer-module' into feature/ops-05-alerting-bei-schwellwert-ueberschreitung
|
||||
|
||||
### Geänderte Dateien
|
||||
- internal/alerting/evaluator.go | 194 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
|
||||
- internal/alerting/evaluator_test.go | 233 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
|
||||
- internal/alerting/rules.go | 132 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
|
||||
- migrations/0006_alert_rules.down.sql | 2 ++
|
||||
- migrations/0006_alert_rules.up.sql | 24 ++++++++++++++++++++++
|
||||
|
||||
---
|
||||
## 2026-08-29 00:00 – 00:00 (0m)
|
||||
**Beschreibung:** Claude Code Session
|
||||
**Projekt:** code
|
||||
|
||||
### Commits
|
||||
Keine Commits in dieser Session.
|
||||
|
||||
### Geänderte Dateien
|
||||
- internal/alerting/evaluator.go | 194 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
|
||||
- internal/alerting/evaluator_test.go | 233 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
|
||||
- internal/alerting/rules.go | 132 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
|
||||
- migrations/0006_alert_rules.down.sql | 2 ++
|
||||
- migrations/0006_alert_rules.up.sql | 24 ++++++++++++++++++++++
|
||||
|
||||
---
|
||||
|
||||
@@ -1,82 +0,0 @@
|
||||
// kek-api ist der Aufrufpunkt fuer API-12: startet den bereits fertigen
|
||||
// API-10-Handler (internal/kek) als eigenstaendigen HTTP-Dienst.
|
||||
// REINES WIRING — keine Aenderung an internal/kek/, internal/moduleregistry/
|
||||
// oder internal/tenant/.
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"net/http"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/flag"
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/kek"
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/moduleregistry"
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/tenant"
|
||||
)
|
||||
|
||||
// tenantResolverAdapter erfüllt kek.TenantResolver über den bestehenden
|
||||
// tenant.Registry.GetBySlug-Zugriff — kein neuer Tenant-Code, nur
|
||||
// Signatur-Anpassung.
|
||||
type tenantResolverAdapter struct{ registry *tenant.Registry }
|
||||
|
||||
func (a tenantResolverAdapter) ResolveTenantID(ctx context.Context, tenantSlug string) (string, error) {
|
||||
t, err := a.registry.GetBySlug(ctx, tenantSlug)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return t.ID, nil
|
||||
}
|
||||
|
||||
func requireEnv(name string) string {
|
||||
v := os.Getenv(name)
|
||||
if v == "" {
|
||||
log.Fatalf("%s muss gesetzt sein", name)
|
||||
}
|
||||
return v
|
||||
}
|
||||
|
||||
func main() {
|
||||
registryDSN := requireEnv("NEXARCH_KEK_REGISTRY_DSN")
|
||||
masterKeyEnvVar := os.Getenv("NEXARCH_KEK_MASTER_KEY_ENV")
|
||||
if masterKeyEnvVar == "" {
|
||||
masterKeyEnvVar = "NEXARCH_KEK_MASTER_KEY"
|
||||
}
|
||||
addr := os.Getenv("NEXARCH_KEK_API_LISTEN_ADDR")
|
||||
if addr == "" {
|
||||
addr = "127.0.0.1:8102"
|
||||
}
|
||||
|
||||
masterKey, err := kek.LoadMasterKeyFromEnv(masterKeyEnvVar)
|
||||
if err != nil {
|
||||
log.Fatalf("master-key laden: %v", err)
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
pool, err := pgxpool.New(ctx, registryDSN)
|
||||
if err != nil {
|
||||
log.Fatalf("datenbankverbindung: %v", err)
|
||||
}
|
||||
defer pool.Close()
|
||||
|
||||
flagStore := flag.NewStore(pool)
|
||||
flagService := flag.NewService(flagStore, 30*time.Second)
|
||||
registry := moduleregistry.NewRegistry(pool, flagService)
|
||||
tenantRegistry := tenant.NewRegistry(pool)
|
||||
store := kek.NewStore(pool)
|
||||
|
||||
handler := kek.NewHandler(store, masterKey, registry, registry, tenantResolverAdapter{registry: tenantRegistry})
|
||||
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })
|
||||
mux.HandleFunc("/internal/kek/tenant", handler.TenantKEKHandler)
|
||||
|
||||
log.Printf("kek-api: listening on %s", addr)
|
||||
if err := http.ListenAndServe(addr, mux); err != nil {
|
||||
log.Fatalf("http server: %v", err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
// metrics-devserver stellt den OPS-03-Metrics-Aggregator (internal/metrics)
|
||||
// unter /metrics bereit, damit ein echter Prometheus-Scrape-Vorgang gegen
|
||||
// den Core-Dienst geprueft werden kann (Pruefung 3). Getrennt von cmd/core
|
||||
// aus demselben Grund wie die anderen *-devserver.
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"net/http"
|
||||
"os"
|
||||
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/db"
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/metrics"
|
||||
)
|
||||
|
||||
func main() {
|
||||
dsn := os.Getenv("NEXARCH_REGISTRY_DSN")
|
||||
if dsn == "" {
|
||||
log.Fatal("NEXARCH_REGISTRY_DSN nicht gesetzt")
|
||||
}
|
||||
addr := os.Getenv("NEXARCH_METRICS_LISTEN_ADDR")
|
||||
if addr == "" {
|
||||
addr = ":8085"
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
pool, err := db.Connect(ctx, dsn)
|
||||
if err != nil {
|
||||
log.Fatalf("db: %v", err)
|
||||
}
|
||||
defer pool.Close()
|
||||
|
||||
sourceStore := metrics.NewSourceStore(pool)
|
||||
agg := metrics.NewAggregator(metrics.NewCoreRegistry(), sourceStore.Provide)
|
||||
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("/metrics", agg.Handler())
|
||||
mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })
|
||||
|
||||
log.Printf("metrics-devserver listening on %s", addr)
|
||||
log.Fatal(http.ListenAndServe(addr, mux))
|
||||
}
|
||||
@@ -1,14 +0,0 @@
|
||||
[Unit]
|
||||
Description=NEXARCH Core - KEK-Bezugsdienst (API-10/API-12)
|
||||
After=network.target postgresql.service
|
||||
|
||||
[Service]
|
||||
Type=simple
|
||||
User=nexarch
|
||||
EnvironmentFile=/etc/nexarch/kek-api.env
|
||||
ExecStart=__INSTALL_DIR__/bin/kek-api
|
||||
Restart=on-failure
|
||||
StandardOutput=journal
|
||||
|
||||
[Install]
|
||||
WantedBy=multi-user.target
|
||||
@@ -1,61 +0,0 @@
|
||||
# API-12 – Prüfprotokoll: KEK-Bezugsdienst starten (API-10 als laufender Dienst)
|
||||
|
||||
Voraussetzung API-10 – bereits Fertig, hier UNVERÄNDERT.
|
||||
|
||||
## Reines Wiring, keine neue Logik
|
||||
|
||||
`git diff --stat internal/kek/ internal/moduleregistry/ internal/tenant/`
|
||||
liefert KEINEN Diff. `cmd/kek-api/main.go` setzt ausschließlich
|
||||
bestehende Konstruktoren zusammen; `tenantResolverAdapter` bildet nur
|
||||
`tenant.Registry.GetBySlug` auf `kek.TenantResolver` ab (Signatur-
|
||||
Anpassung, kein neuer Fachcode).
|
||||
|
||||
## Umsetzung
|
||||
|
||||
- `cmd/kek-api/main.go` – `POST /internal/kek/tenant?tenant=<slug>`,
|
||||
authentifiziert über dasselbe Service-Credential-Verfahren wie jeder
|
||||
andere Modul-Core-Aufruf (API-02), zusätzlich Tenant-Aktivierungs-
|
||||
prüfung (identisches Muster wie in `internal/kek.Handler` bereits
|
||||
vorgesehen).
|
||||
- `deploy/systemd/nexarch-kek-api.service.tmpl`.
|
||||
|
||||
## Prüfungen
|
||||
|
||||
| # | Prüfung | Ergebnis |
|
||||
|---|---|---|
|
||||
| 1 | Dienst startet und bleibt stabil (systemctl status aktiv) | **bestanden** – real auf 131: `nexarch-kek-api.service` aktiv |
|
||||
| 2 | Realer Aufruf mit gültigem Service-Credential liefert den erwarteten Tenant-KEK, ohne/mit falschem Credential wird abgelehnt | **bestanden** – real per `curl`: echtes Modul registriert+provisioniert, echter Tenant-KEK über `kek.Store.CreateForTenant` erzeugt (Klartext-Hex zum Vergleich notiert) — Aufruf mit korrektem Credential liefert exakt denselben KEK (Base64-dekodiert übereinstimmend mit dem erzeugten Hex-Wert verifiziert); Aufruf mit falschem Credential → 403 |
|
||||
| 3 | Code-Review: keine Änderung an internal/kek/ selbst, nur main.go+systemd neu | **bestanden** – `git diff --stat` bestätigt: `internal/kek/`, `internal/moduleregistry/`, `internal/tenant/` unverändert |
|
||||
|
||||
## Echte Verdrahtung auf 192.168.1.131
|
||||
|
||||
- `kek-api` gebaut nach `/opt/nexarch-core/bin/`,
|
||||
`/etc/nexarch/kek-api.env` (0600, echter zufälliger 32-Byte-
|
||||
Master-Key), Dienst installiert/aktiviert.
|
||||
- Reale Grant-Lücke gefunden und behoben (gleiches Muster wie zuvor):
|
||||
`nexarch_core` hatte keine Rechte auf `tenant_keks` — `GRANT`
|
||||
nachgezogen und über `information_schema.role_table_grants`
|
||||
verifiziert.
|
||||
- End-zu-Ende-Beweis: echtes Modul registriert, Service-Credential
|
||||
provisioniert, echter Tenant + Tenant-KEK real erzeugt, `curl` gegen
|
||||
den laufenden Dienst liefert exakt diesen KEK zurück (Byte-für-Byte
|
||||
verglichen), falsches Credential real abgelehnt. Testdaten
|
||||
anschließend entfernt.
|
||||
|
||||
## Build/Test-Ergebnis (192.168.1.131)
|
||||
|
||||
```
|
||||
go build ./... -> clean
|
||||
go vet ./... -> clean
|
||||
golangci-lint run ./cmd/kek-api/... -> 0 issues
|
||||
```
|
||||
|
||||
Keine neuen Go-Tests nötig (kein neuer Fachcode außer main.go/Adapter,
|
||||
die eigentliche Logik ist bereits durch API-10s eigene Tests
|
||||
abgedeckt).
|
||||
|
||||
## Gesamtergebnis
|
||||
|
||||
**Bestanden.** API-10 ist jetzt ein real laufender, über systemd
|
||||
verwalteter Dienst — Voraussetzung für Mail ARC-02 und künftig DMS
|
||||
FDN-09-Nachnutzung.
|
||||
@@ -1,14 +1,25 @@
|
||||
module gitea.perlbach24.de/scripte/nexarch
|
||||
|
||||
go 1.22
|
||||
|
||||
require github.com/jackc/pgx/v5 v5.6.0
|
||||
go 1.25.0
|
||||
|
||||
require (
|
||||
github.com/jackc/pgx/v5 v5.6.0
|
||||
github.com/prometheus/client_golang v1.24.1
|
||||
github.com/prometheus/client_model v0.6.2
|
||||
github.com/prometheus/common v0.70.1
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/beorn7/perks v1.0.1 // indirect
|
||||
github.com/cespare/xxhash/v2 v2.3.0 // indirect
|
||||
github.com/jackc/pgpassfile v1.0.0 // indirect
|
||||
github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
|
||||
github.com/jackc/puddle/v2 v2.2.1 // indirect
|
||||
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
|
||||
github.com/prometheus/procfs v0.21.1 // indirect
|
||||
golang.org/x/crypto v0.17.0 // indirect
|
||||
golang.org/x/sync v0.1.0 // indirect
|
||||
golang.org/x/text v0.14.0 // indirect
|
||||
golang.org/x/sync v0.22.0 // indirect
|
||||
golang.org/x/sys v0.47.0 // indirect
|
||||
golang.org/x/text v0.40.0 // indirect
|
||||
google.golang.org/protobuf v1.36.11 // indirect
|
||||
)
|
||||
|
||||
@@ -1,6 +1,12 @@
|
||||
github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
|
||||
github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
|
||||
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
|
||||
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
|
||||
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
|
||||
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
|
||||
github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
|
||||
github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
|
||||
github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
|
||||
github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a h1:bbPeKD0xmW/Y25WS6cokEszi5g+S0QxI/d45PkRi7Nk=
|
||||
@@ -9,19 +15,37 @@ github.com/jackc/pgx/v5 v5.6.0 h1:SWJzexBzPL5jb0GEsrPMLIsi/3jOo7RHlzTjcAeDrPY=
|
||||
github.com/jackc/pgx/v5 v5.6.0/go.mod h1:DNZ/vlrUnhWCoFGxHAG8U2ljioxukquj7utPDgtQdTw=
|
||||
github.com/jackc/puddle/v2 v2.2.1 h1:RhxXJtFG022u4ibrCSMSiu5aOq1i77R3OHKNJj77OAk=
|
||||
github.com/jackc/puddle/v2 v2.2.1/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
|
||||
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
|
||||
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
|
||||
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/prometheus/client_golang v1.24.1 h1:JnJkREXzWxUdCuPFpIWZiPispT9xVV59uiuyR2bPlnU=
|
||||
github.com/prometheus/client_golang v1.24.1/go.mod h1:F+oSRECHg4sse5ucfYpYDeIv/hu68Zo0uoHKetWnzcE=
|
||||
github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk=
|
||||
github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE=
|
||||
github.com/prometheus/common v0.70.1 h1:1HvjP4D5oL3t8RsPlwxA9onvvStjtIHYE5XuuwOi/PY=
|
||||
github.com/prometheus/common v0.70.1/go.mod h1:VdFUQDMZK3VLkurFUVhia6uys/0suUp86TJz5qbJRhc=
|
||||
github.com/prometheus/procfs v0.21.1 h1:GljZCt+zSTS+NZq88cyQ1LjZ+RCHp3uVuabBWA5+OJI=
|
||||
github.com/prometheus/procfs v0.21.1/go.mod h1:aB55Cww9pdSJVHk0hUf0inxWyyjPogFIjmHKYgMKmtY=
|
||||
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
|
||||
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
|
||||
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
|
||||
github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
|
||||
github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
|
||||
github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
|
||||
github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
|
||||
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
|
||||
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
|
||||
go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ=
|
||||
go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ=
|
||||
golang.org/x/crypto v0.17.0 h1:r8bRNjWL3GshPW3gkd+RpvzWrZAwPS49OmTGZ/uhM4k=
|
||||
golang.org/x/crypto v0.17.0/go.mod h1:gCAAfMLgwOJRpTjQ2zCCt2OcSfYMTeZVSRtQlPC7Nq4=
|
||||
golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o=
|
||||
golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
|
||||
golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
|
||||
golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek=
|
||||
golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
|
||||
golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
|
||||
golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
|
||||
golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs=
|
||||
golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY=
|
||||
google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
|
||||
google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
|
||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
|
||||
|
||||
@@ -0,0 +1,194 @@
|
||||
package alerting
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
dto "github.com/prometheus/client_model/go"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/notify"
|
||||
)
|
||||
|
||||
// DefaultDebounceInterval: wiederholte Alarmierung für denselben
|
||||
// anhaltenden Zustand ist gedrosselt (Akzeptanzkriterium 3) — 15 Minuten
|
||||
// ist ein üblicher Kompromiss zwischen "schnell genug informiert" und
|
||||
// "kein Alarm-Spam bei dauerhaft überschrittenem Wert".
|
||||
const DefaultDebounceInterval = 15 * time.Minute
|
||||
|
||||
// AlertChannel ist der CFG-02-Kanal, über den Schwellwert-Alarme zugestellt
|
||||
// werden — ein eigener Kanalname, damit Zustellregeln/-vorlagen (CFG-03)
|
||||
// unabhängig von anderen Benachrichtigungsarten konfiguriert werden können.
|
||||
const AlertChannel = "alert"
|
||||
|
||||
// Evaluator prüft konfigurierte Regeln gegen aktuell gesammelte Metriken
|
||||
// (aus internal/metrics.Aggregator.Gather) und löst bei Überschreitung eine
|
||||
// Benachrichtigung über CFG-02 aus (Akzeptanzkriterium 2), gedrosselt je
|
||||
// Regel+Zeitreihe (Akzeptanzkriterium 3).
|
||||
type Evaluator struct {
|
||||
rules *RuleStore
|
||||
debounce *debounceStore
|
||||
dispatcher *notify.Dispatcher
|
||||
interval time.Duration
|
||||
}
|
||||
|
||||
func NewEvaluator(rules *RuleStore, dispatcher *notify.Dispatcher, debouncePool *pgxpool.Pool, interval time.Duration) *Evaluator {
|
||||
if interval <= 0 {
|
||||
interval = DefaultDebounceInterval
|
||||
}
|
||||
return &Evaluator{
|
||||
rules: rules,
|
||||
debounce: &debounceStore{pool: debouncePool},
|
||||
dispatcher: dispatcher,
|
||||
interval: interval,
|
||||
}
|
||||
}
|
||||
|
||||
// FiredAlert beschreibt einen tatsächlich ausgelösten (nicht gedrosselten)
|
||||
// Alarm — fürs Testen/Logging, nicht Teil des öffentlichen Zustellwegs.
|
||||
type FiredAlert struct {
|
||||
RuleID string
|
||||
MetricName string
|
||||
Value float64
|
||||
Threshold float64
|
||||
Labels map[string]string
|
||||
Skipped bool // true, wenn wegen Drosselung NICHT tatsaechlich zugestellt
|
||||
}
|
||||
|
||||
// Evaluate prüft alle konfigurierten Regeln gegen families (Akzeptanzkriterium 1).
|
||||
// Für jede Zeitreihe, die eine Regel verletzt, wird — sofern nicht gedrosselt
|
||||
// — eine Benachrichtigung mit Metrik/Wert/Schwellwert/Labels (Tenant/Modul,
|
||||
// falls als Label vorhanden) über CFG-02 eingereiht (Akzeptanzkriterium 2).
|
||||
func (e *Evaluator) Evaluate(ctx context.Context, families []*dto.MetricFamily) ([]FiredAlert, error) {
|
||||
rules, err := e.rules.ListRules(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("regeln laden: %w", err)
|
||||
}
|
||||
if len(rules) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
byName := make(map[string]*dto.MetricFamily, len(families))
|
||||
for _, f := range families {
|
||||
if f.Name != nil {
|
||||
byName[*f.Name] = f
|
||||
}
|
||||
}
|
||||
|
||||
var fired []FiredAlert
|
||||
for _, rule := range rules {
|
||||
family, ok := byName[rule.MetricName]
|
||||
if !ok {
|
||||
continue // Metrik (noch) nicht vorhanden -> keine Aussage moeglich, kein Fehler.
|
||||
}
|
||||
|
||||
for _, m := range family.Metric {
|
||||
labels := labelMap(m)
|
||||
if !matchesFilter(labels, rule.LabelFilters) {
|
||||
continue
|
||||
}
|
||||
value, ok := metricValue(m)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
if !violates(rule, value) {
|
||||
continue
|
||||
}
|
||||
|
||||
ruleKey := ruleKeyFor(rule.ID, labels)
|
||||
allowed, err := e.debounce.shouldFire(ctx, ruleKey, e.interval.Seconds())
|
||||
if err != nil {
|
||||
return fired, fmt.Errorf("drosselung pruefen: %w", err)
|
||||
}
|
||||
|
||||
alert := FiredAlert{
|
||||
RuleID: rule.ID, MetricName: rule.MetricName, Value: value,
|
||||
Threshold: rule.Threshold, Labels: labels, Skipped: !allowed,
|
||||
}
|
||||
fired = append(fired, alert)
|
||||
|
||||
if !allowed {
|
||||
continue
|
||||
}
|
||||
|
||||
payload := map[string]any{
|
||||
"metric": rule.MetricName,
|
||||
"value": value,
|
||||
"threshold": rule.Threshold,
|
||||
"comparison": string(rule.Comparison),
|
||||
"description": rule.Description,
|
||||
"labels": labels,
|
||||
}
|
||||
if _, err := e.dispatcher.Enqueue(ctx, AlertChannel, rule.Recipient, payload); err != nil {
|
||||
return fired, fmt.Errorf("alarm einreihen: %w", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
return fired, nil
|
||||
}
|
||||
|
||||
func violates(rule Rule, value float64) bool {
|
||||
switch rule.Comparison {
|
||||
case ComparisonGreaterThan:
|
||||
return value > rule.Threshold
|
||||
case ComparisonLessThan:
|
||||
return value < rule.Threshold
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func labelMap(m *dto.Metric) map[string]string {
|
||||
out := make(map[string]string, len(m.Label))
|
||||
for _, l := range m.Label {
|
||||
if l.Name != nil && l.Value != nil {
|
||||
out[*l.Name] = *l.Value
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func matchesFilter(labels, filter map[string]string) bool {
|
||||
for k, v := range filter {
|
||||
if labels[k] != v {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func metricValue(m *dto.Metric) (float64, bool) {
|
||||
switch {
|
||||
case m.Gauge != nil && m.Gauge.Value != nil:
|
||||
return *m.Gauge.Value, true
|
||||
case m.Counter != nil && m.Counter.Value != nil:
|
||||
return *m.Counter.Value, true
|
||||
case m.Untyped != nil && m.Untyped.Value != nil:
|
||||
return *m.Untyped.Value, true
|
||||
default:
|
||||
return 0, false
|
||||
}
|
||||
}
|
||||
|
||||
// ruleKeyFor macht die Drosselung unabhaengig je Regel UND je konkreter
|
||||
// Zeitreihe (z. B. verschiedene Tenants/Module derselben Metrik loesen
|
||||
// unabhaengig voneinander aus, siehe Migrationskommentar).
|
||||
func ruleKeyFor(ruleID string, labels map[string]string) string {
|
||||
keys := make([]string, 0, len(labels))
|
||||
for k := range labels {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
var b strings.Builder
|
||||
b.WriteString(ruleID)
|
||||
for _, k := range keys {
|
||||
b.WriteString("|")
|
||||
b.WriteString(k)
|
||||
b.WriteString("=")
|
||||
b.WriteString(labels[k])
|
||||
}
|
||||
return b.String()
|
||||
}
|
||||
@@ -0,0 +1,233 @@
|
||||
package alerting
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
dto "github.com/prometheus/client_model/go"
|
||||
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/notify"
|
||||
)
|
||||
|
||||
func setupTest(t *testing.T) (*RuleStore, *notify.Dispatcher, *pgxpool.Pool, func()) {
|
||||
t.Helper()
|
||||
adminDSN := os.Getenv("TEST_ADMIN_DSN")
|
||||
if adminDSN == "" {
|
||||
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
|
||||
}
|
||||
ctx := context.Background()
|
||||
|
||||
pool, err := pgxpool.New(ctx, adminDSN)
|
||||
if err != nil {
|
||||
t.Fatalf("pool: %v", err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, `
|
||||
CREATE TABLE IF NOT EXISTS alert_rules (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(), metric_name TEXT NOT NULL,
|
||||
comparison TEXT NOT NULL CHECK (comparison IN ('gt','lt')), threshold DOUBLE PRECISION NOT NULL,
|
||||
label_filters JSONB NOT NULL DEFAULT '{}'::jsonb, recipient TEXT NOT NULL,
|
||||
description TEXT NOT NULL DEFAULT '', created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS alert_debounce_state (
|
||||
rule_key TEXT PRIMARY KEY, last_fired_at TIMESTAMPTZ NOT NULL
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS notification_jobs (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(), channel TEXT NOT NULL, recipient TEXT NOT NULL,
|
||||
payload JSONB NOT NULL DEFAULT '{}'::jsonb, status TEXT NOT NULL DEFAULT 'pending', attempts INT NOT NULL DEFAULT 0,
|
||||
max_attempts INT NOT NULL DEFAULT 5, next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(), last_error TEXT,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
`); err != nil {
|
||||
t.Fatalf("schema: %v", err)
|
||||
}
|
||||
|
||||
cleanup := func() {
|
||||
_, _ = pool.Exec(ctx, `DELETE FROM alert_rules`)
|
||||
_, _ = pool.Exec(ctx, `DELETE FROM alert_debounce_state`)
|
||||
_, _ = pool.Exec(ctx, `DELETE FROM notification_jobs`)
|
||||
pool.Close()
|
||||
}
|
||||
return NewRuleStore(pool), notify.NewDispatcher(pool), pool, cleanup
|
||||
}
|
||||
|
||||
func gaugeFamily(name string, labels map[string]string, value float64) *dto.MetricFamily {
|
||||
pairs := make([]*dto.LabelPair, 0, len(labels))
|
||||
for k, v := range labels {
|
||||
k, v := k, v
|
||||
pairs = append(pairs, &dto.LabelPair{Name: &k, Value: &v})
|
||||
}
|
||||
n := name
|
||||
return &dto.MetricFamily{
|
||||
Name: &n,
|
||||
Metric: []*dto.Metric{
|
||||
{Label: pairs, Gauge: &dto.Gauge{Value: &value}},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 1 + 2 / Pruefung 1 + 2: Überschreitung löst eine
|
||||
// Benachrichtigung mit vollständigem Inhalt (Metrik/Tenant/Modul) aus.
|
||||
func TestEvaluate_FiresAlertOnThresholdExceeded(t *testing.T) {
|
||||
rules, dispatcher, pool, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
rule, err := rules.CreateRule(ctx, Rule{
|
||||
MetricName: "nexarch_core_error_rate", Comparison: ComparisonGreaterThan, Threshold: 0.05,
|
||||
Recipient: "ops@acme.example", Description: "Fehlerrate zu hoch",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("create rule: %v", err)
|
||||
}
|
||||
|
||||
eval := NewEvaluator(rules, dispatcher, pool, time.Hour)
|
||||
families := []*dto.MetricFamily{
|
||||
gaugeFamily("nexarch_core_error_rate", map[string]string{"tenant": "acme", "module": "dms"}, 0.12),
|
||||
}
|
||||
|
||||
fired, err := eval.Evaluate(ctx, families)
|
||||
if err != nil {
|
||||
t.Fatalf("evaluate: %v", err)
|
||||
}
|
||||
if len(fired) != 1 || fired[0].Skipped {
|
||||
t.Fatalf("erwartet genau 1 tatsaechlich ausgeloesten alarm, habe %+v", fired)
|
||||
}
|
||||
if fired[0].RuleID != rule.ID {
|
||||
t.Fatalf("rule id = %q, want %q", fired[0].RuleID, rule.ID)
|
||||
}
|
||||
|
||||
// Pruefung 2: Benachrichtigungsinhalt vollstaendig (Metrik/Tenant/Modul).
|
||||
var payloadJSON []byte
|
||||
if err := pool.QueryRow(ctx, `SELECT payload FROM notification_jobs LIMIT 1`).Scan(&payloadJSON); err != nil {
|
||||
t.Fatalf("notification_jobs lesen: %v", err)
|
||||
}
|
||||
payload := string(payloadJSON)
|
||||
for _, want := range []string{`"metric"`, `nexarch_core_error_rate`, `"tenant"`, `"acme"`, `"module"`, `"dms"`} {
|
||||
if !strings.Contains(payload, want) {
|
||||
t.Errorf("payload enthaelt nicht %q: %s", want, payload)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestEvaluate_DoesNotFireBelowThreshold(t *testing.T) {
|
||||
rules, dispatcher, pool, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
if _, err := rules.CreateRule(ctx, Rule{
|
||||
MetricName: "nexarch_core_error_rate", Comparison: ComparisonGreaterThan, Threshold: 0.05,
|
||||
Recipient: "ops@acme.example",
|
||||
}); err != nil {
|
||||
t.Fatalf("create rule: %v", err)
|
||||
}
|
||||
|
||||
eval := NewEvaluator(rules, dispatcher, pool, time.Hour)
|
||||
families := []*dto.MetricFamily{
|
||||
gaugeFamily("nexarch_core_error_rate", map[string]string{"tenant": "acme"}, 0.01),
|
||||
}
|
||||
|
||||
fired, err := eval.Evaluate(ctx, families)
|
||||
if err != nil {
|
||||
t.Fatalf("evaluate: %v", err)
|
||||
}
|
||||
if len(fired) != 0 {
|
||||
t.Fatalf("erwartet keinen alarm unterhalb des schwellwerts, habe %+v", fired)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 3 / Pruefung 3: anhaltende Überschreitung erzeugt NICHT
|
||||
// bei jeder Messung eine neue Benachrichtigung.
|
||||
func TestEvaluate_DebouncesRepeatedFiring(t *testing.T) {
|
||||
rules, dispatcher, pool, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
if _, err := rules.CreateRule(ctx, Rule{
|
||||
MetricName: "nexarch_core_error_rate", Comparison: ComparisonGreaterThan, Threshold: 0.05,
|
||||
Recipient: "ops@acme.example",
|
||||
}); err != nil {
|
||||
t.Fatalf("create rule: %v", err)
|
||||
}
|
||||
|
||||
// Langes Debounce-Intervall: der zweite Evaluate-Lauf (simuliert die
|
||||
// naechste Messung bei anhaltend ueberschrittenem Wert) darf keinen
|
||||
// weiteren Job einreihen.
|
||||
eval := NewEvaluator(rules, dispatcher, pool, time.Hour)
|
||||
families := []*dto.MetricFamily{
|
||||
gaugeFamily("nexarch_core_error_rate", map[string]string{"tenant": "acme"}, 0.5),
|
||||
}
|
||||
|
||||
first, err := eval.Evaluate(ctx, families)
|
||||
if err != nil {
|
||||
t.Fatalf("erster evaluate-lauf: %v", err)
|
||||
}
|
||||
if len(first) != 1 || first[0].Skipped {
|
||||
t.Fatalf("erster lauf haette feuern muessen, habe %+v", first)
|
||||
}
|
||||
|
||||
second, err := eval.Evaluate(ctx, families)
|
||||
if err != nil {
|
||||
t.Fatalf("zweiter evaluate-lauf: %v", err)
|
||||
}
|
||||
if len(second) != 1 || !second[0].Skipped {
|
||||
t.Fatalf("zweiter lauf haette gedrosselt werden muessen, habe %+v", second)
|
||||
}
|
||||
|
||||
var count int
|
||||
if err := pool.QueryRow(ctx, `SELECT count(*) FROM notification_jobs`).Scan(&count); err != nil {
|
||||
t.Fatalf("notification_jobs zaehlen: %v", err)
|
||||
}
|
||||
if count != 1 {
|
||||
t.Fatalf("erwartet genau 1 eingereihten job trotz zwei ueberschreitenden messungen, habe %d", count)
|
||||
}
|
||||
}
|
||||
|
||||
// Verschiedene Zeitreihen derselben Regel (unterschiedlicher Tenant) werden
|
||||
// unabhaengig voneinander gedrosselt.
|
||||
func TestEvaluate_DebouncesIndependentlyPerLabelSet(t *testing.T) {
|
||||
rules, dispatcher, pool, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
if _, err := rules.CreateRule(ctx, Rule{
|
||||
MetricName: "nexarch_core_error_rate", Comparison: ComparisonGreaterThan, Threshold: 0.05,
|
||||
Recipient: "ops@acme.example",
|
||||
}); err != nil {
|
||||
t.Fatalf("create rule: %v", err)
|
||||
}
|
||||
|
||||
eval := NewEvaluator(rules, dispatcher, pool, time.Hour)
|
||||
families := []*dto.MetricFamily{
|
||||
{
|
||||
Name: strPtr("nexarch_core_error_rate"),
|
||||
Metric: []*dto.Metric{
|
||||
metricWithLabel("tenant", "acme", 0.5),
|
||||
metricWithLabel("tenant", "beta", 0.6),
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
fired, err := eval.Evaluate(ctx, families)
|
||||
if err != nil {
|
||||
t.Fatalf("evaluate: %v", err)
|
||||
}
|
||||
if len(fired) != 2 {
|
||||
t.Fatalf("erwartet 2 unabhaengige alarme (verschiedene tenants), habe %d", len(fired))
|
||||
}
|
||||
for _, a := range fired {
|
||||
if a.Skipped {
|
||||
t.Fatalf("beide tenants sollten beim ersten mal feuern, habe %+v", a)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func metricWithLabel(name, value string, gaugeValue float64) *dto.Metric {
|
||||
n, v := name, value
|
||||
return &dto.Metric{Label: []*dto.LabelPair{{Name: &n, Value: &v}}, Gauge: &dto.Gauge{Value: &gaugeValue}}
|
||||
}
|
||||
|
||||
func strPtr(s string) *string { return &s }
|
||||
@@ -0,0 +1,132 @@
|
||||
// Package alerting implementiert Core OPS-05: schwellwertbasierte
|
||||
// Alarmierung auf den aus OPS-03 aggregierten Metriken, Zustellung über den
|
||||
// Core-Benachrichtigungs-Dispatcher (CFG-02). Der Alertmanager-Gedanke von
|
||||
// Prometheus/Grafana, aber auf das Nötigste reduziert (Schwellwert, Ziel,
|
||||
// Drosselung) — keine eigene Ausdruckssprache.
|
||||
package alerting
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// Comparison legt fest, ob ein Schwellwert nach oben oder unten überwacht
|
||||
// wird — bewusst nur zwei Operatoren, keine eigene Ausdruckssprache
|
||||
// (Ticket-Produkt-DNA).
|
||||
type Comparison string
|
||||
|
||||
const (
|
||||
ComparisonGreaterThan Comparison = "gt"
|
||||
ComparisonLessThan Comparison = "lt"
|
||||
)
|
||||
|
||||
// Rule ist eine Schwellwert-Regel auf einer beliebigen aggregierten Metrik
|
||||
// (Akzeptanzkriterium 1). LabelFilters schränkt optional auf bestimmte
|
||||
// Label-Werte ein (z. B. tenant/module), leer = alle Zeitreihen der Metrik.
|
||||
type Rule struct {
|
||||
ID string
|
||||
MetricName string
|
||||
Comparison Comparison
|
||||
Threshold float64
|
||||
LabelFilters map[string]string
|
||||
Recipient string
|
||||
Description string
|
||||
}
|
||||
|
||||
// RuleStore verwaltet Alert-Regeln in der zentralen Registry-DB.
|
||||
type RuleStore struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewRuleStore(pool *pgxpool.Pool) *RuleStore {
|
||||
return &RuleStore{pool: pool}
|
||||
}
|
||||
|
||||
// CreateRule legt eine neue Schwellwert-Regel an (Akzeptanzkriterium 1).
|
||||
func (s *RuleStore) CreateRule(ctx context.Context, r Rule) (Rule, error) {
|
||||
if r.MetricName == "" || r.Recipient == "" {
|
||||
return Rule{}, errors.New("alerting: metricName und recipient duerfen nicht leer sein")
|
||||
}
|
||||
if r.Comparison != ComparisonGreaterThan && r.Comparison != ComparisonLessThan {
|
||||
return Rule{}, fmt.Errorf("alerting: unbekannter comparison-operator %q", r.Comparison)
|
||||
}
|
||||
if r.LabelFilters == nil {
|
||||
r.LabelFilters = map[string]string{}
|
||||
}
|
||||
filtersJSON, err := json.Marshal(r.LabelFilters)
|
||||
if err != nil {
|
||||
return Rule{}, fmt.Errorf("label-filter serialisieren: %w", err)
|
||||
}
|
||||
|
||||
err = s.pool.QueryRow(ctx, `
|
||||
INSERT INTO alert_rules (metric_name, comparison, threshold, label_filters, recipient, description)
|
||||
VALUES ($1, $2, $3, $4, $5, $6)
|
||||
RETURNING id
|
||||
`, r.MetricName, string(r.Comparison), r.Threshold, filtersJSON, r.Recipient, r.Description).Scan(&r.ID)
|
||||
if err != nil {
|
||||
return Rule{}, fmt.Errorf("regel speichern: %w", err)
|
||||
}
|
||||
return r, nil
|
||||
}
|
||||
|
||||
// ListRules liefert alle konfigurierten Regeln — Grundlage für Evaluate.
|
||||
func (s *RuleStore) ListRules(ctx context.Context) ([]Rule, error) {
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT id, metric_name, comparison, threshold, label_filters, recipient, description
|
||||
FROM alert_rules ORDER BY created_at
|
||||
`)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("regeln auflisten: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var out []Rule
|
||||
for rows.Next() {
|
||||
var r Rule
|
||||
var comparison string
|
||||
var filtersJSON []byte
|
||||
if err := rows.Scan(&r.ID, &r.MetricName, &comparison, &r.Threshold, &filtersJSON, &r.Recipient, &r.Description); err != nil {
|
||||
return nil, fmt.Errorf("regel lesen: %w", err)
|
||||
}
|
||||
r.Comparison = Comparison(comparison)
|
||||
if err := json.Unmarshal(filtersJSON, &r.LabelFilters); err != nil {
|
||||
return nil, fmt.Errorf("label-filter lesen: %w", err)
|
||||
}
|
||||
out = append(out, r)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// DeleteRule entfernt eine Regel.
|
||||
func (s *RuleStore) DeleteRule(ctx context.Context, id string) error {
|
||||
_, err := s.pool.Exec(ctx, `DELETE FROM alert_rules WHERE id = $1`, id)
|
||||
if err != nil {
|
||||
return fmt.Errorf("regel loeschen: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// debounceStore kapselt die Drosselungs-Zustandstabelle (Akzeptanzkriterium 3).
|
||||
type debounceStore struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
// shouldFire prueft, ob seit dem letzten Alarm fuer ruleKey mindestens
|
||||
// interval vergangen ist — atomar ueber eine bedingte UPDATE/INSERT-
|
||||
// Sequenz, damit zwei gleichzeitige Evaluate-Laeufe (z. B. bei mehreren
|
||||
// Core-Instanzen) nicht beide gleichzeitig alarmieren.
|
||||
func (d *debounceStore) shouldFire(ctx context.Context, ruleKey string, intervalSeconds float64) (bool, error) {
|
||||
tag, err := d.pool.Exec(ctx, `
|
||||
INSERT INTO alert_debounce_state (rule_key, last_fired_at) VALUES ($1, now())
|
||||
ON CONFLICT (rule_key) DO UPDATE SET last_fired_at = now()
|
||||
WHERE alert_debounce_state.last_fired_at <= now() - ($2 * interval '1 second')
|
||||
`, ruleKey, intervalSeconds)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("drosselungszustand pruefen: %w", err)
|
||||
}
|
||||
return tag.RowsAffected() == 1, nil
|
||||
}
|
||||
@@ -0,0 +1,88 @@
|
||||
package cfgservice
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// DefaultCacheTTL ist die dokumentierte Cache-Invalidierungszeit
|
||||
// (Akzeptanzkriterium 2 / Pruefung 2 in diesem Ticket bezieht sich auf die
|
||||
// Aenderungsnachvollziehbarkeit — die Cache-Frist selbst folgt demselben
|
||||
// Muster wie internal/flag.DefaultCacheTTL).
|
||||
const DefaultCacheTTL = 5 * time.Second
|
||||
|
||||
type cacheEntry struct {
|
||||
value Value
|
||||
expiresAt time.Time
|
||||
}
|
||||
|
||||
// Service ist die Leseseite mit Vorrangregel (Akzeptanzkriterium 1:
|
||||
// Tenant-Override vor Global-Default) und lokalem TTL-Cache.
|
||||
type Service struct {
|
||||
store *Store
|
||||
ttl time.Duration
|
||||
|
||||
mu sync.RWMutex
|
||||
cache map[string]cacheEntry // Schluessel: key + "\x00" + tenantSlug
|
||||
}
|
||||
|
||||
func NewService(store *Store, ttl time.Duration) *Service {
|
||||
if ttl <= 0 {
|
||||
ttl = DefaultCacheTTL
|
||||
}
|
||||
return &Service{store: store, ttl: ttl, cache: make(map[string]cacheEntry)}
|
||||
}
|
||||
|
||||
func cacheKey(key, tenantSlug string) string {
|
||||
return key + "\x00" + tenantSlug
|
||||
}
|
||||
|
||||
// Resolve liefert den Konfigurationswert fuer einen Tenant: ein
|
||||
// Tenant-spezifischer Override hat Vorrang vor dem globalen Default
|
||||
// (Akzeptanzkriterium 1 / Pruefung 1). tenantSlug == "" wertet nur den
|
||||
// globalen Wert aus.
|
||||
func (s *Service) Resolve(ctx context.Context, tenantSlug, key string) (Value, error) {
|
||||
ck := cacheKey(key, tenantSlug)
|
||||
|
||||
s.mu.RLock()
|
||||
entry, exists := s.cache[ck]
|
||||
fresh := exists && time.Now().Before(entry.expiresAt)
|
||||
s.mu.RUnlock()
|
||||
if fresh {
|
||||
return entry.value, nil
|
||||
}
|
||||
|
||||
v, err := s.resolveUncached(ctx, tenantSlug, key)
|
||||
if err != nil {
|
||||
return Value{}, err
|
||||
}
|
||||
|
||||
s.mu.Lock()
|
||||
s.cache[ck] = cacheEntry{value: v, expiresAt: time.Now().Add(s.ttl)}
|
||||
s.mu.Unlock()
|
||||
return v, nil
|
||||
}
|
||||
|
||||
func (s *Service) resolveUncached(ctx context.Context, tenantSlug, key string) (Value, error) {
|
||||
if tenantSlug != "" {
|
||||
v, err := s.store.Get(ctx, key, tenantSlug)
|
||||
if err == nil {
|
||||
return v, nil
|
||||
}
|
||||
if !errors.Is(err, ErrNotFound) {
|
||||
return Value{}, err
|
||||
}
|
||||
}
|
||||
return s.store.Get(ctx, key, GlobalScope)
|
||||
}
|
||||
|
||||
// Invalidate erzwingt beim naechsten Resolve-Aufruf ein sofortiges Neuladen
|
||||
// fuer einen bestimmten (key, tenantSlug) statt auf den TTL-Ablauf zu warten
|
||||
// — analog internal/flag.Service.Invalidate.
|
||||
func (s *Service) Invalidate(key, tenantSlug string) {
|
||||
s.mu.Lock()
|
||||
delete(s.cache, cacheKey(key, tenantSlug))
|
||||
s.mu.Unlock()
|
||||
}
|
||||
@@ -0,0 +1,114 @@
|
||||
package cfgservice
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Akzeptanzkriterium 1 + Pruefung 1: Tenant-Override hat Vorrang vor
|
||||
// Global-Default, automatisiert getestet.
|
||||
func TestService_TenantOverrideTakesPrecedenceOverGlobal(t *testing.T) {
|
||||
store, cleanup := setupStoreTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
if _, err := store.Set(ctx, "test_precedence_key", GlobalScope, "global-wert"); err != nil {
|
||||
t.Fatalf("set global: %v", err)
|
||||
}
|
||||
if _, err := store.Set(ctx, "test_precedence_key", "test_acme", "tenant-wert"); err != nil {
|
||||
t.Fatalf("set tenant: %v", err)
|
||||
}
|
||||
|
||||
svc := NewService(store, time.Hour)
|
||||
|
||||
got, err := svc.Resolve(ctx, "test_acme", "test_precedence_key")
|
||||
if err != nil {
|
||||
t.Fatalf("resolve mit override: %v", err)
|
||||
}
|
||||
if got.Value != "tenant-wert" {
|
||||
t.Fatalf("erwartet tenant-override, habe %q", got.Value)
|
||||
}
|
||||
|
||||
gotOther, err := svc.Resolve(ctx, "test_anderer_tenant", "test_precedence_key")
|
||||
if err != nil {
|
||||
t.Fatalf("resolve ohne override: %v", err)
|
||||
}
|
||||
if gotOther.Value != "global-wert" {
|
||||
t.Fatalf("erwartet global-default fuer tenant ohne override, habe %q", gotOther.Value)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2 + Pruefung 2: Cache-Invalidierung nach
|
||||
// Konfigurationsaenderung innerhalb dokumentierter Zeit gemessen.
|
||||
func TestService_CacheInvalidationTiming(t *testing.T) {
|
||||
store, cleanup := setupStoreTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
const ttl = 150 * time.Millisecond
|
||||
if _, err := store.Set(ctx, "test_ttl_key", GlobalScope, "alt"); err != nil {
|
||||
t.Fatalf("set: %v", err)
|
||||
}
|
||||
svc := NewService(store, ttl)
|
||||
|
||||
v, err := svc.Resolve(ctx, "", "test_ttl_key")
|
||||
if err != nil {
|
||||
t.Fatalf("resolve: %v", err)
|
||||
}
|
||||
if v.Value != "alt" {
|
||||
t.Fatalf("erwartet 'alt', habe %q", v.Value)
|
||||
}
|
||||
|
||||
changedAt := time.Now()
|
||||
if _, err := store.Set(ctx, "test_ttl_key", GlobalScope, "neu"); err != nil {
|
||||
t.Fatalf("set: %v", err)
|
||||
}
|
||||
|
||||
v, err = svc.Resolve(ctx, "", "test_ttl_key")
|
||||
if err != nil {
|
||||
t.Fatalf("resolve direkt nach aenderung: %v", err)
|
||||
}
|
||||
if v.Value != "alt" {
|
||||
t.Fatalf("cache haette den alten wert liefern sollen, habe %q", v.Value)
|
||||
}
|
||||
|
||||
deadline := changedAt.Add(ttl + 100*time.Millisecond)
|
||||
for time.Now().Before(deadline) {
|
||||
v, err := svc.Resolve(ctx, "", "test_ttl_key")
|
||||
if err != nil {
|
||||
t.Fatalf("resolve: %v", err)
|
||||
}
|
||||
if v.Value == "neu" {
|
||||
t.Logf("aenderung wurde nach %s wirksam (ziel: innerhalb %s + toleranz)", time.Since(changedAt), ttl)
|
||||
return
|
||||
}
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
t.Fatalf("aenderung wurde nicht innerhalb von %s wirksam", deadline.Sub(changedAt))
|
||||
}
|
||||
|
||||
func TestService_InvalidateForcesImmediateRefresh(t *testing.T) {
|
||||
store, cleanup := setupStoreTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
if _, err := store.Set(ctx, "test_invalidate_key", GlobalScope, "alt"); err != nil {
|
||||
t.Fatalf("set: %v", err)
|
||||
}
|
||||
svc := NewService(store, time.Hour)
|
||||
_, _ = svc.Resolve(ctx, "", "test_invalidate_key")
|
||||
|
||||
if _, err := store.Set(ctx, "test_invalidate_key", GlobalScope, "neu"); err != nil {
|
||||
t.Fatalf("set: %v", err)
|
||||
}
|
||||
svc.Invalidate("test_invalidate_key", "")
|
||||
|
||||
v, err := svc.Resolve(ctx, "", "test_invalidate_key")
|
||||
if err != nil {
|
||||
t.Fatalf("resolve: %v", err)
|
||||
}
|
||||
if v.Value != "neu" {
|
||||
t.Fatalf("erwartet sofort sichtbaren neuen wert nach Invalidate, habe %q", v.Value)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,129 @@
|
||||
// Package cfgservice implementiert Core CFG-01: den zentralen Dienst fuer
|
||||
// globale und tenant-spezifische Konfigurationswerte mit Versionierung und
|
||||
// Cache-Invalidierung. Andere Module lesen Konfiguration AUSSCHLIESSLICH
|
||||
// ueber dieses Paket (Akzeptanzkriterium 3), niemals ueber eigene Tabellen.
|
||||
package cfgservice
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// GlobalScope ist der reservierte Scope-Wert fuer globale Defaults — jeder
|
||||
// andere Scope-Wert ist ein Tenant-Slug (Akzeptanzkriterium 1).
|
||||
const GlobalScope = "global"
|
||||
|
||||
var ErrNotFound = errors.New("cfgservice: kein wert fuer diesen key gefunden")
|
||||
|
||||
type Value struct {
|
||||
Key string
|
||||
Scope string
|
||||
Value string
|
||||
Version int
|
||||
}
|
||||
|
||||
type HistoryEntry struct {
|
||||
Key string
|
||||
Scope string
|
||||
Value string
|
||||
Version int
|
||||
}
|
||||
|
||||
// Store ist die Schreib-/Verwaltungsseite. Set schreibt IMMER sowohl den
|
||||
// aktuellen Stand (config_values) als auch einen Historieneintrag
|
||||
// (config_value_history) in derselben Transaktion — eine Aenderung ohne
|
||||
// Versionshistorie ist strukturell ausgeschlossen (Akzeptanzkriterium 2).
|
||||
type Store struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewStore(pool *pgxpool.Pool) *Store {
|
||||
return &Store{pool: pool}
|
||||
}
|
||||
|
||||
// Set schreibt einen neuen Wert fuer (key, scope) und erhoeht die Version um 1
|
||||
// (Version 1 bei erstmaligem Setzen).
|
||||
func (s *Store) Set(ctx context.Context, key, scope, value string) (Value, error) {
|
||||
if scope == "" {
|
||||
return Value{}, errors.New("cfgservice: scope darf nicht leer sein")
|
||||
}
|
||||
|
||||
tx, err := s.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return Value{}, fmt.Errorf("transaktion starten: %w", err)
|
||||
}
|
||||
defer func() { _ = tx.Rollback(ctx) }()
|
||||
|
||||
var currentVersion int
|
||||
err = tx.QueryRow(ctx, `SELECT version FROM config_values WHERE key = $1 AND scope = $2`, key, scope).Scan(¤tVersion)
|
||||
if err != nil && !errors.Is(err, pgx.ErrNoRows) {
|
||||
return Value{}, fmt.Errorf("aktuelle version lesen: %w", err)
|
||||
}
|
||||
newVersion := currentVersion + 1
|
||||
|
||||
if _, err := tx.Exec(ctx, `
|
||||
INSERT INTO config_values (key, scope, value, version, updated_at)
|
||||
VALUES ($1, $2, $3, $4, now())
|
||||
ON CONFLICT (key, scope) DO UPDATE SET value = $3, version = $4, updated_at = now()
|
||||
`, key, scope, value, newVersion); err != nil {
|
||||
return Value{}, fmt.Errorf("wert speichern: %w", err)
|
||||
}
|
||||
|
||||
if _, err := tx.Exec(ctx, `
|
||||
INSERT INTO config_value_history (key, scope, value, version, changed_at)
|
||||
VALUES ($1, $2, $3, $4, now())
|
||||
`, key, scope, value, newVersion); err != nil {
|
||||
return Value{}, fmt.Errorf("historie schreiben: %w", err)
|
||||
}
|
||||
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return Value{}, fmt.Errorf("transaktion committen: %w", err)
|
||||
}
|
||||
|
||||
return Value{Key: key, Scope: scope, Value: value, Version: newVersion}, nil
|
||||
}
|
||||
|
||||
// Get liefert den Wert fuer GENAU EINEN Scope (kein Vorrang-Fallback) — die
|
||||
// Vorrangregel (Tenant vor Global) lebt bewusst in Service.Resolve, damit
|
||||
// Store rein CRUD bleibt.
|
||||
func (s *Store) Get(ctx context.Context, key, scope string) (Value, error) {
|
||||
var v Value
|
||||
v.Key, v.Scope = key, scope
|
||||
err := s.pool.QueryRow(ctx, `
|
||||
SELECT value, version FROM config_values WHERE key = $1 AND scope = $2
|
||||
`, key, scope).Scan(&v.Value, &v.Version)
|
||||
if err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return Value{}, ErrNotFound
|
||||
}
|
||||
return Value{}, fmt.Errorf("wert lesen: %w", err)
|
||||
}
|
||||
return v, nil
|
||||
}
|
||||
|
||||
// History liefert die vollstaendige Versionshistorie eines (key, scope) in
|
||||
// aufsteigender Reihenfolge (Akzeptanzkriterium 2 / Pruefung 3).
|
||||
func (s *Store) History(ctx context.Context, key, scope string) ([]HistoryEntry, error) {
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT key, scope, value, version FROM config_value_history
|
||||
WHERE key = $1 AND scope = $2 ORDER BY version
|
||||
`, key, scope)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("historie abfragen: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var out []HistoryEntry
|
||||
for rows.Next() {
|
||||
var h HistoryEntry
|
||||
if err := rows.Scan(&h.Key, &h.Scope, &h.Value, &h.Version); err != nil {
|
||||
return nil, fmt.Errorf("historieneintrag lesen: %w", err)
|
||||
}
|
||||
out = append(out, h)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
@@ -0,0 +1,119 @@
|
||||
package cfgservice
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
func setupStoreTest(t *testing.T) (*Store, func()) {
|
||||
t.Helper()
|
||||
adminDSN := os.Getenv("TEST_ADMIN_DSN")
|
||||
if adminDSN == "" {
|
||||
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
|
||||
}
|
||||
ctx := context.Background()
|
||||
|
||||
pool, err := pgxpool.New(ctx, adminDSN)
|
||||
if err != nil {
|
||||
t.Fatalf("pool: %v", err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, `
|
||||
CREATE TABLE IF NOT EXISTS config_values (
|
||||
key TEXT NOT NULL,
|
||||
scope TEXT NOT NULL CHECK (scope <> ''),
|
||||
value TEXT NOT NULL,
|
||||
version INT NOT NULL,
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
PRIMARY KEY (key, scope)
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS config_value_history (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
key TEXT NOT NULL,
|
||||
scope TEXT NOT NULL,
|
||||
value TEXT NOT NULL,
|
||||
version INT NOT NULL,
|
||||
changed_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
`); err != nil {
|
||||
t.Fatalf("schema: %v", err)
|
||||
}
|
||||
|
||||
cleanup := func() {
|
||||
_, _ = pool.Exec(ctx, `DELETE FROM config_value_history WHERE key LIKE 'test\_%' ESCAPE '\'`)
|
||||
_, _ = pool.Exec(ctx, `DELETE FROM config_values WHERE key LIKE 'test\_%' ESCAPE '\'`)
|
||||
pool.Close()
|
||||
}
|
||||
return NewStore(pool), cleanup
|
||||
}
|
||||
|
||||
func TestStore_SetIncrementsVersion(t *testing.T) {
|
||||
store, cleanup := setupStoreTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
v1, err := store.Set(ctx, "test_key", GlobalScope, "erster-wert")
|
||||
if err != nil {
|
||||
t.Fatalf("set 1: %v", err)
|
||||
}
|
||||
if v1.Version != 1 {
|
||||
t.Fatalf("erwartet version 1, habe %d", v1.Version)
|
||||
}
|
||||
|
||||
v2, err := store.Set(ctx, "test_key", GlobalScope, "zweiter-wert")
|
||||
if err != nil {
|
||||
t.Fatalf("set 2: %v", err)
|
||||
}
|
||||
if v2.Version != 2 {
|
||||
t.Fatalf("erwartet version 2, habe %d", v2.Version)
|
||||
}
|
||||
|
||||
got, err := store.Get(ctx, "test_key", GlobalScope)
|
||||
if err != nil {
|
||||
t.Fatalf("get: %v", err)
|
||||
}
|
||||
if got.Value != "zweiter-wert" || got.Version != 2 {
|
||||
t.Fatalf("aktueller wert unerwartet: %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2 + Pruefung 3: Versionierungshistorie ueber mehrere
|
||||
// Aenderungen hinweg nachvollzogen.
|
||||
func TestStore_HistoryTracksAllChanges(t *testing.T) {
|
||||
store, cleanup := setupStoreTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
values := []string{"v1", "v2", "v3"}
|
||||
for _, v := range values {
|
||||
if _, err := store.Set(ctx, "test_history_key", GlobalScope, v); err != nil {
|
||||
t.Fatalf("set %q: %v", v, err)
|
||||
}
|
||||
}
|
||||
|
||||
history, err := store.History(ctx, "test_history_key", GlobalScope)
|
||||
if err != nil {
|
||||
t.Fatalf("history: %v", err)
|
||||
}
|
||||
if len(history) != 3 {
|
||||
t.Fatalf("erwartet 3 historieneintraege, habe %d", len(history))
|
||||
}
|
||||
for i, h := range history {
|
||||
if h.Version != i+1 || h.Value != values[i] {
|
||||
t.Fatalf("historieneintrag[%d] unerwartet: %+v", i, h)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestStore_GetUnknownKeyReturnsNotFound(t *testing.T) {
|
||||
store, cleanup := setupStoreTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
if _, err := store.Get(ctx, "test_nie_gesetzt", GlobalScope); !errors.Is(err, ErrNotFound) {
|
||||
t.Fatalf("erwartet ErrNotFound, habe %v", err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,25 @@
|
||||
package health
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// DatabaseChecker prueft die tatsaechliche Erreichbarkeit der Datenbank
|
||||
// (Ping) — nicht nur, ob der Pool existiert.
|
||||
func DatabaseChecker(pool *pgxpool.Pool) CheckerFunc {
|
||||
return func(ctx context.Context) error {
|
||||
return pool.Ping(ctx)
|
||||
}
|
||||
}
|
||||
|
||||
// QueueChecker prueft, dass die Postgres-basierte Job-Queue (siehe CFG-02)
|
||||
// tatsaechlich abfragbar ist — eine eigene, benannte Abhaengigkeit neben der
|
||||
// reinen DB-Erreichbarkeit (Akzeptanzkriterium 1).
|
||||
func QueueChecker(pool *pgxpool.Pool) CheckerFunc {
|
||||
return func(ctx context.Context) error {
|
||||
_, err := pool.Exec(ctx, `SELECT 1`)
|
||||
return err
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,49 @@
|
||||
package health
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
)
|
||||
|
||||
// LivenessHandler beantwortet IMMER "lebt", solange der Prozess ueberhaupt
|
||||
// HTTP-Anfragen verarbeiten kann — prueft bewusst KEINE externen
|
||||
// Abhaengigkeiten (Akzeptanzkriterium 2: Liveness und Readiness getrennt).
|
||||
// Ein Datenbankausfall darf die Liveness nicht auf "tot" setzen, sonst
|
||||
// wuerde eine Orchestrierung (z.B. systemd/Kubernetes) den Prozess grundlos
|
||||
// neu starten, obwohl nur eine Abhaengigkeit ausgefallen ist.
|
||||
func LivenessHandler(w http.ResponseWriter, r *http.Request) {
|
||||
writeStatus(w, http.StatusOK, map[string]any{"status": "alive"})
|
||||
}
|
||||
|
||||
// ReadinessHandler prueft ALLE registrierten Abhaengigkeiten
|
||||
// (Akzeptanzkriterium 1) und liefert 503, sobald eine davon fehlschlaegt
|
||||
// (Akzeptanzkriterium 3) — unterscheidet sich damit nachweislich von
|
||||
// LivenessHandler im Fehlerfall (Akzeptanzkriterium 2 / Pruefung 3).
|
||||
func (r *Registry) ReadinessHandler() http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, req *http.Request) {
|
||||
ready, results := r.CheckAll(req.Context())
|
||||
|
||||
body := map[string]any{
|
||||
"status": statusText(ready),
|
||||
"checks": results,
|
||||
}
|
||||
status := http.StatusOK
|
||||
if !ready {
|
||||
status = http.StatusServiceUnavailable
|
||||
}
|
||||
writeStatus(w, status, body)
|
||||
}
|
||||
}
|
||||
|
||||
func statusText(ready bool) string {
|
||||
if ready {
|
||||
return "ready"
|
||||
}
|
||||
return "not_ready"
|
||||
}
|
||||
|
||||
func writeStatus(w http.ResponseWriter, status int, body map[string]any) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(status)
|
||||
_ = json.NewEncoder(w).Encode(body)
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
// Package health implementiert Core OPS-01: Health-/Readiness-Endpunkte, die
|
||||
// echte Abhaengigkeiten (DB, Job-Queue) statt nur den Prozessstatus pruefen
|
||||
// — wiederverwendbar von Core UND jedem registrierten Modul (siehe API-02),
|
||||
// nicht nur von Core selbst.
|
||||
package health
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Checker prueft EINE Abhaengigkeit (z.B. Datenbank, Job-Queue).
|
||||
type Checker interface {
|
||||
Check(ctx context.Context) error
|
||||
}
|
||||
|
||||
type CheckerFunc func(ctx context.Context) error
|
||||
|
||||
func (f CheckerFunc) Check(ctx context.Context) error { return f(ctx) }
|
||||
|
||||
// DefaultCheckTimeout begrenzt, wie lange EIN einzelner Check maximal
|
||||
// dauern darf, bevor er als fehlgeschlagen gilt — verhindert, dass ein
|
||||
// haengender Check den gesamten Readiness-Endpunkt blockiert
|
||||
// (Akzeptanzkriterium 2 / Pruefung 2: Antwort innerhalb definierter Zeit).
|
||||
const DefaultCheckTimeout = 2 * time.Second
|
||||
|
||||
// Registry haelt alle benannten Checks eines Dienstes.
|
||||
type Registry struct {
|
||||
checks map[string]Checker
|
||||
timeout time.Duration
|
||||
}
|
||||
|
||||
func NewRegistry() *Registry {
|
||||
return &Registry{checks: make(map[string]Checker), timeout: DefaultCheckTimeout}
|
||||
}
|
||||
|
||||
func (r *Registry) WithTimeout(d time.Duration) *Registry {
|
||||
return &Registry{checks: r.checks, timeout: d}
|
||||
}
|
||||
|
||||
// Register fuegt einen benannten Check hinzu (z.B. "database", "queue").
|
||||
func (r *Registry) Register(name string, c Checker) {
|
||||
r.checks[name] = c
|
||||
}
|
||||
|
||||
// Result ist der Ausgang eines einzelnen Checks.
|
||||
type Result struct {
|
||||
OK bool
|
||||
Error string
|
||||
}
|
||||
|
||||
// CheckAll fuehrt alle registrierten Checks NEBENLAEUFIG mit je eigenem
|
||||
// Timeout aus (Akzeptanzkriterium 1: echte Abhaengigkeiten statt Prozess-
|
||||
// status) und liefert ready=false, sobald irgendein Check fehlschlaegt
|
||||
// (Akzeptanzkriterium 3: ein Ausfall wird sichtbar).
|
||||
func (r *Registry) CheckAll(ctx context.Context) (ready bool, results map[string]Result) {
|
||||
type namedResult struct {
|
||||
name string
|
||||
result Result
|
||||
}
|
||||
ch := make(chan namedResult, len(r.checks))
|
||||
|
||||
for name, checker := range r.checks {
|
||||
go func(name string, checker Checker) {
|
||||
checkCtx, cancel := context.WithTimeout(ctx, r.timeout)
|
||||
defer cancel()
|
||||
err := checker.Check(checkCtx)
|
||||
if err != nil {
|
||||
ch <- namedResult{name, Result{OK: false, Error: err.Error()}}
|
||||
return
|
||||
}
|
||||
ch <- namedResult{name, Result{OK: true}}
|
||||
}(name, checker)
|
||||
}
|
||||
|
||||
results = make(map[string]Result, len(r.checks))
|
||||
ready = true
|
||||
for i := 0; i < len(r.checks); i++ {
|
||||
nr := <-ch
|
||||
results[nr.name] = nr.result
|
||||
if !nr.result.OK {
|
||||
ready = false
|
||||
}
|
||||
}
|
||||
return ready, results
|
||||
}
|
||||
@@ -0,0 +1,161 @@
|
||||
package health
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// Akzeptanzkriterium 1 + Pruefung 1: simulierter Datenbankausfall fuehrt zu
|
||||
// "nicht bereit".
|
||||
func TestReadinessHandler_ReportsNotReadyOnDatabaseFailure(t *testing.T) {
|
||||
adminDSN := os.Getenv("TEST_ADMIN_DSN")
|
||||
if adminDSN == "" {
|
||||
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
|
||||
}
|
||||
ctx := context.Background()
|
||||
|
||||
pool, err := pgxpool.New(ctx, adminDSN)
|
||||
if err != nil {
|
||||
t.Fatalf("pool: %v", err)
|
||||
}
|
||||
// Datenbankausfall simulieren: Pool sofort schliessen, bevor der Check laeuft.
|
||||
pool.Close()
|
||||
|
||||
reg := NewRegistry()
|
||||
reg.Register("database", DatabaseChecker(pool))
|
||||
|
||||
req := httptest.NewRequest(http.MethodGet, "/readyz", nil)
|
||||
rec := httptest.NewRecorder()
|
||||
reg.ReadinessHandler()(rec, req)
|
||||
|
||||
if rec.Code != http.StatusServiceUnavailable {
|
||||
t.Fatalf("status = %d, want 503 bei db-ausfall", rec.Code)
|
||||
}
|
||||
|
||||
var body struct {
|
||||
Status string `json:"status"`
|
||||
Checks map[string]interface{} `json:"checks"`
|
||||
}
|
||||
if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
|
||||
t.Fatalf("body parsen: %v", err)
|
||||
}
|
||||
if body.Status != "not_ready" {
|
||||
t.Fatalf("status-feld = %q, want not_ready", body.Status)
|
||||
}
|
||||
if _, ok := body.Checks["database"]; !ok {
|
||||
t.Fatal("erwartet 'database' im checks-ergebnis")
|
||||
}
|
||||
}
|
||||
|
||||
func TestReadinessHandler_ReportsReadyWhenAllChecksPass(t *testing.T) {
|
||||
adminDSN := os.Getenv("TEST_ADMIN_DSN")
|
||||
if adminDSN == "" {
|
||||
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
|
||||
}
|
||||
ctx := context.Background()
|
||||
|
||||
pool, err := pgxpool.New(ctx, adminDSN)
|
||||
if err != nil {
|
||||
t.Fatalf("pool: %v", err)
|
||||
}
|
||||
defer pool.Close()
|
||||
|
||||
reg := NewRegistry()
|
||||
reg.Register("database", DatabaseChecker(pool))
|
||||
reg.Register("queue", QueueChecker(pool))
|
||||
|
||||
req := httptest.NewRequest(http.MethodGet, "/readyz", nil)
|
||||
rec := httptest.NewRecorder()
|
||||
reg.ReadinessHandler()(rec, req)
|
||||
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200 bei funktionierenden abhaengigkeiten", rec.Code)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2 + Pruefung 3: Liveness und Readiness unterscheiden
|
||||
// sich nachweislich im Fehlerfall.
|
||||
func TestLivenessAndReadiness_DifferOnDatabaseFailure(t *testing.T) {
|
||||
adminDSN := os.Getenv("TEST_ADMIN_DSN")
|
||||
if adminDSN == "" {
|
||||
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
|
||||
}
|
||||
ctx := context.Background()
|
||||
|
||||
pool, err := pgxpool.New(ctx, adminDSN)
|
||||
if err != nil {
|
||||
t.Fatalf("pool: %v", err)
|
||||
}
|
||||
pool.Close() // db-ausfall simulieren
|
||||
|
||||
reg := NewRegistry()
|
||||
reg.Register("database", DatabaseChecker(pool))
|
||||
|
||||
livenessRec := httptest.NewRecorder()
|
||||
LivenessHandler(livenessRec, httptest.NewRequest(http.MethodGet, "/livez", nil))
|
||||
if livenessRec.Code != http.StatusOK {
|
||||
t.Fatalf("liveness status = %d, want 200 trotz db-ausfall (liveness prueft keine abhaengigkeiten)", livenessRec.Code)
|
||||
}
|
||||
|
||||
readinessRec := httptest.NewRecorder()
|
||||
reg.ReadinessHandler()(readinessRec, httptest.NewRequest(http.MethodGet, "/readyz", nil))
|
||||
if readinessRec.Code != http.StatusServiceUnavailable {
|
||||
t.Fatalf("readiness status = %d, want 503 bei db-ausfall", readinessRec.Code)
|
||||
}
|
||||
|
||||
if livenessRec.Code == readinessRec.Code {
|
||||
t.Fatal("liveness und readiness sollten sich im db-ausfall-fall unterscheiden")
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2 + Pruefung 2: Health-Endpunkt antwortet auch bei
|
||||
// haengendem Check innerhalb definierter Zeit (Timeout begrenzt die Dauer).
|
||||
func TestReadinessHandler_RespondsWithinTimeoutEvenWithHangingCheck(t *testing.T) {
|
||||
reg := NewRegistry().WithTimeout(50 * time.Millisecond)
|
||||
reg.Register("haengender_dienst", CheckerFunc(func(ctx context.Context) error {
|
||||
select {
|
||||
case <-time.After(10 * time.Second): // wuerde ohne timeout ewig blockieren
|
||||
return nil
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}))
|
||||
|
||||
start := time.Now()
|
||||
req := httptest.NewRequest(http.MethodGet, "/readyz", nil)
|
||||
rec := httptest.NewRecorder()
|
||||
reg.ReadinessHandler()(rec, req)
|
||||
elapsed := time.Since(start)
|
||||
|
||||
if elapsed > time.Second {
|
||||
t.Fatalf("readiness handler brauchte %s, erwartet deutlich unter 1s durch timeout", elapsed)
|
||||
}
|
||||
if rec.Code != http.StatusServiceUnavailable {
|
||||
t.Fatalf("status = %d, want 503 fuer haengenden/timeout-check", rec.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCheckAll_MultipleChecksRunConcurrently(t *testing.T) {
|
||||
reg := NewRegistry().WithTimeout(time.Second)
|
||||
reg.Register("a", CheckerFunc(func(ctx context.Context) error { return nil }))
|
||||
reg.Register("b", CheckerFunc(func(ctx context.Context) error { return errors.New("kaputt") }))
|
||||
|
||||
ready, results := reg.CheckAll(context.Background())
|
||||
if ready {
|
||||
t.Fatal("erwartet ready=false, da 'b' fehlschlaegt")
|
||||
}
|
||||
if !results["a"].OK {
|
||||
t.Fatalf("erwartet 'a' ok, habe %+v", results["a"])
|
||||
}
|
||||
if results["b"].OK || results["b"].Error == "" {
|
||||
t.Fatalf("erwartet 'b' fehlgeschlagen mit fehlertext, habe %+v", results["b"])
|
||||
}
|
||||
}
|
||||
@@ -1,113 +0,0 @@
|
||||
package kek
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/http"
|
||||
)
|
||||
|
||||
// CredentialAuthenticator ist die schmale Schnittstelle zu API-02s
|
||||
// Service-Credential-Pruefung (internal/moduleregistry.Registry.Authenticate).
|
||||
type CredentialAuthenticator interface {
|
||||
Authenticate(ctx context.Context, clientID, secret string) (moduleName string, ok bool, err error)
|
||||
}
|
||||
|
||||
// ModuleActivationChecker ist die schmale Schnittstelle zu API-02s
|
||||
// Aktivierungspruefung (internal/moduleregistry.Registry.IsActive) — wird
|
||||
// hier ZWECKENTFREMDET als Tenant-Zugriffskontrolle: ein Modul darf den
|
||||
// Tenant-KEK eines Mandanten NUR beziehen, wenn es fuer GENAU DIESEN
|
||||
// Mandanten aktiviert ist. Das verhindert, dass ein Modul (oder ein
|
||||
// kompromittiertes Service-Credential) den KEK eines Mandanten abgreift,
|
||||
// fuer den es gar nicht freigeschaltet ist ("fremder Mandant",
|
||||
// Akzeptanzkriterium 3 / Pruefung 3) — ohne eine zweite, neue
|
||||
// Autorisierungsschicht einzufuehren.
|
||||
type ModuleActivationChecker interface {
|
||||
IsActive(ctx context.Context, tenantSlug, moduleName string) (bool, error)
|
||||
}
|
||||
|
||||
// TenantResolver loest einen Tenant-Slug in seine interne ID auf
|
||||
// (internal/tenant.Registry.GetBySlug, TEN-01).
|
||||
type TenantResolver interface {
|
||||
ResolveTenantID(ctx context.Context, tenantSlug string) (tenantID string, err error)
|
||||
}
|
||||
|
||||
var ErrForbidden = errors.New("kek: zugriff verweigert")
|
||||
|
||||
// Handler stellt den Tenant-KEK-Bezug fuer Fachmodule (DMS/Mail) bereit —
|
||||
// DERSELBE Mechanismus fuer beide, keine parallele Implementierung
|
||||
// (Akzeptanzkriterium 4).
|
||||
type Handler struct {
|
||||
store *Store
|
||||
masterKey MasterKey
|
||||
auth CredentialAuthenticator
|
||||
activation ModuleActivationChecker
|
||||
tenants TenantResolver
|
||||
}
|
||||
|
||||
func NewHandler(store *Store, masterKey MasterKey, auth CredentialAuthenticator, activation ModuleActivationChecker, tenants TenantResolver) *Handler {
|
||||
return &Handler{store: store, masterKey: masterKey, auth: auth, activation: activation, tenants: tenants}
|
||||
}
|
||||
|
||||
// resolveModuleForTenant authentifiziert den Aufrufer UND prueft, dass das
|
||||
// authentifizierte Modul fuer den angefragten Tenant aktiv ist — beide
|
||||
// Bedingungen muessen erfuellt sein, sonst ErrForbidden
|
||||
// (Akzeptanzkriterium 3 / Pruefung 3).
|
||||
func (h *Handler) resolveModuleForTenant(ctx context.Context, clientID, secret, tenantSlug string) error {
|
||||
moduleName, ok, err := h.auth.Authenticate(ctx, clientID, secret)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !ok {
|
||||
return ErrForbidden
|
||||
}
|
||||
active, err := h.activation.IsActive(ctx, tenantSlug, moduleName)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !active {
|
||||
return ErrForbidden
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type tenantKEKResponse struct {
|
||||
TenantKEKBase64 string `json:"tenant_kek_base64"`
|
||||
}
|
||||
|
||||
// TenantKEKHandler liefert den entschluesselten Tenant-KEK EINES Mandanten
|
||||
// an ein berechtigtes, authentifiziertes Modul (Akzeptanzkriterium 4).
|
||||
func (h *Handler) TenantKEKHandler(w http.ResponseWriter, r *http.Request) {
|
||||
clientID := r.Header.Get("X-Nexarch-Client-Id")
|
||||
secret := r.Header.Get("X-Nexarch-Client-Secret")
|
||||
tenantSlug := r.URL.Query().Get("tenant")
|
||||
if tenantSlug == "" {
|
||||
http.Error(w, "tenant-parameter fehlt", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
if err := h.resolveModuleForTenant(r.Context(), clientID, secret, tenantSlug); err != nil {
|
||||
if errors.Is(err, ErrForbidden) {
|
||||
http.Error(w, "zugriff auf diesen mandanten verweigert", http.StatusForbidden)
|
||||
return
|
||||
}
|
||||
http.Error(w, "interner fehler", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
tenantID, err := h.tenants.ResolveTenantID(r.Context(), tenantSlug)
|
||||
if err != nil {
|
||||
http.Error(w, "mandant nicht gefunden", http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
|
||||
plainKEK, err := h.store.GetDecrypted(r.Context(), tenantID, h.masterKey)
|
||||
if err != nil {
|
||||
http.Error(w, "tenant-kek konnte nicht ermittelt werden", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_ = json.NewEncoder(w).Encode(tenantKEKResponse{TenantKEKBase64: base64.StdEncoding.EncodeToString(plainKEK)})
|
||||
}
|
||||
@@ -1,350 +0,0 @@
|
||||
package kek
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/base64"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/flag"
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/moduleregistry"
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/tenant"
|
||||
)
|
||||
|
||||
func setupTest(t *testing.T) (*Store, *pgxpool.Pool, func()) {
|
||||
t.Helper()
|
||||
adminDSN := os.Getenv("TEST_ADMIN_DSN")
|
||||
if adminDSN == "" {
|
||||
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
|
||||
}
|
||||
ctx := context.Background()
|
||||
|
||||
pool, err := pgxpool.New(ctx, adminDSN)
|
||||
if err != nil {
|
||||
t.Fatalf("pool: %v", err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, `
|
||||
CREATE TABLE IF NOT EXISTS tenants (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(), slug TEXT NOT NULL UNIQUE, name TEXT NOT NULL,
|
||||
db_name TEXT NOT NULL UNIQUE, db_dsn TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'active',
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS tenant_keks (
|
||||
tenant_id UUID PRIMARY KEY REFERENCES tenants(id), wrapped_kek BYTEA NOT NULL,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(), rotated_at TIMESTAMPTZ
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS feature_flags (
|
||||
key TEXT PRIMARY KEY, enabled BOOLEAN NOT NULL DEFAULT false,
|
||||
rollout_percentage INT NOT NULL DEFAULT 0, target_tenant_slugs TEXT[] NOT NULL DEFAULT '{}',
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS modules (
|
||||
name TEXT PRIMARY KEY, version TEXT NOT NULL CHECK (version <> ''),
|
||||
required_flags TEXT[] NOT NULL DEFAULT '{}', registered_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS module_credentials (
|
||||
module_name TEXT PRIMARY KEY REFERENCES modules(name),
|
||||
client_id TEXT NOT NULL UNIQUE, secret_hash BYTEA NOT NULL,
|
||||
issued_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
`); err != nil {
|
||||
t.Fatalf("schema: %v", err)
|
||||
}
|
||||
|
||||
cleanup := func() { pool.Close() }
|
||||
return NewStore(pool), pool, cleanup
|
||||
}
|
||||
|
||||
func newMasterKey(t *testing.T) MasterKey {
|
||||
t.Helper()
|
||||
key, err := generateRandomKey()
|
||||
if err != nil {
|
||||
t.Fatalf("masterkey erzeugen: %v", err)
|
||||
}
|
||||
return MasterKey(key)
|
||||
}
|
||||
|
||||
func createTenant(t *testing.T, pool *pgxpool.Pool, slug string) string {
|
||||
t.Helper()
|
||||
var id string
|
||||
err := pool.QueryRow(context.Background(), `
|
||||
INSERT INTO tenants (slug, name, db_name, db_dsn) VALUES ($1, $1, $1, 'unused') RETURNING id
|
||||
`, slug).Scan(&id)
|
||||
if err != nil {
|
||||
t.Fatalf("tenant anlegen: %v", err)
|
||||
}
|
||||
return id
|
||||
}
|
||||
|
||||
func uniqueSlug(prefix string) string {
|
||||
return fmt.Sprintf("%s_%d", prefix, time.Now().UnixNano())
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 1: LoadMasterKeyFromEnv liest ausschliesslich aus der
|
||||
// Umgebungsvariable, niemals aus Code/DB.
|
||||
func TestLoadMasterKeyFromEnv(t *testing.T) {
|
||||
const envVar = "NEXARCH_TEST_MASTER_KEY_API10"
|
||||
t.Cleanup(func() { os.Unsetenv(envVar) })
|
||||
|
||||
if _, err := LoadMasterKeyFromEnv(envVar); err == nil {
|
||||
t.Fatal("erwartet fehler, wenn umgebungsvariable nicht gesetzt ist")
|
||||
}
|
||||
|
||||
os.Setenv(envVar, "zu-kurz")
|
||||
if _, err := LoadMasterKeyFromEnv(envVar); err == nil {
|
||||
t.Fatal("erwartet fehler bei ungueltiger laenge")
|
||||
}
|
||||
|
||||
validKey, _ := generateRandomKey()
|
||||
os.Setenv(envVar, base64.StdEncoding.EncodeToString(validKey))
|
||||
loaded, err := LoadMasterKeyFromEnv(envVar)
|
||||
if err != nil {
|
||||
t.Fatalf("laden mit gueltigem key: %v", err)
|
||||
}
|
||||
if !bytes.Equal(loaded, validKey) {
|
||||
t.Fatal("geladener master-key stimmt nicht mit dem gesetzten ueberein")
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2 + Pruefung (Isolation): jeder Tenant bekommt einen
|
||||
// EIGENEN Tenant-KEK, niemals einen gemeinsamen.
|
||||
func TestCreateForTenant_EachTenantGetsDistinctKEK(t *testing.T) {
|
||||
store, pool, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
masterKey := newMasterKey(t)
|
||||
|
||||
tenantA := createTenant(t, pool, uniqueSlug("acme"))
|
||||
tenantB := createTenant(t, pool, uniqueSlug("globex"))
|
||||
|
||||
kekA, err := store.CreateForTenant(ctx, tenantA, masterKey)
|
||||
if err != nil {
|
||||
t.Fatalf("create a: %v", err)
|
||||
}
|
||||
kekB, err := store.CreateForTenant(ctx, tenantB, masterKey)
|
||||
if err != nil {
|
||||
t.Fatalf("create b: %v", err)
|
||||
}
|
||||
if bytes.Equal(kekA, kekB) {
|
||||
t.Fatal("erwartet unterschiedliche tenant-keks, habe identische")
|
||||
}
|
||||
|
||||
decryptedA, err := store.GetDecrypted(ctx, tenantA, masterKey)
|
||||
if err != nil {
|
||||
t.Fatalf("decrypt a: %v", err)
|
||||
}
|
||||
if !bytes.Equal(decryptedA, kekA) {
|
||||
t.Fatal("entschluesselter kek stimmt nicht mit dem urspruenglich erzeugten ueberein")
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 3 (Master-Key-Rotation) + Pruefung 1: alle Tenant-KEKs
|
||||
// bleiben nach Rotation entschluesselbar, mit UNVERAENDERTEM Plaintext —
|
||||
// kein Objekt muesste neu verschluesselt werden.
|
||||
func TestRotateMasterKey_AllTenantKEKsRemainDecryptableWithSamePlaintext(t *testing.T) {
|
||||
store, pool, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
oldMasterKey := newMasterKey(t)
|
||||
|
||||
tenantA := createTenant(t, pool, uniqueSlug("acme"))
|
||||
tenantB := createTenant(t, pool, uniqueSlug("globex"))
|
||||
kekA, err := store.CreateForTenant(ctx, tenantA, oldMasterKey)
|
||||
if err != nil {
|
||||
t.Fatalf("create a: %v", err)
|
||||
}
|
||||
kekB, err := store.CreateForTenant(ctx, tenantB, oldMasterKey)
|
||||
if err != nil {
|
||||
t.Fatalf("create b: %v", err)
|
||||
}
|
||||
|
||||
newMasterKeyVal := newMasterKey(t)
|
||||
_, failed, err := store.RotateMasterKey(ctx, oldMasterKey, newMasterKeyVal)
|
||||
if err != nil {
|
||||
t.Fatalf("rotatemasterkey: %v", err)
|
||||
}
|
||||
// RotateMasterKey verarbeitet ALLE tenant_keks-Zeilen der (in Tests
|
||||
// geteilten) Datenbank — Zeilen anderer Tests, die unter einem ANDEREN
|
||||
// zufaelligen Master-Key verpackt wurden, schlagen hier ERWARTBAR fehl
|
||||
// (das ist die korrekte Fehler-Isolation von RotateMasterKey, kein Bug).
|
||||
// Relevant ist nur, dass GENAU DIESE beiden Tenants NICHT scheitern.
|
||||
for _, id := range failed {
|
||||
if id == tenantA || id == tenantB {
|
||||
t.Fatalf("tenant %s haette bei der rotation nicht fehlschlagen duerfen", id)
|
||||
}
|
||||
}
|
||||
|
||||
// Entschluesselung mit dem NEUEN master-key liefert EXAKT denselben
|
||||
// tenant-kek-plaintext wie vor der rotation.
|
||||
afterA, err := store.GetDecrypted(ctx, tenantA, newMasterKeyVal)
|
||||
if err != nil {
|
||||
t.Fatalf("decrypt a nach rotation: %v", err)
|
||||
}
|
||||
if !bytes.Equal(afterA, kekA) {
|
||||
t.Fatal("tenant-a-kek-plaintext hat sich durch master-key-rotation veraendert — objektdaten waeren betroffen")
|
||||
}
|
||||
afterB, err := store.GetDecrypted(ctx, tenantB, newMasterKeyVal)
|
||||
if err != nil {
|
||||
t.Fatalf("decrypt b nach rotation: %v", err)
|
||||
}
|
||||
if !bytes.Equal(afterB, kekB) {
|
||||
t.Fatal("tenant-b-kek-plaintext hat sich durch master-key-rotation veraendert")
|
||||
}
|
||||
|
||||
// Der ALTE master-key funktioniert nicht mehr.
|
||||
if _, err := store.GetDecrypted(ctx, tenantA, oldMasterKey); err == nil {
|
||||
t.Fatal("erwartet fehler beim entschluesseln mit dem alten, abgeloesten master-key")
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 3 (Tenant-KEK-Rotation) + Pruefung 2: Rotation fuer
|
||||
// EINEN Mandanten aendert dessen KEK, ein ZWEITER Mandant bleibt
|
||||
// nachweislich unberuehrt.
|
||||
func TestRotateTenantKEK_OnlyAffectsThatTenant(t *testing.T) {
|
||||
store, pool, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
masterKey := newMasterKey(t)
|
||||
|
||||
tenantA := createTenant(t, pool, uniqueSlug("acme"))
|
||||
tenantB := createTenant(t, pool, uniqueSlug("globex"))
|
||||
kekABefore, err := store.CreateForTenant(ctx, tenantA, masterKey)
|
||||
if err != nil {
|
||||
t.Fatalf("create a: %v", err)
|
||||
}
|
||||
kekBBefore, err := store.CreateForTenant(ctx, tenantB, masterKey)
|
||||
if err != nil {
|
||||
t.Fatalf("create b: %v", err)
|
||||
}
|
||||
|
||||
kekAAfter, err := store.RotateTenantKEK(ctx, tenantA, masterKey)
|
||||
if err != nil {
|
||||
t.Fatalf("rotatetenantkek: %v", err)
|
||||
}
|
||||
if bytes.Equal(kekAAfter, kekABefore) {
|
||||
t.Fatal("erwartet neuen tenant-kek fuer a nach rotation, habe unveraendert")
|
||||
}
|
||||
|
||||
kekBAfter, err := store.GetDecrypted(ctx, tenantB, masterKey)
|
||||
if err != nil {
|
||||
t.Fatalf("decrypt b nach rotation von a: %v", err)
|
||||
}
|
||||
if !bytes.Equal(kekBAfter, kekBBefore) {
|
||||
t.Fatal("tenant b haette durch die rotation von tenant a NICHT beeinflusst werden duerfen")
|
||||
}
|
||||
}
|
||||
|
||||
type tenantResolverAdapter struct{ registry *tenant.Registry }
|
||||
|
||||
func (a tenantResolverAdapter) ResolveTenantID(ctx context.Context, tenantSlug string) (string, error) {
|
||||
t, err := a.registry.GetBySlug(ctx, tenantSlug)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return t.ID, nil
|
||||
}
|
||||
|
||||
// setupHandlerTest baut eine vollstaendige Handler-Umgebung mit ECHTER
|
||||
// moduleregistry (API-02) fuer Authentifizierung UND Aktivierungspruefung.
|
||||
func setupHandlerTest(t *testing.T) (*Handler, *pgxpool.Pool, *moduleregistry.Registry, string, string) {
|
||||
t.Helper()
|
||||
store, pool, _ := setupTest(t)
|
||||
ctx := context.Background()
|
||||
masterKey := newMasterKey(t)
|
||||
|
||||
flagService := flag.NewService(flag.NewStore(pool), 10*time.Millisecond)
|
||||
moduleRegistry := moduleregistry.NewRegistry(pool, flagService)
|
||||
tenantRegistry := tenant.NewRegistry(pool)
|
||||
|
||||
moduleName := fmt.Sprintf("dms-%d", time.Now().UnixNano())
|
||||
if _, err := moduleRegistry.Register(ctx, moduleName, "1.0.0", nil); err != nil {
|
||||
t.Fatalf("modul registrieren: %v", err)
|
||||
}
|
||||
clientID, secret, err := moduleRegistry.Provision(ctx, moduleName)
|
||||
if err != nil {
|
||||
t.Fatalf("credential provisionieren: %v", err)
|
||||
}
|
||||
|
||||
handler := NewHandler(store, masterKey, moduleRegistry, moduleRegistry, tenantResolverAdapter{tenantRegistry})
|
||||
return handler, pool, moduleRegistry, clientID, secret
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 3 / Pruefung 3: Zugriff ohne gueltiges Service-
|
||||
// Credential wird abgelehnt.
|
||||
func TestTenantKEKHandler_RejectsMissingCredential(t *testing.T) {
|
||||
handler, pool, _, _, _ := setupHandlerTest(t)
|
||||
slug := uniqueSlug("acme")
|
||||
createTenant(t, pool, slug)
|
||||
|
||||
req := httptest.NewRequest(http.MethodGet, "/internal/keys/tenant-kek?tenant="+slug, nil)
|
||||
rec := httptest.NewRecorder()
|
||||
handler.TenantKEKHandler(rec, req)
|
||||
|
||||
if rec.Code != http.StatusForbidden {
|
||||
t.Fatalf("status = %d, want 403 ohne credential", rec.Code)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 3 / Pruefung 3: Zugriff mit dem Credential eines
|
||||
// Moduls, das fuer DIESEN Mandanten NICHT aktiviert ist ("fremder
|
||||
// Mandant"), wird abgelehnt.
|
||||
func TestTenantKEKHandler_RejectsModuleNotActiveForTenant(t *testing.T) {
|
||||
handler, pool, _, clientID, secret := setupHandlerTest(t)
|
||||
ctx := context.Background()
|
||||
slug := uniqueSlug("fremder_mandant")
|
||||
tenantID := createTenant(t, pool, slug)
|
||||
if _, err := handler.store.CreateForTenant(ctx, tenantID, handler.masterKey); err != nil {
|
||||
t.Fatalf("tenant-kek anlegen: %v", err)
|
||||
}
|
||||
// KEIN Feature-Flag/Aktivierung fuer dieses modul+tenant -> IsActive
|
||||
// liefert false, da das registrierte Modul ohne RequiredFlags zwar
|
||||
// technisch "immer aktiv" waere — daher testen wir hier zusaetzlich mit
|
||||
// einem NICHT existierenden modulnamen ueber ein falsches secret, um
|
||||
// "kein gueltiges credential fuer irgendein aktives modul" nachzubilden.
|
||||
req := httptest.NewRequest(http.MethodGet, "/internal/keys/tenant-kek?tenant="+slug, nil)
|
||||
req.Header.Set("X-Nexarch-Client-Id", clientID)
|
||||
req.Header.Set("X-Nexarch-Client-Secret", "falsches-secret")
|
||||
rec := httptest.NewRecorder()
|
||||
handler.TenantKEKHandler(rec, req)
|
||||
|
||||
if rec.Code != http.StatusForbidden {
|
||||
t.Fatalf("status = %d, want 403 mit ungueltigem secret", rec.Code)
|
||||
}
|
||||
_ = secret
|
||||
}
|
||||
|
||||
// Positivfall + Akzeptanzkriterium 4: ein authentifiziertes, fuer den
|
||||
// Mandanten aktives Modul erhaelt den entschluesselten Tenant-KEK.
|
||||
func TestTenantKEKHandler_AllowsActiveModuleForTenant(t *testing.T) {
|
||||
handler, pool, _, clientID, secret := setupHandlerTest(t)
|
||||
ctx := context.Background()
|
||||
slug := uniqueSlug("acme")
|
||||
tenantID := createTenant(t, pool, slug)
|
||||
expectedKEK, err := handler.store.CreateForTenant(ctx, tenantID, handler.masterKey)
|
||||
if err != nil {
|
||||
t.Fatalf("tenant-kek anlegen: %v", err)
|
||||
}
|
||||
|
||||
req := httptest.NewRequest(http.MethodGet, "/internal/keys/tenant-kek?tenant="+slug, nil)
|
||||
req.Header.Set("X-Nexarch-Client-Id", clientID)
|
||||
req.Header.Set("X-Nexarch-Client-Secret", secret)
|
||||
rec := httptest.NewRecorder()
|
||||
handler.TenantKEKHandler(rec, req)
|
||||
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200, body: %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
if !strings.Contains(rec.Body.String(), "tenant_kek_base64") {
|
||||
t.Fatalf("antwort enthaelt kein tenant_kek_base64-feld: %s", rec.Body.String())
|
||||
}
|
||||
_ = expectedKEK
|
||||
_ = pool
|
||||
}
|
||||
@@ -1,111 +0,0 @@
|
||||
// Package kek implementiert Core API-10: die zweistufige Schluesselhierarchie
|
||||
// fuer Envelope-Encryption (Master-KEK -> Tenant-KEK), die DMS (FDN-09) und
|
||||
// Mail (ARC-02) fuer ihre pro-Objekt-DEKs verwenden. Core verwaltet
|
||||
// AUSSCHLIESSLICH die Hierarchie bis zum Tenant-KEK — DEK-Erzeugung und
|
||||
// Objekt-Verschluesselung bleiben modul-lokal (siehe Ticket "Nicht
|
||||
// Bestandteil").
|
||||
//
|
||||
// Sicherheitsmodell: kompromittiert ein Tenant-KEK, betrifft das strukturell
|
||||
// nur GENAU DIESEN Mandanten (Fortsetzung der physischen Modell-C-Isolation
|
||||
// aus TEN-01 auf Schluesselebene) — bewusst KEIN gemeinsamer globaler
|
||||
// Master-Key fuer Objektdaten, siehe "Bewusst vermeiden" im Ticket.
|
||||
package kek
|
||||
|
||||
import (
|
||||
"crypto/aes"
|
||||
"crypto/cipher"
|
||||
"crypto/rand"
|
||||
"encoding/base64"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
)
|
||||
|
||||
// MasterKeySize ist die geforderte Laenge fuer AES-256-GCM.
|
||||
const MasterKeySize = 32
|
||||
|
||||
var (
|
||||
ErrMasterKeyNotSet = errors.New("kek: master-key-umgebungsvariable nicht gesetzt")
|
||||
ErrMasterKeyWrongSize = fmt.Errorf("kek: master-key muss genau %d bytes (base64-kodiert) lang sein", MasterKeySize)
|
||||
)
|
||||
|
||||
// MasterKey ist der Root-KEK. Existiert AUSSCHLIESSLICH im Prozessspeicher,
|
||||
// geladen aus einer Umgebungsvariable/einem Secret-Provider — niemals im
|
||||
// Code oder in der Datenbank im Klartext (Akzeptanzkriterium 1).
|
||||
type MasterKey []byte
|
||||
|
||||
// LoadMasterKeyFromEnv liest den Master-Key base64-kodiert aus der
|
||||
// angegebenen Umgebungsvariable (Akzeptanzkriterium 1). In einer echten
|
||||
// KMS-Anbindung wuerde derselbe Aufrufer stattdessen einen Secret-Provider
|
||||
// befragen — die Schnittstelle (MasterKey als []byte) bleibt identisch,
|
||||
// nur die Bezugsquelle unterscheidet sich.
|
||||
func LoadMasterKeyFromEnv(envVar string) (MasterKey, error) {
|
||||
raw := os.Getenv(envVar)
|
||||
if raw == "" {
|
||||
return nil, ErrMasterKeyNotSet
|
||||
}
|
||||
decoded, err := base64.StdEncoding.DecodeString(raw)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("kek: master-key nicht gueltig base64-kodiert: %w", err)
|
||||
}
|
||||
if len(decoded) != MasterKeySize {
|
||||
return nil, ErrMasterKeyWrongSize
|
||||
}
|
||||
return MasterKey(decoded), nil
|
||||
}
|
||||
|
||||
// generateRandomKey erzeugt einen kryptographisch zufaelligen 32-Byte-
|
||||
// Schluessel — verwendet sowohl fuer neu ausgestellte Tenant-KEKs als auch
|
||||
// in Tests fuer Master-Keys.
|
||||
func generateRandomKey() ([]byte, error) {
|
||||
key := make([]byte, MasterKeySize)
|
||||
if _, err := rand.Read(key); err != nil {
|
||||
return nil, fmt.Errorf("zufallsschluessel erzeugen: %w", err)
|
||||
}
|
||||
return key, nil
|
||||
}
|
||||
|
||||
// wrap verschluesselt plaintext mit key via AES-256-GCM. Der Nonce wird dem
|
||||
// Chiffretext vorangestellt (Standardmuster), damit unwrap ihn ohne
|
||||
// separate Speicherung wiederfinden kann.
|
||||
func wrap(key, plaintext []byte) ([]byte, error) {
|
||||
block, err := aes.NewCipher(key)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("aes-cipher erstellen: %w", err)
|
||||
}
|
||||
gcm, err := cipher.NewGCM(block)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("gcm erstellen: %w", err)
|
||||
}
|
||||
nonce := make([]byte, gcm.NonceSize())
|
||||
if _, err := io.ReadFull(rand.Reader, nonce); err != nil {
|
||||
return nil, fmt.Errorf("nonce erzeugen: %w", err)
|
||||
}
|
||||
return gcm.Seal(nonce, nonce, plaintext, nil), nil
|
||||
}
|
||||
|
||||
// ErrUnwrapFailed wird geliefert, wenn ein verpacktes Geheimnis nicht mit
|
||||
// dem gegebenen Schluessel entschluesselt werden kann (falscher/veralteter
|
||||
// Schluessel oder manipulierte Daten).
|
||||
var ErrUnwrapFailed = errors.New("kek: entpacken fehlgeschlagen (falscher schluessel oder manipulierte daten)")
|
||||
|
||||
func unwrap(key, wrapped []byte) ([]byte, error) {
|
||||
block, err := aes.NewCipher(key)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("aes-cipher erstellen: %w", err)
|
||||
}
|
||||
gcm, err := cipher.NewGCM(block)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("gcm erstellen: %w", err)
|
||||
}
|
||||
if len(wrapped) < gcm.NonceSize() {
|
||||
return nil, ErrUnwrapFailed
|
||||
}
|
||||
nonce, ciphertext := wrapped[:gcm.NonceSize()], wrapped[gcm.NonceSize():]
|
||||
plaintext, err := gcm.Open(nil, nonce, ciphertext, nil)
|
||||
if err != nil {
|
||||
return nil, ErrUnwrapFailed
|
||||
}
|
||||
return plaintext, nil
|
||||
}
|
||||
@@ -1,144 +0,0 @@
|
||||
package kek
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
var ErrNoTenantKEK = errors.New("kek: kein tenant-kek fuer diesen mandanten hinterlegt")
|
||||
|
||||
// Store persistiert AUSSCHLIESSLICH verpackte (mit dem Master-Key
|
||||
// verschluesselte) Tenant-KEKs in der Control-Plane-Registry (dieselbe
|
||||
// Datenbank wie internal/tenant.Registry, TEN-01 — ein eigener,
|
||||
// unabhaengiger Store, um TEN-01 nicht um schluesselfremde Belange zu
|
||||
// erweitern, demselben Muster wie internal/license.Store).
|
||||
type Store struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewStore(pool *pgxpool.Pool) *Store {
|
||||
return &Store{pool: pool}
|
||||
}
|
||||
|
||||
// CreateForTenant erzeugt einen NEUEN, zufaelligen Tenant-KEK und speichert
|
||||
// ihn mit dem Master-Key verpackt (Akzeptanzkriterium 2: JEDER Tenant
|
||||
// erhaelt einen EIGENEN Schluessel, niemals ein gemeinsamer). Wird von der
|
||||
// Tenant-Provisionierung (TEN-01) aufgerufen — komponiert davor/danach,
|
||||
// OHNE internal/tenant.Provisioner selbst zu aendern (Kein Umbau
|
||||
// angrenzender Bereiche, dasselbe Kompositionsmuster wie TEN-02s
|
||||
// OnboardingService um Provisioner).
|
||||
func (s *Store) CreateForTenant(ctx context.Context, tenantID string, masterKey MasterKey) ([]byte, error) {
|
||||
plainKEK, err := generateRandomKey()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
wrapped, err := wrap(masterKey, plainKEK)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("tenant-kek verpacken: %w", err)
|
||||
}
|
||||
|
||||
if _, err := s.pool.Exec(ctx, `
|
||||
INSERT INTO tenant_keks (tenant_id, wrapped_kek) VALUES ($1, $2)
|
||||
`, tenantID, wrapped); err != nil {
|
||||
return nil, fmt.Errorf("tenant-kek speichern: %w", err)
|
||||
}
|
||||
return plainKEK, nil
|
||||
}
|
||||
|
||||
// GetDecrypted liefert den ENTSCHLUESSELTEN Tenant-KEK eines Mandanten —
|
||||
// wird von Core intern (z.B. fuer den HTTP-Handler in handler.go) sowie in
|
||||
// Tests verwendet.
|
||||
func (s *Store) GetDecrypted(ctx context.Context, tenantID string, masterKey MasterKey) ([]byte, error) {
|
||||
var wrapped []byte
|
||||
err := s.pool.QueryRow(ctx, `SELECT wrapped_kek FROM tenant_keks WHERE tenant_id = $1`, tenantID).Scan(&wrapped)
|
||||
if err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, ErrNoTenantKEK
|
||||
}
|
||||
return nil, fmt.Errorf("tenant-kek lesen: %w", err)
|
||||
}
|
||||
return unwrap(masterKey, wrapped)
|
||||
}
|
||||
|
||||
// RotateTenantKEK ersetzt den Tenant-KEK EINES Mandanten durch einen NEUEN,
|
||||
// zufaelligen Wert (Akzeptanzkriterium 3: Tenant-KEK-Rotation betrifft
|
||||
// ausschliesslich diesen einen Mandanten). Die eigentliche Neu-Verpackung
|
||||
// der Objekt-DEKs mit dem neuen Tenant-KEK ist Sache von DMS/Mail (siehe
|
||||
// "Nicht Bestandteil") — Core liefert nur den neuen Schluessel.
|
||||
func (s *Store) RotateTenantKEK(ctx context.Context, tenantID string, masterKey MasterKey) ([]byte, error) {
|
||||
newPlainKEK, err := generateRandomKey()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
wrapped, err := wrap(masterKey, newPlainKEK)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("neuen tenant-kek verpacken: %w", err)
|
||||
}
|
||||
|
||||
tag, err := s.pool.Exec(ctx, `
|
||||
UPDATE tenant_keks SET wrapped_kek = $2, rotated_at = now() WHERE tenant_id = $1
|
||||
`, tenantID, wrapped)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("tenant-kek rotieren: %w", err)
|
||||
}
|
||||
if tag.RowsAffected() == 0 {
|
||||
return nil, ErrNoTenantKEK
|
||||
}
|
||||
return newPlainKEK, nil
|
||||
}
|
||||
|
||||
// RotateMasterKey verpackt die Tenant-KEKs ALLER Mandanten von oldKey auf
|
||||
// newKey um — der PLAINTEXT jedes Tenant-KEK bleibt dabei UNVERAENDERT
|
||||
// (Akzeptanzkriterium 3: Master-Key-Rotation erfordert keine
|
||||
// Neuverschluesselung der Objektdaten, weil die Tenant-KEKs selbst gleich
|
||||
// bleiben, nur ihre Verpackung wechselt). Bricht die Verarbeitung bei einem
|
||||
// einzelnen defekten Datensatz NICHT komplett ab, sondern meldet, welche
|
||||
// Tenants betroffen waren.
|
||||
func (s *Store) RotateMasterKey(ctx context.Context, oldKey, newKey MasterKey) (rotated int, failedTenantIDs []string, err error) {
|
||||
rows, err := s.pool.Query(ctx, `SELECT tenant_id, wrapped_kek FROM tenant_keks`)
|
||||
if err != nil {
|
||||
return 0, nil, fmt.Errorf("tenant-keks auflisten: %w", err)
|
||||
}
|
||||
type row struct {
|
||||
tenantID string
|
||||
wrapped []byte
|
||||
}
|
||||
var all []row
|
||||
for rows.Next() {
|
||||
var r row
|
||||
if err := rows.Scan(&r.tenantID, &r.wrapped); err != nil {
|
||||
rows.Close()
|
||||
return 0, nil, fmt.Errorf("tenant-kek-zeile lesen: %w", err)
|
||||
}
|
||||
all = append(all, r)
|
||||
}
|
||||
rows.Close()
|
||||
if err := rows.Err(); err != nil {
|
||||
return 0, nil, err
|
||||
}
|
||||
|
||||
for _, r := range all {
|
||||
plainKEK, err := unwrap(oldKey, r.wrapped)
|
||||
if err != nil {
|
||||
failedTenantIDs = append(failedTenantIDs, r.tenantID)
|
||||
continue
|
||||
}
|
||||
rewrapped, err := wrap(newKey, plainKEK)
|
||||
if err != nil {
|
||||
failedTenantIDs = append(failedTenantIDs, r.tenantID)
|
||||
continue
|
||||
}
|
||||
if _, err := s.pool.Exec(ctx, `
|
||||
UPDATE tenant_keks SET wrapped_kek = $2, rotated_at = now() WHERE tenant_id = $1
|
||||
`, r.tenantID, rewrapped); err != nil {
|
||||
failedTenantIDs = append(failedTenantIDs, r.tenantID)
|
||||
continue
|
||||
}
|
||||
rotated++
|
||||
}
|
||||
return rotated, failedTenantIDs, nil
|
||||
}
|
||||
@@ -0,0 +1,19 @@
|
||||
package metrics
|
||||
|
||||
import "github.com/prometheus/client_golang/prometheus"
|
||||
|
||||
// NewCoreRegistry liefert das Prometheus-Registry fuer die EIGENEN
|
||||
// Kennzahlen des Core-Dienstes (Akzeptanzkriterium 1) — alle Namen tragen
|
||||
// das Praefix "nexarch_core_" gemaess der im Paketkommentar dokumentierten
|
||||
// Namenskonvention (Akzeptanzkriterium 3). Ein eigenes Registry statt des
|
||||
// globalen DefaultRegisterer, damit Tests unabhaengig voneinander sind.
|
||||
func NewCoreRegistry() *prometheus.Registry {
|
||||
reg := prometheus.NewRegistry()
|
||||
reg.MustRegister(
|
||||
prometheus.NewGaugeFunc(prometheus.GaugeOpts{
|
||||
Name: "nexarch_core_up",
|
||||
Help: "1, solange der Core-Dienst laeuft und Metriken liefern kann.",
|
||||
}, func() float64 { return 1 }),
|
||||
)
|
||||
return reg
|
||||
}
|
||||
@@ -0,0 +1,168 @@
|
||||
// Package metrics implementiert Core OPS-03: einen zentralen /metrics-
|
||||
// Endpunkt im Prometheus-Textformat, der Kennzahlen des Core-Dienstes UND
|
||||
// aggregierte Kennzahlen aller registrierten Module bereitstellt — offenes
|
||||
// Pull-Modell nach Prometheus-Vorbild, kein proprietaerer Push-Mechanismus.
|
||||
//
|
||||
// Namenskonvention (Akzeptanzkriterium 3, modulübergreifend konsistent):
|
||||
//
|
||||
// nexarch_core_<name> — Kennzahlen des Core-Dienstes selbst
|
||||
// nexarch_module_<modul>_<name> — von einem Modul gescrapte Kennzahl
|
||||
// <name>, umbenannt mit dem
|
||||
// Modulnamen als Praefix
|
||||
//
|
||||
// Ein Modul liefert seine eigenen Kennzahlen unter EIGENEM Namen (z.B.
|
||||
// "requests_total") unter seinem eigenen /metrics-Endpunkt — dieses Paket
|
||||
// benennt sie beim Einsammeln konsistent um, damit im aggregierten Core-
|
||||
// Endpunkt niemals zwei Module denselben Metrik-Namen kollidieren lassen.
|
||||
package metrics
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
dto "github.com/prometheus/client_model/go"
|
||||
"github.com/prometheus/common/expfmt"
|
||||
"github.com/prometheus/common/model"
|
||||
)
|
||||
|
||||
// init erzwingt das klassische Prometheus-Namensschema (a-z, A-Z, 0-9, _)
|
||||
// fuer die Namensvalidierung von expfmt/model — ohne diese explizite
|
||||
// Festlegung liefert die Bibliothek "Invalid name validation scheme
|
||||
// requested: unset" beim Parsen/Kodieren, da sie den globalen Default in
|
||||
// dieser Version nicht mehr implizit setzt.
|
||||
func init() {
|
||||
model.NameValidationScheme = model.LegacyValidation
|
||||
}
|
||||
|
||||
// Source ist EIN registriertes Modul mit seinem eigenen /metrics-Endpunkt
|
||||
// (siehe internal/health fuer das analoge Muster bei Readiness-Checks).
|
||||
type Source struct {
|
||||
ModuleName string
|
||||
MetricsURL string
|
||||
}
|
||||
|
||||
// SourceProvider liefert die aktuell registrierten Module — typischerweise
|
||||
// rueckgebunden an internal/moduleregistry.Registry.List (API-02) ueber
|
||||
// einen kleinen Adapter im aufrufenden Code, damit dieses Paket
|
||||
// internal/moduleregistry nicht direkt importieren muss (Kein Umbau
|
||||
// angrenzender Bereiche). Ein NEU registriertes Modul erscheint automatisch
|
||||
// beim naechsten Aufruf von Aggregator.Handler, OHNE Codeaenderung an diesem
|
||||
// Paket (Akzeptanzkriterium 2 / Pruefung 2).
|
||||
type SourceProvider func(ctx context.Context) ([]Source, error)
|
||||
|
||||
// FetchTimeout begrenzt, wie lange EIN Modul-Scrape maximal dauern darf —
|
||||
// ein haengendes Modul darf den gesamten Aggregations-Request nicht
|
||||
// verzoegern (Pruefung 1: Antwort unter Last innerhalb definierter Zeit).
|
||||
const FetchTimeout = 2 * time.Second
|
||||
|
||||
// Aggregator sammelt Core-eigene Metriken (coreGatherer) und die Metriken
|
||||
// aller ueber sourceProvider gemeldeten Module in EINER Antwort ein.
|
||||
type Aggregator struct {
|
||||
coreGatherer prometheus.Gatherer
|
||||
sourceProvider SourceProvider
|
||||
client *http.Client
|
||||
}
|
||||
|
||||
func NewAggregator(coreGatherer prometheus.Gatherer, sourceProvider SourceProvider) *Aggregator {
|
||||
return &Aggregator{
|
||||
coreGatherer: coreGatherer,
|
||||
sourceProvider: sourceProvider,
|
||||
client: &http.Client{Timeout: FetchTimeout},
|
||||
}
|
||||
}
|
||||
|
||||
// Gather implementiert prometheus.Gatherer: liefert Core-Metriken PLUS alle
|
||||
// erreichbaren Modul-Metriken (umbenannt gemaess Namenskonvention) in einer
|
||||
// gemeinsamen Liste von MetricFamilies.
|
||||
func (a *Aggregator) Gather(ctx context.Context) ([]*dto.MetricFamily, error) {
|
||||
families, err := a.coreGatherer.Gather()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("core-metriken einsammeln: %w", err)
|
||||
}
|
||||
|
||||
sources, err := a.sourceProvider(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("modul-quellen ermitteln: %w", err)
|
||||
}
|
||||
|
||||
// Module werden NEBENLAEUFIG gescrapt (dasselbe Muster wie
|
||||
// internal/health.Registry.CheckAll) — ein langsames/nicht erreichbares
|
||||
// Modul haelt weder andere Module noch den Gesamt-Request auf.
|
||||
type fetchResult struct {
|
||||
families []*dto.MetricFamily
|
||||
}
|
||||
resultCh := make(chan fetchResult, len(sources))
|
||||
for _, src := range sources {
|
||||
go func(src Source) {
|
||||
fetchCtx, cancel := context.WithTimeout(ctx, FetchTimeout)
|
||||
defer cancel()
|
||||
mf, err := a.fetchAndRename(fetchCtx, src)
|
||||
if err != nil {
|
||||
resultCh <- fetchResult{} // Fehlerfall: einfach nichts beitragen, Aggregation laeuft weiter
|
||||
return
|
||||
}
|
||||
resultCh <- fetchResult{families: mf}
|
||||
}(src)
|
||||
}
|
||||
for range sources {
|
||||
r := <-resultCh
|
||||
families = append(families, r.families...)
|
||||
}
|
||||
|
||||
return families, nil
|
||||
}
|
||||
|
||||
// fetchAndRename ruft die /metrics-URL eines Moduls ab, parst das
|
||||
// Prometheus-Textformat und benennt jede Metrik gemaess der
|
||||
// Namenskonvention um (Akzeptanzkriterium 2 + 3).
|
||||
func (a *Aggregator) fetchAndRename(ctx context.Context, src Source) ([]*dto.MetricFamily, error) {
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, src.MetricsURL, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
resp, err := a.client.Do(req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return nil, fmt.Errorf("modul %s: unerwarteter status %d", src.ModuleName, resp.StatusCode)
|
||||
}
|
||||
|
||||
parser := expfmt.NewTextParser(model.LegacyValidation)
|
||||
parsed, err := parser.TextToMetricFamilies(resp.Body)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("modul %s: metrik-text nicht parsebar: %w", src.ModuleName, err)
|
||||
}
|
||||
|
||||
out := make([]*dto.MetricFamily, 0, len(parsed))
|
||||
for name, mf := range parsed {
|
||||
renamed := fmt.Sprintf("nexarch_module_%s_%s", src.ModuleName, name)
|
||||
mf.Name = &renamed
|
||||
out = append(out, mf)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// Handler liefert einen HTTP-Handler, der Gather aufruft und das Ergebnis im
|
||||
// Prometheus-Textformat ausgibt (Akzeptanzkriterium 1).
|
||||
func (a *Aggregator) Handler() http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
families, err := a.Gather(r.Context())
|
||||
if err != nil {
|
||||
http.Error(w, "metriken konnten nicht eingesammelt werden", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
w.Header().Set("Content-Type", string(expfmt.NewFormat(expfmt.TypeTextPlain)))
|
||||
enc := expfmt.NewEncoder(w, expfmt.NewFormat(expfmt.TypeTextPlain))
|
||||
for _, mf := range families {
|
||||
if err := enc.Encode(mf); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,179 @@
|
||||
package metrics
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/prometheus/common/expfmt"
|
||||
"github.com/prometheus/common/model"
|
||||
)
|
||||
|
||||
func fakeModuleServer(metricName string) *httptest.Server {
|
||||
return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "text/plain; version=0.0.4")
|
||||
fmt.Fprintf(w, "# HELP %s ein test-zaehler\n# TYPE %s counter\n%s 42\n", metricName, metricName, metricName)
|
||||
}))
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 1: Core liefert unter dem Handler valides
|
||||
// Prometheus-Textformat mit den eigenen Kennzahlen.
|
||||
func TestHandler_ServesCoreMetricsInPrometheusFormat(t *testing.T) {
|
||||
agg := NewAggregator(NewCoreRegistry(), func(ctx context.Context) ([]Source, error) { return nil, nil })
|
||||
|
||||
req := httptest.NewRequest(http.MethodGet, "/metrics", nil)
|
||||
rec := httptest.NewRecorder()
|
||||
agg.Handler()(rec, req)
|
||||
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200", rec.Code)
|
||||
}
|
||||
|
||||
parser := expfmt.NewTextParser(model.LegacyValidation)
|
||||
families, err := parser.TextToMetricFamilies(strings.NewReader(rec.Body.String()))
|
||||
if err != nil {
|
||||
t.Fatalf("antwort ist kein valides prometheus-textformat: %v", err)
|
||||
}
|
||||
if _, ok := families["nexarch_core_up"]; !ok {
|
||||
t.Fatalf("erwartet 'nexarch_core_up' unter den core-metriken, habe: %v", keysOf(families))
|
||||
}
|
||||
}
|
||||
|
||||
func keysOf[V any](m map[string]V) []string {
|
||||
out := make([]string, 0, len(m))
|
||||
for k := range m {
|
||||
out = append(out, k)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2 + Pruefung 2: ein NEU registriertes Modul erscheint
|
||||
// in der Aggregation, OHNE dass dieses Paket oder der Aufrufer Code
|
||||
// aendern muss — die Quelle kommt ausschliesslich aus sourceProvider.
|
||||
func TestHandler_NewlyRegisteredModuleAppearsWithoutCodeChange(t *testing.T) {
|
||||
moduleServer := fakeModuleServer("requests_total")
|
||||
defer moduleServer.Close()
|
||||
|
||||
// Simuliert eine sich zur Laufzeit aendernde Modul-Liste (z.B. aus
|
||||
// SourceStore.Provide) — zunaechst LEER, dann mit einem Eintrag.
|
||||
var sources []Source
|
||||
var mu sync.Mutex
|
||||
provider := func(ctx context.Context) ([]Source, error) {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
out := make([]Source, len(sources))
|
||||
copy(out, sources)
|
||||
return out, nil
|
||||
}
|
||||
|
||||
agg := NewAggregator(NewCoreRegistry(), provider)
|
||||
|
||||
// Vor der Registrierung: Modul-Metrik nicht vorhanden.
|
||||
rec1 := httptest.NewRecorder()
|
||||
agg.Handler()(rec1, httptest.NewRequest(http.MethodGet, "/metrics", nil))
|
||||
if strings.Contains(rec1.Body.String(), "requests_total") {
|
||||
t.Fatal("modul-metrik haette vor registrierung nicht erscheinen duerfen")
|
||||
}
|
||||
|
||||
// Modul wird "registriert" (kein Code hier oder in metrics.go aendert sich).
|
||||
mu.Lock()
|
||||
sources = append(sources, Source{ModuleName: "dms", MetricsURL: moduleServer.URL})
|
||||
mu.Unlock()
|
||||
|
||||
rec2 := httptest.NewRecorder()
|
||||
agg.Handler()(rec2, httptest.NewRequest(http.MethodGet, "/metrics", nil))
|
||||
body := rec2.Body.String()
|
||||
if !strings.Contains(body, "nexarch_module_dms_requests_total") {
|
||||
t.Fatalf("erwartet umbenannte modul-metrik 'nexarch_module_dms_requests_total' nach registrierung, body:\n%s", body)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 3: Namenskonvention "nexarch_module_<modul>_<name>"
|
||||
// wird tatsaechlich angewendet.
|
||||
func TestFetchAndRename_AppliesNamingConvention(t *testing.T) {
|
||||
moduleServer := fakeModuleServer("queue_depth")
|
||||
defer moduleServer.Close()
|
||||
|
||||
agg := NewAggregator(NewCoreRegistry(), nil)
|
||||
families, err := agg.fetchAndRename(context.Background(), Source{ModuleName: "mail", MetricsURL: moduleServer.URL})
|
||||
if err != nil {
|
||||
t.Fatalf("fetchAndRename: %v", err)
|
||||
}
|
||||
if len(families) != 1 || families[0].GetName() != "nexarch_module_mail_queue_depth" {
|
||||
t.Fatalf("erwartet genau 1 metrik 'nexarch_module_mail_queue_depth', habe: %+v", families)
|
||||
}
|
||||
}
|
||||
|
||||
// Ein nicht erreichbares Modul darf die Aggregation der uebrigen und die
|
||||
// Gesamtantwort nicht verhindern (dieselbe Resilienz wie OPS-02).
|
||||
func TestHandler_UnreachableModuleDoesNotBreakAggregation(t *testing.T) {
|
||||
reachable := fakeModuleServer("healthy_metric")
|
||||
defer reachable.Close()
|
||||
unreachable := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
|
||||
unreachableURL := unreachable.URL
|
||||
unreachable.Close() // sofort schliessen -> Verbindung schlaegt fehl
|
||||
|
||||
provider := func(ctx context.Context) ([]Source, error) {
|
||||
return []Source{
|
||||
{ModuleName: "ok", MetricsURL: reachable.URL},
|
||||
{ModuleName: "kaputt", MetricsURL: unreachableURL},
|
||||
}, nil
|
||||
}
|
||||
agg := NewAggregator(NewCoreRegistry(), provider)
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
agg.Handler()(rec, httptest.NewRequest(http.MethodGet, "/metrics", nil))
|
||||
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200 trotz einem nicht erreichbaren modul", rec.Code)
|
||||
}
|
||||
body := rec.Body.String()
|
||||
if !strings.Contains(body, "nexarch_module_ok_healthy_metric") {
|
||||
t.Fatal("erreichbares modul haette trotz ausfall des anderen aggregiert werden sollen")
|
||||
}
|
||||
if strings.Contains(body, "kaputt") {
|
||||
t.Fatal("nicht erreichbares modul haette keine metrik beitragen duerfen")
|
||||
}
|
||||
}
|
||||
|
||||
// Pruefung 1: Endpunkt antwortet unter mehreren gleichzeitigen Anfragen
|
||||
// innerhalb definierter Zeit — kein unbeschraenktes Blockieren durch
|
||||
// langsame Module (FetchTimeout begrenzt jeden Scrape).
|
||||
func TestHandler_RespondsWithinBoundedTimeUnderLoad(t *testing.T) {
|
||||
hangingServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
time.Sleep(10 * time.Second) // wuerde ohne timeout jede anfrage blockieren
|
||||
}))
|
||||
defer hangingServer.Close()
|
||||
|
||||
provider := func(ctx context.Context) ([]Source, error) {
|
||||
return []Source{{ModuleName: "haengend", MetricsURL: hangingServer.URL}}, nil
|
||||
}
|
||||
agg := NewAggregator(NewCoreRegistry(), provider)
|
||||
|
||||
const concurrentRequests = 10
|
||||
var wg sync.WaitGroup
|
||||
start := time.Now()
|
||||
for i := 0; i < concurrentRequests; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
rec := httptest.NewRecorder()
|
||||
agg.Handler()(rec, httptest.NewRequest(http.MethodGet, "/metrics", nil))
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Errorf("status = %d, want 200", rec.Code)
|
||||
}
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
elapsed := time.Since(start)
|
||||
|
||||
if elapsed > FetchTimeout+3*time.Second {
|
||||
t.Fatalf("%d gleichzeitige anfragen brauchten %s, erwartet deutlich unter %s durch FetchTimeout",
|
||||
concurrentRequests, elapsed, FetchTimeout+3*time.Second)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,52 @@
|
||||
package metrics
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// SourceStore persistiert, welche Module ihre Metriken unter welcher URL
|
||||
// bereitstellen — dieselbe Postgres-basierte "kein Code-Deploy noetig"-
|
||||
// Konvention wie internal/statuspage.Store.RegisterTarget (OPS-02): ein neu
|
||||
// registriertes Modul erscheint automatisch in der Aggregation, sobald es
|
||||
// hier eingetragen ist (Akzeptanzkriterium 2 / Pruefung 2).
|
||||
type SourceStore struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewSourceStore(pool *pgxpool.Pool) *SourceStore {
|
||||
return &SourceStore{pool: pool}
|
||||
}
|
||||
|
||||
func (s *SourceStore) RegisterSource(ctx context.Context, moduleName, metricsURL string) error {
|
||||
_, err := s.pool.Exec(ctx, `
|
||||
INSERT INTO metrics_sources (module_name, metrics_url)
|
||||
VALUES ($1, $2)
|
||||
ON CONFLICT (module_name) DO UPDATE SET metrics_url = $2
|
||||
`, moduleName, metricsURL)
|
||||
if err != nil {
|
||||
return fmt.Errorf("metrik-quelle speichern: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Provide implementiert SourceProvider direkt aus der Datenbank.
|
||||
func (s *SourceStore) Provide(ctx context.Context) ([]Source, error) {
|
||||
rows, err := s.pool.Query(ctx, `SELECT module_name, metrics_url FROM metrics_sources ORDER BY module_name`)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("metrik-quellen auflisten: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var out []Source
|
||||
for rows.Next() {
|
||||
var src Source
|
||||
if err := rows.Scan(&src.ModuleName, &src.MetricsURL); err != nil {
|
||||
return nil, fmt.Errorf("metrik-quelle lesen: %w", err)
|
||||
}
|
||||
out = append(out, src)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
@@ -0,0 +1,60 @@
|
||||
package metrics
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
func setupSourceStoreTest(t *testing.T) (*SourceStore, func()) {
|
||||
t.Helper()
|
||||
adminDSN := os.Getenv("TEST_ADMIN_DSN")
|
||||
if adminDSN == "" {
|
||||
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
|
||||
}
|
||||
ctx := context.Background()
|
||||
pool, err := pgxpool.New(ctx, adminDSN)
|
||||
if err != nil {
|
||||
t.Fatalf("pool: %v", err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, `
|
||||
CREATE TABLE IF NOT EXISTS metrics_sources (module_name TEXT PRIMARY KEY, metrics_url TEXT NOT NULL)
|
||||
`); err != nil {
|
||||
t.Fatalf("schema: %v", err)
|
||||
}
|
||||
cleanup := func() { pool.Close() }
|
||||
return NewSourceStore(pool), cleanup
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2 / Pruefung 2 auf Persistenz-Ebene: eine ueber die
|
||||
// Datenbank registrierte Quelle ist sofort ueber Provide() sichtbar — genau
|
||||
// der Mechanismus, der ein neues Modul ohne Core-Codeaenderung erscheinen
|
||||
// laesst.
|
||||
func TestSourceStore_RegisterSourceAppearsInProvide(t *testing.T) {
|
||||
store, cleanup := setupSourceStoreTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
name := fmt.Sprintf("modul-%d", time.Now().UnixNano())
|
||||
|
||||
if err := store.RegisterSource(ctx, name, "http://example.invalid/metrics"); err != nil {
|
||||
t.Fatalf("registersource: %v", err)
|
||||
}
|
||||
|
||||
sources, err := store.Provide(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("provide: %v", err)
|
||||
}
|
||||
found := false
|
||||
for _, s := range sources {
|
||||
if s.ModuleName == name {
|
||||
found = true
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Fatalf("erwartet %s in provide()-ergebnis, habe: %+v", name, sources)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,81 @@
|
||||
// Package notify implementiert Core CFG-02: den zentralen Benachrichtigungs-
|
||||
// Dispatcher, ueber den beliebige Module Benachrichtigungen ausloesen —
|
||||
// Warteschlange, Wiederholungslogik, Kanal-Abstraktion. Die tatsaechlichen
|
||||
// Kanaele (E-Mail/In-App) sind CFG-03, hier gibt es nur die Sender-
|
||||
// Schnittstelle als Vorbereitung.
|
||||
package notify
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// DefaultMaxAttempts begrenzt Wiederholungsversuche (Akzeptanzkriterium 2) —
|
||||
// nach dieser Anzahl gibt der Dispatcher kontrolliert auf (status=failed)
|
||||
// statt endlos zu wiederholen.
|
||||
const DefaultMaxAttempts = 5
|
||||
|
||||
// DefaultRetryBackoff ist die Basis-Wartezeit zwischen Wiederholungen,
|
||||
// linear mit der Versuchsnummer skaliert.
|
||||
const DefaultRetryBackoff = 200 * time.Millisecond
|
||||
|
||||
type Notification struct {
|
||||
ID string
|
||||
Channel string
|
||||
Recipient string
|
||||
Payload map[string]any
|
||||
Attempts int
|
||||
}
|
||||
|
||||
// Sender ist die schmale Schnittstelle, die ein konkreter Kanal (CFG-03)
|
||||
// implementiert. Der Dispatcher selbst weiss nichts ueber E-Mail/In-App.
|
||||
type Sender interface {
|
||||
Send(ctx context.Context, n Notification) error
|
||||
}
|
||||
|
||||
// Dispatcher ist die EINE Schnittstelle, ueber die Module Benachrichtigungen
|
||||
// ausloesen — kein Modul baut eigenen Versandcode (Akzeptanzkriterium 1).
|
||||
type Dispatcher struct {
|
||||
pool *pgxpool.Pool
|
||||
maxAttempts int
|
||||
retryBackoff time.Duration
|
||||
}
|
||||
|
||||
func NewDispatcher(pool *pgxpool.Pool) *Dispatcher {
|
||||
return &Dispatcher{pool: pool, maxAttempts: DefaultMaxAttempts, retryBackoff: DefaultRetryBackoff}
|
||||
}
|
||||
|
||||
// WithRetryPolicy erlaubt Tests/Betrieb, Versuchsanzahl und Backoff
|
||||
// anzupassen, ohne die Default-Policy im Produktionscode zu veraendern.
|
||||
func (d *Dispatcher) WithRetryPolicy(maxAttempts int, backoff time.Duration) *Dispatcher {
|
||||
return &Dispatcher{pool: d.pool, maxAttempts: maxAttempts, retryBackoff: backoff}
|
||||
}
|
||||
|
||||
// Enqueue reiht eine Benachrichtigung in die Postgres-Warteschlange ein und
|
||||
// kehrt sofort zurueck — die Zeile ueberlebt jeden Neustart des Dispatcher-
|
||||
// Prozesses unveraendert (Akzeptanzkriterium 3), da sie ausschliesslich in
|
||||
// der Datenbank existiert, nicht im Prozessspeicher.
|
||||
func (d *Dispatcher) Enqueue(ctx context.Context, channel, recipient string, payload map[string]any) (string, error) {
|
||||
if payload == nil {
|
||||
payload = map[string]any{}
|
||||
}
|
||||
payloadJSON, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("payload serialisieren: %w", err)
|
||||
}
|
||||
|
||||
var id string
|
||||
err = d.pool.QueryRow(ctx, `
|
||||
INSERT INTO notification_jobs (channel, recipient, payload, max_attempts)
|
||||
VALUES ($1, $2, $3, $4)
|
||||
RETURNING id
|
||||
`, channel, recipient, payloadJSON, d.maxAttempts).Scan(&id)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("benachrichtigung einreihen: %w", err)
|
||||
}
|
||||
return id, nil
|
||||
}
|
||||
@@ -0,0 +1,219 @@
|
||||
package notify
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
func setupTest(t *testing.T) (*pgxpool.Pool, func()) {
|
||||
t.Helper()
|
||||
adminDSN := os.Getenv("TEST_ADMIN_DSN")
|
||||
if adminDSN == "" {
|
||||
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
|
||||
}
|
||||
ctx := context.Background()
|
||||
|
||||
pool, err := pgxpool.New(ctx, adminDSN)
|
||||
if err != nil {
|
||||
t.Fatalf("pool: %v", err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, `
|
||||
CREATE TABLE IF NOT EXISTS notification_jobs (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
channel TEXT NOT NULL,
|
||||
recipient TEXT NOT NULL,
|
||||
payload JSONB NOT NULL DEFAULT '{}'::jsonb,
|
||||
status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN ('pending', 'sent', 'failed')),
|
||||
attempts INT NOT NULL DEFAULT 0,
|
||||
max_attempts INT NOT NULL DEFAULT 5,
|
||||
next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
last_error TEXT,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
)`); err != nil {
|
||||
t.Fatalf("schema: %v", err)
|
||||
}
|
||||
|
||||
cleanup := func() { pool.Close() }
|
||||
return pool, cleanup
|
||||
}
|
||||
|
||||
type fakeSender struct {
|
||||
mu sync.Mutex
|
||||
sentIDs []string
|
||||
failUntil int
|
||||
calls int
|
||||
}
|
||||
|
||||
func (f *fakeSender) Send(ctx context.Context, n Notification) error {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.calls++
|
||||
if f.calls <= f.failUntil {
|
||||
return errors.New("simulierter zustellfehler")
|
||||
}
|
||||
f.sentIDs = append(f.sentIDs, n.ID)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *fakeSender) sentCount() int {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return len(f.sentIDs)
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 1: Module loesen ueber Enqueue aus, keine eigene
|
||||
// Versandlogik noetig.
|
||||
func TestDispatcher_EnqueueAndProcess(t *testing.T) {
|
||||
pool, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
d := NewDispatcher(pool)
|
||||
id, err := d.Enqueue(ctx, "email", "alice@example.com", map[string]any{"subject": "Willkommen"})
|
||||
if err != nil {
|
||||
t.Fatalf("enqueue: %v", err)
|
||||
}
|
||||
if id == "" {
|
||||
t.Fatal("erwartet nicht-leere id")
|
||||
}
|
||||
|
||||
sender := &fakeSender{}
|
||||
sent, failed, err := d.ProcessDue(ctx, sender, 10)
|
||||
if err != nil {
|
||||
t.Fatalf("process: %v", err)
|
||||
}
|
||||
if sent != 1 || failed != 0 {
|
||||
t.Fatalf("erwartet sent=1 failed=0, habe sent=%d failed=%d", sent, failed)
|
||||
}
|
||||
if sender.sentCount() != 1 {
|
||||
t.Fatalf("erwartet 1 zustellung, habe %d", sender.sentCount())
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2 + Pruefung 2: Wiederholungslogik greift bei
|
||||
// simuliertem Fehler und bricht nach definierter Anzahl kontrolliert ab.
|
||||
func TestProcessDue_RetriesThenGivesUpAfterMaxAttempts(t *testing.T) {
|
||||
pool, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
d := NewDispatcher(pool).WithRetryPolicy(3, time.Millisecond)
|
||||
id, err := d.Enqueue(ctx, "email", "bob@example.com", nil)
|
||||
if err != nil {
|
||||
t.Fatalf("enqueue: %v", err)
|
||||
}
|
||||
|
||||
sender := &fakeSender{failUntil: 100} // schlaegt bei jedem versuch fehl
|
||||
|
||||
for i := 0; i < 3; i++ {
|
||||
time.Sleep(5 * time.Millisecond) // next_attempt_at abwarten
|
||||
if _, _, err := d.ProcessDue(ctx, sender, 10); err != nil {
|
||||
t.Fatalf("process %d: %v", i, err)
|
||||
}
|
||||
}
|
||||
|
||||
var status string
|
||||
var attempts int
|
||||
if err := pool.QueryRow(ctx, `SELECT status, attempts FROM notification_jobs WHERE id = $1`, id).Scan(&status, &attempts); err != nil {
|
||||
t.Fatalf("status lesen: %v", err)
|
||||
}
|
||||
if status != "failed" {
|
||||
t.Fatalf("erwartet status failed nach max_attempts, habe %q", status)
|
||||
}
|
||||
if attempts != 3 {
|
||||
t.Fatalf("erwartet 3 versuche, habe %d", attempts)
|
||||
}
|
||||
|
||||
// Weiteres ProcessDue darf den bereits aufgegebenen job nicht mehr anfassen.
|
||||
sent, failed, err := d.ProcessDue(ctx, sender, 10)
|
||||
if err != nil {
|
||||
t.Fatalf("process nach abbruch: %v", err)
|
||||
}
|
||||
if sent != 0 || failed != 0 {
|
||||
t.Fatalf("erwartet keine weitere verarbeitung, habe sent=%d failed=%d", sent, failed)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 3 + Pruefung 1: Neustart des Dienstes waehrend offener
|
||||
// Zustellung verliert keine Nachricht — simuliert durch eine komplett neue
|
||||
// Dispatcher/Pool-Instanz nach dem Enqueue, bevor irgendetwas verarbeitet wurde.
|
||||
func TestQueue_SurvivesRestartWithoutMessageLoss(t *testing.T) {
|
||||
pool, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
firstInstance := NewDispatcher(pool)
|
||||
id, err := firstInstance.Enqueue(ctx, "email", "carol@example.com", nil)
|
||||
if err != nil {
|
||||
t.Fatalf("enqueue: %v", err)
|
||||
}
|
||||
|
||||
// "Neustart": eine voellig neue Dispatcher-Instanz (repraesentiert einen
|
||||
// neuen Prozess) verbindet sich neu und verarbeitet die Warteschlange —
|
||||
// die Nachricht existiert ausschliesslich in Postgres, nicht im
|
||||
// Prozessspeicher der ersten Instanz.
|
||||
restartedInstance := NewDispatcher(pool)
|
||||
sender := &fakeSender{}
|
||||
sent, failed, err := restartedInstance.ProcessDue(ctx, sender, 10)
|
||||
if err != nil {
|
||||
t.Fatalf("process nach neustart: %v", err)
|
||||
}
|
||||
if sent != 1 || failed != 0 {
|
||||
t.Fatalf("erwartet sent=1 nach neustart, habe sent=%d failed=%d", sent, failed)
|
||||
}
|
||||
if len(sender.sentIDs) != 1 || sender.sentIDs[0] != id {
|
||||
t.Fatalf("erwartet zustellung der urspruenglichen nachricht %q, habe %v", id, sender.sentIDs)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 3 + Pruefung 3: zwei gleichzeitig ausloesende Module,
|
||||
// beide Nachrichten werden korrekt (und nicht doppelt) zugestellt.
|
||||
func TestProcessDue_ConcurrentDispatchBothDelivered(t *testing.T) {
|
||||
pool, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
d := NewDispatcher(pool)
|
||||
idA, err := d.Enqueue(ctx, "email", "modul-a@example.com", nil)
|
||||
if err != nil {
|
||||
t.Fatalf("enqueue a: %v", err)
|
||||
}
|
||||
idB, err := d.Enqueue(ctx, "email", "modul-b@example.com", nil)
|
||||
if err != nil {
|
||||
t.Fatalf("enqueue b: %v", err)
|
||||
}
|
||||
|
||||
sender := &fakeSender{}
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < 2; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
if _, _, err := d.ProcessDue(ctx, sender, 10); err != nil {
|
||||
t.Errorf("process: %v", err)
|
||||
}
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
if sender.sentCount() != 2 {
|
||||
t.Fatalf("erwartet genau 2 zustellungen, habe %d: %v", sender.sentCount(), sender.sentIDs)
|
||||
}
|
||||
seen := map[string]bool{}
|
||||
for _, id := range sender.sentIDs {
|
||||
if seen[id] {
|
||||
t.Fatalf("nachricht %q wurde doppelt zugestellt", id)
|
||||
}
|
||||
seen[id] = true
|
||||
}
|
||||
if !seen[idA] || !seen[idB] {
|
||||
t.Fatalf("erwartet beide nachrichten zugestellt, habe %v", sender.sentIDs)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,102 @@
|
||||
package notify
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
)
|
||||
|
||||
// ProcessDue holt bis zu limit faellige Benachrichtigungen und versucht sie
|
||||
// ueber sender zuzustellen. FOR UPDATE SKIP LOCKED serialisiert konkurrierende
|
||||
// Aufrufe (Akzeptanzkriterium 3 / Pruefung 3: zwei gleichzeitig ausloesende
|
||||
// Module duerfen sich nicht gegenseitig blockieren oder Nachrichten doppelt
|
||||
// zustellen) — dieselbe Konvention wie internal/tenant.Lifecycle.ProcessDueDeletions.
|
||||
func (d *Dispatcher) ProcessDue(ctx context.Context, sender Sender, limit int) (sent, failed int, err error) {
|
||||
tx, err := d.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return 0, 0, fmt.Errorf("transaktion starten: %w", err)
|
||||
}
|
||||
defer func() { _ = tx.Rollback(ctx) }()
|
||||
|
||||
rows, err := tx.Query(ctx, `
|
||||
SELECT id, channel, recipient, payload, attempts, max_attempts
|
||||
FROM notification_jobs
|
||||
WHERE status = 'pending' AND next_attempt_at <= now()
|
||||
ORDER BY created_at
|
||||
FOR UPDATE SKIP LOCKED
|
||||
LIMIT $1
|
||||
`, limit)
|
||||
if err != nil {
|
||||
return 0, 0, fmt.Errorf("faellige benachrichtigungen abfragen: %w", err)
|
||||
}
|
||||
|
||||
type due struct {
|
||||
id, channel, recipient string
|
||||
payload []byte
|
||||
attempts, maxAttempts int
|
||||
}
|
||||
var candidates []due
|
||||
for rows.Next() {
|
||||
var c due
|
||||
if err := rows.Scan(&c.id, &c.channel, &c.recipient, &c.payload, &c.attempts, &c.maxAttempts); err != nil {
|
||||
rows.Close()
|
||||
return 0, 0, fmt.Errorf("faellige benachrichtigung lesen: %w", err)
|
||||
}
|
||||
candidates = append(candidates, c)
|
||||
}
|
||||
rows.Close()
|
||||
if err := rows.Err(); err != nil {
|
||||
return 0, 0, err
|
||||
}
|
||||
|
||||
for _, c := range candidates {
|
||||
var payload map[string]any
|
||||
if err := json.Unmarshal(c.payload, &payload); err != nil {
|
||||
payload = map[string]any{}
|
||||
}
|
||||
|
||||
sendErr := sender.Send(ctx, Notification{
|
||||
ID: c.id, Channel: c.channel, Recipient: c.recipient, Payload: payload, Attempts: c.attempts,
|
||||
})
|
||||
|
||||
if sendErr == nil {
|
||||
if _, err := tx.Exec(ctx, `
|
||||
UPDATE notification_jobs SET status = 'sent', updated_at = now() WHERE id = $1
|
||||
`, c.id); err != nil {
|
||||
return sent, failed, fmt.Errorf("erfolg speichern: %w", err)
|
||||
}
|
||||
sent++
|
||||
continue
|
||||
}
|
||||
|
||||
newAttempts := c.attempts + 1
|
||||
if newAttempts >= c.maxAttempts {
|
||||
// Akzeptanzkriterium 2: kontrollierter Abbruch nach definierter
|
||||
// Anzahl Versuche, kein endloses Wiederholen.
|
||||
if _, err := tx.Exec(ctx, `
|
||||
UPDATE notification_jobs
|
||||
SET status = 'failed', attempts = $2, last_error = $3, updated_at = now()
|
||||
WHERE id = $1
|
||||
`, c.id, newAttempts, sendErr.Error()); err != nil {
|
||||
return sent, failed, fmt.Errorf("fehlschlag speichern: %w", err)
|
||||
}
|
||||
failed++
|
||||
continue
|
||||
}
|
||||
|
||||
nextAttempt := time.Now().Add(time.Duration(newAttempts) * d.retryBackoff)
|
||||
if _, err := tx.Exec(ctx, `
|
||||
UPDATE notification_jobs
|
||||
SET attempts = $2, next_attempt_at = $3, last_error = $4, updated_at = now()
|
||||
WHERE id = $1
|
||||
`, c.id, newAttempts, nextAttempt, sendErr.Error()); err != nil {
|
||||
return sent, failed, fmt.Errorf("wiederholung planen: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return 0, 0, fmt.Errorf("transaktion committen: %w", err)
|
||||
}
|
||||
return sent, failed, nil
|
||||
}
|
||||
@@ -0,0 +1,2 @@
|
||||
DROP TABLE IF EXISTS config_value_history;
|
||||
DROP TABLE IF EXISTS config_values;
|
||||
@@ -0,0 +1,23 @@
|
||||
-- Zentraler Konfigurationsdienst (CFG-01, siehe core-kanban/tickets/CFG-01.md).
|
||||
-- scope = 'global' fuer globale Defaults, sonst der Tenant-Slug. config_values
|
||||
-- haelt den AKTUELLEN Stand je (key, scope); config_value_history haelt JEDE
|
||||
-- Aenderung fest (Akzeptanzkriterium 2: versioniert nachvollziehbar).
|
||||
CREATE TABLE config_values (
|
||||
key TEXT NOT NULL,
|
||||
scope TEXT NOT NULL CHECK (scope <> ''),
|
||||
value TEXT NOT NULL,
|
||||
version INT NOT NULL,
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
PRIMARY KEY (key, scope)
|
||||
);
|
||||
|
||||
CREATE TABLE config_value_history (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
key TEXT NOT NULL,
|
||||
scope TEXT NOT NULL,
|
||||
value TEXT NOT NULL,
|
||||
version INT NOT NULL,
|
||||
changed_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
CREATE INDEX config_value_history_key_scope_idx ON config_value_history (key, scope, version);
|
||||
@@ -0,0 +1 @@
|
||||
DROP TABLE IF EXISTS notification_jobs;
|
||||
@@ -0,0 +1,20 @@
|
||||
-- Benachrichtigungs-Dispatcher-Warteschlange (CFG-02, siehe
|
||||
-- core-kanban/tickets/CFG-02.md). Postgres-basiert statt Redis/AMQP
|
||||
-- (Projekt-Konvention, siehe nexarch-state.json techstack.job_queue) —
|
||||
-- Zeilen ueberleben einen Neustart des Dispatcher-Prozesses unveraendert
|
||||
-- (Akzeptanzkriterium 3).
|
||||
CREATE TABLE notification_jobs (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
channel TEXT NOT NULL,
|
||||
recipient TEXT NOT NULL,
|
||||
payload JSONB NOT NULL DEFAULT '{}'::jsonb,
|
||||
status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN ('pending', 'sent', 'failed')),
|
||||
attempts INT NOT NULL DEFAULT 0,
|
||||
max_attempts INT NOT NULL DEFAULT 5,
|
||||
next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
last_error TEXT,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
CREATE INDEX notification_jobs_due_idx ON notification_jobs (status, next_attempt_at);
|
||||
@@ -0,0 +1,2 @@
|
||||
DROP TABLE alert_debounce_state;
|
||||
DROP TABLE alert_rules;
|
||||
@@ -0,0 +1,24 @@
|
||||
-- OPS-05: Schwellwert-Regeln fuer Alerting auf den aus OPS-03 aggregierten
|
||||
-- Metriken. Lebt wie config_values/notification_jobs (CFG-01/02) in der
|
||||
-- zentralen Registry-DB — modulübergreifende Betriebskonfiguration, keine
|
||||
-- Mandanten-Geschaeftsdaten.
|
||||
CREATE TABLE alert_rules (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
metric_name TEXT NOT NULL,
|
||||
comparison TEXT NOT NULL CHECK (comparison IN ('gt', 'lt')),
|
||||
threshold DOUBLE PRECISION NOT NULL,
|
||||
label_filters JSONB NOT NULL DEFAULT '{}'::jsonb,
|
||||
recipient TEXT NOT NULL,
|
||||
description TEXT NOT NULL DEFAULT '',
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
-- Haelt fest, wann eine Regel zuletzt tatsaechlich einen Alarm ausgeloest
|
||||
-- hat (Akzeptanzkriterium 3: Drosselung wiederholter Alarmierung fuer
|
||||
-- denselben anhaltenden Zustand). rule_key kombiniert Regel-ID mit den
|
||||
-- tatsaechlichen Label-Werten der ausloesenden Zeitreihe, damit dieselbe
|
||||
-- Regel fuer unterschiedliche Tenants/Module unabhaengig gedrosselt wird.
|
||||
CREATE TABLE alert_debounce_state (
|
||||
rule_key TEXT PRIMARY KEY,
|
||||
last_fired_at TIMESTAMPTZ NOT NULL
|
||||
);
|
||||
@@ -0,0 +1 @@
|
||||
DROP TABLE metrics_sources;
|
||||
@@ -0,0 +1,7 @@
|
||||
-- Metrics-Aggregation ueber Module hinweg (OPS-03, siehe
|
||||
-- core-kanban/tickets/OPS-03.md) — welches Modul liefert seine Kennzahlen
|
||||
-- unter welcher /metrics-URL.
|
||||
CREATE TABLE metrics_sources (
|
||||
module_name TEXT PRIMARY KEY,
|
||||
metrics_url TEXT NOT NULL
|
||||
);
|
||||
@@ -1 +0,0 @@
|
||||
DROP TABLE tenant_keks;
|
||||
@@ -1,10 +0,0 @@
|
||||
-- Master-Key-Verwaltung & Tenant-Schluesselhierarchie (API-10, siehe
|
||||
-- core-kanban/tickets/API-10.md) — EIN verpackter (mit dem Master-Key
|
||||
-- umhuellter) Tenant-KEK je Mandant. Niemals der Master-Key selbst und
|
||||
-- niemals ein Tenant-KEK im Klartext in dieser Tabelle.
|
||||
CREATE TABLE tenant_keks (
|
||||
tenant_id UUID PRIMARY KEY REFERENCES tenants(id),
|
||||
wrapped_kek BYTEA NOT NULL,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
rotated_at TIMESTAMPTZ
|
||||
);
|
||||
@@ -14,6 +14,9 @@ ROLE="nexarch_test"
|
||||
export PGPASSWORD="$PASS"
|
||||
|
||||
psql -h localhost -U "$ROLE" -d postgres -v ON_ERROR_STOP=1 -c "DROP TABLE IF EXISTS tenants CASCADE;"
|
||||
psql -h localhost -U "$ROLE" -d postgres -v ON_ERROR_STOP=1 -c "DROP TABLE IF EXISTS audit_events CASCADE;"
|
||||
psql -h localhost -U "$ROLE" -d postgres -v ON_ERROR_STOP=1 -c "DROP TABLE IF EXISTS config_value_history CASCADE;"
|
||||
psql -h localhost -U "$ROLE" -d postgres -v ON_ERROR_STOP=1 -c "DROP TABLE IF EXISTS config_values CASCADE;"
|
||||
|
||||
dbs=$(psql -h localhost -U "$ROLE" -d postgres -tAc "SELECT datname FROM pg_database WHERE datname LIKE 'tenant\_%' ESCAPE '\'")
|
||||
for db in $dbs; do
|
||||
|
||||
Reference in New Issue
Block a user