logstash (empty) → 0.1.0.0

raw patch · 7 files changed

+687/−0 lines, 7 filesdep +aesondep +asyncdep +base

Dependencies added: aeson, async, base, bytestring, data-default-class, exceptions, monad-control, mtl, network, resource-pool, resourcet, retry, stm, stm-chans, time, tls, unbounded-delays, unliftio

Files

ChangeLog.md +5/−0
LICENSE +21/−0
README.md +175/−0
logstash.cabal +59/−0
src/Logstash.hs +244/−0
src/Logstash/Connection.hs +33/−0
src/Logstash/TCP.hs +150/−0

+ ChangeLog.md view

@@ -0,0 +1,5 @@+# Changelog for logstash++## v0.1++- First release with support for connecting to `tcp` (incl. TLS) Logstash inputs and the `json_lines` codec.

+ LICENSE view

@@ -0,0 +1,21 @@+MIT License++Copyright (c) 2020 Michael B. Gale++Permission is hereby granted, free of charge, to any person obtaining a copy+of this software and associated documentation files (the "Software"), to deal+in the Software without restriction, including without limitation the rights+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell+copies of the Software, and to permit persons to whom the Software is+furnished to do so, subject to the following conditions:++The above copyright notice and this permission notice shall be included in all+copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE+SOFTWARE.

+ README.md view

