diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,18 +1,67 @@ +# log-elasticsearch-0.13.0.2 (2025-06-26) +* Switch back to `http-client-tls` by default since `HsOpenSSL` is based on a + deprecated `openssl` API (soon to be removed). + +# log-elasticsearch-0.13.0.1 (2023-01-31) +* Add support for OpenSearch. + +# log-elasticsearch-0.13.0.0 (2022-09-21) +* Remove deprecated `elasticSearchLogger`. +* Use `http-client-openssl` by default instead of `http-client-tls`. +* Generalize logger related functions to `MonadUnliftIO`. + +# log-elasticsearch-0.12.2.0 (2022-04-04) +* Add support for aeson 2.0.1.0. +* Add support for GHC 9.2. +* Drop support for GHC 8.0 and 8.2. + +# log-elasticsearch-0.12.1.1 (2021-10-11) +* Make compatible with log-base-0.11.0.0. + +# log-elasticsearch-0.12.1.0 (2021-06-23) +* Map `time` as `date_nanos` in ElasticSearch >= 7.0. + +# log-elasticsearch-0.12.0.0 (2021-06-09) +* Stop putting `insertion_time` and `insertion_order` in ElasticSearch. + +# log-elasticsearch-0.11.0.0 (2020-08-24) +* Drop dependency on bloodhound +* Unify V1 and V5 specific modules +* Add support for ElasticSearch 7.x +* Try harder to insert problematic messages into ElasticSearch +* Support building with GHC 8.8.4 and GHC 8.10.2 + +# log-elasticsearch-0.10.2.0 (2020-02-03) +* Modify data keys in deterministic manner in case of ES insertion failure + +# log-elasticsearch-0.10.1.0 (2019-05-15) +* Add a Generic instance for ElasticSearchConfig. + +# log-elasticsearch-0.10.0.1 (2019-01-24) +* Compatibility with ES 6.x by using "text" instead of deprecated "string" + +# log-elasticsearch-0.10.0.0 (2018-03-28) +* Expose `checkElasticSearchConnection` and `checkElasticSearchLogin`. +* Add config parameters for number of shards and replicas to + `ElasticSearchConfig` (and make default of 4 shards and 1 replica + explicit in 'defaultElasticSearchConfig'). + # log-elasticsearch-0.9.1.0 (2017-08-10) -* Add 'toEncoding' to 'ToJSON LogsMapping' instance +* Add `toEncoding` to `ToJSON LogsMapping` instance # log-elasticsearch-0.9.0.1 (2017-06-19) -* 'withElasticSearchLogger' no longer fails when the Elasticsearch server is down. +* `withElasticSearchLogger` no longer fails when the Elasticsearch + server is down. # log-elasticsearch-0.9.0.0 (2017-05-04) * Now works with bloodhound-0.14.0.0 (#30). # log-elasticsearch-0.8.1 (2017-03-27) -* Log.Backend.ElasticSearch.Internal now exports 'EsUsername' and - 'EsPassword'. +* `Log.Backend.ElasticSearch.Internal` now exports `EsUsername` and + `EsPassword`. # log-elasticsearch-0.8 (2017-03-16) -* Made ElasticSearchConfig an abstract type (#27). +* Made `ElasticSearchConfig` an abstract type (#27). * Added support for HTTPS and basic auth (#26). # log-elasticsearch-0.7 (2016-11-25) diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -1,3 +1,40 @@ -# log-elasticsearch [![Hackage version](https://img.shields.io/hackage/v/log-elasticsearch.svg?label=Hackage)](https://hackage.haskell.org/package/log-elasticsearch) [![Build Status](https://secure.travis-ci.org/scrive/log.svg?branch=master)](http://travis-ci.org/scrive/log) +# log -Elasticsearch back end for the `log` library suite. +[![Hackage version](https://img.shields.io/hackage/v/log-base.svg?label=Hackage)](https://hackage.haskell.org/package/log-base) +[![Build Status](https://github.com/scrive/log/workflows/Haskell-CI/badge.svg?branch=master)](https://github.com/scrive/log/actions?query=branch%3Amaster) + +A set of libraries that provide a way to record structured log messages with +multiple backends. + +Supported backends: + +* Standard output via + [`log-base`](https://hackage.haskell.org/package/log-base). +* Elasticsearch via + [`log-elasticsearch`](https://hackage.haskell.org/package/log-elasticsearch). +* PostgreSQL via + [`log-postgres`](https://hackage.haskell.org/package/log-postgres). + +## Example + +A sample usage for logging to both standard output and Elasticsearch: + +```haskell +{-# LANGUAGE OverloadedStrings #-} +module Main where + +import Log +import Log.Backend.ElasticSearch +import Log.Backend.StandardOutput + +main :: IO () +main = do + let config = defaultElasticSearchConfig + { esServer = "http://localhost:9200" + , esIndex = "logs" + } + withStdOutLogger $ \stdoutLogger -> do + withElasticSearchLogger config $ \esLogger -> do + runLogT "main" (stdoutLogger <> esLogger) defaultLogLevel $ do + logInfo_ "Hi there" +``` diff --git a/Setup.hs b/Setup.hs deleted file mode 100644 --- a/Setup.hs +++ /dev/null @@ -1,2 +0,0 @@ -import Distribution.Simple -main = defaultMain diff --git a/log-elasticsearch.cabal b/log-elasticsearch.cabal --- a/log-elasticsearch.cabal +++ b/log-elasticsearch.cabal @@ -1,12 +1,13 @@ +cabal-version: 3.0 name: log-elasticsearch -version: 0.9.1.0 +version: 0.13.0.2 synopsis: Structured logging solution (Elasticsearch back end) description: Elasticsearch back end for the 'log' library suite. Depends on 'log-base'. homepage: https://github.com/scrive/log -license: BSD3 +license: BSD-3-Clause license-file: LICENSE author: Scrive AB maintainer: Andrzej Rybczak , @@ -16,44 +17,42 @@ copyright: Scrive AB category: System build-type: Simple -cabal-version: >=1.10 extra-source-files: CHANGELOG.md, README.md -tested-with: GHC == 7.10.3, GHC == 8.0.2, GHC == 8.2.1 +tested-with: GHC == { 8.10.7, 9.0.2, 9.2.8, 9.4.8, 9.6.7, 9.8.4, 9.10.2, 9.12.2 } -Source-repository head - Type: git - Location: https://github.com/scrive/log.git +source-repository head + type: git + location: https://github.com/scrive/log.git +flag openssl + description: Use http-client-openssl instead of http-client-tls + default: False + library - 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 + exposed-modules: Log.Backend.ElasticSearch Log.Backend.ElasticSearch.Internal - build-depends: base <5, - log-base >= 0.7, - aeson >=1.0.0.0, + Log.Backend.ElasticSearch.Lens + build-depends: base >= 4.13 && <5, + aeson >= 1.0, aeson-pretty >=0.8.2, - bytestring, base64-bytestring, - bloodhound >= 0.13 && < 0.15, + bytestring, deepseq, http-client, - http-client-tls, + http-types, + log-base >= 0.10 && <0.13, + network-uri, semigroups, text, text-show >= 2, time >= 1.5, transformers, + unliftio-core >= 0.2, unordered-containers, vector hs-source-dirs: src - ghc-options: -O2 -Wall -funbox-strict-fields + ghc-options: -Wall default-language: Haskell2010 default-extensions: BangPatterns @@ -62,10 +61,16 @@ , GeneralizedNewtypeDeriving , LambdaCase , MultiParamTypeClasses - , NoImplicitPrelude , OverloadedStrings , RankNTypes , RecordWildCards , ScopedTypeVariables + , TupleSections , TypeFamilies , UndecidableInstances + + if flag(openssl) + cpp-options: -DOPENSSL + build-depends: http-client-openssl >= 0.3.2 + else + build-depends: http-client-tls >= 0.3.6.3 diff --git a/src/Log/Backend/ElasticSearch.hs b/src/Log/Backend/ElasticSearch.hs --- a/src/Log/Backend/ElasticSearch.hs +++ b/src/Log/Backend/ElasticSearch.hs @@ -1,5 +1,220 @@ +-- | Elasticsearch logging back-end. module Log.Backend.ElasticSearch - {-# DEPRECATED "Use directly Log.Backend.ElasticSearch.V1 or V5" #-} - ( module Log.Backend.ElasticSearch.V1 ) where + ( ElasticSearchConfig + , esServer + , esIndex + , esShardCount + , esReplicaCount + , esMapping + , esLogin + , esLoginInsecure + , checkElasticSearchLogin + , checkElasticSearchConnection + , defaultElasticSearchConfig + , withElasticSearchLogger + ) where -import Log.Backend.ElasticSearch.V1 +import Control.Applicative +import Control.Concurrent +import Control.Exception +import Control.Monad +import Control.Monad.IO.Unlift +import Data.Aeson +import Data.Aeson.Encode.Pretty +import Data.IORef +import Data.Maybe +import Data.Time +import Log +import Log.Internal.Logger +import Network.HTTP.Client +import qualified Data.Text as T +import qualified Data.Text.IO as T +import qualified Data.Text.Lazy as TL +import qualified Data.Text.Lazy.Builder as T +import qualified Data.Traversable as F +import qualified Data.Vector as V + +import Log.Backend.ElasticSearch.Internal +import qualified Log.Internal.Aeson.Compat as AC + +---------------------------------------- +-- | 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 :: MonadUnliftIO m => ElasticSearchConfig -> (Logger -> m r) -> m r +withElasticSearchLogger conf act = withRunInIO $ \unlift -> do + logger <- elasticSearchLogger conf + withLogger logger (unlift . act) + +-- | 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 Logger +elasticSearchLogger esConf@ElasticSearchConfig{..} = do + checkElasticSearchLogin esConf + env <- mkEsEnv esConf + versionRef <- newIORef Nothing + indexRef <- newIORef T.empty + mkBulkLogger "ElasticSearch" (\msgs -> do + now <- getCurrentTime + oldIndex <- readIORef indexRef + retryOnException versionRef $ do + -- We need to consider version of ES because ES >= 5.0.0 and ES >= 7.0.0 + -- have slight differences in parts of API used for logging. + version <- readIORef versionRef >>= \case + Just version -> pure version + Nothing -> serverInfo env >>= \case + Left (ex :: HttpException) -> error + $ "elasticSearchLogger: unexpected error: " + <> show ex + <> " (is ElasticSearch server running?)" + Right reply -> case parseEsVersion $ responseBody reply of + Nothing -> error + $ "elasticSearchLogger: invalid response when parsing version number: " + <> show reply + Just version -> pure version + -- 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 = T.concat + [ esIndex + , "-" + , T.pack $ formatTime defaultTimeLocale "%F" now + ] + when (oldIndex /= index) $ do + -- There is an obvious race condition in presence of more than one + -- logger instance running, but it's irrelevant as attempting to create + -- index that already exists is harmless. + ixExists <- indexExists env index + unless ixExists $ do + reply <- createIndexWithMapping version env esConf index + unless (isSuccess reply) $ do + printEsError "error while creating index" $ responseBody reply + writeIORef indexRef index + let jsonMsgs = V.fromList $ map toJsonMsg msgs + reply <- responseBody <$> bulkIndex version env esConf index jsonMsgs + -- Try to parse parts of reply to get information about log messages that + -- failed to be inserted for some reason. + case checkForBulkErrors jsonMsgs reply of + Nothing -> printEsError "unexpected response" reply + Just (hasErrors, responses) -> when hasErrors $ do + -- If any message failed to be inserted because of type mismatch, go + -- back to them, log the insertion failure and add type suffix to each + -- of the keys in their "data" fields to work around type errors. + let newMsgs = + let modifyData :: Maybe Value -> Value -> Value + modifyData merr (Object hm) = Object $ + let newData = AC.foldrWithKey keyAddValueTypeSuffix AC.empty hm + in case merr of + -- We have the error message, i.e. we're at the top + -- level object, so add it to the data. + Just err -> newData `AC.union` AC.singleton "__es_error" err + Nothing -> newData + modifyData _ v = v + + keyAddValueTypeSuffix k v acc = AC.insert + (case v of + Object{} -> k <> "_object" + Array{} -> k <> "_array" + String{} -> k <> "_string" + Number{} -> k <> "_number" + Bool{} -> k <> "_bool" + Null{} -> k <> "_null" + ) (modifyData Nothing v) acc + in adjustFailedMessagesWith modifyData jsonMsgs responses + -- Attempt to put modified messages. + newReply <- responseBody <$> bulkIndex version env esConf index newMsgs + case checkForBulkErrors newMsgs newReply of + Nothing -> printEsError "unexpected response" newReply + Just (newHasErrors, newResponses) -> when newHasErrors $ do + -- If some of the messages failed again (it might happen e.g. if + -- data contains an array with elements of different types), drop + -- their data field. + let newerMsgs = + let modifyData :: Maybe Value -> Value -> Value + modifyData (Just err) Object{} = object [ "__es_error" .= err ] + modifyData _ v = v + in adjustFailedMessagesWith modifyData newMsgs newResponses + -- Ignore any further errors. + void $ bulkIndex version env esConf index newerMsgs) + (refreshIndex env =<< readIORef indexRef) + where + -- Process reply of bulk indexing to get responses for each index operation + -- and check whether any insertion failed. + checkForBulkErrors + :: V.Vector a + -> Value + -> Maybe (Bool, V.Vector Object) + checkForBulkErrors jsonMsgs replyBody = do + Object response <- pure replyBody + Bool hasErrors <- "errors" `AC.lookup` response + Array jsonItems <- "items" `AC.lookup` response + items <- F.forM jsonItems $ \v -> do + Object item <- pure v + Object index_ <- "index" `AC.lookup` item + -- ES <= 2.x returns 'create' for some reason, so consider both. + <|> "create" `AC.lookup` item + pure index_ + guard $ V.length items == V.length jsonMsgs + pure (hasErrors, items) + + adjustFailedMessagesWith + :: (Maybe err -> obj -> obj) + -> V.Vector (AC.KeyMap obj) + -> V.Vector (AC.KeyMap err) + -> V.Vector (AC.KeyMap obj) + adjustFailedMessagesWith f jsonMsgs responses = + let failed = V.imapMaybe (\n item -> (n, ) <$> "error" `AC.lookup` item) responses + adjust act key m = case AC.lookup key m of + Nothing -> m + Just v -> AC.insert key (act v) m + in (`V.map` failed) $ \(n, err) -> adjust (f $ Just err) "data" $ jsonMsgs V.! n + + printEsError msg body = + T.putStrLn $ "elasticSearchLogger: " <> msg <> " " <> prettyJson body + + retryOnException :: forall r. IORef (Maybe EsVersion) -> IO r -> IO r + retryOnException versionRef m = try m >>= \case + Left (ex::SomeException) -> do + putStrLn $ "ElasticSearch: unexpected error: " + <> show ex <> ", retrying in 10 seconds" + -- If there was an exception, ElasticSearch version might've changed, so + -- reset it. + writeIORef versionRef Nothing + threadDelay $ 10 * 1000000 + retryOnException versionRef m + Right result -> return result + + prettyJson :: Value -> T.Text + prettyJson = TL.toStrict + . T.toLazyText + . encodePrettyToTextBuilder' defConfig { confIndent = Spaces 2 } + + toJsonMsg :: LogMessage -> Object + toJsonMsg msg = case toJSON msg of + Object jMsg -> jMsg + value -> error $ "unexpected non-object: " ++ show value + +---------------------------------------- + +-- | Check that login credentials are specified properly. +-- +-- @since 0.10.0.0 +checkElasticSearchLogin :: MonadIO m => ElasticSearchConfig -> m () +checkElasticSearchLogin ElasticSearchConfig{..} = liftIO $ do + 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." + +-- | Check that we can connect to the ES server. +-- +-- @since 0.10.0.0 +checkElasticSearchConnection :: MonadIO m => ElasticSearchConfig -> m (Either HttpException ()) +checkElasticSearchConnection esConf = liftIO $ do + fmap (const ()) <$> (serverInfo =<< mkEsEnv esConf) diff --git a/src/Log/Backend/ElasticSearch/Internal.hs b/src/Log/Backend/ElasticSearch/Internal.hs --- a/src/Log/Backend/ElasticSearch/Internal.hs +++ b/src/Log/Backend/ElasticSearch/Internal.hs @@ -1,5 +1,249 @@ +{-# LANGUAGE CPP #-} +{-# LANGUAGE DeriveGeneric #-} module Log.Backend.ElasticSearch.Internal - {-# DEPRECATED "Use directly Log.Backend.ElasticSearch.V1 or V5" #-} - ( module Log.Backend.ElasticSearch.V1.Internal ) where + ( ElasticSearchConfig(..) + , defaultElasticSearchConfig + -- * ES version + , EsVersion(..) + , parseEsVersion + , esV5, esV7 + -- * ES commands + , serverInfo + , indexExists + , createIndexWithMapping + , bulkIndex + , refreshIndex + -- * ES communication details + , EsEnv(..) + , mkEsEnv + , dispatch + , decodeReply + , isSuccess + ) where -import Log.Backend.ElasticSearch.V1.Internal +import Control.Exception +import Control.Monad +import Data.Aeson +import Data.Ix (inRange) +import Data.Maybe +import GHC.Generics (Generic) +import Network.HTTP.Client +import Network.HTTP.Types +#if OPENSSL +import Network.HTTP.Client.OpenSSL (newOpenSSLManager, withOpenSSL) +#else +import Network.HTTP.Client.TLS (tlsManagerSettings) +#endif +import qualified Data.ByteString.Builder as BSB +import qualified Data.ByteString.Lazy.Char8 as BSL +import qualified Data.Text as T +import qualified Data.Text.Encoding as T +import qualified Data.Vector as V + +import qualified Log.Internal.Aeson.Compat as AC + +-- | Configuration for the Elasticsearch 'Logger'. See +-- +-- for the explanation of terms. +data ElasticSearchConfig = ElasticSearchConfig + { esServer :: !T.Text -- ^ Elasticsearch server address. + , esIndex :: !T.Text -- ^ Elasticsearch index name. + , esShardCount :: !Int + -- ^ Elasticsearch shard count for the named index. + -- + -- @since 0.10.0.0 + , esReplicaCount :: !Int + -- ^ Elasticsearch replica count for the named index. + -- + -- @since 0.10.0.0 + , esMapping :: !T.Text + -- ^ Elasticsearch mapping name (unused with ES >= 7.0.0) + , esLogin :: Maybe (T.Text, T.Text) + -- ^ Elasticsearch basic authentication username and password. + , esLoginInsecure :: !Bool + -- ^ Allow basic authentication over non-TLS connections. + } deriving (Eq, Show, Generic) + +-- | Sensible defaults for 'ElasticSearchConfig'. +defaultElasticSearchConfig :: ElasticSearchConfig +defaultElasticSearchConfig = ElasticSearchConfig + { esServer = "http://localhost:9200" + , esIndex = "logs" + , esShardCount = 4 + , esReplicaCount = 1 + , esMapping = "log" + , esLogin = Nothing + , esLoginInsecure = False + } + +---------------------------------------- +-- ES communication + +-- Most of the below code is taken from the bloodhound library +-- (https://github.com/bitemyapp/bloodhound). + +data EsVersion = EsVersion !Int !Int !Int + deriving (Eq, Ord) + +parseEsVersion :: Value -> Maybe EsVersion +parseEsVersion js = do + Object props <- pure js + Object version <- "version" `AC.lookup` props + case "distribution" `AC.lookup` version of + Just "opensearch" -> do + -- OpenSearch is compatible (so far) with esV7 mappings. + pure esV7 + _ -> do + String number <- "number" `AC.lookup` version + [v1, v2, v3] <- mapM (maybeRead . T.unpack) $ T.splitOn "." number + pure $ EsVersion v1 v2 v3 + where + maybeRead s = do + [(v, "")] <- pure $ reads s + pure v + +-- | Minimum version with split 'string' type. +esV5 :: EsVersion +esV5 = EsVersion 5 0 0 + +-- | Minimum version without mapping types. +esV7 :: EsVersion +esV7 = EsVersion 7 0 0 + +---------------------------------------- + +-- | Check the ElasticSearch server for info. Result can be fed to +-- 'parseEsVersion' to determine version of the server. +serverInfo :: EsEnv -> IO (Either HttpException (Response Value)) +serverInfo env = try $ dispatch env methodGet [] Nothing + +-- | Check that given index exists. +indexExists :: EsEnv -> T.Text -> IO Bool +indexExists env index = + isSuccess <$> dispatch env methodHead [index] Nothing + +-- | Create an index with given mapping. +createIndexWithMapping + :: EsVersion + -> EsEnv + -> ElasticSearchConfig + -> T.Text + -> IO (Response Value) +createIndexWithMapping version env ElasticSearchConfig{..} index = do + dispatch env methodPut [index] . Just . encode $ object + [ "settings" .= object + [ "number_of_shards" .= esShardCount + , "number_of_replicas" .= esReplicaCount + ] + , "mappings" .= if version >= esV7 + then logsMapping + else object [ AC.fromText esMapping .= logsMapping ] + ] + where + logsMapping = object + [ "properties" .= object + [ "time" .= object + [ "type" .= timeTy + , "format" .= ("date_time"::T.Text) + ] + , "domain" .= object + [ "type" .= textTy + ] + , "level" .= object + [ "type" .= textTy + ] + , "component" .= object + [ "type" .= textTy + ] + , "message" .= object + [ "type" .= textTy + ] + ] + ] + where + timeTy :: T.Text + timeTy = if version >= esV7 + then "date_nanos" + else "date" + + textTy :: T.Text + textTy = if version >= esV5 + then "text" + else "string" + +-- Index multiple log messages. +bulkIndex + :: EsVersion + -> EsEnv + -> ElasticSearchConfig + -> T.Text + -> V.Vector Object + -> IO (Response Value) +bulkIndex version env conf index objs = do + dispatch env methodPost route . Just . BSB.toLazyByteString $ foldMap ixOp objs + where + route = if version >= esV7 + then [index, "_bulk"] + else [index, esMapping conf, "_bulk"] + + ixOp obj = ixCmd + <> BSB.char8 '\n' + <> BSB.lazyByteString (encode $ Object obj) + <> BSB.char8 '\n' + where + ixCmd = BSB.lazyByteString . encode $ object + [ "index" .= object [] + ] + +-- Refresh given index. +refreshIndex :: EsEnv -> T.Text -> IO () +refreshIndex env index = + void $ dispatch env methodPost [index, "_refresh"] Nothing + +---------------------------------------- + +data EsEnv = EsEnv + { envServer :: !T.Text + , envManager :: !Manager + , envRequestHook :: !(Request -> Request) + } + +mkEsEnv :: ElasticSearchConfig -> IO EsEnv +mkEsEnv ElasticSearchConfig{..} = do +#if OPENSSL + envManager <- withOpenSSL newOpenSSLManager +#else + envManager <- newManager tlsManagerSettings +#endif + let envServer = esServer + envRequestHook = maybe id mkAuthHook esLogin + pure EsEnv{..} + where + mkAuthHook (u, p) = applyBasicAuth (T.encodeUtf8 u) (T.encodeUtf8 p) + +---------------------------------------- + +dispatch :: EsEnv + -> Method + -> [T.Text] + -> Maybe BSL.ByteString + -> IO (Response Value) +dispatch EsEnv{..} dMethod url body = do + initReq <- parseRequest $ T.unpack $ T.intercalate "/" $ envServer : url + let req = envRequestHook . setRequestIgnoreStatus $ initReq + { method = dMethod + , requestBody = RequestBodyLBS $ fromMaybe BSL.empty body + , requestHeaders = ("Content-Type", "application/json") : requestHeaders initReq + } + fmap decodeReply <$> httpLbs req envManager + +decodeReply :: BSL.ByteString -> Value +decodeReply bs = case eitherDecode' bs of + Right js -> js + Left err -> object ["decoding_error" .= err] + +isSuccess :: Response a -> Bool +isSuccess = statusCheck (inRange (200, 299)) + where + statusCheck :: (Int -> Bool) -> Response a -> Bool + statusCheck p = p . statusCode . responseStatus diff --git a/src/Log/Backend/ElasticSearch/Lens.hs b/src/Log/Backend/ElasticSearch/Lens.hs --- a/src/Log/Backend/ElasticSearch/Lens.hs +++ b/src/Log/Backend/ElasticSearch/Lens.hs @@ -1,5 +1,58 @@ +{-# LANGUAGE RankNTypes #-} +-- | Lensified version of "Log.Backend.ElasticSearch". module Log.Backend.ElasticSearch.Lens - {-# DEPRECATED "Use directly Log.Backend.ElasticSearch.V1 or V5" #-} - ( module Log.Backend.ElasticSearch.V1.Lens ) where + ( I.ElasticSearchConfig + , esServer + , esIndex + , esShardCount + , esReplicaCount + , esMapping + , esLogin + , esLoginInsecure + , I.defaultElasticSearchConfig + , I.withElasticSearchLogger + ) where -import Log.Backend.ElasticSearch.V1.Lens +import qualified Data.Text as T +import qualified Log.Backend.ElasticSearch as I + +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 = (\x -> esc { I.esServer = x }) <$> f (I.esServer esc) +{-# INLINE esServer #-} + +-- | Elasticsearch index name. +esIndex :: Lens' I.ElasticSearchConfig T.Text +esIndex f esc = (\x -> esc { I.esIndex = x }) <$> f (I.esIndex esc) +{-# INLINE esIndex #-} + +-- | Elasticsearch shard count for the named index. +-- +-- @since 0.10.0.0 +esShardCount :: Lens' I.ElasticSearchConfig Int +esShardCount f esc = (\x -> esc { I.esShardCount = x }) <$> f (I.esShardCount esc) +{-# INLINE esShardCount #-} + +-- | Elasticsearch replica count for the named index. +-- +-- @since 0.10.0.0 +esReplicaCount :: Lens' I.ElasticSearchConfig Int +esReplicaCount f esc = (\x -> esc { I.esReplicaCount = x }) <$> f (I.esReplicaCount esc) +{-# INLINE esReplicaCount #-} + +-- | Elasticsearch mapping name. +esMapping :: Lens' I.ElasticSearchConfig T.Text +esMapping f esc = (\x -> esc { I.esMapping = x }) <$> f (I.esMapping esc) +{-# INLINE esMapping #-} + +-- | Elasticsearch basic authentication username and password. +esLogin :: Lens' I.ElasticSearchConfig (Maybe (T.Text, T.Text)) +esLogin f esc = (\x -> esc { I.esLogin = x }) <$> f (I.esLogin esc) +{-# INLINE esLogin #-} + +-- | Allow basic authentication over non-TLS connections. +esLoginInsecure :: Lens' I.ElasticSearchConfig Bool +esLoginInsecure f esc = (\x -> esc { I.esLoginInsecure = x }) <$> f (I.esLoginInsecure esc) +{-# INLINE esLoginInsecure #-} diff --git a/src/Log/Backend/ElasticSearch/V1.hs b/src/Log/Backend/ElasticSearch/V1.hs deleted file mode 100644 --- a/src/Log/Backend/ElasticSearch/V1.hs +++ /dev/null @@ -1,283 +0,0 @@ --- | 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 System.IO -import TextShow -import qualified Data.Aeson.Encoding as Aeson -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) -> - hPutStrLn stderr $ "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) - ] - ] - ] - - toEncoding LogsMapping = Aeson.pairs $ mconcat - [ Aeson.pair "properties" $ Aeson.pairs $ mconcat - [ Aeson.pair "insertion_order" $ Aeson.pairs $ mconcat - [ "type" .= ("integer"::T.Text) - ] - , Aeson.pair "insertion_time" $ Aeson.pairs $ mconcat - [ "type" .= ("date"::T.Text) - , "format" .= ("date_time"::T.Text) - ] - , Aeson.pair "time" $ Aeson.pairs $ mconcat - [ "type" .= ("date"::T.Text) - , "format" .= ("date_time"::T.Text) - ] - , Aeson.pair "domain" $ Aeson.pairs $ mconcat - [ "type" .= ("string"::T.Text) - ] - , Aeson.pair "level" $ Aeson.pairs $ mconcat - [ "type" .= ("string"::T.Text) - ] - , Aeson.pair "component" $ Aeson.pairs $ mconcat - [ "type" .= ("string"::T.Text) - ] - , Aeson.pair "message" $ Aeson.pairs $ mconcat - [ "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] diff --git a/src/Log/Backend/ElasticSearch/V1/Internal.hs b/src/Log/Backend/ElasticSearch/V1/Internal.hs deleted file mode 100644 --- a/src/Log/Backend/ElasticSearch/V1/Internal.hs +++ /dev/null @@ -1,31 +0,0 @@ -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 --- --- 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 - } diff --git a/src/Log/Backend/ElasticSearch/V1/Lens.hs b/src/Log/Backend/ElasticSearch/V1/Lens.hs deleted file mode 100644 --- a/src/Log/Backend/ElasticSearch/V1/Lens.hs +++ /dev/null @@ -1,40 +0,0 @@ -{-# 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) diff --git a/src/Log/Backend/ElasticSearch/V5.hs b/src/Log/Backend/ElasticSearch/V5.hs deleted file mode 100644 --- a/src/Log/Backend/ElasticSearch/V5.hs +++ /dev/null @@ -1,254 +0,0 @@ --- | 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 System.IO -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) -> - hPutStrLn stderr $ "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] diff --git a/src/Log/Backend/ElasticSearch/V5/Internal.hs b/src/Log/Backend/ElasticSearch/V5/Internal.hs deleted file mode 100644 --- a/src/Log/Backend/ElasticSearch/V5/Internal.hs +++ /dev/null @@ -1,31 +0,0 @@ -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 --- --- 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 - } diff --git a/src/Log/Backend/ElasticSearch/V5/Lens.hs b/src/Log/Backend/ElasticSearch/V5/Lens.hs deleted file mode 100644 --- a/src/Log/Backend/ElasticSearch/V5/Lens.hs +++ /dev/null @@ -1,40 +0,0 @@ -{-# 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)