diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,9 @@ # Changelog for hs-opentelemetry-sdk +## 0.0.3.6 + +- Raise minimum version bounds for `random` to 1.2.0. This fixes duplicate ID generation issues in highly concurrent systems. + ## 0.0.3.3 - Fix batch processor flush behavior on shutdown to not drop spans diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -94,13 +94,14 @@ -- your existing code here... pure () where + withTracer :: ((TracerOptions -> Tracer) -> IO c) -> IO c withTracer f = bracket -- Install the SDK, pulling configuration from the environment initializeGlobalTracerProvider -- Ensure that any spans that haven't been exported yet are flushed shutdownTracerProvider -- Get a tracer so you can create spans - (\tracerProvider -> f makeTracer tracerProvider "your-app-name-or-subsystem") + (\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.5 +version: 0.0.3.6 synopsis: OpenTelemetry SDK for use in applications. description: Please see the README on GitHub at category: OpenTelemetry, Telemetry, Monitoring, Observability, Metrics @@ -73,15 +73,13 @@ async , base >=4.7 && <5 , bytestring - , hs-opentelemetry-api ==0.0.3.* + , 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.* , http-types - , mwc-random , network-bsd - , random - , random-bytestring + , random >=1.2.0 , stm , text , transformers @@ -116,10 +114,8 @@ , hs-opentelemetry-sdk , hspec , http-types - , mwc-random , network-bsd - , random - , random-bytestring + , random >=1.2.0 , stm , text , transformers 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 @@ -164,13 +164,14 @@ data BoundedMap a = BoundedMap { itemBounds :: !Int + , itemMaxExportBounds :: !Int , itemCount :: !Int , itemMap :: HashMap InstrumentationLibrary (Builder.Builder a) } -boundedMap :: Int -> BoundedMap a -boundedMap bounds = BoundedMap bounds 0 mempty +boundedMap :: Int -> Int -> BoundedMap a +boundedMap bounds exportBounds = BoundedMap bounds exportBounds 0 mempty push :: ImmutableSpan -> BoundedMap ImmutableSpan -> Maybe (BoundedMap ImmutableSpan) @@ -197,7 +198,7 @@ ) -data ProcessorMessage = Flush | Shutdown +data ProcessorMessage = ScheduledFlush | MaxExportFlush | Shutdown -- note: [Unmasking Asyncs] @@ -230,10 +231,10 @@ 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 :: (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 + batch <- newIORef $ boundedMap maxQueueSize maxExportBatchSize workSignal <- newEmptyTMVarIO shutdownSignal <- newEmptyTMVarIO let publish batchToProcess = mask_ $ do @@ -259,10 +260,11 @@ delay <- registerDelay (millisToMicros scheduledDelayMillis) atomically $ do msum - [ Flush <$ do + -- 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 - , Flush <$ takeTMVar workSignal + , MaxExportFlush <$ takeTMVar workSignal , Shutdown <$ takeTMVar shutdownSignal ] @@ -285,11 +287,15 @@ { processorOnStart = \_ _ -> pure () , processorOnEnd = \s -> do span_ <- readIORef s - appendFailed <- atomicModifyIORef' batch $ \builder -> + appendFailedOrExportNeeded <- atomicModifyIORef' batch $ \builder -> case push span_ builder of Nothing -> (builder, True) - Just b' -> (b', False) - when appendFailed $ void $ atomically $ tryPutTMVar workSignal () + 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 = @@ -323,12 +329,12 @@ atomically $ msum [ Just <$> waitCatchSTM worker - , const Nothing <$> do + , Nothing <$ do shouldStop <- readTVar delay check shouldStop ] - -- make sure the worker comes down. + -- make sure the worker comes down if we timed out. cancel worker -- TODO, not convinced we should shut down processor here diff --git a/src/OpenTelemetry/Trace.hs b/src/OpenTelemetry/Trace.hs --- a/src/OpenTelemetry/Trace.hs +++ b/src/OpenTelemetry/Trace.hs @@ -488,12 +488,12 @@ Baggage.values ok -readEnvDefault :: forall a. Read a => String -> a -> IO a +readEnvDefault :: forall a. (Read a) => String -> a -> IO a readEnvDefault k defaultValue = fromMaybe defaultValue . (>>= readMaybe) <$> lookupEnv k -readEnv :: forall a. Read a => String -> IO (Maybe a) +readEnv :: forall a. (Read a) => String -> IO (Maybe a) readEnv k = (>>= readMaybe) <$> lookupEnv k diff --git a/src/OpenTelemetry/Trace/Id/Generator/Default.hs b/src/OpenTelemetry/Trace/Id/Generator/Default.hs --- a/src/OpenTelemetry/Trace/Id/Generator/Default.hs +++ b/src/OpenTelemetry/Trace/Id/Generator/Default.hs @@ -20,14 +20,7 @@ import OpenTelemetry.Trace.Id.Generator (IdGenerator (..)) import System.IO.Unsafe (unsafePerformIO) -import System.Random.MWC - - -#if MIN_VERSION_random(1,2,0) import System.Random.Stateful -#else -import Data.ByteString.Random -#endif {- | The default generator for trace and span ids. @@ -35,20 +28,14 @@ @since 0.1.0.0 -} defaultIdGenerator :: IdGenerator -#if MIN_VERSION_random(1,2,0) defaultIdGenerator = unsafePerformIO $ do - g <- createSystemRandom - pure $ IdGenerator - { generateSpanIdBytes = uniformByteStringM 8 g - , generateTraceIdBytes = uniformByteStringM 16 g - } -{-# NOINLINE defaultIdGenerator #-} -#else -defaultIdGenerator = unsafePerformIO $ do - g <- createSystemRandom - pure $ IdGenerator - { generateSpanIdBytes = randomGen g 8 - , generateTraceIdBytes = randomGen g 16 - } + genBase <- initStdGen + let (spanIdGen, traceIdGen) = split genBase + sg <- newAtomicGenM spanIdGen + tg <- newAtomicGenM traceIdGen + pure $ + IdGenerator + { generateSpanIdBytes = uniformByteStringM 8 sg + , generateTraceIdBytes = uniformByteStringM 16 tg + } {-# NOINLINE defaultIdGenerator #-} -#endif diff --git a/test/OpenTelemetry/TraceSpec.hs b/test/OpenTelemetry/TraceSpec.hs --- a/test/OpenTelemetry/TraceSpec.hs +++ b/test/OpenTelemetry/TraceSpec.hs @@ -197,7 +197,7 @@ addEvent s $ NewEvent { newEventName = "EVENT" - , newEventAttributes = [] + , newEventAttributes = mempty , newEventTimestamp = Nothing } specify "Add order preserved" pending