refactor: move EKS gateway publish helper to the EKS agent tree

Move the eks-gateway-publish implementation to agents/eks/gatewaypublish behind Run(ctx, Config) error. The binary keeps flag definitions, environment defaults, parsing, its log prefix and exit code; each former fatal site returns an error with the same text. The retry wait now also honours context cancellation, which the binary never triggers.
2026-10-15 17:11:30 +00:00 · 2026-10-07 11:06:35 +10:00
parent 7ca1cc91ea
commit fecf0b28a3
3 changed files with 263 additions and 108 deletions
@@ -1,13 +1,6 @@
 // eks-gateway-publish runs inside the EKS K3s control-plane VM and relays a
-// single publication to the host through the AWS gateway, instead of dialing
-// core NATS directly.
-//
-// It reads a JSON payload on stdin, wraps it as
-// {accountId, channel, kind, payload}, SigV4-signs (service "eks") an HTTPS
-// POST to {gateway}/clusters/{cluster}/internal-publish, and retries with
-// backoff until the gateway returns 2xx or the attempt budget is exhausted —
-// so a degraded link surfaces as a non-zero exit rather than a silently
-// dropped message (the failure mode of fire-and-forget `nats pub`).
+// single JSON publication from stdin to the host through the AWS gateway; the
+// implementation is in agents/eks/gatewaypublish.
 //
 // Usage:
 //
@@ -20,115 +13,31 @@
 package main

 import (
-	"bytes"
-	"encoding/json"
+	"context"
 	"flag"
-	"fmt"
-	"io"
 	"log/slog"
 	"os"
-	"time"
-
-	"github.com/mulgadc/spinifex/internal/eksgw"

 	_ "github.com/mulgadc/bluebottle/pkg/fipsboot"
-)

-const (
-	// Retry budget: the control plane reaches readiness before this runs, so a
-	// failing POST means a degraded link, not a cold start. Bounded so a stuck
-	// boot still terminates the OpenRC service.
-	maxAttempts = 30
-	retryDelay  = 5 * time.Second
+	"github.com/mulgadc/spinifex/spinifex/agents/eks/gatewaypublish"
 )

