diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/lib/Vaultaire/Classes/WireFormat.hs b/lib/Vaultaire/Classes/WireFormat.hs new file mode 100644 --- /dev/null +++ b/lib/Vaultaire/Classes/WireFormat.hs @@ -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" diff --git a/lib/Vaultaire/Program.hs b/lib/Vaultaire/Program.hs new file mode 100644 --- /dev/null +++ b/lib/Vaultaire/Program.hs @@ -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 diff --git a/lib/Vaultaire/Types.hs b/lib/Vaultaire/Types.hs new file mode 100644 --- /dev/null +++ b/lib/Vaultaire/Types.hs @@ -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 diff --git a/lib/Vaultaire/Types/Address.hs b/lib/Vaultaire/Types/Address.hs new file mode 100644 --- /dev/null +++ b/lib/Vaultaire/Types/Address.hs @@ -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 + diff --git a/lib/Vaultaire/Types/Common.hs b/lib/Vaultaire/Types/Common.hs new file mode 100644 --- /dev/null +++ b/lib/Vaultaire/Types/Common.hs @@ -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 diff --git a/lib/Vaultaire/Types/ContentsListBypass.hs b/lib/Vaultaire/Types/ContentsListBypass.hs new file mode 100644 --- /dev/null +++ b/lib/Vaultaire/Types/ContentsListBypass.hs @@ -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 diff --git a/lib/Vaultaire/Types/ContentsOperation.hs b/lib/Vaultaire/Types/ContentsOperation.hs new file mode 100644 --- /dev/null +++ b/lib/Vaultaire/Types/ContentsOperation.hs @@ -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 ] diff --git a/lib/Vaultaire/Types/ContentsResponse.hs b/lib/Vaultaire/Types/ContentsResponse.hs new file mode 100644 --- /dev/null +++ b/lib/Vaultaire/Types/ContentsResponse.hs @@ -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 ] + + diff --git a/lib/Vaultaire/Types/DayMap.hs b/lib/Vaultaire/Types/DayMap.hs new file mode 100644 --- /dev/null +++ b/lib/Vaultaire/Types/DayMap.hs @@ -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 diff --git a/lib/Vaultaire/Types/Decoded.hs b/lib/Vaultaire/Types/Decoded.hs new file mode 100644 --- /dev/null +++ b/lib/Vaultaire/Types/Decoded.hs @@ -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) diff --git a/lib/Vaultaire/Types/PassThrough.hs b/lib/Vaultaire/Types/PassThrough.hs new file mode 100644 --- /dev/null +++ b/lib/Vaultaire/Types/PassThrough.hs @@ -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 diff --git a/lib/Vaultaire/Types/ReadRequest.hs b/lib/Vaultaire/Types/ReadRequest.hs new file mode 100644 --- /dev/null +++ b/lib/Vaultaire/Types/ReadRequest.hs @@ -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 ] diff --git a/lib/Vaultaire/Types/ReadStream.hs b/lib/Vaultaire/Types/ReadStream.hs new file mode 100644 --- /dev/null +++ b/lib/Vaultaire/Types/ReadStream.hs @@ -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 ] diff --git a/lib/Vaultaire/Types/SourceDict.hs b/lib/Vaultaire/Types/SourceDict.hs new file mode 100644 --- /dev/null +++ b/lib/Vaultaire/Types/SourceDict.hs @@ -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 diff --git a/lib/Vaultaire/Types/SourceDictCache.hs b/lib/Vaultaire/Types/SourceDictCache.hs new file mode 100644 --- /dev/null +++ b/lib/Vaultaire/Types/SourceDictCache.hs @@ -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 diff --git a/lib/Vaultaire/Types/Telemetry.hs b/lib/Vaultaire/Types/Telemetry.hs new file mode 100644 --- /dev/null +++ b/lib/Vaultaire/Types/Telemetry.hs @@ -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 ] diff --git a/lib/Vaultaire/Types/TimeStamp.hs b/lib/Vaultaire/Types/TimeStamp.hs new file mode 100644 --- /dev/null +++ b/lib/Vaultaire/Types/TimeStamp.hs @@ -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 diff --git a/lib/Vaultaire/Types/WriteResult.hs b/lib/Vaultaire/Types/WriteResult.hs new file mode 100644 --- /dev/null +++ b/lib/Vaultaire/Types/WriteResult.hs @@ -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 ] diff --git a/lib/Vaultaire/Util.hs b/lib/Vaultaire/Util.hs new file mode 100644 --- /dev/null +++ b/lib/Vaultaire/Util.hs @@ -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) diff --git a/tests/InstanceTest.hs b/tests/InstanceTest.hs new file mode 100644 --- /dev/null +++ b/tests/InstanceTest.hs @@ -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" diff --git a/tests/WireFormatsTest.hs b/tests/WireFormatsTest.hs new file mode 100644 --- /dev/null +++ b/tests/WireFormatsTest.hs @@ -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 diff --git a/vaultaire-common.cabal b/vaultaire-common.cabal new file mode 100644 --- /dev/null +++ b/vaultaire-common.cabal @@ -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 +maintainer: Anchor Engineering +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: