endpointslicemirroring.go

Documentation: k8s.io/kubernetes/test/e2e/network

     1  /*
     2  Copyright 2020 The Kubernetes Authors.
     3  
     4  Licensed under the Apache License, Version 2.0 (the "License");
     5  you may not use this file except in compliance with the License.
     6  You may obtain a copy of the License at
     7  
     8      http://www.apache.org/licenses/LICENSE-2.0
     9  
    10  Unless required by applicable law or agreed to in writing, software
    11  distributed under the License is distributed on an "AS IS" BASIS,
    12  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    13  See the License for the specific language governing permissions and
    14  limitations under the License.
    15  */
    16  
    17  package network
    18  
    19  import (
    20  	"context"
    21  	"fmt"
    22  	"net"
    23  	"time"
    24  
    25  	"github.com/onsi/ginkgo/v2"
    26  	v1 "k8s.io/api/core/v1"
    27  	discoveryv1 "k8s.io/api/discovery/v1"
    28  	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    29  	"k8s.io/apimachinery/pkg/util/wait"
    30  	clientset "k8s.io/client-go/kubernetes"
    31  	"k8s.io/kubernetes/test/e2e/framework"
    32  	e2epod "k8s.io/kubernetes/test/e2e/framework/pod"
    33  	"k8s.io/kubernetes/test/e2e/network/common"
    34  	admissionapi "k8s.io/pod-security-admission/api"
    35  )
    36  
    37  var _ = common.SIGDescribe("EndpointSliceMirroring", func() {
    38  	f := framework.NewDefaultFramework("endpointslicemirroring")
    39  	f.NamespacePodSecurityLevel = admissionapi.LevelPrivileged
    40  
    41  	var cs clientset.Interface
    42  
    43  	ginkgo.BeforeEach(func() {
    44  		cs = f.ClientSet
    45  	})
    46  
    47  	/*
    48  		Release: v1.21
    49  		Testname: EndpointSlice Mirroring
    50  		Description: The discovery.k8s.io API group MUST exist in the /apis discovery document.
    51  		The discovery.k8s.io/v1 API group/version MUST exist in the /apis/discovery.k8s.io discovery document.
    52  		The endpointslices resource MUST exist in the /apis/discovery.k8s.io/v1 discovery document.
    53  		The endpointslices mirrorowing must mirror endpoint create, update, and delete actions.
    54  	*/
    55  	framework.ConformanceIt("should mirror a custom Endpoints resource through create update and delete", func(ctx context.Context) {
    56  		svc := createServiceReportErr(ctx, cs, f.Namespace.Name, &v1.Service{
    57  			ObjectMeta: metav1.ObjectMeta{
    58  				Name: "example-custom-endpoints",
    59  			},
    60  			Spec: v1.ServiceSpec{
    61  				Ports: []v1.ServicePort{{
    62  					Name:     "example",
    63  					Port:     80,
    64  					Protocol: v1.ProtocolTCP,
    65  				}},
    66  			},
    67  		})
    68  
    69  		endpoints := &v1.Endpoints{
    70  			ObjectMeta: metav1.ObjectMeta{
    71  				Name: svc.Name,
    72  			},
    73  			Subsets: []v1.EndpointSubset{{
    74  				Ports: []v1.EndpointPort{{
    75  					Port: 80,
    76  				}},
    77  				Addresses: []v1.EndpointAddress{{
    78  					IP: "10.1.2.3",
    79  				}},
    80  			}},
    81  		}
    82  
    83  		ginkgo.By("mirroring a new custom Endpoint", func() {
    84  			_, err := cs.CoreV1().Endpoints(f.Namespace.Name).Create(ctx, endpoints, metav1.CreateOptions{})
    85  			framework.ExpectNoError(err, "Unexpected error creating Endpoints")
    86  
    87  			if err := wait.PollImmediate(2*time.Second, 12*time.Second, func() (bool, error) {
    88  				esList, err := cs.DiscoveryV1().EndpointSlices(f.Namespace.Name).List(ctx, metav1.ListOptions{
    89  					LabelSelector: discoveryv1.LabelServiceName + "=" + svc.Name,
    90  				})
    91  				if err != nil {
    92  					framework.Logf("Error listing EndpointSlices: %v", err)
    93  					return false, nil
    94  				}
    95  				if len(esList.Items) == 0 {
    96  					framework.Logf("Waiting for at least 1 EndpointSlice to exist, got %d", len(esList.Items))
    97  					return false, nil
    98  				}
    99  
   100  				// Due to informer caching, it's possible for the controller
   101  				// to create a second EndpointSlice if it does not see the
   102  				// first EndpointSlice that was created during a sync. All
   103  				// EndpointSlices created should be valid though.
   104  				for _, epSlice := range esList.Items {
   105  					if len(epSlice.Ports) != 1 {
   106  						return false, fmt.Errorf("Expected EndpointSlice to have 1 Port, got %d", len(epSlice.Ports))
   107  					}
   108  					port := epSlice.Ports[0]
   109  					if *port.Port != int32(80) {
   110  						return false, fmt.Errorf("Expected port to be 80, got %d", *port.Port)
   111  					}
   112  					if len(epSlice.Endpoints) != 1 {
   113  						return false, fmt.Errorf("Expected EndpointSlice to have 1 endpoints, got %d", len(epSlice.Endpoints))
   114  					}
   115  					endpoint := epSlice.Endpoints[0]
   116  					if len(endpoint.Addresses) != 1 {
   117  						return false, fmt.Errorf("Expected EndpointSlice endpoint to have 1 address, got %d", len(endpoint.Addresses))
   118  					}
   119  					address := endpoint.Addresses[0]
   120  					if address != "10.1.2.3" {
   121  						return false, fmt.Errorf("Expected EndpointSlice to have 10.1.2.3 as address, got %s", address)
   122  					}
   123  				}
   124  
   125  				return true, nil
   126  			}); err != nil {
   127  				framework.Failf("Did not find matching EndpointSlice for %s/%s: %s", svc.Namespace, svc.Name, err)
   128  			}
   129  		})
   130  
   131  		ginkgo.By("mirroring an update to a custom Endpoint", func() {
   132  			endpoints.Subsets[0].Addresses = []v1.EndpointAddress{{
   133  				IP: "10.2.3.4",
   134  			}}
   135  			_, err := cs.CoreV1().Endpoints(f.Namespace.Name).Update(ctx, endpoints, metav1.UpdateOptions{})
   136  			framework.ExpectNoError(err, "Unexpected error updating Endpoints")
   137  
   138  			// Expect mirrored EndpointSlice resource to be updated.
   139  			if err := wait.PollImmediate(2*time.Second, 12*time.Second, func() (bool, error) {
   140  				esList, err := cs.DiscoveryV1().EndpointSlices(f.Namespace.Name).List(ctx, metav1.ListOptions{
   141  					LabelSelector: discoveryv1.LabelServiceName + "=" + svc.Name,
   142  				})
   143  				if err != nil {
   144  					return false, err
   145  				}
   146  				if len(esList.Items) != 1 {
   147  					framework.Logf("Waiting for 1 EndpointSlice to exist, got %d", len(esList.Items))
   148  					return false, nil
   149  				}
   150  				epSlice := esList.Items[0]
   151  				if len(epSlice.Ports) != 1 {
   152  					framework.Logf("Expected EndpointSlice to have 1 Port, got %d", len(epSlice.Ports))
   153  					return false, nil
   154  				}
   155  				port := epSlice.Ports[0]
   156  				if *port.Port != int32(80) {
   157  					framework.Logf("Expected port to be 80, got %d", *port.Port)
   158  					return false, nil
   159  				}
   160  				if len(epSlice.Endpoints) != 1 {
   161  					framework.Logf("Expected EndpointSlice to have 1 endpoints, got %d", len(epSlice.Endpoints))
   162  					return false, nil
   163  				}
   164  				endpoint := epSlice.Endpoints[0]
   165  				if len(endpoint.Addresses) != 1 {
   166  					framework.Logf("Expected EndpointSlice endpoint to have 1 address, got %d", len(endpoint.Addresses))
   167  					return false, nil
   168  				}
   169  				address := endpoint.Addresses[0]
   170  				if address != "10.2.3.4" {
   171  					framework.Logf("Expected EndpointSlice to have 10.2.3.4 as address, got %s", address)
   172  					return false, nil
   173  				}
   174  
   175  				return true, nil
   176  			}); err != nil {
   177  				framework.Failf("Did not find matching EndpointSlice for %s/%s: %s", svc.Namespace, svc.Name, err)
   178  			}
   179  		})
   180  
   181  		ginkgo.By("mirroring deletion of a custom Endpoint", func() {
   182  			err := cs.CoreV1().Endpoints(f.Namespace.Name).Delete(ctx, endpoints.Name, metav1.DeleteOptions{})
   183  			framework.ExpectNoError(err, "Unexpected error deleting Endpoints")
   184  
   185  			// Expect mirrored EndpointSlice resource to be updated.
   186  			if err := wait.PollImmediate(2*time.Second, 12*time.Second, func() (bool, error) {
   187  				esList, err := cs.DiscoveryV1().EndpointSlices(f.Namespace.Name).List(ctx, metav1.ListOptions{
   188  					LabelSelector: discoveryv1.LabelServiceName + "=" + svc.Name,
   189  				})
   190  				if err != nil {
   191  					return false, err
   192  				}
   193  				if len(esList.Items) != 0 {
   194  					framework.Logf("Waiting for 0 EndpointSlices to exist, got %d", len(esList.Items))
   195  					return false, nil
   196  				}
   197  
   198  				return true, nil
   199  			}); err != nil {
   200  				framework.Failf("Did not find matching EndpointSlice for %s/%s: %s", svc.Namespace, svc.Name, err)
   201  			}
   202  		})
   203  	})
   204  
   205  	ginkgo.It("should mirror a custom Endpoint with multiple subsets and same IP address", func(ctx context.Context) {
   206  		ns := f.Namespace.Name
   207  		svc := createServiceReportErr(ctx, cs, f.Namespace.Name, &v1.Service{
   208  			ObjectMeta: metav1.ObjectMeta{
   209  				Name: "example-custom-endpoints",
   210  			},
   211  			Spec: v1.ServiceSpec{
   212  				Ports: []v1.ServicePort{
   213  					{
   214  						Name:     "port80",
   215  						Port:     80,
   216  						Protocol: v1.ProtocolTCP,
   217  					},
   218  					{
   219  						Name:     "port81",
   220  						Port:     81,
   221  						Protocol: v1.ProtocolTCP,
   222  					},
   223  				},
   224  			},
   225  		})
   226  
   227  		// Add a backend pod to the service in the other node
   228  		port8080 := []v1.ContainerPort{
   229  			{
   230  				ContainerPort: 8090,
   231  				Protocol:      v1.ProtocolTCP,
   232  			},
   233  		}
   234  		port9090 := []v1.ContainerPort{
   235  			{
   236  				ContainerPort: 9090,
   237  				Protocol:      v1.ProtocolTCP,
   238  			},
   239  		}
   240  
   241  		serverPod := e2epod.NewAgnhostPodFromContainers(
   242  			"", "pod-handle-http-request", nil,
   243  			e2epod.NewAgnhostContainer("container-handle-8090-request", nil, port8080, "netexec", "--http-port", "8090", "--udp-port", "-1"),
   244  			e2epod.NewAgnhostContainer("container-handle-9090-request", nil, port9090, "netexec", "--http-port", "9090", "--udp-port", "-1"),
   245  		)
   246  
   247  		pod := e2epod.NewPodClient(f).CreateSync(ctx, serverPod)
   248  
   249  		if pod.Status.PodIP == "" {
   250  			framework.Failf("PodIP not assigned for pod %s", pod.Name)
   251  		}
   252  
   253  		// create custom endpoints
   254  		endpoints := &v1.Endpoints{
   255  			ObjectMeta: metav1.ObjectMeta{
   256  				Name: svc.Name,
   257  			},
   258  			Subsets: []v1.EndpointSubset{
   259  				{
   260  					Ports: []v1.EndpointPort{{
   261  						Name: "port80",
   262  						Port: 8090,
   263  					}},
   264  					Addresses: []v1.EndpointAddress{{
   265  						IP: pod.Status.PodIP,
   266  					}},
   267  				},
   268  				{
   269  					Ports: []v1.EndpointPort{{
   270  						Name: "port81",
   271  						Port: 9090,
   272  					}},
   273  					Addresses: []v1.EndpointAddress{{
   274  						IP: pod.Status.PodIP,
   275  					}},
   276  				},
   277  			},
   278  		}
   279  
   280  		ginkgo.By("mirroring a new custom Endpoint", func() {
   281  			_, err := cs.CoreV1().Endpoints(f.Namespace.Name).Create(context.TODO(), endpoints, metav1.CreateOptions{})
   282  			framework.ExpectNoError(err, "Unexpected error creating Endpoints")
   283  
   284  			if err := wait.PollImmediate(2*time.Second, 12*time.Second, func() (bool, error) {
   285  				esList, err := cs.DiscoveryV1().EndpointSlices(f.Namespace.Name).List(context.TODO(), metav1.ListOptions{
   286  					LabelSelector: discoveryv1.LabelServiceName + "=" + svc.Name,
   287  				})
   288  				if err != nil {
   289  					framework.Logf("Error listing EndpointSlices: %v", err)
   290  					return false, nil
   291  				}
   292  				if len(esList.Items) == 0 {
   293  					framework.Logf("Waiting for at least 1 EndpointSlice to exist, got %d", len(esList.Items))
   294  					return false, nil
   295  				}
   296  				return true, nil
   297  			}); err != nil {
   298  				framework.Failf("Did not find matching EndpointSlice for %s/%s: %s", svc.Namespace, svc.Name, err)
   299  			}
   300  		})
   301  
   302  		// connect to the service must work
   303  		ginkgo.By("Creating a pause pods that will try to connect to the webservers")
   304  		pausePod := e2epod.NewAgnhostPod(ns, "pause-pod-0", nil, nil, nil)
   305  		e2epod.NewPodClient(f).CreateSync(ctx, pausePod)
   306  		dest1 := net.JoinHostPort(svc.Spec.ClusterIP, "80")
   307  		dest2 := net.JoinHostPort(svc.Spec.ClusterIP, "81")
   308  		execHostnameTest(*pausePod, dest1, pod.Name)
   309  		execHostnameTest(*pausePod, dest2, pod.Name)
   310  
   311  		// delete custom endpoints and wait until the endpoint slices are deleted too
   312  		ginkgo.By("mirroring deletion of a custom Endpoint", func() {
   313  			err := cs.CoreV1().Endpoints(f.Namespace.Name).Delete(context.TODO(), endpoints.Name, metav1.DeleteOptions{})
   314  			framework.ExpectNoError(err, "Unexpected error deleting Endpoints")
   315  
   316  			// Expect mirrored EndpointSlice resource to be updated.
   317  			if err := wait.PollImmediate(2*time.Second, 12*time.Second, func() (bool, error) {
   318  				esList, err := cs.DiscoveryV1().EndpointSlices(f.Namespace.Name).List(context.TODO(), metav1.ListOptions{
   319  					LabelSelector: discoveryv1.LabelServiceName + "=" + svc.Name,
   320  				})
   321  				if err != nil {
   322  					return false, err
   323  				}
   324  				if len(esList.Items) != 0 {
   325  					framework.Logf("Waiting for 0 EndpointSlices to exist, got %d", len(esList.Items))
   326  					return false, nil
   327  				}
   328  
   329  				return true, nil
   330  			}); err != nil {
   331  				framework.Failf("Did not find matching EndpointSlice for %s/%s: %s", svc.Namespace, svc.Name, err)
   332  			}
   333  		})
   334  	})
   335  })
   336
View as plain text