Repository: knative/pkg
Pull request #1490 (2020-07-14 20:59:56)

Use hash.BucketSet for standard leader election mode
- Use `hash.Bucket` as the bucket in standard leader election mode.
- Remove the no longer used `leaderelection.bucket` structure.

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

This pull-request has been approved by: *<a href="https://github.com/knative/pkg/pull/1490#" title="Author self-approved">yanweiguo</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **tcnghia**
You can assign the PR to them by writing `/assign @tcnghia` 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:

- **[OWNERS](https://github.com/knative/pkg/blob/master/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":["tcnghia"]} -->
yanweiguo: /assign @mattmoor @vagababov 
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-pkg-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[leaderelection/context.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative_pkg/1490/pull-knative-pkg-go-coverage/1283144364247224321/artifacts/line-cov.html#file0) | 88.2% | 94.4% | 6.2

knative-prow-robot: @yanweiguo: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-pkg-unit-tests | 3988ec77fbad089c719848f5ab789b5d49f2b8f4 | [link](https://prow.knative.dev/view/gcs/knative-prow/pr-logs/pull/knative_pkg/1490/pull-knative-pkg-unit-tests/1283144364247224320) | `/test pull-knative-pkg-unit-tests`

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

<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 -->

Patch:
diff --git a/apis/duck/v1/source_types.go b/apis/duck/v1/source_types.go
index 4ce3eaf60..2f1afe246 100644
--- a/apis/duck/v1/source_types.go
+++ b/apis/duck/v1/source_types.go
@@ -67,7 +67,7 @@ type CloudEventOverrides struct {
 // SourceStatus shows how we expect folks to embed Addressable in
 // their Status field.
 type SourceStatus struct {
-	// inherits duck/v1beta1 Status, which currently provides:
+	// inherits Status, which currently provides:
 	// * ObservedGeneration - the 'Generation' of the Service that was last
 	//   processed by the controller.
 	// * Conditions - the latest available observations of a resource's current
diff --git a/controller/controller.go b/controller/controller.go
index 6024ec253..03f67b6ec 100644
--- a/controller/controller.go
+++ b/controller/controller.go
@@ -452,7 +452,7 @@ func (c *Impl) processNextWorkItem() bool {
 		if err != nil {
 			status = falseString
 		}
-		c.statsReporter.ReportReconcile(time.Since(startTime), keyStr, status)
+		c.statsReporter.ReportReconcile(time.Since(startTime), status)
 	}()
 
 	// Embed the key into the logger and attach that to the context we pass
diff --git a/controller/stats_reporter.go b/controller/stats_reporter.go
index c3445ed3d..6efaeca0f 100644
--- a/controller/stats_reporter.go
+++ b/controller/stats_reporter.go
@@ -46,7 +46,6 @@ var (
 	// - length between 1 and 255 inclusive
 	// - characters are printable US-ASCII
 	reconcilerTagKey = tag.MustNewKey("reconciler")
-	keyTagKey        = tag.MustNewKey("key")
 	successTagKey    = tag.MustNewKey("success")
 )
 
@@ -167,12 +166,12 @@ func init() {
 		Description: "Number of reconcile operations",
 		Measure:     reconcileCountStat,
 		Aggregation: view.Count(),
-		TagKeys:     []tag.Key{reconcilerTagKey, keyTagKey, successTagKey},
+		TagKeys:     []tag.Key{reconcilerTagKey, successTagKey},
 	}, {
 		Description: "Latency of reconcile operations",
 		Measure:     reconcileLatencyStat,
 		Aggregation: reconcileDistribution,
-		TagKeys:     []tag.Key{reconcilerTagKey, keyTagKey, successTagKey},
+		TagKeys:     []tag.Key{reconcilerTagKey, successTagKey},
 	}}
 	views = append(views, wp.DefaultViews()...)
 	views = append(views, rp.DefaultViews()...)
@@ -192,7 +191,7 @@ type StatsReporter interface {
 	ReportQueueDepth(v int64) error
 
 	// ReportReconcile reports the count and latency metrics for a reconcile operation
-	ReportReconcile(duration time.Duration, key, success string) error
+	ReportReconcile(duration time.Duration, success string) error
 }
 
 // Reporter holds cached metric objects to report metrics
@@ -234,11 +233,10 @@ func (r *reporter) ReportQueueDepth(v int64) error {
 }
 
 // ReportReconcile reports the count and latency metrics for a reconcile operation
-func (r *reporter) ReportReconcile(duration time.Duration, key, success string) error {
+func (r *reporter) ReportReconcile(duration time.Duration, success string) error {
 	ctx, err := tag.New(
 		context.Background(),
 		tag.Insert(reconcilerTagKey, r.reconciler),
-		tag.Insert(keyTagKey, key),
 		tag.Insert(successTagKey, success))
 	if err != nil {
 		return err
diff --git a/hash/bucketer.go b/hash/bucketer.go
index bdeeadb4e..37ad83980 100644
--- a/hash/bucketer.go
+++ b/hash/bucketer.go
@@ -19,6 +19,7 @@ limitations under the License.
 package hash
 
 import (
+	"errors"
 	"sync"
 
 	lru "github.com/hashicorp/golang-lru"
@@ -86,36 +87,49 @@ 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) {
+	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
+}
+
 // Owner returns the owner of the key.
 // Owner will cache the results for faster lookup.
-func (b *BucketSet) Owner(key string) string {
-	if v, ok := b.cache.Get(key); ok {
+func (bs *BucketSet) Owner(key string) string {
+	if v, ok := bs.cache.Get(key); ok {
 		return v.(string)
 	}
-	b.mu.RLock()
-	defer b.mu.RUnlock()
-	l := ChooseSubset(b.buckets, 1 /*single query wanted*/, key)
+	bs.mu.RLock()
+	defer bs.mu.RUnlock()
+	l := ChooseSubset(bs.buckets, 1 /*single query wanted*/, key)
 	ret := l.UnsortedList()[0]
-	b.cache.Add(key, ret)
+	bs.cache.Add(key, ret)
 	return ret
 }
 
 // BucketList returns the bucket names of this BucketSet in random order.
-func (b *BucketSet) BucketList() []string {
-	b.mu.RLock()
-	defer b.mu.RUnlock()
+func (bs *BucketSet) BucketList() []string {
+	bs.mu.RLock()
+	defer bs.mu.RUnlock()
 
-	return b.buckets.UnsortedList()
+	return bs.buckets.UnsortedList()
 }
 
 // Update updates the universe of buckets.
-func (b *BucketSet) Update(newB sets.String) {
-	b.mu.Lock()
-	defer b.mu.Unlock()
+func (bs *BucketSet) Update(newB sets.String) {
+	bs.mu.Lock()
+	defer bs.mu.Unlock()
 	// In theory we can iterate over the map and
 	// purge only the keys that moved to a new shard.
 	// But this might be more expensive than re-build
 	// the cache as reconciliations happen.
-	b.cache.Purge()
-	b.buckets = newB
+	bs.cache.Purge()
+	bs.buckets = newB
 }
diff --git a/leaderelection/context.go b/leaderelection/context.go
index eaae67a77..ba9b6dc0d 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,9 @@ 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 {
 		rl, err := resourcelock.New(KnativeResourceLock,
 			system.Namespace(), // use namespace we are running in
 			bkt.Name(),
@@ -168,9 +160,27 @@ 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
+	bs := hash.NewBucketSet(names)
+
+	bkts := make([]reconciler.Bucket, 0, cc.Buckets)
+	for name := range names {
+		bkts = append(bkts, hash.NewBucket(name, bs))
+	}
+	return bkts
+}
+
+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 {
@@ -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
-}
diff --git a/metrics/stackdriver_exporter.go b/metrics/stackdriver_exporter.go
index 1e1a85ae2..96d72dd24 100644
--- a/metrics/stackdriver_exporter.go
+++ b/metrics/stackdriver_exporter.go
@@ -206,76 +206,71 @@ func sdCustomMetricsRecorder(mc metricsConfig, allowCustomMetrics bool) func(con
 		metricskey.LabelClusterName: gm.cluster,
 	}
 	return func(ctx context.Context, mss []stats.Measurement, ros ...stats.Options) error {
-		i := 0
-		var templ *resourceTemplate
-		templateFound := false
+		// Some metrics may be promoted to known Stackdriver schemas, so we may
+		// end up multiple Resources recorded for a single `RecordBatch` call.
+		metricsByResource := map[*resourceTemplate][]stats.Measurement{}
 
-		// This loop serves two purpose, filtering out custom metrics when allowCustomMetrics set to false,
-		// and find the template for the system metrics.
 		for _, m := range mss {
 			metricType := path.Join(mc.stackdriverMetricTypePrefix, m.Measure().Name())
 			t, ok := metricToResourceLabels[metricType]
 			if ok || allowCustomMetrics {
-				mss[i] = m
-				i++
-				if templateFound && templ != t {
-					return fmt.Errorf("mixed resource type measurements in one report: %v", mss)
+				if metricsByResource[t] == nil {
+					metricsByResource[t] = make([]stats.Measurement, 0, len(mss))
 				}
-				templ = t
-				templateFound = true
+				metricsByResource[t] = append(metricsByResource[t], m)
 			}
 		}
-		// Trim the array
-		mss = mss[:i]
-		if i == 0 {
-			return nil
-		}
 
+		baseLabels := map[string]string{}
 		baseResource := metricskey.GetResource(ctx)
-		// Extract resource, if possible
-		if templ != nil {
-			tagMap := tag.FromContext(ctx)
-			r := resource.Resource{
-				Type:   templ.Type,
-				Labels: map[string]string{},
-			}
-			tagMutations := make([]tag.Mutator, 0, len(templ.LabelKeys))
-			if baseResource == nil {
-				baseResource = &resource.Resource{}
-			}
-			for k := range templ.LabelKeys {
-				if v, ok := baseResource.Labels[k]; ok {
-					r.Labels[k] = v
-					continue
+		if baseResource != nil {
+			baseLabels = baseResource.Labels
+		}
+		tagMap := tag.FromContext(ctx)
+		for templ, ms := range metricsByResource {
+			sdResource := baseResource
+			sdCtx := ctx
+			if templ != nil {
+				sdResource = &resource.Resource{
+					Type:   templ.Type,
+					Labels: map[string]string{},
 				}
-				tagKey := tag.MustNewKey(k)
-				if v, ok := tagMap.Value(tagKey); ok {
-					r.Labels[k] = v
-					tagMutations = append(tagMutations, tag.Delete(tagKey))
-					continue
+				tagMutations := make([]tag.Mutator, 0, len(templ.LabelKeys))
+				for k := range templ.LabelKeys {
+					if v, ok := baseLabels[k]; ok {
+						sdResource.Labels[k] = v
+						continue
+					}
+					tagKey := tag.MustNewKey(k)
+					if v, ok := tagMap.Value(tagKey); ok {
+						sdResource.Labels[k] = v
+						tagMutations = append(tagMutations, tag.Delete(tagKey))
+						continue
+					}
+					if v, ok := metadataMap[k]; ok {
+						sdResource.Labels[k] = v
+						continue
+					}
+					sdResource.Labels[k] = metricskey.ValueUnknown
 				}
-				if v, ok := metadataMap[k]; ok {
-					r.Labels[k] = v
-					continue
+				var err error
+				sdCtx, err = tag.New(metricskey.WithResource(ctx, *sdResource), tagMutations...)
+				if err != nil {
+					return err
 				}
-				r.Labels[k] = metricskey.ValueUnknown
 			}
-			var err error
-			ctx, err = tag.New(metricskey.WithResource(ctx, r), tagMutations...)
-			if err != nil {
-				return err
+			if sdResource != nil {
+				opt, err := optionForResource(sdResource)
+				if err != nil {
+					return err
+				}
+				ros = append(ros, opt)
 			}
-			baseResource = &r
-		}
-		if baseResource != nil {
-			opt, err := optionForResource(baseResource)
-			if err != nil {
+			if err := stats.RecordWithOptions(sdCtx, append(ros, stats.WithMeasurements(ms...))...); err != nil {
 				return err
 			}
-			ros = append(ros, opt)
 		}
-
-		return stats.RecordWithOptions(ctx, append(ros, stats.WithMeasurements(mss...))...)
+		return nil
 	}
 }
 
diff --git a/network/network.go b/network/network.go
index 93037189c..c98879abd 100644
--- a/network/network.go
+++ b/network/network.go
@@ -40,7 +40,10 @@ const (
 	// path will wait before shutting down server, but after starting to fail
 	// readiness probes to ensure network layer propagation and so that no requests
 	// are routed to this pod.
-	DefaultDrainTimeout = 30 * time.Second
+	// Note that this was bumped from 30s due to intermittent issues where
+	// the webhook would get a bad request from the API Server when running
+	// under chaos.
+	DefaultDrainTimeout = 45 * time.Second
 
 	// UserAgentKey is the constant for header "User-Agent".
 	UserAgentKey = "User-Agent"
diff --git a/network/transports.go b/network/transports.go
index dcb07a175..c50276259 100644
--- a/network/transports.go
+++ b/network/transports.go
@@ -92,7 +92,7 @@ func dialBackOffHelper(ctx context.Context, network, address string, bo wait.Bac
 	return nil, fmt.Errorf("timed out dialing after %.2fs", elapsed.Seconds())
 }
 
-func newHTTPTransport(connTimeout time.Duration, disableKeepAlives bool) http.RoundTripper {
+func newHTTPTransport(disableKeepAlives bool) http.RoundTripper {
 	return &http.Transport{
 		// Those match net/http/transport.go
 		Proxy:                 http.ProxyFromEnvironment,
@@ -112,7 +112,7 @@ func newHTTPTransport(connTimeout time.Duration, disableKeepAlives bool) http.Ro
 // since it will not cache connections.
 func NewProberTransport() http.RoundTripper {
 	return newAutoTransport(
-		newHTTPTransport(DefaultConnTimeout, true /*disable keep-alives*/),
+		newHTTPTransport(true /*disable keep-alives*/),
 		NewH2CTransport())
 }
 
@@ -120,7 +120,7 @@ func NewProberTransport() http.RoundTripper {
 // based on the request's HTTP version.
 func NewAutoTransport() http.RoundTripper {
 	return newAutoTransport(
-		newHTTPTransport(DefaultConnTimeout, false /*disable keep-alives*/),
+		newHTTPTransport(false /*disable keep-alives*/),
 		NewH2CTransport())
 }
 


Test patch:
diff --git a/apis/testing/roundtrip/roundtrip.go b/apis/testing/roundtrip/roundtrip.go
index ccc030f7e..1bf34e8c2 100644
--- a/apis/testing/roundtrip/roundtrip.go
+++ b/apis/testing/roundtrip/roundtrip.go
@@ -169,7 +169,7 @@ func roundTripViaHub(t *testing.T, gvk schema.GroupVersionKind, scheme, hubs *ru
 	obj = obj.DeepCopyObject().(convertibleObject)
 
 	if !apiequality.Semantic.DeepEqual(original, obj) {
-		t.Errorf("DeepCopy altered the object, diff: %v", diff(original, obj))
+		t.Errorf("DeepCopy modified the original object (DeepCopy should not have side-effects), diff: %v", diff(original, obj))
 		return
 	}
 
@@ -178,7 +178,7 @@ func roundTripViaHub(t *testing.T, gvk schema.GroupVersionKind, scheme, hubs *ru
 	}
 
 	if !apiequality.Semantic.DeepEqual(original, obj) {
-		t.Errorf("Conversion to hub (%s) alterted the object, diff: %v", hubGVK, diff(original, obj))
+		t.Errorf("Conversion to hub (%s) modified the original object (ConvertFrom should not have side-effects), diff: %v", hubGVK, diff(original, obj))
 		return
 	}
 
diff --git a/controller/stats_reporter_test.go b/controller/stats_reporter_test.go
index d82e7fe35..ff56d99a3 100644
--- a/controller/stats_reporter_test.go
+++ b/controller/stats_reporter_test.go
@@ -68,7 +68,6 @@ func TestReportReconcile(t *testing.T) {
 	r, _ := NewStatsReporter("testreconciler")
 	wantTags := map[string]string{
 		"reconciler": "testreconciler",
-		"key":        "test/key",
 		"success":    "true",
 	}
 
@@ -81,11 +80,11 @@ func TestReportReconcile(t *testing.T) {
 		initialReconcileLatency = d[0].Data.(*view.DistributionData).Sum()
 	}
 
-	expectSuccess(t, func() error { return r.ReportReconcile(10*time.Millisecond, "test/key", "true") })
+	expectSuccess(t, func() error { return r.ReportReconcile(10*time.Millisecond, "true") })
 	metricstest.CheckCountData(t, "reconcile_count", wantTags, initialReconcileCount+1)
 	metricstest.CheckDistributionData(t, "reconcile_latency", wantTags, 1, initialReconcileLatency+10, initialReconcileLatency+10)
 
-	expectSuccess(t, func() error { return r.ReportReconcile(15*time.Millisecond, "test/key", "true") })
+	expectSuccess(t, func() error { return r.ReportReconcile(15*time.Millisecond, "true") })
 	metricstest.CheckCountData(t, "reconcile_count", wantTags, initialReconcileCount+2)
 	metricstest.CheckDistributionData(t, "reconcile_latency", wantTags, 2, initialReconcileLatency+10, initialReconcileLatency+15)
 }
diff --git a/controller/testing/fake_stats_reporter.go b/controller/testing/fake_stats_reporter.go
index 506754dc1..528a854b3 100644
--- a/controller/testing/fake_stats_reporter.go
+++ b/controller/testing/fake_stats_reporter.go
@@ -30,8 +30,8 @@ type FakeStatsReporter struct {
 
 // FakeReconcileStatData is used to record the calls to ReportReconcile
 type FakeReconcileStatData struct {
-	Duration     time.Duration
-	Key, Success string
+	Duration time.Duration
+	Success  string
 }
 
 // ReportQueueDepth records the call and returns success.
@@ -43,10 +43,10 @@ func (r *FakeStatsReporter) ReportQueueDepth(v int64) error {
 }
 
 // ReportReconcile records the call and returns success.
-func (r *FakeStatsReporter) ReportReconcile(duration time.Duration, key, success string) error {
+func (r *FakeStatsReporter) ReportReconcile(duration time.Duration, success string) error {
 	r.Lock.Lock()
 	defer r.Lock.Unlock()
-	r.reconcileData = append(r.reconcileData, FakeReconcileStatData{duration, key, success})
+	r.reconcileData = append(r.reconcileData, FakeReconcileStatData{duration, success})
 	return nil
 }
 
diff --git a/controller/testing/fake_stats_reporter_test.go b/controller/testing/fake_stats_reporter_test.go
index dc2f894df..8fde1c7e4 100644
--- a/controller/testing/fake_stats_reporter_test.go
+++ b/controller/testing/fake_stats_reporter_test.go
@@ -38,8 +38,8 @@ func TestReportQueueDepth(t *testing.T) {
 
 func TestReportReconcile(t *testing.T) {
 	r := &FakeStatsReporter{}
-	r.ReportReconcile(time.Duration(123), "testkey", "False")
-	if got, want := r.GetReconcileData(), []FakeReconcileStatData{{time.Duration(123), "testkey", "False"}}; !reflect.DeepEqual(want, got) {
+	r.ReportReconcile(time.Duration(123), "False")
+	if got, want := r.GetReconcileData(), []FakeReconcileStatData{{time.Duration(123), "False"}}; !reflect.DeepEqual(want, got) {
 		t.Errorf("reconcile data len: want: %v, got: %v", want, got)
 	}
 }
diff --git a/hash/bucketer_test.go b/hash/bucketer_test.go
index ee5f074a7..39c0074f3 100644
--- a/hash/bucketer_test.go
+++ b/hash/bucketer_test.go
@@ -101,9 +101,21 @@ func TestBucketSetUpdate(t *testing.T) {
 	}
 }
 
+func TestBucketSetNewBucket(t *testing.T) {
+	bs := NewBucketSet(buckets)
+	_, err := bs.NewBucket(thisBucket)
+	if err != nil {
+		t.Error("NewBucket = ", err)
+	}
+	_, err = bs.NewBucket("giboulée")
+	if err == nil {
+		t.Error("NewBucket unexpectedly succeeded")
+	}
+}
+
 func TestBucketHas(t *testing.T) {
 	bs := NewBucketSet(buckets)
-	b := NewBucket(thisBucket, bs)
+	b, _ := bs.NewBucket(thisBucket)
 	thisNN := types.NamespacedName{Namespace: "snow", Name: "hail"}
 	if !b.Has(thisNN) {
 		t.Errorf("Has(%v) = false", thisNN)
diff --git a/metrics/metricstest/metricstest.go b/metrics/metricstest/metricstest.go
index 9e018604e..83861e61b 100644
--- a/metrics/metricstest/metricstest.go
+++ b/metrics/metricstest/metricstest.go
@@ -30,7 +30,7 @@ import (
 func CheckStatsReported(t test.T, names ...string) {
 	t.Helper()
 	for _, name := range names {
-		d, err := view.RetrieveData(name)
+		d, err := readRowsFromAllMeters(name)
 		if err != nil {
 			t.Error("For metric, Reporter.Report() error", "metric", name, "error", err)
 		}
@@ -45,7 +45,7 @@ func CheckStatsReported(t test.T, names ...string) {
 func CheckStatsNotReported(t test.T, names ...string) {
 	t.Helper()
 	for _, name := range names {
-		d, err := view.RetrieveData(name)
+		d, err := readRowsFromAllMeters(name)
 		// err == nil means a valid stat exists matching "name"
 		// len(d) > 0 means a component recorded metrics for that stat
 		if err == nil && len(d) > 0 {
@@ -166,9 +166,12 @@ func CheckSumData(t test.T, name string, wantTags map[string]string, wantValue f
 //
 // In normal process shutdown, metrics do not need to be unregistered.
 func Unregister(names ...string) {
-	for _, n := range names {
-		if v := view.Find(n); v != nil {
-			view.Unregister(v)
+	for _, producer := range metricproducer.GlobalManager().GetAll() {
+		meter := producer.(view.Meter)
+		for _, n := range names {
+			if v := meter.Find(n); v != nil {
+				meter.Unregister(v)
+			}
 		}
 	}
 }
@@ -180,7 +183,7 @@ func lastRow(t test.T, name string, meter view.Meter) *view.Row {
 	if meter != nil {
 		d, err = meter.RetrieveData(name)
 	} else {
-		d, err = view.RetrieveData(name)
+		d, err = readRowsFromAllMeters(name)
 	}
 	if err != nil {
 		t.Error("Reporter.Report() error", "metric", name, "error", err)
@@ -195,28 +198,32 @@ func lastRow(t test.T, name string, meter view.Meter) *view.Row {
 }
 
 func checkExactlyOneRow(t test.T, name string) (*view.Row, error) {
+	rows, err := readRowsFromAllMeters(name)
+	if err != nil || len(rows) == 0 {
+		return nil, fmt.Errorf("could not find row for %q", name)
+	}
+	if len(rows) > 1 {
+		return nil, fmt.Errorf("expected 1 row for metric %q got %d", name, len(rows))
+	}
+	return rows[0], nil
+}
+
+func readRowsFromAllMeters(name string) ([]*view.Row, error) {
 	// view.Meter implements (and is exposed by) metricproducer.GetAll. Since
 	// this is a test, reach around and cast these to view.Meter.
-	var retval *view.Row
+	var rows []*view.Row
 	for _, producer := range metricproducer.GlobalManager().GetAll() {
 		meter := producer.(view.Meter)
-
 		d, err := meter.RetrieveData(name)
 		if err != nil || len(d) == 0 {
 			continue
 		}
-		if len(d) > 1 {
-			return nil, fmt.Errorf("expected 1 row for metric %q got %d", name, len(d))
-		}
-		if retval != nil {
-			return nil, fmt.Errorf("got 2 rows from different meters: %+v, %+v", *retval, d[0])
+		if rows != nil {
+			return nil, fmt.Errorf("got metrics for the same name from different meters: %+v, %+v", rows, d)
 		}
-		retval = d[0]
-	}
-	if retval == nil {
-		return nil, fmt.Errorf("could not find row for %q", name)
+		rows = d
 	}
-	return retval, nil
+	return rows, nil
 }
 
 func checkRowTags(t test.T, row *view.Row, name string, wantTags map[string]string) {
diff --git a/metrics/resource_view_test.go b/metrics/resource_view_test.go
index e8c5262a1..76b9e23c7 100644
--- a/metrics/resource_view_test.go
+++ b/metrics/resource_view_test.go
@@ -25,6 +25,7 @@ import (
 	"net/http"
 	"os"
 	"sort"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -450,8 +451,11 @@ func TestStackDriverExports(t *testing.T) {
 		"revision_name":      "revision2",
 		"service_name":       "service2",
 	}
-	label3 := map[string]string{
-		"key": "value",
+	batchLabels := map[string]string{
+		"namespace_name":     "ns2",
+		"configuration_name": "config2",
+		"revision_name":      "revision2",
+		"service_name":       "service2",
 	}
 	harness := []struct {
 		name               string
@@ -472,8 +476,8 @@ func TestStackDriverExports(t *testing.T) {
 				2,
 			},
 			{
-				"custom.googleapis.com/knative.dev/autoscaler/testing/value",
-				label3,
+				"custom.googleapis.com/knative.dev/autoscaler/not_ready_pods",
+				batchLabels,
 				3,
 			},
 		},
@@ -516,14 +520,13 @@ func TestStackDriverExports(t *testing.T) {
 				Measure:     desiredPodCountM,
 				Aggregation: view.LastValue(),
 			}
-			customMeasurement := stats.Int64(
-				"testing/value",
-				"Stored value",
+			notReadyPodCountM := stats.Int64(
+				"not_ready_pods",
+				"Number of pods that are not ready",
 				stats.UnitDimensionless)
 			customView := &view.View{
-				Name:        "testing/value",
-				Description: "Test value",
-				Measure:     customMeasurement,
+				Description: "non-knative-revision metric per KnativeRevisionMetrics",
+				Measure:     notReadyPodCountM,
 				Aggregation: view.LastValue(),
 			}
 
@@ -551,23 +554,13 @@ func TestStackDriverExports(t *testing.T) {
 			Record(ctx, actualPodCountM.M(int64(1)))
 
 			r := resource.Resource{
-				Type: "knative_revision",
-				Labels: map[string]string{
-					metricskey.LabelNamespaceName:     "ns2",
-					metricskey.LabelServiceName:       "service2",
-					metricskey.LabelConfigurationName: "config2",
-					metricskey.LabelRevisionName:      "revision2",
-				},
+				Type:   "testing",
+				Labels: batchLabels,
 			}
-			Record(metricskey.WithResource(context.Background(), r), desiredPodCountM.M(int64(2)))
-
-			r = resource.Resource{
-				Type: "custom_resource",
-				Labels: map[string]string{
-					"key": "value",
-				},
-			}
-			Record(metricskey.WithResource(context.Background(), r), customMeasurement.M(int64(3)))
+			RecordBatch(
+				metricskey.WithResource(context.Background(), r),
+				desiredPodCountM.M(int64(2)),
+				notReadyPodCountM.M(int64(3)))
 
 			records := []metricExtract{}
 			for record := range sdFake.published {
@@ -577,6 +570,11 @@ func TestStackDriverExports(t *testing.T) {
 						Labels: ts.Resource.Labels,
 						Value:  ts.Points[0].Value.GetInt64Value(),
 					})
+					if strings.HasPrefix(ts.Metric.Type, "knative.dev/") {
+						if diff := cmp.Diff(ts.Resource.Type, metricskey.ResourceTypeKnativeRevision); diff != "" {
+							t.Errorf("Incorrect resource type for %q: (-want +got): %s", ts.Metric.Type, diff)
+						}
+					}
 				}
 				if len(records) >= 2 {
 					// There's no way to synchronize on the internal timer used
diff --git a/metrics/workqueue_test.go b/metrics/workqueue_test.go
index c08523965..e45359e14 100644
--- a/metrics/workqueue_test.go
+++ b/metrics/workqueue_test.go
@@ -35,6 +35,20 @@ func newFloat64(name string) *stats.Float64Measure {
 	return stats.Float64(name, "bar", "wtfs/s")
 }
 
+type FixedRateLimiter struct {
+	delay time.Duration
+}
+
+func (f *FixedRateLimiter) When(item interface{}) time.Duration {
+	return f.delay
+}
+
+func (f *FixedRateLimiter) Forget(item interface{}) {}
+
+func (f *FixedRateLimiter) NumRequeues(item interface{}) int {
+	return 0
+}
+
 func TestWorkqueueMetrics(t *testing.T) {
 	wp := &WorkqueueProvider{
 		Adds:                           newInt64("adds"),
@@ -60,10 +74,8 @@ func TestWorkqueueMetrics(t *testing.T) {
 	defer view.Unregister(views...)
 
 	queueName := t.Name()
-	wq := workqueue.NewNamedRateLimitingQueue(
-		workqueue.DefaultControllerRateLimiter(),
-		queueName,
-	)
+	limiter := &FixedRateLimiter{delay: 200 * time.Millisecond}
+	wq := workqueue.NewNamedRateLimitingQueue(limiter, queueName)
 
 	metricstest.CheckStatsNotReported(t, "adds", "depth", "latency", "retries", "work_duration",
 		"unfinished_work_seconds", "longest_running_processor_seconds")
@@ -114,7 +126,7 @@ func TestWorkqueueMetrics(t *testing.T) {
 	metricstest.CheckCountData(t, "adds", map[string]string{"name": queueName}, 2)
 
 	// It doesn't show up as an "add" until the rate limit has elapsed.
-	time.Sleep(1 * time.Second)
+	time.Sleep(2 * limiter.delay)
 	metricstest.CheckCountData(t, "adds", map[string]string{"name": queueName}, 3)
 
 	wq.ShutDown()
diff --git a/test/logstream/kubelogs.go b/test/logstream/kubelogs.go
index cd137acdc..c120b7e1c 100644
--- a/test/logstream/kubelogs.go
+++ b/test/logstream/kubelogs.go
@@ -27,6 +27,8 @@ import (
 	"golang.org/x/sync/errgroup"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/apimachinery/pkg/watch"
 
 	"knative.dev/pkg/ptr"
 	"knative.dev/pkg/test"
@@ -35,6 +37,7 @@ import (
 
 type kubelogs struct {
 	namespace string
+	kc        *test.KubeClient
 
 	once sync.Once
 	m    sync.RWMutex
@@ -49,54 +52,80 @@ var _ streamer = (*kubelogs)(nil)
 // timeFormat defines a simple timestamp with millisecond granularity
 const timeFormat = "15:04:05.000"
 
-func (k *kubelogs) init(t test.TLegacy) {
-	k.keys = make(map[string]logger)
+func (k *kubelogs) startForPod(eg *errgroup.Group, pod *corev1.Pod) {
+	// Grab data from all containers in the pods.  We need this in case
+	// an envoy sidecar is injected for mesh installs.  This should be
+	// equivalent to --all-containers.
+	for _, container := range pod.Spec.Containers {
+		// Required for capture below.
+		psn, pn, cn := pod.Namespace, pod.Name, container.Name
+		eg.Go(func() error {
+			options := &corev1.PodLogOptions{
+				Container: cn,
+				// Follow directs the api server to continuously stream logs back.
+				Follow: true,
+				// Only return new logs (this value is being used for "epsilon").
+				SinceSeconds: ptr.Int64(1),
+			}
+
+			req := k.kc.Kube.CoreV1().Pods(psn).GetLogs(pn, options)
+			stream, err := req.Stream()
+			if err != nil {
+				return err
+			}
+			defer stream.Close()
+			// Read this container's stream.
+			scanner := bufio.NewScanner(stream)
+			for scanner.Scan() {
+				k.handleLine(scanner.Bytes())
+			}
+			// Pods get killed with chaos duck, so logs might end
+			// before the test does. So don't report an error here.
+			return nil
+		})
+	}
+}
 
-	kc, err := test.NewKubeClient(test.Flags.Kubeconfig, test.Flags.Cluster)
-	if err != nil {
-		t.Error("Error loading client config", "error", err)
+func podIsReady(p *corev1.Pod) bool {
+	if p.Status.Phase == corev1.PodRunning && p.DeletionTimestamp == nil {
+		for _, cond := range p.Status.Conditions {
+			if cond.Type == corev1.PodReady && cond.Status == corev1.ConditionTrue {
+				return true
+			}
+		}
 	}
+	return false
+}
 
-	// List the pods in the given namespace.
-	pl, err := kc.Kube.CoreV1().Pods(k.namespace).List(metav1.ListOptions{})
+func (k *kubelogs) watchPods(t test.TLegacy) {
+	wi, err := k.kc.Kube.CoreV1().Pods(k.namespace).Watch(metav1.ListOptions{})
 	if err != nil {
-		t.Error("Error listing pods", "error", err)
+		t.Error("Logstream knative pod watch failed, logs might be missing", "error", err)
+		return
 	}
-
 	eg := errgroup.Group{}
-	for _, pod := range pl.Items {
-		// Grab data from all containers in the pods.  We need this in case
-		// an envoy sidecar is injected for mesh installs.  This should be
-		// equivalent to --all-containers.
-		for _, container := range pod.Spec.Containers {
-			// Required for capture below.
-			pod, container := pod, container
-			eg.Go(func() error {
-				options := &corev1.PodLogOptions{
-					Container: container.Name,
-					// Follow directs the api server to continuously stream logs back.
-					Follow: true,
-					// Only return new logs (this value is being used for "epsilon").
-					SinceSeconds: ptr.Int64(1),
-				}
-
-				req := kc.Kube.CoreV1().Pods(k.namespace).GetLogs(pod.Name, options)
-				stream, err := req.Stream()
-				if err != nil {
-					return err
+	go func() {
+		watchedPods := sets.NewString()
+		for ev := range wi.ResultChan() {
+			p := ev.Object.(*corev1.Pod)
+			switch ev.Type {
+			case watch.Deleted:
+				watchedPods.Delete(p.Name)
+			case watch.Added, watch.Modified:
+				if watchedPods.Has(p.Name) {
+					t.Log("Already watching pod", p.Name)
+					continue
 				}
-				defer stream.Close()
-				// Read this container's stream.
-				scanner := bufio.NewScanner(stream)
-				for scanner.Scan() {
-					k.handleLine(scanner.Bytes())
+				if podIsReady(p) {
+					t.Log("Watching logs for pod: ", p.Name)
+					watchedPods.Insert(p.Name)
+					k.startForPod(&eg, p)
+					continue
 				}
-				return fmt.Errorf("logstream completed prematurely for %s/%s: %w",
-					pod.Name, container.Name, scanner.Err())
-			})
+				t.Log("Pod is not yet ready: ", p.Name)
+			}
 		}
-	}
-
+	}()
 	// Monitor the error group in the background and surface an error on the kubelogs
 	// in case anything had an active stream open.
 	go func() {
@@ -108,6 +137,20 @@ func (k *kubelogs) init(t test.TLegacy) {
 	}()
 }
 
+func (k *kubelogs) init(t test.TLegacy) {
+	k.keys = make(map[string]logger, 1)
+
+	kc, err := test.NewKubeClient(test.Flags.Kubeconfig, test.Flags.Cluster)
+	if err != nil {
+		t.Error("Error loading client config", "error", err)
+		return
+	}
+	k.kc = kc
+
+	// watchPods will start logging for existing pods as well.
+	k.watchPods(t)
+}
+
 func (k *kubelogs) handleLine(l []byte) {
 	// This holds the standard structure of our logs.
 	var line struct {
@@ -160,7 +203,7 @@ func (k *kubelogs) handleLine(l []byte) {
 	}
 }
 
-// Start implements streamer
+// Start implements streamer.
 func (k *kubelogs) Start(t test.TLegacy) Canceler {
 	k.once.Do(func() { k.init(t) })
 
@@ -178,7 +221,7 @@ func (k *kubelogs) Start(t test.TLegacy) Canceler {
 		delete(k.keys, name)
 
 		if k.err != nil {
-			t.Error("error during logstream", "error", k.err)
+			t.Error("Error during logstream", "error", k.err)
 		}
 	}
 }