refact(deps): bump k8s and client-go deps to version v0.20.2 (#294)

Signed-off-by: prateekpandey14 <prateek.pandey@mayadata.io>

vendor/k8s.io/client-go/util/workqueue/delaying_queue.go

@@ -35,26 +35,37 @@ type DelayingInterface interface {
 
 // NewDelayingQueue constructs a new workqueue with delayed queuing ability
 func NewDelayingQueue() DelayingInterface {
-	return newDelayingQueue(clock.RealClock{}, "")
+	return NewDelayingQueueWithCustomClock(clock.RealClock{}, "")
 }
 
+// NewDelayingQueueWithCustomQueue constructs a new workqueue with ability to
+// inject custom queue Interface instead of the default one
+func NewDelayingQueueWithCustomQueue(q Interface, name string) DelayingInterface {
+	return newDelayingQueue(clock.RealClock{}, q, name)
+}
+
+// NewNamedDelayingQueue constructs a new named workqueue with delayed queuing ability
 func NewNamedDelayingQueue(name string) DelayingInterface {
-	return newDelayingQueue(clock.RealClock{}, name)
+	return NewDelayingQueueWithCustomClock(clock.RealClock{}, name)
 }
 
-func newDelayingQueue(clock clock.Clock, name string) DelayingInterface {
+// NewDelayingQueueWithCustomClock constructs a new named workqueue
+// with ability to inject real or fake clock for testing purposes
+func NewDelayingQueueWithCustomClock(clock clock.Clock, name string) DelayingInterface {
+	return newDelayingQueue(clock, NewNamed(name), name)
+}
+
+func newDelayingQueue(clock clock.Clock, q Interface, name string) *delayingType {
 	ret := &delayingType{
-		Interface:         NewNamed(name),
-		clock:             clock,
-		heartbeat:         clock.NewTicker(maxWait),
-		stopCh:            make(chan struct{}),
-		waitingForAddCh:   make(chan *waitFor, 1000),
-		metrics:           newRetryMetrics(name),
-		deprecatedMetrics: newDeprecatedRetryMetrics(name),
+		Interface:       q,
+		clock:           clock,
+		heartbeat:       clock.NewTicker(maxWait),
+		stopCh:          make(chan struct{}),
+		waitingForAddCh: make(chan *waitFor, 1000),
+		metrics:         newRetryMetrics(name),
 	}
 
 	go ret.waitingLoop()
-
 	return ret
 }
 
@@ -77,8 +88,7 @@ type delayingType struct {
 	waitingForAddCh chan *waitFor
 
 	// metrics counts the number of retries
-	metrics           retryMetrics
-	deprecatedMetrics retryMetrics
+	metrics retryMetrics
 }
 
 // waitFor holds the data to add and the time it should be added
@@ -154,7 +164,6 @@ func (q *delayingType) AddAfter(item interface{}, duration time.Duration) {
 	}
 
 	q.metrics.retry()
-	q.deprecatedMetrics.retry()
 
 	// immediately add things with no delay
 	if duration <= 0 {
@@ -181,6 +190,9 @@ func (q *delayingType) waitingLoop() {
 	// Make a placeholder channel to use when there are no items in our list
 	never := make(<-chan time.Time)
 
+	// Make a timer that expires when the item at the head of the waiting queue is ready
+	var nextReadyAtTimer clock.Timer
+
 	waitingForQueue := &waitForPriorityQueue{}
 	heap.Init(waitingForQueue)
 
@@ -208,8 +220,12 @@ func (q *delayingType) waitingLoop() {
 		// Set up a wait for the first item's readyAt (if one exists)
 		nextReadyAt := never
 		if waitingForQueue.Len() > 0 {
+			if nextReadyAtTimer != nil {
+				nextReadyAtTimer.Stop()
+			}
 			entry := waitingForQueue.Peek().(*waitFor)
-			nextReadyAt = q.clock.After(entry.readyAt.Sub(now))
+			nextReadyAtTimer = q.clock.NewTimer(entry.readyAt.Sub(now))
+			nextReadyAt = nextReadyAtTimer.C()
 		}
 
 		select {

vendor/k8s.io/client-go/util/workqueue/doc.go

@@ -23,4 +23,4 @@ limitations under the License.
 //  * Multiple consumers and producers. In particular, it is allowed for an
 //      item to be reenqueued while it is being processed.
 //  * Shutdown notifications.
-package workqueue
+package workqueue // import "k8s.io/client-go/util/workqueue"

vendor/k8s.io/client-go/util/workqueue/metrics.go

@@ -87,14 +87,6 @@ type defaultQueueMetrics struct {
 	// how long have current threads been working?
 	unfinishedWorkSeconds   SettableGaugeMetric
 	longestRunningProcessor SettableGaugeMetric
-
-	// TODO(danielqsj): Remove the following metrics, they are deprecated
-	deprecatedDepth                   GaugeMetric
-	deprecatedAdds                    CounterMetric
-	deprecatedLatency                 SummaryMetric
-	deprecatedWorkDuration            SummaryMetric
-	deprecatedUnfinishedWorkSeconds   SettableGaugeMetric
-	deprecatedLongestRunningProcessor SettableGaugeMetric
 }
 
 func (m *defaultQueueMetrics) add(item t) {
@@ -103,9 +95,7 @@ func (m *defaultQueueMetrics) add(item t) {
 	}
 
 	m.adds.Inc()
-	m.deprecatedAdds.Inc()
 	m.depth.Inc()
-	m.deprecatedDepth.Inc()
 	if _, exists := m.addTimes[item]; !exists {
 		m.addTimes[item] = m.clock.Now()
 	}
@@ -117,11 +107,9 @@ func (m *defaultQueueMetrics) get(item t) {
 	}
 
 	m.depth.Dec()
-	m.deprecatedDepth.Dec()
 	m.processingStartTimes[item] = m.clock.Now()
 	if startTime, exists := m.addTimes[item]; exists {
 		m.latency.Observe(m.sinceInSeconds(startTime))
-		m.deprecatedLatency.Observe(m.sinceInMicroseconds(startTime))
 		delete(m.addTimes, item)
 	}
 }
@@ -133,7 +121,6 @@ func (m *defaultQueueMetrics) done(item t) {
 
 	if startTime, exists := m.processingStartTimes[item]; exists {
 		m.workDuration.Observe(m.sinceInSeconds(startTime))
-		m.deprecatedWorkDuration.Observe(m.sinceInMicroseconds(startTime))
 		delete(m.processingStartTimes, item)
 	}
 }
@@ -144,18 +131,14 @@ func (m *defaultQueueMetrics) updateUnfinishedWork() {
 	var total float64
 	var oldest float64
 	for _, t := range m.processingStartTimes {
-		age := m.sinceInMicroseconds(t)
+		age := m.sinceInSeconds(t)
 		total += age
 		if age > oldest {
 			oldest = age
 		}
 	}
-	// Convert to seconds; microseconds is unhelpfully granular for this.
-	total /= 1000000
 	m.unfinishedWorkSeconds.Set(total)
-	m.deprecatedUnfinishedWorkSeconds.Set(total)
-	m.longestRunningProcessor.Set(oldest / 1000000)
-	m.deprecatedLongestRunningProcessor.Set(oldest) // in microseconds.
+	m.longestRunningProcessor.Set(oldest)
 }
 
 type noMetrics struct{}
@@ -165,11 +148,6 @@ func (noMetrics) get(item t)            {}
 func (noMetrics) done(item t)           {}
 func (noMetrics) updateUnfinishedWork() {}
 
-// Gets the time since the specified start in microseconds.
-func (m *defaultQueueMetrics) sinceInMicroseconds(start time.Time) float64 {
-	return float64(m.clock.Since(start).Nanoseconds() / time.Microsecond.Nanoseconds())
-}
-
 // Gets the time since the specified start in seconds.
 func (m *defaultQueueMetrics) sinceInSeconds(start time.Time) float64 {
 	return m.clock.Since(start).Seconds()
@@ -200,13 +178,6 @@ type MetricsProvider interface {
 	NewUnfinishedWorkSecondsMetric(name string) SettableGaugeMetric
 	NewLongestRunningProcessorSecondsMetric(name string) SettableGaugeMetric
 	NewRetriesMetric(name string) CounterMetric
-	NewDeprecatedDepthMetric(name string) GaugeMetric
-	NewDeprecatedAddsMetric(name string) CounterMetric
-	NewDeprecatedLatencyMetric(name string) SummaryMetric
-	NewDeprecatedWorkDurationMetric(name string) SummaryMetric
-	NewDeprecatedUnfinishedWorkSecondsMetric(name string) SettableGaugeMetric
-	NewDeprecatedLongestRunningProcessorMicrosecondsMetric(name string) SettableGaugeMetric
-	NewDeprecatedRetriesMetric(name string) CounterMetric
 }
 
 type noopMetricsProvider struct{}
@@ -239,34 +210,6 @@ func (_ noopMetricsProvider) NewRetriesMetric(name string) CounterMetric {
 	return noopMetric{}
 }
 
-func (_ noopMetricsProvider) NewDeprecatedDepthMetric(name string) GaugeMetric {
-	return noopMetric{}
-}
-
-func (_ noopMetricsProvider) NewDeprecatedAddsMetric(name string) CounterMetric {
-	return noopMetric{}
-}
-
-func (_ noopMetricsProvider) NewDeprecatedLatencyMetric(name string) SummaryMetric {
-	return noopMetric{}
-}
-
-func (_ noopMetricsProvider) NewDeprecatedWorkDurationMetric(name string) SummaryMetric {
-	return noopMetric{}
-}
-
-func (_ noopMetricsProvider) NewDeprecatedUnfinishedWorkSecondsMetric(name string) SettableGaugeMetric {
-	return noopMetric{}
-}
-
-func (_ noopMetricsProvider) NewDeprecatedLongestRunningProcessorMicrosecondsMetric(name string) SettableGaugeMetric {
-	return noopMetric{}
-}
-
-func (_ noopMetricsProvider) NewDeprecatedRetriesMetric(name string) CounterMetric {
-	return noopMetric{}
-}
-
 var globalMetricsFactory = queueMetricsFactory{
 	metricsProvider: noopMetricsProvider{},
 }
@@ -289,21 +232,15 @@ func (f *queueMetricsFactory) newQueueMetrics(name string, clock clock.Clock) qu
 		return noMetrics{}
 	}
 	return &defaultQueueMetrics{
-		clock:                             clock,
-		depth:                             mp.NewDepthMetric(name),
-		adds:                              mp.NewAddsMetric(name),
-		latency:                           mp.NewLatencyMetric(name),
-		workDuration:                      mp.NewWorkDurationMetric(name),
-		unfinishedWorkSeconds:             mp.NewUnfinishedWorkSecondsMetric(name),
-		longestRunningProcessor:           mp.NewLongestRunningProcessorSecondsMetric(name),
-		deprecatedDepth:                   mp.NewDeprecatedDepthMetric(name),
-		deprecatedAdds:                    mp.NewDeprecatedAddsMetric(name),
-		deprecatedLatency:                 mp.NewDeprecatedLatencyMetric(name),
-		deprecatedWorkDuration:            mp.NewDeprecatedWorkDurationMetric(name),
-		deprecatedUnfinishedWorkSeconds:   mp.NewDeprecatedUnfinishedWorkSecondsMetric(name),
-		deprecatedLongestRunningProcessor: mp.NewDeprecatedLongestRunningProcessorMicrosecondsMetric(name),
-		addTimes:                          map[t]time.Time{},
-		processingStartTimes:              map[t]time.Time{},
+		clock:                   clock,
+		depth:                   mp.NewDepthMetric(name),
+		adds:                    mp.NewAddsMetric(name),
+		latency:                 mp.NewLatencyMetric(name),
+		workDuration:            mp.NewWorkDurationMetric(name),
+		unfinishedWorkSeconds:   mp.NewUnfinishedWorkSecondsMetric(name),
+		longestRunningProcessor: mp.NewLongestRunningProcessorSecondsMetric(name),
+		addTimes:                map[t]time.Time{},
+		processingStartTimes:    map[t]time.Time{},
 	}
 }
 
@@ -317,16 +254,6 @@ func newRetryMetrics(name string) retryMetrics {
 	}
 }
 
-func newDeprecatedRetryMetrics(name string) retryMetrics {
-	var ret *defaultRetryMetrics
-	if len(name) == 0 {
-		return ret
-	}
-	return &defaultRetryMetrics{
-		retries: globalMetricsFactory.metricsProvider.NewDeprecatedRetriesMetric(name),
-	}
-}
-
 // SetProvider sets the metrics provider for all subsequently created work
 // queues. Only the first call has an effect.
 func SetProvider(metricsProvider MetricsProvider) {

vendor/k8s.io/client-go/util/workqueue/parallelizer.go

@@ -25,39 +25,77 @@ import (
 
 type DoWorkPieceFunc func(piece int)
 
+type options struct {
+	chunkSize int
+}
+
+type Options func(*options)
+
+// WithChunkSize allows to set chunks of work items to the workers, rather than
+// processing one by one.
+// It is recommended to use this option if the number of pieces significantly
+// higher than the number of workers and the work done for each item is small.
+func WithChunkSize(c int) func(*options) {
+	return func(o *options) {
+		o.chunkSize = c
+	}
+}
+
 // ParallelizeUntil is a framework that allows for parallelizing N
 // independent pieces of work until done or the context is canceled.
-func ParallelizeUntil(ctx context.Context, workers, pieces int, doWorkPiece DoWorkPieceFunc) {
-	var stop <-chan struct{}
-	if ctx != nil {
-		stop = ctx.Done()
+func ParallelizeUntil(ctx context.Context, workers, pieces int, doWorkPiece DoWorkPieceFunc, opts ...Options) {
+	if pieces == 0 {
+		return
+	}
+	o := options{}
+	for _, opt := range opts {
+		opt(&o)
+	}
+	chunkSize := o.chunkSize
+	if chunkSize < 1 {
+		chunkSize = 1
 	}
 
-	toProcess := make(chan int, pieces)
-	for i := 0; i < pieces; i++ {
+	chunks := ceilDiv(pieces, chunkSize)
+	toProcess := make(chan int, chunks)
+	for i := 0; i < chunks; i++ {
 		toProcess <- i
 	}
 	close(toProcess)
 
-	if pieces < workers {
-		workers = pieces
+	var stop <-chan struct{}
+	if ctx != nil {
+		stop = ctx.Done()
+	}
+	if chunks < workers {
+		workers = chunks
 	}
-
 	wg := sync.WaitGroup{}
 	wg.Add(workers)
 	for i := 0; i < workers; i++ {
 		go func() {
 			defer utilruntime.HandleCrash()
 			defer wg.Done()
-			for piece := range toProcess {
-				select {
-				case <-stop:
-					return
-				default:
-					doWorkPiece(piece)
+			for chunk := range toProcess {
+				start := chunk * chunkSize
+				end := start + chunkSize
+				if end > pieces {
+					end = pieces
+				}
+				for p := start; p < end; p++ {
+					select {
+					case <-stop:
+						return
+					default:
+						doWorkPiece(p)
+					}
 				}
 			}
 		}()
 	}
 	wg.Wait()
 }
+
+func ceilDiv(a, b int) int {
+	return (a + b - 1) / b
+}