eventlog-live-otelcol 0.6.1.0 → 0.7.0.0
raw patch · 33 files changed
+1643/−1033 lines, 33 filesdep +case-insensitivedep +http-clientdep +http-client-tlsdep −ghc-stack-profiler-coredep ~eventlog-live
Dependencies added: case-insensitive, http-client, http-client-tls, http-types, ipedb, network-uri
Dependencies removed: ghc-stack-profiler-core
Dependency ranges changed: eventlog-live
Files
- CHANGELOG.md +69/−0
- data/config.schema.json +2/−2
- data/default.yaml +16/−13
- eventlog-live-otelcol.cabal +107/−63
- src-internal/GHC/Debug/Stub/Compat.hs +121/−0
- src-internal/GHC/Eventlog/Socket/Compat.hs +82/−0
- src-internal/Language/Haskell/TH/Lift/Compat.hs +13/−0
- src-internal/Options/Applicative/Compat.hs +34/−0
- src-internal/Options/Applicative/Extra/Feature.hs +62/−0
- src-internal/System/Random/Compat.hs +17/−0
- src/GHC/Debug/Stub/Compat.hs +0/−121
- src/GHC/Eventlog/Live/Otelcol.hs +95/−62
- src/GHC/Eventlog/Live/Otelcol/Config.hs +70/−60
- src/GHC/Eventlog/Live/Otelcol/Config/Types.hs +65/−123
- src/GHC/Eventlog/Live/Otelcol/Exporter/Core.hs +320/−0
- src/GHC/Eventlog/Live/Otelcol/Exporter/Logs.hs +18/−16
- src/GHC/Eventlog/Live/Otelcol/Exporter/Metrics.hs +18/−19
- src/GHC/Eventlog/Live/Otelcol/Exporter/Profiles.hs +18/−18
- src/GHC/Eventlog/Live/Otelcol/Exporter/Traces.hs +15/−13
- src/GHC/Eventlog/Live/Otelcol/Options.hs +128/−72
- src/GHC/Eventlog/Live/Otelcol/Processor/Common/Core.hs +7/−1
- src/GHC/Eventlog/Live/Otelcol/Processor/Common/Metrics.hs +7/−3
- src/GHC/Eventlog/Live/Otelcol/Processor/Common/ProfilesDictionary.hs +25/−13
- src/GHC/Eventlog/Live/Otelcol/Processor/Common/SymbolTable.hs +6/−6
- src/GHC/Eventlog/Live/Otelcol/Processor/Common/Traces.hs +1/−2
- src/GHC/Eventlog/Live/Otelcol/Processor/Heap.hs +8/−4
- src/GHC/Eventlog/Live/Otelcol/Processor/Logs.hs +1/−1
- src/GHC/Eventlog/Live/Otelcol/Processor/Profiles.hs +318/−213
- src/GHC/Eventlog/Socket/Compat.hs +0/−82
- src/Language/Haskell/TH/Lift/Compat.hs +0/−13
- src/Options/Applicative/Compat.hs +0/−34
- src/Options/Applicative/Extra/Feature.hs +0/−62
- src/System/Random/Compat.hs +0/−17
CHANGELOG.md view
@@ -1,3 +1,72 @@+### 0.7.0.0++- Add support for OTLP over HTTP/Protobuf. To use HTTP/Protobuf, use:+ - `--otlp-protocol=http/protobuf`+ - `--otlp-endpoint=http://my-http-endpoint:my-http-port`++ See `demo-http-protobuf` for a demo that uses `eventlog-live-otelcol` to+ sends data directly to Prometheus over HTTP/Protobuf.++- **BREAKING**: Rename `--otelcol` command-line options to `--otlp`, and+ change the separate `host`/`port`/`ssl` options to `--otlp-endpoint`,+ which accepts an URL similar to the OTLP environment variables.++ Moreover, the `certificate-store` and `ssl-key-log`/`ssl-key-log-from-env`+ options are now prefixed by `--otlp-grpc-`, e.g., `--otlp-grpc-ssl-key-log`.++- **BREAKING**: Rename `profiles` processors to `call_stack_profile` and+ `cost_centre_stack_profile`. These processors now respect the name given+ in the configuration file, e.g., using the default configuration, these+ profiles will be available in Pyroscope as...+ - `ghc_eventlog_CallStackProfile:cpu:samples`+ - `ghc_eventlog_CostCentreStackProfile:cpu:samples`++- **BREAKING**: Change default semantics of the configuration file.++ Starting from version 0.7.0.0, if the key for any specific processor+ is present in the file, even if it sets none of the properties, the+ processor is _enabled_, and any missing keys will default to the values+ in this file. Otherwise, the processor is _disabled_. For instance,+ the following configuration will run _only_ the `heap_prof_sample`+ processor with the default configuration.++ ```yaml+ processors:+ metrics:+ heap_prof_sample:+ ```++ Previously, if any key in the configuration file was commented out,+ it was given the value in the default configuration. Unfortunately,+ this was very unintuitive. Consider the following configuration...++ ```yaml+ processors:+ metrics:+ heap_prof_sample:+ name: ghc_eventlog_HeapProfSample+ aggregate: 5s+ export: 5s+ ```++ Previously, this would activate `heap_prof_sample` with the given+ configuration and activate _all other processors_ with their default+ configuration. Moreover, if the user edited the file and commented out+ the `heap_prof_sample` section...++ ```yaml+ processors:+ metrics:+ # heap_prof_sample:+ # name: ghc_eventlog_HeapProfSample+ # aggregate: 5s+ # export: 5s+ ```++ ...this would _not_ deactive the `heap_prof_sample` processor. Rather,+ it would keep the `heap_prof_sample` processor active with its default+ configuration.+ ### 0.6.1.0 - Support eventlog over TCP/IP.
data/config.schema.json view
@@ -45,8 +45,8 @@ "type": "object", "properties": { "defaults": { "#ref": "#/definitions/profile_defaults" },- "stack_sample": { "$ref": "#/definitions/profile" },- "cost_centre_sample": { "$ref": "#/definitions/profile" }+ "call_stack_profile": { "$ref": "#/definitions/profile" },+ "cost_centre_stack_profile": { "$ref": "#/definitions/profile" } }, "additionalProperties": false }
data/default.yaml view
@@ -5,8 +5,8 @@ # blocks_size: # name: ghc_eventlog_BlocksSize # description: The current heap size, calculated by the allocated number of blocks.-# aggregate: 1x-# export: 60x+# aggregate: 1s+# export: 60s # # - The `name` property is added to each piece of telemetry data, # and determines the name under which the data becomes available in your@@ -42,7 +42,7 @@ # processor exports telemetry data as if configured with "1x" (see above). # # The example `blocks_size` processor exports a batch of telemetry data every-# 30 seconds.+# 60 seconds. # # - The `aggregate` property determines whether or not the telemetry data is # aggregated over time, and, if so, over what interval. Only metrics support@@ -60,24 +60,27 @@ # The data produced by our example `blocks_size` processor aggregates data # over 1 second intervals and only keeps the last metric in each interval. #+# If the key for any specific processor is present in the file, even if it sets+# none of the properties, the processor is _enabled_, and any missing keys will+# default to the values in this file. Otherwise, the processor is _disabled_. processors: logs: thread_label: name: ghc_eventlog_ThreadLabel description: A thread label.- export: 5s+ export: 30s user_marker: name: ghc_eventlog_UserMarker description: A user marker.- export: 5s+ export: 30s user_message: name: ghc_eventlog_UserMessage description: A user log message.- export: 5s+ export: 30s internal_log_message: name: eventlog_live_InternalLogMessage description: An internal eventlog-live log message.- export: 5s+ export: 30s metrics: blocks_size: name: ghc_eventlog_BlocksSize@@ -134,11 +137,11 @@ description: A trace of thread state changes (either running or stopped). export: false profiles:- stack_sample:- name: ghc_eventlog_StackSampleProfile- description: A thread RTS callstack sample.+ call_stack_profile:+ name: ghc_eventlog_CallStackProfile+ description: A GHC call-stack profile. export: 30s- cost_centre_sample:- name: ghc_eventlog_CostCentreProfile- description: A cost centre callstack sample.+ cost_centre_stack_profile:+ name: ghc_eventlog_CostCentreStackProfile+ description: A GHC cost-centre stack profile. export: 30s
eventlog-live-otelcol.cabal view
@@ -1,10 +1,16 @@ cabal-version: 3.0 name: eventlog-live-otelcol-version: 0.6.1.0+version: 0.7.0.0 synopsis: Stream eventlog data to the OpenTelemetry Collector. description:- This executable supports live streaming of eventlog data into the OpenTelemetry Collector.+ This executable collects telemetry data from any Haskell application+ and sends it over the OpenTelemetry protocol. The next release of this+ executable will be as part of the+ [@eventlog-live@ package](https://hackage.haskell.org/package/eventlog-live)+ under the name @eventlog-live-otlp@. + For more information, see [the README](https://github.com/well-typed/eventlog-live#readme).+ > Usage: eventlog-live-otelcol (--eventlog-stdin | --eventlog-file FILE | > --eventlog-socket SOCKET | > --eventlog-socket-host HOST@@ -13,73 +19,95 @@ > [--eventlog-socket-exponent NUMBER] > [--eventlog-flush-interval SECONDS] > [--eventlog-log-file FILE] [-h Tcmdyrbi]- > [--service-name STRING]+ > [--service-name STRING] [--ipedb FILE]+ > [--ccdb FILE] > [-v|--verbosity fatal|error|warning|info|debug|trace]- > [-s|--stats] [--config FILE] --otelcol-host HOST- > [--otelcol-port PORT] [--otelcol-authority HOST]- > [--otelcol-ssl] [--otelcol-certificate-store FILE]- > [--otelcol-ssl-key-log FILE |- > --otelcol-ssl-key-log-from-env]- > [--print-defaults] [--print-config-json-schema]+ > [-s|--stats] [--config FILE] [--otlp-protocol ARG]+ > --otlp-endpoint ARG+ > [--otlp-grpc-certificate-store FILE]+ > [--otlp-grpc-ssl-key-log FILE |+ > --otlp-grpc-ssl-key-log-from-env]+ > [--otlp-http-headers ARG] [--print-defaults]+ > [--print-config-json-schema] > > Available options:- > --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-host HOST Read the eventlog from a TCP/IP socket.- > --eventlog-socket-port PORT Read the eventlog from a TCP/IP socket.- > --eventlog-socket-timeout SECONDS Eventlog socket connection retry timeout in seconds.- > --eventlog-socket-exponent NUMBER Eventlog socket connection retry timeout exponent.- > --eventlog-flush-interval SECONDS Eventlog flush interval in seconds.- > Should match the option passed to the application.- > --eventlog-log-file FILE Use file to log binary eventlog data.- > -h Tcmdyrbi Heap profile breakdown.- > Should match the option passed to the application.- > --service-name STRING The name of the profiled service.+ > --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-host HOST+ > Read the eventlog from a TCP/IP socket.+ > --eventlog-socket-port PORT+ > Read the eventlog from a TCP/IP socket.+ > --eventlog-socket-timeout SECONDS+ > Eventlog socket connection retry timeout in seconds.+ > --eventlog-socket-exponent NUMBER+ > Eventlog socket connection retry timeout exponent.+ > --eventlog-flush-interval SECONDS+ > Eventlog flush interval in seconds.+ > Should match the option passed to the application.+ > --eventlog-log-file FILE Use file to log binary eventlog data.+ > -h Tcmdyrbi Heap profile breakdown.+ > Should match the option passed to the application.+ > --service-name STRING The name of the profiled service.+ > --ipedb FILE The path to an IPE database.+ > --ccdb FILE The path a cost-centre database. > -v,--verbosity fatal|error|warning|info|debug|trace- > The severity threshold for logging.- > -s,--stats Display runtime statistics.- > --config FILE The path to a detailed configuration file.- > --print-defaults Print default configuration options.- > --print-config-json-schema Print JSON Schema for configuration format.- > --help Show this help text.- > --version Show version information+ > The severity threshold for logging.+ > -s,--stats Display runtime statistics.+ > --config FILE The path to a detailed configuration file.+ > --print-defaults Print default configuration options.+ > --print-config-json-schema+ > Print JSON Schema for configuration format.+ > --help Show this help text.+ > --version Show version information >- > OpenTelemetry Collector Server Options- > --otelcol-host HOST Otelcol server hostname.- > --otelcol-port PORT Otelcol server TCP port.- > --otelcol-authority HOST Otelcol 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.+ > OTLP Exporter Options+ > --otlp-protocol ARG The OTLP transport protocol to be used for all telemetry data (gRPC, HTTP/Protobuf).+ > Default value: gRPC+ > --otlp-endpoint ARG The OTLP endpoint URL for all telemetry data, with an optionally-specified port number.+ > Default value:+ > gRPC: http://localhost:4317+ > HTTP: http://localhost:4318+ > Example:+ > gRPC: https://my-api-endpoint:443+ > HTTP: http://my-api-endpoint/+ > --otlp-grpc-certificate-store FILE+ > Store for certificate validation.+ > --otlp-grpc-ssl-key-log FILE+ > Use file to log SSL keys.+ > --otlp-grpc-ssl-key-log-from-env+ > Use SSLKEYLOGFILE to log SSL keys.+ > --otlp-http-headers ARG A list of headers to apply to all outgoing data. > > Control Server Options- > --control Requires build with -f+control.- > Start the control server.- > --control-port Requires build with -f+control.- > The port number for the control server.+ > --control Unsupported. Requires build with -f+control.+ > Start the control server.+ > --control-port Unsupported. Requires build with -f+control.+ > The port number for the control server. > --control-cors-allow-origin- > Requires build with -f+control.- > Set the allowed origins for the control server CORS policy.- > --control-cors-max-age Requires build with -f+control.- > Set the maximum age of a cached CORS preflight request for the control server CORS policy.+ > Unsupported. Requires build with -f+control.+ > Set the allowed origins for the control server CORS policy.+ > --control-cors-max-age Unsupported. Requires build with -f+control.+ > Set the maximum age of a cached CORS preflight request for the control server CORS policy. > --control-cors-require-origin- > Requires build with -f+control.- > If enabled, the control server will not accept requests without an Origin header.+ > Unsupported. Requires build with -f+control.+ > If enabled, the control server will not accept requests without an Origin header. > --control-cors-ignore-failure- > Requires build with -f+control.- > If enabled, the control server will accept malformed CORS preflight requests.+ > Unsupported. Requires build with -f+control.+ > If enabled, the control server will accept malformed CORS preflight requests. > > Debug Options- > --my-eventlog-socket-unix Requires build with -f+use-eventlog-socket.- > Open an eventlog socket for this program on the given Unix socket.- > --my-ghc-debug-socket Requires build with -f+use-ghc-debug-stub.- > Open the default ghc-debug socket for this program.- > --my-ghc-debug-socket-unix Requires build with -f+use-ghc-debug-stub.- > Open a ghc-debug Unix domain socket with the given file path.- > --my-ghc-debug-socket-tcp Requires build with -f+use-ghc-debug-stub.- > Open a ghc-debug TCP/IP socket with the given address as 'host:port'.+ > --my-eventlog-socket-unix+ > Unsupported. Requires build with -f+use-eventlog-socket.+ > Open an eventlog socket for this program on the given Unix socket.+ > --my-ghc-debug-socket Unsupported. Requires build with -f+use-ghc-debug-stub.+ > Open the default ghc-debug socket for this program.+ > --my-ghc-debug-socket-unix+ > Unsupported. Requires build with -f+use-ghc-debug-stub.+ > Open a ghc-debug Unix domain socket with the given file path.+ > --my-ghc-debug-socket-tcp+ > Unsupported. Requires build with -f+use-ghc-debug-stub.+ > Open a ghc-debug TCP/IP socket with the given address as 'host:port'. license: BSD-3-Clause license-file: LICENSE@@ -105,7 +133,7 @@ type: git location: https://github.com/well-typed/eventlog-live.git subdir: eventlog-live-otelcol- tag: eventlog-live-otelcol-0.6.1.0+ tag: eventlog-live-otelcol-v0.7.0.0 -- This flag enables the control command server, which is an HTTP endpoint -- that gets started by eventlog-live-otelcol. If +control is set, then@@ -166,6 +194,7 @@ DuplicateRecordFields FlexibleContexts FlexibleInstances+ FunctionalDependencies GADTs GeneralizedNewtypeDeriving ImportQualifiedPost@@ -188,15 +217,19 @@ library import: language- hs-source-dirs: src- exposed-modules: GHC.Eventlog.Live.Otelcol+ hs-source-dirs:+ src++ exposed-modules:+ GHC.Eventlog.Live.Otelcol+ other-modules:- GHC.Debug.Stub.Compat GHC.Eventlog.Live.Otelcol.Config GHC.Eventlog.Live.Otelcol.Config.Default GHC.Eventlog.Live.Otelcol.Config.Default.Raw GHC.Eventlog.Live.Otelcol.Config.Types GHC.Eventlog.Live.Otelcol.Control+ GHC.Eventlog.Live.Otelcol.Exporter.Core GHC.Eventlog.Live.Otelcol.Exporter.Logs GHC.Eventlog.Live.Otelcol.Exporter.Metrics GHC.Eventlog.Live.Otelcol.Exporter.Profiles@@ -213,6 +246,12 @@ GHC.Eventlog.Live.Otelcol.Processor.Profiles GHC.Eventlog.Live.Otelcol.Processor.Threads GHC.Eventlog.Live.Otelcol.Stats++ hs-source-dirs:+ src-internal++ other-modules:+ GHC.Debug.Stub.Compat GHC.Eventlog.Socket.Compat Language.Haskell.TH.Lift.Compat Options.Applicative.Compat@@ -225,19 +264,24 @@ ansi-terminal >=1.1 && <1.2, base >=4.16 && <4.22, bytestring >=0.11 && <0.13,+ case-insensitive >=1.2 && <1.3, containers >=0.6 && <0.8, data-default >=0.2 && <0.9, dlist >=1.0 && <1.1,- eventlog-live >=0.5 && <0.6,+ eventlog-live ==0.6.0.0, file-embed >=0.0.16 && <0.1, ghc-events >=0.20 && <0.21,- ghc-stack-profiler-core >=0.2 && <0.3, grapesy >=1.0.0 && <1.2, hashable >=1.4 && <1.6, hs-opentelemetry-otlp >=0.2.0 && <0.3, HsYAML >=0.2 && <0.3,+ http-client >=0.7 && <0.8,+ http-client-tls >=0.3 && <0.4,+ http-types >=0.12 && <0.13,+ ipedb >=0.2.0.1 && <0.3, lens-family >=2.1.3 && <2.2, machines >=0.7.4 && <0.8,+ network-uri >=2.6 && <2.8, optparse-applicative >=0.17 && <0.20, proto-lens >=0.7.1 && <0.8, random >=1.2 && <1.4,
+ src-internal/GHC/Debug/Stub/Compat.hs view
@@ -0,0 +1,121 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE OverloadedStrings #-}++{- |+Module : GHC.Debug.Stub.Compat+Description : The implementation of @eventlog-live-otelcol@.+Stability : experimental+Portability : portable+-}+module GHC.Debug.Stub.Compat (+ MyGhcDebugSocket (..),+ withMyGhcDebug,+ maybeMyGhcDebugSocketParser,+) where++import Control.Applicative (asum)+import GHC.Eventlog.Live.Logger (Logger)+import Options.Applicative qualified as O+import Options.Applicative.Extra.Feature (Feature (..))+import Options.Applicative.Extra.Feature qualified as OF++#ifdef EVENTLOG_LIVE_OTELCOL_USE_GHC_DEBUG_STUB+import Data.Text qualified as T+import GHC.Debug.Stub qualified as GHC.Debug (withGhcDebug, withGhcDebugTCP, withGhcDebugUnix)+import GHC.Eventlog.Live.Data.Severity (Severity (..))+import GHC.Eventlog.Live.Logger (writeLog)+import System.Exit (exitFailure)+import Text.Read (readEither)+#else+import Data.Maybe (isJust)+import Control.Monad (when)+#endif++--------------------------------------------------------------------------------+-- Feature: use-ghc-debug-stub+--------------------------------------------------------------------------------++useGhcDebugStub :: Feature+useGhcDebugStub = Feature{flag = "use-ghc-debug-stub", isOn = isOn, info = "Cannot open ghc-debug socket."}+ where+ isOn :: Bool+#ifdef EVENTLOG_LIVE_OTELCOL_USE_GHC_DEBUG_STUB+ isOn = True+#else+ isOn = False+#endif++--------------------------------------------------------------------------------+-- My GHC Debug+--------------------------------------------------------------------------------++data MyGhcDebugSocket+ = MyGhcDebugSocketDefault+ | MyGhcDebugSocketUnix FilePath+ | MyGhcDebugSocketTcp String+ deriving (Show)++{- |+Internal helper.+Start @ghc-debug@ on the given `MyGhcDebugSocket`.+-}+withMyGhcDebug :: Logger IO -> Maybe MyGhcDebugSocket -> IO a -> IO a+#ifdef EVENTLOG_LIVE_OTELCOL_USE_GHC_DEBUG_STUB+withMyGhcDebug logger maybeMyGhcDebugSocket action =+ case maybeMyGhcDebugSocket of+ Nothing -> action+ Just MyGhcDebugSocketDefault -> do+ writeLog logger INFO $+ "Start ghc-debug with default socket."+ GHC.Debug.withGhcDebug action+ Just (MyGhcDebugSocketUnix myGhcDebugSocketUnix) -> do+ writeLog logger INFO $+ "Start ghc-debug with Unix domain socket at " <> T.pack myGhcDebugSocketUnix <> "."+ GHC.Debug.withGhcDebugUnix myGhcDebugSocketUnix action+ Just (MyGhcDebugSocketTcp myGhcDebugSocketTcp) -> do+ let (host, port) = break (== ':') myGhcDebugSocketTcp+ writeLog logger INFO $+ "Start ghc-debug with TCP/IP socket at " <> T.pack host <> ":" <> T.pack port <> "."+ case readEither port of+ Left _parseError -> do+ writeLog logger FATAL $+ T.pack $ "Could not parse ghc-debug TCP address " <> myGhcDebugSocketTcp <> "."+ exitFailure+ Right portWord16 ->+ GHC.Debug.withGhcDebugTCP host portWord16 action+#else+withMyGhcDebug logger maybeMyGhcDebugSocket action = do+ when (isJust maybeMyGhcDebugSocket) $+ OF.exitIfUnsupported useGhcDebugStub logger+ action+#endif++--------------------------------------------------------------------------------+-- My GHC Debug++maybeMyGhcDebugSocketParser :: O.Parser (Maybe MyGhcDebugSocket)+maybeMyGhcDebugSocketParser =+ asum $+ [ myGhcDebugSocketDefaultParser+ , myGhcDebugSocketUnixParser+ , myGhcDebugSocketTcpParser+ , pure Nothing+ ]++myGhcDebugSocketDefaultParser :: O.Parser (Maybe MyGhcDebugSocket)+myGhcDebugSocketDefaultParser =+ OF.onlyFor useGhcDebugStub (O.flag' $ Just MyGhcDebugSocketDefault) mempty $+ O.long "my-ghc-debug-socket"+ <> OF.helpFor useGhcDebugStub "Open the default ghc-debug socket for this program."++myGhcDebugSocketUnixParser :: O.Parser (Maybe MyGhcDebugSocket)+myGhcDebugSocketUnixParser =+ OF.onlyFor useGhcDebugStub (O.option (Just . MyGhcDebugSocketUnix <$> O.str)) (O.metavar "FILE") $+ O.long "my-ghc-debug-socket-unix"+ <> OF.helpFor useGhcDebugStub "Open a ghc-debug Unix domain socket with the given file path."++myGhcDebugSocketTcpParser :: O.Parser (Maybe MyGhcDebugSocket)+myGhcDebugSocketTcpParser =+ OF.onlyFor useGhcDebugStub (O.option (Just . MyGhcDebugSocketTcp <$> O.str)) (O.metavar "ADDRESS") $+ O.long "my-ghc-debug-socket-tcp"+ <> OF.helpFor useGhcDebugStub "Open a ghc-debug TCP/IP socket with the given address as 'host:port'."
+ src-internal/GHC/Eventlog/Socket/Compat.hs view
@@ -0,0 +1,82 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE OverloadedStrings #-}++{- |+Module : GHC.Eventlog.Socket.Compat+Description : The implementation of @eventlog-live-otelcol@.+Stability : experimental+Portability : portable+-}+module GHC.Eventlog.Socket.Compat (+ MyEventlogSocket (..),+ maybeMyEventlogSocketParser,+ startMyEventlogSocket,+) where++import Control.Applicative (asum)+import GHC.Eventlog.Live.Logger (Logger)+import Options.Applicative qualified as O+import Options.Applicative.Extra.Feature (Feature (..))+import Options.Applicative.Extra.Feature qualified as OF++#ifdef EVENTLOG_LIVE_OTELCOL_USE_EVENTLOG_SOCKET+import Data.Foldable (for_)+import Data.Text qualified as T+import GHC.Eventlog.Live.Data.Severity (Severity (..))+import GHC.Eventlog.Live.Logger (writeLog)+import GHC.Eventlog.Socket qualified as Eventlog.Socket+#else+import Control.Monad (when)+import Data.Maybe (isJust)+#endif++--------------------------------------------------------------------------------+-- Feature: use-eventlog-socket+--------------------------------------------------------------------------------++useEventlogSocket :: Feature+useEventlogSocket = Feature{flag = "use-eventlog-socket", isOn = isOn, info = "Cannot open eventlog socket."}+ where+ isOn :: Bool+#ifdef EVENTLOG_LIVE_OTELCOL_USE_EVENTLOG_SOCKET+ isOn = True+#else+ isOn = False+#endif++--------------------------------------------------------------------------------+-- My Eventlog Socket+--------------------------------------------------------------------------------++newtype MyEventlogSocket+ = MyEventlogSocketUnix FilePath++maybeMyEventlogSocketParser :: O.Parser (Maybe MyEventlogSocket)+maybeMyEventlogSocketParser =+ asum $+ [ myEventlogSocketUnixParser+ , pure Nothing+ ]++myEventlogSocketUnixParser :: O.Parser (Maybe MyEventlogSocket)+myEventlogSocketUnixParser =+ OF.onlyFor useEventlogSocket (O.option (Just . MyEventlogSocketUnix <$> O.str)) (O.metavar "FILE") $+ O.long "my-eventlog-socket-unix"+ <> OF.helpFor useEventlogSocket "Open an eventlog socket for this program on the given Unix socket."++{- |+Set @eventlog-socket@ as the eventlog writer.+-}+startMyEventlogSocket :: Logger IO -> Maybe MyEventlogSocket -> IO ()+#ifdef EVENTLOG_LIVE_OTELCOL_USE_EVENTLOG_SOCKET+startMyEventlogSocket logger maybeMyEventlogSocket =+ for_ maybeMyEventlogSocket $ \case+ MyEventlogSocketUnix myEventlogSocketUnix -> do+ writeLog logger INFO $+ "Start eventlog-socket with Unix domain socket at " <> T.pack myEventlogSocketUnix <> "."+ Eventlog.Socket.startWait myEventlogSocketUnix+#else+startMyEventlogSocket logger maybeMyEventlogSocket =+ when (isJust maybeMyEventlogSocket) $+ OF.exitIfUnsupported useEventlogSocket logger+#endif
+ src-internal/Language/Haskell/TH/Lift/Compat.hs view
@@ -0,0 +1,13 @@+{-# LANGUAGE CPP #-}++module Language.Haskell.TH.Lift.Compat (+ Exp,+ Lift (..),+ Q,+) where++#if defined(EVENTLOG_LIVE_OTELCOL_USE_TEMPLATE_HASKELL_LIFT)+import Language.Haskell.TH.Lift (Exp, Lift (..), Q)+#else+import Language.Haskell.TH.Syntax (Exp, Lift (..), Q)+#endif
+ src-internal/Options/Applicative/Compat.hs view
@@ -0,0 +1,34 @@+{-# LANGUAGE CPP #-}++module Options.Applicative.Compat (+ parserOptionGroup,+ simpleVersioner,+) where++#if MIN_VERSION_optparse_applicative(0,19,0)+import Options.Applicative (parserOptionGroup)+import Options.Applicative (simpleVersioner)+#else+#if MIN_VERSION_optparse_applicative(0,18,1)+import Options.Applicative (simpleVersioner)+#else+import Options.Applicative (infoOption, long, help, hidden)+#endif+import Options.Applicative (Parser)+#endif++#if MIN_VERSION_optparse_applicative(0,19,0)+#else+-- Prior to optparse-applicative-0.19.0.0, option groups were not supported,+-- so this definition simply drops the group.+parserOptionGroup :: String -> Parser a -> Parser a+parserOptionGroup _ p = p+#if MIN_VERSION_optparse_applicative(0,18,1)+#else+-- Prior to optparse-applicative-0.18.1.0, simpleVersioner was not defined,+-- so this definition is taken verbatim from optparse-applicative-0.18.1.0.+simpleVersioner :: String -> Parser (a -> a)+simpleVersioner version = infoOption version $+ mconcat [long "version", help "Show version information", hidden]+#endif+#endif
+ src-internal/Options/Applicative/Extra/Feature.hs view
@@ -0,0 +1,62 @@+{-# LANGUAGE OverloadedStrings #-}++module Options.Applicative.Extra.Feature (+ Feature (..),+ helpFor,+ onlyFor,+ exitIfUnsupported,+) where++import Control.Monad (unless)+import Data.Default (Default (..))+import Data.Text (Text)+import Data.Text qualified as T+import GHC.Eventlog.Live.Data.Severity (Severity (..))+import GHC.Eventlog.Live.Logger (Logger, writeLog)+import Options.Applicative qualified as O+import Options.Applicative.Help.Pretty qualified as OP+import System.Exit (exitFailure)++data Feature = Feature+ { flag :: !String+ , isOn :: !Bool+ , info :: !String+ }++{- |+Create a command-line help document for an option that depend on a feature flag.+-}+helpFor :: Feature -> String -> O.Mod f a+helpFor feature help+ | feature.isOn = O.help help+ | otherwise = O.helpDoc (Just $ OP.vcat [OP.pretty unsupported, OP.pretty help]) <> O.hidden+ where+ unsupported :: Text+ unsupported = T.pack "Unsupported. Requires build with -f+" <> T.pack feature.flag <> "."++{- |+Create a command-line option that depends on a feature flag.+-}+onlyFor ::+ forall f a.+ (O.HasName f, Default a) =>+ Feature ->+ (O.Mod f a -> O.Parser a) ->+ O.Mod f a ->+ (forall g x. (O.HasName g) => O.Mod g x) ->+ O.Parser a+onlyFor feature opt optIfSupported optAlways+ | feature.isOn = opt (optAlways <> optIfSupported)+ | otherwise = def <$ O.infoOption unsupportedInfo optAlways+ where+ unsupportedInfo :: String+ unsupportedInfo = feature.info <> " Requires build with -f+" <> feature.flag <> "."++{- |+Exit with the feature info.+-}+exitIfUnsupported :: Feature -> Logger IO -> IO ()+exitIfUnsupported feature logger =+ unless feature.isOn $ do+ writeLog logger FATAL (T.pack feature.info)+ exitFailure
+ src-internal/System/Random/Compat.hs view
@@ -0,0 +1,17 @@+{-# LANGUAGE CPP #-}++module System.Random.Compat (+ uniformByteString,+) where++#if MIN_VERSION_random(1,3,0)+import System.Random (uniformByteString)+#else+import Data.Bifunctor (Bifunctor (first))+import Data.ByteString (ByteString)+import Data.ByteString.Short (fromShort)+import System.Random (RandomGen (genShortByteString))++uniformByteString :: RandomGen g => Int -> g -> (ByteString, g)+uniformByteString n g = first fromShort (genShortByteString n g)+#endif
− src/GHC/Debug/Stub/Compat.hs
@@ -1,121 +0,0 @@-{-# LANGUAGE CPP #-}-{-# LANGUAGE OverloadedStrings #-}--{- |-Module : GHC.Debug.Stub.Compat-Description : The implementation of @eventlog-live-otelcol@.-Stability : experimental-Portability : portable--}-module GHC.Debug.Stub.Compat (- MyGhcDebugSocket (..),- withMyGhcDebug,- maybeMyGhcDebugSocketParser,-) where--import Control.Applicative (asum)-import GHC.Eventlog.Live.Logger (Logger)-import Options.Applicative qualified as O-import Options.Applicative.Extra.Feature (Feature (..))-import Options.Applicative.Extra.Feature qualified as OF--#ifdef EVENTLOG_LIVE_OTELCOL_USE_GHC_DEBUG_STUB-import Data.Text qualified as T-import GHC.Debug.Stub qualified as GHC.Debug (withGhcDebug, withGhcDebugTCP, withGhcDebugUnix)-import GHC.Eventlog.Live.Data.Severity (Severity (..))-import GHC.Eventlog.Live.Logger (writeLog)-import System.Exit (exitFailure)-import Text.Read (readEither)-#else-import Data.Maybe (isJust)-import Control.Monad (when)-#endif------------------------------------------------------------------------------------- Feature: use-ghc-debug-stub-----------------------------------------------------------------------------------useGhcDebugStub :: Feature-useGhcDebugStub = Feature{flag = "use-ghc-debug-stub", isOn = isOn, info = "Cannot open ghc-debug socket."}- where- isOn :: Bool-#ifdef EVENTLOG_LIVE_OTELCOL_USE_GHC_DEBUG_STUB- isOn = True-#else- isOn = False-#endif------------------------------------------------------------------------------------- My GHC Debug-----------------------------------------------------------------------------------data MyGhcDebugSocket- = MyGhcDebugSocketDefault- | MyGhcDebugSocketUnix FilePath- | MyGhcDebugSocketTcp String- deriving (Show)--{- |-Internal helper.-Start @ghc-debug@ on the given `MyGhcDebugSocket`.--}-withMyGhcDebug :: Logger IO -> Maybe MyGhcDebugSocket -> IO a -> IO a-#ifdef EVENTLOG_LIVE_OTELCOL_USE_GHC_DEBUG_STUB-withMyGhcDebug logger maybeMyGhcDebugSocket action =- case maybeMyGhcDebugSocket of- Nothing -> action- Just MyGhcDebugSocketDefault -> do- writeLog logger INFO $- "Start ghc-debug with default socket."- GHC.Debug.withGhcDebug action- Just (MyGhcDebugSocketUnix myGhcDebugSocketUnix) -> do- writeLog logger INFO $- "Start ghc-debug with Unix domain socket at " <> T.pack myGhcDebugSocketUnix <> "."- GHC.Debug.withGhcDebugUnix myGhcDebugSocketUnix action- Just (MyGhcDebugSocketTcp myGhcDebugSocketTcp) -> do- let (host, port) = break (== ':') myGhcDebugSocketTcp- writeLog logger INFO $- "Start ghc-debug with TCP/IP socket at " <> T.pack host <> ":" <> T.pack port <> "."- case readEither port of- Left _parseError -> do- writeLog logger FATAL $- T.pack $ "Could not parse ghc-debug TCP address " <> myGhcDebugSocketTcp <> "."- exitFailure- Right portWord16 ->- GHC.Debug.withGhcDebugTCP host portWord16 action-#else-withMyGhcDebug logger maybeMyGhcDebugSocket action = do- when (isJust maybeMyGhcDebugSocket) $- OF.exitIfUnsupported useGhcDebugStub logger- action-#endif------------------------------------------------------------------------------------- My GHC Debug--maybeMyGhcDebugSocketParser :: O.Parser (Maybe MyGhcDebugSocket)-maybeMyGhcDebugSocketParser =- asum $- [ myGhcDebugSocketDefaultParser- , myGhcDebugSocketUnixParser- , myGhcDebugSocketTcpParser- , pure Nothing- ]--myGhcDebugSocketDefaultParser :: O.Parser (Maybe MyGhcDebugSocket)-myGhcDebugSocketDefaultParser =- OF.onlyFor useGhcDebugStub (O.flag' $ Just MyGhcDebugSocketDefault) mempty $- O.long "my-ghc-debug-socket"- <> OF.helpFor useGhcDebugStub "Open the default ghc-debug socket for this program."--myGhcDebugSocketUnixParser :: O.Parser (Maybe MyGhcDebugSocket)-myGhcDebugSocketUnixParser =- OF.onlyFor useGhcDebugStub (O.option (Just . MyGhcDebugSocketUnix <$> O.str)) (O.metavar "FILE") $- O.long "my-ghc-debug-socket-unix"- <> OF.helpFor useGhcDebugStub "Open a ghc-debug Unix domain socket with the given file path."--myGhcDebugSocketTcpParser :: O.Parser (Maybe MyGhcDebugSocket)-myGhcDebugSocketTcpParser =- OF.onlyFor useGhcDebugStub (O.option (Just . MyGhcDebugSocketTcp <$> O.str)) (O.metavar "ADDRESS") $- O.long "my-ghc-debug-socket-tcp"- <> OF.helpFor useGhcDebugStub "Open a ghc-debug TCP/IP socket with the given address as 'host:port'."
src/GHC/Eventlog/Live/Otelcol.hs view
@@ -1,5 +1,4 @@ {-# LANGUAGE OverloadedStrings #-}-{-# OPTIONS_GHC -Wno-name-shadowing -fconstraint-solver-iterations=0 #-} {- | Module : GHC.Eventlog.Live.Otelcol@@ -17,24 +16,25 @@ import Data.DList qualified as D import Data.Default (Default (..)) import Data.Foldable qualified as F-import Data.Machine (Process, ProcessT, asParts, mapping, (~>))-import Data.Maybe (catMaybes, fromMaybe, mapMaybe)+import Data.Machine (Process, ProcessT, asParts, mapping, stopped, (~>))+import Data.Maybe (catMaybes, fromMaybe, isJust, mapMaybe) import Data.Text (Text) import Data.Text qualified as T import Data.Version (showVersion)+import Data.Void (absurd) import GHC.Debug.Stub.Compat (withMyGhcDebug) import GHC.Eventlog.Live.Data.Attribute (AttrValue (AttrText), (~=)) import GHC.Eventlog.Live.Data.LogRecord (LogRecord (..)) import GHC.Eventlog.Live.Data.Severity (Severity (..)) import GHC.Eventlog.Live.Logger (MyTelemetryData, writeLog) import GHC.Eventlog.Live.Logger qualified as M-import GHC.Eventlog.Live.Machine.Analysis.Profile 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 qualified as M import GHC.Eventlog.Live.Otelcol.Config qualified as C import GHC.Eventlog.Live.Otelcol.Config.Types (FullConfig (..)) import GHC.Eventlog.Live.Otelcol.Control (ControlServerApi (..), startControlServer)+import GHC.Eventlog.Live.Otelcol.Exporter.Core (OtlpExporter, parseOtlpExporterOptions, withOtlpExporter) import GHC.Eventlog.Live.Otelcol.Exporter.Logs (exportResourceLogs) import GHC.Eventlog.Live.Otelcol.Exporter.Metrics (exportResourceMetrics) import GHC.Eventlog.Live.Otelcol.Exporter.Profiles (exportResourceProfiles)@@ -43,22 +43,21 @@ import GHC.Eventlog.Live.Otelcol.Processor.Common.Core import GHC.Eventlog.Live.Otelcol.Processor.Common.Logs (ToLogRecord (..), toExportLogsServiceRequest, toResourceLogs, toScopeLogs) import GHC.Eventlog.Live.Otelcol.Processor.Common.Metrics (toExportMetricsServiceRequest, toResourceMetrics, toScopeMetrics)-import GHC.Eventlog.Live.Otelcol.Processor.Common.ProfilesDictionary (toExportProfileServiceRequest) import GHC.Eventlog.Live.Otelcol.Processor.Common.Traces (toExportTracesServiceRequest, toResourceSpans, toScopeSpans) import GHC.Eventlog.Live.Otelcol.Processor.Heap (processHeapEvents) import GHC.Eventlog.Live.Otelcol.Processor.Logs (processLogEvents)-import GHC.Eventlog.Live.Otelcol.Processor.Profiles (processCallStackData, processProfileEvents)+import GHC.Eventlog.Live.Otelcol.Processor.Profiles (Sample, Stack, processProfileEvents, toExportProfileServiceRequest, toProfiles, toProfilesData, toResourceProfiles, toScopeProfiles) import GHC.Eventlog.Live.Otelcol.Processor.Threads (processThreadEvents) import GHC.Eventlog.Live.Otelcol.Stats (Stat (..), eventCountTick, processStats) import GHC.Eventlog.Live.Source (runWithEventlogSourceHandle, withEventlogSourceHandle) import GHC.Eventlog.Socket.Compat (startMyEventlogSocket) import GHC.RTS.Events (Event (..))+import IpeDB.Database qualified as DB+import IpeDB.Types.CostCentre qualified as CC+import IpeDB.Types.InfoProv qualified as IP import Lens.Family2 ((.~))-import Network.GRPC.Client qualified as G-import Network.GRPC.Common qualified as G import Options.Applicative qualified as O import Paths_eventlog_live_otelcol qualified as EventlogLive-import Proto.Opentelemetry.Proto.Collector.Profiles.V1development.ProfilesService_Fields qualified as OPS 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.Logs.V1.Logs qualified as OL@@ -67,6 +66,7 @@ import Proto.Opentelemetry.Proto.Profiles.V1development.Profiles qualified as OP import Proto.Opentelemetry.Proto.Resource.V1.Resource qualified as OR import Proto.Opentelemetry.Proto.Trace.V1.Trace qualified as OT+import System.Exit (die) {- | The main function for @eventlog-live-otelcol@.@@ -74,6 +74,7 @@ main :: IO () main = do Options{..} <- O.execParser options+ otlpExporterOptions' <- either die pure (parseOtlpExporterOptions otlpExporterOptions) -- Construct the logging action myTelemetryDataChan <- newTChanIO@@ -130,13 +131,38 @@ ] ] + -- Create machine that indexes CostCentre data.+ let indexCostCentreEvents ::+ DB.Table CC.CostCentreId CC.CostCentre ->+ ProcessT IO (Tick (M.WithStartTime Event)) (Tick x)+ indexCostCentreEvents ccdb+ -- If a cost-centre database was provided, don't index any new entries.+ | isJust maybeCCDBPath = stopped+ | otherwise = M.liftTick (DB.indexer (CC.toCostCentre . (.value)) def ccdb ~> mapping absurd)++ -- Create machine that indexes InfoProv data.+ let indexInfoProvEvents ::+ DB.Table IP.InfoProvId IP.InfoProv ->+ ProcessT IO (Tick (M.WithStartTime Event)) (Tick x)+ indexInfoProvEvents ipedb+ -- If an IPE database was provided, don't index any new entries.+ | isJust maybeIpeDBPath = stopped+ | otherwise = M.liftTick (DB.indexer (IP.toInfoProv . (.value)) def ipedb ~> mapping absurd)+ -- Create machine that processes eventlog data into telemetry data- let processEventlogTelemetry :: ProcessT IO (Tick Event) (Tick ResourceTelemetryData)- processEventlogTelemetry =+ let processEventlogTelemetry ::+ DB.Table CC.CostCentreId CC.CostCentre ->+ DB.Table IP.InfoProvId IP.InfoProv ->+ ProcessT IO (Tick Event) (Tick ResourceTelemetryData)+ processEventlogTelemetry ccdb ipedb = M.liftTick M.withStartTime ~> M.fanoutTick- [ -- Process the heap events.- processHeapEvents logger maybeHeapProfBreakdown fullConfig+ [ -- Process CostCentre events.+ indexCostCentreEvents ccdb+ , -- Process InfoProv events.+ indexInfoProvEvents ipedb+ , -- Process the heap events.+ processHeapEvents logger (Just ipedb) maybeHeapProfBreakdown fullConfig ~> mapping (fmap (fmap TelemetryData'Metric)) , -- Process the log events. processLogEvents fullConfig@@ -145,8 +171,8 @@ processThreadEvents logger fullConfig ~> mapping (fmap (fmap (either TelemetryData'Metric TelemetryData'Span))) , -- Process the profile events.- processProfileEvents logger fullConfig- ~> mapping (fmap (fmap TelemetryData'Profile))+ processProfileEvents logger ccdb ipedb fullConfig+ ~> mapping (fmap (fmap TelemetryData'Sample)) ] ~> M.liftTick (asResourceTelemetryData eventlogResource eventlogLiveScope) @@ -172,7 +198,7 @@ ~> M.liftTick (asResourceTelemetryData internalResource eventlogLiveScope) -- Create the full machine to process eventlog data.- let processAndExportTelemetry conn =+ let processAndExportTelemetry ccdb ipedb otlpExporter = M.fanoutTick [ -- Log a warning if no input has been received after 10 ticks. M.validateInput logger 10@@ -181,12 +207,12 @@ ~> mapping (fmap (D.singleton . EventCountStat)) , -- Process eventlog and internal telemetry... M.fanoutTickCC- [ processEventlogTelemetry ~> mapping (fmap D.singleton)+ [ processEventlogTelemetry ccdb ipedb ~> mapping (fmap D.singleton) , processInternalTelemetry ~> mapping (fmap D.singleton) ] ~> M.liftTick asParts -- ...and export it.- ~> exportResourceTelemetryData fullConfig conn+ ~> exportResourceTelemetryData fullConfig otlpExporter ] -- Process the statistics -- TODO: windowSize should be the maximum of all aggregation and export intervals@@ -196,32 +222,42 @@ ~> M.dropTick -- Open a connection to the OpenTelemetry Collector.- let OpenTelemetryCollectorOptions{..} = openTelemetryCollectorOptions- G.withConnection G.def openTelemetryCollectorServer $ \conn -> do- withEventlogSourceHandle- logger- eventlogSocketTimeoutS- eventlogSocketTimeoutExponent- eventlogSourceOptions- $ \eventlogSourceHandle -> do- -- Notify the control server of the connection status.- let newConnection = controlServerApi.notifyNewConnection serviceName eventlogSourceHandle- let endConnection = controlServerApi.notifyEndConnection serviceName- bracket_ newConnection endConnection $- -- Run the eventlog processor.- runWithEventlogSourceHandle+ withOtlpExporter otlpExporterOptions' $ \otlpExporter -> do+ DB.withNewSession def $ \session -> do+ let withCostCentreTable =+ case maybeCCDBPath of+ Nothing -> DB.withNewTable session def+ Just ccDBPath -> DB.withTableFrom session ccDBPath def+ let withInfoProvTable =+ case maybeIpeDBPath of+ Nothing -> DB.withNewTable session def+ Just ipeDBPath -> DB.withTableFrom session ipeDBPath def+ withCostCentreTable $ \ccdb ->+ withInfoProvTable $ \ipedb ->+ withEventlogSourceHandle logger- eventlogSourceHandle- fullConfig.batchIntervalMs- Nothing- maybeEventlogLogFile- (processAndExportTelemetry conn)+ eventlogSocketTimeoutS+ eventlogSocketTimeoutExponent+ eventlogSourceOptions+ $ \eventlogSourceHandle -> do+ -- Notify the control server of the connection status.+ let newConnection = controlServerApi.notifyNewConnection serviceName eventlogSourceHandle+ let endConnection = controlServerApi.notifyEndConnection serviceName+ bracket_ newConnection endConnection $+ -- Run the eventlog processor.+ runWithEventlogSourceHandle+ logger+ eventlogSourceHandle+ fullConfig.batchIntervalMs+ Nothing+ maybeEventlogLogFile+ (processAndExportTelemetry ccdb ipedb otlpExporter) data TelemetryData = TelemetryData'Log OL.LogRecord | TelemetryData'Metric OM.Metric | TelemetryData'Span OT.Span- | TelemetryData'Profile M.CallStackData+ | TelemetryData'Sample (Sample Stack) data ResourceTelemetryData = ResourceTelemetryData'Log OL.ResourceLogs@@ -235,9 +271,9 @@ -} exportResourceTelemetryData :: FullConfig ->- G.Connection ->+ OtlpExporter -> ProcessT IO (Tick ResourceTelemetryData) (Tick (DList Stat))-exportResourceTelemetryData fullConfig connection =+exportResourceTelemetryData fullConfig otlpExporter = M.fanoutTick [ -- Export logs. runIf (C.shouldExportLogs fullConfig) $@@ -247,7 +283,7 @@ -- making it impossible to not batch once per interval. ~> M.batchByTick ~> M.liftTick (mapping (toExportLogsServiceRequest . D.toList))- ~> exportResourceLogs connection+ ~> exportResourceLogs otlpExporter ~> M.liftTick (mapping (D.singleton . ExportLogsResultStat)) , -- Export metrics. runIf (C.shouldExportMetrics fullConfig) $@@ -255,7 +291,7 @@ -- NOTE: See note above. ~> M.batchByTick ~> M.liftTick (mapping (toExportMetricsServiceRequest . D.toList))- ~> exportResourceMetrics connection+ ~> exportResourceMetrics otlpExporter ~> M.liftTick (mapping (D.singleton . ExportMetricsResultStat)) , -- Export spans. runIf (C.shouldExportTraces fullConfig) $@@ -263,12 +299,12 @@ -- NOTE: See note above. ~> M.batchByTick ~> M.liftTick (mapping (toExportTracesServiceRequest . D.toList))- ~> exportResourceSpans connection+ ~> exportResourceSpans otlpExporter ~> M.liftTick (mapping (D.singleton . ExportTraceResultStat)) , -- Export profiles. runIf (C.shouldExportProfiles fullConfig) $ M.liftTick (mapping getResourceProfiles ~> asParts ~> mapping toExportProfileServiceRequest)- ~> exportResourceProfiles connection+ ~> exportResourceProfiles otlpExporter ~> M.liftTick (mapping (D.singleton . ExportProfileResultStat)) ] @@ -294,6 +330,7 @@ {- | Internal helper.+ Repack a stream of `TelemetryData` to batched `ResourceTelemetryData`. -} asResourceTelemetryData ::@@ -310,7 +347,7 @@ toResourceTelemetryData telemetryData = catMaybes [maybeResourceLogs, maybeResourceMetrics, maybeResourceSpans, maybeProfiles] where- (logRecords, metrics, spans, profiles) = partitionTelemetryData telemetryData+ (logRecords, metrics, spans, samples) = partitionTelemetryData telemetryData maybeResourceLogs = do scopeLogs <- toScopeLogs instrumentationScope logRecords@@ -325,29 +362,25 @@ resourceSpans <- toResourceSpans resource [scopeSpans] pure $ ResourceTelemetryData'Span resourceSpans maybeProfiles = do- (resourceProfile, dictionary) <-- ifNonEmpty profiles $- processCallStackData resource instrumentationScope profiles- pure $- ResourceTelemetryData'Profile $- messageWith- [ OPS.dictionary .~ dictionary- , OPS.resourceProfiles .~ [resourceProfile]- ]+ (profiles, dictionary) <- toProfiles samples+ scopeProfiles <- toScopeProfiles instrumentationScope profiles+ resourceProfiles <- toResourceProfiles resource [scopeProfiles]+ profilesData <- toProfilesData [resourceProfiles] dictionary+ pure $ ResourceTelemetryData'Profile profilesData {- | Partition a stream of `TelemetryData` batches to individual batches for each kind of telemetry data. -}-partitionTelemetryData :: [TelemetryData] -> ([OL.LogRecord], [OM.Metric], [OT.Span], [M.CallStackData])+partitionTelemetryData :: [TelemetryData] -> ([OL.LogRecord], [OM.Metric], [OT.Span], [Sample Stack]) partitionTelemetryData = go ([], [], [], []) where- go :: ([OL.LogRecord], [OM.Metric], [OT.Span], [M.CallStackData]) -> [TelemetryData] -> ([OL.LogRecord], [OM.Metric], [OT.Span], [M.CallStackData])- go (logs, metrics, spans, profiles) = \case- [] -> (reverse logs, reverse metrics, reverse spans, reverse profiles)- (TelemetryData'Log log : rest) -> go (log : logs, metrics, spans, profiles) rest- (TelemetryData'Metric metric : rest) -> go (logs, metric : metrics, spans, profiles) rest- (TelemetryData'Span span : rest) -> go (logs, metrics, span : spans, profiles) rest- (TelemetryData'Profile profile : rest) -> go (logs, metrics, spans, profile : profiles) rest+ go :: ([OL.LogRecord], [OM.Metric], [OT.Span], [Sample Stack]) -> [TelemetryData] -> ([OL.LogRecord], [OM.Metric], [OT.Span], [Sample Stack])+ go (logsRev, metricsRev, spansRev, samplesRev) = \case+ [] -> (reverse logsRev, reverse metricsRev, reverse spansRev, reverse samplesRev)+ (TelemetryData'Log log_ : rest) -> go (log_ : logsRev, metricsRev, spansRev, samplesRev) rest+ (TelemetryData'Metric metric : rest) -> go (logsRev, metric : metricsRev, spansRev, samplesRev) rest+ (TelemetryData'Span span_ : rest) -> go (logsRev, metricsRev, span_ : spansRev, samplesRev) rest+ (TelemetryData'Sample sample : rest) -> go (logsRev, metricsRev, spansRev, sample : samplesRev) rest {- | Internal helper.
src/GHC/Eventlog/Live/Otelcol/Config.hs view
@@ -58,8 +58,8 @@ Profiles (..), IsProfileProcessorConfig, shouldExportProfiles,- StackSampleProfile (..),- CostCentreSampleProfile (..),+ CallStackProfile (..),+ CostCentreStackProfile (..), -- ** Property types @@ -90,7 +90,7 @@ import Data.Hashable (Hashable) import Data.List.NonEmpty (NonEmpty (..)) import Data.Maybe (catMaybes, fromMaybe, mapMaybe)-import Data.Monoid (Any (..), First (..))+import Data.Monoid (Any (..)) import Data.Semigroup (Semigroup (..)) import Data.Text (Text) import Data.Text qualified as T@@ -254,61 +254,71 @@ def :: ThreadStateSpan def = $(getDefault @'["processors", "traces", "threadState"] defaultConfig) -instance Default StackSampleProfile where- def :: StackSampleProfile- def = $(getDefault @'["processors", "profiles", "stackSample"] defaultConfig)+instance Default CallStackProfile where+ def :: CallStackProfile+ def = $(getDefault @'["processors", "profiles", "callStackProfile"] defaultConfig) -instance Default CostCentreSampleProfile where- def :: CostCentreSampleProfile- def = $(getDefault @'["processors", "profiles", "costCentreSample"] defaultConfig)+instance Default CostCentreStackProfile where+ def :: CostCentreStackProfile+ def = $(getDefault @'["processors", "profiles", "costCentreStackProfile"] defaultConfig) ------------------------------------------------------------------------------- -- Accessors ------------------------------------------------------------------------------- {- |+Get the user-specified processor configuration.+-}+userProcessorConfig ::+ (Processors -> Maybe processorGroup) ->+ (processorGroup -> Maybe processorConfig) ->+ FullConfig ->+ Maybe processorConfig+userProcessorConfig group processor fullConfig =+ processor =<< group =<< fullConfig.config.processors++{- | Get whether or not a processor is enabled. -} processorEnabled ::- (HasField "enabled" b Bool) =>- (Processors -> Maybe a) ->- (a -> Maybe b) ->+ (HasField "export" processorConfig (Maybe ExportStrategy)) =>+ (Processors -> Maybe processorGroup) ->+ (processorGroup -> Maybe processorConfig) -> FullConfig -> Bool-processorEnabled group field =- getAny . with (.processors) (with group (with field (Any . (.enabled)))) . (.config)+processorEnabled group processor =+ isEnabled . userProcessorConfig group processor {- | Get the description corresponding to a processor. -} processorDescription ::+ forall a b. (Default b, HasField "description" b (Maybe Text)) => (Processors -> Maybe a) -> (a -> Maybe b) -> FullConfig -> Maybe Text-processorDescription group field =- (.description) . fromMaybe def . getFirst . with (.processors) (with group (First . field)) . (.config)+processorDescription group processor =+ (.description) . fromMaybe (def :: b) . userProcessorConfig group processor {- | Get the name corresponding to a processor.--__Warning:__ This assumes the value of @`def`.`name`@ is `Just` some `Text`. -} processorName :: forall a b.- (HasCallStack, Default b, HasField "name" b (Maybe Text)) =>+ (HasCallStack, Show b, Default b, HasField "name" b (Maybe Text)) => (Processors -> Maybe a) -> (a -> Maybe b) -> FullConfig -> Text-processorName group field =- fromMaybe defaultName . ((.name) <=< getFirst) . with (.processors) (with group (First . field)) . (.config)+processorName group processor =+ fromMaybe defaultName . ((.name) <=< userProcessorConfig group processor) where- defaultName :: (HasCallStack) => Text- defaultName = case (def :: b).name of- Nothing -> error "The default configuration for this metric has no name."- Just name -> name+ defaultName = fromMaybe (error errMsg) config.name+ where+ config = def :: b+ errMsg = "The default configuration has no name: " <> show config -------------------------------------------------------------------------------- -- Aggregation Strategy@@ -323,7 +333,7 @@ FullConfig -> Maybe AggregationStrategy processorAggregationStrategy group field =- (.aggregate) . fromMaybe def . getFirst . with (.processors) (with group (First . field)) . (.config)+ (.aggregate) . fromMaybe def . userProcessorConfig group field {- | Convert an `AggregationStrategy` to a number of batches.@@ -375,7 +385,7 @@ Config -> [AggregationStrategy] allAggregationStrategies =- catMaybes . with (.processors) (with (.metrics) (forEachMetricProcessor (.aggregate)))+ catMaybes . with (.processors) (with (.metrics) (forEachMetricProcessor ((.aggregate) =<<))) {- | Get the largest aggregation strategy in batches.@@ -400,7 +410,7 @@ FullConfig -> Maybe ExportStrategy processorExportStrategy group field =- (.export) . fromMaybe def . getFirst . with (.processors) (with group (First . field)) . (.config)+ (.export) . fromMaybe def . userProcessorConfig group field {- | Convert an `ExportStrategy` to a number of batches.@@ -449,7 +459,7 @@ Config -> [ExportStrategy] allExportStrategies =- catMaybes . with (.processors) (forEachProcessor (.export))+ catMaybes . with (.processors) (forEachProcessor ((.export) =<<)) {- | Get the largest export strategy in batches.@@ -510,7 +520,7 @@ (.processors) ( with (.logs)- (mconcat . forEachLogProcessor (Any . isEnabled . (.export)))+ (mconcat . forEachLogProcessor (Any . isEnabled)) ) . (.config) @@ -521,7 +531,7 @@ (.processors) ( with (.metrics)- (mconcat . forEachMetricProcessor (Any . isEnabled . (.export)))+ (mconcat . forEachMetricProcessor (Any . isEnabled)) ) . (.config) @@ -532,7 +542,7 @@ (.processors) ( with (.traces)- (mconcat . forEachTraceProcessor (Any . isEnabled . (.export)))+ (mconcat . forEachTraceProcessor (Any . isEnabled)) ) . (.config) @@ -543,7 +553,7 @@ (.processors) ( with (.profiles)- (mconcat . forEachProfileProcessor (Any . isEnabled . (.export)))+ (mconcat . forEachProfileProcessor (Any . isEnabled)) ) . (.config) @@ -557,16 +567,16 @@ forEachProcessor :: ( forall processorConfig. (IsProcessorConfig processorConfig) =>- processorConfig -> a+ Maybe processorConfig -> a ) -> Processors -> [a] forEachProcessor f processors =- mconcat- [ forEachLogProcessor f (fromMaybe def processors.logs)- , forEachMetricProcessor f (fromMaybe def processors.metrics)- , forEachTraceProcessor f (fromMaybe def processors.traces)- , forEachProfileProcessor f (fromMaybe def processors.profiles)+ concatMap (fromMaybe []) $+ [ forEachLogProcessor f <$> processors.logs+ , forEachMetricProcessor f <$> processors.metrics+ , forEachTraceProcessor f <$> processors.traces+ , forEachProfileProcessor f <$> processors.profiles ] {- |@@ -575,16 +585,16 @@ forEachLogProcessor :: ( forall traceProcessorConfig. (IsLogProcessorConfig traceProcessorConfig) =>- traceProcessorConfig -> a+ Maybe traceProcessorConfig -> a ) -> Logs -> [a] forEachLogProcessor f logs = [ -- NOTE: This should be kept in sync with the list of logs.- f $ fromMaybe def logs.threadLabel- , f $ fromMaybe def logs.userMarker- , f $ fromMaybe def logs.userMessage- , f $ fromMaybe def logs.internalLogMessage+ f logs.threadLabel+ , f logs.userMarker+ , f logs.userMessage+ , f logs.internalLogMessage ] {- |@@ -593,21 +603,21 @@ forEachMetricProcessor :: ( forall metricProcessorConfig. (IsMetricProcessorConfig metricProcessorConfig) =>- metricProcessorConfig -> a+ Maybe metricProcessorConfig -> a ) -> Metrics -> [a] forEachMetricProcessor f metrics = [ -- NOTE: This should be kept in sync with the list of metrics.- f $ fromMaybe def metrics.heapAllocated- , f $ fromMaybe def metrics.blocksSize- , f $ fromMaybe def metrics.heapSize- , f $ fromMaybe def metrics.heapLive- , f $ fromMaybe def metrics.memCurrent- , f $ fromMaybe def metrics.memNeeded- , f $ fromMaybe def metrics.memReturned- , f $ fromMaybe def metrics.heapProfSample- , f $ fromMaybe def metrics.capabilityUsage+ f metrics.heapAllocated+ , f metrics.blocksSize+ , f metrics.heapSize+ , f metrics.heapLive+ , f metrics.memCurrent+ , f metrics.memNeeded+ , f metrics.memReturned+ , f metrics.heapProfSample+ , f metrics.capabilityUsage ] {- |@@ -616,14 +626,14 @@ forEachTraceProcessor :: ( forall traceProcessorConfig. (IsTraceProcessorConfig traceProcessorConfig) =>- traceProcessorConfig -> a+ Maybe traceProcessorConfig -> a ) -> Traces -> [a] forEachTraceProcessor f traces = [ -- NOTE: This should be kept in sync with the list of traces.- f $ fromMaybe def traces.capabilityUsage- , f $ fromMaybe def traces.threadState+ f traces.capabilityUsage+ , f traces.threadState ] {- |@@ -632,14 +642,14 @@ forEachProfileProcessor :: ( forall profileProcessorConfig. (IsProfileProcessorConfig profileProcessorConfig) =>- profileProcessorConfig -> a+ Maybe profileProcessorConfig -> a ) -> Profiles -> [a] forEachProfileProcessor f profiles = [ -- NOTE: This should be kept in sync with the list of profiles.- f $ fromMaybe def profiles.stackSample- , f $ fromMaybe def profiles.costCentreSample+ f profiles.callStackProfile+ , f profiles.costCentreStackProfile ] -------------------------------------------------------------------------------
src/GHC/Eventlog/Live/Otelcol/Config/Types.hs view
@@ -46,8 +46,8 @@ -- *** Profile processor configuration types Profiles (..), IsProfileProcessorConfig,- StackSampleProfile (..),- CostCentreSampleProfile (..),+ CallStackProfile (..),+ CostCentreStackProfile (..), -- ** Property types Duration (..),@@ -88,7 +88,7 @@ newtype Config = Config { processors :: Maybe Processors }- deriving (Lift)+ deriving (Lift, Show) instance FromYAML Config where parseYAML :: YAML.Node YAML.Pos -> YAML.Parser Config@@ -112,7 +112,7 @@ , traces :: Maybe Traces , profiles :: Maybe Profiles }- deriving (Lift)+ deriving (Lift, Show) instance FromYAML Processors where parseYAML :: YAML.Node YAML.Pos -> YAML.Parser Processors@@ -149,7 +149,7 @@ , userMessage :: Maybe UserMessage , internalLogMessage :: Maybe InternalLogMessage }- deriving (Lift)+ deriving (Lift, Show) instance FromYAML Logs where parseYAML :: YAML.Node YAML.Pos -> YAML.Parser Logs@@ -194,7 +194,7 @@ , heapProfSample :: Maybe HeapProfSampleMetric , capabilityUsage :: Maybe CapabilityUsageMetric }- deriving (Lift)+ deriving (Lift, Show) instance FromYAML Metrics where parseYAML :: YAML.Node YAML.Pos -> YAML.Parser Metrics@@ -242,7 +242,7 @@ { capabilityUsage :: Maybe CapabilityUsageSpan , threadState :: Maybe ThreadStateSpan }- deriving (Lift)+ deriving (Lift, Show) instance FromYAML Traces where parseYAML :: YAML.Node YAML.Pos -> YAML.Parser Traces@@ -266,10 +266,10 @@ The configuration options for the profile processors. -} data Profiles = Profiles- { stackSample :: Maybe StackSampleProfile- , costCentreSample :: Maybe CostCentreSampleProfile+ { callStackProfile :: Maybe CallStackProfile+ , costCentreStackProfile :: Maybe CostCentreStackProfile }- deriving (Lift)+ deriving (Lift, Show) instance FromYAML Profiles where parseYAML :: YAML.Node YAML.Pos -> YAML.Parser Profiles@@ -277,16 +277,16 @@ -- NOTE: This should be kept in sync with the list of profiles. YAML.withMap "Profiles" $ \m -> Profiles- <$> m .:? "stack_sample"- <*> m .:? "cost_centre_sample"+ <$> m .:? "call_stack_profile"+ <*> m .:? "cost_centre_stack_profile" instance ToYAML Profiles where toYAML :: Profiles -> YAML.Node () toYAML profiles = -- NOTE: This should be kept in sync with the list of profiles. YAML.mapping- [ "stack_sample" .= profiles.stackSample- , "cost_centre_sample" .= profiles.costCentreSample+ [ "call_stack_profile" .= profiles.callStackProfile+ , "cost_centre_stack_profile" .= profiles.costCentreStackProfile ] -------------------------------------------------------------------------------@@ -301,7 +301,7 @@ , description :: Maybe Text , export :: Maybe ExportStrategy }- deriving (Lift)+ deriving (Lift, Show) instance FromYAML ThreadLabel where parseYAML :: YAML.Node YAML.Pos -> YAML.Parser ThreadLabel@@ -311,19 +311,15 @@ toYAML :: ThreadLabel -> YAML.Node () toYAML = genericToYAMLLogProcessorConfig -instance HasField "enabled" ThreadLabel Bool where- getField :: ThreadLabel -> Bool- getField = isEnabled . (.export)- {- |-The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Log.processUserMessageData`.+The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Log.processStackFrame'Message`. -} data UserMessage = UserMessage { name :: Maybe Text , description :: Maybe Text , export :: Maybe ExportStrategy }- deriving (Lift)+ deriving (Lift, Show) instance FromYAML UserMessage where parseYAML :: YAML.Node YAML.Pos -> YAML.Parser UserMessage@@ -333,10 +329,6 @@ toYAML :: UserMessage -> YAML.Node () toYAML = genericToYAMLLogProcessorConfig -instance HasField "enabled" UserMessage Bool where- getField :: UserMessage -> Bool- getField = isEnabled . (.export)- {- | The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Log.processUserMarkerData`. -}@@ -345,7 +337,7 @@ , description :: Maybe Text , export :: Maybe ExportStrategy }- deriving (Lift)+ deriving (Lift, Show) instance FromYAML UserMarker where parseYAML :: YAML.Node YAML.Pos -> YAML.Parser UserMarker@@ -355,10 +347,6 @@ toYAML :: UserMarker -> YAML.Node () toYAML = genericToYAMLLogProcessorConfig -instance HasField "enabled" UserMarker Bool where- getField :: UserMarker -> Bool- getField = isEnabled . (.export)- {- | The configuration options for internal log messages. -}@@ -367,7 +355,7 @@ , description :: Maybe Text , export :: Maybe ExportStrategy }- deriving (Lift)+ deriving (Lift, Show) instance FromYAML InternalLogMessage where parseYAML :: YAML.Node YAML.Pos -> YAML.Parser InternalLogMessage@@ -377,10 +365,6 @@ toYAML :: InternalLogMessage -> YAML.Node () toYAML = genericToYAMLLogProcessorConfig -instance HasField "enabled" InternalLogMessage Bool where- getField :: InternalLogMessage -> Bool- getField = isEnabled . (.export)- ------------------------------------------------------------------------------- -- Metrics -------------------------------------------------------------------------------@@ -394,7 +378,7 @@ , aggregate :: Maybe AggregationStrategy , export :: Maybe ExportStrategy }- deriving (Lift)+ deriving (Lift, Show) instance FromYAML HeapAllocatedMetric where parseYAML :: YAML.Node YAML.Pos -> YAML.Parser HeapAllocatedMetric@@ -404,10 +388,6 @@ toYAML :: HeapAllocatedMetric -> YAML.Node () toYAML = genericToYAMLMetricProcessorConfig -instance HasField "enabled" HeapAllocatedMetric Bool where- getField :: HeapAllocatedMetric -> Bool- getField = isEnabled . (.export)- {- | The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Heap.processHeapSizeData`. -}@@ -417,7 +397,7 @@ , aggregate :: Maybe AggregationStrategy , export :: Maybe ExportStrategy }- deriving (Lift)+ deriving (Lift, Show) instance FromYAML HeapSizeMetric where parseYAML :: YAML.Node YAML.Pos -> YAML.Parser HeapSizeMetric@@ -427,10 +407,6 @@ toYAML :: HeapSizeMetric -> YAML.Node () toYAML = genericToYAMLMetricProcessorConfig -instance HasField "enabled" HeapSizeMetric Bool where- getField :: HeapSizeMetric -> Bool- getField = isEnabled . (.export)- {- | The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Heap.processBlocksSizeData`. -}@@ -440,7 +416,7 @@ , aggregate :: Maybe AggregationStrategy , export :: Maybe ExportStrategy }- deriving (Lift)+ deriving (Lift, Show) instance FromYAML BlocksSizeMetric where parseYAML :: YAML.Node YAML.Pos -> YAML.Parser BlocksSizeMetric@@ -450,10 +426,6 @@ toYAML :: BlocksSizeMetric -> YAML.Node () toYAML = genericToYAMLMetricProcessorConfig -instance HasField "enabled" BlocksSizeMetric Bool where- getField :: BlocksSizeMetric -> Bool- getField = isEnabled . (.export)- {- | The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Heap.processHeapLiveData`. -}@@ -463,7 +435,7 @@ , aggregate :: Maybe AggregationStrategy , export :: Maybe ExportStrategy }- deriving (Lift)+ deriving (Lift, Show) instance FromYAML HeapLiveMetric where parseYAML :: YAML.Node YAML.Pos -> YAML.Parser HeapLiveMetric@@ -473,10 +445,6 @@ toYAML :: HeapLiveMetric -> YAML.Node () toYAML = genericToYAMLMetricProcessorConfig -instance HasField "enabled" HeapLiveMetric Bool where- getField :: HeapLiveMetric -> Bool- getField = isEnabled . (.export)- {- | The configuration options for the @memCurrent@ field `GHC.Eventlog.Live.Machine.Analysis.Heap.processMemReturnData`. -}@@ -486,7 +454,7 @@ , aggregate :: Maybe AggregationStrategy , export :: Maybe ExportStrategy }- deriving (Lift)+ deriving (Lift, Show) instance FromYAML MemCurrentMetric where parseYAML :: YAML.Node YAML.Pos -> YAML.Parser MemCurrentMetric@@ -496,10 +464,6 @@ toYAML :: MemCurrentMetric -> YAML.Node () toYAML = genericToYAMLMetricProcessorConfig -instance HasField "enabled" MemCurrentMetric Bool where- getField :: MemCurrentMetric -> Bool- getField = isEnabled . (.export)- {- | The configuration options for the @memNeeded@ field `GHC.Eventlog.Live.Machine.Analysis.Heap.processMemReturnData`. -}@@ -509,7 +473,7 @@ , aggregate :: Maybe AggregationStrategy , export :: Maybe ExportStrategy }- deriving (Lift)+ deriving (Lift, Show) instance FromYAML MemNeededMetric where parseYAML :: YAML.Node YAML.Pos -> YAML.Parser MemNeededMetric@@ -519,10 +483,6 @@ toYAML :: MemNeededMetric -> YAML.Node () toYAML = genericToYAMLMetricProcessorConfig -instance HasField "enabled" MemNeededMetric Bool where- getField :: MemNeededMetric -> Bool- getField = isEnabled . (.export)- {- | The configuration options for the @memReturned@ field `GHC.Eventlog.Live.Machine.Analysis.Heap.processMemReturnData`. -}@@ -532,7 +492,7 @@ , aggregate :: Maybe AggregationStrategy , export :: Maybe ExportStrategy }- deriving (Lift)+ deriving (Lift, Show) instance FromYAML MemReturnedMetric where parseYAML :: YAML.Node YAML.Pos -> YAML.Parser MemReturnedMetric@@ -542,10 +502,6 @@ toYAML :: MemReturnedMetric -> YAML.Node () toYAML = genericToYAMLMetricProcessorConfig -instance HasField "enabled" MemReturnedMetric Bool where- getField :: MemReturnedMetric -> Bool- getField = isEnabled . (.export)- {- | The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Heap.processHeapProfSampleData`. -}@@ -555,7 +511,7 @@ , aggregate :: Maybe AggregationStrategy , export :: Maybe ExportStrategy }- deriving (Lift)+ deriving (Lift, Show) instance FromYAML HeapProfSampleMetric where parseYAML :: YAML.Node YAML.Pos -> YAML.Parser HeapProfSampleMetric@@ -565,10 +521,6 @@ toYAML :: HeapProfSampleMetric -> YAML.Node () toYAML = genericToYAMLMetricProcessorConfig -instance HasField "enabled" HeapProfSampleMetric Bool where- getField :: HeapProfSampleMetric -> Bool- getField = isEnabled . (.export)- {- | The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Capability.processCapabilityUsageMetrics`. -}@@ -578,7 +530,7 @@ , aggregate :: Maybe AggregationStrategy , export :: Maybe ExportStrategy }- deriving (Lift)+ deriving (Lift, Show) instance FromYAML CapabilityUsageMetric where parseYAML :: YAML.Node YAML.Pos -> YAML.Parser CapabilityUsageMetric@@ -588,10 +540,6 @@ toYAML :: CapabilityUsageMetric -> YAML.Node () toYAML = genericToYAMLMetricProcessorConfig -instance HasField "enabled" CapabilityUsageMetric Bool where- getField :: CapabilityUsageMetric -> Bool- getField = isEnabled . (.export)- ------------------------------------------------------------------------------- -- Traces -------------------------------------------------------------------------------@@ -604,7 +552,7 @@ , description :: Maybe Text , export :: Maybe ExportStrategy }- deriving (Lift)+ deriving (Lift, Show) instance FromYAML CapabilityUsageSpan where parseYAML :: YAML.Node YAML.Pos -> YAML.Parser CapabilityUsageSpan@@ -614,10 +562,6 @@ toYAML :: CapabilityUsageSpan -> YAML.Node () toYAML = genericToYAMLTraceProcessorConfig -instance HasField "enabled" CapabilityUsageSpan Bool where- getField :: CapabilityUsageSpan -> Bool- getField = isEnabled . (.export)- {- | The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Thread.processThreadStateSpan`. -}@@ -626,7 +570,7 @@ , description :: Maybe Text , export :: Maybe ExportStrategy }- deriving (Lift)+ deriving (Lift, Show) instance FromYAML ThreadStateSpan where parseYAML :: YAML.Node YAML.Pos -> YAML.Parser ThreadStateSpan@@ -636,54 +580,46 @@ toYAML :: ThreadStateSpan -> YAML.Node () toYAML = genericToYAMLTraceProcessorConfig -instance HasField "enabled" ThreadStateSpan Bool where- getField :: ThreadStateSpan -> Bool- getField = isEnabled . (.export)+-------------------------------------------------------------------------------+-- Profiles+------------------------------------------------------------------------------- {- | The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Profile.processStackProfSampleData`. -}-data StackSampleProfile = StackSampleProfile+data CallStackProfile = CallStackProfile { name :: Maybe Text , description :: Maybe Text , export :: Maybe ExportStrategy }- deriving (Lift)+ deriving (Lift, Show) -instance FromYAML StackSampleProfile where- parseYAML :: YAML.Node YAML.Pos -> YAML.Parser StackSampleProfile- parseYAML = genericParseYAMLProfilerProcessorConfig "StackSampleProfile" StackSampleProfile+instance FromYAML CallStackProfile where+ parseYAML :: YAML.Node YAML.Pos -> YAML.Parser CallStackProfile+ parseYAML = genericParseYAMLProfilerProcessorConfig "CallStackProfile" CallStackProfile -instance ToYAML StackSampleProfile where- toYAML :: StackSampleProfile -> YAML.Node ()+instance ToYAML CallStackProfile where+ toYAML :: CallStackProfile -> YAML.Node () toYAML = genericToYAMLProfilerProcessorConfig -instance HasField "enabled" StackSampleProfile Bool where- getField :: StackSampleProfile -> Bool- getField = isEnabled . (.export)- {- | The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Profile.processCostCentreProfSampleData`. -}-data CostCentreSampleProfile = CostCentreSampleProfile+data CostCentreStackProfile = CostCentreStackProfile { name :: Maybe Text , description :: Maybe Text , export :: Maybe ExportStrategy }- deriving (Lift)+ deriving (Lift, Show) -instance FromYAML CostCentreSampleProfile where- parseYAML :: YAML.Node YAML.Pos -> YAML.Parser CostCentreSampleProfile- parseYAML = genericParseYAMLProfilerProcessorConfig "CostCentreSampleProfile" CostCentreSampleProfile+instance FromYAML CostCentreStackProfile where+ parseYAML :: YAML.Node YAML.Pos -> YAML.Parser CostCentreStackProfile+ parseYAML = genericParseYAMLProfilerProcessorConfig "CostCentreStackProfile" CostCentreStackProfile -instance ToYAML CostCentreSampleProfile where- toYAML :: CostCentreSampleProfile -> YAML.Node ()+instance ToYAML CostCentreStackProfile where+ toYAML :: CostCentreStackProfile -> YAML.Node () toYAML = genericToYAMLProfilerProcessorConfig -instance HasField "enabled" CostCentreSampleProfile Bool where- getField :: CostCentreSampleProfile -> Bool- getField = isEnabled . (.export)- ------------------------------------------------------------------------------- -- Configuration supertypes -------------------------------------------------------------------------------@@ -695,7 +631,6 @@ type IsProcessorConfig config = ( HasField "name" config (Maybe Text) , HasField "description" config (Maybe Text)- , HasField "enabled" config Bool , HasField "export" config (Maybe ExportStrategy) ) @@ -736,7 +671,7 @@ data Duration = DurationByBatches {batches :: !Int} | DurationBySeconds {seconds :: !Double}- deriving (Lift)+ deriving (Lift, Show) {- | Internal helper.@@ -788,7 +723,7 @@ data AggregationStrategy = AggregationStrategyBool {isOn :: !Bool} | AggregationStrategyDuration {duration :: !Duration}- deriving (Lift)+ deriving (Lift, Show) {- | Convert an `AggregationStrategy` to a number of seconds, if specified in seconds.@@ -827,19 +762,26 @@ data ExportStrategy = ExportStrategyBool {isOn :: !Bool} | ExportStrategyDuration {duration :: !Duration}- deriving (Lift)+ deriving (Lift, Show) {- | Check whether or not a processor is enabled based on its export strategy. -}-isEnabled :: Maybe ExportStrategy -> Bool-isEnabled = \case- Nothing -> False- Just ExportStrategyBool{..} -> isOn- Just ExportStrategyDuration{..} ->- case duration of- DurationByBatches{..} -> batches > 0- DurationBySeconds{..} -> seconds > 0+isEnabled ::+ forall processorConfig.+ (HasField "export" processorConfig (Maybe ExportStrategy)) =>+ Maybe processorConfig -> Bool+isEnabled =+ maybe False (isEnabledByExportStrategy . (.export))+ where+ isEnabledByExportStrategy :: Maybe ExportStrategy -> Bool+ isEnabledByExportStrategy = \case+ Nothing -> True+ Just ExportStrategyBool{..} -> isOn+ Just ExportStrategyDuration{..} ->+ case duration of+ DurationByBatches{..} -> batches > 0+ DurationBySeconds{..} -> seconds > 0 {- | Convert an `ExportStrategy` to a number of seconds, if specified in seconds.
+ src/GHC/Eventlog/Live/Otelcol/Exporter/Core.hs view
@@ -0,0 +1,320 @@+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE OverloadedStrings #-}++module GHC.Eventlog.Live.Otelcol.Exporter.Core (+ OtlpExporter (..),+ parseOtlpExporterOptions,+ withOtlpExporter,+ export,++ -- * Export via gRPC+ CanExportViaGrpc,++ -- * Export via HTTP/Protobuf+ CanExportViaHttpProtobuf (..),+ HttpError (..),+) where++import Control.Exception (Exception (..), throwIO)+import Control.Monad ((<=<))+import Data.ByteString (ByteString)+import Data.ByteString qualified as BS+import Data.ByteString.Char8 qualified as BSC+import Data.ByteString.Lazy qualified as BSL+import Data.CaseInsensitive qualified as CI+import Data.List qualified as L+import Data.Maybe (fromMaybe)+import Data.ProtoLens.Encoding qualified as Proto+import Data.ProtoLens.Message (Message (defMessage))+import Data.ProtoLens.Service.Types (HasMethodImpl (..))+import Data.Word (Word16)+import GHC.Eventlog.Live.Otelcol.Options (OtlpExporterOptions (..), OtlpProtocol (..))+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, StreamingType (..))+import Network.GRPC.Common.Protobuf qualified as G+import Network.GRPC.Common.StreamType qualified as G+import Network.HTTP.Client qualified as H+import Network.HTTP.Client.TLS qualified as H+import Network.HTTP.Types.Header qualified as HTTP+import Network.HTTP.Types.Status qualified as HTTP+import Network.URI qualified as URI+import Text.Read (readMaybe)++--------------------------------------------------------------------------------+-- OTLP Exporter+--------------------------------------------------------------------------------++data OtlpExporter+ = OtlpExporterGrpc !OtlpGrpcExporter+ | OtlpExporterHttpProtobuf !OtlpHttpProtobufExporter++{- |+Construct an t`OtlpExporter` from t`OtlpExporterOptions`.+-}+withOtlpExporter :: OtlpExporterOptions OtlpEndpoint -> (OtlpExporter -> IO a) -> IO a+withOtlpExporter options action =+ case options of+ OtlpExporterOptions{otlpEndpoint = Left otlpGrpcEndpoint, ..} -> do+ let options' = OtlpExporterOptions{otlpEndpoint = otlpGrpcEndpoint, ..}+ withOtlpGrpcExporter options' $ action . OtlpExporterGrpc+ OtlpExporterOptions{otlpEndpoint = Right otlpHttpEndpoint, ..} -> do+ let options' = OtlpExporterOptions{otlpEndpoint = otlpHttpEndpoint, ..}+ withOtlpHttpProtobufExporter options' $ action . OtlpExporterHttpProtobuf++{- |+The options for an OTLP endpoint.+-}+type OtlpEndpoint = Either OtlpGrpcEndpoint OtlpHttpEndpoint++{- |+Parse the OTLP endpoint from a t`String` to an t`OtlpEndpoint`.+-}+parseOtlpExporterOptions :: OtlpExporterOptions String -> Either String (OtlpExporterOptions OtlpEndpoint)+parseOtlpExporterOptions OtlpExporterOptions{..} = do+ otlpEndpoint' <- parseOtlpEndpoint otlpProtocol otlpEndpoint+ let !options' = OtlpExporterOptions{otlpEndpoint = otlpEndpoint', ..}+ case otlpProtocol of+ OtlpProtocolGrpc+ | Just headers <- otlpHttpHeaders+ , not (null headers) -> do+ let showHeaders = L.intercalate "," . map (\(name, value) -> name <> "=" <> value)+ Left $ "The grpc protocol does not support additional HTTP headers, found " <> showHeaders headers+ OtlpProtocolHttpProtobuf+ | Just _sslKeyLog <- otlpGrpcSslKeyLog ->+ Left $ "The http/protobuf protocol does not support the SSL key log."+ OtlpProtocolHttpProtobuf+ | Just certificateStore <- otlpGrpcCertificateStore ->+ Left $ "The http/protobuf protocol does not support the certificate store, found " <> certificateStore+ _otherwise -> pure options'++{- |+Parse an OTLP endpoint string as an t`OtlpEndpoint` depending on the t`OtlpProtocol`.+-}+parseOtlpEndpoint :: OtlpProtocol -> String -> Either String OtlpEndpoint+parseOtlpEndpoint = go True+ where+ go :: Bool -> OtlpProtocol -> String -> Either String OtlpEndpoint+ go retry otlpProtocol url =+ case URI.parseURI url of+ Just URI.URI{..} ->+ case otlpProtocol of+ OtlpProtocolGrpc+ | uriScheme `elem` ["http:", "https:"]+ , null uriPath+ , null uriQuery+ , null uriFragment -> do+ let !host = maybe "localhost" (.uriRegName) uriAuthority+ let !port = fromMaybe 4317 $ uriPortNumber uriAuthority+ let !secure = uriScheme == "https:"+ pure $ Left OtlpGrpcEndpoint{..}+ | otherwise ->+ Left $ "The gRPC protocol only supports HTTP and HTTPS and does not support an URI path, query, or fragment, found: " <> url+ OtlpProtocolHttpProtobuf+ | uriScheme `elem` ["http:", "https:"]+ , null uriQuery+ , null uriFragment -> do+ let !host = maybe "localhost" (.uriRegName) uriAuthority+ let !port = fromMaybe 4317 $ uriPortNumber uriAuthority+ let !auth = URI.nullURIAuth{URI.uriRegName = host, URI.uriPort = ':' : show port}+ let !baseURI = URI.nullURI{URI.uriScheme = uriScheme, URI.uriAuthority = Just auth, URI.uriPath = uriPath}+ pure $ Right OtlpHttpEndpoint{baseUrl = show baseURI}+ | otherwise ->+ Left $ "The HTTP/Protobuf protocol only supports HTTP and HTTPS and does not support an URI query or fragment, found: " <> url+ Nothing+ | retry ->+ go False otlpProtocol $ "http://" <> url+ | otherwise ->+ Left $ "Could not parse url " <> url++{- |+Internal helper.++Extract a t`Word16` port number from a `URI.URIAuth`.+-}+uriPortNumber :: Maybe URI.URIAuth -> Maybe Word16+uriPortNumber = readMaybe @Word16 <=< fmap (dropColon . (.uriPort))+ where+ dropColon :: String -> String+ dropColon = \case (':' : str) -> str; str -> str++{- |+Export telemetry data to the t`OtlpExporter`.+-}+export ::+ forall serv meth.+ ( CanExportViaGrpc serv meth+ , CanExportViaHttpProtobuf serv meth+ ) =>+ -- | The HTTP/Protobuf exporter.+ OtlpExporter ->+ -- | The request message.+ MethodInput serv meth ->+ IO (MethodOutput serv meth)+export = \case+ OtlpExporterGrpc exporter -> exportGrpc @serv @meth exporter+ OtlpExporterHttpProtobuf exporter -> exportHttpProtobuf @serv @meth exporter++--------------------------------------------------------------------------------+-- OTLP gRPC Exporter+--------------------------------------------------------------------------------++{- |+An opaque OTLP gRPC exporter.+-}+newtype OtlpGrpcExporter = OtlpGrpcExporter+ { connection :: G.Connection+ }++{- |+The options for an OTLP gRPC endpoint.+-}+data OtlpGrpcEndpoint = OtlpGrpcEndpoint+ { host :: !String+ , port :: !Word16+ , secure :: !Bool+ }++type CanExportViaGrpc serv meth =+ ( G.SupportsClientRpc (Protobuf serv meth)+ , G.SupportsStreamingType (Protobuf serv meth) 'NonStreaming+ , G.RequestMetadata (Protobuf serv meth) ~ G.NoMetadata+ )++withOtlpGrpcExporter :: OtlpExporterOptions OtlpGrpcEndpoint -> (OtlpGrpcExporter -> IO a) -> IO a+withOtlpGrpcExporter OtlpExporterOptions{otlpEndpoint = OtlpGrpcEndpoint{..}, ..} action =+ G.withConnection G.def server $ \connection -> action OtlpGrpcExporter{..}+ where+ server :: G.Server+ server+ | secure = G.ServerSecure serverValidation sslKeyLog address+ | otherwise = G.ServerInsecure address++ address = G.Address host (fromIntegral port) Nothing+ sslKeyLog = fromMaybe G.SslKeyLogNone otlpGrpcSslKeyLog+ serverValidation = G.ValidateServer $ maybe G.certStoreFromSystem G.certStoreFromPath otlpGrpcCertificateStore++exportGrpc ::+ forall serv meth.+ (CanExportViaGrpc serv meth) =>+ OtlpGrpcExporter ->+ MethodInput serv meth ->+ IO (MethodOutput serv meth)+exportGrpc grpcExporter input =+ G.getProto <$> G.nonStreaming grpcExporter.connection (G.rpc @(G.Protobuf serv meth)) (G.Proto input)++--------------------------------------------------------------------------------+-- OTLP HTTP/Protobuf Exporter+--------------------------------------------------------------------------------++{- |+The options for an OTLP HTTP/Protobuf endpoint.+-}+newtype OtlpHttpEndpoint = OtlpHttpEndpoint+ { baseUrl :: String+ }+ deriving (Show)++data OtlpHttpProtobufExporter = OtlpHttpProtobufExporter+ { manager :: H.Manager+ , baseUrl :: String+ , headers :: HTTP.RequestHeaders+ }++data HttpError+ = HttpStatusError+ { statusCode :: Int+ , statusMessage :: ByteString+ , responseBody :: ByteString+ }+ | HttpDecodeError+ { errorMessage :: String+ }+ deriving (Show)++instance Exception HttpError where+ displayException :: HttpError -> String+ displayException = \case+ HttpStatusError{..} ->+ "OpenTelemetry Collector HTTP/Protobuf endpoint returned status "+ <> show statusCode+ <> " "+ <> BSC.unpack statusMessage+ <> " with body: "+ <> BSC.unpack responseBody+ HttpDecodeError{..} ->+ "Could not decode OpenTelemetry Collector HTTP/Protobuf response: "+ <> errorMessage++{- |+Internal helper.++Run an action with an t`OtlpHttpProtobufExporter`.+-}+withOtlpHttpProtobufExporter :: OtlpExporterOptions OtlpHttpEndpoint -> (OtlpHttpProtobufExporter -> IO a) -> IO a+withOtlpHttpProtobufExporter OtlpExporterOptions{otlpEndpoint = OtlpHttpEndpoint{..}, ..} action = do+ -- Create an HTTP manager.+ manager <- H.newManager H.tlsManagerSettings+ -- Create the HTTP headers.+ let headers = [(CI.mk (BSC.pack name), BSC.pack value) | (name, value) <- fromMaybe [] otlpHttpHeaders]+ -- Run the action.+ action OtlpHttpProtobufExporter{..}++class+ ( Message (MethodInput serv meth)+ , Message (MethodOutput serv meth)+ ) =>+ CanExportViaHttpProtobuf serv meth+ where+ apiPath :: String++{- |+Send a Protobuf message over an HTTP connection.+-}+exportHttpProtobuf ::+ forall serv meth.+ (CanExportViaHttpProtobuf serv meth) =>+ -- | The HTTP/Protobuf exporter.+ OtlpHttpProtobufExporter ->+ -- | The request message.+ MethodInput serv meth ->+ IO (MethodOutput serv meth)+exportHttpProtobuf OtlpHttpProtobufExporter{..} req = do+ baseRequest <- H.parseRequest (baseUrl <> apiPath @serv @meth)+ let request =+ baseRequest+ { H.method = "POST"+ , H.requestBody = H.RequestBodyBS (Proto.encodeMessage req)+ , H.checkResponse = \_ _ -> pure ()+ , H.requestHeaders =+ [ (HTTP.hContentType, "application/x-protobuf")+ , (HTTP.hAccept, "application/x-protobuf")+ ]+ <> headers+ }+ response <- H.httpLbs request manager+ let status = H.responseStatus response+ let body = BSL.toStrict (H.responseBody response)+ if HTTP.statusIsSuccessful status+ then decodeResponseBody body+ else+ throwIO+ HttpStatusError+ { statusCode = HTTP.statusCode status+ , statusMessage = HTTP.statusMessage status+ , responseBody = body+ }++{- |+Internal helper.++Decode the HTTP response body into a Protobuf message.+-}+decodeResponseBody :: (Message msg) => ByteString -> IO msg+decodeResponseBody body+ | BS.null body = pure defMessage+ | otherwise =+ case Proto.decodeMessage body of+ Left errorMessage -> throwIO HttpDecodeError{..}+ Right msg -> pure msg
src/GHC/Eventlog/Live/Otelcol/Exporter/Logs.hs view
@@ -16,12 +16,10 @@ import Data.Text (Text) import Data.Vector qualified as V import GHC.Eventlog.Live.Machine.Core (Tick (..))+import GHC.Eventlog.Live.Otelcol.Exporter.Core (CanExportViaHttpProtobuf (..), OtlpExporter (..), export) import Lens.Family2 ((^.))-import Network.GRPC.Client qualified as G-import Network.GRPC.Client.StreamType.IO qualified as G import Network.GRPC.Common qualified as G import Network.GRPC.Common.Protobuf (Protobuf)-import Network.GRPC.Common.Protobuf qualified as G import Proto.Opentelemetry.Proto.Collector.Logs.V1.LogsService qualified as OLS import Proto.Opentelemetry.Proto.Collector.Logs.V1.LogsService_Fields qualified as OLS import Proto.Opentelemetry.Proto.Logs.V1.Logs qualified as OL@@ -67,9 +65,9 @@ -- OpenTelemetry gRPC Exporter for Logs exportResourceLogs ::- G.Connection ->+ OtlpExporter -> ProcessT IO (Tick OLS.ExportLogsServiceRequest) (Tick ExportLogsResult)-exportResourceLogs conn = construct $ go False+exportResourceLogs exporter = construct $ go False where go exportedResourceLogs = await >>= \case@@ -85,20 +83,20 @@ sendResourceLogs :: OLS.ExportLogsServiceRequest -> IO ExportLogsResult sendResourceLogs exportLogsServiceRequest =- doGrpc `catch` handleSomeException+ doExport `catch` handleSomeException where !exportedLogRecords = countLogRecordsInExportLogsServiceRequest exportLogsServiceRequest - doGrpc :: IO ExportLogsResult- doGrpc = do- G.nonStreaming conn (G.rpc @(Protobuf OLS.LogsService "export")) (G.Proto exportLogsServiceRequest) >>= \case- G.Proto resp- | resp ^. OLS.partialSuccess . OLS.rejectedLogRecords == 0 -> do- pure $ ExportLogsSuccess exportedLogRecords- | otherwise -> do- let !rejectedLogRecords = resp ^. OLS.partialSuccess . OLS.rejectedLogRecords- let !rejectedLogsError = RejectedLogsError{errorMessage = resp ^. OLS.partialSuccess . OLS.errorMessage, ..}- pure $ ExportLogsError exportedLogRecords rejectedLogRecords (SomeException rejectedLogsError)+ doExport :: IO ExportLogsResult+ doExport = do+ resp <- export @OLS.LogsService @"export" exporter exportLogsServiceRequest+ if resp ^. OLS.partialSuccess . OLS.rejectedLogRecords == 0+ then+ pure $ ExportLogsSuccess exportedLogRecords+ else do+ let !rejectedLogRecords = resp ^. OLS.partialSuccess . OLS.rejectedLogRecords+ let !rejectedLogsError = RejectedLogsError{errorMessage = resp ^. OLS.partialSuccess . OLS.errorMessage, ..}+ pure $ ExportLogsError exportedLogRecords rejectedLogRecords (SomeException rejectedLogsError) handleSomeException :: SomeException -> IO ExportLogsResult handleSomeException someException = pure $ ExportLogsError 0 exportedLogRecords someException@@ -106,6 +104,10 @@ type instance G.RequestMetadata (Protobuf OLS.LogsService meth) = G.NoMetadata type instance G.ResponseInitialMetadata (Protobuf OLS.LogsService meth) = G.NoMetadata type instance G.ResponseTrailingMetadata (Protobuf OLS.LogsService meth) = G.NoMetadata++instance CanExportViaHttpProtobuf OLS.LogsService "export" where+ apiPath :: String+ apiPath = "/v1/logs" -------------------------------------------------------------------------------- -- Internal Helpers
src/GHC/Eventlog/Live/Otelcol/Exporter/Metrics.hs view
@@ -16,12 +16,10 @@ import Data.Text (Text) import Data.Vector qualified as V import GHC.Eventlog.Live.Machine.Core (Tick (..))+import GHC.Eventlog.Live.Otelcol.Exporter.Core (CanExportViaHttpProtobuf (..), OtlpExporter (..), export) import Lens.Family2 ((^.))-import Network.GRPC.Client qualified as G-import Network.GRPC.Client.StreamType.IO qualified as G import Network.GRPC.Common qualified as G import Network.GRPC.Common.Protobuf (Protobuf)-import Network.GRPC.Common.Protobuf qualified as G import Proto.Opentelemetry.Proto.Collector.Metrics.V1.MetricsService qualified as OMS import Proto.Opentelemetry.Proto.Collector.Metrics.V1.MetricsService_Fields qualified as OMS import Proto.Opentelemetry.Proto.Metrics.V1.Metrics qualified as OM@@ -67,9 +65,9 @@ -- OpenTelemetry gRPC Exporter for Metrics exportResourceMetrics ::- G.Connection ->+ OtlpExporter -> ProcessT IO (Tick OMS.ExportMetricsServiceRequest) (Tick ExportMetricsResult)-exportResourceMetrics conn = construct $ go False+exportResourceMetrics exporter = construct $ go False where go exportedResourceMetrics = await >>= \case@@ -85,23 +83,20 @@ sendResourceMetrics :: OMS.ExportMetricsServiceRequest -> IO ExportMetricsResult sendResourceMetrics exportMetricsServiceRequest =- doGrpc `catch` handleSomeException+ doExport `catch` handleSomeException where !exportedDataPoints = countDataPointsInExportMetricsServiceRequest exportMetricsServiceRequest - doGrpc :: IO ExportMetricsResult- doGrpc = do- G.nonStreaming conn (G.rpc @(Protobuf OMS.MetricsService "export")) (G.Proto exportMetricsServiceRequest) >>= \case- G.Proto resp- | resp ^. OMS.partialSuccess . OMS.rejectedDataPoints == 0 -> do- pure $ ExportMetricsSuccess exportedDataPoints- | otherwise -> do- let !rejectedDataPoints = resp ^. OMS.partialSuccess . OMS.rejectedDataPoints- let !rejectedMetricsError = RejectedMetricsError{errorMessage = resp ^. OMS.partialSuccess . OMS.errorMessage, ..}- pure $ ExportMetricsError exportedDataPoints rejectedDataPoints (SomeException rejectedMetricsError)-- -- handleGrpcError :: G.GrpcError -> IO ExportMetricsResult- -- handleGrpcError grpcError = pure $ ExportMetricsError 0 exportedDataPoints (SomeException grpcError)+ doExport :: IO ExportMetricsResult+ doExport = do+ resp <- export @OMS.MetricsService @"export" exporter exportMetricsServiceRequest+ if resp ^. OMS.partialSuccess . OMS.rejectedDataPoints == 0+ then+ pure $ ExportMetricsSuccess exportedDataPoints+ else do+ let !rejectedDataPoints = resp ^. OMS.partialSuccess . OMS.rejectedDataPoints+ let !rejectedMetricsError = RejectedMetricsError{errorMessage = resp ^. OMS.partialSuccess . OMS.errorMessage, ..}+ pure $ ExportMetricsError exportedDataPoints rejectedDataPoints (SomeException rejectedMetricsError) handleSomeException :: SomeException -> IO ExportMetricsResult handleSomeException someException = pure $ ExportMetricsError 0 exportedDataPoints someException@@ -109,6 +104,10 @@ 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++instance CanExportViaHttpProtobuf OMS.MetricsService "export" where+ apiPath :: String+ apiPath = "/v1/metrics" -------------------------------------------------------------------------------- -- Internal Helpers
src/GHC/Eventlog/Live/Otelcol/Exporter/Profiles.hs view
@@ -17,12 +17,10 @@ import Data.Text (Text) import Data.Vector qualified as V import GHC.Eventlog.Live.Machine.Core (Tick (..))+import GHC.Eventlog.Live.Otelcol.Exporter.Core (CanExportViaHttpProtobuf (..), OtlpExporter (..), export) import Lens.Family2 ((^.))-import Network.GRPC.Client qualified as G-import Network.GRPC.Client.StreamType.IO qualified as G import Network.GRPC.Common qualified as G import Network.GRPC.Common.Protobuf (Protobuf)-import Network.GRPC.Common.Protobuf qualified as G import Proto.Opentelemetry.Proto.Collector.Profiles.V1development.ProfilesService qualified as OPS import Proto.Opentelemetry.Proto.Collector.Profiles.V1development.ProfilesService_Fields qualified as OPS import Proto.Opentelemetry.Proto.Profiles.V1development.Profiles qualified as OP@@ -61,9 +59,9 @@ -- OpenTelemetry gRPC Exporter for Profiles exportResourceProfiles ::- G.Connection ->+ OtlpExporter -> ProcessT IO (Tick OPS.ExportProfilesServiceRequest) (Tick ExportProfileResult)-exportResourceProfiles conn =+exportResourceProfiles exporter = construct $ go False where go exportedProfiles =@@ -80,29 +78,31 @@ sendResourceProfiles :: OPS.ExportProfilesServiceRequest -> IO ExportProfileResult sendResourceProfiles exportProfilesServiceRequest =- doGrpc `catch` handleSomeException+ doExport `catch` handleSomeException where !exportedProfiles = countSamplesInExportProfileServiceRequest exportProfilesServiceRequest - doGrpc :: IO ExportProfileResult- doGrpc = do- G.nonStreaming conn (G.rpc @(Protobuf OPS.ProfilesService "export")) (G.Proto exportProfilesServiceRequest) >>= \case- G.Proto resp- | resp ^. OPS.partialSuccess . OPS.rejectedProfiles == 0 ->- pure $ ExportProfileSuccess exportedProfiles- | otherwise -> do- let !rejectedProfiles = resp ^. OPS.partialSuccess . OPS.rejectedProfiles- let !rejectedMetricsError = RejectedProfilesError{errorMessage = resp ^. OPS.partialSuccess . OPS.errorMessage, ..}- pure $ ExportProfileError exportedProfiles rejectedProfiles (SomeException rejectedMetricsError)+ doExport :: IO ExportProfileResult+ doExport = do+ resp <- export @OPS.ProfilesService @"export" exporter exportProfilesServiceRequest+ if resp ^. OPS.partialSuccess . OPS.rejectedProfiles == 0+ then+ pure $ ExportProfileSuccess exportedProfiles+ else do+ let !rejectedProfiles = resp ^. OPS.partialSuccess . OPS.rejectedProfiles+ let !rejectedMetricsError = RejectedProfilesError{errorMessage = resp ^. OPS.partialSuccess . OPS.errorMessage, ..}+ pure $ ExportProfileError exportedProfiles rejectedProfiles (SomeException rejectedMetricsError) handleSomeException :: SomeException -> IO ExportProfileResult handleSomeException someException = pure $ ExportProfileError 0 exportedProfiles someException type instance G.RequestMetadata (Protobuf OPS.ProfilesService meth) = G.NoMetadata- type instance G.ResponseInitialMetadata (Protobuf OPS.ProfilesService meth) = G.NoMetadata- type instance G.ResponseTrailingMetadata (Protobuf OPS.ProfilesService meth) = G.NoMetadata++instance CanExportViaHttpProtobuf OPS.ProfilesService "export" where+ apiPath :: String+ apiPath = "/v1development/profiles" {- | Internal helper.
src/GHC/Eventlog/Live/Otelcol/Exporter/Traces.hs view
@@ -16,12 +16,10 @@ import Data.Text (Text) import Data.Vector qualified as V import GHC.Eventlog.Live.Machine.Core (Tick (..))+import GHC.Eventlog.Live.Otelcol.Exporter.Core (CanExportViaHttpProtobuf (..), OtlpExporter (..), export) import Lens.Family2 ((^.))-import Network.GRPC.Client qualified as G-import Network.GRPC.Client.StreamType.IO qualified as G import Network.GRPC.Common qualified as G import Network.GRPC.Common.Protobuf (Protobuf)-import Network.GRPC.Common.Protobuf qualified as G import Proto.Opentelemetry.Proto.Collector.Trace.V1.TraceService qualified as OTS import Proto.Opentelemetry.Proto.Collector.Trace.V1.TraceService_Fields qualified as OTS import Proto.Opentelemetry.Proto.Trace.V1.Trace qualified as OT@@ -63,9 +61,9 @@ -- OpenTelemetry gRPC Exporter for Traces exportResourceSpans ::- G.Connection ->+ OtlpExporter -> ProcessT IO (Tick OTS.ExportTraceServiceRequest) (Tick ExportTraceResult)-exportResourceSpans conn =+exportResourceSpans exporter = construct $ go False where go exportedResourceSpans =@@ -88,14 +86,14 @@ doGrpc :: IO ExportTraceResult doGrpc = do- G.nonStreaming conn (G.rpc @(Protobuf OTS.TraceService "export")) (G.Proto exportTraceServiceRequest) >>= \case- G.Proto resp- | resp ^. OTS.partialSuccess . OTS.rejectedSpans == 0 ->- pure $ ExportTraceSuccess exportedSpans- | otherwise -> do- let !rejectedSpans = resp ^. OTS.partialSuccess . OTS.rejectedSpans- let !rejectedMetricsError = RejectedSpansError{errorMessage = resp ^. OTS.partialSuccess . OTS.errorMessage, ..}- pure $ ExportTraceError exportedSpans rejectedSpans (SomeException rejectedMetricsError)+ resp <- export @OTS.TraceService @"export" exporter exportTraceServiceRequest+ if resp ^. OTS.partialSuccess . OTS.rejectedSpans == 0+ then+ pure $ ExportTraceSuccess exportedSpans+ else do+ let !rejectedSpans = resp ^. OTS.partialSuccess . OTS.rejectedSpans+ let !rejectedMetricsError = RejectedSpansError{errorMessage = resp ^. OTS.partialSuccess . OTS.errorMessage, ..}+ pure $ ExportTraceError exportedSpans rejectedSpans (SomeException rejectedMetricsError) handleSomeException :: SomeException -> IO ExportTraceResult handleSomeException someException = pure $ ExportTraceError 0 exportedSpans someException@@ -103,6 +101,10 @@ 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++instance CanExportViaHttpProtobuf OTS.TraceService "export" where+ apiPath :: String+ apiPath = "/v1/traces" -------------------------------------------------------------------------------- -- Internal Helpers
src/GHC/Eventlog/Live/Otelcol/Options.hs view
@@ -2,12 +2,15 @@ Options (..), MyDebugOptions (..), ServiceName (..),- OpenTelemetryCollectorOptions (..),+ OtlpExporterOptions (..),+ OtlpProtocol (..), options, ) where -import Control.Applicative (asum)+import Data.Char (toLower) import Data.Default (Default (..))+import Data.Functor (void)+import Data.List qualified as L import Data.Text qualified as T import Data.Version (showVersion) import GHC.Debug.Stub.Compat (MyGhcDebugSocket, maybeMyGhcDebugSocketParser)@@ -21,12 +24,14 @@ import GHC.Eventlog.Live.Source.Core (EventlogSourceOptions (..)) import GHC.Eventlog.Socket.Compat (MyEventlogSocket (..), maybeMyEventlogSocketParser) import GHC.RTS.Events (HeapProfBreakdown (..))-import Network.GRPC.Client qualified as G import Network.GRPC.Common qualified as G import Options.Applicative qualified as O import Options.Applicative.Compat qualified as OC import Options.Applicative.Extra qualified as OE+import Options.Applicative.Help.Pretty qualified as OP import Paths_eventlog_live_otelcol qualified as EventlogLive+import Text.ParserCombinators.ReadP (ReadP)+import Text.ParserCombinators.ReadP qualified as P options :: O.ParserInfo Options options =@@ -48,10 +53,12 @@ , maybeEventlogLogFile :: Maybe FilePath , maybeHeapProfBreakdown :: Maybe HeapProfBreakdown , maybeServiceName :: Maybe ServiceName+ , maybeIpeDBPath :: Maybe FilePath+ , maybeCCDBPath :: Maybe FilePath , severityThreshold :: Severity , stats :: Bool , maybeConfigFile :: Maybe FilePath- , openTelemetryCollectorOptions :: OpenTelemetryCollectorOptions+ , otlpExporterOptions :: OtlpExporterOptions String , controlOptions :: ControlOptions , myDebugOptions :: MyDebugOptions }@@ -66,10 +73,12 @@ <*> O.optional eventlogLogFileParser <*> O.optional heapProfBreakdownParser <*> O.optional serviceNameParser+ <*> O.optional ipeDBPathParser+ <*> O.optional ccDBPathParser <*> verbosityParser <*> statsParser <*> O.optional configFileParser- <*> openTelemetryCollectorOptionsParser+ <*> otlpExporterOptionsParser <*> controlOptionsParser <*> myDebugOptionsParser @@ -122,92 +131,127 @@ ) --------------------------------------------------------------------------------+-- InfoProv Tables++ipeDBPathParser :: O.Parser FilePath+ipeDBPathParser =+ O.strOption+ ( O.long "ipedb"+ <> O.metavar "FILE"+ <> O.help "The path to an IPE database."+ )++--------------------------------------------------------------------------------+-- CostCentre Tables++ccDBPathParser :: O.Parser FilePath+ccDBPathParser =+ O.strOption+ ( O.long "ccdb"+ <> O.metavar "FILE"+ <> O.help "The path a cost-centre database."+ )++-------------------------------------------------------------------------------- -- OpenTelemetry Collector configuration -newtype OpenTelemetryCollectorOptions = OpenTelemetryCollectorOptions- { openTelemetryCollectorServer :: G.Server+data OtlpExporterOptions a = OtlpExporterOptions+ { otlpProtocol :: !OtlpProtocol+ , otlpEndpoint :: !a+ , otlpGrpcCertificateStore :: !(Maybe FilePath)+ , otlpGrpcSslKeyLog :: !(Maybe G.SslKeyLog)+ , otlpHttpHeaders :: !(Maybe [(String, String)]) }--openTelemetryCollectorOptionsParser :: O.Parser OpenTelemetryCollectorOptions-openTelemetryCollectorOptionsParser =- OC.parserOptionGroup "OpenTelemetry Collector Server Options" $- OpenTelemetryCollectorOptions- <$> otelcolServerParser+ deriving stock (Functor, Foldable, Traversable) -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+otlpExporterOptionsParser :: O.Parser (OtlpExporterOptions String)+otlpExporterOptionsParser =+ OC.parserOptionGroup "OTLP Exporter Options" $+ OtlpExporterOptions+ <$> otlpProtocolParser+ <*> otlpEndpointParser+ <*> O.optional otlpGrpcCertificateStoreParser+ <*> O.optional otlpGrpcSslKeyLogParser+ <*> O.optional otlpHttpHeadersParser -otelcolAddressParser :: O.Parser G.Address-otelcolAddressParser =- G.Address- <$> O.strOption- ( O.long "otelcol-host"- <> O.metavar "HOST"- <> O.help "Otelcol server hostname."- )- <*> O.option- O.auto- ( O.long "otelcol-port"- <> O.metavar "PORT"- <> O.help "Otelcol server TCP port."- <> O.value 4317- )- <*> O.optional- ( O.strOption- ( O.long "otelcol-authority"- <> O.metavar "HOST"- <> O.help "Otelcol server authority."- )- )+data OtlpProtocol+ = OtlpProtocolGrpc+ | OtlpProtocolHttpProtobuf+ deriving (Show) -otelcolServerValidationParser :: O.Parser G.ServerValidation-otelcolServerValidationParser =- asum- [ G.ValidateServer <$> otelcolCertificateStoreSpecParser- , pure G.NoServerValidation+otlpProtocolParser :: O.Parser OtlpProtocol+otlpProtocolParser =+ O.option (O.maybeReader readOtlpProtocol) . mconcat $+ [ O.long "otlp-protocol"+ , O.helpDoc . Just . OP.vcat . fmap OP.pretty $+ [ "The OTLP transport protocol to be used for all telemetry data (gRPC, HTTP/Protobuf)."+ , "Default value: gRPC"+ ]+ , O.value OtlpProtocolGrpc ] 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+ readOtlpProtocol :: String -> Maybe OtlpProtocol+ readOtlpProtocol protocol =+ case map toLower protocol of+ "grpc" -> Just OtlpProtocolGrpc+ "http/protobuf" -> Just OtlpProtocolHttpProtobuf+ _ -> Nothing -otelcolSslKeyLogParser :: O.Parser G.SslKeyLog-otelcolSslKeyLogParser =- asum+otlpEndpointParser :: O.Parser String+otlpEndpointParser =+ O.strOption . mconcat $+ [ O.long "otlp-endpoint"+ , O.helpDoc . Just . OP.vcat . fmap OP.pretty $+ [ "The OTLP endpoint URL for all telemetry data, with an optionally-specified port number."+ , "Default value:"+ , " gRPC: http://localhost:4317"+ , " HTTP: http://localhost:4318"+ , "Example:"+ , " gRPC: https://my-api-endpoint:443"+ , " HTTP: http://my-api-endpoint/"+ ]+ ]++otlpGrpcCertificateStoreParser :: O.Parser FilePath+otlpGrpcCertificateStoreParser =+ O.strOption+ ( O.long "otlp-grpc-certificate-store"+ <> O.metavar "FILE"+ <> O.help "Store for certificate validation."+ )++otlpGrpcSslKeyLogParser :: O.Parser G.SslKeyLog+otlpGrpcSslKeyLogParser =+ O.asum [ G.SslKeyLogPath <$> O.strOption- ( O.long "otelcol-ssl-key-log"+ ( O.long "otlp-grpc-ssl-key-log" <> O.metavar "FILE" <> O.help "Use file to log SSL keys." )- , O.flag- G.SslKeyLogNone+ , O.flag' G.SslKeyLogFromEnv- ( O.long "otelcol-ssl-key-log-from-env"+ ( O.long "otlp-grpc-ssl-key-log-from-env" <> O.help "Use SSLKEYLOGFILE to log SSL keys." ) ] +otlpHttpHeadersParser :: O.Parser [(String, String)]+otlpHttpHeadersParser =+ O.option (O.maybeReader readHeaders) . mconcat $+ [ O.long "otlp-http-headers"+ , O.help "A list of headers to apply to all outgoing data."+ ]++readHeaders :: String -> Maybe [(String, String)]+readHeaders = runReadP pHeaders+ where+ pHeaders :: ReadP [(String, String)]+ pHeaders = P.many (pHeader <* (void (P.char ',') P.<++ P.eof))++ pHeader :: ReadP (String, String)+ pHeader = (,) <$> P.munch1 (/= '=') <*> P.munch1 (/= ',')+ -------------------------------------------------------------------------------- -- Debug Options @@ -222,3 +266,15 @@ MyDebugOptions <$> maybeMyEventlogSocketParser <*> maybeMyGhcDebugSocketParser++--------------------------------------------------------------------------------+-- Internal helpers+--------------------------------------------------------------------------------++{- |+Internal helper.++Run a ReadP parser.+-}+runReadP :: ReadP a -> String -> Maybe a+runReadP p = fmap fst . L.find (null . snd) . P.readP_to_S p
src/GHC/Eventlog/Live/Otelcol/Processor/Common/Core.hs view
@@ -6,9 +6,11 @@ -} module GHC.Eventlog.Live.Otelcol.Processor.Common.Core ( messageWith,+ (.~?), runIf, ifNonEmpty, toMaybeKeyValue,+ toMaybeAnyValue, ) where @@ -16,13 +18,17 @@ import Data.Machine (MachineT, stopped) import Data.ProtoLens (Message (..)) import GHC.Eventlog.Live.Data.Attribute (Attr, AttrValue (..))-import Lens.Family2 ((.~))+import Lens.Family2 (Setter, (.~)) import Proto.Opentelemetry.Proto.Common.V1.Common qualified as OC import Proto.Opentelemetry.Proto.Common.V1.Common_Fields qualified as OC -- | Construct a message with a list of modifications applied. messageWith :: (Message msg) => [msg -> msg] -> msg messageWith = foldr ($) defMessage++-- | Set a value if it is `Just`.+(.~?) :: Setter s s a a -> Maybe a -> s -> s+setter .~? maybeValue = maybe id (setter .~) maybeValue -- | Run a machine if a boolean is @True@, otherwise stop. runIf :: (Monad m) => Bool -> MachineT m k o -> MachineT m k o
src/GHC/Eventlog/Live/Otelcol/Processor/Common/Metrics.hs view
@@ -82,7 +82,7 @@ -} runMetricProcessor :: forall metricProcessor metricProcessorConfig m a b c d.- (Default metricProcessorConfig) =>+ (Default metricProcessorConfig, Show metricProcessorConfig) => MetricProcessor metricProcessor metricProcessorConfig m a b c d -> -- | The full configuration. FullConfig ->@@ -105,9 +105,13 @@ {-# INLINE runMetricProcessor #-} asMetricWith ::- (Default a, HasField "description" a (Maybe Text), HasField "name" a (Maybe Text)) =>+ ( Show metricProcessorConfig+ , Default metricProcessorConfig+ , HasField "description" metricProcessorConfig (Maybe Text)+ , HasField "name" metricProcessorConfig (Maybe Text)+ ) => FullConfig ->- (C.Metrics -> Maybe a) ->+ (C.Metrics -> Maybe metricProcessorConfig) -> [OM.Metric -> OM.Metric] -> Process OM.Metric'Data OM.Metric asMetricWith fullConfig field f =
src/GHC/Eventlog/Live/Otelcol/Processor/Common/ProfilesDictionary.hs view
@@ -15,14 +15,15 @@ SymbolIndex, getLocation, getFunction,+ getText, getString, getMapping, getLink, getAttribute,+ getAttr, getStack, -- * Convert data to the formats used by @hs-opentelemetry-otlp@.- toExportProfileServiceRequest, toProfilesDictionary, ) where@@ -31,24 +32,17 @@ import Control.Monad.Trans.State.Strict qualified as State import Data.ProtoLens (Message (..)) import Data.Text (Text)-import GHC.Eventlog.Live.Otelcol.Processor.Common.Core (messageWith)+import Data.Text qualified as T+import GHC.Eventlog.Live.Data.Attribute (Attr)+import GHC.Eventlog.Live.Otelcol.Processor.Common.Core (messageWith, toMaybeAnyValue) import GHC.Eventlog.Live.Otelcol.Processor.Common.SymbolTable (SymbolIndex, SymbolTable) import GHC.Eventlog.Live.Otelcol.Processor.Common.SymbolTable qualified as ST import GHC.Generics (Generic) import Lens.Family2 (Lens', (.~), (^.)) import Lens.Family2.Unchecked (lens)-import Proto.Opentelemetry.Proto.Collector.Profiles.V1development.ProfilesService qualified as OPS-import Proto.Opentelemetry.Proto.Collector.Profiles.V1development.ProfilesService_Fields qualified as OPS import Proto.Opentelemetry.Proto.Profiles.V1development.Profiles qualified as OP import Proto.Opentelemetry.Proto.Profiles.V1development.Profiles_Fields qualified as OP -toExportProfileServiceRequest :: OP.ProfilesData -> OPS.ExportProfilesServiceRequest-toExportProfileServiceRequest profilesData =- messageWith- [ OPS.resourceProfiles .~ profilesData ^. OPS.resourceProfiles- , OPS.dictionary .~ profilesData ^. OPS.dictionary- ]- data ProfilesDictionary = ProfilesDictionary { locationTable :: SymbolTable OP.Location {- ^ Common 'OP.Location' table, first entry is the 'defMessage'.@@ -145,9 +139,12 @@ getFunction :: (Monad m) => OP.Function -> StateT ProfilesDictionary m SymbolIndex getFunction = getSymbolIndexFor (lens (.functionTable) (\pd st -> pd{functionTable = st})) -getString :: (Monad m) => Text -> StateT ProfilesDictionary m SymbolIndex-getString = getSymbolIndexFor (lens (.stringTable) (\pd st -> pd{stringTable = st}))+getText :: (Monad m) => Text -> StateT ProfilesDictionary m SymbolIndex+getText = getSymbolIndexFor (lens (.stringTable) (\pd st -> pd{stringTable = st})) +getString :: (Monad m) => String -> StateT ProfilesDictionary m SymbolIndex+getString = getText . T.pack+ getMapping :: (Monad m) => OP.Mapping -> StateT ProfilesDictionary m SymbolIndex getMapping = getSymbolIndexFor (lens (.mappingTable) (\pd st -> pd{mappingTable = st})) @@ -156,6 +153,21 @@ getAttribute :: (Monad m) => OP.KeyValueAndUnit -> StateT ProfilesDictionary m SymbolIndex getAttribute = getSymbolIndexFor (lens (.attributeTable) (\pd st -> pd{attributeTable = st}))++getAttr :: (Monad m) => Attr -> StateT ProfilesDictionary m (Maybe SymbolIndex)+getAttr (key, value) =+ case toMaybeAnyValue value of+ Nothing ->+ pure Nothing+ Just anyValue -> do+ keyStrindex <- getText key+ let keyValueAndUnit :: OP.KeyValueAndUnit+ keyValueAndUnit =+ messageWith+ [ OP.keyStrindex .~ keyStrindex+ , OP.value .~ anyValue+ ]+ Just <$> getAttribute keyValueAndUnit getStack :: (Monad m) => OP.Stack -> StateT ProfilesDictionary m SymbolIndex getStack = getSymbolIndexFor (lens (.stackTable) (\pd st -> pd{stackTable = st}))
src/GHC/Eventlog/Live/Otelcol/Processor/Common/SymbolTable.hs view
@@ -28,7 +28,7 @@ {- ^ The next unused `SymbolIndex`. - > st.nextSymbolIndex == length st.entriesReversed+ > st.nextSymbolIndex == length st.entriesRev -} , entryToSymbolIndex :: !(Map a SymbolIndex) {- ^@@ -36,7 +36,7 @@ > toList st !! (st.entryToSymbolIndex Map.! a) == a -}- , entriesReversed :: ![a] -- reverse order of insertion into entryToSymbolIndex+ , entriesRev :: ![a] -- reverse order of insertion into entryToSymbolIndex {- ^ A list of entries in the `SymbolTable` in reverse order.@@ -49,14 +49,14 @@ SymbolTable { nextSymbolIndex = 0 , entryToSymbolIndex = Map.empty- , entriesReversed = []+ , entriesRev = [] } elemIndex :: (Ord a) => a -> SymbolTable a -> Maybe SymbolIndex elemIndex a st = Map.lookup a st.entryToSymbolIndex toList :: SymbolTable a -> [a]-toList st = reverse st.entriesReversed+toList st = reverse st.entriesRev fromList :: (Ord a) => [a] -> SymbolTable a fromList = foldr (\val st -> snd (insert val st)) empty@@ -65,8 +65,8 @@ insert a st = (si, st'') where ((si, isNew, st'), entryToSymbolIndex') = Map.alterF (updateEntry st) a st.entryToSymbolIndex- entriesReversed' = if isNew then a : st'.entriesReversed else st'.entriesReversed- st'' = st'{entryToSymbolIndex = entryToSymbolIndex', entriesReversed = entriesReversed'}+ entriesRev' = if isNew then a : st'.entriesRev else st'.entriesRev+ st'' = st'{entryToSymbolIndex = entryToSymbolIndex', entriesRev = entriesRev'} updateEntry :: SymbolTable a -> Maybe SymbolIndex -> ((SymbolIndex, Bool, SymbolTable a), Maybe SymbolIndex) updateEntry st Nothing = let (si, st') = freshSymbolIndex st in ((si, True, st'), Just si)
src/GHC/Eventlog/Live/Otelcol/Processor/Common/Traces.hs view
@@ -35,7 +35,6 @@ import Lens.Family2 ((.~)) import Proto.Opentelemetry.Proto.Collector.Trace.V1.TraceService qualified as OTS import Proto.Opentelemetry.Proto.Common.V1.Common qualified as OC-import Proto.Opentelemetry.Proto.Logs.V1.Logs_Fields qualified as OL import Proto.Opentelemetry.Proto.Resource.V1.Resource qualified as OR import Proto.Opentelemetry.Proto.Trace.V1.Trace qualified as OT import Proto.Opentelemetry.Proto.Trace.V1.Trace_Fields qualified as OT@@ -49,7 +48,7 @@ toResourceSpans :: OR.Resource -> [OT.ScopeSpans] -> Maybe OT.ResourceSpans toResourceSpans resource scopeSpans = ifNonEmpty scopeSpans $- messageWith [OL.resource .~ resource, OT.scopeSpans .~ scopeSpans]+ messageWith [OT.resource .~ resource, OT.scopeSpans .~ scopeSpans] toScopeSpans :: OC.InstrumentationScope -> [OT.Span] -> Maybe OT.ScopeSpans toScopeSpans instrumentationScope spans =
src/GHC/Eventlog/Live/Otelcol/Processor/Heap.hs view
@@ -26,6 +26,8 @@ import GHC.Eventlog.Live.Otelcol.Processor.Common.Core (runIf) import GHC.Eventlog.Live.Otelcol.Processor.Common.Metrics (MetricProcessor (..), asGauge, asSum, runMetricProcessor, viaLast, viaSum) import GHC.RTS.Events (Event (..), HeapProfBreakdown (..))+import IpeDB.Database qualified as DB+import IpeDB.Types.InfoProv qualified as IP import Lens.Family2 ((.~)) import Proto.Opentelemetry.Proto.Metrics.V1.Metrics qualified as OM import Proto.Opentelemetry.Proto.Metrics.V1.Metrics_Fields qualified as OM@@ -37,17 +39,18 @@ processHeapEvents :: (MonadIO m) => Logger m ->+ Maybe (DB.Table IP.InfoProvId IP.InfoProv) -> Maybe HeapProfBreakdown -> FullConfig -> ProcessT m (Tick (WithStartTime Event)) (Tick (DList OM.Metric))-processHeapEvents verbosity maybeHeapProfBreakdown fullConfig =+processHeapEvents verbosity maybeInfoProvTable maybeHeapProfBreakdown fullConfig = M.fanoutTick [ processHeapAllocated fullConfig , processBlocksSize fullConfig , processHeapSize fullConfig , processHeapLive fullConfig , processMemReturn fullConfig- , processHeapProfSample verbosity maybeHeapProfBreakdown fullConfig+ , processHeapProfSample verbosity maybeInfoProvTable maybeHeapProfBreakdown fullConfig ] --------------------------------------------------------------------------------@@ -170,14 +173,15 @@ processHeapProfSample :: (MonadIO m) => Logger m ->+ Maybe (DB.Table IP.InfoProvId IP.InfoProv) -> Maybe HeapProfBreakdown -> FullConfig -> ProcessT m (Tick (WithStartTime Event)) (Tick (DList OM.Metric))-processHeapProfSample logger maybeHeapProfBreakdown =+processHeapProfSample logger maybeInfoProvTable maybeHeapProfBreakdown = runMetricProcessor MetricProcessor { metricProcessorProxy = Proxy @"heapProfSample"- , dataProcessor = M.processHeapProfSampleData logger maybeHeapProfBreakdown+ , dataProcessor = M.processHeapProfSampleData logger maybeInfoProvTable maybeHeapProfBreakdown , aggregators = viaLast , postProcessor = mapping M.heapProfSamples ~> asParts , unit = "By"
src/GHC/Eventlog/Live/Otelcol/Processor/Logs.hs view
@@ -46,7 +46,7 @@ processUserMessage :: FullConfig -> Process (Tick (WithStartTime Event)) (Tick (DList OL.LogRecord)) processUserMessage fullConfig = runIf (C.processorEnabled (.logs) (.userMessage) fullConfig) $- M.liftTick M.processUserMessageData+ M.liftTick M.processStackFrame'Message ~> M.liftTick (mapping (D.singleton . toLogRecord)) ~> M.batchByTicks (C.processorExportBatches (.logs) (.userMessage) fullConfig)
src/GHC/Eventlog/Live/Otelcol/Processor/Profiles.hs view
@@ -7,20 +7,37 @@ Portability : portable -} module GHC.Eventlog.Live.Otelcol.Processor.Profiles (+ -- * Profile processing+ Sample (..),+ Stack (..), processProfileEvents,- processCallStackData,+ toProfiles,++ -- * Conversion to OTLP profiles+ toExportProfileServiceRequest,+ toProfilesData,+ toResourceProfiles,+ toScopeProfiles, ) where import Control.Monad.IO.Class (MonadIO (..))-import Control.Monad.Trans.State.Strict (State, runState)+import Control.Monad.Trans.State.Strict (State, StateT (..))+import Data.Bifunctor (Bifunctor (..)) import Data.DList (DList) import Data.DList qualified as D-import Data.Machine (ProcessT, asParts, mapping, (~>))+import Data.Functor.Identity (Identity (..))+import Data.Int (Int64)+import Data.Machine (ProcessT, mapping, (~>))+import Data.Maybe (catMaybes)+import Data.Proxy (Proxy (..)) import Data.Text (Text)-import GHC.Eventlog.Live.Data.Metric (Metric (..))+import Data.Text qualified as T+import Data.Vector (Vector)+import Data.Vector qualified as V+import Data.Word (Word32)+import GHC.Eventlog.Live.Data.Attribute (HasAttrs (..), (~=)) import GHC.Eventlog.Live.Logger (Logger)-import GHC.Eventlog.Live.Machine.Analysis.Heap qualified as M import GHC.Eventlog.Live.Machine.Analysis.Profile qualified as M import GHC.Eventlog.Live.Machine.Core (Tick) import GHC.Eventlog.Live.Machine.Core qualified as M@@ -30,266 +47,354 @@ import GHC.Eventlog.Live.Otelcol.Processor.Common.Core import GHC.Eventlog.Live.Otelcol.Processor.Common.ProfilesDictionary (ProfilesDictionary, SymbolIndex) import GHC.Eventlog.Live.Otelcol.Processor.Common.ProfilesDictionary qualified as PD-import GHC.RTS.Events (Event (..))-import GHC.Stack.Profiler.Core.SourceLocation qualified as Profiler-import Lens.Family2 ((.~))+import GHC.IsList (IsList (..))+import GHC.RTS.Events (Event (..), Timestamp)+import GHC.Records (HasField)+import IpeDB.Database qualified as DB+import IpeDB.Types.CostCentre qualified as CC+import IpeDB.Types.InfoProv qualified as IP+import IpeDB.Types.SrcLoc (Point (..), SrcLoc (..))+import Lens.Family2 ((.~), (^.))+import Proto.Opentelemetry.Proto.Collector.Profiles.V1development.ProfilesService qualified as OPS 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.Profiles.V1development.Profiles qualified as OP import Proto.Opentelemetry.Proto.Profiles.V1development.Profiles_Fields qualified as OP-import Proto.Opentelemetry.Proto.Resource.V1.Resource (Resource)+import Proto.Opentelemetry.Proto.Profiles.V1development.Profiles_Fields qualified as OPS+import Proto.Opentelemetry.Proto.Resource.V1.Resource qualified as OR ----------------------------------------------------------------------------------- processProfileEvents+-- Samples -------------------------------------------------------------------------------- +data Sample a = Sample+ { name :: !Text+ , stack :: !a+ }+ deriving stock (Show, Functor)++data Stack+ = CostCentreStack !M.CostCentreStack+ | CallStack !M.CallStack+ deriving (Show)+ processProfileEvents ::+ forall m. (MonadIO m) => Logger m ->+ DB.Table CC.CostCentreId CC.CostCentre ->+ DB.Table IP.InfoProvId IP.InfoProv -> FullConfig ->- ProcessT m (Tick (WithStartTime Event)) (Tick (DList M.CallStackData))-processProfileEvents verbosity config =+ ProcessT m (Tick (WithStartTime Event)) (Tick (DList (Sample Stack)))+processProfileEvents logger ccdb ipedb config = M.fanoutTick- [ processStackProfSample verbosity config- , processCostCentreProfSample verbosity config+ [ processProfSampleCostCentre logger ccdb config+ ~> mapping (fmap (fmap (fmap CostCentreStack)))+ , processGhcStackProfiler logger ipedb config+ ~> mapping (fmap (fmap (fmap CallStack))) ] ----------------------------------------------------------------------------------- StackProfSample+-- Processor for `ghc-stack-profiler` call-stack samples+-------------------------------------------------------------------------------- -processStackProfSample ::+processGhcStackProfiler ::+ forall m. (MonadIO m) => Logger m ->+ DB.Table IP.InfoProvId IP.InfoProv -> FullConfig ->- ProcessT m (Tick (WithStartTime Event)) (Tick (DList M.CallStackData))-processStackProfSample logger config =- runIf (C.processorEnabled (.profiles) (.stackSample) config) $+ ProcessT m (Tick (WithStartTime Event)) (Tick (DList (Sample M.CallStack)))+processGhcStackProfiler logger ipedb config =+ runIf (C.processorEnabled (.profiles) (.callStackProfile) config) $ M.liftTick- ( M.processStackProfSampleData logger- ~> mapping M.stackProfSamples- ~> asParts- ~> mapping (D.singleton . (.value))+ ( M.processGhcStackProfilerData logger ipedb+ ~> mapping (\stack -> D.singleton Sample{..}) )- ~> M.batchByTicks (C.processorExportBatches (.profiles) (.stackSample) config)+ ~> M.batchByTicks (C.processorExportBatches (.profiles) (.callStackProfile) config)+ where+ !name = C.processorName (.profiles) (.callStackProfile) config -processCostCentreProfSample ::+--------------------------------------------------------------------------------+-- Processor for cost-centre stack samples+--------------------------------------------------------------------------------++processProfSampleCostCentre ::+ forall m. (MonadIO m) => Logger m ->+ DB.Table CC.CostCentreId CC.CostCentre -> FullConfig ->- ProcessT m (Tick (WithStartTime Event)) (Tick (DList M.CallStackData))-processCostCentreProfSample logger config =- runIf (C.processorEnabled (.profiles) (.costCentreSample) config) $+ ProcessT m (Tick (WithStartTime Event)) (Tick (DList (Sample M.CostCentreStack)))+processProfSampleCostCentre logger ccdb config =+ runIf (C.processorEnabled (.profiles) (.costCentreStackProfile) config) $ M.liftTick- ( M.processCostCentreProfSampleData logger- ~> mapping M.stackProfSamples- ~> asParts- ~> mapping (D.singleton . (.value))+ ( M.processProfSampleCostCentreData logger ccdb+ ~> mapping (\stack -> D.singleton Sample{..}) )- ~> M.batchByTicks (C.processorExportBatches (.profiles) (.costCentreSample) config)--processCallStackData :: Resource -> OC.InstrumentationScope -> [M.CallStackData] -> (OP.ResourceProfiles, OP.ProfilesDictionary)-processCallStackData resource instrumentationScope callstacks = (resourceProfiles, profilesDictionary)+ ~> M.batchByTicks (C.processorExportBatches (.profiles) (.costCentreStackProfile) config) where- scopedProfiles =- messageWith- [ OP.profiles .~ [profile]- , OP.scope .~ instrumentationScope- ]-- resourceProfiles =- messageWith- [ OP.scopeProfiles .~ [scopedProfiles]- , OP.resource .~ resource- ]-- profilesDictionary = PD.toProfilesDictionary st+ !name = C.processorName (.profiles) (.costCentreStackProfile) config - (profile, st) = flip runState PD.empty $ do- sampleNameStrId <- PD.getString "__name__"- sampleTypeStrId <- PD.getString "String"- sampleAttrId <-- PD.getAttribute $- messageWith @OP.KeyValueAndUnit- [ OP.keyStrindex .~ sampleNameStrId- , OP.unitStrindex .~ sampleTypeStrId- , OP.value .~ messageWith [OC.stringValue .~ "process_cpu"]- ]+--------------------------------------------------------------------------------+-- Translation to OTLP profiles+-------------------------------------------------------------------------------- - samples <- traverse (asSample sampleAttrId) callstacks- cpuId <- PD.getString "stack"- unitId <- PD.getString "samples"- let sampleType :: OP.ValueType- sampleType =- messageWith- [ OP.typeStrindex .~ cpuId- , OP.unitStrindex .~ unitId- ]+toExportProfileServiceRequest :: OP.ProfilesData -> OPS.ExportProfilesServiceRequest+toExportProfileServiceRequest profilesData =+ messageWith+ [ OPS.resourceProfiles .~ profilesData ^. OPS.resourceProfiles+ , OPS.dictionary .~ profilesData ^. OPS.dictionary+ ] - pure $- messageWith- [ OP.samples .~ samples- , OP.sampleType .~ sampleType- ]+toProfilesData :: [OP.ResourceProfiles] -> OP.ProfilesDictionary -> Maybe OP.ProfilesData+toProfilesData resourceProfiles dictionary =+ ifNonEmpty resourceProfiles $+ messageWith [OP.resourceProfiles .~ resourceProfiles, OP.dictionary .~ dictionary] -asSample :: SymbolIndex -> M.CallStackData -> State ProfilesDictionary OP.Sample-asSample six stackData = do- locIndices <- traverse toIndex stackData.stack- s <-- PD.getStack $- messageWith- [ OP.locationIndices .~ locIndices- ]+toResourceProfiles :: OR.Resource -> [OP.ScopeProfiles] -> Maybe OP.ResourceProfiles+toResourceProfiles resource scopeProfiles =+ ifNonEmpty scopeProfiles $+ messageWith [OP.resource .~ resource, OP.scopeProfiles .~ scopeProfiles] - sampleThreadKeyStrId <- PD.getString "thread"- sampleCapKeyStrId <- PD.getString "capability"- sampleNumberUnitStrId <- PD.getString "Number"+toScopeProfiles :: OC.InstrumentationScope -> [OP.Profile] -> Maybe OP.ScopeProfiles+toScopeProfiles instrumentationScope profiles =+ ifNonEmpty profiles $+ messageWith [OP.scope .~ instrumentationScope, OP.profiles .~ profiles] - threadAttrId <-- PD.getAttribute $- messageWith @OP.KeyValueAndUnit- [ OP.keyStrindex .~ sampleThreadKeyStrId- , OP.unitStrindex .~ sampleNumberUnitStrId- , OP.value .~ messageWith [OC.intValue .~ maybe 0 (fromIntegral . (.value)) stackData.threadId]- ]+toProfiles :: [Sample Stack] -> Maybe ([OP.Profile], OP.ProfilesDictionary)+toProfiles samples = ifNonEmpty profiles profilesData+ where+ (costCentreStacks, callStacks) = partitionSamples samples - capAttrId <-- PD.getAttribute $- messageWith @OP.KeyValueAndUnit- [ OP.keyStrindex .~ sampleCapKeyStrId- , OP.unitStrindex .~ sampleNumberUnitStrId- , OP.value .~ messageWith [OC.intValue .~ fromIntegral stackData.capabilityId.value]- ]+ profilesData@(profiles, _) =+ second PD.toProfilesDictionary . runIdentity . flip runStateT PD.empty $ do+ -- Convert any cost-centre profiles.+ maybeCostCentreProfile <-+ sequence . ifNonEmpty costCentreStacks $+ getProfile costCentreStacks+ -- Convert any call-stack profiles.+ maybeCallStackProfiles <-+ sequence . ifNonEmpty callStacks $+ getProfile callStacks+ pure $ catMaybes [maybeCostCentreProfile, maybeCallStackProfiles] - pure $- messageWith- [ OP.values .~ [1]- , OP.stackIndex .~ s- , OP.attributeIndices- .~ [ six- , threadAttrId- , capAttrId- ]- ]+partitionSamples :: [Sample Stack] -> ([Sample M.CostCentreStack], [Sample M.CallStack])+partitionSamples = go ([], []) where- toIndex :: M.StackItemData -> State ProfilesDictionary SymbolIndex- toIndex = \case- M.IpeData infoProv -> getLocationIndexForInfoTable infoProv- M.UserMessageData message -> getLocationIndexForText message- M.SourceLocationData srcLoc -> getLocationIndexForSourceLocation srcLoc- M.CostCentreData costCentre -> getLocationIndexForCostCentre costCentre+ go :: ([Sample M.CostCentreStack], [Sample M.CallStack]) -> [Sample Stack] -> ([Sample M.CostCentreStack], [Sample M.CallStack])+ go (costCentreStackSamplesRev, callStackSamplesRev) = \case+ [] -> (reverse costCentreStackSamplesRev, reverse callStackSamplesRev)+ (Sample{stack = CostCentreStack costCentreStack, ..} : rest) -> go (Sample{stack = costCentreStack, ..} : costCentreStackSamplesRev, callStackSamplesRev) rest+ (Sample{stack = CallStack callStack, ..} : rest) -> go (costCentreStackSamplesRev, Sample{stack = callStack, ..} : callStackSamplesRev) rest -getLocationIndexForSourceLocation :: Profiler.SourceLocation -> State ProfilesDictionary SymbolIndex-getLocationIndexForSourceLocation srcLoc = do- functionNameId <- PD.getString $ Profiler.functionName srcLoc- fileNameId <- PD.getString $ Profiler.fileName srcLoc- funcIdx <-- PD.getFunction $- messageWith- [ OP.nameStrindex .~ functionNameId- , OP.systemNameStrindex .~ 0 -- 0 means unset- , OP.filenameStrindex .~ fileNameId- , OP.startLine .~ fromIntegral (Profiler.line srcLoc) -- TODO: better casts- ]+--------------------------------------------------------------------------------+-- Translating profiles to OTLP profiles - let line :: OP.Line- line =+{-# SPECIALIZE getProfile ::+ [Sample M.CallStack] -> State ProfilesDictionary OP.Profile+ #-}+{-# SPECIALIZE getProfile ::+ [Sample M.CostCentreStack] -> State ProfilesDictionary OP.Profile+ #-}+getProfile ::+ forall m a.+ (Monad m, ToSample a) =>+ [Sample a] -> StateT ProfilesDictionary m OP.Profile+getProfile xs = do+ samples <- traverse toSample xs+ typeStrindex <- PD.getText (getSampleType (Proxy @a))+ unitStrindex <- PD.getText (getSampleUnit (Proxy @a))+ let sampleType :: OP.ValueType+ sampleType = messageWith- [ OP.functionIndex .~ funcIdx- , OP.line .~ fromIntegral (Profiler.line srcLoc)- , OP.column .~ fromIntegral (Profiler.column srcLoc)+ [ OP.typeStrindex .~ typeStrindex+ , OP.unitStrindex .~ unitStrindex ]+ let profile :: OP.Profile+ profile =+ messageWith+ [ OP.samples .~ samples+ , OP.sampleType .~ sampleType+ ]+ pure profile - PD.getLocation $- messageWith- [ OP.lines .~ [line]- , OP.mappingIndex .~ 0 -- 0 means unset- ]+--------------------------------------------------------------------------------+-- Translating samples to OTLP samples -getLocationIndexForText :: Text -> State ProfilesDictionary SymbolIndex-getLocationIndexForText msg = do- textId <- PD.getString msg- funcIdx <-- PD.getFunction $- messageWith- [ OP.nameStrindex .~ textId- , OP.systemNameStrindex .~ 0 -- 0 means unset- , OP.filenameStrindex .~ 0 -- 0 means unset- , OP.startLine .~ 0 -- 0 means unset- ]+type IsSample a = (IsStack a, HasAttrs a, HasField "maybeTimeUnixNano" a (Maybe Timestamp)) - let line :: OP.Line- line =+class (IsSample a, ToLocation (StackFrame a)) => ToSample a where+ getSampleType :: Proxy a -> Text+ getSampleUnit :: Proxy a -> Text++ getSampleValue :: a -> Int64+ getSampleValue _x = 1+ {-# INLINE getSampleValue #-}++{-# SPECIALIZE toSample ::+ Sample M.CallStack -> State ProfilesDictionary OP.Sample+ #-}+{-# SPECIALIZE toSample ::+ Sample M.CostCentreStack -> State ProfilesDictionary OP.Sample+ #-}+toSample ::+ forall m a.+ (Monad m, ToSample a) =>+ Sample a -> StateT ProfilesDictionary m OP.Sample+toSample x = do+ stackIndex <- PD.getStack =<< toStack x.stack+ let attrs = "__name__" ~= x.name : toList (getAttrs x.stack)+ attributeIndices <- catMaybes <$> traverse PD.getAttr attrs+ let sample :: OP.Sample+ sample = messageWith- [ OP.functionIndex .~ funcIdx- , OP.line .~ 0 -- 0 means unset- , OP.column .~ 0 -- 0 means unset+ [ OP.values .~ [getSampleValue x.stack]+ , OP.stackIndex .~ stackIndex+ , OP.attributeIndices .~ attributeIndices+ , OP.timestampsUnixNano .~? sequence [x.stack.maybeTimeUnixNano] ]+ pure sample - PD.getLocation $- messageWith- [ OP.lines .~ [line]- ]+instance ToSample M.CallStack where+ getSampleType :: Proxy M.CallStack -> Text+ getSampleType _proxy = "cpu"+ {-# INLINE getSampleType #-} -getLocationIndexForInfoTable :: M.InfoProv -> State ProfilesDictionary SymbolIndex-getLocationIndexForInfoTable infoProv = do- ipNameId <- PD.getString infoProv.ipName- let label =- if (infoProv.ipLabel) == ""- then infoProv.ipModule <> ":" <> infoProv.ipName- else infoProv.ipModule <> ":" <> infoProv.ipLabel- infoProvFuncNameId <- PD.getString label- -- tyDesc <- getText infoProv.infoProvTyDesc- --- ipSrcLocId <- PD.getString infoProv.ipSrcLoc- funcIdx <-- PD.getFunction $- messageWith- [ OP.nameStrindex .~ infoProvFuncNameId- , OP.systemNameStrindex .~ ipNameId- , OP.filenameStrindex .~ ipSrcLocId -- 0 means unset- , OP.startLine .~ 0 -- 0 means unset- ]+ getSampleUnit :: Proxy M.CallStack -> Text+ getSampleUnit _proxy = "samples"+ {-# INLINE getSampleUnit #-} - let line :: OP.Line- line =- messageWith- [ OP.functionIndex .~ funcIdx- , OP.line .~ 0 -- 0 means unset- , OP.column .~ 0 -- 0 means unset- ]+instance ToSample M.CostCentreStack where+ getSampleType :: Proxy a -> Text+ getSampleType _proxy = "cpu"+ {-# INLINE getSampleType #-} - PD.getLocation $- messageWith- [ OP.lines .~ [line]- , OP.address .~ 0 -- 0 means unset- ]+ getSampleUnit :: Proxy a -> Text+ getSampleUnit _proxy = "samples"+ {-# INLINE getSampleUnit #-} -getLocationIndexForCostCentre :: M.CostCentre -> State ProfilesDictionary SymbolIndex-getLocationIndexForCostCentre costCentre = do- let label = costCentre.costCentreModule <> ":" <> costCentre.costCentreLabel- costCentreFuncNameId <- PD.getString label- -- tyDesc <- getText infoProv.infoProvTyDesc- --- costCentreSrcLocId <- PD.getString costCentre.costCentreSrcLoc- funcIdx <-- PD.getFunction $- messageWith- [ OP.nameStrindex .~ costCentreFuncNameId- , OP.systemNameStrindex .~ costCentreFuncNameId- , OP.filenameStrindex .~ costCentreSrcLocId -- 0 means unset- , OP.startLine .~ 0 -- 0 means unset- ]+--------------------------------------------------------------------------------+-- Translating stacks to OTLP stacks - let line :: OP.Line- line =+class IsStack a where+ type StackFrame a+ getStackFrames :: a -> Vector (StackFrame a)++instance IsStack M.CallStack where+ type StackFrame M.CallStack = M.CallStackFrame+ getStackFrames :: M.CallStack -> Vector (StackFrame M.CallStack)+ getStackFrames = (.callStack)+ {-# INLINE getStackFrames #-}++instance IsStack M.CostCentreStack where+ type StackFrame M.CostCentreStack = M.CostCentreStackFrame+ getStackFrames :: M.CostCentreStack -> Vector (StackFrame M.CostCentreStack)+ getStackFrames = (.costCentreStack)+ {-# INLINE getStackFrames #-}++{-# SPECIALIZE toStack ::+ M.CallStack -> State ProfilesDictionary OP.Stack+ #-}+{-# SPECIALIZE toStack ::+ M.CostCentreStack -> State ProfilesDictionary OP.Stack+ #-}+toStack ::+ forall m a.+ (Monad m, IsStack a, ToLocation (StackFrame a)) =>+ a -> StateT ProfilesDictionary m OP.Stack+toStack x = do+ locationIndices <- traverse toLocation (getStackFrames x)+ let stack :: OP.Stack+ stack = messageWith- [ OP.functionIndex .~ funcIdx- , OP.line .~ 0 -- 0 means unset- , OP.column .~ 0 -- 0 means unset+ [ OP.vec'locationIndices .~ V.convert locationIndices ]+ pure stack - PD.getLocation $- messageWith- [ OP.lines .~ [line]- , OP.address .~ 0 -- 0 means unset- ]+--------------------------------------------------------------------------------+-- Translating stack frames to OTLP Locations++class ToLocation a where+ toLocation ::+ (Monad m) =>+ a -> StateT ProfilesDictionary m SymbolIndex++instance ToLocation M.CallStackFrame where+ toLocation ::+ (Monad m) =>+ M.CallStackFrame -> StateT ProfilesDictionary m SymbolIndex+ toLocation = \case+ M.CallStackFrame _infoProvId (Just infoProv)+ -- If there's a non-empty ipLabel, use it.+ | not (T.null infoProv.ipLabel) ->+ toLocation (infoProv.ipModule <> ":" <> infoProv.ipLabel, infoProv.ipSrcLoc)+ -- If there's a non-empty ipName, use it.+ | not (T.null infoProv.ipName) ->+ toLocation (infoProv.ipModule <> ":" <> infoProv.ipName, infoProv.ipSrcLoc)+ -- Otherwise, there's no helpful location information.+ M.CallStackFrame infoProvId maybeInfoProv ->+ let name = T.pack (show infoProvId)+ srcLoc = maybe UnhelpfulSrcLoc (.ipSrcLoc) maybeInfoProv+ in toLocation (name, srcLoc)+ M.CallStackMessage name srcLoc -> toLocation (name, srcLoc)+ {-# INLINE toLocation #-}++instance ToLocation M.CostCentreStackFrame where+ toLocation ::+ (Monad m) =>+ M.CostCentreStackFrame -> StateT ProfilesDictionary m SymbolIndex+ toLocation = \case+ M.CostCentreStackFrame _costCentreId (Just costCentre)+ -- If there's a non-empty ccLabel, use it.+ | not (T.null costCentre.ccLabel) ->+ toLocation (costCentre.ccModule <> ":" <> costCentre.ccLabel, costCentre.ccSrcLoc)+ -- Otherwise, there's no helpful location information.+ M.CostCentreStackFrame costCentreId maybeCostCentre ->+ let name = T.pack (show costCentreId)+ srcLoc = maybe UnhelpfulSrcLoc (.ccSrcLoc) maybeCostCentre+ in toLocation (name, srcLoc)+ {-# INLINE toLocation #-}++instance ToLocation (Text, SrcLoc) where+ toLocation ::+ (Monad m) =>+ (Text, SrcLoc) -> StateT ProfilesDictionary m SymbolIndex+ toLocation (name, srcLoc) = do+ -- Encode the filename.+ filenameStrindex <-+ if null srcLoc.srcFilePath+ then pure 0+ else PD.getString srcLoc.srcFilePath++ -- Encode the function name.+ nameStrindex <- PD.getText name++ -- Encode the start point.+ let !maybeStart = (.start) <$> srcLoc.srcRange+ let !maybeStartLine = fromIntegral @Word32 @Int64 . (.line) <$> maybeStart+ let !maybeStartColumn = fromIntegral @Word32 @Int64 . (.column) <$> maybeStart++ -- Encode the function metadata.+ let function :: OP.Function+ function =+ messageWith+ [ OP.nameStrindex .~ nameStrindex+ , OP.filenameStrindex .~ filenameStrindex+ , OP.startLine .~? maybeStartLine+ ]+ functionIndex <- PD.getFunction function++ -- Encode the location metadata.+ let line :: OP.Line+ line =+ messageWith+ [ OP.functionIndex .~ functionIndex+ , OP.line .~? maybeStartLine+ , OP.column .~? maybeStartColumn+ ]+ let location :: OP.Location+ location =+ messageWith+ [ OP.lines .~ [line]+ ]+ PD.getLocation location
− src/GHC/Eventlog/Socket/Compat.hs
@@ -1,82 +0,0 @@-{-# LANGUAGE CPP #-}-{-# LANGUAGE OverloadedStrings #-}--{- |-Module : GHC.Eventlog.Socket.Compat-Description : The implementation of @eventlog-live-otelcol@.-Stability : experimental-Portability : portable--}-module GHC.Eventlog.Socket.Compat (- MyEventlogSocket (..),- maybeMyEventlogSocketParser,- startMyEventlogSocket,-) where--import Control.Applicative (asum)-import GHC.Eventlog.Live.Logger (Logger)-import Options.Applicative qualified as O-import Options.Applicative.Extra.Feature (Feature (..))-import Options.Applicative.Extra.Feature qualified as OF--#ifdef EVENTLOG_LIVE_OTELCOL_USE_EVENTLOG_SOCKET-import Data.Foldable (for_)-import Data.Text qualified as T-import GHC.Eventlog.Live.Data.Severity (Severity (..))-import GHC.Eventlog.Live.Logger (writeLog)-import GHC.Eventlog.Socket qualified as Eventlog.Socket-#else-import Control.Monad (when)-import Data.Maybe (isJust)-#endif------------------------------------------------------------------------------------- Feature: use-eventlog-socket-----------------------------------------------------------------------------------useEventlogSocket :: Feature-useEventlogSocket = Feature{flag = "use-eventlog-socket", isOn = isOn, info = "Cannot open eventlog socket."}- where- isOn :: Bool-#ifdef EVENTLOG_LIVE_OTELCOL_USE_EVENTLOG_SOCKET- isOn = True-#else- isOn = False-#endif------------------------------------------------------------------------------------- My Eventlog Socket-----------------------------------------------------------------------------------newtype MyEventlogSocket- = MyEventlogSocketUnix FilePath--maybeMyEventlogSocketParser :: O.Parser (Maybe MyEventlogSocket)-maybeMyEventlogSocketParser =- asum $- [ myEventlogSocketUnixParser- , pure Nothing- ]--myEventlogSocketUnixParser :: O.Parser (Maybe MyEventlogSocket)-myEventlogSocketUnixParser =- OF.onlyFor useEventlogSocket (O.option (Just . MyEventlogSocketUnix <$> O.str)) (O.metavar "FILE") $- O.long "my-eventlog-socket-unix"- <> OF.helpFor useEventlogSocket "Open an eventlog socket for this program on the given Unix socket."--{- |-Set @eventlog-socket@ as the eventlog writer.--}-startMyEventlogSocket :: Logger IO -> Maybe MyEventlogSocket -> IO ()-#ifdef EVENTLOG_LIVE_OTELCOL_USE_EVENTLOG_SOCKET-startMyEventlogSocket logger maybeMyEventlogSocket =- for_ maybeMyEventlogSocket $ \case- MyEventlogSocketUnix myEventlogSocketUnix -> do- writeLog logger INFO $- "Start eventlog-socket with Unix domain socket at " <> T.pack myEventlogSocketUnix <> "."- Eventlog.Socket.startWait myEventlogSocketUnix-#else-startMyEventlogSocket logger maybeMyEventlogSocket =- when (isJust maybeMyEventlogSocket) $- OF.exitIfUnsupported useEventlogSocket logger-#endif
− src/Language/Haskell/TH/Lift/Compat.hs
@@ -1,13 +0,0 @@-{-# LANGUAGE CPP #-}--module Language.Haskell.TH.Lift.Compat (- Exp,- Lift (..),- Q,-) where--#if defined(EVENTLOG_LIVE_OTELCOL_USE_TEMPLATE_HASKELL_LIFT)-import Language.Haskell.TH.Lift (Exp, Lift (..), Q)-#else-import Language.Haskell.TH.Syntax (Exp, Lift (..), Q)-#endif
− src/Options/Applicative/Compat.hs
@@ -1,34 +0,0 @@-{-# LANGUAGE CPP #-}--module Options.Applicative.Compat (- parserOptionGroup,- simpleVersioner,-) where--#if MIN_VERSION_optparse_applicative(0,19,0)-import Options.Applicative (parserOptionGroup)-import Options.Applicative (simpleVersioner)-#else-#if MIN_VERSION_optparse_applicative(0,18,1)-import Options.Applicative (simpleVersioner)-#else-import Options.Applicative (infoOption, long, help, hidden)-#endif-import Options.Applicative (Parser)-#endif--#if MIN_VERSION_optparse_applicative(0,19,0)-#else--- Prior to optparse-applicative-0.19.0.0, option groups were not supported,--- so this definition simply drops the group.-parserOptionGroup :: String -> Parser a -> Parser a-parserOptionGroup _ p = p-#if MIN_VERSION_optparse_applicative(0,18,1)-#else--- Prior to optparse-applicative-0.18.1.0, simpleVersioner was not defined,--- so this definition is taken verbatim from optparse-applicative-0.18.1.0.-simpleVersioner :: String -> Parser (a -> a)-simpleVersioner version = infoOption version $- mconcat [long "version", help "Show version information", hidden]-#endif-#endif
− src/Options/Applicative/Extra/Feature.hs
@@ -1,62 +0,0 @@-{-# LANGUAGE OverloadedStrings #-}--module Options.Applicative.Extra.Feature (- Feature (..),- helpFor,- onlyFor,- exitIfUnsupported,-) where--import Control.Monad (unless)-import Data.Default (Default (..))-import Data.Text (Text)-import Data.Text qualified as T-import GHC.Eventlog.Live.Data.Severity (Severity (..))-import GHC.Eventlog.Live.Logger (Logger, writeLog)-import Options.Applicative qualified as O-import Options.Applicative.Help.Pretty qualified as OP-import System.Exit (exitFailure)--data Feature = Feature- { flag :: !String- , isOn :: !Bool- , info :: !String- }--{- |-Create a command-line help document for an option that depend on a feature flag.--}-helpFor :: Feature -> String -> O.Mod f a-helpFor feature help- | feature.isOn = O.help help- | otherwise = O.helpDoc (Just $ OP.vcat [OP.pretty unsupported, OP.pretty help]) <> O.hidden- where- unsupported :: Text- unsupported = T.pack "Unsupported. Requires build with -f+" <> T.pack feature.flag <> "."--{- |-Create a command-line option that depends on a feature flag.--}-onlyFor ::- forall f a.- (O.HasName f, Default a) =>- Feature ->- (O.Mod f a -> O.Parser a) ->- O.Mod f a ->- (forall g x. (O.HasName g) => O.Mod g x) ->- O.Parser a-onlyFor feature opt optIfSupported optAlways- | feature.isOn = opt (optAlways <> optIfSupported)- | otherwise = def <$ O.infoOption unsupportedInfo optAlways- where- unsupportedInfo :: String- unsupportedInfo = feature.info <> " Requires build with -f+" <> feature.flag <> "."--{- |-Exit with the feature info.--}-exitIfUnsupported :: Feature -> Logger IO -> IO ()-exitIfUnsupported feature logger =- unless feature.isOn $ do- writeLog logger FATAL (T.pack feature.info)- exitFailure
− src/System/Random/Compat.hs
@@ -1,17 +0,0 @@-{-# LANGUAGE CPP #-}--module System.Random.Compat (- uniformByteString,-) where--#if MIN_VERSION_random(1,3,0)-import System.Random (uniformByteString)-#else-import Data.Bifunctor (Bifunctor (first))-import Data.ByteString (ByteString)-import Data.ByteString.Short (fromShort)-import System.Random (RandomGen (genShortByteString))--uniformByteString :: RandomGen g => Int -> g -> (ByteString, g)-uniformByteString n g = first fromShort (genShortByteString n g)-#endif