From 4141212a5acff460e43c6512c3f049c9b516e707 Mon Sep 17 00:00:00 2001 From: Srinivas Chippagiri Date: Fri, 13 Nov 2026 01:39:11 -0800 Subject: [PATCH] Use EndpointSlices instead of v1 Endpoints in the metadata service v1 Endpoints is deprecated since Kubernetes 1.33 and the metadata service logs a warning for every watch. Watch discovery.k8s.io/v1 EndpointSlices instead and merge the slices of each Service into the existing Endpoints proto, so the rest of the metadata pipeline doesn't change. Terminating endpoints are left out like the Endpoints API does, and pods in dual-stack services are only listed once. The metadata and updater roles get read access to endpointslices. Signed-off-by: Srinivas Chippagiri --- k8s/vizier/base/metadata_role.yaml | 8 + k8s/vizier/bootstrap/updater_role.yaml | 8 + src/shared/k8s/BUILD.bazel | 2 + src/shared/k8s/proto_utils.go | 89 ++++++++++ src/shared/k8s/proto_utils_test.go | 54 ++++++ .../metadata/controllers/k8smeta/BUILD.bazel | 3 + .../k8smeta/k8s_metadata_controller_test.go | 165 +++++++++--------- .../controllers/k8smeta/k8s_metadata_utils.go | 71 +++++++- 8 files changed, 313 insertions(+), 87 deletions(-) diff --git a/k8s/vizier/base/metadata_role.yaml b/k8s/vizier/base/metadata_role.yaml index cf503ad2b7c..972e7dbde01 100644 --- a/k8s/vizier/base/metadata_role.yaml +++ b/k8s/vizier/base/metadata_role.yaml @@ -29,6 +29,14 @@ rules: - watch - get - list +- apiGroups: + - discovery.k8s.io + resources: + - endpointslices + verbs: + - watch + - get + - list --- apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRoleBinding diff --git a/k8s/vizier/bootstrap/updater_role.yaml b/k8s/vizier/bootstrap/updater_role.yaml index 557cafc8d99..de41e208bee 100644 --- a/k8s/vizier/bootstrap/updater_role.yaml +++ b/k8s/vizier/bootstrap/updater_role.yaml @@ -173,6 +173,14 @@ rules: - get - watch - list +- apiGroups: + - discovery.k8s.io + resources: + - endpointslices + verbs: + - get + - watch + - list --- apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRoleBinding diff --git a/src/shared/k8s/BUILD.bazel b/src/shared/k8s/BUILD.bazel index a5a3691b382..f2ffbc17154 100644 --- a/src/shared/k8s/BUILD.bazel +++ b/src/shared/k8s/BUILD.bazel @@ -31,6 +31,7 @@ go_library( "@com_github_sirupsen_logrus//:logrus", "@io_k8s_api//apps/v1:apps", "@io_k8s_api//core/v1:core", + "@io_k8s_api//discovery/v1:discovery", "@io_k8s_apimachinery//pkg/apis/meta/v1:meta", "@io_k8s_apimachinery//pkg/types", "@io_k8s_apimachinery//pkg/util/intstr", @@ -52,6 +53,7 @@ pl_go_test( "@com_github_stretchr_testify//require", "@io_k8s_api//apps/v1:apps", "@io_k8s_api//core/v1:core", + "@io_k8s_api//discovery/v1:discovery", "@io_k8s_apimachinery//pkg/apis/meta/v1:meta", "@io_k8s_apimachinery//pkg/types", "@io_k8s_apimachinery//pkg/util/intstr", diff --git a/src/shared/k8s/proto_utils.go b/src/shared/k8s/proto_utils.go index fb66e8433e4..7c9a2df1e79 100644 --- a/src/shared/k8s/proto_utils.go +++ b/src/shared/k8s/proto_utils.go @@ -23,6 +23,7 @@ import ( apps "k8s.io/api/apps/v1" v1 "k8s.io/api/core/v1" + discoveryv1 "k8s.io/api/discovery/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/intstr" @@ -466,6 +467,94 @@ func EndpointSubsetFromProto(pb *metadatapb.EndpointSubset) *v1.EndpointSubset { } } +// EndpointSlicesToProto builds the Endpoints proto of a Service from its EndpointSlices. +// Each slice becomes one subset. Ready endpoints go to Addresses, terminating ones are +// left out like the v1 Endpoints API does, and the rest go to NotReadyAddresses. With +// dual-stack services a pod shows up in one slice per IP family, so it's only added once. +func EndpointSlicesToProto(serviceName string, slices []*discoveryv1.EndpointSlice) *metadatapb.Endpoints { + meta := metav1.ObjectMeta{Name: serviceName} + seen := make(map[string]bool) + var subsets []*metadatapb.EndpointSubset + + for _, slice := range slices { + meta.Namespace = slice.Namespace + if meta.CreationTimestamp.IsZero() || slice.CreationTimestamp.Before(&meta.CreationTimestamp) { + meta.CreationTimestamp = slice.CreationTimestamp + } + for _, ref := range slice.OwnerReferences { + if ref.Kind == "Service" && ref.Name == serviceName { + meta.UID = ref.UID + } + } + if slice.AddressType == discoveryv1.AddressTypeFQDN { + continue + } + + subset := &metadatapb.EndpointSubset{ + Addresses: []*metadatapb.EndpointAddress{}, + NotReadyAddresses: []*metadatapb.EndpointAddress{}, + Ports: []*metadatapb.EndpointPort{}, + } + for _, ep := range slice.Endpoints { + if ep.Conditions.Terminating != nil && *ep.Conditions.Terminating { + continue + } + if len(ep.Addresses) == 0 { + continue + } + key := ep.Addresses[0] + if ep.TargetRef != nil && ep.TargetRef.UID != "" { + key = string(ep.TargetRef.UID) + } + if seen[key] { + continue + } + seen[key] = true + + addr := &metadatapb.EndpointAddress{IP: ep.Addresses[0]} + if ep.Hostname != nil { + addr.Hostname = *ep.Hostname + } + if ep.NodeName != nil { + addr.NodeName = *ep.NodeName + } + if ep.TargetRef != nil { + addr.TargetRef = ObjectReferenceToProto(ep.TargetRef) + } + if ep.Conditions.Ready == nil || *ep.Conditions.Ready { + subset.Addresses = append(subset.Addresses, addr) + } else { + subset.NotReadyAddresses = append(subset.NotReadyAddresses, addr) + } + } + if len(subset.Addresses) == 0 && len(subset.NotReadyAddresses) == 0 { + continue + } + for _, p := range slice.Ports { + port := &metadatapb.EndpointPort{} + if p.Name != nil { + port.Name = *p.Name + } + if p.Port != nil { + port.Port = *p.Port + } + if p.Protocol != nil { + port.Protocol = ipProtocolObjToPbMap[*p.Protocol] + } + subset.Ports = append(subset.Ports, port) + } + subsets = append(subsets, subset) + } + + if subsets == nil { + subsets = []*metadatapb.EndpointSubset{} + } + return &metadatapb.Endpoints{ + Metadata: ObjectMetadataToProto(&meta), + Subsets: subsets, + } +} + // EndpointsToProto converts an Endpoints into a proto. func EndpointsToProto(e *v1.Endpoints) *metadatapb.Endpoints { subsets := make([]*metadatapb.EndpointSubset, len(e.Subsets)) diff --git a/src/shared/k8s/proto_utils_test.go b/src/shared/k8s/proto_utils_test.go index e94f0c7ead4..52f416fc011 100644 --- a/src/shared/k8s/proto_utils_test.go +++ b/src/shared/k8s/proto_utils_test.go @@ -25,6 +25,7 @@ import ( "github.com/stretchr/testify/assert" apps "k8s.io/api/apps/v1" v1 "k8s.io/api/core/v1" + discoveryv1 "k8s.io/api/discovery/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/intstr" @@ -1307,6 +1308,59 @@ func TestEndpointsToProto(t *testing.T) { assert.Equal(t, expectedPb, oPb) } +func TestEndpointSlicesToProto(t *testing.T) { + yes, no := true, false + tcp := v1.ProtocolTCP + port := int32(80) + portName := "http" + node := "node1" + created := metav1.Unix(0, 10) + owner := []metav1.OwnerReference{{Kind: "Service", Name: "svc", UID: "svc-uid"}} + podA := &v1.ObjectReference{Kind: "Pod", Namespace: "ns", Name: "a", UID: "a-uid"} + podB := &v1.ObjectReference{Kind: "Pod", Namespace: "ns", Name: "b", UID: "b-uid"} + podC := &v1.ObjectReference{Kind: "Pod", Namespace: "ns", Name: "c", UID: "c-uid"} + ports := []discoveryv1.EndpointPort{{Name: &portName, Port: &port, Protocol: &tcp}} + + slices := []*discoveryv1.EndpointSlice{ + { + ObjectMeta: metav1.ObjectMeta{Name: "svc-v4", Namespace: "ns", CreationTimestamp: created, OwnerReferences: owner}, + AddressType: discoveryv1.AddressTypeIPv4, + Endpoints: []discoveryv1.Endpoint{ + {Addresses: []string{"10.0.0.1"}, NodeName: &node, TargetRef: podA, Conditions: discoveryv1.EndpointConditions{Ready: &yes}}, + {Addresses: []string{"10.0.0.2"}, NodeName: &node, TargetRef: podB, Conditions: discoveryv1.EndpointConditions{Ready: &no}}, + {Addresses: []string{"10.0.0.3"}, NodeName: &node, TargetRef: podC, Conditions: discoveryv1.EndpointConditions{Ready: &no, Terminating: &yes}}, + }, + Ports: ports, + }, + { + ObjectMeta: metav1.ObjectMeta{Name: "svc-v6", Namespace: "ns", CreationTimestamp: metav1.Unix(0, 20), OwnerReferences: owner}, + AddressType: discoveryv1.AddressTypeIPv6, + Endpoints: []discoveryv1.Endpoint{ + {Addresses: []string{"fd00::1"}, NodeName: &node, TargetRef: podA, Conditions: discoveryv1.EndpointConditions{Ready: &yes}}, + }, + Ports: ports, + }, + } + + pb := k8s.EndpointSlicesToProto("svc", slices) + + assert.Equal(t, "svc", pb.Metadata.Name) + assert.Equal(t, "ns", pb.Metadata.Namespace) + assert.Equal(t, "svc-uid", pb.Metadata.UID) + assert.Equal(t, int64(10), pb.Metadata.CreationTimestampNS) + // The IPv6 slice only has pod a again, so it adds no subset. + assert.Equal(t, 1, len(pb.Subsets)) + subset := pb.Subsets[0] + assert.Equal(t, 1, len(subset.Addresses)) + assert.Equal(t, "10.0.0.1", subset.Addresses[0].IP) + assert.Equal(t, "node1", subset.Addresses[0].NodeName) + assert.Equal(t, "a", subset.Addresses[0].TargetRef.Name) + // Terminating pod c is left out. + assert.Equal(t, 1, len(subset.NotReadyAddresses)) + assert.Equal(t, "10.0.0.2", subset.NotReadyAddresses[0].IP) + assert.Equal(t, []*metadatapb.EndpointPort{{Name: "http", Port: 80, Protocol: metadatapb.TCP}}, subset.Ports) +} + func TestEndpointsFromProto(t *testing.T) { oPb := &metadatapb.Endpoints{} if err := proto.UnmarshalText(endpointsPb, oPb); err != nil { diff --git a/src/vizier/services/metadata/controllers/k8smeta/BUILD.bazel b/src/vizier/services/metadata/controllers/k8smeta/BUILD.bazel index ba6f5c8498b..df627006cae 100644 --- a/src/vizier/services/metadata/controllers/k8smeta/BUILD.bazel +++ b/src/vizier/services/metadata/controllers/k8smeta/BUILD.bazel @@ -44,10 +44,12 @@ go_library( "@com_github_sirupsen_logrus//:logrus", "@io_k8s_api//apps/v1:apps", "@io_k8s_api//core/v1:core", + "@io_k8s_api//discovery/v1:discovery", "@io_k8s_apimachinery//pkg/labels", "@io_k8s_apimachinery//pkg/watch", "@io_k8s_client_go//informers", "@io_k8s_client_go//kubernetes", + "@io_k8s_client_go//listers/discovery/v1:discovery", "@io_k8s_client_go//rest", "@io_k8s_client_go//tools/cache", ], @@ -77,6 +79,7 @@ pl_go_test( "@com_github_stretchr_testify//require", "@io_k8s_api//apps/v1:apps", "@io_k8s_api//core/v1:core", + "@io_k8s_api//discovery/v1:discovery", "@io_k8s_apimachinery//pkg/apis/meta/v1:meta", "@io_k8s_apimachinery//pkg/util/intstr", "@io_k8s_apimachinery//pkg/watch", diff --git a/src/vizier/services/metadata/controllers/k8smeta/k8s_metadata_controller_test.go b/src/vizier/services/metadata/controllers/k8smeta/k8s_metadata_controller_test.go index b74f23fab94..0dbdb9786f5 100644 --- a/src/vizier/services/metadata/controllers/k8smeta/k8s_metadata_controller_test.go +++ b/src/vizier/services/metadata/controllers/k8smeta/k8s_metadata_controller_test.go @@ -27,6 +27,7 @@ import ( "github.com/stretchr/testify/require" appsv1 "k8s.io/api/apps/v1" v1 "k8s.io/api/core/v1" + discoveryv1 "k8s.io/api/discovery/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/util/intstr" "k8s.io/apimachinery/pkg/watch" @@ -344,31 +345,29 @@ func TestController(t *testing.T) { name: "simple endpoint", updates: []resourceUpdate{ &endpoints{ - e: &v1.Endpoints{ + e: &discoveryv1.EndpointSlice{ ObjectMeta: metav1.ObjectMeta{ - Name: "myendpoint", + Name: "myendpoint-abcde", + Labels: map[string]string{discoveryv1.LabelServiceName: "myendpoint"}, }, - Subsets: []v1.EndpointSubset{ + AddressType: discoveryv1.AddressTypeIPv4, + Endpoints: []discoveryv1.Endpoint{ { - Addresses: []v1.EndpointAddress{ - { - IP: "127.0.0.1", - Hostname: "hostname", - }, - }, - NotReadyAddresses: []v1.EndpointAddress{ - { - IP: "127.0.0.2", - Hostname: "notready", - }, - }, - Ports: []v1.EndpointPort{ - { - Name: "endpointport", - Protocol: v1.ProtocolTCP, - Port: 8081, - }, - }, + Addresses: []string{"127.0.0.1"}, + Hostname: strptr("hostname"), + Conditions: discoveryv1.EndpointConditions{Ready: boolptr(true)}, + }, + { + Addresses: []string{"127.0.0.2"}, + Hostname: strptr("notready"), + Conditions: discoveryv1.EndpointConditions{Ready: boolptr(false)}, + }, + }, + Ports: []discoveryv1.EndpointPort{ + { + Name: strptr("endpointport"), + Protocol: protocolptr(v1.ProtocolTCP), + Port: int32ptr(8081), }, }, }, @@ -844,13 +843,13 @@ func TestController(t *testing.T) { watchersStarted := map[string]bool{ // This list needs to be kept up-to-date with the list of watchers started in k8s_metadata_controller.go - "nodes": false, - "namespaces": false, - "pods": false, - "endpoints": false, - "services": false, - "replicasets": false, - "deployments": false, + "nodes": false, + "namespaces": false, + "pods": false, + "endpointslices": false, + "services": false, + "replicasets": false, + "deployments": false, } allStarted := func() bool { for _, started := range watchersStarted { @@ -962,31 +961,29 @@ func TestControllerWithNotWatchedNameSpaces(t *testing.T) { name: "simple endpoint", updates: []resourceUpdate{ &endpoints{ - e: &v1.Endpoints{ + e: &discoveryv1.EndpointSlice{ ObjectMeta: metav1.ObjectMeta{ - Name: "mynotwatchedendpoint", + Name: "mynotwatchedendpoint-abcde", + Labels: map[string]string{discoveryv1.LabelServiceName: "mynotwatchedendpoint"}, }, - Subsets: []v1.EndpointSubset{ + AddressType: discoveryv1.AddressTypeIPv4, + Endpoints: []discoveryv1.Endpoint{ { - Addresses: []v1.EndpointAddress{ - { - IP: "127.0.0.1", - Hostname: "hostname", - }, - }, - NotReadyAddresses: []v1.EndpointAddress{ - { - IP: "127.0.0.2", - Hostname: "notready", - }, - }, - Ports: []v1.EndpointPort{ - { - Name: "endpointport", - Protocol: v1.ProtocolTCP, - Port: 8081, - }, - }, + Addresses: []string{"127.0.0.1"}, + Hostname: strptr("hostname"), + Conditions: discoveryv1.EndpointConditions{Ready: boolptr(true)}, + }, + { + Addresses: []string{"127.0.0.2"}, + Hostname: strptr("notready"), + Conditions: discoveryv1.EndpointConditions{Ready: boolptr(false)}, + }, + }, + Ports: []discoveryv1.EndpointPort{ + { + Name: strptr("endpointport"), + Protocol: protocolptr(v1.ProtocolTCP), + Port: int32ptr(8081), }, }, }, @@ -994,31 +991,29 @@ func TestControllerWithNotWatchedNameSpaces(t *testing.T) { t: create, }, &endpoints{ - e: &v1.Endpoints{ + e: &discoveryv1.EndpointSlice{ ObjectMeta: metav1.ObjectMeta{ - Name: "myendpoint", + Name: "myendpoint-abcde", + Labels: map[string]string{discoveryv1.LabelServiceName: "myendpoint"}, }, - Subsets: []v1.EndpointSubset{ + AddressType: discoveryv1.AddressTypeIPv4, + Endpoints: []discoveryv1.Endpoint{ { - Addresses: []v1.EndpointAddress{ - { - IP: "127.0.0.1", - Hostname: "hostname", - }, - }, - NotReadyAddresses: []v1.EndpointAddress{ - { - IP: "127.0.0.2", - Hostname: "notready", - }, - }, - Ports: []v1.EndpointPort{ - { - Name: "endpointport", - Protocol: v1.ProtocolTCP, - Port: 8081, - }, - }, + Addresses: []string{"127.0.0.1"}, + Hostname: strptr("hostname"), + Conditions: discoveryv1.EndpointConditions{Ready: boolptr(true)}, + }, + { + Addresses: []string{"127.0.0.2"}, + Hostname: strptr("notready"), + Conditions: discoveryv1.EndpointConditions{Ready: boolptr(false)}, + }, + }, + Ports: []discoveryv1.EndpointPort{ + { + Name: strptr("endpointport"), + Protocol: protocolptr(v1.ProtocolTCP), + Port: int32ptr(8081), }, }, }, @@ -1384,13 +1379,13 @@ func TestControllerWithNotWatchedNameSpaces(t *testing.T) { watchersStarted := map[string]bool{ // This list needs to be kept up-to-date with the list of watchers started in k8s_metadata_controller.go - "nodes": false, - "namespaces": false, - "pods": false, - "endpoints": false, - "services": false, - "replicasets": false, - "deployments": false, + "nodes": false, + "namespaces": false, + "pods": false, + "endpointslices": false, + "services": false, + "replicasets": false, + "deployments": false, } allStarted := func() bool { for _, started := range watchersStarted { @@ -1580,13 +1575,13 @@ func (n *namespace) Apply(ctx context.Context, c kubernetes.Interface) error { } type endpoints struct { - e *v1.Endpoints + e *discoveryv1.EndpointSlice ns string t resourceUpdateType } func (e *endpoints) Apply(ctx context.Context, c kubernetes.Interface) error { - ei := c.CoreV1().Endpoints(e.ns) + ei := c.DiscoveryV1().EndpointSlices(e.ns) switch e.t { case create: _, err := ei.Create(ctx, e.e, metav1.CreateOptions{}) @@ -1601,6 +1596,12 @@ func (e *endpoints) Apply(ctx context.Context, c kubernetes.Interface) error { } } +func strptr(s string) *string { return &s } + +func boolptr(b bool) *bool { return &b } + +func protocolptr(p v1.Protocol) *v1.Protocol { return &p } + type service struct { s *v1.Service ns string diff --git a/src/vizier/services/metadata/controllers/k8smeta/k8s_metadata_utils.go b/src/vizier/services/metadata/controllers/k8smeta/k8s_metadata_utils.go index 1c21e315055..9ac26cac8b2 100644 --- a/src/vizier/services/metadata/controllers/k8smeta/k8s_metadata_utils.go +++ b/src/vizier/services/metadata/controllers/k8smeta/k8s_metadata_utils.go @@ -19,12 +19,16 @@ package k8smeta import ( + "sort" + log "github.com/sirupsen/logrus" apps "k8s.io/api/apps/v1" v1 "k8s.io/api/core/v1" + discoveryv1 "k8s.io/api/discovery/v1" "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/watch" "k8s.io/client-go/informers" + discoverylisters "k8s.io/client-go/listers/discovery/v1" "k8s.io/client-go/tools/cache" "px.dev/pixie/src/shared/k8s" @@ -127,12 +131,43 @@ func startServiceWatcher(ch chan *K8sResourceMessage, quitCh <-chan struct{}, fa func startEndpointsWatcher(ch chan *K8sResourceMessage, quitCh <-chan struct{}, factories []informers.SharedInformerFactory) { for _, factory := range factories { - inf := factory.Core().V1().Endpoints().Informer() - _, _ = inf.AddEventHandler(createHandlers(endpointsConverter, ch)) + slices := factory.Discovery().V1().EndpointSlices() + inf := slices.Informer() + _, _ = inf.AddEventHandler(createEndpointSliceHandlers(slices.Lister(), ch)) go inf.Run(quitCh) } } +// createEndpointSliceHandlers sends an endpoints update for the whole Service whenever one of +// its EndpointSlices changes, so the rest of the metadata service keeps seeing one Endpoints +// object per Service like it did with the v1 Endpoints API. +func createEndpointSliceHandlers(lister discoverylisters.EndpointSliceLister, ch chan *K8sResourceMessage) cache.ResourceEventHandler { + send := func(obj interface{}, eventType watch.EventType) { + if tombstone, ok := obj.(cache.DeletedFinalStateUnknown); ok { + obj = tombstone.Obj + } + slice, ok := obj.(*discoveryv1.EndpointSlice) + if !ok { + return + } + msg, lastSlice := endpointSlicesConverter(lister, slice, eventType == watch.Deleted) + if msg == nil { + return + } + msg.EventType = eventType + if eventType == watch.Deleted && !lastSlice { + // The Service still has other slices, so this is an update, not a delete. + msg.EventType = watch.Modified + } + ch <- msg + } + return cache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { send(obj, watch.Added) }, + UpdateFunc: func(_, newObj interface{}) { send(newObj, watch.Modified) }, + DeleteFunc: func(obj interface{}) { send(obj, watch.Deleted) }, + } +} + func startReplicaSetWatcher(ch chan *K8sResourceMessage, quitCh <-chan struct{}, factories []informers.SharedInformerFactory) { for _, factory := range factories { inf := factory.Apps().V1().ReplicaSets().Informer() @@ -182,15 +217,41 @@ func namespaceConverter(obj interface{}) *K8sResourceMessage { } } -func endpointsConverter(obj interface{}) *K8sResourceMessage { +// endpointSlicesConverter builds the endpoints message for the Service that owns the given slice +// from all of that Service's slices in the informer cache. When the deleted slice was the last +// one, the message is built from it so the delete still carries the last known addresses. +func endpointSlicesConverter(lister discoverylisters.EndpointSliceLister, slice *discoveryv1.EndpointSlice, deleted bool) (*K8sResourceMessage, bool) { + serviceName := slice.Labels[discoveryv1.LabelServiceName] + if serviceName == "" { + return nil, false + } + selector := labels.SelectorFromSet(labels.Set{discoveryv1.LabelServiceName: serviceName}) + slices, err := lister.EndpointSlices(slice.Namespace).List(selector) + if err != nil { + log.WithError(err).Error("Failed to list EndpointSlices") + return nil, false + } + deletedLast := false + if deleted && len(slices) == 0 { + slices = []*discoveryv1.EndpointSlice{slice} + deletedLast = true + } + sort.Slice(slices, func(i, j int) bool { + // Prefer IPv4 addresses for dual-stack services, then keep the order stable. + if slices[i].AddressType != slices[j].AddressType { + return slices[i].AddressType == discoveryv1.AddressTypeIPv4 + } + return slices[i].Name < slices[j].Name + }) + return &K8sResourceMessage{ ObjectType: "endpoints", Object: &storepb.K8SResource{ Resource: &storepb.K8SResource_Endpoints{ - Endpoints: k8s.EndpointsToProto(obj.(*v1.Endpoints)), + Endpoints: k8s.EndpointSlicesToProto(serviceName, slices), }, }, - } + }, deletedLast } func nodeConverter(obj interface{}) *K8sResourceMessage {