Adding integration and e2e tests to cover EndpointSliceMirroring

test/e2e/network/BUILD

@@ -16,6 +16,7 @@ go_library(
         "doc.go",
         "dual_stack.go",
         "endpointslice.go",
+        "endpointslicemirroring.go",
         "example_cluster_dns.go",
         "firewall.go",
         "fixture.go",

test/e2e/network/endpointslicemirroring.go

@@ -0,0 +1,185 @@
+/*
+Copyright 2020 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package network
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/onsi/ginkgo"
+	v1 "k8s.io/api/core/v1"
+	discoveryv1beta1 "k8s.io/api/discovery/v1beta1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
+	clientset "k8s.io/client-go/kubernetes"
+	"k8s.io/kubernetes/test/e2e/framework"
+)
+
+var _ = SIGDescribe("EndpointSliceMirroring", func() {
+	f := framework.NewDefaultFramework("endpointslicemirroring")
+
+	var cs clientset.Interface
+
+	ginkgo.BeforeEach(func() {
+		cs = f.ClientSet
+	})
+
+	ginkgo.It("should mirror a custom Endpoints resource through create update and delete", func() {
+		svc := createServiceReportErr(cs, f.Namespace.Name, &v1.Service{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "example-custom-endpoints",
+			},
+			Spec: v1.ServiceSpec{
+				Ports: []v1.ServicePort{{
+					Name:     "example",
+					Port:     80,
+					Protocol: v1.ProtocolTCP,
+				}},
+			},
+		})
+
+		endpoints := &v1.Endpoints{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: svc.Name,
+			},
+			Subsets: []v1.EndpointSubset{{
+				Ports: []v1.EndpointPort{{
+					Port: 80,
+				}},
+				Addresses: []v1.EndpointAddress{{
+					IP: "10.1.2.3",
+				}},
+			}},
+		}
+
+		ginkgo.By("mirroring a new custom Endpoint", func() {
+			_, err := cs.CoreV1().Endpoints(f.Namespace.Name).Create(context.TODO(), endpoints, metav1.CreateOptions{})
+			framework.ExpectNoError(err, "Unexpected error creating Endpoints")
+
+			if err := wait.PollImmediate(2*time.Second, 12*time.Second, func() (bool, error) {
+				esList, err := cs.DiscoveryV1beta1().EndpointSlices(f.Namespace.Name).List(context.TODO(), metav1.ListOptions{
+					LabelSelector: discoveryv1beta1.LabelServiceName + "=" + svc.Name,
+				})
+				if err != nil {
+					framework.Logf("Error listing EndpointSlices: %v", err)
+					return false, nil
+				}
+				if len(esList.Items) != 1 {
+					framework.Logf("Waiting for 1 EndpointSlice to exist, got %d", len(esList.Items))
+					return false, nil
+				}
+				epSlice := esList.Items[0]
+				if len(epSlice.Ports) != 1 {
+					return false, fmt.Errorf("Expected EndpointSlice to have 1 Port, got %d", len(epSlice.Ports))
+				}
+				port := epSlice.Ports[0]
+				if *port.Port != int32(80) {
+					return false, fmt.Errorf("Expected port to be 80, got %d", *port.Port)
+				}
+				if len(epSlice.Endpoints) != 1 {
+					return false, fmt.Errorf("Expected EndpointSlice to have 1 endpoints, got %d", len(epSlice.Endpoints))
+				}
+				endpoint := epSlice.Endpoints[0]
+				if len(endpoint.Addresses) != 1 {
+					return false, fmt.Errorf("Expected EndpointSlice endpoint to have 1 address, got %d", len(endpoint.Addresses))
+				}
+				address := endpoint.Addresses[0]
+				if address != "10.1.2.3" {
+					return false, fmt.Errorf("Expected EndpointSlice to have 10.1.2.3 as address, got %s", address)
+				}
+
+				return true, nil
+			}); err != nil {
+				framework.Failf("Did not find matching EndpointSlice for %s/%s: %s", svc.Namespace, svc.Name, err)
+			}
+		})
+
+		ginkgo.By("mirroring an update to a custom Endpoint", func() {
+			endpoints.Subsets[0].Addresses = []v1.EndpointAddress{{
+				IP: "10.2.3.4",
+			}}
+			_, err := cs.CoreV1().Endpoints(f.Namespace.Name).Update(context.TODO(), endpoints, metav1.UpdateOptions{})
+			framework.ExpectNoError(err, "Unexpected error updating Endpoints")
+
+			// Expect mirrored EndpointSlice resource to be updated.
+			if err := wait.PollImmediate(2*time.Second, 12*time.Second, func() (bool, error) {
+				esList, err := cs.DiscoveryV1beta1().EndpointSlices(f.Namespace.Name).List(context.TODO(), metav1.ListOptions{
+					LabelSelector: discoveryv1beta1.LabelServiceName + "=" + svc.Name,
+				})
+				if err != nil {
+					return false, err
+				}
+				if len(esList.Items) != 1 {
+					framework.Logf("Waiting for 1 EndpointSlice to exist, got %d", len(esList.Items))
+					return false, nil
+				}
+				epSlice := esList.Items[0]
+				if len(epSlice.Ports) != 1 {
+					framework.Logf("Expected EndpointSlice to have 1 Port, got %d", len(epSlice.Ports))
+					return false, nil
+				}
+				port := epSlice.Ports[0]
+				if *port.Port != int32(80) {
+					framework.Logf("Expected port to be 80, got %d", *port.Port)
+					return false, nil
+				}
+				if len(epSlice.Endpoints) != 1 {
+					framework.Logf("Expected EndpointSlice to have 1 endpoints, got %d", len(epSlice.Endpoints))
+					return false, nil
+				}
+				endpoint := epSlice.Endpoints[0]
+				if len(endpoint.Addresses) != 1 {
+					framework.Logf("Expected EndpointSlice endpoint to have 1 address, got %d", len(endpoint.Addresses))
+					return false, nil
+				}
+				address := endpoint.Addresses[0]
+				if address != "10.2.3.4" {
+					framework.Logf("Expected EndpointSlice to have 10.2.3.4 as address, got %s", address)
+					return false, nil
+				}
+
+				return true, nil
+			}); err != nil {
+				framework.Failf("Did not find matching EndpointSlice for %s/%s: %s", svc.Namespace, svc.Name, err)
+			}
+		})
+
+		ginkgo.By("mirroring deletion of a custom Endpoint", func() {
+			err := cs.CoreV1().Endpoints(f.Namespace.Name).Delete(context.TODO(), endpoints.Name, metav1.DeleteOptions{})
+			framework.ExpectNoError(err, "Unexpected error deleting Endpoints")
+
+			// Expect mirrored EndpointSlice resource to be updated.
+			if err := wait.PollImmediate(2*time.Second, 12*time.Second, func() (bool, error) {
+				esList, err := cs.DiscoveryV1beta1().EndpointSlices(f.Namespace.Name).List(context.TODO(), metav1.ListOptions{
+					LabelSelector: discoveryv1beta1.LabelServiceName + "=" + svc.Name,
+				})
+				if err != nil {
+					return false, err
+				}
+				if len(esList.Items) != 0 {
+					framework.Logf("Waiting for 0 EndpointSlices to exist, got %d", len(esList.Items))
+					return false, nil
+				}
+
+				return true, nil
+			}); err != nil {
+				framework.Failf("Did not find matching EndpointSlice for %s/%s: %s", svc.Namespace, svc.Name, err)
+			}
+		})
+	})
+})

test/integration/BUILD

@@ -50,6 +50,7 @@ filegroup(
         "//test/integration/deployment:all-srcs",
         "//test/integration/disruption:all-srcs",
         "//test/integration/dryrun:all-srcs",
+        "//test/integration/endpointslice:all-srcs",
         "//test/integration/etcd:all-srcs",
         "//test/integration/events:all-srcs",
         "//test/integration/evictions:all-srcs",

test/integration/endpointslice/BUILD

@@ -0,0 +1,37 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_test")
+
+go_test(
+    name = "go_default_test",
+    srcs = [
+        "endpointslicemirroring_test.go",
+        "main_test.go",
+    ],
+    tags = ["integration"],
+    deps = [
+        "//pkg/controller/endpoint:go_default_library",
+        "//pkg/controller/endpointslice:go_default_library",
+        "//pkg/controller/endpointslicemirroring:go_default_library",
+        "//staging/src/k8s.io/api/core/v1:go_default_library",
+        "//staging/src/k8s.io/api/discovery/v1beta1:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
+        "//staging/src/k8s.io/client-go/informers:go_default_library",
+        "//staging/src/k8s.io/client-go/kubernetes:go_default_library",
+        "//staging/src/k8s.io/client-go/rest:go_default_library",
+        "//test/integration/framework:go_default_library",
+    ],
+)
+
+filegroup(
+    name = "package-srcs",
+    srcs = glob(["**"]),
+    tags = ["automanaged"],
+    visibility = ["//visibility:private"],
+)
+
+filegroup(
+    name = "all-srcs",
+    srcs = [":package-srcs"],
+    tags = ["automanaged"],
+    visibility = ["//visibility:public"],
+)

test/integration/endpointslice/endpointslicemirroring_test.go

@@ -0,0 +1,227 @@
+/*
+Copyright 2020 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package endpointslice
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	discovery "k8s.io/api/discovery/v1beta1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/informers"
+	clientset "k8s.io/client-go/kubernetes"
+	restclient "k8s.io/client-go/rest"
+	"k8s.io/kubernetes/pkg/controller/endpoint"
+	"k8s.io/kubernetes/pkg/controller/endpointslice"
+	"k8s.io/kubernetes/pkg/controller/endpointslicemirroring"
+	"k8s.io/kubernetes/test/integration/framework"
+)
+
+func TestEndpointSliceMirroring(t *testing.T) {
+	masterConfig := framework.NewIntegrationTestMasterConfig()
+	_, server, closeFn := framework.RunAMaster(masterConfig)
+	defer closeFn()
+
+	config := restclient.Config{Host: server.URL}
+	client, err := clientset.NewForConfig(&config)
+	if err != nil {
+		t.Fatalf("Error creating clientset: %v", err)
+	}
+
+	resyncPeriod := 12 * time.Hour
+	informers := informers.NewSharedInformerFactory(client, resyncPeriod)
+
+	epController := endpoint.NewEndpointController(
+		informers.Core().V1().Pods(),
+		informers.Core().V1().Services(),
+		informers.Core().V1().Endpoints(),
+		client,
+		1*time.Second)
+
+	epsController := endpointslice.NewController(
+		informers.Core().V1().Pods(),
+		informers.Core().V1().Services(),
+		informers.Core().V1().Nodes(),
+		informers.Discovery().V1beta1().EndpointSlices(),
+		int32(100),
+		client,
+		1*time.Second)
+
+	epsmController := endpointslicemirroring.NewController(
+		informers.Core().V1().Endpoints(),
+		informers.Discovery().V1beta1().EndpointSlices(),
+		informers.Core().V1().Services(),
+		int32(100),
+		client,
+		1*time.Second)
+
+	// Start informer and controllers
+	stopCh := make(chan struct{})
+	defer close(stopCh)
+	informers.Start(stopCh)
+	go epController.Run(5, stopCh)
+	go epsController.Run(5, stopCh)
+	go epsmController.Run(5, stopCh)
+
+	testCases := []struct {
+		testName                     string
+		service                      *corev1.Service
+		endpoints                    *corev1.Endpoints
+		expectEndpointSlice          bool
+		expectEndpointSliceManagedBy string
+	}{{
+		testName: "Service with selector",
+		service: &corev1.Service{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "test-123",
+			},
+			Spec: corev1.ServiceSpec{
+				Ports: []corev1.ServicePort{{
+					Port: int32(80),
+				}},
+				Selector: map[string]string{
+					"foo": "bar",
+				},
+			},
+		},
+		endpoints: &corev1.Endpoints{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "test-123",
+			},
+		},
+		expectEndpointSlice:          true,
+		expectEndpointSliceManagedBy: "endpointslice-controller.k8s.io",
+	}, {
+		testName: "Service without selector",
+		service: &corev1.Service{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "test-123",
+			},
+			Spec: corev1.ServiceSpec{
+				Ports: []corev1.ServicePort{{
+					Port: int32(80),
+				}},
+			},
+		},
+		endpoints: &corev1.Endpoints{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "test-123",
+			},
+			Subsets: []corev1.EndpointSubset{{
+				Ports: []corev1.EndpointPort{{
+					Port: 80,
+				}},
+				Addresses: []corev1.EndpointAddress{{
+					IP: "10.0.0.1",
+				}},
+			}},
+		},
+		expectEndpointSlice:          true,
+		expectEndpointSliceManagedBy: "endpointslicemirroring-controller.k8s.io",
+	}, {
+		testName: "Service without Endpoints",
+		service: &corev1.Service{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "test-123",
+			},
+			Spec: corev1.ServiceSpec{
+				Ports: []corev1.ServicePort{{
+					Port: int32(80),
+				}},
+				Selector: map[string]string{
+					"foo": "bar",
+				},
+			},
+		},
+		endpoints:                    nil,
+		expectEndpointSlice:          true,
+		expectEndpointSliceManagedBy: "endpointslice-controller.k8s.io",
+	}, {
+		testName: "Endpoints without Service",
+		service:  nil,
+		endpoints: &corev1.Endpoints{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "test-123",
+			},
+		},
+		expectEndpointSlice: false,
+	}}
+
+	for i, tc := range testCases {
+		t.Run(tc.testName, func(t *testing.T) {
+			ns := framework.CreateTestingNamespace(fmt.Sprintf("test-endpointslice-mirroring-%d", i), server, t)
+			defer framework.DeleteTestingNamespace(ns, server, t)
+
+			resourceName := ""
+			if tc.service != nil {
+				resourceName = tc.service.Name
+				tc.service.Namespace = ns.Name
+				_, err = client.CoreV1().Services(ns.Name).Create(context.TODO(), tc.service, metav1.CreateOptions{})
+				if err != nil {
+					t.Fatalf("Error creating service: %v", err)
+				}
+			}
+
+			if tc.endpoints != nil {
+				resourceName = tc.endpoints.Name
+				tc.endpoints.Namespace = ns.Name
+				_, err = client.CoreV1().Endpoints(ns.Name).Create(context.TODO(), tc.endpoints, metav1.CreateOptions{})
+				if err != nil {
+					t.Fatalf("Error creating endpoints: %v", err)
+				}
+			}
+
+			err = wait.PollImmediate(1*time.Second, 5*time.Second, func() (bool, error) {
+				lSelector := discovery.LabelServiceName + "=" + resourceName
+				esList, err := client.DiscoveryV1beta1().EndpointSlices(ns.Name).List(context.TODO(), metav1.ListOptions{LabelSelector: lSelector})
+				if err != nil {
+					t.Logf("Error listing EndpointSlices: %v", err)
+					return false, err
+				}
+
+				if tc.expectEndpointSlice {
+					if len(esList.Items) == 0 {
+						t.Logf("Waiting for EndpointSlice to be created")
+						return false, nil
+					}
+					if len(esList.Items) > 1 {
+						return false, fmt.Errorf("Only expected 1 EndpointSlice, got %d", len(esList.Items))
+					}
+					endpointSlice := esList.Items[0]
+					if tc.expectEndpointSliceManagedBy != "" {
+						if endpointSlice.Labels[discovery.LabelManagedBy] != tc.expectEndpointSliceManagedBy {
+							return false, fmt.Errorf("Expected EndpointSlice to be managed by %s, got %s", tc.expectEndpointSliceManagedBy, endpointSlice.Labels[discovery.LabelManagedBy])
+						}
+					}
+				} else if len(esList.Items) > 0 {
+					t.Logf("Waiting for EndpointSlices to be removed, still %d", len(esList.Items))
+					return false, nil
+				}
+
+				return true, nil
+			})
+			if err != nil {
+				t.Fatalf("Timed out waiting for conditions: %v", err)
+			}
+		})
+	}
+
+}

test/integration/endpointslice/main_test.go

@@ -0,0 +1,27 @@
+/*
+Copyright 2020 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package endpointslice
+
+import (
+	"testing"
+
+	"k8s.io/kubernetes/test/integration/framework"
+)
+
+func TestMain(m *testing.M) {
+	framework.EtcdMain(m.Run)
+}