Merge branch 'feature-29'
This commit is contained in:
commit
57448d758f
GPG key ID:
4406BBBF8845D632
31 changed files with 2168 additions and 36 deletions
|
|
@ -1,15 +1,22 @@
|
|||
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"
|
||||
|
|
@ -17,6 +24,8 @@ 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()
|
||||
|
|
@ -30,7 +39,12 @@ func main() {
|
|||
log := logger.New(cfg.Logger.Level, cfg.Logger.Debug)
|
||||
|
||||
db := kv.InitDB(kv.Config{Path: cfg.Database.Path}, false)
|
||||
defer db.Close()
|
||||
closeDB := true
|
||||
defer func() {
|
||||
if closeDB {
|
||||
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.
|
||||
|
|
@ -56,13 +70,69 @@ 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")))
|
||||
go agentapi.New(d, db, log.With(slog.String("component", "api"))).Start(apiAddr)
|
||||
go promserver.Start(promAddr, registry)
|
||||
|
||||
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
|
||||
if cfg.Admin.Enabled {
|
||||
adminAddr := fmt.Sprintf("%s:%d", cfg.Admin.Address, cfg.Admin.Port)
|
||||
go kv.NewAdminServer(db, log.With(slog.String("component", "admin"))).Start(adminAddr)
|
||||
adminSrv = kv.NewAdminServer(db, log.With(slog.String("component", "admin")), adminAddr)
|
||||
go adminSrv.Start()
|
||||
}
|
||||
|
||||
select {}
|
||||
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
|
||||
}
|
||||
}
|
||||
|
|
|
|||
98
cmd/agent/main_test.go
Normal file
98
cmd/agent/main_test.go
Normal file
|
|
@ -0,0 +1,98 @@
|
|||
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")
|
||||
}
|
||||
}
|
||||
|
|
@ -51,6 +51,13 @@ 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
|
||||
|
|
|
|||
|
|
@ -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), db
|
||||
return New(d, db, logger, "127.0.0.1:0"), db
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,8 +1,10 @@
|
|||
package agentapi
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"time"
|
||||
|
|
@ -15,13 +17,11 @@ 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) *Server {
|
||||
return &Server{dispatcher: d, db: db, logger: logger}
|
||||
}
|
||||
|
||||
func (s *Server) Start(address string) {
|
||||
func New(d *dispatcher.Dispatcher, db *badger.DB, logger *slog.Logger, address string) *Server {
|
||||
s := &Server{dispatcher: d, db: db, logger: logger}
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("/vpcs", s.VpcsHandler)
|
||||
mux.HandleFunc("/vpcs/", s.VpcByNameHandler)
|
||||
|
|
@ -29,12 +29,21 @@ func (s *Server) Start(address string) {
|
|||
mux.HandleFunc("/subnets/", s.SubnetByNameHandler)
|
||||
mux.HandleFunc("/vms", s.VmsHandler)
|
||||
mux.HandleFunc("/vms/", s.VmByNameHandler)
|
||||
s.logger.Info("API server listening", "address", address)
|
||||
if err := http.ListenAndServe(address, s.logMiddleware(mux)); err != nil {
|
||||
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.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
|
||||
|
|
|
|||
|
|
@ -81,3 +81,68 @@ 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")
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -44,6 +44,10 @@ 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"`
|
||||
}
|
||||
|
|
@ -69,6 +73,8 @@ 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)
|
||||
|
|
|
|||
|
|
@ -4,6 +4,8 @@ import (
|
|||
"net"
|
||||
)
|
||||
|
||||
const DefaultConfDir = "/etc/dnsmasq.d"
|
||||
|
||||
type Config struct {
|
||||
Network *net.IPNet
|
||||
VPCGateway net.IP // next-hop for VPCRoute (option 121)
|
||||
|
|
|
|||
15
internal/netif/link.go
Normal file
15
internal/netif/link.go
Normal file
|
|
@ -0,0 +1,15 @@
|
|||
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
|
||||
}
|
||||
|
|
@ -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: "/etc/dnsmasq.d",
|
||||
ConfDir: dhcp.DefaultConfDir,
|
||||
}
|
||||
switch d.mode {
|
||||
case "vxlan":
|
||||
|
|
|
|||
|
|
@ -3,7 +3,9 @@ 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"
|
||||
|
|
@ -62,7 +64,7 @@ func stopDHCP(db *badger.DB, subnetName string, d subnetData) error {
|
|||
}
|
||||
}
|
||||
|
||||
if err := os.Remove("/etc/dnsmasq.d/" + d.vpc + "_" + d.bridge + ".conf"); err != nil && !os.IsNotExist(err) {
|
||||
if err := os.Remove(filepath.Join(dhcp.DefaultConfDir, d.vpc+"_"+d.bridge+".conf")); err != nil && !os.IsNotExist(err) {
|
||||
return fmt.Errorf("remove dnsmasq config: %w", err)
|
||||
}
|
||||
|
||||
|
|
|
|||
134
internal/watchdog/check_subnet.go
Normal file
134
internal/watchdog/check_subnet.go
Normal file
|
|
@ -0,0 +1,134 @@
|
|||
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))
|
||||
}
|
||||
}
|
||||
288
internal/watchdog/check_subnet_test.go
Normal file
288
internal/watchdog/check_subnet_test.go
Normal file
|
|
@ -0,0 +1,288 @@
|
|||
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
|
||||
}
|
||||
99
internal/watchdog/check_vm.go
Normal file
99
internal/watchdog/check_vm.go
Normal file
|
|
@ -0,0 +1,99 @@
|
|||
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))
|
||||
}
|
||||
}
|
||||
246
internal/watchdog/check_vm_test.go
Normal file
246
internal/watchdog/check_vm_test.go
Normal file
|
|
@ -0,0 +1,246 @@
|
|||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
71
internal/watchdog/check_vpc.go
Normal file
71
internal/watchdog/check_vpc.go
Normal file
|
|
@ -0,0 +1,71 @@
|
|||
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))
|
||||
}
|
||||
}
|
||||
177
internal/watchdog/check_vpc_test.go
Normal file
177
internal/watchdog/check_vpc_test.go
Normal file
|
|
@ -0,0 +1,177 @@
|
|||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
110
internal/watchdog/helpers_test.go
Normal file
110
internal/watchdog/helpers_test.go
Normal file
|
|
@ -0,0 +1,110 @@
|
|||
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)
|
||||
}
|
||||
}
|
||||
5
internal/watchdog/notify/notify.go
Normal file
5
internal/watchdog/notify/notify.go
Normal file
|
|
@ -0,0 +1,5 @@
|
|||
package notify
|
||||
|
||||
type Notifier interface {
|
||||
Notify(kind, name, problem string)
|
||||
}
|
||||
24
internal/watchdog/notify/stderr.go
Normal file
24
internal/watchdog/notify/stderr.go
Normal file
|
|
@ -0,0 +1,24 @@
|
|||
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,
|
||||
)
|
||||
}
|
||||
120
internal/watchdog/notify/stderr_test.go
Normal file
120
internal/watchdog/notify/stderr_test.go
Normal file
|
|
@ -0,0 +1,120 @@
|
|||
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)
|
||||
}
|
||||
72
internal/watchdog/resources.go
Normal file
72
internal/watchdog/resources.go
Normal file
|
|
@ -0,0 +1,72 @@
|
|||
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 ""
|
||||
}
|
||||
81
internal/watchdog/resources_test.go
Normal file
81
internal/watchdog/resources_test.go
Normal file
|
|
@ -0,0 +1,81 @@
|
|||
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)
|
||||
}
|
||||
}
|
||||
90
internal/watchdog/watchdog.go
Normal file
90
internal/watchdog/watchdog.go
Normal file
|
|
@ -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: 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
|
||||
}
|
||||
200
internal/watchdog/watchdog_test.go
Normal file
200
internal/watchdog/watchdog_test.go
Normal file
|
|
@ -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(), "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)
|
||||
}
|
||||
}
|
||||
|
|
@ -1,6 +1,8 @@
|
|||
package kv
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
|
|
@ -12,21 +14,28 @@ import (
|
|||
type AdminServer struct {
|
||||
db *badger.DB
|
||||
logger *slog.Logger
|
||||
srv *http.Server
|
||||
}
|
||||
|
||||
func NewAdminServer(db *badger.DB, logger *slog.Logger) *AdminServer {
|
||||
return &AdminServer{db: db, logger: logger}
|
||||
}
|
||||
|
||||
func (s *AdminServer) Start(address string) {
|
||||
func NewAdminServer(db *badger.DB, logger *slog.Logger, address string) *AdminServer {
|
||||
s := &AdminServer{db: db, logger: logger}
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("/db", s.dbHandler)
|
||||
s.logger.Info("admin server listening", "address", address)
|
||||
if err := http.ListenAndServe(address, mux); err != nil {
|
||||
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.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)
|
||||
|
|
|
|||
|
|
@ -1,6 +1,8 @@
|
|||
package promserver
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"log"
|
||||
"net/http"
|
||||
|
||||
|
|
@ -8,13 +10,30 @@ import (
|
|||
"github.com/prometheus/client_golang/prometheus/promhttp"
|
||||
)
|
||||
|
||||
// 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) {
|
||||
// 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 {
|
||||
mux := http.NewServeMux()
|
||||
mux.Handle("/metrics", promhttp.HandlerFor(registry, promhttp.HandlerOpts{
|
||||
EnableOpenMetrics: true,
|
||||
}))
|
||||
log.Printf("Prometheus server listening on %s", address)
|
||||
log.Fatal(http.ListenAndServe(address, mux))
|
||||
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)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -99,13 +99,17 @@ func (m *Manager) Status(service string) (*ServiceStatus, error) {
|
|||
return nil, err
|
||||
}
|
||||
|
||||
status := &ServiceStatus{
|
||||
Name: service,
|
||||
LoadState: props["LoadState"].(string),
|
||||
ActiveState: props["ActiveState"].(string),
|
||||
SubState: props["SubState"].(string),
|
||||
}
|
||||
status := &ServiceStatus{Name: service}
|
||||
|
||||
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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,9 @@
|
|||
package worker
|
||||
|
||||
import "log"
|
||||
import (
|
||||
"log"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// Task is a function to be executed asynchronously by a worker.
|
||||
type Task func()
|
||||
|
|
@ -8,6 +11,9 @@ 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
|
||||
}
|
||||
|
||||
// New creates a Queue with the given channel buffer size.
|
||||
|
|
@ -15,8 +21,16 @@ func New(bufferSize int) *Queue {
|
|||
return &Queue{tasks: make(chan Task, bufferSize)}
|
||||
}
|
||||
|
||||
// Submit enqueues a task. Blocks if the queue is full.
|
||||
// Submit enqueues a task. Blocks if the queue is full. Tasks submitted after
|
||||
// Stop are rejected and logged rather than enqueued.
|
||||
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
|
||||
}
|
||||
|
||||
|
|
@ -24,10 +38,28 @@ 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()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -101,3 +101,79 @@ 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()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ After=network.target
|
|||
[Service]
|
||||
Type=simple
|
||||
ExecStart=/opt/two/bin/agent
|
||||
TimeoutStopSec=30
|
||||
|
||||
[Install]
|
||||
WantedBy=multi-user.target
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue