From 6ee7bcb4583575f8b2623bc16d55f92737465217 Mon Sep 17 00:00:00 2001 From: BeckyPauley <64131207+BeckyPauley@users.noreply.github.com> Date: Tue, 14 Jul 2026 13:26:40 +0100 Subject: [PATCH] cmd/{k8s-operator,containerboot,kube}: support IPv6 in egress ProxyGroup (#19898) * cmd/{k8s-operator,containerboot,k8s-proxy},kube: support IPv6 in egress ProxyGroup Add support for dual-stack and IPv6 clusters in egress ProxyGroup. Previously, egress ProxyGroup only supported IPv4: the operator and containerboot assumed IPv4 for ClusterIP Services, EndpointSlices, and health check headers. This change introduces the following: - Create a per-family EndpointSlice instead of a single IPv4 EndpointSlice. - Update the egress services readiness reconciler to account for both IPv4 and IPv6 EndpointSlices. - Update the pod readiness reconciler to use the primary Pod IP (PodIPs[0]) for readiness checks, instead of hard-coding to use IPv4. - Update the /healthz handler to return both PodIPv4Header and PodIPv6Header. - Add an IPv6 address field to egress status. - Update containerboot and k8s-proxy to use the new health check logic. Updates tailscale/corp#41677 Change-Id: If66a3146df48c75b1e65a71632bbc9fc75feded2 Signed-off-by: Becky Pauley * cmd/{k8s-operator,containerboot}: improve dual-stack egress ProxyGroup On dual-stack clusters, an egress ProxyGroup Service has one EndpointSlice per IP family (IPv4 and IPv6). However, EndpointSlices were only recreated when the ExternalName Service configuration changed, so a deleted EndpointSlice was not recreated. The egress readiness reconciler also had no mechanism to identify which IP families should exist (previously only an IPv4 EndpoitSlice was required). We now create an EndpointSlice for every IP family the ClusterIP Service supports. Also mark an egress Service NotReady when an EndpointSlice for an expected IP family (derived from the Service's ClusterIPs) is missing, so a dual-stack Service missing a family's EndpointSlice is no longer reported Ready. Clarify that the egress pre-shutdown and Pod readiness health checks verify only one IP family on dual-stack clusters. Change-Id: I35b03daf76ac817cd516e9a731770b2d85f6ee16 Signed-off-by: Becky Pauley --------- Signed-off-by: Becky Pauley --- cmd/containerboot/egressservices.go | 61 +++-- cmd/containerboot/egressservices_test.go | 4 +- cmd/containerboot/main.go | 5 +- cmd/k8s-operator/egress-eps.go | 35 ++- cmd/k8s-operator/egress-eps_test.go | 122 +++++++++- cmd/k8s-operator/egress-pod-readiness.go | 24 +- cmd/k8s-operator/egress-pod-readiness_test.go | 52 +++- cmd/k8s-operator/egress-services-readiness.go | 82 +++++-- .../egress-services-readiness_test.go | 227 +++++++++++++++++- cmd/k8s-operator/egress-services.go | 65 +++-- cmd/k8s-operator/egress-services_test.go | 160 +++++++++++- cmd/k8s-operator/testutils_test.go | 5 + cmd/k8s-proxy/k8s-proxy.go | 11 +- kube/egressservices/egressservices.go | 1 + kube/health/healthz.go | 11 +- kube/kubetypes/types.go | 2 + 16 files changed, 774 insertions(+), 93 deletions(-) diff --git a/cmd/containerboot/egressservices.go b/cmd/containerboot/egressservices.go index c8b8a770a..0b2ab024e 100644 --- a/cmd/containerboot/egressservices.go +++ b/cmd/containerboot/egressservices.go @@ -57,7 +57,8 @@ type egressProxy struct { netmapChan chan netmapState // chan to receive netmap state updates on - podIPv4 string // never empty string, currently only IPv4 is supported + podIPv4 string // empty if Pod does not have IPv4 address + podIPv6 string // empty if Pod does not have IPv6 address // tailnetFQDNs is the egress service FQDN to tailnet IP mappings that // were last used to configure firewall rules for this proxy. @@ -138,6 +139,7 @@ type egressProxyRunOpts struct { stateSecret string netmapChan chan netmapState podIPv4 string + podIPv6 string tailnetAddrs []netip.Prefix } @@ -150,6 +152,7 @@ func (ep *egressProxy) configure(opts egressProxyRunOpts) { ep.stateSecret = opts.stateSecret ep.netmapChan = opts.netmapChan ep.podIPv4 = opts.podIPv4 + ep.podIPv6 = opts.podIPv6 ep.tailnetAddrs = opts.tailnetAddrs ep.client = &http.Client{} // default HTTP client sleepDuration := time.Second @@ -418,7 +421,7 @@ func (ep *egressProxy) getStatus(ctx context.Context) (*egressservices.Status, e if err := json.Unmarshal([]byte(raw), status); err != nil { return nil, fmt.Errorf("error unmarshalling previous config: %w", err) } - if reflect.DeepEqual(status.PodIPv4, ep.podIPv4) { + if status.PodIPv4 == ep.podIPv4 && status.PodIPv6 == ep.podIPv6 { return status, nil } return nil, nil @@ -432,6 +435,7 @@ func (ep *egressProxy) setStatus(ctx context.Context, status *egressservices.Sta status = &egressservices.Status{} } status.PodIPv4 = ep.podIPv4 + status.PodIPv6 = ep.podIPv6 secret, err := ep.kc.GetSecret(ctx, ep.stateSecret) if err != nil { return fmt.Errorf("error retrieving state Secret: %w", err) @@ -622,6 +626,8 @@ func servicesStatusIsEqual(st, st1 *egressservices.Status) bool { } st.PodIPv4 = "" st1.PodIPv4 = "" + st.PodIPv6 = "" + st1.PodIPv6 = "" return reflect.DeepEqual(*st, *st1) } @@ -673,24 +679,29 @@ func (ep *egressProxy) waitTillSafeToShutdown(ctx context.Context, cfgs egressse continue } svc := s + // TODO(beckypauley): In dual-stack clusters, this is a best-effort check as we do not control which IP family is used. + // This confirms removal from routing on this node for one family only. The other IP family then relies on the longSleep below. wg.Go(func() { log.Printf("Ensuring that cluster traffic is no longer routed to %q via this Pod...", svc) + podIP, header := ep.podIPv4, kubetypes.PodIPv4Header + if podIP == "" { + podIP, header = ep.podIPv6, kubetypes.PodIPv6Header + } + if ep.podDrained(ctx, svc, hep, podIP, header, hp) { + return + } + ticker := time.NewTicker(ep.shortSleep) + defer ticker.Stop() for { - if ctx.Err() != nil { // kubelet's HTTP request timeout + select { + case <-ctx.Done(): // kubelet's HTTP request timeout log.Printf("Cluster traffic for %s did not stop being routed to this Pod.", svc) return + case <-ticker.C: + if ep.podDrained(ctx, svc, hep, podIP, header, hp) { + return + } } - found, err := lookupPodRoute(ctx, hep, ep.podIPv4, hp, ep.client) - if err != nil { - log.Printf("unable to reach endpoint %q, assuming the routing rules for this Pod have been deleted: %v", hep, err) - break - } - if !found { - log.Printf("service %q is no longer routed through this Pod", svc) - break - } - log.Printf("service %q is still routed through this Pod, waiting...", svc) - time.Sleep(ep.shortSleep) } }) } @@ -704,9 +715,9 @@ func (ep *egressProxy) waitTillSafeToShutdown(ctx context.Context, cfgs egressse // lookupPodRoute calls the healthcheck endpoint repeat times and returns true if the endpoint returns with the podIP // header at least once. -func lookupPodRoute(ctx context.Context, hep, podIP string, repeat int, client httpClient) (bool, error) { +func lookupPodRoute(ctx context.Context, hep, podIP, podIPHeader string, repeat int, client httpClient) (bool, error) { for range repeat { - f, err := lookup(ctx, hep, podIP, client) + f, err := lookup(ctx, hep, podIP, podIPHeader, client) if err != nil { return false, err } @@ -718,7 +729,7 @@ func lookupPodRoute(ctx context.Context, hep, podIP string, repeat int, client h } // lookup calls the healthcheck endpoint and returns true if the response contains the podIP header. -func lookup(ctx context.Context, hep, podIP string, client httpClient) (bool, error) { +func lookup(ctx context.Context, hep, podIP, podIPHeader string, client httpClient) (bool, error) { req, err := http.NewRequestWithContext(ctx, httpm.GET, hep, nil) if err != nil { return false, fmt.Errorf("error creating new HTTP request: %v", err) @@ -733,7 +744,7 @@ func lookup(ctx context.Context, hep, podIP string, client httpClient) (bool, er return true, nil } defer resp.Body.Close() - gotIP := resp.Header.Get(kubetypes.PodIPv4Header) + gotIP := resp.Header.Get(podIPHeader) return strings.EqualFold(podIP, gotIP), nil } @@ -762,3 +773,17 @@ func (ep *egressProxy) getHEPPings() (int, error) { } return hp, nil } + +func (ep *egressProxy) podDrained(ctx context.Context, svc, hep, podIP, header string, hp int) bool { + found, err := lookupPodRoute(ctx, hep, podIP, header, hp, ep.client) + if err != nil { + log.Printf("unable to reach endpoint %q, assuming the routing rules for this Pod have been deleted: %v", hep, err) + return true + } + if !found { + log.Printf("service %q is no longer routed through this Pod", svc) + return true + } + log.Printf("service %q is still routed through this Pod, waiting...", svc) + return false +} diff --git a/cmd/containerboot/egressservices_test.go b/cmd/containerboot/egressservices_test.go index b30765f19..e41cc27f6 100644 --- a/cmd/containerboot/egressservices_test.go +++ b/cmd/containerboot/egressservices_test.go @@ -15,6 +15,7 @@ "strings" "sync" "testing" + "time" "tailscale.com/kube/egressservices" "tailscale.com/kube/kubetypes" @@ -269,7 +270,8 @@ func TestWaitTillSafeToShutdown(t *testing.T) { } ep := &egressProxy{ - podIPv4: podIP, + podIPv4: podIP, + shortSleep: time.Millisecond, client: &mockHTTPClient{ podIP: podIP, anotherIP: anotherIP, diff --git a/cmd/containerboot/main.go b/cmd/containerboot/main.go index bb1c8c26a..3afddbe3b 100644 --- a/cmd/containerboot/main.go +++ b/cmd/containerboot/main.go @@ -423,7 +423,7 @@ func run() error { mux := http.NewServeMux() log.Printf("Running healthcheck endpoint at %s/healthz", cfg.HealthCheckAddrPort) - healthCheck = healthz.RegisterHealthHandlers(mux, cfg.PodIPv4, log.Printf) + healthCheck = healthz.RegisterHealthHandlers(mux, cfg.PodIPv4, cfg.PodIPv6, log.Printf) close := runHTTPServer(mux, cfg.HealthCheckAddrPort) defer close() @@ -439,7 +439,7 @@ func run() error { if cfg.localHealthEnabled() { log.Printf("Running healthcheck endpoint at %s/healthz", cfg.LocalAddrPort) - healthCheck = healthz.RegisterHealthHandlers(mux, cfg.PodIPv4, log.Printf) + healthCheck = healthz.RegisterHealthHandlers(mux, cfg.PodIPv4, cfg.PodIPv6, log.Printf) } if cfg.egressSvcsTerminateEPEnabled() { @@ -948,6 +948,7 @@ func run() error { stateSecret: cfg.KubeSecret, netmapChan: egressSvcsNotify, podIPv4: cfg.PodIPv4, + podIPv6: cfg.PodIPv6, tailnetAddrs: addrs, } go func() { diff --git a/cmd/k8s-operator/egress-eps.go b/cmd/k8s-operator/egress-eps.go index b47009e36..1f02cfab9 100644 --- a/cmd/k8s-operator/egress-eps.go +++ b/cmd/k8s-operator/egress-eps.go @@ -106,16 +106,19 @@ func (er *egressEpsReconciler) Reconcile(ctx context.Context, req reconcile.Requ } newEndpoints := make([]discoveryv1.Endpoint, 0) for _, pod := range podList.Items { - ready, err := er.podIsReadyToRouteTraffic(ctx, pod, &cfg, tailnetSvc, lg) + ready, err := er.podIsReadyToRouteTraffic(ctx, pod, &cfg, tailnetSvc, eps.AddressType, lg) if err != nil { return res, fmt.Errorf("error verifying if Pod is ready to route traffic: %w", err) } if !ready { continue // maybe next time } - podIP, err := podIPv4(&pod) // we currently only support IPv4 + podIP, err := podIPForFamily(&pod, eps.AddressType) if err != nil { - return res, fmt.Errorf("error determining IPv4 address for Pod: %w", err) + return res, fmt.Errorf("error determining Pod IP for %s EndpointSlice: %w", eps.AddressType, err) + } + if podIP == "" { + continue // Pod doesn't have an IP for this address family } newEndpoints = append(newEndpoints, discoveryv1.Endpoint{ Hostname: (*string)(&pod.UID), @@ -140,13 +143,16 @@ func (er *egressEpsReconciler) Reconcile(ctx context.Context, req reconcile.Requ return res, nil } -func podIPv4(pod *corev1.Pod) (string, error) { +func podIPForFamily(pod *corev1.Pod, addrType discoveryv1.AddressType) (string, error) { for _, ip := range pod.Status.PodIPs { parsed, err := netip.ParseAddr(ip.IP) if err != nil { return "", fmt.Errorf("error parsing IP address %s: %w", ip, err) } - if parsed.Is4() { + switch { + case addrType == discoveryv1.AddressTypeIPv4 && parsed.Is4(): + return parsed.String(), nil + case addrType == discoveryv1.AddressTypeIPv6 && parsed.Is6(): return parsed.String(), nil } } @@ -156,20 +162,19 @@ func podIPv4(pod *corev1.Pod) (string, error) { // podIsReadyToRouteTraffic returns true if it appears that the proxy Pod has configured firewall rules to be able to // route traffic to the given tailnet service. It retrieves the proxy's state Secret and compares the tailnet service // status written there to the desired service configuration. -func (er *egressEpsReconciler) podIsReadyToRouteTraffic(ctx context.Context, pod corev1.Pod, cfg *egressservices.Config, tailnetSvcName string, lg *zap.SugaredLogger) (bool, error) { +func (er *egressEpsReconciler) podIsReadyToRouteTraffic(ctx context.Context, pod corev1.Pod, cfg *egressservices.Config, tailnetSvcName string, addrType discoveryv1.AddressType, lg *zap.SugaredLogger) (bool, error) { lg = lg.With("proxy_pod", pod.Name) lg.Debug("checking whether proxy is ready to route to egress service") if !pod.DeletionTimestamp.IsZero() { lg.Debug("proxy Pod is being deleted, ignore") return false, nil } - - podIP, err := podIPv4(&pod) + podIP, err := podIPForFamily(&pod, addrType) switch { case err != nil: return false, fmt.Errorf("error determining Pod IP address: %v", err) case podIP == "": - lg.Warn("Pod does not have an IPv4 address, and IPv6 is not currently supported") + lg.Debugf("Pod does not have an address for family %s", addrType) return false, nil } @@ -199,9 +204,15 @@ func (er *egressEpsReconciler) podIsReadyToRouteTraffic(ctx context.Context, pod if err = json.Unmarshal(svcStatusBS, svcStatus); err != nil { return false, fmt.Errorf("error unmarshalling egress service status: %w", err) } - - if !strings.EqualFold(podIP, svcStatus.PodIPv4) { - lg.Infof("proxy's egress service status is for Pod IP %q, current proxy's Pod IP %q, waiting for the proxy to reconfigure...", svcStatus.PodIPv4, podIP) + var statusIP string + switch addrType { + case discoveryv1.AddressTypeIPv4: + statusIP = svcStatus.PodIPv4 + case discoveryv1.AddressTypeIPv6: + statusIP = svcStatus.PodIPv6 + } + if !strings.EqualFold(podIP, statusIP) { + lg.Infof("proxy's egress service status is for Pod IP %q, current proxy's Pod IP %q, waiting for the proxy to reconfigure...", statusIP, podIP) return false, nil } diff --git a/cmd/k8s-operator/egress-eps_test.go b/cmd/k8s-operator/egress-eps_test.go index 6335b4eb8..14ed38d34 100644 --- a/cmd/k8s-operator/egress-eps_test.go +++ b/cmd/k8s-operator/egress-eps_test.go @@ -98,7 +98,7 @@ func TestTailscaleEgressEndpointSlices(t *testing.T) { t.Run("pods_are_ready_to_route_traffic", func(t *testing.T) { pod, stateS := podAndSecretForProxyGroup("foo") - stBs := serviceStatusForPodIP(t, svc, pod.Status.PodIPs[0].IP, port) + stBs := serviceStatusForPodIPs(t, svc, pod.Status.PodIPs[0].IP, "", port) mustUpdate(t, fc, "operator-ns", stateS.Name, func(s *corev1.Secret) { mak.Set(&s.Data, egressservices.KeyEgressServices, stBs) }) @@ -115,8 +115,8 @@ func TestTailscaleEgressEndpointSlices(t *testing.T) { expectEqual(t, fc, eps) }) t.Run("status_does_not_match_pod_ip", func(t *testing.T) { - _, stateS := podAndSecretForProxyGroup("foo") // replica Pod has IP 10.0.0.1 - stBs := serviceStatusForPodIP(t, svc, "10.0.0.2", port) // status is for a Pod with IP 10.0.0.2 + _, stateS := podAndSecretForProxyGroup("foo") // replica Pod has IP 10.0.0.1 + stBs := serviceStatusForPodIPs(t, svc, "10.0.0.2", "", port) // status is for a Pod with IP 10.0.0.2 mustUpdate(t, fc, "operator-ns", stateS.Name, func(s *corev1.Secret) { mak.Set(&s.Data, egressservices.KeyEgressServices, stBs) }) @@ -124,6 +124,117 @@ func TestTailscaleEgressEndpointSlices(t *testing.T) { eps.Endpoints = []discoveryv1.Endpoint{} expectEqual(t, fc, eps) }) + + // Dual-stack. + epsV6 := &discoveryv1.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{ + Name: "foo-ipv6", + Namespace: "operator-ns", + Labels: map[string]string{ + LabelParentName: "test", + LabelParentNamespace: "default", + labelSvcType: typeEgress, + labelProxyGroup: "foo", + }, + }, + AddressType: discoveryv1.AddressTypeIPv6, + } + mustCreate(t, fc, epsV6) + t.Run("dual_stack_pod_ready_to_route", func(t *testing.T) { + mustDeleteAll(t, fc, &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "foo-0", Namespace: "operator-ns"}}) + dualPod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "foo-0", + Namespace: "operator-ns", + Labels: pgLabels("foo", nil), + UID: "foo", + }, + Status: corev1.PodStatus{ + PodIPs: []corev1.PodIP{{IP: "10.0.0.1"}, {IP: "fd00::1"}}, + }, + } + mustCreate(t, fc, dualPod) + stBs := serviceStatusForPodIPs(t, svc, "10.0.0.1", "fd00::1", port) + mustUpdate(t, fc, "operator-ns", "foo-0", func(s *corev1.Secret) { + mak.Set(&s.Data, egressservices.KeyEgressServices, stBs) + }) + expectReconciled(t, er, "operator-ns", "foo") + eps.Endpoints = []discoveryv1.Endpoint{{ + Addresses: []string{"10.0.0.1"}, + Hostname: new("foo"), + Conditions: discoveryv1.EndpointConditions{ + Serving: new(true), + Ready: new(true), + Terminating: new(false), + }, + }} + expectEqual(t, fc, eps) + expectReconciled(t, er, "operator-ns", "foo-ipv6") + epsV6.Endpoints = []discoveryv1.Endpoint{{ + Addresses: []string{"fd00::1"}, + Hostname: new("foo"), + Conditions: discoveryv1.EndpointConditions{ + Serving: new(true), + Ready: new(true), + Terminating: new(false), + }, + }} + expectEqual(t, fc, epsV6) + }) + + // IPv6-only. + t.Run("ipv4_only_pod_skipped_for_ipv6_slice", func(t *testing.T) { + mustDeleteAll(t, fc, &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "foo-0", Namespace: "operator-ns"}}) + ipv4Pod, _ := podAndSecretForProxyGroup("foo") + mustCreate(t, fc, ipv4Pod) + stBs := serviceStatusForPodIPs(t, svc, "10.0.0.1", "", port) + mustUpdate(t, fc, "operator-ns", "foo-0", func(s *corev1.Secret) { + mak.Set(&s.Data, egressservices.KeyEgressServices, stBs) + }) + expectReconciled(t, er, "operator-ns", "foo-ipv6") + // IPv4-only pod should not appear in the IPv6 EndpointSlice. + epsV6.Endpoints = []discoveryv1.Endpoint{} + expectEqual(t, fc, epsV6) + }) + ipv6Pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "foo-0", + Namespace: "operator-ns", + Labels: pgLabels("foo", nil), + UID: "foo", + }, + Status: corev1.PodStatus{ + PodIPs: []corev1.PodIP{{IP: "fd00::1"}}, + }, + } + t.Run("ipv6_status_does_not_match_pod_ip", func(t *testing.T) { + mustDeleteAll(t, fc, &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "foo-0", Namespace: "operator-ns"}}) + mustCreate(t, fc, ipv6Pod) + stBs := serviceStatusForPodIPs(t, svc, "", "fd00::99", port) + mustUpdate(t, fc, "operator-ns", "foo-0", func(s *corev1.Secret) { + mak.Set(&s.Data, egressservices.KeyEgressServices, stBs) + }) + expectReconciled(t, er, "operator-ns", "foo-ipv6") + epsV6.Endpoints = []discoveryv1.Endpoint{} + expectEqual(t, fc, epsV6) + }) + t.Run("ipv6_pod_ready_to_route", func(t *testing.T) { + stBs := serviceStatusForPodIPs(t, svc, "", ipv6Pod.Status.PodIPs[0].IP, port) + mustUpdate(t, fc, "operator-ns", "foo-0", func(s *corev1.Secret) { + mak.Set(&s.Data, egressservices.KeyEgressServices, stBs) + }) + expectReconciled(t, er, "operator-ns", "foo-ipv6") + epsV6.Endpoints = append(epsV6.Endpoints, discoveryv1.Endpoint{ + Addresses: []string{"fd00::1"}, + Hostname: new("foo"), + Conditions: discoveryv1.EndpointConditions{ + Serving: new(true), + Ready: new(true), + Terminating: new(false), + }, + }) + expectEqual(t, fc, epsV6) + }) } func configMapForSvc(t *testing.T, svc *corev1.Service, p uint16) *corev1.ConfigMap { @@ -157,7 +268,7 @@ func configMapForSvc(t *testing.T, svc *corev1.Service, p uint16) *corev1.Config return cm } -func serviceStatusForPodIP(t *testing.T, svc *corev1.Service, ip string, p uint16) []byte { +func serviceStatusForPodIPs(t *testing.T, svc *corev1.Service, ipv4, ipv6 string, p uint16) []byte { t.Helper() ports := make(map[egressservices.PortMap]struct{}) for _, port := range svc.Spec.Ports { @@ -172,7 +283,8 @@ func serviceStatusForPodIP(t *testing.T, svc *corev1.Service, ip string, p uint1 } svcName := tailnetSvcName(svc) st := egressservices.Status{ - PodIPv4: ip, + PodIPv4: ipv4, + PodIPv6: ipv6, Services: map[string]*egressservices.ServiceStatus{svcName: &svcSt}, } bs, err := json.Marshal(st) diff --git a/cmd/k8s-operator/egress-pod-readiness.go b/cmd/k8s-operator/egress-pod-readiness.go index 17ba0c5ec..acd68cb13 100644 --- a/cmd/k8s-operator/egress-pod-readiness.go +++ b/cmd/k8s-operator/egress-pod-readiness.go @@ -10,6 +10,7 @@ "errors" "fmt" "net/http" + "net/netip" "slices" "strings" "sync/atomic" @@ -227,13 +228,24 @@ func (er *egressPodsReconciler) lookupPodRouteViaSvc(ctx context.Context, pod *c lg.Debugf("Pod does not have health check enabled, unable to verify if it is currently routable via Service") return cannotVerify, nil } - wantsIP, err := podIPv4(pod) - if err != nil { - return -1, fmt.Errorf("error determining Pod's IP address: %w", err) - } - if wantsIP == "" { + // Use the Pod's primary IP (PodIPs[0]) to identify this Pod in the health check + // response. The primary IP family is determined by the cluster's IP family configuration. + + // Note: we do not control which IP family the request uses, so on a dual-stack + // cluster either IPv4 or IPv6 could be used. In either case, a matching IP header + // comfirms the request reached this Pod. + if len(pod.Status.PodIPs) == 0 || pod.Status.PodIPs[0].IP == "" { return podNotReady, nil } + wantsIP := pod.Status.PodIPs[0].IP + parsed, err := netip.ParseAddr(wantsIP) + if err != nil { + return -1, fmt.Errorf("error parsing Pod IP %q: %w", wantsIP, err) + } + header := kubetypes.PodIPv4Header + if parsed.Is6() { + header = kubetypes.PodIPv6Header + } ctx, cancel := context.WithTimeout(ctx, time.Second*3) defer cancel() @@ -250,7 +262,7 @@ func (er *egressPodsReconciler) lookupPodRouteViaSvc(ctx context.Context, pod *c return unreachable, nil } defer resp.Body.Close() - gotIP := resp.Header.Get(kubetypes.PodIPv4Header) + gotIP := resp.Header.Get(header) if gotIP == "" { lg.Debugf("Health check does not return Pod's IP header, unable to verify if Pod is currently routable via Service") return cannotVerify, nil diff --git a/cmd/k8s-operator/egress-pod-readiness_test.go b/cmd/k8s-operator/egress-pod-readiness_test.go index 0cf9108f5..6a087031b 100644 --- a/cmd/k8s-operator/egress-pod-readiness_test.go +++ b/cmd/k8s-operator/egress-pod-readiness_test.go @@ -420,6 +420,44 @@ func TestEgressPodReadiness(t *testing.T) { expectEqual(t, fc, pod) mustDeleteAll(t, fc, pod, svc, svc2, svc3) }) + t.Run("ipv6_only_pod_already_routed_to", func(t *testing.T) { + pod := podTemplate.DeepCopy() + pod.Status.PodIPs = []corev1.PodIP{{IP: "fd00::2"}} + + svc, hep := newSvc("svc", 9002) + mustCreateAll(t, fc, svc, pod) + resp := readyRespsV6("fd00::2", 1) + httpCl := fakeHTTPClient{ + t: t, + state: map[string][]fakeResponse{hep: resp}, + } + rec.httpClient = &httpCl + expectReconciled(t, rec, "operator-ns", pod.Name) + + podSetReady(pod, cl) + expectEqual(t, fc, pod) + mustDeleteAll(t, fc, pod, svc) + }) + t.Run("dual_stack_pod", func(t *testing.T) { + pod := podTemplate.DeepCopy() + pod.Status.PodIPs = []corev1.PodIP{{IP: "10.0.0.2"}, {IP: "fd00::2"}} + + svc, hep := newSvc("svc", 9002) + mustCreateAll(t, fc, svc, pod) + // Dual-stack pod: the reconciler uses PodIPs[0] (the primary IP), + // which in this case is IPv4. + resp := readyResps("10.0.0.2", 1) + httpCl := fakeHTTPClient{ + t: t, + state: map[string][]fakeResponse{hep: resp}, + } + rec.httpClient = &httpCl + expectReconciled(t, rec, "operator-ns", pod.Name) + + podSetReady(pod, cl) + expectEqual(t, fc, pod) + mustDeleteAll(t, fc, pod, svc) + }) } func readyResps(ip string, num int) (resps []fakeResponse) { @@ -429,6 +467,13 @@ func readyResps(ip string, num int) (resps []fakeResponse) { return resps } +func readyRespsV6(ip string, num int) (resps []fakeResponse) { + for range num { + resps = append(resps, fakeResponse{statusCode: 200, podIP: ip, header: kubetypes.PodIPv6Header}) + } + return resps +} + func unreadyResps(ip string, num int) (resps []fakeResponse) { for range num { resps = append(resps, fakeResponse{statusCode: 503, podIP: ip}) @@ -513,7 +558,11 @@ func (f *fakeHTTPClient) Do(req *http.Request) (*http.Response, error) { Header: make(http.Header), Body: io.NopCloser(bytes.NewReader([]byte{})), } - r.Header.Add(kubetypes.PodIPv4Header, resp.podIP) + h := kubetypes.PodIPv4Header + if resp.header != "" { + h = resp.header + } + r.Header.Add(h, resp.podIP) return &r, nil } @@ -521,4 +570,5 @@ type fakeResponse struct { err error statusCode int podIP string // for the Pod IP header + header string // header key to use; defaults to PodIPv4Header } diff --git a/cmd/k8s-operator/egress-services-readiness.go b/cmd/k8s-operator/egress-services-readiness.go index 54504af98..76a8fdb21 100644 --- a/cmd/k8s-operator/egress-services-readiness.go +++ b/cmd/k8s-operator/egress-services-readiness.go @@ -9,6 +9,7 @@ "context" "errors" "fmt" + "slices" "strings" "go.uber.org/zap" @@ -24,6 +25,7 @@ tsoperator "tailscale.com/k8s-operator" tsapi "tailscale.com/k8s-operator/apis/v1alpha1" "tailscale.com/tstime" + "tailscale.com/util/set" ) const ( @@ -72,19 +74,57 @@ func (esrr *egressSvcsReadinessReconciler) Reconcile(ctx context.Context, req re }() crl := egressSvcChildResourceLabels(svc) - eps, err := getSingleObject[discoveryv1.EndpointSlice](ctx, esrr.Client, esrr.tsNamespace, crl) - if err != nil { - err = fmt.Errorf("error getting EndpointSlice: %w", err) + epsList := &discoveryv1.EndpointSliceList{} + if err = esrr.List(ctx, epsList, client.InNamespace(esrr.tsNamespace), client.MatchingLabels(crl)); err != nil { + err = fmt.Errorf("error listing EndpointSlices: %w", err) reason = reasonReadinessCheckFailed msg = err.Error() return res, err } - if eps == nil { - lg.Infof("EndpointSlice for Service does not yet exist, waiting...") + if len(epsList.Items) == 0 { + lg.Infof("EndpointSlices for Service do not yet exist, waiting...") reason, msg = reasonClusterResourcesNotReady, reasonClusterResourcesNotReady st = metav1.ConditionFalse return res, nil } + // If an EndpointSlice for an expected family is missing, we mark the Service as NotReady. + // + // Setting the NotReady condition here is also used for best-effort recovery. The + // egress-svcs-reconciler does not watch EndpointSlices, so a deleted EndpointSlice is only + // recreated when this status change re-triggers a Service reconcile. + // + // TODO(beckypauley): refactor so EndpointSlice recovery is not dependent on Service status. + clusterIPSvc, err := getSingleObject[corev1.Service](ctx, esrr.Client, esrr.tsNamespace, crl) + if err != nil { + err = fmt.Errorf("error retrieving ClusterIP Service: %w", err) + reason = reasonReadinessCheckFailed + msg = err.Error() + return res, err + } + if clusterIPSvc == nil { + lg.Infof("ClusterIP Service for egress Service does not yet exist, waiting...") + reason, msg = reasonClusterResourcesNotReady, reasonClusterResourcesNotReady + st = metav1.ConditionFalse + return res, nil + } + gotAddrTypes := make(set.Set[discoveryv1.AddressType], len(epsList.Items)) + for _, eps := range epsList.Items { + gotAddrTypes.Add(eps.AddressType) + } + wantAddrTypes, err := addrTypesForClusterIPSvc(clusterIPSvc) + if err != nil { + reason = reasonReadinessCheckFailed + msg = err.Error() + return res, err + } + for _, wantAddrType := range wantAddrTypes { + if !gotAddrTypes.Contains(wantAddrType) { + lg.Infof("EndpointSlice for %s is missing, waiting...", wantAddrType) + reason, msg = reasonClusterResourcesNotReady, reasonClusterResourcesNotReady + st = metav1.ConditionFalse + return res, nil + } + } pg := &tsapi.ProxyGroup{ ObjectMeta: metav1.ObjectMeta{ Name: svc.Annotations[AnnotationProxyGroup], @@ -119,6 +159,7 @@ func (esrr *egressSvcsReadinessReconciler) Reconcile(ctx context.Context, req re } podLabels := pgLabels(pg.Name, nil) var readyReplicas int32 +nextReplica: for i := range replicas { podLabels[appsv1.PodIndexLabel] = fmt.Sprintf("%d", i) pod, err := getSingleObject[corev1.Pod](ctx, esrr.Client, esrr.tsNamespace, podLabels) @@ -136,18 +177,16 @@ func (esrr *egressSvcsReadinessReconciler) Reconcile(ctx context.Context, req re } lg.Debugf("looking at Pod with IPs %v", pod.Status.PodIPs) - ready := false - for _, ep := range eps.Endpoints { - lg.Debugf("looking at endpoint with addresses %v", ep.Addresses) - if endpointReadyForPod(&ep, pod, lg) { - lg.Debugf("endpoint is ready for Pod") - ready = true - break + for _, eps := range epsList.Items { + lg.Debugf("looking at %s EndpointSlice %s", eps.AddressType, eps.Name) + if !slices.ContainsFunc(eps.Endpoints, func(ep discoveryv1.Endpoint) bool { + return endpointReadyForPod(&ep, pod, eps.AddressType, lg) + }) { + continue nextReplica } } - if ready { - readyReplicas++ - } + lg.Debugf("endpoint is ready for Pod") + readyReplicas++ } msg = fmt.Sprintf(msgReadyToRouteTemplate, readyReplicas, replicas) if readyReplicas == 0 { @@ -164,12 +203,15 @@ func (esrr *egressSvcsReadinessReconciler) Reconcile(ctx context.Context, req re return res, nil } -// endpointReadyForPod returns true if the endpoint is for the Pod's IPv4 address and is ready to serve traffic. -// Endpoint must not be nil. -func endpointReadyForPod(ep *discoveryv1.Endpoint, pod *corev1.Pod, lg *zap.SugaredLogger) bool { - podIP, err := podIPv4(pod) +// endpointReadyForPod returns true if the endpoint is for the Pod's address (for the given address family) +// and is ready to serve traffic. Endpoint must not be nil. +func endpointReadyForPod(ep *discoveryv1.Endpoint, pod *corev1.Pod, addrType discoveryv1.AddressType, lg *zap.SugaredLogger) bool { + podIP, err := podIPForFamily(pod, addrType) if err != nil { - lg.Warnf("error retrieving Pod's IPv4 address: %v", err) + lg.Warnf("error retrieving Pod's %s address: %v", addrType, err) + return false + } + if podIP == "" { return false } diff --git a/cmd/k8s-operator/egress-services-readiness_test.go b/cmd/k8s-operator/egress-services-readiness_test.go index 96d76cc4e..c21a111d5 100644 --- a/cmd/k8s-operator/egress-services-readiness_test.go +++ b/cmd/k8s-operator/egress-services-readiness_test.go @@ -47,7 +47,14 @@ func TestEgressServiceReadiness(t *testing.T) { }, }, } - fakeClusterIPSvc := &corev1.Service{ObjectMeta: metav1.ObjectMeta{Name: "my-app", Namespace: "operator-ns"}} + fakeClusterIPSvc := &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Name: "my-app", + Namespace: "operator-ns", + Labels: egressSvcChildResourceLabels(egressSvc), + }, + Spec: corev1.ServiceSpec{ClusterIPs: []string{"10.0.0.1"}}, + } labels := egressSvcEpsLabels(egressSvc, fakeClusterIPSvc) eps := &discoveryv1.EndpointSlice{ ObjectMeta: metav1.ObjectMeta{ @@ -63,6 +70,7 @@ func TestEgressServiceReadiness(t *testing.T) { }, } mustCreate(t, fc, egressSvc) + mustCreate(t, fc, fakeClusterIPSvc) setClusterNotReady(egressSvc, cl, zl.Sugar()) t.Run("endpointslice_does_not_exist", func(t *testing.T) { expectReconciled(t, rec, "dev", "my-app") @@ -117,6 +125,212 @@ func TestEgressServiceReadiness(t *testing.T) { }) } +func TestEgressServiceReadinessDualStack(t *testing.T) { + fc := fake.NewClientBuilder(). + WithScheme(tsapi.GlobalScheme). + WithStatusSubresource(&tsapi.ProxyGroup{}). + Build() + zl, _ := zap.NewDevelopment() + cl := tstest.NewClock(tstest.ClockOpts{}) + rec := &egressSvcsReadinessReconciler{ + tsNamespace: "operator-ns", + Client: fc, + logger: zl.Sugar(), + clock: cl, + } + tailnetFQDN := "my-app.tailnetxyz.ts.net" + egressSvc := &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Name: "my-app", + Namespace: "dev", + Annotations: map[string]string{ + AnnotationProxyGroup: "dev", + AnnotationTailnetTargetFQDN: tailnetFQDN, + }, + }, + } + fakeClusterIPSvc := &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Name: "my-app", + Namespace: "operator-ns", + Labels: egressSvcChildResourceLabels(egressSvc), + }, + Spec: corev1.ServiceSpec{ClusterIPs: []string{"10.0.0.1", "fd00::1"}}, + } + labels := egressSvcEpsLabels(egressSvc, fakeClusterIPSvc) + epsV4 := &discoveryv1.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{ + Name: "my-app-ipv4", + Namespace: "operator-ns", + Labels: labels, + }, + AddressType: discoveryv1.AddressTypeIPv4, + } + labelsV6 := egressSvcEpsLabels(egressSvc, fakeClusterIPSvc) + epsV6 := &discoveryv1.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{ + Name: "my-app-ipv6", + Namespace: "operator-ns", + Labels: labelsV6, + }, + AddressType: discoveryv1.AddressTypeIPv6, + } + pg := &tsapi.ProxyGroup{ + ObjectMeta: metav1.ObjectMeta{ + Name: "dev", + }, + Spec: tsapi.ProxyGroupSpec{ + Replicas: new(int32(1)), + Type: tsapi.ProxyGroupTypeEgress, + }, + } + mustCreate(t, fc, egressSvc) + mustCreate(t, fc, fakeClusterIPSvc) + mustCreate(t, fc, epsV4) + mustCreate(t, fc, epsV6) + mustCreate(t, fc, pg) + setPGReady(pg, cl, zl.Sugar()) + mustUpdateStatus(t, fc, pg.Namespace, pg.Name, func(p *tsapi.ProxyGroup) { + p.Status = pg.Status + }) + + // Create a dual-stack pod. + p := pod(pg, 0) + p.Status.PodIPs = append(p.Status.PodIPs, corev1.PodIP{IP: "fd00::0"}) + mustCreate(t, fc, p) + mustUpdateStatus(t, fc, p.Namespace, p.Name, func(existing *corev1.Pod) { + existing.Status.PodIPs = p.Status.PodIPs + }) + + t.Run("not_ready_missing_from_ipv6_slice", func(t *testing.T) { + setEndpointForReplicaWithIP("10.0.0.0", epsV4) + mustUpdate(t, fc, epsV4.Namespace, epsV4.Name, func(e *discoveryv1.EndpointSlice) { + e.Endpoints = epsV4.Endpoints + }) + expectReconciled(t, rec, "dev", "my-app") + setNotReady(egressSvc, cl, zl.Sugar(), pgReplicas(pg)) + expectEqual(t, fc, egressSvc) + }) + t.Run("ready_in_both_slices", func(t *testing.T) { + setEndpointForReplicaWithIP("fd00::", epsV6) + mustUpdate(t, fc, epsV6.Namespace, epsV6.Name, func(e *discoveryv1.EndpointSlice) { + e.Endpoints = epsV6.Endpoints + }) + expectReconciled(t, rec, "dev", "my-app") + setReady(egressSvc, cl, zl.Sugar(), pgReplicas(pg), pgReplicas(pg)) + expectEqual(t, fc, egressSvc) + }) + t.Run("not_ready_when_ipv6_slice_missing", func(t *testing.T) { + // Delete the IPv6 EndpointSlice while the ClusterIP Service still + // wants an IPv6 family; the Service should report NotReady even though + // the IPv4 EndpointSlice is healthy. + if err := fc.Delete(t.Context(), epsV6); err != nil { + t.Fatalf("error deleting IPv6 EndpointSlice: %v", err) + } + expectReconciled(t, rec, "dev", "my-app") + setClusterNotReady(egressSvc, cl, zl.Sugar()) + expectEqual(t, fc, egressSvc) + }) +} + +func TestEgressServiceReadinessIPv6Only(t *testing.T) { + fc := fake.NewClientBuilder(). + WithScheme(tsapi.GlobalScheme). + WithStatusSubresource(&tsapi.ProxyGroup{}). + Build() + zl, _ := zap.NewDevelopment() + cl := tstest.NewClock(tstest.ClockOpts{}) + rec := &egressSvcsReadinessReconciler{ + tsNamespace: "operator-ns", + Client: fc, + logger: zl.Sugar(), + clock: cl, + } + egressSvc := &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Name: "my-app", + Namespace: "dev", + Annotations: map[string]string{ + AnnotationProxyGroup: "dev", + AnnotationTailnetTargetFQDN: "my-app.tailnetxyz.ts.net", + }, + }, + } + fakeClusterIPSvc := &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Name: "my-app", + Namespace: "operator-ns", + Labels: egressSvcChildResourceLabels(egressSvc), + }, + Spec: corev1.ServiceSpec{ClusterIPs: []string{"fd00::1"}}, + } + labels := egressSvcEpsLabels(egressSvc, fakeClusterIPSvc) + eps := &discoveryv1.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{ + Name: "my-app-ipv6", + Namespace: "operator-ns", + Labels: labels, + }, + AddressType: discoveryv1.AddressTypeIPv6, + } + pg := &tsapi.ProxyGroup{ + ObjectMeta: metav1.ObjectMeta{ + Name: "dev", + }, + } + mustCreate(t, fc, egressSvc) + mustCreate(t, fc, fakeClusterIPSvc) + mustCreate(t, fc, eps) + mustCreate(t, fc, pg) + setPGReady(pg, cl, zl.Sugar()) + mustUpdateStatus(t, fc, pg.Namespace, pg.Name, func(p *tsapi.ProxyGroup) { + p.Status = pg.Status + }) + + // Create IPv6-only pods. + for i := range pgReplicas(pg) { + p := ipv6OnlyPod(pg, i) + mustCreate(t, fc, p) + mustUpdateStatus(t, fc, p.Namespace, p.Name, func(existing *corev1.Pod) { + existing.Status.PodIPs = p.Status.PodIPs + }) + } + + t.Run("no_ready_replicas", func(t *testing.T) { + expectReconciled(t, rec, "dev", "my-app") + setNotReady(egressSvc, cl, zl.Sugar(), pgReplicas(pg)) + expectEqual(t, fc, egressSvc) + }) + t.Run("all_replicas_ready", func(t *testing.T) { + for i := range pgReplicas(pg) { + p := ipv6OnlyPod(pg, i) + setEndpointForReplicaWithIP(p.Status.PodIPs[0].IP, eps) + } + mustUpdate(t, fc, eps.Namespace, eps.Name, func(e *discoveryv1.EndpointSlice) { + e.Endpoints = eps.Endpoints + }) + setReady(egressSvc, cl, zl.Sugar(), pgReplicas(pg), pgReplicas(pg)) + expectReconciled(t, rec, "dev", "my-app") + expectEqual(t, fc, egressSvc) + }) +} + +func ipv6OnlyPod(pg *tsapi.ProxyGroup, ordinal int32) *corev1.Pod { + labels := pgLabels(pg.Name, nil) + labels[appsv1.PodIndexLabel] = fmt.Sprintf("%d", ordinal) + ip := fmt.Sprintf("fd00::%d", ordinal+1) // +1 to avoid fd00::0 normalization issues + return &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: fmt.Sprintf("%s-%d", pg.Name, ordinal), + Namespace: "operator-ns", + Labels: labels, + }, + Status: corev1.PodStatus{ + PodIPs: []corev1.PodIP{{IP: ip}}, + }, + } +} + func setClusterNotReady(svc *corev1.Service, cl tstime.Clock, lg *zap.SugaredLogger) { tsoperator.SetServiceCondition(svc, tsapi.EgressSvcReady, metav1.ConditionFalse, reasonClusterResourcesNotReady, reasonClusterResourcesNotReady, cl, lg) } @@ -166,3 +380,14 @@ func pod(pg *tsapi.ProxyGroup, ordinal int32) *corev1.Pod { }, } } + +func setEndpointForReplicaWithIP(ip string, eps *discoveryv1.EndpointSlice) { + eps.Endpoints = append(eps.Endpoints, discoveryv1.Endpoint{ + Addresses: []string{ip}, + Conditions: discoveryv1.EndpointConditions{ + Ready: new(true), + Serving: new(true), + Terminating: new(false), + }, + }) +} diff --git a/cmd/k8s-operator/egress-services.go b/cmd/k8s-operator/egress-services.go index b61f0ad1f..f935fad64 100644 --- a/cmd/k8s-operator/egress-services.go +++ b/cmd/k8s-operator/egress-services.go @@ -12,6 +12,7 @@ "errors" "fmt" "math/rand/v2" + "net/netip" "reflect" "slices" "strings" @@ -222,28 +223,56 @@ func (esr *egressSvcsReconciler) maybeProvision(ctx context.Context, svc *corev1 return nil } -// ensureEndpointSlices ensures that an IPv4 EndpointSlice exists for the egress -// service and that its ports are up to date. +// addrTypesForClusterIPSvc returns the EndpointSlice address types (IP families) +// that the given ClusterIP Service supports, derived from its ClusterIPs. +// TODO(beckypauley): this could read Spec.IPFamilies directly instead of parsing +// ClusterIPs to determine the family. +func addrTypesForClusterIPSvc(clusterIPSvc *corev1.Service) ([]discoveryv1.AddressType, error) { + addrTypes := make([]discoveryv1.AddressType, 0, len(clusterIPSvc.Spec.ClusterIPs)) + for _, clusterIP := range clusterIPSvc.Spec.ClusterIPs { + ip, err := netip.ParseAddr(clusterIP) + if err != nil { + return nil, fmt.Errorf("error parsing ClusterIP %q: %w", clusterIP, err) + } + addrType := discoveryv1.AddressTypeIPv4 + if ip.Is6() { + addrType = discoveryv1.AddressTypeIPv6 + } + addrTypes = append(addrTypes, addrType) + } + return addrTypes, nil +} + +// ensureEndpointSlices ensures that EndpointSlices exist for the egress service +// for each IP family supported by the cluster, and that their ports are up to +// date. func (esr *egressSvcsReconciler) ensureEndpointSlices(ctx context.Context, svc, clusterIPSvc *corev1.Service, lg *zap.SugaredLogger) error { crl := egressSvcEpsLabels(svc, clusterIPSvc) - eps := &discoveryv1.EndpointSlice{ - ObjectMeta: metav1.ObjectMeta{ - Name: fmt.Sprintf("%s-ipv4", clusterIPSvc.Name), - Namespace: esr.tsNamespace, - Labels: crl, - }, - AddressType: discoveryv1.AddressTypeIPv4, - Ports: epsPortsFromSvc(clusterIPSvc), + // Only create EndpointSlices for IP families supported by the cluster. + addrTypes, err := addrTypesForClusterIPSvc(clusterIPSvc) + if err != nil { + return err } - if _, err := createOrUpdate(ctx, esr.Client, esr.tsNamespace, eps, func(e *discoveryv1.EndpointSlice) { - e.Labels = eps.Labels - e.AddressType = eps.AddressType - e.Ports = eps.Ports - for _, p := range e.Endpoints { - p.Conditions.Ready = nil + for _, addrType := range addrTypes { + eps := &discoveryv1.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{ + Name: fmt.Sprintf("%s-%s", clusterIPSvc.Name, strings.ToLower(string(addrType))), + Namespace: esr.tsNamespace, + Labels: crl, + }, + AddressType: addrType, + Ports: epsPortsFromSvc(clusterIPSvc), + } + if _, err := createOrUpdate(ctx, esr.Client, esr.tsNamespace, eps, func(e *discoveryv1.EndpointSlice) { + e.Labels = eps.Labels + e.AddressType = eps.AddressType + e.Ports = eps.Ports + for _, p := range e.Endpoints { + p.Conditions.Ready = nil + } + }); err != nil { + return fmt.Errorf("error ensuring %s EndpointSlice: %w", addrType, err) } - }); err != nil { - return fmt.Errorf("error ensuring EndpointSlice: %w", err) } return nil } diff --git a/cmd/k8s-operator/egress-services_test.go b/cmd/k8s-operator/egress-services_test.go index 9017d4807..20efe195d 100644 --- a/cmd/k8s-operator/egress-services_test.go +++ b/cmd/k8s-operator/egress-services_test.go @@ -21,6 +21,7 @@ "k8s.io/apimachinery/pkg/util/intstr" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" tsapi "tailscale.com/k8s-operator/apis/v1alpha1" "tailscale.com/kube/egressservices" @@ -50,6 +51,9 @@ func TestTailscaleEgressServices(t *testing.T) { WithScheme(tsapi.GlobalScheme). WithObjects(pg, cm). WithStatusSubresource(pg). + WithInterceptorFuncs(interceptor.Funcs{ + Create: clusterIPInterceptor("10.96.0.1"), + }). Build() zl, err := zap.NewDevelopment() if err != nil { @@ -152,10 +156,10 @@ func validateReadyService(t *testing.T, fc client.WithWatch, esr *egressSvcsReco expectReconciled(t, esr, "default", "test") // Verify that a ClusterIP Service has been created. name := findGenNameForEgressSvcResources(t, fc, svc) - expectEqual(t, fc, clusterIPSvc(name, svc), removeTargetPortsFromSvc) + expectEqual(t, fc, clusterIPSvc(name, svc), removeTargetPortsFromSvc, removeClusterIPsFromSvc) clusterSvc := mustGetClusterIPSvc(t, fc, name) // Verify that an EndpointSlice has been created. - expectEqual(t, fc, endpointSlice(name, svc, clusterSvc)) + expectEqual(t, fc, endpointSlice(name, svc, clusterSvc, discoveryv1.AddressTypeIPv4)) // Verify that ConfigMap contains configuration for the new egress service. mustHaveConfigForSvc(t, fc, svc, clusterSvc, cm, zl) r := svcConfiguredReason(svc, true, zl.Sugar()) @@ -241,18 +245,22 @@ func mustGetClusterIPSvc(t *testing.T, cl client.Client, name string) *corev1.Se return svc } -func endpointSlice(name string, extNSvc, clusterIPSvc *corev1.Service) *discoveryv1.EndpointSlice { +func endpointSlice(name string, extNSvc, clusterIPSvc *corev1.Service, addrType discoveryv1.AddressType) *discoveryv1.EndpointSlice { labels := egressSvcChildResourceLabels(extNSvc) labels[discoveryv1.LabelManagedBy] = "tailscale.com" labels[discoveryv1.LabelServiceName] = name + suffix := "ipv4" + if addrType == discoveryv1.AddressTypeIPv6 { + suffix = "ipv6" + } return &discoveryv1.EndpointSlice{ ObjectMeta: metav1.ObjectMeta{ - Name: fmt.Sprintf("%s-ipv4", name), + Name: fmt.Sprintf("%s-%s", name, suffix), Namespace: "operator-ns", Labels: labels, }, Ports: portsForEndpointSlice(clusterIPSvc), - AddressType: discoveryv1.AddressTypeIPv4, + AddressType: addrType, } } @@ -312,3 +320,145 @@ func configFromCM(t *testing.T, cm *corev1.ConfigMap, svcName string) *egressser } return nil } + +func TestTailscaleEgressServicesDualStack(t *testing.T) { + pg := &tsapi.ProxyGroup{ + TypeMeta: metav1.TypeMeta{Kind: "ProxyGroup", APIVersion: "tailscale.com/v1alpha1"}, + ObjectMeta: metav1.ObjectMeta{ + Name: "foo", + UID: types.UID("1234-UID"), + }, + Spec: tsapi.ProxyGroupSpec{ + Replicas: pointer.To[int32](3), + Type: tsapi.ProxyGroupTypeEgress, + }, + } + cm := &corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{ + Name: pgEgressCMName("foo"), + Namespace: "operator-ns", + }, + } + fc := fake.NewClientBuilder(). + WithScheme(tsapi.GlobalScheme). + WithObjects(pg, cm). + WithStatusSubresource(pg). + WithInterceptorFuncs(interceptor.Funcs{ + Create: clusterIPInterceptor("10.96.0.1", "fd00::1"), + }). + Build() + zl, err := zap.NewDevelopment() + if err != nil { + t.Fatal(err) + } + clock := tstest.NewClock(tstest.ClockOpts{}) + + esr := &egressSvcsReconciler{ + Client: fc, + logger: zl.Sugar(), + clock: clock, + tsNamespace: "operator-ns", + } + svc := &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test", + Namespace: "default", + UID: types.UID("1234-UID"), + Annotations: map[string]string{ + AnnotationTailnetTargetFQDN: "foo.bar.ts.net.", + AnnotationProxyGroup: "foo", + }, + }, + Spec: corev1.ServiceSpec{ + ExternalName: "placeholder", + Type: corev1.ServiceTypeExternalName, + Selector: nil, + Ports: []corev1.ServicePort{ + { + Protocol: "TCP", + Port: 80, + }, + }, + }, + } + + t.Run("dual_stack_creates_both_endpoint_slices", func(t *testing.T) { + mustCreate(t, fc, svc) + expectReconciled(t, esr, "default", "test") + validateReadyService(t, fc, esr, svc, clock, zl, cm) + // Also verify the IPv6 EndpointSlice was created. + name := findGenNameForEgressSvcResources(t, fc, svc) + clusterSvc := mustGetClusterIPSvc(t, fc, name) + expectEqual(t, fc, endpointSlice(name, svc, clusterSvc, discoveryv1.AddressTypeIPv6)) + }) + + t.Run("dual_stack_endpointslice_deletion_recovery", func(t *testing.T) { + name := findGenNameForEgressSvcResources(t, fc, svc) + // Delete both IPv4 and IPv6 EndpointSlices. + for _, suffix := range []string{"ipv4", "ipv6"} { + epsName := fmt.Sprintf("%s-%s", name, suffix) + eps := &discoveryv1.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{ + Name: epsName, + Namespace: "operator-ns", + }, + } + if err := fc.Delete(t.Context(), eps); err != nil { + t.Fatalf("error deleting EndpointSlice %s: %v", epsName, err) + } + expectMissing[discoveryv1.EndpointSlice](t, fc, "operator-ns", epsName) + } + // Reconcile should recreate both. + validateReadyService(t, fc, esr, svc, clock, zl, cm) + clusterSvc := mustGetClusterIPSvc(t, fc, name) + expectEqual(t, fc, endpointSlice(name, svc, clusterSvc, discoveryv1.AddressTypeIPv6)) + }) + + t.Run("dual_stack_single_endpointslice_deletion_recovery", func(t *testing.T) { + name := findGenNameForEgressSvcResources(t, fc, svc) + // Delete only the IPv6 EndpointSlice. + epsName := fmt.Sprintf("%s-ipv6", name) + eps := &discoveryv1.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{ + Name: epsName, + Namespace: "operator-ns", + }, + } + if err := fc.Delete(t.Context(), eps); err != nil { + t.Fatalf("error deleting EndpointSlice %s: %v", epsName, err) + } + expectMissing[discoveryv1.EndpointSlice](t, fc, "operator-ns", epsName) + // Reconcile should recreate the missing IPv6 EndpointSlice while leaving + // the IPv4 one untouched. + validateReadyService(t, fc, esr, svc, clock, zl, cm) + clusterSvc := mustGetClusterIPSvc(t, fc, name) + expectEqual(t, fc, endpointSlice(name, svc, clusterSvc, discoveryv1.AddressTypeIPv6)) + expectEqual(t, fc, endpointSlice(name, svc, clusterSvc, discoveryv1.AddressTypeIPv4)) + }) + + t.Run("delete_dual_stack_service", func(t *testing.T) { + name := findGenNameForEgressSvcResources(t, fc, svc) + if err := fc.Delete(context.Background(), svc); err != nil { + t.Fatalf("error deleting ExternalName Service: %v", err) + } + expectReconciled(t, esr, "default", "test") + expectMissing[corev1.Service](t, fc, "operator-ns", name) + expectMissing[discoveryv1.EndpointSlice](t, fc, "operator-ns", fmt.Sprintf("%s-ipv4", name)) + expectMissing[discoveryv1.EndpointSlice](t, fc, "operator-ns", fmt.Sprintf("%s-ipv6", name)) + mustNotHaveConfigForSvc(t, fc, svc, cm) + }) +} + +// clusterIPInterceptor returns an interceptor.Funcs Create function that +// simulates the API server assigning ClusterIPs to ClusterIP Services. +// This is required because the reconciler iterates ClusterIPs to create +// per-family EndpointSlices but the fake client does not assign ClusterIPs. +func clusterIPInterceptor(clusterIPs ...string) func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.CreateOption) error { + return func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.CreateOption) error { + if svc, ok := obj.(*corev1.Service); ok && svc.Spec.Type == corev1.ServiceTypeClusterIP { + svc.Spec.ClusterIPs = clusterIPs + svc.Spec.ClusterIP = clusterIPs[0] + } + return c.Create(ctx, obj, opts...) + } +} diff --git a/cmd/k8s-operator/testutils_test.go b/cmd/k8s-operator/testutils_test.go index 074d92094..6c0d77a8c 100644 --- a/cmd/k8s-operator/testutils_test.go +++ b/cmd/k8s-operator/testutils_test.go @@ -984,6 +984,11 @@ func removeTargetPortsFromSvc(svc *corev1.Service) { svc.Spec.Ports = newPorts } +func removeClusterIPsFromSvc(svc *corev1.Service) { + svc.Spec.ClusterIP = "" + svc.Spec.ClusterIPs = nil +} + func removeAuthKeyIfExistsModifier(t *testing.T) func(s *corev1.Secret) { return func(secret *corev1.Secret) { t.Helper() diff --git a/cmd/k8s-proxy/k8s-proxy.go b/cmd/k8s-proxy/k8s-proxy.go index 673493f58..8d9b9faf0 100644 --- a/cmd/k8s-proxy/k8s-proxy.go +++ b/cmd/k8s-proxy/k8s-proxy.go @@ -301,8 +301,15 @@ func run(logger *zap.SugaredLogger) error { } if cfg.Parsed.HealthCheckEnabled.EqualBool(true) { - ipV4, _ := ts.TailscaleIPs() - hz := healthz.RegisterHealthHandlers(mux, ipV4.String(), logger.Infof) + ipV4, ipV6 := ts.TailscaleIPs() + var v4, v6 string + if ipV4.IsValid() { + v4 = ipV4.String() + } + if ipV6.IsValid() { + v6 = ipV6.String() + } + hz := healthz.RegisterHealthHandlers(mux, v4, v6, logger.Infof) group.Go(func() error { err := hz.MonitorHealth(ctx, lc) if err == nil || errors.Is(err, context.Canceled) { diff --git a/kube/egressservices/egressservices.go b/kube/egressservices/egressservices.go index 3828760af..0aeaa9421 100644 --- a/kube/egressservices/egressservices.go +++ b/kube/egressservices/egressservices.go @@ -94,6 +94,7 @@ func (p PortMaps) MarshalJSON() ([]byte, error) { // services for a proxy identified by the PodIP. type Status struct { PodIPv4 string `json:"podIPv4"` + PodIPv6 string `json:"podIPv6,omitempty"` // All egress service status keyed by service name. Services map[string]*ServiceStatus `json:"services"` } diff --git a/kube/health/healthz.go b/kube/health/healthz.go index e9b459fc1..8909e79a4 100644 --- a/kube/health/healthz.go +++ b/kube/health/healthz.go @@ -26,6 +26,7 @@ type Healthz struct { sync.Mutex hasAddrs bool podIPv4 string + podIPv6 string logger logger.Logf } @@ -34,7 +35,12 @@ func (h *Healthz) ServeHTTP(w http.ResponseWriter, r *http.Request) { defer h.Unlock() if h.hasAddrs { - w.Header().Add(kubetypes.PodIPv4Header, h.podIPv4) + if h.podIPv4 != "" { + w.Header().Set(kubetypes.PodIPv4Header, h.podIPv4) + } + if h.podIPv6 != "" { + w.Header().Set(kubetypes.PodIPv6Header, h.podIPv6) + } if _, err := w.Write([]byte("ok")); err != nil { http.Error(w, fmt.Sprintf("error writing status: %v", err), http.StatusInternalServerError) } @@ -74,9 +80,10 @@ func (h *Healthz) MonitorHealth(ctx context.Context, lc *local.Client) error { // RegisterHealthHandlers registers a simple health handler at /healthz. // A containerized tailscale instance is considered healthy if // it has at least one tailnet IP address. -func RegisterHealthHandlers(mux *http.ServeMux, podIPv4 string, logger logger.Logf) *Healthz { +func RegisterHealthHandlers(mux *http.ServeMux, podIPv4, podIPv6 string, logger logger.Logf) *Healthz { h := &Healthz{ podIPv4: podIPv4, + podIPv6: podIPv6, logger: logger, } mux.Handle("GET /healthz", h) diff --git a/kube/kubetypes/types.go b/kube/kubetypes/types.go index 9f1b29064..02ad336e4 100644 --- a/kube/kubetypes/types.go +++ b/kube/kubetypes/types.go @@ -52,6 +52,8 @@ // Pod's IPv4 address header key as returned by containerboot health check endpoint. PodIPv4Header string = "Pod-IPv4" + // Pod's IPv6 address header key as returned by containerboot health check endpoint. + PodIPv6Header string = "Pod-IPv6" EgessServicesPreshutdownEP = "/internal-egress-services-preshutdown"