453 lines
12 KiB
Go
453 lines
12 KiB
Go
package server
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"scm.bstein.dev/bstein/soteria/internal/api"
|
|
"scm.bstein.dev/bstein/soteria/internal/k8s"
|
|
)
|
|
|
|
func (s *Server) runPolicyCycle(ctx context.Context) {
|
|
if !s.beginRun() {
|
|
return
|
|
}
|
|
defer s.endRun()
|
|
|
|
policies := s.activePolicies()
|
|
if len(policies) == 0 {
|
|
return
|
|
}
|
|
|
|
runCtx, cancel := context.WithTimeout(ctx, 3*time.Minute)
|
|
defer cancel()
|
|
|
|
inventory, err := s.buildInventory(runCtx)
|
|
if err != nil {
|
|
log.Printf("policy cycle inventory failed: %v", err)
|
|
s.metrics.RecordPolicyBackup("inventory_error")
|
|
return
|
|
}
|
|
|
|
pvcMap := make(map[string]api.PVCInventory)
|
|
namespaceMap := make(map[string][]api.PVCInventory)
|
|
namespaceNames := make([]string, 0, len(inventory.Namespaces))
|
|
for _, group := range inventory.Namespaces {
|
|
namespaceNames = append(namespaceNames, group.Name)
|
|
for _, pvc := range group.PVCs {
|
|
key := pvc.Namespace + "/" + pvc.PVC
|
|
pvcMap[key] = pvc
|
|
namespaceMap[pvc.Namespace] = append(namespaceMap[pvc.Namespace], pvc)
|
|
}
|
|
}
|
|
|
|
type effectivePolicy struct {
|
|
IntervalHours float64
|
|
Dedupe bool
|
|
KeepLast int
|
|
}
|
|
effectivePolicies := map[string]effectivePolicy{}
|
|
for _, policy := range policies {
|
|
matches := []api.PVCInventory{}
|
|
if policy.PVC != "" {
|
|
if pvc, ok := pvcMap[policy.Namespace+"/"+policy.PVC]; ok {
|
|
matches = append(matches, pvc)
|
|
}
|
|
} else {
|
|
matches = append(matches, namespaceMap[policy.Namespace]...)
|
|
}
|
|
|
|
for _, pvc := range matches {
|
|
key := pvc.Namespace + "/" + pvc.PVC
|
|
current, exists := effectivePolicies[key]
|
|
if !exists || policy.IntervalHours < current.IntervalHours || (policy.IntervalHours == current.IntervalHours && keepLastStricter(policy.KeepLast, current.KeepLast)) {
|
|
effectivePolicies[key] = effectivePolicy{
|
|
IntervalHours: policy.IntervalHours,
|
|
Dedupe: policy.Dedupe,
|
|
KeepLast: policy.KeepLast,
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
keys := make([]string, 0, len(effectivePolicies))
|
|
for key := range effectivePolicies {
|
|
keys = append(keys, key)
|
|
}
|
|
sort.Strings(keys)
|
|
|
|
activeRepositories := map[string]struct{}{}
|
|
if s.cfg.BackupDriver == "restic" {
|
|
var err error
|
|
activeRepositories, err = s.activeResticRepositories(runCtx, namespaceNames)
|
|
if err != nil {
|
|
log.Printf("policy cycle active backup lookup failed: %v", err)
|
|
s.metrics.RecordPolicyBackup("active_lookup_error")
|
|
return
|
|
}
|
|
}
|
|
|
|
started := 0
|
|
for _, key := range keys {
|
|
effective := effectivePolicies[key]
|
|
pvc, ok := pvcMap[key]
|
|
if !ok {
|
|
continue
|
|
}
|
|
if excluded, _ := s.pvcExcluded(pvc.Namespace, pvc.PVC, pvc.StorageClass); excluded {
|
|
s.metrics.RecordPolicyBackup("excluded")
|
|
continue
|
|
}
|
|
blocked, _, err := s.liveExclusivePVCMounted(runCtx, pvc.Namespace, pvc.PVC, pvc.AccessModes)
|
|
if err != nil {
|
|
log.Printf("policy cycle live PVC mount lookup failed for %s/%s: %v", pvc.Namespace, pvc.PVC, err)
|
|
s.metrics.RecordPolicyBackup("active_lookup_error")
|
|
continue
|
|
}
|
|
if blocked {
|
|
s.metrics.RecordPolicyBackup("live_rwo_mount")
|
|
continue
|
|
}
|
|
// Never enqueue a new policy backup while one is already active for this PVC.
|
|
// This prevents runaway job storms when a backup is stuck Pending/Running.
|
|
if pvc.ActiveBackups > 0 {
|
|
s.metrics.RecordPolicyBackup("in_progress")
|
|
continue
|
|
}
|
|
|
|
if !backupDue(policyRunReference(pvc), effective.IntervalHours) {
|
|
s.metrics.RecordPolicyBackup("not_due")
|
|
continue
|
|
}
|
|
if s.cfg.BackupDriver == "longhorn" {
|
|
ready, reason, err := s.longhornVolumeReady(runCtx, pvc.Volume)
|
|
if err != nil {
|
|
log.Printf("policy cycle Longhorn volume lookup failed for %s/%s: %v", pvc.Namespace, pvc.PVC, err)
|
|
s.metrics.RecordPolicyBackup("active_lookup_error")
|
|
continue
|
|
}
|
|
if !ready {
|
|
log.Printf("policy backup skipped for %s/%s: %s", pvc.Namespace, pvc.PVC, reason)
|
|
s.metrics.RecordPolicyBackup("volume_not_ready")
|
|
continue
|
|
}
|
|
}
|
|
if s.cfg.PolicyBackupsPerCycle > 0 && started >= s.cfg.PolicyBackupsPerCycle {
|
|
s.metrics.RecordPolicyBackup("cycle_limit")
|
|
break
|
|
}
|
|
repository := k8s.ResticRepositoryForBackup(s.cfg.ResticRepository, pvc.Namespace, pvc.PVC, effective.Dedupe)
|
|
if _, busy := activeRepositories[repository]; busy {
|
|
s.metrics.RecordPolicyBackup("repo_in_progress")
|
|
continue
|
|
}
|
|
|
|
_, result, err := s.executeBackup(runCtx, api.BackupRequest{
|
|
Namespace: pvc.Namespace,
|
|
PVC: pvc.PVC,
|
|
DryRun: false,
|
|
Dedupe: boolPtr(effective.Dedupe),
|
|
KeepLast: intPtr(effective.KeepLast),
|
|
}, "policy-scheduler")
|
|
started++
|
|
s.metrics.RecordBackupRequest(s.cfg.BackupDriver, result)
|
|
if err != nil {
|
|
s.metrics.RecordPolicyBackup(result)
|
|
log.Printf("policy backup failed for %s/%s: %v", pvc.Namespace, pvc.PVC, err)
|
|
continue
|
|
}
|
|
if s.cfg.BackupDriver == "restic" {
|
|
activeRepositories[repository] = struct{}{}
|
|
}
|
|
s.metrics.RecordPolicyBackup("success")
|
|
}
|
|
}
|
|
|
|
func (s *Server) beginRun() bool {
|
|
s.runMu.Lock()
|
|
defer s.runMu.Unlock()
|
|
if s.running {
|
|
return false
|
|
}
|
|
s.running = true
|
|
return true
|
|
}
|
|
|
|
func (s *Server) endRun() {
|
|
s.runMu.Lock()
|
|
defer s.runMu.Unlock()
|
|
s.running = false
|
|
}
|
|
|
|
func (s *Server) loadPolicies(ctx context.Context) error {
|
|
raw, err := s.client.LoadSecretData(ctx, s.cfg.Namespace, s.cfg.PolicySecretName, policySecretKey)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if len(raw) == 0 {
|
|
return nil
|
|
}
|
|
|
|
var doc struct {
|
|
Policies []struct {
|
|
ID string `json:"id"`
|
|
Namespace string `json:"namespace"`
|
|
PVC string `json:"pvc,omitempty"`
|
|
IntervalHours float64 `json:"interval_hours"`
|
|
Enabled bool `json:"enabled"`
|
|
Dedupe *bool `json:"dedupe,omitempty"`
|
|
KeepLast *int `json:"keep_last,omitempty"`
|
|
CreatedAt string `json:"created_at,omitempty"`
|
|
UpdatedAt string `json:"updated_at,omitempty"`
|
|
} `json:"policies"`
|
|
}
|
|
if err := json.Unmarshal(raw, &doc); err != nil {
|
|
return fmt.Errorf("decode policy document: %w", err)
|
|
}
|
|
|
|
next := map[string]api.BackupPolicy{}
|
|
now := time.Now().UTC().Format(time.RFC3339)
|
|
for _, policy := range doc.Policies {
|
|
namespace := strings.TrimSpace(policy.Namespace)
|
|
pvc := strings.TrimSpace(policy.PVC)
|
|
if namespace == "" {
|
|
continue
|
|
}
|
|
interval := policy.IntervalHours
|
|
if interval <= 0 {
|
|
interval = defaultPolicyHours
|
|
}
|
|
dedupe := true
|
|
if policy.Dedupe != nil {
|
|
dedupe = *policy.Dedupe
|
|
}
|
|
keepLast := keepLastDefault(policy.KeepLast)
|
|
id := policyKey(namespace, pvc)
|
|
createdAt := policy.CreatedAt
|
|
if createdAt == "" {
|
|
createdAt = now
|
|
}
|
|
updatedAt := policy.UpdatedAt
|
|
if updatedAt == "" {
|
|
updatedAt = createdAt
|
|
}
|
|
next[id] = api.BackupPolicy{
|
|
ID: id,
|
|
Namespace: namespace,
|
|
PVC: pvc,
|
|
IntervalHours: interval,
|
|
Enabled: policy.Enabled,
|
|
Dedupe: dedupe,
|
|
KeepLast: keepLast,
|
|
CreatedAt: createdAt,
|
|
UpdatedAt: updatedAt,
|
|
}
|
|
}
|
|
|
|
s.policyMu.Lock()
|
|
s.policies = next
|
|
s.policyMu.Unlock()
|
|
return nil
|
|
}
|
|
|
|
func (s *Server) persistPolicies(ctx context.Context, policies []api.BackupPolicy) error {
|
|
doc := struct {
|
|
Policies []api.BackupPolicy `json:"policies"`
|
|
}{
|
|
Policies: policies,
|
|
}
|
|
payload, err := json.Marshal(doc)
|
|
if err != nil {
|
|
return fmt.Errorf("encode policy document: %w", err)
|
|
}
|
|
return s.client.SaveSecretData(ctx, s.cfg.Namespace, s.cfg.PolicySecretName, policySecretKey, payload, map[string]string{
|
|
"app.kubernetes.io/name": "soteria",
|
|
"app.kubernetes.io/component": "policy-store",
|
|
})
|
|
}
|
|
|
|
func (s *Server) listPolicies() []api.BackupPolicy {
|
|
s.policyMu.RLock()
|
|
defer s.policyMu.RUnlock()
|
|
policies := make([]api.BackupPolicy, 0, len(s.policies))
|
|
for _, policy := range s.policies {
|
|
policies = append(policies, policy)
|
|
}
|
|
sort.Slice(policies, func(i, j int) bool {
|
|
if policies[i].Namespace != policies[j].Namespace {
|
|
return policies[i].Namespace < policies[j].Namespace
|
|
}
|
|
if policies[i].PVC != policies[j].PVC {
|
|
return policies[i].PVC < policies[j].PVC
|
|
}
|
|
return policies[i].ID < policies[j].ID
|
|
})
|
|
return policies
|
|
}
|
|
|
|
func (s *Server) activePolicies() []api.BackupPolicy {
|
|
policies := s.listPolicies()
|
|
filtered := make([]api.BackupPolicy, 0, len(policies))
|
|
for _, policy := range policies {
|
|
if policy.Enabled {
|
|
filtered = append(filtered, policy)
|
|
}
|
|
}
|
|
return filtered
|
|
}
|
|
|
|
func (s *Server) upsertPolicy(ctx context.Context, req api.BackupPolicyUpsertRequest) (api.BackupPolicy, error) {
|
|
namespace := strings.TrimSpace(req.Namespace)
|
|
pvc := strings.TrimSpace(req.PVC)
|
|
if namespace == "" {
|
|
return api.BackupPolicy{}, fmt.Errorf("namespace is required")
|
|
}
|
|
if err := validateKubernetesName("namespace", namespace); err != nil {
|
|
return api.BackupPolicy{}, err
|
|
}
|
|
if pvc != "" {
|
|
if err := validateKubernetesName("pvc", pvc); err != nil {
|
|
return api.BackupPolicy{}, err
|
|
}
|
|
}
|
|
|
|
interval := req.IntervalHours
|
|
if interval <= 0 {
|
|
interval = defaultPolicyHours
|
|
}
|
|
if interval > maxPolicyIntervalHrs {
|
|
return api.BackupPolicy{}, fmt.Errorf("interval_hours must be <= %d", maxPolicyIntervalHrs)
|
|
}
|
|
enabled := true
|
|
if req.Enabled != nil {
|
|
enabled = *req.Enabled
|
|
}
|
|
dedupe := dedupeDefault(req.Dedupe)
|
|
if err := validateKeepLast(req.KeepLast); err != nil {
|
|
return api.BackupPolicy{}, err
|
|
}
|
|
keepLast := keepLastDefault(req.KeepLast)
|
|
|
|
id := policyKey(namespace, pvc)
|
|
now := time.Now().UTC().Format(time.RFC3339)
|
|
|
|
s.policyMu.Lock()
|
|
before := clonePolicyMap(s.policies)
|
|
createdAt := now
|
|
if existing, ok := s.policies[id]; ok && existing.CreatedAt != "" {
|
|
createdAt = existing.CreatedAt
|
|
}
|
|
policy := api.BackupPolicy{
|
|
ID: id,
|
|
Namespace: namespace,
|
|
PVC: pvc,
|
|
IntervalHours: interval,
|
|
Enabled: enabled,
|
|
Dedupe: dedupe,
|
|
KeepLast: keepLast,
|
|
CreatedAt: createdAt,
|
|
UpdatedAt: now,
|
|
}
|
|
s.policies[id] = policy
|
|
snapshot := policySliceFromMap(s.policies)
|
|
s.policyMu.Unlock()
|
|
|
|
if err := s.persistPolicies(ctx, snapshot); err != nil {
|
|
s.policyMu.Lock()
|
|
s.policies = before
|
|
s.policyMu.Unlock()
|
|
return api.BackupPolicy{}, err
|
|
}
|
|
return policy, nil
|
|
}
|
|
|
|
func (s *Server) deletePolicy(ctx context.Context, id string) (bool, error) {
|
|
id = strings.TrimSpace(id)
|
|
if id == "" {
|
|
return false, nil
|
|
}
|
|
|
|
s.policyMu.Lock()
|
|
if _, ok := s.policies[id]; !ok {
|
|
s.policyMu.Unlock()
|
|
return false, nil
|
|
}
|
|
before := clonePolicyMap(s.policies)
|
|
delete(s.policies, id)
|
|
snapshot := policySliceFromMap(s.policies)
|
|
s.policyMu.Unlock()
|
|
|
|
if err := s.persistPolicies(ctx, snapshot); err != nil {
|
|
s.policyMu.Lock()
|
|
s.policies = before
|
|
s.policyMu.Unlock()
|
|
return false, err
|
|
}
|
|
return true, nil
|
|
}
|
|
|
|
func policyKey(namespace, pvc string) string {
|
|
scope := strings.TrimSpace(pvc)
|
|
if scope == "" {
|
|
scope = "_all"
|
|
}
|
|
return strings.TrimSpace(namespace) + "__" + scope
|
|
}
|
|
|
|
func policySliceFromMap(source map[string]api.BackupPolicy) []api.BackupPolicy {
|
|
out := make([]api.BackupPolicy, 0, len(source))
|
|
for _, policy := range source {
|
|
out = append(out, policy)
|
|
}
|
|
sort.Slice(out, func(i, j int) bool {
|
|
return out[i].ID < out[j].ID
|
|
})
|
|
return out
|
|
}
|
|
|
|
func clonePolicyMap(source map[string]api.BackupPolicy) map[string]api.BackupPolicy {
|
|
cloned := make(map[string]api.BackupPolicy, len(source))
|
|
for key, value := range source {
|
|
cloned[key] = value
|
|
}
|
|
return cloned
|
|
}
|
|
|
|
func backupDue(lastBackupAt string, intervalHours float64) bool {
|
|
if intervalHours <= 0 {
|
|
intervalHours = defaultPolicyHours
|
|
}
|
|
if strings.TrimSpace(lastBackupAt) == "" {
|
|
return true
|
|
}
|
|
timestamp, ok := parseBackupTime(lastBackupAt)
|
|
if !ok {
|
|
return true
|
|
}
|
|
interval := time.Duration(intervalHours * float64(time.Hour))
|
|
return time.Since(timestamp) >= interval
|
|
}
|
|
|
|
func policyRunReference(pvc api.PVCInventory) string {
|
|
lastBackup := strings.TrimSpace(pvc.LastBackupAt)
|
|
lastJob := strings.TrimSpace(pvc.LastJobStartedAt)
|
|
backupTime, backupOK := parseBackupTime(lastBackup)
|
|
jobTime, jobOK := parseBackupTime(lastJob)
|
|
switch {
|
|
case backupOK && jobOK:
|
|
if jobTime.After(backupTime) {
|
|
return lastJob
|
|
}
|
|
return lastBackup
|
|
case jobOK:
|
|
return lastJob
|
|
default:
|
|
return lastBackup
|
|
}
|
|
}
|