mirror of
https://github.com/ceph/ceph-csi.git
synced 2025-06-13 18:43:34 +00:00
vendor cleanup: remove unused,non-go and test files
This commit is contained in:
Madhu Rajanna
92
vendor/k8s.io/kubernetes/pkg/controller/nodelifecycle/BUILD
generated
vendored
92
vendor/k8s.io/kubernetes/pkg/controller/nodelifecycle/BUILD
generated
vendored
@ -1,92 +0,0 @@
|
||||
load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
|
||||
|
||||
go_library(
|
||||
name = "go_default_library",
|
||||
srcs = [
|
||||
"metrics.go",
|
||||
"node_lifecycle_controller.go",
|
||||
],
|
||||
importpath = "k8s.io/kubernetes/pkg/controller/nodelifecycle",
|
||||
visibility = ["//visibility:public"],
|
||||
deps = [
|
||||
"//pkg/api/v1/node:go_default_library",
|
||||
"//pkg/cloudprovider:go_default_library",
|
||||
"//pkg/controller:go_default_library",
|
||||
"//pkg/controller/nodelifecycle/scheduler:go_default_library",
|
||||
"//pkg/controller/util/node:go_default_library",
|
||||
"//pkg/scheduler/algorithm:go_default_library",
|
||||
"//pkg/util/metrics:go_default_library",
|
||||
"//pkg/util/node:go_default_library",
|
||||
"//pkg/util/system:go_default_library",
|
||||
"//pkg/util/taints:go_default_library",
|
||||
"//pkg/util/version:go_default_library",
|
||||
"//vendor/github.com/golang/glog:go_default_library",
|
||||
"//vendor/github.com/prometheus/client_golang/prometheus:go_default_library",
|
||||
"//vendor/k8s.io/api/core/v1:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/api/equality:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/api/errors:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/labels:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/util/runtime:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
|
||||
"//vendor/k8s.io/client-go/informers/core/v1:go_default_library",
|
||||
"//vendor/k8s.io/client-go/informers/extensions/v1beta1:go_default_library",
|
||||
"//vendor/k8s.io/client-go/kubernetes:go_default_library",
|
||||
"//vendor/k8s.io/client-go/kubernetes/scheme:go_default_library",
|
||||
"//vendor/k8s.io/client-go/listers/core/v1:go_default_library",
|
||||
"//vendor/k8s.io/client-go/listers/extensions/v1beta1:go_default_library",
|
||||
"//vendor/k8s.io/client-go/tools/cache:go_default_library",
|
||||
"//vendor/k8s.io/client-go/tools/record:go_default_library",
|
||||
"//vendor/k8s.io/client-go/util/flowcontrol:go_default_library",
|
||||
],
|
||||
)
|
||||
|
||||
filegroup(
|
||||
name = "package-srcs",
|
||||
srcs = glob(["**"]),
|
||||
tags = ["automanaged"],
|
||||
visibility = ["//visibility:private"],
|
||||
)
|
||||
|
||||
filegroup(
|
||||
name = "all-srcs",
|
||||
srcs = [
|
||||
":package-srcs",
|
||||
"//pkg/controller/nodelifecycle/scheduler:all-srcs",
|
||||
],
|
||||
tags = ["automanaged"],
|
||||
visibility = ["//visibility:public"],
|
||||
)
|
||||
|
||||
go_test(
|
||||
name = "go_default_test",
|
||||
srcs = ["node_lifecycle_controller_test.go"],
|
||||
embed = [":go_default_library"],
|
||||
deps = [
|
||||
"//pkg/cloudprovider:go_default_library",
|
||||
"//pkg/cloudprovider/providers/fake:go_default_library",
|
||||
"//pkg/controller:go_default_library",
|
||||
"//pkg/controller/nodelifecycle/scheduler:go_default_library",
|
||||
"//pkg/controller/testutil:go_default_library",
|
||||
"//pkg/controller/util/node:go_default_library",
|
||||
"//pkg/kubelet/apis:go_default_library",
|
||||
"//pkg/scheduler/algorithm:go_default_library",
|
||||
"//pkg/util/node:go_default_library",
|
||||
"//pkg/util/taints:go_default_library",
|
||||
"//vendor/k8s.io/api/core/v1:go_default_library",
|
||||
"//vendor/k8s.io/api/extensions/v1beta1:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/api/equality:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/api/resource:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/util/diff:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
|
||||
"//vendor/k8s.io/client-go/informers:go_default_library",
|
||||
"//vendor/k8s.io/client-go/informers/core/v1:go_default_library",
|
||||
"//vendor/k8s.io/client-go/informers/extensions/v1beta1:go_default_library",
|
||||
"//vendor/k8s.io/client-go/kubernetes:go_default_library",
|
||||
"//vendor/k8s.io/client-go/kubernetes/fake:go_default_library",
|
||||
"//vendor/k8s.io/client-go/testing:go_default_library",
|
||||
],
|
||||
)
|
||||
9
vendor/k8s.io/kubernetes/pkg/controller/nodelifecycle/OWNERS
generated
vendored
9
vendor/k8s.io/kubernetes/pkg/controller/nodelifecycle/OWNERS
generated
vendored
@ -1,9 +0,0 @@
|
||||
approvers:
|
||||
- gmarek
|
||||
- bowei
|
||||
reviewers:
|
||||
- gmarek
|
||||
- smarterclayton
|
||||
- ingvagabund
|
||||
- aveshagarwal
|
||||
- k82cn
|
||||
78
vendor/k8s.io/kubernetes/pkg/controller/nodelifecycle/metrics.go
generated
vendored
78
vendor/k8s.io/kubernetes/pkg/controller/nodelifecycle/metrics.go
generated
vendored
@ -1,78 +0,0 @@
|
||||
/*
|
||||
Copyright 2017 The Kubernetes Authors.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
package nodelifecycle
|
||||
|
||||
import (
|
||||
"sync"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
)
|
||||
|
||||
const (
|
||||
nodeControllerSubsystem = "node_collector"
|
||||
zoneHealthStatisticKey = "zone_health"
|
||||
zoneSizeKey = "zone_size"
|
||||
zoneNoUnhealthyNodesKey = "unhealthy_nodes_in_zone"
|
||||
evictionsNumberKey = "evictions_number"
|
||||
)
|
||||
|
||||
var (
|
||||
zoneHealth = prometheus.NewGaugeVec(
|
||||
prometheus.GaugeOpts{
|
||||
Subsystem: nodeControllerSubsystem,
|
||||
Name: zoneHealthStatisticKey,
|
||||
Help: "Gauge measuring percentage of healthy nodes per zone.",
|
||||
},
|
||||
[]string{"zone"},
|
||||
)
|
||||
zoneSize = prometheus.NewGaugeVec(
|
||||
prometheus.GaugeOpts{
|
||||
Subsystem: nodeControllerSubsystem,
|
||||
Name: zoneSizeKey,
|
||||
Help: "Gauge measuring number of registered Nodes per zones.",
|
||||
},
|
||||
[]string{"zone"},
|
||||
)
|
||||
unhealthyNodes = prometheus.NewGaugeVec(
|
||||
prometheus.GaugeOpts{
|
||||
Subsystem: nodeControllerSubsystem,
|
||||
Name: zoneNoUnhealthyNodesKey,
|
||||
Help: "Gauge measuring number of not Ready Nodes per zones.",
|
||||
},
|
||||
[]string{"zone"},
|
||||
)
|
||||
evictionsNumber = prometheus.NewCounterVec(
|
||||
prometheus.CounterOpts{
|
||||
Subsystem: nodeControllerSubsystem,
|
||||
Name: evictionsNumberKey,
|
||||
Help: "Number of Node evictions that happened since current instance of NodeController started.",
|
||||
},
|
||||
[]string{"zone"},
|
||||
)
|
||||
)
|
||||
|
||||
var registerMetrics sync.Once
|
||||
|
||||
// Register the metrics that are to be monitored.
|
||||
func Register() {
|
||||
registerMetrics.Do(func() {
|
||||
prometheus.MustRegister(zoneHealth)
|
||||
prometheus.MustRegister(zoneSize)
|
||||
prometheus.MustRegister(unhealthyNodes)
|
||||
prometheus.MustRegister(evictionsNumber)
|
||||
})
|
||||
}
|
||||
1178
vendor/k8s.io/kubernetes/pkg/controller/nodelifecycle/node_lifecycle_controller.go
generated
vendored
1178
vendor/k8s.io/kubernetes/pkg/controller/nodelifecycle/node_lifecycle_controller.go
generated
vendored
File diff suppressed because it is too large
Load Diff
2538
vendor/k8s.io/kubernetes/pkg/controller/nodelifecycle/node_lifecycle_controller_test.go
generated
vendored
2538
vendor/k8s.io/kubernetes/pkg/controller/nodelifecycle/node_lifecycle_controller_test.go
generated
vendored
File diff suppressed because it is too large
Load Diff
62
vendor/k8s.io/kubernetes/pkg/controller/nodelifecycle/scheduler/BUILD
generated
vendored
62
vendor/k8s.io/kubernetes/pkg/controller/nodelifecycle/scheduler/BUILD
generated
vendored
@ -1,62 +0,0 @@
|
||||
load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
|
||||
|
||||
go_library(
|
||||
name = "go_default_library",
|
||||
srcs = [
|
||||
"rate_limited_queue.go",
|
||||
"taint_manager.go",
|
||||
"timed_workers.go",
|
||||
],
|
||||
importpath = "k8s.io/kubernetes/pkg/controller/nodelifecycle/scheduler",
|
||||
visibility = ["//visibility:public"],
|
||||
deps = [
|
||||
"//pkg/apis/core/helper:go_default_library",
|
||||
"//pkg/apis/core/v1/helper:go_default_library",
|
||||
"//vendor/github.com/golang/glog:go_default_library",
|
||||
"//vendor/k8s.io/api/core/v1:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/fields:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/labels:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
|
||||
"//vendor/k8s.io/client-go/kubernetes:go_default_library",
|
||||
"//vendor/k8s.io/client-go/kubernetes/scheme:go_default_library",
|
||||
"//vendor/k8s.io/client-go/kubernetes/typed/core/v1:go_default_library",
|
||||
"//vendor/k8s.io/client-go/tools/record:go_default_library",
|
||||
"//vendor/k8s.io/client-go/util/flowcontrol:go_default_library",
|
||||
"//vendor/k8s.io/client-go/util/workqueue:go_default_library",
|
||||
],
|
||||
)
|
||||
|
||||
go_test(
|
||||
name = "go_default_test",
|
||||
srcs = [
|
||||
"rate_limited_queue_test.go",
|
||||
"taint_manager_test.go",
|
||||
"timed_workers_test.go",
|
||||
],
|
||||
embed = [":go_default_library"],
|
||||
deps = [
|
||||
"//pkg/controller/testutil:go_default_library",
|
||||
"//vendor/k8s.io/api/core/v1:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
|
||||
"//vendor/k8s.io/client-go/kubernetes/fake:go_default_library",
|
||||
"//vendor/k8s.io/client-go/testing:go_default_library",
|
||||
"//vendor/k8s.io/client-go/util/flowcontrol:go_default_library",
|
||||
],
|
||||
)
|
||||
|
||||
filegroup(
|
||||
name = "package-srcs",
|
||||
srcs = glob(["**"]),
|
||||
tags = ["automanaged"],
|
||||
visibility = ["//visibility:private"],
|
||||
)
|
||||
|
||||
filegroup(
|
||||
name = "all-srcs",
|
||||
srcs = [":package-srcs"],
|
||||
tags = ["automanaged"],
|
||||
visibility = ["//visibility:public"],
|
||||
)
|
||||
309
vendor/k8s.io/kubernetes/pkg/controller/nodelifecycle/scheduler/rate_limited_queue.go
generated
vendored
309
vendor/k8s.io/kubernetes/pkg/controller/nodelifecycle/scheduler/rate_limited_queue.go
generated
vendored
@ -1,309 +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 scheduler
|
||||
|
||||
import (
|
||||
"container/heap"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"k8s.io/apimachinery/pkg/util/sets"
|
||||
"k8s.io/client-go/util/flowcontrol"
|
||||
|
||||
"github.com/golang/glog"
|
||||
)
|
||||
|
||||
const (
|
||||
// NodeStatusUpdateRetry controls the number of retries of writing
|
||||
// NodeStatus update.
|
||||
NodeStatusUpdateRetry = 5
|
||||
// NodeEvictionPeriod controls how often NodeController will try to
|
||||
// evict Pods from non-responsive Nodes.
|
||||
NodeEvictionPeriod = 100 * time.Millisecond
|
||||
// EvictionRateLimiterBurst is the burst value for all eviction rate
|
||||
// limiters
|
||||
EvictionRateLimiterBurst = 1
|
||||
)
|
||||
|
||||
// TimedValue is a value that should be processed at a designated time.
|
||||
type TimedValue struct {
|
||||
Value string
|
||||
// UID could be anything that helps identify the value
|
||||
UID interface{}
|
||||
AddedAt time.Time
|
||||
ProcessAt time.Time
|
||||
}
|
||||
|
||||
// now is used to test time
|
||||
var now = time.Now
|
||||
|
||||
// TimedQueue is a priority heap where the lowest ProcessAt is at the front of the queue
|
||||
type TimedQueue []*TimedValue
|
||||
|
||||
// Len is the length of the queue.
|
||||
func (h TimedQueue) Len() int { return len(h) }
|
||||
|
||||
// Less returns true if queue[i] < queue[j].
|
||||
func (h TimedQueue) Less(i, j int) bool { return h[i].ProcessAt.Before(h[j].ProcessAt) }
|
||||
|
||||
// Swap swaps index i and j.
|
||||
func (h TimedQueue) Swap(i, j int) { h[i], h[j] = h[j], h[i] }
|
||||
|
||||
// Push a new TimedValue on to the queue.
|
||||
func (h *TimedQueue) Push(x interface{}) {
|
||||
*h = append(*h, x.(*TimedValue))
|
||||
}
|
||||
|
||||
// Pop the lowest ProcessAt item.
|
||||
func (h *TimedQueue) Pop() interface{} {
|
||||
old := *h
|
||||
n := len(old)
|
||||
x := old[n-1]
|
||||
*h = old[0 : n-1]
|
||||
return x
|
||||
}
|
||||
|
||||
// UniqueQueue is a FIFO queue which additionally guarantees that any
|
||||
// element can be added only once until it is removed.
|
||||
type UniqueQueue struct {
|
||||
lock sync.Mutex
|
||||
queue TimedQueue
|
||||
set sets.String
|
||||
}
|
||||
|
||||
// Add a new value to the queue if it wasn't added before, or was
|
||||
// explicitly removed by the Remove call. Returns true if new value
|
||||
// was added.
|
||||
func (q *UniqueQueue) Add(value TimedValue) bool {
|
||||
q.lock.Lock()
|
||||
defer q.lock.Unlock()
|
||||
|
||||
if q.set.Has(value.Value) {
|
||||
return false
|
||||
}
|
||||
heap.Push(&q.queue, &value)
|
||||
q.set.Insert(value.Value)
|
||||
return true
|
||||
}
|
||||
|
||||
// Replace replaces an existing value in the queue if it already
|
||||
// exists, otherwise it does nothing. Returns true if the item was
|
||||
// found.
|
||||
func (q *UniqueQueue) Replace(value TimedValue) bool {
|
||||
q.lock.Lock()
|
||||
defer q.lock.Unlock()
|
||||
|
||||
for i := range q.queue {
|
||||
if q.queue[i].Value != value.Value {
|
||||
continue
|
||||
}
|
||||
heap.Remove(&q.queue, i)
|
||||
heap.Push(&q.queue, &value)
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// RemoveFromQueue the value from the queue, but keeps it in the set,
|
||||
// so it won't be added second time. Returns true if something was
|
||||
// removed.
|
||||
func (q *UniqueQueue) RemoveFromQueue(value string) bool {
|
||||
q.lock.Lock()
|
||||
defer q.lock.Unlock()
|
||||
|
||||
if !q.set.Has(value) {
|
||||
return false
|
||||
}
|
||||
for i, val := range q.queue {
|
||||
if val.Value == value {
|
||||
heap.Remove(&q.queue, i)
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// Remove the value from the queue, so Get() call won't return it, and
|
||||
// allow subsequent addition of the given value. If the value is not
|
||||
// present does nothing and returns false.
|
||||
func (q *UniqueQueue) Remove(value string) bool {
|
||||
q.lock.Lock()
|
||||
defer q.lock.Unlock()
|
||||
|
||||
if !q.set.Has(value) {
|
||||
return false
|
||||
}
|
||||
q.set.Delete(value)
|
||||
for i, val := range q.queue {
|
||||
if val.Value == value {
|
||||
heap.Remove(&q.queue, i)
|
||||
return true
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// Get returns the oldest added value that wasn't returned yet.
|
||||
func (q *UniqueQueue) Get() (TimedValue, bool) {
|
||||
q.lock.Lock()
|
||||
defer q.lock.Unlock()
|
||||
if len(q.queue) == 0 {
|
||||
return TimedValue{}, false
|
||||
}
|
||||
result := heap.Pop(&q.queue).(*TimedValue)
|
||||
q.set.Delete(result.Value)
|
||||
return *result, true
|
||||
}
|
||||
|
||||
// Head returns the oldest added value that wasn't returned yet
|
||||
// without removing it.
|
||||
func (q *UniqueQueue) Head() (TimedValue, bool) {
|
||||
q.lock.Lock()
|
||||
defer q.lock.Unlock()
|
||||
if len(q.queue) == 0 {
|
||||
return TimedValue{}, false
|
||||
}
|
||||
result := q.queue[0]
|
||||
return *result, true
|
||||
}
|
||||
|
||||
// Clear removes all items from the queue and duplication preventing
|
||||
// set.
|
||||
func (q *UniqueQueue) Clear() {
|
||||
q.lock.Lock()
|
||||
defer q.lock.Unlock()
|
||||
if q.queue.Len() > 0 {
|
||||
q.queue = make(TimedQueue, 0)
|
||||
}
|
||||
if len(q.set) > 0 {
|
||||
q.set = sets.NewString()
|
||||
}
|
||||
}
|
||||
|
||||
// RateLimitedTimedQueue is a unique item priority queue ordered by
|
||||
// the expected next time of execution. It is also rate limited.
|
||||
type RateLimitedTimedQueue struct {
|
||||
queue UniqueQueue
|
||||
limiterLock sync.Mutex
|
||||
limiter flowcontrol.RateLimiter
|
||||
}
|
||||
|
||||
// NewRateLimitedTimedQueue creates new queue which will use given
|
||||
// RateLimiter to oversee execution.
|
||||
func NewRateLimitedTimedQueue(limiter flowcontrol.RateLimiter) *RateLimitedTimedQueue {
|
||||
return &RateLimitedTimedQueue{
|
||||
queue: UniqueQueue{
|
||||
queue: TimedQueue{},
|
||||
set: sets.NewString(),
|
||||
},
|
||||
limiter: limiter,
|
||||
}
|
||||
}
|
||||
|
||||
// ActionFunc takes a timed value and returns false if the item must
|
||||
// be retried, with an optional time.Duration if some minimum wait
|
||||
// interval should be used.
|
||||
type ActionFunc func(TimedValue) (bool, time.Duration)
|
||||
|
||||
// Try processes the queue.Ends prematurely if RateLimiter forbids an
|
||||
// action and leak is true. Otherwise, requeues the item to be
|
||||
// processed. Each value is processed once if fn returns true,
|
||||
// otherwise it is added back to the queue. The returned remaining is
|
||||
// used to identify the minimum time to execute the next item in the
|
||||
// queue. The same value is processed only once unless Remove is
|
||||
// explicitly called on it (it's done by the cancelPodEviction
|
||||
// function in NodeController when Node becomes Ready again) TODO:
|
||||
// figure out a good way to do garbage collection for all Nodes that
|
||||
// were removed from the cluster.
|
||||
func (q *RateLimitedTimedQueue) Try(fn ActionFunc) {
|
||||
val, ok := q.queue.Head()
|
||||
q.limiterLock.Lock()
|
||||
defer q.limiterLock.Unlock()
|
||||
for ok {
|
||||
// rate limit the queue checking
|
||||
if !q.limiter.TryAccept() {
|
||||
glog.V(10).Infof("Try rate limited for value: %v", val)
|
||||
// Try again later
|
||||
break
|
||||
}
|
||||
|
||||
now := now()
|
||||
if now.Before(val.ProcessAt) {
|
||||
break
|
||||
}
|
||||
|
||||
if ok, wait := fn(val); !ok {
|
||||
val.ProcessAt = now.Add(wait + 1)
|
||||
q.queue.Replace(val)
|
||||
} else {
|
||||
q.queue.RemoveFromQueue(val.Value)
|
||||
}
|
||||
val, ok = q.queue.Head()
|
||||
}
|
||||
}
|
||||
|
||||
// Add value to the queue to be processed. Won't add the same
|
||||
// value(comparison by value) a second time if it was already added
|
||||
// and not removed.
|
||||
func (q *RateLimitedTimedQueue) Add(value string, uid interface{}) bool {
|
||||
now := now()
|
||||
return q.queue.Add(TimedValue{
|
||||
Value: value,
|
||||
UID: uid,
|
||||
AddedAt: now,
|
||||
ProcessAt: now,
|
||||
})
|
||||
}
|
||||
|
||||
// Remove Node from the Evictor. The Node won't be processed until
|
||||
// added again.
|
||||
func (q *RateLimitedTimedQueue) Remove(value string) bool {
|
||||
return q.queue.Remove(value)
|
||||
}
|
||||
|
||||
// Clear removes all items from the queue
|
||||
func (q *RateLimitedTimedQueue) Clear() {
|
||||
q.queue.Clear()
|
||||
}
|
||||
|
||||
// SwapLimiter safely swaps current limiter for this queue with the
|
||||
// passed one if capacities or qps's differ.
|
||||
func (q *RateLimitedTimedQueue) SwapLimiter(newQPS float32) {
|
||||
q.limiterLock.Lock()
|
||||
defer q.limiterLock.Unlock()
|
||||
if q.limiter.QPS() == newQPS {
|
||||
return
|
||||
}
|
||||
var newLimiter flowcontrol.RateLimiter
|
||||
if newQPS <= 0 {
|
||||
newLimiter = flowcontrol.NewFakeNeverRateLimiter()
|
||||
} else {
|
||||
newLimiter = flowcontrol.NewTokenBucketRateLimiter(newQPS, EvictionRateLimiterBurst)
|
||||
|
||||
// If we're currently waiting on limiter, we drain the new one - this is a good approach when Burst value is 1
|
||||
// TODO: figure out if we need to support higher Burst values and decide on the drain logic, should we keep:
|
||||
// - saturation (percentage of used tokens)
|
||||
// - number of used tokens
|
||||
// - number of available tokens
|
||||
// - something else
|
||||
if q.limiter.TryAccept() == false {
|
||||
newLimiter.TryAccept()
|
||||
}
|
||||
}
|
||||
q.limiter.Stop()
|
||||
q.limiter = newLimiter
|
||||
}
|
||||
334
vendor/k8s.io/kubernetes/pkg/controller/nodelifecycle/scheduler/rate_limited_queue_test.go
generated
vendored
334
vendor/k8s.io/kubernetes/pkg/controller/nodelifecycle/scheduler/rate_limited_queue_test.go
generated
vendored
@ -1,334 +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 scheduler
|
||||
|
||||
import (
|
||||
"reflect"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"k8s.io/apimachinery/pkg/util/sets"
|
||||
"k8s.io/client-go/util/flowcontrol"
|
||||
)
|
||||
|
||||
func CheckQueueEq(lhs []string, rhs TimedQueue) bool {
|
||||
for i := 0; i < len(lhs); i++ {
|
||||
if rhs[i].Value != lhs[i] {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func CheckSetEq(lhs, rhs sets.String) bool {
|
||||
return lhs.HasAll(rhs.List()...) && rhs.HasAll(lhs.List()...)
|
||||
}
|
||||
|
||||
func TestAddNode(t *testing.T) {
|
||||
evictor := NewRateLimitedTimedQueue(flowcontrol.NewFakeAlwaysRateLimiter())
|
||||
evictor.Add("first", "11111")
|
||||
evictor.Add("second", "22222")
|
||||
evictor.Add("third", "33333")
|
||||
|
||||
queuePattern := []string{"first", "second", "third"}
|
||||
if len(evictor.queue.queue) != len(queuePattern) {
|
||||
t.Fatalf("Queue %v should have length %d", evictor.queue.queue, len(queuePattern))
|
||||
}
|
||||
if !CheckQueueEq(queuePattern, evictor.queue.queue) {
|
||||
t.Errorf("Invalid queue. Got %v, expected %v", evictor.queue.queue, queuePattern)
|
||||
}
|
||||
|
||||
setPattern := sets.NewString("first", "second", "third")
|
||||
if len(evictor.queue.set) != len(setPattern) {
|
||||
t.Fatalf("Map %v should have length %d", evictor.queue.set, len(setPattern))
|
||||
}
|
||||
if !CheckSetEq(setPattern, evictor.queue.set) {
|
||||
t.Errorf("Invalid map. Got %v, expected %v", evictor.queue.set, setPattern)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDelNode(t *testing.T) {
|
||||
defer func() { now = time.Now }()
|
||||
var tick int64
|
||||
now = func() time.Time {
|
||||
t := time.Unix(tick, 0)
|
||||
tick++
|
||||
return t
|
||||
}
|
||||
evictor := NewRateLimitedTimedQueue(flowcontrol.NewFakeAlwaysRateLimiter())
|
||||
evictor.Add("first", "11111")
|
||||
evictor.Add("second", "22222")
|
||||
evictor.Add("third", "33333")
|
||||
evictor.Remove("first")
|
||||
|
||||
queuePattern := []string{"second", "third"}
|
||||
if len(evictor.queue.queue) != len(queuePattern) {
|
||||
t.Fatalf("Queue %v should have length %d", evictor.queue.queue, len(queuePattern))
|
||||
}
|
||||
if !CheckQueueEq(queuePattern, evictor.queue.queue) {
|
||||
t.Errorf("Invalid queue. Got %v, expected %v", evictor.queue.queue, queuePattern)
|
||||
}
|
||||
|
||||
setPattern := sets.NewString("second", "third")
|
||||
if len(evictor.queue.set) != len(setPattern) {
|
||||
t.Fatalf("Map %v should have length %d", evictor.queue.set, len(setPattern))
|
||||
}
|
||||
if !CheckSetEq(setPattern, evictor.queue.set) {
|
||||
t.Errorf("Invalid map. Got %v, expected %v", evictor.queue.set, setPattern)
|
||||
}
|
||||
|
||||
evictor = NewRateLimitedTimedQueue(flowcontrol.NewFakeAlwaysRateLimiter())
|
||||
evictor.Add("first", "11111")
|
||||
evictor.Add("second", "22222")
|
||||
evictor.Add("third", "33333")
|
||||
evictor.Remove("second")
|
||||
|
||||
queuePattern = []string{"first", "third"}
|
||||
if len(evictor.queue.queue) != len(queuePattern) {
|
||||
t.Fatalf("Queue %v should have length %d", evictor.queue.queue, len(queuePattern))
|
||||
}
|
||||
if !CheckQueueEq(queuePattern, evictor.queue.queue) {
|
||||
t.Errorf("Invalid queue. Got %v, expected %v", evictor.queue.queue, queuePattern)
|
||||
}
|
||||
|
||||
setPattern = sets.NewString("first", "third")
|
||||
if len(evictor.queue.set) != len(setPattern) {
|
||||
t.Fatalf("Map %v should have length %d", evictor.queue.set, len(setPattern))
|
||||
}
|
||||
if !CheckSetEq(setPattern, evictor.queue.set) {
|
||||
t.Errorf("Invalid map. Got %v, expected %v", evictor.queue.set, setPattern)
|
||||
}
|
||||
|
||||
evictor = NewRateLimitedTimedQueue(flowcontrol.NewFakeAlwaysRateLimiter())
|
||||
evictor.Add("first", "11111")
|
||||
evictor.Add("second", "22222")
|
||||
evictor.Add("third", "33333")
|
||||
evictor.Remove("third")
|
||||
|
||||
queuePattern = []string{"first", "second"}
|
||||
if len(evictor.queue.queue) != len(queuePattern) {
|
||||
t.Fatalf("Queue %v should have length %d", evictor.queue.queue, len(queuePattern))
|
||||
}
|
||||
if !CheckQueueEq(queuePattern, evictor.queue.queue) {
|
||||
t.Errorf("Invalid queue. Got %v, expected %v", evictor.queue.queue, queuePattern)
|
||||
}
|
||||
|
||||
setPattern = sets.NewString("first", "second")
|
||||
if len(evictor.queue.set) != len(setPattern) {
|
||||
t.Fatalf("Map %v should have length %d", evictor.queue.set, len(setPattern))
|
||||
}
|
||||
if !CheckSetEq(setPattern, evictor.queue.set) {
|
||||
t.Errorf("Invalid map. Got %v, expected %v", evictor.queue.set, setPattern)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTry(t *testing.T) {
|
||||
evictor := NewRateLimitedTimedQueue(flowcontrol.NewFakeAlwaysRateLimiter())
|
||||
evictor.Add("first", "11111")
|
||||
evictor.Add("second", "22222")
|
||||
evictor.Add("third", "33333")
|
||||
evictor.Remove("second")
|
||||
|
||||
deletedMap := sets.NewString()
|
||||
evictor.Try(func(value TimedValue) (bool, time.Duration) {
|
||||
deletedMap.Insert(value.Value)
|
||||
return true, 0
|
||||
})
|
||||
|
||||
setPattern := sets.NewString("first", "third")
|
||||
if len(deletedMap) != len(setPattern) {
|
||||
t.Fatalf("Map %v should have length %d", evictor.queue.set, len(setPattern))
|
||||
}
|
||||
if !CheckSetEq(setPattern, deletedMap) {
|
||||
t.Errorf("Invalid map. Got %v, expected %v", deletedMap, setPattern)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTryOrdering(t *testing.T) {
|
||||
defer func() { now = time.Now }()
|
||||
current := time.Unix(0, 0)
|
||||
delay := 0
|
||||
// the current time is incremented by 1ms every time now is invoked
|
||||
now = func() time.Time {
|
||||
if delay > 0 {
|
||||
delay--
|
||||
} else {
|
||||
current = current.Add(time.Millisecond)
|
||||
}
|
||||
t.Logf("time %d", current.UnixNano())
|
||||
return current
|
||||
}
|
||||
evictor := NewRateLimitedTimedQueue(flowcontrol.NewFakeAlwaysRateLimiter())
|
||||
evictor.Add("first", "11111")
|
||||
evictor.Add("second", "22222")
|
||||
evictor.Add("third", "33333")
|
||||
|
||||
order := []string{}
|
||||
count := 0
|
||||
hasQueued := false
|
||||
evictor.Try(func(value TimedValue) (bool, time.Duration) {
|
||||
count++
|
||||
t.Logf("eviction %d", count)
|
||||
if value.ProcessAt.IsZero() {
|
||||
t.Fatalf("processAt should not be zero")
|
||||
}
|
||||
switch value.Value {
|
||||
case "first":
|
||||
if !value.AddedAt.Equal(time.Unix(0, time.Millisecond.Nanoseconds())) {
|
||||
t.Fatalf("added time for %s is %v", value.Value, value.AddedAt)
|
||||
}
|
||||
|
||||
case "second":
|
||||
if !value.AddedAt.Equal(time.Unix(0, 2*time.Millisecond.Nanoseconds())) {
|
||||
t.Fatalf("added time for %s is %v", value.Value, value.AddedAt)
|
||||
}
|
||||
if hasQueued {
|
||||
if !value.ProcessAt.Equal(time.Unix(0, 6*time.Millisecond.Nanoseconds())) {
|
||||
t.Fatalf("process time for %s is %v", value.Value, value.ProcessAt)
|
||||
}
|
||||
break
|
||||
}
|
||||
hasQueued = true
|
||||
delay = 1
|
||||
t.Logf("going to delay")
|
||||
return false, 2 * time.Millisecond
|
||||
|
||||
case "third":
|
||||
if !value.AddedAt.Equal(time.Unix(0, 3*time.Millisecond.Nanoseconds())) {
|
||||
t.Fatalf("added time for %s is %v", value.Value, value.AddedAt)
|
||||
}
|
||||
}
|
||||
order = append(order, value.Value)
|
||||
return true, 0
|
||||
})
|
||||
if !reflect.DeepEqual(order, []string{"first", "third"}) {
|
||||
t.Fatalf("order was wrong: %v", order)
|
||||
}
|
||||
if count != 3 {
|
||||
t.Fatalf("unexpected iterations: %d", count)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTryRemovingWhileTry(t *testing.T) {
|
||||
evictor := NewRateLimitedTimedQueue(flowcontrol.NewFakeAlwaysRateLimiter())
|
||||
evictor.Add("first", "11111")
|
||||
evictor.Add("second", "22222")
|
||||
evictor.Add("third", "33333")
|
||||
|
||||
processing := make(chan struct{})
|
||||
wait := make(chan struct{})
|
||||
order := []string{}
|
||||
count := 0
|
||||
queued := false
|
||||
|
||||
// while the Try function is processing "second", remove it from the queue
|
||||
// we should not see "second" retried.
|
||||
go func() {
|
||||
<-processing
|
||||
evictor.Remove("second")
|
||||
close(wait)
|
||||
}()
|
||||
|
||||
evictor.Try(func(value TimedValue) (bool, time.Duration) {
|
||||
count++
|
||||
if value.AddedAt.IsZero() {
|
||||
t.Fatalf("added should not be zero")
|
||||
}
|
||||
if value.ProcessAt.IsZero() {
|
||||
t.Fatalf("next should not be zero")
|
||||
}
|
||||
if !queued && value.Value == "second" {
|
||||
queued = true
|
||||
close(processing)
|
||||
<-wait
|
||||
return false, time.Millisecond
|
||||
}
|
||||
order = append(order, value.Value)
|
||||
return true, 0
|
||||
})
|
||||
|
||||
if !reflect.DeepEqual(order, []string{"first", "third"}) {
|
||||
t.Fatalf("order was wrong: %v", order)
|
||||
}
|
||||
if count != 3 {
|
||||
t.Fatalf("unexpected iterations: %d", count)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClear(t *testing.T) {
|
||||
evictor := NewRateLimitedTimedQueue(flowcontrol.NewFakeAlwaysRateLimiter())
|
||||
evictor.Add("first", "11111")
|
||||
evictor.Add("second", "22222")
|
||||
evictor.Add("third", "33333")
|
||||
|
||||
evictor.Clear()
|
||||
|
||||
if len(evictor.queue.queue) != 0 {
|
||||
t.Fatalf("Clear should remove all elements from the queue.")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSwapLimiter(t *testing.T) {
|
||||
evictor := NewRateLimitedTimedQueue(flowcontrol.NewFakeAlwaysRateLimiter())
|
||||
fakeAlways := flowcontrol.NewFakeAlwaysRateLimiter()
|
||||
qps := evictor.limiter.QPS()
|
||||
if qps != fakeAlways.QPS() {
|
||||
t.Fatalf("QPS does not match create one: %v instead of %v", qps, fakeAlways.QPS())
|
||||
}
|
||||
|
||||
evictor.SwapLimiter(0)
|
||||
qps = evictor.limiter.QPS()
|
||||
fakeNever := flowcontrol.NewFakeNeverRateLimiter()
|
||||
if qps != fakeNever.QPS() {
|
||||
t.Fatalf("QPS does not match create one: %v instead of %v", qps, fakeNever.QPS())
|
||||
}
|
||||
|
||||
createdQPS := float32(5.5)
|
||||
evictor.SwapLimiter(createdQPS)
|
||||
qps = evictor.limiter.QPS()
|
||||
if qps != createdQPS {
|
||||
t.Fatalf("QPS does not match create one: %v instead of %v", qps, createdQPS)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAddAfterTry(t *testing.T) {
|
||||
evictor := NewRateLimitedTimedQueue(flowcontrol.NewFakeAlwaysRateLimiter())
|
||||
evictor.Add("first", "11111")
|
||||
evictor.Add("second", "22222")
|
||||
evictor.Add("third", "33333")
|
||||
evictor.Remove("second")
|
||||
|
||||
deletedMap := sets.NewString()
|
||||
evictor.Try(func(value TimedValue) (bool, time.Duration) {
|
||||
deletedMap.Insert(value.Value)
|
||||
return true, 0
|
||||
})
|
||||
|
||||
setPattern := sets.NewString("first", "third")
|
||||
if len(deletedMap) != len(setPattern) {
|
||||
t.Fatalf("Map %v should have length %d", evictor.queue.set, len(setPattern))
|
||||
}
|
||||
if !CheckSetEq(setPattern, deletedMap) {
|
||||
t.Errorf("Invalid map. Got %v, expected %v", deletedMap, setPattern)
|
||||
}
|
||||
|
||||
evictor.Add("first", "11111")
|
||||
evictor.Try(func(value TimedValue) (bool, time.Duration) {
|
||||
t.Errorf("We shouldn't process the same value if the explicit remove wasn't called.")
|
||||
return true, 0
|
||||
})
|
||||
}
|
||||
486
vendor/k8s.io/kubernetes/pkg/controller/nodelifecycle/scheduler/taint_manager.go
generated
vendored
486
vendor/k8s.io/kubernetes/pkg/controller/nodelifecycle/scheduler/taint_manager.go
generated
vendored
@ -1,486 +0,0 @@
|
||||
/*
|
||||
Copyright 2017 The Kubernetes Authors.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
package scheduler
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"hash/fnv"
|
||||
"io"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"k8s.io/api/core/v1"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/fields"
|
||||
"k8s.io/apimachinery/pkg/labels"
|
||||
"k8s.io/apimachinery/pkg/types"
|
||||
clientset "k8s.io/client-go/kubernetes"
|
||||
"k8s.io/client-go/kubernetes/scheme"
|
||||
v1core "k8s.io/client-go/kubernetes/typed/core/v1"
|
||||
"k8s.io/client-go/tools/record"
|
||||
"k8s.io/client-go/util/workqueue"
|
||||
"k8s.io/kubernetes/pkg/apis/core/helper"
|
||||
v1helper "k8s.io/kubernetes/pkg/apis/core/v1/helper"
|
||||
|
||||
"github.com/golang/glog"
|
||||
)
|
||||
|
||||
const (
|
||||
nodeUpdateChannelSize = 10
|
||||
podUpdateChannelSize = 1
|
||||
retries = 5
|
||||
)
|
||||
|
||||
// Needed to make workqueue work
|
||||
type updateItemInterface interface{}
|
||||
|
||||
type nodeUpdateItem struct {
|
||||
oldNode *v1.Node
|
||||
newNode *v1.Node
|
||||
newTaints []v1.Taint
|
||||
}
|
||||
|
||||
type podUpdateItem struct {
|
||||
oldPod *v1.Pod
|
||||
newPod *v1.Pod
|
||||
newTolerations []v1.Toleration
|
||||
}
|
||||
|
||||
func (n *nodeUpdateItem) name() string {
|
||||
if n.newNode != nil {
|
||||
return n.newNode.ObjectMeta.Name
|
||||
}
|
||||
if n.oldNode != nil {
|
||||
return n.oldNode.ObjectMeta.Name
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func (p *podUpdateItem) nodeName() string {
|
||||
if p.newPod != nil {
|
||||
return p.newPod.Spec.NodeName
|
||||
}
|
||||
if p.oldPod != nil {
|
||||
return p.oldPod.Spec.NodeName
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func hash(val string, max int) int {
|
||||
hasher := fnv.New32a()
|
||||
io.WriteString(hasher, val)
|
||||
return int(hasher.Sum32() % uint32(max))
|
||||
}
|
||||
|
||||
// NoExecuteTaintManager listens to Taint/Toleration changes and is responsible for removing Pods
|
||||
// from Nodes tainted with NoExecute Taints.
|
||||
type NoExecuteTaintManager struct {
|
||||
client clientset.Interface
|
||||
recorder record.EventRecorder
|
||||
|
||||
taintEvictionQueue *TimedWorkerQueue
|
||||
// keeps a map from nodeName to all noExecute taints on that Node
|
||||
taintedNodesLock sync.Mutex
|
||||
taintedNodes map[string][]v1.Taint
|
||||
|
||||
nodeUpdateChannels []chan *nodeUpdateItem
|
||||
podUpdateChannels []chan *podUpdateItem
|
||||
|
||||
nodeUpdateQueue workqueue.Interface
|
||||
podUpdateQueue workqueue.Interface
|
||||
}
|
||||
|
||||
func deletePodHandler(c clientset.Interface, emitEventFunc func(types.NamespacedName)) func(args *WorkArgs) error {
|
||||
return func(args *WorkArgs) error {
|
||||
ns := args.NamespacedName.Namespace
|
||||
name := args.NamespacedName.Name
|
||||
glog.V(0).Infof("NoExecuteTaintManager is deleting Pod: %v", args.NamespacedName.String())
|
||||
if emitEventFunc != nil {
|
||||
emitEventFunc(args.NamespacedName)
|
||||
}
|
||||
var err error
|
||||
for i := 0; i < retries; i++ {
|
||||
err = c.CoreV1().Pods(ns).Delete(name, &metav1.DeleteOptions{})
|
||||
if err == nil {
|
||||
break
|
||||
}
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
func getNoExecuteTaints(taints []v1.Taint) []v1.Taint {
|
||||
result := []v1.Taint{}
|
||||
for i := range taints {
|
||||
if taints[i].Effect == v1.TaintEffectNoExecute {
|
||||
result = append(result, taints[i])
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func getPodsAssignedToNode(c clientset.Interface, nodeName string) ([]v1.Pod, error) {
|
||||
selector := fields.SelectorFromSet(fields.Set{"spec.nodeName": nodeName})
|
||||
pods, err := c.CoreV1().Pods(v1.NamespaceAll).List(metav1.ListOptions{
|
||||
FieldSelector: selector.String(),
|
||||
LabelSelector: labels.Everything().String(),
|
||||
})
|
||||
for i := 0; i < retries && err != nil; i++ {
|
||||
pods, err = c.CoreV1().Pods(v1.NamespaceAll).List(metav1.ListOptions{
|
||||
FieldSelector: selector.String(),
|
||||
LabelSelector: labels.Everything().String(),
|
||||
})
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
}
|
||||
if err != nil {
|
||||
return []v1.Pod{}, fmt.Errorf("failed to get Pods assigned to node %v", nodeName)
|
||||
}
|
||||
return pods.Items, nil
|
||||
}
|
||||
|
||||
// getMinTolerationTime returns minimal toleration time from the given slice, or -1 if it's infinite.
|
||||
func getMinTolerationTime(tolerations []v1.Toleration) time.Duration {
|
||||
minTolerationTime := int64(-1)
|
||||
if len(tolerations) == 0 {
|
||||
return 0
|
||||
}
|
||||
|
||||
for i := range tolerations {
|
||||
if tolerations[i].TolerationSeconds != nil {
|
||||
tolerationSeconds := *(tolerations[i].TolerationSeconds)
|
||||
if tolerationSeconds <= 0 {
|
||||
return 0
|
||||
} else if tolerationSeconds < minTolerationTime || minTolerationTime == -1 {
|
||||
minTolerationTime = tolerationSeconds
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return time.Duration(minTolerationTime) * time.Second
|
||||
}
|
||||
|
||||
// NewNoExecuteTaintManager creates a new NoExecuteTaintManager that will use passed clientset to
|
||||
// communicate with the API server.
|
||||
func NewNoExecuteTaintManager(c clientset.Interface) *NoExecuteTaintManager {
|
||||
eventBroadcaster := record.NewBroadcaster()
|
||||
recorder := eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "taint-controller"})
|
||||
eventBroadcaster.StartLogging(glog.Infof)
|
||||
if c != nil {
|
||||
glog.V(0).Infof("Sending events to api server.")
|
||||
eventBroadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: c.CoreV1().Events("")})
|
||||
} else {
|
||||
glog.Fatalf("kubeClient is nil when starting NodeController")
|
||||
}
|
||||
|
||||
tm := &NoExecuteTaintManager{
|
||||
client: c,
|
||||
recorder: recorder,
|
||||
taintedNodes: make(map[string][]v1.Taint),
|
||||
|
||||
nodeUpdateQueue: workqueue.New(),
|
||||
podUpdateQueue: workqueue.New(),
|
||||
}
|
||||
tm.taintEvictionQueue = CreateWorkerQueue(deletePodHandler(c, tm.emitPodDeletionEvent))
|
||||
|
||||
return tm
|
||||
}
|
||||
|
||||
// Run starts NoExecuteTaintManager which will run in loop until `stopCh` is closed.
|
||||
func (tc *NoExecuteTaintManager) Run(stopCh <-chan struct{}) {
|
||||
glog.V(0).Infof("Starting NoExecuteTaintManager")
|
||||
|
||||
// TODO: Figure out a reasonable number of workers and propagate the
|
||||
// number of workers up making it a paramater of Run() function.
|
||||
workers := 8
|
||||
for i := 0; i < workers; i++ {
|
||||
tc.nodeUpdateChannels = append(tc.nodeUpdateChannels, make(chan *nodeUpdateItem, nodeUpdateChannelSize))
|
||||
tc.podUpdateChannels = append(tc.podUpdateChannels, make(chan *podUpdateItem, podUpdateChannelSize))
|
||||
}
|
||||
|
||||
// Functions that are responsible for taking work items out of the workqueues and putting them
|
||||
// into channels.
|
||||
go func(stopCh <-chan struct{}) {
|
||||
for {
|
||||
item, shutdown := tc.nodeUpdateQueue.Get()
|
||||
if shutdown {
|
||||
break
|
||||
}
|
||||
nodeUpdate := item.(*nodeUpdateItem)
|
||||
hash := hash(nodeUpdate.name(), workers)
|
||||
select {
|
||||
case <-stopCh:
|
||||
tc.nodeUpdateQueue.Done(item)
|
||||
break
|
||||
case tc.nodeUpdateChannels[hash] <- nodeUpdate:
|
||||
}
|
||||
tc.nodeUpdateQueue.Done(item)
|
||||
}
|
||||
}(stopCh)
|
||||
|
||||
go func(stopCh <-chan struct{}) {
|
||||
for {
|
||||
item, shutdown := tc.podUpdateQueue.Get()
|
||||
if shutdown {
|
||||
break
|
||||
}
|
||||
podUpdate := item.(*podUpdateItem)
|
||||
hash := hash(podUpdate.nodeName(), workers)
|
||||
select {
|
||||
case <-stopCh:
|
||||
tc.podUpdateQueue.Done(item)
|
||||
break
|
||||
case tc.podUpdateChannels[hash] <- podUpdate:
|
||||
}
|
||||
tc.podUpdateQueue.Done(item)
|
||||
}
|
||||
}(stopCh)
|
||||
|
||||
wg := sync.WaitGroup{}
|
||||
wg.Add(workers)
|
||||
for i := 0; i < workers; i++ {
|
||||
go tc.worker(i, wg.Done, stopCh)
|
||||
}
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
func (tc *NoExecuteTaintManager) worker(worker int, done func(), stopCh <-chan struct{}) {
|
||||
defer done()
|
||||
|
||||
// When processing events we want to prioritize Node updates over Pod updates,
|
||||
// as NodeUpdates that interest NoExecuteTaintManager should be handled as soon as possible -
|
||||
// we don't want user (or system) to wait until PodUpdate queue is drained before it can
|
||||
// start evicting Pods from tainted Nodes.
|
||||
for {
|
||||
select {
|
||||
case <-stopCh:
|
||||
return
|
||||
case nodeUpdate := <-tc.nodeUpdateChannels[worker]:
|
||||
tc.handleNodeUpdate(nodeUpdate)
|
||||
case podUpdate := <-tc.podUpdateChannels[worker]:
|
||||
// If we found a Pod update we need to empty Node queue first.
|
||||
priority:
|
||||
for {
|
||||
select {
|
||||
case nodeUpdate := <-tc.nodeUpdateChannels[worker]:
|
||||
tc.handleNodeUpdate(nodeUpdate)
|
||||
default:
|
||||
break priority
|
||||
}
|
||||
}
|
||||
// After Node queue is emptied we process podUpdate.
|
||||
tc.handlePodUpdate(podUpdate)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// PodUpdated is used to notify NoExecuteTaintManager about Pod changes.
|
||||
func (tc *NoExecuteTaintManager) PodUpdated(oldPod *v1.Pod, newPod *v1.Pod) {
|
||||
oldTolerations := []v1.Toleration{}
|
||||
if oldPod != nil {
|
||||
oldTolerations = oldPod.Spec.Tolerations
|
||||
}
|
||||
newTolerations := []v1.Toleration{}
|
||||
if newPod != nil {
|
||||
newTolerations = newPod.Spec.Tolerations
|
||||
}
|
||||
|
||||
if oldPod != nil && newPod != nil && helper.Semantic.DeepEqual(oldTolerations, newTolerations) && oldPod.Spec.NodeName == newPod.Spec.NodeName {
|
||||
return
|
||||
}
|
||||
updateItem := &podUpdateItem{
|
||||
oldPod: oldPod,
|
||||
newPod: newPod,
|
||||
newTolerations: newTolerations,
|
||||
}
|
||||
|
||||
tc.podUpdateQueue.Add(updateItemInterface(updateItem))
|
||||
}
|
||||
|
||||
// NodeUpdated is used to notify NoExecuteTaintManager about Node changes.
|
||||
func (tc *NoExecuteTaintManager) NodeUpdated(oldNode *v1.Node, newNode *v1.Node) {
|
||||
oldTaints := []v1.Taint{}
|
||||
if oldNode != nil {
|
||||
oldTaints = oldNode.Spec.Taints
|
||||
}
|
||||
oldTaints = getNoExecuteTaints(oldTaints)
|
||||
|
||||
newTaints := []v1.Taint{}
|
||||
if newNode != nil {
|
||||
newTaints = newNode.Spec.Taints
|
||||
}
|
||||
newTaints = getNoExecuteTaints(newTaints)
|
||||
|
||||
if oldNode != nil && newNode != nil && helper.Semantic.DeepEqual(oldTaints, newTaints) {
|
||||
return
|
||||
}
|
||||
updateItem := &nodeUpdateItem{
|
||||
oldNode: oldNode,
|
||||
newNode: newNode,
|
||||
newTaints: newTaints,
|
||||
}
|
||||
|
||||
tc.nodeUpdateQueue.Add(updateItemInterface(updateItem))
|
||||
}
|
||||
|
||||
func (tc *NoExecuteTaintManager) cancelWorkWithEvent(nsName types.NamespacedName) {
|
||||
if tc.taintEvictionQueue.CancelWork(nsName.String()) {
|
||||
tc.emitCancelPodDeletionEvent(nsName)
|
||||
}
|
||||
}
|
||||
|
||||
func (tc *NoExecuteTaintManager) processPodOnNode(
|
||||
podNamespacedName types.NamespacedName,
|
||||
nodeName string,
|
||||
tolerations []v1.Toleration,
|
||||
taints []v1.Taint,
|
||||
now time.Time,
|
||||
) {
|
||||
if len(taints) == 0 {
|
||||
tc.cancelWorkWithEvent(podNamespacedName)
|
||||
}
|
||||
allTolerated, usedTolerations := v1helper.GetMatchingTolerations(taints, tolerations)
|
||||
if !allTolerated {
|
||||
glog.V(2).Infof("Not all taints are tolerated after update for Pod %v on %v", podNamespacedName.String(), nodeName)
|
||||
// We're canceling scheduled work (if any), as we're going to delete the Pod right away.
|
||||
tc.cancelWorkWithEvent(podNamespacedName)
|
||||
tc.taintEvictionQueue.AddWork(NewWorkArgs(podNamespacedName.Name, podNamespacedName.Namespace), time.Now(), time.Now())
|
||||
return
|
||||
}
|
||||
minTolerationTime := getMinTolerationTime(usedTolerations)
|
||||
// getMinTolerationTime returns negative value to denote infinite toleration.
|
||||
if minTolerationTime < 0 {
|
||||
glog.V(4).Infof("New tolerations for %v tolerate forever. Scheduled deletion won't be cancelled if already scheduled.", podNamespacedName.String())
|
||||
return
|
||||
}
|
||||
|
||||
startTime := now
|
||||
triggerTime := startTime.Add(minTolerationTime)
|
||||
scheduledEviction := tc.taintEvictionQueue.GetWorkerUnsafe(podNamespacedName.String())
|
||||
if scheduledEviction != nil {
|
||||
startTime = scheduledEviction.CreatedAt
|
||||
if startTime.Add(minTolerationTime).Before(triggerTime) {
|
||||
return
|
||||
}
|
||||
tc.cancelWorkWithEvent(podNamespacedName)
|
||||
}
|
||||
tc.taintEvictionQueue.AddWork(NewWorkArgs(podNamespacedName.Name, podNamespacedName.Namespace), startTime, triggerTime)
|
||||
}
|
||||
|
||||
func (tc *NoExecuteTaintManager) handlePodUpdate(podUpdate *podUpdateItem) {
|
||||
// Delete
|
||||
if podUpdate.newPod == nil {
|
||||
pod := podUpdate.oldPod
|
||||
podNamespacedName := types.NamespacedName{Namespace: pod.Namespace, Name: pod.Name}
|
||||
glog.V(4).Infof("Noticed pod deletion: %#v", podNamespacedName)
|
||||
tc.cancelWorkWithEvent(podNamespacedName)
|
||||
return
|
||||
}
|
||||
// Create or Update
|
||||
pod := podUpdate.newPod
|
||||
podNamespacedName := types.NamespacedName{Namespace: pod.Namespace, Name: pod.Name}
|
||||
glog.V(4).Infof("Noticed pod update: %#v", podNamespacedName)
|
||||
nodeName := pod.Spec.NodeName
|
||||
if nodeName == "" {
|
||||
return
|
||||
}
|
||||
taints, ok := func() ([]v1.Taint, bool) {
|
||||
tc.taintedNodesLock.Lock()
|
||||
defer tc.taintedNodesLock.Unlock()
|
||||
taints, ok := tc.taintedNodes[nodeName]
|
||||
return taints, ok
|
||||
}()
|
||||
// It's possible that Node was deleted, or Taints were removed before, which triggered
|
||||
// eviction cancelling if it was needed.
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
tc.processPodOnNode(podNamespacedName, nodeName, podUpdate.newTolerations, taints, time.Now())
|
||||
}
|
||||
|
||||
func (tc *NoExecuteTaintManager) handleNodeUpdate(nodeUpdate *nodeUpdateItem) {
|
||||
// Delete
|
||||
if nodeUpdate.newNode == nil {
|
||||
node := nodeUpdate.oldNode
|
||||
glog.V(4).Infof("Noticed node deletion: %#v", node.Name)
|
||||
tc.taintedNodesLock.Lock()
|
||||
defer tc.taintedNodesLock.Unlock()
|
||||
delete(tc.taintedNodes, node.Name)
|
||||
return
|
||||
}
|
||||
// Create or Update
|
||||
glog.V(4).Infof("Noticed node update: %#v", nodeUpdate)
|
||||
node := nodeUpdate.newNode
|
||||
taints := nodeUpdate.newTaints
|
||||
func() {
|
||||
tc.taintedNodesLock.Lock()
|
||||
defer tc.taintedNodesLock.Unlock()
|
||||
glog.V(4).Infof("Updating known taints on node %v: %v", node.Name, taints)
|
||||
if len(taints) == 0 {
|
||||
delete(tc.taintedNodes, node.Name)
|
||||
} else {
|
||||
tc.taintedNodes[node.Name] = taints
|
||||
}
|
||||
}()
|
||||
pods, err := getPodsAssignedToNode(tc.client, node.Name)
|
||||
if err != nil {
|
||||
glog.Errorf(err.Error())
|
||||
return
|
||||
}
|
||||
if len(pods) == 0 {
|
||||
return
|
||||
}
|
||||
// Short circuit, to make this controller a bit faster.
|
||||
if len(taints) == 0 {
|
||||
glog.V(4).Infof("All taints were removed from the Node %v. Cancelling all evictions...", node.Name)
|
||||
for i := range pods {
|
||||
tc.cancelWorkWithEvent(types.NamespacedName{Namespace: pods[i].Namespace, Name: pods[i].Name})
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
now := time.Now()
|
||||
for i := range pods {
|
||||
pod := &pods[i]
|
||||
podNamespacedName := types.NamespacedName{Namespace: pod.Namespace, Name: pod.Name}
|
||||
tc.processPodOnNode(podNamespacedName, node.Name, pod.Spec.Tolerations, taints, now)
|
||||
}
|
||||
}
|
||||
|
||||
func (tc *NoExecuteTaintManager) emitPodDeletionEvent(nsName types.NamespacedName) {
|
||||
if tc.recorder == nil {
|
||||
return
|
||||
}
|
||||
ref := &v1.ObjectReference{
|
||||
Kind: "Pod",
|
||||
Name: nsName.Name,
|
||||
Namespace: nsName.Namespace,
|
||||
}
|
||||
tc.recorder.Eventf(ref, v1.EventTypeNormal, "TaintManagerEviction", "Marking for deletion Pod %s", nsName.String())
|
||||
}
|
||||
|
||||
func (tc *NoExecuteTaintManager) emitCancelPodDeletionEvent(nsName types.NamespacedName) {
|
||||
if tc.recorder == nil {
|
||||
return
|
||||
}
|
||||
ref := &v1.ObjectReference{
|
||||
Kind: "Pod",
|
||||
Name: nsName.Name,
|
||||
Namespace: nsName.Namespace,
|
||||
}
|
||||
tc.recorder.Eventf(ref, v1.EventTypeNormal, "TaintManagerEviction", "Cancelling deletion of Pod %s", nsName.String())
|
||||
}
|
||||
604
vendor/k8s.io/kubernetes/pkg/controller/nodelifecycle/scheduler/taint_manager_test.go
generated
vendored
604
vendor/k8s.io/kubernetes/pkg/controller/nodelifecycle/scheduler/taint_manager_test.go
generated
vendored
@ -1,604 +0,0 @@
|
||||
/*
|
||||
Copyright 2017 The Kubernetes Authors.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
package scheduler
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sort"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"k8s.io/api/core/v1"
|
||||
"k8s.io/client-go/kubernetes/fake"
|
||||
"k8s.io/kubernetes/pkg/controller/testutil"
|
||||
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
clienttesting "k8s.io/client-go/testing"
|
||||
)
|
||||
|
||||
var timeForControllerToProgress = 500 * time.Millisecond
|
||||
|
||||
func createNoExecuteTaint(index int) v1.Taint {
|
||||
now := metav1.Now()
|
||||
return v1.Taint{
|
||||
Key: "testTaint" + fmt.Sprintf("%v", index),
|
||||
Value: "test" + fmt.Sprintf("%v", index),
|
||||
Effect: v1.TaintEffectNoExecute,
|
||||
TimeAdded: &now,
|
||||
}
|
||||
}
|
||||
|
||||
func addToleration(pod *v1.Pod, index int, duration int64) *v1.Pod {
|
||||
if pod.Annotations == nil {
|
||||
pod.Annotations = map[string]string{}
|
||||
}
|
||||
if duration < 0 {
|
||||
pod.Spec.Tolerations = []v1.Toleration{{Key: "testTaint" + fmt.Sprintf("%v", index), Value: "test" + fmt.Sprintf("%v", index), Effect: v1.TaintEffectNoExecute}}
|
||||
|
||||
} else {
|
||||
pod.Spec.Tolerations = []v1.Toleration{{Key: "testTaint" + fmt.Sprintf("%v", index), Value: "test" + fmt.Sprintf("%v", index), Effect: v1.TaintEffectNoExecute, TolerationSeconds: &duration}}
|
||||
}
|
||||
return pod
|
||||
}
|
||||
|
||||
func addTaintsToNode(node *v1.Node, key, value string, indices []int) *v1.Node {
|
||||
taints := []v1.Taint{}
|
||||
for _, index := range indices {
|
||||
taints = append(taints, createNoExecuteTaint(index))
|
||||
}
|
||||
node.Spec.Taints = taints
|
||||
return node
|
||||
}
|
||||
|
||||
type timestampedPod struct {
|
||||
names []string
|
||||
timestamp time.Duration
|
||||
}
|
||||
|
||||
type durationSlice []timestampedPod
|
||||
|
||||
func (a durationSlice) Len() int { return len(a) }
|
||||
func (a durationSlice) Swap(i, j int) { a[i], a[j] = a[j], a[i] }
|
||||
func (a durationSlice) Less(i, j int) bool { return a[i].timestamp < a[j].timestamp }
|
||||
|
||||
func TestFilterNoExecuteTaints(t *testing.T) {
|
||||
taints := []v1.Taint{
|
||||
{
|
||||
Key: "one",
|
||||
Value: "one",
|
||||
Effect: v1.TaintEffectNoExecute,
|
||||
},
|
||||
{
|
||||
Key: "two",
|
||||
Value: "two",
|
||||
Effect: v1.TaintEffectNoSchedule,
|
||||
},
|
||||
}
|
||||
taints = getNoExecuteTaints(taints)
|
||||
if len(taints) != 1 || taints[0].Key != "one" {
|
||||
t.Errorf("Filtering doesn't work. Got %v", taints)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCreatePod(t *testing.T) {
|
||||
testCases := []struct {
|
||||
description string
|
||||
pod *v1.Pod
|
||||
taintedNodes map[string][]v1.Taint
|
||||
expectDelete bool
|
||||
}{
|
||||
{
|
||||
description: "not scheduled - ignore",
|
||||
pod: testutil.NewPod("pod1", ""),
|
||||
taintedNodes: map[string][]v1.Taint{},
|
||||
expectDelete: false,
|
||||
},
|
||||
{
|
||||
description: "scheduled on untainted Node",
|
||||
pod: testutil.NewPod("pod1", "node1"),
|
||||
taintedNodes: map[string][]v1.Taint{},
|
||||
expectDelete: false,
|
||||
},
|
||||
{
|
||||
description: "schedule on tainted Node",
|
||||
pod: testutil.NewPod("pod1", "node1"),
|
||||
taintedNodes: map[string][]v1.Taint{
|
||||
"node1": {createNoExecuteTaint(1)},
|
||||
},
|
||||
expectDelete: true,
|
||||
},
|
||||
{
|
||||
description: "schedule on tainted Node with finite toleration",
|
||||
pod: addToleration(testutil.NewPod("pod1", "node1"), 1, 100),
|
||||
taintedNodes: map[string][]v1.Taint{
|
||||
"node1": {createNoExecuteTaint(1)},
|
||||
},
|
||||
expectDelete: false,
|
||||
},
|
||||
{
|
||||
description: "schedule on tainted Node with infinite toleration",
|
||||
pod: addToleration(testutil.NewPod("pod1", "node1"), 1, -1),
|
||||
taintedNodes: map[string][]v1.Taint{
|
||||
"node1": {createNoExecuteTaint(1)},
|
||||
},
|
||||
expectDelete: false,
|
||||
},
|
||||
{
|
||||
description: "schedule on tainted Node with infinite ivalid toleration",
|
||||
pod: addToleration(testutil.NewPod("pod1", "node1"), 2, -1),
|
||||
taintedNodes: map[string][]v1.Taint{
|
||||
"node1": {createNoExecuteTaint(1)},
|
||||
},
|
||||
expectDelete: true,
|
||||
},
|
||||
}
|
||||
|
||||
for _, item := range testCases {
|
||||
stopCh := make(chan struct{})
|
||||
fakeClientset := fake.NewSimpleClientset()
|
||||
controller := NewNoExecuteTaintManager(fakeClientset)
|
||||
controller.recorder = testutil.NewFakeRecorder()
|
||||
go controller.Run(stopCh)
|
||||
controller.taintedNodes = item.taintedNodes
|
||||
controller.PodUpdated(nil, item.pod)
|
||||
// wait a bit
|
||||
time.Sleep(timeForControllerToProgress)
|
||||
|
||||
podDeleted := false
|
||||
for _, action := range fakeClientset.Actions() {
|
||||
if action.GetVerb() == "delete" && action.GetResource().Resource == "pods" {
|
||||
podDeleted = true
|
||||
}
|
||||
}
|
||||
if podDeleted != item.expectDelete {
|
||||
t.Errorf("%v: Unexepected test result. Expected delete %v, got %v", item.description, item.expectDelete, podDeleted)
|
||||
}
|
||||
close(stopCh)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDeletePod(t *testing.T) {
|
||||
stopCh := make(chan struct{})
|
||||
fakeClientset := fake.NewSimpleClientset()
|
||||
controller := NewNoExecuteTaintManager(fakeClientset)
|
||||
controller.recorder = testutil.NewFakeRecorder()
|
||||
go controller.Run(stopCh)
|
||||
controller.taintedNodes = map[string][]v1.Taint{
|
||||
"node1": {createNoExecuteTaint(1)},
|
||||
}
|
||||
controller.PodUpdated(testutil.NewPod("pod1", "node1"), nil)
|
||||
// wait a bit to see if nothing will panic
|
||||
time.Sleep(timeForControllerToProgress)
|
||||
close(stopCh)
|
||||
}
|
||||
|
||||
func TestUpdatePod(t *testing.T) {
|
||||
testCases := []struct {
|
||||
description string
|
||||
prevPod *v1.Pod
|
||||
newPod *v1.Pod
|
||||
taintedNodes map[string][]v1.Taint
|
||||
expectDelete bool
|
||||
additionalSleep time.Duration
|
||||
}{
|
||||
{
|
||||
description: "scheduling onto tainted Node",
|
||||
prevPod: testutil.NewPod("pod1", ""),
|
||||
newPod: testutil.NewPod("pod1", "node1"),
|
||||
taintedNodes: map[string][]v1.Taint{
|
||||
"node1": {createNoExecuteTaint(1)},
|
||||
},
|
||||
expectDelete: true,
|
||||
},
|
||||
{
|
||||
description: "scheduling onto tainted Node with toleration",
|
||||
prevPod: addToleration(testutil.NewPod("pod1", ""), 1, -1),
|
||||
newPod: addToleration(testutil.NewPod("pod1", "node1"), 1, -1),
|
||||
taintedNodes: map[string][]v1.Taint{
|
||||
"node1": {createNoExecuteTaint(1)},
|
||||
},
|
||||
expectDelete: false,
|
||||
},
|
||||
{
|
||||
description: "removing toleration",
|
||||
prevPod: addToleration(testutil.NewPod("pod1", "node1"), 1, 100),
|
||||
newPod: testutil.NewPod("pod1", "node1"),
|
||||
taintedNodes: map[string][]v1.Taint{
|
||||
"node1": {createNoExecuteTaint(1)},
|
||||
},
|
||||
expectDelete: true,
|
||||
},
|
||||
{
|
||||
description: "lengthening toleration shouldn't work",
|
||||
prevPod: addToleration(testutil.NewPod("pod1", "node1"), 1, 1),
|
||||
newPod: addToleration(testutil.NewPod("pod1", "node1"), 1, 100),
|
||||
taintedNodes: map[string][]v1.Taint{
|
||||
"node1": {createNoExecuteTaint(1)},
|
||||
},
|
||||
expectDelete: true,
|
||||
additionalSleep: 1500 * time.Millisecond,
|
||||
},
|
||||
}
|
||||
|
||||
for _, item := range testCases {
|
||||
stopCh := make(chan struct{})
|
||||
fakeClientset := fake.NewSimpleClientset()
|
||||
controller := NewNoExecuteTaintManager(fakeClientset)
|
||||
controller.recorder = testutil.NewFakeRecorder()
|
||||
go controller.Run(stopCh)
|
||||
controller.taintedNodes = item.taintedNodes
|
||||
|
||||
controller.PodUpdated(nil, item.prevPod)
|
||||
fakeClientset.ClearActions()
|
||||
time.Sleep(timeForControllerToProgress)
|
||||
controller.PodUpdated(item.prevPod, item.newPod)
|
||||
// wait a bit
|
||||
time.Sleep(timeForControllerToProgress)
|
||||
if item.additionalSleep > 0 {
|
||||
time.Sleep(item.additionalSleep)
|
||||
}
|
||||
|
||||
podDeleted := false
|
||||
for _, action := range fakeClientset.Actions() {
|
||||
if action.GetVerb() == "delete" && action.GetResource().Resource == "pods" {
|
||||
podDeleted = true
|
||||
}
|
||||
}
|
||||
if podDeleted != item.expectDelete {
|
||||
t.Errorf("%v: Unexepected test result. Expected delete %v, got %v", item.description, item.expectDelete, podDeleted)
|
||||
}
|
||||
close(stopCh)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCreateNode(t *testing.T) {
|
||||
testCases := []struct {
|
||||
description string
|
||||
pods []v1.Pod
|
||||
node *v1.Node
|
||||
expectDelete bool
|
||||
}{
|
||||
{
|
||||
description: "Creating Node matching already assigned Pod",
|
||||
pods: []v1.Pod{
|
||||
*testutil.NewPod("pod1", "node1"),
|
||||
},
|
||||
node: testutil.NewNode("node1"),
|
||||
expectDelete: false,
|
||||
},
|
||||
{
|
||||
description: "Creating tainted Node matching already assigned Pod",
|
||||
pods: []v1.Pod{
|
||||
*testutil.NewPod("pod1", "node1"),
|
||||
},
|
||||
node: addTaintsToNode(testutil.NewNode("node1"), "testTaint1", "taint1", []int{1}),
|
||||
expectDelete: true,
|
||||
},
|
||||
{
|
||||
description: "Creating tainted Node matching already assigned tolerating Pod",
|
||||
pods: []v1.Pod{
|
||||
*addToleration(testutil.NewPod("pod1", "node1"), 1, -1),
|
||||
},
|
||||
node: addTaintsToNode(testutil.NewNode("node1"), "testTaint1", "taint1", []int{1}),
|
||||
expectDelete: false,
|
||||
},
|
||||
}
|
||||
|
||||
for _, item := range testCases {
|
||||
stopCh := make(chan struct{})
|
||||
fakeClientset := fake.NewSimpleClientset(&v1.PodList{Items: item.pods})
|
||||
controller := NewNoExecuteTaintManager(fakeClientset)
|
||||
controller.recorder = testutil.NewFakeRecorder()
|
||||
go controller.Run(stopCh)
|
||||
controller.NodeUpdated(nil, item.node)
|
||||
// wait a bit
|
||||
time.Sleep(timeForControllerToProgress)
|
||||
|
||||
podDeleted := false
|
||||
for _, action := range fakeClientset.Actions() {
|
||||
if action.GetVerb() == "delete" && action.GetResource().Resource == "pods" {
|
||||
podDeleted = true
|
||||
}
|
||||
}
|
||||
if podDeleted != item.expectDelete {
|
||||
t.Errorf("%v: Unexepected test result. Expected delete %v, got %v", item.description, item.expectDelete, podDeleted)
|
||||
}
|
||||
close(stopCh)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDeleteNode(t *testing.T) {
|
||||
stopCh := make(chan struct{})
|
||||
fakeClientset := fake.NewSimpleClientset()
|
||||
controller := NewNoExecuteTaintManager(fakeClientset)
|
||||
controller.recorder = testutil.NewFakeRecorder()
|
||||
controller.taintedNodes = map[string][]v1.Taint{
|
||||
"node1": {createNoExecuteTaint(1)},
|
||||
}
|
||||
go controller.Run(stopCh)
|
||||
controller.NodeUpdated(testutil.NewNode("node1"), nil)
|
||||
// wait a bit to see if nothing will panic
|
||||
time.Sleep(timeForControllerToProgress)
|
||||
controller.taintedNodesLock.Lock()
|
||||
if _, ok := controller.taintedNodes["node1"]; ok {
|
||||
t.Error("Node should have been deleted from taintedNodes list")
|
||||
}
|
||||
controller.taintedNodesLock.Unlock()
|
||||
close(stopCh)
|
||||
}
|
||||
|
||||
func TestUpdateNode(t *testing.T) {
|
||||
testCases := []struct {
|
||||
description string
|
||||
pods []v1.Pod
|
||||
oldNode *v1.Node
|
||||
newNode *v1.Node
|
||||
expectDelete bool
|
||||
additionalSleep time.Duration
|
||||
}{
|
||||
{
|
||||
description: "Added taint",
|
||||
pods: []v1.Pod{
|
||||
*testutil.NewPod("pod1", "node1"),
|
||||
},
|
||||
oldNode: testutil.NewNode("node1"),
|
||||
newNode: addTaintsToNode(testutil.NewNode("node1"), "testTaint1", "taint1", []int{1}),
|
||||
expectDelete: true,
|
||||
},
|
||||
{
|
||||
description: "Added tolerated taint",
|
||||
pods: []v1.Pod{
|
||||
*addToleration(testutil.NewPod("pod1", "node1"), 1, 100),
|
||||
},
|
||||
oldNode: testutil.NewNode("node1"),
|
||||
newNode: addTaintsToNode(testutil.NewNode("node1"), "testTaint1", "taint1", []int{1}),
|
||||
expectDelete: false,
|
||||
},
|
||||
{
|
||||
description: "Only one added taint tolerated",
|
||||
pods: []v1.Pod{
|
||||
*addToleration(testutil.NewPod("pod1", "node1"), 1, 100),
|
||||
},
|
||||
oldNode: testutil.NewNode("node1"),
|
||||
newNode: addTaintsToNode(testutil.NewNode("node1"), "testTaint1", "taint1", []int{1, 2}),
|
||||
expectDelete: true,
|
||||
},
|
||||
{
|
||||
description: "Taint removed",
|
||||
pods: []v1.Pod{
|
||||
*addToleration(testutil.NewPod("pod1", "node1"), 1, 1),
|
||||
},
|
||||
oldNode: addTaintsToNode(testutil.NewNode("node1"), "testTaint1", "taint1", []int{1}),
|
||||
newNode: testutil.NewNode("node1"),
|
||||
expectDelete: false,
|
||||
additionalSleep: 1500 * time.Millisecond,
|
||||
},
|
||||
{
|
||||
description: "Pod with multiple tolerations are evicted when first one runs out",
|
||||
pods: []v1.Pod{
|
||||
{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Namespace: "default",
|
||||
Name: "pod1",
|
||||
},
|
||||
Spec: v1.PodSpec{
|
||||
NodeName: "node1",
|
||||
Tolerations: []v1.Toleration{
|
||||
{Key: "testTaint1", Value: "test1", Effect: v1.TaintEffectNoExecute, TolerationSeconds: &[]int64{1}[0]},
|
||||
{Key: "testTaint2", Value: "test2", Effect: v1.TaintEffectNoExecute, TolerationSeconds: &[]int64{100}[0]},
|
||||
},
|
||||
},
|
||||
Status: v1.PodStatus{
|
||||
Conditions: []v1.PodCondition{
|
||||
{
|
||||
Type: v1.PodReady,
|
||||
Status: v1.ConditionTrue,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
oldNode: testutil.NewNode("node1"),
|
||||
newNode: addTaintsToNode(testutil.NewNode("node1"), "testTaint1", "taint1", []int{1, 2}),
|
||||
expectDelete: true,
|
||||
additionalSleep: 1500 * time.Millisecond,
|
||||
},
|
||||
}
|
||||
|
||||
for _, item := range testCases {
|
||||
stopCh := make(chan struct{})
|
||||
fakeClientset := fake.NewSimpleClientset(&v1.PodList{Items: item.pods})
|
||||
controller := NewNoExecuteTaintManager(fakeClientset)
|
||||
controller.recorder = testutil.NewFakeRecorder()
|
||||
go controller.Run(stopCh)
|
||||
controller.NodeUpdated(item.oldNode, item.newNode)
|
||||
// wait a bit
|
||||
time.Sleep(timeForControllerToProgress)
|
||||
if item.additionalSleep > 0 {
|
||||
time.Sleep(item.additionalSleep)
|
||||
}
|
||||
|
||||
podDeleted := false
|
||||
for _, action := range fakeClientset.Actions() {
|
||||
if action.GetVerb() == "delete" && action.GetResource().Resource == "pods" {
|
||||
podDeleted = true
|
||||
}
|
||||
}
|
||||
if podDeleted != item.expectDelete {
|
||||
t.Errorf("%v: Unexepected test result. Expected delete %v, got %v", item.description, item.expectDelete, podDeleted)
|
||||
}
|
||||
close(stopCh)
|
||||
}
|
||||
}
|
||||
|
||||
func TestUpdateNodeWithMultiplePods(t *testing.T) {
|
||||
testCases := []struct {
|
||||
description string
|
||||
pods []v1.Pod
|
||||
oldNode *v1.Node
|
||||
newNode *v1.Node
|
||||
expectedDeleteTimes durationSlice
|
||||
}{
|
||||
{
|
||||
description: "Pods with different toleration times are evicted appropriately",
|
||||
pods: []v1.Pod{
|
||||
*testutil.NewPod("pod1", "node1"),
|
||||
*addToleration(testutil.NewPod("pod2", "node1"), 1, 1),
|
||||
*addToleration(testutil.NewPod("pod3", "node1"), 1, -1),
|
||||
},
|
||||
oldNode: testutil.NewNode("node1"),
|
||||
newNode: addTaintsToNode(testutil.NewNode("node1"), "testTaint1", "taint1", []int{1}),
|
||||
expectedDeleteTimes: durationSlice{
|
||||
{[]string{"pod1"}, 0},
|
||||
{[]string{"pod2"}, time.Second},
|
||||
},
|
||||
},
|
||||
{
|
||||
description: "Evict all pods not matching all taints instantly",
|
||||
pods: []v1.Pod{
|
||||
*testutil.NewPod("pod1", "node1"),
|
||||
*addToleration(testutil.NewPod("pod2", "node1"), 1, 1),
|
||||
*addToleration(testutil.NewPod("pod3", "node1"), 1, -1),
|
||||
},
|
||||
oldNode: testutil.NewNode("node1"),
|
||||
newNode: addTaintsToNode(testutil.NewNode("node1"), "testTaint1", "taint1", []int{1, 2}),
|
||||
expectedDeleteTimes: durationSlice{
|
||||
{[]string{"pod1", "pod2", "pod3"}, 0},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, item := range testCases {
|
||||
t.Logf("Starting testcase %q", item.description)
|
||||
|
||||
stopCh := make(chan struct{})
|
||||
fakeClientset := fake.NewSimpleClientset(&v1.PodList{Items: item.pods})
|
||||
sort.Sort(item.expectedDeleteTimes)
|
||||
controller := NewNoExecuteTaintManager(fakeClientset)
|
||||
controller.recorder = testutil.NewFakeRecorder()
|
||||
go controller.Run(stopCh)
|
||||
controller.NodeUpdated(item.oldNode, item.newNode)
|
||||
|
||||
startedAt := time.Now()
|
||||
for i := range item.expectedDeleteTimes {
|
||||
if i == 0 || item.expectedDeleteTimes[i-1].timestamp != item.expectedDeleteTimes[i].timestamp {
|
||||
// compute a grace duration to give controller time to process updates. Choose big
|
||||
// enough intervals in the test cases above to avoid flakes.
|
||||
var increment time.Duration
|
||||
if i == len(item.expectedDeleteTimes)-1 || item.expectedDeleteTimes[i+1].timestamp == item.expectedDeleteTimes[i].timestamp {
|
||||
increment = 500 * time.Millisecond
|
||||
} else {
|
||||
increment = ((item.expectedDeleteTimes[i+1].timestamp - item.expectedDeleteTimes[i].timestamp) / time.Duration(2))
|
||||
}
|
||||
|
||||
sleepTime := item.expectedDeleteTimes[i].timestamp - time.Since(startedAt) + increment
|
||||
if sleepTime < 0 {
|
||||
sleepTime = 0
|
||||
}
|
||||
t.Logf("Sleeping for %v", sleepTime)
|
||||
time.Sleep(sleepTime)
|
||||
}
|
||||
|
||||
for delay, podName := range item.expectedDeleteTimes[i].names {
|
||||
deleted := false
|
||||
for _, action := range fakeClientset.Actions() {
|
||||
deleteAction, ok := action.(clienttesting.DeleteActionImpl)
|
||||
if !ok {
|
||||
t.Logf("Found not-delete action with verb %v. Ignoring.", action.GetVerb())
|
||||
continue
|
||||
}
|
||||
if deleteAction.GetResource().Resource != "pods" {
|
||||
continue
|
||||
}
|
||||
if podName == deleteAction.GetName() {
|
||||
deleted = true
|
||||
}
|
||||
}
|
||||
if !deleted {
|
||||
t.Errorf("Failed to deleted pod %v after %v", podName, delay)
|
||||
}
|
||||
}
|
||||
for _, action := range fakeClientset.Actions() {
|
||||
deleteAction, ok := action.(clienttesting.DeleteActionImpl)
|
||||
if !ok {
|
||||
t.Logf("Found not-delete action with verb %v. Ignoring.", action.GetVerb())
|
||||
continue
|
||||
}
|
||||
if deleteAction.GetResource().Resource != "pods" {
|
||||
continue
|
||||
}
|
||||
deletedPodName := deleteAction.GetName()
|
||||
expected := false
|
||||
for _, podName := range item.expectedDeleteTimes[i].names {
|
||||
if podName == deletedPodName {
|
||||
expected = true
|
||||
}
|
||||
}
|
||||
if !expected {
|
||||
t.Errorf("Pod %v was deleted even though it shouldn't have", deletedPodName)
|
||||
}
|
||||
}
|
||||
fakeClientset.ClearActions()
|
||||
}
|
||||
|
||||
close(stopCh)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetMinTolerationTime(t *testing.T) {
|
||||
one := int64(1)
|
||||
oneSec := 1 * time.Second
|
||||
|
||||
tests := []struct {
|
||||
tolerations []v1.Toleration
|
||||
expected time.Duration
|
||||
}{
|
||||
{
|
||||
tolerations: []v1.Toleration{},
|
||||
expected: 0,
|
||||
},
|
||||
{
|
||||
tolerations: []v1.Toleration{
|
||||
{
|
||||
TolerationSeconds: &one,
|
||||
},
|
||||
{
|
||||
TolerationSeconds: nil,
|
||||
},
|
||||
},
|
||||
expected: oneSec,
|
||||
},
|
||||
{
|
||||
tolerations: []v1.Toleration{
|
||||
{
|
||||
TolerationSeconds: nil,
|
||||
},
|
||||
{
|
||||
TolerationSeconds: &one,
|
||||
},
|
||||
},
|
||||
expected: oneSec,
|
||||
},
|
||||
}
|
||||
|
||||
for _, test := range tests {
|
||||
got := getMinTolerationTime(test.tolerations)
|
||||
if got != test.expected {
|
||||
t.Errorf("Incorrect min toleration time: got %v, expected %v", got, test.expected)
|
||||
}
|
||||
}
|
||||
}
|
||||
145
vendor/k8s.io/kubernetes/pkg/controller/nodelifecycle/scheduler/timed_workers.go
generated
vendored
145
vendor/k8s.io/kubernetes/pkg/controller/nodelifecycle/scheduler/timed_workers.go
generated
vendored
@ -1,145 +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 scheduler
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"k8s.io/apimachinery/pkg/types"
|
||||
|
||||
"github.com/golang/glog"
|
||||
)
|
||||
|
||||
// WorkArgs keeps arguments that will be passed to the function executed by the worker.
|
||||
type WorkArgs struct {
|
||||
NamespacedName types.NamespacedName
|
||||
}
|
||||
|
||||
// KeyFromWorkArgs creates a key for the given `WorkArgs`
|
||||
func (w *WorkArgs) KeyFromWorkArgs() string {
|
||||
return w.NamespacedName.String()
|
||||
}
|
||||
|
||||
// NewWorkArgs is a helper function to create new `WorkArgs`
|
||||
func NewWorkArgs(name, namespace string) *WorkArgs {
|
||||
return &WorkArgs{types.NamespacedName{Namespace: namespace, Name: name}}
|
||||
}
|
||||
|
||||
// TimedWorker is a responsible for executing a function no earlier than at FireAt time.
|
||||
type TimedWorker struct {
|
||||
WorkItem *WorkArgs
|
||||
CreatedAt time.Time
|
||||
FireAt time.Time
|
||||
Timer *time.Timer
|
||||
}
|
||||
|
||||
// CreateWorker creates a TimedWorker that will execute `f` not earlier than `fireAt`.
|
||||
func CreateWorker(args *WorkArgs, createdAt time.Time, fireAt time.Time, f func(args *WorkArgs) error) *TimedWorker {
|
||||
delay := fireAt.Sub(createdAt)
|
||||
if delay <= 0 {
|
||||
go f(args)
|
||||
return nil
|
||||
}
|
||||
timer := time.AfterFunc(delay, func() { f(args) })
|
||||
return &TimedWorker{
|
||||
WorkItem: args,
|
||||
CreatedAt: createdAt,
|
||||
FireAt: fireAt,
|
||||
Timer: timer,
|
||||
}
|
||||
}
|
||||
|
||||
// Cancel cancels the execution of function by the `TimedWorker`
|
||||
func (w *TimedWorker) Cancel() {
|
||||
if w != nil {
|
||||
w.Timer.Stop()
|
||||
}
|
||||
}
|
||||
|
||||
// TimedWorkerQueue keeps a set of TimedWorkers that are still wait for execution.
|
||||
type TimedWorkerQueue struct {
|
||||
sync.Mutex
|
||||
// map of workers keyed by string returned by 'KeyFromWorkArgs' from the given worker.
|
||||
workers map[string]*TimedWorker
|
||||
workFunc func(args *WorkArgs) error
|
||||
}
|
||||
|
||||
// CreateWorkerQueue creates a new TimedWorkerQueue for workers that will execute
|
||||
// given function `f`.
|
||||
func CreateWorkerQueue(f func(args *WorkArgs) error) *TimedWorkerQueue {
|
||||
return &TimedWorkerQueue{
|
||||
workers: make(map[string]*TimedWorker),
|
||||
workFunc: f,
|
||||
}
|
||||
}
|
||||
|
||||
func (q *TimedWorkerQueue) getWrappedWorkerFunc(key string) func(args *WorkArgs) error {
|
||||
return func(args *WorkArgs) error {
|
||||
err := q.workFunc(args)
|
||||
q.Lock()
|
||||
defer q.Unlock()
|
||||
if err == nil {
|
||||
// To avoid duplicated calls we keep the key in the queue, to prevent
|
||||
// subsequent additions.
|
||||
q.workers[key] = nil
|
||||
} else {
|
||||
delete(q.workers, key)
|
||||
}
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// AddWork adds a work to the WorkerQueue which will be executed not earlier than `fireAt`.
|
||||
func (q *TimedWorkerQueue) AddWork(args *WorkArgs, createdAt time.Time, fireAt time.Time) {
|
||||
key := args.KeyFromWorkArgs()
|
||||
glog.V(4).Infof("Adding TimedWorkerQueue item %v at %v to be fired at %v", key, createdAt, fireAt)
|
||||
|
||||
q.Lock()
|
||||
defer q.Unlock()
|
||||
if _, exists := q.workers[key]; exists {
|
||||
glog.Warningf("Trying to add already existing work for %+v. Skipping.", args)
|
||||
return
|
||||
}
|
||||
worker := CreateWorker(args, createdAt, fireAt, q.getWrappedWorkerFunc(key))
|
||||
q.workers[key] = worker
|
||||
}
|
||||
|
||||
// CancelWork removes scheduled function execution from the queue. Returns true if work was cancelled.
|
||||
func (q *TimedWorkerQueue) CancelWork(key string) bool {
|
||||
q.Lock()
|
||||
defer q.Unlock()
|
||||
worker, found := q.workers[key]
|
||||
result := false
|
||||
if found {
|
||||
glog.V(4).Infof("Cancelling TimedWorkerQueue item %v at %v", key, time.Now())
|
||||
if worker != nil {
|
||||
result = true
|
||||
worker.Cancel()
|
||||
}
|
||||
delete(q.workers, key)
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
// GetWorkerUnsafe returns a TimedWorker corresponding to the given key.
|
||||
// Unsafe method - workers have attached goroutines which can fire afater this function is called.
|
||||
func (q *TimedWorkerQueue) GetWorkerUnsafe(key string) *TimedWorker {
|
||||
q.Lock()
|
||||
defer q.Unlock()
|
||||
return q.workers[key]
|
||||
}
|
||||
141
vendor/k8s.io/kubernetes/pkg/controller/nodelifecycle/scheduler/timed_workers_test.go
generated
vendored
141
vendor/k8s.io/kubernetes/pkg/controller/nodelifecycle/scheduler/timed_workers_test.go
generated
vendored
@ -1,141 +0,0 @@
|
||||
/*
|
||||
Copyright 2017 The Kubernetes Authors.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
package scheduler
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestExecute(t *testing.T) {
|
||||
testVal := int32(0)
|
||||
wg := sync.WaitGroup{}
|
||||
wg.Add(5)
|
||||
queue := CreateWorkerQueue(func(args *WorkArgs) error {
|
||||
atomic.AddInt32(&testVal, 1)
|
||||
wg.Done()
|
||||
return nil
|
||||
})
|
||||
now := time.Now()
|
||||
queue.AddWork(NewWorkArgs("1", "1"), now, now)
|
||||
queue.AddWork(NewWorkArgs("2", "2"), now, now)
|
||||
queue.AddWork(NewWorkArgs("3", "3"), now, now)
|
||||
queue.AddWork(NewWorkArgs("4", "4"), now, now)
|
||||
queue.AddWork(NewWorkArgs("5", "5"), now, now)
|
||||
// Adding the same thing second time should be no-op
|
||||
queue.AddWork(NewWorkArgs("1", "1"), now, now)
|
||||
queue.AddWork(NewWorkArgs("2", "2"), now, now)
|
||||
queue.AddWork(NewWorkArgs("3", "3"), now, now)
|
||||
queue.AddWork(NewWorkArgs("4", "4"), now, now)
|
||||
queue.AddWork(NewWorkArgs("5", "5"), now, now)
|
||||
wg.Wait()
|
||||
lastVal := atomic.LoadInt32(&testVal)
|
||||
if lastVal != 5 {
|
||||
t.Errorf("Espected testVal = 5, got %v", lastVal)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExecuteDelayed(t *testing.T) {
|
||||
testVal := int32(0)
|
||||
wg := sync.WaitGroup{}
|
||||
wg.Add(5)
|
||||
queue := CreateWorkerQueue(func(args *WorkArgs) error {
|
||||
atomic.AddInt32(&testVal, 1)
|
||||
wg.Done()
|
||||
return nil
|
||||
})
|
||||
now := time.Now()
|
||||
then := now.Add(10 * time.Second)
|
||||
queue.AddWork(NewWorkArgs("1", "1"), now, then)
|
||||
queue.AddWork(NewWorkArgs("2", "2"), now, then)
|
||||
queue.AddWork(NewWorkArgs("3", "3"), now, then)
|
||||
queue.AddWork(NewWorkArgs("4", "4"), now, then)
|
||||
queue.AddWork(NewWorkArgs("5", "5"), now, then)
|
||||
queue.AddWork(NewWorkArgs("1", "1"), now, then)
|
||||
queue.AddWork(NewWorkArgs("2", "2"), now, then)
|
||||
queue.AddWork(NewWorkArgs("3", "3"), now, then)
|
||||
queue.AddWork(NewWorkArgs("4", "4"), now, then)
|
||||
queue.AddWork(NewWorkArgs("5", "5"), now, then)
|
||||
wg.Wait()
|
||||
lastVal := atomic.LoadInt32(&testVal)
|
||||
if lastVal != 5 {
|
||||
t.Errorf("Espected testVal = 5, got %v", lastVal)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCancel(t *testing.T) {
|
||||
testVal := int32(0)
|
||||
wg := sync.WaitGroup{}
|
||||
wg.Add(3)
|
||||
queue := CreateWorkerQueue(func(args *WorkArgs) error {
|
||||
atomic.AddInt32(&testVal, 1)
|
||||
wg.Done()
|
||||
return nil
|
||||
})
|
||||
now := time.Now()
|
||||
then := now.Add(10 * time.Second)
|
||||
queue.AddWork(NewWorkArgs("1", "1"), now, then)
|
||||
queue.AddWork(NewWorkArgs("2", "2"), now, then)
|
||||
queue.AddWork(NewWorkArgs("3", "3"), now, then)
|
||||
queue.AddWork(NewWorkArgs("4", "4"), now, then)
|
||||
queue.AddWork(NewWorkArgs("5", "5"), now, then)
|
||||
queue.AddWork(NewWorkArgs("1", "1"), now, then)
|
||||
queue.AddWork(NewWorkArgs("2", "2"), now, then)
|
||||
queue.AddWork(NewWorkArgs("3", "3"), now, then)
|
||||
queue.AddWork(NewWorkArgs("4", "4"), now, then)
|
||||
queue.AddWork(NewWorkArgs("5", "5"), now, then)
|
||||
queue.CancelWork(NewWorkArgs("2", "2").KeyFromWorkArgs())
|
||||
queue.CancelWork(NewWorkArgs("4", "4").KeyFromWorkArgs())
|
||||
wg.Wait()
|
||||
lastVal := atomic.LoadInt32(&testVal)
|
||||
if lastVal != 3 {
|
||||
t.Errorf("Espected testVal = 3, got %v", lastVal)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCancelAndReadd(t *testing.T) {
|
||||
testVal := int32(0)
|
||||
wg := sync.WaitGroup{}
|
||||
wg.Add(4)
|
||||
queue := CreateWorkerQueue(func(args *WorkArgs) error {
|
||||
atomic.AddInt32(&testVal, 1)
|
||||
wg.Done()
|
||||
return nil
|
||||
})
|
||||
now := time.Now()
|
||||
then := now.Add(10 * time.Second)
|
||||
queue.AddWork(NewWorkArgs("1", "1"), now, then)
|
||||
queue.AddWork(NewWorkArgs("2", "2"), now, then)
|
||||
queue.AddWork(NewWorkArgs("3", "3"), now, then)
|
||||
queue.AddWork(NewWorkArgs("4", "4"), now, then)
|
||||
queue.AddWork(NewWorkArgs("5", "5"), now, then)
|
||||
queue.AddWork(NewWorkArgs("1", "1"), now, then)
|
||||
queue.AddWork(NewWorkArgs("2", "2"), now, then)
|
||||
queue.AddWork(NewWorkArgs("3", "3"), now, then)
|
||||
queue.AddWork(NewWorkArgs("4", "4"), now, then)
|
||||
queue.AddWork(NewWorkArgs("5", "5"), now, then)
|
||||
queue.CancelWork(NewWorkArgs("2", "2").KeyFromWorkArgs())
|
||||
queue.CancelWork(NewWorkArgs("4", "4").KeyFromWorkArgs())
|
||||
queue.AddWork(NewWorkArgs("2", "2"), now, then)
|
||||
wg.Wait()
|
||||
lastVal := atomic.LoadInt32(&testVal)
|
||||
if lastVal != 4 {
|
||||
t.Errorf("Espected testVal = 4, got %v", lastVal)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user