diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,13 @@ # Changelog for hs-opentelemetry-sdk +## 0.0.3.3 + +- Fix batch processor flush behavior on shutdown to not drop spans + +## 0.0.3.2 + +- Fix haddock issue + ## 0.0.3.1 - `getTracerProviderInitializationOptions'` introduced to enable custom resource detection diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -65,7 +65,7 @@ The OpenTelemetry Tracing API uses a data type called a `Tracer` to create traces. These `Tracer`s are designed to be associated with one instrumentation library. That way, telemetry they produce can be understood to come from the library or portion of your code base that it instruments. -A `Tracer` is constructed by calling the `getTracer` function, which requires a `TracerProvider`, which we'll discuss next. +A `Tracer` is constructed by calling the `makeTracer` function, which requires a `TracerProvider` and `TracerOptions`, which we'll discuss next. ### TracerProvider @@ -99,11 +99,8 @@ initializeGlobalTracerProvider -- Ensure that any spans that haven't been exported yet are flushed shutdownTracerProvider - (\tracerProvider -> do - -- Get a tracer so you can create spans - tracer <- getTracer tracerProvider "your-app-name-or-subsystem" - f tracer - ) + -- Get a tracer so you can create spans + (\tracerProvider -> f makeTracer tracerProvider "your-app-name-or-subsystem") ``` The primary configuration mechanism for `initializeGlobalTracerProvider` is via the environment variables listed in the official [OpenTelemetry specification](https://github.com/open-telemetry/opentelemetry-specification/blob/6ad485b743553099476d676f1f0369bae0304547/specification/sdk-environment-variables.md). diff --git a/hs-opentelemetry-sdk.cabal b/hs-opentelemetry-sdk.cabal --- a/hs-opentelemetry-sdk.cabal +++ b/hs-opentelemetry-sdk.cabal @@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack name: hs-opentelemetry-sdk -version: 0.0.3.1 +version: 0.0.3.3 synopsis: OpenTelemetry SDK for use in applications. description: Please see the README on GitHub at category: OpenTelemetry, Telemetry, Monitoring, Observability, Metrics @@ -83,6 +83,7 @@ , random-bytestring , stm , text + , transformers , unagi-chan , unix , unordered-containers @@ -119,6 +120,7 @@ , random-bytestring , stm , text + , transformers , unagi-chan , unix , unordered-containers 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 @@ -28,6 +28,7 @@ import Control.Concurrent.Async import Control.Concurrent.MVar (newEmptyMVar, takeMVar, tryPutMVar) import Control.Monad +import Control.Monad.Trans.Except import System.Timeout import Control.Exception import Data.HashMap.Strict (HashMap) @@ -100,15 +101,15 @@ -} --- newGreenBlueBuffer +-- 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 +-- let closestFittingPowerOfTwo = 2 * if (1 `shiftL` logBase2) == maxQueueSize +-- then maxQueueSize -- else 1 `shiftL` (logBase2 + 1) -- readSection <- newTVarIO 0 @@ -151,7 +152,7 @@ -- modifyTVar gbWriteSection (+ 1) -- setTVar gbPendingWrites 0 -- -- TODO slice and freeze appropriate section - -- M.slice (gbSectionSize * (r .&. gbSectionMask) + -- M.slice (gbSectionSize * (r .&. gbSectionMask) -- TODO, counters for dropped spans, exported spans @@ -182,19 +183,31 @@ , Builder.build <$> itemMap m ) +-- | Exitable forever loop +loop :: Monad m => ExceptT e m a -> m e +loop = liftM (either id id) . runExceptT . forever + +data ProcessorMessage = Flush | Shutdown + -- | --- The batch processor accepts spans and places them into batches. Batching helps better compress the data and reduce the number of outgoing connections +-- 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. -- batchProcessor :: MonadIO m => BatchTimeoutConfig -> Exporter ImmutableSpan -> m Processor batchProcessor BatchTimeoutConfig{..} exporter = liftIO $ do batch <- newIORef $ boundedMap maxQueueSize workSignal <- newEmptyMVar - worker <- async $ forever $ do - void $ timeout (millisToMicros scheduledDelayMillis) $ takeMVar workSignal - batchToProcess <- atomicModifyIORef' batch buildExport - Exporter.exporterExport exporter batchToProcess + worker <- async $ loop $ do + req <- liftIO $ timeout (millisToMicros scheduledDelayMillis) + $ takeMVar workSignal + batchToProcess <- liftIO $ atomicModifyIORef' batch buildExport + res <- liftIO $ Exporter.exporterExport exporter batchToProcess + -- if we were asked to shutdown, quit cleanly after this batch + -- FIXME: this could lose batches if there's more than one in queue? + case req of + Just Shutdown -> throwE res + _ -> pure () pure $ Processor { processorOnStart = \_ _ -> pure () @@ -204,13 +217,18 @@ case push span_ builder of Nothing -> (builder, True) Just b' -> (b', False) - when appendFailed $ void $ tryPutMVar workSignal () + when appendFailed $ void $ tryPutMVar workSignal Flush - , processorForceFlush = void $ tryPutMVar workSignal () + , processorForceFlush = void $ tryPutMVar workSignal Flush -- TODO where to call restore, if anywhere? , processorShutdown = async $ mask $ \_restore -> do + -- flush remaining messages + void $ tryPutMVar workSignal Shutdown + shutdownResult <- timeout (millisToMicros exportTimeoutMillis) $ - cancel worker + wait worker + -- make sure the worker comes down + uninterruptibleCancel worker -- TODO, not convinced we should shut down processor here case shutdownResult of @@ -223,7 +241,7 @@ buffer <- newGreenBlueBuffer _ _ batchProcessorAction <- async $ forever $ do -- It would be nice to do an immediate send when possible - chunk <- if (sendDelay == 0) + chunk <- if (sendDelay == 0) else consumeChunk then threadDelay sendDelay >> consumeChunk timeout _ $ export exporter chunk @@ -232,7 +250,7 @@ , onEnd = \s -> void $ tryInsert buffer s , shutdown = do gracefullyShutdownBatchProcessor - + , forceFlush = pure () } 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 RankNTypes #-} {-# LANGUAGE DataKinds #-} {-# LANGUAGE GADTs #-} +{-# LANGUAGE TypeOperators #-} ----------------------------------------------------------------------------- -- | -- Module : OpenTelemetry.Trace @@ -124,6 +125,7 @@ , TracerProviderOptions(..) , emptyTracerProviderOptions , detectBuiltInResources + , detectSampler , createSpan , createSpanWithoutCallStack , endSpan @@ -299,7 +301,7 @@ -- | Detect ptions for initializing a tracer provider from the app environment, taking additional supported resources as well. -- --- @since +-- @since 0.0.3.1 getTracerProviderInitializationOptions' :: (ResourceMerge 'Nothing any ~ 'Nothing) => Resource any -> IO ([Processor], TracerProviderOptions) getTracerProviderInitializationOptions' rs = do sampler <- detectSampler @@ -360,6 +362,11 @@ ] -- TODO MUST log invalid arg + +-- | Detect a sampler from the app environment. If no sampler is specified, +-- the parentbased sampler is used. +-- +-- @since 0.0.3.3 detectSampler :: IO Sampler detectSampler = do envSampler <- lookupEnv "OTEL_TRACES_SAMPLER" diff --git a/test/OpenTelemetry/TraceSpec.hs b/test/OpenTelemetry/TraceSpec.hs --- a/test/OpenTelemetry/TraceSpec.hs +++ b/test/OpenTelemetry/TraceSpec.hs @@ -25,12 +25,13 @@ describe "TracerProvider" $ do specify "Create TracerProvider" $ do void (createTracerProvider [] (emptyTracerProviderOptions :: TracerProviderOptions) :: IO TracerProvider) - specify "Get a Tracer" $ asIO $ do - p <- getGlobalTracerProvider - void $ getTracer p "woo" tracerOptions - specify "Get a Tracer with schema_url" $ asIO $ do - p <- getGlobalTracerProvider - void $ getTracer p "woo" (tracerOptions { tracerSchema = Just "https://woo.com" }) + -- TODO make these tests do something meaningful with makeTracer + -- specify "Get a Tracer" $ asIO $ do + -- p <- getGlobalTracerProvider + -- void $ getTracer p "woo" tracerOptions + -- specify "Get a Tracer with schema_url" $ asIO $ do + -- p <- getGlobalTracerProvider + -- void $ getTracer p "woo" (tracerOptions { tracerSchema = Just "https://woo.com" }) specify "Safe for concurrent calls" pending specify "Shutdown" pending specify "ForceFlush" pending @@ -38,7 +39,7 @@ describe "Trace / Context interaction" $ do specify "Set active span, Get active span" $ do p <- getGlobalTracerProvider - t <- getTracer p "woo" tracerOptions + let t = makeTracer p "woo" tracerOptions s <- createSpan t Context.empty "create_root_span" defaultSpanArguments spanContext1 <- spanContext <$> unsafeReadSpan s let ctxt = Context.insertSpan s mempty @@ -49,7 +50,7 @@ describe "Tracer" $ do specify "Create a new span" $ asIO $ do p <- getGlobalTracerProvider - t <- getTracer p "woo" tracerOptions + let t = makeTracer p "woo" tracerOptions void $ createSpan t Context.empty "create_root_span" defaultSpanArguments specify "Get active new span" pending specify "Mark Span active" pending @@ -76,7 +77,7 @@ describe "Span" $ do specify "Create root span" $ asIO $ do p <- getGlobalTracerProvider - t <- getTracer p "woo" tracerOptions + let t = makeTracer p "woo" tracerOptions void $ createSpan t Context.empty "create_root_span" defaultSpanArguments specify "Create with default parent (active span)" pending specify "Create with parent from Context" pending @@ -84,24 +85,24 @@ specify "Processor.OnStart receives parent Context" pending specify "UpdateName" $ asIO $ do p <- getGlobalTracerProvider - t <- getTracer p "woo" tracerOptions + let t = makeTracer p "woo" tracerOptions s <- createSpan t Context.empty "create_root_span" defaultSpanArguments updateName s "renamed_span" specify "User-defined start timestamp" pending specify "End" $ asIO $ do p <- getGlobalTracerProvider - t <- getTracer p "woo" tracerOptions + let t = makeTracer p "woo" tracerOptions s <- createSpan t Context.empty "create_root_span" defaultSpanArguments endSpan s Nothing specify "End with timestamp" $ asIO $ do p <- getGlobalTracerProvider - t <- getTracer p "woo" tracerOptions + let t = makeTracer p "woo" tracerOptions ts <- getTimestamp s <- createSpan t Context.empty "create_root_span" defaultSpanArguments endSpan s (Just ts) specify "IsRecording" $ asIO $ do p <- getGlobalTracerProvider - t <- getTracer p "woo" tracerOptions + let t = makeTracer p "woo" tracerOptions ts <- getTime Realtime s <- createSpan t Context.empty "create_root_span" defaultSpanArguments recording <- isRecording s @@ -109,7 +110,7 @@ specify "IsRecording becomes false after End" $ do p <- getGlobalTracerProvider - t <- getTracer p "woo" tracerOptions + let t = makeTracer p "woo" tracerOptions ts <- getTime Realtime s <- createSpan t Context.empty "create_root_span" defaultSpanArguments endSpan s Nothing @@ -118,7 +119,7 @@ specify "Set status with StatusCode (Unset, Ok, Error)" $ do p <- getGlobalTracerProvider - t <- getTracer p "woo" tracerOptions + let t = makeTracer p "woo" tracerOptions ts <- getTime Realtime s <- createSpan t Context.empty "create_root_span" defaultSpanArguments @@ -144,44 +145,44 @@ describe "Span attributes" $ do specify "SetAttribute" $ asIO $ do p <- getGlobalTracerProvider - t <- getTracer p "woo" tracerOptions + let t = makeTracer p "woo" tracerOptions s <- createSpan t Context.empty "create_root_span" defaultSpanArguments addAttribute s "attr" (1.0 :: Double) specify "Set order preserved" pending specify "String type" $ asIO $ do p <- getGlobalTracerProvider - t <- getTracer p "woo" tracerOptions + let t = makeTracer p "woo" tracerOptions s <- createSpan t Context.empty "create_root_span" defaultSpanArguments addAttribute s "string_type" ("" :: Text) specify "Boolean type" $ asIO $ do p <- getGlobalTracerProvider - t <- getTracer p "woo" tracerOptions + let t = makeTracer p "woo" tracerOptions s <- createSpan t Context.empty "create_root_span" defaultSpanArguments addAttribute s "bool_type" True specify "Double floating-point type" $ asIO $ do p <- getGlobalTracerProvider - t <- getTracer p "woo" tracerOptions + let t = makeTracer p "woo" tracerOptions s <- createSpan t Context.empty "create_root_span" defaultSpanArguments addAttribute s "attr" (1.0 :: Double) specify "Signed int64 type" $ asIO $ do p <- getGlobalTracerProvider - t <- getTracer p "woo" tracerOptions + let t = makeTracer p "woo" tracerOptions s <- createSpan t Context.empty "create_root_span" defaultSpanArguments addAttribute s "attr" (1 :: Int64) specify "Array of primitives (homegeneous)" $ asIO $ do p <- getGlobalTracerProvider - t <- getTracer p "woo" tracerOptions + let t = makeTracer p "woo" tracerOptions s <- createSpan t Context.empty "create_root_span" defaultSpanArguments addAttribute s "attr" [(1 :: Int64)..10] specify "Unicode support for keys and string values" $ asIO $ do p <- getGlobalTracerProvider - t <- getTracer p "woo" tracerOptions + let t = makeTracer p "woo" tracerOptions s <- createSpan t Context.empty "create_root_span" defaultSpanArguments addAttribute s "🚀" ("🚀" :: Text) -- TODO actually get attributes out @@ -190,7 +191,7 @@ describe "Span events" $ do specify "AddEvent" $ asIO $ do p <- getGlobalTracerProvider - t <- getTracer p "woo" tracerOptions + let t = makeTracer p "woo" tracerOptions s <- createSpan t Context.empty "create_root_span" defaultSpanArguments addEvent s $ NewEvent { newEventName = "EVENT"