From 5fdf9839e81aaf17caedfa9c92bdab8f43c16af0 Mon Sep 17 00:00:00 2001 From: ryskn Date: Sun, 7 Jun 2026 23:00:35 +0900 Subject: [PATCH] feat(srv6): SRv6-native NAT-less (DSR) ClusterIP services Replace cnat DNAT/un-DNAT with Direct Server Return for pod-backed ClusterIP services in the SRv6 data path. Traffic keeps dst=VIP end to end; backend pods accept the VIP on lo and reply from it, so no reverse translation is needed and the cross-node SRv6 un-DNAT failure disappears. Source IP is preserved (no SNAT). - config: SRv6NativeServicesEnabled feature gate + vppSRv6Native annotation - services: classify eligible ClusterIP services (gate + annotation + ClusterIP + port==targetPort), skip cnat, publish DSRService events, reconcile-based diff/withdraw - connectivity/SRv6: per-service ECMP SR policy over backend nodes' End.DT6 SIDs + steering; decap into PodVRF via sw_if_index; reconcile with retain-on-failure retry and periodic re-reconcile - cni: bind VIP on local backend pods (lo) + uRPF allow + PodVRF ECMP delivery route; per-pod cleanup tracking; reconcile on pod add/del Experimental; requires SRv6Enabled. Other services (NodePort/External, port-remapped, host-network backed) keep the cnat path. Signed-off-by: Ryosuke Nakayama --- calico-vpp-agent/cmd/calico_vpp_dataplane.go | 2 +- calico-vpp-agent/cni/cni_dsr.go | 303 ++++++++++++++++++ calico-vpp-agent/cni/cni_server.go | 38 +++ calico-vpp-agent/common/dsr.go | 37 +++ calico-vpp-agent/common/pubsub.go | 7 + .../connectivity/connectivity_server.go | 28 ++ calico-vpp-agent/connectivity/srv6.go | 77 +++-- calico-vpp-agent/connectivity/srv6_dsr.go | 272 ++++++++++++++++ calico-vpp-agent/services/service.go | 3 + calico-vpp-agent/services/service_dsr.go | 177 ++++++++++ calico-vpp-agent/services/service_handler.go | 9 +- calico-vpp-agent/services/service_server.go | 20 +- calico-vpp-agent/services/services_test.go | 2 +- config/config.go | 11 + 14 files changed, 948 insertions(+), 38 deletions(-) create mode 100644 calico-vpp-agent/cni/cni_dsr.go create mode 100644 calico-vpp-agent/common/dsr.go create mode 100644 calico-vpp-agent/connectivity/srv6_dsr.go create mode 100644 calico-vpp-agent/services/service_dsr.go diff --git a/calico-vpp-agent/cmd/calico_vpp_dataplane.go b/calico-vpp-agent/cmd/calico_vpp_dataplane.go index aa6170f77..5fe6ff441 100644 --- a/calico-vpp-agent/cmd/calico_vpp_dataplane.go +++ b/calico-vpp-agent/cmd/calico_vpp_dataplane.go @@ -150,7 +150,6 @@ func main() { bgpFilterWatcher := watchers.NewBGPFilterWatcher(clientv3, k8sclient, log.WithFields(logrus.Fields{"subcomponent": "BGPFilter-watcher"})) netWatcher := watchers.NewNetWatcher(vpp, log.WithFields(logrus.Fields{"component": "net-watcher"})) routingServer := routing.NewRoutingServer(vpp, bgpServer, log.WithFields(logrus.Fields{"component": "routing"})) - serviceServer := services.NewServiceServer(vpp, k8sclient, log.WithFields(logrus.Fields{"component": "services"})) localSIDWatcher := watchers.NewLocalSIDWatcher(vpp, clientv3, log.WithFields(logrus.Fields{"subcomponent": "localsid-watcher"})) felixServer, err := felix.NewFelixServer(vpp, log.WithFields(logrus.Fields{"component": "felix"})) if err != nil { @@ -160,6 +159,7 @@ func main() { if err != nil { log.Fatalf("could not install felix plugin: %s", err) } + serviceServer := services.NewServiceServer(vpp, k8sclient, felixServer, log.WithFields(logrus.Fields{"component": "services"})) connectivityServer := connectivity.NewConnectivityServer(vpp, felixServer, clientv3, log.WithFields(logrus.Fields{"subcomponent": "connectivity"})) cniServer := cni.NewCNIServer(vpp, felixServer, log.WithFields(logrus.Fields{"component": "cni"})) diff --git a/calico-vpp-agent/cni/cni_dsr.go b/calico-vpp-agent/cni/cni_dsr.go new file mode 100644 index 000000000..abe8133ff --- /dev/null +++ b/calico-vpp-agent/cni/cni_dsr.go @@ -0,0 +1,303 @@ +package cni + +import ( + "net" + + "github.com/containernetworking/plugins/pkg/ns" + "github.com/vishvananda/netlink" + + "github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/cni/model" + "github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/common" + "github.com/projectcalico/vpp-dataplane/v3/vpplink" + "github.com/projectcalico/vpp-dataplane/v3/vpplink/types" +) + +// dsrVIPState tracks, for a DSR ClusterIP with backend pods local to this node, +// the shared PodVRFIndex delivery route and the set of local pods that currently +// have the VIP programmed (bound on lo + allowed through uRPF), keyed by pod IP. +// boundPods is authoritative for cleanup: a pod stays in it until its VIP +// binding has been successfully removed (or the pod is gone), so per-pod cleanup +// failures are retried on the next reconcile rather than forgotten. +type dsrVIPState struct { + vip net.IP + delivery *types.Route + boundPods map[string]bool +} + +func (s *Server) handleDSRServiceAdded(svc *common.DSRService) { + if svc == nil || svc.VIP == nil { + return + } + s.lock.Lock() + defer s.lock.Unlock() + s.dsrDesired[svc.VIP.String()] = svc + s.reconcileDSRVIPs() +} + +func (s *Server) handleDSRServiceDeleted(svc *common.DSRService) { + if svc == nil || svc.VIP == nil { + return + } + s.lock.Lock() + defer s.lock.Unlock() + delete(s.dsrDesired, svc.VIP.String()) + s.reconcileDSRVIPs() +} + +// reconcileDSRVIPs drives installed DSR VIP state toward the desired set, +// retrying failed installs and removals (both the delivery route and per-pod +// bindings). It is invoked on every DSR event, the services server's periodic +// re-publish, and pod add/del — so a transient failure is retried rather than +// recorded as reconciled. Caller holds s.lock. +func (s *Server) reconcileDSRVIPs() { + for _, svc := range s.dsrDesired { + s.programDSRVIP(svc) + } + for key := range s.dsrVIPs { + if _, ok := s.dsrDesired[key]; !ok { + s.teardownDSRVIP(key) + } + } +} + +// programDSRVIP binds the VIP on this node's local backend pods (so their kernel +// accepts dst==VIP and the reply is sourced from the VIP, allowed through uRPF), +// unbinds pods no longer backing the service (retaining ones whose cleanup +// fails), and reconciles the PodVRFIndex ECMP delivery route. No NAT is +// involved. Caller holds s.lock. +func (s *Server) programDSRVIP(svc *common.DSRService) { + vipKey := svc.VIP.String() + prev := s.dsrVIPs[vipKey] + boundPods := map[string]bool{} + if prev != nil { + for ip := range prev.boundPods { + boundPods[ip] = true + } + } + + desiredLocal := map[string]bool{} + var deliveryPaths []types.RoutePath + for _, beIP := range svc.Backends { + podSpec := s.findLocalPodByIP(beIP) + if podSpec == nil || podSpec.TunTapSwIfIndex == vpplink.InvalidSwIfIndex { + continue // not a local tun-backed pod + } + desiredLocal[beIP.String()] = true + if err := s.bindVIPInPod(podSpec, svc.VIP, true /* add */); err != nil { + s.log.Errorf("cni(dsr) bind vip=%s pod=%s: %v", svc.VIP, beIP, err) + } + if err := s.dsrRPFRoute(podSpec, svc.VIP, true /* add */); err != nil { + s.log.Errorf("cni(dsr) rpf add vip=%s pod=%s: %v", svc.VIP, beIP, err) + } + boundPods[beIP.String()] = true + deliveryPaths = append(deliveryPaths, types.RoutePath{ + SwIfIndex: podSpec.TunTapSwIfIndex, + Gw: beIP, + }) + } + + // Unbind pods no longer backing the service; keep ones whose cleanup fails so + // a departed-but-running pod can't keep answering as / uRPF-spoofing the VIP. + for ipStr := range boundPods { + if desiredLocal[ipStr] { + continue + } + if s.cleanupDSRPod(ipStr, svc.VIP) { + delete(boundPods, ipStr) + } + } + + st := &dsrVIPState{vip: svc.VIP, boundPods: boundPods} + st.delivery = s.updateDeliveryRoute(svc.VIP, prev, deliveryPaths) + // Drop the entry only once the route is gone AND no pod cleanup is pending. + if st.delivery == nil && len(boundPods) == 0 { + delete(s.dsrVIPs, vipKey) + return + } + s.dsrVIPs[vipKey] = st + s.log.Debugf("cni(dsr) vip=%s local backends=%d bound=%d", svc.VIP, len(desiredLocal), len(boundPods)) +} + +// teardownDSRVIP removes all DSR state for a VIP no longer desired, retaining +// (for retry on the next reconcile) the delivery route and any pod whose cleanup +// fails. Caller holds s.lock. +func (s *Server) teardownDSRVIP(key string) { + st := s.dsrVIPs[key] + if st == nil { + return + } + st.delivery = s.updateDeliveryRoute(st.vip, st, nil /* no paths */) + for ipStr := range st.boundPods { + if s.cleanupDSRPod(ipStr, st.vip) { + delete(st.boundPods, ipStr) + } + } + if st.delivery == nil && len(st.boundPods) == 0 { + delete(s.dsrVIPs, key) + } +} + +// cleanupDSRPod removes the VIP binding (lo) and uRPF allow from a pod that no +// longer backs the service. Returns true when cleanup is complete — the pod is +// gone (its netns + per-pod RPF VRF auto-cleaned) or both removals succeeded — +// and false when it must be retried. +func (s *Server) cleanupDSRPod(ipStr string, vip net.IP) bool { + ip := net.ParseIP(ipStr) + if ip == nil { + return true + } + podSpec := s.findLocalPodByIP(ip) + if podSpec == nil { + return true // pod gone; netns + per-pod RPF VRF auto-cleaned + } + ok := true + if err := s.bindVIPInPod(podSpec, vip, false /* del */); err != nil { + s.log.Errorf("cni(dsr) unbind vip=%s pod=%s: %v", vip, ip, err) + ok = false + } + if err := s.dsrRPFRoute(podSpec, vip, false /* del */); err != nil { + s.log.Errorf("cni(dsr) rpf del vip=%s pod=%s: %v", vip, ip, err) + ok = false + } + return ok +} + +// updateDeliveryRoute reconciles the PodVRFIndex delivery route for a VIP to the +// given paths, retaining state on VPP failure so the next reconcile retries: +// - unchanged paths: keep the existing route (no blackhole churn); +// - changed paths: delete the old (on failure keep it and retry later), then +// add the new (on failure return nil so the add is retried); +// - no paths: delete the old (retain on failure). +func (s *Server) updateDeliveryRoute(vip net.IP, prev *dsrVIPState, paths []types.RoutePath) *types.Route { + if prev != nil && prev.delivery != nil { + if len(paths) > 0 && samePaths(prev.delivery.Paths, paths) { + return prev.delivery + } + if err := s.vpp.RouteDel(prev.delivery); err != nil { + s.log.Errorf("cni(dsr) del delivery route vip=%s: %v", vip, err) + return prev.delivery // keep old; retry replacement next reconcile + } + } + if len(paths) == 0 { + return nil + } + route := &types.Route{ + Dst: common.ToMaxLenCIDR(vip), + Paths: paths, + Table: common.PodVRFIndex, + } + if err := s.vpp.RouteAdd(route); err != nil { + s.log.Errorf("cni(dsr) add delivery route vip=%s: %v", vip, err) + return nil // retry add next reconcile + } + return route +} + +// samePaths reports whether two route path sets are equal (order-independent). +func samePaths(a, b []types.RoutePath) bool { + if len(a) != len(b) { + return false + } + type key struct { + idx uint32 + gw string + } + m := make(map[key]int) + for _, p := range a { + m[key{p.SwIfIndex, p.Gw.String()}]++ + } + for _, p := range b { + m[key{p.SwIfIndex, p.Gw.String()}]-- + } + for _, v := range m { + if v != 0 { + return false + } + } + return true +} + +// findLocalPodByIP returns the local pod spec whose container IP equals ip. +func (s *Server) findLocalPodByIP(ip net.IP) *model.LocalPodSpec { + for key := range s.podInterfaceMap { + spec := s.podInterfaceMap[key] + for _, cip := range spec.GetContainerIPs() { + if cip.IP.Equal(ip) { + return &spec + } + } + } + return nil +} + +// bindVIPInPod adds or removes the VIP on the pod's loopback inside its netns. +// It lists the current addresses first so add is a no-op when already present +// and del is a no-op when already absent — neither reports a spurious failure. +func (s *Server) bindVIPInPod(podSpec *model.LocalPodSpec, vip net.IP, add bool) error { + if podSpec.NetnsName == "" { + return nil + } + return ns.WithNetNSPath(podSpec.NetnsName, func(ns.NetNS) error { + lo, err := netlink.LinkByName("lo") + if err != nil { + return err + } + addrs, err := netlink.AddrList(lo, netlink.FAMILY_V6) + if err != nil { + return err + } + present := false + for _, a := range addrs { + if a.IP.Equal(vip) { + present = true + break + } + } + addr := &netlink.Addr{IPNet: common.ToMaxLenCIDR(vip)} + if add { + if present { + return nil + } + return netlink.AddrAdd(lo, addr) + } + if present { + return netlink.AddrDel(lo, addr) + } + return nil + }) +} + +// dsrRPFRoute adds or removes a route for the VIP in the pod's RPF VRF so the +// loose uRPF check passes (add) / no longer passes (del) a reply sourced from +// the VIP. VPP's route delete is idempotent for an absent route. +func (s *Server) dsrRPFRoute(podSpec *model.LocalPodSpec, vip net.IP, add bool) error { + vipNet := common.ToMaxLenCIDR(vip) + rpfVrfID := podSpec.GetRPFVrfID(vpplink.IPFamilyFromIPNet(vipNet)) + if rpfVrfID == vpplink.InvalidID { + return nil + } + gw := podContainerIPForVIP(podSpec, vip) + if gw == nil { + return nil + } + paths := []types.RoutePath{{SwIfIndex: podSpec.TunTapSwIfIndex, Gw: gw}} + if podSpec.MemifSwIfIndex != vpplink.InvalidSwIfIndex { + paths = append(paths, types.RoutePath{SwIfIndex: podSpec.MemifSwIfIndex, Gw: gw}) + } + route := &types.Route{Dst: vipNet, Paths: paths, Table: rpfVrfID} + if add { + return s.vpp.RouteAdd(route) + } + return s.vpp.RouteDel(route) +} + +// podContainerIPForVIP returns the pod's container IP of the same family as vip. +func podContainerIPForVIP(podSpec *model.LocalPodSpec, vip net.IP) net.IP { + wantV6 := vip.To4() == nil + for _, cip := range podSpec.GetContainerIPs() { + if (cip.IP.To4() == nil) == wantV6 { + return cip.IP + } + } + return nil +} diff --git a/calico-vpp-agent/cni/cni_server.go b/calico-vpp-agent/cni/cni_server.go index 5b1bcc778..e5612b1a4 100644 --- a/calico-vpp-agent/cni/cni_server.go +++ b/calico-vpp-agent/cni/cni_server.go @@ -23,6 +23,7 @@ import ( "os" "sync" "syscall" + "time" "github.com/pkg/errors" calicov3 "github.com/projectcalico/api/pkg/apis/projectcalico/v3" @@ -54,6 +55,12 @@ type Server struct { podInterfaceMap map[string]model.LocalPodSpec lock sync.Mutex /* protects Add/DelVppInterace/RescanState */ cniEventChan chan common.CalicoVppEvent + // dsrVIPs tracks the shared PodVRFIndex delivery route per SRv6-native (DSR) + // ClusterIP VIP for which this node hosts backend pods. + dsrVIPs map[string]*dsrVIPState + // dsrDesired is the desired DSR service set (by VIP), reconciled against + // dsrVIPs so failed installs AND removals are retried, not lost. + dsrDesired map[string]*common.DSRService memifDriver *podinterface.MemifPodInterfaceDriver tuntapDriver *podinterface.TunTapPodInterfaceDriver @@ -147,6 +154,9 @@ func (s *Server) Add(ctx context.Context, request *cniproto.AddRequest) (*cnipro if err != nil { s.log.Errorf("CNI state persist errored %v", err) } + // A newly-added pod may be a backend of a desired DSR service; reconcile so + // its VIP binding + delivery route (and any pending DSR retries) are applied. + s.reconcileDSRVIPs() s.log.Infof("pod(add) Done spec=%s", podSpec.String()) // XXX: container MAC doesn't make sense with tun, we just pass back a constant one. // How does calico / k8s use it? @@ -280,6 +290,10 @@ func (s *Server) Del(ctx context.Context, request *cniproto.DelRequest) (*cnipro s.log.Errorf("CNI state persist errored %v", err) } + // A removed pod may have been a DSR backend; reconcile so the delivery route + // drops its path and any pending DSR delete/install retries are driven. + s.reconcileDSRVIPs() + return &cniproto.DelReply{ Successful: true, }, nil @@ -296,6 +310,8 @@ func NewCNIServer(vpp *vpplink.VppLink, felixServerIpam common.FelixServerIpam, grpcServer: grpc.NewServer(), podInterfaceMap: make(map[string]model.LocalPodSpec), + dsrVIPs: make(map[string]*dsrVIPState), + dsrDesired: make(map[string]*common.DSRService), tuntapDriver: podinterface.NewTunTapPodInterfaceDriver(vpp, log, felixServerIpam), memifDriver: podinterface.NewMemifPodInterfaceDriver(vpp, log, felixServerIpam), vclDriver: podinterface.NewVclPodInterfaceDriver(vpp, log, felixServerIpam), @@ -307,6 +323,8 @@ func NewCNIServer(vpp *vpplink.VppLink, felixServerIpam common.FelixServerIpam, reg.ExpectEvents( common.FelixConfChanged, common.IpamConfChanged, + common.ServiceDSRClusterIPAdded, + common.ServiceDSRClusterIPDeleted, ) regM := common.RegisterHandler(server.cniMultinetEventChan, "CNI server Multinet events") regM.ExpectEvents( @@ -317,11 +335,19 @@ func NewCNIServer(vpp *vpplink.VppLink, felixServerIpam common.FelixServerIpam, return server } func (s *Server) cniServerEventLoop(t *tomb.Tomb) error { + // Periodically reconcile SRv6-native (DSR) VIP state so a withdrawn-service + // cleanup (or install) that failed is retried even without further events. + dsrTicker := time.NewTicker(30 * time.Second) + defer dsrTicker.Stop() forloop: for { select { case <-t.Dying(): break forloop + case <-dsrTicker.C: + s.lock.Lock() + s.reconcileDSRVIPs() + s.lock.Unlock() case evt := <-s.cniEventChan: switch evt.Type { case common.FelixConfChanged: @@ -364,6 +390,18 @@ forloop: s.lock.Lock() s.tuntapDriver.FelixConfigChanged(nil /* felixConfig */, ipipEncapRefCountDelta, vxlanEncapRefCountDelta, s.podInterfaceMap) s.lock.Unlock() + case common.ServiceDSRClusterIPAdded: + if svc, ok := evt.New.(*common.DSRService); ok { + s.handleDSRServiceAdded(svc) + } else { + s.log.Errorf("evt.New is not a *common.DSRService %v", evt.New) + } + case common.ServiceDSRClusterIPDeleted: + if svc, ok := evt.Old.(*common.DSRService); ok { + s.handleDSRServiceDeleted(svc) + } else { + s.log.Errorf("evt.Old is not a *common.DSRService %v", evt.Old) + } } } } diff --git a/calico-vpp-agent/common/dsr.go b/calico-vpp-agent/common/dsr.go new file mode 100644 index 000000000..100f7b95b --- /dev/null +++ b/calico-vpp-agent/common/dsr.go @@ -0,0 +1,37 @@ +package common + +import "net" + +// DSRService describes an SRv6-native / NAT-less (DSR) ClusterIP service. +// +// It is published by the services server (ServiceDSRClusterIPAdded/Deleted) and +// consumed in-process by: +// - the SRv6 connectivity provider, which installs a per-service SR policy +// (weighted ECMP over the backend nodes' End.DT6 SIDs) plus a steering entry +// so that VIP traffic from clients without a local backend is steered to a +// backend node; and +// - the CNI server, which for backends that are local pods binds the VIP +// inside the pod (so its kernel accepts dst==VIP, i.e. DSR), allows VIP as a +// source through uRPF, and installs the pod-VRF delivery route. +// +// There is no NAT anywhere on the path: the inner packet keeps dst==VIP all the +// way to the backend pod, which replies with src==VIP, so no reverse +// translation is needed on the return path. +type DSRService struct { + // VIP is the service ClusterIP. + VIP net.IP + // Backends are the IPs of all pod-backed endpoints of the service (both + // local to this node and remote). Each consumer resolves the ones it cares + // about (CNI: local pods; SRv6: remote backends' nodes). + Backends []net.IP + // ServiceID is namespace/name, for logging only. + ServiceID string +} + +// Key identifies a DSR service by its VIP. +func (d *DSRService) Key() string { + if d == nil || d.VIP == nil { + return "" + } + return d.VIP.String() +} diff --git a/calico-vpp-agent/common/pubsub.go b/calico-vpp-agent/common/pubsub.go index daef558ca..20e94e3ee 100644 --- a/calico-vpp-agent/common/pubsub.go +++ b/calico-vpp-agent/common/pubsub.go @@ -37,6 +37,13 @@ const ( SRv6PolicyAdded CalicoVppEventType = "SRv6PolicyAdded" SRv6PolicyDeleted CalicoVppEventType = "SRv6PolicyDeleted" + // ServiceDSRClusterIP{Added,Deleted} carry a *DSRService describing an + // SRv6-native / NAT-less (DSR) ClusterIP service. Consumed by the SRv6 + // connectivity provider (per-service SR policy + steering) and the CNI + // server (pod VIP bind + uRPF allow + pod-VRF delivery route). + ServiceDSRClusterIPAdded CalicoVppEventType = "ServiceDSRClusterIPAdded" + ServiceDSRClusterIPDeleted CalicoVppEventType = "ServiceDSRClusterIPDeleted" + PodAdded CalicoVppEventType = "PodAdded" PodDeleted CalicoVppEventType = "PodDeleted" diff --git a/calico-vpp-agent/connectivity/connectivity_server.go b/calico-vpp-agent/connectivity/connectivity_server.go index 70e2a034f..fa203354f 100644 --- a/calico-vpp-agent/connectivity/connectivity_server.go +++ b/calico-vpp-agent/connectivity/connectivity_server.go @@ -18,6 +18,7 @@ package connectivity import ( "fmt" "net" + "time" "github.com/pkg/errors" felixConfig "github.com/projectcalico/calico/felix/config" @@ -89,6 +90,8 @@ func NewConnectivityServer(vpp *vpplink.VppLink, felixServerIpam common.FelixSer common.IpamConfChanged, common.SRv6PolicyAdded, common.SRv6PolicyDeleted, + common.ServiceDSRClusterIPAdded, + common.ServiceDSRClusterIPDeleted, common.WireguardPublicKeyChanged, ) @@ -162,11 +165,20 @@ func (s *ConnectivityServer) ServeConnectivity(t *tomb.Tomb) error { for _, provider := range s.providers { provider.RescanState() } + // Periodically reconcile SRv6-native (DSR) services so a withdrawn-service + // cleanup (or install) that failed is retried even when no further events + // arrive (e.g. the last DSR service was deleted in a quiet cluster). + dsrTicker := time.NewTicker(30 * time.Second) + defer dsrTicker.Stop() for { select { case <-t.Dying(): s.log.Warn("Connectivity Server asked to stop") return nil + case <-dsrTicker.C: + if p, ok := s.providers[SRv6].(*SRv6Provider); ok { + p.reconcileDSRServices() + } case evt := <-s.connectivityEventChan: /* Note: we will only receive events we ask for when registering the chan */ switch evt.Type { @@ -288,6 +300,22 @@ func (s *ConnectivityServer) ServeConnectivity(t *tomb.Tomb) error { if err != nil { s.log.Errorf("Error while deleting SRv6 Policy %s", err) } + case common.ServiceDSRClusterIPAdded: + if svc, ok := evt.New.(*common.DSRService); !ok { + s.log.Errorf("evt.New is not a *common.DSRService %v", evt.New) + } else if p, ok := s.providers[SRv6].(*SRv6Provider); ok { + if err := p.UpdateDSRService(svc, false /* isWithdraw */); err != nil { + s.log.Errorf("Error while adding DSR service: %s", err) + } + } + case common.ServiceDSRClusterIPDeleted: + if svc, ok := evt.Old.(*common.DSRService); !ok { + s.log.Errorf("evt.Old is not a *common.DSRService %v", evt.Old) + } else if p, ok := s.providers[SRv6].(*SRv6Provider); ok { + if err := p.UpdateDSRService(svc, true /* isWithdraw */); err != nil { + s.log.Errorf("Error while deleting DSR service: %s", err) + } + } } } } diff --git a/calico-vpp-agent/connectivity/srv6.go b/calico-vpp-agent/connectivity/srv6.go index 4cf355bfe..c8e2b36bb 100644 --- a/calico-vpp-agent/connectivity/srv6.go +++ b/calico-vpp-agent/connectivity/srv6.go @@ -13,6 +13,7 @@ import ( "github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/common" "github.com/projectcalico/vpp-dataplane/v3/config" "github.com/projectcalico/vpp-dataplane/v3/vpplink" + "github.com/projectcalico/vpp-dataplane/v3/vpplink/generated/bindings/interface_types" "github.com/projectcalico/vpp-dataplane/v3/vpplink/generated/bindings/ip_types" "github.com/projectcalico/vpp-dataplane/v3/vpplink/types" ) @@ -44,10 +45,16 @@ type SRv6Provider struct { policyIPPool net.IPNet // localSidIPPool is IP pool for LocalSID's SIDs (SID = IPv6 address in SRv6) localSidIPPool net.IPNet + // dsrServices tracks the SR policy + steering installed for SRv6-native / + // NAT-less (DSR) ClusterIP services, keyed by VIP string. + dsrServices map[string]*dsrServiceState + // dsrDesired is the desired DSR service set (by VIP) — the source of truth + // reconciled against dsrServices, so failed installs AND removals retry. + dsrDesired map[string]*common.DSRService } func NewSRv6Provider(d *ConnectivityProviderData) *SRv6Provider { - p := &SRv6Provider{d, make(map[string]*NodeToPrefixes), make(map[string]*NodeToPolicies), net.IPNet{}, net.IPNet{}} + p := &SRv6Provider{d, make(map[string]*NodeToPrefixes), make(map[string]*NodeToPolicies), net.IPNet{}, net.IPNet{}, make(map[string]*dsrServiceState), make(map[string]*common.DSRService)} if *config.GetCalicoVppFeatureGates().SRv6Enabled { p.localSidIPPool = cnet.MustParseNetwork(config.GetCalicoVppSrv6().LocalsidPool).IPNet p.policyIPPool = cnet.MustParseNetwork(config.GetCalicoVppSrv6().PolicyPool).IPNet @@ -259,6 +266,10 @@ func (p *SRv6Provider) AddConnectivity(cn *common.NodeConnectivity) (err error) p.steerNodeIPViaSID(nodeip) } + // A backend node's End.DT6 SID may have just become available (or a pending + // DSR removal may need retrying); reconcile the DSR services. + p.reconcileDSRServices() + return err } @@ -319,13 +330,11 @@ func (p *SRv6Provider) createLocalSidTunnels(currentLocalSids []*types.SrLocalsi endDt4Exist := false endDt6Exist := false // End.DT{4,6} must decap into the pod VRF (PodVRFIndex), not the main table. - // The decapped inner packet has to be looked up — and CNAT-reverse-translated - // by cnat-output — in the same FIB where the forward DNAT/session was created. - // Pod (and pod-backed service) traffic resolves in calico-pods-ip6 == - // PodVRFIndex, so the CNAT session lives there. With FibTable 0 the return - // path of a cross-node pod-backed ClusterIP misses the CNAT session: the - // SYN-ACK reaches the client sourced from the backend pod IP instead of the - // ClusterIP, so the client RSTs and the connection never establishes. + // The decapped inner packet is looked up there, where pod routes and the + // SRv6-native (DSR) ClusterIP delivery routes live. VPP takes the decap VRF + // from the localsid's sw_if_index (dumped as SwIfIndex), so a correctly + // configured End.DT has SwIfIndex == PodVRFIndex. + podVRF := interface_types.InterfaceIndex(common.PodVRFIndex) for _, localSid := range currentLocalSids { p.log.Debugf("Found existing SRv6Localsid: %s", localSid.String()) @@ -335,34 +344,32 @@ func (p *SRv6Provider) createLocalSidTunnels(currentLocalSids []*types.SrLocalsi continue } - if localSid.FibTable == common.PodVRFIndex { + if localSid.SwIfIndex == podVRF { endDt6Exist = endDt6Exist || isDT6 endDt4Exist = endDt4Exist || isDT4 continue } - // Migrate a stale localsid created in the main table (FibTable 0) by an - // older agent: delete it and re-add it at the same SID address in - // PodVRFIndex, so remote nodes keep encapsulating to the same SID. - if localSid.FibTable == 0 { - p.log.Infof("SRv6Provider migrating End.DT localsid %s from main table to PodVRFIndex", localSid.Localsid.String()) - if err := p.vpp.DelSRv6Localsid(localSid); err != nil { - p.log.Errorf("SRv6Provider migrate: delete stale localsid %s: %v", localSid.Localsid.String(), err) - continue - } - migrated := &types.SrLocalsid{ - Localsid: localSid.Localsid, - EndPsp: false, - FibTable: common.PodVRFIndex, - Behavior: localSid.Behavior, - } - if err := p.vpp.AddSRv6Localsid(migrated); err != nil { - p.log.Errorf("SRv6Provider migrate: re-add localsid %s in PodVRFIndex: %v", localSid.Localsid.String(), err) - continue - } - endDt6Exist = endDt6Exist || isDT6 - endDt4Exist = endDt4Exist || isDT4 + // Migrate a stale localsid created by an older agent to decap into the + // main table: delete it and re-add it at the same SID address decapping + // into PodVRFIndex, so remote nodes keep encapsulating to the same SID. + p.log.Infof("SRv6Provider migrating End.DT localsid %s into PodVRFIndex", localSid.Localsid.String()) + if err := p.vpp.DelSRv6Localsid(localSid); err != nil { + p.log.Errorf("SRv6Provider migrate: delete stale localsid %s: %v", localSid.Localsid.String(), err) + continue + } + migrated := &types.SrLocalsid{ + Localsid: localSid.Localsid, + EndPsp: false, + SwIfIndex: podVRF, + Behavior: localSid.Behavior, + } + if err := p.vpp.AddSRv6Localsid(migrated); err != nil { + p.log.Errorf("SRv6Provider migrate: re-add localsid %s in PodVRFIndex: %v", localSid.Localsid.String(), err) + continue } + endDt6Exist = endDt6Exist || isDT6 + endDt4Exist = endDt4Exist || isDT4 } if !endDt4Exist { if localSidDT4, err := p.setEndDT(4); err != nil { @@ -405,10 +412,12 @@ func (p *SRv6Provider) setEndDT(typeDT int) (newLocalSid *types.SrLocalsid, err newLocalSid = &types.SrLocalsid{ Localsid: newLocalSidAddr, EndPsp: false, - // Decap into the pod VRF so the inner packet (and its CNAT reverse - // translation) is processed in the same FIB as pod/service routing. - FibTable: common.PodVRFIndex, - Behavior: behavior, + // End.DT{4,6} read the decap VRF from sw_if_index (dumped as + // XconnectIfaceOrVrfTable), not fib_table. Point it at PodVRFIndex so the + // decapped inner packet is looked up in the pod VRF, where pod routes and + // SRv6-native (DSR) ClusterIP delivery routes live. + SwIfIndex: interface_types.InterfaceIndex(common.PodVRFIndex), + Behavior: behavior, } if err = p.vpp.AddSRv6Localsid(newLocalSid); err != nil { p.log.Infof("SRv6Provider Error adding LocalSid") diff --git a/calico-vpp-agent/connectivity/srv6_dsr.go b/calico-vpp-agent/connectivity/srv6_dsr.go new file mode 100644 index 000000000..d20a0fb5b --- /dev/null +++ b/calico-vpp-agent/connectivity/srv6_dsr.go @@ -0,0 +1,272 @@ +package connectivity + +import ( + "net" + + "github.com/pkg/errors" + + "github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/common" + "github.com/projectcalico/vpp-dataplane/v3/config" + "github.com/projectcalico/vpp-dataplane/v3/vpplink/generated/bindings/ip_types" + "github.com/projectcalico/vpp-dataplane/v3/vpplink/types" +) + +// dsrServiceState tracks the SR policy + steering installed for a DSR ClusterIP. +type dsrServiceState struct { + vip net.IP + bsid ip_types.IP6Address + policy *types.SrPolicy + steer *types.SrSteer +} + +// UpdateDSRService installs (or withdraws) the client-side steering for an +// SRv6-native / NAT-less (DSR) ClusterIP: a per-service SR policy whose weighted +// SID lists each end at a backend node's End.DT6 SID, plus a steering entry that +// sends VIP traffic into that policy (in the main table). Clients with a local +// backend never reach this steer because the CNI server installs a more-specific +// VIP delivery route in PodVRFIndex; clients without one fall through to here and +// are SRv6-steered to a backend node, which decaps into PodVRFIndex and delivers +// to its local backend. Local backends of this node are handled by the CNI +// server, so they are excluded from the policy here. +func (p *SRv6Provider) UpdateDSRService(svc *common.DSRService, isWithdraw bool) error { + if !*config.GetCalicoVppFeatureGates().SRv6Enabled || svc == nil || svc.VIP == nil { + return nil + } + key := svc.VIP.String() + if isWithdraw { + delete(p.dsrDesired, key) + } else { + p.dsrDesired[key] = svc + } + p.reconcileDSRServices() + return nil +} + +// reconcileDSRServices drives the installed DSR state toward the desired set, +// retrying both failed installs and failed removals. It is invoked on every DSR +// event and on connectivity changes, so a backend node's End.DT6 SID arriving +// late (BGP) or a previously failed delete is eventually applied. +func (p *SRv6Provider) reconcileDSRServices() { + for _, svc := range p.dsrDesired { + if err := p.programDSRService(svc); err != nil { + p.log.Errorf("SRv6Provider DSR program vip %s: %v", svc.VIP, err) + } + } + for key := range p.dsrServices { + if _, ok := p.dsrDesired[key]; !ok { + p.removeDSRService(key) + } + } +} + +// programDSRService installs (idempotently) the per-service SR policy + steering. +func (p *SRv6Provider) programDSRService(svc *common.DSRService) error { + key := svc.VIP.String() + // Group remote backends by node (exclude this node). + _, myIP6 := p.GetNodeIPs() + weights := make(map[string]uint32) + for _, beIP := range svc.Backends { + nodeIP := p.findNodeForPrefix(beIP) + if nodeIP == nil { + p.log.Infof("SRv6Provider DSR: no node for backend %s (vip %s), skipping", beIP, key) + continue + } + if myIP6 != nil && nodeIP.Equal(*myIP6) { + continue // local backend, delivered by CNI in PodVRFIndex + } + weights[nodeIP.String()]++ + } + + if len(weights) == 0 { + // No remote backends: local delivery suffices; drop any stale steer. + p.removeDSRService(key) + return nil + } + + // One weighted SID list per backend node, ending at its End.DT6 SID. + var sidLists []types.Srv6SidList + for nodeIPStr, w := range weights { + policy, err := p.getPolicyNode(nodeIPStr, types.SrBehaviorDT6) + if err != nil || policy == nil || len(policy.SidLists) == 0 || policy.SidLists[0].NumSids == 0 { + p.log.Infof("SRv6Provider DSR: no DT6 SID for node %s yet (vip %s), skipping", nodeIPStr, key) + continue + } + sl := types.Srv6SidList{NumSids: 1, Weight: w} + sl.Sids[0] = policy.SidLists[0].Sids[0] + sidLists = append(sidLists, sl) + } + if len(sidLists) == 0 { + p.log.Infof("SRv6Provider DSR: no usable backend node SIDs yet for vip %s; will retry on next reconcile", key) + return nil + } + + // Already installed identically: skip (lets the periodic re-publish retry + // missing programming without churning steady-state state). Guard st.policy: + // a prior partial removeDSRService may have nil'd it while keeping st.steer. + if st, ok := p.dsrServices[key]; ok && st.steer != nil && st.policy != nil && sameSidLists(st.policy.SidLists, sidLists) { + return nil + } + + // Reuse the previously chosen BSID for stability; derive a fresh + // (collision-avoiding) one only on first install for this VIP. + var bsid ip_types.IP6Address + if st, ok := p.dsrServices[key]; ok { + bsid = st.bsid + } else { + bsid = p.dsrBsidForVIP(svc.VIP) + } + policy := &types.SrPolicy{ + Bsid: bsid, + IsEncap: true, + FibTable: 0, + SidLists: sidLists, + } + if err := p.vpp.AddModSRv6Policy(policy); err != nil { + return errors.Wrapf(err, "SRv6Provider DSR AddModSRv6Policy vip %s", key) + } + + prefix, err := ip_types.ParsePrefix(svc.VIP.String() + "/128") + if err != nil { + return errors.Wrapf(err, "SRv6Provider DSR parse vip prefix %s", key) + } + steer := &types.SrSteer{ + TrafficType: types.SrSteerIPv6, + FibTable: 0, + Prefix: prefix, + Bsid: bsid, + } + if st, ok := p.dsrServices[key]; ok && st.steer != nil { + _ = p.vpp.DelSRv6Steering(st.steer) + } + if err := p.vpp.AddSRv6Steering(steer); err != nil { + return errors.Wrapf(err, "SRv6Provider DSR AddSRv6Steering vip %s", key) + } + p.dsrServices[key] = &dsrServiceState{vip: svc.VIP, bsid: bsid, policy: policy, steer: steer} + p.log.Infof("SRv6Provider DSR: vip %s steered over %d backend node(s)", key, len(sidLists)) + return nil +} + +// removeDSRService withdraws the SR policy + steering for a VIP. On partial +// failure it nils out the parts that succeeded and keeps the rest installed, so +// the next reconcile retries only what's left (rather than re-deleting already +// gone objects or forgetting a still-present one). +func (p *SRv6Provider) removeDSRService(key string) { + st, ok := p.dsrServices[key] + if !ok { + return + } + failed := false + if st.steer != nil { + if err := p.vpp.DelSRv6Steering(st.steer); err != nil { + p.log.Errorf("SRv6Provider DSR DelSRv6Steering vip %s: %v", key, err) + failed = true + } else { + st.steer = nil + } + } + if st.policy != nil { + if err := p.vpp.DelSRv6Policy(st.policy); err != nil { + p.log.Errorf("SRv6Provider DSR DelSRv6Policy vip %s: %v", key, err) + failed = true + } else { + st.policy = nil + } + } + if failed { + return // keep state; retried on next reconcile + } + delete(p.dsrServices, key) +} + +// sameSidLists reports whether two sets of single-SID weighted lists are equal +// (order-independent). ip_types.IP6Address is an array, hence comparable. +func sameSidLists(a, b []types.Srv6SidList) bool { + if len(a) != len(b) { + return false + } + type key struct { + sid ip_types.IP6Address + weight uint32 + } + m := make(map[key]int) + for _, sl := range a { + m[key{sl.Sids[0], sl.Weight}]++ + } + for _, sl := range b { + m[key{sl.Sids[0], sl.Weight}]-- + } + for _, v := range m { + if v != 0 { + return false + } + } + return true +} + +// findNodeForPrefix returns the node IP whose advertised pod prefix contains ip. +func (p *SRv6Provider) findNodeForPrefix(ip net.IP) net.IP { + for _, np := range p.nodePrefixes { + for _, prefix := range np.Prefixes { + ipnet := prefix.ToIPNet() + if ipnet != nil && ipnet.Contains(ip) { + return np.Node + } + } + } + return nil +} + +// dsrBsidForVIP derives a BSID for a service from the policy pool: the pool +// prefix with the VIP's host bytes overlaid, then perturbed if it would collide +// with a BSID already in use by another (per-node pod-connectivity) SR policy. +// Per-node BSIDs are IPAM-allocated from the same pool, so a naive derivation +// could clobber a node's pod-connectivity policy on install/withdraw; avoiding +// in-use BSIDs prevents that. The chosen value is cached per service in +// dsrServices and reused across reconciles for stability. +func (p *SRv6Provider) dsrBsidForVIP(vip net.IP) ip_types.IP6Address { + bsid := make(net.IP, net.IPv6len) + copy(bsid, p.policyIPPool.IP.To16()) + v := vip.To16() + ones, _ := p.policyIPPool.Mask.Size() + for i := ones / 8; i < net.IPv6len && i < len(v); i++ { + bsid[i] = v[i] + } + taken := p.takenBsids() + for tries := 0; tries < 1024 && taken[bsid.String()]; tries++ { + bsid[net.IPv6len-1]++ // perturb low byte until free + } + return types.ToVppIP6Address(bsid) +} + +// takenBsids returns the set of BSIDs already in use by non-DSR SR policies +// (per-node pod-connectivity policies, learned via BGP and/or installed in VPP), +// so DSR BSID derivation can avoid clobbering them. Our own DSR policies are +// excluded so we don't perturb away from an already-chosen stable BSID. +func (p *SRv6Provider) takenBsids() map[string]bool { + m := make(map[string]bool) + for _, np := range p.nodePolices { + for _, t := range np.SRv6Tunnel { + if t.Bsid != nil { + m[t.Bsid.String()] = true + } + } + } + if pols, err := p.vpp.ListSRv6Policies(); err == nil { + for _, pol := range pols { + if p.isOwnDSRBsid(pol.Bsid) { + continue + } + m[pol.Bsid.ToIP().String()] = true + } + } + return m +} + +func (p *SRv6Provider) isOwnDSRBsid(b ip_types.IP6Address) bool { + for _, st := range p.dsrServices { + if st.bsid == b { + return true + } + } + return false +} diff --git a/calico-vpp-agent/services/service.go b/calico-vpp-agent/services/service.go index 79b48fbd8..19c30fee6 100644 --- a/calico-vpp-agent/services/service.go +++ b/calico-vpp-agent/services/service.go @@ -29,4 +29,7 @@ type serviceInfo struct { keepOriginalPacket bool lbType lbType hashConfig types.IPFlowHash + // srv6Native opts this service into the SRv6-native / NAT-less (DSR) path + // (cni.projectcalico.org/vppSRv6Native: "true"). + srv6Native bool } diff --git a/calico-vpp-agent/services/service_dsr.go b/calico-vpp-agent/services/service_dsr.go new file mode 100644 index 000000000..299decd6e --- /dev/null +++ b/calico-vpp-agent/services/service_dsr.go @@ -0,0 +1,177 @@ +package services + +import ( + "net" + + v1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/util/intstr" + + discoveryv1 "k8s.io/api/discovery/v1" + + "github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/common" + "github.com/projectcalico/vpp-dataplane/v3/config" +) + +// dsrEligible reports whether a service should use the SRv6-native / NAT-less +// (DSR) data path instead of cnat. First cut: requires the feature gate + the +// per-service opt-in annotation, must be a ClusterIP service, and every port +// must keep port==targetPort (DSR can't remap L4 ports). Everything else +// (NodePort/External/LoadBalancer, port-remapped, or host-network backed +// services such as the apiserver) keeps the cnat path. The opt-in annotation is +// required so we never accidentally DSR a host-network backed service. +func (s *Server) dsrEligible(service *v1.Service, epSlices []*discoveryv1.EndpointSlice, svcInfo *serviceInfo) bool { + gates := config.GetCalicoVppFeatureGates() + if gates.SRv6NativeServicesEnabled == nil || !*gates.SRv6NativeServicesEnabled { + return false + } + if gates.SRv6Enabled == nil || !*gates.SRv6Enabled { + return false + } + if !svcInfo.srv6Native { + return false + } + if service.Spec.Type != v1.ServiceTypeClusterIP { + return false + } + // ExternalIPs (and LoadBalancer ingress) are still served by cnat, which + // hits the SRv6 un-DNAT issue this feature avoids. Don't DSR a service that + // exposes them — keep the whole service on the cnat path for consistency. + if len(service.Spec.ExternalIPs) > 0 || len(service.Status.LoadBalancer.Ingress) > 0 { + return false + } + for _, port := range service.Spec.Ports { + tp := port.TargetPort + // Named target ports can't be statically verified to equal the service + // port; treat them as not eligible for now. + if tp.Type == intstr.String { + s.log.Warnf("svc(dsr) %s/%s: named targetPort not supported, falling back to cnat", service.Namespace, service.Name) + return false + } + if tp.IntVal != 0 && tp.IntVal != port.Port { + s.log.Warnf("svc(dsr) %s/%s: port remap %d->%d not supported, falling back to cnat", service.Namespace, service.Name, port.Port, tp.IntVal) + return false + } + } + // All (ready, IPv6) backends must be pod-backed: their IPs must fall in a + // Calico IPAM pool. Host-network backed services (endpoints == node IPs) are + // not in any IPAM pool, and DSR cannot deliver to them (no pod interface to + // bind the VIP on). Keep such services on the cnat path rather than silently + // skipping cnat and leaving them with no working path. + if s.felixServerIpam != nil { + for _, epslice := range epSlices { + for _, ep := range epslice.Endpoints { + if ep.Conditions.Ready != nil && !*ep.Conditions.Ready { + continue + } + for _, addr := range ep.Addresses { + ip := net.ParseIP(addr) + if ip == nil || ip.To4() != nil { + continue + } + if s.felixServerIpam.GetPrefixIPPool(common.ToMaxLenCIDR(ip)) == nil { + s.log.Warnf("svc(dsr) %s/%s: backend %s is not pod-backed (not in an IPAM pool); falling back to cnat", service.Namespace, service.Name, addr) + return false + } + } + } + } + } + return true +} + +// buildDSRServices returns one common.DSRService per (IPv6) ClusterIP of a +// DSR-eligible service, carrying all of its ready pod-backed backend IPs. +func (s *Server) buildDSRServices(service *v1.Service, epSlices []*discoveryv1.EndpointSlice, svcInfo *serviceInfo) []common.DSRService { + if !s.dsrEligible(service, epSlices, svcInfo) { + return nil + } + backendSet := make(map[string]net.IP) + for _, epslice := range epSlices { + for _, ep := range epslice.Endpoints { + if ep.Conditions.Ready != nil && !*ep.Conditions.Ready { + continue + } + for _, addr := range ep.Addresses { + ip := net.ParseIP(addr) + if ip == nil || ip.To4() != nil { + continue // IPv6 backends only for now + } + backendSet[ip.String()] = ip + } + } + } + backends := make([]net.IP, 0, len(backendSet)) + for _, ip := range backendSet { + backends = append(backends, ip) + } + + var out []common.DSRService + for _, cip := range service.Spec.ClusterIPs { + vip := net.ParseIP(cip) + if vip == nil || vip.To4() != nil || vip.IsUnspecified() { + continue // IPv6 ClusterIP only for now + } + out = append(out, common.DSRService{ + VIP: vip, + Backends: backends, + ServiceID: objectID(&service.ObjectMeta), + }) + } + return out +} + +// handleDSREntries reconciles the DSR entries of the (current) resolved service +// against the set we previously published for that service id, and emits +// add/delete events consumed by the SRv6 provider and the CNI server. It diffs +// against our own tracked state (s.dsrByServiceID) rather than the passed +// oldService, so it is robust to the caller's argument order. service deletion +// is handled separately in deleteServiceByName. +func (s *Server) handleDSREntries(service *LocalService, _ *LocalService) { + if service == nil { + return + } + serviceID := service.ServiceID + desired := make(map[string]common.DSRService) + for _, d := range service.DSREntries { + desired[d.Key()] = d + } + s.reconcileDSR(serviceID, desired) +} + +// reconcileDSR diffs the desired DSR entries for a service against the tracked +// set and publishes the resulting add/delete events. +func (s *Server) reconcileDSR(serviceID string, desired map[string]common.DSRService) { + prev := s.dsrByServiceID[serviceID] + + // VIPs present before but gone now. + for key, d := range prev { + if _, ok := desired[key]; !ok { + entry := d + s.log.Infof("svc(dsr-del) %s vip=%s", entry.ServiceID, key) + common.SendEvent(common.CalicoVppEvent{ + Type: common.ServiceDSRClusterIPDeleted, + Old: &entry, + }) + } + } + // Always (re)publish the desired VIPs rather than suppressing unchanged ones. + // The consumers (SRv6 provider, CNI) are idempotent and skip when already + // correctly programmed, so the periodic informer resync re-drives this and + // retries any programming that previously failed (e.g. a backend node's + // End.DT6 SID not yet learned via BGP, or a transient VPP error). Suppressing + // here would record such failures as reconciled and never retry them. + for _, d := range desired { + entry := d + s.log.Debugf("svc(dsr-add) %s vip=%s backends=%d", entry.ServiceID, entry.Key(), len(entry.Backends)) + common.SendEvent(common.CalicoVppEvent{ + Type: common.ServiceDSRClusterIPAdded, + New: &entry, + }) + } + + if len(desired) == 0 { + delete(s.dsrByServiceID, serviceID) + } else { + s.dsrByServiceID[serviceID] = desired + } +} diff --git a/calico-vpp-agent/services/service_handler.go b/calico-vpp-agent/services/service_handler.go index eedf92948..3e617415c 100644 --- a/calico-vpp-agent/services/service_handler.go +++ b/calico-vpp-agent/services/service_handler.go @@ -158,6 +158,10 @@ func (s *Server) GetLocalService(service *v1.Service, epSlicesMap map[string]*di } serviceSpec := s.ParseServiceAnnotations(service.Annotations, service.Name) + // DSR-eligible ClusterIPs are steered/delivered over SRv6 (NAT-less) instead + // of being programmed as cnat translations. + dsr := s.dsrEligible(service, epSlices, serviceSpec) + localService.DSREntries = s.buildDSRServices(service, epSlices, serviceSpec) var clusterIPs []net.IP var nodeIPs []net.IP for _, cip := range service.Spec.ClusterIPs { @@ -166,7 +170,8 @@ func (s *Server) GetLocalService(service *v1.Service, epSlicesMap map[string]*di } for _, servicePort := range service.Spec.Ports { for _, cip := range clusterIPs { - if !cip.IsUnspecified() && len(cip) > 0 { + // Skip cnat for ClusterIPs handled by the SRv6-native (DSR) path. + if !dsr && !cip.IsUnspecified() && len(cip) > 0 { entry := s.buildCnatEntryForServicePort(&servicePort, epSlices, cip, false /* isNodePort */, *serviceSpec, InternalIsLocalOnly(service)) localService.Entries = append(localService.Entries, *entry) } @@ -301,6 +306,8 @@ func (s *Server) deleteServiceEntries(entries []types.CnatTranslateEntry, oldSer func (s *Server) deleteServiceByName(serviceID string) { s.lock.Lock() defer s.lock.Unlock() + // Withdraw any SRv6-native (DSR) entries published for this service. + s.reconcileDSR(serviceID, nil) for key := range s.cnatEntryBySidAndKey[serviceID] { s.deleteServiceEntry(key, serviceID) } diff --git a/calico-vpp-agent/services/service_server.go b/calico-vpp-agent/services/service_server.go index 445408dfd..b6a808e79 100644 --- a/calico-vpp-agent/services/service_server.go +++ b/calico-vpp-agent/services/service_server.go @@ -50,6 +50,10 @@ type LocalService struct { Entries []types.CnatTranslateEntry SpecificRoutes []net.IP ServiceID string + // DSREntries holds SRv6-native / NAT-less (DSR) ClusterIP entries for this + // service. When non-empty, the corresponding ClusterIPs are NOT programmed + // as cnat translations; they are steered/delivered over SRv6 instead. + DSREntries []common.DSRService } /** @@ -63,6 +67,7 @@ type cnatEntry struct { type Server struct { log *logrus.Entry vpp *vpplink.VppLink + felixServerIpam common.FelixServerIpam endpointslicesStore cache.Store serviceStore cache.Store serviceInformer cache.Controller @@ -84,6 +89,10 @@ type Server struct { endpointSlicesByService map[string]map[string]*discoveryv1.EndpointSlice endpointSlices map[string]*discoveryv1.EndpointSlice + // dsrByServiceID tracks the DSR entries we've published, keyed by service id + // then VIP, so we can diff/withdraw them (including on service deletion). + dsrByServiceID map[string]map[string]common.DSRService + t tomb.Tomb } @@ -140,6 +149,12 @@ func (s *Server) ParseServiceAnnotations(annotations map[string]string, name str if err1 != nil { err = append(err, errors.Wrapf(err1, "Unknown value %s for key %s", value, key)) } + case config.SRv6NativeAnnotation: + var err1 error + svc.srv6Native, err1 = strconv.ParseBool(value) + if err1 != nil { + err = append(err, errors.Wrapf(err1, "Unknown value %s for key %s", value, key)) + } default: continue } @@ -157,15 +172,17 @@ func (s *Server) resolveLocalServiceFromService(service *v1.Service) *LocalServi return s.GetLocalService(service, s.endpointSlicesByService[objectID(&service.ObjectMeta)]) } -func NewServiceServer(vpp *vpplink.VppLink, k8sclient *kubernetes.Clientset, log *logrus.Entry) *Server { +func NewServiceServer(vpp *vpplink.VppLink, k8sclient *kubernetes.Clientset, felixServerIpam common.FelixServerIpam, log *logrus.Entry) *Server { server := Server{ vpp: vpp, + felixServerIpam: felixServerIpam, log: log, cnatEntryByKeyAndSid: make(map[string]map[string]*cnatEntry), cnatEntryBySidAndKey: make(map[string]map[string]*cnatEntry), serviceIDByKey: make(map[string]string), endpointSlicesByService: make(map[string]map[string]*discoveryv1.EndpointSlice), endpointSlices: make(map[string]*discoveryv1.EndpointSlice), + dsrByServiceID: make(map[string]map[string]common.DSRService), } serviceStore, serviceInformer := cache.NewInformerWithOptions( cache.InformerOptions{ @@ -489,6 +506,7 @@ func (s *Server) handleServiceEndpointEvent(service *LocalService, oldService *L if added, deleted, changed := compareSpecificRoutes(service, oldService); changed { s.advertiseSpecificRoute(added, deleted) } + s.handleDSREntries(service, oldService) } func (s *Server) getServiceIPs() ([]*net.IPNet, []*net.IPNet, []*net.IPNet) { diff --git a/calico-vpp-agent/services/services_test.go b/calico-vpp-agent/services/services_test.go index e7c7f24cb..42a0ecb0e 100644 --- a/calico-vpp-agent/services/services_test.go +++ b/calico-vpp-agent/services/services_test.go @@ -96,7 +96,7 @@ var _ = Describe("Service creation functionality", func() { } _, serviceip, err := net.ParseCIDR("10.96.0.1/24") config.ServiceCIDRs = &[]*net.IPNet{serviceip} - serviceServer = NewServiceServer(vpp, k8sclient, log.WithFields(logrus.Fields{"component": "services"})) + serviceServer = NewServiceServer(vpp, k8sclient, nil /* felixServerIpam */, log.WithFields(logrus.Fields{"component": "services"})) _, ipv4net, err := net.ParseCIDR("1.1.1.1/32") _, ipv6net, err := net.ParseCIDR("f::f/128") serviceServer.SetOurBGPSpec(&common.LocalNodeSpec{ diff --git a/config/config.go b/config/config.go index 247d8e95f..5affee063 100644 --- a/config/config.go +++ b/config/config.go @@ -80,6 +80,10 @@ const ( KeepOriginalPacketAnnotation string = "cni.projectcalico.org/vppKeepOriginalPacket" HashConfigAnnotation string = "cni.projectcalico.org/vppHashConfig" LBTypeAnnotation string = "cni.projectcalico.org/vppLBType" + // SRv6NativeAnnotation opts a ClusterIP service into the SRv6-native / + // NAT-less (DSR) data path (requires the SRv6NativeServicesEnabled feature + // gate). Value "true" enables it. + SRv6NativeAnnotation string = "cni.projectcalico.org/vppSRv6Native" ) type BGPServerModeType string @@ -400,6 +404,12 @@ type CalicoVppFeatureGatesConfigType struct { SRv6Enabled *bool `json:"srv6Enabled,omitempty"` IPSecEnabled *bool `json:"ipsecEnabled,omitempty"` PrometheusEnabled *bool `json:"prometheusEnabled,omitempty"` + // SRv6NativeServicesEnabled turns pod-backed ClusterIP services into + // SRv6-native / NAT-less (DSR) services instead of cnat translations. + // Experimental; requires SRv6Enabled. Only applies to ClusterIP services + // whose backends are all pod-backed and use port==targetPort; other + // services keep the cnat path. + SRv6NativeServicesEnabled *bool `json:"srv6NativeServicesEnabled,omitempty"` } func (cfg *CalicoVppFeatureGatesConfigType) Validate() (err error) { @@ -409,6 +419,7 @@ func (cfg *CalicoVppFeatureGatesConfigType) Validate() (err error) { cfg.SRv6Enabled = DefaultToPtr(cfg.SRv6Enabled, false) cfg.IPSecEnabled = DefaultToPtr(cfg.IPSecEnabled, false) cfg.PrometheusEnabled = DefaultToPtr(cfg.PrometheusEnabled, false) + cfg.SRv6NativeServicesEnabled = DefaultToPtr(cfg.SRv6NativeServicesEnabled, false) return nil }