[WIP] Add NATS Streaming as a new bus implementation · Pull Request #491 · knative/eventing

ghost · 2018-10-05T09:42:09Z

Fixes #

Proposed Changes

add NATS Streaming v.0.11.0 as a new bus implementation.
the messages are persisted 24h
the name of this bus to be used by a channel is "natss"

devguyio

@radufa I aded a couple of cosmetic comments. Will test locally and provide any extra feedback.

devguyio · 2018-10-05T09:52:34Z

+  labels:
+    app: nats-streaming
+spec:
+#  type: NodePort


Please remove commented line

devguyio · 2018-10-05T09:52:49Z

+spec:
+#  type: NodePort
+  type: ClusterIP
+#  clusterIP: None


Please remove commented line

devguyio · 2018-10-05T09:56:10Z

+1. Apply the 'natss' Bus `ko apply -f config/buses/natss/`
+1. Create Channels that reference the 'natss' Bus
+
+The NATSS-Streaming bus uses NATS-Streaming based on a simple setup, see [Natss Streaming](./100-natss.yaml) .


NATS Streaming without a hyphen.

devguyio · 2018-10-05T09:59:58Z

+				}
+			}
+			// subscribe to a NATSS subject
+			if natsStreamingSub, err := (*natsConn).Subscribe(channel.Name, mcb /*nmh.msgh*/, stan.DurableName(subscription.Name)); err != nil {


Remove the comment /*nmh.msgh*/

abbi-gaurav · 2018-10-05T11:02:57Z

+apiVersion: v1
+kind: Service
+metadata:
+  name: nats-streaming-headless


Minor naming comment: I suppose headless is used in case clusterIP:None. Please consider removing headless.

vaikas

I'll review rest later, just got through the yaml files.

vaikas · 2018-10-05T12:25:07Z

+    # configuration file used to override default NATS server settings
+  stan.conf: |
+    # content of configuration file used to override default NATS Streaming server settings
+kind: ConfigMap


bit of a nit, could you organize this:
apiversion
kind
metadata
data

makes it easier to parse mentally :)

vaikas · 2018-10-05T12:26:26Z

+          args:
+          - -D
+          - -SD
+          - --cluster_id=knative-nats-streaming


is this something that needs to be possibly updated when deploying?

Yes, this is a kind of a shared "security context" between all NATS Streaming servers running in a cluster and all their clients. Now it's only a simple deployment without clustering.

vaikas · 2018-10-05T12:27:55Z

+        - "ReadWriteOnce"
+      resources:
+        requests:
+          storage: "1Gi"


Is this for the persistence of the messages? If so, is there a way to get a rough idea how this might be tweaked depending on message rates / sizes, etc.?

Yes, this is for persistence, for all messages... We have to improve the strategy for the persistence layer in the future, but it's enough for this simple example.

ghost · 2018-10-05T14:18:29Z

/retest

ghost · 2018-10-05T14:29:17Z

/retest

grantr · 2018-10-05T19:07:29Z

/cc @adamharwayne

adamharwayne · 2018-10-05T19:09:53Z

+  selector:
+    app: nats-streaming
+  sessionAffinity: None
+  selector:


selector is defined twice.

adamharwayne · 2018-10-05T19:11:32Z

+kind: StatefulSet
+metadata:
+  name: nats-streaming
+  namespace: default


Should this be in the knative-eventing namespace?

it is deployed now in knative-eventing namespace

adamharwayne · 2018-10-05T19:14:03Z

+            requests:
+              cpu: "100m"
+            limits:
+              memory: "32M"          


This seems low, is it enough?

It's enough for this simple, not production ready, "bus" implementation. We will come later with a NATS Streaming cluster which will be deeper integrated into Eventing.

adamharwayne · 2018-10-05T19:18:32Z

