diff --git a/cmd/kube-aggregator/pkg/apiserver/BUILD b/cmd/kube-aggregator/pkg/apiserver/BUILD index 324da4336df..07bd1e60aef 100644 --- a/cmd/kube-aggregator/pkg/apiserver/BUILD +++ b/cmd/kube-aggregator/pkg/apiserver/BUILD @@ -62,7 +62,6 @@ go_library( "//pkg/genericapiserver/registry/generic/rest:go_default_library", "//pkg/genericapiserver/registry/rest:go_default_library", "//pkg/genericapiserver/server:go_default_library", - "//pkg/util/workqueue:go_default_library", "//pkg/version:go_default_library", "//vendor:github.com/golang/glog", "//vendor:k8s.io/apimachinery/pkg/api/errors", @@ -77,6 +76,7 @@ go_library( "//vendor:k8s.io/client-go/rest", "//vendor:k8s.io/client-go/tools/cache", "//vendor:k8s.io/client-go/transport", + "//vendor:k8s.io/client-go/util/workqueue", ], ) diff --git a/cmd/kube-aggregator/pkg/apiserver/apiservice_controller.go b/cmd/kube-aggregator/pkg/apiserver/apiservice_controller.go index a836b5120d0..829f4cf630e 100644 --- a/cmd/kube-aggregator/pkg/apiserver/apiservice_controller.go +++ b/cmd/kube-aggregator/pkg/apiserver/apiservice_controller.go @@ -26,8 +26,8 @@ import ( utilruntime "k8s.io/apimachinery/pkg/util/runtime" "k8s.io/apimachinery/pkg/util/wait" "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" "k8s.io/kubernetes/pkg/controller" - "k8s.io/kubernetes/pkg/util/workqueue" "k8s.io/kubernetes/cmd/kube-aggregator/pkg/apis/apiregistration" informers "k8s.io/kubernetes/cmd/kube-aggregator/pkg/client/informers/apiregistration/internalversion" diff --git a/federation/pkg/federation-controller/deployment/BUILD b/federation/pkg/federation-controller/deployment/BUILD index acf72a0032a..5af892295c4 100644 --- a/federation/pkg/federation-controller/deployment/BUILD +++ b/federation/pkg/federation-controller/deployment/BUILD @@ -27,7 +27,6 @@ go_library( "//pkg/client/clientset_generated/clientset:go_default_library", "//pkg/client/record:go_default_library", "//pkg/controller:go_default_library", - "//pkg/util/workqueue:go_default_library", "//vendor:github.com/golang/glog", "//vendor:k8s.io/apimachinery/pkg/api/errors", "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1", @@ -36,6 +35,7 @@ go_library( "//vendor:k8s.io/apimachinery/pkg/watch", "//vendor:k8s.io/client-go/tools/cache", "//vendor:k8s.io/client-go/util/flowcontrol", + "//vendor:k8s.io/client-go/util/workqueue", ], ) diff --git a/federation/pkg/federation-controller/deployment/deploymentcontroller.go b/federation/pkg/federation-controller/deployment/deploymentcontroller.go index 79ea906deff..e354b8e4364 100644 --- a/federation/pkg/federation-controller/deployment/deploymentcontroller.go +++ b/federation/pkg/federation-controller/deployment/deploymentcontroller.go @@ -32,6 +32,7 @@ import ( "k8s.io/apimachinery/pkg/watch" "k8s.io/client-go/tools/cache" "k8s.io/client-go/util/flowcontrol" + "k8s.io/client-go/util/workqueue" fed "k8s.io/kubernetes/federation/apis/federation" fedv1 "k8s.io/kubernetes/federation/apis/federation/v1beta1" fedclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_clientset" @@ -46,7 +47,6 @@ import ( kubeclientset "k8s.io/kubernetes/pkg/client/clientset_generated/clientset" "k8s.io/kubernetes/pkg/client/record" "k8s.io/kubernetes/pkg/controller" - "k8s.io/kubernetes/pkg/util/workqueue" ) const ( diff --git a/federation/pkg/federation-controller/replicaset/BUILD b/federation/pkg/federation-controller/replicaset/BUILD index 84cd8b86524..925a8c2c11f 100644 --- a/federation/pkg/federation-controller/replicaset/BUILD +++ b/federation/pkg/federation-controller/replicaset/BUILD @@ -28,7 +28,6 @@ go_library( "//pkg/client/legacylisters:go_default_library", "//pkg/client/record:go_default_library", "//pkg/controller:go_default_library", - "//pkg/util/workqueue:go_default_library", "//vendor:github.com/golang/glog", "//vendor:k8s.io/apimachinery/pkg/api/errors", "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1", @@ -37,6 +36,7 @@ go_library( "//vendor:k8s.io/apimachinery/pkg/watch", "//vendor:k8s.io/client-go/tools/cache", "//vendor:k8s.io/client-go/util/flowcontrol", + "//vendor:k8s.io/client-go/util/workqueue", ], ) diff --git a/federation/pkg/federation-controller/replicaset/replicasetcontroller.go b/federation/pkg/federation-controller/replicaset/replicasetcontroller.go index 72a8f2a1f9a..2ae6a842bc3 100644 --- a/federation/pkg/federation-controller/replicaset/replicasetcontroller.go +++ b/federation/pkg/federation-controller/replicaset/replicasetcontroller.go @@ -32,6 +32,7 @@ import ( "k8s.io/apimachinery/pkg/watch" "k8s.io/client-go/tools/cache" "k8s.io/client-go/util/flowcontrol" + "k8s.io/client-go/util/workqueue" fed "k8s.io/kubernetes/federation/apis/federation" fedv1 "k8s.io/kubernetes/federation/apis/federation/v1beta1" fedclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_clientset" @@ -47,7 +48,6 @@ import ( "k8s.io/kubernetes/pkg/client/legacylisters" "k8s.io/kubernetes/pkg/client/record" "k8s.io/kubernetes/pkg/controller" - "k8s.io/kubernetes/pkg/util/workqueue" ) const ( diff --git a/federation/pkg/federation-controller/service/BUILD b/federation/pkg/federation-controller/service/BUILD index 4c3b9ad574c..212f4ebec0c 100644 --- a/federation/pkg/federation-controller/service/BUILD +++ b/federation/pkg/federation-controller/service/BUILD @@ -33,7 +33,6 @@ go_library( "//pkg/client/legacylisters:go_default_library", "//pkg/client/record:go_default_library", "//pkg/controller:go_default_library", - "//pkg/util/workqueue:go_default_library", "//vendor:github.com/golang/glog", "//vendor:k8s.io/apimachinery/pkg/api/errors", "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1", @@ -45,6 +44,7 @@ go_library( "//vendor:k8s.io/apimachinery/pkg/watch", "//vendor:k8s.io/client-go/rest", "//vendor:k8s.io/client-go/tools/cache", + "//vendor:k8s.io/client-go/util/workqueue", ], ) diff --git a/federation/pkg/federation-controller/service/cluster_helper.go b/federation/pkg/federation-controller/service/cluster_helper.go index 0d94a5a712e..180f2e23667 100644 --- a/federation/pkg/federation-controller/service/cluster_helper.go +++ b/federation/pkg/federation-controller/service/cluster_helper.go @@ -25,11 +25,11 @@ import ( "k8s.io/apimachinery/pkg/watch" restclient "k8s.io/client-go/rest" cache "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" v1beta1 "k8s.io/kubernetes/federation/apis/federation/v1beta1" v1 "k8s.io/kubernetes/pkg/api/v1" kubeclientset "k8s.io/kubernetes/pkg/client/clientset_generated/clientset" "k8s.io/kubernetes/pkg/client/legacylisters" - "k8s.io/kubernetes/pkg/util/workqueue" "reflect" diff --git a/federation/pkg/federation-controller/service/servicecontroller.go b/federation/pkg/federation-controller/service/servicecontroller.go index 31a78091408..04b33fd8440 100644 --- a/federation/pkg/federation-controller/service/servicecontroller.go +++ b/federation/pkg/federation-controller/service/servicecontroller.go @@ -33,6 +33,7 @@ import ( "k8s.io/apimachinery/pkg/util/wait" "k8s.io/apimachinery/pkg/watch" cache "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" v1beta1 "k8s.io/kubernetes/federation/apis/federation/v1beta1" federationcache "k8s.io/kubernetes/federation/client/cache" fedclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_clientset" @@ -46,7 +47,6 @@ import ( "k8s.io/kubernetes/pkg/client/legacylisters" "k8s.io/kubernetes/pkg/client/record" "k8s.io/kubernetes/pkg/controller" - "k8s.io/kubernetes/pkg/util/workqueue" ) const ( diff --git a/hack/.linted_packages b/hack/.linted_packages index 9f20d9e383e..99d154ddffb 100644 --- a/hack/.linted_packages +++ b/hack/.linted_packages @@ -237,7 +237,6 @@ pkg/util/validation pkg/util/validation/field pkg/util/version pkg/util/wait -pkg/util/workqueue pkg/util/yaml pkg/version/prometheus pkg/volume @@ -299,6 +298,7 @@ staging/src/k8s.io/client-go/tools/auth staging/src/k8s.io/client-go/tools/metrics staging/src/k8s.io/client-go/util/cert staging/src/k8s.io/client-go/util/homedir +staging/src/k8s.io/client-go/util/workqueue test/e2e/perftype test/e2e_node/runner/local test/images/clusterapi-tester diff --git a/pkg/controller/certificates/BUILD b/pkg/controller/certificates/BUILD index 125579f1768..9729602e39e 100644 --- a/pkg/controller/certificates/BUILD +++ b/pkg/controller/certificates/BUILD @@ -26,7 +26,6 @@ go_library( "//pkg/client/legacylisters:go_default_library", "//pkg/client/record:go_default_library", "//pkg/controller:go_default_library", - "//pkg/util/workqueue:go_default_library", "//vendor:github.com/cloudflare/cfssl/config", "//vendor:github.com/cloudflare/cfssl/helpers", "//vendor:github.com/cloudflare/cfssl/signer", @@ -38,6 +37,7 @@ go_library( "//vendor:k8s.io/apimachinery/pkg/util/wait", "//vendor:k8s.io/apimachinery/pkg/watch", "//vendor:k8s.io/client-go/tools/cache", + "//vendor:k8s.io/client-go/util/workqueue", ], ) diff --git a/pkg/controller/certificates/certificate_controller.go b/pkg/controller/certificates/certificate_controller.go index 750728c9b17..5a3fb0cccc6 100644 --- a/pkg/controller/certificates/certificate_controller.go +++ b/pkg/controller/certificates/certificate_controller.go @@ -26,13 +26,13 @@ import ( "k8s.io/apimachinery/pkg/util/wait" "k8s.io/apimachinery/pkg/watch" "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" certificates "k8s.io/kubernetes/pkg/apis/certificates/v1beta1" "k8s.io/kubernetes/pkg/client/clientset_generated/clientset" v1core "k8s.io/kubernetes/pkg/client/clientset_generated/clientset/typed/core/v1" "k8s.io/kubernetes/pkg/client/legacylisters" "k8s.io/kubernetes/pkg/client/record" "k8s.io/kubernetes/pkg/controller" - "k8s.io/kubernetes/pkg/util/workqueue" "github.com/golang/glog" ) diff --git a/pkg/controller/daemon/BUILD b/pkg/controller/daemon/BUILD index 2fd117ce7d9..229a5df236f 100644 --- a/pkg/controller/daemon/BUILD +++ b/pkg/controller/daemon/BUILD @@ -27,7 +27,6 @@ go_library( "//pkg/controller:go_default_library", "//pkg/controller/informers:go_default_library", "//pkg/util/metrics:go_default_library", - "//pkg/util/workqueue:go_default_library", "//plugin/pkg/scheduler/algorithm/predicates:go_default_library", "//plugin/pkg/scheduler/schedulercache:go_default_library", "//vendor:github.com/golang/glog", @@ -37,6 +36,7 @@ go_library( "//vendor:k8s.io/apimachinery/pkg/util/runtime", "//vendor:k8s.io/apimachinery/pkg/util/wait", "//vendor:k8s.io/client-go/tools/cache", + "//vendor:k8s.io/client-go/util/workqueue", ], ) diff --git a/pkg/controller/daemon/daemoncontroller.go b/pkg/controller/daemon/daemoncontroller.go index 75aec2a3092..9fdbeb89b05 100644 --- a/pkg/controller/daemon/daemoncontroller.go +++ b/pkg/controller/daemon/daemoncontroller.go @@ -29,6 +29,7 @@ import ( utilruntime "k8s.io/apimachinery/pkg/util/runtime" "k8s.io/apimachinery/pkg/util/wait" "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" "k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api/v1" extensions "k8s.io/kubernetes/pkg/apis/extensions/v1beta1" @@ -40,7 +41,6 @@ import ( "k8s.io/kubernetes/pkg/controller" "k8s.io/kubernetes/pkg/controller/informers" "k8s.io/kubernetes/pkg/util/metrics" - "k8s.io/kubernetes/pkg/util/workqueue" "k8s.io/kubernetes/plugin/pkg/scheduler/algorithm/predicates" "k8s.io/kubernetes/plugin/pkg/scheduler/schedulercache" diff --git a/pkg/controller/deployment/BUILD b/pkg/controller/deployment/BUILD index 1a1b371d131..020e7ea3272 100644 --- a/pkg/controller/deployment/BUILD +++ b/pkg/controller/deployment/BUILD @@ -32,7 +32,6 @@ go_library( "//pkg/controller/informers:go_default_library", "//pkg/util/labels:go_default_library", "//pkg/util/metrics:go_default_library", - "//pkg/util/workqueue:go_default_library", "//vendor:github.com/golang/glog", "//vendor:k8s.io/apimachinery/pkg/api/equality", "//vendor:k8s.io/apimachinery/pkg/api/errors", @@ -44,6 +43,7 @@ go_library( "//vendor:k8s.io/apimachinery/pkg/util/wait", "//vendor:k8s.io/client-go/tools/cache", "//vendor:k8s.io/client-go/util/integer", + "//vendor:k8s.io/client-go/util/workqueue", ], ) diff --git a/pkg/controller/deployment/deployment_controller.go b/pkg/controller/deployment/deployment_controller.go index f3e3102be78..c8b96a9cc0f 100644 --- a/pkg/controller/deployment/deployment_controller.go +++ b/pkg/controller/deployment/deployment_controller.go @@ -36,6 +36,7 @@ import ( utilruntime "k8s.io/apimachinery/pkg/util/runtime" "k8s.io/apimachinery/pkg/util/wait" "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" "k8s.io/kubernetes/pkg/api/v1" extensions "k8s.io/kubernetes/pkg/apis/extensions/v1beta1" "k8s.io/kubernetes/pkg/client/clientset_generated/clientset" @@ -46,7 +47,6 @@ import ( "k8s.io/kubernetes/pkg/controller/deployment/util" "k8s.io/kubernetes/pkg/controller/informers" "k8s.io/kubernetes/pkg/util/metrics" - "k8s.io/kubernetes/pkg/util/workqueue" ) const ( diff --git a/pkg/controller/disruption/BUILD b/pkg/controller/disruption/BUILD index a17ad819176..f423b3b773b 100644 --- a/pkg/controller/disruption/BUILD +++ b/pkg/controller/disruption/BUILD @@ -25,7 +25,6 @@ go_library( "//pkg/client/record:go_default_library", "//pkg/controller:go_default_library", "//pkg/util/intstr:go_default_library", - "//pkg/util/workqueue:go_default_library", "//vendor:github.com/golang/glog", "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1", "//vendor:k8s.io/apimachinery/pkg/runtime", @@ -34,6 +33,7 @@ go_library( "//vendor:k8s.io/apimachinery/pkg/util/wait", "//vendor:k8s.io/apimachinery/pkg/watch", "//vendor:k8s.io/client-go/tools/cache", + "//vendor:k8s.io/client-go/util/workqueue", ], ) @@ -52,10 +52,10 @@ go_test( "//pkg/client/record:go_default_library", "//pkg/controller:go_default_library", "//pkg/util/intstr:go_default_library", - "//pkg/util/workqueue:go_default_library", "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1", "//vendor:k8s.io/apimachinery/pkg/util/uuid", "//vendor:k8s.io/client-go/tools/cache", + "//vendor:k8s.io/client-go/util/workqueue", ], ) diff --git a/pkg/controller/disruption/disruption.go b/pkg/controller/disruption/disruption.go index e9c5a67e968..587204ca42e 100644 --- a/pkg/controller/disruption/disruption.go +++ b/pkg/controller/disruption/disruption.go @@ -28,6 +28,7 @@ import ( "k8s.io/apimachinery/pkg/util/wait" "k8s.io/apimachinery/pkg/watch" "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" "k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api/v1" apps "k8s.io/kubernetes/pkg/apis/apps/v1beta1" @@ -40,7 +41,6 @@ import ( "k8s.io/kubernetes/pkg/client/record" "k8s.io/kubernetes/pkg/controller" "k8s.io/kubernetes/pkg/util/intstr" - "k8s.io/kubernetes/pkg/util/workqueue" "github.com/golang/glog" ) diff --git a/pkg/controller/disruption/disruption_test.go b/pkg/controller/disruption/disruption_test.go index 0fad17dc606..d9b3fd446ba 100644 --- a/pkg/controller/disruption/disruption_test.go +++ b/pkg/controller/disruption/disruption_test.go @@ -26,6 +26,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/util/uuid" "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" "k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api/v1" apps "k8s.io/kubernetes/pkg/apis/apps/v1beta1" @@ -35,7 +36,6 @@ import ( "k8s.io/kubernetes/pkg/client/record" "k8s.io/kubernetes/pkg/controller" "k8s.io/kubernetes/pkg/util/intstr" - "k8s.io/kubernetes/pkg/util/workqueue" ) type pdbStates map[string]policy.PodDisruptionBudget diff --git a/pkg/controller/endpoint/BUILD b/pkg/controller/endpoint/BUILD index 9ae14d2d471..97239db6d7d 100644 --- a/pkg/controller/endpoint/BUILD +++ b/pkg/controller/endpoint/BUILD @@ -24,7 +24,6 @@ go_library( "//pkg/controller:go_default_library", "//pkg/controller/informers:go_default_library", "//pkg/util/metrics:go_default_library", - "//pkg/util/workqueue:go_default_library", "//vendor:github.com/golang/glog", "//vendor:k8s.io/apimachinery/pkg/api/errors", "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1", @@ -35,6 +34,7 @@ go_library( "//vendor:k8s.io/apimachinery/pkg/util/wait", "//vendor:k8s.io/apimachinery/pkg/watch", "//vendor:k8s.io/client-go/tools/cache", + "//vendor:k8s.io/client-go/util/workqueue", ], ) diff --git a/pkg/controller/endpoint/endpoints_controller.go b/pkg/controller/endpoint/endpoints_controller.go index b96c2890856..be350f25e9d 100644 --- a/pkg/controller/endpoint/endpoints_controller.go +++ b/pkg/controller/endpoint/endpoints_controller.go @@ -31,6 +31,7 @@ import ( "k8s.io/apimachinery/pkg/util/wait" "k8s.io/apimachinery/pkg/watch" "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" "k8s.io/kubernetes/pkg/api/v1" "k8s.io/kubernetes/pkg/api/v1/endpoints" podutil "k8s.io/kubernetes/pkg/api/v1/pod" @@ -40,7 +41,6 @@ import ( "k8s.io/kubernetes/pkg/controller" "k8s.io/kubernetes/pkg/controller/informers" "k8s.io/kubernetes/pkg/util/metrics" - "k8s.io/kubernetes/pkg/util/workqueue" "github.com/golang/glog" ) diff --git a/pkg/controller/garbagecollector/BUILD b/pkg/controller/garbagecollector/BUILD index 1fa7fbab49f..3430b1cfc37 100644 --- a/pkg/controller/garbagecollector/BUILD +++ b/pkg/controller/garbagecollector/BUILD @@ -21,7 +21,6 @@ go_library( "//pkg/api/v1:go_default_library", "//pkg/controller/garbagecollector/metaonly:go_default_library", "//pkg/util/metrics:go_default_library", - "//pkg/util/workqueue:go_default_library", "//vendor:github.com/golang/glog", "//vendor:github.com/golang/groupcache/lru", "//vendor:github.com/prometheus/client_golang/prometheus", @@ -41,6 +40,7 @@ go_library( "//vendor:k8s.io/client-go/kubernetes", "//vendor:k8s.io/client-go/tools/cache", "//vendor:k8s.io/client-go/util/clock", + "//vendor:k8s.io/client-go/util/workqueue", ], ) @@ -54,7 +54,6 @@ go_test( "//pkg/api/install:go_default_library", "//pkg/api/v1:go_default_library", "//pkg/controller/garbagecollector/metaonly:go_default_library", - "//pkg/util/workqueue:go_default_library", "//vendor:github.com/stretchr/testify/assert", "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1", "//vendor:k8s.io/apimachinery/pkg/runtime/schema", @@ -65,6 +64,7 @@ go_test( "//vendor:k8s.io/client-go/dynamic", "//vendor:k8s.io/client-go/rest", "//vendor:k8s.io/client-go/util/clock", + "//vendor:k8s.io/client-go/util/workqueue", ], ) diff --git a/pkg/controller/garbagecollector/garbagecollector.go b/pkg/controller/garbagecollector/garbagecollector.go index 6a83a1be8a2..c272e30b249 100644 --- a/pkg/controller/garbagecollector/garbagecollector.go +++ b/pkg/controller/garbagecollector/garbagecollector.go @@ -38,9 +38,9 @@ import ( "k8s.io/client-go/dynamic" "k8s.io/client-go/tools/cache" "k8s.io/client-go/util/clock" + "k8s.io/client-go/util/workqueue" "k8s.io/kubernetes/pkg/api/v1" "k8s.io/kubernetes/pkg/controller/garbagecollector/metaonly" - "k8s.io/kubernetes/pkg/util/workqueue" // import known versions _ "k8s.io/client-go/kubernetes" diff --git a/pkg/controller/garbagecollector/garbagecollector_test.go b/pkg/controller/garbagecollector/garbagecollector_test.go index 3655af8f6e6..7f741dddc27 100644 --- a/pkg/controller/garbagecollector/garbagecollector_test.go +++ b/pkg/controller/garbagecollector/garbagecollector_test.go @@ -36,10 +36,10 @@ import ( "k8s.io/client-go/dynamic" restclient "k8s.io/client-go/rest" "k8s.io/client-go/util/clock" + "k8s.io/client-go/util/workqueue" "k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api/v1" "k8s.io/kubernetes/pkg/controller/garbagecollector/metaonly" - "k8s.io/kubernetes/pkg/util/workqueue" ) func TestNewGarbageCollector(t *testing.T) { diff --git a/pkg/controller/job/BUILD b/pkg/controller/job/BUILD index e85d0d2da3a..9956fd115df 100644 --- a/pkg/controller/job/BUILD +++ b/pkg/controller/job/BUILD @@ -27,13 +27,13 @@ go_library( "//pkg/controller:go_default_library", "//pkg/controller/informers:go_default_library", "//pkg/util/metrics:go_default_library", - "//pkg/util/workqueue:go_default_library", "//vendor:github.com/golang/glog", "//vendor:k8s.io/apimachinery/pkg/api/errors", "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1", "//vendor:k8s.io/apimachinery/pkg/util/runtime", "//vendor:k8s.io/apimachinery/pkg/util/wait", "//vendor:k8s.io/client-go/tools/cache", + "//vendor:k8s.io/client-go/util/workqueue", ], ) diff --git a/pkg/controller/job/jobcontroller.go b/pkg/controller/job/jobcontroller.go index 81218639a23..cc45261cd3d 100644 --- a/pkg/controller/job/jobcontroller.go +++ b/pkg/controller/job/jobcontroller.go @@ -28,6 +28,7 @@ import ( utilruntime "k8s.io/apimachinery/pkg/util/runtime" "k8s.io/apimachinery/pkg/util/wait" "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" "k8s.io/kubernetes/pkg/api/v1" batch "k8s.io/kubernetes/pkg/apis/batch/v1" "k8s.io/kubernetes/pkg/client/clientset_generated/clientset" @@ -38,7 +39,6 @@ import ( "k8s.io/kubernetes/pkg/controller" "k8s.io/kubernetes/pkg/controller/informers" "k8s.io/kubernetes/pkg/util/metrics" - "k8s.io/kubernetes/pkg/util/workqueue" "github.com/golang/glog" ) diff --git a/pkg/controller/namespace/BUILD b/pkg/controller/namespace/BUILD index f68aeafb456..5fc46f13803 100644 --- a/pkg/controller/namespace/BUILD +++ b/pkg/controller/namespace/BUILD @@ -21,7 +21,6 @@ go_library( "//pkg/client/clientset_generated/clientset:go_default_library", "//pkg/controller:go_default_library", "//pkg/util/metrics:go_default_library", - "//pkg/util/workqueue:go_default_library", "//vendor:github.com/golang/glog", "//vendor:k8s.io/apimachinery/pkg/api/errors", "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1", @@ -35,6 +34,7 @@ go_library( "//vendor:k8s.io/client-go/discovery", "//vendor:k8s.io/client-go/dynamic", "//vendor:k8s.io/client-go/tools/cache", + "//vendor:k8s.io/client-go/util/workqueue", ], ) diff --git a/pkg/controller/namespace/namespace_controller.go b/pkg/controller/namespace/namespace_controller.go index 1386cc229f4..d2b1d5c0527 100644 --- a/pkg/controller/namespace/namespace_controller.go +++ b/pkg/controller/namespace/namespace_controller.go @@ -28,11 +28,11 @@ import ( "k8s.io/apimachinery/pkg/watch" "k8s.io/client-go/dynamic" "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" "k8s.io/kubernetes/pkg/api/v1" "k8s.io/kubernetes/pkg/client/clientset_generated/clientset" "k8s.io/kubernetes/pkg/controller" "k8s.io/kubernetes/pkg/util/metrics" - "k8s.io/kubernetes/pkg/util/workqueue" "github.com/golang/glog" ) diff --git a/pkg/controller/replicaset/BUILD b/pkg/controller/replicaset/BUILD index 108f004eb5f..e841df4bfa6 100644 --- a/pkg/controller/replicaset/BUILD +++ b/pkg/controller/replicaset/BUILD @@ -28,7 +28,6 @@ go_library( "//pkg/controller:go_default_library", "//pkg/controller/informers:go_default_library", "//pkg/util/metrics:go_default_library", - "//pkg/util/workqueue:go_default_library", "//vendor:github.com/golang/glog", "//vendor:k8s.io/apimachinery/pkg/api/errors", "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1", @@ -38,6 +37,7 @@ go_library( "//vendor:k8s.io/apimachinery/pkg/util/runtime", "//vendor:k8s.io/apimachinery/pkg/util/wait", "//vendor:k8s.io/client-go/tools/cache", + "//vendor:k8s.io/client-go/util/workqueue", ], ) diff --git a/pkg/controller/replicaset/replica_set.go b/pkg/controller/replicaset/replica_set.go index 8151fdbafa3..8bd03b80ab2 100644 --- a/pkg/controller/replicaset/replica_set.go +++ b/pkg/controller/replicaset/replica_set.go @@ -34,6 +34,7 @@ import ( utilruntime "k8s.io/apimachinery/pkg/util/runtime" "k8s.io/apimachinery/pkg/util/wait" "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" "k8s.io/kubernetes/pkg/api/v1" "k8s.io/kubernetes/pkg/apis/extensions/v1beta1" extensions "k8s.io/kubernetes/pkg/apis/extensions/v1beta1" @@ -44,7 +45,6 @@ import ( "k8s.io/kubernetes/pkg/controller" "k8s.io/kubernetes/pkg/controller/informers" "k8s.io/kubernetes/pkg/util/metrics" - "k8s.io/kubernetes/pkg/util/workqueue" ) const ( diff --git a/pkg/controller/replication/BUILD b/pkg/controller/replication/BUILD index 86052a41e40..c6ed5c51306 100644 --- a/pkg/controller/replication/BUILD +++ b/pkg/controller/replication/BUILD @@ -25,7 +25,6 @@ go_library( "//pkg/controller:go_default_library", "//pkg/controller/informers:go_default_library", "//pkg/util/metrics:go_default_library", - "//pkg/util/workqueue:go_default_library", "//vendor:github.com/golang/glog", "//vendor:k8s.io/apimachinery/pkg/api/errors", "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1", @@ -36,6 +35,7 @@ go_library( "//vendor:k8s.io/apimachinery/pkg/util/wait", "//vendor:k8s.io/apiserver/pkg/util/trace", "//vendor:k8s.io/client-go/tools/cache", + "//vendor:k8s.io/client-go/util/workqueue", ], ) diff --git a/pkg/controller/replication/replication_controller.go b/pkg/controller/replication/replication_controller.go index 7c3e474a893..f4ad4a6de70 100644 --- a/pkg/controller/replication/replication_controller.go +++ b/pkg/controller/replication/replication_controller.go @@ -34,6 +34,7 @@ import ( "k8s.io/apimachinery/pkg/util/wait" utiltrace "k8s.io/apiserver/pkg/util/trace" "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" "k8s.io/kubernetes/pkg/api/v1" "k8s.io/kubernetes/pkg/client/clientset_generated/clientset" v1core "k8s.io/kubernetes/pkg/client/clientset_generated/clientset/typed/core/v1" @@ -42,7 +43,6 @@ import ( "k8s.io/kubernetes/pkg/controller" "k8s.io/kubernetes/pkg/controller/informers" "k8s.io/kubernetes/pkg/util/metrics" - "k8s.io/kubernetes/pkg/util/workqueue" ) const ( diff --git a/pkg/controller/resourcequota/BUILD b/pkg/controller/resourcequota/BUILD index d75ec6b1d82..7c3b311fa39 100644 --- a/pkg/controller/resourcequota/BUILD +++ b/pkg/controller/resourcequota/BUILD @@ -25,7 +25,6 @@ go_library( "//pkg/quota:go_default_library", "//pkg/quota/evaluator/core:go_default_library", "//pkg/util/metrics:go_default_library", - "//pkg/util/workqueue:go_default_library", "//vendor:github.com/golang/glog", "//vendor:k8s.io/apimachinery/pkg/api/equality", "//vendor:k8s.io/apimachinery/pkg/api/meta", @@ -36,6 +35,7 @@ go_library( "//vendor:k8s.io/apimachinery/pkg/util/wait", "//vendor:k8s.io/apimachinery/pkg/watch", "//vendor:k8s.io/client-go/tools/cache", + "//vendor:k8s.io/client-go/util/workqueue", ], ) diff --git a/pkg/controller/resourcequota/resource_quota_controller.go b/pkg/controller/resourcequota/resource_quota_controller.go index 55405d07f20..90f442ec8b5 100644 --- a/pkg/controller/resourcequota/resource_quota_controller.go +++ b/pkg/controller/resourcequota/resource_quota_controller.go @@ -29,13 +29,13 @@ import ( "k8s.io/apimachinery/pkg/util/wait" "k8s.io/apimachinery/pkg/watch" "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" "k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api/v1" "k8s.io/kubernetes/pkg/client/clientset_generated/clientset" "k8s.io/kubernetes/pkg/controller" "k8s.io/kubernetes/pkg/quota" "k8s.io/kubernetes/pkg/util/metrics" - "k8s.io/kubernetes/pkg/util/workqueue" ) // ResourceQuotaControllerOptions holds options for creating a quota controller diff --git a/pkg/controller/service/BUILD b/pkg/controller/service/BUILD index 1eacaf01de5..247d9298de4 100644 --- a/pkg/controller/service/BUILD +++ b/pkg/controller/service/BUILD @@ -24,7 +24,6 @@ go_library( "//pkg/cloudprovider:go_default_library", "//pkg/controller:go_default_library", "//pkg/util/metrics:go_default_library", - "//pkg/util/workqueue:go_default_library", "//vendor:github.com/golang/glog", "//vendor:k8s.io/apimachinery/pkg/api/errors", "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1", @@ -34,6 +33,7 @@ go_library( "//vendor:k8s.io/apimachinery/pkg/util/wait", "//vendor:k8s.io/apimachinery/pkg/watch", "//vendor:k8s.io/client-go/tools/cache", + "//vendor:k8s.io/client-go/util/workqueue", ], ) diff --git a/pkg/controller/service/servicecontroller.go b/pkg/controller/service/servicecontroller.go index 84d9061bc71..19ad5afcaf3 100644 --- a/pkg/controller/service/servicecontroller.go +++ b/pkg/controller/service/servicecontroller.go @@ -33,6 +33,7 @@ import ( "k8s.io/apimachinery/pkg/util/wait" "k8s.io/apimachinery/pkg/watch" "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" "k8s.io/kubernetes/pkg/api/v1" "k8s.io/kubernetes/pkg/client/clientset_generated/clientset" unversionedcore "k8s.io/kubernetes/pkg/client/clientset_generated/clientset/typed/core/v1" @@ -41,7 +42,6 @@ import ( "k8s.io/kubernetes/pkg/cloudprovider" "k8s.io/kubernetes/pkg/controller" "k8s.io/kubernetes/pkg/util/metrics" - "k8s.io/kubernetes/pkg/util/workqueue" ) const ( diff --git a/pkg/controller/serviceaccount/BUILD b/pkg/controller/serviceaccount/BUILD index 3462b81d2a5..1bed35c3f22 100644 --- a/pkg/controller/serviceaccount/BUILD +++ b/pkg/controller/serviceaccount/BUILD @@ -31,7 +31,6 @@ go_library( "//pkg/registry/core/serviceaccount/storage:go_default_library", "//pkg/serviceaccount:go_default_library", "//pkg/util/metrics:go_default_library", - "//pkg/util/workqueue:go_default_library", "//vendor:github.com/golang/glog", "//vendor:k8s.io/apimachinery/pkg/api/errors", "//vendor:k8s.io/apimachinery/pkg/api/meta", @@ -47,6 +46,7 @@ go_library( "//vendor:k8s.io/apiserver/pkg/endpoints/request", "//vendor:k8s.io/apiserver/pkg/storage/storagebackend", "//vendor:k8s.io/client-go/tools/cache", + "//vendor:k8s.io/client-go/util/workqueue", ], ) diff --git a/pkg/controller/serviceaccount/serviceaccounts_controller.go b/pkg/controller/serviceaccount/serviceaccounts_controller.go index 01c987367b9..c4fe139c6ab 100644 --- a/pkg/controller/serviceaccount/serviceaccounts_controller.go +++ b/pkg/controller/serviceaccount/serviceaccounts_controller.go @@ -28,12 +28,12 @@ import ( utilruntime "k8s.io/apimachinery/pkg/util/runtime" "k8s.io/apimachinery/pkg/util/wait" "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" "k8s.io/kubernetes/pkg/api/v1" "k8s.io/kubernetes/pkg/client/clientset_generated/clientset" "k8s.io/kubernetes/pkg/client/legacylisters" "k8s.io/kubernetes/pkg/controller/informers" "k8s.io/kubernetes/pkg/util/metrics" - "k8s.io/kubernetes/pkg/util/workqueue" ) // nameIndexFunc is an index function that indexes based on an object's name diff --git a/pkg/controller/serviceaccount/tokens_controller.go b/pkg/controller/serviceaccount/tokens_controller.go index 67cb8898254..f4d3a098746 100644 --- a/pkg/controller/serviceaccount/tokens_controller.go +++ b/pkg/controller/serviceaccount/tokens_controller.go @@ -33,6 +33,7 @@ import ( "k8s.io/apimachinery/pkg/util/wait" "k8s.io/apimachinery/pkg/watch" "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" "k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api/v1" "k8s.io/kubernetes/pkg/client/clientset_generated/clientset" @@ -40,7 +41,6 @@ import ( "k8s.io/kubernetes/pkg/registry/core/secret" "k8s.io/kubernetes/pkg/serviceaccount" "k8s.io/kubernetes/pkg/util/metrics" - "k8s.io/kubernetes/pkg/util/workqueue" ) // RemoveTokenBackoff is the recommended (empirical) retry interval for removing diff --git a/pkg/controller/statefulset/BUILD b/pkg/controller/statefulset/BUILD index 624fe143eaa..138575b239a 100644 --- a/pkg/controller/statefulset/BUILD +++ b/pkg/controller/statefulset/BUILD @@ -29,7 +29,6 @@ go_library( "//pkg/client/legacylisters:go_default_library", "//pkg/client/record:go_default_library", "//pkg/controller:go_default_library", - "//pkg/util/workqueue:go_default_library", "//vendor:github.com/golang/glog", "//vendor:gopkg.in/inf.v0", "//vendor:k8s.io/apimachinery/pkg/api/errors", @@ -43,6 +42,7 @@ go_library( "//vendor:k8s.io/apimachinery/pkg/util/wait", "//vendor:k8s.io/apimachinery/pkg/watch", "//vendor:k8s.io/client-go/tools/cache", + "//vendor:k8s.io/client-go/util/workqueue", ], ) diff --git a/pkg/controller/statefulset/stateful_set.go b/pkg/controller/statefulset/stateful_set.go index 2c74606bd36..fde4e6eb233 100644 --- a/pkg/controller/statefulset/stateful_set.go +++ b/pkg/controller/statefulset/stateful_set.go @@ -36,8 +36,8 @@ import ( utilruntime "k8s.io/apimachinery/pkg/util/runtime" "k8s.io/apimachinery/pkg/util/wait" "k8s.io/apimachinery/pkg/watch" + "k8s.io/client-go/util/workqueue" "k8s.io/kubernetes/pkg/controller" - "k8s.io/kubernetes/pkg/util/workqueue" "github.com/golang/glog" ) diff --git a/pkg/controller/volume/persistentvolume/BUILD b/pkg/controller/volume/persistentvolume/BUILD index 9a9996c4209..af978d75f87 100644 --- a/pkg/controller/volume/persistentvolume/BUILD +++ b/pkg/controller/volume/persistentvolume/BUILD @@ -30,7 +30,6 @@ go_library( "//pkg/util/goroutinemap:go_default_library", "//pkg/util/io:go_default_library", "//pkg/util/mount:go_default_library", - "//pkg/util/workqueue:go_default_library", "//pkg/volume:go_default_library", "//vendor:github.com/golang/glog", "//vendor:k8s.io/apimachinery/pkg/api/errors", @@ -42,6 +41,7 @@ go_library( "//vendor:k8s.io/apimachinery/pkg/util/wait", "//vendor:k8s.io/apimachinery/pkg/watch", "//vendor:k8s.io/client-go/tools/cache", + "//vendor:k8s.io/client-go/util/workqueue", ], ) diff --git a/pkg/controller/volume/persistentvolume/pv_controller.go b/pkg/controller/volume/persistentvolume/pv_controller.go index 20c15160d2a..0f9cf6126ce 100644 --- a/pkg/controller/volume/persistentvolume/pv_controller.go +++ b/pkg/controller/volume/persistentvolume/pv_controller.go @@ -24,6 +24,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" "k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api/v1" storage "k8s.io/kubernetes/pkg/apis/storage/v1beta1" @@ -32,7 +33,6 @@ import ( "k8s.io/kubernetes/pkg/client/record" "k8s.io/kubernetes/pkg/cloudprovider" "k8s.io/kubernetes/pkg/util/goroutinemap" - "k8s.io/kubernetes/pkg/util/workqueue" vol "k8s.io/kubernetes/pkg/volume" "github.com/golang/glog" diff --git a/pkg/controller/volume/persistentvolume/pv_controller_base.go b/pkg/controller/volume/persistentvolume/pv_controller_base.go index 47a004da2b5..13d78de5407 100644 --- a/pkg/controller/volume/persistentvolume/pv_controller_base.go +++ b/pkg/controller/volume/persistentvolume/pv_controller_base.go @@ -28,6 +28,7 @@ import ( "k8s.io/apimachinery/pkg/util/wait" "k8s.io/apimachinery/pkg/watch" "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" "k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api/v1" storage "k8s.io/kubernetes/pkg/apis/storage/v1beta1" @@ -37,7 +38,6 @@ import ( "k8s.io/kubernetes/pkg/cloudprovider" "k8s.io/kubernetes/pkg/controller" "k8s.io/kubernetes/pkg/util/goroutinemap" - "k8s.io/kubernetes/pkg/util/workqueue" vol "k8s.io/kubernetes/pkg/volume" "github.com/golang/glog" diff --git a/pkg/util/BUILD b/pkg/util/BUILD index 2e194816d8b..179c9614e37 100644 --- a/pkg/util/BUILD +++ b/pkg/util/BUILD @@ -98,7 +98,7 @@ filegroup( "//pkg/util/validation:all-srcs", "//pkg/util/version:all-srcs", "//pkg/util/wait:all-srcs", - "//pkg/util/workqueue:all-srcs", + "//pkg/util/workqueue/prometheus:all-srcs", "//pkg/util/yaml:all-srcs", ], tags = ["automanaged"], diff --git a/pkg/util/workqueue/BUILD b/pkg/util/workqueue/BUILD deleted file mode 100644 index 74cc0efbaa2..00000000000 --- a/pkg/util/workqueue/BUILD +++ /dev/null @@ -1,69 +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 = [ - "default_rate_limiters.go", - "delaying_queue.go", - "doc.go", - "metrics.go", - "parallelizer.go", - "queue.go", - "rate_limitting_queue.go", - "timed_queue.go", - ], - tags = ["automanaged"], - deps = [ - "//vendor:github.com/juju/ratelimit", - "//vendor:k8s.io/apimachinery/pkg/util/runtime", - "//vendor:k8s.io/client-go/util/clock", - ], -) - -go_test( - name = "go_default_test", - srcs = [ - "default_rate_limiters_test.go", - "delaying_queue_test.go", - "rate_limitting_queue_test.go", - "timed_queue_test.go", - ], - library = ":go_default_library", - tags = ["automanaged"], - deps = [ - "//pkg/api/v1:go_default_library", - "//vendor:k8s.io/apimachinery/pkg/util/wait", - "//vendor:k8s.io/client-go/util/clock", - ], -) - -go_test( - name = "go_default_xtest", - srcs = ["queue_test.go"], - tags = ["automanaged"], - deps = ["//pkg/util/workqueue:go_default_library"], -) - -filegroup( - name = "package-srcs", - srcs = glob(["**"]), - tags = ["automanaged"], - visibility = ["//visibility:private"], -) - -filegroup( - name = "all-srcs", - srcs = [ - ":package-srcs", - "//pkg/util/workqueue/prometheus:all-srcs", - ], - tags = ["automanaged"], -) diff --git a/pkg/util/workqueue/doc.go b/pkg/util/workqueue/doc.go deleted file mode 100644 index 7d7f733abdb..00000000000 --- a/pkg/util/workqueue/doc.go +++ /dev/null @@ -1,26 +0,0 @@ -/* -Copyright 2014 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 workqueue provides a simple queue that supports the following -// features: -// * Fair: items processed in the order in which they are added. -// * Stingy: a single item will not be processed multiple times concurrently, -// and if an item is added multiple times before it can be processed, it -// will only be processed once. -// * Multiple consumers and producers. In particular, it is allowed for an -// item to be reenqueued while it is being processed. -// * Shutdown notifications. -package workqueue // import "k8s.io/kubernetes/pkg/util/workqueue" diff --git a/pkg/util/workqueue/prometheus/BUILD b/pkg/util/workqueue/prometheus/BUILD index 36b3c9a3124..28ed9754db6 100644 --- a/pkg/util/workqueue/prometheus/BUILD +++ b/pkg/util/workqueue/prometheus/BUILD @@ -12,8 +12,8 @@ go_library( srcs = ["prometheus.go"], tags = ["automanaged"], deps = [ - "//pkg/util/workqueue:go_default_library", "//vendor:github.com/prometheus/client_golang/prometheus", + "//vendor:k8s.io/client-go/util/workqueue", ], ) diff --git a/pkg/util/workqueue/prometheus/prometheus.go b/pkg/util/workqueue/prometheus/prometheus.go index afd83419495..c0d2be761e4 100644 --- a/pkg/util/workqueue/prometheus/prometheus.go +++ b/pkg/util/workqueue/prometheus/prometheus.go @@ -17,7 +17,7 @@ limitations under the License. package prometheus import ( - "k8s.io/kubernetes/pkg/util/workqueue" + "k8s.io/client-go/util/workqueue" "github.com/prometheus/client_golang/prometheus" ) diff --git a/pkg/util/workqueue/queue_test.go b/pkg/util/workqueue/queue_test.go deleted file mode 100644 index 4b8553672e2..00000000000 --- a/pkg/util/workqueue/queue_test.go +++ /dev/null @@ -1,161 +0,0 @@ -/* -Copyright 2015 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 workqueue_test - -import ( - "sync" - "testing" - "time" - - "k8s.io/kubernetes/pkg/util/workqueue" -) - -func TestBasic(t *testing.T) { - // If something is seriously wrong this test will never complete. - q := workqueue.New() - - // Start producers - const producers = 50 - producerWG := sync.WaitGroup{} - producerWG.Add(producers) - for i := 0; i < producers; i++ { - go func(i int) { - defer producerWG.Done() - for j := 0; j < 50; j++ { - q.Add(i) - time.Sleep(time.Millisecond) - } - }(i) - } - - // Start consumers - const consumers = 10 - consumerWG := sync.WaitGroup{} - consumerWG.Add(consumers) - for i := 0; i < consumers; i++ { - go func(i int) { - defer consumerWG.Done() - for { - item, quit := q.Get() - if item == "added after shutdown!" { - t.Errorf("Got an item added after shutdown.") - } - if quit { - return - } - t.Logf("Worker %v: begin processing %v", i, item) - time.Sleep(3 * time.Millisecond) - t.Logf("Worker %v: done processing %v", i, item) - q.Done(item) - } - }(i) - } - - producerWG.Wait() - q.ShutDown() - q.Add("added after shutdown!") - consumerWG.Wait() -} - -func TestAddWhileProcessing(t *testing.T) { - q := workqueue.New() - - // Start producers - const producers = 50 - producerWG := sync.WaitGroup{} - producerWG.Add(producers) - for i := 0; i < producers; i++ { - go func(i int) { - defer producerWG.Done() - q.Add(i) - }(i) - } - - // Start consumers - const consumers = 10 - consumerWG := sync.WaitGroup{} - consumerWG.Add(consumers) - for i := 0; i < consumers; i++ { - go func(i int) { - defer consumerWG.Done() - // Every worker will re-add every item up to two times. - // This tests the dirty-while-processing case. - counters := map[interface{}]int{} - for { - item, quit := q.Get() - if quit { - return - } - counters[item]++ - if counters[item] < 2 { - q.Add(item) - } - q.Done(item) - } - }(i) - } - - producerWG.Wait() - q.ShutDown() - consumerWG.Wait() -} - -func TestLen(t *testing.T) { - q := workqueue.New() - q.Add("foo") - if e, a := 1, q.Len(); e != a { - t.Errorf("Expected %v, got %v", e, a) - } - q.Add("bar") - if e, a := 2, q.Len(); e != a { - t.Errorf("Expected %v, got %v", e, a) - } - q.Add("foo") // should not increase the queue length. - if e, a := 2, q.Len(); e != a { - t.Errorf("Expected %v, got %v", e, a) - } -} - -func TestReinsert(t *testing.T) { - q := workqueue.New() - q.Add("foo") - - // Start processing - i, _ := q.Get() - if i != "foo" { - t.Errorf("Expected %v, got %v", "foo", i) - } - - // Add it back while processing - q.Add(i) - - // Finish it up - q.Done(i) - - // It should be back on the queue - i, _ = q.Get() - if i != "foo" { - t.Errorf("Expected %v, got %v", "foo", i) - } - - // Finish that one up - q.Done(i) - - if a := q.Len(); a != 0 { - t.Errorf("Expected queue to be empty. Has %v items", a) - } -} diff --git a/pkg/util/workqueue/timed_queue_test.go b/pkg/util/workqueue/timed_queue_test.go deleted file mode 100644 index e9b3b73e322..00000000000 --- a/pkg/util/workqueue/timed_queue_test.go +++ /dev/null @@ -1,38 +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 workqueue - -import ( - "testing" - "time" - - "k8s.io/kubernetes/pkg/api/v1" -) - -func TestNoMemoryLeak(t *testing.T) { - timedQueue := NewTimedWorkQueue() - timedQueue.Add(&TimedWorkQueueItem{Object: &v1.Pod{}, StartTime: time.Time{}}) - item, _ := timedQueue.Get() - timedQueue.Add(item) - // The item should still be in the timedQueue. - timedQueue.Done(item) - item, _ = timedQueue.Get() - timedQueue.Done(item) - if len(timedQueue.Type.processing) != 0 { - t.Errorf("expect timedQueue.Type.processing to be empty!") - } -} diff --git a/plugin/pkg/admission/resourcequota/BUILD b/plugin/pkg/admission/resourcequota/BUILD index 776e59faff4..f532bcd83b4 100644 --- a/plugin/pkg/admission/resourcequota/BUILD +++ b/plugin/pkg/admission/resourcequota/BUILD @@ -24,7 +24,6 @@ go_library( "//pkg/quota:go_default_library", "//pkg/quota/install:go_default_library", "//pkg/storage/etcd:go_default_library", - "//pkg/util/workqueue:go_default_library", "//pkg/util/workqueue/prometheus:go_default_library", "//vendor:github.com/golang/glog", "//vendor:github.com/hashicorp/golang-lru", @@ -37,6 +36,7 @@ go_library( "//vendor:k8s.io/apimachinery/pkg/watch", "//vendor:k8s.io/apiserver/pkg/admission", "//vendor:k8s.io/client-go/tools/cache", + "//vendor:k8s.io/client-go/util/workqueue", ], ) diff --git a/plugin/pkg/admission/resourcequota/controller.go b/plugin/pkg/admission/resourcequota/controller.go index 79f355e4d9f..c41d6274754 100644 --- a/plugin/pkg/admission/resourcequota/controller.go +++ b/plugin/pkg/admission/resourcequota/controller.go @@ -30,9 +30,9 @@ import ( "k8s.io/apimachinery/pkg/util/sets" "k8s.io/apimachinery/pkg/util/wait" "k8s.io/apiserver/pkg/admission" + "k8s.io/client-go/util/workqueue" "k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/quota" - "k8s.io/kubernetes/pkg/util/workqueue" _ "k8s.io/kubernetes/pkg/util/workqueue/prometheus" // for workqueue metric registration ) diff --git a/plugin/pkg/scheduler/BUILD b/plugin/pkg/scheduler/BUILD index 70a2867a2b5..e037b495470 100644 --- a/plugin/pkg/scheduler/BUILD +++ b/plugin/pkg/scheduler/BUILD @@ -22,7 +22,6 @@ go_library( "//pkg/client/clientset_generated/clientset:go_default_library", "//pkg/client/record:go_default_library", "//pkg/util/hash:go_default_library", - "//pkg/util/workqueue:go_default_library", "//plugin/pkg/scheduler/algorithm:go_default_library", "//plugin/pkg/scheduler/algorithm/predicates:go_default_library", "//plugin/pkg/scheduler/api:go_default_library", @@ -39,6 +38,7 @@ go_library( "//vendor:k8s.io/apiserver/pkg/util/trace", "//vendor:k8s.io/client-go/rest", "//vendor:k8s.io/client-go/tools/cache", + "//vendor:k8s.io/client-go/util/workqueue", ], ) diff --git a/plugin/pkg/scheduler/algorithm/predicates/BUILD b/plugin/pkg/scheduler/algorithm/predicates/BUILD index 0c68819b94d..a95a40bb3c0 100644 --- a/plugin/pkg/scheduler/algorithm/predicates/BUILD +++ b/plugin/pkg/scheduler/algorithm/predicates/BUILD @@ -21,7 +21,6 @@ go_library( "//pkg/api/v1:go_default_library", "//pkg/client/legacylisters:go_default_library", "//pkg/kubelet/qos:go_default_library", - "//pkg/util/workqueue:go_default_library", "//plugin/pkg/scheduler/algorithm:go_default_library", "//plugin/pkg/scheduler/algorithm/priorities/util:go_default_library", "//plugin/pkg/scheduler/schedulercache:go_default_library", @@ -29,6 +28,7 @@ go_library( "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1", "//vendor:k8s.io/apimachinery/pkg/labels", "//vendor:k8s.io/apimachinery/pkg/util/runtime", + "//vendor:k8s.io/client-go/util/workqueue", ], ) diff --git a/plugin/pkg/scheduler/algorithm/predicates/predicates.go b/plugin/pkg/scheduler/algorithm/predicates/predicates.go index c443964cb53..644d531e027 100644 --- a/plugin/pkg/scheduler/algorithm/predicates/predicates.go +++ b/plugin/pkg/scheduler/algorithm/predicates/predicates.go @@ -28,10 +28,10 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" utilruntime "k8s.io/apimachinery/pkg/util/runtime" + "k8s.io/client-go/util/workqueue" "k8s.io/kubernetes/pkg/api/v1" "k8s.io/kubernetes/pkg/client/legacylisters" "k8s.io/kubernetes/pkg/kubelet/qos" - "k8s.io/kubernetes/pkg/util/workqueue" "k8s.io/kubernetes/plugin/pkg/scheduler/algorithm" priorityutil "k8s.io/kubernetes/plugin/pkg/scheduler/algorithm/priorities/util" "k8s.io/kubernetes/plugin/pkg/scheduler/schedulercache" diff --git a/plugin/pkg/scheduler/algorithm/priorities/BUILD b/plugin/pkg/scheduler/algorithm/priorities/BUILD index 0faeaf6e9ed..f3a18dd86b8 100644 --- a/plugin/pkg/scheduler/algorithm/priorities/BUILD +++ b/plugin/pkg/scheduler/algorithm/priorities/BUILD @@ -28,7 +28,6 @@ go_library( deps = [ "//pkg/api/v1:go_default_library", "//pkg/util/node:go_default_library", - "//pkg/util/workqueue:go_default_library", "//plugin/pkg/scheduler/algorithm:go_default_library", "//plugin/pkg/scheduler/algorithm/predicates:go_default_library", "//plugin/pkg/scheduler/algorithm/priorities/util:go_default_library", @@ -38,6 +37,7 @@ go_library( "//vendor:k8s.io/apimachinery/pkg/api/resource", "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1", "//vendor:k8s.io/apimachinery/pkg/labels", + "//vendor:k8s.io/client-go/util/workqueue", ], ) diff --git a/plugin/pkg/scheduler/algorithm/priorities/interpod_affinity.go b/plugin/pkg/scheduler/algorithm/priorities/interpod_affinity.go index 4fb806961de..7478d63e512 100644 --- a/plugin/pkg/scheduler/algorithm/priorities/interpod_affinity.go +++ b/plugin/pkg/scheduler/algorithm/priorities/interpod_affinity.go @@ -21,8 +21,8 @@ import ( "github.com/golang/glog" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/util/workqueue" "k8s.io/kubernetes/pkg/api/v1" - "k8s.io/kubernetes/pkg/util/workqueue" "k8s.io/kubernetes/plugin/pkg/scheduler/algorithm" "k8s.io/kubernetes/plugin/pkg/scheduler/algorithm/predicates" priorityutil "k8s.io/kubernetes/plugin/pkg/scheduler/algorithm/priorities/util" diff --git a/plugin/pkg/scheduler/algorithm/priorities/selector_spreading.go b/plugin/pkg/scheduler/algorithm/priorities/selector_spreading.go index d1b6a052583..687ca331351 100644 --- a/plugin/pkg/scheduler/algorithm/priorities/selector_spreading.go +++ b/plugin/pkg/scheduler/algorithm/priorities/selector_spreading.go @@ -22,9 +22,9 @@ import ( "github.com/golang/glog" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" + "k8s.io/client-go/util/workqueue" "k8s.io/kubernetes/pkg/api/v1" utilnode "k8s.io/kubernetes/pkg/util/node" - "k8s.io/kubernetes/pkg/util/workqueue" "k8s.io/kubernetes/plugin/pkg/scheduler/algorithm" schedulerapi "k8s.io/kubernetes/plugin/pkg/scheduler/api" "k8s.io/kubernetes/plugin/pkg/scheduler/schedulercache" diff --git a/plugin/pkg/scheduler/generic_scheduler.go b/plugin/pkg/scheduler/generic_scheduler.go index 6d8ab4058ab..a03624c7306 100644 --- a/plugin/pkg/scheduler/generic_scheduler.go +++ b/plugin/pkg/scheduler/generic_scheduler.go @@ -27,8 +27,8 @@ import ( "github.com/golang/glog" "k8s.io/apimachinery/pkg/util/errors" utiltrace "k8s.io/apiserver/pkg/util/trace" + "k8s.io/client-go/util/workqueue" "k8s.io/kubernetes/pkg/api/v1" - "k8s.io/kubernetes/pkg/util/workqueue" "k8s.io/kubernetes/plugin/pkg/scheduler/algorithm" "k8s.io/kubernetes/plugin/pkg/scheduler/algorithm/predicates" schedulerapi "k8s.io/kubernetes/plugin/pkg/scheduler/api" diff --git a/staging/copy.sh b/staging/copy.sh index a52a06d8c5b..7dcfa534480 100755 --- a/staging/copy.sh +++ b/staging/copy.sh @@ -78,7 +78,7 @@ save "tools/cache" save "tools/clientcmd" save "tools/metrics" save "transport" -save "pkg/third_party" +save "third_party" save "plugin" save "util" @@ -101,9 +101,6 @@ mkcp "/pkg/client/record" "/pkg/client" mkcp "/pkg/client/unversioned/portforward" "/pkg/client/unversioned" -mkcp "/pkg/util/workqueue" "pkg/util" -# remove this folder because it imports prometheus -rm -rf "${CLIENT_REPO_TEMP}/pkg/util/workqueue/prometheus" # remove this test because it imports the internal clientset rm "${CLIENT_REPO_TEMP}"/pkg/client/unversioned/portforward/portforward_test.go @@ -146,9 +143,6 @@ find "${CLIENT_REPO_TEMP}"/pkg/client/record -type f -name "*.go" -print0 | xarg find "${CLIENT_REPO_TEMP}"/pkg/client/record -type f -name "*.go" -print0 | xargs -0 sed -i 's,pkg/api",pkg/api/v1",g' # gofmt the changed files -echo "rewrite conflicting Prometheus registration" -sed -i "s/kubernetes_build_info/kubernetes_build_info_copy/g" "${CLIENT_REPO_TEMP}"/pkg/version/version.go - echo "rewrite proto names in proto.RegisterType" find "${CLIENT_REPO_TEMP}" -type f -name "generated.pb.go" -print0 | xargs -0 sed -i "s/k8s\.io\.kubernetes/k8s.io.client-go/g" diff --git a/staging/src/k8s.io/client-go/pkg/third_party/forked/golang/json/fields.go b/staging/src/k8s.io/client-go/pkg/third_party/forked/golang/json/fields.go deleted file mode 100644 index 1d17270ee46..00000000000 --- a/staging/src/k8s.io/client-go/pkg/third_party/forked/golang/json/fields.go +++ /dev/null @@ -1,501 +0,0 @@ -// Copyright 2013 The Go Authors. All rights reserved. -// Use of this source code is governed by a BSD-style -// license that can be found in the LICENSE file. - -// Package json is forked from the Go standard library to enable us to find the -// field of a struct that a given JSON key maps to. -package json - -import ( - "bytes" - "fmt" - "reflect" - "sort" - "strings" - "sync" - "unicode" - "unicode/utf8" -) - -// Finds the patchStrategy and patchMergeKey struct tag fields on a given -// struct field given the struct type and the JSON name of the field. -// TODO: fix the returned errors to be introspectable. -func LookupPatchMetadata(t reflect.Type, jsonField string) (reflect.Type, string, string, error) { - if t.Kind() == reflect.Map { - return t.Elem(), "", "", nil - } - if t.Kind() != reflect.Struct { - return nil, "", "", fmt.Errorf("merging an object in json but data type is not map or struct, instead is: %s", - t.Kind().String()) - } - jf := []byte(jsonField) - // Find the field that the JSON library would use. - var f *field - fields := cachedTypeFields(t) - for i := range fields { - ff := &fields[i] - if bytes.Equal(ff.nameBytes, jf) { - f = ff - break - } - // Do case-insensitive comparison. - if f == nil && ff.equalFold(ff.nameBytes, jf) { - f = ff - } - } - if f != nil { - // Find the reflect.Value of the most preferential struct field. - tjf := t.Field(f.index[0]) - // we must navigate down all the anonymously included structs in the chain - for i := 1; i < len(f.index); i++ { - tjf = tjf.Type.Field(f.index[i]) - } - patchStrategy := tjf.Tag.Get("patchStrategy") - patchMergeKey := tjf.Tag.Get("patchMergeKey") - return tjf.Type, patchStrategy, patchMergeKey, nil - } - return nil, "", "", fmt.Errorf("unable to find api field in struct %s for the json field %q", t.Name(), jsonField) -} - -// A field represents a single field found in a struct. -type field struct { - name string - nameBytes []byte // []byte(name) - equalFold func(s, t []byte) bool // bytes.EqualFold or equivalent - - tag bool - // index is the sequence of indexes from the containing type fields to this field. - // it is a slice because anonymous structs will need multiple navigation steps to correctly - // resolve the proper fields - index []int - typ reflect.Type - omitEmpty bool - quoted bool -} - -func (f field) String() string { - return fmt.Sprintf("{name: %s, type: %v, tag: %v, index: %v, omitEmpty: %v, quoted: %v}", f.name, f.typ, f.tag, f.index, f.omitEmpty, f.quoted) -} - -func fillField(f field) field { - f.nameBytes = []byte(f.name) - f.equalFold = foldFunc(f.nameBytes) - return f -} - -// byName sorts field by name, breaking ties with depth, -// then breaking ties with "name came from json tag", then -// breaking ties with index sequence. -type byName []field - -func (x byName) Len() int { return len(x) } - -func (x byName) Swap(i, j int) { x[i], x[j] = x[j], x[i] } - -func (x byName) Less(i, j int) bool { - if x[i].name != x[j].name { - return x[i].name < x[j].name - } - if len(x[i].index) != len(x[j].index) { - return len(x[i].index) < len(x[j].index) - } - if x[i].tag != x[j].tag { - return x[i].tag - } - return byIndex(x).Less(i, j) -} - -// byIndex sorts field by index sequence. -type byIndex []field - -func (x byIndex) Len() int { return len(x) } - -func (x byIndex) Swap(i, j int) { x[i], x[j] = x[j], x[i] } - -func (x byIndex) Less(i, j int) bool { - for k, xik := range x[i].index { - if k >= len(x[j].index) { - return false - } - if xik != x[j].index[k] { - return xik < x[j].index[k] - } - } - return len(x[i].index) < len(x[j].index) -} - -// typeFields returns a list of fields that JSON should recognize for the given type. -// The algorithm is breadth-first search over the set of structs to include - the top struct -// and then any reachable anonymous structs. -func typeFields(t reflect.Type) []field { - // Anonymous fields to explore at the current level and the next. - current := []field{} - next := []field{{typ: t}} - - // Count of queued names for current level and the next. - count := map[reflect.Type]int{} - nextCount := map[reflect.Type]int{} - - // Types already visited at an earlier level. - visited := map[reflect.Type]bool{} - - // Fields found. - var fields []field - - for len(next) > 0 { - current, next = next, current[:0] - count, nextCount = nextCount, map[reflect.Type]int{} - - for _, f := range current { - if visited[f.typ] { - continue - } - visited[f.typ] = true - - // Scan f.typ for fields to include. - for i := 0; i < f.typ.NumField(); i++ { - sf := f.typ.Field(i) - if sf.PkgPath != "" { // unexported - continue - } - tag := sf.Tag.Get("json") - if tag == "-" { - continue - } - name, opts := parseTag(tag) - if !isValidTag(name) { - name = "" - } - index := make([]int, len(f.index)+1) - copy(index, f.index) - index[len(f.index)] = i - - ft := sf.Type - if ft.Name() == "" && ft.Kind() == reflect.Ptr { - // Follow pointer. - ft = ft.Elem() - } - - // Record found field and index sequence. - if name != "" || !sf.Anonymous || ft.Kind() != reflect.Struct { - tagged := name != "" - if name == "" { - name = sf.Name - } - fields = append(fields, fillField(field{ - name: name, - tag: tagged, - index: index, - typ: ft, - omitEmpty: opts.Contains("omitempty"), - quoted: opts.Contains("string"), - })) - if count[f.typ] > 1 { - // If there were multiple instances, add a second, - // so that the annihilation code will see a duplicate. - // It only cares about the distinction between 1 or 2, - // so don't bother generating any more copies. - fields = append(fields, fields[len(fields)-1]) - } - continue - } - - // Record new anonymous struct to explore in next round. - nextCount[ft]++ - if nextCount[ft] == 1 { - next = append(next, fillField(field{name: ft.Name(), index: index, typ: ft})) - } - } - } - } - - sort.Sort(byName(fields)) - - // Delete all fields that are hidden by the Go rules for embedded fields, - // except that fields with JSON tags are promoted. - - // The fields are sorted in primary order of name, secondary order - // of field index length. Loop over names; for each name, delete - // hidden fields by choosing the one dominant field that survives. - out := fields[:0] - for advance, i := 0, 0; i < len(fields); i += advance { - // One iteration per name. - // Find the sequence of fields with the name of this first field. - fi := fields[i] - name := fi.name - for advance = 1; i+advance < len(fields); advance++ { - fj := fields[i+advance] - if fj.name != name { - break - } - } - if advance == 1 { // Only one field with this name - out = append(out, fi) - continue - } - dominant, ok := dominantField(fields[i : i+advance]) - if ok { - out = append(out, dominant) - } - } - - fields = out - sort.Sort(byIndex(fields)) - - return fields -} - -// dominantField looks through the fields, all of which are known to -// have the same name, to find the single field that dominates the -// others using Go's embedding rules, modified by the presence of -// JSON tags. If there are multiple top-level fields, the boolean -// will be false: This condition is an error in Go and we skip all -// the fields. -func dominantField(fields []field) (field, bool) { - // The fields are sorted in increasing index-length order. The winner - // must therefore be one with the shortest index length. Drop all - // longer entries, which is easy: just truncate the slice. - length := len(fields[0].index) - tagged := -1 // Index of first tagged field. - for i, f := range fields { - if len(f.index) > length { - fields = fields[:i] - break - } - if f.tag { - if tagged >= 0 { - // Multiple tagged fields at the same level: conflict. - // Return no field. - return field{}, false - } - tagged = i - } - } - if tagged >= 0 { - return fields[tagged], true - } - // All remaining fields have the same length. If there's more than one, - // we have a conflict (two fields named "X" at the same level) and we - // return no field. - if len(fields) > 1 { - return field{}, false - } - return fields[0], true -} - -var fieldCache struct { - sync.RWMutex - m map[reflect.Type][]field -} - -// cachedTypeFields is like typeFields but uses a cache to avoid repeated work. -func cachedTypeFields(t reflect.Type) []field { - fieldCache.RLock() - f := fieldCache.m[t] - fieldCache.RUnlock() - if f != nil { - return f - } - - // Compute fields without lock. - // Might duplicate effort but won't hold other computations back. - f = typeFields(t) - if f == nil { - f = []field{} - } - - fieldCache.Lock() - if fieldCache.m == nil { - fieldCache.m = map[reflect.Type][]field{} - } - fieldCache.m[t] = f - fieldCache.Unlock() - return f -} - -func isValidTag(s string) bool { - if s == "" { - return false - } - for _, c := range s { - switch { - case strings.ContainsRune("!#$%&()*+-./:<=>?@[]^_{|}~ ", c): - // Backslash and quote chars are reserved, but - // otherwise any punctuation chars are allowed - // in a tag name. - default: - if !unicode.IsLetter(c) && !unicode.IsDigit(c) { - return false - } - } - } - return true -} - -const ( - caseMask = ^byte(0x20) // Mask to ignore case in ASCII. - kelvin = '\u212a' - smallLongEss = '\u017f' -) - -// foldFunc returns one of four different case folding equivalence -// functions, from most general (and slow) to fastest: -// -// 1) bytes.EqualFold, if the key s contains any non-ASCII UTF-8 -// 2) equalFoldRight, if s contains special folding ASCII ('k', 'K', 's', 'S') -// 3) asciiEqualFold, no special, but includes non-letters (including _) -// 4) simpleLetterEqualFold, no specials, no non-letters. -// -// The letters S and K are special because they map to 3 runes, not just 2: -// * S maps to s and to U+017F 'ſ' Latin small letter long s -// * k maps to K and to U+212A 'K' Kelvin sign -// See http://play.golang.org/p/tTxjOc0OGo -// -// The returned function is specialized for matching against s and -// should only be given s. It's not curried for performance reasons. -func foldFunc(s []byte) func(s, t []byte) bool { - nonLetter := false - special := false // special letter - for _, b := range s { - if b >= utf8.RuneSelf { - return bytes.EqualFold - } - upper := b & caseMask - if upper < 'A' || upper > 'Z' { - nonLetter = true - } else if upper == 'K' || upper == 'S' { - // See above for why these letters are special. - special = true - } - } - if special { - return equalFoldRight - } - if nonLetter { - return asciiEqualFold - } - return simpleLetterEqualFold -} - -// equalFoldRight is a specialization of bytes.EqualFold when s is -// known to be all ASCII (including punctuation), but contains an 's', -// 'S', 'k', or 'K', requiring a Unicode fold on the bytes in t. -// See comments on foldFunc. -func equalFoldRight(s, t []byte) bool { - for _, sb := range s { - if len(t) == 0 { - return false - } - tb := t[0] - if tb < utf8.RuneSelf { - if sb != tb { - sbUpper := sb & caseMask - if 'A' <= sbUpper && sbUpper <= 'Z' { - if sbUpper != tb&caseMask { - return false - } - } else { - return false - } - } - t = t[1:] - continue - } - // sb is ASCII and t is not. t must be either kelvin - // sign or long s; sb must be s, S, k, or K. - tr, size := utf8.DecodeRune(t) - switch sb { - case 's', 'S': - if tr != smallLongEss { - return false - } - case 'k', 'K': - if tr != kelvin { - return false - } - default: - return false - } - t = t[size:] - - } - if len(t) > 0 { - return false - } - return true -} - -// asciiEqualFold is a specialization of bytes.EqualFold for use when -// s is all ASCII (but may contain non-letters) and contains no -// special-folding letters. -// See comments on foldFunc. -func asciiEqualFold(s, t []byte) bool { - if len(s) != len(t) { - return false - } - for i, sb := range s { - tb := t[i] - if sb == tb { - continue - } - if ('a' <= sb && sb <= 'z') || ('A' <= sb && sb <= 'Z') { - if sb&caseMask != tb&caseMask { - return false - } - } else { - return false - } - } - return true -} - -// simpleLetterEqualFold is a specialization of bytes.EqualFold for -// use when s is all ASCII letters (no underscores, etc) and also -// doesn't contain 'k', 'K', 's', or 'S'. -// See comments on foldFunc. -func simpleLetterEqualFold(s, t []byte) bool { - if len(s) != len(t) { - return false - } - for i, b := range s { - if b&caseMask != t[i]&caseMask { - return false - } - } - return true -} - -// tagOptions is the string following a comma in a struct field's "json" -// tag, or the empty string. It does not include the leading comma. -type tagOptions string - -// parseTag splits a struct field's json tag into its name and -// comma-separated options. -func parseTag(tag string) (string, tagOptions) { - if idx := strings.Index(tag, ","); idx != -1 { - return tag[:idx], tagOptions(tag[idx+1:]) - } - return tag, tagOptions("") -} - -// Contains reports whether a comma-separated list of options -// contains a particular substr flag. substr must be surrounded by a -// string boundary or commas. -func (o tagOptions) Contains(optionName string) bool { - if len(o) == 0 { - return false - } - s := string(o) - for s != "" { - var next string - i := strings.Index(s, ",") - if i >= 0 { - s, next = s[:i], s[i+1:] - } - if s == optionName { - return true - } - s = next - } - return false -} diff --git a/staging/src/k8s.io/client-go/pkg/third_party/forked/golang/netutil/addr.go b/staging/src/k8s.io/client-go/pkg/third_party/forked/golang/netutil/addr.go deleted file mode 100644 index c70f431c272..00000000000 --- a/staging/src/k8s.io/client-go/pkg/third_party/forked/golang/netutil/addr.go +++ /dev/null @@ -1,27 +0,0 @@ -package netutil - -import ( - "net/url" - "strings" -) - -// FROM: http://golang.org/src/net/http/client.go -// Given a string of the form "host", "host:port", or "[ipv6::address]:port", -// return true if the string includes a port. -func hasPort(s string) bool { return strings.LastIndex(s, ":") > strings.LastIndex(s, "]") } - -// FROM: http://golang.org/src/net/http/transport.go -var portMap = map[string]string{ - "http": "80", - "https": "443", -} - -// FROM: http://golang.org/src/net/http/transport.go -// canonicalAddr returns url.Host but always with a ":port" suffix -func CanonicalAddr(url *url.URL) string { - addr := url.Host - if !hasPort(addr) { - return addr + ":" + portMap[url.Scheme] - } - return addr -} diff --git a/staging/src/k8s.io/client-go/pkg/util/workqueue/default_rate_limiters.go b/staging/src/k8s.io/client-go/pkg/util/workqueue/default_rate_limiters.go deleted file mode 100644 index 35caed4fa41..00000000000 --- a/staging/src/k8s.io/client-go/pkg/util/workqueue/default_rate_limiters.go +++ /dev/null @@ -1,211 +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 workqueue - -import ( - "math" - "sync" - "time" - - "github.com/juju/ratelimit" -) - -type RateLimiter interface { - // When gets an item and gets to decide how long that item should wait - When(item interface{}) time.Duration - // Forget indicates that an item is finished being retried. Doesn't matter whether its for perm failing - // or for success, we'll stop tracking it - Forget(item interface{}) - // NumRequeues returns back how many failures the item has had - NumRequeues(item interface{}) int -} - -// DefaultControllerRateLimiter is a no-arg constructor for a default rate limiter for a workqueue. It has -// both overall and per-item rate limitting. The overall is a token bucket and the per-item is exponential -func DefaultControllerRateLimiter() RateLimiter { - return NewMaxOfRateLimiter( - NewItemExponentialFailureRateLimiter(5*time.Millisecond, 1000*time.Second), - // 10 qps, 100 bucket size. This is only for retry speed and its only the overall factor (not per item) - &BucketRateLimiter{Bucket: ratelimit.NewBucketWithRate(float64(10), int64(100))}, - ) -} - -// BucketRateLimiter adapts a standard bucket to the workqueue ratelimiter API -type BucketRateLimiter struct { - *ratelimit.Bucket -} - -var _ RateLimiter = &BucketRateLimiter{} - -func (r *BucketRateLimiter) When(item interface{}) time.Duration { - return r.Bucket.Take(1) -} - -func (r *BucketRateLimiter) NumRequeues(item interface{}) int { - return 0 -} - -func (r *BucketRateLimiter) Forget(item interface{}) { -} - -// ItemExponentialFailureRateLimiter does a simple baseDelay*10^ limit -// dealing with max failures and expiration are up to the caller -type ItemExponentialFailureRateLimiter struct { - failuresLock sync.Mutex - failures map[interface{}]int - - baseDelay time.Duration - maxDelay time.Duration -} - -var _ RateLimiter = &ItemExponentialFailureRateLimiter{} - -func NewItemExponentialFailureRateLimiter(baseDelay time.Duration, maxDelay time.Duration) RateLimiter { - return &ItemExponentialFailureRateLimiter{ - failures: map[interface{}]int{}, - baseDelay: baseDelay, - maxDelay: maxDelay, - } -} - -func DefaultItemBasedRateLimiter() RateLimiter { - return NewItemExponentialFailureRateLimiter(time.Millisecond, 1000*time.Second) -} - -func (r *ItemExponentialFailureRateLimiter) When(item interface{}) time.Duration { - r.failuresLock.Lock() - defer r.failuresLock.Unlock() - - exp := r.failures[item] - r.failures[item] = r.failures[item] + 1 - - // The backoff is capped such that 'calculated' value never overflows. - backoff := float64(r.baseDelay.Nanoseconds()) * math.Pow(2, float64(exp)) - if backoff > math.MaxInt64 { - return r.maxDelay - } - - calculated := time.Duration(backoff) - if calculated > r.maxDelay { - return r.maxDelay - } - - return calculated -} - -func (r *ItemExponentialFailureRateLimiter) NumRequeues(item interface{}) int { - r.failuresLock.Lock() - defer r.failuresLock.Unlock() - - return r.failures[item] -} - -func (r *ItemExponentialFailureRateLimiter) Forget(item interface{}) { - r.failuresLock.Lock() - defer r.failuresLock.Unlock() - - delete(r.failures, item) -} - -// ItemFastSlowRateLimiter does a quick retry for a certain number of attempts, then a slow retry after that -type ItemFastSlowRateLimiter struct { - failuresLock sync.Mutex - failures map[interface{}]int - - maxFastAttempts int - fastDelay time.Duration - slowDelay time.Duration -} - -var _ RateLimiter = &ItemFastSlowRateLimiter{} - -func NewItemFastSlowRateLimiter(fastDelay, slowDelay time.Duration, maxFastAttempts int) RateLimiter { - return &ItemFastSlowRateLimiter{ - failures: map[interface{}]int{}, - fastDelay: fastDelay, - slowDelay: slowDelay, - maxFastAttempts: maxFastAttempts, - } -} - -func (r *ItemFastSlowRateLimiter) When(item interface{}) time.Duration { - r.failuresLock.Lock() - defer r.failuresLock.Unlock() - - r.failures[item] = r.failures[item] + 1 - - if r.failures[item] <= r.maxFastAttempts { - return r.fastDelay - } - - return r.slowDelay -} - -func (r *ItemFastSlowRateLimiter) NumRequeues(item interface{}) int { - r.failuresLock.Lock() - defer r.failuresLock.Unlock() - - return r.failures[item] -} - -func (r *ItemFastSlowRateLimiter) Forget(item interface{}) { - r.failuresLock.Lock() - defer r.failuresLock.Unlock() - - delete(r.failures, item) -} - -// MaxOfRateLimiter calls every RateLimiter and returns the worst case response -// When used with a token bucket limiter, the burst could be apparently exceeded in cases where particular items -// were separately delayed a longer time. -type MaxOfRateLimiter struct { - limiters []RateLimiter -} - -func (r *MaxOfRateLimiter) When(item interface{}) time.Duration { - ret := time.Duration(0) - for _, limiter := range r.limiters { - curr := limiter.When(item) - if curr > ret { - ret = curr - } - } - - return ret -} - -func NewMaxOfRateLimiter(limiters ...RateLimiter) RateLimiter { - return &MaxOfRateLimiter{limiters: limiters} -} - -func (r *MaxOfRateLimiter) NumRequeues(item interface{}) int { - ret := 0 - for _, limiter := range r.limiters { - curr := limiter.NumRequeues(item) - if curr > ret { - ret = curr - } - } - - return ret -} - -func (r *MaxOfRateLimiter) Forget(item interface{}) { - for _, limiter := range r.limiters { - limiter.Forget(item) - } -} diff --git a/staging/src/k8s.io/client-go/pkg/util/workqueue/default_rate_limiters_test.go b/staging/src/k8s.io/client-go/pkg/util/workqueue/default_rate_limiters_test.go deleted file mode 100644 index 91d34a31779..00000000000 --- a/staging/src/k8s.io/client-go/pkg/util/workqueue/default_rate_limiters_test.go +++ /dev/null @@ -1,184 +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 workqueue - -import ( - "testing" - "time" -) - -func TestItemExponentialFailureRateLimiter(t *testing.T) { - limiter := NewItemExponentialFailureRateLimiter(1*time.Millisecond, 1*time.Second) - - if e, a := 1*time.Millisecond, limiter.When("one"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - if e, a := 2*time.Millisecond, limiter.When("one"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - if e, a := 4*time.Millisecond, limiter.When("one"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - if e, a := 8*time.Millisecond, limiter.When("one"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - if e, a := 16*time.Millisecond, limiter.When("one"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - if e, a := 5, limiter.NumRequeues("one"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - - if e, a := 1*time.Millisecond, limiter.When("two"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - if e, a := 2*time.Millisecond, limiter.When("two"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - if e, a := 2, limiter.NumRequeues("two"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - - limiter.Forget("one") - if e, a := 0, limiter.NumRequeues("one"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - if e, a := 1*time.Millisecond, limiter.When("one"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - -} - -func TestItemExponentialFailureRateLimiterOverFlow(t *testing.T) { - limiter := NewItemExponentialFailureRateLimiter(1*time.Millisecond, 1000*time.Second) - for i := 0; i < 5; i++ { - limiter.When("one") - } - if e, a := 32*time.Millisecond, limiter.When("one"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - - for i := 0; i < 1000; i++ { - limiter.When("overflow1") - } - if e, a := 1000*time.Second, limiter.When("overflow1"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - - limiter = NewItemExponentialFailureRateLimiter(1*time.Minute, 1000*time.Hour) - for i := 0; i < 2; i++ { - limiter.When("two") - } - if e, a := 4*time.Minute, limiter.When("two"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - - for i := 0; i < 1000; i++ { - limiter.When("overflow2") - } - if e, a := 1000*time.Hour, limiter.When("overflow2"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - -} - -func TestItemFastSlowRateLimiter(t *testing.T) { - limiter := NewItemFastSlowRateLimiter(5*time.Millisecond, 10*time.Second, 3) - - if e, a := 5*time.Millisecond, limiter.When("one"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - if e, a := 5*time.Millisecond, limiter.When("one"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - if e, a := 5*time.Millisecond, limiter.When("one"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - if e, a := 10*time.Second, limiter.When("one"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - if e, a := 10*time.Second, limiter.When("one"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - if e, a := 5, limiter.NumRequeues("one"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - - if e, a := 5*time.Millisecond, limiter.When("two"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - if e, a := 5*time.Millisecond, limiter.When("two"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - if e, a := 2, limiter.NumRequeues("two"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - - limiter.Forget("one") - if e, a := 0, limiter.NumRequeues("one"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - if e, a := 5*time.Millisecond, limiter.When("one"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - -} - -func TestMaxOfRateLimiter(t *testing.T) { - limiter := NewMaxOfRateLimiter( - NewItemFastSlowRateLimiter(5*time.Millisecond, 3*time.Second, 3), - NewItemExponentialFailureRateLimiter(1*time.Millisecond, 1*time.Second), - ) - - if e, a := 5*time.Millisecond, limiter.When("one"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - if e, a := 5*time.Millisecond, limiter.When("one"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - if e, a := 5*time.Millisecond, limiter.When("one"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - if e, a := 3*time.Second, limiter.When("one"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - if e, a := 3*time.Second, limiter.When("one"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - if e, a := 5, limiter.NumRequeues("one"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - - if e, a := 5*time.Millisecond, limiter.When("two"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - if e, a := 5*time.Millisecond, limiter.When("two"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - if e, a := 2, limiter.NumRequeues("two"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - - limiter.Forget("one") - if e, a := 0, limiter.NumRequeues("one"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - if e, a := 5*time.Millisecond, limiter.When("one"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - -} diff --git a/staging/src/k8s.io/client-go/pkg/util/workqueue/delaying_queue.go b/staging/src/k8s.io/client-go/pkg/util/workqueue/delaying_queue.go deleted file mode 100644 index 593ad9ad415..00000000000 --- a/staging/src/k8s.io/client-go/pkg/util/workqueue/delaying_queue.go +++ /dev/null @@ -1,246 +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 workqueue - -import ( - "sort" - "time" - - utilruntime "k8s.io/apimachinery/pkg/util/runtime" - "k8s.io/client-go/util/clock" -) - -// DelayingInterface is an Interface that can Add an item at a later time. This makes it easier to -// requeue items after failures without ending up in a hot-loop. -type DelayingInterface interface { - Interface - // AddAfter adds an item to the workqueue after the indicated duration has passed - AddAfter(item interface{}, duration time.Duration) -} - -// NewDelayingQueue constructs a new workqueue with delayed queuing ability -func NewDelayingQueue() DelayingInterface { - return newDelayingQueue(clock.RealClock{}, "") -} - -func NewNamedDelayingQueue(name string) DelayingInterface { - return newDelayingQueue(clock.RealClock{}, name) -} - -func newDelayingQueue(clock clock.Clock, name string) DelayingInterface { - ret := &delayingType{ - Interface: NewNamed(name), - clock: clock, - heartbeat: clock.Tick(maxWait), - stopCh: make(chan struct{}), - waitingTimeByEntry: map[t]time.Time{}, - waitingForAddCh: make(chan waitFor, 1000), - metrics: newRetryMetrics(name), - } - - go ret.waitingLoop() - - return ret -} - -// delayingType wraps an Interface and provides delayed re-enquing -type delayingType struct { - Interface - - // clock tracks time for delayed firing - clock clock.Clock - - // stopCh lets us signal a shutdown to the waiting loop - stopCh chan struct{} - - // heartbeat ensures we wait no more than maxWait before firing - // - // TODO: replace with Ticker (and add to clock) so this can be cleaned up. - // clock.Tick will leak. - heartbeat <-chan time.Time - - // waitingForAdd is an ordered slice of items to be added to the contained work queue - waitingForAdd []waitFor - // waitingTimeByEntry holds wait time by entry, so we can lookup pre-existing indexes - waitingTimeByEntry map[t]time.Time - // waitingForAddCh is a buffered channel that feeds waitingForAdd - waitingForAddCh chan waitFor - - // metrics counts the number of retries - metrics retryMetrics -} - -// waitFor holds the data to add and the time it should be added -type waitFor struct { - data t - readyAt time.Time -} - -// ShutDown gives a way to shut off this queue -func (q *delayingType) ShutDown() { - q.Interface.ShutDown() - close(q.stopCh) -} - -// AddAfter adds the given item to the work queue after the given delay -func (q *delayingType) AddAfter(item interface{}, duration time.Duration) { - // don't add if we're already shutting down - if q.ShuttingDown() { - return - } - - q.metrics.retry() - - // immediately add things with no delay - if duration <= 0 { - q.Add(item) - return - } - - select { - case <-q.stopCh: - // unblock if ShutDown() is called - case q.waitingForAddCh <- waitFor{data: item, readyAt: q.clock.Now().Add(duration)}: - } -} - -// maxWait keeps a max bound on the wait time. It's just insurance against weird things happening. -// Checking the queue every 10 seconds isn't expensive and we know that we'll never end up with an -// expired item sitting for more than 10 seconds. -const maxWait = 10 * time.Second - -// waitingLoop runs until the workqueue is shutdown and keeps a check on the list of items to be added. -func (q *delayingType) waitingLoop() { - defer utilruntime.HandleCrash() - - // Make a placeholder channel to use when there are no items in our list - never := make(<-chan time.Time) - - for { - if q.Interface.ShuttingDown() { - // discard waiting entries - q.waitingForAdd = nil - q.waitingTimeByEntry = nil - return - } - - now := q.clock.Now() - - // Add ready entries - readyEntries := 0 - for _, entry := range q.waitingForAdd { - if entry.readyAt.After(now) { - break - } - q.Add(entry.data) - delete(q.waitingTimeByEntry, entry.data) - readyEntries++ - } - q.waitingForAdd = q.waitingForAdd[readyEntries:] - - // Set up a wait for the first item's readyAt (if one exists) - nextReadyAt := never - if len(q.waitingForAdd) > 0 { - nextReadyAt = q.clock.After(q.waitingForAdd[0].readyAt.Sub(now)) - } - - select { - case <-q.stopCh: - return - - case <-q.heartbeat: - // continue the loop, which will add ready items - - case <-nextReadyAt: - // continue the loop, which will add ready items - - case waitEntry := <-q.waitingForAddCh: - if waitEntry.readyAt.After(q.clock.Now()) { - q.waitingForAdd = insert(q.waitingForAdd, q.waitingTimeByEntry, waitEntry) - } else { - q.Add(waitEntry.data) - } - - drained := false - for !drained { - select { - case waitEntry := <-q.waitingForAddCh: - if waitEntry.readyAt.After(q.clock.Now()) { - q.waitingForAdd = insert(q.waitingForAdd, q.waitingTimeByEntry, waitEntry) - } else { - q.Add(waitEntry.data) - } - default: - drained = true - } - } - } - } -} - -// inserts the given entry into the sorted entries list -// same semantics as append()... the given slice may be modified, -// and the returned value should be used -// -// TODO: This should probably be converted to use container/heap to improve -// running time for a large number of items. -func insert(entries []waitFor, knownEntries map[t]time.Time, entry waitFor) []waitFor { - // if the entry is already in our retry list and the existing time is before the new one, just skip it - existingTime, exists := knownEntries[entry.data] - if exists && existingTime.Before(entry.readyAt) { - return entries - } - - // if the entry exists and is scheduled for later, go ahead and remove the entry - if exists { - if existingIndex := findEntryIndex(entries, existingTime, entry.data); existingIndex >= 0 && existingIndex < len(entries) { - entries = append(entries[:existingIndex], entries[existingIndex+1:]...) - } - } - - insertionIndex := sort.Search(len(entries), func(i int) bool { - return entry.readyAt.Before(entries[i].readyAt) - }) - - // grow by 1 - entries = append(entries, waitFor{}) - // shift items from the insertion point to the end - copy(entries[insertionIndex+1:], entries[insertionIndex:]) - // insert the record - entries[insertionIndex] = entry - - knownEntries[entry.data] = entry.readyAt - - return entries -} - -// findEntryIndex returns the index for an existing entry -func findEntryIndex(entries []waitFor, existingTime time.Time, data t) int { - index := sort.Search(len(entries), func(i int) bool { - return entries[i].readyAt.After(existingTime) || existingTime == entries[i].readyAt - }) - - // we know this is the earliest possible index, but there could be multiple with the same time - // iterate from here to find the dupe - for ; index < len(entries); index++ { - if entries[index].data == data { - break - } - } - - return index -} diff --git a/staging/src/k8s.io/client-go/pkg/util/workqueue/delaying_queue_test.go b/staging/src/k8s.io/client-go/pkg/util/workqueue/delaying_queue_test.go deleted file mode 100644 index 1b7661064a2..00000000000 --- a/staging/src/k8s.io/client-go/pkg/util/workqueue/delaying_queue_test.go +++ /dev/null @@ -1,236 +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 workqueue - -import ( - "fmt" - "reflect" - "testing" - "time" - - "k8s.io/apimachinery/pkg/util/wait" - "k8s.io/client-go/util/clock" -) - -func TestSimpleQueue(t *testing.T) { - fakeClock := clock.NewFakeClock(time.Now()) - q := newDelayingQueue(fakeClock, "") - - first := "foo" - - q.AddAfter(first, 50*time.Millisecond) - if err := waitForWaitingQueueToFill(q); err != nil { - t.Fatalf("unexpected err: %v", err) - } - - if q.Len() != 0 { - t.Errorf("should not have added") - } - - fakeClock.Step(60 * time.Millisecond) - - if err := waitForAdded(q, 1); err != nil { - t.Errorf("should have added") - } - item, _ := q.Get() - q.Done(item) - - // step past the next heartbeat - fakeClock.Step(10 * time.Second) - - err := wait.Poll(1*time.Millisecond, 30*time.Millisecond, func() (done bool, err error) { - if q.Len() > 0 { - return false, fmt.Errorf("added to queue") - } - - return false, nil - }) - if err != wait.ErrWaitTimeout { - t.Errorf("expected timeout, got: %v", err) - } - - if q.Len() != 0 { - t.Errorf("should not have added") - } -} - -func TestDeduping(t *testing.T) { - fakeClock := clock.NewFakeClock(time.Now()) - q := newDelayingQueue(fakeClock, "") - - first := "foo" - - q.AddAfter(first, 50*time.Millisecond) - if err := waitForWaitingQueueToFill(q); err != nil { - t.Fatalf("unexpected err: %v", err) - } - q.AddAfter(first, 70*time.Millisecond) - if err := waitForWaitingQueueToFill(q); err != nil { - t.Fatalf("unexpected err: %v", err) - } - if q.Len() != 0 { - t.Errorf("should not have added") - } - - // step past the first block, we should receive now - fakeClock.Step(60 * time.Millisecond) - if err := waitForAdded(q, 1); err != nil { - t.Errorf("should have added") - } - item, _ := q.Get() - q.Done(item) - - // step past the second add - fakeClock.Step(20 * time.Millisecond) - if q.Len() != 0 { - t.Errorf("should not have added") - } - - // test again, but this time the earlier should override - q.AddAfter(first, 50*time.Millisecond) - q.AddAfter(first, 30*time.Millisecond) - if err := waitForWaitingQueueToFill(q); err != nil { - t.Fatalf("unexpected err: %v", err) - } - if q.Len() != 0 { - t.Errorf("should not have added") - } - - fakeClock.Step(40 * time.Millisecond) - if err := waitForAdded(q, 1); err != nil { - t.Errorf("should have added") - } - item, _ = q.Get() - q.Done(item) - - // step past the second add - fakeClock.Step(20 * time.Millisecond) - if q.Len() != 0 { - t.Errorf("should not have added") - } - if q.Len() != 0 { - t.Errorf("should not have added") - } -} - -func TestAddTwoFireEarly(t *testing.T) { - fakeClock := clock.NewFakeClock(time.Now()) - q := newDelayingQueue(fakeClock, "") - - first := "foo" - second := "bar" - third := "baz" - - q.AddAfter(first, 1*time.Second) - q.AddAfter(second, 50*time.Millisecond) - if err := waitForWaitingQueueToFill(q); err != nil { - t.Fatalf("unexpected err: %v", err) - } - - if q.Len() != 0 { - t.Errorf("should not have added") - } - - fakeClock.Step(60 * time.Millisecond) - - if err := waitForAdded(q, 1); err != nil { - t.Fatalf("unexpected err: %v", err) - } - item, _ := q.Get() - if !reflect.DeepEqual(item, second) { - t.Errorf("expected %v, got %v", second, item) - } - - q.AddAfter(third, 2*time.Second) - - fakeClock.Step(1 * time.Second) - if err := waitForAdded(q, 1); err != nil { - t.Fatalf("unexpected err: %v", err) - } - item, _ = q.Get() - if !reflect.DeepEqual(item, first) { - t.Errorf("expected %v, got %v", first, item) - } - - fakeClock.Step(2 * time.Second) - if err := waitForAdded(q, 1); err != nil { - t.Fatalf("unexpected err: %v", err) - } - item, _ = q.Get() - if !reflect.DeepEqual(item, third) { - t.Errorf("expected %v, got %v", third, item) - } - -} - -func TestCopyShifting(t *testing.T) { - fakeClock := clock.NewFakeClock(time.Now()) - q := newDelayingQueue(fakeClock, "") - - first := "foo" - second := "bar" - third := "baz" - - q.AddAfter(first, 1*time.Second) - q.AddAfter(second, 500*time.Millisecond) - q.AddAfter(third, 250*time.Millisecond) - if err := waitForWaitingQueueToFill(q); err != nil { - t.Fatalf("unexpected err: %v", err) - } - - if q.Len() != 0 { - t.Errorf("should not have added") - } - - fakeClock.Step(2 * time.Second) - - if err := waitForAdded(q, 3); err != nil { - t.Fatalf("unexpected err: %v", err) - } - actualFirst, _ := q.Get() - if !reflect.DeepEqual(actualFirst, third) { - t.Errorf("expected %v, got %v", third, actualFirst) - } - actualSecond, _ := q.Get() - if !reflect.DeepEqual(actualSecond, second) { - t.Errorf("expected %v, got %v", second, actualSecond) - } - actualThird, _ := q.Get() - if !reflect.DeepEqual(actualThird, first) { - t.Errorf("expected %v, got %v", first, actualThird) - } -} - -func waitForAdded(q DelayingInterface, depth int) error { - return wait.Poll(1*time.Millisecond, 10*time.Second, func() (done bool, err error) { - if q.Len() == depth { - return true, nil - } - - return false, nil - }) -} - -func waitForWaitingQueueToFill(q DelayingInterface) error { - return wait.Poll(1*time.Millisecond, 10*time.Second, func() (done bool, err error) { - if len(q.(*delayingType).waitingForAddCh) == 0 { - return true, nil - } - - return false, nil - }) -} diff --git a/staging/src/k8s.io/client-go/pkg/util/workqueue/metrics.go b/staging/src/k8s.io/client-go/pkg/util/workqueue/metrics.go deleted file mode 100644 index a481bdfb266..00000000000 --- a/staging/src/k8s.io/client-go/pkg/util/workqueue/metrics.go +++ /dev/null @@ -1,195 +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 workqueue - -import ( - "sync" - "time" -) - -// This file provides abstractions for setting the provider (e.g., prometheus) -// of metrics. - -type queueMetrics interface { - add(item t) - get(item t) - done(item t) -} - -// GaugeMetric represents a single numerical value that can arbitrarily go up -// and down. -type GaugeMetric interface { - Inc() - Dec() -} - -// CounterMetric represents a single numerical value that only ever -// goes up. -type CounterMetric interface { - Inc() -} - -// SummaryMetric captures individual observations. -type SummaryMetric interface { - Observe(float64) -} - -type noopMetric struct{} - -func (noopMetric) Inc() {} -func (noopMetric) Dec() {} -func (noopMetric) Observe(float64) {} - -type defaultQueueMetrics struct { - // current depth of a workqueue - depth GaugeMetric - // total number of adds handled by a workqueue - adds CounterMetric - // how long an item stays in a workqueue - latency SummaryMetric - // how long processing an item from a workqueue takes - workDuration SummaryMetric - addTimes map[t]time.Time - processingStartTimes map[t]time.Time -} - -func (m *defaultQueueMetrics) add(item t) { - if m == nil { - return - } - - m.adds.Inc() - m.depth.Inc() - if _, exists := m.addTimes[item]; !exists { - m.addTimes[item] = time.Now() - } -} - -func (m *defaultQueueMetrics) get(item t) { - if m == nil { - return - } - - m.depth.Dec() - m.processingStartTimes[item] = time.Now() - if startTime, exists := m.addTimes[item]; exists { - m.latency.Observe(sinceInMicroseconds(startTime)) - delete(m.addTimes, item) - } -} - -func (m *defaultQueueMetrics) done(item t) { - if m == nil { - return - } - - if startTime, exists := m.processingStartTimes[item]; exists { - m.workDuration.Observe(sinceInMicroseconds(startTime)) - delete(m.processingStartTimes, item) - } -} - -// Gets the time since the specified start in microseconds. -func sinceInMicroseconds(start time.Time) float64 { - return float64(time.Since(start).Nanoseconds() / time.Microsecond.Nanoseconds()) -} - -type retryMetrics interface { - retry() -} - -type defaultRetryMetrics struct { - retries CounterMetric -} - -func (m *defaultRetryMetrics) retry() { - if m == nil { - return - } - - m.retries.Inc() -} - -// MetricsProvider generates various metrics used by the queue. -type MetricsProvider interface { - NewDepthMetric(name string) GaugeMetric - NewAddsMetric(name string) CounterMetric - NewLatencyMetric(name string) SummaryMetric - NewWorkDurationMetric(name string) SummaryMetric - NewRetriesMetric(name string) CounterMetric -} - -type noopMetricsProvider struct{} - -func (_ noopMetricsProvider) NewDepthMetric(name string) GaugeMetric { - return noopMetric{} -} - -func (_ noopMetricsProvider) NewAddsMetric(name string) CounterMetric { - return noopMetric{} -} - -func (_ noopMetricsProvider) NewLatencyMetric(name string) SummaryMetric { - return noopMetric{} -} - -func (_ noopMetricsProvider) NewWorkDurationMetric(name string) SummaryMetric { - return noopMetric{} -} - -func (_ noopMetricsProvider) NewRetriesMetric(name string) CounterMetric { - return noopMetric{} -} - -var metricsFactory = struct { - metricsProvider MetricsProvider - setProviders sync.Once -}{ - metricsProvider: noopMetricsProvider{}, -} - -func newQueueMetrics(name string) queueMetrics { - var ret *defaultQueueMetrics - if len(name) == 0 { - return ret - } - return &defaultQueueMetrics{ - depth: metricsFactory.metricsProvider.NewDepthMetric(name), - adds: metricsFactory.metricsProvider.NewAddsMetric(name), - latency: metricsFactory.metricsProvider.NewLatencyMetric(name), - workDuration: metricsFactory.metricsProvider.NewWorkDurationMetric(name), - addTimes: map[t]time.Time{}, - processingStartTimes: map[t]time.Time{}, - } -} - -func newRetryMetrics(name string) retryMetrics { - var ret *defaultRetryMetrics - if len(name) == 0 { - return ret - } - return &defaultRetryMetrics{ - retries: metricsFactory.metricsProvider.NewRetriesMetric(name), - } -} - -// SetProvider sets the metrics provider of the metricsFactory. -func SetProvider(metricsProvider MetricsProvider) { - metricsFactory.setProviders.Do(func() { - metricsFactory.metricsProvider = metricsProvider - }) -} diff --git a/staging/src/k8s.io/client-go/pkg/util/workqueue/parallelizer.go b/staging/src/k8s.io/client-go/pkg/util/workqueue/parallelizer.go deleted file mode 100644 index be668c42339..00000000000 --- a/staging/src/k8s.io/client-go/pkg/util/workqueue/parallelizer.go +++ /dev/null @@ -1,52 +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 workqueue - -import ( - "sync" - - utilruntime "k8s.io/apimachinery/pkg/util/runtime" -) - -type DoWorkPieceFunc func(piece int) - -// Parallelize is a very simple framework that allow for parallelizing -// N independent pieces of work. -func Parallelize(workers, pieces int, doWorkPiece DoWorkPieceFunc) { - toProcess := make(chan int, pieces) - for i := 0; i < pieces; i++ { - toProcess <- i - } - close(toProcess) - - if pieces < workers { - workers = pieces - } - - wg := sync.WaitGroup{} - wg.Add(workers) - for i := 0; i < workers; i++ { - go func() { - defer utilruntime.HandleCrash() - defer wg.Done() - for piece := range toProcess { - doWorkPiece(piece) - } - }() - } - wg.Wait() -} diff --git a/staging/src/k8s.io/client-go/pkg/util/workqueue/queue.go b/staging/src/k8s.io/client-go/pkg/util/workqueue/queue.go deleted file mode 100644 index 3e1a49fe202..00000000000 --- a/staging/src/k8s.io/client-go/pkg/util/workqueue/queue.go +++ /dev/null @@ -1,172 +0,0 @@ -/* -Copyright 2015 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 workqueue - -import ( - "sync" -) - -type Interface interface { - Add(item interface{}) - Len() int - Get() (item interface{}, shutdown bool) - Done(item interface{}) - ShutDown() - ShuttingDown() bool -} - -// New constructs a new workqueue (see the package comment). -func New() *Type { - return NewNamed("") -} - -func NewNamed(name string) *Type { - return &Type{ - dirty: set{}, - processing: set{}, - cond: sync.NewCond(&sync.Mutex{}), - metrics: newQueueMetrics(name), - } -} - -// Type is a work queue (see the package comment). -type Type struct { - // queue defines the order in which we will work on items. Every - // element of queue should be in the dirty set and not in the - // processing set. - queue []t - - // dirty defines all of the items that need to be processed. - dirty set - - // Things that are currently being processed are in the processing set. - // These things may be simultaneously in the dirty set. When we finish - // processing something and remove it from this set, we'll check if - // it's in the dirty set, and if so, add it to the queue. - processing set - - cond *sync.Cond - - shuttingDown bool - - metrics queueMetrics -} - -type empty struct{} -type t interface{} -type set map[t]empty - -func (s set) has(item t) bool { - _, exists := s[item] - return exists -} - -func (s set) insert(item t) { - s[item] = empty{} -} - -func (s set) delete(item t) { - delete(s, item) -} - -// Add marks item as needing processing. -func (q *Type) Add(item interface{}) { - q.cond.L.Lock() - defer q.cond.L.Unlock() - if q.shuttingDown { - return - } - if q.dirty.has(item) { - return - } - - q.metrics.add(item) - - q.dirty.insert(item) - if q.processing.has(item) { - return - } - - q.queue = append(q.queue, item) - q.cond.Signal() -} - -// Len returns the current queue length, for informational purposes only. You -// shouldn't e.g. gate a call to Add() or Get() on Len() being a particular -// value, that can't be synchronized properly. -func (q *Type) Len() int { - q.cond.L.Lock() - defer q.cond.L.Unlock() - return len(q.queue) -} - -// Get blocks until it can return an item to be processed. If shutdown = true, -// the caller should end their goroutine. You must call Done with item when you -// have finished processing it. -func (q *Type) Get() (item interface{}, shutdown bool) { - q.cond.L.Lock() - defer q.cond.L.Unlock() - for len(q.queue) == 0 && !q.shuttingDown { - q.cond.Wait() - } - if len(q.queue) == 0 { - // We must be shutting down. - return nil, true - } - - item, q.queue = q.queue[0], q.queue[1:] - - q.metrics.get(item) - - q.processing.insert(item) - q.dirty.delete(item) - - return item, false -} - -// Done marks item as done processing, and if it has been marked as dirty again -// while it was being processed, it will be re-added to the queue for -// re-processing. -func (q *Type) Done(item interface{}) { - q.cond.L.Lock() - defer q.cond.L.Unlock() - - q.metrics.done(item) - - q.processing.delete(item) - if q.dirty.has(item) { - q.queue = append(q.queue, item) - q.cond.Signal() - } -} - -// ShutDown will cause q to ignore all new items added to it. As soon as the -// worker goroutines have drained the existing items in the queue, they will be -// instructed to exit. -func (q *Type) ShutDown() { - q.cond.L.Lock() - defer q.cond.L.Unlock() - q.shuttingDown = true - q.cond.Broadcast() -} - -func (q *Type) ShuttingDown() bool { - q.cond.L.Lock() - defer q.cond.L.Unlock() - - return q.shuttingDown -} diff --git a/staging/src/k8s.io/client-go/pkg/util/workqueue/rate_limitting_queue.go b/staging/src/k8s.io/client-go/pkg/util/workqueue/rate_limitting_queue.go deleted file mode 100644 index 417ac001b84..00000000000 --- a/staging/src/k8s.io/client-go/pkg/util/workqueue/rate_limitting_queue.go +++ /dev/null @@ -1,69 +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 workqueue - -// RateLimitingInterface is an interface that rate limits items being added to the queue. -type RateLimitingInterface interface { - DelayingInterface - - // AddRateLimited adds an item to the workqueue after the rate limiter says its ok - AddRateLimited(item interface{}) - - // Forget indicates that an item is finished being retried. Doesn't matter whether its for perm failing - // or for success, we'll stop the rate limiter from tracking it. This only clears the `rateLimiter`, you - // still have to call `Done` on the queue. - Forget(item interface{}) - - // NumRequeues returns back how many times the item was requeued - NumRequeues(item interface{}) int -} - -// NewRateLimitingQueue constructs a new workqueue with rateLimited queuing ability -// Remember to call Forget! If you don't, you may end up tracking failures forever. -func NewRateLimitingQueue(rateLimiter RateLimiter) RateLimitingInterface { - return &rateLimitingType{ - DelayingInterface: NewDelayingQueue(), - rateLimiter: rateLimiter, - } -} - -func NewNamedRateLimitingQueue(rateLimiter RateLimiter, name string) RateLimitingInterface { - return &rateLimitingType{ - DelayingInterface: NewNamedDelayingQueue(name), - rateLimiter: rateLimiter, - } -} - -// rateLimitingType wraps an Interface and provides rateLimited re-enquing -type rateLimitingType struct { - DelayingInterface - - rateLimiter RateLimiter -} - -// AddRateLimited AddAfter's the item based on the time when the rate limiter says its ok -func (q *rateLimitingType) AddRateLimited(item interface{}) { - q.DelayingInterface.AddAfter(item, q.rateLimiter.When(item)) -} - -func (q *rateLimitingType) NumRequeues(item interface{}) int { - return q.rateLimiter.NumRequeues(item) -} - -func (q *rateLimitingType) Forget(item interface{}) { - q.rateLimiter.Forget(item) -} diff --git a/staging/src/k8s.io/client-go/pkg/util/workqueue/rate_limitting_queue_test.go b/staging/src/k8s.io/client-go/pkg/util/workqueue/rate_limitting_queue_test.go deleted file mode 100644 index 47100689d10..00000000000 --- a/staging/src/k8s.io/client-go/pkg/util/workqueue/rate_limitting_queue_test.go +++ /dev/null @@ -1,75 +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 workqueue - -import ( - "testing" - "time" - - "k8s.io/client-go/util/clock" -) - -func TestRateLimitingQueue(t *testing.T) { - limiter := NewItemExponentialFailureRateLimiter(1*time.Millisecond, 1*time.Second) - queue := NewRateLimitingQueue(limiter).(*rateLimitingType) - fakeClock := clock.NewFakeClock(time.Now()) - delayingQueue := &delayingType{ - Interface: New(), - clock: fakeClock, - heartbeat: fakeClock.Tick(maxWait), - stopCh: make(chan struct{}), - waitingForAddCh: make(chan waitFor, 1000), - metrics: newRetryMetrics(""), - } - queue.DelayingInterface = delayingQueue - - queue.AddRateLimited("one") - waitEntry := <-delayingQueue.waitingForAddCh - if e, a := 1*time.Millisecond, waitEntry.readyAt.Sub(fakeClock.Now()); e != a { - t.Errorf("expected %v, got %v", e, a) - } - queue.AddRateLimited("one") - waitEntry = <-delayingQueue.waitingForAddCh - if e, a := 2*time.Millisecond, waitEntry.readyAt.Sub(fakeClock.Now()); e != a { - t.Errorf("expected %v, got %v", e, a) - } - if e, a := 2, queue.NumRequeues("one"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - - queue.AddRateLimited("two") - waitEntry = <-delayingQueue.waitingForAddCh - if e, a := 1*time.Millisecond, waitEntry.readyAt.Sub(fakeClock.Now()); e != a { - t.Errorf("expected %v, got %v", e, a) - } - queue.AddRateLimited("two") - waitEntry = <-delayingQueue.waitingForAddCh - if e, a := 2*time.Millisecond, waitEntry.readyAt.Sub(fakeClock.Now()); e != a { - t.Errorf("expected %v, got %v", e, a) - } - - queue.Forget("one") - if e, a := 0, queue.NumRequeues("one"); e != a { - t.Errorf("expected %v, got %v", e, a) - } - queue.AddRateLimited("one") - waitEntry = <-delayingQueue.waitingForAddCh - if e, a := 1*time.Millisecond, waitEntry.readyAt.Sub(fakeClock.Now()); e != a { - t.Errorf("expected %v, got %v", e, a) - } - -} diff --git a/staging/src/k8s.io/client-go/pkg/util/workqueue/timed_queue.go b/staging/src/k8s.io/client-go/pkg/util/workqueue/timed_queue.go deleted file mode 100644 index 2ad90bfdfcf..00000000000 --- a/staging/src/k8s.io/client-go/pkg/util/workqueue/timed_queue.go +++ /dev/null @@ -1,52 +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 workqueue - -import "time" - -type TimedWorkQueue struct { - *Type -} - -type TimedWorkQueueItem struct { - StartTime time.Time - Object interface{} -} - -func NewTimedWorkQueue() *TimedWorkQueue { - return &TimedWorkQueue{New()} -} - -// Add adds the obj along with the current timestamp to the queue. -func (q TimedWorkQueue) Add(timedItem *TimedWorkQueueItem) { - q.Type.Add(timedItem) -} - -// Get gets the obj along with its timestamp from the queue. -func (q TimedWorkQueue) Get() (timedItem *TimedWorkQueueItem, shutdown bool) { - origin, shutdown := q.Type.Get() - if origin == nil { - return nil, shutdown - } - timedItem, _ = origin.(*TimedWorkQueueItem) - return timedItem, shutdown -} - -func (q TimedWorkQueue) Done(timedItem *TimedWorkQueueItem) error { - q.Type.Done(timedItem) - return nil -} diff --git a/staging/src/k8s.io/client-go/pkg/third_party/forked/golang/template/exec.go b/staging/src/k8s.io/client-go/third_party/forked/golang/template/exec.go similarity index 100% rename from staging/src/k8s.io/client-go/pkg/third_party/forked/golang/template/exec.go rename to staging/src/k8s.io/client-go/third_party/forked/golang/template/exec.go diff --git a/staging/src/k8s.io/client-go/pkg/third_party/forked/golang/template/funcs.go b/staging/src/k8s.io/client-go/third_party/forked/golang/template/funcs.go similarity index 100% rename from staging/src/k8s.io/client-go/pkg/third_party/forked/golang/template/funcs.go rename to staging/src/k8s.io/client-go/third_party/forked/golang/template/funcs.go diff --git a/staging/src/k8s.io/client-go/util/jsonpath/jsonpath.go b/staging/src/k8s.io/client-go/util/jsonpath/jsonpath.go index 11a1c521a10..7158de7c70b 100644 --- a/staging/src/k8s.io/client-go/util/jsonpath/jsonpath.go +++ b/staging/src/k8s.io/client-go/util/jsonpath/jsonpath.go @@ -23,7 +23,7 @@ import ( "reflect" "strings" - "k8s.io/client-go/pkg/third_party/forked/golang/template" + "k8s.io/client-go/third_party/forked/golang/template" ) type JSONPath struct { diff --git a/pkg/util/workqueue/default_rate_limiters.go b/staging/src/k8s.io/client-go/util/workqueue/default_rate_limiters.go similarity index 100% rename from pkg/util/workqueue/default_rate_limiters.go rename to staging/src/k8s.io/client-go/util/workqueue/default_rate_limiters.go diff --git a/pkg/util/workqueue/default_rate_limiters_test.go b/staging/src/k8s.io/client-go/util/workqueue/default_rate_limiters_test.go similarity index 100% rename from pkg/util/workqueue/default_rate_limiters_test.go rename to staging/src/k8s.io/client-go/util/workqueue/default_rate_limiters_test.go diff --git a/pkg/util/workqueue/delaying_queue.go b/staging/src/k8s.io/client-go/util/workqueue/delaying_queue.go similarity index 100% rename from pkg/util/workqueue/delaying_queue.go rename to staging/src/k8s.io/client-go/util/workqueue/delaying_queue.go diff --git a/pkg/util/workqueue/delaying_queue_test.go b/staging/src/k8s.io/client-go/util/workqueue/delaying_queue_test.go similarity index 100% rename from pkg/util/workqueue/delaying_queue_test.go rename to staging/src/k8s.io/client-go/util/workqueue/delaying_queue_test.go diff --git a/staging/src/k8s.io/client-go/pkg/util/workqueue/doc.go b/staging/src/k8s.io/client-go/util/workqueue/doc.go similarity index 100% rename from staging/src/k8s.io/client-go/pkg/util/workqueue/doc.go rename to staging/src/k8s.io/client-go/util/workqueue/doc.go diff --git a/pkg/util/workqueue/metrics.go b/staging/src/k8s.io/client-go/util/workqueue/metrics.go similarity index 100% rename from pkg/util/workqueue/metrics.go rename to staging/src/k8s.io/client-go/util/workqueue/metrics.go diff --git a/pkg/util/workqueue/parallelizer.go b/staging/src/k8s.io/client-go/util/workqueue/parallelizer.go similarity index 100% rename from pkg/util/workqueue/parallelizer.go rename to staging/src/k8s.io/client-go/util/workqueue/parallelizer.go diff --git a/pkg/util/workqueue/queue.go b/staging/src/k8s.io/client-go/util/workqueue/queue.go similarity index 100% rename from pkg/util/workqueue/queue.go rename to staging/src/k8s.io/client-go/util/workqueue/queue.go diff --git a/staging/src/k8s.io/client-go/pkg/util/workqueue/queue_test.go b/staging/src/k8s.io/client-go/util/workqueue/queue_test.go similarity index 98% rename from staging/src/k8s.io/client-go/pkg/util/workqueue/queue_test.go rename to staging/src/k8s.io/client-go/util/workqueue/queue_test.go index eb38c1238fd..131f4a2a599 100644 --- a/staging/src/k8s.io/client-go/pkg/util/workqueue/queue_test.go +++ b/staging/src/k8s.io/client-go/util/workqueue/queue_test.go @@ -21,7 +21,7 @@ import ( "testing" "time" - "k8s.io/client-go/pkg/util/workqueue" + "k8s.io/client-go/util/workqueue" ) func TestBasic(t *testing.T) { diff --git a/pkg/util/workqueue/rate_limitting_queue.go b/staging/src/k8s.io/client-go/util/workqueue/rate_limitting_queue.go similarity index 100% rename from pkg/util/workqueue/rate_limitting_queue.go rename to staging/src/k8s.io/client-go/util/workqueue/rate_limitting_queue.go diff --git a/pkg/util/workqueue/rate_limitting_queue_test.go b/staging/src/k8s.io/client-go/util/workqueue/rate_limitting_queue_test.go similarity index 100% rename from pkg/util/workqueue/rate_limitting_queue_test.go rename to staging/src/k8s.io/client-go/util/workqueue/rate_limitting_queue_test.go diff --git a/pkg/util/workqueue/timed_queue.go b/staging/src/k8s.io/client-go/util/workqueue/timed_queue.go similarity index 100% rename from pkg/util/workqueue/timed_queue.go rename to staging/src/k8s.io/client-go/util/workqueue/timed_queue.go diff --git a/staging/src/k8s.io/client-go/pkg/util/workqueue/timed_queue_test.go b/staging/src/k8s.io/client-go/util/workqueue/timed_queue_test.go similarity index 100% rename from staging/src/k8s.io/client-go/pkg/util/workqueue/timed_queue_test.go rename to staging/src/k8s.io/client-go/util/workqueue/timed_queue_test.go diff --git a/test/e2e/BUILD b/test/e2e/BUILD index c5a7c896a06..4f62b91b1c8 100644 --- a/test/e2e/BUILD +++ b/test/e2e/BUILD @@ -143,7 +143,6 @@ go_library( "//pkg/util/logs:go_default_library", "//pkg/util/system:go_default_library", "//pkg/util/version:go_default_library", - "//pkg/util/workqueue:go_default_library", "//pkg/volume/util/volumehelper:go_default_library", "//plugin/pkg/admission/serviceaccount:go_default_library", "//test/e2e/chaosmonkey:go_default_library", @@ -199,6 +198,7 @@ go_library( "//vendor:k8s.io/client-go/tools/cache", "//vendor:k8s.io/client-go/transport", "//vendor:k8s.io/client-go/util/flowcontrol", + "//vendor:k8s.io/client-go/util/workqueue", ], ) diff --git a/test/e2e/density.go b/test/e2e/density.go index 31172b58f48..d9d3c6fb81b 100644 --- a/test/e2e/density.go +++ b/test/e2e/density.go @@ -35,13 +35,13 @@ import ( utiluuid "k8s.io/apimachinery/pkg/util/uuid" "k8s.io/apimachinery/pkg/watch" "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" "k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api/v1" "k8s.io/kubernetes/pkg/apis/batch" "k8s.io/kubernetes/pkg/apis/extensions" "k8s.io/kubernetes/pkg/client/clientset_generated/clientset" "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset" - "k8s.io/kubernetes/pkg/util/workqueue" "k8s.io/kubernetes/test/e2e/framework" testutils "k8s.io/kubernetes/test/utils" diff --git a/test/utils/BUILD b/test/utils/BUILD index 79ddad1de27..da5efe2ec93 100644 --- a/test/utils/BUILD +++ b/test/utils/BUILD @@ -26,7 +26,6 @@ go_library( "//pkg/apis/extensions/v1beta1:go_default_library", "//pkg/client/clientset_generated/clientset:go_default_library", "//pkg/client/clientset_generated/internalclientset:go_default_library", - "//pkg/util/workqueue:go_default_library", "//vendor:github.com/golang/glog", "//vendor:k8s.io/apimachinery/pkg/api/equality", "//vendor:k8s.io/apimachinery/pkg/api/errors", @@ -42,6 +41,7 @@ go_library( "//vendor:k8s.io/apimachinery/pkg/util/wait", "//vendor:k8s.io/apimachinery/pkg/watch", "//vendor:k8s.io/client-go/tools/cache", + "//vendor:k8s.io/client-go/util/workqueue", ], ) diff --git a/test/utils/runners.go b/test/utils/runners.go index 2778e021d80..f53b0f97d17 100644 --- a/test/utils/runners.go +++ b/test/utils/runners.go @@ -34,6 +34,7 @@ import ( "k8s.io/apimachinery/pkg/util/sets" "k8s.io/apimachinery/pkg/util/uuid" "k8s.io/apimachinery/pkg/util/wait" + "k8s.io/client-go/util/workqueue" "k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api/v1" batchinternal "k8s.io/kubernetes/pkg/apis/batch" @@ -42,7 +43,6 @@ import ( extensions "k8s.io/kubernetes/pkg/apis/extensions/v1beta1" "k8s.io/kubernetes/pkg/client/clientset_generated/clientset" "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset" - "k8s.io/kubernetes/pkg/util/workqueue" "github.com/golang/glog" ) diff --git a/vendor/BUILD b/vendor/BUILD index 69dd02277f7..ca484fa87f9 100644 --- a/vendor/BUILD +++ b/vendor/BUILD @@ -12107,34 +12107,12 @@ go_library( ], ) -go_library( - name = "k8s.io/client-go/pkg/third_party/forked/golang/json", - srcs = ["k8s.io/client-go/pkg/third_party/forked/golang/json/fields.go"], - tags = ["automanaged"], -) - -go_library( - name = "k8s.io/client-go/pkg/third_party/forked/golang/netutil", - srcs = ["k8s.io/client-go/pkg/third_party/forked/golang/netutil/addr.go"], - tags = ["automanaged"], -) - -go_library( - name = "k8s.io/client-go/pkg/third_party/forked/golang/template", - srcs = [ - "k8s.io/client-go/pkg/third_party/forked/golang/template/exec.go", - "k8s.io/client-go/pkg/third_party/forked/golang/template/funcs.go", - ], - tags = ["automanaged"], -) - go_library( name = "k8s.io/client-go/pkg/util", srcs = [ "k8s.io/client-go/pkg/util/doc.go", "k8s.io/client-go/pkg/util/template.go", "k8s.io/client-go/pkg/util/trace.go", - "k8s.io/client-go/pkg/util/trie.go", "k8s.io/client-go/pkg/util/umask.go", "k8s.io/client-go/pkg/util/util.go", ], @@ -12165,9 +12143,9 @@ go_library( "//vendor:k8s.io/apimachinery/pkg/api/errors", "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1", "//vendor:k8s.io/apimachinery/pkg/util/runtime", + "//vendor:k8s.io/apimachinery/third_party/forked/golang/netutil", "//vendor:k8s.io/client-go/pkg/api", "//vendor:k8s.io/client-go/pkg/util/httpstream", - "//vendor:k8s.io/client-go/third_party/forked/golang/netutil", ], ) @@ -12204,62 +12182,6 @@ go_library( deps = ["//vendor:github.com/docker/distribution/reference"], ) -go_library( - name = "k8s.io/client-go/pkg/util/strategicpatch", - srcs = ["k8s.io/client-go/pkg/util/strategicpatch/patch.go"], - tags = ["automanaged"], - deps = [ - "//vendor:github.com/davecgh/go-spew/spew", - "//vendor:github.com/ghodss/yaml", - "//vendor:k8s.io/apimachinery/pkg/util/json", - "//vendor:k8s.io/client-go/third_party/forked/golang/json", - ], -) - -go_test( - name = "k8s.io/client-go/pkg/util/workqueue_test", - srcs = [ - "k8s.io/client-go/pkg/util/workqueue/default_rate_limiters_test.go", - "k8s.io/client-go/pkg/util/workqueue/delaying_queue_test.go", - "k8s.io/client-go/pkg/util/workqueue/rate_limitting_queue_test.go", - "k8s.io/client-go/pkg/util/workqueue/timed_queue_test.go", - ], - library = ":k8s.io/client-go/pkg/util/workqueue", - tags = ["automanaged"], - deps = [ - "//vendor:k8s.io/apimachinery/pkg/util/wait", - "//vendor:k8s.io/client-go/pkg/api/v1", - "//vendor:k8s.io/client-go/util/clock", - ], -) - -go_library( - name = "k8s.io/client-go/pkg/util/workqueue", - srcs = [ - "k8s.io/client-go/pkg/util/workqueue/default_rate_limiters.go", - "k8s.io/client-go/pkg/util/workqueue/delaying_queue.go", - "k8s.io/client-go/pkg/util/workqueue/doc.go", - "k8s.io/client-go/pkg/util/workqueue/metrics.go", - "k8s.io/client-go/pkg/util/workqueue/parallelizer.go", - "k8s.io/client-go/pkg/util/workqueue/queue.go", - "k8s.io/client-go/pkg/util/workqueue/rate_limitting_queue.go", - "k8s.io/client-go/pkg/util/workqueue/timed_queue.go", - ], - tags = ["automanaged"], - deps = [ - "//vendor:github.com/juju/ratelimit", - "//vendor:k8s.io/apimachinery/pkg/util/runtime", - "//vendor:k8s.io/client-go/util/clock", - ], -) - -go_test( - name = "k8s.io/client-go/pkg/util/workqueue_xtest", - srcs = ["k8s.io/client-go/pkg/util/workqueue/queue_test.go"], - tags = ["automanaged"], - deps = ["//vendor:k8s.io/client-go/pkg/util/workqueue"], -) - go_library( name = "k8s.io/client-go/pkg/version", srcs = [ @@ -12745,9 +12667,9 @@ go_test( "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1", "//vendor:k8s.io/apimachinery/pkg/runtime", "//vendor:k8s.io/apimachinery/pkg/util/diff", + "//vendor:k8s.io/apimachinery/pkg/util/strategicpatch", "//vendor:k8s.io/client-go/pkg/api/install", "//vendor:k8s.io/client-go/pkg/api/v1", - "//vendor:k8s.io/client-go/pkg/util/strategicpatch", "//vendor:k8s.io/client-go/rest", "//vendor:k8s.io/client-go/util/clock", ], @@ -12770,9 +12692,9 @@ go_library( "//vendor:k8s.io/apimachinery/pkg/runtime", "//vendor:k8s.io/apimachinery/pkg/util/runtime", "//vendor:k8s.io/apimachinery/pkg/util/sets", + "//vendor:k8s.io/apimachinery/pkg/util/strategicpatch", "//vendor:k8s.io/apimachinery/pkg/watch", "//vendor:k8s.io/client-go/pkg/api/v1", - "//vendor:k8s.io/client-go/pkg/util/strategicpatch", "//vendor:k8s.io/client-go/rest", "//vendor:k8s.io/client-go/util/clock", ], @@ -13205,18 +13127,6 @@ go_library( ], ) -go_library( - name = "k8s.io/client-go/third_party/forked/golang/json", - srcs = ["k8s.io/client-go/third_party/forked/golang/json/fields.go"], - tags = ["automanaged"], -) - -go_library( - name = "k8s.io/client-go/third_party/forked/golang/netutil", - srcs = ["k8s.io/client-go/third_party/forked/golang/netutil/addr.go"], - tags = ["automanaged"], -) - go_library( name = "github.com/armon/circbuf", srcs = ["github.com/armon/circbuf/circbuf.go"], @@ -13355,7 +13265,7 @@ go_library( "k8s.io/client-go/util/jsonpath/parser.go", ], tags = ["automanaged"], - deps = ["//vendor:k8s.io/client-go/pkg/third_party/forked/golang/template"], + deps = ["//vendor:k8s.io/client-go/third_party/forked/golang/template"], ) go_test( @@ -13976,3 +13886,80 @@ go_library( tags = ["automanaged"], deps = ["//vendor:github.com/golang/glog"], ) + +go_library( + name = "k8s.io/client-go/_vendor/k8s.io/apimachinery/pkg/util/strategicpatch", + srcs = ["k8s.io/client-go/_vendor/k8s.io/apimachinery/pkg/util/strategicpatch/patch.go"], + tags = ["automanaged"], + deps = [ + "//vendor:github.com/davecgh/go-spew/spew", + "//vendor:github.com/ghodss/yaml", + "//vendor:k8s.io/apimachinery/pkg/util/json", + "//vendor:k8s.io/apimachinery/third_party/forked/golang/json", + ], +) + +go_library( + name = "k8s.io/client-go/_vendor/k8s.io/apimachinery/third_party/forked/golang/json", + srcs = ["k8s.io/client-go/_vendor/k8s.io/apimachinery/third_party/forked/golang/json/fields.go"], + tags = ["automanaged"], +) + +go_library( + name = "k8s.io/client-go/_vendor/k8s.io/apimachinery/third_party/forked/golang/netutil", + srcs = ["k8s.io/client-go/_vendor/k8s.io/apimachinery/third_party/forked/golang/netutil/addr.go"], + tags = ["automanaged"], +) + +go_library( + name = "k8s.io/client-go/third_party/forked/golang/template", + srcs = [ + "k8s.io/client-go/third_party/forked/golang/template/exec.go", + "k8s.io/client-go/third_party/forked/golang/template/funcs.go", + ], + tags = ["automanaged"], +) + +go_test( + name = "k8s.io/client-go/util/workqueue_test", + srcs = [ + "k8s.io/client-go/util/workqueue/default_rate_limiters_test.go", + "k8s.io/client-go/util/workqueue/delaying_queue_test.go", + "k8s.io/client-go/util/workqueue/rate_limitting_queue_test.go", + "k8s.io/client-go/util/workqueue/timed_queue_test.go", + ], + library = ":k8s.io/client-go/util/workqueue", + tags = ["automanaged"], + deps = [ + "//vendor:k8s.io/apimachinery/pkg/util/wait", + "//vendor:k8s.io/client-go/pkg/api/v1", + "//vendor:k8s.io/client-go/util/clock", + ], +) + +go_library( + name = "k8s.io/client-go/util/workqueue", + srcs = [ + "k8s.io/client-go/util/workqueue/default_rate_limiters.go", + "k8s.io/client-go/util/workqueue/delaying_queue.go", + "k8s.io/client-go/util/workqueue/doc.go", + "k8s.io/client-go/util/workqueue/metrics.go", + "k8s.io/client-go/util/workqueue/parallelizer.go", + "k8s.io/client-go/util/workqueue/queue.go", + "k8s.io/client-go/util/workqueue/rate_limitting_queue.go", + "k8s.io/client-go/util/workqueue/timed_queue.go", + ], + tags = ["automanaged"], + deps = [ + "//vendor:github.com/juju/ratelimit", + "//vendor:k8s.io/apimachinery/pkg/util/runtime", + "//vendor:k8s.io/client-go/util/clock", + ], +) + +go_test( + name = "k8s.io/client-go/util/workqueue_xtest", + srcs = ["k8s.io/client-go/util/workqueue/queue_test.go"], + tags = ["automanaged"], + deps = ["//vendor:k8s.io/client-go/util/workqueue"], +)