sysops
67bcdd1833
Stichprobenbasierter Scrub-Job: nimmt BAK-05s existing_in_storage, priorisiert nach eigenem scrub_state.last_scrubbed_at (nicht file_revisions.created_at, sonst kein echtes Rotationsverhalten), prueft Inhalt per SHA-256 gegen file_revisions.checksum_sha256. Meldung ueber echten dauerhaften /metrics-Endpunkt (Pull-Modell, OPS-03 scrapt, kein Push), Counter monoton steigend. Real registriert in Core metrics_sources, End-zu-Ende ueber OPS-03-Aggregator bestaetigt, realer Befund-Durchlauf mit absichtlich falscher Pruefsumme durchgefuehrt.
75 lines
2.4 KiB
Go
75 lines
2.4 KiB
Go
package scrub
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
// LoadLastScrubbed liefert je storage_key den Zeitpunkt der letzten
|
|
// Pruefung — Grundlage fuer Sample's Cooldown-Filter.
|
|
func LoadLastScrubbed(ctx context.Context, pool *pgxpool.Pool) (map[string]time.Time, error) {
|
|
rows, err := pool.Query(ctx, `SELECT storage_key, last_scrubbed_at FROM scrub_state`)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("scrub: scrub_state lesen: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
|
|
out := make(map[string]time.Time)
|
|
for rows.Next() {
|
|
var key string
|
|
var ts time.Time
|
|
if err := rows.Scan(&key, &ts); err != nil {
|
|
return nil, fmt.Errorf("scrub: scrub_state-zeile lesen: %w", err)
|
|
}
|
|
out[key] = ts
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
// MarkScrubbed vermerkt Ergebnis und Zeitpunkt der Pruefung eines
|
|
// Objekts — idempotent (ON CONFLICT), damit ein unterbrochener und neu
|
|
// gestarteter Lauf keinen inkonsistenten Zustand hinterlaesst
|
|
// (Akzeptanzkriterium: Lauf ist unterbrechbar ohne inkonsistenten
|
|
// Zustand).
|
|
func MarkScrubbed(ctx context.Context, pool *pgxpool.Pool, storageKey string, ok bool, at time.Time) error {
|
|
result := "ok"
|
|
if !ok {
|
|
result = "failed"
|
|
}
|
|
_, err := pool.Exec(ctx, `
|
|
INSERT INTO scrub_state (storage_key, last_scrubbed_at, last_result)
|
|
VALUES ($1, $2, $3)
|
|
ON CONFLICT (storage_key) DO UPDATE SET last_scrubbed_at = $2, last_result = $3
|
|
`, storageKey, at, result)
|
|
if err != nil {
|
|
return fmt.Errorf("scrub: scrub_state schreiben: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// RecordFinding erhoeht den monoton steigenden Befund-Zaehler
|
|
// (scrub_counters.findings_total) um genau 1 — als gueltiger Prometheus-
|
|
// Counter darf dieser Wert nur steigen, niemals sinken, auch wenn ein
|
|
// Befund spaeter behoben wird.
|
|
func RecordFinding(ctx context.Context, pool *pgxpool.Pool) error {
|
|
_, err := pool.Exec(ctx, `UPDATE scrub_counters SET findings_total = findings_total + 1 WHERE id = 1`)
|
|
if err != nil {
|
|
return fmt.Errorf("scrub: befund-zaehler erhoehen: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// FindingsTotal liest den aktuellen Zaehlerstand — genutzt vom
|
|
// /metrics-Endpunkt (cmd/scrub-metrics).
|
|
func FindingsTotal(ctx context.Context, pool *pgxpool.Pool) (int64, error) {
|
|
var total int64
|
|
err := pool.QueryRow(ctx, `SELECT findings_total FROM scrub_counters WHERE id = 1`).Scan(&total)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("scrub: befund-zaehler lesen: %w", err)
|
|
}
|
|
return total, nil
|
|
}
|