feat: config-bounded targeting query (ADR 0011, slice 6b)

cmd/agent/main.go

@@ -19,6 +19,7 @@ import (
 	"flag"
 	"fmt"
 	"log/slog"
+	"net/http"
 	"os"
 	"os/signal"
 	"strings"
@@ -40,6 +41,7 @@ import (
 	"github.com/RebuildStackCo/runtime-agent/internal/nodescan"
 	"github.com/RebuildStackCo/runtime-agent/internal/rollup"
 	"github.com/RebuildStackCo/runtime-agent/internal/sink"
+	"github.com/RebuildStackCo/runtime-agent/internal/targeting"
 )
 
 // version is set at build time via -ldflags.
@@ -212,9 +214,21 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac
 			)
 		}
 	}
+	// The targeting publisher (ADR 0011 §3, §6b) ranks eligible workloads by
+	// consumption from each usage snapshot and publishes the top N for the node
+	// targets query. Created only when profiling is enabled; the snapshot
+	// callback feeds it deep-copied records, so it never races the Accumulator.
+	var targetsPublisher *targeting.Publisher
+	if cfg.Profiling.Enabled {
+		pc := cfg.Profiling.Normalized()
+		targetsPublisher = targeting.NewPublisher(cfg.Profiling.EligibleNamespaces, cfg.Profiling.EligibleWorkloads, pc.TopN)
+	}
 	usagePoller := collector.NewUsagePoller(clientset, nodeWatcher.Names, podWatcher,
 		func(sequence int64, records []*rollup.Record) {
 			logRecords("usage rollup snapshot", sequence, records)
+			if targetsPublisher != nil {
+				targetsPublisher.Publish(records)
+			}
 			if spool != nil {
 				if err := spool.WriteUsageSnapshot(sequence, records); err != nil {
 					logger.Error("spooling usage snapshot", "error", err)
@@ -368,7 +382,13 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac
 				"profiles_received", n)
 		}
 
-		intake, err := buildNodeIntake(logger, restConfig, cfg.NodeIntake, onReport, onProfile)
+		// Pass the publisher as a nil-safe TargetSource: a typed nil pointer
+		// would make a non-nil interface, so only assign when it exists.
+		var targetSource nodeintake.TargetSource
+		if targetsPublisher != nil {
+			targetSource = targetsPublisher
+		}
+		intake, err := buildNodeIntake(logger, restConfig, cfg.NodeIntake, onReport, onProfile, targetSource)
 		if err != nil {
 			return fmt.Errorf("configuring node intake: %w", err)
 		}
@@ -403,7 +423,7 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac
 // by the cluster JWKS (fetched over the API server's in-cluster transport, no
 // TokenReview) and an HTTP server that decodes node reports and hands each to
 // onReport (which joins it into the Go inventory, ADR 0010).
-func buildNodeIntake(logger *slog.Logger, restConfig *rest.Config, cfg config.NodeIntake, onReport func(nodeauth.Identity, nodescan.Report), onProfile func(nodeauth.Identity, nodeintake.ProfileReport)) (*nodeintake.Server, error) {
+func buildNodeIntake(logger *slog.Logger, restConfig *rest.Config, cfg config.NodeIntake, onReport func(nodeauth.Identity, nodescan.Report), onProfile func(nodeauth.Identity, nodeintake.ProfileReport), targets nodeintake.TargetSource) (*nodeintake.Server, error) {
 	addr := cfg.ListenAddress
 	if addr == "" {
 		addr = config.DefaultNodeIntakeListenAddress
@@ -424,8 +444,15 @@ func buildNodeIntake(logger *slog.Logger, restConfig *rest.Config, cfg config.No
 
 	handler := nodeintake.NewHandler(verifier, logger, onReport)
 	profileHandler := nodeintake.NewProfileHandler(verifier, logger, onProfile)
-	logger.Info("node intake enabled", "addr", addr, "audience", audience, "expected_subject", cfg.ExpectedSubject)
-	return nodeintake.NewServer(addr, handler, profileHandler, logger), nil
+	// The targets query endpoint is mounted only when a source is configured
+	// (profiling enabled); a nil interface leaves the route absent.
+	var targetsHandler http.Handler
+	if targets != nil {
+		targetsHandler = nodeintake.NewTargetsHandler(verifier, logger, targets)
+	}
+	logger.Info("node intake enabled", "addr", addr, "audience", audience,
+		"expected_subject", cfg.ExpectedSubject, "targeting", targets != nil)
+	return nodeintake.NewServer(addr, handler, profileHandler, targetsHandler, logger), nil
 }
 
 // podContainerResolver adapts PodWatcher's container lookup to the inventory

cmd/agent/nodeship.go

@@ -108,6 +108,57 @@ func readProjectedToken(path string) (string, error) {
 	return token, nil
 }
 
+// targetsClient asks the controller which workloads to profile (ADR 0011 §3).
+// Unlike the shippers, this reply carries data — the config-bounded top-N list;
+// the node intersects it with its own pods and re-checks it against the eligible
+// set before acting, so a controller can only prioritize within what the node's
+// config already permits.
+type targetsClient struct {
+	endpoint  string
+	tokenPath string
+	client    *http.Client
+}
+
+// newTargetsClient builds a client for endpoint, or returns nil when empty.
+func newTargetsClient(endpoint, tokenPath string) *targetsClient {
+	if endpoint == "" {
+		return nil
+	}
+	return &targetsClient{
+		endpoint:  endpoint,
+		tokenPath: tokenPath,
+		client:    &http.Client{Timeout: 30 * time.Second},
+	}
+}
+
+// fetch queries the controller and returns the published targets.
+func (c *targetsClient) fetch(ctx context.Context) ([]nodeintake.Target, error) {
+	token, err := readProjectedToken(c.tokenPath)
+	if err != nil {
+		return nil, fmt.Errorf("reading controller token: %w", err)
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.endpoint, nil) // #nosec G704 -- endpoint is operator-set config, not tainted input
+	if err != nil {
+		return nil, fmt.Errorf("building request: %w", err)
+	}
+	req.Header.Set("Authorization", "Bearer "+token)
+
+	resp, err := c.client.Do(req) // #nosec G704 -- endpoint is operator-set config, not tainted input
+	if err != nil {
+		return nil, fmt.Errorf("querying targets: %w", err)
+	}
+	defer func() { _ = resp.Body.Close() }()
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 4<<10))
+		return nil, fmt.Errorf("controller rejected targets query: status %d", resp.StatusCode)
+	}
+	var out nodeintake.TargetsResponse
+	if err := json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(&out); err != nil {
+		return nil, fmt.Errorf("decoding targets: %w", err)
+	}
+	return out.Targets, nil
+}
+
 // profileShipper delivers one captured eBPF CPU profile to the controller's
 // profile endpoint (ADR 0011 §5.5), over the same node-initiated, projected-
 // token transport as the inventory shipper. Only already allow-list-filtered

cmd/agent/nodeship_targets_test.go

@@ -0,0 +1,53 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/RebuildStackCo/runtime-agent/internal/nodeintake"
+)
+
+func TestTargetsClientFetch(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodPost {
+			t.Errorf("method = %s, want POST", r.Method)
+		}
+		if got := r.Header.Get("Authorization"); got != "Bearer tok" {
+			t.Errorf("auth = %q, want Bearer tok", got)
+		}
+		w.Header().Set("Content-Type", "application/json")
+		_ = json.NewEncoder(w).Encode(nodeintake.TargetsResponse{
+			Targets: []nodeintake.Target{{Namespace: "shop", WorkloadKind: "Deployment", WorkloadName: "web"}},
+		})
+	}))
+	t.Cleanup(srv.Close)
+
+	c := newTargetsClient(srv.URL, writeToken(t, "tok"))
+	targets, err := c.fetch(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(targets) != 1 || targets[0].WorkloadName != "web" {
+		t.Errorf("targets = %+v, want [web]", targets)
+	}
+}
+
+func TestTargetsClientNilWhenNoEndpoint(t *testing.T) {
+	if newTargetsClient("", "/x") != nil {
+		t.Error("expected nil client for empty endpoint")
+	}
+}
+
+func TestTargetsClientRejectsNon2xx(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusUnauthorized)
+	}))
+	t.Cleanup(srv.Close)
+	c := newTargetsClient(srv.URL, writeToken(t, "tok"))
+	if _, err := c.fetch(context.Background()); err == nil {
+		t.Error("expected error on non-2xx response")
+	}
+}

