diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,9 @@ +### 0.2.0.0 + +- **BREAKING**: Support info and debug verbosity. This changes the semantics of `--verbosity=3` and up. +- Support reading the eventlog from stdin. +- Support reading the eventlog from file. + ### 0.1.0.3 - Support `random-1.2` diff --git a/eventlog-live-otelcol.cabal b/eventlog-live-otelcol.cabal --- a/eventlog-live-otelcol.cabal +++ b/eventlog-live-otelcol.cabal @@ -1,17 +1,18 @@ cabal-version: 3.0 name: eventlog-live-otelcol -version: 0.1.0.4 +version: 0.2.0.0 synopsis: Stream eventlog data to the OpenTelemetry Collector. description: This executable supports live streaming of eventlog data into the OpenTelemetry Collector. - > Usage: eventlog-live-otelcol --eventlog-socket SOCKET + > Usage: eventlog-live-otelcol (--eventlog-stdin | --eventlog-file FILE | + > --eventlog-socket SOCKET) > [--eventlog-socket-timeout NUM] > [--eventlog-socket-exponent NUM] > [--batch-interval NUM] [--eventlog-log-file FILE] > [-h Tcmdyrbi] [--service-name STRING] - > [-v|--verbosity NUM|quiet|error|warning] + > [-v|--verbosity quiet|error|warning|info|debug|0-4] > --otelcol-host HOST [--otelcol-port PORT] > [--otelcol-authority HOST] [--otelcol-ssl] > [--otelcol-certificate-store FILE] @@ -20,17 +21,19 @@ > [--otelcol-no-metrics] [--otelcol-no-traces] > > Available options: - > --eventlog-socket SOCKET Eventlog Unix socket. + > --eventlog-stdin Read the eventlog from stdin. + > --eventlog-file FILE Read the eventlog from a file. + > --eventlog-socket SOCKET Read the eventlog from a Unix socket. > --eventlog-socket-timeout NUM > Eventlog socket connection retry timeout in > microseconds. > --eventlog-socket-exponent NUM > Eventlog socket connection retry timeout exponent. - > --batch-interval NUM Batch interval in microseconds. + > --batch-interval NUM Batch interval in milliseconds. > --eventlog-log-file FILE Use file to log binary eventlog data. > -h Tcmdyrbi Heap profile breakdown. > --service-name STRING The name of the profiled service. - > -v,--verbosity NUM|quiet|error|warning + > -v,--verbosity quiet|error|warning|info|debug|0-4 > The verbosity threshold for logging. > --help Show this help text. > --version Show version information @@ -113,7 +116,7 @@ , base >=4.16 && <4.22 , bytestring >=0.11 && <0.13 , dlist >=1.0 && <1.1 - , eventlog-live >=0.1 && <0.2 + , eventlog-live >=0.2 && <0.3 , eventlog-live:options , eventlog-live:socket , ghc-events >=0.20 && <0.21 @@ -127,6 +130,7 @@ , random >=1.2 && <1.4 , text >=1.2 && <2.2 , unordered-containers >=0.2.20 && <0.3 + , vector >=0.11 && <0.14 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 @@ -30,13 +30,20 @@ import Data.ProtoLens (Message (defMessage)) import Data.Text (Text) import Data.Text qualified as T +import Data.Vector qualified as V import Data.Version (showVersion) import Data.Void (Void) import Data.Word (Word32, Word64) -import GHC.Eventlog.Live.Machines (Attr, AttrValue (..), CapabilityUsageSpan, MemReturnData (..), Metric (..), ThreadStateSpan (..), Tick, WithStartTime (..), (~=)) -import GHC.Eventlog.Live.Machines qualified as ELM +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.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.Socket (runWithEventlogSocket) +import GHC.Eventlog.Live.Socket (runWithEventlogSource) import GHC.Eventlog.Live.Verbosity (Verbosity) import GHC.RTS.Events (Event (..), HeapProfBreakdown (..), ThreadId) import Lens.Family2 ((&), (.~), (^.)) @@ -74,48 +81,75 @@ let OpenTelemetryExporterOptions{..} = openTelemetryExporterOptions let attrServiceName = ("service.name", maybe AttrNull (AttrText . (.serviceName)) maybeServiceName) G.withConnection G.def openTelemetryCollectorServer $ \conn -> do - runWithEventlogSocket + runWithEventlogSource + verbosity eventlogSocket eventlogSocketTimeout eventlogSocketTimeoutExponent batchInterval Nothing maybeEventlogLogFile - $ ELM.liftTick ELM.withStartTime - ~> fanout - [ processHeapEvents verbosity maybeHeapProfBreakdown - ~> mapping (fmap Left) - , processThreadEvents verbosity - ] - ~> mapping (partitionEithers . D.toList) - ~> fanout - [ filtered (const exportMetrics) - ~> mapping fst - ~> asScopeMetrics - [ OM.scope .~ eventlogLiveScope - ] - ~> asResourceMetric [] - ~> asExportMetricServiceRequest - ~> exportResourceMetrics conn - ~> mapping (either displayException displayException) - ~> errorWriter - , filtered (const exportTraces) - ~> mapping snd - ~> asScopeSpans - [ OT.scope .~ eventlogLiveScope - ] - ~> asResourceSpan - [ OT.resource - .~ messageWith - [ OM.attributes .~ mapMaybe toMaybeKeyValue [attrServiceName] - ] - ] - ~> asExportTraceServiceRequest - ~> exportResourceSpans conn - ~> mapping (either displayException displayException) - ~> errorWriter - ] + $ fanout + [ M.validateInput verbosity 10 + , M.counterByTick verbosity "events" + , M.liftTick M.withStartTime + ~> fanout + [ processHeapEvents verbosity maybeHeapProfBreakdown + ~> mapping (fmap Left) + , processThreadEvents 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] + ] + ] + ~> 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) + errorWriter :: (MonadIO m) => ProcessT m String Void errorWriter = repeatedly $ await >>= liftIO . IO.hPutStrLn IO.stderr @@ -133,34 +167,35 @@ Verbosity -> ProcessT m (Tick (WithStartTime Event)) (DList (Either OM.Metric OT.Span)) processThreadEvents verbosity = - ELM.sortByBatchTick (.value.evTime) - ~> ELM.liftTick + M.sortByBatchTick (.value.evTime) + ~> M.liftTick ( fanout - [ ELM.processGCSpans verbosity + [ M.validateOrder verbosity (.value.evTime) + , M.processGCSpans verbosity ~> mapping (D.singleton . A) - , ELM.processThreadStateSpans' ELM.tryGetTimeUnixNano (.value) ELM.setWithStartTime'value verbosity + , M.processThreadStateSpans' M.tryGetTimeUnixNano (.value) M.setWithStartTime'value verbosity ~> fanout - [ ELM.asMutatorSpans' (.value) ELM.setWithStartTime'value + [ M.asMutatorSpans' (.value) M.setWithStartTime'value ~> mapping (D.singleton . B) , mapping (D.singleton . C) ] ] ) - ~> ELM.liftTick + ~> M.liftTick ( asParts ~> mapping repackCapabilityUsageSpanOrThreadStateSpan ) ~> fanout - [ ELM.liftTick + [ M.liftTick ( mapping leftToMaybe ~> asParts ) ~> fanout - [ ELM.liftTick - ( ELM.processCapabilityUsageMetrics + [ M.liftTick + ( M.processCapabilityUsageMetrics ~> asNumberDataPoint ) - ~> ELM.batchByTickList + ~> M.batchByTickList ~> asSum [ OM.aggregationTemporality .~ OM.AGGREGATION_TEMPORALITY_DELTA , OM.isMonotonic .~ True @@ -171,20 +206,20 @@ , OM.unit .~ "ns" ] ~> mapping (D.singleton . Left) - , ELM.liftTick - ( ELM.dropStartTime + , M.liftTick + ( M.dropStartTime ~> asSpan ~> mapping (D.singleton . Right) ) - ~> ELM.batchByTick + ~> M.batchByTick ] - , ELM.liftTick + , M.liftTick ( mapping rightToMaybe ~> asParts ~> asSpan ~> mapping (D.singleton . Right) ) - ~> ELM.batchByTick + ~> M.batchByTick ] where repackCapabilityUsageSpanOrThreadStateSpan = \case @@ -230,8 +265,8 @@ processHeapAllocated :: Process (Tick (WithStartTime Event)) (DList OM.Metric) processHeapAllocated = - ELM.liftTick (ELM.processHeapAllocatedData ~> asNumberDataPoint) - ~> ELM.batchByTickList + M.liftTick (M.processHeapAllocatedData ~> asNumberDataPoint) + ~> M.batchByTickList ~> asSum [ OM.aggregationTemporality .~ OM.AGGREGATION_TEMPORALITY_DELTA , OM.isMonotonic .~ True @@ -248,8 +283,8 @@ processHeapSize :: Process (Tick (WithStartTime Event)) (DList OM.Metric) processHeapSize = - ELM.liftTick (ELM.processHeapSizeData ~> asNumberDataPoint) - ~> ELM.batchByTickList + M.liftTick (M.processHeapSizeData ~> asNumberDataPoint) + ~> M.batchByTickList ~> asGauge ~> asMetric [ OM.name .~ "HeapSize" @@ -263,8 +298,8 @@ processBlocksSize :: Process (Tick (WithStartTime Event)) (DList OM.Metric) processBlocksSize = - ELM.liftTick (ELM.processBlocksSizeData ~> asNumberDataPoint) - ~> ELM.batchByTickList + M.liftTick (M.processBlocksSizeData ~> asNumberDataPoint) + ~> M.batchByTickList ~> asGauge ~> asMetric [ OM.name .~ "BlocksSize" @@ -278,8 +313,8 @@ processHeapLive :: Process (Tick (WithStartTime Event)) (DList OM.Metric) processHeapLive = - ELM.liftTick (ELM.processHeapLiveData ~> asNumberDataPoint) - ~> ELM.batchByTickList + M.liftTick (M.processHeapLiveData ~> asNumberDataPoint) + ~> M.batchByTickList ~> asGauge ~> asMetric [ OM.name .~ "HeapLive" @@ -293,10 +328,10 @@ processMemReturn :: Process (Tick (WithStartTime Event)) (DList OM.Metric) processMemReturn = - ELM.liftTick ELM.processMemReturnData - ~> ELM.batchByTickList + M.liftTick M.processMemReturnData + ~> M.batchByTickList ~> fanout - [ ELM.liftBatch (asMemCurrent ~> asNumberDataPoint) + [ M.liftBatch (asMemCurrent ~> asNumberDataPoint) ~> asGauge ~> asMetric [ OM.name .~ "MemCurrent" @@ -304,7 +339,7 @@ , OM.unit .~ "{mblock}" ] ~> mapping D.singleton - , ELM.liftBatch (asMemNeeded ~> asNumberDataPoint) + , M.liftBatch (asMemNeeded ~> asNumberDataPoint) ~> asGauge ~> asMetric [ OM.name .~ "MemNeeded" @@ -312,7 +347,7 @@ , OM.unit .~ "{mblock}" ] ~> mapping D.singleton - , ELM.liftBatch (asMemReturned ~> asNumberDataPoint) + , M.liftBatch (asMemReturned ~> asNumberDataPoint) ~> asGauge ~> asMetric [ OM.name .~ "MemReturned" @@ -340,8 +375,8 @@ Maybe HeapProfBreakdown -> ProcessT m (Tick (WithStartTime Event)) (DList OM.Metric) processHeapProfSample verbosity maybeHeapProfBreakdown = - ELM.liftTick (ELM.processHeapProfSampleData verbosity maybeHeapProfBreakdown ~> asNumberDataPoint) - ~> ELM.batchByTickList + M.liftTick (M.processHeapProfSampleData verbosity maybeHeapProfBreakdown ~> asNumberDataPoint) + ~> M.batchByTickList ~> asGauge ~> asMetric [ OM.name .~ "HeapProfSample" @@ -490,7 +525,7 @@ messageWith [ OT.traceId .~ traceId , OT.spanId .~ spanId - , OT.name .~ ELM.showCapabilityUserCategory user + , OT.name .~ M.showCapabilityUserCategory user , OT.kind .~ OT.Span'SPAN_KIND_INTERNAL , OT.startTimeUnixNano .~ i.startTimeUnixNano , OT.endTimeUnixNano .~ i.endTimeUnixNano @@ -506,7 +541,7 @@ ] ] where - user = ELM.capabilityUser i + user = M.capabilityUser i -------------------------------------------------------------------------------- -- Interpret thread state spans @@ -522,15 +557,15 @@ messageWith [ OT.traceId .~ traceId , OT.spanId .~ spanId - , OT.name .~ ELM.showThreadStateCategory i.threadState + , OT.name .~ M.showThreadStateCategory i.threadState , OT.kind .~ OT.Span'SPAN_KIND_INTERNAL , OT.startTimeUnixNano .~ i.startTimeUnixNano , OT.endTimeUnixNano .~ i.endTimeUnixNano , OT.attributes .~ mapMaybe toMaybeKeyValue - [ "capability" ~= ELM.threadStateCap i.threadState - , "status" ~= (show <$> ELM.threadStateStatus i.threadState) + [ "capability" ~= M.threadStateCap i.threadState + , "status" ~= (show <$> M.threadStateStatus i.threadState) ] , OT.status .~ messageWith @@ -699,7 +734,7 @@ O.idm data Options = Options - { eventlogSocket :: EventlogSocket + { eventlogSocket :: EventlogSource , eventlogSocketTimeout :: Double , eventlogSocketTimeoutExponent :: Double , batchInterval :: Int @@ -713,7 +748,7 @@ optionsParser :: O.Parser Options optionsParser = Options - <$> eventlogSocketParser + <$> eventlogSourceParser <*> eventlogSocketTimeoutParser <*> eventlogSocketTimeoutExponentParser <*> batchIntervalParser