From 412e7a2418037d43329768ea7aa02829b425800e Mon Sep 17 00:00:00 2001 From: Todd Date: Tue, 4 Oct 2022 11:07:50 -0700 Subject: [PATCH] Add interfaces and error codes for processing downstream router connections (#2511) * Add interfaces and error codes for processing downstream router connections --- internal/daemon/controller/controller.go | 29 +++++++++++++------ internal/daemon/worker/worker.go | 36 ++++++++++++++++-------- internal/errors/code.go | 4 +++ internal/errors/code_test.go | 10 +++++++ internal/errors/info.go | 8 ++++++ 5 files changed, 66 insertions(+), 21 deletions(-) diff --git a/internal/daemon/controller/controller.go b/internal/daemon/controller/controller.go index a6dd86b54..361a20aea 100644 --- a/internal/daemon/controller/controller.go +++ b/internal/daemon/controller/controller.go @@ -49,6 +49,11 @@ type downstreamRouter interface { // StartRouteMgmtTicking starts a ticker which manages the router's // connections. StartRouteMgmtTicking(context.Context, func() string, int) error + + // ProcessPendingConnections starts a function that continually processes + // incoming client connections. This only returns when the provided context + // is done. + StartProcessingPendingConnections(context.Context, func() string) } // downstreamWorkersTicker defines an interface for a ticker that maintains the @@ -425,19 +430,25 @@ func (c *Controller) Start() error { }() if c.downstreamRoutes != nil { - c.tickerWg.Add(1) + c.tickerWg.Add(2) + + servNameFn := func() string { + switch { + case c.conf.RawConfig.Controller.Name != "": + return c.conf.RawConfig.Controller.Name + default: + return "unknown controller name" + } + } + go func() { + defer c.tickerWg.Done() + c.downstreamRoutes.StartProcessingPendingConnections(c.baseContext, servNameFn) + }() go func() { defer c.tickerWg.Done() err := c.downstreamRoutes.StartRouteMgmtTicking( c.baseContext, - func() string { - switch { - case c.conf.RawConfig.Controller.Name != "": - return c.conf.RawConfig.Controller.Name - default: - return "unknown controller name" - } - }, + servNameFn, -1, ) if err != nil { diff --git a/internal/daemon/worker/worker.go b/internal/daemon/worker/worker.go index b88aa864c..e970edfef 100644 --- a/internal/daemon/worker/worker.go +++ b/internal/daemon/worker/worker.go @@ -46,6 +46,11 @@ type downstreamRouter interface { // StartRouteMgmtTicking starts a ticker which manages the router's // connections. StartRouteMgmtTicking(context.Context, func() string, int) error + + // ProcessPendingConnections starts a function that continually processes + // incoming client connections. This only returns when the provided context + // is done. + StartProcessingPendingConnections(context.Context, func() string) } // downstreamers provides at least a minimum interface that must be met by a @@ -389,7 +394,7 @@ func (w *Worker) Start() error { // Rather than deal with some of the potential error conditions for Add on // the waitgroup vs. Done (in case a function exits immediately), we will // always start rotation and simply exit early if we're using KMS - w.tickerWg.Add(3) + w.tickerWg.Add(2) go func() { defer w.tickerWg.Done() w.startStatusTicking(w.baseContext, w.sessionManager, &w.addressReceivers) @@ -398,24 +403,31 @@ func (w *Worker) Start() error { defer w.tickerWg.Done() w.startAuthRotationTicking(w.baseContext) }() - go func() { - defer w.tickerWg.Done() - if w.downstreamRoutes != nil { + + if w.downstreamRoutes != nil { + w.tickerWg.Add(2) + servNameFn := func() string { + if s := w.LastStatusSuccess(); s != nil { + return s.WorkerId + } + return "unknown worker id" + } + go func() { + defer w.tickerWg.Done() + w.downstreamRoutes.StartProcessingPendingConnections(w.baseContext, servNameFn) + }() + go func() { + defer w.tickerWg.Done() err := w.downstreamRoutes.StartRouteMgmtTicking( w.baseContext, - func() string { - if s := w.LastStatusSuccess(); s != nil { - return s.WorkerId - } - return "unknown worker id" - }, + servNameFn, -1, // indicates the ticker should run until cancelled. ) if err != nil { errors.Wrap(w.baseContext, err, op) } - } - }() + }() + } w.workerStartTime = time.Now() w.started.Store(true) diff --git a/internal/errors/code.go b/internal/errors/code.go index 3a1abbc07..acb614ef1 100644 --- a/internal/errors/code.go +++ b/internal/errors/code.go @@ -45,6 +45,10 @@ const ( CycleFound = 121 // CycleFound represents an error when a cycle is found between a parent and child worker WorkerConnNotFound = 122 // WorkerConnNotFound represents an error when a connection to a worker is not found KmsWorkerUnsupportedOperation = 123 // KmsWorkerUnsupportedOperation represents an error when a KMS worker is not supported for an operation + // Note: Currently unused in OSS + RetryLimitExceeded = 124 // RetryLimitExceeded represents an error when a retry limit is exceeded + // Note: Currently unused in OSS + QueueIsFull = 125 // QueueIsFull results in attempting to add an item to a queue which is full AuthAttemptExpired Code = 198 // AuthAttemptExpired represents an expired authentication attempt AuthMethodInactive Code = 199 // AuthMethodInactive represents an error that means the auth method is not active. diff --git a/internal/errors/code_test.go b/internal/errors/code_test.go index e60476574..a13b498f6 100644 --- a/internal/errors/code_test.go +++ b/internal/errors/code_test.go @@ -337,6 +337,16 @@ func TestCode_Both_String_Info(t *testing.T) { c: KmsWorkerUnsupportedOperation, want: KmsWorkerUnsupportedOperation, }, + { + name: "RetryLimitExceeded", + c: RetryLimitExceeded, + want: RetryLimitExceeded, + }, + { + name: "QueueIsFull", + c: QueueIsFull, + want: QueueIsFull, + }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { diff --git a/internal/errors/info.go b/internal/errors/info.go index 80e7c0e7d..c73f6b513 100644 --- a/internal/errors/info.go +++ b/internal/errors/info.go @@ -272,4 +272,12 @@ var errorCodeInfo = map[Code]Info{ Message: "unsupported operation for a kms worker", Kind: State, }, + QueueIsFull: { + Message: "queue is full", + Kind: State, + }, + RetryLimitExceeded: { + Message: "retry limit exceeded", + Kind: State, + }, }