Merge pull request #21469 from wojtek-t/parallel_namespace_deletion

Auto commit by PR queue bot

cluster/gce/config-test.sh

@@ -73,10 +73,11 @@ CONTROLLER_MANAGER_TEST_LOG_LEVEL="${CONTROLLER_MANAGER_TEST_LOG_LEVEL:-$TEST_CL
 SCHEDULER_TEST_LOG_LEVEL="${SCHEDULER_TEST_LOG_LEVEL:-$TEST_CLUSTER_LOG_LEVEL}"
 KUBEPROXY_TEST_LOG_LEVEL="${KUBEPROXY_TEST_LOG_LEVEL:-$TEST_CLUSTER_LOG_LEVEL}"
 
+TEST_CLUSTER_DELETE_COLLECTION_WORKERS="${TEST_CLUSTER_DELETE_COLLECTION_WORKERS:---delete-collection-workers=1}"
 TEST_CLUSTER_RESYNC_PERIOD="${TEST_CLUSTER_RESYNC_PERIOD:---min-resync-period=3m}"
 
 KUBELET_TEST_ARGS="--max-pods=110 --serialize-image-pulls=false --outofdisk-transition-frequency=0"
-APISERVER_TEST_ARGS="--runtime-config=extensions/v1beta1"
+APISERVER_TEST_ARGS="--runtime-config=extensions/v1beta1 ${TEST_CLUSTER_DELETE_COLLECTION_WORKERS}"
 CONTROLLER_MANAGER_TEST_ARGS="${TEST_CLUSTER_RESYNC_PERIOD}"
 SCHEDULER_TEST_ARGS=""
 KUBEPROXY_TEST_ARGS=""

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

