Commit 4855917b authored by Andy Goldstein's avatar Andy Goldstein

Fix possible race in operationNotSupportedCache

Because we can run multiple workers to delete namespaces simultaneously, the operationNotSupportedCache needs to be guarded with a mutex to avoid concurrent map read/write errors.
parent 157b9279
...@@ -50,7 +50,7 @@ type NamespaceController struct { ...@@ -50,7 +50,7 @@ type NamespaceController struct {
// list of preferred group versions and their corresponding resource set for namespace deletion // list of preferred group versions and their corresponding resource set for namespace deletion
groupVersionResources []unversioned.GroupVersionResource groupVersionResources []unversioned.GroupVersionResource
// opCache is a cache to remember if a particular operation is not supported to aid dynamic client. // opCache is a cache to remember if a particular operation is not supported to aid dynamic client.
opCache operationNotSupportedCache opCache *operationNotSupportedCache
// finalizerToken is the finalizer token managed by this controller // finalizerToken is the finalizer token managed by this controller
finalizerToken api.FinalizerName finalizerToken api.FinalizerName
} }
...@@ -70,13 +70,15 @@ func NewNamespaceController( ...@@ -70,13 +70,15 @@ func NewNamespaceController(
// we found in practice though that some auth engines when encountering paths they don't know about may return a 50x. // we found in practice though that some auth engines when encountering paths they don't know about may return a 50x.
// until we have verbs, we pre-populate resources that do not support list or delete for well-known apis rather than // until we have verbs, we pre-populate resources that do not support list or delete for well-known apis rather than
// probing the server once in order to be told no. // probing the server once in order to be told no.
opCache := operationNotSupportedCache{} opCache := &operationNotSupportedCache{
m: make(map[operationKey]bool),
}
ignoredGroupVersionResources := []unversioned.GroupVersionResource{ ignoredGroupVersionResources := []unversioned.GroupVersionResource{
{Group: "", Version: "v1", Resource: "bindings"}, {Group: "", Version: "v1", Resource: "bindings"},
} }
for _, ignoredGroupVersionResource := range ignoredGroupVersionResources { for _, ignoredGroupVersionResource := range ignoredGroupVersionResources {
opCache[operationKey{op: operationDeleteCollection, gvr: ignoredGroupVersionResource}] = true opCache.setNotSupported(operationKey{op: operationDeleteCollection, gvr: ignoredGroupVersionResource})
opCache[operationKey{op: operationList, gvr: ignoredGroupVersionResource}] = true opCache.setNotSupported(operationKey{op: operationList, gvr: ignoredGroupVersionResource})
} }
// create the controller so we can inject the enqueue function // create the controller so we can inject the enqueue function
......
...@@ -158,7 +158,7 @@ func testSyncNamespaceThatIsTerminating(t *testing.T, versions *unversioned.APIV ...@@ -158,7 +158,7 @@ func testSyncNamespaceThatIsTerminating(t *testing.T, versions *unversioned.APIV
mockClient := fake.NewSimpleClientset(testInput.testNamespace) mockClient := fake.NewSimpleClientset(testInput.testNamespace)
clientPool := dynamic.NewClientPool(clientConfig, registered.RESTMapper(), dynamic.LegacyAPIPathResolverFunc) clientPool := dynamic.NewClientPool(clientConfig, registered.RESTMapper(), dynamic.LegacyAPIPathResolverFunc)
err := syncNamespace(mockClient, clientPool, operationNotSupportedCache{}, groupVersionResources, testInput.testNamespace, api.FinalizerKubernetes) err := syncNamespace(mockClient, clientPool, &operationNotSupportedCache{m: make(map[operationKey]bool)}, groupVersionResources, testInput.testNamespace, api.FinalizerKubernetes)
if err != nil { if err != nil {
t.Errorf("scenario %s - Unexpected error when synching namespace %v", scenario, err) t.Errorf("scenario %s - Unexpected error when synching namespace %v", scenario, err)
} }
...@@ -227,7 +227,7 @@ func TestSyncNamespaceThatIsActive(t *testing.T) { ...@@ -227,7 +227,7 @@ func TestSyncNamespaceThatIsActive(t *testing.T) {
Phase: api.NamespaceActive, Phase: api.NamespaceActive,
}, },
} }
err := syncNamespace(mockClient, nil, operationNotSupportedCache{}, testGroupVersionResources(), testNamespace, api.FinalizerKubernetes) err := syncNamespace(mockClient, nil, &operationNotSupportedCache{m: make(map[operationKey]bool)}, testGroupVersionResources(), testNamespace, api.FinalizerKubernetes)
if err != nil { if err != nil {
t.Errorf("Unexpected error when synching namespace %v", err) t.Errorf("Unexpected error when synching namespace %v", err)
} }
......
...@@ -19,6 +19,7 @@ package namespace ...@@ -19,6 +19,7 @@ package namespace
import ( import (
"fmt" "fmt"
"sort" "sort"
"sync"
"time" "time"
"k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api"
...@@ -60,11 +61,22 @@ type operationKey struct { ...@@ -60,11 +61,22 @@ type operationKey struct {
// operationNotSupportedCache is a simple cache to remember if an operation is not supported for a resource. // operationNotSupportedCache is a simple cache to remember if an operation is not supported for a resource.
// if the operationKey maps to true, it means the operation is not supported. // if the operationKey maps to true, it means the operation is not supported.
type operationNotSupportedCache map[operationKey]bool type operationNotSupportedCache struct {
lock sync.RWMutex
m map[operationKey]bool
}
// isSupported returns true if the operation is supported // isSupported returns true if the operation is supported
func (o operationNotSupportedCache) isSupported(key operationKey) bool { func (o *operationNotSupportedCache) isSupported(key operationKey) bool {
return !o[key] o.lock.RLock()
defer o.lock.RUnlock()
return !o.m[key]
}
func (o *operationNotSupportedCache) setNotSupported(key operationKey) {
o.lock.Lock()
defer o.lock.Unlock()
o.m[key] = true
} }
// updateNamespaceFunc is a function that makes an update to a namespace // updateNamespaceFunc is a function that makes an update to a namespace
...@@ -148,7 +160,7 @@ func finalizeNamespace(kubeClient clientset.Interface, namespace *api.Namespace, ...@@ -148,7 +160,7 @@ func finalizeNamespace(kubeClient clientset.Interface, namespace *api.Namespace,
// it returns an error if the operation was supported on the server but was unable to complete. // it returns an error if the operation was supported on the server but was unable to complete.
func deleteCollection( func deleteCollection(
dynamicClient *dynamic.Client, dynamicClient *dynamic.Client,
opCache operationNotSupportedCache, opCache *operationNotSupportedCache,
gvr unversioned.GroupVersionResource, gvr unversioned.GroupVersionResource,
namespace string, namespace string,
) (bool, error) { ) (bool, error) {
...@@ -180,7 +192,7 @@ func deleteCollection( ...@@ -180,7 +192,7 @@ func deleteCollection(
// remember next time that this resource does not support delete collection... // remember next time that this resource does not support delete collection...
if errors.IsMethodNotSupported(err) || errors.IsNotFound(err) { if errors.IsMethodNotSupported(err) || errors.IsNotFound(err) {
glog.V(5).Infof("namespace controller - deleteCollection not supported - namespace: %s, gvr: %v", namespace, gvr) glog.V(5).Infof("namespace controller - deleteCollection not supported - namespace: %s, gvr: %v", namespace, gvr)
opCache[key] = true opCache.setNotSupported(key)
return false, nil return false, nil
} }
...@@ -195,7 +207,7 @@ func deleteCollection( ...@@ -195,7 +207,7 @@ func deleteCollection(
// an error if the operation is supported but could not be completed. // an error if the operation is supported but could not be completed.
func listCollection( func listCollection(
dynamicClient *dynamic.Client, dynamicClient *dynamic.Client,
opCache operationNotSupportedCache, opCache *operationNotSupportedCache,
gvr unversioned.GroupVersionResource, gvr unversioned.GroupVersionResource,
namespace string, namespace string,
) (*runtime.UnstructuredList, bool, error) { ) (*runtime.UnstructuredList, bool, error) {
...@@ -225,7 +237,7 @@ func listCollection( ...@@ -225,7 +237,7 @@ func listCollection(
// remember next time that this resource does not support delete collection... // remember next time that this resource does not support delete collection...
if errors.IsMethodNotSupported(err) || errors.IsNotFound(err) { if errors.IsMethodNotSupported(err) || errors.IsNotFound(err) {
glog.V(5).Infof("namespace controller - listCollection not supported - namespace: %s, gvr: %v", namespace, gvr) glog.V(5).Infof("namespace controller - listCollection not supported - namespace: %s, gvr: %v", namespace, gvr)
opCache[key] = true opCache.setNotSupported(key)
return nil, false, nil return nil, false, nil
} }
...@@ -235,7 +247,7 @@ func listCollection( ...@@ -235,7 +247,7 @@ func listCollection(
// deleteEachItem is a helper function that will list the collection of resources and delete each item 1 by 1. // deleteEachItem is a helper function that will list the collection of resources and delete each item 1 by 1.
func deleteEachItem( func deleteEachItem(
dynamicClient *dynamic.Client, dynamicClient *dynamic.Client,
opCache operationNotSupportedCache, opCache *operationNotSupportedCache,
gvr unversioned.GroupVersionResource, gvr unversioned.GroupVersionResource,
namespace string, namespace string,
) error { ) error {
...@@ -263,7 +275,7 @@ func deleteEachItem( ...@@ -263,7 +275,7 @@ func deleteEachItem(
func deleteAllContentForGroupVersionResource( func deleteAllContentForGroupVersionResource(
kubeClient clientset.Interface, kubeClient clientset.Interface,
clientPool dynamic.ClientPool, clientPool dynamic.ClientPool,
opCache operationNotSupportedCache, opCache *operationNotSupportedCache,
gvr unversioned.GroupVersionResource, gvr unversioned.GroupVersionResource,
namespace string, namespace string,
namespaceDeletedAt unversioned.Time, namespaceDeletedAt unversioned.Time,
...@@ -331,7 +343,7 @@ func deleteAllContentForGroupVersionResource( ...@@ -331,7 +343,7 @@ func deleteAllContentForGroupVersionResource(
func deleteAllContent( func deleteAllContent(
kubeClient clientset.Interface, kubeClient clientset.Interface,
clientPool dynamic.ClientPool, clientPool dynamic.ClientPool,
opCache operationNotSupportedCache, opCache *operationNotSupportedCache,
groupVersionResources []unversioned.GroupVersionResource, groupVersionResources []unversioned.GroupVersionResource,
namespace string, namespace string,
namespaceDeletedAt unversioned.Time, namespaceDeletedAt unversioned.Time,
...@@ -358,7 +370,7 @@ func deleteAllContent( ...@@ -358,7 +370,7 @@ func deleteAllContent(
func syncNamespace( func syncNamespace(
kubeClient clientset.Interface, kubeClient clientset.Interface,
clientPool dynamic.ClientPool, clientPool dynamic.ClientPool,
opCache operationNotSupportedCache, opCache *operationNotSupportedCache,
groupVersionResources []unversioned.GroupVersionResource, groupVersionResources []unversioned.GroupVersionResource,
namespace *api.Namespace, namespace *api.Namespace,
finalizerToken api.FinalizerName, finalizerToken api.FinalizerName,
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment