diff --git a/CHANGELOG.md b/CHANGELOG.md --- 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 --- 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 --- 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 = do [read -> (numPoints :: Int), read -> (batches :: Int)] <- getArgs hSetBuffering stdout NoBuffering - HC.withManager managerSettings $ \manager -> do - config <- newConfig manager + manager' <- HC.newManager managerSettings - 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 + let + ctx = "ctx" + ct1 = "ct1" + qparams = queryParams ctx + & manager .~ Right manager' + & precision .~ RFC3339 - 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 "--" + M.manage qparams $ F.formatQuery ("DROP DATABASE "%F.database) ctx + M.manage qparams $ F.formatQuery ("CREATE DATABASE "%F.database) ctx -newConfig :: HC.Manager -> IO Config -newConfig manager = do - pool <- newServerPool localServer [] - return Config - { configCreds = rootCreds - , configServerPool = pool - , configHttpManager = manager - } + 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 @@ | 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 --- /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 --- 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-file: LICENSE author: Mitsutoshi Aoe maintainer: Mitsutoshi Aoe -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 @@ 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,24 +107,42 @@ 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 type: git branch: develop @@ -142,5 +150,5 @@ 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 --- 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(..) + ( -- $intro - -- ** Encoding - , ToSeriesData(..) - , ToValue(..) + -- * Writing data + -- $write + write + , writeBatch + , writeByteString - -- ** Decoding - , FromSeries(..), fromSeries - , FromSeriesData(..), fromSeriesData - , FromValue(..), fromValue + -- ** Write parameters + , WriteParams + , writeParams + , retentionPolicy - , withValues, (.:), (.:?), (.!=) - , typeMismatch + -- ** The Line protocol + , Line(Line) + , measurement + , tagSet + , fieldSet + , timestamp - -- * HTTP API - -- ** Data types - , Config(..) - , Credentials(..), rootCreds - , TimePrecision(..) - , Server(..), localServer - , ServerPool, newServerPool - , newServerPoolWithRetryPolicy - , Database(..) - , User(..) - , Admin(..) - , Ping(..) - , ShardSpace(..) + , FieldValue(..) + , Timestamp(..) + , precisionScale + , precisionName - -- *** Exception - , InfluxException(..) + -- * Querying data + , Query + , query + , queryChunked - -- ** Writing Data + -- * Query constructor + , formatQuery + , (%) - -- *** Updating Points - , post, postWithPrecision - , SeriesT, PointT - , writeSeries - , writeSeriesData - , withSeries - , writePoints + -- ** Query parameters + , QueryParams + , queryParams + , authentication - -- *** Deleting Points - , deleteSeries + -- ** Parsing results + , QueryResults(..) + , parseResultsWith + , getField + , getTag + , parseTimestamp + , parseFieldValue + , parseKey - -- ** Querying Data - , query - , Stream(..) - , queryChunked + -- * Database management + , manage - -- ** Administration & Security - -- *** Creating and Dropping Databases - , listDatabases - , createDatabase - , dropDatabase + -- * Common data types and classes + , Precision(..) + , Database + , Key - , DatabaseRequest(..) - , configureDatabase + , Server + , host + , port + , ssl + , localServer - -- *** Security - -- **** Shard spaces - , ShardSpaceRequest(..) - , listShardSpaces - , createShardSpace - , dropShardSpace + , Credentials + , user + , password - -- **** Cluster admin - , listClusterAdmins - , authenticateClusterAdmin - , addClusterAdmin - , updateClusterAdminPassword - , deleteClusterAdmin - -- **** Database user - , listDatabaseUsers - , authenticateDatabaseUser - , addDatabaseUser - , updateDatabaseUserPassword - , deleteDatabaseUser - , grantAdminPrivilegeTo - , revokeAdminPrivilegeFrom + -- * Exception + , InfluxException(..) - -- *** Other API - , ping - , isInSync + , 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 --- 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 --- 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 --- /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 --- 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 --- . -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 --- /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 --- 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 --- /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 --- /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 --- /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 --- /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 --- 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 --- 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 --- 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) +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 -#if MIN_VERSION_aeson(0, 7, 0) -import Data.Scientific -#else -import Data.Attoparsec.Number -#endif +newtype Query = Query T.Text deriving IsString ------------------------------------------------------------ --- Compatibility for older GHC +instance Show Query where + show (Query q) = show q -#if __GLASGOW_HASKELL__ < 706 -import Control.Exception (evaluate) +data Server = Server + { _host :: !Text + , _port :: !Int + , _ssl :: !Bool + } deriving (Show, Generic, Eq) -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 ------------------------------------------------------------ +-- | Default server settings. +-- +-- Default parameters: +-- +-- * 'host': @"localhost"@ +-- * 'port': @8086@ +-- * 'ssl': 'False' +localServer :: Server +localServer = Server + { _host = "localhost" + , _port = 8086 + , _ssl = False + } --- | 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) +makeLensesWith (lensRules & generateSignatures .~ False) ''Server --- | Convenient accessor for columns. -seriesColumns :: Series -> Vector Column -seriesColumns = seriesDataColumns . seriesData +-- | Host name of the server +host :: Lens' Server Text --- | Convenient accessor for points. -seriesPoints :: Series -> [Vector Value] -seriesPoints = seriesDataPoints . seriesData +-- | Port number of the server +port :: Lens' Server Int -instance A.ToJSON Series where - toJSON Series {..} = A.object - [ "name" .= seriesName - , "columns" .= seriesDataColumns - , "points" .= seriesDataPoints - ] - where - SeriesData {..} = seriesData +-- | If SSL is enabled +ssl :: Lens' Server Bool -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 +-- | User credentials +data Credentials = Credentials + { _user :: !Text + , _password :: !Text + } --- | 'SeriesData' consists of columns and points. -data SeriesData = SeriesData - { seriesDataColumns :: Vector Column - , seriesDataPoints :: [Vector Value] - } deriving (Eq, Show, Typeable, Generic) +makeLensesWith (lensRules & generateSignatures .~ False) ''Credentials -type Column = Text +-- | User name to access InfluxDB +user :: Lens' Credentials 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) +-- | Password to access InfluxDB +password :: Lens' Credentials Text -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 +-- | Database name +newtype Database = Database { databaseName :: Text } deriving (Eq, Ord) -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) +-- | 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 --- | User credentials. -data Credentials = Credentials - { credsUser :: !Text - , credsPassword :: !Text - } deriving (Show, Typeable, Generic) +instance IsString Key where + fromString xs = Key $ fromNonEmptyString "Key" xs --- | 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) +fromNonEmptyString :: String -> String -> Text +fromNonEmptyString ty xs + | null xs = error $ ty ++ " should never be empty" + | otherwise = fromString xs --- | 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 - } +instance Show Database where + show (Database name) = show name -newtype Database = Database - { databaseName :: Text - } deriving (Show, Typeable, Generic) +instance Show Key where + show (Key name) = show name --- | User -data User = User - { userName :: Text - , userIsAdmin :: Bool - } deriving (Show, Typeable, Generic) +data FieldValue + = FieldInt !Int64 + | FieldFloat !Double + | FieldString !Text + | FieldBool !Bool + | FieldNull + deriving (Eq, Show, Data, Typeable, Generic) --- | Administrator -newtype Admin = Admin - { adminName :: Text - } deriving (Show, Typeable, Generic) +instance IsString FieldValue where + fromString = FieldString . T.pack -newtype Ping = Ping - { pingStatus :: Text - } deriving (Show, Typeable, Generic) +-- | Type of a request +data RequestType + = QueryRequest + -- ^ Request for @/query@ + | WriteRequest + -- ^ Request for @/write@ + deriving Show -type Interface = Text +-- | 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 -data ShardSpace = ShardSpace - { shardSpaceDatabase :: Maybe Text - , shardSpaceName :: Text - , shardSpaceRegex :: Text - , shardSpaceRetentionPolicy :: Text - , shardSpaceShardDuration :: Text - , shardSpaceReplicationFactor :: Word32 - , shardSpaceSplit :: Word32 - } deriving (Show, Typeable, Generic) +deriving instance Show (Precision a) ------------------------------------------------------------ --- Server pool manipulation +precisionName :: Precision ty -> Text +precisionName = \case + Nanosecond -> "n" + Microsecond -> "u" + Millisecond -> "ms" + Second -> "s" + Minute -> "m" + Hour -> "h" + RFC3339 -> "rfc3339" --- | 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 +-- | 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 -newServerPoolWithRetryPolicy - :: RetryPolicy -> Server -> [Server] -> IO (IORef ServerPool) -newServerPoolWithRetryPolicy retryPolicy active backups = - newIORef ServerPool - { serverActive = active - , serverBackup = Seq.fromList backups - , serverRetryPolicy = retryPolicy - } +roundAt :: RealFrac a => a -> a -> a +roundAt scale x = fromIntegral (round (x / scale) :: Int) * scale --- | Get a server from the pool. -activeServer :: IORef ServerPool -> IO Server -activeServer ref = do - ServerPool { serverActive } <- readIORef ref - return serverActive +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 --- | 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 - } +instance Timestamp UTCTime where + roundTo prec = roundTo prec . utcTimeToPOSIXSeconds + scaleTo prec = scaleTo prec . utcTimeToPOSIXSeconds ------------------------------------------------------------ --- Exceptions +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 -seriesDecodeError :: String -> IO a -seriesDecodeError = throwIO . SeriesDecodeError +class HasDatabase a where + database :: Lens' a Database ------------------------------------------------------------ --- Aeson instances +class HasPrecision (ty :: RequestType) a | a -> ty where + -- Time precision parameter + precision :: Lens' a (Precision ty) -deriveFromJSON (stripPrefixOptions "database") ''Database -deriveFromJSON (stripPrefixOptions "admin") ''Admin -deriveFromJSON (stripPrefixOptions "user") ''User -deriveFromJSON (stripPrefixOptions "ping") ''Ping -deriveFromJSON (stripPrefixOptions "shardSpace") ''ShardSpace +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) + +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 --- 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 --- /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 --- /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 --- /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 --- a/tests/test-suite.hs +++ b/tests/test-suite.hs @@ -16,469 +16,8 @@ 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