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