diff --git a/src/yangerd/.gitignore b/src/yangerd/.gitignore index 00713b02..d5b437a9 100644 --- a/src/yangerd/.gitignore +++ b/src/yangerd/.gitignore @@ -1,3 +1,3 @@ -# Build artifacts -yangerd -yangerctl +# Build artifacts (root-level binaries only; do not match cmd/ source dirs) +/yangerd +/yangerctl diff --git a/src/yangerd/cmd/yangerctl/main.go b/src/yangerd/cmd/yangerctl/main.go new file mode 100644 index 00000000..a4ef55d3 --- /dev/null +++ b/src/yangerd/cmd/yangerctl/main.go @@ -0,0 +1,110 @@ +package main + +import ( + "encoding/json" + "fmt" + "os" + "time" + + "github.com/kernelkit/infix/src/yangerd/internal/ipc" +) + +const defaultSocket = "/run/yangerd.sock" +const defaultTimeout = 5 * time.Second + +func main() { + socket := defaultSocket + timeout := defaultTimeout + + args := os.Args[1:] + for len(args) > 0 && len(args[0]) > 0 && args[0][0] == '-' { + switch args[0] { + case "--socket": + if len(args) < 2 { + die("--socket requires an argument") + } + socket = args[1] + args = args[2:] + case "--timeout": + if len(args) < 2 { + die("--timeout requires an argument") + } + d, err := time.ParseDuration(args[1]) + if err != nil { + die("invalid duration: %v", err) + } + timeout = d + args = args[2:] + default: + die("unknown flag: %s", args[0]) + } + } + + if len(args) == 0 { + usage() + } + + client := ipc.NewClient(socket, timeout) + + switch args[0] { + case "get": + if len(args) < 2 { + die("get requires a path argument") + } + resp, err := client.Get(args[1]) + if err != nil { + die("get: %v", err) + } + printResponse(resp) + case "dump": + resp, err := client.Get("/") + if err != nil { + die("dump: %v", err) + } + printResponse(resp) + case "health": + resp, err := client.Health() + if err != nil { + die("health: %v", err) + } + printResponse(resp) + default: + die("unknown command: %s", args[0]) + } +} + +func printResponse(resp *ipc.Response) { + if resp.Code == 503 { + fmt.Fprintf(os.Stderr, "yangerd is starting up\n") + os.Exit(3) + } + if resp.Status == "error" { + fmt.Fprintf(os.Stderr, "error %d: %s\n", resp.Code, resp.Message) + if resp.Code == 404 { + os.Exit(2) + } + os.Exit(1) + } + + var out []byte + if resp.Data != nil { + out, _ = json.MarshalIndent(json.RawMessage(resp.Data), "", " ") + } else { + out, _ = json.MarshalIndent(resp, "", " ") + } + fmt.Println(string(out)) +} + +func usage() { + fmt.Fprintf(os.Stderr, "Usage: yangerctl [--socket path] [--timeout dur] [args]\n\n") + fmt.Fprintf(os.Stderr, "Commands:\n") + fmt.Fprintf(os.Stderr, " get Query a YANG subtree\n") + fmt.Fprintf(os.Stderr, " dump Dump entire tree\n") + fmt.Fprintf(os.Stderr, " health Show daemon health\n") + os.Exit(1) +} + +func die(format string, args ...interface{}) { + fmt.Fprintf(os.Stderr, "yangerctl: "+format+"\n", args...) + os.Exit(1) +} diff --git a/src/yangerd/cmd/yangerd/main.go b/src/yangerd/cmd/yangerd/main.go new file mode 100644 index 00000000..0b5f25b1 --- /dev/null +++ b/src/yangerd/cmd/yangerd/main.go @@ -0,0 +1,427 @@ +package main + +import ( + "context" + "encoding/json" + "log" + "log/slog" + "os" + "os/signal" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "syscall" + "time" + + "github.com/kernelkit/infix/src/yangerd/internal/bridgebatch" + "github.com/kernelkit/infix/src/yangerd/internal/collector" + "github.com/kernelkit/infix/src/yangerd/internal/config" + "github.com/kernelkit/infix/src/yangerd/internal/dbusmonitor" + "github.com/kernelkit/infix/src/yangerd/internal/ethmonitor" + "github.com/kernelkit/infix/src/yangerd/internal/frrvty" + "github.com/kernelkit/infix/src/yangerd/internal/fswatcher" + "github.com/kernelkit/infix/src/yangerd/internal/ipbatch" + "github.com/kernelkit/infix/src/yangerd/internal/ipc" + "github.com/kernelkit/infix/src/yangerd/internal/iwmonitor" + "github.com/kernelkit/infix/src/yangerd/internal/lldpmonitor" + "github.com/kernelkit/infix/src/yangerd/internal/monitor" + "github.com/kernelkit/infix/src/yangerd/internal/sysreaders" + "github.com/kernelkit/infix/src/yangerd/internal/tree" + "github.com/kernelkit/infix/src/yangerd/internal/wgquery" + "github.com/kernelkit/infix/src/yangerd/internal/zapiwatcher" +) + +// osFileChecker implements iface.FileChecker using the real filesystem. +type osFileChecker struct{} + +func (osFileChecker) Exists(path string) bool { + _, err := os.Stat(path) + return err == nil +} + +func (osFileChecker) ReadFile(path string) (string, error) { + b, err := os.ReadFile(path) + if err != nil { + return "", err + } + return string(b), nil +} + +func main() { + cfg := config.Load() + log.SetFlags(0) + + t := tree.New() + ready := &atomic.Bool{} + + srv := ipc.NewServer(t, ready) + if err := srv.Listen(cfg.Socket); err != nil { + log.Fatalf("listen %s: %v", cfg.Socket, err) + } + defer os.Remove(cfg.Socket) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + var wg sync.WaitGroup + cmd := collector.ExecRunner{} + fs := collector.OSFileReader{} + collectors := []collector.Collector{ + collector.NewSystemCollector(cmd, fs, cfg.PollSystem), + collector.NewRoutingCollector(cmd, cfg.PollRouting), + collector.NewNTPCollector(cmd, cfg.PollNTP), + collector.NewHardwareCollector(cmd, fs, cfg.PollHardware, cfg.EnableWifi, cfg.EnableGPS), + } + pokeCh := make(chan struct{}, len(collectors)) + collector.RunAll(ctx, &wg, t, collectors, pokeCh) + + inst := collector.DBusInstaller{} + t.RegisterProvider("ietf-system:system-state", func() json.RawMessage { + live := collector.LiveSystemState(fs) + installerOverlay := collector.MergeInstaller(t.GetCached("ietf-system:system-state"), inst) + if installerOverlay == nil { + return live + } + var base map[string]json.RawMessage + if json.Unmarshal(live, &base) != nil { + return live + } + var overlay map[string]json.RawMessage + if json.Unmarshal(installerOverlay, &overlay) != nil { + return live + } + for k, v := range overlay { + base[k] = v + } + merged, err := json.Marshal(base) + if err != nil { + return live + } + return merged + }) + + if data := collector.BootPlatform(fs); data != nil { + t.Merge("ietf-system:system-state", data) + } + if data := collector.BootSoftware(ctx, cmd); data != nil { + t.Merge("ietf-system:system-state", data) + } + + slogLog := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slogLevel(cfg.LogLevel)})) + + linkBatch, err := ipbatch.New(ctx, slogLog, ipbatch.WithStats(), ipbatch.WithDetails()) + if err != nil { + log.Fatalf("start link batch: %v", err) + } + defer linkBatch.Close() + + addrBatch, err := ipbatch.New(ctx, slogLog, ipbatch.WithDetails()) + if err != nil { + log.Fatalf("start addr batch: %v", err) + } + defer addrBatch.Close() + + neighBatch, err := ipbatch.New(ctx, slogLog) + if err != nil { + log.Fatalf("start neigh batch: %v", err) + } + defer neighBatch.Close() + + brBatch, err := bridgebatch.New(ctx, slogLog) + if err != nil { + log.Fatalf("start bridge batch: %v", err) + } + defer brBatch.Close() + + nlmon := monitor.New(linkBatch, addrBatch, neighBatch, brBatch, t, osFileChecker{}, slogLog) + + ethMon, err := ethmonitor.New(slogLog, cmd) + if err != nil { + slogLog.Warn("ethmonitor unavailable, continuing without it", "err", err) + } else { + ethMon.SetOnUpdate(nlmon.SetEthernetData) + nlmon.SetEthRefresh(ethMon.RefreshInterface) + wg.Add(1) + go func() { + defer wg.Done() + if err := ethMon.Run(ctx); err != nil && ctx.Err() == nil { + slogLog.Error("ethmonitor exited", "err", err) + } + }() + } + + wg.Add(1) + go func() { + defer wg.Done() + ticker := time.NewTicker(10 * time.Second) + defer ticker.Stop() + <-nlmon.WaitReady() + for { + links := nlmon.Links() + for ifname, data := range wgquery.Query(links) { + nlmon.SetWireguardData(ifname, data) + } + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } + }() + + wg.Add(1) + go func() { + defer wg.Done() + ticker := time.NewTicker(cfg.PollSTP) + defer ticker.Stop() + <-nlmon.WaitReady() + for { + nlmon.RefreshSTP() + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } + }() + + wg.Add(1) + go func() { + defer wg.Done() + for { + if err := nlmon.Run(ctx); err != nil { + if ctx.Err() != nil { + return + } + slogLog.Error("nlmonitor exited, restarting in 1s", "err", err) + select { + case <-ctx.Done(): + return + case <-time.After(time.Second): + } + } else { + return + } + } + }() + + if cfg.EnableWifi { + iwmon := iwmonitor.New(slogLog) + iwmon.SetOnUpdate(nlmon.SetWifiData) + iwmon.SetOnPhyChange(func() { + select { + case pokeCh <- struct{}{}: + default: + } + }) + wg.Add(1) + go func() { + defer wg.Done() + if err := iwmon.Run(ctx); err != nil && ctx.Err() == nil { + slogLog.Error("iwmonitor exited", "err", err) + } + }() + } + + if cfg.EnableLLDP { + lldpmon := lldpmonitor.New(t, slogLog) + wg.Add(1) + go func() { + defer wg.Done() + if err := lldpmon.Run(ctx); err != nil && ctx.Err() == nil { + slogLog.Error("lldpmonitor exited", "err", err) + } + }() + } + + zapi := zapiwatcher.New(t, frrvty.New(""), slogLog) + wg.Add(1) + go func() { + defer wg.Done() + if err := zapi.Run(ctx); err != nil && ctx.Err() == nil { + slogLog.Error("zapiwatcher exited", "err", err) + } + }() + + if cfg.EnableDHCP || cfg.EnableFirewall { + dbusMon := dbusmonitor.New(t, slogLog) + wg.Add(1) + go func() { + defer wg.Done() + if err := dbusMon.Run(ctx); err != nil && ctx.Err() == nil { + slogLog.Error("dbusmonitor exited", "err", err) + } + }() + } + + fsw, err := fswatcher.New(t, slogLog) + if err != nil { + log.Fatalf("start fswatcher: %v", err) + } + + fwdAgg := sysreaders.NewForwardingAggregator() + forwardingPaths := []string{ + "/proc/sys/net/ipv4/conf/*/forwarding", + "/proc/sys/net/ipv6/conf/*/forwarding", + } + for _, pattern := range forwardingPaths { + matches, globErr := filepath.Glob(pattern) + if globErr != nil { + slogLog.Warn("fswatcher glob failed", "pattern", pattern, "err", globErr) + continue + } + for _, path := range matches { + if err := fsw.Watch(path, fswatcher.WatchHandler{ + TreeKey: routingTreeKey, + ReadFunc: fwdAgg.HandleForwardingChange, + Debounce: 100 * time.Millisecond, + UseMerge: true, + }); err != nil { + slogLog.Warn("fswatcher watch failed", "path", path, "err", err) + } + } + } + if err := fsw.Watch("/etc/hostname", fswatcher.WatchHandler{ + TreeKey: "ietf-system:system", + ReadFunc: sysreaders.ReadHostname, + Debounce: 200 * time.Millisecond, + UseMerge: true, + }); err != nil { + slogLog.Warn("fswatcher watch failed", "path", "/etc/hostname", "err", err) + } + if err := fsw.WatchSymlink("/etc/localtime", fswatcher.WatchHandler{ + TreeKey: "ietf-system:system", + ReadFunc: sysreaders.ReadTimezone, + Debounce: 200 * time.Millisecond, + UseMerge: true, + }); err != nil { + slogLog.Warn("fswatcher watch failed", "path", "/etc/localtime", "err", err) + } + usersHandler := fswatcher.WatchHandler{ + TreeKey: "ietf-system:system", + ReadFunc: sysreaders.ReadUsers, + Debounce: 200 * time.Millisecond, + UseMerge: true, + } + if err := fsw.Watch("/etc/shadow", usersHandler); err != nil { + slogLog.Warn("fswatcher watch failed", "path", "/etc/shadow", "err", err) + } + if err := fsw.WatchDir(sysreaders.SSHDKeysDir, usersHandler); err != nil { + slogLog.Warn("fswatcher watch failed", "path", sysreaders.SSHDKeysDir, "err", err) + } + bootOrderHandler := fswatcher.WatchHandler{ + TreeKey: "ietf-system:system-state", + ReadFunc: makeBootOrderReader(t, cmd), + Debounce: 200 * time.Millisecond, + UseMerge: true, + } + for _, path := range []string{"/mnt/aux/grub/grubenv", "/mnt/aux/uboot.env"} { + if err := fsw.Watch(path, bootOrderHandler); err != nil { + slogLog.Debug("fswatcher boot-order watch skipped", "path", path, "err", err) + } + } + dnsHandler := fswatcher.WatchHandler{ + TreeKey: "ietf-system:system-state", + ReadFunc: sysreaders.ReadDNSResolver, + Debounce: 200 * time.Millisecond, + UseMerge: true, + } + for _, path := range []string{"/etc/resolv.conf.head", "/var/lib/misc/resolv.conf"} { + if err := fsw.WatchSymlink(path, dnsHandler); err != nil { + slogLog.Warn("fswatcher dns watch failed", "path", path, "err", err) + } + } + if cfg.EnableContainers { + containerHandler := fswatcher.WatchHandler{ + TreeKey: "infix-containers:containers", + ReadFunc: func(_ string) (json.RawMessage, error) { + return collector.CollectContainers(cmd, fs), nil + }, + Debounce: 500 * time.Millisecond, + } + os.MkdirAll("/run/libpod/events", 0755) + if err := fsw.WatchDir("/run/libpod/events", containerHandler); err != nil { + slogLog.Warn("fswatcher container watch failed", "err", err) + } + } + fsw.InitialRead() + wg.Add(1) + go func() { + defer wg.Done() + if err := fsw.Run(ctx); err != nil && ctx.Err() == nil { + slogLog.Error("fswatcher exited", "err", err) + } + }() + + go func() { + <-nlmon.WaitReady() + ready.Store(true) + }() + + sigCh := make(chan os.Signal, 1) + signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT, syscall.SIGHUP) + + go func() { + for sig := range sigCh { + if sig == syscall.SIGHUP { + log.Printf("SIGHUP: triggering immediate re-poll") + for range len(collectors) { + pokeCh <- struct{}{} + } + continue + } + cancel() + return + } + }() + + if err := srv.Serve(ctx); err != nil { + log.Fatalf("serve: %v", err) + } + + wg.Wait() +} + +func slogLevel(s string) slog.Level { + switch strings.ToLower(s) { + case "debug": + return slog.LevelDebug + case "warn", "warning": + return slog.LevelWarn + case "error": + return slog.LevelError + default: + return slog.LevelInfo + } +} + +const routingTreeKey = "ietf-routing:routing" + +func makeBootOrderReader(t *tree.Tree, cmd collector.CommandRunner) func(string) (json.RawMessage, error) { + return func(_ string) (json.RawMessage, error) { + bootOrder := collector.ReadBootOrder(context.TODO(), cmd) + + raw := t.Get("ietf-system:system-state") + var state map[string]interface{} + if raw != nil { + json.Unmarshal(raw, &state) + } + if state == nil { + state = make(map[string]interface{}) + } + + sw, _ := state["infix-system:software"].(map[string]interface{}) + if sw == nil { + sw = make(map[string]interface{}) + } + + if bootOrder != nil { + sw["boot-order"] = bootOrder + } else { + delete(sw, "boot-order") + } + + return json.Marshal(map[string]interface{}{"infix-system:software": sw}) + } +} diff --git a/src/yangerd/internal/config/config.go b/src/yangerd/internal/config/config.go index 2f7b79f4..3d0ea7c0 100644 --- a/src/yangerd/internal/config/config.go +++ b/src/yangerd/internal/config/config.go @@ -15,6 +15,7 @@ type Config struct { PollRouting time.Duration PollNTP time.Duration PollHardware time.Duration + PollSTP time.Duration EnableWifi bool EnableLLDP bool EnableFirewall bool @@ -32,6 +33,7 @@ func Load() *Config { PollRouting: envDur("YANGERD_POLL_INTERVAL_ROUTING", 10*time.Second), PollNTP: envDur("YANGERD_POLL_INTERVAL_NTP", 60*time.Second), PollHardware: envDur("YANGERD_POLL_INTERVAL_HARDWARE", 10*time.Second), + PollSTP: envDur("YANGERD_POLL_INTERVAL_STP", 5*time.Second), EnableWifi: envBool("YANGERD_ENABLE_WIFI", false), EnableLLDP: envBool("YANGERD_ENABLE_LLDP", true), EnableFirewall: envBool("YANGERD_ENABLE_FIREWALL", true), diff --git a/src/yangerd/internal/ethmonitor/ethmonitor.go b/src/yangerd/internal/ethmonitor/ethmonitor.go index eb176d55..3a0c6813 100644 --- a/src/yangerd/internal/ethmonitor/ethmonitor.go +++ b/src/yangerd/internal/ethmonitor/ethmonitor.go @@ -1,5 +1,9 @@ // Package ethmonitor subscribes to ethtool genetlink notifications and // keeps per-interface ethernet settings updated via a callback. +// +// Data is fetched by shelling out to `ethtool --json ` (matching +// the Python yanger approach) while genetlink provides reactive change +// notifications. package ethmonitor import ( @@ -8,40 +12,44 @@ import ( "fmt" "log/slog" "net" + "strings" - "github.com/mdlayher/ethtool" "github.com/mdlayher/genetlink" "github.com/mdlayher/netlink" ) const ( - // ETHTOOL_MSG_LINKINFO_NTF is the ethtool generic netlink notification - // command for link info changes. - ETHTOOL_MSG_LINKINFO_NTF = 28 - // ETHTOOL_MSG_LINKMODES_NTF is the ethtool generic netlink notification - // command for link mode changes. + ETHTOOL_MSG_LINKINFO_NTF = 28 ETHTOOL_MSG_LINKMODES_NTF = 29 ethtoolFamilyName = "ethtool" ethtoolMonitorGroupName = "monitor" nlaHeaderIfindex = 1 + + ethtoolSpeedUnknown = (1 << 32) - 1 ) +// CommandRunner executes external commands and returns stdout. +type CommandRunner interface { + Run(ctx context.Context, name string, args ...string) ([]byte, error) +} + // EthMonitor listens for ethtool genetlink monitor events and updates // interface ethernet operational state via a callback. type EthMonitor struct { - conn *genetlink.Conn - family genetlink.Family - groupID uint32 - etClient *ethtool.Client - log *slog.Logger + conn *genetlink.Conn + family genetlink.Family + groupID uint32 + cmd CommandRunner + ctx context.Context + log *slog.Logger onUpdate func(ifname string, data json.RawMessage) } // New creates an EthMonitor, resolves the ethtool genetlink family, // and joins its "monitor" multicast group. -func New(log *slog.Logger) (*EthMonitor, error) { +func New(log *slog.Logger, cmd CommandRunner) (*EthMonitor, error) { conn, err := genetlink.Dial(nil) if err != nil { return nil, fmt.Errorf("dial genetlink: %w", err) @@ -70,18 +78,12 @@ func New(log *slog.Logger) (*EthMonitor, error) { return nil, fmt.Errorf("join ethtool monitor group %d: %w", groupID, err) } - etClient, err := ethtool.New() - if err != nil { - _ = conn.Close() - return nil, fmt.Errorf("create ethtool client: %w", err) - } - return &EthMonitor{ - conn: conn, - family: family, - groupID: groupID, - etClient: etClient, - log: log, + conn: conn, + family: family, + groupID: groupID, + cmd: cmd, + log: log, }, nil } @@ -93,10 +95,8 @@ func (m *EthMonitor) SetOnUpdate(fn func(string, json.RawMessage)) { // Run starts the ethtool genetlink receive loop and updates interface // ethernet settings when link info or link mode notifications are seen. func (m *EthMonitor) Run(ctx context.Context) error { + m.ctx = ctx defer func() { - if err := m.etClient.Close(); err != nil { - m.log.Warn("ethmonitor: close ethtool client", "err", err) - } if err := m.conn.Close(); err != nil { m.log.Warn("ethmonitor: close genetlink conn", "err", err) } @@ -135,35 +135,48 @@ func (m *EthMonitor) RefreshInterface(ifname string) { m.refreshEthernetSettings(ifname) } +// ethtoolJSON represents the relevant fields from `ethtool --json `. +type ethtoolJSON struct { + Speed int `json:"speed"` + Duplex string `json:"duplex"` + Port string `json:"port"` + AutoNegotiation bool `json:"auto-negotiation"` + SupportedLinkModes []string `json:"supported-link-modes"` + AdvertisedLinkModes []string `json:"advertised-link-modes"` +} + func (m *EthMonitor) refreshEthernetSettings(ifname string) { - iface, err := net.InterfaceByName(ifname) + ctx := m.ctx + if ctx == nil { + ctx = context.Background() + } + + out, err := m.cmd.Run(ctx, "ethtool", "--json", ifname) if err != nil { - m.log.Warn("ethmonitor: lookup interface", "ifname", ifname, "err", err) + m.log.Warn("ethmonitor: run ethtool", "ifname", ifname, "err", err) return } - ethIface := ethtool.Interface{Index: iface.Index, Name: iface.Name} - - if _, err := m.etClient.LinkInfo(ethIface); err != nil { - m.log.Warn("ethmonitor: query link info", "ifname", ifname, "err", err) + var results []ethtoolJSON + if err := json.Unmarshal(out, &results); err != nil { + m.log.Warn("ethmonitor: parse ethtool json", "ifname", ifname, "err", err) + return + } + if len(results) == 0 { return } - mode, err := m.etClient.LinkMode(ethIface) - if err != nil { - m.log.Warn("ethmonitor: query link mode", "ifname", ifname, "err", err) - return + data := results[0] + eth, speedBPS := buildEthernetContainer(data) + + // Marshal the result; include interface-level speed as a special key + // that mergeAugments will lift onto the interface object. + result := map[string]any{"ethernet": eth} + if speedBPS > 0 { + result["speed"] = fmt.Sprintf("%d", speedBPS) } - eth := map[string]any{ - "speed": speedString(mode.SpeedMegabits), - "duplex": duplexString(mode.Duplex), - "auto-negotiation": map[string]any{ - "enable": mode.Autoneg == ethtool.AutonegOn, - }, - } - - raw, err := json.Marshal(eth) + raw, err := json.Marshal(result) if err != nil { m.log.Warn("ethmonitor: marshal ethernet settings", "ifname", ifname, "err", err) return @@ -174,6 +187,55 @@ func (m *EthMonitor) refreshEthernetSettings(ifname string) { } } +// buildEthernetContainer builds the ieee802-ethernet-interface:ethernet +// container and returns (container, interface speed in bits/s or 0). +func buildEthernetContainer(data ethtoolJSON) (map[string]any, int64) { + autoneg := map[string]any{"enable": data.AutoNegotiation} + eth := map[string]any{"auto-negotiation": autoneg} + + duplex := strings.ToLower(data.Duplex) + if duplex == "full" || duplex == "half" { + eth["duplex"] = duplex + } + + // Supported PMD types (config-false leaf-list). + supported := ethtoolModesToPMD(data.SupportedLinkModes) + if len(supported) > 0 { + eth["infix-ethernet-interface:supported-pmd-types"] = supported + } + + // Advertised PMD types — suppress when identical to supported (default). + advertised := ethtoolModesToPMD(data.AdvertisedLinkModes) + if len(advertised) > 0 && !stringSliceEqual(advertised, supported) { + autoneg["infix-ethernet-interface:advertised-pmd-types"] = advertised + } + + // Speed, phy-type, pmd-type. + var speedBPS int64 + speedMbps := data.Speed + if speedMbps > 0 && speedMbps < ethtoolSpeedUnknown { + speedBPS = int64(speedMbps) * 1_000_000 + + // Speed inside the ethernet container (decimal64, Gb/s). + eth["speed"] = fmt.Sprintf("%.3f", float64(speedMbps)/1000.0) + + key := linkModeKey{Port: data.Port, SpeedMbps: speedMbps, Duplex: duplex} + if mapping, ok := linkModes[key]; ok { + eth["phy-type"] = "ieee802-ethernet-phy-type:phy-type-" + mapping.PhyType + if mapping.PMDType != "" { + eth["pmd-type"] = "ieee802-ethernet-phy-type:pmd-type-" + mapping.PMDType + } + } + + // Refine pmd-type when exactly one supported mode (specific SFP). + if len(supported) == 1 { + eth["pmd-type"] = supported[0] + } + } + + return eth, speedBPS +} + func extractIfname(data []byte) (string, error) { ad, err := netlink.NewAttributeDecoder(data) if err != nil { @@ -210,20 +272,118 @@ func extractIfname(data []byte) (string, error) { return "", fmt.Errorf("header ifindex attribute not found") } -func duplexString(d ethtool.Duplex) string { - switch d { - case ethtool.Full: - return "full" - case ethtool.Half: - return "half" - default: - return "unknown" - } +// linkModeKey is the lookup key for phy-type/pmd-type mapping. +type linkModeKey struct { + Port string + SpeedMbps int + Duplex string } -// speedString converts a speed in megabits to a decimal64 string in -// Gb/s with 3 fraction digits, matching the ieee802-ethernet-interface -// YANG model's eth-if-speed-type (decimal64, fraction-digits 3, units Gb/s). -func speedString(megabits int) string { - return fmt.Sprintf("%.3f", float64(megabits)/1000.0) +// linkModeMapping holds the IEEE identity suffixes. +type linkModeMapping struct { + PhyType string + PMDType string // empty means "cannot determine from this tuple alone" +} + +// linkModes maps (port, speed, duplex) → (phy-type, pmd-type) per +// IEEE Std 802.3.2-2025 (ieee802-ethernet-phy-type). +var linkModes = map[linkModeKey]linkModeMapping{ + {"Twisted Pair", 10, "full"}: {"10BASE-T", "10BASE-T"}, + {"Twisted Pair", 10, "half"}: {"10BASE-T", "10BASE-T"}, + {"Twisted Pair", 100, "full"}: {"100BASE-X", "100BASE-TX"}, + {"Twisted Pair", 100, "half"}: {"100BASE-X", "100BASE-TX"}, + {"Twisted Pair", 1000, "full"}: {"1000BASE-T", "1000BASE-T"}, + {"Twisted Pair", 1000, "half"}: {"1000BASE-T", "1000BASE-T"}, + {"Twisted Pair", 2500, "full"}: {"2.5GBASE-T", "2.5GBASE-T"}, + {"Twisted Pair", 5000, "full"}: {"5GBASE-T", "5GBASE-T"}, + {"Twisted Pair", 10000, "full"}: {"10GBASE-T", "10GBASE-T"}, + {"Twisted Pair", 25000, "full"}: {"25GBASE-T", "25GBASE-T"}, + {"Twisted Pair", 40000, "full"}: {"40GBASE-T", "40GBASE-T"}, + {"MII", 10, "full"}: {"10BASE-T", "10BASE-T"}, + {"MII", 10, "half"}: {"10BASE-T", "10BASE-T"}, + {"MII", 100, "full"}: {"100BASE-X", "100BASE-TX"}, + {"MII", 100, "half"}: {"100BASE-X", "100BASE-TX"}, + {"FIBRE", 100, "full"}: {"100BASE-X", ""}, + {"FIBRE", 1000, "full"}: {"1000BASE-X", ""}, + {"FIBRE", 10000, "full"}: {"10GBASE-R", ""}, + {"FIBRE", 25000, "full"}: {"25GBASE-R", ""}, + {"FIBRE", 40000, "full"}: {"40GBASE-R", ""}, + {"FIBRE", 100000, "full"}: {"100GBASE-R", ""}, + {"Direct Attach Copper", 10000, "full"}: {"10GBASE-R", ""}, + {"Direct Attach Copper", 25000, "full"}: {"25GBASE-R", "25GBASE-CR"}, + {"Direct Attach Copper", 40000, "full"}: {"40GBASE-R", "40GBASE-CR4"}, + {"Direct Attach Copper", 100000, "full"}: {"100GBASE-R", "100GBASE-CR4"}, +} + +// ethtoolToPMD maps kernel link-mode base names to IEEE pmd-type +// identity suffixes. The kernel reports modes like "1000baseT/Full"; +// we strip the "/Full" or "/Half" suffix before lookup. +var ethtoolToPMD = map[string]string{ + "10baseT": "10BASE-T", + "10baseT1L": "10BASE-T1L", + "100baseT": "100BASE-TX", + "100baseT1": "100BASE-T1", + "100baseFX": "100BASE-FX", + "1000baseT": "1000BASE-T", + "1000baseT1": "1000BASE-T1", + "1000baseX": "1000BASE-LX", + "1000baseKX": "1000BASE-KX", + "2500baseT": "2.5GBASE-T", + "2500baseX": "2.5GBASE-X", + "5000baseT": "5GBASE-T", + "10000baseT": "10GBASE-T", + "10000baseSR": "10GBASE-SR", + "10000baseLR": "10GBASE-LR", + "10000baseLRM": "10GBASE-LRM", + "10000baseER": "10GBASE-ER", + "10000baseKR": "10GBASE-KR", + "10000baseKX4": "10GBASE-KX4", + "25000baseCR": "25GBASE-CR", + "25000baseSR": "25GBASE-SR", + "25000baseKR": "25GBASE-KR", + "40000baseCR4": "40GBASE-CR4", + "40000baseSR4": "40GBASE-SR4", + "40000baseLR4": "40GBASE-LR4", + "40000baseKR4": "40GBASE-KR4", + "100000baseCR4": "100GBASE-CR4", + "100000baseSR4": "100GBASE-SR4", + "100000baseLR4_ER4": "100GBASE-LR4", + "100000baseKR4": "100GBASE-KR4", +} + +// ethtoolModesToPMD translates a list of ethtool link-mode strings +// (e.g. "1000baseT/Full") into deduped, order-preserving PMD identity +// strings. +func ethtoolModesToPMD(modes []string) []string { + seen := make(map[string]bool) + var out []string + for _, entry := range modes { + base := entry + if idx := strings.IndexByte(entry, '/'); idx >= 0 { + base = entry[:idx] + } + pmd, ok := ethtoolToPMD[base] + if !ok || seen[pmd] { + continue + } + seen[pmd] = true + out = append(out, "ieee802-ethernet-phy-type:pmd-type-"+pmd) + } + return out +} + +func stringSliceEqual(a, b []string) bool { + if len(a) != len(b) { + return false + } + set := make(map[string]bool, len(a)) + for _, s := range a { + set[s] = true + } + for _, s := range b { + if !set[s] { + return false + } + } + return true } diff --git a/src/yangerd/internal/monitor/monitor.go b/src/yangerd/internal/monitor/monitor.go index 5b0c8c19..2fdd575d 100644 --- a/src/yangerd/internal/monitor/monitor.go +++ b/src/yangerd/internal/monitor/monitor.go @@ -7,6 +7,8 @@ import ( "fmt" "log/slog" "net" + "sort" + "strings" "sync" "syscall" "time" @@ -64,6 +66,11 @@ type NLMonitor struct { wireguard map[string]json.RawMessage // ifname → WireGuard peer-status JSON lastOperStatus map[string]string + + // stpMu guards lastSTP, a fingerprint of the most recent mstpd STP + // query, letting the periodic poll rebuild only on actual change. + stpMu sync.Mutex + lastSTP string } // New creates a netlink monitor backed by ip/bridge batch query workers. @@ -249,6 +256,10 @@ func (m *NLMonitor) initialDump() error { if err != nil { neighRaw = json.RawMessage(`[]`) } + mdbRaw, err := m.queryBridge("mdb show") + if err != nil { + mdbRaw = json.RawMessage(`[]`) + } m.log.Debug("initialDump", "linkBytes", len(linkRaw), "addrBytes", len(addrRaw), "neighBytes", len(neighRaw)) m.validateAddrData("initialDump", addrRaw) @@ -257,6 +268,9 @@ func (m *NLMonitor) initialDump() error { m.links = linkRaw m.addrs = addrRaw m.neighs = neighRaw + for _, bridgeName := range mdbBridgeNames(mdbRaw) { + m.mdb[bridgeName] = filterByMDBBridge(mdbRaw, bridgeName) + } for _, name := range interfaceNames(linkRaw) { if st, ok := extractOperStatus(filterByIfName(linkRaw, name)); ok { m.lastOperStatus[name] = st @@ -361,8 +375,8 @@ func (m *NLMonitor) handleMDBUpdate() { } m.mu.Lock() - for _, bridgeName := range bridgeNames(raw) { - m.mdb[bridgeName] = filterByBridge(raw, bridgeName) + for _, bridgeName := range mdbBridgeNames(raw) { + m.mdb[bridgeName] = filterByMDBBridge(raw, bridgeName) } m.mu.Unlock() @@ -448,6 +462,53 @@ func (m *NLMonitor) rebuild() { m.tree.Set(treeKey, doc) } +// RefreshSTP re-queries mstpd and rebuilds only when STP data changed. +// mstpd's control socket is request/response with no event channel, and +// the bridge-level root-id settles via BPDU exchange without any netlink +// event, so STP state must be polled to stay current. No bridges or an +// unchanged result costs one cheap mstpd query and no document re-marshal. +func (m *NLMonitor) RefreshSTP() { + links := m.Links() + resolver := stpquery.NewLinksIfIndexResolver(links) + brSTP, ptSTP := stpquery.Query(links, resolver) + if len(brSTP) == 0 && len(ptSTP) == 0 { + return + } + + fp := stpFingerprint(brSTP, ptSTP) + m.stpMu.Lock() + changed := fp != m.lastSTP + m.lastSTP = fp + m.stpMu.Unlock() + + if changed { + m.rebuild() + } +} + +func stpFingerprint(brSTP, ptSTP map[string]json.RawMessage) string { + var b strings.Builder + writeSortedRaw(&b, "b", brSTP) + writeSortedRaw(&b, "p", ptSTP) + return b.String() +} + +func writeSortedRaw(b *strings.Builder, prefix string, m map[string]json.RawMessage) { + keys := make([]string, 0, len(m)) + for k := range m { + keys = append(keys, k) + } + sort.Strings(keys) + for _, k := range keys { + b.WriteString(prefix) + b.WriteByte(':') + b.WriteString(k) + b.WriteByte('=') + b.Write(m[k]) + b.WriteByte('\n') + } +} + // mergeAugments adds ethernet, wifi, and bridge data into the // complete ietf-interfaces document produced by iface.Transform(). func mergeAugments(doc json.RawMessage, ethernet, wifi, fdb, mdb, bridgeSTP, portSTP, wireguard map[string]json.RawMessage) json.RawMessage { @@ -480,9 +541,20 @@ func mergeAugments(doc json.RawMessage, ethernet, wifi, fdb, mdb, bridgeSTP, por } if ethData, ok := ethernet[name]; ok { - var ethObj any - if err := json.Unmarshal(ethData, ðObj); err == nil { - ifaceObj["ieee802-ethernet-interface:ethernet"] = ethObj + var wrapper map[string]json.RawMessage + if err := json.Unmarshal(ethData, &wrapper); err == nil { + if ethRaw, ok := wrapper["ethernet"]; ok { + var ethObj any + if err := json.Unmarshal(ethRaw, ðObj); err == nil { + ifaceObj["ieee802-ethernet-interface:ethernet"] = ethObj + } + } + if speedRaw, ok := wrapper["speed"]; ok { + var speed string + if err := json.Unmarshal(speedRaw, &speed); err == nil { + ifaceObj["speed"] = speed + } + } } } @@ -503,9 +575,8 @@ func mergeAugments(doc json.RawMessage, ethernet, wifi, fdb, mdb, bridgeSTP, por if mdbData, ok := mdb[name]; ok { bridgeObj := ensureBridgeAugment(ifaceObj) - var mdbObj any - if err := json.Unmarshal(mdbData, &mdbObj); err == nil { - bridgeObj["mdb"] = mdbObj + if mf := transformMDB(mdbData); mf != nil { + bridgeObj["multicast-filters"] = mf } } @@ -982,3 +1053,119 @@ func filterByBridge(raw json.RawMessage, bridgeName string) json.RawMessage { } return json.RawMessage(out) } + +// parseMDBEntries extracts the flat list of MDB entries from the +// bridge batch output format: [{"mdb":[{entries...}],"router":{}}] +func parseMDBEntries(raw json.RawMessage) []map[string]any { + var wrappers []map[string]json.RawMessage + if err := json.Unmarshal(raw, &wrappers); err != nil { + return nil + } + + var all []map[string]any + for _, w := range wrappers { + mdbRaw, ok := w["mdb"] + if !ok { + continue + } + var entries []map[string]any + if err := json.Unmarshal(mdbRaw, &entries); err != nil { + continue + } + all = append(all, entries...) + } + return all +} + +func mdbBridgeNames(raw json.RawMessage) []string { + entries := parseMDBEntries(raw) + seen := make(map[string]struct{}, len(entries)) + var names []string + for _, e := range entries { + name, _ := e["dev"].(string) + if name == "" { + continue + } + if _, ok := seen[name]; ok { + continue + } + seen[name] = struct{}{} + names = append(names, name) + } + return names +} + +func filterByMDBBridge(raw json.RawMessage, bridgeName string) json.RawMessage { + entries := parseMDBEntries(raw) + var filtered []map[string]any + for _, e := range entries { + if dev, _ := e["dev"].(string); dev == bridgeName { + filtered = append(filtered, e) + } + } + if len(filtered) == 0 { + return json.RawMessage(`[]`) + } + out, err := json.Marshal(filtered) + if err != nil { + return json.RawMessage(`[]`) + } + return json.RawMessage(out) +} + +func transformMDB(raw json.RawMessage) map[string]any { + var entries []map[string]any + if err := json.Unmarshal(raw, &entries); err != nil || len(entries) == 0 { + return nil + } + + type portEntry struct { + Port string `json:"port"` + State string `json:"state"` + } + + groups := make(map[string][]portEntry) + var order []string + + for _, e := range entries { + grp, _ := e["grp"].(string) + port, _ := e["port"].(string) + state, _ := e["state"].(string) + if grp == "" || port == "" { + continue + } + + if _, seen := groups[grp]; !seen { + order = append(order, grp) + } + groups[grp] = append(groups[grp], portEntry{ + Port: port, + State: mdbStateToYANG(state), + }) + } + + if len(groups) == 0 { + return nil + } + + filters := make([]map[string]any, 0, len(order)) + for _, grp := range order { + filters = append(filters, map[string]any{ + "group": grp, + "ports": groups[grp], + }) + } + + return map[string]any{"multicast-filter": filters} +} + +func mdbStateToYANG(state string) string { + switch state { + case "temp": + return "temporary" + case "permanent": + return "permanent" + default: + return state + } +} diff --git a/src/yangerd/internal/monitor/monitor_test.go b/src/yangerd/internal/monitor/monitor_test.go index 081a69e3..bd2de1ef 100644 --- a/src/yangerd/internal/monitor/monitor_test.go +++ b/src/yangerd/internal/monitor/monitor_test.go @@ -338,7 +338,7 @@ func TestMergeAugments(t *testing.T) { doc := json.RawMessage(`{"interface":[{"name":"eth0","type":"infix-if-type:ethernet"},{"name":"br0","type":"infix-if-type:bridge"}]}`) eth := map[string]json.RawMessage{ - "eth0": json.RawMessage(`{"speed":1000,"duplex":"full"}`), + "eth0": json.RawMessage(`{"ethernet":{"speed":"1.000","duplex":"full"},"speed":"1000000000"}`), } fdb := map[string]json.RawMessage{ "br0": json.RawMessage(`[{"mac":"00:11:22:33:44:55"}]`), @@ -390,3 +390,69 @@ func TestTreeKey(t *testing.T) { t.Fatalf("treeKey = %q, want %q", treeKey, "ietf-interfaces:interfaces") } } + +func TestTransformMDB(t *testing.T) { + // transformMDB receives the output of filterByMDBBridge: a flat array of entries + raw := json.RawMessage(`[{"dev":"br0","port":"e3","grp":"224.1.1.1","state":"temp"},{"dev":"br0","port":"e4","grp":"224.1.1.1","state":"permanent"},{"dev":"br0","port":"e3","grp":"ff02::6a","state":"temp"}]`) + + result := transformMDB(raw) + if result == nil { + t.Fatal("expected non-nil result") + } + + filters, ok := result["multicast-filter"].([]map[string]any) + if !ok { + t.Fatalf("unexpected type: %T", result["multicast-filter"]) + } + if len(filters) != 2 { + t.Fatalf("expected 2 filters, got %d", len(filters)) + } + + if filters[0]["group"] != "224.1.1.1" { + t.Fatalf("unexpected group: %v", filters[0]["group"]) + } + + out, _ := json.Marshal(result) + if !json.Valid(out) { + t.Fatalf("invalid JSON: %s", out) + } +} + +func TestTransformMDBEmpty(t *testing.T) { + if transformMDB(json.RawMessage(`[]`)) != nil { + t.Fatal("expected nil for empty") + } + if transformMDB(json.RawMessage(`[{"dev":"br0","port":"br0","grp":"ff02::6a","state":"temp"}]`)) == nil { + t.Fatal("expected non-nil for router-only entry") + } +} + +func TestMDBBridgeNames(t *testing.T) { + raw := json.RawMessage(`[{"mdb":[{"dev":"br0","port":"e3","grp":"224.1.1.1","state":"temp"}],"router":{}},{"mdb":[{"dev":"br1","port":"e5","grp":"ff02::1","state":"temp"}],"router":{}}]`) + names := mdbBridgeNames(raw) + if len(names) != 2 || names[0] != "br0" || names[1] != "br1" { + t.Fatalf("unexpected names: %v", names) + } +} + +func TestSTPFingerprintDeterministic(t *testing.T) { + br1 := map[string]json.RawMessage{ + "br0": json.RawMessage(`{"root-id":"1.000.00:a0:85:00:01:00"}`), + "br1": json.RawMessage(`{"root-id":"2.000.00:a0:85:00:02:00"}`), + } + br2 := map[string]json.RawMessage{ + "br1": json.RawMessage(`{"root-id":"2.000.00:a0:85:00:02:00"}`), + "br0": json.RawMessage(`{"root-id":"1.000.00:a0:85:00:01:00"}`), + } + pt1 := map[string]json.RawMessage{"e1": json.RawMessage(`{"state":"forwarding"}`)} + pt2 := map[string]json.RawMessage{"e1": json.RawMessage(`{"state":"forwarding"}`)} + + if stpFingerprint(br1, pt1) != stpFingerprint(br2, pt2) { + t.Fatal("fingerprint must be independent of map iteration order") + } + + br2["br0"] = json.RawMessage(`{"root-id":"8.000.00:a0:85:00:01:00"}`) + if stpFingerprint(br1, pt1) == stpFingerprint(br2, pt2) { + t.Fatal("fingerprint must change when STP data changes") + } +} diff --git a/src/yangerd/internal/zapiwatcher/zapiwatcher.go b/src/yangerd/internal/zapiwatcher/zapiwatcher.go index 69148309..6e4e8c63 100644 --- a/src/yangerd/internal/zapiwatcher/zapiwatcher.go +++ b/src/yangerd/internal/zapiwatcher/zapiwatcher.go @@ -6,9 +6,9 @@ import ( "fmt" "log/slog" "net" - "sort" + "regexp" "strconv" - "sync" + "strings" "time" "github.com/kernelkit/infix/src/yangerd/internal/backoff" @@ -16,8 +16,23 @@ import ( "github.com/kernelkit/infix/src/yangerd/internal/zapi" ) -const zapiSocketPath = "/var/run/frr/zserv.api" +const ( + zapiSocketPath = "/var/run/frr/zserv.api" + routingTreeKey = "ietf-routing:routing" + // debounceDelay coalesces a burst of ZAPI route notifications into a + // single RIB read. FRR emits many RouteAdd/Del messages while a + // protocol converges; without debouncing we would re-read the table + // dozens of times in a few milliseconds. + debounceDelay = 200 * time.Millisecond +) + +// subscribeTypes are the route types we ask zebra to redistribute. We do +// not use the route payloads themselves -- redistribution only ever +// delivers the selected route per destination and does not reliably send +// a delete when a route is superseded. The subscription exists purely so +// zebra notifies us that *something* changed; the authoritative table is +// then read from zebra's vty socket (see RouteQuerier). var subscribeTypes = []zapi.RouteType{ zapi.RouteKernel, zapi.RouteConnect, @@ -27,39 +42,45 @@ var subscribeTypes = []zapi.RouteType{ zapi.RouteOSPF, } -var routeTypeToProtocol = map[zapi.RouteType]string{ - zapi.RouteKernel: "infix-routing:kernel", - zapi.RouteConnect: "ietf-routing:direct", - zapi.RouteLocal: "ietf-routing:direct", - zapi.RouteStatic: "ietf-routing:static", - zapi.RouteOSPF: "ietf-ospf:ospfv2", - zapi.RouteRIP: "ietf-rip:rip", -} - -// routeEntry pairs a ZAPI route with the wall-clock time it was first -// received. The timestamp is used for the YANG last-updated leaf. -type routeEntry struct { - route *zapi.Route - receivedAt time.Time +// RouteQuerier runs a "show ... json" command against FRR and returns its +// raw output. Production code uses an frrvty.Client (zebra's vty socket); +// tests inject a fake. +type RouteQuerier interface { + Query(ctx context.Context, command string) ([]byte, error) } +// ZAPIWatcher keeps the operational RIB (ietf-routing:routing/ribs) in +// sync with FRR. It does NOT reconstruct routes from the ZAPI stream: +// the ZAPI socket is used only as a change trigger, and the full routing +// table -- every candidate per destination, with FRR's own +// selected/installed flags, exactly as "show ip route" renders it -- is +// read from zebra's vty socket on each change. Because every refresh is +// a complete snapshot, a route removed from zebra simply disappears; we +// never depend on receiving a ZAPI delete. type ZAPIWatcher struct { - tree *tree.Tree - log *slog.Logger - mu sync.Mutex - routes map[string]*routeEntry + tree *tree.Tree + querier RouteQuerier + log *slog.Logger + refresh chan struct{} } -const routingTreeKey = "ietf-routing:routing" - -func New(t *tree.Tree, log *slog.Logger) *ZAPIWatcher { +func New(t *tree.Tree, querier RouteQuerier, log *slog.Logger) *ZAPIWatcher { if log == nil { log = slog.Default() } - return &ZAPIWatcher{tree: t, log: log, routes: make(map[string]*routeEntry)} + return &ZAPIWatcher{ + tree: t, + querier: querier, + log: log, + refresh: make(chan struct{}, 1), + } } func (w *ZAPIWatcher) Run(ctx context.Context) error { + // The refresh worker owns all writes to the tree and runs for the + // lifetime of the watcher, independent of the ZAPI connection. + go w.refreshLoop(ctx) + bo := backoff.Default() delay := bo.Initial @@ -81,6 +102,10 @@ func (w *ZAPIWatcher) Run(ctx context.Context) error { delay = bo.Initial w.log.Info("zapi watcher: connected", "socket", zapiSocketPath) + // Read the current table now that we are subscribed, so we have + // data even if no further events arrive. + w.triggerRefresh() + err = w.processMessages(ctx, conn) _ = conn.Close() if ctx.Err() != nil { @@ -88,7 +113,6 @@ func (w *ZAPIWatcher) Run(ctx context.Context) error { } w.log.Warn("zapi watcher: disconnected", "err", err) - w.clearAllRoutes() } } @@ -127,6 +151,9 @@ func (w *ZAPIWatcher) connect(ctx context.Context) (net.Conn, error) { return conn, nil } +// processMessages drains the ZAPI stream. We only care *that* a route +// changed, not what changed -- each route add/delete triggers a debounced +// re-read of the full table. func (w *ZAPIWatcher) processMessages(ctx context.Context, conn net.Conn) error { for { select { @@ -135,235 +162,310 @@ func (w *ZAPIWatcher) processMessages(ctx context.Context, conn net.Conn) error default: } - hdr, body, err := zapi.ReadMessage(conn) + hdr, _, err := zapi.ReadMessage(conn) if err != nil { return fmt.Errorf("read message: %w", err) } - w.handleMessage(hdr, body) + switch hdr.Command { + case zapi.CmdRedistRouteAdd, zapi.CmdRedistRouteDel: + w.log.Debug("zapi watcher: route change", "cmd", hdr.Command, "vrf", hdr.VrfID) + w.triggerRefresh() + } } } -func (w *ZAPIWatcher) handleMessage(hdr zapi.Header, body []byte) { - w.log.Debug("zapi watcher: message", "cmd", hdr.Command, "vrf", hdr.VrfID, "len", hdr.Length) +// triggerRefresh requests a table re-read. The buffered channel collapses +// multiple pending requests into one. +func (w *ZAPIWatcher) triggerRefresh() { + select { + case w.refresh <- struct{}{}: + default: + } +} - switch hdr.Command { - case zapi.CmdRedistRouteAdd: - route, err := zapi.DecodeRouteLog(body, w.log) - if err != nil { - w.log.Warn("zapi watcher: decode route add", "err", err, "bodyLen", len(body)) +func (w *ZAPIWatcher) refreshLoop(ctx context.Context) { + for { + select { + case <-ctx.Done(): return + case <-w.refresh: } - w.log.Debug("zapi watcher: route add", - "type", route.Type, - "prefix", route.Prefix.String(), - "distance", route.Distance, - "metric", route.Metric, - "nexthops", len(route.Nexthops), - "msg", route.Message, - ) - w.addRoute(route) - case zapi.CmdRedistRouteDel: - route, err := zapi.DecodeRoute(body) - if err != nil { - w.log.Warn("zapi watcher: decode route del", "err", err, "bodyLen", len(body)) + + // Let a burst of notifications settle before reading. + select { + case <-ctx.Done(): return + case <-time.After(debounceDelay): } - w.log.Debug("zapi watcher: route del", - "type", route.Type, - "prefix", route.Prefix.String(), - ) - w.deleteRoute(route) - } -} - -// routeKey returns a unique map key for a redistributed route. -// FRR's redistribution sends a single RouteAdd per prefix+protocol — -// nexthop changes are updates (no preceding RouteDel), so the key -// must NOT include nexthops or distance. -func routeKey(route *zapi.Route) string { - return ribName(route.Prefix) + ":" + route.Prefix.String() + ":" + routeProtocol(route.Type) -} - -func (w *ZAPIWatcher) addRoute(route *zapi.Route) { - key := routeKey(route) - now := time.Now() - - w.mu.Lock() - if existing, ok := w.routes[key]; ok { - now = existing.receivedAt - } - w.routes[key] = &routeEntry{route: route, receivedAt: now} - routeCount := len(w.routes) - w.mu.Unlock() - - w.log.Debug("zapi watcher: stored route", "key", key, "totalRoutes", routeCount) - w.writeRibs() -} - -func (w *ZAPIWatcher) deleteRoute(route *zapi.Route) { - key := routeKey(route) - - w.mu.Lock() - delete(w.routes, key) - w.mu.Unlock() - - w.writeRibs() -} - -func (w *ZAPIWatcher) clearAllRoutes() { - w.mu.Lock() - w.routes = make(map[string]*routeEntry) - w.mu.Unlock() - - w.writeRibs() -} - -// destKey returns the grouping key for active route computation: -// "rib:prefix" — all routes sharing the same destination compete. -func destKey(route *zapi.Route) string { - return ribName(route.Prefix) + ":" + route.Prefix.String() -} - -func (w *ZAPIWatcher) writeRibs() { - w.mu.Lock() - - bestDist := make(map[string]uint8) - for _, entry := range w.routes { - dk := destKey(entry.route) - if d, ok := bestDist[dk]; !ok || entry.route.Distance < d { - bestDist[dk] = entry.route.Distance + // Drain a request that arrived during the debounce window; the + // upcoming read already reflects it. + select { + case <-w.refresh: + default: } + + w.writeRibs(ctx) } +} - // Collect entries into a slice for deterministic ordering. - allEntries := make([]*routeEntry, 0, len(w.routes)) - for _, entry := range w.routes { - allEntries = append(allEntries, entry) +// writeRibs reads the full IPv4 and IPv6 routing tables from zebra and +// replaces the ribs subtree. On a query error it leaves the previous +// data untouched rather than blanking the table. +func (w *ZAPIWatcher) writeRibs(ctx context.Context) { + ipv4, err := w.collectRoutes(ctx, "ipv4") + if err != nil { + w.log.Warn("zapi watcher: read ipv4 routes", "err", err) + return } - - // Sort by destination prefix so routes for the same destination are - // grouped together. Within the same prefix, lowest distance first - // (active route on top). - sort.Slice(allEntries, func(i, j int) bool { - a, b := allEntries[i].route, allEntries[j].route - ap, bp := a.Prefix.String(), b.Prefix.String() - if ap != bp { - return ap < bp - } - return a.Distance < b.Distance - }) - - ipv4Routes := make([]json.RawMessage, 0, len(allEntries)) - ipv6Routes := make([]json.RawMessage, 0, len(allEntries)) - - for _, entry := range allEntries { - dk := destKey(entry.route) - active := entry.route.Distance == bestDist[dk] - routeData := transformRoute(entry.route, active, entry.receivedAt) - - if entry.route.Prefix.IP.To4() != nil { - ipv4Routes = append(ipv4Routes, routeData) - } else { - ipv6Routes = append(ipv6Routes, routeData) - } + ipv6, err := w.collectRoutes(ctx, "ipv6") + if err != nil { + w.log.Warn("zapi watcher: read ipv6 routes", "err", err) + return } - w.mu.Unlock() ribs := map[string]any{ "rib": []map[string]any{ { "name": "ipv4", "address-family": "ietf-routing:ipv4", - "routes": map[string]any{ - "route": ipv4Routes, - }, + "routes": map[string]any{"route": ipv4}, }, { "name": "ipv6", "address-family": "ietf-routing:ipv6", - "routes": map[string]any{ - "route": ipv6Routes, - }, + "routes": map[string]any{"route": ipv6}, }, }, } - ribsJSON, err := json.Marshal(map[string]any{"ribs": ribs}) + data, err := json.Marshal(map[string]any{"ribs": ribs}) if err != nil { w.log.Error("zapi watcher: marshal ribs", "err", err) return } - w.tree.Merge(routingTreeKey, ribsJSON) + w.tree.Merge(routingTreeKey, data) } -func ribName(prefix net.IPNet) string { - if prefix.IP.To4() != nil { - return "ipv4" - } - return "ipv6" -} - -func transformRoute(route *zapi.Route, active bool, receivedAt time.Time) json.RawMessage { - isIPv4 := route.Prefix.IP.To4() != nil - - addrKey := "ietf-ipv6-unicast-routing:address" - dpKey := "ietf-ipv6-unicast-routing:destination-prefix" - if isIPv4 { - addrKey = "ietf-ipv4-unicast-routing:address" - dpKey = "ietf-ipv4-unicast-routing:destination-prefix" +// collectRoutes runs "show ip route json" / "show ipv6 route json" and +// transforms every entry into an ietf-routing route node. +func (w *ZAPIWatcher) collectRoutes(ctx context.Context, family string) ([]json.RawMessage, error) { + command := "show ip route json" + if family == "ipv6" { + command = "show ipv6 route json" } - nextHops := make([]map[string]any, 0, len(route.Nexthops)) - for _, nh := range route.Nexthops { - hop := make(map[string]any) - - if len(nh.Gate) > 0 && !nh.Gate.IsUnspecified() { - hop[addrKey] = nh.Gate.String() - } - - if nh.Ifindex > 0 { - ifi, err := net.InterfaceByIndex(int(nh.Ifindex)) - if err == nil && ifi != nil && ifi.Name != "" { - hop["outgoing-interface"] = ifi.Name - } else { - hop["outgoing-interface"] = strconv.FormatUint(uint64(nh.Ifindex), 10) - } - } - - if len(hop) > 0 { - nextHops = append(nextHops, hop) - } - } - - routeNode := map[string]any{ - dpKey: route.Prefix.String(), - "source-protocol": routeProtocol(route.Type), - "route-preference": route.Distance, - "last-updated": receivedAt.Format(time.RFC3339), - "next-hop": map[string]any{ - "next-hop-list": map[string]any{ - "next-hop": nextHops, - }, - }, - } - - if active { - routeNode["active"] = []any{nil} - } - - encoded, err := json.Marshal(routeNode) + out, err := w.querier.Query(ctx, command) if err != nil { - return json.RawMessage(`{}`) + return nil, err } - return json.RawMessage(encoded) + // FRR prints "{}" for an empty table; otherwise a map of + // prefix -> [route, ...] (multiple candidates per prefix). + var table map[string][]map[string]any + if err := json.Unmarshal(out, &table); err != nil { + return nil, fmt.Errorf("parse %q: %w", command, err) + } + + now := time.Now() + routes := make([]json.RawMessage, 0, len(table)) + for prefix, entries := range table { + if !strings.Contains(prefix, "/") { + continue + } + for _, entry := range entries { + routes = append(routes, transformRoute(family, prefix, entry, now)) + } + } + return routes, nil } -func routeProtocol(rt zapi.RouteType) string { - if protocol, ok := routeTypeToProtocol[rt]; ok { - return protocol +// protocolMap maps FRR's protocol names to IETF routing-protocol +// identities. Unknown protocols fall back to kernel so they still +// validate against the model. +var protocolMap = map[string]string{ + "kernel": "infix-routing:kernel", + "connected": "ietf-routing:direct", + "local": "ietf-routing:direct", + "static": "ietf-routing:static", + "ospf": "ietf-ospf:ospfv2", + "ospf6": "ietf-ospf:ospfv3", + "rip": "ietf-rip:rip", + "ripng": "ietf-rip:rip", +} + +func protocolName(frr string) string { + if p, ok := protocolMap[frr]; ok { + return p } return "infix-routing:kernel" } + +// transformRoute converts one FRR JSON route entry into an ietf-routing +// route node. It mirrors the legacy yanger ietf_routing.py:add_protocol. +func transformRoute(family, prefixKey string, route map[string]any, now time.Time) json.RawMessage { + addrKey := "ietf-ipv4-unicast-routing:address" + dpKey := "ietf-ipv4-unicast-routing:destination-prefix" + nhAddrKey := "ietf-ipv4-unicast-routing:next-hop-address" + hostLen := "32" + if family == "ipv6" { + addrKey = "ietf-ipv6-unicast-routing:address" + dpKey = "ietf-ipv6-unicast-routing:destination-prefix" + nhAddrKey = "ietf-ipv6-unicast-routing:next-hop-address" + hostLen = "128" + } + + dst := stringField(route, "prefix") + if dst == "" { + dst = prefixKey + } + if !strings.Contains(dst, "/") { + plen := hostLen + if v, ok := route["prefixLen"]; ok { + plen = strconv.Itoa(toInt(v)) + } + dst = dst + "/" + plen + } + + frr := stringField(route, "protocol") + + node := map[string]any{ + dpKey: dst, + "source-protocol": protocolName(frr), + "route-preference": toInt(route["distance"]), + "last-updated": now.Add(-parseUptime(stringField(route, "uptime"))).Format(time.RFC3339), + } + + // Metric is modelled only for OSPF and RIP routes. + switch { + case strings.Contains(frr, "ospf"): + node["ietf-ospf:metric"] = toInt(route["metric"]) + case strings.Contains(frr, "rip"): + node["ietf-rip:metric"] = toInt(route["metric"]) + } + + // "selected" is FRR's own best-path decision -- the '>' in + // "show ip route". active is a presence leaf, encoded as [null]. + if boolField(route, "selected") { + node["active"] = []any{nil} + } + + installed := boolField(route, "installed") + + nextHops := make([]map[string]any, 0) + if hops, ok := route["nexthops"].([]any); ok { + for _, h := range hops { + hop, ok := h.(map[string]any) + if !ok { + continue + } + nh := map[string]any{} + if ip := stringField(hop, "ip"); ip != "" { + nh[addrKey] = ip + } else if ifn := stringField(hop, "interfaceName"); ifn != "" { + nh["outgoing-interface"] = ifn + } + // zebra marks the nexthop programmed into the FIB with + // "fib":true (see zebra/zebra_vty.c). + if installed && boolField(hop, "fib") { + nh["infix-routing:installed"] = []any{nil} + } + if len(nh) > 0 { + nextHops = append(nextHops, nh) + } + } + } + + if len(nextHops) > 0 { + node["next-hop"] = map[string]any{ + "next-hop-list": map[string]any{ + "next-hop": nextHops, + }, + } + } else { + nh := map[string]any{} + switch frr { + case "blackhole": + nh["special-next-hop"] = "blackhole" + case "unreachable": + nh["special-next-hop"] = "unreachable" + default: + if ifn := stringField(route, "interfaceName"); ifn != "" { + nh["outgoing-interface"] = ifn + } + if gw := stringField(route, "nexthop"); gw != "" { + nh[nhAddrKey] = gw + } + } + node["next-hop"] = nh + } + + encoded, err := json.Marshal(node) + if err != nil { + return json.RawMessage(`{}`) + } + return encoded +} + +func stringField(m map[string]any, key string) string { + if s, ok := m[key].(string); ok { + return s + } + return "" +} + +func boolField(m map[string]any, key string) bool { + b, _ := m[key].(bool) + return b +} + +func toInt(v any) int { + switch n := v.(type) { + case float64: + return int(n) + case int: + return n + case int64: + return int(n) + case json.Number: + i, _ := n.Int64() + return int(i) + case string: + i, _ := strconv.Atoi(n) + return i + } + return 0 +} + +// FRR uptime string formats (frrtime), ported from yanger's +// uptime2datetime: "HH:MM:SS", "XdXXhXXm", "XXwXdXXh". +var ( + uptimeHMS = regexp.MustCompile(`^(\d{2}):(\d{2}):(\d{2})$`) + uptimeDHM = regexp.MustCompile(`^(\d+)d(\d{2})h(\d{2})m$`) + uptimeWDH = regexp.MustCompile(`^(\d{2})w(\d)d(\d{2})h$`) +) + +// parseUptime converts an FRR uptime string into a duration. The +// last-updated leaf is then computed as now-uptime. Unrecognised input +// yields zero (i.e. last-updated == now). +func parseUptime(s string) time.Duration { + atoi := func(x string) int { n, _ := strconv.Atoi(x); return n } + + if m := uptimeHMS.FindStringSubmatch(s); m != nil { + return time.Duration(atoi(m[1]))*time.Hour + + time.Duration(atoi(m[2]))*time.Minute + + time.Duration(atoi(m[3]))*time.Second + } + if m := uptimeDHM.FindStringSubmatch(s); m != nil { + return time.Duration(atoi(m[1]))*24*time.Hour + + time.Duration(atoi(m[2]))*time.Hour + + time.Duration(atoi(m[3]))*time.Minute + } + if m := uptimeWDH.FindStringSubmatch(s); m != nil { + return time.Duration(atoi(m[1]))*7*24*time.Hour + + time.Duration(atoi(m[2]))*24*time.Hour + + time.Duration(atoi(m[3]))*time.Hour + } + return 0 +} diff --git a/src/yangerd/internal/zapiwatcher/zapiwatcher_test.go b/src/yangerd/internal/zapiwatcher/zapiwatcher_test.go index be066f2c..1541809a 100644 --- a/src/yangerd/internal/zapiwatcher/zapiwatcher_test.go +++ b/src/yangerd/internal/zapiwatcher/zapiwatcher_test.go @@ -1,427 +1,269 @@ package zapiwatcher import ( + "context" "encoding/json" - "net" "testing" "time" "github.com/kernelkit/infix/src/yangerd/internal/tree" - "github.com/kernelkit/infix/src/yangerd/internal/zapi" ) -func TestRouteProtocol(t *testing.T) { - tests := []struct { - name string - rt zapi.RouteType - expected string - }{ - {"kernel", zapi.RouteKernel, "infix-routing:kernel"}, - {"connect", zapi.RouteConnect, "ietf-routing:direct"}, - {"static", zapi.RouteStatic, "ietf-routing:static"}, - {"ospf", zapi.RouteOSPF, "ietf-ospf:ospfv2"}, - {"rip", zapi.RouteRIP, "ietf-rip:rip"}, - {"unknown defaults to kernel", zapi.RouteType(99), "infix-routing:kernel"}, - {"zero defaults to kernel", zapi.RouteType(0), "infix-routing:kernel"}, - } - - for _, tc := range tests { - t.Run(tc.name, func(t *testing.T) { - got := routeProtocol(tc.rt) - if got != tc.expected { - t.Errorf("routeProtocol(%d) = %q, want %q", tc.rt, got, tc.expected) - } - }) - } +// fakeQuerier returns canned vtysh JSON per command. +type fakeQuerier struct { + ipv4 string + ipv6 string + err error } -func TestRibName(t *testing.T) { - tests := []struct { - name string - prefix net.IPNet - expected string - }{ - { - "ipv4", - net.IPNet{IP: net.ParseIP("10.0.0.0").To4(), Mask: net.CIDRMask(24, 32)}, - "ipv4", - }, - { - "ipv6", - net.IPNet{IP: net.ParseIP("2001:db8::"), Mask: net.CIDRMask(64, 128)}, - "ipv6", - }, - { - "loopback v4", - net.IPNet{IP: net.ParseIP("127.0.0.0").To4(), Mask: net.CIDRMask(8, 32)}, - "ipv4", - }, +func (f fakeQuerier) Query(_ context.Context, command string) ([]byte, error) { + if f.err != nil { + return nil, f.err } - - for _, tc := range tests { - t.Run(tc.name, func(t *testing.T) { - got := ribName(tc.prefix) - if got != tc.expected { - t.Errorf("ribName(%v) = %q, want %q", tc.prefix, got, tc.expected) - } - }) + if command == "show ipv6 route json" { + if f.ipv6 == "" { + return []byte("{}"), nil + } + return []byte(f.ipv6), nil } + if f.ipv4 == "" { + return []byte("{}"), nil + } + return []byte(f.ipv4), nil } -func TestTransformRoute(t *testing.T) { - route := &zapi.Route{ - Type: zapi.RouteStatic, - Distance: 1, - Metric: 100, - Prefix: net.IPNet{ - IP: net.ParseIP("10.0.0.0").To4(), - Mask: net.CIDRMask(24, 32), - }, - } - - received := time.Date(2025, 6, 15, 10, 30, 0, 0, time.UTC) - result := transformRoute(route, false, received) - - var parsed map[string]any - if err := json.Unmarshal(result, &parsed); err != nil { - t.Fatalf("unmarshal transformRoute result: %v", err) - } - - if got := parsed["ietf-ipv4-unicast-routing:destination-prefix"]; got != "10.0.0.0/24" { - t.Errorf("destination-prefix = %v, want %q", got, "10.0.0.0/24") - } - - if got := parsed["source-protocol"]; got != "ietf-routing:static" { - t.Errorf("source-protocol = %v, want %q", got, "ietf-routing:static") - } - - if got, ok := parsed["route-preference"].(float64); !ok || int(got) != 1 { - t.Errorf("route-preference = %v, want 1 (admin distance)", parsed["route-preference"]) - } - - nhContainer, ok := parsed["next-hop"].(map[string]any) - if !ok { - t.Fatalf("next-hop not a map: %T", parsed["next-hop"]) - } - nhList, ok := nhContainer["next-hop-list"].(map[string]any) - if !ok { - t.Fatalf("next-hop-list not a map: %T", nhContainer["next-hop-list"]) - } - hops, ok := nhList["next-hop"].([]any) - if !ok { - t.Fatalf("next-hop not an array: %T", nhList["next-hop"]) - } - if len(hops) != 0 { - t.Errorf("expected 0 next-hops, got %d", len(hops)) - } - - if _, ok := parsed["active"]; ok { - t.Error("non-selected route should not have 'active' leaf") - } - - lastUpdated, ok := parsed["last-updated"].(string) - if !ok { - t.Fatal("last-updated missing or not a string") - } - if lastUpdated != "2025-06-15T10:30:00Z" { - t.Errorf("last-updated = %q, want %q", lastUpdated, "2025-06-15T10:30:00Z") - } -} - -func TestTransformRouteActiveParam(t *testing.T) { - route := &zapi.Route{ - Type: zapi.RouteStatic, - Distance: 5, - Prefix: net.IPNet{ - IP: net.ParseIP("0.0.0.0").To4(), - Mask: net.CIDRMask(0, 32), - }, - } - - result := transformRoute(route, true, time.Now()) - - var parsed map[string]any - if err := json.Unmarshal(result, &parsed); err != nil { - t.Fatalf("unmarshal: %v", err) - } - - active, ok := parsed["active"].([]any) - if !ok { - t.Fatalf("active leaf missing or wrong type: %v", parsed["active"]) - } - if len(active) != 1 || active[0] != nil { - t.Errorf("active = %v, want [null]", active) - } -} - -func TestRouteKeyDifferentProtocols(t *testing.T) { - ospf := &zapi.Route{ - Type: zapi.RouteOSPF, - Distance: 110, - Prefix: net.IPNet{ - IP: net.IPv4(10, 0, 0, 0), - Mask: net.CIDRMask(24, 32), - }, - Nexthops: []zapi.Nexthop{{ - Gate: net.ParseIP("192.168.10.3").To4(), - }}, - } - static := &zapi.Route{ - Type: zapi.RouteStatic, - Distance: 120, - Prefix: net.IPNet{ - IP: net.IPv4(10, 0, 0, 0), - Mask: net.CIDRMask(24, 32), - }, - Nexthops: []zapi.Nexthop{{ - Gate: net.ParseIP("192.168.50.2").To4(), - }}, - } - - keyOSPF := routeKey(ospf) - keyStatic := routeKey(static) - - if keyOSPF == keyStatic { - t.Errorf("routes with different protocols must have different keys: %q == %q", keyOSPF, keyStatic) - } -} - -func TestRouteKeySamePrefixProtoIgnoresDistance(t *testing.T) { - old := &zapi.Route{ - Type: zapi.RouteStatic, - Distance: 120, - Prefix: net.IPNet{ - IP: net.IPv4(0, 0, 0, 0), - Mask: net.CIDRMask(0, 32), - }, - } - updated := &zapi.Route{ - Type: zapi.RouteStatic, - Distance: 1, - Prefix: net.IPNet{ - IP: net.IPv4(0, 0, 0, 0), - Mask: net.CIDRMask(0, 32), - }, - } - - if routeKey(old) != routeKey(updated) { - t.Errorf("same prefix+proto must produce same key regardless of distance: %q vs %q", routeKey(old), routeKey(updated)) - } -} - -func TestAddRouteSamePrefixDifferentProtocol(t *testing.T) { - tr := tree.New() - w := New(tr, nil) - - ospfRoute := &zapi.Route{ - Type: zapi.RouteOSPF, - Distance: 110, - Prefix: net.IPNet{ - IP: net.IPv4(0, 0, 0, 0), - Mask: net.CIDRMask(0, 32), - }, - Nexthops: []zapi.Nexthop{{ - Type: zapi.NHIPv4IFIndex, - Gate: net.ParseIP("192.168.10.3").To4(), - }}, - } - staticRoute := &zapi.Route{ - Type: zapi.RouteStatic, - Distance: 120, - Prefix: net.IPNet{ - IP: net.IPv4(0, 0, 0, 0), - Mask: net.CIDRMask(0, 32), - }, - Nexthops: []zapi.Nexthop{{ - Type: zapi.NHIPv4IFIndex, - Gate: net.ParseIP("192.168.50.2").To4(), - }}, - } - - w.addRoute(ospfRoute) - w.addRoute(staticRoute) - - w.mu.Lock() - count := len(w.routes) - w.mu.Unlock() - - if count != 2 { - t.Errorf("expected 2 routes (different protocols), got %d", count) - } - +func ipv4Routes(t *testing.T, tr *tree.Tree) []map[string]any { + t.Helper() data := tr.Get(routingTreeKey) if data == nil { t.Fatal("routing tree key not set") } - var routing map[string]any if err := json.Unmarshal(data, &routing); err != nil { t.Fatalf("unmarshal routing: %v", err) } - - ribs, ok := routing["ribs"].(map[string]any) - if !ok { - t.Fatal("ribs not found") - } - ribList, ok := ribs["rib"].([]any) - if !ok { - t.Fatal("rib list not found") - } - - var ipv4Routes []any - for _, rib := range ribList { - ribMap := rib.(map[string]any) - if ribMap["name"] == "ipv4" { - routes := ribMap["routes"].(map[string]any) - ipv4Routes = routes["route"].([]any) - break - } - } - - if len(ipv4Routes) != 2 { - t.Fatalf("expected 2 IPv4 routes, got %d", len(ipv4Routes)) - } - - var activeCount int - for _, r := range ipv4Routes { - rm := r.(map[string]any) - pref := int(rm["route-preference"].(float64)) - _, hasActive := rm["active"] - if hasActive { - activeCount++ - if pref != 110 { - t.Errorf("active route has preference %d, want 110", pref) + ribs := routing["ribs"].(map[string]any) + for _, rib := range ribs["rib"].([]any) { + rm := rib.(map[string]any) + if rm["name"] == "ipv4" { + out := []map[string]any{} + for _, r := range rm["routes"].(map[string]any)["route"].([]any) { + out = append(out, r.(map[string]any)) } - } - if pref == 120 && hasActive { - t.Error("route with preference 120 should not be active") - } - if _, ok := rm["last-updated"].(string); !ok { - t.Errorf("route with preference %d missing last-updated", pref) + return out } } - if activeCount != 1 { - t.Errorf("expected exactly 1 active route, got %d", activeCount) + t.Fatal("ipv4 rib not found") + return nil +} + +func TestProtocolName(t *testing.T) { + cases := map[string]string{ + "kernel": "infix-routing:kernel", + "connected": "ietf-routing:direct", + "local": "ietf-routing:direct", + "static": "ietf-routing:static", + "ospf": "ietf-ospf:ospfv2", + "rip": "ietf-rip:rip", + "bgp": "infix-routing:kernel", // unknown -> kernel + "": "infix-routing:kernel", + } + for in, want := range cases { + if got := protocolName(in); got != want { + t.Errorf("protocolName(%q) = %q, want %q", in, got, want) + } } } -func TestTransformRouteIPv6(t *testing.T) { - route := &zapi.Route{ - Type: zapi.RouteOSPF, - Distance: 110, - Metric: 10, - Prefix: net.IPNet{ - IP: net.ParseIP("2001:db8::").To16(), - Mask: net.CIDRMask(48, 128), +func TestParseUptime(t *testing.T) { + cases := map[string]time.Duration{ + "02:09:02": 2*time.Hour + 9*time.Minute + 2*time.Second, + "00:00:30": 30 * time.Second, + "3d04h05m": 3*24*time.Hour + 4*time.Hour + 5*time.Minute, + "02w3d04h": 2*7*24*time.Hour + 3*24*time.Hour + 4*time.Hour, + "bogus": 0, + "": 0, + } + for in, want := range cases { + if got := parseUptime(in); got != want { + t.Errorf("parseUptime(%q) = %v, want %v", in, got, want) + } + } +} + +// The user's exact bug: a static route is in the FIB (selected, distance +// 120) while a stale OSPF entry (distance 110) is still listed but not +// selected. active must follow FRR's "selected" flag, not the lowest +// admin distance. +func TestActiveFollowsSelectedNotDistance(t *testing.T) { + const j = `{ + "192.168.20.0/24":[ + {"prefix":"192.168.20.0/24","protocol":"ospf","selected":false,"distance":110,"metric":100,"uptime":"02:11:49", + "nexthops":[{"ip":"192.168.60.2","interfaceName":"e3","active":true}]}, + {"prefix":"192.168.20.0/24","protocol":"static","selected":true,"installed":true,"distance":120,"metric":0,"uptime":"02:09:02", + "nexthops":[{"ip":"192.168.50.2","interfaceName":"e7","fib":true,"active":true}]} + ] + }` + + tr := tree.New() + w := New(tr, fakeQuerier{ipv4: j}, nil) + w.writeRibs(context.Background()) + + routes := ipv4Routes(t, tr) + if len(routes) != 2 { + t.Fatalf("expected 2 candidate routes, got %d", len(routes)) + } + + for _, r := range routes { + pref := toInt(r["route-preference"]) + _, active := r["active"] + switch pref { + case 120: + if !active { + t.Error("static route (pref 120, selected) must be active") + } + case 110: + if active { + t.Error("ospf route (pref 110, not selected) must NOT be active") + } + default: + t.Errorf("unexpected route-preference %d", pref) + } + } +} + +// A full snapshot read means a route removed from zebra disappears from +// the cache without any ZAPI delete. +func TestSnapshotPurgesRemovedRoutes(t *testing.T) { + const before = `{ + "192.168.20.0/24":[ + {"prefix":"192.168.20.0/24","protocol":"ospf","selected":true,"distance":110,"uptime":"00:05:00","nexthops":[{"ip":"192.168.60.2"}]} + ] + }` + const after = `{ + "192.168.20.0/24":[ + {"prefix":"192.168.20.0/24","protocol":"static","selected":true,"installed":true,"distance":120,"uptime":"00:01:00","nexthops":[{"ip":"192.168.50.2","fib":true}]} + ] + }` + + tr := tree.New() + New(tr, fakeQuerier{ipv4: before}, nil).writeRibs(context.Background()) + if got := len(ipv4Routes(t, tr)); got != 1 { + t.Fatalf("before: expected 1 route, got %d", got) + } + + // zebra now has only the static route; the OSPF route is gone with no + // delete event. A fresh snapshot must not carry the corpse forward. + New(tr, fakeQuerier{ipv4: after}, nil).writeRibs(context.Background()) + routes := ipv4Routes(t, tr) + if len(routes) != 1 { + t.Fatalf("after: expected 1 route, got %d", len(routes)) + } + if got := protocolName("static"); routes[0]["source-protocol"] != got { + t.Errorf("surviving route protocol = %v, want %v", routes[0]["source-protocol"], got) + } +} + +func TestTransformRouteFields(t *testing.T) { + entry := map[string]any{ + "prefix": "10.0.0.0/24", + "protocol": "ospf", + "selected": true, + "installed": true, + "distance": float64(110), + "metric": float64(20), + "uptime": "01:00:00", + "nexthops": []any{ + map[string]any{"ip": "192.168.1.1", "interfaceName": "e1", "fib": true}, }, } - result := transformRoute(route, false, time.Now()) - + now := time.Date(2026, 6, 10, 12, 0, 0, 0, time.UTC) var parsed map[string]any - if err := json.Unmarshal(result, &parsed); err != nil { + if err := json.Unmarshal(transformRoute("ipv4", "10.0.0.0/24", entry, now), &parsed); err != nil { t.Fatalf("unmarshal: %v", err) } - if got := parsed["ietf-ipv6-unicast-routing:destination-prefix"]; got != "2001:db8::/48" { - t.Errorf("destination-prefix = %v, want %q", got, "2001:db8::/48") + if parsed["ietf-ipv4-unicast-routing:destination-prefix"] != "10.0.0.0/24" { + t.Errorf("destination-prefix = %v", parsed["ietf-ipv4-unicast-routing:destination-prefix"]) } - if got := parsed["source-protocol"]; got != "ietf-ospf:ospfv2" { - t.Errorf("source-protocol = %v, want %q", got, "ietf-ospf:ospfv2") + if parsed["source-protocol"] != "ietf-ospf:ospfv2" { + t.Errorf("source-protocol = %v", parsed["source-protocol"]) + } + if toInt(parsed["route-preference"]) != 110 { + t.Errorf("route-preference = %v", parsed["route-preference"]) + } + if toInt(parsed["ietf-ospf:metric"]) != 20 { + t.Errorf("ietf-ospf:metric = %v", parsed["ietf-ospf:metric"]) + } + if _, ok := parsed["active"]; !ok { + t.Error("selected route must have active leaf") + } + // last-updated = now - 1h + if parsed["last-updated"] != "2026-06-10T11:00:00Z" { + t.Errorf("last-updated = %v, want 2026-06-10T11:00:00Z", parsed["last-updated"]) + } + + hops := parsed["next-hop"].(map[string]any)["next-hop-list"].(map[string]any)["next-hop"].([]any) + if len(hops) != 1 { + t.Fatalf("expected 1 nexthop, got %d", len(hops)) + } + hop := hops[0].(map[string]any) + if hop["ietf-ipv4-unicast-routing:address"] != "192.168.1.1" { + t.Errorf("nexthop address = %v", hop["ietf-ipv4-unicast-routing:address"]) + } + if _, ok := hop["infix-routing:installed"]; !ok { + t.Error("fib nexthop must have infix-routing:installed") } } -func TestAddRouteSamePrefixProtoOverwrites(t *testing.T) { +func TestTransformRouteBlackhole(t *testing.T) { + entry := map[string]any{ + "prefix": "10.1.0.0/24", + "protocol": "blackhole", + "distance": float64(0), + "uptime": "00:00:10", + } + var parsed map[string]any + if err := json.Unmarshal(transformRoute("ipv4", "10.1.0.0/24", entry, time.Now()), &parsed); err != nil { + t.Fatalf("unmarshal: %v", err) + } + nh := parsed["next-hop"].(map[string]any) + if nh["special-next-hop"] != "blackhole" { + t.Errorf("special-next-hop = %v, want blackhole", nh["special-next-hop"]) + } +} + +func TestWriteRibsQueryErrorKeepsData(t *testing.T) { tr := tree.New() - w := New(tr, nil) + // Seed with good data. + New(tr, fakeQuerier{ipv4: `{"10.0.0.0/24":[{"prefix":"10.0.0.0/24","protocol":"static","selected":true,"distance":1,"uptime":"00:00:05","nexthops":[{"ip":"10.0.0.1"}]}]}`}, nil). + writeRibs(context.Background()) + before := tr.Get(routingTreeKey) - gateway := net.ParseIP("192.168.50.2").To4() - prefix := net.IPNet{IP: net.IPv4(0, 0, 0, 0), Mask: net.CIDRMask(0, 32)} + // A failing query must not blank the table. + New(tr, fakeQuerier{err: context.DeadlineExceeded}, nil).writeRibs(context.Background()) + after := tr.Get(routingTreeKey) - w.addRoute(&zapi.Route{ - Type: zapi.RouteStatic, - Distance: 120, - Prefix: prefix, - Nexthops: []zapi.Nexthop{{Type: zapi.NHIPv4IFIndex, Gate: gateway}}, - }) - - newGateway := net.ParseIP("10.0.0.1").To4() - w.addRoute(&zapi.Route{ - Type: zapi.RouteStatic, - Distance: 1, - Prefix: prefix, - Nexthops: []zapi.Nexthop{{Type: zapi.NHIPv4IFIndex, Gate: newGateway}}, - }) - - w.mu.Lock() - count := len(w.routes) - w.mu.Unlock() - - if count != 1 { - t.Errorf("same prefix+proto should overwrite: got %d routes, want 1", count) + if string(before) != string(after) { + t.Error("query error overwrote existing rib data") } +} - data := tr.Get(routingTreeKey) - if data == nil { - t.Fatal("routing tree key not set") - } +func TestWriteRibsBothFamilies(t *testing.T) { + tr := tree.New() + w := New(tr, fakeQuerier{ + ipv4: `{"10.0.0.0/24":[{"prefix":"10.0.0.0/24","protocol":"static","selected":true,"distance":1,"uptime":"00:00:05","nexthops":[{"ip":"10.0.0.1"}]}]}`, + ipv6: `{"2001:db8::/64":[{"prefix":"2001:db8::/64","protocol":"connected","selected":true,"distance":0,"uptime":"00:00:05","nexthops":[{"interfaceName":"e1"}]}]}`, + }, nil) + w.writeRibs(context.Background()) var routing map[string]any - if err := json.Unmarshal(data, &routing); err != nil { + if err := json.Unmarshal(tr.Get(routingTreeKey), &routing); err != nil { t.Fatalf("unmarshal: %v", err) } - - ribs := routing["ribs"].(map[string]any) - ribList := ribs["rib"].([]any) - var ipv4Routes []any - for _, rib := range ribList { - ribMap := rib.(map[string]any) - if ribMap["name"] == "ipv4" { - ipv4Routes = ribMap["routes"].(map[string]any)["route"].([]any) - break - } - } - - if len(ipv4Routes) != 1 { - t.Fatalf("expected 1 route, got %d", len(ipv4Routes)) - } - - rm := ipv4Routes[0].(map[string]any) - if pref := int(rm["route-preference"].(float64)); pref != 1 { - t.Errorf("route-preference = %d, want 1 (latest add wins)", pref) - } -} - -func TestDeleteRouteWithoutNexthops(t *testing.T) { - tr := tree.New() - w := New(tr, nil) - - gateway := net.ParseIP("192.168.10.3").To4() - prefix := net.IPNet{IP: net.IPv4(0, 0, 0, 0), Mask: net.CIDRMask(0, 32)} - - w.addRoute(&zapi.Route{ - Type: zapi.RouteStatic, - Distance: 5, - Prefix: prefix, - Nexthops: []zapi.Nexthop{{Type: zapi.NHIPv4IFIndex, Gate: gateway}}, - }) - - w.mu.Lock() - count := len(w.routes) - w.mu.Unlock() - if count != 1 { - t.Fatalf("expected 1 route after add, got %d", count) - } - - w.deleteRoute(&zapi.Route{ - Type: zapi.RouteStatic, - Prefix: prefix, - }) - - w.mu.Lock() - count = len(w.routes) - w.mu.Unlock() - if count != 0 { - t.Errorf("expected 0 routes after delete without nexthops, got %d", count) + ribs := routing["ribs"].(map[string]any)["rib"].([]any) + if len(ribs) != 2 { + t.Fatalf("expected 2 ribs, got %d", len(ribs)) } }