diff --git a/cmd/eks-gateway-publish/main.go b/cmd/eks-gateway-publish/main.go index d85d38185..ffcc6aec8 100644 --- a/cmd/eks-gateway-publish/main.go +++ b/cmd/eks-gateway-publish/main.go @@ -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(®ion, "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) } diff --git a/spinifex/agents/eks/gatewaypublish/publish.go b/spinifex/agents/eks/gatewaypublish/publish.go new file mode 100644 index 000000000..e83d1d138 --- /dev/null +++ b/spinifex/agents/eks/gatewaypublish/publish.go @@ -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) +} diff --git a/spinifex/agents/eks/gatewaypublish/publish_test.go b/spinifex/agents/eks/gatewaypublish/publish_test.go new file mode 100644 index 000000000..1440857ed --- /dev/null +++ b/spinifex/agents/eks/gatewaypublish/publish_test.go @@ -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()) + } +}