Repository: knative-sandbox/eventing-kafka-broker
Pull request #2180 (2022-05-11 09:22:38)

[release-1.2] Prober targets service instead of pods directly
Manual cherry pick of https://github.com/knative-sandbox/eventing-kafka-broker/pull/2159

```release-note
Receiver's prober targets services instead of pods directly to allow components to be part of an Istio mesh
```

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2180#" title="Author self-approved">aliok</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.2/OWNERS)~~ [aliok]

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[bot]: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2180#" title="Author self-approved">aliok</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2180#pullrequestreview-968976540" title="LGTM">pierDipi</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.2/OWNERS)~~ [aliok,pierDipi]

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":[]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2180?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.2@a1d964e`). [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`.

```diff
@@              Coverage Diff               @@
##             release-1.2    #2180   +/-   ##
==============================================
  Coverage               ?   69.37%           
  Complexity             ?      618           
==============================================
  Files                  ?      131           
  Lines                  ?     5518           
  Branches               ?      180           
==============================================
  Hits                   ?     3828           
  Misses                 ?     1383           
  Partials               ?      307           
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `81.98% <0.00%> (?)` | |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#carryforward-flags-in-the-pull-request-comment) to find out more.


------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2180?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/2180?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [a1d964e...1120ae7](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2180?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).


Patch:
diff --git a/control-plane/pkg/prober/async_prober.go b/control-plane/pkg/prober/async_prober.go
index 1d6d2ed6d..c4f211c09 100644
--- a/control-plane/pkg/prober/async_prober.go
+++ b/control-plane/pkg/prober/async_prober.go
@@ -22,57 +22,55 @@ import (
 	"time"
 
 	"go.uber.org/zap"
-	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
-	podinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/pod"
 	"knative.dev/pkg/logging"
 )
 
+type IPsLister func() ([]string, error)
+
 type asyncProber struct {
 	client    httpClient
 	enqueue   EnqueueFunc
 	logger    *zap.Logger
 	cache     Cache
-	podLister func() ([]*corev1.Pod, error)
+	IPsLister IPsLister
 	port      string
 }
 
 // NewAsync creates an async Prober.
 //
 // It reports status changes using the provided EnqueueFunc.
-func NewAsync(ctx context.Context, client httpClient, port string, podsLabelsSelector labels.Selector, enqueue EnqueueFunc) Prober {
+func NewAsync(ctx context.Context, client httpClient, port string, IPsLister IPsLister, enqueue EnqueueFunc) Prober {
 	logger := logging.FromContext(ctx).Desugar().
 		With(zap.String("scope", "prober"))
 
 	if len(port) == 0 {
 		logger.Fatal("Port is required")
 	}
-	podLister := podinformer.Get(ctx).Lister()
 	return &asyncProber{
 		client:    client,
 		enqueue:   enqueue,
 		logger:    logger,
 		cache:     NewLocalExpiringCache(ctx, 30*time.Minute),
-		podLister: func() ([]*corev1.Pod, error) { return podLister.List(podsLabelsSelector) },
+		IPsLister: IPsLister,
 		port:      port,
 	}
 }
 
 func (a *asyncProber) Probe(ctx context.Context, addressable Addressable, expected Status) Status {
 	address := addressable.Address
-	pods, err := a.podLister()
+	IPs, err := a.IPsLister()
 	if err != nil {
-		a.logger.Error("Failed to list pods", zap.Error(err))
+		a.logger.Error("Failed to list IPs", zap.Error(err))
 		return StatusUnknown
 	}
-	// Return `StatusNotReady` when there are no pods.
-	if len(pods) == 0 {
+	// Return `StatusNotReady` when there are no IPs.
+	if len(IPs) == 0 {
 		return StatusNotReady
 	}
 
 	// aggregatedCurrentKnownStatus keeps track of the current status in the cache excluding `StatusReady`
-	// since we just skip pods that have `StatusReady`.
+	// since we just skip IPs that have `StatusReady`.
 	//
 	// If there is a status that is `StatusUnknown` the final status  we want to return is `StatusUnknown`,
 	// while we return `StatusNotReady` when the status is known and all probes returned `StatusNotReady`.
@@ -83,18 +81,18 @@ func (a *asyncProber) Probe(ctx context.Context, addressable Addressable, expect
 	// It goes to done once we have all probe request results regardless of whether they are coming from
 	// the cache or from an actual request.
 	var wg sync.WaitGroup
-	wg.Add(len(pods))
+	wg.Add(len(IPs))
 
 	// enqueueOnce allows requeuing the resource only once, when we have all probe request results.
 	var enqueueOnce sync.Once
 
-	for _, p := range pods {
+	for _, IP := range IPs {
 		podUrl := *address
-		podUrl.Host = p.Status.PodIP + ":" + a.port
+		podUrl.Host = IP + ":" + a.port
 		address := podUrl.String()
 
 		logger := a.logger.
-			With(zap.String("pod.metadata.name", p.Name)).
+			With(zap.String("IP", IP)).
 			With(zap.String("address", address))
 
 		currentStatus := a.cache.GetStatus(address)
diff --git a/control-plane/pkg/prober/prober.go b/control-plane/pkg/prober/prober.go
index 857693d9c..b0cf2f7d8 100644
--- a/control-plane/pkg/prober/prober.go
+++ b/control-plane/pkg/prober/prober.go
@@ -18,6 +18,7 @@ package prober
 
 import (
 	"context"
+	"fmt"
 	"net/http"
 	"net/url"
 
@@ -89,3 +90,9 @@ func probe(ctx context.Context, client httpClient, logger *zap.Logger, address s
 
 	return StatusReady
 }
+
+func IPsListerFromService(svc types.NamespacedName) IPsLister {
+	return func() ([]string, error) {
+		return []string{fmt.Sprintf("%s.%s.svc", svc.Name, svc.Namespace)}, nil
+	}
+}
diff --git a/control-plane/pkg/reconciler/broker/controller.go b/control-plane/pkg/reconciler/broker/controller.go
index d488bee56..2c3dbbb9a 100644
--- a/control-plane/pkg/reconciler/broker/controller.go
+++ b/control-plane/pkg/reconciler/broker/controller.go
@@ -24,6 +24,7 @@ import (
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/cache"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
@@ -88,7 +89,8 @@ func NewController(ctx context.Context, watcher configmap.Watcher, env *config.E
 	})
 
 	reconciler.Resolver = resolver.NewURIResolverFromTracker(ctx, impl.Tracker)
-	reconciler.Prober = prober.NewAsync(ctx, http.DefaultClient, env.IngressPodPort, reconciler.ReceiverSelector(), impl.EnqueueKey)
+	IPsLister := prober.IPsListerFromService(types.NamespacedName{Namespace: env.SystemNamespace, Name: env.IngressName})
+	reconciler.Prober = prober.NewAsync(ctx, http.DefaultClient, env.IngressPodPort, IPsLister, impl.EnqueueKey)
 	reconciler.IngressHost = network.GetServiceHostname(env.IngressName, env.SystemNamespace)
 
 	brokerInformer := brokerinformer.Get(ctx)
diff --git a/control-plane/pkg/reconciler/channel/controller.go b/control-plane/pkg/reconciler/channel/controller.go
index 8ccfa800b..881023a38 100644
--- a/control-plane/pkg/reconciler/channel/controller.go
+++ b/control-plane/pkg/reconciler/channel/controller.go
@@ -23,6 +23,7 @@ import (
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/cache"
 
 	messagingv1beta "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
@@ -83,7 +84,8 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *conf
 	}
 
 	impl := kafkachannelreconciler.NewImpl(ctx, reconciler)
-	reconciler.Prober = prober.NewAsync(ctx, http.DefaultClient, configs.IngressPodPort, reconciler.ReceiverSelector(), impl.EnqueueKey)
+	IPsLister := prober.IPsListerFromService(types.NamespacedName{Namespace: configs.SystemNamespace, Name: configs.IngressName})
+	reconciler.Prober = prober.NewAsync(ctx, http.DefaultClient, configs.IngressPodPort, IPsLister, impl.EnqueueKey)
 	reconciler.IngressHost = network.GetServiceHostname(configs.IngressName, configs.SystemNamespace)
 
 	channelInformer := kafkachannelinformer.Get(ctx)
diff --git a/control-plane/pkg/reconciler/sink/controller.go b/control-plane/pkg/reconciler/sink/controller.go
index 967dcf4b6..b0de02839 100644
--- a/control-plane/pkg/reconciler/sink/controller.go
+++ b/control-plane/pkg/reconciler/sink/controller.go
@@ -23,6 +23,7 @@ import (
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/cache"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	configmapinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap"
@@ -75,7 +76,8 @@ func NewController(ctx context.Context, _ configmap.Watcher, configs *config.Env
 	}
 
 	impl := sinkreconciler.NewImpl(ctx, reconciler)
-	reconciler.Prober = prober.NewAsync(ctx, http.DefaultClient, configs.IngressPodPort, reconciler.ReceiverSelector(), impl.EnqueueKey)
+	IPsLister := prober.IPsListerFromService(types.NamespacedName{Namespace: configs.SystemNamespace, Name: configs.IngressName})
+	reconciler.Prober = prober.NewAsync(ctx, http.DefaultClient, configs.IngressPodPort, IPsLister, impl.EnqueueKey)
 	reconciler.IngressHost = network.GetServiceHostname(configs.IngressName, configs.SystemNamespace)
 
 	sinkInformer := sinkinformer.Get(ctx)
diff --git a/data-plane/config/broker/500-receiver.yaml b/data-plane/config/broker/500-receiver.yaml
index 7d0ff7262..cdc550d49 100644
--- a/data-plane/config/broker/500-receiver.yaml
+++ b/data-plane/config/broker/500-receiver.yaml
@@ -174,6 +174,10 @@ spec:
       port: 80
       protocol: TCP
       targetPort: 8080
+    - name: http-container
+      port: 8080
+      protocol: TCP
+      targetPort: 8080
     - name: http-metrics
       port: 9090
       protocol: TCP
diff --git a/data-plane/config/channel/500-receiver.yaml b/data-plane/config/channel/500-receiver.yaml
index 317f693a9..88b2cf472 100644
--- a/data-plane/config/channel/500-receiver.yaml
+++ b/data-plane/config/channel/500-receiver.yaml
@@ -174,6 +174,10 @@ spec:
       port: 80
       protocol: TCP
       targetPort: 8080
+    - name: http-container
+      port: 8080
+      protocol: TCP
+      targetPort: 8080
     - name: http-metrics
       port: 9090
       protocol: TCP
diff --git a/data-plane/config/sink/500-receiver.yaml b/data-plane/config/sink/500-receiver.yaml
index 248672c44..b2b34762b 100644
--- a/data-plane/config/sink/500-receiver.yaml
+++ b/data-plane/config/sink/500-receiver.yaml
@@ -174,6 +174,10 @@ spec:
       port: 80
       protocol: TCP
       targetPort: 8080
+    - name: http-container
+      port: 8080
+      protocol: TCP
+      targetPort: 8080
     - name: http-metrics
       port: 9090
       protocol: TCP
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java
index ba9caf8e8..280727369 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java
@@ -142,6 +142,7 @@ public final class WebClientCloudEventSender implements CloudEventSender {
     // From https://github.com/knative/specs/blob/c348f501de9eb998b4fd010c54d9127033ee41be/specs/eventing/data-plane.md#event-acknowledgement-and-delivery-retry
     return statusCode >= 500 || // Generic error
       statusCode == 404 || // Endpoint does not exist
+      statusCode == 408 || // Request Timeout
       statusCode == 409 || // Conflict / Processing in progress
       statusCode == 429;   // Too Many Requests / Overloaded
   }


Test patch:
diff --git a/control-plane/pkg/prober/async_prober_test.go b/control-plane/pkg/prober/async_prober_test.go
index d68fe4481..893a3b6c8 100644
--- a/control-plane/pkg/prober/async_prober_test.go
+++ b/control-plane/pkg/prober/async_prober_test.go
@@ -132,7 +132,18 @@ func TestAsyncProber(t *testing.T) {
 			u, _ := url.Parse(s.URL)
 
 			wantRequeueCountMin := atomic.NewInt64(int64(tc.wantRequeueCountMin))
-			prober := NewAsync(ctx, s.Client(), u.Port(), tc.podsLabelsSelector, func(key types.NamespacedName) {
+			var IPsLister IPsLister = func() ([]string, error) {
+				pods, err := podinformer.Get(ctx).Lister().List(tc.podsLabelsSelector)
+				if err != nil {
+					return nil, err
+				}
+				ips := make([]string, 0, len(pods))
+				for _, p := range pods {
+					ips = append(ips, p.Status.PodIP)
+				}
+				return ips, nil
+			}
+			prober := NewAsync(ctx, s.Client(), u.Port(), IPsLister, func(key types.NamespacedName) {
 				wantRequeueCountMin.Dec()
 			})
 
diff --git a/control-plane/pkg/prober/prober_test.go b/control-plane/pkg/prober/prober_test.go
index bb9bdf30b..596c8985d 100644
--- a/control-plane/pkg/prober/prober_test.go
+++ b/control-plane/pkg/prober/prober_test.go
@@ -20,8 +20,10 @@ import (
 	"context"
 	"testing"
 
+	"github.com/google/go-cmp/cmp"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/atomic"
+	"k8s.io/apimachinery/pkg/types"
 )
 
 func TestFuncProbe(t *testing.T) {
@@ -38,3 +40,33 @@ func TestFuncProbe(t *testing.T) {
 	require.Equal(t, status, s, s.String())
 	require.Equal(t, int32(1), calls.Load())
 }
+
+func TestIPsListerFromService(t *testing.T) {
+	tests := []struct {
+		name    string
+		svc     types.NamespacedName
+		want    []string
+		wantErr bool
+	}{
+		{
+			name: "ok",
+			svc: types.NamespacedName{
+				Namespace: "ns",
+				Name:      "name",
+			},
+			want:    []string{"name.ns.svc"},
+			wantErr: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := IPsListerFromService(tt.svc)()
+			if tt.wantErr != (err != nil) {
+				t.Errorf("Got err %v, wantErr %v", err, tt.wantErr)
+			}
+			if diff := cmp.Diff(tt.want, got); diff != "" {
+				t.Error("(-want, +got)", diff)
+			}
+		})
+	}
+}
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSenderTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSenderTest.java
index 72ee6c55b..5042fdac2 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSenderTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSenderTest.java
@@ -76,7 +76,7 @@ public class WebClientCloudEventSenderTest {
     return Stream.concat(
       Stream.concat(
         IntStream.range(200, 404).boxed(),
-        IntStream.range(405, 409).boxed()
+        IntStream.range(405, 408).boxed()
       ),
       Stream.concat(
         IntStream.range(410, 429).boxed(),
@@ -88,7 +88,7 @@ public class WebClientCloudEventSenderTest {
   public static Stream<Integer> retryableStatusCodes() {
     return Stream.concat(
       IntStream.range(500, 600).boxed(),
-      IntStream.of(404, 409, 429).boxed()
+      IntStream.of(404, 408, 409, 429).boxed()
     );
   }
 }