Kafka Bus Implementation by ericbottard · Pull Request #99 · knative/eventing

ericbottard · 2018-06-20T11:31:39Z

Fixes Issue #83

~~Similar to #98, this is effectively #88 with one commit on top.~~

This introduces an implementation of Bus backed by Kafka.
The bus itself is split between provisioner (which creates/deletes topics on kafka) and dispatcher (which 1. accepts payloads over http and enqueues them in the topic and 2. consumes from topics and broadcasts to subscribers)

A 1-node Kafka broker (that can sit behind a service) setup is provided in kafka-broker.yaml. It is expected to be applied in namespace kafka (change the broker address in the bus declaration accordingly if it is not)

googlebot · 2018-06-20T11:31:50Z

Thanks for your pull request. It looks like this may be your first contribution to a Google open source project (if not, look below for help). Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

📝 Please visit https://cla.developers.google.com/ to sign.

Once you've signed (or fixed any issues), please reply here (e.g. I signed it!) and we'll verify it.

What to do if you already signed the CLA

Individual signers

It's possible we don't have your GitHub username or you're using a different email address on your commit. Check your existing CLA data and verify that your email is set on your git commits.

Corporate signers

Your company has a Point of Contact who decides which employees are authorized to participate. Ask your POC to be added to the group of authorized contributors. If you don't know who your Point of Contact is, direct the Google project maintainer to go/cla#troubleshoot (Public version).
The email used to register you as an authorized contributor must be the email used for the Git commit. Check your existing CLA data and verify that your email is set on your git commits.
The email used to register you as an authorized contributor must also be attached to your GitHub account.

ericbottard · 2018-06-21T08:40:40Z

/check-cla

googlebot · 2018-06-25T13:36:57Z

CLAs look good, thanks!

ericbottard · 2018-06-26T08:48:48Z

Rebased on latest master that now includes #88

scothis · 2018-06-26T14:22:12Z

+    env:
+    - name: KAFKA_BROKERS
+      value: "kafkabroker.kafka:9092"
+    - name: BUS_NAME


BUS_NAME and BUS_NAMESPACE are set by the controller

scothis · 2018-06-26T14:23:53Z

-			*out = new(InjectDelay)
-			**out = **in
-		}
+		*out = new(InjectDelay)


Why is codegen updating istio types?

I really don't know. Maybe we don't reference a fixed version for Istio (or for codegen)?

scothis · 2018-06-26T14:25:58Z

