diff --git a/Changelog.md b/Changelog.md --- a/Changelog.md +++ b/Changelog.md @@ -1,5 +1,10 @@ # Revision history for ekg-influxdb -## 0.1.0.0 -- YYYY-mm-dd +## 0.2.0.0 -- 2019-10-27 -* First version. Released on an unsuspecting world. +* Use InfluxDB WriteParams in configuration. +* Stop reporting mean (https://github.com/OndrejSlamecka/ekg-influxdb/pull/2). + +## 0.1.1.0 -- 2017-08-01 + +* Replace libinfluxdb with influxdb. diff --git a/ekg-influxdb.cabal b/ekg-influxdb.cabal --- a/ekg-influxdb.cabal +++ b/ekg-influxdb.cabal @@ -1,29 +1,30 @@ --- Initial ekg-influxdb.cabal generated by cabal init. For further +-- Initial ekg-influxdb.cabal generated by cabal init. For further -- documentation, see http://haskell.org/cabal/users-guide/ name: ekg-influxdb -version: 0.1.0.0 +version: 0.2.0.0 synopsis: An EKG backend to send statistics to influxdb -homepage: https://github.com/angerman/ekg-influxdb +homepage: https://github.com/OndrejSlamecka/ekg-influxdb license: BSD3 license-file: LICENSE author: Moritz Angermann -maintainer: moritz.angermann@gmail.com +maintainer: Ondrej Slamecka build-type: Simple extra-source-files: Changelog.md cabal-version: >=1.10 library exposed-modules: System.Remote.Monitoring.Influxdb - build-depends: base >=4.9 && <4.10 + build-depends: base >=4.9 && <5 , ekg-core >= 0.1 - , libinfluxdb >= 0.0.4 + , influxdb >= 1.2 , containers >= 0.5.7 , unordered-containers >= 0.2.7 , time >= 1.6 , clock >= 0.7 , text >= 1.2.2 , vector >= 0.11 + , lens hs-source-dirs: src default-language: Haskell2010 ghc-options: -Wall -O2 diff --git a/src/System/Remote/Monitoring/Influxdb.hs b/src/System/Remote/Monitoring/Influxdb.hs --- a/src/System/Remote/Monitoring/Influxdb.hs +++ b/src/System/Remote/Monitoring/Influxdb.hs @@ -3,9 +3,10 @@ -- | This module lets you periodically flush metrics to a influxdb -- backend. Example usage: -- +-- > import qualified Database.Influxdb as Influxdb -- > main = do -- > store <- newStore --- > forkInfluxdb defaultInfluxdbOptions store +-- > forkInfluxdb (defaultInfluxdbOptions (Influxdb.writeParams "database")) store -- -- You probably want to include some of the predefined metrics defined -- in the @ekg-core@ package, by calling e.g. the 'EKG.registerGcMetrics' @@ -23,31 +24,26 @@ ) where import Data.Monoid ((<>)) -import qualified Data.Time.Clock.POSIX as Time +import Data.Time.Clock as Time (getCurrentTime, diffUTCTime, UTCTime) import qualified Data.Text as T import qualified System.Metrics as EKG import qualified System.Metrics.Distribution as Stats -import System.Clock import qualified Data.Map as M import qualified Data.Vector as V -import qualified Database.InfluxDB.Writer as Influxdb +import qualified Database.InfluxDB as Influxdb import Control.Exception (SomeException, try, bracket) import Control.Concurrent (ThreadId, forkIO, myThreadId, threadDelay, throwTo) import Control.Monad (forever) import qualified Data.HashMap.Strict as HashMap -import Data.Int (Int64) +import Data.String (fromString) -------------------------------------------------------------------------------- -- | Options to control how to connect to the Influxdb server and how often to -- flush metrics. The flush interval should match the shortest retention rate of -- the matching retention periods, or you risk over-riding previous samples. data InfluxdbOptions = InfluxdbOptions - { -- | The hostname or IP address of the server running influxdb - host :: !T.Text - -- | Server port of the TCP line receiver interface. - , port :: !Int - -- | The Influxdb database name - , database :: !T.Text + { -- | The write parameters for Influxdb. + writeParams :: !Influxdb.WriteParams -- | The amount of time between sampling EKG metrics and pushing to Influxdb. , flushInterval :: !Int -- | Prefix to add to all matric names. @@ -57,23 +53,17 @@ -- @takeWhile (/= \'.\') \<$\> getHostName@, using @getHostName@ -- from the @Network.BSD@ module in the network package. , suffix :: !T.Text - } deriving (Eq, Show) + } -------------------------------------------------------------------------------- -- | Defaults -- --- * @host@ = @\"127.0.0.1\"@ --- --- * @port@ = @8086@ --- -- * @flushInterval@ = @1000@ -- --- * Empty 'database', 'prefix' and 'suffix'. -defaultInfluxdbOptions :: InfluxdbOptions -defaultInfluxdbOptions = InfluxdbOptions - { host = "127.0.0.1" - , port = 8086 - , database = "" +-- * Empty 'prefix' and 'suffix'. +defaultInfluxdbOptions :: Influxdb.WriteParams -> InfluxdbOptions +defaultInfluxdbOptions params = InfluxdbOptions + { writeParams = params , flushInterval = 1000 , prefix = "" , suffix = "" @@ -108,43 +98,33 @@ -> EKG.Store -> (SomeException -> IO () -> IO ()) -> IO ThreadId -forkInfluxdbRestart opts store exceptionHandler = do - eHandle <- Influxdb.createHandle (Influxdb.Config $ T.unpack ("http://" <> host opts <> ":" <> T.pack (show (port opts)) <> "/" <> database opts)) - handle <- case eHandle of - Right h -> return $ h - _ -> unsupportedAddressError - let go = do +forkInfluxdbRestart opts store exceptionHandler = forkIO go + where + go = do terminated <- try $ bracket - (return handle) + (return (writeParams opts)) (\_ -> return ()) (loop store opts) case terminated of Left exception -> exceptionHandler exception go Right _ -> go - forkIO go - where unsupportedAddressError = - ioError (userError ("unsupported address: " ++ T.unpack (host opts))) - -------------------------------------------------------------------------------- -loop :: EKG.Store -> InfluxdbOptions -> Influxdb.Handle -> IO () -loop store opts handle = forever $ do - start <- time +loop :: EKG.Store -> InfluxdbOptions -> Influxdb.WriteParams -> IO () +loop store opts params = forever $ do + start <- getCurrentTime sample <- EKG.sampleAll store - flushSample sample handle opts - end <- time - threadDelay (flushInterval opts * 1000 - fromIntegral (end - start)) - --- | Microseconds since epoch. -time :: IO Int64 -time = (round . (* 1000000.0) . toDouble) `fmap` Time.getPOSIXTime - where toDouble = realToFrac :: Real a => a -> Double + flushSample sample params opts + end <- getCurrentTime + let diff :: Int + diff = truncate (diffUTCTime end start * 1000) + threadDelay (flushInterval opts * 1000 - diff) -flushSample :: EKG.Sample -> Influxdb.Handle -> InfluxdbOptions -> IO () -flushSample sample handle opts = do - t <- getTime Realtime - sendMetrics handle +flushSample :: EKG.Sample -> Influxdb.WriteParams -> InfluxdbOptions -> IO () +flushSample sample params opts = do + t <- getCurrentTime + sendMetrics params (V.map renamed (HashMap.foldlWithKey' (\ms k v -> metrics k v t <> ms) V.empty @@ -161,8 +141,7 @@ metrics _ (EKG.Label {}) _ = V.empty metrics n (EKG.Distribution stats) t = let f n' v = Metric ("dist." <> n <> "." <> n') v t - in V.fromList [ f "mean" (Stats.mean stats) - , f "variance" (Stats.variance stats) + in V.fromList [ f "variance" (Stats.variance stats) , f "count" (fromIntegral $ Stats.count stats) , f "sum" (Stats.sum stats) , f "min" (Stats.min stats) @@ -172,12 +151,11 @@ data Metric = Metric { path :: !T.Text , value :: !Double - , timestamp :: !TimeSpec + , timestamp :: !UTCTime } deriving (Show) -sendMetrics :: Influxdb.Handle -> V.Vector Metric -> IO () -sendMetrics handle metrics = V.forM_ metrics $ \metric -> do +sendMetrics :: Influxdb.WriteParams -> V.Vector Metric -> IO () +sendMetrics params metrics = V.forM_ metrics $ \metric -> do let tags = M.empty - values = M.singleton "value" (Influxdb.F (value metric)) - ts = fromIntegral . toNanoSecs $ (timestamp metric) - in Influxdb.writePoint' handle (path metric) tags values ts + values = M.singleton "value" (Influxdb.FieldFloat (value metric)) + in Influxdb.write params $ Influxdb.Line (fromString . T.unpack $ path metric) tags values (Just (timestamp metric))