diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,30 @@ +Copyright (c) 2015, Scrive AB + +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are met: + + * Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + + * Redistributions in binary form must reproduce the above + copyright notice, this list of conditions and the following + disclaimer in the documentation and/or other materials provided + with the distribution. + + * Neither the name of Andrzej Rybczak nor the names of other + contributors may be used to endorse or promote products derived + from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/log-elasticsearch.cabal b/log-elasticsearch.cabal new file mode 100644 --- /dev/null +++ b/log-elasticsearch.cabal @@ -0,0 +1,61 @@ +name: log-elasticsearch +version: 0.7 +synopsis: Structured logging solution (Elasticsearch back end) + +description: Elasticsearch back end for the 'log' library. + Depends on 'log-base'. + +homepage: https://github.com/scrive/log +license: BSD3 +license-file: LICENSE +author: Scrive AB +maintainer: Andrzej Rybczak , + Jonathan Jouty , + Mikhail Glushenkov , + Oleg Grenrus +copyright: Scrive AB +category: System +build-type: Simple +cabal-version: >=1.10 +tested-with: GHC == 7.8.4, GHC == 7.10.3, GHC == 8.0.1 + +Source-repository head + Type: git + Location: https://github.com/scrive/log.git + +library + exposed-modules: Log.Backend.ElasticSearch + 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, + deepseq, + http-client, + semigroups, + text, + text-show >= 2, + time >= 1.5, + transformers, + unordered-containers, + vector + hs-source-dirs: src + + ghc-options: -O2 -Wall -funbox-strict-fields + + default-language: Haskell2010 + default-extensions: BangPatterns + , FlexibleContexts + , FlexibleInstances + , GeneralizedNewtypeDeriving + , LambdaCase + , MultiParamTypeClasses + , NoImplicitPrelude + , OverloadedStrings + , RankNTypes + , RecordWildCards + , ScopedTypeVariables + , TypeFamilies + , UndecidableInstances diff --git a/src/Log/Backend/ElasticSearch.hs b/src/Log/Backend/ElasticSearch.hs new file mode 100644 --- /dev/null +++ b/src/Log/Backend/ElasticSearch.hs @@ -0,0 +1,248 @@ +-- | Elasticsearch logging back-end. +module Log.Backend.ElasticSearch ( + ElasticSearchConfig(..) + , 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.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 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 + +-- | 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. + } deriving (Eq, Show) + +-- | Sensible defaults for 'ElasticSearchConfig'. +defaultElasticSearchConfig :: ElasticSearchConfig +defaultElasticSearchConfig = ElasticSearchConfig { + esServer = "http://localhost:9200", + esIndex = "logs", + esMapping = "log" + } + + +---------------------------------------- +-- | 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 + 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 + + 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 = 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]