Detect external bootc status changes via fsnotify (milestone 4d)

.github/workflows/ci.yaml

@@ -8,7 +8,7 @@ on:
     branches: [main]
 
 env:
-  BINK_COMMIT: 2b1bbdca74fc36d319fb1ea6d0523d9a732fcb1f
+  BINK_COMMIT: fdfdc863d9f4e9caf83e56c06bf7f940cfb9f315
 
 permissions: {}
 

Makefile

@@ -6,6 +6,8 @@ CONTAINER_TOOL ?= podman
 BINK_CLUSTER_NAME ?= e2e
 KUBECONFIG_BINK ?= ./kubeconfig-$(BINK_CLUSTER_NAME)
 ARTIFACTS ?= $(abspath _output/logs)
+BINK_NODE_DISK_IMAGE ?= ghcr.io/alicefr/bink/node:v1.35-fedora-44-disk
+BINK_LOCAL_REGISTRY_NODE_IMAGE ?= registry.cluster.local:5000/node
 # YEAR defines the year value used for substituting the YEAR placeholder in the boilerplate header.
 YEAR ?= $(shell date +%Y)
 
@@ -62,7 +64,11 @@ e2e: ## Run e2e tests (requires: make deploy-bink). V=1 for verbose. RUN=<regex>
 	rm -rf $(ARTIFACTS)
 	cd test/e2e && KUBECONFIG=$(abspath $(KUBECONFIG_BINK)) BINK_CLUSTER_NAME=$(BINK_CLUSTER_NAME) \
 		$(if $(BINK_NODE_IMAGE),BINK_NODE_IMAGE=$(BINK_NODE_IMAGE)) \
+		BINK_NODE_DISK_IMAGE=$(BINK_NODE_DISK_IMAGE) \
+		BINK_LOCAL_REGISTRY_NODE_IMAGE=$(BINK_LOCAL_REGISTRY_NODE_IMAGE) \
 		ARTIFACTS=$(ARTIFACTS) \
