<commit_before>--
-- Copyright (C) 2012 <NAME>
--
-- Licensed under the Apache License, Version 2.0 (the "License");
-- you may not use this file except in compliance with the License.
-- You may obtain a copy of the License at
--
--     http://www.apache.org/licenses/LICENSE-2.0
--
-- Unless required by applicable law or agreed to in writing, software
-- distributed under the License is distributed on an "AS IS" BASIS,
-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-- See the License for the specific language governing permissions and
-- limitations under the License.
--

-- | Fluent Logger for Haskell
module Network.Fluent.Logger
    ( -- * Logger
      FluentLogger
    , withFluentLogger
    , newFluentLogger
    , closeFluentLogger
      -- * Settings
    , FluentSettings(..)
    , defaultFluentSettings
      -- * Post
    , post
    , postWithTime
    ) where

import qualified Data.ByteString.Char8 as BS ( ByteString, pack, unpack, empty, null)
import qualified Data.ByteString.Lazy as LBS ( ByteString, length )
import Data.Monoid ( mconcat )
import qualified Network.Socket as NS
import Network.Socket.Options ( setRecvTimeout, setSendTimeout )
import Network.Socket.ByteString.Lazy ( sendAll )
import Control.Monad ( void, forever, when )
import Control.Applicative ( (<$>) )
import Control.Concurrent ( ThreadId, forkIO, killThread, threadDelay )
import Control.Concurrent.STM ( atomically
                              , TChan, newTChanIO, readTChan, peekTChan, writeTChan
                              , TVar, newTVarIO, readTVar, modifyTVar )
import Control.Exception ( SomeException, handle, bracket, throwIO )
import Data.MessagePack
import Data.Serialize hiding (label)
import Data.Int ( Int64 )
import Data.Time.Clock.POSIX ( getPOSIXTime )
import System.Random ( randomRIO )

import Network.Fluent.Logger.Packable

-- | Fluent logger settings
--
-- Since 0.1.0.0
--
data FluentSettings =
    FluentSettings
    { fluentSettingsTag :: BS.ByteString
    , fluentSettingsHost :: BS.ByteString
    , fluentSettingsPort :: Int
    , fluentSettingsTimeout :: Double
    , fluentSettingsBufferLimit :: Int64
    }

-- | Default fluent logger settings
--
-- Since 0.1.0.0
--
defaultFluentSettings :: FluentSettings
defaultFluentSettings =
    FluentSettings
    { fluentSettingsTag = BS.empty
    , fluentSettingsHost = BS.pack "localhost"
    , fluentSettingsPort = 24224
    , fluentSettingsTimeout = 3.0
    , fluentSettingsBufferLimit = 1024*1024
    }

-- | Fluent logger
--
-- Since 0.1.0.0
--
data FluentLogger =
    FluentLogger
    { fluentLoggerSender :: FluentLoggerSender
    , fluentLoggerThread :: ThreadId
    }

data FluentLoggerSender =
    FluentLoggerSender
    { fluentLoggerSenderChan :: TChan LBS.ByteString
    , fluentLoggerSenderBuffered :: TVar Int64
    , fluentLoggerSenderSettings :: FluentSettings
    }

getSocket :: BS.ByteString -> Int -> Int64 -> IO NS.Socket
getSocket host port timeout = do
  let hints = NS.defaultHints { NS.addrFlags = [NS.AI_ADDRCONFIG]
                              , NS.addrSocketType = NS.Stream
                              }
  (addr:_) <- NS.getAddrInfo (Just hints) (Just $ BS.unpack host) (Just $ show port)
  sock <- NS.socket (NS.addrFamily addr) (NS.addrSocketType addr) (NS.addrProtocol addr)
  setRecvTimeout sock timeout
  setSendTimeout sock timeout
  let onErr :: SomeException -> IO a
      onErr e = NS.sClose sock >> throwIO e
  handle onErr $ do
    NS.connect sock $ NS.addrAddress addr
    return sock

