ekg-carbon 1.0.3 → 1.0.4

raw patch · 3 files changed

+60/−23 lines, 3 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• Changelog.md +4/−0
• ekg-carbon.cabal +1/−1
• src/System/Remote/Monitoring/Carbon.hs +55/−22

@@ -1,3 +1,7 @@+## 1.0.4++* Added `forkCarbonRestart`, which allows caller's to have a little more control over what happens when the Carbon thread crashes.+ ## 1.0.3  * Increased upper-bounds of:

@@ -1,5 +1,5 @@ name: ekg-carbon-version: 1.0.3+version: 1.0.4 synopsis: An EKG backend to send statistics to Carbon (part of Graphite monitoring tools) homepage: http://github.com/ocharles/ekg-carbon license: BSD3

@@ -14,9 +14,11 @@   ( CarbonOptions(..)   , defaultCarbonOptions   , forkCarbon+  , forkCarbonRestart   ) where -import Control.Concurrent (ThreadId, forkFinally, myThreadId, threadDelay, throwTo)+import Control.Exception (SomeException, try, bracket)+import Control.Concurrent (ThreadId, forkIO, myThreadId, threadDelay, throwTo) import Control.Monad (forever) import Data.Int (Int64) import Data.Monoid ((<>))@@ -79,31 +81,62 @@  -------------------------------------------------------------------------------- -- | Create a thread that periodically flushes the metrics in 'EKG.Store' to--- Carbon.-forkCarbon :: CarbonOptions -> EKG.Store -> IO (ThreadId)-forkCarbon opts store = do-  addrInfos <- Network.getAddrInfo Nothing-                                   (Just $ T.unpack $ host opts)-                                   (Just $ show $ port opts)-  c <- case addrInfos of-    (addrInfo : _) -> Carbon.connect (Network.addrAddress addrInfo)-    _ -> unsupportedAddressError--  parent <- myThreadId-  forkFinally (loop store c opts)-              (\r -> do Carbon.disconnect c-                        case r of-                          Left e  -> throwTo parent e-                          Right _ -> return ())+-- Carbon. If the thread flushing statistics throws an exception (for example, the+-- network connection is lost), this exception will be thrown up to the thread+-- that called 'forkCarbon'. For more control, see 'forkCarbonRestart'.+forkCarbon :: CarbonOptions -> EKG.Store -> IO ThreadId+forkCarbon opts store =+  do parent <- myThreadId+     forkCarbonRestart opts+                       store+                       (\e _ -> throwTo parent e) -  where-  unsupportedAddressError = ioError $ userError $-      "unsupported address: " ++ T.unpack (host opts)+--------------------------------------------------------------------------------+-- | Create a thread that periodically flushes the metrics in 'EKG.Store' to+-- Carbon. If the thread flushing statistics throws an exception (for example, the+-- network connection is lost), the callback function will be invoked with the+-- exception that was thrown, and an 'IO' computation to restart the handler.+--+-- For example, you can use 'forkCarbonRestart' to log failures and restart+-- logging:+--+-- @+-- 'forkCarbonRestart' opts+--                     store+--                     (\ex restart -> do hPutStrLn stderr ("ekg-carbon: " ++ show ex)+--                                        restart)+-- @+forkCarbonRestart :: CarbonOptions+                  -> EKG.Store+                  -> (SomeException -> IO () -> IO ())+                  -> IO ThreadId+forkCarbonRestart opts store exceptionHandler =+  do addrInfos <-+       Network.getAddrInfo Nothing+                           (Just (T.unpack (host opts)))+                           (Just (show (port opts)))+     addrInfo <-+       case addrInfos of+         (addrInfo:_) -> return $ Network.addrAddress addrInfo+         _ -> unsupportedAddressError+     let go =+           do terminated <-+                try $ bracket+                  (Carbon.connect addrInfo)+                  Carbon.disconnect+                  (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 -> Carbon.Connection -> CarbonOptions -> IO ()-loop store socket opts = forever $ do+loop :: EKG.Store -> CarbonOptions -> Carbon.Connection -> IO ()+loop store opts socket = forever $ do   start <- time   sample <- EKG.sampleAll store   flushSample sample socket opts