hnormalise 0.4.7.0 → 0.4.8.0
raw patch · 7 files changed
+407/−103 lines, 7 filesdep +monad-pardep −deepseq
Dependencies added: monad-par
Dependencies removed: deepseq
Files
- 0mq/Main.hs +290/−48
- README.md +6/−3
- app/Main.hs +28/−22
- hnormalise.cabal +13/−10
- src/HNormalise.hs +5/−5
- src/HNormalise/Torque/Parser.hs +57/−4
- test/Bench.hs +8/−11
0mq/Main.hs view
@@ -1,62 +1,304 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE OverloadedStrings #-}+ module Main where -import System.Posix.Signals (installHandler, Handler(Catch), Handler(CatchOnce), sigINT, sigTERM)-import Control.Concurrent.MVar (modifyMVar_, newMVar, withMVar, tryTakeMVar, putMVar, MVar, newEmptyMVar)+--------------------------------------------------------------------------------+import Control.Applicative ((<$>), (<*>))+import Control.Concurrent.MVar (modifyMVar_, newMVar, newEmptyMVar, putMVar, readMVar, tryTakeMVar, withMVar, takeMVar, MVar)+import Control.Monad+import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Monad.Loops (whileM_, untilM_)+import Control.Monad.Par+import Control.Monad.Trans (lift)+import Control.Monad.Trans.Resource (runResourceT)+import Data.Aeson+import Data.Attoparsec.Text+import qualified Data.ByteString.Char8 as SBS+import qualified Data.ByteString.Lazy.Char8 as BS+import Data.Conduit+import Data.Conduit.Binary (sinkFile)+import qualified Data.Conduit.Binary as CB+import qualified Data.Conduit.Combinators as C+import Data.Conduit.Network+import qualified Data.Conduit.Text as DCT+import qualified Data.Conduit.ZMQ4 as ZMQC+import Data.Maybe (fromJust, fromMaybe)+import Data.Monoid ((<>))+import qualified Data.Text as T+import qualified Data.Text.Encoding as TE+import Data.Time+import Data.Time.Format+import Data.Version (showVersion)+import qualified Options.Applicative as OA+import qualified Paths_hnormalise+import System.Exit (exitFailure, exitSuccess)+import System.Posix.Signals (installHandler, Handler(CatchOnce), sigINT, sigTERM)+import qualified System.ZMQ4 as ZMQ (withContext, withSocket, bind, send, receive, connect, Push(..), Pull(..))+import Text.Printf (printf) -import System.ZMQ4-import Debug.Trace+import Debug.Trace+--------------------------------------------------------------------------------+import HNormalise+import HNormalise.Config ( Config (..)+ , ConnectionType(..)+ , InputConfig(..)+ , LoggingConfig(..)+ , OutputConfig(..)+ , TcpOutputConfig(..)+ , TcpPortConfig(..)+ , ZeroMQOutputConfig(..)+ , ZeroMQPortConfig(..)+ , connectionType+ , defaultLoggingFrequency+ , loadConfig)+import HNormalise.Internal (Rsyslog (..))+import HNormalise.Json+import HNormalise.Verbose -handler :: MVar () -> Handler-handler s_interrupted = CatchOnce $ trace "modifying" $ putMVar s_interrupted ()+--------------------------------------------------------------------------------+data Options = Options+ { oConfigFilePath :: !(Maybe FilePath)+ , oJsonInput :: !Bool+ , oVersion :: !Bool+ , oTestFilePath :: !(Maybe FilePath)+ , oVerbose :: !Bool+ } deriving (Show) -main :: IO ()-main = do- withContext $ \ctx -> do- withSocket ctx Pull $ \socket -> do- bind socket "tcp://*:5555"- s_interrupted <- newEmptyMVar- installHandler sigINT (handler s_interrupted) Nothing- installHandler sigTERM (handler s_interrupted) Nothing- recvFunction s_interrupted socket- trace "outside withSocket" $ return ()- trace "outside withContext" $ return ()+--------------------------------------------------------------------------------+parserOptions :: OA.Parser Main.Options+parserOptions = Options+ <$> OA.optional ( OA.strOption $+ OA.long "configfile" <>+ OA.short 'c' <>+ OA.metavar "FILENAME" <>+ OA.help "configuration file location ")+ <*> OA.switch (+ OA.long "jsoninput" <>+ OA.short 'j' <>+ OA.help "Input will be delivered as JSON (slower)")+ <*> OA.switch (+ OA.long "version" <>+ OA.short 'v' <>+ OA.help "Display version and exit" <>+ OA.hidden)+ <*> OA.optional ( OA.strOption $+ OA.long "test" <>+ OA.short 't' <>+ OA.metavar "OUTPUT FILENAME" <>+ OA.help "run in test modus, sinking output to the given file")+ <*> (OA.switch $+ OA.long "verbose" <>+ OA.help "Verbose output" <>+ OA.hidden) -recvFunction :: (Receiver b) => MVar () -> Socket b -> IO ()-recvFunction mi sock = do- msg <- receive sock- putStrLn $ "message received: " ++ show msg- val <- tryTakeMVar mi- case val of- Just _ -> putStrLn "W: Interrupt received. Killing Server"- Nothing -> recvFunction mi sock-{-} withMVar mi (\val -> if val > 0- then putStrLn "W: Interrupt Received. Killing Server"- else recvFunction mi sock)+--------------------------------------------------------------------------------+parserInfo :: OA.ParserInfo Main.Options+parserInfo = OA.info (OA.helper <*> parserOptions)+ (OA.fullDesc+ <> OA.progDesc "Normalise rsyslog messages"+ <> OA.header ("hNormalise v" <> showVersion Paths_hnormalise.version)+ )++--------------------------------------------------------------------------------+-- | 'handler' will catch SIGTERM and modify the MVar to allow the upstream+-- connection to be closed cleanly when using ZeroMQ. Code taken from+-- http://zguide.zeromq.org/hs:interrupt+handler :: MVar () -> IO ()+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+messageSink success failure messageCount frequency = loop+ where+ loop = do+ v <- await+ case v of+ Just (Transformed json) -> do+ yield json $$ success+ yield (SBS.pack "\n") $$ success+ liftIO $ increaseCount (1, 0) messageCount frequency+ loop+ Just (Original l) -> do+ yield l $$ failure+ yield (SBS.pack "\n") $$ failure+ liftIO $ increaseCount (0, 1) messageCount frequency+ loop+ Nothing -> return ()++increaseCount (s, f) messageCount frequency = do+ (s', f') <- takeMVar messageCount+ when ((s' + f') `mod` frequency == 0) $ do+ epoch_int <- (read . formatTime defaultTimeLocale "%s" <$> getCurrentTime) :: IO Int+ printf "%ld - message count: %10d (success: %10d, fail: %10d)\n" epoch_int (s' + f') s' f'+ putMVar messageCount (s' + s, f' + f)++--------------------------------------------------------------------------------+-- | 'mySink' yields the results downstream with the addition of a string mentioning success or failure+-- for testing purposes+mySink messageCount frequency = loop+ where+ loop = do+ v <- await+ case v of+ Just (Transformed json) -> do+ yield (SBS.pack "success: ")+ yield json+ yield (SBS.pack "\n")+ liftIO $ increaseCount (1, 0) messageCount frequency+ loop+ Just (Original l) -> do+ yield (SBS.pack "fail - original: ")+ yield l+ yield (SBS.pack "\n")+ liftIO $ increaseCount (0, 1) messageCount frequency+ loop+ Nothing -> return ()++--------------------------------------------------------------------------------+{-runTCPConnection options config messageCount = do+ let listenHost = fromJust $ input config >>= (\(InputConfig t _) -> t) >>= (\(TcpPortConfig h p) -> h)+ let listenPort = fromJust $ input config >>= (\(InputConfig t _) -> t) >>= (\(TcpPortConfig h p) -> p)+ let frequency = fromMaybe defaultLoggingFrequency (logging config >>= (\(LoggingConfig f) -> f))+ runTCPServer (serverSettings listenPort "*") $ \appData -> do+ let fs = fields config+ case oTestFilePath options of+ Nothing -> do+ let successHost = TE.encodeUtf8 $ fromJust $ output config >>= (\(OutputConfig t _) -> t) >>= (\(TcpOutputConfig s f) -> s) >>= (\(TcpPortConfig h p) -> h)+ let successPort = fromJust $ output config >>= (\(OutputConfig t _) -> t) >>= (\(TcpOutputConfig s f) -> s) >>= (\(TcpPortConfig h p) -> p)+ let failureHost = TE.encodeUtf8 $ fromJust $ output config >>= (\(OutputConfig t _) -> t) >>= (\(TcpOutputConfig s f) -> f) >>= (\(TcpPortConfig h p) -> h)+ let failurePort = fromJust $ output config >>= (\(OutputConfig t _) -> t) >>= (\(TcpOutputConfig s f) -> f) >>= (\(TcpPortConfig h p) -> p)++ runTCPClient (clientSettings successPort successHost) $ \successServer ->+ runTCPClient (clientSettings failurePort failureHost) $ \failServer ->+ appSource appData+ $= normalisationConduit options config+ $$ messageSink (appSink successServer) (appSink failServer) messageCount frequency+ Just testSinkFileName -> runResourceT $ appSource appData+ $= normalisationConduit options config+ $= mySink messageCount frequency+ $$ sinkFile testSinkFileName -}-{-+-- | 'zmqInterruptibleSource' converts a regular 0mq recieve operation on a socket into a conduit source+-- The source is halted when something is put into the MVar, effectively stopping the program from checking+-- for new incoming messages.+zmqInterruptibleSource m s = do+ whileM_+ (liftIO $ do+ val <- tryTakeMVar m+ case val of+ Just _ -> return False+ Nothing -> return True)+ (liftIO (ZMQ.receive s) >>= yield)+ liftIO $ putStrLn "Done!" -import System.Posix.Signals-import Control.Concurrent (threadDelay)-import Control.Concurrent.MVar+--------------------------------------------------------------------------------+normalisationConduit options config =+ let fs = fields config+ in if oJsonInput options+ then CB.lines $= C.map (normaliseJsonInput fs)+ --else CB.lines $= conduitPooledMapBuffered 20 (normaliseText fs)+ else CB.lines $= C.map (normaliseText fs) -termHandler :: MVar () -> Handler-termHandler v = CatchOnce $ do- putStrLn "Caught SIGTERM"- putMVar v ()+--------------------------------------------------------------------------------+{- runZeroMQConnection :: Main.Options+ -> Config+ -> MVar a1+ -> MVar (Int, Int)+ -> Verbose+ -> IO ()+runZeroMQConnection options config s_interrupted messageCount verbose' = do+ 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)+ let frequency = fromMaybe defaultLoggingFrequency (logging config >>= (\(LoggingConfig f) -> f)) -loop :: MVar () -> IO ()-loop v = do- putStrLn "Still running"- threadDelay 1000000- val <- tryTakeMVar v- case val of- Just _ -> putStrLn "Quitting" >> return ()- Nothing -> loop v+ case oTestFilePath options of+ Nothing -> ZMQ.withContext $ \ctx ->+ ZMQ.withSocket ctx ZMQ.Pull $ \s -> do+ ZMQ.bind s $ printf "tcp://%s:%d" listenHost listenPort+ verbose' $ printf "Listening on tcp://%s:%d" listenHost listenPort -main = do- v <- newEmptyMVar- installHandler sigTERM (termHandler v) Nothing- loop v⏎+ let successHost = fromJust $ output config >>= (\(OutputConfig _ z) -> z) >>= (\(ZeroMQOutputConfig s f) -> s) >>= (\(ZeroMQPortConfig m h p) -> h)+ let successPort = fromJust $ output config >>= (\(OutputConfig _ z) -> z) >>= (\(ZeroMQOutputConfig s f) -> s) >>= (\(ZeroMQPortConfig m h p) -> p)+ let failureHost = fromJust $ output config >>= (\(OutputConfig _ z) -> z) >>= (\(ZeroMQOutputConfig s f) -> f) >>= (\(ZeroMQPortConfig m h p) -> h)+ let failurePort = fromJust $ output config >>= (\(OutputConfig _ z) -> z) >>= (\(ZeroMQOutputConfig s f) -> f) >>= (\(ZeroMQPortConfig m h p) -> p) + ZMQ.withSocket ctx ZMQ.Push $ \successSocket ->+ ZMQ.withSocket ctx ZMQ.Push $ \failureSocket -> do+ ZMQ.connect successSocket $ printf "tcp://%s:%d" successHost successPort+ verbose' $ printf "Pushing successful parses on tcp://%s:%d" successHost successPort++ ZMQ.connect failureSocket $ printf "tcp://%s:%d" failureHost failurePort+ verbose' $ printf "Pushing failed parses on tcp://%s:%d" failureHost failurePort+ liftIO $ zmqInterruptibleSource s_interrupted s+ $= normalisationConduit options config+ $$ messageSink (ZMQC.zmqSink successSocket []) (ZMQC.zmqSink failureSocket []) messageCount frequency++ Just testSinkFileName ->+ ZMQ.withContext $ \ctx ->+ ZMQ.withSocket ctx ZMQ.Pull $ \s -> do+ ZMQ.bind s $ printf "tcp://%s:%d" listenHost listenPort+ verbose' $ printf "Listening on tcp://%s:%d" listenHost listenPort+ runResourceT $ zmqInterruptibleSource s_interrupted s+ $= normalisationConduit options config+ $= mySink messageCount frequency+ $$ sinkFile testSinkFileName -}++passSink messageCount frequency = loop+ where+ loop = do+ v <- await+ case v of+ Just s -> do+ yield s+ liftIO $ increaseCount (1, 0) messageCount frequency+ loop+ Nothing -> return ()++runFromSourceFile options config messageCount verbose' =+ runResourceT $ C.sourceFile "/tmp/hnorm_test_input"+ $= normalisationConduit options config+ -- $= DCT.decode DCT.utf8 $= DCT.lines $= DCT.encode DCT.utf8 -- 2.2s+ -- $= DCT.decode DCT.utf8 $= DCT.lines $= DCT.encode DCT.utf8 -- 2.2s+ -- $= CB.lines $= DCT.decode DCT.utf8 $= DCT.encode DCT.utf8+ -- $= passSink messageCount 10000+ $= mySink messageCount 10000+ $$ sinkFile "/dev/null" -- "/tmp/hnorm_test_output"++--------------------------------------------------------------------------------+-- | 'main' starts a TCP server, listening to incoming data and connecting to TCP servers downstream to+-- for the pipeline.+main :: IO ()+main = do+ options <- OA.execParser parserInfo++ when (oVersion options) $ do+ putStrLn $ showVersion Paths_hnormalise.version+ exitSuccess++ let verbose' = makeVerbose (oVerbose options)+ config <- loadConfig (oConfigFilePath options)++ verbose' $ "Config: " ++ show config++ -- For now, we only support input and output configurations of the same type,+ -- i.e., both TCP, both ZeroMQ, etc.+ messageCount <- newMVar ((0,0) :: (Int, Int))+ {-case connectionType config of+ TCP -> void $ runTCPConnection options config messageCount+ ZeroMQ -> do+ verbose' "ZeroMQ connections"+ s_interrupted <- newEmptyMVar+ installHandler sigINT (CatchOnce $ handler s_interrupted) Nothing+ verbose' "SIGINT handler installed"+ installHandler sigTERM (CatchOnce $ handler s_interrupted) Nothing+ verbose' "SIGTERM handler installed"+ runZeroMQConnection options config s_interrupted messageCount verbose'+ -}+ runFromSourceFile options config messageCount verbose'+ exitSuccess
README.md view
@@ -16,6 +16,7 @@ including rsyslog, [logstash](http://www.elastic.co/products/logstash), ... - sends out original messages to a (different) TCP or ZeroMQ port in case the parsing fails, allowing other services to process the information.+- can process up to 18K Torque log messages/s on a 2.7GHz 2015 Corei5 (mid 2015 MacBook Pro) when streaming from an to a file. Usage and configuration -----------------------@@ -58,9 +59,11 @@ no definite ordering. Note that this _will_ slow down the parsing. -Caveat: at this point, we do not a priori restrict the possible parsers we unleash on each message. However, if the inbound-data can be tagged properly, we could reduce the maximal number of parsers tried and avoid extensive backtracking.-Using ZeroMQ, this could be done by using topics to tag inbound information.+Caveat: at this point, we do not a priori restrict the possible parsers we+unleash on each message. However, if the inbound data can be tagged properly,+we could reduce the maximal number of parsers tried and avoid extensive+backtracking. Using ZeroMQ, this could be done by using topics to tag inbound+information. ### Adding a new parser
app/Main.hs view
@@ -65,7 +65,7 @@ } deriving (Show) ---------------------------------------------------------------------------------parserOptions :: OA.Parser Options+parserOptions :: OA.Parser Main.Options parserOptions = Options <$> OA.optional ( OA.strOption $ OA.long "configfile" <>@@ -93,7 +93,7 @@ ---------------------------------------------------------------------------------parserInfo :: OA.ParserInfo Options+parserInfo :: OA.ParserInfo Main.Options parserInfo = OA.info (OA.helper <*> parserOptions) (OA.fullDesc <> OA.progDesc "Normalise rsyslog messages"@@ -119,26 +119,26 @@ Just (Transformed json) -> do yield json $$ success yield (SBS.pack "\n") $$ success- increaseCount (1, 0)+ liftIO $ increaseCount (1, 0) messageCount frequency loop Just (Original l) -> do yield l $$ failure yield (SBS.pack "\n") $$ failure- increaseCount (0, 1)+ liftIO $ increaseCount (0, 1) messageCount frequency loop Nothing -> return ()- increaseCount (s, f) =- liftIO $ do- (s', f') <- takeMVar messageCount- when ((s' + f') `mod` frequency == 0) $ do- epoch_int <- (read . formatTime defaultTimeLocale "%s" <$> getCurrentTime) :: IO Int- printf "%ld - message count: %10d (success: %10d, fail: %10d)\n" epoch_int (s' + f') s' f'- putMVar messageCount (s' + s, f' + f) +increaseCount (s, f) messageCount frequency = do+ (s', f') <- takeMVar messageCount+ when ((s' + f') `mod` frequency == 0) $ do+ epoch_int <- (read . formatTime defaultTimeLocale "%s" <$> getCurrentTime) :: IO Int+ printf "%ld - message count: %10d (success: %10d, fail: %10d)\n" epoch_int (s' + f') s' f'+ putMVar messageCount (s' + s, f' + f)+ -------------------------------------------------------------------------------- -- | 'mySink' yields the results downstream with the addition of a string mentioning success or failure -- for testing purposes-mySink = loop+mySink messageCount frequency = loop where loop = do v <- await@@ -147,11 +147,13 @@ yield (SBS.pack "success: ") yield json yield (SBS.pack "\n")+ -- liftIO $ increaseCount (1, 0) messageCount frequency loop Just (Original l) -> do yield (SBS.pack "fail - original: ") yield l yield (SBS.pack "\n")+ --liftIO $ increaseCount (0, 1) messageCount frequency loop Nothing -> return () @@ -171,12 +173,12 @@ runTCPClient (clientSettings successPort successHost) $ \successServer -> runTCPClient (clientSettings failurePort failureHost) $ \failServer ->- appSource appData+ runResourceT $ appSource appData $= normalisationConduit options config $$ messageSink (appSink successServer) (appSink failServer) messageCount frequency Just testSinkFileName -> runResourceT $ appSource appData $= normalisationConduit options config- $= mySink+ $= mySink messageCount frequency $$ sinkFile testSinkFileName -- | 'zmqInterruptibleSource' converts a regular 0mq recieve operation on a socket into a conduit source@@ -189,7 +191,7 @@ case val of Just _ -> return False Nothing -> return True)- (lift (ZMQ.receive s) >>= yield)+ (liftIO (ZMQ.receive s) >>= yield) liftIO $ putStrLn "Done!" --------------------------------------------------------------------------------@@ -197,10 +199,15 @@ let fs = fields config in if oJsonInput options then CB.lines $= C.map (normaliseJsonInput fs)- else DCT.decode DCT.utf8 $= DCT.lines $= C.map (normaliseText fs)-+ else CB.lines $= C.map (normaliseText fs) --------------------------------------------------------------------------------+runZeroMQConnection :: Main.Options+ -> Config+ -> MVar a1+ -> MVar (Int, Int)+ -> Verbose+ -> IO () runZeroMQConnection options config s_interrupted messageCount verbose' = do let fs = fields config let listenHost = fromJust $ input config >>= (\(InputConfig _ z) -> z) >>= (\(ZeroMQPortConfig m h p) -> h)@@ -225,20 +232,19 @@ ZMQ.connect failureSocket $ printf "tcp://%s:%d" failureHost failurePort verbose' $ printf "Pushing failed parses on tcp://%s:%d" failureHost failurePort- liftIO $ zmqInterruptibleSource s_interrupted s+ runResourceT $ zmqInterruptibleSource s_interrupted s $= normalisationConduit options config $$ messageSink (ZMQC.zmqSink successSocket []) (ZMQC.zmqSink failureSocket []) messageCount frequency -{- -- FIXME: This does not work, gives a type error Just testSinkFileName -> ZMQ.withContext $ \ctx -> ZMQ.withSocket ctx ZMQ.Pull $ \s -> do ZMQ.bind s $ printf "tcp://%s:%d" listenHost listenPort- liftIO $ zmqInterruptibleSource s_interrupted s+ verbose' $ printf "Listening on tcp://%s:%d" listenHost listenPort+ runResourceT $ zmqInterruptibleSource s_interrupted s $= normalisationConduit options config- $= mySink+ $= mySink messageCount frequency $$ sinkFile testSinkFileName--} -------------------------------------------------------------------------------- -- | 'main' starts a TCP server, listening to incoming data and connecting to TCP servers downstream to -- for the pipeline.
hnormalise.cabal view
@@ -1,5 +1,5 @@ name: hnormalise-version: 0.4.7.0+version: 0.4.8.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@@ -39,6 +39,7 @@ , HNormalise.Torque.Json , HNormalise.Torque.Parser , HNormalise.Verbose+ other-modules: Paths_hnormalise build-depends: base >= 4.7 && < 5 , aeson , aeson-pretty@@ -49,14 +50,16 @@ , directory , ip , permute+ , resourcet , text , time , unordered-containers , yaml default-language: Haskell2010 -executable hnormalise- hs-source-dirs: app++executable hnormalise-0mq+ hs-source-dirs: 0mq main-is: Main.hs ghc-options: -threaded -rtsopts -with-rtsopts=-N -funbox-strict-fields -optc-O2 build-depends: base@@ -72,7 +75,9 @@ , containers , directory , ip+ , lifted-base , monad-loops+ , monad-par , mtl , optparse-applicative , resourcet@@ -83,14 +88,10 @@ , yaml , zeromq4-haskell , zeromq4-conduit- , lifted-base- , monad-control- , transformers-base default-language: Haskell2010 --executable hnormalise-0mq- hs-source-dirs: 0mq+executable hnormalise+ hs-source-dirs: app main-is: Main.hs ghc-options: -threaded -rtsopts -with-rtsopts=-N -funbox-strict-fields -optc-O2 build-depends: base@@ -117,6 +118,9 @@ , yaml , zeromq4-haskell , zeromq4-conduit+ , lifted-base+ , monad-control+ , transformers-base default-language: Haskell2010 @@ -154,7 +158,6 @@ , aeson , attoparsec , criterion- , deepseq , hnormalise , random , text
src/HNormalise.hs view
@@ -44,13 +44,12 @@ import Control.Applicative ((<|>)) import Data.Aeson (ToJSON) import qualified Data.Aeson as Aeson-import Data.Aeson.Text (encodeToLazyText) import Data.Attoparsec.Combinator (lookAhead, manyTill) import Data.Attoparsec.Text import qualified Data.ByteString.Char8 as SBS import qualified Data.ByteString.Lazy.Char8 as BS import Data.Text (Text, empty)-import Data.Text.Encoding (encodeUtf8)+import Data.Text.Encoding (encodeUtf8, decodeUtf8) import Data.Text.Lazy (toStrict) --------------------------------------------------------------------------------@@ -86,17 +85,18 @@ -- | The 'normaliseText' function converts a 'Text' to a normalised message or keeps the original (in 'ByteString') -- format if the conversion fails normaliseText :: Maybe [(Text, Text)] -- ^ Output fields- -> Text -- ^ Input+ -> SBS.ByteString -- ^ Input -> Normalised -- ^ Transformed or Original result normaliseText fs logLine =- case parse (parseRsyslogLogstashString fs) logLine of+ let !p = parse (parseRsyslogLogstashString fs) $ decodeUtf8 logLine+ in case p of Done _ r -> Transformed r Partial c -> case c empty of Done _ r -> Transformed r _ -> original _ -> original where- original = Original $ encodeUtf8 logLine+ original = Original logLine -------------------------------------------------------------------------------- -- | The 'convertMessage' function transforms the actual message to a 'Maybe' 'ParseResult'. If parsing fails,
src/HNormalise/Torque/Parser.hs view
@@ -42,6 +42,7 @@ import Control.Monad (join) import Data.Attoparsec.Combinator (lookAhead, manyTill) import Data.Attoparsec.Text+import qualified Data.Attoparsec.Text as AT import Data.Char (isDigit, isSpace) import Data.List (concatMap, groupBy, sort) import qualified Data.Map as M@@ -136,7 +137,7 @@ ppn <- maybeOption $ char ':' *> string "ppn=" *> decimal return $ TSN TorqueJobShortNode { number = number, ppn = ppn } else TFN <$> sepBy (do- fqdn <- Data.Attoparsec.Text.takeWhile (\c -> c /= ':' && c /= ' ')+ fqdn <- AT.takeWhile (\c -> c /= ':' && c /= ' ') ppn <- maybeOption $ char ':' *> kvNumParser "ppn" return TorqueJobFQNode { name = fqdn, ppn = ppn}) (char '+') @@ -166,8 +167,59 @@ -- the guesses as to the most common ordering and the mandatory fields is based on 5 years of log data from Torque -- 4.x to 6.0 parseTorqueResourceRequest :: Parser TorqueResourceRequest-parseTorqueResourceRequest =- permute $ TorqueResourceRequest+parseTorqueResourceRequest = try+ (do+ mem <- maybeOption (skipSpace *> string "Resource_List.mem=" *> parseTorqueMemory)+ advres <- maybeOption (skipSpace *> kvTextParser "Resource_List.advres")+ naccesspolicy <- maybeOption $ skipSpace *> kvTextParser "Resource_List.naccesspolicy"+ ncpus <- maybeOption $ skipSpace *> kvNumParser "Resource_List.ncpus"+ cputime <- maybeOption $ skipSpace *> string "Resource_List.cput=" *> parseTorqueWalltime+ prologue <- maybeOption $ skipSpace *> kvTextParser "Resource_List.prologue"+ epilogue <- maybeOption $ skipSpace *> kvTextParser "Resource_List.epilogue"+ neednodes <- maybeOption $ skipSpace *> string "Resource_List.neednodes=" *> parseTorqueResourceNodeList+ nice <- maybeOption $ skipSpace *> kvNumParser "Resource_List.nice"+ nodect <- skipSpace *> kvNumParser "Resource_List.nodect"+ nodes <- skipSpace *> string "Resource_List.nodes=" *> parseTorqueResourceNodeList+ select <- maybeOption $ skipSpace *> kvTextParser "Resource_List.select"+ qos <- maybeOption $ skipSpace *> kvTextParser "Resource_List.qos"+ other <- maybeOption $ skipSpace *> kvTextParser "Resource_List.other"+ feature <- maybeOption $ skipSpace *> kvTextParser "Resource_List.feature"+ host <- maybeOption $ skipSpace *> kvTextParser "Resource_List.host"+ procs <- maybeOption $ skipSpace *> kvTextParser "Resource_List.procs"+ nodeset <- maybeOption $ skipSpace *> kvTextParser "Resource_List.nodeset"+ tpn <- maybeOption $ skipSpace *> kvTextParser "Resource_List.tpn"+ pmem <- maybeOption $ skipSpace *> string "Resource_List.pmem=" *> parseTorqueMemory+ vmem <- maybeOption $ skipSpace *> string "Resource_List.vmem=" *> parseTorqueMemory+ pvmem <- maybeOption $ skipSpace *> string "Resource_List.pvmem=" *> parseTorqueMemory+ mppmem <- maybeOption $ skipSpace *> string "Resource_List.mppmem=" *> parseTorqueMemory+ walltime <- skipSpace *> string "Resource_List.walltime=" *> parseTorqueWalltime+ return TorqueResourceRequest+ { mem = mem+ , advres = advres+ , naccesspolicy = naccesspolicy+ , ncpus = ncpus+ , cputime = cputime+ , prologue = prologue+ , epilogue = epilogue+ , neednodes = neednodes+ , nice = nice+ , nodeCount = nodect+ , nodes = nodes+ , select = select+ , qos = qos+ , other = other+ , feature = feature+ , host = host+ , procs = procs+ , nodeset = nodeset+ , tpn = tpn+ , pmem = pmem+ , vmem = vmem+ , pvmem = pvmem+ , mppmem = mppmem+ , walltime = walltime+ })+ <|> ( permute $ TorqueResourceRequest <$?> (Nothing, Just `fmap` (skipSpace *> string "Resource_List.mem=" *> parseTorqueMemory)) <|?> (Nothing, Just `fmap` (skipSpace *> kvTextParser "Resource_List.advres")) <|?> (Nothing, Just `fmap` (skipSpace *> kvTextParser "Resource_List.naccesspolicy"))@@ -192,6 +244,7 @@ <|?> (Nothing, Just `fmap` (skipSpace *> string "Resource_List.pvmem=" *> parseTorqueMemory)) <|?> (Nothing, Just `fmap` (skipSpace *> string "Resource_List.mppmem=" *> parseTorqueMemory)) <||> skipSpace *> string "Resource_List.walltime=" *> parseTorqueWalltime+ ) -------------------------------------------------------------------------------- -- | 'parseTorqueCpuTime' parses the cpu time spent@@ -238,7 +291,7 @@ parseTorqueHostList = do string "exec_host=" hosts <- flip sepBy (char '+') $ do- fqdn <- Data.Attoparsec.Text.takeWhile (/= '/')+ fqdn <- AT.takeWhile (/= '/') char '/' cores <- parseCores return TorqueExecHost { name = fqdn, cores = cores}
test/Bench.hs view
@@ -44,6 +44,8 @@ import qualified Data.Aeson as Aeson import qualified Data.Attoparsec.Text as AT import Data.Text (Text)+import Data.Text.Encoding (encodeUtf8)+ -------------------------------------------------------------------------------- import qualified HNormalise.Lmod.Parser as LmodP import qualified HNormalise.Shorewall.Parser as ShorewallP@@ -52,13 +54,8 @@ import qualified HNormalise as H ---------------------------------------------------------------------------------instance NFData H.Normalised where- rnf (H.Transformed s) = rnf s- rnf (H.Original s) = rnf s-----------------------------------------------------------------------------------torqueJobExitInput1 = "torque: 04/05/2017 13:06:53;E;45.master23.banette.gent.vsc;user=vsc40075 group=vsc40075 jobname=STDIN queue=short ctime=1491390300 qtime=1491390300 etime=1491390300 start=1491390307 owner=vsc40075@gligar01.gligar.gent.vsc exec_host=node2801.banette.gent.vsc/0-1+node2803.banette.gent.vsc/0-1 Resource_List.nodes=node2801.banette.gent.vsc:ppn=2+node2803.banette.gent.vsc:ppn=2 Resource_List.vmem=1gb Resource_List.nodect=2 Resource_List.neednodes=node2801.banette.gent.vsc:ppn=2+node2803.banette.gent.vsc:ppn=2 Resource_List.nice=0 Resource_List.walltime=01:00:00 session=15273 total_execution_slots=4 unique_node_count=2 end=1491390413 Exit_status=0 resources_used.cput=0 resources_used.energy_used=0 resources_used.mem=55048kb resources_used.vmem=92488kb resources_used.walltime=00:01:44" :: Text-torqueJobExitInput2 = "torque: 04/05/2017 13:06:53;E;45.master23.banette.gent.vsc;user=vsc40075 group=vsc40075 jobname=STDIN queue=short ctime=1491390300 qtime=1491390300 etime=1491390300 start=1491390307 owner=vsc40075@gligar01.gligar.gent.vsc exec_host=node2801.banette.gent.vsc/0-1+node2803.banette.gent.vsc/0-1 Resource_List.nodes=2 Resource_List.vmem=1gb Resource_List.nodect=2 Resource_List.neednodes=2 Resource_List.nice=0 Resource_List.walltime=01:00:00 session=15273 total_execution_slots=4 unique_node_count=2 end=1491390413 Exit_status=0 resources_used.cput=0 resources_used.energy_used=0 resources_used.mem=55048kb resources_used.vmem=92488kb resources_used.walltime=00:01:44" :: Text+torqueJobExitInput1 = "torque: 04/05/2017 13:06:53;E;45.master23.banette.gent.vsc;user=vsc40075 group=vsc40075 jobname=STDIN queue=short ctime=1491390300 qtime=1491390300 etime=1491390300 start=1491390307 owner=vsc40075@gligar01.gligar.gent.vsc exec_host=node2801.banette.gent.vsc/0-1+node2803.banette.gent.vsc/0-1 Resource_List.nodes=node2801.banette.gent.vsc:ppn=2+node2803.banette.gent.vsc:ppn=2 Resource_List.vmem=1gb Resource_List.nodect=2 Resource_List.neednodes=node2801.banette.gent.vsc:ppn=2+node2803.banette.gent.vsc:ppn=2 Resource_List.nice=0 Resource_List.walltime=01:00:00 session=15273 total_execution_slots=4 unique_node_count=2 end=1491390413 Exit_status=0 resources_used.cput=0 resources_used.energy_used=0 resources_used.mem=55048kb resources_used.vmem=92488kb resources_used.walltime=00:01:44"+torqueJobExitInput2 = "torque: 04/05/2017 13:06:53;E;45.master23.banette.gent.vsc;user=vsc40075 group=vsc40075 jobname=STDIN queue=short ctime=1491390300 qtime=1491390300 etime=1491390300 start=1491390307 owner=vsc40075@gligar01.gligar.gent.vsc exec_host=node2801.banette.gent.vsc/0-1+node2803.banette.gent.vsc/0-1 Resource_List.nodes=2 Resource_List.vmem=1gb Resource_List.nodect=2 Resource_List.neednodes=2 Resource_List.nice=0 Resource_List.walltime=01:00:00 session=15273 total_execution_slots=4 unique_node_count=2 end=1491390413 Exit_status=0 resources_used.cput=0 resources_used.energy_used=0 resources_used.mem=55048kb resources_used.vmem=92488kb resources_used.walltime=00:01:44" torqueJobExitFailInput1 = "torque: 04/05/2017 13:06:53;E;45.master23.banette.gent.vsc;user=vsc40075 group=vsc40075 jobname=STDIN queue=short HUPPEL" @@ -66,15 +63,15 @@ torqueJobDeleteInput = "torque: 06/28/2017 15:44:02;D;81.master23.banette.gent.vsc;requestor=vsc40075@gligar02.gligar.gent.vsc" torqueJobStartInput = "torque: 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+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"+fullLmodInput = encodeUtf8 "<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" shorewallTCPInput = "kernel:: Shorewall:ext2fw:REJECT:IN=em3 OUT= MAC=aa:aa:bb:ff:88:bc:bc:15:80:8b:f8:f8:80:00 SRC=78.0.0.1 DST=150.0.0.1 LEN=52 TOS=0x00 PREC=0x00 TTL=117 ID=7564 DF PROTO=TCP SPT=60048 DPT=22 WINDOW=65535 RES=0x00 SYN URGP=0" shorewallUDPInput = "kernel:: Shorewall:ipmi2int:REJECT:IN=em4 OUT=em1 SRC=10.0.0.2 DST=10.0.0.1 LEN=57 TOS=0x00 PREC=0x00 TTL=63 ID=62392 PROTO=UDP SPT=57002 DPT=53 LEN=37" shorewallICMPInput = "kernel:: Shorewall:ipmi2ext:REJECT:IN=em4 OUT=em3 SRC=10.0.0.2 DST=10.0.0.1 LEN=28 TOS=0x00 PREC=0x00 TTL=63 ID=36216 PROTO=ICMP TYPE=8 CODE=0 ID=0 SEQ=1421" -snoopyInput = "snoopy[27316]:: [uid:110 sid:9379 tty:(none) cwd:/ filename:/usr/lib64/nagios/plugins/hpc/check_ifutil.pl]: /usr/lib64/nagios/plugins/hpc/check_ifutil.pl -i em1.295 -w 90 -c 95 -p -b 10000m" :: Text-fullSnoopyInput = "<86>1 2015-12-20T09:59:40.844711+01:00 gligar03 snoopy[46513]: - snoopy[46513]:: [uid:2540337 sid:19403 tty:ERROR(ttyname_r->EUNKNOWN) cwd:/vscmnt/gent_vulpix/_/user/home/gent/vsc403/vsc40337/UCS_LABELLED_NEW/20000_to_30000 filename:/usr/bin/qsub]: qsub -l walltime=72:00:00 job7_21293_30000_doit" :: Text+snoopyInput = "snoopy[27316]:: [uid:110 sid:9379 tty:(none) cwd:/ filename:/usr/lib64/nagios/plugins/hpc/check_ifutil.pl]: /usr/lib64/nagios/plugins/hpc/check_ifutil.pl -i em1.295 -w 90 -c 95 -p -b 10000m"+fullSnoopyInput = encodeUtf8 "<86>1 2015-12-20T09:59:40.844711+01:00 gligar03 snoopy[46513]: - snoopy[46513]:: [uid:2540337 sid:19403 tty:ERROR(ttyname_r->EUNKNOWN) cwd:/vscmnt/gent_vulpix/_/user/home/gent/vsc403/vsc40337/UCS_LABELLED_NEW/20000_to_30000 filename:/usr/bin/qsub]: qsub -l walltime=72:00:00 job7_21293_30000_doit" --------------------------------------------------------------------------------