feat: add authenticated controller status reporting

README.md

@@ -175,7 +175,7 @@ Supported deployment shapes include virtual machines, dedicated physical machine
 | Build a compatible project contract | [Project CI standard](docs/PROJECT-STANDARD.md) |
 | Verify project compliance | [Compliance checklist](docs/COMPLIANCE-CHECKLIST.md) |
 | Configure upgrades, cleanup, draining, and rebooting | [Host maintenance](docs/HOST-MAINTENANCE.md) |
-| Configure host health and external missed-heartbeat detection | [Fleet health monitoring](docs/HEALTH-MONITORING.md) |
+| Configure host health and authenticated outbound observation | [Fleet health monitoring](docs/HEALTH-MONITORING.md) and [status reporting](docs/STATUS-REPORTING.md) |
 | Understand secret storage and injection | [Secrets model](docs/SECRETS.md) |
 | Use private fleet workers for a public project | [Public projects and private delivery](docs/PUBLIC-PRIVATE-CONFIGURATION.md) |
 | See planned work | [Roadmap](docs/ROADMAP.md) |

controller/Dockerfile

@@ -12,7 +12,9 @@ FROM debian:13.6-slim
 ARG CI_FLEET_COMMIT=unknown
 LABEL org.opencontainers.image.revision="${CI_FLEET_COMMIT}" \
       io.randomdevelopment.ci-fleet.managed="true"
