diff --git a/internal/watchdog/watchdog.go b/internal/watchdog/watchdog.go new file mode 100644 index 0000000..0dbb85b --- /dev/null +++ b/internal/watchdog/watchdog.go @@ -0,0 +1,90 @@ +package watchdog + +import ( + "context" + "log/slog" + "time" + + configuration "git.g3e.fr/syonad/two/internal/config/agent" + "git.g3e.fr/syonad/two/internal/watchdog/notify" + "git.g3e.fr/syonad/two/pkg/systemd" + + "github.com/dgraph-io/badger/v4" +) + +const defaultInterval = 60 * time.Second + +type Watchdog struct { + db *badger.DB + cfg *configuration.Config + notifier notify.Notifier + logger *slog.Logger + interval time.Duration + dbusDown bool +} + +func New(db *badger.DB, cfg *configuration.Config, n notify.Notifier, logger *slog.Logger, interval time.Duration) *Watchdog { + if logger == nil { + logger = slog.Default() + } + if interval <= 0 { + logger.Warn("watchdog: intervalle invalide, valeur par défaut appliquée", + "interval", interval, "default", defaultInterval) + interval = defaultInterval + } + return &Watchdog{ + db: db, + cfg: cfg, + notifier: n, + logger: logger, + interval: interval, + } +} + +func (w *Watchdog) Run(ctx context.Context) { + ticker := time.NewTicker(w.interval) + defer ticker.Stop() + + w.logger.Info("watchdog: démarrage", "interval", w.interval) + + for { + select { + case <-ctx.Done(): + w.logger.Info("watchdog: arrêt") + return + case <-ticker.C: + w.tick() + } + } +} + +func (w *Watchdog) tick() { + u, closeUnits := w.units() + defer closeUnits() + + if err := CheckVPCs(w.db, w.notifier); err != nil { + w.logger.Error("watchdog: vérification des vpc", "err", err) + } + if err := CheckSubnets(w.db, u, w.notifier); err != nil { + w.logger.Error("watchdog: vérification des subnets", "err", err) + } + if err := CheckVMs(w.db, w.cfg, u, w.notifier); err != nil { + w.logger.Error("watchdog: vérification des vm", "err", err) + } +} + +func (w *Watchdog) units() (unitChecker, func()) { + m, err := systemd.New() + if err != nil { + if !w.dbusDown { + w.logger.Warn("watchdog: connexion systemd impossible, vérification des units désactivée", "err", err) + w.dbusDown = true + } + return nil, func() {} + } + if w.dbusDown { + w.logger.Info("watchdog: connexion systemd rétablie") + w.dbusDown = false + } + return m, m.Close +} diff --git a/internal/watchdog/watchdog_test.go b/internal/watchdog/watchdog_test.go new file mode 100644 index 0000000..c8cea47 --- /dev/null +++ b/internal/watchdog/watchdog_test.go @@ -0,0 +1,200 @@ +package watchdog + +import ( + "bytes" + "context" + "encoding/json" + "log/slog" + "strings" + "testing" + "time" + + "git.g3e.fr/syonad/two/internal/state" + "git.g3e.fr/syonad/two/internal/watchdog/notify" +) + +type chanNotifier struct { + seen chan notification +} + +var _ notify.Notifier = (*chanNotifier)(nil) + +func newChanNotifier() *chanNotifier { + return &chanNotifier{seen: make(chan notification, 256)} +} + +func (c *chanNotifier) Notify(kind, name, problem string) { + select { + case c.seen <- notification{kind: kind, name: name, problem: problem}: + default: + } +} + +func testLogger() (*slog.Logger, *bytes.Buffer) { + var buf bytes.Buffer + return slog.New(slog.NewJSONHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug})), &buf +} + +func logLines(t *testing.T, buf *bytes.Buffer) []map[string]any { + t.Helper() + var out []map[string]any + for _, line := range strings.Split(strings.TrimSpace(buf.String()), "\n") { + if line == "" { + continue + } + var m map[string]any + if err := json.Unmarshal([]byte(line), &m); err != nil { + t.Fatalf("ligne non JSON %q: %v", line, err) + } + out = append(out, m) + } + return out +} + +func TestNew_IntervalleInvalideUtiliseLeDefaut(t *testing.T) { + l, buf := testLogger() + + for _, interval := range []time.Duration{0, -time.Second} { + w := New(newTestDB(t), testCfg(t), &recorder{}, l, interval) + if w.interval != defaultInterval { + t.Errorf("interval = %v, attendu %v", w.interval, defaultInterval) + } + } + + if !strings.Contains(buf.String(), "intervalle invalide") { + t.Error("un intervalle invalide devrait être signalé dans les logs") + } +} + +func TestNew_IntervalleValideConserve(t *testing.T) { + l, _ := testLogger() + w := New(newTestDB(t), testCfg(t), &recorder{}, l, 5*time.Second) + if w.interval != 5*time.Second { + t.Errorf("interval = %v, attendu 5s", w.interval) + } +} + +func TestNew_LoggerNil(t *testing.T) { + w := New(newTestDB(t), testCfg(t), &recorder{}, nil, time.Second) + if w.logger == nil { + t.Fatal("logger nil non remplacé par slog.Default()") + } +} + +func TestRun_SArreteSurContexteAnnule(t *testing.T) { + l, _ := testLogger() + w := New(newTestDB(t), testCfg(t), &recorder{}, l, time.Hour) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + w.Run(ctx) + close(done) + }() + + cancel() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("Run n'a pas rendu la main après annulation du contexte") + } +} + +func TestRun_VerifieLesTroisTypesDeRessource(t *testing.T) { + db := newTestDB(t) + seedResource(t, db, prefixVPC, "vp-fantome", state.Running) + seedSubnet(t, db, "br-000042", "vp-admin", modeBridge) + seedVM(t, db, "i-test1", "br-000042", "vp-admin", "12345678") + + n := newChanNotifier() + l, _ := testLogger() + w := New(db, testCfg(t), n, l, 10*time.Millisecond) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go w.Run(ctx) + + kinds := map[string]bool{} + deadline := time.After(5 * time.Second) + for len(kinds) < 3 { + select { + case c := <-n.seen: + kinds[c.kind] = true + case <-deadline: + t.Fatalf("types de ressource vérifiés: %v, attendu vpc + subnet + vm", kinds) + } + } +} + +func TestRun_ErreurDeBaseLogueeEtBoucleContinue(t *testing.T) { + db := newTestDB(t) + seedResource(t, db, prefixVPC, "vp-fantome", state.Running) + db.Close() + + n := newChanNotifier() + l, buf := testLogger() + w := New(db, testCfg(t), n, l, 10*time.Millisecond) + + ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond) + defer cancel() + w.Run(ctx) + + var vpc, subnet, vm bool + for _, line := range logLines(t, buf) { + msg, _ := line["msg"].(string) + switch { + case strings.Contains(msg, "vérification des vpc"): + vpc = true + case strings.Contains(msg, "vérification des subnets"): + subnet = true + case strings.Contains(msg, "vérification des vm"): + vm = true + } + } + if !vpc || !subnet || !vm { + t.Errorf("les trois erreurs devraient être loguées (vpc=%v subnet=%v vm=%v)", vpc, subnet, vm) + } + if len(n.seen) != 0 { + t.Errorf("une erreur d'accès à la base ne doit pas produire de notification, obtenu %d", len(n.seen)) + } +} + +func TestRun_LogueDemarrageEtArret(t *testing.T) { + l, buf := testLogger() + w := New(newTestDB(t), testCfg(t), &recorder{}, l, time.Hour) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + w.Run(ctx) + + out := buf.String() + if !strings.Contains(out, "watchdog: démarrage") { + t.Error("le démarrage devrait être logué") + } + if !strings.Contains(out, "watchdog: arrêt") { + t.Error("l'arrêt devrait être logué") + } +} + +func TestUnits_ConnexionSystemdIndisponibleSignaleeUneSeuleFois(t *testing.T) { + l, buf := testLogger() + w := New(newTestDB(t), testCfg(t), &recorder{}, l, time.Hour) + + for range 3 { + u, close := w.units() + close() + if u != nil { + t.Skip("systemd joignable sur cette machine, cas non testable") + } + } + + var warnings int + for _, line := range logLines(t, buf) { + if msg, _ := line["msg"].(string); strings.Contains(msg, "connexion systemd impossible") { + warnings++ + } + } + if warnings != 1 { + t.Errorf("attendu 1 avertissement pour 3 tentatives, obtenu %d", warnings) + } +}