diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -1,31 +1,8 @@ # Introduction -Please read the **detailed tutorial** available at [Odd Jobs Tutorial](https://www.haskelltutorials.com/odd-jobs). Broadly, here is the **most common** way to get started with this library: +- [Odd Jobs home page](https://www.haskelltutorials.com/odd-jobs) - contains a description of top-level features of this library +- [Getting started & implementation guide for Odd Jobs](https://www.haskelltutorials.com/odd-jobs/guide.html) +- [Haskell Job Queues: An Ultimate Guide](https://www.haskelltutorials.com/odd-jobs/haskell-job-queues-ultimate-guide.html) - A detailed writeup on why we built Odd Jobs and how it compares against other similar libraries. +- [Start reading Hackage documentation from `OddJobs.Job`](https://hackage.haskell.org/package/odd-jobs-0.2.0/docs/OddJobs-Job.html) (**Note:** Please ensure you're reading docs for the correct version of the library) +- Open an issue on [Odd Jobs Github repo](https://github.com/saurabhnanda/odd-jobs) if you need help, or want to collaborate. -- (For a working version of the steps described below, please check the [simple-example](https://github.com/saurabhnanda/odd-jobs/tree/master/simple-example) directory in the project repo). -- Create a DB table to hold your jobs using the `OddJobs.Migrations.createJobTable` function. -- Create a sum-type to represent your job payloads, for example: - ``` - data MyJob = SendWelcomeEmail Int - | SendPasswordResetEmail Text - | SetupSampleData Int - deriving (Eq, Show, Generic, ToJSON, FromJSON) - ``` -- Ensure that you use a "tagged" JSON encoding for this data-type for best results (this is the default behaviour in Aeson). For example: - ``` - Aeson.encode (SendWelcomeEmail 10) == "{\"tag\":\"SendWelcomeEmail\", \"contents\":10}" - ``` - -- Create you "core" job-runner function: - ``` - myJobRunner :: Job -> IO () - myJobRunner Job{jobPayload} = - case jobPayload of - SendWelcomeEmail userId -> sendWelcomeEmail userId - SendPasswordResetEmail tkn -> sendPasswordResetEmail tkn - SetupSampleData userId -> sendPasswordResetEmail userId - ``` - -- Use `OddJobs.Job.createJob` and `OddJobs.Job.scheduleJob` within your app whenever you want to enqueue a job. -- Use `OddJobs.Cli` to wrap all of this together into a nice CLI that can fork itself as a background daemon. -- Hook this up to your [deployment scripts](https://www.haskelltutorials.com/odd-jobs/deployment.html) and you're good to go! diff --git a/dev/DevelMain.hs b/dev/DevelMain.hs --- a/dev/DevelMain.hs +++ b/dev/DevelMain.hs @@ -30,47 +30,48 @@ import Network.Wai.Handler.Warp (defaultSettings, runSettings, setPort) -import OddJobs.Endpoints (startApp, stopApp) +-- import OddJobs.Endpoints (startApp, stopApp) -- import qualified ElmCodeGen -- | Start or restart the server. -- newStore is from foreign-store. -- A Store holds onto some data across ghci reloads update :: IO () -update = do - mtidStore <- lookupStore tidStoreNum - case mtidStore of - -- no server running - Nothing -> do - done <- storeAction doneStore newEmptyMVar - tid <- start done - _ <- storeAction (Store tidStoreNum) (newIORef tid) - return () - -- server is already running - Just tidStore -> restartAppInNewThread tidStore - where - doneStore :: Store (MVar ()) - doneStore = Store 0 +update = undefined +-- update = do +-- mtidStore <- lookupStore tidStoreNum +-- case mtidStore of +-- -- no server running +-- Nothing -> do +-- done <- storeAction doneStore newEmptyMVar +-- tid <- start done +-- _ <- storeAction (Store tidStoreNum) (newIORef tid) +-- return () +-- -- server is already running +-- Just tidStore -> restartAppInNewThread tidStore +-- where +-- doneStore :: Store (MVar ()) +-- doneStore = Store 0 - -- shut the server down with killThread and wait for the done signal - restartAppInNewThread :: Store (IORef ThreadId) -> IO () - restartAppInNewThread tidStore = modifyStoredIORef tidStore $ \tid -> do - killThread tid - withStore doneStore takeMVar - readStore doneStore >>= start +-- -- shut the server down with killThread and wait for the done signal +-- restartAppInNewThread :: Store (IORef ThreadId) -> IO () +-- restartAppInNewThread tidStore = modifyStoredIORef tidStore $ \tid -> do +-- killThread tid +-- withStore doneStore takeMVar +-- readStore doneStore >>= start - -- | Start the server in a separate thread. - start :: MVar () -- ^ Written to when the thread is killed. - -> IO ThreadId - start done = do - -- (port, config, app) <- initialize - -- ElmCodeGen.updatea - forkIO (finally startApp - -- Note that this implies concurrency - -- between shutdownApp and the next app that is starting. - -- Normally this should be fine - (putMVar done () >> stopApp)) +-- -- | Start the server in a separate thread. +-- start :: MVar () -- ^ Written to when the thread is killed. +-- -> IO ThreadId +-- start done = do +-- -- (port, config, app) <- initialize +-- -- ElmCodeGen.updatea +-- forkIO (finally startApp +-- -- Note that this implies concurrency +-- -- between shutdownApp and the next app that is starting. +-- -- Normally this should be fine +-- (putMVar done () >> stopApp)) -- | kill the server shutdown :: IO () diff --git a/examples/OddJobsCliExample.lhs b/examples/OddJobsCliExample.lhs new file mode 100644 --- /dev/null +++ b/examples/OddJobsCliExample.lhs @@ -0,0 +1,162 @@ +=== 1. Create a table to store jobs + +In this example, our jobs table will be called `jobs_test` + +
+``` +ghci> import Datasbe.PostgreSQL.Simple (connectPostgreSQL) +ghci> import OddJobs.Migrations +ghci> conn <- connectPostgreSQL "dbname=jobs_test user=jobs_test password=jobs_test host=localhost" +ghci> createJobTable conn "jobs_test" +``` +
+ +=== 2. Create a module for your job-runner + +Ideally, this module should be compiled into a separate executable and should depend on your application's library module. Please refer to TODO for an example of how to setup your `package.yaml` (or cabal) file. + +| If you do not wish to deploy odd-jobs as an independent executable, you may embed it within your main application's executable as well. This is described in TODO. + +\begin{code} +{-# LANGUAGE DeriveAnyClass #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE LambdaCase #-} + +module OddJobsCliExample where + +import OddJobs.Job (Job(..), ConcurrencyControl(..), Config(..), throwParsePayload) +import OddJobs.ConfigBuilder (mkConfig, withConnectionPool, defaultTimedLogger, defaultLogStr, defaultJobType) +import OddJobs.Cli (defaultMain) + +-- Note: It is not necessary to use fast-logger. You can use any logging library +-- that can give you a logging function in the IO monad. +import System.Log.FastLogger(withTimedFastLogger, LogType'(..), defaultBufSize) +import System.Log.FastLogger.Date (newTimeCache, simpleTimeFormat) + +import Data.Text (Text) +import Data.Aeson as Aeson +import GHC.Generics + +-- This example is using these functions to introduce an artificial delay of a +-- few seconds in one of the jobs. Otherwise it is not really needed. +import OddJobs.Types (delaySeconds, Seconds(..)) +\end{code} + +=== 3. Set-up a Haskell type to represent your job-payload + +- Ideally, this data-type should be defined _inside_ your application's code and the module containing this type-definition should be part of the `exposed-modules` stanza. Again, please look at TODO for an example of how to setup your `package.yaml` (or cabal) file. +- To work with all the default settings provided by 'OddJobs.ConfigBuilder' this data-type should have a **"tagged" JSON serialisation,** i.e.: + + ```json + {"tag": "SendWelcomEmail", "contents": 10} + ``` + + In case your JSON payload does not conform to this structure, please look at TODO. + +- In this example, we are _blindly_ deriving `ToJSON` and `FromJSON` instances because the default behaviour of Aeson is to generate a tagged JSON as-per the example given above. + +\begin{code} +data MyJob + = SendWelcomeEmail Int + | SendPasswordResetEmail Text + | SetupSampleData Int + deriving (Eq, Show, Generic, ToJSON, FromJSON) +\end{code} + +=== 4. Write the core job-runner function + +In this example, the core job-runner function is in the `IO` monad. In all probability, you application's code will be in a custom monad, and not IO. Pleae refer to TODO, on how to work with custom monads. + +\begin{code} +myJobRunner :: Job -> IO () +myJobRunner job = do + (throwParsePayload job) >>= \case + SendWelcomeEmail userId -> do + putStrLn $ "This should call the function that actually sends the welcome email. " <> + "\nWe are purposely waiting 60 seconds before completing this job so that graceful shutdown can be demonstrated." + delaySeconds (Seconds 60) + putStrLn "60 second wait is now over..." + SendPasswordResetEmail tkn -> + putStrLn "This should call the function that actually sends the password-reset email" + SetupSampleData userId -> do + Prelude.error "User onboarding is incomplete" + putStrLn "This should call the function that actually sets up sample data in a newly registered user's account" +\end{code} + +=== 5. Write the main function using `OddJobs.Cli` + +\begin{code} +main :: IO () +main = do + defaultMain startJobMonitor + where + -- A callback-within-callback function. If the commands-line args contain a + -- `start` command, this function will be called. Once this function has + -- constructed the 'Config' (which requires setting up a logging function, + -- and a DB pool) it needs to execute the `callback` function that is passed + -- to it. + startJobMonitor callback = + + -- a utility function provided by `OddJobs.ConfigBuilder` which ensures + -- that the DB pool is gracefully destroyed upon shutdown. + withConnectionPool (Left "dbname=jobs_test user=jobs_test password=jobs_test host=localhost")$ \dbPool -> do + + -- Boilerplate code to setup a TimedFastLogger (from the fast-logger library) + tcache <- newTimeCache simpleTimeFormat + withTimedFastLogger tcache (LogFileNoRotate "oddjobs.log" defaultBufSize) $ \logger -> do + + -- Using the default string-based logging provided by + -- `OddJobs.ConfigBuilder`. If you want to actually use + -- structured-logging you'll need to define your own logging function. + let jobLogger = defaultTimedLogger logger (defaultLogStr defaultJobType) + cfg = mkConfig jobLogger "jobs" dbPool (MaxConcurrentJobs 50) myJobRunner Prelude.id + + -- Finally, executing the callback function that was passed to me... + callback cfg +\end{code} + +=== 6. Compile and start the Odd Jobs runner + +
+``` +$ stack install :exe:odd-jobs-cli +$ odd-jobs-cli start --daemonize --web-ui-basic-auth=oddjobs --web-ui-basic-password=awesome +``` +
+ + +=== 7. Enqueue some jobs from within your application's code + +
+``` +ghci> import OddJobs.Job (createJob) +ghci> import Database.PostgreSQL.Simple +ghci> conn <- connectPostgreSQL "dbname=jobs_test user=jobs_test password=jobs_test host=localhost" +ghci> createJob conn $ SendWelcomeEmail 10 +ghci> createJob conn $ SetupSampleData 10 +``` +
+ +=== 8. Check-out the awesome web UI + +Visit [http://localhost:7777](http://localhost:7777) (`username=oddjobs` / `password=awesome` as configured earlier). + +=== 9. Check-out the log file to see what Odd Jobs is doing + +
+``` +$ tail -f oddjobs.log +``` +
+ +=== 10. Finally, shutdown Odd Jobs _gracefully_ + +Please read [graceful shutdown](#graceful-shutdown) to know more. + +
+``` +$ odd-jobs-cli stop --timeout 65 +``` +
diff --git a/odd-jobs.cabal b/odd-jobs.cabal --- a/odd-jobs.cabal +++ b/odd-jobs.cabal @@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack -- --- hash: 4ca0abdf0789c7f27fe483c4da9d3dca5a1c1d05d10eb630e77c370297e77368 +-- hash: 55fcabc52c82895fc22a613b9b0536420144ff36e138ed38153b02b1babde122 name: odd-jobs -version: 0.1.0 +version: 0.2.0 synopsis: A full-featured PostgreSQL-backed job queue (with an admin UI) description: - Background jobs library for Haskell. - Extracted from production code at [Vacation Labs](https://www.vacationlabs.com). @@ -38,8 +38,8 @@ OddJobs.Endpoints OddJobs.Cli OddJobs.Types + OddJobs.ConfigBuilder other-modules: - OddJobs.Links UI Paths_odd_jobs hs-source-dirs: @@ -56,6 +56,7 @@ , fast-logger , filepath , friendly-time + , generic-deriving , hostname , lucid , monad-control @@ -68,6 +69,7 @@ , servant , servant-lucid , servant-server + , servant-static-th , string-conv , text , text-conversions @@ -77,15 +79,25 @@ , unliftio , unliftio-core , unordered-containers + , wai , warp default-language: Haskell2010 executable devel main-is: DevelMain.hs other-modules: + OddJobs.Cli + OddJobs.ConfigBuilder + OddJobs.Endpoints + OddJobs.Job + OddJobs.Migrations + OddJobs.Types + OddJobs.Web + UI Paths_odd_jobs hs-source-dirs: dev + src default-extensions: NamedFieldPuns LambdaCase TemplateHaskell ScopedTypeVariables GeneralizedNewtypeDeriving QuasiQuotes OverloadedStrings ghc-options: -Wall -fno-warn-orphans -fno-warn-unused-imports -fno-warn-dodgy-exports -Werror=missing-fields -threaded -with-rtsopts=-N -main-is DevelMain build-depends: @@ -99,6 +111,7 @@ , filepath , foreign-store , friendly-time + , generic-deriving , hostname , lucid , monad-control @@ -112,6 +125,7 @@ , servant , servant-lucid , servant-server + , servant-static-th , string-conv , text , text-conversions @@ -121,9 +135,56 @@ , unliftio , unliftio-core , unordered-containers + , wai , warp default-language: Haskell2010 +executable odd-jobs-cli-example + main-is: OddJobsCliExample.lhs + other-modules: + Paths_odd_jobs + hs-source-dirs: + examples + default-extensions: NamedFieldPuns LambdaCase TemplateHaskell ScopedTypeVariables GeneralizedNewtypeDeriving QuasiQuotes OverloadedStrings + ghc-options: -Wall -fno-warn-orphans -fno-warn-unused-imports -fno-warn-dodgy-exports -Werror=missing-fields -threaded -with-rtsopts=-N -main-is OddJobsCliExample + build-depends: + aeson + , base >=4.7 && <5 + , bytestring + , direct-daemonize + , directory + , either + , fast-logger + , filepath + , friendly-time + , generic-deriving + , hostname + , lucid + , monad-control + , monad-logger + , mtl + , odd-jobs + , optparse-applicative + , postgresql-simple + , resource-pool + , safe + , servant + , servant-lucid + , servant-server + , servant-static-th + , string-conv + , text + , text-conversions + , time + , timing-convenience + , unix + , unliftio + , unliftio-core + , unordered-containers + , wai + , warp + default-language: Haskell2010 + test-suite jobrunner type: exitcode-stdio-1.0 main-is: Test.hs @@ -147,6 +208,7 @@ , fast-logger , filepath , friendly-time + , generic-deriving , hedgehog , hostname , lifted-async @@ -165,6 +227,7 @@ , servant , servant-lucid , servant-server + , servant-static-th , string-conv , tasty , tasty-discover @@ -178,5 +241,6 @@ , unliftio , unliftio-core , unordered-containers + , wai , warp default-language: Haskell2010 diff --git a/src/OddJobs/Cli.hs b/src/OddJobs/Cli.hs --- a/src/OddJobs/Cli.hs +++ b/src/OddJobs/Cli.hs @@ -1,9 +1,12 @@ {-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE PartialTypeSignatures #-} +{-# LANGUAGE TypeOperators #-} +{-# LANGUAGE DataKinds #-} module OddJobs.Cli where import Options.Applicative as Opts import Data.Text -import OddJobs.Job (startJobRunner, Config(..), defaultLockTimeout) +import OddJobs.Job (startJobRunner, Config(..)) import System.Daemonize (DaemonOptions(..), daemonize) import System.FilePath (FilePath) import System.Posix.Process (getProcessID) @@ -13,7 +16,13 @@ import OddJobs.Types (Seconds(..), delaySeconds) import qualified System.Posix.Signals as Sig import qualified UnliftIO.Async as Async - +import qualified OddJobs.Endpoints as UI +import Servant.Server as Servant +import Servant.API +import Data.Proxy +import Data.Text.Encoding (decodeUtf8) +import Network.Wai.Handler.Warp as Warp +import Debug.Trace -- * Introduction -- @@ -31,11 +40,15 @@ -- It is __highly recommended__ that you read the following links before putting -- odd-jobs into production. -- --- * A [simple-example](https://github.com/saurabhnanda/odd-jobs/blob/master/simple-example/src/Main.hs) --- of how to use the 'defaultMain' function, that should make the callback-within-callback --- more understandable. +-- * The [getting started +-- guide](https://www.haskelltutorials.com/odd-jobs/guide.html#getting-started) +-- which will walk you through a bare-bones example of using the +-- 'defaultMain' function. It should make the callback-within-callback +-- clearer. -- --- * TODO: link-off to \"Deployment\" in the tutorial +-- * The [section on +-- deployment](https://www.haskelltutorials.com/odd-jobs/guide.html#deployment) +-- in the guide. -- * Default behaviour -- @@ -64,15 +77,16 @@ All of these resource allocations need to be bracketed so that when the job-runner exits, they may be cleaned-up gracefully. -Please take a look at @simple-example@ for how to use this function in -practice. (TODO: link-off to the example). +Please take a look at the [getting started +guide](https://www.haskelltutorials.com/odd-jobs/guide.html#getting-started) for +an example of how to use this function. -} defaultMain :: ((Config -> IO ()) -> IO ()) -- ^ A callback function that will be executed once the dameon has -- forked into the background. -> IO () defaultMain startFn = do - Args{argsCommand} <- customExecParser defaultCliParserPrefs (defaultCliInfo defaultLockTimeout) + Args{argsCommand} <- customExecParser defaultCliParserPrefs defaultCliInfo case argsCommand of Start cmdArgs -> do defaultStartCommand cmdArgs startFn @@ -95,15 +109,16 @@ -- ^ the same callback-within-callback function described in -- 'defaultMain' -> IO () -defaultStartCommand StartArgs{..} startFn = do +defaultStartCommand args@StartArgs{..} startFn = do progName <- getProgName case startDaemonize of False -> do - startFn startJobRunner + startFn coreStartupFn True -> do (Dir.doesPathExist startPidFile) >>= \case True -> do - putStrLn $ "PID file already exists. Please check if " <> progName <> " is still running in the background." <> + putStrLn $ + "PID file already exists. Please check if " <> progName <> " is still running in the background." <> " If not, you can safely delete this file and start " <> progName <> " again: " <> startPidFile Exit.exitWith (Exit.ExitFailure 1) False -> do @@ -111,8 +126,33 @@ pid <- getProcessID writeFile startPidFile (show pid) putStrLn $ "Started " <> progName <> " in background with PID=" <> show pid <> ". PID written to " <> startPidFile - startFn $ \jm -> startJobRunner jm{cfgPidFile = Just startPidFile} + startFn $ \cfg -> coreStartupFn cfg{cfgPidFile = Just startPidFile} + where + coreStartupFn cfg = do + Async.withAsync (defaultWebUi args cfg) $ \_ -> do + startJobRunner cfg + +defaultWebUi :: StartArgs + -> Config + -> IO () +defaultWebUi StartArgs{..} cfg@Config{..} = do + env <- UI.mkEnv cfg ("/" <>) + case startWebUiAuth of + Nothing -> pure () + Just AuthNone -> + let app = UI.server cfg env Prelude.id + in Warp.run startWebUiPort $ + Servant.serve (Proxy :: Proxy UI.FinalAPI) app + Just (AuthBasic u p) -> + let api = Proxy :: Proxy (BasicAuth "OddJobs Admin UI" OddJobsUser :> UI.FinalAPI) + ctx = defaultBasicAuth (u, p) :. EmptyContext + -- Now the app will receive an extra argument for OddJobsUser, + -- which we aren't really interested in. + app _ = UI.server cfg env Prelude.id + in Warp.run startWebUiPort $ + Servant.serveWithContext api ctx app + {-| Used by 'defaultMain' if 'Stop' command is issued via the CLI. Sends a @SIGINT@ signal to the process indicated by 'shutPidFile'. Waits for a maximum of 'shutTimeout' seconds (controller by @--timeout@) for the daemon to shutdown @@ -166,11 +206,8 @@ -- | The top-level command-line parser -argParser :: Seconds - -- ^ the default value for 'shutTimeout' - -> Parser Args -argParser defaultTimeout = Args - <$> (commandParser defaultTimeout) +argParser :: Parser Args +argParser = Args <$> commandParser -- ** Top-level command parser @@ -182,11 +219,10 @@ deriving (Eq, Show) -- Parser for 'argsCommand' -commandParser :: Seconds -- ^ default value for 'shutTimeout' - -> Parser Command -commandParser defaultTimeout = hsubparser +commandParser :: Parser Command +commandParser = hsubparser ( command "start" (info startParser (progDesc "start the odd-jobs runner")) <> - command "stop" (info (stopParser defaultTimeout) (progDesc "stop the odd-jobs runner")) <> + command "stop" (info stopParser (progDesc "stop the odd-jobs runner")) <> command "status" (info statusParser (progDesc "print status of all active jobs")) ) @@ -195,8 +231,13 @@ -- | @start@ command is parsed into this data-structure by 'startParser' data StartArgs = StartArgs { - -- | Switch to enable/disable the web UI (the web UI is still WIP) - startWebUiEnable :: !Bool + -- | Switch to pick the authentication mechanism for web UI. __Note:__ We + -- don't have a separate switch to enable\/disable the web UI. Picking an + -- authentication mechanism by specifying the relevant options, + -- automatically enables the web UI. Ref: 'webUiAuthParser' + startWebUiAuth :: !(Maybe WebUiAuth) + -- | Port on which the web UI will run. + , startWebUiPort :: !Int -- | You'll need to pass the @--daemonize@ switch to fork the job-runner as -- a background daemon, else it will keep running as a foreground process. , startDaemonize :: !Bool @@ -206,22 +247,51 @@ startParser :: Parser Command startParser = fmap Start $ StartArgs - <$> switch ( long "web-ui-enable" <> - help "Please look at other web-ui-* options to configure the Web UI" - ) + <$> webUiAuthParser + <*> option auto ( long "web-ui-port" <> + metavar "PORT" <> + value 7777 <> + showDefault <> + help "The port on which the Web UI listens. Please note, to actually enable the Web UI you need to pick one of the available auth schemes" + ) <*> switch ( long "daemonize" <> help "Fork the job-runner as a background daemon. If omitted, the job-runner remains in the foreground." ) <*> pidFileParser +data WebUiAuth + = AuthNone + | AuthBasic !Text !Text + deriving (Eq, Show) +-- | Pick one of the following auth mechanisms for the web UI: +-- +-- * No auth - @--web-ui-no-auth@ __NOT RECOMMENDED__ +-- * Basic auth - @--web-ui-basic-auth-user @ and +-- @--web-ui-basic-auth-password @ +webUiAuthParser :: Parser (Maybe WebUiAuth) +webUiAuthParser = basicAuthParser <|> noAuthParser <|> (pure Nothing) + where + basicAuthParser = fmap Just $ AuthBasic + <$> strOption ( long "web-ui-basic-auth-user" <> + metavar "USER" <> + help "Username for basic auth" + ) + <*> strOption ( long "web-ui-basic-auth-password" <> + metavar "PASS" <> + help "Password for basic auth" + ) + noAuthParser = flag' (Just AuthNone) + ( long "web-ui-no-auth" <> + help "Start the web UI with any authentication. NOT RECOMMENDED." + ) + -- ** Stop command -- | @stop@ command is parsed into this data-structure by 'stopParser'. Please -- note, that this command first sends a @SIGINT@ to the daemon and waits for --- 'shutTimeout' seconds (which defaults to 'defaultLockTimeout'). If the daemon --- doesn't shut down cleanly within that time, it sends a @SIGKILL@ to kill --- immediately. +-- 'shutTimeout' seconds. If the daemon doesn't shut down cleanly within that +-- time, it sends a @SIGKILL@ to kill immediately. data StopArgs = StopArgs { -- | After sending a @SIGINT@, how many seconds to wait before sending a -- @SIGKILL@ @@ -230,13 +300,13 @@ , shutPidFile :: !FilePath } deriving (Eq, Show) -stopParser :: Seconds -> Parser Command -stopParser defaultTimeout = fmap Stop $ StopArgs +stopParser :: Parser Command +stopParser = fmap Stop $ StopArgs <$> option (Seconds <$> auto) ( long "timeout" <> metavar "TIMEOUT" <> - help "Maximum seconds to wait before force-killing the background daemon." <> - value defaultTimeout <> - showDefaultWith (show . unSeconds) + help "Maximum seconds to wait before force-killing the background daemon." + -- value defaultTimeout <> + -- showDefaultWith (show . unSeconds) ) <*> pidFileParser @@ -265,11 +335,9 @@ showHelpOnError <> showHelpOnEmpty -defaultCliInfo :: Seconds - -- ^ default value for 'shutTimeout' - -> ParserInfo Args -defaultCliInfo defaultTimeout = - info ((argParser defaultTimeout) <**> helper) fullDesc +defaultCliInfo :: ParserInfo Args +defaultCliInfo = + info (argParser <**> helper) fullDesc defaultDaemonOptions :: DaemonOptions defaultDaemonOptions = DaemonOptions @@ -281,3 +349,16 @@ } +-- ** Auth implementations for the default Web UI + +-- *** Basic Auth + +data OddJobsUser = OddJobsUser !Text !Text deriving (Eq, Show) + +defaultBasicAuth :: (Text, Text) -> BasicAuthCheck OddJobsUser +defaultBasicAuth (user, pass) = BasicAuthCheck $ \b -> + let u = decodeUtf8 (basicAuthUsername b) + p = decodeUtf8 (basicAuthPassword b) + in if u==user && p==pass + then pure (Authorized $ OddJobsUser u p) + else pure BadPassword diff --git a/src/OddJobs/ConfigBuilder.hs b/src/OddJobs/ConfigBuilder.hs new file mode 100644 --- /dev/null +++ b/src/OddJobs/ConfigBuilder.hs @@ -0,0 +1,294 @@ +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE FlexibleContexts #-} + +module OddJobs.ConfigBuilder where + +import OddJobs.Types +import Database.PostgreSQL.Simple as PGS +import Data.Pool +import Control.Monad.Logger (LogLevel(..), LogStr, toLogStr) +import Data.Text (Text) +import Lucid (Html, toHtml, class_, div_, span_, br_, button_, a_, href_, onclick_) +import Data.Maybe (fromMaybe) +import Data.List as DL +import Data.Aeson as Aeson hiding (Success) +import qualified Data.Text as T +import qualified Data.HashMap.Lazy as HM +import GHC.Generics +import Data.Proxy (Proxy(..)) +import Generics.Deriving.ConNames +import Control.Monad +import Data.String.Conv +import GHC.Exts (toList) +import qualified Data.ByteString as BS +import UnliftIO (MonadUnliftIO, withRunInIO, bracket, liftIO) +import qualified System.Log.FastLogger as FLogger + + +-- | This function gives you a 'Config' with a bunch of sensible defaults +-- already applied. It requires the bare minimum configuration parameters that +-- this library cannot assume on your behalf. +-- +-- It makes a few __important assumptions__ about your 'jobPayload 'JSON, which +-- are documented in 'defaultJobType'. +mkConfig :: (LogLevel -> LogEvent -> IO ()) + -- ^ "Structured logging" function. Ref: 'cfgLogger' + -> TableName + -- ^ DB table which holds your jobs. Ref: 'cfgTableName' + -> Pool Connection + -- ^ DB connection-pool to be used by job-runner. Ref: 'cfgDbPool' + -> ConcurrencyControl + -- ^ Concurrency configuration. Ref: 'cfgConcurrencyControl' + -> (Job -> IO ()) + -- ^ The actual "job runner" which contains your application code. Ref: 'cfgJobRunner' + -> (Config -> Config) + -- ^ A function that allows you to modify the \"interim config\". The + -- \"interim config\" will cotain a bunch of in-built default config + -- params, along with the config params that you\'ve just provided + -- (i.e. logging function, table name, DB pool, etc). You can use this + -- function to override values in the \"interim config\". If you do not + -- wish to modify the \"interim config\" just pass 'Prelude.id' as an + -- argument to this parameter. __Note:__ it is strongly recommended + -- that you __do not__ modify the generated 'Config' outside of this + -- function, unless you know what you're doing. + -> Config + -- ^ The final 'Config' that can be used to start various job-runners +mkConfig logger tname dbpool ccControl jrunner configOverridesFn = + let cfg = configOverridesFn $ Config + { cfgPollingInterval = defaultPollingInterval + , cfgOnJobSuccess = (const $ pure ()) + , cfgOnJobFailed = [] + , cfgJobRunner = jrunner + , cfgLogger = logger + , cfgDbPool = dbpool + , cfgOnJobStart = (const $ pure ()) + , cfgDefaultMaxAttempts = 10 + , cfgTableName = tname + , cfgOnJobTimeout = (const $ pure ()) + , cfgConcurrencyControl = ccControl + , cfgPidFile = Nothing + , cfgJobType = defaultJobType + , cfgDefaultJobTimeout = Seconds 600 + , cfgJobToHtml = defaultJobToHtml (cfgJobType cfg) + , cfgAllJobTypes = (defaultDynamicJobTypes (cfgTableName cfg) (cfgJobTypeSql cfg)) + , cfgJobTypeSql = defaultJobTypeSql + } + in cfg + + + +-- | If you aren't interested in structured logging, you can use this function +-- to emit plain-text logs (or define your own). +defaultLogStr :: (Job -> Text) + -> LogLevel + -> LogEvent + -> LogStr +defaultLogStr jobTypeFn logLevel logEvent = + (toLogStr $ show logLevel) <> " | " <> str + where + jobToLogStr job@Job{jobId} = + "JobId=" <> (toLogStr $ show jobId) <> " JobType=" <> (toLogStr $ jobTypeFn job) + + str = case logEvent of + LogJobStart j -> + "Started | " <> jobToLogStr j + LogJobFailed j e fm t -> + let tag = case fm of + FailWithRetry -> "Failed (retry)" + FailPermanent -> "Failed (permanent)" + in tag <> " | " <> jobToLogStr j <> " | runtime=" <> (toLogStr $ show t) <> " | error=" <> (toLogStr $ show e) + LogJobSuccess j t -> + "Success | " <> (jobToLogStr j) <> " | runtime=" <> (toLogStr $ show t) + LogJobTimeout j@Job{jobLockedAt, jobLockedBy} -> + "Timeout | " <> jobToLogStr j <> " | lockedBy=" <> (toLogStr $ maybe "unknown" unJobRunnerName jobLockedBy) <> + " lockedAt=" <> (toLogStr $ maybe "unknown" show jobLockedAt) + LogPoll -> + "Polling jobs table" + LogWebUIRequest -> + "WebUIRequest (TODO: Log the actual request)" + LogText t -> + toLogStr t + +defaultJobToHtml :: (Job -> Text) + -> [Job] + -> IO [Html ()] +defaultJobToHtml jobType js = + pure $ DL.map jobToHtml js + where + jobToHtml :: Job -> Html () + jobToHtml j = do + div_ [ class_ "job" ] $ do + div_ [ class_ "job-type" ] $ do + toHtml $ jobType j + div_ [ class_ "job-payload" ] $ do + defaultPayloadToHtml $ defaultJobContent $ jobPayload j + case jobLastError j of + Nothing -> mempty + Just e -> do + div_ [ class_ "job-error collapsed" ] $ do + a_ [ href_ "javascript: void(0);", onclick_ "toggleError(this)" ] $ do + span_ [ class_ "badge badge-secondary error-expand" ] "+ Last error" + span_ [ class_ "badge badge-secondary error-collapse d-none" ] "- Last error" + " " + defaultErrorToHtml e + + +defaultErrorToHtml :: Value -> Html () +defaultErrorToHtml e = + case e of + Aeson.String s -> handleLineBreaks s + Aeson.Bool b -> toHtml $ show b + Aeson.Number n -> toHtml $ show n + Aeson.Null -> toHtml ("(null)" :: Text) + Aeson.Object o -> toHtml $ show o -- TODO: handle this properly + Aeson.Array a -> toHtml $ show a -- TODO: handle this properly + where + handleLineBreaks s = do + forM_ (T.splitOn "\n" s) $ \x -> do + toHtml x + br_ [] + +defaultJobContent :: Value -> Value +defaultJobContent v = case v of + Aeson.Object o -> case HM.lookup "contents" o of + Nothing -> v + Just c -> c + _ -> v + +defaultPayloadToHtml :: Value -> Html () +defaultPayloadToHtml v = case v of + Aeson.Object o -> do + toHtml ("{ " :: Text) + forM_ (HM.toList o) $ \(k, v2) -> do + span_ [ class_ " key-value-pair " ] $ do + span_ [ class_ "key" ] $ toHtml $ k <> ":" + span_ [ class_ "value" ] $ defaultPayloadToHtml v2 + toHtml (" }" :: Text) + Aeson.Array a -> do + toHtml ("[" :: Text) + forM_ (toList a) $ \x -> do + defaultPayloadToHtml x + toHtml (", " :: Text) + toHtml ("]" :: Text) + Aeson.String t -> toHtml t + Aeson.Number n -> toHtml $ show n + Aeson.Bool b -> toHtml $ show b + Aeson.Null -> toHtml ("null" :: Text) + +defaultJobTypeSql :: PGS.Query +defaultJobTypeSql = "payload->>'tag'" + +defaultConstantJobTypes :: forall a . (Generic a, ConNames (Rep a)) + => Proxy a + -> AllJobTypes +defaultConstantJobTypes _ = + AJTFixed $ DL.map toS $ conNames (undefined :: a) + +defaultDynamicJobTypes :: TableName + -> PGS.Query + -> AllJobTypes +defaultDynamicJobTypes tname jobTypeSql = AJTSql $ \conn -> do + fmap (DL.map ((fromMaybe "(unknown)") . fromOnly)) $ PGS.query_ conn $ "select distinct(" <> jobTypeSql <> ") from " <> tname <> " order by 1 nulls last" + +-- | This makes __two important assumptions__. First, this /assumes/ that jobs +-- in your app are represented by a sum-type. For example: +-- +-- @ +-- data MyJob = SendWelcomeEmail Int +-- | SendPasswordResetEmail Text +-- | SetupSampleData Int +-- @ +-- +-- Second, it /assumes/ that the JSON representatin of this sum-type is +-- "tagged". For example, the following... +-- +-- > let pload = SendWelcomeEmail 10 +-- +-- ...when converted to JSON, would look like... +-- +-- > {"tag":"SendWelcomeEmail", "contents":10} +-- +-- It uses this assumption to extract the "job type" from a 'Data.Aeson.Value' +-- (which would be @SendWelcomeEmail@ in the example given above). This is used +-- in logging and the admin UI. +-- +-- Even if tihs assumption is violated, the job-runner /should/ continue to +-- function. It's just that you won't get very useful log messages. +-- +-- __Note:__ If your job payload does not conform to the structure described +-- above, please read the section on [customising the job payload's +-- structure](https://www.haskelltutorials.com/odd-jobs/guide.html#custom-payload-structure) +-- in the implementation guide. +defaultJobType :: Job -> Text +defaultJobType Job{jobPayload} = + case jobPayload of + Aeson.Object hm -> case HM.lookup "tag" hm of + Just (Aeson.String t) -> t + _ -> "unknown" + _ -> "unknown" + + +-- | As the name says. Ref: 'cfgPollingInterval' +defaultPollingInterval :: Seconds +defaultPollingInterval = Seconds 5 + +-- | Convenience function to create a DB connection-pool with some sensible +-- defaults. Please see the source-code of this function to understand what it's +-- doing. +withConnectionPool :: (MonadUnliftIO m) + => Either BS.ByteString PGS.ConnectInfo + -> (Pool PGS.Connection -> m a) + -> m a +withConnectionPool connConfig action = withRunInIO $ \runInIO -> do + bracket poolCreator destroyAllResources (runInIO . action) + where + poolCreator = liftIO $ + case connConfig of + Left connString -> + createPool (PGS.connectPostgreSQL connString) PGS.close 1 (fromIntegral $ 2 * (unSeconds defaultPollingInterval)) 8 + Right connInfo -> + createPool (PGS.connect connInfo) PGS.close 1 (fromIntegral $ 2 * (unSeconds defaultPollingInterval)) 8 + +-- | A convenience function to help you define a timed-logger with some sensible +-- defaults. +defaultTimedLogger :: FLogger.TimedFastLogger + -> (LogLevel -> LogEvent -> LogStr) + -> LogLevel + -> LogEvent + -> IO () +defaultTimedLogger logger logStrFn logLevel logEvent = + if logLevel == LevelDebug + then pure () + else logger $ \t -> (toLogStr t) <> " | " <> + (logStrFn logLevel logEvent) <> + "\n" + + +defaultJsonLogEvent :: LogEvent -> Aeson.Value +defaultJsonLogEvent logEvent = + case logEvent of + LogJobStart job -> + Aeson.object [ "tag" Aeson..= ("LogJobStart" :: Text) + , "contents" Aeson..= (defaultJsonJob job) ] + LogJobSuccess job runTime -> + Aeson.object [ "tag" Aeson..= ("LogJobSuccess" :: Text) + , "contents" Aeson..= (defaultJsonJob job, runTime) ] + LogJobFailed job e fm runTime -> + Aeson.object [ "tag" Aeson..= ("LogJobFailed" :: Text) + , "contents" Aeson..= (defaultJsonJob job, show e, defaultJsonFailureMode fm, runTime) ] + LogJobTimeout job -> + Aeson.object [ "tag" Aeson..= ("LogJobTimeout" :: Text) + , "contents" Aeson..= (defaultJsonJob job) ] + LogPoll -> + Aeson.object [ "tag" Aeson..= ("LogJobPoll" :: Text)] + LogWebUIRequest -> + Aeson.object [ "tag" Aeson..= ("LogWebUIRequest" :: Text)] + LogText t -> + Aeson.object [ "tag" Aeson..= ("LogText" :: Text) + , "contents" Aeson..= t ] + +defaultJsonJob :: Job -> Aeson.Value +defaultJsonJob job = genericToJSON Aeson.defaultOptions job + +defaultJsonFailureMode :: FailureMode -> Aeson.Value +defaultJsonFailureMode fm = genericToJSON Aeson.defaultOptions fm diff --git a/src/OddJobs/Endpoints.hs b/src/OddJobs/Endpoints.hs --- a/src/OddJobs/Endpoints.hs +++ b/src/OddJobs/Endpoints.hs @@ -1,8 +1,12 @@ -{-# LANGUAGE TypeOperators, DeriveGeneric, NamedFieldPuns, DataKinds, StandaloneDeriving, FlexibleContexts #-} +{-# LANGUAGE TypeOperators, DeriveGeneric, NamedFieldPuns, DataKinds, StandaloneDeriving, FlexibleContexts, RecordWildCards, RankNTypes #-} + +-- | TODO: Rename this to OddJobs.Servant + module OddJobs.Endpoints where -import OddJobs.Web as Web +import OddJobs.Web as Web hiding (Routes(..)) +import qualified OddJobs.Web as Web import OddJobs.Job as Job import OddJobs.Types import GHC.Generics @@ -25,20 +29,22 @@ import Data.String.Conv (toS) import Control.Monad.Except import Data.Time as Time -import Data.Time.Format.Human (humanReadableTime') import Data.Aeson as Aeson import qualified Data.HashMap.Strict as HM import GHC.Exts (toList) -import Data.Maybe (fromMaybe) -import Data.Text.Conversions (fromText, toText) +import Data.Maybe (fromMaybe, mapMaybe) import Control.Applicative ((<|>)) -import Data.Time.Convenience (timeSince, Unit(..), Direction(..)) -import qualified OddJobs.Links as Links +-- import qualified OddJobs.Links as Links import Data.List ((\\)) import qualified System.Log.FastLogger as FLogger import qualified System.Log.FastLogger.Date as FLogger import Control.Monad.Logger as MLogger import qualified Data.ByteString.Lazy as BSL +import qualified Data.List as DL +import UnliftIO.IORef +import Debug.Trace +import qualified OddJobs.ConfigBuilder as Builder +import Servant.Static.TH (createApiAndServerDecs) -- startApp :: IO () -- startApp = undefined @@ -46,410 +52,160 @@ -- stopApp :: IO () -- stopApp = undefined -tname :: TableName -tname = "jobs_aqgrqtaowi" +$(createApiAndServerDecs "StaticAssetRoutes" "staticAssetServer" "assets") -startApp :: IO () -startApp = do - let connInfo = ConnectInfo - { connectHost = "localhost" - , connectPort = fromIntegral 5432 - , connectUser = "jobs_test" - , connectPassword = "jobs_test" - , connectDatabase = "jobs_test" - } - dbPool <- createPool - (PGS.connect connInfo) -- cretea a new resource - (PGS.close) -- destroy resource - 1 -- stripes - (fromRational 10) -- number of seconds unused resources are kept around - 5 -- maximum open connections +data Routes route = Routes + { rFilterResults :: route :- QueryParam "filters" Web.Filter :> Get '[HTML] (Html ()) + , rEnqueue :: route :- "enqueue" :> Capture "jobId" JobId :> Post '[HTML] NoContent + , rRunNow :: route :- "run" :> Capture "jobId" JobId :> Post '[HTML] NoContent + , rCancel :: route :- "cancel" :> Capture "jobId" JobId :> Post '[HTML] NoContent + , rRefreshJobTypes :: route :- "refresh-job-types" :> Post '[HTML] NoContent + , rRefreshJobRunners :: route :- "refresh-job-runners" :> Post '[HTML] NoContent + } deriving (Generic) - tcache <- FLogger.newTimeCache FLogger.simpleTimeFormat' - (tlogger, cleanup) <- FLogger.newTimedFastLogger tcache (FLogger.LogStdout FLogger.defaultBufSize) - let flogger = Job.defaultTimedLogger tlogger (Job.defaultLogStr (Job.defaultJobToText Job.defaultJobType)) - jm = Job.defaultConfig flogger tname dbPool Job.UnlimitedConcurrentJobs (const $ pure ()) - let nt :: ReaderT Job.Config IO a -> Servant.Handler a - nt action = (liftIO $ try $ runReaderT action jm) >>= \case - Left (e :: SomeException) -> Servant.Handler $ ExceptT $ pure $ Left $ err500 { errBody = toS $ show e } - Right a -> Servant.Handler $ ExceptT $ pure $ Right a - appProxy = (Proxy :: Proxy (ToServant Routes AsApi)) - - finally - (run 8080 $ genericServe (server dbPool)) - (cleanup >> (Pool.destroyAllResources dbPool)) - -stopApp :: IO () -stopApp = pure () +type FinalAPI = + (ToServant Routes AsApi) :<|> + "assets" :> StaticAssetRoutes -server :: Pool Connection - -> Routes AsServer -server dbPool = Routes - { rFilterResults = (\mFilter -> filterResults dbPool mFilter) - , rStaticAssets = serveDirectoryFileServer "assets" +data Env = Env + { envRoutes :: Web.Routes + , envJobTypesRef :: IORef [Text] + , envJobRunnersRef :: IORef [JobRunnerName] } - --- withDbConnection :: HasJobMonitor m --- => (Connection -> m a) --- -> m a --- withDbConnection fn = getDbPool >>= \pool -> Pool.withResource pool fn - -filterResults :: Pool Connection - -> Maybe Filter - -> Handler (Html ()) -filterResults dbPool mFilter = do - let filters = fromMaybe mempty mFilter - (jobs, runningCount) <- liftIO $ Pool.withResource dbPool $ \conn -> (,) - <$> (filterJobs conn tname filters) - <*> (countJobs conn tname filters{ filterStatuses = [Job.Locked] }) - t <- liftIO getCurrentTime - pure $ pageLayout $ do - searchBar t filters - resultsPanel t filters jobs runningCount - +mkEnv :: (MonadIO m) => Config -> (Text -> Text) -> m Env +mkEnv cfg@Config{..} linksFn = do + allJobTypes <- fetchAllJobTypes cfg + allJobRunners <- fetchAllJobRunners cfg + envJobTypesRef <- newIORef allJobTypes + envJobRunnersRef <- newIORef allJobRunners + let envRoutes = routes linksFn -pageNav :: Html () -pageNav = do - div_ $ nav_ [ class_ "navbar navbar-default navigation-clean" ] $ div_ [ class_ "container" ] $ do - div_ [ class_ "navbar-header" ] $ do - a_ [ class_ "navbar-brand navbar-link", href_ "#", style_ "padding: 0px;" ] $ img_ [ src_ "/assets/odd-jobs-color-logo.png", title_ "Odd Jobs Logo" ] - button_ [ class_ "navbar-toggle collapsed", data_ "toggle" "collapse", data_ "target" "#navcol-1" ] $ do - span_ [ class_ "sr-only" ] $ "Toggle navigation" - span_ [ class_ "icon-bar" ] $ "" - span_ [ class_ "icon-bar" ] $ "" - span_ [ class_ "icon-bar" ] $ "" - -- div_ [ class_ "collapse navbar-collapse", id_ "navcol-1" ] $ ul_ [ class_ "nav navbar-nav navbar-right" ] $ do - -- li_ [ class_ "active", role_ "presentation" ] $ a_ [ href_ "#" ] $ "First Item" - -- li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "Second Item" - -- li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "Third Item" - -- li_ [ class_ "dropdown" ] $ do - -- a_ [ class_ "dropdown-toggle", data_ "toggle" "dropdown", ariaExpanded_ "false", href_ "#" ] $ do - -- "Dropdown" - -- span_ [ class_ "caret" ] $ "" - -- ul_ [ class_ "dropdown-menu", role_ "menu" ] $ do - -- li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "First Item" - -- li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "Second Item" - -- li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "Third Item" + pure Env{..} + -- TODO: remove hard-coded port + -- run 8080 $ genericServe (server cfg Env{..}) -pageLayout :: Html () -> Html () -pageLayout inner = do - doctype_ - html_ $ do - head_ $ do - meta_ [ charset_ "utf-8" ] - meta_ [ name_ "viewport", content_ "width=device-width, initial-scale=1.0" ] - title_ "haskell-pg-queue" - link_ [ rel_ "stylesheet", href_ "assets/bootstrap/css/bootstrap.min.css" ] - link_ [ rel_ "stylesheet", href_ "https://fonts.googleapis.com/css?family=Lato:100i,300,300i,400,700,900" ] - link_ [ rel_ "stylesheet", href_ "assets/css/logo-slider.css" ] - link_ [ rel_ "stylesheet", href_ "assets/css/Navigation-Clean1.css" ] - link_ [ rel_ "stylesheet", href_ "assets/css/styles.css" ] - body_ $ do - pageNav - div_ $ div_ [ class_ "container", style_ "/*background-color:#f2f2f2;*/" ] $ div_ [ class_ "row" ] $ div_ [ class_ "col-md-12" ] $ do - inner - script_ [ src_ "assets/js/jquery.min.js" ] $ ("" :: Text) - script_ [ src_ "assets/bootstrap/js/bootstrap.min.js" ] $ ("" :: Text) - script_ [ src_ "https://cdnjs.cloudflare.com/ajax/libs/slick-carousel/1.6.0/slick.js" ] $ ("" :: Text) - script_ [ src_ "assets/js/logo-slider.js" ] $ ("" :: Text) + -- let nt :: ReaderT Job.Config IO a -> Servant.Handler a + -- nt action = (liftIO $ try $ runReaderT action jm) >>= \case + -- Left (e :: SomeException) -> Servant.Handler $ ExceptT $ pure $ Left $ err500 { errBody = toS $ show e } + -- Right a -> Servant.Handler $ ExceptT $ pure $ Right a + -- appProxy = (Proxy :: Proxy (ToServant Routes AsApi)) -searchBar :: UTCTime -> Filter -> Html () -searchBar t filter@Filter{filterStatuses, filterCreatedAfter, filterCreatedBefore, filterUpdatedAfter, filterUpdatedBefore, filterJobTypes, filterRunAfter} = do - form_ [ style_ "padding-top: 2em;" ] $ do - div_ [ class_ "form-group" ] $ do - div_ [ class_ "search-container" ] $ do - ul_ [ class_ "list-inline search-bar" ] $ do - forM_ filterStatuses $ \s -> renderFilter "Status" (toText s) (Links.rFilterResults $ Just filter{filterStatuses = filterStatuses \\ [s]}) - maybe mempty (\x -> renderFilter "Created after" (showText x) (Links.rFilterResults $ Just filter{filterCreatedAfter = Nothing})) filterCreatedAfter - maybe mempty (\x -> renderFilter "Created before" (showText x) (Links.rFilterResults $ Just filter{filterCreatedBefore = Nothing})) filterCreatedBefore - maybe mempty (\x -> renderFilter "Updated after" (showText x) (Links.rFilterResults $ Just filter{filterUpdatedAfter = Nothing})) filterUpdatedAfter - maybe mempty (\x -> renderFilter "Updated before" (showText x) (Links.rFilterResults $ Just filter{filterUpdatedBefore = Nothing})) filterUpdatedBefore - maybe mempty (\x -> renderFilter "Run after" (showText x) (Links.rFilterResults $ Just filter{filterRunAfter = Nothing})) filterRunAfter - forM_ filterJobTypes $ \x -> renderFilter "Job type" x (Links.rFilterResults $ Just filter{filterJobTypes = filterJobTypes \\ [x]}) + -- finally + -- (run 8080 $ genericServe (server jm dbPool jobTypesRef jobRunnerRef)) + -- (cleanup >> (Pool.destroyAllResources dbPool)) - button_ [ class_ "btn btn-default search-button", type_ "button" ] $ "Search" - ul_ [ class_ "list-inline" ] $ do - li_ $ span_ $ strong_ "Common searches:" - li_ $ a_ [ href_ (Links.rFilterResults $ Just mempty) ] $ "All jobs" - li_ $ a_ [ href_ (Links.rFilterResults $ Just $ filter{ filterStatuses = [Job.Locked] }) ] $ "Currently running" - li_ $ a_ [ href_ (Links.rFilterResults $ Just $ filter{ filterStatuses = [Job.Success] }) ] $ "Successful" - li_ $ a_ [ href_ (Links.rFilterResults $ Just $ filter{ filterStatuses = [Job.Failed] }) ] $ "Failed" - li_ $ a_ [ href_ (Links.rFilterResults $ Just $ filter{ filterRunAfter = Just t }) ] $ "Future" - -- li_ $ a_ [ href_ "#" ] $ "Retried" - li_ $ a_ [ href_ (Links.rFilterResults $ Just $ filter{ filterStatuses = [Job.Queued] }) ] $ "Queued" - li_ $ a_ [ href_ (Links.rFilterResults $ Just $ filter{ filterUpdatedAfter = Just $ timeSince t 10 Minutes Ago }) ] $ "Last 10 mins" - li_ $ a_ [ href_ (Links.rFilterResults $ Just $ filter{ filterCreatedAfter = Just $ timeSince t 10 Minutes Ago }) ] $ "Recently created" - where - renderFilter :: Text -> Text -> Text -> Html () - renderFilter k v u = do - li_ [ class_ "search-filter" ] $ do - span_ [ class_ "filter-name" ] $ toHtml k - span_ [ class_ "filter-value" ] $ do - toHtml v - a_ [ href_ u, class_ "text-danger" ] $ i_ [ class_ "glyphicon glyphicon-remove" ] $ "" +stopApp :: IO () +stopApp = pure () -timeDuration :: UTCTime -> UTCTime -> (Int, String) -timeDuration from to = (diff, str) +server :: forall m . (MonadIO m) + => Config + -> Env + -> (forall a . Handler a -> m a) + -> ServerT FinalAPI m +server cfg env nt = + (toServant routeServer) :<|> staticAssetServer where - str = if diff <= 0 - then "under 1s" - else (if d>0 then (show d) <> "d" else "") <> - (if m>0 then (show m) <> "m" else "") <> - (if s>0 then (show s) <> "s" else "") - diff = (abs $ round $ diffUTCTime from to) - (m', s) = diff `divMod` 60 - (h', m) = m' `divMod` 60 - (d, h) = h' `divMod` 24 - -showText :: (Show a) => a -> Text -showText a = toS $ show a - -jobContent :: Value -> Value -jobContent v = case v of - Aeson.Object o -> case HM.lookup "contents" o of - Nothing -> v - Just c -> c - _ -> v + routeServer :: Routes (AsServerT m) + routeServer = Routes + { rFilterResults = nt . (filterResults cfg env) + , rEnqueue = nt . (enqueueJob cfg env) + , rCancel = nt . (cancelJob cfg env) + , rRunNow = nt . (runJobNow cfg env) + , rRefreshJobTypes = nt $ refreshJobTypes cfg env + , rRefreshJobRunners = nt $ refreshJobRunners cfg env + } -rowSuccess :: UTCTime -> Job -> Html () -rowSuccess t job@Job{jobStatus, jobCreatedAt, jobUpdatedAt, jobPayload, jobAttempts, jobRunAt} = do - tr_ $ do - td_ [ class_ "job-type" ] $ case jobStatus of - Job.Success -> statusSuccess - Job.Failed -> statusFailed - Job.Queued -> if jobRunAt > t - then statusFuture - else statusWaiting - Job.Retry -> statusRetry - Job.Locked -> statusLocked +server2 :: Config + -> Env + -> Routes AsServer +server2 cfg env = Routes + { rFilterResults = (filterResults cfg env) + , rEnqueue = (enqueueJob cfg env) + , rCancel = (cancelJob cfg env) + , rRunNow = (runJobNow cfg env) + , rRefreshJobTypes = refreshJobTypes cfg env + , rRefreshJobRunners = refreshJobRunners cfg env + } - -- span_ [ class_ "label label-success" ] $ "Success" - -- span_ [ class_ "job-run-time" ] $ "Completed 23h ago. Took 3 sec." - - td_ $ toHtml $ Job.defaultJobType job -- TODO: this needs to be changed - td_ $ div_ [ class_ "job-payload" ] $ payloadToHtml $ jobContent jobPayload - -- span_ [ class_ "key-value-pair" ] $ do - -- span_ [ class_ "key" ] $ "args" - -- span_ [ class_ "value" ] $ do - -- "[\"flexi_payment_reminder\", 3432423," - -- a_ [ href_ "#", class_ "json-ellipsis" ] $ "\8230" - -- "]" - td_ "Text" - td_ $ case jobStatus of - Job.Success -> mempty - Job.Failed -> actionsFailed - Job.Queued -> if jobRunAt > t - then actionsFuture - else mempty - Job.Retry -> actionsRetry - Job.Locked -> mempty - where - - actionsFailed = do - button_ [ class_ "btn btn-default", type_ "button" ] $ "Retry again" - - actionsRetry = do - button_ [ class_ "btn btn-default", type_ "button" ] $ "Retry now" - - actionsFuture = do - button_ [ class_ "btn btn-default", type_ "button" ] $ "Run now" - - payloadToHtml :: Value -> Html () - payloadToHtml v = case v of - Aeson.Object o -> do - toHtml ("{ " :: Text) - forM_ (HM.toList o) $ \(k, v) -> do - span_ [ class_ " key-value-pair " ] $ do - span_ [ class_ "key" ] $ toHtml $ k <> ":" - span_ [ class_ "value" ] $ payloadToHtml v - toHtml (" }" :: Text) - Aeson.Array a -> do - toHtml ("[" :: Text) - forM_ (toList a) $ \x -> do - payloadToHtml x - toHtml (", " :: Text) - toHtml ("]" :: Text) - Aeson.String t -> toHtml t - Aeson.Number n -> toHtml $ show n - Aeson.Bool b -> toHtml $ show b - Aeson.Null -> toHtml ("null" :: Text) - - statusSuccess = do - span_ [ class_ "label label-success" ] $ "Success" - span_ [ class_ "job-run-time" ] $ do - let (d, s) = timeDuration jobCreatedAt jobUpdatedAt - abbr_ [ title_ (showText jobUpdatedAt) ] $ toHtml $ "Completed " <> humanReadableTime' t jobUpdatedAt <> ". " - abbr_ [ title_ (showText d <> " seconds")] $ toHtml $ "Took " <> s - - statusFailed = do - span_ [ class_ "label label-danger" ] $ "Failed" - span_ [ class_ "job-run-time" ] $ do - abbr_ [ title_ (showText jobUpdatedAt) ] $ toHtml $ "Failed " <> humanReadableTime' t jobUpdatedAt <> " after " <> show jobAttempts <> " attempts" - - statusFuture = do - span_ [ class_ "label" ] $ "Future" - span_ [ class_ "job-run-time" ] $ do - abbr_ [ title_ (showText jobRunAt) ] $ toHtml $ humanReadableTime' t jobRunAt - - statusWaiting = do - span_ [ class_ "label label-warning" ] $ "Waiting" - -- span_ [ class_ "job-run-time" ] ("Waiting to be picked up" :: Text) - - statusRetry = do - span_ [ class_ "label label-info" ] $ toHtml $ "Retries (" <> show jobAttempts <> ")" - span_ [ class_ "job-run-time" ] $ do - abbr_ [ title_ (showText jobUpdatedAt) ] $ toHtml $ "Retried " <> humanReadableTime' t jobUpdatedAt <> ". " - abbr_ [ title_ (showText jobRunAt)] $ toHtml $ "Next retry in " <> humanReadableTime' t jobRunAt - - statusLocked = do - span_ [ class_ "label label-warning" ] $ toHtml ("Locked" :: Text) - -- span_ [ class_ "job-run-time" ] $ do - -- abbr_ [ title_ (showText jobUpdatedAt) ] $ toHtml $ "Retried " <> humanReadableTime' t jobUpdatedAt <> ". " - -- abbr_ [ title_ (showText jobRunAt)] $ toHtml $ "Next retry in " <> humanReadableTime' t jobRunAt +refreshJobRunners :: Config + -> Env + -> Handler NoContent +refreshJobRunners cfg@Config{..} Env{envRoutes=Web.Routes{..}, envJobRunnersRef} = do + allJobRunners <- fetchAllJobRunners cfg + atomicModifyIORef' envJobRunnersRef (\_ -> (allJobRunners, ())) + throwError $ err302{errHeaders=[("Location", toS $ rFilterResults Nothing)]} +refreshJobTypes :: Config + -> Env + -> Handler NoContent +refreshJobTypes cfg Env{envRoutes=Web.Routes{..}, envJobTypesRef} = do + allJobTypes <- fetchAllJobTypes cfg + atomicModifyIORef' envJobTypesRef (\_ -> (allJobTypes, ())) + throwError $ err302{errHeaders=[("Location", toS $ rFilterResults Nothing)]} -rowRetry :: Html () -rowRetry = do - tr_ $ do - td_ [ class_ "job-type" ] $ do - span_ [ class_ "label label-info" ] $ "Retried (5)" - span_ [ class_ "job-run-time" ] $ "23 mins ago. Next retry in 90 min." - td_ "Queued Mail" - td_ $ div_ [ class_ "job-payload" ] $ do - span_ [ class_ "key-value-pair" ] $ do - span_ [ class_ "key" ] $ "client_id" - span_ [ class_ "value" ] $ "456" - span_ [ class_ "key-value-pair" ] $ do - span_ [ class_ "key" ] $ "user_id" - span_ [ class_ "value" ] $ "123" - span_ [ class_ "key-value-pair" ] $ do - span_ [ class_ "key" ] $ "args" - span_ [ class_ "value" ] $ do - "[\"flexi_payment_reminder\", 3432423," - a_ [ href_ "#", class_ "json-ellipsis" ] $ "\8230" - "]" - td_ "Text" - td_ $ div_ [ class_ "btn-group" ] $ do - button_ [ class_ "btn btn-default", type_ "button" ] $ "Retry now" - button_ [ class_ "btn btn-default dropdown-toggle", data_ "toggle" "dropdown", ariaExpanded_ "false", type_ "button" ] $ span_ [ class_ "caret" ] $ "" - ul_ [ class_ "dropdown-menu", role_ "menu" ] $ do - li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "First Item" - li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "Second Item" - li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "Third Item" +cancelJob :: Config + -> Env + -> JobId + -> Handler NoContent +cancelJob Config{..} env jid = do + liftIO $ withResource cfgDbPool $ \conn -> void $ cancelJobIO conn cfgTableName jid + redirectToHome env +runJobNow :: Config + -> Env + -> JobId + -> Handler NoContent +runJobNow Config{..} env jid = do + liftIO $ withResource cfgDbPool $ \conn -> void $ runJobNowIO conn cfgTableName jid + redirectToHome env -rowFailed :: Html () -rowFailed = do - tr_ $ do - td_ [ class_ "job-type" ] $ do - span_ [ class_ "label label-danger" ] $ "Failed" - span_ [ class_ "job-run-time" ] $ "23 mins ago. After 25 attempts." - td_ "Queued Mail" - td_ $ div_ [ class_ "job-payload" ] $ do - span_ [ class_ "key-value-pair" ] $ do - span_ [ class_ "key" ] $ "client_id" - span_ [ class_ "value" ] $ "456" - span_ [ class_ "key-value-pair" ] $ do - span_ [ class_ "key" ] $ "user_id" - span_ [ class_ "value" ] $ "123" - span_ [ class_ "key-value-pair" ] $ do - span_ [ class_ "key" ] $ "args" - span_ [ class_ "value" ] $ do - "[\"flexi_payment_reminder\", 3432423," - a_ [ href_ "#", class_ "json-ellipsis" ] $ "\8230" - "]" - td_ "Text" - td_ $ button_ [ class_ "btn btn-default", type_ "button" ] $ "Retry again" +enqueueJob :: Config + -> Env + -> JobId + -> Handler NoContent +enqueueJob Config{..} env jid = do + liftIO $ withResource cfgDbPool $ \conn -> do + void $ unlockJobIO conn cfgTableName jid + void $ runJobNowIO conn cfgTableName jid + redirectToHome env -rowFuture :: Html () -rowFuture = do - tr_ $ do - td_ [ class_ "job-type" ] $ do - span_ [ class_ "label label-default" ] $ "Future" - span_ [ class_ "job-run-time" ] $ "37 mins from now" - td_ "Queued Mail" - td_ $ div_ [ class_ "job-payload" ] $ do - span_ [ class_ "key-value-pair" ] $ do - span_ [ class_ "key" ] $ "client_id" - span_ [ class_ "value" ] $ "456" - span_ [ class_ "key-value-pair" ] $ do - span_ [ class_ "key" ] $ "user_id" - span_ [ class_ "value" ] $ "123" - span_ [ class_ "key-value-pair" ] $ do - span_ [ class_ "key" ] $ "args" - span_ [ class_ "value" ] $ do - "[\"flexi_payment_reminder\", 3432423," - a_ [ href_ "#", class_ "json-ellipsis" ] $ "\8230" - "]" - td_ "Text" - td_ $ div_ [ class_ "btn-group" ] $ do - button_ [ class_ "btn btn-default", type_ "button" ] $ "Run now" - button_ [ class_ "btn btn-default dropdown-toggle", data_ "toggle" "dropdown", ariaExpanded_ "false", type_ "button" ] $ span_ [ class_ "caret" ] $ "" - ul_ [ class_ "dropdown-menu", role_ "menu" ] $ do - li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "First Item" - li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "Second Item" - li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "Third Item" +redirectToHome :: Env -> Handler NoContent +redirectToHome Env{envRoutes=Web.Routes{..}} = do + throwError $ err301{errHeaders=[("Location", toS $ rFilterResults Nothing)]} -rowLocked :: Html () -rowLocked = do - tr_ $ do - td_ [ class_ "job-type" ] $ do - span_ [ class_ "label label-warning" ] $ "Locked" - span_ [ class_ "job-run-time" ] $ "Since 2min by" - span_ [ class_ "job-runner-name" ] $ "hostname:3242" - td_ "Queued Mail" - td_ $ div_ [ class_ "job-payload" ] $ do - span_ [ class_ "key-value-pair" ] $ do - span_ [ class_ "key" ] $ "client_id" - span_ [ class_ "value" ] $ "456" - span_ [ class_ "key-value-pair" ] $ do - span_ [ class_ "key" ] $ "user_id" - span_ [ class_ "value" ] $ "123" - span_ [ class_ "key-value-pair" ] $ do - span_ [ class_ "key" ] $ "args" - span_ [ class_ "value" ] $ do - "[\"flexi_payment_reminder\", 3432423," - a_ [ href_ "#", class_ "json-ellipsis" ] $ "\8230" - "]" - td_ "Text" - td_ $ button_ [ class_ "btn btn-default", type_ "button" ] $ "Unlock" - -resultsPanel :: UTCTime -> Filter -> [Job] -> Int -> Html () -resultsPanel t filter@Filter{filterPage} jobs runningCount = do - div_ [ class_ "panel panel-default" ] $ do - div_ [ class_ "panel-heading" ] $ h3_ [ class_ "panel-title" ] $ do - "Currently running " - span_ [ class_ "badge" ] $ toHtml (show runningCount) - div_ [ class_ "panel-body" ] $ div_ [ class_ "currently-running" ] $ div_ [ class_ "table-responsive" ] $ table_ [ class_ "table" ] $ do - thead_ $ tr_ $ do - th_ "Job status" - th_ "Job type" - th_ "Job payload" - th_ "Last error" - th_ "Actions" - tbody_ $ do - forM_ jobs $ \j -> case jobStatus j of - Job.Success -> rowSuccess t j - _ -> rowSuccess t j - -- rowLocked - -- rowSuccess - -- rowFuture - -- rowRetry - -- rowFailed - div_ [ class_ "panel-footer" ] $ nav_ $ ul_ [ class_ "pager", style_ "margin:0px;" ] $ do - li_ [ class_ "previous" ] $ case filterPage of - Nothing -> a_ [ disabled_ "disabled" ] $ "Prev" - Just (l, 0) -> a_ [ disabled_ "disabled" ] $ "Prev" - Just (l, o) -> a_ [ href_ (Links.rFilterResults $ Just $ filter {filterPage = Just (l, max 0 $ o - l)}) ] $ "Prev" - li_ [ class_ "next" ] $ case filterPage of - Nothing -> a_ [ href_ (Links.rFilterResults $ Just $ filter {filterPage = Just (10, 10)}) ] $ "Next" - Just (l, o) -> a_ [ href_ (Links.rFilterResults $ Just $ filter {filterPage = Just (l, o + l)}) ] $ "Next" +filterResults :: Config + -> Env + -> Maybe Filter + -> Handler (Html ()) +filterResults cfg@Config{cfgJobToHtml, cfgDbPool} Env{..} mFilter = do + let filters = fromMaybe mempty mFilter + (jobs, runningCount) <- liftIO $ Pool.withResource cfgDbPool $ \conn -> (,) + <$> (filterJobs cfg conn filters) + <*> (countJobs cfg conn filters{ filterStatuses = [Job.Locked] }) + t <- liftIO getCurrentTime + js <- liftIO $ fmap (DL.zip jobs) $ cfgJobToHtml jobs + allJobTypes <- readIORef envJobTypesRef + let navHtml = Web.sideNav envRoutes allJobTypes [] t filters + bodyHtml = Web.resultsPanel envRoutes t filters js runningCount + pure $ Web.pageLayout envRoutes navHtml bodyHtml +routes :: (Text -> Text) -> Web.Routes +routes linkFn = Web.Routes + { Web.rFilterResults = rFilterResults + , Web.rEnqueue = rEnqueue + , Web.rRunNow = rRunNow + , Web.rCancel = rCancel + , Web.rRefreshJobTypes = rRefreshJobTypes + , Web.rRefreshJobRunners = rRefreshJobRunners + , Web.rStaticAsset = linkFn + } + where + OddJobs.Endpoints.Routes{..} = allFieldLinks' (linkFn . toS . show . linkURI) :: OddJobs.Endpoints.Routes (AsLink Text) -ariaExpanded_ :: Text -> Attribute -ariaExpanded_ v = makeAttribute "aria-expanded" v +-- absText :: Link -> Text +-- absText l = "/" <> (toS $ show $ linkURI l) diff --git a/src/OddJobs/Job.hs b/src/OddJobs/Job.hs --- a/src/OddJobs/Job.hs +++ b/src/OddJobs/Job.hs @@ -1,4 +1,7 @@ {-# LANGUAGE RankNTypes, FlexibleInstances, FlexibleContexts, PartialTypeSignatures, TupleSections, DeriveGeneric, UndecidableInstances #-} +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE ScopedTypeVariables #-} + module OddJobs.Job ( -- * Starting the job-runner @@ -9,21 +12,9 @@ -- * Configuring the job-runner -- -- $config - , Config(..) + , Config(..) , ConcurrencyControl(..) - -- ** Configuration helpers - -- - -- $configHelpers - , defaultConfig - , defaultJobToText - , defaultJobType - , defaultTimedLogger - , defaultLogStr - , defaultPollingInterval - , defaultLockTimeout - , withConnectionPool - -- * Creating/scheduling jobs -- -- $createJobs @@ -36,9 +27,13 @@ , Job(..) , JobId , Status(..) + , JobRunnerName(..) , TableName , delaySeconds , Seconds(..) + , JobErrHandler(..) + , AllJobTypes(..) + -- ** Structured logging -- -- $logging @@ -58,12 +53,15 @@ -- * Database helpers -- -- $dbHelpers - , findJobById , findJobByIdIO - , saveJob , saveJobIO + , runJobNowIO + , unlockJobIO + , cancelJobIO , jobDbColumns , concatJobDbColumns + , fetchAllJobTypes + , fetchAllJobRunners -- * JSON helpers -- @@ -80,9 +78,6 @@ import Data.Text as T import Database.PostgreSQL.Simple as PGS import Database.PostgreSQL.Simple.Notification -import Database.PostgreSQL.Simple.FromField as FromField -import Database.PostgreSQL.Simple.ToField as ToField -import Database.PostgreSQL.Simple.FromRow as FromRow import UnliftIO.Async import UnliftIO.Concurrent (threadDelay, myThreadId) import Data.String @@ -90,7 +85,7 @@ import Network.HostName (getHostName) import UnliftIO.MVar import Debug.Trace -import Control.Monad.Logger as MLogger (LogLevel(..), LogStr(..), toLogStr) +import Control.Monad.Logger as MLogger (LogLevel(..), LogStr, toLogStr) import UnliftIO.IORef import UnliftIO.Exception ( SomeException(..), try, catch, finally , catchAny, bracket, Exception(..), throwIO @@ -107,7 +102,7 @@ import Data.String.Conv (StringConv(..), toS) import Data.Functor (void) import Control.Monad (forever) -import Data.Maybe (isNothing, maybe, fromMaybe) +import Data.Maybe (isNothing, maybe, fromMaybe, listToMaybe, mapMaybe) import Data.Either (either) import Control.Monad.Reader import GHC.Generics @@ -118,9 +113,10 @@ import System.FilePath (FilePath) import qualified System.Directory as Dir import Data.Aeson.Internal (iparse, IResult(..), formatError) -import qualified System.Log.FastLogger as FLogger import Prelude hiding (log) - +import GHC.Exts (toList) +import Database.PostgreSQL.Simple.Types as PGS (Identifier(..)) +import Database.PostgreSQL.Simple.ToField as PGS (toField) -- | The documentation of odd-jobs currently promotes 'startJobRunner', which -- expects a fairly detailed 'Config' record, as a top-level function for @@ -128,237 +124,38 @@ -- an enviroment for a 'ReaderT', and almost all functions are written in this -- 'ReaderT' monad which impleents an instance of the 'HasJobRunner' type-class. -- --- **In future,** this /internal/ implementation detail will allow us to offer a +-- __In future,__ this /internal/ implementation detail will allow us to offer a -- type-class based interface as well (similar to what -- 'Yesod.JobQueue.YesodJobQueue' provides). class (MonadUnliftIO m, MonadBaseControl IO m) => HasJobRunner m where getPollingInterval :: m Seconds onJobSuccess :: Job -> m () - onJobFailed :: Job -> m () - onJobPermanentlyFailed :: Job -> m () + onJobFailed :: forall a . m [JobErrHandler a] getJobRunner :: m (Job -> IO ()) getDbPool :: m (Pool Connection) getTableName :: m TableName onJobStart :: Job -> m () getDefaultMaxAttempts :: m Int - onJobTimeout :: Job -> m () getRunnerEnv :: m RunnerEnv getConcurrencyControl :: m ConcurrencyControl getPidFile :: m (Maybe FilePath) - getJobToText :: m (Job -> Text) log :: LogLevel -> LogEvent -> m () + getDefaultJobTimeout :: m Seconds -- $logging -- --- TODO: Complete the prose here - - -data LogEvent - -- | Emitted when a job starts execution - = LogJobStart !Job - -- | Emitted when a job succeeds along with the time taken for execution. - | LogJobSuccess !Job !NominalDiffTime - -- | Emitted when a job fails (but will be retried) along with the time taken for - -- /this/ attempt - | LogJobFailed !Job !NominalDiffTime - -- | Emitted when a job fails permanently (and will no longer be retried) along - -- with the time taken for /this/ attempt (i.e. final attempt) - | LogJobPermanentlyFailed !Job !NominalDiffTime - -- | Emitted when a job times out and is picked-up again for execution - | LogJobTimeout !Job - -- | Emitted whenever 'jobPoller' polls the DB table - | LogPoll - -- | Emitted whenever any other event occurs - | LogText !Text - deriving (Eq, Show, Generic) - - --- | While odd-jobs is highly configurable and the 'Config' data-type might seem --- daunting at first, it is not necessary to tweak every single configuration --- parameter by hand. Please start-off by using the sensible defaults provided --- by the [configuration helpers](#configHelpers), and tweaking --- config parameters on a case-by-case basis. -data Config = Config - { -- | The DB table which holds your jobs. Please note, this should have been - -- created by the 'OddJobs.Migrations.createJobTable' function. - cfgTableName :: TableName - - -- | The actualy "job-runner" that __you__ need to provide. Please look at - -- the examples/tutorials if your applicaton's code is not in the @IO@ - -- monad. - , cfgJobRunner :: Job -> IO () - - -- | The number of times a failing job is retried before it is considered is - -- "permanently failed" and ignored by the job-runner. This config parameter - -- is called "/default/ max attempts" because, in the future, it would be - -- possible to specify the number of retry-attemps on a per-job basis - -- (__Note:__ per-job retry-attempts has not been implemented yet) - , cfgDefaultMaxAttempts :: Int - - -- | Controls how many jobs can be run concurrently by /this instance/ of - -- the job-runner. __Please note,__ this is NOT the global concurrency of - -- entire job-queue. It is possible to have job-runners running on multiple - -- machines, and each will apply the concurrency control independnt of other - -- job-runners. TODO: Link-off to relevant section in the tutorial. - , cfgConcurrencyControl :: ConcurrencyControl - - - -- | The DB connection-pool to use for the job-runner. __Note:__ in case - -- your jobs require a DB connection, please create a separate - -- connection-pool for them. This pool will be used ONLY for monitoring jobs - -- and changing their status. We need to have _at least 4 connections__ in - -- this connection-pool for the job-runner to work as expected. (TODO: - -- Link-off to tutorial) - , cfgDbPool :: Pool Connection - - -- | How frequently should the 'jobPoller' check for jobs where the Job's - -- 'jobRunAt' field indicates that it's time for the job to be executed. - -- TODO: link-off to the tutorial. - , cfgPollingInterval :: Seconds - - -- | User-defined callback function that is called whenever a job succeeds. - , cfgOnJobSuccess :: Job -> IO () - - -- | User-defined callback function that is called whenever a job fails. This - -- does not indicate permanent failure and means the job will be retried. It - -- is a good idea to log the failures to Airbrake, NewRelic, Sentry, or some - -- other error monitoring tool. - , cfgOnJobFailed :: Job -> IO () - - -- | User-defined callback function that is called whenever a job fails - -- /permanently/ (i.e. number of retry-attempts have crossed the configured - -- threshold). It is a good idea to log the failures to Airbrake, NewRelic, - -- Sentry, or some other error monitoring tool. - , cfgOnJobPermanentlyFailed :: Job -> IO () - - -- | User-defined callback function that is called whenever a job starts - -- execution. - , cfgOnJobStart :: Job -> IO () - - -- | User-defined callback function that is called whenever a job times-out. - , cfgOnJobTimeout :: Job -> IO () - - -- | File to store the PID of the job-runner process. This is used only when - -- invoking the job-runner as an independent background deemon (the usual mode - -- of deployment). (TODO: Link-off to tutorial). - , cfgPidFile :: Maybe FilePath - - -- | A "structured logging" function that __you__ need to provide. The - -- @odd-jobs@ library does NOT use the standard logging interface provided by - -- 'monad-logger' on purpose. TODO: link-off to tutorial. Please also read - -- 'cfgJobToText' and 'cfgJobType' - , cfgLogger :: LogLevel -> LogEvent -> IO () - - -- | When emitting certain text messages in logs, how should the 'Job' be - -- summarized in a textual format? Related: 'defaultJobToText' - , cfgJobToText :: Job -> Text - - -- | How to extract the "job type" from a 'Job'. Related: 'defaultJobType' - , cfgJobType :: Job -> Text - } - -data ConcurrencyControl - -- | The maximum number of concurrent jobs that /this instance/ of the - -- job-runner can execute. TODO: Link-off to tutorial. - = MaxConcurrentJobs Int - -- | __Not recommended:__ Please do not use this in production unless you know - -- what you're doing. No machine can support unlimited concurrency. If your - -- jobs are doing anything worthwhile, running a sufficiently large number - -- concurrently is going to max-out /some/ resource of the underlying machine, - -- such as, CPU, memory, disk IOPS, or network bandwidth. - | UnlimitedConcurrentJobs - - -- | Use this to dynamically determine if the next job should be picked-up, or - -- not. This is useful to write custom-logic to determine whether a limited - -- resource is below a certain usage threshold (eg. CPU usage is below 80%). - -- __Caveat:__ This feature has not been tested in production, yet. TODO: - -- Link-off to tutorial. - | DynamicConcurrency (IO Bool) - -instance Show ConcurrencyControl where - show cc = case cc of - MaxConcurrentJobs n -> "MaxConcurrentJobs " <> show n - UnlimitedConcurrentJobs -> "UnlimitedConcurrentJobs" - DynamicConcurrency _ -> "DynamicConcurrency (IO Bool)" - - --- $configHelpers +-- OddJobs uses \"structured logging\" for important events that occur during +-- the life-cycle of a job-runner. This is useful if you're using JSON\/XML for +-- aggegating logs of your entire application to something like Kibana, AWS +-- CloudFront, GCP StackDriver Logging, etc. -- --- #configHelpers# - --- | This function gives you a 'Config' with a bunch of sensible defaults --- already applied. It requies the bare minimum arguments that this library --- cannot assume on your behalf. +-- If you're not interested in using structured logging, look at +-- 'OddJobs.ConfigBuilder.defaultLogStr' to output plain-text logs (or you can +-- write your own function, as well). -- --- It makes a few __important assumptions__ about your 'jobPayload 'JSON, which --- are documented in 'defaultJobType'. -defaultConfig :: (LogLevel -> LogEvent -> IO ()) -- ^ "Structured logging" function. Ref: 'cfgLogger' - -> TableName -- ^ DB table which holds your jobs. Ref: 'cfgTableName' - -> Pool Connection -- ^ DB connection-pool to be used by job-runner. Ref: 'cfgDbPool' - -> ConcurrencyControl -- ^ Concurrency configuration. Ref: 'cfgConcurrencyControl' - -> (Job -> IO ()) -- ^ The actual "job runner" which contains your application code. Ref: 'cfgJobRunner' - -> Config -defaultConfig logger tname dbpool ccControl jrunner = - let cfg = Config - { cfgPollingInterval = defaultPollingInterval - , cfgOnJobSuccess = (const $ pure ()) - , cfgOnJobFailed = (const $ pure ()) - , cfgOnJobPermanentlyFailed = (const $ pure ()) - , cfgJobRunner = jrunner - , cfgLogger = logger - , cfgDbPool = dbpool - , cfgOnJobStart = (const $ pure ()) - , cfgDefaultMaxAttempts = 10 - , cfgTableName = tname - , cfgOnJobTimeout = (const $ pure ()) - , cfgConcurrencyControl = ccControl - , cfgPidFile = Nothing - , cfgJobToText = defaultJobToText (cfgJobType cfg) - , cfgJobType = defaultJobType - } - in cfg --- | Used only by 'defaultLogStr' now. TODO: Is this even required anymore? --- Should this be removed? -defaultJobToText :: (Job -> Text) -> Job -> Text -defaultJobToText jobTypeFn job@Job{jobId} = - "JobId=" <> (toS $ show jobId) <> " JobType=" <> jobTypeFn job --- | This makes __two important assumptions__. First, this /assumes/ that jobs --- in your app are represented by a sum-type. For example: --- --- @ --- data MyJob = SendWelcomeEmail Int --- | SendPasswordResetEmail Text --- | SetupSampleData Int --- @ --- --- Second, it /assumes/ that the JSON representatin of this sum-type is --- "tagged". For example, the following... --- --- > let pload = SendWelcomeEmail 10 --- --- ...when converted to JSON, would look like... --- --- > {"tag":"SendWelcomeEmail", "contents":10} --- --- It uses this assumption to extract the "job type" from a 'Data.Aeson.Value' --- (which would be @SendWelcomeEmail@ in the example given above). This is used --- in logging and the admin UI. --- --- Even if tihs assumption is violated, the job-runner /should/ continue to --- function. It's just that you won't get very useful log messages. -defaultJobType :: Job -> Text -defaultJobType Job{jobPayload} = - case jobPayload of - Aeson.Object hm -> case HM.lookup "tag" hm of - Just (Aeson.String t) -> t - _ -> "unknown" - _ -> "unknown" - - - data RunnerEnv = RunnerEnv { envConfig :: !Config , envJobThreadsRef :: !(IORef [Async ()]) @@ -371,15 +168,10 @@ instance HasJobRunner RunnerM where getPollingInterval = cfgPollingInterval . envConfig <$> ask - onJobFailed job = do - fn <- cfgOnJobFailed . envConfig <$> ask - logCallbackErrors (jobId job) "onJobFailed" $ liftIO $ fn job + onJobFailed = cfgOnJobFailed . envConfig <$> ask onJobSuccess job = do fn <- cfgOnJobSuccess . envConfig <$> ask logCallbackErrors (jobId job) "onJobSuccess" $ liftIO $ fn job - onJobPermanentlyFailed job = do - fn <- cfgOnJobPermanentlyFailed . envConfig <$> ask - logCallbackErrors (jobId job) "onJobPermanentlyFailed" $ liftIO $ fn job getJobRunner = cfgJobRunner . envConfig <$> ask getDbPool = cfgDbPool . envConfig <$> ask getTableName = cfgTableName . envConfig <$> ask @@ -387,10 +179,6 @@ fn <- cfgOnJobStart . envConfig <$> ask logCallbackErrors (jobId job) "onJobStart" $ liftIO $ fn job - onJobTimeout job = do - fn <- cfgOnJobTimeout . envConfig <$> ask - logCallbackErrors (jobId job) "onJobTimeout" $ liftIO $ fn job - getDefaultMaxAttempts = cfgDefaultMaxAttempts . envConfig <$> ask getRunnerEnv = ask @@ -398,12 +186,13 @@ getConcurrencyControl = (cfgConcurrencyControl . envConfig <$> ask) getPidFile = cfgPidFile . envConfig <$> ask - getJobToText = cfgJobToText . envConfig <$> ask log logLevel logEvent = do loggerFn <- cfgLogger . envConfig <$> ask liftIO $ loggerFn logLevel logEvent + getDefaultJobTimeout = cfgDefaultJobTimeout . envConfig <$> ask + -- | Start the job-runner in the /current/ thread, i.e. you'll need to use -- 'forkIO' or 'async' manually, if you want the job-runner to run in the -- background. Consider using 'OddJobs.Cli' to rapidly build your own @@ -417,107 +206,18 @@ } runReaderT jobMonitor monitorEnv --- | Convenience function to create a DB connection-pool with some sensible --- defaults. Please see the source-code of this function to understand what it's --- doing. TODO: link-off to tutorial. -withConnectionPool :: (MonadUnliftIO m) - => Either BS.ByteString PGS.ConnectInfo - -> (Pool PGS.Connection -> m a) - -> m a -withConnectionPool connConfig action = withRunInIO $ \runInIO -> do - bracket poolCreator destroyAllResources (runInIO . action) - where - poolCreator = liftIO $ - case connConfig of - Left connString -> - createPool (PGS.connectPostgreSQL connString) PGS.close 1 (fromIntegral $ 2 * (unSeconds defaultPollingInterval)) 5 - Right connInfo -> - createPool (PGS.connect connInfo) PGS.close 1 (fromIntegral $ 2 * (unSeconds defaultPollingInterval)) 5 - --- | As the name says. Ref: 'cfgPollingInterval' -defaultPollingInterval :: Seconds -defaultPollingInterval = Seconds 5 - -type JobId = Int - -data Status = Success - | Queued - | Failed - | Retry - | Locked - deriving (Eq, Show, Generic, Enum) - -instance Ord Status where - compare x y = compare (toText x) (toText y) - -data Job = Job - { jobId :: JobId - , jobCreatedAt :: UTCTime - , jobUpdatedAt :: UTCTime - , jobRunAt :: UTCTime - , jobStatus :: Status - , jobPayload :: Aeson.Value - , jobLastError :: Maybe Value - , jobAttempts :: Int - , jobLockedAt :: Maybe UTCTime - , jobLockedBy :: Maybe Text - } deriving (Eq, Show) - -instance ToText Status where - toText s = case s of - Success -> "success" - Queued -> "queued" - Retry -> "retry" - Failed -> "failed" - Locked -> "locked" - -instance (StringConv Text a) => FromText (Either a Status) where - fromText t = case t of - "success" -> Right Success - "queued" -> Right Queued - "failed" -> Right Failed - "retry" -> Right Retry - "locked" -> Right Locked - x -> Left $ toS $ "Unknown job status: " <> x - -instance FromField Status where - fromField f mBS = (fromText <$> (fromField f mBS)) >>= \case - Left e -> FromField.returnError PGS.ConversionFailed f e - Right s -> pure s - -instance ToField Status where - toField s = toField $ toText s - -instance FromRow Job where - fromRow = Job - <$> field -- jobId - <*> field -- createdAt - <*> field -- updatedAt - <*> field -- runAt - <*> field -- status - <*> field -- payload - <*> field -- lastError - <*> field -- attempts - <*> field -- lockedAt - <*> field -- lockedBy - --- TODO: Add a sum-type for return status which can signal the monitor about --- whether the job needs to be retried, marked successfull, or whether it has --- completed failed. -type JobRunner = Job -> IO () - - jobWorkerName :: IO String jobWorkerName = do pid <- getProcessID hname <- getHostName pure $ hname ++ ":" ++ (show pid) --- | TODO: Make this configurable for the job-runner, why is this still --- hard-coded? -defaultLockTimeout :: Seconds -defaultLockTimeout = Seconds 600 - +-- | If you are writing SQL queries where you want to return ALL columns from +-- the jobs table it is __recommended__ that you do not issue a @SELECT *@ or +-- @RETURNIG *@. List out specific DB columns using 'jobDbColumns' and +-- 'concatJobDbColumns' instead. This will insulate you from runtime errors +-- caused by addition of new columns to 'cfgTableName' in future versions of +-- OddJobs. jobDbColumns :: (IsString s, Semigroup s) => [s] jobDbColumns = [ "id" @@ -532,6 +232,11 @@ , "locked_by" ] +-- | All 'jobDbColumns' joined together with commas. Useful for constructing SQL +-- queries, eg: +-- +-- @'query_' conn $ "SELECT " <> concatJobDbColumns <> "FROM jobs"@ + concatJobDbColumns :: (IsString s, Semigroup s) => s concatJobDbColumns = concatJobDbColumns_ jobDbColumns "" where @@ -550,6 +255,18 @@ pool <- getDbPool withResource pool action +-- +-- $dbHelpers +-- +-- A bunch of functions that help you query 'cfgTableName' and change the status +-- of individual jobs. Most of these functions are in @IO@ and you /might/ want +-- to write wrappers that lift them into you application's custom monad. +-- +-- __Note:__ When passing a 'Connection' to these function, it is +-- __recommended__ __to not__ take a connection from 'cfgDbPool'. Use your +-- application\'s database pool instead. +-- + findJobById :: (HasJobRunner m) => JobId -> m (Maybe Job) @@ -567,6 +284,8 @@ saveJobQuery :: TableName -> PGS.Query saveJobQuery tname = "UPDATE " <> tname <> " set run_at = ?, status = ?, payload = ?, last_error = ?, attempts = ?, locked_at = ?, locked_by = ? WHERE id = ? RETURNING " <> concatJobDbColumns +deleteJobQuery :: TableName -> PGS.Query +deleteJobQuery tname = "delete " <> tname <> " Where id = ?" saveJob :: (HasJobRunner m) => Job -> m Job saveJob j = do @@ -590,6 +309,45 @@ [j] -> pure j js -> Prelude.error $ "Not expecting multiple rows to ber returned when updating job id=" <> (show jobId) +deleteJob :: (HasJobRunner m) => JobId -> m () +deleteJob jid = do + tname <- getTableName + withDbConnection $ \conn -> liftIO $ deleteJobIO conn tname jid + +deleteJobIO :: Connection -> TableName -> JobId -> IO () +deleteJobIO conn tname jid = do + void $ PGS.execute conn (deleteJobQuery tname) (Only jid) + +runJobNowIO :: Connection -> TableName -> JobId -> IO (Maybe Job) +runJobNowIO conn tname jid = do + t <- getCurrentTime + updateJobHelper tname conn (Queued, [Queued, Retry, Failed], Just t, jid) + +-- | TODO: First check in all job-runners if this job is still running, or not, +-- and somehow send an uninterruptibleCancel to that thread. +unlockJobIO :: Connection -> TableName -> JobId -> IO (Maybe Job) +unlockJobIO conn tname jid = do + fmap listToMaybe $ PGS.query conn q (Retry, jid, In [Locked]) + where + q = "update " <> tname <> " set status=?, run_at=now(), locked_at=null, locked_by=null where id=? and status in ? returning " <> concatJobDbColumns + +cancelJobIO :: Connection -> TableName -> JobId -> IO (Maybe Job) +cancelJobIO conn tname jid = + updateJobHelper tname conn (Failed, [Queued, Retry], Nothing, jid) + +updateJobHelper :: TableName + -> Connection + -> (Status, [Status], Maybe UTCTime, JobId) + -> IO (Maybe Job) +updateJobHelper tname conn (newStatus, existingStates, mRunAt, jid) = + fmap listToMaybe $ PGS.query conn q (newStatus, runAt, jid, PGS.In existingStates) + where + q = "update " <> tname <> " set attempts=0, status=?, run_at=? where id=? and status in ? returning " <> concatJobDbColumns + runAt = case mRunAt of + Nothing -> PGS.toField $ PGS.Identifier "run_at" + Just t -> PGS.toField t + + data TimeoutException = TimeoutException deriving (Eq, Show) instance Exception TimeoutException @@ -623,42 +381,44 @@ Nothing -> Prelude.error $ "Could not find job id=" <> show jid Just job -> do startTime <- liftIO getCurrentTime + lockTimeout <- getDefaultJobTimeout (flip catches) [Handler $ timeoutHandler job startTime, Handler $ exceptionHandler job startTime] $ do - runJobWithTimeout defaultLockTimeout job + runJobWithTimeout lockTimeout job endTime <- liftIO getCurrentTime - newJob <- saveJob job{jobStatus=Success, jobLockedBy=Nothing, jobLockedAt=Nothing} + deleteJob jid + let newJob = job{jobStatus=Success, jobLockedBy=Nothing, jobLockedAt=Nothing} log LevelInfo $ LogJobSuccess newJob (diffUTCTime endTime startTime) onJobSuccess newJob pure () where - timeoutHandler job startTime (e :: TimeoutException) = retryOrFail (show e) job onJobTimeout onJobPermanentlyFailed startTime - exceptionHandler job startTime (e :: SomeException) = retryOrFail (show e) job onJobFailed onJobPermanentlyFailed startTime - retryOrFail errStr job@Job{jobAttempts} onFail onPermanentFail startTime = do + timeoutHandler job startTime (e :: TimeoutException) = retryOrFail (toException e) job startTime + exceptionHandler job startTime (e :: SomeException) = retryOrFail (toException e) job startTime + retryOrFail e job@Job{jobAttempts} startTime = do endTime <- liftIO getCurrentTime defaultMaxAttempts <- getDefaultMaxAttempts - jobToText <- getJobToText let runTime = diffUTCTime endTime startTime - (newStatus, action, logAction) = if jobAttempts >= defaultMaxAttempts - then ( Failed - , onPermanentFail - , log LevelError $ LogJobPermanentlyFailed job runTime - ) - else ( Retry - , onFail - , log LevelWarn $ LogJobFailed job runTime - ) + (newStatus, failureMode, logLevel) = if jobAttempts >= defaultMaxAttempts + then ( Failed, FailPermanent, LevelError ) + else ( Retry, FailWithRetry, LevelWarn ) t <- liftIO getCurrentTime newJob <- saveJob job{ jobStatus=newStatus , jobLockedBy=Nothing , jobLockedAt=Nothing - , jobLastError=(Just $ toJSON errStr) -- TODO: convert errors to json properly + , jobLastError=(Just $ toJSON $ show e) -- TODO: convert errors to json properly , jobRunAt=(addUTCTime (fromIntegral $ (2::Int) ^ jobAttempts) t) } - logAction - void $ action newJob - pure () + case fromException e :: Maybe TimeoutException of + Nothing -> log logLevel $ LogJobFailed newJob e failureMode runTime + Just _ -> log logLevel $ LogJobTimeout newJob + let tryHandler (JobErrHandler handler) res = case fromException e of + Nothing -> res + Just e_ -> handler e_ newJob failureMode + handlers <- onJobFailed + liftIO $ void $ Prelude.foldr tryHandler (throwIO e) handlers + pure () +-- TODO: This might have a resource leak. restartUponCrash :: (HasJobRunner m, Show a) => Text -> m a -> m () restartUponCrash name_ action = do a <- async action @@ -704,7 +464,7 @@ [] -> log LevelInfo $ LogText "All job-threads exited" as -> do tid <- myThreadId - (a, _) <- waitAnyCatch as + void $ waitAnyCatch as log LevelDebug $ LogText $ toS $ "Waiting for " <> show (DL.length as) <> " jobs to complete before shutting down. myThreadId=" <> (show tid) delaySeconds (Seconds 1) waitForJobs @@ -738,6 +498,7 @@ processName <- liftIO jobWorkerName pool <- getDbPool tname <- getTableName + lockTimeout <- getDefaultJobTimeout log LevelInfo $ LogText $ toS $ "Starting the job monitor via DB polling with processName=" <> processName concurrencyControlFn <- getConcurrencyControlFn withResource pool $ \pollerDbConn -> forever $ concurrencyControlFn >>= \case @@ -748,7 +509,7 @@ t <- liftIO getCurrentTime r <- liftIO $ PGS.query pollerDbConn (jobPollingSql tname) - (Locked, t, processName, t, (In [Queued, Retry]), Locked, (addUTCTime (fromIntegral $ negate $ unSeconds defaultLockTimeout) t)) + (Locked, t, processName, t, (In [Queued, Retry]), Locked, (addUTCTime (fromIntegral $ negate $ unSeconds lockTimeout) t)) case r of -- When we don't have any jobs to run, we can relax a bit... [] -> pure delayAction @@ -902,41 +663,26 @@ either throwString (pure . Prelude.id) (eitherParsePayloadWith parser job) --- | TODO: Should the library be doing this? -defaultTimedLogger :: FLogger.TimedFastLogger - -> (LogLevel -> LogEvent -> LogStr) - -> LogLevel - -> LogEvent - -> IO () -defaultTimedLogger logger logStrFn logLevel logEvent = - if logLevel == LevelDebug - then pure () - else logger $ \t -> (toLogStr t) <> " | " <> - (logStrFn logLevel logEvent) <> - "\n" +-- | Used by the web\/admin UI to fetch a \"master list\" of all known +-- job-types. Ref: 'cfgAllJobTypes' +fetchAllJobTypes :: (MonadIO m) + => Config + -> m [Text] +fetchAllJobTypes Config{cfgAllJobTypes, cfgDbPool} = liftIO $ do + case cfgAllJobTypes of + AJTFixed jts -> pure jts + AJTSql fn -> withResource cfgDbPool fn + AJTCustom fn -> fn - -- TODO; Should the library be doing this? -defaultLogStr :: (Job -> Text) - -> LogLevel - -> LogEvent - -> LogStr -defaultLogStr jobToText logLevel logEvent = - (toLogStr $ show logLevel) <> " | " <> str - where - jobToLogStr j = toLogStr $ jobToText j - str = case logEvent of - LogJobStart j -> - "Started | " <> jobToLogStr j - LogJobFailed j t -> - "Failed (retry) | " <> jobToLogStr j <> " | runtime=" <> (toLogStr $ show t) - LogJobSuccess j t -> - "Success | " <> (jobToLogStr j) <> " | runtime=" <> (toLogStr $ show t) - LogJobPermanentlyFailed j t -> - "Failed (permanent) | " <> jobToLogStr j <> " | runtime=" <> (toLogStr $ show t) - LogJobTimeout j@Job{jobLockedAt, jobLockedBy} -> - "Timeout | " <> jobToLogStr j <> " | lockedBy=" <> (toLogStr $ fromMaybe "unknown" jobLockedBy) <> - " lockedAt=" <> (toLogStr $ maybe "unknown" show jobLockedAt) - LogPoll -> - "Polling jobs table" - LogText t -> - toLogStr t +-- | Used by web\/admin IO to fetch a \"master list\" of all known job-runners. +-- There is a known issue with the way this has been implemented: +-- +-- * Since this looks at the 'jobLockedBy' column of 'cfgTableName', it will +-- discover only those job-runners that are actively executing at least one +-- job at the time this function is executed. +fetchAllJobRunners :: (MonadIO m) + => Config + -> m [JobRunnerName] +fetchAllJobRunners Config{cfgTableName, cfgDbPool} = liftIO $ withResource cfgDbPool $ \conn -> do + fmap (mapMaybe fromOnly) $ PGS.query_ conn $ "select distinct locked_by from " <> cfgTableName + diff --git a/src/OddJobs/Links.hs b/src/OddJobs/Links.hs deleted file mode 100644 --- a/src/OddJobs/Links.hs +++ /dev/null @@ -1,14 +0,0 @@ -{-# LANGUAGE RecordWildCards, OverloadedStrings #-} - -module OddJobs.Links where - -import Servant -import Servant.API.Generic -import OddJobs.Web (Routes(..)) -import Data.String.Conv -import Data.Text - -Routes{..} = allFieldLinks' absText :: Routes (AsLink Text) - -absText :: Link -> Text -absText l = "/" <> (toS $ show $ linkURI l) diff --git a/src/OddJobs/Types.hs b/src/OddJobs/Types.hs --- a/src/OddJobs/Types.hs +++ b/src/OddJobs/Types.hs @@ -1,8 +1,27 @@ +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE RankNTypes #-} + module OddJobs.Types where import Database.PostgreSQL.Simple as PGS import UnliftIO (MonadIO) import UnliftIO.Concurrent (threadDelay) +import Data.Text.Conversions +import Database.PostgreSQL.Simple.FromField as FromField +import Database.PostgreSQL.Simple.ToField as ToField +import Database.PostgreSQL.Simple.FromRow as FromRow +import Data.Time +import UnliftIO.Exception +import Data.Text (Text) +import GHC.Generics +import Data.Aeson as Aeson hiding (Success) +import Data.String.Conv +import Lucid (Html(..)) +import Data.Pool (Pool) +import Control.Monad.Logger (LogLevel) -- | An alias for 'Query' type. Since this type has an instance of 'IsString' -- you do not need to do anything special to create a value for this type. Just @@ -29,3 +48,300 @@ oneSec :: Int oneSec = 1000000 + +data LogEvent + -- | Emitted when a job starts execution + = LogJobStart !Job + -- | Emitted when a job succeeds along with the time taken for execution. + | LogJobSuccess !Job !NominalDiffTime + -- | Emitted when a job fails (but will be retried) along with the time taken for + -- /this/ attempt + | LogJobFailed !Job !SomeException !FailureMode !NominalDiffTime + -- | Emitted when a job times out and is picked-up again for execution + | LogJobTimeout !Job + -- | Emitted whenever 'OddJobs.Job.jobPoller' polls the DB table + | LogPoll + -- | TODO + | LogWebUIRequest + -- | Emitted whenever any other event occurs + | LogText !Text + deriving (Show, Generic) + +-- | Used by 'JobErrHandler' and 'LogEvent' to indicate the nature of failure. +data FailureMode + -- The job failed, but will be retried in the future. + = FailWithRetry + -- | The job failed and will no longer be retried (probably because it has + -- been tried 'cfgDefaultMaxAttempts' times already). + | FailPermanent deriving (Eq, Show, Generic) + +-- | Exception handler for jobs. This is conceptually very similar to how +-- 'Control.Exception.Handler' and 'Control.Exception.catches' (from +-- 'Control.Exception') work in-tandem. Using 'cfgOnJobFailed' you can install +-- /multiple/ exception handlers, where each handler is responsible for one type +-- of exception. OddJobs will execute the correct exception handler on the basis +-- of the type of runtime exception raised. For example: +-- +-- @ +-- cfgOnJobFailed = +-- [ JobErrHandler $ \(e :: HttpException) job failMode -> ... +-- , JobErrHandler $ \(e :: SqlException) job failMode -> ... +-- , JobErrHandler $ \(e :: ) job failMode -> ... +-- ] +-- @ +-- +-- __Note:__ Please refer to the section on [alerts and +-- notifications](https://www.haskelltutorials.com/odd-jobs/guide.html#alerts) +-- in the implementation guide to understand how to use the machinery provided +-- by 'JobErrHandler' and 'cfgOnJobFailed'. +data JobErrHandler a = forall e . (Exception e) => JobErrHandler (e -> Job -> FailureMode -> IO a) + + +-- | __Note:__ Please read the section on [controlling +-- concurrency](https://www.haskelltutorials.com/odd-jobs/guide.html#controlling-concurrency) +-- in the implementation guide to understand the implications of each option +-- given by the data-type. +data ConcurrencyControl + -- | The maximum number of concurrent jobs that /this instance/ of the + -- job-runner can execute. + = MaxConcurrentJobs Int + -- | __Not recommended:__ Please do not use this in production unless you know + -- what you're doing. No machine can support unlimited concurrency. If your + -- jobs are doing anything worthwhile, running a sufficiently large number + -- concurrently is going to max-out /some/ resource of the underlying machine, + -- such as, CPU, memory, disk IOPS, or network bandwidth. + | UnlimitedConcurrentJobs + + -- | Use this to dynamically determine if the next job should be picked-up, or + -- not. This is useful to write custom-logic to determine whether a limited + -- resource is below a certain usage threshold (eg. CPU usage is below 80%). + -- __Caveat:__ This feature has not been tested in production, yet. + | DynamicConcurrency (IO Bool) + +instance Show ConcurrencyControl where + show cc = case cc of + MaxConcurrentJobs n -> "MaxConcurrentJobs " <> show n + UnlimitedConcurrentJobs -> "UnlimitedConcurrentJobs" + DynamicConcurrency _ -> "DynamicConcurrency (IO Bool)" + +type JobId = Int + +data Status + -- | In the current version of odd-jobs you /should not/ find any jobs having + -- the 'Success' status, because successful jobs are immediately deleted. + -- However, in the future, we may keep them around for a certain time-period + -- before removing them from the jobs table. + = Success + -- | Jobs in 'Queued' status /may/ be picked up by the job-runner on the basis + -- of the 'jobRunAt' field. + | Queued + -- | Jobs in 'Failed' status will will not be retried by the job-runner. + | Failed + -- | Jobs in 'Retry' status will be retried by the job-runner on the basis of + -- the 'jobRunAt' field. + | Retry + -- | Jobs in 'Locked' status are currently being executed by a job-runner, + -- which is identified by the 'jobLockedBy' field. The start of job-execution + -- is indicated by the 'jobLocketAt' field. + | Locked + deriving (Eq, Show, Generic, Enum, Bounded) + +instance Ord Status where + compare x y = compare (toText x) (toText y) + +instance ToJSON Status where + toJSON s = toJSON $ toText s + +instance FromJSON Status where + parseJSON = withText "Expecting text to convert into Job.Status" $ \t -> do + case (fromText t :: Either String Status) of + Left e -> fail e + Right r -> pure r + + +newtype JobRunnerName = JobRunnerName { unJobRunnerName :: Text } deriving (Eq, Show, FromField, ToField, Generic, ToJSON, FromJSON) + +data Job = Job + { jobId :: JobId + , jobCreatedAt :: UTCTime + , jobUpdatedAt :: UTCTime + , jobRunAt :: UTCTime + , jobStatus :: Status + , jobPayload :: Aeson.Value + , jobLastError :: Maybe Value + , jobAttempts :: Int + , jobLockedAt :: Maybe UTCTime + , jobLockedBy :: Maybe JobRunnerName + } deriving (Eq, Show, Generic) + +instance ToText Status where + toText s = case s of + Success -> "success" + Queued -> "queued" + Retry -> "retry" + Failed -> "failed" + Locked -> "locked" + +instance (StringConv Text a) => FromText (Either a Status) where + fromText t = case t of + "success" -> Right Success + "queued" -> Right Queued + "failed" -> Right Failed + "retry" -> Right Retry + "locked" -> Right Locked + x -> Left $ toS $ "Unknown job status: " <> x + +instance FromField Status where + fromField f mBS = (fromText <$> (fromField f mBS)) >>= \case + Left e -> FromField.returnError PGS.ConversionFailed f e + Right s -> pure s + +instance ToField Status where + toField s = toField $ toText s + +instance FromRow Job where + fromRow = Job + <$> field -- jobId + <*> field -- createdAt + <*> field -- updatedAt + <*> field -- runAt + <*> field -- status + <*> field -- payload + <*> field -- lastError + <*> field -- attempts + <*> field -- lockedAt + <*> field -- lockedBy + +-- TODO: Add a sum-type for return status which can signal the monitor about +-- whether the job needs to be retried, marked successfull, or whether it has +-- completed failed. +type JobRunner = Job -> IO () + +-- | The web\/admin UI needs to know a \"master list\" of all job-types to be +-- able to power the \"filter by job-type\" feature. This data-type helps in +-- letting odd-jobs know /how/ to get such a master-list. The function specified +-- by this type is run once when the job-runner starts (and stored in an +-- internal @IORef@). After that the list of job-types needs to be updated +-- manually by pressing the appropriate \"refresh\" link in the admin\/web UI. +data AllJobTypes + -- | A fixed-list of job-types. If you don't want to increase boilerplate, + -- consider using 'OddJobs.ConfigBuilder.defaultConstantJobTypes' which will + -- automatically generate the list of available job-types based on a sum-type + -- that represents your job payload. + = AJTFixed [Text] + -- | Construct the list of job-types dynamically by looking at the actual + -- payloads in 'cfgTableName' (using an SQL query). + | AJTSql (Connection -> IO [Text]) + -- | A custom 'IO' action for fetching the list of job-types. + | AJTCustom (IO [Text]) + +-- | While odd-jobs is highly configurable and the 'Config' data-type might seem +-- daunting at first, it is not necessary to tweak every single configuration +-- parameter by hand. +-- +-- __Recommendation:__ Please start-off by building a 'Config' by using the +-- 'OddJobs.ConfigBuilder.mkConfig' function (to get something with sensible +-- defaults) and then tweaking config parameters on a case-by-case basis. +data Config = Config + { -- | The DB table which holds your jobs. Please note, this should have been + -- created by the 'OddJobs.Migrations.createJobTable' function. + cfgTableName :: TableName + + -- | The actualy "job-runner" that __you__ need to provide. If this function + -- throws a runtime exception, the job will be retried + -- 'cfgDefaultMaxAttempts' times. Please look at the examples/tutorials if + -- your applicaton's code is not in the @IO@ monad. + , cfgJobRunner :: Job -> IO () + + -- | The number of times a failing job is retried before it is considered is + -- "permanently failed" and ignored by the job-runner. This config parameter + -- is called "/default/ max attempts" because, in the future, it would be + -- possible to specify the number of retry-attemps on a per-job basis + -- (__Note:__ per-job retry-attempts has not been implemented yet) + , cfgDefaultMaxAttempts :: Int + + -- | Controls how many jobs can be run concurrently by /this instance/ of + -- the job-runner. __Please note,__ this is NOT the global concurrency of + -- entire job-queue. It is possible to have job-runners running on multiple + -- machines, and each will apply the concurrency control independnt of other + -- job-runners. __Ref:__ Section on [controllng + -- concurrency](https://www.haskelltutorials.com/odd-jobs/guide.html#controlling-concurrency) + -- in the implementtion guide. + , cfgConcurrencyControl :: ConcurrencyControl + + -- | The DB connection-pool to use for the job-runner. __Note:__ in case + -- your jobs require a DB connection, please create a separate + -- connection-pool for them. This pool will be used ONLY for monitoring jobs + -- and changing their status. We need to have __at least 4 connections__ in + -- this connection-pool for the job-runner to work as expected. + , cfgDbPool :: Pool Connection + + -- | How frequently should the 'jobPoller' check for jobs where the Job's + -- 'jobRunAt' field indicates that it's time for the job to be executed. + -- __Ref:__ Please read the section on [how Odd Jobs works + -- (architecture)](https://www.haskelltutorials.com/odd-jobs/guide.html#architecture) + -- to find out more. + , cfgPollingInterval :: Seconds + + -- | User-defined callback function that is called whenever a job succeeds. + , cfgOnJobSuccess :: Job -> IO () + + -- | User-defined error-handler that is called whenever a job fails (indicated + -- by 'cfgJobRunner' throwing an unhandled runtime exception). Please refer to + -- 'JobErrHandler' for documentation on how to use this. + , cfgOnJobFailed :: forall a . [JobErrHandler a] + + -- | User-defined callback function that is called whenever a job starts + -- execution. + , cfgOnJobStart :: Job -> IO () + + -- | User-defined callback function that is called whenever a job times-out. + -- Also check 'cfgDefaultJobTimeout' + , cfgOnJobTimeout :: Job -> IO () + + -- | File to store the PID of the job-runner process. This is used only when + -- invoking the job-runner as an independent background deemon (the usual mode + -- of deployment). + , cfgPidFile :: Maybe FilePath + + -- | A "structured logging" function that __you__ need to provide. The + -- @odd-jobs@ library does NOT use the standard logging interface provided by + -- 'monad-logger' on purpose. Also look at 'cfgJobType' and 'defaultLogStr' + -- + -- __Note:__ Please take a look at the section on [structured + -- logging](https://www.haskelltutorials.com/odd-jobs/guide.html#structured-logging) + -- to find out how to use this to log in JSON. + , cfgLogger :: LogLevel -> LogEvent -> IO () + + -- | How to extract the "job type" from a 'Job'. If you are overriding this, + -- please consider overriding 'cfgJobTypeSql' as well. Related: + -- 'OddJobs.ConfigBuilder.defaultJobType' + , cfgJobType :: Job -> Text + + -- | How to extract the \"job type\" directly in SQL. There are many places, + -- especially in the web\/admin UI, where we need to know a job's type + -- directly in SQL (because transferrring the entire @payload@ column to + -- Haskell, and then parsing it into JSON, and then applying the + -- 'cfgJobType' function on it would be too inefficient). Ref: + -- 'OddJobs.ConfigBuilder.defaultJobTypeSql' and 'cfgJobType' + , cfgJobTypeSql :: PGS.Query + + -- | How long can a job run after which it is considered to be "crashed" and + -- picked up for execution again + , cfgDefaultJobTimeout :: Seconds + + -- | How to convert a list of 'Job's to a list of HTML fragments. This is + -- used in the Web\/Admin UI. This function accepts a /list/ of jobs and + -- returns a /list/ of 'Html' fragments, because, in case, you need to query + -- another table to fetch some metadata (eg. convert a primary-key to a + -- human-readable name), you can do it efficiently instead of resulting in + -- an N+1 SQL bug. Ref: 'defaultJobToHtml' + , cfgJobToHtml :: [Job] -> IO [Html ()] + + -- | How to get a list of all known job-types? This is used by the + -- Web\/Admin UI to power the \"filter by job-type\" functionality. The + -- default value for this is 'OddJobs.ConfigBuilder.defaultDynamicJobTypes' + -- which does a @SELECT DISTINCT payload ->> ...@ to get a list of job-types + -- directly from the DB. + , cfgAllJobTypes :: AllJobTypes + } diff --git a/src/OddJobs/Web.hs b/src/OddJobs/Web.hs --- a/src/OddJobs/Web.hs +++ b/src/OddJobs/Web.hs @@ -1,12 +1,13 @@ -{-# LANGUAGE DeriveGeneric, NamedFieldPuns, TypeOperators, DataKinds #-} +{-# LANGUAGE DeriveGeneric, NamedFieldPuns, TypeOperators, DataKinds, RecordWildCards, DeriveAnyClass #-} module OddJobs.Web where import OddJobs.Types import OddJobs.Job as Job import Data.Time import Data.Aeson as Aeson -import Data.Text as T -import GHC.Generics +import qualified Data.Text as T +import Data.Text (Text) +import GHC.Generics hiding (from, to) import Database.PostgreSQL.Simple as PGS import Database.PostgreSQL.Simple.ToRow as PGS import Database.PostgreSQL.Simple.ToField as PGS @@ -20,7 +21,16 @@ import Servant.API.Generic import Servant.HTML.Lucid import Lucid +import Lucid.Html5 +import Lucid.Base import Data.String.Conv +import qualified Data.HashMap.Strict as HM +import Data.List as DL hiding (filter, and) +import Control.Monad +import Data.Time.Format.Human (humanReadableTime') +import Data.Time.Convenience (timeSince, Unit(..), Direction(..)) +import Data.Text.Conversions (fromText, toText) +import Prelude hiding (filter, and) data OrderDirection = Asc | Desc deriving (Eq, Show, Generic, Enum) @@ -41,6 +51,7 @@ , filterOrder :: Maybe (OrderByField, OrderDirection) , filterPage :: Maybe (Int, Int) , filterRunAfter :: Maybe UTCTime + , filterJobRunner :: [JobRunnerName] } deriving (Eq, Show, Generic) instance Semigroup Filter where @@ -54,6 +65,7 @@ , filterOrder = filterOrder b <|> filterOrder a , filterPage = filterPage b <|> filterPage a , filterRunAfter = filterRunAfter b <|> filterRunAfter a + , filterJobRunner = nub (filterJobRunner b <> filterJobRunner a) } instance Monoid Filter where @@ -70,12 +82,9 @@ , filterOrder = Nothing , filterPage = Just (10, 0) , filterRunAfter = Nothing + , filterJobRunner = [] } - -instance ToJSON Status -instance FromJSON Status - instance ToJSON OrderDirection instance FromJSON OrderDirection instance ToJSON OrderByField @@ -95,15 +104,31 @@ instance ToHttpApiData Filter where toQueryParam x = toS $ Aeson.encode x -data Routes route = Routes - { rFilterResults :: route :- QueryParam "filters" Filter :> Get '[HTML] (Html ()) - , rStaticAssets :: route :- "assets" :> Raw - } deriving (Generic) +-- data Routes route = Routes +-- { rFilterResults :: route :- QueryParam "filters" Filter :> Get '[HTML] (Html ()) +-- , rStaticAssets :: route :- "assets" :> Raw +-- , rEnqueue :: route :- "enqueue" :> Capture "jobId" JobId :> Post '[HTML] NoContent +-- , rRunNow :: route :- "run" :> Capture "jobId" JobId :> Post '[HTML] NoContent +-- , rCancel :: route :- "cancel" :> Capture "jobId" JobId :> Post '[HTML] NoContent +-- , rRefreshJobTypes :: route :- "refresh-job-types" :> Post '[HTML] NoContent +-- , rRefreshJobRunners :: route :- "refresh-job-runners" :> Post '[HTML] NoContent +-- } deriving (Generic) -filterJobsQuery :: TableName -> Filter -> (PGS.Query, [Action]) -filterJobsQuery tname Filter{filterStatuses, filterCreatedBefore, filterCreatedAfter, filterUpdatedBefore, filterUpdatedAfter, filterJobTypes, filterOrder, filterPage, filterRunAfter} = - ( "SELECT " <> Job.concatJobDbColumns <> " FROM " <> tname <> whereClause <> " " <> (orderClause $ fromMaybe (OrdUpdatedAt, Desc) filterOrder) <> " " <> limitOffsetClause +data Routes = Routes + { rFilterResults :: Maybe Filter -> Text + , rEnqueue :: JobId -> Text + , rRunNow :: JobId -> Text + , rCancel :: JobId -> Text + , rRefreshJobTypes :: Text + , rRefreshJobRunners :: Text + , rStaticAsset :: Text -> Text + } + + +filterJobsQuery :: Config -> Filter -> (PGS.Query, [Action]) +filterJobsQuery Config{cfgTableName, cfgJobTypeSql} Filter{..} = + ( "SELECT " <> Job.concatJobDbColumns <> " FROM " <> cfgTableName <> whereClause <> " " <> (orderClause $ fromMaybe (OrdUpdatedAt, Desc) filterOrder) <> " " <> limitOffsetClause , whereActions ) where @@ -124,9 +149,14 @@ Nothing -> mempty Just (l, o) -> "LIMIT " <> fromString (show l) <> " OFFSET " <> fromString (show o) - (whereClause, whereActions) = case statusClause `and` createdAfterClause `and` createdBeforeClause `and` updatedBeforeClause `and` updatedAfterClause `and` jobTypeClause `and` runAfterClause of - Nothing -> (mempty, toRow ()) - Just (q, as) -> (" WHERE " <> q, as) + (whereClause, whereActions) = + let finalClause = statusClause `and` createdAfterClause `and` + createdBeforeClause `and` updatedBeforeClause `and` + updatedAfterClause `and` jobTypeClause `and` + runAfterClause `and` jobRunnerClause + in case finalClause of + Nothing -> (mempty, toRow ()) + Just (q, as) -> (" WHERE " <> q, as) statusClause = if Prelude.null filterStatuses then Nothing @@ -142,14 +172,18 @@ jobTypeClause = case filterJobTypes of [] -> Nothing xs -> - let qFragment = "payload @> ?" - valBuilder v = toField $ Aeson.object ["tag" .= v] + let qFragment = "(" <> cfgJobTypeSql <> ")=?" build ys (q, vs) = case ys of [] -> (q, vs) - (y:[]) -> (qFragment <> q, (valBuilder y):vs) - (y:ys_) -> build ys_ (" OR " <> qFragment <> q, (valBuilder y):vs) + (y:[]) -> (qFragment <> q, (toField y):vs) + (y:ys_) -> build ys_ (" OR " <> qFragment <> q, (toField y):vs) in Just $ build xs (mempty, []) + jobRunnerClause :: Maybe (Query, [Action]) + jobRunnerClause = case filterJobRunner of + [] -> Nothing + xs -> Just ("locked_by in ?", toRow $ Only $ In xs) + and :: Maybe (Query, [PGS.Action]) -> Maybe (Query, [PGS.Action]) -> Maybe (Query, [PGS.Action]) and Nothing Nothing = Nothing and Nothing (Just (q, as)) = Just (q, as) @@ -158,14 +192,14 @@ -- orderClause = _ -filterJobs :: Connection -> TableName -> Filter -> IO [Job] -filterJobs conn tname f = do - let (q, queryArgs) = filterJobsQuery tname f +filterJobs :: Config -> Connection -> Filter -> IO [Job] +filterJobs cfg conn f = do + let (q, queryArgs) = filterJobsQuery cfg f PGS.query conn q queryArgs -countJobs :: Connection -> TableName -> Filter -> IO Int -countJobs conn tname f = do - let (q, queryArgs) = filterJobsQuery tname f +countJobs :: Config -> Connection -> Filter -> IO Int +countJobs cfg conn f = do + let (q, queryArgs) = filterJobsQuery cfg f finalqry = "SELECT count(*) FROM (" <> q <> ") a" [Only r] <- PGS.query conn finalqry queryArgs pure r @@ -177,3 +211,293 @@ -- { filterStatuses = [Job.Success, Queued] -- , filterJobTypes = ["QueuedMail", "ConfirmBooking"] -- } + + + +pageNav :: Routes -> Html () +pageNav Routes{..} = do + div_ $ nav_ [ class_ "navbar navbar-default navigation-clean" ] $ div_ [ class_ "container-fluid" ] $ do + div_ [ class_ "navbar-header" ] $ do + a_ [ class_ "navbar-brand navbar-link", href_ "#", style_ "margin-left: 2px; padding: 0px;" ] $ img_ [ src_ $ rStaticAsset "assets/odd-jobs-color-logo.png", title_ "Odd Jobs Logo" ] + button_ [ class_ "navbar-toggle collapsed", data_ "toggle" "collapse", data_ "target" "#navcol-1" ] $ do + span_ [ class_ "sr-only" ] $ "Toggle navigation" + span_ [ class_ "icon-bar" ] $ "" + span_ [ class_ "icon-bar" ] $ "" + span_ [ class_ "icon-bar" ] $ "" + -- div_ [ class_ "collapse navbar-collapse", id_ "navcol-1" ] $ ul_ [ class_ "nav navbar-nav navbar-right" ] $ do + -- li_ [ class_ "active", role_ "presentation" ] $ a_ [ href_ "#" ] $ "First Item" + -- li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "Second Item" + -- li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "Third Item" + -- li_ [ class_ "dropdown" ] $ do + -- a_ [ class_ "dropdown-toggle", data_ "toggle" "dropdown", ariaExpanded_ "false", href_ "#" ] $ do + -- "Dropdown" + -- span_ [ class_ "caret" ] $ "" + -- ul_ [ class_ "dropdown-menu", role_ "menu" ] $ do + -- li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "First Item" + -- li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "Second Item" + -- li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "Third Item" + +pageLayout :: Routes -> Html() -> Html () -> Html () +pageLayout routes@Routes{..} navHtml bodyHtml = do + doctype_ + html_ $ do + head_ $ do + meta_ [ charset_ "utf-8" ] + meta_ [ name_ "viewport", content_ "width=device-width, initial-scale=1.0" ] + title_ "haskell-pg-queue" + link_ [ rel_ "stylesheet", href_ $ rStaticAsset "assets/bootstrap/css/bootstrap.min.css" ] + link_ [ rel_ "stylesheet", href_ $ rStaticAsset "https://fonts.googleapis.com/css?family=Lato:100i,300,300i,400,700,900" ] + link_ [ rel_ "stylesheet", href_ $ rStaticAsset "assets/css/styles.css" ] + body_ $ do + pageNav routes + div_ $ div_ [ class_ "container-fluid", style_ "/*background-color:#f2f2f2;*/" ] $ div_ [ class_ "row" ] $ do + div_ [ class_ "d-none d-md-block col-md-2" ] navHtml + div_ [ class_ "col-12 col-md-10" ] bodyHtml + script_ [ src_ $ rStaticAsset "assets/js/jquery.min.js" ] $ ("" :: Text) + script_ [ src_ $ rStaticAsset "assets/bootstrap/js/bootstrap.min.js" ] $ ("" :: Text) + script_ [ src_ $ rStaticAsset "assets/js/custom.js" ] $ ("" :: Text) + +sideNav :: Routes -> [Text] -> [JobRunnerName] -> UTCTime -> Filter -> Html () +sideNav Routes{..} jobTypes jobRunnerNames t filter@Filter{..} = do + div_ [ class_ "filters mt-3" ] $ do + jobStatusFilters + jobTypeFilters + jobRunnerFilters + where + jobStatusFilters = do + h6_ "Filter by job status" + div_ [ class_ "card" ] $ do + ul_ [ class_ "list-group list-group-flush" ] $ do + li_ [ class_ ("list-group-item " <> if filterStatuses == [] then "active-nav" else "") ] $ do + let lnk = (rFilterResults $ Just filter{filterStatuses = [], filterPage = (OddJobs.Web.filterPage blankFilter)}) + a_ [ href_ lnk ] $ do + "all" + -- span_ [ class_ "badge badge-pill badge-secondary float-right" ] "12" + forM_ ((\\) (enumFrom minBound) [Job.Success]) $ \st -> do + li_ [ class_ ("list-group-item " <> if (st `elem` filterStatuses) then "active-nav" else "") ] $ do + let lnk = (rFilterResults $ Just filter{filterStatuses = [st], filterPage = Nothing}) + a_ [ href_ lnk ] $ do + toHtml $ toText st + -- span_ [ class_ "badge badge-pill badge-secondary float-right" ] "12" + + jobRunnerFilters = do + h6_ [ class_ "mt-3" ] $ do + "Filter by job-runner" + form_ [ method_ "post", action_ rRefreshJobRunners, class_ "d-inline"] $ do + button_ [ type_ "submit", class_ "btn btn-link m-0 p-0 ml-1 float-right"] $ do + small_ "refresh" + + div_ [ class_ "card" ] $ do + ul_ [ class_ "list-group list-group-flush" ] $ do + li_ [ class_ ("list-group-item " <> if filterJobRunner == [] then "active-nav" else "") ] $ do + let lnk = (rFilterResults $ Just filter{filterJobRunner = [], filterPage = (OddJobs.Web.filterPage blankFilter)}) + a_ [ href_ lnk ] "all" + forM_ jobRunnerNames $ \jr -> do + li_ [ class_ ("list-group-item" <> if (jr `elem` filterJobRunner) then " active-nav" else "")] $ do + a_ [ href_ "#" ] $ toHtml $ unJobRunnerName jr + + jobTypeFilters = do + h6_ [ class_ "mt-3" ] $ do + "Filter by job-type" + form_ [ method_ "post", action_ rRefreshJobTypes, class_ "d-inline"] $ do + button_ [ type_ "submit", class_ "btn btn-link m-0 p-0 ml-1 float-right"] $ do + small_ "refresh" + + div_ [ class_ "card" ] $ do + ul_ [ class_ "list-group list-group-flush" ] $ do + li_ [ class_ ("list-group-item " <> if filterJobTypes == [] then "active-nav" else "") ] $ do + let lnk = (rFilterResults $ Just filter{filterJobTypes = [], filterPage = (OddJobs.Web.filterPage blankFilter)}) + a_ [ href_ lnk ] "all" + forM_ jobTypes $ \jt -> do + li_ [ class_ ("list-group-item" <> if (jt `elem` filterJobTypes) then " active-nav" else "")] $ do + a_ [ href_ (rFilterResults $ Just filter{filterJobTypes=[jt]}) ] $ toHtml jt + +searchBar :: Routes -> UTCTime -> Filter -> Html () +searchBar Routes{..} t filter@Filter{filterStatuses, filterCreatedAfter, filterCreatedBefore, filterUpdatedAfter, filterUpdatedBefore, filterJobTypes, filterRunAfter} = do + form_ [ style_ "padding-top: 2em;" ] $ do + div_ [ class_ "form-group" ] $ do + div_ [ class_ "search-container" ] $ do + ul_ [ class_ "list-inline search-bar" ] $ do + forM_ filterStatuses $ \s -> renderFilter "Status" (toText s) (rFilterResults $ Just filter{filterStatuses = filterStatuses \\ [s]}) + maybe mempty (\x -> renderFilter "Created after" (showText x) (rFilterResults $ Just filter{filterCreatedAfter = Nothing})) filterCreatedAfter + maybe mempty (\x -> renderFilter "Created before" (showText x) (rFilterResults $ Just filter{filterCreatedBefore = Nothing})) filterCreatedBefore + maybe mempty (\x -> renderFilter "Updated after" (showText x) (rFilterResults $ Just filter{filterUpdatedAfter = Nothing})) filterUpdatedAfter + maybe mempty (\x -> renderFilter "Updated before" (showText x) (rFilterResults $ Just filter{filterUpdatedBefore = Nothing})) filterUpdatedBefore + maybe mempty (\x -> renderFilter "Run after" (showText x) (rFilterResults $ Just filter{filterRunAfter = Nothing})) filterRunAfter + forM_ filterJobTypes $ \x -> renderFilter "Job type" x (rFilterResults $ Just filter{filterJobTypes = filterJobTypes \\ [x]}) + + button_ [ class_ "btn btn-default search-button", type_ "button" ] $ "Search" + -- ul_ [ class_ "list-inline" ] $ do + -- li_ $ span_ $ strong_ "Common searches:" + -- li_ $ a_ [ href_ (rFilterResults $ Just mempty) ] $ "All jobs" + -- li_ $ a_ [ href_ (rFilterResults $ Just $ filter{ filterStatuses = [Job.Locked] }) ] $ "Currently running" + -- li_ $ a_ [ href_ (rFilterResults $ Just $ filter{ filterStatuses = [Job.Success] }) ] $ "Successful" + -- li_ $ a_ [ href_ (rFilterResults $ Just $ filter{ filterStatuses = [Job.Failed] }) ] $ "Failed" + -- li_ $ a_ [ href_ (rFilterResults $ Just $ filter{ filterRunAfter = Just t }) ] $ "Future" + -- -- li_ $ a_ [ href_ "#" ] $ "Retried" + -- li_ $ a_ [ href_ (rFilterResults $ Just $ filter{ filterStatuses = [Job.Queued] }) ] $ "Queued" + -- li_ $ a_ [ href_ (rFilterResults $ Just $ filter{ filterUpdatedAfter = Just $ timeSince t 10 Minutes Ago }) ] $ "Last 10 mins" + -- li_ $ a_ [ href_ (rFilterResults $ Just $ filter{ filterCreatedAfter = Just $ timeSince t 10 Minutes Ago }) ] $ "Recently created" + where + renderFilter :: Text -> Text -> Text -> Html () + renderFilter k v u = do + li_ [ class_ "search-filter" ] $ do + span_ [ class_ "filter-name" ] $ toHtml k + span_ [ class_ "filter-value" ] $ do + toHtml v + a_ [ href_ u, class_ "text-danger" ] $ i_ [ class_ "glyphicon glyphicon-remove" ] $ "" + + +timeDuration :: UTCTime -> UTCTime -> (Int, String) +timeDuration from to = (diff, str) + where + str = if diff <= 0 + then "under 1s" + else (if d>0 then (show d) <> "d" else "") <> + (if m>0 then (show m) <> "m" else "") <> + (if s>0 then (show s) <> "s" else "") + diff = (abs $ round $ diffUTCTime from to) + (m', s) = diff `divMod` 60 + (h', m) = m' `divMod` 60 + (d, h) = h' `divMod` 24 + +showText :: (Show a) => a -> Text +showText a = toS $ show a + +jobContent :: Value -> Value +jobContent v = case v of + Aeson.Object o -> case HM.lookup "contents" o of + Nothing -> v + Just c -> c + _ -> v + +jobRow :: Routes -> UTCTime -> (Job, Html ()) -> Html () +jobRow routes t (job@Job{..}, jobHtml) = do + tr_ $ do + td_ [ class_ "job-type" ] $ do + let statusFn = case jobStatus of + Job.Success -> statusSuccess + Job.Failed -> statusFailed + Job.Queued -> if jobRunAt > t + then statusFuture + else statusWaiting + Job.Retry -> statusRetry + Job.Locked -> statusLocked + statusFn t job + + td_ jobHtml + td_ $ do + let actionsFn = case jobStatus of + Job.Success -> (const mempty) + Job.Failed -> actionsFailed + Job.Queued -> if jobRunAt > t + then actionsFuture + else actionsWaiting + Job.Retry -> actionsRetry + Job.Locked -> (const mempty) + actionsFn routes job + + +actionsFailed :: Routes -> Job -> Html () +actionsFailed Routes{..} Job{..} = do + form_ [ action_ (rEnqueue jobId), method_ "post" ] $ do + button_ [ class_ "btn btn-secondary", type_ "submit" ] $ "Enqueue again" + +actionsRetry :: Routes -> Job -> Html () +actionsRetry Routes{..} Job{..} = do + form_ [ action_ (rRunNow jobId), method_ "post" ] $ do + button_ [ class_ "btn btn-secondary", type_ "submit" ] $ "Run now" + +actionsFuture :: Routes -> Job -> Html () +actionsFuture Routes{..} Job{..} = do + form_ [ action_ (rRunNow jobId), method_ "post" ] $ do + button_ [ class_ "btn btn-secondary", type_ "submit" ] $ "Run now" + +actionsWaiting :: Routes -> Job -> Html () +actionsWaiting Routes{..} Job{..} = do + form_ [ action_ (rCancel jobId), method_ "post" ] $ do + button_ [ class_ "btn btn-danger", type_ "submit" ] $ "Cancel" + +statusSuccess :: UTCTime -> Job -> Html () +statusSuccess t Job{..} = do + span_ [ class_ "badge badge-success" ] $ "Success" + span_ [ class_ "job-run-time" ] $ do + let (d, s) = timeDuration jobCreatedAt jobUpdatedAt + abbr_ [ title_ (showText jobUpdatedAt) ] $ toHtml $ "Completed " <> humanReadableTime' t jobUpdatedAt <> ". " + abbr_ [ title_ (showText d <> " seconds")] $ toHtml $ "Took " <> s + +statusFailed :: UTCTime -> Job -> Html () +statusFailed t Job{..} = do + span_ [ class_ "badge badge-danger" ] $ "Failed" + span_ [ class_ "job-run-time" ] $ do + abbr_ [ title_ (showText jobUpdatedAt) ] $ toHtml $ "Failed " <> humanReadableTime' t jobUpdatedAt <> " after " <> show jobAttempts <> " attempts" + +statusFuture :: UTCTime -> Job -> Html () +statusFuture t Job{..} = do + span_ [ class_ "badge badge-secondary" ] $ "Future" + span_ [ class_ "job-run-time" ] $ do + abbr_ [ title_ (showText jobRunAt) ] $ toHtml $ humanReadableTime' t jobRunAt + +statusWaiting :: UTCTime -> Job -> Html () +statusWaiting t Job{..} = do + span_ [ class_ "badge badge-warning" ] $ "Waiting" + -- span_ [ class_ "job-run-time" ] ("Waiting to be picked up" :: Text) + +statusRetry :: UTCTime -> Job -> Html () +statusRetry t Job{..} = do + span_ [ class_ "badge badge-warning" ] $ toHtml $ "Retries (" <> show jobAttempts <> ")" + span_ [ class_ "job-run-time" ] $ do + abbr_ [ title_ (showText jobUpdatedAt) ] $ toHtml $ "Retried " <> humanReadableTime' t jobUpdatedAt <> ". " + abbr_ [ title_ (showText jobRunAt)] $ toHtml $ "Next retry in " <> humanReadableTime' t jobRunAt + +statusLocked :: UTCTime -> Job -> Html () +statusLocked t Job{..} = do + span_ [ class_ "badge badge-info" ] $ toHtml ("Locked" :: Text) + -- span_ [ class_ "job-run-time" ] $ do + -- abbr_ [ title_ (showText jobUpdatedAt) ] $ toHtml $ "Retried " <> humanReadableTime' t jobUpdatedAt <> ". " + -- abbr_ [ title_ (showText jobRunAt)] $ toHtml $ "Next retry in " <> humanReadableTime' t jobRunAt + +resultsPanel :: Routes -> UTCTime -> Filter -> [(Job, Html ())] -> Int -> Html () +resultsPanel routes@Routes{..} t filter@Filter{filterPage} js runningCount = do + div_ [ class_ "card mt-3" ] $ do + div_ [ class_ "card-header bg-secondary text-white" ] $ do + "Currently running " + span_ [ class_ "badge badge-primary badge-primary" ] $ toHtml (show runningCount) + div_ [ class_ "currently-running" ] $ div_ [ class_ "" ] $ table_ [ class_ "table table-striped table-hover" ] $ do + thead_ [ class_ "thead-dark"] $ do + tr_ $ do + th_ "Job status" + th_ "Job" + th_ [ style_ "min-width: 12em;" ] "Actions" + tbody_ $ do + forM_ js (jobRow routes t) + div_ [ class_ "card-footer" ] $ do + nav_ $ do + ul_ [ class_ "pagination" ] $ do + prevLink + nextLink + where + prevLink = do + let (extraClass, lnk) = case filterPage of + Nothing -> ("disabled", "") + Just (l, 0) -> ("disabled", "") + Just (l, o) -> ("", rFilterResults $ Just $ filter {filterPage = Just (l, max 0 $ o - l)}) + li_ [ class_ ("page-item previous " <> extraClass) ] $ do + a_ [ class_ "page-link", href_ lnk ] $ "Prev" + + nextLink = do + let (extraClass, lnk) = case filterPage of + Nothing -> + if (DL.length js) < 10 + then ("disabled", "") + else ("", (rFilterResults $ Just $ filter {filterPage = Just (10, 10)})) + Just (l, o) -> + if (DL.length js) < l + then ("disabled", "") + else ("", (rFilterResults $ Just $ filter {filterPage = Just (l, o + l)})) + li_ [ class_ ("page-item next " <> extraClass) ] $ do + a_ [ class_ "page-link", href_ lnk ] $ "Next" + +ariaExpanded_ :: Text -> Attribute +ariaExpanded_ v = makeAttribute "aria-expanded" v + diff --git a/test/Test.hs b/test/Test.hs --- a/test/Test.hs +++ b/test/Test.hs @@ -37,9 +37,7 @@ import qualified Data.Text as T import Data.Ord (comparing, Down(..)) import Data.Maybe (fromMaybe) -import qualified System.Random as R -import Data.IORef (IORef(..), atomicModifyIORef', newIORef, readIORef) -import Data.Either (isRight) + $(Aeson.deriveJSON Aeson.defaultOptions ''Seconds) main :: IO () @@ -200,6 +198,7 @@ t <- getCurrentTime j1 <- Job.createJob conn tname (PayloadSucceed $ 2 * Job.defaultPollingInterval) j2 <- Job.scheduleJob conn tname (PayloadSucceed 0) (addUTCTime (fromIntegral $ unSeconds $ Job.defaultPollingInterval) t) + liftIO $ putStrLn "created" pure (j1, j2) Pool.withResource appPool $ \conn -> do delaySeconds 1 @@ -222,9 +221,9 @@ testJobFailure appPool jobPool = testCase "job retry" $ withNewJobMonitor jobPool $ \tname -> Pool.withResource appPool $ \conn -> do Job{jobId} <- Job.createJob conn tname (PayloadAlwaysFail 0) - delaySeconds $ Seconds 20 + delaySeconds $ Seconds 15 Job{jobAttempts, jobStatus} <- ensureJobId conn tname jobId - assertEqual "Expecteding job to be in Failed status" Job.Failed jobStatus + assertEqual "Exepcting job to be in Failed status" Job.Failed jobStatus assertEqual ("Expecting job attempts to be 3. Found " <> show jobAttempts) 3 jobAttempts data JobEvent = JobStart @@ -235,29 +234,12 @@ testEverything appPool jobPool = testProperty "test everything" $ property $ do - jobPayloads <- forAll $ Gen.list (Range.linear 1 1000) payloadGen - jobsRef <- liftIO $ newIORef (Right 0 :: Either String Int, Map.empty :: Map.IntMap [(JobEvent, Job.Job)]) - ccRef <- liftIO $ newIORef True - ccControl <- forAll $ genConcurrencyControl ccRef + jobPayloads <- forAll $ Gen.list (Range.linear 300 1000) payloadGen + jobsMVar <- liftIO $ newMVar (Map.empty :: Map.IntMap [(JobEvent, Job.Job)]) let maxDelay = sum $ map (payloadDelay jobPollingInterval) jobPayloads completeRun = ((unSeconds maxDelay) `div` concurrencyFactor) + (2 * (unSeconds testPollingInterval)) - onJobEvent :: JobEvent -> (Int -> Int) -> Job.Job -> IO () - onJobEvent evt jobCountFn job@Job{jobId} = - void $ atomicModifyIORef' jobsRef $ \(ccViolated, jobMap) -> - let newCCViolated = (jobCountFn <$> ccViolated) >>= \activeJobCount -> - case ccControl of - Job.MaxConcurrentJobs mj -> - if activeJobCount > mj - then Left $ "Concurrent jobs should NOT be more than " <> show mj <> ". Found: " <> show activeJobCount - else Right activeJobCount - Job.UnlimitedConcurrentJobs -> Right activeJobCount - Job.DynamicConcurrency _ -> Right activeJobCount - in ( (newCCViolated, Map.insertWith (++) jobId [(evt, job)] jobMap) - , () - ) - shutdownAfter <- forAll $ Gen.choice [ pure $ Seconds completeRun -- Either we allow all jobs to be completed properly , (Seconds <$> (Gen.int $ Range.linear 1 completeRun)) -- Or, we shutdown early after a random number of seconds ] @@ -266,48 +248,37 @@ (defaults, cleanup) <- liftIO $ Test.defaultJobMonitor tname jobPool let jobMonitorSettings = defaults { Job.monitorJobRunner = jobRunner , Job.monitorTableName = tname - , Job.monitorOnJobStart = onJobEvent JobStart (1+) - , Job.monitorOnJobFailed = onJobEvent JobRetry (\x -> x-1) - , Job.monitorOnJobPermanentlyFailed = onJobEvent JobFailed (\x -> x-1) - , Job.monitorOnJobSuccess = onJobEvent JobSuccess (\x -> x-1) + , Job.monitorOnJobStart = onJobEvent JobStart jobsMVar + , Job.monitorOnJobFailed = onJobEvent JobRetry jobsMVar + , Job.monitorOnJobPermanentlyFailed = onJobEvent JobFailed jobsMVar + , Job.monitorOnJobSuccess = onJobEvent JobSuccess jobsMVar , Job.monitorDefaultMaxAttempts = 3 , Job.monitorPollingInterval = jobPollingInterval - , Job.monitorConcurrencyControl = ccControl } (jobs :: [Job]) <- withAsync - (liftIO $ Job.runJobMonitor jobMonitorSettings) - (const $ finally - (liftIO $ actualTest shutdownAfter jobPayloads tname jobsRef) - (liftIO cleanup)) + (liftIO $ Job.runJobMonitor jobMonitorSettings) + (const $ finally + (liftIO $ actualTest shutdownAfter jobPayloads tname jobsMVar) + (liftIO cleanup)) - (ccViolated, jobAudit) <- liftIO $ readIORef jobsRef + jobAudit <- takeMVar jobsMVar [(Only (lockedJobCount :: Int))] <- liftIO $ Pool.withResource appPool $ \conn -> PGS.query conn ("SELECT coalesce(count(id), 0) FROM " <> tname <> " where status=?") (Only Job.Locked) - annotate $ "Job-payload count: " <> show (DL.length jobPayloads) - annotate $ "Concurrency control: " <> show ccControl - annotate $ "Shutdown delay (sec): " <> show shutdownAfter - annotate $ "Complete-run time estimate (sec): " <> show completeRun - - -- No job should be in a locked state - 0 === lockedJobCount - -- ALL jobs should show up in the audit, which means they should have -- been attempted /at least/ once - -- when (shutdownAfter >= completeRun) $ (DL.sort $ map jobId jobs) === (DL.sort $ Map.keys jobAudit) + -- No job should be in a locked state + 0 === lockedJobCount -- No job should've been simultaneously picked-up by more than one -- worker True === (Map.foldl (\m js -> m && noRaceCondition js) True jobAudit) - -- Concurrency control should not have been violated - Hedgehog.assert $ isRight ccViolated - - -- liftIO $ print $ "Test passed with job-count = " <> show (length jobPayloads) + liftIO $ print $ "Test passed with job-count = " <> show (length jobPayloads) where @@ -318,18 +289,11 @@ jobPollingInterval = Seconds 2 - -- onJobEvent :: JobEvent - -- -> Job.ConcurrencyControl - -- -> IORef (Either String Int) - -- -> IORef (Int, Map.IntMap [(JobEvent, Job.Job)]) -> (Int -> Int) -> Job.Job -> IO () - -- onJobEvent evt ccViolatedRef jobsRef countFn job@Job{jobId} = do - -- void $ atomicModifyIORef' jobsRef $ \(activeJobCount, jobMap) -> - -- ( ( countFn activeJobCount, Map.insertWith (++) jobId [(evt, job)] jobMap ) - -- , () - -- ) + onJobEvent evt jobsMVar job@Job{jobId} = void $ modifyMVar_ jobsMVar $ \jobMap -> do + pure $ Map.insertWith (++) jobId [(evt, job)] jobMap - actualTest :: Seconds -> [JobPayload] -> Job.TableName -> IORef (Either String Int, Map.IntMap [(JobEvent, Job.Job)]) -> IO [Job] - actualTest shutdownAfter jobPayloads tname jobsRef = do + actualTest :: Seconds -> [JobPayload] -> Job.TableName -> MVar (Map.IntMap [(JobEvent, Job.Job)]) -> IO [Job] + actualTest shutdownAfter jobPayloads tname jobsMVar = do jobs <- forConcurrently jobPayloads $ \pload -> Pool.withResource appPool $ \conn -> liftIO $ Job.createJob conn tname pload @@ -339,13 +303,13 @@ if remaining == Seconds 0 then pure (Right $ Seconds 0) else do delaySeconds testPollingInterval - -- print $ "------- Polling (remaining = " <> show (unSeconds remaining) <> " sec)------" - (_, jobMap) <- liftIO $ readIORef jobsRef - x <- if (Map.foldl (\m js -> m && (isJobTerminalState js)) True jobMap) - then pure (Right $ Seconds 0) - else if remaining < testPollingInterval - then pure (Left $ "Timeout. Job count=" <> show (length jobPayloads) <> " shutdownAfter=" <> show shutdownAfter) - else pure $ Right (remaining - testPollingInterval) + print $ "------- Polling (remaining = " <> show (unSeconds remaining) <> " sec)------" + x <- withMVar jobsMVar $ \jobMap -> + if (Map.foldl (\m js -> m && (isJobTerminalState js)) True jobMap) + then pure (Right $ Seconds 0) + else if remaining < testPollingInterval + then pure (Left $ "Timeout. Job count=" <> show (length jobPayloads) <> " shutdownAfter=" <> show shutdownAfter) + else pure $ Right (remaining - testPollingInterval) poller x poller (Right shutdownAfter) >>= \case @@ -450,19 +414,7 @@ , filterRunAfter = runAfter } -dynamicConcurrencyFn :: IORef Bool -> IO Bool -dynamicConcurrencyFn ccRef = do - f <- R.randomIO - atomicModifyIORef' ccRef $ const (f, f) -genConcurrencyControl :: MonadGen m => IORef Bool -> m Job.ConcurrencyControl -genConcurrencyControl ccRef = Gen.choice - [ Job.MaxConcurrentJobs <$> (Gen.int $ Range.linear 1 20) - , pure Job.UnlimitedConcurrentJobs - , pure $ Job.DynamicConcurrency $ dynamicConcurrencyFn ccRef - ] - - filterJobs :: Filter -> [Job] -> [Job] filterJobs Web.Filter{filterStatuses, filterCreatedAfter, filterCreatedBefore, filterUpdatedAfter, filterUpdatedBefore, filterOrder, filterPage, filterRunAfter} js = applyLimitOffset $ @@ -509,6 +461,6 @@ tcache <- newTimeCache simpleTimeFormat' (tlogger, cleanup) <- newTimedFastLogger tcache LogNone let flogger loc lsource llevel lstr = tlogger $ \t -> toLogStr t <> " | " <> defaultLogStr loc lsource llevel lstr - pure ( Job.defaultJobMonitor flogger tname pool Job.UnlimitedConcurrentJobs + pure ( Job.defaultJobMonitor flogger tname pool , cleanup )