Repository: knative/pkg
Pull request #1500 (2020-07-15 18:33:39)

Add a new method to permit creation of buckets from the bucketset
This is a natural factory.

I don't want to mess with Yanwei's PR, so I didn't remove the old method.
But we can reconcile that in the end.

/assign @yanweiguo @evankanderson 

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

This pull-request has been approved by: *<a href="https://github.com/knative/pkg/pull/1500#" 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:

- ~~[hash/OWNERS](https://github.com/knative/pkg/blob/master/hash/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: New changes are detected. LGTM label has been removed.
vagababov: /unhold

Patch:
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/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"


Test patch:
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/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)
 		}
 	}
 }