summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorMitsutoshiAoe <>2017-03-03 02:17:00 (GMT)
committerhdiff <hdiff@hdiff.luite.com>2017-03-03 02:17:00 (GMT)
commitdeb9d3b1fc87150ebcf73215c369114648c9e223 (patch)
tree18e5fd56ce129ff8152332f637223f32dfed830f
parent1a20351ac288260a9c115e807eb0383d27e8b896 (diff)
version 1.0.01.0.0
-rw-r--r--CHANGELOG.md8
-rw-r--r--README.md9
-rw-r--r--examples/random-points.hs158
-rw-r--r--examples/write-udp.hs36
-rw-r--r--influxdb.cabal120
-rw-r--r--src/Database/InfluxDB.hs169
-rw-r--r--src/Database/InfluxDB/Decode.hs232
-rw-r--r--src/Database/InfluxDB/Encode.hs95
-rw-r--r--src/Database/InfluxDB/Format.hs108
-rw-r--r--src/Database/InfluxDB/Http.hs792
-rw-r--r--src/Database/InfluxDB/JSON.hs207
-rw-r--r--src/Database/InfluxDB/Lens.hs66
-rw-r--r--src/Database/InfluxDB/Line.hs124
-rw-r--r--src/Database/InfluxDB/Manage.hs158
-rw-r--r--src/Database/InfluxDB/Ping.hs101
-rw-r--r--src/Database/InfluxDB/Query.hs367
-rw-r--r--src/Database/InfluxDB/Stream.hs41
-rw-r--r--src/Database/InfluxDB/TH.hs141
-rw-r--r--src/Database/InfluxDB/Types.hs483
-rw-r--r--src/Database/InfluxDB/Types/Internal.hs49
-rw-r--r--src/Database/InfluxDB/Write.hs209
-rw-r--r--src/Database/InfluxDB/Write/UDP.hs87
-rw-r--r--src/Network/HTTP/Client/Compat.hs16
-rw-r--r--tests/test-suite.hs461
24 files changed, 1843 insertions, 2394 deletions
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 186fa93..cb37654 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,11 @@
+## TBD
+
+## v1.0.0 - 2017-03-03
+
+The library was completely rewritten and support for older InfluxDB has been dropped.
+
+* Support for InfluxDB 1.2
+
## v0.10.0 - 2016-05-17
* Fix a typo in a Haddock comment (#28)
diff --git a/README.md b/README.md
index f8c4097..f61d2d9 100644
--- a/README.md
+++ b/README.md
@@ -1,10 +1,11 @@
Haskell client library for InfluxDB
==========
-[![Build Status](https://travis-ci.org/maoe/influxdb-haskell.svg?branch=develop)](https://travis-ci.org/maoe/influxdb-haskell?branch=develop)
-[![Coverage Status](https://coveralls.io/repos/maoe/influxdb-haskell/badge.png?branch=develop)](https://coveralls.io/r/maoe/influxdb-haskell?branch=develop)
-[![Gitter chat](https://badges.gitter.im/maoe/influxdb-haskell.png)](https://gitter.im/maoe/influxdb-haskell)
+[![Build Status](https://travis-ci.org/maoe/influxdb-haskell.svg?branch=master)](https://travis-ci.org/maoe/influxdb-haskell)
+[![Hackage](https://img.shields.io/hackage/v/influxdb.svg)](https://hackage.haskell.org/package/influxdb)
+[![Hackage-Deps](https://img.shields.io/hackage-deps/v/influxdb.svg)](http://packdeps.haskellers.com/feed?needle=influxdbhttp://packdeps.haskellers.com/feed?needle=influxdb)
+[![Gitter](https://badges.gitter.im/maoe/influxdb-haskell.svg)](https://gitter.im/maoe/influxdb-haskell?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge)
-Support for current version of InfluxDB is under development.
+Currently this library supports InfluxDB 1.2.
Contact information
----------
diff --git a/examples/random-points.hs b/examples/random-points.hs
index 50a8162..4e41845 100644
--- a/examples/random-points.hs
+++ b/examples/random-points.hs
@@ -1,26 +1,30 @@
{-# LANGUAGE BangPatterns #-}
{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE RecordWildCards #-}
{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TemplateHaskell #-}
{-# LANGUAGE TypeSynonymInstances #-}
{-# LANGUAGE ViewPatterns #-}
-{-# LANGUAGE TemplateHaskell #-}
-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 Data.Foldable
+import Data.Traversable
import System.Environment
import System.IO
-import qualified Data.Text as T
+import Text.Printf (printf)
+import Control.Lens
+import Data.Aeson
+import Data.Optional (Optional(Default))
+import Data.Time.Clock.POSIX
import System.Random.MWC (Variate(..))
+import qualified Control.Foldl as L
+import qualified Data.Map.Strict as Map
+import qualified Data.Text as T
import qualified Network.HTTP.Client as HC
import qualified System.Random.MWC as MWC
import Database.InfluxDB
-import Database.InfluxDB.TH
-import qualified Database.InfluxDB.Stream as S
+import qualified Database.InfluxDB.Format as F
+import qualified Database.InfluxDB.Manage as M
oneWeekInSeconds :: Int
oneWeekInSeconds = 7*24*60*60
@@ -29,74 +33,60 @@ main :: IO ()
main = do
[read -> (numPoints :: Int), read -> (batches :: Int)] <- getArgs
hSetBuffering stdout NoBuffering
- HC.withManager managerSettings $ \manager -> do
- config <- newConfig manager
-
- let db = "ctx"
- dropDatabase config db
- `E.catch`
- -- Ignore exceptions here
- \(_ :: HC.HttpException) -> return ()
- createDatabase config "ctx"
- gen <- MWC.create
- flip fix batches $ \outerLoop !m -> when (m > 0) $ do
- postWithPrecision config 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 (Time timestamp)
- innerLoop $ n - 1
- outerLoop $ m - 1
-
- result <- query config db "select count(value) from ct1;"
- case result of
- [] -> putStrLn "Empty series"
- series:_ -> do
- print $ seriesColumns series
- print $ seriesPoints series
- -- Streaming output
- queryChunked config db "select * from ct1;" $ S.fold step ()
- where
- step _ series = do
- case fromSeriesData series of
- Left reason -> hPutStrLn stderr reason
- Right points -> mapM_ print (points :: [Point])
- putStrLn "--"
-
-newConfig :: HC.Manager -> IO Config
-newConfig manager = do
- pool <- newServerPool localServer []
- return Config
- { configCreds = rootCreds
- , configServerPool = pool
- , configHttpManager = manager
- }
+ manager' <- HC.newManager managerSettings
+
+ let
+ ctx = "ctx"
+ ct1 = "ct1"
+ qparams = queryParams ctx
+ & manager .~ Right manager'
+ & precision .~ RFC3339
+
+ M.manage qparams $ F.formatQuery ("DROP DATABASE "%F.database) ctx
+ M.manage qparams $ F.formatQuery ("CREATE DATABASE "%F.database) ctx
+
+ let wparams = writeParams ctx & manager .~ Right manager'
+
+ gen <- MWC.create
+ for_ [1..batches] $ \_ -> do
+ batch <- for [1..numPoints] $ \_ -> do
+ !time <- (-)
+ <$> getPOSIXTime
+ <*> (fromIntegral <$> uniformR (0, oneWeekInSeconds) gen)
+ !value <- uniform gen
+ return (time, value)
+ writeBatch wparams $ flip map batch $ \(time, value) ->
+ Line ct1
+ (Map.fromList [])
+ (Map.fromList [("value", nameToFVal value)])
+ (Just time)
+
+ queryChunked qparams Default (F.formatQuery ("SELECT * FROM "%F.key) ct1) $
+ L.mapM_ $ traverse_ $ \Row {..} ->
+ printf "%s:\t%s\n"
+ (show $ posixSecondsToUTCTime rowTime)
+ (show rowValue)
managerSettings :: HC.ManagerSettings
managerSettings = HC.defaultManagerSettings
- { HC.managerResponseTimeout = Just $ 60*(10 :: Int)^(6 :: Int)
- }
-data Point = Point
- { pointValue :: !Name
- , pointTime :: !Time
+data Row = Row
+ { rowTime :: POSIXTime
+ , rowValue :: Name
} deriving Show
-newtype Time = Time POSIXTime
- deriving Show
-
-instance ToValue Time where
- toValue (Time epoch) = toValue $ epochInSeconds epoch
- where
- epochInSeconds :: POSIXTime -> Value
- epochInSeconds = Int . floor
-
-instance FromValue Time where
- parseValue (Int n) = return $ Time $ fromIntegral n
- parseValue (Float d) = return $ Time $ realToFrac d
- parseValue v = typeMismatch "Int or Float" v
+instance QueryResults Row where
+ parseResults prec = parseResultsWith $ \_ _ columns fields -> do
+ rowTime <- getField "time" columns fields >>= parseTimestamp prec
+ String name <- getField "value" columns fields
+ rowValue <- case name of
+ "foo" -> return Foo
+ "bar" -> return Bar
+ "baz" -> return Baz
+ "quu" -> return Quu
+ "qux" -> return Qux
+ _ -> fail $ "unknown name: " ++ show name
+ return Row {..}
data Name
= Foo
@@ -106,31 +96,11 @@ data Name
| 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
+nameToFVal :: Name -> FieldValue
+nameToFVal = FieldString . T.toLower . T.pack . show
instance Variate Name where
uniform = uniformR (minBound, maxBound)
uniformR (lower, upper) g = do
name <- uniformR (fromEnum lower, fromEnum upper) g
return $! toEnum name
-
--- Instance deriving
-
-deriveSeriesData defaultOptions
- { fieldLabelModifier = stripPrefixLower "point" }
- ''Point
diff --git a/examples/write-udp.hs b/examples/write-udp.hs
new file mode 100644
index 0000000..490c50a
--- /dev/null
+++ b/examples/write-udp.hs
@@ -0,0 +1,36 @@
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE OverloadedLists #-}
+{-# LANGUAGE OverloadedStrings #-}
+module Main where
+import Control.Exception
+
+import Control.Lens
+import Data.Time.Clock
+import Network.Socket
+
+import Database.InfluxDB.Types
+import Database.InfluxDB.Line
+import qualified Database.InfluxDB.Write.UDP as UDP
+
+main :: IO ()
+main = bracket (socket AF_INET Datagram defaultProtocol) close $ \sock -> do
+ localhost <- inet_addr "127.0.0.1"
+ let params = UDP.writeParams sock $ SockAddrInet 8089 localhost
+ tags1 =
+ [ ("tag1", "A")
+ , ("tag2", "B")
+ ]
+ fields1 =
+ [ ("val1", FieldInt 10)
+ , ("val2", FieldBool True)
+ ]
+ fields2 =
+ [ ("val1", FieldInt 1)
+ , ("val2", FieldBool False)
+ ]
+ UDP.write params $
+ Line "measurement1" tags1 fields1 (Nothing :: Maybe UTCTime)
+ now <- getCurrentTime
+ UDP.write
+ (params & UDP.precision .~ Millisecond)
+ (Line "measurement1" tags1 fields2 (Just now))
diff --git a/influxdb.cabal b/influxdb.cabal
index 4a90469..6d63fe8 100644
--- a/influxdb.cabal
+++ b/influxdb.cabal
@@ -1,5 +1,5 @@
name: influxdb
-version: 0.10.0
+version: 1.0.0
synopsis: Haskell client library for InfluxDB
description: Haskell client library for InfluxDB
homepage: https://github.com/maoe/influxdb-haskell
@@ -7,10 +7,11 @@ license: BSD3
license-file: LICENSE
author: Mitsutoshi Aoe
maintainer: Mitsutoshi Aoe <maoe@foldr.in>
-copyright: Copyright (C) 2014-2015 Mitsutoshi Aoe
+copyright: Copyright (C) 2014-2017 Mitsutoshi Aoe
category: Database
build-type: Simple
cabal-version: >= 1.10
+tested-with: GHC >= 7.10 && < 8.1
extra-source-files:
README.md
@@ -21,80 +22,69 @@ flag examples
default: False
manual: True
-flag aeson-070
- description: Use aeson >= 0.7.0
+flag http-client-05
default: True
manual: False
-flag network-uri
- description: Get Network.URI from the network-uri package
- default: True
-
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.Format
+ Database.InfluxDB.JSON
+ Database.InfluxDB.Line
+ Database.InfluxDB.Manage
+ Database.InfluxDB.Ping
+ Database.InfluxDB.Query
Database.InfluxDB.Types
- Database.InfluxDB.TH
+ Database.InfluxDB.Write
+ Database.InfluxDB.Write.UDP
other-modules:
- Database.InfluxDB.Types.Internal
+ Network.HTTP.Client.Compat
other-extensions:
BangPatterns
CPP
- ConstraintKinds
+ DataKinds
DeriveDataTypeable
+ DeriveGeneric
+ ExistentialQuantification
FlexibleInstances
+ FunctionalDependencies
+ GADTs
GeneralizedNewtypeDeriving
+ KindSignatures
+ LambdaCase
+ MultiParamTypeClasses
NamedFieldPuns
OverloadedStrings
- RankNTypes
RecordWildCards
ScopedTypeVariables
+ StandaloneDeriving
TemplateHaskell
- TypeSynonymInstances
ViewPatterns
ghc-options: -Wall
build-depends:
- base >= 4 && < 5.0
+ base >= 4 && < 4.10
+ , aeson >= 0.7 && < 1.2
, attoparsec < 0.14
- , bytestring
- , containers
- , data-default-class
- , dlist
- , exceptions >= 0.5 && < 0.9
- , http-client < 0.5
- , mtl < 2.3
- , retry >= 0.7 && < 0.8
- , tagged
- , template-haskell
+ , bytestring >= 0.10 && < 0.11
+ , clock >= 0.7 && < 0.8
+ , containers >= 0.5 && < 0.6
+ , foldl < 1.3
+ , http-types >= 0.8.6 && < 0.10
+ , lens >= 4.9 && < 4.16
+ , network >= 2.6 && < 2.7
+ , optional-args >= 1.0 && < 1.1
+ , scientific >= 0.3.3 && < 0.4
, text < 1.3
- , vector
-
- if flag(aeson-070)
- build-depends:
- aeson >= 0.7.0 && < 0.12
- , scientific >= 0.2
- else
- build-depends:
- aeson >= 0.6.1.0 && < 0.7.0
-
- if flag(network-uri)
- build-depends:
- network-uri >= 2.6
+ , time >= 1.5 && < 1.9
+ , unordered-containers < 0.3
+ , vector >= 0.10 && < 0.13
+ if flag(http-client-05)
+ build-depends: http-client >= 0.5 && < 0.6
else
build-depends:
- network < 2.6
-
- if impl(ghc < 7.6)
- build-depends:
- ghc-prim
-
+ data-default-class
+ , http-client >= 0.4.10 && < 0.5
hs-source-dirs: src
default-language: Haskell2010
@@ -117,22 +107,40 @@ test-suite test-suite
default-language: Haskell2010
executable influx-random-points
- if flag(examples)
- buildable: True
- else
+ if !flag(examples)
buildable: False
hs-source-dirs: examples
main-is: random-points.hs
ghc-options: -Wall
build-depends:
- base
+ aeson
+ , base
, bytestring
+ , containers
+ , foldl >= 1.1.3
, http-client
, influxdb
- , mtl
+ , lens
, mwc-random
+ , optional-args
, text
, time
+ , vector
+ default-language: Haskell2010
+
+executable influx-write-udp
+ if !flag(examples)
+ buildable: False
+ hs-source-dirs: examples
+ main-is: write-udp.hs
+ ghc-options: -Wall
+ build-depends:
+ base
+ , containers
+ , influxdb
+ , lens
+ , network
+ , time
default-language: Haskell2010
source-repository head
@@ -142,5 +150,5 @@ source-repository head
source-repository this
type: git
- tag: v0.10.0
+ tag: v1.0.0
location: https://github.com/maoe/influxdb-haskell.git
diff --git a/src/Database/InfluxDB.hs b/src/Database/InfluxDB.hs
index d3b5836..aef0159 100644
--- a/src/Database/InfluxDB.hs
+++ b/src/Database/InfluxDB.hs
@@ -1,94 +1,97 @@
+{- |
+stability: experimental
+portability: GHC
+-}
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
- , newServerPoolWithRetryPolicy
- , Database(..)
- , User(..)
- , Admin(..)
- , Ping(..)
- , ShardSpace(..)
-
- -- *** Exception
- , InfluxException(..)
+ ( -- $intro
+
+ -- * Writing data
+ -- $write
+ write
+ , writeBatch
+ , writeByteString
- -- ** Writing Data
+ -- ** Write parameters
+ , WriteParams
+ , writeParams
+ , retentionPolicy
- -- *** Updating Points
- , post, postWithPrecision
- , SeriesT, PointT
- , writeSeries
- , writeSeriesData
- , withSeries
- , writePoints
+ -- ** The Line protocol
+ , Line(Line)
+ , measurement
+ , tagSet
+ , fieldSet
+ , timestamp
- -- *** Deleting Points
- , deleteSeries
+ , FieldValue(..)
+ , Timestamp(..)
+ , precisionScale
+ , precisionName
- -- ** Querying Data
+ -- * Querying data
+ , Query
, query
- , Stream(..)
, queryChunked
- -- ** Administration & Security
- -- *** Creating and Dropping Databases
- , listDatabases
- , createDatabase
- , dropDatabase
-
- , DatabaseRequest(..)
- , configureDatabase
-
- -- *** Security
- -- **** Shard spaces
- , ShardSpaceRequest(..)
- , listShardSpaces
- , createShardSpace
- , dropShardSpace
-
- -- **** Cluster admin
- , listClusterAdmins
- , authenticateClusterAdmin
- , addClusterAdmin
- , updateClusterAdminPassword
- , deleteClusterAdmin
- -- **** Database user
- , listDatabaseUsers
- , authenticateDatabaseUser
- , addDatabaseUser
- , updateDatabaseUserPassword
- , deleteDatabaseUser
- , grantAdminPrivilegeTo
- , revokeAdminPrivilegeFrom
-
- -- *** Other API
- , ping
- , isInSync
+ -- * Query constructor
+ , formatQuery
+ , (%)
+
+ -- ** Query parameters
+ , QueryParams
+ , queryParams
+ , authentication
+
+ -- ** Parsing results
+ , QueryResults(..)
+ , parseResultsWith
+ , getField
+ , getTag
+ , parseTimestamp
+ , parseFieldValue
+ , parseKey
+
+ -- * Database management
+ , manage
+
+ -- * Common data types and classes
+ , Precision(..)
+ , Database
+ , Key
+
+ , Server
+ , host
+ , port
+ , ssl
+ , localServer
+
+ , Credentials
+ , user
+ , password
+
+ -- * Exception
+ , InfluxException(..)
+
+ , HasServer(..)
+ , HasDatabase(..)
+ , HasPrecision(..)
+ , HasManager(..)
) where
-import Database.InfluxDB.Decode
-import Database.InfluxDB.Encode
-import Database.InfluxDB.Http
+import Database.InfluxDB.Format ((%), formatQuery)
+import Database.InfluxDB.JSON
+import Database.InfluxDB.Line
+import Database.InfluxDB.Manage (manage)
+import Database.InfluxDB.Query
import Database.InfluxDB.Types
+import Database.InfluxDB.Write
+
+{- $write
+InfluxDB has two ways to write data into it, via HTTP and UDP. This module
+only exports functions for the HTTP API. For UDP, you can use a qualified
+import:
+
+@
+import qualified "Database.InfluxDB.Write.UDP" as UDP
+@
+-}
diff --git a/src/Database/InfluxDB/Decode.hs b/src/Database/InfluxDB/Decode.hs
deleted file mode 100644
index 94c0eda..0000000
--- a/src/Database/InfluxDB/Decode.hs
+++ /dev/null
@@ -1,232 +0,0 @@
-{-# LANGUAGE FlexibleInstances #-}
-{-# LANGUAGE GeneralizedNewtypeDeriving #-}
-{-# LANGUAGE RecordWildCards #-}
-{-# LANGUAGE TypeSynonymInstances #-}
-module Database.InfluxDB.Decode
- ( FromSeries(..), fromSeries
- , FromSeriesData(..), fromSeriesData, fromSeriesData_
- , withValues, (.:), (.:?), (.!=)
- , FromValue(..), fromValue
- , Parser, ValueParser, typeMismatch
- ) where
-import Control.Applicative
-import Control.Monad.Reader
-import Data.Either (rights)
-import Data.Int
-import Data.Word
-import Data.Map (Map)
-import Data.Maybe (fromMaybe)
-import Data.Vector (Vector)
-import Data.Tuple (swap)
-import qualified Data.Map as Map
-import qualified Data.Text as T
-import qualified Data.Text.Lazy as TL
-import qualified Data.Vector as V
-import Prelude
-
-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 Event 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 = [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)
- seriesDataPoints
-
--- | Same as @fromSeriesData@ but ignores parse errors and returns only
--- successful data.
-fromSeriesData_ :: FromSeriesData a => SeriesData -> [a]
-fromSeriesData_ SeriesData {..} = rights $ map
- (runParser . parseSeriesData seriesDataColumns)
- seriesDataPoints
-
--- | Helper function to define 'parseSeriesData' from 'ValueParser's.
-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
-
--- | Retrieve the value associated with the given column. The result is 'empty'
--- if the column is not present or the value cannot be converted to the desired
--- type.
-(.:) :: FromValue a => Vector Value -> Column -> ValueParser a
-values .: column = do
- found <- asks $ Map.lookup column
- case found of
- Nothing -> liftParser $ parseError $ "No such column: " ++ T.unpack column
- Just idx -> do
- value <- V.indexM values idx
- liftParser $ parseValue value
-
--- | Retrieve the value associated with the given column. The result is
--- 'Nothing' if the column is not present or the value cannot be converted to
--- the desired type.
-(.:?) :: FromValue a => Vector Value -> Column -> ValueParser (Maybe a)
-values .:? column = do
- found <- asks $ Map.lookup column
- case found of
- Nothing -> return Nothing
- Just idx ->
- case values V.!? idx of
- Nothing -> return Nothing
- Just value -> liftParser $ parseValue value
-
--- | Helper for use in combination with '.:?' to provide default values for
--- optional columns.
-(.!=) :: Parser (Maybe a) -> a -> Parser a
-p .!= def = fromMaybe def <$> p
-
--- | 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 Int8 where
- parseValue (Int n)
- | n <= fromIntegral (maxBound :: Int8) = return $ fromIntegral n
- | otherwise = parseError $ "Larger than the maximum Int8: " ++ show n
- parseValue v = typeMismatch "Int8" v
-
-instance FromValue Int16 where
- parseValue (Int n)
- | n <= fromIntegral (maxBound :: Int16) = return $ fromIntegral n
- | otherwise = parseError $ "Larger than the maximum Int16: " ++ show n
- parseValue v = typeMismatch "Int16" v
-
-instance FromValue Int32 where
- parseValue (Int n)
- | n <= fromIntegral (maxBound :: Int32) = return $ fromIntegral n
- | otherwise = parseError $ "Larger than the maximum Int32: " ++ show n
- parseValue v = typeMismatch "Int32" v
-
-instance FromValue Int64 where
- parseValue (Int n)
- | n <= fromIntegral (maxBound :: Int64) = return $ fromIntegral n
- | otherwise = parseError $ "Larger than the maximum Int64: " ++ show n
- parseValue v = typeMismatch "Int64" v
-
-instance FromValue Word8 where
- parseValue (Int n)
- | n <= fromIntegral (maxBound :: Word8) = return $ fromIntegral n
- | otherwise = parseError $ "Larger than the maximum Word8: " ++ show n
- parseValue v = typeMismatch "Word8" v
-
-instance FromValue Word16 where
- parseValue (Int n)
- | n <= fromIntegral (maxBound :: Word16) = return $ fromIntegral n
- | otherwise = parseError $ "Larger than the maximum Word16: " ++ show n
- parseValue v = typeMismatch "Word16" v
-
-instance FromValue Word32 where
- parseValue (Int n)
- | n <= fromIntegral (maxBound :: Word32) = return $ fromIntegral n
- | otherwise = parseError $ "Larger than the maximum Word32: " ++ show n
- parseValue v = typeMismatch "Word32" v
-
-instance FromValue Double where
- parseValue (Float d) = return d
- -- If the floating number happens to be a whole number, it must
- -- have encoded as an integer. We should decode it back as a floating
- -- number here.
- parseValue (Int n) = return $ fromIntegral n
- parseValue v = typeMismatch "Float" v
-
-instance FromValue T.Text where
- parseValue (String xs) = return xs
- parseValue v = typeMismatch "Text" v
-
-instance FromValue TL.Text where
- parseValue (String xs) = return $ TL.fromStrict xs
- parseValue v = typeMismatch "lazy Text" v
-
-instance FromValue String where
- parseValue (String xs) = return $ T.unpack xs
- parseValue v = typeMismatch "String" v
-
-typeMismatch
- :: String
- -> Value
- -> Parser a
-typeMismatch expected actual = parseError $
- "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
-
-parseError :: String -> Parser a
-parseError = Parser . Left
diff --git a/src/Database/InfluxDB/Encode.hs b/src/Database/InfluxDB/Encode.hs
deleted file mode 100644
index 6b28c90..0000000
--- a/src/Database/InfluxDB/Encode.hs
+++ /dev/null
@@ -1,95 +0,0 @@
-{-# 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.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 Event where
--- > toSeriesColumns _ = 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 = [toSeriesPoints a]
- }
-
--- | A type that can be stored in InfluxDB.
-class ToValue a where
- toValue :: a -> Value
-
-instance ToValue Value where
- toValue = id
-
-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 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
diff --git a/src/Database/InfluxDB/Format.hs b/src/Database/InfluxDB/Format.hs
new file mode 100644
index 0000000..42d8c7e
--- /dev/null
+++ b/src/Database/InfluxDB/Format.hs
@@ -0,0 +1,108 @@
+{-# LANGUAGE GADTs #-}
+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE OverloadedStrings #-}
+module Database.InfluxDB.Format
+ ( Query
+ , fromQuery
+
+ , Format
+ , makeFormat
+ , (%)
+ , formatQuery
+ , formatDatabase
+ , formatKey
+
+ , database
+ , key
+ , keys
+ , fieldVal
+ , decimal
+ , realFloat
+ , text
+ , time
+ ) where
+import Control.Category
+import Data.Monoid
+import Data.String
+import Prelude hiding ((.), id)
+
+import Data.Time
+import qualified Data.ByteString as B
+import qualified Data.ByteString.Lazy as BL
+import qualified Data.ByteString.Lazy.Builder as BL
+import qualified Data.List as L
+import qualified Data.Text as T
+import qualified Data.Text.Encoding as T
+import qualified Data.Text.Lazy as TL
+import qualified Data.Text.Lazy.Builder as TL
+import qualified Data.Text.Lazy.Builder.Int as TL
+import qualified Data.Text.Lazy.Builder.RealFloat as TL
+
+import Database.InfluxDB.Types hiding (database)
+
+fromQuery :: Query -> B.ByteString
+fromQuery (Query q) =
+ BL.toStrict $ BL.toLazyByteString $ T.encodeUtf8Builder q
+
+newtype Format a b = Format { runFormat :: (TL.Builder -> a) -> b }
+
+instance Category Format where
+ id = Format (\k -> k "")
+ fmt1 . fmt2 = Format $ \k ->
+ runFormat fmt1 $ \a ->
+ runFormat fmt2 $ \b ->
+ k (a <> b)
+
+instance a ~ b => IsString (Format a b) where
+ fromString xs = Format $ \k -> k $ fromString xs
+
+(%) :: Format b c -> Format a b -> Format a c
+(%) = (.)
+
+formatQuery :: Format Query r -> r
+formatQuery fmt = runFormat fmt (Query . TL.toStrict . TL.toLazyText)
+
+formatDatabase :: Format Database r -> r
+formatDatabase fmt = runFormat fmt (Database . TL.toStrict . TL.toLazyText)
+
+formatKey :: Format Key r -> r
+formatKey fmt = runFormat fmt (Key . TL.toStrict . TL.toLazyText)
+
+makeFormat :: (a -> TL.Builder) -> Format r (a -> r)
+makeFormat build = Format $ \k a -> k $ build a
+
+database :: Format r (Database -> r)
+database = makeFormat $ \(Database name) -> "\"" <> TL.fromText name <> "\""
+
+keyBuilder :: Key -> TL.Builder
+keyBuilder (Key name) = "\"" <> TL.fromText name <> "\""
+
+key :: Format r (Key -> r)
+key = makeFormat keyBuilder
+
+keys :: Format r ([Key] -> r)
+keys = makeFormat $ mconcat . L.intersperse "," . map keyBuilder
+
+fieldVal :: Format r (FieldValue -> r)
+fieldVal = makeFormat $ \case
+ FieldInt n -> TL.decimal n
+ FieldFloat d -> TL.realFloat d
+ FieldString s -> "'" <> TL.fromText s <> "'"
+ FieldBool b -> if b then "true" else "false"
+ FieldNull -> "null"
+
+decimal :: Integral a => Format r (a -> r)
+decimal = makeFormat TL.decimal
+
+realFloat :: RealFloat a => Format r (a -> r)
+realFloat = makeFormat TL.realFloat
+
+text :: Format r (T.Text -> r)
+text = makeFormat TL.fromText
+
+time :: FormatTime time => Format r (time -> r)
+time = makeFormat $ \t ->
+ "'" <> TL.fromString (formatTime defaultTimeLocale fmt t) <> "'"
+ where
+ fmt = "%F %X%Q" -- YYYY-MM-DD HH:MM:SS.nnnnnnnnn
diff --git a/src/Database/InfluxDB/Http.hs b/src/Database/InfluxDB/Http.hs
deleted file mode 100644
index 93931e3..0000000
--- a/src/Database/InfluxDB/Http.hs
+++ /dev/null
@@ -1,792 +0,0 @@
-{-# LANGUAGE CPP #-}
-{-# LANGUAGE GeneralizedNewtypeDeriving #-}
-{-# LANGUAGE NamedFieldPuns #-}
-{-# LANGUAGE OverloadedStrings #-}
-{-# LANGUAGE RecordWildCards #-}
-{-# LANGUAGE ScopedTypeVariables #-}
-{-# LANGUAGE TemplateHaskell #-}
-module Database.InfluxDB.Http
- ( Config(..)
- , Credentials(..), rootCreds
- , Server(..), localServer
- , TimePrecision(..)
-
- -- * Writing Data
-
- -- ** Updating Points
- , post, postWithPrecision
- , SeriesT, PointT
- , writeSeries
- , writeSeriesData
- , withSeries
- , writePoints
-
- -- ** Deleting Points
- , deleteSeries
-
- -- * Querying Data
- , query
- , Stream(..)
- , queryChunked
-
- -- * Administration & Security
- -- ** Creating and Dropping Databases
- , listDatabases
- , createDatabase
- , dropDatabase
-
- , DatabaseRequest(..)
- , configureDatabase
-
- -- ** Security
- -- *** Shard spaces
- , ShardSpaceRequest(..)
- , listShardSpaces
- , createShardSpace
- , dropShardSpace
-
- -- *** Cluster admin
- , listClusterAdmins
- , authenticateClusterAdmin
- , addClusterAdmin
- , updateClusterAdminPassword
- , deleteClusterAdmin
- -- *** Database user
- , listDatabaseUsers
- , authenticateDatabaseUser
- , addDatabaseUser
- , updateDatabaseUserPassword
- , deleteDatabaseUser
- , grantAdminPrivilegeTo
- , revokeAdminPrivilegeFrom
-
- -- ** Other API
- , ping
- , isInSync
- ) where
-
-import Control.Applicative
-import Control.Monad.Identity
-import Control.Monad.Writer
-import Data.DList (DList)
-import Data.IORef
-import Data.Proxy
-import Data.Text (Text)
-import Data.Vector (Vector)
-import Data.Word (Word32)
-import Network.URI (escapeURIString, isAllowedInURI)
-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 Text.Printf (printf)
-import Prelude
-
-import Control.Monad.Catch (Handler(..))
-import Control.Retry (recovering)
-import Data.Aeson ((.=))
-import Data.Aeson.TH (deriveToJSON)
-import Data.Default.Class (Default(def))
-import qualified Data.Aeson as A
-import qualified Data.Aeson.Encode as AE
-import qualified Data.Aeson.Parser as AP
-import qualified Data.Aeson.Types as AT
-import qualified Data.Attoparsec.ByteString as P
-import qualified Data.Attoparsec.ByteString.Lazy as PL
-import qualified Network.HTTP.Client as HC
-
-import Database.InfluxDB.Decode
-import Database.InfluxDB.Encode
-import Database.InfluxDB.Types
-import Database.InfluxDB.Types.Internal (stripPrefixOptions)
-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)
- , configHttpManager :: !HC.Manager
- }
-
--- | 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
-
-timePrecString :: TimePrecision -> String
-timePrecString SecondsPrecision = "s"
-timePrecString MillisecondsPrecision = "ms"
-timePrecString MicrosecondsPrecision = "u"
-
------------------------------------------------------------
--- Writing Data
-
--- | Post a bunch of writes for (possibly multiple) series into a database.
-post
- :: Config
- -> Text
- -> SeriesT IO a
- -> IO a
-post config databaseName =
- postGeneric config databaseName Nothing
-
--- | Post a bunch of writes for (possibly multiple) series into a database like
--- 'post' but with time precision.
-postWithPrecision
- :: Config
- -> Text -- ^ Database name
- -> TimePrecision
- -> SeriesT IO a
- -> IO a
-postWithPrecision config databaseName timePrec =
- postGeneric config databaseName (Just timePrec)
-
-postGeneric
- :: Config
- -> Text -- ^ Database name
- -> Maybe TimePrecision
- -> SeriesT IO a
- -> IO a
-postGeneric Config {..} databaseName timePrec write = do
- (a, series) <- runSeriesT write
- void $ httpLbsWithRetry configServerPool
- (makeRequest series)
- configHttpManager
- 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=%s" . timePrecString)
- timePrec :: String)
- }
- 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 = writeSeriesData name . toSeriesData
-
--- | Write a single series data.
-writeSeriesData
- :: Monad m
- => Text
- -- ^ Series name
- -> SeriesData
- -- ^ Series data
- -> SeriesT m ()
-writeSeriesData name a = tell . DL.singleton $ Series
- { seriesName = name
- , seriesData = 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 = DL.toList values
- }
- }
-
--- | Write a data into a series.
-writePoints
- :: (Monad m, ToSeriesData a)
- => a
- -> PointT a m ()
-writePoints = tell . DL.singleton . toSeriesPoints
-
-deleteSeries
- :: Config
- -> Text -- ^ Database name
- -> Text -- ^ Series name
- -> IO ()
-deleteSeries config databaseName seriesName = runRequest_ config request
- where
- request = def
- { HC.method = "DELETE"
- , HC.path = escapeString $ printf "/db/%s/series/%s"
- (T.unpack databaseName)
- (T.unpack seriesName)
- , HC.queryString = escapeString $ printf "u=%s&p=%s"
- (T.unpack credsUser)
- (T.unpack credsPassword)
- }
- Credentials {..} = configCreds config
-
------------------------------------------------------------
--- 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
- -> Text -- ^ Database name
- -> Text -- ^ Query text
- -> IO [a]
-query config databaseName q = do
- xs <- runRequest config request
- case mapM fromSeries xs of
- Left reason -> seriesDecodeError 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)
- }
- Credentials {..} = configCreds config
-
--- | 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 -> jsonDecodeError message
- decode (P.Partial k) = demandPayload (decode . k)
- decode (P.Fail _ _ message) = jsonDecodeError message
- parseAsJson = P.parse A.json
-
--- | Query a specified database like 'query' but in a streaming fashion.
-queryChunked
- :: FromSeries a
- => Config
- -> Text -- ^ Database name
- -> Text -- ^ Query text
- -> (Stream IO a -> IO b)
- -- ^ Action to handle the resulting stream of series
- -> IO b
-queryChunked Config {..} databaseName q f =
- withPool configServerPool request $ \request' ->
- HC.withResponse request' configHttpManager $
- responseStream . HC.responseBody >=> S.mapM parse >=> f
- where
- parse series = case fromSeries series of
- Left reason -> seriesDecodeError 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)
- }
- Credentials {..} = configCreds
-
------------------------------------------------------------
--- Administration & Security
-
--- | List existing databases.
-listDatabases :: Config -> IO [Database]
-listDatabases config = runRequest config request
- where
- request = def
- { HC.path = "/db"
- , HC.queryString = escapeString $ printf "u=%s&p=%s"
- (T.unpack credsUser)
- (T.unpack credsPassword)
- }
- Credentials {..} = configCreds config
-
--- | Create a new database. Requires cluster admin privileges.
-createDatabase :: Config -> Text -> IO ()
-createDatabase config name = runRequest_ config request
- where
- request = 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 config
-
--- | Drop a database. Requires cluster admin privileges.
-dropDatabase
- :: Config
- -> Text -- ^ Database name
- -> IO ()
-dropDatabase config databaseName = runRequest_ config request
- where
- request = 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)
- }
- Credentials {..} = configCreds config
-
-
-data DatabaseRequest = DatabaseRequest
- { databaseRequestSpaces :: [ShardSpaceRequest]
- , databaseRequestContinuousQueries :: [Text]
- } deriving Show
-
-configureDatabase
- :: Config
- -> Text -- ^ Database name
- -> DatabaseRequest
- -> IO ()
-configureDatabase config databaseName databaseRequest =
- runRequest_ config request
- where
- request = def
- { HC.method = "POST"
- , HC.requestBody = HC.RequestBodyLBS $ AE.encode databaseRequest
- , HC.path = escapeString $ printf "/cluster/database_configs/%s"
- (T.unpack databaseName)
- , HC.queryString = escapeString $ printf "u=%s&p=%s"
- (T.unpack credsUser)
- (T.unpack credsPassword)
- }
- Credentials {..} = configCreds config
-
--- | List shard spaces.
-listShardSpaces :: Config -> IO [ShardSpace]
-listShardSpaces config = runRequest config request
- where
- request = def
- { HC.path = "/cluster/shard_spaces"
- , HC.queryString = escapeString $ printf "u=%s&p=%s"
- (T.unpack credsUser)
- (T.unpack credsPassword)
- }
- Credentials {..} = configCreds config
-
-data ShardSpaceRequest = ShardSpaceRequest
- { shardSpaceRequestName :: Text
- , shardSpaceRequestRegex :: Text
- , shardSpaceRequestRetentionPolicy :: Text
- , shardSpaceRequestShardDuration :: Text
- , shardSpaceRequestReplicationFactor :: Word32
- , shardSpaceRequestSplit :: Word32
- } deriving Show
-
--- | Create a shard space.
-createShardSpace
- :: Config
- -> Text -- ^ Database
- -> ShardSpaceRequest
- -> IO ()
-createShardSpace config databaseName shardSpace = runRequest_ config request
- where
- request = def
- { HC.method = "POST"
- , HC.requestBody = HC.RequestBodyLBS $ AE.encode shardSpace
- , HC.path = escapeString $ printf "/cluster/shard_spaces/%s"
- (T.unpack databaseName)
- , HC.queryString = escapeString $ printf "u=%s&p=%s"
- (T.unpack credsUser)
- (T.unpack credsPassword)
- }
- Credentials {..} = configCreds config
-
-dropShardSpace
- :: Config
- -> Text -- ^ Database name
- -> Text -- ^ Shard space name
- -> IO ()
-dropShardSpace config databaseName shardSpaceName = runRequest_ config request
- where
- request = def
- { HC.method = "DELETE"
- , HC.path = escapeString $ printf "/cluster/shard_spaces/%s/%s"
- (T.unpack databaseName)
- (T.unpack shardSpaceName)
- , HC.queryString = escapeString $ printf "u=%s&p=%s"
- (T.unpack credsUser)
- (T.unpack credsPassword)
- }
- Credentials {..} = configCreds config
-
--- | List cluster administrators.
-listClusterAdmins :: Config -> IO [Admin]
-listClusterAdmins config = runRequest config request
- where
- request = def
- { HC.path = "/cluster_admins"
- , HC.queryString = escapeString $ printf "u=%s&p=%s"
- (T.unpack credsUser)
- (T.unpack credsPassword)
- }
- Credentials {..} = configCreds config
-
-authenticateClusterAdmin :: Config -> IO ()
-authenticateClusterAdmin config = runRequest_ config request
- where
- request = def
- { HC.path = "/cluster_admins/authenticate"
- , HC.queryString = escapeString $ printf "u=%s&p=%s"
- (T.unpack credsUser)
- (T.unpack credsPassword)
- }
- Credentials {..} = configCreds config
-
--- | Add a new cluster administrator. Requires cluster admin privilege.
-addClusterAdmin
- :: Config
- -> Text -- ^ Admin name
- -> Text -- ^ Password
- -> IO Admin
-addClusterAdmin config name password = do
- runRequest_ config request
- return Admin
- { adminName = name
- }
- where
- request = def
- { HC.method = "POST"
- , HC.requestBody = HC.RequestBodyLBS $ AE.encode $ A.object
- [ "name" .= name
- , "password" .= password
- ]
- , HC.path = "/cluster_admins"
- , HC.queryString = escapeString $ printf "u=%s&p=%s"
- (T.unpack credsUser)
- (T.unpack credsPassword)
- }
- Credentials {..} = configCreds config
-
--- | Update a cluster administrator's password. Requires cluster admin
--- privilege.
-updateClusterAdminPassword
- :: Config
- -> Admin
- -> Text -- ^ New password
- -> IO ()
-updateClusterAdminPassword Config {..} admin password =
- void $ httpLbsWithRetry configServerPool makeRequest configHttpManager
- 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 adminName)
- , HC.queryString = escapeString $ printf "u=%s&p=%s"
- (T.unpack credsUser)
- (T.unpack credsPassword)
- }
- Admin {adminName} = admin
- Credentials {..} = configCreds
-
--- | Delete a cluster administrator. Requires cluster admin privilege.
-deleteClusterAdmin
- :: Config
- -> Admin
- -> IO ()
-deleteClusterAdmin Config {..} admin =
- void $ httpLbsWithRetry configServerPool makeRequest configHttpManager
- where
- makeRequest = def
- { HC.method = "DELETE"
- , HC.path = escapeString $ printf "/cluster_admins/%s"
- (T.unpack adminName)
- , HC.queryString = escapeString $ printf "u=%s&p=%s"
- (T.unpack credsUser)
- (T.unpack credsPassword)
- }
- Admin {adminName} = admin
- Credentials {..} = configCreds
-
--- | List database users.
-listDatabaseUsers
- :: Config
- -> Text
- -> IO [User]
-listDatabaseUsers config@Config {..} database = runRequest config makeRequest
- 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
-
-authenticateDatabaseUser
- :: Config
- -> Text -- ^ Database name
- -> IO ()
-authenticateDatabaseUser Config {..} database =
- void $ httpLbsWithRetry configServerPool makeRequest configHttpManager
- where
- makeRequest = def
- { HC.path = escapeString $ printf "/db/%s/authenticate"
- (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
- -> Text -- ^ Database name
- -> Text -- ^ User name
- -> Text -- ^ Password
- -> IO ()
-addDatabaseUser config databaseName name password = runRequest_ config request
- where
- request = def
- { HC.method = "POST"
- , HC.requestBody = HC.RequestBodyLBS $ AE.encode $ A.object
- [ "name" .= name
- , "password" .= password
- ]
- , HC.path = escapeString $ printf "/db/%s/users"
- (T.unpack databaseName)
- , HC.queryString = escapeString $ printf "u=%s&p=%s"
- (T.unpack credsUser)
- (T.unpack credsPassword)
- }
- Credentials {..} = configCreds config
-
--- | Delete an user from the database users.
-deleteDatabaseUser
- :: Config
- -> Text -- ^ Database name
- -> Text -- ^ User name
- -> IO ()
-deleteDatabaseUser config databaseName userName = runRequest_ config request
- where
- request = (makeRequestFromDatabaseUser config databaseName userName)
- { HC.method = "DELETE"
- }
-
--- | Update password for the database user.
-updateDatabaseUserPassword
- :: Config
- -> Text -- ^ Database name
- -> Text -- ^ User name
- -> Text -- ^ New password
- -> IO ()
-updateDatabaseUserPassword config databaseName userName password =
- runRequest_ config request
- where
- request = (makeRequestFromDatabaseUser config databaseName userName)
- { HC.method = "POST"
- , HC.requestBody = HC.RequestBodyLBS $ AE.encode $ A.object
- [ "password" .= password
- ]
- }
-
--- | Give admin privilege to the user.
-grantAdminPrivilegeTo
- :: Config
- -> Text -- ^ Database name
- -> Text -- ^ User name
- -> IO ()
-grantAdminPrivilegeTo config databaseName userName = runRequest_ config request
- where
- request = (makeRequestFromDatabaseUser config databaseName userName)
- { HC.method = "POST"
- , HC.requestBody = HC.RequestBodyLBS $ AE.encode $ A.object
- [ "admin" .= True
- ]
- }
-
--- | Remove admin privilege from the user.
-revokeAdminPrivilegeFrom
- :: Config
- -> Text -- ^ Database name
- -> Text -- ^ User name
- -> IO ()
-revokeAdminPrivilegeFrom config databaseName userName =
- runRequest_ config request
- where
- request = (makeRequestFromDatabaseUser config databaseName userName)
- { HC.method = "POST"
- , HC.requestBody = HC.RequestBodyLBS $ AE.encode $ A.object
- [ "admin" .= False
- ]
- }
-
-makeRequestFromDatabaseUser
- :: Config
- -> Text -- ^ Database name
- -> Text -- ^ User name
- -> HC.Request
-makeRequestFromDatabaseUser Config {..} databaseName userName = 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
- Credentials {..} = configCreds
-
-ping :: Config -> IO Ping
-ping config = runRequest config request
- where
- request = def
- { HC.path = "/ping"
- }
-
-isInSync :: Config -> IO Bool
-isInSync Config {..} = do
- response <- httpLbsWithRetry configServerPool makeRequest configHttpManager
- case eitherDecodeBool (HC.responseBody response) of
- Left reason -> jsonDecodeError reason
- Right status -> return status
- where
- makeRequest = def
- { HC.path = "/sync"
- , HC.queryString = escapeString $ printf "u=%s&p=%s"
- (T.unpack credsUser)
- (T.unpack credsPassword)
- }
- Credentials {..} = configCreds
- eitherDecodeBool lbs = do
- val <- PL.eitherResult $ PL.parse AP.value lbs
- AT.parseEither A.parseJSON val
-
------------------------------------------------------------
-
-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 = do
- retryPolicy <- serverRetryPolicy <$> readIORef pool
- recovering retryPolicy handlers $ \_ -> do
- server <- activeServer pool
- f $ makeRequest server
- where
- makeRequest Server {..} = request
- { HC.host = escapeText serverHost
- , HC.port = serverPort
- , HC.secure = serverSsl
- }
- handlers =
- [ const $ Handler $ \e -> case e of
- HC.FailedConnectionException {} -> retry
- HC.FailedConnectionException2 {} -> retry
- HC.InternalIOException {} -> retry
- HC.ResponseTimeout {} -> retry
- _ -> return False
- ]
- retry = True <$ failover pool
-
-escapeText :: Text -> BS.ByteString
-escapeText = escapeString . T.unpack
-
-escapeString :: String -> BS.ByteString
-escapeString = BS8.pack . escapeURIString isAllowedInURI
-
-decodeJsonResponse
- :: A.FromJSON a
- => HC.Response BL.ByteString
- -> IO a
-decodeJsonResponse response =
- case A.eitherDecode (HC.responseBody response) of
- Left reason -> jsonDecodeError reason
- Right a -> return a
-
-runRequest :: A.FromJSON a => Config -> HC.Request -> IO a
-runRequest Config {..} req = do
- response <- httpLbsWithRetry configServerPool req configHttpManager
- decodeJsonResponse response
-
-runRequest_ :: Config -> HC.Request -> IO ()
-runRequest_ Config {..} req =
- void $ httpLbsWithRetry configServerPool req configHttpManager
-
------------------------------------------------------------
--- Aeson instances
-
-deriveToJSON (stripPrefixOptions "shardSpaceRequest") ''ShardSpaceRequest
-deriveToJSON (stripPrefixOptions "databaseRequest") ''DatabaseRequest
diff --git a/src/Database/InfluxDB/JSON.hs b/src/Database/InfluxDB/JSON.hs
new file mode 100644
index 0000000..544a898
--- /dev/null
+++ b/src/Database/InfluxDB/JSON.hs
@@ -0,0 +1,207 @@
+{-# LANGUAGE BangPatterns #-}
+{-# LANGUAGE ExistentialQuantification #-}
+{-# LANGUAGE FlexibleInstances #-}
+{-# LANGUAGE GADTs #-}
+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE RecordWildCards #-}
+{-# LANGUAGE TemplateHaskell #-}
+{-# LANGUAGE ViewPatterns #-}
+module Database.InfluxDB.JSON
+ ( parseResultsWith
+ , parseResultsWithDecoder
+ , Decoder(..)
+ , strictDecoder
+ , lenientDecoder
+
+ , getField
+ , getTag
+
+ , parseTimestamp
+ , parsePOSIXTime
+ , parseRFC3339
+ , parseFieldValue
+
+ , parseResultsObject
+ , parseSeriesObject
+ , parseSeriesBody
+ , parseErrorObject
+ ) where
+import Control.Applicative
+import Control.Exception
+import Control.Monad
+import Data.Maybe
+
+import Data.Aeson
+import Data.HashMap.Strict (HashMap)
+import Data.Text (Text)
+import Data.Time.Clock.POSIX
+import Data.Time.Format
+import Data.Vector (Vector)
+import qualified Data.Aeson.Types as A
+import qualified Data.HashMap.Strict as HashMap
+import qualified Data.Scientific as Sci
+import qualified Data.Text as T
+import qualified Data.Vector as V
+
+import Database.InfluxDB.Types
+
+-- | Parse a JSON response
+parseResultsWith
+ :: (Maybe Text -> HashMap Text Text -> Vector Text -> Array -> A.Parser a)
+ -- ^ A parser that takes
+ --
+ -- 1. an optional name of the series
+ -- 2. a map of tags
+ -- 3. an array of field names
+ -- 4. an array of values
+ --
+ -- to construct a value.
+ -> Value
+ -> A.Parser (Vector a)
+parseResultsWith = parseResultsWithDecoder lenientDecoder
+
+-- | Parse a JSON response with specified decoder settings.
+parseResultsWithDecoder
+ :: Decoder a
+ -> (Maybe Text -> HashMap Text Text -> Vector Text -> Array -> A.Parser a)
+ -- ^ A parser that takes
+ --
+ -- 1. an optional name of the series
+ -- 2. a map of tags
+ -- 3. an array of field names
+ -- 4. an array of values
+ --
+ -- to construct a value.
+ -> Value
+ -> A.Parser (Vector a)
+parseResultsWithDecoder Decoder {..} row val0 = success
+ where
+ success = do
+ results <- parseResultsObject val0
+
+ (join -> series) <- V.forM results $ \val ->
+ parseSeriesObject val <|> parseErrorObject val
+ values <- V.forM series $ \val -> do
+ (name, tags, columns, values) <- parseSeriesBody val
+ decodeFold $ V.forM values $ A.withArray "values" $ \fields -> do
+ assert (V.length columns == V.length fields) $ return ()
+ decodeEach $ row name tags columns fields
+ return $! join values
+
+-- | Decoder settings
+data Decoder a = forall b. Decoder
+ { decodeEach :: A.Parser a -> A.Parser b
+ -- ^ How to turn a parser for each element into another. For example, a
+ -- failure can be turned into 'Nothing'.
+ , decodeFold :: A.Parser (Vector b) -> A.Parser (Vector a)
+ -- ^ How to aggregate all results from 'decodeEach' into a vector of results.
+ }
+
+-- | Fail immediately if there's any parse failure.
+strictDecoder :: Decoder a
+strictDecoder = Decoder
+ { decodeEach = id
+ , decodeFold = id
+ }
+
+-- | Ignore parse failures and return successful results.
+lenientDecoder :: Decoder a
+lenientDecoder = Decoder
+ { decodeEach = optional
+ , decodeFold = \p -> do
+ bs <- p
+ return $! V.map fromJust $ V.filter isJust bs
+ }
+
+-- | Get a field value from a column name
+getField
+ :: Text -- ^ Column name
+ -> Vector Text -- ^ Columns
+ -> Array -- ^ Fields
+ -> A.Parser Value
+getField column columns fields =
+ case V.elemIndex column columns of
+ Nothing -> fail $ "getField: no such column " ++ show column
+ Just idx -> case V.indexM fields idx of
+ Nothing -> fail $ "getField: index out of bound for " ++ show column
+ Just field -> return field
+
+-- | Get a tag value from a tag name
+getTag
+ :: Monad m
+ => Text -- ^ Tag name
+ -> HashMap Text Text -- ^ Tags
+ -> m Text
+getTag tag tags = case HashMap.lookup tag tags of
+ Nothing -> fail $ "getTag: no such tag " ++ show tag
+ Just val -> return val
+
+parseResultsObject :: Value -> A.Parser (Vector A.Value)
+parseResultsObject = A.withObject "results" $ \obj -> obj .: "results"
+
+parseSeriesObject :: Value -> A.Parser (Vector A.Value)
+parseSeriesObject = A.withObject "series" $ \obj ->
+ fromMaybe V.empty <$> obj .:? "series"
+
+parseSeriesBody
+ :: Value
+ -> A.Parser (Maybe Text, HashMap Text Text, Vector Text, Array)
+parseSeriesBody = A.withObject "series" $ \obj -> do
+ !name <- obj .:? "name"
+ !columns <- obj .: "columns"
+ !values <- obj .: "values"
+ !tags <- obj .:? "tags" .!= HashMap.empty
+ return (name, tags, columns, values)
+
+parseErrorObject :: A.Value -> A.Parser a
+parseErrorObject = A.withObject "error" $ \obj -> do
+ message <- obj .: "error"
+ fail $ T.unpack message
+
+-- | Parse either a POSIX timestamp or RFC3339 formatted timestamp.
+parseTimestamp :: Precision ty -> A.Value -> A.Parser POSIXTime
+parseTimestamp prec val = case prec of
+ RFC3339 -> utcTimeToPOSIXSeconds <$!> parseRFC3339 val
+ _ -> parsePOSIXTime prec val
+
+-- | Parse an integer POSIX timestamp in given time precision.
+parsePOSIXTime :: Precision ty -> A.Value -> A.Parser POSIXTime
+parsePOSIXTime prec val = case prec of
+ RFC3339 -> A.typeMismatch err val
+ _ -> A.withScientific err
+ (\s -> case timestampToUTC s of
+ Nothing -> A.typeMismatch err val
+ Just !utc -> return utc)
+ val
+ where
+ err = "POSIX timestamp in " ++ T.unpack (precisionName prec)
+ timestampToUTC s = do
+ n <- Sci.toBoundedInteger s
+ return $! fromIntegral (n :: Int) * precisionScale prec
+
+-- | Parse a RFC3339-formatted timestamp.
+--
+-- Note that this parser is slow as it converts a 'T.Text' input to a 'String'
+-- before parsing.
+parseRFC3339 :: ParseTime time => A.Value -> A.Parser time
+parseRFC3339 val = A.withText err
+ (\text -> maybe (A.typeMismatch err val) (return $!) $
+ parseTimeM True defaultTimeLocale fmt $ T.unpack text)
+ val
+ where
+ fmt, err :: String
+ fmt = "%FT%X%QZ"
+ err = "RFC3339-formatted timestamp"
+
+parseFieldValue :: A.Value -> A.Parser FieldValue
+parseFieldValue val = case val of
+ A.Number sci ->
+ return $! either FieldFloat FieldInt $ Sci.floatingOrInteger sci
+ A.String txt ->
+ return $! FieldString txt
+ A.Bool b ->
+ return $! FieldBool b
+ A.Null ->
+ return FieldNull
+ _ -> fail "parseFieldValue: expected a flat data structure"
diff --git a/src/Database/InfluxDB/Lens.hs b/src/Database/InfluxDB/Lens.hs
deleted file mode 100644
index a7ae64a..0000000
--- a/src/Database/InfluxDB/Lens.hs
+++ /dev/null
@@ -1,66 +0,0 @@
-{-# LANGUAGE RankNTypes #-}
-module Database.InfluxDB.Lens
- ( Lens, Lens'
-
- -- * Lenses for 'Config'
- , credentials
- , httpManager
-
- -- * Lenses for 'Credentials'
- , user, password
-
- -- * Lenses for 'Server'
- , host, port, ssl
- ) where
-import Control.Applicative
-import Data.Text (Text)
-import Prelude
-
-import Network.HTTP.Client (Manager)
-
-import Database.InfluxDB.Http
-
-type Lens s t a b = forall f. Functor f => (a -> f b) -> s -> f t
-type Lens' s a = Lens s s a a
-
--- | User credentials for authentication
-credentials :: Lens' Config Credentials
-credentials f r = set <$> f (configCreds r)
- where
- set c = r { configCreds = c }
-
--- | An instance of 'Manager' from @http-client@ package
-httpManager :: Lens' Config Manager
-httpManager f c = set <$> f (configHttpManager c)
- where
- set m = c { configHttpManager = m }
-
--- | User name to be used for authentication
-user :: Lens' Credentials Text
-user f c = set <$> f (credsUser c)
- where
- set u = c { credsUser = u }
-
--- | Password to be used for authentication
-password :: Lens' Credentials Text
-password f s = set <$> f (credsPassword s)
- where
- set p = s { credsPassword = p }
-
--- | Host name or IP address of an InfluxDB
-host :: Lens' Server Text
-host f s = set <$> f (serverHost s)
- where
- set h = s { serverHost = h }
-
--- | Port number to be used to connect to an InfluxDB
-port :: Lens' Server Int
-port f s = set <$> f (serverPort s)
- where
- set p = s { serverPort = p }
-
--- | Whether or not to enable SSL connection
-ssl :: Lens' Server Bool
-ssl f s = set <$> f (serverSsl s)
- where
- set s' = s { serverSsl = s' }
diff --git a/src/Database/InfluxDB/Line.hs b/src/Database/InfluxDB/Line.hs
new file mode 100644
index 0000000..6597cd6
--- /dev/null
+++ b/src/Database/InfluxDB/Line.hs
@@ -0,0 +1,124 @@
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE KindSignatures #-}
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE RecordWildCards #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TemplateHaskell #-}
+module Database.InfluxDB.Line
+ ( Line(Line)
+ , measurement
+ , tagSet
+ , fieldSet
+ , timestamp
+
+ , buildLine
+ , buildLines
+ , encodeLine
+ , encodeLines
+ ) where
+import Data.List (intersperse)
+import Data.Int (Int64)
+import Data.Monoid
+
+import Control.Lens
+import Data.Map (Map)
+import Data.Text (Text)
+import qualified Data.ByteString.Builder as B
+import qualified Data.ByteString.Lazy as L
+import qualified Data.Map.Strict as Map
+import qualified Data.Text as T
+import qualified Data.Text.Encoding as TE
+
+import Database.InfluxDB.Types
+
+-- | Placeholder for the Line Protocol
+--
+-- See https://docs.influxdata.com/influxdb/v1.0/write_protocols/line_protocol_tutorial/ for the
+-- concrete syntax.
+data Line time = Line
+ { _measurement :: !Key
+ -- ^ Measurement name
+ , _tagSet :: !(Map Key Text)
+ -- ^ Set of tags (optional)
+ , _fieldSet :: !(Map Key FieldValue)
+ -- ^ Set of fields
+ --
+ -- It shouldn't be empty.
+ , _timestamp :: !(Maybe time)
+ -- ^ Timestamp (optional)
+ }
+
+encodeLine
+ :: (time -> Int64)
+ -> Line time
+ -> L.ByteString
+encodeLine toTimestamp = B.toLazyByteString . buildLine toTimestamp
+
+encodeLines
+ :: Traversable f
+ => (time -> Int64)
+ -> f (Line time)
+ -> L.ByteString
+encodeLines toTimestamp = B.toLazyByteString . buildLines toTimestamp
+
+buildLine
+ :: (time -> Int64)
+ -> Line time
+ -> B.Builder
+buildLine toTimestamp Line {..} =
+ key <> " " <> fields <> maybe "" (" " <>) timestamp
+ where
+ measurement = buildKey _measurement
+ tags = buildMap TE.encodeUtf8Builder _tagSet
+ key = if Map.null _tagSet
+ then measurement
+ else measurement <> "," <> tags
+ fields = buildMap buildFieldValue _fieldSet
+ timestamp = B.int64Dec . toTimestamp <$> _timestamp
+ buildMap encodeVal =
+ mconcat . intersperse "," . map encodeKeyVal . Map.toList
+ where
+ encodeKeyVal (name, val) = mconcat
+ [ buildKey name
+ , "="
+ , encodeVal val
+ ]
+
+buildKey :: Key -> B.Builder
+buildKey = TE.encodeUtf8Builder . escapeKey
+
+escapeKey :: Key -> Text
+escapeKey (Key text) = T.replace " " "\\ " $ T.replace "," "\\," text
+
+buildFieldValue :: FieldValue -> B.Builder
+buildFieldValue = \case
+ FieldInt i -> B.int64Dec i <> "i"
+ FieldFloat d -> B.doubleDec d
+ FieldString t -> "\"" <> TE.encodeUtf8Builder t <> "\""
+ FieldBool b -> if b then "true" else "false"
+ FieldNull -> "null"
+
+buildLines
+ :: Traversable f
+ => (time -> Int64)
+ -> f (Line time)
+ -> B.Builder
+buildLines toTimestamp = foldMap ((<> "\n") . buildLine toTimestamp)
+
+makeLensesWith (lensRules & generateSignatures .~ False) ''Line
+
+-- | Name of the measurement that you want to write your data to.
+measurement :: Lens' (Line time) Key
+
+-- | Tag(s) that you want to include with your data point. Tags are optional in
+-- the Line Protocol, so you can set it 'empty'.
+tagSet :: Lens' (Line time) (Map Key Text)
+
+-- | Field(s) for your data point. Every data point requires at least one field
+-- in the Line Protocol, so it shouldn't be 'empty'.
+fieldSet :: Lens' (Line time) (Map Key FieldValue)
+
+-- | Timestamp for your data point. You can put whatever type of timestamp that
+-- is an instance of the 'Timestamp' class.
+timestamp :: Lens' (Line time) (Maybe time)
diff --git a/src/Database/InfluxDB/Manage.hs b/src/Database/InfluxDB/Manage.hs
new file mode 100644
index 0000000..9042f6e
--- /dev/null
+++ b/src/Database/InfluxDB/Manage.hs
@@ -0,0 +1,158 @@
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE TemplateHaskell #-}
+{-# LANGUAGE ViewPatterns #-}
+{-# LANGUAGE RecordWildCards #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+module Database.InfluxDB.Manage
+ ( manage
+
+ , ShowQuery
+ , qid
+ , queryText
+ , Types.database
+ , duration
+
+ , ShowSeries
+ , key
+ ) where
+import Control.Applicative
+import Control.Exception
+import Control.Monad
+
+import Control.Lens
+import Data.Aeson
+import Data.Scientific (toBoundedInteger)
+import Data.Text (Text)
+import Data.Time.Clock
+import Data.Void
+import qualified Data.Aeson.Types as A
+import qualified Data.Attoparsec.Combinator as AC
+import qualified Data.Attoparsec.Text as AT
+import qualified Data.Text.Encoding as TE
+import qualified Data.Vector as V
+import qualified Network.HTTP.Types as HT
+
+import Database.InfluxDB.Types as Types
+import Database.InfluxDB.Query hiding (query)
+import qualified Database.InfluxDB.Format as F
+import qualified Network.HTTP.Client.Compat as HC
+
+-- | Send a database management query to InfluxDB.
+manage :: QueryParams -> Query -> IO ()
+manage params q = do
+ manager' <- either HC.newManager return $ params^.manager
+ response <- HC.httpLbs request manager'
+ let body = HC.responseBody response
+ case eitherDecode' body of
+ Left message -> do
+ throwIO $ IllformedJSON message body
+ Right val -> case A.parse (parseResults (params^.precision)) val of
+ A.Success (_ :: V.Vector Void) -> return ()
+ A.Error message -> do
+ let status = HC.responseStatus response
+ when (HT.statusIsServerError status) $
+ throwIO $ ServerError message
+ when (HT.statusIsClientError status) $
+ throwIO $ BadRequest message request
+ fail $ "BUG: " ++ message ++ " in Database.InfluxDB.Manage.manage"
+ where
+ request = HC.setQueryString qs $ manageRequest params
+ qs =
+ [ ("q", Just $ F.fromQuery q)
+ ]
+
+manageRequest :: QueryParams -> HC.Request
+manageRequest params = HC.defaultRequest
+ { HC.host = TE.encodeUtf8 $ params^.server.host
+ , HC.port = fromIntegral $ params^.server.port
+ , HC.secure = params^.server.ssl
+ , HC.method = "POST"
+ , HC.path = "/query"
+ }
+ where
+ Server {..} = params^.server
+
+data ShowQuery = ShowQuery
+ { _qid :: !Int
+ , _queryText :: !Query
+ , _database :: !Database
+ , _duration :: !NominalDiffTime
+ } deriving Show
+
+instance QueryResults ShowQuery where
+ parseResults _ = parseResultsWith $ \_ _ columns fields ->
+ maybe (fail "parseResults: parse error") return $ do
+ Number (toBoundedInteger -> Just _qid) <-
+ V.elemIndex "qid" columns >>= V.indexM fields
+ String (F.formatQuery F.text -> _queryText) <-
+ V.elemIndex "query" columns >>= V.indexM fields
+ String (F.formatDatabase F.text -> _database) <-
+ V.elemIndex "database" columns >>= V.indexM fields
+ String (parseDuration -> Right _duration) <-
+ V.elemIndex "duration" columns >>= V.indexM fields
+ return ShowQuery {..}
+
+parseDuration :: Text -> Either String NominalDiffTime
+parseDuration = AT.parseOnly $ sum <$!> durations
+ where
+ durations = some $ (*)
+ <$> fmap fromIntegral int
+ <*> unit
+ where
+ int :: AT.Parser Int
+ int = AT.decimal
+ unit = AC.choice
+ [ 10^^(-6 :: Int) <$ AT.char 'u'
+ , 1 <$ AT.char 's'
+ , 60 <$ AT.char 'm'
+ , 3600 <$ AT.char 'h'
+ ]
+
+newtype ShowSeries = ShowSeries
+ { _key :: Key
+ } deriving Show
+
+instance QueryResults ShowSeries where
+ parseResults _ = parseResultsWith $ \_ _ columns fields ->
+ ShowSeries <$> parseKey "key" columns fields
+
+makeLensesWith (lensRules & generateSignatures .~ False) ''ShowQuery
+
+-- | Query ID
+--
+-- >>> v <- query (queryParams "_internal") "SHOW QUERIES" :: IO (V.Vector ShowQuery)
+-- >>> v ^.. each.qid
+-- [149250]
+qid :: Lens' ShowQuery Int
+
+-- | Query text
+--
+-- >>> v <- query (queryParams "_internal") "SHOW QUERIES" :: IO (V.Vector ShowQuery)
+-- >>> v ^.. each.queryText
+-- ["SHOW QUERIES"]
+queryText :: Lens' ShowQuery Query
+
+database :: Lens' ShowQuery Database
+
+-- |
+-- >>> v <- query (queryParams "_internal") "SHOW QUERIES" :: IO (V.Vector ShowQuery)
+-- >>> v ^.. each.database
+-- ["_internal"]
+instance HasDatabase ShowQuery where
+ database = Database.InfluxDB.Manage.database
+
+-- | Duration of the query
+--
+-- >>> v <- query (queryParams "_internal") "SHOW QUERIES" :: IO (V.Vector ShowQuery)
+-- >>> v ^.. each.duration
+-- [0.06062s]
+duration :: Lens' ShowQuery NominalDiffTime
+
+makeLensesWith (lensRules & generateSignatures .~ False) ''ShowSeries
+
+-- | Series name
+--
+-- >>> v <- query (queryParams "_internal") "SHOW SERIES" :: IO (V.Vector ShowSeries)
+-- >>> length $ v ^.. each.key
+-- 755
+key :: Lens' ShowSeries Key
diff --git a/src/Database/InfluxDB/Ping.hs b/src/Database/InfluxDB/Ping.hs
new file mode 100644
index 0000000..cb75b31
--- /dev/null
+++ b/src/Database/InfluxDB/Ping.hs
@@ -0,0 +1,101 @@
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE RecordWildCards #-}
+{-# LANGUAGE TemplateHaskell #-}
+
+module Database.InfluxDB.Ping
+ (
+ -- * Ping interface
+ ping
+
+ -- * Ping parameters
+ , PingParams(..)
+ , pingParams
+ , Types.server
+ , Types.manager
+ , waitForLeader
+
+ -- * Ping result
+ , PingResult(..)
+ , roundtripTime
+ , influxdbVersion
+ ) where
+
+import Control.Lens
+import qualified Data.ByteString as BS
+import qualified Data.Text.Encoding as TE
+import qualified Network.HTTP.Client.Compat as HC
+import System.Clock
+
+import Database.InfluxDB.Types as Types
+
+
+-- Ping requests do not require authentication
+-- | The full set of parameters for the ping API
+data PingParams = PingParams
+ { _server :: !Server
+ , _manager :: !(Either HC.ManagerSettings HC.Manager)
+ -- ^ HTTP connection manager
+ , _waitForLeader :: !(Maybe Int)
+ -- ^ the number of seconds to wait
+ }
+
+makeLensesWith (lensRules & generateSignatures .~ False) ''PingParams
+
+server :: Lens' PingParams Server
+
+instance HasServer PingParams where
+ server = Database.InfluxDB.Ping.server
+
+manager :: Lens' PingParams (Either HC.ManagerSettings HC.Manager)
+
+instance HasManager PingParams where
+ manager = Database.InfluxDB.Ping.manager
+
+-- | The number of seconds to wait before returning a response
+waitForLeader :: Lens' PingParams (Maybe Int)
+
+pingParams :: PingParams
+pingParams =
+ PingParams
+ { _server = localServer
+ , _manager = Left HC.defaultManagerSettings
+ , _waitForLeader = Nothing
+ }
+
+pingRequest :: PingParams -> HC.Request
+pingRequest PingParams {..} = HC.defaultRequest
+ { HC.host = TE.encodeUtf8 _host
+ , HC.port = fromIntegral _port
+ , HC.secure = _ssl
+ , HC.method = "GET"
+ , HC.path = "/ping"
+ }
+ where
+ Server {..} = _server
+
+data PingResult = PingResult
+ { _roundtripTime :: !TimeSpec
+ , _influxdbVersion :: !BS.ByteString
+ } deriving (Show, Eq, Ord)
+
+makeLensesWith (lensRules & generateSignatures .~ False) ''PingResult
+
+-- | Roundtrip time of the ping
+roundtripTime :: Lens' PingResult TimeSpec
+
+-- | Version string returned by the InfluxDB header
+influxdbVersion :: Lens' PingResult BS.ByteString
+
+ping :: PingParams -> IO PingResult
+ping params = do
+ manager' <- either HC.newManager return $ _manager params
+ startTime <- getTime'
+ HC.withResponse request manager' (\response -> do
+ endTime <- getTime'
+ let headers = HC.responseHeaders response
+ case lookup "X-Influxdb-Version" headers of
+ Just version -> pure (PingResult (diffTimeSpec endTime startTime) version)
+ Nothing -> error "A response by influxdb should always contain a version header.")
+ where
+ request = pingRequest params
+ getTime' = getTime Monotonic
diff --git a/src/Database/InfluxDB/Query.hs b/src/Database/InfluxDB/Query.hs
new file mode 100644
index 0000000..19f261d
--- /dev/null
+++ b/src/Database/InfluxDB/Query.hs
@@ -0,0 +1,367 @@
+{-# LANGUAGE BangPatterns #-}
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE FlexibleInstances #-}
+{-# LANGUAGE GADTs #-}
+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE MultiParamTypeClasses #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE RecordWildCards #-}
+{-# LANGUAGE TemplateHaskell #-}
+{-# LANGUAGE ViewPatterns #-}
+module Database.InfluxDB.Query
+ (
+ -- * Query interface
+ Query
+ , query
+ , queryChunked
+
+ -- * Query parameters
+ , QueryParams
+ , queryParams
+ , Types.server
+ , Types.database
+ , Types.precision
+ , Types.manager
+
+ -- * Parsing results
+ , QueryResults(..)
+ , parseResultsWith
+ , parseKey
+
+ -- * Low-level functions
+ , withQueryResponse
+ ) where
+import Control.Exception
+import Control.Monad
+import Text.Printf
+
+import Control.Lens
+import Data.Aeson
+import Data.Optional (Optional(..), optional)
+import Data.Text (Text)
+import Data.Vector (Vector)
+import Data.Void
+import qualified Control.Foldl as L
+import qualified Data.Aeson.Parser as A
+import qualified Data.Aeson.Types as A
+import qualified Data.Attoparsec.ByteString as AB
+import qualified Data.ByteString as B
+import qualified Data.ByteString.Builder as BB
+import qualified Data.ByteString.Lazy as BL
+import qualified Data.Text.Encoding as TE
+import qualified Data.Text as T
+import qualified Data.Vector as V
+import qualified Network.HTTP.Types as HT
+
+import Database.InfluxDB.JSON
+import Database.InfluxDB.Types as Types
+import qualified Database.InfluxDB.Format as F
+import qualified Network.HTTP.Client.Compat as HC
+
+class QueryResults a where
+ parseResults
+ :: Precision 'QueryRequest
+ -> Value
+ -> A.Parser (Vector a)
+
+instance QueryResults Void where
+ parseResults _ = A.withObject "error" $ \obj -> obj .:? "error"
+ >>= maybe (pure V.empty) (withText "error" $ fail . T.unpack)
+
+instance (a ~ Value, b ~ Value) => QueryResults (a, b) where
+ parseResults _ = parseResultsWith $ \_ _ _ fields ->
+ maybe (fail $ "invalid fields: " ++ show fields) return $ do
+ a <- fields V.!? 0
+ b <- fields V.!? 1
+ return (a, b)
+
+instance (a ~ Value, b ~ Value, c ~ Value)
+ => QueryResults (a, b, c) where
+ parseResults _ = parseResultsWith $ \_ _ _ fields ->
+ maybe (fail $ "invalid fields: " ++ show fields) return $ do
+ a <- fields V.!? 0
+ b <- fields V.!? 1
+ c <- fields V.!? 2
+ return (a, b, c)
+
+instance (a ~ Value, b ~ Value, c ~ Value, d ~ Value)
+ => QueryResults (a, b, c, d) where
+ parseResults _ = parseResultsWith $ \_ _ _ fields ->
+ maybe (fail $ "invalid fields: " ++ show fields) return $ do
+ a <- fields V.!? 0
+ b <- fields V.!? 1
+ c <- fields V.!? 2
+ d <- fields V.!? 3
+ return (a, b, c, d)
+
+instance (a ~ Value, b ~ Value, c ~ Value, d ~ Value, e ~ Value)
+ => QueryResults (a, b, c, d, e) where
+ parseResults _ = parseResultsWith $ \_ _ _ fields ->
+ maybe (fail $ "invalid fields: " ++ show fields) return $ do
+ a <- fields V.!? 0
+ b <- fields V.!? 1
+ c <- fields V.!? 2
+ d <- fields V.!? 3
+ e <- fields V.!? 4
+ return (a, b, c, d, e)
+
+instance (a ~ Value, b ~ Value, c ~ Value, d ~ Value, e ~ Value, f ~ Value)
+ => QueryResults (a, b, c, d, e, f) where
+ parseResults _ = parseResultsWith $ \_ _ _ fields ->
+ maybe (fail $ "invalid fields: " ++ show fields) return $ do
+ a <- fields V.!? 0
+ b <- fields V.!? 1
+ c <- fields V.!? 2
+ d <- fields V.!? 3
+ e <- fields V.!? 4
+ f <- fields V.!? 5
+ return (a, b, c, d, e, f)
+
+instance
+ ( a ~ Value, b ~ Value, c ~ Value, d ~ Value, e ~ Value, f ~ Value
+ , g ~ Value )
+ => QueryResults (a, b, c, d, e, f, g) where
+ parseResults _ = parseResultsWith $ \_ _ _ fields ->
+ maybe (fail $ "invalid fields: " ++ show fields) return $ do
+ a <- fields V.!? 0
+ b <- fields V.!? 1
+ c <- fields V.!? 2
+ d <- fields V.!? 3
+ e <- fields V.!? 4
+ f <- fields V.!? 5
+ g <- fields V.!? 6
+ return (a, b, c, d, e, f, g)
+
+instance
+ ( a ~ Value, b ~ Value, c ~ Value, d ~ Value, e ~ Value, f ~ Value
+ , g ~ Value, h ~ Value )
+ => QueryResults (a, b, c, d, e, f, g, h) where
+ parseResults _ = parseResultsWith $ \_ _ _ fields ->
+ maybe (fail $ "invalid fields: " ++ show fields) return $ do
+ a <- fields V.!? 0
+ b <- fields V.!? 1
+ c <- fields V.!? 2
+ d <- fields V.!? 3
+ e <- fields V.!? 4
+ f <- fields V.!? 5
+ g <- fields V.!? 6
+ h <- fields V.!? 7
+ return (a, b, c, d, e, f, g, h)
+
+parseKey :: Key -> Vector Text -> Array -> A.Parser Key
+parseKey (Key name) columns fields = do
+ case V.elemIndex name columns >>= V.indexM fields of
+ Just (String (F.formatKey F.text -> key)) -> return key
+ _ -> fail $ printf "parseKey: %s not found in columns" $ show name
+
+-- | The full set of parameters for the query API
+data QueryParams = QueryParams
+ { _server :: !Server
+ , _database :: !Database
+ , _precision :: !(Precision 'QueryRequest)
+ -- ^ Timestamp precision
+ --
+ -- InfluxDB uses nanosecond precision if nothing is specified.
+ , _authentication :: !(Maybe Credentials)
+ -- ^ No authentication by default
+ , _manager :: !(Either HC.ManagerSettings HC.Manager)
+ -- ^ HTTP connection manager
+ }
+
+-- | Smart constructor for 'QueryParams'
+--
+-- Default parameters:
+--
+-- ['L.server'] 'localServer'
+-- ['L.precision'] 'RFC3339'
+-- ['authentication'] 'Nothing'
+-- ['L.manager'] @'Left' 'HC.defaultManagerSettings'@
+queryParams :: Database -> QueryParams
+queryParams _database = QueryParams
+ { _server = localServer
+ , _precision = RFC3339
+ , _authentication = Nothing
+ , _manager = Left HC.defaultManagerSettings
+ , ..
+ }
+
+-- | Query data from InfluxDB.
+--
+-- It may throw 'InfluxException'.
+query :: QueryResults a => QueryParams -> Query -> IO (Vector a)
+query params q = withQueryResponse params Nothing q go
+ where
+ go request response = do
+ chunks <- HC.brConsume $ HC.responseBody response
+ let body = BL.fromChunks chunks
+ case eitherDecode' body of
+ Left message -> throwIO $ IllformedJSON message body
+ Right val -> case A.parse (parseResults (_precision params)) val of
+ A.Success vec -> return vec
+ A.Error message ->
+ errorQuery request response message
+
+setPrecision
+ :: Precision 'QueryRequest
+ -> [(B.ByteString, Maybe B.ByteString)]
+ -> [(B.ByteString, Maybe B.ByteString)]
+setPrecision prec qs = maybe qs (\p -> ("epoch", Just p):qs) $
+ precisionParam prec
+
+precisionParam :: Precision 'QueryRequest -> Maybe B.ByteString
+precisionParam = \case
+ Nanosecond -> return "ns"
+ Microsecond -> return "u"
+ Millisecond -> return "ms"
+ Second -> return "s"
+ Minute -> return "m"
+ Hour -> return "h"
+ RFC3339 -> Nothing
+
+-- | Same as 'query' but it instructs InfluxDB to stream chunked responses
+-- rather than returning a huge JSON object. This can be lot more efficient than
+-- 'query' if the result is huge.
+--
+-- It may throw 'InfluxException'.
+queryChunked
+ :: QueryResults a
+ => QueryParams
+ -> Optional Int
+ -- ^ Chunk size
+ --
+ -- By 'Default', InfluxDB chunks responses by series or by every 10,000
+ -- points, whichever occurs first. If it set to a 'Specific' value, InfluxDB
+ -- chunks responses by series or by that number of points.
+ -> Query
+ -> L.FoldM IO (Vector a) r
+ -> IO r
+queryChunked params chunkSize q (L.FoldM step initialize extract) =
+ withQueryResponse params (Just chunkSize) q go
+ where
+ go request response = do
+ x0 <- initialize
+ chunk0 <- HC.responseBody response
+ x <- loop x0 k0 chunk0
+ extract x
+ where
+ k0 = AB.parse A.json
+ loop x k chunk
+ | B.null chunk = return x
+ | otherwise = case k chunk of
+ AB.Fail unconsumed _contexts message ->
+ throwIO $ IllformedJSON message $ BL.fromStrict unconsumed
+ AB.Partial k' -> do
+ chunk' <- HC.responseBody response
+ loop x k' chunk'
+ AB.Done leftover val ->
+ case A.parse (parseResults (_precision params)) val of
+ A.Success vec -> do
+ x' <- step x vec
+ loop x' k0 leftover
+ A.Error message ->
+ errorQuery request response message
+
+withQueryResponse
+ :: QueryParams
+ -> Maybe (Optional Int)
+ -- ^ Chunk size
+ --
+ -- By 'Nothing', InfluxDB returns all matching data points at once.
+ -- By @'Just' 'Default'@, InfluxDB chunks responses by series or by every
+ -- 10,000 points, whichever occurs first. If it set to a 'Specific' value,
+ -- InfluxDB chunks responses by series or by that number of points.
+ -> Query
+ -> (HC.Request -> HC.Response HC.BodyReader -> IO r)
+ -> IO r
+withQueryResponse params chunkSize q f = do
+ manager' <- either HC.newManager return $ _manager params
+ HC.withResponse request manager' (f request)
+ where
+ request =
+ HC.setQueryString (setPrecision (_precision params) queryString) $
+ queryRequest params
+ queryString = addChunkedParam
+ [ ("q", Just $ F.fromQuery q)
+ , ("db", Just db)
+ ]
+ where
+ !db = TE.encodeUtf8 $ databaseName $ _database params
+ addChunkedParam ps = case chunkSize of
+ Nothing -> ps
+ Just size ->
+ let !chunked = optional "true" (decodeChunkSize . max 1) size
+ in ("chunked", Just chunked) : ps
+ where
+ decodeChunkSize = BL.toStrict . BB.toLazyByteString . BB.intDec
+
+
+queryRequest :: QueryParams -> HC.Request
+queryRequest QueryParams {..} = HC.defaultRequest
+ { HC.host = TE.encodeUtf8 _host
+ , HC.port = fromIntegral _port
+ , HC.secure = _ssl
+ , HC.method = "GET"
+ , HC.path = "/query"
+ }
+ where
+ Server {..} = _server
+
+errorQuery :: HC.Request -> HC.Response body -> String -> IO a
+errorQuery request response message = do
+ let status = HC.responseStatus response
+ when (HT.statusIsServerError status) $
+ throwIO $ ServerError message
+ when (HT.statusIsClientError status) $
+ throwIO $ BadRequest message request
+ fail $ "BUG: " ++ message ++ " in Database.InfluxDB.Query.query - "
+ ++ show request
+
+makeLensesWith (lensRules & generateSignatures .~ False) ''QueryParams
+
+server :: Lens' QueryParams Server
+
+-- |
+-- >>> let p = queryParams "foo"
+-- >>> p ^. server.host
+-- "localhost"
+instance HasServer QueryParams where
+ server = Database.InfluxDB.Query.server
+
+database :: Lens' QueryParams Database
+
+-- |
+-- >>> let p = queryParams "foo"
+-- >>> p ^. database
+-- "foo"
+instance HasDatabase QueryParams where
+ database = Database.InfluxDB.Query.database
+
+precision :: Lens' QueryParams (Precision 'QueryRequest)
+
+-- | Returning JSON responses contain timestamps in the specified
+-- precision/format.
+--
+-- >>> let p = queryParams "foo"
+-- >>> p ^. precision
+-- Nanosecond
+instance HasPrecision 'QueryRequest QueryParams where
+ precision = Database.InfluxDB.Query.precision
+
+manager :: Lens' QueryParams (Either HC.ManagerSettings HC.Manager)
+
+-- |
+-- >>> let p = queryParams "foo"
+-- >>> p & manager .~ Left HC.defaultManagerSettings
+instance HasManager QueryParams where
+ manager = Database.InfluxDB.Query.manager
+
+-- | Authentication info for the query
+--
+-- >>> let p = queryParams "foo"
+-- >>> p ^. authentication
+-- Nothing
+authentication :: Lens' QueryParams (Maybe Credentials)
+
+instance HasCredentials QueryParams where
+ authentication = Database.InfluxDB.Query.authentication
diff --git a/src/Database/InfluxDB/Stream.hs b/src/Database/InfluxDB/Stream.hs
deleted file mode 100644
index 7820616..0000000
--- a/src/Database/InfluxDB/Stream.hs
+++ /dev/null
@@ -1,41 +0,0 @@
-{-# LANGUAGE BangPatterns #-}
-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)
-
--- | Monadic left fold for 'Stream'.
-fold :: Monad m => (b -> a -> m b) -> b -> Stream m a -> m b
-fold f = loop
- where
- loop z stream = case stream of
- Done -> return z
- Yield a nextStream -> do
- b <- f z a
- stream' <- nextStream
- loop b stream'
-
--- | Strict version of 'fold'.
-fold' :: Monad m => (b -> a -> m b) -> b -> Stream m a -> m b
-fold' f = loop
- where
- loop z stream = case stream of
- Done -> return z
- Yield a nextStream -> do
- !b <- f z a
- stream' <- nextStream
- loop b stream'
diff --git a/src/Database/InfluxDB/TH.hs b/src/Database/InfluxDB/TH.hs
deleted file mode 100644
index ab86e09..0000000
--- a/src/Database/InfluxDB/TH.hs
+++ /dev/null
@@ -1,141 +0,0 @@
-{-# LANGUAGE CPP #-}
-{-# LANGUAGE TemplateHaskell #-}
-
-#if __GLASGOW_HASKELL__ == 704
-{-# LANGUAGE ConstraintKinds #-}
-#endif
-
-module Database.InfluxDB.TH
- ( Options(..), defaultOptions
- , deriveSeriesData
- , deriveToSeriesData
- , deriveFromSeriesData
-
- , stripPrefixLower
- , stripPrefixSnake
- ) where
-import Control.Applicative
-import Language.Haskell.TH
-import Language.Haskell.TH.Syntax (VarStrictType)
-import Prelude
-
-import qualified Data.Vector as V
-
-import Database.InfluxDB.Decode
-import Database.InfluxDB.Encode
-import Database.InfluxDB.Types.Internal
-
-data Options = Options
- { fieldLabelModifier :: String -> String
- }
-
-defaultOptions :: Options
-defaultOptions = Options
- { fieldLabelModifier = id
- }
-
-deriveSeriesData :: Options -> Name -> Q [Dec]
-deriveSeriesData opts name = (++)
- <$> deriveToSeriesData opts name
- <*> deriveFromSeriesData opts name
-
-deriveToSeriesData :: Options -> Name -> Q [Dec]
-deriveToSeriesData opts name = do
- info <- reify name
- case info of
- TyConI dec -> pure <$> deriveWith toSeriesDataBody opts dec
- _ -> fail $ "Expected a type constructor, but got " ++ show info
-
-deriveFromSeriesData :: Options -> Name -> Q [Dec]
-deriveFromSeriesData opts name = do
- info <- reify name
- case info of
- TyConI dec -> pure <$> deriveWith fromSeriesDataBody opts dec
- _ -> fail $ "Expected a type constructor, but got " ++ show info
-
-deriveWith
- :: (Options -> Name -> [TyVarBndr] -> Con -> Q Dec)
- -> Options -> Dec -> Q Dec
-deriveWith f opts dec = case dec of
-#if MIN_VERSION_template_haskell(2, 11, 0)
- DataD _ tyName tyVars _ [con] _ -> f opts tyName tyVars con
- NewtypeD _ tyName tyVars _ con _ -> f opts tyName tyVars con
-#else
- DataD _ tyName tyVars [con] _ -> f opts tyName tyVars con
- NewtypeD _ tyName tyVars con _ -> f opts tyName tyVars con
-#endif
- _ -> fail $ "Expected a data or newtype declaration, but got " ++ show dec
-
-toSeriesDataBody :: Options -> Name -> [TyVarBndr] -> Con -> Q Dec
-toSeriesDataBody opts tyName tyVars con = do
- case con of
- RecC conName vars -> InstanceD
-#if MIN_VERSION_template_haskell(2, 11, 0)
- Nothing
-#endif
- <$> mapM tyVarToPred tyVars
- <*> [t| ToSeriesData $(conT tyName) |]
- <*> deriveDecs conName vars
- _ -> fail $ "Expected a record, but got " ++ show con
- where
- tyVarToPred tv = case tv of
-#if MIN_VERSION_template_haskell(2, 10, 0)
- PlainTV name -> conT ''FromValue `appT` varT name
- KindedTV name _ -> conT ''FromValue `appT` varT name
-#else
- PlainTV name -> classP ''FromValue [varT name]
- KindedTV name _ -> classP ''FromValue [varT name]
-#endif
- deriveDecs _conName vars = do
- a <- newName "a"
- sequence
- [ funD 'toSeriesColumns
- [ clause [wildP]
- (normalB [| V.fromList $(listE columns) |]) []
- ]
- , funD 'toSeriesPoints
- [ clause [varP a]
- (normalB [| V.fromList $(listE $ map (applyToValue a) vars) |]) []
- ]
- ]
- where
- applyToValue a (name, _, _) = [| toValue ($(varE name) $(varE a)) |]
- columns = map (varStrictTypeToColumn opts) vars
-
-fromSeriesDataBody :: Options -> Name -> [TyVarBndr] -> Con -> Q Dec
-fromSeriesDataBody opts tyName tyVars con = do
- case con of
- RecC conName vars -> instanceD
- (mapM tyVarToPred tyVars)
- [t| FromSeriesData $(conT tyName) |]
- [deriveDec conName vars]
- _ -> fail $ "Expected a record, but got " ++ show con
- where
- tyVarToPred tv = case tv of
-#if MIN_VERSION_template_haskell(2, 10, 0)
- PlainTV name -> conT ''FromValue `appT` varT name
- KindedTV name _ -> conT ''FromValue `appT` varT name
-#else
- PlainTV name -> classP ''FromValue [varT name]
- KindedTV name _ -> classP ''FromValue [varT name]
-#endif
- deriveDec conName vars = funD 'parseSeriesData
- [ clause [] (normalB deriveBody) []
- ]
- where
- deriveBody = do
- values <- newName "values"
- appE (varE 'withValues) $ lamE [varP values] $
- foldl (go values) [| pure $(conE conName) |] columns
- where
- go :: Name -> Q Exp -> Q Exp -> Q Exp
- go values expQ col = [| $expQ <*> $(varE values) .: $col |]
- columns = map (varStrictTypeToColumn opts) vars
-
-varStrictTypeToColumn :: Options -> VarStrictType -> Q Exp
-varStrictTypeToColumn opts = column opts . f
- where
- f (var, _, _) = var
-
-column :: Options -> Name -> Q Exp
-column opts = litE . stringL . fieldLabelModifier opts . nameBase
diff --git a/src/Database/InfluxDB/Types.hs b/src/Database/InfluxDB/Types.hs
index 9a46864..27fd118 100644
--- a/src/Database/InfluxDB/Types.hs
+++ b/src/Database/InfluxDB/Types.hs
@@ -1,302 +1,225 @@
-{-# LANGUAGE CPP #-}
+{-# LANGUAGE DataKinds #-}
{-# LANGUAGE DeriveDataTypeable #-}
{-# LANGUAGE DeriveGeneric #-}
-{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE FlexibleInstances #-}
+{-# LANGUAGE FunctionalDependencies #-}
+{-# LANGUAGE GADTs #-}
+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
+{-# LANGUAGE KindSignatures #-}
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE MultiParamTypeClasses #-}
{-# LANGUAGE OverloadedStrings #-}
-{-# LANGUAGE RankNTypes #-}
-{-# LANGUAGE RecordWildCards #-}
+{-# LANGUAGE StandaloneDeriving #-}
{-# 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(..)
- , User(..)
- , Admin(..)
- , Ping(..)
- , Interface
- , ShardSpace(..)
-
- -- * Server pool
- , ServerPool
- , serverRetryPolicy
- , newServerPool
- , newServerPoolWithRetryPolicy
- , activeServer
- , failover
-
- -- * Exceptions
- , InfluxException(..)
- , jsonDecodeError
- , seriesDecodeError
- ) where
-
-import Control.Applicative (empty)
-import Control.Exception (Exception, throwIO)
+module Database.InfluxDB.Types where
+import Control.Exception
import Data.Data (Data)
-import Data.IORef
import Data.Int (Int64)
-import Data.Monoid ((<>))
-import Data.Sequence (Seq, ViewL(..), (|>))
-import Data.Text (Text)
+import Data.String
import Data.Typeable (Typeable)
-import Data.Vector (Vector)
-import Data.Word (Word32)
import GHC.Generics (Generic)
-import qualified Data.Sequence as Seq
-
-import Control.Retry (RetryPolicy, limitRetries, exponentialBackoff)
-import Data.Aeson ((.=), (.:))
-import Data.Aeson.TH
-import qualified Data.Aeson as A
-
-import Database.InfluxDB.Types.Internal (stripPrefixOptions)
-
-#if MIN_VERSION_aeson(0, 7, 0)
-import Data.Scientific
-#else
-import Data.Attoparsec.Number
-#endif
-
------------------------------------------------------------
--- 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
- } deriving (Typeable, Generic)
-
--- | Convenient accessor for columns.
-seriesColumns :: Series -> Vector Column
-seriesColumns = seriesDataColumns . seriesData
-
--- | Convenient accessor for points.
-seriesPoints :: Series -> [Vector Value]
-seriesPoints = seriesDataPoints . seriesData
-
-instance A.ToJSON Series where
- toJSON Series {..} = A.object
- [ "name" .= seriesName
- , "columns" .= seriesDataColumns
- , "points" .= 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 = points
- }
- }
- parseJSON _ = empty
-
--- | 'SeriesData' consists of columns and points.
-data SeriesData = SeriesData
- { seriesDataColumns :: Vector Column
- , seriesDataPoints :: [Vector Value]
- } deriving (Eq, Show, Typeable, Generic)
-
-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, Generic)
-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 $! numberToValue
- where
-#if MIN_VERSION_aeson(0, 7, 0)
- numberToValue
- -- If the number is larger than Int64, it must be
- -- a float64 (Double in Haskell).
- | n > maxInt = Float $ toRealFloat n
- | e < 0 = Float $ realToFrac n
- | otherwise = Int $ fromIntegral $ coefficient n * 10 ^ e
- where
- e = base10Exponent n
-#if !MIN_VERSION_scientific(0, 3, 0)
- toRealFloat = realToFrac
--- scientific
-#endif
-#else
- numberToValue = case n of
- I i
- -- If the number is larger than Int64, it must be
- -- a float64 (Double in Haskell).
- | i > maxInt -> Float $ fromIntegral i
- | otherwise -> Int $ fromIntegral i
- D d -> Float d
--- aeson
-#endif
- maxInt = fromIntegral (maxBound :: Int64)
-
------------------------------------------------------------
-
--- | User credentials.
-data Credentials = Credentials
- { credsUser :: !Text
- , credsPassword :: !Text
- } deriving (Show, Typeable, Generic)
+import Control.Lens
+import Data.Text (Text)
+import Data.Time.Clock
+import Data.Time.Clock.POSIX
+import Network.HTTP.Client (Manager, ManagerSettings, Request)
+import qualified Data.ByteString.Lazy as BL
+import qualified Data.Text as T
+
+newtype Query = Query T.Text deriving IsString
+
+instance Show Query where
+ show (Query q) = show q
--- | 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, Typeable, Generic)
-
--- | 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.
- , serverRetryPolicy :: !RetryPolicy
+ { _host :: !Text
+ , _port :: !Int
+ , _ssl :: !Bool
+ } deriving (Show, Generic, Eq)
+
+-- | Default server settings.
+--
+-- Default parameters:
+--
+-- * 'host': @"localhost"@
+-- * 'port': @8086@
+-- * 'ssl': 'False'
+localServer :: Server
+localServer = Server
+ { _host = "localhost"
+ , _port = 8086
+ , _ssl = False
+ }
+
+makeLensesWith (lensRules & generateSignatures .~ False) ''Server
+
+-- | Host name of the server
+host :: Lens' Server Text
+
+-- | Port number of the server
+port :: Lens' Server Int
+
+-- | If SSL is enabled
+ssl :: Lens' Server Bool
+
+-- | User credentials
+data Credentials = Credentials
+ { _user :: !Text
+ , _password :: !Text
}
-newtype Database = Database
- { databaseName :: Text
- } deriving (Show, Typeable, Generic)
-
--- | User
-data User = User
- { userName :: Text
- , userIsAdmin :: Bool
- } deriving (Show, Typeable, Generic)
-
--- | Administrator
-newtype Admin = Admin
- { adminName :: Text
- } deriving (Show, Typeable, Generic)
-
-newtype Ping = Ping
- { pingStatus :: Text
- } deriving (Show, Typeable, Generic)
-
-type Interface = Text
-
-data ShardSpace = ShardSpace
- { shardSpaceDatabase :: Maybe Text
- , shardSpaceName :: Text
- , shardSpaceRegex :: Text
- , shardSpaceRetentionPolicy :: Text
- , shardSpaceShardDuration :: Text
- , shardSpaceReplicationFactor :: Word32
- , shardSpaceSplit :: Word32
- } deriving (Show, Typeable, Generic)
-
------------------------------------------------------------
--- 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 = newServerPoolWithRetryPolicy defaultRetryPolicy
- where
- defaultRetryPolicy :: RetryPolicy
- defaultRetryPolicy = limitRetries 5 <> exponentialBackoff 50
-
-newServerPoolWithRetryPolicy
- :: RetryPolicy -> Server -> [Server] -> IO (IORef ServerPool)
-newServerPoolWithRetryPolicy retryPolicy active backups =
- newIORef ServerPool
- { serverActive = active
- , serverBackup = Seq.fromList backups
- , serverRetryPolicy = retryPolicy
- }
-
--- | 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 -> (newPool, ())
- where
- newPool = pool
- { serverActive = active
- , serverBackup = rest |> serverActive
- }
-
------------------------------------------------------------
--- Exceptions
+makeLensesWith (lensRules & generateSignatures .~ False) ''Credentials
+-- | User name to access InfluxDB
+user :: Lens' Credentials Text
+
+-- | Password to access InfluxDB
+password :: Lens' Credentials Text
+
+-- | Database name
+newtype Database = Database { databaseName :: Text } deriving (Eq, Ord)
+
+-- | String type that is used for measurements, tag keys and field keys.
+newtype Key = Key Text deriving (Eq, Ord)
+
+instance IsString Database where
+ fromString xs = Database $ fromNonEmptyString "Database" xs
+
+instance IsString Key where
+ fromString xs = Key $ fromNonEmptyString "Key" xs
+
+fromNonEmptyString :: String -> String -> Text
+fromNonEmptyString ty xs
+ | null xs = error $ ty ++ " should never be empty"
+ | otherwise = fromString xs
+
+instance Show Database where
+ show (Database name) = show name
+
+instance Show Key where
+ show (Key name) = show name
+
+data FieldValue
+ = FieldInt !Int64
+ | FieldFloat !Double
+ | FieldString !Text
+ | FieldBool !Bool
+ | FieldNull
+ deriving (Eq, Show, Data, Typeable, Generic)
+
+instance IsString FieldValue where
+ fromString = FieldString . T.pack
+
+-- | Type of a request
+data RequestType
+ = QueryRequest
+ -- ^ Request for @/query@
+ | WriteRequest
+ -- ^ Request for @/write@
+ deriving Show
+
+-- | Predefined set of time precision.
+--
+-- 'RFC3339' is only available for 'QueryRequest's.
+data Precision (ty :: RequestType) where
+ -- | POSIX time in ns
+ Nanosecond :: Precision ty
+ -- | POSIX time in μs
+ Microsecond :: Precision ty
+ -- | POSIX time in ms
+ Millisecond :: Precision ty
+ -- | POSIX time in s
+ Second :: Precision ty
+ -- | POSIX time in minutes
+ Minute :: Precision ty
+ -- | POSIX time in hours
+ Hour :: Precision ty
+ -- | Nanosecond precision time in a human readable format, like
+ -- @2016-01-04T00:00:23.135623Z@. This is the default format for @/query@.
+ RFC3339 :: Precision 'QueryRequest
+
+deriving instance Show (Precision a)
+
+precisionName :: Precision ty -> Text
+precisionName = \case
+ Nanosecond -> "n"
+ Microsecond -> "u"
+ Millisecond -> "ms"
+ Second -> "s"
+ Minute -> "m"
+ Hour -> "h"
+ RFC3339 -> "rfc3339"
+
+-- | A 'Timestamp' is something that can be converted to a valid
+-- InfluxDB timestamp, which is represented as a 64-bit integer.
+class Timestamp time where
+ -- | Round a time to the given precision and scale it to nanoseconds
+ roundTo :: Precision 'WriteRequest -> time -> Int64
+ -- | Scale a time to the given precision
+ scaleTo :: Precision 'WriteRequest -> time -> Int64
+
+roundAt :: RealFrac a => a -> a -> a
+roundAt scale x = fromIntegral (round (x / scale) :: Int) * scale
+
+precisionScale :: Fractional a => Precision ty -> a
+precisionScale = \case
+ RFC3339 -> 10^^(-9 :: Int)
+ Nanosecond -> 10^^(-9 :: Int)
+ Microsecond -> 10^^(-6 :: Int)
+ Millisecond -> 10^^(-3 :: Int)
+ Second -> 1
+ Minute -> 60
+ Hour -> 60 * 60
+
+instance Timestamp UTCTime where
+ roundTo prec = roundTo prec . utcTimeToPOSIXSeconds
+ scaleTo prec = scaleTo prec . utcTimeToPOSIXSeconds
+
+instance Timestamp NominalDiffTime where
+ roundTo prec time =
+ round $ 10^(9 :: Int) * roundAt (precisionScale prec) time
+ scaleTo prec time = round $ time / precisionScale prec
+
+-- | Exceptions used in this library.
+--
+-- In general, the library tries to convert exceptions from the dependent
+-- libraries to the following types of errors.
data InfluxException
- = JsonDecodeError String
- | SeriesDecodeError String
+ = ServerError String
+ -- ^ Server side error.
+ --
+ -- You can expect to get a successful response once the issue is resolved on
+ -- the server side.
+ | BadRequest String Request
+ -- ^ Client side error.
+ --
+ -- You need to fix your query to get a successful response.
+ | IllformedJSON String BL.ByteString
+ -- ^ Unexpected JSON response.
+ --
+ -- This can happen e.g. when the response from InfluxDB is incompatible with
+ -- what this library expects due to an upstream format change etc.
deriving (Show, Typeable)
instance Exception InfluxException
-jsonDecodeError :: String -> IO a
-jsonDecodeError = throwIO . JsonDecodeError
+class HasServer a where
+ server :: Lens' a Server
+
+class HasDatabase a where
+ database :: Lens' a Database
-seriesDecodeError :: String -> IO a
-seriesDecodeError = throwIO . SeriesDecodeError
+class HasPrecision (ty :: RequestType) a | a -> ty where
+ -- Time precision parameter
+ precision :: Lens' a (Precision ty)
------------------------------------------------------------
--- Aeson instances
+class HasManager a where
+ -- | HTTP manager settings or a manager itself.
+ --
+ -- If it's set to 'ManagerSettings', the library will create a 'Manager' from
+ -- the settings for you.
+ manager :: Lens' a (Either ManagerSettings Manager)
-deriveFromJSON (stripPrefixOptions "database") ''Database
-deriveFromJSON (stripPrefixOptions "admin") ''Admin
-deriveFromJSON (stripPrefixOptions "user") ''User
-deriveFromJSON (stripPrefixOptions "ping") ''Ping
-deriveFromJSON (stripPrefixOptions "shardSpace") ''ShardSpace
+class HasCredentials a where
+ authentication :: Lens' a (Maybe Credentials)
diff --git a/src/Database/InfluxDB/Types/Internal.hs b/src/Database/InfluxDB/Types/Internal.hs
deleted file mode 100644
index 0707d93..0000000
--- a/src/Database/InfluxDB/Types/Internal.hs
+++ /dev/null
@@ -1,49 +0,0 @@
-{-# LANGUAGE CPP #-}
-module Database.InfluxDB.Types.Internal
- ( stripPrefixOptions
- , stripPrefixLower
- , stripPrefixSnake
- ) where
-import Data.Char (isUpper, toLower)
-
--------------------------------------------------
--- Conditional imports
-
-#if MIN_VERSION_aeson(0, 6, 2)
-import Data.Aeson.TH (Options(..), defaultOptions)
-#endif
-
--------------------------------------------------
-
-#if MIN_VERSION_aeson(0, 6, 2)
-stripPrefixOptions :: String -> Options
-stripPrefixOptions name = defaultOptions
- { fieldLabelModifier = stripPrefixLower name
- }
-#else
-stripPrefixOptions :: String -> String -> String
-stripPrefixOptions = stripPrefixLower
-#endif
-
--- | Strip the prefix then convert to 'lowerCamelCase'.
-stripPrefixLower
- :: String -- ^ Prefix to be stripped
- -> String -- ^ Input string
- -> String
-stripPrefixLower prefix xs = case drop (length prefix) xs of
- [] -> error "Insufficient length of field name"
- c:cs -> toLower c : cs
-
--- | Strip the prefix then convert to 'snake_case'.
-stripPrefixSnake
- :: String -- ^ Prefix to be stripped
- -> String -- ^ Input string
- -> String
-stripPrefixSnake prefix xs = case drop (length prefix) xs of
- [] -> error "Insufficient length of field name"
- cs -> toSnake cs
- where
- toSnake = dropWhile (== '_') . foldr f []
- f c cs
- | isUpper c = '_':toLower c:cs
- | otherwise = c:cs
diff --git a/src/Database/InfluxDB/Write.hs b/src/Database/InfluxDB/Write.hs
new file mode 100644
index 0000000..650d457
--- /dev/null
+++ b/src/Database/InfluxDB/Write.hs
@@ -0,0 +1,209 @@
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE MultiParamTypeClasses #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE RecordWildCards #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TemplateHaskell #-}
+{-# LANGUAGE ViewPatterns #-}
+module Database.InfluxDB.Write
+ ( -- * Writers
+ write
+ , writeBatch
+ , writeByteString
+
+ -- * Writer parameters
+ , WriteParams
+ , writeParams
+ , Types.server
+ , Types.database
+ , retentionPolicy
+ , Types.precision
+ , Types.manager
+) where
+import Control.Exception
+import Control.Monad
+import Data.Maybe
+
+import Control.Lens
+import qualified Data.Aeson as A
+import qualified Data.Aeson.Types as A
+import qualified Data.ByteString.Char8 as B8
+import qualified Data.ByteString.Lazy as BL
+import qualified Data.Text.Encoding as TE
+import qualified Network.HTTP.Types as HT
+
+import Database.InfluxDB.Line
+import Database.InfluxDB.Types as Types
+import Database.InfluxDB.JSON
+import qualified Network.HTTP.Client.Compat as HC
+
+-- | The full set of parameters for the HTTP writer.
+data WriteParams = WriteParams
+ { _server :: !Server
+ , _database :: !Database
+ -- ^ Database to be written
+ , _retentionPolicy :: !(Maybe Key)
+ -- ^ 'Nothing' means the default retention policy for the database.
+ , _precision :: !(Precision 'WriteRequest)
+ -- ^ Timestamp precision
+ --
+ -- In the HTTP API, timestamps are scaled by the given precision.
+ , _authentication :: !(Maybe Credentials)
+ -- ^ No authentication by default
+ , _manager :: !(Either HC.ManagerSettings HC.Manager)
+ -- ^ HTTP connection manager
+ }
+
+-- | Smart constructor for 'WriteParams'
+--
+-- Default parameters:
+--
+-- ['L.server'] 'localServer'
+-- ['L.precision'] 'Nanosecond'
+-- ['retentionPolicy'] 'Nothing'
+-- ['L.manager'] @'Left' 'HC.defaultManagerSettings'@
+writeParams :: Database -> WriteParams
+writeParams _database = WriteParams
+ { _server = localServer
+ , _precision = Nanosecond
+ , _retentionPolicy = Nothing
+ , _authentication = Nothing
+ , _manager = Left HC.defaultManagerSettings
+ , ..
+ }
+
+-- | Write a 'Line'
+write
+ :: Timestamp time
+ => WriteParams
+ -> Line time
+ -> IO ()
+write p@WriteParams {_precision} =
+ writeByteString p . encodeLine (scaleTo _precision)
+
+-- | Write 'Line's in a batch
+--
+-- This is more efficient than 'write'.
+writeBatch
+ :: (Timestamp time, Traversable f)
+ => WriteParams
+ -> f (Line time)
+ -> IO ()
+writeBatch p@WriteParams {_precision} =
+ writeByteString p . encodeLines (scaleTo _precision)
+
+-- | Write a raw 'BL.ByteString'
+writeByteString :: WriteParams -> BL.ByteString -> IO ()
+writeByteString params payload = do
+ manager' <- either HC.newManager return $ _manager params
+ response <- HC.httpLbs request manager'
+ let body = HC.responseBody response
+ status = HC.responseStatus response
+ if BL.null body
+ then do
+ let message = B8.unpack $ HT.statusMessage status
+ when (HT.statusIsServerError status) $
+ throwIO $ ServerError message
+ when (HT.statusIsClientError status) $
+ throwIO $ BadRequest message request
+ else case A.eitherDecode' body of
+ Left message ->
+ throwIO $ IllformedJSON message body
+ Right val -> case A.parse parseErrorObject val of
+ A.Success _ ->
+ fail $ "BUG: impossible code path in Database.InfluxDB.Write.writeByteString"
+ A.Error message -> do
+ when (HT.statusIsServerError status) $
+ throwIO $ ServerError message
+ when (HT.statusIsClientError status) $
+ throwIO $ BadRequest message request
+ fail $ "BUG: " ++ message
+ ++ " in Database.InfluxDB.Write.writeByteString"
+
+ where
+ request = (writeRequest params)
+ { HC.requestBody = HC.RequestBodyLBS payload
+ }
+
+writeRequest :: WriteParams -> HC.Request
+writeRequest WriteParams {..} =
+ HC.setQueryString qs HC.defaultRequest
+ { HC.host = TE.encodeUtf8 _host
+ , HC.port = fromIntegral _port
+ , HC.secure = _ssl
+ , HC.method = "POST"
+ , HC.path = "/write"
+ }
+ where
+ Server {..} = _server
+ qs = concat
+ [ [("db", Just $ TE.encodeUtf8 $ databaseName _database)]
+ , fromMaybe [] $ do
+ Key name <- _retentionPolicy
+ return [("rp", Just (TE.encodeUtf8 name))]
+ , fromMaybe [] $ do
+ Credentials { _user = u, _password = p } <- _authentication
+ return
+ [ ("u", Just (TE.encodeUtf8 u))
+ , ("p", Just (TE.encodeUtf8 p))
+ ]
+ ]
+
+makeLensesWith (lensRules & generateSignatures .~ False) ''WriteParams
+
+server :: Lens' WriteParams Server
+
+-- |
+-- >>> let p = writeParams "foo"
+-- >>> p ^. server.host
+-- "localhost"
+instance HasServer WriteParams where
+ server = Database.InfluxDB.Write.server
+
+database :: Lens' WriteParams Database
+
+-- |
+-- >>> let p = writeParams "foo"
+-- >>> p ^. database
+-- "foo"
+instance HasDatabase WriteParams where
+ database = Database.InfluxDB.Write.database
+
+-- | Target retention policy for the write.
+--
+-- InfluxDB writes to the @default@ retention policy if this parameter is set
+-- to 'Nothing'.
+--
+-- >>> let p = writeParams "foo"
+-- >>> let p' = p & retentionPolicy .~ Just "two_hours"
+-- >>> p' ^. retentionPolicy
+-- Just "two_hours"
+retentionPolicy :: Lens' WriteParams (Maybe Key)
+
+precision :: Lens' WriteParams (Precision 'WriteRequest)
+
+-- |
+-- >>> let p = writeParams "foo"
+-- >>> p ^. precision
+-- Nanosecond
+instance HasPrecision 'WriteRequest WriteParams where
+ precision = Database.InfluxDB.Write.precision
+
+manager :: Lens' WriteParams (Either HC.ManagerSettings HC.Manager)
+
+-- |
+-- >>> let p = writeParams "foo"
+-- >>> p & manager .~ Left HC.defaultManagerSettings
+instance HasManager WriteParams where
+ manager = Database.InfluxDB.Write.manager
+
+-- | Authentication info for the write
+--
+-- >>> let p = writeParams "foo"
+-- >>> p ^. authentication
+-- Nothing
+authentication :: Lens' WriteParams (Maybe Credentials)
+
+instance HasCredentials WriteParams where
+ authentication = Database.InfluxDB.Write.authentication
diff --git a/src/Database/InfluxDB/Write/UDP.hs b/src/Database/InfluxDB/Write/UDP.hs
new file mode 100644
index 0000000..7803ccc
--- /dev/null
+++ b/src/Database/InfluxDB/Write/UDP.hs
@@ -0,0 +1,87 @@
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE MultiParamTypeClasses #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE RecordWildCards #-}
+{-# LANGUAGE TemplateHaskell #-}
+module Database.InfluxDB.Write.UDP
+ ( -- * Writers
+ write
+ , writeBatch
+ , writeByteString
+
+ -- * Writer parameters
+ , WriteParams
+ , writeParams
+ , socket
+ , sockAddr
+ , Types.precision
+ ) where
+
+import Control.Lens
+import Network.Socket (SockAddr, Socket)
+import Network.Socket.ByteString (sendManyTo)
+import qualified Data.ByteString.Lazy as BL
+
+import Database.InfluxDB.Line
+import Database.InfluxDB.Types as Types
+
+-- | The full set of parameters for the UDP writer.
+data WriteParams = WriteParams
+ { _socket :: !Socket
+ , _sockAddr :: !SockAddr
+ , _precision :: !(Precision 'WriteRequest)
+ }
+
+-- | Smart constructor for 'WriteParams'
+--
+-- Default parameters:
+--
+-- ['L.precision'] 'Nanosecond'
+writeParams :: Socket -> SockAddr -> WriteParams
+writeParams _socket _sockAddr = WriteParams
+ { _precision = Nanosecond
+ , ..
+ }
+
+-- | Write a 'Line'
+write
+ :: Timestamp time
+ => WriteParams
+ -> Line time
+ -> IO ()
+write p@WriteParams {_precision} =
+ writeByteString p . encodeLine (roundTo _precision)
+
+-- | Write 'Line's in a batch
+--
+-- This is more efficient than 'write'.
+writeBatch
+ :: (Timestamp time, Traversable f)
+ => WriteParams
+ -> f (Line time)
+ -> IO ()
+writeBatch p@WriteParams {_precision} =
+ writeByteString p . encodeLines (roundTo _precision)
+
+-- | Write a raw 'L.ByteString'
+writeByteString :: WriteParams -> BL.ByteString -> IO ()
+writeByteString WriteParams {..} payload =
+ sendManyTo _socket (BL.toChunks payload) _sockAddr
+
+makeLensesWith (lensRules & generateSignatures .~ False) ''WriteParams
+
+-- | Open UDP socket
+socket :: Lens' WriteParams Socket
+
+-- | UDP endopoint of the database
+sockAddr :: Lens' WriteParams SockAddr
+
+precision :: Lens' WriteParams (Precision 'WriteRequest)
+
+-- | Timestamp precision.
+--
+-- In the UDP API, all timestamps are sent in nanosecond but you can specify
+-- lower precision. The writer just rounds timestamps to the specified
+-- precision.
+instance HasPrecision 'WriteRequest WriteParams where
+ precision = Database.InfluxDB.Write.UDP.precision
diff --git a/src/Network/HTTP/Client/Compat.hs b/src/Network/HTTP/Client/Compat.hs
new file mode 100644
index 0000000..01a8e4e
--- /dev/null
+++ b/src/Network/HTTP/Client/Compat.hs
@@ -0,0 +1,16 @@
+{-# LANGUAGE CPP #-}
+module Network.HTTP.Client.Compat
+ ( defaultRequest
+ , module X
+ ) where
+
+#if MIN_VERSION_http_client(0, 5, 0)
+import Network.HTTP.Client (defaultRequest)
+import Network.HTTP.Client as X hiding (defaultRequest)
+#else
+import Data.Default.Class (def)
+import Network.HTTP.Client as X
+
+defaultRequest :: Request
+defaultRequest = def
+#endif
diff --git a/tests/test-suite.hs b/tests/test-suite.hs
index b730e1e..7f6e3c6 100644
--- a/tests/test-suite.hs
+++ b/tests/test-suite.hs
@@ -16,469 +16,8 @@ import qualified Data.Vector as V
import Test.Tasty.HUnit
import Test.Tasty.TH
-import Test.Tasty.QuickCheck hiding (reason)
-import qualified Network.HTTP.Client as HC
import Database.InfluxDB
-import Database.InfluxDB.TH
-import qualified Database.InfluxDB.Stream as S
-
-prop_fromValue_toValue_identity_Value :: Value -> Bool
-prop_fromValue_toValue_identity_Value = fromValueToValueIdentity
-
-prop_fromValue_toValue_identity_Bool :: Bool -> Bool
-prop_fromValue_toValue_identity_Bool = fromValueToValueIdentity
-
-prop_fromValue_toValue_identity_Int :: Int -> Bool
-prop_fromValue_toValue_identity_Int = fromValueToValueIdentity
-
-prop_fromValue_toValue_identity_Int8 :: Int8 -> Bool
-prop_fromValue_toValue_identity_Int8 = fromValueToValueIdentity
-
-prop_fromValue_toValue_identity_Int16 :: Int16 -> Bool
-prop_fromValue_toValue_identity_Int16 = fromValueToValueIdentity
-
-prop_fromValue_toValue_identity_Int32 :: Int32 -> Bool
-prop_fromValue_toValue_identity_Int32 = fromValueToValueIdentity
-
-prop_fromValue_toValue_identity_Int64 :: Int64 -> Bool
-prop_fromValue_toValue_identity_Int64 = fromValueToValueIdentity
-
-prop_fromValue_toValue_identity_Word8 :: Word8 -> Bool
-prop_fromValue_toValue_identity_Word8 = fromValueToValueIdentity
-
-prop_fromValue_toValue_identity_Word16 :: Word16 -> Bool
-prop_fromValue_toValue_identity_Word16 = fromValueToValueIdentity
-
-prop_fromValue_toValue_identity_Word32 :: Word32 -> Bool
-prop_fromValue_toValue_identity_Word32 = fromValueToValueIdentity
-
-prop_fromValue_toValue_identity_Double :: Double -> Bool
-prop_fromValue_toValue_identity_Double = fromValueToValueIdentity
-
-prop_fromValue_toValue_identity_Text :: T.Text -> Bool
-prop_fromValue_toValue_identity_Text = fromValueToValueIdentity
-
-prop_fromValue_toValue_identity_LazyText :: TL.Text -> Bool
-prop_fromValue_toValue_identity_LazyText = fromValueToValueIdentity
-
-prop_fromValue_toValue_identity_String :: String -> Bool
-prop_fromValue_toValue_identity_String = fromValueToValueIdentity
-
-prop_fromValue_toValue_identity_Maybe_Int :: Maybe Int -> Bool
-prop_fromValue_toValue_identity_Maybe_Int = fromValueToValueIdentity
-
--------------------------------------------------
-
-instance Arbitrary Value where
- arbitrary = oneof
- [ Int <$> arbitrary
- , Float <$> arbitrary
- , String <$> arbitrary
- , Bool <$> arbitrary
- , pure Null
- ]
-
-instance Arbitrary T.Text where
- arbitrary = T.pack <$> arbitrary
-
-instance Arbitrary TL.Text where
- arbitrary = TL.pack <$> arbitrary
-
-fromValueToValueIdentity :: (Eq a, FromValue a, ToValue a) => a -> Bool
-fromValueToValueIdentity a = fromValue (toValue a) == Right a
-
--------------------------------------------------
-
-case_ping :: Assertion
-case_ping = runTest $ \config -> do
- Ping status <- ping config
- status @?= "ok"
-
-case_isInSync :: Assertion
-case_isInSync = runTest $ \config -> do
- inSync <- isInSync config
- assertBool "The database is not in sync." inSync
-
-case_post :: Assertion
-case_post = runTest $ \config ->
- withTestDatabase config $ \database -> do
- name <- liftIO newName
- post config database $
- writeSeries name $ Val 42
- ss <- query config database $ "select value from " <> name
- case ss of
- [series] -> fromSeriesData series @?= Right [Val 42]
- _ -> assertFailure $ "Expect one series, but got: " ++ show ss
-
-case_post_multi_series :: Assertion
-case_post_multi_series = runTest $ \config ->
- withTestDatabase config $ \database -> do
- name <- liftIO newName
- post config database $ do
- writeSeries name $ Val 42
- writeSeries name $ Val 42
- writeSeries name $ Val 42
- ss <- query config database $ "select value from " <> name
- case ss of
- [series] -> fromSeriesData series @?= Right [Val 42, Val 42, Val 42]
- _ -> assertFailure $ "Expect one series, but got: " ++ show ss
-
-case_post_multi_points :: Assertion
-case_post_multi_points = runTest $ \config ->
- withTestDatabase config $ \database -> do
- name <- liftIO newName
- post config database $ withSeries name $ do
- writePoints $ Val 42
- writePoints $ Val 42
- writePoints $ Val 42
- ss <- query config database $ "select value from " <> name
- case ss of
- [series] -> fromSeriesData series @?= Right [Val 42, Val 42, Val 42]
- _ -> assertFailure $ "Expect one series, but got: " ++ show ss
-
-case_query_nonexistent_series :: Assertion
-case_query_nonexistent_series = runTest $ \config ->
- withTestDatabase config $ \database -> do
- name <- liftIO newName
- assertStatusCodeException
- (query config database $ "select * from " <> name :: IO [SeriesData])
-
-case_query_empty_series :: Assertion
-case_query_empty_series = runTest $ \config ->
- withTestDatabase config $ \database -> do
- name <- liftIO newName
- post config database $
- writeSeries name $ Val 42
- ss1 <- query config database $ "delete from " <> name
- ss1 @?= ([] :: [SeriesData])
- ss2 <- query config database $ "select * from " <> name
- ss2 @?= ([] :: [SeriesData])
-
-case_queryChunked :: Assertion
-case_queryChunked = runTest $ \config ->
- withTestDatabase config $ \database -> do
- name <- liftIO newName
- post config database $ withSeries name $ do
- writePoints $ Val 42
- writePoints $ Val 42
- writePoints $ Val 42
- ss <- queryChunked config database ("select value from " <> name) $
- S.fold step []
- mapM fromSeriesData ss @?= Right [[Val 42], [Val 42], [Val 42]]
- where
- step xs series = case fromSeriesData series of
- Left reason -> throwIO $ HUnitFailure reason
- Right values -> return $ xs ++ values
-
-case_post_with_precision :: Assertion
-case_post_with_precision = runTest $ \config ->
- withTestDatabase config $ \database -> do
- name <- liftIO newName
- postWithPrecision config database SecondsPrecision $
- writeSeries name $ Val 42
- ss <- query config database $ "select value from " <> name
- case ss of
- [series] -> fromSeriesData series @?= Right [Val 42]
- _ -> assertFailure $ "Expect one series, but got: " ++ show ss
-
-case_delete_series :: Assertion
-case_delete_series = runTest $ \config ->
- withTestDatabase config $ \database -> do
- name <- liftIO newName
- post config database $
- writeSeries name $ Val 42
- ss <- query config database $ "select value from " <> name
- case ss of
- [series] -> fromSeriesData series @?= Right [Val 42]
- _ -> assertFailure $ "Expect one series, but got: " ++ show ss
- deleteSeries config database name
- assertStatusCodeException
- (query config database $ "select value from " <> name :: IO [SeriesData])
-
-case_listDatabases :: Assertion
-case_listDatabases = runTest $ \config ->
- withTestDatabase config $ \name -> do
- databases <- listDatabases config
- assertBool ("No such database: " ++ T.unpack name) $
- any ((name ==) . databaseName) databases
-
-case_configureDatabase :: Assertion
-case_configureDatabase = runTest $ \config -> do
- dbName <- newName
- do
- configureDatabase config dbName $ DatabaseRequest shardSpaces contQueries
- listDatabases config >>= \databases ->
- assertBool ("No such database: " ++ T.unpack dbName) $
- any ((dbName ==) . databaseName) databases
- listShardSpaces config >>= \spaces ->
- assertBool "Missing shard space(s)" $
- any ((`elem` spaceNames) . shardSpaceName) spaces
- `finally`
- dropDatabase config dbName
- where
- spaceNames = map shardSpaceRequestName shardSpaces
- shardSpaces =
- [ ShardSpaceRequest
- { shardSpaceRequestName = "everything_30d"
- , shardSpaceRequestRetentionPolicy = "30d"
- , shardSpaceRequestShardDuration = "7d"
- , shardSpaceRequestRegex = "/.*/"
- , shardSpaceRequestReplicationFactor = 1
- , shardSpaceRequestSplit = 1
- }
- , ShardSpaceRequest
- { shardSpaceRequestName = "forever"
- , shardSpaceRequestRetentionPolicy = "inf"
- , shardSpaceRequestShardDuration = "7d"
- , shardSpaceRequestRegex = "/^_.*/"
- , shardSpaceRequestReplicationFactor = 1
- , shardSpaceRequestSplit = 1
- }
- , ShardSpaceRequest
- { shardSpaceRequestName = "rollups"
- , shardSpaceRequestRetentionPolicy = "365d"
- , shardSpaceRequestShardDuration = "30d"
- , shardSpaceRequestRegex = "/^\\d+.*/"
- , shardSpaceRequestReplicationFactor = 1
- , shardSpaceRequestSplit = 1
- }
- ]
- contQueries =
- [ "select * from events into events.[id]"
- , "select count(value) from events group by time(5m) into 5m.count.events"
- ]
-
-case_shardSpaces :: Assertion
-case_shardSpaces = runTest $ \config ->
- withTestDatabase config $ \name -> do
- spaceName <- newName
- createShardSpace config name $ ShardSpaceRequest
- { shardSpaceRequestName = spaceName
- , shardSpaceRequestRegex = "^[a-z].*"
- , shardSpaceRequestRetentionPolicy = "7d"
- , shardSpaceRequestShardDuration = "1d"
- , shardSpaceRequestReplicationFactor = 1
- , shardSpaceRequestSplit = 1
- }
- listShardSpaces config >>= \spaces ->
- assertBool ("No such shard space: " ++ T.unpack spaceName) $
- any ((spaceName ==) . shardSpaceName) spaces
- dropShardSpace config name spaceName
- listShardSpaces config >>= \spaces ->
- assertBool ("Found a dropped shard space: " ++ T.unpack spaceName) $
- all ((spaceName /=) . shardSpaceName) spaces
-
-case_create_then_drop_database :: Assertion
-case_create_then_drop_database = runTest $ \config -> do
- name <- newName
- dropDatabaseIfExists config name
- createDatabase config name
- listDatabases config >>= \databases ->
- assertBool ("No such database: " ++ T.unpack name) $
- any ((name ==) . databaseName) databases
- dropDatabase config name
- listDatabases config >>= \databases ->
- assertBool ("Found a dropped database: " ++ T.unpack name) $
- all ((name /=) . databaseName) databases
-
-case_list_cluster_admins :: Assertion
-case_list_cluster_admins = runTest $ \config -> do
- admins <- listClusterAdmins config
- assertBool "No root admin" $
- any (("root" ==) . adminName) admins
-
-case_authenticate_cluster_admin :: Assertion
-case_authenticate_cluster_admin = runTest authenticateClusterAdmin
-
-case_add_then_delete_cluster_admin :: Assertion
-case_add_then_delete_cluster_admin = runTest $ \config -> do
- name <- newName
- admin <- addClusterAdmin config name "somePassword"
- listClusterAdmins config >>= \admins ->
- assertBool ("No such admin: " ++ T.unpack name) $
- any ((name ==) . adminName) admins
- deleteClusterAdmin config admin
- listClusterAdmins config >>= \admins ->
- assertBool ("Found a deleted admin: " ++ T.unpack name) $
- all ((name /=) . adminName) admins
-
-case_update_cluster_admin_password :: Assertion
-case_update_cluster_admin_password = runTest $ \config -> do
- let curPassword = "somePassword"
- newPassword = "otherPassword"
- name <- newName
- deleteClusterAdminIfExists config name
- admin <- addClusterAdmin config name curPassword
- updateClusterAdminPassword config admin newPassword
- let newCreds = Credentials name newPassword
- newConfig = config { configCreds = newCreds }
- name' <- newName
- dropDatabaseIfExists config name'
- createDatabase newConfig name'
- listDatabases newConfig >>= \databases ->
- assertBool ("No such database: " ++ T.unpack name') $
- any ((name' ==) . databaseName) databases
- dropDatabase newConfig name'
- listDatabases newConfig >>= \databases ->
- assertBool ("Found a dropped database: " ++ T.unpack name') $
- all ((name' /=) . databaseName) databases
- deleteClusterAdmin config admin
-
-case_add_then_delete_database_users :: Assertion
-case_add_then_delete_database_users = runTest $ \config ->
- withTestDatabase config $ \name -> do
- listDatabaseUsers config name >>= \users ->
- assertBool "There shouldn't be any users" $ null users
- newUserName <- newName
- addDatabaseUser config name newUserName "somePassword"
- let newCreds = rootCreds
- { credsUser = newUserName
- , credsPassword = "somePassword" }
- newConfig = config { configCreds = newCreds }
- authenticateDatabaseUser newConfig name
- listDatabaseUsers config name >>= \users ->
- assertBool ("No such user: " <> T.unpack newUserName) $
- any ((newUserName ==) . userName) users
- deleteDatabaseUser config name newUserName
- listDatabaseUsers config name >>= \users ->
- assertBool ("Found a deleted user: " <> T.unpack newUserName) $
- all ((newUserName /=) . userName) users
-
-case_update_database_user_password :: Assertion
-case_update_database_user_password = runTest $ \config ->
- withTestDatabase config $ \name -> do
- newUserName <- newName
- addDatabaseUser config name newUserName "somePassword"
- listDatabaseUsers config name >>= \users ->
- assertBool ("No such user: " <> T.unpack newUserName) $
- any ((newUserName ==) . userName) users
- updateDatabaseUserPassword config name newUserName "otherPassword"
- deleteDatabaseUser config name newUserName
-
-case_grant_revoke_database_user :: Assertion
-case_grant_revoke_database_user = runTest $ \config ->
- withTestDatabase config $ \name -> do
- newUserName <- newName
- addDatabaseUser config name newUserName "somePassword"
- listDatabaseUsers config name >>= \users ->
- assertBool ("No such user: " <> T.unpack newUserName) $
- any ((newUserName ==) . userName) users
- grantAdminPrivilegeTo config name newUserName
- listDatabaseUsers config name >>= \users ->
- case find ((newUserName ==) . userName) users of
- Nothing -> assertFailure $ "No such user: " <> T.unpack newUserName
- Just user -> assertBool
- ("User is not privileged: " <> T.unpack newUserName)
- (userIsAdmin user)
- revokeAdminPrivilegeFrom config name newUserName
- listDatabaseUsers config name >>= \users ->
- case find ((newUserName ==) . userName) users of
- Nothing -> assertFailure $ "No such user: " <> T.unpack newUserName
- Just user -> assertBool
- ("User is still privileged: " <> T.unpack newUserName)
- (not $ userIsAdmin user)
- deleteDatabaseUser config name newUserName
-
--------------------------------------------------
--- Regressions
-
-newtype WholeFloat = WholeFloat
- { wholeFloatValue :: Double
- } deriving (Eq, Show)
-
--- #14: InfluxDB may return Int instead of Float when
--- the WholeFloat value happens to be a whole number.
-case_regression_whole_Float_number :: Assertion
-case_regression_whole_Float_number = runTest $ \config ->
- withTestDatabase config $ \database -> do
- series <- newName
- post config database $
- writeSeries series $ WholeFloat 42.0
- ss <- query config database $ "select value from " <> series
- case ss of
- [sd] -> fromSeriesData sd @?= Right [WholeFloat 42]
- _ -> assertFailure $ "Expect one series, but got: " ++ show ss
-
-case_regression_really_big_Float_number :: Assertion
-case_regression_really_big_Float_number = runTest $ \config ->
- withTestDatabase config $ \database -> do
- series <- newName
- post config database $
- writeSeries series $ WholeFloat 42e100
- ss <- query config database $ "select value from " <> series
- case ss of
- [sd] -> fromSeriesData sd @?= Right [WholeFloat 42e100]
- _ -> assertFailure $ "Expect one series, but got: " ++ show ss
-
--------------------------------------------------
-
-data Val = Val Int deriving (Eq, Show)
-
-instance ToSeriesData Val where
- toSeriesColumns _ = V.fromList ["value"]
- toSeriesPoints (Val n) = V.fromList [toValue n]
-
-instance FromSeriesData Val where
- parseSeriesData = withValues $ \values -> Val <$> values .: "value"
-
--------------------------------------------------
-
-dropDatabaseIfExists :: Config -> Text -> IO ()
-dropDatabaseIfExists config name =
- dropDatabase config name
- `catchAll` \_ -> return ()
-
-deleteClusterAdminIfExists :: Config -> Text -> IO ()
-deleteClusterAdminIfExists config name =
- deleteClusterAdmin config (Admin name)
- `catchAll` \_ -> return ()
-
--------------------------------------------------
-
-runTest :: (Config -> IO a) -> IO a
-runTest f = do
- pool <- newServerPool localServer []
- HC.withManager settings (f . Config rootCreds pool)
- where
- settings = HC.defaultManagerSettings
-
-newName :: IO Text
-newName = do
- uniq <- newUnique
- return $ T.pack $ "test_" ++ show (hashUnique uniq)
-
-withTestDatabase :: Config -> (Text -> IO a) -> IO a
-withTestDatabase config = bracket acquire release
- where
- acquire = do
- name <- newName
- dropDatabaseIfExists config name
- createDatabase config name
- return name
- release = dropDatabase config
-
-catchAll :: IO a -> (SomeException -> IO a) -> IO a
-catchAll = E.catch
-
-assertStatusCodeException :: Show a => IO a -> IO ()
-assertStatusCodeException io = do
- r <- try io
- case r of
- Left e -> case fromException e of
- Just HC.StatusCodeException {} -> return ()
- _ ->
- assertFailure $ "Expect a StatusCodeException, but got " ++ show e
- Right ss -> assertFailure $ "Expect an exception, but got " ++ show ss
-
--------------------------------------------------
main :: IO ()
main = $defaultMainGenerator
-
--------------------------------------------------
--- Instance deriving
-
-deriveSeriesData defaultOptions
- { fieldLabelModifier = stripPrefixLower "wholeFloat" }
- ''WholeFloat