vaultaire-common (empty) → 2.9.1

raw patch · 24 files changed

+1612/−0 lines, 24 filesdep +QuickCheckdep +asyncdep +attoparsecsetup-changed

Dependencies added: QuickCheck, async, attoparsec, base, blaze-builder, bytestring, cereal, containers, hashable, hslogger, hspec, locators, mtl, old-locale, packer, siphash, text, time, transformers, unix, unordered-containers, vaultaire-common

Files

• LICENSE +32/−0
• Setup.hs +2/−0
• lib/Vaultaire/Classes/WireFormat.hs +18/−0
• lib/Vaultaire/Program.hs +110/−0
• lib/Vaultaire/Types.hs +101/−0
• lib/Vaultaire/Types/Address.hs +58/−0
• lib/Vaultaire/Types/Common.hs +67/−0
• lib/Vaultaire/Types/ContentsListBypass.hs +48/−0
• lib/Vaultaire/Types/ContentsOperation.hs +70/−0
• lib/Vaultaire/Types/ContentsResponse.hs +85/−0
• lib/Vaultaire/Types/DayMap.hs +71/−0
• lib/Vaultaire/Types/Decoded.hs +33/−0
• lib/Vaultaire/Types/PassThrough.hs +28/−0
• lib/Vaultaire/Types/ReadRequest.hs +74/−0
• lib/Vaultaire/Types/ReadStream.hs +58/−0
• lib/Vaultaire/Types/SourceDict.hs +101/−0
• lib/Vaultaire/Types/SourceDictCache.hs +38/−0
• lib/Vaultaire/Types/Telemetry.hs +213/−0
• lib/Vaultaire/Types/TimeStamp.hs +125/−0
• lib/Vaultaire/Types/WriteResult.hs +35/−0
• lib/Vaultaire/Util.hs +24/−0
• tests/InstanceTest.hs +39/−0
• tests/WireFormatsTest.hs +71/−0
• vaultaire-common.cabal +111/−0

@@ -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.

@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain

@@ -0,0 +1,18 @@+module Vaultaire.Classes.WireFormat+(+    WireFormat(..),+) where++import Control.Exception (SomeException)+import Data.ByteString (ByteString)++-- | This typeclass encapsulates all wire encoding/decoding, with the+-- possibility of a decode failing.+class WireFormat operation where+    fromWire :: ByteString -> Either SomeException operation+    toWire   :: operation -> ByteString++-- | Default dummy wire format+instance WireFormat () where+    fromWire = error "() has no wire format"+    toWire = error "() has no wire format"

@@ -0,0 +1,110 @@+--+-- 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.+--++--+-- | Common program initalization for Vaultaire binaries+--+module Vaultaire.Program+(+    initializeProgram,+    Verbosity(..)+)+where++import Control.Concurrent.MVar+import Control.Monad+import GHC.Conc+import System.Environment+import System.IO (hFlush, hPutStrLn, stdout)+import System.Log.Formatter+import System.Log.Handler (setFormatter)+import System.Log.Handler.Simple+import System.Log.Logger+import System.Posix.Signals+++--+-- Main program entry point+--++interruptHandler :: MVar () -> Handler+interruptHandler semaphore = Catch $ do+    putStrLn "\nInterrupt"+    hFlush stdout+    putMVar semaphore ()++terminateHandler :: MVar () -> Handler+terminateHandler semaphore = Catch $ do+    putStrLn "Terminating"+    hFlush stdout+    putMVar semaphore ()++quitHandler :: Handler+quitHandler = Catch $ do+    putStrLn ""+    hFlush stdout+    logger <- getLogger rootLoggerName+    let level   = getLevel logger+        level'  = case level of+                    Just DEBUG  -> INFO+                    Just INFO   -> DEBUG+                    _           -> DEBUG+        logger' = setLevel level' logger+    saveGlobalLogger logger'+    infoM "Main.quitHandler" ("Change log level to " ++ show level')++data Verbosity = Debug | Normal | Quiet deriving Show++--+-- | Initialize a program. Call this from your 'main' program entry point+-- before doing anything else. Indicate the logging verbosity you want, along+-- with an identification of your program. Returns an MVar which will be set to+-- unit if one of the installed signal handlers catches a signal and requests a+-- shutdown as a result.+--+initializeProgram :: String -> Verbosity -> IO (MVar ())+initializeProgram banner verbosity = do+    -- Indicate startup+    name <- getProgName+    case verbosity of+        Quiet -> return ()+        _     -> putStrLn $ name ++ " (" ++ banner ++ ") starting"++    -- command line +RTS -Nn -RTS value+    when (numCapabilities == 1) (getNumProcessors >>= setNumCapabilities)++    -- Start and configure logger, deleting the default handler in favour of+    -- our own formatter outputting to stdout with timestamps. Run in Zulu time.+    setEnv "TZ" "UTC"++    let level = case verbosity of+                   Debug  -> DEBUG+                   Normal -> INFO+                   Quiet  -> WARNING++    logger  <- getRootLogger+    handler <- streamHandler stdout DEBUG+    let handler' = setFormatter handler (tfLogFormatter "%Y-%m-%dT%H:%M:%SZ" "$time  $msg")+    let logger' = (setHandlers [handler'] . setLevel level) logger+    saveGlobalLogger logger'++    debugM "Program.initialize" "Logging initialized"++    quit <- newEmptyMVar++    _ <- installHandler sigINT  (interruptHandler quit) Nothing+    _ <- installHandler sigTERM (terminateHandler quit) Nothing+    _ <- installHandler sigQUIT quitHandler Nothing+++    debugM "Program.initialize" "Signal handlers installed"++    return quit

