Repository: knative-sandbox/eventing-kafka
Pull request #451 (2021-03-10 11:25:04)

Separate consolidated dispatcher subscription endpoint
Fixes #

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Separate consolidated dispatcher subscription endpoint into its own type
-
-

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- 🎁 Add new feature
- 🐛 Fix bug
- 🧽 Update or clean up current behavior
- 🗑️ Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
🗒️ If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note

```

**Docs**

<!--
📖 If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


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

This pull-request has been approved by:
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **aliok** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @aliok` 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).

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

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka/blob/main/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":["aliok"]} -->
knative-prow-robot: @aliok: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | 5ed391faf8974ab27be069fe989806a85b1cc575 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/451/pull-knative-sandbox-eventing-kafka-unit-tests/1369610262432190464) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/451). [Your PR dashboard](https://gubernator.knative.dev/pr/aliok).

<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: @aliok: PR needs rebase.

<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.
</details>
knative-prow-robot: @aliok: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | 8f4ebaa84c1138142e7613019b4d7327c0ce9941 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/451/pull-knative-sandbox-eventing-kafka-unit-tests/1369917077367820288) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/451). [Your PR dashboard](https://gubernator.knative.dev/pr/aliok).

<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 -->
aliok: /assign @matzew 
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/451/pull-knative-sandbox-eventing-kafka-go-coverage/1369922136352755712/artifacts/line-cov.html#file0) | 71.3% | 67.2% | -4.1
[pkg/channel/consolidated/dispatcher/subscription_endpoint.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/451/pull-knative-sandbox-eventing-kafka-go-coverage/1369922136352755712/artifacts/line-cov.html#file1) | Do not exist | 85.2% | 

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/451?src=pr&el=h1) Report
> Merging [#451](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/451?src=pr&el=desc) (3e15c7d) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/5d8683b226e0eb26037e957bf5790cfbed798546?el=desc) (5d8683b) will **increase** coverage by `0.07%`.
> The diff coverage is `68.75%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/451/graphs/tree.svg?width=650&height=150&src=pr&token=W765CT93QC)](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/451?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##             main     #451      +/-   ##
==========================================
+ Coverage   73.80%   73.88%   +0.07%     
==========================================
  Files         138      139       +1     
  Lines        5315     5319       +4     
==========================================
+ Hits         3923     3930       +7     
+ Misses       1147     1144       -3     
  Partials      245      245              
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/451?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [pkg/channel/consolidated/dispatcher/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/451/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL2Rpc3BhdGNoZXIvZGlzcGF0Y2hlci5nbw==) | `58.45% <0.00%> (-4.81%)` | :arrow_down: |
| [...l/consolidated/dispatcher/subscription\_endpoint.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/451/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL2Rpc3BhdGNoZXIvc3Vic2NyaXB0aW9uX2VuZHBvaW50Lmdv) | `78.57% <78.57%> (ø)` | |
| [...el/distributed/dispatcher/dispatcher/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/451/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvZGlzcGF0Y2hlci9kaXNwYXRjaGVyL2Rpc3BhdGNoZXIuZ28=) | `76.80% <0.00%> (+5.59%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/451?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/451?src=pr&el=footer). Last update [5d8683b...3e15c7d](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/451?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

lberk: /hold
/lgtm
as we chatted, holding to make sure #433 lands smoothly before adding this.
knative-prow-robot: @aliok: PR needs rebase.

<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.
</details>
knative-prow-robot: New changes are detected. LGTM label has been removed.
knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by:
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please ask for approval from **aliok** after the PR has been reviewed.

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

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

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka/blob/main/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":["aliok"]} -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/451/pull-knative-sandbox-eventing-kafka-go-coverage/1373909306465849344/artifacts/line-cov.html#file0) | 71.9% | 67.7% | -4.2
[pkg/channel/consolidated/dispatcher/subscription_endpoint.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/451/pull-knative-sandbox-eventing-kafka-go-coverage/1373909306465849344/artifacts/line-cov.html#file1) | Do not exist | 85.7% | 

aliok: @slinkydeveloper this is rebased.

I guess it is fine if don't ask for others' review since this is only a refactoring w/o any business logic change.
knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/451#pullrequestreview-617277481" title="Approved">slinkydeveloper</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).

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/blob/main/OWNERS)~~ [slinkydeveloper]

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":[]} -->
aliok: /unhold

Patch:
diff --git a/pkg/channel/consolidated/dispatcher/dispatcher.go b/pkg/channel/consolidated/dispatcher/dispatcher.go
index 2a83999c1..a3927949e 100644
--- a/pkg/channel/consolidated/dispatcher/dispatcher.go
+++ b/pkg/channel/consolidated/dispatcher/dispatcher.go
@@ -17,11 +17,9 @@ package dispatcher
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
 	"fmt"
 	nethttp "net/http"
-	"strings"
 	"sync"
 	"sync/atomic"
 
@@ -108,8 +106,13 @@ func NewDispatcher(ctx context.Context, args *KafkaDispatcherArgs) (*KafkaDispat
 		topicFunc:            args.TopicFunc,
 	}
 
+	// initialize and start the subscription endpoint server
+	subscriptionEndpoint := &subscriptionEndpoint{
+		dispatcher: dispatcher,
+		logger:     args.Logger,
+	}
 	go func() {
-		dispatcher.logger.Fatal(nethttp.ListenAndServe(":8081", dispatcher))
+		subscriptionEndpoint.start()
 	}()
 
 	podName, err := env.GetRequiredConfigValue(args.Logger.Desugar(), env.PodNameEnvVarKey)
@@ -166,45 +169,6 @@ func (d *KafkaDispatcher) Start(ctx context.Context) error {
 	return d.receiver.Start(ctx)
 }
 
-func (d *KafkaDispatcher) ServeHTTP(w nethttp.ResponseWriter, r *nethttp.Request) {
-	if r.Method != nethttp.MethodGet {
-		w.WriteHeader(nethttp.StatusMethodNotAllowed)
-		d.logger.Errorf("Received request method that wasn't GET: %s", r.Method)
-		return
-	}
-	uriSplit := strings.Split(r.RequestURI, "/")
-	if len(uriSplit) != 3 {
-		w.WriteHeader(nethttp.StatusNotFound)
-		d.logger.Errorf("Unable to process request: %s", r.RequestURI)
-		return
-	}
-	channelRefNamespace, channelRefName := uriSplit[1], uriSplit[2]
-	channelRef := eventingchannels.ChannelReference{
-		Name:      channelRefName,
-		Namespace: channelRefNamespace,
-	}
-	if _, ok := d.channelSubscriptions[channelRef]; !ok {
-		w.WriteHeader(nethttp.StatusNotFound)
-		return
-	}
-	d.channelSubscriptions[channelRef].readySubscriptionsLock.RLock()
-	defer d.channelSubscriptions[channelRef].readySubscriptionsLock.RUnlock()
-	var subscriptions = make(map[string][]int32)
-	w.Header().Set(dispatcherReadySubHeader, channelRefName)
-	for s, ps := range d.channelSubscriptions[channelRef].channelReadySubscriptions {
-		subscriptions[s] = ps.List()
-	}
-	jsonResult, err := json.Marshal(subscriptions)
-	if err != nil {
-		d.logger.Errorf("Error marshalling json for sub-status channelref: %s/%s, %w", channelRefNamespace, channelRefName, err)
-		return
-	}
-	_, err = w.Write(jsonResult)
-	if err != nil {
-		d.logger.Errorf("Error writing jsonResult to serveHTTP writer: %w", err)
-	}
-}
-
 // UpdateKafkaConsumers will be called by new CRD based kafka channel dispatcher controller.
 func (d *KafkaDispatcher) UpdateKafkaConsumers(config *Config) (map[types.UID]error, error) {
 	if config == nil {
diff --git a/pkg/channel/consolidated/dispatcher/subscription_endpoint.go b/pkg/channel/consolidated/dispatcher/subscription_endpoint.go
new file mode 100644
index 000000000..3612e4464
--- /dev/null
+++ b/pkg/channel/consolidated/dispatcher/subscription_endpoint.go
@@ -0,0 +1,60 @@
+package dispatcher
+
+import (
+	"encoding/json"
+	nethttp "net/http"
+	"strings"
+
+	"go.uber.org/zap"
+	eventingchannels "knative.dev/eventing/pkg/channel"
+)
+
+// subscriptionEndpoint is serving the subscription status of the Kafka channel.
+// A prober in the controller calls the endpoint to see if the subscriper is ready.
+type subscriptionEndpoint struct {
+	dispatcher *KafkaDispatcher
+	logger     *zap.SugaredLogger
+}
+
+func (d *subscriptionEndpoint) ServeHTTP(w nethttp.ResponseWriter, r *nethttp.Request) {
+	if r.Method != nethttp.MethodGet {
+		w.WriteHeader(nethttp.StatusMethodNotAllowed)
+		d.logger.Errorf("Received request method that wasn't GET: %s", r.Method)
+		return
+	}
+	uriSplit := strings.Split(r.RequestURI, "/")
+	if len(uriSplit) != 3 {
+		w.WriteHeader(nethttp.StatusNotFound)
+		d.logger.Errorf("Unable to process request: %s", r.RequestURI)
+		return
+	}
+	channelRefNamespace, channelRefName := uriSplit[1], uriSplit[2]
+	channelRef := eventingchannels.ChannelReference{
+		Name:      channelRefName,
+		Namespace: channelRefNamespace,
+	}
+	if _, ok := d.dispatcher.channelSubscriptions[channelRef]; !ok {
+		w.WriteHeader(nethttp.StatusNotFound)
+		return
+	}
+	d.dispatcher.channelSubscriptions[channelRef].readySubscriptionsLock.RLock()
+	defer d.dispatcher.channelSubscriptions[channelRef].readySubscriptionsLock.RUnlock()
+	var subscriptions = make(map[string][]int32)
+	w.Header().Set(dispatcherReadySubHeader, channelRefName)
+	for s, ps := range d.dispatcher.channelSubscriptions[channelRef].channelReadySubscriptions {
+		subscriptions[s] = ps.List()
+	}
+	jsonResult, err := json.Marshal(subscriptions)
+	if err != nil {
+		d.logger.Errorf("Error marshalling json for sub-status channelref: %s/%s, %w", channelRefNamespace, channelRefName, err)
+		return
+	}
+	_, err = w.Write(jsonResult)
+	if err != nil {
+		d.logger.Errorf("Error writing jsonResult to serveHTTP writer: %w", err)
+	}
+}
+
+func (d *subscriptionEndpoint) start() {
+	d.logger.Fatal(nethttp.ListenAndServe(":8081", d))
+}


Test patch:
diff --git a/pkg/channel/consolidated/dispatcher/dispatcher_test.go b/pkg/channel/consolidated/dispatcher/dispatcher_test.go
index f968bba86..39ce1a12f 100644
--- a/pkg/channel/consolidated/dispatcher/dispatcher_test.go
+++ b/pkg/channel/consolidated/dispatcher/dispatcher_test.go
@@ -19,10 +19,6 @@ package dispatcher
 import (
 	"context"
 	"errors"
-	"fmt"
-	"io/ioutil"
-	"net/http"
-	"net/http/httptest"
 	"net/url"
 	"testing"
 
@@ -569,141 +565,6 @@ func TestSetReady(t *testing.T) {
 	}
 }
 
-func TestServeHTTP(t *testing.T) {
-
-	httpGet := "GET"
-	httpPost := "POST"
-	testCases := []struct {
-		name               string
-		responseReturnCode int
-		desiredJson        []byte
-		channelSubs        map[eventingchannels.ChannelReference]*KafkaSubscription
-		requestURI         string
-		httpMethod         string
-	}{
-		{
-			name:               "channelref not found",
-			httpMethod:         httpGet,
-			responseReturnCode: http.StatusNotFound,
-			desiredJson:        []byte{},
-			requestURI:         "/exist/thisDoesNot",
-		}, {
-			name:               "nop",
-			httpMethod:         httpGet,
-			responseReturnCode: http.StatusNotFound,
-			desiredJson:        []byte{},
-			requestURI:         "///",
-		}, {
-			name:               "no ready subscribers",
-			httpMethod:         httpGet,
-			responseReturnCode: http.StatusOK,
-			desiredJson:        []byte(`{}`),
-			channelSubs: map[eventingchannels.ChannelReference]*KafkaSubscription{
-				{Name: "foo", Namespace: "bar"}: {
-					subs:                      []types.UID{},
-					channelReadySubscriptions: map[string]sets.Int32{},
-				},
-			},
-			requestURI: "/bar/foo",
-		}, {
-			name:               "different channelref called from populated channref (different ns)",
-			httpMethod:         httpGet,
-			desiredJson:        []byte{},
-			responseReturnCode: http.StatusNotFound,
-			channelSubs: map[eventingchannels.ChannelReference]*KafkaSubscription{
-				{Name: "foo", Namespace: "baz"}: {
-					subs: []types.UID{"a", "b"},
-					channelReadySubscriptions: map[string]sets.Int32{
-						"a": sets.NewInt32(0),
-						"b": sets.NewInt32(0),
-					},
-				},
-			},
-			requestURI: "/bar/foo",
-		}, {
-			name:               "return correct subscription",
-			httpMethod:         httpGet,
-			desiredJson:        []byte(`{"a":[0],"b":[0,2,5]}`),
-			responseReturnCode: http.StatusOK,
-			channelSubs: map[eventingchannels.ChannelReference]*KafkaSubscription{
-				{Name: "foo", Namespace: "bar"}: {
-					subs: []types.UID{"a", "b"},
-					channelReadySubscriptions: map[string]sets.Int32{
-						"a": sets.NewInt32(0),
-						"b": sets.NewInt32(0, 2, 5),
-					},
-				},
-			},
-			requestURI: "/bar/foo",
-		}, {
-			name:               "return correct subscription from multiple chanrefs",
-			httpMethod:         httpGet,
-			desiredJson:        []byte(`{"a":[0],"b":[0,2,5]}`),
-			responseReturnCode: http.StatusOK,
-			channelSubs: map[eventingchannels.ChannelReference]*KafkaSubscription{
-				{Name: "table", Namespace: "flip"}: {
-					subs: []types.UID{"c", "d"},
-					channelReadySubscriptions: map[string]sets.Int32{
-						"c": sets.NewInt32(0),
-						"d": sets.NewInt32(0),
-					}},
-				{Name: "foo", Namespace: "bar"}: {
-					subs: []types.UID{"a", "b"},
-					channelReadySubscriptions: map[string]sets.Int32{
-						"a": sets.NewInt32(0),
-						"b": sets.NewInt32(0, 2, 5),
-					},
-				},
-			},
-			requestURI: "/bar/foo",
-		}, {
-			name:               "bad request uri",
-			httpMethod:         httpGet,
-			desiredJson:        []byte{},
-			responseReturnCode: http.StatusNotFound,
-			requestURI:         "/here/be/dragons/there/are/too/many/slashes",
-		}, {
-			name:               "bad request method (POST)",
-			httpMethod:         httpPost,
-			desiredJson:        []byte{},
-			responseReturnCode: http.StatusMethodNotAllowed,
-		},
-	}
-	d := &KafkaDispatcher{
-		channelSubscriptions: make(map[eventingchannels.ChannelReference]*KafkaSubscription),
-		logger:               klogtesting.TestLogger(t),
-	}
-	ts := httptest.NewServer(d)
-	defer ts.Close()
-	for _, tc := range testCases {
-		t.Run(tc.name, func(t *testing.T) {
-			t.Logf("Running %s", t.Name())
-			d.channelSubscriptions = tc.channelSubs
-
-			request, _ := http.NewRequest(tc.httpMethod, fmt.Sprintf("%s%s", ts.URL, tc.requestURI), nil)
-			//			resp, err := http.Get(fmt.Sprintf("%s%s", ts.URL, tc.requestURI))
-			resp, err := http.DefaultClient.Do(request)
-			if err != nil {
-				t.Errorf("Could not send request to subscriber endpoint: %v", err)
-			}
-			if resp.StatusCode != tc.responseReturnCode {
-				t.Errorf("unepxected status returned: want: %d, got: %d", tc.responseReturnCode, resp.StatusCode)
-			}
-			respBody, err := ioutil.ReadAll(resp.Body)
-			defer resp.Body.Close()
-			if err != nil {
-				t.Errorf("Could not read response from subscriber endpoint: %v", err)
-			}
-			if testing.Verbose() && len(respBody) > 0 {
-				t.Logf("http response: %s\n", string(respBody))
-			}
-			if diff := cmp.Diff(tc.desiredJson, respBody); diff != "" {
-				t.Errorf("unexpected readysubscriber status response: (-want, +got) = %v", diff)
-			}
-		})
-	}
-}
-
 var sortStrings = cmpopts.SortSlices(func(x, y string) bool {
 	return x < y
 })
diff --git a/pkg/channel/consolidated/dispatcher/subscription_endpoint_test.go b/pkg/channel/consolidated/dispatcher/subscription_endpoint_test.go
new file mode 100644
index 000000000..219cf1dc7
--- /dev/null
+++ b/pkg/channel/consolidated/dispatcher/subscription_endpoint_test.go
@@ -0,0 +1,156 @@
+package dispatcher
+
+import (
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
+	eventingchannels "knative.dev/eventing/pkg/channel"
+	klogtesting "knative.dev/pkg/logging/testing"
+)
+
+func TestServeHTTP(t *testing.T) {
+
+	httpGet := "GET"
+	httpPost := "POST"
+	testCases := []struct {
+		name               string
+		responseReturnCode int
+		desiredJson        []byte
+		channelSubs        map[eventingchannels.ChannelReference]*KafkaSubscription
+		requestURI         string
+		httpMethod         string
+	}{
+		{
+			name:               "channelref not found",
+			httpMethod:         httpGet,
+			responseReturnCode: http.StatusNotFound,
+			desiredJson:        []byte{},
+			requestURI:         "/exist/thisDoesNot",
+		}, {
+			name:               "nop",
+			httpMethod:         httpGet,
+			responseReturnCode: http.StatusNotFound,
+			desiredJson:        []byte{},
+			requestURI:         "///",
+		}, {
+			name:               "no ready subscribers",
+			httpMethod:         httpGet,
+			responseReturnCode: http.StatusOK,
+			desiredJson:        []byte(`{}`),
+			channelSubs: map[eventingchannels.ChannelReference]*KafkaSubscription{
+				{Name: "foo", Namespace: "bar"}: {
+					subs:                      []types.UID{},
+					channelReadySubscriptions: map[string]sets.Int32{},
+				},
+			},
+			requestURI: "/bar/foo",
+		}, {
+			name:               "different channelref called from populated channref (different ns)",
+			httpMethod:         httpGet,
+			desiredJson:        []byte{},
+			responseReturnCode: http.StatusNotFound,
+			channelSubs: map[eventingchannels.ChannelReference]*KafkaSubscription{
+				{Name: "foo", Namespace: "baz"}: {
+					subs: []types.UID{"a", "b"},
+					channelReadySubscriptions: map[string]sets.Int32{
+						"a": sets.NewInt32(0),
+						"b": sets.NewInt32(0),
+					},
+				},
+			},
+			requestURI: "/bar/foo",
+		}, {
+			name:               "return correct subscription",
+			httpMethod:         httpGet,
+			desiredJson:        []byte(`{"a":[0],"b":[0,2,5]}`),
+			responseReturnCode: http.StatusOK,
+			channelSubs: map[eventingchannels.ChannelReference]*KafkaSubscription{
+				{Name: "foo", Namespace: "bar"}: {
+					subs: []types.UID{"a", "b"},
+					channelReadySubscriptions: map[string]sets.Int32{
+						"a": sets.NewInt32(0),
+						"b": sets.NewInt32(0, 2, 5),
+					},
+				},
+			},
+			requestURI: "/bar/foo",
+		}, {
+			name:               "return correct subscription from multiple chanrefs",
+			httpMethod:         httpGet,
+			desiredJson:        []byte(`{"a":[0],"b":[0,2,5]}`),
+			responseReturnCode: http.StatusOK,
+			channelSubs: map[eventingchannels.ChannelReference]*KafkaSubscription{
+				{Name: "table", Namespace: "flip"}: {
+					subs: []types.UID{"c", "d"},
+					channelReadySubscriptions: map[string]sets.Int32{
+						"c": sets.NewInt32(0),
+						"d": sets.NewInt32(0),
+					}},
+				{Name: "foo", Namespace: "bar"}: {
+					subs: []types.UID{"a", "b"},
+					channelReadySubscriptions: map[string]sets.Int32{
+						"a": sets.NewInt32(0),
+						"b": sets.NewInt32(0, 2, 5),
+					},
+				},
+			},
+			requestURI: "/bar/foo",
+		}, {
+			name:               "bad request uri",
+			httpMethod:         httpGet,
+			desiredJson:        []byte{},
+			responseReturnCode: http.StatusNotFound,
+			requestURI:         "/here/be/dragons/there/are/too/many/slashes",
+		}, {
+			name:               "bad request method (POST)",
+			httpMethod:         httpPost,
+			desiredJson:        []byte{},
+			responseReturnCode: http.StatusMethodNotAllowed,
+		},
+	}
+	logger := klogtesting.TestLogger(t)
+	d := &KafkaDispatcher{
+		channelSubscriptions: make(map[eventingchannels.ChannelReference]*KafkaSubscription),
+		logger:               logger,
+	}
+	subscriptionEndpoint := &subscriptionEndpoint{
+		dispatcher: d,
+		logger:     logger,
+	}
+
+	ts := httptest.NewServer(subscriptionEndpoint)
+	defer ts.Close()
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Logf("Running %s", t.Name())
+			d.channelSubscriptions = tc.channelSubs
+
+			request, _ := http.NewRequest(tc.httpMethod, fmt.Sprintf("%s%s", ts.URL, tc.requestURI), nil)
+			//			resp, err := http.Get(fmt.Sprintf("%s%s", ts.URL, tc.requestURI))
+			resp, err := http.DefaultClient.Do(request)
+			if err != nil {
+				t.Errorf("Could not send request to subscriber endpoint: %v", err)
+			}
+			if resp.StatusCode != tc.responseReturnCode {
+				t.Errorf("unepxected status returned: want: %d, got: %d", tc.responseReturnCode, resp.StatusCode)
+			}
+			respBody, err := ioutil.ReadAll(resp.Body)
+			defer resp.Body.Close()
+			if err != nil {
+				t.Errorf("Could not read response from subscriber endpoint: %v", err)
+			}
+			if testing.Verbose() && len(respBody) > 0 {
+				t.Logf("http response: %s\n", string(respBody))
+			}
+			if diff := cmp.Diff(tc.desiredJson, respBody); diff != "" {
+				t.Errorf("unexpected readysubscriber status response: (-want, +got) = %v", diff)
+			}
+		})
+	}
+}