@@ -52,6 +52,7 @@ type APIServer struct {
 	CloudConfigFile            string
 	CloudProvider              string
 	CorsAllowedOriginList      []string
+	DeleteCollectionWorkers    int
 	DeprecatedStorageVersion   string
 	EnableLogsSupport          bool
 	EnableProfiling            bool
@@ -97,6 +98,7 @@ func NewAPIServer() *APIServer {
 		APIPrefix:               "/api",
 		AdmissionControl:        "AlwaysAdmit",
 		AuthorizationMode:       "AlwaysAllow",
+		DeleteCollectionWorkers: 1,
 		EnableLogsSupport:       true,
 		EtcdPathPrefix:          genericapiserver.DefaultEtcdPathPrefix,
 		EventTTL:                1 * time.Hour,
@@ -232,6 +234,7 @@ func (s *APIServer) AddFlags(fs *pflag.FlagSet) {
 	fs.MarkDeprecated("service-node-ports", "see --service-node-port-range instead.")
 	fs.StringVar(&s.MasterServiceNamespace, "master-service-namespace", s.MasterServiceNamespace, "The namespace from which the kubernetes master services should be injected into pods")
 	fs.IntVar(&s.MasterCount, "apiserver-count", s.MasterCount, "The number of apiservers running in the cluster")
+	fs.IntVar(&s.DeleteCollectionWorkers, "delete-collection-workers", s.DeleteCollectionWorkers, "Number of workers spawned for DeleteCollection call. These are used to speed up namespace cleanup.")
 	fs.Var(&s.RuntimeConfig, "runtime-config", "A set of key=value pairs that describe runtime configuration that may be passed to apiserver. apis/<groupVersion> key can be used to turn on/off specific api versions. apis/<groupVersion>/<resource> can be used to turn on/off specific resources. api/all and api/legacy are special keys to control all and legacy api versions respectively.")
 	fs.BoolVar(&s.EnableProfiling, "profiling", true, "Enable profiling via web interface host:port/debug/pprof/")
 	// TODO: enable cache in integration tests.

cmd/kube-apiserver/app/server.go

@@ -470,6 +470,7 @@ func Run(s *options.APIServer) error {
 			Serializer:                api.Codecs,
 		},
 		EnableCoreControllers:   true,
+		DeleteCollectionWorkers: s.DeleteCollectionWorkers,
 		EventTTL:                s.EventTTL,
 		KubeletClient:           kubeletClient,
 

docs/admin/kube-apiserver.md

@@ -66,6 +66,7 @@ kube-apiserver
       --cloud-config="": The path to the cloud provider configuration file.  Empty string for no configuration file.
       --cloud-provider="": The provider for cloud services.  Empty string for no provider.
       --cors-allowed-origins=[]: List of allowed origins for CORS, comma separated.  An allowed origin can be a regular expression to support subdomain matching.  If this list is empty CORS will not be enabled.
+      --delete-collection-workers=1: Number of workers spawned for DeleteCollection call. These are used to speed up namespace cleanup.
       --etcd-prefix="/registry": The prefix for all resource paths in etcd.
       --etcd-quorum-read[=false]: If true, enable quorum read
       --etcd-servers=[]: List of etcd servers to watch (http://ip:port), comma separated. Mutually exclusive with -etcd-config
@@ -111,7 +112,7 @@ kube-apiserver
       --watch-cache-sizes=[]: List of watch cache sizes for every resource (pods, nodes, etc.), comma separated. The individual override format: resource#size, where size is a number. It takes effect when watch-cache is enabled.
 ```
 
-###### Auto generated by spf13/cobra on 18-Feb-2016
+###### Auto generated by spf13/cobra on 24-Feb-2016
 
 
 <!-- BEGIN MUNGE: GENERATED_ANALYTICS -->

hack/jenkins/job-configs/kubernetes-e2e.yaml

@@ -142,6 +142,8 @@
                 export KUBELET_TEST_LOG_LEVEL="--v=4"
                 # Increase resync period to simulate production
                 export TEST_CLUSTER_RESYNC_PERIOD="--min-resync-period=12h"
+                # Increase delete collection parallelism
+                export TEST_CLUSTER_DELETE_COLLECTION_WORKERS="--delete-collection-workers=16"
         - 'gce-examples':
             description: 'Run E2E examples test on GCE.'
             timeout: 90

hack/verify-flags/known-flags.txt

@@ -70,6 +70,7 @@ current-replicas
 default-container-cpu-limit
 default-container-mem-limit
 delay-shutdown
+delete-collection-workers
 delete-namespace
 deleting-pods-burst
 deleting-pods-qps

pkg/controller/serviceaccount/tokengetter.go

@@ -71,7 +71,7 @@ func (r *registryGetter) GetSecret(namespace, name string) (*api.Secret, error)
 // uses the specified storage to retrieve service accounts and secrets.
 func NewGetterFromStorageInterface(s storage.Interface) serviceaccount.ServiceAccountTokenGetter {
 	return NewGetterFromRegistries(
-		serviceaccountregistry.NewRegistry(serviceaccountetcd.NewREST(generic.RESTOptions{s, generic.UndecoratedStorage})),
+		serviceaccountregistry.NewRegistry(serviceaccountetcd.NewREST(generic.RESTOptions{Storage: s, Decorator: generic.UndecoratedStorage})),
-		secret.NewRegistry(secretetcd.NewREST(generic.RESTOptions{s, generic.UndecoratedStorage})),
+		secret.NewRegistry(secretetcd.NewREST(generic.RESTOptions{Storage: s, Decorator: generic.UndecoratedStorage})),
 	)
 }

pkg/master/master.go

@@ -92,6 +92,7 @@ type Config struct {
 	*genericapiserver.Config
 
 	EnableCoreControllers   bool
+	DeleteCollectionWorkers int
 	EventTTL                time.Duration
 	KubeletClient           kubeletclient.KubeletClient
 	// Used to start and monitor tunneling
@@ -106,6 +107,7 @@ type Master struct {
 	v1ResourcesStorage map[string]rest.Storage
 
 	enableCoreControllers   bool
+	deleteCollectionWorkers int
 	// registries are internal client APIs for accessing the storage layer
 	// TODO: define the internal typed interface in a way that clients can
 	// also be replaced
@@ -151,6 +153,7 @@ func New(c *Config) (*Master, error) {
 	m := &Master{
 		GenericAPIServer:        s,
 		enableCoreControllers:   c.EnableCoreControllers,
+		deleteCollectionWorkers: c.DeleteCollectionWorkers,
 		tunneler:                c.Tunneler,
 	}
 	m.InstallAPIs(c)
@@ -329,6 +332,7 @@ func (m *Master) initV1ResourcesStorage(c *Config) {
 		return generic.RESTOptions{
 			Storage:                 dbClient(resource),
 			Decorator:               m.StorageDecorator(),
+			DeleteCollectionWorkers: m.deleteCollectionWorkers,
 		}
 	}
 
@@ -612,7 +616,8 @@ func (m *Master) InstallThirdPartyResource(rsrc *extensions.ThirdPartyResource)
 }
 
 func (m *Master) thirdpartyapi(group, kind, version string) *apiserver.APIGroupVersion {
-	resourceStorage := thirdpartyresourcedataetcd.NewREST(generic.RESTOptions{m.thirdPartyStorage, generic.UndecoratedStorage}, group, kind)
+	resourceStorage := thirdpartyresourcedataetcd.NewREST(
+		generic.RESTOptions{m.thirdPartyStorage, generic.UndecoratedStorage, m.deleteCollectionWorkers}, group, kind)
 
 	apiRoot := makeThirdPartyPath("")
 
@@ -664,6 +669,7 @@ func (m *Master) getExtensionResources(c *Config) map[string]rest.Storage {
 		return generic.RESTOptions{
 			Storage:                 c.StorageDestinations.Get(extensions.GroupName, resource),
 			Decorator:               m.StorageDecorator(),
+			DeleteCollectionWorkers: m.deleteCollectionWorkers,
 		}
 	}
 
@@ -671,7 +677,7 @@ func (m *Master) getExtensionResources(c *Config) map[string]rest.Storage {
 	if isEnabled("horizontalpodautoscalers") {
 		m.constructHPAResources(c, storage)
 		controllerStorage := expcontrolleretcd.NewStorage(
-			generic.RESTOptions{c.StorageDestinations.Get("", "replicationControllers"), m.StorageDecorator()})
+			generic.RESTOptions{c.StorageDestinations.Get("", "replicationControllers"), m.StorageDecorator(), m.deleteCollectionWorkers})
 		storage["replicationcontrollers"] = controllerStorage.ReplicationController
 		storage["replicationcontrollers/scale"] = controllerStorage.Scale
 	}
@@ -737,6 +743,7 @@ func (m *Master) constructHPAResources(c *Config, restStorage map[string]rest.St
 		return generic.RESTOptions{
 			Storage:                 c.StorageDestinations.Search([]string{autoscaling.GroupName, extensions.GroupName}, resource),
 			Decorator:               m.StorageDecorator(),
+			DeleteCollectionWorkers: m.deleteCollectionWorkers,
 		}
 	}
 	autoscalerStorage, autoscalerStatusStorage := horizontalpodautoscaleretcd.NewREST(restOptions("horizontalpodautoscalers"))
@@ -773,6 +780,7 @@ func (m *Master) constructJobResources(c *Config, restStorage map[string]rest.St
 		return generic.RESTOptions{
 			Storage:                 c.StorageDestinations.Search([]string{batch.GroupName, extensions.GroupName}, resource),
 			Decorator:               m.StorageDecorator(),
+			DeleteCollectionWorkers: m.deleteCollectionWorkers,
 		}
 	}
 	jobStorage, jobStatusStorage := jobetcd.NewREST(restOptions("jobs"))

pkg/registry/configmap/etcd/etcd.go

@@ -68,6 +68,8 @@ func NewREST(opts generic.RESTOptions) *REST {
 
 		QualifiedResource: api.Resource("configmaps"),
 
+		DeleteCollectionWorkers: opts.DeleteCollectionWorkers,
+
 		CreateStrategy: configmap.Strategy,
 		UpdateStrategy: configmap.Strategy,
 

pkg/registry/configmap/etcd/etcd_test.go

@@ -30,7 +30,7 @@ import (
 
 func newStorage(t *testing.T) (*REST, *etcdtesting.EtcdTestServer) {
 	etcdStorage, server := registrytest.NewEtcdStorage(t, "")
-	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage}
+	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage, 1}
 	return NewREST(restOptions), server
 }
 

pkg/registry/controller/etcd/etcd.go

@@ -66,6 +66,8 @@ func NewREST(opts generic.RESTOptions) (*REST, *StatusREST) {
 		},
 		QualifiedResource: api.Resource("replicationcontrollers"),
 
+		DeleteCollectionWorkers: opts.DeleteCollectionWorkers,
+
 		// Used to validate controller creation
 		CreateStrategy: controller.Strategy,
 

pkg/registry/controller/etcd/etcd_test.go

@@ -30,7 +30,7 @@ import (
 
 func newStorage(t *testing.T) (*REST, *StatusREST, *etcdtesting.EtcdTestServer) {
 	etcdStorage, server := registrytest.NewEtcdStorage(t, "")
-	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage}
+	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage, 1}
 	controllerStorage, statusStorage := NewREST(restOptions)
 	return controllerStorage, statusStorage, server
 }

pkg/registry/daemonset/etcd/etcd.go

@@ -65,6 +65,7 @@ func NewREST(opts generic.RESTOptions) (*REST, *StatusREST) {
 			return daemonset.MatchDaemonSet(label, field)
 		},
 		QualifiedResource:       extensions.Resource("daemonsets"),
+		DeleteCollectionWorkers: opts.DeleteCollectionWorkers,
 
 		// Used to validate daemon set creation
 		CreateStrategy: daemonset.Strategy,

pkg/registry/daemonset/etcd/etcd_test.go

@@ -32,7 +32,7 @@ import (
 
 func newStorage(t *testing.T) (*REST, *StatusREST, *etcdtesting.EtcdTestServer) {
 	etcdStorage, server := registrytest.NewEtcdStorage(t, extensions.GroupName)
-	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage}
+	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage, 1}
 	daemonSetStorage, statusStorage := NewREST(restOptions)
 	return daemonSetStorage, statusStorage, server
 }

pkg/registry/deployment/etcd/etcd.go

@@ -90,6 +90,7 @@ func NewREST(opts generic.RESTOptions) (*REST, *StatusREST, *RollbackREST) {
 			return deployment.MatchDeployment(label, field)
 		},
 		QualifiedResource:       extensions.Resource("deployments"),
+		DeleteCollectionWorkers: opts.DeleteCollectionWorkers,
 
 		// Used to validate deployment creation.
 		CreateStrategy: deployment.Strategy,

pkg/registry/deployment/etcd/etcd_test.go

@@ -36,7 +36,7 @@ import (
 
 func newStorage(t *testing.T) (*DeploymentStorage, *etcdtesting.EtcdTestServer) {
 	etcdStorage, server := registrytest.NewEtcdStorage(t, extensions.GroupName)
-	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage}
+	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage, 1}
 	deploymentStorage := NewStorage(restOptions)
 	return &deploymentStorage, server
 }

pkg/registry/endpoint/etcd/etcd.go

@@ -55,6 +55,7 @@ func NewREST(opts generic.RESTOptions) *REST {
 			return endpoint.MatchEndpoints(label, field)
 		},
 		QualifiedResource:       api.Resource("endpoints"),
+		DeleteCollectionWorkers: opts.DeleteCollectionWorkers,
 
 		CreateStrategy: endpoint.Strategy,
 		UpdateStrategy: endpoint.Strategy,

pkg/registry/endpoint/etcd/etcd_test.go

@@ -30,7 +30,7 @@ import (
 
 func newStorage(t *testing.T) (*REST, *etcdtesting.EtcdTestServer) {
 	etcdStorage, server := registrytest.NewEtcdStorage(t, "")
-	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage}
+	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage, 1}
 	return NewREST(restOptions), server
 }
 

pkg/registry/event/etcd/etcd.go

@@ -57,6 +57,7 @@ func NewREST(opts generic.RESTOptions, ttl uint64) *REST {
 			return ttl, nil
 		},
 		QualifiedResource:       api.Resource("events"),
+		DeleteCollectionWorkers: opts.DeleteCollectionWorkers,
 
 		CreateStrategy: event.Strategy,
 		UpdateStrategy: event.Strategy,

pkg/registry/event/etcd/etcd_test.go

@@ -30,7 +30,7 @@ var testTTL uint64 = 60
 
 func newStorage(t *testing.T) (*REST, *etcdtesting.EtcdTestServer) {
 	etcdStorage, server := registrytest.NewEtcdStorage(t, "")
-	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage}
+	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage, 1}
 	return NewREST(restOptions, testTTL), server
 }
 

pkg/registry/experimental/controller/etcd/etcd_test.go

@@ -30,7 +30,7 @@ import (
 
 func newStorage(t *testing.T) (*ScaleREST, *etcdtesting.EtcdTestServer, storage.Interface) {
 	etcdStorage, server := registrytest.NewEtcdStorage(t, "")
-	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage}
+	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage, 1}
 	return NewStorage(restOptions).Scale, server, etcdStorage
 }
 

pkg/registry/generic/etcd/etcd.go

@@ -19,6 +19,7 @@ package etcd
 import (
 	"fmt"
 	"reflect"
+	"sync"
 
 	"k8s.io/kubernetes/pkg/api"
 	kubeerr "k8s.io/kubernetes/pkg/api/errors"
@@ -32,6 +33,7 @@ import (
 	"k8s.io/kubernetes/pkg/registry/generic"
 	"k8s.io/kubernetes/pkg/runtime"
 	"k8s.io/kubernetes/pkg/storage"
+	utilruntime "k8s.io/kubernetes/pkg/util/runtime"
 	"k8s.io/kubernetes/pkg/util/validation/field"
 	"k8s.io/kubernetes/pkg/watch"
 
@@ -84,6 +86,10 @@ type Etcd struct {
 	// Returns a matcher corresponding to the provided labels and fields.
 	PredicateFunc func(label labels.Selector, field fields.Selector) generic.Matcher
 
+	// DeleteCollectionWorkers is the maximum number of workers in a single
+	// DeleteCollection call.
+	DeleteCollectionWorkers int
+
 	// Called on all objects returned from the underlying store, after
 	// the exit hooks are invoked. Decorators are intended for integrations
 	// that are above etcd and should only be used for specific cases where
@@ -459,16 +465,63 @@ func (e *Etcd) DeleteCollection(ctx api.Context, options *api.DeleteOptions, lis
 	if err != nil {
 		return nil, err
 	}
-	for _, item := range items {
+	// Spawn a number of goroutines, so that we can issue requests to etcd
-		accessor, err := meta.Accessor(item)
+	// in parallel to speed up deletion.
+	// TODO: Make this proportional to the number of items to delete, up to
+	// DeleteCollectionWorkers (it doesn't make much sense to spawn 16
+	// workers to delete 10 items).
+	workersNumber := e.DeleteCollectionWorkers
+	if workersNumber < 1 {
+		workersNumber = 1
+	}
+	wg := sync.WaitGroup{}
+	toProcess := make(chan int, 2*workersNumber)
+	errs := make(chan error, workersNumber+1)
+
+	go func() {
+		defer utilruntime.HandleCrash(func(panicReason interface{}) {
+			errs <- fmt.Errorf("DeleteCollection distributor panicked: %v", panicReason)
+		})
+		for i := 0; i < len(items); i++ {
+			toProcess <- i
+		}
+		close(toProcess)
+	}()
+
+	wg.Add(workersNumber)
+	for i := 0; i < workersNumber; i++ {
+		go func() {
+			// panics don't cross goroutine boundaries
+			defer utilruntime.HandleCrash(func(panicReason interface{}) {
+				errs <- fmt.Errorf("DeleteCollection goroutine panicked: %v", panicReason)
+			})
+			defer wg.Done()
+
+			for {
+				index, ok := <-toProcess
+				if !ok {
+					return
+				}
+				accessor, err := meta.Accessor(items[index])
 				if err != nil {
-			return nil, err
+					errs <- err
+					return
 				}
 				if _, err := e.Delete(ctx, accessor.GetName(), options); err != nil && !kubeerr.IsNotFound(err) {
+					glog.V(4).Infof("Delete %s in DeleteCollection failed: %v", accessor.GetName(), err)
+					errs <- err
+					return
+				}
+			}
+		}()
+	}
+	wg.Wait()
+	select {
+	case err := <-errs:
 		return nil, err
-		}
+	default:
-	}
 		return listObj, nil
+	}
 }
 
 func (e *Etcd) finalizeDelete(obj runtime.Object, runHooks bool) (runtime.Object, error) {

pkg/registry/generic/options.go

@@ -24,4 +24,5 @@ import (
 type RESTOptions struct {
 	Storage                 pkgstorage.Interface
 	Decorator               StorageDecorator
+	DeleteCollectionWorkers int
 }

pkg/registry/horizontalpodautoscaler/etcd/etcd.go

@@ -63,6 +63,7 @@ func NewREST(opts generic.RESTOptions) (*REST, *StatusREST) {
 			return horizontalpodautoscaler.MatchAutoscaler(label, field)
 		},
 		QualifiedResource:       extensions.Resource("horizontalpodautoscalers"),
+		DeleteCollectionWorkers: opts.DeleteCollectionWorkers,
 
 		// Used to validate autoscaler creation
 		CreateStrategy: horizontalpodautoscaler.Strategy,

pkg/registry/horizontalpodautoscaler/etcd/etcd_test.go

@@ -33,7 +33,7 @@ import (
 
 func newStorage(t *testing.T) (*REST, *StatusREST, *etcdtesting.EtcdTestServer) {
 	etcdStorage, server := registrytest.NewEtcdStorage(t, extensions.GroupName)
-	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage}
+	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage, 1}
 	horizontalPodAutoscalerStorage, statusStorage := NewREST(restOptions)
 	return horizontalPodAutoscalerStorage, statusStorage, server
 }

pkg/registry/ingress/etcd/etcd.go

@@ -65,6 +65,7 @@ func NewREST(opts generic.RESTOptions) (*REST, *StatusREST) {
 			return ingress.MatchIngress(label, field)
 		},
 		QualifiedResource:       extensions.Resource("ingresses"),
+		DeleteCollectionWorkers: opts.DeleteCollectionWorkers,
 
 		// Used to validate controller creation
 		CreateStrategy: ingress.Strategy,

pkg/registry/ingress/etcd/etcd_test.go

@@ -32,7 +32,7 @@ import (
 
 func newStorage(t *testing.T) (*REST, *StatusREST, *etcdtesting.EtcdTestServer) {
 	etcdStorage, server := registrytest.NewEtcdStorage(t, extensions.GroupName)
-	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage}
+	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage, 1}
 	ingressStorage, statusStorage := NewREST(restOptions)
 	return ingressStorage, statusStorage, server
 }

pkg/registry/job/etcd/etcd.go

@@ -65,6 +65,7 @@ func NewREST(opts generic.RESTOptions) (*REST, *StatusREST) {
 			return job.MatchJob(label, field)
 		},
 		QualifiedResource:       extensions.Resource("jobs"),
+		DeleteCollectionWorkers: opts.DeleteCollectionWorkers,
 
 		// Used to validate job creation
 		CreateStrategy: job.Strategy,

pkg/registry/job/etcd/etcd_test.go

@@ -34,7 +34,7 @@ import (
 
 func newStorage(t *testing.T) (*REST, *StatusREST, *etcdtesting.EtcdTestServer) {
 	etcdStorage, server := registrytest.NewEtcdStorage(t, extensions.GroupName)
-	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage}
+	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage, 1}
 	jobStorage, statusStorage := NewREST(restOptions)
 	return jobStorage, statusStorage, server
 }

pkg/registry/limitrange/etcd/etcd.go

@@ -55,6 +55,7 @@ func NewREST(opts generic.RESTOptions) *REST {
 			return limitrange.MatchLimitRange(label, field)
 		},
 		QualifiedResource:       api.Resource("limitranges"),
+		DeleteCollectionWorkers: opts.DeleteCollectionWorkers,
 
 		CreateStrategy: limitrange.Strategy,
 		UpdateStrategy: limitrange.Strategy,

pkg/registry/limitrange/etcd/etcd_test.go

@@ -31,7 +31,7 @@ import (
 
 func newStorage(t *testing.T) (*REST, *etcdtesting.EtcdTestServer) {
 	etcdStorage, server := registrytest.NewEtcdStorage(t, "")
-	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage}
+	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage, 1}
 	return NewREST(restOptions), server
 }
 

pkg/registry/namespace/etcd/etcd.go

@@ -71,6 +71,7 @@ func NewREST(opts generic.RESTOptions) (*REST, *StatusREST, *FinalizeREST) {
 			return namespace.MatchNamespace(label, field)
 		},
 		QualifiedResource:       api.Resource("namespaces"),
+		DeleteCollectionWorkers: opts.DeleteCollectionWorkers,
 
 		CreateStrategy:      namespace.Strategy,
 		UpdateStrategy:      namespace.Strategy,

pkg/registry/namespace/etcd/etcd_test.go

@@ -31,7 +31,7 @@ import (
 
 func newStorage(t *testing.T) (*REST, *etcdtesting.EtcdTestServer) {
 	etcdStorage, server := registrytest.NewEtcdStorage(t, "")
-	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage}
+	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage, 1}
 	namespaceStorage, _, _ := NewREST(restOptions)
 	return namespaceStorage, server
 }

pkg/registry/node/etcd/etcd.go

@@ -81,6 +81,7 @@ func NewStorage(opts generic.RESTOptions, connection client.ConnectionInfoGetter
 		},
 		PredicateFunc:           node.MatchNode,
 		QualifiedResource:       api.Resource("nodes"),
+		DeleteCollectionWorkers: opts.DeleteCollectionWorkers,
 
 		CreateStrategy: node.Strategy,
 		UpdateStrategy: node.Strategy,

pkg/registry/node/etcd/etcd_test.go

@@ -39,7 +39,7 @@ func (fakeConnectionInfoGetter) GetConnectionInfo(ctx api.Context, nodeName stri
 
 func newStorage(t *testing.T) (*REST, *etcdtesting.EtcdTestServer) {
 	etcdStorage, server := registrytest.NewEtcdStorage(t, "")
-	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage}
+	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage, 1}
 	storage := NewStorage(restOptions, fakeConnectionInfoGetter{}, nil)
 	return storage.Node, server
 }

pkg/registry/persistentvolume/etcd/etcd.go

@@ -55,6 +55,7 @@ func NewREST(opts generic.RESTOptions) (*REST, *StatusREST) {
 			return persistentvolume.MatchPersistentVolumes(label, field)
 		},
 		QualifiedResource:       api.Resource("persistentvolumes"),
+		DeleteCollectionWorkers: opts.DeleteCollectionWorkers,
 
 		CreateStrategy:      persistentvolume.Strategy,
 		UpdateStrategy:      persistentvolume.Strategy,

pkg/registry/persistentvolume/etcd/etcd_test.go

@@ -33,7 +33,7 @@ import (
 
 func newStorage(t *testing.T) (*REST, *StatusREST, *etcdtesting.EtcdTestServer) {
 	etcdStorage, server := registrytest.NewEtcdStorage(t, "")
-	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage}
+	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage, 1}
 	persistentVolumeStorage, statusStorage := NewREST(restOptions)
 	return persistentVolumeStorage, statusStorage, server
 }

pkg/registry/persistentvolumeclaim/etcd/etcd.go

@@ -55,6 +55,7 @@ func NewREST(opts generic.RESTOptions) (*REST, *StatusREST) {
 			return persistentvolumeclaim.MatchPersistentVolumeClaim(label, field)
 		},
 		QualifiedResource:       api.Resource("persistentvolumeclaims"),
+		DeleteCollectionWorkers: opts.DeleteCollectionWorkers,
 
 		CreateStrategy:      persistentvolumeclaim.Strategy,
 		UpdateStrategy:      persistentvolumeclaim.Strategy,

pkg/registry/persistentvolumeclaim/etcd/etcd_test.go

@@ -33,7 +33,7 @@ import (
 
 func newStorage(t *testing.T) (*REST, *StatusREST, *etcdtesting.EtcdTestServer) {
 	etcdStorage, server := registrytest.NewEtcdStorage(t, "")
-	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage}
+	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage, 1}
 	persistentVolumeClaimStorage, statusStorage := NewREST(restOptions)
 	return persistentVolumeClaimStorage, statusStorage, server
 }

pkg/registry/pod/etcd/etcd.go

@@ -81,6 +81,7 @@ func NewStorage(opts generic.RESTOptions, k client.ConnectionInfoGetter, proxyTr
 			return pod.MatchPod(label, field)
 		},
 		QualifiedResource:       api.Resource("pods"),
+		DeleteCollectionWorkers: opts.DeleteCollectionWorkers,
 
 		CreateStrategy:      pod.Strategy,
 		UpdateStrategy:      pod.Strategy,

pkg/registry/pod/etcd/etcd_test.go

@@ -37,7 +37,7 @@ import (
 
 func newStorage(t *testing.T) (*REST, *BindingREST, *StatusREST, *etcdtesting.EtcdTestServer) {
 	etcdStorage, server := registrytest.NewEtcdStorage(t, "")
-	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage}
+	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage, 3}
 	storage := NewStorage(restOptions, nil, nil)
 	return storage.Pod, storage.Binding, storage.Status, server
 }

pkg/registry/podsecuritypolicy/etcd/etcd.go

@@ -56,6 +56,7 @@ func NewREST(opts generic.RESTOptions) *REST {
 			return podsecuritypolicy.MatchPodSecurityPolicy(label, field)
 		},
 		QualifiedResource:       extensions.Resource("podsecuritypolicies"),
+		DeleteCollectionWorkers: opts.DeleteCollectionWorkers,
 
 		CreateStrategy:      podsecuritypolicy.Strategy,
 		UpdateStrategy:      podsecuritypolicy.Strategy,

pkg/registry/podsecuritypolicy/etcd/etcd_test.go

@@ -33,7 +33,7 @@ import (
 
 func newStorage(t *testing.T) (*REST, *etcdtesting.EtcdTestServer) {
 	etcdStorage, server := registrytest.NewEtcdStorage(t, "extensions")
-	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage}
+	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage, 1}
 	return NewREST(restOptions), server
 }
 

pkg/registry/podtemplate/etcd/etcd.go

@@ -55,6 +55,7 @@ func NewREST(opts generic.RESTOptions) *REST {
 			return podtemplate.MatchPodTemplate(label, field)
 		},
 		QualifiedResource:       api.Resource("podtemplates"),
+		DeleteCollectionWorkers: opts.DeleteCollectionWorkers,
 
 		CreateStrategy: podtemplate.Strategy,
 		UpdateStrategy: podtemplate.Strategy,

pkg/registry/podtemplate/etcd/etcd_test.go

@@ -30,7 +30,7 @@ import (
 
 func newStorage(t *testing.T) (*REST, *etcdtesting.EtcdTestServer) {
 	etcdStorage, server := registrytest.NewEtcdStorage(t, "")
-	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage}
+	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage, 1}
 	return NewREST(restOptions), server
 }
 

pkg/registry/replicaset/etcd/etcd.go

@@ -81,6 +81,7 @@ func NewREST(opts generic.RESTOptions) (*REST, *StatusREST) {
 			return replicaset.MatchReplicaSet(label, field)
 		},
 		QualifiedResource:       api.Resource("replicasets"),
+		DeleteCollectionWorkers: opts.DeleteCollectionWorkers,
 
 		// Used to validate ReplicaSet creation
 		CreateStrategy: replicaset.Strategy,

pkg/registry/replicaset/etcd/etcd_test.go

@@ -32,7 +32,7 @@ import (
 
 func newStorage(t *testing.T) (*ReplicaSetStorage, *etcdtesting.EtcdTestServer) {
 	etcdStorage, server := registrytest.NewEtcdStorage(t, "extensions")
-	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage}
+	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage, 1}
 	replicaSetStorage := NewStorage(restOptions)
 	return &replicaSetStorage, server
 }

pkg/registry/resourcequota/etcd/etcd.go

@@ -55,6 +55,7 @@ func NewREST(opts generic.RESTOptions) (*REST, *StatusREST) {
 			return resourcequota.MatchResourceQuota(label, field)
 		},
 		QualifiedResource:       api.Resource("resourcequotas"),
+		DeleteCollectionWorkers: opts.DeleteCollectionWorkers,
 
 		CreateStrategy:      resourcequota.Strategy,
 		UpdateStrategy:      resourcequota.Strategy,

pkg/registry/resourcequota/etcd/etcd_test.go

@@ -32,7 +32,7 @@ import (
 
 func newStorage(t *testing.T) (*REST, *StatusREST, *etcdtesting.EtcdTestServer) {
 	etcdStorage, server := registrytest.NewEtcdStorage(t, "")
-	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage}
+	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage, 1}
 	resourceQuotaStorage, statusStorage := NewREST(restOptions)
 	return resourceQuotaStorage, statusStorage, server
 }

pkg/registry/secret/etcd/etcd.go

@@ -55,6 +55,7 @@ func NewREST(opts generic.RESTOptions) *REST {
 			return secret.Matcher(label, field)
 		},
 		QualifiedResource:       api.Resource("secrets"),
+		DeleteCollectionWorkers: opts.DeleteCollectionWorkers,
 
 		CreateStrategy: secret.Strategy,
 		UpdateStrategy: secret.Strategy,

pkg/registry/secret/etcd/etcd_test.go

@@ -30,7 +30,7 @@ import (
 
 func newStorage(t *testing.T) (*REST, *etcdtesting.EtcdTestServer) {
 	etcdStorage, server := registrytest.NewEtcdStorage(t, "")
-	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage}
+	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage, 1}
 	return NewREST(restOptions), server
 }
 

pkg/registry/service/etcd/etcd.go

@@ -55,6 +55,7 @@ func NewREST(opts generic.RESTOptions) (*REST, *StatusREST) {
 			return service.MatchServices(label, field)
 		},
 		QualifiedResource:       api.Resource("services"),
+		DeleteCollectionWorkers: opts.DeleteCollectionWorkers,
 
 		CreateStrategy: service.Strategy,
 		UpdateStrategy: service.Strategy,

pkg/registry/service/etcd/etcd_test.go

@@ -31,7 +31,7 @@ import (
 
 func newStorage(t *testing.T) (*REST, *StatusREST, *etcdtesting.EtcdTestServer) {
 	etcdStorage, server := registrytest.NewEtcdStorage(t, "")
-	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage}
+	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage, 1}
 	serviceStorage, statusStorage := NewREST(restOptions)
 	return serviceStorage, statusStorage, server
 }

pkg/registry/serviceaccount/etcd/etcd.go

@@ -55,6 +55,7 @@ func NewREST(opts generic.RESTOptions) *REST {
 			return serviceaccount.Matcher(label, field)
 		},
 		QualifiedResource:       api.Resource("serviceaccounts"),
+		DeleteCollectionWorkers: opts.DeleteCollectionWorkers,
 
 		CreateStrategy:      serviceaccount.Strategy,
 		UpdateStrategy:      serviceaccount.Strategy,

pkg/registry/serviceaccount/etcd/etcd_test.go

@@ -30,7 +30,7 @@ import (
 
 func newStorage(t *testing.T) (*REST, *etcdtesting.EtcdTestServer) {
 	etcdStorage, server := registrytest.NewEtcdStorage(t, "")
-	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage}
+	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage, 1}
 	return NewREST(restOptions), server
 }
 

pkg/registry/thirdpartyresource/etcd/etcd.go

@@ -55,6 +55,7 @@ func NewREST(opts generic.RESTOptions) *REST {
 			return thirdpartyresource.Matcher(label, field)
 		},
 		QualifiedResource:       extensions.Resource("thirdpartyresources"),
+		DeleteCollectionWorkers: opts.DeleteCollectionWorkers,
 		CreateStrategy:          thirdpartyresource.Strategy,
 		UpdateStrategy:          thirdpartyresource.Strategy,
 

pkg/registry/thirdpartyresource/etcd/etcd_test.go

@@ -33,7 +33,7 @@ import (
 
 func newStorage(t *testing.T) (*REST, *etcdtesting.EtcdTestServer) {
 	etcdStorage, server := registrytest.NewEtcdStorage(t, extensions.GroupName)
-	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage}
+	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage, 1}
 	return NewREST(restOptions), server
 }
 

pkg/registry/thirdpartyresourcedata/etcd/etcd.go

@@ -58,6 +58,7 @@ func NewREST(opts generic.RESTOptions, group, kind string) *REST {
 			return thirdpartyresourcedata.Matcher(label, field)
 		},
 		QualifiedResource:       extensions.Resource("thirdpartyresourcedatas"),
+		DeleteCollectionWorkers: opts.DeleteCollectionWorkers,
 		CreateStrategy:          thirdpartyresourcedata.Strategy,
 		UpdateStrategy:          thirdpartyresourcedata.Strategy,
 

pkg/registry/thirdpartyresourcedata/etcd/etcd_test.go

@@ -33,7 +33,7 @@ import (
 
 func newStorage(t *testing.T) (*REST, *etcdtesting.EtcdTestServer) {
 	etcdStorage, server := registrytest.NewEtcdStorage(t, extensions.GroupName)
-	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage}
+	restOptions := generic.RESTOptions{etcdStorage, generic.UndecoratedStorage, 1}
 	return NewREST(restOptions, "foo", "bar"), server
 }