batchd 0.1.0.0 → 0.1.1.0
raw patch · 30 files changed
+6152/−127 lines, 30 filesdep +batchd-amazonkadep +batchd-coredep +batchd-dockerdep −monad-loggerdep −monad-logger-syslogdep −readlinedep ~containersdep ~data-defaultdep ~directory
Dependencies added: batchd-amazonka, batchd-core, batchd-docker, batchd-libvirt, batchd-linode, boxes, conduit, conduit-combinators, conduit-extra, ekg-core, ekg-json, exceptions, fast-logger, haskeline, heavy-logger, heavy-logger-instances, hsyslog, libssh2-conduit, localize, microlens, monad-metrics, text-format-heavy, unliftio-core, vector, wai-middleware-metrics
Dependencies removed: monad-logger, monad-logger-syslog, readline, template, template-haskell, transformers
Dependency ranges changed: containers, data-default, directory, esqueleto, libssh2, optparse-applicative, persistent, scotty, syb, time
Files
- batchd.cabal +142/−54
- src/Batchd/Client/Actions.hs +569/−0
- src/Batchd/Client/CmdLine.hs +368/−0
- src/Batchd/Client/Config.hs +122/−0
- src/Batchd/Client/Http.hs +244/−0
- src/Batchd/Client/Logging.hs +27/−0
- src/Batchd/Client/Monad.hs +95/−0
- src/Batchd/Client/Shell.hs +105/−0
- src/Batchd/Client/Types.hs +30/−0
- src/Batchd/Common/Config.hs +61/−0
- src/Batchd/Common/Data.hs +256/−0
- src/Batchd/Common/Schedule.hs +100/−0
- src/Batchd/Common/Types.hs +732/−0
- src/Batchd/Daemon/Auth.hs +475/−0
- src/Batchd/Daemon/CmdLine.hs +89/−0
- src/Batchd/Daemon/Crypto.hs +24/−0
- src/Batchd/Daemon/Database.hs +521/−0
- src/Batchd/Daemon/Dispatcher.hs +175/−0
- src/Batchd/Daemon/Executor.hs +115/−0
- src/Batchd/Daemon/Hosts.hs +290/−0
- src/Batchd/Daemon/Local.hs +91/−0
- src/Batchd/Daemon/Logging.hs +48/−0
- src/Batchd/Daemon/Manager.hs +665/−0
- src/Batchd/Daemon/Monitoring.hs +180/−0
- src/Batchd/Daemon/SSH.hs +164/−0
- src/Batchd/Daemon/Schedule.hs +62/−0
- src/Batchd/Daemon/Types.hs +316/−0
- src/batch.hs +12/−11
- src/batchd-admin.hs +35/−28
- src/batchd.hs +39/−34
batchd.cabal view
@@ -1,5 +1,5 @@ name: batchd-version: 0.1.0.0+version: 0.1.1.0 synopsis: Batch processing toolset for Linux / Unix description: The batchd is a toolset for batch processing for Linux / Unix operating systems. It enables one to:@@ -34,7 +34,6 @@ <https://github.com/portnov/batchd/blob/master/README.md README> and <https://github.com/portnov/batchd/wiki Wiki> on GitHub. - Homepage: https://github.com/portnov/batchd license: BSD3 license-file: LICENSE@@ -46,118 +45,205 @@ -- extra-source-files: cabal-version: >=1.10 +flag libvirt+ description: Enable support of libvirt host controller+ default: False++flag docker+ description: Enable support of docker host controller+ default: False++flag ec2+ description: Enable support of AWS EC2 host controller+ default: False++flag linode+ description: Enable support of Linode.com host conrtoller+ default: False+ executable batchd main-is: batchd.hs hs-source-dirs: src/- -- other-modules: + other-modules: Batchd.Common.Data+ Batchd.Common.Schedule+ Batchd.Daemon.Auth+ Batchd.Daemon.Crypto+ Batchd.Daemon.Database+ Batchd.Daemon.Dispatcher+ Batchd.Daemon.Executor+ Batchd.Daemon.Manager+ Batchd.Daemon.Hosts+ Batchd.Daemon.SSH+ Batchd.Daemon.Local+ Batchd.Daemon.Schedule+ Batchd.Common.Config+ Batchd.Common.Types+ Batchd.Daemon.Logging+ Batchd.Daemon.Types+ Batchd.Daemon.Monitoring+ Batchd.Daemon.CmdLine other-extensions: TypeFamilies, DeriveDataTypeable, TemplateHaskell, StandaloneDeriving, RecordWildCards build-depends: base >=4.7 && <5.0,- persistent >= 2.2,+ data-default >= 0.7.1,+ exceptions >= 0.8.3,+ conduit >= 1.2,+ conduit-extra >= 1.1,+ conduit-combinators >= 1.1,+ persistent >= 2.13.3.0, persistent-sqlite >= 2.2, persistent-postgresql >= 2.2, persistent-template >= 2,- esqueleto >= 2.4,- monad-logger >= 0.3,- monad-logger-syslog >= 0.1,+ esqueleto >= 3.5,+ hsyslog >= 5,+ text-format-heavy >= 0.1.5.3,+ heavy-logger >= 0.3.2.2,+ heavy-logger-instances >= 0.1.2.0,+ localize >= 0.2, data-default >= 0.5, mtl >=2.2 && <2.3,- transformers >= 0.5,+ unliftio-core >= 0.1.2.0, dates >=0.2 && <0.3, filepath >= 1.3,- directory >= 1.2, Glob >= 0.7.5,- time >=1.4 && <1.7,- syb >=0.6 && <0.7,- containers >=0.5 && <0.6,+ time >=1.4 && <1.10,+ syb >=0.6,+ containers >=0.5 && <0.7, unordered-containers >= 0.2,+ vector >= 0.12, resourcet >= 1.1.7, http-types >= 0.9, wai >= 3.0, wai-extra >= 3.0, wai-cors >= 0.2.5, wai-middleware-static >= 0.8.1,+ ekg-core >= 0.1.1.3,+ ekg-json >= 0.1.0.6,+ monad-metrics >= 0.2.1.2,+ wai-middleware-metrics >= 0.2.4,+ microlens >= 0.4.8.1, vault,- scotty >= 0.10,+ scotty >= 0.12, warp >= 3.2, aeson >= 0.11, yaml >= 0.8.4, text >= 1.2, parsec >= 3.1, bytestring >= 0.10,- optparse-applicative >= 0.13.2.0,- template >= 0.2,+ optparse-applicative >= 0.14.3.0, process >= 1.2.3.0,- libssh2 >= 0.2,+ libssh2 >= 0.2.0.8,+ libssh2-conduit >= 0.2.1, cryptonite >= 0.23,- template-haskell >= 2.8,- th-lift >= 0.6- -- hs-source-dirs: + directory >= 1.1.3.1,+ th-lift >= 0.6,+ batchd-core >= 0.1.0.0+ if flag(libvirt)+ cpp-options: -DLIBVIRT+ build-depends: batchd-libvirt++ if flag(docker)+ cpp-options: -DDOCKER+ build-depends: batchd-docker++ if flag(ec2)+ cpp-options: -DAWSEC2+ build-depends: batchd-amazonka++ if flag(linode)+ cpp-options: -DLINODE+ build-depends: batchd-linode+ ghc-options: -fwarn-unused-imports -fwarn-missing-signatures -threaded default-language: Haskell2010 executable batchd-admin main-is: batchd-admin.hs hs-source-dirs: src/- -- other-modules: + other-modules: + Batchd.Common.Data+ Batchd.Common.Schedule+ Batchd.Daemon.Auth+ Batchd.Daemon.Crypto+ Batchd.Daemon.Database+ Batchd.Daemon.Schedule+ Batchd.Common.Config+ Batchd.Common.Types+ Batchd.Daemon.Logging+ Batchd.Daemon.Types+ Batchd.Daemon.CmdLine other-extensions: TypeFamilies, DeriveDataTypeable, TemplateHaskell, StandaloneDeriving, RecordWildCards build-depends: base >=4.7 && <5.0,+ data-default >= 0.7.1,+ exceptions >= 0.8.3, persistent >= 2.2, persistent-sqlite >= 2.2, persistent-postgresql >= 2.2, persistent-template >= 2,- esqueleto >= 2.4,- monad-logger >= 0.3,- monad-logger-syslog >= 0.1,- data-default >= 0.5,+ esqueleto >= 3.5,+ hsyslog >= 5,+ heavy-logger >= 0.3.2.2,+ heavy-logger-instances >= 0.1.2.0,+ text-format-heavy >= 0.1.5.3,+ localize >= 0.2, mtl >=2.2 && <2.3,- transformers >= 0.5,+ unliftio-core >= 0.1.2.0, dates >=0.2 && <0.3,- filepath >= 1.3,- directory >= 1.2,- Glob >= 0.7.5,- time >=1.4 && <1.7,- syb >=0.6 && <0.7,- containers >=0.5 && <0.6,+ time >=1.4 && <1.10,+ syb >=0.6,+ containers >=0.5 && <0.7, unordered-containers >= 0.2,+ vector >= 0.12, resourcet >= 1.1.7, http-types >= 0.9, wai >= 3.0, wai-extra >= 3.0,+ ekg-core >= 0.1.1.3,+ ekg-json >= 0.1.0.6,+ monad-metrics >= 0.2.1.2,+ wai-middleware-metrics >= 0.2.4, vault, scotty >= 0.10,- warp >= 3.2, aeson >= 0.11, yaml >= 0.8.4, text >= 1.2, parsec >= 3.1, bytestring >= 0.10,- optparse-applicative >= 0.13.2.0,- template >= 0.2,- process >= 1.2.3.0,- libssh2 >= 0.2,+ optparse-applicative >= 0.14.3.0, cryptonite >= 0.23,- template-haskell >= 2.8,- th-lift >= 0.6+ th-lift >= 0.6,+ batchd-core >= 0.1.0.0 -- hs-source-dirs: ghc-options: -fwarn-unused-imports -fwarn-missing-signatures -threaded default-language: Haskell2010+ executable batch main-is: batch.hs hs-source-dirs: src/- -- other-modules: + other-modules: + Batchd.Client.Actions+ Batchd.Client.CmdLine+ Batchd.Client.Config+ Batchd.Client.Http+ Batchd.Client.Monad+ Batchd.Client.Shell+ Batchd.Client.Types+ Batchd.Client.Logging+ Batchd.Common.Data+ Batchd.Common.Schedule+ Batchd.Common.Config+ Batchd.Common.Types+ build-depends: base >=4.7 && <5.0, data-default >= 0.5, mtl >=2.2 && <2.3,- transformers >= 0.5,+ unliftio-core >= 0.1.2.0, dates >=0.2 && <0.3, filepath >= 1.3,- directory >= 1.2,- time >=1.4 && <1.7,- syb >=0.6 && <0.7,- containers >=0.5 && <0.6,+ time >=1.4 && <1.10,+ syb >=0.6,+ containers >=0.5 && <0.7, unordered-containers >= 0.2,- monad-logger >= 0.3,- readline >= 1.0.3.0,+ haskeline >= 0.8.2, x509-store >= 1.6.2, tls >= 1.3.10, connection >= 0.2.8,@@ -169,17 +255,19 @@ text >= 1.2, parsec >= 3.1, bytestring >= 0.10,- optparse-applicative >= 0.13.2.0,+ optparse-applicative >= 0.14.3.0, persistent >= 2.2, persistent-template >= 2,- persistent-sqlite >= 2.2,- persistent-postgresql >= 2.2,- esqueleto >= 2.4,- monad-logger >= 0.3,- monad-logger-syslog >= 0.1,- resourcet >= 1.1.7,+ hsyslog >= 5,+ text-format-heavy >= 0.1.5.2,+ heavy-logger >= 0.3.1.0,+ heavy-logger-instances >= 0.1.2.0,+ hsyslog >= 5,+ fast-logger >= 2.4,+ localize >= 0.2, unix >= 2.7.0,- scotty >= 0.10+ boxes >= 0.1.4,+ batchd-core >= 0.1.0.0 ghc-options: -fwarn-unused-imports -fwarn-missing-signatures default-language: Haskell2010
+ src/Batchd/Client/Actions.hs view
@@ -0,0 +1,569 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE DeriveGeneric #-}+-- | This module contains definitions of client actions.+-- These functions prepare REST request structure, execute HTTP request+-- and show response in human-readable form to stdout.+module Batchd.Client.Actions where++import Prelude hiding ((<>))+import Control.Monad+import Control.Monad.State+import Data.Int+import Data.Maybe+import Data.Time.Clock+import Data.Time.LocalTime+import qualified Data.Map as M+import qualified Data.Text as T+import qualified Data.Text.Lazy as TL+import qualified Data.Text.IO as TIO+import qualified Data.Text.Lazy.IO as TLIO+import Data.Text.Format.Heavy+import Data.Char+import Data.List (intercalate, transpose)+import Data.Aeson+import System.FilePath+import System.Exit+import Text.Printf+import Text.PrettyPrint.Boxes++import Batchd.Core.Common.Localize+import qualified Batchd.Common.Data as Database+import Batchd.Common.Types+import Batchd.Common.Schedule+import Batchd.Common.Config+import Batchd.Client.CmdLine+import Batchd.Client.Config+import Batchd.Client.Http+import Batchd.Client.Monad++localTimeToUTC' :: Maybe (Maybe LocalTime) -> Client (Maybe (Maybe UTCTime))+localTimeToUTC' Nothing = return Nothing+localTimeToUTC' (Just Nothing) = return (Just Nothing)+localTimeToUTC' (Just (Just local)) = do+ tz <- liftIO $ getCurrentTimeZone+ return $ Just $ Just $ localTimeToUTC tz local++-- | Put job to the queue.+doEnqueue :: Client ()+doEnqueue = do+ cfg <- gets csConfig+ baseUrl <- getBaseUrl+ creds <- getCredentials+ opts <- gets csCmdline+ t <- liftIO $ getTypeName opts cfg+ jtr <- liftIO $ loadTemplate (T.pack t)+ case jtr of+ Left err -> throwC =<< (__f "Can't load job type description: {}" (Single $ Shown err))+ Right jtype -> do+ qname <- liftIO $ getQueueName opts cfg+ host <- liftIO $ getHostName opts cfg+ jobStartTime <- join `fmap` localTimeToUTC' (startTime $ cmdCommand opts)++ let job = JobInfo {+ jiId = 0,+ jiQueue = qname,+ jiType = t,+ jiSeq = 0,+ jiUserName = fst creds,+ jiCreateTime = zeroUtcTime,+ jiStartTime = jobStartTime,+ jiStatus = New,+ jiTryCount = 0,+ jiHostName = host,+ jiNotes = jobNotes (cmdCommand opts),+ jiResultTime = Nothing,+ jiExitCode = Nothing,+ jiStdout = Nothing,+ jiStderr = Nothing,+ jiParams = parseParams (jtParams jtype) (cmdCommand opts)+ }+ -- debug (__ "Job to be queued: {}") (Single $ show job)++ let url = baseUrl </> "queue" </> qname+ doPost url job++mkTable :: [[String]] -> Box+mkTable table = hsep 1 top [vcat left (map text column) | column <- table]++printTable :: Int -> [[String]] -> IO ()+printTable indent table =+ printBox $ emptyBox 0 indent <> mkTable table++-- | List queues or jobs.+doList :: Client ()+doList = do+ baseUrl <- getBaseUrl+ opts <- gets csCmdline+ let command = cmdCommand opts+ case queueToList command of+ [] -> do+ let url = baseUrl </> "queue"+ response <- doGet url+ let queuesTable = flip map (response :: [Database.Queue]) $ \queue ->+ [(if Database.queueEnabled queue then "[*]" else "[ ]") :: String,+ Database.queueName queue,+ Database.queueTitle queue,+ Database.queueScheduleName queue,+ fromMaybe "*" $ Database.queueHostName queue,+ maybe "no" show $ Database.queueAutostartJobCount queue]+ + liftIO $ printTable 0 $ transpose queuesTable++ qnames ->+ forM_ qnames $ \qname -> do+ statusOpt <- parseStatus (Just New) (throwC =<< (__ "Invalid status")) (status command)+ let statusStr = case statusOpt of+ Nothing -> "?status=all"+ Just st -> case status command of+ Nothing -> ""+ _ -> "?status=" ++ map toLower (show st)+ let url = baseUrl </> "queue" </> qname </> "jobs" ++ statusStr+ response <- doGet url++ liftIO $ forM_ (response :: [JobInfo]) $ \job -> do+ printf "#%d: [%d]\t%s\t%s\t%s\n" (jiId job) (jiSeq job) (jiType job) (show $ jiStatus job) (fromMaybe "" $ jiNotes job)+ forM_ (M.assocs $ jiParams job) $ \(name, value) -> do+ printf "\t%s:\t%s\n" name value++-- | Retrieve statistics+doStats :: Client ()+doStats = do+ baseUrl <- getBaseUrl+ opts <- gets csCmdline+ let command = cmdCommand opts+ case queueToStat command of+ [] -> do+ let url = baseUrl </> "stats"+ response <- doGet url+ liftIO $ forM_ (M.assocs response) $ \record -> do+ let qname = fst record :: String+ stats = snd record+ + liftIO $ putStrLn $ qname ++ ":"+ liftIO $ printStats stats+ qnames -> do+ forM_ qnames $ \qname -> do+ liftIO $ putStrLn $ qname ++ ":"+ let url = baseUrl </> "stats" </> qname+ response <- doGet url+ liftIO $ printStats response+ where+ printStats :: ByStatus Int -> IO ()+ printStats (ByStatus stat) = do+ printTable 4 $ transpose $ [[show status ++ ":", show count] | (status, count) <- M.assocs stat]++printField :: Formatable value => TL.Text -> IO TL.Text -> value -> IO ()+printField sep ioName value = do+ name <- ioName+ TLIO.putStrLn $ format "{}{}:\t{}" (sep, name, value)++translateTable :: [(IO TL.Text, String)] -> IO [[String]]+translateTable pairs =+ forM pairs $ \(ioTitle, value) -> do+ title <- ioTitle+ return [TL.unpack title ++ ":", value]++translateTableT :: [(IO TL.Text, T.Text)] -> IO [[T.Text]]+translateTableT pairs =+ forM pairs $ \(ioTitle, value) -> do+ title <- ioTitle+ return [TL.toStrict title `T.append` ":", value]++-- | View job details or results.+viewJob :: Client ()+viewJob = do+ baseUrl <- getBaseUrl+ opts <- gets csCmdline+ let command = cmdCommand opts+ let showDescription = if viewDescription command+ then True+ else if not (viewResult command) && not (viewAll command)+ then True+ else viewDescription command+ if (showDescription || viewResult command) && not (viewAll command)+ then do+ let url = baseUrl </> "job" </> show (jobId command)+ response <- doGet url+ when (showDescription) $+ liftIO $ printJob response+ when (viewResult command) $+ liftIO $ printLastResult response+ else do+ when (showDescription) $ do+ let url = baseUrl </> "job" </> show (jobId command)+ response <- doGet url+ liftIO $ printJob response+ when (viewAll command) $ do+ let url = baseUrl </> "job" </> show (jobId command) </> "results"+ response <- doGet url+ let showEc Nothing = "-"+ showEc (Just ExitSuccess) = "0"+ showEc (Just (ExitFailure n)) = show n+ let responseTable = flip map (response :: [Database.JobResult]) $ \result ->+ [show (Database.jobResultTime result),+ showEc (Database.jobResultExitCode result),+ T.unpack (Database.jobResultStdout result)+ ]+ liftIO $ printTable 0 $ transpose responseTable+ where+ printJob :: JobInfo -> IO ()+ printJob job = do+ let host = fromMaybe "*" $ jiHostName job+ table <- translateTable $ [+ ((__ "Order"), show $ jiSeq job),+ ((__ "Type"), jiType job),+ ((__ "Queue"), jiQueue job),+ ((__ "Host"), host),+ ((__ "Notes"), fromMaybe "-" $ jiNotes job),+ ((__ "User"), jiUserName job),+ ((__ "Created"), show $ jiCreateTime job),+ ((__ "Start"), maybe "*" show $ jiStartTime job),+ ((__ "Status"), show $ jiStatus job),+ ((__ "Try count"), show $ jiTryCount job)+ ]+ printTable 0 $ transpose table+ let params = [[TL.unpack name ++ ":", T.unpack value] | (name, value) <- M.assocs (jiParams job)]+ printTable 4 $ transpose params++ printLastResult :: JobInfo -> IO ()+ printLastResult job = do+ let time = case jiResultTime job of+ Nothing -> "-"+ Just t -> show t+ let code = show (jiExitCode job)+ TLIO.putStrLn =<< (__f "Exit code:\t{}\nTime:\t{}\n\n" (code, time))+ case jiStdout job of+ Nothing -> return ()+ Just text -> putStrLn $ T.unpack text++ printResult :: Database.JobResult -> IO ()+ printResult r = do+ let time = show (Database.jobResultTime r)+ let code = show (Database.jobResultExitCode r)+ TLIO.putStrLn =<< (__f "Exit code:\t{}\nTime:\t{}\n\n" (code, time))+ putStrLn $ T.unpack $ Database.jobResultStdout r++-- checkIncompatibleOptions :: [(String, Bool)] -> Client ()+-- checkIncompatibleOptions opts =+-- when (length (filter snd opts) > 1) $ do+-- throwC $ printf "Only one option of %s can be specified at once" enabled+-- where+-- enabled = intercalate ", " $ map fst $ filter snd opts++checkModes :: [Client (Maybe a)] -> Client a+checkModes list = go Nothing list+ where+ go Nothing [] = throwC =<< (__ "No mode is selected")+ go (Just m) [] = return m+ go Nothing (m:ms) = do+ result <- m+ go result ms+ go selected@(Just _) (m:ms) = do+ result <- m+ case result of+ Nothing -> go selected ms+ Just _ -> throwC =<< (__ "Only one mode can be selected")++-- | Update job.+updateJob :: Client ()+updateJob = do+ baseUrl <- getBaseUrl+ opts <- gets csCmdline+ let command = cmdCommand opts+ job <- checkModes [mMove command, mPrioritize command, mUpdate command]+ let url = baseUrl </> "job" </> show (jobId command)+ doPut url job++ where+ mMove command = do+ case queueName command of+ Just qname -> return $ Just $ object ["move" .= qname]+ Nothing -> return Nothing++ mPrioritize command = do+ case prioritize command of+ Just action -> return $ Just $ object ["priority" .= action]+ Nothing -> return Nothing++ mUpdate command = do+ if isNothing (queueName command) && isNothing (prioritize command)+ then do+ jobStartTime <- localTimeToUTC' (startTime command)+ return $ Just $ object $+ toList "status" (status command) +++ toList "host_name" (hostName command) +++ toList "notes" (jobNotes command) +++ toList "start_time" jobStartTime+ else return Nothing++-- | Delete job.+deleteJob :: Client ()+deleteJob = do+ baseUrl <- getBaseUrl+ opts <- gets csCmdline+ let command = cmdCommand opts+ let url = baseUrl </> "job" </> show (jobId command)+ doDelete url++-- | Create queue.+addQueue :: Client ()+addQueue = do+ baseUrl <- getBaseUrl+ opts <- gets csCmdline+ let command = cmdCommand opts+ let queue = Database.Queue {+ Database.queueName = queueObject command,+ Database.queueTitle = fromMaybe (queueObject command) (title command),+ Database.queueEnabled = fromMaybe True (enabled command),+ Database.queueScheduleName = fromMaybe "anytime" (scheduleName command),+ Database.queueHostName = hostName command,+ Database.queueAutostartJobCount = join (autostartCount command)+ }+ let url = baseUrl </> "queue"+ doPost url queue++-- | Update queue.+updateQueue :: Client ()+updateQueue = do+ baseUrl <- getBaseUrl+ opts <- gets csCmdline+ let command = cmdCommand opts+ let queue = object $+ toList "enabled" (enabled command) +++ toList "title" (title command) +++ toList "schedule_name" (scheduleName command) +++ toList "host_name" (hostName command) ++ + toList "autostart_job_count" (autostartCount command)+ -- print queue+ let url = baseUrl </> "queue" </> queueObject command+ doPut url queue++toList :: (ToJSON v, KeyValue t) => T.Text -> Maybe v -> [t]+toList _ Nothing = []+toList name (Just str) = [name .= str]++-- | Delete queue.+deleteQueue :: Client ()+deleteQueue = do+ baseUrl <- getBaseUrl+ opts <- gets csCmdline+ let command = cmdCommand opts+ let forceStr = if force command then "?forced=true" else ""+ let url = baseUrl </> "queue" </> queueObject command ++ forceStr+ doDelete url++-- | List schedules+doListSchedules :: Client ()+doListSchedules = do+ baseUrl <- getBaseUrl+ opts <- gets csCmdline+ let command = cmdCommand opts+ let url = baseUrl </> "schedule"+ response <- doGet url+ let check = if null (scheduleNames command)+ then const True+ else \si -> sName si `elem` scheduleNames command+ liftIO $ forM_ (response :: [ScheduleInfo]) $ \si -> do+ when (check si) $ do+ putStrLn $ sName si ++ ":"+ case sWeekdays si of+ Nothing -> putStrLn "\tany weekday"+ Just lst -> putStrLn $ "\t" ++ intercalate ", " (map show lst)+ case sTime si of+ Nothing -> putStrLn "\tany time of day"+ Just lst -> putStrLn $ "\t" ++ intercalate ", " (map show lst)++-- | Create a schedule.+doAddSchedule :: Client ()+doAddSchedule = do+ baseUrl <- getBaseUrl+ opts <- gets csCmdline+ let command = cmdCommand opts+ when (length (scheduleNames command) /= 1) $+ throwC =<< (__ "Exactly one schedule name must be specified when creating a schedule")+ let ts = forM (periods command) $ \str -> do+ parsePeriod str+ case ts of+ Left err -> throwC =<< (__f "Can't parse period description: {}" (Single $ show err))+ Right times -> do+ let url = baseUrl </> "schedule"+ let schedule = ScheduleInfo {+ sName = head (scheduleNames command),+ sWeekdays = if null (weekdays command) then Nothing else Just (weekdays command),+ sTime = if null times then Nothing else Just times+ }+ doPost url schedule++-- | Delete schedule.+doDeleteSchedule :: Client ()+doDeleteSchedule = do+ baseUrl <- getBaseUrl+ opts <- gets csCmdline+ let command = cmdCommand opts+ when (length (scheduleNames command) /= 1) $+ throwC =<< (__ "Exactly one schedule name must be specified when deleting a schedule")+ let sname = head (scheduleNames command)+ let forceStr = if force command then "?forced=true" else ""+ let url = baseUrl </> "schedule" </> sname ++ forceStr+ doDelete url++-- | List job types.+doType :: Client ()+doType = do+ baseUrl <- getBaseUrl+ opts <- gets csCmdline+ let command = cmdCommand opts+ let url = baseUrl </> "type"+ response <- doGet url+ let check = if null (types command)+ then const True+ else \jt -> jtName jt `elem` types command+ liftIO $ forM_ (response :: [JobType]) $ \jt -> do+ when (check jt) $ do+ let title = fromMaybe (jtName jt) (jtTitle jt)+ putStrLn $ jtName jt ++ ":"+ table <- translateTable $+ [(__ "Title", title)] +++ translateTemplate (jtTemplate jt) +++ [(__ "On fail", show (jtOnFail jt)),+ (__ "Host", fromMaybe "*" (jtHostName jt))]+ printTable 4 $ transpose table+ paramsLine <- (__ "Parameters:")+ TLIO.putStrLn $ " " `TL.append` paramsLine+ forM_ (jtParams jt) $ \desc -> do+ paramsTable <- translateTable $ [+ (__ "Name", TL.unpack $ piName desc),+ (__ "Type", show (piType desc)),+ (__ "Title", TL.unpack $ piTitle desc),+ (__ "Default", T.unpack $ piDefault desc)+ ]+ let box = emptyBox 0 4 <> char '*' <+> mkTable (transpose paramsTable)+ printBox box+ where+ translateTemplate (line : lines) =+ (__ "Template", T.unpack line) :+ [(return "", T.unpack l) | l <- lines]++-- | List defined hosts.+doHosts :: Client ()+doHosts = do+ baseUrl <- getBaseUrl+ opts <- gets csCmdline+ let command = cmdCommand opts+ let url = baseUrl </> "host"+ response <- doGet url+ liftIO $ forM_ (response :: [T.Text]) $ \hostName -> do+ TIO.putStrLn hostName++doMonitor :: Client ()+doMonitor = do+ baseUrl <- getBaseUrl+ opts <- gets csCmdline+ let command = cmdCommand opts+ let monitor = baseUrl </> "monitor"+ url <- case metricTime command of+ LastSample -> case metricPrefix command of+ Nothing -> fail "metric name must be specified for --last mode"+ Just name -> return $ monitor </> name </> "last"+ CurrentSample -> + let byPrefix = case metricPrefix command of+ Nothing -> monitor </> "current"+ Just prefix -> monitor </> prefix </> "current"+ in return $ case metricView command of+ Plain -> byPrefix </> "plain"+ Tree -> byPrefix </> "tree"+ response <- doGet url+-- liftIO $ putStrLn $ show (response :: Value)+ liftIO $ forM_ (response :: [Database.MetricRecord]) $ \r -> do+ TIO.putStr $ Database.metricRecordName r `T.append` ": "+ case Database.metricRecordKind r of+ Counter -> TLIO.putStrLn =<< (__f "count: {value}" r)+ Gauge -> TLIO.putStrLn =<< (__f "value: {value}" r)+ Label -> TLIO.putStrLn =<< (__f "text: {text}" r)+ Distribution -> do+ TLIO.putStrLn =<< (__f "mean: {mean}; variance: {variance}; count: {count}; sum: {sum}; min: {min}; max: {max}" r)++-- | List users.+doListUsers :: Client ()+doListUsers = do+ baseUrl <- getBaseUrl+ opts <- gets csCmdline+ let url = baseUrl </> "user"+ response <- doGet url+ liftIO $ forM_ (response :: [String]) $ \name -> putStrLn name++-- | Create user+doAddUser :: Client ()+doAddUser = do+ baseUrl <- getBaseUrl+ opts <- gets csCmdline+ let command = cmdCommand opts+ let url = baseUrl </> "user"+ pwd <- liftIO $ getPassword2+ name <- case objectUserName command of+ [n] -> return n+ _ -> fail "user name must be provided"+ let user = UserInfo name pwd+ doPost url user++-- | Change user password+doChangePassword :: Client ()+doChangePassword = do+ baseUrl <- getBaseUrl+ opts <- gets csCmdline+ let command = cmdCommand opts+ creds <- getCredentials+ name <- case objectUserName command of+ [n] -> return n+ _ -> return $ fst creds+ let url = baseUrl </> "user" </> name+ pwd <- liftIO $ getPassword2+ let user = UserInfo name pwd+ doPut url user++-- | List user permissions+doListPermissions :: Client ()+doListPermissions = do+ baseUrl <- getBaseUrl+ opts <- gets csCmdline+ let command = cmdCommand opts+ let url = baseUrl </> "user" </> grantUserName command </> "permissions"+ response <- doGet url+ liftIO $ forM_ (response :: [(Int64, Database.UserPermission)]) $ \(id, perm) -> do+ table <- translateTable $ [+ (__ "ID", show id),+ (__ "Permission", show $ Database.userPermissionPermission perm),+ (__ "Queue", fromMaybe "*" $ Database.userPermissionQueueName perm),+ (__ "Job type", fromMaybe "*" $ Database.userPermissionTypeName perm),+ (__ "Host", T.unpack $ fromMaybe "*" $ Database.userPermissionHostName perm)+ ]+ printTable 0 $ transpose table+ putStrLn ""++-- | Add user permission+doAddPermission :: Client ()+doAddPermission = do+ baseUrl <- getBaseUrl+ opts <- gets csCmdline+ let command = cmdCommand opts+ let name = grantUserName command+ url = baseUrl </> "user" </> name </> "permissions"+ perm <- case permission command of+ Just p -> return $ Database.UserPermission name p (queueName command) (typeName command) (T.pack <$> hostName command)+ Nothing -> throwC =<< (__ "permission (-p) must be specified")+ doPost url perm++-- | Revoke user permission.+doRevokePermission :: Client ()+doRevokePermission = do+ baseUrl <- getBaseUrl+ opts <- gets csCmdline+ let command = cmdCommand opts+ let name = grantUserName command+ url <- case grantPermissionId command of+ Just permId -> return $ baseUrl </> "user" </> name </> "permissions" </> show permId+ Nothing -> throwC =<< (__ "permission ID (-i) must be specified")+ -- liftIO $ print url+ doDelete url+
+ src/Batchd/Client/CmdLine.hs view
@@ -0,0 +1,368 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE DeriveGeneric #-}+-- | This module contains definitions for client command-line parsing+-- by using optparse-applicative package.+module Batchd.Client.CmdLine where++import Data.Generics hiding (Generic)+import qualified Data.Map as M+import Data.Int+import Data.Time.LocalTime+import Data.Dates+import Data.Char (toLower)+import qualified Data.Text as T+import qualified Data.Text.Lazy as TL+import Options.Applicative+import System.Log.Heavy+import qualified Text.Parsec as Parsec ++import System.IO.Unsafe (unsafePerformIO)++import Batchd.Core.Common.Types+import Batchd.Common.Types+import Batchd.Common.Data (MoveAction (..))+import Batchd.Client.Types++data CmdLine = CmdLine {+ cmdCommon :: CommonOpts,+ cmdCommand :: Command+ } deriving (Eq, Show)++data CommonOpts = CommonOpts {+ managerUrl :: Maybe String,+ username :: Maybe String,+ password :: Maybe String,+ logLevel :: Level+ }+ deriving (Eq, Show)+ ++data Command =+ Enqueue {+ queueName :: Maybe String,+ typeName :: Maybe String,+ hostName :: Maybe String,+ startTime :: Maybe (Maybe LocalTime),+ jobNotes :: Maybe String,+ jobCommand :: [T.Text],+ parameters :: [String]+ }+ | List {+ status :: Maybe String,+ queueToList :: [String]+ }+ | Queue {+ queueMode :: CrudMode,+ queueObject :: String,+ scheduleName :: Maybe String,+ hostName :: Maybe String,+ title :: Maybe String,+ enabled :: Maybe Bool,+ autostartCount :: Maybe (Maybe Int),+ force :: Bool+ }+ | Job {+ jobId :: Int,+ queueName :: Maybe String,+ prioritize :: Maybe MoveAction,+ startTime :: Maybe (Maybe LocalTime),+ status :: Maybe String,+ hostName :: Maybe String,+ jobNotes :: Maybe String,+ viewDescription :: Bool,+ viewResult :: Bool,+ viewAll :: Bool,+ jobMode :: CrudMode+ }+ | Schedule {+ scheduleMode :: CrudMode,+ scheduleNames :: [String],+ periods :: [String],+ weekdays :: [WeekDay],+ force :: Bool+ }+ | Monitor {+ metricPrefix :: Maybe String,+ metricView :: MetricView,+ metricTime :: MetricTime+ }+ | User {+ userMode :: CrudMode,+ objectUserName :: [String]+ }+ | Grant {+ grantMode :: CrudMode,+ grantUserName :: String,+ grantPermissionId :: Maybe Int64,+ permission :: Maybe Permission,+ queueName :: Maybe String,+ typeName :: Maybe String,+ hostName :: Maybe String+ }+ | Type {+ types :: [String]+ }+ | Stats {+ queueToStat :: [String]+ }+ | Hosts+ | RunShell+ deriving (Eq, Show, Data, Typeable)++data MetricTime = CurrentSample | LastSample+ deriving (Eq, Show, Data, Typeable)++data MetricView = Tree | Plain+ deriving (Eq, Show, Data, Typeable)++defaultUrl :: String+defaultUrl = "http://localhost:" ++ show defaultManagerPort++defaultQueue :: String+defaultQueue = "default"++defaultType :: String+defaultType = "command"++-- managerUrlAnn = Nothing &= name "url" &= typ defaultUrl &= help "batchd manager API URL"++-- usernameAnn = Nothing &= name "user" &= typ "USER" &= help "batchd user name"++-- passwordAnn = Nothing &= name "password" &= typ "PASSWORD" &= help "batchd user password"++parser :: Parser CmdLine+parser = CmdLine <$> commonOpts <*> commands++commonOpts :: Parser CommonOpts+commonOpts = CommonOpts+ <$> optional (strOption+ ( long "manager-url"+ <> short 'M'+ <> metavar "URL"+ <> help "batchd manager API URL"))+ <*> optional (strOption+ ( long "username"+ <> short 'U'+ <> metavar "NAME"+ <> help "batchd user name"))+ <*> optional (strOption+ ( long "password"+ <> short 'P'+ <> metavar "PASSWORD"+ <> help "batchd user password"))+ <*> verbosity++verbosity :: Parser Level+verbosity =+ flag' debug_level (short 'd' <> long "debug" <> help "enable client debug output")+ <|> flag' verbose_level (short 'v' <> long "verbose" <> help "be verbose")+ <|> flag info_level disable_logging (short 'q' <> long "quiet" <> help "be quiet")++pMetricView :: Parser MetricView+pMetricView =+ flag Plain Tree (short 't' <> long "tree" <> help "output data in tree view")++pMetricTime :: Parser MetricTime+pMetricTime =+ flag CurrentSample LastSample (short 'l' <> long "last" <> help "retrieve last metrics sample from database")++required :: Read a => String -> Char -> String -> String -> Parser a+required longName shortName meta helpText =+ option auto (long longName <> short shortName <> metavar meta <> help helpText)++optionalF :: Read a => String -> Char -> String -> String -> Parser (Maybe a)+optionalF longName shortName meta helpText =+ optional $ required longName shortName meta helpText++requiredString :: String -> Char -> String -> String -> Parser String+requiredString longName shortName meta helpText =+ strOption (long longName <> short shortName <> metavar meta <> help helpText)++optionalString :: String -> Char -> String -> String -> Parser (Maybe String)+optionalString longName shortName meta helpText =+ optional $ requiredString longName shortName meta helpText++enqueue :: Parser Command+enqueue = Enqueue+ <$> optionalString "queue" 'q' "QUEUE" "queue name"+ <*> optionalString "type" 't' "TYPE" "job type name"+ <*> optionalString "host" 'h' "HOST" "worker host name"+ <*> (optional $ option timeReader (long "at" <> long "start" <> metavar "YYYY-MM-DD HH:MM:SS" <> help "set job start time"))+ <*> optionalString "notes" 'n' "TEXT" "job description"+ <*> many (strArgument (metavar "COMMAND"))+ <*> many (requiredString "parameter" 'p' "NAME=VALUE" "job parameters specified by name")++list :: Parser Command+list = List+ <$> (optionalString "status" 's' "STATUS" "list only jobs of specified status"+ <|> flag Nothing (Just "all") (long "all" <> short 'a' <> help "synonym for --status=all"))+ <*> many (strArgument (metavar "QUEUE" <> help "queue to list"))++crudMode :: [(CrudMode, String)] -> Parser CrudMode+crudMode [] = error "crudMode called with empty list"+crudMode modes@((dflt,_):_) = foldr1 (<|>) $ map go modes+ where+ go (mode, helpText) =+ let (longName, shortName) = case mode of+ View -> ("view", 'v')+ Add -> ("add", 'a')+ Update -> ("update", 'u')+ Delete -> ("delete", 'd')+ in flag dflt mode+ ( long longName + <> short shortName+ <> help helpText )++priorityReader :: ReadM MoveAction+priorityReader = maybeReader (check . map toLower)+ where+ check s+ | s `elem` ["first", "f"] = Just First+ | s `elem` ["more", "up", "u"] = Just More+ | s `elem` ["less", "down", "d"] = Just Less+ | s `elem` ["last", "l"] = Just Last+ | otherwise = Nothing++startupTime :: DateTime+startupTime = unsafePerformIO $ getCurrentDateTime+{-# NOINLINE startupTime #-}++timeReader :: ReadM (Maybe LocalTime)+timeReader = (maybeReader nothing) <|> (Just <$> toUtc <$> eitherReader readDateTime)+ where+ toUtc :: DateTime -> LocalTime+ toUtc dt = LocalTime day time+ where+ day = dateTimeToDay dt+ time = TimeOfDay (hour dt) (minute dt) (fromIntegral $ second dt)++ nothing :: String -> Maybe (Maybe LocalTime)+ nothing "any" = Just Nothing+ nothing "now" = Just Nothing+ nothing "anytime" = Just Nothing+ nothing _ = Nothing++ readDateTime :: String -> Either String DateTime+ readDateTime str = case Parsec.runParser (do {t <- pDateTime startupTime; Parsec.eof; return t}) () "<start time>" str of+ Left err -> Left (show err)+ Right time -> Right time++queue :: Parser Command+queue = Queue+ <$> crudMode [(Update, "modify queue"), (Add, "create new queue"), (Delete, "delete queue")]+ <*> strArgument (metavar "QUEUE" <> help "queue to operate on")+ <*> optionalString "schedule" 's' "SCHEDULE" "queue schedule name"+ <*> optionalString "host" 'h' "HOST" "default host name for queue"+ <*> optionalString "name" 'n' "TITLE" "set queue title"+ <*> ( flag Nothing (Just True) (long "enable" <> short 'e' <> help "enable queue")+ <|> flag Nothing (Just False) (long "disable" <> short 'D' <> help "disable queue")+ )+ <*> (optional $ option autostartReader (long "autostart" <> metavar "N"+ <> help "automatically enable queue if it has N new jobs. Use `--autostart no' to disable."))+ <*> switch (long "force" <> short 'f' <> help "force non-empty queue deletion")++autostartReader :: ReadM (Maybe Int)+autostartReader = (Just <$> auto) <|> maybeReader disable+ where+ disable :: String -> Maybe (Maybe Int)+ disable "no" = Just Nothing+ disable "disable" = Just Nothing+ disable "false" = Just Nothing+ disable _ = Nothing++job :: Parser Command+job = Job+ <$> argument auto (metavar "ID" <> help "job ID")+ <*> optionalString "move" 'm' "QUEUE" "move job to other queue"+ <*> (optional $ option priorityReader (long "priority" <> short 'p' <> metavar "ACTION" <> help "change job priority. ACTION is one of: up, down, first, last"))+ <*> (optional $ option timeReader (long "at" <> long "start" <> metavar "YYYY-MM-DD HH:MM:SS" <> help "set job start time. Use `any' to specify that job can be run at any time."))+ <*> optionalString "status" 's' "STATUS" "set job status"+ <*> optionalString "host" 'h' "HOST" "set job host"+ <*> optionalString "notes" 'n' "TEXT" "job description"+ <*> switch (long "description" <> help "view job description")+ <*> switch (long "result" <> short 'r' <> help "view job result")+ <*> switch (long "all" <> short 'a' <> help "view all job results")+ <*> crudMode [(View, "view job description or result"), (Update, "modify job"), (Delete, "delete job")]+ +schedule :: Parser Command+schedule = Schedule+ <$> crudMode [(View, "list available schedules"), (Add, "create new schedule"), (Update, "modify schedule"), (Delete, "delete (unused) schedule")]+ <*> many (strArgument (metavar "SCHEDULE" <> help "schedules to operate on"))+ <*> many (requiredString "period" 'p' "HH:MM:SS HH:MM:SS" "time of day period(s)")+ <*> many (required "weekday" 'w' "WEEKDAY" "week day(s)")+ <*> switch (long "force" <> short 'f' <> help "delete also all queues which use this schedule and their jobs")++user :: Parser Command+user = User+ <$> crudMode [(View, "list existing users"), (Add, "create new user"), (Update, "change user password")]+ <*> many (strArgument (metavar "NAME" <> help "name of user to operate on"))++grant :: Parser Command+grant = Grant+ <$> crudMode [(View, "view permissions of user"), (Add, "add permissions to user"), (Delete, "revoke permission from user")]+ <*> strArgument (metavar "NAME" <> help "name of user to operate on")+ <*> optionalF "id" 'i' "ID" "permission ID to operate on. mandatory for revoke operation."+ <*> optionalF "permission" 'p' "PERMISSION" "specify permission, for example ViewJobs"+ <*> optionalString "queue" 'q' "QUEUE" "queue to grant permission to, by default - any"+ <*> optionalString "type" 't' "TYPE" "job type to grant permission to, usable for CreateJobs permission"+ <*> optionalString "host" 'h' "HOST" "name of host to grant permission to, usable for CreateJobs permission. Use `__default__' value to allow default host of queue."++typesList :: Parser Command+typesList = Type+ <$> many (strArgument (metavar "TYPE"))++stats :: Parser Command+stats = Stats+ <$> many (strArgument (metavar "QUEUE"))++monitor :: Parser Command+monitor = Monitor+ <$> (optional $ strArgument (metavar "PREFIX" <> help "prefix of names of metrics to retrieve, for example batchd.wai"))+ <*> pMetricView+ <*> pMetricTime++shell :: Parser Command+shell = pure RunShell++hostsList :: Parser Command+hostsList = pure Hosts++parseParams :: [ParamDesc] -> Command -> JobParamInfo+parseParams desc e = + let posNames = map piName desc+ ordered = M.fromList $ zip posNames (jobCommand e)+ byName = M.fromList $ map parseOne (parameters e)+ in M.union byName ordered+ where+ parseOne :: String -> (TL.Text, T.Text)+ parseOne s = case break (== '=') s of+ (key, (_:value)) -> (TL.pack key, T.pack value)+ (key, []) -> (TL.pack key, T.empty)++commands :: Parser Command+commands = hsubparser + ( cmd "enqueue" enqueue "put a new job into queue"+ <> cmd "ls" list "list queues or jobs"+ <> cmd "job" job "update or delete jobs"+ <> cmd "queue" queue "create, update or delete queues"+ <> cmd "schedule" schedule "create, update or delete schedules"+ <> cmd "monitor" monitor "retrieve monitoring metrics"+ <> cmd "type" typesList "view job types"+ <> cmd "host" hostsList "view defined hosts"+ <> cmd "stats" stats "print statistics on queue or on all jobs"+ <> cmd "user" user "create, update or delete users"+ <> cmd "grant" grant "create, update or delete user permissions"+ <> cmd "shell" shell "run interactive shell" )+ where+ cmd name func helpText = command name (info func (progDesc helpText))++parserInfo :: ParserInfo CmdLine+parserInfo = info (parser <**> helper)+ (fullDesc+ <> header "batch - the batchd toolset command-line client program"+ <> progDesc "operate on batch jobs, queues, schedules etc.")++getCmdArgs :: IO CmdLine+getCmdArgs = execParser parserInfo+
+ src/Batchd/Client/Config.hs view
@@ -0,0 +1,122 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE DeriveGeneric #-}+-- | This module contains definitions for dealing with+-- client configuration file.+module Batchd.Client.Config where++import GHC.Generics+import Control.Exception+import Data.Yaml+import Data.Generics hiding (Generic)+import Data.Maybe+import Data.Text.Format.Heavy+import Text.Localize+import Text.Localize.IO () -- import instances only+import System.Environment+import System.Posix.User (getLoginName)++import Batchd.Core.Common.Config+import Batchd.Client.Types+import Batchd.Client.CmdLine++-- | Client configuration+data ClientConfig = ClientConfig {+ ccManagerUrl :: Maybe String -- ^ batchd manager URL+ , ccQueue :: Maybe String -- ^ Default queue to put jobs to+ , ccType :: Maybe String -- ^ Default job type+ , ccHost :: Maybe String -- ^ Default host+ , ccDisableAuth :: Bool -- ^ True if authentication is disabled+ , ccCertificate :: Maybe FilePath -- ^ Path to client HTTPS certificate file+ , ccKey :: Maybe FilePath -- ^ Path to client HTTPS private key file+ , ccCaCertificate :: Maybe FilePath -- ^ Path to CA certificate for server certificate check+ , ccDisableServerCertificateCheck :: Bool -- ^ If true, then check of client certificate is disabled+ , ccUsername :: Maybe String -- ^ User name+ , ccPassword :: Maybe String -- ^ Password+ }+ deriving (Show, Data, Typeable, Generic)++-- | Default client configuration+defaultConfig :: ClientConfig+defaultConfig = ClientConfig {+ ccManagerUrl = Nothing,+ ccQueue = Nothing,+ ccType = Nothing,+ ccHost = Nothing,+ ccDisableAuth = False,+ ccCertificate = Nothing,+ ccKey = Nothing,+ ccCaCertificate = Nothing,+ ccDisableServerCertificateCheck = False,+ ccUsername = Nothing,+ ccPassword = Nothing+ }++instance FromJSON ClientConfig where+ parseJSON (Object v) =+ ClientConfig+ <$> v .:? "manager_url"+ <*> v .:? "queue"+ <*> v .:? "type"+ <*> v .:? "host"+ <*> v .:? "disable_auth" .!= False+ <*> v .:? "certificate"+ <*> v .:? "key"+ <*> v .:? "ca_certificate"+ <*> v .:? "disable_server_certificate_check" .!= False+ <*> v .:? "username"+ <*> v .:? "password"++getConfigParam :: Maybe String -> String -> Maybe String -> String -> IO String+getConfigParam cmdline varname cfg dflt = do+ case cmdline of+ Just val -> return val+ Nothing -> do+ mbEnv <- lookupEnv varname+ case mbEnv of+ Just val -> return val+ Nothing -> return $ fromMaybe dflt cfg+ +getConfigParam' :: Maybe String -> String -> Maybe String -> IO (Maybe String)+getConfigParam' cmdline varname cfg = do+ case cmdline of+ Just val -> return $ Just val+ Nothing -> do+ mbEnv <- lookupEnv varname+ case mbEnv of+ Just val -> return $ Just val+ Nothing -> return cfg++-- | Load client configuration file.+loadClientConfig :: IO ClientConfig+loadClientConfig = do+ mbPath <- locateConfig "" "client.yaml"+ case mbPath of+ Nothing -> return defaultConfig+ Just path -> do+ r <- decodeFileEither path+ case r of+ Left err -> throw =<< ClientException `fmap` (__f "Can't parse client configuration file:\n{}" (Single $ show err))+ Right cfg -> return cfg+ +getManagerUrl :: CmdLine -> ClientConfig -> IO String+getManagerUrl (CmdLine o _) cfg =+ getConfigParam (managerUrl o) "BATCH_MANAGER_URL" (ccManagerUrl cfg) defaultUrl++getHostName :: CmdLine -> ClientConfig -> IO (Maybe String)+getHostName (CmdLine _ c) cfg =+ getConfigParam' (hostName c) "BATCH_HOST" (ccHost cfg)++getTypeName :: CmdLine -> ClientConfig -> IO String+getTypeName (CmdLine _ c) cfg =+ getConfigParam (typeName c) "BATCH_TYPE" (ccType cfg) defaultType++getQueueName :: CmdLine -> ClientConfig -> IO String+getQueueName (CmdLine _ c) cfg =+ getConfigParam (queueName c) "BATCH_QUEUE" (ccQueue cfg) defaultQueue++getUserName :: CmdLine -> ClientConfig -> IO String+getUserName (CmdLine o _) cfg = do+ osuser <- getLoginName+ getConfigParam (username o) "BATCH_USERNAME" (ccUsername cfg) osuser+
+ src/Batchd/Client/Http.hs view
@@ -0,0 +1,244 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE DeriveGeneric #-}+-- | This module contains definitions for HTTP\/HTTPS protocol handling.+module Batchd.Client.Http where++import Control.Exception+import Control.Monad.State+import qualified Data.ByteString.Lazy as L+import qualified Data.Text.Lazy.Encoding as TLE+import Data.Text.Format.Heavy+import Data.Default+import Data.List+import Data.Aeson as Aeson+import Data.X509.CertificateStore+import Network.HTTP.Client+import Network.HTTP.Types.Status+import Network.HTTP.Client.TLS+import Network.TLS as TLS+import Network.TLS.Extra.Cipher+import Network.Connection++import Batchd.Core.Common.Types+import Batchd.Core.Common.Localize+import Batchd.Common.Types+import Batchd.Common.Config (getPassword)+import Batchd.Client.Types as C+import Batchd.Client.Config+import Batchd.Client.CmdLine+import Batchd.Client.Monad+import Batchd.Client.Logging++-- | Create a Manager - a structure which contains all required information to+-- maintain HTTP connection, over TLS if required.+makeClientManager :: CmdLine -> Client Manager+makeClientManager opts = do+ cfg <- liftIO loadClientConfig+ url <- liftIO $ getManagerUrl opts cfg+ if "https" `isPrefixOf` url+ then mkMngr (ccDisableServerCertificateCheck cfg) (ccCertificate cfg) (ccKey cfg) (ccCertificate cfg)+ else liftIO $ newManager defaultManagerSettings -- for HTTP everything is easy+ where+ mkMngr :: Bool -> Maybe FilePath -> Maybe FilePath -> Maybe FilePath -> Client Manager+ mkMngr disableServerCertCheck mbCertFile mbKeyFile mbCaFile = do+ when disableServerCertCheck $ do+ verbose (__ "Server certificate check disabled") ()+ -- load Credentials from certificate and private key file (.pem and .key usually)+ creds <- case (mbCertFile, mbKeyFile) of+ (Just certFile, Just keyFile) -> do+ verbose (__ "Loading HTTPS certificate: {}, key: {}") (certFile, keyFile)+ r <- liftIO $ credentialLoadX509 certFile keyFile+ case r of+ Right cr -> return $ Just cr+ Left err -> fail =<< (__sf "Can't load certificate/key: {}" (Single $ show err))+ _ -> do+ message (__ "Certificate or key file is not specified, try to access HTTPS without them") ()+ return Nothing+ -- load trusted CA store if specified+ mbStore <- case mbCaFile of+ Nothing -> return Nothing+ Just caFile -> do+ r <- liftIO $ readCertificateStore caFile+ case r of+ Just store -> return $ Just store+ Nothing -> fail =<< (__s "cannot read specified CA store")+ let shared = case mbStore of+ Nothing -> def+ Just store -> def { sharedCAStore = store }+ + let clientCertHook _ = return creds+ skip _ _ _ _ = do+ return []++ -- handlers for TLS protocol events on client side+ hooks =+ if disableServerCertCheck+ then def {+ onCertificateRequest = clientCertHook, -- return loaded client certificate+ onServerCertificate = skip+ }+ else def { onCertificateRequest = clientCertHook }++ clientParams = (defaultParamsClient "" "")+ { clientHooks = hooks,+ clientUseServerNameIndication = True, -- SNI+ clientShared = shared,+ -- This is probably to be configurable+ clientSupported = def { supportedCiphers = ciphersuite_all, supportedVersions = [TLS12] }+ }+ tlsSettings = TLSSettings clientParams++ liftIO $ newManager $ mkManagerSettings tlsSettings Nothing++-- | Obtain list of authentication methods supported by server.+-- This is done by requesting OPTIONS /.+getAuthMethods :: Client [AuthMethod]+getAuthMethods = do+ mbMethods <- gets csAuthMethods+ case mbMethods of+ Just methods -> return methods+ Nothing -> do+ baseUrl <- getBaseUrl+ methods <- doOptions baseUrl+ modify $ \st -> st {csAuthMethods = Just methods}+ return methods++-- | Obtain user name and password if required+obtainCredentials :: Client C.Credentials+obtainCredentials = do+ cfg <- gets csConfig+ opts@(CmdLine o _) <- gets csCmdline+ methods <- getAuthMethods+ verbose (__ "Supported authentication methods: {}") (Single $ show methods)+ let needPassword = BasicAuth `elem` methods -- only basic auth of currently supported methods requires a password.+ name <- liftIO $ getUserName opts cfg+ pass <- if ccDisableAuth cfg || not needPassword+ then do+ debug (__ "Password is not required") ()+ return ""+ else do+ mbPassword <- liftIO $ getConfigParam' (password o) "BATCH_PASSWORD" (ccPassword cfg)+ case mbPassword of+ Just p -> return p+ Nothing -> liftIO $ getPassword =<< (__sf "{} password: " (Single name))+ let creds = (name, pass)+ -- remember credentials in client state+ modify $ \st -> st {csCredentials = Just creds}+ return creds++-- | Obain and cache user name and password, if required+getCredentials :: Client C.Credentials+getCredentials = do+ mbCreds <- gets csCredentials+ case mbCreds of+ Just creds -> return creds+ Nothing -> obtainCredentials++-- | Add authentication headers to HTTP request+applyAuth :: Request -> Client Request+applyAuth rq = do+ methods <- getAuthMethods+ -- liftIO $ print methods+ if BasicAuth `elem` methods+ then do+ (name, password) <- getCredentials+ return $ applyBasicAuth (stringToBstr name) (stringToBstr password) rq+ else if HeaderAuth `elem` methods+ then do+ (name, _) <- getCredentials+ return $ rq {requestHeaders = ("X-Auth-User", stringToBstr name) : requestHeaders rq}+ else return rq++handleStatus :: Response L.ByteString -> Client L.ByteString+handleStatus rs = do+ debug (__ "Received response: {}") (Single $ Shown rs)+ verbose (__ "Server response code is {} {}") (statusCode $ responseStatus rs, statusMessage $ responseStatus rs)+ if responseStatus rs == ok200+ then return $ responseBody rs+ else do+ liftIO $ throw $ ClientException $ TLE.decodeUtf8 $ responseBody rs++allowAny :: Request -> Response BodyReader -> IO ()+allowAny _ _ = return ()++getManager :: Client Manager+getManager = do+ mbManager <- gets csManager+ case mbManager of+ Just manager -> return manager+ Nothing -> do+ cmdline <- gets csCmdline+ manager <- makeClientManager cmdline+ modify $ \st -> st {csManager = Just manager}+ return manager++-- Generic HTTP request wrapper, useful mostly for debugging+doHttp :: Request -> Manager -> Client L.ByteString+doHttp request manager = do+ debug (__ "Sending request: {}") (Single $ Shown request)+ handleStatus =<< (liftIO $ httpLbs request manager)++-- PUT request+doPut :: ToJSON a => String -> a -> Client ()+doPut urlStr object = do+ manager <- getManager+ url <- liftIO $ parseUrlThrow urlStr+ let json = Aeson.encode object+ debug (__ "Request JSON: {}") (Single json)+ request <- applyAuth $ url {+ method="PUT",+ checkResponse = allowAny,+ requestBody = RequestBodyLBS json+ }+ doHttp request manager+ return ()++-- POST request+doPost :: ToJSON a => String -> a -> Client ()+doPost urlStr object = do+ manager <- getManager+ url <- liftIO $ parseUrlThrow urlStr+ let json = Aeson.encode object+ debug (__ "Request JSON: {}") (Single json)+ request <- applyAuth $ url {+ method="POST",+ checkResponse = allowAny,+ requestBody = RequestBodyLBS json+ }+ doHttp request manager+ return ()++-- DELETE request+doDelete :: String -> Client ()+doDelete urlStr = do+ manager <- getManager+ url <- liftIO $ parseUrlThrow urlStr+ request <- applyAuth $ url { method="DELETE",+ checkResponse = allowAny+ }+ doHttp request manager+ return ()++-- GET request+doGet :: FromJSON a => String -> Client a+doGet urlStr = do+ manager <- getManager+ url <- liftIO $ parseUrlThrow urlStr+ request <- applyAuth $ url {checkResponse = allowAny}+ responseLbs <- doHttp request manager+ case Aeson.eitherDecode responseLbs of+ Left err -> throwC =<< (__f "Can't parse server response for GET request: {}" (Single err))+ Right res -> return res++-- OPTIONS request+doOptions :: FromJSON a => String -> Client a+doOptions urlStr = do+ manager <- getManager+ url <- liftIO $ parseUrlThrow urlStr+ let request = url {checkResponse = allowAny, method = "OPTIONS"}+ responseLbs <- doHttp request manager+ case Aeson.eitherDecode responseLbs of+ Left err -> throwC =<< (__f "Can't parse server response for OPTIONS request: {}" (Single err))+ Right res -> return res+
+ src/Batchd/Client/Logging.hs view
@@ -0,0 +1,27 @@+{-# LANGUAGE FlexibleInstances #-}+-- | This module contains definitions for logging in client.+-- Log messages in client are localized.+module Batchd.Client.Logging where++import qualified Data.Text.Lazy as TL+import Data.Text.Format.Heavy+import System.Log.Heavy++import Batchd.Core.Common.Types+import Batchd.Client.Monad++putMessage :: (ClosedVarContainer vars) => Level -> Client TL.Text -> vars -> Client ()+putMessage v msg vars = do+ localizedMessage <- msg+ let msg = LogMessage v [] undefined localizedMessage vars []+ logMessage' msg++message :: ClosedVarContainer vars => Client TL.Text -> vars -> Client ()+message msg vars = putMessage info_level msg vars++verbose :: ClosedVarContainer vars => Client TL.Text -> vars -> Client ()+verbose msg vars = putMessage verbose_level msg vars++debug :: ClosedVarContainer vars => Client TL.Text -> vars -> Client ()+debug msg vars = putMessage debug_level msg vars+
+ src/Batchd/Client/Monad.hs view
@@ -0,0 +1,95 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE RankNTypes #-}+-- | This module contains defintion of @Client@ data type and utility functions for it.+module Batchd.Client.Monad where++import Control.Monad.State+import Control.Exception+import qualified Data.Text.Lazy as TL+import Network.HTTP.Client+import System.Log.Heavy+import qualified System.Log.FastLogger as FL++import Batchd.Core.Common.Localize+import Batchd.Common.Types+import Batchd.Client.Config+import Batchd.Client.Types+import Batchd.Client.CmdLine++-- | Client state+data ClientState = ClientState {+ csCmdline :: CmdLine -- ^ Parsed command line+ , csConfig :: ClientConfig -- ^ Loaded configuration file+ , csCredentials :: Maybe Credentials -- ^ Cached user credentials+ , csAuthMethods :: Maybe [AuthMethod] -- ^ Cached list of authentication methods supported by server+ , csLogger :: Maybe SpecializedLogger -- ^ Logging function+ , csManager :: Maybe Manager -- ^ HTTP\/HTTPS manager+ }++-- | Client monad+type Client a = StateT ClientState IO a++instance Localized (StateT ClientState IO) where+ getLanguage = lift getLanguage+ getTranslations = lift getTranslations+ getContext = lift getContext++instance HasLogger (StateT ClientState IO) where+ getLogger = do+ mbLogger <- gets csLogger+ case mbLogger of+ Just logger -> return logger+ Nothing -> fail "Logger is not initialized yet"++ localLogger logger actions = do+ oldLogger <- gets csLogger+ modify $ \st -> st {csLogger = Just logger}+ result <- actions+ modify $ \st -> st {csLogger = oldLogger}+ return result++-- | Execute actions within Client monad.+runClient :: ClientState -> Client a -> IO a+runClient st action =+ evalStateT (withLogging (LoggingSettings logSettings) action) st+ where+ logSettings =+ filtering [([], verbosity)] $+ FastLoggerSettings logFormat (FL.LogStderr 0)+ verbosity = logLevel $ cmdCommon $ csCmdline st+ logFormat = "{level:~l}: {message}\n"++-- | Get manager base URL.+getBaseUrl :: Client String+getBaseUrl = do+ cfg <- gets csConfig+ opts <- gets csCmdline+ liftIO $ getManagerUrl opts cfg++-- | Throw an exception.+throwC :: TL.Text -> Client a+throwC msg = lift $ throw $ ClientException msg++wrapClient :: (forall a. IO a -> IO a) -> Client b -> Client b+wrapClient wrapper actions = do+ state <- get+ (result, state') <- liftIO $ wrapper $ do+ runStateT actions state+ put state'+ return result++-- | Catch an exception.+catchC :: Exception e+ => Client a+ -> (e -> Client a)+ -> Client a+catchC action handler = do+ state <- get+ (result, state') <- liftIO $ runStateT action state `catch` \e ->+ runStateT (handler e) state+ put state'+ return result+
+ src/Batchd/Client/Shell.hs view
@@ -0,0 +1,105 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE RankNTypes #-}+-- | This module contains definitions for client interactive shell implementation.+module Batchd.Client.Shell (commandHandler) where++import Control.Monad.State+import Data.Maybe+import qualified Data.Text.Lazy.IO as TLIO+import Data.Text.Format.Heavy+import Options.Applicative+import System.Console.Haskeline+import System.Environment (getProgName)++import Batchd.Core.Common.Localize+import Batchd.Client.Types+import Batchd.Client.Logging+import Batchd.Client.Actions+import Batchd.Client.CmdLine+import Batchd.Client.Monad+import Batchd.Client.Http++detectMode :: CrudMode -> [Command -> Bool] -> Command -> CrudMode+detectMode mode mods cmd =+ if mode == Delete+ then Delete+ else if mode == Add+ then Add+ else if mode == Update || or [mod cmd | mod <- mods]+ then Update+ else View++-- | Handle client command line.+commandHandler :: Client ()+commandHandler = do+ opts <- gets (cmdCommand . csCmdline)+ case opts of+ Enqueue {} -> doEnqueue+ List {} -> doList+ Stats {} -> doStats+ Type {} -> doType+ Hosts -> doHosts+ Job {} -> do+ let mode = detectMode (jobMode opts) [isJust . status, isJust . hostName, isJust . jobNotes, isJust . startTime, isJust . queueName, isJust . prioritize] opts+ case mode of+ View -> viewJob+ Update -> updateJob+ Delete -> deleteJob+ Queue {} -> do+ let mode = detectMode (queueMode opts) [isJust . scheduleName, isJust . hostName, isJust . title, isJust . enabled] opts+ case mode of+ Add -> addQueue+ Update -> updateQueue+ Delete -> deleteQueue+ Schedule {} -> + case scheduleMode opts of+ View -> doListSchedules+ Add -> doAddSchedule+ Delete -> doDeleteSchedule+ Monitor {} -> doMonitor+ User {} ->+ case userMode opts of+ View -> doListUsers+ Add -> doAddUser+ Update -> doChangePassword+ Grant {} ->+ case grantMode opts of+ View -> doListPermissions+ Add -> doAddPermission+ Delete -> doRevokePermission+ RunShell {} -> do+ obtainCredentials+ message (__ "This is batch client shell. Type `--help' for list of available commands or `some_command --help' (without quotes) for help message on particular command.") ()+ runShell++errorHandler :: ClientException -> Client ()+errorHandler (ClientException e) =+ liftIO $ TLIO.putStrLn =<< (__f "Error: {}" (Single e))++-- | Interactive shell REPL.+runShell :: Client ()+runShell = runInputT defaultSettings loop+ where+ loop = do+ mbLine <- getInputLine "batch> "+ case mbLine of+ Nothing -> return ()+ Just "exit" -> return ()+ Just "quit" -> return ()+ Just "" -> loop+ Just line -> do+ let res = execParserPure Options.Applicative.defaultPrefs parserInfo (words line)+ case res of+ (Success cmd) -> lift $ do+ modify $ \st -> st {csCmdline = cmd}+ commandHandler `catchC` errorHandler+ (Failure failure) -> do+ progn <- liftIO $ getProgName+ let (msg,_) = renderFailure failure progn+ liftIO $ putStrLn msg+ (CompletionInvoked _) -> do+ liftIO $ putStrLn "bash-completion is not supported in interactive shell."+ loop+
+ src/Batchd/Client/Types.hs view
@@ -0,0 +1,30 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE DeriveGeneric #-}+-- | This module contains basic type definitions used by client.+module Batchd.Client.Types where++import GHC.Generics+import Control.Exception+import Data.Generics hiding (Generic)+import qualified Data.Text.Lazy as TL++-- | Client-side exception.+data ClientException = ClientException TL.Text+ deriving (Data, Typeable, Generic)++instance Exception ClientException++instance Show ClientException where+ show (ClientException e) = TL.unpack e++data CrudMode =+ View+ | Add+ | Update+ | Delete+ deriving (Eq, Show, Data, Typeable)++-- | User credentials.+type Credentials = (String, String)+
+ src/Batchd/Common/Config.hs view
@@ -0,0 +1,61 @@+{-# LANGUAGE OverloadedStrings #-}+-- | This module contains utility functions for locating and loading+-- configuration files.+module Batchd.Common.Config where++import Control.Exception+import Data.Yaml+import qualified Data.Text as T+import System.IO++import Batchd.Core.Common.Types+import Batchd.Core.Common.Localize+import Batchd.Core.Common.Config+import Batchd.Common.Types++-- | Load job type config file+loadTemplate :: T.Text -> IO (Either Error JobType)+loadTemplate name = loadConfig "jobtypes" name InvalidJobType++-- | Load global config file (@"batchd.yaml"@)+loadGlobalConfig :: Maybe FilePath -> IO (Either Error GlobalConfig)+loadGlobalConfig mbCmdlinePath = do+ mbPath <- case mbCmdlinePath of+ Just path -> return $ Just path+ Nothing -> locateConfig "" "batchd.yaml"+ case mbPath of+ Nothing -> return $ Left $ FileNotExists "batchd.yaml"+ Just path -> do+ r <- decodeFileEither path+ case r of+ Left err -> return $ Left $ InvalidDbCfg err+ Right cfg -> do+ return $ Right cfg++-- | Read password from stdout.+getPassword :: String -- ^ Prompt+ -> IO String+getPassword prompt = do+ putStr prompt+ hFlush stdout+ pass <- withEcho False getLine+ putChar '\n'+ return pass++-- | Ask password from stdout twice.+-- Make sure that entered passwords are equal.+getPassword2 :: IO String+getPassword2 = do+ pwd1 <- getPassword =<< (__s "Password: ")+ pwd2 <- getPassword =<< (__s "Password again: ")+ if pwd1 /= pwd2+ then fail =<< (__s "passwords do not match")+ else return pwd1++-- | Execute IO actions with enabled\/disabled terminal echo,+-- and then return echo state to previous.+withEcho :: Bool -> IO a -> IO a+withEcho echo action = do+ old <- hGetEcho stdin+ bracket_ (hSetEcho stdin echo) (hSetEcho stdin old) action+
+ src/Batchd/Common/Data.hs view
@@ -0,0 +1,256 @@+{-# LANGUAGE EmptyDataDecls #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE QuasiQuotes #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE TypeSynonymInstances #-}+{-# LANGUAGE NoMonomorphismRestriction #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE DataKinds #-}+-- | This module contains data type definitions for most entities stored in DB.+module Batchd.Common.Data where++import GHC.Generics++import Data.Time+import Data.Dates+import Database.Persist+import Data.Maybe+import Data.Int+import Data.Aeson as Aeson+import Data.Aeson.Types+import qualified Data.Text as T+import qualified Data.Text.Lazy as TL+import Data.Text.Format.Heavy+import Data.Generics hiding (Generic)++import Database.Persist.TH+import System.Exit++import Batchd.Common.Types++share [mkPersist sqlSettings, mkMigrate "migrateAll", mkDeleteCascade sqlSettings] [persistLowerCase|+JobParam+ jobId JobId+ name TL.Text+ value T.Text+ UniqParam jobId name++Job+ typeName String+ queueName String+ seq Int+ userName String default='root'+ createTime UTCTime default=CURRENT_TIMESTAMP+ startTime UTCTime Maybe+ status JobStatus default='New'+ tryCount Int default=0+ hostName String Maybe+ notes String Maybe+ UniqJobSeq queueName seq+ Foreign User user userName++JobResult+ jobId JobId+ time UTCTime default=CURRENT_TIMESTAMP+ exitCode ExitCode Maybe+ stdout T.Text sqltype=TEXT+ stderr T.Text sqltype=TEXT+ Primary jobId time++Queue+ name String+ title String default=''+ enabled Bool default=True+ scheduleName String+ hostName String Maybe+ autostartJobCount Int Maybe+ Primary name+ Foreign Schedule schedule scheduleName++Schedule+ name String+ Primary name++ScheduleTime+ scheduleName String+ begin TimeOfDay+ end TimeOfDay+ Foreign Schedule schedule scheduleName++ScheduleWeekDay+ scheduleName String+ weekDay WeekDay+ Foreign Schedule schedule scheduleName++MetricRecord+ name T.Text+ time UTCTime+ daemon DaemonMode+ kind MetricKind+ value Int64 Maybe+ text T.Text Maybe+ mean Double Maybe+ variance Double Maybe+ count Int64 Maybe+ sum Double Maybe+ min Double Maybe+ max Double Maybe+ Primary name time daemon++User+ name String+ pwdHash String+ salt String+ Primary name++UserPermission+ userName String+ permission Permission+ queueName String Maybe+ typeName String Maybe+ hostName T.Text Maybe+ Foreign User user userName+ Foreign Queue queue queueName+|]++deriving instance Eq ScheduleTime+deriving instance Show ScheduleTime++deriving instance Generic Queue+deriving instance Generic MetricKind+deriving instance Generic MetricRecord++instance ToJSON Queue where+ toJSON = genericToJSON (jsonOptions "queue")++instance FromJSON Queue where+ parseJSON = genericParseJSON (jsonOptions "queue")++newtype UpdateList a = UpdateList [Update a]++instance FromJSON (UpdateList ScheduleTime) where+ parseJSON o = do+ uBegin <- parseUpdate ScheduleTimeBegin "begin" o+ uEnd <- parseUpdate ScheduleTimeEnd "end" o+ return $ UpdateList $ catMaybes [uBegin, uEnd]++instance FromJSON (UpdateList Queue) where+ parseJSON o = do+ uSchedule <- parseUpdate QueueScheduleName "schedule_name" o+ uTitle <- parseUpdate QueueTitle "title" o+ uEnable <- parseUpdate QueueEnabled "enabled" o+ uHostName <- parseUpdateStar QueueHostName "host_name" o+ uAutostart <- parseUpdateMaybe QueueAutostartJobCount "autostart_job_count" o+ return $ UpdateList $ catMaybes [uEnable, uTitle, uSchedule, uHostName, uAutostart]++instance FromJSON (UpdateList Job) where+ parseJSON o = do+ uQueue <- parseUpdate JobQueueName "queue_name" o+ uStatus <- parseUpdate JobStatus "status" o+ uHost <- parseUpdateStar JobHostName "host_name" o+ uNotes <- parseUpdateMaybe JobNotes "notes" o+ uStart <- parseUpdateMaybe JobStartTime "start_time" o+ return $ UpdateList $ catMaybes [uQueue, uStatus, uHost, uNotes, uStart]++deriving instance Generic JobResult+deriving instance Generic UserPermission++instance ToJSON JobResult where+ toJSON = genericToJSON (jsonOptions "jobResult")++instance FromJSON JobResult where+ parseJSON = genericParseJSON (jsonOptions "jobResult")++instance ToJSON UserPermission where+ toJSON = genericToJSON (jsonOptions "userPermission")++instance FromJSON UserPermission where+ parseJSON = genericParseJSON (jsonOptions "userPermission")++data MoveAction =+ First+ | More+ | Less+ | Last+ deriving (Eq, Show, Data, Typeable, Generic)++instance FromJSON MoveAction where+ parseJSON (Aeson.String "first") = return First+ parseJSON (Aeson.String "more") = return More+ parseJSON (Aeson.String "less") = return Less+ parseJSON (Aeson.String "last") = return Last+ parseJSON invalid = typeMismatch "job priority direction" invalid++instance ToJSON MoveAction where+ toJSON First = toJSON ("first" :: String)+ toJSON More = toJSON ("more" :: String)+ toJSON Less = toJSON ("less" :: String)+ toJSON Last = toJSON ("last" :: String)++data JobUpdate =+ Prioritize MoveAction+ | Move String+ | UpdateJob (UpdateList Job)++instance FromJSON JobUpdate where+ parseJSON o@(Object v) = do+ pr <- v .:? "priority"+ case pr of+ Just action -> return $ Prioritize action+ Nothing -> do+ mv <- v .:? "move"+ case mv of+ Just qname -> return $ Move qname+ Nothing -> do+ upd <- parseJSON o+ return $ UpdateJob upd+ parseJSON invalid = typeMismatch "job update query" invalid++instance ToJSON MetricKind where+ toJSON Counter = Aeson.String "c"+ toJSON Gauge = Aeson.String "g"+ toJSON Label = Aeson.String "l"+ toJSON Distribution = Aeson.String "d"++instance FromJSON MetricKind where+ parseJSON (Aeson.String "c") = pure Counter+ parseJSON (Aeson.String "g") = pure Gauge+ parseJSON (Aeson.String "l") = pure Label+ parseJSON (Aeson.String "d") = pure Distribution+ parseJSON v = typeMismatch "metric kind" v++instance ToJSON MetricRecord where+ toJSON = genericToJSON (jsonOptions "metricRecord")++instance FromJSON MetricRecord where+ parseJSON = genericParseJSON (jsonOptions "metricRecord")++instance VarContainer MetricRecord where+ lookupVar "value" r = Just $ Variable $ metricRecordValue r+ lookupVar "text" r = Just $ Variable $ metricRecordText r+ lookupVar "mean" r = Just $ Variable $ metricRecordMean r+ lookupVar "variance" r = Just $ Variable $ metricRecordVariance r+ lookupVar "count" r = Just $ Variable $ metricRecordCount r+ lookupVar "sum" r = Just $ Variable $ metricRecordSum r+ lookupVar "min" r = Just $ Variable $ metricRecordMin r+ lookupVar "max" r = Just $ Variable $ metricRecordMax r+ lookupVar _ _ = Nothing++instance ClosedVarContainer MetricRecord where+ allVarNames _ = ["value", "text", "mean", "variance", "count", "sum", "min", "max"]++-- instance ToJSON JobUpdate where+-- toJSON (Prioritize action) = object ["priority" .= action]+-- toJSON (Move qname) = object ["move" .= qname]+-- toJSON (UpdateJob lst) = toJSON lst+
+ src/Batchd/Common/Schedule.hs view
@@ -0,0 +1,100 @@+{-# LANGUAGE DeriveDataTypeable, StandaloneDeriving, RecordWildCards, DeriveGeneric #-}+-- | This module contains definitions for working with schedules+module Batchd.Common.Schedule where++import GHC.Generics+import Data.Time.Clock+import Data.Time.LocalTime+import Data.Dates+import Data.Dates.Formats+import Data.Aeson+import Text.Parsec+import Text.Parsec.String++import Batchd.Common.Types (jsonOptions)+import Batchd.Common.Data++-- | Time period within day.+data Period =+ Period {+ periodBegin :: TimeOfDay,+ periodEnd :: TimeOfDay+ }+ deriving (Eq, Generic)++instance Show Period where+ show (Period begin end) = show begin ++ " -- " ++ show end++toPeriod :: ScheduleTime -> Period+toPeriod (ScheduleTime {..}) = Period scheduleTimeBegin scheduleTimeEnd++instance ToJSON Period where+ toJSON = genericToJSON (jsonOptions "period")++instance FromJSON Period where+ parseJSON = genericParseJSON (jsonOptions "period")++data ScheduleInfo =+ ScheduleInfo {+ sName :: String+ , sWeekdays :: Maybe [WeekDay]+ , sTime :: Maybe [Period]+ }+ deriving (Eq, Show, Generic)++instance ToJSON ScheduleInfo where+ toJSON = genericToJSON (jsonOptions "s")++instance FromJSON ScheduleInfo where+ parseJSON = genericParseJSON (jsonOptions "s")++anytime :: ScheduleInfo+anytime = ScheduleInfo "anytime" Nothing Nothing++getTime :: DateTime -> TimeOfDay+getTime (DateTime {..}) = TimeOfDay hour minute (fromIntegral second)++parsePeriod :: String -> Either ParseError Period+parsePeriod str = parse parser "(period description)" str + where+ format :: Format+ format = [HOUR True 2,Fixed True ":",MINUTE True 2,Fixed True ":",SECOND True 2]++ parser :: Parser Period+ parser = do+ begin <- formatParser format+ spaces+ end <- formatParser format+ return $ Period (getTime begin) (getTime end)++allows :: ScheduleInfo -> DateTime -> Bool+allows (ScheduleInfo {..}) dt = weekdayOk && timeOk+ where+ weekdayOk = case sWeekdays of+ Nothing -> True+ Just weekdays -> dateWeekDay dt `elem` weekdays++ timeOk = case sTime of+ Nothing -> True+ Just lst -> any goodTime lst++ goodTime (Period {..}) = + getTime dt > periodBegin && getTime dt <= periodEnd++allowsU :: ScheduleInfo -> UTCTime -> Bool+allowsU (ScheduleInfo {..}) dt = weekdayOk && timeOk+ where+ date = dayToDateTime (utctDay dt)+ time = timeToTimeOfDay (utctDayTime dt)++ weekdayOk = case sWeekdays of+ Nothing -> True+ Just weekdays -> dateWeekDay date `elem` weekdays++ timeOk = case sTime of+ Nothing -> True+ Just lst -> any goodTime lst++ goodTime (Period {..}) = + time > periodBegin && time <= periodEnd+
+ src/Batchd/Common/Types.hs view
@@ -0,0 +1,732 @@+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE RecordWildCards #-}+-- | This module contains data type declarations that are used both by batchd daemon and client.+module Batchd.Common.Types+ (+ -- * Data types+ TemplateSyntax (..),+ JobType (..), OnFailAction (..),+ ParamType (..), ParamDesc (..),+ JobStatus (..),+ JobParamInfo, JobInfo (..),+ UserInfo (..), Permission (..),+ DbDriver (..), DaemonMode (..),+ AuthMode (..), AuthMethod (..),+ LogTarget (..), LogConfig (..),+ WebClientConfig (..),+ ManagerConfig (..), DispatcherConfig (..),+ MetricsConfig (..),+ StorageConfig (..),+ GlobalConfig (..),+ MetricKind (..),+ ByStatus (..),+ -- * Exceptions+ UploadException (..), DownloadException (..), ExecException (..),+ -- * Utility functions+ lookupParam, getParamType,+ parseUpdate, parseUpdateMaybe,+ parseUpdateStar, parseStatus,+ jsonOptions, authMethods,+ -- * Some default settings+ defaultManagerPort, zeroUtcTime,+ defaultAuthMode, defaultLogConfig,+ defaultLogFormat, defaultStaticSalt,+ getAuthStaticSalt+ ) where++import GHC.Generics+import Control.Exception+import Control.Monad+import Data.Generics hiding (Generic)+import Data.Int+import Data.Char+import Data.String+import Data.Time+import Data.List (isPrefixOf)+import Database.Persist+import Database.Persist.TH+import qualified Data.Map as M+import qualified Data.HashMap.Strict as H+import qualified Data.Text as T+import qualified Data.Text.Lazy as TL+import Data.Aeson as Aeson+import Data.Aeson.Types+import Data.Default+import qualified Data.Text.Format.Heavy as F+import qualified Data.Text.Format.Heavy.Parse.Braces as PF+import Data.Dates+import qualified System.Posix.Syslog as Syslog+import System.Log.Heavy+import System.Exit++import Batchd.Core.Common.Types++-- | Default manager port - 9681.+defaultManagerPort :: Int+defaultManagerPort = 9681++-- | Type of job parameter+data ParamType =+ String+ | Integer+ | InputFile+ | OutputFile+ deriving (Eq, Show, Data, Typeable, Generic)++-- | Description of job parameter+data ParamDesc = ParamDesc {+ piName :: TL.Text -- ^ Parameter name (identifier)+ , piType :: ParamType -- ^ Parameter type+ , piTitle :: TL.Text -- ^ Parameter title (to show in client)+ , piDefault :: T.Text -- ^ Default value of the parameter+ }+ deriving (Eq, Show, Data, Typeable, Generic)++data TemplateSyntax = Shell | Python+ deriving (Eq, Show, Data, Typeable, Generic)++instance FromJSON TemplateSyntax where+ parseJSON = genericParseJSON defaultOptions++instance ToJSON TemplateSyntax where+ toJSON = genericToJSON defaultOptions++-- | Job type description+data JobType = JobType {+ jtName :: String -- ^ Job type name (identifier)+ , jtTitle :: Maybe String -- ^ Job type title (to show in client)+ , jtSyntax :: Maybe TemplateSyntax+ , jtTemplate :: [T.Text] -- ^ Template of command line to execute+ , jtOnFail :: OnFailAction -- ^ What to do if execution failed+ , jtHostName :: Maybe String -- ^ Name of host where to execute jobs.+ -- Nothing means use default host from queue or localhost.+ , jtMaxJobs :: Maybe Int -- ^ Maximum count of jobs of this type which+ -- can be executed in parallel on one host.+ -- Can be overruled in host settings.+ , jtParams :: [ParamDesc] -- ^ Set of job parameters+ }+ deriving (Eq, Show, Data, Typeable, Generic)++stripPrefix :: String -> String -> String+stripPrefix prefix str =+ if prefix `isPrefixOf` str+ then drop (length prefix) str+ else str++camelCaseToUnderscore :: String -> String+camelCaseToUnderscore = go False+ where+ go _ [] = []+ go False (x:xs) = toLower x : go True xs+ go True (x:xs)+ | isUpper x = '_' : toLower x : go True xs+ | otherwise = x : go True xs++jsonOptions :: String -> Data.Aeson.Types.Options+jsonOptions prefix = defaultOptions {fieldLabelModifier = camelCaseToUnderscore . stripPrefix prefix}++instance FromJSON ParamType where+ parseJSON = genericParseJSON $ defaultOptions {fieldLabelModifier = camelCaseToUnderscore}++instance ToJSON ParamType where+ toJSON = genericToJSON $ defaultOptions {fieldLabelModifier = camelCaseToUnderscore}++instance FromJSON ParamDesc where+ parseJSON (Object v) = do+ name <- v .: "name" + tp <- v .: "type"+ title <- v .:? "title" .!= name+ dflt <- v .:? "default" .!= ""+ return $ ParamDesc name tp title dflt+ parseJSON invalid = typeMismatch "parameter description" invalid++instance ToJSON ParamDesc where+ toJSON = genericToJSON (jsonOptions "pi")++repackLines :: [T.Text] -> [T.Text]+repackLines texts = concatMap T.lines texts++instance FromJSON JobType where+ parseJSON (Object v) = do+ name <- v .: "name"+ title <- v .:? "title"+ tmp <- v .: "template"+ syntax <- v .:? "syntax"+ template <- case tmp of+ Aeson.String str -> return [str]+ _ -> parseJSON tmp+ on_fail <- v .:? "on_fail" .!= Continue+ host_name <- v .:? "host_name"+ max_jobs <- v .:? "max_jobs"+ params <- v .: "params"+ return $ JobType name title syntax (repackLines template) on_fail host_name max_jobs params++instance ToJSON JobType where+ toJSON = genericToJSON (jsonOptions "jt")++-- | What to do if job execution fails+data OnFailAction =+ Continue -- ^ Continue to the next job+ | RetryNow Int -- ^ Leave the job in the queue and retry execution.+ -- Not more than @n@ times.+ | RetryLater Int -- ^ Put the job to the end of queue, to be executed later.+ -- Not more than @n@ times.+ deriving (Eq, Show, Read, Data, Typeable, Generic)++instance ToJSON OnFailAction where+ toJSON Continue = Aeson.String "continue"+ toJSON (RetryNow n) = object ["retry" .= object ["when" .= ("now" :: T.Text), "count" .= n]]+ toJSON (RetryLater n) = object ["retry" .= object ["when" .= ("later" :: T.Text), "count" .= n]]++instance FromJSON OnFailAction where+ parseJSON (Aeson.String "continue") = return Continue+ parseJSON (Aeson.String "retry") = return (RetryNow 1)+ parseJSON (Object v) = do+ r <- v .: "retry"+ case r of+ Object retry -> do+ nowStr <- retry .:? "when" .!= "now"+ now <- case nowStr of+ "now" -> return True+ "later" -> return False+ _ -> fail $ "Unknown retry type specification: " ++ nowStr+ count <- retry .:? "count" .!= 1+ if now+ then return $ RetryNow count+ else return $ RetryLater count+ _ -> typeMismatch "retry" r+ parseJSON invalid = typeMismatch "on fail" invalid++-- | Job execution status+data JobStatus =+ New -- ^ Just created, waiting for polling process to peek it.+ | Waiting -- ^ Waiting for free worker.+ | Processing -- ^ Being processed by the worker.+ | Done -- ^ Successfully executed.+ | Failed -- ^ Execution failed.+ | Postponed -- ^ Execution postponed. This status can be set only manually.+ deriving (Eq, Ord, Show, Read, Data, Typeable, Generic)++instance ToJSON JobStatus+instance FromJSON JobStatus++newtype ByStatus a = ByStatus (M.Map JobStatus a)+ deriving (Eq, Show, Data, Typeable, Generic)++instance ToJSON a => ToJSON (ByStatus a) where+ toJSON (ByStatus m) = object $ map go $ M.assocs m+ where+ go (st,x) = (T.pack $ map toLower $ show st) .= toJSON x++instance FromJSON a => FromJSON (ByStatus a) where+ parseJSON (Object v) = do+ pairs <- forM (H.toList v) go+ return $ ByStatus $ M.fromList pairs+ where+ go (key,val) = do+ Just st <- parseStatus Nothing (fail "invalid status") (Just key) + cnt <- parseJSON val+ return (st, cnt)++deriving instance Generic WeekDay+instance ToJSON WeekDay+instance FromJSON WeekDay++-- | Job parameter values+type JobParamInfo = Variables++-- | Job information. This data type+-- unites information from @Job@ and connected+-- @JobParam@.+data JobInfo = JobInfo {+ jiId :: Int64,+ jiQueue :: String,+ jiType :: String,+ jiSeq :: Int,+ jiUserName :: String,+ jiCreateTime :: UTCTime,+ jiStartTime :: Maybe UTCTime,+ jiStatus :: JobStatus,+ jiTryCount :: Int,+ jiHostName :: Maybe String,+ jiNotes :: Maybe String,+ jiResultTime :: Maybe UTCTime,+ jiExitCode :: Maybe ExitCode,+ jiStdout :: Maybe T.Text,+ jiStderr :: Maybe T.Text,+ jiParams :: JobParamInfo+ }+ deriving (Generic, Show)++instance ToJSON JobInfo where+ toJSON = genericToJSON (jsonOptions "ji")++zeroUtcTime :: UTCTime+zeroUtcTime = UTCTime (ModifiedJulianDay 0) 0++instance FromJSON JobInfo where+ parseJSON (Object v) =+ JobInfo+ <$> v .:? "id" .!= 0+ <*> v .:? "queue" .!= ""+ <*> v .: "type"+ <*> v .:? "seq" .!= 0+ <*> v .:? "user_name" .!= "<unknown>"+ <*> v .:? "create_time" .!= zeroUtcTime+ <*> v .:? "start_time"+ <*> v .:? "status" .!= New+ <*> v .:? "try_count" .!= 0+ <*> v .:? "host_name"+ <*> v .:? "notes"+ <*> v .:? "result_time"+ <*> v .:? "exit_code"+ <*> v .:? "stdout"+ <*> v .:? "stderr"+ <*> v .:? "params" .!= M.empty+ parseJSON invalid = typeMismatch "job" invalid++-- | User name and password+data UserInfo = UserInfo {+ uiName :: String,+ uiPassword :: String+ } deriving (Generic, Show)++instance ToJSON UserInfo where+ toJSON = genericToJSON (jsonOptions "ui")++instance FromJSON UserInfo where+ parseJSON = genericParseJSON (jsonOptions "ui")++-- | Lookup for parameter description by name.+lookupParam :: TL.Text -- ^ Parameter name+ -> [ParamDesc] -- ^ List of parameter descriptions+ -> Maybe ParamDesc+lookupParam _ [] = Nothing+lookupParam name (p:ps)+ | piName p == name = Just p+ | otherwise = lookupParam name ps++-- | Lookup for parameter type by name+getParamType :: JobType -> TL.Text -> Maybe ParamType+getParamType jt name = piType `fmap` lookupParam name (jtParams jt)++-- | Supported database drivers+data DbDriver = Sqlite | PostgreSql+ deriving (Eq, Show, Data, Typeable, Generic)++instance ToJSON DbDriver+instance FromJSON DbDriver++-- | Daemon execution mode+data DaemonMode =+ Manager -- ^ Manager thread only+ | Dispatcher -- ^ Dispatcher thread only+ | Both -- ^ Both manager and dispatcher threads+ deriving (Data, Typeable, Show, Read, Eq, Ord, Generic)++instance ToJSON DaemonMode+instance FromJSON DaemonMode++-- | Authentication methods+data AuthMode =+ AuthDisabled -- ^ Authentication is disabled, all users to be authorized as superusers+ | AuthConfig {+ authBasicEnabled :: Bool -- ^ Is HTTP Basic authentication enabled+ , authHeaderEnabled :: Bool -- ^ Is use of @X-Auth-User:@ HTTP header enabled+ , authStaticSalt :: String -- ^ Static salt value+ }+ deriving (Data, Typeable, Show, Eq, Generic)++getAuthStaticSalt :: AuthMode -> String+getAuthStaticSalt AuthDisabled = defaultStaticSalt+getAuthStaticSalt auth = authStaticSalt auth++instance ToJSON AuthMode where+ toJSON AuthDisabled = Aeson.String "disable"+ toJSON (AuthConfig {..}) = object ["basic" .= authBasicEnabled, "header" .= authHeaderEnabled, "staticSalt" .= authStaticSalt]++instance FromJSON AuthMode where+ parseJSON (Aeson.String "disable") = return AuthDisabled+ parseJSON (Object v) =+ AuthConfig+ <$> v .:? "basic" .!= True+ <*> v .:? "header" .!= False+ <*> v .:? "static_salt" .!= defaultStaticSalt++-- | Default autentication mode: only HTTP basic auth is enabled.+defaultAuthMode :: AuthMode+defaultAuthMode = AuthConfig True False defaultStaticSalt++-- | Supported authentication methods+data AuthMethod = BasicAuth | HeaderAuth+ deriving (Data, Typeable, Show, Read, Eq, Generic)++instance ToJSON AuthMethod where+ toJSON BasicAuth = Aeson.String "basic"+ toJSON HeaderAuth = Aeson.String "header"++instance FromJSON AuthMethod where+ parseJSON (Aeson.String "basic") = return BasicAuth+ parseJSON (Aeson.String "header") = return HeaderAuth+ parseJSON x = typeMismatch "auth method" x++-- | Get enabled authentication methods from specified mode.+authMethods :: AuthMode -> [AuthMethod]+authMethods AuthDisabled = []+authMethods (AuthConfig {..}) =+ (if authBasicEnabled then [BasicAuth] else []) +++ (if authHeaderEnabled then [HeaderAuth] else [])++-- | Supported logging targets+data LogTarget =+ LogSyslog+ | LogStdout+ | LogStderr+ | LogFile FilePath+ deriving (Eq, Show, Data, Typeable, Generic)++instance ToJSON LogTarget where+ toJSON LogSyslog = Aeson.String "syslog"+ toJSON LogStdout = Aeson.String "stdout"+ toJSON LogStderr = Aeson.String "stderr"+ toJSON (LogFile path) = toJSON path++instance FromJSON LogTarget where+ parseJSON (Aeson.String "syslog") = return LogSyslog+ parseJSON (Aeson.String "stdout") = return LogStdout+ parseJSON (Aeson.String "stderr") = return LogStderr+ parseJSON (Aeson.String path) = return $ LogFile $ T.unpack path+ parseJSON invalid = typeMismatch "log target" invalid++-- | Logging configuration+data LogConfig = LogConfig {+ lcTarget :: LogTarget -- ^ Logging target. Default is syslog.+ , lcFormat :: F.Format -- ^ Log message format.+ , lcLevel :: Level -- ^ General filter+ , lcFilter :: [(String, Level)] -- ^ Filter by source prefixes+ }+ deriving (Eq, Show, Typeable, Generic)++-- | Default logging config+defaultLogConfig :: LogConfig+defaultLogConfig = LogConfig LogSyslog defaultLogFormat info_level []++-- | Default log messages format:+-- @"{time} [{level:~l}] {source} ({fullcontext}): {message}\n"@+defaultLogFormat :: F.Format+defaultLogFormat = "{time} [{level:~l}] {source} ({fullcontext}): {message}\n"++instance ToJSON LogConfig where+ toJSON = genericToJSON (jsonOptions "lc")++instance ToJSON F.Format where+ toJSON fmt = toJSON (show fmt)++instance FromJSON LogConfig where+ parseJSON (Object v) = LogConfig+ <$> v .:? "target" .!= LogSyslog+ <*> parseLogFormat (v .:? "format")+ <*> v .:? "level" .!= info_level+ <*> parseFilter (v .:? "filter" .!= M.empty)+ where+ parseFilter :: Parser (M.Map String Level) -> Parser [(String, Level)]+ parseFilter = fmap M.assocs++ parseLogFormat :: Parser (Maybe TL.Text) -> Parser F.Format+ parseLogFormat p = do+ mbString <- p+ case mbString of+ Nothing -> return defaultLogFormat+ Just str -> case PF.parseFormat str of+ Left err -> fail $ show err+ Right fmt -> return fmt++-- | Global daemon configuration+data GlobalConfig = GlobalConfig {+ dbcDaemonMode :: DaemonMode -- ^ Daemon execution mode+ , dbcDriver :: DbDriver -- ^ Type of DB backend+ , dbcConnectionString :: T.Text -- ^ DB connection string+ , dbcConnectionPoolSize :: Int -- ^ Database connections pool size+ , dbcLogging :: LogConfig -- ^ Logging configuration+ , dbcManager :: ManagerConfig+ , dbcDispatcher :: DispatcherConfig+ , dbcDefTemplateSyntax :: TemplateSyntax+ , dbcDefScriptsDirectory :: FilePath+ , dbcMetrics :: MetricsConfig+ , dbcStorage :: StorageConfig+ , dbcVariables :: Variables+ }+ deriving (Eq, Show, Typeable, Generic)++instance Default GlobalConfig where+ def = GlobalConfig {+ dbcDaemonMode = Both+ , dbcDriver = Sqlite+ , dbcConnectionString = ":memory:"+ , dbcConnectionPoolSize = 10+ , dbcLogging = defaultLogConfig+ , dbcManager = def+ , dbcDispatcher = def+ , dbcDefTemplateSyntax = Shell+ , dbcDefScriptsDirectory = "./scripts"+ , dbcMetrics = def+ , dbcStorage = def+ , dbcVariables = def+ }++data MetricsConfig = MetricsConfig {+ mcGcMetrics :: Bool -- ^ Whether to enable GC metrics (default true)+ , mcHttpMetrics :: Bool -- ^ Whether to enable HTTP metrics for manager REST API (default true)+ , mcStorePrefixOnly :: Maybe T.Text -- ^ Only store metrics, names of which starts with specified prefix.+ , mcDumpTimeout :: Int -- ^ How often to dump metrics data to DB, in seconds (default - each 10 seconds)+ }+ deriving (Eq, Show, Typeable, Generic)++instance Default MetricsConfig where+ def = MetricsConfig {+ mcGcMetrics = True+ , mcHttpMetrics = True+ , mcStorePrefixOnly = Nothing+ , mcDumpTimeout = 10+ }++data WebClientConfig = WebClientConfig {+ wcAllowedOrigin :: Maybe String -- ^ Allowed Origin for CORS+ , wcPath :: String -- ^ Path to web client HTML\/JS\/CSS files+ }+ deriving (Eq, Show, Typeable, Generic)++instance Default WebClientConfig where+ def = WebClientConfig Nothing ""++data ManagerConfig = ManagerConfig {+ mcPort :: Int -- ^ Network port for manager to listen+ , mcAuth :: AuthMode -- ^ Authentication configuration+ , mcWebClient :: Maybe WebClientConfig+ }+ deriving (Eq, Show, Typeable, Generic)++instance Default ManagerConfig where+ def = ManagerConfig {+ mcPort = defaultManagerPort+ , mcAuth = defaultAuthMode+ , mcWebClient = Nothing+ }++data DispatcherConfig = DispatcherConfig {+ dcWorkers :: Int+ , dcPollTimeout :: Int+ }+ deriving (Eq, Show, Typeable, Generic)++instance Default DispatcherConfig where+ def = DispatcherConfig {+ dcWorkers = 1+ , dcPollTimeout = 10+ }++data StorageConfig = StorageConfig {+ scDoneJobs :: Int -- ^ How long to store executed jobs, in days.+ , scMetricRecords :: Int -- ^ How long to store metric records, in days.+ }+ deriving (Eq, Show, Typeable, Generic)++instance Default StorageConfig where+ def = StorageConfig {+ scDoneJobs = 2+ , scMetricRecords = 3+ }++-- | Default static salt value.+defaultStaticSalt :: String+defaultStaticSalt = "1234567890abcdef"++instance ToJSON MetricsConfig where+ toJSON = genericToJSON (jsonOptions "mc")++instance FromJSON MetricsConfig where+ parseJSON (Object v) =+ MetricsConfig+ <$> v .:? "gc_metrics" .!= mcGcMetrics def+ <*> v .:? "http_metrics" .!= mcHttpMetrics def+ <*> v .:? "store_prefix_only" .!= mcStorePrefixOnly def+ <*> v .:? "dump_timeout" .!= mcDumpTimeout def+ parseJSON invalid = typeMismatch "metrics configuration" invalid++instance ToJSON WebClientConfig where+ toJSON = genericToJSON (jsonOptions "wc")++instance FromJSON WebClientConfig where+ parseJSON (Object v) =+ WebClientConfig+ <$> v .:? "allowed_origin"+ <*> v .: "path"+ parseJSON invalid = typeMismatch "web client configuration" invalid++instance ToJSON ManagerConfig where+ toJSON = genericToJSON (jsonOptions "mc")++instance FromJSON ManagerConfig where+ parseJSON (Object v) =+ ManagerConfig+ <$> v .:? "port" .!= defaultManagerPort+ <*> v .:? "auth" .!= defaultAuthMode+ <*> v .:? "web_client"+ parseJSON invalid = typeMismatch "manager configuration" invalid++instance ToJSON DispatcherConfig where+ toJSON = genericToJSON (jsonOptions "dc")+ +instance FromJSON DispatcherConfig where+ parseJSON (Object v) =+ DispatcherConfig+ <$> v .:? "workers" .!= dcWorkers def+ <*> v .:? "poll_timeout" .!= dcPollTimeout def+ parseJSON invalid = typeMismatch "dispatcher configuration" invalid++instance ToJSON StorageConfig where+ toJSON = genericToJSON (jsonOptions "sc")+ +instance FromJSON StorageConfig where+ parseJSON (Object v) =+ StorageConfig+ <$> v .:? "done_jobs" .!= scDoneJobs def+ <*> v .:? "metric_records" .!= scMetricRecords def+ parseJSON invalid = typeMismatch "storage configuration" invalid++instance ToJSON GlobalConfig where+ toJSON = genericToJSON (jsonOptions "dbc")++instance FromJSON GlobalConfig where+ parseJSON (Object v) =+ GlobalConfig+ <$> v .:? "daemon" .!= dbcDaemonMode def+ <*> v .:? "driver" .!= dbcDriver def+ <*> v .:? "connection_string" .!= dbcConnectionString def+ <*> v .:? "connection_pool_size" .!= dbcConnectionPoolSize def+ <*> v .:? "logging" .!= defaultLogConfig+ <*> v .:? "manager" .!= def+ <*> v .:? "dispatcher" .!= def+ <*> v .:? "default_template_syntax" .!= Shell+ <*> v .:? "default_scripts_directory" .!= "./scripts"+ <*> v .:? "metrics" .!= def+ <*> v .:? "storage" .!= def+ <*> v .:? "variables" .!= def+ parseJSON invalid = typeMismatch "global configuration" invalid++data MetricKind =+ Counter+ | Gauge+ | Label+ | Distribution+ deriving (Eq, Ord, Show, Read)++deriving instance Data Syslog.Priority+deriving instance Typeable Syslog.Priority++deriving instance Data Level+deriving instance Typeable Level++parseStatus :: (Eq s, IsString s, Monad m) => Maybe JobStatus -> m (Maybe JobStatus) -> Maybe s -> m (Maybe JobStatus)+parseStatus dflt _ Nothing = return dflt+parseStatus _ _ (Just "all") = return Nothing+parseStatus _ _ (Just "new") = return $ Just New+parseStatus _ _ (Just "waiting") = return $ Just Waiting+parseStatus _ _ (Just "processing") = return $ Just Processing+parseStatus _ _ (Just "done") = return $ Just Done+parseStatus _ _ (Just "failed") = return $ Just Failed+parseStatus _ _ (Just "postponed") = return $ Just Postponed+parseStatus _ handle (Just _) = handle++instance ToJSON ExitCode where+ toJSON ExitSuccess = Number (fromIntegral 0)+ toJSON (ExitFailure n) = Number (fromIntegral n)++instance FromJSON ExitCode where+ parseJSON (Number 0) = return ExitSuccess+ parseJSON (Number n) = return $ ExitFailure $ round n+ parseJSON x = typeMismatch "exit code" x++data ExecException = ExecException SomeException+ deriving (Typeable)++instance Exception ExecException++instance Show ExecException where+ show (ExecException e) = "Exception during command execution: " ++ show e++data UploadException = UploadException FilePath SomeException+ deriving (Typeable)++instance Exception UploadException++instance Show UploadException where+ show (UploadException path e) = "Exception uploading file `" ++ path ++ "': " ++ show e++data DownloadException = DownloadException FilePath SomeException+ deriving (Typeable)++instance Exception DownloadException++instance Show DownloadException where+ show (DownloadException path e) = "Exception downloading file `" ++ path ++ "': " ++ show e++derivePersistField "WeekDay"+derivePersistField "JobStatus"+derivePersistField "ExitCode"+derivePersistField "DaemonMode"+derivePersistField "MetricKind"++-- | Supported user permissions.+data Permission =+ SuperUser+ | CreateJobs+ | ViewJobs+ | ManageJobs+ | ViewQueues+ | ManageQueues+ | ViewSchedules+ | ManageSchedules+ deriving (Eq, Show, Read, Data, Typeable, Generic)++derivePersistField "Permission"++instance FromJSON Permission where+ parseJSON = genericParseJSON $ defaultOptions {fieldLabelModifier = camelCaseToUnderscore}++instance ToJSON Permission where+ toJSON = genericToJSON $ defaultOptions {fieldLabelModifier = camelCaseToUnderscore}++parseUpdate :: (PersistField t, FromJSON t) => EntityField v t -> T.Text -> Value -> Parser (Maybe (Update v))+parseUpdate field label (Object v) = do+ mbValue <- v .:? label+ let upd = case mbValue of+ Nothing -> Nothing+ Just value -> Just (field =. value)+ return upd++parseUpdateMaybe :: (PersistField t, FromJSON t) => EntityField v (Maybe t) -> T.Text -> Value -> Parser (Maybe (Update v))+parseUpdateMaybe field label (Object v) = do+ if label `H.member` v+ then do+ mbValue <- v .:? label+ return $ Just $ field =. mbValue+ else return Nothing++parseUpdateStar :: (PersistField t, FromJSON t, IsString t, Eq t)+ => EntityField v (Maybe t) -> T.Text -> Value -> Parser (Maybe (Update v))+parseUpdateStar field label (Object v) = do+ mbValue <- v .:? label+ let upd = case mbValue of+ Nothing -> Nothing+ Just "*" -> Just (field =. Nothing)+ Just value -> Just (field =. Just value)+ return upd+
+ src/Batchd/Daemon/Auth.hs view
@@ -0,0 +1,475 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TemplateHaskell #-}+module Batchd.Daemon.Auth where++import Control.Monad+import Control.Monad.Reader+import Data.Maybe+import Data.Int+import qualified Data.Text as T+import qualified Data.Vault.Lazy as V+import Database.Persist+import Database.Persist.Sql as Sql hiding (Single)+import qualified Data.ByteString as B+import Network.HTTP.Types (status401, hAuthorization)+import Network.Wai+import qualified Network.Wai.Middleware.HttpAuth as HA+import qualified Web.Scotty.Trans as Scotty+import System.Log.Heavy (LoggingTState)+import Data.Text.Format.Heavy++import System.IO.Unsafe (unsafePerformIO)++import Batchd.Core.Common.Types+import Batchd.Common.Types+import Batchd.Common.Data+import Batchd.Daemon.Types+import Batchd.Daemon.Crypto+import Batchd.Daemon.Database+import Batchd.Core.Daemon.Logging++-- | WAI Request vault key for currently authenticated user name+usernameKey :: V.Key String+usernameKey = unsafePerformIO V.newKey+{-# NOINLINE usernameKey #-}++---------- Users manipulation --------------------++-- | Create user+createUserDb :: String -- ^ User name+ -> String -- ^ Password+ -> String -- ^ Static part of salt+ -> DB (Key User)+createUserDb name password staticSalt = do+ dynamicSalt <- liftIO randomSalt+ let hash = calcHash password dynamicSalt staticSalt+ let user = User name hash dynamicSalt+ insert user++-- | Create superuser+createSuperUserDb :: String -- ^ User name+ -> String -- ^ Password+ -> String -- ^ Static part of salt+ -> DB (Key User)+createSuperUserDb name password staticSalt = do+ userKey <- createUserDb name password staticSalt+ let perm = UserPermission name SuperUser Nothing Nothing Nothing+ insert perm+ return userKey++-- | Create user+createUser :: GlobalConfig+ -> LoggingTState+ -> String -- ^ User name+ -> String -- ^ Password+ -> IO Bool+createUser gcfg lts name password = do+ pool <- getPool gcfg lts+ let staticSalt = getAuthStaticSalt $ mcAuth $ dbcManager gcfg+ res <- runDBIO gcfg pool lts (createUserDb name password staticSalt)+ case res of+ Left _ -> return False+ Right _ -> return True++-- | Create superuser+createSuperUser :: GlobalConfig+ -> LoggingTState+ -> String -- ^ User name+ -> String -- ^ Password+ -> IO Bool+createSuperUser gcfg lts name password = do+ pool <- getPool gcfg lts+ let staticSalt = getAuthStaticSalt $ mcAuth $ dbcManager gcfg+ res <- runDBIO gcfg pool lts (createSuperUserDb name password staticSalt)+ case res of+ Left _ -> return False+ Right _ -> return True++-- | Create permission+createPermission :: String -- ^ User name+ -> UserPermission+ -> DB (Key UserPermission)+createPermission name perm = do+ let perm' = perm {userPermissionUserName = name}+ insert perm'++-- | Get list of permissions and their IDs for specified user+getPermissions :: String -- ^ User name+ -> DB [(Int64, UserPermission)]+getPermissions name = do+ res <- selectList [UserPermissionUserName ==. name] []+ let getKey (UserPermissionKey (SqlBackendKey id)) = id+ return [(getKey (entityKey e), entityVal e) | e <- res]++-- | Delete permission. Fail if specified permission does not belong to specified user.+deletePermission :: Int64 -- ^ Permission ID+ -> String -- ^ User name+ -> DB ()+deletePermission id name = do+ let pid = UserPermissionKey (SqlBackendKey id)+ mbPerm <- get pid+ case mbPerm of+ Nothing -> throwR $ UnknownError "Permission does not exist"+ Just perm -> do+ if userPermissionUserName perm == name+ then delete pid+ else throwR $ UnknownError "Permission does not belong to specified user"++-- | Change user password+changePasswordDb :: String -- ^ User name+ -> String -- ^ New password+ -> String -- ^ Static part of salt+ -> DB ()+changePasswordDb name password staticSalt = do+ dynamicSalt <- liftIO randomSalt+ let hash = calcHash password dynamicSalt staticSalt+ key = UserKey name+ update key [UserPwdHash =. hash, UserSalt =. dynamicSalt]++changePassword :: GlobalConfig+ -> LoggingTState+ -> String -- ^ User name+ -> String -- ^ Password+ -> IO Bool+changePassword gcfg lts name password = do+ pool <- getPool gcfg lts+ let staticSalt = getAuthStaticSalt $ mcAuth $ dbcManager gcfg+ res <- runDBIO gcfg pool lts (changePasswordDb name password staticSalt)+ case res of+ Left _ -> return False+ Right _ -> return True++-- | Get list of all users+getUsers :: DB [String]+getUsers = do+ res <- selectList [] []+ return $ map (userName . entityVal) res++------------------ Check user -----------------------++checkUserDb :: String -> String -> String -> DB Bool+checkUserDb name password staticSalt = do+ mbUser <- get (UserKey name)+ case mbUser of+ Nothing -> return False+ Just user -> do+ let hash = calcHash password (userSalt user) staticSalt+ return $ hash == userPwdHash user++checkUserExistsDb :: String -> DB Bool+checkUserExistsDb name = do+ mbUser <- get (UserKey name)+ return $ isJust mbUser++checkUser :: GlobalConfig -> LoggingTState -> (B.ByteString -> B.ByteString -> IO Bool)+checkUser gcfg lts nameBstr passwordBstr = do+ pool <- getPool gcfg lts+ let name = bstrToString nameBstr+ password = bstrToString passwordBstr+ salt = getAuthStaticSalt $ mcAuth $ dbcManager gcfg+ res <- runDBIO gcfg pool lts (checkUserDb name password salt)+ case res of+ Left _ -> return False+ Right r -> return r++checkUserExists :: GlobalConfig -> LoggingTState -> B.ByteString -> IO Bool+checkUserExists gcfg lts nameBstr = do+ pool <- getPool gcfg lts+ let name = bstrToString nameBstr+ res <- runDBIO gcfg pool lts (checkUserExistsDb name)+ case res of+ Left _ -> return False+ Right r -> return r++------------------ Authentication ---------------------++anonymousUserName :: String+anonymousUserName = "anonymous"++anonymousUser :: User+anonymousUser = User anonymousUserName "" ""++extractBasicUser :: Request -> String+extractBasicUser req =+ case (lookup hAuthorization $ requestHeaders req) >>= HA.extractBasicAuth of+ Nothing -> anonymousUserName+ Just (name,_) -> bstrToString name++-- | Returns true if this is OPTIONS request for @/@+isRootOptions :: Request -> Bool+isRootOptions rq = requestMethod rq == "OPTIONS" && pathInfo rq == []++-- | HTTP basic auth middleware+basicAuth :: GlobalConfig -> LoggingTState -> Middleware+basicAuth gcfg lts app req sendResponse =+ let settings = "batchd" :: HA.AuthSettings+ username = extractBasicUser req+ -- put user name extracted from header to vault+ req' = req {vault = V.insert usernameKey username $ vault req}+ in if isRootOptions req+ then app req sendResponse -- OPTIONS / is available without auth+ else case getAuthUserRq req of+ Nothing -> do+ infoIO lts $(here) "Will try to authenticate user with basic auth: {}" (Single username)+ HA.basicAuth (checkUser gcfg lts) settings app req' sendResponse+ Just _ -> app req sendResponse++-- | Authentication by X-Auth-User HTTP header+headerAuth :: GlobalConfig -> LoggingTState -> Middleware+headerAuth gcfg lts app req sendResponse = do+ -- liftIO $ putStrLn $ "X-Auth-User: " ++ show req+ if isRootOptions req+ then app req sendResponse -- OPTIONS / is available without auth+ else case lookup "X-Auth-User" (requestHeaders req) of+ Nothing -> do+ debugIO lts $(here) "No X-Auth-User header" ()+ app req sendResponse+ -- case getAuthUserRq req of+ -- Nothing -> sendResponse $ responseLBS status401 [] "User name not provided"+ -- Just _ -> app req sendResponse+ Just name -> do+ let username = bstrToString name+ -- put user name extracted from header to vault+ req' = req {vault = V.insert usernameKey username $ vault req}+ ok <- liftIO $ checkUserExists gcfg lts name+ infoIO lts $(here) "User from X-AUth-User header authenticated: {}" (Single username)+ if ok+ then app req' sendResponse+ else sendResponse $ responseLBS status401 [] "Specified user does not exist"++-- | Unconditional authentication+noAuth :: GlobalConfig -> LoggingTState -> Middleware+noAuth gcfg lts app req sendResponse = do+ let username = case lookup "X-Auth-User" (requestHeaders req) of+ Just n -> bstrToString n+ Nothing -> extractBasicUser req+ -- if by some condition username appeared in header, put it to vault.+ -- otherwise user name will be anonymous.+ req' = req {vault = V.insert usernameKey username $ vault req}+ infoIO lts $(here) "Authentication is disabled. User treated as superuser: {}" (Single username)+ app req' sendResponse++-- authentication :: GlobalConfig -> Middleware+-- authentication gcfg =+-- let m1 = if dbcEnableHeaderAuth gcfg+-- then headerAuth gcfg+-- else id+-- m2 = if dbcEnableBasicAuth gcfg+-- then basicAuth gcfg+-- else id+-- in m2 . m1++------------ Get current user ------------++-- | Get user name from WAI Request vault+getAuthUserRq :: Request -> Maybe String+getAuthUserRq req = V.lookup usernameKey $ vault req++-- | Get currently authenticated user name.+-- Nothing means user was not authenticated.+getAuthUserName :: Action (Maybe String)+getAuthUserName = do+ rq <- Scotty.request+ return $ getAuthUserRq rq++-- | Get currently authenticated user.+-- Fail if user was not authenticated.+getAuthUser :: Action User+getAuthUser = do+ cfg <- askConfigA+ if isAuthDisabled $ mcAuth $ dbcManager cfg+ then return anonymousUser+ else do+ mbName <- getAuthUserName+ case mbName of+ Nothing -> Scotty.raise $ InsufficientRights "user has to be authenticated"+ Just name -> do+ mbUser <- runDBA $ get (UserKey name)+ case mbUser of+ Nothing -> Scotty.raise $ UnknownError $ "user does not exist: " <> name+ Just user -> return user++isAuthDisabled :: AuthMode -> Bool+isAuthDisabled AuthDisabled = True+isAuthDisabled _ = False++whenAuthEnabled :: Action () -> Action ()+whenAuthEnabled actions = do+ cfg <- askConfigA+ when (not $ isAuthDisabled $ mcAuth $ dbcManager cfg) $ actions++whenAuthEnabledCheck :: Action Bool -> Action Bool+whenAuthEnabledCheck check = do+ cfg <- askConfigA+ if isAuthDisabled $ mcAuth $ dbcManager cfg+ then return True+ else check++askAuthEnabled :: Action Bool+askAuthEnabled = do+ cfg <- askConfigA+ return (not $ isAuthDisabled $ mcAuth $ dbcManager cfg)++----------- Check user rights ---------------++-- | Check if user has SuperUser permission+isSuperUserDb :: String -> DB Bool+isSuperUserDb name = do+ res <- selectList [UserPermissionUserName ==. name, UserPermissionPermission ==. SuperUser] []+ return $ not $ null res++isSuperUser :: String -> Action Bool+isSuperUser name = do+ whenAuthEnabledCheck $ runDBA $ isSuperUserDb name++-- | Check if user is superuser. Fail otherwise.+checkSuperUser :: Action ()+checkSuperUser = do+ whenAuthEnabled $ do+ user <- getAuthUser+ ok <- runDBA $ isSuperUserDb (userName user)+ when (not ok) $ do+ Scotty.raise $ InsufficientRights "user has to be superuser"++-- | Check that user is authenticated. Fail otherwise.+checkUserAuthenticated :: Action ()+checkUserAuthenticated = do+ _ <- getAuthUser+ return ()++-- | Check if user has specified permission+hasPermissionDb :: String -- ^ User name+ -> Permission+ -> String -- ^ Queue name+ -> DB Bool+hasPermissionDb name perm qname = do+ super <- isSuperUserDb name+ if super+ then return True+ else do+ exact <- selectList [UserPermissionUserName ==. name, UserPermissionPermission ==. perm, UserPermissionQueueName ==. Just qname] []+ if not (null exact)+ then return True+ else do+ any <- selectList [UserPermissionUserName ==. name, UserPermissionPermission ==. perm, UserPermissionQueueName ==. Nothing] []+ return $ not $ null any++-- | Check if user has permissions to create jobs in certain queue+hasCreatePermissionDb :: String -- ^ User name+ -> String -- ^ Queue name+ -> Maybe String -- ^ Just type name, or Nothing if you want to check that user has permission+ -- to create jobs of at least some type+ -> Maybe T.Text -- ^ Just host name, or Nothing if you want to check that user has permission+ -- to create jobs on at least some host. Use special @__default__@ value+ -- for default host of queue.+ -> DB Bool+hasCreatePermissionDb name qname mbTypename mbHostname = do+ super <- isSuperUserDb name+ if super+ then return True+ else do+ let byUser = [UserPermissionUserName ==. name, UserPermissionPermission ==. CreateJobs]+ byQueue = [UserPermissionQueueName ==. Nothing] ||. [UserPermissionQueueName ==. Just qname]+ byType = case mbTypename of+ Nothing -> []+ Just typename -> [UserPermissionTypeName ==. Nothing] ||. [UserPermissionTypeName ==. Just typename]+ byHost = case mbHostname of+ Nothing -> []+ Just hostname -> [UserPermissionHostName ==. Nothing] ||. [UserPermissionHostName ==. Just hostname]+ let filtr = byUser ++ byQueue ++ byType ++ byHost+ res <- selectList filtr []+ return $ not $ null res++-- | Check if user has permissions to create jobs in certain queue+hasCreatePermission :: String -- ^ User name+ -> String -- ^ Queue name+ -> Maybe String -- ^ Just type name, or Nothing if you want to check that user has permission+ -- to create jobs of at least some type+ -> Maybe T.Text -- ^ Just host name, or Nothing if you want to check that user has permission+ -- to create jobs on at least some host. Use special @__default__@ value+ -- for default host of queue.+ -> Action Bool+hasCreatePermission name qname mbTypename mbHostname =+ whenAuthEnabledCheck $ runDBA $ hasCreatePermissionDb name qname mbTypename mbHostname++-- | List names of hosts where user can create jobs of specified type in certain queue.+-- Returns Nothing if user can create jobs on any host.+listAllowedHostsDb :: String -- ^ User name+ -> String -- ^ Queue name+ -> String -- ^ Job type name+ -> DB (Maybe [T.Text])+listAllowedHostsDb name qname typename = do+ super <- isSuperUserDb name+ if super+ then return Nothing+ else do+ let byUser = [UserPermissionUserName ==. name, UserPermissionPermission ==. CreateJobs]+ byQueue = [UserPermissionQueueName ==. Nothing] ||. [UserPermissionQueueName ==. Just qname]+ byType = [UserPermissionTypeName ==. Nothing] ||. [UserPermissionTypeName ==. Just typename]+ fltr = byUser ++ byQueue ++ byType+ qryResult <- selectList fltr []+ let mbHosts = map (userPermissionHostName . entityVal) qryResult+ result = if any isNothing mbHosts -- in this case user can create jobs on any host+ then Nothing+ else Just $ map fromJust mbHosts+ return result++-- | List names of hosts where user can create jobs of specified type in certain queue.+-- Returns Nothing if user can create jobs on any host.+listAllowedHosts :: String -- ^ User name+ -> String -- ^ Queue name+ -> String -- ^ Job type name+ -> Action (Maybe [T.Text])+listAllowedHosts name qname typename = do+ auth <- askAuthEnabled+ if auth+ then runDBA $ listAllowedHostsDb name qname typename+ else return Nothing++-- | Check if user has permission to the full list of objects.+hasPermissionToListDb :: String -- ^ User name+ -> Permission+ -> DB Bool+hasPermissionToListDb name perm = do+ super <- isSuperUserDb name+ if super+ then return True+ else do+ any <- selectList [UserPermissionUserName ==. name, UserPermissionPermission ==. perm, UserPermissionQueueName ==. Nothing] []+ return $ not $ null any++-- | Check that user has specified permission. Fail otherwise.+checkPermission :: String -- ^ Error message for case of insufficient privileges+ -> Permission+ -> String -- ^ Queue name+ -> Action ()+checkPermission message perm qname = do+ whenAuthEnabled $ do+ user <- getAuthUser+ ok <- runDBA $ hasPermissionDb (userName user) perm qname+ when (not ok) $ do+ Scotty.raise $ InsufficientRights message++-- | Check if user can create jobs in given conditions. Fail otherwise.+checkCanCreateJobs :: String -- ^ Queue name+ -> String -- ^ Job type name+ -> T.Text -- ^ Host name. Use @__default__@ for default host of queue.+ -> Action ()+checkCanCreateJobs qname typename hostname = do+ whenAuthEnabled $ do+ user <- getAuthUser+ ok <- runDBA $ hasCreatePermissionDb (userName user) qname (Just typename) (Just hostname)+ when (not ok) $ do+ Scotty.raise $ InsufficientRights "create jobs"++-- | Check that user has permission to full list of objects. Fail otherwise.+checkPermissionToList :: String -- ^ Error message for case of insufficient privileges+ -> Permission+ -> Action ()+checkPermissionToList message perm = do+ whenAuthEnabled $ do+ user <- getAuthUser+ ok <- runDBA $ hasPermissionToListDb (userName user) perm+ when (not ok) $ do+ Scotty.raise $ InsufficientRights message+
+ src/Batchd/Daemon/CmdLine.hs view
@@ -0,0 +1,89 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE RecordWildCards #-}++module Batchd.Daemon.CmdLine+ (CommonOpts (..),+ DaemonCmdLine (..),+ Admin (..), AdminCmdLine (..),+ daemonParserInfo,+ adminParserInfo+ ) where++import Data.Semigroup ((<>))+import Options.Applicative++import Batchd.Common.Types++data CommonOpts = CommonOpts {+ globalConfigPath :: Maybe FilePath+ }+ deriving (Eq, Show)++data DaemonCmdLine = DaemonCmdLine {+ daemonCommon :: CommonOpts+ , daemonMode :: DaemonMode+ }+ deriving (Eq, Show)++data AdminCmdLine = AdminCmdLine {+ adminCommon :: CommonOpts+ , adminCommand :: Admin+ }+ deriving (Eq, Show)++data Admin =+ CreateSuperuser {username :: String}+ | Passwd {username :: String}+ | Migrate+ deriving (Show, Eq)++commonOpts :: Parser CommonOpts+commonOpts = CommonOpts+ <$> optional (strOption+ ( long "config"+ <> short 'c'+ <> metavar "PATH"+ <> help "path to global configuration file (batchd.yaml)"))++pDaemonMode :: Parser DaemonMode+pDaemonMode =+ hsubparser+ ( command "both" (info (pure Both) (progDesc "run both manager and dispatcher"))+ <> command "manager" (info (pure Manager) (progDesc "run manager"))+ <> command "dispatcher" (info (pure Dispatcher) (progDesc "run dispatcher"))+ )+ <|> pure Both++daemonParser :: Parser DaemonCmdLine+daemonParser = DaemonCmdLine <$> commonOpts <*> pDaemonMode++daemonParserInfo :: ParserInfo DaemonCmdLine+daemonParserInfo = info (daemonParser <**> helper)+ (fullDesc+ <> header "batchd - the batchd toolset daemon server-side program"+ <> progDesc "process client requests and / or execute batch jobs" )++createSuperuser :: Parser Admin+createSuperuser = CreateSuperuser+ <$> strArgument (metavar "NAME" <> help "name of user to create" <> value "root" <> showDefault)++passwd :: Parser Admin+passwd = Passwd <$> strArgument (metavar "NAME" <> help "name of user to update" <> value "root" <> showDefault)++pAdminCommand :: Parser Admin+pAdminCommand =+ hsubparser+ ( command "create-superuser" (info createSuperuser (progDesc "create super user"))+ <> command "passwd" (info passwd (progDesc "change user password"))+ <> command "upgrade-db" (info (pure Migrate) (progDesc "upgrade database to current version of batchd")))++adminParser :: Parser AdminCmdLine+adminParser = AdminCmdLine <$> commonOpts <*> pAdminCommand++adminParserInfo :: ParserInfo AdminCmdLine+adminParserInfo = info (adminParser <**> helper)+ (fullDesc+ <> header "batchd-admin - the batchd toolset administrative utility")+
+ src/Batchd/Daemon/Crypto.hs view
@@ -0,0 +1,24 @@++module Batchd.Daemon.Crypto where++import Crypto.Hash+import Crypto.Random++import Batchd.Core.Common.Types++-- for generating the salt+randomString :: Int -> IO String+randomString size = do+ drg <- getSystemDRG+ let (bytes, _) = randomBytesGenerate size drg+ return $ bstrToString bytes++randomSalt :: IO String+randomSalt = randomString 16++calcHash :: String -> String -> String -> String+calcHash password dynamicSalt staticSalt =+ let str = password <> dynamicSalt <> staticSalt+ bstr = stringToBstr str+ in show (hash bstr :: Digest SHA3_256)+
+ src/Batchd/Daemon/Database.hs view
@@ -0,0 +1,521 @@+{-# LANGUAGE EmptyDataDecls #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE QuasiQuotes #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE TypeSynonymInstances #-}+{-# LANGUAGE NoMonomorphismRestriction #-}+module Batchd.Daemon.Database where++import Control.Monad+import Control.Monad.Reader+import qualified Control.Monad.State as State++import Data.Time+import Database.Persist+import Data.Maybe+import Data.Int+import Data.Monoid ((<>))+import qualified Data.Map as M+import qualified Data.Text as T+import qualified Data.Text.Encoding as TE++import Database.Persist.Sql as Sql+import Database.Persist.Sqlite as Sqlite+import Database.Persist.Postgresql as Postgres+import qualified Database.Persist.SqlBackend as SqlBackend+import qualified Database.Esqueleto as E+import Database.Esqueleto ((^.))+import System.Log.Heavy++import Batchd.Core.Common.Types+import Batchd.Daemon.Types+import Batchd.Common.Types+import Batchd.Common.Data+import Batchd.Common.Schedule+import Batchd.Daemon.Schedule++getPool :: GlobalConfig -> LoggingTState -> IO Sql.ConnectionPool+getPool cfg lts =+ case dbcDriver cfg of+ Sqlite -> runLoggingT (Sqlite.createSqlitePool (dbcConnectionString cfg) (dbcConnectionPoolSize cfg)) lts+ PostgreSql -> do+ let str = TE.encodeUtf8 (dbcConnectionString cfg)+ runLoggingT (Postgres.createPostgresqlPool str (dbcConnectionPoolSize cfg)) lts++connectPool :: Daemon Sql.ConnectionPool+connectPool = do+ cfg <- askConfig+ lts <- askLoggingStateM+ pool <- liftIO $ getPool cfg lts+ Daemon $ lift $ State.modify $ \st -> st {ciPool = Just pool}+ return pool++buildJobInfo :: Int64 -> Job -> Maybe JobResult -> JobParamInfo -> JobInfo+buildJobInfo jid j mbr params =+ JobInfo {+ jiId = jid,+ jiQueue = jobQueueName j,+ jiType = jobTypeName j,+ jiCreateTime = jobCreateTime j,+ jiStartTime = jobStartTime j,+ jiSeq = jobSeq j,+ jiUserName = jobUserName j,+ jiStatus = jobStatus j,+ jiTryCount = jobTryCount j,+ jiHostName = jobHostName j,+ jiNotes = jobNotes j,+ jiResultTime = fmap jobResultTime mbr,+ jiExitCode = jobResultExitCode =<< mbr,+ jiStdout = fmap jobResultStdout mbr,+ jiStderr = fmap jobResultStderr mbr,+ jiParams = params+ }++loadJob :: Key Job -> DB JobInfo+loadJob jkey@(JobKey (SqlBackendKey jid)) = do+ mbJob <- get jkey+ case mbJob of+ Nothing -> throwR JobNotExists+ Just j -> do+ mbr <- selectFirst [JobResultJobId ==. jkey] [Desc JobResultTime]+ ps <- selectList [JobParamJobId ==. jkey] []+ let params = M.fromList [(jobParamName p, jobParamValue p) | p <- map entityVal ps]+ return $ buildJobInfo jid j (fmap entityVal mbr) params++loadJob' :: Int64 -> DB JobInfo+loadJob' jid = loadJob (JobKey (SqlBackendKey jid))++canLock :: DB Bool+canLock = do+ -- Dirty hack: we can't do "for update" in sqlite+ backend <- ask+ return $ SqlBackend.getRDBMS backend /= "sqlite"++lockJob :: JobInfo -> DB ()+lockJob ji = do+ ok <- canLock+ when ok $ do+ let qname = jiQueue ji+ seq = jiSeq ji+ E.select $+ E.from $ \job -> do+ E.where_ $ (job ^. JobQueueName `equals` E.val qname) `eand` (job ^. JobSeq `equals` E.val seq)+ E.locking E.ForUpdate+ return ()++lockQueue :: String -> DB ()+lockQueue qname = do+ ok <- canLock+ when ok $ do+ E.select $+ E.from $ \queue -> do+ E.where_ (queue ^. QueueName `equals` E.val qname)+ E.locking E.ForUpdate+ return ()++loadJobSeq :: String -> Int -> DB JobInfo+loadJobSeq qname seq = do+ mbJe <- getBy (UniqJobSeq qname seq)+ case mbJe of+ Nothing -> throwR JobNotExists+ Just je -> do+ let jkey = entityKey je+ let JobKey (SqlBackendKey jid) = jkey+ mbr <- selectFirst [JobResultJobId ==. jkey] [Desc JobResultTime]+ ps <- selectList [JobParamJobId ==. jkey] []+ let j = entityVal je+ let params = M.fromList [(jobParamName p, jobParamValue p) | p <- map entityVal ps]+ return $ buildJobInfo jid j (fmap entityVal mbr) params++updateJob :: Int64 -> UpdateList Job -> DB ()+updateJob jid (UpdateList updates) = do+ let jkey = JobKey (SqlBackendKey jid)+ update jkey updates++moveJob :: Int64 -> String -> DB ()+moveJob jid qname = do+ let jkey = JobKey (SqlBackendKey jid)+ oldJob <- loadJob' jid+ lockQueue (jiQueue oldJob)+ lockQueue qname+ seq <- getLastJobSeq qname+ update jkey [JobQueueName =. qname, JobSeq =. (seq+1)]++prioritizeJob :: Int64 -> MoveAction -> DB ()+prioritizeJob jid action = do+ let jkey = JobKey (SqlBackendKey jid)+ oldJob <- loadJob' jid+ let qname = jiQueue oldJob+ let oldSeq = jiSeq oldJob+ lockQueue qname+ case action of+ Last -> do+ lastSeq <- getLastJobSeq qname+ let seq' = lastSeq + 1+ update jkey [JobSeq =. seq']+ First -> do+ lastSeq <- getFirstJobSeq qname+ let seq' = lastSeq - 1+ update jkey [JobSeq =. seq']+ _ -> do+ let next = action == Less+ (mbJid, seq') <- getAdjJob next qname oldSeq+ update jkey [JobSeq =. seq']+ case mbJid of+ Nothing -> return ()+ Just jid' -> do+ let jkey' = JobKey (SqlBackendKey jid')+ update jkey' [JobSeq =. oldSeq]+ ++setJobStatus :: JobInfo -> JobStatus -> DB ()+setJobStatus ji status = do+ updateWhere [JobQueueName ==. jiQueue ji, JobSeq ==. jiSeq ji] [JobStatus =. status]++increaseTryCount :: JobInfo -> DB Int+increaseTryCount ji = do+ let qname = jiQueue ji+ seq = jiSeq ji+ mbJe <- getBy (UniqJobSeq qname seq)+ case mbJe of+ Nothing -> throwR JobNotExists+ Just je -> do+ let count = jobTryCount (entityVal je)+ count' = count + 1+ updateWhere [JobQueueName ==. qname, JobSeq ==. seq] [JobTryCount =. count']+ return count'++moveToEnd :: JobInfo -> DB ()+moveToEnd ji = do+ let qname = jiQueue ji+ seq = jiSeq ji+ lockQueue qname+ mbJe <- getBy (UniqJobSeq qname seq)+ case mbJe of+ Nothing -> throwR JobNotExists+ Just je -> do+ lastSeq <- getLastJobSeq qname+ let seq' = lastSeq + 1+ updateWhere [JobQueueName ==. qname, JobSeq ==. seq] [JobSeq =. seq', JobStatus =. New]++getAllQueues :: DB [Entity Queue]+getAllQueues = selectList [] []++getEnabledQueues :: DB [Entity Queue]+getEnabledQueues = selectList [QueueEnabled ==. True] []++getDisabledQueues :: DB [Entity Queue]+getDisabledQueues = selectList [QueueEnabled ==. False] []++getDisabledQueues' :: DB [Queue]+getDisabledQueues' = map entityVal `fmap` getDisabledQueues++getAllQueues' :: DB [Queue]+getAllQueues' = map entityVal `fmap` selectList [] []++getAllowedQueues :: String -> Permission -> DB [Queue]+getAllowedQueues name perm = do+ lst <- E.select $ E.distinct $+ E.from $ \(queue `E.InnerJoin` uperm) -> do+ E.on $ (uperm ^. UserPermissionUserName `equals` E.val name)+ `eand` (uperm ^. UserPermissionPermission `equals` E.val perm)+ `eand` ((uperm ^. UserPermissionQueueName `equals` E.just (queue ^. QueueName))+ `eor` (E.isNothing $ uperm ^. UserPermissionQueueName))+ return queue+ return $ map entityVal lst++enableQueue :: String -> DB ()+enableQueue qname = do+ update (QueueKey qname) [QueueEnabled =. True]++getAllJobs :: String -> DB [Entity Job]+getAllJobs qid = selectList [JobQueueName ==. qid] [Asc JobSeq]++getJobs :: String -> Maybe JobStatus -> DB [Entity Job]+getJobs qname mbStatus = do+ let filt = case mbStatus of+ Nothing -> []+ Just status -> [JobStatus ==. status]+ selectList ([JobQueueName ==. qname] ++ filt) [Asc JobSeq]++loadJobs :: String -> Maybe JobStatus -> DB [JobInfo]+loadJobs qname mbStatus = do+ mbQ <- getQueue qname+ case mbQ of+ Nothing -> throwR (QueueNotExists qname)+ Just _ -> do+ jes <- getJobs qname mbStatus+ forM jes $ \je -> do+ loadJob (entityKey je)++loadJobsByStatus :: Maybe JobStatus -> DB [JobInfo]+loadJobsByStatus mbStatus = do+ let filt = case mbStatus of+ Nothing -> []+ Just status -> [JobStatus ==. status]+ jes <- selectList filt [Asc JobId]+ forM jes $ \je -> do+ loadJob (entityKey je)++getNewJobsCount :: String -> DB Int+getNewJobsCount qname = do+ lst <- E.select $+ E.from $ \job -> do+ E.where_ $ job ^. JobStatus `equals` E.val New+ return E.countRows+ case lst of+ [] -> return 0+ [E.Value cnt] -> return cnt+ _ -> throwR $ UnknownError "impossible: query for new jobs count returned more than one row"++getJobResult :: Int64 -> DB JobResult+getJobResult jid = do+ let jkey = JobKey (SqlBackendKey jid)+ r <- selectFirst [JobResultJobId ==. jkey] [Desc JobResultTime]+ case r of+ Nothing -> throwR JobNotExists+ Just res -> return $ entityVal res++getJobResultE :: Int64 -> DB (Entity JobResult)+getJobResultE jid = do+ let jkey = JobKey (SqlBackendKey jid)+ r <- selectFirst [JobResultJobId ==. jkey] [Desc JobResultTime]+ case r of+ Nothing -> throwR JobNotExists+ Just res -> return res++getJobResults :: Int64 -> DB [JobResult]+getJobResults jid = do+ let jkey = JobKey (SqlBackendKey jid)+ res <- selectList [JobResultJobId ==. jkey] [Asc JobResultTime]+ return $ map entityVal res++equals :: (PersistField typ) =>+ E.SqlExpr (E.Value typ) -> E.SqlExpr (E.Value typ) -> E.SqlExpr (E.Value Bool)+equals = (E.==.)+infix 4 `equals`++leq :: (PersistField typ) =>+ E.SqlExpr (E.Value typ) -> E.SqlExpr (E.Value typ) -> E.SqlExpr (E.Value Bool)+leq = (E.<=.)+infix 4 `leq`++geq :: (PersistField typ) =>+ E.SqlExpr (E.Value typ) -> E.SqlExpr (E.Value typ) -> E.SqlExpr (E.Value Bool)+geq = (E.>=.)+infix 4 `geq`++eand :: E.SqlExpr (E.Value Bool) -> E.SqlExpr (E.Value Bool) -> E.SqlExpr (E.Value Bool)+eand = (E.&&.)+infixr 3 `eand`++eor :: E.SqlExpr (E.Value Bool) -> E.SqlExpr (E.Value Bool) -> E.SqlExpr (E.Value Bool)+eor = (E.||.)+infixr 3 `eor`++getLastJobSeq :: String -> DB Int+getLastJobSeq qid = do+ lst <- E.select $+ E.from $ \job -> do+ E.where_ (job ^. JobQueueName `equals` E.val qid)+ return $ E.max_ (job ^. JobSeq)+ case map E.unValue lst of+ (Just r:_) -> return r+ _ -> return 0++getFirstJobSeq :: String -> DB Int+getFirstJobSeq qid = do+ lst <- E.select $+ E.from $ \job -> do+ E.where_ (job ^. JobQueueName `equals` E.val qid)+ return $ E.min_ (job ^. JobSeq)+ case map E.unValue lst of+ (Just r:_) -> return r+ _ -> return 1++getAdjJob :: Bool -> String -> Int -> DB (Maybe Int64, Int)+getAdjJob next qname seq = do+ let cmp = if next then (E.>.) else (E.<.)+ let aggr = if next then E.min_ else E.max_+ lst <- E.select $+ E.from $ \job -> do+ E.where_ ((job ^. JobQueueName `equals` E.val qname) `eand` (job ^. JobSeq `cmp` E.val seq))+ return $ aggr (job ^. JobSeq)+ seq' <- case map E.unValue lst of+ (Just r:_) -> return r+ _ -> return $ if next then (seq+1) else (seq-1)+ mbJob <- getBy (UniqJobSeq qname seq')+ let mbJid = case entityKey `fmap` mbJob of+ Nothing -> Nothing+ Just (JobKey (SqlBackendKey jid)) -> Just jid+ return (mbJid, seq')++getNextJob :: Key Queue -> DB (Maybe JobInfo)+getNextJob (QueueKey qname) = do+ now <- liftIO $ getCurrentTime+ lst <- E.select $+ E.from $ \job -> do+ E.where_ $ (job ^. JobQueueName `equals` E.val qname)+ `eand` (job ^. JobStatus `equals` E.val New)+ `eand` ((E.isNothing (job ^. JobStartTime)) `eor` ((job ^. JobStartTime) `leq` (E.val (Just now))))+ return $ E.min_ (job ^. JobSeq)+ case map E.unValue lst of+ (Just seq:_) -> do+ jinfo <- loadJobSeq qname seq+ return $ Just jinfo+ _ -> return Nothing++deleteQueue :: String -> Bool -> DB ()+deleteQueue name forced = do+ mbQueue <- get (QueueKey name)+ case mbQueue of+ Nothing -> throwR (QueueNotExists name)+ Just qe -> do+ js <- selectFirst [JobQueueName ==. name] []+ if isNothing js || forced+ then delete (QueueKey name)+ else throwR QueueNotEmpty++addQueue :: Queue -> DB (Key Queue)+addQueue q = do+ r <- getQueue (queueName q)+ case r of+ Nothing -> do+ mbSchedule <- get (queueSchedule q)+ case mbSchedule of+ Nothing -> throwR (ScheduleNotExists (queueScheduleName q))+ otherwise -> insert q+ Just _ -> throwR (QueueExists (queueName q))++addQueue' :: String -> String -> DB (Key Queue)+addQueue' name scheduleId = do+ r <- insertUnique $ Queue name name True scheduleId Nothing Nothing+ case r of+ Just qid -> return qid+ Nothing -> throwR $ QueueExists name++getQueue :: String -> DB (Maybe Queue)+getQueue name = get (QueueKey name)++updateQueue :: String -> (UpdateList Queue) -> DB ()+updateQueue qname (UpdateList updates) = do+ update (QueueKey qname) updates++getQueueStats :: String -> DB (ByStatus Int)+getQueueStats qname = do+ let sql = "select status, count(1) from job where queue_name = ? group by status"+ pvs <- Sql.rawSql sql [toPersistValue qname]+ return $ ByStatus $ M.fromList [(unSingle st, unSingle cnt) | (st,cnt) <- pvs]++getStats :: DB (M.Map String (ByStatus Int))+getStats = do+ queues <- getAllQueues'+ rs <- forM queues $ \q -> do+ st <- getQueueStats (queueName q)+ return (queueName q, st)+ return $ M.fromList rs++enqueue :: String -> String -> JobInfo -> DB (Key Job)+enqueue username qname jinfo = do+ mbQueue <- getQueue qname+ lockQueue qname+ case mbQueue of+ Nothing -> throwR (QueueNotExists qname)+ Just qe -> do+ seq <- getLastJobSeq qname+ now <- liftIO getCurrentTime+ checkStartTime (jiStartTime jinfo) (queueSchedule qe)+ let job = Job (jiType jinfo) qname (seq+1) username now (jiStartTime jinfo) (jiStatus jinfo) (jiTryCount jinfo) (jiHostName jinfo) (jiNotes jinfo)+ jid <- insert job+ forM_ (M.assocs $ jiParams jinfo) $ \(name,value) -> do+ let param = JobParam jid name value+ insert_ param+ return jid++checkStartTime :: Maybe UTCTime -> Key Schedule -> DB ()+checkStartTime Nothing _ = return ()+checkStartTime (Just startTime) scheduleId = do+ schedule <- loadSchedule scheduleId+ when (not (schedule `allowsU` startTime)) $+ throwR InvalidStartTime++removeJob :: String -> Int -> DB ()+removeJob qname jseq = do+ deleteCascadeWhere [JobQueueName ==. qname, JobSeq ==. jseq]++removeJobById :: Int64 -> DB ()+removeJobById jid = do+ let key = JobKey (SqlBackendKey jid)+ deleteCascade key++removeJobs :: String -> JobStatus -> DB ()+removeJobs qname status = do+ deleteCascadeWhere [JobQueueName ==. qname, JobStatus ==. status]++cleanupJobResults :: Int -> DB ()+cleanupJobResults days = do+ now <- liftIO $ getCurrentTime+ let delta = fromIntegral $ days * 24 * 3600+ let edge = addUTCTime (negate delta) now+ -- Delete jobs with obsolete results+ -- let deleteParams = "delete from job_param where job_id in (select job_id from job_result where time < ?) and (select status from job where id = job_param.job_id) in ('Done', 'Failed')"+ -- Sql.rawExecute deleteParams [toPersistValue edge]+ -- liftIO $ putStrLn deleteParams+ -- let deleteJobs = "delete from job where id in (select job_id from job_result where time < ?) and status in ('Done', 'Failed')"+ -- liftIO $ putStrLn deleteJobs+ -- Sql.rawExecute deleteJobs [toPersistValue edge]+ -- Delete obsolete job results+ deleteCascadeWhere [JobResultTime <. edge]+ deleteCascadeWhere [JobCreateTime <. edge, JobStatus <-. [Done, Failed]]++-- cleanupJobs :: Int -> DB ()+-- cleanupJobs days = do+-- now <- liftIO $ getCurrentTime+-- let delta = fromIntegral $ days * 3600+-- let edge = addUTCTime (negate delta) now+ -- let deleteResults = "delete from job_result where job_id in (select job_id from job_result where time < ?)"+ -- Sql.rawExecute deleteResults [toPersistValue edge]+ -- let deleteJobs = "delete from job where time ++getMetrics :: Maybe T.Text -> DB [MetricRecord]+getMetrics mbPrefix = do+ lst <- E.select $+ E.from $ \record -> do+ case mbPrefix of+ Nothing -> return ()+ Just prefix -> E.where_ (record ^. MetricRecordName `E.like` E.val (prefix <> "%"))+ E.orderBy [E.desc (record ^. MetricRecordTime), E.asc (record ^. MetricRecordName)]+ return record+ return $ map entityVal lst++getLastMetric :: T.Text -> DB MetricRecord+getLastMetric name = do+ lst <- E.select $+ E.from $ \record -> do+ E.where_ (record ^. MetricRecordName `equals` E.val name)+ E.orderBy [E.desc (record ^. MetricRecordTime), E.asc (record ^. MetricRecordName)]+ return record+ case lst of+ [] -> throwR $ MetricNotExists (T.unpack name)+ (r:_) -> return $ entityVal r++queryMetric :: T.Text -> UTCTime -> UTCTime -> DB [MetricRecord]+queryMetric nameLike from to = do+ lst <- E.select $+ E.from $ \record -> do+ E.where_ ((record ^. MetricRecordName `E.like` E.val nameLike)+ `eand` (record ^. MetricRecordTime `geq` E.val from)+ `eand` (record ^. MetricRecordTime `leq` E.val to))+ E.orderBy [E.asc (record ^. MetricRecordTime)]+ return record+ return $ map entityVal lst+
+ src/Batchd/Daemon/Dispatcher.hs view
@@ -0,0 +1,175 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE TemplateHaskell #-}++module Batchd.Daemon.Dispatcher (runDispatcher) where++import Control.Concurrent+import Control.Monad+import Control.Monad.Reader+import qualified Data.Text as T+import qualified Data.Map as M+import Data.Time+import Data.Dates+import Database.Persist+import qualified Database.Persist.Sql as Sql hiding (Single)+import System.Exit+import Data.Text.Format.Heavy+import System.Log.Heavy++import Batchd.Daemon.Types+import Batchd.Common.Types+import Batchd.Common.Data+import Batchd.Common.Config as Config+import Batchd.Daemon.Database+import Batchd.Common.Schedule+import Batchd.Daemon.Schedule+import Batchd.Daemon.Executor+import Batchd.Daemon.Hosts (hostCleaner, hostsMetricDumper)+import Batchd.Core.Daemon.Logging+import Batchd.Core.Daemon.Hosts++-- | Set up workers, callback listener and dispatcher itself.+runDispatcher :: Daemon ()+runDispatcher = do+ cfg <- askConfig+ jobsChan <- liftIO newChan+ resChan <- liftIO newChan+ counters <- liftIO $ newMVar M.empty+ $info "Starting {} workers..." (Single $ dcWorkers $ dbcDispatcher cfg)+ -- Each worker will run in separate thread.+ -- All workers read jobs to be executed from single Chan.+ -- So job put into Chan will be executed by first worker who sees it.+ forM_ [1.. dcWorkers (dbcDispatcher cfg)] $ \idx -> do+ $debug " Starting worker #{}" (Single idx)+ forkDaemon "worker" $ worker idx counters jobsChan resChan++ -- Listen for job results+ forkDaemon "job results listener" $ callbackListener resChan++ -- Stop hosts when they are not needed anymore+ withLogVariable "thread" ("host cleaner" :: String) $ do+ lts <- askLoggingStateM+ liftIO $ forkIO $ hostCleaner lts counters++ forkDaemon "hosts metrics dumper" $ hostsMetricDumper counters++ dispatcher jobsChan++-- | Dispatcher main loop itself.+dispatcher :: Chan (Queue, JobInfo) -> Daemon ()+dispatcher jobsChan = withLogVariable "thread" ("dispatcher" :: String) $ do+ forever $ do+ qesr <- runDB getEnabledQueues+ cfg <- askConfig+ case qesr of+ Left err -> $reportError "Can't get list of enabled queues: {}" (Single $ Shown err)+ Right qes -> do+ forM_ qes $ \qe -> runDB $ do+ schedule <- loadSchedule (queueSchedule $ entityVal qe)+ now <- liftIO $ getCurrentDateTime+ when (schedule `allows` now) $ do+ -- will work only with queues, shedules of which allow current time+ let QueueKey qname = entityKey qe+ -- pick next job from the queue+ mbJob <- getNextJob (entityKey qe)+ case mbJob of+ Nothing -> $debug "Queue `{}' exhaused." (Single qname)+ Just job -> do+ -- Waiting means that Dispatcher saw job and put it to Chan to be+ -- picked by some of workers.+ setJobStatus job Waiting+ liftIO $ writeChan jobsChan (entityVal qe, job)+ -- Sleep for poll timeout+ liftIO $ threadDelay $ (dcPollTimeout $ dbcDispatcher cfg) * 1000*1000++-- | This listens for job results Chan and writes results to DB.+-- It also reschedules failed jobs if needed.+callbackListener :: ResultsChan -> Daemon ()+callbackListener resChan = forever $ do+ (job, command) <- liftIO $ readChan resChan+ let jid = jiId job+ let jkey = JobKey (Sql.SqlBackendKey jid)+ now <- liftIO getCurrentTime+ + withJobContext job $ do+ case command of+ StartExecution -> do+ -- let result = JobResult jkey now Nothing T.empty T.empty+ -- catchDbError $ runDB $ insert_ result+ return ()++ StdoutLine line -> do+ let result = JobResult jkey now Nothing line T.empty+ catchDbError $ runDB $ do+ insert_ result++ StderrLine line -> do+ let result = JobResult jkey now Nothing T.empty line+ catchDbError $ runDB $ do+ insert_ result++ ExecError msg onFail -> do+ let result = JobResult jkey now Nothing T.empty msg+ catchDbError $ runDB $ do+ insert_ result+ processFail job onFail++ Exited ec onFail -> do+ let result = JobResult jkey now (Just ec) T.empty T.empty+ catchDbError $ runDB $ do+ insert_ result+ if ec == ExitSuccess+ then setJobStatus job Done+ else processFail job onFail+ where+ catchDbError db = do+ res <- db+ case res of+ Right _ -> return ()+ Left err -> do+ $reportError "Can't update job result: {}" (Single $ show err)+ return ()++ processFail job Continue = setJobStatus job Failed -- just mark job as Failed+ processFail job (RetryNow m) = do+ count <- increaseTryCount job+ if count <= m+ then do+ $info "Retry now" ()+ setJobStatus job New -- job will be picked up by dispatcher at nearest iteration.+ else setJobStatus job Failed+ processFail job (RetryLater m) = do+ count <- increaseTryCount job+ if count <= m+ then do+ $info "Retry later" ()+ moveToEnd job -- put the job to the end of queue.+ else setJobStatus job Failed++withJobContext :: HasLogContext m => JobInfo -> m a -> m a+withJobContext job =+ withLogContext (LogContextFrame vars noChange)+ where+ vars = [("job", Variable (jiId job)),+ ("user", Variable (jiUserName job))]++-- | Worker loop executes jobs themeselves+worker :: Int -> HostsPool -> Chan (Queue, JobInfo) -> ResultsChan -> Daemon ()+worker idx hosts jobsChan resChan = forever $ withLogVariable "worker" idx $ do+ (queue, job) <- liftIO $ readChan jobsChan+ withJobContext job $ do+ $info "got job #{}" (Single $ jiId job)+ -- now job is picked up by worker, mark it as Processing+ runDB $ setJobStatus job Processing+ jtypeR <- liftIO $ Config.loadTemplate (T.pack $ jiType job)+ case jtypeR of+ Left err -> do -- we could not load job type description+ $reportError "Invalid job type {}: {}" (jiType job, show err)+ liftIO $ writeChan resChan (job, ExecError (T.pack $ show err) Continue)++ Right jtype -> do+ executeJob hosts queue jtype job resChan++ $info "done job #{}." (Single $ jiId job)+
+ src/Batchd/Daemon/Executor.hs view
@@ -0,0 +1,115 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TemplateHaskell #-}++module Batchd.Daemon.Executor where++import Control.Monad+import Control.Monad.Trans+import Control.Concurrent+import Data.Maybe (fromMaybe)+import qualified Data.Map as M+import qualified Data.Text as T+import qualified Data.Text.Lazy as TL+import Data.Text.Format.Heavy+import Data.Text.Format.Heavy.Parse.Shell+import Data.Text.Format.Heavy.Parse.Braces (parseFormat')+import qualified Database.Persist.Sql as Sql hiding (Single)+-- import Data.Time+import System.FilePath+import System.Exit (ExitCode (..))+import Network.SSH.Client.LibSSH2.Conduit (execCommand)++import Batchd.Core.Common.Types+import Batchd.Core.Common.Config+import Batchd.Core.Daemon.Logging+import Batchd.Common.Types+import Batchd.Common.Data+import Batchd.Daemon.Types+import Batchd.Core.Daemon.Hosts+import Batchd.Daemon.SSH+import Batchd.Daemon.Local (processOnLocalhost, withLocalScript)+import Batchd.Daemon.Hosts (loadHostController, withHost)+import Batchd.Daemon.Monitoring as Monitoring++getCommands :: GlobalConfig -> Maybe Host -> JobType -> JobInfo -> [T.Text]+getCommands cfg mbHost jt job =+ let context = mkContext $ hostContext mbHost jt $ jiParams job+ syntax = fromMaybe (dbcDefTemplateSyntax cfg) (jtSyntax jt)+ parse = case syntax of+ Shell -> parseShellFormat'+ Python -> parseFormat'+ in [TL.toStrict $ format (parse $ TL.fromStrict line) context | line <- jtTemplate jt]+ where+ mkContext m = optional $ m `ThenCheck` hostVars `ThenCheck` dbcVariables cfg+ hostVars = maybe M.empty hVariables mbHost++getHostName :: Queue -> JobType -> JobInfo -> Maybe T.Text+getHostName q jt job = T.pack <$>+ msum [jiHostName job, jtHostName jt, queueHostName q]++hostContext :: Maybe Host -> JobType -> JobParamInfo -> JobParamInfo+hostContext Nothing jt params = params+hostContext (Just host) jt params = M.fromList $ map update $ M.assocs params+ where+ update (key, value) =+ case getParamType jt key of+ Just InputFile -> (key, T.pack $ hInputDirectory host </> takeFileName (T.unpack value))+ Just OutputFile -> (key, T.pack $ hOutputDirectory host </> takeFileName (T.unpack value))+ _ -> (key, value)++processOnHost :: HostsPool -> Host -> JobType -> JobInfo -> ResultsChan -> FilePath -> [T.Text] -> Daemon ExitCode+processOnHost counters host jtype job resultChan scriptsDir commands = do+ host' <- liftIO $ accountForDefaultHostKeys host+ lts <- askLoggingStateM+ controller <- liftIO $ loadHostController lts (hController host')+ r <- withHost counters host' jtype $ do+ withSshOnHost controller host' $ \session -> do+ withRemoteScript session scriptsDir (jiId job) commands $ \command -> do+ uploadFiles (getInputFiles jtype job) (hInputDirectory host') session+ $info "EXECUTING: {}" (Single command)+ (Just commandHandle, commandOutput) <- liftIO $ execCommand True session command+ ec <- liftIO $ retrieveOutput job jtype commandHandle commandOutput resultChan+ $info "Done, exit code is {}." (Single $ show ec)+ downloadFiles (hOutputDirectory host') (getOutputFiles jtype job) session+ return ec+ case r of+ Left e -> do+ $reportError "Error while executing job at host `{}': {}" (hName host', show e)+ liftIO $ writeChan resultChan (job, ExecError (T.pack $ show e) (jtOnFail jtype))+ return $ ExitFailure (-1)+ Right result -> return result++executeJob :: HostsPool -> Queue -> JobType -> JobInfo -> ResultsChan -> Daemon ()+executeJob counters q jt job resultChan = do+ let mbHostName = getHostName q jt job+ hostForMetric = fromMaybe "localhost" mbHostName+ let metrics = ["batchd.job.duration", + "batchd.job.duration.host." <> hostForMetric,+ "batchd.job.duration.type." <> T.pack (jtName jt)]+ Monitoring.timedN metrics $ do+ cfg <- askConfig+ liftIO $ writeChan resultChan (job, StartExecution)+ let jid = JobKey (Sql.SqlBackendKey $ jiId job)+ case mbHostName of+ Nothing -> do -- localhost+ let commands = getCommands cfg Nothing jt job+ let scriptsDir = dbcDefScriptsDirectory cfg+ liftIO $ withLocalScript scriptsDir (jiId job) commands $ \script ->+ processOnLocalhost job (jtOnFail jt) script resultChan++ Just hostname -> do+ hostR <- liftIO $ loadHost hostname+ case hostR of+ Right host -> do+ let commands = getCommands cfg (Just host) jt job+ let scriptsDir = fromMaybe (dbcDefScriptsDirectory cfg) (hScriptsDirectory host)+ -- now <- liftIO $ getCurrentTime+ -- let result = JobResult jid now (ExitFailure (-2)) T.empty T.empty+ $(putMessage config_level) "Loaded host configuration: {}" (Single $ show host)+ processOnHost counters host jt job resultChan scriptsDir commands+ return ()+ Left err -> do+ $reportError "Error while executing job: {}" (Single $ Shown err)+ liftIO $ writeChan resultChan (job, ExecError (T.pack $ show err) (jtOnFail jt))+ return ()+
+ src/Batchd/Daemon/Hosts.hs view
@@ -0,0 +1,290 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Batchd.Daemon.Hosts where++import Control.Monad+import Control.Concurrent+import Control.Exception+import Control.Monad.Trans+import qualified Control.Monad.Catch as MC+import qualified Data.Map as M+import qualified Data.HashMap.Strict as H+import Data.Time+import Data.Aeson as Aeson+import qualified Data.Text as T+import qualified Data.Text.Lazy as TL+import Data.Text.Format.Heavy+import Data.Text.Format.Heavy.Time () -- import instances only+import Data.Text.Format.Heavy.Parse.Shell+import System.Log.Heavy+import System.Exit (ExitCode (..))++import Batchd.Core.Common.Types+import Batchd.Core.Common.Config+import Batchd.Daemon.Types+import Batchd.Core.Daemon.Hosts+import Batchd.Core.Daemon.Logging+import Batchd.Common.Types+import Batchd.Daemon.Monitoring as Monitoring+import Batchd.Daemon.SSH (execCommandsOnHost, accountForDefaultHostKeys)+import Batchd.Daemon.Local (execLocalCommands)++#ifdef LIBVIRT+import Batchd.Ext.LibVirt+#endif+#ifdef DOCKER+import Batchd.Ext.Docker+#endif+#ifdef AWSEC2+import Batchd.Ext.AWS+#endif+#ifdef LINODE+import Batchd.Ext.Linode+#endif++supportedDrivers :: [HostDriver]+supportedDrivers =+ [+#ifdef LIBVIRT+ libVirtDriver,+#endif+#ifdef DOCKER+ dockerDriver,+#endif+#ifdef AWSEC2+ awsEc2Driver,+#endif+#ifdef LINODE+ linodeDriver,+#endif+ localDriver+ ]++selectDriver :: String -> Maybe HostDriver+selectDriver name = go supportedDrivers+ where+ go [] = Nothing+ go (driver : rest)+ | driverName driver == name = Just driver+ | otherwise = go rest++loadHostController :: LoggingTState -> T.Text -> IO HostController+loadHostController lts "local" = do+ case initController localDriver lts undefined of+ Right controller -> return controller+ Left err -> throw $ UnknownError $ "Can't initialize local host controller: " ++ show err+loadHostController lts name = do+ cfgr <- loadHostControllerConfig name+ case cfgr of+ Left err -> throw err+ Right value@(Object cfg) -> do+ case H.lookup "driver" cfg of+ Nothing -> throw $ UnknownError $ "Invalid host controller config: driver not specified"+ Just (Aeson.String dname) -> do+ case selectDriver (T.unpack dname) of+ Nothing -> throw $ UnknownError $ "Invalid host controller config: unsupported driver name: " ++ T.unpack dname+ Just driver -> do+ case initController driver lts value of+ Right controller -> return controller+ Left err -> throw err+ Right _ -> throw $ UnknownError "Invalid host controller config: it must be a dictionary"++getMaxJobs :: Host -> JobType -> Maybe Int+getMaxJobs host jtype =+ case (hMaxJobs host, jtMaxJobs jtype) of+ (Nothing, Nothing) -> Nothing+ (Just m, Nothing) -> Just m+ (Nothing, Just m) -> Just m+ (Just x, Just y) -> Just $ max x y++waitForStatus :: MVar HostState -> [HostStatus] -> (HostState -> Daemon HostState) -> Daemon ()+waitForStatus mvar targetStatuses actions = do+ connInfo <- askConnectionInfo+ lts <- askLoggingStateM+ ok <- liftIO $ modifyMVar mvar $ \st ->+ if hsStatus st `elem` targetStatuses+ then do+ result <- runDaemonIO connInfo lts $ actions st+ return (result, True)+ else do+ debugIO lts $(here) "Host `{}' has status {}, waiting for one of {}..."+ (hName $ hsHostConfig st, show (hsStatus st), show targetStatuses)+ return (st, False)++ when (not ok) $ do+ liftIO $ threadDelay $ 10 * 1000 * 1000+ waitForStatus mvar targetStatuses actions++increaseJobCount :: HostState -> Maybe Int -> Daemon HostState+increaseJobCount st mbMaxJobs = do+ $debug "Host `{}' had {} jobs, got new one, set status to {}"+ (hName $ hsHostConfig st, hsJobCount st, show activeOrBusy)+ return $ st {hsStatus = activeOrBusy, hsJobCount = hsJobCount st + 1}+ where+ activeOrBusy =+ case mbMaxJobs of+ Nothing -> Active+ Just maxJobs -> if hsJobCount st + 1 >= maxJobs + then Busy+ else Active++decreaseJobCount :: HostState -> Daemon HostState+decreaseJobCount st = do+ let newStatus = if hsJobCount st - 1 <= 0+ then Released+ else Active+ released <- if newStatus == Released+ then Just <$> liftIO getCurrentTime+ else return Nothing+ $debug "Host `{}' had {} jobs, one done, set status to {}"+ (hName $ hsHostConfig st, hsJobCount st, show newStatus)+ return $ st {hsStatus = newStatus, hsJobCount = hsJobCount st - 1, hsReleaseTime = released}++setHostStatus :: MVar HostState -> HostStatus -> IO ()+setHostStatus mvar status =+ modifyMVar_ mvar $ \st -> return $ st {hsStatus = status}++formatHostCommand :: GlobalConfig -> Host -> T.Text -> T.Text+formatHostCommand cfg host template =+ TL.toStrict $ format (parseShellFormat' $ TL.fromStrict template) context+ where+ context = optional $ hVariables host `ThenCheck` hostParams `ThenCheck` dbcVariables cfg+ hostParams :: Variables+ hostParams = M.fromList $ [+ ("name", hName host),+ ("hostname", hHostName host),+ ("controllerId", hControllerId host),+ -- ("publicKey", hPublicKey host),+ -- ("privateKey", hPrivateKey host),+ ("passphrase", hPassphrase host),+ ("user", hUserName host),+ ("port", T.pack $ show $ hPort host),+ ("inputDirectory", T.pack $ hInputDirectory host),+ ("outputDirectory", T.pack $ hOutputDirectory host)+ ]++ensureHostStarted :: Host -> Daemon ()+ensureHostStarted host = do+ lts <- askLoggingStateM+ cfg <- askConfig+ host' <- liftIO $ accountForDefaultHostKeys host+ controller <- liftIO $ loadHostController lts (hController host')+ let name = hName host'+ $debug "Host `{}' is controlled by `{}'" (name, show controller)+ if doesSupportStartStop controller+ then do+ $debug "Starting host `{}'" (Single name)+ r <- liftIO $ startHost controller host'+ case r of+ Right _ -> do+ $debug "Waiting for host `{}' to initialize for {} seconds..." + (name, hStartupTime host')+ liftIO $ threadDelay $ 1000 * 1000 * hStartupTime host'+ let localCommands = map (formatHostCommand cfg host') (hStartupDispatcherCommands host')+ ec <- liftIO $ execLocalCommands lts localCommands+ case ec of+ ExitFailure rc -> throw $ HostInitializationError rc+ ExitSuccess -> do+ let commandsOnHost = map (formatHostCommand cfg host') (hStartupHostCommands host')+ ec <- execCommandsOnHost controller host' commandsOnHost+ case ec of+ ExitSuccess -> return ()+ ExitFailure rc -> throw $ HostInitializationError rc+ Left err -> throw err+ else $debug "Controller does not support starting hosts" ()++ensureHostStopped :: LoggingTState -> Host -> IO ()+ensureHostStopped lts host = do+ controller <- loadHostController lts (hController host)+ debugIO lts $(here) "Host `{}' is controlled by `{}'" (hName host, show controller)+ let name = hName host+ if doesSupportStartStop controller+ then do+ debugIO lts $(here) "Stopping host `{}'" (Single name)+ r <- stopHost controller (hControllerId host)+ case r of+ Right _ -> return ()+ Left err -> throw err+ else debugIO lts $(here) "Controller does not support stopping hosts" ()++acquireHost :: HostsPool -> Host -> JobType -> Daemon ()+acquireHost mvar host jtype = do+ check (hName host)+ return ()+ where+ check :: T.Text -> Daemon()+ check name = do+ counter <- liftIO $ modifyMVar mvar $ \m -> do+ case M.lookup name m of+ Just c -> do+ return (m, c)+ Nothing -> do+ c <- newMVar $ HostState Free Nothing 0 host+ let m' = M.insert name c m+ return (m', c)++ waitForStatus counter [Free, Active, Released] $ \st -> do+ when (hsStatus st == Free) $ do+ ensureHostStarted host+ increaseJobCount st $ getMaxJobs host jtype++releaseHost :: HostsPool -> Host -> Daemon ()+releaseHost mvar host = do+ let name = hName host+ counter <- liftIO $ withMVar mvar $ \m -> do+ case M.lookup name m of+ Nothing -> fail $ "Can't release host: " <> T.unpack name <> "; map: " <> show (M.keys m)+ Just c -> return c+ wrapDaemon_ (modifyMVar_ counter) $ \st -> decreaseJobCount st+ return ()++withHost :: HostsPool -> Host -> JobType -> Daemon a -> Daemon (Either SomeException a)+withHost mvar host jtype actions = withLogVariable "host" (hName host) $ do+ MC.try $ MC.bracket_ (acquireHost mvar host jtype) (releaseHost mvar host) actions++hostsMetricDumper :: HostsPool -> Daemon ()+hostsMetricDumper pool = forever $ do+ liftIO $ threadDelay $ 10 * 1000 * 1000+ hosts <- liftIO $ readMVar pool+ statuses <- forM (M.assocs hosts) $ \(name, stVar) -> do+ st <- liftIO $ readMVar stVar+ Monitoring.gauge ("batchd.host." <> name <> ".jobs") $ hsJobCount st+ Monitoring.label ("batchd.host." <> name <> ".status") $ T.pack $ show $ hsStatus st+ return $ hsStatus st+ let active = length $ filter (== Active) statuses+ let busy = length $ filter (== Busy) statuses+ Monitoring.gauge "batchd.hosts.active" active+ Monitoring.gauge "batchd.hosts.busy" busy++hostCleaner :: LoggingTState -> HostsPool -> IO ()+hostCleaner lts mvar = forever $ do+ threadDelay $ 60 * 1000 * 1000+ hosts <- readMVar mvar+ forM_ (M.assocs hosts) $ \(name, counter) -> do+ modifyMVar_ counter $ \st -> do+ if hsStatus st == Released+ then case hsReleaseTime st of+ Nothing -> do+ reportErrorIO lts $(here) "Host `{}' status is Released, but it's release time is not set"+ (Single name)+ return st+ Just releaseTime -> do+ now <- getCurrentTime+ if ceiling (diffUTCTime now releaseTime) >= hShutdownTimeout (hsHostConfig st)+ then do+ debugIO lts $(here) "Host `{}' was released at {}, it's time to shut it down"+ (name, releaseTime)+ ensureHostStopped lts (hsHostConfig st)+ return $ st {hsStatus = Free, hsReleaseTime = Nothing}+ else do+ debugIO lts $(here) "Host `{}' was relesed at {}, it's not time to shut it down yet"+ (name, releaseTime)+ return st+ else return st+
+ src/Batchd/Daemon/Local.hs view
@@ -0,0 +1,91 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TemplateHaskell #-}++module Batchd.Daemon.Local where++import Control.Monad+import Control.Monad.Trans+import Control.Concurrent+import qualified Control.Exception as E+import Data.Conduit+import Data.Maybe (fromMaybe)+import Data.Int+import qualified Data.Conduit.Combinators as C+import qualified Data.Conduit.List as CL+import Data.Conduit.Binary (sourceHandle)+import qualified Data.Map as M+import qualified Data.Text as T+import qualified Data.Text.IO as TIO+import qualified Data.Text.Lazy as TL+import Data.Text.Format.Heavy+import Data.Text.Format.Heavy.Parse.Shell+import Data.Time+import System.Process+import System.FilePath+import System.Directory+import System.IO+import System.Exit (ExitCode (..))+import System.Log.Heavy++import Batchd.Core.Common.Types+import Batchd.Core.Common.Config+import Batchd.Core.Daemon.Logging+import Batchd.Core.Daemon.Hosts+import Batchd.Common.Types+import Batchd.Common.Data+import Batchd.Daemon.Types++execLocalCommand :: LoggingTState -> T.Text -> IO ExitCode+execLocalCommand lts command = do+ infoIO lts $(here) "Executing on dispatcher host: {}" (Single command)+ let opts = (shell $ T.unpack command) {std_out = CreatePipe, std_err = CreatePipe}+ withCreateProcess opts $ \_ (Just stdout) (Just stderr) process -> do+ forkIO $ retrieveOutput "error" stderr+ retrieveOutput "output" stdout+ ec <- waitForProcess process+ infoIO lts $(here) "Exit code: {}" (Single $ show ec)+ return ec+ where+ retrieveOutput :: T.Text -> Handle -> IO ()+ retrieveOutput kind handle = do+ sourceHandle handle =$= C.decodeUtf8 =$= C.linesUnbounded $$ CL.mapM_ $ \line ->+ infoIO lts $(here) "{}> {}" (kind, line)+ hClose handle++execLocalCommands :: LoggingTState -> [T.Text] -> IO ExitCode+execLocalCommands _ [] = return ExitSuccess+execLocalCommands lts (command : commands) = do+ ec <- execLocalCommand lts command+ if ec == ExitSuccess+ then execLocalCommands lts commands+ else return ec++withLocalScript :: FilePath -> Int64 -> [T.Text] -> (String -> IO ()) -> IO ()+withLocalScript scriptsDir jobId commands actions = do+ if length commands == 1+ then actions (T.unpack $ head commands)+ else do+ let scriptText = formatScript commands+ scriptName = TL.unpack $ format "batchd_job_{}.script" (Single jobId)+ scriptPath = scriptsDir </> scriptName+ chmodX = do+ perm <- getPermissions scriptPath+ setPermissions scriptPath (setOwnerExecutable True perm)+ E.bracket_ (TIO.writeFile scriptPath scriptText >> chmodX)+ (removeFile scriptPath)+ (actions scriptPath)++processOnLocalhost :: JobInfo -> OnFailAction -> String -> ResultsChan -> IO ()+processOnLocalhost job onFail command resultChan = do+ let opts = (shell command) {std_out = CreatePipe, std_err = CreatePipe}+ withCreateProcess opts $ \_ (Just stdout) (Just stderr) process -> do+ forkIO $ retrieveOutput StderrLine stderr+ retrieveOutput StdoutLine stdout+ ec <- waitForProcess process+ writeChan resultChan (job, Exited ec onFail)+ where+ retrieveOutput cons handle = do+ sourceHandle handle =$= C.decodeUtf8 =$= C.linesUnbounded $$ CL.mapM_ $ \line ->+ writeChan resultChan (job, cons line)+ hClose handle+
+ src/Batchd/Daemon/Logging.hs view
@@ -0,0 +1,48 @@+{-# LANGUAGE TemplateHaskell, TypeFamilies, OverloadedStrings #-}+-- | This module contains utilities for logging, used by batchd daemon.+module Batchd.Daemon.Logging+ (+ getLoggingSettings+ ) where++import qualified Data.Text.Format.Heavy as F+import Language.Haskell.TH.Lift+import System.Log.Heavy+import System.Log.Heavy.TH () -- import instances only++import Batchd.Common.Types++deriveLift ''DaemonMode++deriveLift ''DbDriver++deriveLift ''AuthMode+deriveLift ''LogTarget+deriveLift ''F.FormatItem+deriveLift ''F.Format+deriveLift ''LogConfig+deriveLift ''MetricsConfig+deriveLift ''WebClientConfig+deriveLift ''ManagerConfig+deriveLift ''DispatcherConfig+deriveLift ''StorageConfig+deriveLift ''TemplateSyntax+deriveLift ''GlobalConfig++-- | Get logging settings from global config.+getLoggingSettings :: GlobalConfig -> LoggingSettings+getLoggingSettings cfg =+ case lcTarget $ dbcLogging cfg of+ LogSyslog -> LoggingSettings $ filtering fltr $ defaultSyslogSettings {ssIdent = "batchd", ssFormat = logFormat}+ LogStdout -> LoggingSettings $ filtering fltr $ defStdoutSettings {lsFormat = logFormat}+ LogStderr -> LoggingSettings $ filtering fltr $ defStderrSettings {lsFormat = logFormat}+ LogFile path -> LoggingSettings $ filtering fltr $ (defFileSettings path) {lsFormat = logFormat}+ where+ fltr :: LogFilter+ fltr = map toFilter (lcFilter $ dbcLogging cfg) ++ [([], lcLevel $ dbcLogging cfg)]++ logFormat = lcFormat (dbcLogging cfg)++ toFilter :: (String, Level) -> (LogSource, Level)+ toFilter (src, level) = (splitDots src, level)+
+ src/Batchd/Daemon/Manager.hs view
@@ -0,0 +1,665 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE RecordWildCards #-}++module Batchd.Daemon.Manager where++import Control.Concurrent+import Control.Monad+import Control.Applicative (optional)+import Control.Monad.Reader+import qualified Control.Monad.State as State+import qualified Data.Map as M+import qualified Data.ByteString as B+import qualified Data.Text as T+import qualified Data.Text.Lazy as TL+import Data.Text.Format.Heavy hiding (optional)+import Data.Text.Format.Heavy.Parse+import Data.Char (isDigit)+import Data.Maybe+import Data.Default+import Data.Yaml+import Data.Time+import Network.HTTP.Types+import qualified Network.Wai as Wai+import Network.Wai.Handler.Warp (defaultSettings, setPort)+import Network.Wai.Middleware.Cors+import Network.Wai.Middleware.Static as Static+import Web.Scotty.Trans as Scotty+import qualified Text.Parsec as Parsec+import qualified Text.Parsec.Text as Parsec+import System.FilePath+import System.FilePath.Glob+import System.Log.Heavy.Types+import System.Log.Heavy+import System.Metrics.Json as EKG++import Batchd.Core.Common.Types+import Batchd.Core.Common.Localize+import Batchd.Core.Common.Config+import Batchd.Core.Daemon.Logging+import Batchd.Common.Types+import Batchd.Common.Data+import Batchd.Common.Config+import Batchd.Daemon.Types+import Batchd.Daemon.Database+import Batchd.Daemon.Schedule+import Batchd.Daemon.Auth+import Batchd.Daemon.Monitoring as Monitoring++corsPolicy :: GlobalConfig -> CorsResourcePolicy+corsPolicy cfg =+ let origins = case wcAllowedOrigin =<< (mcWebClient $ dbcManager cfg) of+ Nothing -> Nothing+ Just url -> Just ([stringToBstr url], not (isAuthDisabled $ mcAuth $ dbcManager cfg))+ in simpleCorsResourcePolicy {+ corsOrigins = origins,+ corsMethods = ["GET", "POST", "PUT", "DELETE"]+ }++routes :: GlobalConfig -> LoggingTState -> Maybe Wai.Middleware -> ScottyT Error Daemon ()+routes cfg lts mbWaiMetrics = do+ Scotty.defaultHandler raiseError++ Scotty.middleware $ cors $ const $ Just $ corsPolicy cfg+ Scotty.middleware $ requestLogger cfg lts+ case mbWaiMetrics of+ Nothing -> return ()+ Just waiMetrics -> do+ Scotty.middleware waiMetrics+ + case mcAuth $ dbcManager cfg of+ AuthDisabled -> Scotty.middleware (noAuth cfg lts)+ AuthConfig {..} -> do+ when authHeaderEnabled $+ Scotty.middleware (headerAuth cfg lts)+ when authBasicEnabled $+ Scotty.middleware (basicAuth cfg lts)+ + case wcPath `fmap` (mcWebClient $ dbcManager cfg) of+ Just path -> do+ Scotty.middleware $ staticPolicy (addBase path)+ Scotty.get "/" $ file $ path </> "batch.html"+ Scotty.get "/monitor" $ file $ path </> "monitor.html"+ Nothing -> return ()++ Scotty.get "/stats" getStatsA+ Scotty.get "/stats/:name" getQueueStatsA++ Scotty.get "/queue" getQueuesA+ Scotty.get "/queue/:name" getQueueA+ Scotty.get "/queue/:name/jobs" getQueueJobsA+ Scotty.get "/queue/:name/type" $ getAllowedJobTypesA+ Scotty.get "/queue/:qname/type/:tname/host" $ getAllowedHostsForTypeA+ Scotty.get "/queue/:name/host" $ getAllowedHostsA+ Scotty.put "/queue/:name" updateQueueA+ Scotty.post "/queue" addQueueA+ Scotty.post "/queue/:name" enqueueA+ Scotty.delete "/queue/:name/:seq" removeJobA+ Scotty.delete "/queue/:name" removeQueueA++ Scotty.get "/job/:id" getJobA+ Scotty.get "/job/:id/results" getJobResultsA+ Scotty.get "/job/:id/results/last" getJobLastResultA+ Scotty.put "/job/:id" updateJobA+ Scotty.delete "/job/:id" removeJobByIdA+ Scotty.get "/jobs" getJobsA++ Scotty.get "/schedule" getSchedulesA+ Scotty.post "/schedule" addScheduleA+ Scotty.delete "/schedule/:name" removeScheduleA++ Scotty.get "/type" $ getJobTypesA+ Scotty.get "/type/:name" getJobTypeA++ Scotty.get "/host" $ getHostsA++ Scotty.post "/user" createUserA+ Scotty.get "/user" getUsersA+ Scotty.put "/user/:name" changePasswordA+ Scotty.get "/user/:name/permissions" getPermissionsA+ Scotty.post "/user/:name/permissions" createPermissionA+ Scotty.delete "/user/:name/permissions/:id" deletePermissionA++ Scotty.get "/monitor/current/tree" currentMetricsTreeA + Scotty.get "/monitor/current/plain" currentMetricsPlainA+ Scotty.get "/monitor/jobs" getQueueJobsCountA+ Scotty.get "/monitor/:prefix/current/tree" currentMetricsTreeA + Scotty.get "/monitor/:prefix/current/plain" currentMetricsPlainA+ Scotty.get "/monitor/:name/last" lastMetricA + Scotty.get "/monitor/:prefix/query" queryMetricA++ Scotty.options "/" $ getAuthOptionsA+ Scotty.options (Scotty.regex "/.*") $ done++requestLogger :: GlobalConfig -> LoggingTState -> Wai.Middleware+requestLogger gcfg lts app req sendResponse = do+ -- putStrLn "request logger"+ debugIO lts $(here) "Request: {method} {path}. User-Agent: {useragent}" req+ app req sendResponse++runManager :: Daemon ()+runManager = do+ connInfo <- Daemon $ lift State.get+ cfg <- askConfig+ let options = def {Scotty.settings = setPort (mcPort $ dbcManager cfg) defaultSettings}+ lts <- askLoggingStateM+ waiMetrics <- getWaiMetricsMiddleware+ forkDaemon "job metrics calculator" jobMetricsCalculator+ forkDaemon "metrics dumper" metricsDumper+ forkDaemon "queue runner" queueRunner+ forkDaemon "maintainer" maintainer+ forkDaemon "metrics cleaner" metricsCleaner+ liftIO $ do+ scottyOptsT options (runService connInfo lts) $ routes cfg lts waiMetrics+ where+ runService connInfo lts actions =+ runDaemonIO connInfo lts $+ withLogVariable "thread" ("REST service" :: String) $ actions++jobMetricsCalculator :: Daemon ()+jobMetricsCalculator = forever $ do+ liftIO $ threadDelay $ 60 * 1000*1000+ r <- runDB getStats+ case r of+ Left err -> $reportError "Can't get job statistics" ()+ Right byQueue -> do+ forM_ (M.assocs byQueue) $ \(name, ByStatus byStatus) -> do+ forM_ (M.assocs byStatus) $ \(status, count) -> do+ let gname = "batchd.queue." <> T.pack name <> ".jobs." <> T.pack (show status)+ Monitoring.gauge gname count+ let totals = M.unionsWith (+) [byStatus | ByStatus byStatus <- M.elems byQueue]+ forM_ (M.assocs totals) $ \(status, count) -> do+ let gname = "batchd.jobs." <> T.pack (show status)+ Monitoring.gauge gname count++maintainer :: Daemon ()+maintainer = forever $ do+ cfg <- askConfig+ runDB $ cleanupJobResults (scDoneJobs $ dbcStorage cfg)+ liftIO $ threadDelay $ 60 * 1000*1000++queueRunner :: Daemon ()+queueRunner = forever $ do+ qesr <- runDB getDisabledQueues'+ case qesr of+ Left err -> $reportError "Can't get list of disabled queues: {}" (Single $ show err)+ Right queues -> do+ forM_ queues $ \queue -> do+ case queueAutostartJobCount queue of+ Nothing -> return ()+ Just minCount -> do+ countRes <- runDB $ getNewJobsCount (queueName queue)+ case countRes of+ Left err -> $reportError "Can't get list of new jobs in queue `{}': {}" (queueName queue, show err)+ Right currentCount -> do+ if currentCount >= minCount+ then do+ $debug "Disabled queue `{}' has {} new jobs, enable it" (queueName queue, currentCount)+ r <- runDB $ enableQueue (queueName queue)+ case r of+ Left err -> $reportError "Can't enable queue `{}': {}" (queueName queue, show err)+ Right _ -> return ()+ else $debug "Disabled queue `{}' has {} new jobs, do not enable it" (queueName queue, currentCount)+ liftIO $ threadDelay $ 60 * 1000*1000++-- | Get URL parameter in form ?name=value+getUrlParam :: B.ByteString -> Action (Maybe B.ByteString)+getUrlParam key = do+ rq <- Scotty.request+ let qry = Wai.queryString rq+ return $ join $ lookup key qry++raise404 :: Action TL.Text -> Maybe String -> Action ()+raise404 message mbs = do+ localizedMessage <- message+ Scotty.status status404+ case mbs of+ Nothing -> Scotty.text localizedMessage+ Just name -> do+ let msg' = format (parseFormat' localizedMessage) (Single name)+ Scotty.text msg'++raiseError :: Error -> Action ()+raiseError (QueueNotExists name) = raise404 (__ "Queue not found: `{}'") (Just name)+raiseError JobNotExists = raise404 (__ "Specified job not found") Nothing+raiseError (FileNotExists name) = raise404 (__ "File not found: `{}'") (Just name)+raiseError (MetricNotExists name) = raise404 (__ "Metric not found: `{}'") (Just name)+raiseError QueueNotEmpty = Scotty.status status403+raiseError (InsufficientRights msg) = do+ Scotty.status status403+ Scotty.text $ TL.pack msg+raiseError e = do+ Scotty.status status500+ Scotty.text $ TL.pack $ show e++done :: Action ()+done = Scotty.json ("done" :: String)++inUserContext :: Action a -> Action a+inUserContext action = do+ name <- getAuthUserName+ withLogVariable "user" (fromMaybe "<unauthorized>" name) $ do+ -- req <- Scotty.request+ -- context <- getLogContext+ -- liftIO $ putStrLn $ show context+ -- \$debug "Request: {method} {path}. User-Agent: {useragent}" req+ action++getQueuesA :: Action ()+getQueuesA = inUserContext $ do+ $debug "getting queues list" ()+ user <- getAuthUser+ let name = userName user+ super <- isSuperUser name+ qes <- if super+ then runDBA getAllQueues'+ else runDBA $ getAllowedQueues name ViewQueues+ Scotty.json qes++getQueueA :: Action ()+getQueueA = inUserContext $ do+ qname <- Scotty.param "name"+ checkPermission "view queue" ViewQueues qname+ mbQueue <- runDBA $ getQueue qname+ case mbQueue of+ Nothing -> raise (QueueNotExists qname)+ Just queue -> Scotty.json queue++parseStatus' :: Maybe JobStatus -> Maybe B.ByteString -> Action (Maybe JobStatus)+parseStatus' dflt str = parseStatus dflt (raise (InvalidJobStatus str)) str++getQueueJobsA :: Action ()+getQueueJobsA = inUserContext $ do+ qname <- Scotty.param "name"+ checkPermission "view queue jobs" ViewJobs qname+ st <- getUrlParam "status"+ fltr <- parseStatus' (Just New) st+ jobs <- runDBA $ loadJobs qname fltr+ Scotty.json jobs++getQueueStatsA :: Action ()+getQueueStatsA = inUserContext $ do+ qname <- Scotty.param "name"+ checkPermission "view queue statistics" ManageJobs qname+ stats <- runDBA $ getQueueStats qname+ Scotty.json stats++getStatsA :: Action ()+getStatsA = inUserContext $ do+ checkPermissionToList "view queue statistics" ManageJobs+ stats <- runDBA getStats+ Scotty.json stats++enqueueA :: Action ()+enqueueA = inUserContext $ do+ jinfo <- jsonData+ qname <- Scotty.param "name"+ user <- getAuthUser+ -- special name for default host of queue+ let hostToCheck = fromMaybe defaultHostOfQueue (T.pack <$> jiHostName jinfo)+ checkCanCreateJobs qname (jiType jinfo) hostToCheck+ r <- runDBA $ enqueue (userName user) qname jinfo+ Scotty.json r++removeJobA :: Action ()+removeJobA = inUserContext $ do+ qname <- Scotty.param "name"+ checkPermission "delete jobs from queue" ManageJobs qname+ jseq <- Scotty.param "seq"+ runDBA $ removeJob qname jseq+ done++removeJobByIdA :: Action ()+removeJobByIdA = inUserContext $ do+ checkPermissionToList "delete jobs" ManageJobs+ jid <- Scotty.param "id"+ runDBA $ removeJobById jid+ done++getJobLastResultA :: Action ()+getJobLastResultA = inUserContext $ do+ jid <- Scotty.param "id"+ job <- runDBA $ loadJob' jid+ checkPermission "view job result" ViewJobs (jiQueue job)+ res <- runDBA $ getJobResult jid+ Scotty.json res++getJobResultsA :: Action ()+getJobResultsA = inUserContext $ do+ jid <- Scotty.param "id"+ job <- runDBA $ loadJob' jid+ checkPermission "view job result" ViewJobs (jiQueue job)+ res <- runDBA $ getJobResults jid+ Scotty.json res++removeQueueA :: Action ()+removeQueueA = inUserContext $ do+ qname <- Scotty.param "name"+ checkPermission "delete queue" ManageQueues qname+ forced <- getUrlParam "forced"+ st <- getUrlParam "status"+ fltr <- parseStatus' Nothing st+ case fltr of+ Nothing -> do+ r <- runDBA' $ deleteQueue qname (forced == Just "true")+ case r of+ Left QueueNotEmpty -> do+ Scotty.status status403+ Left e -> Scotty.raise e+ Right _ -> done+ Just status -> do+ runDBA $ removeJobs qname status+ done++getSchedulesA :: Action ()+getSchedulesA = inUserContext $ do+ checkPermissionToList "get list of schedules" ViewSchedules+ ss <- runDBA loadAllSchedules+ Scotty.json ss++addScheduleA :: Action ()+addScheduleA = inUserContext $ do+ checkPermissionToList "create schedule" ManageSchedules+ sd <- jsonData+ name <- runDBA $ addSchedule sd+ Scotty.json name++removeScheduleA :: Action ()+removeScheduleA = inUserContext $ do+ checkPermissionToList "delete schedule" ManageSchedules+ name <- Scotty.param "name"+ forced <- getUrlParam "forced"+ r <- runDBA' $ removeSchedule name (forced == Just "true")+ case r of+ Left ScheduleUsed -> Scotty.status status403+ Left e -> Scotty.raise e+ Right _ -> done++addQueueA :: Action ()+addQueueA = inUserContext $ do+ checkPermissionToList "create queue" ManageQueues+ qd <- jsonData+ name <- runDBA $ addQueue qd+ Scotty.json name++updateQueueA :: Action ()+updateQueueA = inUserContext $ do+ name <- Scotty.param "name"+ checkPermission "modify queue" ManageQueues name+ upd <- jsonData+ runDBA $ updateQueue name upd+ done++getJobA :: Action ()+getJobA = inUserContext $ do+ jid <- Scotty.param "id"+ job <- runDBA $ loadJob' jid+ checkPermission "view job" ViewJobs (jiQueue job)+ Scotty.json job++updateJobA :: Action ()+updateJobA = inUserContext $ do+ jid <- Scotty.param "id"+ job <- runDBA $ loadJob' jid+ checkPermission "modify job" ManageJobs (jiQueue job)+ qry <- jsonData+ runDBA $ case qry of+ UpdateJob upd -> updateJob jid upd+ Move qname -> moveJob jid qname+ Prioritize action -> prioritizeJob jid action+ done++getJobsA :: Action ()+getJobsA = inUserContext $ do+ checkPermissionToList "view jobs from all queues" ViewJobs+ st <- getUrlParam "status"+ fltr <- parseStatus' (Just New) st+ jobs <- runDBA $ loadJobsByStatus fltr+ Scotty.json jobs++deleteJobsA :: Action ()+deleteJobsA = inUserContext $ do+ name <- Scotty.param "name"+ checkPermission "delete jobs" ManageJobs name+ st <- getUrlParam "status"+ fltr <- parseStatus' Nothing st+ case fltr of+ Nothing -> raise $ InvalidJobStatus st+ Just status -> runDBA $ removeJobs name status++getJobTypesA :: Action ()+getJobTypesA = inUserContext $ do+ cfg <- askConfigA+ lts <- askLoggingState+ types <- liftIO $ listJobTypes cfg lts+ Scotty.json types++getAllowedJobTypesA :: Action ()+getAllowedJobTypesA = inUserContext $ do+ user <- getAuthUser+ qname <- Scotty.param "name"+ let name = userName user+ cfg <- askConfigA+ lts <- askLoggingState+ types <- liftIO $ listJobTypes cfg lts+ allowedTypes <- flip filterM types $ \jt -> do+ hasCreatePermission name qname (Just $ jtName jt) Nothing+ Scotty.json allowedTypes++listJobTypes :: GlobalConfig -> LoggingTState -> IO [JobType]+listJobTypes cfg lts = do+ dirs <- getConfigDirs "jobtypes"+ files <- forM dirs $ \dir -> glob (dir </> "*.yaml")+ ts <- forM (concat files) $ \path -> do+ r <- decodeFileEither path+ case r of+ Left err -> do+ reportErrorIO lts $(here) "Can't parse job type description file: {}" (Single $ Shown err)+ return []+ Right jt -> return [jt]+ let types = concat ts :: [JobType]+ return types++getJobTypeA :: Action ()+getJobTypeA = inUserContext $ do+ name <- Scotty.param "name"+ r <- liftIO $ loadTemplate name+ case r of+ Left err -> raise err+ Right jt -> Scotty.json jt++listHosts :: GlobalConfig -> LoggingTState -> IO [Host]+listHosts cfg lts = do+ dirs <- getConfigDirs "hosts"+ files <- forM dirs $ \dir -> glob (dir </> "*.yaml")+ hs <- forM (concat files) $ \path -> do+ r <- decodeFileEither path+ case r of+ Left err -> do+ reportErrorIO lts $(here) "Can't parse host description file: {}" (Single $ Shown err)+ return []+ Right host -> return [host]+ let hosts = concat hs :: [Host]+ return hosts++getHostsA :: Action ()+getHostsA = inUserContext $ do+ cfg <- askConfigA+ lts <- askLoggingState+ hosts <- liftIO $ listHosts cfg lts+ Scotty.json $ map hName hosts++getAllowedHostsA :: Action ()+getAllowedHostsA = inUserContext $ do+ user <- getAuthUser+ qname <- Scotty.param "name"+ let name = userName user+ cfg <- askConfigA+ lts <- askLoggingState+ hosts <- liftIO $ listHosts cfg lts+ let allHostNames = defaultHostOfQueue : map hName hosts+ allowedHosts <- flip filterM allHostNames $ \hostname -> do+ hasCreatePermission name qname Nothing (Just hostname)+ Scotty.json allowedHosts++getAllowedHostsForTypeA :: Action ()+getAllowedHostsForTypeA = inUserContext $ do+ user <- getAuthUser+ qname <- Scotty.param "qname"+ tname <- Scotty.param "tname"+ mbAllowedHosts <- listAllowedHosts (userName user) qname tname+ allowedHosts <- case mbAllowedHosts of+ Just list -> return list -- user is restricted to list of hosts+ Nothing -> do -- user can create jobs on any defined host+ cfg <- askConfigA+ lts <- askLoggingState+ hosts <- liftIO $ listHosts cfg lts+ return $ defaultHostOfQueue : map hName hosts+ Scotty.json allowedHosts++getUsersA :: Action ()+getUsersA = inUserContext $ do+ checkSuperUser+ names <- runDBA getUsers+ Scotty.json names++createUserA :: Action ()+createUserA = inUserContext $ do+ checkSuperUser+ user <- jsonData+ cfg <- askConfigA+ let staticSalt = getAuthStaticSalt $ mcAuth $ dbcManager cfg+ name <- runDBA $ createUserDb (uiName user) (uiPassword user) staticSalt+ Scotty.json name++changePasswordA :: Action ()+changePasswordA = inUserContext $ do+ name <- Scotty.param "name"+ curUser <- getAuthUser+ when (userName curUser /= name) $+ checkSuperUser+ user <- jsonData+ cfg <- askConfigA+ let staticSalt = getAuthStaticSalt $ mcAuth $ dbcManager cfg+ runDBA $ changePasswordDb name (uiPassword user) staticSalt+ done++createPermissionA :: Action ()+createPermissionA = inUserContext $ do+ checkSuperUser+ name <- Scotty.param "name"+ perm <- jsonData+ id <- runDBA $ createPermission name perm+ Scotty.json id++getPermissionsA :: Action ()+getPermissionsA = inUserContext $ do+ checkSuperUser+ name <- Scotty.param "name"+ perms <- runDBA $ getPermissions name+ Scotty.json perms++deletePermissionA :: Action ()+deletePermissionA = inUserContext $ do+ checkSuperUser+ name <- Scotty.param "name"+ id <- Scotty.param "id"+ runDBA $ deletePermission id name+ done++getAuthOptionsA :: Action ()+getAuthOptionsA = inUserContext $ do+ cfg <- askConfigA+ let methods = authMethods $ mcAuth $ dbcManager cfg+ Scotty.json methods++currentMetricsPlainA :: Action ()+currentMetricsPlainA = inUserContext $ do+ mbPrefix <- optional $ Scotty.param "prefix"+ metrics <- lift $ getCurrentMetrics mbPrefix+ now <- liftIO $ getCurrentTime+ Scotty.json $ sampleToJsonPlain now metrics++currentMetricsTreeA :: Action ()+currentMetricsTreeA = inUserContext $ do+ mbPrefix <- optional $ Scotty.param "prefix"+ metrics <- lift $ getCurrentMetrics mbPrefix+ Scotty.json $ EKG.sampleToJson metrics++lastMetricA :: Action ()+lastMetricA = inUserContext $ do+ name <- Scotty.param "name"+ metric <- runDBA $ getLastMetric name+ Scotty.json $ metricRecordToJsonTree metric++queryMetricA :: Action ()+queryMetricA = inUserContext $ do+ prefix <- Scotty.param "prefix"+ (from, to) <- getPeriod+ records <- runDBA $ queryMetric (prefix <> "%") from to+ Scotty.json $ map metricRecordToJsonPlain records++getPeriod :: Action (UTCTime, UTCTime)+getPeriod = do+ mbLast <- getUrlParam "last"+ mbFrom <- getUrlParam "from"+ mbTo <- getUrlParam "to"+ now <- liftIO $ getCurrentTime+ case (mbFrom, mbTo, mbLast) of+ (Just f, Nothing, Nothing) -> liftM2 (,) (parse f) (return now)+ (Just f, Just t, Nothing) -> liftM2 (,) (parse f) (parse t)+ (Nothing, Nothing, Just l) -> getLast l now+ _ -> raise $ UnknownError "Time period is not specified for metrics request"++ where++ parse :: B.ByteString -> Action UTCTime+ parse bstr = do+ let str = bstrToString bstr+ local <- case readSTime True defaultTimeLocale "%FT%T" str of+ [(result, "")] -> return result+ _ -> raise $ UnknownError $ "Can't parse time: " ++ str + tz <- liftIO $ getCurrentTimeZone+ return $ localTimeToUTC tz local++ getLast :: B.ByteString -> UTCTime -> Action (UTCTime, UTCTime)+ getLast bstr now = do+ let str = bstrToString bstr+ seconds <- if all isDigit str+ then return $ read str+ else raise $ UnknownError $ "Can't parse period: " ++ str+ let start = addUTCTime (fromIntegral $ negate seconds) now+ return (start, now)++getQueueJobsCountA :: Action ()+getQueueJobsCountA = do+ (from, to) <- getPeriod+ records <- runDBA $ queryMetric "batchd.queue.%.jobs.%" from to+ result <- forM records parseRecord+ Scotty.json result+ where+ + parseRecord :: MetricRecord -> Action Data.Yaml.Value+ parseRecord r = do+ (queue, status) <- parseName (metricRecordName r)+ return $ object [+ "time" .= metricRecordTime r,+ "queue" .= queue,+ "status" .= status,+ "jobs" .= metricRecordValue r]+ + parseName :: T.Text -> Action (T.Text, JobStatus)+ parseName name = case Parsec.runParser pName () "<metric name>" name of+ Left err -> raise $ UnknownError $ show err+ Right res -> return res++ pName :: Parsec.Parser (T.Text, JobStatus)+ pName = do+ Parsec.string "batchd.queue."+ queue <- Parsec.many1 (Parsec.noneOf ".")+ Parsec.string ".jobs."+ status <- Parsec.many1 Parsec.alphaNum+ return (T.pack queue, read status)+
+ src/Batchd/Daemon/Monitoring.hs view
@@ -0,0 +1,180 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TemplateHaskell #-}++module Batchd.Daemon.Monitoring + (+ WaiMetrics,+ setupMetrics,+ metricsDumper, metricsCleaner,+ getWaiMetricsMiddleware,+ getCurrentMetrics,+ metricRecordToJsonTree,+ metricRecordToJsonPlain,+ sampleToJsonPlain,+ Metrics.counter,+ Metrics.gauge,+ Metrics.timed,+ timedN,+ Metrics.distribution,+ Metrics.label+ ) where++import Control.Monad+import Control.Monad.State+import Control.Concurrent+import qualified Control.Monad.Metrics as Metrics+import Control.Monad.Catch as MC+import Lens.Micro+import Data.Time+import qualified Data.Vector as V+import Data.Aeson as Aeson+import Data.Aeson.Types as Aeson+import qualified Data.Text as T+import Data.Text.Format.Heavy+import qualified Data.HashMap.Lazy as H+import qualified Data.HashMap.Strict as M+import Database.Persist+import qualified System.Metrics as EKG+import qualified System.Metrics.Distribution as EKG+import qualified Network.Wai as Wai+import Network.Wai.Metrics++import Batchd.Core.Daemon.Logging+import Batchd.Common.Types+import Batchd.Common.Data+import Batchd.Daemon.Types++setupMetrics :: Daemon ()+setupMetrics = do+ store <- liftIO EKG.newStore+ cfg <- askConfig+ when (mcGcMetrics $ dbcMetrics cfg) $ do+ liftIO $ EKG.registerGcMetrics store+ mbWaiMetrics <- if mcHttpMetrics $ dbcMetrics cfg+ then Just `fmap` (liftIO $ registerNamedWaiMetrics "batchd" store)+ else return Nothing+ metrics <- liftIO $ Metrics.initializeWith store+ Daemon $ lift $ modify $ \st -> st {ciMetrics = Just metrics, ciWaiMetrics = mbWaiMetrics}++instance Metrics.MonadMetrics Daemon where+ getMetrics = do+ ci <- askConnectionInfo+ case ciMetrics ci of+ Nothing -> fail $ "Metrics are not initialized yet!"+ Just m -> return m++getWaiMetricsMiddleware :: Daemon (Maybe Wai.Middleware)+getWaiMetricsMiddleware = do+ ci <- askConnectionInfo+ case ciWaiMetrics ci of+ Nothing -> return Nothing+ Just wm -> return $ Just $ metrics wm++getCurrentMetrics :: Maybe T.Text -> Daemon EKG.Sample+getCurrentMetrics mbPrefix = do+ metrics <- Metrics.getMetrics+ let store = metrics ^. Metrics.metricsStore+ sample <- liftIO $ EKG.sampleAll store+ let good = case mbPrefix of+ Just prefix -> H.filterWithKey (\name _ -> prefix `T.isPrefixOf` name) sample+ Nothing -> sample+ return good++metricsDumper :: Daemon ()+metricsDumper = do+ cfg <- askConfig+ let mode = dbcDaemonMode cfg+ let timeout = mcDumpTimeout $ dbcMetrics cfg+ forever $ do+ liftIO $ threadDelay $ timeout * 1000 * 1000+ now <- liftIO $ getCurrentTime+ sample <- getCurrentMetrics Nothing+ r <- runDB $ do+ forM_ (H.toList sample) $ \(name, value) -> do+ let ok = case mcStorePrefixOnly $ dbcMetrics cfg of+ Nothing -> True+ Just prefix -> prefix `T.isPrefixOf` name+ when ok $ do+ let record = mkRecord mode now name value+ insert_ record+ case r of+ Left err -> $reportError "Can't insert metric record to DB: {}" (Single $ show err)+ Right _ -> return ()++mkRecord :: DaemonMode -> UTCTime -> T.Text -> EKG.Value -> MetricRecord+mkRecord mode time name value =+ case value of+ EKG.Counter n -> (emptyRecord Counter) {metricRecordValue = Just n}+ EKG.Gauge n -> (emptyRecord Gauge) {metricRecordValue = Just n}+ EKG.Label text -> (emptyRecord Label) {metricRecordText = Just text}+ EKG.Distribution st ->+ (emptyRecord Distribution) {+ metricRecordMean = Just $ EKG.mean st,+ metricRecordVariance = Just $ EKG.variance st,+ metricRecordCount = Just $ EKG.count st,+ metricRecordSum = Just $ EKG.sum st,+ metricRecordMin = Just $ EKG.min st,+ metricRecordMax = Just $ EKG.max st+ }+ where+ emptyRecord kind = MetricRecord name time mode kind Nothing Nothing Nothing Nothing Nothing Nothing Nothing Nothing++metricsCleaner :: Daemon ()+metricsCleaner = do+ cfg <- askConfig+ let days = scMetricRecords $ dbcStorage cfg+ forever $ do+ liftIO $ threadDelay $ 60 * 60 * 1000*1000+ now <- liftIO $ getCurrentTime+ let delta = fromIntegral $ days * 24 * 3600+ let edge = addUTCTime (negate delta) now+ r <- runDB $ do+ deleteWhere [MetricRecordTime <. edge]+ case r of+ Left err -> $reportError "Can't clean metrics history: {}" (Single $ show err)+ Right _ -> return ()++-- metricValueToJson :: MetricRecord -> Aeson.Value+-- metricValueToJson r =+-- case metricRecordKind r of+-- Counter -> object ["type" .= ("c" :: T.Text), "val" .= metricRecordValue r]+-- Gauge -> object ["type" .= ("g" :: T.Text), "val" .= metricRecordValue r]+-- Label -> object ["type" .= ("l" :: T.Text), "val" .= metricRecordText r]+-- Distribution ->+-- object [+-- "type" .= ("d" :: T.Text), +-- "mean" .= metricRecordMean r,+-- "variance" .= metricRecordVariance r,+-- "count" .= metricRecordCount r,+-- "sum" .= metricRecordSum r,+-- "min" .= metricRecordMin r,+-- "max" .= metricRecordMax r+-- ]++metricRecordToJsonPlain :: MetricRecord -> Aeson.Value+metricRecordToJsonPlain r = toJSON r++sampleToJsonPlain :: UTCTime -> EKG.Sample -> Aeson.Value+sampleToJsonPlain time sample = Aeson.Array $ V.fromList $ map convert $ H.toList sample+ where+ convert (name, value) = toJSON $ mkRecord Manager time name value++metricRecordToJsonTree :: MetricRecord -> Aeson.Value+metricRecordToJsonTree r =+ let metricValue = toJSON r++ build :: Aeson.Value -> T.Text -> Aeson.Value -> Aeson.Value+ build m name val = go m (T.splitOn "." name) val++ go :: Aeson.Value -> [T.Text] -> Aeson.Value -> Aeson.Value+ go (Aeson.Object m) [str] val = Aeson.Object $ M.insert str val m+ go (Aeson.Object m) (str:rest) val = case M.lookup str m of+ Nothing -> Aeson.Object $ M.insert str (go Aeson.emptyObject rest val) m+ Just m' -> Aeson.Object $ M.insert str (go m' rest val) m+ go v _ _ = error $ "metricToRecordJson.go: unexpected: " ++ show v++ in build Aeson.emptyObject (metricRecordName r) metricValue++timedN :: (MonadIO m, Metrics.MonadMetrics m, MonadMask m) => [T.Text] -> m a -> m a+timedN = Metrics.timedList Metrics.Seconds+
+ src/Batchd/Daemon/SSH.hs view
@@ -0,0 +1,164 @@+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE OverloadedStrings #-}++module Batchd.Daemon.SSH where++import Control.Monad+import Control.Monad.Trans+import Control.Exception as E+import Control.Concurrent+import qualified Control.Monad.Catch as MC+import Data.Maybe+import Data.Int+import qualified Data.Map as M+import qualified Data.Text as T+import qualified Data.Text.Lazy as TL+import qualified Data.Text.IO as TIO+import qualified Data.ByteString as B+import Data.Conduit+import qualified Data.Conduit.Combinators as C+import qualified Data.Conduit.List as CL+import Data.Text.Format.Heavy+import Network.SSH.Client.LibSSH2+import Network.SSH.Client.LibSSH2.Conduit+import System.Log.Heavy+import System.FilePath+import System.Environment+import System.Exit+import System.IO (hClose, openTempFile, Handle)+import System.Directory (getTemporaryDirectory, removeFile)++import Batchd.Core.Common.Types+import Batchd.Daemon.Types+import Batchd.Core.Daemon.Hosts+import Batchd.Core.Daemon.Logging+import Batchd.Common.Types++getKnownHosts :: IO FilePath+getKnownHosts = do+ home <- getEnv "HOME"+ return $ home </> ".ssh" </> "known_hosts"++getDfltPublicKey :: IO FilePath+getDfltPublicKey = do+ home <- getEnv "HOME"+ return $ home </> ".ssh" </> "id_rsa.pub"++getDfltPrivateKey :: IO FilePath+getDfltPrivateKey = do+ home <- getEnv "HOME"+ return $ home </> ".ssh" </> "id_rsa"++accountForDefaultHostKeys :: Host -> IO Host+accountForDefaultHostKeys host = do+ def_public_key <- liftIO getDfltPublicKey+ def_private_key <- liftIO getDfltPrivateKey+ return $ host {+ hPublicKey = Just $ fromMaybe def_public_key (hPublicKey host)+ , hPrivateKey = Just $ fromMaybe def_private_key (hPrivateKey host)+ }++withSshOnHost :: HostController -> Host -> (Session -> Daemon a) -> Daemon a+withSshOnHost controller host actions = do+ known_hosts <- liftIO getKnownHosts+ def_public_key <- liftIO getDfltPublicKey+ def_private_key <- liftIO getDfltPrivateKey+ let passphrase = hPassphrase host+ public_key = fromMaybe def_public_key $ hPublicKey host+ private_key = fromMaybe def_private_key $ hPrivateKey host+ user = hUserName host+ port = hPort host+ original_hostname = hHostName host++ $(putMessage config_level) "Target host settings: {}" (Single $ Shown host)+ mbActualHostName <- liftIO $ getActualHostName controller (hControllerId host)+ hostname <- case mbActualHostName of+ Nothing -> return original_hostname+ Just actual -> do+ $debug "Actual hostname of host `{}' is {}" (hName host, actual)+ return actual+ $info "CONNECTING TO {} => {}@{}:{}" (original_hostname, user, hostname, port)+ wrapDaemon (withSSH2 known_hosts public_key private_key (T.unpack passphrase) (T.unpack user) (T.unpack hostname) port) $ \session -> do+ $debug "Connected to {}:{}." (hostname, port)+ actions session++ignoringIOErrors :: MC.MonadCatch m => m () -> m ()+ignoringIOErrors ioe = ioe `MC.catch` (\e -> const (return ()) (e :: IOError))++withTempFile :: FilePath -> (FilePath -> Handle -> Daemon a) -> Daemon a+withTempFile name actions = do+ tmpDir <- liftIO $ getTemporaryDirectory+ MC.bracket (liftIO $ openTempFile tmpDir name)+ (\(name, handle) -> ignoringIOErrors $ liftIO $ removeFile name)+ (uncurry actions)++withRemoteScript :: Session -> FilePath -> Int64 -> [T.Text] -> (String -> Daemon a) -> Daemon a+withRemoteScript session scriptsDir jobId commands actions = do+ if length commands == 1+ then actions (T.unpack $ head commands)+ else do+ let scriptText = formatScript commands+ scriptName = TL.unpack $ format "batchd_job_{}.script" (Single jobId)+ withTempFile scriptName $ \tmpPath tmpHandle -> do+ liftIO $ TIO.hPutStr tmpHandle scriptText+ liftIO $ hClose tmpHandle+ uploadFiles [tmpPath] scriptsDir session+ actions (scriptsDir </> takeFileName tmpPath)++execCommandsOnHost :: HostController -> Host -> [T.Text] -> Daemon ExitCode+execCommandsOnHost controller host commands =+ withSshOnHost controller host $ \session -> go ExitSuccess session commands+ where+ go :: ExitCode -> Session -> [T.Text] -> Daemon ExitCode+ go prevEc _ [] = return prevEc+ go prevEc session (command : commands) = do+ $info "Executing at {}: {}" (hName host, command)+ (Just commandHandle, commandOutput) <- liftIO $ execCommand True session (T.unpack command)+ lts <- askLoggingStateM+ liftIO $+ commandOutput =$= C.decodeUtf8 =$= C.linesUnbounded $$ CL.mapM_ $ \line ->+ infoIO lts $(here) "output> {}" (Single line)+ rc <- liftIO $ getReturnCode commandHandle+ $info "Exit code: {}" (Single rc)+ if rc == 0+ then go ExitSuccess session commands+ else return (ExitFailure rc)++retrieveOutput :: JobInfo -> JobType -> CommandsHandle -> Source IO B.ByteString -> ResultsChan -> IO ExitCode+retrieveOutput job jtype commandHandle commandOutput resultChan = do+ commandOutput =$= C.decodeUtf8 =$= C.linesUnbounded $$ CL.mapM_ $ \line ->+ writeChan resultChan (job, StdoutLine line)+ rc <- getReturnCode commandHandle+ let ec = if rc == 0+ then ExitSuccess+ else ExitFailure rc+ writeChan resultChan (job, Exited ec (jtOnFail jtype))+ return ec++getInputFiles :: JobType -> JobInfo -> [FilePath]+getInputFiles jt job =+ [T.unpack value | (name, value) <- M.assocs (jiParams job), getParamType jt name == Just InputFile]++getOutputFiles :: JobType -> JobInfo -> [FilePath]+getOutputFiles jt job =+ [T.unpack value | (name, value) <- M.assocs (jiParams job), getParamType jt name == Just OutputFile]++uploadFiles :: [FilePath] -> FilePath -> Session -> Daemon ()+uploadFiles files input_directory session =+ forM_ files $ \path -> do+ let remotePath = input_directory </> takeFileName path+ $info "Uploading `{}' to `{}'" (path, remotePath)+ size <- liftIO $ scpSendFile session 0o777 path remotePath+ `catch` (\(e :: SomeException) -> throw (UploadException path e))+ $debug "Done ({} bytes)." (Single size)++downloadFiles :: FilePath -> [FilePath] -> Session -> Daemon ()+downloadFiles output_directory files session =+ forM_ files $ \path -> do+ let remotePath = output_directory </> takeFileName path+ $info "Downlooading `{}' to `{}'" (remotePath, path)+ liftIO $ scpReceiveFile session remotePath path+ `catch` (\(e :: SomeException) -> throw (DownloadException path e))+ $debug "Done." ()+
+ src/Batchd/Daemon/Schedule.hs view
@@ -0,0 +1,62 @@+{-# LANGUAGE DeriveDataTypeable, StandaloneDeriving, RecordWildCards, DeriveGeneric #-}++module Batchd.Daemon.Schedule where++import Control.Monad.Reader+import Data.Maybe+import Database.Persist++import Batchd.Core.Common.Types+import Batchd.Daemon.Types+import Batchd.Common.Data+import Batchd.Common.Schedule++loadSchedule :: Key Schedule -> DB ScheduleInfo+loadSchedule scheduleId@(ScheduleKey sname) = do+ mbSchedule <- get scheduleId+ case mbSchedule of+ Nothing -> throwR (ScheduleNotExists sname)+ Just s -> do+ let name = scheduleName s+ ts <- selectList [ScheduleTimeScheduleName ==. sname] []+ let times = map (toPeriod . entityVal) ts+ ws <- selectList [ScheduleWeekDayScheduleName ==. sname] []+ let weekdays = map (scheduleWeekDayWeekDay . entityVal) ws+ return $ ScheduleInfo {+ sName = name,+ sWeekdays = if null weekdays then Nothing else Just weekdays,+ sTime = if null times then Nothing else Just times+ }+ +addSchedule :: ScheduleInfo -> DB String+addSchedule si = do+ ScheduleKey sid <- insert $ Schedule (sName si)+ case sWeekdays si of+ Nothing -> return ()+ Just weekdays ->+ forM_ weekdays $ \wd -> do+ insert_ $ ScheduleWeekDay sid wd+ case sTime si of+ Nothing -> return ()+ Just times -> + forM_ times $ \(Period {..}) -> do+ let time = ScheduleTime sid periodBegin periodEnd+ insert_ time+ return sid++loadAllSchedules :: DB [ScheduleInfo]+loadAllSchedules = do+ sids <- selectKeysList [] []+ forM sids $ \sid -> loadSchedule sid++removeSchedule :: String -> Bool -> DB ()+removeSchedule name forced = do+ qs <- selectFirst [QueueScheduleName ==. name] []+ let key = ScheduleKey name+ if isNothing qs || forced+ then do+ deleteWhere [ScheduleTimeScheduleName ==. name]+ deleteWhere [ScheduleWeekDayScheduleName ==. name]+ deleteCascade key+ else throwR ScheduleUsed+
+ src/Batchd/Daemon/Types.hs view
@@ -0,0 +1,316 @@+{-# LANGUAGE DeriveDataTypeable, ScopedTypeVariables, TemplateHaskell, GeneralizedNewtypeDeriving, DeriveGeneric, StandaloneDeriving, OverloadedStrings, TypeSynonymInstances, FlexibleInstances, MultiParamTypeClasses #-}+-- | This module contains type declarations and utilities that are used only by batchd daemon.+module Batchd.Daemon.Types where++import Control.Exception (catch)+import Control.Monad.Reader+import Control.Monad.State+import qualified Control.Monad.State.Strict as SSt+import Control.Monad.Except+import qualified Control.Monad.Catch as MC+import Control.Monad.Trans.Resource+import qualified Control.Monad.Metrics as Metrics+import Control.Concurrent+import qualified Data.Text as T+import qualified Data.Text.Lazy as TL+import qualified Data.Text.Format.Heavy as F+import qualified Database.Persist.Sql as Sql+import Network.Wai+import Network.Wai.Metrics+import qualified Web.Scotty.Trans as Scotty+import qualified Web.Scotty.Internal.Types as SI+import System.Exit (ExitCode (..))+import System.Log.Heavy+import System.Log.Heavy.Instances.UnliftIO () -- import instances only+import Text.Localize+import Text.Localize.IO++import Batchd.Core.Common.Types+import Batchd.Common.Types++instance Scotty.ScottyError Error where+ stringError e = UnknownError e+ showError e = TL.pack (show e)++-- deriving instance MonadUnliftIO (LoggingT (ResourceT IO))++-- | Standard monad for database actions+type DB a = ReaderT Sql.SqlBackend (LoggingT (ExceptT Error (ResourceT IO))) a++-- | Temporary monad type for usage of DB from IO directly+type DBIO a = ReaderT Sql.SqlBackend (LoggingT (ResourceT IO)) a++-- | Throw an error in DB monad+throwR :: Error -> DB a+throwR ex = lift $ lift $ throwError ex++-- | Convert DB monad into DBIO+dbio :: DB a -> DBIO (Either Error a)+dbio action = do+ backend <- ask+ lts <- lift $ ask+ x <- liftIO $ runResourceT $ runExceptT $ runLoggingT (runReaderT action backend) lts+ case x of+ Left err -> do+ return $ Left err+ Right r -> return $ Right r++-- | Database connection information and configuration+data ConnectionInfo = ConnectionInfo {+ ciGlobalConfig :: GlobalConfig -- ^ Global configuration+ , ciPool :: Maybe Sql.ConnectionPool -- ^ DB connection pool+ , ciTranslations :: Maybe Translations+ , ciMetrics :: Maybe Metrics.Metrics+ , ciWaiMetrics :: Maybe WaiMetrics+ }++data JobResultUpdate =+ StartExecution+ | StdoutLine T.Text+ | StderrLine T.Text+ | Exited ExitCode OnFailAction+ | ExecError T.Text OnFailAction+ deriving (Eq, Show)++type ResultsChan = Chan (JobInfo, JobResultUpdate)++deriving instance MonadThrow m => MC.MonadThrow (LoggingT m)+deriving instance MC.MonadCatch m => MC.MonadCatch (LoggingT m)+deriving instance MC.MonadMask m => MC.MonadMask (LoggingT m)++deriving instance MonadFail m => MonadFail (LoggingT m)++-- | Main monad for daemon actions (both Manager and Dispatcher). This handles logging+-- and DB connection.+newtype Daemon a = Daemon {+ runDaemonT :: LoggingT (StateT ConnectionInfo IO) a+ }+ deriving (Applicative,Functor,Monad,MonadIO, MonadFail, MonadReader LoggingTState,+ MC.MonadThrow, MC.MonadCatch, MC.MonadMask,+ HasLogContext, HasLogger)++instance Localized Daemon where+ getLanguage = liftIO $ languageFromLocale++ getTranslations = do+ ci <- askConnectionInfo+ case ciTranslations ci of+ Just tr -> return tr+ Nothing -> fail $ "Translations are not loaded yet"++ getContext = return Nothing++instance Localized (ReaderT Sql.SqlBackend (LoggingT (ExceptT Error (ResourceT IO)))) where+ getLanguage = liftIO $ languageFromLocale++ getTranslations = liftIO getTranslations++ getContext = return Nothing++-- | Run daemon actions within IO monad+runDaemonIO :: ConnectionInfo -> LoggingTState -> Daemon a -> IO a+runDaemonIO connInfo lts actions =+ evalStateT (runLoggingT (runDaemonT actions) lts) connInfo++-- | REST handler monad type+type Action a = Scotty.ActionT Error Daemon a++-- TODO: this should respect Accept-Language HTTP header+instance Localized (Scotty.ActionT Error Daemon) where+ getLanguage = lift getLanguage+ getTranslations = lift getTranslations+ getContext = lift getContext++instance HasLogContext (Scotty.ActionT Error Daemon) where+ getLogContext = lift getLogContext++ withLogContext frame actions =+ SI.ActionT $ ExceptT $ ReaderT $ \env -> SSt.StateT $ \rs ->+ let actions' = SI.runAM actions+ r = (runReaderT (runExcept actions') env)+ in withLogContext frame $ SSt.runStateT r rs+ where+ runExcept (ExceptT m) = m++instance HasLogger (Scotty.ActionT Error Daemon) where+ getLogger = lift getLogger++ localLogger logger actions = + SI.ActionT $ ExceptT $ ReaderT $ \env -> SSt.StateT $ \rs ->+ localLogger logger $ SSt.runStateT (runReaderT (runExcept $ SI.runAM actions) env) rs+ where+ runExcept (ExceptT m) = m++instance HasLogContext (ReaderT Sql.SqlBackend (LoggingT (ExceptT Error (ResourceT IO)))) where+ getLogContext = lift getLogContext++ withLogContext frame actions =+ ReaderT $ \backend -> withLogContext frame $ runReaderT actions backend++instance HasLogger (ReaderT Sql.SqlBackend (LoggingT (ExceptT Error (ResourceT IO)))) where + getLogger = lift getLogger++ localLogger logger actions = + ReaderT $ \backend -> localLogger logger $ runReaderT actions backend++instance F.VarContainer Request where+ lookupVar "method" rq = Just $ F.Variable $ show $ requestMethod rq+ lookupVar "path" rq = Just $ F.Variable $ rawPathInfo rq+ lookupVar "secure" rq = Just $ F.Variable $ isSecure rq+ lookupVar "referer" rq = Just $ F.Variable $ requestHeaderReferer rq+ lookupVar "useragent" rq = Just $ F.Variable $ requestHeaderUserAgent rq+ lookupVar _ rq = Nothing++instance F.ClosedVarContainer Request where+ allVarNames _ = ["method", "path", "secure", "referer", "useragent"]++-- | Set up translations+setupTranslations :: LocatePolicy -> Daemon ()+setupTranslations p = do+ liftIO $ Text.Localize.IO.setupTranslations p+ translations <- liftIO $ locateTranslations p+ Daemon $ lift $ modify $ \st -> st {ciTranslations = Just translations}++-- | Obtain DB connection+askConnectionInfo :: Daemon ConnectionInfo+askConnectionInfo = Daemon $ lift get++-- | Obtain field of DB connection within Action monad+asksConnectionInfo :: (ConnectionInfo -> a) -> Action a+asksConnectionInfo fn = do+ ci <- lift askConnectionInfo+ return $ fn ci++-- | Obtain logger function.+-- Note: if logger is called directly, it does not perform+-- context-level filter check.+askLoggerM :: Daemon SpecializedLogger+askLoggerM = asks ltsLogger++-- | Obtain logging state within Daemon monad+askLoggingStateM :: Daemon LoggingTState+askLoggingStateM = ask++-- | Obtain logger function.+-- Note: if logger is called directly, it does not perform+-- context-level filter check.+askLogger :: Action SpecializedLogger+askLogger = lift $ askLoggerM++-- | Obtain logging state within Action monad+askLoggingState :: Action LoggingTState+askLoggingState = lift $ askLoggingStateM++-- | Obtain DB connections pool+askPool :: Daemon Sql.ConnectionPool+askPool = do+ mbPool <- Daemon $ lift $ gets ciPool+ case mbPool of+ Nothing -> fail =<< TL.unpack `fmap` (__ "Database connection is not established yet")+ Just pool -> return pool++-- | Obtain global daemon configuration+askConfig :: Daemon GlobalConfig+askConfig = Daemon $ lift $ gets ciGlobalConfig++-- | Obtain DB connections pool within Action monad+askPoolA :: Action Sql.ConnectionPool+askPoolA = do+ mbPool <- asksConnectionInfo ciPool+ case mbPool of+ Nothing -> fail =<< TL.unpack `fmap` (__ "Database connection is not established yet")+ Just pool -> return pool++-- | Obtain global daemon configuration within Action monad+askConfigA :: Action GlobalConfig+askConfigA = asksConnectionInfo ciGlobalConfig++-- | Run DB action within Action monad. Raise HTTP-level error if DB action fails.+runDBA :: DB a -> Action a+runDBA qry = do+ pool <- askPoolA+ cfg <- askConfigA+ lts <- askLoggingState+ r <- liftIO $ do+ (runResourceT $ runLoggingT (Sql.runSqlPool (dbio qry) pool) lts)+ `catch` (\e -> return $ Left $ SqlError e)+ case r of+ Left err -> Scotty.raise err+ Right x -> return x++-- | Run DB action within Action monad.+-- This version does not raise HTTP-level error on DB action fail.+runDBA' :: DB a -> Action (Either Error a)+runDBA' qry = do+ pool <- askPoolA+ cfg <- asksConnectionInfo ciGlobalConfig+ lts <- askLoggingState+ r <- liftIO $ do+ (runResourceT $ runLoggingT (Sql.runSqlPool (dbio qry) pool) lts)+ `catch` (\e -> return $ Left $ SqlError e)+ return r++-- | Run DB action within Daemon monad.+runDB :: DB a -> Daemon (Either Error a)+runDB qry = do+ pool <- askPool+ cfg <- askConfig+ lts <- askLoggingStateM+ liftIO $ do+ (runResourceT $ runLoggingT (Sql.runSqlPool (dbio qry) pool) lts)+ `catch` (\e -> return $ Left $ SqlError e)++-- | Run DB action within IO monad.+runDBIO :: GlobalConfig -> Sql.ConnectionPool -> LoggingTState -> DB a -> IO (Either Error a)+runDBIO cfg pool lts qry = do+ (runResourceT $ runLoggingT (Sql.runSqlPool (dbio qry) pool) lts)+ `catch` (\e -> return $ Left $ SqlError e)++-- | Run Daemon action within IO monad.+-- This is generally used to spawn main daemon threads.+runDaemon :: GlobalConfig -> Maybe Sql.ConnectionPool -> LoggingSettings -> Daemon a -> IO a+runDaemon cfg mbPool backend daemon =+ runner $ withLoggingT backend $ + withLogContext (LogContextFrame [] noChange) $ runDaemonT daemon+ where+ runner r = evalStateT r initState+ initState = ConnectionInfo cfg mbPool Nothing Nothing Nothing+ -- undefinedLogger = error "Internal error: logger is not defined yet"++-- | Wrap Daemon action with some @with@-style function in IO monad.+-- Example: @wrapDaemon (withResource x) $ \resource -> do ...@.+wrapDaemon :: ((c -> IO a) -> IO a) -> (c -> Daemon a) -> Daemon a+wrapDaemon wrapper daemon = do+ lts <- askLoggingStateM+ connInfo <- askConnectionInfo+ result <- liftIO $ wrapper $ \c -> runDaemonIO connInfo lts (daemon c)+ return result++wrapDaemon_ :: ((c -> IO a) -> IO ()) -> (c -> Daemon a) -> Daemon ()+wrapDaemon_ wrapper daemon = do+ lts <- askLoggingStateM+ connInfo <- askConnectionInfo+ liftIO $ wrapper $ \c -> runDaemonIO connInfo lts (daemon c)++-- | forkIO for Daemon monad.+forkDaemon :: String -> Daemon a -> Daemon ()+forkDaemon name daemon = do+ lts <- askLoggingStateM+ connInfo <- askConnectionInfo+ withLogVariable "thread" name $+ liftIO $ forkIO $ do+ runDaemonIO connInfo lts daemon+ return ()+ return ()++-- | Special constant for default host of the queue+defaultHostOfQueue :: T.Text+defaultHostOfQueue = "__default__"++formatScript :: [T.Text] -> T.Text+formatScript lines =+ let lines' = if "#!" `T.isPrefixOf` head lines+ then lines+ else "#!/bin/sh" : lines+ in T.unlines lines'+
src/batch.hs view
@@ -3,30 +3,31 @@ {-# LANGUAGE DeriveGeneric #-} import Control.Exception-import Data.Maybe+import qualified Data.Text.Lazy.IO as TLIO+import Data.Text.Format.Heavy -import Client.Types-import Client.Actions-import Client.CmdLine-import Client.Config-import Client.Monad-import Client.Http-import Client.Shell+import Batchd.Core.Common.Localize+import Batchd.Client.Types+import Batchd.Client.CmdLine+import Batchd.Client.Config+import Batchd.Client.Monad+import Batchd.Client.Shell main :: IO () main = realMain `catch` errorHandler errorHandler :: ClientException -> IO ()-errorHandler (ClientException e) = putStrLn $ "Error: " ++ e+errorHandler (ClientException e) = TLIO.putStrLn =<< (__f "Error: {}" (Single e)) realMain :: IO () realMain = do+ setupTranslations translationPolicy opts <- getCmdArgs -- print opts - manager <- makeClientManager opts+ -- manager <- makeClientManager opts cfg <- loadClientConfig- let state = ClientState opts cfg Nothing Nothing manager+ let state = ClientState opts cfg Nothing Nothing Nothing Nothing runClient state $ commandHandler
src/batchd-admin.hs view
@@ -2,40 +2,47 @@ {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE DeriveDataTypeable #-} -import Data.Semigroup ((<>))+import Control.Monad.Trans+import Control.Monad.Reader import Options.Applicative--import Common.Types-import Common.Config-import Daemon.Database-import Daemon.Auth--data Admin =- CreateSuperuser {userName :: String}- deriving (Show)--createSuperuser :: Parser Admin-createSuperuser = CreateSuperuser- <$> strArgument (metavar "NAME" <> help "name of user to create" <> value "root" <> showDefault)--parser :: Parser Admin-parser =- hsubparser- (command "create-superuser" (info createSuperuser (progDesc "create super user")))+import qualified Database.Persist.Sql as Sql+import System.Log.Heavy+import Text.Localize.IO (setupTranslations) -parserInfo :: ParserInfo Admin-parserInfo = info (parser <**> helper)- (fullDesc- <> header "batchd-admin - the batchd toolset administrative utility")+import Batchd.Core.Common.Localize (translationPolicy)+import Batchd.Common.Config+import Batchd.Common.Data (migrateAll)+import Batchd.Daemon.Logging (getLoggingSettings)+import Batchd.Daemon.Database+import Batchd.Daemon.Auth+import Batchd.Daemon.CmdLine main :: IO () main = do- cmd <- execParser parserInfo- cfgR <- loadGlobalConfig+ cmdline <- execParser adminParserInfo+ cfgR <- loadGlobalConfig (globalConfigPath $ adminCommon cmdline)+ setupTranslations translationPolicy+ let cmd = adminCommand cmdline case cfgR of Left err -> fail $ show err Right cfg -> do- password <- getPassword2- createSuperUser cfg (userName cmd) password- return ()+ let logSettings = getLoggingSettings cfg+ case cmd of+ CreateSuperuser {} -> do+ password <- getPassword2+ withLoggingT logSettings $ do+ logger <- ask+ liftIO $ createSuperUser cfg logger (username cmd) password+ return ()+ Passwd name -> do+ password <- getPassword2+ withLoggingT logSettings $ do+ logger <- ask+ liftIO $ changePassword cfg logger name password+ return ()+ Migrate -> do+ withLoggingT logSettings $ do+ logger <- ask+ pool <- liftIO $ getPool cfg logger+ Sql.runSqlPool (Sql.runMigration migrateAll) pool
src/batchd.hs view
@@ -1,47 +1,52 @@ {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE TemplateHaskell #-} -import Control.Concurrent-import Data.Semigroup ((<>)) import Options.Applicative--import Common.Types-import Common.Config-import Daemon.Database-import Daemon.Manager as Manager-import Daemon.Dispatcher as Dispatcher--parser :: Parser DaemonMode-parser =- hsubparser- ( command "both" (info (pure Both) (progDesc "run both manager and dispatcher"))- <> command "manager" (info (pure Manager) (progDesc "run manager"))- <> command "dispatcher" (info (pure Dispatcher) (progDesc "run dispatcher"))- )- <|> pure Both+import Data.Text.Format.Heavy hiding (optional)+import System.Log.Heavy -parserInfo :: ParserInfo DaemonMode-parserInfo = info (parser <**> helper)- (fullDesc- <> header "batchd - the batchd toolset daemon server-side program"- <> progDesc "process client requests and / or execute batch jobs" )+import Batchd.Core.Common.Types+import Batchd.Core.Common.Localize+import qualified Batchd.Core.Daemon.Logging as Log+import Batchd.Daemon.Types (runDaemon, forkDaemon, setupTranslations)+import qualified Batchd.Daemon.Logging as Log+import Batchd.Common.Config+import Batchd.Common.Types+import Batchd.Daemon.Database+import Batchd.Daemon.Monitoring+import Batchd.Daemon.CmdLine+import Batchd.Daemon.Manager as Manager+import Batchd.Daemon.Dispatcher as Dispatcher main :: IO () main = do- cmd <- execParser parserInfo- cfgR <- loadGlobalConfig+ cmdline <- execParser daemonParserInfo+ let cmd = daemonMode cmdline+ cfgR <- loadGlobalConfig (globalConfigPath $ daemonCommon cmdline) case cfgR of Left err -> fail $ show err Right cfg -> do- let mode = if cmd == Both- then dbcDaemonMode cfg- else cmd- pool <- getPool cfg- case mode of- Manager -> Manager.runManager cfg pool- Dispatcher -> Dispatcher.runDispatcher cfg pool- Both -> do- forkIO $ Manager.runManager cfg pool- Dispatcher.runDispatcher cfg pool+ let cfg' = if cmd == Both+ then cfg+ else cfg {dbcDaemonMode = cmd}+ let mode = dbcDaemonMode cfg'+ let logSettings = Log.getLoggingSettings cfg'+ runDaemon cfg' Nothing logSettings $ do+ Batchd.Daemon.Types.setupTranslations translationPolicy+ tr <- getTranslations+ $(Log.putMessage config_level) "Loaded translations: {}" (Single $ show tr)+ case globalConfigPath $ daemonCommon cmdline of+ Nothing -> return ()+ Just path -> $(Log.putMessage config_level) "Using global configuration file: {}" (Single path)+ $(Log.putMessage config_level) "Loaded global configuration file: {}" (Single $ show cfg')+ connectPool+ setupMetrics+ case mode of+ Manager -> Manager.runManager+ Dispatcher -> Dispatcher.runDispatcher+ Both -> do+ forkDaemon "manager" $ Manager.runManager+ Dispatcher.runDispatcher