eventlog-live-otelcol (empty) → 0.1.0.0
raw patch · 5 files changed
+1015/−0 lines, 5 filesdep +basedep +bytestringdep +dlist
Dependencies added: base, bytestring, dlist, eventlog-live, eventlog-live-otelcol, ghc-events, grapesy, hashable, hs-opentelemetry-otlp, lens-family, machines, optparse-applicative, proto-lens, random, text, unordered-containers
Files
- CHANGELOG.md +3/−0
- LICENSE +30/−0
- app/Main.hs +8/−0
- eventlog-live-otelcol.cabal +131/−0
- src/GHC/Eventlog/Live/Otelcol.hs +843/−0
+ CHANGELOG.md view
@@ -0,0 +1,3 @@+### 0.1.0.0++- Initial release.
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) 2021 Well-Typed++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of Oleg Grenrus nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ app/Main.hs view
@@ -0,0 +1,8 @@+{-# LANGUAGE NoImplicitPrelude #-}+{-# OPTIONS_GHC -Wno-missing-signatures #-}++module Main where++import GHC.Eventlog.Live.Otelcol qualified (main)++main = GHC.Eventlog.Live.Otelcol.main
+ eventlog-live-otelcol.cabal view
@@ -0,0 +1,131 @@+cabal-version: 3.12+name: eventlog-live-otelcol+version: 0.1.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+ > [--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]+ > --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]+ >+ > Available options:+ > --eventlog-socket SOCKET Eventlog 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.+ > --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+ > The verbosity threshold for logging.+ > --help Show this help text.+ > --version Show version information+ >+ > OpenTelemetry Collector Server Options+ > --otelcol-host HOST Server hostname.+ > --otelcol-port PORT Server TCP port.+ > --otelcol-authority HOST Server authority.+ > --otelcol-ssl Use SSL.+ > --otelcol-certificate-store FILE+ > Store for certificate validation.+ > --otelcol-ssl-key-log FILE+ > 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: AGPL-3.0-only+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++source-repository head+ type: git+ location: https://github.com/well-typed/eventlog-live.git+ subdir: eventlog-live-otelcol++common language+ ghc-options:+ -Wall -Wcompat -Widentities -Wprepositive-qualified-module+ -Wredundant-constraints -Wunticked-promoted-constructors+ -Wunused-packages++ default-language: Haskell2010+ default-extensions:+ BangPatterns+ DataKinds+ DefaultSignatures+ DeriveFoldable+ DeriveFunctor+ DeriveTraversable+ DerivingStrategies+ DuplicateRecordFields+ FlexibleContexts+ FlexibleInstances+ GADTs+ GeneralizedNewtypeDeriving+ ImportQualifiedPost+ InstanceSigs+ KindSignatures+ LambdaCase+ MultiParamTypeClasses+ NoFieldSelectors+ NumericUnderscores+ OverloadedRecordDot+ RankNTypes+ RecordWildCards+ ScopedTypeVariables+ TupleSections+ TypeApplications+ TypeFamilies++library+ import: language+ hs-source-dirs: src+ exposed-modules: GHC.Eventlog.Live.Otelcol+ other-modules: PackageInfo_eventlog_live_otelcol+ autogen-modules: PackageInfo_eventlog_live_otelcol+ build-depends:+ , base >=4.16 && <4.22+ , bytestring >=0.11 && <0.13+ , dlist >=1.0 && <1.1+ , eventlog-live >=0.1 && <0.2+ , eventlog-live:options+ , eventlog-live:socket+ , ghc-events >=0.20 && <0.21+ , grapesy >=1.1.0 && <1.2+ , hashable >=1.4 && <1.6+ , hs-opentelemetry-otlp >=0.1.0 && <0.2+ , lens-family >=2.1.3 && <2.2+ , machines >=0.7.4 && <0.8+ , optparse-applicative >=0.19 && <0.20+ , proto-lens >=0.7.1 && <0.8+ , random >=1.3 && <1.4+ , text >=1.2 && <2.2+ , unordered-containers >=0.2.20 && <0.3++executable eventlog-live-otelcol+ import: language+ main-is: Main.hs+ hs-source-dirs: app+ build-depends: eventlog-live-otelcol
+ src/GHC/Eventlog/Live/Otelcol.hs view
@@ -0,0 +1,843 @@+{-# LANGUAGE OverloadedStrings #-}+{-# OPTIONS_GHC -Wno-name-shadowing #-}+{-# OPTIONS_GHC -Wno-orphans #-}++{- |+Module : GHC.Eventlog.Live.Otelcol+Description : The implementation of @eventlog-live-otelcol@.+Stability : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Otelcol (+ main,+) where++import Control.Applicative (asum)+import Control.Exception (Exception (..), catch)+import Control.Monad.IO.Class (MonadIO (..))+import Data.ByteString (ByteString)+import Data.DList (DList)+import Data.DList qualified as D+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.Fanout (fanout)+import Data.Maybe (fromMaybe, mapMaybe)+import Data.ProtoLens (Message (defMessage))+import Data.Text (Text)+import Data.Text qualified as T+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.Options+import GHC.Eventlog.Live.Socket (runWithEventlogSocket)+import GHC.Eventlog.Live.Verbosity (Verbosity)+import GHC.RTS.Events (Event (..), HeapProfBreakdown (..), ThreadId)+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.Extra qualified as O+import PackageInfo_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+import Proto.Opentelemetry.Proto.Common.V1.Common_Fields qualified as OC+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 System.IO qualified as IO+import System.Random (StdGen, initStdGen, uniformByteString)+import Text.Printf (printf)++{- |+The main function for @eventlog-live-otelcol@.+-}+main :: IO ()+main = do+ Options{..} <- O.execParser options+ let OpenTelemetryCollectorOptions{..} = openTelemetryCollectorOptions+ let OpenTelemetryExporterOptions{..} = openTelemetryExporterOptions+ let attrServiceName = ("service.name", maybe AttrNull (AttrText . (.serviceName)) maybeServiceName)+ G.withConnection G.def openTelemetryCollectorServer $ \conn -> do+ runWithEventlogSocket+ 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+ ]++errorWriter :: (MonadIO m) => ProcessT m String Void+errorWriter = repeatedly $ await >>= liftIO . IO.hPutStrLn IO.stderr++-- dumpBatch :: (MonadIO m, Show a) => ProcessT m [a] [a]+-- dumpBatch = traversing (\as -> for_ as (liftIO . print) >> pure as)++--------------------------------------------------------------------------------+-- processThreadEvents+--------------------------------------------------------------------------------++data OneOf a b c = A !a | B !b | C !c++processThreadEvents ::+ (MonadIO m) =>+ Verbosity ->+ ProcessT m (Tick (WithStartTime Event)) (DList (Either OM.Metric OT.Span))+processThreadEvents verbosity =+ ELM.sortByBatchTick (.value.evTime)+ ~> ELM.liftTick+ ( fanout+ [ ELM.processGCSpans verbosity+ ~> mapping (D.singleton . A)+ , ELM.processThreadStateSpans' ELM.tryGetTimeUnixNano (.value) ELM.setWithStartTime'value verbosity+ ~> fanout+ [ ELM.asMutatorSpans' (.value) ELM.setWithStartTime'value+ ~> mapping (D.singleton . B)+ , mapping (D.singleton . C)+ ]+ ]+ )+ ~> ELM.liftTick+ ( asParts+ ~> mapping repackCapabilityUsageSpanOrThreadStateSpan+ )+ ~> fanout+ [ ELM.liftTick+ ( mapping leftToMaybe+ ~> asParts+ )+ ~> fanout+ [ ELM.liftTick+ ( ELM.processCapabilityUsageMetrics+ ~> asNumberDataPoint+ )+ ~> ELM.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)+ , ELM.liftTick+ ( ELM.dropStartTime+ ~> asSpan+ ~> mapping (D.singleton . Right)+ )+ ~> ELM.batchByTick+ ]+ , ELM.liftTick+ ( mapping rightToMaybe+ ~> asParts+ ~> asSpan+ ~> mapping (D.singleton . Right)+ )+ ~> ELM.batchByTick+ ]+ where+ repackCapabilityUsageSpanOrThreadStateSpan = \case+ A i -> Left $ fmap Left i+ B i -> Left $ fmap Right i+ C i -> Right i.value++{- |+Internal helper.+Get the `Left` value, if any.+-}+leftToMaybe :: Either a b -> Maybe a+leftToMaybe = either Just (const Nothing)++{- |+Internal helper.+Get the `Right` value, if any.+-}+rightToMaybe :: Either a b -> Maybe b+rightToMaybe = either (const Nothing) Just++--------------------------------------------------------------------------------+-- processHeapEvents+--------------------------------------------------------------------------------++processHeapEvents ::+ (MonadIO m) =>+ Verbosity ->+ Maybe HeapProfBreakdown ->+ ProcessT m (Tick (WithStartTime Event)) (DList OM.Metric)+processHeapEvents verbosity maybeHeapProfBreakdown =+ fanout+ [ processHeapAllocated+ , processBlocksSize+ , processHeapSize+ , processHeapLive+ , processMemReturn+ , processHeapProfSample verbosity maybeHeapProfBreakdown+ ]++--------------------------------------------------------------------------------+-- HeapAllocated++processHeapAllocated :: Process (Tick (WithStartTime Event)) (DList OM.Metric)+processHeapAllocated =+ ELM.liftTick (ELM.processHeapAllocatedData ~> asNumberDataPoint)+ ~> ELM.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++--------------------------------------------------------------------------------+-- HeapSize++processHeapSize :: Process (Tick (WithStartTime Event)) (DList OM.Metric)+processHeapSize =+ ELM.liftTick (ELM.processHeapSizeData ~> asNumberDataPoint)+ ~> ELM.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++--------------------------------------------------------------------------------+-- BlocksSize++processBlocksSize :: Process (Tick (WithStartTime Event)) (DList OM.Metric)+processBlocksSize =+ ELM.liftTick (ELM.processBlocksSizeData ~> asNumberDataPoint)+ ~> ELM.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++--------------------------------------------------------------------------------+-- HeapLive++processHeapLive :: Process (Tick (WithStartTime Event)) (DList OM.Metric)+processHeapLive =+ ELM.liftTick (ELM.processHeapLiveData ~> asNumberDataPoint)+ ~> ELM.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++--------------------------------------------------------------------------------+-- MemReturn++processMemReturn :: Process (Tick (WithStartTime Event)) (DList OM.Metric)+processMemReturn =+ ELM.liftTick ELM.processMemReturnData+ ~> ELM.batchByTickList+ ~> fanout+ [ ELM.liftBatch (asMemCurrent ~> asNumberDataPoint)+ ~> asGauge+ ~> asMetric+ [ OM.name .~ "MemCurrent"+ , OM.description .~ "Report the number of megablocks currently allocated."+ , OM.unit .~ "{mblock}"+ ]+ ~> mapping D.singleton+ , ELM.liftBatch (asMemNeeded ~> asNumberDataPoint)+ ~> asGauge+ ~> asMetric+ [ OM.name .~ "MemNeeded"+ , OM.description .~ "Report the number of megablocks currently needed."+ , OM.unit .~ "{mblock}"+ ]+ ~> mapping D.singleton+ , ELM.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+ ]++asMemCurrent :: Process (Metric MemReturnData) (Metric Word32)+asMemCurrent = mapping (fmap (.current))++asMemNeeded :: Process (Metric MemReturnData) (Metric Word32)+asMemNeeded = mapping (fmap (.needed))++asMemReturned :: Process (Metric MemReturnData) (Metric Word32)+asMemReturned = mapping (fmap (.returned))++--------------------------------------------------------------------------------+-- HeapProfSample++processHeapProfSample ::+ (MonadIO m) =>+ Verbosity ->+ Maybe HeapProfBreakdown ->+ ProcessT m (Tick (WithStartTime Event)) (DList OM.Metric)+processHeapProfSample verbosity maybeHeapProfBreakdown =+ ELM.liftTick (ELM.processHeapProfSampleData verbosity maybeHeapProfBreakdown ~> asNumberDataPoint)+ ~> ELM.batchByTickList+ ~> asGauge+ ~> asMetric+ [ OM.name .~ "HeapProfSample"+ , OM.description .~ "Report a heap profile sample."+ , OM.unit .~ "By"+ ]+ ~> mapping D.singleton++--------------------------------------------------------------------------------+-- Machines+--------------------------------------------------------------------------------++eventlogLiveScope :: OC.InstrumentationScope+eventlogLiveScope =+ messageWith+ [ OC.name .~ T.pack EventlogLive.name+ , OC.version .~ T.pack (showVersion EventlogLive.version)+ ]++asExportMetricsServiceRequest :: Process [OM.ResourceMetrics] OMS.ExportMetricsServiceRequest+asExportMetricsServiceRequest = mapping $ (defMessage &) . (OM.resourceMetrics .~)++asExportTracesServiceRequest :: Process [OT.ResourceSpans] OTS.ExportTraceServiceRequest+asExportTracesServiceRequest = mapping $ (defMessage &) . (OTS.resourceSpans .~)++asExportMetricServiceRequest :: Process OM.ResourceMetrics OMS.ExportMetricsServiceRequest+asExportMetricServiceRequest = mapping (: []) ~> asExportMetricsServiceRequest++asExportTraceServiceRequest :: Process OT.ResourceSpans OTS.ExportTraceServiceRequest+asExportTraceServiceRequest = mapping (: []) ~> asExportTracesServiceRequest++asResourceMetrics :: [OM.ResourceMetrics -> OM.ResourceMetrics] -> Process [OM.ScopeMetrics] OM.ResourceMetrics+asResourceMetrics mod = mapping $ \scopeMetrics ->+ messageWith ((OM.scopeMetrics .~ scopeMetrics) : mod)++asResourceSpans :: [OT.ResourceSpans -> OT.ResourceSpans] -> Process [OT.ScopeSpans] OT.ResourceSpans+asResourceSpans mod = mapping $ \scopeSpans ->+ messageWith ((OT.scopeSpans .~ scopeSpans) : mod)++asResourceSpan :: [OT.ResourceSpans -> OT.ResourceSpans] -> Process OT.ScopeSpans OT.ResourceSpans+asResourceSpan mod = mapping (: []) ~> asResourceSpans mod++asResourceMetric :: [OM.ResourceMetrics -> OM.ResourceMetrics] -> Process OM.ScopeMetrics OM.ResourceMetrics+asResourceMetric mod = mapping (: []) ~> asResourceMetrics mod++asScopeSpans :: [OT.ScopeSpans -> OT.ScopeSpans] -> Process [OT.Span] OT.ScopeSpans+asScopeSpans mod = mapping $ \spans ->+ messageWith ((OT.spans .~ spans) : mod)++asScopeMetrics :: [OM.ScopeMetrics -> OM.ScopeMetrics] -> Process [OM.Metric] OM.ScopeMetrics+asScopeMetrics mod = mapping $ \metrics ->+ messageWith ((OM.metrics .~ metrics) : mod)++asMetric :: [OM.Metric -> OM.Metric] -> Process OM.Metric'Data OM.Metric+asMetric mod = mapping $ toMetric mod++toMetric :: [OM.Metric -> OM.Metric] -> OM.Metric'Data -> OM.Metric+toMetric mod metric'data = messageWith ((OM.maybe'data' .~ Just metric'data) : mod)++asGauge :: Process [OM.NumberDataPoint] OM.Metric'Data+asGauge =+ repeatedly $+ await >>= \case+ dataPoints+ | null dataPoints -> pure ()+ | otherwise -> yield . OM.Metric'Gauge . messageWith $ [OM.dataPoints .~ dataPoints]++asSum :: [OM.Sum -> OM.Sum] -> Process [OM.NumberDataPoint] OM.Metric'Data+asSum mod = repeatedly $ await >>= \dataPoints -> for_ (toSum mod dataPoints) yield++toSum :: [OM.Sum -> OM.Sum] -> [OM.NumberDataPoint] -> Maybe OM.Metric'Data+toSum mod dataPoints+ | 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+--------------------------------------------------------------------------------++--------------------------------------------------------------------------------+-- Interpret spans++class AsSpan v where+ -- | The `Key` type is used to index a `HashMap` in the default definition of `asSpan`.+ type Key v++ -- | The `toKey` function extracts a `Key` from the input value.+ toKey ::+ -- | The input value.+ v ->+ Key v++ toSpan ::+ -- | The input value.+ v ->+ -- | The trace ID.+ ByteString ->+ -- | The span ID.+ ByteString ->+ 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)+ where+ -- go :: (HashMap (Key v) ByteString, Maybe StdGen) -> PlanT (Is v) OT.Span m Void+ go (traceIds, maybeGen) = do+ -- Ensure the StdGen is initialised+ gen0 <- maybe (liftIO initStdGen) pure maybeGen+ -- Receive the next value+ i <- await+ -- Ensure the next value has a trace ID+ let ensureTraceId :: Maybe ByteString -> ((ByteString, StdGen), Maybe ByteString)+ ensureTraceId = wrap . maybe (uniformByteString 16 gen0) (,gen0)+ where+ wrap out@(traceId, _gen) = (out, Just traceId)+ let ((traceId, gen1), traceIds') = M.alterF ensureTraceId (toKey i) traceIds+ -- Ensure the next value has a span ID+ let (spanId, gen2) = uniformByteString 8 gen1+ -- Yield a span+ yield $ toSpan i traceId spanId+ -- Continue+ go (traceIds', Just gen2)++--------------------------------------------------------------------------------+-- Interpret capability usage spans++instance AsSpan CapabilityUsageSpan where+ type Key CapabilityUsageSpan = Int++ toKey :: CapabilityUsageSpan -> Int+ toKey = (.cap)++ toSpan :: CapabilityUsageSpan -> ByteString -> ByteString -> OT.Span+ toSpan i traceId spanId =+ messageWith+ [ OT.traceId .~ traceId+ , OT.spanId .~ spanId+ , OT.name .~ ELM.showCapabilityUserCategory user+ , OT.kind .~ OT.Span'SPAN_KIND_INTERNAL+ , OT.startTimeUnixNano .~ i.startTimeUnixNano+ , OT.endTimeUnixNano .~ i.endTimeUnixNano+ , OT.attributes+ .~ mapMaybe+ toMaybeKeyValue+ [ "capability" ~= i.cap+ , "user" ~= user+ ]+ , OT.status+ .~ messageWith+ [ OT.code .~ OT.Status'STATUS_CODE_OK+ ]+ ]+ where+ user = ELM.capabilityUser i++--------------------------------------------------------------------------------+-- Interpret thread state spans++instance AsSpan ThreadStateSpan where+ type Key ThreadStateSpan = ThreadId++ toKey :: ThreadStateSpan -> ThreadId+ toKey = (.thread)++ toSpan :: ThreadStateSpan -> ByteString -> ByteString -> OT.Span+ toSpan i traceId spanId =+ messageWith+ [ OT.traceId .~ traceId+ , OT.spanId .~ spanId+ , OT.name .~ ELM.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)+ ]+ , OT.status+ .~ messageWith+ [ OT.code .~ OT.Status'STATUS_CODE_OK+ ]+ ]++--------------------------------------------------------------------------------+-- Interpret metrics++class IsNumberDataPoint'Value v where+ toNumberDataPoint'Value :: v -> OM.NumberDataPoint'Value++instance IsNumberDataPoint'Value Double where+ toNumberDataPoint'Value :: Double -> OM.NumberDataPoint'Value+ toNumberDataPoint'Value = OM.NumberDataPoint'AsDouble++instance IsNumberDataPoint'Value Word32 where+ toNumberDataPoint'Value :: Word32 -> OM.NumberDataPoint'Value+ toNumberDataPoint'Value = OM.NumberDataPoint'AsInt . fromIntegral++-- | __Warning__: This instance may cause overflow.+instance IsNumberDataPoint'Value Word64 where+ toNumberDataPoint'Value :: Word64 -> OM.NumberDataPoint'Value+ toNumberDataPoint'Value = OM.NumberDataPoint'AsInt . fromIntegral++toNumberDataPoint :: (IsNumberDataPoint'Value v) => Metric v -> OM.NumberDataPoint+toNumberDataPoint i =+ messageWith+ [ 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+ ]++toMaybeKeyValue :: Attr -> Maybe OC.KeyValue+toMaybeKeyValue (k, v) =+ toMaybeAnyValue v <&> \v ->+ messageWith+ [ OC.key .~ k+ , OC.value .~ v+ ]++toMaybeAnyValue :: AttrValue -> Maybe OC.AnyValue+toMaybeAnyValue = \case+ AttrInt v -> Just $ messageWith [OC.intValue .~ fromIntegral v]+ AttrInt8 v -> Just $ messageWith [OC.intValue .~ fromIntegral v]+ AttrInt16 v -> Just $ messageWith [OC.intValue .~ fromIntegral v]+ AttrInt32 v -> Just $ messageWith [OC.intValue .~ fromIntegral v]+ AttrInt64 v -> Just $ messageWith [OC.intValue .~ v]+ AttrWord v -> Just $ messageWith [OC.intValue .~ fromIntegral v]+ AttrWord8 v -> Just $ messageWith [OC.intValue .~ fromIntegral v]+ AttrWord16 v -> Just $ messageWith [OC.intValue .~ fromIntegral v]+ AttrWord32 v -> Just $ messageWith [OC.intValue .~ fromIntegral v]+ AttrWord64 v -> Just $ messageWith [OC.intValue .~ fromIntegral v]+ AttrDouble v -> Just $ messageWith [OC.doubleValue .~ v]+ AttrText v -> Just $ messageWith [OC.stringValue .~ v]+ AttrNull -> Nothing++--------------------------------------------------------------------------------+-- DSL for writing messages++-- | Construct a message with a list of modifications applied.+messageWith :: (Message msg) => [msg -> msg] -> msg+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+--------------------------------------------------------------------------------++options :: O.ParserInfo Options+options =+ O.info+ ( optionsParser+ O.<**> O.helperWith (O.long "help" <> O.help "Show this help text.")+ O.<**> O.simpleVersioner (showVersion EventlogLive.version)+ )+ O.idm++data Options = Options+ { eventlogSocket :: EventlogSocket+ , eventlogSocketTimeout :: Double+ , eventlogSocketTimeoutExponent :: Double+ , batchInterval :: Int+ , maybeEventlogLogFile :: Maybe FilePath+ , maybeHeapProfBreakdown :: Maybe HeapProfBreakdown+ , maybeServiceName :: Maybe ServiceName+ , verbosity :: Verbosity+ , openTelemetryCollectorOptions :: OpenTelemetryCollectorOptions+ }++optionsParser :: O.Parser Options+optionsParser =+ Options+ <$> eventlogSocketParser+ <*> eventlogSocketTimeoutParser+ <*> eventlogSocketTimeoutExponentParser+ <*> batchIntervalParser+ <*> O.optional eventlogLogFileParser+ <*> O.optional heapProfBreakdownParser+ <*> O.optional serviceNameParser+ <*> verbosityParser+ <*> openTelemetryCollectorOptionsParser++--------------------------------------------------------------------------------+-- Service Name++newtype ServiceName = ServiceName {serviceName :: Text}++serviceNameParser :: O.Parser ServiceName+serviceNameParser =+ ServiceName+ <$> O.strOption+ ( O.long "service-name"+ <> O.metavar "STRING"+ <> O.help "The name of the profiled service."+ )++--------------------------------------------------------------------------------+-- OpenTelemetry Collector configuration++data OpenTelemetryCollectorOptions = OpenTelemetryCollectorOptions+ { openTelemetryCollectorServer :: G.Server+ , openTelemetryExporterOptions :: OpenTelemetryExporterOptions+ }++openTelemetryCollectorOptionsParser :: O.Parser OpenTelemetryCollectorOptions+openTelemetryCollectorOptionsParser =+ O.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+ <$> otelcolAddressParser+ <*> O.switch (O.long "otelcol-ssl" <> O.help "Use SSL.")+ <*> otelcolServerValidationParser+ <*> otelcolSslKeyLogParser+ where+ makeServer :: G.Address -> Bool -> G.ServerValidation -> G.SslKeyLog -> G.Server+ makeServer address ssl serverValidation sslKeyLog+ | ssl = G.ServerSecure serverValidation sslKeyLog address+ | otherwise = G.ServerInsecure address++otelcolAddressParser :: O.Parser G.Address+otelcolAddressParser =+ G.Address+ <$> O.strOption+ ( O.long "otelcol-host"+ <> O.metavar "HOST"+ <> O.help "Server hostname."+ )+ <*> O.option+ O.auto+ ( O.long "otelcol-port"+ <> O.metavar "PORT"+ <> O.help "Server TCP port."+ <> O.value 4317+ )+ <*> O.optional+ ( O.strOption+ ( O.long "otelcol-authority"+ <> O.metavar "HOST"+ <> O.help "Server authority."+ )+ )++otelcolServerValidationParser :: O.Parser G.ServerValidation+otelcolServerValidationParser =+ asum+ [ G.ValidateServer <$> otelcolCertificateStoreSpecParser+ , pure G.NoServerValidation+ ]+ where+ otelcolCertificateStoreSpecParser :: O.Parser G.CertificateStoreSpec+ otelcolCertificateStoreSpecParser =+ makeCertificateStoreSpec+ <$> O.optional+ ( O.strOption+ ( O.long "otelcol-certificate-store"+ <> O.metavar "FILE"+ <> O.help "Store for certificate validation."+ )+ )+ where+ makeCertificateStoreSpec :: Maybe FilePath -> G.CertificateStoreSpec+ makeCertificateStoreSpec = maybe G.certStoreFromSystem G.certStoreFromPath++otelcolSslKeyLogParser :: O.Parser G.SslKeyLog+otelcolSslKeyLogParser =+ asum+ [ G.SslKeyLogPath+ <$> O.strOption+ ( O.long "otelcol-ssl-key-log"+ <> O.metavar "FILE"+ <> O.help "Use file to log SSL keys."+ )+ , O.flag+ G.SslKeyLogNone+ G.SslKeyLogFromEnv+ ( O.long "otelcol-ssl-key-log-from-env"+ <> O.help "Use SSLKEYLOGFILE to log SSL keys."+ )+ ]