diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,13 @@ # Changelog for hs-opentelemetry-sdk +## 0.1.0.0 + +- Support new versions of dependencies. +- Windows: Replace POSIX-only functionality with a stub, so the package could be built at all (#114). +- Support `OTEL_SDK_DISABLED` (#148). +- Add Datadog as a known propagator (#117). +- Documentation improvements + ## 0.0.3.6 - Raise minimum version bounds for `random` to 1.2.0. This fixes duplicate ID generation issues in highly concurrent systems. @@ -30,7 +38,3 @@ ## 0.0.1.0 - Initial release - -## Unreleased changes - -- Documentation improvements diff --git a/hs-opentelemetry-sdk.cabal b/hs-opentelemetry-sdk.cabal --- a/hs-opentelemetry-sdk.cabal +++ b/hs-opentelemetry-sdk.cabal @@ -1,11 +1,11 @@ cabal-version: 1.22 --- This file has been generated from package.yaml by hpack version 0.35.2. +-- This file has been generated from package.yaml by hpack version 0.36.0. -- -- see: https://github.com/sol/hpack name: hs-opentelemetry-sdk -version: 0.0.3.6 +version: 0.1.0.0 synopsis: OpenTelemetry SDK for use in applications. description: Please see the README on GitHub at category: OpenTelemetry, Telemetry, Monitoring, Observability, Metrics @@ -13,7 +13,7 @@ bug-reports: https://github.com/iand675/hs-opentelemetry/issues author: Ian Duncan, Jade Lovelace maintainer: ian@iankduncan.com -copyright: 2023 Ian Duncan, Mercury Technologies +copyright: 2024 Ian Duncan, Mercury Technologies license: BSD3 license-file: LICENSE build-type: Simple @@ -30,7 +30,11 @@ library exposed-modules: OpenTelemetry.Processor.Batch + OpenTelemetry.Processor.Batch.LogRecord + OpenTelemetry.Processor.Batch.Span OpenTelemetry.Processor.Simple + OpenTelemetry.Processor.Simple.LogRecord + OpenTelemetry.Processor.Simple.Span OpenTelemetry.Resource.Host.Detector OpenTelemetry.Resource.OperatingSystem.Detector OpenTelemetry.Resource.Process.Detector @@ -45,7 +49,11 @@ , OpenTelemetry.Baggage , OpenTelemetry.Context , OpenTelemetry.Context.ThreadLocal + , OpenTelemetry.Exporter.Span + , OpenTelemetry.Exporter.LogRecord , OpenTelemetry.Exporter + , OpenTelemetry.Processor.Span + , OpenTelemetry.Processor.LogRecord , OpenTelemetry.Processor , OpenTelemetry.Propagator , OpenTelemetry.Resource @@ -73,22 +81,34 @@ async , base >=4.7 && <5 , bytestring - , hs-opentelemetry-api >=0.0.3 && <0.2 - , hs-opentelemetry-exporter-otlp ==0.0.1.* - , hs-opentelemetry-propagator-b3 ==0.0.1.* - , hs-opentelemetry-propagator-w3c ==0.0.1.* + , hs-opentelemetry-api ==0.2.* + , hs-opentelemetry-exporter-otlp ==0.1.* + , hs-opentelemetry-propagator-b3 >=0.0.1 && <0.1 + , hs-opentelemetry-propagator-datadog >=0.0.0 && <0.1 + , hs-opentelemetry-propagator-w3c >=0.0.1 && <0.1 , http-types , network-bsd , random >=1.2.0 , stm , text - , transformers , unagi-chan - , unix + , unix-compat >=0.7.1 , unordered-containers , vector , vector-builder default-language: Haskell2010 + if os(windows) + other-modules: + OpenTelemetry.Platform + hs-source-dirs: + src/platform/windows + else + other-modules: + OpenTelemetry.Platform + hs-source-dirs: + src/platform/unix + build-depends: + unix test-suite hs-opentelemetry-sdk-test type: exitcode-stdio-1.0 @@ -103,25 +123,11 @@ test ghc-options: -threaded -rtsopts -with-rtsopts=-N build-depends: - async - , base >=4.7 && <5 - , bytestring + base >=4.7 && <5 , clock , hs-opentelemetry-api - , hs-opentelemetry-exporter-otlp ==0.0.1.* - , hs-opentelemetry-propagator-b3 ==0.0.1.* - , hs-opentelemetry-propagator-w3c ==0.0.1.* , hs-opentelemetry-sdk , hspec - , http-types - , network-bsd - , random >=1.2.0 - , stm , text - , transformers - , unagi-chan - , unix , unordered-containers - , vector - , vector-builder default-language: Haskell2010 diff --git a/src/OpenTelemetry/Processor/Batch.hs b/src/OpenTelemetry/Processor/Batch.hs --- a/src/OpenTelemetry/Processor/Batch.hs +++ b/src/OpenTelemetry/Processor/Batch.hs @@ -1,372 +1,6 @@ -{-# LANGUAGE RecordWildCards #-} - ------------------------------------------------------------------------------ - ------------------------------------------------------------------------------ - -{- | - Module : OpenTelemetry.Processor.Batch - Copyright : (c) Ian Duncan, 2021 - License : BSD-3 - Description : Performant exporting of spans in time & space-bounded batches. - Maintainer : Ian Duncan - Stability : experimental - Portability : non-portable (GHC extensions) - - This is an implementation of the Span Processor which create batches of finished spans and passes the export-friendly span data representations to the configured Exporter. --} module OpenTelemetry.Processor.Batch ( - BatchTimeoutConfig (..), - batchTimeoutConfig, - batchProcessor, - -- , BatchProcessorOperations + module OpenTelemetry.Processor.Batch.Span, ) where -import Control.Concurrent (rtsSupportsBoundThreads) -import Control.Concurrent.Async -import Control.Concurrent.STM -import Control.Exception -import Control.Monad -import Control.Monad.IO.Class -import Data.HashMap.Strict (HashMap) -import qualified Data.HashMap.Strict as HashMap -import Data.IORef (atomicModifyIORef', newIORef, readIORef) -import Data.Vector (Vector) -import OpenTelemetry.Exporter (Exporter) -import qualified OpenTelemetry.Exporter as Exporter -import OpenTelemetry.Processor -import OpenTelemetry.Trace.Core -import VectorBuilder.Builder as Builder -import VectorBuilder.Vector as Builder - - --- | Configurable options for batch exporting frequence and size -data BatchTimeoutConfig = BatchTimeoutConfig - { maxQueueSize :: Int - -- ^ The maximum queue size. After the size is reached, spans are dropped. - , scheduledDelayMillis :: Int - -- ^ The delay interval in milliseconds between two consective exports. - -- The default value is 5000. - , exportTimeoutMillis :: Int - -- ^ How long the export can run before it is cancelled. - -- The default value is 30000. - , maxExportBatchSize :: Int - -- ^ The maximum batch size of every export. It must be - -- smaller or equal to 'maxQueueSize'. The default value is 512. - } - deriving (Show) - - --- | Default configuration values -batchTimeoutConfig :: BatchTimeoutConfig -batchTimeoutConfig = - BatchTimeoutConfig - { maxQueueSize = 1024 - , scheduledDelayMillis = 5000 - , exportTimeoutMillis = 30000 - , maxExportBatchSize = 512 - } - - --- type BatchProcessorOperations = () - --- A multi-producer single-consumer green/blue buffer. --- Write requests that cannot fit in the live chunk will be dropped --- --- TODO, would be cool to use AtomicCounters for this if possible --- data GreenBlueBuffer a = GreenBlueBuffer --- { gbReadSection :: !(TVar Word) --- , gbWriteGreenOrBlue :: !(TVar Word) --- , gbPendingWrites :: !(TVar Word) --- , gbSectionSize :: !Int --- , gbVector :: !(M.IOVector a) --- } - -{- brainstorm: Single Word64 state sketch - - 63 (high bit): green or blue - 32-62: read section - 0-32: write count --} - -{- - -Green - 512 512 512 512 -\|---------|---------|---------|---------| - 0 1 2 3 - -Blue - 512 512 512 512 -\|---------|---------|---------|---------| - 0 1 2 3 - -The current read section denotes one chunk of length gbSize, which gets flushed -to the span exporter. Once the vector has been copied for export, gbReadSection -will be incremented. - --} - --- newGreenBlueBuffer --- :: Int -- Max queue size (2048) --- -> Int -- Export batch size (512) --- -> IO (GreenBlueBuffer a) --- newGreenBlueBuffer maxQueueSize batchSize = do --- let logBase2 = finiteBitSize maxQueueSize - 1 - countLeadingZeros maxQueueSize - --- let closestFittingPowerOfTwo = 2 * if (1 `shiftL` logBase2) == maxQueueSize --- then maxQueueSize --- else 1 `shiftL` (logBase2 + 1) - --- readSection <- newTVarIO 0 --- writeSection <- newTVarIO 0 --- writeCount <- newTVarIO 0 --- buf <- M.new closestFittingPowerOfTwo --- pure $ GreenBlueBuffer --- { gbSize = maxQueueSize --- , gbVector = buf --- , gbReadSection = readSection --- , gbPendingWrites = writeCount --- } - --- isEmpty :: GreenBlueBuffer a -> STM Bool --- isEmpty = do --- c <- readTVar gbPendingWrites --- pure (c == 0) - --- data InsertResult = ValueDropped | ValueInserted - --- tryInsert :: GreenBlueBuffer a -> a -> IO InsertResult --- tryInsert GreenBlueBuffer{..} x = atomically $ do --- c <- readTVar gbPendingWrites --- if c == gbMaxLength --- then pure ValueDropped --- else do --- greenOrBlue <- readTVar gbWriteGreenOrBlue --- let i = c + ((M.length gbVector `shiftR` 1) `shiftL` (greenOrBlue `mod` 2)) --- M.write gbVector i x --- writeTVar gbPendingWrites (c + 1) --- pure ValueInserted - --- Caution, single writer means that this can't be called concurrently --- consumeChunk :: GreenBlueBuffer a -> IO (V.Vector a) --- consumeChunk GreenBlueBuffer{..} = atomically $ do --- r <- readTVar gbReadSection --- w <- readTVar gbWriteSection --- c <- readTVar gbPendingWrites --- when (r == w) $ do --- modifyTVar gbWriteSection (+ 1) --- setTVar gbPendingWrites 0 --- -- TODO slice and freeze appropriate section --- M.slice (gbSectionSize * (r .&. gbSectionMask) - --- TODO, counters for dropped spans, exported spans - -data BoundedMap a = BoundedMap - { itemBounds :: !Int - , itemMaxExportBounds :: !Int - , itemCount :: !Int - , itemMap :: HashMap InstrumentationLibrary (Builder.Builder a) - } - - -boundedMap :: Int -> Int -> BoundedMap a -boundedMap bounds exportBounds = BoundedMap bounds exportBounds 0 mempty - - -push :: ImmutableSpan -> BoundedMap ImmutableSpan -> Maybe (BoundedMap ImmutableSpan) -push s m = - if itemCount m + 1 >= itemBounds m - then Nothing - else - Just $! - m - { itemCount = itemCount m + 1 - , itemMap = - HashMap.insertWith - (<>) - (tracerName $ spanTracer s) - (Builder.singleton s) - $ itemMap m - } - - -buildExport :: BoundedMap a -> (BoundedMap a, HashMap InstrumentationLibrary (Vector a)) -buildExport m = - ( m {itemCount = 0, itemMap = mempty} - , Builder.build <$> itemMap m - ) - - -data ProcessorMessage = ScheduledFlush | MaxExportFlush | Shutdown - - --- note: [Unmasking Asyncs] --- --- It is possible to create unkillable asyncs. Behold: --- --- ``` --- a <- uninterruptibleMask_ $ do --- async $ do --- forever $ do --- threadDelay 10_000 --- putStrLn "still alive" --- cancel a --- ``` --- --- The prior code block will never successfully cancel `a` and will be --- blocked forever. The reason is that `cancel` sends an async exception to --- the thread performing the action, but the `uninterruptibleMask` state is --- inherited by the forked thread. This means that *no async exceptions* --- can reach it, and `cancel` will therefore run forever. --- --- This also affects `timeout`, which uses an async exception to kill the --- running job. If the action is done in an uninterruptible masked state, --- then the timeout will not successfully kill the running action. - -{- | - The batch processor accepts spans and places them into batches. Batching helps better compress the data and reduce the number of outgoing connections - required to transmit the data. This processor supports both size and time based batching. - - NOTE: this function requires the program be compiled with the @-threaded@ GHC - option and will throw an error if this is not the case. --} -batchProcessor :: (MonadIO m) => BatchTimeoutConfig -> Exporter ImmutableSpan -> m Processor -batchProcessor BatchTimeoutConfig {..} exporter = liftIO $ do - unless rtsSupportsBoundThreads $ error "The hs-opentelemetry batch processor does not work without the -threaded GHC flag!" - batch <- newIORef $ boundedMap maxQueueSize maxExportBatchSize - workSignal <- newEmptyTMVarIO - shutdownSignal <- newEmptyTMVarIO - let publish batchToProcess = mask_ $ do - -- we mask async exceptions in this, so that a buggy exporter that - -- catches async exceptions won't swallow them. since we use - -- an interruptible mask, blocking calls can still be killed, like - -- `threadDelay` or `putMVar` or most file I/O operations. - -- - -- if we've received a shutdown, then we should be expecting - -- a `cancel` anytime now. - Exporter.exporterExport exporter batchToProcess - - let flushQueueImmediately ret = do - batchToProcess <- atomicModifyIORef' batch buildExport - if null batchToProcess - then do - pure ret - else do - ret' <- publish batchToProcess - flushQueueImmediately ret' - - let waiting = do - delay <- registerDelay (millisToMicros scheduledDelayMillis) - atomically $ do - msum - -- Flush every scheduled delay time, when we've reached the max export size, or when the shutdown signal is received. - [ ScheduledFlush <$ do - continue <- readTVar delay - check continue - , MaxExportFlush <$ takeTMVar workSignal - , Shutdown <$ takeTMVar shutdownSignal - ] - - let workerAction = do - req <- waiting - batchToProcess <- atomicModifyIORef' batch buildExport - res <- publish batchToProcess - - -- if we were asked to shutdown, stop waiting and flush it all out - case req of - Shutdown -> - flushQueueImmediately res - _ -> - workerAction - -- see note [Unmasking Asyncs] - worker <- asyncWithUnmask $ \unmask -> unmask workerAction - - pure $ - Processor - { processorOnStart = \_ _ -> pure () - , processorOnEnd = \s -> do - span_ <- readIORef s - appendFailedOrExportNeeded <- atomicModifyIORef' batch $ \builder -> - case push span_ builder of - Nothing -> (builder, True) - Just b' -> - if itemCount b' >= itemMaxExportBounds b' - then -- If the batch has grown to the maximum export size, prompt the worker to export it. - (b', True) - else (b', False) - when appendFailedOrExportNeeded $ void $ atomically $ tryPutTMVar workSignal () - , processorForceFlush = void $ atomically $ tryPutTMVar workSignal () - , -- TODO where to call restore, if anywhere? - processorShutdown = - asyncWithUnmask $ \unmask -> unmask $ do - -- we call asyncWithUnmask here because the shutdown action is - -- likely to happen inside of a `finally` or `bracket`. the - -- @safe-exceptions@ pattern (followed by unliftio as well) - -- will use uninterruptibleMask in an exception cleanup. the - -- uninterruptibleMask state means that the `timeout` call - -- below will never exit, because `wait worker` will be in the - -- `uninterruptibleMasked` state, and the timeout async - -- exception will not be delivered. - -- - -- see note [Unmasking Asyncs] - mask $ \_restore -> do - -- is it a little silly that we unmask and remask? seems - -- silly! but the `mask` here is doing an interruptible mask. - -- which means that async exceptions can still be delivered - -- if a process is blocking. - - -- flush remaining messages and signal the worker to shutdown - void $ atomically $ putTMVar shutdownSignal () - - -- gracefully wait for the worker to stop. we may be in - -- a `bracket` or responding to an async exception, so we - -- must be very careful not to wait too long. the following - -- STM action will block, so we'll be susceptible to an async - -- exception. - delay <- registerDelay (millisToMicros exportTimeoutMillis) - shutdownResult <- - atomically $ - msum - [ Just <$> waitCatchSTM worker - , Nothing <$ do - shouldStop <- readTVar delay - check shouldStop - ] - - -- make sure the worker comes down if we timed out. - cancel worker - -- TODO, not convinced we should shut down processor here - - pure $ case shutdownResult of - Nothing -> - ShutdownTimeout - Just er -> - case er of - Left _ -> - ShutdownFailure - Right _ -> - ShutdownSuccess - } - where - millisToMicros = (* 1000) - -{- -buffer <- newGreenBlueBuffer _ _ -batchProcessorAction <- async $ forever $ do - -- It would be nice to do an immediate send when possible - chunk <- if (sendDelay == 0) - else consumeChunk - then threadDelay sendDelay >> consumeChunk - timeout _ $ export exporter chunk -pure $ Processor - { onStart = \_ _ -> pure () - , onEnd = \s -> void $ tryInsert buffer s - , shutdown = do - gracefullyShutdownBatchProcessor +import OpenTelemetry.Processor.Batch.Span - , forceFlush = pure () - } -where - sendDelay = scheduledDelayMilis * 1_000 --} diff --git a/src/OpenTelemetry/Processor/Batch/LogRecord.hs b/src/OpenTelemetry/Processor/Batch/LogRecord.hs new file mode 100644 --- /dev/null +++ b/src/OpenTelemetry/Processor/Batch/LogRecord.hs @@ -0,0 +1,2 @@ +module OpenTelemetry.Processor.Batch.LogRecord () where + diff --git a/src/OpenTelemetry/Processor/Batch/Span.hs b/src/OpenTelemetry/Processor/Batch/Span.hs new file mode 100644 --- /dev/null +++ b/src/OpenTelemetry/Processor/Batch/Span.hs @@ -0,0 +1,372 @@ +{-# LANGUAGE RecordWildCards #-} + +----------------------------------------------------------------------------- + +----------------------------------------------------------------------------- + +{- | + Module : OpenTelemetry.Processor.Batch.Span + Copyright : (c) Ian Duncan, 2021 + License : BSD-3 + Description : Performant exporting of spans in time & space-bounded batches. + Maintainer : Ian Duncan + Stability : experimental + Portability : non-portable (GHC extensions) + + This is an implementation of the Span Processor which create batches of finished spans and passes the export-friendly span data representations to the configured Exporter. +-} +module OpenTelemetry.Processor.Batch.Span ( + BatchTimeoutConfig (..), + batchTimeoutConfig, + batchProcessor, + -- , BatchProcessorOperations +) where + +import Control.Concurrent (rtsSupportsBoundThreads) +import Control.Concurrent.Async +import Control.Concurrent.STM +import Control.Exception +import Control.Monad +import Control.Monad.IO.Class +import Data.HashMap.Strict (HashMap) +import qualified Data.HashMap.Strict as HashMap +import Data.IORef (atomicModifyIORef', newIORef, readIORef) +import Data.Vector (Vector) +import OpenTelemetry.Exporter.Span (SpanExporter) +import qualified OpenTelemetry.Exporter.Span as SpanExporter +import OpenTelemetry.Processor.Span +import OpenTelemetry.Trace.Core +import VectorBuilder.Builder as Builder +import VectorBuilder.Vector as Builder + + +-- | Configurable options for batch exporting frequence and size +data BatchTimeoutConfig = BatchTimeoutConfig + { maxQueueSize :: Int + -- ^ The maximum queue size. After the size is reached, spans are dropped. + , scheduledDelayMillis :: Int + -- ^ The delay interval in milliseconds between two consective exports. + -- The default value is 5000. + , exportTimeoutMillis :: Int + -- ^ How long the export can run before it is cancelled. + -- The default value is 30000. + , maxExportBatchSize :: Int + -- ^ The maximum batch size of every export. It must be + -- smaller or equal to 'maxQueueSize'. The default value is 512. + } + deriving (Show) + + +-- | Default configuration values +batchTimeoutConfig :: BatchTimeoutConfig +batchTimeoutConfig = + BatchTimeoutConfig + { maxQueueSize = 1024 + , scheduledDelayMillis = 5000 + , exportTimeoutMillis = 30000 + , maxExportBatchSize = 512 + } + + +-- type BatchProcessorOperations = () + +-- A multi-producer single-consumer green/blue buffer. +-- Write requests that cannot fit in the live chunk will be dropped +-- +-- TODO, would be cool to use AtomicCounters for this if possible +-- data GreenBlueBuffer a = GreenBlueBuffer +-- { gbReadSection :: !(TVar Word) +-- , gbWriteGreenOrBlue :: !(TVar Word) +-- , gbPendingWrites :: !(TVar Word) +-- , gbSectionSize :: !Int +-- , gbVector :: !(M.IOVector a) +-- } + +{- brainstorm: Single Word64 state sketch + + 63 (high bit): green or blue + 32-62: read section + 0-32: write count +-} + +{- + +Green + 512 512 512 512 +\|---------|---------|---------|---------| + 0 1 2 3 + +Blue + 512 512 512 512 +\|---------|---------|---------|---------| + 0 1 2 3 + +The current read section denotes one chunk of length gbSize, which gets flushed +to the span exporter. Once the vector has been copied for export, gbReadSection +will be incremented. + +-} + +-- newGreenBlueBuffer +-- :: Int -- Max queue size (2048) +-- -> Int -- Export batch size (512) +-- -> IO (GreenBlueBuffer a) +-- newGreenBlueBuffer maxQueueSize batchSize = do +-- let logBase2 = finiteBitSize maxQueueSize - 1 - countLeadingZeros maxQueueSize + +-- let closestFittingPowerOfTwo = 2 * if (1 `shiftL` logBase2) == maxQueueSize +-- then maxQueueSize +-- else 1 `shiftL` (logBase2 + 1) + +-- readSection <- newTVarIO 0 +-- writeSection <- newTVarIO 0 +-- writeCount <- newTVarIO 0 +-- buf <- M.new closestFittingPowerOfTwo +-- pure $ GreenBlueBuffer +-- { gbSize = maxQueueSize +-- , gbVector = buf +-- , gbReadSection = readSection +-- , gbPendingWrites = writeCount +-- } + +-- isEmpty :: GreenBlueBuffer a -> STM Bool +-- isEmpty = do +-- c <- readTVar gbPendingWrites +-- pure (c == 0) + +-- data InsertResult = ValueDropped | ValueInserted + +-- tryInsert :: GreenBlueBuffer a -> a -> IO InsertResult +-- tryInsert GreenBlueBuffer{..} x = atomically $ do +-- c <- readTVar gbPendingWrites +-- if c == gbMaxLength +-- then pure ValueDropped +-- else do +-- greenOrBlue <- readTVar gbWriteGreenOrBlue +-- let i = c + ((M.length gbVector `shiftR` 1) `shiftL` (greenOrBlue `mod` 2)) +-- M.write gbVector i x +-- writeTVar gbPendingWrites (c + 1) +-- pure ValueInserted + +-- Caution, single writer means that this can't be called concurrently +-- consumeChunk :: GreenBlueBuffer a -> IO (V.Vector a) +-- consumeChunk GreenBlueBuffer{..} = atomically $ do +-- r <- readTVar gbReadSection +-- w <- readTVar gbWriteSection +-- c <- readTVar gbPendingWrites +-- when (r == w) $ do +-- modifyTVar gbWriteSection (+ 1) +-- setTVar gbPendingWrites 0 +-- -- TODO slice and freeze appropriate section +-- M.slice (gbSectionSize * (r .&. gbSectionMask) + +-- TODO, counters for dropped spans, exported spans + +data BoundedMap a = BoundedMap + { itemBounds :: !Int + , itemMaxExportBounds :: !Int + , itemCount :: !Int + , itemMap :: HashMap InstrumentationLibrary (Builder.Builder a) + } + + +boundedMap :: Int -> Int -> BoundedMap a +boundedMap bounds exportBounds = BoundedMap bounds exportBounds 0 mempty + + +push :: ImmutableSpan -> BoundedMap ImmutableSpan -> Maybe (BoundedMap ImmutableSpan) +push s m = + if itemCount m + 1 >= itemBounds m + then Nothing + else + Just $! + m + { itemCount = itemCount m + 1 + , itemMap = + HashMap.insertWith + (<>) + (tracerName $ spanTracer s) + (Builder.singleton s) + $ itemMap m + } + + +buildExport :: BoundedMap a -> (BoundedMap a, HashMap InstrumentationLibrary (Vector a)) +buildExport m = + ( m {itemCount = 0, itemMap = mempty} + , Builder.build <$> itemMap m + ) + + +data ProcessorMessage = ScheduledFlush | MaxExportFlush | Shutdown + + +-- note: [Unmasking Asyncs] +-- +-- It is possible to create unkillable asyncs. Behold: +-- +-- ``` +-- a <- uninterruptibleMask_ $ do +-- async $ do +-- forever $ do +-- threadDelay 10_000 +-- putStrLn "still alive" +-- cancel a +-- ``` +-- +-- The prior code block will never successfully cancel `a` and will be +-- blocked forever. The reason is that `cancel` sends an async exception to +-- the thread performing the action, but the `uninterruptibleMask` state is +-- inherited by the forked thread. This means that *no async exceptions* +-- can reach it, and `cancel` will therefore run forever. +-- +-- This also affects `timeout`, which uses an async exception to kill the +-- running job. If the action is done in an uninterruptible masked state, +-- then the timeout will not successfully kill the running action. + +{- | + The batch processor accepts spans and places them into batches. Batching helps better compress the data and reduce the number of outgoing connections + required to transmit the data. This processor supports both size and time based batching. + + NOTE: this function requires the program be compiled with the @-threaded@ GHC + option and will throw an error if this is not the case. +-} +batchProcessor :: (MonadIO m) => BatchTimeoutConfig -> SpanExporter -> m SpanProcessor +batchProcessor BatchTimeoutConfig {..} exporter = liftIO $ do + unless rtsSupportsBoundThreads $ error "The hs-opentelemetry batch processor does not work without the -threaded GHC flag!" + batch <- newIORef $ boundedMap maxQueueSize maxExportBatchSize + workSignal <- newEmptyTMVarIO + shutdownSignal <- newEmptyTMVarIO + let publish batchToProcess = mask_ $ do + -- we mask async exceptions in this, so that a buggy exporter that + -- catches async exceptions won't swallow them. since we use + -- an interruptible mask, blocking calls can still be killed, like + -- `threadDelay` or `putMVar` or most file I/O operations. + -- + -- if we've received a shutdown, then we should be expecting + -- a `cancel` anytime now. + SpanExporter.spanExporterExport exporter batchToProcess + + let flushQueueImmediately ret = do + batchToProcess <- atomicModifyIORef' batch buildExport + if null batchToProcess + then do + pure ret + else do + ret' <- publish batchToProcess + flushQueueImmediately ret' + + let waiting = do + delay <- registerDelay (millisToMicros scheduledDelayMillis) + atomically $ do + msum + -- Flush every scheduled delay time, when we've reached the max export size, or when the shutdown signal is received. + [ ScheduledFlush <$ do + continue <- readTVar delay + check continue + , MaxExportFlush <$ takeTMVar workSignal + , Shutdown <$ takeTMVar shutdownSignal + ] + + let workerAction = do + req <- waiting + batchToProcess <- atomicModifyIORef' batch buildExport + res <- publish batchToProcess + + -- if we were asked to shutdown, stop waiting and flush it all out + case req of + Shutdown -> + flushQueueImmediately res + _ -> + workerAction + -- see note [Unmasking Asyncs] + worker <- asyncWithUnmask $ \unmask -> unmask workerAction + + pure $ + SpanProcessor + { spanProcessorOnStart = \_ _ -> pure () + , spanProcessorOnEnd = \s -> do + span_ <- readIORef s + appendFailedOrExportNeeded <- atomicModifyIORef' batch $ \builder -> + case push span_ builder of + Nothing -> (builder, True) + Just b' -> + if itemCount b' >= itemMaxExportBounds b' + then -- If the batch has grown to the maximum export size, prompt the worker to export it. + (b', True) + else (b', False) + when appendFailedOrExportNeeded $ void $ atomically $ tryPutTMVar workSignal () + , spanProcessorForceFlush = void $ atomically $ tryPutTMVar workSignal () + , -- TODO where to call restore, if anywhere? + spanProcessorShutdown = + asyncWithUnmask $ \unmask -> unmask $ do + -- we call asyncWithUnmask here because the shutdown action is + -- likely to happen inside of a `finally` or `bracket`. the + -- @safe-exceptions@ pattern (followed by unliftio as well) + -- will use uninterruptibleMask in an exception cleanup. the + -- uninterruptibleMask state means that the `timeout` call + -- below will never exit, because `wait worker` will be in the + -- `uninterruptibleMasked` state, and the timeout async + -- exception will not be delivered. + -- + -- see note [Unmasking Asyncs] + mask $ \_restore -> do + -- is it a little silly that we unmask and remask? seems + -- silly! but the `mask` here is doing an interruptible mask. + -- which means that async exceptions can still be delivered + -- if a process is blocking. + + -- flush remaining messages and signal the worker to shutdown + void $ atomically $ putTMVar shutdownSignal () + + -- gracefully wait for the worker to stop. we may be in + -- a `bracket` or responding to an async exception, so we + -- must be very careful not to wait too long. the following + -- STM action will block, so we'll be susceptible to an async + -- exception. + delay <- registerDelay (millisToMicros exportTimeoutMillis) + shutdownResult <- + atomically $ + msum + [ Just <$> waitCatchSTM worker + , Nothing <$ do + shouldStop <- readTVar delay + check shouldStop + ] + + -- make sure the worker comes down if we timed out. + cancel worker + -- TODO, not convinced we should shut down processor here + + pure $ case shutdownResult of + Nothing -> + ShutdownTimeout + Just er -> + case er of + Left _ -> + ShutdownFailure + Right _ -> + ShutdownSuccess + } + where + millisToMicros = (* 1000) + +{- +buffer <- newGreenBlueBuffer _ _ +batchProcessorAction <- async $ forever $ do + -- It would be nice to do an immediate send when possible + chunk <- if (sendDelay == 0) + else consumeChunk + then threadDelay sendDelay >> consumeChunk + timeout _ $ export exporter chunk +pure $ Processor + { onStart = \_ _ -> pure () + , onEnd = \s -> void $ tryInsert buffer s + , shutdown = do + gracefullyShutdownBatchProcessor + + , forceFlush = pure () + } +where + sendDelay = scheduledDelayMilis * 1_000 +-} diff --git a/src/OpenTelemetry/Processor/Simple.hs b/src/OpenTelemetry/Processor/Simple.hs --- a/src/OpenTelemetry/Processor/Simple.hs +++ b/src/OpenTelemetry/Processor/Simple.hs @@ -1,64 +1,6 @@ -{-# LANGUAGE RecordWildCards #-} -{-# LANGUAGE ScopedTypeVariables #-} - module OpenTelemetry.Processor.Simple ( - SimpleProcessorConfig (..), - simpleProcessor, + module OpenTelemetry.Processor.Simple.Span, ) where -import Control.Concurrent.Async -import Control.Concurrent.Chan.Unagi -import Control.Exception -import Control.Monad -import qualified Data.HashMap.Strict as HashMap -import Data.IORef -import qualified OpenTelemetry.Exporter as Exporter -import OpenTelemetry.Processor -import OpenTelemetry.Trace.Core (ImmutableSpan, spanTracer, tracerName) - - -newtype SimpleProcessorConfig = SimpleProcessorConfig - { exporter :: Exporter.Exporter ImmutableSpan - -- ^ The exporter where the spans are pushed. - } - - -{- | This is an implementation of SpanProcessor which passes finished spans - and passes the export-friendly span data representation to the configured SpanExporter, - as soon as they are finished. - - @since 0.0.1.0 --} -simpleProcessor :: SimpleProcessorConfig -> IO Processor -simpleProcessor SimpleProcessorConfig {..} = do - (inChan :: InChan (IORef ImmutableSpan), outChan :: OutChan (IORef ImmutableSpan)) <- newChan - exportWorker <- async $ forever $ do - -- TODO, masking vs bracket here, not sure what's the right choice - spanRef <- readChanOnException outChan (>>= writeChan inChan) - span_ <- readIORef spanRef - mask_ (exporter `Exporter.exporterExport` HashMap.singleton (tracerName $ spanTracer span_) (pure span_)) +import OpenTelemetry.Processor.Simple.Span - pure $ - Processor - { processorOnStart = \_ _ -> pure () - , processorOnEnd = writeChan inChan - , processorShutdown = async $ mask $ \restore -> do - cancel exportWorker - -- TODO handle timeouts - restore $ do - -- TODO, not convinced we should shut down processor here - shutdownProcessor outChan `finally` Exporter.exporterShutdown exporter - pure ShutdownSuccess - , processorForceFlush = pure () - } - where - shutdownProcessor :: OutChan (IORef ImmutableSpan) -> IO () - shutdownProcessor outChan = do - (Element m, _) <- tryReadChan outChan - mSpan <- m - case mSpan of - Nothing -> pure () - Just spanRef -> do - span_ <- readIORef spanRef - _ <- exporter `Exporter.exporterExport` HashMap.singleton (tracerName $ spanTracer span_) (pure span_) - shutdownProcessor outChan diff --git a/src/OpenTelemetry/Processor/Simple/LogRecord.hs b/src/OpenTelemetry/Processor/Simple/LogRecord.hs new file mode 100644 --- /dev/null +++ b/src/OpenTelemetry/Processor/Simple/LogRecord.hs @@ -0,0 +1,2 @@ +module OpenTelemetry.Processor.Simple.LogRecord () where + diff --git a/src/OpenTelemetry/Processor/Simple/Span.hs b/src/OpenTelemetry/Processor/Simple/Span.hs new file mode 100644 --- /dev/null +++ b/src/OpenTelemetry/Processor/Simple/Span.hs @@ -0,0 +1,64 @@ +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE ScopedTypeVariables #-} + +module OpenTelemetry.Processor.Simple.Span ( + SimpleProcessorConfig (..), + simpleProcessor, +) where + +import Control.Concurrent.Async +import Control.Concurrent.Chan.Unagi +import Control.Exception +import Control.Monad +import qualified Data.HashMap.Strict as HashMap +import Data.IORef +import qualified OpenTelemetry.Exporter.Span as SpanExporter +import OpenTelemetry.Processor.Span +import OpenTelemetry.Trace.Core (ImmutableSpan, spanTracer, tracerName) + + +newtype SimpleProcessorConfig = SimpleProcessorConfig + { spanExporter :: SpanExporter.SpanExporter + -- ^ The exporter where the spans are pushed. + } + + +{- | This is an implementation of SpanProcessor which passes finished spans + and passes the export-friendly span data representation to the configured SpanExporter, + as soon as they are finished. + + @since 0.0.1.0 +-} +simpleProcessor :: SimpleProcessorConfig -> IO SpanProcessor +simpleProcessor SimpleProcessorConfig {..} = do + (inChan :: InChan (IORef ImmutableSpan), outChan :: OutChan (IORef ImmutableSpan)) <- newChan + exportWorker <- async $ forever $ do + -- TODO, masking vs bracket here, not sure what's the right choice + spanRef <- readChanOnException outChan (>>= writeChan inChan) + span_ <- readIORef spanRef + mask_ (spanExporter `SpanExporter.spanExporterExport` HashMap.singleton (tracerName $ spanTracer span_) (pure span_)) + + pure $ + SpanProcessor + { spanProcessorOnStart = \_ _ -> pure () + , spanProcessorOnEnd = writeChan inChan + , spanProcessorShutdown = async $ mask $ \restore -> do + cancel exportWorker + -- TODO handle timeouts + restore $ do + -- TODO, not convinced we should shut down processor here + shutdownProcessor outChan `finally` SpanExporter.spanExporterShutdown spanExporter + pure ShutdownSuccess + , spanProcessorForceFlush = pure () + } + where + shutdownProcessor :: OutChan (IORef ImmutableSpan) -> IO () + shutdownProcessor outChan = do + (Element m, _) <- tryReadChan outChan + mSpan <- m + case mSpan of + Nothing -> pure () + Just spanRef -> do + span_ <- readIORef spanRef + _ <- spanExporter `SpanExporter.spanExporterExport` HashMap.singleton (tracerName $ spanTracer span_) (pure span_) + shutdownProcessor outChan diff --git a/src/OpenTelemetry/Resource/Process/Detector.hs b/src/OpenTelemetry/Resource/Process/Detector.hs --- a/src/OpenTelemetry/Resource/Process/Detector.hs +++ b/src/OpenTelemetry/Resource/Process/Detector.hs @@ -1,18 +1,16 @@ module OpenTelemetry.Resource.Process.Detector where -import Control.Exception (throwIO, try) import qualified Data.Text as T import Data.Version +import OpenTelemetry.Platform (tryGetUser) import OpenTelemetry.Resource.Process import System.Environment ( getArgs, getExecutablePath, getProgName, ) -import System.IO.Error import System.Info -import System.Posix.Process (getProcessID) -import System.Posix.User (getEffectiveUserName) +import System.PosixCompat.Process (getProcessID) {- | Create a 'Process' 'Resource' based off of the current process' knowledge @@ -30,17 +28,6 @@ <*> pure Nothing <*> (Just . map T.pack <$> getArgs) <*> tryGetUser - - -tryGetUser :: IO (Maybe T.Text) -tryGetUser = do - eResult <- try getEffectiveUserName - case eResult of - Left err -> - if isDoesNotExistError err - then pure Nothing - else throwIO err - Right ok -> pure $ Just $ T.pack ok {- | A 'ProcessRuntime' 'Resource' populated with the current process' knoweldge diff --git a/src/OpenTelemetry/Trace.hs b/src/OpenTelemetry/Trace.hs --- a/src/OpenTelemetry/Trace.hs +++ b/src/OpenTelemetry/Trace.hs @@ -3,6 +3,7 @@ {-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RankNTypes #-} +{-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeOperators #-} ----------------------------------------------------------------------------- @@ -107,6 +108,7 @@ tracerOptions, HasTracer (..), InstrumentationLibrary (..), + detectInstrumentationLibrary, -- * 'Span' operations Span, @@ -115,6 +117,7 @@ SpanArguments (..), SpanKind (..), NewLink (..), + addLink, inSpan', updateName, addAttribute, @@ -160,12 +163,14 @@ import OpenTelemetry.Baggage (decodeBaggageHeader) import qualified OpenTelemetry.Baggage as Baggage import OpenTelemetry.Context (Context) -import OpenTelemetry.Exporter (Exporter) -import OpenTelemetry.Exporter.OTLP (loadExporterEnvironmentVariables, otlpExporter) -import OpenTelemetry.Processor (Processor) -import OpenTelemetry.Processor.Batch (BatchTimeoutConfig (..), batchProcessor, batchTimeoutConfig) +import OpenTelemetry.Environment +import OpenTelemetry.Exporter.OTLP.Span (loadExporterEnvironmentVariables, otlpExporter) +import OpenTelemetry.Exporter.Span (SpanExporter) +import OpenTelemetry.Processor.Batch.Span (BatchTimeoutConfig (..), batchProcessor, batchTimeoutConfig) +import OpenTelemetry.Processor.Span (SpanProcessor) import OpenTelemetry.Propagator (Propagator) import OpenTelemetry.Propagator.B3 (b3MultiTraceContextPropagator, b3TraceContextPropagator) +import OpenTelemetry.Propagator.Datadog (datadogTraceContextPropagator) import OpenTelemetry.Propagator.W3CBaggage (w3cBaggagePropagator) import OpenTelemetry.Propagator.W3CTraceContext (w3cTraceContextPropagator) import OpenTelemetry.Resource @@ -221,21 +226,23 @@ {- $envGeneral - +---------------------------+---------------------------------------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------------------------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ - | Name | Description | Default | Notes | - +===========================+===============================================================================================================+================================================================================================================================================+================================================================================================================================================================================================================================================================================+ - | OTEL_RESOURCE_ATTRIBUTES | Key-value pairs to be used as resource attributes | See [Resource semantic conventions](resource/semantic_conventions/README.md#semantic-attributes-with-sdk-provided-default-value) for details. | See [Resource SDK](./resource/sdk.md#specifying-resource-information-via-an-environment-variable) for more details. | - +---------------------------+---------------------------------------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------------------------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ - | OTEL_SERVICE_NAME | Sets the value of the [`service.name`](./resource/semantic_conventions/README.md#service) resource attribute | | If `service.name` is also provided in `OTEL_RESOURCE_ATTRIBUTES`, then `OTEL_SERVICE_NAME` takes precedence. | - +---------------------------+---------------------------------------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------------------------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ - | OTEL_LOG_LEVEL | Log level used by the SDK logger | "info" | | - +---------------------------+---------------------------------------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------------------------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ - | OTEL_PROPAGATORS | Propagators to be used as a comma-separated list | "tracecontext,baggage" | Values MUST be deduplicated in order to register a `Propagator` only once. | - +---------------------------+---------------------------------------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------------------------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ - | OTEL_TRACES_SAMPLER | Sampler to be used for traces | "parentbased_always_on" | See [Sampling](./trace/sdk.md#sampling) | - +---------------------------+---------------------------------------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------------------------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ - | OTEL_TRACES_SAMPLER_ARG | String value to be used as the sampler argument | | The specified value will only be used if OTEL_TRACES_SAMPLER is set. Each Sampler type defines its own expected input, if any. Invalid or unrecognized input MUST be logged and MUST be otherwise ignored, i.e. the SDK MUST behave as if OTEL_TRACES_SAMPLER_ARG is not set. | - +---------------------------+---------------------------------------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------------------------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ + +---------------------------+---------------------------------------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ + | Name | Description | Default | Notes | + +===========================+===============================================================================================================+================================================================================================================================================+==========================================================================================================================================================================================================================================================================================+ + | OTEL_SDK_DISABLED | Disable the SDK for all signals | false | Boolean value. If “true”, a no-op SDK implementation will be used for all telemetry signals. Any other value or absence of the variable will have no effect and the SDK will remain enabled. This setting has no effect on propagators configured through the OTEL_PROPAGATORS variable. | + +---------------------------+---------------------------------------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ + | OTEL_RESOURCE_ATTRIBUTES | Key-value pairs to be used as resource attributes | See [Resource semantic conventions](resource/semantic_conventions/README.md#semantic-attributes-with-sdk-provided-default-value) for details. | See [Resource SDK](./resource/sdk.md#specifying-resource-information-via-an-environment-variable) for more details. | + +---------------------------+---------------------------------------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ + | OTEL_SERVICE_NAME | Sets the value of the [`service.name`](./resource/semantic_conventions/README.md#service) resource attribute | | If `service.name` is also provided in `OTEL_RESOURCE_ATTRIBUTES`, then `OTEL_SERVICE_NAME` takes precedence. | + +---------------------------+---------------------------------------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ + | OTEL_LOG_LEVEL | Log level used by the SDK logger | "info" | | + +---------------------------+---------------------------------------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ + | OTEL_PROPAGATORS | Propagators to be used as a comma-separated list | "tracecontext,baggage" | Values MUST be deduplicated in order to register a `Propagator` only once. | + +---------------------------+---------------------------------------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ + | OTEL_TRACES_SAMPLER | Sampler to be used for traces | "parentbased_always_on" | See [Sampling](./trace/sdk.md#sampling) | + +---------------------------+---------------------------------------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ + | OTEL_TRACES_SAMPLER_ARG | String value to be used as the sampler argument | | The specified value will only be used if OTEL_TRACES_SAMPLER is set. Each Sampler type defines its own expected input, if any. Invalid or unrecognized input MUST be logged and MUST be otherwise ignored, i.e. the SDK MUST behave as if OTEL_TRACES_SAMPLER_ARG is not set. | + +---------------------------+---------------------------------------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ -} @@ -290,6 +297,7 @@ , ("baggage", w3cBaggagePropagator) , ("b3", b3TraceContextPropagator) , ("b3multi", b3MultiTraceContextPropagator) + , ("datadog", datadogTraceContextPropagator) , ("jaeger", error "Jaeger not yet implemented") ] @@ -321,7 +329,7 @@ createTracerProvider processors opts -getTracerProviderInitializationOptions :: IO ([Processor], TracerProviderOptions) +getTracerProviderInitializationOptions :: IO ([SpanProcessor], TracerProviderOptions) getTracerProviderInitializationOptions = getTracerProviderInitializationOptions' (mempty :: Resource 'Nothing) @@ -329,32 +337,42 @@ @since 0.0.3.1 -} -getTracerProviderInitializationOptions' :: (ResourceMerge 'Nothing any ~ 'Nothing) => Resource any -> IO ([Processor], TracerProviderOptions) +getTracerProviderInitializationOptions' + :: forall schema + . (MaterializeResource schema) + => Resource schema + -> IO ([SpanProcessor], TracerProviderOptions) getTracerProviderInitializationOptions' rs = do - sampler <- detectSampler - attrLimits <- detectAttributeLimits - spanLimits <- detectSpanLimits - propagators <- detectPropagators - processorConf <- detectBatchProcessorConfig - exporters <- detectExporters - builtInRs <- detectBuiltInResources - envVarRs <- mkResource . map Just <$> detectResourceAttributes - let allRs = mergeResources (builtInRs <> envVarRs) rs - processors <- case exporters of - [] -> do - pure [] - e : _ -> do - pure <$> batchProcessor processorConf e - let providerOpts = - emptyTracerProviderOptions - { tracerProviderOptionsIdGenerator = defaultIdGenerator - , tracerProviderOptionsSampler = sampler - , tracerProviderOptionsAttributeLimits = attrLimits - , tracerProviderOptionsSpanLimits = spanLimits - , tracerProviderOptionsPropagators = propagators - , tracerProviderOptionsResources = materializeResources allRs - } - pure (processors, providerOpts) + disabled <- lookupBooleanEnv "OTEL_SDK_DISABLED" + if disabled + then pure ([], emptyTracerProviderOptions) + else do + sampler <- detectSampler + attrLimits <- detectAttributeLimits + spanLimits <- detectSpanLimits + propagators <- detectPropagators + processorConf <- detectBatchProcessorConfig + exporters <- detectExporters + builtInRs <- detectBuiltInResources + envVarRs <- mkResource . map Just <$> detectResourceAttributes + let + -- NB: Resource merge prioritizes the left value on attribute key conflict. + allRs = mergeResources rs (envVarRs <> builtInRs) + processors <- case exporters of + [] -> do + pure [] + e : _ -> do + pure <$> batchProcessor processorConf e + let providerOpts = + emptyTracerProviderOptions + { tracerProviderOptionsIdGenerator = defaultIdGenerator + , tracerProviderOptionsSampler = sampler + , tracerProviderOptionsAttributeLimits = attrLimits + , tracerProviderOptionsSpanLimits = spanLimits + , tracerProviderOptionsPropagators = propagators + , tracerProviderOptionsResources = materializeResources allRs + } + pure (processors, providerOpts) detectPropagators :: IO (Propagator Context RequestHeaders ResponseHeaders) @@ -441,7 +459,7 @@ <*> readEnv "OTEL_LINK_ATTRIBUTE_COUNT_LIMIT" -knownExporters :: [(T.Text, IO (Exporter ImmutableSpan))] +knownExporters :: [(T.Text, IO SpanExporter)] knownExporters = [ ( "otlp" @@ -455,7 +473,7 @@ -- TODO, support multiple exporters -detectExporters :: IO [Exporter ImmutableSpan] +detectExporters :: IO [SpanExporter] detectExporters = do exportersInEnv <- fmap (T.splitOn "," . T.pack) <$> lookupEnv "OTEL_TRACES_EXPORTER" if exportersInEnv == Just ["none"] diff --git a/src/platform/unix/OpenTelemetry/Platform.hs b/src/platform/unix/OpenTelemetry/Platform.hs new file mode 100644 --- /dev/null +++ b/src/platform/unix/OpenTelemetry/Platform.hs @@ -0,0 +1,17 @@ +module OpenTelemetry.Platform where + +import Control.Exception (throwIO, try) +import qualified Data.Text as T +import System.IO.Error (isDoesNotExistError) +import System.Posix.User (getEffectiveUserName) + + +tryGetUser :: IO (Maybe T.Text) +tryGetUser = do + eResult <- try getEffectiveUserName + case eResult of + Left err -> + if isDoesNotExistError err + then pure Nothing + else throwIO err + Right ok -> pure $ Just $ T.pack ok diff --git a/src/platform/windows/OpenTelemetry/Platform.hs b/src/platform/windows/OpenTelemetry/Platform.hs new file mode 100644 --- /dev/null +++ b/src/platform/windows/OpenTelemetry/Platform.hs @@ -0,0 +1,7 @@ +module OpenTelemetry.Platform where + +import qualified Data.Text as T + + +tryGetUser :: IO (Maybe T.Text) +tryGetUser = pure Nothing diff --git a/test/OpenTelemetry/TraceSpec.hs b/test/OpenTelemetry/TraceSpec.hs --- a/test/OpenTelemetry/TraceSpec.hs +++ b/test/OpenTelemetry/TraceSpec.hs @@ -1,13 +1,22 @@ {-# LANGUAGE DataKinds #-} {-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE PatternSynonyms #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TypeApplications #-} module OpenTelemetry.TraceSpec where import Control.Monad +import Data.Foldable (traverse_) +import qualified Data.HashMap.Strict as HM import Data.IORef import Data.Int import Data.Text (Text) +import qualified Data.Text as Text +import GHC.Stack (withFrozenCallStack) +import OpenTelemetry.Attributes (AttributeLimits (..), Attributes, defaultAttributeLimits, lookupAttribute) import qualified OpenTelemetry.Context as Context +import OpenTelemetry.Resource import OpenTelemetry.Trace import OpenTelemetry.Trace.Core import OpenTelemetry.Trace.Id @@ -22,6 +31,22 @@ asIO = id +pattern HostName :: Text +pattern HostName = "host.name" + + +pattern TelemetrySdkLanguage :: Text +pattern TelemetrySdkLanguage = "telemetry.sdk.language" + + +pattern ExampleName :: Text +pattern ExampleName = "example.name" + + +pattern ExampleCount :: Text +pattern ExampleCount = "example.count" + + spec :: Spec spec = describe "Trace" $ do describe "TracerProvider" $ do @@ -37,6 +62,37 @@ specify "Safe for concurrent calls" pending specify "Shutdown" pending specify "ForceFlush" pending + specify "Resource initialization prioritizes user override, then OTEL_RESOURCE_ATTRIBUTES env var" $ do + let getInitialResourceAttrs :: Resource 'Nothing -> IO Attributes + getInitialResourceAttrs resource = do + opts <- snd <$> getTracerProviderInitializationOptions' resource + pure . getMaterializedResourcesAttributes $ tracerProviderOptionsResources opts + shouldHaveAttrPair :: Attributes -> (Text, Attribute) -> Expectation + shouldHaveAttrPair attrs (k, v) = lookupAttribute attrs k `shouldBe` Just v + attrsFromEnv <- getInitialResourceAttrs mempty + traverse_ + (attrsFromEnv `shouldHaveAttrPair`) + [ (HostName, toAttribute @Text "env_host_name") + , (TelemetrySdkLanguage, toAttribute @Text "haskell") + , (ExampleName, toAttribute @Text "env_example_name") + , -- OTEL_RESOURCE_ATTRIBUTES uses Baggage format, where attribute values are always text + (ExampleCount, toAttribute @Text "42") + ] + attrsFromUser <- + getInitialResourceAttrs $ + mkResource @'Nothing + [ HostName .= toAttribute @Text "user_host_name" + , TelemetrySdkLanguage .= toAttribute @Text "GHC2021" + , ExampleCount .= toAttribute @Int 11 + ] + traverse_ + (attrsFromUser `shouldHaveAttrPair`) + [ (HostName, toAttribute @Text "user_host_name") + , (TelemetrySdkLanguage, toAttribute @Text "GHC2021") + , -- No user override for "example.name", so the value from OTEL_RESOURCES_ATTRIBUTES shines through + (ExampleName, toAttribute @Text "env_example_name") + , (ExampleCount, toAttribute @Int 11) + ] describe "Trace / Context interaction" $ do specify "Set active span, Get active span" $ do @@ -187,8 +243,40 @@ let t = makeTracer p "woo" tracerOptions s <- createSpan t Context.empty "create_root_span" defaultSpanArguments addAttribute s "🚀" ("🚀" :: Text) - -- TODO actually get attributes out + -- TODO actually get attributes out + specify "Source code attributes are added correctly" $ asIO $ do + p <- getGlobalTracerProvider + let t = makeTracer p "woo" tracerOptions + s <- unsafeReadSpan =<< f t + spanAttributes s `shouldSatisfy` \attrs -> + (lookupAttribute attrs "code.function") == Just (toAttribute @Text "f") + && (lookupAttribute attrs "code.namespace") == Just (toAttribute @Text "OpenTelemetry.TraceSpec") + + specify "Source code attributes are added correctly in the presence of frozen call stacks" $ asIO $ do + p <- getGlobalTracerProvider + let t = makeTracer p "woo" tracerOptions + s <- unsafeReadSpan =<< g3 t + spanAttributes s `shouldSatisfy` \attrs -> + (lookupAttribute attrs "code.function") == Just (toAttribute @Text "g") + && (lookupAttribute attrs "code.namespace") == Just (toAttribute @Text "OpenTelemetry.TraceSpec") + + specify "Attribute length limit is respected" $ asIO $ do + p <- getGlobalTracerProvider + let t = makeTracer p "woo" tracerOptions + s <- createSpan t Context.empty "test_span" defaultSpanArguments + let + longAttribute :: Text = "looooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooong" + -- We set this limit in Spec.hs + -- The long attribute should be truncated down to this length + truncatedAttribute :: Text = Text.take 50 longAttribute + addAttribute s "attr1" longAttribute + addAttributes s (HM.singleton "attr2" (toAttribute longAttribute)) + s <- unsafeReadSpan s + + lookupAttribute (spanAttributes s) "attr1" `shouldBe` Just (toAttribute truncatedAttribute) + lookupAttribute (spanAttributes s) "attr2" `shouldBe` Just (toAttribute truncatedAttribute) + describe "Span events" $ do specify "AddEvent" $ asIO $ do p <- getGlobalTracerProvider @@ -217,3 +305,27 @@ specify "SpanLimits" pending specify "Built-in Processors implement ForceFlush spec" pending specify "Attribute Limits" pending + + +f :: HasCallStack => Tracer -> IO Span +f tracer = + createSpan tracer Context.empty "name" defaultSpanArguments + + +helper :: HasCallStack => Tracer -> IO Span +helper tracer = + createSpan tracer Context.empty "name" defaultSpanArguments + + +g :: HasCallStack => Tracer -> IO Span +-- block createSpan and callerAttributes from appearing in the call stack +g tracer = withFrozenCallStack $ helper tracer + + +g2 :: HasCallStack => Tracer -> IO Span +g2 tracer = g tracer + + +-- Make a 3-deep call stack +g3 :: HasCallStack => Tracer -> IO Span +g3 tracer = g2 tracer diff --git a/test/Spec.hs b/test/Spec.hs --- a/test/Spec.hs +++ b/test/Spec.hs @@ -3,11 +3,16 @@ import qualified OpenTelemetry.ResourceSpec as ResourceSpec import OpenTelemetry.Trace (initializeGlobalTracerProvider) import qualified OpenTelemetry.TraceSpec as TraceSpec +import System.Environment (setEnv) import Test.Hspec main :: IO () main = do + -- For the attribute length limit test + setEnv "OTEL_ATTRIBUTE_VALUE_LENGTH_LIMIT" "50" + -- For the resource attributes initialization test + setEnv "OTEL_RESOURCE_ATTRIBUTES" "host.name=env_host_name,example.name=env_example_name,example.count=42" initializeGlobalTracerProvider hspec $ do BaggageSpec.spec