scripte/nexarch
1
0 Fork 0
Code Issues 1 Pull Requests 6 Actions Packages Projects Releases Wiki Activity

dms: S3-Treiber in dms-app und dms-worker waehlbar #431

Open
patrick wants to merge 1 commits from feat/dms-s3-verdrahten into main
pull from: feat/dms-s3-verdrahten
merge into: scripte:main
Conversation 0 Commits 1 Files Changed 6
+129 -28
6 changed files with 129 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

dms/README.md
+4
View File
@@ -29,6 +29,10 @@ Fachliche Routen (z. B. `POST /retention/destroy-callback`) liegen unter dem Pr
`scripts/check-route-prefixes.sh`); nur `/healthz` und `/metrics` bleiben auf Wurzelebene. Die bei archive
registrierte Rückruf-URL (`NEXARCH_DMS_APP_RETENTION_CALLBACK_URL`) muss den Präfix enthalten.
## Objekt-Speicher
`NEXARCH_DMS_APP_STORAGE_DRIVER` und `NEXARCH_DMS_WORKER_STORAGE_DRIVER` wählen den Treiber: `local` (Standard, braucht `…_STORAGE_DIR`) oder `s3`. Für `s3` sind `…_S3_BUCKET`, `…_S3_ACCESS_KEY_ID` und `…_S3_SECRET_ACCESS_KEY` Pflicht; optional `…_S3_REGION` (Standard `us-east-1`), `…_S3_ENDPOINT_URL` (MinIO, Ceph) und `…_S3_PATH_STYLE=1`. App und Worker müssen denselben Bucket nutzen.
## Struktur
- `cmd/app` — Anfrage-Dienst (HTTP), blockiert nie durch lange Aufgaben
dms/cmd/app/api.go
+4 -4
View File
@@ -41,10 +41,10 @@ type uploadStack struct {
}
func newUploadStack(pool *pgxpool.Pool, stagingDir string) uploadStack {
storageDir := os.Getenv("NEXARCH_DMS_APP_STORAGE_DIR")
signingSecret := os.Getenv("NEXARCH_DMS_APP_STORAGE_SIGNING_SECRET")
publicBaseURL := os.Getenv("NEXARCH_DMS_APP_STORAGE_PUBLIC_BASE_URL")
driver := storage.NewLocalDriver(storageDir, []byte(signingSecret), publicBaseURL, signedURLMaxTTL(os.Getenv("NEXARCH_DMS_APP_SIGNED_URL_MAX_TTL")))
driver, err := storage.NewDriverFromEnv(context.Background(), "NEXARCH_DMS_APP")
if err != nil {
log.Fatalf("dms-app: %v", err)
}
usageReporter := storage.NewHTTPUsageReporter(
os.Getenv("NEXARCH_DMS_APP_USAGE_ENDPOINT_URL"),
os.Getenv("NEXARCH_DMS_APP_USAGE_CLIENT_ID"),
dms/cmd/app/main.go
+4 -12
View File
@@ -194,10 +194,10 @@ func setupRetention(mux *http.ServeMux, dsn string) {
log.Fatalf("dms-app: datenbankverbindung: %v", err)
}
storageDir := os.Getenv("NEXARCH_DMS_APP_STORAGE_DIR")
signingSecret := os.Getenv("NEXARCH_DMS_APP_STORAGE_SIGNING_SECRET")
publicBaseURL := os.Getenv("NEXARCH_DMS_APP_STORAGE_PUBLIC_BASE_URL")
driver := storage.NewLocalDriver(storageDir, []byte(signingSecret), publicBaseURL, signedURLMaxTTL(os.Getenv("NEXARCH_DMS_APP_SIGNED_URL_MAX_TTL")))
driver, err := storage.NewDriverFromEnv(ctx, "NEXARCH_DMS_APP")
if err != nil {
log.Fatalf("dms-app: %v", err)
}
usageEndpoint := os.Getenv("NEXARCH_DMS_APP_USAGE_ENDPOINT_URL")
usageClientID := os.Getenv("NEXARCH_DMS_APP_USAGE_CLIENT_ID")
@@ -360,11 +360,3 @@ func setupWatchFolder(ctx context.Context, dir string) {
}
}()
}
func signedURLMaxTTL(raw string) time.Duration {
ttl, err := storage.ParseMaxSignedURLTTL(raw)
if err != nil {
log.Fatalf("dms-app: %v", err)
}
return ttl
}
dms/cmd/worker/main.go
+6 -12
View File
@@ -133,14 +133,16 @@ func processRetentionRegisterJobs(ctx context.Context, queue *jobqueue.Queue, cl
// startJobWorkers verdrahtet alle Fachjobs (FDN-17). Fehlt die Storage-/Schluessel-
// Konfiguration, werden sie nicht gestartet und das wird gemeldet.
func startJobWorkers(ctx context.Context, pool *pgxpool.Pool, queue *jobqueue.Queue, tracker *errortracking.Tracker) {
storageDir := os.Getenv("NEXARCH_DMS_WORKER_STORAGE_DIR")
tenantSlug := os.Getenv("NEXARCH_DMS_WORKER_TENANT_SLUG")
kekURL := os.Getenv("NEXARCH_DMS_WORKER_KEK_ENDPOINT_URL")
if storageDir == "" || tenantSlug == "" || kekURL == "" {
log.Print("dms-worker: NEXARCH_DMS_WORKER_STORAGE_DIR/_TENANT_SLUG/_KEK_ENDPOINT_URL unvollstaendig, Fachjobs (Vorschau, Konvertierung, Virenscan, Export, Texterkennung) werden NICHT verarbeitet")
if !storage.ConfiguredFromEnv("NEXARCH_DMS_WORKER") || tenantSlug == "" || kekURL == "" {
log.Print("dms-worker: NEXARCH_DMS_WORKER_STORAGE_DIR (oder _STORAGE_DRIVER=s3 mit _S3_BUCKET)/_TENANT_SLUG/_KEK_ENDPOINT_URL unvollstaendig, Fachjobs (Vorschau, Konvertierung, Virenscan, Export, Texterkennung) werden NICHT verarbeitet")
return
}
driver := storage.NewLocalDriver(storageDir, []byte(os.Getenv("NEXARCH_DMS_WORKER_STORAGE_SIGNING_SECRET")), os.Getenv("NEXARCH_DMS_WORKER_STORAGE_PUBLIC_BASE_URL"), signedURLMaxTTL(os.Getenv("NEXARCH_DMS_WORKER_SIGNED_URL_MAX_TTL")))
driver, err := storage.NewDriverFromEnv(ctx, "NEXARCH_DMS_WORKER")
if err != nil {
log.Fatalf("dms-worker: %v", err)
}
usage := storage.NewHTTPUsageReporter(os.Getenv("NEXARCH_DMS_WORKER_USAGE_ENDPOINT_URL"), os.Getenv("NEXARCH_DMS_WORKER_USAGE_CLIENT_ID"), os.Getenv("NEXARCH_DMS_WORKER_USAGE_CLIENT_SECRET"), nil)
cryptoSvc := dmscrypto.NewService(dmscrypto.NewHTTPKEKProvider(kekURL, os.Getenv("NEXARCH_DMS_WORKER_KEK_CLIENT_ID"), os.Getenv("NEXARCH_DMS_WORKER_KEK_CLIENT_SECRET"), nil))
@@ -304,11 +306,3 @@ func startJobWorkers(ctx context.Context, pool *pgxpool.Pool, queue *jobqueue.Qu
go integrity.RunScheduler(ctx, pool, queue, integritySched, 30*time.Second)
}
}
func signedURLMaxTTL(raw string) time.Duration {
ttl, err := storage.ParseMaxSignedURLTTL(raw)
if err != nil {
log.Fatalf("dms-worker: %v", err)
}
return ttl
}
dms/internal/storage/fromenv.go
+42
View File
@@ -0,0 +1,42 @@
package storage
import (
"context"
"fmt"
"os"
"strings"
)
// NewDriverFromEnv wählt den Treiber über <prefix>_STORAGE_DRIVER ("local",
// Standard, oder "s3"). Zugangsdaten kommen nur aus der Umgebung.
func NewDriverFromEnv(ctx context.Context, prefix string) (Driver, error) {
env := func(name string) string { return os.Getenv(prefix + "_" + name) }
ttl, err := ParseMaxSignedURLTTL(env("SIGNED_URL_MAX_TTL"))
if err != nil {
return nil, err
}
switch strings.ToLower(env("STORAGE_DRIVER")) {
case "", "local":
return NewLocalDriver(env("STORAGE_DIR"), []byte(env("STORAGE_SIGNING_SECRET")), env("STORAGE_PUBLIC_BASE_URL"), ttl), nil
case "s3":
bucket, key, secret := env("S3_BUCKET"), env("S3_ACCESS_KEY_ID"), env("S3_SECRET_ACCESS_KEY")
if bucket == "" || key == "" || secret == "" {
return nil, fmt.Errorf("storage: %s_S3_BUCKET, _S3_ACCESS_KEY_ID und _S3_SECRET_ACCESS_KEY sind erforderlich", prefix)
}
region := env("S3_REGION")
if region == "" {
region = "us-east-1"
}
return NewS3Driver(ctx, bucket, region, env("S3_ENDPOINT_URL"), key, secret, env("S3_PATH_STYLE") == "1", ttl)
default:
return nil, fmt.Errorf("storage: %s_STORAGE_DRIVER %q unbekannt (local oder s3)", prefix, env("STORAGE_DRIVER"))
}
}
// ConfiguredFromEnv meldet, ob für prefix ein Speicher konfiguriert ist.
func ConfiguredFromEnv(prefix string) bool {
if strings.EqualFold(os.Getenv(prefix+"_STORAGE_DRIVER"), "s3") {
return os.Getenv(prefix+"_S3_BUCKET") != ""
}
return os.Getenv(prefix+"_STORAGE_DIR") != ""
}
dms/internal/storage/fromenv_test.go
+69
View File
@@ -0,0 +1,69 @@
package storage
import (
"context"
"testing"
)
func TestNewDriverFromEnv(t *testing.T) {
ctx := context.Background()
t.Run("Standard ist lokal", func(t *testing.T) {
t.Setenv("X_STORAGE_DIR", t.TempDir())
d, err := NewDriverFromEnv(ctx, "X")
if err != nil {
t.Fatal(err)
}
if _, ok := d.(*LocalDriver); !ok {
t.Fatalf("erwartet *LocalDriver, bekam %T", d)
}
})
t.Run("s3 mit Zugangsdaten", func(t *testing.T) {
t.Setenv("X_STORAGE_DRIVER", "S3")
t.Setenv("X_S3_BUCKET", "b")
t.Setenv("X_S3_ACCESS_KEY_ID", "k")
t.Setenv("X_S3_SECRET_ACCESS_KEY", "s")
t.Setenv("X_S3_ENDPOINT_URL", "http://127.0.0.1:9000")
d, err := NewDriverFromEnv(ctx, "X")
if err != nil {
t.Fatal(err)
}
if _, ok := d.(*S3Driver); !ok {
t.Fatalf("erwartet *S3Driver, bekam %T", d)
}
})
t.Run("s3 ohne Zugangsdaten schlaegt fehl", func(t *testing.T) {
t.Setenv("X_STORAGE_DRIVER", "s3")
t.Setenv("X_S3_BUCKET", "b")
if _, err := NewDriverFromEnv(ctx, "X"); err == nil {
t.Fatal("erwartet Fehler ohne Schluessel")
}
})
t.Run("unbekannter Treiber", func(t *testing.T) {
t.Setenv("X_STORAGE_DRIVER", "ftp")
if _, err := NewDriverFromEnv(ctx, "X"); err == nil {
t.Fatal("erwartet Fehler fuer unbekannten Treiber")
}
})
}
func TestConfiguredFromEnv(t *testing.T) {
if ConfiguredFromEnv("Y") {
t.Fatal("ohne Variablen nicht konfiguriert")
}
t.Setenv("Y_STORAGE_DIR", "/x")
if !ConfiguredFromEnv("Y") {
t.Fatal("STORAGE_DIR zaehlt als konfiguriert")
}
t.Setenv("Y_STORAGE_DRIVER", "s3")
if ConfiguredFromEnv("Y") {
t.Fatal("s3 braucht den Bucket")
}
t.Setenv("Y_S3_BUCKET", "b")
if !ConfiguredFromEnv("Y") {
t.Fatal("s3 mit Bucket ist konfiguriert")
}
}