log 0.4 → 0.5.0
raw patch · 9 files changed
+235/−23 lines, 9 filesdep +bloodhounddep +conddep +http-clientdep ~hpqtypesdep ~time
Dependencies added: bloodhound, cond, http-client, text-show, transformers
Dependency ranges changed: hpqtypes, time
Files
- log.cabal +24/−5
- src/Log/Backend/ElasticSearch.hs +198/−0
- src/Log/Backend/PostgreSQL.hs +4/−6
- src/Log/Backend/StandardOutput.hs +1/−1
- src/Log/Class.hs +2/−2
- src/Log/Data.hs +1/−4
- src/Log/Internal/Logger.hs +2/−1
- src/Log/Logger.hs +1/−1
- src/Log/Monad.hs +2/−3
log.cabal view
@@ -1,5 +1,5 @@ name: log-version: 0.4+version: 0.5.0 synopsis: Structured logging solution with multiple backends description: A library that provides a way to record structured log messages with multiple backends.@@ -14,7 +14,7 @@ license: BSD3 license-file: LICENSE author: Scrive-maintainer: andrzej@scrive.com+maintainer: Andrzej Rybczak <andrzej@rybczak.net> category: System build-type: Simple cabal-version: >=1.10@@ -25,6 +25,7 @@ library exposed-modules: Log,+ Log.Backend.ElasticSearch, Log.Backend.PostgreSQL, Log.Backend.StandardOutput, Log.Class,@@ -37,10 +38,13 @@ aeson >=0.6.2.0, aeson-pretty >=0.7, base64-bytestring,+ bloodhound, bytestring,+ cond, deepseq, exceptions >=0.6,- hpqtypes >=1.4,+ hpqtypes >=1.5,+ http-client, lifted-base, monad-control >=0.3, monad-time >= 0.2,@@ -49,7 +53,9 @@ split, stm >=2.4, text,- time,+ text-show,+ time >= 1.5,+ transformers, transformers-base, unordered-containers, vector@@ -58,4 +64,17 @@ ghc-options: -O2 -Wall -funbox-strict-fields - default-language: Haskell2010+ default-language: Haskell2010+ default-extensions: BangPatterns+ , FlexibleContexts+ , FlexibleInstances+ , GeneralizedNewtypeDeriving+ , LambdaCase+ , MultiParamTypeClasses+ , NoImplicitPrelude+ , OverloadedStrings+ , RankNTypes+ , RecordWildCards+ , ScopedTypeVariables+ , TypeFamilies+ , UndecidableInstances
+ src/Log/Backend/ElasticSearch.hs view
@@ -0,0 +1,198 @@+module Log.Backend.ElasticSearch (+ ElasticSearchConfig(..)+ , elasticSearchLogger+ ) where++import Control.Arrow (second)+import Control.Concurrent+import Control.Conditional (unlessM)+import Control.Exception+import Control.Monad.IO.Class+import Data.Aeson+import Data.Aeson.Encode.Pretty+import Data.Bits+import Data.IORef+import Data.Monoid+import Data.Monoid.Utils+import Control.Monad+import Data.Time+import Data.Time.Clock.POSIX+import Data.Word+import Database.Bloodhound hiding (Status)+import Log+import Network.HTTP.Client+import Prelude+import TextShow+import qualified Data.ByteString as BS+import qualified Data.ByteString.Base64 as B64+import qualified Data.ByteString.Lazy.Char8 as BSL+import qualified Data.HashMap.Strict as H+import qualified Data.Text as T+import qualified Data.Text.Encoding as T+import qualified Data.Traversable as F+import qualified Data.Vector as V++data ElasticSearchConfig = ElasticSearchConfig {+ esServer :: !T.Text+ , esIndex :: !T.Text+ , esMapping :: !T.Text+ } deriving (Eq, Show)++----------------------------------------++elasticSearchLogger :: ElasticSearchConfig -> IO Word32 -> IO Logger+elasticSearchLogger ElasticSearchConfig{..} genRandomWord = do+ checkElasticSearchConnection+ indexRef <- newIORef $ IndexName T.empty+ mkBulkLogger "ElasticSearch" $ \msgs -> do+ now <- getCurrentTime+ oldIndex <- readIORef indexRef+ -- Bloodhound doesn't support letting ES autogenerate IDs, so let's generate+ -- them ourselves. An ID of a log message is 12 bytes (4 bytes: random, 4+ -- bytes: current time as epoch, 4 bytes: insertion order) encoded as+ -- Base64. This makes eventual collisions practically impossible.+ baseID <- (<>)+ <$> (littleEndianRep <$> liftIO genRandomWord)+ <*> pure (littleEndianRep . floor $ timeToDouble now)+ retryOnException . runBH_ $ do+ -- ElasticSearch index names are additionally indexed by date so that each+ -- day is logged to a separate index to make log management easier.+ let index = IndexName $ T.concat [+ esIndex+ , "-"+ , T.pack $ formatTime defaultTimeLocale "%F" now+ ]+ when (oldIndex /= index) $ do+ -- There is an obvious race condition in the presence of more than one+ -- logger instance running, but it's irrelevant as attempting to create+ -- index that already exists is harmless.+ unlessM (indexExists index) $ do+ -- Bloodhound is weird and won't let us create index using default+ -- settings, so pass these as the default ones.+ let indexSettings = IndexSettings {+ indexShards = ShardCount 4+ , indexReplicas = ReplicaCount 1+ }+ void $ createIndex indexSettings index+ reply <- putMapping index mapping LogsMapping+ when (not $ isSuccess reply) $ do+ error $ "ElasticSearch: error while creating mapping:" <+> T.unpack (T.decodeUtf8 . BSL.toStrict . jsonToBSL $ decodeReply reply)+ liftIO $ writeIORef indexRef index+ let jsonMsgs = V.fromList $ map (toJsonMsg now) $ zip [1..] msgs+ reply <- bulk $ V.map (toBulk index baseID) jsonMsgs+ -- Try to parse parts of reply to get information about log messages that+ -- failed to be inserted for some reason.+ let replyBody = decodeReply reply+ result = do+ Object response <- return replyBody+ Bool hasErrors <- "errors" `H.lookup` response+ Array jsonItems <- "items" `H.lookup` response+ items <- F.forM jsonItems $ \v -> do+ Object item <- return v+ Object index_ <- "index" `H.lookup` item+ return index_+ guard $ V.length items == V.length jsonMsgs+ return (hasErrors, items)+ case result of+ Nothing -> liftIO . BSL.putStrLn $ "ElasticSearch: unexpected response:" <+> jsonToBSL replyBody+ Just (hasErrors, items) -> when hasErrors $ do+ -- If any message failed to be inserted because of type mismatch, go+ -- back to them, replace their data with elastic search error and put+ -- old data into its own namespace to work around insertion errors.+ let failed = V.findIndices (H.member "error") items+ dummyMsgs <- V.forM failed $ \n -> do+ dataNamespace <- liftIO genRandomWord+ let modifyData oldData = object [+ "__es_error" .= H.lookup "error" (items V.! n)+ , "__es_modified" .= True+ , ("__data_" <> showt dataNamespace) .= oldData+ ]+ return . second (H.adjust modifyData "data") $ jsonMsgs V.! n+ -- Attempt to put modified messages and ignore any further errors.+ void $ bulk (V.map (toBulk index baseID) dummyMsgs)+ where+ server = Server esServer+ mapping = MappingName esMapping++ checkElasticSearchConnection :: IO ()+ checkElasticSearchConnection = try (void $ runBH_ listIndices) >>= \case+ Left (ex::HttpException) -> error $ "ElasticSearch: unexpected error: " <> show ex <> " (is ElasticSearch server running?)"+ Right () -> return ()++ retryOnException :: forall r. IO r -> IO r+ retryOnException m = try m >>= \case+ Left (ex::SomeException) -> do+ putStrLn $ "ElasticSearch: unexpected error: " <> show ex <> ", retrying in 10 seconds"+ threadDelay $ 10 * 1000000+ retryOnException m+ Right result -> return result++ timeToDouble :: UTCTime -> Double+ timeToDouble = realToFrac . utcTimeToPOSIXSeconds++ runBH_ :: forall r. BH IO r -> IO r+ runBH_ = withBH defaultManagerSettings server++ jsonToBSL :: Value -> BSL.ByteString+ jsonToBSL = encodePretty' defConfig { confIndent = 2 }++ toJsonMsg :: UTCTime -> (Word32, LogMessage) -> (Word32, H.HashMap T.Text Value)+ toJsonMsg now (n, msg) = (n, H.union jMsg $ H.fromList [+ ("insertion_order", toJSON n)+ , ("insertion_time", toJSON now)+ ])+ where+ Object jMsg = toJSON msg++ mkDocId :: BS.ByteString -> Word32 -> DocId+ mkDocId baseID insertionOrder = DocId . T.decodeUtf8 . B64.encode $ BS.concat [+ baseID+ , littleEndianRep insertionOrder+ ]++ toBulk :: IndexName -> BS.ByteString -> (Word32, H.HashMap T.Text Value) -> BulkOperation+ toBulk index baseID (n, obj) = BulkIndex index mapping (mkDocId baseID n) $ Object obj++data LogsMapping = LogsMapping+instance ToJSON LogsMapping where+ toJSON LogsMapping = object [+ "logs" .= object [+ "properties" .= object [+ "insertion_order" .= object [+ "type" .= ("integer"::T.Text)+ ]+ , "insertion_time" .= object [+ "type" .= ("date"::T.Text)+ , "format" .= ("strict_date_time"::T.Text)+ ]+ , "time" .= object [+ "type" .= ("date"::T.Text)+ , "format" .= ("strict_date_time"::T.Text)+ ]+ , "domain" .= object [+ "type" .= ("string"::T.Text)+ ]+ , "level" .= object [+ "type" .= ("string"::T.Text)+ ]+ , "component" .= object [+ "type" .= ("string"::T.Text)+ ]+ , "message" .= object [+ "type" .= ("string"::T.Text)+ ]+ ]+ ]+ ]++----------------------------------------++littleEndianRep :: Word32 -> BS.ByteString+littleEndianRep = fst . BS.unfoldrN 4 step+ where+ step n = Just (fromIntegral $ n .&. 0xff, n `shiftR` 8)++decodeReply :: Reply -> Value+decodeReply reply = case eitherDecode' $ responseBody reply of+ Right body -> body+ Left err -> object ["decoding_error" .= err]
src/Log/Backend/PostgreSQL.hs view
@@ -1,5 +1,3 @@-{-# LANGUAGE BangPatterns, GeneralizedNewtypeDeriving, OverloadedStrings- , RecordWildCards, ScopedTypeVariables #-} module Log.Backend.PostgreSQL (pgLogger) where import Control.Applicative@@ -7,13 +5,13 @@ import Control.Exception.Lifted import Control.Monad.State.Lazy import Data.Aeson-import Data.ByteString.Lazy (toStrict) import Data.List.Split import Data.Monoid import Data.Monoid.Utils import Data.String import Data.Typeable-import Database.PostgreSQL.PQTypes hiding (put)+import Database.PostgreSQL.PQTypes+import Prelude import qualified Data.ByteString.Base64 as B64 import qualified Data.Foldable as F import qualified Data.HashMap.Strict as H@@ -28,7 +26,7 @@ deriving Enum -- | Create logger that inserts log messages into PostgreSQL database.-pgLogger :: ConnectionSource -> IO Logger+pgLogger :: ConnectionSourceM IO -> IO Logger pgLogger cs = mkBulkLogger loggerName $ mapM_ (serialize $ Attempt 1) . chunksOf 1000 where@@ -137,7 +135,7 @@ , "," <?> lmComponent , "," <?> Array1 lmDomain , "," <?> lmMessage- , "," <?> toStrict (encode lmData) <> "::jsonb"+ , "," <?> JSONB (encode lmData) , ")" ]
src/Log/Backend/StandardOutput.hs view
@@ -1,6 +1,6 @@-{-# LANGUAGE OverloadedStrings #-} module Log.Backend.StandardOutput (stdoutLogger) where +import Prelude import qualified Data.Text.IO as T import Log.Data
src/Log/Class.hs view
@@ -1,6 +1,5 @@ {-# OPTIONS_GHC -fno-warn-deprecated-flags #-}-{-# LANGUAGE FlexibleContexts, FlexibleInstances, OverlappingInstances- , RankNTypes, UndecidableInstances #-}+{-# LANGUAGE OverlappingInstances #-} module Log.Class ( UTCTime , MonadTime(..)@@ -19,6 +18,7 @@ import Data.Aeson import Data.Aeson.Types import Data.Time+import Prelude import qualified Data.Text as T import Log.Data
src/Log/Data.hs view
@@ -1,4 +1,3 @@-{-# LANGUAGE CPP, OverloadedStrings, RecordWildCards #-} module Log.Data ( LogLevel(..) , showLogLevel@@ -13,9 +12,7 @@ import Data.Aeson.Types import Data.ByteString.Lazy (toStrict) import Data.Time-#if !MIN_VERSION_time(1,5,0)-import System.Locale-#endif+import Prelude import qualified Data.Text as T import qualified Data.Text.Encoding as T
src/Log/Internal/Logger.hs view
@@ -1,4 +1,3 @@-{-# LANGUAGE RecordWildCards #-} module Log.Internal.Logger ( Logger(..) , execLogger@@ -7,6 +6,8 @@ import Control.Concurrent.STM import Data.IORef+import Data.Monoid+import Prelude import Log.Data
src/Log/Logger.hs view
@@ -1,4 +1,3 @@-{-# LANGUAGE OverloadedStrings, RecordWildCards #-} module Log.Logger ( Logger , mkLogger@@ -14,6 +13,7 @@ import Control.Monad import Data.IORef import Data.Monoid+import Prelude import qualified Data.Text as T import qualified Data.Text.IO as T
src/Log/Monad.hs view
@@ -1,6 +1,4 @@-{-# LANGUAGE CPP, FlexibleContexts, FlexibleInstances- , GeneralizedNewtypeDeriving, MultiParamTypeClasses, OverloadedStrings- , RecordWildCards, TypeFamilies, UndecidableInstances #-}+{-# LANGUAGE CPP #-} module Log.Monad ( Logger , LoggerEnv(..)@@ -19,6 +17,7 @@ import Data.Aeson import Data.Aeson.Types import Data.Text (Text)+import Prelude import qualified Control.Exception as E import qualified Data.HashMap.Strict as H