Changes to subscriber pattern #828

maleck13 · 2018-03-09T15:51:08Z

Describe what this PR does and why we need it:
Updates to how we handle the channels and subscribers based on the proposal outlined as part of
#638

@jmrodri @eriknelson initial WIP work here for early feedback. Not complete yet but also not miles away. Would like any feedback you have to ensure we are still on the same page and I am not heading down any rabbit holes 😄

jmrodri

I'm going to mark this as request changes because this is only a first pass review. I need to look at the engine a little closer, I'm not understanding something that I want to look at a bit closer. Overall it's a nice job (no pun intended).

jmrodri · 2018-03-09T18:53:09Z

pkg/app/app.go

@@ -62,8 +62,6 @@ var (
 )

 const (
-	// MsgBufferSize - The buffer for the message channel.
-	MsgBufferSize = 20


So we're not using a buffered channel anymore? I guess if each one has its own channel it makes sense to be a blocking one item channel.

Yes also we are reading from that channel and handing of to the subscribers async, so we are always ready to read the next message. If we wanted to slow throughput we could very easily use a sync.WaitGroup in combination with a buffered channel.

I don't see any reason to artificially slow the processing of msgs.

jmrodri · 2018-03-09T18:54:44Z

pkg/app/app.go

@@ -213,43 +211,43 @@ func CreateApp() App {
 	validateRegistryNames(app.registry)

 	log.Debug("Initializing WorkEngine")
-	app.engine = broker.NewWorkEngine(MsgBufferSize)
+	stateSubscriber := broker.NewJobStateSubscriber(app.dao)


jmrodri · 2018-03-09T19:02:22Z

pkg/broker/broker.go

@@ -1059,13 +1059,6 @@ func (a AnsibleBroker) Unbind(
 	} else {
 		log.Warning("Broker configured to *NOT* launch and run APB unbind")
 	}
-
-	err = cleanupUnbind(&bindInstance, &serviceInstance, bindExtCreds, a.dao)


Why is this gone? It was part of the unbind_subscriber before, what is doing this now?

I see it was moved to cleanupAfterUnbind in the job_state_subscriber.

jmrodri · 2018-03-09T19:09:12Z

pkg/broker/types.go

@@ -304,4 +304,12 @@ type SubscriberDAO interface {
 	SetState(id string, state apb.JobState) (string, error)
 	GetServiceInstance(id string) (*apb.ServiceInstance, error)
 	DeleteServiceInstance(id string) error
+	GetBindInstance(id string) (*apb.BindInstance, error)
+	DeleteBindInstance(id string) error
+	SetServiceInstance(id string, serviceInstance *apb.ServiceInstance) error


I'm not a fan of the SubscriberDAO. It feels weird to me. But that's for a different day.

As this is adding more and more methods, why don't we just make DAO now that DAO is an interface itself?

We either have or easily can create a mock for it as well.

Yep no issue with changing that as we already have the DAO interface

@shawn-hurley How do you feel about me creating a follow on issue to remove the subscriber dao and create the mocks etc? I don't see an existing mock at the moment and quite a few tests would need updating. I would like to get this landed and base the apb state work from master with these changes already available?

jmrodri · 2018-03-09T19:09:34Z

pkg/broker/types.go

+	SetServiceInstance(id string, serviceInstance *apb.ServiceInstance) error
+}
+
+// WorkSubscriber - Interface tha wraps the Subscribe method


typo: tha -> that

also it says that wraps the Subscribe method, is that correct?

nope comment needs changing 👍

jmrodri · 2018-03-09T19:10:37Z

pkg/broker/work_engine.go

-func NewWorkEngine(bufferSize int) *WorkEngine {
-	return &WorkEngine{topics: make(map[WorkTopic]chan JobMsg), bufsz: bufferSize}
+func NewWorkEngine() *WorkEngine {
+	return &WorkEngine{jobs: make(map[string]chan JobMsg), subscribers: map[WorkTopic][]WorkSubscriber{}}


again, why is the channel a non buffered channel?

As currently we are handing off the messages immediately and asynchronously we should always be able to read from and send to the channel so in this case there is no benefit to a buffered channel.

Where we may benefit from a buffered channel is if we used it in conjunction with a sync.WaitGroup

Something like:

wg := sync.WaitGroup{} for msg := range engine.jobs[token] { wg.Add(len(engine.subscribers)) for _, sub := range engine.subscribers[topic] { go sub.Notify(msg, wg) // All subscribers receive msg at same time, and each subscriber will call Done() when it is finished } wg.Wait() // wait for all subscribers to be done }

This design would allow each of the subscribers to receive the messages at the same time, but force us to wait until they are all complete before accepting the next message. In this case having a buffer would be valuable as it would allow the Job to continue to send messages even if our subscribers were running a little slow.
The question really is if there is value in waiting for all the subscribers to finish before accepting the next message.
I am leaning towards yes on this due to the fact that state will be changed and modified and there is no guarantee on that the order a set of go routines were created will also be the order they are executed in.

jmrodri · 2018-03-09T19:19:53Z

pkg/broker/work_engine.go

@@ -50,10 +50,33 @@ func (engine *WorkEngine) StartNewAsyncJob(
 	if token == "" {
 		token = engine.Token()
 	}
-	go work.Run(token, engine.topic(topic))
+	go engine.start(token, work, topic)


This feels really weird to me. Starting the engine for each job. Feels like the engine should always be running and waiting for new jobs to be added to it to be run.

This could be called startSubscriberListeningAndRun(). I would prefer to keep in smaller maybe something like startJob?

jmrodri · 2018-03-09T19:20:33Z

pkg/mock/dao.go

@@ -86,6 +86,49 @@ func (mp *SubscriberDAO) GetServiceInstance(id string) (*apb.ServiceInstance, er
 	return retOb.(*apb.ServiceInstance), mp.Errs["GetServiceInstance"]
 }

+//DeleteBindInstance mock impl


spaced after the //

shawn-hurley · 2018-03-09T21:25:41Z

pkg/broker/types.go

@@ -304,4 +304,12 @@ type SubscriberDAO interface {
 	SetState(id string, state apb.JobState) (string, error)
 	GetServiceInstance(id string) (*apb.ServiceInstance, error)
 	DeleteServiceInstance(id string) error
+	GetBindInstance(id string) (*apb.BindInstance, error)
+	DeleteBindInstance(id string) error
+	SetServiceInstance(id string, serviceInstance *apb.ServiceInstance) error


As this is adding more and more methods, why don't we just make DAO now that DAO is an interface itself?

We either have or easily can create a mock for it as well.

shawn-hurley · 2018-03-09T21:36:12Z

pkg/broker/work_engine.go

+		for msg := range engine.jobs[token] {
+			for _, sub := range engine.subscribers[topic] {
+				//TODO edge case consider the fact that a subscriber may never exit and so we would leak go routines
+				go sub.Notify(msg)


In regards to the comment, I do think that having a context timeout makes sense here.

I think a simple implementation would be to wrap this in a func() that waited for a configurable amount of time.

The notify interface could then have an error return that we would log if something does go wrong. and if it returns with nil then everythng is all good.

... for _, sub := range engine.subscribers[topic] { c := make(chan error, 1) go waitForNotify(engine.duration, sub, msg) .... func waitForNotify(d time.Duration, w WorkSubscriber, j JobMessage) { select { case err := w.Notify(j): handle error or if nil return case <-d: } }

Going to chew this one over 🤔 I am not sure you can do the function call this way in the select statement

I think your right, I think you need to pass a channel to waitForNotify function and add the timeout bits to a function that gets launched with the go routine above.

Something like this?
https://play.golang.org/p/pRjyCXfZgSi

updated to take into account your feedback plus look at adding in the wait group and using context https://gist.github.com/maleck13/d2c903eac6064422966d64715118345e

shawn-hurley · 2018-03-09T21:59:21Z

pkg/broker/work_engine.go

@@ -50,10 +50,33 @@ func (engine *WorkEngine) StartNewAsyncJob(
 	if token == "" {
 		token = engine.Token()
 	}
-	go work.Run(token, engine.topic(topic))
+	go engine.start(token, work, topic)


This could be called startSubscriberListeningAndRun(). I would prefer to keep in smaller maybe something like startJob?

shawn-hurley · 2018-03-09T22:02:37Z

I think this is looking really good, does a good job of reducing complexity!

maleck13 · 2018-03-12T16:30:47Z

Ok so I have updated based on feedback and discussions. It is still WIP, but again wanted to give opportunity for feedback.
After much discussion (thanks guys) the design changed to make the start job more complex than the original commit, but I think it provides a more robust solution.

Main changes
We have introduced a WaitGroup for each message that comes in, along with a context object with timeout.
We now have the following on a per job basis:

msg-> buffered-channel -> fanout-to-subscribers-async -> lock-until-all-subs-done-> read-next-msg

The key concern for me here was to avoid a situation where writes could be executed in the wrong order.
There is no guarantee on the order that go routines will execute. So without the synchronization, there would be the possibility that a write could happen in the wrong order. Now however I do not believe this is the case as we use the WaitGroup to ensure subscribers are finished and the order messages come off a channel is FIFO

ping @shawn-hurley @jmrodri @eriknelson

Will be adding a number of unit tests here and also looking at a metrics subscriber in the next day or so.

jmrodri

Some minor comment changes, variable name change (or question). Overall I like this updated version better.

jmrodri · 2018-03-12T21:32:57Z

pkg/app/app.go

 	// defaultClusterURLPreFix - prefix for the ansible service broker.
 	defaultClusterURLPreFix = "/ansible-service-broker"
+	// MsgBufferSize - The buffer for the message channel.
+	MsgBufferSize = 20


+1 this makes me feel better :)

jmrodri · 2018-03-12T21:33:19Z

pkg/app/app.go

@@ -213,43 +213,43 @@ func CreateApp() App {
 	validateRegistryNames(app.registry)

 	log.Debug("Initializing WorkEngine")
+	stateSubscriber := broker.NewJobStateSubscriber(app.dao)


👍 to the single stateSubscriber.

jmrodri · 2018-03-12T21:37:30Z

pkg/broker/job_state_subscriber.go

+	log.Debugf("JobStateSubscriber Notify : msg state %v ", msg.State)
+	id := msg.InstanceUUID
+	if isBinding(msg) {
+		id = msg.BindingUUID


Can msg.BindingUUID ever be ""? I would probably change isBinding to also ensure BindingUUID is not empty. Otherwise, we would store an empty string as the key. Maybe there is no way this could happen, I'll see how the flow goes.

It seems if it can ever be empty that it should fail before getting here, similar with instanceUUID.

jmrodri · 2018-03-12T21:45:44Z

pkg/broker/work_engine.go

-	topics map[WorkTopic]chan JobMsg
-	bufsz  int
+	subscribers   map[WorkTopic][]WorkSubscriber
+	jobs          map[string]chan JobMsg


jobs is weird because these aren't the jobs. They're the channels used by the jobs for messages. The name implies the job structs are in this map but that is not the case. Calling jobs["token"] will not return me the job I created. Consider renaming it.

changed to jobChannels

jmrodri · 2018-03-12T21:46:06Z

pkg/broker/work_engine.go

@@ -50,10 +54,65 @@ func (engine *WorkEngine) StartNewAsyncJob(
 	if token == "" {
 		token = engine.Token()
 	}
-	go work.Run(token, engine.topic(topic))
+	go engine.startJob(token, work, topic)


startJob is better and more clear.

👍 I'm not sure it should be a private fn though? I suppose it's fine since we're really the only ones using it but I thought it was useful to design it from a perspective of "this is a public lib". No requested changes, but spitballing.

jmrodri · 2018-03-12T21:47:54Z

pkg/broker/work_engine.go

+
+	go func() {
+		// listen for a new message for the job keyed to this token and hand off to the subscribers async. Wait for them all to be done before accepting
+		// the next message


reformat this comment to be a couple shorter lines. The first one is really long.

jmrodri · 2018-03-12T21:48:06Z

pkg/broker/work_engine.go

+					}()
+					// notify the subscriber
+					go waitForNotify(sub, msg, notifySignal)
+					//act on whichever happens first the subscriber's notify method completing or the timeout


space after //

jmrodri · 2018-03-12T21:48:18Z

pkg/broker/work_engine.go

+					ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) //TODO make configurable
+					// used to tell us when the subscribers notify method is completed
+					notifySignal := make(chan struct{})
+					//If our subscriber times out or returns normally we will always clean up


space after //

jmrodri · 2018-03-12T21:48:50Z

pkg/broker/work_engine.go

@@ -66,7 +125,7 @@ func (engine *WorkEngine) StartNewSyncJob(
 		token = engine.Token()
 	}

-	work.Run(token, engine.topic(topic))
+	engine.startJob(token, work, topic)


jmrodri · 2018-03-12T21:50:35Z

pkg/broker/work_engine.go

-	return engine.topics
+// GetActiveJobs - Get list of active jobs
+func (engine *WorkEngine) GetActiveJobs() map[string]chan JobMsg {
+	return engine.jobs


is this really active jobs? I would expect it to return an array of jobs.

changed to GetActiveJobChannels

maleck13 · 2018-03-27T15:02:14Z

@jmrodri I added back the clean up of unbind state when launch_apb_on_bind is false. I think this will fix the CI failure. At lease it does locally when running make ci
Wondering what you think to moving the unbind clean up into the dao layer as it is essentially a fascade around a set of dao calls?

Maybe

func(dao d)DeleteBinding(bindingInstance, serviceInstance)error{
    if err := d.DeleteBindInstance(bindInstance.ID.String()); err != nil {
        return nil, false, err
    }
    serviceInstance.RemoveBinding(bindInstance.ID)
    if err = d.SetServiceInstance(serviceInstance.ID.String(), serviceInstance); err != nil {
        return nil, false, err
    }
    return nil
}

maleck13 · 2018-03-27T16:23:13Z

added to DAO can remove if need be

maleck13 · 2018-03-28T08:31:28Z

@shawn-hurley @jmrodri @eriknelson this is no longer WIP please review when ready.

shawn-hurley · 2018-03-28T16:13:04Z

pkg/dao/dao.go

@@ -79,6 +79,9 @@ type Dao interface {
 	// DeleteBindInstance - Delete the binding instance for an id in the kvp API.
 	DeleteBindInstance(string) error

+	// DeleteBinding - Delete the binding instance and remove the assocation with the service instance.
+	DeleteBinding(apb.BindInstance, apb.ServiceInstance) error


shawn-hurley · 2018-03-28T20:42:52Z

@maleck13 If you could rebase this, the build should start passing

…emoving concerns around race conditions

eriknelson

No requests for change; I love how much code this removed, it's a great improvement. Thanks @maleck13

eriknelson · 2018-04-04T16:49:49Z

pkg/broker/work_engine.go

+	case <-ctx.Done():
+		return
+	default:
+		signal <- struct{}{}


Is this just a trigger? Usually I see bools used.

eriknelson · 2018-04-04T16:50:39Z

pkg/broker/work_engine.go

+					wg.Add(1)
+					// ensure things don't get locked up.
+					// Each subscriber has up to the configured amount of time to complete its action
+					ctx, cancel := context.WithTimeout(context.Background(), engine.subscriberTimeout*time.Second)


This is so much better 👍

openshift-ci-robot added do-not-merge/work-in-progress Indicates that a PR should not merge because it is a work in progress. size/XXL Denotes a PR that changes 1000+ lines, ignoring generated files. labels Mar 9, 2018

jmrodri suggested changes Mar 9, 2018

View reviewed changes

shawn-hurley approved these changes Mar 9, 2018

View reviewed changes

maleck13 force-pushed the issue-638-subcriber-improvements branch from f7be0e0 to 308ef44 Compare March 12, 2018 16:16

jmrodri suggested changes Mar 12, 2018

View reviewed changes

maleck13 force-pushed the issue-638-subcriber-improvements branch from 308ef44 to 3213f38 Compare March 20, 2018 13:23

openshift-bot added the needs-rebase Indicates a PR cannot be merged because it has merge conflicts with HEAD. label Mar 24, 2018

maleck13 force-pushed the issue-638-subcriber-improvements branch from 3213f38 to 2f8adeb Compare March 26, 2018 12:23

openshift-bot removed the needs-rebase Indicates a PR cannot be merged because it has merge conflicts with HEAD. label Mar 26, 2018

maleck13 mentioned this pull request Mar 27, 2018

config for ci launch_apb_on_bind = false seems to cause issues #858

Closed

maleck13 force-pushed the issue-638-subcriber-improvements branch from bfcf48a to 6211d29 Compare March 27, 2018 16:22

maleck13 changed the title ~~[wip] initial changes to subscriber pattern~~ Changes to subscriber pattern Mar 27, 2018

openshift-ci-robot removed the do-not-merge/work-in-progress Indicates that a PR should not merge because it is a work in progress. label Mar 27, 2018

shawn-hurley approved these changes Mar 28, 2018

View reviewed changes

maleck13 added 4 commits March 29, 2018 08:15

[wip] initial changes to subscriber pattern

67b031b

and sync waitGroup and context support, allowing better control and r…

f2097b7

…emoving concerns around race conditions

changes based on feedback

e51808f

Add DeleteBinding to dao

13be109

maleck13 force-pushed the issue-638-subcriber-improvements branch from 6211d29 to 13be109 Compare March 29, 2018 07:16

jmrodri approved these changes Mar 29, 2018

View reviewed changes

eriknelson approved these changes Apr 4, 2018

View reviewed changes

eriknelson merged commit a3b1ceb into openshift:master Apr 4, 2018

Changes to subscriber pattern #828

Changes to subscriber pattern #828

Conversation

maleck13 commented Mar 9, 2018

jmrodri left a 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

maleck13 Mar 12, 2018 • edited

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

maleck13 Mar 12, 2018 • edited

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

shawn-hurley commented Mar 9, 2018

maleck13 commented Mar 12, 2018

jmrodri left a 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

maleck13 commented Mar 27, 2018 • edited

maleck13 commented Mar 27, 2018

maleck13 commented Mar 28, 2018

Choose a reason for hiding this comment

shawn-hurley commented Mar 28, 2018

eriknelson left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

maleck13 Mar 12, 2018 •

edited

maleck13 Mar 12, 2018 •

edited

maleck13 commented Mar 27, 2018 •

edited