From ba94be3c84e457456390c9a4fa8288e425e8d424 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mattias=20Walstr=C3=B6m?= Date: Wed, 20 May 2026 16:16:44 +0200 Subject: [PATCH] yanger: Remove code duplication and fix upgrade Remove a lot of duplication and move to seperate libs in internal/ Also, instead of polling upgrade status using src/rauc-installation-status, hook into the dbus monitor. --- src/yangerd/internal/backoff/backoff.go | 51 ++++ src/yangerd/internal/collector/boot.go | 22 -- src/yangerd/internal/collector/boot_test.go | 21 -- src/yangerd/internal/collector/live.go | 72 +++++ src/yangerd/internal/collector/live_test.go | 74 +++++ src/yangerd/internal/collector/runner.go | 49 +++- src/yangerd/internal/config/config.go | 8 - .../internal/dbusmonitor/dbusmonitor.go | 40 +-- .../internal/dbusmonitor/dbusmonitor_test.go | 19 +- .../internal/lldpmonitor/lldpmonitor.go | 34 +-- src/yangerd/internal/monitor/monitor.go | 3 +- src/yangerd/internal/sysreaders/sysreaders.go | 276 ++++++++++++++++++ src/yangerd/internal/tree/tree.go | 15 + .../internal/zapiwatcher/zapiwatcher.go | 23 +- 14 files changed, 574 insertions(+), 133 deletions(-) create mode 100644 src/yangerd/internal/backoff/backoff.go create mode 100644 src/yangerd/internal/sysreaders/sysreaders.go diff --git a/src/yangerd/internal/backoff/backoff.go b/src/yangerd/internal/backoff/backoff.go new file mode 100644 index 00000000..a12a2951 --- /dev/null +++ b/src/yangerd/internal/backoff/backoff.go @@ -0,0 +1,51 @@ +// Package backoff provides exponential backoff retry logic with +// context-aware sleep, shared across reactive monitors. +package backoff + +import ( + "context" + "math" + "time" +) + +// Backoff implements exponential backoff with a configurable initial +// delay, maximum delay, and growth factor. +type Backoff struct { + Initial time.Duration + Max time.Duration + Factor float64 +} + +// Default returns a Backoff with the standard yangerd parameters: +// 100ms initial, 30s max, factor 2. +func Default() *Backoff { + return &Backoff{ + Initial: 100 * time.Millisecond, + Max: 30 * time.Second, + Factor: 2.0, + } +} + +// Next returns the next delay value after current. If current is +// zero, Initial is returned. +func (b *Backoff) Next(current time.Duration) time.Duration { + if current <= 0 { + return b.Initial + } + next := time.Duration(math.Min(float64(current)*b.Factor, float64(b.Max))) + if next <= 0 { + return b.Initial + } + return next +} + +// Sleep waits for duration d or until ctx is cancelled, whichever +// comes first. Returns ctx.Err() if the context was cancelled. +func Sleep(ctx context.Context, d time.Duration) error { + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(d): + return nil + } +} diff --git a/src/yangerd/internal/collector/boot.go b/src/yangerd/internal/collector/boot.go index f7cf31b4..5c03c6ca 100644 --- a/src/yangerd/internal/collector/boot.go +++ b/src/yangerd/internal/collector/boot.go @@ -55,28 +55,6 @@ func BootSoftware(ctx context.Context, cmd CommandRunner) json.RawMessage { software["boot-order"] = bootOrder } - installer := make(map[string]interface{}) - instOut, err := cmd.Run(ctx, "rauc-installation-status") - if err == nil { - var instData map[string]interface{} - if json.Unmarshal(instOut, &instData) == nil { - if op, ok := instData["operation"]; ok && op != "" { - installer["operation"] = op - } - if prog, ok := instData["progress"].(map[string]interface{}); ok { - progress := make(map[string]interface{}) - if pct, ok := prog["percentage"]; ok { - progress["percentage"] = toInt(pct) - } - if msg, ok := prog["message"]; ok { - progress["message"] = msg - } - installer["progress"] = progress - } - } - } - software["installer"] = installer - result, _ := json.Marshal(map[string]interface{}{"infix-system:software": software}) return result } diff --git a/src/yangerd/internal/collector/boot_test.go b/src/yangerd/internal/collector/boot_test.go index 0ab2c3e5..9b53c0bf 100644 --- a/src/yangerd/internal/collector/boot_test.go +++ b/src/yangerd/internal/collector/boot_test.go @@ -75,14 +75,6 @@ HOME_URL="https://kernelkit.github.io" ] }` - testRaucInstallStatus = `{ - "operation": "idle", - "progress": { - "percentage": 100, - "message": "Installation complete" - } -}` - testBootOrder = "BOOT_ORDER=A B\n" ) @@ -134,7 +126,6 @@ func TestBootSoftware(t *testing.T) { runner := &testutil.MockRunner{ Results: map[string][]byte{ "rauc status --detailed --output-format=json": []byte(testRaucStatus), - "rauc-installation-status": []byte(testRaucInstallStatus), "fw_printenv BOOT_ORDER": []byte(testBootOrder), }, Errors: map[string]error{}, @@ -174,14 +165,6 @@ func TestBootSoftware(t *testing.T) { if !ok || len(slots) != 2 { t.Fatalf("expected 2 slots, got %v", sw["slot"]) } - - installer, ok := sw["installer"].(map[string]interface{}) - if !ok { - t.Fatal("missing installer") - } - if installer["operation"] != "idle" { - t.Fatalf("installer operation: expected 'idle', got %v", installer["operation"]) - } } func TestBootSoftwareAllCommandsFail(t *testing.T) { @@ -189,7 +172,6 @@ func TestBootSoftwareAllCommandsFail(t *testing.T) { Results: map[string][]byte{}, Errors: map[string]error{ "rauc status --detailed --output-format=json": fmt.Errorf("not found"), - "rauc-installation-status": fmt.Errorf("not found"), "fw_printenv BOOT_ORDER": fmt.Errorf("not found"), "grub-editenv /mnt/aux/grub/grubenv list": fmt.Errorf("not found"), }, @@ -206,9 +188,6 @@ func TestBootSoftwareAllCommandsFail(t *testing.T) { if _, ok := sw["boot-order"]; ok { t.Fatal("boot-order should not be present when commands fail") } - if _, ok := sw["installer"]; !ok { - t.Fatal("installer key should always be present") - } } func TestReadBootOrderFwPrintenv(t *testing.T) { diff --git a/src/yangerd/internal/collector/live.go b/src/yangerd/internal/collector/live.go index c49373d3..764fcf59 100644 --- a/src/yangerd/internal/collector/live.go +++ b/src/yangerd/internal/collector/live.go @@ -11,6 +11,9 @@ import ( // LiveSystemState computes the on-demand portion of ietf-system:system-state. // It reads uptime, current time, memory, load average from procfs and // filesystem usage via statfs — all computed fresh on each call. +// +// Installer status is handled separately via MergeInstaller to avoid +// shallow-merge clobbering the boot-time software data. func LiveSystemState(fs FileReader) json.RawMessage { state := make(map[string]interface{}) @@ -39,6 +42,75 @@ func LiveSystemState(fs FileReader) json.RawMessage { return data } +// MergeInstaller reads the cached software data from the tree and +// overlays the live installer status into it, returning the merged +// infix-system:software object as a top-level system-state fragment. +func MergeInstaller(cached json.RawMessage, inst InstallerStatus) json.RawMessage { + if inst == nil { + return nil + } + installer := liveInstaller(inst) + if len(installer) == 0 { + return nil + } + + var base map[string]json.RawMessage + if len(cached) > 0 { + json.Unmarshal(cached, &base) + } + if base == nil { + base = make(map[string]json.RawMessage) + } + + sw := make(map[string]interface{}) + if raw, ok := base["infix-system:software"]; ok { + json.Unmarshal(raw, &sw) + } + if sw == nil { + sw = make(map[string]interface{}) + } + sw["installer"] = installer + + swJSON, err := json.Marshal(sw) + if err != nil { + return nil + } + + result := map[string]json.RawMessage{ + "infix-system:software": swJSON, + } + out, err := json.Marshal(result) + if err != nil { + return nil + } + return out +} + +func liveInstaller(inst InstallerStatus) map[string]interface{} { + op, lastErr, pct, msg, err := inst.GetInstallStatus() + if err != nil { + return nil + } + installer := make(map[string]interface{}) + if op != "" { + installer["operation"] = op + } + if lastErr != "" { + installer["last-error"] = lastErr + } + if pct > 0 || msg != "" { + progress := make(map[string]interface{}) + if pct > 0 { + progress["percentage"] = pct + } + if msg != "" { + progress["message"] = msg + } + installer["progress"] = progress + } + return installer +} + func liveClock(fs FileReader) map[string]interface{} { data, err := fs.ReadFile("/proc/uptime") if err != nil { diff --git a/src/yangerd/internal/collector/live_test.go b/src/yangerd/internal/collector/live_test.go index 38ac43c7..5daa0b06 100644 --- a/src/yangerd/internal/collector/live_test.go +++ b/src/yangerd/internal/collector/live_test.go @@ -188,3 +188,77 @@ func TestLiveSystemStatePartialFailure(t *testing.T) { t.Fatal("load-average should be present") } } + +type mockInstaller struct { + op, lastErr, msg string + pct int + err error +} + +func (m *mockInstaller) GetInstallStatus() (string, string, int, string, error) { + return m.op, m.lastErr, m.pct, m.msg, m.err +} + +func TestMergeInstaller(t *testing.T) { + cached := json.RawMessage(`{"infix-system:software":{"compatible":"infix-x86_64","booted":{"slot":"rootfs.0"}}}`) + inst := &mockInstaller{op: "installing", pct: 45, msg: "Writing rootfs"} + + raw := MergeInstaller(cached, inst) + if raw == nil { + t.Fatal("expected non-nil") + } + + var result map[string]json.RawMessage + json.Unmarshal(raw, &result) + + var sw map[string]interface{} + json.Unmarshal(result["infix-system:software"], &sw) + if sw == nil { + t.Fatal("missing infix-system:software") + } + if sw["compatible"] != "infix-x86_64" { + t.Fatalf("cached 'compatible' was lost: %v", sw["compatible"]) + } + if sw["booted"] == nil { + t.Fatal("cached 'booted' was lost") + } + installer, ok := sw["installer"].(map[string]interface{}) + if !ok { + t.Fatal("missing installer") + } + if installer["operation"] != "installing" { + t.Fatalf("operation = %v, want 'installing'", installer["operation"]) + } + progress, ok := installer["progress"].(map[string]interface{}) + if !ok { + t.Fatal("missing progress") + } + if toInt(progress["percentage"]) != 45 { + t.Fatalf("percentage = %v, want 45", progress["percentage"]) + } + if progress["message"] != "Writing rootfs" { + t.Fatalf("message = %v", progress["message"]) + } +} + +func TestMergeInstallerNilCached(t *testing.T) { + inst := &mockInstaller{op: "idle"} + raw := MergeInstaller(nil, inst) + if raw == nil { + t.Fatal("expected non-nil even with nil cached") + } + var result map[string]json.RawMessage + json.Unmarshal(raw, &result) + var sw map[string]interface{} + json.Unmarshal(result["infix-system:software"], &sw) + if sw["installer"] == nil { + t.Fatal("missing installer") + } +} + +func TestMergeInstallerNilInst(t *testing.T) { + cached := json.RawMessage(`{"infix-system:software":{"compatible":"infix-x86_64"}}`) + if raw := MergeInstaller(cached, nil); raw != nil { + t.Fatalf("expected nil with nil installer, got %s", raw) + } +} diff --git a/src/yangerd/internal/collector/runner.go b/src/yangerd/internal/collector/runner.go index 94e4da71..6776b4e7 100644 --- a/src/yangerd/internal/collector/runner.go +++ b/src/yangerd/internal/collector/runner.go @@ -5,6 +5,8 @@ import ( "os" "os/exec" "path/filepath" + + "github.com/godbus/dbus/v5" ) // CommandRunner executes external commands and returns their stdout. @@ -18,10 +20,14 @@ type FileReader interface { Glob(pattern string) ([]string, error) } +// InstallerStatus queries RAUC installation progress. +type InstallerStatus interface { + GetInstallStatus() (operation string, lastError string, percentage int, message string, err error) +} + // ExecRunner is the production CommandRunner using os/exec. type ExecRunner struct{} -// Run executes name with args and returns combined stdout. func (ExecRunner) Run(ctx context.Context, name string, args ...string) ([]byte, error) { return exec.CommandContext(ctx, name, args...).Output() } @@ -29,12 +35,49 @@ func (ExecRunner) Run(ctx context.Context, name string, args ...string) ([]byte, // OSFileReader is the production FileReader using the os package. type OSFileReader struct{} -// ReadFile reads the named file. func (OSFileReader) ReadFile(path string) ([]byte, error) { return os.ReadFile(path) } -// Glob returns filenames matching the pattern. func (OSFileReader) Glob(pattern string) ([]string, error) { return filepath.Glob(pattern) } + +// DBusInstaller reads RAUC installation status from D-Bus properties. +type DBusInstaller struct{} + +func (DBusInstaller) GetInstallStatus() (string, string, int, string, error) { + conn, err := dbus.ConnectSystemBus() + if err != nil { + return "", "", 0, "", err + } + defer conn.Close() + + obj := conn.Object("de.pengutronix.rauc", "/") + + operation, _ := obj.GetProperty("de.pengutronix.rauc.Installer.Operation") + lastError, _ := obj.GetProperty("de.pengutronix.rauc.Installer.LastError") + + var pct int + var msg string + progress, err := obj.GetProperty("de.pengutronix.rauc.Installer.Progress") + if err == nil { + if vals, ok := progress.Value().([]interface{}); ok && len(vals) >= 2 { + if p, ok := vals[0].(int32); ok { + pct = int(p) + } + if s, ok := vals[1].(string); ok { + msg = s + } + } + } + + return variantString(operation), variantString(lastError), pct, msg, nil +} + +func variantString(v dbus.Variant) string { + if s, ok := v.Value().(string); ok { + return s + } + return "" +} diff --git a/src/yangerd/internal/config/config.go b/src/yangerd/internal/config/config.go index 4926698e..2f7b79f4 100644 --- a/src/yangerd/internal/config/config.go +++ b/src/yangerd/internal/config/config.go @@ -11,14 +11,10 @@ import ( type Config struct { Socket string LogLevel string - ZebraSocket string - LLDPCommand string - StartupTimeout time.Duration PollSystem time.Duration PollRouting time.Duration PollNTP time.Duration PollHardware time.Duration - PollContainers time.Duration EnableWifi bool EnableLLDP bool EnableFirewall bool @@ -32,14 +28,10 @@ func Load() *Config { return &Config{ Socket: envStr("YANGERD_SOCKET", "/run/yangerd.sock"), LogLevel: envStr("YANGERD_LOG_LEVEL", "info"), - ZebraSocket: envStr("YANGERD_ZEBRA_SOCKET", "/var/run/frr/zserv.api"), - LLDPCommand: envStr("YANGERD_LLDP_COMMAND", "lldpcli"), - StartupTimeout: envDur("YANGERD_STARTUP_TIMEOUT", 5*time.Second), PollSystem: envDur("YANGERD_POLL_INTERVAL_SYSTEM", 60*time.Second), 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), - PollContainers: envDur("YANGERD_POLL_INTERVAL_CONTAINERS", 10*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/dbusmonitor/dbusmonitor.go b/src/yangerd/internal/dbusmonitor/dbusmonitor.go index 99b52de4..b66c3b42 100644 --- a/src/yangerd/internal/dbusmonitor/dbusmonitor.go +++ b/src/yangerd/internal/dbusmonitor/dbusmonitor.go @@ -7,13 +7,13 @@ import ( "encoding/json" "fmt" "log/slog" - "math" "os" "strconv" "strings" "time" "github.com/godbus/dbus/v5" + "github.com/kernelkit/infix/src/yangerd/internal/backoff" "github.com/kernelkit/infix/src/yangerd/internal/tree" ) @@ -33,10 +33,6 @@ const ( dhcpTreeKey = "infix-dhcp-server:dhcp-server" firewallTreeKey = "infix-firewall:firewall" - - reconnectInitial = 100 * time.Millisecond - reconnectMax = 30 * time.Second - reconnectFactor = 2.0 ) // DBusMonitor subscribes to dnsmasq and firewalld D-Bus signals and @@ -55,7 +51,8 @@ func New(t *tree.Tree, log *slog.Logger) *DBusMonitor { // to relevant signals, loads initial DHCP/firewall data, and reconnects // with exponential backoff on failures until ctx is cancelled. func (m *DBusMonitor) Run(ctx context.Context) error { - delay := reconnectInitial + bo := backoff.Default() + delay := bo.Initial for { if err := ctx.Err(); err != nil { @@ -65,24 +62,24 @@ func (m *DBusMonitor) Run(ctx context.Context) error { conn, err := dbus.ConnectSystemBus() if err != nil { m.log.Warn("dbus monitor: connect system bus failed", "err", err, "delay", delay) - if err := sleepOrDone(ctx, delay); err != nil { + if err := backoff.Sleep(ctx, delay); err != nil { return err } - delay = nextDelay(delay) + delay = bo.Next(delay) continue } if err := m.subscribe(conn); err != nil { m.log.Warn("dbus monitor: subscribe failed", "err", err, "delay", delay) _ = conn.Close() - if err := sleepOrDone(ctx, delay); err != nil { + if err := backoff.Sleep(ctx, delay); err != nil { return err } - delay = nextDelay(delay) + delay = bo.Next(delay) continue } - delay = reconnectInitial + delay = bo.Initial if err := m.refreshDHCP(conn); err != nil { m.log.Warn("dbus monitor: initial dhcp refresh failed", "err", err) @@ -98,10 +95,10 @@ func (m *DBusMonitor) Run(ctx context.Context) error { } m.log.Warn("dbus monitor: signal loop ended, reconnecting", "err", err, "delay", delay) - if err := sleepOrDone(ctx, delay); err != nil { + if err := backoff.Sleep(ctx, delay); err != nil { return err } - delay = nextDelay(delay) + delay = bo.Next(delay) } } @@ -1049,20 +1046,3 @@ func toUint64(v any) uint64 { } return 0 } - -func sleepOrDone(ctx context.Context, d time.Duration) error { - select { - case <-ctx.Done(): - return ctx.Err() - case <-time.After(d): - return nil - } -} - -func nextDelay(delay time.Duration) time.Duration { - next := time.Duration(math.Min(float64(delay)*reconnectFactor, float64(reconnectMax))) - if next <= 0 { - return reconnectInitial - } - return next -} diff --git a/src/yangerd/internal/dbusmonitor/dbusmonitor_test.go b/src/yangerd/internal/dbusmonitor/dbusmonitor_test.go index b11e6dde..2bddf3ea 100644 --- a/src/yangerd/internal/dbusmonitor/dbusmonitor_test.go +++ b/src/yangerd/internal/dbusmonitor/dbusmonitor_test.go @@ -7,6 +7,8 @@ import ( "reflect" "testing" "time" + + "github.com/kernelkit/infix/src/yangerd/internal/backoff" ) func TestParseDnsmasqLeases(t *testing.T) { @@ -586,7 +588,7 @@ func TestSleepOrDone(t *testing.T) { defer cancel() } - err := sleepOrDone(ctx, tc.delay) + err := backoff.Sleep(ctx, tc.delay) if tc.wantErr { if !errors.Is(err, context.Canceled) { t.Fatalf("expected context.Canceled, got %v", err) @@ -668,22 +670,23 @@ func TestDecodeActiveZones(t *testing.T) { } func TestNextDelay(t *testing.T) { + b := backoff.Default() tests := []struct { name string in time.Duration want time.Duration }{ - {name: "doubles normal delay", in: reconnectInitial, want: reconnectInitial * 2}, - {name: "caps at reconnectMax", in: reconnectMax, want: reconnectMax}, - {name: "near max also caps", in: reconnectMax - time.Second, want: reconnectMax}, - {name: "zero becomes reconnectInitial", in: 0, want: reconnectInitial}, - {name: "negative becomes reconnectInitial", in: -time.Second, want: reconnectInitial}, + {name: "doubles normal delay", in: b.Initial, want: b.Initial * 2}, + {name: "caps at max", in: b.Max, want: b.Max}, + {name: "near max also caps", in: b.Max - time.Second, want: b.Max}, + {name: "zero becomes initial", in: 0, want: b.Initial}, + {name: "negative becomes initial", in: -time.Second, want: b.Initial}, } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { - if got := nextDelay(tc.in); got != tc.want { - t.Fatalf("nextDelay(%v) = %v, want %v", tc.in, got, tc.want) + if got := b.Next(tc.in); got != tc.want { + t.Fatalf("Next(%v) = %v, want %v", tc.in, got, tc.want) } }) } diff --git a/src/yangerd/internal/lldpmonitor/lldpmonitor.go b/src/yangerd/internal/lldpmonitor/lldpmonitor.go index b0b77211..430ca9ad 100644 --- a/src/yangerd/internal/lldpmonitor/lldpmonitor.go +++ b/src/yangerd/internal/lldpmonitor/lldpmonitor.go @@ -11,21 +11,16 @@ import ( "encoding/json" "fmt" "log/slog" - "math" "os/exec" "regexp" "strconv" "strings" - "time" + "github.com/kernelkit/infix/src/yangerd/internal/backoff" "github.com/kernelkit/infix/src/yangerd/internal/tree" ) const ( - reconnectInitial = 100 * time.Millisecond - reconnectMax = 30 * time.Second - reconnectFactor = 2.0 - lldpMulticastMAC = "01:80:C2:00:00:0E" treeKey = "ieee802-dot1ab-lldp:lldp" ) @@ -44,7 +39,8 @@ func New(t *tree.Tree, log *slog.Logger) *LLDPMonitor { // Run starts the LLDP monitor. It blocks until ctx is cancelled. func (m *LLDPMonitor) Run(ctx context.Context) error { - delay := reconnectInitial + bo := backoff.Default() + delay := bo.Initial for { err := m.runOnce(ctx) @@ -54,14 +50,10 @@ func (m *LLDPMonitor) Run(ctx context.Context) error { m.log.Warn("lldp monitor: subprocess exited, restarting", "err", err, "delay", delay) - select { - case <-ctx.Done(): - return ctx.Err() - case <-time.After(delay): + if err := backoff.Sleep(ctx, delay); err != nil { + return err } - delay = time.Duration(math.Min( - float64(delay)*reconnectFactor, - float64(reconnectMax))) + delay = bo.Next(delay) } } @@ -262,15 +254,7 @@ func transformLLDPEvent(data []byte) json.RawMessage { return json.RawMessage(out) } -var chassisSubtypeMap = map[string]string{ - "ifalias": "interface-alias", - "mac": "mac-address", - "ip": "network-address", - "ifname": "interface-name", - "local": "local", -} - -var portSubtypeMap = map[string]string{ +var idSubtypeMap = map[string]string{ "ifalias": "interface-alias", "mac": "mac-address", "ip": "network-address", @@ -279,14 +263,14 @@ var portSubtypeMap = map[string]string{ } func chassisIDSubtype(t string) string { - if v, ok := chassisSubtypeMap[t]; ok { + if v, ok := idSubtypeMap[t]; ok { return v } return "unknown" } func portIDSubtype(t string) string { - if v, ok := portSubtypeMap[t]; ok { + if v, ok := idSubtypeMap[t]; ok { return v } return "unknown" diff --git a/src/yangerd/internal/monitor/monitor.go b/src/yangerd/internal/monitor/monitor.go index 4e80abf6..fb8cec1c 100644 --- a/src/yangerd/internal/monitor/monitor.go +++ b/src/yangerd/internal/monitor/monitor.go @@ -409,7 +409,8 @@ func (m *NLMonitor) refreshInterface(name string) { func (m *NLMonitor) rebuild() { m.mu.Lock() linksCopy := append(json.RawMessage{}, m.links...) - doc := iface.Transform(m.links, m.addrs, m.links, m.fc) + addrsCopy := append(json.RawMessage{}, m.addrs...) + doc := iface.Transform(linksCopy, addrsCopy, linksCopy, m.fc) eth := copyStringMap(m.ethernet) wfi := copyStringMap(m.wifi) fdb := copyStringMap(m.fdb) diff --git a/src/yangerd/internal/sysreaders/sysreaders.go b/src/yangerd/internal/sysreaders/sysreaders.go new file mode 100644 index 00000000..b103d979 --- /dev/null +++ b/src/yangerd/internal/sysreaders/sysreaders.go @@ -0,0 +1,276 @@ +package sysreaders + +import ( + "bufio" + "bytes" + "encoding/json" + "fmt" + "os" + "path/filepath" + "regexp" + "strconv" + "strings" + "sync" +) + +var gmtOffsetRe = regexp.MustCompile(`Etc/GMT([+-]\d{1,2})$`) + +var zonePrefixes = []string{ + "/usr/share/zoneinfo/posix/", + "/usr/share/zoneinfo/right/", + "/usr/share/zoneinfo/", +} + +var userShellMap = map[string]string{ + "/bin/bash": "infix-system:bash", + "/bin/sh": "infix-system:sh", + "/usr/bin/clish": "infix-system:clish", + "/bin/false": "infix-system:false", + "/sbin/nologin": "infix-system:false", + "/usr/sbin/nologin": "infix-system:false", +} + +const SSHDKeysDir = "/var/run/sshd" + +func ReadHostname(path string) (json.RawMessage, error) { + data, err := os.ReadFile(path) + if err != nil { + return nil, err + } + name := strings.TrimSpace(string(data)) + return json.Marshal(map[string]string{"hostname": name}) +} + +func ReadTimezone(path string) (json.RawMessage, error) { + target, err := filepath.EvalSymlinks(path) + if err != nil { + return nil, err + } + + var tz string + for _, p := range zonePrefixes { + if strings.HasPrefix(target, p) { + tz = target[len(p):] + break + } + } + if tz == "" { + return nil, fmt.Errorf("unrecognized zoneinfo path: %s", target) + } + + clock := make(map[string]interface{}) + if m := gmtOffsetRe.FindStringSubmatch(tz); m != nil { + offset, _ := strconv.Atoi(m[1]) + clock["timezone-utc-offset"] = -offset + } else if tz == "Etc/UTC" { + clock["timezone-utc-offset"] = 0 + } else { + clock["timezone-name"] = tz + } + + return json.Marshal(map[string]interface{}{"clock": clock}) +} + +func ReadUsers(_ string) (json.RawMessage, error) { + passwdData, err := os.ReadFile("/etc/passwd") + if err != nil { + return nil, err + } + + passwdUsers := make(map[string]string) + scanner := bufio.NewScanner(bytes.NewReader(passwdData)) + for scanner.Scan() { + parts := strings.Split(scanner.Text(), ":") + if len(parts) < 7 { + continue + } + uid, err := strconv.Atoi(parts[2]) + if err != nil || uid < 1000 || uid >= 10000 { + continue + } + shell := strings.TrimSpace(parts[6]) + mapped, ok := userShellMap[shell] + if !ok { + mapped = "infix-system:false" + } + passwdUsers[parts[0]] = mapped + } + + shadowHashes := make(map[string]string) + shadowData, err := os.ReadFile("/etc/shadow") + if err == nil { + scanner = bufio.NewScanner(bytes.NewReader(shadowData)) + for scanner.Scan() { + parts := strings.SplitN(scanner.Text(), ":", 3) + if len(parts) < 2 { + continue + } + hash := parts[1] + if hash == "" || strings.HasPrefix(hash, "*") || strings.HasPrefix(hash, "!") { + continue + } + shadowHashes[parts[0]] = hash + } + } + + users := make([]interface{}, 0) + for username, shell := range passwdUsers { + user := map[string]interface{}{ + "name": username, + "infix-system:shell": shell, + } + if hash, ok := shadowHashes[username]; ok { + user["password"] = hash + } + + keysData, err := os.ReadFile(filepath.Join(SSHDKeysDir, username+".keys")) + if err == nil { + var authKeys []interface{} + for _, line := range strings.Split(string(keysData), "\n") { + line = strings.TrimSpace(line) + if line == "" || strings.HasPrefix(line, "#") { + continue + } + parts := strings.SplitN(line, " ", 3) + if len(parts) < 2 { + continue + } + keyName := fmt.Sprintf("%s-key-%d", username, len(authKeys)) + if len(parts) > 2 { + keyName = parts[2] + } + authKeys = append(authKeys, map[string]interface{}{ + "name": keyName, + "algorithm": parts[0], + "key-data": parts[1], + }) + } + if len(authKeys) > 0 { + user["authorized-key"] = authKeys + } + } + users = append(users, user) + } + + return json.Marshal(map[string]interface{}{ + "authentication": map[string]interface{}{ + "user": users, + }, + }) +} + +func ReadDNSResolver(_ string) (json.RawMessage, error) { + servers := make([]interface{}, 0) + var search []string + options := make(map[string]interface{}) + seen := make(map[string]bool) + + for _, path := range []string{"/etc/resolv.conf.head", "/var/lib/misc/resolv.conf"} { + data, err := os.ReadFile(path) + if err != nil { + continue + } + ParseResolvConf(string(data), &servers, &search, options, seen) + } + + dns := make(map[string]interface{}) + dns["server"] = servers + if len(search) > 0 { + dns["search"] = search + } + if len(options) > 0 { + dns["options"] = options + } + + return json.Marshal(map[string]interface{}{"infix-system:dns-resolver": dns}) +} + +func ParseResolvConf(data string, servers *[]interface{}, search *[]string, options map[string]interface{}, seen map[string]bool) { + for _, line := range strings.Split(data, "\n") { + line = strings.TrimSpace(line) + switch { + case strings.HasPrefix(line, "nameserver"): + ip := strings.TrimSpace(strings.TrimPrefix(line, "nameserver")) + if ip != "" && ip != "127.0.0.1" && ip != "::1" && !seen[ip] { + seen[ip] = true + *servers = append(*servers, map[string]interface{}{ + "address": ip, + }) + } + case strings.HasPrefix(line, "search"): + *search = append(*search, strings.Fields(line)[1:]...) + case strings.HasPrefix(line, "options"): + for _, opt := range strings.Fields(line)[1:] { + if strings.HasPrefix(opt, "timeout:") { + if v, err := strconv.Atoi(strings.TrimPrefix(opt, "timeout:")); err == nil { + options["timeout"] = v + } + } else if strings.HasPrefix(opt, "attempts:") { + if v, err := strconv.Atoi(strings.TrimPrefix(opt, "attempts:")); err == nil { + options["attempts"] = v + } + } + } + } + } +} + +// ForwardingAggregator tracks all /proc/sys/net/ipv{4,6}/conf/*/forwarding +// files and rebuilds the complete interfaces list on every change. +type ForwardingAggregator struct { + mu sync.Mutex +} + +func NewForwardingAggregator() *ForwardingAggregator { + return &ForwardingAggregator{} +} + +func (fa *ForwardingAggregator) HandleForwardingChange(_ string) (json.RawMessage, error) { + fa.mu.Lock() + defer fa.mu.Unlock() + + enabled := make(map[string]bool) + + for _, family := range []string{"ipv4", "ipv6"} { + sysctl := "forwarding" + if family == "ipv6" { + sysctl = "force_forwarding" + } + pattern := fmt.Sprintf("/proc/sys/net/%s/conf/*/%s", family, sysctl) + matches, err := filepath.Glob(pattern) + if err != nil { + continue + } + for _, path := range matches { + b, err := os.ReadFile(path) + if err != nil { + continue + } + if strings.TrimSpace(string(b)) != "1" { + continue + } + parts := strings.Split(filepath.Clean(path), string(os.PathSeparator)) + if len(parts) >= 7 { + ifname := parts[len(parts)-2] + if ifname != "all" && ifname != "default" && ifname != "lo" { + enabled[ifname] = true + } + } + } + } + + ifnames := make([]string, 0) + for name := range enabled { + ifnames = append(ifnames, name) + } + + data, err := json.Marshal(map[string]interface{}{ + "interfaces": map[string]interface{}{ + "interface": ifnames, + }, + }) + if err != nil { + return nil, err + } + return json.RawMessage(data), nil +} diff --git a/src/yangerd/internal/tree/tree.go b/src/yangerd/internal/tree/tree.go index 75ea450b..5873af87 100644 --- a/src/yangerd/internal/tree/tree.go +++ b/src/yangerd/internal/tree/tree.go @@ -226,6 +226,21 @@ func shallowMerge(base, overlay json.RawMessage) json.RawMessage { return merged } +// GetCached returns the raw cached JSON for the given module key +// WITHOUT invoking any registered provider. This is safe to call +// from inside a provider closure (no recursion risk). +func (t *Tree) GetCached(key string) json.RawMessage { + t.mu.RLock() + entry, ok := t.models[key] + t.mu.RUnlock() + if !ok { + return nil + } + entry.mu.RLock() + defer entry.mu.RUnlock() + return entry.data +} + // Info returns metadata for the given module key. func (t *Tree) Info(key string) (ModelInfo, bool) { t.mu.RLock() diff --git a/src/yangerd/internal/zapiwatcher/zapiwatcher.go b/src/yangerd/internal/zapiwatcher/zapiwatcher.go index 851d8135..69148309 100644 --- a/src/yangerd/internal/zapiwatcher/zapiwatcher.go +++ b/src/yangerd/internal/zapiwatcher/zapiwatcher.go @@ -5,24 +5,18 @@ import ( "encoding/json" "fmt" "log/slog" - "math" "net" "sort" "strconv" "sync" "time" + "github.com/kernelkit/infix/src/yangerd/internal/backoff" "github.com/kernelkit/infix/src/yangerd/internal/tree" "github.com/kernelkit/infix/src/yangerd/internal/zapi" ) -const ( - zapiSocketPath = "/var/run/frr/zserv.api" - - reconnectInitial = 100 * time.Millisecond - reconnectMax = 30 * time.Second - reconnectFactor = 2.0 -) +const zapiSocketPath = "/var/run/frr/zserv.api" var subscribeTypes = []zapi.RouteType{ zapi.RouteKernel, @@ -66,7 +60,8 @@ func New(t *tree.Tree, log *slog.Logger) *ZAPIWatcher { } func (w *ZAPIWatcher) Run(ctx context.Context) error { - delay := reconnectInitial + bo := backoff.Default() + delay := bo.Initial for { conn, err := w.connect(ctx) @@ -76,16 +71,14 @@ func (w *ZAPIWatcher) Run(ctx context.Context) error { } w.log.Warn("zapi watcher: connect failed", "err", err, "delay", delay) - select { - case <-ctx.Done(): - return ctx.Err() - case <-time.After(delay): + if err := backoff.Sleep(ctx, delay); err != nil { + return err } - delay = time.Duration(math.Min(float64(delay)*reconnectFactor, float64(reconnectMax))) + delay = bo.Next(delay) continue } - delay = reconnectInitial + delay = bo.Initial w.log.Info("zapi watcher: connected", "socket", zapiSocketPath) err = w.processMessages(ctx, conn)