github/kubernetes
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fix Endpoint/EndpointSlice pod change detection

Browse Source 
The endpoint controllers responded to Pod changes by trying to figure
out if the generated endpoint resource would change, rather than just
checking if the Pod had changed, but since the set of Pod fields that
need to be checked depend on the Service and Node as well, the code
ended up only checking for a subset of the changes it should have.

In particular, EndpointSliceController ended up only looking at IPv4
Pod IPs when processing Pod update events, so when a Pod went from
having no IP to having only an IPv6 IP, EndpointSliceController would
think it hadn't changed.
This commit is contained in:
Dan Winship
2020-05-24 11:55:09 -04:00
parent f9ad7db9a6
commit 9fb6e2ef55
8 changed files with 217 additions and 310 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
pkg/controller/endpoint/endpoints_controller.go
View File

@@ -19,7 +19,6 @@ package endpoint
import (
"context"
"fmt"
"reflect"
"strconv"
"time"
@@ -213,39 +212,33 @@ func (e *EndpointController) addPod(obj interface{}) {
}
func podToEndpointAddressForService(svc *v1.Service, pod *v1.Pod) (*v1.EndpointAddress, error) {
var endpointIP string
if !utilfeature.DefaultFeatureGate.Enabled(features.IPv6DualStack) {
return podToEndpointAddress(pod), nil
}
endpointIP = pod.Status.PodIP
} else {
// api-server service controller ensured that the service got the correct IP Family
// according to user setup, here we only need to match EndPoint IPs' family to service
// actual IP family. as in, we don't need to check service.IPFamily
// api-server service controller ensured that the service got the correct IP Family
// according to user setup, here we only need to match EndPoint IPs' family to service
// actual IP family. as in, we don't need to check service.IPFamily
ipv6ClusterIP := utilnet.IsIPv6String(svc.Spec.ClusterIP)
for _, podIP := range pod.Status.PodIPs {
ipv6PodIP := utilnet.IsIPv6String(podIP.IP)
// same family?
// TODO (khenidak) when we remove the max of 2 PodIP limit from pods
// we will have to return multiple endpoint addresses
if ipv6ClusterIP == ipv6PodIP {
return &v1.EndpointAddress{
IP: podIP.IP,
NodeName: &pod.Spec.NodeName,
TargetRef: &v1.ObjectReference{
Kind: "Pod",
Namespace: pod.ObjectMeta.Namespace,
Name: pod.ObjectMeta.Name,
UID: pod.ObjectMeta.UID,
ResourceVersion: pod.ObjectMeta.ResourceVersion,
}}, nil
ipv6ClusterIP := utilnet.IsIPv6String(svc.Spec.ClusterIP)
for _, podIP := range pod.Status.PodIPs {
ipv6PodIP := utilnet.IsIPv6String(podIP.IP)
// same family?
// TODO (khenidak) when we remove the max of 2 PodIP limit from pods
// we will have to return multiple endpoint addresses
if ipv6ClusterIP == ipv6PodIP {
endpointIP = podIP.IP
break
}
}
if endpointIP == "" {
return nil, fmt.Errorf("failed to find a matching endpoint for service %v", svc.Name)
}
}
return nil, fmt.Errorf("failed to find a matching endpoint for service %v", svc.Name)
}
func podToEndpointAddress(pod *v1.Pod) *v1.EndpointAddress {
return &v1.EndpointAddress{
IP: pod.Status.PodIP,
IP: endpointIP,
NodeName: &pod.Spec.NodeName,
TargetRef: &v1.ObjectReference{
Kind: "Pod",
@@ -253,24 +246,15 @@ func podToEndpointAddress(pod *v1.Pod) *v1.EndpointAddress {
Name: pod.ObjectMeta.Name,
UID: pod.ObjectMeta.UID,
ResourceVersion: pod.ObjectMeta.ResourceVersion,
}}
}
func endpointChanged(pod1, pod2 *v1.Pod) bool {
endpointAddress1 := podToEndpointAddress(pod1)
endpointAddress2 := podToEndpointAddress(pod2)
endpointAddress1.TargetRef.ResourceVersion = ""
endpointAddress2.TargetRef.ResourceVersion = ""
return !reflect.DeepEqual(endpointAddress1, endpointAddress2)
},
}, nil
}
// When a pod is updated, figure out what services it used to be a member of
// and what services it will be a member of, and enqueue the union of these.
// old and cur must be *v1.Pod types.
func (e *EndpointController) updatePod(old, cur interface{}) {
services := endpointutil.GetServicesToUpdateOnPodChange(e.serviceLister, e.serviceSelectorCache, old, cur, endpointChanged)
services := endpointutil.GetServicesToUpdateOnPodChange(e.serviceLister, e.serviceSelectorCache, old, cur)
for key := range services {
e.queue.AddAfter(key, e.endpointUpdatesBatchPeriod)
}