packages feed

influxdb (empty) → 0.0.0

raw patch · 12 files changed

+1493/−0 lines, 12 filesdep +aesondep +attoparsecdep +basesetup-changed

Dependencies added: aeson, attoparsec, base, bytestring, containers, data-default-class, dlist, http-client, influxdb, lifted-base, mtl, mwc-random, network, retry, scientific, tagged, text, time, vector

Files

+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) 2014, Mitsutoshi Aoe++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++    * Redistributions of source code must retain the above copyright+      notice, this list of conditions and the following disclaimer.++    * 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.++    * Neither the name of Mitsutoshi Aoe nor the names of other+      contributors may be used to endorse or promote products derived+      from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ examples/random-points.hs view
@@ -0,0 +1,132 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeSynonymInstances #-}+{-# LANGUAGE ViewPatterns #-}+{-# OPTIONS_GHC -fno-warn-orphans #-}+import Control.Applicative+import Control.Exception as E+import Control.Monad+import Control.Monad.Trans+import Data.Function (fix)+import Data.Time.Clock.POSIX+import System.Environment+import System.IO+import qualified Data.Text as T+import qualified Data.Vector as V++import System.Random.MWC (Variate(..))+import qualified Network.HTTP.Client as HC+import qualified System.Random.MWC as MWC++import Database.InfluxDB++oneWeekInSeconds :: Int+oneWeekInSeconds = 7*24*60*60++main :: IO ()+main = do+  [read -> (numPoints :: Int), read -> (batches :: Int)] <- getArgs+  hSetBuffering stdout NoBuffering+  config <- newConfig+  HC.withManager managerSettings $ \manager -> do+    dropDatabase config manager (Database "ctx" Nothing)+      `E.catch`+        -- Ignore exceptions here+        \(_ :: HC.HttpException) -> return ()+    db <- createDatabase config manager "ctx"+    gen <- MWC.create+    flip fix batches $ \outerLoop !m -> when (m > 0) $ do+      postWithPrecision config manager db SecondsPrecision $ withSeries "ct1" $+        flip fix numPoints $ \innerLoop !n -> when (n > 0) $ do+          !timestamp <- liftIO $ (-)+            <$> getPOSIXTime+            <*> (fromIntegral <$> uniformR (0, oneWeekInSeconds) gen)+          !value <- liftIO $ uniform gen+          writePoints $ Point value timestamp+          innerLoop $ n - 1+      outerLoop $ m - 1++    result <- query config manager db "select count(value) from ct1;"+    case result of+      [] -> putStrLn "Empty series"+      series:_ -> do+        print $ seriesColumns series+        print $ seriesPoints series+    -- Streaming output+    queryChunked config manager db "select * from ct1;" $ \stream0 ->+      flip fix stream0 $ \loop stream -> case stream of+        Done -> return ()+        Yield series next -> do+          case fromSeriesData series of+            Left reason -> hPutStrLn stderr reason+            Right points -> mapM_ print (points :: [Point])+          putStrLn "--"+          stream' <- next+          loop stream'++newConfig :: IO Config+newConfig = do+  pool <- newServerPool localServer [] -- no backup servers+  return Config+    { configCreds = rootCreds+    , configServerPool = pool+    }++managerSettings :: HC.ManagerSettings+managerSettings = HC.defaultManagerSettings+  { HC.managerResponseTimeout = Just $ 60*(10 :: Int)^(6 :: Int)+  }++data Point = Point !Name !POSIXTime deriving Show++instance ToSeriesData Point where+  toSeriesColumns _ = V.fromList ["value", "time"]+  toSeriesPoints (Point value time) = V.fromList+    [ toValue value+    , epochInSeconds time+    ]++instance FromSeriesData Point where+  parseSeriesData = withValues $ \values -> Point+    <$> values .: "value"+    <*> values .: "time"++epochInSeconds :: POSIXTime -> Value+epochInSeconds = Int . floor++data Name+  = Foo+  | Bar+  | Baz+  | Quu+  | Qux+  deriving (Enum, Bounded, Show)++instance ToValue Name where+  toValue Foo = String "foo"+  toValue Bar = String "bar"+  toValue Baz = String "baz"+  toValue Quu = String "quu"+  toValue Qux = String "qux"++instance FromValue Name where+  parseValue (String name) = case name of+    "foo" -> return Foo+    "bar" -> return Bar+    "baz" -> return Baz+    "quu" -> return Quu+    "qux" -> return Qux+    _ -> fail $ "Incorrect string: " ++ T.unpack name+  parseValue v = typeMismatch "String" v++instance Variate Name where+  uniform = uniformR (minBound, maxBound)+  uniformR (lower, upper) g = do+    name <- uniformR (fromEnum lower, fromEnum upper) g+    return $! toEnum name++instance FromValue POSIXTime where+  parseValue (Int n) = return $ fromIntegral n+  parseValue (Float d) = return $ realToFrac d+  parseValue v = typeMismatch "Int or Float" v
+ influxdb.cabal view
@@ -0,0 +1,81 @@+name: influxdb+version: 0.0.0+synopsis: Haskell client library for InfluxDB+description: Haskell client library for InfluxDB+homepage: https://github.com/maoe/influxdb-haskell+license: BSD3+license-file: LICENSE+author: Mitsutoshi Aoe+maintainer: Mitsutoshi Aoe <maoe@foldr.in>+copyright: Copyright (C) 2014 Mitsutoshi Aoe+category: Database+build-type: Simple+cabal-version: >= 1.10++flag examples+  default: False++library+  exposed-modules:+    Database.InfluxDB+    Database.InfluxDB.Decode+    Database.InfluxDB.Encode+    Database.InfluxDB.Http+    Database.InfluxDB.Lens+    -- Database.InfluxDB.Protobuf+    -- Database.InfluxDB.Query+    Database.InfluxDB.Stream+    Database.InfluxDB.Types+  other-modules:+    Database.InfluxDB.Types.Internal+  ghc-options: -Wall+  build-depends:+      base >= 4 && < 4.8+    , aeson >= 0.7+    , attoparsec+    , bytestring+    , containers+    , data-default-class+    , dlist+    , http-client+    , lifted-base+    , mtl+    , network+    , retry+    , scientific+    , tagged+    , text+    , time+    , vector+  hs-source-dirs: src+  default-language: Haskell2010++executable influx-random-points+  if flag(examples)+    buildable: True+  else+    buildable: False+  hs-source-dirs: examples+  main-is: random-points.hs+  ghc-options: -Wall+  build-depends:+      base+    , bytestring+    , http-client+    , influxdb+    , mtl+    , mwc-random+    , text+    , time+    , vector+  default-language: Haskell2010++source-repository head+  type: git+  branch: develop+  location: https://github.com/maoe/influxdb-haskell.git++source-repository this+  type: git+  tag: v0.0.0+  location: https://github.com/maoe/influxdb-haskell.git
+ src/Database/InfluxDB.hs view
@@ -0,0 +1,77 @@+module Database.InfluxDB+  (+  -- * Series data types+    Series(..), seriesColumns, seriesPoints+  , SeriesData(..)+  , Value(..)++  -- ** Encoding+  , ToSeriesData(..)+  , ToValue(..)++  -- ** Decoding+  , FromSeries(..), fromSeries+  , FromSeriesData(..), fromSeriesData+  , FromValue(..), fromValue++  , withValues, (.:)+  , typeMismatch++  -- * HTTP API+  -- ** Data types+  , Config(..)+  , Credentials(..), rootCreds+  , TimePrecision(..)+  , Server(..), localServer+  , ServerPool, newServerPool+  , Database(..)+  , User(..)+  , Admin(..)++  -- ** Writing Data++  -- *** Updating Points+  , post, postWithPrecision+  , SeriesT, PointT+  , writeSeries+  , withSeries+  , writePoints++  -- *** Deleting Points+  -- **** One Time Deletes (not implemented)+  -- , deleteSeries+  -- **** Regularly Scheduled Deletes (not implemented)+  -- , getScheduledDeletes+  -- , addScheduledDelete+  -- , removeScheduledDelete++  -- ** Querying Data+  , query+  , Stream(..)+  , queryChunked++  -- ** Administration & Security+  -- *** Creating and Dropping Databases+  , listDatabases+  , createDatabase+  , dropDatabase++  -- *** Security+  -- **** Cluster admin+  , listClusterAdmins+  , addClusterAdmin+  , updateClusterAdminPassword+  , deleteClusterAdmin+  -- **** Database user+  , listDatabaseUsers+  , addDatabaseUser+  , updateDatabaseUserPassword+  , deleteDatabaseUser+  , grantAdminPrivilegeTo+  , revokeAdminPrivilegeFrom+  ) where++import Database.InfluxDB.Decode+import Database.InfluxDB.Encode+import Database.InfluxDB.Http+import Database.InfluxDB.Types
+ src/Database/InfluxDB/Decode.hs view
@@ -0,0 +1,135 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE RecordWildCards #-}+module Database.InfluxDB.Decode+  ( FromSeries(..), fromSeries+  , FromSeriesData(..), fromSeriesData+  , withValues, (.:)+  , FromValue(..), fromValue+  , Parser, ValueParser, typeMismatch+  ) where+import Control.Applicative+import Control.Monad.Reader+import Data.Map (Map)+import Data.Vector (Vector)+import Data.Tuple (swap)+import qualified Data.DList as DL+import qualified Data.Map as Map+import qualified Data.Text as T+import qualified Data.Vector as V++import Database.InfluxDB.Types++-- | A type that can be converted from a @Series@.+class FromSeries a where+  parseSeries :: Series -> Parser a++instance FromSeries Series where+  parseSeries = return++instance FromSeries SeriesData where+  parseSeries = return . seriesData++-- | Converte a value from a @Series@, failing if the types do not match.+fromSeries :: FromSeries a => Series -> Either String a+fromSeries = runParser . parseSeries++-- | A type that can be converted from a @SeriesData@. A typical implementation+-- is as follows.+--+-- > import Control.Applicative ((<$>), (<*>))+-- > import qualified Data.Vector as V+-- >+-- > data Event = Event Text EventType+-- > data EventType = Login | Logout+-- >+-- > instance FromSeriesData where+-- >   parseSeriesData = withValues $ values -> Event+-- >     <$> values .: "user"+-- >     <*> values .: "type"+-- >+-- > instance FromValue EventType+class FromSeriesData a where+  parseSeriesData :: Vector Column -> Vector Value -> Parser a++instance FromSeriesData SeriesData where+  parseSeriesData columns values = return SeriesData+    { seriesDataColumns = columns+    , seriesDataPoints = DL.singleton values+    }++-- | Converte a value from a @SeriesData@, failing if the types do not match.+fromSeriesData :: FromSeriesData a => SeriesData -> Either String [a]+fromSeriesData SeriesData {..} = mapM+  (runParser . parseSeriesData seriesDataColumns)+  (DL.toList seriesDataPoints)++withValues+  :: (Vector Value -> ValueParser a)+  -> Vector Column -> Vector Value -> Parser a+withValues f columns values =+  runReaderT m $ Map.fromList $ map swap $ V.toList $ V.indexed columns+  where+    ValueParser m = f values++(.:) :: FromValue a => Vector Value -> Column -> ValueParser a+values .: column = do+  found <- asks $ Map.lookup column+  case found of+    Nothing -> fail $ "No such column: " ++ T.unpack column+    Just idx -> do+      value <- V.indexM values idx+      liftParser $ parseValue value++-- | A type that can be converted from a @Value@.+class FromValue a where+  parseValue :: Value -> Parser a++-- | Converte a value from a @Value@, failing if the types do not match.+fromValue :: FromValue a => Value -> Either String a+fromValue = runParser . parseValue++instance FromValue Value where+  parseValue = return++instance FromValue Bool where+  parseValue (Bool b) = return b+  parseValue v = typeMismatch "Bool" v++instance FromValue a => FromValue (Maybe a) where+  parseValue Null = return Nothing+  parseValue v = Just <$> parseValue v++instance FromValue Int where+  parseValue (Int n) = return $ fromIntegral n+  parseValue v = typeMismatch "Int" v++instance FromValue Double where+  parseValue (Float d) = return d+  parseValue v = typeMismatch "Float" v++typeMismatch+  :: String+  -> Value+  -> Parser a+typeMismatch expected actual = fail $+  "when expecting a " ++ expected +++  ", encountered " ++ name ++ " instead"+  where+    name = case actual of+      Int _ -> "Int"+      Float _ -> "Float"+      String _ -> "String"+      Bool _ -> "Bool"+      Null -> "Null"++newtype Parser a = Parser+  { runParser :: Either String a+  } deriving (Functor, Applicative, Monad)++type ColumnIndex = Map Column Int++newtype ValueParser a = ValueParser (ReaderT ColumnIndex Parser a)+  deriving (Functor, Applicative, Monad, MonadReader ColumnIndex)++liftParser :: Parser a -> ValueParser a+liftParser = ValueParser . ReaderT . const
+ src/Database/InfluxDB/Encode.hs view
@@ -0,0 +1,96 @@+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeSynonymInstances #-}+module Database.InfluxDB.Encode+  ( ToSeries(..)+  , ToSeriesData(..), toSeriesData+  , ToValue(..)+  ) where+import Data.Int (Int8, Int16, Int32, Int64)+import Data.Proxy+import Data.Vector (Vector)+import Data.Word (Word8, Word16, Word32)+import qualified Data.DList as DL+import qualified Data.Text as T+import qualified Data.Text.Lazy as TL++import Database.InfluxDB.Types++-- | A type that can be converted to a @Series@.+class ToSeries a where+  toSeries :: a -> Series++-- | A type that can be converted to a @SeriesData@. A typical implementation is+-- as follows.+--+-- > import qualified Data.Vector as V+-- >+-- > data Event = Event Text EventType+-- > data EventType = Login | Logout+-- >+-- > instance ToSeriesData where+-- >   toSeriesColumn _ = V.fromList ["user", "type"]+-- >   toSeriesPoints (Event user ty) = V.fromList [toValue user, toValue ty]+-- >+-- > instance ToValue EventType+class ToSeriesData a where+  -- | Column names. You can safely ignore the proxy agument.+  toSeriesColumns :: Proxy a -> Vector Column+  -- | Data points.+  toSeriesPoints :: a -> Vector Value++toSeriesData :: forall a. ToSeriesData a => a -> SeriesData+toSeriesData a = SeriesData+  { seriesDataColumns = toSeriesColumns (Proxy :: Proxy a)+  , seriesDataPoints = DL.singleton (toSeriesPoints a)+  }++-- | A type that can be stored in InfluxDB.+class ToValue a where+  toValue :: a -> Value++instance ToValue Bool where+  toValue = Bool++instance ToValue a => ToValue (Maybe a) where+  toValue Nothing = Null+  toValue (Just a) = toValue a++instance ToValue Int where+  toValue = Int . fromIntegral++instance ToValue Int8 where+  toValue = Int . fromIntegral++instance ToValue Int16 where+  toValue = Int . fromIntegral++instance ToValue Int32 where+  toValue = Int . fromIntegral++instance ToValue Int64 where+  toValue = Int++instance ToValue Word8 where+  toValue = Int . fromIntegral++instance ToValue Word16 where+  toValue = Int . fromIntegral++instance ToValue Word32 where+  toValue = Int . fromIntegral++instance ToValue Float where+  toValue = Float . realToFrac++instance ToValue Double where+  toValue = Float++instance ToValue T.Text where+  toValue = String++instance ToValue TL.Text where+  toValue = String . TL.toStrict++instance ToValue String where+  toValue = String . T.pack
+ src/Database/InfluxDB/Http.hs view
@@ -0,0 +1,659 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}+module Database.InfluxDB.Http+  ( Config(..)+  , Credentials(..), rootCreds+  , Server(..), localServer+  , TimePrecision(..)++  -- * Writing Data++  -- ** Updating Points+  , post, postWithPrecision+  , SeriesT, PointT+  , writeSeries+  , withSeries+  , writePoints++  -- ** Deleting Points+  -- *** One Time Deletes (not implemented)+  -- , deleteSeries+  -- *** Regularly Scheduled Deletes (not implemented)+  -- , getScheduledDeletes+  -- , addScheduledDelete+  -- , removeScheduledDelete++  -- * Querying Data+  , query+  , Stream(..)+  , queryChunked++  -- * Administration & Security+  -- ** Creating and Dropping Databases+  , listDatabases+  , createDatabase+  , dropDatabase++  -- ** Security+  -- *** Cluster admin+  , listClusterAdmins+  , addClusterAdmin+  , updateClusterAdminPassword+  , deleteClusterAdmin+  -- *** Database user+  , listDatabaseUsers+  , addDatabaseUser+  , updateDatabaseUserPassword+  , deleteDatabaseUser+  , grantAdminPrivilegeTo+  , revokeAdminPrivilegeFrom+  ) where++import Control.Applicative+import Control.Monad.Identity+import Control.Monad.Writer+import Data.DList (DList)+import Data.IORef (IORef)+import Data.Proxy+import Data.Text (Text)+import Data.Vector (Vector)+import Network.URI (escapeURIString, isAllowedInURI)+import Text.Printf (printf)+import qualified Data.ByteString as BS+import qualified Data.ByteString.Char8 as BS8+import qualified Data.ByteString.Lazy as BL+import qualified Data.DList as DL+import qualified Data.Text as T++import Control.Exception.Lifted (Handler(..))+import Control.Retry+import Data.Aeson ((.=))+import Data.Default.Class (Default(def))+import qualified Data.Aeson as A+import qualified Data.Aeson.Encode as AE+import qualified Data.Attoparsec as P+import qualified Network.HTTP.Client as HC++import Database.InfluxDB.Decode+import Database.InfluxDB.Encode+import Database.InfluxDB.Types+import Database.InfluxDB.Stream (Stream(..))+import qualified Database.InfluxDB.Stream as S++-- | Configurations for HTTP API client.+data Config = Config+  { configCreds :: !Credentials+  , configServerPool :: IORef ServerPool+  }++-- | Default credentials.+rootCreds :: Credentials+rootCreds = Credentials+  { credsUser = "root"+  , credsPassword = "root"+  }++-- | Default server location.+localServer :: Server+localServer = Server+  { serverHost = "localhost"+  , serverPort = 8086+  , serverSsl = False+  }++data TimePrecision+  = SecondsPrecision+  | MillisecondsPrecision+  | MicrosecondsPrecision++timePrecChar :: TimePrecision -> Char+timePrecChar SecondsPrecision = 's'+timePrecChar MillisecondsPrecision = 'm'+timePrecChar MicrosecondsPrecision = 'u'++-----------------------------------------------------------+-- Writing Data++-- | Post a bunch of writes for (possibly multiple) series into a database.+post+  :: Config+  -> HC.Manager+  -> Database+  -> SeriesT IO a+  -> IO a+post config manager database =+  postGeneric config manager database Nothing++-- | Post a bunch of writes for (possibly multiple) series into a database like+-- @post@ but with time precision.+postWithPrecision+  :: Config+  -> HC.Manager+  -> Database+  -> TimePrecision+  -> SeriesT IO a+  -> IO a+postWithPrecision config manager database timePrec =+  postGeneric config manager database (Just timePrec)++postGeneric+  :: Config+  -> HC.Manager+  -> Database+  -> Maybe TimePrecision+  -> SeriesT IO a+  -> IO a+postGeneric Config {..} manager database timePrec write = do+  (a, series) <- runSeriesT write+  void $ httpLbsWithRetry configServerPool (makeRequest series) manager+  return a+  where+    makeRequest series = def+      { HC.method = "POST"+      , HC.requestBody = HC.RequestBodyLBS $ AE.encode series+      , HC.path = escapeString $ printf "/db/%s/series"+          (T.unpack databaseName)+      , HC.queryString = escapeString $ printf "u=%s&p=%s%s"+          (T.unpack credsUser)+          (T.unpack credsPassword)+          (maybe+            ""+            (printf "&time_precision=%c" . timePrecChar)+            timePrec :: String)+      }+    Database {databaseName} = database+    Credentials {..} = configCreds++-- | Monad transformer to batch up multiple writes of series to speed up+-- insertions.+newtype SeriesT m a = SeriesT (WriterT (DList Series) m a)+  deriving+    ( Functor, Applicative, Monad, MonadIO, MonadTrans+    , MonadWriter (DList Series)+    )++-- | Monad transformer to batch up multiple writes of points to speed up+-- insertions.+newtype PointT p m a = PointT (WriterT (DList (Vector Value)) m a)+  deriving+    ( Functor, Applicative, Monad, MonadIO, MonadTrans+    , MonadWriter (DList (Vector Value))+    )++runSeriesT :: Monad m => SeriesT m a -> m (a, [Series])+runSeriesT (SeriesT w) = do+  (a, series) <- runWriterT w+  return (a, DL.toList series)++-- | Write a single series data.+writeSeries+  :: (Monad m, ToSeriesData a)+  => Text+  -- ^ Series name+  -> a+  -- ^ Series data+  -> SeriesT m ()+writeSeries name a = tell . DL.singleton $ Series+  { seriesName = name+  , seriesData = toSeriesData a+  }++-- | Write a bunch of data for a single series. Columns for the points don't+-- need to be specified because they can be inferred from the type of @a@.+withSeries+  :: forall m a. (Monad m, ToSeriesData a)+  => Text+  -- ^ Series name+  -> PointT a m ()+  -> SeriesT m ()+withSeries name (PointT w) = do+  (_, values) <- lift $ runWriterT w+  tell $ DL.singleton Series+    { seriesName = name+    , seriesData = SeriesData+        { seriesDataColumns = toSeriesColumns (Proxy :: Proxy a)+        , seriesDataPoints = values+        }+    }++-- | Write a data into a series.+writePoints+  :: (Monad m, ToSeriesData a)+  => a+  -> PointT a m ()+writePoints = tell . DL.singleton . toSeriesPoints++-- TODO: Delete API hasn't been implemented in InfluxDB yet+--+-- deleteSeries+--   :: Config+--   -> HC.Manager+--   -> Series+--   -> IO ()+-- deleteSeries Config {..} manager =+--   error "deleteSeries: not implemented"+--+-- getScheduledDeletes+--   :: Config+--   -> HC.Manager+--   -> IO [ScheduledDelete]+-- getScheduledDeletes = do+--   error "getScheduledDeletes: not implemented"+--+-- addScheduledDelete+--   :: Config+--   -> HC.Manager+--   -> IO ScheduledDelete+-- addScheduledDelete =+--   error "addScheduledDeletes: not implemented"+--+-- removeScheduledDelete+--   :: Config+--   -> HC.Manager+--   -> ScheduledDeletes+--   -> IO ()+-- removeScheduledDelete =+--   error "removeScheduledDelete: not implemented"++-----------------------------------------------------------+-- Querying Data++-- | Query a specified database.+--+-- The query format is specified in the+-- <http://influxdb.org/docs/query_language/ InfluxDB Query Language>.+query+  :: FromSeries a+  => Config+  -> HC.Manager+  -> Database+  -> Text -- ^ Query text+  -> IO [a]+query Config {..} manager database q = do+  response <- httpLbsWithRetry configServerPool request manager+  case A.decode (HC.responseBody response) of+    Nothing -> fail $ show response+    Just xs -> case mapM fromSeries xs of+      Left reason -> fail reason+      Right ys -> return ys+  where+    request = def+      { HC.path = escapeString $ printf "/db/%s/series"+          (T.unpack databaseName)+      , HC.queryString = escapeString $ printf "u=%s&p=%s&q=%s"+          (T.unpack credsUser)+          (T.unpack credsPassword)+          (T.unpack q)+      }+    Database {databaseName} = database+    Credentials {..} = configCreds++-- | Construct streaming output+responseStream :: A.FromJSON a => HC.BodyReader -> IO (Stream IO a)+responseStream body = demandPayload $ \payload ->+  if BS.null payload+    then return Done+    else decode $ parseAsJson payload+  where+    demandPayload k = HC.brRead body >>= k+    decode (P.Done leftover value) = case A.fromJSON value of+      A.Success a -> return $ Yield a $ if BS.null leftover+        then responseStream body+        else decode $ parseAsJson leftover+      A.Error message -> fail message+    decode (P.Partial k) = demandPayload (decode . k)+    decode (P.Fail _ _ message) = fail message+    parseAsJson = P.parse A.json++-- | Query a specified database like @query@ but in a streaming fashion.+queryChunked+  :: FromSeries a+  => Config+  -> HC.Manager+  -> Database+  -> Text -- ^ Query text+  -> (Stream IO a -> IO b)+  -- ^ Action to handle the resulting stream of series+  -> IO b+queryChunked Config {..} manager database q f =+  withPool configServerPool request $ \request' ->+    HC.withResponse request' manager $+      responseStream . HC.responseBody >=> S.mapM parse >=> f+  where+    parse series = case fromSeries series of+      Left reason -> fail reason+      Right a -> return a+    request = def+      { HC.path = escapeString $ printf "/db/%s/series"+          (T.unpack databaseName)+      , HC.queryString = escapeString $ printf "u=%s&p=%s&q=%s&chunked=true"+          (T.unpack credsUser)+          (T.unpack credsPassword)+          (T.unpack q)+      }+    Database {databaseName} = database+    Credentials {..} = configCreds++-----------------------------------------------------------+-- Administration & Security++-- | List existing databases.+listDatabases :: Config -> HC.Manager -> IO [Database]+listDatabases Config {..} manager = do+  response <- httpLbsWithRetry configServerPool makeRequest manager+  case A.decode (HC.responseBody response) of+    Nothing -> fail $ show response+    Just xs -> return xs+  where+    makeRequest = def+      { HC.path = "/db"+      , HC.queryString = escapeString $ printf "u=%s&p=%s"+          (T.unpack credsUser)+          (T.unpack credsPassword)+      }+    Credentials {..} = configCreds++-- | Create a new database. Requires cluster admin privileges.+createDatabase :: Config -> HC.Manager -> Text -> IO Database+createDatabase Config {..} manager name = do+  void $ httpLbsWithRetry configServerPool makeRequest manager+  return Database+    { databaseName = name+    , databaseReplicationFactor = Nothing+    }+  where+    makeRequest = def+      { HC.method = "POST"+      , HC.requestBody = HC.RequestBodyLBS $ AE.encode $ A.object+          [ "name" .= name+          ]+      , HC.path = "/db"+      , HC.queryString = escapeString $ printf "u=%s&p=%s"+          (T.unpack credsUser)+          (T.unpack credsPassword)+      }+    Credentials {..} = configCreds++-- | Drop a database. Requires cluster admin privileges.+dropDatabase :: Config -> HC.Manager -> Database -> IO ()+dropDatabase Config {..} manager database =+  void $ httpLbsWithRetry configServerPool makeRequest manager+  where+    makeRequest = def+      { HC.method = "DELETE"+      , HC.path = escapeString $ printf "/db/%s"+          (T.unpack databaseName)+      , HC.queryString = escapeString $ printf "u=%s&p=%s"+          (T.unpack credsUser)+          (T.unpack credsPassword)+      }+    Database {databaseName} = database+    Credentials {..} = configCreds++-- | List cluster administrators.+listClusterAdmins+  :: Config+  -> HC.Manager+  -> IO [Admin]+listClusterAdmins Config {..} manager = do+  response <- httpLbsWithRetry configServerPool makeRequest manager+  case A.decode (HC.responseBody response) of+    Nothing -> fail $ show response+    Just xs -> return xs+  where+    makeRequest = def+      { HC.path = "/cluster_admins"+      , HC.queryString = escapeString $ printf "u=%s&p=%s"+          (T.unpack credsUser)+          (T.unpack credsPassword)+      }+    Credentials {..} = configCreds++-- | Add a new cluster administrator. Requires cluster admin privilege.+addClusterAdmin+  :: Config+  -> HC.Manager+  -> Text+  -> IO Admin+addClusterAdmin Config {..} manager name = do+  void $ httpLbsWithRetry configServerPool makeRequest manager+  return Admin+    { adminUsername = name+    }+  where+    makeRequest = def+      { HC.requestBody = HC.RequestBodyLBS $ AE.encode $ A.object+          [ "name" .= name+          ]+      , HC.path = "/cluster_admins"+      , HC.queryString = escapeString $ printf "u=%s&p=%s"+          (T.unpack credsUser)+          (T.unpack credsPassword)+      }+    Credentials {..} = configCreds++-- | Update a cluster administrator's password. Requires cluster admin+-- privilege.+updateClusterAdminPassword+  :: Config+  -> HC.Manager+  -> Admin+  -> Text+  -> IO ()+updateClusterAdminPassword Config {..} manager admin password =+  void $ httpLbsWithRetry configServerPool makeRequest manager+  where+    makeRequest = def+      { HC.method = "POST"+      , HC.requestBody = HC.RequestBodyLBS $ AE.encode $ A.object+          [ "password" .= password+          ]+      , HC.path = escapeString $ printf "/cluster_admins/%s"+          (T.unpack adminUsername)+      , HC.queryString = escapeString $ printf "u=%s&p=%s"+          (T.unpack credsUser)+          (T.unpack credsPassword)+      }+    Admin {adminUsername} = admin+    Credentials {..} = configCreds++-- | Delete a cluster administrator. Requires cluster admin privilege.+deleteClusterAdmin+  :: Config+  -> HC.Manager+  -> Admin+  -> IO ()+deleteClusterAdmin Config {..} manager admin =+  void $ httpLbsWithRetry configServerPool makeRequest manager+  where+    makeRequest = def+      { HC.method = "DELETE"+      , HC.path = escapeString $ printf "/cluster_admins/%s"+          (T.unpack adminUsername)+      , HC.queryString = escapeString $ printf "u=%s&p=%s"+          (T.unpack credsUser)+          (T.unpack credsPassword)+      }+    Admin {adminUsername} = admin+    Credentials {..} = configCreds++-- | List database users.+listDatabaseUsers+  :: Config+  -> HC.Manager+  -> Text+  -> IO [User]+listDatabaseUsers Config {..} manager database = do+  response <- httpLbsWithRetry configServerPool makeRequest manager+  case A.decode (HC.responseBody response) of+    Nothing -> fail $ show response+    Just xs -> return xs+  where+    makeRequest = def+      { HC.path = escapeString $ printf "/db/%s/users"+          (T.unpack database)+      , HC.queryString = escapeString $ printf "u=%s&p=%s"+          (T.unpack credsUser)+          (T.unpack credsPassword)+      }+    Credentials {..} = configCreds++-- | Add an user to the database users.+addDatabaseUser+  :: Config+  -> HC.Manager+  -> Database+  -> Text+  -> IO User+addDatabaseUser Config {..} manager database name = do+  void $ httpLbsWithRetry configServerPool makeRequest manager+  return User+    { userName = name+    }+  where+    makeRequest = def+      { HC.requestBody = HC.RequestBodyLBS $ AE.encode $ A.object+          [ "name" .= name+          ]+      , HC.path = escapeString $ printf "/db/%s/users"+          (T.unpack databaseName)+      , HC.queryString = escapeString $ printf "u=%s&p=%s"+          (T.unpack credsUser)+          (T.unpack credsPassword)+      }+    Database {databaseName} = database+    Credentials {..} = configCreds++-- | Delete an user from the database users.+deleteDatabaseUser+  :: Config+  -> HC.Manager+  -> Database+  -> User+  -> IO ()+deleteDatabaseUser config manager database user =+  void $ httpLbsWithRetry (configServerPool config) request manager+  where+    request = (makeRequestFromDatabaseUser config database user)+      { HC.method = "DELETE"+      }++-- | Update password for the database user.+updateDatabaseUserPassword+  :: Config+  -> HC.Manager+  -> Database+  -> User+  -> Text+  -> IO ()+updateDatabaseUserPassword config manager database user password =+  void $ httpLbsWithRetry (configServerPool config) request manager+  where+    request = (makeRequestFromDatabaseUser config database user)+      { HC.method = "POST"+      , HC.requestBody = HC.RequestBodyLBS $ AE.encode $ A.object+          [ "password" .= password+          ]+      }++-- | Give admin privilege to the user.+grantAdminPrivilegeTo+  :: Config+  -> HC.Manager+  -> Database+  -> User+  -> IO ()+grantAdminPrivilegeTo config manager database user =+  void $ httpLbsWithRetry (configServerPool config) request manager+  where+    request = (makeRequestFromDatabaseUser config database user)+      { HC.method = "POST"+      , HC.requestBody = HC.RequestBodyLBS $ AE.encode $ A.object+          [ "admin" .= True+          ]+      }++-- | Remove admin privilege from the user.+revokeAdminPrivilegeFrom+  :: Config+  -> HC.Manager+  -> Database+  -> User+  -> IO ()+revokeAdminPrivilegeFrom config manager database user =+  void $ httpLbsWithRetry (configServerPool config) request manager+  where+    request = (makeRequestFromDatabaseUser config database user)+      { HC.method = "POST"+      , HC.requestBody = HC.RequestBodyLBS $ AE.encode $ A.object+          [ "admin" .= False+          ]+      }++makeRequestFromDatabaseUser+  :: Config+  -> Database+  -> User+  -> HC.Request+makeRequestFromDatabaseUser Config {..} database user = def+  { HC.path = escapeString $ printf "/db/%s/users/%s"+      (T.unpack databaseName)+      (T.unpack userName)+  , HC.queryString = escapeString $ printf "u=%s&p=%s"+      (T.unpack credsUser)+      (T.unpack credsPassword)+  }+  where+    Database {databaseName} = database+    User {userName} = user+    Credentials {..} = configCreds++-----------------------------------------------------------++httpLbsWithRetry+  :: IORef ServerPool+  -> HC.Request+  -> HC.Manager+  -> IO (HC.Response BL.ByteString)+httpLbsWithRetry pool request manager =+  withPool pool request $ \request' ->+    HC.httpLbs request' manager++withPool+  :: IORef ServerPool+  -> HC.Request+  -> (HC.Request -> IO a)+  -> IO a+withPool pool request f =+  recovering defaultRetrySettings handlers $ do+    server <- activeServer pool+    f $ makeRequest server+  where+    makeRequest Server {..} = request+      { HC.host = escapeText serverHost+      , HC.port = serverPort+      , HC.secure = serverSsl+      }+    handlers =+      [ Handler $ \e -> case e of+          HC.InternalIOException _ -> do+            failover pool+            return True+          _ -> return False+      ]++defaultRetrySettings :: RetrySettings+defaultRetrySettings = RetrySettings+  { numRetries = limitedRetries 5+  , backoff = True+  , baseDelay = 50+  }++escapeText :: Text -> BS.ByteString+escapeText = escapeString . T.unpack++escapeString :: String -> BS.ByteString+escapeString = BS8.pack . escapeURIString isAllowedInURI
+ src/Database/InfluxDB/Lens.hs view
@@ -0,0 +1,26 @@+{-# LANGUAGE RankNTypes #-}+module Database.InfluxDB.Lens+  ( credentials, user, password+  ) where+import Control.Applicative+import Data.Text (Text)++import Database.InfluxDB.Http++type Lens s t a b = Functor f => (a -> f b) -> s -> f t+type Lens' s a = Lens s s a a++credentials :: Lens' Config Credentials+credentials f r = set <$> f (configCreds r)+  where+    set c = r { configCreds = c }++user :: Lens' Credentials Text+user f c = set <$> f (credsUser c)+  where+    set u = c { credsUser = u }++password :: Lens' Credentials Text+password f s = set <$> f (credsPassword s)+  where+    set p = s { credsPassword = p }
+ src/Database/InfluxDB/Stream.hs view
@@ -0,0 +1,18 @@+module Database.InfluxDB.Stream where+import Prelude hiding (mapM)++-- | Effectful stream+data Stream m a+  = Yield a (m (Stream m a))+  -- ^ Yield a value. The stream will be continued.+  | Done+  -- ^ The end of the stream.++-- | Map each element of a stream to a monadic action, evaluate these actions+-- from left to right, and collect the results as a stream.+mapM :: Monad m => (a -> m b) -> Stream m a -> m (Stream m b)+mapM _ Done = return Done+mapM f (Yield a mb) = do+  a' <- f a+  b <- mb+  return $ Yield a' (mapM f b)
+ src/Database/InfluxDB/Types.hs view
@@ -0,0 +1,222 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE TemplateHaskell #-}+module Database.InfluxDB.Types+  ( -- * Series, columns and data points+    Series(..)+  , seriesColumns+  , seriesPoints+  , SeriesData(..)+  , Column+  , Value(..)++  -- * Data types for HTTP API+  , Credentials(..)+  , Server(..)+  , Database(..)+  , ScheduledDelete(..)+  , User(..)+  , Admin(..)++  -- * Server pool+  , ServerPool+  , newServerPool+  , activeServer+  , failover+  ) where++import Control.Applicative (empty)+import Data.DList (DList)+import Data.Data (Data)+import Data.IORef+import Data.Int (Int64)+import Data.Sequence (Seq, ViewL(..), (|>))+import Data.Text (Text)+import Data.Typeable (Typeable)+import Data.Vector (Vector)+import qualified Data.DList as DL+import qualified Data.Sequence as Seq++import Data.Aeson ((.=), (.:))+import Data.Aeson.TH+import Data.Scientific+import qualified Data.Aeson as A++import Database.InfluxDB.Types.Internal (stripPrefixOptions)++-----------------------------------------------------------+-- Compatibility for older GHC++#if __GLASGOW_HASKELL__ < 706+import Control.Exception (evaluate)++atomicModifyIORef' :: IORef a -> (a -> (a, b)) -> IO b+atomicModifyIORef' ref f = do+    b <- atomicModifyIORef ref $ \x ->+      let (a, b) = f x+      in (a, a `seq` b)+    evaluate b+#endif+-----------------------------------------------------------++-- | A series consists of name, columns and points. The columns and points are+-- expressed in a separate type @SeriesData@.+data Series = Series+  { seriesName :: {-# UNPACK #-} !Text+  -- ^ Series name+  , seriesData :: {-# UNPACK #-} !SeriesData+  -- ^ Columns and data points in the series+  }++-- | Convenient accessor for columns.+seriesColumns :: Series -> Vector Column+seriesColumns = seriesDataColumns . seriesData++-- | Convenient accessor for points.+seriesPoints :: Series -> DList (Vector Value)+seriesPoints = seriesDataPoints . seriesData++instance A.ToJSON Series where+  toJSON Series {..} = A.object+    [ "name" .= seriesName+    , "columns" .= seriesDataColumns+    , "points" .= DL.toList seriesDataPoints+    ]+    where+      SeriesData {..} = seriesData++instance A.FromJSON Series where+  parseJSON (A.Object v) = do+    name <- v .: "name"+    columns <- v .: "columns"+    points <- v .: "points"+    return Series+      { seriesName = name+      , seriesData = SeriesData+          { seriesDataColumns = columns+          , seriesDataPoints = DL.fromList points+          }+      }+  parseJSON _ = empty++-- | @SeriesData@ consists of columns and points.+data SeriesData = SeriesData+  { seriesDataColumns :: Vector Column+  , seriesDataPoints :: DList (Vector Value)+  }++type Column = Text++-- | An InfluxDB value represented as a Haskell value.+data Value+  = Int !Int64+  | Float !Double+  | String !Text+  | Bool !Bool+  | Null+  deriving (Eq, Show, Data, Typeable)++instance A.ToJSON Value where+  toJSON (Int n) = A.toJSON n+  toJSON (Float d) = A.toJSON d+  toJSON (String xs) = A.toJSON xs+  toJSON (Bool b) = A.toJSON b+  toJSON Null = A.Null++instance A.FromJSON Value where+  parseJSON (A.Object o) = fail $ "Unexpected object: " ++ show o+  parseJSON (A.Array a) = fail $ "Unexpected array: " ++ show a+  parseJSON (A.String xs) = return $ String xs+  parseJSON (A.Bool b) = return $ Bool b+  parseJSON A.Null = return Null+  parseJSON (A.Number n) = return $! if base10Exponent n == 0+    then Int $ fromIntegral $ coefficient n+    else Float $ realToFrac n++-----------------------------------------------------------++-- | User credentials.+data Credentials = Credentials+  { credsUser :: !Text+  , credsPassword :: !Text+  } deriving Show++-- | Server location.+data Server = Server+  { serverHost :: !Text+  -- ^ Hostname or IP address+  , serverPort :: !Int+  , serverSsl :: !Bool+  -- ^ SSL is enabled or not in the server side+  } deriving Show++-- | Non-empty set of server locations. The active server will always be used+-- until any HTTP communications fail.+data ServerPool = ServerPool+  { serverActive :: !Server+  -- ^ Current active server+  , serverBackup :: !(Seq Server)+  -- ^ The rest of the servers in the pool.+  }++-- | Database consits of name and replication factor.+data Database = Database+  { databaseName :: !Text+  , databaseReplicationFactor :: !(Maybe Int)+  } deriving Show++newtype ScheduledDelete = ScheduledDelete+  { scheduledDeleteId :: Int+  } deriving Show++-- | User+newtype User = User+  { userName :: Text+  } deriving Show++-- | Administrator+newtype Admin = Admin+  { adminUsername :: Text+  } deriving Show+++-----------------------------------------------------------+-- Server pool manipulation++-- | Create a non-empty server pool. You must specify at least one server+-- location to create a pool.+newServerPool :: Server -> [Server] -> IO (IORef ServerPool)+newServerPool active backups = newIORef ServerPool+  { serverActive = active+  , serverBackup = Seq.fromList backups+  }++-- | Get a server from the pool.+activeServer :: IORef ServerPool -> IO Server+activeServer ref = do+  ServerPool { serverActive } <- readIORef ref+  return serverActive++-- | Move the current server to the backup pool and pick one of the backup+-- server as the new active server. Currently the scheduler works in+-- round-robin fashion.+failover :: IORef ServerPool -> IO ()+failover ref = atomicModifyIORef' ref $ \pool@ServerPool {..} ->+  case Seq.viewl serverBackup of+    EmptyL -> (pool, ())+    active :< rest -> (pool', ())+      where+        pool' = ServerPool+          { serverActive = active+          , serverBackup = rest |> serverActive+          }++-----------------------------------------------------------+-- Aeson instances++deriveFromJSON (stripPrefixOptions "database") ''Database+deriveFromJSON (stripPrefixOptions "admin") ''Admin+deriveFromJSON (stripPrefixOptions "user") ''User
+ src/Database/InfluxDB/Types/Internal.hs view
@@ -0,0 +1,15 @@+module Database.InfluxDB.Types.Internal+  ( stripPrefixOptions+  ) where+import Data.Char (toLower)++import Data.Aeson.TH++stripPrefixOptions :: String -> Options+stripPrefixOptions name = defaultOptions+  { fieldLabelModifier = modifier+  }+  where+    modifier xs = case drop (length name) xs of+      [] -> error "Insufficient length of field name"+      c:cs -> toLower c:cs