Api server data partition reset on the fly (CentaurusInfra#123)

* Api server data partition reset on the fly.
    . Add data partition configuration and serviceGroups for api server
    . Reset etcd data watch based on partition changes in api server
    . Fix unit tests and integration tests

* Api server data partition reset on the fly - auto generated code

* Bring back one commented out integration test. Fix copyright header.

* Add integration test for api server data partition reset

.gitignore

@@ -121,6 +121,7 @@ zz_generated_*_test.go
 
 # Just in time generated data in the source, should never be committed
 /test/e2e/generated/bindata.go
+/cmd/kube-apiserver/app/testing/testdata/*
 
 # This file used by some vendor repos (e.g. github.com/go-openapi/...) to store secret variables and should not be ignored
 !\.drone\.sec

cmd/kube-apiserver/app/BUILD

@@ -54,6 +54,7 @@ go_library(
         "//staging/src/k8s.io/apiserver/pkg/server/healthz:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/server/options:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/server/storage:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/storage/datapartition:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/storage/etcd3/preflight:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/util/term:go_default_library",

cmd/kube-apiserver/app/aggregator.go

@@ -1,5 +1,6 @@
 /*
 Copyright 2017 The Kubernetes Authors.
+Copyright 2020 Authors of Arktos - file modified.
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.

cmd/kube-apiserver/app/options/options.go

@@ -73,6 +73,8 @@ type ServerRunOptions struct {
 	ServiceAccountSigningKeyFile     string
 	ServiceAccountIssuer             serviceaccount.TokenGenerator
 	ServiceAccountTokenMaxExpiration time.Duration
+
+	ServiceGroupId string
 }
 
 // NewServerRunOptions creates a new ServerRunOptions object with default parameters

cmd/kube-apiserver/app/server.go

@@ -24,6 +24,7 @@ import (
 	"crypto/tls"
 	"fmt"
 	"io/ioutil"
+	"k8s.io/apiserver/pkg/storage/datapartition"
 	"net"
 	"net/http"
 	"net/url"
@@ -178,7 +179,7 @@ func CreateServerChain(completedOptions completedServerRunOptions, stopCh <-chan
 		return nil, err
 	}
 
-	kubeAPIServer, err := CreateKubeAPIServer(kubeAPIServerConfig, apiExtensionsServer.GenericAPIServer, admissionPostStartHook)
+	kubeAPIServer, err := CreateKubeAPIServer(kubeAPIServerConfig, apiExtensionsServer.GenericAPIServer, admissionPostStartHook, stopCh)
 	if err != nil {
 		return nil, err
 	}
@@ -212,13 +213,14 @@ func CreateServerChain(completedOptions completedServerRunOptions, stopCh <-chan
 }
 
 // CreateKubeAPIServer creates and wires a workable kube-apiserver
-func CreateKubeAPIServer(kubeAPIServerConfig *master.Config, delegateAPIServer genericapiserver.DelegationTarget, admissionPostStartHook genericapiserver.PostStartHookFunc) (*master.Master, error) {
+func CreateKubeAPIServer(kubeAPIServerConfig *master.Config, delegateAPIServer genericapiserver.DelegationTarget, admissionPostStartHook genericapiserver.PostStartHookFunc, stopCh <-chan struct{}) (*master.Master, error) {
 	kubeAPIServer, err := kubeAPIServerConfig.Complete().New(delegateAPIServer)
 	if err != nil {
 		return nil, err
 	}
 
 	kubeAPIServer.GenericAPIServer.AddPostStartHookOrDie("start-kube-apiserver-admission-initializer", admissionPostStartHook)
+	go kubeAPIServerConfig.ExtraConfig.DataPartitionManager.Run(stopCh)
 
 	return kubeAPIServer, nil
 }
@@ -351,6 +353,7 @@ func CreateKubeAPIServerConfig(
 
 			EndpointReconcilerType: reconcilers.Type(s.EndpointReconcilerType),
 			MasterCount:            s.MasterCount,
+			ServiceGroupId:         s.ServiceGroupId,
 
 			ServiceAccountIssuer:        s.ServiceAccountIssuer,
 			ServiceAccountMaxExpiration: s.ServiceAccountTokenMaxExpiration,
@@ -359,6 +362,12 @@ func CreateKubeAPIServerConfig(
 		},
 	}
 
+	config.ExtraConfig.DataPartitionManager = datapartition.GetDataPartitionConfigManager()
+	if config.ExtraConfig.DataPartitionManager == nil {
+		config.ExtraConfig.DataPartitionManager = datapartition.NewDataPartitionConfigManager(
+			config.ExtraConfig.ServiceGroupId, config.ExtraConfig.VersionedInformers.Core().V1().DataPartitionConfigs())
+	}
+
 	if nodeTunneler != nil {
 		// Use the nodeTunneler's dialer to connect to the kubelet
 		config.ExtraConfig.KubeletClientConfig.Dial = nodeTunneler.Dial
@@ -629,6 +638,10 @@ func Complete(s *options.ServerRunOptions) (completedServerRunOptions, error) {
 			}
 		}
 	}
+
+	// service group - for data partition
+	// TODO - read service group id from commandline config
+	s.ServiceGroupId = "0"
 	options.ServerRunOptions = s
 	return options, nil
 }

pkg/apis/core/register.go

@@ -100,6 +100,8 @@ func addKnownTypes(scheme *runtime.Scheme) error {
 		&CustomAction{},
 		&Action{},
 		&ActionList{},
+		&DataPartitionConfig{},
+		&DataPartitionConfigList{},
 	)
 
 	return nil

pkg/apis/core/types.go

@@ -5246,3 +5246,42 @@ type ControllerInstanceList struct {
 	// List of controller instance
 	Items []ControllerInstance
 }
+
+// +genclient
+// +genclient:nonNamespaced
+// +genclient:nonTenanted
+
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+
+// DataPartitionConfig contains data partition configuration instance with name
+type DataPartitionConfig struct {
+	metav1.TypeMeta
+	metav1.ObjectMeta
+
+	// Start tenant is inclusive
+	StartTenant string
+
+	// Whether this is an open end start
+	IsStartTenantValid bool
+
+	// End tenant is exclusive
+	EndTenant string
+
+	// Whether this is an open end end
+	IsEndTenantValid bool
+
+	// Which service group is using this data configuration
+	ServiceGroupId string
+}
+
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+
+// DataPartitionConfigList holds the data partition configuration list
+type DataPartitionConfigList struct {
+	metav1.TypeMeta
+	// +optional
+	metav1.ListMeta
+
+	// List of data partition configuration
+	Items []DataPartitionConfig
+}

pkg/apis/core/validation/validation.go

@@ -206,6 +206,9 @@ var ValidateReplicationControllerName = apimachineryvalidation.NameIsDNSSubdomai
 // trailing dashes are allowed.
 var ValidateServiceName = apimachineryvalidation.NameIsDNS1035Label
 
+// ValidateDataPartitionName can be used to check whether given data partition name is valid.
+var ValidateDataPartitionName = apimachineryvalidation.NameIsDNSSubdomain
+
 // ValidateNodeName can be used to check whether the given node name is valid.
 // Prefix indicates this name will be used as part of generation, in which case
 // trailing dashes are allowed.
@@ -4166,6 +4169,21 @@ func ValidateServiceStatusUpdate(service, oldService *core.Service) field.ErrorL
 	return allErrs
 }
 
+func ValidateDataPartitionConfig(dataPartitionConfig *core.DataPartitionConfig) field.ErrorList {
+	allErrs := ValidateObjectMeta(&dataPartitionConfig.ObjectMeta, false, false, ValidateDataPartitionName, field.NewPath("metadata"))
+	return allErrs
+}
+
+func ValidateDataPartitionConfigUpdate(newDataPartitionConfig *core.DataPartitionConfig, oldDataPartitionConfig *core.DataPartitionConfig) field.ErrorList {
+	allErrs := ValidateObjectMeta(&newDataPartitionConfig.ObjectMeta, false, false, ValidateDataPartitionName, field.NewPath("metadata"))
+	if newDataPartitionConfig.Name != oldDataPartitionConfig.Name {
+		klog.Infof("Intended to update data partition config name. Not allowed. new name [%v], old name [%v]", newDataPartitionConfig.Name, oldDataPartitionConfig.Name)
+		nameDiff := diff.ObjectDiff(newDataPartitionConfig.Name, oldDataPartitionConfig.Name)
+		allErrs = append(allErrs, field.Forbidden(field.NewPath("Name"), fmt.Sprintf("Update data configuration name is not allowed. %v", nameDiff)))
+	}
+	return allErrs
+}
+
 // ValidateReplicationController tests if required fields in the replication controller are set.
 func ValidateReplicationController(controller *core.ReplicationController) field.ErrorList {
 	allErrs := ValidateObjectMeta(&controller.ObjectMeta, true, true, ValidateReplicationControllerName, field.NewPath("metadata"))

pkg/cloudfabric-controller/controllerframework/controller_framework.go

@@ -184,7 +184,7 @@ func (c *ControllerBase) WatchInstanceUpdate(stopCh <-chan struct{}) {
 			break
 		case updatedType, ok := <-c.controllerInstanceUpdateCh.Read:
 			if !ok {
-				klog.Errorf("Unexpected controller instance update message")
+				klog.Error("Unexpected controller instance update message")
 				return
 			}
 			klog.Infof("Got controller instance update massage. Updated Controller Type %s, current controller instance type %s, key %d", updatedType, c.controllerType, c.controllerKey)