peter/infix
1
0
Fork 0
mirror of https://github.com/kernelkit/infix.git synced 2026-07-29 20:23:01 +02:00
Code Issues Packages Projects Releases Wiki Activity

Fix interface and bridge race conditions

Browse Source
This commit is contained in:
Mattias Walström
2026-06-27 08:39:48 +02:00
parent 7b45f2589c
commit 4cc9751c7b
3 changed files with 148 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/yangerd/internal/bridgebatch/bridgebatch.go
+33 -11
View File
@@ -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.
src/yangerd/internal/ipbatch/ipbatch.go
+37 -19
View File
@@ -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 <nonexistent>"). 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)
src/yangerd/internal/monitor/monitor.go
+78 -7
View File
@@ -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