@@ -0,0 +1,101 @@+--+-- |+-- Maintainer: The Vaultaire Team+-- Stability: Experimental+--+-- /Overview/+--+-- When communicating with a Vaultaire installation, you need to serialize+-- requests and deserialize responses. The bytes used over the wire are all+-- formed by making the various types involved instances of class 'WireFormat'.+--+-- As it happens, these types are also the same ones used in the internals of+-- Vaultaire as it persists measurements and metadata to disk.+--++module Vaultaire.Types+(+    -- * Identification of measurements+    Address(..),+    calculateBucketNumber,+    isAddressExtended,++    -- * Time of a measurement+    TimeStamp(..),+    convertToDiffTime,+    convertToTimeStamp,+    getCurrentTimeNanoseconds,++    -- * Namespacing and authentication+    Origin(..),+    makeOrigin,++    -- * Metadata about sources+    SourceDict,+    unionSource,+    diffSource,+    lookupSource,+    makeSourceDict,++    -- * Caching SourceDicts+    hashSource,+    SourceDictCache,+    emptySourceCache,+    insertSourceCache,+    memberSourceCache,+    sizeOfSourceCache,++    -- * Operations with the contents store+    ContentsOperation(..),+    ContentsResponse(..),+    ContentsListBypass(..),++    -- * Streaming reads+    ReadRequest(..),+    ReadStream(..),+    SimpleBurst(..),+    ExtendedBurst(..),++    -- * Decoded reads+    SimplePoint(..),+    ExtendedPoint(..),++    -- * Writes+    WriteResult(..),++    -- * Hacks+    PassThrough(..),++    -- * Conversion to and from wire format+    WireFormat(fromWire, toWire),++    -- * Internal+    DayMap(..),+    TeleResp(..),+    TeleMsg(..),+    TeleMsgType(..),+    TeleMsgUOM(..),+    msgTypeUOM,+    AgentID, agentIDLength, agentID,++    -- * Convenience/clarity+    Epoch,+    NumBuckets+) where++import Vaultaire.Classes.WireFormat+import Vaultaire.Types.Address+import Vaultaire.Types.Common+import Vaultaire.Types.ContentsListBypass+import Vaultaire.Types.ContentsOperation+import Vaultaire.Types.ContentsResponse+import Vaultaire.Types.DayMap+import Vaultaire.Types.Decoded+import Vaultaire.Types.PassThrough+import Vaultaire.Types.ReadRequest+import Vaultaire.Types.ReadStream+import Vaultaire.Types.SourceDict+import Vaultaire.Types.SourceDictCache+import Vaultaire.Types.Telemetry+import Vaultaire.Types.TimeStamp+import Vaultaire.Types.WriteResult

