From 8555e529a3a084832f8e562f15c0c8ce1ea5cb41 Mon Sep 17 00:00:00 2001 From: Peter Sprygada Date: Thu, 15 Oct 2026 10:53:08 -0400 Subject: [PATCH] feat(router): Apply traffic policy bindings on each node galactic-router gains the node side of interface traffic policy, behind GALACTIC_ROUTER_TRAFFIC_POLICY (off by default, refused on a route reflector). For every VPCAttachment on its node it resolves the binding's ingress and egress references, compiles each policy, and publishes it into the uSID datapath; a direction with no reference gets the built-in default-deny-all. - Counter rows: each rule keeps its row while its name and action stay the same within one policy UID and interface token, so editing a match keeps its counts. Replaced rows are freed only after the outer-map update returns; a failed publish frees what it allocated. - Fail closed: a missing or reserved-conflict reference, and a rebind that cannot be installed, unbind the direction (policy_unavailable). A failed update or an invalid edit of the same policy keeps its last snapshot. - Restart: snapshots record the policy UID and generation beside the token. A restarted router adopts a snapshot only when it holds exactly what the policy compiles to today, so counts continue and an older compiler's output is republished. Startup reads every bound snapshot or nothing, and a snapshot nothing claims within a resync is unbound and its rows freed. - Status: this node's entry in the binding's status.nodes, applied with server-side apply, written only when it changes, after the datapath lock is released. - Metrics: galactic_cni_traffic_policy_* on the router's metrics port, labeled from what the router installed, plus a gauge of counter rows a scrape could not read. The design said galactic-cni would serve them; the router allocates the rows and knows their rule and interface, so it does, under the names the design fixed. docs/metrics.md documents them. - Diagnostics: `galactic-router policy explain` decides a packet with the Go evaluator and the compiled entries and fails if they disagree; `policy show` prints what the node holds for one attachment. RBAC gains read access to policies and bindings and patch on binding status. The root unit job gains envtest assets so the reconciler's envtest-and-BPF test runs in CI. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/ci.yaml | 8 + AGENTS.md | 2 +- cmd/galactic-router/main.go | 15 +- cmd/galactic-router/policy.go | 419 +++++++++ cmd/galactic-router/policy_test.go | 106 +++ cmd/galactic-router/root.go | 28 +- cmd/galactic-router/root_test.go | 1 + config/galactic-router/rbac.yaml | 12 + docs/README.md | 4 + docs/agents/ARCHITECTURE-ROUTER.md | 54 +- .../interface-traffic-policy/README.md | 39 +- docs/metrics.md | 75 +- docs/router/configuration.md | 20 +- docs/traffic-policy/controller.md | 4 +- docs/traffic-policy/node.md | 126 +++ go.mod | 2 +- internal/config/router.go | 14 + internal/config/router_test.go | 19 + internal/plumbing/ebpf/prog/usid.c | 9 +- internal/plumbing/ebpf/tpmap/tpmap.go | 158 +++- internal/trafficpolicy/node/metrics.go | 150 ++++ internal/trafficpolicy/node/reconciler.go | 798 ++++++++++++++++++ .../trafficpolicy/node/reconciler_test.go | 667 +++++++++++++++ 23 files changed, 2668 insertions(+), 62 deletions(-) create mode 100644 cmd/galactic-router/policy.go create mode 100644 cmd/galactic-router/policy_test.go create mode 100644 docs/traffic-policy/node.md create mode 100644 internal/trafficpolicy/node/metrics.go create mode 100644 internal/trafficpolicy/node/reconciler.go create mode 100644 internal/trafficpolicy/node/reconciler_test.go 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 +}