diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,8 @@ +### 0.3.0.0 + +- **BREAKING**: Remove `--otelcol-no-metrics` and `--otelcol-no-traces` flags. +- Support configuration file. + ### 0.2.0.0 - **BREAKING**: Support info and debug verbosity. This changes the semantics of `--verbosity=3` and up. diff --git a/eventlog-live-otelcol.cabal b/eventlog-live-otelcol.cabal --- a/eventlog-live-otelcol.cabal +++ b/eventlog-live-otelcol.cabal @@ -1,6 +1,6 @@ cabal-version: 3.0 name: eventlog-live-otelcol -version: 0.2.0.0 +version: 0.3.0.0 synopsis: Stream eventlog data to the OpenTelemetry Collector. description: This executable supports live streaming of eventlog data into @@ -13,12 +13,12 @@ > [--batch-interval NUM] [--eventlog-log-file FILE] > [-h Tcmdyrbi] [--service-name STRING] > [-v|--verbosity quiet|error|warning|info|debug|0-4] - > --otelcol-host HOST [--otelcol-port PORT] - > [--otelcol-authority HOST] [--otelcol-ssl] - > [--otelcol-certificate-store FILE] + > [--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] - > [--otelcol-no-metrics] [--otelcol-no-traces] + > [--print-defaults] > > Available options: > --eventlog-stdin Read the eventlog from stdin. @@ -35,6 +35,9 @@ > --service-name STRING The name of the profiled service. > -v,--verbosity quiet|error|warning|info|debug|0-4 > The verbosity threshold for logging. + > --config FILE The path to a detailed configuration file. + > --print-defaults Print default configuration options that can be used + > in config.yaml > --help Show this help text. > --version Show version information > @@ -49,8 +52,6 @@ > Use file to log SSL keys. > --otelcol-ssl-key-log-from-env > Use SSLKEYLOGFILE to log SSL keys. - > --otelcol-no-metrics Disable metrics exporter. - > --otelcol-no-traces Disable traces exporter. license: BSD-3-Clause license-file: LICENSE @@ -80,6 +81,7 @@ DefaultSignatures DeriveFoldable DeriveFunctor + DeriveGeneric DeriveTraversable DerivingStrategies DuplicateRecordFields @@ -105,20 +107,23 @@ library import: language hs-source-dirs: src - autogen-modules: Paths_eventlog_live_otelcol + exposed-modules: + GHC.Eventlog.Live.Otelcol + GHC.Eventlog.Live.Otelcol.Config + other-modules: Options.Applicative.Compat Paths_eventlog_live_otelcol System.Random.Compat - exposed-modules: GHC.Eventlog.Live.Otelcol + autogen-modules: Paths_eventlog_live_otelcol build-depends: + , aeson >=2.2 && <2.3 , base >=4.16 && <4.22 , bytestring >=0.11 && <0.13 + , data-default >=0.8 && <0.9 , dlist >=1.0 && <1.1 - , eventlog-live >=0.2 && <0.3 - , eventlog-live:options - , eventlog-live:socket + , eventlog-live >=0.3 && <0.4 , ghc-events >=0.20 && <0.21 , grapesy >=1.0.0 && <1.2 , hashable >=1.4 && <1.6 @@ -131,6 +136,7 @@ , text >=1.2 && <2.2 , unordered-containers >=0.2.20 && <0.3 , vector >=0.11 && <0.14 + , yaml >=0.11 && <0.12 executable eventlog-live-otelcol import: language 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 @@ -18,34 +18,44 @@ import Data.ByteString (ByteString) 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.Functor ((<&>)) import Data.HashMap.Strict qualified as M import Data.Hashable (Hashable) import Data.Int (Int64) -import Data.Machine (Process, ProcessT, asParts, await, construct, filtered, mapping, repeatedly, yield, (~>)) +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.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.Void (Void) import Data.Word (Word32, Word64) +import Data.Yaml qualified as Y import GHC.Eventlog.Live.Data.Attribute import GHC.Eventlog.Live.Data.Metric -import GHC.Eventlog.Live.Machine (CapabilityUsageSpan, MemReturnData (..), ThreadStateSpan (..)) -import GHC.Eventlog.Live.Machine qualified as M +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 +import GHC.Eventlog.Live.Machine.Analysis.Heap (MemReturnData (..)) +import GHC.Eventlog.Live.Machine.Analysis.Heap qualified as M +import GHC.Eventlog.Live.Machine.Analysis.Thread (ThreadStateSpan (..)) +import GHC.Eventlog.Live.Machine.Analysis.Thread qualified as M import GHC.Eventlog.Live.Machine.Core (Tick) import GHC.Eventlog.Live.Machine.Core qualified as M import GHC.Eventlog.Live.Machine.WithStartTime (WithStartTime (..)) import GHC.Eventlog.Live.Machine.WithStartTime qualified as M 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.Socket (runWithEventlogSource) import GHC.Eventlog.Live.Verbosity (Verbosity) import GHC.RTS.Events (Event (..), HeapProfBreakdown (..), ThreadId) +import GHC.Records (HasField) import Lens.Family2 ((&), (.~), (^.)) import Network.GRPC.Client qualified as G import Network.GRPC.Client.StreamType.IO qualified as G @@ -66,7 +76,6 @@ 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.IO qualified as IO import System.Random (StdGen, initStdGen) import System.Random.Compat (uniformByteString) import Text.Printf (printf) @@ -78,7 +87,14 @@ main = do Options{..} <- O.execParser options let OpenTelemetryCollectorOptions{..} = openTelemetryCollectorOptions - let OpenTelemetryExporterOptions{..} = openTelemetryExporterOptions + -- 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 @@ -94,43 +110,43 @@ , M.counterByTick verbosity "events" , M.liftTick M.withStartTime ~> fanout - [ processHeapEvents verbosity maybeHeapProfBreakdown + [ processHeapEvents config verbosity maybeHeapProfBreakdown ~> mapping (fmap Left) - , processThreadEvents verbosity + , processThreadEvents config verbosity ] ~> mapping (partitionEithers . D.toList) ~> fanout - [ filtered (const exportMetrics) - ~> mapping fst - ~> fanout - [ M.counterBy verbosity "metrics" (sum . fmap countMetricDataPoints) - , asScopeMetrics - [ OM.scope .~ eventlogLiveScope - ] - ~> asResourceMetric [] - ~> asExportMetricServiceRequest - ~> exportResourceMetrics conn - ~> mapping (either displayException displayException) - ~> errorWriter - ] - , filtered (const exportTraces) - ~> mapping snd - ~> fanout - [ M.counterBy verbosity "spans" (fromIntegral . length) - , asScopeSpans - [ OT.scope .~ eventlogLiveScope - ] - ~> asResourceSpan - [ OT.resource - .~ messageWith - [ OM.attributes .~ mapMaybe toMaybeKeyValue [attrServiceName] - ] + [ runIf (shouldExportMetrics config) $ + mapping fst + ~> fanout + [ M.counterBy verbosity "metrics" (sum . fmap countMetricDataPoints) + , asScopeMetrics + [ OM.scope .~ eventlogLiveScope ] - ~> asExportTraceServiceRequest - ~> exportResourceSpans conn - ~> mapping (either displayException displayException) - ~> errorWriter - ] + ~> 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 + ] ] ] @@ -150,11 +166,32 @@ Just (OM.Metric'Summary summary) -> V.length (summary ^. OM.vec'dataPoints) -errorWriter :: (MonadIO m) => ProcessT m String Void -errorWriter = repeatedly $ await >>= liftIO . IO.hPutStrLn IO.stderr +{- | +Internal helper. +Determine whether or not any spans should be exported. +-} +shouldExportMetrics :: Config -> Bool +shouldExportMetrics config = + or + [ C.processorEnabled (.metrics) (.heapAllocated) config + , C.processorEnabled (.metrics) (.blocksSize) config + , C.processorEnabled (.metrics) (.heapSize) config + , C.processorEnabled (.metrics) (.heapLive) config + , C.processorEnabled (.metrics) (.memCurrent) config + , C.processorEnabled (.metrics) (.memNeeded) config + , C.processorEnabled (.metrics) (.memReturned) config + , C.processorEnabled (.metrics) (.heapProfSample) config + , C.processorEnabled (.metrics) (.capabilityUsage) config + ] --- dumpBatch :: (MonadIO m, Show a) => ProcessT m [a] [a] --- dumpBatch = traversing (\as -> for_ as (liftIO . print) >> pure as) +{- | +Internal helper. +Determine whether or not any spans should be exported. +-} +shouldExportSpans :: Config -> Bool +shouldExportSpans config = + C.processorEnabled (.spans) (.capabilityUsage) config + || C.processorEnabled (.spans) (.threadState) config -------------------------------------------------------------------------------- -- processThreadEvents @@ -164,63 +201,66 @@ processThreadEvents :: (MonadIO m) => + Config -> Verbosity -> ProcessT m (Tick (WithStartTime Event)) (DList (Either OM.Metric OT.Span)) -processThreadEvents verbosity = - M.sortByBatchTick (.value.evTime) - ~> M.liftTick - ( fanout - [ M.validateOrder verbosity (.value.evTime) - , M.processGCSpans verbosity - ~> mapping (D.singleton . A) - , M.processThreadStateSpans' M.tryGetTimeUnixNano (.value) M.setWithStartTime'value verbosity - ~> fanout - [ M.asMutatorSpans' (.value) M.setWithStartTime'value - ~> mapping (D.singleton . B) - , mapping (D.singleton . C) - ] - ] - ) - ~> M.liftTick - ( asParts - ~> mapping repackCapabilityUsageSpanOrThreadStateSpan - ) - ~> fanout - [ M.liftTick - ( mapping leftToMaybe - ~> asParts - ) - ~> fanout - [ M.liftTick - ( M.processCapabilityUsageMetrics - ~> asNumberDataPoint - ) - ~> M.batchByTickList - ~> asSum - [ OM.aggregationTemporality .~ OM.AGGREGATION_TEMPORALITY_DELTA - , OM.isMonotonic .~ True - ] - ~> asMetric - [ OM.name .~ "CapabilityUsage" - , OM.description .~ "Report the duration of each capability usage span." - , OM.unit .~ "ns" - ] - ~> mapping (D.singleton . Left) - , M.liftTick - ( M.dropStartTime - ~> asSpan - ~> mapping (D.singleton . Right) - ) - ~> M.batchByTick +processThreadEvents config verbosity = + runIf (shouldProcessThreadEvents config) $ + M.sortByBatchTick (.value.evTime) + ~> M.liftTick + ( fanout + [ M.validateOrder verbosity (.value.evTime) + , runIf (shouldComputeCapabilityUsageSpan config) $ + M.processGCSpans verbosity + ~> mapping (D.singleton . A) + , runIf (shouldComputeThreadStateSpan config) $ + M.processThreadStateSpans' M.tryGetTimeUnixNano (.value) M.setWithStartTime'value verbosity + ~> fanout + [ M.asMutatorSpans' (.value) M.setWithStartTime'value + ~> mapping (D.singleton . B) + , mapping (D.singleton . C) + ] ] - , M.liftTick - ( mapping rightToMaybe - ~> asParts - ~> asSpan - ~> mapping (D.singleton . Right) - ) - ~> M.batchByTick - ] + ) + ~> M.liftTick + ( asParts + ~> mapping repackCapabilityUsageSpanOrThreadStateSpan + ) + ~> fanout + [ M.liftTick + ( mapping leftToMaybe + ~> asParts + ) + ~> fanout + [ runIf (C.processorEnabled (.metrics) (.capabilityUsage) config) $ + M.liftTick + ( M.processCapabilityUsageMetrics + ~> asNumberDataPoint + ) + ~> M.batchByTickList + ~> asSum + [ OM.aggregationTemporality .~ OM.AGGREGATION_TEMPORALITY_DELTA + , OM.isMonotonic .~ True + ] + ~> asMetricWith config (.capabilityUsage) [OM.unit .~ "ns"] + ~> mapping (D.singleton . Left) + , runIf (C.processorEnabled (.spans) (.capabilityUsage) config) $ + M.liftTick + ( M.dropStartTime + ~> asSpan config + ~> mapping (D.singleton . Right) + ) + ~> M.batchByTick + ] + , runIf (C.processorEnabled (.spans) (.threadState) config) $ + M.liftTick + ( mapping rightToMaybe + ~> asParts + ~> asSpan config + ~> mapping (D.singleton . Right) + ) + ~> M.batchByTick + ] where repackCapabilityUsageSpanOrThreadStateSpan = \case A i -> Left $ fmap Left i @@ -241,122 +281,141 @@ rightToMaybe :: Either a b -> Maybe b rightToMaybe = either (const Nothing) Just +{- | +Internal helper. +Determine whether or not any thread events should be processed at all. +-} +shouldProcessThreadEvents :: Config -> Bool +shouldProcessThreadEvents config = + C.processorEnabled (.metrics) (.capabilityUsage) config + || C.processorEnabled (.spans) (.capabilityUsage) config + || C.processorEnabled (.spans) (.threadState) config + +{- | +Internal helper. +Determine whether or not the capability usage spans should be computed. +-} +shouldComputeCapabilityUsageSpan :: Config -> Bool +shouldComputeCapabilityUsageSpan config = + C.processorEnabled (.spans) (.capabilityUsage) config + || C.processorEnabled (.metrics) (.capabilityUsage) config + +{- | +Internal helper. +Determine whether or not the thread state spans should be computed. +-} +shouldComputeThreadStateSpan :: Config -> Bool +shouldComputeThreadStateSpan config = + C.processorEnabled (.spans) (.threadState) config + || shouldComputeCapabilityUsageSpan config + -------------------------------------------------------------------------------- -- processHeapEvents -------------------------------------------------------------------------------- processHeapEvents :: (MonadIO m) => + Config -> Verbosity -> Maybe HeapProfBreakdown -> ProcessT m (Tick (WithStartTime Event)) (DList OM.Metric) -processHeapEvents verbosity maybeHeapProfBreakdown = +processHeapEvents config verbosity maybeHeapProfBreakdown = fanout - [ processHeapAllocated - , processBlocksSize - , processHeapSize - , processHeapLive - , processMemReturn - , processHeapProfSample verbosity maybeHeapProfBreakdown + [ processHeapAllocated config + , processBlocksSize config + , processHeapSize config + , processHeapLive config + , processMemReturn config + , processHeapProfSample config verbosity maybeHeapProfBreakdown ] -------------------------------------------------------------------------------- -- HeapAllocated -processHeapAllocated :: Process (Tick (WithStartTime Event)) (DList OM.Metric) -processHeapAllocated = - M.liftTick (M.processHeapAllocatedData ~> asNumberDataPoint) - ~> M.batchByTickList - ~> asSum - [ OM.aggregationTemporality .~ OM.AGGREGATION_TEMPORALITY_DELTA - , OM.isMonotonic .~ True - ] - ~> asMetric - [ OM.name .~ "HeapAllocated" - , OM.description .~ "Report when a new chunk of heap has been allocated by the indicated capability set." - , OM.unit .~ "By" - ] - ~> mapping D.singleton +processHeapAllocated :: Config -> Process (Tick (WithStartTime Event)) (DList OM.Metric) +processHeapAllocated config = + runIf (C.processorEnabled (.metrics) (.heapAllocated) config) $ + M.liftTick (M.processHeapAllocatedData ~> asNumberDataPoint) + ~> M.batchByTickList + ~> asSum + [ OM.aggregationTemporality .~ OM.AGGREGATION_TEMPORALITY_DELTA + , OM.isMonotonic .~ True + ] + ~> asMetricWith config (.heapAllocated) [OM.unit .~ "By"] + ~> mapping D.singleton -------------------------------------------------------------------------------- -- HeapSize -processHeapSize :: Process (Tick (WithStartTime Event)) (DList OM.Metric) -processHeapSize = - M.liftTick (M.processHeapSizeData ~> asNumberDataPoint) - ~> M.batchByTickList - ~> asGauge - ~> asMetric - [ OM.name .~ "HeapSize" - , OM.description .~ "Report the current heap size, calculated by the allocated number of megablocks." - , OM.unit .~ "By" - ] - ~> mapping D.singleton +processHeapSize :: Config -> Process (Tick (WithStartTime Event)) (DList OM.Metric) +processHeapSize config = + runIf (C.processorEnabled (.metrics) (.heapSize) config) $ + M.liftTick (M.processHeapSizeData ~> asNumberDataPoint) + ~> M.batchByTickList + ~> asGauge + ~> asMetricWith config (.heapSize) [OM.unit .~ "By"] + ~> mapping D.singleton -------------------------------------------------------------------------------- -- BlocksSize -processBlocksSize :: Process (Tick (WithStartTime Event)) (DList OM.Metric) -processBlocksSize = - M.liftTick (M.processBlocksSizeData ~> asNumberDataPoint) - ~> M.batchByTickList - ~> asGauge - ~> asMetric - [ OM.name .~ "BlocksSize" - , OM.description .~ "Report the current heap size, calculated by the allocated number of blocks." - , OM.unit .~ "By" - ] - ~> mapping D.singleton +processBlocksSize :: Config -> Process (Tick (WithStartTime Event)) (DList OM.Metric) +processBlocksSize config = + runIf (C.processorEnabled (.metrics) (.blocksSize) config) $ + M.liftTick (M.processBlocksSizeData ~> asNumberDataPoint) + ~> M.batchByTickList + ~> asGauge + ~> asMetricWith config (.blocksSize) [OM.unit .~ "By"] + ~> mapping D.singleton -------------------------------------------------------------------------------- -- HeapLive -processHeapLive :: Process (Tick (WithStartTime Event)) (DList OM.Metric) -processHeapLive = - M.liftTick (M.processHeapLiveData ~> asNumberDataPoint) - ~> M.batchByTickList - ~> asGauge - ~> asMetric - [ OM.name .~ "HeapLive" - , OM.description .~ "Report the current heap size, calculated by the allocated number of megablocks." - , OM.unit .~ "By" - ] - ~> mapping D.singleton +processHeapLive :: Config -> Process (Tick (WithStartTime Event)) (DList OM.Metric) +processHeapLive config = + runIf (C.processorEnabled (.metrics) (.heapLive) config) $ + M.liftTick (M.processHeapLiveData ~> asNumberDataPoint) + ~> M.batchByTickList + ~> asGauge + ~> asMetricWith config (.heapLive) [OM.unit .~ "By"] + ~> mapping D.singleton -------------------------------------------------------------------------------- -- MemReturn -processMemReturn :: Process (Tick (WithStartTime Event)) (DList OM.Metric) -processMemReturn = - M.liftTick M.processMemReturnData - ~> M.batchByTickList - ~> fanout - [ M.liftBatch (asMemCurrent ~> asNumberDataPoint) - ~> asGauge - ~> asMetric - [ OM.name .~ "MemCurrent" - , OM.description .~ "Report the number of megablocks currently allocated." - , OM.unit .~ "{mblock}" - ] - ~> mapping D.singleton - , M.liftBatch (asMemNeeded ~> asNumberDataPoint) - ~> asGauge - ~> asMetric - [ OM.name .~ "MemNeeded" - , OM.description .~ "Report the number of megablocks currently needed." - , OM.unit .~ "{mblock}" - ] - ~> mapping D.singleton - , M.liftBatch (asMemReturned ~> asNumberDataPoint) - ~> asGauge - ~> asMetric - [ OM.name .~ "MemReturned" - , OM.description .~ "Report the number of megablocks currently being returned to the OS." - , OM.unit .~ "{mblock}" - ] - ~> mapping D.singleton - ] +processMemReturn :: Config -> Process (Tick (WithStartTime Event)) (DList OM.Metric) +processMemReturn config = + runIf (shouldComputeMemReturn config) $ + M.liftTick M.processMemReturnData + ~> M.batchByTickList + ~> fanout + [ runIf (C.processorEnabled (.metrics) (.memCurrent) config) $ + M.liftBatch (asMemCurrent ~> asNumberDataPoint) + ~> asGauge + ~> asMetricWith config (.memCurrent) [OM.unit .~ "{mblock}"] + ~> mapping D.singleton + , runIf (C.processorEnabled (.metrics) (.memNeeded) config) $ + M.liftBatch (asMemNeeded ~> asNumberDataPoint) + ~> asGauge + ~> asMetricWith config (.memNeeded) [OM.unit .~ "{mblock}"] + ~> mapping D.singleton + , runIf (C.processorEnabled (.metrics) (.memReturned) config) $ + M.liftBatch (asMemReturned ~> asNumberDataPoint) + ~> asGauge + ~> asMetricWith config (.memReturned) [OM.unit .~ "{mblock}"] + ~> mapping D.singleton + ] +{- | +Internal helper. +Determine whether the MemReturn data should be computed. +-} +shouldComputeMemReturn :: Config -> Bool +shouldComputeMemReturn config = + C.processorEnabled (.metrics) (.memCurrent) config + || C.processorEnabled (.metrics) (.memNeeded) config + || C.processorEnabled (.metrics) (.memReturned) config + asMemCurrent :: Process (Metric MemReturnData) (Metric Word32) asMemCurrent = mapping (fmap (.current)) @@ -371,19 +430,17 @@ processHeapProfSample :: (MonadIO m) => + Config -> Verbosity -> Maybe HeapProfBreakdown -> ProcessT m (Tick (WithStartTime Event)) (DList OM.Metric) -processHeapProfSample verbosity maybeHeapProfBreakdown = - M.liftTick (M.processHeapProfSampleData verbosity maybeHeapProfBreakdown ~> asNumberDataPoint) - ~> M.batchByTickList - ~> asGauge - ~> asMetric - [ OM.name .~ "HeapProfSample" - , OM.description .~ "Report a heap profile sample." - , OM.unit .~ "By" - ] - ~> mapping D.singleton +processHeapProfSample config verbosity maybeHeapProfBreakdown = + runIf (C.processorEnabled (.metrics) (.heapProfSample) config) $ + M.liftTick (M.processHeapProfSampleData verbosity maybeHeapProfBreakdown ~> asNumberDataPoint) + ~> M.batchByTickList + ~> asGauge + ~> asMetricWith config (.heapProfSample) [OM.unit .~ "By"] + ~> mapping D.singleton -------------------------------------------------------------------------------- -- Machines @@ -436,6 +493,19 @@ asScopeMetrics mod = mapping $ \metrics -> messageWith ((OM.metrics .~ metrics) : mod) +asMetricWith :: + (Default a, HasField "description" a (Maybe Text), HasField "name" a Text) => + Config -> + (C.Metrics -> Maybe a) -> + [OM.Metric -> OM.Metric] -> + Process OM.Metric'Data OM.Metric +asMetricWith config field mod = + asMetric $ + [ OM.name .~ C.processorName (.metrics) field config + , maybe id (OM.description .~) $ C.processorDescription (.metrics) field config + ] + <> mod + asMetric :: [OM.Metric -> OM.Metric] -> Process OM.Metric'Data OM.Metric asMetric mod = mapping $ toMetric mod @@ -479,6 +549,8 @@ Key v toSpan :: + -- | The configuration. + Config -> -- | The input value. v -> -- | The trace ID. @@ -488,9 +560,9 @@ OT.Span -- | The `asSpan` machine processes values @v@ into OpenTelemetry spans `OT.Span`. - asSpan :: (MonadIO m) => ProcessT m v OT.Span - default asSpan :: (MonadIO m, Hashable (Key v)) => ProcessT m v OT.Span - asSpan = construct $ go (mempty, Nothing) + asSpan :: (MonadIO m) => Config -> ProcessT m v OT.Span + default asSpan :: (MonadIO m, Hashable (Key v)) => Config -> ProcessT m v OT.Span + asSpan config = construct $ go (mempty, Nothing) where -- go :: (HashMap (Key v) ByteString, Maybe StdGen) -> PlanT (Is v) OT.Span m Void go (traceIds, maybeGen) = do @@ -507,7 +579,7 @@ -- Ensure the next value has a span ID let (spanId, gen2) = uniformByteString 8 gen1 -- Yield a span - yield $ toSpan i traceId spanId + yield $ toSpan config i traceId spanId -- Continue go (traceIds', Just gen2) @@ -520,12 +592,12 @@ toKey :: CapabilityUsageSpan -> Int toKey = (.cap) - toSpan :: CapabilityUsageSpan -> ByteString -> ByteString -> OT.Span - toSpan i traceId spanId = + toSpan :: Config -> CapabilityUsageSpan -> ByteString -> ByteString -> OT.Span + toSpan config i traceId spanId = messageWith [ OT.traceId .~ traceId , OT.spanId .~ spanId - , OT.name .~ M.showCapabilityUserCategory user + , OT.name .~ C.processorName (.spans) (.capabilityUsage) config <> " " <> M.showCapabilityUserCategory user , OT.kind .~ OT.Span'SPAN_KIND_INTERNAL , OT.startTimeUnixNano .~ i.startTimeUnixNano , OT.endTimeUnixNano .~ i.endTimeUnixNano @@ -552,12 +624,12 @@ toKey :: ThreadStateSpan -> ThreadId toKey = (.thread) - toSpan :: ThreadStateSpan -> ByteString -> ByteString -> OT.Span - toSpan i traceId spanId = + toSpan :: Config -> ThreadStateSpan -> ByteString -> ByteString -> OT.Span + toSpan config i traceId spanId = messageWith [ OT.traceId .~ traceId , OT.spanId .~ spanId - , OT.name .~ M.showThreadStateCategory i.threadState + , OT.name .~ C.processorName (.spans) (.threadState) config <> " " <> M.showThreadStateCategory i.threadState , OT.kind .~ OT.Span'SPAN_KIND_INTERNAL , OT.startTimeUnixNano .~ i.startTimeUnixNano , OT.endTimeUnixNano .~ i.endTimeUnixNano @@ -728,6 +800,7 @@ options = O.info ( optionsParser + O.<**> defaultsPrinter O.<**> OE.helperWith (O.long "help" <> O.help "Show this help text.") O.<**> OC.simpleVersioner (showVersion EventlogLive.version) ) @@ -742,6 +815,7 @@ , maybeHeapProfBreakdown :: Maybe HeapProfBreakdown , maybeServiceName :: Maybe ServiceName , verbosity :: Verbosity + , maybeConfigFile :: Maybe FilePath , openTelemetryCollectorOptions :: OpenTelemetryCollectorOptions } @@ -756,9 +830,31 @@ <*> O.optional heapProfBreakdownParser <*> O.optional serviceNameParser <*> verbosityParser + <*> O.optional configFileParser <*> openTelemetryCollectorOptionsParser -------------------------------------------------------------------------------- +-- Configuration + +configFileParser :: O.Parser FilePath +configFileParser = + O.strOption + ( O.long "config" + <> O.metavar "FILE" + <> O.help "The path to a detailed configuration file." + ) + +defaultsPrinter :: O.Parser (a -> a) +defaultsPrinter = + O.infoOption defaults . 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 newtype ServiceName = ServiceName {serviceName :: Text} @@ -775,9 +871,8 @@ -------------------------------------------------------------------------------- -- OpenTelemetry Collector configuration -data OpenTelemetryCollectorOptions = OpenTelemetryCollectorOptions +newtype OpenTelemetryCollectorOptions = OpenTelemetryCollectorOptions { openTelemetryCollectorServer :: G.Server - , openTelemetryExporterOptions :: OpenTelemetryExporterOptions } openTelemetryCollectorOptionsParser :: O.Parser OpenTelemetryCollectorOptions @@ -785,32 +880,7 @@ OC.parserOptionGroup "OpenTelemetry Collector Server Options" $ OpenTelemetryCollectorOptions <$> otelcolServerParser - <*> otelcolExporterOptionsParsers -data OpenTelemetryExporterOptions = OpenTelemetryExporterOptions - { exportMetrics :: Bool - , exportTraces :: Bool - } - -otelcolExporterOptionsParsers :: O.Parser OpenTelemetryExporterOptions -otelcolExporterOptionsParsers = - OpenTelemetryExporterOptions - <$> exportMetricsParser - <*> exportTracesParser - where - exportMetricsParser = - -- flipped from O.switch to match negated semantics - O.flag True False . mconcat $ - [ O.long "otelcol-no-metrics" - , O.help "Disable metrics exporter." - ] - exportTracesParser = - -- flipped from O.switch to match negated semantics - O.flag True False . mconcat $ - [ O.long "otelcol-no-traces" - , O.help "Disable traces exporter." - ] - otelcolServerParser :: O.Parser G.Server otelcolServerParser = makeServer @@ -884,3 +954,10 @@ <> O.help "Use SSLKEYLOGFILE to log SSL keys." ) ] + +------------------------------------------------------------------------------- +-- Internal Helpers +------------------------------------------------------------------------------- + +runIf :: (Monad m) => Bool -> MachineT m k o -> MachineT m k o +runIf b m = if b then m else stopped diff --git a/src/GHC/Eventlog/Live/Otelcol/Config.hs b/src/GHC/Eventlog/Live/Otelcol/Config.hs new file mode 100644 --- /dev/null +++ b/src/GHC/Eventlog/Live/Otelcol/Config.hs @@ -0,0 +1,532 @@ +{-# LANGUAGE OverloadedStrings #-} + +{- | +Module : GHC.Eventlog.Live.Otelcol.Config +Description : The implementation of @eventlog-live-otelcol@. +Stability : experimental +Portability : portable +-} +module GHC.Eventlog.Live.Otelcol.Config ( + readConfig, + Config (..), + Processors (..), + Metrics (..), + Spans (..), + HeapAllocatedMetric (..), + BlocksSizeMetric (..), + HeapSizeMetric (..), + HeapLiveMetric (..), + MemCurrentMetric (..), + MemNeededMetric (..), + MemReturnedMetric (..), + HeapProfSampleMetric (..), + CapabilityUsageMetric (..), + CapabilityUsageSpan (..), + ThreadStateSpan (..), + processorEnabled, + processorDescription, + processorName, +) 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.Records (HasField) + +{- | +Read a `Config` from a configuration file. +-} +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 + +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 + +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 + +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 + +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 + +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 + +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 + +instance ToJSON HeapLiveMetric where + toJSON :: HeapLiveMetric -> Value + toJSON = genericToJSON encodingOptions + toEncoding :: HeapLiveMetric -> Encoding + toEncoding = genericToEncoding encodingOptions + +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 + +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 + +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 + +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 + +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 + +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 + +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 + +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" + } + +------------------------------------------------------------------------------- +-- Accessors +------------------------------------------------------------------------------- + +{- | +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)))) + +{- | +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)) + +{- | +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 +------------------------------------------------------------------------------- + +{- | +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. +-} +with :: (Foldable f, Monoid r) => (s -> f t) -> (t -> r) -> s -> r +with = flip ((.) . foldMap)