log-elasticsearch 0.8.1 → 0.9.0.0

raw patch · 10 files changed

+669/−324 lines, 10 filesdep ~bloodhoundPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: bloodhound

API changes (from Hackage documentation)

Files

• log-elasticsearch.cabal +10/−4
• src/Log/Backend/ElasticSearch.hs +4/−252
• src/Log/Backend/ElasticSearch/Internal.hs +3/−29
• src/Log/Backend/ElasticSearch/Lens.hs +4/−39
• src/Log/Backend/ElasticSearch/V1.hs +253/−0
• src/Log/Backend/ElasticSearch/V1/Internal.hs +31/−0
• src/Log/Backend/ElasticSearch/V1/Lens.hs +40/−0
• src/Log/Backend/ElasticSearch/V5.hs +253/−0
• src/Log/Backend/ElasticSearch/V5/Internal.hs +31/−0
• src/Log/Backend/ElasticSearch/V5/Lens.hs +40/−0

@@ -1,5 +1,5 @@ name:                log-elasticsearch-version:             0.8.1+version:             0.9.0.0 synopsis:            Structured logging solution (Elasticsearch back end)  description:         Elasticsearch back end for the 'log' library.@@ -24,16 +24,22 @@   Location: https://github.com/scrive/log.git  library-  exposed-modules:     Log.Backend.ElasticSearch-                       Log.Backend.ElasticSearch.Internal+  exposed-modules:     Log.Backend.ElasticSearch.V1+                       Log.Backend.ElasticSearch.V1.Internal+                       Log.Backend.ElasticSearch.V1.Lens+                       Log.Backend.ElasticSearch.V5+                       Log.Backend.ElasticSearch.V5.Internal+                       Log.Backend.ElasticSearch.V5.Lens+                       Log.Backend.ElasticSearch                        Log.Backend.ElasticSearch.Lens+                       Log.Backend.ElasticSearch.Internal   build-depends:       base <5,                        log-base >= 0.7,                        aeson >=0.11.0.0,                        aeson-pretty >=0.8.2,                        bytestring,                        base64-bytestring,-                       bloodhound >= 0.11.1 && < 0.13,+                       bloodhound >= 0.13 && < 0.15,                        deepseq,                        http-client,                        http-client-tls,

