From 137c13fa98959901aa39e8ce958719cf7de1f9cd Mon Sep 17 00:00:00 2001 From: Philipp Date: Thu, 11 Jun 2026 14:48:50 +0200 Subject: [PATCH] feat: add proxmox reconciliation job --- CHANGELOG.md | 1 + README.md | 3 +- TODO.md | 9 +- platform/jobs/proxmox_task.go | 1 + platform/proxmox/client.go | 35 +++ platform/proxmox/client_test.go | 19 ++ worker/cmd/worker/main.go | 24 ++ worker/internal/tasks/proxmox_reconcile.go | 211 ++++++++++++++++++ .../tasks/proxmox_reconcile_resolver.go | 62 +++++ .../internal/tasks/proxmox_reconcile_test.go | 124 ++++++++++ 10 files changed, 487 insertions(+), 2 deletions(-) create mode 100644 worker/internal/tasks/proxmox_reconcile.go create mode 100644 worker/internal/tasks/proxmox_reconcile_resolver.go create mode 100644 worker/internal/tasks/proxmox_reconcile_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 72f2fe5..b710a9d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## Unreleased +- Proxmox-Reconciliation-Job fuer aktive Cluster angelegt; Drift wird report-only als Audit-Event geloggt. - VM-Power-Endpunkte fuer start/stop/reboot mit Proxmox-UPID, Queue-Enqueue und Audit-Write angelegt. - VM-Read-Endpunkte fuer Projekt-Listen und VM-Details mit Membership-gefilterten SQL-Queries angelegt. - Cluster-, Krypto- und Proxmox-Client-Bausteine nach `platform/` verschoben und im Worker fuer echte UPID-Polls verdrahtet. diff --git a/README.md b/README.md index 4770030..f1ef0b6 100644 --- a/README.md +++ b/README.md @@ -40,7 +40,7 @@ Der Proxmox-Client nutzt HTTPS mit normaler Zertifikatsvalidierung plus SHA-256- Interne Cluster-Admin-Routen unter `/internal/clusters` sind nicht tenant-gebunden. Sie erfordern den Header `X-ProxUI-Operator-Token` mit `OPERATOR_TOKEN` und die isolierte Rolle `operator`; Responses enthalten nie `token_secret`. -Der Worker nutzt `asynq` gegen `REDIS_ADDR`, pingt beim Start dieselbe Supabase-DB ueber `DATABASE_URL`, entschluesselt Cluster-Credentials mit `MASTER_KEY_BASE64` und nutzt den gemeinsamen Proxmox-Client aus `platform/proxmox` fuer UPID-Polls. +Der Worker nutzt `asynq` gegen `REDIS_ADDR`, pingt beim Start dieselbe Supabase-DB ueber `DATABASE_URL`, entschluesselt Cluster-Credentials mit `MASTER_KEY_BASE64` und nutzt den gemeinsamen Proxmox-Client aus `platform/proxmox` fuer UPID-Polls und Reconciliation. Lokale Dienste: @@ -66,6 +66,7 @@ Worker-Jobs: - `proxui.dummy`: Dummy-Job mit Payload `{"message":"..."}` fuer Smoke-Tests der asynq-Verarbeitung - `proxmox.task.poll`: Pollt einen Proxmox-UPID ueber den gespeicherten Cluster, setzt VM-Status auf Erfolgs- oder Fehlerstatus und schreibt Audit-Metadaten. +- `proxmox.reconcile.all`: Laeuft alle 5 Minuten, prueft aktive Cluster gegen bekannte VMs und schreibt Drift als Audit-Event `vm.reconcile.drift`. Backend-Endpunkte: diff --git a/TODO.md b/TODO.md index bc47606..d1d4fd8 100644 --- a/TODO.md +++ b/TODO.md @@ -151,6 +151,12 @@ Arbeitsliste auf Basis von `proxmox-console-entwicklungsplan.md`. Die Entwurfsda - [x] `vm.power` wird anhand der Membership-Rolle geprueft - [x] Proxmox-Call liefert UPID und queued `proxmox.task.poll` - [x] Audit-Log-Eintrag fuer ausgeloeste Power-Aktion wird geschrieben +- [x] E5-T03: Reconciliation-Job + - [x] Periodischen asynq-Scheduler fuer `proxmox.reconcile.all` angelegt + - [x] Aktive Cluster werden aus der DB geladen + - [x] Bekannte VMs je Cluster werden gegen Proxmox `status/current` geprueft + - [x] Drift wird als Audit-Event `vm.reconcile.drift` geloggt + - [x] Kein Auto-Fix: DB-Status wird im Reconciliation-Lauf nicht ueberschrieben ## MVP-Backlog @@ -159,7 +165,7 @@ Arbeitsliste auf Basis von `proxmox-console-entwicklungsplan.md`. Die Entwurfsda - [x] E3: Envelope-Encryption, Cluster-Repository, Proxmox-Client und interne Cluster-Verwaltung - [ ] E4: RBAC, Membership-Resolver, Policy-Funktion und Autorisierungs-Middleware - [ ] E6: Worker-Grundgeruest und UPID-Polling -- [ ] E5: VM-Liste, Detail, Power-Aktionen und Reconciliation +- [x] E5: VM-Liste, Detail, Power-Aktionen und Reconciliation - [ ] E7: SSH-Keys, Templates und Provisioning - [ ] E8: Konsolen-Tickets und Websocket-Proxy - [ ] E9: Audit-Writer und Audit-Anzeige @@ -213,3 +219,4 @@ Arbeitsliste auf Basis von `proxmox-console-entwicklungsplan.md`. Die Entwurfsda - 2026-06-11: Shared Cluster-/Krypto-/Proxmox-Packages nach `platform/` verschoben und Worker-UPID-Polling an echte Cluster-Aufloesung angeschlossen. - 2026-06-11: VM-Read-Endpunkte fuer Projekt-Listen und Details mit Membership-gefilterten Queries angelegt. - 2026-06-11: VM-Power-Endpunkte fuer start/stop/reboot mit Proxmox-UPID, Queue-Enqueue und Audit-Write angelegt. +- 2026-06-11: Reconciliation-Job fuer aktive Proxmox-Cluster angelegt; Drift wird report-only auditiert. diff --git a/platform/jobs/proxmox_task.go b/platform/jobs/proxmox_task.go index dae2414..449928a 100644 --- a/platform/jobs/proxmox_task.go +++ b/platform/jobs/proxmox_task.go @@ -1,6 +1,7 @@ package jobs const TypeProxmoxTaskPoll = "proxmox.task.poll" +const TypeProxmoxReconcileAll = "proxmox.reconcile.all" type ProxmoxTaskPollPayload struct { ClusterID string `json:"cluster_id"` diff --git a/platform/proxmox/client.go b/platform/proxmox/client.go index cc1870c..22ad3ba 100644 --- a/platform/proxmox/client.go +++ b/platform/proxmox/client.go @@ -131,6 +131,10 @@ type TaskStatus struct { ExitStatus string } +type VMStatus struct { + Status string +} + func (c *Client) GetTaskStatus(ctx context.Context, node string, upid string) (TaskStatus, error) { response, err := c.Get(ctx, fmt.Sprintf( "/nodes/%s/tasks/%s/status", @@ -193,6 +197,37 @@ func (c *Client) PowerVM(ctx context.Context, node string, vmid int, action stri return body.Data, nil } +func (c *Client) GetVMStatus(ctx context.Context, node string, vmid int) (VMStatus, error) { + response, err := c.Get(ctx, fmt.Sprintf( + "/nodes/%s/qemu/%d/status/current", + url.PathEscape(node), + vmid, + )) + if err != nil { + return VMStatus{}, err + } + defer response.Body.Close() + + if response.StatusCode >= http.StatusBadRequest { + _, _ = io.Copy(io.Discard, response.Body) + return VMStatus{}, fmt.Errorf("proxmox returned %s", response.Status) + } + + var body struct { + Data struct { + Status string `json:"status"` + } `json:"data"` + } + if err := json.NewDecoder(response.Body).Decode(&body); err != nil { + return VMStatus{}, err + } + if body.Data.Status == "" { + return VMStatus{}, fmt.Errorf("proxmox response missing VM status") + } + + return VMStatus{Status: body.Data.Status}, nil +} + func (c *Client) Do(ctx context.Context, method string, path string, body []byte) (*http.Response, error) { var lastErr error attempts := c.retries + 1 diff --git a/platform/proxmox/client_test.go b/platform/proxmox/client_test.go index cbbe8f1..28ce142 100644 --- a/platform/proxmox/client_test.go +++ b/platform/proxmox/client_test.go @@ -130,6 +130,25 @@ func TestClientPowersVM(t *testing.T) { } } +func TestClientGetsVMStatus(t *testing.T) { + server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api2/json/nodes/pve/qemu/100/status/current" { + t.Fatalf("path = %q", r.URL.Path) + } + _, _ = w.Write([]byte(`{"data":{"status":"running"}}`)) + })) + defer server.Close() + + client := newTestClient(t, server, fingerprintForServer(server)) + status, err := client.GetVMStatus(context.Background(), "pve", 100) + if err != nil { + t.Fatalf("GetVMStatus() error = %v", err) + } + if status.Status != "running" { + t.Fatalf("status = %q, want running", status.Status) + } +} + func TestNewClientRejectsInvalidFingerprint(t *testing.T) { _, err := NewClient(cluster.Cluster{ APIEndpoint: "https://pve.example.test:8006/api2/json", diff --git a/worker/cmd/worker/main.go b/worker/cmd/worker/main.go index e5cafab..c80e37f 100644 --- a/worker/cmd/worker/main.go +++ b/worker/cmd/worker/main.go @@ -50,9 +50,14 @@ func main() { server := asynq.NewServer(redis, asynq.Config{ Concurrency: cfg.WorkerConcurrency, }) + scheduler := asynq.NewScheduler(redis, &asynq.SchedulerOpts{}) mux := asynq.NewServeMux() clusterRepository := cluster.NewRepository(db, tokenCipher) registerHandlers(mux, db, clusterRepository, logger) + if _, err := scheduler.Register("@every 5m", tasks.NewProxmoxReconcileAllTask()); err != nil { + logger.Error("failed to register reconciliation schedule", "error", err) + os.Exit(1) + } go func() { logger.Info("worker started", "concurrency", cfg.WorkerConcurrency, "redis_addr", cfg.RedisAddr) @@ -61,8 +66,16 @@ func main() { stop() } }() + go func() { + logger.Info("worker scheduler started", "reconcile_interval", "5m") + if err := scheduler.Run(); err != nil { + logger.Error("worker scheduler failed", "error", err) + stop() + } + }() <-ctx.Done() + scheduler.Shutdown() server.Shutdown() logger.Info("worker stopped") } @@ -78,6 +91,17 @@ func registerHandlers(mux *asynq.ServeMux, db *sql.DB, clusters cluster.Reposito logger, ), ) + reconcileStore := tasks.NewSQLReconcileStore(db) + mux.Handle( + tasks.TypeProxmoxReconcileAll, + tasks.NewProxmoxReconcileHandler( + reconcileStore, + reconcileStore, + tasks.NewDefaultClusterReconcileClientResolver(clusters), + tasks.NewSQLAuditWriter(db), + logger, + ), + ) } func openDatabase(databaseURL string) (*sql.DB, error) { diff --git a/worker/internal/tasks/proxmox_reconcile.go b/worker/internal/tasks/proxmox_reconcile.go new file mode 100644 index 0000000..c1f4419 --- /dev/null +++ b/worker/internal/tasks/proxmox_reconcile.go @@ -0,0 +1,211 @@ +package tasks + +import ( + "context" + "database/sql" + "fmt" + "log/slog" + + "forgejo.digital-droplets.de/philschlo/proxui/platform/jobs" + "github.com/hibiken/asynq" +) + +const TypeProxmoxReconcileAll = jobs.TypeProxmoxReconcileAll + +type ReconcileClusterStore interface { + ListActiveClusterIDs(ctx context.Context) ([]string, error) +} + +type ReconcileVMStore interface { + ListClusterVMs(ctx context.Context, clusterID string) ([]ReconcileVM, error) +} + +type ReconcileVM struct { + ID string + TenantID string + Node string + ProxmoxVMID int + Status string +} + +type ProxmoxReconcileClient interface { + GetVMStatus(ctx context.Context, node string, vmid int) (ProxmoxVMStatus, error) +} + +type ProxmoxVMStatus struct { + Status string +} + +type ProxmoxReconcileClientResolver interface { + ResolveReconcileClient(ctx context.Context, clusterID string) (ProxmoxReconcileClient, error) +} + +type ProxmoxReconcileHandler struct { + clusters ReconcileClusterStore + vms ReconcileVMStore + resolver ProxmoxReconcileClientResolver + auditWriter AuditWriter + logger *slog.Logger +} + +func NewProxmoxReconcileHandler( + clusters ReconcileClusterStore, + vms ReconcileVMStore, + resolver ProxmoxReconcileClientResolver, + auditWriter AuditWriter, + logger *slog.Logger, +) ProxmoxReconcileHandler { + return ProxmoxReconcileHandler{ + clusters: clusters, + vms: vms, + resolver: resolver, + auditWriter: auditWriter, + logger: logger, + } +} + +func NewProxmoxReconcileAllTask() *asynq.Task { + return asynq.NewTask(TypeProxmoxReconcileAll, nil) +} + +func (h ProxmoxReconcileHandler) ProcessTask(ctx context.Context, _ *asynq.Task) error { + clusterIDs, err := h.clusters.ListActiveClusterIDs(ctx) + if err != nil { + return fmt.Errorf("list active clusters: %w", err) + } + + for _, clusterID := range clusterIDs { + if err := h.reconcileCluster(ctx, clusterID); err != nil { + return err + } + } + + h.logger.Info("completed proxmox reconciliation", "cluster_count", len(clusterIDs)) + return nil +} + +func (h ProxmoxReconcileHandler) reconcileCluster(ctx context.Context, clusterID string) error { + client, err := h.resolver.ResolveReconcileClient(ctx, clusterID) + if err != nil { + return fmt.Errorf("resolve proxmox reconcile client for cluster %s: %w", clusterID, err) + } + + vms, err := h.vms.ListClusterVMs(ctx, clusterID) + if err != nil { + return fmt.Errorf("list cluster vms for cluster %s: %w", clusterID, err) + } + + driftCount := 0 + for _, vm := range vms { + proxmoxStatus, err := client.GetVMStatus(ctx, vm.Node, vm.ProxmoxVMID) + if err != nil { + return fmt.Errorf("get proxmox vm status for vm %s: %w", vm.ID, err) + } + + currentStatus := normalizeProxmoxVMStatus(proxmoxStatus.Status) + if currentStatus == "" || currentStatus == vm.Status { + continue + } + + driftCount++ + if err := h.auditWriter.WriteAudit(ctx, AuditEvent{ + TenantID: vm.TenantID, + Action: "vm.reconcile.drift", + TargetType: "vm", + TargetID: vm.ID, + Metadata: map[string]any{ + "cluster_id": clusterID, + "node": vm.Node, + "proxmox_vmid": vm.ProxmoxVMID, + "db_status": vm.Status, + "proxmox_status": currentStatus, + }, + }); err != nil { + return fmt.Errorf("write reconcile drift audit for vm %s: %w", vm.ID, err) + } + } + + h.logger.Info("reconciled proxmox cluster", "cluster_id", clusterID, "vm_count", len(vms), "drift_count", driftCount) + return nil +} + +func normalizeProxmoxVMStatus(status string) string { + switch status { + case "running": + return "running" + case "stopped": + return "stopped" + case "paused": + return "suspended" + default: + return "" + } +} + +type SQLReconcileStore struct { + db *sql.DB +} + +func NewSQLReconcileStore(db *sql.DB) SQLReconcileStore { + return SQLReconcileStore{db: db} +} + +func (s SQLReconcileStore) ListActiveClusterIDs(ctx context.Context) ([]string, error) { + rows, err := s.db.QueryContext(ctx, ` + select id::text + from public.clusters + where status = 'active' + order by created_at asc + `) + if err != nil { + return nil, err + } + defer rows.Close() + + var clusterIDs []string + for rows.Next() { + var clusterID string + if err := rows.Scan(&clusterID); err != nil { + return nil, err + } + clusterIDs = append(clusterIDs, clusterID) + } + if err := rows.Err(); err != nil { + return nil, err + } + return clusterIDs, nil +} + +func (s SQLReconcileStore) ListClusterVMs(ctx context.Context, clusterID string) ([]ReconcileVM, error) { + rows, err := s.db.QueryContext(ctx, ` + select + v.id::text, + p.tenant_id::text, + v.node, + v.proxmox_vmid, + v.status::text + from public.vms v + join public.projects p + on p.id = v.project_id + where v.cluster_id = $1 + and v.status not in ('deleting', 'deleted') + order by v.created_at asc + `, clusterID) + if err != nil { + return nil, err + } + defer rows.Close() + + var vms []ReconcileVM + for rows.Next() { + var vm ReconcileVM + if err := rows.Scan(&vm.ID, &vm.TenantID, &vm.Node, &vm.ProxmoxVMID, &vm.Status); err != nil { + return nil, err + } + vms = append(vms, vm) + } + if err := rows.Err(); err != nil { + return nil, err + } + return vms, nil +} diff --git a/worker/internal/tasks/proxmox_reconcile_resolver.go b/worker/internal/tasks/proxmox_reconcile_resolver.go new file mode 100644 index 0000000..5b3a19b --- /dev/null +++ b/worker/internal/tasks/proxmox_reconcile_resolver.go @@ -0,0 +1,62 @@ +package tasks + +import ( + "context" + + "forgejo.digital-droplets.de/philschlo/proxui/platform/cluster" + "forgejo.digital-droplets.de/philschlo/proxui/platform/proxmox" +) + +type ProxmoxReconcileClientFactory func(cluster cluster.Cluster) (PlatformProxmoxReconcileClient, error) + +type PlatformProxmoxReconcileClient interface { + GetVMStatus(ctx context.Context, node string, vmid int) (proxmox.VMStatus, error) +} + +type ClusterReconcileClientResolver struct { + clusters ClusterRepository + factory ProxmoxReconcileClientFactory +} + +func NewClusterReconcileClientResolver(clusters ClusterRepository, factory ProxmoxReconcileClientFactory) ClusterReconcileClientResolver { + return ClusterReconcileClientResolver{ + clusters: clusters, + factory: factory, + } +} + +func NewDefaultClusterReconcileClientResolver(clusters ClusterRepository) ClusterReconcileClientResolver { + return NewClusterReconcileClientResolver(clusters, func(cluster cluster.Cluster) (PlatformProxmoxReconcileClient, error) { + return proxmox.NewClient(cluster) + }) +} + +func (r ClusterReconcileClientResolver) ResolveReconcileClient(ctx context.Context, clusterID string) (ProxmoxReconcileClient, error) { + cluster, found, err := r.clusters.GetCluster(ctx, clusterID) + if err != nil { + return nil, err + } + if !found { + return nil, ErrClusterNotFound + } + + client, err := r.factory(cluster) + if err != nil { + return nil, err + } + + return platformReconcileClientAdapter{client: client}, nil +} + +type platformReconcileClientAdapter struct { + client PlatformProxmoxReconcileClient +} + +func (a platformReconcileClientAdapter) GetVMStatus(ctx context.Context, node string, vmid int) (ProxmoxVMStatus, error) { + status, err := a.client.GetVMStatus(ctx, node, vmid) + if err != nil { + return ProxmoxVMStatus{}, err + } + + return ProxmoxVMStatus{Status: status.Status}, nil +} diff --git a/worker/internal/tasks/proxmox_reconcile_test.go b/worker/internal/tasks/proxmox_reconcile_test.go new file mode 100644 index 0000000..9f3c8a0 --- /dev/null +++ b/worker/internal/tasks/proxmox_reconcile_test.go @@ -0,0 +1,124 @@ +package tasks + +import ( + "context" + "log/slog" + "testing" +) + +func TestProxmoxReconcileHandlerWritesAuditForDrift(t *testing.T) { + client := &stubReconcileClient{status: ProxmoxVMStatus{Status: "stopped"}} + auditWriter := &stubAuditWriter{} + handler := NewProxmoxReconcileHandler( + &stubReconcileClusterStore{clusterIDs: []string{"cluster-1"}}, + &stubReconcileVMStore{vms: []ReconcileVM{{ + ID: "vm-1", + TenantID: "tenant-1", + Node: "pve", + ProxmoxVMID: 100, + Status: "running", + }}}, + &stubReconcileClientResolver{client: client}, + auditWriter, + slog.Default(), + ) + + if err := handler.ProcessTask(context.Background(), NewProxmoxReconcileAllTask()); err != nil { + t.Fatalf("ProcessTask() error = %v", err) + } + + if client.node != "pve" { + t.Fatalf("client node = %q, want pve", client.node) + } + if auditWriter.event.Action != "vm.reconcile.drift" { + t.Fatalf("audit action = %q, want vm.reconcile.drift", auditWriter.event.Action) + } + if auditWriter.event.Metadata["db_status"] != "running" { + t.Fatalf("audit db_status = %v, want running", auditWriter.event.Metadata["db_status"]) + } + if auditWriter.event.Metadata["proxmox_status"] != "stopped" { + t.Fatalf("audit proxmox_status = %v, want stopped", auditWriter.event.Metadata["proxmox_status"]) + } +} + +func TestProxmoxReconcileHandlerSkipsMatchingStatus(t *testing.T) { + auditWriter := &stubAuditWriter{} + handler := NewProxmoxReconcileHandler( + &stubReconcileClusterStore{clusterIDs: []string{"cluster-1"}}, + &stubReconcileVMStore{vms: []ReconcileVM{{ + ID: "vm-1", + TenantID: "tenant-1", + Node: "pve", + ProxmoxVMID: 100, + Status: "running", + }}}, + &stubReconcileClientResolver{client: &stubReconcileClient{status: ProxmoxVMStatus{Status: "running"}}}, + auditWriter, + slog.Default(), + ) + + if err := handler.ProcessTask(context.Background(), NewProxmoxReconcileAllTask()); err != nil { + t.Fatalf("ProcessTask() error = %v", err) + } + + if auditWriter.event.Action != "" { + t.Fatalf("audit action = %q, want empty", auditWriter.event.Action) + } +} + +func TestNormalizeProxmoxVMStatus(t *testing.T) { + tests := []struct { + name string + status string + want string + }{ + {name: "running", status: "running", want: "running"}, + {name: "stopped", status: "stopped", want: "stopped"}, + {name: "paused", status: "paused", want: "suspended"}, + {name: "unknown", status: "unknown", want: ""}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if got := normalizeProxmoxVMStatus(test.status); got != test.want { + t.Fatalf("normalizeProxmoxVMStatus() = %q, want %q", got, test.want) + } + }) + } +} + +type stubReconcileClusterStore struct { + clusterIDs []string +} + +func (s *stubReconcileClusterStore) ListActiveClusterIDs(context.Context) ([]string, error) { + return s.clusterIDs, nil +} + +type stubReconcileVMStore struct { + vms []ReconcileVM +} + +func (s *stubReconcileVMStore) ListClusterVMs(context.Context, string) ([]ReconcileVM, error) { + return s.vms, nil +} + +type stubReconcileClientResolver struct { + client ProxmoxReconcileClient +} + +func (s *stubReconcileClientResolver) ResolveReconcileClient(context.Context, string) (ProxmoxReconcileClient, error) { + return s.client, nil +} + +type stubReconcileClient struct { + node string + vmid int + status ProxmoxVMStatus +} + +func (s *stubReconcileClient) GetVMStatus(_ context.Context, node string, vmid int) (ProxmoxVMStatus, error) { + s.node = node + s.vmid = vmid + return s.status, nil +}