@@ -0,0 +1,175 @@+# Haskell client library for Logstash++![MIT](https://img.shields.io/github/license/mbg/logstash)+![CI](https://github.com/mbg/logstash/workflows/Build/badge.svg?branch=master)+![stackage-nightly](https://github.com/mbg/logstash/workflows/stackage-nightly/badge.svg)++This library implements a client for Logstash in Haskell. The following features are currently supported:++- Connections to Logstash via TCP or TLS (`tcp` input type).+- Support for the `json_lines` codec out of the box and custom codecs can be implemented (arbitrary `ByteString` data can be sent across the connections).+- This library can either be used without any logging framework or as a backend for [`monad-logger`](http://hackage.haskell.org/package/monad-logger/).+- Log messages can either be written synchronously or asynchronously.++For example, to connect to a Logstash server via TCP at `127.0.0.1:5000` (configuration given by `def`) and send a JSON document synchronously with a timeout of 1s and the default retry policy from [`Control.Retry`](https://hackage.haskell.org/package/retry/docs/Control-Retry.html):++```haskell+data Doc = Doc String++instance ToJSON Doc where+    toJSON (Doc msg) = object [ "message" .= msg ]++main :: IO ()+main = runLogstashConn (logstashTcp def) retryPolicyDefault 1000000 $+    stashJsonLine (Doc "Hello World")+```++Only the `tcp` input type (with or without TLS) is currently supported. For example, without TLS, the Logstash input configuration should roughly be:++```conf+input {+    tcp {+        port => 5000+        codec => "json_lines"+    }+}+```++With TLS, the expected Logstash configuration should roughly be:++```conf+input {+    tcp {+        port => 5000+        ssl_cert => "/usr/share/logstash/tls/cert.pem"+        ssl_key => "/usr/share/logstash/tls/key.pem"+        ssl_key_passphrase => "foobar"+        ssl_enable => true +        ssl_verify => false+        codec => "json_lines"+    }+}+```++## Configuring connections++Connections to Logstash are represented by the `LogstashConnection` type. To connect to Logstash via `tcp` use the `Logstash.TCP` module which exports four principal functions. Note that none of these functions establish any connections when they are called - instead, they allow `runLogstashConn` and `runLogstashPool` to establish connections/reuse them as needed:++- `logstashTcp` which, given a hostname and a port, will produce an `Acquire` that can be used with `runLogstashConn`.+- `logstashTcpPool` which, given a hostname and a port, will produce a `Pool` that can be used with `runLogstashPool`.+- `logstashTls` which, given a hostname, a port, and TLS client parameters, will produce an `Acquire` that can be used with `runLogstashConn`.+- `logstashTlsPool` which, given a hostname, a port, and TLS client parameters, will produce a `Pool` that can be used with `runLogstashPool`.++For `logstashTls` and `logstashTlsPool`, TLS `ClientParams` are required. It is worth noting that the `defaultParamsClient` function in the `tls` package does **not** set any supported ciphers and does **not** load the system trust store by default. For relatively sane defaults, it is worth using `newDefaultClientParams` from [`network-simple-tls`](http://hackage.haskell.org/package/network-simple-tls/) instead. For example:++```haskell+main :: IO ()+main = do +    params <- newDefaultClientParams ("127.0.0.1", "")++    runLogstashConn (logstashTls def params) retryPolicyDefault 1000000 $ +        stashJsonLine myDocument+```++## Logging things++The `Logstash` module exports functions for synchronous and asynchronous logging. Synchronous logging is acceptable for applications or parts of applications that are largely single-threaded where blocking on writes to Logstash is not an issue. For multi-threaded applications, such as web applications or services, you may wish to write log messages to Logstash asynchronously instead. In the latter model, log messages are added to a bounded queue which is processed asynchronously by worker threads. ++### Synchronously++The logging functions exported by the `Logstash` module are backend-independent can be invoked synchronously with `runLogstash`, which is overloaded to work with either `Acquire LogstashConnection` or `LogstashPool` (`Pool LogstashConnection`) values and maps to one of the two implementations described below. In either case, you must supply a retry policy and a timeout (in microseconds). The retry policy determines whether performing the logging action should be re-attempted if an exception occurs. The order of operations is:++1. The retry policy is applied.+2. A connection is established using the provided Logstash context.+3. The timeout is applied.+4. The Logstash computation is executed.++If the computation is successful, each step will only be executed once. If an exception is raised by the computation or the timeout, the connection to the Logstash server is terminated and the exception propagated to the retry policy. If the retry policy determines that the computation should be re-attempted, steps 2-4 will happen again. The timeout applies to every attempt individually and should be chosen appropriately in conjunction with the retry policy in mind.++Depending on whether the Logstash context is a `Acquire LogstashConnection` value or a `LogstashPool` (`Pool LogstashConnection`) value, the `runLogstash` functions maps to one of:++- `runLogstashConn` for `Acquire LogstashConnection` (e.g. the result of `logstashTcp` or `logstashTls`).+- `runLogstashPool` for `Pool LogstashConnection` (e.g. the result of `logstashTcpPool` or `logstashTlsPool`). If a connection is available in the pool, that connection will be used. If no connection is available but there is an empty space in the pool, a new connection will be established. If neither is true, this function blocks until a connection is available. The computation that is provided as the second argument is then run with the connection. In the event of an exception, the connection is not returned to the pool.++#### Stashing things by hand++The following functions allow sending data synchronously via the Logstash connection:++- `stash` is a general-purpose function for sending `ByteString` data to the server. No further processing is performed on the data.+- `stashJsonLine` is for use with the `json_line` codec. The argument is encoded as JSON and a `\n` character is appended, which is then sent to the server. ++Any exception raised by the above `stash`ing functions will likely be due to a bad connection. The `runLogstash` functions apply the retry policy before establishing a connection, so in the event that an exception is raised, a new connection will be established for the next attempt.++### Asynchronously++The `withLogstashQueue` function is used for asynchronous logging. When called, it sets up a bounded queue that is then used to communicate log messages to worker threads which dispatch them to Logstash. A minimal example with default settings is shown below:++```haskell+data Doc = Doc String++instance ToJSON Doc where+    toJSON (Doc msg) = object [ "message" .= msg ]++main :: IO ()+main = do+    let ctx = logstashTcp def+    let cfg = defaultLogstashQueueCfg ctx+    +    withLogstashQueue cfg (const stashJsonLine) [] $ \queue -> do+        atomically $ writeTBMQueue queue (Doc "Hello World")+```++The `[]` given to `withLogstashQueue` allows installing exception handlers that are used to handle the case where a log message has exhausted the retry policy. This can e.g. be used to fall back to the standard output for logging as a last resort and to stop the worker thread from getting terminated by an exception that may be recoverable.++The queue is automatically closed when the inner computation returns. The worker threads will continue running until the queue is empty and then terminate. `withLogstashQueue` will not return until all worker threads have returned.++### Usage with `monad-logger`++The `monad-logger-logstash` package provides convenience functions and types for working with [`monad-logger`](http://hackage.haskell.org/package/monad-logger/). ++#### Synchronous logging++The following example demonstrates how to use the `runLogstashLoggerT` function with a TCP connection to Logstash, the default retry policy from [`Control.Retry`](https://hackage.haskell.org/package/retry/docs/Control-Retry.html), a 1s timeout for each attempt, and the `json_lines` codec:++```haskell+main :: IO ()+main = do +    let ctx = logstashTcp def+    runLogstashLoggerT ctx retryPolicyDefault 1000000 (const stashJsonLine) $ +        logInfoN "Hello World"+```++Each call to a logging function such as `logInfoN` in the example will result in the log message being written to Logstash synchronously. ++#### Asynchronous logging++The `withLogstashLoggerT` function is the analogue of `withLogstashQueue` for `monad-logger`. It performs the same setup as `withLogstashQueue`, but automatically adds all log messages from logging functions to the queue. A minimal example with default settings is:++```haskell+main :: IO ()+main = do +    let ctx = logstashTcp def+    withLogstashLoggerT (defaultLogstashQueueCfg ctx) (const stashJsonLine) [] $ +        logInfoN "Hello World"+```++### Usage with `katip`++The `katip-logstash` package provides convenience functions and types for working with [`katip`](http://hackage.haskell.org/package/katip/). ++#### Asynchronous logging++The `withLogstashScribe` function is the analogue of `withLogstashQueue` for `katip`. It performs the same setup as `withLogstashQueue`, but provides a `Scribe` instead of the raw queue. A minimal example with default settings is (adapted from the `katip` documentation):++```haskell+main :: IO ()+main = do +    let ctx = logstashTcp def+    withLogstashScribe (defaultLogstashQueueCfg ctx) (const $ pure True) (itemJson V3) (const stashJsonLine) [] $ \logstashScribe -> do+        let makeLogEnv = registerScribe "logstash" logstashScribe defaultScribeSettings =<< initLogEnv "MyApp" "production"+        bracket makeLogEnv closeScribes $ \le -> do+            let initialContext = ()+            let initialNamespace = "main"+            runKatipContextT le initialContext initialNamespace $ do+                $(logTM) InfoS "Hello World"+```

