mirror of https://github.com/k3s-io/k3s
make CRD apiservice controller
parent
a637c49c8d
commit
446e959bf7
|
@ -93,6 +93,7 @@ go_library(
|
|||
"//vendor/k8s.io/kube-aggregator/pkg/client/clientset_generated/internalclientset/typed/apiregistration/internalversion:go_default_library",
|
||||
"//vendor/k8s.io/kube-aggregator/pkg/controllers/autoregister:go_default_library",
|
||||
"//vendor/k8s.io/kube-apiextensions-server/pkg/apiserver:go_default_library",
|
||||
"//vendor/k8s.io/kube-apiextensions-server/pkg/client/informers/internalversion:go_default_library",
|
||||
"//vendor/k8s.io/kube-apiextensions-server/pkg/cmd/server:go_default_library",
|
||||
],
|
||||
)
|
||||
|
|
|
@ -36,6 +36,7 @@ import (
|
|||
aggregatorapiserver "k8s.io/kube-aggregator/pkg/apiserver"
|
||||
apiregistrationclient "k8s.io/kube-aggregator/pkg/client/clientset_generated/internalclientset/typed/apiregistration/internalversion"
|
||||
"k8s.io/kube-aggregator/pkg/controllers/autoregister"
|
||||
apiextensionsinformers "k8s.io/kube-apiextensions-server/pkg/client/informers/internalversion"
|
||||
"k8s.io/kubernetes/cmd/kube-apiserver/app/options"
|
||||
informers "k8s.io/kubernetes/pkg/client/informers/informers_generated/internalversion"
|
||||
"k8s.io/kubernetes/pkg/master/thirdparty"
|
||||
|
@ -89,7 +90,7 @@ func createAggregatorConfig(kubeAPIServerConfig genericapiserver.Config, command
|
|||
|
||||
}
|
||||
|
||||
func createAggregatorServer(aggregatorConfig *aggregatorapiserver.Config, delegateAPIServer genericapiserver.DelegationTarget, sharedInformers informers.SharedInformerFactory) (*aggregatorapiserver.APIAggregator, error) {
|
||||
func createAggregatorServer(aggregatorConfig *aggregatorapiserver.Config, delegateAPIServer genericapiserver.DelegationTarget, kubeInformers informers.SharedInformerFactory, apiExtensionInformers apiextensionsinformers.SharedInformerFactory) (*aggregatorapiserver.APIAggregator, error) {
|
||||
aggregatorServer, err := aggregatorConfig.Complete().NewWithDelegate(delegateAPIServer)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
@ -102,7 +103,10 @@ func createAggregatorServer(aggregatorConfig *aggregatorapiserver.Config, delega
|
|||
}
|
||||
autoRegistrationController := autoregister.NewAutoRegisterController(aggregatorServer.APIRegistrationInformers.Apiregistration().InternalVersion().APIServices(), apiRegistrationClient)
|
||||
apiServices := apiServicesToRegister(delegateAPIServer, autoRegistrationController)
|
||||
tprRegistrationController := thirdparty.NewAutoRegistrationController(sharedInformers.Extensions().InternalVersion().ThirdPartyResources(), autoRegistrationController)
|
||||
tprRegistrationController := thirdparty.NewAutoRegistrationController(
|
||||
kubeInformers.Extensions().InternalVersion().ThirdPartyResources(),
|
||||
apiExtensionInformers.Apiextensions().InternalVersion().CustomResourceDefinitions(),
|
||||
autoRegistrationController)
|
||||
|
||||
aggregatorServer.GenericAPIServer.AddPostStartHook("kube-apiserver-autoregistration", func(context genericapiserver.PostStartHookContext) error {
|
||||
go autoRegistrationController.Run(5, context.StopCh)
|
||||
|
|
|
@ -111,7 +111,24 @@ func Run(runOptions *options.ServerRunOptions, stopCh <-chan struct{}) error {
|
|||
return err
|
||||
}
|
||||
|
||||
// TPRs are enabled and not yet beta, since this these are the successor, they fall under the same rule
|
||||
// if we're starting up a hacked up version of this API server for a weird test case,
|
||||
// just start the API server as is because clients don't get built correctly when you do this
|
||||
if len(os.Getenv("KUBE_API_VERSIONS")) > 0 {
|
||||
if insecureServingOptions != nil {
|
||||
insecureHandlerChain := kubeserver.BuildInsecureHandlerChain(kubeAPIServer.GenericAPIServer.UnprotectedHandler(), kubeAPIServerConfig.GenericConfig)
|
||||
if err := kubeserver.NonBlockingRun(insecureServingOptions, insecureHandlerChain, stopCh); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return kubeAPIServer.GenericAPIServer.PrepareRun().Run(stopCh)
|
||||
}
|
||||
|
||||
// otherwise go down the normal path of standing the aggregator up in front of the API server
|
||||
// this wires up openapi
|
||||
kubeAPIServer.GenericAPIServer.PrepareRun()
|
||||
|
||||
// TPRs are enabled and not yet beta, since this these are the successor, they fall under the same enablement rule
|
||||
// Subsequent API servers in between here and kube-apiserver will need to be gated.
|
||||
// These come first so that if someone registers both a TPR and a CRD, the CRD is preferred.
|
||||
apiExtensionsConfig, err := createAPIExtensionsConfig(*kubeAPIServerConfig.GenericConfig, runOptions)
|
||||
|
@ -123,34 +140,24 @@ func Run(runOptions *options.ServerRunOptions, stopCh <-chan struct{}) error {
|
|||
return err
|
||||
}
|
||||
|
||||
// run the insecure server now, don't block. It doesn't have any aggregator goodies since authentication wouldn't work
|
||||
if insecureServingOptions != nil {
|
||||
insecureHandlerChain := kubeserver.BuildInsecureHandlerChain(kubeAPIServer.GenericAPIServer.UnprotectedHandler(), kubeAPIServerConfig.GenericConfig)
|
||||
if err := kubeserver.NonBlockingRun(insecureServingOptions, insecureHandlerChain, stopCh); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// if we're starting up a hacked up version of this API server for a weird test case,
|
||||
// just start the API server as is because clients don't get built correctly when you do this
|
||||
if len(os.Getenv("KUBE_API_VERSIONS")) > 0 {
|
||||
return kubeAPIServer.GenericAPIServer.PrepareRun().Run(stopCh)
|
||||
}
|
||||
|
||||
// otherwise go down the normal path of standing the aggregator up in front of the API server
|
||||
// this wires up openapi
|
||||
kubeAPIServer.GenericAPIServer.PrepareRun()
|
||||
|
||||
// aggregator comes last in the chain
|
||||
aggregatorConfig, err := createAggregatorConfig(*kubeAPIServerConfig.GenericConfig, runOptions)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
aggregatorServer, err := createAggregatorServer(aggregatorConfig, apiExtensionsServer.GenericAPIServer, sharedInformers)
|
||||
aggregatorServer, err := createAggregatorServer(aggregatorConfig, apiExtensionsServer.GenericAPIServer, sharedInformers, apiExtensionsServer.Informers)
|
||||
if err != nil {
|
||||
// we don't need special handling for innerStopCh because the aggregator server doesn't create any go routines
|
||||
return err
|
||||
}
|
||||
|
||||
if insecureServingOptions != nil {
|
||||
insecureHandlerChain := kubeserver.BuildInsecureHandlerChain(aggregatorServer.GenericAPIServer.UnprotectedHandler(), kubeAPIServerConfig.GenericConfig)
|
||||
if err := kubeserver.NonBlockingRun(insecureServingOptions, insecureHandlerChain, stopCh); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return aggregatorServer.GenericAPIServer.PrepareRun().Run(stopCh)
|
||||
}
|
||||
|
||||
|
|
|
@ -67,6 +67,7 @@ static="static"
|
|||
storageclass="storageclass"
|
||||
subjectaccessreviews="subjectaccessreviews"
|
||||
thirdpartyresources="thirdpartyresources"
|
||||
customresourcedefinitions="customresourcedefinitions"
|
||||
daemonsets="daemonsets"
|
||||
|
||||
|
||||
|
@ -1286,6 +1287,57 @@ run_kubectl_request_timeout_tests() {
|
|||
kubectl delete pods valid-pod "${kube_flags[@]}"
|
||||
}
|
||||
|
||||
run_crd_tests() {
|
||||
create_and_use_new_namespace
|
||||
kubectl "${kube_flags_with_token[@]}" create -f - << __EOF__
|
||||
{
|
||||
"kind": "CustomResourceDefinition",
|
||||
"apiVersion": "apiextensions.k8s.io/v1alpha1",
|
||||
"metadata": {
|
||||
"name": "foos.company.com"
|
||||
},
|
||||
"spec": {
|
||||
"group": "company.com",
|
||||
"version": "v1",
|
||||
"names": {
|
||||
"plural": "foos",
|
||||
"kind": "Foo"
|
||||
}
|
||||
}
|
||||
}
|
||||
__EOF__
|
||||
|
||||
# Post-Condition: assertion object exist
|
||||
kube::test::get_object_assert customresourcedefinitions "{{range.items}}{{$id_field}}:{{end}}" 'foos.company.com:'
|
||||
|
||||
kubectl "${kube_flags_with_token[@]}" create -f - << __EOF__
|
||||
{
|
||||
"kind": "CustomResourceDefinition",
|
||||
"apiVersion": "apiextensions.k8s.io/v1alpha1",
|
||||
"metadata": {
|
||||
"name": "bars.company.com"
|
||||
},
|
||||
"spec": {
|
||||
"group": "company.com",
|
||||
"version": "v1",
|
||||
"names": {
|
||||
"plural": "bars",
|
||||
"kind": "Bar"
|
||||
}
|
||||
}
|
||||
}
|
||||
__EOF__
|
||||
|
||||
# Post-Condition: assertion object exist
|
||||
kube::test::get_object_assert customresourcedefinitions "{{range.items}}{{$id_field}}:{{end}}" 'bars.company.com:foos.company.com:'
|
||||
|
||||
run_non_native_resource_tests
|
||||
|
||||
# teardown
|
||||
kubectl delete customresourcedefinitions/foos.company.com "${kube_flags_with_token[@]}"
|
||||
kubectl delete customresourcedefinitions/bars.company.com "${kube_flags_with_token[@]}"
|
||||
}
|
||||
|
||||
run_tpr_tests() {
|
||||
create_and_use_new_namespace
|
||||
kubectl "${kube_flags[@]}" create -f - "${kube_flags[@]}" << __EOF__
|
||||
|
@ -1324,11 +1376,39 @@ __EOF__
|
|||
# Post-Condition: assertion object exist
|
||||
kube::test::get_object_assert thirdpartyresources "{{range.items}}{{$id_field}}:{{end}}" 'bar.company.com:foo.company.com:'
|
||||
|
||||
kube::util::wait_for_url "http://127.0.0.1:${API_PORT}/apis/company.com/v1" "third party api"
|
||||
run_non_native_resource_tests
|
||||
|
||||
kube::util::wait_for_url "http://127.0.0.1:${API_PORT}/apis/company.com/v1/foos" "third party api Foo"
|
||||
# teardown
|
||||
kubectl delete thirdpartyresources/foo.company.com "${kube_flags[@]}"
|
||||
kubectl delete thirdpartyresources/bar.company.com "${kube_flags[@]}"
|
||||
}
|
||||
|
||||
kube::util::wait_for_url "http://127.0.0.1:${API_PORT}/apis/company.com/v1/bars" "third party api Bar"
|
||||
|
||||
kube::util::non_native_resources() {
|
||||
local times
|
||||
local wait
|
||||
local failed
|
||||
times=30
|
||||
wait=10
|
||||
local i
|
||||
for i in $(seq 1 $times); do
|
||||
failed=""
|
||||
kubectl "${kube_flags[@]}" get --raw '/apis/company.com/v1' || failed=true
|
||||
kubectl "${kube_flags[@]}" get --raw '/apis/company.com/v1/foos' || failed=true
|
||||
kubectl "${kube_flags[@]}" get --raw '/apis/company.com/v1/bars' || failed=true
|
||||
|
||||
if [ -z "${failed}" ]; then
|
||||
return 0
|
||||
fi
|
||||
sleep ${wait}
|
||||
done
|
||||
|
||||
kube::log::error "Timed out waiting for non-native-resources; tried ${times} waiting ${wait}s between each"
|
||||
return 1
|
||||
}
|
||||
|
||||
run_non_native_resource_tests() {
|
||||
kube::util::non_native_resources
|
||||
|
||||
# Test that we can list this new third party resource (foos)
|
||||
kube::test::get_object_assert foos "{{range.items}}{{$id_field}}:{{end}}" ''
|
||||
|
@ -1363,7 +1443,7 @@ __EOF__
|
|||
kubectl "${kube_flags[@]}" get foos/test -o "jsonpath={.someField}" --allow-missing-template-keys=false
|
||||
kubectl "${kube_flags[@]}" get foos -o "go-template={{range .items}}{{.someField}}{{end}}" --allow-missing-template-keys=false
|
||||
kubectl "${kube_flags[@]}" get foos/test -o "go-template={{.someField}}" --allow-missing-template-keys=false
|
||||
output_message=$(kubectl get foos/test -o name)
|
||||
output_message=$(kubectl "${kube_flags[@]}" get foos/test -o name)
|
||||
kube::test::if_has_string "${output_message}" 'foos/test'
|
||||
|
||||
# Test patching
|
||||
|
@ -1558,10 +1638,6 @@ __EOF__
|
|||
# Make sure it's gone
|
||||
kube::test::get_object_assert foos "{{range.items}}{{$id_field}}:{{end}}" ''
|
||||
kube::test::get_object_assert bars "{{range.items}}{{$id_field}}:{{end}}" ''
|
||||
|
||||
# teardown
|
||||
kubectl delete thirdpartyresources foo.company.com "${kube_flags[@]}"
|
||||
kubectl delete thirdpartyresources bar.company.com "${kube_flags[@]}"
|
||||
}
|
||||
|
||||
run_recursive_resources_tests() {
|
||||
|
@ -3172,6 +3248,11 @@ runTests() {
|
|||
# Third Party Resources #
|
||||
#####################################
|
||||
|
||||
# customresourcedefinitions cleanup after themselves. Run these first, then TPRs
|
||||
if kube::test::if_supports_resource "${customresourcedefinitions}" ; then
|
||||
run_crd_tests
|
||||
fi
|
||||
|
||||
if kube::test::if_supports_resource "${thirdpartyresources}" ; then
|
||||
run_tpr_tests
|
||||
fi
|
||||
|
|
|
@ -43,6 +43,9 @@ go_library(
|
|||
"//vendor/k8s.io/client-go/tools/cache:go_default_library",
|
||||
"//vendor/k8s.io/client-go/util/workqueue:go_default_library",
|
||||
"//vendor/k8s.io/kube-aggregator/pkg/apis/apiregistration:go_default_library",
|
||||
"//vendor/k8s.io/kube-apiextensions-server/pkg/apis/apiextensions:go_default_library",
|
||||
"//vendor/k8s.io/kube-apiextensions-server/pkg/client/informers/internalversion/apiextensions/internalversion:go_default_library",
|
||||
"//vendor/k8s.io/kube-apiextensions-server/pkg/client/listers/apiextensions/internalversion:go_default_library",
|
||||
],
|
||||
)
|
||||
|
||||
|
@ -72,5 +75,7 @@ go_test(
|
|||
"//vendor/k8s.io/client-go/tools/cache:go_default_library",
|
||||
"//vendor/k8s.io/client-go/util/workqueue:go_default_library",
|
||||
"//vendor/k8s.io/kube-aggregator/pkg/apis/apiregistration:go_default_library",
|
||||
"//vendor/k8s.io/kube-apiextensions-server/pkg/apis/apiextensions:go_default_library",
|
||||
"//vendor/k8s.io/kube-apiextensions-server/pkg/client/listers/apiextensions/internalversion:go_default_library",
|
||||
],
|
||||
)
|
||||
|
|
|
@ -30,6 +30,9 @@ import (
|
|||
"k8s.io/client-go/tools/cache"
|
||||
"k8s.io/client-go/util/workqueue"
|
||||
"k8s.io/kube-aggregator/pkg/apis/apiregistration"
|
||||
"k8s.io/kube-apiextensions-server/pkg/apis/apiextensions"
|
||||
crdinformers "k8s.io/kube-apiextensions-server/pkg/client/informers/internalversion/apiextensions/internalversion"
|
||||
crdlisters "k8s.io/kube-apiextensions-server/pkg/client/listers/apiextensions/internalversion"
|
||||
"k8s.io/kubernetes/pkg/apis/extensions"
|
||||
informers "k8s.io/kubernetes/pkg/client/informers/informers_generated/internalversion/extensions/internalversion"
|
||||
listers "k8s.io/kubernetes/pkg/client/listers/extensions/internalversion"
|
||||
|
@ -49,6 +52,8 @@ type AutoAPIServiceRegistration interface {
|
|||
type tprRegistrationController struct {
|
||||
tprLister listers.ThirdPartyResourceLister
|
||||
tprSynced cache.InformerSynced
|
||||
crdLister crdlisters.CustomResourceDefinitionLister
|
||||
crdSynced cache.InformerSynced
|
||||
|
||||
apiServiceRegistration AutoAPIServiceRegistration
|
||||
|
||||
|
@ -61,14 +66,18 @@ type tprRegistrationController struct {
|
|||
|
||||
// NewAutoRegistrationController returns a controller which will register TPR GroupVersions with the auto APIService registration
|
||||
// controller so they automatically stay in sync.
|
||||
func NewAutoRegistrationController(tprInformer informers.ThirdPartyResourceInformer, apiServiceRegistration AutoAPIServiceRegistration) *tprRegistrationController {
|
||||
// In order to stay sane with both TPR and CRD present, we have a single controller that manages both. When choosing whether to have an
|
||||
// APIService, we simply iterate through both.
|
||||
func NewAutoRegistrationController(tprInformer informers.ThirdPartyResourceInformer, crdinformer crdinformers.CustomResourceDefinitionInformer, apiServiceRegistration AutoAPIServiceRegistration) *tprRegistrationController {
|
||||
c := &tprRegistrationController{
|
||||
tprLister: tprInformer.Lister(),
|
||||
tprSynced: tprInformer.Informer().HasSynced,
|
||||
crdLister: crdinformer.Lister(),
|
||||
crdSynced: crdinformer.Informer().HasSynced,
|
||||
apiServiceRegistration: apiServiceRegistration,
|
||||
queue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "tpr-autoregister"),
|
||||
}
|
||||
c.syncHandler = c.handleTPR
|
||||
c.syncHandler = c.handleVersionUpdate
|
||||
|
||||
tprInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
|
||||
AddFunc: func(obj interface{}) {
|
||||
|
@ -97,6 +106,33 @@ func NewAutoRegistrationController(tprInformer informers.ThirdPartyResourceInfor
|
|||
},
|
||||
})
|
||||
|
||||
crdinformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
|
||||
AddFunc: func(obj interface{}) {
|
||||
cast := obj.(*apiextensions.CustomResourceDefinition)
|
||||
c.enqueueCRD(cast)
|
||||
},
|
||||
UpdateFunc: func(_, obj interface{}) {
|
||||
cast := obj.(*apiextensions.CustomResourceDefinition)
|
||||
c.enqueueCRD(cast)
|
||||
},
|
||||
DeleteFunc: func(obj interface{}) {
|
||||
cast, ok := obj.(*apiextensions.CustomResourceDefinition)
|
||||
if !ok {
|
||||
tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
|
||||
if !ok {
|
||||
glog.V(2).Infof("Couldn't get object from tombstone %#v", obj)
|
||||
return
|
||||
}
|
||||
cast, ok = tombstone.Obj.(*apiextensions.CustomResourceDefinition)
|
||||
if !ok {
|
||||
glog.V(2).Infof("Tombstone contained unexpected object: %#v", obj)
|
||||
return
|
||||
}
|
||||
}
|
||||
c.enqueueCRD(cast)
|
||||
},
|
||||
})
|
||||
|
||||
return c
|
||||
}
|
||||
|
||||
|
@ -173,14 +209,19 @@ func (c *tprRegistrationController) enqueueTPR(tpr *extensions.ThirdPartyResourc
|
|||
}
|
||||
}
|
||||
|
||||
func (c *tprRegistrationController) handleTPR(groupVersion schema.GroupVersion) error {
|
||||
func (c *tprRegistrationController) enqueueCRD(crd *apiextensions.CustomResourceDefinition) {
|
||||
c.queue.Add(schema.GroupVersion{Group: crd.Spec.Group, Version: crd.Spec.Version})
|
||||
}
|
||||
|
||||
func (c *tprRegistrationController) handleVersionUpdate(groupVersion schema.GroupVersion) error {
|
||||
found := false
|
||||
apiServiceName := groupVersion.Version + "." + groupVersion.Group
|
||||
|
||||
// check all TPRs. There shouldn't that many, but if we have problems later we can index them
|
||||
tprs, err := c.tprLister.List(labels.Everything())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
found := false
|
||||
for _, tpr := range tprs {
|
||||
_, group, err := thirdpartyresourcedata.ExtractApiGroupAndKind(tpr)
|
||||
if err != nil {
|
||||
|
@ -194,7 +235,17 @@ func (c *tprRegistrationController) handleTPR(groupVersion schema.GroupVersion)
|
|||
}
|
||||
}
|
||||
|
||||
apiServiceName := groupVersion.Version + "." + groupVersion.Group
|
||||
// check all CRDs. There shouldn't that many, but if we have problems later we can index them
|
||||
crds, err := c.crdLister.List(labels.Everything())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, crd := range crds {
|
||||
if crd.Spec.Version == groupVersion.Version && crd.Spec.Group == groupVersion.Group {
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
if !found {
|
||||
c.apiServiceRegistration.RemoveAPIServiceToSync(apiServiceName)
|
||||
|
|
|
@ -25,6 +25,8 @@ import (
|
|||
"k8s.io/client-go/tools/cache"
|
||||
"k8s.io/client-go/util/workqueue"
|
||||
"k8s.io/kube-aggregator/pkg/apis/apiregistration"
|
||||
"k8s.io/kube-apiextensions-server/pkg/apis/apiextensions"
|
||||
crdlisters "k8s.io/kube-apiextensions-server/pkg/client/listers/apiextensions/internalversion"
|
||||
"k8s.io/kubernetes/pkg/apis/extensions"
|
||||
listers "k8s.io/kubernetes/pkg/client/listers/extensions/internalversion"
|
||||
)
|
||||
|
@ -56,17 +58,18 @@ func TestEnqueue(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func TestHandleTPR(t *testing.T) {
|
||||
func TestHandleVersionUpdate(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
startingTPRs []*extensions.ThirdPartyResource
|
||||
startingCRDs []*apiextensions.CustomResourceDefinition
|
||||
version schema.GroupVersion
|
||||
|
||||
expectedAdded []*apiregistration.APIService
|
||||
expectedRemoved []string
|
||||
}{
|
||||
{
|
||||
name: "simple add",
|
||||
name: "simple add tpr",
|
||||
startingTPRs: []*extensions.ThirdPartyResource{
|
||||
{
|
||||
ObjectMeta: metav1.ObjectMeta{Name: "resource.group.com"},
|
||||
|
@ -89,7 +92,7 @@ func TestHandleTPR(t *testing.T) {
|
|||
},
|
||||
},
|
||||
{
|
||||
name: "simple remove",
|
||||
name: "simple remove tpr",
|
||||
startingTPRs: []*extensions.ThirdPartyResource{
|
||||
{
|
||||
ObjectMeta: metav1.ObjectMeta{Name: "resource.group.com"},
|
||||
|
@ -100,6 +103,43 @@ func TestHandleTPR(t *testing.T) {
|
|||
},
|
||||
version: schema.GroupVersion{Group: "group.com", Version: "v2"},
|
||||
|
||||
expectedRemoved: []string{"v2.group.com"},
|
||||
},
|
||||
{
|
||||
name: "simple add crd",
|
||||
startingCRDs: []*apiextensions.CustomResourceDefinition{
|
||||
{
|
||||
Spec: apiextensions.CustomResourceDefinitionSpec{
|
||||
Group: "group.com",
|
||||
Version: "v1",
|
||||
},
|
||||
},
|
||||
},
|
||||
version: schema.GroupVersion{Group: "group.com", Version: "v1"},
|
||||
|
||||
expectedAdded: []*apiregistration.APIService{
|
||||
{
|
||||
ObjectMeta: metav1.ObjectMeta{Name: "v1.group.com"},
|
||||
Spec: apiregistration.APIServiceSpec{
|
||||
Group: "group.com",
|
||||
Version: "v1",
|
||||
Priority: 500,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "simple remove crd",
|
||||
startingCRDs: []*apiextensions.CustomResourceDefinition{
|
||||
{
|
||||
Spec: apiextensions.CustomResourceDefinitionSpec{
|
||||
Group: "group.com",
|
||||
Version: "v1",
|
||||
},
|
||||
},
|
||||
},
|
||||
version: schema.GroupVersion{Group: "group.com", Version: "v2"},
|
||||
|
||||
expectedRemoved: []string{"v2.group.com"},
|
||||
},
|
||||
}
|
||||
|
@ -108,15 +148,21 @@ func TestHandleTPR(t *testing.T) {
|
|||
registration := &fakeAPIServiceRegistration{}
|
||||
tprCache := cache.NewIndexer(cache.DeletionHandlingMetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc})
|
||||
tprLister := listers.NewThirdPartyResourceLister(tprCache)
|
||||
crdCache := cache.NewIndexer(cache.DeletionHandlingMetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc})
|
||||
crdLister := crdlisters.NewCustomResourceDefinitionLister(crdCache)
|
||||
c := tprRegistrationController{
|
||||
tprLister: tprLister,
|
||||
crdLister: crdLister,
|
||||
apiServiceRegistration: registration,
|
||||
}
|
||||
for i := range test.startingTPRs {
|
||||
tprCache.Add(test.startingTPRs[i])
|
||||
}
|
||||
for i := range test.startingCRDs {
|
||||
crdCache.Add(test.startingCRDs[i])
|
||||
}
|
||||
|
||||
c.handleTPR(test.version)
|
||||
c.handleVersionUpdate(test.version)
|
||||
|
||||
if !reflect.DeepEqual(test.expectedAdded, registration.added) {
|
||||
t.Errorf("%s expected %v, got %v", test.name, test.expectedAdded, registration.added)
|
||||
|
|
|
@ -79,6 +79,9 @@ type Config struct {
|
|||
|
||||
type CustomResourceDefinitions struct {
|
||||
GenericAPIServer *genericapiserver.GenericAPIServer
|
||||
|
||||
// provided for easier embedding
|
||||
Informers internalinformers.SharedInformerFactory
|
||||
}
|
||||
|
||||
type completedConfig struct {
|
||||
|
@ -126,11 +129,11 @@ func (c completedConfig) New(delegationTarget genericapiserver.DelegationTarget)
|
|||
return nil, err
|
||||
}
|
||||
|
||||
customResourceDefinitionClient, err := internalclientset.NewForConfig(s.GenericAPIServer.LoopbackClientConfig)
|
||||
crdClient, err := internalclientset.NewForConfig(s.GenericAPIServer.LoopbackClientConfig)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
customResourceDefinitionInformers := internalinformers.NewSharedInformerFactory(customResourceDefinitionClient, 5*time.Minute)
|
||||
s.Informers = internalinformers.NewSharedInformerFactory(crdClient, 5*time.Minute)
|
||||
|
||||
delegateHandler := delegationTarget.UnprotectedHandler()
|
||||
if delegateHandler == nil {
|
||||
|
@ -145,31 +148,31 @@ func (c completedConfig) New(delegationTarget genericapiserver.DelegationTarget)
|
|||
discovery: map[string]*discovery.APIGroupHandler{},
|
||||
delegate: delegateHandler,
|
||||
}
|
||||
customResourceDefinitionHandler := NewCustomResourceDefinitionHandler(
|
||||
crdHandler := NewCustomResourceDefinitionHandler(
|
||||
versionDiscoveryHandler,
|
||||
groupDiscoveryHandler,
|
||||
s.GenericAPIServer.RequestContextMapper(),
|
||||
customResourceDefinitionInformers.Apiextensions().InternalVersion().CustomResourceDefinitions().Lister(),
|
||||
s.Informers.Apiextensions().InternalVersion().CustomResourceDefinitions().Lister(),
|
||||
delegateHandler,
|
||||
c.CRDRESTOptionsGetter,
|
||||
c.GenericConfig.AdmissionControl,
|
||||
)
|
||||
s.GenericAPIServer.Handler.PostGoRestfulMux.Handle("/apis", customResourceDefinitionHandler)
|
||||
s.GenericAPIServer.Handler.PostGoRestfulMux.HandlePrefix("/apis/", customResourceDefinitionHandler)
|
||||
s.GenericAPIServer.Handler.PostGoRestfulMux.Handle("/apis", crdHandler)
|
||||
s.GenericAPIServer.Handler.PostGoRestfulMux.HandlePrefix("/apis/", crdHandler)
|
||||
|
||||
customResourceDefinitionController := NewDiscoveryController(customResourceDefinitionInformers.Apiextensions().InternalVersion().CustomResourceDefinitions(), versionDiscoveryHandler, groupDiscoveryHandler)
|
||||
namingController := status.NewNamingConditionController(customResourceDefinitionInformers.Apiextensions().InternalVersion().CustomResourceDefinitions(), customResourceDefinitionClient)
|
||||
crdController := NewDiscoveryController(s.Informers.Apiextensions().InternalVersion().CustomResourceDefinitions(), versionDiscoveryHandler, groupDiscoveryHandler)
|
||||
namingController := status.NewNamingConditionController(s.Informers.Apiextensions().InternalVersion().CustomResourceDefinitions(), crdClient)
|
||||
finalizingController := finalizer.NewCRDFinalizer(
|
||||
customResourceDefinitionInformers.Apiextensions().InternalVersion().CustomResourceDefinitions(),
|
||||
customResourceDefinitionClient,
|
||||
s.Informers.Apiextensions().InternalVersion().CustomResourceDefinitions(),
|
||||
crdClient,
|
||||
dynamic.NewDynamicClientPool(s.GenericAPIServer.LoopbackClientConfig))
|
||||
|
||||
s.GenericAPIServer.AddPostStartHook("start-apiextensions-informers", func(context genericapiserver.PostStartHookContext) error {
|
||||
customResourceDefinitionInformers.Start(context.StopCh)
|
||||
s.Informers.Start(context.StopCh)
|
||||
return nil
|
||||
})
|
||||
s.GenericAPIServer.AddPostStartHook("start-apiextensions-controllers", func(context genericapiserver.PostStartHookContext) error {
|
||||
go customResourceDefinitionController.Run(context.StopCh)
|
||||
go crdController.Run(context.StopCh)
|
||||
go namingController.Run(context.StopCh)
|
||||
go finalizingController.Run(5, context.StopCh)
|
||||
return nil
|
||||
|
|
|
@ -127,10 +127,6 @@ func (r *crdHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
|
|||
r.delegate.ServeHTTP(w, req)
|
||||
return
|
||||
}
|
||||
if len(requestInfo.Subresource) > 0 {
|
||||
http.NotFound(w, req)
|
||||
return
|
||||
}
|
||||
|
||||
crdName := requestInfo.Resource + "." + requestInfo.APIGroup
|
||||
crd, err := r.crdLister.Get(crdName)
|
||||
|
@ -150,6 +146,10 @@ func (r *crdHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
|
|||
if !apiextensions.IsCRDConditionFalse(crd, apiextensions.NameConflict) {
|
||||
r.delegate.ServeHTTP(w, req)
|
||||
}
|
||||
if len(requestInfo.Subresource) > 0 {
|
||||
http.NotFound(w, req)
|
||||
return
|
||||
}
|
||||
|
||||
terminating := apiextensions.IsCRDConditionTrue(crd, apiextensions.Terminating)
|
||||
|
||||
|
|
|
@ -34,6 +34,7 @@ go_test(
|
|||
"//vendor/k8s.io/apimachinery/pkg/util/diff:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
|
||||
"//vendor/k8s.io/apiserver/pkg/server:go_default_library",
|
||||
"//vendor/k8s.io/apiserver/pkg/storage/storagebackend:go_default_library",
|
||||
"//vendor/k8s.io/client-go/kubernetes:go_default_library",
|
||||
"//vendor/k8s.io/client-go/rest:go_default_library",
|
||||
|
|
|
@ -39,6 +39,7 @@ import (
|
|||
"k8s.io/apimachinery/pkg/util/diff"
|
||||
"k8s.io/apimachinery/pkg/util/sets"
|
||||
"k8s.io/apimachinery/pkg/util/wait"
|
||||
genericapiserver "k8s.io/apiserver/pkg/server"
|
||||
"k8s.io/apiserver/pkg/storage/storagebackend"
|
||||
kclient "k8s.io/client-go/kubernetes"
|
||||
restclient "k8s.io/client-go/rest"
|
||||
|
@ -559,7 +560,7 @@ func startRealMasterOrDie(t *testing.T, certDir string) (*allClient, clientv3.KV
|
|||
|
||||
kubeAPIServerConfig.APIResourceConfigSource = &allResourceSource{} // force enable all resources
|
||||
|
||||
kubeAPIServer, err := app.CreateKubeAPIServer(kubeAPIServerConfig, sharedInformers)
|
||||
kubeAPIServer, err := app.CreateKubeAPIServer(kubeAPIServerConfig, genericapiserver.EmptyDelegate, sharedInformers)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
|
|
@ -21,6 +21,7 @@ go_test(
|
|||
"//vendor/github.com/stretchr/testify/assert:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
|
||||
"//vendor/k8s.io/apiserver/pkg/server:go_default_library",
|
||||
"//vendor/k8s.io/client-go/kubernetes:go_default_library",
|
||||
"//vendor/k8s.io/client-go/rest:go_default_library",
|
||||
"//vendor/k8s.io/client-go/tools/clientcmd:go_default_library",
|
||||
|
|
|
@ -33,6 +33,7 @@ import (
|
|||
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/util/wait"
|
||||
genericapiserver "k8s.io/apiserver/pkg/server"
|
||||
client "k8s.io/client-go/kubernetes"
|
||||
"k8s.io/client-go/rest"
|
||||
"k8s.io/client-go/tools/clientcmd"
|
||||
|
@ -117,7 +118,7 @@ func TestAggregatedAPIServer(t *testing.T) {
|
|||
}
|
||||
kubeClientConfigValue.Store(kubeAPIServerConfig.GenericConfig.LoopbackClientConfig)
|
||||
|
||||
kubeAPIServer, err := app.CreateKubeAPIServer(kubeAPIServerConfig, sharedInformers)
|
||||
kubeAPIServer, err := app.CreateKubeAPIServer(kubeAPIServerConfig, genericapiserver.EmptyDelegate, sharedInformers)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
|
Loading…
Reference in New Issue