Use EndpointSlices instead of v1 Endpoints in the metadata service

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

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

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",

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))

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 {

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",