feat(agent): worker cloud-init + systemd + bearer auth (FJB-94 Phase B)

cmd/fj-bellows/main.go

@@ -46,6 +46,12 @@ func main() {
 	controlListen := flag.String("control-listen", "127.0.0.1:9876", "control plane listen address (TCP); empty disables")
 	controlTokenFile := flag.String("control-token-file", "", "bearer-token file for the control plane (required for non-loopback binds; mode 0600)")
 	enableControlWrites := flag.Bool("enable-control-writes", false, "expose mutating control RPCs (ForceReap, ForceProvision); off by default")
+	// fjbagent (FJB-94). The agent's version implicitly tracks this
+	// orchestrator's build (main.version), so there is no per-deployment
+	// version choice — only a token file (required) and an optional URL
+	// template override (default points at the matching github release).
+	fjbAgentTokenFile := flag.String("fjbagent-token-file", "", "bearer-token file for fjbagent on workers (mode 0600); empty disables agent install")
+	fjbAgentURLTmpl := flag.String("fjbagent-url", "", "URL template for the fjbagent binary; supports {{.Version}} (resolved from this build) and $rarch (resolved by cloud-init). Empty uses the default github releases URL.")
 	flag.Parse()
 
 	// Wrap the stderr text handler with a logbus tee so the control plane's
@@ -66,6 +72,8 @@ func main() {
 		controlListen:       *controlListen,
 		controlTokenFile:    *controlTokenFile,
 		enableControlWrites: *enableControlWrites,
+		fjbAgentTokenFile:   *fjbAgentTokenFile,
+		fjbAgentURLTmpl:     *fjbAgentURLTmpl,
 	}
 	if err := run(opts, log, logBus); err != nil {
 		log.Error("fatal", "err", err)
@@ -83,8 +91,17 @@ type runOpts struct {
 	controlListen       string
 	controlTokenFile    string
 	enableControlWrites bool
+	fjbAgentTokenFile   string
+	fjbAgentURLTmpl     string
 }
 
+// version is the fj-bellows daemon's build version, stamped at link
+// time via -ldflags "-X main.version=<git describe output>". It also
+// drives the version of fjbagent installed on workers via cloud-init,
+// since agent and orchestrator share a proto and must be built from
+// the same commit.
+var version = "dev"
+
 func run(opts runOpts, log *slog.Logger, logBus *logbus.Bus) error {
 	cfg, err := config.Load(opts.configPath)
 	if err != nil {
@@ -102,31 +119,16 @@ func run(opts runOpts, log *slog.Logger, logBus *logbus.Bus) error {
 
 	// SSH key is required only for providers that dispatch over SSH. A docker
 	// deployment passes nothing into cloud-init and execs into containers.
-	var (
-		signer  ssh.Signer
-		authKey string
-	)
-	if cfg.Provider != config.ProviderDocker {
-		signer, authKey, err = loadSSHKey(cfg.SSH.PrivateKeyFile)
-		if err != nil {
-			return err
-		}
+	signer, authKey, err := loadSSHKeyForProvider(cfg)
+	if err != nil {
+		return err
 	}
 
 	prov, err := provider.New(cfg.Provider)
 	if err != nil {
 		return err
 	}
-	// Hand the Linode provider the orchestrator's SSH public key so
-	// the managed cache VM (if `cache:` is set) accepts ssh from the
-	// operator for debugging. No tunnel; this is inbound-only debug
-	// access. No-op for non-Linode providers.
-	if l, ok := prov.(*linodeprov.Linode); ok && authKey != "" {
-		// ssh.MarshalAuthorizedKey appends a trailing newline; Linode's
-		// authorized_keys API rejects multi-line values with a 400.
-		// The worker Provision path already does this trim on spec.AuthorizedKey.
-		l.SetSSHAuthorizedKey(strings.TrimSpace(authKey))
-	}
+	applyAuthorizedKeyToLinodeProvider(prov, authKey)
 	// Propagate transport mode into the Linode provider so its managed
 	// firewall synthesizes the right ACCEPT rules (tcp/22 for legacy SSH,
 	// IPsec ports for cache-gateway). Duck-typed so providers that don't
@@ -179,25 +181,11 @@ func run(opts runOpts, log *slog.Logger, logBus *logbus.Bus) error {
 		return err
 	}
 
-	orch := orchestrator.New(orchestrator.Config{
-		Tag:           cfg.Tag,
-		MaxScale:      cfg.Scale.Max,
-		Labels:        cfg.Forgejo.Labels,
-		PollInterval:  cfg.Poll.Interval.D(),
-		RunnerVersion: opts.runnerVersion,
-		ReadyFile:     bootstrap.DefaultReadyFile,
-		AuthorizedKey: authKey,
-		TransportMode: cfg.Transport.Mode,
-		Teardown: orchestrator.TeardownPolicy{
-			Model:       prov.BillingModel(),
-			IdleTimeout: cfg.Poll.IdleTimeout.D(),
-			HourMargin:  cfg.Poll.HourMargin.D(),
-			BillingHour: cfg.Poll.BillingHour.D(),
-		},
-		DrainOnShutdown: opts.drain,
-		DrainTimeout:    opts.drainTimeout,
-		DestroyOnExit:   opts.destroyOnExit,
-	}, prov, fj, dispatcher, log)
+	orchCfg, err := buildOrchestratorConfig(cfg, opts, version, authKey, prov)
+	if err != nil {
+		return err
+	}
+	orch := orchestrator.New(orchCfg, prov, fj, dispatcher, log)
 
 	if err := startControlPlane(ctx, controlOpts{
 		listen:       opts.controlListen,
@@ -825,3 +813,90 @@ func acquireLock(path string) (func(), error) {
 		_ = f.Close()
 	}, nil
 }
+
+// loadFJBAgentSettings reads the fjbagent token from disk (if configured)
+// and resolves the binary download URL against the daemon's own build
+// version. Returns the empty-string pair when -fjbagent-token-file is
+// unset (agent install disabled).
+func loadFJBAgentSettings(opts runOpts, buildVersion string) (token, url string, err error) {
+	if opts.fjbAgentTokenFile == "" {
+		// Empty token file = agent install disabled; nothing to load.
+		return "", "", nil
+	}
+	tok, err := os.ReadFile(opts.fjbAgentTokenFile)
+	if err != nil {
+		return "", "", fmt.Errorf("read fjbagent token: %w", err)
+	}
+	t := strings.TrimSpace(string(tok))
+	if t == "" {
+		return "", "", fmt.Errorf("fjbagent token file %s is empty", opts.fjbAgentTokenFile)
+	}
+	resolvedURL, err := bootstrap.ResolveAgentDownloadURL(opts.fjbAgentURLTmpl, buildVersion)
+	if err != nil {
+		return "", "", fmt.Errorf("resolve fjbagent URL: %w", err)
+	}
+	return t, resolvedURL, nil
+}
+
+// buildOrchestratorConfig assembles the orchestrator.Config from the
+// loaded config + flags. Extracted from run() so the latter stays under
+// the gocyclo budget after Phase B added the FJB-94 token-load branch.
+func buildOrchestratorConfig(cfg *config.Config, opts runOpts, buildVersion, authKey string, prov provider.Provider) (orchestrator.Config, error) {
+	fjbAgentToken, fjbAgentURL, err := loadFJBAgentSettings(opts, buildVersion)
+	if err != nil {
+		return orchestrator.Config{}, err
+	}
+	_ = prov // reserved for Phase C: provider-aware SetFJBAgent wiring lives in its own helper there.
+	return orchestrator.Config{
+		Tag:                 cfg.Tag,
+		MaxScale:            cfg.Scale.Max,
+		Labels:              cfg.Forgejo.Labels,
+		PollInterval:        cfg.Poll.Interval.D(),
+		RunnerVersion:       opts.runnerVersion,
+		ReadyFile:           bootstrap.DefaultReadyFile,
+		AuthorizedKey:       authKey,
+		TransportMode:       cfg.Transport.Mode,
+		FJBAgentDownloadURL: fjbAgentURL,
+		FJBAgentToken:       fjbAgentToken,
+		Teardown: orchestrator.TeardownPolicy{
+			Model:       prov.BillingModel(),
+			IdleTimeout: cfg.Poll.IdleTimeout.D(),
+			HourMargin:  cfg.Poll.HourMargin.D(),
+			BillingHour: cfg.Poll.BillingHour.D(),
+		},
+		DrainOnShutdown: opts.drain,
+		DrainTimeout:    opts.drainTimeout,
+		DestroyOnExit:   opts.destroyOnExit,
+	}, nil
+}
+
+// applyAuthorizedKeyToLinodeProvider hands the operator's SSH authorized-key
+// line to the Linode provider so the managed cache VM accepts ssh debug
+// access. No-op for non-Linode providers, no-op when authKey is empty
+// (docker-only deployments). Extracted to keep run()'s cyclomatic
+// complexity under the gocyclo budget after the FJB-94 Phase B token-load
+// branch landed.
+func applyAuthorizedKeyToLinodeProvider(prov provider.Provider, authKey string) {
+	if authKey == "" {
+		return
+	}
+	l, ok := prov.(*linodeprov.Linode)
+	if !ok {
+		return
+	}
+	// ssh.MarshalAuthorizedKey appends a trailing newline; Linode's
+	// authorized_keys API rejects multi-line values with a 400. The
+	// worker Provision path already does this trim on spec.AuthorizedKey.
+	l.SetSSHAuthorizedKey(strings.TrimSpace(authKey))
+}
+
+// loadSSHKeyForProvider loads the SSH key only for providers that dispatch
+// over SSH; docker dispatches via container exec and needs nothing. Empty
+// returns are safe for both signer and authKey on the docker path. Extracted
+// to keep run() under the gocyclo budget after FJB-94 Phase B.
+func loadSSHKeyForProvider(cfg *config.Config) (ssh.Signer, string, error) {
+	if cfg.Provider == config.ProviderDocker {
+		return nil, "", nil
+	}
+	return loadSSHKey(cfg.SSH.PrivateKeyFile)
+}

internal/agent/sdnotify.go

@@ -0,0 +1,33 @@
+package agent
+
+import (
+	"net"
+	"os"
+)
+
+// sdNotifyReady signals systemd that the service is ready, when the unit
+// uses Type=notify. Reads NOTIFY_SOCKET from the env (set by systemd) and
+// writes "READY=1\n" to it. No-op when the env var is unset (running
+// outside systemd) so unit tests don't need to fake it.
+//
+// We inline the protocol rather than depend on github.com/coreos/go-systemd:
+// it's a Unix datagram socket and four bytes of payload. Adding the
+// transitive dependency surface isn't worth it.
+func sdNotifyReady() {
+	sock := os.Getenv("NOTIFY_SOCKET")
+	if sock == "" {
+		return
+	}
+	addr := &net.UnixAddr{Name: sock, Net: "unixgram"}
+	conn, err := net.DialUnix("unixgram", nil, addr)
+	if err != nil {
+		// systemd is responsible for cleaning up if we never report
+		// ready; on this failure we just lose the up-signal. Quiet
+		// failure is intentional — the agent should still serve
+		// requests in test/dev environments where NOTIFY_SOCKET
+		// happens to be set but unreachable.
+		return
+	}
+	defer func() { _ = conn.Close() }()
+	_, _ = conn.Write([]byte("READY=1\n"))
+}

internal/agent/server.go

@@ -96,6 +96,12 @@ func (s *Server) Run(ctx context.Context) error {
 	}
 	s.log.Info("agent listening", "addr", ln.Addr().String())
 
+	// Tell systemd we're ready as soon as the listener is bound. Under
+	// Type=notify this is what unblocks `systemctl start fjbagent` and
+	// what the orchestrator's readiness gate watches via the unit
+	// active state. No-op outside systemd.
+	sdNotifyReady()
+
 	serveErr := make(chan error, 1)
 	go func() {
 		err := s.srv.Serve(ln)

internal/bootstrap/bootstrap.go

@@ -8,16 +8,41 @@ import (
 	"encoding/base64"
 	"errors"
 	"fmt"
+	"strings"
 	"text/template"
 )
 
 //go:embed cloud-init.yaml.tmpl
 var cloudInitTemplate string
 
+//go:embed fjbagent.service
+var fjbagentServiceUnit string
+
 // DefaultReadyFile is touched by cloud-init once the worker is provisioned.
 // The orchestrator polls for it over SSH to decide a node is ready.
 const DefaultReadyFile = "/run/fj-bellows-ready"
 
+// DefaultAgentListenPort is the TCP port fjbagent binds on the worker's
+// VPC IP (cache binds the same port on its WG inner address). Exposed as
+// a constant so the orchestrator dialer and the cloud-init renderer agree
+// without an extra config knob.
+const DefaultAgentListenPort = 9001
+
+// DefaultAgentDownloadURLTemplate is the URL the orchestrator resolves
+// against its own build version (via ResolveAgentDownloadURL) before
+// passing the result into Params.FJBAgentDownloadURL. The agent is part
+// of the same fj-bellows codebase as the orchestrator — there is no
+// per-deployment choice of agent version, only a per-build one. {{.Version}}
+// resolves to the orchestrator's main.version (linker-stamped); $rarch
+// stays a shell literal so cloud-init substitutes it at boot from the
+// worker's actual architecture.
+//
+// Pre-existing workers from before an orchestrator upgrade keep running
+// the agent version they were provisioned with — proto-wire compatibility
+// between agent versions is the contract that lets the orchestrator dial
+// older workers without a forced reap.
+const DefaultAgentDownloadURLTemplate = "https://github.com/hstern/fj-bellows/releases/download/v{{.Version}}/fjbagent-linux-$rarch"
+
 // Params fills the cloud-init template.
 type Params struct {
 	// RunnerVersion is the forgejo-runner release to install, without a
@@ -31,6 +56,20 @@ type Params struct {
 	// dial is verified, eliminating the trust-on-first-use window. When empty the
 	// worker keeps the host key cloud-init generates on its own.
 	HostPrivateKey string
+
+	// FJBAgentDownloadURL is the fully-resolved URL the worker fetches the
+	// fjbagent binary from. Use ResolveAgentDownloadURL to substitute the
+	// orchestrator's version into a template first; $rarch stays a shell
+	// literal cloud-init substitutes at boot. Empty disables agent install
+	// entirely (transitional deployments).
+	FJBAgentDownloadURL string
+
+	// FJBAgentToken is the per-deployment shared-secret bearer token the
+	// agent and orchestrator agree on. Written to /etc/fjbagent/auth.token
+	// (mode 0600, owned by the fjb user) and presented by the orchestrator
+	// in the Authorization header on every Connect RPC. Required when
+	// FJBAgentDownloadURL is non-empty.
+	FJBAgentToken string
 }
 
 // Render produces the cloud-init user-data for a worker.
@@ -41,18 +80,66 @@ func Render(p Params) (string, error) {
 	if p.ReadyFile == "" {
 		p.ReadyFile = DefaultReadyFile
 	}
+	if p.FJBAgentDownloadURL != "" && p.FJBAgentToken == "" {
+		return "", errors.New("bootstrap: FJBAgentToken is required when FJBAgentDownloadURL is set")
+	}
+
+	tmplData := struct {
+		Params
+		FJBAgentServiceUnit string
+		FJBAgentListenPort  int
+	}{
+		Params:              p,
+		FJBAgentServiceUnit: fjbagentServiceUnit,
+		FJBAgentListenPort:  DefaultAgentListenPort,
+	}
+
 	funcs := template.FuncMap{
 		"b64enc": func(s string) string {
 			return base64.StdEncoding.EncodeToString([]byte(s))
 		},
+		"indent": func(spaces int, s string) string {
+			prefix := strings.Repeat(" ", spaces)
+			lines := strings.Split(strings.TrimRight(s, "\n"), "\n")
+			for i, line := range lines {
+				lines[i] = prefix + line
+			}
+			return strings.Join(lines, "\n")
+		},
 	}
 	tmpl, err := template.New("cloud-init").Funcs(funcs).Parse(cloudInitTemplate)
 	if err != nil {
 		return "", fmt.Errorf("parse cloud-init template: %w", err)
 	}
 	var buf bytes.Buffer
-	if err := tmpl.Execute(&buf, p); err != nil {
+	if err := tmpl.Execute(&buf, tmplData); err != nil {
 		return "", fmt.Errorf("render cloud-init: %w", err)
 	}
 	return buf.String(), nil
 }
+
+// ResolveAgentDownloadURL substitutes the {{.Version}} placeholder in
+// urlTemplate with the orchestrator's own build version. $rarch stays a
+// shell literal cloud-init substitutes at boot. The agent version is
+// tied to the orchestrator's build by construction — there is no
+// per-deployment version choice.
+func ResolveAgentDownloadURL(urlTemplate, version string) (string, error) {
+	if urlTemplate == "" {
+		urlTemplate = DefaultAgentDownloadURLTemplate
+	}
+	tmpl, err := template.New("agent-url").Parse(urlTemplate)
+	if err != nil {
+		return "", fmt.Errorf("parse agent download URL: %w", err)
+	}
+	var buf bytes.Buffer
+	// Arch stays as a shell variable ($rarch) the cloud-init replaces at
+	// boot time; we only resolve the version here.
+	data := struct {
+		Version string
+		Arch    string
+	}{Version: version, Arch: "$rarch"}
+	if err := tmpl.Execute(&buf, data); err != nil {
+		return "", fmt.Errorf("execute agent download URL: %w", err)
+	}
+	return buf.String(), nil
+}