+
+import (
+	"flag"
+		"github.com/golang/glog"


did gofmt do this?

No, somehow Goland did this. Fixing

scothis · 2018-06-26T14:39:13Z

+
+func main() {
+
+	kubeconfig := flag.String("kubeconfig", "", "Path to a kubeconfig. Only required if out-of-cluster.")


the convention is to define flags in the init func.

That's only relevant if flags are not in the main package (which really shouldn't ever happen if you ask me), in case the package ends up being imported several times.

Putting flag declarations in init really has no benefit whatsoever. Worse, it encourages global vars

scothis · 2018-06-28T19:00:48Z

/retest

evankanderson · 2018-07-02T22:24:54Z

/assign @scothis
/assign @n3wscott

rootfs · 2018-07-05T14:15:51Z

@@ -0,0 +1,305 @@
+/*
+ * Copyright 2018 the original author or authors.


Knative authors, see #115

rootfs · 2018-07-05T14:25:28Z

+	} else {
+		return 0, fmt.Errorf("unsupported initialOffset value. Must be one of %s or %s", Oldest, Newest)
+	}
+	return initialOffset, nil


this is not reachable.

Good catch. This is a leftover from when I thought you could pass an actual int offset here, not just Oldest|Newest

Will fix

ericbottard · 2018-07-09T11:56:28Z

Rebased on latest master

jldec · 2018-07-09T13:57:57Z

@evankanderson do you think we could add this PR to be included in the Knative launch project for demos?

ericbottard · 2018-07-10T10:04:47Z

Rebased on latest master

ericbottard · 2018-07-10T14:19:51Z

@vaikas-google could you please have a look at this PR, having a second non-trivial bus implementation seems valuable for launch

/assign @vaikas-google

vaikas

Couple of small questions, nits. Just curious about the failure model for messages that fail to be delivered.

vaikas · 2018-07-10T15:10:13Z

+	}
+	req, err := http.NewRequest(http.MethodPost, url.String(), bytes.NewReader(body))
+	if err != nil {
+		glog.Errorf("Unable to create subscriber request %v", err)


Are these just dropped on the floor? How do we communicate errors from here and what happens in the error cases?

vaikas · 2018-07-10T15:10:56Z

+}
+
+func channelFromHost(host string) (channel string, namespace string) {
+	chunks := strings.Split(host, ".")


Should we check to make sure the host is valid form? Or check that we actually get two chunks back, otherwise L268 will panic?

Will add code to prevent the panic. Paths that end up returning the zero string for either name of namespace will end up in HTTP 404 thanks to code further down

vaikas · 2018-07-10T15:11:43Z

+				hs := kafka2HttpHeaders(msg)
+				svc := fmt.Sprintf("%s.%s", subscription.Spec.Subscriber, subscription.Namespace)
+				d.dispatchEvent(svc, msg.Value, hs)
+				consumer.MarkOffset(msg, "") // Mark message as processed


Since dispatchEvent does not return error, seems like any delivery failure will result in dropped messages since we mark them as processed?

True, (for now).

Could change signature to propagate errors, but infinite retry is not better. Would require some retry-and-giveup logic that seems out of scope of this initial PR, wdyt?

Will at least mark as TODO

vaikas · 2018-07-10T15:13:00Z

+	conf.ClientID = name + "-dispatcher"
+	kafka_client, err := sarama.NewClient(brokers, conf)
+	if err != nil {
+		glog.Fatalf("Error building kafka client: %s", err.Error())


Just curious why not just log this as "...%v", err)

vaikas · 2018-07-10T15:14:39Z

+subscriber.
+
+To view logs:
+- for the dispatcher `kail -d kafka-bus -c dispatcher`


We should probably add a pointer on how to install kail? Since it's used throughout the examples, wonder if this should be made part of the DEVELOPMENT.md (as optional?).

Adding the same snippet that was just merged in other READMEs

vaikas · 2018-07-10T15:16:05Z

+}
+
+func topicNameFromChannel(channel *channelsv1alpha1.Channel) string {
+	topicName := fmt.Sprintf("%s.%s", channel.Namespace, channel.Name)


maybe just combine these two lines:
return fmt.Sprintf("...

Don't version IDE files Add sarama + sarama-cluster Handle initialOffset

ericbottard · 2018-07-10T15:48:48Z

Addressed review comments, rebased on latest master

ericbottard · 2018-07-10T15:51:28Z

Created #167 to track dispatch error handling

vaikas · 2018-07-10T16:02:48Z

I would prefer changing the signature to return an error, and then ignroing it with a TODO as you changed it.

…

On Tue, Jul 10, 2018 at 8:41 AM Eric Bottard ***@***.***> wrote: ***@***.**** commented on this pull request. ------------------------------ In pkg/buses/kafka/dispatcher/dispatcher.go <#99 (comment)>: > + if err != nil { + return err + } + + d.consumers[subscriptionKeyFor(subscription)] = consumer + + go func() { + for { + msg, more := <-consumer.Messages() + if more { + glog.Infof("Dispatching a message for subscription %s/%s: %s -> %s", subscription.Namespace, + subscription.Name, subscription.Spec.Channel, subscription.Spec.Subscriber) + hs := kafka2HttpHeaders(msg) + svc := fmt.Sprintf("%s.%s", subscription.Spec.Subscriber, subscription.Namespace) + d.dispatchEvent(svc, msg.Value, hs) + consumer.MarkOffset(msg, "") // Mark message as processed True, (for now). Could change signature to propagate errors, but infinite retry is not better. Would require some retry-and-giveup logic that seems out of scope of this initial PR, wdyt? Will at least mark as TODO — You are receiving this because you were assigned. Reply to this email directly, view it on GitHub <#99 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AKwedJkHZ6L_TJyuckurcC7aU8MXve62ks5uFMs0gaJpZM4UvGO1> .

vaikas · 2018-07-10T16:38:18Z

/lgtm
/approve

google-prow-robot · 2018-07-10T16:38:24Z

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: ericbottard, vaikas-google

The full list of commands accepted by this bot can be found here.

The pull request process is described here

Details

Needs approval from an approver in each of these files:

~~OWNERS~~ [vaikas-google]

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

ericbottard · 2018-07-10T17:29:22Z

Awesome, thanks for you thorough review(s) @vaikas-google

@evankanderson

@evankanderson

From [ROLES.MD](https://github.com/knative/docs/blob/dfc53c67c8e80d30b8863353c9e9b4ad00c41fa0/community/ROLES.md#approver): > Reviewer of the codebase for at least 3 months or 50% of project lifetime, whichever is shorter - [First Issue](#80). Opened 6/11 - [First PR](#66). Opened 5/31 - [First Review](#79 (review)) 6/11 > Primary reviewer for at least 10 substantial PRs to the codebase - #422 (review) - #414 (review) - #325 (review) - #225 (review) - #189 (review) - #168 (review) - #165 (review) - #99 (review) - #79 (review) - #111 (review) > Reviewed or merged at least 30 PRs to the codebase - [Reviewed 23 PRs](https://github.com/knative/eventing/pulls?utf8=✓&q=is%3Apr+reviewed-by%3Ascothis) - [Authored 34 merged PRs](https://github.com/knative/eventing/pulls?utf8=✓&q=is%3Apr+author%3Ascothis+is%3Amerged) - [Authored 5 open PRs](https://github.com/knative/eventing/pulls/scothis) > Nominated by an area lead From [WORKING_GROUPS.MD](https://github.com/knative/docs/blob/dfc53c67c8e80d30b8863353c9e9b4ad00c41fa0/community/WORKING-GROUPS.md#events) /assign @vaikas-google > With no objections from other leads 🤞 /cc @evankanderson @grantr @inlined @mattmoor

* Introduce bus backed by Kafka Don't version IDE files Add sarama + sarama-cluster Handle initialOffset * Address review comments * Anticipate ClusterBus * Less invasive update-deps * Rebase on latest master * Update old copyright headers * Remove unreachable code * Adapt to latest bus changes * Rebase against latest master * Use separate directory for kafka bus. Add README. * Address review comments * Add embryo of error handling while dispatching

@evankanderson

latest, greatest filez

google-prow-robot added the size/XXL Denotes a PR that changes 1000+ lines, ignoring generated files. label Jun 20, 2018

ericbottard changed the title ~~Issue 83~~ [DNM] Kafka Bus Implementation Jun 20, 2018

scothis mentioned this pull request Jun 21, 2018

Refine bus monitor #88

Merged

ericbottard force-pushed the issue_83 branch 3 times, most recently from 7c97c14 to 078ba14 Compare June 25, 2018 13:36

ericbottard force-pushed the issue_83 branch 2 times, most recently from 4e7e396 to 848bf9e Compare June 26, 2018 08:48

ericbottard changed the title ~~[DNM] Kafka Bus Implementation~~ Kafka Bus Implementation Jun 26, 2018

scothis reviewed Jun 26, 2018

View reviewed changes

ericbottard force-pushed the issue_83 branch from 9c360ba to 2b3a2fd Compare June 27, 2018 09:34

ericbottard force-pushed the issue_83 branch from 2b3a2fd to 7d3ab73 Compare June 29, 2018 13:57

google-prow-robot assigned n3wscott and scothis Jul 2, 2018

ericbottard force-pushed the issue_83 branch from 7d3ab73 to c3af737 Compare July 5, 2018 12:36

rootfs reviewed Jul 5, 2018

View reviewed changes

ericbottard force-pushed the issue_83 branch from 41023b1 to c7fe8c9 Compare July 9, 2018 11:56

ericbottard force-pushed the issue_83 branch from c7fe8c9 to d8f81ae Compare July 10, 2018 10:03

google-prow-robot assigned vaikas Jul 10, 2018

vaikas reviewed Jul 10, 2018

View reviewed changes

ericbottard and others added 11 commits July 10, 2018 17:26

Introduce bus backed by Kafka

a1987cb

Don't version IDE files Add sarama + sarama-cluster Handle initialOffset

Address review comments

d77f223

Anticipate ClusterBus

f66a0c6

Less invasive update-deps

9f9d823

Rebase on latest master

b766d26

Update old copyright headers

ab326e2

Remove unreachable code

2f1dfcb

Adapt to latest bus changes

812d056

Rebase against latest master

93709f7

Use separate directory for kafka bus. Add README.

acd949e

Address review comments

4e7c4fb

ericbottard force-pushed the issue_83 branch from cfb6f6c to 4e7c4fb Compare July 10, 2018 15:48

Add embryo of error handling while dispatching

81b1ec0

google-prow-robot added the lgtm Indicates that a PR is ready to be merged. label Jul 10, 2018

google-prow-robot added the approved Indicates a PR has been approved by an approver from all required OWNERS files. label Jul 10, 2018

google-prow-robot merged commit 7dc29a5 into knative:master Jul 10, 2018

scothis mentioned this pull request Sep 11, 2018

Add scothis as an approver #426

Merged

matzew pushed a commit to matzew/eventing that referenced this pull request May 16, 2019

Merge pull request knative#99 from matzew/moar_updates

8c7b287

latest, greatest filez


		func main() {

		kubeconfig := flag.String("kubeconfig", "", "Path to a kubeconfig. Only required if out-of-cluster.")

		@@ -0,0 +1,305 @@
		/*
		* Copyright 2018 the original author or authors.

Conversation

ericbottard commented Jun 20, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

googlebot commented Jun 20, 2018

What to do if you already signed the CLA

Individual signers

Corporate signers

Uh oh!

ericbottard commented Jun 21, 2018

Uh oh!

googlebot commented Jun 25, 2018

Uh oh!

ericbottard commented Jun 26, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

scothis commented Jun 28, 2018

Uh oh!

evankanderson commented Jul 2, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ericbottard commented Jul 9, 2018

Uh oh!

jldec commented Jul 9, 2018

Uh oh!

ericbottard commented Jul 10, 2018

Uh oh!

ericbottard commented Jul 10, 2018

Uh oh!

vaikas left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ericbottard commented Jul 10, 2018

Uh oh!

ericbottard commented Jul 10, 2018

Uh oh!

vaikas commented Jul 10, 2018 via email

Uh oh!

vaikas commented Jul 10, 2018

Uh oh!

google-prow-robot commented Jul 10, 2018

Uh oh!

ericbottard commented Jun 20, 2018 •

edited

Loading