+ logstash.cabal view

@@ -0,0 +1,59 @@+cabal-version: 1.12++-- This file has been generated from package.yaml by hpack version 0.33.0.+--+-- see: https://github.com/sol/hpack+--+-- hash: b7aa35eeee7789f4d5593a17d55eee75a1241ad7b1549bb21a1b45c6056ca022++name:           logstash+version:        0.1.0.0+synopsis:       Logstash client library for Haskell+description:    Please see the README on GitHub at <https://github.com/mbg/logstash#readme>+category:       Network+homepage:       https://github.com/mbg/logstash#readme+bug-reports:    https://github.com/mbg/logstash/issues+author:         Michael B. Gale+maintainer:     m.gale@warwick.ac.uk+copyright:      Copyright (c) 2020 Michael B. Gale+license:        MIT+license-file:   LICENSE+build-type:     Simple+extra-source-files:+    README.md+    ChangeLog.md++source-repository head+  type: git+  location: https://github.com/mbg/logstash++library+  exposed-modules:+      Logstash+      Logstash.Connection+      Logstash.TCP+  other-modules:+      Paths_logstash+  hs-source-dirs:+      src+  default-extensions: OverloadedStrings RecordWildCards FlexibleContexts FlexibleInstances+  build-depends:+      aeson+    , async+    , base >=4.7 && <5+    , bytestring+    , data-default-class+    , exceptions+    , monad-control+    , mtl+    , network+    , resource-pool+    , resourcet+    , retry+    , stm+    , stm-chans+    , time+    , tls+    , unbounded-delays+    , unliftio+  default-language: Haskell2010

