add internal queue for kubernetes watcher updates #2123

aanm · 2017-11-23T10:58:15Z

Summary of changes:

Created a function serializer that will execute the functions the same order as they were received. This will allow us to retry those same functions in case of a failure.

Fixes: #1966

Add internal queue for k8s watcher updates #1966

houndci-bot · 2017-11-23T10:58:21Z

pkg/serializer/func_serializer.go

+
+// NewFuncSerializer returns a funcSerializer that will be used to execute
+// functions in the same order they are enqueued.
+func NewFuncSerializer(size uint) *funcSerializer {


exported func NewFuncSerializer returns unexported type *serializer.funcSerializer, which can be annoying to use

tgraf · 2017-11-23T12:16:18Z

pkg/serializer/func_serializer.go

+
+// NewFuncSerializer returns a funcSerializer that will be used to execute
+// functions in the same order they are enqueued.
+func NewFuncSerializer(size uint) *funcSerializer {


s/size/queueSize/

tgraf · 2017-11-23T12:16:46Z

pkg/serializer/func_serializer.go

+
+// NewFuncSerializer returns a funcSerializer that will be used to execute
+// functions in the same order they are enqueued.
+func NewFuncSerializer(size uint) *funcSerializer {


I would call the whole thing FunctionQueue

tgraf · 2017-11-23T12:17:56Z

pkg/serializer/func_serializer.go

+// The waitFunc will be executed with the number of times `f` had returned an
+// error, in case `f` returns an error. waitFunc should return either true or
+// false depending if `f` should be executed again or not.
+func (fs *funcSerializer) Enqueue(f func() error, waitFunc func(int) bool) {


Please define a documented type for waitFunc which explains what int will be and what it has to return

tgraf · 2017-11-23T12:18:48Z

pkg/serializer/func_serializer.go

+// amount of functions queued, this function can block until the function
+// serializer is ready to receive more requests. In case `f` returns an error,
+// `f` won't be executed again.
+func (fs *funcSerializer) EnqueueOrDie(f func() error) {


This is not needed, just expose the NoRetry function

tgraf · 2017-11-23T12:21:11Z

pkg/serializer/func_serializer.go

+}
+
+// Run starts the funcSerializer internal worker. It will be stopped once
+// `stopCh` is closed or receives a value.


This is very misleading, stopped indicates that this is a go routine which it is not. Why not just spawn the routine in NewFuncSerializer? I would also store the stop channel in the funcSerializer struct and provide a Stop() API so the channel is hidden.

tgraf · 2017-11-23T12:24:50Z

pkg/serializer/func_serializer.go

+	noRetry = func(i int) bool { return false }
+)
+
+type funcProp struct {


Call this queuedFunction

tgraf · 2017-11-23T12:24:59Z

pkg/serializer/func_serializer.go

+}
+
+type funcSerializer struct {
+	funcs chan funcProp


Call this queue

tgraf · 2017-11-23T12:25:38Z

pkg/serializer/func_serializer.go

+				default:
+				}
+				retries++
+				if err := f.f(); err != nil {


We want to log these errors, right?

I don't think it's a good idea to do it here since this might be used on different places of the code. When calling the Enqueue function it should be done as follow:

serNodes.Enqueue(func() error { err := d.updateK8sNodeV1(oldK8sNode, newK8sNode) if err != nil { log.Debugf("this is a dedicated k8s function that failed") } return err }, serializer.NoRetry)

OK, that's fine, but then document that the error will only be used to determine failure if non-nil and that it is the responsibility of the function to log the error.

houndci-bot · 2017-11-23T13:54:17Z

pkg/serializer/func_serializer.go

+
+// NewFunctionQueue returns a functionQueue that will be used to execute
+// functions in the same order they are enqueued.
+func NewFunctionQueue(queueSize uint) *functionQueue {


exported func NewFunctionQueue returns unexported type *serializer.functionQueue, which can be annoying to use

houndci-bot · 2017-11-23T13:54:18Z

pkg/serializer/func_serializer.go

+
+// NewFunctionQueue returns a functionQueue that will be used to execute
+// functions in the same order they are enqueued.
+func NewFunctionQueue(queueSize uint) *functionQueue {


exported func NewFunctionQueue returns unexported type *serializer.functionQueue, which can be annoying to use

houndci-bot · 2017-11-23T13:56:04Z

pkg/serializer/func_queue.go

+
+// NewFunctionQueue returns a functionQueue that will be used to execute
+// functions in the same order they are enqueued.
+func NewFunctionQueue(queueSize uint) *functionQueue {


exported func NewFunctionQueue returns unexported type *serializer.functionQueue, which can be annoying to use

tgraf · 2017-11-24T10:06:00Z

pkg/serializer/func_queue.go

+package serializer
+
+var (
+	// NoRetry always returns false independently the number of retries.


independently of the number

tgraf · 2017-11-24T10:08:32Z

pkg/serializer/func_queue.go

+	NoRetry = func(int) bool { return false }
+)
+
+// WaitFunc will be executed with the number of times `f` had returned an


f is undocumented so I would rewrite this as follows:

WaitFunc will be invoked each time a queued function has returned an error. nRetries will be set to the number of consecutive execution failures that have occurred so far. The WaitFunc must return true if execution must be retried or false if the function must be returned from the queue.

Add function serializer that allows functions to be executed in order with the ability to retry the execution in case of failure. Signed-off-by: André Martins <andre@cilium.io>

Signed-off-by: André Martins <andre@cilium.io>

aanm requested a review from a team November 23, 2017 10:58

aanm requested a review from a team as a code owner November 23, 2017 10:58

houndci-bot reviewed Nov 23, 2017

View reviewed changes

aanm force-pushed the 1966-add-k8s-watcher-queue branch from d9d27f9 to bd4cad2 Compare November 23, 2017 11:55

tgraf requested changes Nov 23, 2017

View reviewed changes

aanm requested a review from tgraf November 23, 2017 13:54

aanm force-pushed the 1966-add-k8s-watcher-queue branch from bd4cad2 to 1952aba Compare November 23, 2017 13:54

houndci-bot reviewed Nov 23, 2017

View reviewed changes

aanm force-pushed the 1966-add-k8s-watcher-queue branch from 1952aba to 8d73ae9 Compare November 23, 2017 13:55

houndci-bot reviewed Nov 23, 2017

View reviewed changes

aanm force-pushed the 1966-add-k8s-watcher-queue branch from 8d73ae9 to a46e3ed Compare November 23, 2017 16:20

tgraf requested changes Nov 24, 2017

View reviewed changes

aanm added 2 commits November 24, 2017 12:19

pkg/serializer: add serializer to executed functions in order

dd6fa8b

Add function serializer that allows functions to be executed in order with the ability to retry the execution in case of failure. Signed-off-by: André Martins <andre@cilium.io>

k8s: add function serializer for all requests received by k8s watchers

1543401

Signed-off-by: André Martins <andre@cilium.io>

aanm force-pushed the 1966-add-k8s-watcher-queue branch from a46e3ed to 1543401 Compare November 24, 2017 11:20

aanm requested a review from tgraf November 24, 2017 11:21

aanm added ready-to-merge This PR has passed all tests and received consensus from code owners to merge. and removed pending-review labels Nov 24, 2017

tgraf approved these changes Nov 24, 2017

View reviewed changes

tgraf merged commit a631ce2 into master Nov 24, 2017

tgraf deleted the 1966-add-k8s-watcher-queue branch November 24, 2017 12:42

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

add internal queue for kubernetes watcher updates #2123

add internal queue for kubernetes watcher updates #2123

aanm commented Nov 23, 2017 •

edited

houndci-bot Nov 23, 2017

tgraf Nov 23, 2017

aanm Nov 23, 2017

tgraf Nov 23, 2017

aanm Nov 23, 2017

tgraf Nov 23, 2017

aanm Nov 23, 2017

tgraf Nov 23, 2017

aanm Nov 23, 2017

tgraf Nov 23, 2017

aanm Nov 23, 2017

tgraf Nov 23, 2017

aanm Nov 23, 2017

tgraf Nov 23, 2017

aanm Nov 23, 2017

tgraf Nov 23, 2017

aanm Nov 23, 2017

tgraf Nov 23, 2017

aanm Nov 23, 2017

houndci-bot Nov 23, 2017

houndci-bot Nov 23, 2017

houndci-bot Nov 23, 2017

tgraf Nov 24, 2017

tgraf Nov 24, 2017

add internal queue for kubernetes watcher updates #2123

add internal queue for kubernetes watcher updates #2123

Conversation

aanm commented Nov 23, 2017 • 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

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

aanm commented Nov 23, 2017 •

edited