Implement the ChannelSource #6

lichuqiang · 2018-06-08T11:06:58Z

Address: kubernetes-sigs/kubebuilder#254

pwittrock · 2018-06-08T13:32:33Z

pkg/controller/controller_manager.go

@@ -166,6 +175,9 @@ func (cm *controllerManager) Start(stop <-chan struct{}) error {
 	cm.stop = stop
 	cm.informers.Start(stop)

+	// Start the outer informer
+	cm.outerInformer.Run(stop)


We should be consistent about Start vs Run in our naming

pwittrock · 2018-06-08T14:03:39Z

pkg/controller/source/channel_source.go

+// (eh.g. GitHub Webhook callback).  ChannelSource requires the user to wire the external
+// source (eh.g. http handler) to write GenericEvents to the underlying channel.
+type ChannelSource struct {
+	informer *ChannelSourceInformer


Why do we need an informer for this? I envisioned reading events from the Channel and passing them directly to the Event handler

pwittrock · 2018-06-08T14:24:15Z

pkg/controller/source/channel_source.go

+		return fmt.Errorf("must call InjectInformer on ChannelSource before calling Start")
+	}
+
+	cs.informer.AddEventHandler(internal.EventHandler{Queue: queue, EventHandler: handler})


Instead of adding an EventHandler, why not add the event directly to the Channel. e.g.

type ChannelSource struct { once sync.Once Source chan event.GenericEvent stop chan <-struct{} dest []chan event.GenericEvent } func (cs *ChannelSource) InjectStop(stop chan <-struct{}) {cs.stop = stop} func (cs *ChannelSource) Start( handler eventhandler.EventHandler, queue workqueue.RateLimitingInterface) error { cs.once.Do(func() { // Fanout GenericEvents to all EventHandler / Queue pairs Watching this source go func() { // TODO: Make this select on stop and end the routine // TODO: Handle closed channel for _, evt := range cs.Source { for _, dst := range cs.dest { // TODO: Make this select on stop and end the routine go func() { dst <- evt } } } } }) dst := make(chan event.GenericEvent) go func() { for evt := range dst {handler.Generic(evt, queue)} } cs.dst = append(cs.dst, dst) }

pwittrock · 2018-06-08T17:46:33Z

Hey can we keep this in another PR. I think it would be useful for a SlackBotSource or a GitHubSource of events.

lichuqiang

Changed to a simpler way.
Will keep the "informer mode" in my private repo, in case we need it someday :)

Still I'm not very happy with the once.Do part. It's in fact similar to the startup behavior of an informer.
Maybe we should explicitly perform it outside the func, rather than inject a stop channel into the source

lichuqiang · 2018-06-09T08:15:24Z

pkg/controller/source/source.go

+				select {
+				case <-cs.stop:
+					// Close source channel.
+					// TODO: should we close the source channel here?


Not sure if we need to close the source channel here.
Maybe users meant to use it somewhere out of the controller, and handle its lifecycle themselves?

I think the lifecycle of the source channel should be completely managed by the user. We shouldn't create it or close it for them.

lichuqiang · 2018-06-09T08:19:24Z

pkg/controller/manager.go

@@ -139,6 +143,12 @@ func (cm *controllerManager) injectInto(i interface{}) error {
 	if _, err := inject.DoInformers(cm.informers, i); err != nil {


I guess we need a fix here.
We expect to have it inject whichever params that i implement, rather than return directly once an error occur

pwittrock · 2018-06-09T16:12:11Z

pkg/controller/source/source.go

+	}
+
+	// stop should have been injected before Start was called
+	if cs.stop == nil {


This is a bug in the error you copied :P This should be InjectSourceChannel instead of DoStop

pwittrock · 2018-06-09T16:14:06Z

pkg/controller/source/source.go

+
+// InjectSourceChannel is internal should be called only by the Controller.
+// It is used to inject the source event channel initialized by the ControllerManager.
+func (cs *ChannelSource) InjectSourceChannel(c chan event.GenericEvent) error {


What do you think of having the user provide the channel when creating the object (like Type on KindSource)?

Private Injected fields are initialized by the ControllerManager. Public fields should be supplied by the user.

I'd say I haven't got enough knowledge on how users will use it.
I think you have more say over this point, so let's make it the way you suggested :)

pwittrock · 2018-06-09T16:15:38Z

pkg/controller/source/source.go

+	// once ensures the event distribution goroutine will be performed only once
+	once sync.Once
+	// source is the source channel to fetch GenericEvents
+	source chan event.GenericEvent


Wondering if this should be a read channel only. What do you think?

pwittrock · 2018-06-09T16:18:38Z

pkg/controller/source/source.go

+					close(cs.source)
+
+					// Close destination channels
+					cs.destLock.Lock()


not sure if this is a common pattern, but if we wrap this in an in-line function we can call defer unlock.

func() { cs.destLock.Lock() defer cs.destLock.Unlock() for _, dst := range cs.dest { close(dst) } }()

Graceful vs Easy to understand
I'm fine with both

or

case <-cs.stop: cs.doStop() case evt := <-cs.source: cs.DoEvent(evt)

that makes sense

pwittrock · 2018-06-09T16:19:48Z

pkg/controller/source/source.go

+				case evt := <-cs.source:
+					cs.destLock.Lock()
+					for _, dst := range cs.dest {
+						dst <- evt


Should this be synchronous or asynchronous? With synchronous if one of the EventHandlers blocks, it will block all of them. Is there any advantage to this?

That's what I meant to discuss with you. I guess we might have to tolerate this.
if put it under a goroutine, we'll have to face the race condition of write message to closed channel, even if we have lock here, it's out of control.

Most obviously, you'll see error in go vet if we make it asynchronous

I guess this is still asynchronous because of the buffer. Is the buffer a more common / better pattern? Please leave a comment with the information for future readers.

yeah, I think the buffer should have been big enough normally, and if we still see it blocked, the controller is thought to be in an abnormal state.
Will add comment for clarification.

pwittrock · 2018-06-09T17:02:18Z

pkg/runtime/inject/inject.go

+
+// DoSourceChannel will set source channel on i and return the result if it implements SourceChannel.
+// Returns false if i does not implement SourceChannel.
+func DoSourceChannel(ch chan event.GenericEvent, i interface{}) (bool, error) {


Lets no inject this.

pwittrock · 2018-06-09T17:05:53Z

pkg/controller/source/source.go

@@ -31,6 +32,11 @@ import (
 	logf "github.com/kubernetes-sigs/controller-runtime/pkg/runtime/log"
 )

+const (
+	// initialBufferSize is the initial number of event notifications that can be buffered.
+	initialBufferSize = 1024


lets make this a variable on the ChannelSource and write a test for what happens when we run out of buffer (e.g. we can set the buffer size to 1)

So when do you think it should be specified?
Or we'll in fact always use the default value, and the field will only be changed in test?

pwittrock · 2018-06-09T17:06:24Z

pkg/controller/source/source_test.go

@@ -203,4 +204,57 @@ var _ = Describe("Source", func() {
 			})
 		})
 	})
+	Describe("ChannelSource", func() {


Add a test for when the ChannelSource is reused by multiple watches with different EventHandlers.

pwittrock · 2018-06-09T17:06:59Z

pkg/controller/source/source_test.go

+		var c chan struct{}
+		var ch chan event.GenericEvent
+
+		BeforeEach(func() {


Add tests for error conditions where the stop or channel are not specified.

pwittrock · 2018-06-09T17:07:43Z

pkg/controller/source/source_test.go


 	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"


Lets also add a controller_integration_test.go case that exercises the end-to-end flow.

pwittrock · 2018-06-09T17:08:40Z

This looks good. A few minor comments.

pwittrock · 2018-06-11T21:35:27Z

pkg/controller/source/source.go

+	dest []chan event.GenericEvent
+
+	// DestBufferSize is the specified buffer size of dest channels
+	DestBufferSize int


followup: Include the default value in the comment.

pwittrock · 2018-06-11T21:39:44Z

pkg/controller/source/source_test.go

+			It("should block if exceed buffer size", func(done Done) {
+				ch := make(chan event.GenericEvent)
+				evt := event.GenericEvent{}
+				interval := 5 * time.Second


This will make the test run a really long time. Instead of sleeping, consider waiting on a signal / channel.

pwittrock · 2018-06-11T21:40:27Z

pkg/controller/source/source_test.go

+					},
+					GenericFunc: func(q2 workqueue.RateLimitingInterface, evt event.GenericEvent) {
+						defer GinkgoRecover()
+						time.Sleep(interval)


Don't sleep, wait on a channel or something

pwittrock · 2018-06-11T21:40:40Z

pkg/controller/source/source_test.go

+				distributeInterval := time.Now().Sub(beforeEvent)
+				Expect(distributeInterval >= interval).To(BeTrue())
+				close(done)
+			}, 15)


keep the timeout at 1 second

pwittrock · 2018-06-11T21:50:44Z

pkg/controller/source/source_test.go

+				<-c
+				close(done)
+			})
+			It("should block if exceed buffer size", func(done Done) {


I think we care less that it blocks, and more that it doesn't blow up. I think the main thing we want to test is that we can write 3 events to the source channel while the eventhandler is blocked, then unblock the eventhandler, and all the 3 events will get processed.

pwittrock · 2018-06-11T21:51:51Z

lgtm once the test run time is reduce back down to sub second

droot

Changes look ok. However I am still struggling to come up with a concrete use case :).

Thinking more about this.

droot · 2018-06-11T22:09:25Z

pkg/controller/source/source.go

+
+// InjectStop is internal should be called only by the Controller.
+// It is used to inject the stop channel initialized by the ControllerManager.
+func (cs *ChannelSource) InjectStop(stop <-chan struct{}) error {


Better name it explicit InjectStopChannel ?

droot · 2018-06-11T22:37:20Z

pkg/runtime/inject/inject.go

+
+// Stop is used by the ControllerManager to inject stop channel into Sources, EventHandlers, Predicates, and
+// Reconciles
+type Stop interface {


Stoppable ?

droot · 2018-06-11T22:41:58Z

pkg/controller/source/source.go

@@ -45,21 +51,129 @@ type Source interface {
 	Start(eventhandler.EventHandler, workqueue.RateLimitingInterface, ...predicate.Predicate) error
 }

+var _ Source = &ChannelSource{}
+
 // ChannelSource is used to provide a source of events originating outside the cluster


I am trying to understand the correct usage pattern for the channel-source so that we can advise users accordingly.

For external triggers which maps to changes in a type in K8s, controller will automatically pick them up because it is watching that type.
Now left are the ones which don't result in any data change in the type directly but impact the type, channel source is the right fit for those.

We have to come up with a better example demonstrating this source.

Yeah, this is of great importance. Maybe we can make it when constructing sample controller.

lichuqiang · 2018-06-12T03:24:56Z

Rebased on the latest code; also, addressed the comments :)

droot

/lgtm

pwittrock · 2018-06-12T16:17:13Z

/approve

Implement the ChannelSource

Add links to releases

…min-cluster Create minimal clusterscoped cache

) * Add missing godocs and provide default for HTTPClient

k8s-ci-robot added do-not-merge/work-in-progress Indicates that a PR should not merge because it is a work in progress. cncf-cla: yes Indicates the PR's author has signed the CNCF CLA. size/L Denotes a PR that changes 100-499 lines, ignoring generated files. labels Jun 8, 2018

pwittrock reviewed Jun 8, 2018

View reviewed changes

lichuqiang force-pushed the master branch 2 times, most recently from fa5d8e9 to 6cab6c3 Compare June 9, 2018 08:05

lichuqiang commented Jun 9, 2018

View reviewed changes

lichuqiang force-pushed the master branch 2 times, most recently from 3d5cfcd to 039baf6 Compare June 9, 2018 10:19

lichuqiang changed the title ~~WIP: Implement the ChannelSource~~ Implement the ChannelSource Jun 9, 2018

k8s-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 Jun 9, 2018

pwittrock reviewed Jun 9, 2018

View reviewed changes

lichuqiang force-pushed the master branch 3 times, most recently from c421290 to e844cac Compare June 11, 2018 11:37

pwittrock reviewed Jun 11, 2018

View reviewed changes

droot reviewed Jun 11, 2018

View reviewed changes

lichuqiang force-pushed the master branch from 5499aa8 to 5988a09 Compare June 12, 2018 03:09

implement the ChannelSource

b0b0957

lichuqiang force-pushed the master branch from 5988a09 to b0b0957 Compare June 12, 2018 03:17

k8s-ci-robot assigned droot Jun 12, 2018

droot reviewed Jun 12, 2018

View reviewed changes

k8s-ci-robot added the lgtm "Looks good to me", indicates that a PR is ready to be merged. label Jun 12, 2018

pwittrock added the approved Indicates a PR has been approved by an approver from all required OWNERS files. label Jun 12, 2018

k8s-ci-robot merged commit b5597f5 into kubernetes-sigs:master Jun 12, 2018

pwittrock mentioned this pull request Jun 12, 2018

Fix coverage regressions #24

Closed

justinsb pushed a commit to justinsb/controller-runtime that referenced this pull request Dec 7, 2018

Merge pull request kubernetes-sigs#6 from lichuqiang/master

9afc9fb

Implement the ChannelSource

DirectXMan12 pushed a commit that referenced this pull request Jan 31, 2020

Merge pull request #6 from pwittrock/iterations

105ee20

Add links to releases

varshaprasad96 pushed a commit to varshaprasad96/controller-runtime that referenced this pull request Mar 14, 2022

Merge pull request kubernetes-sigs#6 from varshaprasad96/fix-watch-ad…

f3ffab7

…min-cluster Create minimal clusterscoped cache

ncdc pushed a commit to ncdc/controller-runtime that referenced this pull request May 27, 2022

Add missing godocs and provide default for HTTPClient (kubernetes-sigs#6

7a42e65

) * Add missing godocs and provide default for HTTPClient

		@@ -139,6 +143,12 @@ func (cm *controllerManager) injectInto(i interface{}) error {
		if _, err := inject.DoInformers(cm.informers, i); err != nil {


		corev1 "k8s.io/api/core/v1"
		metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

Implement the ChannelSource #6

Implement the ChannelSource #6

Uh oh!

Conversation

lichuqiang commented Jun 8, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

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

Uh oh!

Choose a reason for hiding this comment

Uh oh!

pwittrock commented Jun 8, 2018

Uh oh!

lichuqiang left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

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

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!

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

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!

pwittrock commented Jun 9, 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!

pwittrock Jun 8, 2018 •

edited

Loading

lichuqiang left a comment •

edited

Loading

lichuqiang Jun 9, 2018 •

edited

Loading

lichuqiang Jun 9, 2018 •

edited

Loading