mirror of
https://github.com/k3s-io/kubernetes.git
synced 2025-07-30 23:15:14 +00:00
fed: Move namespace propagation to the sync controller
This commit is contained in:
Maru Newby
parent
7e772ee3eb
commit
228ab0d882
@ -24,7 +24,6 @@ go_library(
|
||||
"//federation/pkg/federatedtypes:go_default_library",
|
||||
"//federation/pkg/federation-controller/cluster:go_default_library",
|
||||
"//federation/pkg/federation-controller/ingress:go_default_library",
|
||||
"//federation/pkg/federation-controller/namespace:go_default_library",
|
||||
"//federation/pkg/federation-controller/service:go_default_library",
|
||||
"//federation/pkg/federation-controller/service/dns:go_default_library",
|
||||
"//federation/pkg/federation-controller/sync:go_default_library",
|
||||
@ -40,7 +39,6 @@ go_library(
|
||||
"//vendor/k8s.io/apiserver/pkg/server/healthz:go_default_library",
|
||||
"//vendor/k8s.io/apiserver/pkg/util/flag:go_default_library",
|
||||
"//vendor/k8s.io/client-go/discovery:go_default_library",
|
||||
"//vendor/k8s.io/client-go/dynamic:go_default_library",
|
||||
"//vendor/k8s.io/client-go/rest:go_default_library",
|
||||
"//vendor/k8s.io/client-go/tools/clientcmd:go_default_library",
|
||||
],
|
||||
|
||||
@ -30,7 +30,6 @@ import (
|
||||
"k8s.io/apimachinery/pkg/util/wait"
|
||||
"k8s.io/apiserver/pkg/server/healthz"
|
||||
utilflag "k8s.io/apiserver/pkg/util/flag"
|
||||
"k8s.io/client-go/dynamic"
|
||||
restclient "k8s.io/client-go/rest"
|
||||
"k8s.io/client-go/tools/clientcmd"
|
||||
federationclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_clientset"
|
||||
@ -38,7 +37,6 @@ import (
|
||||
"k8s.io/kubernetes/federation/pkg/federatedtypes"
|
||||
clustercontroller "k8s.io/kubernetes/federation/pkg/federation-controller/cluster"
|
||||
ingresscontroller "k8s.io/kubernetes/federation/pkg/federation-controller/ingress"
|
||||
namespacecontroller "k8s.io/kubernetes/federation/pkg/federation-controller/namespace"
|
||||
servicecontroller "k8s.io/kubernetes/federation/pkg/federation-controller/service"
|
||||
servicednscontroller "k8s.io/kubernetes/federation/pkg/federation-controller/service/dns"
|
||||
synccontroller "k8s.io/kubernetes/federation/pkg/federation-controller/sync"
|
||||
@ -151,14 +149,6 @@ func StartControllers(s *options.CMServer, restClientCfg *restclient.Config) err
|
||||
go serviceController.Run(s.ConcurrentServiceSyncs, wait.NeverStop)
|
||||
}
|
||||
|
||||
if controllerEnabled(s.Controllers, serverResources, namespacecontroller.ControllerName, namespacecontroller.RequiredResources, true) {
|
||||
glog.V(3).Infof("Loading client config for namespace controller %q", namespacecontroller.UserAgentName)
|
||||
nsClientset := federationclientset.NewForConfigOrDie(restclient.AddUserAgent(restClientCfg, namespacecontroller.UserAgentName))
|
||||
namespaceController := namespacecontroller.NewNamespaceController(nsClientset, dynamic.NewDynamicClientPool(restclient.AddUserAgent(restClientCfg, namespacecontroller.UserAgentName)))
|
||||
glog.V(3).Infof("Running namespace controller")
|
||||
namespaceController.Run(wait.NeverStop)
|
||||
}
|
||||
|
||||
for kind, federatedType := range federatedtypes.FederatedTypes() {
|
||||
if controllerEnabled(s.Controllers, serverResources, federatedType.ControllerName, federatedType.RequiredResources, true) {
|
||||
synccontroller.StartFederationSyncController(kind, federatedType.AdapterFactory, restClientCfg, stopChan, minimizeLatency)
|
||||
|
||||
@ -15,6 +15,7 @@ go_library(
|
||||
"configmap.go",
|
||||
"daemonset.go",
|
||||
"deployment.go",
|
||||
"namespace.go",
|
||||
"qualifiedname.go",
|
||||
"registry.go",
|
||||
"replicaset.go",
|
||||
@ -30,7 +31,9 @@ go_library(
|
||||
"//federation/pkg/federation-controller/util/planner:go_default_library",
|
||||
"//federation/pkg/federation-controller/util/podanalyzer:go_default_library",
|
||||
"//federation/pkg/federation-controller/util/replicapreferences:go_default_library",
|
||||
"//pkg/api:go_default_library",
|
||||
"//pkg/client/clientset_generated/clientset:go_default_library",
|
||||
"//pkg/controller/namespace/deletion:go_default_library",
|
||||
"//vendor/github.com/golang/glog:go_default_library",
|
||||
"//vendor/k8s.io/api/core/v1:go_default_library",
|
||||
"//vendor/k8s.io/api/extensions/v1beta1:go_default_library",
|
||||
@ -39,7 +42,9 @@ go_library(
|
||||
"//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/watch:go_default_library",
|
||||
"//vendor/k8s.io/client-go/dynamic:go_default_library",
|
||||
"//vendor/k8s.io/client-go/rest:go_default_library",
|
||||
"//vendor/k8s.io/client-go/tools/record:go_default_library",
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
215
federation/pkg/federatedtypes/namespace.go
Normal file
215
federation/pkg/federatedtypes/namespace.go
Normal file
@ -0,0 +1,215 @@
|
||||
/*
|
||||
Copyright 2017 The Kubernetes Authors.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
package federatedtypes
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
apiv1 "k8s.io/api/core/v1"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
pkgruntime "k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/apimachinery/pkg/watch"
|
||||
"k8s.io/client-go/dynamic"
|
||||
restclient "k8s.io/client-go/rest"
|
||||
"k8s.io/client-go/tools/record"
|
||||
federationclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_clientset"
|
||||
"k8s.io/kubernetes/federation/pkg/federation-controller/util"
|
||||
"k8s.io/kubernetes/pkg/api"
|
||||
kubeclientset "k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
|
||||
"k8s.io/kubernetes/pkg/controller/namespace/deletion"
|
||||
|
||||
"github.com/golang/glog"
|
||||
)
|
||||
|
||||
const (
|
||||
NamespaceKind = "namespace"
|
||||
NamespaceControllerName = "namespaces"
|
||||
)
|
||||
|
||||
func init() {
|
||||
RegisterFederatedType(NamespaceKind, NamespaceControllerName, []schema.GroupVersionResource{apiv1.SchemeGroupVersion.WithResource(NamespaceControllerName)}, NewNamespaceAdapter)
|
||||
}
|
||||
|
||||
type NamespaceAdapter struct {
|
||||
client federationclientset.Interface
|
||||
deleter deletion.NamespacedResourcesDeleterInterface
|
||||
}
|
||||
|
||||
func NewNamespaceAdapter(client federationclientset.Interface, config *restclient.Config) FederatedTypeAdapter {
|
||||
dynamicClientPool := dynamic.NewDynamicClientPool(config)
|
||||
discoverResourcesFunc := client.Discovery().ServerPreferredNamespacedResources
|
||||
deleter := deletion.NewNamespacedResourcesDeleter(
|
||||
client.Core().Namespaces(),
|
||||
dynamicClientPool,
|
||||
nil,
|
||||
discoverResourcesFunc,
|
||||
apiv1.FinalizerKubernetes,
|
||||
false)
|
||||
return &NamespaceAdapter{client: client, deleter: deleter}
|
||||
}
|
||||
|
||||
func (a *NamespaceAdapter) Kind() string {
|
||||
return NamespaceKind
|
||||
}
|
||||
|
||||
func (a *NamespaceAdapter) ObjectType() pkgruntime.Object {
|
||||
return &apiv1.Namespace{}
|
||||
}
|
||||
|
||||
func (a *NamespaceAdapter) IsExpectedType(obj interface{}) bool {
|
||||
_, ok := obj.(*apiv1.Namespace)
|
||||
return ok
|
||||
}
|
||||
|
||||
func (a *NamespaceAdapter) Copy(obj pkgruntime.Object) pkgruntime.Object {
|
||||
namespace := obj.(*apiv1.Namespace)
|
||||
return &apiv1.Namespace{
|
||||
ObjectMeta: util.DeepCopyRelevantObjectMeta(namespace.ObjectMeta),
|
||||
Spec: *(util.DeepCopyApiTypeOrPanic(&namespace.Spec).(*apiv1.NamespaceSpec)),
|
||||
}
|
||||
}
|
||||
|
||||
func (a *NamespaceAdapter) Equivalent(obj1, obj2 pkgruntime.Object) bool {
|
||||
return util.ObjectMetaAndSpecEquivalent(obj1, obj2)
|
||||
}
|
||||
|
||||
func (a *NamespaceAdapter) QualifiedName(obj pkgruntime.Object) QualifiedName {
|
||||
namespace := obj.(*apiv1.Namespace)
|
||||
return QualifiedName{Name: namespace.Name}
|
||||
}
|
||||
|
||||
func (a *NamespaceAdapter) ObjectMeta(obj pkgruntime.Object) *metav1.ObjectMeta {
|
||||
return &obj.(*apiv1.Namespace).ObjectMeta
|
||||
}
|
||||
|
||||
func (a *NamespaceAdapter) FedCreate(obj pkgruntime.Object) (pkgruntime.Object, error) {
|
||||
namespace := obj.(*apiv1.Namespace)
|
||||
return a.client.CoreV1().Namespaces().Create(namespace)
|
||||
}
|
||||
|
||||
func (a *NamespaceAdapter) FedDelete(qualifiedName QualifiedName, options *metav1.DeleteOptions) error {
|
||||
return a.client.CoreV1().Namespaces().Delete(qualifiedName.Name, options)
|
||||
}
|
||||
|
||||
func (a *NamespaceAdapter) FedGet(qualifiedName QualifiedName) (pkgruntime.Object, error) {
|
||||
return a.client.CoreV1().Namespaces().Get(qualifiedName.Name, metav1.GetOptions{})
|
||||
}
|
||||
|
||||
func (a *NamespaceAdapter) FedList(namespace string, options metav1.ListOptions) (pkgruntime.Object, error) {
|
||||
return a.client.CoreV1().Namespaces().List(options)
|
||||
}
|
||||
|
||||
func (a *NamespaceAdapter) FedUpdate(obj pkgruntime.Object) (pkgruntime.Object, error) {
|
||||
namespace := obj.(*apiv1.Namespace)
|
||||
return a.client.CoreV1().Namespaces().Update(namespace)
|
||||
}
|
||||
|
||||
func (a *NamespaceAdapter) FedWatch(namespace string, options metav1.ListOptions) (watch.Interface, error) {
|
||||
return a.client.CoreV1().Namespaces().Watch(options)
|
||||
}
|
||||
|
||||
func (a *NamespaceAdapter) ClusterCreate(client kubeclientset.Interface, obj pkgruntime.Object) (pkgruntime.Object, error) {
|
||||
namespace := obj.(*apiv1.Namespace)
|
||||
return client.CoreV1().Namespaces().Create(namespace)
|
||||
}
|
||||
|
||||
func (a *NamespaceAdapter) ClusterDelete(client kubeclientset.Interface, qualifiedName QualifiedName, options *metav1.DeleteOptions) error {
|
||||
return client.CoreV1().Namespaces().Delete(qualifiedName.Name, options)
|
||||
}
|
||||
|
||||
func (a *NamespaceAdapter) ClusterGet(client kubeclientset.Interface, qualifiedName QualifiedName) (pkgruntime.Object, error) {
|
||||
return client.CoreV1().Namespaces().Get(qualifiedName.Name, metav1.GetOptions{})
|
||||
}
|
||||
|
||||
func (a *NamespaceAdapter) ClusterList(client kubeclientset.Interface, namespace string, options metav1.ListOptions) (pkgruntime.Object, error) {
|
||||
return client.CoreV1().Namespaces().List(options)
|
||||
}
|
||||
|
||||
func (a *NamespaceAdapter) ClusterUpdate(client kubeclientset.Interface, obj pkgruntime.Object) (pkgruntime.Object, error) {
|
||||
namespace := obj.(*apiv1.Namespace)
|
||||
return client.CoreV1().Namespaces().Update(namespace)
|
||||
}
|
||||
|
||||
func (a *NamespaceAdapter) ClusterWatch(client kubeclientset.Interface, namespace string, options metav1.ListOptions) (watch.Interface, error) {
|
||||
return client.CoreV1().Namespaces().Watch(options)
|
||||
}
|
||||
|
||||
func (a *NamespaceAdapter) IsSchedulingAdapter() bool {
|
||||
return false
|
||||
}
|
||||
|
||||
func (a *NamespaceAdapter) NewTestObject(namespace string) pkgruntime.Object {
|
||||
return &apiv1.Namespace{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
GenerateName: "test-namespace-",
|
||||
},
|
||||
Spec: apiv1.NamespaceSpec{
|
||||
Finalizers: []apiv1.FinalizerName{apiv1.FinalizerKubernetes},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// CleanUpNamespace deletes all resources in a given namespace.
|
||||
func (a *NamespaceAdapter) CleanUpNamespace(obj pkgruntime.Object, eventRecorder record.EventRecorder) (pkgruntime.Object, error) {
|
||||
namespace := obj.(*apiv1.Namespace)
|
||||
name := namespace.Name
|
||||
|
||||
// Set Terminating status.
|
||||
updatedNamespace := &apiv1.Namespace{
|
||||
ObjectMeta: namespace.ObjectMeta,
|
||||
Spec: namespace.Spec,
|
||||
Status: apiv1.NamespaceStatus{
|
||||
Phase: apiv1.NamespaceTerminating,
|
||||
},
|
||||
}
|
||||
var err error
|
||||
if namespace.Status.Phase != apiv1.NamespaceTerminating {
|
||||
glog.V(2).Infof("Marking ns %s as terminating", name)
|
||||
eventRecorder.Event(namespace, api.EventTypeNormal, "DeleteNamespace", fmt.Sprintf("Marking for deletion"))
|
||||
_, err = a.FedUpdate(updatedNamespace)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to update namespace: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
if hasFinalizerInSpec(updatedNamespace, apiv1.FinalizerKubernetes) {
|
||||
// Delete resources in this namespace.
|
||||
err = a.deleter.Delete(name)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("error in deleting resources in namespace %s: %v", name, err)
|
||||
}
|
||||
glog.V(2).Infof("Removed kubernetes finalizer from ns %s", name)
|
||||
// Fetch the updated Namespace.
|
||||
obj, err = a.FedGet(QualifiedName{Name: name})
|
||||
updatedNamespace = obj.(*apiv1.Namespace)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("error in fetching updated namespace %s: %s", name, err)
|
||||
}
|
||||
}
|
||||
|
||||
return updatedNamespace, nil
|
||||
}
|
||||
|
||||
func hasFinalizerInSpec(namespace *apiv1.Namespace, finalizer apiv1.FinalizerName) bool {
|
||||
for i := range namespace.Spec.Finalizers {
|
||||
if namespace.Spec.Finalizers[i] == finalizer {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@ -26,7 +26,6 @@ filegroup(
|
||||
":package-srcs",
|
||||
"//federation/pkg/federation-controller/cluster:all-srcs",
|
||||
"//federation/pkg/federation-controller/ingress:all-srcs",
|
||||
"//federation/pkg/federation-controller/namespace:all-srcs",
|
||||
"//federation/pkg/federation-controller/service:all-srcs",
|
||||
"//federation/pkg/federation-controller/sync:all-srcs",
|
||||
"//federation/pkg/federation-controller/util:all-srcs",
|
||||
|
||||
@ -1,75 +0,0 @@
|
||||
package(default_visibility = ["//visibility:public"])
|
||||
|
||||
licenses(["notice"])
|
||||
|
||||
load(
|
||||
"@io_bazel_rules_go//go:def.bzl",
|
||||
"go_library",
|
||||
"go_test",
|
||||
)
|
||||
|
||||
go_library(
|
||||
name = "go_default_library",
|
||||
srcs = ["namespace_controller.go"],
|
||||
tags = ["automanaged"],
|
||||
deps = [
|
||||
"//federation/apis/federation/v1beta1:go_default_library",
|
||||
"//federation/client/clientset_generated/federation_clientset:go_default_library",
|
||||
"//federation/pkg/federation-controller/util:go_default_library",
|
||||
"//federation/pkg/federation-controller/util/deletionhelper:go_default_library",
|
||||
"//federation/pkg/federation-controller/util/eventsink:go_default_library",
|
||||
"//pkg/api:go_default_library",
|
||||
"//pkg/client/clientset_generated/clientset:go_default_library",
|
||||
"//pkg/controller:go_default_library",
|
||||
"//pkg/controller/namespace/deletion:go_default_library",
|
||||
"//vendor/github.com/golang/glog:go_default_library",
|
||||
"//vendor/k8s.io/api/core/v1:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/api/errors:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/watch:go_default_library",
|
||||
"//vendor/k8s.io/client-go/dynamic:go_default_library",
|
||||
"//vendor/k8s.io/client-go/tools/cache:go_default_library",
|
||||
"//vendor/k8s.io/client-go/tools/record:go_default_library",
|
||||
"//vendor/k8s.io/client-go/util/flowcontrol:go_default_library",
|
||||
],
|
||||
)
|
||||
|
||||
go_test(
|
||||
name = "go_default_test",
|
||||
srcs = ["namespace_controller_test.go"],
|
||||
library = ":go_default_library",
|
||||
tags = ["automanaged"],
|
||||
deps = [
|
||||
"//federation/apis/federation/v1beta1:go_default_library",
|
||||
"//federation/client/clientset_generated/federation_clientset/fake:go_default_library",
|
||||
"//federation/pkg/federation-controller/util:go_default_library",
|
||||
"//federation/pkg/federation-controller/util/deletionhelper:go_default_library",
|
||||
"//federation/pkg/federation-controller/util/test:go_default_library",
|
||||
"//pkg/client/clientset_generated/clientset:go_default_library",
|
||||
"//pkg/client/clientset_generated/clientset/fake:go_default_library",
|
||||
"//vendor/github.com/stretchr/testify/assert:go_default_library",
|
||||
"//vendor/github.com/stretchr/testify/require:go_default_library",
|
||||
"//vendor/k8s.io/api/core/v1:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
|
||||
"//vendor/k8s.io/client-go/dynamic:go_default_library",
|
||||
"//vendor/k8s.io/client-go/rest:go_default_library",
|
||||
"//vendor/k8s.io/client-go/testing:go_default_library",
|
||||
],
|
||||
)
|
||||
|
||||
filegroup(
|
||||
name = "package-srcs",
|
||||
srcs = glob(["**"]),
|
||||
tags = ["automanaged"],
|
||||
visibility = ["//visibility:private"],
|
||||
)
|
||||
|
||||
filegroup(
|
||||
name = "all-srcs",
|
||||
srcs = [":package-srcs"],
|
||||
tags = ["automanaged"],
|
||||
)
|
||||
@ -1,460 +0,0 @@
|
||||
/*
|
||||
Copyright 2016 The Kubernetes Authors.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
package namespace
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
apiv1 "k8s.io/api/core/v1"
|
||||
clientv1 "k8s.io/api/core/v1"
|
||||
"k8s.io/apimachinery/pkg/api/errors"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/apimachinery/pkg/watch"
|
||||
"k8s.io/client-go/dynamic"
|
||||
"k8s.io/client-go/tools/cache"
|
||||
"k8s.io/client-go/tools/record"
|
||||
"k8s.io/client-go/util/flowcontrol"
|
||||
federationapi "k8s.io/kubernetes/federation/apis/federation/v1beta1"
|
||||
federationclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_clientset"
|
||||
"k8s.io/kubernetes/federation/pkg/federation-controller/util"
|
||||
"k8s.io/kubernetes/federation/pkg/federation-controller/util/deletionhelper"
|
||||
"k8s.io/kubernetes/federation/pkg/federation-controller/util/eventsink"
|
||||
"k8s.io/kubernetes/pkg/api"
|
||||
kubeclientset "k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
|
||||
"k8s.io/kubernetes/pkg/controller"
|
||||
"k8s.io/kubernetes/pkg/controller/namespace/deletion"
|
||||
|
||||
"github.com/golang/glog"
|
||||
)
|
||||
|
||||
const (
|
||||
allClustersKey = "ALL_CLUSTERS"
|
||||
ControllerName = "namespaces"
|
||||
UserAgentName = "federation-namespace-controller"
|
||||
)
|
||||
|
||||
var (
|
||||
RequiredResources = []schema.GroupVersionResource{apiv1.SchemeGroupVersion.WithResource("namespaces")}
|
||||
)
|
||||
|
||||
type NamespaceController struct {
|
||||
// For triggering single namespace reconciliation. This is used when there is an
|
||||
// add/update/delete operation on a namespace in either federated API server or
|
||||
// in some member of the federation.
|
||||
namespaceDeliverer *util.DelayingDeliverer
|
||||
|
||||
// For triggering all namespaces reconciliation. This is used when
|
||||
// a new cluster becomes available.
|
||||
clusterDeliverer *util.DelayingDeliverer
|
||||
|
||||
// Contains namespaces present in members of federation.
|
||||
namespaceFederatedInformer util.FederatedInformer
|
||||
// For updating members of federation.
|
||||
federatedUpdater util.FederatedUpdater
|
||||
// Definitions of namespaces that should be federated.
|
||||
namespaceInformerStore cache.Store
|
||||
// Informer controller for namespaces that should be federated.
|
||||
namespaceInformerController cache.Controller
|
||||
|
||||
// Client to federated api server.
|
||||
federatedApiClient federationclientset.Interface
|
||||
|
||||
// Backoff manager for namespaces
|
||||
namespaceBackoff *flowcontrol.Backoff
|
||||
|
||||
// For events
|
||||
eventRecorder record.EventRecorder
|
||||
|
||||
deletionHelper *deletionhelper.DeletionHelper
|
||||
|
||||
// Helper to delete all resources in a namespace.
|
||||
namespacedResourcesDeleter deletion.NamespacedResourcesDeleterInterface
|
||||
|
||||
namespaceReviewDelay time.Duration
|
||||
clusterAvailableDelay time.Duration
|
||||
smallDelay time.Duration
|
||||
updateTimeout time.Duration
|
||||
}
|
||||
|
||||
// NewNamespaceController returns a new namespace controller
|
||||
func NewNamespaceController(client federationclientset.Interface, dynamicClientPool dynamic.ClientPool) *NamespaceController {
|
||||
broadcaster := record.NewBroadcaster()
|
||||
broadcaster.StartRecordingToSink(eventsink.NewFederatedEventSink(client))
|
||||
recorder := broadcaster.NewRecorder(api.Scheme, clientv1.EventSource{Component: UserAgentName})
|
||||
|
||||
nc := &NamespaceController{
|
||||
federatedApiClient: client,
|
||||
namespaceReviewDelay: time.Second * 10,
|
||||
clusterAvailableDelay: time.Second * 20,
|
||||
smallDelay: time.Second * 3,
|
||||
updateTimeout: time.Second * 30,
|
||||
namespaceBackoff: flowcontrol.NewBackOff(5*time.Second, time.Minute),
|
||||
eventRecorder: recorder,
|
||||
}
|
||||
|
||||
// Build deliverers for triggering reconciliations.
|
||||
nc.namespaceDeliverer = util.NewDelayingDeliverer()
|
||||
nc.clusterDeliverer = util.NewDelayingDeliverer()
|
||||
|
||||
// Start informer in federated API servers on namespaces that should be federated.
|
||||
nc.namespaceInformerStore, nc.namespaceInformerController = cache.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
|
||||
return client.Core().Namespaces().List(options)
|
||||
},
|
||||
WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
|
||||
return client.Core().Namespaces().Watch(options)
|
||||
},
|
||||
},
|
||||
&apiv1.Namespace{},
|
||||
controller.NoResyncPeriodFunc(),
|
||||
util.NewTriggerOnAllChanges(func(obj runtime.Object) { nc.deliverNamespaceObj(obj, 0, false) }))
|
||||
|
||||
// Federated informer on namespaces in members of federation.
|
||||
nc.namespaceFederatedInformer = util.NewFederatedInformer(
|
||||
client,
|
||||
func(cluster *federationapi.Cluster, targetClient kubeclientset.Interface) (cache.Store, cache.Controller) {
|
||||
return cache.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
|
||||
return targetClient.Core().Namespaces().List(options)
|
||||
},
|
||||
WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
|
||||
return targetClient.Core().Namespaces().Watch(options)
|
||||
},
|
||||
},
|
||||
&apiv1.Namespace{},
|
||||
controller.NoResyncPeriodFunc(),
|
||||
// Trigger reconciliation whenever something in federated cluster is changed. In most cases it
|
||||
// would be just confirmation that some namespace operation succeeded.
|
||||
util.NewTriggerOnMetaAndSpecChanges(
|
||||
func(obj runtime.Object) { nc.deliverNamespaceObj(obj, nc.namespaceReviewDelay, false) },
|
||||
))
|
||||
},
|
||||
&util.ClusterLifecycleHandlerFuncs{
|
||||
ClusterAvailable: func(cluster *federationapi.Cluster) {
|
||||
// When new cluster becomes available process all the namespaces again.
|
||||
nc.clusterDeliverer.DeliverAfter(allClustersKey, nil, nc.clusterAvailableDelay)
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
// Federated updater along with Create/Update/Delete operations.
|
||||
nc.federatedUpdater = util.NewFederatedUpdater(nc.namespaceFederatedInformer, "namespace", nc.updateTimeout, nc.eventRecorder,
|
||||
func(client kubeclientset.Interface, obj runtime.Object) error {
|
||||
namespace := obj.(*apiv1.Namespace)
|
||||
_, err := client.Core().Namespaces().Create(namespace)
|
||||
return err
|
||||
},
|
||||
func(client kubeclientset.Interface, obj runtime.Object) error {
|
||||
namespace := obj.(*apiv1.Namespace)
|
||||
_, err := client.Core().Namespaces().Update(namespace)
|
||||
return err
|
||||
},
|
||||
func(client kubeclientset.Interface, obj runtime.Object) error {
|
||||
namespace := obj.(*apiv1.Namespace)
|
||||
orphanDependents := false
|
||||
err := client.Core().Namespaces().Delete(namespace.Name, &metav1.DeleteOptions{OrphanDependents: &orphanDependents})
|
||||
// IsNotFound error is fine since that means the object is deleted already.
|
||||
if errors.IsNotFound(err) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
})
|
||||
|
||||
nc.deletionHelper = deletionhelper.NewDeletionHelper(
|
||||
nc.updateNamespace,
|
||||
// objNameFunc
|
||||
func(obj runtime.Object) string {
|
||||
namespace := obj.(*apiv1.Namespace)
|
||||
return fmt.Sprintf("%s/%s", namespace.Namespace, namespace.Name)
|
||||
},
|
||||
nc.namespaceFederatedInformer,
|
||||
nc.federatedUpdater,
|
||||
)
|
||||
|
||||
discoverResourcesFn := nc.federatedApiClient.Discovery().ServerPreferredNamespacedResources
|
||||
nc.namespacedResourcesDeleter = deletion.NewNamespacedResourcesDeleter(
|
||||
client.Core().Namespaces(), dynamicClientPool, nil,
|
||||
discoverResourcesFn, apiv1.FinalizerKubernetes, false)
|
||||
return nc
|
||||
}
|
||||
|
||||
// Sends the given update object to apiserver.
|
||||
// Assumes that the given object is a namespace.
|
||||
func (nc *NamespaceController) updateNamespace(obj runtime.Object) (runtime.Object, error) {
|
||||
namespace := obj.(*apiv1.Namespace)
|
||||
return nc.federatedApiClient.Core().Namespaces().Update(namespace)
|
||||
}
|
||||
|
||||
// Returns true if the given object has the given finalizer in its NamespaceSpec.
|
||||
func (nc *NamespaceController) hasFinalizerFuncInSpec(obj runtime.Object, finalizer apiv1.FinalizerName) bool {
|
||||
namespace := obj.(*apiv1.Namespace)
|
||||
for i := range namespace.Spec.Finalizers {
|
||||
if namespace.Spec.Finalizers[i] == finalizer {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// Removes the finalizer from the given objects NamespaceSpec.
|
||||
func (nc *NamespaceController) removeFinalizerFromSpec(namespace *apiv1.Namespace, finalizer apiv1.FinalizerName) (*apiv1.Namespace, error) {
|
||||
updatedFinalizers := []apiv1.FinalizerName{}
|
||||
for i := range namespace.Spec.Finalizers {
|
||||
if namespace.Spec.Finalizers[i] != finalizer {
|
||||
updatedFinalizers = append(updatedFinalizers, namespace.Spec.Finalizers[i])
|
||||
}
|
||||
}
|
||||
namespace.Spec.Finalizers = updatedFinalizers
|
||||
updatedNamespace, err := nc.federatedApiClient.Core().Namespaces().Finalize(namespace)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to remove finalizer %s from namespace %s: %v", string(finalizer), namespace.Name, err)
|
||||
}
|
||||
return updatedNamespace, nil
|
||||
}
|
||||
|
||||
func (nc *NamespaceController) Run(stopChan <-chan struct{}) {
|
||||
go nc.namespaceInformerController.Run(stopChan)
|
||||
nc.namespaceFederatedInformer.Start()
|
||||
go func() {
|
||||
<-stopChan
|
||||
nc.namespaceFederatedInformer.Stop()
|
||||
}()
|
||||
nc.namespaceDeliverer.StartWithHandler(func(item *util.DelayingDelivererItem) {
|
||||
namespace := item.Value.(string)
|
||||
nc.reconcileNamespace(namespace)
|
||||
})
|
||||
nc.clusterDeliverer.StartWithHandler(func(_ *util.DelayingDelivererItem) {
|
||||
nc.reconcileNamespacesOnClusterChange()
|
||||
})
|
||||
util.StartBackoffGC(nc.namespaceBackoff, stopChan)
|
||||
}
|
||||
|
||||
func (nc *NamespaceController) deliverNamespaceObj(obj interface{}, delay time.Duration, failed bool) {
|
||||
namespace := obj.(*apiv1.Namespace)
|
||||
nc.deliverNamespace(namespace.Name, delay, failed)
|
||||
}
|
||||
|
||||
// Adds backoff to delay if this delivery is related to some failure. Resets backoff if there was no failure.
|
||||
func (nc *NamespaceController) deliverNamespace(namespace string, delay time.Duration, failed bool) {
|
||||
if failed {
|
||||
nc.namespaceBackoff.Next(namespace, time.Now())
|
||||
delay = delay + nc.namespaceBackoff.Get(namespace)
|
||||
} else {
|
||||
nc.namespaceBackoff.Reset(namespace)
|
||||
}
|
||||
nc.namespaceDeliverer.DeliverAfter(namespace, namespace, delay)
|
||||
}
|
||||
|
||||
// Check whether all data stores are in sync. False is returned if any of the informer/stores is not yet
|
||||
// synced with the corresponding api server.
|
||||
func (nc *NamespaceController) isSynced() bool {
|
||||
if !nc.namespaceFederatedInformer.ClustersSynced() {
|
||||
glog.V(2).Infof("Cluster list not synced")
|
||||
return false
|
||||
}
|
||||
clusters, err := nc.namespaceFederatedInformer.GetReadyClusters()
|
||||
if err != nil {
|
||||
glog.Errorf("Failed to get ready clusters: %v", err)
|
||||
return false
|
||||
}
|
||||
if !nc.namespaceFederatedInformer.GetTargetStore().ClustersSynced(clusters) {
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// The function triggers reconciliation of all federated namespaces.
|
||||
func (nc *NamespaceController) reconcileNamespacesOnClusterChange() {
|
||||
if !nc.isSynced() {
|
||||
nc.clusterDeliverer.DeliverAfter(allClustersKey, nil, nc.clusterAvailableDelay)
|
||||
}
|
||||
for _, obj := range nc.namespaceInformerStore.List() {
|
||||
namespace := obj.(*apiv1.Namespace)
|
||||
nc.deliverNamespace(namespace.Name, nc.smallDelay, false)
|
||||
}
|
||||
}
|
||||
|
||||
func (nc *NamespaceController) reconcileNamespace(namespace string) {
|
||||
if !nc.isSynced() {
|
||||
nc.deliverNamespace(namespace, nc.clusterAvailableDelay, false)
|
||||
return
|
||||
}
|
||||
|
||||
namespaceObjFromStore, exist, err := nc.namespaceInformerStore.GetByKey(namespace)
|
||||
if err != nil {
|
||||
glog.Errorf("Failed to query main namespace store for %v: %v", namespace, err)
|
||||
nc.deliverNamespace(namespace, 0, true)
|
||||
return
|
||||
}
|
||||
|
||||
if !exist {
|
||||
// Not federated namespace, ignoring.
|
||||
return
|
||||
}
|
||||
// Create a copy before modifying the namespace to prevent race condition with
|
||||
// other readers of namespace from store.
|
||||
namespaceObj, err := api.Scheme.DeepCopy(namespaceObjFromStore)
|
||||
baseNamespace, ok := namespaceObj.(*apiv1.Namespace)
|
||||
if err != nil || !ok {
|
||||
glog.Errorf("Error in retrieving obj from store: %v, %v", ok, err)
|
||||
nc.deliverNamespace(namespace, 0, true)
|
||||
return
|
||||
}
|
||||
if baseNamespace.DeletionTimestamp != nil {
|
||||
if err := nc.delete(baseNamespace); err != nil {
|
||||
glog.Errorf("Failed to delete %s: %v", namespace, err)
|
||||
nc.eventRecorder.Eventf(baseNamespace, api.EventTypeWarning, "DeleteFailed",
|
||||
"Namespace delete failed: %v", err)
|
||||
nc.deliverNamespace(namespace, 0, true)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
glog.V(3).Infof("Ensuring delete object from underlying clusters finalizer for namespace: %s",
|
||||
baseNamespace.Name)
|
||||
// Add the required finalizers before creating a namespace in
|
||||
// underlying clusters.
|
||||
// This ensures that the dependent namespaces are deleted in underlying
|
||||
// clusters when the federated namespace is deleted.
|
||||
updatedNamespaceObj, err := nc.deletionHelper.EnsureFinalizers(baseNamespace)
|
||||
if err != nil {
|
||||
glog.Errorf("Failed to ensure delete object from underlying clusters finalizer in namespace %s: %v",
|
||||
baseNamespace.Name, err)
|
||||
nc.deliverNamespace(namespace, 0, false)
|
||||
return
|
||||
}
|
||||
baseNamespace = updatedNamespaceObj.(*apiv1.Namespace)
|
||||
|
||||
glog.V(3).Infof("Syncing namespace %s in underlying clusters", baseNamespace.Name)
|
||||
// Sync the namespace in all underlying clusters.
|
||||
clusters, err := nc.namespaceFederatedInformer.GetReadyClusters()
|
||||
if err != nil {
|
||||
glog.Errorf("Failed to get cluster list: %v", err)
|
||||
nc.deliverNamespace(namespace, nc.clusterAvailableDelay, false)
|
||||
return
|
||||
}
|
||||
|
||||
operations := make([]util.FederatedOperation, 0)
|
||||
for _, cluster := range clusters {
|
||||
clusterNamespaceObj, found, err := nc.namespaceFederatedInformer.GetTargetStore().GetByKey(cluster.Name, namespace)
|
||||
if err != nil {
|
||||
glog.Errorf("Failed to get %s from %s: %v", namespace, cluster.Name, err)
|
||||
nc.deliverNamespace(namespace, 0, true)
|
||||
return
|
||||
}
|
||||
// The object should not be modified.
|
||||
desiredNamespace := &apiv1.Namespace{
|
||||
ObjectMeta: util.DeepCopyRelevantObjectMeta(baseNamespace.ObjectMeta),
|
||||
Spec: *(util.DeepCopyApiTypeOrPanic(&baseNamespace.Spec).(*apiv1.NamespaceSpec)),
|
||||
}
|
||||
glog.V(5).Infof("Desired namespace in underlying clusters: %+v", desiredNamespace)
|
||||
|
||||
if !found {
|
||||
operations = append(operations, util.FederatedOperation{
|
||||
Type: util.OperationTypeAdd,
|
||||
Obj: desiredNamespace,
|
||||
ClusterName: cluster.Name,
|
||||
Key: namespace,
|
||||
})
|
||||
} else {
|
||||
clusterNamespace := clusterNamespaceObj.(*apiv1.Namespace)
|
||||
|
||||
// Update existing namespace, if needed.
|
||||
if !util.ObjectMetaAndSpecEquivalent(desiredNamespace, clusterNamespace) {
|
||||
operations = append(operations, util.FederatedOperation{
|
||||
Type: util.OperationTypeUpdate,
|
||||
Obj: desiredNamespace,
|
||||
ClusterName: cluster.Name,
|
||||
Key: namespace,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if len(operations) == 0 {
|
||||
// Everything is in order
|
||||
return
|
||||
}
|
||||
glog.V(2).Infof("Updating namespace %s in underlying clusters. Operations: %d", baseNamespace.Name, len(operations))
|
||||
|
||||
err = nc.federatedUpdater.Update(operations)
|
||||
if err != nil {
|
||||
glog.Errorf("Failed to execute updates for %s: %v", namespace, err)
|
||||
nc.deliverNamespace(namespace, 0, true)
|
||||
return
|
||||
}
|
||||
|
||||
// Everything is in order but lets be double sure
|
||||
nc.deliverNamespace(namespace, nc.namespaceReviewDelay, false)
|
||||
}
|
||||
|
||||
// delete deletes the given namespace or returns error if the deletion was not complete.
|
||||
func (nc *NamespaceController) delete(namespace *apiv1.Namespace) error {
|
||||
// Set Terminating status.
|
||||
updatedNamespace := &apiv1.Namespace{
|
||||
ObjectMeta: namespace.ObjectMeta,
|
||||
Spec: namespace.Spec,
|
||||
Status: apiv1.NamespaceStatus{
|
||||
Phase: apiv1.NamespaceTerminating,
|
||||
},
|
||||
}
|
||||
var err error
|
||||
if namespace.Status.Phase != apiv1.NamespaceTerminating {
|
||||
glog.V(2).Infof("Marking ns %s as terminating", namespace.Name)
|
||||
nc.eventRecorder.Event(namespace, api.EventTypeNormal, "DeleteNamespace", fmt.Sprintf("Marking for deletion"))
|
||||
_, err = nc.federatedApiClient.Core().Namespaces().Update(updatedNamespace)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to update namespace: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
if nc.hasFinalizerFuncInSpec(updatedNamespace, apiv1.FinalizerKubernetes) {
|
||||
// Delete resources in this namespace.
|
||||
err = nc.namespacedResourcesDeleter.Delete(updatedNamespace.Name)
|
||||
if err != nil {
|
||||
return fmt.Errorf("error in deleting resources in namespace %s: %v", namespace.Name, err)
|
||||
}
|
||||
glog.V(2).Infof("Removed kubernetes finalizer from ns %s", namespace.Name)
|
||||
// Fetch the updated Namespace.
|
||||
updatedNamespace, err = nc.federatedApiClient.Core().Namespaces().Get(updatedNamespace.Name, metav1.GetOptions{})
|
||||
if err != nil {
|
||||
return fmt.Errorf("error in fetching updated namespace %s: %s", updatedNamespace.Name, err)
|
||||
}
|
||||
}
|
||||
|
||||
// Delete the namespace from all underlying clusters.
|
||||
_, err = nc.deletionHelper.HandleObjectInUnderlyingClusters(updatedNamespace)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
err = nc.federatedApiClient.Core().Namespaces().Delete(namespace.Name, nil)
|
||||
if err != nil {
|
||||
// Its all good if the error is not found error. That means it is deleted already and we do not have to do anything.
|
||||
// This is expected when we are processing an update as a result of namespace finalizer deletion.
|
||||
// The process that deleted the last finalizer is also going to delete the namespace and we do not have to do anything.
|
||||
if !errors.IsNotFound(err) {
|
||||
return fmt.Errorf("failed to delete namespace: %v", err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@ -1,188 +0,0 @@
|
||||
/*
|
||||
Copyright 2016 The Kubernetes Authors.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
package namespace
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
apiv1 "k8s.io/api/core/v1"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/util/wait"
|
||||
"k8s.io/client-go/dynamic"
|
||||
restclient "k8s.io/client-go/rest"
|
||||
core "k8s.io/client-go/testing"
|
||||
federationapi "k8s.io/kubernetes/federation/apis/federation/v1beta1"
|
||||
fakefedclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_clientset/fake"
|
||||
"k8s.io/kubernetes/federation/pkg/federation-controller/util"
|
||||
"k8s.io/kubernetes/federation/pkg/federation-controller/util/deletionhelper"
|
||||
. "k8s.io/kubernetes/federation/pkg/federation-controller/util/test"
|
||||
kubeclientset "k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
|
||||
fakekubeclientset "k8s.io/kubernetes/pkg/client/clientset_generated/clientset/fake"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
const (
|
||||
namespaces string = "namespaces"
|
||||
clusters string = "clusters"
|
||||
)
|
||||
|
||||
func TestNamespaceController(t *testing.T) {
|
||||
cluster1 := NewCluster("cluster1", apiv1.ConditionTrue)
|
||||
cluster2 := NewCluster("cluster2", apiv1.ConditionTrue)
|
||||
ns1 := apiv1.Namespace{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "test-namespace",
|
||||
SelfLink: "/api/v1/namespaces/test-namespace",
|
||||
},
|
||||
Spec: apiv1.NamespaceSpec{
|
||||
Finalizers: []apiv1.FinalizerName{apiv1.FinalizerKubernetes},
|
||||
},
|
||||
}
|
||||
|
||||
fakeClient := &fakefedclientset.Clientset{}
|
||||
RegisterFakeList(clusters, &fakeClient.Fake, &federationapi.ClusterList{Items: []federationapi.Cluster{*cluster1}})
|
||||
RegisterFakeList(namespaces, &fakeClient.Fake, &apiv1.NamespaceList{Items: []apiv1.Namespace{}})
|
||||
namespaceWatch := RegisterFakeWatch(namespaces, &fakeClient.Fake)
|
||||
namespaceUpdateChan := RegisterFakeCopyOnUpdate(namespaces, &fakeClient.Fake, namespaceWatch)
|
||||
clusterWatch := RegisterFakeWatch(clusters, &fakeClient.Fake)
|
||||
|
||||
cluster1Client := &fakekubeclientset.Clientset{}
|
||||
cluster1Watch := RegisterFakeWatch(namespaces, &cluster1Client.Fake)
|
||||
RegisterFakeList(namespaces, &cluster1Client.Fake, &apiv1.NamespaceList{Items: []apiv1.Namespace{}})
|
||||
cluster1CreateChan := RegisterFakeCopyOnCreate(namespaces, &cluster1Client.Fake, cluster1Watch)
|
||||
cluster1UpdateChan := RegisterFakeCopyOnUpdate(namespaces, &cluster1Client.Fake, cluster1Watch)
|
||||
|
||||
cluster2Client := &fakekubeclientset.Clientset{}
|
||||
cluster2Watch := RegisterFakeWatch(namespaces, &cluster2Client.Fake)
|
||||
RegisterFakeList(namespaces, &cluster2Client.Fake, &apiv1.NamespaceList{Items: []apiv1.Namespace{}})
|
||||
cluster2CreateChan := RegisterFakeCopyOnCreate(namespaces, &cluster2Client.Fake, cluster2Watch)
|
||||
|
||||
nsDeleteChan := RegisterDelete(&fakeClient.Fake, namespaces)
|
||||
namespaceController := NewNamespaceController(fakeClient, dynamic.NewDynamicClientPool(&restclient.Config{}))
|
||||
informerClientFactory := func(cluster *federationapi.Cluster) (kubeclientset.Interface, error) {
|
||||
switch cluster.Name {
|
||||
case cluster1.Name:
|
||||
return cluster1Client, nil
|
||||
case cluster2.Name:
|
||||
return cluster2Client, nil
|
||||
default:
|
||||
return nil, fmt.Errorf("Unknown cluster")
|
||||
}
|
||||
}
|
||||
setClientFactory(namespaceController.namespaceFederatedInformer, informerClientFactory)
|
||||
namespaceController.clusterAvailableDelay = time.Second
|
||||
namespaceController.namespaceReviewDelay = 50 * time.Millisecond
|
||||
namespaceController.smallDelay = 20 * time.Millisecond
|
||||
namespaceController.updateTimeout = 5 * time.Second
|
||||
|
||||
stop := make(chan struct{})
|
||||
namespaceController.Run(stop)
|
||||
|
||||
// Test add federated namespace.
|
||||
namespaceWatch.Add(&ns1)
|
||||
// Verify that the DeleteFromUnderlyingClusters finalizer is added to the namespace.
|
||||
updatedNamespace := GetNamespaceFromChan(namespaceUpdateChan)
|
||||
require.NotNil(t, updatedNamespace)
|
||||
AssertHasFinalizer(t, updatedNamespace, deletionhelper.FinalizerDeleteFromUnderlyingClusters)
|
||||
ns1 = *updatedNamespace
|
||||
|
||||
// Verify that the namespace is created in underlying cluster1.
|
||||
createdNamespace := GetNamespaceFromChan(cluster1CreateChan)
|
||||
require.NotNil(t, createdNamespace)
|
||||
assert.Equal(t, ns1.Name, createdNamespace.Name)
|
||||
|
||||
// Wait for the namespace to appear in the informer store
|
||||
err := WaitForStoreUpdate(
|
||||
namespaceController.namespaceFederatedInformer.GetTargetStore(),
|
||||
cluster1.Name, ns1.Name, wait.ForeverTestTimeout)
|
||||
assert.Nil(t, err, "namespace should have appeared in the informer store")
|
||||
|
||||
// Test update federated namespace.
|
||||
ns1.Annotations = map[string]string{
|
||||
"A": "B",
|
||||
}
|
||||
namespaceWatch.Modify(&ns1)
|
||||
assert.NoError(t, CheckObjectFromChan(cluster1UpdateChan, MetaAndSpecCheckingFunction(&ns1)))
|
||||
|
||||
// Test add cluster
|
||||
clusterWatch.Add(cluster2)
|
||||
createdNamespace2 := GetNamespaceFromChan(cluster2CreateChan)
|
||||
require.NotNil(t, createdNamespace2)
|
||||
assert.Equal(t, ns1.Name, createdNamespace2.Name)
|
||||
assert.Contains(t, createdNamespace2.Annotations, "A")
|
||||
|
||||
// Delete the namespace with orphan finalizer (let namespaces
|
||||
// in underlying clusters be as is).
|
||||
// TODO: Add a test without orphan finalizer.
|
||||
ns1.ObjectMeta.Finalizers = append(ns1.ObjectMeta.Finalizers, metav1.FinalizerOrphanDependents)
|
||||
ns1.DeletionTimestamp = &metav1.Time{Time: time.Now()}
|
||||
namespaceWatch.Modify(&ns1)
|
||||
assert.Equal(t, ns1.Name, GetStringFromChan(nsDeleteChan))
|
||||
// TODO: Add a test for verifying that resources in the namespace are deleted
|
||||
// when the namespace is deleted.
|
||||
// Need a fake dynamic client to mock list and delete actions to be able to test this.
|
||||
// TODO: Add a fake dynamic client and test this.
|
||||
// In the meantime, e2e test verify that the resources in a namespace are
|
||||
// deleted when the namespace is deleted.
|
||||
close(stop)
|
||||
}
|
||||
|
||||
func setClientFactory(informer util.FederatedInformer, informerClientFactory func(*federationapi.Cluster) (kubeclientset.Interface, error)) {
|
||||
testInformer := ToFederatedInformerForTestOnly(informer)
|
||||
testInformer.SetClientFactory(informerClientFactory)
|
||||
}
|
||||
|
||||
func RegisterDeleteCollection(client *core.Fake, resource string) chan string {
|
||||
deleteChan := make(chan string, 100)
|
||||
client.AddReactor("delete-collection", resource, func(action core.Action) (bool, runtime.Object, error) {
|
||||
deleteChan <- "all"
|
||||
return true, nil, nil
|
||||
})
|
||||
return deleteChan
|
||||
}
|
||||
|
||||
func RegisterDelete(client *core.Fake, resource string) chan string {
|
||||
deleteChan := make(chan string, 100)
|
||||
client.AddReactor("delete", resource, func(action core.Action) (bool, runtime.Object, error) {
|
||||
deleteAction := action.(core.DeleteAction)
|
||||
deleteChan <- deleteAction.GetName()
|
||||
return true, nil, nil
|
||||
})
|
||||
return deleteChan
|
||||
}
|
||||
|
||||
func GetStringFromChan(c chan string) string {
|
||||
select {
|
||||
case str := <-c:
|
||||
return str
|
||||
case <-time.After(5 * time.Second):
|
||||
return "timedout"
|
||||
}
|
||||
}
|
||||
|
||||
func GetNamespaceFromChan(c chan runtime.Object) *apiv1.Namespace {
|
||||
if namespace := GetObjectFromChan(c); namespace == nil {
|
||||
return nil
|
||||
} else {
|
||||
return namespace.(*apiv1.Namespace)
|
||||
}
|
||||
}
|
||||
@ -416,6 +416,17 @@ func (s *FederationSyncController) objFromCache(kind, key string) (pkgruntime.Ob
|
||||
// delete deletes the given resource or returns error if the deletion was not complete.
|
||||
func (s *FederationSyncController) delete(obj pkgruntime.Object, kind string, qualifiedName federatedtypes.QualifiedName) error {
|
||||
glog.V(3).Infof("Handling deletion of %s %q", kind, qualifiedName)
|
||||
|
||||
// Perform pre-deletion cleanup for the namespace adapter
|
||||
namespaceAdapter, ok := s.adapter.(*federatedtypes.NamespaceAdapter)
|
||||
if ok {
|
||||
var err error
|
||||
obj, err = namespaceAdapter.CleanUpNamespace(obj, s.eventRecorder)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
_, err := s.deletionHelper.HandleObjectInUnderlyingClusters(obj)
|
||||
if err != nil {
|
||||
return err
|
||||
|
||||
@ -66,39 +66,6 @@ var _ = framework.KubeDescribe("Federation namespace [Feature:Federation]", func
|
||||
}
|
||||
})
|
||||
|
||||
It("should be created and deleted successfully", func() {
|
||||
fedframework.SkipUnlessFederated(f.ClientSet)
|
||||
|
||||
nsName = createNamespace(f.FederationClientset.Core().Namespaces())
|
||||
|
||||
By(fmt.Sprintf("Deleting namespace %s", nsName))
|
||||
deleteNamespace(nil, nsName,
|
||||
f.FederationClientset.Core().Namespaces().Get,
|
||||
f.FederationClientset.Core().Namespaces().Delete)
|
||||
By(fmt.Sprintf("Verified that deletion succeeded"))
|
||||
})
|
||||
|
||||
It("should be deleted from underlying clusters when OrphanDependents is false", func() {
|
||||
fedframework.SkipUnlessFederated(f.ClientSet)
|
||||
orphanDependents := false
|
||||
nsName = verifyNsCascadingDeletion(f.FederationClientset.Core().Namespaces(), clusters, &orphanDependents)
|
||||
By(fmt.Sprintf("Verified that namespaces were deleted from underlying clusters"))
|
||||
})
|
||||
|
||||
It("should not be deleted from underlying clusters when OrphanDependents is true", func() {
|
||||
fedframework.SkipUnlessFederated(f.ClientSet)
|
||||
orphanDependents := true
|
||||
nsName = verifyNsCascadingDeletion(f.FederationClientset.Core().Namespaces(), clusters, &orphanDependents)
|
||||
By(fmt.Sprintf("Verified that namespaces were not deleted from underlying clusters"))
|
||||
})
|
||||
|
||||
It("should not be deleted from underlying clusters when OrphanDependents is nil", func() {
|
||||
fedframework.SkipUnlessFederated(f.ClientSet)
|
||||
|
||||
nsName = verifyNsCascadingDeletion(f.FederationClientset.Core().Namespaces(), clusters, nil)
|
||||
By(fmt.Sprintf("Verified that namespaces were not deleted from underlying clusters"))
|
||||
})
|
||||
|
||||
// See https://github.com/kubernetes/kubernetes/issues/38225
|
||||
It("deletes replicasets in the namespace when the namespace is deleted", func() {
|
||||
fedframework.SkipUnlessFederated(f.ClientSet)
|
||||
|
||||
@ -599,7 +599,6 @@ k8s.io/kubernetes/federation/pkg/dnsprovider/providers/google/clouddns,madhusuda
|
||||
k8s.io/kubernetes/federation/pkg/federation-controller/cluster,nikhiljindal,0,
|
||||
k8s.io/kubernetes/federation/pkg/federation-controller/deployment,zmerlynn,1,
|
||||
k8s.io/kubernetes/federation/pkg/federation-controller/ingress,vishh,1,
|
||||
k8s.io/kubernetes/federation/pkg/federation-controller/namespace,rrati,0,
|
||||
k8s.io/kubernetes/federation/pkg/federation-controller/replicaset,roberthbailey,1,
|
||||
k8s.io/kubernetes/federation/pkg/federation-controller/secret,apelisse,1,
|
||||
k8s.io/kubernetes/federation/pkg/federation-controller/service,pmorie,1,
|
||||
|
||||
|
Loading…
Reference in New Issue
Block a user