hs-opentelemetry-sdk 0.0.3.5 → 0.0.3.6
raw patch · 7 files changed
+40/−46 lines, 7 filesdep −mwc-randomdep −random-bytestringdep ~hs-opentelemetry-apidep ~random
Dependencies removed: mwc-random, random-bytestring
Dependency ranges changed: hs-opentelemetry-api, random
Files
- ChangeLog.md +4/−0
- README.md +2/−1
- hs-opentelemetry-sdk.cabal +4/−8
- src/OpenTelemetry/Processor/Batch.hs +18/−12
- src/OpenTelemetry/Trace.hs +2/−2
- src/OpenTelemetry/Trace/Id/Generator/Default.hs +9/−22
- test/OpenTelemetry/TraceSpec.hs +1/−1
ChangeLog.md view
@@ -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
README.md view
@@ -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).
hs-opentelemetry-sdk.cabal view
@@ -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 <https://github.com/iand675/hs-opentelemetry/tree/main/sdk#readme> 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
src/OpenTelemetry/Processor/Batch.hs view
@@ -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
src/OpenTelemetry/Trace.hs view
@@ -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
src/OpenTelemetry/Trace/Id/Generator/Default.hs view
@@ -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
test/OpenTelemetry/TraceSpec.hs view
@@ -197,7 +197,7 @@ addEvent s $ NewEvent { newEventName = "EVENT"- , newEventAttributes = []+ , newEventAttributes = mempty , newEventTimestamp = Nothing } specify "Add order preserved" pending