+ src/Logstash.hs view

@@ -0,0 +1,244 @@+--------------------------------------------------------------------------------+-- Logstash client for Haskell                                                --+--------------------------------------------------------------------------------+-- This source code is licensed under the MIT license found in the LICENSE    --+-- file in the root directory of this source tree.                            --+--------------------------------------------------------------------------------++-- | A simple Logstash client.+module Logstash (+    module Logstash.Connection,+    +    -- * Running Logstash actions+    LogstashContext(..),+    runLogstashConn,+    runLogstashPool,++    LogstashQueueCfg(..),+    defaultLogstashQueueCfg,+    withLogstashQueue,++    -- * Codecs+    stash, +    stashJsonLine+) where ++--------------------------------------------------------------------------------++import Control.Concurrent+import Control.Concurrent.Async+import Control.Concurrent.STM+import Control.Concurrent.STM.TBMQueue+import Control.Concurrent.Timeout+import Control.Exception+import Control.Monad.Catch (MonadMask)+import Control.Monad.Reader+import Control.Monad.Trans.Control+import Control.Retry++import Data.Aeson+import Data.Acquire+import qualified Data.ByteString.Lazy as BSL+import Data.Either (isRight)+import Data.Pool++import UnliftIO (MonadUnliftIO(..))++import Logstash.Connection++--------------------------------------------------------------------------------++-- | `stash` @timeout message@ is a computation which sends @message@ to +-- the Logstash server. +stash +    :: MonadIO m +    => BSL.ByteString +    -> ReaderT LogstashConnection m ()+stash msg = ask >>= \con -> liftIO $ writeData con msg++-- | `stashJsonLine` @timeout document@ is a computation which serialises+-- @document@ and sends it to the Logstash server. This function is intended+-- to be used with the `json_lines` codec.+stashJsonLine +    :: (MonadIO m, ToJSON a) +    => a +    -> ReaderT LogstashConnection m ()+stashJsonLine = stash . (<> "\n") . encode++--------------------------------------------------------------------------------++-- | A type of exceptions that can occur related to Logstash connections.+data LogstashException = LogstashTimeout+    deriving (Eq, Show) ++instance Exception LogstashException where +    displayException _ = "Writing to Logstash timed out."++-- | A type class of types that provide Logstash connections.+class LogstashContext ctx where +    runLogstash +        :: (MonadMask m, MonadUnliftIO m) +        => ctx +        -> RetryPolicyM m+        -> Integer+        -> (RetryStatus -> ReaderT LogstashConnection m a) +        -> m a++instance LogstashContext (Acquire LogstashConnection) where +    runLogstash = runLogstashConn++instance LogstashContext LogstashPool where+    runLogstash = runLogstashPool++-- | `runLogstashConn` @connectionAcquire computation@ runs @computation@ using+-- a Logstash connection produced by @connectionAcquire@.+runLogstashConn +    :: (MonadMask m, MonadUnliftIO m)+    => Acquire LogstashConnection+    -> RetryPolicyM m+    -> Integer +    -> (RetryStatus -> ReaderT LogstashConnection m a) +    -> m a +runLogstashConn acq policy time action = +    recoverAll policy $ \s -> +    withRunInIO $ \runInIO -> do +    -- run the Logstash action with the specified timeout+    mr <- timeout time $ runInIO $ withAcquire acq $ runReaderT (action s)+    +    -- raise an exception if a timeout occurred+    maybe (throw LogstashTimeout) pure mr++-- | `runLogstashPool` @pool computation@ takes a `LogstashConnection` from+-- @pool@ and runs @computation@ with it.+runLogstashPool +    :: (MonadMask m, MonadUnliftIO m) +    => LogstashPool +    -> RetryPolicyM m+    -> Integer+    -> (RetryStatus -> ReaderT LogstashConnection m a)+    -> m a+runLogstashPool pool policy time action =+    recoverAll policy $ \s -> +    withRunInIO $ \runInIO -> +    mask $ \restore -> do+        -- acquire a connection from the resource pool+        (resource, local) <- takeResource pool++        mr <- restore (timeout time $ runInIO $ runReaderT (action s) resource) +            `onException` destroyResource pool local resource++        -- raise an exception if a timeout occurred+        maybe (throw LogstashTimeout) pure mr++--------------------------------------------------------------------------------++-- | Configurations for `withLogstashQueue` which control the general Logstash+-- configuration as well as the size of the bounded queue and the number of+-- worker threads.+data LogstashQueueCfg ctx = MkLogstashQueueCfg {+    -- | The connection context for the worker threads.+    logstashQueueContext :: ctx,+    -- | The size of the queue.+    logstashQueueSize :: Int,+    -- | The number of workers. This must be at least 1.+    logstashQueueWorkers :: Int,+    -- | The retry policy.+    logstashQueueRetryPolicy :: RetryPolicyM IO,+    -- | The timeout for each attempt at sending data to the Logstash server.+    logstashQueueTimeout :: Integer+}++-- | `defaultLogstashQueueCfg` @ctx@ constructs a `LogstashQueueCfg` with+-- some default values for the Logstash context given by @ctx@:+-- +-- - A queue size limited to 1000 entries+-- - Two worker threads+-- - 25ms exponential backoff with a maximum of five retries as retry policy+-- - 1s timeout for each logging attempt+--+-- You may wish to customise these settings to suit your needs.+defaultLogstashQueueCfg :: LogstashContext ctx => ctx -> LogstashQueueCfg ctx+defaultLogstashQueueCfg ctx = MkLogstashQueueCfg{+    logstashQueueContext = ctx,+    logstashQueueSize = 1000,+    logstashQueueWorkers = 2,+    logstashQueueRetryPolicy = exponentialBackoff 25000 <> limitRetries 5,+    logstashQueueTimeout = 1000000+}++-- | `withLogstashQueue` @cfg codec exceptionHandlers action@ initialises a+-- queue with space for a finite number of log messages given by @cfg@ to allow +-- for log messages to be sent to Logstash asynchronously. This addresses the +-- following issues with synchronous logging:+--+-- - Since writing log messages to Logstash involves network I/O, this may +--   be slower than queueing up messages in memory and therefore synchronously +--   sending messages may delay the computation that originated the log +--   message.+-- - With a finite number of Logstash connections, synchronous logging may also+--   get blocked until a connection is available.+--+-- The queue is read from by a configurable number of worker threads which +-- use Logstash connections from a `LogstashContext`. The queue is given+-- to @action@ as an argument. The @retryPolicy@ and @timeout@ parameters+-- serve the same purpose as for `runLogstash`. We recommend that, if the+-- `LogstashContext` is a `LogstashPool`, it should contain at least as many +-- connections as the number of works to avoid contention between worker +-- threads.+--+-- @codec@ is the handler for how messages should be sent to the Logstash +-- server, this is typically a codec like `stashJsonLine`. The `RetryStatus`+-- is provided as an additional argument to @codec@ in case a handler wishes+-- to inspect this.+withLogstashQueue +    :: (LogstashContext ctx, MonadUnliftIO m)+    => LogstashQueueCfg ctx+    -> (RetryStatus -> item -> ReaderT LogstashConnection IO ())+    -> [item -> Handler ()]+    -> (TBMQueue item -> m a)+    -> m a+withLogstashQueue MkLogstashQueueCfg{..} dispatch handlers action = +    withRunInIO $ \runInIO -> do+        -- initialise a bounded queue with the specified size+        queue <- newTBMQueueIO logstashQueueSize++        let worker = do +                -- [blocking] read the next item from the queue+                mitem <- atomically $ readTBMQueue queue ++                -- the item will be Nothing if the queue is empty and has+                -- been closed; otherwise we have an item that should be+                -- dispatched to Logstash+                case mitem of +                    Nothing -> pure ()+                    Just item -> do +                        -- dispatch the item using the policies from the+                        -- configuration and the provided dispatch handler+                        -- this may fail if the retry policy is exhausted,+                        -- in which case we use the provided exception+                        -- handlers to try and catch the exception+                        runLogstash logstashQueueContext +                                    logstashQueueRetryPolicy +                                    logstashQueueTimeout +                                    (\status -> dispatch status item)+                            `catches` (map ($ item) handlers)+                        +                        -- loop+                        worker++        -- initialise the requested number of worker threads+        workers <- replicateM logstashQueueWorkers $ async worker++        -- run the main computation in the current thread with its original+        -- monad stack and give it the the queue to write log messages to+        runInIO (action queue) `finally` do +            -- close the queue to allow the worker threads to gracefully +            -- shut down+            atomically $ closeTBMQueue queue++            -- wait for the worker threads to terminate; we use `waitCatch` +            -- here instead of `wait` because `waitCatch` does not raise any +            -- exceptions that may occur in the worker threads in this thread+            mapM_ waitCatch workers++--------------------------------------------------------------------------------

