66 "time"
77
88 v1 "k8s.io/api/core/v1"
9+ discoveryv1 "k8s.io/api/discovery/v1"
910 "k8s.io/apimachinery/pkg/api/errors"
1011 metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
1112 "k8s.io/apimachinery/pkg/labels"
@@ -17,29 +18,40 @@ import (
1718 "github.com/kubeovn/kube-ovn/pkg/util"
1819)
1920
20- func (c * Controller ) enqueueAddEndpoint (obj any ) {
21- key := cache .MetaObjectToName (obj .(* v1.Endpoints )).String ()
22- klog .V (3 ).Infof ("enqueue add endpoint %s" , key )
23- c .addOrUpdateEndpointQueue .Add (key )
21+ func findServiceKey (endpointSlice * discoveryv1.EndpointSlice ) string {
22+ if endpointSlice != nil && endpointSlice .Labels != nil && endpointSlice .Labels [discoveryv1 .LabelServiceName ] != "" {
23+ return endpointSlice .Namespace + "/" + endpointSlice .Labels [discoveryv1 .LabelServiceName ]
24+ }
25+ return ""
2426}
2527
26- func (c * Controller ) enqueueUpdateEndpoint (oldObj , newObj any ) {
27- oldEp := oldObj .(* v1.Endpoints )
28- newEp := newObj .(* v1.Endpoints )
29- if oldEp .ResourceVersion == newEp .ResourceVersion {
28+ func (c * Controller ) enqueueAddEndpointSlice (obj any ) {
29+ key := findServiceKey (obj .(* discoveryv1.EndpointSlice ))
30+ if key != "" {
31+ klog .V (3 ).Infof ("enqueue add endpointSlice %s" , key )
32+ c .addOrUpdateEndpointSliceQueue .Add (key )
33+ }
34+ }
35+
36+ func (c * Controller ) enqueueUpdateEndpointSlice (oldObj , newObj any ) {
37+ oldEndpointSlice := oldObj .(* discoveryv1.EndpointSlice )
38+ newEndpointSlice := newObj .(* discoveryv1.EndpointSlice )
39+ if oldEndpointSlice .ResourceVersion == newEndpointSlice .ResourceVersion {
3040 return
3141 }
3242
33- if len (oldEp . Subsets ) == 0 && len (newEp . Subsets ) == 0 {
43+ if len (oldEndpointSlice . Endpoints ) == 0 && len (newEndpointSlice . Endpoints ) == 0 {
3444 return
3545 }
3646
37- key := cache .MetaObjectToName (newEp ).String ()
38- klog .V (3 ).Infof ("enqueue update endpoint %s" , key )
39- c .addOrUpdateEndpointQueue .Add (key )
47+ key := findServiceKey (newEndpointSlice )
48+ if key != "" {
49+ klog .V (3 ).Infof ("enqueue update endpointSlice for service %s" , key )
50+ c .addOrUpdateEndpointSliceQueue .Add (key )
51+ }
4052}
4153
42- func (c * Controller ) handleUpdateEndpoint (key string ) error {
54+ func (c * Controller ) handleUpdateEndpointSlice (key string ) error {
4355 namespace , name , err := cache .SplitMetaNamespaceKey (key )
4456 if err != nil {
4557 utilruntime .HandleError (fmt .Errorf ("invalid resource key: %s" , key ))
@@ -48,9 +60,9 @@ func (c *Controller) handleUpdateEndpoint(key string) error {
4860
4961 c .epKeyMutex .LockKey (key )
5062 defer func () { _ = c .epKeyMutex .UnlockKey (key ) }()
51- klog .Infof ("handle update endpoint %s" , key )
63+ klog .Infof ("handle update endpointSlice for service %s" , key )
5264
53- ep , err := c .endpointsLister . Endpoints (namespace ).Get ( name )
65+ endpointSlices , err := c .endpointSlicesLister . EndpointSlices (namespace ).List ( labels . Set ( map [ string ] string { discoveryv1 . LabelServiceName : name }). AsSelector () )
5466 if err != nil {
5567 if errors .IsNotFound (err ) {
5668 return nil
@@ -81,11 +93,10 @@ func (c *Controller) handleUpdateEndpoint(key string) error {
8193 if vip , ok = svc .Annotations [util .SwitchLBRuleVipsAnnotation ]; ok {
8294 lbVips = []string {vip }
8395
84- for _ , subset := range ep .Subsets {
85- for _ , address := range subset .Addresses {
86- // TODO: IPv6
96+ for _ , endpointSlice := range endpointSlices {
97+ for _ , endpoint := range endpointSlice .Endpoints {
8798 if util .CheckProtocol (vip ) == kubeovnv1 .ProtocolIPv4 &&
88- address .TargetRef .Name != "" {
99+ endpoint .TargetRef .Name != "" {
89100 ignoreHealthCheck = false
90101 }
91102 }
@@ -113,32 +124,7 @@ func (c *Controller) handleUpdateEndpoint(key string) error {
113124 klog .Errorf ("failed to get pods for service %s in namespace %s: %v" , name , namespace , err )
114125 return err
115126 }
116- for i , pod := range pods {
117- if pod .Status .PodIP != "" || len (pod .Status .PodIPs ) != 0 {
118- continue
119- }
120-
121- for _ , subset := range ep .Subsets {
122- for _ , addr := range subset .Addresses {
123- if addr .TargetRef == nil || addr .TargetRef .Kind != "Pod" || addr .TargetRef .Name != pod .Name {
124- continue
125- }
126-
127- p , err := c .config .KubeClient .CoreV1 ().Pods (pod .Namespace ).Get (context .Background (), pod .Name , metav1.GetOptions {})
128- if err != nil {
129- klog .Errorf ("failed to get pod %s/%s: %v" , pod .Namespace , pod .Name , err )
130- return err
131- }
132- pods [i ] = p .DeepCopy ()
133- break
134- }
135- if pods [i ] != pod {
136- break
137- }
138- }
139- }
140-
141- vpcName , subnetName = c .getVpcSubnetName (pods , ep , svc )
127+ vpcName , subnetName = c .getVpcSubnetName (pods , endpointSlices , svc )
142128
143129 var (
144130 vpc * kubeovnv1.Vpc
@@ -189,7 +175,7 @@ func (c *Controller) handleUpdateEndpoint(key string) error {
189175 checkIP = util .MasqueradeCheckIP
190176 }
191177 isGenIPPortMapping := ! ignoreHealthCheck || isPreferLocalBackend
192- ipPortMapping , backends = getIPPortMappingBackend (ep , pods , port , lbVip , checkIP , isGenIPPortMapping )
178+ ipPortMapping , backends = getIPPortMappingBackend (endpointSlices , port , lbVip , checkIP , isGenIPPortMapping )
193179 // for performance reason delete lb with no backends
194180 if len (backends ) != 0 {
195181 vip = util .JoinHostPort (lbVip , port .Port )
@@ -252,7 +238,7 @@ func (c *Controller) handleUpdateEndpoint(key string) error {
252238 return nil
253239}
254240
255- func (c * Controller ) getVpcSubnetName (pods []* v1.Pod , endpoints * v1. Endpoints , service * v1.Service ) (string , string ) {
241+ func (c * Controller ) getVpcSubnetName (pods []* v1.Pod , endpointSlices [] * discoveryv1. EndpointSlice , service * v1.Service ) (string , string ) {
256242 var (
257243 vpcName string
258244 subnetName string
@@ -267,14 +253,16 @@ func (c *Controller) getVpcSubnetName(pods []*v1.Pod, endpoints *v1.Endpoints, s
267253 }
268254
269255 LOOP:
270- for _ , subset := range endpoints .Subsets {
271- for _ , addr := range subset .Addresses {
272- if addr .IP == pod .Status .PodIP {
273- if vpcName == "" {
274- vpcName = pod .Annotations [util .LogicalRouterAnnotation ]
275- }
276- if vpcName != "" {
277- break LOOP
256+ for _ , endpointSlice := range endpointSlices {
257+ for _ , endpoint := range endpointSlice .Endpoints {
258+ for _ , addr := range endpoint .Addresses {
259+ if addr == pod .Status .PodIP {
260+ if vpcName == "" {
261+ vpcName = pod .Annotations [util .LogicalRouterAnnotation ]
262+ }
263+ if vpcName != "" {
264+ break LOOP
265+ }
278266 }
279267 }
280268 }
@@ -361,56 +349,50 @@ func (c *Controller) getHealthCheckVip(subnetName, lbVip string) (string, error)
361349 return checkIP , nil
362350}
363351
364- func getIPPortMappingBackend (endpoints * v1. Endpoints , pods []* v1. Pod , servicePort v1.ServicePort , serviceIP , checkVip string , isGenIPPortMapping bool ) (map [string ]string , []string ) {
352+ func getIPPortMappingBackend (endpointSlices []* discoveryv1. EndpointSlice , servicePort v1.ServicePort , serviceIP , checkVip string , isGenIPPortMapping bool ) (map [string ]string , []string ) {
365353 var (
366354 ipPortMapping = map [string ]string {}
367355 backends = []string {}
368356 protocol = util .CheckProtocol (serviceIP )
369357 )
370358
371- for _ , subset := range endpoints . Subsets {
359+ for _ , endpointSlice := range endpointSlices {
372360 var targetPort int32
373- for _ , port := range subset .Ports {
374- if port .Name == servicePort .Name {
375- targetPort = port .Port
361+ for _ , port := range endpointSlice .Ports {
362+ if port .Name != nil && * port . Name == servicePort .Name {
363+ targetPort = * port .Port
376364 break
377365 }
378366 }
379367 if targetPort == 0 {
380368 continue
381369 }
382370
383- for _ , address := range subset .Addresses {
384- if isGenIPPortMapping && address .TargetRef .Name != "" {
385- ipName := fmt .Sprintf ("%s.%s" , address .TargetRef .Name , address .TargetRef .Namespace )
386- ipPortMapping [address .IP ] = fmt .Sprintf (util .HealthCheckNamedVipTemplate , ipName , checkVip )
387- }
388- if address .TargetRef == nil || address .TargetRef .Kind != "Pod" {
389- if util .CheckProtocol (address .IP ) == protocol {
390- backends = append (backends , util .JoinHostPort (address .IP , targetPort ))
371+ for _ , endpoint := range endpointSlice .Endpoints {
372+ if isGenIPPortMapping && endpoint .TargetRef .Name != "" {
373+ ipName := fmt .Sprintf ("%s.%s" , endpoint .TargetRef .Name , endpoint .TargetRef .Namespace )
374+ for _ , address := range endpoint .Addresses {
375+ ipPortMapping [address ] = fmt .Sprintf (util .HealthCheckNamedVipTemplate , ipName , checkVip )
391376 }
377+ }
378+ }
379+
380+ for _ , endpoint := range endpointSlice .Endpoints {
381+ if ! endpointReady (endpoint ) {
392382 continue
393383 }
394- var ip string
395- for _ , pod := range pods {
396- if pod .Name == address .TargetRef .Name {
397- for _ , podIP := range util .PodIPs (* pod ) {
398- if util .CheckProtocol (podIP ) == protocol {
399- ip = podIP
400- break
401- }
402- }
403- break
384+
385+ for _ , address := range endpoint .Addresses {
386+ if util .CheckProtocol (address ) == protocol {
387+ backends = append (backends , util .JoinHostPort (address , targetPort ))
404388 }
405389 }
406- if ip == "" && util .CheckProtocol (address .IP ) == protocol {
407- ip = address .IP
408- }
409- if ip != "" {
410- backends = append (backends , util .JoinHostPort (ip , targetPort ))
411- }
412390 }
413391 }
414392
415393 return ipPortMapping , backends
416394}
395+
396+ func endpointReady (endpoint discoveryv1.Endpoint ) bool {
397+ return endpoint .Conditions .Ready == nil || * endpoint .Conditions .Ready
398+ }
0 commit comments