From 5088740bef0bcc53c921aa41c797a7f80f39cc98 Mon Sep 17 00:00:00 2001 From: Peter Stanko Date: Sat, 4 Feb 2023 16:44:24 +0100 Subject: [PATCH 1/2] feat: migrate to async runner --- .github/workflows/unit-test.yml | 3 + cmd/serve.go | 7 +- pkg/apprun/start.go | 76 --------- pkg/apprun/start_test.go | 249 --------------------------- pkg/asyncrun/async_run.go | 104 ++++++++++++ pkg/asyncrun/async_run_test.go | 274 ++++++++++++++++++++++++++++++ pkg/rest/chiapp/web_server_run.go | 14 +- 7 files changed, 390 insertions(+), 337 deletions(-) delete mode 100644 pkg/apprun/start.go delete mode 100644 pkg/apprun/start_test.go create mode 100644 pkg/asyncrun/async_run.go create mode 100644 pkg/asyncrun/async_run_test.go diff --git a/.github/workflows/unit-test.yml b/.github/workflows/unit-test.yml index c32f26e..d5e85f2 100644 --- a/.github/workflows/unit-test.yml +++ b/.github/workflows/unit-test.yml @@ -22,8 +22,11 @@ jobs: with: go-version: ^1.19 + - name: Install Task uses: arduino/setup-task@v1 + with: + repo-token: ${{ secrets.GITHUB_TOKEN }} - name: Unit test run: | diff --git a/cmd/serve.go b/cmd/serve.go index 973fde6..f811dd4 100644 --- a/cmd/serve.go +++ b/cmd/serve.go @@ -33,12 +33,9 @@ var serveCmd = &cobra.Command{ GraceFullTimeout: 30 * time.Second, } - errC, err := chiapp.RunWebServer(ctx, server, ops) - if err != nil { - return err - } + errC := chiapp.RunWebServer(ctx, server, ops) - if err = <-errC; errC != nil { + if err := <-errC; errC != nil { return err } diff --git a/pkg/apprun/start.go b/pkg/apprun/start.go deleted file mode 100644 index 6d5ff2a..0000000 --- a/pkg/apprun/start.go +++ /dev/null @@ -1,76 +0,0 @@ -package apprun - -import ( - "context" - "time" - - "github.com/rs/zerolog/log" -) - -const defaultTimeout = 30 * time.Second - -// StartParams parameters for the Start function; -// Start defines the start callback - function that will be executed async way -// Stop defines a stop/shutdown callback - function that will executed to cleanup/stop the Start -// function -type StartParams struct { - Start func(ctx context.Context) error - Stop func(ctx context.Context) error - GraceTimeout time.Duration -} - -// Start executing the StartParams.Start function asynchronously and then clean up/shutdown the -// execution -func Start(appCtx context.Context, params StartParams) (chan error, error) { - if params.GraceTimeout == 0 { - params.GraceTimeout = defaultTimeout - } - - if params.Start == nil { - params.Start = noopFn - } - - if params.Stop == nil { - params.Stop = noopFn - } - - errC := make(chan error) - runtimeConnect := context.Background() - - // run the Start function - go func() { - if err := params.Start(runtimeConnect); err != nil { - errC <- err - } - }() - - // run the Stop function - go func() { - // if the application context is closed, the function will continue - <-appCtx.Done() - // Stop signal with grace period of 30 seconds - shutdownCtx, cancelCallback := context.WithTimeout(runtimeConnect, params.GraceTimeout) - defer cancelCallback() - - go func() { - <-shutdownCtx.Done() - if shutdownCtx.Err() == context.DeadlineExceeded { - log.Err(shutdownCtx.Err()).Msg("graceful shutdown timed out.. forcing exit.") - errC <- shutdownCtx.Err() - } - }() - - if err := params.Stop(runtimeConnect); err != nil { - errC <- err - } - - close(errC) - - }() - - return errC, nil -} - -func noopFn(ctx context.Context) error { - return nil -} diff --git a/pkg/apprun/start_test.go b/pkg/apprun/start_test.go deleted file mode 100644 index 6a21883..0000000 --- a/pkg/apprun/start_test.go +++ /dev/null @@ -1,249 +0,0 @@ -package apprun - -import ( - "context" - "fmt" - "testing" - "time" - - "github.com/stretchr/testify/assert" -) - -const deadlineTimeout = 5 * time.Second - -func TestStart(t *testing.T) { - t.Run("no start and stop function, close asap", func(t *testing.T) { - params := StartParams{ - Start: nil, - Stop: nil, - GraceTimeout: 0, - } - - appCtx, cancelCallback := context.WithCancel(context.Background()) - cancelCallback() - - errC, err := Start(appCtx, params) - assert.NoError(t, err) - - select { - case err := <-errC: - assert.NoError(t, err) - case <-time.After(deadlineTimeout): - assert.Fail(t, "test timeout") - } - }) - - t.Run("start and stop, with channels", func(t *testing.T) { - appCtx, cancelCallback := context.WithCancel(context.Background()) - unlockStartChan := make(chan bool) - unlockStopChan := make(chan bool) - params := StartParams{ - Start: startWithTimeout(t, unlockStartChan), - Stop: stopWithTimeout(t, unlockStopChan, false), - GraceTimeout: 1 * time.Second, - } - - go func() { - select { - case <-unlockStartChan: - assert.True(t, true, "start channel should unblock") - cancelCallback() - case <-time.After(deadlineTimeout): - assert.Fail(t, "test timeout for start") - } - }() - - go func() { - select { - case <-unlockStopChan: - assert.True(t, true, "stop channel should unblock") - case <-time.After(deadlineTimeout): - assert.Fail(t, "test timeout for stop") - } - }() - - errC, err := Start(appCtx, params) - assert.NoError(t, err) - - select { - case err := <-errC: - assert.NoError(t, err) - case <-time.After(deadlineTimeout): - assert.Fail(t, "test timeout") - } - }) - - t.Run("start context timeout", func(t *testing.T) { - appCtx, cancelCallback := context.WithTimeout(context.Background(), 500*time.Millisecond) - unlockStartChan := make(chan bool) - unlockStopChan := make(chan bool) - params := StartParams{ - Start: startWithTimeout(t, unlockStartChan), - Stop: stopWithTimeout(t, unlockStopChan, false), - GraceTimeout: 1 * time.Second, - } - - go func() { - select { - case <-unlockStartChan: - assert.True(t, true, "start channel should unblock") - cancelCallback() - case <-time.After(deadlineTimeout): - assert.Fail(t, "test timeout for start") - } - }() - - go func() { - select { - case <-unlockStopChan: - assert.True(t, true, "stop channel should unblock") - case <-time.After(deadlineTimeout): - assert.Fail(t, "test timeout for stop") - } - }() - - errC, err := Start(appCtx, params) - assert.NoError(t, err) - - select { - case err := <-errC: - assert.NoError(t, err) - case <-time.After(deadlineTimeout): - assert.Fail(t, "test timeout") - } - }) - - t.Run("stop graceful timeout", func(t *testing.T) { - appCtx, cancelCallback := context.WithTimeout(context.Background(), 500*time.Millisecond) - unlockStartChan := make(chan bool) - unlockStopChan := make(chan bool) - params := StartParams{ - Start: startWithTimeout(t, unlockStartChan), - Stop: stopWithTimeout(t, unlockStopChan, true), - GraceTimeout: 800 * time.Millisecond, - } - - go func() { - select { - case <-unlockStartChan: - assert.True(t, true, "start channel should unblock") - cancelCallback() - case <-time.After(deadlineTimeout): - assert.Fail(t, "test timeout for start") - } - }() - - go func() { - select { - case <-unlockStopChan: - assert.True(t, true, "stop channel should unblock") - case <-time.After(deadlineTimeout): - assert.Fail(t, "test timeout for stop") - } - }() - - errC, err := Start(appCtx, params) - assert.NoError(t, err) - - select { - case err := <-errC: - assert.Error(t, err) - case <-time.After(deadlineTimeout): - assert.Fail(t, "test timeout") - } - }) - - t.Run("start returns error", func(t *testing.T) { - appCtx, cancelCallback := context.WithTimeout(context.Background(), 500*time.Millisecond) - defer cancelCallback() - - unlockStopChan := make(chan bool) - startErr := fmt.Errorf("some start error") - params := StartParams{ - Start: func(ctx context.Context) error { - return startErr - }, - Stop: stopWithTimeout(t, unlockStopChan, true), - GraceTimeout: 800 * time.Millisecond, - } - - go func() { - select { - case <-unlockStopChan: - assert.True(t, true, "stop channel should unblock") - case <-time.After(deadlineTimeout): - assert.Fail(t, "test timeout for stop") - } - }() - - errC, err := Start(appCtx, params) - assert.NoError(t, err) - - select { - case err := <-errC: - assert.ErrorIs(t, err, startErr) - case <-time.After(deadlineTimeout): - assert.Fail(t, "test timeout") - } - }) - - t.Run("stop error", func(t *testing.T) { - appCtx, cancelCallback := context.WithTimeout(context.Background(), 500*time.Millisecond) - unlockStartChan := make(chan bool) - unlockStopChan := make(chan bool) - params := StartParams{ - Start: startWithTimeout(t, unlockStartChan), - Stop: stopWithTimeout(t, unlockStartChan, true), - GraceTimeout: 800 * time.Millisecond, - } - - go func() { - select { - case <-unlockStartChan: - assert.True(t, true, "start channel should unblock") - cancelCallback() - case <-time.After(deadlineTimeout): - assert.Fail(t, "test timeout for start") - } - }() - - go func() { - select { - case <-unlockStopChan: - assert.True(t, true, "stop channel should unblock") - case <-time.After(deadlineTimeout): - assert.Fail(t, "test timeout for stop") - } - }() - - errC, err := Start(appCtx, params) - assert.NoError(t, err) - - select { - case err := <-errC: - assert.Error(t, err) - case <-time.After(deadlineTimeout): - assert.Fail(t, "test timeout") - } - }) -} - -func startWithTimeout(t *testing.T, unlock chan bool) func(ctx context.Context) error { - return func(ctx context.Context) error { - unlock <- true - time.Sleep(10 * time.Second) - assert.Fail(t, "Start overtime!") - return nil - } -} - -func stopWithTimeout(t *testing.T, unlock chan bool, sleep bool) func(ctx context.Context) error { - return func(ctx context.Context) error { - unlock <- true - if sleep { - time.Sleep(2 * time.Second) - assert.Fail(t, "Stop overtime!") - } - return nil - } -} diff --git a/pkg/asyncrun/async_run.go b/pkg/asyncrun/async_run.go new file mode 100644 index 0000000..5d94b7e --- /dev/null +++ b/pkg/asyncrun/async_run.go @@ -0,0 +1,104 @@ +// Package asyncrun contains tha asynchronous runner, +// it's responsibilities are: +// - run the callback asynchronously, with the Stop/Shutdown functionality +package asyncrun + +import ( + "context" + "fmt" + "time" + + "github.com/rs/zerolog/log" +) + +const defaultTimeout = 30 * time.Second + +// ErrShutdownTimeout this error will be returned if the graceful timeout period is exceeded +var ErrShutdownTimeout = fmt.Errorf("runtime context timeout: %w", context.DeadlineExceeded) + +// Params parameters for the AsyncRun function; +// Run defines the main callback - function that will be executed async way +// Shutdown defines a stop/shutdown callback - function that will do a cleanup/shutdown +// of the Run function +type Params struct { + Run func(ctx context.Context) error + Shutdown func(ctx context.Context) error + GraceTimeout time.Duration +} + +// AsyncRun will start the async execution of the params.Run callback provided through the +// params argument +// The function returns an error channel - the caller should be checking for the runtime errors +// and wait until there is some error message or the channel is closed +// most general usage: +// ```go +// errC := AsyncRun(ctx, params) +// +// if err = <-errC; err != nil { +// return err +// } +// +// / ``` +func AsyncRun(appCtx context.Context, params Params) chan error { + if params.GraceTimeout == 0 { + params.GraceTimeout = defaultTimeout + } + + if params.Run == nil { + params.Run = noopFn + } + + if params.Shutdown == nil { + params.Shutdown = noopFn + } + + errC := make(chan error) + runtimeCtx, runtimeCancelCallback := context.WithCancel(context.Background()) + + // run the Run function + go func() { + if err := params.Run(runtimeCtx); err != nil { + errC <- err + } + }() + + // run the Stop function + go func() { + // if the application context is closed, the function will continue + // we start with the shutdown process + <-appCtx.Done() + + // Either wait for shutdown to complete + // or timeout in params.GraceTimeout seconds (default: 30) + go func() { + defer close(errC) + + select { + // shutdown completed + case <-runtimeCtx.Done(): + log.Debug().Msg("graceful shutdown completed in time") + + // graceful shutdown timeout + case <-time.After(params.GraceTimeout): + defer runtimeCancelCallback() + err := ErrShutdownTimeout + log.Error().Err(err). + Msg("graceful shutdown timed out.. forcing exit.") + errC <- err + } + }() + + if err := params.Shutdown(runtimeCtx); err != nil { + errC <- err + } + + defer runtimeCancelCallback() + }() + + return errC +} + +// noopFn represents a no operation function +func noopFn(ctx context.Context) error { + return nil +} diff --git a/pkg/asyncrun/async_run_test.go b/pkg/asyncrun/async_run_test.go new file mode 100644 index 0000000..1829785 --- /dev/null +++ b/pkg/asyncrun/async_run_test.go @@ -0,0 +1,274 @@ +package asyncrun + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/rs/zerolog/log" + "github.com/stretchr/testify/assert" +) + +const testDeadline = 5 * time.Second +const gracefulTimeout = 1 * time.Second + +func TestAsyncRun(t *testing.T) { + t.Run("all params are empty or 0, the context is closed before calling", func(t *testing.T) { + // all params are empty + params := Params{ + Run: nil, + Shutdown: nil, + GraceTimeout: 0, + } + + appCtx, cancelCallback := context.WithCancel(context.Background()) + // we close the context before calling + cancelCallback() + + // we start the execution + errC := AsyncRun(appCtx, params) + + // assert there was no execution error + // the start and shutdown will do nothing + assertErrChan(t, errC, nil) + }) + + t.Run( + "Run the params.Run and shutdown functions with direct ctx cancel", + func(t *testing.T) { + // It Tests: it uses the direct context cancellation the assertion + appCtx, cancelCallback := context.WithCancel(context.Background()) + defer cancelCallback() + // this channel is used whether the params.Run has been called + unlockRunC := make(chan bool) + // this channel is used whether the params.Shutdown has been called + unlockShutdownC := make(chan bool) + params := Params{ + // run func is an example function that will do unlockRunC<-true when called, + // then sleep + Run: runFunc(t, unlockRunC), + // run func is an example function that will do unlockShutdownC<-true when called, + // since it should not timeout (3-rd parameter) there is no sleep, + // it would end immediately + Shutdown: shutdownFunc(t, unlockShutdownC, false), + GraceTimeout: gracefulTimeout, + } + + // assert that unlock channel contains value a.k.a the Run has been called + assertRunUnlock(t, unlockRunC, cancelCallback) + + // assert that unlock channel contains value a.k.a the Shutdown has been called + assertShutdownUnlock(t, unlockShutdownC) + + // Execute the function + errC := AsyncRun(appCtx, params) + + // assert there was no execution error + assertErrChan(t, errC, nil) + }, + ) + + t.Run("Run the params.Run function context timeout", func(t *testing.T) { + // It Tests: it uses context.WithTimeout instead of direct context cancellation + + appCtx, cancelCallback := context.WithTimeout(context.Background(), 500*time.Millisecond) + defer cancelCallback() + + unlockRunC := make(chan bool) + unlockShutdownC := make(chan bool) + params := Params{ + Run: runFunc(t, unlockRunC), + Shutdown: shutdownFunc(t, unlockShutdownC, false), + GraceTimeout: gracefulTimeout, + } + + // do not call the cancel callback, it should timeout + assertRunUnlock(t, unlockRunC, nil) + + assertShutdownUnlock(t, unlockShutdownC) + + errC := AsyncRun(appCtx, params) + + // assert there was no execution error + assertErrChan(t, errC, nil) + }) + + t.Run("Shutdown function will exceed the graceful timeout period", func(t *testing.T) { + // It Tests: it tests the graceful timeout for the shutdown function + // the Shutdown function will wait (using sleep) until the Graceful period ends + + appCtx, cancelCallback := context.WithTimeout(context.Background(), 300*time.Millisecond) + defer cancelCallback() + + unlockRunC := make(chan bool) + unlockShutdownC := make(chan bool) + params := Params{ + Run: runFunc(t, unlockRunC), + Shutdown: shutdownFunc(t, unlockShutdownC, true), + GraceTimeout: gracefulTimeout, + } + + // do not call the cancel callback, it should timeout + assertRunUnlock(t, unlockRunC, nil) + + assertShutdownUnlock(t, unlockShutdownC) + + errC := AsyncRun(appCtx, params) + + // assert there was execution error: shutdown timeout + assertErrChan(t, errC, func(err error) { + assert.ErrorIs(t, err, ErrShutdownTimeout) + }) + }) + + t.Run("The Run function returns error", func(t *testing.T) { + // It tests: The Run function returns an error + // this error is sent to `errC<-` channel and we are asserting it + + appCtx, cancelCallback := context.WithTimeout(context.Background(), 500*time.Millisecond) + defer cancelCallback() + + // only the shutdown is called + unlockShutdownC := make(chan bool) + // error that will be returned after the Run starts + startErr := fmt.Errorf("some run error") + params := Params{ + // the Run will return an error immediately after call + Run: func(ctx context.Context) error { + return startErr + }, + Shutdown: shutdownFunc(t, unlockShutdownC, true), + GraceTimeout: gracefulTimeout, + } + + assertShutdownUnlock(t, unlockShutdownC) + + errC := AsyncRun(appCtx, params) + + // assert there was execution error: startErr + assertErrChan(t, errC, func(err error) { + assert.ErrorIs(t, err, startErr) + }) + }) + + t.Run("The Shutdown function will produce an error", func(t *testing.T) { + // It tests: The Shutdown function returns an error + // this error is sent to `errC<-` channel and we are asserting it + appCtx, cancelCallback := context.WithTimeout(context.Background(), 500*time.Millisecond) + unlockRunC := make(chan bool) + shutdownErr := fmt.Errorf("unable to shutdown") + params := Params{ + Run: runFunc(t, unlockRunC), + Shutdown: func(ctx context.Context) error { + return shutdownErr + }, + GraceTimeout: gracefulTimeout, + } + + assertRunUnlock(t, unlockRunC, cancelCallback) + + errC := AsyncRun(appCtx, params) + + // assert there was execution error: shutdownErr + assertErrChan(t, errC, func(err error) { + assert.ErrorIs(t, err, shutdownErr) + }) + }) +} + +// assertErrChan assert whether the errC has been set and to what it has been set +// otherwise the test would timeout +// match - if nil we assert that there is no error, otherwise it takes a match function +func assertErrChan(t *testing.T, errC chan error, match func(err error)) { + if match == nil { + match = func(err error) { + assert.NoError(t, err) + } + } + + select { + case err := <-errC: + match(err) + // If there is no error or error channel is not closed, test will timeout + case <-time.After(testDeadline): + assert.Fail(t, "test timeout") + } +} + +// assertRunUnlock check whether the Run function channel has been unlocked +// it has to be unlocked - it means that function has been called +// after it was called, we can call the `cancelCallback` that would manually stop the execution +func assertRunUnlock(t *testing.T, unlockRunC chan bool, cancelCallback context.CancelFunc) { + go func() { + select { + // wait until the Run channel unlocks + case <-unlockRunC: + assert.True(t, true, "run channel should unblock") + if cancelCallback != nil { + cancelCallback() + } + // this case will happen only if the Run function is not called + // Run function should be called always + case <-time.After(testDeadline): + assert.Fail(t, "test timeout for run callback - THIS SHOULD NEVER HAPPEN") + } + }() +} + +func assertShutdownUnlock(t *testing.T, unlockShutdownC chan bool) { + go func() { + select { + case <-unlockShutdownC: + assert.True(t, true, "shutdown channel should unblock") + // this case will happen only if the Shutdown function is not called + // Shutdown function should be called always + case <-time.After(testDeadline): + assert.Fail(t, "test timeout for shutdown - THIS SHOULD NEVER HAPPEN") + } + }() +} + +// runFunc test implementation of the Run function +// unlock channel is set in order to check whether the function has been called +func runFunc(t *testing.T, unlock chan bool) func(ctx context.Context) error { + return func(ctx context.Context) error { + unlock <- true + + select { + case <-ctx.Done(): + log.Info().Msg("context is done run shutdown") + // execution ended successfully + case <-time.After(10 * time.Second): + log.Info().Msg("Should not happen - for run func") + assert.Fail(t, "Run function overtime!") + } + return nil + } +} + +// shutdownFunc test implementation of the shutdown function +// unlock channel is set in order to check whether the function has been called +// shouldTimeout if true - the function should timeout and the exec. should end with GrateTimeout +func shutdownFunc( + t *testing.T, + unlock chan bool, + shouldTimeout bool, +) func(ctx context.Context) error { + return func(ctx context.Context) error { + unlock <- true + + if shouldTimeout { + select { + case <-ctx.Done(): // this should happen after `gracefulTimeout` -> which is 1 second + log.Info().Msg("context is done for shutdown") + assert.True(t, true, "Shutdown context should be done") + case <-time.After(5 * time.Second): // 5 second is >> then 1 second (gracefulTimeout) + log.Info().Msg("Should not happen - for shutdown func") + assert.Fail(t, "Shutdown function overtime!") + } + } + + return nil + } +} diff --git a/pkg/rest/chiapp/web_server_run.go b/pkg/rest/chiapp/web_server_run.go index 5739d67..6440a01 100644 --- a/pkg/rest/chiapp/web_server_run.go +++ b/pkg/rest/chiapp/web_server_run.go @@ -2,10 +2,9 @@ package chiapp import ( "context" + "github.com/pestanko/miniscrape/pkg/asyncrun" "net/http" "time" - - "github.com/pestanko/miniscrape/pkg/apprun" ) // RunOps defines runtime options @@ -15,21 +14,22 @@ type RunOps struct { GraceFullTimeout time.Duration } -func RunWebServer(appCtx context.Context, handler http.Handler, ops RunOps) (chan error, error) { +func RunWebServer(appCtx context.Context, handler http.Handler, ops RunOps) chan error { server := http.Server{ Addr: ops.ListenAddr, Handler: handler, ReadTimeout: ops.ReadTimeout, } - params := apprun.StartParams{ - Start: func(ctx context.Context) error { + params := asyncrun.Params{ + Run: func(ctx context.Context) error { return server.ListenAndServe() }, - Stop: func(ctx context.Context) error { + Shutdown: func(ctx context.Context) error { return server.Shutdown(ctx) }, GraceTimeout: ops.GraceFullTimeout, } - return apprun.Start(appCtx, params) + + return asyncrun.AsyncRun(appCtx, params) } From 4bbc58f992d980220841639b074a9c1315c9f10d Mon Sep 17 00:00:00 2001 From: Peter Stanko Date: Sat, 4 Feb 2023 16:55:34 +0100 Subject: [PATCH 2/2] refactor: add LOG_LEVEL support --- cmd/root.go | 7 ++++++- pkg/utils/applog/logging.go | 3 ++- 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/cmd/root.go b/cmd/root.go index fc7a293..6ef3ebf 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -39,10 +39,15 @@ func init() { // Cobra supports persistent flags, which, if defined here, // will be global for your application. + defaultLogLevel := os.Getenv("LOG_LEVEL") + if defaultLogLevel == "" { + defaultLogLevel = "info" + } + rootCmd.PersistentFlags().StringVar(&cfgFile, "config-file", "", "config file (default is ./config/food.yml)") rootCmd.PersistentFlags().StringVarP(&logLevel, "log", "L", - "info", "Set log level") + defaultLogLevel, "Set log level") } // initConfig reads in config file and ENV variables if set. diff --git a/pkg/utils/applog/logging.go b/pkg/utils/applog/logging.go index a796c35..1f971f7 100644 --- a/pkg/utils/applog/logging.go +++ b/pkg/utils/applog/logging.go @@ -12,8 +12,9 @@ import ( // LogConfig logger configuration type LogConfig struct { + Level string `json:"level" env:"LOG_LEVEL,default=info"` // Dir where to store log files - Dir string `json:"dir"` + Dir string `json:"dir" env:"LOG_OUT_DIR,default=./log"` // ConsoleLoggingEnabled whether logger should use console logging ConsoleLoggingEnabled bool `json:"console_logging_enabled"` }