diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,30 @@ +Copyright Li Meng Jun (c) 2017 + +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are met: + + * Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + + * Redistributions in binary form must reproduce the above + copyright notice, this list of conditions and the following + disclaimer in the documentation and/or other materials provided + with the distribution. + + * Neither the name of Li Meng Jun nor the names of other + contributors may be used to endorse or promote products derived + from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,3 @@ +# periodic-client-exe + +Periodic task system haskell client executables. diff --git a/app/periodic-http-bridge.hs b/app/periodic-http-bridge.hs new file mode 100644 --- /dev/null +++ b/app/periodic-http-bridge.hs @@ -0,0 +1,202 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} + +module Main + ( main + ) where + +import Control.Monad (when) +import Control.Monad.IO.Class (liftIO) +import qualified Data.ByteString.Char8 as B (pack) +import qualified Data.ByteString.Lazy as LB (empty, fromStrict, + toStrict) +import Data.Default.Class (def) +import Data.List (isPrefixOf) +import Data.Maybe (fromMaybe) +import Data.Streaming.Network.Internal (HostPreference (Host)) +import Data.String (fromString) +import Metro.Class (Transport) +import Metro.Socket (getHost, getService) +import Metro.TP.Socket (socket) +import Metro.TP.TLS (makeClientParams', tlsConfig) +import Metro.TP.WebSockets (clientConfig) +import Metro.TP.XOR (xorConfig) +import Network.HTTP.Types (status204, status500) +import Network.Wai.Handler.Warp (setHost, setPort) +import Periodic.Trans.ClientPool +import Periodic.Types.Job +import System.Environment (getArgs, lookupEnv) +import System.Exit (exitSuccess) +import Web.Scotty (ActionM, ScottyM, body, get, + param, post, raw, rescue, + scottyOpts, settings) +import qualified Web.Scotty as WS (status) + + +data Options = Options + { host :: String + , xorFile :: FilePath + , useTls :: Bool + , useWs :: Bool + , hostName :: String + , certKey :: FilePath + , cert :: FilePath + , caStore :: FilePath + , httpHost :: String + , httpPort :: Int + , poolSize :: Int + , showHelp :: Bool + } + +options :: Maybe String -> Maybe String -> Options +options h f = Options + { host = fromMaybe "unix:///tmp/periodic.sock" h + , xorFile = fromMaybe "" f + , useTls = False + , useWs = False + , hostName = "localhost" + , certKey = "client-key.pem" + , cert = "client.pem" + , caStore = "ca.pem" + , httpHost = "127.0.0.1" + , httpPort = 8080 + , poolSize = 10 + , showHelp = False + } + +parseOptions :: [String] -> Options -> Options +parseOptions [] opt = opt +parseOptions ("-H":x:xs) opt = parseOptions xs opt { host = x } +parseOptions ("--host":x:xs) opt = parseOptions xs opt { host = x } +parseOptions ("--xor":x:xs) opt = parseOptions xs opt { xorFile = x } +parseOptions ("--tls":xs) opt = parseOptions xs opt { useTls = True } +parseOptions ("--ws":xs) opt = parseOptions xs opt { useWs = True } +parseOptions ("--hostname":x:xs) opt = parseOptions xs opt { hostName = x } +parseOptions ("--cert-key":x:xs) opt = parseOptions xs opt { certKey = x } +parseOptions ("--cert":x:xs) opt = parseOptions xs opt { cert = x } +parseOptions ("--ca":x:xs) opt = parseOptions xs opt { caStore = x } +parseOptions ("--http-host":x:xs) opt = parseOptions xs opt { httpHost = x } +parseOptions ("--http-port":x:xs) opt = parseOptions xs opt { httpPort = read x } +parseOptions ("--pool-size":x:xs) opt = parseOptions xs opt { poolSize = read x } +parseOptions ("--help":xs) opt = parseOptions xs opt { showHelp = True } +parseOptions ("-h":xs) opt = parseOptions xs opt { showHelp = True } +parseOptions (_:xs) opt = parseOptions xs opt + +printHelp :: IO () +printHelp = do + putStrLn "periodic-http-bridge - Periodic task system client http bridge" + putStrLn "" + putStrLn "Usage: periodic [--host|-H HOST] [--http-host HOST] [--http-port PORT] [--pool-size SIZE] [--xor FILE|--ws|--tls [--hostname HOSTNAME] [--cert-key FILE] [--cert FILE] [--ca FILE]]" + putStrLn "" + putStrLn "Available options:" + putStrLn " -H --host Socket path [$PERIODIC_PORT]" + putStrLn " eg: tcp://:5000 (optional: unix:///tmp/periodic.sock) " + putStrLn " --xor XOR Transport encode file [$XOR_FILE]" + putStrLn " --tls Use tls transport" + putStrLn " --ws Use websockets transport" + putStrLn " --hostname Host name" + putStrLn " --cert-key Private key associated" + putStrLn " --cert Public certificate (X.509 format)" + putStrLn " --ca trusted certificates" + putStrLn " --http-host HTTP host (optional: 127.0.0.1)" + putStrLn " --http-port HTTP port (optional: 8080)" + putStrLn " --pool-size Connection pool size" + putStrLn " -h --help Display help message" + putStrLn "" + putStrLn "Version: v1.1.7.1" + putStrLn "" + exitSuccess + +main :: IO () +main = do + h <- lookupEnv "PERIODIC_PORT" + f <- lookupEnv "XOR_FILE" + + opts@Options{..} <- flip parseOptions (options h f) <$> getArgs + + when showHelp printHelp + + when (not ("tcp" `isPrefixOf` host) && not ("unix" `isPrefixOf` host)) $ do + putStrLn $ "Invalid host " ++ host + printHelp + + run opts def + { settings = setPort httpPort + $ setHost (Host httpHost) (settings def)} + +run Options {useTls = True, ..} sopts = do + prms <- makeClientParams' cert [] certKey caStore (hostName, B.pack $ fromMaybe "" $ getService host) + clientEnv <- openPool (tlsConfig prms (socket host)) poolSize + scottyOpts sopts $ application clientEnv + +run Options {useWs = True, ..} sopts = do + clientEnv <- openPool (clientConfig (socket host) (fromMaybe "0.0.0.0" $ getHost host) (fromMaybe "" $ getService host)) poolSize + scottyOpts sopts $ application clientEnv + +run Options {xorFile = "", ..} sopts = do + clientEnv <- openPool (socket host) poolSize + scottyOpts sopts $ application clientEnv + +run Options {..} sopts = do + clientEnv <- openPool (xorConfig xorFile $ socket host) poolSize + scottyOpts sopts $ application clientEnv + +application :: Transport tp => ClientPoolEnv tp -> ScottyM () +application clientEnv = do + get "/periodic/status/" $ statusHandler clientEnv + post "/periodic/submit/:func_name/:job_name/" $ submitJobHandler clientEnv + post "/periodic/run/:func_name/:job_name/" $ runJobHandler clientEnv + post "/periodic/remove/:func_name/:job_name/" $ removeJobHandler clientEnv + post "/periodic/drop/:func_name/" $ dropFuncHandler clientEnv + +paramJob :: ActionM (FuncName, JobName) +paramJob = do + fn <- fromString <$> param "func_name" + jn <- fromString <$> param "job_name" + return (fn, jn) + +paramJob_ :: ActionM Job +paramJob_ = do + (fn, jn) <- paramJob + schedAt <- param "sched_at" `rescue` const (return 0) + timeout <- param "timeout" `rescue` const (return 0) + wb <- Workload . LB.toStrict <$> body + pure $ setSchedAt schedAt $ setTimeout timeout $ setWorkload wb $ initJob fn jn + +sampleRet :: Bool -> ActionM () +sampleRet r = do + WS.status $ if r then status204 else status500 + raw LB.empty + +submitJobHandler :: Transport tp => ClientPoolEnv tp -> ActionM () +submitJobHandler clientEnv = do + job <- paramJob_ + r <- liftIO $ runClientPoolT clientEnv $ submitJob_ job + sampleRet r + +removeJobHandler :: Transport tp => ClientPoolEnv tp -> ActionM () +removeJobHandler clientEnv = do + (fn, jn) <- paramJob + r <- liftIO $ runClientPoolT clientEnv $ removeJob fn jn + sampleRet r + +runJobHandler :: Transport tp => ClientPoolEnv tp -> ActionM () +runJobHandler clientEnv = do + job <- paramJob_ + r <- liftIO $ runClientPoolT clientEnv $ runJob_ job + case r of + Nothing -> do + WS.status status500 + raw "error" + Just bs -> raw $ LB.fromStrict bs + +dropFuncHandler :: Transport tp => ClientPoolEnv tp -> ActionM () +dropFuncHandler clientEnv = do + fn <- fromString <$> param "func_name" + r <- liftIO $ runClientPoolT clientEnv $ dropFunc fn + sampleRet r + +statusHandler :: Transport tp => ClientPoolEnv tp -> ActionM () +statusHandler clientEnv = do + r <- liftIO $ runClientPoolT clientEnv status + raw $ LB.fromStrict r diff --git a/app/periodic-run.hs b/app/periodic-run.hs new file mode 100644 --- /dev/null +++ b/app/periodic-run.hs @@ -0,0 +1,224 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} + +module Main + ( main + ) where + +import Control.Concurrent (forkIO, killThread) +import Control.DeepSeq (rnf) +import Control.Monad (unless, void, when) +import Control.Monad.IO.Class (liftIO) +import qualified Data.ByteString.Char8 as B (ByteString, pack) +import qualified Data.ByteString.Lazy as LB (null, toStrict) +import qualified Data.ByteString.Lazy.Char8 as LB (hGetContents, hPut) +import Data.List (isPrefixOf) +import Data.Maybe (fromMaybe) +import qualified Data.Text as T (unpack) +import Data.Text.Encoding (decodeUtf8With) +import Data.Text.Encoding.Error (ignore) +import Metro.Class (Transport) +import Metro.Socket (getHost, getService) +import Metro.TP.Socket (socket) +import Metro.TP.TLS (makeClientParams', tlsConfig) +import Metro.TP.WebSockets (clientConfig) +import Metro.TP.XOR (xorConfig) +import Periodic.Trans.Job (JobT, name, withLock_, workDone, + workDone_, workFail, workload) +import Periodic.Trans.Worker (WorkerT, addFunc, broadcast, + startWorkerT, work) +import Periodic.Types (FuncName (..), LockName (..)) +import System.Environment (getArgs, lookupEnv) +import System.Exit (ExitCode (..), exitSuccess) +import System.IO (hClose) +import System.Process (CreateProcess (std_in, std_out), + StdStream (CreatePipe, Inherit), + proc, waitForProcess, + withCreateProcess) +import UnliftIO (MVar, SomeException, evaluate, + mask, newEmptyMVar, onException, + putMVar, takeMVar, throwIO, try, + tryIO) + + +data Options = Options + { host :: String + , xorFile :: FilePath + , useTls :: Bool + , useWs :: Bool + , hostName :: String + , certKey :: FilePath + , cert :: FilePath + , caStore :: FilePath + , thread :: Int + , lockCount :: Int + , lockName :: Maybe LockName + , notify :: Bool + , useData :: Bool + , useName :: Bool + , showHelp :: Bool + } + +options :: Maybe Int -> Maybe String -> Maybe String -> Options +options t h f = Options + { host = fromMaybe "unix:///tmp/periodic.sock" h + , xorFile = fromMaybe "" f + , useTls = False + , useWs = False + , hostName = "localhost" + , certKey = "client-key.pem" + , cert = "client.pem" + , caStore = "ca.pem" + , thread = fromMaybe 1 t + , lockCount = 1 + , lockName = Nothing + , notify = False + , useData = False + , useName = True + , showHelp = False + } + +parseOptions :: [String] -> Options -> (Options, FuncName, String, [String]) +parseOptions ("-H":x:xs) opt = parseOptions xs opt { host = x } +parseOptions ("--host":x:xs) opt = parseOptions xs opt { host = x } +parseOptions ("--xor":x:xs) opt = parseOptions xs opt { xorFile = x } +parseOptions ("--tls":xs) opt = parseOptions xs opt { useTls = True } +parseOptions ("--ws":xs) opt = parseOptions xs opt { useWs = True } +parseOptions ("--hostname":x:xs) opt = parseOptions xs opt { hostName = x } +parseOptions ("--cert-key":x:xs) opt = parseOptions xs opt { certKey = x } +parseOptions ("--cert":x:xs) opt = parseOptions xs opt { cert = x } +parseOptions ("--ca":x:xs) opt = parseOptions xs opt { caStore = x } +parseOptions ("--thread":x:xs) opt = parseOptions xs opt { thread = read x } +parseOptions ("--lock-count":x:xs) opt = parseOptions xs opt { lockCount = read x } +parseOptions ("--lock-name":x:xs) opt = parseOptions xs opt { lockName = Just (LockName $ B.pack x) } +parseOptions ("--help":xs) opt = parseOptions xs opt { showHelp = True } +parseOptions ("--broadcast":xs) opt = parseOptions xs opt { notify = True } +parseOptions ("--data":xs) opt = parseOptions xs opt { useData = True } +parseOptions ("--no-name":xs) opt = parseOptions xs opt { useName = False } +parseOptions ("-h":xs) opt = parseOptions xs opt { showHelp = True } +parseOptions [] opt = (opt { showHelp = True }, "", "", []) +parseOptions [_] opt = (opt { showHelp = True }, "", "", []) +parseOptions (x:y:xs) opt = (opt, FuncName $ B.pack x, y, xs) + +printHelp :: IO () +printHelp = do + putStrLn "periodic-run - Periodic task system worker" + putStrLn "" + putStrLn "Usage: periodic-run [--host|-H HOST] [--xor FILE|--ws|--tls [--hostname HOSTNAME] [--cert-key FILE] [--cert FILE] [--ca FILE] [--thread THREAD] [--lock-name NAME] [--lock-count COUNT] [--broadcast] [--data] [--no-name]] funcname command [options]" + putStrLn "" + putStrLn "Available options:" + putStrLn " -H --host Socket path [$PERIODIC_PORT]" + putStrLn " Eg: tcp://:5000 (optional: unix:///tmp/periodic.sock) " + putStrLn " --xor XOR Transport encode file [$XOR_FILE]" + putStrLn " --tls Use tls transport" + putStrLn " --ws Use websockets transport" + putStrLn " --hostname Host name" + putStrLn " --cert-key Private key associated" + putStrLn " --cert Public certificate (X.509 format)" + putStrLn " --ca Trusted certificates" + putStrLn " --thread Worker thread [$THREAD]" + putStrLn " --lock-count Max lock count (optional: 1)" + putStrLn " --lock-name The lock name (optional: no lock)" + putStrLn " --broadcast Is broadcast worker" + putStrLn " --data Send work data to client" + putStrLn " --no-name Ignore the job name" + putStrLn " -h --help Display help message" + putStrLn "" + putStrLn "Version: v1.1.7.1" + putStrLn "" + exitSuccess + +main :: IO () +main = do + h <- lookupEnv "PERIODIC_PORT" + f <- lookupEnv "XOR_FILE" + t <- fmap read <$> lookupEnv "THREAD" + + (opts@Options {..}, func, cmd, argv) <- flip parseOptions (options t h f) <$> getArgs + + when showHelp printHelp + + when (not ("tcp" `isPrefixOf` host) && not ("unix" `isPrefixOf` host)) $ do + putStrLn $ "Invalid host " ++ host + printHelp + + run opts func cmd argv + +doWork :: Transport tp => Options -> FuncName -> String -> [String] -> WorkerT tp IO () +doWork opts@Options{..} func cmd argv = do + let w = processWorker opts cmd argv + if notify then broadcast func w + else + case lockName of + Nothing -> addFunc func w + Just n -> addFunc func $ withLock_ n lockCount w + liftIO $ putStrLn "Worker started." + work thread + +run :: Options -> FuncName -> String -> [String] -> IO () +run opts@Options {useTls = True, ..} func cmd argv = do + prms <- makeClientParams' cert [] certKey caStore (hostName, B.pack $ fromMaybe "" $ getService host) + startWorkerT (tlsConfig prms (socket host)) $ doWork opts func cmd argv + +run opts@Options {useWs = True, ..} func cmd argv = + startWorkerT (clientConfig (socket host) (fromMaybe "0.0.0.0" $ getHost host) (fromMaybe "" $ getService host)) $ doWork opts func cmd argv + +run opts@Options {xorFile = "", ..} func cmd argv = + startWorkerT (socket host) $ doWork opts func cmd argv + +run opts@Options {..} func cmd argv = + startWorkerT (xorConfig xorFile $ socket host) $ doWork opts func cmd argv + +processWorker :: Transport tp => Options -> String -> [String] -> JobT tp IO () +processWorker Options{..} cmd argv = do + n <- name + rb <- workload + let argv' = if useName then argv ++ [n] else argv + cp = (proc cmd argv') {std_in = CreatePipe, std_out= if useData then CreatePipe else Inherit} + + (code, out) <- liftIO $ withCreateProcess cp $ \mb_inh mb_outh _ ph -> + case (mb_inh, mb_outh) of + (Nothing, _) -> error "processWorker: Failed to get a stdin handle." + (Just inh, Nothing) -> do + unless (LB.null rb) $ void $ tryIO $ LB.hPut inh rb + void $ tryIO $ hClose inh + code <- waitForProcess ph + return (code, Nothing) + + (Just inh, Just outh) -> do + output <- LB.hGetContents outh + withForkWait (evaluate $ rnf output) $ \waitOut -> do + unless (LB.null rb) $ void $ tryIO $ LB.hPut inh rb + void $ tryIO $ hClose inh + + waitOut + hClose outh + + code <- waitForProcess ph + return (code, Just output) + + + case code of + ExitFailure _ -> workFail + ExitSuccess -> + case out of + Nothing -> workDone + Just wl -> workDone_ $ LB.toStrict wl + +unpackBS :: B.ByteString -> String +unpackBS = T.unpack . decodeUtf8With ignore + +-- | Fork a thread while doing something else, but kill it if there's an +-- exception. +-- +-- This is important in the cases above because we want to kill the thread +-- that is holding the Handle lock, because when we clean up the process we +-- try to close that handle, which could otherwise deadlock. +-- +withForkWait :: IO () -> (IO () -> IO a) -> IO a +withForkWait async body = do + waitVar <- newEmptyMVar :: IO (MVar (Either SomeException ())) + mask $ \restore -> do + tid <- forkIO $ try (restore async) >>= putMVar waitVar + let wait = takeMVar waitVar >>= either throwIO return + restore (body wait) `onException` killThread tid diff --git a/app/periodic.hs b/app/periodic.hs new file mode 100644 --- /dev/null +++ b/app/periodic.hs @@ -0,0 +1,375 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} + +module Main + ( main + ) where + +import Control.Monad (void, when) +import Control.Monad.IO.Class (liftIO) +import Data.Binary (decodeFile, encodeFile) +import Data.ByteString (ByteString) +import qualified Data.ByteString.Char8 as B (lines, pack, putStr, readFile, + split, unpack) +import Data.Int (Int64) +import Data.List (isPrefixOf, transpose) +import Data.Maybe (fromMaybe) +import Metro.Class (Transport) +import Metro.Socket (getHost, getService) +import Metro.TP.Socket (socket) +import Metro.TP.TLS (makeClientParams', tlsConfig) +import Metro.TP.WebSockets (clientConfig) +import Metro.TP.XOR (xorConfig) +import Periodic.Trans.Client +import Periodic.Types (Workload (..)) +import System.Environment (getArgs, lookupEnv) +import System.Exit (exitSuccess) +import Text.Read (readMaybe) + +import Data.String (fromString) +import Data.UnixTime +import System.IO.Unsafe (unsafePerformIO) +import qualified Text.PrettyPrint.Boxes as T + + +data Command = Status + | Submit + | Run + | Remove + | Drop + | Config + | Shutdown + | Dump + | Load + | Help + deriving (Eq) + +parseCommand :: String -> Command +parseCommand "status" = Status +parseCommand "submit" = Submit +parseCommand "run" = Run +parseCommand "remove" = Remove +parseCommand "drop" = Drop +parseCommand "config" = Config +parseCommand "shutdown" = Shutdown +parseCommand "dump" = Dump +parseCommand "load" = Load +parseCommand _ = Help + +data Options = Options + { host :: String + , xorFile :: FilePath + , useTls :: Bool + , useWs :: Bool + , hostName :: String + , certKey :: FilePath + , cert :: FilePath + , caStore :: FilePath + } + +options :: Maybe String -> Maybe String -> Options +options h f = Options { host = fromMaybe "unix:///tmp/periodic.sock" h + , xorFile = fromMaybe "" f + , useTls = False + , useWs = False + , hostName = "localhost" + , certKey = "client-key.pem" + , cert = "client.pem" + , caStore = "ca.pem" + } + +parseOptions :: [String] -> Options -> (Command, Options, [String]) +parseOptions [] opt = (Help, opt, []) +parseOptions ("-H":x:xs) opt = parseOptions xs opt { host = x } +parseOptions ("--host":x:xs) opt = parseOptions xs opt { host = x } +parseOptions ("--xor":x:xs) opt = parseOptions xs opt { xorFile = x } +parseOptions ("--tls":xs) opt = parseOptions xs opt { useTls = True } +parseOptions ("--ws":xs) opt = parseOptions xs opt { useWs = True } +parseOptions ("--hostname":x:xs) opt = parseOptions xs opt { hostName = x } +parseOptions ("--cert-key":x:xs) opt = parseOptions xs opt { certKey = x } +parseOptions ("--cert":x:xs) opt = parseOptions xs opt { cert = x } +parseOptions ("--ca":x:xs) opt = parseOptions xs opt { caStore = x } +parseOptions (x:xs) opt = (parseCommand x, opt, xs) + +printHelp :: IO () +printHelp = do + putStrLn "periodic - Periodic task system client" + putStrLn "" + putStrLn "Usage: periodic [--host|-H HOST] [--xor FILE|--ws|--tls [--hostname HOSTNAME] [--cert-key FILE] [--cert FILE] [--ca FILE]] command [options]" + putStrLn "" + putStrLn "Commands:" + putStrLn " status Show status" + putStrLn " submit Submit job" + putStrLn " run Run job and output result" + putStrLn " remove Remove job" + putStrLn " drop Drop function" + putStrLn " config Set or Get config" + putStrLn " shutdown Shutdown periodicd" + putStrLn " dump Dump jobs" + putStrLn " load Load jobs" + putStrLn " help Shows a list of commands or help for one command" + putStrLn "" + putStrLn "Available options:" + putStrLn " -H --host Socket path [$PERIODIC_PORT]" + putStrLn " eg: tcp://:5000 (optional: unix:///tmp/periodic.sock) " + putStrLn " --xor XOR Transport encode file [$XOR_FILE]" + putStrLn " --tls Use tls transport" + putStrLn " --ws Use websockets transport" + putStrLn " --hostname Host name" + putStrLn " --cert-key Private key associated" + putStrLn " --cert Public certificate (X.509 format)" + putStrLn " --ca trusted certificates" + putStrLn "" + putStrLn "Version: v1.1.7.1" + putStrLn "" + exitSuccess + +printWorkloadHelp :: IO () +printWorkloadHelp = do + putStrLn "" + putStrLn "Available options:" + putStrLn " -w --workload WORKLOAD or @FILE" + +printSubmitHelp :: IO () +printSubmitHelp = do + putStrLn "periodic submit - Submit job" + putStrLn "" + putStrLn "Usage: periodic submit funcname jobname [-w|--workload WORKLOAD|@FILE] [--later 0]" + printWorkloadHelp + putStrLn " --later Sched job later" + putStrLn "" + exitSuccess + +printRunHelp :: IO () +printRunHelp = do + putStrLn "periodic run - Run job and output result" + putStrLn "" + putStrLn "Usage: periodic run funcname jobname [-w|--workload WORKLOAD|@FILE]" + printWorkloadHelp + putStrLn "" + exitSuccess + +printRemoveHelp :: IO () +printRemoveHelp = do + putStrLn "periodic remove - Remove job" + putStrLn "" + putStrLn "Usage: periodic remove funcname jobname [jobname...]" + putStrLn "" + exitSuccess + +printDropHelp :: IO () +printDropHelp = do + putStrLn "periodic drop - Drop function" + putStrLn "" + putStrLn "Usage: periodic drop funcname [funcname...]" + putStrLn "" + exitSuccess + +printConfigHelp :: IO () +printConfigHelp = do + putStrLn "periodic config - Set or get config" + putStrLn "" + putStrLn "Usage: periodic config command [options]" + putStrLn "" + putStrLn "Commands:" + putStrLn " get Get config" + putStrLn " set Set config" + putStrLn "" + exitSuccess + +printConfigKeyList :: IO () +printConfigKeyList = do + putStrLn "" + putStrLn "Available keys:" + putStrLn " poll-interval - poll loop every time interval" + putStrLn " revert-interval - revert process queue loop every time interval" + putStrLn " timeout - job process timeout" + putStrLn " keepalive - client keepalive" + putStrLn " max-batch-size - max poll batch size" + putStrLn " expiration - run job result cache expiration" + putStrLn "" + exitSuccess + +printConfigGetHelp :: IO () +printConfigGetHelp = do + putStrLn "periodic config get - Get config" + putStrLn "" + putStrLn "Usage: periodic config get key" + printConfigKeyList + +printConfigSetHelp :: IO () +printConfigSetHelp = do + putStrLn "periodic config set - Set config" + putStrLn "" + putStrLn "Usage: periodic config set key val" + printConfigKeyList + +printLoadHelp :: IO () +printLoadHelp = do + putStrLn "periodic load - Load jobs" + putStrLn "" + putStrLn "Usage: periodic load file" + putStrLn "" + exitSuccess + +printDumpHelp :: IO () +printDumpHelp = do + putStrLn "periodic dump - Dump jobs" + putStrLn "" + putStrLn "Usage: periodic dump file" + putStrLn "" + exitSuccess + +main :: IO () +main = do + h <- lookupEnv "PERIODIC_PORT" + f <- lookupEnv "XOR_FILE" + + (cmd, opts@Options {..}, argv) <- flip parseOptions (options h f) <$> getArgs + + let argc = length argv + + when (cmd == Help) printHelp + + when (cmd == Submit && argc < 2) printSubmitHelp + when (cmd == Run && argc < 2) printRunHelp + when (cmd == Remove && argc < 2) printRemoveHelp + when (cmd == Drop && argc < 1) printDropHelp + when (cmd == Config && argc < 1) printConfigHelp + when (cmd == Load && argc < 1) printLoadHelp + when (cmd == Dump && argc < 1) printDumpHelp + + when (not ("tcp" `isPrefixOf` host) && not ("unix" `isPrefixOf` host)) $ do + putStrLn $ "Invalid host " ++ host + printHelp + + run opts cmd argv + +run Options {useTls = True, ..} cmd argv = do + prms <- makeClientParams' cert [] certKey caStore (hostName, B.pack $ fromMaybe "" $ getService host) + clientEnv <- open (tlsConfig prms (socket host)) + runClientT clientEnv $ processCommand cmd argv + +run Options {useWs = True, ..} cmd argv = do + clientEnv <- open (clientConfig (socket host) (fromMaybe "0.0.0.0" $ getHost host) (fromMaybe "" $ getService host)) + runClientT clientEnv $ processCommand cmd argv + +run Options {xorFile = "", ..} cmd argv = do + clientEnv <- open (socket host) + runClientT clientEnv $ processCommand cmd argv + +run Options {..} cmd argv = do + clientEnv <- open (xorConfig xorFile $ socket host) + runClientT clientEnv $ processCommand cmd argv + +processCommand :: Transport tp => Command -> [String] -> ClientT tp IO () +processCommand Help _ = liftIO printHelp +processCommand Status _ = doStatus +processCommand Submit xs = doSubmitJob xs +processCommand Run xs = doRunJob xs +processCommand Remove xs = doRemoveJob xs +processCommand Drop xs = doDropFunc xs +processCommand Config xs = doConfig xs +processCommand Load xs = doLoad xs +processCommand Dump xs = doDump xs +processCommand Shutdown _ = shutdown + +doRemoveJob :: Transport tp => [String] -> ClientT tp IO () +doRemoveJob (x:xs) = mapM_ (removeJob (fromString x) . fromString) xs +doRemoveJob [] = liftIO printRemoveHelp + +doDropFunc :: Transport tp => [String] -> ClientT tp IO () +doDropFunc = mapM_ (dropFunc . fromString) + +doConfig :: Transport tp => [String] -> ClientT tp IO () +doConfig ["get"] = liftIO printConfigGetHelp +doConfig ["get", k] = do + v <- configGet k + liftIO $ print v +doConfig ("get":_:_) = liftIO printConfigGetHelp +doConfig ["set"] = liftIO printConfigSetHelp +doConfig ["set", _] = liftIO printConfigSetHelp +doConfig ["set", k, v] = void $ configSet k (read v) +doConfig ("set":_:_:_) = liftIO printConfigSetHelp +doConfig _ = liftIO printConfigHelp + +doLoad :: Transport tp => [String] -> ClientT tp IO () +doLoad [fn] = do + jobs <- liftIO $ decodeFile fn + void $ load jobs + +doLoad _ = liftIO printLoadHelp + +doDump :: Transport tp => [String] -> ClientT tp IO () +doDump [fn] = do + jobs <- dump + liftIO $ encodeFile fn jobs + +doDump _ = liftIO printDumpHelp + +doSubmitJob :: Transport tp => [String] -> ClientT tp IO () +doSubmitJob [] = liftIO printSubmitHelp +doSubmitJob [_] = liftIO printSubmitHelp +doSubmitJob (x:y:xs) = do + (wl, l) <- liftIO $ go (Nothing, Nothing) xs + void $ submitJob (fromString x) (fromString y) wl l + + where go :: (Maybe Workload, Maybe Int64) -> [String] -> IO (Maybe Workload, Maybe Int64) + go v [] = pure v + go (_, l0) ("-w":('@':f):ys) = do + w <- B.readFile f + go (Just (Workload w), l0) ys + go (_, l0) ("--workload":('@':f):ys) = do + w <- B.readFile f + go (Just (Workload w), l0) ys + go (_, l0) ("-w":w:ys) = go (Just (fromString w), l0) ys + go (_, l0) ("--workload":w:ys) = go (Just (fromString w), l0) ys + go (w, _) ("--later":l0:ys) = go (w, readMaybe l0) ys + go v (_:ys) = go v ys + +doRunJob :: Transport tp => [String] -> ClientT tp IO () +doRunJob [] = liftIO printRunHelp +doRunJob [_] = liftIO printRunHelp +doRunJob (x:y:xs) = + liftIO . putR + =<< runJob (fromString x) (fromString y) + =<< liftIO (go xs) + + where go :: [String] -> IO (Maybe Workload) + go [] = pure Nothing + go ("-w":('@':f):_) = getFile f + go ("--workload":('@':f):_) = getFile f + go ("-w":w:_) = pure $ Just (fromString w) + go ("--workload":w:_) = pure $ Just (fromString w) + go (_:ys) = go ys + + getFile :: String -> IO (Maybe Workload) + getFile fn = Just . Workload <$> B.readFile fn + + putR :: Maybe ByteString -> IO () + putR Nothing = putStrLn "Error: run job failed" + putR (Just bs) = B.putStr bs + +doStatus :: Transport tp => ClientT tp IO () +doStatus = do + st <- map formatTime . unpackBS . map (B.split ',') . B.lines <$> status + liftIO $ printTable (["FUNCTIONS", "WORKERS", "JOBS", "PROCESSING", "LOCKING", "SCHEDAT"]:st) + +unpackBS :: [[ByteString]] -> [[String]] +unpackBS = map (map B.unpack) + +formatTime :: [String] -> [String] +formatTime [] = [] +formatTime [x] = [formatUnixTimeLocal x] +formatTime (x:xs) = x:formatTime xs + +formatUnixTimeLocal :: String -> String +formatUnixTimeLocal = B.unpack + . unsafePerformIO + . formatUnixTime "%Y-%m-%d %H:%M:%S" + . fromEpochTime + . fromIntegral + . read + +printTable :: [[String]] -> IO () +printTable rows = T.printBox $ T.hsep 2 T.left (map (T.vcat T.left . map T.text) (transpose rows)) diff --git a/periodic-client-exe.cabal b/periodic-client-exe.cabal new file mode 100644 --- /dev/null +++ b/periodic-client-exe.cabal @@ -0,0 +1,76 @@ +name: periodic-client-exe +version: 1.1.7.1 +synopsis: Periodic task system haskell client executables +description: Periodic task system haskell client executables. +homepage: https://github.com/Lupino/haskell-periodic/tree/master/periodic-client-exe#readme +license: BSD3 +license-file: LICENSE +author: Li Meng Jun +maintainer: lmjubuntu@gmail.com +copyright: MIT +category: System,Web +build-type: Simple +extra-source-files: README.md +cabal-version: >=1.10 + +executable periodic + hs-source-dirs: app + main-is: periodic.hs + ghc-options: -threaded -rtsopts -with-rtsopts=-N + build-depends: base >= 4.7 && < 5 + , periodic-client + , periodic-common + , bytestring + , boxes + , unix-time + , binary + , metro + , metro-socket + , metro-transport-xor + , metro-transport-tls + , metro-transport-websockets + default-language: Haskell2010 + +executable periodic-run + hs-source-dirs: app + main-is: periodic-run.hs + ghc-options: -threaded -rtsopts -with-rtsopts=-N + build-depends: base >= 4.7 && < 5 + , periodic-client + , periodic-common + , bytestring + , text + , process + , unliftio + , deepseq + , metro + , metro-socket + , metro-transport-xor + , metro-transport-tls + , metro-transport-websockets + default-language: Haskell2010 + +executable periodic-http-bridge + hs-source-dirs: app + main-is: periodic-http-bridge.hs + ghc-options: -threaded -rtsopts -with-rtsopts=-N + build-depends: base >= 4.7 && < 5 + , periodic-client + , periodic-common + , bytestring + , scotty + , warp + , data-default-class + , streaming-commons + , http-types + , metro + , metro-socket + , metro-transport-xor + , metro-transport-tls + , metro-transport-websockets + , websockets + default-language: Haskell2010 + +source-repository head + type: git + location: https://github.com/Lupino/haskell-periodic