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 {