runSender :: FluentLoggerSender -> IO ()
runSender logger = forever $ connectFluent logger >>= sendFluent logger

connectFluent :: FluentLoggerSender -> IO NS.Socket
connectFluent logger = exponentialBackoff $ getSocket host port timeout where
    set = fluentLoggerSenderSettings logger
    host = fluentSettingsHost set
    port = fluentSettingsPort set
    timeout = round $ fluentSettingsTimeout set * 1000000

exponentialBackoff :: IO a -> IO a
exponentialBackoff action = handle (retry 100000) action where
    retry failCount exception =
        let _ = exception :: SomeException
        in exponentialBackoff' failCount
    exponentialBackoff' interval = do
      delay <- randomRIO (interval `div` 2, interval * 3 `div` 2)
      threadDelay delay
      handle (retry $ min 60000000 $ interval * 3 `div` 2) action

sendFluent :: FluentLoggerSender -> NS.Socket -> IO ()
sendFluent logger sock = handle (done sock) toSender where
    chan = fluentLoggerSenderChan logger
    buffered = fluentLoggerSenderBuffered logger
    done :: NS.Socket -> SomeException -> IO ()
    done = const . NS.sClose
    toSender = do
      entry <- atomically $ peekTChan chan
      sendAll sock entry
      atomically $ do
        void $ readTChan chan
        modifyTVar buffered (subtract $ LBS.length entry)
      sendFluent logger sock

