diff --git a/zrpc/resolver/internal/kube/eventhandler.go b/zrpc/resolver/internal/kube/eventhandler.go index 3e0c6b14fd51..1c0180ac97f2 100644 --- a/zrpc/resolver/internal/kube/eventhandler.go +++ b/zrpc/resolver/internal/kube/eventhandler.go @@ -5,7 +5,7 @@ import ( "github.com/zeromicro/go-zero/core/lang" "github.com/zeromicro/go-zero/core/logx" - v1 "k8s.io/api/core/v1" + discoveryv1 "k8s.io/api/discovery/v1" "k8s.io/client-go/tools/cache" ) @@ -28,9 +28,9 @@ func NewEventHandler(update func([]string)) *EventHandler { // OnAdd handles the endpoints add events. func (h *EventHandler) OnAdd(obj any, _ bool) { - endpoints, ok := obj.(*v1.Endpoints) + endpoints, ok := obj.(*discoveryv1.EndpointSlice) if !ok { - logx.Errorf("%v is not an object with type *v1.Endpoints", obj) + logx.Errorf("%v is not an object with type *discoveryv1.EndpointSlice", obj) return } @@ -38,10 +38,10 @@ func (h *EventHandler) OnAdd(obj any, _ bool) { defer h.lock.Unlock() var changed bool - for _, sub := range endpoints.Subsets { - for _, point := range sub.Addresses { - if _, ok := h.endpoints[point.IP]; !ok { - h.endpoints[point.IP] = lang.Placeholder + for _, point := range endpoints.Endpoints { + for _, address := range point.Addresses { + if _, ok := h.endpoints[address]; !ok { + h.endpoints[address] = lang.Placeholder changed = true } } @@ -54,9 +54,9 @@ func (h *EventHandler) OnAdd(obj any, _ bool) { // OnDelete handles the endpoints delete events. func (h *EventHandler) OnDelete(obj any) { - endpoints, ok := obj.(*v1.Endpoints) + endpoints, ok := obj.(*discoveryv1.EndpointSlice) if !ok { - logx.Errorf("%v is not an object with type *v1.Endpoints", obj) + logx.Errorf("%v is not an object with type *discoveryv1.EndpointSlice", obj) return } @@ -64,10 +64,10 @@ func (h *EventHandler) OnDelete(obj any) { defer h.lock.Unlock() var changed bool - for _, sub := range endpoints.Subsets { - for _, point := range sub.Addresses { - if _, ok := h.endpoints[point.IP]; ok { - delete(h.endpoints, point.IP) + for _, point := range endpoints.Endpoints { + for _, address := range point.Addresses { + if _, ok := h.endpoints[address]; ok { + delete(h.endpoints, address) changed = true } } @@ -80,35 +80,35 @@ func (h *EventHandler) OnDelete(obj any) { // OnUpdate handles the endpoints update events. func (h *EventHandler) OnUpdate(oldObj, newObj any) { - oldEndpoints, ok := oldObj.(*v1.Endpoints) + oldEndpointSlices, ok := oldObj.(*discoveryv1.EndpointSlice) if !ok { - logx.Errorf("%v is not an object with type *v1.Endpoints", oldObj) + logx.Errorf("%v is not an object with type *discoveryv1.EndpointSlice", oldObj) return } - newEndpoints, ok := newObj.(*v1.Endpoints) + newEndpointSlices, ok := newObj.(*discoveryv1.EndpointSlice) if !ok { - logx.Errorf("%v is not an object with type *v1.Endpoints", newObj) + logx.Errorf("%v is not an object with type *discoveryv1.EndpointSlice", newObj) return } - if oldEndpoints.ResourceVersion == newEndpoints.ResourceVersion { + if oldEndpointSlices.ResourceVersion == newEndpointSlices.ResourceVersion { return } - h.Update(newEndpoints) + h.Update(newEndpointSlices) } // Update updates the endpoints. -func (h *EventHandler) Update(endpoints *v1.Endpoints) { +func (h *EventHandler) Update(endpoints *discoveryv1.EndpointSlice) { h.lock.Lock() defer h.lock.Unlock() old := h.endpoints h.endpoints = make(map[string]lang.PlaceholderType) - for _, sub := range endpoints.Subsets { - for _, point := range sub.Addresses { - h.endpoints[point.IP] = lang.Placeholder + for _, point := range endpoints.Endpoints { + for _, address := range point.Addresses { + h.endpoints[address] = lang.Placeholder } } diff --git a/zrpc/resolver/internal/kube/eventhandler_test.go b/zrpc/resolver/internal/kube/eventhandler_test.go index fed504ab31e6..62d46efda5d0 100644 --- a/zrpc/resolver/internal/kube/eventhandler_test.go +++ b/zrpc/resolver/internal/kube/eventhandler_test.go @@ -4,7 +4,7 @@ import ( "testing" "github.com/stretchr/testify/assert" - v1 "k8s.io/api/core/v1" + discoveryv1 "k8s.io/api/discovery/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) @@ -14,21 +14,19 @@ func TestAdd(t *testing.T) { endpoints = change }) h.OnAdd("bad", false) - h.OnAdd(&v1.Endpoints{Subsets: []v1.EndpointSubset{ - { - Addresses: []v1.EndpointAddress{ - { - IP: "0.0.0.1", - }, - { - IP: "0.0.0.2", - }, - { - IP: "0.0.0.3", - }, + h.OnAdd(&discoveryv1.EndpointSlice{ + Endpoints: []discoveryv1.Endpoint{ + { + Addresses: []string{"0.0.0.1"}, + }, + { + Addresses: []string{"0.0.0.2"}, + }, + { + Addresses: []string{"0.0.0.3"}, }, }, - }}, false) + }, false) assert.ElementsMatch(t, []string{"0.0.0.1", "0.0.0.2", "0.0.0.3"}, endpoints) } @@ -37,34 +35,30 @@ func TestDelete(t *testing.T) { h := NewEventHandler(func(change []string) { endpoints = change }) - h.OnAdd(&v1.Endpoints{Subsets: []v1.EndpointSubset{ - { - Addresses: []v1.EndpointAddress{ - { - IP: "0.0.0.1", - }, - { - IP: "0.0.0.2", - }, - { - IP: "0.0.0.3", - }, + h.OnAdd(&discoveryv1.EndpointSlice{ + Endpoints: []discoveryv1.Endpoint{ + { + Addresses: []string{"0.0.0.1"}, + }, + { + Addresses: []string{"0.0.0.2"}, + }, + { + Addresses: []string{"0.0.0.3"}, }, }, - }}, false) + }, false) h.OnDelete("bad") - h.OnDelete(&v1.Endpoints{Subsets: []v1.EndpointSubset{ - { - Addresses: []v1.EndpointAddress{ - { - IP: "0.0.0.1", - }, - { - IP: "0.0.0.2", - }, + h.OnDelete(&discoveryv1.EndpointSlice{ + Endpoints: []discoveryv1.Endpoint{ + { + Addresses: []string{"0.0.0.1"}, + }, + { + Addresses: []string{"0.0.0.2"}, }, }, - }}) + }) assert.ElementsMatch(t, []string{"0.0.0.3"}, endpoints) } @@ -73,36 +67,28 @@ func TestUpdate(t *testing.T) { h := NewEventHandler(func(change []string) { endpoints = change }) - h.OnUpdate(&v1.Endpoints{ - Subsets: []v1.EndpointSubset{ + h.OnUpdate(&discoveryv1.EndpointSlice{ + Endpoints: []discoveryv1.Endpoint{ + { + Addresses: []string{"0.0.0.1"}, + }, { - Addresses: []v1.EndpointAddress{ - { - IP: "0.0.0.1", - }, - { - IP: "0.0.0.2", - }, - }, + Addresses: []string{"0.0.0.2"}, }, }, ObjectMeta: metav1.ObjectMeta{ ResourceVersion: "1", }, - }, &v1.Endpoints{ - Subsets: []v1.EndpointSubset{ + }, &discoveryv1.EndpointSlice{ + Endpoints: []discoveryv1.Endpoint{ { - Addresses: []v1.EndpointAddress{ - { - IP: "0.0.0.1", - }, - { - IP: "0.0.0.2", - }, - { - IP: "0.0.0.3", - }, - }, + Addresses: []string{"0.0.0.1"}, + }, + { + Addresses: []string{"0.0.0.2"}, + }, + { + Addresses: []string{"0.0.0.3"}, }, }, ObjectMeta: metav1.ObjectMeta{ @@ -116,33 +102,25 @@ func TestUpdateNoChange(t *testing.T) { h := NewEventHandler(func(change []string) { assert.Fail(t, "should not called") }) - h.OnUpdate(&v1.Endpoints{ - Subsets: []v1.EndpointSubset{ + h.OnUpdate(&discoveryv1.EndpointSlice{ + Endpoints: []discoveryv1.Endpoint{ { - Addresses: []v1.EndpointAddress{ - { - IP: "0.0.0.1", - }, - { - IP: "0.0.0.2", - }, - }, + Addresses: []string{"0.0.0.1"}, + }, + { + Addresses: []string{"0.0.0.2"}, }, }, ObjectMeta: metav1.ObjectMeta{ ResourceVersion: "1", }, - }, &v1.Endpoints{ - Subsets: []v1.EndpointSubset{ + }, &discoveryv1.EndpointSlice{ + Endpoints: []discoveryv1.Endpoint{ + { + Addresses: []string{"0.0.0.1"}, + }, { - Addresses: []v1.EndpointAddress{ - { - IP: "0.0.0.1", - }, - { - IP: "0.0.0.2", - }, - }, + Addresses: []string{"0.0.0.2"}, }, }, ObjectMeta: metav1.ObjectMeta{ @@ -156,45 +134,35 @@ func TestUpdateChangeWithDifferentVersion(t *testing.T) { h := NewEventHandler(func(change []string) { endpoints = change }) - h.OnAdd(&v1.Endpoints{Subsets: []v1.EndpointSubset{ - { - Addresses: []v1.EndpointAddress{ - { - IP: "0.0.0.1", - }, - { - IP: "0.0.0.3", - }, + h.OnAdd(&discoveryv1.EndpointSlice{ + Endpoints: []discoveryv1.Endpoint{ + { + Addresses: []string{"0.0.0.1"}, + }, + { + Addresses: []string{"0.0.0.3"}, }, }, - }}, false) - h.OnUpdate(&v1.Endpoints{ - Subsets: []v1.EndpointSubset{ + }, false) + h.OnUpdate(&discoveryv1.EndpointSlice{ + Endpoints: []discoveryv1.Endpoint{ + { + Addresses: []string{"0.0.0.1"}, + }, { - Addresses: []v1.EndpointAddress{ - { - IP: "0.0.0.1", - }, - { - IP: "0.0.0.3", - }, - }, + Addresses: []string{"0.0.0.3"}, }, }, ObjectMeta: metav1.ObjectMeta{ ResourceVersion: "1", }, - }, &v1.Endpoints{ - Subsets: []v1.EndpointSubset{ + }, &discoveryv1.EndpointSlice{ + Endpoints: []discoveryv1.Endpoint{ + { + Addresses: []string{"0.0.0.1"}, + }, { - Addresses: []v1.EndpointAddress{ - { - IP: "0.0.0.1", - }, - { - IP: "0.0.0.2", - }, - }, + Addresses: []string{"0.0.0.2"}, }, }, ObjectMeta: metav1.ObjectMeta{ @@ -209,63 +177,49 @@ func TestUpdateNoChangeWithDifferentVersion(t *testing.T) { h := NewEventHandler(func(change []string) { endpoints = change }) - h.OnAdd(&v1.Endpoints{Subsets: []v1.EndpointSubset{ - { - Addresses: []v1.EndpointAddress{ - { - IP: "0.0.0.1", - }, - { - IP: "0.0.0.2", - }, + h.OnAdd(&discoveryv1.EndpointSlice{ + Endpoints: []discoveryv1.Endpoint{ + { + Addresses: []string{"0.0.0.1"}, + }, + { + Addresses: []string{"0.0.0.2"}, }, }, - }}, false) - h.OnUpdate("bad", &v1.Endpoints{Subsets: []v1.EndpointSubset{ - { - Addresses: []v1.EndpointAddress{ - { - IP: "0.0.0.1", - }, + }, false) + h.OnUpdate("bad", &discoveryv1.EndpointSlice{ + Endpoints: []discoveryv1.Endpoint{ + { + Addresses: []string{"0.0.0.1"}, }, }, - }}) - h.OnUpdate(&v1.Endpoints{Subsets: []v1.EndpointSubset{ - { - Addresses: []v1.EndpointAddress{ - { - IP: "0.0.0.1", - }, + }) + h.OnUpdate(&discoveryv1.EndpointSlice{ + Endpoints: []discoveryv1.Endpoint{ + { + Addresses: []string{"0.0.0.1"}, }, }, - }}, "bad") - h.OnUpdate(&v1.Endpoints{ - Subsets: []v1.EndpointSubset{ + }, "bad") + h.OnUpdate(&discoveryv1.EndpointSlice{ + Endpoints: []discoveryv1.Endpoint{ + { + Addresses: []string{"0.0.0.1"}, + }, { - Addresses: []v1.EndpointAddress{ - { - IP: "0.0.0.1", - }, - { - IP: "0.0.0.2", - }, - }, + Addresses: []string{"0.0.0.2"}, }, }, ObjectMeta: metav1.ObjectMeta{ ResourceVersion: "1", }, - }, &v1.Endpoints{ - Subsets: []v1.EndpointSubset{ + }, &discoveryv1.EndpointSlice{ + Endpoints: []discoveryv1.Endpoint{ + { + Addresses: []string{"0.0.0.1"}, + }, { - Addresses: []v1.EndpointAddress{ - { - IP: "0.0.0.1", - }, - { - IP: "0.0.0.2", - }, - }, + Addresses: []string{"0.0.0.2"}, }, }, ObjectMeta: metav1.ObjectMeta{ diff --git a/zrpc/resolver/internal/kubebuilder.go b/zrpc/resolver/internal/kubebuilder.go index c24988652578..caa12f990dd6 100644 --- a/zrpc/resolver/internal/kubebuilder.go +++ b/zrpc/resolver/internal/kubebuilder.go @@ -18,8 +18,8 @@ import ( ) const ( - resyncInterval = 5 * time.Minute - nameSelector = "metadata.name=" + resyncInterval = 5 * time.Minute + serviceSelector = "kubernetes.io/service-name=" ) type kubeResolver struct { @@ -60,14 +60,19 @@ func (b *kubeBuilder) Build(target resolver.Target, cc resolver.ClientConn, } if svc.Port == 0 { - // getting endpoints is only to get the port - endpoints, err := cs.CoreV1().Endpoints(svc.Namespace).Get( - context.Background(), svc.Name, v1.GetOptions{}) + endpointSlices, err := cs.DiscoveryV1().EndpointSlices(svc.Namespace).List(context.Background(), v1.ListOptions{ + LabelSelector: serviceSelector + svc.Name, + }) if err != nil { return nil, err } + if len(endpointSlices.Items) == 0 { + return nil, fmt.Errorf("no endpoint slices found for service %s in namespace %s", svc.Name, svc.Namespace) + } - svc.Port = int(endpoints.Subsets[0].Ports[0].Port) + // Since this resolver is used for in-cluster service discovery, + // there should be at least one port available. + svc.Port = int(*endpointSlices.Items[0].Ports[0].Port) } handler := kube.NewEventHandler(func(endpoints []string) { @@ -88,23 +93,27 @@ func (b *kubeBuilder) Build(target resolver.Target, cc resolver.ClientConn, inf := informers.NewSharedInformerFactoryWithOptions(cs, resyncInterval, informers.WithNamespace(svc.Namespace), informers.WithTweakListOptions(func(options *v1.ListOptions) { - options.FieldSelector = nameSelector + svc.Name + options.LabelSelector = serviceSelector + svc.Name })) - in := inf.Core().V1().Endpoints() + in := inf.Discovery().V1().EndpointSlices() _, err = in.Informer().AddEventHandler(handler) if err != nil { return nil, err } - // get the initial endpoints, cannot use the previous endpoints, - // because the endpoints may be updated before/after the informer is started. - endpoints, err := cs.CoreV1().Endpoints(svc.Namespace).Get( - context.Background(), svc.Name, v1.GetOptions{}) + // get the initial endpoint slices, cannot use the previous endpoint slices, + // because the endpoint slices may be updated before/after the informer is started. + endpointSlices, err := cs.DiscoveryV1().EndpointSlices(svc.Namespace).List( + context.Background(), v1.ListOptions{ + LabelSelector: serviceSelector + svc.Name, + }) if err != nil { return nil, err } - handler.Update(endpoints) + for _, endpointSlice := range endpointSlices.Items { + handler.Update(&endpointSlice) + } r := &kubeResolver{ cc: cc,