allocator_test.go

Documentation: k8s.io/kubernetes/test/integration/servicecidr

     1  /*
     2  Copyright 2023 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 servicecidr
    18  
    19  import (
    20  	"context"
    21  	"fmt"
    22  	"strings"
    23  	"testing"
    24  	"time"
    25  
    26  	v1 "k8s.io/api/core/v1"
    27  	apierrors "k8s.io/apimachinery/pkg/api/errors"
    28  	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    29  	"k8s.io/apimachinery/pkg/runtime"
    30  	"k8s.io/apimachinery/pkg/util/wait"
    31  	utilfeature "k8s.io/apiserver/pkg/util/feature"
    32  	"k8s.io/client-go/kubernetes"
    33  	featuregatetesting "k8s.io/component-base/featuregate/testing"
    34  	"k8s.io/kubernetes/cmd/kube-apiserver/app/options"
    35  	kubeapiservertesting "k8s.io/kubernetes/cmd/kube-apiserver/app/testing"
    36  	"k8s.io/kubernetes/pkg/api/legacyscheme"
    37  	"k8s.io/kubernetes/pkg/features"
    38  	"k8s.io/kubernetes/test/integration/framework"
    39  	"k8s.io/kubernetes/test/utils/ktesting"
    40  	netutils "k8s.io/utils/net"
    41  )
    42  
    43  func TestServiceAlloc(t *testing.T) {
    44  	// Create an IPv4 single stack control-plane
    45  	serviceCIDR := "192.168.0.0/29"
    46  
    47  	tCtx := ktesting.Init(t)
    48  	client, _, tearDownFn := framework.StartTestServer(tCtx, t, framework.TestServerSetup{
    49  		ModifyServerRunOptions: func(opts *options.ServerRunOptions) {
    50  			opts.ServiceClusterIPRanges = serviceCIDR
    51  		},
    52  	})
    53  	defer tearDownFn()
    54  
    55  	svc := func(i int) *v1.Service {
    56  		return &v1.Service{
    57  			ObjectMeta: metav1.ObjectMeta{
    58  				Name: fmt.Sprintf("svc-%v", i),
    59  			},
    60  			Spec: v1.ServiceSpec{
    61  				Type: v1.ServiceTypeClusterIP,
    62  				Ports: []v1.ServicePort{
    63  					{Port: 80},
    64  				},
    65  			},
    66  		}
    67  	}
    68  
    69  	// Wait until the default "kubernetes" service is created.
    70  	if err := wait.Poll(250*time.Millisecond, time.Minute, func() (bool, error) {
    71  		_, err := client.CoreV1().Services(metav1.NamespaceDefault).Get(context.TODO(), "kubernetes", metav1.GetOptions{})
    72  		if err != nil && !apierrors.IsNotFound(err) {
    73  			return false, err
    74  		}
    75  		return !apierrors.IsNotFound(err), nil
    76  	}); err != nil {
    77  		t.Fatalf("creating kubernetes service timed out")
    78  	}
    79  
    80  	// make 5 more services to take up all IPs
    81  	for i := 0; i < 5; i++ {
    82  		if _, err := client.CoreV1().Services(metav1.NamespaceDefault).Create(context.TODO(), svc(i), metav1.CreateOptions{}); err != nil {
    83  			t.Error(err)
    84  		}
    85  	}
    86  
    87  	// Make another service. It will fail because we're out of cluster IPs
    88  	if _, err := client.CoreV1().Services(metav1.NamespaceDefault).Create(context.TODO(), svc(8), metav1.CreateOptions{}); err != nil {
    89  		if !strings.Contains(err.Error(), "range is full") {
    90  			t.Errorf("unexpected error text: %v", err)
    91  		}
    92  	} else {
    93  		svcs, err := client.CoreV1().Services(metav1.NamespaceAll).List(context.TODO(), metav1.ListOptions{})
    94  		if err != nil {
    95  			t.Fatalf("unexpected success, and error getting the services: %v", err)
    96  		}
    97  		allIPs := []string{}
    98  		for _, s := range svcs.Items {
    99  			allIPs = append(allIPs, s.Spec.ClusterIP)
   100  		}
   101  		t.Fatalf("unexpected creation success. The following IPs exist: %#v. It should only be possible to allocate 2 IP addresses in this cluster.\n\n%#v", allIPs, svcs)
   102  	}
   103  
   104  	// Delete the first service.
   105  	if err := client.CoreV1().Services(metav1.NamespaceDefault).Delete(context.TODO(), svc(1).ObjectMeta.Name, metav1.DeleteOptions{}); err != nil {
   106  		t.Fatalf("got unexpected error: %v", err)
   107  	}
   108  
   109  	// This time creating the second service should work.
   110  	if _, err := client.CoreV1().Services(metav1.NamespaceDefault).Create(context.TODO(), svc(8), metav1.CreateOptions{}); err != nil {
   111  		t.Fatalf("got unexpected error: %v", err)
   112  	}
   113  }
   114  
   115  func TestServiceAllocIPAddress(t *testing.T) {
   116  	// Create an IPv6 single stack control-plane with a large range
   117  	serviceCIDR := "2001:db8::/64"
   118  	defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.MultiCIDRServiceAllocator, true)()
   119  
   120  	tCtx := ktesting.Init(t)
   121  	client, _, tearDownFn := framework.StartTestServer(tCtx, t, framework.TestServerSetup{
   122  		ModifyServerRunOptions: func(opts *options.ServerRunOptions) {
   123  			opts.ServiceClusterIPRanges = serviceCIDR
   124  			opts.GenericServerRunOptions.AdvertiseAddress = netutils.ParseIPSloppy("2001:db8::10")
   125  			opts.APIEnablement.RuntimeConfig.Set("networking.k8s.io/v1alpha1=true")
   126  		},
   127  	})
   128  	defer tearDownFn()
   129  
   130  	svc := func(i int) *v1.Service {
   131  		return &v1.Service{
   132  			ObjectMeta: metav1.ObjectMeta{
   133  				Name: fmt.Sprintf("svc-%v", i),
   134  			},
   135  			Spec: v1.ServiceSpec{
   136  				Type: v1.ServiceTypeClusterIP,
   137  				Ports: []v1.ServicePort{
   138  					{Port: 80},
   139  				},
   140  			},
   141  		}
   142  	}
   143  
   144  	// Wait until the default "kubernetes" service is created.
   145  	if err := wait.Poll(250*time.Millisecond, time.Minute, func() (bool, error) {
   146  		_, err := client.CoreV1().Services(metav1.NamespaceDefault).Get(tCtx, "kubernetes", metav1.GetOptions{})
   147  		if err != nil && !apierrors.IsNotFound(err) {
   148  			return false, err
   149  		}
   150  		return !apierrors.IsNotFound(err), nil
   151  	}); err != nil {
   152  		t.Fatalf("creating kubernetes service timed out")
   153  	}
   154  
   155  	// create 5 random services and check that the Services have an IP associated
   156  	for i := 0; i < 5; i++ {
   157  		svc, err := client.CoreV1().Services(metav1.NamespaceDefault).Create(tCtx, svc(i), metav1.CreateOptions{})
   158  		if err != nil {
   159  			t.Error(err)
   160  		}
   161  		_, err = client.NetworkingV1alpha1().IPAddresses().Get(tCtx, svc.Spec.ClusterIP, metav1.GetOptions{})
   162  		if err != nil {
   163  			t.Error(err)
   164  		}
   165  	}
   166  
   167  	// Make a service in the top of the range to verify we can allocate in the whole range
   168  	// because it is not reasonable to create 2^64 services
   169  	lastSvc := svc(8)
   170  	lastSvc.Spec.ClusterIP = "2001:db8::ffff:ffff:ffff:ffff"
   171  	if _, err := client.CoreV1().Services(metav1.NamespaceDefault).Create(context.TODO(), lastSvc, metav1.CreateOptions{}); err != nil {
   172  		t.Errorf("unexpected error text: %v", err)
   173  	}
   174  	_, err := client.NetworkingV1alpha1().IPAddresses().Get(context.TODO(), lastSvc.Spec.ClusterIP, metav1.GetOptions{})
   175  	if err != nil {
   176  		t.Error(err)
   177  	}
   178  
   179  }
   180  
   181  func TestMigrateService(t *testing.T) {
   182  	defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.MultiCIDRServiceAllocator, true)()
   183  	//logs.GlogSetter("7")
   184  
   185  	etcdOptions := framework.SharedEtcd()
   186  	apiServerOptions := kubeapiservertesting.NewDefaultTestServerOptions()
   187  	s := kubeapiservertesting.StartTestServerOrDie(t,
   188  		apiServerOptions,
   189  		[]string{
   190  			"--runtime-config=networking.k8s.io/v1alpha1=true",
   191  			"--service-cluster-ip-range=10.0.0.0/24",
   192  			"--advertise-address=10.1.1.1",
   193  			"--disable-admission-plugins=ServiceAccount",
   194  		},
   195  		etcdOptions)
   196  	defer s.TearDownFn()
   197  	serviceName := "test-old-service"
   198  	namespace := "old-service-ns"
   199  	// Create a service and store it in etcd
   200  	svc := &v1.Service{
   201  		ObjectMeta: metav1.ObjectMeta{
   202  			Name:              serviceName,
   203  			Namespace:         namespace,
   204  			CreationTimestamp: metav1.Now(),
   205  			UID:               "08675309-9376-9376-9376-086753099999",
   206  		},
   207  		Spec: v1.ServiceSpec{
   208  			ClusterIP: "10.0.0.11",
   209  			Ports: []v1.ServicePort{
   210  				{
   211  					Name: "test-port",
   212  					Port: 81,
   213  				},
   214  			},
   215  		},
   216  	}
   217  	svcJSON, err := runtime.Encode(legacyscheme.Codecs.LegacyCodec(v1.SchemeGroupVersion), svc)
   218  	if err != nil {
   219  		t.Fatalf("Failed creating service JSON: %v", err)
   220  	}
   221  	key := "/" + etcdOptions.Prefix + "/services/specs/" + namespace + "/" + serviceName
   222  	if _, err := s.EtcdClient.Put(context.Background(), key, string(svcJSON)); err != nil {
   223  		t.Error(err)
   224  	}
   225  	t.Logf("Service stored in etcd %v", string(svcJSON))
   226  
   227  	kubeclient, err := kubernetes.NewForConfig(s.ClientConfig)
   228  	if err != nil {
   229  		t.Fatalf("Unexpected error: %v", err)
   230  	}
   231  	ns := framework.CreateNamespaceOrDie(kubeclient, namespace, t)
   232  	defer framework.DeleteNamespaceOrDie(kubeclient, ns, t)
   233  
   234  	// TODO: Understand why the Service can not be obtained with a List, it only works if we trigger an event
   235  	// by updating the Service.
   236  	_, err = kubeclient.CoreV1().Services(namespace).Update(context.Background(), svc, metav1.UpdateOptions{})
   237  	if err != nil {
   238  		t.Error(err)
   239  	}
   240  
   241  	err = wait.PollImmediate(1*time.Second, 10*time.Second, func() (bool, error) {
   242  		// The repair loop must create the IP address associated
   243  		_, err = kubeclient.NetworkingV1alpha1().IPAddresses().Get(context.TODO(), svc.Spec.ClusterIP, metav1.GetOptions{})
   244  		if err != nil {
   245  			return false, nil
   246  		}
   247  		return true, nil
   248  	})
   249  	if err != nil {
   250  		t.Error(err)
   251  	}
   252  
   253  }
   254  
   255  func TestSkewedAllocators(t *testing.T) {
   256  	svc := func(i int) *v1.Service {
   257  		return &v1.Service{
   258  			ObjectMeta: metav1.ObjectMeta{
   259  				Name: fmt.Sprintf("svc-%v", i),
   260  			},
   261  			Spec: v1.ServiceSpec{
   262  				Type: v1.ServiceTypeClusterIP,
   263  				Ports: []v1.ServicePort{
   264  					{Port: 80},
   265  				},
   266  			},
   267  		}
   268  	}
   269  
   270  	etcdOptions := framework.SharedEtcd()
   271  	apiServerOptions := kubeapiservertesting.NewDefaultTestServerOptions()
   272  	// s1 uses IPAddress allocator
   273  	s1 := kubeapiservertesting.StartTestServerOrDie(t, apiServerOptions,
   274  		[]string{
   275  			"--runtime-config=networking.k8s.io/v1alpha1=true",
   276  			"--service-cluster-ip-range=10.0.0.0/24",
   277  			"--disable-admission-plugins=ServiceAccount",
   278  			fmt.Sprintf("--feature-gates=%s=true", features.MultiCIDRServiceAllocator)},
   279  		etcdOptions)
   280  	defer s1.TearDownFn()
   281  
   282  	kubeclient1, err := kubernetes.NewForConfig(s1.ClientConfig)
   283  	if err != nil {
   284  		t.Fatalf("Unexpected error: %v", err)
   285  	}
   286  
   287  	// create 5 random services and check that the Services have an IP associated
   288  	for i := 0; i < 5; i++ {
   289  		service, err := kubeclient1.CoreV1().Services(metav1.NamespaceDefault).Create(context.TODO(), svc(i), metav1.CreateOptions{})
   290  		if err != nil {
   291  			t.Error(err)
   292  			continue
   293  		}
   294  		_, err = kubeclient1.NetworkingV1alpha1().IPAddresses().Get(context.TODO(), service.Spec.ClusterIP, metav1.GetOptions{})
   295  		if err != nil {
   296  			t.Error(err)
   297  		}
   298  	}
   299  
   300  	// s2 uses bitmap allocator
   301  	s2 := kubeapiservertesting.StartTestServerOrDie(t, apiServerOptions,
   302  		[]string{
   303  			"--runtime-config=networking.k8s.io/v1alpha1=false",
   304  			"--service-cluster-ip-range=10.0.0.0/24",
   305  			"--disable-admission-plugins=ServiceAccount",
   306  			fmt.Sprintf("--feature-gates=%s=false", features.MultiCIDRServiceAllocator)},
   307  		etcdOptions)
   308  	defer s2.TearDownFn()
   309  
   310  	kubeclient2, err := kubernetes.NewForConfig(s2.ClientConfig)
   311  	if err != nil {
   312  		t.Fatalf("Unexpected error: %v", err)
   313  	}
   314  
   315  	// create 5 random services and check that the Services have an IP associated
   316  	for i := 5; i < 10; i++ {
   317  		service, err := kubeclient2.CoreV1().Services(metav1.NamespaceDefault).Create(context.TODO(), svc(i), metav1.CreateOptions{})
   318  		if err != nil {
   319  			t.Error(err)
   320  		}
   321  
   322  		err = wait.PollImmediate(1*time.Second, 10*time.Second, func() (bool, error) {
   323  			// The repair loop must create the IP address associated
   324  			_, err = kubeclient1.NetworkingV1alpha1().IPAddresses().Get(context.TODO(), service.Spec.ClusterIP, metav1.GetOptions{})
   325  			if err != nil {
   326  				return false, nil
   327  			}
   328  			return true, nil
   329  		})
   330  		if err != nil {
   331  			t.Error(err)
   332  		}
   333  
   334  	}
   335  
   336  }
   337  
   338  func TestFlagsIPAllocator(t *testing.T) {
   339  	svc := func(i int) *v1.Service {
   340  		return &v1.Service{
   341  			ObjectMeta: metav1.ObjectMeta{
   342  				Name: fmt.Sprintf("svc-%v", i),
   343  			},
   344  			Spec: v1.ServiceSpec{
   345  				Type: v1.ServiceTypeClusterIP,
   346  				Ports: []v1.ServicePort{
   347  					{Port: 80},
   348  				},
   349  			},
   350  		}
   351  	}
   352  
   353  	etcdOptions := framework.SharedEtcd()
   354  	apiServerOptions := kubeapiservertesting.NewDefaultTestServerOptions()
   355  	// s1 uses IPAddress allocator
   356  	s1 := kubeapiservertesting.StartTestServerOrDie(t, apiServerOptions,
   357  		[]string{
   358  			"--runtime-config=networking.k8s.io/v1alpha1=true",
   359  			"--service-cluster-ip-range=10.0.0.0/24",
   360  			fmt.Sprintf("--feature-gates=%s=true", features.MultiCIDRServiceAllocator)},
   361  		etcdOptions)
   362  	defer s1.TearDownFn()
   363  
   364  	kubeclient1, err := kubernetes.NewForConfig(s1.ClientConfig)
   365  	if err != nil {
   366  		t.Fatalf("Unexpected error: %v", err)
   367  	}
   368  
   369  	// create 5 random services and check that the Services have an IP associated
   370  	for i := 0; i < 5; i++ {
   371  		service, err := kubeclient1.CoreV1().Services(metav1.NamespaceDefault).Create(context.TODO(), svc(i), metav1.CreateOptions{})
   372  		if err != nil {
   373  			t.Error(err)
   374  			continue
   375  		}
   376  		_, err = kubeclient1.NetworkingV1alpha1().IPAddresses().Get(context.TODO(), service.Spec.ClusterIP, metav1.GetOptions{})
   377  		if err != nil {
   378  			t.Error(err)
   379  		}
   380  	}
   381  
   382  }
   383
View as plain text