diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index 5cf48097..08c75138 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -164,6 +164,14 @@ jobs: # `env "PATH=$PATH"`) already covers. uses: ./.github/actions/install-ebpf-deps + - name: Install envtest binaries + # The traffic policy node reconciler's test needs a real API server + # as well as root, and skips without one. + run: | + assets="$(go run sigs.k8s.io/controller-runtime/tools/setup-envtest@release-0.24 use 1.35.x --bin-dir "$RUNNER_TEMP/envtest" -p path)" + test -x "$assets/kube-apiserver" + echo "KUBEBUILDER_ASSETS=$assets" >> "$GITHUB_ENV" + - name: Run root-gated unit tests as root # sudo's secure_path policy overrides -E's preserved PATH for # exactly the PATH variable, which would hide the go/task binaries diff --git a/AGENTS.md b/AGENTS.md index f36fb1e3..a2577f4e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -100,7 +100,7 @@ See [docs/node-labels.md](docs/node-labels.md) for the full strategy — why `ga - **`config/fabric-router/`** — the FRR underlay eBGP DaemonSet (`fabric-router`; `galactic-router` needs a working underlay before it can start). Unlike `config/galactic-router/`, this is a single flat DaemonSet with no `default`/`rr`-style role split — its affinity matches any node labeled `galactic.datumapis.com/fabric: router` (a dedicated mode-enum label, independent of whatever `galactic.datumapis.com/node` role value or `galactic.datumapis.com/galactic` mode that node also carries) directly, since there's no env/config difference between running on a regular node vs. those roles — fabric-router itself is identical everywhere; only the per-node BGP underlay config it reads from the ConfigMap differs. That affinity can legitimately match more than one node per cluster, and BGP underlay config (hostname, router-id, interface addresses, remote-AS) inherently differs per physical node — so each node's config lives in its own `fabric-router.` ConfigMap (a `frr.conf` key, plus optional `daemons`/`vtysh.conf` overrides of the image defaults) rather than one shared ConfigMap for the whole DaemonSet. `fabric-config-agent` (`cmd/fabric-config-agent`, `internal/fabricconfig`, built into the `fabric-router` image) reads it from the API: as the `frr-init` init container it validates (`vtysh -C`) and installs it, holding the pod in `Init` until a valid one exists; as the `config-agent` sidecar it applies later changes live with `frr-reload.py`, so an edit to one node's ConfigMap neither restarts the pod nor touches other nodes, and reports `Applied`/`Rejected`/`ReloadFailed`/`RestartRequired` events on the ConfigMap. The sidecar also hard-resets any IPv6 session announcing an IPv4-mapped next hop (`internal/fabricconfig/nexthop.go`) and serves that check's metrics on the node's port `9343`. A third container, `frr-exporter` (upstream `tynany/frr_exporter`, pinned by digest), reads the daemons' vty sockets in the same pod-scoped `/run/frr` and serves BGP peer state, message counters and prefix counts for both underlay families on the node's port `9342` (`hostNetwork`); `task verify:fabric-metrics` in `deploy/containerlab/` checks it on every lab node. The image's default `daemons` loads bgpd's BMP module (`-M bmp`), so a node's `frr.conf` can stream to a BMP collector (`config/bmp-collector/`) with a `bmp targets` block; loading it alone streams nothing, and a change to the modules needs a pod restart rather than a reload. Every lab node's `frr.conf` streams its pre-policy Adj-RIB-In and Loc-RIB for both families, sourced from `lo` (`source-interface lo`), since the underlay carries loopbacks but not link subnets. Production `frr.conf`s come from infra's renderer, which has to emit the same block for production nodes to stream. A node with no ConfigMap of its own falls back, with a `DeprecatedConfigSource` warning, to key `frr.conf.` of the legacy shared `fabric-config` ConfigMap. **Not** part of the root `config/kustomization.yaml` and not covered by `kubectl apply -k config/` — unlike every other component here, it has no generic default: the deployer must hand-author one `fabric-router.` ConfigMap per matching node before applying `kubectl apply -k config/fabric-router/` (which also applies the agent's ServiceAccount and namespaced Role). The opt-in Component `config/fabric-router/components/fabric-api` adds the fabric-api sidecar to this DaemonSet's pod (see `config/fabric-api/` below); applying it, and every later fabric-api image change, recreates the pod and restarts FRR with it, so roll it out as a routing change. - **`config/bmp-collector/`** — a single-replica GoBMP Deployment and its `bmp-collector` Service (TCP `5000`), the BMP (RFC 7854) collector `galactic-router` and `fabric-router` stream to; `galactic-router` does so when `GALACTIC_ROUTER_BMP_STATIONS` names it (`bmp-collector.galactic-system.svc:5000`; see [docs/router/configuration.md](docs/router/configuration.md)). It parses every Peer Up/Down, route-monitoring and statistics message to JSON on stdout, so the cluster's log pipeline is its store; an overlay swapping `--dump=console` for `--kafka-server` publishes to Kafka instead. Opt-in, like `config/fabric-router/`: not part of the root `config/kustomization.yaml`, since nothing streams to it until a router is configured to. The containerlab lab runs one on `iad-worker3`'s host network (`deploy/containerlab/resources/bmp-collector/iad/`), because its three clusters cannot reach each other's Services, and points every node's `fabric-router` and the EVPN route reflector at it. Of the `galactic-router`s, only the reflector streams: its pre-policy Adj-RIB-In already holds every node's advertisements. - **`config/fabric-api/`** — the fabric looking glass's cell-wide prerequisites and per-cell gateway. `kubectl apply -k config/fabric-api/` applies only the `fabric-api-gateway` and `fabric-api-janitor` ServiceAccounts and RBAC (the janitor's delete access is deliberately a separate identity from the gateway's, which can only read `FabricQuery` and write its status); safe and idempotent in any cell. `config/fabric-api/base/` (the 2-replica leader-elected gateway Deployment, its ClusterIP debug Service, a PodDisruptionBudget and the `*/15` janitor CronJob) is **not** included in that kustomization and is **not** applied as-is, for the same reason as `config/galactic-gateway/base/`: it needs per-cell values — a `fabric-api` ConfigMap (`site`, `cell`, `cluster-name`) and the cell name in its certificate's URI SAN — supplied by a per-cell overlay; see `deploy/containerlab/resources/fabric-api/gateway/` for a worked example. `config/fabric-api/crd/` is a generated `FabricQuery` CRD for the lab and tests only; infra owns the CRD (and the per-cell `fabric-api` cert-manager `Issuer`) in every cell. The node sidecar is the `config/fabric-router/components/fabric-api` Component, which also adds a `fabric-api-certs` DaemonSet (`fabric-api certsync`) that owns the cert-manager csi-driver volume and copies the certificate to the node's `/run/fabric-api/tls` for the sidecar: a csi volume blocks its pod until the certificate is issued, so keeping it out of the `fabric-router` pod means a broken Issuer stalls only the certs pod, never FRR. The overlay's `CELL_NAME` replacement applies to that DaemonSet's `fabric-api-csi` volume. Not part of the root `config/kustomization.yaml`: fabric-api is opt-in. See [docs/fabric-api/configuration.md](docs/fabric-api/configuration.md) and [docs/fabric-api/api.md](docs/fabric-api/api.md). -- **`config/crd/`** and **`config/galactic-controller/`** — the `galactic.datumapis.com` CRDs (TrafficPolicy, TrafficPolicyBinding) and `galactic-controller --role=cell`, the TrafficPolicy validating webhook (fail closed), default-policy provisioning and binding resolution. `kubectl apply -k config/galactic-controller/` applies only its ServiceAccount and RBAC; `base/` (two-replica Deployment, webhook Service, cert-manager serving certificate, ValidatingWebhookConfiguration) is installed with the CRDs (including the cloud operator's `VPCAttachment`, which it watches) and cert-manager in place. Neither is in the root `config/kustomization.yaml`: nothing enforces traffic policy yet. See [docs/traffic-policy/controller.md](docs/traffic-policy/controller.md). +- **`config/crd/`** and **`config/galactic-controller/`** — the `galactic.datumapis.com` CRDs (TrafficPolicy, TrafficPolicyBinding) and `galactic-controller --role=cell`, the TrafficPolicy validating webhook (fail closed), default-policy provisioning and binding resolution. `kubectl apply -k config/galactic-controller/` applies only its ServiceAccount and RBAC; `base/` (two-replica Deployment, webhook Service, cert-manager serving certificate, ValidatingWebhookConfiguration) is installed with the CRDs (including the cloud operator's `VPCAttachment`, which it watches) and cert-manager in place. Neither is in the root `config/kustomization.yaml`. The node side is `galactic-router` with `GALACTIC_ROUTER_TRAFFIC_POLICY=true` (off by default; `internal/trafficpolicy/node`), which publishes each attachment's bound policies into the uSID datapath, writes the binding's `status.nodes` and exports `galactic_cni_traffic_policy_*`; galactic-cni's `GALACTIC_CNI_TRAFFIC_POLICY` decides whether the datapath enforces them. See [docs/traffic-policy/controller.md](docs/traffic-policy/controller.md) and [docs/traffic-policy/node.md](docs/traffic-policy/node.md). - **`config/monitoring/`** — Prometheus Operator resources for BGP and NAT: a `fabric-router` PodMonitor (frr-exporter, port `frr-metrics`/`9342`, as `job=fabric-router`; fabric-config-agent, port `agent-metrics`/`9343`, as `job=fabric-config-agent`), a `galactic-router` PodMonitor covering both roles (port `metrics`/`9179`), a `galactic-nat` PodMonitor for the egress shards (port `metrics`/`9182`), each adding a `node` label, and the `galactic-bgp` PrometheusRule: underlay session down, administratively down for an hour, flapping, received prefixes halved, exporter collector or scrape failing, a session announcing an IPv4-mapped next hop, BGP routes not installed, the agent's metrics scrape down, a `galactic-router` BMP station or metrics scrape down, `galactic-router` routes failing to install, an egress shard's session table evicting live sessions, an egress shard out of masquerade ports, a stalled or persistently lossy NIC receive queue on a `galactic-nat` uplink (#673), and a `galactic-nat` metrics scrape down. For the opt-in fabric-api it adds the `fabric-router` PodMonitor's `api-metrics` endpoint (the sidecar, port `api-metrics`/`9345`, as `job=fabric-api-node`), a `fabric-api-gateway` PodMonitor (port `metrics`/`9347`) and a `fabric-api` rule group (stalled execution, node diagnostics unavailable, nodes unreachable, oversized results, cleanup backlog, certificate renewal failing, credentials missing, search index unsynced, scrape down — none fires merely because nobody ran a query). Requires the `monitoring.coreos.com` CRDs, which this repo does not install, so it is not part of the root `config/kustomization.yaml`. `task test:alerts` unit-tests the rules with `promtool` (`config/monitoring/tests/`). Every alert carries what infra's alerting standard checks for (`service: galactic`, `team: connect`, `runbook_url` into infra's `docs/runbooks/galactic/README.md`). The PrometheusRule here is the tested source, not where the rules run on Datum's infra: there the PodMonitors are converted and scraped on each edge cluster, but edge vmalert never notifies, so infra evaluates a copy of the rules centrally. Change the rules here first, with their tests, then update infra's copy. - **`deploy/containerlab/`** — ContainerLab topology (`gvpc.clab.yaml`) for three Kind clusters (dfw, iad, sjc) wired over an SRv6 transit mesh whose underlay is dual-stack — every transit link and loopback carries an IPv4 and an IPv6 address, with one BGP session per address family (the SRv6 data plane itself stays IPv6-only; tenant IPv4 rides inside the encapsulation). Each site has one `compute` worker and at least one `edge` worker (dfw has two); every worker has its own eBGP uplink to its site's transit router, with FRR running as a hostNetwork DaemonSet. `galactic-router` (plain mode) distributes EVPN paths over iBGP to one route reflector for the whole lab (`iad-worker3`, `config/galactic-router/overlays/control/`) — compute and edge nodes never peer directly. Every `edge` node runs an egress shard (`galactic-nat`) and a `galactic-gateway`, which share the node's uplinks through its XDP dispatcher (`GALACTIC_NAT_XDP_ATTACH=dispatch`). Each site's compute node egresses only through its own site's edge shards, and each site has its own VIP `/64`, served only through its own edge nodes to one `ns60` backend in that site. A standalone off-cluster host on the fourth transit router gives tenant egress a real destination and is the client for the VIP checks (`verify:gateway-ingress`). See `deploy/containerlab/README.md` and `deploy/containerlab/Taskfile.yaml` for bring-up commands. diff --git a/cmd/galactic-router/main.go b/cmd/galactic-router/main.go index cd39acee..a55bc229 100644 --- a/cmd/galactic-router/main.go +++ b/cmd/galactic-router/main.go @@ -14,6 +14,8 @@ import ( authorizationv1 "k8s.io/api/authorization/v1" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" + + galacticv1alpha1 "go.datum.net/galactic/api/v1alpha1" ) const ( @@ -45,7 +47,7 @@ func main() { // resource the manager watches. A denial means the informer cache never syncs // and every reconciler is silently blocked, so this logs a clear, actionable // message at startup instead. -func checkWatchPermissions(mgr ctrl.Manager) { +func checkWatchPermissions(mgr ctrl.Manager, trafficPolicy bool) { 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") @@ -76,6 +78,17 @@ func checkWatchPermissions(mgr ctrl.Manager) { {version: "v1", resource: resourceSecrets}, {version: "v1", resource: resourceNodes}, } + if trafficPolicy { + resources = append(resources, + struct{ group, version, resource, namespace string }{ + group: galacticv1alpha1.GroupVersion.Group, version: galacticv1alpha1.GroupVersion.Version, + resource: "trafficpolicies", + }, + struct{ group, version, resource, namespace string }{ + group: galacticv1alpha1.GroupVersion.Group, version: galacticv1alpha1.GroupVersion.Version, + resource: "trafficpolicybindings", + }) + } for _, r := range resources { review := &authorizationv1.SelfSubjectAccessReview{ diff --git a/cmd/galactic-router/policy.go b/cmd/galactic-router/policy.go new file mode 100644 index 00000000..b278a30a --- /dev/null +++ b/cmd/galactic-router/policy.go @@ -0,0 +1,419 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package main + +import ( + "context" + "errors" + "fmt" + "io" + "net/netip" + "os" + "strconv" + "strings" + "text/tabwriter" + + "github.com/google/uuid" + "github.com/spf13/cobra" + "github.com/vishvananda/netlink" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/util/validation/field" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/yaml" + + cloudv1alpha1 "go.datum.net/cloud/api/v1alpha1" + galacticv1alpha1 "go.datum.net/galactic/api/v1alpha1" + "go.datum.net/galactic/internal/plumbing/ebpf/attach" + "go.datum.net/galactic/internal/plumbing/ebpf/tpmap" + "go.datum.net/galactic/internal/trafficpolicy" + tpnode "go.datum.net/galactic/internal/trafficpolicy/node" +) + +// newPolicyCommand builds the "policy" subcommand group, the traffic policy +// diagnostics. +func newPolicyCommand() *cobra.Command { + cmd := &cobra.Command{ + Use: "policy", + Short: "Explain a TrafficPolicy's verdict or show what this node applies", + } + cmd.AddCommand(newPolicyExplainCommand(), newPolicyShowCommand()) + return cmd +} + +// explainFlags is a packet tuple and the policy to decide it against. +type explainFlags struct { + file, policy string + path, src, dst, proto string + srcPort, dstPort, icmpType, icmpC uint16 +} + +func newPolicyExplainCommand() *cobra.Command { + var f explainFlags + cmd := &cobra.Command{ + Use: "explain", + Short: "Decide one packet against a TrafficPolicy", + Long: `explain decides a packet tuple against a TrafficPolicy with the Go evaluator +and against the entries it compiles to, and prints the winning rule. It reads +the policy from a file (--file) or from the cluster (--policy namespace/name), +and never touches this node's datapath or counters. The two results always +agree; a disagreement is a compiler bug and fails the command.`, + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + spec, name, err := loadPolicy(cmd.Context(), f) + if err != nil { + return err + } + pkt, err := f.packet() + if err != nil { + return err + } + return explain(cmd.OutOrStdout(), name, spec, pkt) + }, + } + fl := cmd.Flags() + fl.StringVar(&f.file, "file", "", "a TrafficPolicy manifest") + fl.StringVar(&f.policy, "policy", "", "a TrafficPolicy in the cluster, as namespace/name") + fl.StringVar(&f.path, "path", "", "the packet's path: Private, PublicNAT, PublicService, or empty for unknown") + fl.StringVar(&f.src, "src", "", "source address (required)") + fl.StringVar(&f.dst, "dst", "", "destination address (required)") + fl.StringVar(&f.proto, "proto", "tcp", "protocol: tcp, udp, icmp, icmpv6 or a number") + fl.Uint16Var(&f.srcPort, "sport", 0, "TCP or UDP source port") + fl.Uint16Var(&f.dstPort, "dport", 0, "TCP or UDP destination port") + fl.Uint16Var(&f.icmpType, "icmp-type", 0, "ICMP type") + fl.Uint16Var(&f.icmpC, "icmp-code", 0, "ICMP code") + cmd.MarkFlagsMutuallyExclusive("file", "policy") + cmd.MarkFlagsOneRequired("file", "policy") + return cmd +} + +// loadPolicy reads the policy f names and returns its spec and a display name. +func loadPolicy(ctx context.Context, f explainFlags) (galacticv1alpha1.TrafficPolicySpec, string, error) { + if f.file != "" { + raw, err := os.ReadFile(f.file) + if err != nil { + return galacticv1alpha1.TrafficPolicySpec{}, "", err + } + var p galacticv1alpha1.TrafficPolicy + if err := yaml.UnmarshalStrict(raw, &p); err != nil { + return galacticv1alpha1.TrafficPolicySpec{}, "", fmt.Errorf("parse %s: %w", f.file, err) + } + return p.Spec, p.Name, nil + } + key, err := parseNamespacedName(f.policy) + if err != nil { + return galacticv1alpha1.TrafficPolicySpec{}, "", err + } + c, err := policyClient() + if err != nil { + return galacticv1alpha1.TrafficPolicySpec{}, "", err + } + var p galacticv1alpha1.TrafficPolicy + if err := c.Get(ctx, key, &p); err != nil { + return galacticv1alpha1.TrafficPolicySpec{}, "", err + } + return p.Spec, p.Name, nil +} + +func (f explainFlags) packet() (trafficpolicy.Packet, error) { + var pkt trafficpolicy.Packet + switch f.path { + case "": + pkt.Path = trafficpolicy.PathUnknown + case string(galacticv1alpha1.PathPrivate): + pkt.Path = trafficpolicy.PathPrivate + case string(galacticv1alpha1.PathPublicNAT): + pkt.Path = trafficpolicy.PathPublicNAT + case string(galacticv1alpha1.PathPublicService): + pkt.Path = trafficpolicy.PathPublicService + default: + return pkt, fmt.Errorf("unknown path %q", f.path) + } + var err error + if pkt.Src, err = netip.ParseAddr(f.src); err != nil { + return pkt, fmt.Errorf("--src: %w", err) + } + if pkt.Dst, err = netip.ParseAddr(f.dst); err != nil { + return pkt, fmt.Errorf("--dst: %w", err) + } + pkt.Src, pkt.Dst = pkt.Src.Unmap(), pkt.Dst.Unmap() + if pkt.Src.Is4() != pkt.Dst.Is4() { + return pkt, errors.New("--src and --dst must be the same address family") + } + if pkt.Proto, err = parseProto(f.proto); err != nil { + return pkt, err + } + if f.icmpType > 255 || f.icmpC > 255 { + return pkt, errors.New("--icmp-type and --icmp-code must be at most 255") + } + pkt.SrcPort, pkt.DstPort = f.srcPort, f.dstPort + pkt.ICMPType, pkt.ICMPCode = uint8(f.icmpType), uint8(f.icmpC) //nolint:gosec // checked above + return pkt, nil +} + +func parseProto(s string) (uint8, error) { + switch strings.ToLower(s) { + case "tcp": + return trafficpolicy.ProtoTCP, nil + case "udp": + return trafficpolicy.ProtoUDP, nil + case "icmp": + return trafficpolicy.ProtoICMP, nil + case "icmpv6": + return trafficpolicy.ProtoICMPv6, nil + } + n, err := strconv.ParseUint(s, 10, 8) + if err != nil { + return 0, fmt.Errorf("unknown protocol %q", s) + } + return uint8(n), nil +} + +// explain prints spec's verdict for pkt, from the evaluator and from the +// compiled entries, and fails when they disagree. +func explain(w io.Writer, name string, spec galacticv1alpha1.TrafficPolicySpec, pkt trafficpolicy.Packet) error { + model, errs := trafficpolicy.FromAPI(spec, field.NewPath("spec")) + if len(errs) > 0 { + return fmt.Errorf("policy %s is invalid: %w", name, errs.ToAggregate()) + } + want := trafficpolicy.Evaluate(model, pkt) + entries, err := trafficpolicy.Compile(model) + if err != nil { + return err + } + got := trafficpolicy.EvaluateEntries(entries, pkt) + + action := "deny" + if want.Allow { + action = "allow" + } + rule := "-" + if want.Rule >= 0 { + rule = fmt.Sprintf("%s (#%d)", model.Rules[want.Rule].Name, want.Rule) + } + tw := tabwriter.NewWriter(w, 0, 4, 2, ' ', 0) + out := &printer{w: tw} + out.printf("policy\t%s\n", name) + out.printf("path\t%s\n", pkt.Path) + out.printf("verdict\t%s\n", action) + out.printf("reason\t%s\n", want.Reason) + out.printf("rule\t%s\n", rule) + out.printf("compiled entries\t%d\n", len(entries)) + if err := out.flush(tw); err != nil { + return err + } + if got != want { + return fmt.Errorf("the compiled entries decide %+v but the evaluator decides %+v: report this as a bug", + got, want) + } + return nil +} + +func newPolicyShowCommand() *cobra.Command { + var pinDir string + cmd := &cobra.Command{ + Use: "show /", + Short: "Show what this node applies to one VPCAttachment's interface", + Long: `show prints a VPCAttachment's binding, this node's status entry for it, and +the snapshot each direction of its interface holds in this node's datapath, +with every rule's counters. Run it in the galactic-router pod on the node +the attachment is on.`, + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + key, err := parseNamespacedName(args[0]) + if err != nil { + return err + } + c, err := policyClient() + if err != nil { + return err + } + maps, err := tpmap.OpenPinned(pinDir) + if err != nil { + return fmt.Errorf("open traffic policy maps: %w", err) + } + defer maps.Close() + return show(cmd.Context(), cmd.OutOrStdout(), c, maps, key) + }, + } + cmd.Flags().StringVar(&pinDir, "pin-dir", attach.PinDir, + "bpffs directory the uSID datapath's maps are pinned under") + return cmd +} + +func show(ctx context.Context, w io.Writer, c client.Client, maps *tpmap.Maps, key types.NamespacedName) error { + var att cloudv1alpha1.VPCAttachment + if err := c.Get(ctx, key, &att); err != nil { + return err + } + if att.Status.HostInterface == "" { + return fmt.Errorf("VPCAttachment %s has no host interface yet", key) + } + link, err := netlink.LinkByName(att.Status.HostInterface) + if err != nil { + return fmt.Errorf("interface %s is not on this node (VPCAttachment is on %q): %w", + att.Status.HostInterface, att.Status.Node, err) + } + ifindex := uint32(link.Attrs().Index) //nolint:gosec // a kernel ifindex + var binding galacticv1alpha1.TrafficPolicyBinding + bindingFound := c.Get(ctx, key, &binding) == nil + + tw := tabwriter.NewWriter(w, 0, 4, 2, ' ', 0) + out := &printer{w: tw} + out.printf("attachment\t%s (uid %s)\n", key, att.UID) + out.printf("interface\t%s (ifindex %d)\n", att.Status.HostInterface, ifindex) + for _, d := range []struct { + dir uint32 + name string + ref *galacticv1alpha1.LocalPolicyReference + st func(galacticv1alpha1.NodeBindingStatus) *galacticv1alpha1.AppliedPolicy + }{ + {tpmap.DirIngress, "ingress", binding.Spec.IngressPolicyRef, + func(n galacticv1alpha1.NodeBindingStatus) *galacticv1alpha1.AppliedPolicy { return n.Ingress }}, + {tpmap.DirEgress, "egress", binding.Spec.EgressPolicyRef, + func(n galacticv1alpha1.NodeBindingStatus) *galacticv1alpha1.AppliedPolicy { return n.Egress }}, + } { + bound := galacticv1alpha1.DefaultPolicyName + " (built in)" + if bindingFound && d.ref != nil { + bound = d.ref.Name + } + out.printf("\n%s\tbound to %s\n", d.name, bound) + for _, n := range binding.Status.Nodes { + if a := d.st(n); n.Node == att.Status.Node && a != nil { + out.printf(" status\t%s %s\n", a.State, a.Message) + } + } + snap, ok, err := maps.ReadSnapshot(ifindex, d.dir) + if err != nil { + return err + } + if !ok { + out.printf(" snapshot\tnone: denies as policy_unavailable\n") + continue + } + out.printf(" snapshot\tpolicy uid %s, generation %d, %d entries\n", + snapshotUID(snap.PolicyUID), snap.Generation, len(snap.Entries)) + names := ruleNames(ctx, c, key.Namespace, d.ref, snap) + seen := map[uint32]bool{} + for _, e := range snap.Entries { + row := snap.CounterBase + uint32(e.Rule) + if seen[row] { + continue + } + seen[row] = true + packets, bytes, err := maps.RuleCounter(row) + if err != nil { + return err + } + action := "deny" + if e.Flags&trafficpolicy.FlagAllow != 0 { + action = "allow" + } + name, ok := names[row] + if !ok { + name = fmt.Sprintf("row %d", row) + } + out.printf(" rule %s\t%s: %d packets, %d bytes\n", name, action, packets, bytes) + } + for reason, rn := range tpmap.ReasonNames { + packets, bytes, err := maps.FallbackCounter(ifindex, snap.Token, d.dir, uint8(reason)) //nolint:gosec // tiny + if err == nil && packets > 0 { + out.printf(" fallback %s\t%d packets, %d bytes\n", rn, packets, bytes) + } + } + } + return out.flush(tw) +} + +// ruleNames names a snapshot's counter rows by rule when the snapshot holds +// exactly what the policy now in the cluster compiles to, and returns nothing +// otherwise, so a stale snapshot is never labeled with today's rule names. +func ruleNames(ctx context.Context, c client.Client, ns string, ref *galacticv1alpha1.LocalPolicyReference, + snap tpmap.Snapshot, +) map[uint32]string { + identity := galacticv1alpha1.DefaultPolicyUID + model := trafficpolicy.Default() + if ref != nil { + var p galacticv1alpha1.TrafficPolicy + if err := c.Get(ctx, client.ObjectKey{Namespace: ns, Name: ref.Name}, &p); err != nil || + p.Generation != snap.Generation { + return nil + } + var errs field.ErrorList + if model, errs = trafficpolicy.FromAPI(p.Spec, field.NewPath("spec")); len(errs) > 0 { + return nil + } + identity = string(p.UID) + } + if tpnode.PolicyUIDBytes(identity) != snap.PolicyUID { + return nil + } + entries, err := trafficpolicy.Compile(model) + if err != nil { + return nil + } + rows, ok := tpnode.RuleRows(entries, snap, len(model.Rules)) + if !ok { + return nil + } + names := map[uint32]string{} + for i, row := range rows { + if row != tpnode.NoRow { + names[row] = model.Rules[i].Name + } + } + return names +} + +// snapshotUID prints a snapshot's recorded policy identity. +func snapshotUID(b [16]byte) string { + if b == tpnode.PolicyUIDBytes(galacticv1alpha1.DefaultPolicyUID) { + return galacticv1alpha1.DefaultPolicyUID + } + return uuid.UUID(b).String() +} + +func parseNamespacedName(s string) (types.NamespacedName, error) { + ns, name, ok := strings.Cut(s, "/") + if !ok || ns == "" || name == "" { + return types.NamespacedName{}, fmt.Errorf("%q is not namespace/name", s) + } + return types.NamespacedName{Namespace: ns, Name: name}, nil +} + +func policyClient() (client.Client, error) { + cfg, err := ctrl.GetConfig() + if err != nil { + return nil, err + } + scheme := runtime.NewScheme() + if err := galacticv1alpha1.AddToScheme(scheme); err != nil { + return nil, err + } + if err := cloudv1alpha1.AddToScheme(scheme); err != nil { + return nil, err + } + return client.New(cfg, client.Options{Scheme: scheme}) +} + +// printer writes formatted lines and keeps the first error. +type printer struct { + w io.Writer + err error +} + +func (p *printer) printf(format string, args ...any) { + if p.err == nil { + _, p.err = fmt.Fprintf(p.w, format, args...) + } +} + +func (p *printer) flush(tw *tabwriter.Writer) error { + if p.err != nil { + return p.err + } + return tw.Flush() +} diff --git a/cmd/galactic-router/policy_test.go b/cmd/galactic-router/policy_test.go new file mode 100644 index 00000000..0b9548b8 --- /dev/null +++ b/cmd/galactic-router/policy_test.go @@ -0,0 +1,106 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package main + +import ( + "bytes" + "os" + "path/filepath" + "strings" + "testing" +) + +const explainPolicy = `apiVersion: galactic.datumapis.com/v1alpha1 +kind: TrafficPolicy +metadata: + name: web + namespace: tenant +spec: + rules: + - name: private-http + action: Allow + match: + path: Private + protocol: TCP + destinationPorts: + - start: 80 + end: 80 + - name: block-dns + action: Deny + match: + protocol: UDP + destinationPorts: + - start: 53 + end: 53 +` + +const verdictDeny = "verdict deny" + +func runExplain(t *testing.T, args ...string) (string, error) { + t.Helper() + file := filepath.Join(t.TempDir(), "policy.yaml") + if err := os.WriteFile(file, []byte(explainPolicy), 0o600); err != nil { + t.Fatal(err) + } + cmd := newPolicyCommand() + var out bytes.Buffer + cmd.SetOut(&out) + cmd.SetErr(&out) + cmd.SetArgs(append([]string{"explain", "--file", file}, args...)) + err := cmd.Execute() + return out.String(), err +} + +// tuple is explain's packet flags: path (empty for none), addresses, and +// any further flags. +func tuple(path, src, dst string, extra ...string) []string { + args := []string{"--src", src, "--dst", dst} + if path != "" { + args = append(args, "--path", path) + } + return append(args, extra...) +} + +func TestPolicyExplain(t *testing.T) { + for _, tt := range []struct { + name string + args []string + want []string + }{ + {"rule allows", tuple("Private", "10.0.0.1", "10.0.0.2", "--dport", "80"), + []string{"verdict allow", "private-http (#0)"}}, + {"rule denies", tuple("Private", "fd00::1", "fd00::2", "--proto", "udp", "--dport", "53"), + []string{verdictDeny, "block-dns (#1)", "reason rule"}}, + {"no rule matches", tuple("PublicNAT", "10.0.0.1", "10.0.0.2", "--dport", "80"), + []string{verdictDeny, "reason no_match"}}, + {"unknown path at a path rule", tuple("", "10.0.0.1", "10.0.0.2", "--dport", "80"), + []string{verdictDeny, "reason classification_error"}}, + } { + t.Run(tt.name, func(t *testing.T) { + out, err := runExplain(t, tt.args...) + if err != nil { + t.Fatalf("explain: %v\n%s", err, out) + } + for _, w := range tt.want { + if !strings.Contains(out, w) { + t.Errorf("output lacks %q:\n%s", w, out) + } + } + }) + } +} + +func TestPolicyExplainRejectsBadInput(t *testing.T) { + for _, args := range [][]string{ + tuple("", "10.0.0.1", "fd00::2"), + tuple("Elsewhere", "10.0.0.1", "10.0.0.2"), + tuple("", "10.0.0.1", "10.0.0.2", "--proto", "sctp"), + tuple("", "nope", "10.0.0.2"), + } { + if _, err := runExplain(t, args...); err == nil { + t.Errorf("explain %v succeeded", args) + } + } +} diff --git a/cmd/galactic-router/root.go b/cmd/galactic-router/root.go index cd97d97d..16c7d9f6 100644 --- a/cmd/galactic-router/root.go +++ b/cmd/galactic-router/root.go @@ -27,6 +27,7 @@ import ( metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server" cloudv1alpha1 "go.datum.net/cloud/api/v1alpha1" + galacticv1alpha1 "go.datum.net/galactic/api/v1alpha1" "go.datum.net/galactic/internal/config" "go.datum.net/galactic/internal/controller" "go.datum.net/galactic/internal/hash" @@ -38,6 +39,7 @@ import ( galacticruntime "go.datum.net/galactic/internal/runtime" "go.datum.net/galactic/internal/runtime/gobgp" "go.datum.net/galactic/internal/serviceroute" + tpnode "go.datum.net/galactic/internal/trafficpolicy/node" bgpv1alpha1 "go.datum.net/network/api/v1alpha1" ) @@ -83,6 +85,7 @@ func runCmd(cfg *config.RouterConfig) error { utilruntime.Must(clientgoscheme.AddToScheme(scheme)) utilruntime.Must(bgpv1alpha1.AddToScheme(scheme)) utilruntime.Must(cloudv1alpha1.AddToScheme(scheme)) + utilruntime.Must(galacticv1alpha1.AddToScheme(scheme)) mgrOptions := ctrl.Options{ Scheme: scheme, @@ -131,7 +134,7 @@ func runCmd(cfg *config.RouterConfig) error { }() // Pre-flight RBAC check. - checkWatchPermissions(mgr) + checkWatchPermissions(mgr, cfg.TrafficPolicy) // Open this node's own vip_xlat_table handle for the ServiceVIPBinding // reconciler's tap branch. The datapath program is loaded and attached @@ -278,6 +281,12 @@ func runCmd(cfg *config.RouterConfig) error { ctrl.Log.Info("route reflector mode: skipping the node-local ServiceRoutePolicy controller") } + if cfg.TrafficPolicy { + if err := setupTrafficPolicyController(mgr, nodeName); err != nil { + return err + } + } + // Register GC controller for cleaning up orphaned BGP CRDs and VRFs. gcRec := &controller.GCReconciler{ Client: mgr.GetClient(), @@ -389,6 +398,19 @@ func setupVIPBindingControllers( return nil } +// setupTrafficPolicyController registers the node's traffic policy +// reconciler and exports its counters on the manager's metrics endpoint. +func setupTrafficPolicyController(mgr ctrl.Manager, nodeName string) error { + rec := &tpnode.Reconciler{Client: mgr.GetClient(), NodeName: nodeName, PinDir: attach.PinDir} + if err := rec.SetupWithManager(mgr); err != nil { + return fmt.Errorf("setup traffic policy controller: %w", err) + } + if err := ctrlmetrics.Registry.Register(rec); err != nil { + return fmt.Errorf("register traffic policy metrics: %w", err) + } + return nil +} + // serviceRouteControllerEnabled reports whether this router programs // node-local service routes. Route reflectors have no CNI attach point and // therefore no service route maps. @@ -465,6 +487,8 @@ func newRootCommand() *cobra.Command { "Cleanup interval") cmd.Flags().Bool("service-frontend-enabled", false, "Enable consumer-facing private-service address translation") + cmd.Flags().Bool("traffic-policy", false, + "Enforce TrafficPolicy on this node's tenant interfaces; needs the galactic.datumapis.com CRDs") cmd.Flags().String("bmp-stations", "", "Comma-separated BMP collectors to stream to, each host:port; empty disables BMP") cmd.Flags().String("bmp-policy", config.DefaultRouterBMPPolicy, @@ -474,6 +498,6 @@ func newRootCommand() *cobra.Command { cmd.Flags().Bool("build-info", false, "Print build information and exit") cmd.Flags().BoolP("version", "V", false, "Print version and exit") - cmd.AddCommand(newVIPCommand()) + cmd.AddCommand(newVIPCommand(), newPolicyCommand()) return cmd } diff --git a/cmd/galactic-router/root_test.go b/cmd/galactic-router/root_test.go index f5b852ca..46ce3b7f 100644 --- a/cmd/galactic-router/root_test.go +++ b/cmd/galactic-router/root_test.go @@ -27,6 +27,7 @@ func testCmd(t *testing.T) *cobra.Command { cmd.Flags().IntP("metrics-port", "", config.DefaultRouterMetricsPort, "Metrics listen port") cmd.Flags().IntP("grpc-health-port", "", config.DefaultRouterGRPCHealthPort, "gRPC health check port") cmd.Flags().StringP("gc-namespace", "", config.DefaultRouterGCNamespace, "Namespace for orphaned CRD cleanup") + cmd.Flags().Bool("traffic-policy", false, "Enforce TrafficPolicy") cmd.Flags().DurationP("gc-interval", "", config.DefaultRouterGCInterval, "Cleanup interval") return cmd } diff --git a/config/galactic-router/rbac.yaml b/config/galactic-router/rbac.yaml index bfd4090b..dbd93771 100644 --- a/config/galactic-router/rbac.yaml +++ b/config/galactic-router/rbac.yaml @@ -62,6 +62,18 @@ rules: - vpcs - vpcattachments verbs: ["get", "list", "watch"] + # The traffic policy reconciler (GALACTIC_ROUTER_TRAFFIC_POLICY) reads + # policies and bindings and applies this node's entry in each binding's + # status.nodes. It never writes either resource itself. + - apiGroups: ["galactic.datumapis.com"] + resources: + - trafficpolicies + - trafficpolicybindings + verbs: ["get", "list", "watch"] + - apiGroups: ["galactic.datumapis.com"] + resources: + - trafficpolicybindings/status + verbs: ["get", "patch"] - apiGroups: [""] resources: - secrets diff --git a/docs/README.md b/docs/README.md index 12a55eae..fed97eba 100644 --- a/docs/README.md +++ b/docs/README.md @@ -60,6 +60,10 @@ data flow, module reference, known constraints): - [traffic-policy/controller.md](traffic-policy/controller.md) — `galactic-controller --role=cell`: the TrafficPolicy webhook, the reserved default policy, binding resolution status, flags and install order. +- [traffic-policy/node.md](traffic-policy/node.md) — galactic-router's node + side: enabling it, what each binding applies, per-node status, the + `galactic_cni_traffic_policy_*` metrics and the `policy explain` and + `policy show` diagnostics. - [api/galactic.md](api/galactic.md) — the generated `galactic.datumapis.com` API reference (TrafficPolicy, TrafficPolicyBinding). - [cni/README.md](cni/README.md) — entry point for the `galactic-cni` docs diff --git a/docs/agents/ARCHITECTURE-ROUTER.md b/docs/agents/ARCHITECTURE-ROUTER.md index 1d3eadf9..3861196f 100644 --- a/docs/agents/ARCHITECTURE-ROUTER.md +++ b/docs/agents/ARCHITECTURE-ROUTER.md @@ -197,19 +197,20 @@ there instead. ### galactic-router environment variables -| Variable | Required | Default | Description | -| ----------------------------------------- | -------- | ----------------- | ---------------------------------------------------------------------------------------------------- | -| `GALACTIC_ROUTER_NODE_NAME` | Yes | — | Kubernetes node name; filters which BGPRouter CRDs this instance owns | -| `GALACTIC_ROUTER_REFLECTOR` | No | `false` | Marks every peer as an iBGP route-reflector client; independent of `GALACTIC_ROUTER_BGP_LISTEN_PORT` | -| `GALACTIC_ROUTER_BGP_LISTEN_PORT` | No | `179` | BGP TCP listen port; `-1` disables inbound connections (outbound-only) | -| `GALACTIC_ROUTER_BGP_LOCAL_ADDRESS` | No | — | Source address for outgoing BGP TCP connections (numbered underlay use) | -| `GALACTIC_ROUTER_METRICS_PORT` | No | `9179` | controller-runtime Prometheus metrics port | -| `GALACTIC_ROUTER_GRPC_HEALTH_PORT` | No | `5179` | gRPC health check port (liveness/readiness probes) | -| `GALACTIC_ROUTER_GC_NAMESPACE` | No | `galactic-system` | Namespace the GC controller scans for orphaned CRDs | -| `GALACTIC_ROUTER_GC_INTERVAL` | No | `5m` | GC controller sweep interval | -| `GALACTIC_ROUTER_BMP_STATIONS` | No | — | BMP collectors to stream to, comma-separated `host:port`; empty disables BMP | -| `GALACTIC_ROUTER_BMP_POLICY` | No | `pre-policy` | BMP route monitoring policy: `pre-policy`, `post-policy`, `local-rib`, or `all` | -| `GALACTIC_ROUTER_BMP_STATISTICS_INTERVAL` | No | `60s` | Interval between BMP statistics reports per peer; `0s` disables them | +| Variable | Required | Default | Description | +| ----------------------------------------- | -------- | ----------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------- | +| `GALACTIC_ROUTER_NODE_NAME` | Yes | — | Kubernetes node name; filters which BGPRouter CRDs this instance owns | +| `GALACTIC_ROUTER_REFLECTOR` | No | `false` | Marks every peer as an iBGP route-reflector client; independent of `GALACTIC_ROUTER_BGP_LISTEN_PORT` | +| `GALACTIC_ROUTER_BGP_LISTEN_PORT` | No | `179` | BGP TCP listen port; `-1` disables inbound connections (outbound-only) | +| `GALACTIC_ROUTER_BGP_LOCAL_ADDRESS` | No | — | Source address for outgoing BGP TCP connections (numbered underlay use) | +| `GALACTIC_ROUTER_METRICS_PORT` | No | `9179` | controller-runtime Prometheus metrics port | +| `GALACTIC_ROUTER_GRPC_HEALTH_PORT` | No | `5179` | gRPC health check port (liveness/readiness probes) | +| `GALACTIC_ROUTER_GC_NAMESPACE` | No | `galactic-system` | Namespace the GC controller scans for orphaned CRDs | +| `GALACTIC_ROUTER_GC_INTERVAL` | No | `5m` | GC controller sweep interval | +| `GALACTIC_ROUTER_BMP_STATIONS` | No | — | BMP collectors to stream to, comma-separated `host:port`; empty disables BMP | +| `GALACTIC_ROUTER_BMP_POLICY` | No | `pre-policy` | BMP route monitoring policy: `pre-policy`, `post-policy`, `local-rib`, or `all` | +| `GALACTIC_ROUTER_BMP_STATISTICS_INTERVAL` | No | `60s` | Interval between BMP statistics reports per peer; `0s` disables them | +| `GALACTIC_ROUTER_TRAFFIC_POLICY` | No | `false` | Runs the node's TrafficPolicy reconciler and exports its counters; needs the `galactic.datumapis.com` CRDs; refused with `GALACTIC_ROUTER_REFLECTOR` | See [docs/router/configuration.md](../router/configuration.md) for the full reference, including CLI flags and precedence. @@ -226,19 +227,20 @@ co-located `galactic-gateway` container's own health port on the same ## Module / Package Reference -| Package | Binary | Responsibility | Owns state | -| ------------------------ | ---------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------- | -| `internal/controller` | galactic-router | controller-runtime reconcilers (BGPRouter, BGPPeer, BGPAdvertisement, BGPVRFInstance, BGPPolicy, Node, Secret, ServiceVIPBinding, NetworkRuleBinding, NodeDeparture, GC); field index registration; CRD status helpers | No | -| `internal/reconcile` | galactic-router | Translates BGPRouter + related CRDs into `model.DesiredRouter`; enforces node targeting, timer validation, AFI validation | No | -| `internal/runtime` | galactic-router | `RouterRuntime` interface; `RuntimeManager` (keyed map of live runtimes, double-checked lock create) | Yes (runtime map) | -| `internal/runtime/gobgp` | galactic-router | Embeds GoBGP v4; lazy-starts on first Apply; handles peer/VRF/EVPN-path/policy add/update/delete; tracks established timestamps; logs peer FSM transitions (`peer_monitor.go`); keeps BMP stations registered across server replacement and collector address changes (`bmp.go`) | Yes (per-router) | -| `internal/model` | galactic-router | `DesiredRouter`, `DesiredPeer`, `DesiredAdvertisement`, `DesiredPolicy`, `DesiredVRFInstance`, `RuntimeStatus`; re-exports BGP API enums | No | -| `internal/hash` | galactic-router | SHA-256 fingerprint of `DesiredRouter` for no-op suppression | No | -| `internal/metadata` | every binary | Build-time vars (`Version`, `GitCommit`, `GitTreeState`, `BuildDate`) stamped via `-ldflags` | No | -| `internal/gc` | galactic-router | Collects orphaned `BGPAdvertisement`/`BGPVRFInstance` CRDs, stale kernel VRFs, and stale eBPF `vrf_table` entries; invoked by the GC controller's ticker | No | -| `internal/plumbing/srv6` | galactic-router | SID computation (`ComputeSID`) for the router's own BGP Prefix-SID path attribute | No | -| `internal/plumbing/intf` | galactic-router + every CNI-chain binary | Deterministic interface naming (`G{vpc9}{att3}V/H/G`); base62↔hex encoding | No | -| `internal/plumbing/vrf` | galactic-router + every CNI-chain binary | Linux VRF create/delete/lookup via netlink | No | +| Package | Binary | Responsibility | Owns state | +| ----------------------------- | ---------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------ | +| `internal/controller` | galactic-router | controller-runtime reconcilers (BGPRouter, BGPPeer, BGPAdvertisement, BGPVRFInstance, BGPPolicy, Node, Secret, ServiceVIPBinding, NetworkRuleBinding, NodeDeparture, GC); field index registration; CRD status helpers | No | +| `internal/reconcile` | galactic-router | Translates BGPRouter + related CRDs into `model.DesiredRouter`; enforces node targeting, timer validation, AFI validation | No | +| `internal/runtime` | galactic-router | `RouterRuntime` interface; `RuntimeManager` (keyed map of live runtimes, double-checked lock create) | Yes (runtime map) | +| `internal/runtime/gobgp` | galactic-router | Embeds GoBGP v4; lazy-starts on first Apply; handles peer/VRF/EVPN-path/policy add/update/delete; tracks established timestamps; logs peer FSM transitions (`peer_monitor.go`); keeps BMP stations registered across server replacement and collector address changes (`bmp.go`) | Yes (per-router) | +| `internal/model` | galactic-router | `DesiredRouter`, `DesiredPeer`, `DesiredAdvertisement`, `DesiredPolicy`, `DesiredVRFInstance`, `RuntimeStatus`; re-exports BGP API enums | No | +| `internal/hash` | galactic-router | SHA-256 fingerprint of `DesiredRouter` for no-op suppression | No | +| `internal/metadata` | every binary | Build-time vars (`Version`, `GitCommit`, `GitTreeState`, `BuildDate`) stamped via `-ldflags` | No | +| `internal/gc` | galactic-router | Collects orphaned `BGPAdvertisement`/`BGPVRFInstance` CRDs, stale kernel VRFs, and stale eBPF `vrf_table` entries; invoked by the GC controller's ticker | No | +| `internal/plumbing/srv6` | galactic-router | SID computation (`ComputeSID`) for the router's own BGP Prefix-SID path attribute | No | +| `internal/plumbing/intf` | galactic-router + every CNI-chain binary | Deterministic interface naming (`G{vpc9}{att3}V/H/G`); base62↔hex encoding | No | +| `internal/plumbing/vrf` | galactic-router + every CNI-chain binary | Linux VRF create/delete/lookup via netlink | No | +| `internal/trafficpolicy/node` | galactic-router | Traffic policy node reconciler (`GALACTIC_ROUTER_TRAFFIC_POLICY`): compiles each attachment's bound TrafficPolicy per direction, publishes it into the uSID datapath keeping each rule's counter row, adopts snapshots across restarts, writes the binding's `status.nodes` entry, and exports the counters | Yes (counter rows) | --- diff --git a/docs/enhancements/networking/interface-traffic-policy/README.md b/docs/enhancements/networking/interface-traffic-policy/README.md index 5940797c..873d8c35 100644 --- a/docs/enhancements/networking/interface-traffic-policy/README.md +++ b/docs/enhancements/networking/interface-traffic-policy/README.md @@ -699,23 +699,33 @@ TrafficPolicyReconciler: - compiles each direction, publishes and retires snapshots as above; - reports the applied identity and generation in `status.nodes`. -It reconciles on start, after every informer resync, and on a periodic sweep -that also removes outer-map rows for interfaces that no longer exist. It writes -no status on VPCAttachment. +It reconciles on start, on every watch event, and every five minutes. After a +restart it adopts each snapshot whose token, policy UID and generation still +match, so counters continue; one no attachment claims within a resync is +unbound and its rows freed. galactic-cni removes the outer-map rows of +interfaces that no longer exist. It writes no status on VPCAttachment. It runs +only when `GALACTIC_ROUTER_TRAFFIC_POLICY` is set, because it watches CRDs a +cluster may not have installed. A diagnostic command, `galactic-router policy show /`, -prints both bindings, the applied generations, the compiled entries and each -rule's counters. `galactic-router policy explain` evaluates a packet tuple, -direction and path against a policy file or a live binding using the Go -evaluator, and never touches live counters. +prints both bindings, this node's status entry, the bound snapshots and each +rule's counters. `galactic-router policy explain` evaluates a packet tuple and +path against a policy file or a live policy with both the Go evaluator and the +compiled entries, fails if they disagree, and never touches live counters. ### Metrics -Exported from galactic-cni's existing `:9180/metrics` endpoint, which already -has a PodMonitor: - -| Metric | Labels | -| ------------------------------------------------------ | ------------------------------------------------------------------------------------------------- | +Exported from galactic-router's existing `:9179/metrics` endpoint, which +already has a PodMonitor. galactic-router allocates every counter row and +knows which rule, policy and interface it belongs to, so it is the one +process that can label a row without a second lookup table. The +`galactic_cni_` prefix names the datapath the counters belong to, not the +process that serves them. A row the collector cannot read is left out, not +reported as zero, and counted in the `galactic_cni_traffic_policy_counter_read_errors` +gauge. + +| Metric | Labels | +| --------------------------------------------------------- | ------------------------------------------------------------------------------------------------- | | `galactic_cni_traffic_policy_rule_hits_total` | namespace, vpc, attachment, interface, interface_uid, direction, policy, policy_uid, rule, action | | `galactic_cni_traffic_policy_rule_bytes_total` | the same | | `galactic_cni_traffic_policy_fallback_packets_total` | namespace, vpc, attachment, interface, interface_uid, direction, reason | @@ -726,8 +736,9 @@ has a PodMonitor: - `reason` on the fallback metrics is one of `no_match`, `policy_unavailable`, `parse_error` or `classification_error`. - `interface_uid` is the VPCAttachment UID. -- The collector resolves labels from the snapshot metadata galactic-router - writes beside each pinned binding, never from packet fields. +- The collector labels each row from what galactic-router installed, never + from packet fields. After a restart, a row counts on in the kernel and is + exported again once its interface is reconciled and its snapshot adopted. - Each evaluation produces exactly one outcome: a winning rule's hit and bytes, or one fallback pair. The one exception is a packet captured while the mode was off and decided just after it turned on, at most one per CPU at the diff --git a/docs/metrics.md b/docs/metrics.md index fb82b78f..fef2133f 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -14,10 +14,11 @@ the 9182 default with `galactic-nat` without clashing, even on the edge nodes that run both. | Area | Component | Metric prefix | Port | `job` from `config/monitoring/` | -|------------------------|-----------------------------------------------------------|-----------------------------------------|----------------|------------------------------------------| +| ---------------------- | --------------------------------------------------------- | --------------------------------------- | -------------- | ---------------------------------------- | | VPC dataplane | `galactic-cni` (`credential-refresh`, the datapath owner) | `galactic_usid_*` | 9180 | none shipped | | VPC ingress | `galactic-vrf` | `galactic_ingress_sidecar_*` | 9182 (default) | none shipped | | Tenant routing (GoBGP) | `galactic-router`, `galactic-router-rr` | `galactic_router_*`, controller-runtime | 9179 | `galactic-router` / `galactic-router-rr` | +| Traffic policy | `galactic-router` with `GALACTIC_ROUTER_TRAFFIC_POLICY` | `galactic_cni_traffic_policy_*` | 9179 | `galactic-router` | | Underlay routing (FRR) | `fabric-router` `frr-exporter` sidecar | `frr_*` | 9342 | `fabric-router` | | Underlay config | `fabric-router` `fabric-config-agent` | `fabric_router_*` | 9343 | `fabric-config-agent` | | NAT66 / NAT64 | `galactic-nat` (one per egress shard) | `galactic_nat_*` | 9182 | `galactic-nat` | @@ -209,6 +210,78 @@ histogram_quantile(0.99, sum by (le, node) (rate(galactic_ingress_sidecar_reconc min_over_time(galactic_ingress_sidecar_vrf_teardown_pending[1h]) > 0 ``` +## Traffic policy — `galactic_cni_traffic_policy_*` + +Interface traffic policy decisions in the uSID datapath. The counters live in +per-CPU BPF maps the datapath writes; `galactic-router`, which allocates +every counter row and knows the rule, policy and interface it belongs to, +reads and labels them, so they are served on its port (9179) and scraped by +the `galactic-router` PodMonitor. The prefix names the datapath, not the +process. Only routers running with `GALACTIC_ROUTER_TRAFFIC_POLICY=true` +export them. See [traffic-policy/node.md](traffic-policy/node.md). + +| Metric | Type | Labels | Meaning | +| ------------------------------------------------------ | ------- | ------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------- | +| `galactic_cni_traffic_policy_rule_hits_total` | counter | namespace, vpc, attachment, interface, interface_uid, direction, policy, policy_uid, rule, action | Packets evaluated with this rule as the winning rule | +| `galactic_cni_traffic_policy_rule_bytes_total` | counter | the same | Tenant IP bytes evaluated with this rule as the winning rule | +| `galactic_cni_traffic_policy_fallback_packets_total` | counter | namespace, vpc, attachment, interface, interface_uid, direction, reason | Packets denied without a winning rule | +| `galactic_cni_traffic_policy_fallback_bytes_total` | counter | the same | Tenant IP bytes denied without a winning rule | +| `galactic_cni_traffic_policy_unattributed_drops_total` | counter | direction, reason | Policy denials whose interface or fallback row could not be resolved | +| `galactic_cni_traffic_policy_accounting_errors_total` | counter | direction, action, reason | Verdicts whose rule counter row was missing; the verdict stood, the count was lost | +| `galactic_cni_traffic_policy_counter_read_errors` | gauge | none | Counter rows the last scrape could not read, and so left out rather than reporting zero | + +- `direction` is `ingress` or `egress`; `action` is `allow` or `deny`. +- `reason` is one of `no_match`, `policy_unavailable`, `parse_error` or + `classification_error`; `accounting_errors_total` uses + `missing_rule_counter`. +- `interface_uid` is the VPCAttachment UID. `policy_uid` is the policy's + UID, or `builtin:default-deny-all:v1` for a direction with no reference, + which gets the built-in default. Its deny-all rule is a rule hit, not a + fallback, so the default stays visible per interface and direction. +- One evaluation produces one outcome: a rule's hit and bytes, or one + fallback pair. A packet between two managed interfaces is two + evaluations, the sender's egress and the receiver's ingress. +- A hit is a policy decision, not forwarded traffic: an allowed packet + dropped later still counts. +- Bytes are the tenant IP packet at the evaluation point, without Ethernet + or outer SRv6. A GSO packet counts once, at its buffer length. A packet + whose IP boundary cannot be found adds zero bytes. +- A rule keeps its series while its name and action stay the same within + one policy and interface lifetime, so editing its match keeps counting. + Rename a rule to start a fresh measurement. Rebinding another policy, + removing a rule, flipping its action, or recreating the interface starts + from zero. +- A node reboot or a datapath reload resets them. Use `rate()` or + `increase()`. + +```promql +# Rule hits per second, aggregated across nodes and interfaces. +sum by (namespace, policy, rule, direction, action) ( + rate(galactic_cni_traffic_policy_rule_hits_total[5m]) +) + +# Tenant bytes per second explicitly denied by a rule on each interface. +sum by (namespace, attachment, interface, interface_uid, direction, rule) ( + rate(galactic_cni_traffic_policy_rule_bytes_total{action="deny"}[5m]) +) + +# Failures and no-match denies, which are not attributed to a winning rule. +sum by (namespace, direction, reason) ( + rate(galactic_cni_traffic_policy_fallback_packets_total[5m]) +) + +# Interfaces still on the built-in default deny. +count by (namespace, attachment, direction) ( + galactic_cni_traffic_policy_rule_hits_total{policy_uid="builtin:default-deny-all:v1"} +) + +# Counts the datapath lost or could not attribute. +sum by (node, direction, reason) ( + rate(galactic_cni_traffic_policy_unattributed_drops_total[5m]) + or rate(galactic_cni_traffic_policy_accounting_errors_total[5m]) +) > 0 +``` + ## Routing Two planes: FRR (`fabric-router`) carries the underlay to the upstream diff --git a/docs/router/configuration.md b/docs/router/configuration.md index 7d8f4127..5e04b648 100644 --- a/docs/router/configuration.md +++ b/docs/router/configuration.md @@ -15,7 +15,7 @@ and the `EnvRouter*` constants). Always use the exact name from the table below. | Option | Environment Variable | CLI Flag | Default | -|--------------------------|-------------------------------------------|-----------------------------|-----------------------------| +| ------------------------ | ----------------------------------------- | --------------------------- | --------------------------- | | Node name | `GALACTIC_ROUTER_NODE_NAME` | `--node-name` | _(required)_ | | Route reflector | `GALACTIC_ROUTER_REFLECTOR` | `--reflector` | `false` | | BGP listen port | `GALACTIC_ROUTER_BGP_LISTEN_PORT` | `--bgp-listen-port` | `179` | @@ -27,6 +27,7 @@ below. | BMP stations | `GALACTIC_ROUTER_BMP_STATIONS` | `--bmp-stations` | _(empty; BMP disabled)_ | | BMP policy | `GALACTIC_ROUTER_BMP_POLICY` | `--bmp-policy` | `pre-policy` | | BMP statistics interval | `GALACTIC_ROUTER_BMP_STATISTICS_INTERVAL` | `--bmp-statistics-interval` | `60s` | +| Traffic policy | `GALACTIC_ROUTER_TRAFFIC_POLICY` | `--traffic-policy` | `false` | ## Required Options @@ -209,6 +210,23 @@ disables them. **Default:** `60s` **Valid values:** `0s`–`65535s` +### `--traffic-policy` / `GALACTIC_ROUTER_TRAFFIC_POLICY` + +Runs the node's traffic policy reconciler. For every VPCAttachment whose +`status.node` is this node, it compiles the TrafficPolicy each direction of +the interface is bound to, publishes it into the uSID datapath, and reports +what it applied in the TrafficPolicyBinding's `status.nodes`. It also exports +the policy counters on the metrics port, as +`galactic_cni_traffic_policy_*`, and enables the `galactic-router policy` +diagnostics' live lookups. + +Off by default, because it watches the `galactic.datumapis.com` CRDs, which a +cluster may not have installed: an informer for a missing CRD never syncs. A +route reflector has no tenant interfaces, so `--reflector` with this option is +a configuration error. Whether the datapath enforces what this publishes is +galactic-cni's setting, not this one. See +[the traffic policy design](../enhancements/networking/interface-traffic-policy/README.md). + ## Configuration Precedence Values are resolved in the following order (highest to lowest priority): diff --git a/docs/traffic-policy/controller.md b/docs/traffic-policy/controller.md index e03cfd5a..5d0aeffe 100644 --- a/docs/traffic-policy/controller.md +++ b/docs/traffic-policy/controller.md @@ -2,8 +2,8 @@ `galactic-controller --role=cell` runs once per edge cell and owns the API side of interface traffic policy. It does not program any node: the per-node -datapath state belongs to `galactic-router`, which a later change teaches to -read the bindings this controller resolves. The design is in +datapath state belongs to `galactic-router`, which reads the bindings this +controller resolves (see [Traffic policy on the node](node.md)). The design is in [the traffic policy proposal](../enhancements/networking/interface-traffic-policy/README.md); the API reference is [docs/api/galactic.md](../api/galactic.md). diff --git a/docs/traffic-policy/node.md b/docs/traffic-policy/node.md new file mode 100644 index 00000000..5c6bc011 --- /dev/null +++ b/docs/traffic-policy/node.md @@ -0,0 +1,126 @@ +# Traffic policy on the node + +`galactic-router` programs interface traffic policy on each node. For every +VPCAttachment whose `status.node` is its node, it compiles the TrafficPolicy +each direction of the interface is bound to, publishes it into the uSID +datapath, reports what it applied in the TrafficPolicyBinding's +`status.nodes`, and exports the policy counters. The cell side, which +validates policies and resolves bindings, is +[galactic-controller](controller.md); the design is in +[the traffic policy proposal](../enhancements/networking/interface-traffic-policy/README.md). + +## Enabling it + +Set `GALACTIC_ROUTER_TRAFFIC_POLICY=true` on the plain `galactic-router` +DaemonSet, never on the route reflector, which refuses it. Install first: + +1. the `galactic.datumapis.com` CRDs (`config/crd/`), since an informer for + a missing CRD never syncs; +2. the ClusterRole in `config/galactic-router/rbac.yaml`, which grants read + access to policies and bindings and `patch` on the bindings' status. + +Whether the datapath enforces, audits or ignores what galactic-router +publishes is galactic-cni's setting, `GALACTIC_CNI_TRAFFIC_POLICY`. With it +off, galactic-router still publishes and reports, and nothing is enforced. + +## What it applies + +| Binding | The direction gets | +| --------------------------------------------- | ----------------------------------------------------------------------------------- | +| No TrafficPolicyBinding | The built-in `default-deny-all`, which denies everything | +| A binding with no reference for a direction | The built-in `default-deny-all` | +| A reference to a valid policy | That policy, at its current generation | +| A reference to a missing policy | Nothing: the direction denies as `policy_unavailable` and its status is `Pending` | +| A reference to a policy that does not compile | The last snapshot of the same policy, if any, otherwise nothing; status is `Failed` | + +A reference to a `default-deny-all` the controller does not own resolves to +nothing, as on the cell side. + +Each rule keeps its counter row while its name and action stay the same +within one policy and one interface lifetime, so editing a match does not +reset its counts. Renaming a rule, flipping its action, binding a different +policy or recreating the interface starts a new series. + +galactic-router reconciles on every change and every five minutes. When it +restarts, it adopts each snapshot still bound in the datapath that holds +exactly what the current policy compiles to, for the same token, policy UID +and generation, so counts continue; a snapshot an older galactic-router +compiled differently is republished, starting its rules from zero. A +snapshot on an interface no attachment claims within one resync is unbound, +and its rows are freed. + +A rebind that cannot be installed, such as one that runs out of counter +rows, unbinds the direction rather than keeping the old policy's grants, +and reports `Failed`. A failed update of the same policy keeps its last +snapshot. Status is written only when this node's entry changes. + +## Status + +Each node owns one entry in the binding's `status.nodes`, written with +server-side apply as field manager `galactic-router/`: + +```yaml +status: + nodes: + - node: worker-1 + ingress: + state: Applied + policy: {name: default-deny-all, uid: ..., generation: 1} + observedGeneration: 3 + lastTransitionTime: "2026-10-15T12:00:00Z" + egress: + state: Applied + policy: {name: web, uid: ..., generation: 7} + observedGeneration: 3 + lastTransitionTime: "2026-10-15T12:00:00Z" +``` + +`Pending` means the interface's datapath state is not ready yet, or a +reference resolves to nothing. `Failed` carries the reason in `message`. The +entry disappears when the attachment leaves the node. + +## Metrics + +galactic-router serves the policy counters on its metrics port (`9179`), +which the `galactic-router` PodMonitor already scrapes, as +`galactic_cni_traffic_policy_*`: the prefix names the datapath that counts, +not the process that serves them. [docs/metrics.md](../metrics.md#traffic-policy--galactic_cni_traffic_policy_) +lists every series, its labels and example queries. + +Every installed rule has a series from the moment it is published, at zero, +including the built-in default's deny-all rule, whose `policy_uid` is +`builtin:default-deny-all:v1`; an interface still on the default is visible +by that label. A hit is one evaluation, not one forwarded packet. The +supported budget is 27,200 series per node, 100 interfaces at the 64-rule +limit. After a restart, series return once galactic-router reconciles each +interface and adopts its snapshot, with their counts intact. A counter row +the collector cannot read is left out, never reported as zero, and counted +in `galactic_cni_traffic_policy_counter_read_errors`. + +## Diagnostics + +Run these in the galactic-router pod. + +`galactic-router policy explain` decides one packet against a policy, from a +file or the cluster, without touching the datapath: + +```console +$ galactic-router policy explain --policy tenant/web \ + --path Private --src 10.0.0.1 --dst 10.0.0.2 --proto tcp --dport 80 +policy web +path Private +verdict allow +reason rule +rule http (#0) +compiled entries 1 +``` + +It decides with both the Go evaluator and the compiled entries the datapath +runs, and fails if they disagree. Leave `--path` out to see what an +unclassified packet gets. + +`galactic-router policy show /` prints what this node +holds for one attachment: each direction's binding, status, bound snapshot +and every rule's counters, plus any fallback counts. Rules are named when the +policy in the cluster is still the generation the snapshot was compiled from, +and numbered otherwise. diff --git a/go.mod b/go.mod index c7bf7deb..a61d7f23 100644 --- a/go.mod +++ b/go.mod @@ -8,6 +8,7 @@ require ( github.com/containernetworking/plugins v1.9.1 github.com/coreos/go-iptables v0.8.0 github.com/go-logr/logr v1.4.4 + github.com/google/uuid v1.6.0 github.com/kenshaw/baseconv v0.1.1 github.com/lorenzosaino/go-sysctl v0.3.1 github.com/mdlayher/genetlink v1.4.0 @@ -74,7 +75,6 @@ require ( github.com/go-viper/mapstructure/v2 v2.5.0 // indirect github.com/google/gnostic-models v0.7.1 // indirect github.com/google/go-cmp v0.7.0 // indirect - github.com/google/uuid v1.6.0 // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.30.0 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect github.com/json-iterator/go v1.1.12 // indirect diff --git a/internal/config/router.go b/internal/config/router.go index 56c1cb3d..c7dae8d5 100644 --- a/internal/config/router.go +++ b/internal/config/router.go @@ -56,6 +56,10 @@ const ( EnvRouterGCNamespace = "GALACTIC_ROUTER_GC_NAMESPACE" EnvRouterGCInterval = "GALACTIC_ROUTER_GC_INTERVAL" EnvRouterServiceFrontendEnabled = "GALACTIC_ROUTER_SERVICE_FRONTEND_ENABLED" + // EnvRouterTrafficPolicy runs the node's traffic policy reconciler. Off by + // default, because it watches the galactic.datumapis.com CRDs, which a + // cluster may not have installed. + EnvRouterTrafficPolicy = "GALACTIC_ROUTER_TRAFFIC_POLICY" // EnvRouterBMPStations lists the BMP collectors to stream to, as a // comma-separated list of host:port. Empty, the default, disables BMP. @@ -89,6 +93,9 @@ type RouterConfig struct { GCNamespace string GCInterval time.Duration ServiceFrontendEnabled bool + // TrafficPolicy runs the node's traffic policy reconciler. It needs the + // galactic.datumapis.com CRDs, and a route reflector cannot run it. + TrafficPolicy bool // BMPStations, BMPPolicy, and BMPStatisticsInterval configure BMP export to // collectors. BMPStations holds each collector as host:port; empty @@ -115,6 +122,7 @@ func NewRouterConfig() *RouterConfig { v.SetDefault("gc_namespace", DefaultRouterGCNamespace) v.SetDefault("gc_interval", DefaultRouterGCInterval.String()) v.SetDefault("service_frontend_enabled", false) + v.SetDefault("traffic_policy", false) v.SetDefault("bmp_stations", "") v.SetDefault("bmp_policy", DefaultRouterBMPPolicy) v.SetDefault("bmp_statistics_interval", DefaultRouterBMPStatisticsInterval.String()) @@ -143,6 +151,7 @@ func (c *RouterConfig) BindFlags(flags *pflag.FlagSet) { {"gc-namespace", "gc_namespace"}, {"gc-interval", "gc_interval"}, {"service-frontend-enabled", "service_frontend_enabled"}, + {"traffic-policy", "traffic_policy"}, {"bmp-stations", "bmp_stations"}, {"bmp-policy", "bmp_policy"}, {"bmp-statistics-interval", "bmp_statistics_interval"}, @@ -169,6 +178,7 @@ func (c *RouterConfig) readFields() { c.GCNamespace = c.v.GetString("gc_namespace") c.GCInterval = c.v.GetDuration("gc_interval") c.ServiceFrontendEnabled = c.v.GetBool("service_frontend_enabled") + c.TrafficPolicy = c.v.GetBool("traffic_policy") c.BMPStations = splitCommaList(c.v.GetString("bmp_stations")) c.BMPPolicy = c.v.GetString("bmp_policy") c.BMPStatisticsInterval = c.v.GetDuration("bmp_statistics_interval") @@ -189,6 +199,10 @@ func (c *RouterConfig) Validate() error { if c.GRPCHealthPort < 1 || c.GRPCHealthPort > 65535 { return errors.New("grpc health port must be between 1 and 65535") } + if c.TrafficPolicy && c.Reflector { + return fmt.Errorf("%s needs a node with tenant interfaces, so a route reflector cannot enable it", + EnvRouterTrafficPolicy) + } return c.validateBMP() } diff --git a/internal/config/router_test.go b/internal/config/router_test.go index 05213f58..e321486d 100644 --- a/internal/config/router_test.go +++ b/internal/config/router_test.go @@ -290,3 +290,22 @@ func TestRouterConfigServiceFrontendEnv(t *testing.T) { t.Fatal("explicit service frontend enablement was ignored") } } + +func TestRouterConfigTrafficPolicy(t *testing.T) { + if NewRouterConfig().TrafficPolicy { + t.Fatal("traffic policy is on by default") + } + t.Setenv(EnvRouterNodeName, "node-a") + t.Setenv(EnvRouterTrafficPolicy, testBoolTrue) + cfg := NewRouterConfig() + if !cfg.TrafficPolicy { + t.Fatal("explicit traffic policy enablement was ignored") + } + if err := cfg.Validate(); err != nil { + t.Fatalf("Validate() = %v", err) + } + cfg.Reflector = true + if err := cfg.Validate(); err == nil { + t.Fatal("a route reflector accepted traffic policy") + } +} diff --git a/internal/plumbing/ebpf/prog/usid.c b/internal/plumbing/ebpf/prog/usid.c index c5bb2df1..af35644c 100644 --- a/internal/plumbing/ebpf/prog/usid.c +++ b/internal/plumbing/ebpf/prog/usid.c @@ -1943,12 +1943,17 @@ struct tp_entry { __u64 dst_mask[2]; }; -// struct tp_header is slot 0 of a snapshot, the same size as an entry. +// struct tp_header is slot 0 of a snapshot, the same size as an entry. The +// datapath reads only the token, counter_base and count; generation and +// policy_uid record which policy the snapshot was compiled from, so a +// restarted galactic-router can adopt it and its counter rows. struct tp_header { __u64 policy_token; __u32 counter_base; __u32 count; - __u8 pad[72]; + __u64 generation; + __u8 policy_uid[16]; + __u8 pad[48]; }; union tp_slot { diff --git a/internal/plumbing/ebpf/tpmap/tpmap.go b/internal/plumbing/ebpf/tpmap/tpmap.go index 3c6173f8..51299a2c 100644 --- a/internal/plumbing/ebpf/tpmap/tpmap.go +++ b/internal/plumbing/ebpf/tpmap/tpmap.go @@ -88,18 +88,23 @@ type Maps struct { NATLocators *ebpf.Map RuleCounters *ebpf.Map FallbackCounters *ebpf.Map + // Unattributed and AccountingErrorCounts are the node-level counters. + Unattributed *ebpf.Map + AccountingErrorCounts *ebpf.Map } // FromObjects returns the traffic policy maps of objs. func FromObjects(objs *prog.UsidObjects) *Maps { return &Maps{ - State: objs.TpState, - Interfaces: objs.TpIfaceTable, - InterfaceAddrs: objs.TpIfaceAddrTable, - Bindings: objs.TpBindingTable, - NATLocators: objs.TpNatLocatorTable, - RuleCounters: objs.TpRuleCounters, - FallbackCounters: objs.TpFallbackCounters, + State: objs.TpState, + Interfaces: objs.TpIfaceTable, + InterfaceAddrs: objs.TpIfaceAddrTable, + Bindings: objs.TpBindingTable, + NATLocators: objs.TpNatLocatorTable, + RuleCounters: objs.TpRuleCounters, + FallbackCounters: objs.TpFallbackCounters, + Unattributed: objs.TpUnattributedCounters, + AccountingErrorCounts: objs.TpAccountingErrors, } } @@ -156,6 +161,8 @@ func OpenPinned(pinDir string) (*Maps, error) { {prog.UsidMapTpNatLocatorTable, &m.NATLocators}, {prog.UsidMapTpRuleCounters, &m.RuleCounters}, {prog.UsidMapTpFallbackCounters, &m.FallbackCounters}, + {prog.UsidMapTpUnattributedCounters, &m.Unattributed}, + {prog.UsidMapTpAccountingErrors, &m.AccountingErrorCounts}, } { mp, err := ebpf.LoadPinnedMap(filepath.Join(pinDir, p.name), nil) if err != nil { @@ -170,7 +177,7 @@ func OpenPinned(pinDir string) (*Maps, error) { // Close releases maps opened by OpenPinned. func (m *Maps) Close() { for _, mp := range []*ebpf.Map{m.State, m.Interfaces, m.InterfaceAddrs, m.Bindings, m.NATLocators, - m.RuleCounters, m.FallbackCounters} { + m.RuleCounters, m.FallbackCounters, m.Unattributed, m.AccountingErrorCounts} { if mp != nil { _ = mp.Close() } @@ -288,17 +295,40 @@ func (m *Maps) AddNATLocator(sid netip.Addr) error { return nil } +// Snapshot is one interface direction's compiled policy as the datapath +// holds it. +type Snapshot struct { + Token uint64 + // CounterBase is added to every entry's Rule to find its counter row. + CounterBase uint32 + // Generation and PolicyUID identify the policy the entries were compiled + // from; the datapath ignores them. + Generation int64 + PolicyUID [16]byte + Entries []trafficpolicy.Entry + // Raw is each entry as the datapath holds it, filled by ReadSnapshot + // only, so a reader can check the entries are what a policy compiles to. + Raw []prog.UsidTpEntry +} + // Publish installs entries as ifindex's snapshot in dir, its rules counting // into rows counterBase onward. The rows must already be zeroed (see // ZeroCounters). When Publish returns, no packet can still be evaluating the // snapshot it replaced. func (m *Maps) Publish(ifindex, dir uint32, token uint64, counterBase uint32, entries []trafficpolicy.Entry) error { + return m.PublishSnapshot(ifindex, dir, Snapshot{Token: token, CounterBase: counterBase, Entries: entries}) +} + +// PublishSnapshot is Publish with the policy identity recorded in the header. +func (m *Maps) PublishSnapshot(ifindex, dir uint32, snap Snapshot) error { + entries := snap.Entries if len(entries) > galacticv1alpha1.MaxCompiledEntries { return fmt.Errorf("%d entries, more than the limit of %d", len(entries), galacticv1alpha1.MaxCompiledEntries) } for _, e := range entries { - if uint64(counterBase)+uint64(e.Rule) >= RuleCounterRows { - return fmt.Errorf("rule %d's counter row %d is outside tp_rule_counters", e.Rule, uint64(counterBase)+uint64(e.Rule)) + if uint64(snap.CounterBase)+uint64(e.Rule) >= RuleCounterRows { + return fmt.Errorf("rule %d's counter row %d is outside tp_rule_counters", e.Rule, + uint64(snap.CounterBase)+uint64(e.Rule)) } } inner, err := ebpf.NewMap(&SnapshotSpec) @@ -308,7 +338,10 @@ func (m *Maps) Publish(ifindex, dir uint32, token uint64, counterBase uint32, en defer inner.Close() //nolint:errcheck // the outer map holds its own reference once inserted count := uint32(len(entries)) //nolint:gosec // bounded above - header := prog.UsidTpHeader{PolicyToken: token, CounterBase: counterBase, Count: count} + header := prog.UsidTpHeader{ + PolicyToken: snap.Token, CounterBase: snap.CounterBase, Count: count, + Generation: uint64(snap.Generation), PolicyUid: snap.PolicyUID, //nolint:gosec // a generation is never negative + } if err := inner.Put(uint32(0), header); err != nil { return fmt.Errorf("write snapshot header: %w", err) } @@ -323,6 +356,109 @@ func (m *Maps) Publish(ifindex, dir uint32, token uint64, counterBase uint32, en return nil } +// ReadSnapshot reads ifindex's snapshot in dir, reporting false when none is +// bound. Entries carry only each entry's Rule, which names its counter row, +// and its flags; Raw holds every entry in full. +func (m *Maps) ReadSnapshot(ifindex, dir uint32) (Snapshot, bool, error) { + var inner *ebpf.Map + err := m.Bindings.Lookup(prog.UsidTpBindingKey{Ifindex: ifindex, Direction: dir}, &inner) + if errors.Is(err, ebpf.ErrKeyNotExist) || errors.Is(err, os.ErrNotExist) { + return Snapshot{}, false, nil + } + if err != nil { + return Snapshot{}, false, fmt.Errorf("look up binding of ifindex %d direction %d: %w", ifindex, dir, err) + } + defer inner.Close() //nolint:errcheck // read-only handle + var header prog.UsidTpHeader + if err := inner.Lookup(uint32(0), &header); err != nil { + return Snapshot{}, false, fmt.Errorf("read snapshot header: %w", err) + } + snap := Snapshot{ + Token: header.PolicyToken, CounterBase: header.CounterBase, + Generation: int64(header.Generation), PolicyUID: header.PolicyUid, //nolint:gosec // written from an int64 + } + if header.Count > galacticv1alpha1.MaxCompiledEntries { + return Snapshot{}, false, fmt.Errorf("snapshot of ifindex %d direction %d claims %d entries, more than the limit "+ + "of %d", ifindex, dir, header.Count, galacticv1alpha1.MaxCompiledEntries) + } + for i := uint32(1); i <= header.Count; i++ { + var e prog.UsidTpEntry + if err := inner.Lookup(i, &e); err != nil { + return Snapshot{}, false, fmt.Errorf("read snapshot entry %d: %w", i, err) + } + snap.Entries = append(snap.Entries, trafficpolicy.Entry{Rule: e.Rule, Flags: trafficpolicy.EntryFlag(e.Flags)}) + snap.Raw = append(snap.Raw, e) + } + return snap, true, nil +} + +// BoundKeys lists every bound interface direction. +func (m *Maps) BoundKeys() ([]prog.UsidTpBindingKey, error) { + var ( + key prog.UsidTpBindingKey + inner *ebpf.Map + out []prog.UsidTpBindingKey + ) + it := m.Bindings.Iterate() + for it.Next(&key, &inner) { + out = append(out, key) + _ = inner.Close() + } + if err := it.Err(); err != nil { + return nil, fmt.Errorf("iterate tp_binding_table: %w", err) + } + return out, nil +} + +// RuleCounter is the sum over CPUs of one rule counter row. +func (m *Maps) RuleCounter(row uint32) (packets, bytes uint64, err error) { + return sumPerCPU(m.RuleCounters, row) +} + +// FallbackCounter is the sum over CPUs of one interface lifetime's fallback +// row. +func (m *Maps) FallbackCounter(ifindex uint32, token uint64, dir uint32, reason uint8) ( + packets, bytes uint64, err error, +) { + return sumPerCPU(m.FallbackCounters, FallbackKey(ifindex, token, dir, reason)) +} + +// UnattributedCounter is the sum over CPUs of the node-level fallback count +// for dir and reason. +func (m *Maps) UnattributedCounter(dir uint32, reason uint8) (packets, bytes uint64, err error) { + return sumPerCPU(m.Unattributed, dir*reasonCount+uint32(reason)) +} + +// AccountingErrors is the sum over CPUs of the missing-rule-counter count for +// dir and verdict. +func (m *Maps) AccountingErrors(dir uint32, allow bool) (uint64, error) { + k := dir * 2 + if allow { + k++ + } + var per []uint64 + if err := m.AccountingErrorCounts.Lookup(k, &per); err != nil { + return 0, fmt.Errorf("read tp_accounting_errors[%d]: %w", k, err) + } + var n uint64 + for _, v := range per { + n += v + } + return n, nil +} + +func sumPerCPU(m *ebpf.Map, key any) (packets, bytes uint64, err error) { + var per []prog.UsidTpCounter + if err := m.Lookup(key, &per); err != nil { + return 0, 0, err + } + for _, c := range per { + packets += c.Packets + bytes += c.Bytes + } + return packets, bytes, nil +} + // Invalidate removes ifindex's snapshot in dir, so it denies as // policy_unavailable. It needs no allocation, so it works when publishing a // replacement cannot. diff --git a/internal/trafficpolicy/node/metrics.go b/internal/trafficpolicy/node/metrics.go new file mode 100644 index 00000000..4ed849dd --- /dev/null +++ b/internal/trafficpolicy/node/metrics.go @@ -0,0 +1,150 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package node + +import ( + "slices" + "strings" + + "github.com/prometheus/client_golang/prometheus" + + "go.datum.net/galactic/internal/plumbing/ebpf/tpmap" +) + +// Label names. +const ( + labelAttachment = "attachment" + labelDirection = "direction" + labelAction = "action" + labelReason = "reason" +) + +var ( + interfaceLabelNames = []string{"namespace", "vpc", labelAttachment, "interface", "interface_uid"} + ruleLabels = append(slices.Clone(interfaceLabelNames), labelDirection, "policy", "policy_uid", "rule", + labelAction) + fallbackLabels = append(slices.Clone(interfaceLabelNames), labelDirection, labelReason) + + ruleHitsDesc = prometheus.NewDesc("galactic_cni_traffic_policy_rule_hits_total", + "Packets evaluated with this rule as the winning rule, one per interface and direction evaluation. "+ + "A GSO packet counts once. Not forwarded traffic: an allowed packet dropped later still counts.", + ruleLabels, nil) + ruleBytesDesc = prometheus.NewDesc("galactic_cni_traffic_policy_rule_bytes_total", + "Tenant IP bytes evaluated with this rule as the winning rule, without Ethernet or outer SRv6.", + ruleLabels, nil) + fallbackPacketsDesc = prometheus.NewDesc("galactic_cni_traffic_policy_fallback_packets_total", + "Packets denied without a winning rule, by reason: no_match, policy_unavailable, parse_error or "+ + "classification_error.", + fallbackLabels, nil) + fallbackBytesDesc = prometheus.NewDesc("galactic_cni_traffic_policy_fallback_bytes_total", + "Tenant IP bytes denied without a winning rule, by reason; a packet whose IP boundary could not be "+ + "found adds none.", + fallbackLabels, nil) + unattributedDesc = prometheus.NewDesc("galactic_cni_traffic_policy_unattributed_drops_total", + "Policy denials on this node whose interface or fallback row could not be resolved.", + []string{labelDirection, labelReason}, nil) + accountingErrorsDesc = prometheus.NewDesc("galactic_cni_traffic_policy_accounting_errors_total", + "Verdicts whose rule counter row was missing; the verdict stood, the count was lost.", + []string{labelDirection, labelAction, labelReason}, nil) + readErrorsDesc = prometheus.NewDesc("galactic_cni_traffic_policy_counter_read_errors", + "Counter rows the last scrape could not read and left out, rather than reporting zero.", + nil, nil) +) + +var directionNames = [2]string{"ingress", "egress"} + +// Describe implements prometheus.Collector. +func (r *Reconciler) Describe(ch chan<- *prometheus.Desc) { + for _, d := range []*prometheus.Desc{ruleHitsDesc, ruleBytesDesc, fallbackPacketsDesc, fallbackBytesDesc, + unattributedDesc, accountingErrorsDesc, readErrorsDesc} { + ch <- d + } +} + +// Collect implements prometheus.Collector. It reads every installed rule's +// row, and every known interface's fallback rows, from the per-CPU counters, +// labeled from what this reconciler installed, never from packet fields. Only +// the interface an attachment currently has is exported, so a direction left +// bound on an old ifindex never duplicates its labels. A row it cannot read +// is left out rather than reported as zero, and counted in +// counter_read_errors. +func (r *Reconciler) Collect(ch chan<- prometheus.Metric) { + r.mu.Lock() + defer r.mu.Unlock() + if r.maps == nil { + return + } + readErrors := 0 + for key, a := range r.state { + if cur, ok := r.ifaces[a.attachment]; !ok || cur.ifindex != key.ifindex { + continue + } + for i, rule := range a.rules { + packets, bytes, err := r.maps.RuleCounter(a.rows[i]) + if err != nil { + readErrors++ + continue + } + l := append(a.labels.values(), directionNames[key.dir], a.policy.Name, a.policy.UID, rule.name, + actionName(rule.allow)) + ch <- prometheus.MustNewConstMetric(ruleHitsDesc, prometheus.CounterValue, float64(packets), l...) + ch <- prometheus.MustNewConstMetric(ruleBytesDesc, prometheus.CounterValue, float64(bytes), l...) + } + } + for _, i := range r.ifaces { + for dir, dirName := range directionNames { + for reason, reasonName := range tpmap.ReasonNames { + packets, bytes, err := r.maps.FallbackCounter(i.ifindex, i.token, uint32(dir), uint8(reason)) //nolint:gosec // tiny + if err != nil { + readErrors++ + continue + } + l := append(i.labels.values(), dirName, reasonName) + ch <- prometheus.MustNewConstMetric(fallbackPacketsDesc, prometheus.CounterValue, float64(packets), l...) + ch <- prometheus.MustNewConstMetric(fallbackBytesDesc, prometheus.CounterValue, float64(bytes), l...) + } + } + } + for dir, dirName := range directionNames { + for reason, reasonName := range tpmap.ReasonNames { + packets, _, err := r.maps.UnattributedCounter(uint32(dir), uint8(reason)) //nolint:gosec // tiny + if err != nil { + readErrors++ + continue + } + ch <- prometheus.MustNewConstMetric(unattributedDesc, prometheus.CounterValue, float64(packets), + dirName, reasonName) + } + for _, allow := range []bool{false, true} { + n, err := r.maps.AccountingErrors(uint32(dir), allow) //nolint:gosec // tiny + if err != nil { + readErrors++ + continue + } + ch <- prometheus.MustNewConstMetric(accountingErrorsDesc, prometheus.CounterValue, float64(n), + dirName, actionName(allow), "missing_rule_counter") + } + } + ch <- prometheus.MustNewConstMetric(readErrorsDesc, prometheus.GaugeValue, float64(readErrors)) +} + +func actionName(allow bool) string { + if allow { + return "allow" + } + return "deny" +} + +// values are the interface labels, in label order, with an empty value +// spelled out so it stays distinguishable in queries. +func (l interfaceLabels) values() []string { + out := []string{l.namespace, l.vpc, l.attachment, l.iface, l.ifaceUID} + for i, v := range out { + if strings.TrimSpace(v) == "" { + out[i] = "unknown" + } + } + return out +} diff --git a/internal/trafficpolicy/node/reconciler.go b/internal/trafficpolicy/node/reconciler.go new file mode 100644 index 00000000..782c9450 --- /dev/null +++ b/internal/trafficpolicy/node/reconciler.go @@ -0,0 +1,798 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +// Package node is galactic-router's traffic policy reconciler: for every +// VPCAttachment on this node, it compiles the policy each direction of its +// interface is bound to and publishes it into the uSID datapath, keeps +// each rule's counter row across updates, reports what it applied in the +// binding's per-node status, and exports the counters. +package node + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "sync" + "time" + + "github.com/cilium/ebpf" + "github.com/google/uuid" + "github.com/vishvananda/netlink" + "k8s.io/apimachinery/pkg/api/equality" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/util/validation/field" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + cloudv1alpha1 "go.datum.net/cloud/api/v1alpha1" + galacticv1alpha1 "go.datum.net/galactic/api/v1alpha1" + "go.datum.net/galactic/internal/plumbing/ebpf/prog" + "go.datum.net/galactic/internal/plumbing/ebpf/tpmap" + "go.datum.net/galactic/internal/trafficpolicy" + "go.datum.net/galactic/internal/trafficpolicy/cell" +) + +// resync bounds how long a missed event can leave an interface stale. +const resync = 5 * time.Minute + +// pendingRetry is how soon an interface whose datapath state is not ready +// yet is looked at again. +const pendingRetry = 5 * time.Second + +// policyRefIndex indexes bindings by every policy name they reference. +const policyRefIndex = "galactic.datumapis.com/node-policy-refs" + +var directions = [2]struct { + dir uint32 + name galacticv1alpha1.Direction + ref func(*galacticv1alpha1.TrafficPolicyBindingSpec) *galacticv1alpha1.LocalPolicyReference +}{ + {tpmap.DirIngress, galacticv1alpha1.DirectionIngress, + func(s *galacticv1alpha1.TrafficPolicyBindingSpec) *galacticv1alpha1.LocalPolicyReference { + return s.IngressPolicyRef + }}, + {tpmap.DirEgress, galacticv1alpha1.DirectionEgress, + func(s *galacticv1alpha1.TrafficPolicyBindingSpec) *galacticv1alpha1.LocalPolicyReference { + return s.EgressPolicyRef + }}, +} + +// ruleKey identifies a rule's counter interval within one interface +// lifetime, direction and policy: renaming the rule or changing its action +// starts a new one. +type ruleKey struct { + name string + allow bool +} + +// applied is what is installed for one interface direction. +type applied struct { + attachment types.NamespacedName + labels interfaceLabels + token uint64 + policy galacticv1alpha1.PolicyIdentity + rules []ruleKey + rows []uint32 // rows[i] is rules[i]'s counter row +} + +type interfaceLabels struct { + namespace, vpc, attachment, iface, ifaceUID string +} + +// iface is an attachment's interface on this node, as last reconciled. +type iface struct { + ifindex uint32 + token uint64 + labels interfaceLabels +} + +type bindingKey struct { + ifindex uint32 + dir uint32 +} + +// Reconciler is galactic-router's traffic policy reconciler. Each request +// names a VPCAttachment. +type Reconciler struct { + Client client.Client + NodeName string + PinDir string + + mu sync.Mutex + maps *tpmap.Maps + alloc *tpmap.CounterAllocator + state map[bindingKey]*applied + adopted map[bindingKey]tpmap.Snapshot // bound before a restart, not yet reconciled + ifaces map[types.NamespacedName]iface + openedAt time.Time + + // openMaps opens the datapath's maps, and counterRows sizes the row + // allocator, zero meaning the whole of tp_rule_counters; tests set both. + openMaps func(pinDir string) (*tpmap.Maps, error) + counterRows uint32 +} + +// SetupWithManager registers the reconciler and its collector. +func (r *Reconciler) SetupWithManager(mgr ctrl.Manager) error { + if err := mgr.GetFieldIndexer().IndexField(context.Background(), &galacticv1alpha1.TrafficPolicyBinding{}, + policyRefIndex, func(o client.Object) []string { + b := o.(*galacticv1alpha1.TrafficPolicyBinding) + var refs []string + for _, d := range directions { + if ref := d.ref(&b.Spec); ref != nil { + refs = append(refs, ref.Name) + } + } + return refs + }); err != nil { + return fmt.Errorf("index bindings by policy: %w", err) + } + sameName := handler.EnqueueRequestsFromMapFunc(func(_ context.Context, o client.Object) []reconcile.Request { + return []reconcile.Request{{NamespacedName: client.ObjectKeyFromObject(o)}} + }) + byPolicy := handler.EnqueueRequestsFromMapFunc(func(ctx context.Context, o client.Object) []reconcile.Request { + var bindings galacticv1alpha1.TrafficPolicyBindingList + if err := mgr.GetClient().List(ctx, &bindings, client.InNamespace(o.GetNamespace()), + client.MatchingFields{policyRefIndex: o.GetName()}); err != nil { + ctrl.LoggerFrom(ctx).Error(err, "list bindings referencing policy", "policy", o.GetName()) + return nil + } + out := make([]reconcile.Request, 0, len(bindings.Items)) + for i := range bindings.Items { + out = append(out, reconcile.Request{NamespacedName: client.ObjectKeyFromObject(&bindings.Items[i])}) + } + return out + }) + return ctrl.NewControllerManagedBy(mgr). + Named("traffic-policy-node"). + For(&cloudv1alpha1.VPCAttachment{}). + Watches(&galacticv1alpha1.TrafficPolicyBinding{}, sameName). + Watches(&galacticv1alpha1.TrafficPolicy{}, byPolicy). + Complete(r) +} + +// Reconcile implements reconcile.Reconciler. It changes the datapath under +// the lock and writes status after releasing it, so a scrape never waits on +// the API server. +func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { + update, res, err := r.reconcile(ctx, req) + if update != nil { + if serr := r.writeStatus(ctx, req.NamespacedName, update.results); serr != nil { + err = errors.Join(err, serr) + } + } + return res, err +} + +// statusUpdate is this node's entry for a binding: both directions, or, when +// results is nil, no entry. +type statusUpdate struct { + results *[2]galacticv1alpha1.AppliedPolicy +} + +var removeStatus = &statusUpdate{} + +func pendingStatus(msg string) *statusUpdate { + p := galacticv1alpha1.AppliedPolicy{State: galacticv1alpha1.StatePending, Message: msg} + return &statusUpdate{results: &[2]galacticv1alpha1.AppliedPolicy{p, p}} +} + +func (r *Reconciler) reconcile(ctx context.Context, req ctrl.Request) (*statusUpdate, ctrl.Result, error) { + log := ctrl.LoggerFrom(ctx) + r.mu.Lock() + defer r.mu.Unlock() + if err := r.open(); err != nil { + // The datapath pins its maps when galactic-cni loads it, which may be + // after this process starts. + log.Info("traffic policy maps are not available yet", "error", err.Error()) + return nil, ctrl.Result{RequeueAfter: pendingRetry}, nil + } + r.sweepAdopted(ctx) + + var att cloudv1alpha1.VPCAttachment + if err := r.Client.Get(ctx, req.NamespacedName, &att); err != nil { + if !apierrors.IsNotFound(err) { + return nil, ctrl.Result{}, err + } + r.forget(ctx, req.NamespacedName) + return removeStatus, ctrl.Result{}, nil + } + if att.Status.Node != r.NodeName || att.Status.HostInterface == "" { + r.forget(ctx, req.NamespacedName) + return removeStatus, ctrl.Result{}, nil + } + link, err := netlink.LinkByName(att.Status.HostInterface) + if err != nil { + var notFound netlink.LinkNotFoundError + if !errors.As(err, ¬Found) { + // Keep what is installed through a transient failure. + return nil, ctrl.Result{}, fmt.Errorf("find host interface %q: %w", att.Status.HostInterface, err) + } + r.forget(ctx, req.NamespacedName) + return pendingStatus("the interface is not on this node yet"), ctrl.Result{RequeueAfter: pendingRetry}, nil + } + ifindex := uint32(link.Attrs().Index) //nolint:gosec // a kernel ifindex + var row prog.UsidTpIfaceValue + if err := r.maps.Interfaces.Lookup(ifindex, &row); err != nil { + if !errors.Is(err, ebpf.ErrKeyNotExist) { + return nil, ctrl.Result{}, fmt.Errorf("read tp_iface_table[%d]: %w", ifindex, err) + } + // CNI ADD has not written the row yet; the interface denies as + // policy_unavailable meanwhile. + return pendingStatus("the interface's datapath state is not ready"), ctrl.Result{RequeueAfter: pendingRetry}, nil + } + if prev, ok := r.ifaces[req.NamespacedName]; ok && prev.ifindex != ifindex { + r.drop(ctx, prev.ifindex, req.NamespacedName) + } + + var binding galacticv1alpha1.TrafficPolicyBinding + bindingFound := true + if err := r.Client.Get(ctx, req.NamespacedName, &binding); err != nil { + if !apierrors.IsNotFound(err) { + return nil, ctrl.Result{}, err + } + bindingFound = false + } + labels := interfaceLabels{ + namespace: att.Namespace, vpc: att.Spec.VPC.Name, attachment: att.Name, + iface: att.Spec.Interface.Name, ifaceUID: string(att.UID), + } + r.ifaces[req.NamespacedName] = iface{ifindex: ifindex, token: row.PolicyToken, labels: labels} + + var results [2]galacticv1alpha1.AppliedPolicy + var errs []error + for i, d := range directions { + var ref *galacticv1alpha1.LocalPolicyReference + if bindingFound { + ref = d.ref(&binding.Spec) + } + res, err := r.reconcileDirection(ctx, bindingKey{ifindex, d.dir}, req.NamespacedName, labels, + row.PolicyToken, att.Namespace, ref) + res.ObservedGeneration = binding.Generation + results[i] = res + if err != nil { + errs = append(errs, fmt.Errorf("%s: %w", d.name, err)) + } + } + var update *statusUpdate + if bindingFound { + update = &statusUpdate{results: &results} + } + return update, ctrl.Result{RequeueAfter: resync}, errors.Join(errs...) +} + +// reconcileDirection resolves, compiles and publishes one direction, and +// returns what the node now applies there. +func (r *Reconciler) reconcileDirection(ctx context.Context, key bindingKey, att types.NamespacedName, + labels interfaceLabels, token uint64, ns string, ref *galacticv1alpha1.LocalPolicyReference, +) (galacticv1alpha1.AppliedPolicy, error) { + identity, model, unresolved, invalid, err := r.resolve(ctx, ns, ref) + if err != nil { + // An API outage keeps what is installed. + return failed(r.state[key], err.Error()), err + } + prev := r.state[key] + sameLifetime := prev != nil && prev.policy.UID == identity.UID && prev.token == token + if unresolved != "" { + // A missing or reserved-conflict reference never keeps old grants. + if err := r.invalidate(key); err != nil { + return failed(prev, err.Error()), err + } + return galacticv1alpha1.AppliedPolicy{State: galacticv1alpha1.StatePending, Message: unresolved}, nil + } + if invalid != "" { + // The same policy keeps its last good snapshot; any other identity, + // including one adopted across a restart, is invalidated. + if sameLifetime { + return failed(prev, invalid), nil + } + if err := r.invalidate(key); err != nil { + return failed(nil, err.Error()), err + } + return failed(nil, invalid), nil + } + if prev == nil && r.adopt(key, att, labels, token, identity, model) { + return appliedState(r.state[key]), nil // bound before a restart, and still current + } + if prev != nil && prev.policy == identity && prev.token == token { + return appliedState(prev), nil // nothing changed + } + if err := r.publish(key, att, labels, token, identity, model); err != nil { + if sameLifetime { + return failed(prev, err.Error()), err // an update of the same policy keeps its last snapshot + } + // A rebind that cannot be installed must not leave the old grants. + if ierr := r.invalidate(key); ierr != nil { + return failed(prev, err.Error()), errors.Join(err, ierr) + } + return failed(nil, err.Error()), err + } + return appliedState(r.state[key]), nil +} + +// resolve reads what a reference names. unresolved explains a reference that +// resolves to nothing; invalid explains a policy that does not compile. +func (r *Reconciler) resolve(ctx context.Context, ns string, ref *galacticv1alpha1.LocalPolicyReference) ( + identity galacticv1alpha1.PolicyIdentity, model trafficpolicy.Policy, unresolved, invalid string, err error, +) { + if ref == nil { + builtin := galacticv1alpha1.PolicyIdentity{ + Name: galacticv1alpha1.DefaultPolicyName, UID: galacticv1alpha1.DefaultPolicyUID, + } + return builtin, trafficpolicy.Default(), "", "", nil + } + var p galacticv1alpha1.TrafficPolicy + if err := r.Client.Get(ctx, client.ObjectKey{Namespace: ns, Name: ref.Name}, &p); err != nil { + if apierrors.IsNotFound(err) { + return identity, model, "TrafficPolicy " + ref.Name + " does not exist", "", nil + } + return identity, model, "", "", err + } + if cell.IsReservedConflict(&p) { + return identity, model, "TrafficPolicy " + ref.Name + " holds the reserved default name without being " + + "the controller's default policy", "", nil + } + identity = galacticv1alpha1.PolicyIdentity{Name: p.Name, UID: string(p.UID), Generation: p.Generation} + model, errs := trafficpolicy.FromAPI(p.Spec, field.NewPath("spec")) + if len(errs) > 0 { + return identity, model, "", "TrafficPolicy " + ref.Name + " is invalid: " + errs.ToAggregate().Error(), nil + } + return identity, model, "", "", nil +} + +// publish installs model for key, reusing the counter row of every rule whose +// name and action are unchanged within the same policy and interface +// lifetime, and freeing the rows the replaced snapshot alone used once the +// kernel has stopped every program that could still count into them. On +// failure nothing it allocated stays allocated and the bound snapshot is +// unchanged. +func (r *Reconciler) publish(key bindingKey, att types.NamespacedName, labels interfaceLabels, token uint64, + identity galacticv1alpha1.PolicyIdentity, model trafficpolicy.Policy, +) error { + prev := r.state[key] + reuse := map[ruleKey]uint32{} + if prev != nil && prev.policy.UID == identity.UID && prev.token == token { + for i, k := range prev.rules { + reuse[k] = prev.rows[i] + } + } + entries, err := trafficpolicy.Compile(model) + if err != nil { + return err + } + next := &applied{attachment: att, labels: labels, token: token, policy: identity} + var fresh []uint32 + rollback := func() { + for _, row := range fresh { + r.alloc.Free(row) + } + } + for _, rule := range model.Rules { + k := ruleKey{rule.Name, rule.Allow} + row, ok := reuse[k] + if !ok { + var err error + if row, err = r.alloc.Allocate(1); err != nil { + rollback() + return err + } + fresh = append(fresh, row) + if err := r.maps.ZeroCounters(row, 1); err != nil { + rollback() + return err + } + } + delete(reuse, k) + next.rules = append(next.rules, k) + next.rows = append(next.rows, row) + } + for i := range entries { + entries[i].Rule = uint16(next.rows[entries[i].Rule]) //nolint:gosec // rows < RuleCounterRows + } + snap := tpmap.Snapshot{ + Token: token, Generation: identity.Generation, PolicyUID: PolicyUIDBytes(identity.UID), Entries: entries, + } + if err := r.maps.PublishSnapshot(key.ifindex, key.dir, snap); err != nil { + rollback() + return err + } + // Publish returned, so no program can still be counting into a row only + // the replaced snapshot named. + if prev != nil { + if prev.policy.UID == identity.UID && prev.token == token { + for _, row := range reuse { + r.alloc.Free(row) + } + } else { + for _, row := range prev.rows { + r.alloc.Free(row) + } + } + } + r.state[key] = next + r.freeAdopted(key) + return nil +} + +// invalidate unbinds key, which denies as policy_unavailable, then frees the +// rows its snapshot used. +func (r *Reconciler) invalidate(key bindingKey) error { + if err := r.maps.Invalidate(key.ifindex, key.dir); err != nil { + return err + } + if prev := r.state[key]; prev != nil { + for _, row := range prev.rows { + r.alloc.Free(row) + } + } + delete(r.state, key) + r.freeAdopted(key) + return nil +} + +// forget drops what this node installed for an attachment that left it. +// CNI DEL removes the interface's datapath state. +func (r *Reconciler) forget(ctx context.Context, att types.NamespacedName) { + if i, ok := r.ifaces[att]; ok { + r.drop(ctx, i.ifindex, att) + delete(r.ifaces, att) + } +} + +// drop unbinds att's directions on ifindex so their rows can be freed. A +// direction another attachment now holds, after the ifindex was reused, is +// left alone. A direction that cannot be unbound keeps its rows, and Collect +// stops exporting it because att no longer owns the ifindex. +func (r *Reconciler) drop(ctx context.Context, ifindex uint32, att types.NamespacedName) { + for _, d := range directions { + key := bindingKey{ifindex, d.dir} + if a := r.state[key]; a != nil && a.attachment != att { + continue + } + if err := r.invalidate(key); err != nil { + ctrl.LoggerFrom(ctx).Error(err, "unbind traffic policy", "ifindex", ifindex, "direction", d.name) + } + } +} + +// open loads the pinned maps and reserves the rows of every snapshot already +// bound, adopting them, on the first reconcile; it retries until the +// datapath has pinned its maps. Nothing is kept unless every bound snapshot +// was read, since a row left unreserved could be handed to another rule +// while a bound snapshot still counts into it. +func (r *Reconciler) open() error { + if r.maps != nil { + return nil + } + openMaps := r.openMaps + if openMaps == nil { + openMaps = tpmap.OpenPinned + } + maps, err := openMaps(r.PinDir) + if err != nil { + return fmt.Errorf("open traffic policy maps: %w", err) + } + rows := r.counterRows + if rows == 0 { + rows = tpmap.RuleCounterRows + } + alloc := tpmap.NewCounterAllocator(rows) + adopted := map[bindingKey]tpmap.Snapshot{} + if err := readBound(maps, alloc, adopted); err != nil { + if r.openMaps == nil { + maps.Close() + } + return err + } + r.maps, r.alloc, r.adopted = maps, alloc, adopted + r.state = map[bindingKey]*applied{} + r.ifaces = map[types.NamespacedName]iface{} + r.openedAt = time.Now() + return nil +} + +func readBound(maps *tpmap.Maps, alloc *tpmap.CounterAllocator, adopted map[bindingKey]tpmap.Snapshot) error { + keys, err := maps.BoundKeys() + if err != nil { + return err + } + for _, k := range keys { + snap, ok, err := maps.ReadSnapshot(k.Ifindex, k.Direction) + if err != nil { + return err + } + if !ok { + continue // unbound since it was listed + } + for _, row := range snapshotRows(snap) { + if err := alloc.Reserve(row, 1); err != nil { + return fmt.Errorf("reserve row %d of ifindex %d direction %d: %w", row, k.Ifindex, k.Direction, err) + } + } + adopted[bindingKey{k.Ifindex, k.Direction}] = snap + } + return nil +} + +// freeAdopted releases the rows of a snapshot bound before a restart, once +// it has been replaced or unbound, except those the current state reuses. +func (r *Reconciler) freeAdopted(key bindingKey) { + snap, ok := r.adopted[key] + if !ok { + return + } + keep := map[uint32]bool{} + if cur := r.state[key]; cur != nil { + for _, row := range cur.rows { + keep[row] = true + } + } + for _, row := range snapshotRows(snap) { + if !keep[row] { + r.alloc.Free(row) + } + } + delete(r.adopted, key) +} + +// sweepAdopted unbinds the snapshots bound before a restart that no +// attachment has claimed within a full resync, such as one whose +// VPCAttachment was deleted while galactic-router was down, so their rows +// are not held forever. Their interfaces deny as policy_unavailable. An +// interface this node has reconciled is never swept: a direction whose +// reconcile keeps failing keeps what it had. +func (r *Reconciler) sweepAdopted(ctx context.Context) { + if len(r.adopted) == 0 || time.Since(r.openedAt) < resync+pendingRetry { + return + } + known := map[uint32]bool{} + for _, i := range r.ifaces { + known[i.ifindex] = true + } + for key := range r.adopted { + if known[key.ifindex] { + continue + } + if err := r.invalidate(key); err != nil { + ctrl.LoggerFrom(ctx).Error(err, "unbind unclaimed traffic policy", "ifindex", key.ifindex) + } + } +} + +// snapshotRows is every counter row a snapshot's entries name. +func snapshotRows(s tpmap.Snapshot) []uint32 { + var rows []uint32 + seen := map[uint32]bool{} + for _, e := range s.Entries { + row := s.CounterBase + uint32(e.Rule) + if !seen[row] { + seen[row] = true + rows = append(rows, row) + } + } + return rows +} + +// NoRow marks a rule that compiles to no entries, and so names no row in a +// snapshot. +const NoRow = ^uint32(0) + +// RuleRows maps each of a policy's rules to the counter row a bound snapshot +// gives it, from entries, what the policy compiles to with rule indexes. It +// reports false unless the snapshot holds exactly those entries, so a +// snapshot an older compiler produced, or one of another policy, never +// matches. +func RuleRows(entries []trafficpolicy.Entry, snap tpmap.Snapshot, rules int) ([]uint32, bool) { + if len(entries) != len(snap.Raw) { + return nil, false + } + rows := make([]uint32, rules) + for i := range rows { + rows[i] = NoRow + } + owner := map[uint32]uint16{} + for i, e := range entries { + got := snap.Raw[i] + want := tpmap.EncodeEntry(e) + row := snap.CounterBase + uint32(got.Rule) + got.Rule, want.Rule = 0, 0 + if got != want || int(e.Rule) >= rules { + return nil, false + } + if prev, ok := owner[row]; ok && prev != e.Rule { + return nil, false // two rules share a row + } + if rows[e.Rule] != NoRow && rows[e.Rule] != row { + return nil, false // one rule names two rows + } + owner[row] = e.Rule + rows[e.Rule] = row + } + return rows, true +} + +// adopt takes over a snapshot bound before a restart when it holds exactly +// what the policy now desired compiles to, so its counters continue. A rule +// that compiles to no entries counts nothing and gets a fresh row. +func (r *Reconciler) adopt(key bindingKey, att types.NamespacedName, labels interfaceLabels, token uint64, + identity galacticv1alpha1.PolicyIdentity, model trafficpolicy.Policy, +) bool { + snap, ok := r.adopted[key] + if !ok || snap.Token != token || snap.Generation != identity.Generation || + snap.PolicyUID != PolicyUIDBytes(identity.UID) { + return false + } + entries, err := trafficpolicy.Compile(model) + if err != nil { + return false + } + rows, ok := RuleRows(entries, snap, len(model.Rules)) + if !ok { + return false + } + var fresh []uint32 + for i, row := range rows { + if row != NoRow { + continue + } + row, err := r.alloc.Allocate(1) + if err == nil { + err = r.maps.ZeroCounters(row, 1) + fresh = append(fresh, row) + } + if err != nil { + for _, f := range fresh { + r.alloc.Free(f) + } + return false + } + rows[i] = row + } + a := &applied{attachment: att, labels: labels, token: token, policy: identity, rows: rows} + for _, rule := range model.Rules { + a.rules = append(a.rules, ruleKey{rule.Name, rule.Allow}) + } + r.state[key] = a + delete(r.adopted, key) + return true +} + +// PolicyUIDBytes is the 16-byte policy identity a snapshot header records: a +// Kubernetes UID's own bytes, or a digest of the synthetic built-in identity. +func PolicyUIDBytes(uid string) [16]byte { + if u, err := uuid.Parse(uid); err == nil { + return u + } + sum := sha256.Sum256([]byte(uid)) + var b [16]byte + copy(b[:], sum[:]) + return b +} + +func appliedState(a *applied) galacticv1alpha1.AppliedPolicy { + p := a.policy + return galacticv1alpha1.AppliedPolicy{Policy: &p, State: galacticv1alpha1.StateApplied} +} + +func failed(last *applied, msg string) galacticv1alpha1.AppliedPolicy { + out := galacticv1alpha1.AppliedPolicy{State: galacticv1alpha1.StateFailed, Message: truncate(msg)} + if last != nil { + p := last.policy + out.Policy = &p + } + return out +} + +func truncate(s string) string { + if len(s) > 1024 { + return s[:1021] + "..." + } + return s +} + +// fieldOwner is this node's server-side apply field manager, within the +// 128-byte limit for any node name. +func fieldOwner(node string) string { + owner := "galactic-router/" + node + if len(owner) <= 128 { + return owner + } + sum := sha256.Sum256([]byte(node)) + return "galactic-router/" + node[:64] + "-" + hex.EncodeToString(sum[:8]) +} + +// writeStatus applies this node's entry in the binding's status.nodes, or, +// when results is nil, removes it. It writes only when the entry would +// change, so resyncs, pending retries and attachments on other nodes cost no +// API writes. +func (r *Reconciler) writeStatus(ctx context.Context, binding types.NamespacedName, + results *[2]galacticv1alpha1.AppliedPolicy, +) error { + var cur galacticv1alpha1.TrafficPolicyBinding + if err := r.Client.Get(ctx, binding, &cur); err != nil { + if apierrors.IsNotFound(err) { + return nil // no binding: both directions use the built-in default + } + return err + } + var existing *galacticv1alpha1.NodeBindingStatus + for i := range cur.Status.Nodes { + if cur.Status.Nodes[i].Node == r.NodeName { + existing = &cur.Status.Nodes[i] + } + } + nodes := []galacticv1alpha1.NodeBindingStatus{} + if results == nil { + if existing == nil { + return nil + } + } else { + in, eg := results[0], results[1] + now := metav1.Now() + for _, a := range []*galacticv1alpha1.AppliedPolicy{&in, &eg} { + if a.ObservedGeneration == 0 { + a.ObservedGeneration = cur.Generation + } + } + if existing != nil { + keepTransitionTimes(&[2]*galacticv1alpha1.AppliedPolicy{&in, &eg}, *existing) + } + for _, a := range []*galacticv1alpha1.AppliedPolicy{&in, &eg} { + if a.LastTransitionTime == nil { + a.LastTransitionTime = &now + } + } + entry := galacticv1alpha1.NodeBindingStatus{Node: r.NodeName, Ingress: &in, Egress: &eg} + if existing != nil && equality.Semantic.DeepEqual(*existing, entry) { + return nil + } + nodes = append(nodes, entry) + } + body, err := json.Marshal(map[string]any{ + "apiVersion": galacticv1alpha1.GroupVersion.String(), + "kind": "TrafficPolicyBinding", + "metadata": metav1.ObjectMeta{Name: binding.Name, Namespace: binding.Namespace}, + "status": map[string]any{"nodes": nodes}, + }) + if err != nil { + return err + } + obj := &galacticv1alpha1.TrafficPolicyBinding{} + obj.Name, obj.Namespace = binding.Name, binding.Namespace + err = r.Client.Status().Patch(ctx, obj, client.RawPatch(types.ApplyPatchType, body), + client.FieldOwner(fieldOwner(r.NodeName)), client.ForceOwnership) + if apierrors.IsNotFound(err) { + return nil + } + if err != nil { + return fmt.Errorf("apply node status: %w", err) + } + return nil +} + +// keepTransitionTimes carries over lastTransitionTime for each direction +// whose state and policy are unchanged from this node's existing entry. +func keepTransitionTimes(results *[2]*galacticv1alpha1.AppliedPolicy, existing galacticv1alpha1.NodeBindingStatus) { + for i, old := range []*galacticv1alpha1.AppliedPolicy{existing.Ingress, existing.Egress} { + cur := results[i] + if old == nil || old.State != cur.State || !samePolicy(old.Policy, cur.Policy) { + continue + } + cur.LastTransitionTime = old.LastTransitionTime + } +} + +func samePolicy(a, b *galacticv1alpha1.PolicyIdentity) bool { + if a == nil || b == nil { + return a == b + } + return *a == *b +} diff --git a/internal/trafficpolicy/node/reconciler_test.go b/internal/trafficpolicy/node/reconciler_test.go new file mode 100644 index 00000000..f3ed01e2 --- /dev/null +++ b/internal/trafficpolicy/node/reconciler_test.go @@ -0,0 +1,667 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package node + +import ( + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "fmt" + "os" + "os/exec" + "path/filepath" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/cilium/ebpf" + "github.com/cilium/ebpf/rlimit" + "github.com/google/uuid" + "github.com/prometheus/client_golang/prometheus" + dto "github.com/prometheus/client_model/go" + "github.com/vishvananda/netlink" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + 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/client" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" + "sigs.k8s.io/controller-runtime/pkg/envtest" + + cloudv1alpha1 "go.datum.net/cloud/api/v1alpha1" + galacticv1alpha1 "go.datum.net/galactic/api/v1alpha1" + "go.datum.net/galactic/internal/plumbing/ebpf/prog" + "go.datum.net/galactic/internal/plumbing/ebpf/tpmap" + "go.datum.net/galactic/internal/trafficpolicy" +) + +const ( + testNode = "node-a" + testNS = "tenant" + testAtt = "att-1" + testVPC = "vpc-1" + + webPolicy = "web" +) + +func TestKeepTransitionTimes(t *testing.T) { + old := metav1.NewTime(metav1.Now().Add(-3600e9)) + id := &galacticv1alpha1.PolicyIdentity{Name: "p", UID: "u", Generation: 2} + existing := galacticv1alpha1.NodeBindingStatus{Node: testNode, + Ingress: &galacticv1alpha1.AppliedPolicy{State: galacticv1alpha1.StateApplied, Policy: id, + LastTransitionTime: &old}, + Egress: &galacticv1alpha1.AppliedPolicy{State: galacticv1alpha1.StateApplied, Policy: id, + LastTransitionTime: &old}, + } + newer := *id + newer.Generation = 3 + in := galacticv1alpha1.AppliedPolicy{State: galacticv1alpha1.StateApplied, Policy: id} + eg := galacticv1alpha1.AppliedPolicy{State: galacticv1alpha1.StateApplied, Policy: &newer} + keepTransitionTimes(&[2]*galacticv1alpha1.AppliedPolicy{&in, &eg}, existing) + if in.LastTransitionTime == nil || !in.LastTransitionTime.Equal(&old) { + t.Errorf("unchanged ingress: lastTransitionTime = %v, want the old one", in.LastTransitionTime) + } + if eg.LastTransitionTime != nil { + t.Errorf("egress moved to a new generation but kept lastTransitionTime %v", eg.LastTransitionTime) + } +} + +func TestPolicyUIDBytes(t *testing.T) { + u := "4b6f0e1c-2a3d-4e5f-8a9b-0c1d2e3f4a5b" + if got := PolicyUIDBytes(u); uuid.UUID(got).String() != u { + t.Errorf("a Kubernetes UID round-trips to %s", uuid.UUID(got)) + } + if PolicyUIDBytes("builtin:default-deny-all:v1") == PolicyUIDBytes("builtin:default-deny-all:v2") { + t.Error("two versions of the built-in identity record the same bytes") + } +} + +func TestFieldOwner(t *testing.T) { + if got := fieldOwner("node-a"); got != "galactic-router/node-a" { + t.Errorf("fieldOwner = %q", got) + } + long := strings.Repeat("n", 253) + a, b := fieldOwner(long), fieldOwner(long[:252]+"m") + if len(a) > 128 || a == b { + t.Errorf("long names: %q (%d bytes) and %q", a, len(a), b) + } +} + +// TestRuleRows checks a snapshot is matched to a policy only when it holds +// exactly what the policy compiles to, and that a rule compiling to no +// entries is reported as having no row. +func TestRuleRows(t *testing.T) { + // The middle rule is one the API refuses, kept here because the + // compiler gives it no entries: ICMPv6 under IPv4. + v4, icmp6, tcp := trafficpolicy.FamilyIPv4, trafficpolicy.ProtoICMPv6, trafficpolicy.ProtoTCP + model := trafficpolicy.Policy{Rules: []trafficpolicy.Rule{ + {Name: "dns", Allow: true, Match: trafficpolicy.Match{DestPorts: []trafficpolicy.PortRange{{Start: 53, End: 53}}}}, + {Name: "never", Allow: true, Match: trafficpolicy.Match{Family: &v4, Proto: &icmp6}}, + {Name: "http", Allow: true, Match: trafficpolicy.Match{Proto: &tcp}}, + }} + entries, err := trafficpolicy.Compile(model) + if err != nil { + t.Fatal(err) + } + snap := tpmap.Snapshot{CounterBase: 100} + for _, e := range entries { + rowed := e + rowed.Rule = []uint16{7, 0, 3}[e.Rule] + snap.Raw = append(snap.Raw, tpmap.EncodeEntry(rowed)) + } + rows, ok := RuleRows(entries, snap, len(model.Rules)) + if !ok || rows[0] != 107 || rows[1] != NoRow || rows[2] != 103 { + t.Fatalf("RuleRows = %v, %v; want [107 NoRow 103]", rows, ok) + } + snap.Raw[0].DportLo++ + if _, ok := RuleRows(entries, snap, len(model.Rules)); ok { + t.Error("a snapshot with a different entry matched") + } +} + +// TestSweepKeepsReconciledInterfaces: a snapshot bound before a restart on +// an interface this node has reconciled is never swept, even unclaimed, so a +// direction whose reconcile keeps failing keeps what it had. +func TestSweepKeepsReconciledInterfaces(t *testing.T) { + key := bindingKey{7, tpmap.DirEgress} + r := &Reconciler{ + adopted: map[bindingKey]tpmap.Snapshot{key: {}}, + ifaces: map[types.NamespacedName]iface{{Namespace: testNS, Name: testAtt}: {ifindex: 7}}, + openedAt: time.Now().Add(-2 * resync), + } + r.sweepAdopted(context.Background()) // would need maps to unbind anything + if _, ok := r.adopted[key]; !ok { + t.Error("swept a snapshot on a reconciled interface") + } +} + +func requireRoot(t *testing.T) { + t.Helper() + if os.Geteuid() != 0 { + t.Skip("test requires root to load BPF maps and create an interface; re-run via sudo") + } + if err := rlimit.RemoveMemlock(); err != nil { + t.Fatalf("rlimit.RemoveMemlock: %v", err) + } +} + +// fixture is a real API server, the uSID datapath's maps, and a dummy +// interface standing in for a tenant interface. +type fixture struct { + c client.Client + patches atomic.Int32 // status patches through c + maps *tpmap.Maps + link string + ifindex uint32 + token uint64 +} + +func moduleDir(t *testing.T, module string) string { + t.Helper() + out, err := exec.CommandContext(context.Background(), "go", "list", "-m", "-f", "{{.Dir}}", module).Output() + if err != nil { + t.Fatalf("locate module %s: %v", module, err) + } + return strings.TrimSpace(string(out)) +} + +func newFixture(t *testing.T) *fixture { + t.Helper() + requireRoot(t) + if os.Getenv("KUBEBUILDER_ASSETS") == "" { + if os.Getenv("CI") != "" { + t.Fatal("KUBEBUILDER_ASSETS not set in CI") + } + t.Skip("KUBEBUILDER_ASSETS not set") + } + var objs prog.UsidObjects + if err := prog.LoadUsidObjects(&objs, nil); err != nil { + t.Fatalf("load usid objects: %v", err) + } + t.Cleanup(func() { _ = objs.Close() }) + + te := &envtest.Environment{ + CRDDirectoryPaths: []string{ + filepath.Join("..", "..", "..", "config", "crd"), + filepath.Join(moduleDir(t, "go.datum.net/cloud"), "config", "crd"), + }, + ErrorIfCRDPathMissing: true, + } + cfg, err := te.Start() + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = te.Stop() }) + scheme := runtime.NewScheme() + utilruntime.Must(clientgoscheme.AddToScheme(scheme)) + utilruntime.Must(galacticv1alpha1.AddToScheme(scheme)) + utilruntime.Must(cloudv1alpha1.AddToScheme(scheme)) + direct, err := client.NewWithWatch(cfg, client.Options{Scheme: scheme}) + if err != nil { + t.Fatal(err) + } + + suffix := make([]byte, 4) + _, _ = rand.Read(suffix) + f := &fixture{maps: tpmap.FromObjects(&objs), link: "tpt" + hex.EncodeToString(suffix)} + f.c = interceptor.NewClient(direct, interceptor.Funcs{ + SubResourcePatch: func(ctx context.Context, c client.Client, sub string, obj client.Object, patch client.Patch, + opts ...client.SubResourcePatchOption, + ) error { + f.patches.Add(1) + return c.SubResource(sub).Patch(ctx, obj, patch, opts...) + }, + }) + c := f.c + if err := netlink.LinkAdd(&netlink.Dummy{LinkAttrs: netlink.LinkAttrs{Name: f.link}}); err != nil { + t.Fatalf("create %s: %v", f.link, err) + } + t.Cleanup(func() { + if l, err := netlink.LinkByName(f.link); err == nil { + _ = netlink.LinkDel(l) + } + }) + l, err := netlink.LinkByName(f.link) + if err != nil { + t.Fatal(err) + } + f.ifindex = uint32(l.Attrs().Index) //nolint:gosec // a kernel ifindex + if f.token, err = tpmap.NewPolicyToken(); err != nil { + t.Fatal(err) + } + if err := f.maps.PutInterface(f.ifindex, f.token, tpmap.KindTenant); err != nil { + t.Fatal(err) + } + + ctx := context.Background() + if err := c.Create(ctx, &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: testNS}}); err != nil { + t.Fatal(err) + } + att := &cloudv1alpha1.VPCAttachment{ObjectMeta: metav1.ObjectMeta{Name: testAtt, Namespace: testNS}} + att.Spec.VPC.Name = testVPC + att.Spec.Interface.Name = "eth0" + if err := c.Create(ctx, att); err != nil { + t.Fatal(err) + } + att.Status.Node = testNode + att.Status.HostInterface = f.link + if err := c.Status().Update(ctx, att); err != nil { + t.Fatal(err) + } + return f +} + +func (f *fixture) reconciler() *Reconciler { + return &Reconciler{Client: f.c, NodeName: testNode, PinDir: "unused", + openMaps: func(string) (*tpmap.Maps, error) { return f.maps, nil }} +} + +func (f *fixture) reconcile(t *testing.T, r *Reconciler) { + t.Helper() + req := ctrl.Request{NamespacedName: types.NamespacedName{Namespace: testNS, Name: testAtt}} + if _, err := r.Reconcile(context.Background(), req); err != nil { + t.Fatalf("reconcile: %v", err) + } +} + +func (f *fixture) snapshot(t *testing.T, dir uint32) (tpmap.Snapshot, bool) { + t.Helper() + snap, ok, err := f.maps.ReadSnapshot(f.ifindex, dir) + if err != nil { + t.Fatal(err) + } + return snap, ok +} + +func (f *fixture) applyPolicy(t *testing.T, name, rules string) *galacticv1alpha1.TrafficPolicy { + t.Helper() + var spec galacticv1alpha1.TrafficPolicySpec + if err := json.Unmarshal([]byte(`{"rules":`+rules+`}`), &spec); err != nil { + t.Fatal(err) + } + ctx := context.Background() + p := &galacticv1alpha1.TrafficPolicy{ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: testNS}} + if err := f.c.Get(ctx, client.ObjectKeyFromObject(p), p); err == nil { + p.Spec = spec + if err := f.c.Update(ctx, p); err != nil { + t.Fatal(err) + } + return p + } + p.Spec = spec + if err := f.c.Create(ctx, p); err != nil { + t.Fatal(err) + } + return p +} + +func (f *fixture) binding(t *testing.T) *galacticv1alpha1.TrafficPolicyBinding { + t.Helper() + var b galacticv1alpha1.TrafficPolicyBinding + if err := f.c.Get(context.Background(), types.NamespacedName{Namespace: testNS, Name: testAtt}, &b); err != nil { + t.Fatal(err) + } + return &b +} + +func nodeStatus(t *testing.T, b *galacticv1alpha1.TrafficPolicyBinding) galacticv1alpha1.NodeBindingStatus { + t.Helper() + for _, n := range b.Status.Nodes { + if n.Node == testNode { + return n + } + } + t.Fatalf("binding has no status entry for %s: %+v", testNode, b.Status.Nodes) + return galacticv1alpha1.NodeBindingStatus{} +} + +const ( + httpRule = `{"name":"http","action":"Allow",` + + `"match":{"protocol":"TCP","destinationPorts":[{"start":80,"end":80}]}}` + dnsRule = `{"name":"dns","action":"Allow",` + + `"match":{"protocol":"UDP","destinationPorts":[{"start":53,"end":53}]}}` +) + +// TestReconciler runs the node reconciler through a binding's lifecycle +// against a real API server and the real datapath maps. Each step builds on +// the state the previous one left, so they run in order without subtests. +func TestReconciler(t *testing.T) { + f := newFixture(t) + r := f.reconciler() + stepDefault(t, f, r) + httpRow := stepBindPolicy(t, f, r) + stepMetrics(t, r) + stepOtherNode(t, f, r) + p := stepInsertRule(t, f, r, httpRow) + r = stepRestart(t, f, httpRow) + stepRenameRule(t, f, r, httpRow) + r = stepRebindFailure(t, f) + stepMissingPolicy(t, f, r, p) + stepNodeMove(t, f, r) + stepAttachmentDeleted(t, f, r) +} + +// stepDefault: with no binding, both directions get the built-in default. +func stepDefault(t *testing.T, f *fixture, r *Reconciler) { + t.Helper() + f.reconcile(t, r) + for _, dir := range []uint32{tpmap.DirIngress, tpmap.DirEgress} { + snap, ok := f.snapshot(t, dir) + if !ok || snap.Token != f.token || len(snap.Entries) == 0 { + t.Fatalf("direction %d: default snapshot = %+v, bound %v", dir, snap, ok) + } + } +} + +// stepBindPolicy: a binding naming a policy for egress publishes it and +// reports it, and a resync with nothing changed leaves the status alone. It +// returns the http rule's counter row. +func stepBindPolicy(t *testing.T, f *fixture, r *Reconciler) uint32 { + t.Helper() + p := f.applyPolicy(t, webPolicy, "["+httpRule+"]") + b := &galacticv1alpha1.TrafficPolicyBinding{ObjectMeta: metav1.ObjectMeta{Name: testAtt, Namespace: testNS}} + b.Spec.EgressPolicyRef = &galacticv1alpha1.LocalPolicyReference{Name: webPolicy} + if err := f.c.Create(context.Background(), b); err != nil { + t.Fatal(err) + } + f.reconcile(t, r) + snap, ok := f.snapshot(t, tpmap.DirEgress) + if !ok || snap.Generation != p.Generation || uuid.UUID(snap.PolicyUID).String() != string(p.UID) { + t.Fatalf("egress snapshot = %+v, want generation %d uid %s", snap, p.Generation, p.UID) + } + st := nodeStatus(t, f.binding(t)) + if st.Egress == nil || st.Egress.State != galacticv1alpha1.StateApplied || st.Egress.Policy == nil || + st.Egress.Policy.UID != string(p.UID) { + t.Fatalf("egress status = %+v", st.Egress) + } + if st.Ingress == nil || st.Ingress.Policy == nil || st.Ingress.Policy.Name != galacticv1alpha1.DefaultPolicyName { + t.Fatalf("ingress status = %+v, want the built-in default", st.Ingress) + } + + before := f.patches.Load() + f.reconcile(t, r) + if n := f.patches.Load() - before; n != 0 { + t.Errorf("a no-op resync patched the binding's status %d times", n) + } + return r.state[f.egressKey()].rows[0] +} + +// stepMetrics: the collector exports the installed rule and the interface's +// fallback rows with the interface's labels. +func stepMetrics(t *testing.T, r *Reconciler) { + t.Helper() + reg := prometheus.NewPedanticRegistry() + reg.MustRegister(r) + families, err := reg.Gather() + if err != nil { + t.Fatal(err) + } + if !hasSeries(families, "galactic_cni_traffic_policy_rule_hits_total", map[string]string{ + "namespace": testNS, labelAttachment: testAtt, "vpc": testVPC, labelDirection: "egress", "policy": webPolicy, + "rule": "http", labelAction: "allow", + }) { + t.Errorf("no rule hit series for web/http") + } + if !hasSeries(families, "galactic_cni_traffic_policy_fallback_packets_total", map[string]string{ + labelAttachment: testAtt, labelDirection: "ingress", labelReason: "policy_unavailable", + }) { + t.Errorf("no fallback series for the interface") + } +} + +// stepInsertRule: a new rule ahead of http keeps http's counter row. +func stepInsertRule(t *testing.T, f *fixture, r *Reconciler, httpRow uint32) *galacticv1alpha1.TrafficPolicy { + t.Helper() + p := f.applyPolicy(t, webPolicy, "["+dnsRule+","+httpRule+"]") + f.reconcile(t, r) + a := r.state[f.egressKey()] + if len(a.rows) != 2 || a.rows[1] != httpRow || a.rows[0] == httpRow { + t.Fatalf("rows after inserting dns = %v, want http to keep row %d", a.rows, httpRow) + } + if snap, _ := f.snapshot(t, tpmap.DirEgress); snap.Generation != p.Generation { + t.Fatalf("egress generation = %d, want %d", snap.Generation, p.Generation) + } + return p +} + +// stepOtherNode: an attachment on another node, whose binding carries no +// entry for this node, costs no API write. +func stepOtherNode(t *testing.T, f *fixture, r *Reconciler) { + t.Helper() + ctx := context.Background() + other := types.NamespacedName{Namespace: testNS, Name: "att-2"} + att := &cloudv1alpha1.VPCAttachment{ObjectMeta: metav1.ObjectMeta{Name: other.Name, Namespace: testNS}} + att.Spec.VPC.Name = testVPC + att.Spec.Interface.Name = "eth0" + if err := f.c.Create(ctx, att); err != nil { + t.Fatal(err) + } + att.Status.Node = "node-b" + att.Status.HostInterface = "elsewhere" + if err := f.c.Status().Update(ctx, att); err != nil { + t.Fatal(err) + } + b := &galacticv1alpha1.TrafficPolicyBinding{ObjectMeta: metav1.ObjectMeta{Name: other.Name, Namespace: testNS}} + if err := f.c.Create(ctx, b); err != nil { + t.Fatal(err) + } + before := b.ResourceVersion + if _, err := r.Reconcile(ctx, ctrl.Request{NamespacedName: other}); err != nil { + t.Fatal(err) + } + if err := f.c.Get(ctx, other, b); err != nil { + t.Fatal(err) + } + if b.ResourceVersion != before { + t.Errorf("reconciling another node's attachment wrote its binding: %+v", b.Status) + } +} + +// foreignIfindex is an interface no attachment names, with a snapshot left +// bound from before a restart. +const foreignIfindex = 0x7ffffff0 + +// stepRestart: a restarted reconciler adopts the bound snapshot and its +// rows, with their counts, and unbinds a snapshot no attachment claims +// within a resync. +func stepRestart(t *testing.T, f *fixture, httpRow uint32) *Reconciler { + t.Helper() + f.setCounter(t, httpRow, 5, 500) + entries, err := trafficpolicy.Compile(trafficpolicy.Default()) + if err != nil { + t.Fatal(err) + } + if err := f.maps.Publish(foreignIfindex, tpmap.DirEgress, 1, 200, entries); err != nil { + t.Fatal(err) + } + + r := f.reconciler() + f.reconcile(t, r) + if got := r.state[f.egressKey()]; got == nil || len(got.rows) != 2 || got.rows[1] != httpRow { + t.Fatalf("after restart, egress state = %+v, want http on row %d", got, httpRow) + } + if packets, bytes, err := f.maps.RuleCounter(httpRow); err != nil || packets != 5 || bytes != 500 { + t.Errorf("http's counts after restart = %d, %d, %v; want 5, 500", packets, bytes, err) + } + foreign := bindingKey{foreignIfindex, tpmap.DirEgress} + if _, ok := r.adopted[foreign]; !ok || len(r.adopted) != 1 { + t.Fatalf("adopted = %v, want only the unclaimed snapshot", r.adopted) + } + + r.openedAt = r.openedAt.Add(-2 * resync) + f.reconcile(t, r) + if _, ok, _ := f.maps.ReadSnapshot(foreignIfindex, tpmap.DirEgress); ok { + t.Error("the unclaimed snapshot is still bound after a resync") + } + if _, ok := f.snapshot(t, tpmap.DirEgress); !ok { + t.Error("the sweep unbound a claimed snapshot") + } + if err := r.alloc.Reserve(200, 1); err != nil { + t.Errorf("the unclaimed snapshot's row was not freed: %v", err) + } + r.alloc.Free(200) + return r +} + +// stepRenameRule: renaming a rule starts a new series on a fresh, zeroed +// row, and its old row is freed. +func stepRenameRule(t *testing.T, f *fixture, r *Reconciler, httpRow uint32) { + t.Helper() + f.applyPolicy(t, webPolicy, "["+dnsRule+","+strings.Replace(httpRule, `"http"`, `"`+webPolicy+`"`, 1)+"]") + f.reconcile(t, r) + a := r.state[f.egressKey()] + if a.rules[1].name != webPolicy || a.rows[1] == httpRow { + t.Fatalf("after renaming http, rules %v on rows %v; want a new row", a.rules, a.rows) + } + if packets, _, _ := f.maps.RuleCounter(a.rows[1]); packets != 0 { + t.Errorf("the renamed rule starts at %d packets, want 0", packets) + } + if err := r.alloc.Reserve(httpRow, 1); err != nil { + t.Errorf("http's old row %d was not freed: %v", httpRow, err) + } + r.alloc.Free(httpRow) +} + +// stepRebindFailure: a rebind to a policy that cannot be installed, here +// for want of counter rows, unbinds the direction rather than leaving the +// old policy's grants. The binding then returns to the old policy. +func stepRebindFailure(t *testing.T, f *fixture) *Reconciler { + t.Helper() + ctx := context.Background() + rules := make([]string, 0, 40) + for i := range 40 { + rules = append(rules, fmt.Sprintf(`{"name":"r%d","action":"Allow"}`, i)) + } + f.applyPolicy(t, "big", "["+strings.Join(rules, ",")+"]") + r := f.reconciler() + r.counterRows = 32 + f.reconcile(t, r) + + f.setEgressRef(t, "big") + req := ctrl.Request{NamespacedName: types.NamespacedName{Namespace: testNS, Name: testAtt}} + if _, err := r.Reconcile(ctx, req); err == nil { + t.Fatal("a rebind that cannot allocate its rows reconciled cleanly") + } + if _, ok := f.snapshot(t, tpmap.DirEgress); ok { + t.Fatal("egress kept the old policy's snapshot after a failed rebind") + } + if st := nodeStatus(t, f.binding(t)).Egress; st.State != galacticv1alpha1.StateFailed || st.Policy != nil { + t.Fatalf("egress status = %+v, want Failed with nothing installed", st) + } + + f.setEgressRef(t, webPolicy) + f.reconcile(t, r) + if _, ok := f.snapshot(t, tpmap.DirEgress); !ok { + t.Fatal("egress is unbound after returning to web") + } + return r +} + +func (f *fixture) setEgressRef(t *testing.T, name string) { + t.Helper() + b := f.binding(t) + b.Spec.EgressPolicyRef = &galacticv1alpha1.LocalPolicyReference{Name: name} + if err := f.c.Update(context.Background(), b); err != nil { + t.Fatal(err) + } +} + +// setCounter writes a rule counter row's totals onto CPU 0. +func (f *fixture) setCounter(t *testing.T, row uint32, packets, bytes uint64) { + t.Helper() + per := make([]prog.UsidTpCounter, ebpf.MustPossibleCPU()) + per[0] = prog.UsidTpCounter{Packets: packets, Bytes: bytes} + if err := f.maps.RuleCounters.Put(row, per); err != nil { + t.Fatal(err) + } +} + +// stepMissingPolicy: a reference to a missing policy unbinds the direction, +// which then denies as policy_unavailable, and reports why. +func stepMissingPolicy(t *testing.T, f *fixture, r *Reconciler, p *galacticv1alpha1.TrafficPolicy) { + t.Helper() + if err := f.c.Delete(context.Background(), p); err != nil { + t.Fatal(err) + } + f.reconcile(t, r) + if _, ok := f.snapshot(t, tpmap.DirEgress); ok { + t.Fatal("egress is still bound after its policy was deleted") + } + if st := nodeStatus(t, f.binding(t)); st.Egress.State != galacticv1alpha1.StatePending { + t.Fatalf("egress status = %+v, want Pending", st.Egress) + } +} + +// stepNodeMove: an attachment that moves to another node is unbound here and +// this node's status entry is removed. +func stepNodeMove(t *testing.T, f *fixture, r *Reconciler) { + t.Helper() + ctx := context.Background() + var att cloudv1alpha1.VPCAttachment + if err := f.c.Get(ctx, types.NamespacedName{Namespace: testNS, Name: testAtt}, &att); err != nil { + t.Fatal(err) + } + att.Status.Node = "node-b" + if err := f.c.Status().Update(ctx, &att); err != nil { + t.Fatal(err) + } + f.reconcile(t, r) + for _, dir := range []uint32{tpmap.DirIngress, tpmap.DirEgress} { + if _, ok := f.snapshot(t, dir); ok { + t.Errorf("direction %d is still bound after the attachment moved", dir) + } + } + for _, n := range f.binding(t).Status.Nodes { + if n.Node == testNode { + t.Errorf("this node's status entry outlived the move: %+v", n) + } + } + if len(r.state) != 0 { + t.Errorf("state left after the move: %v", r.state) + } +} + +// stepAttachmentDeleted: a deleted attachment is forgotten. +func stepAttachmentDeleted(t *testing.T, f *fixture, r *Reconciler) { + t.Helper() + if err := f.c.Delete(context.Background(), &cloudv1alpha1.VPCAttachment{ + ObjectMeta: metav1.ObjectMeta{Name: testAtt, Namespace: testNS}, + }); err != nil { + t.Fatal(err) + } + f.reconcile(t, r) + if _, ok := r.ifaces[types.NamespacedName{Namespace: testNS, Name: testAtt}]; ok { + t.Error("the deleted attachment is still tracked") + } +} + +func (f *fixture) egressKey() bindingKey { return bindingKey{f.ifindex, tpmap.DirEgress} } + +func hasSeries(families []*dto.MetricFamily, name string, want map[string]string) bool { + for _, mf := range families { + if mf.GetName() != name { + continue + } + metrics: + for _, m := range mf.GetMetric() { + got := map[string]string{} + for _, l := range m.GetLabel() { + got[l.GetName()] = l.GetValue() + } + for k, v := range want { + if got[k] != v { + continue metrics + } + } + return true + } + } + return false +}