master count and lease endpoint tests

test/integration/master/BUILD

@@ -78,6 +78,7 @@ go_test(
         "//vendor/k8s.io/apiserver/pkg/authorization/authorizer:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/authorization/authorizerfactory:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/features:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/registry/generic/registry:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/server/options/encryptionconfig:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/storage/value:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/storage/value/encrypt/aes:go_default_library",

test/integration/master/crd_test.go

@@ -41,7 +41,7 @@ import (
 )
 
 func TestCRDShadowGroup(t *testing.T) {
-	result := kubeapiservertesting.StartTestServerOrDie(t, nil, framework.SharedEtcd())
+	result := kubeapiservertesting.StartTestServerOrDie(t, nil, nil, framework.SharedEtcd())
 	defer result.TearDownFn()
 
 	kubeclient, err := kubernetes.NewForConfig(result.ClientConfig)
@@ -109,7 +109,7 @@ func TestCRDShadowGroup(t *testing.T) {
 func TestCRD(t *testing.T) {
 	defer utilfeaturetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.Initializers, true)()
 
-	result := kubeapiservertesting.StartTestServerOrDie(t, []string{"--admission-control", "Initializers"}, framework.SharedEtcd())
+	result := kubeapiservertesting.StartTestServerOrDie(t, nil, []string{"--admission-control", "Initializers"}, framework.SharedEtcd())
 	defer result.TearDownFn()
 
 	kubeclient, err := kubernetes.NewForConfig(result.ClientConfig)

test/integration/master/kube_apiserver_test.go

@@ -18,13 +18,18 @@ package master
 
 import (
 	"encoding/json"
+	"fmt"
+	"reflect"
 	"strings"
 	"testing"
+	"time"
 
 	appsv1beta1 "k8s.io/api/apps/v1beta1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apiextensions-apiserver/pkg/apis/apiextensions"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/apiserver/pkg/registry/generic/registry"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/kube-aggregator/pkg/apis/apiregistration"
 	kubeapiservertesting "k8s.io/kubernetes/cmd/kube-apiserver/app/testing"
@@ -32,7 +37,7 @@ import (
 )
 
 func TestRun(t *testing.T) {
-	server := kubeapiservertesting.StartTestServerOrDie(t, nil, framework.SharedEtcd())
+	server := kubeapiservertesting.StartTestServerOrDie(t, nil, nil, framework.SharedEtcd())
 	defer server.TearDownFn()
 
 	client, err := kubernetes.NewForConfig(server.ClientConfig)
@@ -82,7 +87,7 @@ func TestRun(t *testing.T) {
 // apiextensions-server and the kube-aggregator server, both part of
 // the delegation chain in kube-apiserver.
 func TestOpenAPIDelegationChainPlumbing(t *testing.T) {
-	server := kubeapiservertesting.StartTestServerOrDie(t, nil, framework.SharedEtcd())
+	server := kubeapiservertesting.StartTestServerOrDie(t, nil, nil, framework.SharedEtcd())
 	defer server.TearDownFn()
 
 	kubeclient, err := kubernetes.NewForConfig(server.ClientConfig)
@@ -138,3 +143,110 @@ func TestOpenAPIDelegationChainPlumbing(t *testing.T) {
 		t.Errorf("missing path: %q", registrationPrefix)
 	}
 }
+
+// return the unique endpoint IPs
+func getEndpointIPs(endpoints *corev1.Endpoints) []string {
+	endpointMap := make(map[string]bool)
+	ips := make([]string, 0)
+	for _, subset := range endpoints.Subsets {
+		for _, address := range subset.Addresses {
+			if _, ok := endpointMap[address.IP]; !ok {
+				endpointMap[address.IP] = true
+				ips = append(ips, address.IP)
+			}
+		}
+	}
+	return ips
+}
+
+func verifyEndpointsWithIPs(servers []*kubeapiservertesting.TestServer, ips []string) bool {
+	listenAddresses := make([]string, 0)
+	for _, server := range servers {
+		listenAddresses = append(listenAddresses, server.ServerOpts.GenericServerRunOptions.AdvertiseAddress.String())
+	}
+	return reflect.DeepEqual(listenAddresses, ips)
+}
+
+func testReconcilersMasterLease(t *testing.T, leaseCount int, masterCount int) {
+	var leaseServers []*kubeapiservertesting.TestServer
+	var masterCountServers []*kubeapiservertesting.TestServer
+	etcd := framework.SharedEtcd()
+
+	instanceOptions := &kubeapiservertesting.TestServerInstanceOptions{
+		DisableStorageCleanup: true,
+	}
+
+	// cleanup the registry storage
+	defer registry.CleanupStorage()
+
+	// 1. start masterCount api servers
+	for i := 0; i < masterCount; i++ {
+		// start master count api server
+		server := kubeapiservertesting.StartTestServerOrDie(t, instanceOptions, []string{
+			"--endpoint-reconciler-type", "master-count",
+			"--advertise-address", fmt.Sprintf("10.0.1.%v", i+1),
+			"--apiserver-count", fmt.Sprintf("%v", masterCount),
+		}, etcd)
+		masterCountServers = append(masterCountServers, server)
+	}
+
+	// 2. verify master count servers have registered
+	if err := wait.PollImmediate(3*time.Second, 2*time.Minute, func() (bool, error) {
+		client, err := kubernetes.NewForConfig(masterCountServers[0].ClientConfig)
+		endpoints, err := client.CoreV1().Endpoints("default").Get("kubernetes", metav1.GetOptions{})
+		if err != nil {
+			t.Logf("error fetching endpoints: %v", err)
+			return false, nil
+		}
+		return verifyEndpointsWithIPs(masterCountServers, getEndpointIPs(endpoints)), nil
+	}); err != nil {
+		t.Fatalf("master count endpoints failed to register: %v", err)
+	}
+
+	// 3. start lease api servers
+	for i := 0; i < leaseCount; i++ {
+		options := []string{
+			"--endpoint-reconciler-type", "lease",
+			"--advertise-address", fmt.Sprintf("10.0.1.%v", i+10),
+		}
+		server := kubeapiservertesting.StartTestServerOrDie(t, instanceOptions, options, etcd)
+		defer server.TearDownFn()
+		leaseServers = append(leaseServers, server)
+	}
+
+	time.Sleep(3 * time.Second)
+
+	// 4. Shutdown the masterCount server
+	for _, server := range masterCountServers {
+		server.TearDownFn()
+	}
+
+	// 5. verify only leaseEndpoint servers left
+	if err := wait.PollImmediate(3*time.Second, 2*time.Minute, func() (bool, error) {
+		client, err := kubernetes.NewForConfig(leaseServers[0].ClientConfig)
+		if err != nil {
+			t.Logf("create client error: %v", err)
+			return false, nil
+		}
+		endpoints, err := client.CoreV1().Endpoints("default").Get("kubernetes", metav1.GetOptions{})
+		if err != nil {
+			t.Logf("error fetching endpoints: %v", err)
+			return false, nil
+		}
+		return verifyEndpointsWithIPs(leaseServers, getEndpointIPs(endpoints)), nil
+	}); err != nil {
+		t.Fatalf("did not find only lease endpoints: %v", err)
+	}
+}
+
+func TestReconcilerMasterLeaseCombined(t *testing.T) {
+	testReconcilersMasterLease(t, 1, 3)
+}
+
+func TestReconcilerMasterLeaseMultiMoreMasters(t *testing.T) {
+	testReconcilersMasterLease(t, 3, 2)
+}
+
+func TestReconcilerMasterLeaseMultiCombined(t *testing.T) {
+	testReconcilersMasterLease(t, 3, 3)
+}

test/integration/master/transformation_testcase.go

@@ -78,7 +78,7 @@ func newTransformTest(l kubeapiservertesting.Logger, transformerConfigYAML strin
 		}
 	}
 
-	if e.kubeAPIServer, err = kubeapiservertesting.StartTestServer(l, e.getEncryptionOptions(), e.storageConfig); err != nil {
+	if e.kubeAPIServer, err = kubeapiservertesting.StartTestServer(l, nil, e.getEncryptionOptions(), e.storageConfig); err != nil {
 		return nil, fmt.Errorf("failed to start KubeAPI server: %v", err)
 	}