diff --git a/app/Main.hs b/app/Main.hs --- a/app/Main.hs +++ b/app/Main.hs @@ -97,7 +97,9 @@ -- connection to be closed cleanly when using ZeroMQ. Code taken from -- http://zguide.zeromq.org/hs:interrupt handler :: MVar () -> IO () -handler s_interrupted = trace "Interrupt received" $ putMVar s_interrupted () +handler s_interrupted = do + putStrLn "Interrupt received" + putMVar s_interrupted () -------------------------------------------------------------------------------- -- | 'messageSink' yields the parsed JSON downstream, or if parsing fails, yields the original message downstream @@ -126,12 +128,12 @@ v <- await case v of Just (Transformed json) -> do - trace "successfull parse" $ yield (SBS.pack "success: ") + yield (SBS.pack "success: ") yield json yield (SBS.pack "\n") loop Just (Original l) -> do - trace "failed parse" $ yield (SBS.pack "fail - original: ") + yield (SBS.pack "fail - original: ") yield l yield (SBS.pack "\n") loop @@ -175,8 +177,8 @@ (liftIO $ do val <- tryTakeMVar m case val of - Just _ -> trace "interrupt received, stopping source" $ return False - Nothing -> trace "still going ..." $ return True) + Just _ -> return False + Nothing -> return True) (lift (ZMQ.receive s) >>= yield) liftIO $ putStrLn "Done!" @@ -185,7 +187,7 @@ let fs = fields config let listenHost = fromJust $ input config >>= (\(InputConfig _ z) -> z) >>= (\(ZeroMQPortConfig m h p) -> h) let listenPort = fromJust $ input config >>= (\(InputConfig _ z) -> z) >>= (\(ZeroMQPortConfig m h p) -> p) - trace (printf "listening on tcp://%s:%d" listenHost listenPort) $ return () + liftIO $ putStrLn (printf "listening on tcp://%s:%d" listenHost listenPort) case oTestFilePath options of Nothing -> do @@ -201,9 +203,9 @@ ZMQ.withSocket ctx ZMQ.Push $ \successSocket -> ZMQ.withSocket ctx ZMQ.Push $ \failureSocket -> do ZMQ.connect successSocket $ printf "tcp://%s:%d" successHost successPort - trace (printf "connected to tcp://%s:%d" successHost successPort) $ return () + liftIO $ putStrLn (printf "connected to tcp://%s:%d" successHost successPort) ZMQ.connect failureSocket $ printf "tcp://%s:%d" failureHost failurePort - trace (printf "connected to tcp://%s:%d" failureHost failurePort) $ return () + liftIO $ putStrLn (printf "connected to tcp://%s:%d" failureHost failurePort) let normalisationConduit = case oJsonInput options of True -> CB.lines $= C.map (normaliseJsonInput fs) @@ -238,9 +240,6 @@ exitSuccess config <- loadConfig (oConfigFilePath options) - --trace (show config) $ return () - -- install the signal handlers for a clean shutdown - -- For now, we only support input and output configurations of the same type, -- i.e., both TCP, both ZeroMQ, etc. diff --git a/hnormalise.cabal b/hnormalise.cabal --- a/hnormalise.cabal +++ b/hnormalise.cabal @@ -1,5 +1,5 @@ name: hnormalise -version: 0.4.1.1 +version: 0.4.2.0 synopsis: Log message normalisation tool producing structured JSON messages description: Log message normalisation tool producing structured JSON messages homepage: https://github.com/itkovian/hnormalise#readme diff --git a/src/HNormalise/Torque/Internal.hs b/src/HNormalise/Torque/Internal.hs --- a/src/HNormalise/Torque/Internal.hs +++ b/src/HNormalise/Torque/Internal.hs @@ -108,7 +108,7 @@ , qtime :: !Integer , etime :: !Integer , startTime :: !Integer - , endTime :: !Integer + , endTime :: !(Maybe Integer) } deriving (Show, Eq, Generic) -------------------------------------------------------------------------------- @@ -136,4 +136,35 @@ , array_id :: !(Maybe Integer) , master :: !Text , cluster :: !Text + } deriving (Show, Eq, Generic) + +-------------------------------------------------------------------------------- +data TorqueJobQueue = TorqueJobQueue + { name :: !TorqueJobName + , queue :: !Text + } deriving (Show, Eq, Generic) + +-------------------------------------------------------------------------------- +data TorqueJobStart = TorqueJobStart + { name :: !TorqueJobName + , user :: !Text + , group :: !Text + , jobname :: !Text + , queue :: !Text + , owner :: !Text + , times :: !TorqueJobTime + , execHost :: ![TorqueExecHost] + , resourceRequest :: !TorqueResourceRequest + } deriving (Show, Eq, Generic) + +-------------------------------------------------------------------------------- +data TorqueRequestor = TorqueRequestor + { user :: !Text + , whence :: !Text + } deriving (Show, Eq, Generic) + +-------------------------------------------------------------------------------- +data TorqueJobDelete = TorqueJobDelete + { name :: !TorqueJobName + , requestor :: !TorqueRequestor } deriving (Show, Eq, Generic) diff --git a/src/HNormalise/Torque/Parser.hs b/src/HNormalise/Torque/Parser.hs --- a/src/HNormalise/Torque/Parser.hs +++ b/src/HNormalise/Torque/Parser.hs @@ -43,7 +43,7 @@ import Control.Applicative ((<|>)) import Data.Attoparsec.Combinator (lookAhead, manyTill) import Data.Attoparsec.Text -import Data.Char (isDigit) +import Data.Char (isDigit, isSpace) import qualified Data.Map as M import Data.Text (Text) import qualified Data.Text as T @@ -212,7 +212,20 @@ c <- decimal return [c] +-------------------------------------------------------------------------------- +-- | `parseTorqueRequestor` parses a requestor string, i.e., the user plus machine issueing e.g., a delete request +parseTorqueRequestor :: Parser TorqueRequestor +parseTorqueRequestor = do + string "requestor=" + user <- takeTill (== '@') + char '@' + whence <- takeTill isSpace -- FIXME: this works in the given contexts, but might not be general enough. + return TorqueRequestor + { user = user + , whence = whence + } + -------------------------------------------------------------------------------- -- | 'parseTorqueExit' parses a complete log line denoting a job exit. Tested with Torque 6.1.x. parseTorqueExit :: Parser (Text, TorqueJobExit) @@ -252,7 +265,7 @@ , qtime = qtime , etime = etime , startTime = start - , endTime = end + , endTime = Just end } , execHost = exec_host , resourceRequest = request @@ -260,4 +273,66 @@ , totalExecutionSlots = total_execution_slots , uniqueNodeCount = unique_node_count , exitStatus = exit_status + }) + +-------------------------------------------------------------------------------- +-- | `parseTorqueDelete` parses a complete log line denoting a deleted job. Tested with Torue 6.1.x +parseTorqueDelete :: Parser (Text, TorqueJobDelete) +parseTorqueDelete = do + takeTill (== ';') *> string ";D;" -- drop the prefix + name <- parseTorqueJobName + requestor <- parseTorqueRequestor + + return ("torque", TorqueJobDelete + { name = name + , requestor = requestor + }) + +-------------------------------------------------------------------------------- +-- | `parseTorqueQueue` parses a complete log line denoting a queued job. Tested with Torue 6.1.x +parseTorqueQueue :: Parser (Text, TorqueJobQueue) +parseTorqueQueue = do + takeTill (== ';') *> string ";Q;" -- drop the prefix + name <- parseTorqueJobName + queue <- kvTextParser "queue" + + return ("torque", TorqueJobQueue + { name = name + , queue = queue + }) + +-------------------------------------------------------------------------------- +-- | `parseTorqueStart` parses a complete log line denoting a started job. Tested with Torque 6.1.x +parseTorqueStart :: Parser (Text, TorqueJobStart) +parseTorqueStart = do + takeTill (== ';') *> string ";S;" -- drop the prefix + name <- parseTorqueJobName + user <- kvTextParser "user" + group <- skipSpace *> kvTextParser "group" + jobname <- skipSpace *> kvTextParser "jobname" + queue <- skipSpace *> kvTextParser "queue" + ctime <- skipSpace *> kvNumParser "ctime" + qtime <- skipSpace *> kvNumParser "qtime" + etime <- skipSpace *> kvNumParser "etime" + start <- skipSpace *> kvNumParser "start" + owner <- skipSpace *> kvTextParser "owner" + exec_host <- skipSpace *> parseTorqueHostList + request <- parseTorqueResourceRequest + + return $ ("torque", TorqueJobStart + { name = name + , user = user + , group = group + , jobname = jobname + , queue = queue + , owner = owner + , times = TorqueJobTime + { ctime = ctime + , qtime = qtime + , etime = etime + , startTime = start + , endTime = Nothing + } + , execHost = exec_host + , resourceRequest = request }) diff --git a/test/Bench.hs b/test/Bench.hs --- a/test/Bench.hs +++ b/test/Bench.hs @@ -62,6 +62,10 @@ torqueJobExitFailInput1 = "04/05/2017 13:06:53;E;45.master23.banette.gent.vsc;user=vsc40075 group=vsc40075 jobname=STDIN queue=short HUPPEL" +torqueJobQueueInput = "06/28/2017 14:31:09;Q;80.master23.banette.gent.vsc;queue=default" +torqueJobDeleteInput = "06/28/2017 15:44:02;D;81.master23.banette.gent.vsc;requestor=vsc40075@gligar02.gligar.gent.vsc" +torqueJobStartInput = "06/20/2017 11:24:49;S;63.master23.banette.gent.vsc;user=vsc40075 group=vsc40075 jobname=STDIN queue=short ctime=1497950675 qtime=1497950675 etime=1497950675 start=1497950689 owner=vsc40075@gligar01.gligar.gent.vsc exec_host=node2801.banette.gent.vsc/0 Resource_List.vmem=4224531456b Resource_List.nodes=1:ppn=1 Resource_List.walltime=00:10:00 Resource_List.nodect=1 Resource_List.neednodes=1:ppn=1 Resource_List.nice=0" + lmodLoadInput1 = "lmod:: username=myuser, cluster=mycluster, jobid=3230905.master.mycluster.mydomain, userload=yes, module=GSL/2.3-intel-2016b, fn=/apps/gent/CO7/sandybridge/modules/all/GSL/2.3-intel-2016b" :: Text fullLmodInput = "<13>1 2016-06-07T17:50:22.495571+02:00 node2159 lmod: - lmod:: username=vsc40307, cluster=delcatty, jobid=434.master16.delcatty.gent.vsc, userload=no, module=binutils/2.25-GCCcore-4.9.3, fn=/apps/gent/SL6/sandybridge/modules/all/binutils/2.25-GCCcore-4.9.3" :: Text @@ -80,6 +84,9 @@ [ bench "jobexit full resource node list" $ whnf (AT.parse TorqueP.parseTorqueExit) torqueJobExitInput1 , bench "jobexit short resource node number" $ whnf (AT.parse TorqueP.parseTorqueExit) torqueJobExitInput1 , bench "jobexit borked input" $ whnf (AT.parse TorqueP.parseTorqueExit) torqueJobExitFailInput1 + , bench "jobqueue input" $ whnf (AT.parse TorqueP.parseTorqueQueue) torqueJobQueueInput + , bench "jobdelete input" $ whnf (AT.parse TorqueP.parseTorqueDelete) torqueJobDeleteInput + , bench "jobstart input" $ whnf (AT.parse TorqueP.parseTorqueStart) torqueJobStartInput ] , bgroup "parse lmod" [ bench "lmod successfull module load parse" $ whnf (AT.parse LmodP.parseLmodLoad) lmodLoadInput1] diff --git a/test/HNormalise/Torque/ParserSpec.hs b/test/HNormalise/Torque/ParserSpec.hs --- a/test/HNormalise/Torque/ParserSpec.hs +++ b/test/HNormalise/Torque/ParserSpec.hs @@ -283,7 +283,7 @@ , qtime = 1491390300 , etime = 1491390300 , startTime = 1491390307 - , endTime = 1491390413 + , endTime = Just 1491390413 } , execHost = [ TorqueExecHost @@ -339,4 +339,71 @@ , totalExecutionSlots = 4 , uniqueNodeCount = 2 , exitStatus = 0 + }) + + describe "parseTorqueQueue" $ do + it "parse job queue entry" $ do + let s = "06/28/2017 14:31:09;Q;80.master23.banette.gent.vsc;queue=default" :: Text + s ~> parseTorqueQueue `shouldParse` ("torque", TorqueJobQueue + { name = TorqueJobName { number = 80, array_id = Nothing, master = "master23", cluster = "banette" } + , queue = "default" + }) + + describe "parseTorqueDelete" $ do + it "parse job delete entry" $ do + let s = "06/28/2017 15:44:02;D;81.master23.banette.gent.vsc;requestor=vsc40075@gligar02.gligar.gent.vsc" :: Text + s ~> parseTorqueDelete `shouldParse` ("torque", TorqueJobDelete + { name = TorqueJobName { number = 81, array_id = Nothing, master = "master23", cluster = "banette" } + , requestor = TorqueRequestor { user = "vsc40075", whence = "gligar02.gligar.gent.vsc" } + }) + + describe "parseTorqueStart" $ do + it "parse job start" $ do + let s = "06/20/2017 11:24:49;S;63.master23.banette.gent.vsc;user=vsc40075 group=vsc40075 jobname=STDIN queue=short ctime=1497950675 qtime=1497950675 etime=1497950675 start=1497950689 owner=vsc40075@gligar01.gligar.gent.vsc exec_host=node2801.banette.gent.vsc/0 Resource_List.vmem=4224531456b Resource_List.nodes=1:ppn=1 Resource_List.walltime=00:10:00 Resource_List.nodect=1 Resource_List.neednodes=1:ppn=1 Resource_List.nice=0" :: Text + s ~> parseTorqueStart `shouldParse` ("torque", TorqueJobStart + { name = TorqueJobName { number = 63, array_id = Nothing, master = "master23", cluster = "banette" } + , user = "vsc40075" + , group = "vsc40075" + , jobname = "STDIN" + , queue = "short" + , owner = "vsc40075@gligar01.gligar.gent.vsc" + , times = TorqueJobTime + { ctime = 1497950675 + , qtime = 1497950675 + , etime = 1497950675 + , startTime = 1497950689 + , endTime = Nothing + } + , execHost = + [ TorqueExecHost + { name = "node2801.banette.gent.vsc" + , cores = [0] + } + ] + , resourceRequest = TorqueResourceRequest + { mem = Nothing + , advres = Nothing + , naccesspolicy = Nothing + , ncpus = Nothing + , neednodes = TSN + ( TorqueJobShortNode + { number = 1 + , ppn = Just 1 + } + ) + , nice = Just 0 + , nodeCount = 1 + , nodes = TSN + ( TorqueJobShortNode + { number = 1 + , ppn = Just 1 + } + ) + , select = Nothing + , qos = Nothing + , vmem = Just $ 4224531456 + , pmem = Nothing + , pvmem = Nothing + , walltime = TorqueWalltime { days = 0, hours = 0, minutes = 10, seconds = 0} + } })