-- | Create a fluent logger
--
-- Since 0.1.0.0
--
newFluentLogger :: FluentSettings -> IO FluentLogger
newFluentLogger set = do
  tchan <- newTChanIO
  tvar <- newTVarIO 0
  let sender = FluentLoggerSender
               { fluentLoggerSenderChan = tchan<commit_msg>Use bracket in runSender so that it terminates the thread and closes the socket when ThreadKilled async exception is raised.<commit_after>--
-- Copyright (C) 2012 <NAME>
--
-- Licensed under the Apache License, Version 2.0 (the "License");
-- you may not use this file except in compliance with the License.
-- You may obtain a copy of the License at
--
--     http://www.apache.org/licenses/LICENSE-2.0
--
-- Unless required by applicable law or agreed to in writing, software
-- distributed under the License is distributed on an "AS IS" BASIS,
-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-- See the License for the specific language governing permissions and
-- limitations under the License.
--

-- | Fluent Logger for Haskell
module Network.Fluent.Logger
    ( -- * Logger
      FluentLogger
    , withFluentLogger
    , newFluentLogger
    , closeFluentLogger
      -- * Settings
    , FluentSettings(..)
    , defaultFluentSettings
      -- * Post
    , post
    , postWithTime
    ) where

import qualified Data.ByteString.Char8 as BS ( ByteString, pack, unpack, empty, null)
import qualified Data.ByteString.Lazy as LBS ( ByteString, length )
import Data.Monoid ( mconcat )
import qualified Network.Socket as NS
import Network.Socket.Options ( setRecvTimeout, setSendTimeout )
import Network.Socket.ByteString.Lazy ( sendAll )
import Control.Monad ( void, forever, when )
import Control.Applicative ( (<$>) )
import Control.Concurrent ( ThreadId, forkIO, killThread, threadDelay )
import Control.Concurrent.STM ( atomically
                              , TChan, newTChanIO, readTChan, peekTChan, writeTChan
                              , TVar, newTVarIO, readTVar, modifyTVar )
import Control.Exception ( SomeException, handle, bracket, throwIO )
import Data.MessagePack
import Data.Serialize hiding (label)
import Data.Int ( Int64 )
import Data.Time.Clock.POSIX ( getPOSIXTime )
import System.Random ( randomRIO )

import Network.Fluent.Logger.Packable

-- | Fluent logger settings
--
-- Since 0.1.0.0
--
data FluentSettings =
    FluentSettings
    { fluentSettingsTag :: BS.ByteString
    , fluentSettingsHost :: BS.ByteString
    , fluentSettingsPort :: Int
    , fluentSettingsTimeout :: Double
    , fluentSettingsBufferLimit :: Int64
    }

-- | Default fluent logger settings
--
-- Since 0.1.0.0
--
defaultFluentSettings :: FluentSettings
defaultFluentSettings =
    FluentSettings
    { fluentSettingsTag = BS.empty
    , fluentSettingsHost = BS.pack "localhost"
    , fluentSettingsPort = 24224
    , fluentSettingsTimeout = 3.0
    , fluentSettingsBufferLimit = 1024*1024
    }

-- | Fluent logger
--
-- Since 0.1.0.0
--
data FluentLogger =
    FluentLogger
    { fluentLoggerSender :: FluentLoggerSender
    , fluentLoggerThread :: ThreadId
    }

data FluentLoggerSender =
    FluentLoggerSender
    { fluentLoggerSenderChan :: TChan LBS.ByteString
    , fluentLoggerSenderBuffered :: TVar Int64
    , fluentLoggerSenderSettings :: FluentSettings
    }

getSocket :: BS.ByteString -> Int -> Int64 -> IO NS.Socket
getSocket host port timeout = do
  let hints = NS.defaultHints { NS.addrFlags = [NS.AI_ADDRCONFIG]
                              , NS.addrSocketType = NS.Stream
                              }
  (addr:_) <- NS.getAddrInfo (Just hints) (Just $ BS.unpack host) (Just $ show port)
  sock <- NS.socket (NS.addrFamily addr) (NS.addrSocketType addr) (NS.addrProtocol addr)
  setRecvTimeout sock timeout
  setSendTimeout sock timeout
  let onErr :: SomeException -> IO a
      onErr e = NS.sClose sock >> throwIO e
  handle onErr $ do
    NS.connect sock $ NS.addrAddress addr
    return sock

runSender :: FluentLoggerSender -> IO ()
runSender logger = forever $ bracket (connectFluent logger) NS.sClose (sendFluent logger)

connectFluent :: FluentLoggerSender -> IO NS.Socket
connectFluent logger = exponentialBackoff $ getSocket host port timeout where
    set = fluentLoggerSenderSettings logger
    host = fluentSettingsHost set
    port = fluentSettingsPort set
    timeout = round $ fluentSettingsTimeout set * 1000000

exponentialBackoff :: IO a -> IO a
exponentialBackoff action = handle (retry 100000) action where
    retry failCount exception =
        let _ = exception :: SomeException
        in exponentialBackoff' failCount
    exponentialBackoff' interval = do
      delay <- randomRIO (interval `div` 2, interval * 3 `div` 2)
      threadDelay delay
      handle (retry $ min 60000000 $ interval * 3 `div` 2) action

sendFluent :: FluentLoggerSender -> NS.Socket -> IO ()
sendFluent logger sock = toSender where
    chan = fluentLoggerSenderChan logger
    buffered = fluentLoggerSenderBuffered logger
    toSender = do
      entry <- atomically $ peekTChan chan
      sendAll sock entry
      atomically $ do
        void $ readTChan chan
        modifyTVar buffered (subtract $ LBS.length entry)
      sendFluent logger sock

-- | Create a fluent logger
--
-- Since 0.1.0.0
--
newFluentLogger :: FluentSettings -> IO FluentLogger
newFluentLogger set = do
  tchan <- newTChanIO
  tvar <- newTVarIO 0
  let sender = FluentLoggerSender
               { fluentLoggerSenderChan = tchan
{'fraction_non_alphanumeric': 0.08520353638395026, 'fraction_numerical': 0.01457301078402798, 'mean_word_length': 3.065560821484992, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 35, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 8, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4711893', 'n_tokens_mistral': 3060, 'n_tokens_neox': 2892, 'n_words': 918}