diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,14 @@ +# Changelog + +All notable changes to `atelier-monitoring` will be documented in this file. + +The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), +and this project adheres to the [PVP](https://pvp.haskell.org/). + +## [Unreleased] + +## [0.1.0.0] - 2026-08-14 + +### Added + +- All modules relating to monitoring and observability from `atelier-core`. diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2025 Tweag + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,12 @@ +# atelier-monitoring + +Moitoring, metrics, and tracing effects and utilities for +[Effectful](https://github.com/haskell-effectful/effectful)-based applications +— part of the \**atelie*r\* toolkit. + +`atelier-monitoring` provides a set of composable Effectful effects and +supporting types for building structured, observable applications. + +## License + +MIT — see [LICENSE](LICENSE). diff --git a/atelier-monitoring.cabal b/atelier-monitoring.cabal new file mode 100644 --- /dev/null +++ b/atelier-monitoring.cabal @@ -0,0 +1,171 @@ +cabal-version: 2.0 + +-- This file has been generated from package.yaml by hpack version 0.38.3. +-- +-- see: https://github.com/sol/hpack + +name: atelier-monitoring +version: 0.1.0.0 +synopsis: Effectful-based monitoring suite +description: Moitoring, metrics, and tracing effects and utilities for Effectful-based + applications — part of the atelier toolkit. +category: OpenTelemetry, + Observability, + Monitoring, + Tracing +homepage: https://github.com/tweag/tricorder#readme +bug-reports: https://github.com/tweag/tricorder/issues +author: Victor Nascimento Bakke +maintainer: victor.bakke@tweag.io +license: MIT +license-file: LICENSE +build-type: Simple +tested-with: + GHC == 9.10.3 + , GHC == 9.6.7 + , GHC == 9.8.4 + , GHC == 9.12.4 +extra-doc-files: + CHANGELOG.md + README.md + +source-repository head + type: git + location: https://github.com/tweag/tricorder + +library + exposed-modules: + Atelier.Component + Atelier.Effects.Cache.Singleflight.Traced + Atelier.Effects.Conc.Traced + Atelier.Effects.Monitoring.Metrics + Atelier.Effects.Monitoring.Metrics.Registry + Atelier.Effects.Monitoring.Metrics.Server + Atelier.Effects.Monitoring.Tracing + Atelier.Effects.Monitoring.Tracing.Provider + Atelier.Effects.Publishing.Traced + other-modules: + Paths_atelier_monitoring + autogen-modules: + Paths_atelier_monitoring + hs-source-dirs: + src + default-extensions: + BlockArguments + DataKinds + DeriveAnyClass + DerivingStrategies + DerivingVia + DuplicateRecordFields + FlexibleContexts + GADTs + LambdaCase + MultiWayIf + OverloadedLabels + OverloadedRecordDot + OverloadedStrings + StrictData + TemplateHaskell + TypeFamilies + ghc-options: -Weverything -Wno-unsafe -Wno-missing-safe-haskell-mode -Wno-monomorphism-restriction -Wno-missing-kind-signatures -Wno-missing-local-signatures -Wno-missing-import-lists -Wno-implicit-prelude -Wno-unticked-promoted-constructors -Wno-unused-packages -Wno-all-missed-specialisations -Wno-missed-specialisations -fplugin=Effectful.Plugin -threaded + build-depends: + aeson >=2.2 && <2.4 + , atelier-core >=0.3 && <0.5 + , atelier-prelude >=0.1 && <0.3 + , base >=4.18 && <4.23 + , base64-bytestring ==1.2.* + , bytestring >=0.11 && <0.13 + , casing ==0.1.* + , containers >=0.6 && <0.9 + , daemons ==0.4.* + , data-default ==0.8.* + , directory ==1.3.* + , effectful ==2.6.* + , effectful-core ==2.6.* + , effectful-plugin >=2.0 && <2.2 + , effectful-th ==1.0.* + , filepath >=1.4 && <1.6 + , fsnotify ==0.4.* + , hs-opentelemetry-api ==0.3.* + , hs-opentelemetry-sdk ==0.1.* + , http-api-data >=0.5.1 && <0.8 + , http-types ==0.12.* + , ki ==1.0.* + , list-t ==1.0.* + , optparse-applicative ==0.19.* + , process ==1.6.* + , prometheus-client ==1.1.* + , prometheus-metrics-ghc ==1.0.* + , stm ==2.5.* + , stm-containers ==1.2.* + , text ==2.1.* + , time >=1.12 && <1.17 + , time-units ==1.0.* + , typed-process ==0.2.* + , unagi-chan ==0.4.* + , unix ==2.8.* + , unordered-containers ==0.2.* + , uuid ==1.3.* + , wai ==3.2.* + , warp ==3.4.* + mixins: + base hiding (Prelude) + default-language: GHC2021 + if impl(GHC >= 9.8) + ghc-options: -Wno-missing-poly-kind-signatures -Wno-missing-role-annotations + +test-suite atelier-monitoring-test + type: exitcode-stdio-1.0 + main-is: Driver.hs + other-modules: + Unit.Atelier.Effects.Conc.TracedSpec + Paths_atelier_monitoring + autogen-modules: + Paths_atelier_monitoring + hs-source-dirs: + test + default-extensions: + BlockArguments + DataKinds + DeriveAnyClass + DerivingStrategies + DerivingVia + DuplicateRecordFields + FlexibleContexts + GADTs + LambdaCase + MultiWayIf + OverloadedLabels + OverloadedRecordDot + OverloadedStrings + StrictData + TemplateHaskell + TypeFamilies + ghc-options: -Weverything -Wno-unsafe -Wno-missing-safe-haskell-mode -Wno-monomorphism-restriction -Wno-missing-kind-signatures -Wno-missing-local-signatures -Wno-missing-import-lists -Wno-implicit-prelude -Wno-unticked-promoted-constructors -Wno-unused-packages -Wno-all-missed-specialisations -Wno-missed-specialisations -fplugin=Effectful.Plugin -threaded -Wno-prepositive-qualified-module + build-tool-depends: + tasty-discover:tasty-discover + build-depends: + aeson >=2.2 && <2.4 + , atelier-monitoring ==0.1.* + , atelier-prelude >=0.1 && <0.3 + , base >=4.18 && <4.23 + , bytestring >=0.11 && <0.13 + , containers >=0.6 && <0.9 + , data-default ==0.8.* + , effectful ==2.6.* + , effectful-core ==2.6.* + , effectful-plugin >=2.0 && <2.2 + , hedgehog ==1.7.* + , hs-opentelemetry-api ==0.3.* + , hspec ==2.11.* + , hspec-hedgehog ==0.3.* + , stm ==2.5.* + , stm-containers ==1.2.* + , tasty ==1.5.* + , tasty-hspec ==1.2.* + , time >=1.12 && <1.17 + mixins: + base hiding (Prelude) + default-language: GHC2021 + if impl(GHC >= 9.8) + ghc-options: -Wno-missing-poly-kind-signatures -Wno-missing-role-annotations diff --git a/src/Atelier/Component.hs b/src/Atelier/Component.hs new file mode 100644 --- /dev/null +++ b/src/Atelier/Component.hs @@ -0,0 +1,132 @@ +-- | A small component model for assembling long-running applications. +-- +-- A 'Component' bundles a named unit of work with a lifecycle: one-off 'setup', +-- a set of 'listeners' and 'triggers' that run as forked threads, and a +-- post-'start' action. 'runSystem' drives a collection of components through +-- these phases in lockstep, so that (for example) every component has finished +-- 'setup' before any 'start' action runs and publishes events the others react +-- to. +module Atelier.Component + ( -- * Components + Component (..) + , defaultComponent + , Listener + , Trigger + + -- * Running + , runComponent + , runSystem + ) where + +import Text.Casing (fromHumps, toQuietSnake) + +import Atelier.Effects.Conc (Conc) +import Atelier.Effects.Log (Log) +import Atelier.Effects.Monitoring.Tracing (Tracing, withSpan) + +import Atelier.Effects.Conc qualified as Conc +import Atelier.Effects.Log qualified as Log + + +-- | A listener reacts to events and runs forever; it never returns normally, +-- hence the 'Void' result. +type Listener es = Eff es Void + + +-- | A trigger initiates periodic or scheduled work and, like a 'Listener', runs +-- forever and never returns normally. +type Trigger es = Eff es Void + + +-- | A named unit of application work together with its lifecycle hooks. +-- +-- Build one by overriding the fields of 'defaultComponent' you care about. +data Component es = Component + { name :: ~Text + -- ^ Component name for tracing + , setup :: Eff es () + -- ^ Setup component (runs before listeners/triggers start) + , listeners :: Eff es [Listener es] + -- ^ Event listeners (react to events) + , triggers :: Eff es [Trigger es] + -- ^ Triggers (initiate periodic/scheduled work) + , start :: Eff es () + -- ^ Post-start actions (runs after all components have started) + } + + +-- | A 'Component' with no-op lifecycle hooks and no listeners or triggers. +-- +-- Override the fields you need. 'name' is deliberately left as an 'error' so a +-- component created without a name fails fast rather than tracing anonymously. +defaultComponent :: (HasCallStack) => Component es +defaultComponent = + Component + { name = error "Missing component name" + , setup = pure () + , listeners = pure [] + , triggers = pure [] + , start = pure () + } + + +-- | Run a component by forking its listeners and triggers +runComponent :: (Conc :> es, Tracing :> es) => Component es -> Eff es () +runComponent c = withSpan c.name $ do + ls <- c.listeners + ts <- c.triggers + traverse_ Conc.fork_ ls + traverse_ Conc.fork_ ts + + +-- | Run multiple components with structured lifecycle coordination +-- +-- Executes components in three sequential phases: +-- 1. Run @setup for all components (initialization before activation) +-- 2. Fork all listeners and triggers (components become active) +-- 3. Fork @start for all components (background startup actions) +-- +-- This separation allows components to prepare resources during setup, then perform +-- work during start that depends on listeners already running (e.g., publishing events). +-- The start phase is forked to allow parallel execution across components. +runSystem + :: forall es + . (Conc :> es, Log :> es, Tracing :> es) + => [Component es] + -> Eff es () +runSystem components = Conc.scoped do + -- Phase 1: Setup all components + traverse_ setupComponent components + + -- Phase 2: Start all components (fork listeners/triggers) + traverse_ startComponent components + + -- Phase 3: Fork post-start actions + traverse_ postStartComponent components + + Conc.awaitAll + where + setupComponent :: Component es -> Eff es () + setupComponent c = do + let name = formatName c.name + Log.debug $ "Setting up component: " <> name + withSpan (name <> ":setup") + $ c.setup + + startComponent :: Component es -> Eff es () + startComponent c = do + let name = formatName c.name + Log.debug $ "Starting listeners/triggers for component: " <> name + runComponent c + + postStartComponent :: Component es -> Eff es () + postStartComponent c = do + let name = formatName c.name + Log.debug $ "Forking start phase for component: " <> name + void . Conc.fork + $ withSpan (name <> ":start") + $ c.start + + +formatName :: Text -> Text +formatName = toText . toQuietSnake . fromHumps . toString diff --git a/src/Atelier/Effects/Cache/Singleflight/Traced.hs b/src/Atelier/Effects/Cache/Singleflight/Traced.hs new file mode 100644 --- /dev/null +++ b/src/Atelier/Effects/Cache/Singleflight/Traced.hs @@ -0,0 +1,124 @@ +-- | A singleflight cache effect: deduplicate concurrent computations of the +-- same key. +-- +-- When several threads request the same key at once, the first runs the +-- computation while the rest wait and share its result — so an expensive lookup +-- happens once per key per in-flight window. Results (and exceptions) can also +-- be seeded with 'updateCache' or invalidated with 'removeFromCache'. Each +-- operation is traced (see "Atelier.Effects.Monitoring.Tracing"). +module Atelier.Effects.Cache.Singleflight.Traced + ( Singleflight + , withCache + , updateCache + , removeFromCache + , runSingleflight + ) where + +import Atelier.Effects.Cache.Singleflight + ( Singleflight (..) + , removeFromCache + , updateCache + , withCache + ) +import Effectful.Concurrent (Concurrent) +import Effectful.Concurrent.STM (TMVar) +import Effectful.Dispatch.Dynamic (interpretWith, localSeqUnlift) +import Effectful.Exception (throwIO, trySync) +import StmContainers.Map (Map) +import Prelude hiding (Map) + +import Effectful.Concurrent.STM qualified as STM +import StmContainers.Map qualified as Map + +import Atelier.Effects.Monitoring.Tracing (Tracing, addAttribute, withSpan) + + +type InFlightMap key value = Map key (TMVar (Either SomeException value)) + + +-- | Run the Singleflight effect with an in-memory cache +runSingleflight + :: forall key value es a + . (Concurrent :> es, Hashable key, Tracing :> es) + => Eff (Singleflight key value : es) a + -> Eff es a +runSingleflight action = do + -- Initialize the cache + cache :: InFlightMap key value <- STM.atomically Map.new + + -- Run with the cache in Reader context, interpreting Singleflight operations + interpretWith action $ \env -> \case + WithCache key computation -> localSeqUnlift env $ \unlift -> withSpan "singleflight.with_cache" do + -- Singleflight pattern: check if computation is already in-flight. + -- Also attempt a non-blocking read of any existing result in the same transaction. + (mvar, isFirst, mResult) <- STM.atomically $ do + existing <- Map.lookup key cache + case existing of + Just tmvar -> do + mResult <- STM.tryReadTMVar tmvar + pure (tmvar, False, mResult) + Nothing -> do + -- Create new empty TMVar and insert into cache + tmvar <- STM.newEmptyTMVar + Map.insert tmvar key cache + pure (tmvar, True, Nothing) + + case (isFirst, mResult) of + (_, Just (Right value)) -> do + addAttribute @Text "singleflight.outcome" "hit" + pure value + (_, Just (Left exception)) -> do + addAttribute @Text "singleflight.outcome" "hit" + throwIO exception + (True, Nothing) -> do + addAttribute @Text "singleflight.outcome" "compute" + result <- unlift $ trySync computation + + -- Try to fill the TMVar with result (success or failure) for waiters + -- Use tryPutTMVar in case updateCache already filled it + filled <- STM.atomically $ STM.tryPutTMVar mvar result + + if filled then do + -- We successfully filled the TMVar with our result + -- If it was an exception, remove from cache so future requests can retry + case result of + Left _exception -> do + STM.atomically $ Map.delete key cache + Right _ -> pure () + + -- Return the result or re-throw the exception + case result of + Left exception -> throwIO exception + Right value -> pure value + else do + -- updateCache filled it before us - read and use that value + finalResult <- STM.atomically $ STM.readTMVar mvar + case finalResult of + Left exception -> throwIO exception + Right value -> pure value + (False, Nothing) -> do + addAttribute @Text "singleflight.outcome" "wait" + result <- + withSpan "singleflight.wait" + $ STM.atomically + $ STM.readTMVar mvar + case result of + Left exception -> throwIO exception + Right value -> pure value + UpdateCache entries -> withSpan "singleflight.update_cache" do + STM.atomically $ do + forM_ entries $ \(key, value) -> do + existing <- Map.lookup key cache + case existing of + Just existingTMVar -> do + -- In-flight computation exists: update its result + -- Try to take whatever value is there (or Nothing if empty) + _ <- STM.tryTakeTMVar existingTMVar + -- Put the correct value (wrapped in Right for success) + STM.putTMVar existingTMVar (Right value) + Nothing -> do + -- No in-flight computation: create fresh TMVar with value + tmvar <- STM.newTMVar (Right value) + Map.insert tmvar key cache + RemoveFromCache keys -> withSpan "singleflight.remove_from_cache" do + STM.atomically $ forM_ keys $ \key -> Map.delete key cache diff --git a/src/Atelier/Effects/Conc/Traced.hs b/src/Atelier/Effects/Conc/Traced.hs new file mode 100644 --- /dev/null +++ b/src/Atelier/Effects/Conc/Traced.hs @@ -0,0 +1,90 @@ +-- | Tracing integration for the 'Conc' effect. +-- +-- Provides interpreters and interposers that automatically propagate +-- OpenTelemetry trace context across thread boundaries using span links. +module Atelier.Effects.Conc.Traced + ( module ExConc + , ExConc.await + + -- * Interpreters + , runConc + , runConcByConfig + , runConcTraced + + -- * Interposers + , withConcTracingLinks + ) +where + +import Atelier.Effects.Conc (Conc (..), Scope (..), concStrat, fork, forkTry, fork_, runConcBase) +import Effectful (IOE, raise, withEffToIO) +import Effectful.Concurrent (Concurrent) +import Effectful.Dispatch.Dynamic (interpose, localLend, localUnlift, passthrough) +import Effectful.Reader.Static (Reader, asks) + +import Atelier.Effects.Conc qualified as Conc +import Atelier.Effects.Conc qualified as ExConc hiding (runConc) +import Ki qualified + +import Atelier.Effects.Monitoring.Tracing (Tracing, TracingConfig (..)) + +import Atelier.Effects.Monitoring.Tracing qualified as Tracing + + +-- | Run 'Conc' effect with automatic trace context propagation in a new scope. +runConc :: (Concurrent :> es, IOE :> es, Tracing :> es) => Eff (Conc : es) a -> Eff es a +runConc eff = withEffToIO concStrat $ \unlift -> + Ki.scoped $ \scope -> + unlift $ runConcTraced (Scope scope) eff + + +-- | Run 'Conc' effect, selecting the interpreter based on tracing config. +-- +-- If tracing is enabled, uses 'runConc' for automatic span link propagation. +-- If tracing is disabled, falls back to 'Conc.runConc' to skip the overhead. +runConcByConfig + :: ( Concurrent :> es + , IOE :> es + , Reader TracingConfig :> es + , Tracing :> es + ) + => Eff (Conc : es) a -> Eff es a +runConcByConfig eff = do + tracingEnabled <- asks @TracingConfig (.enabled) + if tracingEnabled then runConc eff else Conc.runConc eff + + +-- | Run 'Conc' effect with automatic trace context propagation. +-- +-- All fork variants ('fork', 'fork_', 'forkTry') use span links, i.e. forked +-- threads start a fresh trace whose root spans carry a link back to the +-- originating span. +runConcTraced + :: ( Concurrent :> es + , IOE :> es + , Tracing :> es + ) + => Scope -> Eff (Conc : es) a -> Eff es a +runConcTraced scope = runConcBase scope . withConcTracingLinks + + +-- | Intercept fork operations and wrap the forked action with span link propagation. +-- +-- Re-dispatches to the underlying 'Conc' handler via 'fork', 'fork_', 'forkTry'. +withConcTracingLinks :: (Conc :> es, Tracing :> es) => Eff es a -> Eff es a +withConcTracingLinks = interpose \env -> \case + Fork action -> tracedFork env action fork + Fork_ action -> tracedFork env action fork_ + ForkTry action -> tracedFork env action forkTry + other -> passthrough env other + where + tracedFork env action kiOp = do + parentCtx <- Tracing.getSpanContext + localUnlift env concStrat \unliftEff -> + localLend @'[Tracing] env concStrat \lend -> + kiOp + $ unliftEff + . lend + . Tracing.withLinkPropagation parentCtx + . raise @Tracing + $ action diff --git a/src/Atelier/Effects/Monitoring/Metrics.hs b/src/Atelier/Effects/Monitoring/Metrics.hs new file mode 100644 --- /dev/null +++ b/src/Atelier/Effects/Monitoring/Metrics.hs @@ -0,0 +1,125 @@ +-- | Metrics effect for Prometheus metrics collection. +-- +-- Provides operations for tracking application metrics. +-- +-- == Basic Usage +-- +-- @ +-- myComponent :: (Metrics :> es) => Eff es () +-- myComponent = do +-- gaugeSet "hoard_connected_peers" 5.0 +-- counterInc "hoard_blocks_received_total" +-- @ +-- +-- == Available Metric Types +-- +-- * Gauges: Point-in-time values that can go up or down +-- * Counters: Monotonically increasing values +-- * Histograms: Distributions with buckets for measuring durations +module Atelier.Effects.Monitoring.Metrics + ( -- * Effect + Metrics + + -- * Gauge Operations + , gaugeSet + , gaugeInc + , gaugeDec + + -- * Counter Operations + , counterInc + , counterAdd + + -- * Histogram Operations + , histogramObserve + , withHistogramTiming + + -- * Export Operations + , exportMetrics + + -- * Interpreters + , runMetrics + , runMetricsNoOp + ) where + +import Data.Time.Clock (diffUTCTime) +import Effectful (Effect, IOE) +import Effectful.Dispatch.Dynamic (interpret, interpretWith, localSeqUnlift) +import Effectful.TH (makeEffect) + +import Prometheus qualified as Prom +import Prometheus.Metric.GHC qualified as GHC + +import Atelier.Effects.Clock (Clock, currentTime) +import Atelier.Effects.Monitoring.Tracing (Tracing, withSpan) + +import Atelier.Effects.Monitoring.Metrics.Registry qualified as Registry + + +-- | Metrics effect for tracking application metrics +data Metrics :: Effect where + -- | Set a named gauge to a specific value. + GaugeSet :: Text -> Double -> Metrics m () + -- | Increment a named gauge by 1. + GaugeInc :: Text -> Metrics m () + -- | Decrement a named gauge by 1. + GaugeDec :: Text -> Metrics m () + -- | Increment a named counter by 1. + CounterInc :: Text -> Metrics m () + -- | Add a value to a named counter. + CounterAdd :: Text -> Double -> Metrics m () + -- | Observe a value in a named histogram. + HistogramObserve :: Text -> Double -> Metrics m () + -- | Time an action and record its duration to a histogram metric + WithHistogramTiming :: Text -> m a -> Metrics m a + -- | Export all collected metrics in Prometheus text format. + ExportMetrics :: Metrics m Text + + +makeEffect ''Metrics + + +-- | Run the Metrics effect with prometheus-client +-- +-- Initializes the metric registry and registers GHC metrics automatically. +runMetrics + :: forall es a + . (Clock :> es, IOE :> es, Tracing :> es) + => Eff (Metrics : es) a + -> Eff es a +runMetrics action = do + -- Initialize metrics registry and register GHC metrics + + handles <- withSpan "metrics.setup" do + void $ liftIO $ Prom.register GHC.ghcMetrics + liftIO Registry.initMetricHandles + + interpretWith action \env -> \case + GaugeSet name value -> withSpan "metrics.gauge_set" $ liftIO $ Registry.setGauge handles name value + GaugeInc name -> withSpan "metrics.gauge_inc" $ liftIO $ Registry.incGauge handles name + GaugeDec name -> withSpan "metrics.gauge_dec" $ liftIO $ Registry.decGauge handles name + CounterInc name -> withSpan "metrics.counter_inc" $ liftIO $ Registry.incCounter handles name + CounterAdd name value -> withSpan "metrics.counter_add" $ liftIO $ Registry.addCounter handles name value + HistogramObserve name value -> withSpan "metrics.histogram_observe" $ liftIO $ Registry.observeHistogram handles name value + WithHistogramTiming metricName eff -> do + start <- currentTime + result <- localSeqUnlift env \unlift -> unlift eff + end <- currentTime + let duration = realToFrac $ diffUTCTime end start + withSpan "metrics.with_histogram_timing.histogram_observe" + $ liftIO + $ Registry.observeHistogram handles metricName duration + pure result + ExportMetrics -> withSpan "metrics.export_metrics" $ liftIO $ decodeUtf8 <$> Prom.exportMetricsAsText + + +-- | No-op interpreter that discards all metrics operations +runMetricsNoOp :: Eff (Metrics : es) a -> Eff es a +runMetricsNoOp = interpret \env -> \case + GaugeSet _ _ -> pure () + GaugeInc _ -> pure () + GaugeDec _ -> pure () + CounterInc _ -> pure () + CounterAdd _ _ -> pure () + HistogramObserve _ _ -> pure () + WithHistogramTiming _ eff -> localSeqUnlift env \unlift -> unlift eff + ExportMetrics -> pure "" diff --git a/src/Atelier/Effects/Monitoring/Metrics/Registry.hs b/src/Atelier/Effects/Monitoring/Metrics/Registry.hs new file mode 100644 --- /dev/null +++ b/src/Atelier/Effects/Monitoring/Metrics/Registry.hs @@ -0,0 +1,119 @@ +-- | Metric registry for managing Prometheus metrics. +-- +-- The actual implementation of the `runMetrics` handler. +module Atelier.Effects.Monitoring.Metrics.Registry + ( MetricHandles + , initMetricHandles + , setGauge + , incGauge + , decGauge + , incCounter + , addCounter + , observeHistogram + ) where + +import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef) + +import Data.Map.Strict qualified as Map +import Prometheus qualified as Prom + + +-- | Mutable handles to the registered Prometheus metrics, keyed by name and +-- created on first use. +data MetricHandles = MetricHandles + { gauges :: IORef (Map Text Prom.Gauge) + , counters :: IORef (Map Text Prom.Counter) + , histograms :: IORef (Map Text Prom.Histogram) + } + + +-- | Initialize empty metric handles +initMetricHandles :: IO MetricHandles +initMetricHandles = do + MetricHandles + <$> newIORef mempty + <*> newIORef mempty + <*> newIORef mempty + + +-- | Set a gauge to a specific value +setGauge :: MetricHandles -> Text -> Double -> IO () +setGauge handles name value = do + g <- getOrCreateGauge handles name + Prom.setGauge g value + + +-- | Increment a gauge by 1 +incGauge :: MetricHandles -> Text -> IO () +incGauge handles name = do + g <- getOrCreateGauge handles name + Prom.incGauge g + + +-- | Decrement a gauge by 1 +decGauge :: MetricHandles -> Text -> IO () +decGauge handles name = do + g <- getOrCreateGauge handles name + Prom.decGauge g + + +-- | Increment a counter by 1 +incCounter :: MetricHandles -> Text -> IO () +incCounter handles name = do + c <- getOrCreateCounter handles name + Prom.incCounter c + + +-- | Add a value to a counter +addCounter :: MetricHandles -> Text -> Double -> IO () +addCounter handles name value = do + c <- getOrCreateCounter handles name + void $ Prom.addCounter c value + + +-- | Observe a value in a histogram +observeHistogram :: MetricHandles -> Text -> Double -> IO () +observeHistogram handles name value = do + h <- getOrCreateHistogram handles name + Prom.observe h value + + +-- | Get or create a gauge metric +getOrCreateGauge :: MetricHandles -> Text -> IO Prom.Gauge +getOrCreateGauge handles name = do + gaugeMap <- readIORef handles.gauges + case Map.lookup name gaugeMap of + Just g -> pure g + Nothing -> do + g <- Prom.register $ Prom.gauge (Prom.Info name "") + atomicModifyIORef' handles.gauges $ \m -> + (Map.insert name g m, ()) + pure g + + +-- | Get or create a counter metric +getOrCreateCounter :: MetricHandles -> Text -> IO Prom.Counter +getOrCreateCounter handles name = do + counterMap <- readIORef handles.counters + case Map.lookup name counterMap of + Just c -> pure c + Nothing -> do + c <- Prom.register $ Prom.counter (Prom.Info name "") + atomicModifyIORef' handles.counters $ \m -> + (Map.insert name c m, ()) + pure c + + +-- | Get or create a histogram metric +getOrCreateHistogram :: MetricHandles -> Text -> IO Prom.Histogram +getOrCreateHistogram handles name = do + histogramMap <- readIORef handles.histograms + case Map.lookup name histogramMap of + Just h -> pure h + Nothing -> do + -- Default buckets for duration metrics: 1ms, 10ms, 100ms, 1s, 10s + let buckets = [0.001, 0.01, 0.1, 1.0, 10.0] + h <- Prom.register $ Prom.histogram (Prom.Info name "") buckets + atomicModifyIORef' handles.histograms $ \m -> + (Map.insert name h m, ()) + pure h diff --git a/src/Atelier/Effects/Monitoring/Metrics/Server.hs b/src/Atelier/Effects/Monitoring/Metrics/Server.hs new file mode 100644 --- /dev/null +++ b/src/Atelier/Effects/Monitoring/Metrics/Server.hs @@ -0,0 +1,50 @@ +-- | Effect for running the Prometheus metrics HTTP server. +-- +-- Wraps a Warp server that exposes the current metrics registry over HTTP, so +-- callers start it through the effect system rather than raw 'IO'. +module Atelier.Effects.Monitoring.Metrics.Server + ( -- * Effect + MetricsServer + + -- * Operations + , runMetricsServer + + -- * Interpreters + , runMetricsServerIO + ) where + +import Effectful (Effect, IOE) +import Effectful.Dispatch.Dynamic (interpret_) +import Effectful.TH (makeEffect) +import Network.HTTP.Types (status200) +import Network.Wai (Application, responseLBS) + +import Network.Wai.Handler.Warp qualified as Warp +import Prometheus qualified as Prom + + +-- | Running the Prometheus metrics HTTP server. +data MetricsServer :: Effect where + -- | Start an HTTP server that exposes Prometheus metrics at any path on the + -- given port. Blocks until the server stops, which normally only happens on + -- error; intended to be run in a background thread. + RunMetricsServer :: Int -> MetricsServer m () + + +makeEffect ''MetricsServer + + +-- | Interpret 'MetricsServer' by running a real Warp HTTP server. +runMetricsServerIO :: (IOE :> es) => Eff (MetricsServer : es) a -> Eff es a +runMetricsServerIO = interpret_ \case + RunMetricsServer port -> liftIO $ Warp.run port metricsApp + + +metricsApp :: Application +metricsApp _req respond = do + payload <- Prom.exportMetricsAsText + respond + $ responseLBS + status200 + [("Content-Type", "text/plain; version=0.0.4; charset=utf-8")] + payload diff --git a/src/Atelier/Effects/Monitoring/Tracing.hs b/src/Atelier/Effects/Monitoring/Tracing.hs new file mode 100644 --- /dev/null +++ b/src/Atelier/Effects/Monitoring/Tracing.hs @@ -0,0 +1,293 @@ +-- | Tracing effect for distributed tracing with OpenTelemetry. +-- +-- Provides operations for creating spans, adding attributes, and propagating trace context. +-- +-- == Basic Usage +-- +-- @ +-- myHandler :: (Tracing :> es) => Eff es () +-- myHandler = do +-- withSpan "api.create_user" $ do +-- addAttribute "user.id" "123" +-- -- ... do work ... +-- setStatus Ok +-- @ +module Atelier.Effects.Monitoring.Tracing + ( -- * Effect + Tracing (..) + + -- * Span Operations + , withSpan + , withSpanLinked + , withLinkPropagation + , addAttribute + , addEvent + , setStatus + , getSpanContext + , OT.ToAttribute (..) + , Attr (..) + , ToAttributeShow (..) + + -- * Interpreters + , runTracing + , runTracingFromConfig + , runTracingNoOp + + -- * Configuration + , TracingConfig (..) + + -- * Re-exports + , SpanStatus (..) + , SpanContext + ) where + +import Data.Aeson (FromJSON, ToJSON) +import Data.Default (Default (..)) +import Effectful (Effect, IOE, Limit (..), Persistence (..), UnliftStrategy (..)) +import Effectful.Dispatch.Dynamic (interposeWith, interpret, interpretWith, localSeqUnlift, localUnlift) +import Effectful.Exception (bracket, onException) +import Effectful.Reader.Static (Reader, ask) +import Effectful.TH (makeEffect) + +import Data.HashMap.Strict qualified as HashMap +import OpenTelemetry.Context qualified as Context +import OpenTelemetry.Context.ThreadLocal qualified as ThreadLocal +import OpenTelemetry.Trace qualified as OT +import OpenTelemetry.Trace.Core qualified as OT + +import Atelier.Effects.Timeout (Timeout, timeout) +import Atelier.Time (Second) +import Atelier.Types.QuietSnake (QuietSnake (..)) +import Atelier.Types.WithDefaults (WithDefaults (..)) + +import Atelier.Effects.Monitoring.Tracing.Provider qualified as Provider + + +-- | Tracing configuration for OpenTelemetry +data TracingConfig = TracingConfig + { enabled :: Bool + -- ^ Enable tracing + , serviceName :: Text + -- ^ Service name for traces + , otlpEndpoint :: Text + -- ^ OTLP endpoint (e.g., "http://localhost:4318") + } + deriving stock (Eq, Generic, Show) + deriving (ToJSON) via QuietSnake TracingConfig + deriving (FromJSON) via WithDefaults (QuietSnake TracingConfig) + + +instance Default TracingConfig where + def = + TracingConfig + { enabled = False + , serviceName = "hoard" + , otlpEndpoint = "http://localhost:4318" + } + + +-- | Span status for indicating success or failure +data SpanStatus + = -- | The operation completed successfully. + Ok + | -- | The operation failed, with an explanatory message. + Error Text + deriving stock (Eq, Show) + + +-- | Opaque span context for correlating traces with metrics +type SpanContext = OT.SpanContext + + +-- | Tracing effect for distributed tracing +data Tracing :: Effect where + -- | Execute an action within a named span (bracket-style, automatic cleanup) + WithSpan :: Text -> m a -> Tracing m a + -- | Execute an action within a named span with links to other span contexts + WithSpanLinked :: Text -> [SpanContext] -> m a -> Tracing m a + -- | Add an attribute to the current span + AddAttribute :: (OT.ToAttribute attr) => Text -> attr -> Tracing m () + -- | Add an event to the current span + AddEvent :: (OT.ToAttribute attr) => Text -> [(Text, attr)] -> Tracing m () + -- | Set the status of the current span + SetStatus :: SpanStatus -> Tracing m () + -- | Get the current span context (for exemplars) + GetSpanContext :: Tracing m (Maybe SpanContext) + -- | Get the current OpenTelemetry context (internal use) + GetCurrentContext :: Tracing m Context.Context + + +-- | Useful to create a heterogeneous list of attribute values. These two are equivalent: +-- +-- @ +-- addEvent "foo" [OT.toAttribute 1, OT.toAttribute "foo"] +-- addEvent "foo" [Attr 1, Attr "foo"] +-- @ +data Attr where + Attr :: (OT.ToAttribute a) => a -> Attr + + +instance OT.ToAttribute Attr where + toAttribute (Attr a) = OT.toAttribute a + + +-- | Wrapper that turns any 'Show'able value into an OpenTelemetry attribute via +-- its 'Show' instance. Derive an attribute instance @via 'ToAttributeShow' T@, +-- or wrap a value directly. +newtype ToAttributeShow a = ToAttributeShow + { getToAttributeShow :: a + } + + +instance (Show a) => OT.ToPrimitiveAttribute (ToAttributeShow a) where + toPrimitiveAttribute = OT.TextAttribute . show . getToAttributeShow + + +instance (Show a) => OT.ToAttribute (ToAttributeShow a) + + +makeEffect ''Tracing + + +-- | Run an action with automatic span link propagation for fire-and-forget forks. +-- +-- Any span created with no current parent (i.e., a root span) will automatically +-- receive a link to @parentCtx@. Nested spans inside those are unaffected — they +-- already have a parent and follow normal child semantics. +-- +-- This avoids the trace growth problem caused by parent-child propagation in +-- long-running loops: each loop iteration's work becomes its own trace, linked +-- back to the originating span rather than piling spans onto a single trace. +withLinkPropagation :: (Tracing :> es) => Maybe SpanContext -> Eff es a -> Eff es a +withLinkPropagation Nothing action = action +withLinkPropagation (Just parentSpanCtx) action = + interposeWith action $ \env -> \case + WithSpan name m -> do + currentCtx <- getCurrentContext + localUnlift env (ConcUnlift Persistent Unlimited) $ \unlift -> + case Context.lookupSpan currentCtx of + Nothing -> withSpanLinked name [parentSpanCtx] (unlift m) + Just _ -> withSpan name (unlift m) + WithSpanLinked name ctxs m -> + localUnlift env (ConcUnlift Persistent Unlimited) $ \unlift -> + withSpanLinked name ctxs (unlift m) + AddAttribute key val -> addAttribute key val + AddEvent name attrs -> addEvent name attrs + SetStatus status -> setStatus status + GetSpanContext -> getSpanContext + GetCurrentContext -> getCurrentContext + + +-- | Run the Tracing effect with OpenTelemetry +-- +-- Initializes the tracer provider and manages span lifecycle. +runTracing + :: (IOE :> es, Timeout :> es) + => Bool + -- ^ Tracing enabled flag + -> Text + -- ^ Service name + -> Text + -- ^ OTLP endpoint + -> Eff (Tracing : es) a + -> Eff es a +runTracing enabled serviceName otlpEndpoint action + | not enabled = runTracingNoOp action + | otherwise = + bracket + (liftIO $ Provider.initTracingState serviceName otlpEndpoint) + (\tracingState -> void $ timeout (3 :: Second) $ liftIO $ Provider.shutdownTracingState tracingState) + $ \tracingState -> interpretWith action $ \env -> \case + WithSpan spanName innerAction -> localSeqUnlift env $ \unlift -> do + currentCtx <- liftIO ThreadLocal.getContext + newSpan <- liftIO $ OT.createSpan tracingState.tracer currentCtx spanName OT.defaultSpanArguments + let newCtx = Context.insertSpan newSpan currentCtx + oldCtx <- liftIO $ ThreadLocal.attachContext newCtx + innerResult <- + unlift innerAction `onException` do + liftIO $ OT.setStatus newSpan (OT.Error "Exception occurred") + -- Restore the old context + liftIO $ void $ case oldCtx of + Just ctx -> ThreadLocal.attachContext ctx + Nothing -> ThreadLocal.detachContext + liftIO $ OT.endSpan newSpan Nothing + pure innerResult + WithSpanLinked spanName linkedContexts innerAction -> localSeqUnlift env $ \unlift -> do + currentCtx <- liftIO ThreadLocal.getContext + let spanArgs = + OT.defaultSpanArguments + { OT.links = map (\ctx -> OT.NewLink ctx mempty) linkedContexts + } + newSpan <- liftIO $ OT.createSpan tracingState.tracer currentCtx spanName spanArgs + let newCtx = Context.insertSpan newSpan currentCtx + oldCtx <- liftIO $ ThreadLocal.attachContext newCtx + innerResult <- + unlift innerAction `onException` do + liftIO $ OT.setStatus newSpan (OT.Error "Exception occurred") + + -- Restore the old context + liftIO $ void $ case oldCtx of + Just ctx -> ThreadLocal.attachContext ctx + Nothing -> ThreadLocal.detachContext + liftIO $ OT.endSpan newSpan Nothing + pure innerResult + AddAttribute key value -> do + currentCtx <- liftIO ThreadLocal.getContext + case Context.lookupSpan currentCtx of + Just currentSpan -> + liftIO $ OT.addAttribute currentSpan key (OT.toAttribute value) + Nothing -> + -- No active span, ignore + pure () + AddEvent eventName attributes -> do + currentCtx <- liftIO ThreadLocal.getContext + case Context.lookupSpan currentCtx of + Just currentSpan -> do + let attrMap = HashMap.fromList $ map (\(k, v) -> (k, OT.toAttribute v)) attributes + let event = OT.NewEvent eventName attrMap Nothing + liftIO $ OT.addEvent currentSpan event + Nothing -> + -- No active span, ignore + pure () + SetStatus status -> do + currentCtx <- liftIO ThreadLocal.getContext + case Context.lookupSpan currentCtx of + Just currentSpan -> + liftIO $ case status of + Ok -> OT.setStatus currentSpan OT.Ok + Error msg -> OT.setStatus currentSpan (OT.Error msg) + Nothing -> + -- No active span, ignore + pure () + GetSpanContext -> do + currentCtx <- liftIO ThreadLocal.getContext + case Context.lookupSpan currentCtx of + Just currentSpan -> do + spanCtx <- liftIO $ OT.getSpanContext currentSpan + pure $ Just spanCtx + Nothing -> pure Nothing + GetCurrentContext -> liftIO ThreadLocal.getContext + + +-- | Run the Tracing effect with config from Reader +-- +-- Convenience wrapper that reads TracingConfig from the Reader effect. +runTracingFromConfig + :: (IOE :> es, Reader TracingConfig :> es, Timeout :> es) + => Eff (Tracing : es) a + -> Eff es a +runTracingFromConfig action = do + TracingConfig {enabled, serviceName, otlpEndpoint} <- ask + runTracing enabled serviceName otlpEndpoint action + + +-- | No-op interpreter that discards all tracing operations +runTracingNoOp :: Eff (Tracing : es) a -> Eff es a +runTracingNoOp = interpret $ \env -> \case + WithSpan _ act -> localSeqUnlift env $ \unlift -> unlift act + WithSpanLinked _ _ act -> localSeqUnlift env $ \unlift -> unlift act + AddAttribute _ _ -> pure () + AddEvent _ _ -> pure () + SetStatus _ -> pure () + GetSpanContext -> pure Nothing + GetCurrentContext -> pure Context.empty diff --git a/src/Atelier/Effects/Monitoring/Tracing/Provider.hs b/src/Atelier/Effects/Monitoring/Tracing/Provider.hs new file mode 100644 --- /dev/null +++ b/src/Atelier/Effects/Monitoring/Tracing/Provider.hs @@ -0,0 +1,70 @@ +-- | Tracing provider for managing OpenTelemetry TracerProvider and span stack. +-- +-- The actual implementation backing the `runTracing` handler. +module Atelier.Effects.Monitoring.Tracing.Provider + ( TracingState (..) + , initTracingState + , shutdownTracingState + ) where + +import System.Environment (setEnv) + +import OpenTelemetry.Attributes qualified as OT +import OpenTelemetry.Trace qualified as OT + + +-- | Handles to the initialised OpenTelemetry tracer provider and a tracer +-- derived from it. +data TracingState = TracingState + { tracerProvider :: OT.TracerProvider + -- ^ The global tracer provider; flushed and shut down on teardown. + , tracer :: OT.Tracer + -- ^ The tracer used to create spans. + } + + +-- | Initialize tracing state using global tracer provider +-- +-- This uses initializeGlobalTracerProvider which reads configuration from +-- environment variables (OTEL_SERVICE_NAME, OTEL_EXPORTER_OTLP_ENDPOINT) +-- and properly sets up ID generation. +initTracingState + :: Text + -- ^ Service name + -> Text + -- ^ OTLP endpoint (e.g., "http://localhost:4318") + -> IO TracingState +initTracingState serviceName otlpEndpoint = do + -- Set environment variables for OpenTelemetry SDK + setEnv "OTEL_SERVICE_NAME" (toString serviceName) + setEnv "OTEL_EXPORTER_OTLP_ENDPOINT" (toString otlpEndpoint) + + -- Initialize the global tracer provider + -- This reads from environment variables and sets up proper ID generation + provider <- OT.initializeGlobalTracerProvider + + -- Create instrumentation library + let instrumentationLibrary = + OT.InstrumentationLibrary + serviceName -- Library name + "" -- Version (empty for now) + "" -- Schema URL (empty for now) + OT.emptyAttributes -- Attributes + + -- Get a tracer from the provider + let tracerInstance = OT.makeTracer provider instrumentationLibrary OT.tracerOptions + + pure + TracingState + { tracerProvider = provider + , tracer = tracerInstance + } + + +-- | Shutdown tracing state gracefully +-- +-- Flushes any remaining spans and cleans up resources. +shutdownTracingState :: TracingState -> IO () +shutdownTracingState tracingState = do + -- Force shutdown of the tracer provider to flush spans + OT.shutdownTracerProvider tracingState.tracerProvider diff --git a/src/Atelier/Effects/Publishing/Traced.hs b/src/Atelier/Effects/Publishing/Traced.hs new file mode 100644 --- /dev/null +++ b/src/Atelier/Effects/Publishing/Traced.hs @@ -0,0 +1,57 @@ +module Atelier.Effects.Publishing.Traced + ( runPubSub + ) +where + +import Atelier.Effects.Chan (Chan) +import Atelier.Effects.Clock (Clock) +import Atelier.Effects.Publishing.Pub (Pub (..)) +import Atelier.Effects.Publishing.Sub (Sub (..)) +import Data.Time (UTCTime) +import Effectful.Dispatch.Dynamic (interpretWith, interpretWith_, localSeqUnlift) + +import Atelier.Effects.Chan qualified as Chan +import Atelier.Effects.Clock qualified as Clock + +import Atelier.Effects.Monitoring.Tracing (SpanContext, Tracing) + +import Atelier.Effects.Monitoring.Tracing qualified as Tracing + + +-- | Internal wrapper for events with trace context +data TracedEvent event = TracedEvent + { event :: event + , timestamp :: UTCTime + , publisherSpanContext :: Maybe SpanContext + } + + +-- | Runs 'Pub' and 'Sub' effects with an internal channel for a specific event +-- type. Automatically captures span context from the publisher and creates +-- linked spans in listeners. +runPubSub + :: forall event es a + . ( Chan :> es + , Clock :> es + , Tracing :> es + ) + => Eff (Pub event : Sub event : es) a -> Eff es a +runPubSub action = do + (inChan, _) <- Chan.newChan @(TracedEvent event) + + let handlePub eff = interpretWith_ eff \case + Publish event -> do + timestamp <- Clock.currentTime + -- Capture the current span context from the publisher + publisherSpanContext <- Tracing.getSpanContext + Chan.writeChan inChan TracedEvent {event, timestamp, publisherSpanContext} + + handleSub eff = interpretWith eff \env -> \case + ListenWith onSubscribed listener -> localSeqUnlift env \unlift -> do + chan <- Chan.dupChan inChan + unlift onSubscribed + forever do + TracedEvent {event, timestamp, publisherSpanContext} <- Chan.readChan chan + Tracing.withLinkPropagation publisherSpanContext $ unlift $ listener timestamp event + + handleSub . handlePub $ action diff --git a/test/Driver.hs b/test/Driver.hs new file mode 100644 --- /dev/null +++ b/test/Driver.hs @@ -0,0 +1,2 @@ +{-# OPTIONS_GHC -F -pgmF tasty-discover #-} + diff --git a/test/Unit/Atelier/Effects/Conc/TracedSpec.hs b/test/Unit/Atelier/Effects/Conc/TracedSpec.hs new file mode 100644 --- /dev/null +++ b/test/Unit/Atelier/Effects/Conc/TracedSpec.hs @@ -0,0 +1,130 @@ +module Unit.Atelier.Effects.Conc.TracedSpec (spec_ConcTraced) where + +import Control.Concurrent (newEmptyMVar, putMVar, takeMVar, threadDelay) +import Data.IORef (IORef, modifyIORef', newIORef, readIORef) +import Effectful (IOE, runEff) +import Effectful.Concurrent (runConcurrent) +import Effectful.Dispatch.Dynamic (interpret, localUnlift) +import OpenTelemetry.Common (TraceFlags (..)) +import OpenTelemetry.Internal.Trace.Id (SpanId (..), TraceId (..)) +import OpenTelemetry.Trace.TraceState (TraceState (..)) +import Test.Hspec (Spec, describe, it, shouldReturn, shouldSatisfy) + +import OpenTelemetry.Context qualified as Context +import OpenTelemetry.Context.ThreadLocal qualified as ThreadLocal +import OpenTelemetry.Trace.Core qualified as OT + +import Atelier.Effects.Conc.Traced (await, concStrat, fork, fork_, runConc) +import Atelier.Effects.Monitoring.Tracing (SpanContext, Tracing (..), withLinkPropagation, withSpan) + + +spec_ConcTraced :: Spec +spec_ConcTraced = do + describe "withLinkPropagation" $ do + it "passes through when no parent context" $ do + ops <- newIORef [] + runEff . runTracingCapture ops + $ withLinkPropagation Nothing + $ withSpan "foo" + $ pure () + readIORef ops `shouldReturn` [PlainSpan "foo"] + + it "converts root span to linked span when parent context given" $ do + ops <- newIORef [] + runEff . runTracingCapture ops + $ withLinkPropagation (Just fakeSpanCtx) + $ withSpan "foo" + $ pure () + readIORef ops `shouldReturn` [LinkedSpan "foo"] + + it "does not convert nested spans (they already have a parent)" $ do + ops <- newIORef [] + runEff . runTracingCapture ops + $ withSpan "outer" + $ withLinkPropagation (Just fakeSpanCtx) + $ withSpan "inner" + $ pure () + readIORef ops `shouldReturn` [PlainSpan "outer", PlainSpan "inner"] + + describe "withConcTracingLinks" $ do + it "does not link when forking outside any span" $ do + ops <- newIORef [] + runEff . runConcurrent . runTracingCapture ops . runConc $ do + t <- fork $ withSpan "child" $ pure () + await t + readIORef ops `shouldReturn` [PlainSpan "child"] + + it "links forked thread's root span to the parent span" $ do + ops <- newIORef [] + runEff . runConcurrent . runTracingCapture ops . runConc $ do + withSpan "parent" $ do + t <- fork $ withSpan "child" $ pure () + await t + readIORef ops `shouldReturn` [PlainSpan "parent", LinkedSpan "child"] + + it "links fork_ thread's root span to the parent span" $ do + ops <- newIORef [] + spanRecorded <- newEmptyMVar + runEff . runConcurrent . runTracingCapture ops . runConc $ do + withSpan "parent" $ do + fork_ $ do + withSpan "bg" $ pure () + liftIO $ putMVar spanRecorded () + liftIO $ forever $ threadDelay maxBound + liftIO $ takeMVar spanRecorded + readIORef ops >>= (`shouldSatisfy` elem (LinkedSpan "bg")) + + +-------------------------------------------------------------------------------- +-- Test Infrastructure +-------------------------------------------------------------------------------- + +data SpanOp + = PlainSpan Text + | LinkedSpan Text + deriving stock (Eq, Show) + + +-- | A fake but valid span context for testing. +fakeSpanCtx :: SpanContext +fakeSpanCtx = + OT.SpanContext + (TraceFlags 0x01) + False + (TraceId "\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\1") + (SpanId "\0\0\0\0\0\0\0\1") + (TraceState []) + + +-- | Test interpreter that records 'PlainSpan' vs 'LinkedSpan' operations. +-- +-- Uses real thread-local storage for context tracking so that forked threads +-- correctly start with an empty context (as they would in production), allowing +-- 'withLinkPropagation' to distinguish root spans from nested ones. +runTracingCapture + :: (IOE :> es) + => IORef [SpanOp] + -> Eff (Tracing : es) a + -> Eff es a +runTracingCapture log = interpret $ \env -> \case + WithSpan name act -> do + liftIO $ modifyIORef' log (<> [PlainSpan name]) + currentCtx <- liftIO ThreadLocal.getContext + let newCtx = Context.insertSpan (OT.wrapSpanContext fakeSpanCtx) currentCtx + oldCtx <- liftIO $ ThreadLocal.attachContext newCtx + result <- localUnlift env concStrat $ \unlift -> unlift act + liftIO $ void $ case oldCtx of + Just ctx -> ThreadLocal.attachContext ctx + Nothing -> ThreadLocal.detachContext + pure result + WithSpanLinked name _ act -> do + liftIO $ modifyIORef' log (<> [LinkedSpan name]) + localUnlift env concStrat $ \unlift -> unlift act + GetSpanContext -> do + ctx <- liftIO ThreadLocal.getContext + liftIO $ traverse OT.getSpanContext (Context.lookupSpan ctx) + GetCurrentContext -> + liftIO ThreadLocal.getContext + AddAttribute _ _ -> pure () + AddEvent _ _ -> pure () + SetStatus _ -> pure ()