diff --git a/LICENCE b/LICENCE new file mode 100644 --- /dev/null +++ b/LICENCE @@ -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. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -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 + diff --git a/lib/Marquise/Classes.hs b/lib/Marquise/Classes.hs new file mode 100644 --- /dev/null +++ b/lib/Marquise/Classes.hs @@ -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 diff --git a/lib/Marquise/Client.hs b/lib/Marquise/Client.hs new file mode 100644 --- /dev/null +++ b/lib/Marquise/Client.hs @@ -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 diff --git a/lib/Marquise/Client/Core.hs b/lib/Marquise/Client/Core.hs new file mode 100644 --- /dev/null +++ b/lib/Marquise/Client/Core.hs @@ -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 diff --git a/lib/Marquise/IO.hs b/lib/Marquise/IO.hs new file mode 100644 --- /dev/null +++ b/lib/Marquise/IO.hs @@ -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 () diff --git a/lib/Marquise/IO/Connection.hs b/lib/Marquise/IO/Connection.hs new file mode 100644 --- /dev/null +++ b/lib/Marquise/IO/Connection.hs @@ -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 diff --git a/lib/Marquise/IO/Contents.hs b/lib/Marquise/IO/Contents.hs new file mode 100644 --- /dev/null +++ b/lib/Marquise/IO/Contents.hs @@ -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 diff --git a/lib/Marquise/IO/FFI.hsc b/lib/Marquise/IO/FFI.hsc new file mode 100644 --- /dev/null +++ b/lib/Marquise/IO/FFI.hsc @@ -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 + +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 () diff --git a/lib/Marquise/IO/Reader.hs b/lib/Marquise/IO/Reader.hs new file mode 100644 --- /dev/null +++ b/lib/Marquise/IO/Reader.hs @@ -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 diff --git a/lib/Marquise/IO/SpoolFile.hs b/lib/Marquise/IO/SpoolFile.hs new file mode 100644 --- /dev/null +++ b/lib/Marquise/IO/SpoolFile.hs @@ -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] diff --git a/lib/Marquise/IO/Writer.hs b/lib/Marquise/IO/Writer.hs new file mode 100644 --- /dev/null +++ b/lib/Marquise/IO/Writer.hs @@ -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" diff --git a/lib/Marquise/Server.hs b/lib/Marquise/Server.hs new file mode 100644 --- /dev/null +++ b/lib/Marquise/Server.hs @@ -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 diff --git a/lib/Marquise/Types.hs b/lib/Marquise/Types.hs new file mode 100644 --- /dev/null +++ b/lib/Marquise/Types.hs @@ -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 diff --git a/marquise.cabal b/marquise.cabal new file mode 100644 --- /dev/null +++ b/marquise.cabal @@ -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 +maintainer: Anchor Engineering +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 diff --git a/src/DataProgram.hs b/src/DataProgram.hs new file mode 100644 --- /dev/null +++ b/src/DataProgram.hs @@ -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" diff --git a/src/MarquiseDaemon.hs b/src/MarquiseDaemon.hs new file mode 100644 --- /dev/null +++ b/src/MarquiseDaemon.hs @@ -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" diff --git a/tests/Spool.hs b/tests/Spool.hs new file mode 100644 --- /dev/null +++ b/tests/Spool.hs @@ -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