Repository: knative-sandbox/eventing-kafka-broker
Pull request #1917 (2022-02-16 11:55:57)

[release-1.1] Add expected status to prober
This is an automated cherry-pick of #1914

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1917#" title="Author self-approved">knative-prow-robot</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **markusthoemmes** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @markusthoemmes` 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-sandbox%2Feventing-kafka-broker).

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

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/release-1.1/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":["markusthoemmes"]} -->
knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1917#" title="Author self-approved">knative-prow-robot</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1917#pullrequestreview-884424509" title="Approved">matzew</a>*

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

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:

- ~~[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/release-1.1/OWNERS)~~ [matzew]

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-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/channel/channel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1917/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1493917063188582400/artifacts/line-cov.html#file4) | 81.9% | 78.6% | -3.3

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-build-tests | b28517fe7e4e60867b9028e8c370c4b045c38d40 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1917/pull-knative-sandbox-eventing-kafka-broker-build-tests/1493917063020810240) | true | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1917). [Your PR dashboard](https://gubernator.knative.dev/pr/knative-prow-robot).

<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 -->
knative-prow-robot: @knative-prow-robot: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-build-tests | b28517fe7e4e60867b9028e8c370c4b045c38d40 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1917/pull-knative-sandbox-eventing-kafka-broker-build-tests/1493917063020810240) | true | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`
pull-knative-sandbox-eventing-kafka-broker-unit-tests | b28517fe7e4e60867b9028e8c370c4b045c38d40 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1917/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1493917063058558976) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1917). [Your PR dashboard](https://gubernator.knative.dev/pr/knative-prow-robot).

<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 -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1917?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> :exclamation: No coverage uploaded for pull request base (`release-1.1@752199a`). [Click here to learn what that means](https://docs.codecov.io/docs/error-reference?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#section-missing-base-commit).
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1917/graphs/tree.svg?width=650&height=150&src=pr&token=xyPIgGOiwy&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1917?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@              Coverage Diff               @@
##             release-1.1    #1917   +/-   ##
==============================================
  Coverage               ?   66.61%           
==============================================
  Files                  ?       52           
  Lines                  ?     2480           
  Branches               ?        0           
==============================================
  Hits                   ?     1652           
  Misses                 ?      653           
  Partials               ?      175           
```



------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1917?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1917?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [752199a...b28517f](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1917?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).

