diff --git a/LICENSE b/LICENSE --- a/LICENSE +++ b/LICENSE @@ -1,5 +1,4 @@ Copyright (c) 2013 Joseph Abrahamson, Reify Health LLC - 2015 Trevis Elser Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the diff --git a/riemann.cabal b/riemann.cabal --- a/riemann.cabal +++ b/riemann.cabal @@ -1,21 +1,16 @@ name: riemann -version: 0.1.0.2 +version: 0.1.1.0 synopsis: A Riemann client for Haskell -description: Very simple event sending to a Riemann server. -homepage: https://github.com/telser/riemann-hs +description: A Riemann client for Haskell +homepage: https://github.com/tel/riemann-hs license: MIT license-file: LICENSE -author: Joseph Abrahamson, Trevis Elser -maintainer: trevis@clickscape.com -copyright: 2013 Joseph Abrahamson, 2015 Trevis Elser +author: Joseph Abrahamson +maintainer: me@jspha.com +copyright: 2013 Joseph Abrahamson category: Network build-type: Simple -cabal-version: >=1.10 - - -source-repository head - type: git - location: https://github.com/telser/riemann-hs/ +cabal-version: >=1.20 library hs-source-dirs: src @@ -23,51 +18,70 @@ Network.Monitoring.Riemann Network.Monitoring.Riemann.Types other-modules: + Network.Monitoring.Riemann.TCP build-depends: - base >= 4.8.0 && < 5.0 - , network >= 2.4.0.0 - , text >= 0.11.2.3 - , containers >= 0.4.2.0 - , protobuf >= 0.2 - , transformers >= 0.4.3.0 && < 0.5 - , either - , errors < 2.0 - , lens - , time - , data-default - , cereal + base >= 4.5 && < 5, + network >= 2.4.0.0, + text >= 0.11.2.3, + containers >= 0.4.2.0, + protobuf >= 0.2, + bytestring >= 0.10, + lens, + time, + errors, + data-default, + mtl, + transformers, + cereal default-language: Haskell2010 --- test-suite doctest --- type: exitcode-stdio-1.0 --- ghc-options: -threaded --- hs-source-dirs: test --- main-is: Doctest.hs --- build-depends: --- base --- , transformers >= 0.4.3.0 && < 0.5 --- , either --- , errors < 2.0 --- , directory --- , either --- , doctest --- , filepath --- default-language: Haskell2010 +executable riemann-client + main-is: riemann.hs + hs-source-dirs: src + ghc-options: -threaded + build-depends: + base, + riemann, + network >= 2.4.0.0, + text >= 0.11.2.3, + containers >= 0.4.2.0, + protobuf >= 0.2, + wreq, + http-client, + bytestring, + lens, + time, + errors, + data-default, + mtl, + transformers, + cereal + default-language: Haskell2010 + ghc-options: -Wall -fwarn-tabs +test-suite doctest + type: exitcode-stdio-1.0 + ghc-options: -threaded + hs-source-dirs: test + main-is: Doctest.hs + build-depends: + base, + directory, + doctest, + filepath + default-language: Haskell2010 + test-suite property type: exitcode-stdio-1.0 ghc-options: -threaded hs-source-dirs: test main-is: Property.hs build-depends: - base - , transformers >= 0.4.3.0 && < 0.5 - , either - , errors < 2.0 - , QuickCheck >= 2.5.1.0 - , test-framework - , test-framework-quickcheck2 - , riemann + base, + QuickCheck >= 2.5.1.0, + test-framework, + test-framework-quickcheck2, + riemann default-language: Haskell2010 test-suite unit @@ -76,12 +90,9 @@ hs-source-dirs: test main-is: Unit.hs build-depends: - base - , transformers >= 0.4.3.0 && < 0.5 - , either - , errors < 2.0 - , HUnit - , test-framework - , test-framework-hunit - , riemann + base, + HUnit, + test-framework, + test-framework-hunit, + riemann default-language: Haskell2010 diff --git a/src/Network/Monitoring/Riemann.hs b/src/Network/Monitoring/Riemann.hs --- a/src/Network/Monitoring/Riemann.hs +++ b/src/Network/Monitoring/Riemann.hs @@ -1,30 +1,31 @@ -{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE ScopedTypeVariables #-} module Network.Monitoring.Riemann ( module Network.Monitoring.Riemann.Types, module Data.Int, - Client, makeClient, - sendEvent, sendEvent' - {-, sendEvent'-} - ) where + Client(..), makeUDPClient, makeTCPClient, isError, + closeClient, + sendEventT, sendEvent ) where +import Network.Monitoring.Riemann.TCP import Network.Monitoring.Riemann.Types import Data.Default import Data.Int +import Data.IORef (IORef, atomicModifyIORef', + newIORef, readIORef, + writeIORef) import Data.ProtocolBuffers import Data.Serialize.Put -import Data.Time.Clock import Data.Time.Clock.POSIX -import Control.Applicative -import qualified Control.Error as Error +import Control.Error import Control.Exception import Control.Lens -import qualified Control.Monad as CM -import qualified Control.Monad.IO.Class as CMIC -import Control.Monad.Trans.Either -import qualified Control.Monad.Trans.Except as Except +import Control.Monad +import Control.Monad.Trans import Network.Socket hiding (recv, recvFrom, send, sendTo) @@ -113,18 +114,30 @@ -} -data Client = UDP (Maybe (Socket, AddrInfo)) - deriving (Show, Eq) +data Client = UDP { riemannHost :: Hostname + , riemannPort :: Port + , udpClient :: Either IOException (Socket, AddrInfo) + } + | TCP { riemannHost :: Hostname + , riemannPort :: Port + , tcpClient :: IORef TCPState + -- ^ TCP connection's state is maintained into a mutable reference in order to allow + -- not mutating the @Client@ structure + } -type Hostname = String -type Port = Int +-- |Checks whether or not a given @Client@ is in an error state. +isError :: (MonadIO m) => Client -> m Bool +isError (UDP _ _ (Left _)) = return True +isError (UDP _ _ (Right _ )) = return False +isError (TCP _ _ r) = isTCPError <$> liftIO (readIORef r) + -- | Attempts to bind a UDP client at the passed 'Hostname' and -- 'Port'. Failures are silently ignored---failure in monitoring -- should not cause an application failure... -makeClient :: Hostname -> Port -> IO Client -makeClient hn po = UDP . Error.rightMay <$> sock - where sock :: IO (Either SomeException (Socket, AddrInfo)) +makeUDPClient :: Hostname -> Port -> IO Client +makeUDPClient hn po = UDP hn po <$> sock + where sock :: IO (Either IOException (Socket, AddrInfo)) sock = try $ do addrs <- getAddrInfo (Just $ defaultHints { @@ -134,21 +147,64 @@ case addrs of [] -> fail "No accessible addresses" (addy:_) -> do - s <- socket (addrFamily addy) - Datagram - (addrProtocol addy) + s <- socket AF_INET + Datagram + defaultProtocol return (s, addy) +-- | Attempts to connect to given riemann server +-- Returns an initialised TCP client that can later be used to send events to given riemann host/port. +-- TCP Clients always try to reconnect to server when an event is sent and connection is either closed +-- or in error. +makeTCPClient :: Hostname -> Port -> IO Client +makeTCPClient hn po = do + ref <- newIORef CnxClosed + cnx <- tcpConnect hn po CnxClosed + writeIORef ref cnx + return $ TCP hn po ref + + +closeClient :: Client -> IO () +closeClient (UDP _ _ cnx) = either (const $ return ()) (close . fst) cnx +closeClient (TCP _ _ r ) = do + tcp <- readIORef r + case tcp of + CnxOpen (s,_) -> close s + _ -> return () + atomicModifyIORef' r (const $ (CnxClosed, ())) + + -- | Attempts to forward an event to a client. Fails silently. -sendEvent :: CMIC.MonadIO m => Client -> Event -> m () -sendEvent c = CMIC.liftIO . CM.void . runEitherT . sendEvent' c +sendEvent :: MonadIO m => Client -> Event -> m () +sendEvent c = liftIO . void . runExceptT . sendEventT c -- | Attempts to forward an event to a client. If it fails, it'll --- return an 'IOException' in the 'Either'. -sendEvent' :: Client -> Event -> EitherT IOException IO () -sendEvent' (UDP Nothing) _ = return () -sendEvent' (UDP (Just (s, addy))) e = Error.tryIO $ do - current <- getCurrentTime - let now = round (utcTimeToPOSIXSeconds current) +-- return an 'IOException' in the 'ExceptT', otherwise it returns the passed @Client@, possibly +-- modified to cope for changes in state. +sendEventT :: Client -> Event -> ExceptT IOException IO Client +sendEventT client@(UDP _ _ _) event = sendUDPEvent client event +sendEventT client@(TCP _ _ _ ) event = sendTCPEvent client event + +sendTCPEvent :: (MonadIO m) => Client -> Event -> ExceptT IOException m Client +sendTCPEvent c@(TCP h n r) event = tryIO $ do + tcp <- readIORef r + case tcp of + CnxOpen (s,_) -> doSendTCPEvent r s event >> return c + s -> do + s' <- tcpConnect h n s + writeIORef r s' + case s' of + CnxOpen (sock,_) -> doSendTCPEvent r sock event >> return c + CnxError e -> throw e + _ -> fail "connection is closed after retry" +sendTCPEvent _ _ = fail "trying to send TCP event through UDP client" + + +sendUDPEvent :: (MonadIO m) => Client -> Event -> ExceptT IOException m Client +sendUDPEvent (UDP _ _ (Left e)) _ = throwE e +sendUDPEvent c@(UDP _ _ (Right (s, addy))) e = tryIO $ do + now <- fmap round getPOSIXTime let msg = def & events .~ [e & time ?~ now] - CM.void $ sendTo s (runPut $ encodeMessage msg) (addrAddress addy) + void $ sendTo s (runPut $ encodeMessage msg) (addrAddress addy) + return c +sendUDPEvent _ _ = fail "trying to send UDP event through TCP client" diff --git a/src/Network/Monitoring/Riemann/TCP.hs b/src/Network/Monitoring/Riemann/TCP.hs new file mode 100644 --- /dev/null +++ b/src/Network/Monitoring/Riemann/TCP.hs @@ -0,0 +1,100 @@ +-- | TCP Specific connection handling for monitoring with Riemann +module Network.Monitoring.Riemann.TCP(tcpConnect, isTCPError, TCPState(..), doSendTCPEvent) where + +import Network.Monitoring.Riemann.Types + +import qualified Data.ByteString as BS +import Data.Default +import Data.IORef (IORef, writeIORef) +import Data.ProtocolBuffers +import Data.Serialize.Get +import Data.Serialize.Put +import Data.Time.Clock.POSIX + +import Control.Exception +import Control.Lens +import Control.Monad + +import Network.Socket hiding (recv, recvFrom, send, + sendTo) +import Network.Socket.ByteString + + +-- | High-level and probably too naive state of TCP connection to Riemann server. +data TCPState = CnxClosed + -- ^Connection is known to be closed + | CnxOpen (Socket, AddrInfo) + -- ^Connection is expected to be usable with given socket and address + | CnxError IOException + -- ^Connection failed with given exception + deriving Show + +-- |Is given state in error? +isTCPError :: TCPState -> Bool +isTCPError (CnxError _) = True +isTCPError _ = False + +-- | Try connecting with TCP to given host/port. +-- +-- * If the passed @TCPState@ is already connected then no new connection attempt is made +-- * Otherwise, we try to connect to given host/port and update @TCPState@ accordingly +-- +-- '''Note''': We never use IPv6 address resolved for given hostname. +tcpConnect :: Hostname -> Port -> TCPState -> IO TCPState +tcpConnect _ _ s@(CnxOpen _) = return s +tcpConnect hn po _ = do + res <- try $ doConnect hn po + return $ case res of + Left e -> CnxError e + Right s -> CnxOpen s + +tcpv4 :: AddrInfo -> Bool +tcpv4 addr = addrSocketType addr == Stream && + addrFamily addr == AF_INET + +doConnect :: HostName -> Port -> IO (Socket, AddrInfo) +doConnect hn po = do addrs <- getAddrInfo + (Just $ defaultHints { + addrFlags = [AI_NUMERICSERV] }) + (Just hn) + (Just $ show po) + case (filter tcpv4 addrs) of + [] -> fail ("No accessible addresses in " ++ show addrs) + (addy:_) -> do + s <- socket AF_INET + Stream + defaultProtocol + connect s (addrAddress addy) + return (s, addy) + + +-- | Try sending a riemann event to given @Socket@, updating given state accordingly. +doSendTCPEvent :: IORef TCPState -> Socket -> Event -> IO () +doSendTCPEvent r s event = do + sending <- try $ do + now <- fmap round getPOSIXTime + let msg = def & events .~ [event & time ?~ now] + bytes = runPut $ encodeMessage msg + bytesWithLen = runPut (putWord32be (fromIntegral $ BS.length bytes) >> putByteString bytes) + void $ send s bytesWithLen + case sending of + Left e -> writeIORef r (CnxError e) + Right _ -> doReceiveAck r s + +-- | Re +doReceiveAck :: IORef TCPState -> Socket -> IO () +doReceiveAck r s = do + bs <- recv s 4096 + if BS.length bs == 0 + then close s >> writeIORef r CnxClosed + else case decoded bs of + Left _ -> return () -- TODO something useful, but what ? Close the socket ? + Right msg -> do + let st = msg ^. msgState + case st of + Ok -> return () + Error _ -> return () -- TODO something useful + Unknown -> return () + + where + decoded bs = runGet decodeMessage bs :: Either String Msg diff --git a/src/Network/Monitoring/Riemann/Types.hs b/src/Network/Monitoring/Riemann/Types.hs --- a/src/Network/Monitoring/Riemann/Types.hs +++ b/src/Network/Monitoring/Riemann/Types.hs @@ -1,11 +1,11 @@ -{-# LANGUAGE DataKinds #-} -{-# LANGUAGE DeriveGeneric #-} -{-# LANGUAGE TypeOperators #-} -{-# LANGUAGE OverloadedStrings #-} -{-# LANGUAGE TemplateHaskell #-} -{-# LANGUAGE TupleSections #-} -{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE TupleSections #-} +{-# LANGUAGE TypeOperators #-} module Network.Monitoring.Riemann.Types ( HasState (..), @@ -14,29 +14,24 @@ State, Event, Query, Msg, ev, once, attributes, - MsgState, msgState, states, events + MsgState(..), msgState, states, events, + Hostname, Port ) where -import Data.ProtocolBuffers - -import GHC.Generics hiding (to, from) -import qualified GHC.Generics as G - -import Data.Int -import Data.Monoid -import Data.Maybe -import Data.Time.Clock -import Data.Default -import Data.List -import Data.Text (Text) -import qualified Data.Text as T -import Data.Map (Map) -import qualified Data.Map as M - -import Control.Lens -import Control.Monad -import Control.Arrow -import Control.Applicative +import Control.Arrow +import Control.Lens +import Control.Monad +import Data.Default +import Data.Int +import Data.List +import Data.Map (Map) +import qualified Data.Map as M +import Data.Maybe +import Data.ProtocolBuffers +import Data.Text (Text) +import qualified Data.Text as T +import GHC.Generics hiding (from, to) +import qualified GHC.Generics as G -- $class @@ -45,7 +40,7 @@ -- 'time'. These shared types give rise to restrictedly polymorphic -- lenses. class HasState a where - time :: Lens' a (Maybe (Signed Int64)) + time :: Lens' a (Maybe Int64) -- ^ The time of the event, in unix epoch seconds state :: Lens' a (Maybe Text) -- ^ Any string less than 255 bytes, e.g. "ok", "warning", @@ -100,7 +95,7 @@ -- | 'State' is an object within Riemann's index, a result from a -- 'Query'. data State = State { - _stateTime :: Optional 1 (Value (Signed Int64)), + _stateTime :: Optional 1 (Value Int64), _stateState :: Optional 2 (Value Text), _stateService :: Optional 3 (Value Text), _stateHost :: Optional 4 (Value Text), @@ -113,14 +108,14 @@ -- | 'Event' is a description of an application-level event, emitted -- to Riemann for indexing. data Event = Event { - _eventTime :: Optional 1 (Value (Signed Int64)), + _eventTime :: Optional 1 (Value Int64), _eventState :: Optional 2 (Value Text), _eventService :: Optional 3 (Value Text), _eventHost :: Optional 4 (Value Text), _eventDescription :: Optional 5 (Value Text), _eventTags :: Repeated 7 (Value Text), _eventTtl :: Optional 8 (Value Float), - + _eventAttributes :: Repeated 9 (Message Attribute), _eventMetricSInt :: Optional 13 (Value (Signed Int64)), _eventMetricD :: Optional 14 (Value Double), @@ -166,7 +161,7 @@ once = stateOnce . field instance Show State where - show s = "State { " ++ intercalate ", " innards ++ " }" + show s = "State { " ++ intercalate ", " innards ++ " }" where innards = catMaybes [ showM "time" time, showM "state" state, @@ -178,7 +173,7 @@ showM "once" once ] showM name l = (\x -> name ++ " = " ++ x) . show <$> s ^. l - showL name l = let lst = s ^. l + showL name l = let lst = s ^. l in if null lst then Nothing else Just $ name ++ " = " ++ show lst instance Default State where @@ -233,11 +228,11 @@ class AMetric a where metric :: Lens' Event (Maybe a) -instance AMetric Int where +instance AMetric Int where metric = eventMetricSInt . field . mapping (iso fromIntegral fromIntegral) -instance AMetric Integer where +instance AMetric Integer where metric = eventMetricSInt . field . mapping (iso fromIntegral fromIntegral) -instance AMetric (Signed Int64) where +instance AMetric (Signed Int64) where metric = eventMetricSInt . field instance AMetric Double where metric = eventMetricD . field @@ -256,7 +251,7 @@ sequen (a, fb) = (a,) <$> fb instance Show Event where - show s = "Event { " ++ intercalate ", " innards ++ " }" + show s = "Event { " ++ intercalate ", " innards ++ " }" where innards = catMaybes [ showM "time" time, showM "state" state, @@ -271,13 +266,13 @@ showM "metric_d" (metric :: Lens' Event (Maybe Double)) ] showM name l = (\x -> name ++ " = " ++ x) . show <$> s ^. l - showMap name l = let mp = s ^. l - in if M.null mp then Nothing + showMap name l = let mp = s ^. l + in if M.null mp then Nothing else Just . (\x -> name ++ " = " ++ show x) $ mp - showL name l = let lst = s ^. l - in if null lst then Nothing + showL name l = let lst = s ^. l + in if null lst then Nothing else Just $ name ++ " = " ++ show lst - + instance Default Event where def = Event { _eventTime = putField Nothing, @@ -301,22 +296,22 @@ -- | Create a simple 'Event' with state "ok". -- --- >>> get state $ ev "service" (0 :: (Signed Int64)) +-- >>> view state $ ev "service" (0 :: (Signed Int64)) -- Just "ok" -- --- >>> get service $ ev "service" (0 :: (Signed Int64)) +-- >>> view service $ ev "service" (0 :: (Signed Int64)) -- Just "service" -- --- >>> get metric $ ev "service" (0 :: (Signed Int64)) :: Maybe (Signed Int64) --- Just 0 +-- >>> view metric $ ev "service" (0 :: (Signed Int64)) :: Maybe (Signed Int64) +-- Just (Signed 0) -- --- >>> get tags $ ev "service" (0 :: (Signed Int64)) +-- >>> view tags $ ev "service" (0 :: (Signed Int64)) -- [] ev :: AMetric a => String -> a -> Event ev serv met = - def - & state ?~ "ok" - & service ?~ T.pack serv + def + & state ?~ "ok" + & service ?~ T.pack serv & metric ?~ met -- $query @@ -336,7 +331,7 @@ mappend = defMappend instance Show Query where - show s = "Query { " ++ intercalate ", " innards ++ " }" + show s = "Query { " ++ intercalate ", " innards ++ " }" where innards = catMaybes [ showM "query" query ] @@ -370,7 +365,7 @@ events = msgEvents . field instance Show Msg where - show s = "Msg { " ++ intercalate ", " innards ++ " }" + show s = "Msg { " ++ intercalate ", " innards ++ " }" where innards = catMaybes [ showMsgState, showL "states" states, @@ -378,7 +373,7 @@ showM "query" query ] showM name l = (\x -> name ++ " = " ++ x) . show <$> s ^. l - showL name l = let lst = s ^. l + showL name l = let lst = s ^. l in if null lst then Nothing else Just $ name ++ " = " ++ show lst showMsgState = ("msgState = " ++) <$> case s ^. msgState of Ok -> Just "Ok" @@ -402,3 +397,7 @@ instance Monoid Msg where mempty = def mappend = defMappend + + +type Hostname = String +type Port = Int diff --git a/src/riemann.hs b/src/riemann.hs new file mode 100644 --- /dev/null +++ b/src/riemann.hs @@ -0,0 +1,43 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} + +module Main where + +import Control.Concurrent +import Control.Error +import Control.Exception (catch) +import Control.Lens +import Control.Monad +import Control.Monad.IO.Class +import Data.Monoid ((<>)) +import Data.Time.Clock +import Network.HTTP.Client (HttpException (..)) +import Network.Monitoring.Riemann +import Network.Wreq +import System.Environment + +main :: IO () +main = do + [rhost,rport,uriList] <- getArgs + c <- makeTCPClient rhost (read rport) + uris <- lines <$> readFile uriList + forever $ checkAll c uris + + +checkAll :: Client -> [ String ] -> IO () +checkAll c uris = forM_ uris (check c) >> threadDelay (60 * 1000000) + +check :: Client -> String -> IO () +check c uri = do + putStr $ "checking " <> uri <> " " + st <- getCurrentTime + ok <- (get uri >> return True) `catch` \ (_ :: HttpException) -> return False + en <- getCurrentTime + let event = ev ("http " <> uri) (realToFrac (diffUTCTime en st) :: Double) & state ?~ if ok then "ok" else "error" + res <- liftIO $ runExceptT $ sendEventT c event + -- not very useful error handling, if socket fails to connect at startup it cannot reconnects later + either + (\e -> putStrLn ("error connecting to riemann server: " <> show e)) + (const $ putStrLn "sent event to riemann") + res + diff --git a/test/Doctest.hs b/test/Doctest.hs new file mode 100644 --- /dev/null +++ b/test/Doctest.hs @@ -0,0 +1,20 @@ +import Control.Applicative +import Control.Monad +import Data.List +import System.Directory +import System.FilePath +import Test.DocTest + +main = getSources >>= doctest + +getSources :: IO [FilePath] +getSources = filter (isSuffixOf ".hs") <$> go "src" + where + go dir = do + (dirs, files) <- getFilesAndDirectories dir + (files ++) . concat <$> mapM go dirs + +getFilesAndDirectories :: FilePath -> IO ([FilePath], [FilePath]) +getFilesAndDirectories dir = do + c <- map (dir ) . filter (`notElem` ["..", "."]) <$> getDirectoryContents dir + (,) <$> filterM doesDirectoryExist c <*> filterM doesFileExist c diff --git a/test/Property.hs b/test/Property.hs --- a/test/Property.hs +++ b/test/Property.hs @@ -1,19 +1,10 @@ import Test.Framework (defaultMain, testGroup) import Test.QuickCheck import Test.Framework.Providers.QuickCheck2 (testProperty) -import Network.Monitoring.Riemann --- main :: IO () --- main = defaultMain [ --- testGroup "(default)" [ --- testProperty "isGood" (\a -> a == (a `asTypeOf` True)) --- ] --- ] - main :: IO () -main = do - client <- makeClient "127.0.0.1" 5555 - print "Client made now attempting to send event" - sendEvent client (ev "test_service" (0.0 :: Double)) - print "post event send" - print $ show client +main = defaultMain [ + testGroup "(default)" [ + testProperty "isGood" (\a -> a == (a `asTypeOf` True)) + ] + ]