marquise (empty) → 4.0.0
raw patch · 18 files changed
+1909/−0 lines, 18 filesdep +asyncdep +attoparsecdep +basebuild-type:Customsetup-changed
Dependencies added: async, attoparsec, base, binary, bytestring, containers, cryptohash, data-binary-ieee754, directory, either, errors, fast-logger, filepath, hashable, hslogger, hspec, lifted-async, marquise, mmorph, monad-control, monad-logger, mtl, old-locale, optparse-applicative, packer, pipes, pipes-attoparsec, pipes-bytestring, pipes-group, semigroups, siphash, text, time, transformers, transformers-base, unix, unordered-containers, vaultaire-common, zeromq4-haskell
Files
- LICENCE +32/−0
- Setup.hs +71/−0
- lib/Marquise/Classes.hs +77/−0
- lib/Marquise/Client.hs +59/−0
- lib/Marquise/Client/Core.hs +311/−0
- lib/Marquise/IO.hs +18/−0
- lib/Marquise/IO/Connection.hs +69/−0
- lib/Marquise/IO/Contents.hs +22/−0
- lib/Marquise/IO/FFI.hsc +58/−0
- lib/Marquise/IO/Reader.hs +25/−0
- lib/Marquise/IO/SpoolFile.hs +142/−0
- lib/Marquise/IO/Writer.hs +36/−0
- lib/Marquise/Server.hs +308/−0
- lib/Marquise/Types.hs +47/−0
- marquise.cabal +161/−0
- src/DataProgram.hs +308/−0
- src/MarquiseDaemon.hs +114/−0
- tests/Spool.hs +51/−0
+ LICENCE view
@@ -0,0 +1,32 @@+A data vault for time series metrics data.++Copyright © 2013 Anchor Systems, Pty Ltd+All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions+are met:++ 1. Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ 2. 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.+ + 3. Neither the name of the project nor the names of its 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.
+ Setup.hs view
@@ -0,0 +1,71 @@+--+-- Common Haskell build tools+--+-- Copyright © 2013-2014 Operational Dynamics Consulting, Pty Ltd and Others+--+-- The code in this file, and the program it is a part of, is+-- made available to you by its authors as open source software:+-- you can redistribute it and/or modify it under the terms of+-- the 3-clause BSD licence.+--++import Data.Char (toUpper)+import Distribution.PackageDescription (PackageDescription (..))+import Distribution.Simple+import Distribution.Simple.LocalBuildInfo (LocalBuildInfo)+import Distribution.Simple.Setup (ConfigFlags)+import Distribution.System (OS (..), buildOS)+import Distribution.Text (display)+import System.IO (Handle, IOMode (..), hPutStrLn, withFile)++main :: IO ()+main = defaultMainWithHooks $ simpleUserHooks {+ postConf = configure+ }++{-+ Simple detection of which operating system we're building on;+ there's no need to link the Cabal logic into our library, so+ we output to a .hs file in src/+-}++configure :: Args -> ConfigFlags -> PackageDescription -> LocalBuildInfo -> IO ()+configure _ _ p _ = do+ withFile "src/Package.hs" WriteMode (\h -> do+ outputModuleHeader h+ discoverOperatingSystem h+ discoverProgramVersion h p)++ return ()++outputModuleHeader :: Handle -> IO ()+outputModuleHeader h = do+ hPutStrLn h "module Package where"++discoverOperatingSystem :: Handle -> IO ()+discoverOperatingSystem h = do+ hPutStrLn h "build :: String"+ hPutStrLn h ("build = \"" ++ s ++ "\"")+ where+ o = buildOS++ s = case o of+ Linux -> "Linux"+ OSX -> "Mac OS X"+ Windows -> "Windows"+ _ -> up o++ up x = map toUpper (show x)++discoverProgramVersion :: Handle -> PackageDescription -> IO ()+discoverProgramVersion h p = do+ hPutStrLn h "package :: String"+ hPutStrLn h ("package = \"" ++ n ++ "\"")+ hPutStrLn h "version :: String"+ hPutStrLn h ("version = \"" ++ s ++ "\"")+ where+ i = package p+ (PackageName n) = pkgName i+ v = pkgVersion i+ s = display v+
+ lib/Marquise/Classes.hs view
@@ -0,0 +1,77 @@+--+-- Copyright © 2013-2014 Anchor Systems, Pty Ltd and Others+--+-- The code in this file, and the program it is a part of, is+-- made available to you by its authors as open source software:+-- you can redistribute it and/or modify it under the terms of+-- the 3-clause BSD licence.+--++{-# LANGUAGE FunctionalDependencies #-}+{-# LANGUAGE MultiParamTypeClasses #-}++module Marquise.Classes+(+ MarquiseWriterMonad(..),+ MarquiseSpoolFileMonad(..),+ MarquiseReaderMonad(..),+ MarquiseContentsMonad(..),+) where++import Data.ByteString (ByteString)+import qualified Data.ByteString.Lazy as LB++import Marquise.Types+import Vaultaire.Types++-- | This class is for convenience of testing. It encapsulates all IO+-- interaction that the client and server will do.+class Monad m => MarquiseSpoolFileMonad m where+ randomSpoolFiles :: SpoolName -> m SpoolFiles++ createDirectories :: SpoolName -> m ()++ -- | Append to the spool file for points, i.e. data.+ --+ -- This append does not imply that the given data is synced to disk, just+ -- that it is queued to do so. This assumes no state, so any file handles+ -- must be stashed globally or re-opened and closed.+ appendPoints :: SpoolFiles -> ByteString -> m ()++ -- | Append to the spool file for contents updates, i.e. metadata.+ appendContents :: SpoolFiles -> ByteString -> m ()++ -- | Return an lazy bytestring and an IO action to signify that the burst+ -- has been completely sent.+ --+ -- May block until something is actually spooled up.+ nextPoints :: SpoolName -> m (Maybe (LB.ByteString, m ()))+ nextContents :: SpoolName -> m (Maybe (LB.ByteString, m ()))++ -- | Close any open handles and flush all previously appended datum to disk+ close :: SpoolFiles -> m ()++-- | Monad encapsulating writer operations. Note there is an instance for IO+-- in IO/Writer.hs+class Monad m => MarquiseWriterMonad m where+ -- | Send bytes upstream.+ -- returns: - result when an ACK is received.+ -- - error when an exception happens.+ transmitBytes :: String -- ^ Broker address+ -> Origin -- ^ Origin+ -> ByteString -- ^ Bytes to send+ -> m ()++-- | Monad encapsulating reader operations. Note there is an instance for+-- IO SocketState in IO/Contents.hs+class Monad m => MarquiseContentsMonad m connection | m -> connection where+ sendContentsRequest :: ContentsOperation -> Origin -> connection -> m ()+ recvContentsResponse :: connection -> m ContentsResponse+ withContentsConnection :: String -> (connection -> m a) -> m a++-- | Monad encapsulating reader operations. Note there is an instance for+-- IO SocketState in IO/Reader.hs+class Monad m => MarquiseReaderMonad m connection | m -> connection where+ sendReaderRequest :: ReadRequest -> Origin -> connection -> m ()+ recvReaderResponse :: connection -> m ReadStream+ withReaderConnection :: String -> (connection -> m a) -> m a
+ lib/Marquise/Client.hs view
@@ -0,0 +1,59 @@+{-# LANGUAGE MultiParamTypeClasses #-}++-- Hide warnings for the deprecated ErrorT transformer:+{-# OPTIONS_GHC -fno-warn-warnings-deprecations #-}++module Marquise.Client (+ -- | * Utility functions+ -- Note: You may read MarquiseSpoolFileMonad m as IO.+ hashIdentifier+ , makeSpoolName+ , makeOrigin+ , newRandomPointsSpoolFile+ , newRandomContentsSpoolFile++ -- | * Contents daemon requests+ , withContentsConnection+ , requestUnique+ , makeSourceDict+ , updateSourceDict+ , removeSourceDict+ , enumerateOrigin++ -- | * Queuing data to be sent to vaultaire+ , createSpoolFiles+ , queueSimple+ , queueExtended+ , queueSourceDictUpdate+ , flush++ -- | Reading from Vaultaire+ , withReaderConnection+ , readExtended+ , readExtendedPoints+ , readSimple+ , readSimplePoints+ , decodeExtended+ , decodeSimple++ -- * Types+ , SourceDict+ , SpoolName+ , SpoolFiles+ , Address(..)+ , Origin(..)+ , TimeStamp(..)+ , ExtendedBurst(..)+ , ExtendedPoint(..)+ , SimpleBurst(..)+ , SimplePoint(..)+ , SocketState(..)+ ) where++import Marquise.Classes+import Marquise.Client.Core+import Marquise.IO ()+import Marquise.IO.Connection+import Marquise.IO.SpoolFile+import Marquise.Types+import Vaultaire.Types
+ lib/Marquise/Client/Core.hs view
@@ -0,0 +1,311 @@+--+-- Copyright © 2013-2014 Anchor Systems, Pty Ltd and Others+--+-- The code in this file, and the program it is a part of, is+-- made available to you by its authors as open source software:+-- you can redistribute it and/or modify it under the terms of+-- the 3-clause BSD licence.+--++-- | client interface for sending data to the vault.+--+-- This module provides functions for preparing and queuing points to be sent+-- by a server to the vault.+--+-- If you call close, you can be assured that your data is safe and will at+-- some point end up in the data vault (excluding local disk failure). This+-- assumption is based on a functional marquise daemon with connectivity+-- eventually running within your namespace.+--+-- We provide no way to *absolutely* ensure that a point is currently written+-- to the vault. Such a guarantee would require blocking and complex queuing,+-- or observing various underlying mechanisms that should ideally remain+-- abstract.+--++{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- Hide warnings for the deprecated ErrorT transformer:+{-# OPTIONS_GHC -fno-warn-warnings-deprecations #-}++module Marquise.Client.Core where++import Control.Applicative+import qualified Control.Exception as E+import Control.Monad.Error+import Crypto.MAC.SipHash+import Data.Bits+import Data.ByteString (ByteString)+import qualified Data.ByteString as BS+import Data.Char (isAlphaNum)+import Data.Packer+import Data.Word (Word64)+import Pipes++import Marquise.Classes+import Marquise.Types+import Vaultaire.Types+++-- | Create a SpoolName. Only alphanumeric characters are allowed, max length+-- is 32 characters.+makeSpoolName :: Monad m => String -> m SpoolName+makeSpoolName s+ | any (not . isAlphaNum) s = E.throw $ MarquiseException s+ | otherwise = return (SpoolName s)++-- | Create a name in the spool. Only alphanumeric characters are allowed, max length+-- is 32 characters.+createSpoolFiles :: MarquiseSpoolFileMonad m+ => String+ -> m SpoolFiles+createSpoolFiles s = do+ n <- makeSpoolName s+ createDirectories n+ randomSpoolFiles n++-- | Deterministically convert a ByteString to an Address by taking the+-- most significant 63 bytes of its SipHash-2-4[0] with a zero key. The+-- LSB of the resulting 64-bit value is not part of the unique portion+-- of the address; it is set when queueing writes, depending on the+-- point type (simple or extended) being written.+--+-- [0] https://131002.net/siphash/+hashIdentifier :: ByteString -> Address+hashIdentifier = Address . (`clearBit` 0) . unSipHash . hash iv+ where+ iv = SipKey 0 0+ unSipHash (SipHash h) = h :: Word64++-- | Generate an un-used Address. You will need to store this for later re-use.+requestUnique :: MarquiseContentsMonad m conn+ => Origin+ -> conn+ -> m Address+requestUnique origin conn = do+ sendContentsRequest GenerateNewAddress origin conn+ response <- recvContentsResponse conn+ case response of+ RandomAddress addr -> return addr+ _ -> error "requestUnique: Invalid response"++-- | Set the key,value tags as metadata on the given Address.+updateSourceDict :: MarquiseContentsMonad m conn+ => Address+ -> SourceDict+ -> Origin+ -> conn+ -> m ()+updateSourceDict addr source_dict origin conn = do+ sendContentsRequest (UpdateSourceTag addr source_dict) origin conn+ response <- recvContentsResponse conn+ case response of+ UpdateSuccess -> return ()+ _ -> error "requestSourceDictUpdate: Invalid response"++-- | Remove the supplied key,value tags from metadata on the Address, if present.+removeSourceDict :: MarquiseContentsMonad m conn+ => Address+ -> SourceDict+ -> Origin+ -> conn+ -> m ()+removeSourceDict addr source_dict origin conn = do+ sendContentsRequest (RemoveSourceTag addr source_dict) origin conn+ response <- recvContentsResponse conn+ case response of+ RemoveSuccess -> return ()+ _ -> error "requestSourceDictRemoval: Invalid response"++-- | Stream read every Address associated with the given Origin+enumerateOrigin :: MarquiseContentsMonad m conn+ => Origin+ -> conn+ -> Producer (Address, SourceDict) m ()+enumerateOrigin origin conn = do+ lift $ sendContentsRequest ContentsListRequest origin conn+ loop+ where+ loop = do+ resp <- lift $ recvContentsResponse conn+ case resp of+ ContentsListEntry addr dict -> do+ yield (addr, dict)+ loop+ EndOfContentsList -> return ()+ _ -> error "enumerateOrigin loop: Invalid response"++-- | Stream read every SimpleBurst from the Address between the given times+readSimple :: MarquiseReaderMonad m conn+ => Address+ -> TimeStamp+ -> TimeStamp+ -> Origin+ -> conn+ -> Producer' SimpleBurst m ()+readSimple addr start end origin conn = do+ lift $ sendReaderRequest (SimpleReadRequest addr start end) origin conn+ loop+ where+ loop = do+ response <- lift $ recvReaderResponse conn+ case response of+ SimpleStream burst ->+ yield burst >> loop+ EndOfStream ->+ return ()+ InvalidReadOrigin ->+ error "readSimple loop: Invalid origin"+ _ ->+ error "readSimple loop: Invalid response"++-- | Like @readSimple@ but also decodes the points.+--+readSimplePoints+ :: MarquiseReaderMonad m conn+ => Address+ -> TimeStamp+ -> TimeStamp+ -> Origin+ -> conn+ -> Producer' SimplePoint m ()+readSimplePoints addr start end origin conn+ = for (readSimple addr start end origin conn >-> decodeSimple) yield++-- | Stream read every ExtendedBurst from the Address between the given times+readExtended :: MarquiseReaderMonad m conn+ => Address+ -> TimeStamp+ -> TimeStamp+ -> Origin+ -> conn+ -> Producer' ExtendedBurst m ()+readExtended addr start end origin conn = do+ lift $ sendReaderRequest (ExtendedReadRequest addr start end) origin conn+ loop+ where+ loop = do+ response <- lift $ recvReaderResponse conn+ case response of+ ExtendedStream burst ->+ yield burst >> loop+ EndOfStream ->+ return ()+ _ ->+ error "readExtended loop: Invalid response"++-- | Like @readExtended@ but also decodes the points.+--+readExtendedPoints+ :: MarquiseReaderMonad m conn+ => Address+ -> TimeStamp+ -> TimeStamp+ -> Origin+ -> conn+ -> Producer' ExtendedPoint m ()+readExtendedPoints addr start end origin conn+ = for (readExtended addr start end origin conn >-> decodeExtended) yield++-- | Stream converts raw SimpleBursts into SimplePoints+decodeSimple :: Monad m => Pipe SimpleBurst SimplePoint m ()+decodeSimple = forever (unSimpleBurst <$> await >>= emitFrom 0)+ where+ emitFrom os chunk+ | os >= BS.length chunk = return ()+ | otherwise = do+ yield $ flip runUnpacking chunk $ do+ unpackSetPosition os+ addr <- Address <$> getWord64LE+ time <- TimeStamp <$> getWord64LE+ payload <- getWord64LE+ return $ SimplePoint addr time payload++ emitFrom (os + 24) chunk+++-- | Stream converts raw ExtendedBursts into ExtendedPoints+decodeExtended :: Monad m => Pipe ExtendedBurst ExtendedPoint m ()+decodeExtended = forever (unExtendedBurst <$> await >>= emitFrom 0)+ where+ emitFrom os chunk+ | os >= BS.length chunk = return ()+ | otherwise = do+ let result = runUnpacking (unpack os) chunk+ yield result++ let size = BS.length (extendedPayload result) + 24+ emitFrom (os + size) chunk++ unpack os = do+ unpackSetPosition os+ addr <- Address <$> getWord64LE+ time <- TimeStamp <$> getWord64LE+ len <- fromIntegral <$> getWord64LE+ payload <- if len == 0+ then return BS.empty+ else getBytes len++ return $ ExtendedPoint addr time payload++-- | Send a "simple" data point. Interpretation of this point, e.g.+-- float/signed is up to you, but it must be sent in the form of a Word64.+-- Clears the least-significant bit of the address to indicate that this+-- is a simple datapoint.+queueSimple+ :: MarquiseSpoolFileMonad m+ => SpoolFiles+ -> Address+ -> TimeStamp+ -> Word64+ -> m ()+queueSimple sfs (Address ad) (TimeStamp ts) w = appendPoints sfs bytes+ where+ bytes = runPacking 24 $ do+ putWord64LE (ad `clearBit` 0)+ putWord64LE ts+ putWord64LE w++-- | Send an "extended" data point. Again, representation is up to you.+-- Sets the least-significant bit of the address to indicate that this is+-- an extended data point.+queueExtended+ :: MarquiseSpoolFileMonad m+ => SpoolFiles+ -> Address+ -> TimeStamp+ -> ByteString+ -> m ()+queueExtended sfs (Address ad) (TimeStamp ts) bs = appendPoints sfs bytes+ where+ len = BS.length bs+ bytes = runPacking (24 + len) $ do+ putWord64LE (ad `setBit` 0)+ putWord64LE ts+ putWord64LE $ fromIntegral len+ putBytes bs++-- | Updates the SourceDict at addr with source_dict+queueSourceDictUpdate+ :: MarquiseSpoolFileMonad m+ => SpoolFiles+ -> Address+ -> SourceDict+ -> m ()+queueSourceDictUpdate sfs (Address addr) source_dict = appendContents sfs bytes+ where+ source_dict_bytes = toWire source_dict+ source_dict_len = BS.length source_dict_bytes+ bytes = runPacking (source_dict_len + 16) $ do+ putWord64LE addr+ putWord64LE (fromIntegral source_dict_len)+ putBytes source_dict_bytes++-- | Ensure that all sent points have hit the local disk.+flush+ :: MarquiseSpoolFileMonad m+ => SpoolFiles+ -> m ()+flush = close
+ lib/Marquise/IO.hs view
@@ -0,0 +1,18 @@+-- Data vault for metrics+--+-- Copyright © 2013-2014 Anchor Systems, Pty Ltd and Others+--+-- The code in this file, and the program it is a part of, is+-- made available to you by its authors as open source software:+-- you can redistribute it and/or modify it under the terms of+-- the 3-clause BSD licence.+--++-- | IO interactions for Marquise+module Marquise.IO+where++import Marquise.IO.Contents ()+import Marquise.IO.Reader ()+import Marquise.IO.SpoolFile ()+import Marquise.IO.Writer ()
+ lib/Marquise/IO/Connection.hs view
@@ -0,0 +1,69 @@+--+-- Copyright © 2013-2014 Anchor Systems, Pty Ltd and Others+--+-- The code in this file, and the program it is a part of, is+-- made available to you by its authors as open source software:+-- you can redistribute it and/or modify it under the terms of+-- the 3-clause BSD licence.+--++{-# LANGUAGE OverloadedStrings #-}+{-# OPTIONS_GHC -fno-warn-orphans #-}+-- Hide warnings for the deprecated ErrorT transformer:+{-# OPTIONS_GHC -fno-warn-warnings-deprecations #-}++module Marquise.IO.Connection+( withConnection,+ send,+ recv,+ SocketState(..),+) where++import qualified Control.Exception as E+import Data.List.NonEmpty (fromList)+import System.ZMQ4 (Dealer (..), Event (..), Poll (..), Socket)+import qualified System.ZMQ4 as Z++import Marquise.Types+import Vaultaire.Types++-- | Wrapped ZMQ4 Socket + broker/IP+data SocketState = SocketState (Socket Dealer) String++-- | Performs operation f through broker.+withConnection :: String -> (SocketState -> IO a) -> IO a+withConnection broker f =+ Z.withContext $ \ctx ->+ Z.withSocket ctx Dealer $ \s -> do+ Z.connect s broker+ f (SocketState s broker)++send :: WireFormat request+ => request+ -> Origin+ -> SocketState+ -> IO ()+send request (Origin origin) (SocketState sock _)+ = Z.sendMulti sock (fromList [origin, toWire request])++recv :: WireFormat response+ => SocketState+ -> IO response+recv (SocketState sock endpoint) = do+ poll_result <- Z.poll timeout [Sock sock [In] Nothing]+ case poll_result of+ [[In]] -> do+ resp <- Z.receiveMulti sock+ case resp of+ [msg] -> either E.throw return $ fromWire msg+ [] -> E.throw $ MarquiseException "expected one message, received none"+ _ -> E.throw $ MarquiseException "expected one message, received multiple"+ [[]] -> do+ -- Timeout, reconnect the socket so that we can be sure that a late+ -- response on the current connection isn't confused with a+ -- response to a later request.+ Z.disconnect sock endpoint+ Z.connect sock endpoint+ E.throw $ MarquiseException "timeout"+ _ -> E.throw $ MarquiseException "Marquise.IO.Connection.recv: impossible"+ where timeout = 30 * 60 * 1000 -- milliseconds, 30m
+ lib/Marquise/IO/Contents.hs view
@@ -0,0 +1,22 @@+--+-- Copyright © 2013-2014 Anchor Systems, Pty Ltd and Others+--+-- The code in this file, and the program it is a part of, is+-- made available to you by its authors as open source software:+-- you can redistribute it and/or modify it under the terms of+-- the 3-clause BSD licence.+--++{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# OPTIONS_GHC -fno-warn-orphans #-}++module Marquise.IO.Contents where++import Marquise.Classes+import Marquise.IO.Connection++instance MarquiseContentsMonad IO SocketState where+ withContentsConnection broker = withConnection ("tcp://" ++ broker ++ ":5580")+ sendContentsRequest = send+ recvContentsResponse = recv
+ lib/Marquise/IO/FFI.hsc view
@@ -0,0 +1,58 @@+--+-- Copyright © 2013-2014 Anchor Systems, Pty Ltd and Others+--+-- The code in this file, and the program it is a part of, is+-- made available to you by its authors as open source software:+-- you can redistribute it and/or modify it under the terms of+-- the 3-clause BSD licence.+--++#include <sys/file.h>++module Marquise.IO.FFI+(+ tryLock,+ c_sync,+) where++import Control.Exception(onException)+import Data.Bits+import Foreign.C.Error+import Foreign.C.Types+import System.Posix.Files+import System.Posix.IO (openFd, closeFd, defaultFileFlags, OpenMode(..))+import System.Posix.Types+++open :: FilePath -> IO Fd+open path = openFd path ReadOnly (Just stdFileMode) defaultFileFlags+++-- Some locking bits were inspired by Takano Akio's filelock package+tryLock :: FilePath -> IO (Maybe Fd)+tryLock path = do+ fd <- open path+ (`onException` closeFd fd) $ do+ locked <- flock fd+ if locked+ then return $ Just fd+ else closeFd fd >> return Nothing++flock :: Fd -> IO Bool+flock fd@(Fd fd_int) = do+ r <- c_flock fd_int (#{const LOCK_EX} .|. #{const LOCK_NB})+ if r == 0+ then return True -- success+ else do+ errno <- getErrno+ case () of+ _ | errno == eWOULDBLOCK+ -> return False+ | errno == eINTR+ -> flock fd+ | otherwise -> throwErrno "flock"++foreign import ccall "flock"+ c_flock :: CInt -> CInt -> IO CInt++foreign import ccall "unistd.h sync" c_sync :: IO ()
+ lib/Marquise/IO/Reader.hs view
@@ -0,0 +1,25 @@+--+-- Copyright © 2013-2014 Anchor Systems, Pty Ltd and Others+--+-- The code in this file, and the program it is a part of, is+-- made available to you by its authors as open source software:+-- you can redistribute it and/or modify it under the terms of+-- the 3-clause BSD licence.+--++{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# OPTIONS_GHC -fno-warn-orphans #-}++module Marquise.IO.Reader+(+) where++import Marquise.Classes+import Marquise.IO.Connection++instance MarquiseReaderMonad IO SocketState where+ withReaderConnection broker =+ withConnection ("tcp://" ++ broker ++ ":5570")+ sendReaderRequest = send+ recvReaderResponse = recv
+ lib/Marquise/IO/SpoolFile.hs view
@@ -0,0 +1,142 @@+--+-- Copyright © 2013-2014 Anchor Systems, Pty Ltd and Others+--+-- The code in this file, and the program it is a part of, is+-- made available to you by its authors as open source software:+-- you can redistribute it and/or modify it under the terms of+-- the 3-clause BSD licence.+--++{-# OPTIONS_GHC -fno-warn-orphans #-}++module Marquise.IO.SpoolFile+( newRandomPointsSpoolFile+, newRandomContentsSpoolFile+) where++import Control.Applicative+import Control.Concurrent (threadDelay)+import Control.Monad.State+import Data.ByteString (ByteString)+import qualified Data.ByteString as S+import qualified Data.ByteString.Lazy as L+import qualified Data.ByteString.Lazy as LB+import Data.Maybe+import Marquise.Classes+import Marquise.IO.FFI+import Marquise.Types+import System.Directory+import System.FilePath.Posix+import System.IO+import System.IO.Unsafe+import System.Posix.Files+import System.Posix.IO (closeFd)+import System.Posix.Temp+import System.Posix.Types (Fd)++instance MarquiseSpoolFileMonad IO where+ randomSpoolFiles sn =+ SpoolFiles <$> newRandomPointsSpoolFile sn+ <*> newRandomContentsSpoolFile sn++ createDirectories sn =+ mapM_ (createDirectoryIfMissing True . ($sn))+ [ newPointsDir+ , newContentsDir+ , curPointsDir+ , curContentsDir]++ appendPoints spools = doAppend (pointsSpoolFile spools)+ appendContents spools = doAppend (contentsSpoolFile spools)++ nextPoints sn = nextSpoolContents (newPointsDir sn) (curPointsDir sn)+ nextContents sn = nextSpoolContents (newContentsDir sn) (curContentsDir sn)++ close _ = c_sync++newRandomSpoolFile :: FilePath -> IO FilePath+newRandomSpoolFile path = do+ (spool_file, handle) <- mkstemp path+ hClose handle+ return spool_file++-- | Creates and returns a new points spool file from a namespace+newRandomPointsSpoolFile :: SpoolName -> IO FilePath+newRandomPointsSpoolFile = newRandomSpoolFile . newPointsDir++-- | Creates and returns a new contents spool file from a namespace+newRandomContentsSpoolFile :: SpoolName -> IO FilePath+newRandomContentsSpoolFile = newRandomSpoolFile . newContentsDir++-- | Grab the next avaliable spool file, providing that file as a lazy+-- bytestring and an action to close it, wiping the file.+nextSpoolContents :: FilePath -> FilePath -> IO (Maybe (L.ByteString, IO ()))+nextSpoolContents new_dir cur_dir = do+ -- First check for any work already in the work spool dir.+ work <- tryCurDir cur_dir+ case work of+ Nothing ->+ -- No existing work, get some new work out of the spool+ -- directory then.+ rotate new_dir cur_dir >> return Nothing+ Just (fp, lock_fd) -> do+ threadDelay 100000 -- Ensure that any slow writes are done+ contents <- LB.readFile fp+ let close_f = removeLink fp >> closeFd lock_fd+ return $ Just (contents, close_f)++-- | Check the work directory for any outstanding work, if there is a potential+-- candidate, lock it. If that fails, try the next.+tryCurDir :: FilePath -> IO (Maybe (FilePath, Fd))+tryCurDir cur_dir =+ listToMaybe . catMaybes <$> (getAbsoluteDirectoryFiles cur_dir+ >>= mapM lazyLock)+ where+ lazyLock :: FilePath -> IO (Maybe (FilePath, Fd))+ lazyLock fp = unsafeInterleaveIO $ do+ lock <- tryLock fp+ case lock of+ Nothing -> return Nothing+ Just lock_fd -> return . Just $ (fp, lock_fd)++-- Attempt to rotate all files from src folder to dst folder.+rotate :: FilePath -> FilePath -> IO ()+rotate src dst = do+ candidates <- getAbsoluteDirectoryFiles src+ unless (null candidates)+ (mapM_ doMove candidates)+ where+ doMove src_file = do+ (new_path, h) <- mkstemp dst+ hClose h+ renameFile src_file new_path++getAbsoluteDirectoryFiles :: FilePath -> IO [FilePath]+getAbsoluteDirectoryFiles =+ getAbsoluteDirectoryContents >=> filterM doesFileExist++getAbsoluteDirectoryContents :: FilePath -> IO [FilePath]+getAbsoluteDirectoryContents fp =+ map (\rel -> joinPath [fp, rel]) <$> getDirectoryContents fp++doAppend :: FilePath -> ByteString -> IO ()+doAppend = S.appendFile++newPointsDir :: SpoolName -> FilePath+newPointsDir = specificSpoolDir ["points", "new"]++newContentsDir :: SpoolName -> FilePath+newContentsDir = specificSpoolDir ["contents", "new"]++curPointsDir :: SpoolName -> FilePath+curPointsDir = specificSpoolDir ["points", "cur"]++curContentsDir :: SpoolName -> FilePath+curContentsDir = specificSpoolDir ["contents", "cur"]++specificSpoolDir :: [String] -> SpoolName -> FilePath+specificSpoolDir subdirs sn =+ addTrailingPathSeparator (joinPath (baseSpoolDir sn:subdirs))++baseSpoolDir :: SpoolName -> FilePath+baseSpoolDir (SpoolName sn) = joinPath ["/var/spool/marquise/", sn]
+ lib/Marquise/IO/Writer.hs view
@@ -0,0 +1,36 @@+--+-- Copyright © 2013-2014 Anchor Systems, Pty Ltd and Others+--+-- The code in this file, and the program it is a part of, is+-- made available to you by its authors as open source software:+-- you can redistribute it and/or modify it under the terms of+-- the 3-clause BSD licence.+--++{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TupleSections #-}+{-# OPTIONS_GHC -fno-warn-orphans #-}+-- Hide warnings for the deprecated ErrorT transformer:+{-# OPTIONS_GHC -fno-warn-warnings-deprecations #-}++module Marquise.IO.Writer+(+) where++import qualified Control.Exception as E++import Marquise.Classes+import Marquise.IO.Connection+import Marquise.Types+import Vaultaire.Types++instance MarquiseWriterMonad IO where+ transmitBytes broker origin bytes =+ withConnection ("tcp://" ++ broker ++ ":5560") $ \c -> do+ send (PassThrough bytes) origin c+ ack <- recv c+ case ack of+ OnDisk -> return ()+ InvalidWriteOrigin -> E.throw $ MarquiseException "invalid origin"
+ lib/Marquise/Server.hs view
@@ -0,0 +1,308 @@+--+-- Data vault for metrics+--+-- Copyright © 2013-2014 Anchor Systems, Pty Ltd and Others+--+-- The code in this file, and the program it is a part of, is+-- made available to you by its authors as open source software:+-- you can redistribute it and/or modify it under the terms of+-- the 3-clause BSD licence.+--++{-# LANGUAGE MultiParamTypeClasses #-}++-- Hide warnings for the deprecated ErrorT transformer:+{-# OPTIONS_GHC -fno-warn-warnings-deprecations #-}++-- | Marquise server library, for transmission of queued data to the vault.+module Marquise.Server+(+ runMarquiseDaemon,+ parseContentsRequests,+ breakInToChunks,+)+where++import Control.Applicative+import Control.Concurrent (threadDelay)+import Control.Concurrent.Async+import qualified Control.Concurrent.Async.Lifted as AL+import Control.Concurrent.MVar+import Control.Exception (throw)+import Control.Monad+import Control.Monad.Error+import Control.Monad.State.Lazy+import Data.Attoparsec.ByteString.Lazy (Parser)+import Data.Attoparsec.Combinator (eitherP)+import qualified Data.Attoparsec.Lazy as Parser+import Data.ByteString.Builder (Builder, byteString, toLazyByteString)+import qualified Data.ByteString.Char8 as S+import qualified Data.ByteString.Lazy as L+import Data.Maybe+import Data.Monoid+import Data.Packer+import Data.Time.Clock+import Pipes+import Pipes.Attoparsec (parsed)+import qualified Pipes.ByteString as PB+import Pipes.Group (FreeF (..), FreeT (..))+import qualified Pipes.Group as PG+import qualified Pipes.Lift as P+import System.IO+import System.Log.Logger++import Marquise.Classes+import Marquise.Client (makeSpoolName, updateSourceDict)+import Marquise.IO ()+import Marquise.Types+import Vaultaire.Types++data ContentsRequest = ContentsRequest Address SourceDict+ deriving Show++runMarquiseDaemon :: String -> Origin -> String -> MVar () -> String -> Integer -> IO (Async ())+runMarquiseDaemon broker origin namespace shutdown cache_file cache_flush_period =+ async $ startMarquise broker origin namespace shutdown cache_file cache_flush_period++startMarquise :: String -> Origin -> String -> MVar () -> String -> Integer -> IO ()+startMarquise broker origin name shutdown cache_file cache_flush_period = do+ infoM "Server.startMarquise" $ "Reading SourceDict cache from " ++ cache_file+ init_cache <- withFile cache_file ReadWriteMode $ \h -> do+ result <- fromWire <$> S.hGetContents h+ case result of+ Left e -> do+ warningM "Server.startMarquise" $+ concat ["Error decoding hash file: "+ , show e+ , " Continuing with empty initial cache"+ ]+ return emptySourceCache+ Right cache -> do+ debugM "Server.startMarquise" $+ concat ["Read "+ , show (sizeOfSourceCache cache)+ , " hashes from source dict cache."+ ]+ return cache+ infoM "Server.startMarquise" "Marquise daemon started"++ (points_loop, final_cache) <- do+ sn <- makeSpoolName name+ debugM "Server.startMarquise" "Creating spool directories"+ createDirectories sn+ debugM "Server.startMarquise" "Starting point transmitting thread"+ points_loop <- AL.async (sendPoints broker origin sn shutdown)+ currTime <- do+ link points_loop+ debugM "Server.startMarquise" "Starting contents transmitting thread"+ getCurrentTime+ final_cache <- sendContents broker origin sn init_cache cache_file cache_flush_period currTime shutdown+ return (points_loop, final_cache)++ debugM "Server.startMarquise" "Send loop shut down gracefully, writing out cache"+ S.writeFile cache_file $ toWire final_cache++ debugM "Server.startMarquise" "Waiting for points loop thread"+ AL.wait points_loop++sendPoints :: String -> Origin -> SpoolName -> MVar () -> IO ()+sendPoints broker origin sn shutdown = do+ nexts <- nextPoints sn+ case nexts of+ Just (bytes, seal) -> do+ debugM "Server.sendPoints" "Got points, starting transmission pipe"+ runEffect $ for (breakInToChunks bytes) sendChunk+ debugM "Server.sendPoints" "Transmission complete, cleaning up"+ seal+ Nothing -> threadDelay idleTime++ done <- isJust <$> tryReadMVar shutdown+ unless done (sendPoints broker origin sn shutdown)+ where+ sendChunk chunk = do+ let size = show . S.length $ chunk+ liftIO (debugM "Server.sendPoints" $ "Sending chunk of " ++ size ++ " bytes.")+ lift (transmitBytes broker origin chunk)++sendContents :: String+ -> Origin+ -> SpoolName+ -> SourceDictCache+ -> String+ -> Integer+ -> UTCTime+ -> MVar ()+ -> IO SourceDictCache+sendContents broker origin sn initial cache_file cache_flush_period flush_time shutdown = do+ nexts <- nextContents sn+ (final, newFlushTime) <- case nexts of+ Just (bytes, seal) -> do+ debugM "Server.sendContents" $+ concat+ [ "Got contents, starting transmission pipe with "+ , show $ sizeOfSourceCache initial+ , " cached sources."+ ]++ ((), final') <- withContentsConnection broker $ \c ->+ runEffect $ for (P.runStateP initial (parseContentsRequests bytes >-> filterSeen))+ (sendSourceDictUpdate c)++ newFlushTime' <- do+ debugM "Server.sendContents" "Contents transmission complete, cleaning up."+ debugM "Server.sendContents" $+ concat+ [ "Saw "+ , show $ sizeOfSourceCache final' - sizeOfSourceCache initial+ , " new sources."+ ]+ seal+ currTime <- getCurrentTime+ if currTime > flush_time+ then do+ debugM "Server.setContents" "Performing periodic cache writeout."+ S.writeFile cache_file $ toWire final'+ return $ addUTCTime (fromInteger cache_flush_period) currTime+ else do+ debugM "Server.sendContents" $ concat ["Next cache flush at ", show flush_time, "."]+ return flush_time+ return (final', newFlushTime')++ Nothing -> do+ threadDelay idleTime+ return (initial, flush_time)++ done <- isJust <$> tryReadMVar shutdown++ if done+ then return final+ else sendContents broker origin sn final cache_file cache_flush_period newFlushTime shutdown++ where+ filterSeen = forever $ do+ req@(ContentsRequest addr sd) <- await+ cache <- get+ let currHash = hashSource sd+ if memberSourceCache currHash cache then+ liftIO $ debugM "Server.filterSeen" $ "Seen source dict with address " ++ show addr ++ " before, ignoring."+ else do+ put (insertSourceCache currHash cache)+ yield req+ sendSourceDictUpdate conn (ContentsRequest addr source_dict) = do+ liftIO (debugM "Server.sendContents" $ "Sending contents update for " ++ show addr)+ lift (updateSourceDict addr source_dict origin conn)++parseContentsRequests :: Monad m => L.ByteString -> Producer ContentsRequest m ()+parseContentsRequests bs =+ parsed parseContentsRequest (PB.fromLazy bs)+ >>= either (throw . fst) return++parseContentsRequest :: Parser ContentsRequest+parseContentsRequest = do+ addr <- fromWire <$> Parser.take 8+ len <- runUnpacking getWord64LE <$> Parser.take 8+ source_dict <- fromWire <$> Parser.take (fromIntegral len)+ case ContentsRequest <$> addr <*> source_dict of+ Left e -> fail (show e)+ Right request -> return request++idleTime :: Int+idleTime = 1000000 -- 1 second++breakInToChunks :: Monad m => L.ByteString -> Producer S.ByteString m ()+breakInToChunks bs =+ chunkBuilder (parsed parsePoint (PB.fromLazy bs))+ >>= either (throw . fst) return++-- Take a producer of (Int, Builder), where Int is the number of bytes in the+-- builder and produce chunks of n bytes.+--+-- This could be done with explicit recursion and next, but, then we would not+-- get to apply a fold over a FreeT stack of producers. This is almost+-- generalizable, at a stretch.+chunkBuilder :: Monad m => Producer (Int, Builder) m r -> Producer S.ByteString m r+chunkBuilder = PG.folds (<>) mempty (L.toStrict . toLazyByteString)+ -- Fold over each producer of counted Builders, turning it into+ -- a contigous strict ByteString ready for transmission.+ . builderChunks idealBurstSize+ -- Split the builder producer into FreeT+ where+ builderChunks :: Monad m+ => Int+ -- ^ The size to split a stream of builders at+ -> Producer (Int, Builder) m r+ -- ^ The input producer+ -> FreeT (Producer Builder m) m r+ -- ^ The FreeT delimited chunks of that producer, split into+ -- the desired chunk length+ builderChunks max_size p = FreeT $ do+ -- Try to grab the next value from the Producer+ x <- next p+ return $ case x of+ Left r -> Pure r+ Right (a, p') -> Free $ do+ -- Pass the re-joined Producer to go, which will yield values+ -- from it until the desired chunk size is reached.+ p'' <- go max_size (yield a >> p')+ -- The desired chunk size has been reached, loop and try again+ -- with the rest of the stream (possibly empty)+ return (builderChunks max_size p'')++ -- We take a Producer and pass along its values until we've passed along+ -- enough bytes (at least the initial bytes_left).+ --+ -- When done, returns the remainder of the unconsumed Producer+ go :: Monad m+ => Int+ -> Producer (Int, Builder) m r+ -> Producer Builder m (Producer (Int, Builder) m r)+ go bytes_left p =+ if bytes_left < 0+ then return p+ else do+ x <- lift (next p)+ case x of+ Left r ->+ return . return $ r+ Right ((size, builder), p') -> do+ yield builder+ go (bytes_left - size) p'++-- Parse a single point, returning the size of the point and the bytes as a+-- builder.+parsePoint :: Parser (Int, Builder)+parsePoint = do+ packet <- Parser.take 24++ case extendedSize packet of+ Just len -> do+ -- We must ensure that we get this many bytes now, or attoparsec+ -- will just backtrack on us. We do this with a dummy parser inside+ -- an eitherP+ --+ -- This is only to get good error messages.+ extended <- eitherP (Parser.take len) (return ())+ case extended of+ Left bytes ->+ let b = byteString packet <> byteString bytes+ in return (24 + len, b)+ Right () ->+ fail "not enough bytes in alleged extended burst"+ Nothing ->+ return (24, byteString packet)++-- Return the size of the extended segment, if the point is an extended one.+extendedSize :: S.ByteString -> Maybe Int+extendedSize packet = flip runUnpacking packet $ do+ addr <- Address <$> getWord64LE+ if isAddressExtended addr+ then do+ unpackSkip 8+ Just . fromIntegral <$> getWord64LE -- length+ else+ return Nothing++-- A burst should be, at maximum, very close to this size, unless the user+-- decides to send a very long extended point.+idealBurstSize :: Int+idealBurstSize = 16 * 1048576
+ lib/Marquise/Types.hs view
@@ -0,0 +1,47 @@+--+-- Copyright © 2013-2014 Anchor Systems, Pty Ltd and Others+--+-- The code in this file, and the program it is a part of, is+-- made available to you by its authors as open source software:+-- you can redistribute it and/or modify it under the terms of+-- the 3-clause BSD licence.+--++{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE TypeFamilies #-}++module Marquise.Types+ ( -- * Data+ SpoolName(..)+ , SpoolFiles(..)+ , TimeStamp(..)+ , SimplePoint(..), ExtendedPoint(..)+ , MarquiseException(..)+ ) where++import Control.Exception+import Data.Typeable++import Vaultaire.Types+++-- | A NameSpace implies a certain amount of Marquise server-side state. This+-- state being the Marquise server's authentication and origin configuration.+newtype SpoolName = SpoolName { unSpoolName :: String }+ deriving (Eq, Show)++-- | SpoolFiles simple wraps around two file paths.+-- One for queuing points updates, one for queuing contents updates+data SpoolFiles = SpoolFiles { pointsSpoolFile :: FilePath+ , contentsSpoolFile :: FilePath }+ deriving (Eq, Show)++newtype MarquiseException = MarquiseException String+ deriving (Show, Typeable)++instance Exception MarquiseException
+ marquise.cabal view
@@ -0,0 +1,161 @@+cabal-version: >= 1.10+name: marquise+version: 4.0.0+license-file: LICENCE+synopsis: Client library for Vaultaire+description: Marquise is a collection of a library and two executables for use with Vaultaire.+ .+ 1. A client and server library for reading/writing to the vault and spool files.+ This provides streaming reads and writes to the vault using+ [pipes](https://hackage.haskell.org/package/pipes) as well as writing to spool+ files with automatic caching and rotation.+ .+ 2. An executable `marquised`, a daemon which writes data to the vault from spool+ files generated from users of the marquise library.+ .+ 3. An executable `data`, used for easily inspecting data in the vault as well as+ marquise cache files.+license: BSD3+author: Anchor Engineering <engineering@anchor.com.au>+maintainer: Anchor Engineering <engineering@anchor.com.au>+copyright: © 2013-2015 Anchor Systems, Pty Ltd and Others+category: Other+tested-with: GHC == 7.8.3+stability: experimental++build-type: Custom++source-repository head+ type: git+ location: git@github.com:anchor/marquise.git+++library+ hs-source-dirs: lib+ default-language: Haskell2010++ exposed-modules: Marquise.Client,+ Marquise.Classes,+ Marquise.Types,+ Marquise.Server++ other-modules: Marquise.IO,+ Marquise.IO.FFI,+ Marquise.IO.Writer,+ Marquise.IO.Reader,+ Marquise.IO.Contents,+ Marquise.IO.SpoolFile,+ Marquise.IO.Connection,+ Marquise.Client.Core++ build-depends: base >=3 && <5,+ mtl,+ transformers,+ monad-logger,+ fast-logger,+ either,+ errors,+ transformers-base,+ monad-control,+ mmorph,+ lifted-async,+ containers,+ unordered-containers,+ bytestring >= 0.10.2,+ text,+ attoparsec,+ unix,+ pipes,+ pipes-group,+ pipes-bytestring,+ pipes-attoparsec >= 0.5,+ siphash,+ async,+ hslogger,+ hashable,+ binary,+ packer,+ filepath,+ directory,+ semigroups,+ cryptohash,+ zeromq4-haskell,+ time,+ vaultaire-common >= 2.8.3++ ghc-options: -Wall+ -O2+ -Wwarn+ -fwarn-tabs++ ghc-prof-options: -fprof-auto+++executable marquised+ hs-source-dirs: src+ main-is: MarquiseDaemon.hs+ default-language: Haskell2010++ build-depends: base >=3 && <5,+ bytestring,+ hslogger,+ optparse-applicative >= 0.11.0,+ unix,+ async,+ containers,+ vaultaire-common >= 2.8.3,+ marquise++ ghc-options: -threaded+ -O2+ -Wall+ -Wwarn+ -fwarn-tabs++ ghc-prof-options: -fprof-auto++executable data+ hs-source-dirs: src+ main-is: DataProgram.hs+ default-language: Haskell2010++ build-depends: base >=3 && <5,+ bytestring,+ hslogger,+ optparse-applicative >= 0.11.0,+ unix,+ pipes,+ text,+ attoparsec,+ unordered-containers,+ time,+ old-locale,+ data-binary-ieee754,+ packer,+ vaultaire-common >= 2.8.3,+ marquise++ ghc-options: -threaded+ -O2+ -Wall+ -Wwarn+ -fwarn-tabs++ ghc-prof-options: -fprof-auto++test-suite spool-test+ hs-source-dirs: tests+ main-is: Spool.hs+ type: exitcode-stdio-1.0+ default-language: Haskell2010++ build-depends: base >=3 && <5,+ hspec,+ bytestring,+ marquise++ ghc-options: -threaded+ -O2+ -Wall+ -Wwarn+ -fwarn-tabs
+ src/DataProgram.hs view
@@ -0,0 +1,308 @@+--+-- Data vault for metrics+--+-- Copyright © 2013-2014 Anchor Systems, Pty Ltd and Others+--+-- The code in this file, and the program it is a part of, is+-- made available to you by its authors as open source software:+-- you can redistribute it and/or modify it under the terms of+-- the 3-clause BSD licence.+--++{-# LANGUAGE DoAndIfThenElse #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE RecordWildCards #-}++module Main where++import Control.Concurrent.MVar+import qualified Data.Attoparsec.Text as PT+import Data.Binary.IEEE754+import qualified Data.ByteString.Char8 as S+import qualified Data.HashMap.Strict as HT+import Data.String+import Data.Text (Text)+import qualified Data.Text as T+import Data.Time+import Data.Time.Clock.POSIX+import Data.Word+import Options.Applicative+import Options.Applicative.Types+import Pipes+import qualified Pipes.Prelude as P+import System.IO+import System.Locale+import System.Log.Logger+import Text.Printf++import Marquise.Classes+import Marquise.Client+import Package (package, version)+import Vaultaire.Program+import Vaultaire.Types+import Vaultaire.Util++--+-- Component line option parsing+--++data Options = Options+ { broker :: String+ , debug :: Bool+ , component :: Component }++data Component =+ Now+ | Read { raw :: Bool+ , origin :: Origin+ , address :: Address+ , start :: TimeStamp+ , end :: TimeStamp }+ | List { origin :: Origin }+ | Add { origin :: Origin+ , addr :: Address+ , dict :: [Tag] }+ | Remove { origin :: Origin+ , addr :: Address+ , dict :: [Tag] }+ | SourceCache { cacheFile :: FilePath }++type Tag = (Text, Text)++justRead :: Read a => ReadM a+justRead = readerAsk >>= return . read++helpfulParser :: ParserInfo Options+helpfulParser = info (helper <*> optionsParser) fullDesc++optionsParser :: Parser Options+optionsParser = Options <$> parseBroker+ <*> parseDebug+ <*> parseComponents+ where+ parseBroker = strOption $+ long "broker"+ <> short 'b'+ <> metavar "BROKER"+ <> value "localhost"+ <> showDefault+ <> help "Vaultaire broker hostname or IP address"++ parseDebug = switch $+ long "debug"+ <> short 'd'+ <> help "Output lots of debugging information"++ parseComponents = subparser+ ( parseTimeComponent+ <> parseReadComponent+ <> parseListComponent+ <> parseAddComponent+ <> parseRemoveComponent+ <> parseSourceCacheComponent )++ parseTimeComponent =+ componentHelper "now" (pure Now) "Display the current time"++ parseReadComponent =+ componentHelper "read" readOptionsParser "Read points from a given address and time range"++ parseListComponent =+ componentHelper "list" listOptionsParser "List addresses and metadata in origin"++ parseAddComponent =+ componentHelper "add-source" addOptionsParser "Add some tags to an address"++ parseRemoveComponent =+ componentHelper "remove-source" addOptionsParser "Remove some tags from an address, does nothing if the tags do not exist"++ parseSourceCacheComponent =+ componentHelper "source-cache" sourceCacheOptionsParser "Validate the contents of a Marquise daemon source cache file"++ componentHelper cmd_name parser desc =+ command cmd_name (info (helper <*> parser) (progDesc desc))+++parseAddress :: Parser Address+parseAddress = argument (readerAsk >>= return . fromString) (metavar "ADDRESS")++parseOrigin :: Parser Origin+parseOrigin = argument (readerAsk >>= return . mkOrigin) (metavar "ORIGIN")+ where+ mkOrigin = Origin . S.pack++parseTags :: Parser [Tag]+parseTags = many $ argument (readerAsk >>= return . mkTag) (metavar "TAG")+ where+ mkTag x = case PT.parseOnly tag $ T.pack x of+ Left _ -> error "data: invalid tag format"+ Right y -> y+ tag = (,) <$> PT.takeWhile (/= ':')+ <* ":"+ <*> PT.takeWhile (/= ',')++readOptionsParser :: Parser Component+readOptionsParser = Read <$> parseRaw+ <*> parseOrigin+ <*> parseAddress+ <*> parseStart+ <*> parseEnd+ where+ parseRaw = switch $+ long "raw"+ <> short 'r'+ <> help "Output values in a raw form (human-readable otherwise)"++ parseStart = option justRead $+ long "start"+ <> short 's'+ <> value 0+ <> showDefault+ <> help "Start time in nanoseconds since epoch"++ parseEnd = option justRead $+ long "end"+ <> short 'e'+ <> value maxBound+ <> showDefault+ <> help "End time in nanoseconds since epoch"++listOptionsParser :: Parser Component+listOptionsParser = List <$> parseOrigin++addOptionsParser :: Parser Component+addOptionsParser = Add <$> parseOrigin <*> parseAddress <*> parseTags++removeOptionsParser :: Parser Component+removeOptionsParser = Remove <$> parseOrigin <*> parseAddress <*> parseTags++sourceCacheOptionsParser :: Parser Component+sourceCacheOptionsParser = SourceCache <$> parseFilePath+ where+ parseFilePath = argument str $ metavar "CACHEFILE"++--+-- Actual tools+--++runPrintDate :: IO ()+runPrintDate = do+ now <- getCurrentTime+ let time = formatTime defaultTimeLocale "%FT%TZ" now+ putStrLn time++runReadPoints :: String -> Bool -> Origin -> Address -> TimeStamp -> TimeStamp -> IO ()+runReadPoints broker raw origin addr start end =+ withReaderConnection broker $ \c ->+ runEffect $ readSimplePoints addr start end origin c+ >-> P.map (displayPoint raw)+ >-> P.print++displayPoint :: Bool -> SimplePoint -> String+displayPoint raw (SimplePoint address timestamp payload) =+ if raw+ then+ show address ++ "," ++ show timestamp ++ "," ++ show payload+ else+ show address ++ " " ++ formatTimestamp timestamp ++ " " ++ formatValue payload+ where+ formatTimestamp :: TimeStamp -> String+ formatTimestamp (TimeStamp t) =+ let+ seconds = posixSecondsToUTCTime $ realToFrac (fromIntegral t / 1000000000 :: Rational)+ iso8601 = formatTime defaultTimeLocale "%FT%T.%q" seconds+ in+ take 29 iso8601 ++ "Z"++{-+ Take a stab at differentiating between raw integers and encoded floats.+ The 2^51 is somewhat arbitrary, being one less bit than the size of the+ significand in an IEEE 754 64-bit double. Seems safe to assume anything+ larger than that was in fact an encoded float; 2^32 (aka 10^9) is too small+ — we have counters bigger than that — but nothing has gone past 10^15 so+ there seemed plenty of headroom. At the end of the day this is just a+ convenience; if you need the real value and know its interpretation then+ you can request raw (in this program) output or actual bytes (via reader+ daemon).+-}+ formatValue :: Word64 -> String+ formatValue v = if v > (2^(51 :: Int) :: Word64)+ then+ printf "% 24.6f" (wordToDouble v)+ else+ printf "% 17d" v+++runListContents :: String -> Origin -> IO ()+runListContents broker origin =+ withContentsConnection broker $ \c ->+ runEffect $ enumerateOrigin origin c >-> P.print++runAddTags, runRemoveTags :: String -> Origin -> Address -> [Tag] -> IO ()+runAddTags = run updateSourceDict+runRemoveTags = run removeSourceDict++runSourceCache :: FilePath -> IO ()+runSourceCache cacheFile = do+ bits <- withFile cacheFile ReadMode S.hGetContents+ case fromWire bits of+ Left e -> putStrLn . concat $+ [ "Error parsing cache file: "+ , show e+ ]+ Right cache -> putStrLn . concat $+ [ "Valid Marquise source cache. Contains "+ , show . sizeOfSourceCache $ cache+ , " entries."+ ]++run :: (MarquiseContentsMonad m connection, Functor m)+ => (Address -> SourceDict -> Origin -> connection -> m a)+ -> String -> Origin -> Address -> [(Text, Text)] -> m a+run op broker origin addr sdPairs = do+ let dict = case makeSourceDict $ HT.fromList sdPairs of+ Left e -> error e+ Right a -> a+ withContentsConnection broker $ \c ->+ op addr dict origin c++--+-- Main program entry point+--++main :: IO ()+main = do+ Options{..} <- execParser helpfulParser++ let level = if debug+ then Debug+ else Quiet++ quit <- initializeProgram (package ++ "-" ++ version) level++ -- Run selected component.+ debugM "Main.main" "Running command"++ -- Although none of the components are running in the background, we get off+ -- of the main thread so that we can block the main thread on the quit+ -- semaphore, such that a user interrupt will kill the program.++ linkThread $ do+ _ <- case component of+ Now ->+ runPrintDate+ Read human origin addr start end ->+ runReadPoints broker human origin addr start end+ List origin ->+ runListContents broker origin+ Add origin addr tags ->+ runAddTags broker origin addr tags+ Remove origin addr tags ->+ runRemoveTags broker origin addr tags+ SourceCache cacheFile ->+ runSourceCache cacheFile+ putMVar quit ()++ takeMVar quit+ debugM "Main.main" "End"
+ src/MarquiseDaemon.hs view
@@ -0,0 +1,114 @@+--+-- Data vault for metrics+--+-- Copyright © 2013-2014 Anchor Systems, Pty Ltd and Others+--+-- The code in this file, and the program it is a part of, is+-- made available to you by its authors as open source software:+-- you can redistribute it and/or modify it under the terms of+-- the 3-clause BSD licence.+--++{-# LANGUAGE RecordWildCards #-}++module Main where++import Control.Concurrent.Async+import qualified Data.ByteString.Char8 as S+import Data.Monoid+import Options.Applicative hiding (Parser, option)+import qualified Options.Applicative as O+import Options.Applicative.Types+import System.Log.Logger++import Marquise.Client+import Marquise.Server+import Package (package, version)+import Vaultaire.Program++data Options = Options+ { broker :: String+ , debug :: Bool+ , quiet :: Bool+ , cacheFile :: String+ , cacheFlushFreq :: Integer+ , origin :: String+ , namespace :: String }++justRead :: Read a => ReadM a+justRead = readerAsk >>= return . read++helpfulParser :: O.ParserInfo Options+helpfulParser = info (helper <*> optionsParser) fullDesc++optionsParser :: O.Parser Options+optionsParser = Options <$> parseBroker+ <*> parseDebug+ <*> parseQuiet+ <*> parseCacheFile+ <*> parseCacheFlushFreq+ <*> parseOrigin+ <*> parseNameSpace+ where+ parseBroker = strOption $+ long "broker"+ <> short 'b'+ <> metavar "BROKER"+ <> value "localhost"+ <> showDefault+ <> help "Vault broker host name or IP address"++ parseDebug = switch $+ long "debug"+ <> short 'd'+ <> help "Output lots of debugging information"++ parseQuiet = switch $+ long "quiet"+ <> short 'q'+ <> help "Only emit warnings or fatal messages"++ parseCacheFile = strOption $+ long "cache-file"+ <> short 'c'+ <> value ""+ <> help "Location to read/write cached SourceDicts"++ -- This doesn't mean that marquised will rigorously flush the cache+ -- every `t` seconds. To clarify: marquised consideres flushing the+ -- cache after it finishes reading every spool file. If the cache+ -- hasn't been flushed in the last `t` seconds, it will be flushed+ -- before processing the next spool file.+ parseCacheFlushFreq = O.option justRead $+ long "cache-flush-freq"+ <> short 't'+ <> help "Period of time to wait between cache writes, in seconds"+ <> value 42++ parseNameSpace = argument str (metavar "NAMESPACE")++ parseOrigin = argument str (metavar "ORIGIN")++defaultCacheLoc :: String -> String+defaultCacheLoc = (++) "/var/cache/marquise/"++main :: IO ()+main = do+ Options{..} <- execParser $ helpfulParser++ let level+ | debug = Debug+ | quiet = Quiet+ | otherwise = Normal++ quit <- initializeProgram (package ++ "-" ++ version) level++ cacheFile' <- return $ case cacheFile of+ "" -> defaultCacheLoc origin+ x -> x++ a <- runMarquiseDaemon broker (Origin $ S.pack origin) namespace quit cacheFile' cacheFlushFreq++ -- wait forever+ wait a+ debugM "Main.main" "End"
+ tests/Spool.hs view
@@ -0,0 +1,51 @@+{-# LANGUAGE OverloadedStrings #-}+{-# OPTIONS_GHC -fno-warn-orphans #-}++import Marquise.Classes+import Marquise.Client+import Marquise.Types+import Test.Hspec++-- we assume successes in this test, tests for failures (timeouts etc) need to be added++ns1, ns2 :: Monad m => m SpoolName+ns1 = makeSpoolName "ns1"+ns2 = makeSpoolName "ns2"++main :: IO ()+main = hspec suite++suite :: Spec+suite =+ describe "IO MarquiseClientMonad and MarquiseServerMonad" $+ it "reads appends, then cleans up when nextBurst is called" $ do+ (bytes1, bytes2, contents) <- do+ sf1 <- createSpoolFiles "ns1"+ sf2 <- createSpoolFiles "ns2"++ appendPoints sf1 "BBBBBBBBAAAAAAAACCCCCCCC"+ appendPoints sf1 "DBBBBBBBAAAAAAAACCCCCCCC"+ appendPoints sf2 "FBBBBBBBAAAAAAAACCCCCCCC"+ appendContents sf1 "contents"++ (bytes1,close_f1) <- loop (nextPoints =<< ns1)+ (bytes2,close_f2) <- loop (nextPoints =<< ns2)+ (contents,close_f3) <- loop (nextContents =<< ns1)++ do close_f1+ close_f2+ close_f3++ return (bytes1, bytes2, contents)++ bytes1 `shouldBe` "BBBBBBBBAAAAAAAACCCCCCCC\+ \DBBBBBBBAAAAAAAACCCCCCCC"+ bytes2 `shouldBe` "FBBBBBBBAAAAAAAACCCCCCCC"+ contents `shouldBe` "contents"++ where+ loop x = do+ x' <- x+ case x' of+ Just result -> return result+ Nothing -> loop x