-type publishBody struct {
-	AccountID string          `json:"accountId"`
-	Channel   string          `json:"channel"`
-	Kind      string          `json:"kind"`
-	Payload   json.RawMessage `json:"payload"`
-}
-
 func main() {
-	var (
-		gatewayURL  string
-		gatewayCA   string
-		accessKey   string
-		secretKey   string
-		region      string
-		accountID   string
-		clusterName string
-		channel     string
-		kind        string
-	)
-
-	flag.StringVar(&gatewayURL, "gateway", os.Getenv("EKS_GATEWAY_URL"), "Gateway URL (e.g. https://10.15.8.1:9999)")
-	flag.StringVar(&gatewayCA, "gateway-ca", os.Getenv("EKS_GATEWAY_CA"), "Path to gateway TLS CA PEM (optional; falls back to system trust)")
-	flag.StringVar(&accessKey, "access-key", os.Getenv("EKS_ACCESS_KEY"), "AWS access key ID")
-	flag.StringVar(&secretKey, "secret-key", os.Getenv("EKS_SECRET_KEY"), "AWS secret access key")
-	flag.StringVar(&region, "region", os.Getenv("EKS_REGION"), "AWS region for SigV4 signing")
-	flag.StringVar(&accountID, "account-id", os.Getenv("EKS_ACCOUNT_ID"), "Cluster account ID")
-	flag.StringVar(&clusterName, "cluster", os.Getenv("EKS_CLUSTER_NAME"), "Cluster name")
-	flag.StringVar(&channel, "channel", "", "Publish channel: bootstrap|state|addon")
-	flag.StringVar(&kind, "kind", "", "Bootstrap subject kind (bootstrap channel only)")
+	cfg := gatewaypublish.Config{Payload: os.Stdin}
+	flag.StringVar(&cfg.GatewayURL, "gateway", os.Getenv("EKS_GATEWAY_URL"), "Gateway URL (e.g. https://10.15.8.1:9999)")
+	flag.StringVar(&cfg.GatewayCA, "gateway-ca", os.Getenv("EKS_GATEWAY_CA"), "Path to gateway TLS CA PEM (optional; falls back to system trust)")
+	flag.StringVar(&cfg.AccessKey, "access-key", os.Getenv("EKS_ACCESS_KEY"), "AWS access key ID")
+	flag.StringVar(&cfg.SecretKey, "secret-key", os.Getenv("EKS_SECRET_KEY"), "AWS secret access key")
+	flag.StringVar(&cfg.Region, "region", os.Getenv("EKS_REGION"), "AWS region for SigV4 signing")
+	flag.StringVar(&cfg.AccountID, "account-id", os.Getenv("EKS_ACCOUNT_ID"), "Cluster account ID")
+	flag.StringVar(&cfg.ClusterName, "cluster", os.Getenv("EKS_CLUSTER_NAME"), "Cluster name")
+	flag.StringVar(&cfg.Channel, "channel", "", "Publish channel: bootstrap|state|addon")
+	flag.StringVar(&cfg.Kind, "kind", "", "Bootstrap subject kind (bootstrap channel only)")
 	flag.Parse()

-	switch {
-	case accountID == "":
-		fatal("--account-id is required (or set EKS_ACCOUNT_ID)")
-	case clusterName == "":
-		fatal("--cluster is required (or set EKS_CLUSTER_NAME)")
-	case channel != "bootstrap" && channel != "state" && channel != "addon":
-		fatal("--channel must be bootstrap, state or addon")
-	case channel == "bootstrap" && kind == "":
-		fatal("--kind is required for the bootstrap channel")
+	if err := gatewaypublish.Run(context.Background(), cfg); err != nil {
+		slog.Error("eks-gateway-publish: " + err.Error())
+		os.Exit(1)
 	}
-
-	payload, err := io.ReadAll(os.Stdin)
-	if err != nil {
-		fatal(fmt.Sprintf("read stdin payload: %v", err))
-	}
-	if len(bytes.TrimSpace(payload)) == 0 {
-		fatal("empty stdin payload")
-	}
-	if !json.Valid(payload) {
-		fatal("stdin payload is not valid JSON")
-	}
-
-	body, err := json.Marshal(publishBody{
-		AccountID: accountID,
-		Channel:   channel,
-		Kind:      kind,
-		Payload:   json.RawMessage(payload),
-	})
-	if err != nil {
-		fatal(fmt.Sprintf("marshal request body: %v", err))
-	}
-
-	client, err := eksgw.New(gatewayURL, gatewayCA, accessKey, secretKey, region)
-	if err != nil {
-		fatal(fmt.Sprintf("build gateway client: %v", err))
-	}
-	path := "/clusters/" + clusterName + "/internal-publish"
-
-	var lastErr error
-	for attempt := 1; attempt <= maxAttempts; attempt++ {
-		if _, err := client.Post(path, body); err != nil {
-			lastErr = err
-			slog.Warn("eks-gateway-publish: attempt failed",
-				"channel", channel, "kind", kind, "attempt", attempt, "err", err)
-			if attempt < maxAttempts {
-				time.Sleep(retryDelay)
-			}
-			continue
-		}
-		slog.Info("eks-gateway-publish: published", "channel", channel, "kind", kind)
-		return
-	}
-	fatal(fmt.Sprintf("publish failed after %d attempts: %v", maxAttempts, lastErr))
-}
-
-func fatal(msg string) {
-	slog.Error("eks-gateway-publish: " + msg)
-	os.Exit(1)
 }
@@ -0,0 +1,115 @@
+// Package gatewaypublish implements eks-gateway-publish, which runs inside the
+// EKS K3s control-plane VM and relays a single publication to the host through
+// the AWS gateway, instead of dialing core NATS directly.
+//
+// It reads a JSON payload, wraps it as {accountId, channel, kind, payload},
+// SigV4-signs (service "eks") an HTTPS POST to
+// {gateway}/clusters/{cluster}/internal-publish, and retries with backoff until
+// the gateway returns 2xx or the attempt budget is exhausted — so a degraded
+// link surfaces as an error rather than a silently dropped message (the failure
+// mode of fire-and-forget `nats pub`).
+package gatewaypublish
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"log/slog"
+	"time"
+
+	"github.com/mulgadc/spinifex/internal/eksgw"
+)
+
+// Retry budget: the control plane reaches readiness before this runs, so a
+// failing POST means a degraded link, not a cold start. Bounded so a stuck
+// boot still terminates the OpenRC service.
+const maxAttempts = 30
+
+var retryDelay = 5 * time.Second
+
+// Config is one publication and the gateway it is sent through.
+type Config struct {
+	GatewayURL  string
+	GatewayCA   string
+	AccessKey   string
+	SecretKey   string
+	Region      string
+	AccountID   string
+	ClusterName string
+	Channel     string
+	Kind        string
+	// Payload supplies the JSON document to publish.
+	Payload io.Reader
+}
+
+type publishBody struct {
+	AccountID string          `json:"accountId"`
+	Channel   string          `json:"channel"`
+	Kind      string          `json:"kind"`
+	Payload   json.RawMessage `json:"payload"`
+}
+
+// Run validates cfg, reads the payload and publishes it, retrying until the
+// gateway accepts it, the attempt budget is spent, or ctx is cancelled.
+func Run(ctx context.Context, cfg Config) error {
+	switch {
+	case cfg.AccountID == "":
+		return errors.New("--account-id is required (or set EKS_ACCOUNT_ID)")
+	case cfg.ClusterName == "":
+		return errors.New("--cluster is required (or set EKS_CLUSTER_NAME)")
+	case cfg.Channel != "bootstrap" && cfg.Channel != "state" && cfg.Channel != "addon":
+		return errors.New("--channel must be bootstrap, state or addon")
+	case cfg.Channel == "bootstrap" && cfg.Kind == "":
+		return errors.New("--kind is required for the bootstrap channel")
+	}
+
+	payload, err := io.ReadAll(cfg.Payload)
+	if err != nil {
+		return fmt.Errorf("read stdin payload: %w", err)
+	}
+	if len(bytes.TrimSpace(payload)) == 0 {
+		return errors.New("empty stdin payload")
+	}
+	if !json.Valid(payload) {
+		return errors.New("stdin payload is not valid JSON")
+	}
+
+	body, err := json.Marshal(publishBody{
+		AccountID: cfg.AccountID,
+		Channel:   cfg.Channel,
+		Kind:      cfg.Kind,
+		Payload:   json.RawMessage(payload),
+	})
+	if err != nil {
+		return fmt.Errorf("marshal request body: %w", err)
+	}
+
+	client, err := eksgw.New(cfg.GatewayURL, cfg.GatewayCA, cfg.AccessKey, cfg.SecretKey, cfg.Region)
+	if err != nil {
+		return fmt.Errorf("build gateway client: %w", err)
+	}
+	path := "/clusters/" + cfg.ClusterName + "/internal-publish"
+
+	var lastErr error
+	for attempt := 1; attempt <= maxAttempts; attempt++ {
+		if _, err := client.Post(path, body); err != nil {
+			lastErr = err
+			slog.Warn("eks-gateway-publish: attempt failed",
+				"channel", cfg.Channel, "kind", cfg.Kind, "attempt", attempt, "err", err)
+			if attempt < maxAttempts {
+				select {
+				case <-ctx.Done():
+					return ctx.Err()
+				case <-time.After(retryDelay):
+				}
+			}
+			continue
+		}
+		slog.Info("eks-gateway-publish: published", "channel", cfg.Channel, "kind", cfg.Kind)
+		return nil
+	}
+	return fmt.Errorf("publish failed after %d attempts: %w", maxAttempts, lastErr)
+}
@@ -0,0 +1,131 @@
+package gatewaypublish
+
+import (
+	"context"
+	"encoding/json"
+	"encoding/pem"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+type errReader struct{}
+
+func (errReader) Read([]byte) (int, error) { return 0, errors.New("boom") }
+
+// gateway starts a TLS gateway answering every POST with status, and returns a
+// config pointing at it with the CA written to disk.
+func gateway(t *testing.T, status int, got *publishBody, hits *atomic.Int32) Config {
+	t.Helper()
+	srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		hits.Add(1)
+		if r.URL.Path != "/clusters/demo/internal-publish" {
+			w.WriteHeader(http.StatusNotFound)
+			return
+		}
+		if got != nil {
+			b, _ := io.ReadAll(r.Body)
+			_ = json.Unmarshal(b, got)
+		}
+		w.WriteHeader(status)
+	}))
+	t.Cleanup(srv.Close)
+	ca := filepath.Join(t.TempDir(), "ca.pem")
+	if err := os.WriteFile(ca, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: srv.Certificate().Raw}), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	return Config{
+		GatewayURL: srv.URL, GatewayCA: ca, AccessKey: "AKIATEST", SecretKey: "secret",
+		Region: "us-east-1", AccountID: "123456789012", ClusterName: "demo", Channel: "state",
+		Payload: strings.NewReader(`{"ok":true}`),
+	}
+}
+
+func TestRun_RejectsBeforeContactingGateway(t *testing.T) {
+	valid := Config{AccountID: "123456789012", ClusterName: "demo", Channel: "state", Payload: strings.NewReader(`{}`)}
+	cases := []struct {
+		name string
+		mod  func(*Config)
+		want string
+	}{
+		{"account", func(c *Config) { c.AccountID = "" }, "--account-id is required (or set EKS_ACCOUNT_ID)"},
+		{"cluster", func(c *Config) { c.ClusterName = "" }, "--cluster is required (or set EKS_CLUSTER_NAME)"},
+		{"channel", func(c *Config) { c.Channel = "events" }, "--channel must be bootstrap, state or addon"},
+		{"bootstrap kind", func(c *Config) { c.Channel = "bootstrap" }, "--kind is required for the bootstrap channel"},
+		{"read error", func(c *Config) { c.Payload = errReader{} }, "read stdin payload: boom"},
+		{"empty", func(c *Config) { c.Payload = strings.NewReader(" \n") }, "empty stdin payload"},
+		{"invalid json", func(c *Config) { c.Payload = strings.NewReader("{") }, "stdin payload is not valid JSON"},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			cfg := valid
+			tc.mod(&cfg)
+			err := Run(context.Background(), cfg)
+			if err == nil || err.Error() != tc.want {
+				t.Fatalf("Run error = %v, want %q", err, tc.want)
+			}
+		})
+	}
+}
+
+func TestRun_PublishesWrappedPayload(t *testing.T) {
+	var got publishBody
+	var hits atomic.Int32
+	cfg := gateway(t, http.StatusOK, &got, &hits)
+	cfg.Channel, cfg.Kind = "bootstrap", "k3s-bootstrap-token"
+
+	if err := Run(context.Background(), cfg); err != nil {
+		t.Fatalf("Run: %v", err)
+	}
+	if hits.Load() != 1 {
+		t.Fatalf("gateway hit %d times, want 1", hits.Load())
+	}
+	if got.AccountID != "123456789012" || got.Channel != "bootstrap" || got.Kind != "k3s-bootstrap-token" || string(got.Payload) != `{"ok":true}` {
+		t.Fatalf("published body = %+v", got)
+	}
+}
+
+func TestRun_FailsAfterAttemptBudget(t *testing.T) {
+	prev := retryDelay
+	retryDelay = 0
+	t.Cleanup(func() { retryDelay = prev })
+	var hits atomic.Int32
+	cfg := gateway(t, http.StatusServiceUnavailable, nil, &hits)
+
+	err := Run(context.Background(), cfg)
+	if err == nil || !strings.HasPrefix(err.Error(), "publish failed after 30 attempts: ") {
+		t.Fatalf("Run error = %v, want attempt-budget failure", err)
+	}
+	if hits.Load() != maxAttempts {
+		t.Fatalf("gateway hit %d times, want %d", hits.Load(), maxAttempts)
+	}
+}
+
+func TestRun_StopsRetryingWhenCancelled(t *testing.T) {
+	prev := retryDelay
+	retryDelay = time.Hour
+	t.Cleanup(func() { retryDelay = prev })
+	var hits atomic.Int32
+	cfg := gateway(t, http.StatusServiceUnavailable, nil, &hits)
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		for hits.Load() == 0 {
+			time.Sleep(time.Millisecond)
+		}
+		cancel()
+	}()
+
+	if err := Run(ctx, cfg); !errors.Is(err, context.Canceled) {
+		t.Fatalf("Run error = %v, want context.Canceled", err)
+	}
+	if hits.Load() != 1 {
+		t.Fatalf("gateway hit %d times, want 1", hits.Load())
+	}
+}