Remove delay when deleting namespaces, move to new controller framework

This commit is contained in:
derekwaynecarr 2015-04-13 13:15:27 -04:00
parent 810ad7116d
commit c1a3fa0dae
4 changed files with 35 additions and 58 deletions

View File

@ -81,7 +81,7 @@ func NewCMServer() *CMServer {
Address: util.IP(net.ParseIP("127.0.0.1")), Address: util.IP(net.ParseIP("127.0.0.1")),
NodeSyncPeriod: 10 * time.Second, NodeSyncPeriod: 10 * time.Second,
ResourceQuotaSyncPeriod: 10 * time.Second, ResourceQuotaSyncPeriod: 10 * time.Second,
NamespaceSyncPeriod: 1 * time.Minute, NamespaceSyncPeriod: 5 * time.Minute,
RegisterRetryCount: 10, RegisterRetryCount: 10,
PodEvictionTimeout: 5 * time.Minute, PodEvictionTimeout: 5 * time.Minute,
NodeMilliCPU: 1000, NodeMilliCPU: 1000,
@ -207,8 +207,8 @@ func (s *CMServer) Run(_ []string) error {
resourceQuotaManager := resourcequota.NewResourceQuotaManager(kubeClient) resourceQuotaManager := resourcequota.NewResourceQuotaManager(kubeClient)
resourceQuotaManager.Run(s.ResourceQuotaSyncPeriod) resourceQuotaManager.Run(s.ResourceQuotaSyncPeriod)
namespaceManager := namespace.NewNamespaceManager(kubeClient) namespaceManager := namespace.NewNamespaceManager(kubeClient, s.NamespaceSyncPeriod)
namespaceManager.Run(s.NamespaceSyncPeriod) namespaceManager.Run()
select {} select {}
return nil return nil

View File

@ -172,7 +172,7 @@ func DeletionHandlingMetaNamespaceKeyFunc(obj interface{}) (string, error) {
// Parameters: // Parameters:
// * lw is list and watch functions for the source of the resource you want to // * lw is list and watch functions for the source of the resource you want to
// be informed of. // be informed of.
// * objType is an object of the type that you expect to receieve. // * objType is an object of the type that you expect to receive.
// * resyncPeriod: if non-zero, will re-list this often (you will get OnUpdate // * resyncPeriod: if non-zero, will re-list this often (you will get OnUpdate
// calls, even if nothing changed). Otherwise, re-list will be delayed as // calls, even if nothing changed). Otherwise, re-list will be delayed as
// long as possible (until the upstream source closes the watch or times out, // long as possible (until the upstream source closes the watch or times out,

View File

@ -17,34 +17,28 @@ limitations under the License.
package namespace package namespace
import ( import (
"sync"
"time" "time"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api" "github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/client" "github.com/GoogleCloudPlatform/kubernetes/pkg/client"
"github.com/GoogleCloudPlatform/kubernetes/pkg/client/cache" "github.com/GoogleCloudPlatform/kubernetes/pkg/client/cache"
"github.com/GoogleCloudPlatform/kubernetes/pkg/controller/framework"
"github.com/GoogleCloudPlatform/kubernetes/pkg/fields" "github.com/GoogleCloudPlatform/kubernetes/pkg/fields"
"github.com/GoogleCloudPlatform/kubernetes/pkg/labels" "github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime" "github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util" "github.com/GoogleCloudPlatform/kubernetes/pkg/util"
"github.com/GoogleCloudPlatform/kubernetes/pkg/watch" "github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
"github.com/golang/glog"
) )
// NamespaceManager is responsible for performing actions dependent upon a namespace phase // NamespaceManager is responsible for performing actions dependent upon a namespace phase
type NamespaceManager struct { type NamespaceManager struct {
kubeClient client.Interface controller *framework.Controller
store cache.Store StopEverything chan struct{}
syncTime <-chan time.Time
// To allow injection for testing.
syncHandler func(namespace api.Namespace) error
} }
// NewNamespaceManager creates a new NamespaceManager // NewNamespaceManager creates a new NamespaceManager
func NewNamespaceManager(kubeClient client.Interface) *NamespaceManager { func NewNamespaceManager(kubeClient client.Interface, resyncPeriod time.Duration) *NamespaceManager {
store := cache.NewStore(cache.MetaNamespaceKeyFunc) _, controller := framework.NewInformer(
reflector := cache.NewReflector(
&cache.ListWatch{ &cache.ListWatch{
ListFunc: func() (runtime.Object, error) { ListFunc: func() (runtime.Object, error) {
return kubeClient.Namespaces().List(labels.Everything(), fields.Everything()) return kubeClient.Namespaces().List(labels.Everything(), fields.Everything())
@ -54,42 +48,28 @@ func NewNamespaceManager(kubeClient client.Interface) *NamespaceManager {
}, },
}, },
&api.Namespace{}, &api.Namespace{},
store, resyncPeriod,
0, framework.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) {
namespace := obj.(*api.Namespace)
syncNamespace(kubeClient, *namespace)
},
UpdateFunc: func(oldObj, newObj interface{}) {
namespace := newObj.(*api.Namespace)
syncNamespace(kubeClient, *namespace)
},
},
) )
reflector.Run()
nm := &NamespaceManager{ return &NamespaceManager{
kubeClient: kubeClient, controller: controller,
store: store, StopEverything: make(chan struct{}),
} }
// set the synchronization handler
nm.syncHandler = nm.syncNamespace
return nm
} }
// Run begins syncing at the specified period interval // Run begins observing the system. It starts a goroutine and returns immediately.
func (nm *NamespaceManager) Run(period time.Duration) { func (nm *NamespaceManager) Run() {
nm.syncTime = time.Tick(period) go nm.controller.Run(nm.StopEverything)
go util.Forever(func() { nm.synchronize() }, period)
}
// Iterate over the each namespace that is in terminating phase and perform necessary clean-up
func (nm *NamespaceManager) synchronize() {
namespaceObjs := nm.store.List()
wg := sync.WaitGroup{}
wg.Add(len(namespaceObjs))
for ix := range namespaceObjs {
go func(ix int) {
defer wg.Done()
namespace := namespaceObjs[ix].(*api.Namespace)
glog.V(4).Infof("periodic sync of namespace: %v", namespace.Name)
err := nm.syncHandler(*namespace)
if err != nil {
glog.Errorf("Error synchronizing: %v", err)
}
}(ix)
}
wg.Wait()
} }
// finalized returns true if the spec.finalizers is empty list // finalized returns true if the spec.finalizers is empty list
@ -153,7 +133,7 @@ func deleteAllContent(kubeClient client.Interface, namespace string) (err error)
} }
// syncNamespace makes namespace life-cycle decisions // syncNamespace makes namespace life-cycle decisions
func (nm *NamespaceManager) syncNamespace(namespace api.Namespace) (err error) { func syncNamespace(kubeClient client.Interface, namespace api.Namespace) (err error) {
if namespace.DeletionTimestamp == nil { if namespace.DeletionTimestamp == nil {
return nil return nil
} }
@ -164,7 +144,7 @@ func (nm *NamespaceManager) syncNamespace(namespace api.Namespace) (err error) {
newNamespace.ObjectMeta = namespace.ObjectMeta newNamespace.ObjectMeta = namespace.ObjectMeta
newNamespace.Status = namespace.Status newNamespace.Status = namespace.Status
newNamespace.Status.Phase = api.NamespaceTerminating newNamespace.Status.Phase = api.NamespaceTerminating
result, err := nm.kubeClient.Namespaces().Status(&newNamespace) result, err := kubeClient.Namespaces().Status(&newNamespace)
if err != nil { if err != nil {
return err return err
} }
@ -174,25 +154,25 @@ func (nm *NamespaceManager) syncNamespace(namespace api.Namespace) (err error) {
// if the namespace is already finalized, delete it // if the namespace is already finalized, delete it
if finalized(namespace) { if finalized(namespace) {
err = nm.kubeClient.Namespaces().Delete(namespace.Name) err = kubeClient.Namespaces().Delete(namespace.Name)
return err return err
} }
// there may still be content for us to remove // there may still be content for us to remove
err = deleteAllContent(nm.kubeClient, namespace.Name) err = deleteAllContent(kubeClient, namespace.Name)
if err != nil { if err != nil {
return err return err
} }
// we have removed content, so mark it finalized by us // we have removed content, so mark it finalized by us
result, err := finalize(nm.kubeClient, namespace) result, err := finalize(kubeClient, namespace)
if err != nil { if err != nil {
return err return err
} }
// now check if all finalizers have reported that we delete now // now check if all finalizers have reported that we delete now
if finalized(*result) { if finalized(*result) {
err = nm.kubeClient.Namespaces().Delete(namespace.Name) err = kubeClient.Namespaces().Delete(namespace.Name)
return err return err
} }

View File

@ -20,7 +20,6 @@ import (
"testing" "testing"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api" "github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/client/cache"
"github.com/GoogleCloudPlatform/kubernetes/pkg/client/testclient" "github.com/GoogleCloudPlatform/kubernetes/pkg/client/testclient"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util" "github.com/GoogleCloudPlatform/kubernetes/pkg/util"
) )
@ -69,7 +68,6 @@ func TestFinalize(t *testing.T) {
func TestSyncNamespaceThatIsTerminating(t *testing.T) { func TestSyncNamespaceThatIsTerminating(t *testing.T) {
mockClient := &testclient.Fake{} mockClient := &testclient.Fake{}
nm := NamespaceManager{kubeClient: mockClient, store: cache.NewStore(cache.MetaNamespaceKeyFunc)}
now := util.Now() now := util.Now()
testNamespace := api.Namespace{ testNamespace := api.Namespace{
ObjectMeta: api.ObjectMeta{ ObjectMeta: api.ObjectMeta{
@ -84,7 +82,7 @@ func TestSyncNamespaceThatIsTerminating(t *testing.T) {
Phase: api.NamespaceTerminating, Phase: api.NamespaceTerminating,
}, },
} }
err := nm.syncNamespace(testNamespace) err := syncNamespace(mockClient, testNamespace)
if err != nil { if err != nil {
t.Errorf("Unexpected error when synching namespace %v", err) t.Errorf("Unexpected error when synching namespace %v", err)
} }
@ -109,7 +107,6 @@ func TestSyncNamespaceThatIsTerminating(t *testing.T) {
func TestSyncNamespaceThatIsActive(t *testing.T) { func TestSyncNamespaceThatIsActive(t *testing.T) {
mockClient := &testclient.Fake{} mockClient := &testclient.Fake{}
nm := NamespaceManager{kubeClient: mockClient, store: cache.NewStore(cache.MetaNamespaceKeyFunc)}
testNamespace := api.Namespace{ testNamespace := api.Namespace{
ObjectMeta: api.ObjectMeta{ ObjectMeta: api.ObjectMeta{
Name: "test", Name: "test",
@ -122,7 +119,7 @@ func TestSyncNamespaceThatIsActive(t *testing.T) {
Phase: api.NamespaceActive, Phase: api.NamespaceActive,
}, },
} }
err := nm.syncNamespace(testNamespace) err := syncNamespace(mockClient, testNamespace)
if err != nil { if err != nil {
t.Errorf("Unexpected error when synching namespace %v", err) t.Errorf("Unexpected error when synching namespace %v", err)
} }