feat(agent): client helper + cache cloud-init install (FJB-94 Phase C)

cmd/fj-bellows/main.go

@@ -846,7 +846,12 @@ func buildOrchestratorConfig(cfg *config.Config, opts runOpts, buildVersion, aut
 	if err != nil {
 		return orchestrator.Config{}, err
 	}
-	_ = prov // reserved for Phase C: provider-aware SetFJBAgent wiring lives in its own helper there.
+	// Phase C: push the resolved knobs into the provider via duck-typing
+	// so the managed cache renders the same install block as workers.
+	// Providers that don't implement SetFJBAgent (docker) are unaffected.
+	if a, ok := prov.(interface{ SetFJBAgent(string, string) }); ok {
+		a.SetFJBAgent(fjbAgentURL, fjbAgentToken)
+	}
 	return orchestrator.Config{
 		Tag:                 cfg.Tag,
 		MaxScale:            cfg.Scale.Max,

internal/agent/client.go

@@ -0,0 +1,118 @@
+package agent
+
+import (
+	"context"
+	"crypto/tls"
+	"net"
+	"net/http"
+	"time"
+
+	"connectrpc.com/connect"
+	"golang.org/x/net/http2"
+
+	"github.com/hstern/fj-bellows/gen/fjbellows/agent/v1/agentv1connect"
+)
+
+// DialContextFunc is the network-dial primitive a Client uses to open
+// TCP connections. The orchestrator passes wg.Tunnel.DialContext so
+// agent traffic rides the WireGuard netstack; tests pass an httptest-
+// or bufconn-backed dialer. Signature matches net.Dialer.DialContext.
+type DialContextFunc func(ctx context.Context, network, address string) (net.Conn, error)
+
+// ClientOptions configures NewClient.
+type ClientOptions struct {
+	// Addr is the agent's host:port. For workers, the worker's VPC IP +
+	// the agent listen port; for cache, the cache's WG inner address +
+	// the agent listen port.
+	Addr string
+
+	// Token is the per-deployment bearer secret presented on every RPC.
+	// Empty disables the Authorization header — useful in tests.
+	Token string
+
+	// DialContext opens TCP connections. Required.
+	DialContext DialContextFunc
+
+	// RequestTimeout caps how long a single RPC may take. Defaults to
+	// 30s when zero — long enough for a slow Health call on a busy
+	// worker, short enough that callers don't sit forever on a dead
+	// agent. Bidi-streaming RPCs (Exec) should use a per-stream
+	// context instead.
+	RequestTimeout time.Duration
+}
+
+// NewClient builds a ConnectRPC AgentServiceClient that dials through
+// the supplied DialContext and presents the bearer token. The wire is
+// HTTP/2 cleartext (h2c) — the orchestrator↔agent leg is already
+// authenticated and encrypted by WG.
+func NewClient(opts ClientOptions) agentv1connect.AgentServiceClient {
+	if opts.RequestTimeout == 0 {
+		opts.RequestTimeout = 30 * time.Second
+	}
+
+	httpClient := &http.Client{
+		Transport: &http2.Transport{
+			// AllowHTTP + DialTLSContext-on-http2 over a plain TCP
+			// dialer is how Connect speaks h2c. We pretend tls is
+			// configured but the DialTLS just opens a TCP conn.
+			AllowHTTP: true,
+			DialTLSContext: func(ctx context.Context, network, addr string, _ *tls.Config) (net.Conn, error) {
+				return opts.DialContext(ctx, network, addr)
+			},
+		},
+		Timeout: opts.RequestTimeout,
+	}
+
+	connectOpts := []connect.ClientOption{}
+	if opts.Token != "" {
+		connectOpts = append(connectOpts, connect.WithInterceptors(bearerInterceptor(opts.Token)))
+	}
+
+	return agentv1connect.NewAgentServiceClient(httpClient, "http://"+opts.Addr, connectOpts...)
+}
+
+// bearerInterceptor attaches `Authorization: Bearer <token>` to every
+// outgoing request and stream.
+func bearerInterceptor(token string) connect.Interceptor {
+	auth := "Bearer " + token
+	return interceptorFunc{
+		unary: func(next connect.UnaryFunc) connect.UnaryFunc {
+			return func(ctx context.Context, req connect.AnyRequest) (connect.AnyResponse, error) {
+				req.Header().Set("Authorization", auth)
+				return next(ctx, req)
+			}
+		},
+		streamClient: func(next connect.StreamingClientFunc) connect.StreamingClientFunc {
+			return func(ctx context.Context, spec connect.Spec) connect.StreamingClientConn {
+				conn := next(ctx, spec)
+				conn.RequestHeader().Set("Authorization", auth)
+				return conn
+			}
+		},
+	}
+}
+
+// interceptorFunc adapts plain funcs into connect.Interceptor without
+// implementing the full type for every middleware.
+type interceptorFunc struct {
+	unary        func(connect.UnaryFunc) connect.UnaryFunc
+	streamClient func(connect.StreamingClientFunc) connect.StreamingClientFunc
+}
+
+func (f interceptorFunc) WrapUnary(next connect.UnaryFunc) connect.UnaryFunc {
+	if f.unary == nil {
+		return next
+	}
+	return f.unary(next)
+}
+
+func (f interceptorFunc) WrapStreamingClient(next connect.StreamingClientFunc) connect.StreamingClientFunc {
+	if f.streamClient == nil {
+		return next
+	}
+	return f.streamClient(next)
+}
+
+func (f interceptorFunc) WrapStreamingHandler(next connect.StreamingHandlerFunc) connect.StreamingHandlerFunc {
+	return next
+}

internal/agent/client_test.go

@@ -0,0 +1,202 @@
+package agent
+
+import (
+	"context"
+	"errors"
+	"net"
+	"net/http"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"connectrpc.com/connect"
+
+	agentv1 "github.com/hstern/fj-bellows/gen/fjbellows/agent/v1"
+)
+
+// newTestServerForClient starts an in-process agent server bound to
+// 127.0.0.1:0 with HTTP/2 cleartext (h2c) so bidi-streaming RPCs work
+// without TLS — same shape as production where the wire rides WG. The
+// returned DialContext bypasses normal DNS and connects directly to
+// the bound listener, mirroring how wg.Tunnel.DialContext will route in
+// production.
+func newTestServerForClient(t *testing.T, token string) (addr string, dial DialContextFunc, teardown func()) {
+	t.Helper()
+	h := NewHandler("test", time.Now())
+	var opts []Option
+	if token != "" {
+		opts = append(opts, WithBearerToken(token))
+	}
+	srv := NewServer("127.0.0.1:0", h, nil, opts...)
+
+	var lc net.ListenConfig
+	ln, err := lc.Listen(context.Background(), "tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	var protos http.Protocols
+	protos.SetHTTP1(true)
+	protos.SetUnencryptedHTTP2(true)
+	hsrv := &http.Server{
+		Handler:           srv.Handler(),
+		Protocols:         &protos,
+		ReadHeaderTimeout: 5 * time.Second,
+	}
+	serveErr := make(chan error, 1)
+	go func() {
+		err := hsrv.Serve(ln)
+		if errors.Is(err, http.ErrServerClosed) {
+			err = nil
+		}
+		serveErr <- err
+	}()
+
+	host := ln.Addr().String()
+	dial = func(ctx context.Context, _, _ string) (net.Conn, error) {
+		var d net.Dialer
+		return d.DialContext(ctx, "tcp", host)
+	}
+
+	var once sync.Once
+	teardown = func() {
+		once.Do(func() {
+			ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+			defer cancel()
+			_ = hsrv.Shutdown(ctx)
+			<-serveErr
+		})
+	}
+	return host, dial, teardown
+}
+
+func TestClient_Health_NoAuth(t *testing.T) {
+	t.Parallel()
+	host, dial, teardown := newTestServerForClient(t, "")
+	defer teardown()
+
+	client := NewClient(ClientOptions{
+		Addr:        host,
+		DialContext: dial,
+	})
+	resp, err := client.Health(context.Background(), connect.NewRequest(&agentv1.HealthRequest{}))
+	if err != nil {
+		t.Fatalf("Health: %v", err)
+	}
+	if !resp.Msg.Ready {
+		t.Errorf("ready = false, want true")
+	}
+	if resp.Msg.BuildVersion != "test" {
+		t.Errorf("build_version = %q, want %q", resp.Msg.BuildVersion, "test")
+	}
+}
+
+func TestClient_Health_BearerToken_OK(t *testing.T) {
+	t.Parallel()
+	host, dial, teardown := newTestServerForClient(t, "s3cret")
+	defer teardown()
+
+	client := NewClient(ClientOptions{
+		Addr:        host,
+		Token:       "s3cret",
+		DialContext: dial,
+	})
+	resp, err := client.Health(context.Background(), connect.NewRequest(&agentv1.HealthRequest{}))
+	if err != nil {
+		t.Fatalf("Health: %v", err)
+	}
+	if !resp.Msg.Ready {
+		t.Errorf("ready = false, want true")
+	}
+}
+
+func TestClient_Health_WrongToken_Rejected(t *testing.T) {
+	t.Parallel()
+	host, dial, teardown := newTestServerForClient(t, "right")
+	defer teardown()
+
+	client := NewClient(ClientOptions{
+		Addr:        host,
+		Token:       "wrong",
+		DialContext: dial,
+	})
+	_, err := client.Health(context.Background(), connect.NewRequest(&agentv1.HealthRequest{}))
+	if err == nil {
+		t.Fatal("Health succeeded with wrong token, want CodeUnauthenticated")
+	}
+	if connect.CodeOf(err) != connect.CodeUnauthenticated {
+		t.Errorf("code = %v, want CodeUnauthenticated", connect.CodeOf(err))
+	}
+}
+
+// Confirm the bearer interceptor sets the header even on stream RPCs
+// (Exec is bidi; the orchestrator side will use this client for it).
+func TestClient_Exec_BearerToken_OK(t *testing.T) {
+	t.Parallel()
+	host, dial, teardown := newTestServerForClient(t, "s3cret")
+	defer teardown()
+
+	client := NewClient(ClientOptions{
+		Addr:        host,
+		Token:       "s3cret",
+		DialContext: dial,
+	})
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	stream := client.Exec(ctx)
+	defer func() { _ = stream.CloseRequest() }()
+
+	if err := stream.Send(&agentv1.ShellMsg{Kind: &agentv1.ShellMsg_Open{
+		Open: &agentv1.ExecOpen{Argv: []string{execTestArgvBin, "via-client-helper"}},
+	}}); err != nil {
+		t.Fatalf("Send Open: %v", err)
+	}
+	var got strings.Builder
+	for {
+		ev, err := stream.Receive()
+		if err != nil {
+			break
+		}
+		switch k := ev.GetKind().(type) {
+		case *agentv1.ShellEvent_Stdout:
+			got.Write(k.Stdout.GetData())
+		case *agentv1.ShellEvent_Exit:
+			if k.Exit.GetCode() != 0 {
+				t.Errorf("exit = %d, want 0", k.Exit.GetCode())
+			}
+			if want := "via-client-helper\n"; got.String() != want {
+				t.Errorf("stdout = %q, want %q", got.String(), want)
+			}
+			return
+		}
+	}
+}
+
+// Sanity: confirm we route via DialContext, not via the default dialer.
+// We pass a DialContext that errors and assert the call fails with that
+// error.
+func TestClient_UsesDialContext(t *testing.T) {
+	t.Parallel()
+	sentinel := dialSentinelError("dial sentinel reached")
+	client := NewClient(ClientOptions{
+		Addr: "host-that-should-not-be-resolved-by-default-dialer:9001",
+		DialContext: func(_ context.Context, _, _ string) (net.Conn, error) {
+			return nil, sentinel
+		},
+		RequestTimeout: 2 * time.Second,
+	})
+	_, err := client.Health(context.Background(), connect.NewRequest(&agentv1.HealthRequest{}))
+	if err == nil {
+		t.Fatal("Health succeeded; want sentinel error from DialContext")
+	}
+	if !strings.Contains(err.Error(), "dial sentinel reached") {
+		t.Errorf("expected sentinel in error, got %v", err)
+	}
+}
+
+type dialSentinelError string
+
+func (e dialSentinelError) Error() string { return string(e) }
+
+// Sanity: the helper does not import http blindly.
+var _ = http.NoBody

internal/agent/exec_test.go

@@ -17,6 +17,10 @@ import (
 	"github.com/hstern/fj-bellows/gen/fjbellows/agent/v1/agentv1connect"
 )
 
+// execTestArgvBin is the binary used across exec tests. Extracted so
+// goconst doesn't complain about repeated string literals.
+const execTestArgvBin = "echo"
+
 // execTestSetup spins up an in-process agent server reachable via H2C
 // (Connect bidi streaming needs HTTP/2). Returns a connected client and
 // a teardown.
@@ -86,7 +90,7 @@ func TestExec_Echo(t *testing.T) {
 	defer teardown()
 
 	out, errOut, code, sig, err := runExec(t, client,
-		&agentv1.ExecOpen{Argv: []string{"echo", "hello"}}, nil)
+		&agentv1.ExecOpen{Argv: []string{execTestArgvBin, "hello"}}, nil)
 	if err != nil {
 		t.Fatalf("Exec: %v", err)
 	}
@@ -190,7 +194,7 @@ func TestExec_RejectsTTY(t *testing.T) {
 	defer teardown()
 
 	_, _, _, _, err := runExec(t, client,
-		&agentv1.ExecOpen{Argv: []string{"echo"}, Tty: true}, nil)
+		&agentv1.ExecOpen{Argv: []string{execTestArgvBin}, Tty: true}, nil)
 	if err == nil {
 		t.Fatal("Exec with tty=true succeeded, want CodeUnimplemented")
 	}
@@ -300,7 +304,7 @@ func TestExec_BearerAuth(t *testing.T) {
 	stream := client.Exec(context.Background())
 	defer func() { _ = stream.CloseRequest() }()
 	_ = stream.Send(&agentv1.ShellMsg{Kind: &agentv1.ShellMsg_Open{
-		Open: &agentv1.ExecOpen{Argv: []string{"echo"}},
+		Open: &agentv1.ExecOpen{Argv: []string{execTestArgvBin}},
 	}})
 	_, err := stream.Receive()
 	if err == nil {

internal/bootstrap/bootstrap.go

@@ -18,6 +18,12 @@ var cloudInitTemplate string
 //go:embed fjbagent.service
 var fjbagentServiceUnit string
 
+// FJBAgentServiceUnit returns the embedded fjbagent systemd unit so
+// other cloud-init renderers (the linode cache, etc.) can write the
+// same unit without duplicating the file. Returning a copy keeps the
+// embedded variable read-only to callers.
+func FJBAgentServiceUnit() string { return fjbagentServiceUnit }
+
 // 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"