ekg-influxdb 0.1.0.0 → 0.2.0.0

raw patch · 3 files changed

+48/−64 lines, 3 filesdep +influxdbdep +lensdep −libinfluxdbdep ~basenew-uploaderPVP ok

version bump matches the API change (PVP)

Dependencies added: influxdb, lens

Dependencies removed: libinfluxdb

Dependency ranges changed: base

API changes (from Hackage documentation)

Files

• Changelog.md +7/−2
• ekg-influxdb.cabal +7/−6
• src/System/Remote/Monitoring/Influxdb.hs +34/−56

@@ -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.

@@ -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 <ondrej@slamecka.cz> 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

@@ -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))