+			}
+			// subscribe to a NATSS subject
+			if natsStreamingSub, err := (*natsConn).Subscribe(channel.Name, mcb, stan.DurableName(subscription.Name)); err != nil {
+				bus.logger.Errorf(" Create new NATSS Subscription failed: %+v", err)


adamharwayne · 2018-10-05T19:22:01Z

+			// unsubscribe from a NATSS subject
+			if natsStreamingSub, ok := bus.subscribers[subscription.Name]; ok {
+				if err := (*natsStreamingSub).Unsubscribe(); err  != nil {
+					bus.logger.Errorf(" Unsubscribe() failed: %+v", err)


adamharwayne · 2018-10-05T20:01:24Z

+		ProvisionFunc: func(channel buses.ChannelReference, parameters buses.ResolvedParameters) error {
+			bus.logger.Infof("Provision channel %q\n", channel.Name)
+			bus.logger.Info("channel=%+v; parameters=%+v", channel, parameters)
+			// TODO create a NATSS subject


Is there anything to do? Aren't the subjects created with the subscriptions, rather than the bus itself? If so, I think we can simply not assign ProvisionFunc and UnprovisionFunc.

It's only a place holder now, cause the creation of a NATS Streaming subjects will be validated in the future, avoiding automatically (uncontrolled) creation.
For this simple implementation, this code is not necessary now. A subject is created, if necessary, with the subscription.
I'll delete it, if it has an impact of the clarity of this example.

My preference is to either fill out the TODO with what is actually planned or to remove them. But it is weakly held.

adamharwayne · 2018-10-05T20:06:57Z

+}
+
+// IsConnected ....
+func IsConnected(sc *stan.Conn) (ok bool) {


Is this used?

Not used, removed

adamharwayne · 2018-10-05T20:08:10Z

+const BusType = "natss"
+
+type NatssBus struct {
+	natsConn *stan.Conn


I don't think this is used.

adamharwayne · 2018-10-05T20:11:42Z

+			// TODO delete a NATSS subject
+			return nil
+		},
+		SubscribeFunc: func(channel buses.ChannelReference, subscription buses.SubscriptionReference, parameters buses.ResolvedParameters) error {


For ease of reading, can these be moved to functions on NatssBus?

SubscribeFunc: bus.subscribeFunc,
...

func (b *NatssBus) subscribeFunc(...)

adamharwayne · 2018-10-05T20:18:05Z

+			// TODO delete a NATSS subject
+			return nil
+		},
+		SubscribeFunc: func(channel buses.ChannelReference, subscription buses.SubscriptionReference, parameters buses.ResolvedParameters) error {


I think SubscribeFunc and UnsubscribeFunc should be set in the Dispatcher, not the Provisioner.

ghost · 2018-10-08T13:45:18Z

/retest

bbrowning · 2018-10-08T16:15:12Z

The last integration test failures were due to an unrelated issue impacting all PRs that has been fixed. Submitting this one for testing again.

/test pull-knative-eventing-integration-tests

ghost · 2018-10-09T06:53:40Z

@adamharwayne I have done the changes proposed by you. Please have a look.

n3wscott · 2018-10-09T14:20:19Z

Should we add a new bus vs provisioner?

ghost · 2018-10-09T14:31:12Z

@n3wscott Are you referring to the Provisioner part of the bus implementation (see in this case the discussion with @adamharwayne)?

adamharwayne

Overall looks good to me. Please add some unit tests and I will be happy to approve.

adamharwayne · 2018-10-09T21:30:42Z

+			return err
+		}
+		delete(bus.subscribers, subscription.Name)
+		bus.logger.Errorf(" Unsubscribe() successful.")


adamharwayne · 2018-10-09T21:33:37Z

+		}
+	}
+	// subscribe to a NATSS subject
+	aw, _ := time.ParseDuration("60s")


1 * time.Minute

adamharwayne · 2018-10-09T21:36:56Z

+		ProvisionFunc: func(channel buses.ChannelReference, parameters buses.ResolvedParameters) error {
+			bus.logger.Infof("Provision channel %q\n", channel.Name)
+			bus.logger.Info("channel=%+v; parameters=%+v", channel, parameters)
+			// TODO create a NATSS subject


My preference is to either fill out the TODO with what is actually planned or to remove them. But it is weakly held.

adamharwayne · 2018-10-09T21:40:46Z

+ * limitations under the License.
+ */
+
+package stanutil


Please add unit tests for this package.

Done. The unit tests are using NATS Streaming embedded.

adamharwayne · 2018-10-09T21:41:15Z

+
+import "testing"
+
+func TestStub(t *testing.T) {


Please add in some tests.

@adamharwayne the NATS Streaming server can be embedded inside of the test process to run unit tests again a real NATS Streaming server. This implies vendoring the NATS Streaming server code in Eventing.
Of course, mocking NATS Streaming could be another solution.
Cause this is the first bus implementation which should add tests, please advise.

So far we haven't run the real implementations in unit tests, but it seems like a good idea as long as:

It isn't flaky

It doesn't take too long (upper bound of 1 minute to start with?).

I didn't realize that the other buses didn't have tests. If you don't vendor the NATS streaming server, then I would focus on testing the event handler functions (e.g. handles errors gracefully, only ACKs the message if the downstream acked, etc.).

Thanks! I'll use the embedded NATS Streaming for the unit tests. It's stable and the startup time is aprox. 10s.

googlebot · 2018-10-17T07:00:42Z

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.

ghost · 2018-10-17T09:09:11Z

I signed it!

googlebot · 2018-10-17T09:09:16Z

CLAs look good, thanks!

ghost · 2018-10-17T09:17:00Z

/cc @adamharwayne

adamharwayne

Thanks for adding the tests! I suddenly see why none of our existing buses have tests...

adamharwayne · 2018-10-18T19:39:36Z

@@ -0,0 +1,11 @@
+# Stub - Knative Bus


Suggested change

# Stub - Knative Bus

# NATS Streaming - Knative Bus

adamharwayne · 2018-10-18T19:46:28Z

+)
+
+const (
+	clusterId = "knative-eventing"


adamharwayne · 2018-10-18T19:55:24Z

+func TestConnectPublishClose(t *testing.T) {
+	// connect
+	natsConn, err := Connect(clusterId, clientId, natssUrl, logger)
+	assert.Nil(t, err)


I think our standard way to do this is:

if err != nil { t.Error("Connect returned nil") }

Replacing the error message as needed.

Done, also I've removed the "testify" from vendor

adamharwayne · 2018-10-18T19:58:25Z

+
+func TestConnectPublishClose(t *testing.T) {
+	// connect
+	natsConn, err := Connect(clusterId, clientId, natssUrl, logger)


I would expect this to defer closing here (well, just after checking if err is not nil).

defer Close(natsConn, logger)

adamharwayne · 2018-10-18T19:59:49Z

+	if err != nil {
+		panic(err)
+	}
+	retCode := m.Run()


I would expect this to defer stopNatss here, now that we know the err is nil.

defer stopNatss(stanServer) retCode := m.Run()

adamharwayne · 2018-10-18T20:55:38Z

+	busProv.Run(1, stopCh, "test-provisioner-natss")
+	busDisp.Run(1, stopCh, "test-dispatcher-natss")
+
+	// create adummy channel


adummy -> a dummy

adamharwayne · 2018-10-18T20:56:25Z

+	// wait for subscriber to respond
+	select {
+	case <-done: logger.Infof("Subscriber finished")
+	case <-time.After(5 * time.Second): logger.Info("Subscriber timeout")


Shouldn't this fail the test?

adamharwayne · 2018-10-18T20:58:01Z

+func SetNewBusProvisioner(opts *buses.BusOpts) SetupNatssBus {
+	return func(b *NatssBus) error {
+		b.natsUrl = NatssUrl
+		b.provisioner = buses.NewBusProvisioner(b.ref, b.dispEventHandler, opts)


b.dispEventHandler -> b.provEventHandler

adamharwayne · 2018-10-18T21:00:16Z

+	busDisp, err := NewNatssBusDispatcher(ref, setNewTestBusDispatcher(t, opts))
+	assert.Nil(t, err)
+
+	stopCh := signals.SetupSignalHandler()


Should we defer closing/sending to this Channel?

adamharwayne · 2018-10-18T21:09:02Z

+
+	// wait for subscriber to respond
+	select {
+	case <-done: logger.Infof("Subscriber finished")


Check the message payload.

ghost · 2018-10-19T14:26:18Z

/retest

ghost · 2018-10-19T14:36:59Z

/test pull-knative-eventing-unit-tests

ghost · 2018-10-19T14:55:01Z

/test pull-knative-eventing-unit-tests

adamharwayne

/lgtm

adamharwayne · 2018-10-24T17:34:59Z

+)
+
+func TestMain(m *testing.M) {
+	logger = buses.NewBusLoggerFromConfig(buses.NewLoggingConfig())


Does anything rely on the logs being written? E.g. something searches the log output to see if 'foobar' was written. Or does it just need a struct that has the right methods?

zap.NewNop() returns a valid logger, but the logs don't get written anywhere.

adamharwayne · 2018-10-24T17:38:31Z

+)
+
+func TestMain(m *testing.M) {
+	logger = buses.NewBusLoggerFromConfig(buses.NewLoggingConfig())


Same as above. Do we need the actual logs? Or just a logger (that writes to /dev/null)?

adamharwayne · 2018-10-24T18:02:24Z

 	}
 }

+func SetNewBusProvisioner(opts *buses.BusOpts) SetupNatssBus {


Thanks for investigating. I think this is fine as-is for this PR.

I think we need to create a fake Reconciler (not in this PR, just in general to make this stuff testable). Then we could set it in the options and not need this workaround.

opts.Reconciler = FakeReconciler(),

As you call bus.Run(), just creating an empty Reconciler{} doesn't work either.

Random musing: Maybe we just need to make Reconciler an interface and swap out Run() to do nothing (as is done in the test file).

n3wscott

Apologies for the churn, we need to refactor this PR to use ClusterChannelProvisioners.

As it is, I don't think we want to commit any new changes to bus impls, we have changed directions a bit on what it means to back a channel.

n3wscott · 2018-10-24T17:46:34Z

@@ -0,0 +1,101 @@
+apiVersion: v1


It should actually be embedded into the controller for the provisioner for the natss channels. And then published into kubernetes when a channel is attempted to be provisioned with the natss cluster channel provisioner

n3wscott · 2018-10-24T17:49:24Z

@@ -0,0 +1,9 @@
+# NATS Streaming - Knative Bus


The current model does not have busses and sorry to say that we will have to pivot this implementation to conform to the ideas of ClusterChannelProvisioners

n3wscott · 2018-10-24T17:51:08Z

+		ReceiveMessageFunc: func(channel buses.ChannelReference, message *buses.Message) error {
+			bus.logger.Infof("Recieved message from %q channel", channel.String())
+			if err := stanutil.Publish(natsConn, channel.Name, &message.Payload, bus.logger); err != nil {
+				bus.logger.Errorf("Error during publish: %+v", err)


Zap wants you to wrap err in zap.Error(err) for pretty print.

n3wscott · 2018-10-24T17:51:56Z

+				bus.logger.Errorf("Error during publish: %+v", err)
+				return err
+			}
+			bus.logger.Infof("Published [%s] : '%s'", channel.String(), message)


We should not log the contents of the message. There is a huge risk to leak PII here.

n3wscott · 2018-10-24T17:54:06Z

+	var err error
+	for i := 0; i < 60; i++ {
+		if natsConn, err = stanutil.Connect("knative-nats-streaming", clientId, b.natssUrl, b.logger); err != nil {
+			b.logger.Errorf(" Create new connection failed: %+v", err)


zap.Error(err) for pretty print.

n3wscott · 2018-10-24T18:13:29Z

+}
+
+// Publish a message to a subject
+func Publish(sc *stan.Conn, subj string, msg *[]byte, logger *zap.SugaredLogger) (err error) {


in general, loggers should not be passed around, only context and then pull the logger from the context. There are examples in the repo to do this with the knative/pkg/logging helpers.

n3wscott · 2018-10-24T18:13:39Z

+	defer func() {
+		if r := recover(); r != nil {
+			err = fmt.Errorf("recovered from: %v", r)
+			logger.Errorf("Publish(): %v", err.Error())


zap.Error(err) for pretty print.

n3wscott · 2018-10-24T18:34:35Z

+	}
+	err = (*sc).Publish(subj, *msg)
+	if err != nil {
+		logger.Errorf("Error during publish: %v\n", err)


zap.Error(err) for pretty print.

n3wscott · 2018-10-24T18:35:04Z

+	}()
+
+	if sc == nil {
+		err = errors.New("cant'publish on empty stan connection")


typo: -> "can't publish on empty stan connection"

likely it should be "can not publish on empty stan connection"

n3wscott · 2018-10-24T18:35:52Z

@@ -0,0 +1,73 @@
+package stanutil


needs knative header

ghost · 2018-10-29T07:59:22Z

Thanks a lot for reviews!
We will change the implementation to be aligned with the new ClusterChannelProvisioners.

- move NATSS in knative-eventing namespace - use zap logger in instanutil - use manual Ack for NATSS messages

- remove "testify" from vendor - correct log messages - update Readme.md

knative-metrics-robot · 2018-10-31T10:10:39Z

The following is the coverage report on pkg/.
Say /test pull-knative-eventing-go-coverage to re-run this coverage report

File	Old Coverage	New Coverage
pkg/buses/natss/bus.go	Do not exist	61.2%
pkg/buses/natss/dispatcher/main.go	Do not exist	0.0%
pkg/buses/natss/provisioner/main.go	Do not exist	0.0%
pkg/buses/natss/stanutil/stanutil.go	Do not exist	57.7%

knative-prow-robot · 2018-10-31T10:11:26Z

@radufa: The following test failed, say /retest to rerun them all:

Test name	Commit	Details	Rerun command
pull-knative-eventing-build-tests	`8a77c45`	link	`/test pull-knative-eventing-build-tests`

Full PR test history. Your PR dashboard.

Details

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes/test-infra repository. I understand the commands that are listed here.

matzew · 2018-11-19T14:18:51Z

@radufa I guess this can be closed - due to #624 ?

ghost · 2018-11-19T14:29:45Z

Close this PR cause it is replaced by the new one: "Add NATS Streaming provisioner", see: #624

…knative#7250) (knative#491) * Fixed eventtype create-delete loop on built in sources * Fix unit tests * Fixed source finalizers * Fixed tests --------- Signed-off-by: Calum Murray <cmurray@redhat.com> Co-authored-by: Calum Murray <cmurray@redhat.com>

knative-prow-robot requested review from inlined and vaikas October 5, 2018 09:42

knative-prow-robot added the size/XL Denotes a PR that changes 500-999 lines, ignoring generated files. label Oct 5, 2018

devguyio reviewed Oct 5, 2018

View reviewed changes

abbi-gaurav suggested changes Oct 5, 2018

View reviewed changes

vaikas reviewed Oct 5, 2018

View reviewed changes

knative-prow-robot added size/XXL Denotes a PR that changes 1000+ lines, ignoring generated files. and removed size/XL Denotes a PR that changes 500-999 lines, ignoring generated files. labels Oct 5, 2018

knative-prow-robot requested a review from adamharwayne October 5, 2018 19:07

adamharwayne reviewed Oct 5, 2018

View reviewed changes

ghost mentioned this pull request Oct 9, 2018

Create a proposal for Knative eventing bus implementation which can be used for Kyma business features kyma-project/kyma#402

Closed

adamharwayne reviewed Oct 9, 2018

View reviewed changes

knative-prow-robot requested a review from adamharwayne October 17, 2018 09:17

adamharwayne reviewed Oct 18, 2018

View reviewed changes

googlebot added the cla: yes Indicates the PR's author has signed the CLA. label Oct 19, 2018

adamharwayne reviewed Oct 24, 2018

View reviewed changes

n3wscott reviewed Oct 24, 2018

View reviewed changes

n3wscott mentioned this pull request Oct 24, 2018

adding n3wscott to approvers. #465

Merged

Radu Fantaziu added 16 commits October 31, 2018 10:36

Add NATS Streaming as a new bus implementation

7452871

Add vendoring for nats-io

d816c69

Fix comments regarding reordering of entities

1d03981

Add more details to README.md

02068d9

Add "gogoproto" to vendor

17eb19d

Use [override] instead of [constraint] in toml file

1fb8398

Fix missed gogo package in vendor

6ff96de

Resolve reviews

d9f067d

- move NATSS in knative-eventing namespace - use zap logger in instanutil - use manual Ack for NATSS messages

Remove natsConn from NatsBus struct

83a46bc

Improve code style and change log level

6f668a2

Add unit tests based on the embedded NATS Streaming

418c398

Add NATS Streaming server to vendor allowing to embed it for unit tests

4fe954c

Update unit tests and format code

3d0dd59

- remove "testify" from vendor - correct log messages - update Readme.md

Update the main tests

f85e7ca

Retry if embedded NATSS failed to start

1a8185c

Update Gopkg.lock

8a77c45

ghost changed the title ~~Add NATS Streaming as a new bus implementation~~ [WIP] Add NATS Streaming as a new bus implementation Nov 16, 2018

knative-prow-robot added the do-not-merge/work-in-progress Indicates that a PR should not merge because it is a work in progress. label Nov 16, 2018

ghost closed this Nov 19, 2018

Harwayne mentioned this pull request Feb 7, 2019

Add Harwayne as an approver. #782

Merged

devguyio mentioned this pull request Jan 6, 2021

Add devguyio to approvers #4700

Closed

This pull request was closed.

Conversation

ghost commented Oct 5, 2018

Proposed Changes

Uh oh!

devguyio 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!

Choose a reason for hiding this comment

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!

ghost Oct 5, 2018 • edited by ghost Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ghost commented Oct 5, 2018

Uh oh!

ghost commented Oct 5, 2018

Uh oh!

grantr commented Oct 5, 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!

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!

ghost Oct 5, 2018 •

edited by ghost

Loading