Repository: knative/pkg
Pull request #1533 (2020-07-21 16:22:56)

Fix #1531 error with stdout/stderr pipes
The #1531 logs to the provided writer the stdout/stderr, providing an utility to log everything that comes in a reader to a logf function. This approach is not working, so i removed the new logging utils and i ported the code to log the result of the stdout/stderr pipes.

knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative/pkg/pull/1533#" title="Author self-approved">slinkydeveloper</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **chaodaig**
You can assign the PR to them by writing `/assign @chaodaig` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative%2Fpkg).

<details open>
Needs approval from an approver in each of these files:

- **[test/OWNERS](https://github.com/knative/pkg/blob/master/test/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["chaodaig"]} -->
n3wscott: Can you add a description on why these are deleted?
knative-prow-robot: @slinkydeveloper: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-pkg-unit-tests | 3f72d43ebd0d09f625fa4300579f4613343ba2a9 | [link](https://prow.knative.dev/view/gcs/knative-prow/pr-logs/pull/knative_pkg/1533/pull-knative-pkg-unit-tests/1285611368610795520) | `/test pull-knative-pkg-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative_pkg/1533). [Your PR dashboard](https://gubernator.knative.dev/pr/slinkydeveloper).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
slinkydeveloper: done
slinkydeveloper: /retest

Patch:
diff --git a/controller/controller.go b/controller/controller.go
index 966a1ffff..551088341 100644
--- a/controller/controller.go
+++ b/controller/controller.go
@@ -30,7 +30,6 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/runtime"
-	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/workqueue"
@@ -186,12 +185,14 @@ type Impl struct {
 	// from the workqueue to process.  Public for testing.
 	Reconciler Reconciler
 
-	// WorkQueue is a rate limited work queue. This is used to queue work to be
-	// processed instead of performing it as soon as a change happens. This
-	// means we can ensure we only process a fixed amount of resources at a
-	// time, and makes it easy to ensure we are never processing the same item
-	// simultaneously in two different workers.
-	WorkQueue workqueue.RateLimitingInterface
+	// workQueue is a rate-limited two-lane work queue.
+	// This is used to queue work to be processed instead of performing it as
+	// soon as a change happens. This means we can ensure we only process a
+	// fixed amount of resources at a time, and makes it easy to ensure we are
+	// never processing the same item simultaneously in two different workers.
+	// The slow queue is used for global resync and other background processes
+	// which are not required to complete at the highest priority.
+	workQueue *twoLaneQueue
 
 	// Sugared logger is easier to use but is not as performant as the
 	// raw logger. In performance critical paths, call logger.Desugar()
@@ -215,12 +216,17 @@ func NewImplWithStats(r Reconciler, logger *zap.SugaredLogger, workQueueName str
 	return &Impl{
 		Name:          workQueueName,
 		Reconciler:    r,
-		WorkQueue:     newTwoLaneWorkQueue(workQueueName),
+		workQueue:     newTwoLaneWorkQueue(workQueueName),
 		logger:        logger,
 		statsReporter: reporter,
 	}
 }
 
+// WorkQueue permits direct access to the work queue.
+func (c *Impl) WorkQueue() workqueue.RateLimitingInterface {
+	return c.workQueue
+}
+
 // EnqueueAfter takes a resource, converts it into a namespace/name string,
 // and passes it to EnqueueKey.
 func (c *Impl) EnqueueAfter(obj interface{}, after time.Duration) {
@@ -232,6 +238,25 @@ func (c *Impl) EnqueueAfter(obj interface{}, after time.Duration) {
 	c.EnqueueKeyAfter(types.NamespacedName{Namespace: object.GetNamespace(), Name: object.GetName()}, after)
 }
 
+// EnqueueSlowKey takes a resource, converts it into a namespace/name string,
+// and enqueues that key in the slow lane.
+func (c *Impl) EnqueueSlowKey(key types.NamespacedName) {
+	c.workQueue.SlowLane().Add(key)
+	c.logger.Debugf("Adding to the slow queue %s (depth(total/slow): %d/%d)", safeKey(key), c.workQueue.Len(), c.workQueue.SlowLane().Len())
+}
+
+// EnqueueSlow extracts namesspeced name from the object and enqueues it on the slow
+// work queue.
+func (c *Impl) EnqueueSlow(obj interface{}) {
+	object, err := kmeta.DeletionHandlingAccessor(obj)
+	if err != nil {
+		c.logger.Errorw("EnqueueSlow", zap.Error(err))
+		return
+	}
+	key := types.NamespacedName{Namespace: object.GetNamespace(), Name: object.GetName()}
+	c.EnqueueSlowKey(key)
+}
+
 // Enqueue takes a resource, converts it into a namespace/name string,
 // and passes it to EnqueueKey.
 func (c *Impl) Enqueue(obj interface{}) {
@@ -342,23 +367,23 @@ func (c *Impl) EnqueueNamespaceOf(obj interface{}) {
 
 // EnqueueKey takes a namespace/name string and puts it onto the work queue.
 func (c *Impl) EnqueueKey(key types.NamespacedName) {
-	c.WorkQueue.Add(key)
-	c.logger.Debugf("Adding to queue %s (depth: %d)", safeKey(key), c.WorkQueue.Len())
+	c.workQueue.Add(key)
+	c.logger.Debugf("Adding to queue %s (depth: %d)", safeKey(key), c.workQueue.Len())
 }
 
-// MaybeEnqueueBucketKey takes a Bucket and namespace/name string and puts it onto the work queue.
+// MaybeEnqueueBucketKey takes a Bucket and namespace/name string and puts it onto
+// the slow work queue.
 func (c *Impl) MaybeEnqueueBucketKey(bkt reconciler.Bucket, key types.NamespacedName) {
 	if bkt.Has(key) {
-		c.WorkQueue.Add(key)
-		c.logger.Debugf("Adding to queue %s (depth: %d)", safeKey(key), c.WorkQueue.Len())
+		c.EnqueueSlowKey(key)
 	}
 }
 
 // EnqueueKeyAfter takes a namespace/name string and schedules its execution in
 // the work queue after given delay.
 func (c *Impl) EnqueueKeyAfter(key types.NamespacedName, delay time.Duration) {
-	c.WorkQueue.AddAfter(key, delay)
-	c.logger.Debugf("Adding to queue %s (delay: %v, depth: %d)", safeKey(key), delay, c.WorkQueue.Len())
+	c.workQueue.AddAfter(key, delay)
+	c.logger.Debugf("Adding to queue %s (delay: %v, depth: %d)", safeKey(key), delay, c.workQueue.Len())
 }
 
 // RunContext starts the controller's worker threads, the number of which is threadiness.
@@ -372,8 +397,8 @@ func (c *Impl) RunContext(ctx context.Context, threadiness int) error {
 	sg := sync.WaitGroup{}
 	defer sg.Wait()
 	defer func() {
-		c.WorkQueue.ShutDown()
-		for c.WorkQueue.Len() > 0 {
+		c.workQueue.ShutDown()
+		for c.workQueue.Len() > 0 {
 			time.Sleep(time.Millisecond * 100)
 		}
 	}()
@@ -423,25 +448,25 @@ func (c *Impl) Run(threadiness int, stopCh <-chan struct{}) error {
 // processNextWorkItem will read a single work item off the workqueue and
 // attempt to process it, by calling Reconcile on our Reconciler.
 func (c *Impl) processNextWorkItem() bool {
-	obj, shutdown := c.WorkQueue.Get()
+	obj, shutdown := c.workQueue.Get()
 	if shutdown {
 		return false
 	}
 	key := obj.(types.NamespacedName)
 	keyStr := safeKey(key)
 
-	c.logger.Debugf("Processing from queue %s (depth: %d)", safeKey(key), c.WorkQueue.Len())
+	c.logger.Debugf("Processing from queue %s (depth: %d)", safeKey(key), c.workQueue.Len())
 
 	startTime := time.Now()
 	// Send the metrics for the current queue depth
-	c.statsReporter.ReportQueueDepth(int64(c.WorkQueue.Len()))
+	c.statsReporter.ReportQueueDepth(int64(c.workQueue.Len()))
 
 	// We call Done here so the workqueue knows we have finished
 	// processing this item. We also must remember to call Forget if
 	// reconcile succeeds. If a transient error occurs, we do not call
 	// Forget and put the item back to the queue with an increased
 	// delay.
-	defer c.WorkQueue.Done(key)
+	defer c.workQueue.Done(key)
 
 	var err error
 	defer func() {
@@ -467,7 +492,7 @@ func (c *Impl) processNextWorkItem() bool {
 
 	// Finally, if no error occurs we Forget this item so it does not
 	// have any delay when another change happens.
-	c.WorkQueue.Forget(key)
+	c.workQueue.Forget(key)
 	logger.Info("Reconcile succeeded. Time taken: ", time.Since(startTime))
 
 	return true
@@ -480,16 +505,16 @@ func (c *Impl) handleErr(err error, key types.NamespacedName) {
 	// We want to check that the queue is shutting down here
 	// since controller Run might have exited by now (since while this item was
 	// being processed, queue.Len==0).
-	if !IsPermanentError(err) && !c.WorkQueue.ShuttingDown() {
-		c.WorkQueue.AddRateLimited(key)
-		c.logger.Debugf("Requeuing key %s due to non-permanent error (depth: %d)", safeKey(key), c.WorkQueue.Len())
+	if !IsPermanentError(err) && !c.workQueue.ShuttingDown() {
+		c.workQueue.AddRateLimited(key)
+		c.logger.Debugf("Requeuing key %s due to non-permanent error (depth: %d)", safeKey(key), c.workQueue.Len())
 		return
 	}
 
-	c.WorkQueue.Forget(key)
+	c.workQueue.Forget(key)
 }
 
-// GlobalResync enqueues (with a delay) all objects from the passed SharedInformer
+// GlobalResync enqueues into the slow lane all objects from the passed SharedInformer
 func (c *Impl) GlobalResync(si cache.SharedInformer) {
 	alwaysTrue := func(interface{}) bool { return true }
 	c.FilteredGlobalResync(alwaysTrue, si)
@@ -498,14 +523,13 @@ func (c *Impl) GlobalResync(si cache.SharedInformer) {
 // FilteredGlobalResync enqueues (with a delay) all objects from the
 // SharedInformer that pass the filter function
 func (c *Impl) FilteredGlobalResync(f func(interface{}) bool, si cache.SharedInformer) {
-	if c.WorkQueue.ShuttingDown() {
+	if c.workQueue.ShuttingDown() {
 		return
 	}
 	list := si.GetStore().List()
-	count := float64(len(list))
 	for _, obj := range list {
 		if f(obj) {
-			c.EnqueueAfter(obj, wait.Jitter(time.Second, count))
+			c.EnqueueSlow(obj)
 		}
 	}
 }
diff --git a/controller/two_lane_queue.go b/controller/two_lane_queue.go
index 165eb2df3..ebd53df5b 100644
--- a/controller/two_lane_queue.go
+++ b/controller/two_lane_queue.go
@@ -16,9 +16,7 @@ limitations under the License.
 
 package controller
 
-import (
-	"k8s.io/client-go/util/workqueue"
-)
+import "k8s.io/client-go/util/workqueue"
 
 // twoLaneQueue is a rate limited queue that wraps around two queues
 // -- fast queue (anonymously aliased), whose contents are processed with priority.
@@ -72,8 +70,8 @@ func process(q workqueue.Interface, ch chan interface{}) {
 		if d {
 			break
 		}
-		ch <- i
 		q.Done(i)
+		ch <- i
 	}
 }
 
diff --git a/hash/bucketer.go b/hash/bucketer.go
index 37ad83980..d0c1460a8 100644
--- a/hash/bucketer.go
+++ b/hash/bucketer.go
@@ -19,7 +19,6 @@ limitations under the License.
 package hash
 
 import (
-	"errors"
 	"sync"
 
 	lru "github.com/hashicorp/golang-lru"
@@ -67,15 +66,6 @@ func NewBucketSet(bucketList sets.String) *BucketSet {
 	}
 }
 
-// NewBucket creates a new bucket. Caller MUST make sure that
-// the given `name` is in the given `bl.buckets`.
-func NewBucket(name string, bl *BucketSet) *Bucket {
-	return &Bucket{
-		name:    name,
-		buckets: bl,
-	}
-}
-
 // Name implements Bucket.
 func (b *Bucket) Name() string {
 	return b.name
@@ -87,17 +77,24 @@ func (b *Bucket) Has(nn types.NamespacedName) bool {
 	return b.buckets.Owner(nn.String()) == b.name
 }
 
-// NewBucket creates a new Bucket with given name based on this bucketset.
-func (bs *BucketSet) NewBucket(name string) (*Bucket, error) {
+// Buckets creates a new list of all possible Bucket based on this bucketset
+// ordered by bucket name.
+func (bs *BucketSet) Buckets() []reconciler.Bucket {
+	bkts := make([]reconciler.Bucket, len(bs.buckets))
+	for i, n := range bs.sortedBucketNames() {
+		bkts[i] = &Bucket{
+			name:    n,
+			buckets: bs,
+		}
+	}
+	return bkts
+}
+
+func (bs *BucketSet) sortedBucketNames() []string {
 	bs.mu.RLock()
 	defer bs.mu.RUnlock()
-	if !bs.buckets.Has(name) {
-		return nil, errors.New(name + " is not a valid bucket in the bucketset")
-	}
-	return &Bucket{
-		name:    name,
-		buckets: bs,
-	}, nil
+
+	return bs.buckets.List()
 }
 
 // Owner returns the owner of the key.
diff --git a/leaderelection/context.go b/leaderelection/context.go
index eaae67a77..5e897bcc7 100644
--- a/leaderelection/context.go
+++ b/leaderelection/context.go
@@ -19,7 +19,6 @@ package leaderelection
 import (
 	"context"
 	"fmt"
-	"hash/fnv"
 	"strings"
 	"sync"
 
@@ -82,11 +81,11 @@ type Elector interface {
 
 // BuildElector builds a leaderelection.LeaderElector for the named LeaderAware
 // reconciler using a builder added to the context via WithStandardLeaderElectorBuilder.
-func BuildElector(ctx context.Context, la reconciler.LeaderAware, name string, enq func(reconciler.Bucket, types.NamespacedName)) (Elector, error) {
+func BuildElector(ctx context.Context, la reconciler.LeaderAware, queueName string, enq func(reconciler.Bucket, types.NamespacedName)) (Elector, error) {
 	if val := ctx.Value(builderKey{}); val != nil {
 		switch builder := val.(type) {
 		case *standardBuilder:
-			return builder.buildElector(ctx, la, name, enq)
+			return builder.buildElector(ctx, la, queueName, enq)
 		case *statefulSetBuilder:
 			return builder.buildElector(ctx, la, enq)
 		}
@@ -107,7 +106,7 @@ type standardBuilder struct {
 }
 
 func (b *standardBuilder) buildElector(ctx context.Context, la reconciler.LeaderAware,
-	name string, enq func(reconciler.Bucket, types.NamespacedName)) (Elector, error) {
+	queueName string, enq func(reconciler.Bucket, types.NamespacedName)) (Elector, error) {
 	logger := logging.FromContext(ctx)
 
 	id, err := UniqueID()
@@ -115,16 +114,12 @@ func (b *standardBuilder) buildElector(ctx context.Context, la reconciler.Leader
 		return nil, err
 	}
 
-	buckets := make([]Elector, 0, b.lec.Buckets)
-	for i := uint32(0); i < b.lec.Buckets; i++ {
-		bkt := &bucket{
-			// The resource name is the lowercase:
-			//   {component}.{workqueue}.{index}-of-{total}
-			name:  strings.ToLower(fmt.Sprintf("%s.%s.%02d-of-%02d", b.lec.Component, name, i, b.lec.Buckets)),
-			index: i,
-			total: b.lec.Buckets,
-		}
-
+	bkts := newStandardBuckets(queueName, b.lec)
+	electors := make([]Elector, 0, b.lec.Buckets)
+	for _, bkt := range bkts {
+		// Use a local var which won't change across the for loop since it is
+		// used in a callback asynchronously.
+		bkt := bkt
 		rl, err := resourcelock.New(KnativeResourceLock,
 			system.Namespace(), // use namespace we are running in
 			bkt.Name(),
@@ -168,9 +163,22 @@ func (b *standardBuilder) buildElector(ctx context.Context, la reconciler.Leader
 		// if lec.WatchDog != nil {
 		// 	lec.WatchDog.SetLeaderElection(le)
 		// }
-		buckets = append(buckets, &runUntilCancelled{Elector: le})
+		electors = append(electors, &runUntilCancelled{Elector: le})
+	}
+	return &runAll{les: electors}, nil
+}
+
+func newStandardBuckets(queueName string, cc ComponentConfig) []reconciler.Bucket {
+	names := make(sets.String, cc.Buckets)
+	for i := uint32(0); i < cc.Buckets; i++ {
+		names.Insert(standardBucketName(i, queueName, cc))
 	}
-	return &runAll{les: buckets}, nil
+
+	return hash.NewBucketSet(names).Buckets()
+}
+
+func standardBucketName(ordinal uint32, queueName string, cc ComponentConfig) string {
+	return strings.ToLower(fmt.Sprintf("%s.%s.%02d-of-%02d", cc.Component, queueName, ordinal, cc.Buckets))
 }
 
 type statefulSetBuilder struct {
@@ -210,7 +218,9 @@ func NewStatefulSetBucketAndSet(buckets int) (reconciler.Bucket, *hash.BucketSet
 	}
 
 	bs := hash.NewBucketSet(names)
-	return hash.NewBucket(statefulSetPodDNS(ssc.StatefulSetID.ordinal, ssc), bs), bs, nil
+	// Buckets is sorted in order of names so we can use ordinal to
+	// get the correct Bucket for this binary.
+	return bs.Buckets()[ssc.StatefulSetID.ordinal], bs, nil
 }
 
 func statefulSetPodDNS(ordinal int, ssc *statefulSetConfig) string {
@@ -269,26 +279,3 @@ func (ruc *runUntilCancelled) Run(ctx context.Context) {
 		}
 	}
 }
-
-type bucket struct {
-	name string
-
-	// We are bucket {index} of {total}
-	index uint32
-	total uint32
-}
-
-var _ reconciler.Bucket = (*bucket)(nil)
-
-// Name implements reconciler.Bucket
-func (b *bucket) Name() string {
-	return b.name
-}
-
-// Has implements reconciler.Bucket
-func (b *bucket) Has(nn types.NamespacedName) bool {
-	h := fnv.New32a()
-	h.Write([]byte(nn.Namespace + "." + nn.Name))
-	ii := h.Sum32() % b.total
-	return b.index == ii
-}


Test patch:
diff --git a/controller/controller_test.go b/controller/controller_test.go
index 629ecc0f0..56e63c676 100644
--- a/controller/controller_test.go
+++ b/controller/controller_test.go
@@ -464,6 +464,17 @@ func TestEnqueues(t *testing.T) {
 			})
 		},
 		wantQueue: []types.NamespacedName{{Namespace: "bar", Name: "foo"}},
+	}, {
+		name: "enqueue resource slow",
+		work: func(impl *Impl) {
+			impl.EnqueueSlow(&Resource{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "foo",
+					Namespace: "bar",
+				},
+			})
+		},
+		wantQueue: []types.NamespacedName{{Namespace: "bar", Name: "foo"}},
 	}, {
 		name: "enqueue sentinel resource",
 		work: func(impl *Impl) {
@@ -720,8 +731,8 @@ func TestEnqueues(t *testing.T) {
 
 			// The rate limit on our queue delays when things are added to the queue.
 			time.Sleep(50 * time.Millisecond)
-			impl.WorkQueue.ShutDown()
-			gotQueue := drainWorkQueue(impl.WorkQueue)
+			impl.WorkQueue().ShutDown()
+			gotQueue := drainWorkQueue(impl.WorkQueue())
 
 			if diff := cmp.Diff(test.wantQueue, gotQueue); diff != "" {
 				t.Errorf("unexpected queue (-want +got): %s", diff)
@@ -730,7 +741,7 @@ func TestEnqueues(t *testing.T) {
 	}
 }
 
-func TestEnqeueAfter(t *testing.T) {
+func TestEnqueueAfter(t *testing.T) {
 	t.Cleanup(ClearAll)
 	impl := NewImplWithStats(&NopReconciler{}, TestLogger(t), "Testing", &FakeStatsReporter{})
 	impl.EnqueueAfter(&Resource{
@@ -752,16 +763,16 @@ func TestEnqeueAfter(t *testing.T) {
 		},
 	}, 20*time.Second)
 	time.Sleep(10 * time.Millisecond)
-	if got, want := impl.WorkQueue.Len(), 0; got != want {
+	if got, want := impl.WorkQueue().Len(), 0; got != want {
 		t.Errorf("|Queue| = %d, want: %d", got, want)
 	}
 	// Sleep the remaining time.
 	time.Sleep(time.Second)
-	if got, want := impl.WorkQueue.Len(), 1; got != want {
+	if got, want := impl.WorkQueue().Len(), 1; got != want {
 		t.Errorf("|Queue| = %d, want: %d", got, want)
 	}
-	impl.WorkQueue.ShutDown()
-	if got, want := drainWorkQueue(impl.WorkQueue), []types.NamespacedName{{Namespace: "the", Name: "waterfall"}}; !cmp.Equal(got, want) {
+	impl.WorkQueue().ShutDown()
+	if got, want := drainWorkQueue(impl.WorkQueue()), []types.NamespacedName{{Namespace: "the", Name: "waterfall"}}; !cmp.Equal(got, want) {
 		t.Errorf("Queue = %v, want: %v, diff: %s", got, want, cmp.Diff(got, want))
 	}
 }
@@ -773,16 +784,16 @@ func TestEnqeueKeyAfter(t *testing.T) {
 	impl.EnqueueKeyAfter(types.NamespacedName{Namespace: "the", Name: "waterfall"}, 500*time.Millisecond)
 	impl.EnqueueKeyAfter(types.NamespacedName{Namespace: "to", Name: "fall"}, 20*time.Second)
 	time.Sleep(10 * time.Millisecond)
-	if got, want := impl.WorkQueue.Len(), 0; got != want {
+	if got, want := impl.WorkQueue().Len(), 0; got != want {
 		t.Errorf("|Queue| = %d, want: %d", got, want)
 	}
 	// Sleep the remaining time.
 	time.Sleep(time.Second)
-	if got, want := impl.WorkQueue.Len(), 1; got != want {
+	if got, want := impl.WorkQueue().Len(), 1; got != want {
 		t.Errorf("|Queue| = %d, want: %d", got, want)
 	}
-	impl.WorkQueue.ShutDown()
-	if got, want := drainWorkQueue(impl.WorkQueue), []types.NamespacedName{{Namespace: "the", Name: "waterfall"}}; !cmp.Equal(got, want) {
+	impl.WorkQueue().ShutDown()
+	if got, want := drainWorkQueue(impl.WorkQueue()), []types.NamespacedName{{Namespace: "the", Name: "waterfall"}}; !cmp.Equal(got, want) {
 		t.Errorf("Queue = %v, want: %v, diff: %s", got, want, cmp.Diff(got, want))
 	}
 }
@@ -1007,7 +1018,7 @@ func TestStartAndShutdownWithWork(t *testing.T) {
 	if got, want := r.count, 1; got != want {
 		t.Errorf("reconcile count = %v, wanted %v", got, want)
 	}
-	if got, want := impl.WorkQueue.NumRequeues(types.NamespacedName{Namespace: "foo", Name: "bar"}), 0; got != want {
+	if got, want := impl.WorkQueue().NumRequeues(types.NamespacedName{Namespace: "foo", Name: "bar"}), 0; got != want {
 		t.Errorf("requeues = %v, wanted %v", got, want)
 	}
 
@@ -1093,7 +1104,7 @@ func TestStartAndShutdownWithErroringWork(t *testing.T) {
 	// As NumRequeues can't fully reflect the real state of queue length.
 	// Here we need to wait for NumRequeues to be more than 1, to ensure
 	// the key get re-queued and reprocessed as expect.
-	if got, wantAtLeast := impl.WorkQueue.NumRequeues(types.NamespacedName{Namespace: "", Name: "bar"}), 2; got < wantAtLeast {
+	if got, wantAtLeast := impl.WorkQueue().NumRequeues(types.NamespacedName{Namespace: "", Name: "bar"}), 2; got < wantAtLeast {
 		t.Errorf("Requeue count = %v, wanted at least %v", got, wantAtLeast)
 	}
 }
@@ -1137,7 +1148,7 @@ func TestStartAndShutdownWithPermanentErroringWork(t *testing.T) {
 	}
 
 	// Check that the work was not requeued in RateLimiter.
-	if got, want := impl.WorkQueue.NumRequeues(types.NamespacedName{Namespace: "foo", Name: "bar"}), 0; got != want {
+	if got, want := impl.WorkQueue().NumRequeues(types.NamespacedName{Namespace: "foo", Name: "bar"}), 0; got != want {
 		t.Errorf("Requeue count = %v, wanted %v", got, want)
 	}
 
diff --git a/controller/two_lane_queue_test.go b/controller/two_lane_queue_test.go
index 0c8bb1260..9c512f35f 100644
--- a/controller/two_lane_queue_test.go
+++ b/controller/two_lane_queue_test.go
@@ -20,14 +20,20 @@ import (
 	"strconv"
 	"testing"
 	"time"
+
+	"k8s.io/apimachinery/pkg/util/wait"
 )
 
 func TestSlowQueue(t *testing.T) {
 	q := newTwoLaneWorkQueue("live-in-the-fast-lane")
 	q.SlowLane().Add("1")
-	if got, want := q.Len(), 1; got != want {
-		t.Errorf("Len = %d, want: 1", got)
+	// Queue has async moving parts so if we check at the wrong moment, this might still be 0.
+	if wait.PollImmediate(10*time.Millisecond, 250*time.Millisecond, func() (bool, error) {
+		return q.Len() == 1, nil
+	}) != nil {
+		t.Error("Queue length was never 1")
 	}
+
 	k, done := q.Get()
 	if got, want := k.(string), "1"; got != want {
 		t.Errorf(`Got = %q, want: "1"`, got)
diff --git a/hash/bucketer_test.go b/hash/bucketer_test.go
index 39c0074f3..18497a087 100644
--- a/hash/bucketer_test.go
+++ b/hash/bucketer_test.go
@@ -21,6 +21,7 @@ import (
 	"sort"
 	"testing"
 
+	"github.com/google/go-cmp/cmp"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
 )
@@ -70,7 +71,7 @@ func TestBucketSetList(t *testing.T) {
 	got := bs.BucketList()
 	sort.Strings(got)
 	if want := buckets.List(); !reflect.DeepEqual(got, want) {
-		t.Errorf("Name = %q, want: %q", got, want)
+		t.Errorf("Name = %q, want: %q, diff(-want,+got):\n%s", got, want, cmp.Diff(want, got))
 	}
 }
 
@@ -101,26 +102,36 @@ func TestBucketSetUpdate(t *testing.T) {
 	}
 }
 
-func TestBucketSetNewBucket(t *testing.T) {
+func TestBucketSetBuckets(t *testing.T) {
 	bs := NewBucketSet(buckets)
-	_, err := bs.NewBucket(thisBucket)
-	if err != nil {
-		t.Error("NewBucket = ", err)
+	bkts := bs.Buckets()
+
+	// Sorted
+	want := []string{"aguacero", "chaparrón", "chubasco", "monsoon"}
+	got := make([]string, len(bkts))
+	for i, b := range bkts {
+		got[i] = b.Name()
 	}
-	_, err = bs.NewBucket("giboulée")
-	if err == nil {
-		t.Error("NewBucket unexpectedly succeeded")
+
+	if !reflect.DeepEqual(got, want) {
+		t.Errorf("Got %q, want %q, diff(-want,+got):\n%s", got, want, cmp.Diff(want, got))
 	}
 }
 
 func TestBucketHas(t *testing.T) {
 	bs := NewBucketSet(buckets)
-	b, _ := bs.NewBucket(thisBucket)
+	b := Bucket{
+		name:    thisBucket,
+		buckets: bs,
+	}
 	thisNN := types.NamespacedName{Namespace: "snow", Name: "hail"}
 	if !b.Has(thisNN) {
 		t.Errorf("Has(%v) = false", thisNN)
 	}
-	b = NewBucket(otherBucket, bs)
+	b = Bucket{
+		name:    otherBucket,
+		buckets: bs,
+	}
 	if b.Has(thisNN) {
 		t.Errorf("Other bucket Has(%v) = true", thisNN)
 	}
@@ -128,7 +139,10 @@ func TestBucketHas(t *testing.T) {
 
 func TestBucketName(t *testing.T) {
 	bs := NewBucketSet(buckets)
-	b := NewBucket(thisBucket, bs)
+	b := Bucket{
+		name:    thisBucket,
+		buckets: bs,
+	}
 	if got, want := b.Name(), thisBucket; got != want {
 		t.Errorf("Name = %q, want: %q", got, want)
 	}
diff --git a/leaderelection/context_test.go b/leaderelection/context_test.go
index e80229557..f2d082e4f 100644
--- a/leaderelection/context_test.go
+++ b/leaderelection/context_test.go
@@ -29,6 +29,7 @@ import (
 
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
 	fakekube "k8s.io/client-go/kubernetes/fake"
 	ktesting "k8s.io/client-go/testing"
 	"knative.dev/pkg/reconciler"
@@ -36,9 +37,10 @@ import (
 )
 
 func TestWithBuilder(t *testing.T) {
+	buckets := 3
 	cc := ComponentConfig{
 		Component:     "the-component",
-		Buckets:       1,
+		Buckets:       uint32(buckets),
 		LeaseDuration: 15 * time.Second,
 		RenewDeadline: 10 * time.Second,
 		RetryPeriod:   2 * time.Second,
@@ -46,15 +48,17 @@ func TestWithBuilder(t *testing.T) {
 	kc := fakekube.NewSimpleClientset()
 	ctx := context.Background()
 
+	gotNames := make(sets.String, 3)
 	promoted := make(chan struct{})
 	demoted := make(chan struct{})
 	laf := &reconciler.LeaderAwareFuncs{
 		PromoteFunc: func(bkt reconciler.Bucket, enq func(reconciler.Bucket, types.NamespacedName)) error {
-			close(promoted)
+			gotNames.Insert(bkt.Name())
+			promoted <- struct{}{}
 			return nil
 		},
 		DemoteFunc: func(bkt reconciler.Bucket) {
-			close(demoted)
+			demoted <- struct{}{}
 		},
 	}
 	enq := func(reconciler.Bucket, types.NamespacedName) {}
@@ -62,7 +66,7 @@ func TestWithBuilder(t *testing.T) {
 	created := make(chan struct{})
 	kc.PrependReactor("create", "leases",
 		func(action ktesting.Action) (bool, runtime.Object, error) {
-			close(created)
+			created <- struct{}{}
 			return false, nil, nil
 		},
 	)
@@ -70,7 +74,7 @@ func TestWithBuilder(t *testing.T) {
 	updated := make(chan struct{})
 	kc.PrependReactor("update", "leases",
 		func(action ktesting.Action) (bool, runtime.Object, error) {
-			// Only close update once.
+			// Only close updated once.
 			select {
 			case <-updated:
 			default:
@@ -116,17 +120,21 @@ func TestWithBuilder(t *testing.T) {
 	t.Cleanup(cancel)
 	go le.Run(ctx)
 
-	select {
-	case <-created:
-		// We expect the lease to be created.
-	case <-time.After(1 * time.Second):
-		t.Fatal("Timed out waiting for lease creation.")
+	// We expect 3 lease to be created.
+	for i := 0; i < buckets; i++ {
+		select {
+		case <-created:
+		case <-time.After(1 * time.Second):
+			t.Fatal("Timed out waiting for lease creation.")
+		}
 	}
-	select {
-	case <-promoted:
-		// We expect to have been promoted.
-	case <-time.After(1 * time.Second):
-		t.Fatal("Timed out waiting for promotion.")
+	// We expect to have been promoted 3 times.
+	for i := 0; i < buckets; i++ {
+		select {
+		case <-promoted:
+		case <-time.After(1 * time.Second):
+			t.Fatal("Timed out waiting for promotion.")
+		}
 	}
 
 	// Cancelling the context should case us to give up leadership.
@@ -138,11 +146,22 @@ func TestWithBuilder(t *testing.T) {
 	case <-time.After(1 * time.Second):
 		t.Fatal("Timed out waiting for lease update.")
 	}
-	select {
-	case <-demoted:
-		// We expect to have been demoted.
-	case <-time.After(1 * time.Second):
-		t.Fatal("Timed out waiting for demotion.")
+	// We expect to have been demoted 3 times.
+	for i := 0; i < buckets; i++ {
+		select {
+		case <-demoted:
+		case <-time.After(1 * time.Second):
+			t.Fatal("Timed out waiting for demotion.")
+		}
+	}
+
+	want := []string{
+		"the-component.name.00-of-03",
+		"the-component.name.01-of-03",
+		"the-component.name.02-of-03",
+	}
+	if got := gotNames.List(); !reflect.DeepEqual(got, want) {
+		t.Errorf("BucketSet.BucketList() = %q, want: %q", got, want)
 	}
 }
 
@@ -181,7 +200,6 @@ func TestNewStatefulSetBucketAndSet(t *testing.T) {
 	if !reflect.DeepEqual(gotNames, wantNames) {
 		t.Errorf("BucketSet.BucketList() = %q, want: %q", gotNames, wantNames)
 	}
-
 }
 
 func TestWithStatefulSetBuilder(t *testing.T) {
diff --git a/test/logging/logger_writer.go b/test/logging/logger_writer.go
deleted file mode 100644
index 92bcc1d83..000000000
--- a/test/logging/logger_writer.go
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
-Copyright 2020 The Knative 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 logging
-
-import (
-	"bytes"
-	"io"
-	"strings"
-)
-
-type loggerWriter struct {
-	prepend string
-	logf    FormatLogger
-
-	buf bytes.Buffer
-}
-
-func NewLoggerWriter(prepend string, logf FormatLogger) io.Writer {
-	return &loggerWriter{
-		prepend: prepend,
-		logf:    logf,
-	}
-}
-
-func (l *loggerWriter) Write(p []byte) (n int, err error) {
-	s := string(p)
-	splitted := strings.Split(s, "\n")
-
-	// Get the remaining of the previous write
-	splitted[0] = l.buf.String() + splitted[0]
-	l.buf.Reset()
-
-	// Write out the lines
-	for i := 0; i < len(splitted)-1; i++ {
-		l.logf(l.prepend + splitted[i])
-	}
-
-	n, err = l.buf.WriteString(splitted[len(splitted)-1])
-	return
-}
diff --git a/test/logging/logger_writer_test.go b/test/logging/logger_writer_test.go
deleted file mode 100644
index bd3d26aa1..000000000
--- a/test/logging/logger_writer_test.go
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
-Copyright 2020 The Knative 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 logging
-
-import (
-	"strings"
-	"testing"
-)
-
-func TestLoggerWriter(t *testing.T) {
-	var strs []string
-	fn := FormatLogger(func(template string, args ...interface{}) {
-		strs = append(strs, template)
-	})
-
-	loggerWriter := NewLoggerWriter("hello: ", fn)
-	_, err := loggerWriter.Write([]byte("aaaa"))
-	if err != nil {
-		t.Fatalf("Error while writing: %v", err)
-	}
-	_, err = loggerWriter.Write([]byte("bbb\nbbb\nbbb"))
-	if err != nil {
-		t.Fatalf("Error while writing: %v", err)
-	}
-	_, err = loggerWriter.Write([]byte("\nccc\n"))
-	if err != nil {
-		t.Fatalf("Error while writing: %v", err)
-	}
-	_, err = loggerWriter.Write([]byte("ddd\n"))
-	if err != nil {
-		t.Fatalf("Error while writing: %v", err)
-	}
-
-	actual := strings.Join(strs, "-")
-	expected := "hello: aaaabbb-hello: bbb-hello: bbb-hello: ccc-hello: ddd"
-	if actual != expected {
-		t.Fatalf("expected: %s, actual: %s", expected, actual)
-	}
-
-}
diff --git a/test/monitoring/monitoring.go b/test/monitoring/monitoring.go
index f0bbafb68..f2fc3e1b8 100644
--- a/test/monitoring/monitoring.go
+++ b/test/monitoring/monitoring.go
@@ -17,6 +17,7 @@ limitations under the License.
 package monitoring
 
 import (
+	"bufio"
 	"fmt"
 	"io"
 	"net"
@@ -60,10 +61,12 @@ func Cleanup(pid int) error {
 }
 
 // PortForward sets up local port forward to the pod specified by the "app" label in the given namespace
-func PortForward(logf logging.FormatLogger, podList *v1.PodList, localPort, remotePort int, namespace string, stdout io.Writer, stderr io.Writer) (int, error) {
+func PortForward(logf logging.FormatLogger, podList *v1.PodList, localPort, remotePort int, namespace string) (int, error) {
 	podName := podList.Items[0].Name
 	portFwdCmd := fmt.Sprintf("kubectl port-forward %s %d:%d -n %s", podName, localPort, remotePort, namespace)
-	portFwdProcess, err := executeCmdBackground(logf, stdout, stderr, portFwdCmd)
+	portFwdProcess, err := executeCmdBackground(func(template string, args ...interface{}) {
+		logf("port-forward-"+podName+": "+template, args...)
+	}, portFwdCmd)
 
 	if err != nil {
 		return 0, fmt.Errorf("failed to port forward: %w", err)
@@ -74,15 +77,31 @@ func PortForward(logf logging.FormatLogger, podList *v1.PodList, localPort, remo
 }
 
 // RunBackground starts a background process and returns the Process if succeed
-func executeCmdBackground(logf logging.FormatLogger, stdout io.Writer, stderr io.Writer, format string, args ...interface{}) (*os.Process, error) {
+func executeCmdBackground(logf logging.FormatLogger, format string, args ...interface{}) (*os.Process, error) {
 	cmd := fmt.Sprintf(format, args...)
 	logf("Executing command: %s", cmd)
 	parts := strings.Split(cmd, " ")
 	c := exec.Command(parts[0], parts[1:]...) // #nosec
-	c.Stdout = stdout
-	c.Stderr = stderr
+	outPipe, err := c.StdoutPipe()
+	if err != nil {
+		return nil, fmt.Errorf("%s cannot get stdout pipe: %w", cmd, err)
+	}
+	go printPipe(logf, outPipe)
+	errPipe, err := c.StderrPipe()
+	if err != nil {
+		return nil, fmt.Errorf("%s cannot get stderr pipe: %w", cmd, err)
+	}
+	go printPipe(logf, errPipe)
 	if err := c.Start(); err != nil {
 		return nil, fmt.Errorf("%s command failed: %w", cmd, err)
 	}
 	return c.Process, nil
 }
+
+func printPipe(logf logging.FormatLogger, reader io.ReadCloser) {
+	defer reader.Close()
+	scanner := bufio.NewScanner(reader)
+	for scanner.Scan() {
+		logf(scanner.Text())
+	}
+}
diff --git a/test/prometheus/prometheus.go b/test/prometheus/prometheus.go
index f6b0ee50b..c04fb6fee 100644
--- a/test/prometheus/prometheus.go
+++ b/test/prometheus/prometheus.go
@@ -70,8 +70,6 @@ func (p *PromProxy) Setup(kubeClientset *kubernetes.Clientset, logf logging.Form
 			prometheusPort,
 			prometheusPort,
 			p.Namespace,
-			logging.NewLoggerWriter("prometheus-port-forward-stdout", logf),
-			logging.NewLoggerWriter("prometheus-port-forward-stderr", logf),
 		)
 		if err != nil {
 			logf("Error starting kubectl port-forward command: %v", err)
diff --git a/test/zipkin/util.go b/test/zipkin/util.go
index 51ecf5c59..155a70d4f 100644
--- a/test/zipkin/util.go
+++ b/test/zipkin/util.go
@@ -132,8 +132,6 @@ func SetupZipkinTracing(kubeClientset *kubernetes.Clientset, logf logging.Format
 			ZipkinPort,
 			zipkinRemotePort,
 			zipkinNamespace,
-			logging.NewLoggerWriter("zipkin-port-forward-stdout", logf),
-			logging.NewLoggerWriter("zipkin-port-forward-stderr", logf),
 		)
 		if e != nil {
 			err = fmt.Errorf("error starting kubectl port-forward command: %w", err)
diff --git a/webhook/certificates/certificates_test.go b/webhook/certificates/certificates_test.go
index 776a8f1bf..c614fc2fe 100644
--- a/webhook/certificates/certificates_test.go
+++ b/webhook/certificates/certificates_test.go
@@ -229,7 +229,7 @@ func TestNew(t *testing.T) {
 		t.Fatal("Expected NewController to return a non-nil value")
 	}
 
-	if want, got := 0, c.WorkQueue.Len(); want != got {
+	if want, got := 0, c.WorkQueue().Len(); want != got {
 		t.Errorf("WorkQueue.Len() = %d, wanted %d", got, want)
 	}
 
@@ -244,7 +244,7 @@ func TestNew(t *testing.T) {
 
 	// Queue has async moving parts so if we check at the wrong moment, this might still be 0.
 	if wait.PollImmediate(10*time.Millisecond, 250*time.Millisecond, func() (bool, error) {
-		return c.WorkQueue.Len() == 1, nil
+		return c.WorkQueue().Len() == 1, nil
 	}) != nil {
 		t.Error("Queue length was never 1")
 	}
diff --git a/webhook/configmaps/table_test.go b/webhook/configmaps/table_test.go
index abb2e9ff7..ff9548b61 100644
--- a/webhook/configmaps/table_test.go
+++ b/webhook/configmaps/table_test.go
@@ -327,7 +327,7 @@ func TestNew(t *testing.T) {
 		t.Fatal("Expected NewController to return a non-nil value")
 	}
 
-	if want, got := 0, c.WorkQueue.Len(); want != got {
+	if want, got := 0, c.WorkQueue().Len(); want != got {
 		t.Errorf("WorkQueue.Len() = %d, wanted %d", got, want)
 	}
 
@@ -342,7 +342,7 @@ func TestNew(t *testing.T) {
 
 	// Queue has async moving parts so if we check at the wrong moment, this might still be 0.
 	if wait.PollImmediate(10*time.Millisecond, 250*time.Millisecond, func() (bool, error) {
-		return c.WorkQueue.Len() == 1, nil
+		return c.WorkQueue().Len() == 1, nil
 	}) != nil {
 		t.Error("Queue length was never 1")
 	}
diff --git a/webhook/psbinding/table_test.go b/webhook/psbinding/table_test.go
index 86952af91..1ae11be65 100644
--- a/webhook/psbinding/table_test.go
+++ b/webhook/psbinding/table_test.go
@@ -1039,7 +1039,7 @@ func TestNew(t *testing.T) {
 		t.Fatal("Expected NewController to return a non-nil value")
 	}
 
-	if want, got := 0, c.WorkQueue.Len(); want != got {
+	if want, got := 0, c.WorkQueue().Len(); want != got {
 		t.Errorf("WorkQueue.Len() = %d, wanted %d", got, want)
 	}
 
@@ -1054,7 +1054,7 @@ func TestNew(t *testing.T) {
 
 	// Queue has async moving parts so if we check at the wrong moment, this might still be 0.
 	if wait.PollImmediate(10*time.Millisecond, 250*time.Millisecond, func() (bool, error) {
-		return c.WorkQueue.Len() == 1, nil
+		return c.WorkQueue().Len() == 1, nil
 	}) != nil {
 		t.Error("Queue length was never 1")
 	}
diff --git a/webhook/resourcesemantics/defaulting/table_test.go b/webhook/resourcesemantics/defaulting/table_test.go
index 3aae9f28d..37c1bedea 100644
--- a/webhook/resourcesemantics/defaulting/table_test.go
+++ b/webhook/resourcesemantics/defaulting/table_test.go
@@ -357,7 +357,7 @@ func TestNew(t *testing.T) {
 		t.Fatal("Expected NewController to return a non-nil value")
 	}
 
-	if want, got := 0, c.WorkQueue.Len(); want != got {
+	if want, got := 0, c.WorkQueue().Len(); want != got {
 		t.Errorf("WorkQueue.Len() = %d, wanted %d", got, want)
 	}
 
@@ -372,7 +372,7 @@ func TestNew(t *testing.T) {
 
 	// Queue has async moving parts so if we check at the wrong moment, this might still be 0.
 	if wait.PollImmediate(10*time.Millisecond, 250*time.Millisecond, func() (bool, error) {
-		return c.WorkQueue.Len() == 1, nil
+		return c.WorkQueue().Len() == 1, nil
 	}) != nil {
 		t.Error("Queue length was never 1")
 	}
diff --git a/webhook/resourcesemantics/validation/reconcile_config_test.go b/webhook/resourcesemantics/validation/reconcile_config_test.go
index ba419df78..bb803208d 100644
--- a/webhook/resourcesemantics/validation/reconcile_config_test.go
+++ b/webhook/resourcesemantics/validation/reconcile_config_test.go
@@ -358,7 +358,7 @@ func TestNew(t *testing.T) {
 		t.Fatal("Expected NewController to return a non-nil value")
 	}
 
-	if want, got := 0, c.WorkQueue.Len(); want != got {
+	if want, got := 0, c.WorkQueue().Len(); want != got {
 		t.Errorf("WorkQueue.Len() = %d, wanted %d", got, want)
 	}
 
@@ -373,7 +373,7 @@ func TestNew(t *testing.T) {
 
 	// Queue has async moving parts so if we check at the wrong moment, thist might still be 0.
 	if wait.PollImmediate(10*time.Millisecond, 250*time.Millisecond, func() (bool, error) {
-		return c.WorkQueue.Len() == 1, nil
+		return c.WorkQueue().Len() == 1, nil
 	}) != nil {
 		t.Error("Queue length was never 1")
 	}
diff --git a/webhook/resourcesemantics/validation/validation_admit_test.go b/webhook/resourcesemantics/validation/validation_admit_test.go
index 87c9dc624..47dc65522 100644
--- a/webhook/resourcesemantics/validation/validation_admit_test.go
+++ b/webhook/resourcesemantics/validation/validation_admit_test.go
@@ -648,7 +648,7 @@ func NewTestResourceAdmissionController(t *testing.T) *reconciler {
 		t.Fatal("Expected NewController to return a non-nil value")
 	}
 
-	if want, got := 0, c.WorkQueue.Len(); want != got {
+	if want, got := 0, c.WorkQueue().Len(); want != got {
 		t.Errorf("WorkQueue.Len() = %d, wanted %d", got, want)
 	}
 
@@ -663,7 +663,7 @@ func NewTestResourceAdmissionController(t *testing.T) *reconciler {
 
 	// Queue has async moving parts so if we check at the wrong moment, this might still be 0.
 	if wait.PollImmediate(10*time.Millisecond, 250*time.Millisecond, func() (bool, error) {
-		return c.WorkQueue.Len() == 1, nil
+		return c.WorkQueue().Len() == 1, nil
 	}) != nil {
 		t.Error("Queue length was never 1")
 	}