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 <becky@tailscale.com>

* 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 <becky@tailscale.com>

---------

Signed-off-by: Becky Pauley <becky@tailscale.com>
This commit is contained in:
BeckyPauley
2026-07-14 13:26:40 +01:00
committed by GitHub
parent 9711883a2f
commit 6ee7bcb458
16 changed files with 774 additions and 93 deletions
+43 -18
View File
@@ -57,7 +57,8 @@ type egressProxy struct {
netmapChan chan netmapState // chan to receive netmap state updates on 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 // tailnetFQDNs is the egress service FQDN to tailnet IP mappings that
// were last used to configure firewall rules for this proxy. // were last used to configure firewall rules for this proxy.
@@ -138,6 +139,7 @@ type egressProxyRunOpts struct {
stateSecret string stateSecret string
netmapChan chan netmapState netmapChan chan netmapState
podIPv4 string podIPv4 string
podIPv6 string
tailnetAddrs []netip.Prefix tailnetAddrs []netip.Prefix
} }
@@ -150,6 +152,7 @@ func (ep *egressProxy) configure(opts egressProxyRunOpts) {
ep.stateSecret = opts.stateSecret ep.stateSecret = opts.stateSecret
ep.netmapChan = opts.netmapChan ep.netmapChan = opts.netmapChan
ep.podIPv4 = opts.podIPv4 ep.podIPv4 = opts.podIPv4
ep.podIPv6 = opts.podIPv6
ep.tailnetAddrs = opts.tailnetAddrs ep.tailnetAddrs = opts.tailnetAddrs
ep.client = &http.Client{} // default HTTP client ep.client = &http.Client{} // default HTTP client
sleepDuration := time.Second 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 { if err := json.Unmarshal([]byte(raw), status); err != nil {
return nil, fmt.Errorf("error unmarshalling previous config: %w", err) 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 status, nil
} }
return nil, nil return nil, nil
@@ -432,6 +435,7 @@ func (ep *egressProxy) setStatus(ctx context.Context, status *egressservices.Sta
status = &egressservices.Status{} status = &egressservices.Status{}
} }
status.PodIPv4 = ep.podIPv4 status.PodIPv4 = ep.podIPv4
status.PodIPv6 = ep.podIPv6
secret, err := ep.kc.GetSecret(ctx, ep.stateSecret) secret, err := ep.kc.GetSecret(ctx, ep.stateSecret)
if err != nil { if err != nil {
return fmt.Errorf("error retrieving state Secret: %w", err) return fmt.Errorf("error retrieving state Secret: %w", err)
@@ -622,6 +626,8 @@ func servicesStatusIsEqual(st, st1 *egressservices.Status) bool {
} }
st.PodIPv4 = "" st.PodIPv4 = ""
st1.PodIPv4 = "" st1.PodIPv4 = ""
st.PodIPv6 = ""
st1.PodIPv6 = ""
return reflect.DeepEqual(*st, *st1) return reflect.DeepEqual(*st, *st1)
} }
@@ -673,24 +679,29 @@ func (ep *egressProxy) waitTillSafeToShutdown(ctx context.Context, cfgs egressse
continue continue
} }
svc := s 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() { wg.Go(func() {
log.Printf("Ensuring that cluster traffic is no longer routed to %q via this Pod...", svc) 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 { 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) log.Printf("Cluster traffic for %s did not stop being routed to this Pod.", svc)
return 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 // lookupPodRoute calls the healthcheck endpoint repeat times and returns true if the endpoint returns with the podIP
// header at least once. // 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 { for range repeat {
f, err := lookup(ctx, hep, podIP, client) f, err := lookup(ctx, hep, podIP, podIPHeader, client)
if err != nil { if err != nil {
return false, err 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. // 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) req, err := http.NewRequestWithContext(ctx, httpm.GET, hep, nil)
if err != nil { if err != nil {
return false, fmt.Errorf("error creating new HTTP request: %v", err) 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 return true, nil
} }
defer resp.Body.Close() defer resp.Body.Close()
gotIP := resp.Header.Get(kubetypes.PodIPv4Header) gotIP := resp.Header.Get(podIPHeader)
return strings.EqualFold(podIP, gotIP), nil return strings.EqualFold(podIP, gotIP), nil
} }
@@ -762,3 +773,17 @@ func (ep *egressProxy) getHEPPings() (int, error) {
} }
return hp, nil 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
}
+3 -1
View File
@@ -15,6 +15,7 @@ import (
"strings" "strings"
"sync" "sync"
"testing" "testing"
"time"
"tailscale.com/kube/egressservices" "tailscale.com/kube/egressservices"
"tailscale.com/kube/kubetypes" "tailscale.com/kube/kubetypes"
@@ -269,7 +270,8 @@ func TestWaitTillSafeToShutdown(t *testing.T) {
} }
ep := &egressProxy{ ep := &egressProxy{
podIPv4: podIP, podIPv4: podIP,
shortSleep: time.Millisecond,
client: &mockHTTPClient{ client: &mockHTTPClient{
podIP: podIP, podIP: podIP,
anotherIP: anotherIP, anotherIP: anotherIP,
+3 -2
View File
@@ -423,7 +423,7 @@ func run() error {
mux := http.NewServeMux() mux := http.NewServeMux()
log.Printf("Running healthcheck endpoint at %s/healthz", cfg.HealthCheckAddrPort) 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) close := runHTTPServer(mux, cfg.HealthCheckAddrPort)
defer close() defer close()
@@ -439,7 +439,7 @@ func run() error {
if cfg.localHealthEnabled() { if cfg.localHealthEnabled() {
log.Printf("Running healthcheck endpoint at %s/healthz", cfg.LocalAddrPort) 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() { if cfg.egressSvcsTerminateEPEnabled() {
@@ -948,6 +948,7 @@ runLoop:
stateSecret: cfg.KubeSecret, stateSecret: cfg.KubeSecret,
netmapChan: egressSvcsNotify, netmapChan: egressSvcsNotify,
podIPv4: cfg.PodIPv4, podIPv4: cfg.PodIPv4,
podIPv6: cfg.PodIPv6,
tailnetAddrs: addrs, tailnetAddrs: addrs,
} }
go func() { go func() {
+23 -12
View File
@@ -106,16 +106,19 @@ func (er *egressEpsReconciler) Reconcile(ctx context.Context, req reconcile.Requ
} }
newEndpoints := make([]discoveryv1.Endpoint, 0) newEndpoints := make([]discoveryv1.Endpoint, 0)
for _, pod := range podList.Items { 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 { if err != nil {
return res, fmt.Errorf("error verifying if Pod is ready to route traffic: %w", err) return res, fmt.Errorf("error verifying if Pod is ready to route traffic: %w", err)
} }
if !ready { if !ready {
continue // maybe next time continue // maybe next time
} }
podIP, err := podIPv4(&pod) // we currently only support IPv4 podIP, err := podIPForFamily(&pod, eps.AddressType)
if err != nil { 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{ newEndpoints = append(newEndpoints, discoveryv1.Endpoint{
Hostname: (*string)(&pod.UID), Hostname: (*string)(&pod.UID),
@@ -140,13 +143,16 @@ func (er *egressEpsReconciler) Reconcile(ctx context.Context, req reconcile.Requ
return res, nil 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 { for _, ip := range pod.Status.PodIPs {
parsed, err := netip.ParseAddr(ip.IP) parsed, err := netip.ParseAddr(ip.IP)
if err != nil { if err != nil {
return "", fmt.Errorf("error parsing IP address %s: %w", ip, err) 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 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 // 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 // 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. // 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 = lg.With("proxy_pod", pod.Name)
lg.Debug("checking whether proxy is ready to route to egress service") lg.Debug("checking whether proxy is ready to route to egress service")
if !pod.DeletionTimestamp.IsZero() { if !pod.DeletionTimestamp.IsZero() {
lg.Debug("proxy Pod is being deleted, ignore") lg.Debug("proxy Pod is being deleted, ignore")
return false, nil return false, nil
} }
podIP, err := podIPForFamily(&pod, addrType)
podIP, err := podIPv4(&pod)
switch { switch {
case err != nil: case err != nil:
return false, fmt.Errorf("error determining Pod IP address: %v", err) return false, fmt.Errorf("error determining Pod IP address: %v", err)
case podIP == "": 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 return false, nil
} }
@@ -199,9 +204,15 @@ func (er *egressEpsReconciler) podIsReadyToRouteTraffic(ctx context.Context, pod
if err = json.Unmarshal(svcStatusBS, svcStatus); err != nil { if err = json.Unmarshal(svcStatusBS, svcStatus); err != nil {
return false, fmt.Errorf("error unmarshalling egress service status: %w", err) return false, fmt.Errorf("error unmarshalling egress service status: %w", err)
} }
var statusIP string
if !strings.EqualFold(podIP, svcStatus.PodIPv4) { switch addrType {
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) 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 return false, nil
} }
+117 -5
View File
@@ -98,7 +98,7 @@ func TestTailscaleEgressEndpointSlices(t *testing.T) {
t.Run("pods_are_ready_to_route_traffic", func(t *testing.T) { t.Run("pods_are_ready_to_route_traffic", func(t *testing.T) {
pod, stateS := podAndSecretForProxyGroup("foo") 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) { mustUpdate(t, fc, "operator-ns", stateS.Name, func(s *corev1.Secret) {
mak.Set(&s.Data, egressservices.KeyEgressServices, stBs) mak.Set(&s.Data, egressservices.KeyEgressServices, stBs)
}) })
@@ -115,8 +115,8 @@ func TestTailscaleEgressEndpointSlices(t *testing.T) {
expectEqual(t, fc, eps) expectEqual(t, fc, eps)
}) })
t.Run("status_does_not_match_pod_ip", func(t *testing.T) { t.Run("status_does_not_match_pod_ip", func(t *testing.T) {
_, stateS := podAndSecretForProxyGroup("foo") // replica Pod has IP 10.0.0.1 _, 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 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) { mustUpdate(t, fc, "operator-ns", stateS.Name, func(s *corev1.Secret) {
mak.Set(&s.Data, egressservices.KeyEgressServices, stBs) mak.Set(&s.Data, egressservices.KeyEgressServices, stBs)
}) })
@@ -124,6 +124,117 @@ func TestTailscaleEgressEndpointSlices(t *testing.T) {
eps.Endpoints = []discoveryv1.Endpoint{} eps.Endpoints = []discoveryv1.Endpoint{}
expectEqual(t, fc, eps) 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 { 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 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() t.Helper()
ports := make(map[egressservices.PortMap]struct{}) ports := make(map[egressservices.PortMap]struct{})
for _, port := range svc.Spec.Ports { 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) svcName := tailnetSvcName(svc)
st := egressservices.Status{ st := egressservices.Status{
PodIPv4: ip, PodIPv4: ipv4,
PodIPv6: ipv6,
Services: map[string]*egressservices.ServiceStatus{svcName: &svcSt}, Services: map[string]*egressservices.ServiceStatus{svcName: &svcSt},
} }
bs, err := json.Marshal(st) bs, err := json.Marshal(st)
+18 -6
View File
@@ -10,6 +10,7 @@ import (
"errors" "errors"
"fmt" "fmt"
"net/http" "net/http"
"net/netip"
"slices" "slices"
"strings" "strings"
"sync/atomic" "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") lg.Debugf("Pod does not have health check enabled, unable to verify if it is currently routable via Service")
return cannotVerify, nil return cannotVerify, nil
} }
wantsIP, err := podIPv4(pod) // Use the Pod's primary IP (PodIPs[0]) to identify this Pod in the health check
if err != nil { // response. The primary IP family is determined by the cluster's IP family configuration.
return -1, fmt.Errorf("error determining Pod's IP address: %w", err)
} // Note: we do not control which IP family the request uses, so on a dual-stack
if wantsIP == "" { // 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 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) ctx, cancel := context.WithTimeout(ctx, time.Second*3)
defer cancel() defer cancel()
@@ -250,7 +262,7 @@ func (er *egressPodsReconciler) lookupPodRouteViaSvc(ctx context.Context, pod *c
return unreachable, nil return unreachable, nil
} }
defer resp.Body.Close() defer resp.Body.Close()
gotIP := resp.Header.Get(kubetypes.PodIPv4Header) gotIP := resp.Header.Get(header)
if gotIP == "" { if gotIP == "" {
lg.Debugf("Health check does not return Pod's IP header, unable to verify if Pod is currently routable via Service") lg.Debugf("Health check does not return Pod's IP header, unable to verify if Pod is currently routable via Service")
return cannotVerify, nil return cannotVerify, nil
+51 -1
View File
@@ -420,6 +420,44 @@ func TestEgressPodReadiness(t *testing.T) {
expectEqual(t, fc, pod) expectEqual(t, fc, pod)
mustDeleteAll(t, fc, pod, svc, svc2, svc3) 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) { func readyResps(ip string, num int) (resps []fakeResponse) {
@@ -429,6 +467,13 @@ func readyResps(ip string, num int) (resps []fakeResponse) {
return resps 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) { func unreadyResps(ip string, num int) (resps []fakeResponse) {
for range num { for range num {
resps = append(resps, fakeResponse{statusCode: 503, podIP: ip}) 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), Header: make(http.Header),
Body: io.NopCloser(bytes.NewReader([]byte{})), 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 return &r, nil
} }
@@ -521,4 +570,5 @@ type fakeResponse struct {
err error err error
statusCode int statusCode int
podIP string // for the Pod IP header podIP string // for the Pod IP header
header string // header key to use; defaults to PodIPv4Header
} }
+62 -20
View File
@@ -9,6 +9,7 @@ import (
"context" "context"
"errors" "errors"
"fmt" "fmt"
"slices"
"strings" "strings"
"go.uber.org/zap" "go.uber.org/zap"
@@ -24,6 +25,7 @@ import (
tsoperator "tailscale.com/k8s-operator" tsoperator "tailscale.com/k8s-operator"
tsapi "tailscale.com/k8s-operator/apis/v1alpha1" tsapi "tailscale.com/k8s-operator/apis/v1alpha1"
"tailscale.com/tstime" "tailscale.com/tstime"
"tailscale.com/util/set"
) )
const ( const (
@@ -72,19 +74,57 @@ func (esrr *egressSvcsReadinessReconciler) Reconcile(ctx context.Context, req re
}() }()
crl := egressSvcChildResourceLabels(svc) crl := egressSvcChildResourceLabels(svc)
eps, err := getSingleObject[discoveryv1.EndpointSlice](ctx, esrr.Client, esrr.tsNamespace, crl) epsList := &discoveryv1.EndpointSliceList{}
if err != nil { if err = esrr.List(ctx, epsList, client.InNamespace(esrr.tsNamespace), client.MatchingLabels(crl)); err != nil {
err = fmt.Errorf("error getting EndpointSlice: %w", err) err = fmt.Errorf("error listing EndpointSlices: %w", err)
reason = reasonReadinessCheckFailed reason = reasonReadinessCheckFailed
msg = err.Error() msg = err.Error()
return res, err return res, err
} }
if eps == nil { if len(epsList.Items) == 0 {
lg.Infof("EndpointSlice for Service does not yet exist, waiting...") lg.Infof("EndpointSlices for Service do not yet exist, waiting...")
reason, msg = reasonClusterResourcesNotReady, reasonClusterResourcesNotReady reason, msg = reasonClusterResourcesNotReady, reasonClusterResourcesNotReady
st = metav1.ConditionFalse st = metav1.ConditionFalse
return res, nil 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{ pg := &tsapi.ProxyGroup{
ObjectMeta: metav1.ObjectMeta{ ObjectMeta: metav1.ObjectMeta{
Name: svc.Annotations[AnnotationProxyGroup], Name: svc.Annotations[AnnotationProxyGroup],
@@ -119,6 +159,7 @@ func (esrr *egressSvcsReadinessReconciler) Reconcile(ctx context.Context, req re
} }
podLabels := pgLabels(pg.Name, nil) podLabels := pgLabels(pg.Name, nil)
var readyReplicas int32 var readyReplicas int32
nextReplica:
for i := range replicas { for i := range replicas {
podLabels[appsv1.PodIndexLabel] = fmt.Sprintf("%d", i) podLabels[appsv1.PodIndexLabel] = fmt.Sprintf("%d", i)
pod, err := getSingleObject[corev1.Pod](ctx, esrr.Client, esrr.tsNamespace, podLabels) 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) lg.Debugf("looking at Pod with IPs %v", pod.Status.PodIPs)
ready := false for _, eps := range epsList.Items {
for _, ep := range eps.Endpoints { lg.Debugf("looking at %s EndpointSlice %s", eps.AddressType, eps.Name)
lg.Debugf("looking at endpoint with addresses %v", ep.Addresses) if !slices.ContainsFunc(eps.Endpoints, func(ep discoveryv1.Endpoint) bool {
if endpointReadyForPod(&ep, pod, lg) { return endpointReadyForPod(&ep, pod, eps.AddressType, lg)
lg.Debugf("endpoint is ready for Pod") }) {
ready = true continue nextReplica
break
} }
} }
if ready { lg.Debugf("endpoint is ready for Pod")
readyReplicas++ readyReplicas++
}
} }
msg = fmt.Sprintf(msgReadyToRouteTemplate, readyReplicas, replicas) msg = fmt.Sprintf(msgReadyToRouteTemplate, readyReplicas, replicas)
if readyReplicas == 0 { if readyReplicas == 0 {
@@ -164,12 +203,15 @@ func (esrr *egressSvcsReadinessReconciler) Reconcile(ctx context.Context, req re
return res, nil return res, nil
} }
// endpointReadyForPod returns true if the endpoint is for the Pod's IPv4 address and is ready to serve traffic. // endpointReadyForPod returns true if the endpoint is for the Pod's address (for the given address family)
// Endpoint must not be nil. // and is ready to serve traffic. Endpoint must not be nil.
func endpointReadyForPod(ep *discoveryv1.Endpoint, pod *corev1.Pod, lg *zap.SugaredLogger) bool { func endpointReadyForPod(ep *discoveryv1.Endpoint, pod *corev1.Pod, addrType discoveryv1.AddressType, lg *zap.SugaredLogger) bool {
podIP, err := podIPv4(pod) podIP, err := podIPForFamily(pod, addrType)
if err != nil { 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 return false
} }
@@ -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) labels := egressSvcEpsLabels(egressSvc, fakeClusterIPSvc)
eps := &discoveryv1.EndpointSlice{ eps := &discoveryv1.EndpointSlice{
ObjectMeta: metav1.ObjectMeta{ ObjectMeta: metav1.ObjectMeta{
@@ -63,6 +70,7 @@ func TestEgressServiceReadiness(t *testing.T) {
}, },
} }
mustCreate(t, fc, egressSvc) mustCreate(t, fc, egressSvc)
mustCreate(t, fc, fakeClusterIPSvc)
setClusterNotReady(egressSvc, cl, zl.Sugar()) setClusterNotReady(egressSvc, cl, zl.Sugar())
t.Run("endpointslice_does_not_exist", func(t *testing.T) { t.Run("endpointslice_does_not_exist", func(t *testing.T) {
expectReconciled(t, rec, "dev", "my-app") 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) { func setClusterNotReady(svc *corev1.Service, cl tstime.Clock, lg *zap.SugaredLogger) {
tsoperator.SetServiceCondition(svc, tsapi.EgressSvcReady, metav1.ConditionFalse, reasonClusterResourcesNotReady, reasonClusterResourcesNotReady, cl, lg) 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),
},
})
}
+47 -18
View File
@@ -12,6 +12,7 @@ import (
"errors" "errors"
"fmt" "fmt"
"math/rand/v2" "math/rand/v2"
"net/netip"
"reflect" "reflect"
"slices" "slices"
"strings" "strings"
@@ -222,28 +223,56 @@ func (esr *egressSvcsReconciler) maybeProvision(ctx context.Context, svc *corev1
return nil return nil
} }
// ensureEndpointSlices ensures that an IPv4 EndpointSlice exists for the egress // addrTypesForClusterIPSvc returns the EndpointSlice address types (IP families)
// service and that its ports are up to date. // 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 { func (esr *egressSvcsReconciler) ensureEndpointSlices(ctx context.Context, svc, clusterIPSvc *corev1.Service, lg *zap.SugaredLogger) error {
crl := egressSvcEpsLabels(svc, clusterIPSvc) crl := egressSvcEpsLabels(svc, clusterIPSvc)
eps := &discoveryv1.EndpointSlice{ // Only create EndpointSlices for IP families supported by the cluster.
ObjectMeta: metav1.ObjectMeta{ addrTypes, err := addrTypesForClusterIPSvc(clusterIPSvc)
Name: fmt.Sprintf("%s-ipv4", clusterIPSvc.Name), if err != nil {
Namespace: esr.tsNamespace, return err
Labels: crl,
},
AddressType: discoveryv1.AddressTypeIPv4,
Ports: epsPortsFromSvc(clusterIPSvc),
} }
if _, err := createOrUpdate(ctx, esr.Client, esr.tsNamespace, eps, func(e *discoveryv1.EndpointSlice) { for _, addrType := range addrTypes {
e.Labels = eps.Labels eps := &discoveryv1.EndpointSlice{
e.AddressType = eps.AddressType ObjectMeta: metav1.ObjectMeta{
e.Ports = eps.Ports Name: fmt.Sprintf("%s-%s", clusterIPSvc.Name, strings.ToLower(string(addrType))),
for _, p := range e.Endpoints { Namespace: esr.tsNamespace,
p.Conditions.Ready = nil 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 return nil
} }
+155 -5
View File
@@ -21,6 +21,7 @@ import (
"k8s.io/apimachinery/pkg/util/intstr" "k8s.io/apimachinery/pkg/util/intstr"
"sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/client/fake" "sigs.k8s.io/controller-runtime/pkg/client/fake"
"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
tsapi "tailscale.com/k8s-operator/apis/v1alpha1" tsapi "tailscale.com/k8s-operator/apis/v1alpha1"
"tailscale.com/kube/egressservices" "tailscale.com/kube/egressservices"
@@ -50,6 +51,9 @@ func TestTailscaleEgressServices(t *testing.T) {
WithScheme(tsapi.GlobalScheme). WithScheme(tsapi.GlobalScheme).
WithObjects(pg, cm). WithObjects(pg, cm).
WithStatusSubresource(pg). WithStatusSubresource(pg).
WithInterceptorFuncs(interceptor.Funcs{
Create: clusterIPInterceptor("10.96.0.1"),
}).
Build() Build()
zl, err := zap.NewDevelopment() zl, err := zap.NewDevelopment()
if err != nil { if err != nil {
@@ -152,10 +156,10 @@ func validateReadyService(t *testing.T, fc client.WithWatch, esr *egressSvcsReco
expectReconciled(t, esr, "default", "test") expectReconciled(t, esr, "default", "test")
// Verify that a ClusterIP Service has been created. // Verify that a ClusterIP Service has been created.
name := findGenNameForEgressSvcResources(t, fc, svc) 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) clusterSvc := mustGetClusterIPSvc(t, fc, name)
// Verify that an EndpointSlice has been created. // 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. // Verify that ConfigMap contains configuration for the new egress service.
mustHaveConfigForSvc(t, fc, svc, clusterSvc, cm, zl) mustHaveConfigForSvc(t, fc, svc, clusterSvc, cm, zl)
r := svcConfiguredReason(svc, true, zl.Sugar()) r := svcConfiguredReason(svc, true, zl.Sugar())
@@ -241,18 +245,22 @@ func mustGetClusterIPSvc(t *testing.T, cl client.Client, name string) *corev1.Se
return svc 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 := egressSvcChildResourceLabels(extNSvc)
labels[discoveryv1.LabelManagedBy] = "tailscale.com" labels[discoveryv1.LabelManagedBy] = "tailscale.com"
labels[discoveryv1.LabelServiceName] = name labels[discoveryv1.LabelServiceName] = name
suffix := "ipv4"
if addrType == discoveryv1.AddressTypeIPv6 {
suffix = "ipv6"
}
return &discoveryv1.EndpointSlice{ return &discoveryv1.EndpointSlice{
ObjectMeta: metav1.ObjectMeta{ ObjectMeta: metav1.ObjectMeta{
Name: fmt.Sprintf("%s-ipv4", name), Name: fmt.Sprintf("%s-%s", name, suffix),
Namespace: "operator-ns", Namespace: "operator-ns",
Labels: labels, Labels: labels,
}, },
Ports: portsForEndpointSlice(clusterIPSvc), 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 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...)
}
}
+5
View File
@@ -984,6 +984,11 @@ func removeTargetPortsFromSvc(svc *corev1.Service) {
svc.Spec.Ports = newPorts 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) { func removeAuthKeyIfExistsModifier(t *testing.T) func(s *corev1.Secret) {
return func(secret *corev1.Secret) { return func(secret *corev1.Secret) {
t.Helper() t.Helper()
+9 -2
View File
@@ -301,8 +301,15 @@ func run(logger *zap.SugaredLogger) error {
} }
if cfg.Parsed.HealthCheckEnabled.EqualBool(true) { if cfg.Parsed.HealthCheckEnabled.EqualBool(true) {
ipV4, _ := ts.TailscaleIPs() ipV4, ipV6 := ts.TailscaleIPs()
hz := healthz.RegisterHealthHandlers(mux, ipV4.String(), logger.Infof) 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 { group.Go(func() error {
err := hz.MonitorHealth(ctx, lc) err := hz.MonitorHealth(ctx, lc)
if err == nil || errors.Is(err, context.Canceled) { if err == nil || errors.Is(err, context.Canceled) {
+1
View File
@@ -94,6 +94,7 @@ func (p PortMaps) MarshalJSON() ([]byte, error) {
// services for a proxy identified by the PodIP. // services for a proxy identified by the PodIP.
type Status struct { type Status struct {
PodIPv4 string `json:"podIPv4"` PodIPv4 string `json:"podIPv4"`
PodIPv6 string `json:"podIPv6,omitempty"`
// All egress service status keyed by service name. // All egress service status keyed by service name.
Services map[string]*ServiceStatus `json:"services"` Services map[string]*ServiceStatus `json:"services"`
} }
+9 -2
View File
@@ -26,6 +26,7 @@ type Healthz struct {
sync.Mutex sync.Mutex
hasAddrs bool hasAddrs bool
podIPv4 string podIPv4 string
podIPv6 string
logger logger.Logf logger logger.Logf
} }
@@ -34,7 +35,12 @@ func (h *Healthz) ServeHTTP(w http.ResponseWriter, r *http.Request) {
defer h.Unlock() defer h.Unlock()
if h.hasAddrs { 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 { if _, err := w.Write([]byte("ok")); err != nil {
http.Error(w, fmt.Sprintf("error writing status: %v", err), http.StatusInternalServerError) 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. // RegisterHealthHandlers registers a simple health handler at /healthz.
// A containerized tailscale instance is considered healthy if // A containerized tailscale instance is considered healthy if
// it has at least one tailnet IP address. // 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{ h := &Healthz{
podIPv4: podIPv4, podIPv4: podIPv4,
podIPv6: podIPv6,
logger: logger, logger: logger,
} }
mux.Handle("GET /healthz", h) mux.Handle("GET /healthz", h)
+2
View File
@@ -52,6 +52,8 @@ const (
// Pod's IPv4 address header key as returned by containerboot health check endpoint. // Pod's IPv4 address header key as returned by containerboot health check endpoint.
PodIPv4Header string = "Pod-IPv4" 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" EgessServicesPreshutdownEP = "/internal-egress-services-preshutdown"