-RUN apt-get update && apt-get install -y --no-install-recommends ca-certificates && rm -rf /var/lib/apt/lists/*
+RUN apt-get update && apt-get install -y --no-install-recommends ca-certificates \
+    && install -d -o 65532 -g 65532 /run/ci-fleet \
+    && rm -rf /var/lib/apt/lists/*
 COPY --from=build /out/ci-fleet-controller /usr/local/bin/ci-fleet-controller
 USER 65532:65532
 ENTRYPOINT ["/usr/local/bin/ci-fleet-controller"]

controller/config.go

@@ -27,6 +27,7 @@ type Config struct {
 	RunnerMemory    int64
 	DockerGID       string
 	RunnerTTL       time.Duration
+	StatusFile      string
 }
 
 func configFromEnv() (Config, error) {
@@ -85,6 +86,7 @@ func configFromEnv() (Config, error) {
 		RunnerMemory: runnerMemoryMiB * 1024 * 1024,
 		DockerGID:    os.Getenv("CI_FLEET_DOCKER_GID"),
 		RunnerTTL:    runnerTTL,
+		StatusFile:   getenv("CI_FLEET_STATUS_FILE", "/run/ci-fleet/status.json"),
 	}
 	return cfg, cfg.Validate()
 }

controller/main.go

@@ -10,6 +10,7 @@ import (
 	"os"
 	"os/signal"
 	"syscall"
+	"time"
 
 	"github.com/actions/scaleset"
 	"github.com/actions/scaleset/listener"
@@ -74,6 +75,8 @@ func run(ctx context.Context) error {
 
 	scaler := &Scaler{runners: newRunnerState(), dockerClient: docker, scalesetClient: client, logger: logger, config: cfg, scaleSetID: set.ID}
 	if err := scaler.recoverStale(ctx); err != nil { return err }
+	scaler.writeStatus()
+	go scaler.publishStatus(ctx, time.Minute)
 	defer scaler.shutdown(context.WithoutCancel(ctx))
 	hostname, err := os.Hostname()
 	if err != nil { return fmt.Errorf("get hostname: %w", err) }

controller/scaler.go

@@ -15,6 +15,7 @@ import (
 	"github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/api/types/filters"
 	dockerclient "github.com/docker/docker/client"
+	"github.com/docker/docker/errdefs"
 	"github.com/google/uuid"
 )
 
@@ -30,6 +31,7 @@ type Scaler struct {
 }
 
 func (s *Scaler) HandleDesiredRunnerCount(ctx context.Context, count int) (int, error) {
+	defer s.writeStatus()
 	current := s.runners.count()
 	target := min(s.config.MaxRunners, s.config.MinRunners+count)
 	for i := current; i < target; i++ {
@@ -44,16 +46,19 @@ func (s *Scaler) HandleJobStarted(_ context.Context, job *scaleset.JobStarted) e
 	if !s.runners.markBusy(job.RunnerName) {
 		return fmt.Errorf("job started for unknown runner %q", job.RunnerName)
 	}
+	s.writeStatus()
 	s.logger.Info("job started", "runner", job.RunnerName, "jobID", job.JobID)
 	return nil
 }
 
 func (s *Scaler) HandleJobCompleted(ctx context.Context, job *scaleset.JobCompleted) error {
-	id, ok := s.runners.markDone(job.RunnerName)
+	id, cleanup, ok := s.runners.markDone(job.RunnerName)
 	if !ok {
 		return fmt.Errorf("job completed for unknown runner %q", job.RunnerName)
 	}
+	s.writeStatus()
 	s.logger.Info("job completed", "runner", job.RunnerName, "jobID", job.JobID)
+	if !cleanup { return nil }
 	return s.logAndRemove(ctx, job.RunnerName, id)
 }
 
@@ -92,10 +97,50 @@ func (s *Scaler) startRunner(ctx context.Context) (string, error) {
 		return "", fmt.Errorf("start runner container: %w", err)
 	}
 	s.runners.addIdle(name, created.ID)
+	go s.watchRunner(context.WithoutCancel(ctx), name, created.ID)
 	s.logger.Info("runner started", "runner", name, "containerID", created.ID)
 	return name, nil
 }
 
+func (s *Scaler) watchRunner(ctx context.Context, name, id string) {
+waitLoop:
+	for {
+		stopped, errors := s.dockerClient.ContainerWait(ctx, id, container.WaitConditionNotRunning)
+		select {
+		case err, ok := <-errors:
+			if !s.runners.contains(name, id) { return }
+			if ok && errdefs.IsNotFound(err) { break waitLoop }
+			if ok { s.logger.Warn("watch runner container", "runner", name, "error", err) }
+			time.Sleep(time.Minute)
+		case response, ok := <-stopped:
+			if !ok {
+				if !s.runners.contains(name, id) { return }
+				time.Sleep(time.Minute)
+				continue
+			}
+			if response.Error != nil {
+				if !s.runners.contains(name, id) { return }
+				s.logger.Warn("watch runner container", "runner", name, "error", response.Error.Message)
+				time.Sleep(time.Minute)
+				continue
+			}
+			break waitLoop
+		}
+	}
+	if !s.runners.markExited(name, id) { return }
+	s.writeStatus()
+	s.logger.Warn("runner container exited before job completion", "runner", name)
+	for {
+		if err := s.logAndRemove(context.WithoutCancel(ctx), name, id); err == nil {
+			time.AfterFunc(10*time.Minute, func() { s.runners.forgetExited(name, id) })
+			return
+		} else {
+			s.logger.Warn("retry exited runner cleanup", "runner", name, "error", err)
+		}
+		time.Sleep(time.Minute)
+	}
+}
+
 func (s *Scaler) recoverStale(ctx context.Context) error {
 	f := filters.NewArgs(
 		filters.Arg("label", labelPrefix+"managed=true"),
@@ -122,13 +167,14 @@ func (s *Scaler) logAndRemove(ctx context.Context, name, id string) error {
 	} else {
 		s.logger.Warn("could not collect runner logs", "runner", name, "error", err)
 	}
-	if err := s.dockerClient.ContainerRemove(ctx, id, container.RemoveOptions{Force: true, RemoveVolumes: true}); err != nil {
+	if err := s.dockerClient.ContainerRemove(ctx, id, container.RemoveOptions{Force: true, RemoveVolumes: true}); err != nil && !errdefs.IsNotFound(err) {
 		return fmt.Errorf("remove runner %s: %w", name, err)
 	}
 	return nil
 }
 
 func (s *Scaler) shutdown(ctx context.Context) {
+	defer s.writeStatus()
 	for name, id := range s.runners.drain() {
 		if err := s.logAndRemove(ctx, name, id); err != nil {
 			s.logger.Error("runner shutdown failed", slog.String("runner", name), slog.String("error", err.Error()))

controller/state.go

@@ -3,19 +3,25 @@ package main
 import "sync"
 
 type runnerState struct {
-	mu   sync.Mutex
-	idle map[string]string
-	busy map[string]string
+	mu     sync.Mutex
+	idle   map[string]string
+	busy   map[string]string
+	exited map[string]string
 }
 
 func newRunnerState() runnerState {
-	return runnerState{idle: make(map[string]string), busy: make(map[string]string)}
+	return runnerState{idle: make(map[string]string), busy: make(map[string]string), exited: make(map[string]string)}
 }
 
 func (r *runnerState) count() int {
+	current, _ := r.counts()
+	return current
+}
+
+func (r *runnerState) counts() (int, int) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
-	return len(r.idle) + len(r.busy)
+	return len(r.idle) + len(r.busy), len(r.busy)
 }
 
 func (r *runnerState) addIdle(name, id string) {
@@ -29,34 +35,69 @@ func (r *runnerState) markBusy(name string) bool {
 	defer r.mu.Unlock()
 	id, ok := r.idle[name]
 	if !ok {
-		return false
+		_, ok = r.exited[name]
+		return ok
 	}
 	delete(r.idle, name)
 	r.busy[name] = id
 	return true
 }
 
-func (r *runnerState) markDone(name string) (string, bool) {
+func (r *runnerState) contains(name, expectedID string) bool {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return r.idle[name] == expectedID || r.busy[name] == expectedID || r.exited[name] == expectedID
+}
+
+func (r *runnerState) markExited(name, expectedID string) bool {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if id, ok := r.busy[name]; ok && id == expectedID {
+		delete(r.busy, name)
+		r.exited[name] = id
+		return true
+	}
+	if id, ok := r.idle[name]; ok && id == expectedID {
+		delete(r.idle, name)
+		r.exited[name] = id
+		return true
+	}
+	return false
+}
+
+func (r *runnerState) forgetExited(name, expectedID string) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.exited[name] == expectedID { delete(r.exited, name) }
+}
+
+func (r *runnerState) markDone(name string) (string, bool, bool) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 	if id, ok := r.busy[name]; ok {
 		delete(r.busy, name)
-		return id, true
+		return id, true, true
 	}
 	if id, ok := r.idle[name]; ok {
 		delete(r.idle, name)
-		return id, true
+		return id, true, true
+	}
+	if id, ok := r.exited[name]; ok {
+		delete(r.exited, name)
+		return id, false, true
 	}
-	return "", false
+	return "", false, false
 }
 
 func (r *runnerState) drain() map[string]string {
 	r.mu.Lock()
 	defer r.mu.Unlock()
-	all := make(map[string]string, len(r.idle)+len(r.busy))
+	all := make(map[string]string, len(r.idle)+len(r.busy)+len(r.exited))
 	for name, id := range r.idle { all[name] = id }
 	for name, id := range r.busy { all[name] = id }
+	for name, id := range r.exited { all[name] = id }
 	clear(r.idle)
 	clear(r.busy)
+	clear(r.exited)
 	return all
 }

controller/status.go

@@ -0,0 +1,67 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"sync"
+	"time"
+)
+
+var statusWriteMu sync.Mutex
+var encodeControllerStatus = func(file *os.File, value controllerStatus) error {
+	return json.NewEncoder(file).Encode(value)
+}
+
+type controllerStatus struct {
+	Controller      string `json:"controller"`
+	SoftwareVersion string `json:"software_version"`
+	Current         int    `json:"current"`
+	Busy            int    `json:"busy"`
+	Maximum         int    `json:"maximum"`
+	GeneratedAt     int64  `json:"generated_at"`
+}
+
+func (s *Scaler) writeStatus() {
+	statusWriteMu.Lock()
+	defer statusWriteMu.Unlock()
+	current, busy := s.runners.counts()
+	softwareVersion := commitSHA
+	if softwareVersion == "unknown" { softwareVersion = version }
+	value := controllerStatus{
+		Controller: s.config.FleetInstance, SoftwareVersion: softwareVersion,
+		Current: current, Busy: busy, Maximum: s.config.MaxRunners, GeneratedAt: time.Now().Unix(),
+	}
+	directory := filepath.Dir(s.config.StatusFile)
+	if err := os.MkdirAll(directory, 0o755); err != nil {
+		s.logger.Warn("write controller status", "error", err)
+		return
+	}
+	temporary, err := os.CreateTemp(directory, ".status-*")
+	if err != nil {
+		s.logger.Warn("write controller status", "error", err)
+		return
+	}
+	name := temporary.Name()
+	defer os.Remove(name)
+	if err = temporary.Chmod(0o644); err == nil {
+		err = encodeControllerStatus(temporary, value)
+	}
+	if closeErr := temporary.Close(); err == nil { err = closeErr }
+	if err == nil { err = os.Rename(name, s.config.StatusFile) }
+	if err != nil { s.logger.Warn("write controller status", "error", err) }
+}
+
+func (s *Scaler) publishStatus(ctx context.Context, interval time.Duration) {
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			s.writeStatus()
+		}
+	}
+}