module Database.InfluxDB.Http
( Config(..)
, Credentials(..), rootCreds
, Server(..), localServer
, TimePrecision(..)
, post, postWithPrecision
, SeriesT, PointT
, writeSeries
, writeSeriesData
, withSeries
, writePoints
, deleteSeries
, query
, Stream(..)
, queryChunked
, listDatabases
, createDatabase
, dropDatabase
, DatabaseRequest(..)
, configureDatabase
, ShardSpaceRequest(..)
, listShardSpaces
, createShardSpace
, dropShardSpace
, listClusterAdmins
, authenticateClusterAdmin
, addClusterAdmin
, updateClusterAdminPassword
, deleteClusterAdmin
, listDatabaseUsers
, authenticateDatabaseUser
, addDatabaseUser
, updateDatabaseUserPassword
, deleteDatabaseUser
, grantAdminPrivilegeTo
, revokeAdminPrivilegeFrom
, 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
data Config = Config
{ configCreds :: !Credentials
, configServerPool :: !(IORef ServerPool)
, configHttpManager :: !HC.Manager
}
rootCreds :: Credentials
rootCreds = Credentials
{ credsUser = "root"
, credsPassword = "root"
}
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"
post
:: Config
-> Text
-> SeriesT IO a
-> IO a
post config databaseName =
postGeneric config databaseName Nothing
postWithPrecision
:: Config
-> Text
-> TimePrecision
-> SeriesT IO a
-> IO a
postWithPrecision config databaseName timePrec =
postGeneric config databaseName (Just timePrec)
postGeneric
:: Config
-> Text
-> 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
newtype SeriesT m a = SeriesT (WriterT (DList Series) m a)
deriving
( Functor, Applicative, Monad, MonadIO, MonadTrans
, MonadWriter (DList Series)
)
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)
writeSeries
:: (Monad m, ToSeriesData a)
=> Text
-> a
-> SeriesT m ()
writeSeries name = writeSeriesData name . toSeriesData
writeSeriesData
:: Monad m
=> Text
-> SeriesData
-> SeriesT m ()
writeSeriesData name a = tell . DL.singleton $ Series
{ seriesName = name
, seriesData = a
}
withSeries
:: forall m a. (Monad m, ToSeriesData a)
=> Text
-> 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
}
}
writePoints
:: (Monad m, ToSeriesData a)
=> a
-> PointT a m ()
writePoints = tell . DL.singleton . toSeriesPoints
deleteSeries
:: Config
-> Text
-> Text
-> 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
query
:: FromSeries a
=> Config
-> Text
-> 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
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
queryChunked
:: FromSeries a
=> Config
-> Text
-> Text
-> (Stream IO a -> IO b)
-> 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
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
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
dropDatabase
:: Config
-> Text
-> 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
-> 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
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
createShardSpace
:: Config
-> Text
-> 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
-> Text
-> 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
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
addClusterAdmin
:: Config
-> Text
-> Text
-> 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
updateClusterAdminPassword
:: Config
-> Admin
-> Text
-> 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
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
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
-> 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
addDatabaseUser
:: Config
-> Text
-> Text
-> Text
-> 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
deleteDatabaseUser
:: Config
-> Text
-> Text
-> IO ()
deleteDatabaseUser config databaseName userName = runRequest_ config request
where
request = (makeRequestFromDatabaseUser config databaseName userName)
{ HC.method = "DELETE"
}
updateDatabaseUserPassword
:: Config
-> Text
-> Text
-> Text
-> 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
]
}
grantAdminPrivilegeTo
:: Config
-> Text
-> Text
-> 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
]
}
revokeAdminPrivilegeFrom
:: Config
-> Text
-> Text
-> 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
-> Text
-> 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
deriveToJSON (stripPrefixOptions "shardSpaceRequest") ''ShardSpaceRequest
deriveToJSON (stripPrefixOptions "databaseRequest") ''DatabaseRequest