knative-prow-robot: @knative-prow-robot: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-build-tests | b28517fe7e4e60867b9028e8c370c4b045c38d40 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1917/pull-knative-sandbox-eventing-kafka-broker-build-tests/1493917063020810240) | true | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`
pull-knative-sandbox-eventing-kafka-broker-unit-tests | b28517fe7e4e60867b9028e8c370c4b045c38d40 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1917/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1493917063058558976) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-integration-tests | b28517fe7e4e60867b9028e8c370c4b045c38d40 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1917/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1493917063092113408) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1917). [Your PR dashboard](https://gubernator.knative.dev/pr/knative-prow-robot).

<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 -->
pierDipi: /retest-required

Patch:
diff --git a/control-plane/pkg/prober/async_prober.go b/control-plane/pkg/prober/async_prober.go
index 56daf37f5..1d6d2ed6d 100644
--- a/control-plane/pkg/prober/async_prober.go
+++ b/control-plane/pkg/prober/async_prober.go
@@ -59,7 +59,7 @@ func NewAsync(ctx context.Context, client httpClient, port string, podsLabelsSel
 	}
 }
 
-func (a *asyncProber) Probe(ctx context.Context, addressable Addressable) Status {
+func (a *asyncProber) Probe(ctx context.Context, addressable Addressable, expected Status) Status {
 	address := addressable.Address
 	pods, err := a.podLister()
 	if err != nil {
@@ -98,8 +98,8 @@ func (a *asyncProber) Probe(ctx context.Context, addressable Addressable) Status
 			With(zap.String("address", address))
 
 		currentStatus := a.cache.GetStatus(address)
-		if currentStatus == StatusReady {
-			logger.Debug("Skip probing", zap.String("status", "ready"))
+		if currentStatus == expected {
+			logger.Debug("Skip probing, status in cache is equal to the expected status", zap.String("status", currentStatus.String()))
 			wg.Done()
 			continue
 		}
@@ -112,7 +112,7 @@ func (a *asyncProber) Probe(ctx context.Context, addressable Addressable) Status
 		// We want to requeue the resource only once when we have all probe request
 		// results.
 		enqueueOnce.Do(func() {
-			// Wait in a separate gorouting.
+			// Wait in a separate goroutine.
 			go func() {
 				// Wait for all the prober request results and then requeue the
 				// resource.
@@ -133,7 +133,7 @@ func (a *asyncProber) Probe(ctx context.Context, addressable Addressable) Status
 
 	if aggregatedCurrentKnownStatus == nil {
 		// Every status is ready, return ready.
-		return StatusReady
+		return expected
 	}
 	return *aggregatedCurrentKnownStatus
 }
diff --git a/control-plane/pkg/prober/prober.go b/control-plane/pkg/prober/prober.go
index 59679808b..857693d9c 100644
--- a/control-plane/pkg/prober/prober.go
+++ b/control-plane/pkg/prober/prober.go
@@ -40,18 +40,18 @@ type EnqueueFunc func(key types.NamespacedName)
 // Prober probes an addressable resource.
 type Prober interface {
 	// Probe probes the provided Addressable resource and returns its Status.
-	Probe(ctx context.Context, addressable Addressable) Status
+	Probe(ctx context.Context, addressable Addressable, expected Status) Status
 }
 
 // Func type is an adapter to allow the use of
 // ordinary functions as Prober. If f is a function
 // with the appropriate signature, Func(f) is a
 // Prober that calls f.
-type Func func(ctx context.Context, addressable Addressable) Status
+type Func func(ctx context.Context, addressable Addressable, expected Status) Status
 
 // Probe implements the Prober interface for Func.
-func (p Func) Probe(ctx context.Context, addressable Addressable) Status {
-	return p(ctx, addressable)
+func (p Func) Probe(ctx context.Context, addressable Addressable, expected Status) Status {
+	return p(ctx, addressable, expected)
 }
 
 // httpClient interface is an interface for an HTTP client.
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index 47175d86d..60dc38987 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -232,7 +232,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 		},
 	}
 
-	if status := r.Prober.Probe(ctx, proberAddressable); status != prober.StatusReady {
+	if status := r.Prober.Probe(ctx, proberAddressable, prober.StatusReady); status != prober.StatusReady {
 		statusConditionManager.ProbesStatusNotReady(status)
 		return nil // Object will get re-queued once probe status changes.
 	}
@@ -299,7 +299,7 @@ func (r *Reconciler) finalizeKind(ctx context.Context, broker *eventing.Broker)
 			Name:      broker.GetName(),
 		},
 	}
-	if status := r.Prober.Probe(ctx, proberAddressable); status != prober.StatusNotReady {
+	if status := r.Prober.Probe(ctx, proberAddressable, prober.StatusNotReady); status != prober.StatusNotReady {
 		// Return a requeueKeyError that doesn't generate an event and it re-queues the object
 		// for a new reconciliation.
 		return controller.NewRequeueAfter(5 * time.Second)
diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index 45f78be55..9e3917944 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -275,7 +275,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, channel *messagingv1beta
 		},
 	}
 
-	if status := r.Prober.Probe(ctx, proberAddressable); status != prober.StatusReady {
+	if status := r.Prober.Probe(ctx, proberAddressable, prober.StatusReady); status != prober.StatusReady {
 		statusConditionManager.ProbesStatusNotReady(status)
 		return nil // Object will get re-queued once probe status changes.
 	}
@@ -352,7 +352,7 @@ func (r *Reconciler) finalizeKind(ctx context.Context, channel *messagingv1beta1
 			Name:      channel.GetName(),
 		},
 	}
-	if status := r.Prober.Probe(ctx, proberAddressable); status != prober.StatusNotReady {
+	if status := r.Prober.Probe(ctx, proberAddressable, prober.StatusNotReady); status != prober.StatusNotReady {
 		// Return a requeueKeyError that doesn't generate an event and it re-queues the object
 		// for a new reconciliation.
 		return controller.NewRequeueAfter(5 * time.Second)
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink.go b/control-plane/pkg/reconciler/sink/kafka_sink.go
index 64e14fe58..19ee66d0f 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink.go
@@ -232,7 +232,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *eventing.KafkaSink)
 		},
 	}
 
-	if status := r.Prober.Probe(ctx, proberAddressable); status != prober.StatusReady {
+	if status := r.Prober.Probe(ctx, proberAddressable, prober.StatusReady); status != prober.StatusReady {
 		statusConditionManager.ProbesStatusNotReady(status)
 		return nil // Object will get re-queued once probe status changes.
 	}
@@ -298,7 +298,7 @@ func (r *Reconciler) finalizeKind(ctx context.Context, ks *eventing.KafkaSink) e
 			Name:      ks.GetName(),
 		},
 	}
-	if status := r.Prober.Probe(ctx, proberAddressable); status != prober.StatusNotReady {
+	if status := r.Prober.Probe(ctx, proberAddressable, prober.StatusNotReady); status != prober.StatusNotReady {
 		// Return a requeueKeyError that doesn't generate an event and it re-queues the object
 		// for a new reconciliation.
 		return controller.NewRequeueAfter(5 * time.Second)


Test patch:
diff --git a/control-plane/pkg/prober/async_prober_test.go b/control-plane/pkg/prober/async_prober_test.go
index e27b94c56..d68fe4481 100644
--- a/control-plane/pkg/prober/async_prober_test.go
+++ b/control-plane/pkg/prober/async_prober_test.go
@@ -36,6 +36,7 @@ import (
 )
 
 func TestAsyncProber(t *testing.T) {
+	t.Parallel()
 
 	tt := []struct {
 		name                string
@@ -106,12 +107,13 @@ func TestAsyncProber(t *testing.T) {
 	}
 
 	for _, tc := range tt {
+		tc := tc
 		t.Run(tc.name, func(t *testing.T) {
 			ctx, _ := reconcilertesting.SetupFakeContext(t)
-			ctx, cacel := context.WithCancel(ctx)
+			ctx, cancel := context.WithCancel(ctx)
 			defer func() {
 				time.Sleep(time.Second)
-				cacel()
+				cancel()
 			}()
 			wantRequestCountMin := atomic.NewInt64(int64(tc.wantRequestCountMin))
 			h := http.HandlerFunc(func(writer http.ResponseWriter, r *http.Request) {
@@ -121,7 +123,6 @@ func TestAsyncProber(t *testing.T) {
 			})
 			s := httptest.NewUnstartedServer(h)
 			s.Start()
-			time.Sleep(time.Second) // Wait for the server time to start.
 			defer s.Close()
 
 			for _, p := range tc.pods {
@@ -134,14 +135,15 @@ func TestAsyncProber(t *testing.T) {
 			prober := NewAsync(ctx, s.Client(), u.Port(), tc.podsLabelsSelector, func(key types.NamespacedName) {
 				wantRequeueCountMin.Dec()
 			})
+
 			probeFunc := func() bool {
-				status := prober.Probe(ctx, tc.addressable)
+				status := prober.Probe(ctx, tc.addressable, tc.wantStatus)
 				return status == tc.wantStatus
 			}
 
 			require.Eventuallyf(t, probeFunc, time.Second, 10*time.Millisecond, "")
-			require.GreaterOrEqual(t, int64(0), wantRequeueCountMin.Load())
-			require.GreaterOrEqual(t, int64(0), wantRequestCountMin.Load())
+			require.GreaterOrEqual(t, int64(0), wantRequestCountMin.Load(), "")
+			require.GreaterOrEqual(t, int64(0), wantRequeueCountMin.Load(), "")
 		})
 	}
 }
diff --git a/control-plane/pkg/prober/prober_test.go b/control-plane/pkg/prober/prober_test.go
index 850bd1192..bb9bdf30b 100644
--- a/control-plane/pkg/prober/prober_test.go
+++ b/control-plane/pkg/prober/prober_test.go
@@ -28,12 +28,12 @@ func TestFuncProbe(t *testing.T) {
 
 	calls := atomic.NewInt32(0)
 	status := StatusReady
-	p := Func(func(ctx context.Context, addressable Addressable) Status {
+	p := Func(func(ctx context.Context, addressable Addressable, expected Status) Status {
 		calls.Inc()
 		return status
 	})
 
-	s := p.Probe(context.Background(), Addressable{})
+	s := p.Probe(context.Background(), Addressable{}, status)
 
 	require.Equal(t, status, s, s.String())
 	require.Equal(t, int32(1), calls.Load())
diff --git a/control-plane/pkg/prober/probertesting/mock_prober.go b/control-plane/pkg/prober/probertesting/mock_prober.go
index 0111c4850..c13110a7d 100644
--- a/control-plane/pkg/prober/probertesting/mock_prober.go
+++ b/control-plane/pkg/prober/probertesting/mock_prober.go
@@ -24,7 +24,7 @@ import (
 
 // MockProber returns a prober that always returns the provided status.
 func MockProber(status prober.Status) prober.Prober {
-	return prober.Func(func(ctx context.Context, addressable prober.Addressable) prober.Status {
+	return prober.Func(func(ctx context.Context, addressable prober.Addressable, expected prober.Status) prober.Status {
 		return status
 	})
 }
diff --git a/control-plane/pkg/prober/probertesting/mock_prober_test.go b/control-plane/pkg/prober/probertesting/mock_prober_test.go
index 560afebb0..c2c6aba91 100644
--- a/control-plane/pkg/prober/probertesting/mock_prober_test.go
+++ b/control-plane/pkg/prober/probertesting/mock_prober_test.go
@@ -53,7 +53,7 @@ func TestMockProber(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got := MockProber(tt.status).Probe(tt.ctx, tt.addressable)
+			got := MockProber(tt.status).Probe(tt.ctx, tt.addressable, tt.status)
 			if diff := cmp.Diff(tt.status, got); diff != "" {
 				t.Error("(-want, got)", diff)
 			}