Repository: knative/pkg
Pull request #1534 (2020-07-21 20:29:46)

Reorder statements for more precise behavior
So we have a Len() race here. Before we could have len=2, now we can have len=0. I don't know which one's better/worse, but I prefer 0.
Also harden the test to not flake.
/assign mattmoor

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

This pull-request has been approved by: *<a href="https://github.com/knative/pkg/pull/1534#" title="Author self-approved">vagababov</a>*

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

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

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

- ~~[controller/OWNERS](https://github.com/knative/pkg/blob/master/controller/OWNERS)~~ [vagababov]

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":[]} -->
knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative/pkg/pull/1534#pullrequestreview-452818630" title="Approved">mattmoor</a>*, *<a href="https://github.com/knative/pkg/pull/1534#" title="Author self-approved">vagababov</a>*

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

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

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

- ~~[controller/OWNERS](https://github.com/knative/pkg/blob/master/controller/OWNERS)~~ [mattmoor,vagababov]

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":[]} -->

Patch:
diff --git a/controller/two_lane_queue.go b/controller/two_lane_queue.go
index 340a958c4..ebd53df5b 100644
--- a/controller/two_lane_queue.go
+++ b/controller/two_lane_queue.go
@@ -70,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/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) {