diff --git a/src/yangerd/cmd/yangerd/main.go b/src/yangerd/cmd/yangerd/main.go index 16577ad1..65077290 100644 --- a/src/yangerd/cmd/yangerd/main.go +++ b/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() { diff --git a/src/yangerd/internal/containermonitor/containermonitor.go b/src/yangerd/internal/containermonitor/containermonitor.go new file mode 100644 index 00000000..70e1556b --- /dev/null +++ b/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") +} diff --git a/src/yangerd/internal/containermonitor/containermonitor_test.go b/src/yangerd/internal/containermonitor/containermonitor_test.go new file mode 100644 index 00000000..6634f4bd --- /dev/null +++ b/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) + } +}