yangerd: Change containers from polled to reactive

Author: mattiaswal

src/yangerd/cmd/yangerd/main.go

@@ -17,6 +17,7 @@ import (
 	"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/containermonitor"
 	"github.com/kernelkit/infix/src/yangerd/internal/dbusmonitor"
 	"github.com/kernelkit/infix/src/yangerd/internal/ethmonitor"
 	"github.com/kernelkit/infix/src/yangerd/internal/frrvty"
@@ -235,6 +236,17 @@ func main() {
 		}()
 	}
 
+	if cfg.EnableContainers {
+		ctrmon := containermonitor.New(t, cmd, fs, slogLog)
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			if err := ctrmon.Run(ctx); err != nil && ctx.Err() == nil {
+				slogLog.Error("containermonitor exited", "err", err)
+			}
+		}()
+	}
+
 	zapi := zapiwatcher.New(t, frrvty.New(""), slogLog)
 	wg.Add(1)
 	go func() {
@@ -338,19 +350,8 @@ func main() {
 			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)
-		}
-	}
+	// Container operational data is handled by containermonitor (a
+	// `podman events` stream), not the fswatcher.
 	fsw.InitialRead()
 	wg.Add(1)
 	go func() {

src/yangerd/internal/containermonitor/containermonitor.go

@@ -0,0 +1,181 @@
+// Package containermonitor keeps the infix-containers subtree in the tree
+// in sync with podman.  A persistent `podman events` subprocess is used
+// purely as a change trigger; on every event the full container table is
+// re-read with `podman ps` (via collector.CollectContainers) and the
+// subtree replaced, so removed containers disappear and containers present
+// before yangerd started are picked up.
+//
+// This replaces an earlier inotify watch on /run/libpod/events, which was
+// reactive-only and silently went stale whenever an event was missed
+// (debounce coalescing, inotify overflow, a removal racing the re-read, or
+// yangerd starting after the container).  `podman events` reads whichever
+// events backend podman is configured for (file or journald), so it does
+// not depend on a specific on-disk layout.
+package containermonitor
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"log/slog"
+	"os/exec"
+	"time"
+
+	"github.com/kernelkit/infix/src/yangerd/internal/backoff"
+	"github.com/kernelkit/infix/src/yangerd/internal/collector"
+	"github.com/kernelkit/infix/src/yangerd/internal/tree"
+)
+
+const (
+	treeKey = "infix-containers:containers"
+
+	// debounceDelay coalesces bursts of events into one re-read.
+	debounceDelay = 200 * time.Millisecond
+)
+
+// ContainerMonitor subscribes to container lifecycle events via a
+// persistent `podman events` subprocess and re-reads the full container
+// table on every event.
+type ContainerMonitor struct {
+	tree    *tree.Tree
+	log     *slog.Logger
+	refresh chan struct{}
+
+	// collect returns the current container subtree, or nil when there are
+	// no containers; overridable in tests.
+	collect func() json.RawMessage
+}
+
+// New creates a ContainerMonitor.
+func New(t *tree.Tree, cmd collector.CommandRunner, fs collector.FileReader, log *slog.Logger) *ContainerMonitor {
+	if log == nil {
+		log = slog.Default()
+	}
+	return &ContainerMonitor{
+		tree:    t,
+		log:     log,
+		refresh: make(chan struct{}, 1),
+		collect: func() json.RawMessage { return collector.CollectContainers(cmd, fs) },
+	}
+}
+
+// Run starts the container monitor.  It blocks until ctx is cancelled,
+// restarting the events subprocess with backoff if it exits.
+func (m *ContainerMonitor) Run(ctx context.Context) error {
+	go m.refreshLoop(ctx)
+
+	bo := backoff.Default()
+	delay := bo.Initial
+
+	for {
+		err := m.runOnce(ctx)
+		if ctx.Err() != nil {
+			return ctx.Err()
+		}
+
+		m.log.Warn("container monitor: subprocess exited, restarting",
+			"err", err, "delay", delay)
+		if err := backoff.Sleep(ctx, delay); err != nil {
+			return err
+		}
+		delay = bo.Next(delay)
+	}
+}
+
+func (m *ContainerMonitor) runOnce(ctx context.Context) error {
+	cmd := exec.CommandContext(ctx, "podman", "events", "--filter", "type=container", "--format", "json")
+	stdout, err := cmd.StdoutPipe()
+	if err != nil {
+		return fmt.Errorf("stdout pipe: %w", err)
+	}
+	if err := cmd.Start(); err != nil {
+		return fmt.Errorf("start podman events: %w", err)
+	}
+	defer cmd.Wait()
+
+	// Pick up containers that existed before we attached.
+	m.triggerRefresh()
+
+	return m.readEvents(stdout)
+}
+
+// readEvents consumes the newline-delimited JSON event stream.  Each event
+// is only a trigger; the payload is never used to build state.
+func (m *ContainerMonitor) readEvents(r io.Reader) error {
+	scanner := bufio.NewScanner(r)
+	scanner.Buffer(make([]byte, 0, 64*1024), 1*1024*1024)
+
+	for scanner.Scan() {
+		line := scanner.Bytes()
+		if len(line) == 0 {
+			continue
+		}
+		if status := eventStatus(line); status != "" {
+			m.log.Debug("container monitor: event", "status", status)
+		}
+		m.triggerRefresh()
+	}
+	if err := scanner.Err(); err != nil {
+		return fmt.Errorf("read podman events: %w", err)
+	}
+	return fmt.Errorf("podman events process exited")
+}
+
+// eventStatus extracts the event status for logging; best-effort only.
+func eventStatus(line []byte) string {
+	var ev struct {
+		Status string `json:"Status"`
+	}
+	if json.Unmarshal(line, &ev) != nil {
+		return ""
+	}
+	return ev.Status
+}
+
+// triggerRefresh requests a table re-read; the buffered channel collapses
+// pending requests into one.
+func (m *ContainerMonitor) triggerRefresh() {
+	select {
+	case m.refresh <- struct{}{}:
+	default:
+	}
+}
+
+func (m *ContainerMonitor) refreshLoop(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-m.refresh:
+		}
+
+		// Let a burst of events settle before reading.
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(debounceDelay):
+		}
+		select {
+		case <-m.refresh:
+		default:
+		}
+
+		m.updateTree()
+	}
+}
+
+// updateTree re-reads the full container table and replaces the subtree.
+// With no containers the key is deleted rather than left as an empty node,
+// so an idle-but-enabled container feature reads as absent.
+func (m *ContainerMonitor) updateTree() {
+	data := m.collect()
+	if len(data) == 0 {
+		m.tree.Delete(treeKey)
+		m.log.Debug("container monitor: no containers, key removed")
+		return
+	}
+	m.tree.Set(treeKey, data)
+	m.log.Debug("container monitor: tree updated")
+}

