Port "Service endpoints latency should not be very high" to EndpointSlices

This changes the conformance test to measure the latency of the
EndpointSlice controller rather than the Endpoints controller. This
should not have any real effect since the two controllers run in the
same process and do mostly the same thing (especially in tests like
this one which don't test multi-slice services).

test/e2e/network/service_latency.go

@@ -25,6 +25,7 @@ import (
 
 	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/runtime"
 	"k8s.io/apimachinery/pkg/util/sets"
@@ -59,8 +60,8 @@ var _ = common.SIGDescribe("Service endpoints latency", func() {
 	framework.ConformanceIt("should not be very high", func(ctx context.Context) {
 		const (
 			// These are very generous criteria. Ideally we will
-			// get this much lower in the future. See issue
-			// #10436.
+			// get this much lower in the future. See
+			// https://issues.k8s.io/10436.
 			limitMedian = time.Second * 20
 			limitTail   = time.Second * 50
 
@@ -144,7 +145,7 @@ func runServiceLatencies(ctx context.Context, f *framework.Framework, inParallel
 	framework.ExpectNoError(err)
 	// Run a single watcher, to reduce the number of API calls we have to
 	// make; this is to minimize the timing error. It's how kube-proxy
-	// consumes the endpoints data, so it seems like the right thing to
+	// consumes the endpoint data, so it seems like the right thing to
 	// test.
 	endpointQueries := newQuerier()
 	startEndpointWatcher(ctx, f, endpointQueries)
@@ -196,9 +197,9 @@ func runServiceLatencies(ctx context.Context, f *framework.Framework, inParallel
 }
 
 type endpointQuery struct {
-	endpointsName string
-	endpoints     *v1.Endpoints
-	result        chan<- struct{}
+	serviceName string
+	slice       *discoveryv1.EndpointSlice
+	result      chan<- struct{}
 }
 
 type endpointQueries struct {
@@ -206,7 +207,7 @@ type endpointQueries struct {
 
 	stop        chan struct{}
 	requestChan chan *endpointQuery
-	seenChan    chan *v1.Endpoints
+	seenChan    chan *discoveryv1.EndpointSlice
 }
 
 func newQuerier() *endpointQueries {
@@ -215,7 +216,7 @@ func newQuerier() *endpointQueries {
 
 		stop:        make(chan struct{}, 100),
 		requestChan: make(chan *endpointQuery),
-		seenChan:    make(chan *v1.Endpoints, 100),
+		seenChan:    make(chan *discoveryv1.EndpointSlice, 100),
 	}
 	go eq.join()
 	return eq
@@ -225,6 +226,9 @@ func newQuerier() *endpointQueries {
 // nice properties like:
 //   - remembering an endpoint if it happens to arrive before it is requested.
 //   - closing all outstanding requests (returning nil) if it is stopped.
+//
+// Note that this test case uses a single-stack Service with a single endpoint. Thus, it's
+// guaranteed it will have only a single endpoint IP and thus a single EndpointSlice.
 func (eq *endpointQueries) join() {
 	defer func() {
 		// Terminate all pending requests, so that no goroutine will
@@ -241,22 +245,23 @@ func (eq *endpointQueries) join() {
 		case <-eq.stop:
 			return
 		case req := <-eq.requestChan:
-			if cur, ok := eq.requests[req.endpointsName]; ok && cur.endpoints != nil {
+			if cur, ok := eq.requests[req.serviceName]; ok && cur.slice != nil {
 				// We've already gotten the result, so we can
 				// immediately satisfy this request.
-				delete(eq.requests, req.endpointsName)
-				req.endpoints = cur.endpoints
+				delete(eq.requests, req.serviceName)
+				req.slice = cur.slice
 				close(req.result)
 			} else {
 				// Save this request.
-				eq.requests[req.endpointsName] = req
+				eq.requests[req.serviceName] = req
 			}
 		case got := <-eq.seenChan:
-			if req, ok := eq.requests[got.Name]; ok {
+			serviceName := got.Labels[discoveryv1.LabelServiceName]
+			if req, ok := eq.requests[serviceName]; ok {
 				if req.result != nil {
 					// Satisfy a request.
-					delete(eq.requests, got.Name)
-					req.endpoints = got
+					delete(eq.requests, serviceName)
+					req.slice = got
 					close(req.result)
 				}
 				// We've already recorded a result, but
@@ -265,8 +270,8 @@ func (eq *endpointQueries) join() {
 			} else {
 				// We haven't gotten the corresponding request
 				// yet, save this result.
-				eq.requests[got.Name] = &endpointQuery{
-					endpoints: got,
+				eq.requests[serviceName] = &endpointQuery{
+					slice: got,
 				}
 			}
 		}
@@ -274,19 +279,19 @@ func (eq *endpointQueries) join() {
 }
 
 // request blocks until the requested endpoint is seen.
-func (eq *endpointQueries) request(endpointsName string) *v1.Endpoints {
+func (eq *endpointQueries) request(serviceName string) *discoveryv1.EndpointSlice {
 	result := make(chan struct{})
 	req := &endpointQuery{
-		endpointsName: endpointsName,
-		result:        result,
+		serviceName: serviceName,
+		result:      result,
 	}
 	eq.requestChan <- req
 	<-result
-	return req.endpoints
+	return req.slice
 }
 
 // marks e as added; does not block.
-func (eq *endpointQueries) added(e *v1.Endpoints) {
+func (eq *endpointQueries) added(e *discoveryv1.EndpointSlice) {
 	eq.seenChan <- e
 }
 
@@ -295,26 +300,26 @@ func startEndpointWatcher(ctx context.Context, f *framework.Framework, q *endpoi
 	_, controller := cache.NewInformer(
 		&cache.ListWatch{
 			ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
-				obj, err := f.ClientSet.CoreV1().Endpoints(f.Namespace.Name).List(ctx, options)
+				obj, err := f.ClientSet.DiscoveryV1().EndpointSlices(f.Namespace.Name).List(ctx, options)
 				return runtime.Object(obj), err
 			},
 			WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
-				return f.ClientSet.CoreV1().Endpoints(f.Namespace.Name).Watch(ctx, options)
+				return f.ClientSet.DiscoveryV1().EndpointSlices(f.Namespace.Name).Watch(ctx, options)
 			},
 		},
-		&v1.Endpoints{},
+		&discoveryv1.EndpointSlice{},
 		0,
 		cache.ResourceEventHandlerFuncs{
 			AddFunc: func(obj interface{}) {
-				if e, ok := obj.(*v1.Endpoints); ok {
-					if len(e.Subsets) > 0 && len(e.Subsets[0].Addresses) > 0 {
+				if e, ok := obj.(*discoveryv1.EndpointSlice); ok {
+					if len(e.Endpoints) > 0 {
 						q.added(e)
 					}
 				}
 			},
 			UpdateFunc: func(old, cur interface{}) {
-				if e, ok := cur.(*v1.Endpoints); ok {
-					if len(e.Subsets) > 0 && len(e.Subsets[0].Addresses) > 0 {
+				if e, ok := cur.(*discoveryv1.EndpointSlice); ok {
+					if len(e.Endpoints) > 0 {
 						q.added(e)
 					}
 				}