diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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. diff --git a/data/config.schema.json b/data/config.schema.json --- a/data/config.schema.json +++ b/data/config.schema.json @@ -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 } diff --git a/data/default.yaml b/data/default.yaml --- a/data/default.yaml +++ b/data/default.yaml @@ -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 diff --git a/eventlog-live-otelcol.cabal b/eventlog-live-otelcol.cabal --- a/eventlog-live-otelcol.cabal +++ b/eventlog-live-otelcol.cabal @@ -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, diff --git a/src-internal/GHC/Debug/Stub/Compat.hs b/src-internal/GHC/Debug/Stub/Compat.hs new file mode 100644 --- /dev/null +++ b/src-internal/GHC/Debug/Stub/Compat.hs @@ -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'." diff --git a/src-internal/GHC/Eventlog/Socket/Compat.hs b/src-internal/GHC/Eventlog/Socket/Compat.hs new file mode 100644 --- /dev/null +++ b/src-internal/GHC/Eventlog/Socket/Compat.hs @@ -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 diff --git a/src-internal/Language/Haskell/TH/Lift/Compat.hs b/src-internal/Language/Haskell/TH/Lift/Compat.hs new file mode 100644 --- /dev/null +++ b/src-internal/Language/Haskell/TH/Lift/Compat.hs @@ -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 diff --git a/src-internal/Options/Applicative/Compat.hs b/src-internal/Options/Applicative/Compat.hs new file mode 100644 --- /dev/null +++ b/src-internal/Options/Applicative/Compat.hs @@ -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 diff --git a/src-internal/Options/Applicative/Extra/Feature.hs b/src-internal/Options/Applicative/Extra/Feature.hs new file mode 100644 --- /dev/null +++ b/src-internal/Options/Applicative/Extra/Feature.hs @@ -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 diff --git a/src-internal/System/Random/Compat.hs b/src-internal/System/Random/Compat.hs new file mode 100644 --- /dev/null +++ b/src-internal/System/Random/Compat.hs @@ -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 diff --git a/src/GHC/Debug/Stub/Compat.hs b/src/GHC/Debug/Stub/Compat.hs deleted file mode 100644 --- a/src/GHC/Debug/Stub/Compat.hs +++ /dev/null @@ -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'." diff --git a/src/GHC/Eventlog/Live/Otelcol.hs b/src/GHC/Eventlog/Live/Otelcol.hs --- a/src/GHC/Eventlog/Live/Otelcol.hs +++ b/src/GHC/Eventlog/Live/Otelcol.hs @@ -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. diff --git a/src/GHC/Eventlog/Live/Otelcol/Config.hs b/src/GHC/Eventlog/Live/Otelcol/Config.hs --- a/src/GHC/Eventlog/Live/Otelcol/Config.hs +++ b/src/GHC/Eventlog/Live/Otelcol/Config.hs @@ -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 ] ------------------------------------------------------------------------------- diff --git a/src/GHC/Eventlog/Live/Otelcol/Config/Types.hs b/src/GHC/Eventlog/Live/Otelcol/Config/Types.hs --- a/src/GHC/Eventlog/Live/Otelcol/Config/Types.hs +++ b/src/GHC/Eventlog/Live/Otelcol/Config/Types.hs @@ -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. diff --git a/src/GHC/Eventlog/Live/Otelcol/Exporter/Core.hs b/src/GHC/Eventlog/Live/Otelcol/Exporter/Core.hs new file mode 100644 --- /dev/null +++ b/src/GHC/Eventlog/Live/Otelcol/Exporter/Core.hs @@ -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 diff --git a/src/GHC/Eventlog/Live/Otelcol/Exporter/Logs.hs b/src/GHC/Eventlog/Live/Otelcol/Exporter/Logs.hs --- a/src/GHC/Eventlog/Live/Otelcol/Exporter/Logs.hs +++ b/src/GHC/Eventlog/Live/Otelcol/Exporter/Logs.hs @@ -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 diff --git a/src/GHC/Eventlog/Live/Otelcol/Exporter/Metrics.hs b/src/GHC/Eventlog/Live/Otelcol/Exporter/Metrics.hs --- a/src/GHC/Eventlog/Live/Otelcol/Exporter/Metrics.hs +++ b/src/GHC/Eventlog/Live/Otelcol/Exporter/Metrics.hs @@ -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 diff --git a/src/GHC/Eventlog/Live/Otelcol/Exporter/Profiles.hs b/src/GHC/Eventlog/Live/Otelcol/Exporter/Profiles.hs --- a/src/GHC/Eventlog/Live/Otelcol/Exporter/Profiles.hs +++ b/src/GHC/Eventlog/Live/Otelcol/Exporter/Profiles.hs @@ -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. diff --git a/src/GHC/Eventlog/Live/Otelcol/Exporter/Traces.hs b/src/GHC/Eventlog/Live/Otelcol/Exporter/Traces.hs --- a/src/GHC/Eventlog/Live/Otelcol/Exporter/Traces.hs +++ b/src/GHC/Eventlog/Live/Otelcol/Exporter/Traces.hs @@ -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 diff --git a/src/GHC/Eventlog/Live/Otelcol/Options.hs b/src/GHC/Eventlog/Live/Otelcol/Options.hs --- a/src/GHC/Eventlog/Live/Otelcol/Options.hs +++ b/src/GHC/Eventlog/Live/Otelcol/Options.hs @@ -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 diff --git a/src/GHC/Eventlog/Live/Otelcol/Processor/Common/Core.hs b/src/GHC/Eventlog/Live/Otelcol/Processor/Common/Core.hs --- a/src/GHC/Eventlog/Live/Otelcol/Processor/Common/Core.hs +++ b/src/GHC/Eventlog/Live/Otelcol/Processor/Common/Core.hs @@ -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 diff --git a/src/GHC/Eventlog/Live/Otelcol/Processor/Common/Metrics.hs b/src/GHC/Eventlog/Live/Otelcol/Processor/Common/Metrics.hs --- a/src/GHC/Eventlog/Live/Otelcol/Processor/Common/Metrics.hs +++ b/src/GHC/Eventlog/Live/Otelcol/Processor/Common/Metrics.hs @@ -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 = diff --git a/src/GHC/Eventlog/Live/Otelcol/Processor/Common/ProfilesDictionary.hs b/src/GHC/Eventlog/Live/Otelcol/Processor/Common/ProfilesDictionary.hs --- a/src/GHC/Eventlog/Live/Otelcol/Processor/Common/ProfilesDictionary.hs +++ b/src/GHC/Eventlog/Live/Otelcol/Processor/Common/ProfilesDictionary.hs @@ -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})) diff --git a/src/GHC/Eventlog/Live/Otelcol/Processor/Common/SymbolTable.hs b/src/GHC/Eventlog/Live/Otelcol/Processor/Common/SymbolTable.hs --- a/src/GHC/Eventlog/Live/Otelcol/Processor/Common/SymbolTable.hs +++ b/src/GHC/Eventlog/Live/Otelcol/Processor/Common/SymbolTable.hs @@ -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) diff --git a/src/GHC/Eventlog/Live/Otelcol/Processor/Common/Traces.hs b/src/GHC/Eventlog/Live/Otelcol/Processor/Common/Traces.hs --- a/src/GHC/Eventlog/Live/Otelcol/Processor/Common/Traces.hs +++ b/src/GHC/Eventlog/Live/Otelcol/Processor/Common/Traces.hs @@ -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 = diff --git a/src/GHC/Eventlog/Live/Otelcol/Processor/Heap.hs b/src/GHC/Eventlog/Live/Otelcol/Processor/Heap.hs --- a/src/GHC/Eventlog/Live/Otelcol/Processor/Heap.hs +++ b/src/GHC/Eventlog/Live/Otelcol/Processor/Heap.hs @@ -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" diff --git a/src/GHC/Eventlog/Live/Otelcol/Processor/Logs.hs b/src/GHC/Eventlog/Live/Otelcol/Processor/Logs.hs --- a/src/GHC/Eventlog/Live/Otelcol/Processor/Logs.hs +++ b/src/GHC/Eventlog/Live/Otelcol/Processor/Logs.hs @@ -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) diff --git a/src/GHC/Eventlog/Live/Otelcol/Processor/Profiles.hs b/src/GHC/Eventlog/Live/Otelcol/Processor/Profiles.hs --- a/src/GHC/Eventlog/Live/Otelcol/Processor/Profiles.hs +++ b/src/GHC/Eventlog/Live/Otelcol/Processor/Profiles.hs @@ -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 diff --git a/src/GHC/Eventlog/Socket/Compat.hs b/src/GHC/Eventlog/Socket/Compat.hs deleted file mode 100644 --- a/src/GHC/Eventlog/Socket/Compat.hs +++ /dev/null @@ -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 diff --git a/src/Language/Haskell/TH/Lift/Compat.hs b/src/Language/Haskell/TH/Lift/Compat.hs deleted file mode 100644 --- a/src/Language/Haskell/TH/Lift/Compat.hs +++ /dev/null @@ -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 diff --git a/src/Options/Applicative/Compat.hs b/src/Options/Applicative/Compat.hs deleted file mode 100644 --- a/src/Options/Applicative/Compat.hs +++ /dev/null @@ -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 diff --git a/src/Options/Applicative/Extra/Feature.hs b/src/Options/Applicative/Extra/Feature.hs deleted file mode 100644 --- a/src/Options/Applicative/Extra/Feature.hs +++ /dev/null @@ -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 diff --git a/src/System/Random/Compat.hs b/src/System/Random/Compat.hs deleted file mode 100644 --- a/src/System/Random/Compat.hs +++ /dev/null @@ -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