+		BINK_NODE_IMAGE_DIGEST=$$(skopeo inspect --tls-verify=false --format '{{.Digest}}' docker://localhost:5000/node:latest) \
+		UPDATE_IMAGE_DIGEST=$$(skopeo inspect --tls-verify=false docker://localhost:5000/node:update | jq -r '.Digest') \
 		go test -timeout 10m -count=1 $(if $(V),-v) $(if $(RUN),-run $(RUN)) .
 
 ##@ Build
@@ -82,6 +88,12 @@ build-daemon: ## Build daemon binary.
 buildimg: ## Build container image.
 	$(CONTAINER_TOOL) build -t $(IMG) .
 
+.PHONY: build-update-image
+build-update-image: ## Build a derived node image for update testing and push to bink registry.
+	@printf 'FROM localhost:5000/node:latest\nRUN touch /usr/share/update-marker\n' | \
+		podman build -t localhost:5000/node:update -f - .
+	podman push --tls-verify=false localhost:5000/node:update
+
 ##@ Deployment
 
 ifndef ignore-not-found
@@ -111,16 +123,26 @@ undeploy: kustomize ## Undeploy controller from the K8s cluster specified in ~/.
 # Note the :latest tag here: this makes the pull policy be Always.
 IMG_BINK ?= registry.cluster.local:5000/bootc-operator-e2e:latest
 
-.PHONY: start-bink
-start-bink: ## Start a bink cluster (idempotent).
+.PHONY: seed-node-image
+seed-node-image: ## Pull the bootc node image by digest and push to the bink registry.
 	bink registry start
-	bink cluster list 2>&1 | grep -qw $(BINK_CLUSTER_NAME) || \
+	podman pull $(BINK_NODE_DISK_IMAGE)
+	bootc_img=$$(podman inspect --format '{{index .Config.Labels "bink.bootc-image"}}' $(BINK_NODE_DISK_IMAGE)) && \
+		bootc_digest=$$(podman inspect --format '{{index .Config.Labels "bink.bootc-image-digest"}}' $(BINK_NODE_DISK_IMAGE)) && \
+		podman pull "$$bootc_img@$$bootc_digest" && \
+		podman tag "$$bootc_img@$$bootc_digest" localhost:5000/node:latest
+	podman push --tls-verify=false localhost:5000/node:latest
+
+.PHONY: start-bink
+start-bink: seed-node-image ## Start a bink cluster (idempotent).
+	bink cluster list 2>&1 | grep -qw $(BINK_CLUSTER_NAME) || { \
+		node_digest=$$(skopeo inspect --tls-verify=false --format '{{.Digest}}' docker://localhost:5000/node:latest) && \
 		bink cluster start --cluster-name $(BINK_CLUSTER_NAME) --node-name controller --api-port 0 --expose $(KUBECONFIG_BINK) \
-		$(if $(BINK_NODE_IMAGE),--node-image $(BINK_NODE_IMAGE))
+		--node-image $(BINK_NODE_DISK_IMAGE) --target-imgref $(BINK_LOCAL_REGISTRY_NODE_IMAGE)@$$node_digest; }
 	kubectl --kubeconfig $(KUBECONFIG_BINK) wait --for=condition=Ready node/controller --timeout=5m
 
 .PHONY: deploy-bink
-deploy-bink: start-bink kustomize ## Deploy to a bink cluster (requires: buildimg).
+deploy-bink: start-bink build-update-image kustomize ## Deploy to a bink cluster (requires: buildimg).
 	podman push --tls-verify=false $(IMG) localhost:5000/bootc-operator-e2e:latest
 	# On re-deploy, restart the rollout to force a re-pull of the :latest tag.
 	# On fresh deploy, skip the restart -- the pod is already pulling the correct image.

cmd/daemon/main.go

@@ -6,6 +6,7 @@ import (
 	"flag"
 	"fmt"
 	"os"
+	"time"
 
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -14,6 +15,7 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 
 	bootcv1alpha1 "github.com/jlebon/bootc-operator/api/v1alpha1"
@@ -32,6 +34,9 @@ func init() {
 }
 
 func main() {
+	var pollInterval time.Duration
+	flag.DurationVar(&pollInterval, "poll-interval", 5*time.Minute, "Interval for polling bootc status as a fallback to fsnotify")
+
 	opts := zap.Options{
 		Development: true,
 	}
@@ -62,17 +67,32 @@ func main() {
 		os.Exit(1)
 	}
 
+	statusChanged := make(chan event.GenericEvent, 1)
+
 	if err := (&daemon.BootcNodeReconciler{
-		Client:   mgr.GetClient(),
-		Scheme:   mgr.GetScheme(),
-		NodeName: nodeName,
-		Executor: bootc.NewHostExecutor(),
+		Client:        mgr.GetClient(),
+		Scheme:        mgr.GetScheme(),
+		NodeName:      nodeName,
+		Executor:      bootc.NewHostExecutor(),
+		StatusChanged: statusChanged,
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "Failed to create controller", "controller", "bootcnode")
 		os.Exit(1)
 	}
 
-	setupLog.Info("Starting daemon", "node", nodeName)
+	watcher := &daemon.StatusWatcher{
+		PollInterval: pollInterval,
+		PrimaryPath:  daemon.DefaultPrimaryPath,
+		FallbackPath: daemon.DefaultFallbackPath,
+		Events:       statusChanged,
+		NodeName:     nodeName,
+	}
+	if err := mgr.Add(watcher); err != nil {
+		setupLog.Error(err, "Failed to add status watcher")
+		os.Exit(1)
+	}
+
+	setupLog.Info("Starting daemon", "node", nodeName, "pollInterval", pollInterval)
 	if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
 		setupLog.Error(err, "Failed to run daemon")
 		os.Exit(1)

config/daemon/daemon.yaml

@@ -38,7 +38,7 @@ spec:
         resources:
           limits:
             cpu: 500m
-            memory: 128Mi
+            memory: 512Mi
           requests:
             cpu: 10m
             memory: 64Mi

internal/bootc/executor.go

@@ -6,13 +6,18 @@ import (
 	"context"
 	"fmt"
 	"os/exec"
+	"strings"
+
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
 )
 
 // Executor abstracts the execution of bootc commands on the host.
 // The real implementation uses nsenter to enter the host's mount and
 // PID namespaces. Tests can provide a fake implementation.
 type Executor interface {
 	Status(ctx context.Context) ([]byte, error)
+	Switch(ctx context.Context, image string) error
+	Reboot(ctx context.Context) error
 }
 
 // HostExecutor runs bootc commands on the host via nsenter.
@@ -23,21 +28,45 @@ func NewHostExecutor() *HostExecutor {
 	return &HostExecutor{}
 }
 
-func (e *HostExecutor) Status(ctx context.Context) ([]byte, error) {
-	cmd := exec.CommandContext(ctx,
-		"nsenter",
+func (e *HostExecutor) nsenterCmd(ctx context.Context, args ...string) *exec.Cmd {
+	base := []string{
 		"--target", "1",
-		"--mount",
-		"--pid",
-		"--setuid", "0",
-		"--setgid", "0",
-		"--env",
-		"--",
-		"bootc", "status", "--json", "--format-version", "1",
-	)
+		"--mount", "--pid",
+		"--setuid", "0", "--setgid", "0",
+		"--env", "--",
+	}
+	return exec.CommandContext(ctx, "nsenter", append(base, args...)...)
+}
+
+func (e *HostExecutor) Status(ctx context.Context) ([]byte, error) {
+	cmd := e.nsenterCmd(ctx, "bootc", "status", "--json", "--format-version", "1")
 	out, err := cmd.Output()
 	if err != nil {
 		return nil, fmt.Errorf("running bootc status: %w", err)
 	}
 	return out, nil
 }
+
+func (e *HostExecutor) Switch(ctx context.Context, image string) error {
+	log := logf.FromContext(ctx)
+
+	cmd := e.nsenterCmd(ctx, "bootc", "switch", image)
+	log.Info("Executing", "cmd", strings.Join(cmd.Args, " "))
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		return fmt.Errorf("running bootc switch: %s: %w", out, err)
+	}
+	return nil
+}
+
+func (e *HostExecutor) Reboot(ctx context.Context) error {
+	log := logf.FromContext(ctx)
+
+	cmd := e.nsenterCmd(ctx, "systemctl", "reboot")
+	log.Info("Executing", "cmd", strings.Join(cmd.Args, " "))
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		return fmt.Errorf("running systemctl reboot: %s: %w", out, err)
+	}
+	return nil
+}

internal/daemon/fake_test.go

@@ -4,24 +4,128 @@ package daemon
 
 import (
 	"context"
+	"encoding/json"
+	"strings"
 	"sync"
+
+	testutil "github.com/jlebon/bootc-operator/test/util"
+
+	"github.com/jlebon/bootc-operator/internal/bootc"
 )
 
 type fakeExecutor struct {
-	mu   sync.Mutex
-	data []byte
-	err  error
+	mu        sync.Mutex
+	status    bootc.Status
+	statusErr error
+
+	switchErr  error
+	switchImg  string
+	switchHook func()
+	rebooted   bool
 }
 
 func (f *fakeExecutor) Status(_ context.Context) ([]byte, error) {
 	f.mu.Lock()
 	defer f.mu.Unlock()
-	return f.data, f.err
+	if f.statusErr != nil {
+		return nil, f.statusErr
+	}
+	data, err := json.Marshal(f.status)
+	if err != nil {
+		return nil, err
+	}
+	return data, nil
+}
+
+func (f *fakeExecutor) Switch(_ context.Context, image string) error {
+	f.mu.Lock()
+	f.switchImg = image
+	hook := f.switchHook
+	err := f.switchErr
+	f.mu.Unlock()
+
+	if hook != nil {
+		hook()
+	}
+	if err != nil {
+		return err
+	}
+
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	_, digest, _ := strings.Cut(image, "@")
+	f.status.Status.Staged = newBootEntry(image, digest)
+	return nil
+}
+
+func (f *fakeExecutor) Reboot(_ context.Context) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.rebooted = true
+	return nil
+}
+
+func (f *fakeExecutor) setStatusErr(err error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.statusErr = err
+}
+
+func (f *fakeExecutor) setSwitchErr(err error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.switchErr = err
+}
+
+func (f *fakeExecutor) setSwitchHook(hook func()) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.switchHook = hook
 }
 
-func (f *fakeExecutor) set(data []byte, err error) {
+func (f *fakeExecutor) getSwitchImg() string {
 	f.mu.Lock()
 	defer f.mu.Unlock()
-	f.data = data
-	f.err = err
+	return f.switchImg
+}
+
+func (f *fakeExecutor) getRebooted() bool {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.rebooted
+}
+
+func (f *fakeExecutor) reset() {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.status = bootc.Status{}
+	f.statusErr = nil
+	f.switchErr = nil
+	f.switchImg = ""
+	f.switchHook = nil
+	f.rebooted = false
+}
+
+func newBootEntry(image, digest string) *bootc.BootEntry {
+	return &bootc.BootEntry{
+		Image: &bootc.ImageStatus{
+			Image:        bootc.ImageReference{Image: image, Transport: "registry"},
+			ImageDigest:  digest,
+			Architecture: "amd64",
+		},
+	}
+}
+
+func newBootcStatus(bootedDigest string) bootc.Status {
+	return bootc.Status{
+		APIVersion: "org.containers.bootc/v1alpha1",
+		Kind:       "BootcHost",
+		Spec: bootc.StatusSpec{
+			Image:     &bootc.ImageReference{Image: testutil.ImageTaggedRef, Transport: "registry"},
+			BootOrder: "default",
+		},
+		Status: bootc.StatusBody{
+			Booted: newBootEntry(testutil.ImageTaggedRef, bootedDigest),
+		},
+	}
 }