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"