From 9c132876f4a23afc8ff0341b03fcb71712cd64d2 Mon Sep 17 00:00:00 2001 From: Peter Sprygada Date: Fri, 9 Oct 2026 20:31:38 -0400 Subject: [PATCH] feat(cni): Spread egress across the cluster's EgressShards in hashed mode galactic-cni run now owns how tenant VRFs reach the egress shards. GALACTIC_CNI_EGRESS_MODE=ordered, the default, keeps today's behaviour: each VRF uses the first reachable shard in GALACTIC_CNI_EGRESS_SHARD_SIDS. GALACTIC_CNI_EGRESS_MODE=hashed makes every eligible EgressShard in the cluster a member of the shard groups usid_egress picks from, so each shard carries about 1/N of the metro's egress and losing one costs only its sessions. In hashed mode the installer watches EgressShards. A shard is a candidate once its status carries a shardSID with Programmed and Ready true; it is draining while spec.drain is set, and a terminating shard is left out. It joins a translation class's group only if its status can translate that class: an IPv6 address for NAT66; an IPv4 address and the matching nat64Prefix, or translatesWellKnownPrefix for 64:ff9b::/96, for each NAT64 prefix. More than three NAT64 prefixes in hashed mode stops the container at startup. A candidate takes tenants only while its SID resolves to an uplink neighbor, so a dead node's stale status does not matter. A netlink route and neighbor watch rebuilds the groups as soon as a route moves, and the 30s sweep stays as a backstop. Until the watch has listed the cluster once, the groups keep their last members. Going back to ordered mode publishes every enabled group as CLOSING, moves every route back to the static list, and disables the groups only after a sweep that read and wrote every route without error and found none still naming a group, other namespaces' routes included. A failed sweep leaves the groups forwarding and is retried. GALACTIC_CNI_EGRESS_HASH picks source (default, one public address per tenant address) or flow hashing, GALACTIC_CNI_EGRESS_PIN_IDLE sets how long a tenant address stays pinned to its shard after its last packet (default 2h4m, the shard's established-TCP timeout; 0 turns pins off), and GALACTIC_CNI_EGRESS_POOL_MIN_ACTIVE sets the floor the pool alert compares against. An invalid value stops the container at startup. New metrics: galactic_cni_egress_shard_{packets,bytes}_total{shard_sid}, galactic_cni_egress_shard_selections_total{result}, galactic_cni_egress_pool_members{class,state} (with an ineligible state), galactic_cni_egress_pool_min_active, galactic_cni_egress_pool_enabled{class}, and the control loop's own state: galactic_cni_egress_mode, galactic_cni_egress_shard_watch_synced, galactic_cni_egress_group_routes{class}, galactic_cni_egress_sweep_errors_total{stage} and the last successful sweep's timestamp. A node told to run hashed mode that never published a group says so instead of exporting nothing. galactic-cni's ClusterRole gains get, list and watch on egressshards. go.datum.net/network moves to the commit that adds EgressShard spec.drain (datum-cloud/network#32); it moves to the release once that is cut. Co-Authored-By: Claude Opus 5.5 --- config/galactic-cni/daemonset.yaml | 12 +- config/galactic-cni/rbac.yaml | 7 + go.mod | 3 +- go.sum | 113 +-- internal/config/cni.go | 109 +++ internal/config/cni_test.go | 46 ++ internal/installer/egressgroup.go | 682 ++++++++++++++++++ internal/installer/egressgroup_test.go | 535 ++++++++++++++ internal/installer/installer.go | 148 ++-- .../plumbing/ebpf/metrics/egresscontrol.go | 117 +++ internal/plumbing/ebpf/metrics/egressshard.go | 197 +++++ .../plumbing/ebpf/metrics/egressshard_test.go | 201 ++++++ internal/plumbing/ebpf/metrics/metrics.go | 4 +- 13 files changed, 2002 insertions(+), 172 deletions(-) create mode 100644 internal/installer/egressgroup.go create mode 100644 internal/installer/egressgroup_test.go create mode 100644 internal/plumbing/ebpf/metrics/egresscontrol.go create mode 100644 internal/plumbing/ebpf/metrics/egressshard.go create mode 100644 internal/plumbing/ebpf/metrics/egressshard_test.go diff --git a/config/galactic-cni/daemonset.yaml b/config/galactic-cni/daemonset.yaml index f4a26ab4..e31cf86f 100644 --- a/config/galactic-cni/daemonset.yaml +++ b/config/galactic-cni/daemonset.yaml @@ -97,6 +97,16 @@ spec: # placeholder, not a published tag. image: ghcr.io/datum-cloud/galactic-cni:latest command: ["/galactic-cni", "run"] + env: + # How tenant VRFs reach the cluster's egress shards. "ordered" + # sends each VRF to the first reachable shard in + # GALACTIC_CNI_EGRESS_SHARD_SIDS. "hashed" spreads tenants across + # every EgressShard in this namespace, and also reads + # GALACTIC_CNI_EGRESS_HASH, GALACTIC_CNI_EGRESS_PIN_IDLE and + # GALACTIC_CNI_EGRESS_POOL_MIN_ACTIVE. See + # docs/nat/configuration.md#active-active-egress. + - name: GALACTIC_CNI_EGRESS_MODE + value: ordered # Documentation only: hostNetwork: true means these are the # process's actual listen ports on the node, not something # Kubernetes publishes -- containerPort here doesn't change @@ -128,7 +138,7 @@ spec: readOnlyRootFilesystem: true # The kernel charges eBPF map memory to the cgroup of the process # that creates the map, and this container creates the uSID - # datapath's maps (about 59Mi) on a node that has none pinned. + # datapath's maps (about 67Mi) on a node that has none pinned. # TestUsid_MapMemoryFitsGalacticCNILimit fails when the maps plus # the process outgrow this limit -- see config/galactic-cni/README.md. resources: diff --git a/config/galactic-cni/rbac.yaml b/config/galactic-cni/rbac.yaml index 823ebf54..aa2197ae 100644 --- a/config/galactic-cni/rbac.yaml +++ b/config/galactic-cni/rbac.yaml @@ -12,6 +12,13 @@ rules: - bgpadvertisements - bgpvrfinstances verbs: ["get", "list", "create", "update", "patch", "delete"] + # egressshards (read only): in hashed egress mode, galactic-cni run spreads + # tenant egress across every EgressShard in the cluster and watches them for + # membership and drain changes. See internal/installer/egressgroup.go. + - apiGroups: ["network.datumapis.com"] + resources: + - egressshards + verbs: ["get", "list", "watch"] - apiGroups: ["k8s.cni.cncf.io"] resources: - network-attachment-definitions diff --git a/go.mod b/go.mod index 4aa7bb3d..bacbf5a8 100644 --- a/go.mod +++ b/go.mod @@ -22,7 +22,7 @@ require ( github.com/spf13/viper v1.21.0 github.com/vishvananda/netlink v1.3.2-0.20261002201547-1d1c62507e3c go.datum.net/cloud v0.1.2 - go.datum.net/network v0.4.0 + go.datum.net/network v0.4.1-0.20261009192728-36ee528904ff go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.72.0 go.opentelemetry.io/otel v1.47.0 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.47.0 @@ -63,7 +63,6 @@ require ( github.com/go-openapi/swag/cmdutils v0.28.0 // indirect github.com/go-openapi/swag/conv v0.28.0 // indirect github.com/go-openapi/swag/fileutils v0.28.0 // indirect - github.com/go-openapi/swag/jsonname v0.26.0 // indirect github.com/go-openapi/swag/jsonutils v0.28.0 // indirect github.com/go-openapi/swag/loading v0.28.0 // indirect github.com/go-openapi/swag/mangling v0.28.0 // indirect diff --git a/go.sum b/go.sum index edf9cd50..0632385e 100644 --- a/go.sum +++ b/go.sum @@ -42,84 +42,52 @@ github.com/fxamacker/cbor/v2 v2.9.0/go.mod h1:vM4b+DJCtHn+zz7h3FFp/hDAI9WNWCsZj2 github.com/gaissmai/bart v0.26.1 h1:+w4rnLGNlA2GDVn382Tfe3jOsK5vOr5n4KmigJ9lbTo= github.com/gaissmai/bart v0.26.1/go.mod h1:GREWQfTLRWz/c5FTOsIw+KkscuFkIV5t8Rp7Nd1Td5c= github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= -github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= -github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/go-logr/logr v1.4.4 h1:tG4xh9yMsRCAiodLVTxyrkzSZ9+o0L1Kg/+cPVcbP/8= github.com/go-logr/logr v1.4.4/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= github.com/go-logr/zapr v1.3.0 h1:XGdV8XW8zdwFiwOA2Dryh1gj2KRQyOOoNmBy4EplIcQ= github.com/go-logr/zapr v1.3.0/go.mod h1:YKepepNBd1u/oyhd/yQmtjVXmm9uML4IXUgMOwR8/Gg= -github.com/go-openapi/jsonpointer v0.23.1 h1:1HBACs7XIwR2RcmItfdSFlALhGbe6S92p0ry4d1GWg4= -github.com/go-openapi/jsonpointer v0.23.1/go.mod h1:iWRmZTrGn7XwYhtPt/fvdSFj1OfNBngqRT2UG3BxSqY= github.com/go-openapi/jsonpointer v1.0.0 h1:kR9tHqY0CtZaOPVFm622dPVNhrvYpwr4uCxgL3h1H8s= github.com/go-openapi/jsonpointer v1.0.0/go.mod h1:Z3rw7dWu1p9IgitXCFamSlA5lmDiklEB6vkaxcNZW5Y= -github.com/go-openapi/jsonreference v0.21.6 h1:NZ5nGfnaM1n4I43Xjm1e5/M2GjOwQwndQz22uhxwD+Y= -github.com/go-openapi/jsonreference v0.21.6/go.mod h1:xzbgtQ3ZbWxvET3AxdzCJlJt6vkovbf+IfSPJjD0tUY= github.com/go-openapi/jsonreference v1.0.0 h1:jlmTr6torcd1YgDQvSfNmRtKzYDO4FGBkrAdlAVWnpY= github.com/go-openapi/jsonreference v1.0.0/go.mod h1:jtwdyGbJk0Xhe5Y+rwtglQP6Sb1WZST4rT32LWB+sv0= -github.com/go-openapi/swag v0.26.0 h1:GVDXCmfvhfu1BxiHo8/FA+BbKmhecHnG3varjON5/RI= -github.com/go-openapi/swag v0.26.0/go.mod h1:82g3193sZJRbocs7bNCqGfIgq8pkuwVwCfhKIRlEQF0= github.com/go-openapi/swag v0.28.0 h1:xkgbOSKj6DZziNpyqRRAOt3GJGtgjgsd2RoyT30VWuw= github.com/go-openapi/swag v0.28.0/go.mod h1:4qYnT3Cqr1p1VknOdPo70evN4rgQnAg6jwApHyxSGIg= -github.com/go-openapi/swag/cmdutils v0.26.0 h1:iowihOcvq7y4egO8cOq0dmfohz6wfeQ63U1EnuhO2TU= -github.com/go-openapi/swag/cmdutils v0.26.0/go.mod h1:Sm1MVFMkF6guJJ+pQqHnQA3N0j9qALV3NxzDSv6bETM= github.com/go-openapi/swag/cmdutils v0.28.0 h1:7TOeNtkYru1SG8Y34tDh9WBbLsMqGnptuxWiHREPZ4Q= github.com/go-openapi/swag/cmdutils v0.28.0/go.mod h1:Sm1MVFMkF6guJJ+pQqHnQA3N0j9qALV3NxzDSv6bETM= -github.com/go-openapi/swag/conv v0.26.0 h1:5yGGsPYI1ZCva93U0AoKi/iZrNhaJEjr324YVsiD89I= -github.com/go-openapi/swag/conv v0.26.0/go.mod h1:tpAmIL7X58VPnHHiSO4uE3jBeRamGsFsfdDeDtb5ECE= github.com/go-openapi/swag/conv v0.28.0 h1:GtqqbyFe7vR5Y7ehxG9W6/OvrSFdf1OLeTGp40TqxH8= github.com/go-openapi/swag/conv v0.28.0/go.mod h1:mbUE+mzctnhxi864m0Q07SpN8OowD9JhxmxuYvZZD/k= -github.com/go-openapi/swag/fileutils v0.26.0 h1:WJoPRvsA7QRiiWluowkLJa9jaYR7FCuxmDvnCgaRRxU= -github.com/go-openapi/swag/fileutils v0.26.0/go.mod h1:0WDJ7lp67eNjPMO50wAWYlKvhOb6CQ37rzR7wrgI8Tc= github.com/go-openapi/swag/fileutils v0.28.0 h1:Z04XWQD7R8Eq+7GnOrjovBxPPmZzsS4gt2H2GPGIViU= github.com/go-openapi/swag/fileutils v0.28.0/go.mod h1:VvJFZLTZS0AI854gEQz5tk7dBESdLjiNUMSZ/th2ry8= -github.com/go-openapi/swag/jsonname v0.26.0 h1:gV1NFX9M8avo0YSpmWogqfQISigCmpaiNci8cGECU5w= -github.com/go-openapi/swag/jsonname v0.26.0/go.mod h1:urBBR8bZNoDYGr653ynhIx+gTeIz0ARZxHkAPktJK2M= -github.com/go-openapi/swag/jsonutils v0.26.0 h1:FawFML2iAXsPqmERscuMPIHmFsoP1tOqWkxBaKNMsnA= -github.com/go-openapi/swag/jsonutils v0.26.0/go.mod h1:2VmA0CJlyFqgawOaPI9psnjFDqzyivIqLYN34t9p91E= github.com/go-openapi/swag/jsonutils v0.28.0 h1:YIch6FwO7RXzeAnbO8Tu7dWBZeUEH+4nA0HXltVTnv4= github.com/go-openapi/swag/jsonutils v0.28.0/go.mod h1:CYM3WlTUcagR2ZoHdz54di/cbBqt82tuxuXgAjxw+mg= -github.com/go-openapi/swag/jsonutils/fixtures_test v0.26.0 h1:apqeINu/ICHouqiRZbyFvuDge5jCmmLTqGQ9V95EaOM= -github.com/go-openapi/swag/jsonutils/fixtures_test v0.26.0/go.mod h1:AyM6QT8uz5IdKxk5akv0y6u4QvcL9GWERt0Jx/F/R8Y= -github.com/go-openapi/swag/loading v0.26.0 h1:Apg6zaKhCJurpJer0DCxq99qwmhFddBhaMX7kilDcko= -github.com/go-openapi/swag/loading v0.26.0/go.mod h1:dBxQ/6V2uBaAQdevN18VELE6xSpJWZxLX4txe12JwDg= +github.com/go-openapi/swag/jsonutils/fixtures_test v0.28.0 h1:qV+VVUAx5Oro8WjVWpZeql7YReTKhT4smR4zhcOQZr0= +github.com/go-openapi/swag/jsonutils/fixtures_test v0.28.0/go.mod h1:mofwUWx70wvskwESqRJ//k/9kURmCgyJl5m5Ppoh5kY= github.com/go-openapi/swag/loading v0.28.0 h1:td8QZdZC9MIYGGSnSPKShKiK22I2tU5UQvuUhIBPRLU= github.com/go-openapi/swag/loading v0.28.0/go.mod h1:rXB0QiQX5mMveXEA7ouM4KiiM9jVJe4K6BVbwhD1M4k= -github.com/go-openapi/swag/mangling v0.26.0 h1:Du2YC4YLA/Y5m/YKQd7AnY5qq0wRKSFZTTt8ktFaXcQ= -github.com/go-openapi/swag/mangling v0.26.0/go.mod h1:jifS7W9vbg+pw63bT+GI53otluMQL3CeemuyCHKwVx0= github.com/go-openapi/swag/mangling v0.28.0 h1:pH8eyeNO9SLYsTMWJrurnNfKmDa28XrlA+HePVD53VM= github.com/go-openapi/swag/mangling v0.28.0/go.mod h1:jtBE2+V+3pILxOR7Vgce+Cwp6A2PgZbvVqfNntbVs0w= -github.com/go-openapi/swag/netutils v0.26.0 h1:CmZp+ZT7HrmFwrC3GdGsXBq2+42T1bjKBapcqVpIs3c= -github.com/go-openapi/swag/netutils v0.26.0/go.mod h1:5iK+Ok3ZohWWex1C50BFTPexi03UaPwjW4Oj8kgrpwo= github.com/go-openapi/swag/netutils v0.28.0 h1:YXN6TALEi2pzts8/8GNm6T61HTAZsieukGZidap989k= github.com/go-openapi/swag/netutils v0.28.0/go.mod h1:J+WYyFMLtvtCGqa6jLv+YNUmIKI3ZRQRrvfNDMoQoEQ= github.com/go-openapi/swag/pools v0.28.0 h1:HPMZWSAfce3rdVTFcjFiCIBtDg9h4x2QlRrHipwhxeU= github.com/go-openapi/swag/pools v0.28.0/go.mod h1:kVQefhSK5RWuRe7BXsL8htgBPAMpN7HDGpGEknqugeE= -github.com/go-openapi/swag/stringutils v0.26.0 h1:qZQngLxs5s7SLijc3N2ZO+fUq2o8LjuWAASSrJuh+xg= -github.com/go-openapi/swag/stringutils v0.26.0/go.mod h1:sWn5uY+QIIspwPhvgnqJsH8xqFT2ZbYcvbcFanRyhFE= github.com/go-openapi/swag/stringutils v0.28.0 h1:ixsc9iYgDPubHL/8nSkbnryEHpD2VRlBMLKpQyPXcDU= github.com/go-openapi/swag/stringutils v0.28.0/go.mod h1:lzRN95CxXmA03XcDWHLOb6nOMcxCqR5rGY0lOgsfRoM= -github.com/go-openapi/swag/typeutils v0.26.0 h1:2kdEwdiNWy+JJdOvu5MA2IIg2SylWAFuuyQIKYybfq4= -github.com/go-openapi/swag/typeutils v0.26.0/go.mod h1:oovDuIUvTrEHVMqWilQzKzV4YlSKgyZmFh7AlfABNVE= github.com/go-openapi/swag/typeutils v0.28.0 h1:nRBKSBXjDgf01VDPB3fWeD9nQuhCOVeIYAkUx2tbkyY= github.com/go-openapi/swag/typeutils v0.28.0/go.mod h1:Srm0xFNRZ1Y+vCxJclo5qzx8aj+1pAKda/YfFPrG0dQ= -github.com/go-openapi/swag/yamlutils v0.26.0 h1:H7O8l/8NJJQ/oiReEN+oMpnGMyt8G0hl460nRZxhLMQ= -github.com/go-openapi/swag/yamlutils v0.26.0/go.mod h1:1evKEGAtP37Pkwcc7EWMF0hedX0/x3Rkvei2wtG/TbU= github.com/go-openapi/swag/yamlutils v0.28.0 h1:TV3JXH6DS46KUroDtMLAYHGkdWf5VDq3wVWFirmzROY= github.com/go-openapi/swag/yamlutils v0.28.0/go.mod h1:x0q/yndZHEgk9Rx3DyDqzFUmHy55KTvIZldvF2dTJXs= -github.com/go-openapi/testify/enable/yaml/v2 v2.4.2 h1:5zRca5jw7lzVREKCZVNBpysDNBjj74rBh0N2BGQbSR0= -github.com/go-openapi/testify/enable/yaml/v2 v2.4.2/go.mod h1:XVevPw5hUXuV+5AkI1u1PeAm27EQVrhXTTCPAF85LmE= -github.com/go-openapi/testify/v2 v2.5.1 h1:TMdhCaw8fUNraVSf3Omoob1dO/AzBfhtFAPW0an6sBo= -github.com/go-openapi/testify/v2 v2.5.1/go.mod h1:SgsVHtfooshd0tublTtJ50FPKhujf47YRqauXXOUxfw= +github.com/go-openapi/testify/enable/yaml/v2 v2.6.0 h1:gGHwAJ0R/5jU8BEGDbfRNR3hL68dAVi84WuOApp29B0= +github.com/go-openapi/testify/enable/yaml/v2 v2.6.0/go.mod h1:tY+St1SGq4NFl0QIqdTY4aEdbChAHxhyB77XQi9iJCo= +github.com/go-openapi/testify/v2 v2.6.0 h1:5PKH2HE7YJ/LuRPQGvSxBRlFXNQhSetBLlGAgUEu3ug= +github.com/go-openapi/testify/v2 v2.6.0/go.mod h1:SgsVHtfooshd0tublTtJ50FPKhujf47YRqauXXOUxfw= github.com/go-quicktest/qt v1.101.1-0.20240301121107-c6c8733fa1e6 h1:teYtXy9B7y5lHTp8V9KPxpYRAVA7dozigQcMiBust1s= github.com/go-quicktest/qt v1.101.1-0.20240301121107-c6c8733fa1e6/go.mod h1:p4lGIVX+8Wa6ZPNDvqcxq36XpUDLh42FLetFU7odllI= github.com/go-task/slim-sprig/v3 v3.0.0 h1:sUs3vkvUymDpBKi3qH1YSqBQk9+9D/8M2mN1vB6EwHI= github.com/go-task/slim-sprig/v3 v3.0.0/go.mod h1:W848ghGpv3Qj3dhTPRyJypKRiqCdHZiAzKg9hl15HA8= github.com/go-test/deep v1.1.1 h1:0r/53hagsehfO4bzD2Pgr/+RgHqhmf+k1Bpse2cTu1U= github.com/go-test/deep v1.1.1/go.mod h1:5C2ZWiW0ErCdrYzpqxLbTX7MG14M9iiw8DgHncVwcsE= -github.com/go-viper/mapstructure/v2 v2.4.0 h1:EBsztssimR/CONLSZZ04E8qAkxNYq4Qp9LvH92wZUgs= -github.com/go-viper/mapstructure/v2 v2.4.0/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM= github.com/go-viper/mapstructure/v2 v2.5.0 h1:vM5IJoUAy3d7zRSVtIwQgBj7BiWtMPfmPEgAXnvj1Ro= github.com/go-viper/mapstructure/v2 v2.5.0/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM= github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= @@ -135,8 +103,6 @@ github.com/google/pprof v0.0.0-20260115054156-294ebfa9ad83 h1:z2ogiKUYzX5Is6zr/v github.com/google/pprof v0.0.0-20260115054156-294ebfa9ad83/go.mod h1:MxpfABSjhmINe3F1It9d+8exIHFvUqtLIRCdOGNXqiI= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 h1:5VipnvEpbqr2gA2VbM+nYVbkIF28c5ZQfqCBQ5g2xfk= -github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0/go.mod h1:Hyl3n6Twe1hvtd9XUXDec4pTvgMSEixRuQKPTMH2bNs= github.com/grpc-ecosystem/grpc-gateway/v2 v2.30.0 h1:/Tnpcb2E0Pz/tN9s3bfEY2Q8ePCEX9iuS+cneUwncnw= github.com/grpc-ecosystem/grpc-gateway/v2 v2.30.0/go.mod h1:zOBXOsUaBSjKgmH4OGzV1esUpR3oUSCPYVd2cUBjKYY= github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8= @@ -149,8 +115,8 @@ github.com/k-sone/critbitgo v1.4.0 h1:l71cTyBGeh6X5ATh6Fibgw3+rtNT80BA0uNNWgkPrb github.com/k-sone/critbitgo v1.4.0/go.mod h1:7E6pyoyADnFxlUBEKcnfS49b7SUAQGMK+OAp/UQvo0s= github.com/kenshaw/baseconv v0.1.1 h1:oAu/C7ipUT2PqT9DT0mZDGDg4URIglizZMjPv9oCu0E= github.com/kenshaw/baseconv v0.1.1/go.mod h1:yy9zGmnnR6vgOxOQb702nVdAG30JhyYZpj/5/m0siRI= -github.com/klauspost/compress v1.19.1 h1:VsB4HPswih7mmZ8WleSFQ75c/Ui1M4trX5oAsJnhSlk= -github.com/klauspost/compress v1.19.1/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/compress v1.20.1 h1:T7kKElXUMXrUJ2E9QhQhxFtcK5rPyLdsGZvdbLMPdiQ= +github.com/klauspost/compress v1.20.1/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= @@ -192,25 +158,17 @@ github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINE github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U= github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= -github.com/prometheus/client_golang v1.24.1 h1:JnJkREXzWxUdCuPFpIWZiPispT9xVV59uiuyR2bPlnU= -github.com/prometheus/client_golang v1.24.1/go.mod h1:F+oSRECHg4sse5ucfYpYDeIv/hu68Zo0uoHKetWnzcE= github.com/prometheus/client_golang v1.25.0 h1:cDtW1TcFsMi1RJc8khwCKB4VhzwV04aKqbYtNOXuiK0= github.com/prometheus/client_golang v1.25.0/go.mod h1:j2X/4/kaQHLPnmyEU8zjmkndo1T317VpCIQEml8LTKk= github.com/prometheus/client_model v0.6.3 h1:O0jaTVAYNxTHYInEPFJt5I3+sN8zqBtVMPTB1qyxiEo= github.com/prometheus/client_model v0.6.3/go.mod h1:gpN5P9S7Rr6Yr92PiQ+Ixvhf6JZEkF1dnxsYL2aPBEM= -github.com/prometheus/common v0.70.1 h1:1HvjP4D5oL3t8RsPlwxA9onvvStjtIHYE5XuuwOi/PY= -github.com/prometheus/common v0.70.1/go.mod h1:VdFUQDMZK3VLkurFUVhia6uys/0suUp86TJz5qbJRhc= github.com/prometheus/common v0.72.0 h1:tAYsE+sPJxIncDAobm4H5aQjmox9ZxEIIqPbiffa8G4= github.com/prometheus/common v0.72.0/go.mod h1:77NWqAQ2tXT7BIK40qjJdw5Acrsrg1TlHAnsQi3i6mk= -github.com/prometheus/procfs v0.21.1 h1:GljZCt+zSTS+NZq88cyQ1LjZ+RCHp3uVuabBWA5+OJI= -github.com/prometheus/procfs v0.21.1/go.mod h1:aB55Cww9pdSJVHk0hUf0inxWyyjPogFIjmHKYgMKmtY= github.com/prometheus/procfs v0.22.0 h1:6q9+/JL9IKAPbCmBrv9n5O5Ty3NKnciV5X7YGw0oics= github.com/prometheus/procfs v0.22.0/go.mod h1:CvmFr/GVhIjIvWJZW3tgkODBQMRIf0EyWMQLHCHab58= github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= -github.com/safchain/ethtool v0.6.2 h1:O3ZPFAKEUEfbtE6J/feEe2Ft7dIJ2Sy8t4SdMRiIMHY= -github.com/safchain/ethtool v0.6.2/go.mod h1:VS7cn+bP3Px3rIq55xImBiZGHVLNyBh5dqG6dDQy8+I= github.com/safchain/ethtool v0.7.0 h1:rlJzfDetsVvT61uz8x1YIcFn12akMfuPulHtZjtb7Is= github.com/safchain/ethtool v0.7.0/go.mod h1:MenQKEjXdfkjD3mp2QdCk8B/hwvkrlOTm/FD4gTpFxQ= github.com/sagikazarmark/locafero v0.11.0 h1:1iurJgmM9G3PA/I+wWYIOw/5SyBtxapeHDcg+AAIFXc= @@ -234,8 +192,8 @@ github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+ github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY= github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= -github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= -github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE= +github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg= github.com/subosito/gotenv v1.6.0 h1:9NlTDc1FTs4qu0DDq7AEtTPNw6SVm7uBMsUCUjABIf8= github.com/subosito/gotenv v1.6.0/go.mod h1:Dk4QP5c2W3ibzajGcXpNraDfq2IrhjMIvMSWPKKo0FU= github.com/vishvananda/netlink v1.3.2-0.20261002201547-1d1c62507e3c h1:88GLXNKgc6pwIU/4yq+gleF5ISv3AsJkYmFlS1GjHvo= @@ -246,44 +204,28 @@ github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM= github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg= go.datum.net/cloud v0.1.2 h1:Bs8qR+ZCxf70bZGvjscyXeqoIIy8uNo2zjfdrWzIlGE= go.datum.net/cloud v0.1.2/go.mod h1:HTeaepXL0EqiqNsQ13K4/kGEwv+VLJ//BGrUfVIFfEA= -go.datum.net/network v0.4.0 h1:JeNcJD7Ln9QFnGdbr2sjSWOy4p96gHoUXr3S6tEaqJM= -go.datum.net/network v0.4.0/go.mod h1:LawI34k+ddtv3kRbmjiss0Lb6QKpSr60YP3fX1IG5RU= +go.datum.net/network v0.4.1-0.20261009192728-36ee528904ff h1:tHhqHnOxM3nDs7eCNeoUOcLTyTV05M3QXy4tR6CV9Hc= +go.datum.net/network v0.4.1-0.20261009192728-36ee528904ff/go.mod h1:LawI34k+ddtv3kRbmjiss0Lb6QKpSr60YP3fX1IG5RU= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= -go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.69.0 h1:2yEATaop1/a1I4psnSLgWVPLWwCzkqWakgJy7xTDVy0= -go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.69.0/go.mod h1:D7J12YRapIekYyPWgGPlA/23pRmpSEZC5xJC/TTLI9U= go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.72.0 h1:Tq+E65HaNbZTeqcLw6QUhsV+/EIWdkfUH2CQzcrlE5M= go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.72.0/go.mod h1:A+EVhDakAj1waJBPhmTszZaSr08ww8MRRKy1nB1rqck= -go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU= -go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc= go.opentelemetry.io/otel v1.47.0 h1:j7ALJ/zgkS7Z6aeJW09p8VC9804bC+PpeTfCD4XPnOM= go.opentelemetry.io/otel v1.47.0/go.mod h1:8wS9O2qfXrYrzp6hIF/HOYJJf/wIhFPhR2xLuP+iXQU= -go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0 h1:4YsVu3B8+3qtWYYrsUYgn0OG78pN0rnNPRGX4SbokQI= -go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0/go.mod h1:+wnlSn0mD1ADVMe3v9Z/WIaiz6q6gL2J/ejaAmdmv80= go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.47.0 h1:julhjPeUH/q/7hinbSdDdqt5h7Zw9YWmRlWRhI0jd54= go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.47.0/go.mod h1:Ao2mz688LH/tFf0yMAenidq6k2YNSx6SIY2q6jDACck= -go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.44.0 h1:qazEJlUOQzhCpzQpFETGby7EdqjI1wsd0W+6Gg1SCTU= -go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.44.0/go.mod h1:fOD2Yefuxixkx3ahVNf0O/PERb6r4OlbxfATVnYvzCo= go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.47.0 h1:UFpxOpYPmMNUtOWhdb+nC1WELIgVf8z9higURrbzYU4= go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.47.0/go.mod h1:YBGjxe3lt0jtXoEXxGrGhv/jM7oUBUDVK7zhAvNyEjc= go.opentelemetry.io/otel/log v1.47.0 h1:cOTS1CcLbSQeZKanGJ+0JpF/+t4PELi3O3bbl2lqCcI= go.opentelemetry.io/otel/log v1.47.0/go.mod h1:9byitSQ5pLC6PpqwGXjqdMKya6ZTswHRZh2vvXT33nw= -go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc= -go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo= go.opentelemetry.io/otel/metric v1.47.0 h1:4PptaldXx3Eat1XjMZ68pPJEs5wrhlemctZE9a3UdWY= go.opentelemetry.io/otel/metric v1.47.0/go.mod h1:ADGSXxRrXM6bjbvLo535EstVFlPpPYZm4LBKixjDHwU= -go.opentelemetry.io/otel/sdk v1.44.0 h1:nHYwb9lK+fJPU/dnT6s7W7Z8itMWyqrnVfbheVYrZ58= -go.opentelemetry.io/otel/sdk v1.44.0/go.mod h1:Osuydd3Se74nqjAKxid74N5eC+jfEqfTegHRnq58oK0= go.opentelemetry.io/otel/sdk v1.47.0 h1:zWXEr4j2lFefG87TU6Yg8a7ngfohIKFZHKp0Hf5hC6I= go.opentelemetry.io/otel/sdk v1.47.0/go.mod h1:VUc24kiOeoGsxG8G9ULx3fWKvB7jMhnGE8Oi607lgR0= -go.opentelemetry.io/otel/sdk/metric v1.44.0 h1:3LlKgI+VjbVsjNRFZJZAJ30WjXC5VkNRks6si09iEfI= -go.opentelemetry.io/otel/sdk/metric v1.44.0/go.mod h1:5B5pMARnXxKhltooO4xUuCBorl65a4EpnTalObqOigA= -go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk= -go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE= +go.opentelemetry.io/otel/sdk/metric v1.47.0 h1:lfISg2j93VT6yqdk9OfUaZmw/GfcZqCCV3jdXtsPnKw= +go.opentelemetry.io/otel/sdk/metric v1.47.0/go.mod h1:ypLp+mW1Nt2x+Szt3b5/i1syodyts49lMOwxpDI3VGw= go.opentelemetry.io/otel/trace v1.47.0 h1:JOjX/Oci8K94QHddo+bbfya/Ai/nf6/dt9ZfrFNWSrM= go.opentelemetry.io/otel/trace v1.47.0/go.mod h1:jNaSLa2PZEYFG6fRjJABAu+bw4FS08uDmPg28lTghu0= -go.opentelemetry.io/proto/otlp v1.10.0 h1:IQRWgT5srOCYfiWnpqUYz9CVmbO8bFmKcwYxpuCSL2g= -go.opentelemetry.io/proto/otlp v1.10.0/go.mod h1:/CV4QoCR/S9yaPj8utp3lvQPoqMtxXdzn7ozvvozVqk= go.opentelemetry.io/proto/otlp v1.11.0 h1:5rrYs0Ykyj50sdU/JU0x8etU+LubXWb+gED6TbEdMIk= go.opentelemetry.io/proto/otlp v1.11.0/go.mod h1:SmVizdCOAm3XBtG1g1NnOdhW6jtddT72hLMhv8VwA8E= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= @@ -294,7 +236,6 @@ go.uber.org/zap v1.28.0 h1:IZzaP1Fv73/T/pBMLk4VutPl36uNC+OSUh3JLG3FIjo= go.uber.org/zap v1.28.0/go.mod h1:rDLpOi171uODNm/mxFcuYWxDsqWSAVkFdX4XojSKg/Q= go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ= go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ= -go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc= go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw= go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg= @@ -305,47 +246,29 @@ golang.org/x/exp/typeparams v0.0.0-20260209203927-2842357ff358/go.mod h1:4Mzdyp/ golang.org/x/lint v0.0.0-20210508222113-6edffad5e616 h1:VLliZ0d+/avPrXXH+OakdXhpJuEoBZuwh1m2j7U6Iug= golang.org/x/lint v0.0.0-20210508222113-6edffad5e616/go.mod h1:3xt1FjdF8hUf6vQPIChWIBhFzV8gjjsPE/fR3IyQdNY= golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg= -golang.org/x/mod v0.38.0 h1:MECBjubtXD7yj4HrhIUcywNaGeNVUdfVnxmPajOk4yk= -golang.org/x/mod v0.38.0/go.mod h1:V6Xz0pq8TQ3dGqVQ1FVHuelZpAL0uNhSkk9ogYP3c40= golang.org/x/mod v0.41.0 h1:qJmnOUb4YB+FsEuM3HcWucdZASCPGhsX6uljO6pog0c= golang.org/x/mod v0.41.0/go.mod h1:Ek9pY8RKWXwsWvd3rQiHYtMqkjSUV+s1Rj7j4H5Ur6o= golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= -golang.org/x/net v0.58.0 h1:ynWG7rqYi4ccpTEuPZ2QGWHktVEM9DMCj9yzDE0Q7To= -golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU= golang.org/x/net v0.61.0 h1:RR+6j/BTBCrggOLugxa+66R8FvoN6MHlzDlZIbN9YFw= golang.org/x/net v0.61.0/go.mod h1:WgDCOTH8iwtB66T4/GfI94AHo/hhdb7Hah9NDseQfiI= -golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs= -golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q= golang.org/x/oauth2 v0.37.0 h1:JUlcxA8oAtauLfiH8FX2/FkAWHAdi0QtGCGc+hofE98= golang.org/x/oauth2 v0.37.0/go.mod h1:IxwZNxUULJmpBFf9K/9NTMSIfZZuvuTy1gGxhigP/58= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= -golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= golang.org/x/sync v0.24.0 h1:iqiDrGoDgyuvAKrhLQthUEzU/R378N78gXbBjxed7oA= golang.org/x/sync v0.24.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= -golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= golang.org/x/sys v0.49.0 h1:XbzkgYJHdqh/8m2Uu0W/dQv8nktxx4BFHp1M0gROTXA= golang.org/x/sys v0.49.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= -golang.org/x/term v0.46.0 h1:3+OXuTbaKDgwk8jTi3aSLHRlmWqHEUDUtxnbFigO4YE= -golang.org/x/term v0.46.0/go.mod h1:+K02xbkittuwc0Am4abfA3Fc+XRGXkvBXNO88NCXPoc= golang.org/x/term v0.47.0 h1:cpyOwQkr2Nzxe2p6KnrTG7Ei2wtzjsuSjhpXXepZ/BM= golang.org/x/term v0.47.0/go.mod h1:efAfAuG6RKABvg1ArGebRkvX6LXul14SM67o2e4mUCY= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= -golang.org/x/text v0.41.0 h1:vz/seA0lnX87Othu2f/0L24RcgrXD9/YFTSuGjj3rH8= -golang.org/x/text v0.41.0/go.mod h1:jvf1O8ajNzZqhSrQBPbutR/EB83Cc0CFrezNQIwbb5M= golang.org/x/text v0.43.0 h1:1QivrlhwAsnMOcqgOdl9my7s9OyZl8O5/UGUTJVXoWc= golang.org/x/text v0.43.0/go.mod h1:hbSIYA/amXcRXTFSZqh8tWHEoqRvvbKPJ2uz/KU+a+s= -golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= -golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= golang.org/x/time v0.16.0 h1:vMb6ptszcQMkcwiRTAuNNU50gom6++Q/6gY2hDM6VDE= golang.org/x/time v0.16.0/go.mod h1:rVKOqvZeKvrDKTQiAHJ7wmwP0RzleSphoEA9RcdLA0s= golang.org/x/tools v0.0.0-20200130002326-2f3ba24bd6e7/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28= -golang.org/x/tools v0.48.0 h1:3+hClM1aLL5mjMKm5ovokw9epgRXPuu2tILgismM6RE= -golang.org/x/tools v0.48.0/go.mod h1:08xX0orndb/F7jJxGDicx061tyd5pcMto75YMAXr6lk= golang.org/x/tools v0.51.0 h1:k4Xc/1Om9jwkBJBo4NVLMSARBoWtK10mx+W5BnXCeAI= golang.org/x/tools v0.51.0/go.mod h1:9eEncMayCV6zRMGhR5eZEC2iBx98qWcF1HZ9Z7wJOoA= golang.org/x/tools/go/expect v0.1.1-deprecated h1:jpBZDwmgPhXsKZC6WhL20P4b/wmnpsEAGHaNy0n/rJM= @@ -355,12 +278,8 @@ gomodules.xyz/jsonpatch/v2 v2.5.0 h1:JELs8RLM12qJGXU4u/TO3V25KW8GreMKl9pdkk14RM0 gomodules.xyz/jsonpatch/v2 v2.5.0/go.mod h1:AH3dM2RI6uoBZxn3LVrfvJ3E0/9dG4cSrbuBJT4moAY= gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= -google.golang.org/genproto/googleapis/api v0.0.0-20260706201446-f0a921348800 h1:admdQBe8jR3VWhBsUrAOaF2Qw6K/+p5pSm1GN8+6Fw4= -google.golang.org/genproto/googleapis/api v0.0.0-20260706201446-f0a921348800/go.mod h1:FPk7EXUKMtImne7AmknoYjT4QXqKIzzRbeQIXzLk6fQ= google.golang.org/genproto/googleapis/api v0.0.0-20260928230214-8a89bd6388cc h1:rZHRz0ogq4Pid1IKtA/ivHi8XONVF5tqUQXA2TtgpkE= google.golang.org/genproto/googleapis/api v0.0.0-20260928230214-8a89bd6388cc/go.mod h1:K/n0XLazCJHjD73YuMyNJsN38BPPy1kfFnKxUXaThFU= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260706201446-f0a921348800 h1:qEHAMpSaUhtD0p3NbEEI83HwNGFxEwaSJ1G9PLnCBZE= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260706201446-f0a921348800/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= google.golang.org/genproto/googleapis/rpc v0.0.0-20261005182115-fad411399dd8 h1:0SXBqli5dnuCKMjI/X3FM226zEaEa0GUcSCJtN1vL2o= google.golang.org/genproto/googleapis/rpc v0.0.0-20261005182115-fad411399dd8/go.mod h1:OaIUM3+LpYcK2GXM4FTmhWoIq371Owdr+Cc7/BsYHHc= google.golang.org/grpc v1.84.0 h1:soMyaPJ8pAak5PIQ0DGBUir0XRo2fRoMqhNWMLlLxO0= @@ -368,14 +287,10 @@ google.golang.org/grpc v1.84.0/go.mod h1:ljCht0DrxQrXBDRTZp52Qxh3Ffk8CdYm2sj4O2Q google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc= google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= -gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= -gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= gopkg.in/evanphx/json-patch.v4 v4.13.0 h1:czT3CmqEaQ1aanPc5SdlgQrrEIb8w/wwCvWWnfEbYzo= gopkg.in/evanphx/json-patch.v4 v4.13.0/go.mod h1:p8EYWUEYMpynmqDbY58zCKCFZw8pRWMG4EsWvDvM72M= gopkg.in/inf.v0 v0.9.1 h1:73M5CoZyi3ZLMOyDlQh031Cx6N9NDJ2Vvfl76EDAgDc= gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw= -gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= -gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= honnef.co/go/tools v0.7.0 h1:w6WUp1VbkqPEgLz4rkBzH/CSU6HkoqNLp6GstyTx3lU= honnef.co/go/tools v0.7.0/go.mod h1:pm29oPxeP3P82ISxZDgIYeOaf9ta6Pi0EWvCFoLG2vc= k8s.io/api v0.36.3 h1:NxB+05W2UGqXWFXcLO0RB5cnqnUPP5v5sVlaOH0Iz4w= diff --git a/internal/config/cni.go b/internal/config/cni.go index dcf21279..10dd07b1 100644 --- a/internal/config/cni.go +++ b/internal/config/cni.go @@ -5,11 +5,14 @@ package config import ( + "errors" "fmt" "net" "net/netip" "os" + "strconv" "strings" + "time" "go.datum.net/galactic/internal/plumbing/dan" ) @@ -95,8 +98,114 @@ const ( // Unset or empty means this fabric has no NAT64, and a VRF gets no such // route -- not an error. EnvCNINAT64Prefix = "GALACTIC_CNI_NAT64_PREFIX" + + // EnvCNIEgressMode selects how a tenant VRF's egress reaches the + // cluster's egress shards. EgressModeOrdered, the default, sends each VRF + // to the first shard in EnvCNIEgressShardSIDs that resolves. + // EgressModeHashed spreads tenants across every healthy EgressShard in the + // cluster, picking a shard per tenant address in the datapath, and does not + // read EnvCNIEgressShardSIDs. Read only by galactic-cni run, which owns + // the shard group CNI ADD then follows. + EnvCNIEgressMode = "GALACTIC_CNI_EGRESS_MODE" + + // EnvCNIEgressHash selects what hashed mode hashes: "source" (the + // default) keeps every connection from one tenant address on one shard + // and one public address, as RFC 4787 REQ-2 asks; "flow" hashes the + // 5-tuple, spreading one heavy address over every shard at the cost of + // that pairing. + EnvCNIEgressHash = "GALACTIC_CNI_EGRESS_HASH" + + // EnvCNIEgressPinIdle is how long, as a Go duration, hashed mode keeps a + // tenant address on the shard it was first sent to after its last packet, + // so adding or draining a shard does not move established sessions. "0" + // turns pinning off. The default, DefaultEgressPinIdle, is the egress + // shard's longest idle timeout. + EnvCNIEgressPinIdle = "GALACTIC_CNI_EGRESS_PIN_IDLE" + + // EnvCNIEgressPoolMinActive is the fewest active egress shards the cluster + // is expected to have. It is exported as a metric for the pool alert to + // compare against, and does not change forwarding. Default 1. + EnvCNIEgressPoolMinActive = "GALACTIC_CNI_EGRESS_POOL_MIN_ACTIVE" +) + +// EgressMode is how tenant VRFs reach the cluster's egress shards; see +// EnvCNIEgressMode. +type EgressMode string + +// The EgressMode values. +const ( + EgressModeOrdered EgressMode = "ordered" + EgressModeHashed EgressMode = "hashed" ) +// The EnvCNIEgressHash values. +const ( + EgressHashSource = "source" + EgressHashFlow = "flow" +) + +// DefaultEgressPinIdle is EnvCNIEgressPinIdle's default: 2 h 4 min, the +// egress shard's established-TCP idle timeout (RFC 5382 REQ-5). +const DefaultEgressPinIdle = 2*time.Hour + 4*time.Minute + +// EgressGroupConfig is galactic-cni run's hashed-mode configuration. +type EgressGroupConfig struct { + Mode EgressMode + Hash string + PinIdle time.Duration + // MinActive is EnvCNIEgressPoolMinActive. + MinActive int +} + +// ParseEgressGroupConfig reads the hashed-mode settings from getenv. Unset +// values take their defaults; a set but invalid one is an error, so a typo +// fails the installer at startup instead of silently running ordered. +func ParseEgressGroupConfig(getenv func(string) string) (EgressGroupConfig, error) { + cfg := EgressGroupConfig{ + Mode: EgressModeOrdered, + Hash: EgressHashSource, + PinIdle: DefaultEgressPinIdle, + MinActive: 1, + } + var errs []error + switch mode := EgressMode(strings.TrimSpace(getenv(EnvCNIEgressMode))); mode { + case "": + case EgressModeOrdered, EgressModeHashed: + cfg.Mode = mode + default: + errs = append(errs, fmt.Errorf("%s=%q: want %q or %q", EnvCNIEgressMode, mode, EgressModeOrdered, EgressModeHashed)) + } + switch hash := strings.TrimSpace(getenv(EnvCNIEgressHash)); hash { + case "": + case EgressHashSource, EgressHashFlow: + cfg.Hash = hash + default: + errs = append(errs, fmt.Errorf("%s=%q: want %q or %q", EnvCNIEgressHash, hash, EgressHashSource, EgressHashFlow)) + } + if raw := strings.TrimSpace(getenv(EnvCNIEgressPinIdle)); raw != "" { + idle, err := time.ParseDuration(raw) + switch { + case raw == "0": + cfg.PinIdle = 0 + case err != nil: + errs = append(errs, fmt.Errorf("%s=%q: %w", EnvCNIEgressPinIdle, raw, err)) + case idle < time.Second: + errs = append(errs, fmt.Errorf("%s=%q: want 0 or at least 1s", EnvCNIEgressPinIdle, raw)) + default: + cfg.PinIdle = idle + } + } + if raw := strings.TrimSpace(getenv(EnvCNIEgressPoolMinActive)); raw != "" { + n, err := strconv.Atoi(raw) + if err != nil || n < 0 { + errs = append(errs, fmt.Errorf("%s=%q: want a non-negative integer", EnvCNIEgressPoolMinActive, raw)) + } else { + cfg.MinActive = n + } + } + return cfg, errors.Join(errs...) +} + // --- CNIConfig ------------------------------------------------------------- // CNIConfig resolves CNI configuration with three-tier precedence: environment diff --git a/internal/config/cni_test.go b/internal/config/cni_test.go index 8f717352..0c962aa9 100644 --- a/internal/config/cni_test.go +++ b/internal/config/cni_test.go @@ -6,6 +6,7 @@ package config import ( "testing" + "time" "go.datum.net/galactic/internal/plumbing/dan" ) @@ -288,3 +289,48 @@ func TestParseNAT64Prefixes(t *testing.T) { }) } } + +func TestParseEgressGroupConfig(t *testing.T) { + tests := []struct { + name string + env map[string]string + want EgressGroupConfig + wantErr bool + }{ + { + name: "DefaultsToOrderedSourceWithPins", + want: EgressGroupConfig{ + Mode: EgressModeOrdered, Hash: EgressHashSource, PinIdle: DefaultEgressPinIdle, MinActive: 1, + }, + }, + { + name: "HashedFlowWithoutPins", + env: map[string]string{ + EnvCNIEgressMode: "hashed", EnvCNIEgressHash: "flow", EnvCNIEgressPinIdle: "0", + EnvCNIEgressPoolMinActive: "3", + }, + want: EgressGroupConfig{Mode: EgressModeHashed, Hash: EgressHashFlow, PinIdle: 0, MinActive: 3}, + }, + { + name: "CustomPinIdle", + env: map[string]string{EnvCNIEgressPinIdle: "45m"}, + want: EgressGroupConfig{Mode: EgressModeOrdered, Hash: EgressHashSource, PinIdle: 45 * time.Minute, MinActive: 1}, + }, + {name: "UnknownModeFails", env: map[string]string{EnvCNIEgressMode: "anycast"}, wantErr: true}, + {name: "UnknownHashFails", env: map[string]string{EnvCNIEgressHash: "5tuple"}, wantErr: true}, + {name: "BadPinIdleFails", env: map[string]string{EnvCNIEgressPinIdle: "forever"}, wantErr: true}, + {name: "SubSecondPinIdleFails", env: map[string]string{EnvCNIEgressPinIdle: "10ms"}, wantErr: true}, + {name: "NegativeMinActiveFails", env: map[string]string{EnvCNIEgressPoolMinActive: "-1"}, wantErr: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := ParseEgressGroupConfig(func(k string) string { return tt.env[k] }) + if (err != nil) != tt.wantErr { + t.Fatalf("ParseEgressGroupConfig() error = %v, wantErr %v", err, tt.wantErr) + } + if !tt.wantErr && got != tt.want { + t.Errorf("ParseEgressGroupConfig() = %+v, want %+v", got, tt.want) + } + }) + } +} diff --git a/internal/installer/egressgroup.go b/internal/installer/egressgroup.go new file mode 100644 index 00000000..5bd8e47a --- /dev/null +++ b/internal/installer/egressgroup.go @@ -0,0 +1,682 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package installer + +import ( + "context" + "errors" + "fmt" + "log/slog" + "net" + "net/netip" + "os" + "sync" + "sync/atomic" + "time" + + "github.com/cilium/ebpf" + "github.com/vishvananda/netlink" + "k8s.io/apimachinery/pkg/api/meta" + "k8s.io/client-go/rest" + toolscache "k8s.io/client-go/tools/cache" + "sigs.k8s.io/controller-runtime/pkg/cache" + "sigs.k8s.io/controller-runtime/pkg/client" + + "go.datum.net/galactic/internal/config" + "go.datum.net/galactic/internal/plumbing/ebpf/attach" + "go.datum.net/galactic/internal/plumbing/ebpf/egressroutemap" + "go.datum.net/galactic/internal/plumbing/ebpf/metrics" + "go.datum.net/galactic/internal/plumbing/ebpf/prog" + "go.datum.net/galactic/internal/plumbing/ebpf/usidmap" + bgpv1alpha1 "go.datum.net/network/api/v1alpha1" +) + +// egressGroupDebounce is how long a burst of route, neighbor or EgressShard +// events is collected before one sweep handles all of it. A BGP withdrawal +// removes a shard's route and then its neighbor in quick succession, and a +// sweep for each would only repeat the work. +var egressGroupDebounce = 250 * time.Millisecond + +// egressGroup keeps egress_route_table and the egress shard groups in line +// with the configured egress mode. +// +// In ordered mode it re-resolves every entry and keeps each VRF on the first +// reachable shard of the static list, as the egress route sweep always has. +// Leaving hashed mode takes three steps, each only after the one before it +// succeeded in full: every enabled group is marked closing, so no CNI ADD +// writes a new route naming it while it keeps forwarding; every route naming +// a group is moved onto the static list; and only once a sweep that started +// after the groups closed has moved every one, and found none left, are the +// groups disabled. A failed scan or write leaves the groups forwarding. +// +// In hashed mode there is one group per translation class: NAT66 for ::/0, +// and one per NAT64 prefix. A class's members are the cluster's EgressShards +// that can translate it, read from a watch, and a member is alive only while +// its SID resolves to an uplink neighbor. Every VRF route bound to a shard is +// moved onto its class's group, after which usid_egress picks a shard per +// tenant address. +// +// All of it runs on one goroutine, started by run, and is triggered by kick: +// on the egress route ticker, on any EgressShard change, and on any route or +// neighbor change that can move a shard's reachability, so a withdrawn shard +// leaves the group as soon as BGP withdraws its route rather than at the next +// tick. A kick during a sweep queues exactly one more. +type egressGroup struct { + cfg config.EgressGroupConfig + staticSIDs []net.IP + pinDir string + trigger chan struct{} + + // classes are the translation classes this node's VRFs need, in group + // order: NAT66, then each configured NAT64 prefix. + classes []egressroutemap.TranslationClass + + // shards lists the cluster's EgressShards. Nil in ordered mode, and in + // hashed mode until the watch has started. + shards egressShardLister + + // counters is egress_shard_counters, whose slots Apply hands to new + // shards. Nil when the datapath is a test fake. + counters *ebpf.Map + + // ctl reports the control loop's state. Nil in tests. + ctl *metrics.EgressControl + + // locators is the set of /64 locators the groups' candidates live in, + // for the netlink filter. Written by the sweep goroutine. + locators atomic.Pointer[[]netip.Prefix] + + mu sync.Mutex + // active and ineligible are each class's last applied counts, for + // logging a change once rather than every sweep. + active map[string]int + ineligible map[string]string +} + +// egressShardLister reads the cluster's EgressShards. synced is false until +// the first full list has arrived; an unsynced lister's empty answer is not +// "no shards". +type egressShardLister interface { + ListEgressShards(ctx context.Context) (shards []bgpv1alpha1.EgressShard, synced bool, err error) +} + +// newEgressGroup builds the sweep for cfg. staticSIDs is the conflist's shard +// list, which ordered mode places VRFs on, and which hashed mode still +// recognizes so a VRF written in ordered mode moves onto a group. nat64 is the +// conflist's NAT64 prefix list, which decides the classes. +func newEgressGroup( + cfg config.EgressGroupConfig, staticSIDs []net.IP, nat64 []netip.Prefix, pinDir string, +) *egressGroup { + classes := make([]egressroutemap.TranslationClass, 0, 1+len(nat64)) + classes = append(classes, egressroutemap.NAT66Class) + for _, p := range nat64 { + classes = append(classes, egressroutemap.NAT64Class(p)) + } + return &egressGroup{ + cfg: cfg, staticSIDs: staticSIDs, pinDir: pinDir, classes: classes, trigger: make(chan struct{}, 1), + active: map[string]int{}, ineligible: map[string]string{}, + } +} + +// checkEgressClasses refuses a hashed-mode configuration with more NAT64 +// prefixes than there are shard groups for. +func checkEgressClasses(cfg config.EgressGroupConfig, nat64 []netip.Prefix) error { + if cfg.Mode == config.EgressModeHashed && len(nat64) > egressroutemap.MaxNAT64Groups { + return fmt.Errorf("hashed egress mode serves at most %d NAT64 prefixes, and %s lists %d", + egressroutemap.MaxNAT64Groups, config.EnvCNINAT64Prefix, len(nat64)) + } + return nil +} + +// kick asks for a sweep without blocking. +func (g *egressGroup) kick() { + select { + case g.trigger <- struct{}{}: + default: + } +} + +// run sweeps on every kick until ctx ends. +func (g *egressGroup) run(ctx context.Context) { + for { + select { + case <-ctx.Done(): + return + case <-g.trigger: + } + select { + case <-ctx.Done(): + return + case <-time.After(egressGroupDebounce): + } + g.sweep(ctx) + } +} + +// policy is the group policy cfg asks for, for class. +func (g *egressGroup) policy(class egressroutemap.TranslationClass, ineligible int) egressroutemap.GroupPolicy { + policy := egressroutemap.GroupPolicy{Class: class, PinIdle: g.cfg.PinIdle, Ineligible: ineligible} + if g.cfg.Hash == config.EgressHashFlow { + policy.Hash = egressroutemap.HashFlow + } + return policy +} + +// sweep runs one pass. A missing pinned map means the datapath is not loaded +// on this node, and there is nothing to do. +// +// The ingress sidecar shares these maps from inside a pod network namespace, +// so the sweep first reads back which VRF routing tables that writer owns and +// leaves them alone. Failing to read that set skips the whole sweep: a sweep +// that cannot tell the two writers apart rewrites the sidecar's entries to +// host interfaces the pod does not have, and a stale next hop is recoverable +// where that is not. +func (g *egressGroup) sweep(ctx context.Context) { + table, closer, err := egressroutemap.OpenPinnedEgressRouteTable(g.pinDir) + if err != nil { + return + } + defer func() { _ = closer.Close() }() + + registry, registryCloser, err := usidmap.OpenPinnedRegistry(g.pinDir) + if err != nil { + return + } + defer func() { _ = registryCloser.Close() }() + + foreignTableIDs, err := egressroutemap.SidecarOwnedTableIDs(registry.VRF) + if err != nil { + slog.Error("egress_route_table refresh sweep skipped: could not resolve entry ownership", "err", err) + return + } + + groups, groupsCloser, err := egressroutemap.OpenPinnedShardGroupTable(g.pinDir) + if err != nil { + // A datapath older than the shard group map: ordered mode is all it + // can run. + if !errors.Is(err, os.ErrNotExist) { + slog.Error("egress shard groups unavailable; running the egress route sweep in ordered mode", "err", err) + } + groups = nil + } else { + defer func() { _ = groupsCloser.Close() }() + } + + if err := g.sweepWith(ctx, table, groups, foreignTableIDs); err == nil { + g.ctl.SweepSucceeded(time.Now()) + } +} + +// sweepWith is one sweep against already-open maps. groups is nil on a +// datapath without the shard group map. It returns the first error, after +// counting it. +func (g *egressGroup) sweepWith( + ctx context.Context, table *egressroutemap.EgressRouteTable, groups *egressroutemap.ShardGroupTable, + foreignTableIDs map[uint32]struct{}, +) error { + if groups == nil { + _, err := g.refresh(table, foreignTableIDs, g.staticSIDs) + return err + } + if g.cfg.Mode == config.EgressModeHashed { + return g.sweepHashed(ctx, table, groups, foreignTableIDs) + } + return g.sweepOrdered(table, groups, foreignTableIDs) +} + +// sweepOrdered moves every route off the shard groups and retires them. +// +// The groups are marked closing before the scan that moves routes off them, +// and disabled only if that scan read the whole table and wrote every move, +// and found no route still naming a group. A CNI ADD that read the groups +// before they closed checks again after writing its routes and rewrites them +// in ordered mode if they closed meanwhile, so no route naming a group is +// written after a scan that started after they closed. A scan that fails, or +// leaves a route it could not move, leaves the groups forwarding, and the +// next sweep tries again. +func (g *egressGroup) sweepOrdered( + table *egressroutemap.EgressRouteTable, groups *egressroutemap.ShardGroupTable, + foreignTableIDs map[uint32]struct{}, +) error { + var enabled []uint32 + closed := true + for id := range uint32(egressroutemap.MaxShardGroups) { + status, err := groups.Status(id) + if err != nil { + g.ctl.SweepError("close") + return fmt.Errorf("read egress shard group %d: %w", id, err) + } + if !status.Enabled { + continue + } + enabled = append(enabled, id) + if ok, err := groups.Close(id); err != nil || !ok { + g.ctl.SweepError("close") + slog.Error("Failed to mark an egress shard group closing; it stays enabled", "group", id, "err", err) + closed = false + } + } + + result, err := g.refresh(table, foreignTableIDs, g.staticSIDs) + if err != nil { + return err + } + if len(enabled) == 0 { + return nil + } + if !closed || result.SentinelRoutes() > 0 { + slog.Info("Egress shard groups stay enabled while routes still name them", + "routes", result.SentinelRoutes(), "allClosed", closed) + return nil + } + for _, id := range enabled { + if changed, err := groups.Disable(id); err != nil { + g.ctl.SweepError("disable") + return fmt.Errorf("disable egress shard group %d: %w", id, err) + } else if changed { + slog.Info("Egress shard group disabled: ordered egress mode, no route names it", "group", id) + } + } + return nil +} + +// sweepHashed applies the cluster's shards to each class's group and moves +// every VRF route onto its class's group. Until the watch has listed the +// cluster once, each group keeps whatever it last held: an empty answer then +// means "not known yet", and applying it would drop the cluster's egress. +func (g *egressGroup) sweepHashed( + ctx context.Context, table *egressroutemap.EgressRouteTable, groups *egressroutemap.ShardGroupTable, + foreignTableIDs map[uint32]struct{}, +) error { + recognized := append([]net.IP(nil), g.staticSIDs...) + var firstErr error + if g.shards == nil { + g.ctl.SetWatchSynced(false) + } else { + shards, synced, err := g.shards.ListEgressShards(ctx) + g.ctl.SetWatchSynced(synced && err == nil) + switch { + case err != nil: + g.ctl.SweepError("watch") + slog.Error("Failed to list EgressShards; the egress shard groups keep their members", "err", err) + firstErr = err + case synced: + sids, err := g.applyClasses(groups, shards) + recognized = append(recognized, sids...) + if err != nil { + firstErr = err + } + } + } + + // Each route goes to its class's group only while that group is open and + // has a candidate. Until then a route keeps the shard ordered mode gave + // it, rather than be moved onto a group with nothing to send it to. + byClass := map[string]uint32{} + for id := range uint32(egressroutemap.MaxShardGroups) { + status, err := groups.Status(id) + if err != nil { + return errors.Join(firstErr, err) + } + if status.Open() && status.Candidates > 0 { + byClass[status.Class.String()] = id + } + } + groupFor := func(prefix *net.IPNet) (uint32, bool) { + class, ok := egressroutemap.ClassForRoute(prefix) + if !ok { + return 0, false + } + id, ok := byClass[class.String()] + return id, ok + } + var opts []egressroutemap.RefreshOption + if len(byClass) > 0 { + opts = append(opts, egressroutemap.WithShardGroups(groupFor)) + } + if _, err := g.refresh(table, foreignTableIDs, recognized, opts...); err != nil { + return errors.Join(firstErr, err) + } + return firstErr +} + +// applyClasses publishes each class's group from the cluster's shards and +// returns every candidate's SID. +func (g *egressGroup) applyClasses( + groups *egressroutemap.ShardGroupTable, shards []bgpv1alpha1.EgressShard, +) ([]net.IP, error) { + var ( + sids []net.IP + locators []netip.Prefix + errs []error + ) + for i, class := range g.classes { + eligible, ineligible := egressShardCandidates(shards, class) + g.reportIneligible(class, ineligible) + result, err := groups.Apply(uint32(i), g.policy(class, len(ineligible)), eligible) //nolint:gosec // < 4 + if err != nil { + g.ctl.SweepError("apply") + slog.Error("Failed to publish an egress shard group", "class", class.String(), "err", err) + errs = append(errs, err) + continue + } + for _, s := range result.Shards { + locators = append(locators, netip.PrefixFrom(s.SID, 64).Masked()) + if s.Assigned { + g.resetCounter(uint32(i), s.Slot) //nolint:gosec // < 4 + } + } + for _, c := range eligible { + sids = append(sids, c.SID) + } + g.logApplied(class, result) + } + g.locators.Store(&locators) + return sids, errors.Join(errs...) +} + +// reportIneligible logs the shards that cannot serve class, once per change. +func (g *egressGroup) reportIneligible(class egressroutemap.TranslationClass, ineligible []string) { + key := fmt.Sprint(ineligible) + g.mu.Lock() + previous, seen := g.ineligible[class.String()] + g.ineligible[class.String()] = key + g.mu.Unlock() + if seen && previous == key { + return + } + if len(ineligible) > 0 { + slog.Warn("EgressShards that cannot translate this class are kept out of its group", + "class", class.String(), "shards", ineligible) + } +} + +// logApplied logs a group's membership when it changed. +func (g *egressGroup) logApplied(class egressroutemap.TranslationClass, result egressroutemap.GroupResult) { + g.mu.Lock() + previous, seen := g.active[class.String()] + g.active[class.String()] = result.Active() + g.mu.Unlock() + if !result.Changed && seen && previous == result.Active() { + return + } + for _, s := range result.Shards { + if s.Err != nil { + slog.Warn("Egress shard unreachable; it takes no tenant until its SID resolves", + "class", class.String(), "sid", s.SID, "slot", s.Slot, "err", s.Err) + } + } + slog.Info("Egress shard group updated", "class", class.String(), + "generation", result.Generation, "candidates", len(result.Shards), "active", result.Active(), + "hash", g.cfg.Hash, "pinIdle", g.cfg.PinIdle) +} + +// resetCounter zeroes a slot's per-CPU packet and byte counters, so a shard +// that takes a slot does not inherit its previous occupant's share. +func (g *egressGroup) resetCounter(group uint32, slot int) { + if g.counters == nil { + return + } + cpus, err := ebpf.PossibleCPU() + if err != nil { + slog.Warn("Could not reset an egress shard slot's counters", "slot", slot, "err", err) + return + } + key := group*egressroutemap.MaxShardsPerGroup + uint32(slot) //nolint:gosec // < 32 + if err := g.counters.Put(key, make([]prog.UsidEgressShardCounter, cpus)); err != nil { + slog.Warn("Could not reset an egress shard slot's counters", "slot", slot, "err", err) + } +} + +// refresh runs one Refresh pass, logs anything it did, and records how many +// routes still name each class's group. Refresh's error is returned, never +// swallowed: a caller deciding whether a group can be retired must know the +// scan was complete. +func (g *egressGroup) refresh( + table *egressroutemap.EgressRouteTable, foreignTableIDs map[uint32]struct{}, shardSIDs []net.IP, + opts ...egressroutemap.RefreshOption, +) (egressroutemap.RefreshResult, error) { + result, err := table.Refresh(foreignTableIDs, shardSIDs, opts...) + if err != nil { + g.ctl.SweepError("refresh") + slog.Error("egress_route_table refresh sweep failed", "err", err, + "scanned", result.Scanned, "refreshed", result.Refreshed) + return result, err + } + if result.Refreshed > 0 || result.Unresolved > 0 { + slog.Info("egress_route_table refresh sweep complete", + "scanned", result.Scanned, "refreshed", result.Refreshed, "reselected", result.Reselected, + "grouped", result.Grouped, "ungrouped", result.Ungrouped, "inGroup", result.InGroup, + "unresolved", result.Unresolved, "skipped", result.Skipped) + } + byClass := map[string]int{} + for i, class := range g.classes { + byClass[class.String()] = result.GroupRoutes[uint32(i)] //nolint:gosec // < 4 + } + g.ctl.SetGroupRoutes(byClass) + return result, nil +} + +// egressShardCandidates returns the cluster's EgressShards that can serve +// class as group candidates, and the names of those that are candidates in +// every other respect but cannot translate class. +// +// A shard is a candidate once its datapath reports the SID it translates with, +// in status, and both Programmed and Ready are true: the same evidence the +// shard's own advertisement is built from. Status rather than spec, since only +// status says what the shard's datapath claims. A status gone stale on a dead +// node is caught by liveness instead: its route is withdrawn with its BGP +// session, and a candidate whose SID does not resolve takes no tenant. +// +// It can serve NAT66 when its status carries an IPv6 masquerade address, and +// NAT64 toward a prefix when it carries an IPv4 one and translates that +// prefix: its own nat64Prefix, or the Well-Known Prefix when it says it +// translates that too. +// +// One with spec.drain set is draining. One being deleted is no candidate at +// all: deletion withdraws its advertisement and clears its translation at +// once, so it has nothing to drain with. +func egressShardCandidates( + shards []bgpv1alpha1.EgressShard, class egressroutemap.TranslationClass, +) (eligible []egressroutemap.ShardCandidate, ineligible []string) { + for i := range shards { + shard := &shards[i] + if !shard.DeletionTimestamp.IsZero() || shard.Status.ShardSID == "" || + !meta.IsStatusConditionTrue(shard.Status.Conditions, bgpv1alpha1.ConditionTypeProgrammed) || + !meta.IsStatusConditionTrue(shard.Status.Conditions, bgpv1alpha1.ConditionTypeReady) { + continue + } + sid := net.ParseIP(shard.Status.ShardSID) + if sid == nil { + slog.Warn("Ignoring EgressShard with an unparseable status.shardSID", + "egressShard", shard.Namespace+"/"+shard.Name, "shardSID", shard.Status.ShardSID) + continue + } + if !translates(shard.Status, class) { + ineligible = append(ineligible, shard.Name) + continue + } + eligible = append(eligible, egressroutemap.ShardCandidate{SID: sid, Draining: shard.Spec.Drain}) + } + return eligible, ineligible +} + +// translates reports whether a shard with status can translate class. +func translates(status bgpv1alpha1.EgressShardStatus, class egressroutemap.TranslationClass) bool { + if !class.NAT64 { + return status.ShardAddressIPv6 != "" + } + if status.ShardAddressIPv4 == "" { + return false + } + if p, err := netip.ParsePrefix(status.NAT64Prefix); err == nil && p.Masked() == class.Prefix { + return true + } + return class.Prefix == egressroutemap.WellKnownPrefix && status.TranslatesWellKnownPrefix +} + +// newRestConfigFn is an override point so tests can avoid an in-cluster +// config. +var newRestConfigFn = rest.InClusterConfig + +// startEgressShardWatchFn starts the EgressShard watch hashed mode reads its +// members from, calling onChange on every add, update and delete. A variable +// so tests can substitute a fake lister. +var startEgressShardWatchFn = startEgressShardWatch + +// cacheShardLister lists EgressShards from a controller-runtime cache. +type cacheShardLister struct { + reader client.Reader + namespace string + synced atomic.Bool +} + +func (l *cacheShardLister) ListEgressShards(ctx context.Context) ([]bgpv1alpha1.EgressShard, bool, error) { + if !l.synced.Load() { + return nil, false, nil + } + var list bgpv1alpha1.EgressShardList + if err := l.reader.List(ctx, &list, client.InNamespace(l.namespace)); err != nil { + return nil, true, fmt.Errorf("list EgressShards in %s: %w", l.namespace, err) + } + return list.Items, true, nil +} + +// startEgressShardWatch starts an informer on namespace's EgressShards for the +// life of ctx. It returns once the informer is registered, not once it has +// synced; the lister reports synced from then on. +func startEgressShardWatch(ctx context.Context, namespace string, onChange func()) (egressShardLister, error) { + restConfig, err := newRestConfigFn() + if err != nil { + return nil, fmt.Errorf("load in-cluster config: %w", err) + } + c, err := cache.New(restConfig, cache.Options{ + Scheme: scheme, + DefaultNamespaces: map[string]cache.Config{namespace: {}}, + }) + if err != nil { + return nil, fmt.Errorf("create EgressShard cache: %w", err) + } + informer, err := c.GetInformer(ctx, &bgpv1alpha1.EgressShard{}, cache.BlockUntilSynced(false)) + if err != nil { + return nil, fmt.Errorf("get EgressShard informer: %w", err) + } + if _, err := informer.AddEventHandler(toolscache.ResourceEventHandlerFuncs{ + AddFunc: func(any) { onChange() }, + UpdateFunc: func(any, any) { onChange() }, + DeleteFunc: func(any) { onChange() }, + }); err != nil { + return nil, fmt.Errorf("watch EgressShards: %w", err) + } + + lister := &cacheShardLister{reader: c, namespace: namespace} + go func() { + if err := c.Start(ctx); err != nil { + slog.Error("EgressShard watch stopped", "err", err) + } + }() + go func() { + if c.WaitForCacheSync(ctx) { + lister.synced.Store(true) + onChange() + } + }() + return lister, nil +} + +// watchShardReachability kicks g whenever a route or neighbor changes in a way +// that can move a group member's reachability: a route into one of the +// members' /64 locators, the default route, or any IPv6 neighbor on an SRv6 +// uplink. Subscription errors are logged and the subscription retried, since +// the egress route ticker still covers the gap. +func (g *egressGroup) watchShardReachability(ctx context.Context) { + for ctx.Err() == nil { + if err := g.subscribeReachability(ctx); err != nil { + slog.Warn("Egress shard reachability watch failed; retrying", "err", err) + } + select { + case <-ctx.Done(): + return + case <-time.After(5 * time.Second): + } + } +} + +// subscribeReachability runs one route and neighbor subscription until ctx +// ends or either closes. +func (g *egressGroup) subscribeReachability(ctx context.Context) error { + done := make(chan struct{}) + defer close(done) + + routes := make(chan netlink.RouteUpdate, 64) + if err := netlink.RouteSubscribe(routes, done); err != nil { + return fmt.Errorf("subscribe to route updates: %w", err) + } + neighs := make(chan netlink.NeighUpdate, 64) + if err := netlink.NeighSubscribe(neighs, done); err != nil { + return fmt.Errorf("subscribe to neighbor updates: %w", err) + } + + for { + select { + case <-ctx.Done(): + return nil + case update, ok := <-routes: + if !ok { + return errors.New("route subscription closed") + } + if g.routeAffectsShards(update.Dst) { + g.kick() + } + case update, ok := <-neighs: + if !ok { + return errors.New("neighbor subscription closed") + } + if update.Family == netlink.FAMILY_V6 && isUplink(update.LinkIndex) { + g.kick() + } + } + } +} + +// routeAffectsShards reports whether a route to dst can change how a group +// member resolves: the default route, or any prefix overlapping a member's +// locator. +func (g *egressGroup) routeAffectsShards(dst *net.IPNet) bool { + if dst == nil { + return true + } + addr, ok := netip.AddrFromSlice(dst.IP) + if !ok { + return false + } + ones, _ := dst.Mask.Size() + prefix := netip.PrefixFrom(addr.Unmap(), ones) + if prefix.Bits() == 0 { + return true + } + locators := g.locators.Load() + if locators == nil { + return false + } + for _, l := range *locators { + if l.Overlaps(prefix) { + return true + } + } + return false +} + +// isUplink reports whether ifindex is an interface the SRv6 datapath is +// attached to. A lookup failure answers true, so a neighbor change is never +// missed for want of the uplink set. +func isUplink(ifindex int) bool { + uplinks, err := attach.UplinkIndexes() + if err != nil { + return true + } + _, ok := uplinks[ifindex] + return ok +} diff --git a/internal/installer/egressgroup_test.go b/internal/installer/egressgroup_test.go new file mode 100644 index 00000000..dd8ab644 --- /dev/null +++ b/internal/installer/egressgroup_test.go @@ -0,0 +1,535 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package installer + +import ( + "context" + "errors" + "net" + "net/netip" + "reflect" + "testing" + + "github.com/cilium/ebpf" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + "go.datum.net/galactic/internal/config" + "go.datum.net/galactic/internal/plumbing/ebpf/egressroutemap" + "go.datum.net/galactic/internal/plumbing/ebpf/prog" + "go.datum.net/galactic/internal/plumbing/ebpf/uformat" + "go.datum.net/galactic/internal/plumbing/ebpf/usidmap" + bgpv1alpha1 "go.datum.net/network/api/v1alpha1" +) + +var ( + testNSP = netip.MustParsePrefix("2001:db8:64::/96") + testNSPNet = &net.IPNet{IP: net.IP(testNSP.Addr().AsSlice()), Mask: net.CIDRMask(96, 128)} + nat64NSP = egressroutemap.NAT64Class(testNSP) + nat64WKP = egressroutemap.NAT64Class(egressroutemap.WellKnownPrefix) + testHashed = config.EgressGroupConfig{Mode: config.EgressModeHashed, Hash: config.EgressHashSource} + testOrdered = config.EgressGroupConfig{Mode: config.EgressModeOrdered, Hash: config.EgressHashSource} +) + +// shardObject is an EgressShard whose status reports sid, the masquerade +// addresses v6 and v4 (either may be empty), a NAT64 prefix and the two +// conditions. +func shardObject(name, sid, v6, v4, nat64 string, programmed, ready bool) bgpv1alpha1.EgressShard { + status := func(ok bool) metav1.ConditionStatus { + if ok { + return metav1.ConditionTrue + } + return metav1.ConditionFalse + } + return bgpv1alpha1.EgressShard{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "galactic-system"}, + Status: bgpv1alpha1.EgressShardStatus{ + ShardSID: sid, ShardAddressIPv6: v6, ShardAddressIPv4: v4, NAT64Prefix: nat64, + Conditions: []metav1.Condition{ + {Type: bgpv1alpha1.ConditionTypeProgrammed, Status: status(programmed)}, + {Type: bgpv1alpha1.ConditionTypeReady, Status: status(ready)}, + }, + }, + } +} + +func shardSIDFor(nodeID uint16) string { + addr, _ := uformat.Encode(uformat.Fields{Block: 0x20010db8ff09, NodeID: nodeID, Function: uformat.FunctionEndDT46}) + return addr.String() +} + +func dual(name string, nodeID uint16) bgpv1alpha1.EgressShard { + return shardObject(name, shardSIDFor(nodeID), "2001:db8:9966::1", "192.0.2.1", testNSP.String(), true, true) +} + +func candidateNames(t *testing.T, got []egressroutemap.ShardCandidate) map[string]bool { + t.Helper() + out := map[string]bool{} + for _, c := range got { + out[c.SID.String()] = c.Draining + } + return out +} + +// TestEgressShardCandidates covers pool admission (F04): each class admits only +// shards that can translate it, readiness and SID are required, drain is +// reported, and a shard being deleted is no candidate at all (F03). +func TestEgressShardCandidates(t *testing.T) { + nat66Only := shardObject("nat66-only", shardSIDFor(0x11), "2001:db8:9966::2", "", "", true, true) + nat64Only := shardObject("nat64-only", shardSIDFor(0x12), "", "192.0.2.3", testNSP.String(), true, true) + otherPrefix := shardObject("other-prefix", shardSIDFor(0x13), "2001:db8:9966::4", "192.0.2.4", + "2001:db8:65::/96", true, true) + wkp := shardObject("wkp", shardSIDFor(0x14), "", "192.0.2.5", testNSP.String(), true, true) + wkp.Status.TranslatesWellKnownPrefix = true + draining := dual("draining", 0x15) + draining.Spec.Drain = true + deleting := dual("deleting", 0x16) + now := metav1.Now() + deleting.DeletionTimestamp = &now + shards := []bgpv1alpha1.EgressShard{ + dual("dual", 0x10), nat66Only, nat64Only, otherPrefix, wkp, draining, deleting, + shardObject("not-programmed", shardSIDFor(0x20), "2001:db8:9966::9", "", "", false, true), + shardObject("not-ready", shardSIDFor(0x21), "2001:db8:9966::9", "", "", true, false), + shardObject("no-sid", "", "2001:db8:9966::9", "", "", true, true), + shardObject("bad-sid", "not-an-address", "2001:db8:9966::9", "", "", true, true), + } + + tests := []struct { + class egressroutemap.TranslationClass + eligible map[string]bool + ineligible []string + }{ + { + class: egressroutemap.NAT66Class, + eligible: map[string]bool{ + shardSIDFor(0x10): false, shardSIDFor(0x11): false, shardSIDFor(0x13): false, shardSIDFor(0x15): true, + }, + ineligible: []string{"nat64-only", "wkp"}, + }, + { + class: nat64NSP, + eligible: map[string]bool{ + shardSIDFor(0x10): false, shardSIDFor(0x12): false, shardSIDFor(0x14): false, shardSIDFor(0x15): true, + }, + ineligible: []string{"nat66-only", "other-prefix"}, + }, + { + class: nat64WKP, + eligible: map[string]bool{shardSIDFor(0x14): false}, + ineligible: []string{"dual", "nat66-only", "nat64-only", "other-prefix", "draining"}, + }, + } + for _, tt := range tests { + t.Run(tt.class.String(), func(t *testing.T) { + eligible, ineligible := egressShardCandidates(shards, tt.class) + if got := candidateNames(t, eligible); !reflect.DeepEqual(got, tt.eligible) { + t.Errorf("eligible = %v, want %v", got, tt.eligible) + } + if !reflect.DeepEqual(ineligible, tt.ineligible) { + t.Errorf("ineligible = %v, want %v", ineligible, tt.ineligible) + } + }) + } +} + +func TestCheckEgressClasses(t *testing.T) { + four := []netip.Prefix{testNSP, egressroutemap.WellKnownPrefix, + netip.MustParsePrefix("2001:db8:65::/96"), netip.MustParsePrefix("2001:db8:66::/96")} + if err := checkEgressClasses(testHashed, four[:3]); err != nil { + t.Errorf("three NAT64 prefixes in hashed mode: %v, want accepted", err) + } + if err := checkEgressClasses(testHashed, four); err == nil { + t.Error("four NAT64 prefixes in hashed mode accepted, want an error") + } + if err := checkEgressClasses(testOrdered, four); err != nil { + t.Errorf("four NAT64 prefixes in ordered mode: %v, want accepted", err) + } +} + +func TestRouteAffectsShards(t *testing.T) { + g := newEgressGroup(config.EgressGroupConfig{}, nil, nil, "") + locators := []netip.Prefix{netip.MustParsePrefix("2001:db8:ff09:10::/64")} + g.locators.Store(&locators) + + cidr := func(s string) *net.IPNet { + _, n, err := net.ParseCIDR(s) + if err != nil { + t.Fatal(err) + } + return n + } + tests := []struct { + dst *net.IPNet + want bool + }{ + {nil, true}, + {cidr("::/0"), true}, + {cidr("2001:db8:ff09:10::/64"), true}, + {cidr("2001:db8:ff09::/48"), true}, + {cidr("2001:db8:ff09:10:e000::/128"), true}, + {cidr("2001:db8:ff09:11::/64"), false}, + {cidr("10.0.0.0/8"), false}, + } + for _, tt := range tests { + if got := g.routeAffectsShards(tt.dst); got != tt.want { + t.Errorf("routeAffectsShards(%v) = %v, want %v", tt.dst, got, tt.want) + } + } +} + +func TestEgressGroupKickCoalesces(t *testing.T) { + g := newEgressGroup(config.EgressGroupConfig{}, nil, nil, "") + for range 5 { + g.kick() + } + if len(g.trigger) != 1 { + t.Errorf("queued kicks = %d, want 1", len(g.trigger)) + } +} + +// --- Sweep sequencing, against in-memory maps with injected faults --------- + +// memStore is an in-memory GroupStore. +type memStore map[uint32]prog.UsidEgressShardGroupValue + +func (s memStore) Load(id uint32) (prog.UsidEgressShardGroupValue, bool, error) { + v, ok := s[id] + return v, ok, nil +} + +func (s memStore) Publish(id uint32, v prog.UsidEgressShardGroupValue) error { + s[id] = v + return nil +} + +// routeTable is an in-memory egress_route_table with injected faults. +// iterErr fails the next iteration; putFailures fails that many Puts; +// afterScan runs once an iteration has read every entry, standing in for a +// concurrent writer. +type routeTable struct { + keys []prog.UsidEgressRouteKey + values map[prog.UsidEgressRouteKey]prog.UsidEgressRouteValue + iterErr error + putFailures int + afterScan func() +} + +func newRouteTable() *routeTable { + return &routeTable{values: map[prog.UsidEgressRouteKey]prog.UsidEgressRouteValue{}} +} + +var errInjected = errors.New("injected map failure") + +func (r *routeTable) Put(key, value any) error { + if r.putFailures > 0 { + r.putFailures-- + return errInjected + } + k := key.(prog.UsidEgressRouteKey) + if _, ok := r.values[k]; !ok { + r.keys = append(r.keys, k) + } + r.values[k] = value.(prog.UsidEgressRouteValue) + return nil +} + +func (r *routeTable) Lookup(key, out any) error { + v, ok := r.values[key.(prog.UsidEgressRouteKey)] + if !ok { + return ebpf.ErrKeyNotExist + } + reflect.ValueOf(out).Elem().Set(reflect.ValueOf(v)) + return nil +} + +func (r *routeTable) Delete(key any) error { + delete(r.values, key.(prog.UsidEgressRouteKey)) + return nil +} + +func (r *routeTable) Iterate() usidmap.Iterator { + keys := append([]prog.UsidEgressRouteKey(nil), r.keys...) + err := r.iterErr + r.iterErr = nil + return &routeIter{table: r, keys: keys, err: err} +} + +type routeIter struct { + table *routeTable + keys []prog.UsidEgressRouteKey + i int + err error + done bool +} + +func (it *routeIter) Next(keyOut, valueOut any) bool { + if it.i >= len(it.keys) { + if !it.done && it.table.afterScan != nil { + it.done = true + hook := it.table.afterScan + it.table.afterScan = nil + hook() + } + return false + } + k := it.keys[it.i] + it.i++ + reflect.ValueOf(keyOut).Elem().Set(reflect.ValueOf(k)) + reflect.ValueOf(valueOut).Elem().Set(reflect.ValueOf(it.table.values[k])) + return true +} + +func (it *routeIter) Err() error { return it.err } + +// fixture is one node: its egress_route_table, its shard groups and an +// egressGroup sweeping them. +type fixture struct { + raw *routeTable + routes *egressroutemap.EgressRouteTable + groups *egressroutemap.ShardGroupTable + g *egressGroup +} + +func newFixture(t *testing.T, cfg config.EgressGroupConfig, static []string, nat64 []netip.Prefix) *fixture { + t.Helper() + restore := egressroutemap.OverrideResolverForTest(func(net.IP) (int, net.HardwareAddr, net.HardwareAddr, error) { + return 7, net.HardwareAddr{2, 0, 0, 0, 0, 1}, net.HardwareAddr{2, 0, 0, 0, 0, 2}, nil + }) + t.Cleanup(restore) + sids := make([]net.IP, 0, len(static)) + for _, s := range static { + sids = append(sids, net.ParseIP(s)) + } + raw := newRouteTable() + return &fixture{ + raw: raw, + routes: egressroutemap.NewEgressRouteTable(raw), + groups: egressroutemap.NewShardGroupTable(memStore{}), + g: newEgressGroup(cfg, sids, nat64, ""), + } +} + +func (f *fixture) sweep(t *testing.T) error { + t.Helper() + return f.g.sweepWith(context.Background(), f.routes, f.groups, nil) +} + +// sentinels counts the routes naming any group. +func (f *fixture) sentinels() int { + n := 0 + for _, v := range f.raw.values { + if v.LinkIfindex == 0xFFFFFFFF { + n++ + } + } + return n +} + +// hashedNode leaves the fixture as hashed mode would: a NAT66 group with one +// shard, and VRFs 1-3 routed to it. +func (f *fixture) hashedNode(t *testing.T) { + t.Helper() + shard := net.ParseIP(shardSIDFor(0x10)) + candidates := []egressroutemap.ShardCandidate{{SID: shard}} + if _, err := f.groups.Apply(0, egressroutemap.GroupPolicy{}, candidates); err != nil { + t.Fatal(err) + } + for vrf := uint32(1); vrf <= 3; vrf++ { + argument := uint16(0x10 + vrf) //nolint:gosec // test + if err := f.routes.RegisterGroup(vrf, egressroutemap.DefaultPrefix, 0, argument); err != nil { + t.Fatal(err) + } + } +} + +func status(t *testing.T, groups *egressroutemap.ShardGroupTable, id uint32) egressroutemap.GroupStatus { + t.Helper() + s, err := groups.Status(id) + if err != nil { + t.Fatal(err) + } + return s +} + +// TestSweepOrdered_RetiresGroupsOnlyAfterACompleteConversion covers F01: a +// switch to ordered mode marks the groups closing, and disables them only +// after a sweep that read every route and wrote every move. An iteration +// failure or a failed write leaves the groups forwarding; the next sweep +// finishes the job. +func TestSweepOrdered_RetiresGroupsOnlyAfterACompleteConversion(t *testing.T) { + static := []string{shardSIDFor(0x10)} + t.Run("IterationFailure", func(t *testing.T) { + f := newFixture(t, testOrdered, static, nil) + f.hashedNode(t) + f.raw.iterErr = errInjected + if err := f.sweep(t); !errors.Is(err, errInjected) { + t.Fatalf("sweep = %v, want the iteration failure", err) + } + if s := status(t, f.groups, 0); !s.Enabled || !s.Closing || f.sentinels() != 3 { + t.Errorf("after a failed scan: status %+v with %d sentinels, want closing, enabled, 3 sentinels", + s, f.sentinels()) + } + }) + t.Run("PartialWriteThenRetry", func(t *testing.T) { + f := newFixture(t, testOrdered, static, nil) + f.hashedNode(t) + f.raw.putFailures = 1 + if err := f.sweep(t); !errors.Is(err, errInjected) { + t.Fatalf("sweep = %v, want the write failure", err) + } + if s := status(t, f.groups, 0); !s.Enabled || f.sentinels() == 0 { + t.Fatalf("after a failed write: status %+v with %d sentinels, want the group still enabled", s, + f.sentinels()) + } + if err := f.sweep(t); err != nil { + t.Fatalf("retry = %v", err) + } + if s := status(t, f.groups, 0); s.Enabled || f.sentinels() != 0 { + t.Errorf("after the retry: status %+v with %d sentinels, want disabled and none left", s, f.sentinels()) + } + }) + t.Run("NoStaticShardKeepsTheGroup", func(t *testing.T) { + f := newFixture(t, testOrdered, nil, nil) + f.hashedNode(t) + if err := f.sweep(t); err != nil { + t.Fatal(err) + } + if s := status(t, f.groups, 0); !s.Enabled || f.sentinels() != 3 { + t.Errorf("with nothing to move routes to: status %+v with %d sentinels, want the group kept", s, + f.sentinels()) + } + }) + t.Run("ConcurrentADDAfterTheScan", func(t *testing.T) { + // An ADD that read the groups open, and writes its sentinel after the + // sweep's iterator has passed. It must find the groups already + // closing when it checks again, and so rewrite its route itself. + f := newFixture(t, testOrdered, static, nil) + f.hashedNode(t) + f.raw.afterScan = func() { + if err := f.routes.RegisterGroup(9, egressroutemap.DefaultPrefix, 0, 0x99); err != nil { + t.Error(err) + } + if open, _ := f.groups.AnyOpen(); open { + t.Error("the groups were still open when the sweep's scan finished; a late ADD would not rewrite") + return + } + sid, _ := egressroutemap.BaseShardSID(net.ParseIP(static[0])) + b := sid.As16() + b[9] = 0x99 + if err := f.routes.Register(9, egressroutemap.DefaultPrefix, net.IP(b[:])); err != nil { + t.Error(err) + } + } + if err := f.sweep(t); err != nil { + t.Fatal(err) + } + if f.sentinels() != 0 { + t.Errorf("%d sentinels survive the sweep that disabled their group", f.sentinels()) + } + }) + t.Run("Success", func(t *testing.T) { + f := newFixture(t, testOrdered, static, nil) + f.hashedNode(t) + if err := f.sweep(t); err != nil { + t.Fatal(err) + } + if s := status(t, f.groups, 0); s.Enabled || f.sentinels() != 0 { + t.Errorf("status %+v with %d sentinels, want disabled and none left", s, f.sentinels()) + } + }) +} + +// fakeLister is an egressShardLister over a fixed answer. +type fakeLister struct { + shards []bgpv1alpha1.EgressShard + synced bool + err error +} + +func (l *fakeLister) ListEgressShards(context.Context) ([]bgpv1alpha1.EgressShard, bool, error) { + return l.shards, l.synced, l.err +} + +// TestSweepHashed_ClassGroups checks hashed mode end to end on the control +// side: each class's group holds only the shards that translate it, routes +// move onto their class's group, and a membership change removes a shard +// that is no longer eligible. +func TestSweepHashed_ClassGroups(t *testing.T) { + f := newFixture(t, testHashed, nil, []netip.Prefix{testNSP}) + nat66Only := shardObject("nat66-only", shardSIDFor(0x11), "2001:db8:9966::2", "", "", true, true) + lister := &fakeLister{shards: []bgpv1alpha1.EgressShard{dual("dual", 0x10), nat66Only}, synced: true} + f.g.shards = lister + + staticSID, _ := egressroutemap.BaseShardSID(net.ParseIP(shardSIDFor(0x10))) + withArg := staticSID.As16() + withArg[9] = 0x21 + if err := f.routes.Register(1, egressroutemap.DefaultPrefix, net.IP(withArg[:])); err != nil { + t.Fatal(err) + } + if err := f.routes.Register(1, testNSPNet, net.IP(withArg[:])); err != nil { + t.Fatal(err) + } + + if err := f.sweep(t); err != nil { + t.Fatal(err) + } + nat66, nat64 := status(t, f.groups, 0), status(t, f.groups, 1) + if nat66.Class != egressroutemap.NAT66Class || nat66.Candidates != 2 || nat66.Ineligible != 0 { + t.Errorf("NAT66 group = %+v, want both shards", nat66) + } + if nat64.Class != nat64NSP || nat64.Candidates != 1 || nat64.Ineligible != 1 { + t.Errorf("NAT64 group = %+v, want only the dual shard, one ineligible", nat64) + } + if f.sentinels() != 2 { + t.Errorf("%d routes on groups, want both", f.sentinels()) + } + + // The dual shard stops being Programmed: it leaves both groups. + lister.shards[0].Status.Conditions[0].Status = metav1.ConditionFalse + if err := f.sweep(t); err != nil { + t.Fatal(err) + } + if s := status(t, f.groups, 1); s.Candidates != 0 { + t.Errorf("NAT64 group after its only shard left = %+v, want no candidate", s) + } + members, _ := f.groups.Members(0) + if len(members) != 1 || members[0].SID.String() != staticSIDFor(0x11) { + t.Errorf("NAT66 members = %+v, want only the NAT66-only shard", members) + } +} + +func staticSIDFor(nodeID uint16) string { + sid, _ := egressroutemap.BaseShardSID(net.ParseIP(shardSIDFor(nodeID))) + return sid.String() +} + +// TestSweepHashed_UnsyncedOrFailingWatch checks that until the watch has listed +// the cluster, and while listing fails, no group is published or changed and +// no route moves: an unknown membership is not an empty one. +func TestSweepHashed_UnsyncedOrFailingWatch(t *testing.T) { + f := newFixture(t, testHashed, []string{shardSIDFor(0x10)}, nil) + f.g.shards = &fakeLister{synced: false} + if err := f.routes.Register(1, egressroutemap.DefaultPrefix, net.ParseIP(shardSIDFor(0x10))); err != nil { + t.Fatal(err) + } + if err := f.sweep(t); err != nil { + t.Fatal(err) + } + if s := status(t, f.groups, 0); s.Present || f.sentinels() != 0 { + t.Errorf("unsynced watch published %+v and moved %d routes, want nothing", s, f.sentinels()) + } + + f.g.shards = &fakeLister{shards: []bgpv1alpha1.EgressShard{dual("dual", 0x10)}, synced: true} + if err := f.sweep(t); err != nil { + t.Fatal(err) + } + before := status(t, f.groups, 0) + f.g.shards = &fakeLister{synced: true, err: errInjected} + if err := f.sweep(t); !errors.Is(err, errInjected) { + t.Fatalf("sweep with a failing watch = %v, want the error", err) + } + if after := status(t, f.groups, 0); after != before { + t.Errorf("a failing watch changed the group: %+v -> %+v", before, after) + } +} diff --git a/internal/installer/installer.go b/internal/installer/installer.go index fb7006a5..f675b0d3 100644 --- a/internal/installer/installer.go +++ b/internal/installer/installer.go @@ -12,6 +12,7 @@ import ( "log/slog" "net" "net/http" + "net/netip" "os" "path/filepath" "strings" @@ -35,10 +36,8 @@ import ( "go.datum.net/galactic/internal/hostgw" "go.datum.net/galactic/internal/plumbing/ebpf/attach" "go.datum.net/galactic/internal/plumbing/ebpf/attachreg" - "go.datum.net/galactic/internal/plumbing/ebpf/egressroutemap" "go.datum.net/galactic/internal/plumbing/ebpf/metrics" "go.datum.net/galactic/internal/plumbing/ebpf/prog" - "go.datum.net/galactic/internal/plumbing/ebpf/usidmap" "go.datum.net/galactic/internal/plumbing/radv" bgpv1alpha1 "go.datum.net/network/api/v1alpha1" ) @@ -501,10 +500,10 @@ type ebpfDatapathState struct { // Nil when the datapath is a test fake. mssClamp *mssClampState - // egressShardSIDs is the configured egress shard list, in preference - // order, from the same host conflist CNI ADD reads it from. The egress - // route sweep uses it to keep each VRF on the first reachable shard. - egressShardSIDs []net.IP + // egressGroup is the egress route sweep: it keeps each VRF on the first + // reachable shard of the conflist's list in ordered mode, and on the + // cluster's shard group in hashed mode. + egressGroup *egressGroup // egress is the same host conflist's raw egress configuration, which the // datapath repair rebuilds a VRF's shard routes from exactly as CNI ADD @@ -524,6 +523,11 @@ type ebpfDatapathState struct { // only forwarding path, so there is no partial or legacy state to fall back // to. func startEBPFDatapath(ctx context.Context, m *metrics.Metrics) (ebpfDatapathState, io.Closer, error) { + egressCfg, err := config.ParseEgressGroupConfig(os.Getenv) + if err != nil { + return ebpfDatapathState{}, nil, fmt.Errorf("egress shard configuration: %w", err) + } + attach.SetHooks(m.Events.Hooks()) datapath, ifaces, watcher, err := ebpfStartFn(ctx, attach.PinDir) @@ -532,7 +536,12 @@ func startEBPFDatapath(ctx context.Context, m *metrics.Metrics) (ebpfDatapathSta } slog.Info("eBPF uSID datapath loaded, pinned, and attached", "interfaces", ifaces, "pinDir", attach.PinDir) - state := ebpfDatapathState{ifaces: ifaces, watcher: watcher, sidecarReap: &sidecarReaper{}} + state := ebpfDatapathState{ + ifaces: ifaces, watcher: watcher, sidecarReap: &sidecarReaper{}, + egressGroup: newEgressGroup(egressCfg, nil, nil, attach.PinDir), + } + state.egressGroup.ctl = m.Egress + m.Egress.SetMode(string(egressCfg.Mode), string(config.EgressModeOrdered), string(config.EgressModeHashed)) watcher.OnChange(rewriteEBPFInterfaces) // The closer is the loaded objects in production. A test fake stands in a @@ -542,9 +551,13 @@ func startEBPFDatapath(ctx context.Context, m *metrics.Metrics) (ebpfDatapathSta state.objs = objs state.mssClamp = newMSSClampState(objs.MssClampTable, objs.EncapMtuTable) state.mssClamp.reconcile() + state.egressGroup.counters = objs.EgressShardCounters if err := m.RegisterDatapathCollector(objs); err != nil { slog.Warn("Failed to register eBPF datapath metrics collector", "err", err) } + if err := m.RegisterEgressShardCollector(objs, egressCfg.MinActive); err != nil { + slog.Warn("Failed to register egress shard metrics collector", "err", err) + } } // Best-effort setup for the eBPF map GC sweeps. Unlike the datapath start @@ -557,17 +570,64 @@ func startEBPFDatapath(ctx context.Context, m *metrics.Metrics) (ebpfDatapathSta "err", err) return state, datapath, nil } - state.egressShardSIDs = loadEgressShardSIDs(hostConf.EgressShardSIDs) + nat64, err := egressNAT64Prefixes(hostConf.NAT64Prefix) + if err != nil { + slog.Warn("Egress shard groups serve NAT66 only: the conflist's NAT64 prefix list does not parse", "err", err) + } + if err := checkEgressClasses(egressCfg, nat64); err != nil { + return state, datapath, fmt.Errorf("egress shard configuration: %w", err) + } + state.egressGroup = newEgressGroup(egressCfg, loadEgressShardSIDs(hostConf.EgressShardSIDs), nat64, attach.PinDir) + state.egressGroup.ctl = m.Egress + if state.objs != nil { + state.egressGroup.counters = state.objs.EgressShardCounters + } state.egress = attachreg.EgressConfig{ShardSIDs: hostConf.EgressShardSIDs, NAT64Prefix: hostConf.NAT64Prefix} if k8sClient, err := newK8sClientFn(); err != nil { slog.Warn("eBPF vrf_table GC sweep disabled: failed to create k8s client", "err", err) } else { state.k8sClient, state.namespace, state.nodeName = k8sClient, hostConf.Namespace, hostConf.NodeName } + if egressCfg.Mode == config.EgressModeHashed { + startHashedEgress(ctx, state.egressGroup, hostConf.Namespace) + } return state, datapath, nil } +// egressNAT64Prefixes parses the conflist's NAT64 prefix list into the +// prefixes the NAT64 shard groups serve, in order. +func egressNAT64Prefixes(raw string) ([]netip.Prefix, error) { + parsed, err := config.ParseNAT64Prefixes(raw) + if err != nil { + return nil, err + } + prefixes := make([]netip.Prefix, 0, len(parsed)) + for _, p := range parsed { + addr, _ := netip.AddrFromSlice(p.IP) + ones, _ := p.Mask.Size() + prefixes = append(prefixes, netip.PrefixFrom(addr, ones).Masked()) + } + return prefixes, nil +} + +// startHashedEgress starts what hashed mode needs beyond the sweep itself: the +// EgressShard watch its members come from, and the netlink watch that makes a +// shard's withdrawal take effect at once. A watch that fails to start is +// logged; the group keeps its last members and the ticker keeps sweeping. +func startHashedEgress(ctx context.Context, g *egressGroup, namespace string) { + lister, err := startEgressShardWatchFn(ctx, namespace, g.kick) + if err != nil { + slog.Error("Hashed egress: EgressShard watch failed to start; the shard group keeps its last members", + "err", err) + } else { + g.shards = lister + } + go g.watchShardReachability(ctx) + slog.Info("Hashed egress mode: spreading tenant egress across the cluster's EgressShards", + "namespace", namespace, "hash", g.cfg.Hash, "pinIdle", g.cfg.PinIdle) +} + // loadEgressShardSIDs parses the host conflist's egress shard list for the // egress route sweep. A list that fails to parse also fails every ADD on this // node, which is where it gets reported; the sweep just keeps each entry's @@ -632,64 +692,6 @@ func startSidecarReturnSweep(ctx context.Context, sem chan struct{}, st ebpfData } } -// startEgressRouteRefreshSweep runs one egress_route_table re-resolution pass, -// which also moves each VRF onto the first reachable of shardSIDs, off Run's -// goroutine, for the same reason the two sweeps above run off it: an -// entry whose next hop has no neighbor costs a solicit plus a poll, so a node -// that has lost its fabric uplink would hold the select loop past the next tick -// and starve the credential refresh, GC sweeps, and health check with it. -// -// sem is a size-1 semaphore, dropping a tick rather than queueing it when the -// previous pass is still running. -// -// A missing pinned map is not an error worth logging on every tick: it means -// the datapath is not loaded on this node, and the sweep has nothing to do. -// -// The ingress sidecar shares these pinned maps from inside a pod network -// namespace, so the sweep first reads back which VRF routing tables that writer -// owns and passes them to Refresh to be left alone. Failing to read that set -// skips the whole sweep: a sweep that cannot tell the two writers apart -// rewrites the sidecar's entries to host interfaces the pod does not have, and -// a stale next hop is recoverable where that is not. -func startEgressRouteRefreshSweep(sem chan struct{}, shardSIDs []net.IP) { - select { - case sem <- struct{}{}: - go func() { - defer func() { <-sem }() - table, closer, err := egressroutemap.OpenPinnedEgressRouteTable(attach.PinDir) - if err != nil { - return - } - defer func() { _ = closer.Close() }() - - registry, registryCloser, err := usidmap.OpenPinnedRegistry(attach.PinDir) - if err != nil { - return - } - defer func() { _ = registryCloser.Close() }() - - foreignTableIDs, err := egressroutemap.SidecarOwnedTableIDs(registry.VRF) - if err != nil { - slog.Error("egress_route_table refresh sweep skipped: could not resolve entry ownership", "err", err) - return - } - - result, err := table.Refresh(foreignTableIDs, shardSIDs) - if err != nil { - slog.Error("egress_route_table refresh sweep failed", "err", err, - "scanned", result.Scanned, "refreshed", result.Refreshed) - return - } - if result.Refreshed > 0 || result.Unresolved > 0 { - slog.Info("egress_route_table refresh sweep complete", - "scanned", result.Scanned, "refreshed", result.Refreshed, "reselected", result.Reselected, - "unresolved", result.Unresolved, "skipped", result.Skipped) - } - }() - default: - } -} - // radvActorSet tracks the running radv.RunActor goroutines, keyed by host // interface name, plus the retry state of attachments not currently served. It // is Run's local state, reconciled against the recorded attachments on every @@ -1189,7 +1191,8 @@ func Run(ctx context.Context, grpcHealthPort, metricsPort int) error { egressRouteRefreshTicker := time.NewTicker(egressRouteRefreshInterval) defer egressRouteRefreshTicker.Stop() - egressRouteRefreshSem := make(chan struct{}, 1) + go ebpfState.egressGroup.run(ctx) + ebpfState.egressGroup.kick() for { select { @@ -1283,8 +1286,15 @@ func Run(ctx context.Context, grpcHealthPort, metricsPort int) error { // stay wrong for the life of the node. The same pass moves each // VRF onto the first reachable shard in the configured order, // since the one ADD picked may just have been the first whose - // route happened to arrive. - startEgressRouteRefreshSweep(egressRouteRefreshSem, ebpfState.egressShardSIDs) + // route happened to arrive. In hashed mode it also re-resolves + // the cluster's shard group; see egressGroup. + // + // Off this goroutine, on egressGroup's own: an entry whose next + // hop has no neighbor costs a solicit plus a poll, so a node that + // has lost its fabric uplink would otherwise hold the select loop + // past the next tick and starve the credential refresh, GC + // sweeps and health check with it. + ebpfState.egressGroup.kick() case failure := <-radvActors.failed: radvActorFailed(radvActors, failure) diff --git a/internal/plumbing/ebpf/metrics/egresscontrol.go b/internal/plumbing/ebpf/metrics/egresscontrol.go new file mode 100644 index 00000000..9985a325 --- /dev/null +++ b/internal/plumbing/ebpf/metrics/egresscontrol.go @@ -0,0 +1,117 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package metrics + +import ( + "time" + + "github.com/prometheus/client_golang/prometheus" +) + +// egressSubsystem names galactic-cni's egress control-loop series. +const egressSubsystem = "egress" + +// EgressControl is galactic-cni's own view of the egress shard groups: which +// mode it was configured for, whether its EgressShard watch has ever listed +// the cluster, how many routes still name a group, and whether its sweeps are +// failing. These come from the control loop rather than from the maps, so a +// node that was told to run hashed mode but never got as far as publishing a +// group still says so: the absence of the map-derived series is not health. +// +// All methods are safe on a nil receiver, which tests use. +type EgressControl struct { + mode *prometheus.GaugeVec + watchSynced prometheus.Gauge + groupRoutes *prometheus.GaugeVec + sweepErrors *prometheus.CounterVec + lastSweep prometheus.Gauge +} + +// NewEgressControl builds the series and registers them with reg. +func NewEgressControl(reg prometheus.Registerer) *EgressControl { + c := &EgressControl{ + mode: prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Namespace: egressNamespace, Subsystem: egressSubsystem, Name: "mode", + Help: "1 for the egress mode galactic-cni is configured with (GALACTIC_CNI_EGRESS_MODE), 0 for the other.", + }, []string{"mode"}), + watchSynced: prometheus.NewGauge(prometheus.GaugeOpts{ + Namespace: egressNamespace, Subsystem: egressSubsystem, Name: "shard_watch_synced", + Help: "1 once galactic-cni's EgressShard watch has listed the cluster, else 0. Hashed mode keeps the " + + "shard groups it last published until then.", + }), + groupRoutes: prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Namespace: egressNamespace, Subsystem: egressSubsystem, Name: "group_routes", + Help: "egress_route_table entries naming an egress shard group after the last successful sweep, by " + + "translation class. Must be 0 before rolling back to a release that predates shard groups.", + }, []string{labelClass}), + sweepErrors: prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: egressNamespace, Subsystem: egressSubsystem, Name: "sweep_errors_total", + Help: "Egress route sweep failures, by stage: apply (publishing a shard group), refresh (reading or " + + "rewriting egress_route_table), watch (listing EgressShards), close and disable (retiring groups).", + }, []string{"stage"}), + lastSweep: prometheus.NewGauge(prometheus.GaugeOpts{ + Namespace: egressNamespace, Subsystem: egressSubsystem, Name: "last_successful_sweep_timestamp_seconds", + Help: "Unix time of the last egress route sweep that finished without an error.", + }), + } + reg.MustRegister(c.mode, c.watchSynced, c.groupRoutes, c.sweepErrors, c.lastSweep) + for _, stage := range []string{"apply", "refresh", "watch", "close", "disable"} { + c.sweepErrors.WithLabelValues(stage) + } + return c +} + +// SetMode records the configured mode. +func (c *EgressControl) SetMode(mode string, all ...string) { + if c == nil { + return + } + for _, m := range all { + v := 0.0 + if m == mode { + v = 1 + } + c.mode.WithLabelValues(m).Set(v) + } +} + +// SetWatchSynced records whether the EgressShard watch has synced. +func (c *EgressControl) SetWatchSynced(synced bool) { + if c == nil { + return + } + v := 0.0 + if synced { + v = 1 + } + c.watchSynced.Set(v) +} + +// SetGroupRoutes records the routes naming each class's group. +func (c *EgressControl) SetGroupRoutes(byClass map[string]int) { + if c == nil { + return + } + c.groupRoutes.Reset() + for class, n := range byClass { + c.groupRoutes.WithLabelValues(class).Set(float64(n)) + } +} + +// SweepError counts a failure at stage. +func (c *EgressControl) SweepError(stage string) { + if c == nil { + return + } + c.sweepErrors.WithLabelValues(stage).Inc() +} + +// SweepSucceeded records a sweep that finished without an error. +func (c *EgressControl) SweepSucceeded(at time.Time) { + if c == nil { + return + } + c.lastSweep.Set(float64(at.Unix())) +} diff --git a/internal/plumbing/ebpf/metrics/egressshard.go b/internal/plumbing/ebpf/metrics/egressshard.go new file mode 100644 index 00000000..295429c5 --- /dev/null +++ b/internal/plumbing/ebpf/metrics/egressshard.go @@ -0,0 +1,197 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package metrics + +import ( + "fmt" + + "github.com/prometheus/client_golang/prometheus" + + "go.datum.net/galactic/internal/plumbing/ebpf/egressroutemap" + "go.datum.net/galactic/internal/plumbing/ebpf/prog" +) + +// egressNamespace prefixes the hashed egress series. They describe the +// galactic-cni installer's view of the cluster's egress shard pool, which is +// what the pool alert reads. +const egressNamespace = "galactic_cni" + +// labelClass is the Prometheus label name for an egress translation class. +const labelClass = "class" + +// The pool member states galactic_cni_egress_pool_members reports. +const ( + poolStateActive = "active" + poolStateDraining = "draining" + poolStateUnreachable = "unreachable" + poolStateIneligible = "ineligible" +) + +var ( + egressShardPacketsDesc = prometheus.NewDesc( + prometheus.BuildFQName(egressNamespace, "egress_shard", "packets_total"), + "Tenant packets this node encapsulated toward each egress shard, by the translation class whose group "+ + "placed them, counted since the shard took its slot.", + []string{"shard_sid", labelClass}, nil, + ) + egressShardBytesDesc = prometheus.NewDesc( + prometheus.BuildFQName(egressNamespace, "egress_shard", "bytes_total"), + "Tenant bytes this node encapsulated toward each egress shard, by the translation class whose group "+ + "placed them, counted since the shard took its slot.", + []string{"shard_sid", labelClass}, nil, + ) + egressShardSelectionsDesc = prometheus.NewDesc( + prometheus.BuildFQName(egressNamespace, "egress_shard", "selections_total"), + "How usid_egress placed each packet whose route names a shard group (egress_shard_stats map). "+ + "pin_hit and maglev forwarded it; pin_stale re-placed it; group_empty, shard_dead and "+ + "parse_failed dropped it.", + []string{labelResult}, nil, + ) + egressPoolMembersDesc = prometheus.NewDesc( + prometheus.BuildFQName(egressNamespace, "egress_pool", "members"), + "The cluster's egress shards as this node's group for each translation class sees them: active shards "+ + "take new tenant addresses, draining ones serve only the tenants pinned to them, unreachable ones "+ + "have no route from this node, and ineligible ones cannot translate the class (no address in its "+ + "family, or another NAT64 prefix).", + []string{labelClass, "state"}, nil, + ) + egressPoolMinActiveDesc = prometheus.NewDesc( + prometheus.BuildFQName(egressNamespace, "egress_pool", "min_active"), + "The fewest active egress shards each class is expected to have (GALACTIC_CNI_EGRESS_POOL_MIN_ACTIVE). "+ + "The pool alert compares galactic_cni_egress_pool_members{state=\"active\"} against it.", + nil, nil, + ) + egressPoolEnabledDesc = prometheus.NewDesc( + prometheus.BuildFQName(egressNamespace, "egress_pool", "enabled"), + "1 while the class's shard group takes new routes (hashed mode), else 0. A group being retired for the "+ + "move back to ordered mode still forwards but reads 0.", + []string{labelClass}, nil, + ) +) + +// EgressShardCollector reads the hashed egress maps at every scrape: each +// class's shard group, each shard slot's packet and byte counters, and how +// usid_egress placed packets. +type EgressShardCollector struct { + groups *egressroutemap.ShardGroupTable + counters DropReasonsReader + stats DropReasonsReader + minActive int +} + +// NewEgressShardCollector builds a collector from already-constructed readers, +// so tests can pass fakes. +func NewEgressShardCollector( + groups *egressroutemap.ShardGroupTable, counters, stats DropReasonsReader, minActive int, +) *EgressShardCollector { + return &EgressShardCollector{groups: groups, counters: counters, stats: stats, minActive: minActive} +} + +// RegisterEgressShardCollector registers an EgressShardCollector reading objs' +// maps, with minActive as the pool's expected floor. +func (m *Metrics) RegisterEgressShardCollector(objs *prog.UsidObjects, minActive int) error { + groups := egressroutemap.NewShardGroupTable(egressroutemap.NewKernelGroupStore(objs.EgressShardGroups)) + return m.Registry.Register( + NewEgressShardCollector(groups, objs.EgressShardCounters, objs.EgressShardStats, minActive)) +} + +// Describe implements prometheus.Collector. +func (c *EgressShardCollector) Describe(ch chan<- *prometheus.Desc) { + ch <- egressShardPacketsDesc + ch <- egressShardBytesDesc + ch <- egressShardSelectionsDesc + ch <- egressPoolMembersDesc + ch <- egressPoolMinActiveDesc + ch <- egressPoolEnabledDesc +} + +// Collect implements prometheus.Collector. A group that has never been +// published reports nothing; galactic_cni_egress_mode says whether one was +// expected. +func (c *EgressShardCollector) Collect(ch chan<- prometheus.Metric) { + ch <- prometheus.MustNewConstMetric(egressPoolMinActiveDesc, prometheus.GaugeValue, float64(c.minActive)) + c.collectSelections(ch) + + for group := range uint32(egressroutemap.MaxShardGroups) { + status, err := c.groups.Status(group) + if err != nil { + ch <- prometheus.NewInvalidMetric(egressPoolEnabledDesc, err) + continue + } + if !status.Present || (!status.Enabled && status.Candidates == 0 && status.Ineligible == 0) { + continue + } + class := status.Class.String() + enabled := 0.0 + if status.Open() { + enabled = 1 + } + ch <- prometheus.MustNewConstMetric(egressPoolEnabledDesc, prometheus.GaugeValue, enabled, class) + + members, err := c.groups.Members(group) + if err != nil { + ch <- prometheus.NewInvalidMetric(egressPoolMembersDesc, err) + continue + } + states := map[string]int{ + poolStateActive: 0, poolStateDraining: 0, poolStateUnreachable: 0, + poolStateIneligible: status.Ineligible, + } + for _, member := range members { + switch { + case !member.Alive: + states[poolStateUnreachable]++ + case member.Draining: + states[poolStateDraining]++ + default: + states[poolStateActive]++ + } + c.collectShardCounters(ch, group, class, member) + } + for state, n := range states { + ch <- prometheus.MustNewConstMetric(egressPoolMembersDesc, prometheus.GaugeValue, float64(n), class, state) + } + } +} + +func (c *EgressShardCollector) collectShardCounters( + ch chan<- prometheus.Metric, group uint32, class string, member egressroutemap.ShardState, +) { + key := group*egressroutemap.MaxShardsPerGroup + uint32(member.Slot) //nolint:gosec // slot < 32 + var perCPU []prog.UsidEgressShardCounter + if err := c.counters.Lookup(key, &perCPU); err != nil { + ch <- prometheus.NewInvalidMetric(egressShardPacketsDesc, + fmt.Errorf("lookup egress_shard_counters[%d]: %w", key, err)) + return + } + var packets, bytes uint64 + for _, v := range perCPU { + packets += v.Packets + bytes += v.Bytes + } + sid := member.SID.String() + ch <- prometheus.MustNewConstMetric(egressShardPacketsDesc, prometheus.CounterValue, float64(packets), sid, class) + ch <- prometheus.MustNewConstMetric(egressShardBytesDesc, prometheus.CounterValue, float64(bytes), sid, class) +} + +func (c *EgressShardCollector) collectSelections(ch chan<- prometheus.Metric) { + for i := range prog.EgressShardStatCount { + var perCPU []uint64 + if err := c.stats.Lookup(i, &perCPU); err != nil { + ch <- prometheus.NewInvalidMetric(egressShardSelectionsDesc, + fmt.Errorf("lookup egress_shard_stats[%d]: %w", i, err)) + continue + } + var total uint64 + for _, v := range perCPU { + total += v + } + name := prog.EgressShardStatNames[i] + if name == "" { + name = fmt.Sprintf("unknown_%d", i) + } + ch <- prometheus.MustNewConstMetric(egressShardSelectionsDesc, prometheus.CounterValue, float64(total), name) + } +} diff --git a/internal/plumbing/ebpf/metrics/egressshard_test.go b/internal/plumbing/ebpf/metrics/egressshard_test.go new file mode 100644 index 00000000..54334c59 --- /dev/null +++ b/internal/plumbing/ebpf/metrics/egressshard_test.go @@ -0,0 +1,201 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package metrics + +import ( + "net/netip" + "reflect" + "strings" + "testing" + + "github.com/prometheus/client_golang/prometheus" + dto "github.com/prometheus/client_model/go" + + "go.datum.net/galactic/internal/plumbing/ebpf/egressroutemap" + "go.datum.net/galactic/internal/plumbing/ebpf/prog" +) + +// snapshotStore is a GroupStore over fixed snapshots. +type snapshotStore map[uint32]prog.UsidEgressShardGroupValue + +func (s snapshotStore) Load(id uint32) (prog.UsidEgressShardGroupValue, bool, error) { + v, ok := s[id] + return v, ok, nil +} + +func (s snapshotStore) Publish(id uint32, v prog.UsidEgressShardGroupValue) error { + s[id] = v + return nil +} + +// perCPUTable answers a per-CPU lookup with a fixed slice per key. +type perCPUTable map[uint32]any + +func (p perCPUTable) Lookup(key, out any) error { + v, ok := p[key.(uint32)] + if !ok { + v = reflect.MakeSlice(reflect.TypeOf(out).Elem(), 2, 2).Interface() + } + reflect.ValueOf(out).Elem().Set(reflect.ValueOf(v)) + return nil +} + +func sidBytes(s string) [16]byte { return netip.MustParseAddr(s).As16() } + +func TestEgressShardCollector(t *testing.T) { + nat66 := prog.UsidEgressShardGroupValue{Flags: 1, Candidates: 3, Active: 1, ClassKind: 1} + nat66.Shards[0] = prog.UsidEgressShardMember{Sid: sidBytes("2001:db8:ff09:10:e000::"), Flags: 1, Generation: 1} + nat66.Shards[1] = prog.UsidEgressShardMember{Sid: sidBytes("2001:db8:ff09:11:e000::"), Flags: 3, Generation: 2} + nat66.Shards[4] = prog.UsidEgressShardMember{Sid: sidBytes("2001:db8:ff09:12:e000::"), Flags: 0, Generation: 3} + // A NAT64 class no shard can serve, being retired: it forwards but is + // not open. + nat64 := prog.UsidEgressShardGroupValue{Flags: 1 | 4, Ineligible: 3, ClassKind: 2, + ClassPrefix: sidBytes("2001:db8:64::")} + store := snapshotStore{0: nat66, 1: nat64, 2: {ClassKind: 2}} + counters := perCPUTable{ + uint32(0): []prog.UsidEgressShardCounter{{Packets: 3, Bytes: 300}, {Packets: 4, Bytes: 400}}, + } + stats := perCPUTable{ + prog.EgressShardStatMaglev: []uint64{5, 6}, + prog.EgressShardStatGroupEmpty: []uint64{1, 0}, + } + c := NewEgressShardCollector(egressroutemap.NewShardGroupTable(store), counters, stats, 2) + + got := map[string]float64{} + for _, m := range collectNamed(t, c) { + got[m.name] = m.value + } + const nat64Class = `class="nat64:2001:db8:64::/96"` + want := map[string]float64{ + `galactic_cni_egress_pool_min_active`: 2, + `galactic_cni_egress_pool_enabled{class="nat66"}`: 1, + `galactic_cni_egress_pool_enabled{` + nat64Class + `}`: 0, + `galactic_cni_egress_pool_members{class="nat66",state="active"}`: 1, + `galactic_cni_egress_pool_members{class="nat66",state="draining"}`: 1, + `galactic_cni_egress_pool_members{class="nat66",state="unreachable"}`: 1, + `galactic_cni_egress_pool_members{class="nat66",state="ineligible"}`: 0, + `galactic_cni_egress_pool_members{` + nat64Class + `,state="active"}`: 0, + `galactic_cni_egress_pool_members{` + nat64Class + `,state="ineligible"}`: 3, + `galactic_cni_egress_shard_packets_total{class="nat66",shard_sid="2001:db8:ff09:10:e000::"}`: 7, + `galactic_cni_egress_shard_bytes_total{class="nat66",shard_sid="2001:db8:ff09:10:e000::"}`: 700, + `galactic_cni_egress_shard_packets_total{class="nat66",shard_sid="2001:db8:ff09:12:e000::"}`: 0, + `galactic_cni_egress_shard_selections_total{result="maglev"}`: 11, + `galactic_cni_egress_shard_selections_total{result="group_empty"}`: 1, + `galactic_cni_egress_shard_selections_total{result="pin_hit"}`: 0, + } + for name, value := range want { + if v, ok := got[name]; !ok || v != value { + t.Errorf("%s = %v (present %v), want %v", name, v, ok, value) + } + } + for name := range got { + if strings.Contains(name, `class="nat64:::/96"`) { + t.Errorf("an empty, disabled group was reported: %s", name) + } + } +} + +// TestEgressShardCollectorReportsNothingUnpublished checks that a node with no +// published group reports no pool series at all, rather than zeros that read +// as an empty but healthy pool. +func TestEgressShardCollectorReportsNothingUnpublished(t *testing.T) { + c := NewEgressShardCollector(egressroutemap.NewShardGroupTable(snapshotStore{}), perCPUTable{}, perCPUTable{}, 1) + for _, m := range collectNamed(t, c) { + if strings.HasPrefix(m.name, "galactic_cni_egress_pool_members") || + strings.HasPrefix(m.name, "galactic_cni_egress_pool_enabled") { + t.Errorf("unpublished node reported %s", m.name) + } + } +} + +func TestEgressControlSeries(t *testing.T) { + reg := prometheus.NewRegistry() + c := NewEgressControl(reg) + c.SetMode("hashed", "ordered", "hashed") + c.SetWatchSynced(false) + c.SetGroupRoutes(map[string]int{"nat66": 4}) + c.SweepError("refresh") + got := map[string]float64{} + for _, m := range collectNamed(t, registryCollector{reg}) { + got[m.name] = m.value + } + for name, value := range map[string]float64{ + `galactic_cni_egress_mode{mode="hashed"}`: 1, + `galactic_cni_egress_mode{mode="ordered"}`: 0, + `galactic_cni_egress_shard_watch_synced`: 0, + `galactic_cni_egress_group_routes{class="nat66"}`: 4, + `galactic_cni_egress_sweep_errors_total{stage="refresh"}`: 1, + `galactic_cni_egress_sweep_errors_total{stage="apply"}`: 0, + } { + if v, ok := got[name]; !ok || v != value { + t.Errorf("%s = %v (present %v), want %v", name, v, ok, value) + } + } + var nilControl *EgressControl + nilControl.SweepError("apply") // must not panic +} + +// registryCollector adapts a registry to prometheus.Collector for collectNamed. +type registryCollector struct{ reg *prometheus.Registry } + +func (r registryCollector) Describe(chan<- *prometheus.Desc) {} + +func (r registryCollector) Collect(ch chan<- prometheus.Metric) { + families, _ := r.reg.Gather() + for _, f := range families { + for _, m := range f.GetMetric() { + labels := map[string]string{} + names := make([]string, 0, len(m.GetLabel())) + for _, lp := range m.GetLabel() { + labels[lp.GetName()] = lp.GetValue() + names = append(names, lp.GetName()) + } + desc := prometheus.NewDesc(f.GetName(), f.GetHelp(), names, nil) + values := make([]string, 0, len(names)) + for _, n := range names { + values = append(values, labels[n]) + } + v := m.GetGauge().GetValue() + if m.GetCounter() != nil { + v = m.GetCounter().GetValue() + } + ch <- prometheus.MustNewConstMetric(desc, prometheus.UntypedValue, v, values...) + } + } +} + +type namedMetric struct { + name string + value float64 +} + +// collectNamed collects c and names each metric as name{label="value",...}. +func collectNamed(t *testing.T, c prometheus.Collector) []namedMetric { + t.Helper() + ch := make(chan prometheus.Metric, 256) + go func() { + c.Collect(ch) + close(ch) + }() + var out []namedMetric + for m := range ch { + var pb dto.Metric + if err := m.Write(&pb); err != nil { + t.Fatalf("write metric: %v", err) + } + desc := m.Desc().String() + start := strings.Index(desc, `fqName: "`) + len(`fqName: "`) + name := desc[start : start+strings.Index(desc[start:], `"`)] + labels := make([]string, 0, len(pb.GetLabel())) + for _, lp := range pb.GetLabel() { + labels = append(labels, lp.GetName()+`="`+lp.GetValue()+`"`) + } + if len(labels) > 0 { + name += "{" + strings.Join(labels, ",") + "}" + } + out = append(out, namedMetric{name: name, value: metricValue(&pb)}) + } + return out +} diff --git a/internal/plumbing/ebpf/metrics/metrics.go b/internal/plumbing/ebpf/metrics/metrics.go index 299f50fc..873b6a22 100644 --- a/internal/plumbing/ebpf/metrics/metrics.go +++ b/internal/plumbing/ebpf/metrics/metrics.go @@ -21,6 +21,8 @@ import ( type Metrics struct { Registry *prometheus.Registry Events *EventCounters + // Egress is galactic-cni's egress shard group control-loop state. + Egress *EgressControl } // New builds a Metrics with the event counters already registered. Call @@ -30,7 +32,7 @@ func New() *Metrics { reg := prometheus.NewRegistry() events := NewEventCounters() events.MustRegister(reg) - return &Metrics{Registry: reg, Events: events} + return &Metrics{Registry: reg, Events: events, Egress: NewEgressControl(reg)} } // RegisterDatapathCollector registers a collector reading live map state from