eventlog-live-otelcol 0.1.0.4 → 0.2.0.0
raw patch · 3 files changed
+127/−82 lines, 3 filesdep +vectordep ~eventlog-live
Dependencies added: vector
Dependency ranges changed: eventlog-live
Files
- CHANGELOG.md +6/−0
- eventlog-live-otelcol.cabal +11/−7
- src/GHC/Eventlog/Live/Otelcol.hs +110/−75
CHANGELOG.md view
@@ -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`
eventlog-live-otelcol.cabal view
@@ -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
src/GHC/Eventlog/Live/Otelcol.hs view
@@ -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