internal/nodeintake/server.go

@@ -14,6 +14,7 @@ import (
 const (
 	reportPath  = "/v1/node-inventory"
 	profilePath = "/v1/node-profile"
+	targetsPath = "/v1/node-targets"
 )
 
 // Server runs the node-intake HTTP receiver until its context is canceled.
@@ -25,16 +26,22 @@ type Server struct {
 	logger  *slog.Logger
 }
 
-// NewServer wires a receiver listening on addr that dispatches the inventory
-// path to inventory and, when non-nil, the profile path to profile. addr follows
-// net/http conventions (e.g. ":8080"). Both endpoints share the one port, the
-// one token audience, and the one NetworkPolicy.
-func NewServer(addr string, inventory, profile http.Handler, logger *slog.Logger) *Server {
+// NewServer wires a receiver listening on addr. It always dispatches the
+// inventory path to inventory, and mounts the profile and targets paths when
+// their handlers are non-nil. addr follows net/http conventions (e.g. ":8080").
+// All endpoints share the one port, the one token audience, and the one
+// NetworkPolicy. The inventory and profile endpoints are one-way pushes (the
+// reply is an ack); the targets endpoint is the config-bounded query whose reply
+// carries data (ADR 0011 §3).
+func NewServer(addr string, inventory, profile, targets http.Handler, logger *slog.Logger) *Server {
 	mux := http.NewServeMux()
 	mux.Handle(reportPath, inventory)
 	if profile != nil {
 		mux.Handle(profilePath, profile)
 	}
+	if targets != nil {
+		mux.Handle(targetsPath, targets)
+	}
 	return &Server{addr: addr, handler: mux, logger: logger}
 }
 

internal/nodeintake/targets.go

@@ -0,0 +1,77 @@
+package nodeintake
+
+import (
+	"encoding/json"
+	"log/slog"
+	"net/http"
+)
+
+// Target names one workload the controller suggests the node profile. It carries
+// identifiers only — nothing the node executes.
+type Target struct {
+	Namespace    string `json:"namespace"`
+	WorkloadKind string `json:"workload_kind"`
+	WorkloadName string `json:"workload_name"`
+}
+
+// TargetsResponse is the reply to a node's targets query.
+type TargetsResponse struct {
+	Targets []Target `json:"targets"`
+}
+
+// TargetSource yields the currently published top-N targets. The controller's
+// publisher implements it by reading an atomically published snapshot — never
+// the live rollup state (ADR 0011 §6b concurrency note).
+type TargetSource interface {
+	Snapshot() []Target
+}
+
+// TargetsHandler answers a node's targets query with the top-N workloads by
+// consumption.
+//
+// This is the one node↔controller endpoint whose reply carries data the node
+// acts on, and it is a deliberate, config-bounded exception to the one-way reply
+// discipline of ADR 0010 §1 (ADR 0011 §3). It does not break invariant 1: the
+// reply is data derived from the cluster's own rollups (workload identifiers),
+// never configuration or a command; the external backend is untouched; and the
+// node's ConfigMap — not this reply — bounds what may be profiled and how much.
+// The published snapshot is already filtered to the eligible set and capped at
+// TopN, and the node re-checks eligibility before it acts. The worst a rogue
+// controller can do is reorder already-permitted targets within already-set
+// ceilings.
+type TargetsHandler struct {
+	verifier TokenVerifier
+	source   TargetSource
+	logger   *slog.Logger
+}
+
+// NewTargetsHandler builds the targets query handler.
+func NewTargetsHandler(verifier TokenVerifier, logger *slog.Logger, source TargetSource) *TargetsHandler {
+	return &TargetsHandler{verifier: verifier, source: source, logger: logger}
+}
+
+// ServeHTTP authenticates the caller and returns the published targets. The
+// query is parameterless — the top-N is cluster-wide — so no request body is
+// read; the node intersects the reply with its own pods locally.
+func (h *TargetsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		w.Header().Set("Allow", http.MethodPost)
+		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+	token, ok := bearerToken(r)
+	if !ok {
+		http.Error(w, "missing bearer token", http.StatusUnauthorized)
+		return
+	}
+	if _, err := h.verifier.Verify(r.Context(), token); err != nil {
+		h.logger.Warn("targets query rejected: token verification failed", "error", err)
+		http.Error(w, "unauthorized", http.StatusUnauthorized)
+		return
+	}
+
+	resp := TargetsResponse{Targets: h.source.Snapshot()}
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(http.StatusOK)
+	_ = json.NewEncoder(w).Encode(resp)
+}

internal/nodeintake/targets_test.go

@@ -0,0 +1,53 @@
+package nodeintake
+
+import (
+	"encoding/json"
+	"io"
+	"log/slog"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/RebuildStackCo/runtime-agent/internal/nodeauth"
+)
+
+type fakeSource struct{ targets []Target }
+
+func (f fakeSource) Snapshot() []Target { return f.targets }
+
+func testTargetsHandler(t *testing.T, src TargetSource) *TargetsHandler {
+	t.Helper()
+	v := fakeVerifier{accept: goodToken, identity: nodeauth.Identity{Subject: "sub"}}
+	return NewTargetsHandler(v, slog.New(slog.NewTextHandler(io.Discard, nil)), src)
+}
+
+func TestTargetsHandlerReturnsSnapshot(t *testing.T) {
+	src := fakeSource{targets: []Target{{Namespace: "shop", WorkloadKind: "Deployment", WorkloadName: "web"}}}
+	rec := post(t, testTargetsHandler(t, src), goodToken, "")
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status = %d, want 200; body %s", rec.Code, rec.Body.String())
+	}
+	var resp TargetsResponse
+	if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil {
+		t.Fatal(err)
+	}
+	if len(resp.Targets) != 1 || resp.Targets[0].WorkloadName != "web" {
+		t.Errorf("targets = %+v, want [web]", resp.Targets)
+	}
+}
+
+func TestTargetsHandlerRejectsBadToken(t *testing.T) {
+	if rec := post(t, testTargetsHandler(t, fakeSource{}), "wrong", ""); rec.Code != http.StatusUnauthorized {
+		t.Errorf("status = %d, want 401", rec.Code)
+	}
+}
+
+func TestTargetsHandlerRejectsGet(t *testing.T) {
+	req := httptest.NewRequest(http.MethodGet, targetsPath, nil)
+	req.Header.Set("Authorization", "Bearer "+goodToken)
+	rec := httptest.NewRecorder()
+	testTargetsHandler(t, fakeSource{}).ServeHTTP(rec, req)
+	if rec.Code != http.StatusMethodNotAllowed {
+		t.Errorf("status = %d, want 405", rec.Code)
+	}
+}