diff --git a/cmd/agent/main.go b/cmd/agent/main.go index 1cea7ec..0425d4e 100644 --- a/cmd/agent/main.go +++ b/cmd/agent/main.go @@ -1,22 +1,15 @@ package main import ( - "context" "flag" "fmt" "log/slog" - "os" - "os/signal" - "syscall" - "time" agentapi "git.g3e.fr/syonad/two/internal/api/agent" configuration "git.g3e.fr/syonad/two/internal/config/agent" dispatcher "git.g3e.fr/syonad/two/internal/dispatcher/agent" "git.g3e.fr/syonad/two/internal/migration" agentmetrics "git.g3e.fr/syonad/two/internal/prometheus/agent" - "git.g3e.fr/syonad/two/internal/watchdog" - "git.g3e.fr/syonad/two/internal/watchdog/notify" "git.g3e.fr/syonad/two/pkg/db/kv" "git.g3e.fr/syonad/two/pkg/logger" promserver "git.g3e.fr/syonad/two/pkg/prometheus" @@ -24,8 +17,6 @@ import ( "github.com/prometheus/client_golang/prometheus" ) -const shutdownTimeout = 20 * time.Second - func main() { confFile := flag.String("config", "/etc/two/agent.yml", "config file path") flag.Parse() @@ -39,12 +30,7 @@ func main() { log := logger.New(cfg.Logger.Level, cfg.Logger.Debug) db := kv.InitDB(kv.Config{Path: cfg.Database.Path}, false) - closeDB := true - defer func() { - if closeDB { - db.Close() - } - }() + defer db.Close() // Avant tout démarrage de service : la DB peut porter l'ancien vocabulaire // d'états, et des ressources transitoires orphelines d'un arrêt précédent. @@ -70,69 +56,13 @@ func main() { "debug", cfg.Logger.Debug, ) - ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) - defer stop() - d := dispatcher.New(q, db, cfg, log.With(slog.String("component", "dispatcher"))) - - apiSrv := agentapi.New(d, db, log.With(slog.String("component", "api")), apiAddr) - go apiSrv.Start() - - promSrv := promserver.New(promAddr, registry) - go promSrv.Start() - - var adminSrv *kv.AdminServer + go agentapi.New(d, db, log.With(slog.String("component", "api"))).Start(apiAddr) + go promserver.Start(promAddr, registry) if cfg.Admin.Enabled { adminAddr := fmt.Sprintf("%s:%d", cfg.Admin.Address, cfg.Admin.Port) - adminSrv = kv.NewAdminServer(db, log.With(slog.String("component", "admin")), adminAddr) - go adminSrv.Start() + go kv.NewAdminServer(db, log.With(slog.String("component", "admin"))).Start(adminAddr) } - if cfg.Watchdog.Enabled { - wlog := log.With(slog.String("component", "watchdog")) - go watchdog.New(db, cfg, notify.NewStderr(wlog), wlog, - time.Duration(cfg.Watchdog.IntervalSeconds)*time.Second, - ).Run(ctx) - } - - <-ctx.Done() - stop() - - servers := map[string]httpShutdowner{"api": apiSrv, "prometheus": promSrv} - if adminSrv != nil { - servers["admin"] = adminSrv - } - closeDB = shutdown(log, q, servers, shutdownTimeout) -} - -type httpShutdowner interface { - Shutdown(context.Context) error -} - -func shutdown(log *slog.Logger, q *worker.Queue, servers map[string]httpShutdowner, timeout time.Duration) bool { - log.Info("shutting down", "timeout", timeout) - - ctx, cancel := context.WithTimeout(context.Background(), timeout) - defer cancel() - - for name, srv := range servers { - if err := srv.Shutdown(ctx); err != nil { - log.Error("http server shutdown", "server", name, "error", err) - } - } - - drained := make(chan struct{}) - go func() { - q.Stop() - close(drained) - }() - - select { - case <-drained: - log.Info("workers drained") - return true - case <-ctx.Done(): - log.Error("workers still running after timeout, leaving database untouched") - return false - } + select {} } diff --git a/cmd/agent/main_test.go b/cmd/agent/main_test.go deleted file mode 100644 index 12291c7..0000000 --- a/cmd/agent/main_test.go +++ /dev/null @@ -1,98 +0,0 @@ -package main - -import ( - "context" - "errors" - "io" - "log/slog" - "strings" - "sync/atomic" - "testing" - "time" - - "git.g3e.fr/syonad/two/pkg/worker" -) - -type fakeServer struct { - called atomic.Bool - err error -} - -func (f *fakeServer) Shutdown(context.Context) error { - f.called.Store(true) - return f.err -} - -func discardLogger() *slog.Logger { - return slog.New(slog.NewTextHandler(io.Discard, nil)) -} - -func TestShutdown_DrainReussi(t *testing.T) { - q := worker.New(10) - q.Start(2) - - var done atomic.Int32 - for range 3 { - q.Submit(func() { - time.Sleep(20 * time.Millisecond) - done.Add(1) - }) - } - - api, prom := &fakeServer{}, &fakeServer{} - servers := map[string]httpShutdowner{"api": api, "prometheus": prom} - - if !shutdown(discardLogger(), q, servers, 5*time.Second) { - t.Fatal("un drainage réussi doit autoriser la fermeture de la base") - } - if !api.called.Load() || !prom.called.Load() { - t.Error("tous les serveurs HTTP doivent être arrêtés") - } - if got := done.Load(); got != 3 { - t.Errorf("les 3 tâches devaient se terminer, %d terminées", got) - } -} - -func TestShutdown_TimeoutLaisseLaBaseIntacte(t *testing.T) { - q := worker.New(10) - q.Start(1) - q.Submit(func() { time.Sleep(2 * time.Second) }) - - var buf strings.Builder - log := slog.New(slog.NewTextHandler(&buf, nil)) - - if shutdown(log, q, map[string]httpShutdowner{}, 50*time.Millisecond) { - t.Fatal("un drainage incomplet ne doit pas autoriser la fermeture de la base") - } - if !strings.Contains(buf.String(), "leaving database untouched") { - t.Errorf("le dépassement devrait être logué, obtenu %q", buf.String()) - } -} - -func TestShutdown_ErreurServeurNEmpechePasLeDrainage(t *testing.T) { - q := worker.New(10) - q.Start(1) - - var buf strings.Builder - log := slog.New(slog.NewTextHandler(&buf, nil)) - servers := map[string]httpShutdowner{ - "api": &fakeServer{err: errors.New("boom")}, - "prometheus": &fakeServer{}, - } - - if !shutdown(log, q, servers, 5*time.Second) { - t.Fatal("une erreur d'arrêt HTTP ne doit pas empêcher le drainage") - } - if !strings.Contains(buf.String(), "http server shutdown") { - t.Errorf("l'erreur devrait être loguée, obtenu %q", buf.String()) - } -} - -func TestShutdown_SansServeur(t *testing.T) { - q := worker.New(10) - q.Start(1) - - if !shutdown(discardLogger(), q, map[string]httpShutdowner{}, 5*time.Second) { - t.Fatal("l'absence de serveur ne doit pas empêcher un arrêt propre") - } -} diff --git a/conf/agent/config.exemple.yml b/conf/agent/config.exemple.yml index 72e88f8..0517eac 100644 --- a/conf/agent/config.exemple.yml +++ b/conf/agent/config.exemple.yml @@ -51,13 +51,6 @@ qemu: monitor_dir: "/run/two/vms/monitor" qmp_dir: "/run/two/vms/qmp" -# Consistency watchdog: periodically checks that resources marked running in the -# database still exist on the system, and reports the gaps. Read-only, never repairs. -# A persisting gap is reported at every tick (no deduplication). -watchdog: - enabled: true - interval_seconds: 60 - # Admin API (read-only DB inspection, loopback only) admin: enabled: false diff --git a/internal/api/agent/helpers_test.go b/internal/api/agent/helpers_test.go index c7162c7..206874c 100644 --- a/internal/api/agent/helpers_test.go +++ b/internal/api/agent/helpers_test.go @@ -23,5 +23,5 @@ func newTestServer(t *testing.T) (*Server, *badger.DB) { cfg := &configuration.Config{DefaultInterface: "br-test"} logger := slog.New(slog.NewTextHandler(io.Discard, nil)) d := dispatcher.New(q, db, cfg, logger) - return New(d, db, logger, "127.0.0.1:0"), db + return New(d, db, logger), db } diff --git a/internal/api/agent/server.go b/internal/api/agent/server.go index 2b892bd..4a0fba4 100644 --- a/internal/api/agent/server.go +++ b/internal/api/agent/server.go @@ -1,10 +1,8 @@ package agentapi import ( - "context" "crypto/rand" "encoding/hex" - "errors" "log/slog" "net/http" "time" @@ -17,11 +15,13 @@ type Server struct { dispatcher *dispatcher.Dispatcher db *badger.DB logger *slog.Logger - srv *http.Server } -func New(d *dispatcher.Dispatcher, db *badger.DB, logger *slog.Logger, address string) *Server { - s := &Server{dispatcher: d, db: db, logger: logger} +func New(d *dispatcher.Dispatcher, db *badger.DB, logger *slog.Logger) *Server { + return &Server{dispatcher: d, db: db, logger: logger} +} + +func (s *Server) Start(address string) { mux := http.NewServeMux() mux.HandleFunc("/vpcs", s.VpcsHandler) mux.HandleFunc("/vpcs/", s.VpcByNameHandler) @@ -29,21 +29,12 @@ func New(d *dispatcher.Dispatcher, db *badger.DB, logger *slog.Logger, address s mux.HandleFunc("/subnets/", s.SubnetByNameHandler) mux.HandleFunc("/vms", s.VmsHandler) mux.HandleFunc("/vms/", s.VmByNameHandler) - s.srv = &http.Server{Addr: address, Handler: s.logMiddleware(mux)} - return s -} - -func (s *Server) Start() { - s.logger.Info("API server listening", "address", s.srv.Addr) - if err := s.srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + s.logger.Info("API server listening", "address", address) + if err := http.ListenAndServe(address, s.logMiddleware(mux)); err != nil { s.logger.Error("API server stopped", "error", err) } } -func (s *Server) Shutdown(ctx context.Context) error { - return s.srv.Shutdown(ctx) -} - type statusWriter struct { http.ResponseWriter status int diff --git a/internal/config/agent/config_test.go b/internal/config/agent/config_test.go index a4d5b22..d0f3d31 100644 --- a/internal/config/agent/config_test.go +++ b/internal/config/agent/config_test.go @@ -81,68 +81,3 @@ database: t.Errorf("attendu %q, obtenu %q", "/opt/two/data", cfg.Database.Path) } } - -func TestLoadConfig_WatchdogDefauts(t *testing.T) { - path := writeYAML(t, "") - cfg, err := LoadConfig(path) - if err != nil { - t.Fatalf("LoadConfig a échoué : %v", err) - } - if cfg.Watchdog.Enabled { - t.Error("watchdog.enabled devrait être false par défaut") - } - if cfg.Watchdog.IntervalSeconds != 60 { - t.Errorf("watchdog.interval_seconds attendu 60, obtenu %d", cfg.Watchdog.IntervalSeconds) - } -} - -func TestLoadConfig_WatchdogValeursExplicites(t *testing.T) { - path := writeYAML(t, ` -watchdog: - enabled: true - interval_seconds: 30 -`) - cfg, err := LoadConfig(path) - if err != nil { - t.Fatalf("LoadConfig a échoué : %v", err) - } - if !cfg.Watchdog.Enabled { - t.Error("watchdog.enabled attendu true") - } - if cfg.Watchdog.IntervalSeconds != 30 { - t.Errorf("watchdog.interval_seconds attendu 30, obtenu %d", cfg.Watchdog.IntervalSeconds) - } -} - -func TestLoadConfig_WatchdogActiveSansIntervalle(t *testing.T) { - path := writeYAML(t, ` -watchdog: - enabled: true -`) - cfg, err := LoadConfig(path) - if err != nil { - t.Fatalf("LoadConfig a échoué : %v", err) - } - if !cfg.Watchdog.Enabled { - t.Error("watchdog.enabled attendu true") - } - if cfg.Watchdog.IntervalSeconds != 60 { - t.Errorf("watchdog.interval_seconds attendu 60 (défaut viper), obtenu %d", cfg.Watchdog.IntervalSeconds) - } -} - -func TestLoadConfig_ExempleFourniEstValide(t *testing.T) { - cfg, err := LoadConfig("../../../conf/agent/config.exemple.yml") - if err != nil { - t.Fatalf("config.exemple.yml illisible : %v", err) - } - if !cfg.Watchdog.Enabled { - t.Error("config.exemple.yml devrait activer le watchdog") - } - if cfg.Watchdog.IntervalSeconds != 60 { - t.Errorf("config.exemple.yml : interval_seconds attendu 60, obtenu %d", cfg.Watchdog.IntervalSeconds) - } - if cfg.QEMU.QMPDir == "" { - t.Error("config.exemple.yml devrait définir qemu.qmp_dir") - } -} diff --git a/internal/config/agent/struct.go b/internal/config/agent/struct.go index 92a8677..8b5d406 100644 --- a/internal/config/agent/struct.go +++ b/internal/config/agent/struct.go @@ -44,10 +44,6 @@ type Config struct { MonitorDir string `mapstructure:"monitor_dir"` QMPDir string `mapstructure:"qmp_dir"` } `mapstructure:"qemu"` - Watchdog struct { - Enabled bool `mapstructure:"enabled"` - IntervalSeconds int `mapstructure:"interval_seconds"` - } `mapstructure:"watchdog"` DefaultInterface string `mapstructure:"default_interface"` Interfaces map[string]string `mapstructure:"interfaces"` } @@ -73,8 +69,6 @@ func LoadConfig(path string) (*Config, error) { v.SetDefault("qemu.serial_dir", "/run/two/vms/serial") v.SetDefault("qemu.monitor_dir", "/run/two/vms/monitor") v.SetDefault("qemu.qmp_dir", "/run/two/vms/qmp") - v.SetDefault("watchdog.enabled", false) - v.SetDefault("watchdog.interval_seconds", 60) v.SetDefault("admin.enabled", false) v.SetDefault("admin.address", "127.0.0.1") v.SetDefault("admin.port", 9091) diff --git a/internal/dhcp/struct.go b/internal/dhcp/struct.go index bf6ace9..316667c 100644 --- a/internal/dhcp/struct.go +++ b/internal/dhcp/struct.go @@ -4,8 +4,6 @@ import ( "net" ) -const DefaultConfDir = "/etc/dnsmasq.d" - type Config struct { Network *net.IPNet VPCGateway net.IP // next-hop for VPCRoute (option 121) diff --git a/internal/netif/link.go b/internal/netif/link.go deleted file mode 100644 index 4e1a2b4..0000000 --- a/internal/netif/link.go +++ /dev/null @@ -1,15 +0,0 @@ -package netif - -import ( - "net" - - "github.com/vishvananda/netlink" -) - -func LinkIsUp(name string) (bool, error) { - link, err := netlink.LinkByName(name) - if err != nil { - return false, err - } - return link.Attrs().Flags&net.FlagUp != 0, nil -} diff --git a/internal/subnet/create.go b/internal/subnet/create.go index e7b0c70..75ef62e 100644 --- a/internal/subnet/create.go +++ b/internal/subnet/create.go @@ -138,7 +138,7 @@ func startDHCP(db *badger.DB, subnetName string, d subnetData) error { conf := dhcp.Config{ Network: d.cidr, Name: d.vpc + "_" + d.bridge, - ConfDir: dhcp.DefaultConfDir, + ConfDir: "/etc/dnsmasq.d", } switch d.mode { case "vxlan": diff --git a/internal/subnet/delete.go b/internal/subnet/delete.go index 7c26a71..1175fd5 100644 --- a/internal/subnet/delete.go +++ b/internal/subnet/delete.go @@ -3,9 +3,7 @@ package subnet import ( "fmt" "os" - "path/filepath" - "git.g3e.fr/syonad/two/internal/dhcp" "git.g3e.fr/syonad/two/internal/ebtables" "git.g3e.fr/syonad/two/internal/netif" "git.g3e.fr/syonad/two/internal/netns" @@ -64,7 +62,7 @@ func stopDHCP(db *badger.DB, subnetName string, d subnetData) error { } } - if err := os.Remove(filepath.Join(dhcp.DefaultConfDir, d.vpc+"_"+d.bridge+".conf")); err != nil && !os.IsNotExist(err) { + if err := os.Remove("/etc/dnsmasq.d/" + d.vpc + "_" + d.bridge + ".conf"); err != nil && !os.IsNotExist(err) { return fmt.Errorf("remove dnsmasq config: %w", err) } diff --git a/internal/watchdog/check_subnet.go b/internal/watchdog/check_subnet.go deleted file mode 100644 index fafb2ff..0000000 --- a/internal/watchdog/check_subnet.go +++ /dev/null @@ -1,134 +0,0 @@ -package watchdog - -import ( - "fmt" - "os" - "path/filepath" - "strconv" - "strings" - - "git.g3e.fr/syonad/two/internal/dhcp" - "git.g3e.fr/syonad/two/internal/netns" - "git.g3e.fr/syonad/two/internal/state" - "git.g3e.fr/syonad/two/internal/watchdog/notify" - "git.g3e.fr/syonad/two/pkg/db/kv" - - "github.com/dgraph-io/badger/v4" -) - -const ( - modeVxlan = "vxlan" - modeBridge = "bridge" -) - -func subnetIfaceNames(subnetName string) (hostVeth, nsVeth, bridge string, err error) { - parts := strings.SplitN(subnetName, "-", 2) - if len(parts) < 2 || parts[1] == "" { - return "", "", "", fmt.Errorf("subnet name %q has no identifier after the dash, interface names cannot be derived", subnetName) - } - id := parts[1] - return "v-" + id + "-e", "v-" + id + "-i", "br-" + id, nil -} - -func dnsmasqName(vpc, bridge string) string { - return vpc + "_" + bridge -} - -func CheckSubnets(db *badger.DB, u unitChecker, n notify.Notifier) error { - pairs, err := kv.ListByPrefix(db, prefixSubnet) - if err != nil { - return fmt.Errorf("watchdog: listing subnets: %w", err) - } - - for _, name := range resourceNames(pairs, prefixSubnet) { - st, err := state.Get(db, prefixSubnet+name) - if err != nil { - n.Notify(kindSubnet, name, fmt.Sprintf("state unreadable in database: %v", err)) - continue - } - if st != state.Running { - continue - } - checkSubnet(db, name, u, n) - } - return nil -} - -func checkSubnet(db *badger.DB, name string, u unitChecker, n notify.Notifier) { - hostVeth, nsVeth, bridge, err := subnetIfaceNames(name) - if err != nil { - n.Notify(kindSubnet, name, err.Error()) - return - } - - vpc, err := kv.GetFromDB(db, prefixSubnet+name+"/vpc") - if err != nil { - n.Notify(kindSubnet, name, fmt.Sprintf("vpc unreadable in database: %v", err)) - return - } - - mode, err := kv.GetFromDB(db, prefixSubnet+name+"/mode") - if err != nil { - n.Notify(kindSubnet, name, fmt.Sprintf("mode unreadable in database: %v", err)) - return - } - - if p := linkProblem(hostVeth); p != "" { - n.Notify(kindSubnet, name, p) - } - - switch mode { - case modeVxlan: - if p := linkProblem(bridge); p != "" { - n.Notify(kindSubnet, name, p+" (host)") - } - checkVxlanIface(db, name, n) - case modeBridge: - default: - n.Notify(kindSubnet, name, fmt.Sprintf("unknown mode %q", mode)) - } - - checkSubnetNetns(name, vpc, nsVeth, bridge, n) - - dnsName := dnsmasqName(vpc, bridge) - conf := filepath.Join(dhcp.DefaultConfDir, dnsName+".conf") - if _, err := os.Stat(conf); err != nil { - n.Notify(kindSubnet, name, fmt.Sprintf("dnsmasq config missing (%s): %v", conf, err)) - } - - checkUnit(kindSubnet, name, "dnsmasq@"+dnsName+".service", u, n) -} - -func checkVxlanIface(db *badger.DB, name string, n notify.Notifier) { - raw, err := kv.GetFromDB(db, prefixSubnet+name+"/vxlan_id") - if err != nil { - n.Notify(kindSubnet, name, fmt.Sprintf("vxlan_id unreadable in database: %v", err)) - return - } - id, err := strconv.Atoi(raw) - if err != nil { - n.Notify(kindSubnet, name, fmt.Sprintf("invalid vxlan_id %q: %v", raw, err)) - return - } - if p := linkProblem(fmt.Sprintf("vxlan-%d", id)); p != "" { - n.Notify(kindSubnet, name, p) - } -} - -func checkSubnetNetns(name, vpc, nsVeth, bridge string, n notify.Notifier) { - if !netns.Exist(vpc) { - n.Notify(kindSubnet, name, "netns "+vpc+" missing (/var/run/netns/"+vpc+")") - return - } - - if err := netns.Call(vpc, func() error { - for _, iface := range []string{nsVeth, bridge} { - if p := linkProblem(iface); p != "" { - n.Notify(kindSubnet, name, p+" (in netns "+vpc+")") - } - } - return nil - }); err != nil { - n.Notify(kindSubnet, name, fmt.Sprintf("cannot enter netns %s: %v", vpc, err)) - } -} diff --git a/internal/watchdog/check_subnet_test.go b/internal/watchdog/check_subnet_test.go deleted file mode 100644 index 6928e6f..0000000 --- a/internal/watchdog/check_subnet_test.go +++ /dev/null @@ -1,288 +0,0 @@ -package watchdog - -import ( - "errors" - "strings" - "testing" - - "git.g3e.fr/syonad/two/internal/state" - - "github.com/dgraph-io/badger/v4" -) - -func seedSubnet(t *testing.T, db *badger.DB, name, vpc, mode string) { - t.Helper() - seedResource(t, db, prefixSubnet, name, state.Running) - seedKV(t, db, prefixSubnet+name+"/vpc", vpc) - seedKV(t, db, prefixSubnet+name+"/mode", mode) -} - -func TestSubnetIfaceNames_NomStandard(t *testing.T) { - hostVeth, nsVeth, bridge, err := subnetIfaceNames("br-000042") - if err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - if hostVeth != "v-000042-e" || nsVeth != "v-000042-i" || bridge != "br-000042" { - t.Errorf("obtenu (%q, %q, %q), attendu (v-000042-e, v-000042-i, br-000042)", hostVeth, nsVeth, bridge) - } -} - -func TestSubnetIfaceNames_NomSansTiret(t *testing.T) { - if _, _, _, err := subnetIfaceNames("subnet"); err == nil { - t.Fatal("un nom sans tiret devrait produire une erreur, pas une panique") - } -} - -func TestSubnetIfaceNames_TiretFinal(t *testing.T) { - if _, _, _, err := subnetIfaceNames("br-"); err == nil { - t.Fatal("un identifiant vide devrait produire une erreur") - } -} - -func TestDnsmasqName(t *testing.T) { - if got := dnsmasqName("vp-admin", "br-000000"); got != "vp-admin_br-000000" { - t.Errorf("dnsmasqName = %q, attendu vp-admin_br-000000", got) - } -} - -func TestCheckSubnets_BaseVide(t *testing.T) { - db := newTestDB(t) - r := &recorder{} - - if err := CheckSubnets(db, newFakeUnits(), r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - if len(r.calls) != 0 { - t.Errorf("aucune notification attendue, obtenu %v", r.calls) - } -} - -func TestCheckSubnets_IgnoreLesEtatsNonRunning(t *testing.T) { - db := newTestDB(t) - for _, s := range []state.State{state.Creating, state.Deleting, state.Error, state.Deleted} { - seedResource(t, db, prefixSubnet, "br-"+string(s), s) - } - r := &recorder{} - - if err := CheckSubnets(db, newFakeUnits(), r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - if len(r.calls) != 0 { - t.Errorf("aucune notification attendue, obtenu %v", r.calls) - } -} - -func TestCheckSubnets_VPCManquantEnBase(t *testing.T) { - db := newTestDB(t) - seedResource(t, db, prefixSubnet, "br-000042", state.Running) - r := &recorder{} - - if err := CheckSubnets(db, newFakeUnits(), r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - - got := r.forName("br-000042") - if len(got) != 1 { - t.Fatalf("attendu 1 notification, obtenu %d : %v", len(got), got) - } - if !strings.Contains(got[0].problem, "vpc unreadable") { - t.Errorf("problem = %q, devrait porter sur le vpc", got[0].problem) - } -} - -func TestCheckSubnets_ModeManquantEnBase(t *testing.T) { - db := newTestDB(t) - seedResource(t, db, prefixSubnet, "br-000042", state.Running) - seedKV(t, db, prefixSubnet+"br-000042/vpc", "vp-admin") - r := &recorder{} - - if err := CheckSubnets(db, newFakeUnits(), r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - - if !r.hasProblemContaining("mode unreadable") { - t.Errorf("devrait signaler un mode illisible, obtenu %v", r.calls) - } -} - -func TestCheckSubnets_ModeInconnu(t *testing.T) { - db := newTestDB(t) - seedSubnet(t, db, "br-000042", "vp-admin", "macvlan") - r := &recorder{} - - if err := CheckSubnets(db, newFakeUnits(), r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - - if !r.hasProblemContaining(`unknown mode "macvlan"`) { - t.Errorf("devrait signaler un mode inconnu, obtenu %v", r.calls) - } -} - -func TestCheckSubnets_ModeBridgeNeVerifiePasDeVxlan(t *testing.T) { - db := newTestDB(t) - seedSubnet(t, db, "br-000042", "vp-admin", modeBridge) - r := &recorder{} - - if err := CheckSubnets(db, newFakeUnits(), r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - - for _, c := range r.calls { - if strings.Contains(c.problem, "vxlan") { - t.Errorf("un subnet en mode bridge ne doit rien signaler sur vxlan: %q", c.problem) - } - if strings.Contains(c.problem, "(host)") { - t.Errorf("un subnet en mode bridge n'a pas de bridge host: %q", c.problem) - } - } -} - -func TestCheckSubnets_ModeVxlanSansVxlanID(t *testing.T) { - db := newTestDB(t) - seedSubnet(t, db, "br-000042", "vp-admin", modeVxlan) - r := &recorder{} - - if err := CheckSubnets(db, newFakeUnits(), r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - - if !r.hasProblemContaining("vxlan_id unreadable") { - t.Errorf("devrait signaler un vxlan_id illisible, obtenu %v", r.calls) - } -} - -func TestCheckSubnets_ModeVxlanVxlanIDInvalide(t *testing.T) { - db := newTestDB(t) - seedSubnet(t, db, "br-000042", "vp-admin", modeVxlan) - seedKV(t, db, prefixSubnet+"br-000042/vxlan_id", "pas-un-nombre") - r := &recorder{} - - if err := CheckSubnets(db, newFakeUnits(), r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - - if !r.hasProblemContaining("invalid vxlan_id") { - t.Errorf("devrait signaler un vxlan_id invalide, obtenu %v", r.calls) - } -} - -func TestCheckSubnets_ModeVxlanVerifieLInterfaceVxlan(t *testing.T) { - db := newTestDB(t) - seedSubnet(t, db, "br-000042", "vp-admin", modeVxlan) - seedKV(t, db, prefixSubnet+"br-000042/vxlan_id", "42") - r := &recorder{} - - if err := CheckSubnets(db, newFakeUnits(), r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - - if !r.hasProblemContaining("vxlan-42") { - t.Errorf("devrait signaler l'interface vxlan-42, obtenu %v", r.calls) - } -} - -func TestCheckSubnets_ConfigDnsmasqAbsente(t *testing.T) { - db := newTestDB(t) - seedSubnet(t, db, "br-000042", "vp-admin", modeBridge) - r := &recorder{} - - if err := CheckSubnets(db, newFakeUnits(), r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - - if !r.hasProblemContaining("dnsmasq config missing") { - t.Errorf("devrait signaler la config dnsmasq absente, obtenu %v", r.calls) - } - if !r.hasProblemContaining("vp-admin_br-000042.conf") { - t.Errorf("devrait nommer le fichier attendu, obtenu %v", r.calls) - } -} - -func TestCheckSubnets_UnitDnsmasqInterrogee(t *testing.T) { - db := newTestDB(t) - seedSubnet(t, db, "br-000042", "vp-admin", modeBridge) - u := newFakeUnits().active("dnsmasq@vp-admin_br-000042.service") - r := &recorder{} - - if err := CheckSubnets(db, u, r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - - if len(u.asked) != 1 || u.asked[0] != "dnsmasq@vp-admin_br-000042.service" { - t.Errorf("units interrogées = %v, attendu [dnsmasq@vp-admin_br-000042.service]", u.asked) - } - if r.hasProblemContaining("unit ") { - t.Errorf("une unit active ne doit rien signaler, obtenu %v", r.calls) - } -} - -func TestCheckSubnets_UnitDnsmasqInactive(t *testing.T) { - db := newTestDB(t) - seedSubnet(t, db, "br-000042", "vp-admin", modeBridge) - u := newFakeUnits().inactive("dnsmasq@vp-admin_br-000042.service", "failed") - r := &recorder{} - - if err := CheckSubnets(db, u, r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - - if !r.hasProblemContaining("unit dnsmasq@vp-admin_br-000042.service inactive (failed)") { - t.Errorf("devrait signaler l'unit inactive, obtenu %v", r.calls) - } -} - -func TestCheckSubnets_UnitIllisible(t *testing.T) { - db := newTestDB(t) - seedSubnet(t, db, "br-000042", "vp-admin", modeBridge) - u := newFakeUnits().failing("dnsmasq@vp-admin_br-000042.service", errors.New("dbus indisponible")) - r := &recorder{} - - if err := CheckSubnets(db, u, r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - - if !r.hasProblemContaining("unit dnsmasq@vp-admin_br-000042.service unreadable") { - t.Errorf("devrait signaler l'unit illisible, obtenu %v", r.calls) - } -} - -func TestCheckSubnets_SansUnitCheckerPasDeVerificationDUnit(t *testing.T) { - db := newTestDB(t) - seedSubnet(t, db, "br-000042", "vp-admin", modeBridge) - r := &recorder{} - - if err := CheckSubnets(db, nil, r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - - if r.hasProblemContaining("unit ") { - t.Errorf("sans unitChecker, aucune notification d'unit attendue, obtenu %v", r.calls) - } -} - -func TestCheckSubnets_EtatCorrompuNInterrompPasLaBoucle(t *testing.T) { - db := newTestDB(t) - seedKV(t, db, prefixSubnet+"br-corrompu"+stateSuffix, "n_importe_quoi") - seedSubnet(t, db, "br-000042", "vp-admin", modeBridge) - r := &recorder{} - - if err := CheckSubnets(db, newFakeUnits(), r); err != nil { - t.Fatalf("un état corrompu ne doit pas faire échouer CheckSubnets: %v", err) - } - - if !strings.Contains(strings.Join(problems(r.forName("br-corrompu")), " "), "state unreadable") { - t.Errorf("devrait signaler l'état corrompu, obtenu %v", r.calls) - } - if len(r.forName("br-000042")) == 0 { - t.Error("une clé corrompue ne doit pas empêcher l'examen des subnets suivants") - } -} - -func problems(ns []notification) []string { - out := make([]string, 0, len(ns)) - for _, n := range ns { - out = append(out, n.problem) - } - return out -} diff --git a/internal/watchdog/check_vm.go b/internal/watchdog/check_vm.go deleted file mode 100644 index 3c850a3..0000000 --- a/internal/watchdog/check_vm.go +++ /dev/null @@ -1,99 +0,0 @@ -package watchdog - -import ( - "errors" - "fmt" - "path/filepath" - "strconv" - - configuration "git.g3e.fr/syonad/two/internal/config/agent" - "git.g3e.fr/syonad/two/internal/netns" - "git.g3e.fr/syonad/two/internal/qemu" - "git.g3e.fr/syonad/two/internal/qmp" - "git.g3e.fr/syonad/two/internal/state" - "git.g3e.fr/syonad/two/internal/watchdog/notify" - "git.g3e.fr/syonad/two/pkg/db/kv" - - "github.com/dgraph-io/badger/v4" -) - -func tapName(tapID int) string { - return fmt.Sprintf("tap%d", tapID) -} - -func CheckVMs(db *badger.DB, cfg *configuration.Config, u unitChecker, n notify.Notifier) error { - if cfg == nil { - return errors.New("watchdog: configuration required to check vms") - } - - pairs, err := kv.ListByPrefix(db, prefixVM) - if err != nil { - return fmt.Errorf("watchdog: listing vms: %w", err) - } - - for _, name := range resourceNames(pairs, prefixVM) { - st, err := state.Get(db, prefixVM+name) - if err != nil { - n.Notify(kindVM, name, fmt.Sprintf("state unreadable in database: %v", err)) - continue - } - if st != state.Running { - continue - } - checkVM(db, cfg, name, u, n) - } - return nil -} - -func checkVM(db *badger.DB, cfg *configuration.Config, name string, u unitChecker, n notify.Notifier) { - subnetName, err := kv.GetFromDB(db, prefixVM+name+"/subnet") - if err != nil { - n.Notify(kindVM, name, fmt.Sprintf("subnet unreadable in database: %v", err)) - return - } - - vpc, err := kv.GetFromDB(db, prefixSubnet+subnetName+"/vpc") - if err != nil { - n.Notify(kindVM, name, fmt.Sprintf("vpc of subnet %s unreadable in database: %v", subnetName, err)) - return - } - - checkVMTap(db, name, vpc, n) - checkVMQemu(cfg, name, n) - checkUnit(kindVM, name, "metadata@"+name+".service", u, n) - checkUnit(kindVM, name, qemu.ScopeName(name), u, n) -} - -func checkVMTap(db *badger.DB, name, vpc string, n notify.Notifier) { - raw, err := kv.GetFromDB(db, prefixVM+name+"/tap_id") - if err != nil { - n.Notify(kindVM, name, fmt.Sprintf("tap_id unreadable in database: %v", err)) - return - } - tapID, err := strconv.Atoi(raw) - if err != nil { - n.Notify(kindVM, name, fmt.Sprintf("invalid tap_id %q: %v", raw, err)) - return - } - - if !netns.Exist(vpc) { - n.Notify(kindVM, name, "netns "+vpc+" missing (/var/run/netns/"+vpc+")") - return - } - - if err := netns.Call(vpc, func() error { - if p := linkProblem(tapName(tapID)); p != "" { - n.Notify(kindVM, name, p+" (in netns "+vpc+")") - } - return nil - }); err != nil { - n.Notify(kindVM, name, fmt.Sprintf("cannot enter netns %s: %v", vpc, err)) - } -} - -func checkVMQemu(cfg *configuration.Config, name string, n notify.Notifier) { - sock := filepath.Join(cfg.QEMU.QMPDir, name+".sock") - if _, err := qmp.Send(sock, nil); err != nil { - n.Notify(kindVM, name, fmt.Sprintf("qemu not responding on %s: %v", sock, err)) - } -} diff --git a/internal/watchdog/check_vm_test.go b/internal/watchdog/check_vm_test.go deleted file mode 100644 index eb62b59..0000000 --- a/internal/watchdog/check_vm_test.go +++ /dev/null @@ -1,246 +0,0 @@ -package watchdog - -import ( - "strings" - "testing" - - configuration "git.g3e.fr/syonad/two/internal/config/agent" - "git.g3e.fr/syonad/two/internal/state" - - "github.com/dgraph-io/badger/v4" -) - -func testCfg(t *testing.T) *configuration.Config { - t.Helper() - cfg := &configuration.Config{} - cfg.QEMU.QMPDir = t.TempDir() - return cfg -} - -func seedVM(t *testing.T, db *badger.DB, name, subnetName, vpc, tapID string) { - t.Helper() - seedResource(t, db, prefixVM, name, state.Running) - seedKV(t, db, prefixVM+name+"/subnet", subnetName) - seedKV(t, db, prefixSubnet+subnetName+"/vpc", vpc) - if tapID != "" { - seedKV(t, db, prefixVM+name+"/tap_id", tapID) - } -} - -func TestTapName(t *testing.T) { - if got := tapName(12345678); got != "tap12345678" { - t.Errorf("tapName = %q, attendu tap12345678", got) - } -} - -func TestCheckVMs_ConfigNil(t *testing.T) { - db := newTestDB(t) - r := &recorder{} - - err := CheckVMs(db, nil, newFakeUnits(), r) - if err == nil { - t.Fatal("une config nil doit produire une erreur, pas une panique") - } - if len(r.calls) != 0 { - t.Errorf("aucune notification attendue, obtenu %v", r.calls) - } -} - -func TestCheckVMs_BaseVide(t *testing.T) { - db := newTestDB(t) - r := &recorder{} - - if err := CheckVMs(db, testCfg(t), newFakeUnits(), r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - if len(r.calls) != 0 { - t.Errorf("aucune notification attendue, obtenu %v", r.calls) - } -} - -func TestCheckVMs_IgnoreLesEtatsNonRunning(t *testing.T) { - db := newTestDB(t) - for _, s := range []state.State{state.Creating, state.Deleting, state.Error, state.Deleted} { - seedResource(t, db, prefixVM, "i-"+string(s), s) - } - r := &recorder{} - - if err := CheckVMs(db, testCfg(t), newFakeUnits(), r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - if len(r.calls) != 0 { - t.Errorf("aucune notification attendue, obtenu %v", r.calls) - } -} - -func TestCheckVMs_SubnetManquantEnBase(t *testing.T) { - db := newTestDB(t) - seedResource(t, db, prefixVM, "i-test1", state.Running) - r := &recorder{} - - if err := CheckVMs(db, testCfg(t), newFakeUnits(), r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - - got := r.forName("i-test1") - if len(got) != 1 { - t.Fatalf("attendu 1 notification, obtenu %d : %v", len(got), got) - } - if !strings.Contains(got[0].problem, "subnet unreadable") { - t.Errorf("problem = %q, devrait porter sur le subnet", got[0].problem) - } -} - -func TestCheckVMs_VPCDuSubnetManquant(t *testing.T) { - db := newTestDB(t) - seedResource(t, db, prefixVM, "i-test1", state.Running) - seedKV(t, db, prefixVM+"i-test1/subnet", "br-000042") - r := &recorder{} - - if err := CheckVMs(db, testCfg(t), newFakeUnits(), r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - - if !r.hasProblemContaining("vpc of subnet br-000042 unreadable") { - t.Errorf("devrait signaler le vpc introuvable, obtenu %v", r.calls) - } -} - -func TestCheckVMs_TapIDManquant(t *testing.T) { - db := newTestDB(t) - seedVM(t, db, "i-test1", "br-000042", "vp-admin", "") - r := &recorder{} - - if err := CheckVMs(db, testCfg(t), newFakeUnits(), r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - - if !r.hasProblemContaining("tap_id unreadable") { - t.Errorf("devrait signaler un tap_id illisible, obtenu %v", r.calls) - } -} - -func TestCheckVMs_TapIDInvalide(t *testing.T) { - db := newTestDB(t) - seedVM(t, db, "i-test1", "br-000042", "vp-admin", "pas-un-nombre") - r := &recorder{} - - if err := CheckVMs(db, testCfg(t), newFakeUnits(), r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - - if !r.hasProblemContaining("invalid tap_id") { - t.Errorf("devrait signaler un tap_id invalide, obtenu %v", r.calls) - } -} - -func TestCheckVMs_QemuNeRepondPas(t *testing.T) { - db := newTestDB(t) - seedVM(t, db, "i-test1", "br-000042", "vp-admin", "12345678") - cfg := testCfg(t) - r := &recorder{} - - if err := CheckVMs(db, cfg, newFakeUnits(), r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - - if !r.hasProblemContaining("qemu not responding") { - t.Errorf("devrait signaler que qemu ne répond pas, obtenu %v", r.calls) - } - if !r.hasProblemContaining("i-test1.sock") { - t.Errorf("devrait nommer la socket attendue, obtenu %v", r.calls) - } -} - -func TestCheckVMs_UnitsInterrogees(t *testing.T) { - db := newTestDB(t) - seedVM(t, db, "i-test1", "br-000042", "vp-admin", "12345678") - u := newFakeUnits().active("metadata@i-test1.service").active("two-vm-i-test1.scope") - r := &recorder{} - - if err := CheckVMs(db, testCfg(t), u, r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - - want := map[string]bool{"metadata@i-test1.service": false, "two-vm-i-test1.scope": false} - for _, asked := range u.asked { - if _, ok := want[asked]; !ok { - t.Errorf("unit inattendue interrogée: %q", asked) - continue - } - want[asked] = true - } - for unit, seen := range want { - if !seen { - t.Errorf("unit %q non interrogée (interrogées: %v)", unit, u.asked) - } - } - if r.hasProblemContaining("unit ") { - t.Errorf("des units actives ne doivent rien signaler, obtenu %v", r.calls) - } -} - -func TestCheckVMs_MetadataInactive(t *testing.T) { - db := newTestDB(t) - seedVM(t, db, "i-test1", "br-000042", "vp-admin", "12345678") - u := newFakeUnits().inactive("metadata@i-test1.service", "dead").active("two-vm-i-test1.scope") - r := &recorder{} - - if err := CheckVMs(db, testCfg(t), u, r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - - if !r.hasProblemContaining("unit metadata@i-test1.service inactive (dead)") { - t.Errorf("devrait signaler metadata inactive, obtenu %v", r.calls) - } -} - -func TestCheckVMs_ScopeAbsent(t *testing.T) { - db := newTestDB(t) - seedVM(t, db, "i-test1", "br-000042", "vp-admin", "12345678") - u := newFakeUnits().active("metadata@i-test1.service") - r := &recorder{} - - if err := CheckVMs(db, testCfg(t), u, r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - - if !r.hasProblemContaining("unit two-vm-i-test1.scope inactive") { - t.Errorf("devrait signaler le scope absent, obtenu %v", r.calls) - } -} - -func TestCheckVMs_EtatCorrompuNInterrompPasLaBoucle(t *testing.T) { - db := newTestDB(t) - seedKV(t, db, prefixVM+"i-corrompu"+stateSuffix, "n_importe_quoi") - seedVM(t, db, "i-test1", "br-000042", "vp-admin", "12345678") - r := &recorder{} - - if err := CheckVMs(db, testCfg(t), newFakeUnits(), r); err != nil { - t.Fatalf("un état corrompu ne doit pas faire échouer CheckVMs: %v", err) - } - - if !strings.Contains(strings.Join(problems(r.forName("i-corrompu")), " "), "state unreadable") { - t.Errorf("devrait signaler l'état corrompu, obtenu %v", r.calls) - } - if len(r.forName("i-test1")) == 0 { - t.Error("une clé corrompue ne doit pas empêcher l'examen des VMs suivantes") - } -} - -func TestCheckVMs_ClesDeDisqueNeCreentPasDeFausseVM(t *testing.T) { - db := newTestDB(t) - seedVM(t, db, "i-test1", "br-000042", "vp-admin", "12345678") - seedKV(t, db, prefixVM+"i-test1/disk/vda/state", "attached") - r := &recorder{} - - if err := CheckVMs(db, testCfg(t), newFakeUnits(), r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - - for _, c := range r.calls { - if c.name != "i-test1" { - t.Errorf("ressource inattendue signalée: %q", c.name) - } - } -} diff --git a/internal/watchdog/check_vpc.go b/internal/watchdog/check_vpc.go deleted file mode 100644 index c344561..0000000 --- a/internal/watchdog/check_vpc.go +++ /dev/null @@ -1,71 +0,0 @@ -package watchdog - -import ( - "fmt" - "strings" - - "git.g3e.fr/syonad/two/internal/netns" - "git.g3e.fr/syonad/two/internal/state" - "git.g3e.fr/syonad/two/internal/watchdog/notify" - "git.g3e.fr/syonad/two/pkg/db/kv" - - "github.com/dgraph-io/badger/v4" -) - -const vpcBridge = "br-public" - -func vpcIfaceNames(vpcName string) (host, ns string, err error) { - parts := strings.SplitN(vpcName, "-", 2) - if len(parts) < 2 || parts[1] == "" { - return "", "", fmt.Errorf("vpc name %q has no identifier after the dash, interface names cannot be derived", vpcName) - } - return "vp-" + parts[1] + "-e", "vp-" + parts[1] + "-i", nil -} - -func CheckVPCs(db *badger.DB, n notify.Notifier) error { - pairs, err := kv.ListByPrefix(db, prefixVPC) - if err != nil { - return fmt.Errorf("watchdog: listing vpcs: %w", err) - } - - for _, name := range resourceNames(pairs, prefixVPC) { - st, err := state.Get(db, prefixVPC+name) - if err != nil { - n.Notify(kindVPC, name, fmt.Sprintf("state unreadable in database: %v", err)) - continue - } - if st != state.Running { - continue - } - checkVPC(name, n) - } - return nil -} - -func checkVPC(name string, n notify.Notifier) { - hostVeth, nsVeth, err := vpcIfaceNames(name) - if err != nil { - n.Notify(kindVPC, name, err.Error()) - return - } - - if !netns.Exist(name) { - n.Notify(kindVPC, name, "netns missing (/var/run/netns/"+name+")") - return - } - - if p := linkProblem(hostVeth); p != "" { - n.Notify(kindVPC, name, p) - } - - if err := netns.Call(name, func() error { - for _, iface := range []string{nsVeth, vpcBridge} { - if p := linkProblem(iface); p != "" { - n.Notify(kindVPC, name, p+" (in netns)") - } - } - return nil - }); err != nil { - n.Notify(kindVPC, name, fmt.Sprintf("cannot enter netns: %v", err)) - } -} diff --git a/internal/watchdog/check_vpc_test.go b/internal/watchdog/check_vpc_test.go deleted file mode 100644 index 3ae8a07..0000000 --- a/internal/watchdog/check_vpc_test.go +++ /dev/null @@ -1,177 +0,0 @@ -package watchdog - -import ( - "strings" - "testing" - - "git.g3e.fr/syonad/two/internal/state" - "git.g3e.fr/syonad/two/pkg/db/kv" -) - -func TestVPCIfaceNames_NomStandard(t *testing.T) { - host, ns, err := vpcIfaceNames("vp-admin") - if err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - if host != "vp-admin-e" || ns != "vp-admin-i" { - t.Errorf("obtenu (%q, %q), attendu (vp-admin-e, vp-admin-i)", host, ns) - } -} - -func TestVPCIfaceNames_IdentifiantNumerique(t *testing.T) { - host, ns, err := vpcIfaceNames("vpc-000003") - if err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - if host != "vp-000003-e" || ns != "vp-000003-i" { - t.Errorf("obtenu (%q, %q), attendu (vp-000003-e, vp-000003-i)", host, ns) - } -} - -func TestVPCIfaceNames_PlusieursTirets(t *testing.T) { - host, _, err := vpcIfaceNames("vp-admin-prod") - if err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - if host != "vp-admin-prod-e" { - t.Errorf("obtenu %q, attendu vp-admin-prod-e", host) - } -} - -func TestVPCIfaceNames_NomSansTiret(t *testing.T) { - if _, _, err := vpcIfaceNames("admin"); err == nil { - t.Fatal("un nom sans tiret devrait produire une erreur, pas une panique") - } -} - -func TestVPCIfaceNames_TiretFinal(t *testing.T) { - if _, _, err := vpcIfaceNames("vp-"); err == nil { - t.Fatal("un identifiant vide devrait produire une erreur") - } -} - -func TestVPCIfaceNames_NomVide(t *testing.T) { - if _, _, err := vpcIfaceNames(""); err == nil { - t.Fatal("un nom vide devrait produire une erreur") - } -} - -func TestCheckVPCs_BaseVide(t *testing.T) { - db := newTestDB(t) - r := &recorder{} - - if err := CheckVPCs(db, r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - if len(r.calls) != 0 { - t.Errorf("aucune notification attendue, obtenu %v", r.calls) - } -} - -func TestCheckVPCs_IgnoreLesEtatsNonRunning(t *testing.T) { - db := newTestDB(t) - for _, s := range []state.State{state.Creating, state.Deleting, state.Error, state.Deleted} { - seedResource(t, db, prefixVPC, "vp-"+string(s), s) - } - r := &recorder{} - - if err := CheckVPCs(db, r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - if len(r.calls) != 0 { - t.Errorf("aucune notification attendue pour des états non-running, obtenu %v", r.calls) - } -} - -func TestCheckVPCs_SignaleUnVPCRunningAbsent(t *testing.T) { - db := newTestDB(t) - seedResource(t, db, prefixVPC, "vp-fantome", state.Running) - r := &recorder{} - - if err := CheckVPCs(db, r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - - got := r.forName("vp-fantome") - if len(got) == 0 { - t.Fatal("un VPC running inexistant sur le système doit être signalé") - } - for _, c := range got { - if c.kind != kindVPC { - t.Errorf("kind = %q, attendu %q", c.kind, kindVPC) - } - if c.problem == "" { - t.Error("problem ne doit pas être vide") - } - } -} - -func TestCheckVPCs_EtatCorrompu(t *testing.T) { - db := newTestDB(t) - if err := kv.AddInDB(db, prefixVPC+"vp-corrompu"+stateSuffix, "n_importe_quoi"); err != nil { - t.Fatalf("préparation: %v", err) - } - seedResource(t, db, prefixVPC, "vp-suivant", state.Running) - r := &recorder{} - - if err := CheckVPCs(db, r); err != nil { - t.Fatalf("un état corrompu ne doit pas faire échouer CheckVPCs: %v", err) - } - - got := r.forName("vp-corrompu") - if len(got) != 1 { - t.Fatalf("attendu 1 notification pour l'état corrompu, obtenu %d", len(got)) - } - if !strings.Contains(got[0].problem, "state unreadable") { - t.Errorf("problem = %q, devrait mentionner un état illisible", got[0].problem) - } - if len(r.forName("vp-suivant")) == 0 { - t.Error("une clé corrompue ne doit pas empêcher l'examen des VPC suivants") - } -} - -func TestCheckVPCs_NomIndeductibleNePaniquePas(t *testing.T) { - db := newTestDB(t) - seedResource(t, db, prefixVPC, "admin", state.Running) - r := &recorder{} - - if err := CheckVPCs(db, r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - - got := r.forName("admin") - if len(got) != 1 { - t.Fatalf("attendu 1 notification, obtenu %d : %v", len(got), got) - } - if !strings.Contains(got[0].problem, "cannot be derived") { - t.Errorf("problem = %q, devrait porter sur les interfaces indéductibles", got[0].problem) - } -} - -func TestCheckVPCs_OrdreDeterministe(t *testing.T) { - db := newTestDB(t) - for _, name := range []string{"vp-c", "vp-a", "vp-b"} { - seedResource(t, db, prefixVPC, name, state.Running) - } - - var first []string - for i := range 5 { - r := &recorder{} - if err := CheckVPCs(db, r); err != nil { - t.Fatalf("erreur inattendue: %v", err) - } - var order []string - for _, c := range r.calls { - if len(order) == 0 || order[len(order)-1] != c.name { - order = append(order, c.name) - } - } - if i == 0 { - first = order - continue - } - if strings.Join(order, ",") != strings.Join(first, ",") { - t.Fatalf("itération %d : ordre %v, attendu %v", i, order, first) - } - } -} diff --git a/internal/watchdog/helpers_test.go b/internal/watchdog/helpers_test.go deleted file mode 100644 index 891eeba..0000000 --- a/internal/watchdog/helpers_test.go +++ /dev/null @@ -1,110 +0,0 @@ -package watchdog - -import ( - "strings" - "testing" - - "git.g3e.fr/syonad/two/internal/state" - "git.g3e.fr/syonad/two/internal/watchdog/notify" - "git.g3e.fr/syonad/two/pkg/db/kv" - "git.g3e.fr/syonad/two/pkg/systemd" - - "github.com/dgraph-io/badger/v4" -) - -func seedKV(t *testing.T, db *badger.DB, key, value string) { - t.Helper() - if err := kv.AddInDB(db, key, value); err != nil { - t.Fatalf("seedKV %s: %v", key, err) - } -} - -type notification struct { - kind string - name string - problem string -} - -type recorder struct { - calls []notification -} - -var _ notify.Notifier = (*recorder)(nil) - -func (r *recorder) Notify(kind, name, problem string) { - r.calls = append(r.calls, notification{kind: kind, name: name, problem: problem}) -} - -func (r *recorder) forName(name string) []notification { - var out []notification - for _, c := range r.calls { - if c.name == name { - out = append(out, c) - } - } - return out -} - -func (r *recorder) hasProblemContaining(substr string) bool { - for _, c := range r.calls { - if strings.Contains(c.problem, substr) { - return true - } - } - return false -} - -type fakeUnits struct { - status map[string]*systemd.ServiceStatus - err map[string]error - asked []string -} - -var _ unitChecker = (*fakeUnits)(nil) - -func newFakeUnits() *fakeUnits { - return &fakeUnits{ - status: map[string]*systemd.ServiceStatus{}, - err: map[string]error{}, - } -} - -func (f *fakeUnits) active(unit string) *fakeUnits { - f.status[unit] = &systemd.ServiceStatus{Name: unit, LoadState: "loaded", ActiveState: "active", SubState: "running"} - return f -} - -func (f *fakeUnits) inactive(unit, sub string) *fakeUnits { - f.status[unit] = &systemd.ServiceStatus{Name: unit, LoadState: "loaded", ActiveState: "inactive", SubState: sub} - return f -} - -func (f *fakeUnits) failing(unit string, err error) *fakeUnits { - f.err[unit] = err - return f -} - -func (f *fakeUnits) Status(unit string) (*systemd.ServiceStatus, error) { - f.asked = append(f.asked, unit) - if err, ok := f.err[unit]; ok { - return nil, err - } - if st, ok := f.status[unit]; ok { - return st, nil - } - return &systemd.ServiceStatus{Name: unit, LoadState: "not-found", ActiveState: "inactive", SubState: "dead"}, nil -} - -func newTestDB(t *testing.T) *badger.DB { - t.Helper() - db := kv.InitDB(kv.Config{Path: t.TempDir()}, false) - t.Cleanup(func() { db.Close() }) - return db -} - -func seedResource(t *testing.T, db *badger.DB, prefix, name string, s state.State) { - t.Helper() - if err := state.Set(db, prefix+name, s); err != nil { - t.Fatalf("seedResource %s%s: %v", prefix, name, err) - } -} diff --git a/internal/watchdog/notify/notify.go b/internal/watchdog/notify/notify.go deleted file mode 100644 index fbf75e2..0000000 --- a/internal/watchdog/notify/notify.go +++ /dev/null @@ -1,5 +0,0 @@ -package notify - -type Notifier interface { - Notify(kind, name, problem string) -} diff --git a/internal/watchdog/notify/stderr.go b/internal/watchdog/notify/stderr.go deleted file mode 100644 index cac8174..0000000 --- a/internal/watchdog/notify/stderr.go +++ /dev/null @@ -1,24 +0,0 @@ -package notify - -import "log/slog" - -var _ Notifier = (*StderrNotifier)(nil) - -type StderrNotifier struct { - logger *slog.Logger -} - -func NewStderr(l *slog.Logger) *StderrNotifier { - if l == nil { - l = slog.Default() - } - return &StderrNotifier{logger: l} -} - -func (n *StderrNotifier) Notify(kind, name, problem string) { - n.logger.Error("watchdog: inconsistency detected", - "kind", kind, - "name", name, - "problem", problem, - ) -} diff --git a/internal/watchdog/notify/stderr_test.go b/internal/watchdog/notify/stderr_test.go deleted file mode 100644 index cfc45a4..0000000 --- a/internal/watchdog/notify/stderr_test.go +++ /dev/null @@ -1,120 +0,0 @@ -package notify - -import ( - "bytes" - "encoding/json" - "log/slog" - "strings" - "testing" -) - -func newTestNotifier(t *testing.T) (*StderrNotifier, *bytes.Buffer) { - t.Helper() - var buf bytes.Buffer - l := slog.New(slog.NewJSONHandler(&buf, &slog.HandlerOptions{Level: slog.LevelError})) - return NewStderr(l), &buf -} - -func decodeLines(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 TestNotify_EmetLesTroisChampsStructures(t *testing.T) { - n, buf := newTestNotifier(t) - - n.Notify("vm", "i-test1", "tap12 absente du netns vp-admin") - - lines := decodeLines(t, buf) - if len(lines) != 1 { - t.Fatalf("attendu 1 ligne, obtenu %d", len(lines)) - } - - for field, want := range map[string]string{ - "kind": "vm", - "name": "i-test1", - "problem": "tap12 absente du netns vp-admin", - } { - got, ok := lines[0][field] - if !ok { - t.Errorf("champ %q absent de la sortie", field) - continue - } - if got != want { - t.Errorf("champ %q = %v, attendu %q", field, got, want) - } - } -} - -func TestNotify_NiveauError(t *testing.T) { - n, buf := newTestNotifier(t) - - n.Notify("vpc", "vp-admin", "netns absent") - - lines := decodeLines(t, buf) - if len(lines) != 1 { - t.Fatalf("attendu 1 ligne, obtenu %d", len(lines)) - } - if lines[0]["level"] != "ERROR" { - t.Errorf("level = %v, attendu ERROR", lines[0]["level"]) - } -} - -func TestNotify_MessageNeContientPasLesValeurs(t *testing.T) { - n, buf := newTestNotifier(t) - - n.Notify("subnet", "br-000042", "bridge down") - - lines := decodeLines(t, buf) - msg, _ := lines[0]["msg"].(string) - for _, v := range []string{"br-000042", "bridge down"} { - if strings.Contains(msg, v) { - t.Errorf("msg %q ne devrait pas contenir la valeur %q (elle doit être un attribut)", msg, v) - } - } -} - -func TestNotify_AucuneDeduplication(t *testing.T) { - n, buf := newTestNotifier(t) - - for range 3 { - n.Notify("vm", "i-test1", "qemu ne répond pas sur QMP") - } - - if lines := decodeLines(t, buf); len(lines) != 3 { - t.Errorf("attendu 3 lignes (une par appel), obtenu %d", len(lines)) - } -} - -func TestNotify_ChampsVides(t *testing.T) { - n, buf := newTestNotifier(t) - - n.Notify("", "", "") - - if lines := decodeLines(t, buf); len(lines) != 1 { - t.Errorf("attendu 1 ligne, obtenu %d", len(lines)) - } -} - -func TestNewStderr_LoggerNil(t *testing.T) { - n := NewStderr(nil) - if n.logger == nil { - t.Fatal("logger nil non remplacé par slog.Default()") - } - n.Notify("vpc", "vp-admin", "netns absent") -} - -func TestNewStderr_ImplementeNotifier(t *testing.T) { - var _ Notifier = NewStderr(nil) -} diff --git a/internal/watchdog/resources.go b/internal/watchdog/resources.go deleted file mode 100644 index 37219a5..0000000 --- a/internal/watchdog/resources.go +++ /dev/null @@ -1,72 +0,0 @@ -package watchdog - -import ( - "fmt" - "sort" - "strings" - - "git.g3e.fr/syonad/two/internal/netif" - "git.g3e.fr/syonad/two/internal/watchdog/notify" - "git.g3e.fr/syonad/two/pkg/systemd" -) - -const ( - prefixVPC = "vpc/" - prefixSubnet = "subnet/" - prefixVM = "vm/" - - kindVPC = "vpc" - kindSubnet = "subnet" - kindVM = "vm" - - stateSuffix = "/state" -) - -func resourceNames(pairs map[string]string, prefix string) []string { - names := make([]string, 0, len(pairs)) - for key := range pairs { - name, ok := strings.CutPrefix(key, prefix) - if !ok { - continue - } - name, ok = strings.CutSuffix(name, stateSuffix) - if !ok { - continue - } - if name == "" || strings.Contains(name, "/") { - continue - } - names = append(names, name) - } - sort.Strings(names) - return names -} - -type unitChecker interface { - Status(unit string) (*systemd.ServiceStatus, error) -} - -func checkUnit(kind, name, unit string, u unitChecker, n notify.Notifier) { - if u == nil { - return - } - st, err := u.Status(unit) - if err != nil { - n.Notify(kind, name, fmt.Sprintf("unit %s unreadable: %v", unit, err)) - return - } - if st.ActiveState != "active" { - n.Notify(kind, name, fmt.Sprintf("unit %s %s (%s)", unit, st.ActiveState, st.SubState)) - } -} - -func linkProblem(iface string) string { - up, err := netif.LinkIsUp(iface) - switch { - case err != nil: - return fmt.Sprintf("interface %s not found: %v", iface, err) - case !up: - return fmt.Sprintf("interface %s down", iface) - } - return "" -} diff --git a/internal/watchdog/resources_test.go b/internal/watchdog/resources_test.go deleted file mode 100644 index 6d0ed5c..0000000 --- a/internal/watchdog/resources_test.go +++ /dev/null @@ -1,81 +0,0 @@ -package watchdog - -import ( - "reflect" - "strings" - "testing" -) - -func TestResourceNames_ExtraitDepuisLesClesState(t *testing.T) { - pairs := map[string]string{ - "vpc/vp-admin/state": "running", - "vpc/vp-lab/state": "creating", - } - want := []string{"vp-admin", "vp-lab"} - if got := resourceNames(pairs, prefixVPC); !reflect.DeepEqual(got, want) { - t.Errorf("resourceNames = %v, attendu %v", got, want) - } -} - -func TestResourceNames_IgnoreLesAutresCles(t *testing.T) { - pairs := map[string]string{ - "subnet/br-000042/state": "running", - "subnet/br-000042/vpc": "vp-admin", - "subnet/br-000042/cidr": "10.0.0.0/24", - "subnet/br-000042/local_iface": "br-000042", - } - want := []string{"br-000042"} - if got := resourceNames(pairs, prefixSubnet); !reflect.DeepEqual(got, want) { - t.Errorf("resourceNames = %v, attendu %v", got, want) - } -} - -func TestResourceNames_IgnoreLesClesImbriquees(t *testing.T) { - pairs := map[string]string{ - "vm/i-test1/state": "running", - "vm/i-test1/disk/vda/state": "attached", - "vm/i-test1/disk/vdb/state": "attached", - "vm/i-test1/dhcp/10.0.0.5": "00:22:33:44:55:66", - } - want := []string{"i-test1"} - if got := resourceNames(pairs, prefixVM); !reflect.DeepEqual(got, want) { - t.Errorf("resourceNames = %v, attendu %v", got, want) - } -} - -func TestResourceNames_Trie(t *testing.T) { - pairs := map[string]string{ - "vpc/vp-c/state": "running", - "vpc/vp-a/state": "running", - "vpc/vp-b/state": "running", - } - want := []string{"vp-a", "vp-b", "vp-c"} - for i := range 20 { - if got := resourceNames(pairs, prefixVPC); !reflect.DeepEqual(got, want) { - t.Fatalf("itération %d : resourceNames = %v, attendu %v", i, got, want) - } - } -} - -func TestResourceNames_MapVide(t *testing.T) { - if got := resourceNames(map[string]string{}, prefixVPC); len(got) != 0 { - t.Errorf("resourceNames = %v, attendu vide", got) - } -} - -func TestResourceNames_PrefixeNonCorrespondant(t *testing.T) { - pairs := map[string]string{"vm/i-test1/state": "running"} - if got := resourceNames(pairs, prefixVPC); len(got) != 0 { - t.Errorf("resourceNames = %v, attendu vide", got) - } -} - -func TestLinkProblem_InterfaceInexistante(t *testing.T) { - p := linkProblem("interface-qui-nexiste-pas-42") - if p == "" { - t.Fatal("linkProblem devrait signaler un problème pour une interface inexistante") - } - if !strings.Contains(p, "interface-qui-nexiste-pas-42") { - t.Errorf("le message %q devrait nommer l'interface", p) - } -} diff --git a/internal/watchdog/watchdog.go b/internal/watchdog/watchdog.go deleted file mode 100644 index 3ab1d52..0000000 --- a/internal/watchdog/watchdog.go +++ /dev/null @@ -1,90 +0,0 @@ -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: invalid interval, default applied", - "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: starting", "interval", w.interval) - - for { - select { - case <-ctx.Done(): - w.logger.Info("watchdog: stopping") - 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: vpc check failed", "err", err) - } - if err := CheckSubnets(w.db, u, w.notifier); err != nil { - w.logger.Error("watchdog: subnet check failed", "err", err) - } - if err := CheckVMs(w.db, w.cfg, u, w.notifier); err != nil { - w.logger.Error("watchdog: vm check failed", "err", err) - } -} - -func (w *Watchdog) units() (unitChecker, func()) { - m, err := systemd.New() - if err != nil { - if !w.dbusDown { - w.logger.Warn("watchdog: systemd unreachable, unit checks disabled", "err", err) - w.dbusDown = true - } - return nil, func() {} - } - if w.dbusDown { - w.logger.Info("watchdog: systemd connection restored") - w.dbusDown = false - } - return m, m.Close -} diff --git a/internal/watchdog/watchdog_test.go b/internal/watchdog/watchdog_test.go deleted file mode 100644 index 019ddb5..0000000 --- a/internal/watchdog/watchdog_test.go +++ /dev/null @@ -1,200 +0,0 @@ -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(), "invalid interval") { - 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, "vpc check failed"): - vpc = true - case strings.Contains(msg, "subnet check failed"): - subnet = true - case strings.Contains(msg, "vm check failed"): - 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: starting") { - t.Error("le démarrage devrait être logué") - } - if !strings.Contains(out, "watchdog: stopping") { - 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, "systemd unreachable") { - warnings++ - } - } - if warnings != 1 { - t.Errorf("attendu 1 avertissement pour 3 tentatives, obtenu %d", warnings) - } -} diff --git a/pkg/db/kv/admin_server.go b/pkg/db/kv/admin_server.go index db37f6c..1216244 100644 --- a/pkg/db/kv/admin_server.go +++ b/pkg/db/kv/admin_server.go @@ -1,8 +1,6 @@ package kv import ( - "context" - "errors" "fmt" "log/slog" "net/http" @@ -14,28 +12,21 @@ import ( type AdminServer struct { db *badger.DB logger *slog.Logger - srv *http.Server } -func NewAdminServer(db *badger.DB, logger *slog.Logger, address string) *AdminServer { - s := &AdminServer{db: db, logger: logger} +func NewAdminServer(db *badger.DB, logger *slog.Logger) *AdminServer { + return &AdminServer{db: db, logger: logger} +} + +func (s *AdminServer) Start(address string) { mux := http.NewServeMux() mux.HandleFunc("/db", s.dbHandler) - s.srv = &http.Server{Addr: address, Handler: mux} - return s -} - -func (s *AdminServer) Start() { - s.logger.Info("admin server listening", "address", s.srv.Addr) - if err := s.srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + s.logger.Info("admin server listening", "address", address) + if err := http.ListenAndServe(address, mux); err != nil { s.logger.Error("admin server stopped", "error", err) } } -func (s *AdminServer) Shutdown(ctx context.Context) error { - return s.srv.Shutdown(ctx) -} - func (s *AdminServer) dbHandler(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) diff --git a/pkg/prometheus/server.go b/pkg/prometheus/server.go index f6c34b3..9f8e557 100644 --- a/pkg/prometheus/server.go +++ b/pkg/prometheus/server.go @@ -1,8 +1,6 @@ package promserver import ( - "context" - "errors" "log" "net/http" @@ -10,30 +8,13 @@ import ( "github.com/prometheus/client_golang/prometheus/promhttp" ) -// Server exposes the Prometheus metrics endpoint. -type Server struct { - srv *http.Server -} - -// New builds the metrics server for the given address and registry. -func New(address string, registry *prometheus.Registry) *Server { +// Start launches the Prometheus metrics HTTP server on the given address. +// The provided registry is used to expose metrics at /metrics. +func Start(address string, registry *prometheus.Registry) { mux := http.NewServeMux() mux.Handle("/metrics", promhttp.HandlerFor(registry, promhttp.HandlerOpts{ EnableOpenMetrics: true, })) - return &Server{srv: &http.Server{Addr: address, Handler: mux}} -} - -// Start blocks until the server stops. A failure is logged, never fatal: an -// unavailable metrics endpoint must not bring the whole agent down. -func (s *Server) Start() { - log.Printf("Prometheus server listening on %s", s.srv.Addr) - if err := s.srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { - log.Printf("Prometheus server stopped: %v", err) - } -} - -// Shutdown stops the server, waiting for in-flight requests until ctx expires. -func (s *Server) Shutdown(ctx context.Context) error { - return s.srv.Shutdown(ctx) + log.Printf("Prometheus server listening on %s", address) + log.Fatal(http.ListenAndServe(address, mux)) } diff --git a/pkg/systemd/main.go b/pkg/systemd/main.go index 768294f..e89af6e 100644 --- a/pkg/systemd/main.go +++ b/pkg/systemd/main.go @@ -99,17 +99,13 @@ func (m *Manager) Status(service string) (*ServiceStatus, error) { return nil, err } - status := &ServiceStatus{Name: service} + status := &ServiceStatus{ + Name: service, + LoadState: props["LoadState"].(string), + ActiveState: props["ActiveState"].(string), + SubState: props["SubState"].(string), + } - if v, ok := props["LoadState"].(string); ok { - status.LoadState = v - } - if v, ok := props["ActiveState"].(string); ok { - status.ActiveState = v - } - if v, ok := props["SubState"].(string); ok { - status.SubState = v - } if pid, ok := props["MainPID"].(uint32); ok { status.MainPID = pid } diff --git a/pkg/worker/queue.go b/pkg/worker/queue.go index 40e1443..109c726 100644 --- a/pkg/worker/queue.go +++ b/pkg/worker/queue.go @@ -1,19 +1,13 @@ package worker -import ( - "log" - "sync" -) +import "log" // Task is a function to be executed asynchronously by a worker. type Task func() // Queue is a FIFO channel-backed task queue consumed by worker goroutines. type Queue struct { - tasks chan Task - wg sync.WaitGroup - mu sync.RWMutex - stopped bool + tasks chan Task } // New creates a Queue with the given channel buffer size. @@ -21,16 +15,8 @@ func New(bufferSize int) *Queue { return &Queue{tasks: make(chan Task, bufferSize)} } -// Submit enqueues a task. Blocks if the queue is full. Tasks submitted after -// Stop are rejected and logged rather than enqueued. +// Submit enqueues a task. Blocks if the queue is full. func (q *Queue) Submit(t Task) { - q.mu.RLock() - defer q.mu.RUnlock() - - if q.stopped { - log.Print("worker: queue stopped, task rejected") - return - } q.tasks <- t } @@ -38,28 +24,10 @@ func (q *Queue) Submit(t Task) { func (q *Queue) Start(n int) { log.Printf("worker: starting %d workers", n) for i := range n { - q.wg.Add(1) go func(id int) { - defer q.wg.Done() for task := range q.tasks { task() } }(i) } } - -// Stop rejects new tasks, then waits for the queued and in-flight ones to -// finish. The write lock is what makes closing the channel safe: it is only -// taken once every in-flight Submit has released its read lock. -func (q *Queue) Stop() { - q.mu.Lock() - if q.stopped { - q.mu.Unlock() - return - } - q.stopped = true - close(q.tasks) - q.mu.Unlock() - - q.wg.Wait() -} diff --git a/pkg/worker/queue_test.go b/pkg/worker/queue_test.go index 4e463d6..5353b06 100644 --- a/pkg/worker/queue_test.go +++ b/pkg/worker/queue_test.go @@ -101,79 +101,3 @@ func TestQueue_SubmitBlocksWhenFull(t *testing.T) { t.Error("Submit aurait dû se débloquer après démarrage d'un worker") } } - -func TestStop_AttendLesTachesEnCours(t *testing.T) { - q := New(10) - q.Start(2) - - var mu sync.Mutex - done := 0 - for range 5 { - q.Submit(func() { - time.Sleep(20 * time.Millisecond) - mu.Lock() - done++ - mu.Unlock() - }) - } - - q.Stop() - - mu.Lock() - defer mu.Unlock() - if done != 5 { - t.Errorf("Stop devrait attendre les 5 tâches, %d terminées", done) - } -} - -func TestStop_RejetteLesTachesSuivantes(t *testing.T) { - q := New(10) - q.Start(1) - q.Stop() - - executed := make(chan struct{}, 1) - q.Submit(func() { executed <- struct{}{} }) - - select { - case <-executed: - t.Error("une tâche soumise après Stop ne doit pas être exécutée") - case <-time.After(100 * time.Millisecond): - } -} - -func TestStop_Idempotent(t *testing.T) { - q := New(10) - q.Start(1) - q.Stop() - q.Stop() -} - -func TestStop_SansTacheEnCours(t *testing.T) { - q := New(10) - q.Start(3) - - done := make(chan struct{}) - go func() { q.Stop(); close(done) }() - - select { - case <-done: - case <-time.After(2 * time.Second): - t.Fatal("Stop n'a pas rendu la main sur une file vide") - } -} - -func TestSubmit_ConcurrentAvecStop(t *testing.T) { - q := New(100) - q.Start(4) - - var wg sync.WaitGroup - for range 20 { - wg.Add(1) - go func() { - defer wg.Done() - q.Submit(func() {}) - }() - } - q.Stop() - wg.Wait() -} diff --git a/systemd/agent.service b/systemd/agent.service index 95ccf7e..37715c4 100644 --- a/systemd/agent.service +++ b/systemd/agent.service @@ -5,7 +5,6 @@ After=network.target [Service] Type=simple ExecStart=/opt/two/bin/agent -TimeoutStopSec=30 [Install] WantedBy=multi-user.target