Repository: knative/pkg
Pull request #1525 (2020-07-20 21:59:02)

Harden the drain tests more
- introduce synthetic timer that gives us the capability to actually trigger events in the drain itself
- unfortunately we still need time.After calls in the test due to the timing for go routines to finish (they race the default: case and lose).

OTOH, the happy runs take virtually no time now.
```
>rt -count=2200
PASS
ok      knative.dev/pkg/network/handlers        1.414s
```

/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/1525#" 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:

- ~~[network/OWNERS](https://github.com/knative/pkg/blob/master/network/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-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
---- |:------------:|:------------:|:-----:
[network/handlers/drain.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative_pkg/1525/pull-knative-pkg-go-coverage/1285333539151155200/artifacts/line-cov.html#file0) | 96.4% | 90.0% | -6.4

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
---- |:------------:|:------------:|:-----:
[network/handlers/drain.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative_pkg/1525/pull-knative-pkg-go-coverage/1285335550672244736/artifacts/line-cov.html#file0) | 96.4% | 90.0% | -6.4

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
---- |:------------:|:------------:|:-----:
[network/handlers/drain.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative_pkg/1525/pull-knative-pkg-go-coverage/1285340080608645120/artifacts/line-cov.html#file0) | 96.4% | 90.0% | -6.4

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
---- |:------------:|:------------:|:-----:
[network/handlers/drain.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative_pkg/1525/pull-knative-pkg-go-coverage/1285342849189023747/artifacts/line-cov.html#file0) | 96.4% | 90.0% | -6.4

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
---- |:------------:|:------------:|:-----:
[network/handlers/drain.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative_pkg/1525/pull-knative-pkg-go-coverage/1285343855960395777/artifacts/line-cov.html#file0) | 96.4% | 93.3% | -3.1

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

This pull-request has been approved by: *<a href="https://github.com/knative/pkg/pull/1525#pullrequestreview-452010134" title="Approved">mattmoor</a>*, *<a href="https://github.com/knative/pkg/pull/1525#" 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:

- ~~[network/OWNERS](https://github.com/knative/pkg/blob/master/network/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/network/handlers/drain.go b/network/handlers/drain.go
index b85516088..42a6f25a6 100644
--- a/network/handlers/drain.go
+++ b/network/handlers/drain.go
@@ -24,6 +24,28 @@ import (
 	"knative.dev/pkg/network"
 )
 
+type timer interface {
+	Stop() bool
+	Reset(time.Duration) bool
+	tickChan() <-chan time.Time
+}
+
+type sysTimer struct {
+	*time.Timer
+}
+
+func (s *sysTimer) tickChan() <-chan time.Time {
+	return s.Timer.C
+}
+
+// This constructor is overridden in tests to control the progress
+// of time in the test.
+var newTimer = func(d time.Duration) timer {
+	return &sysTimer{
+		time.NewTimer(d),
+	}
+}
+
 // Drainer wraps an inner http.Handler to support responding to kubelet
 // probes with a "200 OK" until the handler is told to Drain.
 // When the Drainer is told to Drain, it will immediately start to fail
@@ -45,7 +67,7 @@ type Drainer struct {
 	once sync.Once
 
 	// timer is used to orchestrate the drain.
-	timer *time.Timer
+	timer timer
 }
 
 // Ensure Drainer implements http.Handler
@@ -69,18 +91,20 @@ func (d *Drainer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 // Drain blocks until QuietPeriod has elapsed since the last request,
 // starting when this is invoked.
 func (d *Drainer) Drain() {
+	// Note: until the first caller exits, the others
+	// will wait blocked as well.
 	d.once.Do(func() {
-		t := func() *time.Timer {
+		t := func() timer {
 			d.Lock()
 			defer d.Unlock()
 			if d.QuietPeriod <= 0 {
 				d.QuietPeriod = network.DefaultDrainTimeout
 			}
-			d.timer = time.NewTimer(d.QuietPeriod)
+			d.timer = newTimer(d.QuietPeriod)
 			return d.timer
 		}()
 
-		<-t.C
+		<-t.tickChan()
 	})
 }
 


Test patch:
diff --git a/network/handlers/drain_test.go b/network/handlers/drain_test.go
index e73331b34..fd200d7e4 100644
--- a/network/handlers/drain_test.go
+++ b/network/handlers/drain_test.go
@@ -25,6 +25,44 @@ import (
 	"knative.dev/pkg/network"
 )
 
+type mockTimer struct {
+	now        time.Time // our current time.
+	deadline   time.Time // when we're supposed to fire
+	c          chan time.Time
+	resetCalls int
+	stopped    bool
+}
+
+func (mt *mockTimer) advance(d time.Duration) {
+	mt.now = mt.now.Add(d)
+	if !mt.now.Before(mt.deadline) {
+		mt.stopped = true
+		mt.c <- mt.now
+	}
+}
+
+func (mt *mockTimer) Reset(d time.Duration) bool {
+	mt.resetCalls++
+	if mt.stopped {
+		mt.now = time.Now()
+		mt.deadline = mt.now.Add(d)
+		mt.stopped = false
+	}
+	return !mt.stopped
+}
+
+func (mt *mockTimer) Stop() bool {
+	if mt.stopped {
+		return false
+	}
+	mt.stopped = true
+	return true
+}
+
+func (mt *mockTimer) tickChan() <-chan time.Time {
+	return mt.c
+}
+
 func TestDrainMechanics(t *testing.T) {
 	var (
 		w     http.ResponseWriter
@@ -34,19 +72,45 @@ func TestDrainMechanics(t *testing.T) {
 				"User-Agent": []string{network.KubeProbeUAPrefix},
 			},
 		}
+		cnt   = 0
+		inner = http.HandlerFunc(func(http.ResponseWriter, *http.Request) { cnt++ })
 	)
 
-	inner := http.HandlerFunc(func(http.ResponseWriter, *http.Request) {})
+	const (
+		timeout = 100 * time.Millisecond
+		epsilon = time.Nanosecond
+	)
 
+	// We need init channel to signal the main thread that the drain
+	// has been initialized in the background thread.
+	init := make(chan struct{})
+	nt := newTimer
+	t.Cleanup(func() {
+		newTimer = nt
+	})
+	// The mock timer will only fire when we advance it past timeout.
+	mt := &mockTimer{
+		c: make(chan time.Time),
+	}
+	newTimer = func(d time.Duration) timer {
+		// When we close the init channel, we know that first drain has been called, and the test can progress.
+		defer close(init)
+		mt.now = time.Now()
+		mt.deadline = mt.now.Add(d)
+		return mt
+	}
 	drainer := &Drainer{
 		Inner:       inner,
-		QuietPeriod: 100 * time.Millisecond,
+		QuietPeriod: timeout,
 	}
 
 	// Works before Drain is called.
 	drainer.ServeHTTP(w, req)
 	drainer.ServeHTTP(w, req)
 	drainer.ServeHTTP(w, req)
+	if cnt != 3 {
+		t.Error("Inner handler was not properly invoked")
+	}
 
 	// Check for 200 OK.
 	resp := httptest.NewRecorder()
@@ -55,7 +119,7 @@ func TestDrainMechanics(t *testing.T) {
 		t.Errorf("Probe status = %d, wanted %d", got, want)
 	}
 
-	// Start to drain, and cancel the context when it returns.
+	// Start to drain, and close the channel when it returns.
 	done := make(chan struct{})
 	go func() {
 		defer close(done)
@@ -63,13 +127,19 @@ func TestDrainMechanics(t *testing.T) {
 	}()
 
 	select {
-	case <-time.After(40 * time.Millisecond):
-		// Drain is blocking.
 	case <-done:
 		t.Error("Drain terminated prematurely.")
+	case <-init:
+		// OK.
 	}
+	mt.advance(timeout - epsilon)
+
 	// Now send a request to reset things.
+	rc := mt.resetCalls
 	drainer.ServeHTTP(w, req)
+	if mt.resetCalls != rc+1 {
+		t.Errorf("ResetCalls = %d, want: %d", mt.resetCalls, rc+1)
+	}
 
 	// Check for 503 as a probe response when shutting down.
 	resp = httptest.NewRecorder()
@@ -77,19 +147,32 @@ func TestDrainMechanics(t *testing.T) {
 	if got, want := resp.Code, http.StatusServiceUnavailable; got != want {
 		t.Errorf("Probe status = %d, wanted %d", got, want)
 	}
+	// Verify no reset was called.
+	if got, want := mt.resetCalls, rc+1; got != want {
+		t.Errorf("ResetCalls = %d, want: %d", got, want)
+	}
+	rc++
 
 	for i := 0; i < 3; i++ {
+		mt.advance(timeout - epsilon)
 		select {
-		case <-time.After(40 * time.Millisecond):
-			// Drain is blocking.
 		case <-done:
 			t.Error("Drain terminated prematurely.")
+		default:
+			// OK
 		}
 		// For the last one we don't want to reset the drain timer.
 		if i < 2 {
 			drainer.ServeHTTP(w, req)
+
+			// Two more drains should have been called.
+			if got, want := mt.resetCalls, rc+1; got != want {
+				t.Errorf("ResetCalls = %d, want: %d", got, want)
+			}
+			rc++
 		}
 	}
+
 	// Probing does not reset the clock.
 	// Check for 503 on a probe when shutting down.
 	resp = httptest.NewRecorder()
@@ -116,13 +199,23 @@ func TestDrainMechanics(t *testing.T) {
 	}()
 
 	select {
-	case <-time.After(80 * time.Millisecond):
-		t.Error("Timed out waiting for Drain to return.")
 	case <-done:
 	case <-done1:
 	case <-done2:
 	case <-done3:
-		// Once the first context is cancelled, check that all of them are cancelled.
+	default:
+		// Expected.
+	}
+
+	// Finally we made it there!
+	mt.advance(epsilon)
+	select {
+	case <-done:
+	case <-done1:
+	case <-done2:
+	case <-done3:
+	case <-time.After(time.Second): // We can't use default here, since it will race the tick in the drainer.
+		t.Error("Drains should have happened!")
 	}
 
 	// Check that a 4th and final one after things complete finishes instantly.
@@ -132,15 +225,43 @@ func TestDrainMechanics(t *testing.T) {
 		drainer.Drain()
 	}()
 
-	// Give the test a short window to launch and execute the go routine.
-	time.Sleep(5 * time.Millisecond)
-
+	// We need to ensure all the go routines complete, so give them ample time.
 	for idx, dch := range []chan struct{}{done, done1, done2, done3, done4} {
 		select {
 		case <-dch:
 			// Should be done.
-		default:
+		case <-time.After(time.Second):
 			t.Errorf("Drain[%d] did not complete.", idx)
 		}
 	}
 }
+
+func TestDefaultQuietPeriod(t *testing.T) {
+	nt := newTimer
+	t.Cleanup(func() {
+		newTimer = nt
+	})
+	mt := &mockTimer{
+		c: make(chan time.Time),
+	}
+	init := make(chan struct{})
+	newTimer = func(d time.Duration) timer {
+		defer close(init)
+		mt.now = time.Now()
+		mt.deadline = mt.now.Add(d)
+		return mt
+	}
+	drainer := &Drainer{
+		Inner: http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}),
+	}
+	go drainer.Drain()
+	select {
+	case <-init:
+		if got, want := mt.deadline.Sub(mt.now), network.DefaultDrainTimeout; got != want {
+			t.Errorf("DefaultDrainTimeout = %v, want: %v", got, want)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("Failed to call drain in 1s")
+	}
+	mt.advance(network.DefaultDrainTimeout)
+}