diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,7 +1,30 @@ +### 0.4.0.0 + +- Fix space leak in `processHeapProfSampleData`. +- Support aggregation (via configuration files). +- Support statistics with the `--stats` flag: + + ``` + ┌────────┬──────────┬──────────────┬───────────────┬───────────────────┐ + │ Item │ Action │ Total (item) │ Rate (item/s) │ Peak (item/batch) │ + ╞════════╪══════════╪══════════════╪═══════════════╪═══════════════════╡ + │ Event │ Received │ 66036 │ 6603/s │ 22197/s │ + ├────────┼──────────┼──────────────┼───────────────┼───────────────────┤ + │ Metric │ Exported │ 274 │ 45/s │ 59/s │ + ├────────┼──────────┼──────────────┼───────────────┼───────────────────┤ + │ │ Rejected │ 0 │ 0/s │ 0/s │ + ├────────┼──────────┼──────────────┼───────────────┼───────────────────┤ + │ Span │ Exported │ 19372 │ 3228/s │ 5979/s │ + ├────────┼──────────┼──────────────┼───────────────┼───────────────────┤ + │ │ Rejected │ 0 │ 0/s │ 0/s │ + └────────┴──────────┴──────────────┴───────────────┴───────────────────┘ + ``` + ### 0.3.0.0 +- Support configuration files (via `--config`; see `--print-defaults`). - **BREAKING**: Remove `--otelcol-no-metrics` and `--otelcol-no-traces` flags. -- Support configuration file. + The behaviour of these flags is superceded by the configuration files. ### 0.2.0.0 diff --git a/data/default.yaml b/data/default.yaml new file mode 100644 --- /dev/null +++ b/data/default.yaml @@ -0,0 +1,56 @@ +processors: + metrics: + blocks_size: + aggregate: by_batch # one of [null, by_batch] + description: The current heap size, calculated by the allocated number of blocks. + enabled: true + name: ghc_eventlog_BlocksSize + capability_usage: + aggregate: by_batch # one of [null, by_batch] + description: The duration of each capability usage span. + enabled: true + name: ghc_eventlog_CapabilityUsageDuration + heap_allocated: + aggregate: by_batch # one of [null, by_batch] + description: The size of a newly allocated chunk of heap. + enabled: true + name: ghc_eventlog_HeapAllocated + heap_live: + aggregate: by_batch + description: The current heap size, calculated by the allocated number of megablocks. + enabled: true + name: ghc_eventlog_HeapLive + heap_prof_sample: + aggregate: by_batch # one of [null, by_batch] + description: A heap profile sample. + enabled: true + name: ghc_eventlog_HeapProfSample + heap_size: + aggregate: by_batch # one of [null, by_batch] + description: The current heap size, calculated by the allocated number of megablocks. + enabled: true + name: ghc_eventlog_HeapSize + mem_current: + aggregate: by_batch # one of [null, by_batch] + description: The number of megablocks currently allocated. + enabled: true + name: ghc_eventlog_MemCurrent + mem_needed: + aggregate: by_batch # one of [null, by_batch] + description: The number of megablocks currently needed. + enabled: true + name: ghc_eventlog_MemNeeded + mem_returned: + aggregate: by_batch # one of [null, by_batch] + description: The number of megablocks currently being returned to the OS. + enabled: true + name: ghc_eventlog_MemReturned + spans: + capability_usage: + description: A span of capability usage (either mutator thread or garbage collection). + enabled: true + name: ghc_eventlog_CapabilityUsage + thread_state: + description: A span of thread state changes (either running or stopped). + enabled: true + name: ghc_eventlog_ThreadState diff --git a/eventlog-live-otelcol.cabal b/eventlog-live-otelcol.cabal --- a/eventlog-live-otelcol.cabal +++ b/eventlog-live-otelcol.cabal @@ -1,7 +1,7 @@ -cabal-version: 3.0 -name: eventlog-live-otelcol -version: 0.3.0.0 -synopsis: Stream eventlog data to the OpenTelemetry Collector. +cabal-version: 3.0 +name: eventlog-live-otelcol +version: 0.4.0.0 +synopsis: Stream eventlog data to the OpenTelemetry Collector. description: This executable supports live streaming of eventlog data into the OpenTelemetry Collector. @@ -13,11 +13,15 @@ > [--batch-interval NUM] [--eventlog-log-file FILE] > [-h Tcmdyrbi] [--service-name STRING] > [-v|--verbosity quiet|error|warning|info|debug|0-4] - > [--config FILE] --otelcol-host HOST + > [-s|--stats] [--config FILE] --otelcol-host HOST > [--otelcol-port PORT] [--otelcol-authority HOST] > [--otelcol-ssl] [--otelcol-certificate-store FILE] > [--otelcol-ssl-key-log FILE | > --otelcol-ssl-key-log-from-env] + > [--enable-my-eventlog-socket-unix SOCKET] + > [--enable-my-ghc-debug-socket | + > --enable-my-ghc-debug-socket-unix SOCKET | + > --enable-my-ghc-debug-socket-tcp ADDRESS] > [--print-defaults] > > Available options: @@ -35,6 +39,7 @@ > --service-name STRING The name of the profiled service. > -v,--verbosity quiet|error|warning|info|debug|0-4 > The verbosity threshold for logging. + > -s,--stats Display runtime statistics. > --config FILE The path to a detailed configuration file. > --print-defaults Print default configuration options that can be used > in config.yaml @@ -52,22 +57,45 @@ > Use file to log SSL keys. > --otelcol-ssl-key-log-from-env > Use SSLKEYLOGFILE to log SSL keys. + > + > Debug Options + > --enable-my-eventlog-socket-unix SOCKET + > Enable the eventlog socket for this program on the + > given Unix socket. + > --enable-my-ghc-debug-socket + > Enable ghc-debug for this program. + > --enable-my-ghc-debug-socket-unix SOCKET + > Enable ghc-debug for this program on the given Unix + > socket. + > --enable-my-ghc-debug-socket-tcp ADDRESS + > Enable ghc-debug for this program on the given TCP + > socket specified as 'host:port'. -license: BSD-3-Clause -license-file: LICENSE -author: Wen Kokke -maintainer: wen@well-typed.com -copyright: (c) 2025 Well-Typed -build-type: Simple -category: Debug, Monitoring, System -extra-doc-files: CHANGELOG.md -tested-with: GHC ==9.2.8 || ==9.4.8 || ==9.6.7 || ==9.8.4 || ==9.10.2 +license: BSD-3-Clause +license-file: LICENSE +author: Wen Kokke +maintainer: wen@well-typed.com +copyright: (c) 2025 Well-Typed +build-type: Simple +category: Debug, Monitoring, System +extra-doc-files: CHANGELOG.md +extra-source-files: data/default.yaml +tested-with: + GHC ==9.2.8 || ==9.4.8 || ==9.6.7 || ==9.8.4 || ==9.10.2 source-repository head type: git location: https://github.com/well-typed/eventlog-live.git subdir: eventlog-live-otelcol +-- 2025-11-06: +-- This flag should enable switching between template-haskell and +-- template-haskell-lift, because the latter is not on nixpkgs. +flag use-template-haskell-lift + description: Use template-haskell-lift in place of template-haskell + default: False + manual: False + common language ghc-options: -Wall -Wcompat -Widentities -Wprepositive-qualified-module @@ -82,6 +110,7 @@ DeriveFoldable DeriveFunctor DeriveGeneric + DeriveLift DeriveTraversable DerivingStrategies DuplicateRecordFields @@ -97,12 +126,14 @@ NoFieldSelectors NumericUnderscores OverloadedRecordDot + PatternSynonyms RankNTypes RecordWildCards ScopedTypeVariables TupleSections TypeApplications TypeFamilies + TypeOperators library import: language @@ -110,8 +141,14 @@ exposed-modules: GHC.Eventlog.Live.Otelcol GHC.Eventlog.Live.Otelcol.Config + GHC.Eventlog.Live.Otelcol.Exporter + GHC.Eventlog.Live.Otelcol.Stats other-modules: + GHC.Eventlog.Live.Otelcol.Config.Default + GHC.Eventlog.Live.Otelcol.Config.Default.Raw + GHC.Eventlog.Live.Otelcol.Config.Types + Language.Haskell.TH.Lift.Compat Options.Applicative.Compat Paths_eventlog_live_otelcol System.Random.Compat @@ -119,11 +156,15 @@ autogen-modules: Paths_eventlog_live_otelcol build-depends: , aeson >=2.2 && <2.3 + , ansi-terminal >=1.1 && <1.2 , base >=4.16 && <4.22 , bytestring >=0.11 && <0.13 - , data-default >=0.8 && <0.9 + , data-default >=0.2 && <0.9 , dlist >=1.0 && <1.1 - , eventlog-live >=0.3 && <0.4 + , eventlog-live >=0.4 && <0.5 + , eventlog-socket >=0.1.0 && <0.2 + , file-embed >=0.0.16 && <0.1 + , ghc-debug-stub >=0.1 && <1.0 , ghc-events >=0.20 && <0.21 , grapesy >=1.0.0 && <1.2 , hashable >=1.4 && <1.6 @@ -133,13 +174,23 @@ , optparse-applicative >=0.17 && <0.20 , proto-lens >=0.7.1 && <0.8 , random >=1.2 && <1.4 + , strict-list >=0.1 && <0.2 + , table-layout >=1.0 && <1.1 , text >=1.2 && <2.2 , unordered-containers >=0.2.20 && <0.3 , vector >=0.11 && <0.14 , yaml >=0.11 && <0.12 + if flag(use-template-haskell-lift) + build-depends: template-haskell-lift >=0.1 && <0.2 + cpp-options: -DEVENTLOG_LIVE_OTELCOL_USE_TEMPLATE_HASKELL_LIFT + + else + build-depends: template-haskell >=2.2 && <3.0 + executable eventlog-live-otelcol import: language main-is: Main.hs hs-source-dirs: app build-depends: eventlog-live-otelcol + ghc-options: -threaded -rtsopts -finfo-table-map diff --git a/src/GHC/Eventlog/Live/Otelcol.hs b/src/GHC/Eventlog/Live/Otelcol.hs --- a/src/GHC/Eventlog/Live/Otelcol.hs +++ b/src/GHC/Eventlog/Live/Otelcol.hs @@ -1,6 +1,5 @@ {-# LANGUAGE OverloadedStrings #-} {-# OPTIONS_GHC -Wno-name-shadowing #-} -{-# OPTIONS_GHC -Wno-orphans #-} {- | Module : GHC.Eventlog.Live.Otelcol @@ -12,32 +11,35 @@ main, ) where -import Control.Applicative (asum) -import Control.Exception (Exception (..), catch) +import Control.Applicative (Alternative (..), asum) import Control.Monad.IO.Class (MonadIO (..)) import Data.ByteString (ByteString) +import Data.Coerce (Coercible, coerce) import Data.DList (DList) import Data.DList qualified as D import Data.Default (Default (..)) import Data.Either (partitionEithers) -import Data.Foldable (for_, traverse_) +import Data.Foldable (for_) import Data.Functor ((<&>)) import Data.HashMap.Strict qualified as M import Data.Hashable (Hashable) -import Data.Int (Int64) import Data.Machine (MachineT, Process, ProcessT, asParts, await, construct, mapping, repeatedly, stopped, yield, (~>)) import Data.Machine.Fanout (fanout) import Data.Maybe (fromMaybe, mapMaybe) import Data.ProtoLens (Message (defMessage)) +import Data.Proxy (Proxy (..)) +import Data.Semigroup (Last (..), Sum (..)) import Data.Text (Text) import Data.Text qualified as T import Data.Text.Encoding qualified as TE -import Data.Vector qualified as V import Data.Version (showVersion) import Data.Word (Word32, Word64) import Data.Yaml qualified as Y +import GHC.Debug.Stub qualified as GHC.Debug (withGhcDebug, withGhcDebugTCP, withGhcDebugUnix) import GHC.Eventlog.Live.Data.Attribute -import GHC.Eventlog.Live.Data.Metric +import GHC.Eventlog.Live.Data.Group (Group, GroupBy, GroupedBy) +import GHC.Eventlog.Live.Data.Group qualified as DG +import GHC.Eventlog.Live.Data.Metric (Metric (..)) import GHC.Eventlog.Live.Logger (logDebug, logError) import GHC.Eventlog.Live.Machine.Analysis.Capability (CapabilityUsageSpan) import GHC.Eventlog.Live.Machine.Analysis.Capability qualified as M @@ -52,22 +54,22 @@ import GHC.Eventlog.Live.Options import GHC.Eventlog.Live.Otelcol.Config (Config) import GHC.Eventlog.Live.Otelcol.Config qualified as C +import GHC.Eventlog.Live.Otelcol.Config.Default.Raw (defaultConfigString) +import GHC.Eventlog.Live.Otelcol.Exporter (exportResourceMetrics, exportResourceSpans) +import GHC.Eventlog.Live.Otelcol.Stats (Stat (..), eventCountTick, processStats) import GHC.Eventlog.Live.Socket (runWithEventlogSource) import GHC.Eventlog.Live.Verbosity (Verbosity) +import GHC.Eventlog.Socket qualified as Eventlog.Socket import GHC.RTS.Events (Event (..), HeapProfBreakdown (..), ThreadId) import GHC.Records (HasField) -import Lens.Family2 ((&), (.~), (^.)) +import Lens.Family2 ((&), (.~)) import Network.GRPC.Client qualified as G -import Network.GRPC.Client.StreamType.IO qualified as G import Network.GRPC.Common qualified as G -import Network.GRPC.Common.Protobuf (Protobuf) -import Network.GRPC.Common.Protobuf qualified as G import Options.Applicative qualified as O import Options.Applicative.Compat qualified as OC import Options.Applicative.Extra qualified as OE import Paths_eventlog_live_otelcol qualified as EventlogLive import Proto.Opentelemetry.Proto.Collector.Metrics.V1.MetricsService qualified as OMS -import Proto.Opentelemetry.Proto.Collector.Metrics.V1.MetricsService_Fields qualified as OMS import Proto.Opentelemetry.Proto.Collector.Trace.V1.TraceService qualified as OTS import Proto.Opentelemetry.Proto.Collector.Trace.V1.TraceService_Fields qualified as OTS import Proto.Opentelemetry.Proto.Common.V1.Common qualified as OC @@ -76,9 +78,10 @@ import Proto.Opentelemetry.Proto.Metrics.V1.Metrics_Fields qualified as OM import Proto.Opentelemetry.Proto.Trace.V1.Trace qualified as OT import Proto.Opentelemetry.Proto.Trace.V1.Trace_Fields qualified as OT +import System.Exit (exitFailure) import System.Random (StdGen, initStdGen) import System.Random.Compat (uniformByteString) -import Text.Printf (printf) +import Text.Read (readEither) {- | The main function for @eventlog-live-otelcol@. @@ -86,86 +89,76 @@ main :: IO () main = do Options{..} <- O.execParser options - let OpenTelemetryCollectorOptions{..} = openTelemetryCollectorOptions - -- Define the error writer: - let errorWriter = repeatedly $ await >>= logError verbosity . T.pack - -- Read the configuration file: - config <- flip (maybe (pure def)) maybeConfigFile $ \configFile -> do - logDebug verbosity $ "Reading configuration file from " <> T.pack configFile - config <- C.readConfig configFile - logDebug verbosity $ "Configuration file:\n" <> (TE.decodeUtf8Lenient . Y.encode $ config) - pure config - let attrServiceName = ("service.name", maybe AttrNull (AttrText . (.serviceName)) maybeServiceName) - G.withConnection G.def openTelemetryCollectorServer $ \conn -> do - runWithEventlogSource - verbosity - eventlogSocket - eventlogSocketTimeout - eventlogSocketTimeoutExponent - batchInterval - Nothing - maybeEventlogLogFile - $ fanout - [ M.validateInput verbosity 10 - , M.counterByTick verbosity "events" - , M.liftTick M.withStartTime - ~> fanout - [ processHeapEvents config verbosity maybeHeapProfBreakdown - ~> mapping (fmap Left) - , processThreadEvents config verbosity - ] - ~> mapping (partitionEithers . D.toList) - ~> fanout - [ runIf (shouldExportMetrics config) $ - mapping fst - ~> fanout - [ M.counterBy verbosity "metrics" (sum . fmap countMetricDataPoints) - , asScopeMetrics - [ OM.scope .~ eventlogLiveScope - ] - ~> asResourceMetric [] - ~> asExportMetricServiceRequest - ~> exportResourceMetrics conn - ~> mapping (either displayException displayException) - ~> errorWriter - ] - , runIf (shouldExportSpans config) $ - mapping snd - ~> fanout - [ M.counterBy verbosity "spans" (fromIntegral . length) - , asScopeSpans - [ OT.scope .~ eventlogLiveScope - ] - ~> asResourceSpan - [ OT.resource - .~ messageWith - [ OM.attributes .~ mapMaybe toMaybeKeyValue [attrServiceName] - ] - ] - ~> asExportTraceServiceRequest - ~> exportResourceSpans conn - ~> mapping (either displayException displayException) - ~> errorWriter - ] - ] - ] -countMetricDataPoints :: OM.Metric -> Word -countMetricDataPoints metric = - fromIntegral $ - case metric ^. OM.maybe'data' of - Nothing -> 0 - Just (OM.Metric'Gauge gauge) -> - V.length (gauge ^. OM.vec'dataPoints) - Just (OM.Metric'Sum sum) -> - V.length (sum ^. OM.vec'dataPoints) - Just (OM.Metric'Histogram histogram) -> - V.length (histogram ^. OM.vec'dataPoints) - Just (OM.Metric'ExponentialHistogram exponentialHistogram) -> - V.length (exponentialHistogram ^. OM.vec'dataPoints) - Just (OM.Metric'Summary summary) -> - V.length (summary ^. OM.vec'dataPoints) + -- Instument THIS PROGRAM with eventlog-socket and/or ghc-debug. + let MyDebugOptions{..} = myDebugOptions + withMyEventlogSocket maybeMyEventlogSocket + withMyGhcDebug verbosity maybeMyGhcDebugSocket $ do + -- + -- Read the configuration file. + config <- flip (maybe (pure def)) maybeConfigFile $ \configFile -> do + logDebug verbosity $ "Reading configuration file from " <> T.pack configFile + config <- C.readConfig configFile + logDebug verbosity $ "Configuration file:\n" <> (TE.decodeUtf8Lenient . Y.encode $ config) + pure config + -- Create the service name attribute. + let attrServiceName = ("service.name", maybe AttrNull (AttrText . (.serviceName)) maybeServiceName) + + -- Open a connection to the OpenTelemetry Collector. + let OpenTelemetryCollectorOptions{..} = openTelemetryCollectorOptions + G.withConnection G.def openTelemetryCollectorServer $ \conn -> do + runWithEventlogSource + verbosity + eventlogSocket + eventlogSocketTimeout + eventlogSocketTimeoutExponent + batchInterval + Nothing + maybeEventlogLogFile + $ fanout + [ M.validateInput verbosity 10 + , eventCountTick ~> mapping (D.singleton . EventCountStat) + , M.liftTick M.withStartTime + ~> fanout + [ processHeapEvents config verbosity maybeHeapProfBreakdown + ~> mapping (fmap Left) + , processThreadEvents config verbosity + ] + ~> mapping (partitionEithers . D.toList) + ~> fanout + [ runIf (shouldExportMetrics config) $ + mapping fst + ~> fanout + [ asScopeMetrics + [ OM.scope .~ eventlogLiveScope + ] + ~> asResourceMetric [] + ~> asExportMetricServiceRequest + ~> exportResourceMetrics conn + ~> mapping (D.singleton . ExportMetricsResultStat) + ] + , runIf (shouldExportSpans config) $ + mapping snd + ~> fanout + [ asScopeSpans + [ OT.scope .~ eventlogLiveScope + ] + ~> asResourceSpan + [ OT.resource + .~ messageWith + [ OM.attributes .~ mapMaybe toMaybeKeyValue [attrServiceName] + ] + ] + ~> asExportTraceServiceRequest + ~> exportResourceSpans conn + ~> mapping (D.singleton . ExportTraceResultStat) + ] + ] + ] + ~> asParts + ~> processStats verbosity stats batchInterval 10 + {- | Internal helper. Determine whether or not any spans should be exported. @@ -233,11 +226,9 @@ ) ~> fanout [ runIf (C.processorEnabled (.metrics) (.capabilityUsage) config) $ - M.liftTick - ( M.processCapabilityUsageMetrics - ~> asNumberDataPoint - ) - ~> M.batchByTickList + M.liftTick M.processCapabilityUsageMetrics + ~> aggregate viaSum (C.processorAggregationStrategy (.metrics) (.capabilityUsage) config) + ~> mapping (fmap toNumberDataPoint) ~> asSum [ OM.aggregationTemporality .~ OM.AGGREGATION_TEMPORALITY_DELTA , OM.isMonotonic .~ True @@ -330,13 +321,73 @@ ] -------------------------------------------------------------------------------- +-- Metric Aggregation + +data Aggregators a b = Aggregators + { nothing :: Process (Tick a) b + , byBatch :: Process (Tick a) b + } + +{- | +Internal helper. +Aggregate items based on the provided aggregators and aggregation strategy. +-} +aggregate :: Aggregators a b -> Maybe C.AggregationStrategy -> Process (Tick a) b +aggregate Aggregators{..} = \case + Nothing -> nothing + Just C.AggregationStrategyByBatch -> byBatch + +{- | +Internal helper. +Metric aggregators via the `Semigroup` instance for `Sum`. +-} +viaSum :: forall a. (Num a) => Aggregators (Metric a) [Metric a] +viaSum = + Aggregators + { nothing = M.batchByTickList + , byBatch = + -- TODO: emit group sample counts as separate metric + byBatchVia (Proxy @(Metric (Sum a))) + ~> mapping (fmap (.representative)) + } + +{- | +Internal helper. +Metric aggregators via the `Semigroup` instance for `Last`. +-} +viaLast :: forall a. (GroupBy a) => Aggregators a [a] +viaLast = + Aggregators + { nothing = M.batchByTickList + , byBatch = + -- TODO: emit group sample counts as separate metric + byBatchVia (Proxy @(Last a)) + ~> mapping (fmap (.representative)) + } + +{- | +Internal helper. +This function aggregates items via a `Semigroup` instance and grouped by the `GroupBy` instance. +-} +byBatchVia :: + forall a b. + (Coercible a b, GroupBy b, Semigroup b) => + Proxy b -> + Process (Tick a) [Group a] +byBatchVia (Proxy :: Proxy b) = + mapping (fmap DG.singleton . coerce @(Tick a) @(Tick b)) + ~> M.aggregateByTick @(GroupedBy b) + ~> mapping (coerce @[Group b] @[Group a] . DG.groups) + +-------------------------------------------------------------------------------- -- HeapAllocated processHeapAllocated :: Config -> Process (Tick (WithStartTime Event)) (DList OM.Metric) processHeapAllocated config = runIf (C.processorEnabled (.metrics) (.heapAllocated) config) $ - M.liftTick (M.processHeapAllocatedData ~> asNumberDataPoint) - ~> M.batchByTickList + M.liftTick M.processHeapAllocatedData + ~> aggregate viaSum (C.processorAggregationStrategy (.metrics) (.heapAllocated) config) + ~> mapping (fmap toNumberDataPoint) ~> asSum [ OM.aggregationTemporality .~ OM.AGGREGATION_TEMPORALITY_DELTA , OM.isMonotonic .~ True @@ -350,8 +401,9 @@ processHeapSize :: Config -> Process (Tick (WithStartTime Event)) (DList OM.Metric) processHeapSize config = runIf (C.processorEnabled (.metrics) (.heapSize) config) $ - M.liftTick (M.processHeapSizeData ~> asNumberDataPoint) - ~> M.batchByTickList + M.liftTick M.processHeapSizeData + ~> aggregate viaLast (C.processorAggregationStrategy (.metrics) (.heapSize) config) + ~> mapping (fmap toNumberDataPoint) ~> asGauge ~> asMetricWith config (.heapSize) [OM.unit .~ "By"] ~> mapping D.singleton @@ -362,8 +414,9 @@ processBlocksSize :: Config -> Process (Tick (WithStartTime Event)) (DList OM.Metric) processBlocksSize config = runIf (C.processorEnabled (.metrics) (.blocksSize) config) $ - M.liftTick (M.processBlocksSizeData ~> asNumberDataPoint) - ~> M.batchByTickList + M.liftTick M.processBlocksSizeData + ~> aggregate viaLast (C.processorAggregationStrategy (.metrics) (.blocksSize) config) + ~> mapping (fmap toNumberDataPoint) ~> asGauge ~> asMetricWith config (.blocksSize) [OM.unit .~ "By"] ~> mapping D.singleton @@ -374,8 +427,9 @@ processHeapLive :: Config -> Process (Tick (WithStartTime Event)) (DList OM.Metric) processHeapLive config = runIf (C.processorEnabled (.metrics) (.heapLive) config) $ - M.liftTick (M.processHeapLiveData ~> asNumberDataPoint) - ~> M.batchByTickList + M.liftTick M.processHeapLiveData + ~> aggregate viaLast (C.processorAggregationStrategy (.metrics) (.heapLive) config) + ~> mapping (fmap toNumberDataPoint) ~> asGauge ~> asMetricWith config (.heapLive) [OM.unit .~ "By"] ~> mapping D.singleton @@ -387,20 +441,25 @@ processMemReturn config = runIf (shouldComputeMemReturn config) $ M.liftTick M.processMemReturnData - ~> M.batchByTickList ~> fanout [ runIf (C.processorEnabled (.metrics) (.memCurrent) config) $ - M.liftBatch (asMemCurrent ~> asNumberDataPoint) + M.liftTick asMemCurrent + ~> aggregate viaLast (C.processorAggregationStrategy (.metrics) (.memCurrent) config) + ~> mapping (fmap toNumberDataPoint) ~> asGauge ~> asMetricWith config (.memCurrent) [OM.unit .~ "{mblock}"] ~> mapping D.singleton , runIf (C.processorEnabled (.metrics) (.memNeeded) config) $ - M.liftBatch (asMemNeeded ~> asNumberDataPoint) + M.liftTick asMemNeeded + ~> aggregate viaLast (C.processorAggregationStrategy (.metrics) (.memNeeded) config) + ~> mapping (fmap toNumberDataPoint) ~> asGauge ~> asMetricWith config (.memNeeded) [OM.unit .~ "{mblock}"] ~> mapping D.singleton , runIf (C.processorEnabled (.metrics) (.memReturned) config) $ - M.liftBatch (asMemReturned ~> asNumberDataPoint) + M.liftTick asMemReturned + ~> aggregate viaLast (C.processorAggregationStrategy (.metrics) (.memReturned) config) + ~> mapping (fmap toNumberDataPoint) ~> asGauge ~> asMetricWith config (.memReturned) [OM.unit .~ "{mblock}"] ~> mapping D.singleton @@ -428,6 +487,20 @@ -------------------------------------------------------------------------------- -- HeapProfSample +heapProfSampleDataAggregators :: Aggregators M.HeapProfSampleData [Metric Word64] +heapProfSampleDataAggregators = Aggregators{..} + where + nothing :: Process (Tick M.HeapProfSampleData) [Metric Word64] + nothing = + M.batchByTickList + ~> mapping (concatMap M.heapProfSamples) + byBatch :: Process (Tick M.HeapProfSampleData) [Metric Word64] + byBatch = + -- TODO: emit group sample counts as separate metric + mapping (fmap (DG.singleton . Last)) + ~> M.aggregateByTick @(GroupedBy (Last M.HeapProfSampleData)) + ~> mapping (concatMap (M.heapProfSamples . getLast) . DG.elems) + processHeapProfSample :: (MonadIO m) => Config -> @@ -436,8 +509,9 @@ ProcessT m (Tick (WithStartTime Event)) (DList OM.Metric) processHeapProfSample config verbosity maybeHeapProfBreakdown = runIf (C.processorEnabled (.metrics) (.heapProfSample) config) $ - M.liftTick (M.processHeapProfSampleData verbosity maybeHeapProfBreakdown ~> asNumberDataPoint) - ~> M.batchByTickList + M.liftTick (M.processHeapProfSampleData verbosity maybeHeapProfBreakdown) + ~> aggregate heapProfSampleDataAggregators (C.processorAggregationStrategy (.metrics) (.heapProfSample) config) + ~> mapping (fmap toNumberDataPoint) ~> asGauge ~> asMetricWith config (.heapProfSample) [OM.unit .~ "By"] ~> mapping D.singleton @@ -528,9 +602,6 @@ | null dataPoints = Nothing | otherwise = Just . OM.Metric'Sum . messageWith $ (OM.dataPoints .~ dataPoints) : mod -asNumberDataPoint :: (IsNumberDataPoint'Value v) => Process (Metric v) OM.NumberDataPoint -asNumberDataPoint = mapping toNumberDataPoint - -------------------------------------------------------------------------------- -- Interpret data -------------------------------------------------------------------------------- @@ -670,7 +741,7 @@ [ OM.maybe'value .~ Just (toNumberDataPoint'Value i.value) , OM.timeUnixNano .~ fromMaybe 0 i.maybeTimeUnixNano , OM.startTimeUnixNano .~ fromMaybe 0 i.maybeStartTimeUnixNano - , OM.attributes .~ mapMaybe toMaybeKeyValue i.attr + , OM.attributes .~ mapMaybe toMaybeKeyValue (toList i.attrs) ] toMaybeKeyValue :: Attr -> Maybe OC.KeyValue @@ -705,94 +776,6 @@ messageWith = foldr ($) defMessage -------------------------------------------------------------------------------- --- OpenTelemetry Collector exporters --------------------------------------------------------------------------------- - --------------------------------------------------------------------------------- --- OpenTelemetry Collector Trace Exporter - -data ExportTraceError - = ExportTraceError - { rejectedSpans :: Int64 - , errorMessage :: Text - } - deriving (Show) - -instance Exception ExportTraceError where - displayException :: ExportTraceError -> String - displayException ExportTraceError{..} = - printf "Error: OpenTelemetry Collector rejected %d data points with message: %s" rejectedSpans errorMessage - -sendResourceSpans :: G.Connection -> OTS.ExportTraceServiceRequest -> IO (Maybe (Either G.GrpcError ExportTraceError)) -sendResourceSpans conn exportTraceServiceRequest = - fmap (fmap Right) doGrpc `catch` handleGrpcError - where - doGrpc :: IO (Maybe ExportTraceError) - doGrpc = do - G.nonStreaming conn (G.rpc @(Protobuf OTS.TraceService "export")) (G.Proto exportTraceServiceRequest) >>= \case - G.Proto resp - | resp ^. OTS.partialSuccess . OTS.rejectedSpans == 0 -> pure Nothing - | otherwise -> pure $ Just ExportTraceError{..} - where - rejectedSpans = resp ^. OTS.partialSuccess . OTS.rejectedSpans - errorMessage = resp ^. OTS.partialSuccess . OTS.errorMessage - handleGrpcError :: G.GrpcError -> IO (Maybe (Either G.GrpcError ExportTraceError)) - handleGrpcError = pure . pure . Left - -exportResourceSpans :: G.Connection -> ProcessT IO OTS.ExportTraceServiceRequest (Either G.GrpcError ExportTraceError) -exportResourceSpans conn = - repeatedly $ - await >>= \exportTraceServiceRequest -> - liftIO (sendResourceSpans conn exportTraceServiceRequest) - >>= traverse_ yield - -type instance G.RequestMetadata (Protobuf OTS.TraceService meth) = G.NoMetadata -type instance G.ResponseInitialMetadata (Protobuf OTS.TraceService meth) = G.NoMetadata -type instance G.ResponseTrailingMetadata (Protobuf OTS.TraceService meth) = G.NoMetadata - --------------------------------------------------------------------------------- --- OpenTelemetry Collector Metrics Exporter - -data ExportMetricsError - = ExportMetricsError - { rejectedDataPoints :: Int64 - , errorMessage :: Text - } - deriving (Show) - -instance Exception ExportMetricsError where - displayException :: ExportMetricsError -> String - displayException ExportMetricsError{..} = - printf "Error: OpenTelemetry Collector rejected %d data points with message: %s" rejectedDataPoints errorMessage - -sendResourceMetrics :: G.Connection -> OMS.ExportMetricsServiceRequest -> IO (Maybe (Either G.GrpcError ExportMetricsError)) -sendResourceMetrics conn exportMetricsServiceRequest = - fmap (fmap Right) doGrpc `catch` handleGrpcError - where - doGrpc :: IO (Maybe ExportMetricsError) - doGrpc = do - G.nonStreaming conn (G.rpc @(Protobuf OMS.MetricsService "export")) (G.Proto exportMetricsServiceRequest) >>= \case - G.Proto resp - | resp ^. OMS.partialSuccess . OMS.rejectedDataPoints == 0 -> pure Nothing - | otherwise -> pure $ Just ExportMetricsError{..} - where - rejectedDataPoints = resp ^. OMS.partialSuccess . OMS.rejectedDataPoints - errorMessage = resp ^. OMS.partialSuccess . OMS.errorMessage - handleGrpcError :: G.GrpcError -> IO (Maybe (Either G.GrpcError ExportMetricsError)) - handleGrpcError = pure . pure . Left - -exportResourceMetrics :: G.Connection -> ProcessT IO OMS.ExportMetricsServiceRequest (Either G.GrpcError ExportMetricsError) -exportResourceMetrics conn = - repeatedly $ - await >>= \exportMetricsServiceRequest -> - liftIO (sendResourceMetrics conn exportMetricsServiceRequest) - >>= traverse_ yield - -type instance G.RequestMetadata (Protobuf OMS.MetricsService meth) = G.NoMetadata -type instance G.ResponseInitialMetadata (Protobuf OMS.MetricsService meth) = G.NoMetadata -type instance G.ResponseTrailingMetadata (Protobuf OMS.MetricsService meth) = G.NoMetadata - --------------------------------------------------------------------------------- -- Options -------------------------------------------------------------------------------- @@ -815,8 +798,10 @@ , maybeHeapProfBreakdown :: Maybe HeapProfBreakdown , maybeServiceName :: Maybe ServiceName , verbosity :: Verbosity + , stats :: Bool , maybeConfigFile :: Maybe FilePath , openTelemetryCollectorOptions :: OpenTelemetryCollectorOptions + , myDebugOptions :: MyDebugOptions } optionsParser :: O.Parser Options @@ -830,10 +815,108 @@ <*> O.optional heapProfBreakdownParser <*> O.optional serviceNameParser <*> verbosityParser + <*> statsParser <*> O.optional configFileParser <*> openTelemetryCollectorOptionsParser + <*> myDebugOptionsParser -------------------------------------------------------------------------------- +-- Debug Options + +data MyDebugOptions = MyDebugOptions + { maybeMyEventlogSocket :: Maybe MyEventlogSocket + , maybeMyGhcDebugSocket :: Maybe MyGhcDebugSocket + } + +myDebugOptionsParser :: O.Parser MyDebugOptions +myDebugOptionsParser = + OC.parserOptionGroup "Debug Options" $ + MyDebugOptions + <$> O.optional myEventlogSocketParser + <*> O.optional myGhcDebugSocketParser + +-------------------------------------------------------------------------------- +-- My Eventlog Socket + +newtype MyEventlogSocket + = MyEventlogSocketUnix FilePath + +myEventlogSocketParser :: O.Parser MyEventlogSocket +myEventlogSocketParser = + MyEventlogSocketUnix + <$> O.strOption + ( O.long "enable-my-eventlog-socket-unix" + <> O.metavar "SOCKET" + <> O.help "Enable the eventlog socket for this program on the given Unix socket." + ) + +{- | +Set @eventlog-socket@ as the eventlog writer. +-} +withMyEventlogSocket :: Maybe MyEventlogSocket -> IO () +withMyEventlogSocket maybeMyEventlogSocket = + for_ maybeMyEventlogSocket $ \(MyEventlogSocketUnix myEventlogSocket) -> + Eventlog.Socket.startWait myEventlogSocket + +-------------------------------------------------------------------------------- +-- My GHC Debug + +data MyGhcDebugSocket + = MyGhcDebugSocketDefault + | MyGhcDebugSocketUnix FilePath + | MyGhcDebugSocketTcp String + deriving (Show) + +myGhcDebugSocketParser :: O.Parser MyGhcDebugSocket +myGhcDebugSocketParser = + myGhcDebugSocketDefaultParser + <|> myGhcDebugSocketUnixParser + <|> myGhcDebugSocketTcpParser + where + myGhcDebugSocketDefaultParser = + O.flag' + MyGhcDebugSocketDefault + ( O.long "enable-my-ghc-debug-socket" + <> O.help "Enable ghc-debug for this program." + ) + myGhcDebugSocketUnixParser = + MyGhcDebugSocketUnix + <$> O.strOption + ( O.long "enable-my-ghc-debug-socket-unix" + <> O.metavar "SOCKET" + <> O.help "Enable ghc-debug for this program on the given Unix socket." + ) + myGhcDebugSocketTcpParser = + MyGhcDebugSocketUnix + <$> O.strOption + ( O.long "enable-my-ghc-debug-socket-tcp" + <> O.metavar "ADDRESS" + <> O.help "Enable ghc-debug for this program on the given TCP socket specified as 'host:port'." + ) + +{- | +Internal helper. +Start @ghc-debug@ on the given `MyGhcDebugSocket`. +-} +withMyGhcDebug :: Verbosity -> Maybe MyGhcDebugSocket -> IO a -> IO a +withMyGhcDebug verbosity maybeMyGhcDebugSocket action = + case maybeMyGhcDebugSocket of + Nothing -> action + Just MyGhcDebugSocketDefault -> + GHC.Debug.withGhcDebug action + Just (MyGhcDebugSocketUnix myGhcDebugSocketUnix) -> + GHC.Debug.withGhcDebugUnix myGhcDebugSocketUnix action + Just (MyGhcDebugSocketTcp myGhcDebugSocketTcp) -> do + let (host, port) = break (== ':') myGhcDebugSocketTcp + case readEither port of + Left _parseError -> do + logError verbosity $ + "Could not parse ghc-debug TCP address " <> T.pack myGhcDebugSocketTcp <> "." + exitFailure + Right portWord16 -> + GHC.Debug.withGhcDebugTCP host portWord16 action + +-------------------------------------------------------------------------------- -- Configuration configFileParser :: O.Parser FilePath @@ -846,13 +929,10 @@ defaultsPrinter :: O.Parser (a -> a) defaultsPrinter = - O.infoOption defaults . mconcat $ + O.infoOption defaultConfigString . mconcat $ [ O.long "print-defaults" , O.help "Print default configuration options that can be used in config.yaml" ] - where - defaults :: String - defaults = T.unpack (TE.decodeUtf8Lenient (Y.encode (def :: Config))) -------------------------------------------------------------------------------- -- Service Name diff --git a/src/GHC/Eventlog/Live/Otelcol/Config.hs b/src/GHC/Eventlog/Live/Otelcol/Config.hs --- a/src/GHC/Eventlog/Live/Otelcol/Config.hs +++ b/src/GHC/Eventlog/Live/Otelcol/Config.hs @@ -1,4 +1,6 @@ {-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE TemplateHaskell #-} +{-# OPTIONS_GHC -Wno-orphans #-} {- | Module : GHC.Eventlog.Live.Otelcol.Config @@ -12,6 +14,7 @@ Processors (..), Metrics (..), Spans (..), + AggregationStrategy (..), HeapAllocatedMetric (..), BlocksSizeMetric (..), HeapSizeMetric (..), @@ -26,16 +29,17 @@ processorEnabled, processorDescription, processorName, + processorAggregationStrategy, ) where import Control.Monad.IO.Class (MonadIO (..)) -import Data.Aeson.Types (Encoding, FromJSON (..), Options (..), Parser, SumEncoding (..), ToJSON (..), Value (..), camelTo2, defaultOptions, genericParseJSON, genericToEncoding, genericToJSON) import Data.Default (Default (..)) import Data.Maybe (fromMaybe) import Data.Monoid (Any (..), First (..)) import Data.Text (Text) import Data.Yaml qualified as Y -import GHC.Generics (Generic) +import GHC.Eventlog.Live.Otelcol.Config.Default (defaultConfig, getDefault) +import GHC.Eventlog.Live.Otelcol.Config.Types import GHC.Records (HasField) {- | @@ -44,444 +48,69 @@ readConfig :: (MonadIO m) => FilePath -> m Config readConfig = Y.decodeFileThrow -{- | -The configuration for @eventlog-live-otelcol@. --} -newtype Config = Config - { processors :: Maybe Processors - } - deriving (Generic) - -instance FromJSON Config where - parseJSON :: Value -> Parser Config - parseJSON = genericParseJSON encodingOptions - -instance ToJSON Config where - toJSON :: Config -> Value - toJSON = genericToJSON encodingOptions - toEncoding :: Config -> Encoding - toEncoding = genericToEncoding encodingOptions +------------------------------------------------------------------------------- +-- Default Instances +------------------------------------------------------------------------------- instance Default Config where def :: Config - def = - Config - { processors = Just def - } - -{- | -The configuration options for the processors. --} -data Processors = Processors - { metrics :: Maybe Metrics - , spans :: Maybe Spans - } - deriving (Generic) - -instance FromJSON Processors where - parseJSON :: Value -> Parser Processors - parseJSON = genericParseJSON encodingOptions - -instance ToJSON Processors where - toJSON :: Processors -> Value - toJSON = genericToJSON encodingOptions - toEncoding :: Processors -> Encoding - toEncoding = genericToEncoding encodingOptions + def = defaultConfig instance Default Processors where def :: Processors - def = - Processors - { metrics = Just def - , spans = Just def - } - -{- | -The configuration options for the metric processors. --} -data Metrics = Metrics - { heapAllocated :: Maybe HeapAllocatedMetric - , blocksSize :: Maybe BlocksSizeMetric - , heapSize :: Maybe HeapSizeMetric - , heapLive :: Maybe HeapLiveMetric - , memCurrent :: Maybe MemCurrentMetric - , memNeeded :: Maybe MemNeededMetric - , memReturned :: Maybe MemReturnedMetric - , heapProfSample :: Maybe HeapProfSampleMetric - , capabilityUsage :: Maybe CapabilityUsageMetric - } - deriving (Generic) - -instance FromJSON Metrics where - parseJSON :: Value -> Parser Metrics - parseJSON = genericParseJSON encodingOptions - -instance ToJSON Metrics where - toJSON :: Metrics -> Value - toJSON = genericToJSON encodingOptions - toEncoding :: Metrics -> Encoding - toEncoding = genericToEncoding encodingOptions + def = $(getDefault @'["processors"] defaultConfig) instance Default Metrics where def :: Metrics - def = - Metrics - { heapAllocated = Just def - , blocksSize = Just def - , heapSize = Just def - , heapLive = Just def - , memCurrent = Just def - , memNeeded = Just def - , memReturned = Just def - , heapProfSample = Just def - , capabilityUsage = Just def - } - -{- | -The configuration options for the span processors. --} -data Spans = Spans - { capabilityUsage :: Maybe CapabilityUsageSpan - , threadState :: Maybe ThreadStateSpan - } - deriving (Generic) - -instance FromJSON Spans where - parseJSON :: Value -> Parser Spans - parseJSON = genericParseJSON encodingOptions - -instance ToJSON Spans where - toJSON :: Spans -> Value - toJSON = genericToJSON encodingOptions - toEncoding :: Spans -> Encoding - toEncoding = genericToEncoding encodingOptions + def = $(getDefault @'["processors", "metrics"] defaultConfig) instance Default Spans where def :: Spans - def = - Spans - { capabilityUsage = Just def - , threadState = Just def - } - -{- | -The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Heap.processHeapAllocatedData`. --} -data HeapAllocatedMetric = HeapAllocatedMetric - { description :: Maybe Text - , enabled :: Bool - , name :: Text - } - deriving (Generic) - -instance FromJSON HeapAllocatedMetric where - parseJSON :: Value -> Parser HeapAllocatedMetric - parseJSON = genericParseJSON encodingOptions - -instance ToJSON HeapAllocatedMetric where - toJSON :: HeapAllocatedMetric -> Value - toJSON = genericToJSON encodingOptions - toEncoding :: HeapAllocatedMetric -> Encoding - toEncoding = genericToEncoding encodingOptions + def = $(getDefault @'["processors", "spans"] defaultConfig) instance Default HeapAllocatedMetric where def :: HeapAllocatedMetric - def = - HeapAllocatedMetric - { description = Just "The size of a newly allocated chunk of heap." - , enabled = True - , name = "ghc_eventlog_HeapAllocated" - } - -{- | -The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Heap.processHeapSizeData`. --} -data HeapSizeMetric = HeapSizeMetric - { description :: Maybe Text - , enabled :: Bool - , name :: Text - } - deriving (Generic) - -instance FromJSON HeapSizeMetric where - parseJSON :: Value -> Parser HeapSizeMetric - parseJSON = genericParseJSON encodingOptions - -instance ToJSON HeapSizeMetric where - toJSON :: HeapSizeMetric -> Value - toJSON = genericToJSON encodingOptions - toEncoding :: HeapSizeMetric -> Encoding - toEncoding = genericToEncoding encodingOptions - -instance Default HeapSizeMetric where - def :: HeapSizeMetric - def = - HeapSizeMetric - { description = Just "The current heap size, calculated by the allocated number of megablocks." - , enabled = True - , name = "ghc_eventlog_HeapSize" - } - -{- | -The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Heap.processBlocksSizeData`. --} -data BlocksSizeMetric = BlocksSizeMetric - { description :: Maybe Text - , enabled :: Bool - , name :: Text - } - deriving (Generic) - -instance FromJSON BlocksSizeMetric where - parseJSON :: Value -> Parser BlocksSizeMetric - parseJSON = genericParseJSON encodingOptions - -instance ToJSON BlocksSizeMetric where - toJSON :: BlocksSizeMetric -> Value - toJSON = genericToJSON encodingOptions - toEncoding :: BlocksSizeMetric -> Encoding - toEncoding = genericToEncoding encodingOptions + def = $(getDefault @'["processors", "metrics", "heapAllocated"] defaultConfig) instance Default BlocksSizeMetric where def :: BlocksSizeMetric - def = - BlocksSizeMetric - { description = Just "The current heap size, calculated by the allocated number of blocks." - , enabled = True - , name = "ghc_eventlog_BlocksSize" - } - -{- | -The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Heap.processHeapLiveData`. --} -data HeapLiveMetric = HeapLiveMetric - { description :: Maybe Text - , enabled :: Bool - , name :: Text - } - deriving (Generic) - -instance FromJSON HeapLiveMetric where - parseJSON :: Value -> Parser HeapLiveMetric - parseJSON = genericParseJSON encodingOptions + def = $(getDefault @'["processors", "metrics", "blocksSize"] defaultConfig) -instance ToJSON HeapLiveMetric where - toJSON :: HeapLiveMetric -> Value - toJSON = genericToJSON encodingOptions - toEncoding :: HeapLiveMetric -> Encoding - toEncoding = genericToEncoding encodingOptions +instance Default HeapSizeMetric where + def :: HeapSizeMetric + def = $(getDefault @'["processors", "metrics", "heapSize"] defaultConfig) instance Default HeapLiveMetric where def :: HeapLiveMetric - def = - HeapLiveMetric - { description = Just "The current heap size, calculated by the allocated number of megablocks." - , enabled = True - , name = "ghc_eventlog_HeapLive" - } - -{- | -The configuration options for the @memCurrent@ field `GHC.Eventlog.Live.Machine.Analysis.Heap.processMemReturnData`. --} -data MemCurrentMetric = MemCurrentMetric - { description :: Maybe Text - , enabled :: Bool - , name :: Text - } - deriving (Generic) - -instance FromJSON MemCurrentMetric where - parseJSON :: Value -> Parser MemCurrentMetric - parseJSON = genericParseJSON encodingOptions - -instance ToJSON MemCurrentMetric where - toJSON :: MemCurrentMetric -> Value - toJSON = genericToJSON encodingOptions - toEncoding :: MemCurrentMetric -> Encoding - toEncoding = genericToEncoding encodingOptions + def = $(getDefault @'["processors", "metrics", "heapLive"] defaultConfig) instance Default MemCurrentMetric where def :: MemCurrentMetric - def = - MemCurrentMetric - { description = Just "The number of megablocks currently allocated." - , enabled = True - , name = "ghc_eventlog_MemCurrent" - } - -{- | -The configuration options for the @memNeeded@ field `GHC.Eventlog.Live.Machine.Analysis.Heap.processMemReturnData`. --} -data MemNeededMetric = MemNeededMetric - { description :: Maybe Text - , enabled :: Bool - , name :: Text - } - deriving (Generic) - -instance FromJSON MemNeededMetric where - parseJSON :: Value -> Parser MemNeededMetric - parseJSON = genericParseJSON encodingOptions - -instance ToJSON MemNeededMetric where - toJSON :: MemNeededMetric -> Value - toJSON = genericToJSON encodingOptions - toEncoding :: MemNeededMetric -> Encoding - toEncoding = genericToEncoding encodingOptions + def = $(getDefault @'["processors", "metrics", "memCurrent"] defaultConfig) instance Default MemNeededMetric where def :: MemNeededMetric - def = - MemNeededMetric - { description = Just "The number of megablocks currently needed." - , enabled = True - , name = "ghc_eventlog_MemNeeded" - } - -{- | -The configuration options for the @memReturned@ field `GHC.Eventlog.Live.Machine.Analysis.Heap.processMemReturnData`. --} -data MemReturnedMetric = MemReturnedMetric - { description :: Maybe Text - , enabled :: Bool - , name :: Text - } - deriving (Generic) - -instance FromJSON MemReturnedMetric where - parseJSON :: Value -> Parser MemReturnedMetric - parseJSON = genericParseJSON encodingOptions - -instance ToJSON MemReturnedMetric where - toJSON :: MemReturnedMetric -> Value - toJSON = genericToJSON encodingOptions - toEncoding :: MemReturnedMetric -> Encoding - toEncoding = genericToEncoding encodingOptions + def = $(getDefault @'["processors", "metrics", "memNeeded"] defaultConfig) instance Default MemReturnedMetric where def :: MemReturnedMetric - def = - MemReturnedMetric - { description = Just "The number of megablocks currently being returned to the OS." - , enabled = True - , name = "ghc_eventlog_MemReturned" - } - -{- | -The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Heap.processHeapProfSampleData`. --} -data HeapProfSampleMetric = HeapProfSampleMetric - { description :: Maybe Text - , enabled :: Bool - , name :: Text - } - deriving (Generic) - -instance FromJSON HeapProfSampleMetric where - parseJSON :: Value -> Parser HeapProfSampleMetric - parseJSON = genericParseJSON encodingOptions - -instance ToJSON HeapProfSampleMetric where - toJSON :: HeapProfSampleMetric -> Value - toJSON = genericToJSON encodingOptions - toEncoding :: HeapProfSampleMetric -> Encoding - toEncoding = genericToEncoding encodingOptions + def = $(getDefault @'["processors", "metrics", "memReturned"] defaultConfig) instance Default HeapProfSampleMetric where def :: HeapProfSampleMetric - def = - HeapProfSampleMetric - { description = Just "A heap profile sample." - , enabled = True - , name = "ghc_eventlog_HeapProfSample" - } - -{- | -The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Capability.processCapabilityUsageMetrics`. --} -data CapabilityUsageMetric = CapabilityUsageMetric - { description :: Maybe Text - , enabled :: Bool - , name :: Text - } - deriving (Generic) - -instance FromJSON CapabilityUsageMetric where - parseJSON :: Value -> Parser CapabilityUsageMetric - parseJSON = genericParseJSON encodingOptions - -instance ToJSON CapabilityUsageMetric where - toJSON :: CapabilityUsageMetric -> Value - toJSON = genericToJSON encodingOptions - toEncoding :: CapabilityUsageMetric -> Encoding - toEncoding = genericToEncoding encodingOptions + def = $(getDefault @'["processors", "metrics", "heapProfSample"] defaultConfig) instance Default CapabilityUsageMetric where def :: CapabilityUsageMetric - def = - CapabilityUsageMetric - { description = Just "The duration of each capability usage span." - , enabled = True - , name = "ghc_eventlog_CapabilityUsageDuration" - } - -{- | -The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Capability.processCapabilityUsageSpans`. --} -data CapabilityUsageSpan = CapabilityUsageSpan - { description :: Maybe Text - , enabled :: Bool - , name :: Text - } - deriving (Generic) - -instance FromJSON CapabilityUsageSpan where - parseJSON :: Value -> Parser CapabilityUsageSpan - parseJSON = genericParseJSON encodingOptions - -instance ToJSON CapabilityUsageSpan where - toJSON :: CapabilityUsageSpan -> Value - toJSON = genericToJSON encodingOptions - toEncoding :: CapabilityUsageSpan -> Encoding - toEncoding = genericToEncoding encodingOptions + def = $(getDefault @'["processors", "metrics", "capabilityUsage"] defaultConfig) instance Default CapabilityUsageSpan where def :: CapabilityUsageSpan - def = - CapabilityUsageSpan - { description = Just "A span of capability usage (either by a mutator thread or the garbage collector)." - , enabled = True - , name = "ghc_eventlog_CapabilityUsage" - } - -{- | -The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Thread.processThreadStateSpans`. --} -data ThreadStateSpan = ThreadStateSpan - { description :: Maybe Text - , enabled :: Bool - , name :: Text - } - deriving (Generic) - -instance FromJSON ThreadStateSpan where - parseJSON :: Value -> Parser ThreadStateSpan - parseJSON = genericParseJSON encodingOptions - -instance ToJSON ThreadStateSpan where - toJSON :: ThreadStateSpan -> Value - toJSON = genericToJSON encodingOptions - toEncoding :: ThreadStateSpan -> Encoding - toEncoding = genericToEncoding encodingOptions + def = $(getDefault @'["processors", "spans", "capabilityUsage"] defaultConfig) instance Default ThreadStateSpan where def :: ThreadStateSpan - def = - ThreadStateSpan - { description = Just "A span of thread state changes (either running or stopped)." - , enabled = True - , name = "ghc_eventlog_ThreadState" - } + def = $(getDefault @'["processors", "spans", "threadState"] defaultConfig) ------------------------------------------------------------------------------- -- Accessors @@ -490,40 +119,50 @@ {- | Get whether or not a processor is enabled. -} -processorEnabled :: (HasField "enabled" b Bool) => (Processors -> Maybe a) -> (a -> Maybe b) -> Config -> Bool -processorEnabled group field = getAny . with (.processors) (with group (with field (Any . (.enabled)))) +processorEnabled :: + (HasField "enabled" b Bool) => + (Processors -> Maybe a) -> + (a -> Maybe b) -> + Config -> + Bool +processorEnabled group field = + getAny . with (.processors) (with group (with field (Any . (.enabled)))) {- | Get the description corresponding to a processor. -} -processorDescription :: (Default b, HasField "description" b (Maybe Text)) => (Processors -> Maybe a) -> (a -> Maybe b) -> Config -> Maybe Text -processorDescription group field = (.description) . fromMaybe def . getFirst . with (.processors) (with group (First . field)) +processorDescription :: + (Default b, HasField "description" b (Maybe Text)) => + (Processors -> Maybe a) -> + (a -> Maybe b) -> + Config -> + Maybe Text +processorDescription group field = + (.description) . fromMaybe def . getFirst . with (.processors) (with group (First . field)) {- | Get the name corresponding to a processor. -} -processorName :: (Default b, HasField "name" b Text) => (Processors -> Maybe a) -> (a -> Maybe b) -> Config -> Text -processorName group field = (.name) . fromMaybe def . getFirst . with (.processors) (with group (First . field)) - -------------------------------------------------------------------------------- --- Internal Helpers -------------------------------------------------------------------------------- +processorName :: + (Default b, HasField "name" b Text) => + (Processors -> Maybe a) -> + (a -> Maybe b) -> + Config -> + Text +processorName group field = + (.name) . fromMaybe def . getFirst . with (.processors) (with group (First . field)) {- | -Internal helper. -The encoding options that should be used by every `FromJSON` and `ToJSON` instance for the configuration. +Get the aggregation strategy corresponding to a metric processor. -} -encodingOptions :: Options -encodingOptions = - defaultOptions - { fieldLabelModifier = camelTo2 '_' - , constructorTagModifier = camelTo2 '_' - , allNullaryToStringTag = True - , omitNothingFields = False - , sumEncoding = UntaggedValue - , tagSingleConstructors = False - , unwrapUnaryRecords = False - } +processorAggregationStrategy :: + (Default b, HasField "aggregate" b (Maybe AggregationStrategy)) => + (Processors -> Maybe a) -> + (a -> Maybe b) -> + Config -> + Maybe AggregationStrategy +processorAggregationStrategy group field = + (.aggregate) . fromMaybe def . getFirst . with (.processors) (with group (First . field)) {- | Internal helper. diff --git a/src/GHC/Eventlog/Live/Otelcol/Config/Default.hs b/src/GHC/Eventlog/Live/Otelcol/Config/Default.hs new file mode 100644 --- /dev/null +++ b/src/GHC/Eventlog/Live/Otelcol/Config/Default.hs @@ -0,0 +1,97 @@ +{-# LANGUAGE AllowAmbiguousTypes #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE UndecidableInstances #-} +{-# OPTIONS_GHC -Wno-orphans #-} + +{- | +Module : GHC.Eventlog.Live.Otelcol.Config +Description : The implementation of @eventlog-live-otelcol@. +Stability : experimental +Portability : portable +-} +module GHC.Eventlog.Live.Otelcol.Config.Default ( + defaultConfig, + + -- * Internal helpers for defining `Default` instances + MissingDefaultException (..), + getDefault, +) where + +import Control.Exception (Exception (..), throw) +import Data.Bifunctor (Bifunctor (..)) +import Data.Kind (Type) +import Data.List (intercalate) +import Data.Proxy (Proxy (..)) +import Data.Yaml qualified as Y +import GHC.Eventlog.Live.Otelcol.Config.Default.Raw (defaultConfigByteString) +import GHC.Eventlog.Live.Otelcol.Config.Types (Config) +import GHC.Records (HasField (..)) +import GHC.TypeLits (KnownSymbol, Symbol, symbolVal) +import Language.Haskell.TH.Lift.Compat (Exp, Lift (..), Q) + +{- | +Internal helper. +The default configuration. +-} +defaultConfig :: Config +defaultConfig = $(lift =<< Y.decodeThrow @Q @Config defaultConfigByteString) + +{- | +__Warning:__ Compile-time only. + +Internal helper. +This exception is thrown when a property is missing from the default configuration file. +-} +newtype MissingDefaultException + = MissingDefaultException + { accessors :: [String] + } + +instance Show MissingDefaultException where + show :: MissingDefaultException -> String + show e = + "Missing property '" <> intercalate "." e.accessors <> "' in default configuration" + +instance Exception MissingDefaultException + +{- | +__Warning:__ Compile-time only. + +Internal helper. +Get the default value under the given path of accessors and `lift` it. +-} +getDefault :: forall xs a b. (GetDefault xs a b, Lift b) => a -> Q Exp +getDefault a = either throw lift (getDefaultEither @xs @a @b a) + +{- | +__Warning:__ Compile-time only. + +Internal helper. +Get the default value under the given accessor after calling a recursor. +This function is used to implement the `GetDefault` instances. +-} +getDefaultEither' :: forall x a b c. (KnownSymbol x, HasField x a (Maybe b)) => (b -> Either MissingDefaultException c) -> a -> Either MissingDefaultException c +getDefaultEither' rec a = + first addAccessor . maybe (Left $ MissingDefaultException []) rec $ getField @x a + where + addAccessor :: MissingDefaultException -> MissingDefaultException + addAccessor e = MissingDefaultException{accessors = symbolVal (Proxy @x) : e.accessors} + +{- | +__Warning:__ Compile-time only. + +Internal helper. +This class guides the search for `getDefaultEither` functions from a list of accessors. +-} +class GetDefault (xs :: [Symbol]) (a :: Type) (b :: Type) where + getDefaultEither :: a -> Either MissingDefaultException b + +instance (KnownSymbol x, HasField x a (Maybe b)) => GetDefault (x ': '[]) a b where + getDefaultEither :: a -> Either MissingDefaultException b + getDefaultEither = getDefaultEither' @x Right + {-# INLINE getDefaultEither #-} + +instance (KnownSymbol x, HasField x a (Maybe b), GetDefault (y ': ys) b c) => GetDefault (x ': y ': ys) a c where + getDefaultEither :: a -> Either MissingDefaultException c + getDefaultEither = getDefaultEither' @x (getDefaultEither @(y ': ys)) + {-# INLINE getDefaultEither #-} diff --git a/src/GHC/Eventlog/Live/Otelcol/Config/Default/Raw.hs b/src/GHC/Eventlog/Live/Otelcol/Config/Default/Raw.hs new file mode 100644 --- /dev/null +++ b/src/GHC/Eventlog/Live/Otelcol/Config/Default/Raw.hs @@ -0,0 +1,40 @@ +{-# LANGUAGE TemplateHaskell #-} + +{- | +Module : GHC.Eventlog.Live.Otelcol.Config.Default.Raw +Description : The implementation of @eventlog-live-otelcol@. +Stability : experimental +Portability : portable +-} +module GHC.Eventlog.Live.Otelcol.Config.Default.Raw ( + defaultConfigByteString, + defaultConfigString, + defaultConfigText, +) where + +import Data.ByteString (ByteString) +import Data.FileEmbed (embedFileRelative) +import Data.Text (Text) +import Data.Text qualified as T +import Data.Text.Encoding qualified as TE (decodeUtf8Lenient) + +{- | +Internal helper. +The default configuration as a `ByteString`. +-} +defaultConfigByteString :: ByteString +defaultConfigByteString = $(embedFileRelative "data/default.yaml") + +{- | +Internal helper. +The default configuration as a `String`. +-} +defaultConfigString :: String +defaultConfigString = T.unpack defaultConfigText + +{- | +Internal helper. +The default configuration as a `Text`. +-} +defaultConfigText :: Text +defaultConfigText = TE.decodeUtf8Lenient defaultConfigByteString diff --git a/src/GHC/Eventlog/Live/Otelcol/Config/Types.hs b/src/GHC/Eventlog/Live/Otelcol/Config/Types.hs new file mode 100644 --- /dev/null +++ b/src/GHC/Eventlog/Live/Otelcol/Config/Types.hs @@ -0,0 +1,391 @@ +{- | +Module : GHC.Eventlog.Live.Otelcol.Config.Types +Description : The implementation of @eventlog-live-otelcol@. +Stability : experimental +Portability : portable +-} +module GHC.Eventlog.Live.Otelcol.Config.Types ( + Config (..), + Processors (..), + Metrics (..), + Spans (..), + AggregationStrategy (..), + HeapAllocatedMetric (..), + BlocksSizeMetric (..), + HeapSizeMetric (..), + HeapLiveMetric (..), + MemCurrentMetric (..), + MemNeededMetric (..), + MemReturnedMetric (..), + HeapProfSampleMetric (..), + CapabilityUsageMetric (..), + CapabilityUsageSpan (..), + ThreadStateSpan (..), +) where + +import Data.Aeson.Types (Encoding, FromJSON (..), Options (..), Parser, SumEncoding (..), ToJSON (..), Value (..), camelTo2, defaultOptions, genericParseJSON, genericToEncoding, genericToJSON) +import Data.Text (Text) +import GHC.Generics (Generic) +import Language.Haskell.TH.Lift.Compat (Lift) + +{- | +The configuration for @eventlog-live-otelcol@. +-} +newtype Config = Config + { processors :: Maybe Processors + } + deriving (Generic, Lift) + +instance FromJSON Config where + parseJSON :: Value -> Parser Config + parseJSON = genericParseJSON encodingOptions + +instance ToJSON Config where + toJSON :: Config -> Value + toJSON = genericToJSON encodingOptions + toEncoding :: Config -> Encoding + toEncoding = genericToEncoding encodingOptions + +{- | +The configuration options for the processors. +-} +data Processors = Processors + { metrics :: Maybe Metrics + , spans :: Maybe Spans + } + deriving (Generic, Lift) + +instance FromJSON Processors where + parseJSON :: Value -> Parser Processors + parseJSON = genericParseJSON encodingOptions + +instance ToJSON Processors where + toJSON :: Processors -> Value + toJSON = genericToJSON encodingOptions + toEncoding :: Processors -> Encoding + toEncoding = genericToEncoding encodingOptions + +{- | +The configuration options for the metric processors. +-} +data Metrics = Metrics + { heapAllocated :: Maybe HeapAllocatedMetric + , blocksSize :: Maybe BlocksSizeMetric + , heapSize :: Maybe HeapSizeMetric + , heapLive :: Maybe HeapLiveMetric + , memCurrent :: Maybe MemCurrentMetric + , memNeeded :: Maybe MemNeededMetric + , memReturned :: Maybe MemReturnedMetric + , heapProfSample :: Maybe HeapProfSampleMetric + , capabilityUsage :: Maybe CapabilityUsageMetric + } + deriving (Generic, Lift) + +instance FromJSON Metrics where + parseJSON :: Value -> Parser Metrics + parseJSON = genericParseJSON encodingOptions + +instance ToJSON Metrics where + toJSON :: Metrics -> Value + toJSON = genericToJSON encodingOptions + toEncoding :: Metrics -> Encoding + toEncoding = genericToEncoding encodingOptions + +{- | +The configuration options for the span processors. +-} +data Spans = Spans + { capabilityUsage :: Maybe CapabilityUsageSpan + , threadState :: Maybe ThreadStateSpan + } + deriving (Generic, Lift) + +instance FromJSON Spans where + parseJSON :: Value -> Parser Spans + parseJSON = genericParseJSON encodingOptions + +instance ToJSON Spans where + toJSON :: Spans -> Value + toJSON = genericToJSON encodingOptions + toEncoding :: Spans -> Encoding + toEncoding = genericToEncoding encodingOptions + +{- | +The options for metric aggregation. +-} +data AggregationStrategy + = AggregationStrategyByBatch + deriving (Generic, Lift) + +instance FromJSON AggregationStrategy where + parseJSON :: Value -> Parser AggregationStrategy + parseJSON = genericParseJSON aggregationStrategyEncodingOptions + +instance ToJSON AggregationStrategy where + toJSON :: AggregationStrategy -> Value + toJSON = genericToJSON aggregationStrategyEncodingOptions + toEncoding :: AggregationStrategy -> Encoding + toEncoding = genericToEncoding aggregationStrategyEncodingOptions + +{- | +The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Heap.processHeapAllocatedData`. +-} +data HeapAllocatedMetric = HeapAllocatedMetric + { description :: Maybe Text + , enabled :: Bool + , name :: Text + , aggregate :: Maybe AggregationStrategy + } + deriving (Generic, Lift) + +instance FromJSON HeapAllocatedMetric where + parseJSON :: Value -> Parser HeapAllocatedMetric + parseJSON = genericParseJSON encodingOptions + +instance ToJSON HeapAllocatedMetric where + toJSON :: HeapAllocatedMetric -> Value + toJSON = genericToJSON encodingOptions + toEncoding :: HeapAllocatedMetric -> Encoding + toEncoding = genericToEncoding encodingOptions + +{- | +The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Heap.processHeapSizeData`. +-} +data HeapSizeMetric = HeapSizeMetric + { description :: Maybe Text + , enabled :: Bool + , name :: Text + , aggregate :: Maybe AggregationStrategy + } + deriving (Generic, Lift) + +instance FromJSON HeapSizeMetric where + parseJSON :: Value -> Parser HeapSizeMetric + parseJSON = genericParseJSON encodingOptions + +instance ToJSON HeapSizeMetric where + toJSON :: HeapSizeMetric -> Value + toJSON = genericToJSON encodingOptions + toEncoding :: HeapSizeMetric -> Encoding + toEncoding = genericToEncoding encodingOptions + +{- | +The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Heap.processBlocksSizeData`. +-} +data BlocksSizeMetric = BlocksSizeMetric + { description :: Maybe Text + , enabled :: Bool + , name :: Text + , aggregate :: Maybe AggregationStrategy + } + deriving (Generic, Lift) + +instance FromJSON BlocksSizeMetric where + parseJSON :: Value -> Parser BlocksSizeMetric + parseJSON = genericParseJSON encodingOptions + +instance ToJSON BlocksSizeMetric where + toJSON :: BlocksSizeMetric -> Value + toJSON = genericToJSON encodingOptions + toEncoding :: BlocksSizeMetric -> Encoding + toEncoding = genericToEncoding encodingOptions + +{- | +The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Heap.processHeapLiveData`. +-} +data HeapLiveMetric = HeapLiveMetric + { description :: Maybe Text + , enabled :: Bool + , name :: Text + , aggregate :: Maybe AggregationStrategy + } + deriving (Generic, Lift) + +instance FromJSON HeapLiveMetric where + parseJSON :: Value -> Parser HeapLiveMetric + parseJSON = genericParseJSON encodingOptions + +instance ToJSON HeapLiveMetric where + toJSON :: HeapLiveMetric -> Value + toJSON = genericToJSON encodingOptions + toEncoding :: HeapLiveMetric -> Encoding + toEncoding = genericToEncoding encodingOptions + +{- | +The configuration options for the @memCurrent@ field `GHC.Eventlog.Live.Machine.Analysis.Heap.processMemReturnData`. +-} +data MemCurrentMetric = MemCurrentMetric + { description :: Maybe Text + , enabled :: Bool + , name :: Text + , aggregate :: Maybe AggregationStrategy + } + deriving (Generic, Lift) + +instance FromJSON MemCurrentMetric where + parseJSON :: Value -> Parser MemCurrentMetric + parseJSON = genericParseJSON encodingOptions + +instance ToJSON MemCurrentMetric where + toJSON :: MemCurrentMetric -> Value + toJSON = genericToJSON encodingOptions + toEncoding :: MemCurrentMetric -> Encoding + toEncoding = genericToEncoding encodingOptions + +{- | +The configuration options for the @memNeeded@ field `GHC.Eventlog.Live.Machine.Analysis.Heap.processMemReturnData`. +-} +data MemNeededMetric = MemNeededMetric + { description :: Maybe Text + , enabled :: Bool + , name :: Text + , aggregate :: Maybe AggregationStrategy + } + deriving (Generic, Lift) + +instance FromJSON MemNeededMetric where + parseJSON :: Value -> Parser MemNeededMetric + parseJSON = genericParseJSON encodingOptions + +instance ToJSON MemNeededMetric where + toJSON :: MemNeededMetric -> Value + toJSON = genericToJSON encodingOptions + toEncoding :: MemNeededMetric -> Encoding + toEncoding = genericToEncoding encodingOptions + +{- | +The configuration options for the @memReturned@ field `GHC.Eventlog.Live.Machine.Analysis.Heap.processMemReturnData`. +-} +data MemReturnedMetric = MemReturnedMetric + { description :: Maybe Text + , enabled :: Bool + , name :: Text + , aggregate :: Maybe AggregationStrategy + } + deriving (Generic, Lift) + +instance FromJSON MemReturnedMetric where + parseJSON :: Value -> Parser MemReturnedMetric + parseJSON = genericParseJSON encodingOptions + +instance ToJSON MemReturnedMetric where + toJSON :: MemReturnedMetric -> Value + toJSON = genericToJSON encodingOptions + toEncoding :: MemReturnedMetric -> Encoding + toEncoding = genericToEncoding encodingOptions + +{- | +The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Heap.processHeapProfSampleData`. +-} +data HeapProfSampleMetric = HeapProfSampleMetric + { description :: Maybe Text + , enabled :: Bool + , name :: Text + , aggregate :: Maybe AggregationStrategy + } + deriving (Generic, Lift) + +instance FromJSON HeapProfSampleMetric where + parseJSON :: Value -> Parser HeapProfSampleMetric + parseJSON = genericParseJSON encodingOptions + +instance ToJSON HeapProfSampleMetric where + toJSON :: HeapProfSampleMetric -> Value + toJSON = genericToJSON encodingOptions + toEncoding :: HeapProfSampleMetric -> Encoding + toEncoding = genericToEncoding encodingOptions + +{- | +The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Capability.processCapabilityUsageMetrics`. +-} +data CapabilityUsageMetric = CapabilityUsageMetric + { description :: Maybe Text + , enabled :: Bool + , name :: Text + , aggregate :: Maybe AggregationStrategy + } + deriving (Generic, Lift) + +instance FromJSON CapabilityUsageMetric where + parseJSON :: Value -> Parser CapabilityUsageMetric + parseJSON = genericParseJSON encodingOptions + +instance ToJSON CapabilityUsageMetric where + toJSON :: CapabilityUsageMetric -> Value + toJSON = genericToJSON encodingOptions + toEncoding :: CapabilityUsageMetric -> Encoding + toEncoding = genericToEncoding encodingOptions + +{- | +The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Capability.processCapabilityUsageSpans`. +-} +data CapabilityUsageSpan = CapabilityUsageSpan + { description :: Maybe Text + , enabled :: Bool + , name :: Text + } + deriving (Generic, Lift) + +instance FromJSON CapabilityUsageSpan where + parseJSON :: Value -> Parser CapabilityUsageSpan + parseJSON = genericParseJSON encodingOptions + +instance ToJSON CapabilityUsageSpan where + toJSON :: CapabilityUsageSpan -> Value + toJSON = genericToJSON encodingOptions + toEncoding :: CapabilityUsageSpan -> Encoding + toEncoding = genericToEncoding encodingOptions + +{- | +The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Thread.processThreadStateSpans`. +-} +data ThreadStateSpan = ThreadStateSpan + { description :: Maybe Text + , enabled :: Bool + , name :: Text + } + deriving (Generic, Lift) + +instance FromJSON ThreadStateSpan where + parseJSON :: Value -> Parser ThreadStateSpan + parseJSON = genericParseJSON encodingOptions + +instance ToJSON ThreadStateSpan where + toJSON :: ThreadStateSpan -> Value + toJSON = genericToJSON encodingOptions + toEncoding :: ThreadStateSpan -> Encoding + toEncoding = genericToEncoding encodingOptions + +------------------------------------------------------------------------------- +-- Internal Helpers +------------------------------------------------------------------------------- + +{- | +Internal helper. +The encoding options that should be used by every `FromJSON` and `ToJSON` instance for the configuration. +-} +encodingOptions :: Options +encodingOptions = + defaultOptions + { fieldLabelModifier = camelTo2 '_' + , constructorTagModifier = camelTo2 '_' + , allNullaryToStringTag = True + , omitNothingFields = False + , sumEncoding = UntaggedValue + , tagSingleConstructors = False + , unwrapUnaryRecords = False + } + +{- | +Internal helper. +The encoding options that should be used by the `FromJSON` and `ToJSON` instances for `AggregationStrategy`. +-} +aggregationStrategyEncodingOptions :: Options +aggregationStrategyEncodingOptions = + encodingOptions + { constructorTagModifier = camelTo2 '_' . stripAggregationStrategy + , tagSingleConstructors = True + } + where + stripAggregationStrategy :: String -> String + stripAggregationStrategy = drop (length ("AggregationStrategy" :: String)) diff --git a/src/GHC/Eventlog/Live/Otelcol/Exporter.hs b/src/GHC/Eventlog/Live/Otelcol/Exporter.hs new file mode 100644 --- /dev/null +++ b/src/GHC/Eventlog/Live/Otelcol/Exporter.hs @@ -0,0 +1,276 @@ +{-# OPTIONS_GHC -Wno-orphans #-} + +module GHC.Eventlog.Live.Otelcol.Exporter ( + -- * Metrics + ExportMetricsResult (..), + RejectedMetricsError (..), + exportResourceMetrics, + + -- * Traces + ExportTraceResult (..), + RejectedSpansError (..), + exportResourceSpans, + + -- * Helpers + + -- ** Counting Metrics + countDataPointsInExportMetricsServiceRequest, + countDataPointsInResourceMetrics, + countDataPointsInScopeMetrics, + countDataPointsInMetric, + + -- ** Counting Spans + countSpansInExportTraceServiceRequest, + countSpansInResourceSpans, + countSpansInScopeSpans, +) where + +import Control.Exception (Exception (..), SomeException (..), catch) +import Control.Monad.IO.Class (MonadIO (..)) +import Data.Int (Int64) +import Data.Machine (ProcessT, await, repeatedly, yield) +import Data.Semigroup (Sum (..)) +import Data.Text (Text) +import Data.Vector qualified as V +import Lens.Family2 ((^.)) +import Network.GRPC.Client qualified as G +import Network.GRPC.Client.StreamType.IO qualified as G +import Network.GRPC.Common qualified as G +import Network.GRPC.Common.Protobuf (Protobuf) +import Network.GRPC.Common.Protobuf qualified as G +import Proto.Opentelemetry.Proto.Collector.Metrics.V1.MetricsService qualified as OMS +import Proto.Opentelemetry.Proto.Collector.Metrics.V1.MetricsService_Fields qualified as OMS +import Proto.Opentelemetry.Proto.Collector.Trace.V1.TraceService qualified as OTS +import Proto.Opentelemetry.Proto.Collector.Trace.V1.TraceService_Fields qualified as OTS +import Proto.Opentelemetry.Proto.Metrics.V1.Metrics qualified as OM +import Proto.Opentelemetry.Proto.Metrics.V1.Metrics_Fields qualified as OM +import Proto.Opentelemetry.Proto.Trace.V1.Trace qualified as OT +import Proto.Opentelemetry.Proto.Trace.V1.Trace_Fields qualified as OT +import Text.Printf (printf) + +-------------------------------------------------------------------------------- +-- OpenTelemetry gRPC Exporters +-------------------------------------------------------------------------------- + +-------------------------------------------------------------------------------- +-- OpenTelemetry Exporter Result for Metrics + +data ExportMetricsResult + = ExportMetricsResult + { exportedDataPoints :: !Int64 + , rejectedDataPoints :: !Int64 + , maybeSomeException :: Maybe SomeException + } + deriving (Show) + +pattern ExportMetricsSuccess :: Int64 -> ExportMetricsResult +pattern ExportMetricsSuccess exportedDataPoints = + ExportMetricsResult exportedDataPoints 0 Nothing + +pattern ExportMetricsError :: Int64 -> Int64 -> SomeException -> ExportMetricsResult +pattern ExportMetricsError exportedDataPoints rejectedDataPoints someException = + ExportMetricsResult exportedDataPoints rejectedDataPoints (Just someException) + +data RejectedMetricsError + = RejectedMetricsError + { rejectedDataPoints :: !Int64 + , errorMessage :: !Text + } + deriving (Show) + +instance Exception RejectedMetricsError where + displayException :: RejectedMetricsError -> String + displayException RejectedMetricsError{..} = + printf "Error: OpenTelemetry Collector rejected %d data points with message: %s" rejectedDataPoints errorMessage + +-------------------------------------------------------------------------------- +-- OpenTelemetry gRPC Exporter for Metrics + +exportResourceMetrics :: + G.Connection -> + ProcessT IO OMS.ExportMetricsServiceRequest ExportMetricsResult +exportResourceMetrics conn = + repeatedly $ + await >>= \exportMetricsServiceRequest -> + liftIO (sendResourceMetrics exportMetricsServiceRequest) + >>= yield + where + sendResourceMetrics :: OMS.ExportMetricsServiceRequest -> IO ExportMetricsResult + sendResourceMetrics exportMetricsServiceRequest = + doGrpc `catch` handleGrpcError + where + !totalDataPoints = countDataPointsInExportMetricsServiceRequest exportMetricsServiceRequest + + doGrpc :: IO ExportMetricsResult + doGrpc = do + G.nonStreaming conn (G.rpc @(Protobuf OMS.MetricsService "export")) (G.Proto exportMetricsServiceRequest) >>= \case + G.Proto resp + | resp ^. OMS.partialSuccess . OMS.rejectedDataPoints == 0 -> do + pure $ ExportMetricsSuccess totalDataPoints + | otherwise -> do + let !rejectedDataPoints = resp ^. OMS.partialSuccess . OMS.rejectedDataPoints + let !exportedDataPoints = totalDataPoints - rejectedDataPoints + let !rejectedMetricsError = RejectedMetricsError{errorMessage = resp ^. OMS.partialSuccess . OMS.errorMessage, ..} + pure $ ExportMetricsError exportedDataPoints rejectedDataPoints (SomeException rejectedMetricsError) + + handleGrpcError :: G.GrpcError -> IO ExportMetricsResult + handleGrpcError grpcError = pure $ ExportMetricsError 0 totalDataPoints (SomeException grpcError) + +type instance G.RequestMetadata (Protobuf OMS.MetricsService meth) = G.NoMetadata +type instance G.ResponseInitialMetadata (Protobuf OMS.MetricsService meth) = G.NoMetadata +type instance G.ResponseTrailingMetadata (Protobuf OMS.MetricsService meth) = G.NoMetadata + +-------------------------------------------------------------------------------- +-- OpenTelemetry Exporter Result for Traces + +data ExportTraceResult + = ExportTraceResult + { exportedSpans :: !Int64 + , rejectedSpans :: !Int64 + , maybeSomeException :: Maybe SomeException + } + deriving (Show) + +pattern ExportTraceSuccess :: Int64 -> ExportTraceResult +pattern ExportTraceSuccess exportedSpans = + ExportTraceResult exportedSpans 0 Nothing + +pattern ExportTraceError :: Int64 -> Int64 -> SomeException -> ExportTraceResult +pattern ExportTraceError exportedSpans rejectedSpans someException = + ExportTraceResult exportedSpans rejectedSpans (Just someException) + +data RejectedSpansError + = RejectedSpansError + { rejectedSpans :: !Int64 + , errorMessage :: !Text + } + deriving (Show) + +instance Exception RejectedSpansError where + displayException :: RejectedSpansError -> String + displayException RejectedSpansError{..} = + printf "Error: OpenTelemetry Collector rejectedSpans %d data points with message: %s" rejectedSpans errorMessage + +-------------------------------------------------------------------------------- +-- OpenTelemetry gRPC Exporter for Traces + +exportResourceSpans :: + G.Connection -> + ProcessT IO OTS.ExportTraceServiceRequest ExportTraceResult +exportResourceSpans conn = + repeatedly $ + await >>= \exportTraceServiceRequest -> + liftIO (sendResourceSpans exportTraceServiceRequest) + >>= yield + where + sendResourceSpans :: OTS.ExportTraceServiceRequest -> IO ExportTraceResult + sendResourceSpans exportTraceServiceRequest = + doGrpc `catch` handleGrpcError + where + !totalSpans = countSpansInExportTraceServiceRequest exportTraceServiceRequest + + doGrpc :: IO ExportTraceResult + doGrpc = do + G.nonStreaming conn (G.rpc @(Protobuf OTS.TraceService "export")) (G.Proto exportTraceServiceRequest) >>= \case + G.Proto resp + | resp ^. OTS.partialSuccess . OTS.rejectedSpans == 0 -> + pure $ ExportTraceSuccess totalSpans + | otherwise -> do + let !rejectedSpans = resp ^. OTS.partialSuccess . OTS.rejectedSpans + let !exportedSpans = totalSpans - rejectedSpans + let !rejectedMetricsError = RejectedSpansError{errorMessage = resp ^. OTS.partialSuccess . OTS.errorMessage, ..} + pure $ ExportTraceError exportedSpans rejectedSpans (SomeException rejectedMetricsError) + + handleGrpcError :: G.GrpcError -> IO ExportTraceResult + handleGrpcError grpcError = pure $ ExportTraceError 0 totalSpans (SomeException grpcError) + +type instance G.RequestMetadata (Protobuf OTS.TraceService meth) = G.NoMetadata +type instance G.ResponseInitialMetadata (Protobuf OTS.TraceService meth) = G.NoMetadata +type instance G.ResponseTrailingMetadata (Protobuf OTS.TraceService meth) = G.NoMetadata + +-------------------------------------------------------------------------------- +-- Internal Helpers +-------------------------------------------------------------------------------- + +{- | +Internal helper. +Count the number of `OM.NumberDataPoint` values in an `OMS.ExportMetricsServiceRequest`. +-} +{-# SPECIALIZE countDataPointsInExportMetricsServiceRequest :: OMS.ExportMetricsServiceRequest -> Int64 #-} +{-# SPECIALIZE countDataPointsInExportMetricsServiceRequest :: OMS.ExportMetricsServiceRequest -> Word #-} +countDataPointsInExportMetricsServiceRequest :: (Integral i) => OMS.ExportMetricsServiceRequest -> i +countDataPointsInExportMetricsServiceRequest exportMetricsServiceRequest = + getSum $ foldMap (Sum . countDataPointsInResourceMetrics) (exportMetricsServiceRequest ^. OMS.vec'resourceMetrics) + +{- | +Internal helper. +Count the number of `OM.NumberDataPoint` values in an `OM.ResourceMetrics`. +-} +{-# SPECIALIZE countDataPointsInResourceMetrics :: OM.ResourceMetrics -> Int64 #-} +{-# SPECIALIZE countDataPointsInResourceMetrics :: OM.ResourceMetrics -> Word #-} +countDataPointsInResourceMetrics :: (Integral i) => OM.ResourceMetrics -> i +countDataPointsInResourceMetrics resourceMetrics = + getSum $ foldMap (Sum . countDataPointsInScopeMetrics) (resourceMetrics ^. OM.vec'scopeMetrics) + +{- | +Internal helper. +Count the number of `OM.NumberDataPoint` values in an `OM.ScopeMetrics`. +-} +{-# SPECIALIZE countDataPointsInScopeMetrics :: OM.ScopeMetrics -> Int64 #-} +{-# SPECIALIZE countDataPointsInScopeMetrics :: OM.ScopeMetrics -> Word #-} +countDataPointsInScopeMetrics :: (Integral i) => OM.ScopeMetrics -> i +countDataPointsInScopeMetrics scopeMetrics = + getSum $ foldMap (Sum . countDataPointsInMetric) (scopeMetrics ^. OM.vec'metrics) + +{- | +Internal helper. +Count the number of `OM.NumberDataPoint` values in an `OM.Metric`. +-} +{-# SPECIALIZE countDataPointsInMetric :: OM.Metric -> Int64 #-} +{-# SPECIALIZE countDataPointsInMetric :: OM.Metric -> Word #-} +countDataPointsInMetric :: (Integral i) => OM.Metric -> i +countDataPointsInMetric metric = + fromIntegral $ + case metric ^. OM.maybe'data' of + Nothing -> 0 + Just (OM.Metric'Gauge gauge) -> + V.length (gauge ^. OM.vec'dataPoints) + Just (OM.Metric'Sum sum_) -> + V.length (sum_ ^. OM.vec'dataPoints) + Just (OM.Metric'Histogram histogram) -> + V.length (histogram ^. OM.vec'dataPoints) + Just (OM.Metric'ExponentialHistogram exponentialHistogram) -> + V.length (exponentialHistogram ^. OM.vec'dataPoints) + Just (OM.Metric'Summary summary) -> + V.length (summary ^. OM.vec'dataPoints) + +{- | +Internal helper. +Count the number of `OT.Span` values in an `OTS.ExportTraceServiceRequest`. +-} +{-# SPECIALIZE countSpansInExportTraceServiceRequest :: OTS.ExportTraceServiceRequest -> Int64 #-} +{-# SPECIALIZE countSpansInExportTraceServiceRequest :: OTS.ExportTraceServiceRequest -> Word #-} +countSpansInExportTraceServiceRequest :: (Integral i) => OTS.ExportTraceServiceRequest -> i +countSpansInExportTraceServiceRequest exportTraceServiceRequest = + getSum $ foldMap (Sum . countSpansInResourceSpans) (exportTraceServiceRequest ^. OTS.vec'resourceSpans) + +{- | +Internal helper. +Count the number of `OT.Span` values in an `OT.ResourceSpans`. +-} +{-# SPECIALIZE countSpansInResourceSpans :: OT.ResourceSpans -> Int64 #-} +{-# SPECIALIZE countSpansInResourceSpans :: OT.ResourceSpans -> Word #-} +countSpansInResourceSpans :: (Integral i) => OT.ResourceSpans -> i +countSpansInResourceSpans resourceSpans = + getSum $ foldMap (Sum . countSpansInScopeSpans) (resourceSpans ^. OT.vec'scopeSpans) + +{- | +Internal helper. +Count the number of `OT.Span` values in an `OT.ScopeSpans`. +-} +{-# SPECIALIZE countSpansInScopeSpans :: OT.ScopeSpans -> Int64 #-} +{-# SPECIALIZE countSpansInScopeSpans :: OT.ScopeSpans -> Word #-} +countSpansInScopeSpans :: (Integral i) => OT.ScopeSpans -> i +countSpansInScopeSpans scopeSpans = + fromIntegral $ + V.length (scopeSpans ^. OT.vec'spans) diff --git a/src/GHC/Eventlog/Live/Otelcol/Stats.hs b/src/GHC/Eventlog/Live/Otelcol/Stats.hs new file mode 100644 --- /dev/null +++ b/src/GHC/Eventlog/Live/Otelcol/Stats.hs @@ -0,0 +1,344 @@ +{-# LANGUAGE OverloadedStrings #-} + +{- | +Module : GHC.Eventlog.Live.Otelcol.Stats +Description : The implementation of @eventlog-live-otelcol@. +Stability : experimental +Portability : portable +-} +module GHC.Eventlog.Live.Otelcol.Stats ( + EventCount, + eventCountTick, + Stat (..), + processStats, +) where + +import Control.Exception (Exception (..)) +import Control.Monad (when) +import Control.Monad.IO.Class (MonadIO (..)) +import Data.Default (Default (..)) +import Data.Foldable (for_) +import Data.Int (Int64) +import Data.Machine (ProcessT, await, construct, mapping, repeatedly, (~>)) +import Data.Monoid (First (..)) +import Data.Text (Text) +import Data.Text qualified as T +import Data.Text.IO qualified as TIO +import Data.Void (Void) +import GHC.Eventlog.Live.Logger (logDebug, logError) +import GHC.Eventlog.Live.Machine.Core (Tick) +import GHC.Eventlog.Live.Machine.Core qualified as M +import GHC.Eventlog.Live.Otelcol.Exporter (ExportMetricsResult (..), ExportTraceResult (..)) +import GHC.Eventlog.Live.Verbosity (Verbosity) +import GHC.Records (HasField (..)) +import StrictList qualified as Strict +import System.Console.ANSI (hNowSupportsANSI) +import System.Console.ANSI qualified as ANSI +import System.IO qualified as IO +import Text.Layout.Table qualified as TBL + +{- | +This type represents a count of input events. +-} +newtype EventCount = EventCount {value :: Int64} + +{- | +Count the number of events seen between each tick. +-} +eventCountTick :: (Monad m) => ProcessT m (Tick a) EventCount +eventCountTick = M.batchByTickList ~> mapping (EventCount . fromIntegral . length) + +{- | +This type represents the various stats produced by the pipeline. +-} +data Stat + = EventCountStat !EventCount + | ExportMetricsResultStat !ExportMetricsResult + | ExportTraceResultStat !ExportTraceResult + +{- | +Internal helper. +This type represents the aggregate stats kept by the stats processor. +-} +data Stats = Stats + { eventCounts :: Row + , exportedDataPoints :: Row + , rejectedDataPoints :: Row + , exportedSpans :: Row + , rejectedSpans :: Row + , errors :: !(Strict.List Text) + , displayedLines :: !(First Int) + } + deriving (Show) + +{- | +Internal helper. +This type represents a single row of statistics. +-} +data Row = Row + { total :: !Int64 + , peakRatePerBatch :: !Int64 + , window :: !(Strict.List Int64) + } + deriving (Show) + +instance HasField "ratePerBatch" Row Int64 where + getField :: Row -> Int64 + getField row = ratePerBatch row.window + +{- | +Internal helper. +Computes the rate per batch from a window. +-} +ratePerBatch :: Strict.List Int64 -> Int64 +ratePerBatch window = + let !n = length window + in if n <= 0 then 0 else sum window `div` fromIntegral n + +{- | +Internal helper. +Create a singleton `Row`. +-} +singletonRow :: Int64 -> Row +singletonRow total = Row{..} + where + peakRatePerBatch = total + window = singletonStrictList total + +{- | +Internal helper. +This implements the left-biased union of rows. +In @unionRow new old@, the @new@ argument should contain the latest data. +-} +unionRow :: Int -> Row -> Row -> Row +unionRow windowSize new old = Row{..} + where + total = new.total + old.total + window = Strict.take windowSize (new.window <> old.window) + peakRatePerBatch = maximum [new.peakRatePerBatch, old.peakRatePerBatch, ratePerBatch window] + +{- | +Internal helper. +This instance implements the left-biased union of stats. +In @new <> old@, the @new@ argument should contain the latest data. +-} +unionStats :: Int -> Stats -> Stats -> Stats +unionStats windowSize new old = Stats{..} + where + eventCounts = unionRow windowSize new.eventCounts old.eventCounts + exportedDataPoints = unionRow windowSize new.exportedDataPoints old.exportedDataPoints + rejectedDataPoints = unionRow windowSize new.rejectedDataPoints old.rejectedDataPoints + exportedSpans = unionRow windowSize new.exportedSpans old.exportedSpans + rejectedSpans = unionRow windowSize new.rejectedSpans old.rejectedSpans + errors = Strict.take windowSize (new.errors <> old.errors) + displayedLines = new.displayedLines <> old.displayedLines + +{- | +Internal helper. +Construct a `Stats` object from an `EventCount`. +-} +fromEventCount :: EventCount -> Stats +fromEventCount eventCount = def{eventCounts = singletonRow eventCount.value} + +{- | +Internal helper. +Construct a `Stats` object from an `ExportMetricsResult`. +-} +fromExportMetricsResult :: ExportMetricsResult -> Stats +fromExportMetricsResult exportMetricResult = + def + { exportedDataPoints = singletonRow exportMetricResult.exportedDataPoints + , rejectedDataPoints = singletonRow exportMetricResult.rejectedDataPoints + , errors = maybeToStrictList $ T.pack . displayException <$> exportMetricResult.maybeSomeException + } + +{- | +Internal helper. +Construct a `Stats` object from an `ExportTraceResult`. +-} +fromExportTraceResult :: ExportTraceResult -> Stats +fromExportTraceResult exportTraceResult = + def + { exportedSpans = singletonRow exportTraceResult.exportedSpans + , rejectedSpans = singletonRow exportTraceResult.rejectedSpans + , errors = maybeToStrictList $ T.pack . displayException <$> exportTraceResult.maybeSomeException + } + +{- | +Internal helper. +Construct a singleton `Strict.List`. +-} +singletonStrictList :: a -> Strict.List a +singletonStrictList = (`Strict.Cons` Strict.Nil) + +{- | +Internal helper. +Variant of `Data.Maybe.maybeToList` for `Strict.List`. +-} +maybeToStrictList :: Maybe a -> Strict.List a +maybeToStrictList = maybe Strict.Nil singletonStrictList + +{- | +Internal helper. +This instance implements the empty row. +-} +instance Default Row where + def :: Row + def = Row{..} + where + total = 0 + peakRatePerBatch = 0 + window = Strict.Nil + +{- | +Internal helper. +This implements the empty stats. +-} +instance Default Stats where + def :: Stats + def = Stats{..} + where + eventCounts = def + exportedDataPoints = def + rejectedDataPoints = def + exportedSpans = def + rejectedSpans = def + errors = mempty + displayedLines = First Nothing + +{- | +Process and display stats. + +__Warning:__ This machine prints to stdout and is intended to be the /only/ function printing to stdout. +-} +processStats :: + (MonadIO m) => + Verbosity -> + Bool -> + Int -> + Int -> + ProcessT m Stat Void +processStats verbosity stats batchIntervalMs windowSize + | stats = + -- If --stats is ENABLED, maintain and display `Stats`. + let go stats0 = + await >>= \stat -> do + let stats1 = updateStats windowSize stats0 stat + stats2 <- liftIO (displayStats verbosity batchIntervalMs stats1) + go stats2 + in construct $ go def + | otherwise = + -- If --stats is DISABLED, log all incoming `Stat` values. + repeatedly $ await >>= logStat verbosity + +{- | +Internal helper. +Update the current stats based on new input. +-} +updateStats :: Int -> Stats -> Stat -> Stats +updateStats windowSize old = \case + EventCountStat eventCount -> unionStats windowSize (fromEventCount eventCount) old + ExportMetricsResultStat exportMetricsResult -> unionStats windowSize (fromExportMetricsResult exportMetricsResult) old + ExportTraceResultStat exportTraceResult -> unionStats windowSize (fromExportTraceResult exportTraceResult) old + +{- | +Internal helper. +Log a statistic. +-} +logStat :: (MonadIO m) => Verbosity -> Stat -> m () +logStat verbosity = \case + EventCountStat eventCount -> + logDebug verbosity $ "received " <> showText eventCount.value <> " events" + ExportMetricsResultStat exportMetricsResult -> do + -- Log exported events. + logDebug verbosity $ "exported " <> showText exportMetricsResult.exportedDataPoints <> " metrics" + -- Log rejected events. + when (exportMetricsResult.rejectedDataPoints > 0) $ + logError verbosity $ + "rejected " <> showText exportMetricsResult.rejectedDataPoints <> " metrics" + -- Log exception. + for_ exportMetricsResult.maybeSomeException $ \someException -> do + logError verbosity . T.pack $ displayException someException + ExportTraceResultStat exportTraceResult -> do + -- Log exported events. + logDebug verbosity $ "exported " <> showText exportTraceResult.exportedSpans <> " metrics" + -- Log rejected events. + when (exportTraceResult.rejectedSpans > 0) $ + logError verbosity $ + "rejected " <> showText exportTraceResult.rejectedSpans <> " metrics" + -- Log exception. + for_ exportTraceResult.maybeSomeException $ \someException -> do + logError verbosity . T.pack $ displayException someException + +{- | +Internal helper. +Display the current stats. +This is intented to be the *only* function printing to the terminal. + +TODO: The stats printer should only overwrite the numbers. +-} +displayStats :: Verbosity -> Int -> Stats -> IO Stats +displayStats verbosity intervalMs stats = do + -- Check if `displayedLines` is empty... + case stats.displayedLines of + First Nothing -> + -- ...if so, this is the first time this function has been evaluated... + -- ...so we should perform the `warnIfStderrSupportsANSI` check... + warnIfStderrSupportsANSI verbosity + First (Just numberOfLines) -> do + -- ...if not, we should clear the previous lines of output... + ANSI.cursorUp numberOfLines + ANSI.clearFromCursorToScreenEnd + + -- Compute the moving average count of items _per second_, + -- by computing the adjusted average of counts over n batches. + let rate :: Row -> Text + rate row = + let !r = (row.ratePerBatch * 1_000) `div` fromIntegral intervalMs + in showText r <> "/s" + let peak :: Row -> Text + peak row = + let !r = row.peakRatePerBatch + in showText r <> "/s" + + let cSpec :: [TBL.ColSpec] + cSpec = [TBL.defColSpec, TBL.defColSpec, TBL.numCol, TBL.numCol, TBL.numCol] + let hSpec :: TBL.HeaderSpec TBL.LineStyle (Maybe Text) + hSpec = TBL.titlesH [Just "Item", Just "Action", Just "Total (item)", Just "Rate (item/s)", Just "Peak (item/batch)"] + let mkRow :: Maybe Text -> Maybe Text -> Row -> TBL.RowGroup (Maybe Text) + mkRow item result row = TBL.rowG [item, result, Just (showText row.total), Just (rate row), Just (peak row)] + let rSpec :: [TBL.RowGroup (Maybe Text)] + rSpec = + [ mkRow (Just "Event") (Just "Received") stats.eventCounts + , mkRow (Just "Metric") (Just "Exported") stats.exportedDataPoints + , mkRow Nothing (Just "Rejected") stats.rejectedDataPoints + , mkRow (Just "Span") (Just "Exported") stats.exportedSpans + , mkRow Nothing (Just "Rejected") stats.rejectedSpans + ] + let tSpec :: TBL.TableSpec TBL.LineStyle TBL.LineStyle String (Maybe Text) (Maybe Text) + tSpec = TBL.columnHeaderTableS cSpec TBL.unicodeS hSpec rSpec + let table = TBL.tableLinesB tSpec :: [Text] + for_ table TIO.putStrLn + pure stats{displayedLines = First (Just $ length table)} + +{- | +Check if `IO.stderr` supports ANSI codes. If it does, it is likely printed to +the same terminal as `IO.stdout`, which causes issues if @--stats@ is enabled. +-} +warnIfStderrSupportsANSI :: Verbosity -> IO () +warnIfStderrSupportsANSI verbosity = do + supportsANSI <- hNowSupportsANSI IO.stderr + when supportsANSI $ do + logError verbosity $ + "When statistics are enabled, stderr should be redirected to a file." + +------------------------------------------------------------------------------- +-- Internal Helpers +------------------------------------------------------------------------------- + +{- | +Internal helper. +Show a value as `Text`. +-} +showText :: (Show a) => a -> Text +showText = T.pack . show diff --git a/src/Language/Haskell/TH/Lift/Compat.hs b/src/Language/Haskell/TH/Lift/Compat.hs new file mode 100644 --- /dev/null +++ b/src/Language/Haskell/TH/Lift/Compat.hs @@ -0,0 +1,13 @@ +{-# LANGUAGE CPP #-} + +module Language.Haskell.TH.Lift.Compat ( + Exp, + Lift (..), + Q, +) where + +#if defined(EVENTLOG_LIVE_OTELCOL_USE_TEMPLATE_HASKELL_LIFT) +import Language.Haskell.TH.Lift (Exp, Lift (..), Q) +#else +import Language.Haskell.TH.Syntax (Exp, Lift (..), Q) +#endif