diff --git a/src/yangerd/internal/bridgebatch/bridgebatch.go b/src/yangerd/internal/bridgebatch/bridgebatch.go index 0ab85fb5..2d5795f2 100644 --- a/src/yangerd/internal/bridgebatch/bridgebatch.go +++ b/src/yangerd/internal/bridgebatch/bridgebatch.go @@ -2,6 +2,10 @@ // subprocess for querying bridge FDB, VLAN, MDB, and STP state. // Identical design to ipbatch: mutex-serialized queries, dead/alive // state management, and exponential backoff restart. +// +// Like ipbatch, `bridge -batch -` produces NO stdout for commands +// that fail. Query uses a read timeout to detect this and kills +// the subprocess so restartLoop can recover. package bridgebatch import ( @@ -25,6 +29,7 @@ var ErrBatchDead = errors.New("bridge batch process is dead") const ( canaryCommand = "vlan show" + queryTimeout = 5 * time.Second reconnectInitial = 100 * time.Millisecond reconnectMax = 30 * time.Second reconnectFactor = 2.0 @@ -34,7 +39,7 @@ const ( type BridgeBatch struct { cmd *exec.Cmd stdin io.WriteCloser - stdout *bufio.Scanner + lines chan []byte stderr io.ReadCloser mu sync.Mutex alive atomic.Bool @@ -79,15 +84,26 @@ func (b *BridgeBatch) start() error { b.mu.Lock() b.cmd = cmd b.stdin = stdin - b.stdout = bufio.NewScanner(stdout) - b.stdout.Buffer(make([]byte, 0, 4*1024*1024), 4*1024*1024) + b.lines = make(chan []byte, 8) b.stderr = stderr b.alive.Store(true) b.mu.Unlock() + go b.readLines(stdout) go b.drainStderr() return nil } +func (b *BridgeBatch) readLines(r io.Reader) { + scanner := bufio.NewScanner(r) + scanner.Buffer(make([]byte, 0, 4*1024*1024), 4*1024*1024) + for scanner.Scan() { + line := make([]byte, len(scanner.Bytes())) + copy(line, scanner.Bytes()) + b.lines <- line + } + b.alive.Store(false) +} + // Query sends a command to the bridge batch process and returns the // JSON response. func (b *BridgeBatch) Query(command string) (json.RawMessage, error) { @@ -105,16 +121,22 @@ func (b *BridgeBatch) Query(command string) (json.RawMessage, error) { b.alive.Store(false) return nil, fmt.Errorf("write command: %w", err) } - if !b.stdout.Scan() { - b.alive.Store(false) - if err := b.stdout.Err(); err != nil { - return nil, fmt.Errorf("read response: %w", err) + + select { + case line, ok := <-b.lines: + if !ok { + b.alive.Store(false) + return nil, fmt.Errorf("bridge batch process exited") } - return nil, fmt.Errorf("bridge batch process exited") + return json.RawMessage(line), nil + case <-time.After(queryTimeout): + b.log.Warn("bridge batch query timeout, killing subprocess", "cmd", command) + b.alive.Store(false) + if b.cmd != nil && b.cmd.Process != nil { + b.cmd.Process.Kill() + } + return nil, fmt.Errorf("timeout waiting for response to: %s", command) } - raw := make([]byte, len(b.stdout.Bytes())) - copy(raw, b.stdout.Bytes()) - return json.RawMessage(raw), nil } // Close terminates the subprocess and cancels the restart loop. diff --git a/src/yangerd/internal/ipbatch/ipbatch.go b/src/yangerd/internal/ipbatch/ipbatch.go index dcb06e07..eb36635e 100644 --- a/src/yangerd/internal/ipbatch/ipbatch.go +++ b/src/yangerd/internal/ipbatch/ipbatch.go @@ -9,6 +9,10 @@ // Query. Address queries must therefore use a separate IPBatch instance // that omits -s (use WithDetails only). // +// IMPORTANT: `ip -force -batch -` produces NO stdout for commands that +// fail (e.g. "link show dev "). Query uses a read timeout +// to detect this and kills the subprocess so restartLoop can recover. +// // On subprocess death the manager enters a dead state and attempts // automatic restart with exponential backoff. package ipbatch @@ -35,6 +39,7 @@ var ErrBatchDead = errors.New("ip batch process is dead") const ( canaryCommand = "link show lo" + queryTimeout = 5 * time.Second reconnectInitial = 100 * time.Millisecond reconnectMax = 30 * time.Second reconnectFactor = 2.0 @@ -53,7 +58,7 @@ func WithDetails() Option { return func(b *IPBatch) { b.details = true } } type IPBatch struct { cmd *exec.Cmd stdin io.WriteCloser - stdout *bufio.Scanner + lines chan []byte stderr io.ReadCloser mu sync.Mutex // serializes queries alive atomic.Bool @@ -114,18 +119,31 @@ func (b *IPBatch) start() error { b.mu.Lock() b.cmd = cmd b.stdin = stdin - b.stdout = bufio.NewScanner(stdout) - b.stdout.Buffer(make([]byte, 0, 4*1024*1024), 4*1024*1024) // 4 MiB max line + b.lines = make(chan []byte, 8) b.stderr = stderr b.alive.Store(true) b.mu.Unlock() + go b.readLines(stdout) go b.drainStderr() return nil } +func (b *IPBatch) readLines(r io.Reader) { + scanner := bufio.NewScanner(r) + scanner.Buffer(make([]byte, 0, 4*1024*1024), 4*1024*1024) + for scanner.Scan() { + line := make([]byte, len(scanner.Bytes())) + copy(line, scanner.Bytes()) + b.lines <- line + } + b.alive.Store(false) +} + // Query sends a command to the ip batch process and returns the JSON // response. Commands are newline-terminated (e.g. "link show dev eth0"). -// Each command produces exactly one line of JSON array output. +// Each command produces exactly one line of JSON array output. If the +// subprocess produces no output (e.g. querying a non-existent device), +// Query times out and kills the subprocess for recovery. func (b *IPBatch) Query(command string) (json.RawMessage, error) { if !b.alive.Load() { return nil, ErrBatchDead @@ -141,19 +159,23 @@ func (b *IPBatch) Query(command string) (json.RawMessage, error) { b.alive.Store(false) return nil, fmt.Errorf("write command: %w", err) } - if !b.stdout.Scan() { - b.alive.Store(false) - if err := b.stdout.Err(); err != nil { - return nil, fmt.Errorf("read response: %w", err) + + select { + case line, ok := <-b.lines: + if !ok { + b.alive.Store(false) + return nil, fmt.Errorf("ip batch process exited") } - return nil, fmt.Errorf("ip batch process exited") + b.log.Debug("ipbatch query", "cmd", command, "respLen", len(line)) + return json.RawMessage(line), nil + case <-time.After(queryTimeout): + b.log.Warn("ip batch query timeout, killing subprocess", "cmd", command) + b.alive.Store(false) + if b.cmd != nil && b.cmd.Process != nil { + b.cmd.Process.Kill() + } + return nil, fmt.Errorf("timeout waiting for response to: %s", command) } - raw := make([]byte, len(b.stdout.Bytes())) - copy(raw, b.stdout.Bytes()) - - b.log.Debug("ipbatch query", "cmd", command, "respLen", len(raw)) - - return json.RawMessage(raw), nil } // Close terminates the subprocess and cancels the restart loop. @@ -198,8 +220,6 @@ func (b *IPBatch) restartLoop() { } if b.alive.Load() { - // Wait for death or context cancellation. - // Poll periodically since there's no notification channel. select { case <-b.ctx.Done(): return @@ -215,7 +235,6 @@ func (b *IPBatch) restartLoop() { case <-time.After(delay): } - // Kill old process if lingering. b.mu.Lock() if b.cmd != nil && b.cmd.Process != nil { b.cmd.Process.Kill() @@ -231,7 +250,6 @@ func (b *IPBatch) restartLoop() { continue } - // Canary query to validate the new process. if _, err := b.Query(canaryCommand); err != nil { b.log.Warn("ip batch: canary query failed", "err", err) b.alive.Store(false) diff --git a/src/yangerd/internal/monitor/monitor.go b/src/yangerd/internal/monitor/monitor.go index e8738831..e32d7994 100644 --- a/src/yangerd/internal/monitor/monitor.go +++ b/src/yangerd/internal/monitor/monitor.go @@ -38,7 +38,15 @@ type NLMonitor struct { fc iface.FileChecker // initDone is closed after the first initialDump completes. - initDone chan struct{} + // initDoneOnce ensures it is only closed once across restarts. + initDone chan struct{} + initDoneOnce sync.Once + + // redumpCh is written by the errorCallback goroutine to ask the + // event loop goroutine to run initialDump. Buffer of 1 so signals + // are coalesced — if a re-dump is already pending there is no point + // queuing another one. + redumpCh chan struct{} // staging holds raw ip-json data used as input to iface.Transform(). // Protected by mu. @@ -65,6 +73,7 @@ func New(linkBatch, addrBatch *ipbatch.IPBatch, brBatch *bridgebatch.BridgeBatch fc: fc, log: log, initDone: make(chan struct{}), + redumpCh: make(chan struct{}, 1), fdb: make(map[string]json.RawMessage), mdb: make(map[string]json.RawMessage), ethernet: make(map[string]json.RawMessage), @@ -115,6 +124,18 @@ func (m *NLMonitor) Run(ctx context.Context) error { if err == nil { return } + if errors.Is(err, syscall.ENOBUFS) { + // Kernel dropped events due to socket buffer overflow. + // Signal the event loop goroutine to do a full re-dump so + // that the snapshot is taken in the same goroutine as event + // processing — avoiding a race with concurrent refreshInterface + // calls that could otherwise be overwritten by the re-dump. + select { + case m.redumpCh <- struct{}{}: + default: // re-dump already pending + } + return + } m.log.Error("netlink subscription error", "err", err) cancel() } @@ -125,17 +146,23 @@ func (m *NLMonitor) Run(ctx context.Context) error { mdbCh := make(chan struct{}, 32) if err := netlink.LinkSubscribeWithOptions(linkCh, done, netlink.LinkSubscribeOptions{ - ErrorCallback: errorCallback, + ErrorCallback: errorCallback, + ReceiveBufferSize: 32 * 1024 * 1024, + ReceiveBufferForceSize: true, }); err != nil { return fmt.Errorf("subscribe link updates: %w", err) } if err := netlink.AddrSubscribeWithOptions(addrCh, done, netlink.AddrSubscribeOptions{ - ErrorCallback: errorCallback, + ErrorCallback: errorCallback, + ReceiveBufferSize: 32 * 1024 * 1024, + ReceiveBufferForceSize: true, }); err != nil { return fmt.Errorf("subscribe addr updates: %w", err) } if err := netlink.NeighSubscribeWithOptions(neighCh, done, netlink.NeighSubscribeOptions{ - ErrorCallback: errorCallback, + ErrorCallback: errorCallback, + ReceiveBufferSize: 32 * 1024 * 1024, + ReceiveBufferForceSize: true, }); err != nil { return fmt.Errorf("subscribe neigh updates: %w", err) } @@ -146,7 +173,7 @@ func (m *NLMonitor) Run(ctx context.Context) error { if err := m.initialDump(); err != nil { m.log.Error("initial dump failed", "err", err) } - close(m.initDone) + m.initDoneOnce.Do(func() { close(m.initDone) }) for { select { @@ -175,6 +202,12 @@ func (m *NLMonitor) Run(ctx context.Context) error { return fmt.Errorf("bridge mdb update channel closed") } m.handleMDBUpdate() + case <-m.redumpCh: + m.log.Warn("re-dumping all interfaces") + if err := m.initialDump(); err != nil { + m.log.Error("re-dump failed, will retry", "err", err) + m.requestRedump() + } } } } @@ -213,6 +246,11 @@ func (m *NLMonitor) handleLinkUpdate(update netlink.LinkUpdate) { return } + if update.Header.Type == syscall.RTM_DELLINK { + m.removeInterface(name) + return + } + m.refreshInterface(name) if m.ethRefresh != nil { m.ethRefresh(name) @@ -229,6 +267,9 @@ func (m *NLMonitor) handleAddrUpdate(update netlink.AddrUpdate) { m.log.Debug("handleAddrUpdate", "ifname", ifname) raw, err := m.queryAddr("addr show dev " + ifname) if err != nil { + if errors.Is(err, ipbatch.ErrBatchDead) { + m.requestRedump() + } m.log.Error("handleAddrUpdate queryAddr failed", "ifname", ifname, "err", err) return } @@ -255,6 +296,9 @@ func (m *NLMonitor) handleNeighUpdate(update netlink.NeighUpdate) { raw, err := m.queryBridge("fdb show br " + bridgeName) if err != nil { + if errors.Is(err, bridgebatch.ErrBatchDead) { + m.requestRedump() + } return } @@ -283,9 +327,37 @@ func (m *NLMonitor) handleMDBUpdate() { m.rebuild() } +// removeInterface purges all staged data for the named interface and +// triggers a rebuild. Used for RTM_DELLINK events where querying the +// device via ip-batch would produce no output and hang the batch process. +func (m *NLMonitor) removeInterface(name string) { + m.mu.Lock() + m.links = replaceByIfName(m.links, name, json.RawMessage(`[]`)) + m.addrs = replaceByIfName(m.addrs, name, json.RawMessage(`[]`)) + delete(m.fdb, name) + delete(m.mdb, name) + delete(m.ethernet, name) + delete(m.wifi, name) + delete(m.lastOperStatus, name) + m.mu.Unlock() + + m.log.Debug("removeInterface", "ifname", name) + m.rebuild() +} + +func (m *NLMonitor) requestRedump() { + select { + case m.redumpCh <- struct{}{}: + default: + } +} + func (m *NLMonitor) refreshInterface(name string) { linkRaw, err := m.queryLink("link show dev " + name) if err != nil { + if errors.Is(err, ipbatch.ErrBatchDead) { + m.requestRedump() + } return } @@ -319,10 +391,9 @@ func (m *NLMonitor) rebuild() { wfi := copyStringMap(m.wifi) fdb := copyStringMap(m.fdb) mdb := copyStringMap(m.mdb) - m.mu.Unlock() - doc = mergeAugments(doc, eth, wfi, fdb, mdb) m.tree.Set(treeKey, doc) + m.mu.Unlock() } // mergeAugments adds ethernet, wifi, and bridge data into the