src/yangerd/internal/containermonitor/containermonitor_test.go

@@ -0,0 +1,88 @@
+package containermonitor
+
+import (
+	"context"
+	"encoding/json"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/kernelkit/infix/src/yangerd/internal/tree"
+)
+
+// newTestMonitor builds a monitor whose collect() is driven by the test.
+// cmd/fs are nil since collect is overridden, so the default closure that
+// would use them is never called.
+func newTestMonitor(t *testing.T, collect func() json.RawMessage) (*ContainerMonitor, *tree.Tree) {
+	t.Helper()
+	tr := tree.New()
+	m := New(tr, nil, nil, nil)
+	m.collect = collect
+	return m, tr
+}
+
+func TestUpdateTreeSetsContainers(t *testing.T) {
+	m, tr := newTestMonitor(t, func() json.RawMessage {
+		return json.RawMessage(`{"container":[{"name":"web"}]}`)
+	})
+
+	m.updateTree()
+
+	got := tr.Get(treeKey)
+	if got == nil || !strings.Contains(string(got), "web") {
+		t.Fatalf("expected container data, got %s", got)
+	}
+}
+
+// With no containers the key must be deleted, not left as an empty node,
+// so an idle-but-enabled container feature reads as absent.
+func TestUpdateTreeDeletesWhenEmpty(t *testing.T) {
+	m, tr := newTestMonitor(t, func() json.RawMessage { return nil })
+
+	tr.Set(treeKey, json.RawMessage(`{"container":[{"name":"old"}]}`))
+	m.updateTree()
+
+	if got := tr.Get(treeKey); got != nil {
+		t.Fatalf("expected key removed when no containers, got %s", got)
+	}
+}
+
+// An event in the stream must trigger a re-read; here the re-read clears a
+// previously-present container, proving the stream drives reconciliation.
+func TestEventTriggersRefresh(t *testing.T) {
+	calls := 0
+	m, tr := newTestMonitor(t, func() json.RawMessage {
+		calls++
+		return nil // container is gone
+	})
+	tr.Set(treeKey, json.RawMessage(`{"container":[{"name":"gone"}]}`))
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go m.refreshLoop(ctx)
+
+	// A container "died" event, newline-framed as podman emits it.
+	go m.readEvents(strings.NewReader(`{"Type":"container","Status":"died","Name":"gone"}` + "\n"))
+
+	deadline := time.After(2 * time.Second)
+	for {
+		if tr.Get(treeKey) == nil && calls > 0 {
+			break
+		}
+		select {
+		case <-deadline:
+			t.Fatalf("event did not trigger reconcile; calls=%d tree=%s", calls, tr.Get(treeKey))
+		default:
+			time.Sleep(10 * time.Millisecond)
+		}
+	}
+}
+
+func TestEventStatus(t *testing.T) {
+	if s := eventStatus([]byte(`{"Status":"start"}`)); s != "start" {
+		t.Errorf("eventStatus = %q, want start", s)
+	}
+	if s := eventStatus([]byte(`not json`)); s != "" {
+		t.Errorf("eventStatus on garbage = %q, want empty", s)
+	}
+}