Restart dispatcher on connection loss #796

sbezverk · 2019-02-09T00:35:26Z

Signed-off-by: Serguei Bezverkhi sbezverk@cisco.com

NATSS channels now recover from connection loss.

sbezverk · 2019-02-09T00:35:37Z

/assign @Harwayne

Harwayne · 2019-02-09T00:45:05Z

contrib/natss/pkg/dispatcher/dispatcher/dispatcher.go

@@ -69,6 +69,9 @@ func createReceiverFunction(s *SubscriptionsSupervisor, logger *zap.SugaredLogge
 		ch := getSubject(channel)
 		if err := stanutil.Publish(s.natssConn, ch, &m.Payload, logger); err != nil {
 			logger.Errorf("Error during publish: %v", err)
+			if err.Error() == stan.ErrConnectionClosed.Error() {
+				logger.Fatalf("Connection to NATS was lost, dispatcher is exiting.")


Add a TODO that states we should reconnect rather than restarting, but for now restarting is better than staying in a broken state forever.

Sounds good. Will do.

Harwayne · 2019-02-09T00:46:30Z

Looks good to me. It's not in anyway ideal, because requests from upstream will get rejected as the Pod restarts, but is distinctly better than now, where once the connection closes, the dispatcher will not work again.

I would say that this shouldn't close the bug, but should reference it. Then we still have the bug if anyone wants to tackle it.

/assign @radufa
As he did the initial NATSS work.

sbezverk · 2019-02-10T04:47:44Z

@Harwayne I refactored it for a dynamic reconnect. Appreciate your review and feedback.

ghost · 2019-02-11T15:19:02Z

contrib/natss/pkg/dispatcher/dispatcher/dispatcher.go

+		nConn, err := stanutil.Connect(clusterchannelprovisioner.ClusterId, clientID, s.natssURL, s.logger.Sugar())
+		if err == nil {
+			s.subscriptionsMux.Lock()
+			s.natssConn = nConn


Perhaps i better to use:

defer s.subscriptionsMux.Unlock()

right after Lock()

Harwayne · 2019-02-11T17:48:41Z

contrib/natss/pkg/dispatcher/dispatcher/dispatcher.go

+	for {
+		nConn, err := stanutil.Connect(clusterchannelprovisioner.ClusterId, clientID, s.natssURL, s.logger.Sugar())
+		if err == nil {
+			s.subscriptionsMux.Lock()


Is this now intended to lock s.natssConn as well?

Right, adding a separate mutex just for natssConn does not seem to be efficient.

My preference would be to have a second mux, natssConnMux for natssConn and natssConnInProgress.

Harwayne · 2019-02-11T17:50:49Z

contrib/natss/pkg/dispatcher/dispatcher/dispatcher.go

@@ -67,8 +64,17 @@ func createReceiverFunction(s *SubscriptionsSupervisor, logger *zap.SugaredLogge
 		logger.Infof("Received message from %q channel", channel.String())
 		// publish to Natss
 		ch := getSubject(channel)
+		if s.natssConn == nil {
+			return fmt.Errorf("No Connection to NATS")


NATS -> NATSS

Harwayne · 2019-02-11T17:50:56Z

contrib/natss/pkg/dispatcher/dispatcher/dispatcher.go

 		if err := stanutil.Publish(s.natssConn, ch, &m.Payload, logger); err != nil {
 			logger.Errorf("Error during publish: %v", err)
+			if err.Error() == stan.ErrConnectionClosed.Error() {
+				logger.Error("Connection to NATS has been lost, attempting to reconnect.")


NATS -> NATSS

Here and all the other comments and log lines.

Harwayne · 2019-02-11T17:51:15Z

contrib/natss/pkg/dispatcher/dispatcher/dispatcher.go

 		if err := stanutil.Publish(s.natssConn, ch, &m.Payload, logger); err != nil {
 			logger.Errorf("Error during publish: %v", err)
+			if err.Error() == stan.ErrConnectionClosed.Error() {
+				logger.Error("Connection to NATS has been lost, attempting to reconnect.")
+				// Informing SubscriptionsSupervisor to re-establish connection to NATS


NATS -> NATSS

And a trailing period.

Harwayne · 2019-02-11T17:54:42Z

contrib/natss/pkg/dispatcher/dispatcher/dispatcher.go

+		select {
+		case <-s.connect:
+			// Establishing connection only if there is no already connection to NATS
+			if s.natssConn == nil {


What sets s.natssConn to nil? It looks like it will be nil when the struct is initialized, but after that, what sets it to nil?

Harwayne · 2019-02-11T17:56:24Z

contrib/natss/pkg/dispatcher/dispatcher/dispatcher.go

@@ -77,10 +83,45 @@ func createReceiverFunction(s *SubscriptionsSupervisor, logger *zap.SugaredLogge
 }

 func (s *SubscriptionsSupervisor) Start(stopCh <-chan struct{}) error {
+	// Trigger Connect to establish connection with NATS
+	s.connect <- struct{}{}


We should probably kick off s.Connect() here and pass in the stopCh.

ok, I will give it a try.

Harwayne · 2019-02-11T17:57:59Z

contrib/natss/pkg/dispatcher/main.go

@@ -68,6 +68,8 @@ func main() {
 	if err != nil {
 		logger.Fatal("Unable to create NATSS dispatcher.", zap.Error(err))
 	}
+	// Starting Connect to establish connection with NATS
+	go dispatcher.Connect()


My preference is to put this inside the dispatcher's Start(stopCh).

ok, I will give it a try.

I don't see this change in, did it not work?

sbezverk · 2019-02-12T01:04:31Z

@radufa I have extensively tested the latest revision, the connection get reestablished everytime. Appreciate if you try it in your test bed.

Harwayne · 2019-02-12T07:25:17Z

contrib/natss/pkg/dispatcher/dispatcher/dispatcher.go

@@ -67,8 +65,20 @@ func createReceiverFunction(s *SubscriptionsSupervisor, logger *zap.SugaredLogge
 		logger.Infof("Received message from %q channel", channel.String())
 		// publish to Natss
 		ch := getSubject(channel)
+		if s.natssConn == nil {


When is the lock required to be held to access or mutate s.natssConn?

The reason I did not put lock here is because the code which is setting this value is in the same goroutine. I tested this code with go tests --race it is all clean.

I don't think I follow your explanation. The only place I see s.natssConn set is connectWithRetry, is that running in this goroutine?

Based on the comment above, we only need to hold the lock while writing s.natssConn, is that why we don't grab the lock here?

Harwayne · 2019-02-12T07:31:23Z

contrib/natss/pkg/dispatcher/dispatcher/dispatcher.go

+	// re-attempting evey 60 seconds until the connection is established.
+	ticker := time.NewTicker(60 * time.Second)
+	defer ticker.Stop()
+	s.subscriptionsMux.Lock()


This will be held for potentially a long time, is that safe? Does it need to be held when we are not actually manipulating s.natssConn?

I will move it close to the place where the value gets set.

Harwayne · 2019-02-12T07:40:23Z

contrib/natss/pkg/dispatcher/dispatcher/dispatcher.go

+	dispatcher          *provisioners.MessageDispatcher
+	connect             chan struct{}
+	natssURL            string
+	subscriptionsMux    sync.Mutex


Add a comment saying when this has to be held.

sbezverk · 2019-02-12T14:47:19Z

/test pull-knative-eventing-integration-tests

chaodaiG · 2019-02-12T19:20:13Z

/test pull-knative-eventing-integration-tests

chaodaiG · 2019-02-12T20:11:04Z

/test pull-knative-eventing-integration-tests

Harwayne · 2019-02-13T07:06:47Z

contrib/natss/pkg/dispatcher/main.go

@@ -68,6 +68,8 @@ func main() {
 	if err != nil {
 		logger.Fatal("Unable to create NATSS dispatcher.", zap.Error(err))
 	}
+	// Starting Connect to establish connection with NATS
+	go dispatcher.Connect()


I don't see this change in, did it not work?

Harwayne · 2019-02-13T07:08:28Z

contrib/natss/pkg/dispatcher/dispatcher/dispatcher_test.go

+			logger.Fatalf("Failed to connect to NATSS!")
+		}
+	}
+	logger.Infof("><SB> (*s.natssConn).NatsConn().IsConnected(): %t", (*s.natssConn).NatsConn().IsConnected())


These log lines don't look like they should make it in as-is (I'm guessing is so you can find them quickly with grep).

Harwayne · 2019-02-13T07:09:33Z

contrib/natss/pkg/dispatcher/dispatcher/dispatcher.go

 }

 // NewDispatcher returns a new SubscriptionsSupervisor.
 func NewDispatcher(natssUrl string, logger *zap.Logger) (*SubscriptionsSupervisor, error) {
 	d := &SubscriptionsSupervisor{
 		logger:        logger,
 		dispatcher:    provisioners.NewMessageDispatcher(logger.Sugar()),
+		connect:       make(chan struct{}),


I think this is a blocking Channel, should it be non-blocking?

If so, then the various writes to it below can be made into non-blocking writes, because as long as there is at least one item in the channel, a reconnection attempt will occur, so multiple items in the channel aren't needed.

Harwayne · 2019-02-13T07:13:15Z

contrib/natss/pkg/dispatcher/dispatcher/dispatcher.go

@@ -67,8 +65,20 @@ func createReceiverFunction(s *SubscriptionsSupervisor, logger *zap.SugaredLogge
 		logger.Infof("Received message from %q channel", channel.String())
 		// publish to Natss
 		ch := getSubject(channel)
+		if s.natssConn == nil {


I don't think I follow your explanation. The only place I see s.natssConn set is connectWithRetry, is that running in this goroutine?

Based on the comment above, we only need to hold the lock while writing s.natssConn, is that why we don't grab the lock here?

Harwayne · 2019-02-13T07:13:45Z

contrib/natss/pkg/dispatcher/dispatcher/dispatcher.go

@@ -67,8 +68,20 @@ func createReceiverFunction(s *SubscriptionsSupervisor, logger *zap.SugaredLogge
 		logger.Infof("Received message from %q channel", channel.String())
 		// publish to Natss
 		ch := getSubject(channel)
+		if s.natssConn == nil {


Save s.natssConn to a local variable so that all the checks are done against the same value (i.e. we check for nil and then use the checked object).

Harwayne · 2019-02-13T07:18:11Z

contrib/natss/pkg/dispatcher/dispatcher/dispatcher.go

+	for {
+		nConn, err := stanutil.Connect(clusterchannelprovisioner.ClusterId, clientID, s.natssURL, s.logger.Sugar())
+		if err == nil {
+			s.subscriptionsMux.Lock()


My preference would be to have a second mux, natssConnMux for natssConn and natssConnInProgress.

Harwayne · 2019-02-13T07:19:15Z

contrib/natss/pkg/dispatcher/dispatcher/dispatcher.go

 	s.receiver.Start(stopCh)
 	return nil
 }

+func (s *SubscriptionsSupervisor) connectWithRetry() {


This should probably have a stop channel, even if only for the unit tests.

Signed-off-by: Serguei Bezverkhi <sbezverk@cisco.com>

Harwayne · 2019-02-14T19:11:46Z

contrib/natss/pkg/dispatcher/dispatcher/dispatcher.go

+		if currentNatssConn == nil {
+			return fmt.Errorf("No Connection to NATSS")
+		}
+		if reflect.ValueOf(currentNatssConn).IsNil() {


I'm not familiar with Go reflection, how does this differ from if currentNatssConn == nil three lines above?

this check helps to prevent panic when we do (*currentNattsConn).NatssCon().IsConnected(), as during debugging I cold see currentNattsConn is not nil but it point to nil interface when the connection gets lost. reflect.ValueOf allows us safely evaluate that the interface is not nil. Let me know if it makes sense.

Harwayne · 2019-02-14T19:15:33Z

contrib/natss/pkg/dispatcher/dispatcher/dispatcher.go

+			if err.Error() == stan.ErrConnectionClosed.Error() {
+				logger.Error("Connection to NATSS has been lost, attempting to reconnect.")
+				// Informing SubscriptionsSupervisor to re-establish connection to NATSS.
+				s.connect <- struct{}{}


We can be fancy and make this a non-blocking insertion. Something like replace this with s.signalReconnect() (not a good name). And then have:

func (s *SubscriptionsSupervisor) signalReconnect() { select { case s.connect <- struct{}{}: // Sent. default: // The Channel is already full, so a reconnection attempt will occur. } }

If we do this, then all writing to the channel should be via the s.signalReconnect() function.

ok, sounds good thanks for suggestion.

Harwayne · 2019-02-14T19:18:15Z

contrib/natss/pkg/dispatcher/dispatcher/dispatcher.go

+	for {
+		select {
+		case <-s.connect:
+			if s.natssConn == nil && !s.natssConnInProgress {


s.natssConn is protected by the lock, acquire the lock before reading it.

Harwayne · 2019-02-14T19:19:04Z

contrib/natss/pkg/dispatcher/dispatcher/dispatcher.go

+				go s.connectWithRetry(stopCh)
+				continue
+			}
+			if !(*s.natssConn).NatsConn().IsConnected() &&


If s.natssConn is nil and s.natssConnInProgress is true, will this panic (because it tries to dereference s.natssConn)?

provided a protection against this corner case.

sbezverk · 2019-02-14T20:28:01Z

@Harwayne Addressed your comments.

Signed-off-by: Serguei Bezverkhi <sbezverk@cisco.com>

ghost · 2019-02-15T08:25:23Z

I've tested the last changes, using the same scenario as discussed yesterday and it panics too, see please my last comments in #781

/hold

Signed-off-by: Serguei Bezverkhi <sbezverk@cisco.com>

ghost · 2019-02-19T09:46:56Z

contrib/natss/pkg/dispatcher/dispatcher/dispatcher.go

@@ -19,7 +19,7 @@ package dispatcher
 import (
 	"encoding/json"
 	"fmt"
-	"reflect"
+	// "reflect"


Please delete the commented line

ghost · 2019-02-19T09:50:02Z

contrib/natss/pkg/dispatcher/dispatcher/dispatcher.go

 	s.receiver.Start(stopCh)
 	return nil
 }

+func (s *SubscriptionsSupervisor) connectWithRetry(stopCh <-chan struct{}) {
+	// re-attempting evey 60 seconds until the connection is established.
+	ticker := time.NewTicker(60 * time.Second)


Please change it to 1s, 60s is too big and it's the only place which creates a new connection.

done, but do you think 1 second is not too aggressive?

ghost · 2019-02-19T09:51:22Z

contrib/natss/pkg/dispatcher/dispatcher/dispatcher.go

+		select {
+		case <-s.connect:
+			s.natssConnMux.Lock()
+			// currentNatssConn := s.natssConn


Please delete the commented line:

// currentNatssConn := s.natssConn

Signed-off-by: Serguei Bezverkhi <sbezverk@cisco.com>

ghost · 2019-02-19T13:16:37Z

contrib/natss/pkg/dispatcher/dispatcher/dispatcher.go

+			s.natssConnMux.Unlock()
+			return
+		}
+		s.logger.Sugar().Errorf("Connect() failed with error: %+v, retrying in 60 seconds", err)


Please change the message too: "retrying in 1 second"

ghost · 2019-02-19T13:17:57Z

/lgtm

Signed-off-by: Serguei Bezverkhi <sbezverk@cisco.com>

ghost · 2019-02-19T13:39:56Z

/lgtm

Harwayne · 2019-02-19T16:03:20Z

contrib/natss/pkg/dispatcher/dispatcher/dispatcher.go

 	d.receiver = provisioners.NewMessageReceiver(createReceiverFunction(d, logger.Sugar()), logger.Sugar())

 	return d, nil
 }

+func (s *SubscriptionsSupervisor) signalReconnect() {
+	s.connect <- struct{}{}


Let's make this non-blocking:

func (s *SubscriptionsSupervisor) signalReconnect() { select { case s.connect <- struct{}{}: // Sent. default: // The Channel is already full, so a reconnection attempt will occur. } }

will do in the follow up PR

Harwayne · 2019-02-19T16:24:14Z

contrib/natss/pkg/dispatcher/dispatcher/dispatcher.go

+	for {
+		nConn, err := stanutil.Connect(clusterchannelprovisioner.ClusterId, clientID, s.natssURL, s.logger.Sugar())
+		if err == nil {
+			// Locking here in order to reduce time in locked state


Add a trailing period.

Signed-off-by: Serguei Bezverkhi <sbezverk@cisco.com>

knative-metrics-robot · 2019-02-19T16:39:25Z

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	Delta
contrib/natss/pkg/dispatcher/dispatcher/dispatcher.go	56.5%	57.5%	1.0
contrib/natss/pkg/stanutil/stanutil.go	60.0%	58.6%	-1.4

Harwayne

/lgtm

sbezverk · 2019-02-19T17:23:41Z

@vaikas-google for approval

vaikas · 2019-02-19T18:04:52Z

/lgtm
/approve

knative-prow-robot · 2019-02-19T18:05:40Z

[APPROVALNOTIFIER] This PR is APPROVED

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

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

The pull request process is described here

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

googlebot added the cla: yes Indicates the PR's author has signed the CLA. label Feb 9, 2019

knative-prow-robot requested review from inlined and n3wscott February 9, 2019 00:35

knative-prow-robot added the size/XS Denotes a PR that changes 0-9 lines, ignoring generated files. label Feb 9, 2019

knative-prow-robot assigned Harwayne Feb 9, 2019

Harwayne reviewed Feb 9, 2019

View reviewed changes

knative-prow-robot assigned ghost Feb 9, 2019

knative-prow-robot added size/M Denotes a PR that changes 30-99 lines, ignoring generated files. and removed size/XS Denotes a PR that changes 0-9 lines, ignoring generated files. labels Feb 10, 2019

knative-prow-robot added size/L Denotes a PR that changes 100-499 lines, ignoring generated files. and removed size/M Denotes a PR that changes 30-99 lines, ignoring generated files. labels Feb 10, 2019

ghost reviewed Feb 11, 2019

View reviewed changes

Harwayne reviewed Feb 11, 2019

View reviewed changes

Harwayne reviewed Feb 12, 2019

View reviewed changes

Harwayne reviewed Feb 13, 2019

View reviewed changes

sbezverk added 8 commits February 13, 2019 15:53

Restart dispatcher on connection loss

f4728ab

Signed-off-by: Serguei Bezverkhi <sbezverk@cisco.com>

Addressing comments

165176c

Signed-off-by: Serguei Bezverkhi <sbezverk@cisco.com>

Refactor dispatcher for a dynamic reconnect

65c404e

Signed-off-by: Serguei Bezverkhi <sbezverk@cisco.com>

Removing mutes

420e9a0

Signed-off-by: Serguei Bezverkhi <sbezverk@cisco.com>

Fixing unit test

377d24c

Signed-off-by: Serguei Bezverkhi <sbezverk@cisco.com>

Fixing unit tests part 2

4cfdb95

Signed-off-by: Serguei Bezverkhi <sbezverk@cisco.com>

Fixing unit tests part 3 racing issue

a67ef6d

Signed-off-by: Serguei Bezverkhi <sbezverk@cisco.com>

Removing redundant mutex

b9f75a7

Signed-off-by: Serguei Bezverkhi <sbezverk@cisco.com>

Harwayne reviewed Feb 14, 2019

View reviewed changes

Addressing comments

d4ae9d7

Signed-off-by: Serguei Bezverkhi <sbezverk@cisco.com>

ghost mentioned this pull request Feb 15, 2019

NATSS Dispatcher cannot reconnect if NATS Streaming is restarted #781

Closed

Addressing racing conditions

bb5ae46

Signed-off-by: Serguei Bezverkhi <sbezverk@cisco.com>

ghost reviewed Feb 19, 2019

View reviewed changes

Addressing comments

aabb1c2

Signed-off-by: Serguei Bezverkhi <sbezverk@cisco.com>

ghost reviewed Feb 19, 2019

View reviewed changes

knative-prow-robot added the lgtm Indicates that a PR is ready to be merged. label Feb 19, 2019

Addressing more comments

2ceb894

Signed-off-by: Serguei Bezverkhi <sbezverk@cisco.com>

knative-prow-robot removed the lgtm Indicates that a PR is ready to be merged. label Feb 19, 2019

knative-prow-robot added the lgtm Indicates that a PR is ready to be merged. label Feb 19, 2019

Harwayne reviewed Feb 19, 2019

View reviewed changes

Addressing more comments part3

ea7e7b6

Signed-off-by: Serguei Bezverkhi <sbezverk@cisco.com>

knative-prow-robot removed the lgtm Indicates that a PR is ready to be merged. label Feb 19, 2019

Harwayne reviewed Feb 19, 2019

View reviewed changes

knative-prow-robot added the lgtm Indicates that a PR is ready to be merged. label Feb 19, 2019

knative-prow-robot assigned vaikas Feb 19, 2019

knative-prow-robot added the approved Indicates a PR has been approved by an approver from all required OWNERS files. label Feb 19, 2019

knative-prow-robot merged commit 2142e53 into knative:master Feb 19, 2019

sbezverk deleted the dispatcher_restart_on_disconnect branch February 19, 2019 23:51

Restart dispatcher on connection loss #796

Restart dispatcher on connection loss #796

Conversation

sbezverk commented Feb 9, 2019 • edited by grantr Loading

sbezverk commented Feb 9, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Harwayne commented Feb 9, 2019

sbezverk commented Feb 10, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sbezverk commented Feb 12, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sbezverk commented Feb 12, 2019

chaodaiG commented Feb 12, 2019

chaodaiG commented Feb 12, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sbezverk commented Feb 14, 2019

ghost commented Feb 15, 2019 • edited by ghost Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ghost Feb 19, 2019 • edited by ghost Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ghost commented Feb 19, 2019

ghost commented Feb 19, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

knative-metrics-robot commented Feb 19, 2019

Harwayne left a comment

Choose a reason for hiding this comment

sbezverk commented Feb 19, 2019

vaikas commented Feb 19, 2019

knative-prow-robot commented Feb 19, 2019

sbezverk commented Feb 9, 2019 •

edited by grantr

Loading

ghost commented Feb 15, 2019 •

edited by ghost

Loading

ghost Feb 19, 2019 •

edited by ghost

Loading