diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,13 @@ +* 0.3.0 (2023-09-28) + + - GHC 8.x is no longer tested, the minimum version is 9.2.8 + - Various fixes the Jaeger backend + - Fix Zipkin backend to use POST (#49) + - Use a bounded queue for the batch reporter (#46) + + This release is considered _breaking_ due to #46, and (potential) + incompatibility with older GHCs. + * 0.2.2 (2022-05-09) Support for base 4.16 / GHC 9.2 diff --git a/OpenTracing/Reporting/Batch.hs b/OpenTracing/Reporting/Batch.hs --- a/OpenTracing/Reporting/Batch.hs +++ b/OpenTracing/Reporting/Batch.hs @@ -16,11 +16,15 @@ module OpenTracing.Reporting.Batch ( BatchOptions , batchOptions + , boptAtCapacity , boptBatchSize - , boptTimeoutSec - , boptReporter , boptErrorLog + , boptQueueSize + , boptReporter + , boptTimeoutSec + , AtCapacity (..) + , defaultErrorLog , BatchEnv @@ -31,20 +35,21 @@ ) where -import Control.Concurrent.Async -import Control.Concurrent.STM -import Control.Exception (AsyncException (ThreadKilled)) -import Control.Exception.Safe -import Control.Lens -import Control.Monad -import Control.Monad.IO.Class -import Data.ByteString.Builder -import Data.Time (NominalDiffTime) -import Data.Word -import OpenTracing.Span -import OpenTracing.Time -import System.IO (stderr) -import System.Timeout +import Control.Concurrent.Async +import Control.Concurrent.STM +import Control.Exception (AsyncException (ThreadKilled)) +import Control.Exception.Safe +import Control.Lens +import Control.Monad +import Control.Monad.IO.Class +import Data.ByteString.Builder +import Data.Time (NominalDiffTime) +import Data.Word +import Numeric.Natural (Natural) +import OpenTracing.Span +import OpenTracing.Time +import System.IO (stderr) +import System.Timeout -- | Options available to construct a batch reporter. Default options are -- available with `batchOptions` @@ -59,8 +64,15 @@ -- to _boptBatchSize. No default. , _boptErrorLog :: Builder -> IO () -- ^ What to do with errors. Default print to stderr. + , _boptQueueSize :: Natural + -- ^ Size of the queue holding batched spans. Default 1000 + , _boptAtCapacity :: AtCapacity + -- ^ What to do when the queue is at capacity. Default: Drop } +-- | Policy to apply to new spans when the internal queue is at capacity. +data AtCapacity = Drop | Block + -- | Default batch options which can be overridden via lenses. batchOptions :: ([FinishedSpan] -> IO ()) -> BatchOptions batchOptions f = BatchOptions @@ -68,6 +80,8 @@ , _boptTimeoutSec = 5 , _boptReporter = f , _boptErrorLog = defaultErrorLog + , _boptQueueSize = 1000 + , _boptAtCapacity = Drop } -- | An error logging function which prints to stderr. @@ -76,38 +90,67 @@ makeLenses ''BatchOptions - -- | The environment of a batch reporter. data BatchEnv = BatchEnv - { envQ :: TQueue FinishedSpan + { envQ :: TBQueue FinishedSpan -- ^ The queue of spans to be reported , envRep :: Async () -- ^ Asynchronous consumer of the queue + , envCap :: AtCapacity + -- ^ Policy to apply when the queue is at capacity + , envLog :: Builder -> IO () + -- ^ Where to report errors } -- | Create a new batch environment newBatchEnv :: BatchOptions -> IO BatchEnv newBatchEnv opt = do - q <- newTQueueIO - BatchEnv q <$> consumer opt q + q <- newTBQueueIO (_boptQueueSize opt) + c <- consumer opt q + pure BatchEnv + { envQ = q + , envRep = c + , envCap = _boptAtCapacity opt + , envLog = _boptErrorLog opt + } -- | Close a batch reporter, stop consuming any new spans. Any -- spans in the queue will be drained. closeBatchEnv :: BatchEnv -> IO () closeBatchEnv = cancel . envRep --- | An implementation of `OpenTracing.Tracer.tracerReport` that batches the finished --- spans for transimission to their destination. +-- | An implementation of `OpenTracing.Tracer.tracerReport` that batches the +-- finished spans for transimission to their destination. +-- +-- If the underlying queue is currently at capacity, the behaviour depends on +-- the setting of `boptAtCapacity`: if the value is `Drop`, `fspan` is dropped, +-- otherwise, if the value is `Block`, the reporter will block until the queue +-- has enough space to accept the span. +-- +-- In either case, a log record is emitted. batchReporter :: MonadIO m => BatchEnv -> FinishedSpan -> m () -batchReporter BatchEnv{envQ} = liftIO . atomically . writeTQueue envQ +batchReporter BatchEnv{envCap = Block, envQ, envLog} fspan = liftIO $ do + full <- atomically $ isFullTBQueue envQ + when full $ + envLog "Queue at capacity, enqueueing span may block\n" + atomically $ writeTBQueue envQ fspan -consumer :: BatchOptions -> TQueue FinishedSpan -> IO (Async ()) +batchReporter BatchEnv{envCap = Drop, envQ, envLog} fspan = liftIO $ do + full <- atomically $ do + full <- isFullTBQueue envQ + unless full $ + writeTBQueue envQ fspan + pure full + when full $ + envLog "Queue at capacity, span was dropped\n" + +consumer :: BatchOptions -> TBQueue FinishedSpan -> IO (Async ()) consumer opt@BatchOptions{..} q = async . forever $ do xs <- popBlocking go False xs where popBlocking = atomically $ do - x <- readTQueue q + x <- readTBQueue q (x:) <$> pop (_boptBatchSize - 1) q popNonblock = atomically $ pop _boptBatchSize q @@ -138,10 +181,10 @@ timeoutMicros = micros @NominalDiffTime $ fromIntegral _boptTimeoutSec -pop :: Word16 -> TQueue a -> STM [a] +pop :: Word16 -> TBQueue a -> STM [a] pop 0 _ = pure [] pop n q = do - v <- tryReadTQueue q + v <- tryReadTBQueue q case v of Nothing -> pure [] Just v' -> (v' :) <$> pop (n-1) q diff --git a/OpenTracing/Tags.hs b/OpenTracing/Tags.hs --- a/OpenTracing/Tags.hs +++ b/OpenTracing/Tags.hs @@ -2,12 +2,11 @@ {-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE PatternSynonyms #-} +{-# LANGUAGE RankNTypes #-} {-# LANGUAGE StrictData #-} {-# LANGUAGE TupleSections #-} {-# LANGUAGE ViewPatterns #-} -{-# OPTIONS_GHC -fno-warn-missing-pattern-synonym-signatures #-} - module OpenTracing.Tags ( Tags(fromTags) , Tag @@ -90,6 +89,7 @@ import qualified Data.HashMap.Strict as HashMap import Data.Int (Int64) import Data.Monoid (First) +import Data.String (IsString) import Data.Text (Text) import qualified Data.Text as Text import Data.Text.Encoding (decodeUtf8, encodeUtf8) @@ -140,6 +140,26 @@ getTagReify :: Getting (First b) Tag b -> Text -> Tags -> Maybe b getTagReify p k ts = getTag k ts >>= preview p . (k,) +pattern + ComponentKey + , DbInstanceKey + , DbStatementKey + , DbTypeKey + , DbUserKey + , ErrorKey + , HttpMethodKey + , HttpStatusCodeKey + , HttpUrlKey + , MessageBusDestinationKey + , PeerAddressKey + , PeerHostnameKey + , PeerIPv4Key + , PeerIPv6Key + , PeerPortKey + , PeerServiceKey + , SamplingPriorityKey + , SpanKindKey + :: forall a. (Eq a, IsString a) => a pattern ComponentKey = "component" pattern DbInstanceKey = "db.instance" @@ -165,6 +185,7 @@ (k, StringT v) | k == ComponentKey -> Just v _ -> Nothing +pattern Component :: Text -> Tag pattern Component v <- (preview _Component -> Just v) where Component v = review _Component v @@ -173,6 +194,7 @@ (k, StringT v) | k == DbInstanceKey -> Just v _ -> Nothing +pattern DbInstance :: Text -> Tag pattern DbInstance v <- (preview _DbInstance -> Just v) where DbInstance v = review _DbInstance v @@ -181,6 +203,7 @@ (k, StringT v) | k == DbStatementKey -> Just v _ -> Nothing +pattern DbStatement :: Text -> Tag pattern DbStatement v <- (preview _DbStatement -> Just v) where DbStatement v = review _DbStatement v @@ -189,6 +212,7 @@ (k, StringT v) | k == DbTypeKey -> Just v _ -> Nothing +pattern DbType :: Text -> Tag pattern DbType v <- (preview _DbType -> Just v) where DbType v = review _DbType v @@ -197,6 +221,7 @@ (k, StringT v) | k == DbUserKey -> Just v _ -> Nothing +pattern DbUser :: Text -> Tag pattern DbUser v <- (preview _DbUser -> Just v) where DbUser v = review _DbUser v @@ -205,6 +230,7 @@ (k, BoolT v) | k == ErrorKey -> Just v _ -> Nothing +pattern Error :: Bool -> Tag pattern Error v <- (preview _Error -> Just v) where Error v = review _Error v @@ -213,6 +239,7 @@ (k, StringT v) | k == HttpUrlKey -> Just v _ -> Nothing +pattern HttpUrl :: Text -> Tag pattern HttpUrl v <- (preview _HttpUrl -> Just v) where HttpUrl v = review _HttpUrl v @@ -221,6 +248,7 @@ (k, StringT v) | k == MessageBusDestinationKey -> Just v _ -> Nothing +pattern MessageBusDestination :: Text -> Tag pattern MessageBusDestination v <- (preview _MessageBusDestination -> Just v) where MessageBusDestination v = review _MessageBusDestination v @@ -229,6 +257,7 @@ (k, StringT v) | k == PeerAddressKey -> Just v _ -> Nothing +pattern PeerAddress :: Text -> Tag pattern PeerAddress v <- (preview _PeerAddress -> Just v) where PeerAddress v = review _PeerAddress v @@ -237,6 +266,7 @@ (k, StringT v) | k == PeerHostnameKey -> Just v _ -> Nothing +pattern PeerHostname :: Text -> Tag pattern PeerHostname v <- (preview _PeerHostname -> Just v) where PeerHostname v = review _PeerHostname v @@ -245,6 +275,7 @@ (k, StringT v) | k == PeerServiceKey -> Just v _ -> Nothing +pattern PeerService :: Text -> Tag pattern PeerService v <- (preview _PeerService -> Just v) where PeerService v = review _PeerService v @@ -254,6 +285,7 @@ either (const Nothing) (const (Just x)) $ parseMethod x _ -> Nothing +pattern HttpMethod :: Method -> Tag pattern HttpMethod v <- (preview _HttpMethod -> Just v) where HttpMethod v = review _HttpMethod v @@ -262,6 +294,7 @@ (k, IntT x) | k == HttpStatusCodeKey -> Just . toEnum . fromIntegral $ x _ -> Nothing +pattern HttpStatusCode :: Status -> Tag pattern HttpStatusCode v <- (preview _HttpStatusCode -> Just v) where HttpStatusCode v = review _HttpStatusCode v @@ -270,6 +303,7 @@ (k, StringT x) | k == PeerIPv4Key -> readMaybe (Text.unpack x) _ -> Nothing +pattern PeerIPv4 :: IPv4 -> Tag pattern PeerIPv4 v <- (preview _PeerIPv4 -> Just v) where PeerIPv4 v = review _PeerIPv4 v @@ -278,6 +312,7 @@ (k, StringT x) | k == PeerIPv6Key -> readMaybe (Text.unpack x) _ -> Nothing +pattern PeerIPv6 :: IPv6 -> Tag pattern PeerIPv6 v <- (preview _PeerIPv6 -> Just v) where PeerIPv6 v = review _PeerIPv6 v @@ -286,6 +321,7 @@ (k, IntT x) | k == PeerPortKey -> Just . toEnum . fromIntegral $ x _ -> Nothing +pattern PeerPort :: Port -> Tag pattern PeerPort v <- (preview _PeerPort -> Just v) where PeerPort v = review _PeerPort v @@ -294,6 +330,7 @@ (k, IntT x) | k == SamplingPriorityKey -> Just . fromIntegral $ x _ -> Nothing +pattern SamplingPriority :: Word8 -> Tag pattern SamplingPriority v <- (preview _SamplingPriority -> Just v) where SamplingPriority v = review _SamplingPriority v @@ -302,6 +339,7 @@ (k, StringT x) | k == SpanKindKey -> fromSpanKindLabel x _ -> Nothing +pattern SpanKind :: SpanKinds -> Tag pattern SpanKind v <- (preview _SpanKind -> Just v) where SpanKind = review _SpanKind diff --git a/opentracing.cabal b/opentracing.cabal --- a/opentracing.cabal +++ b/opentracing.cabal @@ -1,13 +1,13 @@ cabal-version: 2.2 name: opentracing -version: 0.2.2 +version: 0.3.0 synopsis: OpenTracing for Haskell homepage: https://github.com/kim/opentracing bug-reports: https://github.com/kim/opentracing/issues author: Kim Altintop -maintainer: Kim Altintop -copyright: Copyright (c) 2017-2018 Kim Altintop +maintainer: Kim Altintop +copyright: Copyright (c) 2017-2023 Kim Altintop license: Apache-2.0 license-file: LICENSE build-type: Simple @@ -49,7 +49,7 @@ build-depends: aeson >= 1.1 , async >= 2.1 - , base >= 4.9 && < 4.17 + , base >= 4.9 && < 4.19 , base64-bytestring >= 1.0 , bytestring >= 0.10 , case-insensitive >= 1.2