+ src/Logstash/Connection.hs view

@@ -0,0 +1,33 @@+--------------------------------------------------------------------------------+-- Logstash client for Haskell                                                --+--------------------------------------------------------------------------------+-- This source code is licensed under the MIT license found in the LICENSE    --+-- file in the root directory of this source tree.                            --+--------------------------------------------------------------------------------++module Logstash.Connection (+    LogstashConnection(..),+    LogstashPool+) where++--------------------------------------------------------------------------------++import qualified Data.ByteString.Lazy as BSL+import Data.Pool++--------------------------------------------------------------------------------++-- | Represents an abstract interface for Logstash connections that hides+-- details about the nature of the connection.+data LogstashConnection = LogstashConnection {+    -- | A computation which sends data to the logstash server.+    writeData :: BSL.ByteString -> IO (),+    -- | A computation which closes the connection.+    closeConnection :: IO ()+}++-- | For convenience so that importing modules do not have to import +-- `Data.Pool`, a type alias for a `Pool` of `LogstashConnection`.+type LogstashPool = Pool LogstashConnection++--------------------------------------------------------------------------------

+ src/Logstash/TCP.hs view

@@ -0,0 +1,150 @@+--------------------------------------------------------------------------------+-- Logstash client for Haskell                                                --+--------------------------------------------------------------------------------+-- This source code is licensed under the MIT license found in the LICENSE    --+-- file in the root directory of this source tree.                            --+--------------------------------------------------------------------------------++-- | This module implements a logstash client for the tcp input plugin:+-- https://www.elastic.co/guide/en/logstash/7.10/plugins-inputs-tcp.html+module Logstash.TCP (+    LogstashTcpConfig(..),+    logstashTcp,+    logstashTcpPool,+    logstashTls,+    logstashTlsPool+) where ++--------------------------------------------------------------------------------++import Control.Monad.IO.Class++import Data.Acquire+import qualified Data.ByteString as BS+import qualified Data.ByteString.Lazy as BSL+import Data.Default.Class+import Data.Time+import Data.Pool++import Network.Socket+import Network.Socket.ByteString (sendAll)+import Network.TLS++import Logstash.Connection++--------------------------------------------------------------------------------++-- | Represents configurations for Logstash TCP inputs.+data LogstashTcpConfig = LogstashTcpConfig {+    -- | The hostname of the server to connect to.+    logstashTcpHost :: String,+    -- | The port of the server to connect to.+    logstashTcpPort :: Int+} deriving (Eq, Show)++instance Default LogstashTcpConfig where +    def = LogstashTcpConfig{+        logstashTcpHost = "127.0.0.1",+        logstashTcpPort = 5000+    }++-- | `connectTCP` @config@ establishes a TCP socket connection to the server+-- configured by @config@.+connectTcpSocket +    :: MonadIO m +    => LogstashTcpConfig +    -> m Socket+connectTcpSocket LogstashTcpConfig{..} = liftIO $ withSocketsDo $ do+    -- initialise a TCP socket for the given host and port+    let hints = defaultHints{ addrSocketType = Stream }+    addr <- head <$> getAddrInfo +        (Just hints) (Just logstashTcpHost) (Just $ show logstashTcpPort)+    sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)+    connect sock $ addrAddress addr++    -- return the socket+    pure sock++-- | `createTcpConnection` @config@ establishes a `LogstashConnection` via+-- TCP to the server configured by @config@.+createTcpConnection +    :: MonadIO m +    => LogstashTcpConfig +    -> m LogstashConnection+createTcpConnection cfg = do +    -- establish a TCP connection+    sock <- connectTcpSocket cfg++    -- return the Logstash connection+    pure $ LogstashConnection{+        writeData = sendAll sock . BSL.toStrict,+        closeConnection = close sock+    }++-- | `logstashTcp` @config@ produces an `Acquire` for establishing+-- `LogstashConnection` values for the given TCP @config@.+logstashTcp +    :: LogstashTcpConfig +    -> Acquire LogstashConnection+logstashTcp cfg = mkAcquire (createTcpConnection cfg) closeConnection ++-- | `logstashTcpPool` @config stripes ttl resources@ produces a `Pool`+-- of `LogstashConnection` values for the given TCP @config@. The other+-- parameters are passed on directly to `createPool`.+logstashTcpPool +    :: LogstashTcpConfig+    -> Int +    -> NominalDiffTime+    -> Int +    -> IO LogstashPool+logstashTcpPool cfg = createPool (createTcpConnection cfg) closeConnection++--------------------------------------------------------------------------------++-- | `createTlsConnection` @config params@ establishes a `LogstashConnection` via+-- TLS to the server configured by @config@ and the TLS configuration given by+-- @params@.+createTlsConnection +    :: MonadIO m +    => LogstashTcpConfig +    -> ClientParams+    -> m LogstashConnection+createTlsConnection cfg params = do +    -- establish a TCP connection+    sock <- connectTcpSocket cfg++    -- establish a TLS connection on top+    ctx <- contextNew sock params+    handshake ctx++    -- return the Logstash connection+    pure $ LogstashConnection{+        writeData = sendData ctx,+        closeConnection = do +            bye ctx +            close sock+    }++-- | `logstashTls` @config params@ produces an `Acquire` for establishing+-- `LogstashConnection` values for the given TCP @config@ and TLS @params@.+logstashTls +    :: LogstashTcpConfig +    -> ClientParams +    -> Acquire LogstashConnection +logstashTls cfg params = +    mkAcquire (createTlsConnection cfg params) closeConnection ++-- | `logstashTlsPool` @config params stripes ttl resources@ produces a +-- `Pool` of `LogstashConnection` values for the given TCP @config@ and TLS+-- @params@. The other parameters are passed on directly to `createPool`.+logstashTlsPool+    :: LogstashTcpConfig +    -> ClientParams +    -> Int +    -> NominalDiffTime+    -> Int+    -> IO LogstashPool+logstashTlsPool cfg params = +    createPool (createTlsConnection cfg params) closeConnection ++--------------------------------------------------------------------------------