From de11cb45dadd02144e695c76a37319154e046d0c Mon Sep 17 00:00:00 2001 From: Peter Sprygada Date: Wed, 12 Aug 2026 13:44:12 -0400 Subject: [PATCH 1/2] feat(gateway): add galactic-gateway binary Introduces galactic-gateway as its own binary, hosting the edge XDP NAT+LB gateway engine (previous two branches) and the NetworkGateway/NetworkRule reconcilers (previous branch) separately from galactic-router's tenant BGP process, so a crash on either side no longer takes the other down with it. Tenant BGP (GoBGP + the BGPRouter/BGPPeer/BGPAdvertisement/BGPPolicy/BGPVRFInstance reconcilers) still runs in galactic-router, co-located on the same gateway node. - cmd/galactic-gateway/main.go, root.go: manager setup, flags/env config, RBAC pre-flight -- no BGP runtime/RuntimeManager/BGP-family reconciler here at all. - cmd/galactic-gateway/gateway.go: setupGatewayDatapath, the load/attach/wire-up entry point, moved unchanged from cmd/galactic-router/gateway.go. - internal/config/gateway.go: GatewayConfig (metrics/grpc-health ports distinct from every other galactic-* process on the same hostNetwork node, node name, public interface, SRv6 address). - Taskfile.yaml: builds bin/galactic-gateway alongside the other binaries. Fourth branch in the edge-gateway stack; builds on feat/edge-gateway-03-controllers. --- Taskfile.yaml | 1 + cmd/galactic-gateway/gateway.go | 116 +++++++++++++++++ cmd/galactic-gateway/gateway_test.go | 24 ++++ cmd/galactic-gateway/main.go | 104 +++++++++++++++ cmd/galactic-gateway/root.go | 182 +++++++++++++++++++++++++++ cmd/galactic-gateway/root_test.go | 104 +++++++++++++++ internal/config/gateway.go | 154 +++++++++++++++++++++++ internal/config/gateway_test.go | 136 ++++++++++++++++++++ internal/config/router_test.go | 5 + 9 files changed, 826 insertions(+) create mode 100644 cmd/galactic-gateway/gateway.go create mode 100644 cmd/galactic-gateway/gateway_test.go create mode 100644 cmd/galactic-gateway/main.go create mode 100644 cmd/galactic-gateway/root.go create mode 100644 cmd/galactic-gateway/root_test.go create mode 100644 internal/config/gateway.go create mode 100644 internal/config/gateway_test.go diff --git a/Taskfile.yaml b/Taskfile.yaml index d53f8b0..496c86a 100644 --- a/Taskfile.yaml +++ b/Taskfile.yaml @@ -140,6 +140,7 @@ tasks: - go build -ldflags "{{.LDFLAGS}}" -o bin/galactic-bgp ./cmd/galactic-bgp - go build -ldflags "{{.LDFLAGS}}" -o bin/galactic-route ./cmd/galactic-route - go build -ldflags "{{.LDFLAGS}}" -o bin/galactic-router ./cmd/galactic-router + - go build -ldflags "{{.LDFLAGS}}" -o bin/galactic-gateway ./cmd/galactic-gateway - go build -ldflags "{{.LDFLAGS}}" -o bin/vmtap-cni ./cmd/vmtap-cni - GOBIN={{.LOCALBIN}} go install github.com/containernetworking/plugins/plugins/main/host-device@v1.9.1 diff --git a/cmd/galactic-gateway/gateway.go b/cmd/galactic-gateway/gateway.go new file mode 100644 index 0000000..973a7e0 --- /dev/null +++ b/cmd/galactic-gateway/gateway.go @@ -0,0 +1,116 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package main + +import ( + "fmt" + "net" + "net/netip" + + "github.com/cilium/ebpf/link" + "github.com/prometheus/client_golang/prometheus" + + "go.datum.net/galactic/internal/gateway" + "go.datum.net/galactic/internal/plumbing/ebpf/edgeattach" + "go.datum.net/galactic/internal/plumbing/ebpf/edgemetrics" + "go.datum.net/galactic/internal/plumbing/ebpf/edgeprog" +) + +// gatewayDatapathKeepAlive holds the loaded *edgeprog.EdgenatObjects and +// the attached link.Link for the life of this process, once +// setupGatewayDatapath's attach path succeeds. Neither is Closed anywhere +// in this file — see setupGatewayDatapath's doc comment for why — but a +// value that isn't stored somewhere reachable is exactly as good as +// Closed: cilium/ebpf's *ebpf.Program, *ebpf.Map, and link.Link types all +// register a runtime finalizer that closes their underlying fd once the +// garbage collector determines nothing reachable still points at them, +// with no error surfaced anywhere when that happens. gateway.KernelDatapath +// only keeps objs.RuleTable (via edgemap.KernelTable) alive on its own, so +// without this package-level var, objs.EdgeNat (the program) and the +// link.Link returned by Attach — the two things actually keeping this +// node's XDP attachment live on the wire — would eventually get GC'd and +// silently detached, with every control-plane signal (the DaemonSet pod +// healthy, ApplyRule succeeding, rule_table metrics populated) still +// looking completely normal. Confirmed live: this is exactly what happened +// the first time this path was ever exercised against a real interface +// (ingress traffic for a registered rule was never intercepted at all, +// bouncing between this node and its transit-facing peer via ordinary +// kernel routing instead) — no unit test exercises this path with a real +// attach for a GC cycle to occur during, and Phase D's "manifests and the +// live pod/eBPF path" validation predates any live underlay BGP peering +// that would have delivered real traffic to notice the gap. +// +// This var, and the rest of this file, moved here unchanged from +// cmd/galactic-router/gateway.go: the edge NAT+LB gateway datapath now +// lives in its own process rather than sharing one with the tenant BGP +// reconcilers. +var gatewayDatapathKeepAlive struct { + objs *edgeprog.EdgenatObjects + link link.Link +} + +// setupGatewayDatapath loads and attaches the edge NAT+LB eBPF datapath to +// publicInterface and returns the gateway.Datapath this node's Engine +// should use. Unlike cmd/galactic-router's identically-named predecessor +// (now removed), publicInterface and +// srv6Address are both required here, not a jointly-optional pair with a +// gateway.NoopDatapath{} fallback: config.GatewayConfig.Validate already +// rejects either being empty before runCmd ever calls this function, since +// this binary only exists to run the gateway role. +// +// The loaded *edgeprog.EdgenatObjects and the returned link.Link are +// stashed in gatewayDatapathKeepAlive (see that var's doc comment for why) +// rather than Closed here: they, and the XDP attachment itself, must +// survive for the life of this process — same convention as +// internal/plumbing/ebpf/attach.Start's identical choice for the SRv6 uSID +// datapath. +// +// metricsReg additionally gets an edgemetrics.Collector registered against +// it once objs is loaded, reading rule_table/conn_table/drop_reasons live +// at every scrape — see that package's doc comment for why this is a +// pull-based Collector rather than incrementally-updated Gauges. +func setupGatewayDatapath( + publicInterface, srv6Address string, metricsReg prometheus.Registerer, +) (gateway.Datapath, error) { + gwAddr, err := netip.ParseAddr(srv6Address) + if err != nil { + return nil, fmt.Errorf("parse gateway SRv6 address %q: %w", srv6Address, err) + } + + objs, err := edgeattach.Load(edgeattach.PinDir) + if err != nil { + return nil, fmt.Errorf("load edge gateway eBPF datapath: %w", err) + } + + xdpLink, err := edgeattach.Attach(objs.EdgeNat, publicInterface) + if err != nil { + _ = objs.Close() + return nil, fmt.Errorf("attach edge gateway datapath to public interface %q: %w", publicInterface, err) + } + + if _, err := net.InterfaceByName(publicInterface); err != nil { + _ = xdpLink.Close() + _ = objs.Close() + return nil, fmt.Errorf("resolve public interface %q: %w", publicInterface, err) + } + + datapath, err := gateway.NewKernelDatapath(objs, gwAddr) + if err != nil { + _ = xdpLink.Close() + _ = objs.Close() + return nil, fmt.Errorf("construct kernel datapath: %w", err) + } + + if err := metricsReg.Register(edgemetrics.NewCollectorFromObjects(objs)); err != nil { + _ = xdpLink.Close() + _ = objs.Close() + return nil, fmt.Errorf("register edge gateway metrics collector: %w", err) + } + + gatewayDatapathKeepAlive.objs = objs + gatewayDatapathKeepAlive.link = xdpLink + + return datapath, nil +} diff --git a/cmd/galactic-gateway/gateway_test.go b/cmd/galactic-gateway/gateway_test.go new file mode 100644 index 0000000..3feffef --- /dev/null +++ b/cmd/galactic-gateway/gateway_test.go @@ -0,0 +1,24 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package main + +import ( + "testing" + + "github.com/prometheus/client_golang/prometheus" +) + +// TestSetupGatewayDatapath_InvalidAddressIsError covers the address-parse +// failure path, which runs before any kernel/eBPF interaction -- no root +// needed. Unlike cmd/galactic-router's removed identically-named test +// file, there is no "empty interface is a no-op" case to cover here: this +// binary's config.GatewayConfig.Validate rejects an empty +// PublicInterface/SRv6Address before setupGatewayDatapath is ever called. +func TestSetupGatewayDatapath_InvalidAddressIsError(t *testing.T) { + _, err := setupGatewayDatapath("eth0", "not-an-ip-address", prometheus.NewRegistry()) + if err == nil { + t.Error("setupGatewayDatapath with an invalid SRv6 address: want an error, got nil") + } +} diff --git a/cmd/galactic-gateway/main.go b/cmd/galactic-gateway/main.go new file mode 100644 index 0000000..830a67b --- /dev/null +++ b/cmd/galactic-gateway/main.go @@ -0,0 +1,104 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +// Command galactic-gateway is the edge XDP NAT+LB gateway control plane of +// the Galactic data plane. It loads and attaches the edge NAT+LB eBPF +// program (internal/plumbing/ebpf/edgeprog) to a gateway node's public/ +// underlay-facing uplink interface and drives it via the +// NetworkGateway/NetworkRule reconcilers. +// +// This binary was split out of galactic-router so that a crash on either +// side (tenant BGP vs. the XDP-holding gateway engine) no longer takes the +// other down with it. Tenant BGP (the embedded GoBGP server and the +// BGPRouter/BGPPeer/ +// BGPAdvertisement/BGPPolicy/BGPVRFInstance reconcilers) still runs in +// galactic-router, co-located on the same gateway node. +package main + +import ( + "context" + "os" + "time" + + authorizationv1 "k8s.io/api/authorization/v1" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +const ( + networkAPIGroup = "network.datumapis.com" + networkAPIVersion = "v1alpha1" + + resourceNetworkGateways = "networkgateways" + resourceNetworkRules = "networkrules" + resourceBGPAdvertisements = "bgpadvertisements" + resourceBGPRouters = "bgprouters" +) + +func main() { + cmd := newRootCommand() + if err := cmd.Execute(); err != nil { + os.Exit(1) + } +} + +// checkWatchPermissions issues a SelfSubjectAccessReview for each resource +// type the manager watches, checking the watch verb. If any review denies +// the request the informer cache will never sync and all reconcilers will +// be silently blocked; this logs a clear, actionable message at startup so +// the problem is immediately obvious. Mirrors +// cmd/galactic-router/main.go's identically-named function, scoped to this +// binary's own resource set instead of the BGP-family CRDs. +// +// resourceBGPRouters is included even though this binary has no BGP +// client of its own: internal/controller/usidresolver.go's +// buildBackendSIDIndex lists +// BGPRouter CRDs directly (alongside BGPAdvertisement) to resolve a +// NetworkRule backend's SRv6 uSID, so read access to bgprouters is a real +// RBAC requirement here regardless of not needing a BGP runtime. +func checkWatchPermissions(mgr ctrl.Manager) { + c, err := client.New(mgr.GetConfig(), client.Options{Scheme: mgr.GetScheme()}) + if err != nil { + ctrl.Log.Error(err, "RBAC pre-flight: cannot create client, skipping check") + return + } + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + logger := ctrl.Log.WithName("rbac-preflight") + + resources := []struct { + group string + version string + resource string + }{ + {group: networkAPIGroup, version: networkAPIVersion, resource: resourceNetworkGateways}, + {group: networkAPIGroup, version: networkAPIVersion, resource: resourceNetworkRules}, + {group: networkAPIGroup, version: networkAPIVersion, resource: resourceBGPAdvertisements}, + {group: networkAPIGroup, version: networkAPIVersion, resource: resourceBGPRouters}, + } + + for _, r := range resources { + review := &authorizationv1.SelfSubjectAccessReview{ + Spec: authorizationv1.SelfSubjectAccessReviewSpec{ + ResourceAttributes: &authorizationv1.ResourceAttributes{ + Verb: "watch", + Group: r.group, + Version: r.version, + Resource: r.resource, + }, + }, + } + if err := c.Create(ctx, review); err != nil { + logger.Error(err, "RBAC pre-flight: failed to submit access review for "+r.resource, "verb", "watch") + continue + } + if review.Status.Allowed { + continue + } + logger.Error(nil, "missing watch RBAC for "+r.resource, + "verb", "watch", "detail", "informer cache will not sync; add resource to ServiceAccount ClusterRole and restart") + } +} diff --git a/cmd/galactic-gateway/root.go b/cmd/galactic-gateway/root.go new file mode 100644 index 0000000..1bdfba8 --- /dev/null +++ b/cmd/galactic-gateway/root.go @@ -0,0 +1,182 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package main + +import ( + "fmt" + "log" + "net" + "strings" + + "github.com/spf13/cobra" + "google.golang.org/grpc" + grpchealth "google.golang.org/grpc/health" + "google.golang.org/grpc/health/grpc_health_v1" + "k8s.io/apimachinery/pkg/runtime" + utilruntime "k8s.io/apimachinery/pkg/util/runtime" + clientgoscheme "k8s.io/client-go/kubernetes/scheme" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/log/zap" + ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/metrics" + metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server" + + "go.datum.net/galactic/internal/config" + "go.datum.net/galactic/internal/controller" + "go.datum.net/galactic/internal/gateway" + "go.datum.net/galactic/internal/metadata" + bgpv1alpha1 "go.datum.net/network/api/v1alpha1" +) + +const ( + appName = "galactic-gateway" + + appDesc = `Galactic edge XDP NAT+LB gateway + + Find more information at: https://www.datum.net/docs` +) + +// runCmd contains the application startup logic: it loads and attaches the +// edge NAT+LB eBPF datapath to this node's public interface and registers +// the NetworkGateway/NetworkRule reconcilers that drive it. Unlike +// cmd/galactic-router's runCmd, there is no BGP runtime, RuntimeManager, or +// BGP-family reconciler here at all: NetworkGatewayReconciler/ +// NetworkRuleReconciler need no BGP client of their own — they only +// create/update/delete BGPAdvertisement CRDs, which the co-located +// galactic-router (tenant role) picks up via its own +// BGPAdvertisementReconciler. +func runCmd(cfg *config.GatewayConfig) error { + nodeName := cfg.NodeName + metricsPort := cfg.MetricsPort + grpcHealthPort := cfg.GRPCHealthPort + + ctrl.SetLogger(zap.New(zap.UseDevMode(true))) + + scheme := runtime.NewScheme() + utilruntime.Must(clientgoscheme.AddToScheme(scheme)) + utilruntime.Must(bgpv1alpha1.AddToScheme(scheme)) + + mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{ + Scheme: scheme, + HealthProbeBindAddress: "0", + Metrics: metricsserver.Options{ + BindAddress: fmt.Sprintf(":%d", metricsPort), + }, + }) + if err != nil { + return fmt.Errorf("create manager: %w", err) + } + + ctx := ctrl.SetupSignalHandler() + + // Start gRPC health server. + lis, err := (&net.ListenConfig{}).Listen(ctx, "tcp", fmt.Sprintf(":%d", grpcHealthPort)) + if err != nil { + return fmt.Errorf("listen on gRPC health port %d: %w", grpcHealthPort, err) + } + grpcSrv := grpc.NewServer() + healthSrv := grpchealth.NewServer() + grpc_health_v1.RegisterHealthServer(grpcSrv, healthSrv) + healthSrv.SetServingStatus("", grpc_health_v1.HealthCheckResponse_SERVING) + go func() { + if serveErr := grpcSrv.Serve(lis); serveErr != nil { + log.Printf("gRPC health server: %v", serveErr) + } + }() + go func() { + <-ctx.Done() + grpcSrv.GracefulStop() + }() + + // Pre-flight RBAC check. + checkWatchPermissions(mgr) + + // Load and attach the edge NAT+LB eBPF datapath. Always loads a real + // gateway.KernelDatapath -- unlike cmd/galactic-router's removed + // setupGatewayDatapath, there is no gateway.NoopDatapath{} case here: + // config.GatewayConfig.Validate already rejects an empty + // PublicInterface/SRv6Address before runCmd is ever reached. + gwDatapath, err := setupGatewayDatapath(cfg.PublicInterface, cfg.SRv6Address, ctrlmetrics.Registry) + if err != nil { + return fmt.Errorf("setup edge gateway eBPF datapath: %w", err) + } + + // Real (not stubbed) QuotaEnforcer/TelemetryEmitter — see + // internal/gateway/quota.go and telemetry.go's doc comments for what + // each does and does not cover. + gwQuota := gateway.NewNodeQuotaEnforcer(gateway.DefaultMaxRulesPerTenant, gateway.DefaultMaxRuleTableEntries) + gwTelemetry := gateway.NewPrometheusTelemetryEmitter() + gwTelemetry.MustRegister(ctrlmetrics.Registry) + gwEngine := gateway.NewEngine(gwDatapath, gwQuota, gwTelemetry) + + // Register NetworkGateway controller (edge XDP NAT+LB gateway engine). + if err := (&controller.NetworkGatewayReconciler{ + Client: mgr.GetClient(), + Scheme: mgr.GetScheme(), + Engine: gwEngine, + NodeName: nodeName, + SRv6Address: cfg.SRv6Address, + }).SetupWithManager(mgr); err != nil { + return fmt.Errorf("setup NetworkGateway controller: %w", err) + } + + // Register NetworkRule controller (finalizer-guarded teardown ordering + // and one-time primary_node assignment; see + // internal/controller/networkrule_controller.go). + if err := (&controller.NetworkRuleReconciler{ + Client: mgr.GetClient(), + Scheme: mgr.GetScheme(), + NodeName: nodeName, + }).SetupWithManager(mgr); err != nil { + return fmt.Errorf("setup NetworkRule controller: %w", err) + } + + if err := mgr.Start(ctx); err != nil { + return fmt.Errorf("manager exited: %w", err) + } + + return nil +} + +// newRootCommand builds the root cobra command with all flags and the +// application startup logic. +func newRootCommand() *cobra.Command { + cmd := &cobra.Command{ + Use: appName, + Short: strings.Split(appDesc, "\n")[0], + Long: appDesc, + RunE: func(cmd *cobra.Command, _ []string) error { + if ok, _ := cmd.Flags().GetBool("build-info"); ok { + fmt.Println(metadata.BuildInfo(appName)) + return nil + } + if ok, _ := cmd.Flags().GetBool("version"); ok { + fmt.Printf("galactic-gateway version %s\n", metadata.Version) + return nil + } + + cfg := config.NewGatewayConfig() + cfg.BindFlags(cmd.Flags()) + if err := cfg.Validate(); err != nil { + return err + } + return runCmd(cfg) + }, + } + + cmd.Flags().StringP("node-name", "n", "", "Kubernetes node name (required)") + cmd.Flags().IntP("metrics-port", "", + config.DefaultGatewayMetricsPort, + "Metrics listen port") + cmd.Flags().IntP("grpc-health-port", "", + config.DefaultGatewayGRPCHealthPort, + "gRPC health check port") + cmd.Flags().StringP("gateway-public-interface", "", "", + "Public/underlay-facing uplink interface for the edge NAT+LB gateway datapath (required)") + cmd.Flags().StringP("gateway-srv6-address", "", "", + "This gateway node's own SRv6-reachable address, used as the Full-NAT SNAT source (required)") + cmd.Flags().Bool("build-info", false, "Print build information and exit") + cmd.Flags().BoolP("version", "V", false, "Print version and exit") + return cmd +} diff --git a/cmd/galactic-gateway/root_test.go b/cmd/galactic-gateway/root_test.go new file mode 100644 index 0000000..b296241 --- /dev/null +++ b/cmd/galactic-gateway/root_test.go @@ -0,0 +1,104 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package main + +import ( + "testing" + + "github.com/spf13/cobra" + + "go.datum.net/galactic/internal/config" + "go.datum.net/galactic/internal/metadata" +) + +const testCmdUse = "test" + +// testCmd creates a cobra command with the same flags as newRootCommand. +func testCmd(t *testing.T) *cobra.Command { + t.Helper() + cmd := &cobra.Command{Use: testCmdUse} + cmd.Flags().StringP("node-name", "n", "", "Kubernetes node name (required)") + cmd.Flags().IntP("metrics-port", "", config.DefaultGatewayMetricsPort, "Metrics listen port") + cmd.Flags().IntP("grpc-health-port", "", config.DefaultGatewayGRPCHealthPort, "gRPC health check port") + cmd.Flags().StringP("gateway-public-interface", "", "", "Edge gateway public interface") + cmd.Flags().StringP("gateway-srv6-address", "", "", "Edge gateway SRv6 address") + return cmd +} + +func TestFlagDefaults(t *testing.T) { + cfg := config.NewGatewayConfig() + cmd := testCmd(t) + cfg.BindFlags(cmd.Flags()) + + if cfg.MetricsPort != config.DefaultGatewayMetricsPort { + t.Errorf("MetricsPort = %d, want %d", cfg.MetricsPort, config.DefaultGatewayMetricsPort) + } + if cfg.GRPCHealthPort != config.DefaultGatewayGRPCHealthPort { + t.Errorf("GRPCHealthPort = %d, want %d", cfg.GRPCHealthPort, config.DefaultGatewayGRPCHealthPort) + } +} + +func TestRequiredFlags(t *testing.T) { + cfg := config.NewGatewayConfig() + cmd := testCmd(t) + cfg.BindFlags(cmd.Flags()) + if err := cfg.Validate(); err == nil { + t.Error("Validate() with empty node-name/public-interface/srv6-address returned nil error") + } +} + +func TestEnvVarDefaults(t *testing.T) { + t.Setenv(config.EnvGatewayNodeName, "test-node") + t.Setenv(config.EnvGatewayPublicInterface, "eth0") + t.Setenv(config.EnvGatewaySRv6Address, "2001:db8::1") + + cfg := config.NewGatewayConfig() + cmd := testCmd(t) + cfg.BindFlags(cmd.Flags()) + if err := cfg.Validate(); err != nil { + t.Errorf("Validate() with valid env vars: %v", err) + } +} + +func TestNodeNameRequired(t *testing.T) { + t.Setenv(config.EnvGatewayPublicInterface, "eth0") + t.Setenv(config.EnvGatewaySRv6Address, "2001:db8::1") + + cfg := config.NewGatewayConfig() + cmd := testCmd(t) + cfg.BindFlags(cmd.Flags()) + if err := cfg.Validate(); err == nil { + t.Error("Validate() with empty node-name returned nil error") + } +} + +func TestGatewayFlagsOverrideEnv(t *testing.T) { + t.Setenv(config.EnvGatewayNodeName, "test-node") + t.Setenv(config.EnvGatewayPublicInterface, "env-eth0") + t.Setenv(config.EnvGatewaySRv6Address, "2001:db8:3::1") + + cfg := config.NewGatewayConfig() + cmd := testCmd(t) + if err := cmd.Flags().Set("gateway-public-interface", "flag-eth0"); err != nil { + t.Fatalf("set --gateway-public-interface flag: %v", err) + } + if err := cmd.Flags().Set("gateway-srv6-address", "2001:db8:3::2"); err != nil { + t.Fatalf("set --gateway-srv6-address flag: %v", err) + } + cfg.BindFlags(cmd.Flags()) + + if cfg.PublicInterface != "flag-eth0" { + t.Errorf("PublicInterface = %q, want %q (flag should override env var)", cfg.PublicInterface, "flag-eth0") + } + if cfg.SRv6Address != "2001:db8:3::2" { + t.Errorf("SRv6Address = %q, want %q (flag should override env var)", cfg.SRv6Address, "2001:db8:3::2") + } +} + +func TestVersionFlag(t *testing.T) { + if metadata.Version == "" { + t.Error("metadata.Version should not be empty") + } +} diff --git a/internal/config/gateway.go b/internal/config/gateway.go new file mode 100644 index 0000000..d612e96 --- /dev/null +++ b/internal/config/gateway.go @@ -0,0 +1,154 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package config + +import ( + "errors" + "fmt" + + "github.com/spf13/pflag" + "github.com/spf13/viper" +) + +// --- Gateway defaults -------------------------------------------------- + +const ( + // DefaultGatewayMetricsPort/DefaultGatewayGRPCHealthPort deliberately + // differ from DefaultRouterMetricsPort (8080) and + // DefaultRouterGRPCHealthPort (5000, overridden to 5179 on Talos by + // config/router/base/daemonset.yaml) and from galactic-cni's + // credential-refresh grpc-health port (5180, + // config/cni/daemonset.yaml): galactic-gateway is deployed as a + // second container in the same hostNetwork: true pod as + // galactic-router, so every port it binds is on the same network + // namespace as every other galactic-* process already running on + // that node and must not collide with any of them. + DefaultGatewayMetricsPort = 8081 + DefaultGatewayGRPCHealthPort = 5181 +) + +// --- Gateway environment variable keys ----------------------------------- + +const ( + EnvGatewayNodeName = "GALACTIC_GATEWAY_NODE_NAME" + EnvGatewayMetricsPort = "GALACTIC_GATEWAY_METRICS_PORT" + EnvGatewayGRPCHealthPort = "GALACTIC_GATEWAY_GRPC_HEALTH_PORT" + + // EnvGatewayPublicInterface names this node's single public/underlay- + // facing uplink interface for the edge NAT+LB gateway datapath + // (internal/plumbing/ebpf/edgeprog). Unlike + // config.EnvRouterGatewayPublicInterface's predecessor field, this is + // required: galactic-gateway only ever runs as the gateway role, so + // there is no "not gateway role, return gateway.NoopDatapath{}" case + // to support — see GatewayConfig.Validate. + EnvGatewayPublicInterface = "GALACTIC_GATEWAY_PUBLIC_INTERFACE" + + // EnvGatewaySRv6Address is this gateway node's own SRv6-reachable + // address (network.datumapis.com/v1alpha1's NetworkGatewayStatus. + // SRv6Address), used as the Full-NAT SNAT source for every ingress + // flow this node translates. Required — see EnvGatewayPublicInterface. + EnvGatewaySRv6Address = "GALACTIC_GATEWAY_SRV6_ADDRESS" +) + +// --- GatewayConfig ----------------------------------------------------- + +// GatewayConfig resolves galactic-gateway configuration with three-tier +// precedence: CLI flag > env var > compiled-in default. Create once via +// NewGatewayConfig(), call BindFlags() to layer CLI flags, then read the +// exported fields. +type GatewayConfig struct { + v *viper.Viper + prefix string + + // Resolved fields. + NodeName string + MetricsPort int + GRPCHealthPort int + + // PublicInterface/SRv6Address configure the edge NAT+LB gateway + // datapath -- see EnvGatewayPublicInterface's doc comment. Both + // required; GatewayConfig.Validate rejects either being empty. + PublicInterface string + SRv6Address string +} + +// NewGatewayConfig creates a gateway config resolver with the +// GALACTIC_GATEWAY env prefix and AutomaticEnv enabled. Exported fields are +// populated from env vars and defaults; call BindFlags() to layer CLI +// overrides. +func NewGatewayConfig() *GatewayConfig { + v := viper.New() + v.SetEnvPrefix("GALACTIC_GATEWAY") + v.AutomaticEnv() + + v.SetDefault("node_name", "") + v.SetDefault("metrics_port", DefaultGatewayMetricsPort) + v.SetDefault("grpc_health_port", DefaultGatewayGRPCHealthPort) + v.SetDefault("public_interface", "") + v.SetDefault("srv6_address", "") + + cfg := &GatewayConfig{ + v: v, + prefix: "GALACTIC_GATEWAY", + } + cfg.readFields() + return cfg +} + +// BindFlags binds Cobra/pflag flags to the config resolver and re-reads the +// exported fields. Each flag is bound to a Viper key using the key argument. +func (c *GatewayConfig) BindFlags(flags *pflag.FlagSet) { + bindings := []struct { + flag string + key string + }{ + {"node-name", "node_name"}, + {"metrics-port", "metrics_port"}, + {"grpc-health-port", "grpc_health_port"}, + {"gateway-public-interface", "public_interface"}, + {"gateway-srv6-address", "srv6_address"}, + } + for _, b := range bindings { + if flags.Changed(b.flag) { + c.v.Set(b.key, flags.Lookup(b.flag).Value.String()) + } else { + //nolint:errcheck // controlled keys, BindPFlag cannot fail here + c.v.BindPFlag(b.key, flags.Lookup(b.flag)) + } + } + c.readFields() +} + +// readFields populates the exported fields from the current Viper state. +func (c *GatewayConfig) readFields() { + c.NodeName = c.v.GetString("node_name") + c.MetricsPort = c.v.GetInt("metrics_port") + c.GRPCHealthPort = c.v.GetInt("grpc_health_port") + c.PublicInterface = c.v.GetString("public_interface") + c.SRv6Address = c.v.GetString("srv6_address") +} + +// Validate checks that the required configuration fields are set. +func (c *GatewayConfig) Validate() error { + if c.NodeName == "" { + return fmt.Errorf("node name is required (use --node-name flag or %s env var)", EnvGatewayNodeName) + } + if c.PublicInterface == "" { + return fmt.Errorf( + "public interface is required (use --gateway-public-interface flag or %s env var)", + EnvGatewayPublicInterface) + } + if c.SRv6Address == "" { + return fmt.Errorf( + "SRv6 address is required (use --gateway-srv6-address flag or %s env var)", EnvGatewaySRv6Address) + } + if c.MetricsPort < 1 || c.MetricsPort > 65535 { + return errors.New("metrics port must be between 1 and 65535") + } + if c.GRPCHealthPort < 1 || c.GRPCHealthPort > 65535 { + return errors.New("grpc health port must be between 1 and 65535") + } + return nil +} diff --git a/internal/config/gateway_test.go b/internal/config/gateway_test.go new file mode 100644 index 0000000..47ad13b --- /dev/null +++ b/internal/config/gateway_test.go @@ -0,0 +1,136 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package config + +import ( + "strings" + "testing" +) + +const ( + testGatewayNodeName = "test-gateway-node" +) + +func TestGatewayConfigDefaults(t *testing.T) { + cfg := NewGatewayConfig() + + if cfg.MetricsPort != DefaultGatewayMetricsPort { + t.Errorf("MetricsPort = %d, want %d", cfg.MetricsPort, DefaultGatewayMetricsPort) + } + if cfg.GRPCHealthPort != DefaultGatewayGRPCHealthPort { + t.Errorf("GRPCHealthPort = %d, want %d", cfg.GRPCHealthPort, DefaultGatewayGRPCHealthPort) + } + if cfg.NodeName != "" { + t.Errorf("NodeName = %q, want empty", cfg.NodeName) + } + if cfg.PublicInterface != "" { + t.Errorf("PublicInterface = %q, want empty", cfg.PublicInterface) + } + if cfg.SRv6Address != "" { + t.Errorf("SRv6Address = %q, want empty", cfg.SRv6Address) + } +} + +func TestGatewayConfigEnvOverride(t *testing.T) { + t.Setenv(EnvGatewayNodeName, "env-node") + t.Setenv(EnvGatewayMetricsPort, "9090") + t.Setenv(EnvGatewayGRPCHealthPort, "9091") + t.Setenv(EnvGatewayPublicInterface, testGatewayIface) + t.Setenv(EnvGatewaySRv6Address, testGatewaySRv6) + + cfg := NewGatewayConfig() + + if cfg.NodeName != "env-node" { + t.Errorf("NodeName = %q, want %q", cfg.NodeName, "env-node") + } + if cfg.MetricsPort != 9090 { + t.Errorf("MetricsPort = %d, want 9090", cfg.MetricsPort) + } + if cfg.GRPCHealthPort != 9091 { + t.Errorf("GRPCHealthPort = %d, want 9091", cfg.GRPCHealthPort) + } + if cfg.PublicInterface != testGatewayIface { + t.Errorf("PublicInterface = %q, want %q", cfg.PublicInterface, testGatewayIface) + } + if cfg.SRv6Address != testGatewaySRv6 { + t.Errorf("SRv6Address = %q, want %q", cfg.SRv6Address, testGatewaySRv6) + } +} + +func TestGatewayConfigValidate(t *testing.T) { + tests := []struct { + name string + envVars map[string]string + wantErr string + }{ + { + name: "missing node name", + envVars: map[string]string{EnvGatewayPublicInterface: testGatewayIface, EnvGatewaySRv6Address: testGatewaySRv6}, + wantErr: "node name is required", + }, + { + name: "missing public interface", + envVars: map[string]string{EnvGatewayNodeName: testGatewayNodeName, EnvGatewaySRv6Address: testGatewaySRv6}, + wantErr: "public interface is required", + }, + { + name: "missing srv6 address", + envVars: map[string]string{EnvGatewayNodeName: testGatewayNodeName, EnvGatewayPublicInterface: testGatewayIface}, + wantErr: "SRv6 address is required", + }, + { + name: "invalid metrics port", + envVars: map[string]string{ + EnvGatewayNodeName: testGatewayNodeName, + EnvGatewayPublicInterface: testGatewayIface, + EnvGatewaySRv6Address: testGatewaySRv6, + EnvGatewayMetricsPort: "0", + }, + wantErr: "metrics port must be between", + }, + { + name: "invalid grpc health port", + envVars: map[string]string{ + EnvGatewayNodeName: testGatewayNodeName, + EnvGatewayPublicInterface: testGatewayIface, + EnvGatewaySRv6Address: testGatewaySRv6, + EnvGatewayGRPCHealthPort: "0", + }, + wantErr: "grpc health port must be between", + }, + { + name: "valid config", + envVars: map[string]string{ + EnvGatewayNodeName: testGatewayNodeName, + EnvGatewayPublicInterface: testGatewayIface, + EnvGatewaySRv6Address: testGatewaySRv6, + }, + wantErr: "", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + for k, v := range tc.envVars { + t.Setenv(k, v) + } + cfg := NewGatewayConfig() + err := cfg.Validate() + if tc.wantErr == "" { + if err != nil { + t.Errorf("Validate() = %v, want nil", err) + } + return + } + if err == nil { + t.Errorf("Validate() = nil, want error containing %q", tc.wantErr) + return + } + if !strings.Contains(err.Error(), tc.wantErr) { + t.Errorf("Validate() = %q, want error containing %q", err, tc.wantErr) + } + }) + } +} diff --git a/internal/config/router_test.go b/internal/config/router_test.go index f341584..8d7fce7 100644 --- a/internal/config/router_test.go +++ b/internal/config/router_test.go @@ -13,6 +13,11 @@ import ( const ( testRouterNodeName = "test-node" testBoolTrue = "true" + // testGatewayIface/testGatewaySRv6 are shared with gateway_test.go's + // GatewayConfig tests, which took over the equivalent fields dropped + // from RouterConfig by the galactic-gateway split. + testGatewayIface = "eth0" + testGatewaySRv6 = "2001:db8:3::1" ) func TestRouterConfigDefaults(t *testing.T) { From beb2198f67977026eaf10eabf15ce25a40eb3560 Mon Sep 17 00:00:00 2001 From: Peter Sprygada Date: Wed, 12 Aug 2026 17:18:48 -0400 Subject: [PATCH 2/2] fix(gateway): report healthy only once the datapath is attached galactic-gateway's health server set SERVING right after registering it, before setupGatewayDatapath ever ran. grpchealth.NewServer() additionally defaults the "" overall-health service to SERVING on its own, so the process was already answering healthy from the moment the server started -- not merely from that explicit call -- for the entire window before the XDP program was loaded and attached. Any probe watching during that window saw ready before the thing being probed existed, masking a silent failure: traffic that should have been intercepted routed normally instead. Three smaller issues sat alongside it: - The health server's Serve() goroutine logged a failure and kept going. If it stopped serving, the process ran on with no health signal at all and nothing restarted it. - The gateway's own SRv6 address was checked for being parseable, not for being the right family. An IPv4 address passed GatewayConfig.Validate and only failed later, deeper in, at kerneldatapath.go's identical Is6()/Is4In6() check. - setupGatewayDatapath looked up the public interface with net.InterfaceByName after edgeattach.Attach had already resolved the same interface. Attach would already have failed if it were missing, so the check could never fire. Fixes, in cmd/galactic-gateway/root.go, gateway.go, and internal/config/gateway.go: - Explicitly set NOT_SERVING when the health server is registered (overriding grpchealth's own SERVING default), and only flip to SERVING once setupGatewayDatapath has attached the datapath and constructed the rule table. - Wrap ctrl.SetupSignalHandler()'s context with context.WithCancelCause. A Serve() failure now cancels it with that error as the cause, which mgr.Start propagates out of runCmd as a fatal error -- the same as any other startup failure -- instead of a dropped log line. context.Cause is checked against context.Canceled to still treat an ordinary signal-triggered shutdown as success. - GatewayConfig.Validate now parses SRv6Address and rejects anything that isn't a native IPv6 address, with a message that names the problem instead of surfacing it later as a kernel-datapath error. - Dropped the dead net.InterfaceByName check. Adds two GatewayConfig.Validate cases (unparseable address, IPv4 address) covering the new validation. Fixes #360 Co-Authored-By: Claude Sonnet 5 --- cmd/galactic-gateway/gateway.go | 7 ------ cmd/galactic-gateway/root.go | 38 +++++++++++++++++++++++++++------ internal/config/gateway.go | 13 +++++++++++ internal/config/gateway_test.go | 22 +++++++++++++++++++ 4 files changed, 67 insertions(+), 13 deletions(-) diff --git a/cmd/galactic-gateway/gateway.go b/cmd/galactic-gateway/gateway.go index 973a7e0..1fd11da 100644 --- a/cmd/galactic-gateway/gateway.go +++ b/cmd/galactic-gateway/gateway.go @@ -6,7 +6,6 @@ package main import ( "fmt" - "net" "net/netip" "github.com/cilium/ebpf/link" @@ -90,12 +89,6 @@ func setupGatewayDatapath( return nil, fmt.Errorf("attach edge gateway datapath to public interface %q: %w", publicInterface, err) } - if _, err := net.InterfaceByName(publicInterface); err != nil { - _ = xdpLink.Close() - _ = objs.Close() - return nil, fmt.Errorf("resolve public interface %q: %w", publicInterface, err) - } - datapath, err := gateway.NewKernelDatapath(objs, gwAddr) if err != nil { _ = xdpLink.Close() diff --git a/cmd/galactic-gateway/root.go b/cmd/galactic-gateway/root.go index 1bdfba8..74e2a17 100644 --- a/cmd/galactic-gateway/root.go +++ b/cmd/galactic-gateway/root.go @@ -5,8 +5,9 @@ package main import ( + "context" + "errors" "fmt" - "log" "net" "strings" @@ -68,9 +69,19 @@ func runCmd(cfg *config.GatewayConfig) error { return fmt.Errorf("create manager: %w", err) } - ctx := ctrl.SetupSignalHandler() - - // Start gRPC health server. + // ctx's cause distinguishes a normal signal-triggered shutdown from the + // health server's own Serve failure below (#360) -- see the cause check + // after mgr.Start. + ctx, cancel := context.WithCancelCause(ctrl.SetupSignalHandler()) + defer cancel(nil) + + // Start gRPC health server. grpchealth.NewServer() defaults the "" + // overall-health service to SERVING, so that must be overridden to + // NOT_SERVING here, explicitly and immediately: otherwise a probe could + // see "healthy" for the entire window before the datapath below is + // even attached, which is exactly the failure #360 describes. Only once + // the datapath is attached and the rule table is reachable does this + // flip to SERVING, further down. lis, err := (&net.ListenConfig{}).Listen(ctx, "tcp", fmt.Sprintf(":%d", grpcHealthPort)) if err != nil { return fmt.Errorf("listen on gRPC health port %d: %w", grpcHealthPort, err) @@ -78,10 +89,15 @@ func runCmd(cfg *config.GatewayConfig) error { grpcSrv := grpc.NewServer() healthSrv := grpchealth.NewServer() grpc_health_v1.RegisterHealthServer(grpcSrv, healthSrv) - healthSrv.SetServingStatus("", grpc_health_v1.HealthCheckResponse_SERVING) + healthSrv.SetServingStatus("", grpc_health_v1.HealthCheckResponse_NOT_SERVING) go func() { + // A Serve failure here is fatal, not merely logged: with no health + // server left running and nothing to notice, the process would + // otherwise carry on with no health signal at all. Canceling ctx + // with this error as its cause carries it out through mgr.Start + // below, the same way any other fatal startup error does. if serveErr := grpcSrv.Serve(lis); serveErr != nil { - log.Printf("gRPC health server: %v", serveErr) + cancel(fmt.Errorf("gRPC health server: %w", serveErr)) } }() go func() { @@ -101,6 +117,9 @@ func runCmd(cfg *config.GatewayConfig) error { if err != nil { return fmt.Errorf("setup edge gateway eBPF datapath: %w", err) } + // Only now is the datapath attached and its rule table reachable -- + // report serving from here on, not from process start (#360). + healthSrv.SetServingStatus("", grpc_health_v1.HealthCheckResponse_SERVING) // Real (not stubbed) QuotaEnforcer/TelemetryEmitter — see // internal/gateway/quota.go and telemetry.go's doc comments for what @@ -135,6 +154,13 @@ func runCmd(cfg *config.GatewayConfig) error { if err := mgr.Start(ctx); err != nil { return fmt.Errorf("manager exited: %w", err) } + // mgr.Start only returns nil once ctx is Done, and by then ctx always + // has a cause: either context.Canceled (an ordinary signal-triggered + // shutdown) or the gRPC health server's own fatal Serve error from + // above. Only the latter should fail the process. + if cause := context.Cause(ctx); cause != nil && !errors.Is(cause, context.Canceled) { + return cause + } return nil } diff --git a/internal/config/gateway.go b/internal/config/gateway.go index d612e96..6f79be5 100644 --- a/internal/config/gateway.go +++ b/internal/config/gateway.go @@ -7,6 +7,7 @@ package config import ( "errors" "fmt" + "net/netip" "github.com/spf13/pflag" "github.com/spf13/viper" @@ -144,6 +145,18 @@ func (c *GatewayConfig) Validate() error { return fmt.Errorf( "SRv6 address is required (use --gateway-srv6-address flag or %s env var)", EnvGatewaySRv6Address) } + addr, err := netip.ParseAddr(c.SRv6Address) + if err != nil { + return fmt.Errorf("SRv6 address %q is not a valid IP address: %w", c.SRv6Address, err) + } + // Caught eventually anyway by internal/gateway/kerneldatapath.go's own + // identical Is6()/Is4In6() check, but late: only once setupGatewayDatapath + // has already loaded and attached the eBPF datapath. Reject it here + // instead, at startup, with a message that names the actual problem + // rather than surfacing as a deeper, less obvious kernel-datapath error. + if !addr.Is6() || addr.Is4In6() { + return fmt.Errorf("SRv6 address %q must be a native IPv6 address, not IPv4", c.SRv6Address) + } if c.MetricsPort < 1 || c.MetricsPort > 65535 { return errors.New("metrics port must be between 1 and 65535") } diff --git a/internal/config/gateway_test.go b/internal/config/gateway_test.go index 47ad13b..499830a 100644 --- a/internal/config/gateway_test.go +++ b/internal/config/gateway_test.go @@ -80,6 +80,28 @@ func TestGatewayConfigValidate(t *testing.T) { envVars: map[string]string{EnvGatewayNodeName: testGatewayNodeName, EnvGatewayPublicInterface: testGatewayIface}, wantErr: "SRv6 address is required", }, + { + name: "unparseable srv6 address", + envVars: map[string]string{ + EnvGatewayNodeName: testGatewayNodeName, + EnvGatewayPublicInterface: testGatewayIface, + EnvGatewaySRv6Address: "not-an-ip-address", + }, + wantErr: "is not a valid IP address", + }, + { + // Regression test for #360: an IPv4 address used to pass this + // Validate() call and only fail later, deeper in, at + // internal/gateway/kerneldatapath.go's identical Is6()/Is4In6() + // check -- reject it here instead, at startup. + name: "ipv4 srv6 address is wrong family", + envVars: map[string]string{ + EnvGatewayNodeName: testGatewayNodeName, + EnvGatewayPublicInterface: testGatewayIface, + EnvGatewaySRv6Address: "203.0.113.1", + }, + wantErr: "must be a native IPv6 address", + }, { name: "invalid metrics port", envVars: map[string]string{