@@ -1,253 +1,5 @@--- | Elasticsearch logging back-end.-module Log.Backend.ElasticSearch (-    ElasticSearchConfig-  , esServer-  , esIndex-  , esMapping-  , esLogin-  , esLoginInsecure-  , defaultElasticSearchConfig-  , withElasticSearchLogger-  , elasticSearchLogger-  ) where--import Control.Applicative-import Control.Arrow (second)-import Control.Concurrent-import Control.Exception-import Control.Monad-import Control.Monad.IO.Class-import Data.Aeson-import Data.Aeson.Encode.Pretty-import Data.Bits-import Data.IORef-import Data.Maybe (isJust)-import Data.Semigroup-import Data.Time-import Data.Time.Clock.POSIX-import Data.Word-import Database.Bloodhound hiding (Status)-import Log-import Log.Internal.Logger-import Network.HTTP.Client-import Network.HTTP.Client.TLS (tlsManagerSettings)-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--import Log.Backend.ElasticSearch.Internal--------------------------------------------- | Create an 'elasticSearchLogger' for the duration of the given--- action, and shut it down afterwards, making sure that all buffered--- messages are actually written to the Elasticsearch store.-withElasticSearchLogger :: ElasticSearchConfig -> IO Word32 -> (Logger -> IO r)-                        -> IO r-withElasticSearchLogger conf randGen act = do-  logger <- elasticSearchLogger conf randGen-  withLogger logger act--{-# DEPRECATED elasticSearchLogger "Use 'withElasticSearchLogger' instead!" #-}---- | Start an asynchronous logger thread that stores messages using--- Elasticsearch.------ Please use 'withElasticSearchLogger' instead, which is more--- exception-safe (see the note attached to 'mkBulkLogger').-elasticSearchLogger ::-  ElasticSearchConfig -- ^ Configuration.-  -> IO Word32        -- ^ Generate a random 32-bit word for use in-                      -- document IDs.-  -> IO Logger-elasticSearchLogger ElasticSearchConfig{..} genRandomWord = do-  checkElasticSearchLogin-  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.-        indexExists' <- indexExists index-        unless indexExists' $ 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))-    (elasticSearchSync indexRef)-  where-    server  = Server esServer-    mapping = MappingName esMapping--    elasticSearchSync :: IORef IndexName -> IO ()-    elasticSearchSync indexRef = do-      indexName <- readIORef indexRef-      void . runBH_ $ refreshIndex indexName--    checkElasticSearchLogin :: IO ()-    checkElasticSearchLogin =-      when (isJust esLogin-            && not esLoginInsecure-            && not ("https:" `T.isPrefixOf` esServer)) $-        error $ "ElasticSearch: insecure login: "-          <> "Attempting to send login credentials over an insecure connection. "-          <> "Set esLoginInsecure = True to disable this check."--    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_ f = do-      mgr <- newManager tlsManagerSettings-      let hook = maybe return (uncurry basicAuthHook) esLogin-      let env = (mkBHEnv server mgr) { bhRequestHook = hook }-      runBH env f---    jsonToBSL :: Value -> BSL.ByteString-    jsonToBSL = encodePretty' defConfig { confIndent = Spaces 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 [-    "properties" .= object [-        "insertion_order" .= object [-            "type" .= ("integer"::T.Text)-          ]-        , "insertion_time" .= object [-            "type"   .= ("date"::T.Text)-          , "format" .= ("date_time"::T.Text)-          ]-        , "time" .= object [-            "type"   .= ("date"::T.Text)-          , "format" .= ("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)+module Log.Backend.ElasticSearch+    {-# DEPRECATED "Use directly Log.Backend.ElasticSearch.V1 or V5" #-}+    ( module Log.Backend.ElasticSearch.V1 ) where -decodeReply :: Reply -> Value-decodeReply reply = case eitherDecode' $ responseBody reply of-  Right body -> body-  Left  err  -> object ["decoding_error" .= err]+import Log.Backend.ElasticSearch.V1

@@ -1,31 +1,5 @@ module Log.Backend.ElasticSearch.Internal-  (ElasticSearchConfig(..)-  ,defaultElasticSearchConfig-  ,EsUsername(..)-  ,EsPassword(..))-where--import Database.Bloodhound hiding (Status)-import Prelude-import qualified Data.Text as T---- | Configuration for the Elasticsearch 'Logger'. See--- <https://www.elastic.co/guide/en/elasticsearch/reference/current/glossary.html>--- for the explanation of terms.-data ElasticSearchConfig = ElasticSearchConfig {-    esServer        :: !T.Text -- ^ Elasticsearch server address.-  , esIndex         :: !T.Text -- ^ Elasticsearch index name.-  , esMapping       :: !T.Text -- ^ Elasticsearch mapping name.-  , esLogin         :: Maybe (EsUsername, EsPassword) -- ^ Elasticsearch basic authentication username and password.-  , esLoginInsecure :: !Bool   -- ^ Allow basic authentication over non-TLS connections.-  } deriving (Eq, Show)+    {-# DEPRECATED "Use directly Log.Backend.ElasticSearch.V1 or V5" #-}+    ( module Log.Backend.ElasticSearch.V1.Internal ) where --- | Sensible defaults for 'ElasticSearchConfig'.-defaultElasticSearchConfig :: ElasticSearchConfig-defaultElasticSearchConfig = ElasticSearchConfig {-  esServer        = "http://localhost:9200",-  esIndex         = "logs",-  esMapping       = "log",-  esLogin         = Nothing,-  esLoginInsecure = False-  }+import Log.Backend.ElasticSearch.V1.Internal

@@ -1,40 +1,5 @@-{-# LANGUAGE RankNTypes #-}--- | Lensified version of "Log.Backend.ElasticSearch".-module Log.Backend.ElasticSearch.Lens (-    I.ElasticSearchConfig-  , esServer-  , esIndex-  , esMapping-  , esLogin-  , esLoginInsecure-  , I.defaultElasticSearchConfig-  , I.withElasticSearchLogger-  ) where--import Database.Bloodhound hiding (Status)-import Prelude-import qualified Data.Text as T-import qualified Log.Backend.ElasticSearch as I-import qualified Log.Backend.ElasticSearch.Internal ()--type Lens' s a = forall f. Functor f => (a -> f a) -> s -> f s---- | Elasticsearch server address.-esServer :: Lens' I.ElasticSearchConfig T.Text-esServer f esc = fmap (\x -> esc { I.esServer = x }) $ f (I.esServer esc)---- | Elasticsearch index name.-esIndex :: Lens' I.ElasticSearchConfig T.Text-esIndex f esc = fmap (\x -> esc { I.esIndex = x }) $ f (I.esIndex esc)---- | Elasticsearch mapping name.-esMapping :: Lens' I.ElasticSearchConfig T.Text-esMapping f esc = fmap (\x -> esc { I.esMapping = x }) $ f (I.esMapping esc)---- |  Elasticsearch basic authentication username and password.-esLogin :: Lens' I.ElasticSearchConfig (Maybe (EsUsername, EsPassword))-esLogin f esc = fmap (\x -> esc { I.esLogin = x }) $ f (I.esLogin esc)+module Log.Backend.ElasticSearch.Lens+    {-# DEPRECATED "Use directly Log.Backend.ElasticSearch.V1 or V5" #-}+    ( module Log.Backend.ElasticSearch.V1.Lens ) where --- | Allow basic authentication over non-TLS connections.-esLoginInsecure :: Lens' I.ElasticSearchConfig Bool-esLoginInsecure f esc = fmap (\x -> esc { I.esLoginInsecure = x }) $ f (I.esLoginInsecure esc)+import Log.Backend.ElasticSearch.V1.Lens

@@ -0,0 +1,253 @@+-- | Elasticsearch logging back-end.+module Log.Backend.ElasticSearch.V1 (+    ElasticSearchConfig+  , esServer+  , esIndex+  , esMapping+  , esLogin+  , esLoginInsecure+  , defaultElasticSearchConfig+  , withElasticSearchLogger+  , elasticSearchLogger+  ) where++import Control.Applicative+import Control.Arrow (second)+import Control.Concurrent+import Control.Exception+import Control.Monad+import Control.Monad.IO.Class+import Data.Aeson+import Data.Aeson.Encode.Pretty+import Data.Bits+import Data.IORef+import Data.Maybe (isJust)+import Data.Semigroup+import Data.Time+import Data.Time.Clock.POSIX+import Data.Word+import Database.V1.Bloodhound hiding (Status)+import Log+import Log.Internal.Logger+import Network.HTTP.Client+import Network.HTTP.Client.TLS (tlsManagerSettings)+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++import Log.Backend.ElasticSearch.V1.Internal++----------------------------------------+-- | Create an 'elasticSearchLogger' for the duration of the given+-- action, and shut it down afterwards, making sure that all buffered+-- messages are actually written to the Elasticsearch store.+withElasticSearchLogger :: ElasticSearchConfig -> IO Word32 -> (Logger -> IO r)+                        -> IO r+withElasticSearchLogger conf randGen act = do+  logger <- elasticSearchLogger conf randGen+  withLogger logger act++{-# DEPRECATED elasticSearchLogger "Use 'withElasticSearchLogger' instead!" #-}++-- | Start an asynchronous logger thread that stores messages using+-- Elasticsearch.+--+-- Please use 'withElasticSearchLogger' instead, which is more+-- exception-safe (see the note attached to 'mkBulkLogger').+elasticSearchLogger ::+  ElasticSearchConfig -- ^ Configuration.+  -> IO Word32        -- ^ Generate a random 32-bit word for use in+                      -- document IDs.+  -> IO Logger+elasticSearchLogger ElasticSearchConfig{..} genRandomWord = do+  checkElasticSearchLogin+  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.+        indexExists' <- indexExists index+        unless indexExists' $ 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))+    (elasticSearchSync indexRef)+  where+    server  = Server esServer+    mapping = MappingName esMapping++    elasticSearchSync :: IORef IndexName -> IO ()+    elasticSearchSync indexRef = do+      indexName <- readIORef indexRef+      void . runBH_ $ refreshIndex indexName++    checkElasticSearchLogin :: IO ()+    checkElasticSearchLogin =+      when (isJust esLogin+            && not esLoginInsecure+            && not ("https:" `T.isPrefixOf` esServer)) $+        error $ "ElasticSearch: insecure login: "+          <> "Attempting to send login credentials over an insecure connection. "+          <> "Set esLoginInsecure = True to disable this check."++    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_ f = do+      mgr <- newManager tlsManagerSettings+      let hook = maybe return (uncurry basicAuthHook) esLogin+      let env = (mkBHEnv server mgr) { bhRequestHook = hook }+      runBH env f+++    jsonToBSL :: Value -> BSL.ByteString+    jsonToBSL = encodePretty' defConfig { confIndent = Spaces 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 [+    "properties" .= object [+        "insertion_order" .= object [+            "type" .= ("integer"::T.Text)+          ]+        , "insertion_time" .= object [+            "type"   .= ("date"::T.Text)+          , "format" .= ("date_time"::T.Text)+          ]+        , "time" .= object [+            "type"   .= ("date"::T.Text)+          , "format" .= ("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]

@@ -0,0 +1,31 @@+module Log.Backend.ElasticSearch.V1.Internal+  (ElasticSearchConfig(..)+  ,defaultElasticSearchConfig+  ,EsUsername(..)+  ,EsPassword(..))+where++import Database.V1.Bloodhound hiding (Status)+import Prelude+import qualified Data.Text as T++-- | Configuration for the Elasticsearch 'Logger'. See+-- <https://www.elastic.co/guide/en/elasticsearch/reference/current/glossary.html>+-- for the explanation of terms.+data ElasticSearchConfig = ElasticSearchConfig {+    esServer        :: !T.Text -- ^ Elasticsearch server address.+  , esIndex         :: !T.Text -- ^ Elasticsearch index name.+  , esMapping       :: !T.Text -- ^ Elasticsearch mapping name.+  , esLogin         :: Maybe (EsUsername, EsPassword) -- ^ Elasticsearch basic authentication username and password.+  , esLoginInsecure :: !Bool   -- ^ Allow basic authentication over non-TLS connections.+  } deriving (Eq, Show)++-- | Sensible defaults for 'ElasticSearchConfig'.+defaultElasticSearchConfig :: ElasticSearchConfig+defaultElasticSearchConfig = ElasticSearchConfig {+  esServer        = "http://localhost:9200",+  esIndex         = "logs",+  esMapping       = "log",+  esLogin         = Nothing,+  esLoginInsecure = False+  }

@@ -0,0 +1,40 @@+{-# LANGUAGE RankNTypes #-}+-- | Lensified version of "Log.Backend.ElasticSearch".+module Log.Backend.ElasticSearch.V1.Lens (+    I.ElasticSearchConfig+  , esServer+  , esIndex+  , esMapping+  , esLogin+  , esLoginInsecure+  , I.defaultElasticSearchConfig+  , I.withElasticSearchLogger+  ) where++import Database.V1.Bloodhound hiding (Status)+import Prelude+import qualified Data.Text as T+import qualified Log.Backend.ElasticSearch.V1 as I+import qualified Log.Backend.ElasticSearch.V1.Internal ()++type Lens' s a = forall f. Functor f => (a -> f a) -> s -> f s++-- | Elasticsearch server address.+esServer :: Lens' I.ElasticSearchConfig T.Text+esServer f esc = fmap (\x -> esc { I.esServer = x }) $ f (I.esServer esc)++-- | Elasticsearch index name.+esIndex :: Lens' I.ElasticSearchConfig T.Text+esIndex f esc = fmap (\x -> esc { I.esIndex = x }) $ f (I.esIndex esc)++-- | Elasticsearch mapping name.+esMapping :: Lens' I.ElasticSearchConfig T.Text+esMapping f esc = fmap (\x -> esc { I.esMapping = x }) $ f (I.esMapping esc)++-- |  Elasticsearch basic authentication username and password.+esLogin :: Lens' I.ElasticSearchConfig (Maybe (EsUsername, EsPassword))+esLogin f esc = fmap (\x -> esc { I.esLogin = x }) $ f (I.esLogin esc)++-- | Allow basic authentication over non-TLS connections.+esLoginInsecure :: Lens' I.ElasticSearchConfig Bool+esLoginInsecure f esc = fmap (\x -> esc { I.esLoginInsecure = x }) $ f (I.esLoginInsecure esc)

@@ -0,0 +1,253 @@+-- | Elasticsearch logging back-end.+module Log.Backend.ElasticSearch.V5 (+    ElasticSearchConfig+  , esServer+  , esIndex+  , esMapping+  , esLogin+  , esLoginInsecure+  , defaultElasticSearchConfig+  , withElasticSearchLogger+  , elasticSearchLogger+  ) where++import Control.Applicative+import Control.Arrow (second)+import Control.Concurrent+import Control.Exception+import Control.Monad+import Control.Monad.IO.Class+import Data.Aeson+import Data.Aeson.Encode.Pretty+import Data.Bits+import Data.IORef+import Data.Maybe (isJust)+import Data.Semigroup+import Data.Time+import Data.Time.Clock.POSIX+import Data.Word+import Database.V5.Bloodhound hiding (Status)+import Log+import Log.Internal.Logger+import Network.HTTP.Client+import Network.HTTP.Client.TLS (tlsManagerSettings)+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++import Log.Backend.ElasticSearch.V5.Internal++----------------------------------------+-- | Create an 'elasticSearchLogger' for the duration of the given+-- action, and shut it down afterwards, making sure that all buffered+-- messages are actually written to the Elasticsearch store.+withElasticSearchLogger :: ElasticSearchConfig -> IO Word32 -> (Logger -> IO r)+                        -> IO r+withElasticSearchLogger conf randGen act = do+  logger <- elasticSearchLogger conf randGen+  withLogger logger act++{-# DEPRECATED elasticSearchLogger "Use 'withElasticSearchLogger' instead!" #-}++-- | Start an asynchronous logger thread that stores messages using+-- Elasticsearch.+--+-- Please use 'withElasticSearchLogger' instead, which is more+-- exception-safe (see the note attached to 'mkBulkLogger').+elasticSearchLogger ::+  ElasticSearchConfig -- ^ Configuration.+  -> IO Word32        -- ^ Generate a random 32-bit word for use in+                      -- document IDs.+  -> IO Logger+elasticSearchLogger ElasticSearchConfig{..} genRandomWord = do+  checkElasticSearchLogin+  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.+        indexExists' <- indexExists index+        unless indexExists' $ 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))+    (elasticSearchSync indexRef)+  where+    server  = Server esServer+    mapping = MappingName esMapping++    elasticSearchSync :: IORef IndexName -> IO ()+    elasticSearchSync indexRef = do+      indexName <- readIORef indexRef+      void . runBH_ $ refreshIndex indexName++    checkElasticSearchLogin :: IO ()+    checkElasticSearchLogin =+      when (isJust esLogin+            && not esLoginInsecure+            && not ("https:" `T.isPrefixOf` esServer)) $+        error $ "ElasticSearch: insecure login: "+          <> "Attempting to send login credentials over an insecure connection. "+          <> "Set esLoginInsecure = True to disable this check."++    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_ f = do+      mgr <- newManager tlsManagerSettings+      let hook = maybe return (uncurry basicAuthHook) esLogin+      let env = (mkBHEnv server mgr) { bhRequestHook = hook }+      runBH env f+++    jsonToBSL :: Value -> BSL.ByteString+    jsonToBSL = encodePretty' defConfig { confIndent = Spaces 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 [+    "properties" .= object [+        "insertion_order" .= object [+            "type" .= ("integer"::T.Text)+          ]+        , "insertion_time" .= object [+            "type"   .= ("date"::T.Text)+          , "format" .= ("date_time"::T.Text)+          ]+        , "time" .= object [+            "type"   .= ("date"::T.Text)+          , "format" .= ("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]

@@ -0,0 +1,31 @@+module Log.Backend.ElasticSearch.V5.Internal+  (ElasticSearchConfig(..)+  ,defaultElasticSearchConfig+  ,EsUsername(..)+  ,EsPassword(..))+where++import Database.V5.Bloodhound hiding (Status)+import Prelude+import qualified Data.Text as T++-- | Configuration for the Elasticsearch 'Logger'. See+-- <https://www.elastic.co/guide/en/elasticsearch/reference/current/glossary.html>+-- for the explanation of terms.+data ElasticSearchConfig = ElasticSearchConfig {+    esServer        :: !T.Text -- ^ Elasticsearch server address.+  , esIndex         :: !T.Text -- ^ Elasticsearch index name.+  , esMapping       :: !T.Text -- ^ Elasticsearch mapping name.+  , esLogin         :: Maybe (EsUsername, EsPassword) -- ^ Elasticsearch basic authentication username and password.+  , esLoginInsecure :: !Bool   -- ^ Allow basic authentication over non-TLS connections.+  } deriving (Eq, Show)++-- | Sensible defaults for 'ElasticSearchConfig'.+defaultElasticSearchConfig :: ElasticSearchConfig+defaultElasticSearchConfig = ElasticSearchConfig {+  esServer        = "http://localhost:9200",+  esIndex         = "logs",+  esMapping       = "log",+  esLogin         = Nothing,+  esLoginInsecure = False+  }

@@ -0,0 +1,40 @@+{-# LANGUAGE RankNTypes #-}+-- | Lensified version of "Log.Backend.ElasticSearch".+module Log.Backend.ElasticSearch.V5.Lens (+    I.ElasticSearchConfig+  , esServer+  , esIndex+  , esMapping+  , esLogin+  , esLoginInsecure+  , I.defaultElasticSearchConfig+  , I.withElasticSearchLogger+  ) where++import Database.V5.Bloodhound hiding (Status)+import Prelude+import qualified Data.Text as T+import qualified Log.Backend.ElasticSearch.V5 as I+import qualified Log.Backend.ElasticSearch.V5.Internal ()++type Lens' s a = forall f. Functor f => (a -> f a) -> s -> f s++-- | Elasticsearch server address.+esServer :: Lens' I.ElasticSearchConfig T.Text+esServer f esc = fmap (\x -> esc { I.esServer = x }) $ f (I.esServer esc)++-- | Elasticsearch index name.+esIndex :: Lens' I.ElasticSearchConfig T.Text+esIndex f esc = fmap (\x -> esc { I.esIndex = x }) $ f (I.esIndex esc)++-- | Elasticsearch mapping name.+esMapping :: Lens' I.ElasticSearchConfig T.Text+esMapping f esc = fmap (\x -> esc { I.esMapping = x }) $ f (I.esMapping esc)++-- |  Elasticsearch basic authentication username and password.+esLogin :: Lens' I.ElasticSearchConfig (Maybe (EsUsername, EsPassword))+esLogin f esc = fmap (\x -> esc { I.esLogin = x }) $ f (I.esLogin esc)++-- | Allow basic authentication over non-TLS connections.+esLoginInsecure :: Lens' I.ElasticSearchConfig Bool+esLoginInsecure f esc = fmap (\x -> esc { I.esLoginInsecure = x }) $ f (I.esLoginInsecure esc)