@@ -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.+--++{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE TupleSections              #-}++module Vaultaire.Types.Address+(+    Address(..),+    calculateBucketNumber,+    isAddressExtended,+) where++import Control.Applicative+import Data.Bits+import Data.Hashable+import Data.Locator+import Data.Packer (getWord64LE, putWord64LE, runPacking, tryUnpacking)+import Data.String+import Data.Word (Word64)+import Test.QuickCheck++import Vaultaire.Classes.WireFormat++newtype Address = Address {+    unAddress :: Word64+} deriving (Eq, Ord, Hashable, Num, Bounded)++instance Read Address where+    readsPrec _ = pure . (,"") . Address . fromInteger . fromBase62++instance Show Address where+    show = padWithZeros 11 . toBase62 . toInteger . unAddress++instance IsString Address where+    fromString = fromIntegral . fromBase62++-- | There are assumptions made that the encoding of Address is fixed-length (8+-- bytes). Changing that will break things subtly.+instance WireFormat Address where+    toWire = runPacking 8 . putWord64LE . unAddress+    fromWire = tryUnpacking (Address `fmap` getWord64LE)++instance Arbitrary Address where+    arbitrary = Address <$> arbitrary++calculateBucketNumber :: Word64 -> Address -> Word64+calculateBucketNumber num_buckets (Address addr) = (addr `clearBit` 0) `mod` num_buckets++isAddressExtended :: Address -> Bool+isAddressExtended (Address addr) = addr `testBit` 0+

@@ -0,0 +1,67 @@+--+-- 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 GeneralizedNewtypeDeriving #-}+{-# LANGUAGE TupleSections              #-}++module Vaultaire.Types.Common+(+    Origin(..),+    makeOrigin,+    Epoch,+    NumBuckets+) where++import Control.Applicative+import Control.Exception (Exception, SomeException (..))+import Data.ByteString (ByteString)+import qualified Data.ByteString.Char8 as S+import Data.Char+import Data.Either+import Data.Hashable (Hashable)+import Data.Locator+import Data.String (IsString)+import Data.Typeable (Typeable)+import Data.Word (Word64)+import Test.QuickCheck++-- |Origin is a ByteString representing a data origin. It must be+--  between one and eight bytes.+newtype Origin = Origin { unOrigin :: ByteString }+    deriving (Eq, Ord, IsString, Hashable)++instance Arbitrary Origin where+    -- suchThat condition should be removed once locators package is fixed+    arbitrary = Origin . S.pack . toLocator16a 6 <$> arbitrary `suchThat` (>0)++instance Read Origin where+    readsPrec _ = fmap (,"") . rights . (:[]) . makeOrigin . S.pack++instance Show Origin where+    show = S.unpack . unOrigin++-- | Invalid origin Exception+data BadOrigin = NullOrigin | NonAlphaNumOrigin | OriginTooLong+    deriving (Show, Typeable)++instance Exception BadOrigin++makeOrigin :: ByteString -> Either SomeException Origin+makeOrigin bs+    | S.null bs = Left (SomeException NullOrigin)+    | S.any (not . isAlphaNum) bs = Left (SomeException NonAlphaNumOrigin)+    | S.length bs > 8 = Left (SomeException OriginTooLong)+    | otherwise = Right (Origin bs)++-- These can all be newtype wrapped as make work, perhaps excluding DayMap.+-- They have no reason to be inter-mixed.++type Epoch = Word64+type NumBuckets = Word64

@@ -0,0 +1,48 @@+--+-- 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.+--++-- | This module provides a ContentsListBypass type which is required to+-- somewhat subvert the type system we use to ensure that anything passed to+-- 'reply' is a valid WireFormat instance.+--+-- In the case of reading contents lists from Ceph, we already have a valid+-- wireformat bytestring. It is just missing the Address. So given an (Address,+-- ByteString) tuple, we can transform it one-way into a ContentsListBypass in+-- such a way that:+--+-- forall SourceDict s.+-- forall Address a.+-- toWire (ContentsListBypass (a, toWire s)) == toWire (ContentsListEntry a s)+--+-- The point of this is to avoid re-parsing on egress from disk.+module Vaultaire.Types.ContentsListBypass+(+    ContentsListBypass(..)+) where++import Data.ByteString (ByteString)+import qualified Data.ByteString as S+import Data.Packer (putBytes, putWord64LE, putWord8, runPacking)+import Vaultaire.Classes.WireFormat+import Vaultaire.Types.Address++data ContentsListBypass = ContentsListBypass Address ByteString++instance WireFormat ContentsListBypass where+    -- There is no fromWire, you want to use ContentsListEntry's fromWire as+    -- the wire representation is identical. This is the whole point.+    fromWire = error "fromWire for ContentsListBypass is not implemented"+    toWire (ContentsListBypass a s) =+        let addr_bytes = toWire a+            len = S.length s+        in runPacking (len + 17) $ do+                putWord8 0x2+                putBytes addr_bytes+                putWord64LE $ fromIntegral len+                putBytes s

@@ -0,0 +1,70 @@+--+-- 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 OverloadedStrings #-}+module Vaultaire.Types.ContentsOperation+(+    ContentsOperation(..),+    SourceDict,+) where++import Control.Applicative ((<$>), (<*>))+import qualified Data.ByteString as S+import Data.Packer (getBytes, getWord64LE, getWord8, putBytes, putWord64LE,+                    putWord8, runPacking, tryUnpacking)+import Test.QuickCheck+import Vaultaire.Classes.WireFormat+import Vaultaire.Types.Address+import Vaultaire.Types.SourceDict (SourceDict)++data ContentsOperation = ContentsListRequest+                       | GenerateNewAddress+                       | UpdateSourceTag Address SourceDict+                       | RemoveSourceTag Address SourceDict+  deriving (Show, Eq)++instance WireFormat ContentsOperation where+    fromWire bs = flip tryUnpacking bs $ do+        header <- getWord8+        case header of+            0x0 -> return ContentsListRequest+            0x1 -> return GenerateNewAddress+            0x2 -> UpdateSourceTag <$> getAddr <*> getSourceDict+            0x3 -> RemoveSourceTag <$> getAddr <*> getSourceDict+            _   -> fail "Illegal op code"+      where+        getAddr = Address <$> getWord64LE+        getSourceDict = do+            len <- fromIntegral <$> getWord64LE+            fromWire <$> getBytes len >>= either (fail . show) return++    toWire op =+        case op of+            ContentsListRequest   -> "\x00"+            GenerateNewAddress    -> "\x01"+            UpdateSourceTag addr dict -> sourceOpToWire 0x2 addr dict+            RemoveSourceTag addr dict -> sourceOpToWire 0x3 addr dict+      where+        sourceOpToWire header (Address addr) dict =+            let dict_bytes = toWire dict in+            let dict_len = S.length dict_bytes in+            runPacking (17 + dict_len) $ do+                putWord8 header+                putWord64LE addr+                putWord64LE (fromIntegral dict_len)+                putBytes dict_bytes++instance Arbitrary ContentsOperation where+    arbitrary = oneof [ return ContentsListRequest+                      , return GenerateNewAddress+                      , UpdateSourceTag <$> arbitrary <*> arbitrary+                      , RemoveSourceTag <$> arbitrary <*> arbitrary ]

@@ -0,0 +1,85 @@+--+-- 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 OverloadedStrings #-}++module Vaultaire.Types.ContentsResponse+(+    ContentsResponse(..),+) where++import Control.Applicative ((<$>), (<*>))+import Control.Exception (SomeException (..))+import qualified Data.ByteString as S+import Data.Packer (getBytes, getWord64LE, putBytes, putWord64LE, putWord8,+                    runPacking, runUnpacking)+import Test.QuickCheck+import Vaultaire.Classes.WireFormat+import Vaultaire.Types.Address+import Vaultaire.Types.SourceDict++data ContentsResponse = RandomAddress Address+                      | InvalidContentsOrigin+                      | ContentsListEntry Address SourceDict+                      | EndOfContentsList+                      | UpdateSuccess+                      | RemoveSuccess+  deriving (Show, Eq)+++instance WireFormat ContentsResponse where+    fromWire bs+        | bs == "\x00" = Right InvalidContentsOrigin+        | bs == "\x03" = Right EndOfContentsList+        | bs == "\x04" = Right UpdateSuccess+        | bs == "\x05" = Right RemoveSuccess+        | S.take 1 bs == "\x01" = RandomAddress <$> fromWire (S.drop 1 bs)+        -- This relies on address being fixed-length when encoded+        | S.take 1 bs == "\x02" = do+            let body = S.drop 1 bs+            let unpacker = (,) <$> getBytes 8 <*> (getWord64LE >>= getBytes . fromIntegral)+            let (addr_bytes, dict_bytes ) = runUnpacking unpacker body+            ContentsListEntry <$> fromWire addr_bytes <*> fromWire dict_bytes++        | otherwise = Left $ SomeException $+                        userError "Invalid ContentsResponse packet"++    toWire InvalidContentsOrigin = "\x00"+    toWire (RandomAddress addr)  = "\x01" `S.append` toWire addr+    toWire (ContentsListEntry addr dict) =+        let addr_bytes = toWire addr+            dict_bytes = toWire dict+            dict_len = S.length dict_bytes+        in runPacking (dict_len + 17) $ do+            putWord8 0x2+            putBytes addr_bytes+            putWord64LE $ fromIntegral dict_len+            putBytes dict_bytes++--  Be aware there is also case such that:+--  toWire (ContentsListBypass addr b) =+--      "\x02" ...+--  so that raw encoded bytes stored on disk can be tunnelled through. See+--  Vaultaire.Types.ContentsListBypass for details++    toWire EndOfContentsList = "\x03"+    toWire UpdateSuccess     = "\x04"+    toWire RemoveSuccess     = "\x05"++instance Arbitrary ContentsResponse where+    arbitrary = oneof [ RandomAddress  <$> arbitrary+                      , ContentsListEntry <$> arbitrary <*> arbitrary+                      , return EndOfContentsList+                      , return UpdateSuccess+                      , return RemoveSuccess ]++

@@ -0,0 +1,71 @@+--+-- 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 SD licence.+--++{-# LANGUAGE GeneralizedNewtypeDeriving #-}++module Vaultaire.Types.DayMap+(+    DayMap(..)+) where++import Control.Applicative+import Control.Exception+import Control.Monad+import Data.ByteString (ByteString)+import qualified Data.ByteString as S+import Data.List+import Data.Map.Strict (Map)+import qualified Data.Map.Strict as Map+import Data.Monoid+import Data.Packer+import Test.QuickCheck+import Vaultaire.Classes.WireFormat+import Vaultaire.Types.Common+++newtype DayMap = DayMap { unDayMap :: Map Epoch NumBuckets }+    deriving (Monoid, Eq)++instance Show DayMap where+    show = intercalate "\n"+         . map (\(k,v) -> show k ++ "," ++ show v)+         . Map.toAscList+         . unDayMap++instance Arbitrary DayMap where+    -- Valid first entry followed by whatever+    arbitrary =+        DayMap . Map.fromList . ((0, 128):) <$> arbitrary++instance WireFormat DayMap where+    fromWire bs+        | S.null bs =+            Left . toException . userError $ "empty daymap file"+        | S.length bs `rem` 16 /= 0 =+            Left . toException . userError $ "corrupt contents, should be multiple of 16"+        | otherwise =+            let loaded = mustLoadDayMap bs+                (first, _) = Map.findMin (unDayMap loaded)+            in if first == 0+                then Right loaded+                else Left . toException . userError $ "bad first entry, must start at zero."++    toWire (DayMap m)+        | Map.null m = error "cannot toWire empty DayMap"+        | otherwise =+            runPacking (Map.size m * 16) $+                forM_ (Map.toAscList m)+                      (\(k,v) -> putWord64LE k >> putWord64LE v)+++mustLoadDayMap :: ByteString -> DayMap+mustLoadDayMap =+    DayMap . Map.fromList . runUnpacking parse+  where+    parse = many $ (,) <$> getWord64LE <*> getWord64LE

@@ -0,0 +1,33 @@+module Vaultaire.Types.Decoded+  ( SimplePoint(..)+  , ExtendedPoint(..)+  ) where++import Data.Word (Word64)+import Data.ByteString (ByteString)+import Vaultaire.Types.Address+import Vaultaire.Types.TimeStamp++-- | SimplePoints are simply wrapped packets for Vaultaire+-- Each consists of 24 bytes:+-- An 8 byte Address+-- An 8 byte Timestamp (nanoseconds since Unix epoch)+-- An 8 byte Payload+data SimplePoint = SimplePoint { simpleAddress :: {-# UNPACK #-} !Address+                               , simpleTime    :: {-# UNPACK #-} !TimeStamp+                               , simplePayload :: {-# UNPACK #-} !Word64 }+  deriving (Show, Eq)+++-- | ExtendedPoints are simply wrapped packets for Vaultaire+-- Each consists of 16 + 'length' bytes:+-- An 8 byte Address+-- An 8 byte Time (in nanoseconds since Unix epoch)+-- A 'length' byte Payload+-- On the wire their equivalent representation takes up+-- 24 + 'length' bytes with format:+-- 8 byte Address, 8 byte Time, 8 byte Length, Payload+data ExtendedPoint = ExtendedPoint { extendedAddress :: Address+                                   , extendedTime    :: TimeStamp+                                   , extendedPayload :: ByteString }+  deriving (Show, Eq)

@@ -0,0 +1,28 @@+--+-- 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.+--++module Vaultaire.Types.PassThrough+(+    PassThrough(..)+) where++import Control.Applicative+import Data.ByteString (ByteString, pack)+import Test.QuickCheck+import Vaultaire.Classes.WireFormat++newtype PassThrough = PassThrough { unPassThrough :: ByteString }+  deriving (Eq, Show)++instance WireFormat PassThrough where+    toWire = unPassThrough+    fromWire = Right . PassThrough++instance Arbitrary PassThrough where+    arbitrary = PassThrough . pack <$> arbitrary

@@ -0,0 +1,74 @@+--+-- 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 #-}++module Vaultaire.Types.ReadRequest+(+    ReadRequest(..),+) where++import Control.Applicative+import Control.Exception+import Data.ByteString (ByteString)+import qualified Data.ByteString as S+import Data.Packer (getBytes, getWord64LE, getWord8, putBytes, putWord64LE,+                    putWord8, runPacking, tryUnpacking)+import Data.Word (Word8)+import Test.QuickCheck+import Text.Printf+import Vaultaire.Classes.WireFormat+import Vaultaire.Types.Address+import Vaultaire.Types.TimeStamp++data ReadRequest = SimpleReadRequest Address TimeStamp TimeStamp+                 | ExtendedReadRequest Address TimeStamp TimeStamp+  deriving (Eq)++-- For use in debugging output in the reader daemon. Could use TimeStamp's Show+-- instance, but that's in ISO8601, and we just want a Unix timestamp here.+instance Show ReadRequest where+    show (SimpleReadRequest   addr start end) = show addr ++ " (s) " ++ format start ++ " to " ++ format end+    show (ExtendedReadRequest addr start end) = show addr ++ " (e) " ++ format start ++ " to " ++ format end++format :: TimeStamp -> String+format (TimeStamp t) = printf "%010d" (t `div` 1000000000)++instance WireFormat ReadRequest where+    toWire (SimpleReadRequest addr start end) =+        packWithHeaderByte 0 addr start end+    toWire (ExtendedReadRequest addr start end) =+        packWithHeaderByte 1 addr start end++    fromWire bs+        | S.length bs /= 25 = Left . SomeException $ userError "expected 25 bytes"+        | otherwise = flip tryUnpacking bs $ do+            header <- getWord8+            addr_bytes <- getBytes 8+            addr <- either (fail . show) return $ fromWire addr_bytes+            start <- TimeStamp <$> getWord64LE+            end <- TimeStamp <$> getWord64LE+            case header of+                0 -> return $ SimpleReadRequest addr start end+                1 -> return $ ExtendedReadRequest addr start end+                _ -> fail "invalid header byte"++packWithHeaderByte :: Word8 -> Address -> TimeStamp -> TimeStamp -> ByteString+packWithHeaderByte header addr start end =+    let addr_bytes = toWire addr+    in runPacking 25 $ do+        putWord8 header+        putBytes addr_bytes+        putWord64LE (unTimeStamp start)+        putWord64LE (unTimeStamp end)++instance Arbitrary ReadRequest where+    arbitrary =+        oneof [ SimpleReadRequest <$> arbitrary <*> arbitrary <*> arbitrary+              , ExtendedReadRequest <$> arbitrary <*> arbitrary <*> arbitrary ]

@@ -0,0 +1,58 @@+{-# LANGUAGE OverloadedStrings #-}+--+-- 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 SD licence.+--++module Vaultaire.Types.ReadStream+(+    ReadStream(..),+    ExtendedBurst(..),+    SimpleBurst(..),+) where++import Control.Applicative+import Control.Exception (SomeException (..))+import Data.ByteString (ByteString)+import qualified Data.ByteString as S+import Test.QuickCheck+import Vaultaire.Classes.WireFormat++newtype ExtendedBurst = ExtendedBurst { unExtendedBurst :: ByteString }+  deriving (Show, Eq)++newtype SimpleBurst = SimpleBurst { unSimpleBurst :: ByteString }+  deriving (Show, Eq)++data ReadStream = InvalidReadOrigin+                | SimpleStream { unSimpleStream :: SimpleBurst }+                | ExtendedStream { unExtendedStream :: ExtendedBurst }+                | EndOfStream+  deriving (Show, Eq)++instance WireFormat ReadStream where+    fromWire bs+        | bs == "\x00" = Right InvalidReadOrigin+        | bs == "\x01" = Right EndOfStream+        | S.take 1 bs == "\x02" =+            Right . SimpleStream . SimpleBurst $ S.drop 1 bs+        | S.take 1 bs == "\x03" =+            Right . ExtendedStream. ExtendedBurst $ S.drop 1 bs+        | otherwise = Left $ SomeException $ userError "Invalid ReadStream packet"+    toWire InvalidReadOrigin = "\x00"+    toWire EndOfStream = "\x01"+    toWire (SimpleStream (SimpleBurst bs)) = "\x02" `S.append` bs+    toWire (ExtendedStream (ExtendedBurst bs)) = "\x03" `S.append` bs++instance Arbitrary ReadStream where+    arbitrary = oneof [ return InvalidReadOrigin+                      , SimpleStream . SimpleBurst . S.pack <$> arbitrary+                      , ExtendedStream . ExtendedBurst . S.pack <$> arbitrary+                      , return EndOfStream ]

@@ -0,0 +1,101 @@+--+-- 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 GeneralizedNewtypeDeriving #-}+{-# LANGUAGE OverloadedStrings          #-}++module Vaultaire.Types.SourceDict+(+    SourceDict,+    unionSource,+    diffSource,+    lookupSource,+    hashSource,+    makeSourceDict+) where++import Blaze.ByteString.Builder (fromByteString, toByteString)+import Blaze.ByteString.Builder.Char8 (fromChar)+import Control.Applicative (many, optional, (<$>), (<*), (<*>))+import Control.Arrow ((***))+import Control.Exception (SomeException (..))+import Crypto.MAC.SipHash+import Data.Attoparsec.Text (parseOnly)+import qualified Data.Attoparsec.Text as PT+import Data.HashMap.Strict (HashMap, difference, foldlWithKey', fromList,+                            lookup, toList, union)+import Data.List (sortBy)+import Data.Maybe (isNothing)+import Data.Monoid (Monoid, mempty, (<>))+import Data.Ord (comparing)+import Data.Serialize+import Data.Text (Text, find, pack, unpack)+import Data.Text.Encoding (decodeUtf8', encodeUtf8)+import Data.Word+import Prelude hiding (lookup)+import Test.QuickCheck+import Vaultaire.Classes.WireFormat++newtype SourceDict = SourceDict { unSourceDict :: HashMap Text Text }+  deriving (Eq, Monoid)++makeSourceDict :: HashMap Text Text -> Either String SourceDict+makeSourceDict hm = if foldlWithKey' allGoodKV True hm+                    then Right $ SourceDict hm+                    else Left "Bad character in source dict,\+                              \ no ',' or ':' allowed."+  where allGoodKV acc k v = acc && (allGoodChars k && allGoodChars v)+        allGoodChars = isNothing . find (\c -> c == ':' || c == ',')++instance Show SourceDict where+  show (SourceDict sd) = "dict=" <> show (toList sd)++instance WireFormat SourceDict where+    fromWire bs = either (Left . SomeException) parse (decodeUtf8' bs)+      where+        parse t = either (Left . SomeException . userError)+                         (Right . SourceDict . fromList)+                         (parseOnly tagParser t)++        tagParser = many $ (,) <$> k <*> v+          where+            k = PT.takeWhile (/= ':') <* ":"+            v = PT.takeWhile (/= ',') <* optional ","++    toWire = toByteString . foldlWithKey' f mempty . unSourceDict+      where+        f acc k v = acc <> text k <> fromChar ':' <> text v <> fromChar ','+        text = fromByteString . encodeUtf8++instance Arbitrary SourceDict where+    arbitrary = do+        attempt <- fromList . map (pack *** pack) <$> arbitrary+        either (const arbitrary) return $ makeSourceDict attempt++-- | Wrapped HashMap.union for SourceDicts+unionSource :: SourceDict -> SourceDict -> SourceDict+unionSource (SourceDict a) (SourceDict b) = SourceDict $ union a b++-- | Wrapped HashMap.difference for SourceDicts+diffSource :: SourceDict -> SourceDict -> SourceDict+diffSource (SourceDict a) (SourceDict b) = SourceDict $ difference a b++-- | Wrapped HashMap.lookup for SourceDicts+lookupSource :: Text -> SourceDict -> Maybe Text+lookupSource key sd = lookup key $ unSourceDict sd++-- | Hashes the sourcedict using SipHash+-- Hashes are used primarily to avoid redundant updates+hashSource :: SourceDict -> Word64+hashSource (SourceDict sd) =+    let canonicalList = sortBy (comparing fst) (map (unpack *** unpack) $ toList sd) in+    let (SipHash ret) = hash (SipKey 0 0) (encode canonicalList) in+    ret

@@ -0,0 +1,38 @@+module Vaultaire.Types.SourceDictCache+(+    SourceDictCache,+    emptySourceCache,+    insertSourceCache,+    memberSourceCache,+    sizeOfSourceCache+) where++import Control.Applicative+import Data.Packer+import Data.Set (Set)+import qualified Data.Set as Set+import Data.Word++import Vaultaire.Classes.WireFormat++newtype SourceDictCache = SourceDictCache {+    unSourceDictCache :: Set Word64+}++instance WireFormat SourceDictCache where+    toWire (SourceDictCache sdc) =+        let elems = Set.elems sdc in+        runPacking (length elems * 8) (mapM putWord64 elems)+    fromWire bs = Right $ SourceDictCache $ Set.fromList $ runUnpacking (many getWord64) bs++emptySourceCache :: SourceDictCache+emptySourceCache = SourceDictCache Set.empty++insertSourceCache :: Word64 -> SourceDictCache -> SourceDictCache+insertSourceCache x (SourceDictCache sdc) = SourceDictCache (Set.insert x sdc)++memberSourceCache :: Word64 -> SourceDictCache -> Bool+memberSourceCache x (SourceDictCache sdc) = Set.member x sdc++sizeOfSourceCache :: SourceDictCache -> Int+sizeOfSourceCache (SourceDictCache sdc) = Set.size sdc

@@ -0,0 +1,213 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+module Vaultaire.Types.Telemetry+     ( TeleResp(..)+     , TeleMsg(..)+     , TeleMsgType(..)+     , TeleMsgUOM(..)+     , msgTypeUOM+     , AgentID, agentIDLength, agentID )+where++import Control.Applicative+import Control.Exception+import Control.Monad+import Data.ByteString (ByteString)+import qualified Data.ByteString.Char8 as S+import Data.Monoid+import Data.Packer+import Data.Word+import Test.QuickCheck++import Vaultaire.Classes.WireFormat+import Vaultaire.Types.Common+import Vaultaire.Types.TimeStamp++-- | ID string associated with a running daemon, so telemetry messages+--   can be associated with the processes which sent them.+newtype AgentID = AgentID String+        deriving (Eq, Ord, Monoid)++-- | Response for a telemetry request, sent by the profiler to clients.+data TeleResp = TeleResp+     { _timestamp :: TimeStamp+     , _aid       :: AgentID+     , _msg       :: TeleMsg+     } deriving Eq++-- | The actual telemetric data, reported by Vaultaire worker threads+--   to their profiler.+--+data TeleMsg = TeleMsg+     { _origin  :: Origin+     , _type    :: TeleMsgType+     , _payload :: Word64+     } deriving Eq++-- | Telemetry types. All counts are absolute and all latencies are in milliseconds.+data TeleMsgType+   = WriterSimplePoints       -- ^ Total number of simple points written since last message+   | WriterExtendedPoints     -- ^ Total number of extended points written since last message+   | WriterRequest            -- ^ Total number of write requests received since last message+   | WriterRequestLatency     -- ^ Mean latency for one request+   | WriterCephLatency        -- ^ Mean Ceph latency for one request+   | ReaderSimplePoints       -- ^ Total number of simple points read since last message+   | ReaderExtendedPoints     -- ^ Total number of extended points read since last message+   | ReaderRequest            -- ^ Total number of read requests received since last message+   | ReaderRequestLatency     -- ^ Mean latency for one request+   | ReaderCephLatency        -- ^ Mean Ceph latency for one request+   | ContentsEnumerate        -- ^ Total number of enumerate requests received since last message+   | ContentsUpdate           -- ^ Total number of update requests received since last message+   | ContentsEnumerateLatency -- ^ Mean latency for one enumerate request+   | ContentsUpdateLatency    -- ^ Mean latency for one update request+   | ContentsEnumerateCeph    -- ^ Mean Ceph latency for one enumerate request+   | ContentsUpdateCeph       -- ^ Mean Ceph latency for one update request+   deriving (Enum, Bounded, Eq, Ord)++data TeleMsgUOM+    = Points+    | Requests+    | Milliseconds+    deriving (Enum, Bounded, Eq, Ord)++instance Show TeleMsgUOM where+  show Points       = "points"+  show Requests     = "requests"+  show Milliseconds = "ms"++-- | Map a telemetry message type onto its associated UOM.+msgTypeUOM :: TeleMsgType -> TeleMsgUOM+msgTypeUOM WriterSimplePoints       = Points+msgTypeUOM WriterExtendedPoints     = Points+msgTypeUOM WriterRequest            = Requests+msgTypeUOM WriterRequestLatency     = Milliseconds+msgTypeUOM WriterCephLatency        = Milliseconds+msgTypeUOM ReaderSimplePoints       = Points+msgTypeUOM ReaderExtendedPoints     = Points+msgTypeUOM ReaderRequest            = Requests+msgTypeUOM ReaderRequestLatency     = Milliseconds+msgTypeUOM ReaderCephLatency        = Milliseconds+msgTypeUOM ContentsEnumerate        = Requests+msgTypeUOM ContentsUpdate           = Requests+msgTypeUOM ContentsEnumerateLatency = Milliseconds+msgTypeUOM ContentsUpdateLatency    = Milliseconds+msgTypeUOM ContentsEnumerateCeph    = Milliseconds+msgTypeUOM ContentsUpdateCeph       = Milliseconds++-- | Return (possibly empty) prefix component of a ByteString terminated+--   by one or more null bytes.+chomp :: ByteString -> ByteString+chomp = S.takeWhile (/='\0')++-- | Agent IDs are a maximum of 64 bytes.+agentIDLength :: Int+agentIDLength = 64++-- | An agent ID has to fit in 64 characters and does not contain \NUL.+agentID :: String -> Maybe AgentID+agentID s | length s <= agentIDLength && notElem '\0' s+          = Just $ AgentID s+          | otherwise = Nothing++putAgentID :: AgentID -> Packing ()+putAgentID (AgentID x)+  = putBytes $ S.pack $ x ++ replicate (agentIDLength - length x) '\0'++getAgentID :: Unpacking AgentID+getAgentID = AgentID . S.unpack . chomp <$> getBytes agentIDLength++-- | Pack a telemetry message. Assumes the origin is no longer than+--   eight bytes.+putTeleMsg :: TeleMsg -> Packing ()+putTeleMsg x = do+    -- 8 bytes for the origin.+    let o = unOrigin $ _origin x+    putBytes    $ S.append o $ S.pack $ replicate (8 - S.length o) '\0'+    -- 8 bytes for the message type.+    putWord64LE $ fromIntegral $ fromEnum $ _type x+    -- 8 bytes for the payload+    putWord64LE $ _payload x++getTeleMsg :: Unpacking (Either SomeException TeleMsg)+getTeleMsg = do+    o <- makeOrigin . chomp <$> getBytes 8+    t <- toEnum . fromIntegral <$> getWord64LE+    p <- getWord64LE+    return $ fmap (\org -> TeleMsg org t p) o++instance WireFormat AgentID where+  toWire   = runPacking agentIDLength . putAgentID+  fromWire = tryUnpacking    getAgentID++instance WireFormat TeleMsg where+  toWire   = runPacking 24 . putTeleMsg+  fromWire = runUnpacking getTeleMsg++instance WireFormat TeleResp where+  toWire x = runPacking 96 $ do+    -- 8 bytes for the timestamp+    putWord64LE $ unTimeStamp $ _timestamp x+    -- 64 bytes for the agent ID, padded out with nuls+    putAgentID  $ _aid x+    -- 24 bytes for the message (origin, type and payload)+    putTeleMsg  $ _msg x+  fromWire x = join $ flip tryUnpacking x $ do+    s <- TimeStamp <$> getWord64LE+    a <- getAgentID+    m <- getTeleMsg+    return $ TeleResp s a <$> m+++instance Arbitrary TeleMsg where+  arbitrary =   TeleMsg+            <$> arbitrary+            <*> arbitrary+            <*> arbitrary++instance Arbitrary TeleResp where+  arbitrary =   TeleResp+            <$> arbitrary+            <*> arbitrary+            <*> arbitrary++instance Arbitrary TeleMsgType where+  arbitrary = arbitraryBoundedEnum++instance Arbitrary AgentID where+  arbitrary = untilG agentID arbitrary+    where untilG :: (Arbitrary a, Arbitrary b) => (a -> Maybe b) -> Gen a -> Gen b+          untilG f a = a >>= maybe arbitrary return . f+++instance Show AgentID where+  show (AgentID s) = s++instance Show TeleMsgType where+  show WriterSimplePoints       = "writer-count-simple-point      "+  show WriterExtendedPoints     = "writer-count-extended-point    "+  show WriterRequest            = "writer-count-request           "+  show WriterRequestLatency     = "writer-latency-request         "+  show WriterCephLatency        = "writer-latency-ceph            "+  show ReaderSimplePoints       = "reader-count-simple-point      "+  show ReaderExtendedPoints     = "reader-count-extended-point    "+  show ReaderRequest            = "reader-count-request           "+  show ReaderRequestLatency     = "reader-latency-request         "+  show ReaderCephLatency        = "reader-latency-ceph            "+  show ContentsEnumerate        = "contents-count-enumerate       "+  show ContentsUpdate           = "contents-count-update          "+  show ContentsEnumerateLatency = "contents-latency-enumerate     "+  show ContentsUpdateLatency    = "contents-latency-update        "+  show ContentsEnumerateCeph    = "contents-latency-ceph-enumerate"+  show ContentsUpdateCeph       = "contents-latency-ceph-update   "++instance Show TeleResp where+  show r = concat [ show $ _timestamp r, " "+                  , show $ _aid r,       " "+                  , show $ _msg r]++instance Show TeleMsg where+  show m = concat [ show $ _origin m, " "+                  , show $ _type m,   " "+                  , let s = show (fromIntegral $ _payload m :: Int)+                    in  replicate (8 - length s) ' ' ++ s+                  , " "+                  , show $ msgTypeUOM $ _type m ]

@@ -0,0 +1,125 @@+--+-- 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 GeneralizedNewtypeDeriving #-}+{-# LANGUAGE TupleSections              #-}++module Vaultaire.Types.TimeStamp+(+    TimeStamp(..),+    convertToDiffTime,+    convertToTimeStamp,+    getCurrentTimeNanoseconds+) where++import Control.Applicative+import Data.Maybe+import Data.Packer (getWord64LE, putWord64LE, runPacking, tryUnpacking)+import Data.Time.Calendar+import Data.Time.Clock+import Data.Time.Clock.POSIX+import Data.Time.Format+import Data.Word (Word64)+import System.Locale+import Test.QuickCheck++import Vaultaire.Classes.WireFormat++--+-- | Number of nanoseconds since the Unix epoch, stored in a Word64.+--+-- The Show instance displays the TimeStamp as seconds with the nanosecond+-- precision expressed as a decimal amount after the interger, ie:+--+-- >>> t <- getCurrentTimeNanoseconds+-- >>> show t+-- 2014-07-31T23:09:35.274387000Z+--+-- However this doesn't change the fact the underlying representation counts+-- nanoseconds since epoch:+--+-- >>> show $ unTimeStamp t+-- 1406848175274387000+--+-- There is a Read instance that is reasonably accommodating.+--+-- >>> read "2014-07-31T13:05:04.942089001Z" ::TimeStamp+-- 2014-07-31T13:05:04.942089001Z+--+-- >>> read "1406811904.942089001" :: TimeStamp+-- 2014-07-31T13:05:04.942089001Z+--+-- >>> read "1406811904" :: TimeStamp+-- 2014-07-31T13:05:04.000000000Z+--+newtype TimeStamp = TimeStamp {+    unTimeStamp :: Word64+} deriving (Eq, Ord, Enum, Arbitrary, Num, Real, Integral, Bounded)++instance Show TimeStamp where+    show (TimeStamp t) =+      let+        seconds = posixSecondsToUTCTime $ realToFrac (fromIntegral t / 1000000000 :: Rational)+        iso8601 = formatTime defaultTimeLocale "%FT%T.%q" seconds+      in+        -- trim to nanoseconds+        take 29 iso8601 ++ "Z"++instance Read TimeStamp where+    readsPrec _ s = maybeToList $ (,"") <$> convertToTimeStamp <$> parse s+      where+        parse :: String -> Maybe UTCTime+        parse x =   parseTime defaultTimeLocale "%FT%T%Q%Z" x+                <|> parseTime defaultTimeLocale "%F" x+                <|> parseTime defaultTimeLocale "%s%Q" x++instance WireFormat TimeStamp where+    toWire = runPacking 8 . putWord64LE . unTimeStamp+    fromWire = tryUnpacking (TimeStamp `fmap` getWord64LE)++--+-- | Utility function to convert nanoseconds since Unix epoch to a+-- 'NominalDiffTime', allowing you to then use the time manipulation+-- functions in "Data.Time.Clock"+--+convertToDiffTime :: TimeStamp -> NominalDiffTime+convertToDiffTime = fromRational . (/ 1e9) . fromIntegral++--+-- | Get the current system time, expressed as a 'TimeStamp' (which is to+-- say, number of nanoseconds since the Unix epoch).+--+{-+    getPOSIXTime returns a NominalDiffTime with picosecond precision. So+    convert it to nanoseconds, and discard any remaining fractional amount.+-}+getCurrentTimeNanoseconds :: IO TimeStamp -- Word64+getCurrentTimeNanoseconds = do+    u <- getCurrentTime+    return $ convertToTimeStamp u++{-+    This code adapted from the implementation in Data.Time.Clock.POSIX. The+    time types in base are hopeless. Julian days? Really? We'll replace this+    with hs-hourglass shortly.+-}++secondsPerDay :: Integer+secondsPerDay = 86400++unixEpochDay :: Day+unixEpochDay = ModifiedJulianDay 40587++convertToTimeStamp :: UTCTime -> TimeStamp+convertToTimeStamp (UTCTime day secs) =+  let+    mark = diffDays day unixEpochDay * secondsPerDay * 1000000000+    nano = floor $ (*1000000000) $ toRational secs+  in+    TimeStamp $ fromIntegral $ mark + nano

@@ -0,0 +1,35 @@+{-# LANGUAGE OverloadedStrings #-}+--+-- 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.+--++module Vaultaire.Types.WriteResult+(+    WriteResult(..),+) where++import Control.Exception (SomeException (..))+import Test.QuickCheck+import Vaultaire.Classes.WireFormat++data WriteResult = InvalidWriteOrigin | OnDisk+  deriving (Show, Eq)++instance WireFormat WriteResult where+    fromWire bs+        | bs == "\x00" = Right OnDisk+        | bs == "\x01" = Right InvalidWriteOrigin+        | otherwise = Left $ SomeException $ userError "Invalid WriteResult packet"+    toWire OnDisk = "\x00"+    toWire InvalidWriteOrigin = "\x01"++instance Arbitrary WriteResult where+    arbitrary = oneof [ return InvalidWriteOrigin, return OnDisk ]

@@ -0,0 +1,24 @@+module Vaultaire.Util+(+    linkThread,+    waitForever,+    fatal+) where++import Control.Concurrent+import Control.Concurrent.Async+import Control.Monad+import Control.Monad.IO.Class+import System.IO.Unsafe (unsafePerformIO)+import System.Log.Logger++linkThread :: MonadIO m => IO a -> m ()+linkThread = liftIO . (async >=> link)++waitForever :: MonadIO m => m ()+waitForever = liftIO (threadDelay maxBound) >> waitForever++fatal :: String -> String -> a+fatal component err =+    unsafePerformIO (criticalM component err) `seq`+        error ("fatal error in " ++ component ++ ": " ++ err)

@@ -0,0 +1,39 @@+--+-- 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.+--++-- | Test serialisation/deserialiastion for Vaultaire types.++import Test.Hspec+import Vaultaire.Types++main :: IO ()+main = hspec suite++suite :: Spec+suite =+  describe "Round trip through Read and Show instances" $ do+    it "outputs a correctly formated ISO 8601 timestamp when Shown" $ do+      show (TimeStamp 1406849015948797001) `shouldBe` "2014-07-31T23:23:35.948797001Z"+      show (TimeStamp 1406849015948797001) `shouldBe` "2014-07-31T23:23:35.948797001Z"+      show (TimeStamp 0) `shouldBe` "1970-01-01T00:00:00.000000000Z"++    it "Reads ISO 8601 timestamps" $ do+      read "2014-07-31T23:23:35.948797001Z" `shouldBe` TimeStamp 1406849015948797001+      read "2014-07-31T23:23:35Z" `shouldBe` TimeStamp 1406849015000000000+      read "2014-07-31" `shouldBe` TimeStamp 1406764800000000000++    it "reads the Unix epoch date" $+      read "1970-01-01" `shouldBe` TimeStamp 0++    it "permissively reads various formats" $ do+      show (read "1970-01-01T00:00:00.000000000Z" :: TimeStamp) `shouldBe` "1970-01-01T00:00:00.000000000Z"+      show (read "1970-01-01" :: TimeStamp) `shouldBe` "1970-01-01T00:00:00.000000000Z"+      show (read "0" :: TimeStamp) `shouldBe` "1970-01-01T00:00:00.000000000Z"

@@ -0,0 +1,71 @@+--+-- 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 OverloadedStrings #-}++module Main where++import Control.Exception (throw)+import Data.HashMap.Strict (fromList)+import Test.Hspec+import Test.Hspec.QuickCheck+import Vaultaire.Types++main :: IO ()+main = hspec suite++suite :: Spec+suite = do+    describe "WireFormat identity tests" $ do+        prop "Daymap" (wireId :: DayMap -> Bool)+        prop "ContentsOperation" (wireId :: ContentsOperation -> Bool)+        prop "ContentsResponse" (wireId :: ContentsResponse -> Bool)+        prop "WriteResult" (wireId :: WriteResult -> Bool)+        prop "ReadStream" (wireId :: ReadStream -> Bool)+        prop "Address" (wireId :: Address -> Bool)+        prop "SourceDict" (wireId :: SourceDict -> Bool)+        prop "ReadRequest" (wireId :: ReadRequest -> Bool)+        prop "PassThrough" (wireId :: PassThrough -> Bool)+        prop "TelemetryResponse" (wireId :: TeleResp -> Bool)++    describe "source dict wire format" $+        it "parses string to map" $ do+            let hm =  fromList [ ("metric","cpu")+                            , ("server","example")]+            let expected = either error id $ makeSourceDict hm+            let wire = either throw id $ fromWire "server:example,metric:cpu"+            wire `shouldBe` expected++    describe "Contents list bypass" $ do+        prop "has same wire format for all" contentsListBypassId+        it "has same wire format for known case" $ do+            let al = [("metric","cpu"), ("server","www.example.com")]+            let (Right source_dict) = makeSourceDict $ fromList al+            let encoded = toWire source_dict+            let expected = "\x02\+                           \\x01\x00\x00\x00\x00\x00\x00\x00\+                           \\x22\x00\x00\x00\x00\x00\x00\x00\+                           \metric:cpu,server:www.example.com,"++            toWire (ContentsListEntry 1 source_dict) `shouldBe` expected+            toWire (ContentsListBypass 1 encoded) `shouldBe` expected++contentsListBypassId :: SourceDict -> Address -> Bool+contentsListBypassId dict addr = toWire entry == toWire bypass+  where+    entry = ContentsListEntry addr dict+    bypass = ContentsListBypass addr (toWire dict)++wireId :: (Eq w, WireFormat w) => w -> Bool+wireId op = id' op == op+  where+    id' = fromRight . fromWire . toWire+    fromRight = either (error . show) id

@@ -0,0 +1,111 @@+cabal-version:       >= 1.10+name:                vaultaire-common+version:             2.9.1+synopsis:            Common types and instances for Vaultaire+description:         Defines a set of types, typeclasses and instances for+                     Vaultaire, intended for use with Marquise and other+                     Vaultaire related libraries and executables+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+license-file:        LICENSE+tested-with:         GHC == 7.8.3+stability:           experimental++build-type:          Simple++source-repository    head+  type:              git+  location:          git@github.com:anchor/vaultaire-common.git+++library+  hs-source-dirs:    lib+  default-language:  Haskell2010++  exposed-modules:   Vaultaire.Program,+                     Vaultaire.Types,+                     Vaultaire.Util++  other-modules:     Vaultaire.Classes.WireFormat,+                     Vaultaire.Types.Address,+                     Vaultaire.Types.ContentsOperation,+                     Vaultaire.Types.ContentsResponse,+                     Vaultaire.Types.SourceDict,+                     Vaultaire.Types.SourceDictCache,+                     Vaultaire.Types.ReadStream,+                     Vaultaire.Types.ReadRequest,+                     Vaultaire.Types.Common,+                     Vaultaire.Types.WriteResult,+                     Vaultaire.Types.ContentsListBypass,+                     Vaultaire.Types.DayMap,+                     Vaultaire.Types.Decoded,+                     Vaultaire.Types.TimeStamp,+                     Vaultaire.Types.PassThrough,+                     Vaultaire.Types.Telemetry++  build-depends:     base >=3 && <5,+                     bytestring,+                     QuickCheck,+                     packer,+                     async,+                     transformers,+                     unix,+                     hslogger,+                     hashable,+                     containers,+                     unordered-containers,+                     blaze-builder,+                     attoparsec,+                     text,+                     time,+                     old-locale,+                     locators >= 0.2.4,+                     siphash,+                     cereal+  ghc-options:       -Wall+                     -Wwarn+                     -fwarn-tabs++  ghc-prof-options:  -fprof-auto++test-suite           wire-formats-test+  hs-source-dirs:    tests+  main-is:           WireFormatsTest.hs+  type:              exitcode-stdio-1.0+  default-language:  Haskell2010++  build-depends:     base >=4.7 && <5,+                     hspec,+                     containers,+                     unordered-containers,+                     text,+                     locators >= 0.2.4,+                     QuickCheck,+                     mtl,+                     bytestring,+                     vaultaire-common++  ghc-options:       -threaded+                     -Wall+                     -Wwarn+                     -fwarn-tabs++test-suite           instance-test+  hs-source-dirs:    tests+  main-is:           InstanceTest.hs+  type:              exitcode-stdio-1.0+  default-language:  Haskell2010++  build-depends:     base >=4.7 && <5,+                     hspec,+                     vaultaire-common++  ghc-options:       -threaded+                     -Wall+                     -Wwarn+                     -fwarn-tabs++-- vim: set tabstop=21 expandtab: