packages feed

eventlog-live-otelcol 0.5.0.0 → 0.6.0.0

raw patch · 29 files changed

+5860/−2018 lines, 29 filesdep +HsYAMLdep +binarydep +containersdep −yamldep ~eventlog-livedep ~eventlog-socketdep ~ghc-debug-stub

Dependencies added: HsYAML, binary, containers, eventlog-socket-control, fast-logger, ghc-stack-profiler-core, http-api-data, network, servant, servant-server, stm, transformers, wai, wai-cors, wai-extra, warp

Dependencies removed: yaml

Dependency ranges changed: eventlog-live, eventlog-socket, ghc-debug-stub, hs-opentelemetry-otlp, template-haskell

Files

CHANGELOG.md view
@@ -1,3 +1,14 @@+### 0.6.0.0++- **BREAKING**: Drop `enabled` from configuration files.+- **BREAKING**: Change value of `aggregate` in configuration files to ``(boolean | `${number}x`)``.+- Support logs and markers (`thread_label`, `user_marker`, `user_message`, `internal_log_message`).+- Support internal telemetry (currently only `internal_log_message`).+- Support cost-centre profiles.+- Support `ghc-stack-profiler` call-stack profiles.+- **BREAKING**: Use OTLP v1.9.+- Support an optional control server via a REST API.+ ### 0.5.0.0  - **BREAKING**: Put `ghc-debug-stub` support behind `use-ghc-debug-stub` feature flag.
+ data/config.schema.json view
@@ -0,0 +1,169 @@+{+  "$schema": "https://json-schema.org/draft-07/schema",+  "title": "Config",+  "description": "Configuration for the eventlog-live-otelcol program",+  "type": "object",+  "properties": {+    "processors": {+      "type": "object",+      "properties": {+        "defaults": { "#ref": "#/definitions/processor_defaults" },+        "logs": {+          "type": "object",+          "defaults": { "#ref": "#/definitions/log_defaults" },+          "thread_label": { "#ref": "#/definitions/log" },+          "user_marker": { "#ref": "#/definitions/log" },+          "user_message": { "#ref": "#/definitions/log" },+          "internal_log_message": { "#ref": "#/definitions/log" }+        },+        "metrics": {+          "type": "object",+          "properties": {+            "defaults": { "#ref": "#/definitions/metric_defaults" },+            "blocks_size": { "$ref": "#/definitions/metric" },+            "capability_usage": { "$ref": "#/definitions/metric" },+            "heap_allocated": { "$ref": "#/definitions/metric" },+            "heap_live": { "$ref": "#/definitions/metric" },+            "heap_prof_sample": { "$ref": "#/definitions/metric" },+            "heap_size": { "$ref": "#/definitions/metric" },+            "mem_current": { "$ref": "#/definitions/metric" },+            "mem_needed": { "$ref": "#/definitions/metric" },+            "mem_returned": { "$ref": "#/definitions/metric" }+          },+          "additionalProperties": false+        },+        "traces": {+          "type": "object",+          "properties": {+            "defaults": { "#ref": "#/definitions/trace_defaults" },+            "capability_usage": { "$ref": "#/definitions/trace" },+            "thread_state": { "$ref": "#/definitions/trace" }+          },+          "additionalProperties": false+        },+        "profiles": {+          "type": "object",+          "properties": {+            "defaults": { "#ref": "#/definitions/profile_defaults" },+            "stack_sample": { "$ref": "#/definitions/profile" },+            "cost_centre_sample": { "$ref": "#/definitions/profile" }+          },+          "additionalProperties": false+        }+      }+    }+  },+  "definitions": {+    "duration": {},+    "duration_by_batches": {+      "type": "string",+      "pattern": "^([0-9]+)(x)$"+    },+    "duration_by_seconds": {+      "type": "string",+      "pattern": "^([0-9]+(\\.[0-9]+)?)(s)$"+    },+    "aggregation_strategy": {+      "oneOf": [+        { "type": "null" },+        { "$ref": "#/definitions/aggregation_strategy_bool" },+        { "$ref": "#/definitions/aggregation_strategy_duration" }+      ]+    },+    "aggregation_strategy_bool": {+      "type": "boolean"+    },+    "aggregation_strategy_duration": {+      "oneOf": [+        { "$ref": "#/definitions/duration_by_batches" },+        { "$ref": "#/definitions/duration_by_seconds" }+      ]+    },+    "export_strategy": {+      "oneOf": [+        { "type": "null" },+        { "$ref": "#/definitions/export_strategy_bool" },+        { "$ref": "#/definitions/export_strategy_duration" }+      ]+    },+    "export_strategy_bool": {+      "type": "boolean"+    },+    "export_strategy_duration": {+      "oneOf": [+        { "$ref": "#/definitions/duration_by_batches" },+        { "$ref": "#/definitions/duration_by_seconds" }+      ]+    },+    "processor_defaults": {+      "description": "An optional 'defaults' block for defining YAML anchors.",+      "properties": {+        "export": { "$ref": "#/definitions/export_strategy" },+        "aggregate": { "$ref": "#/definitions/aggregation_strategy" }+      },+      "additionalProperties": false+    },+    "log": {+      "properties": {+        "name": { "type": "string" },+        "description": { "type": "string" },+        "export": { "$ref": "#/definitions/export_strategy" }+      },+      "additionalProperties": false+    },+    "log_defaults": {+      "description": "An optional 'defaults' block for defining YAML anchors.",+      "properties": {+        "export": { "$ref": "#/definitions/export_strategy" }+      },+      "additionalProperties": false+    },+    "metric": {+      "properties": {+        "name": { "type": "string" },+        "description": { "type": "string" },+        "export": { "$ref": "#/definitions/export_strategy" },+        "aggregate": { "$ref": "#/definitions/aggregation_strategy" }+      },+      "additionalProperties": false+    },+    "metric_defaults": {+      "description": "An optional 'defaults' block for defining YAML anchors.",+      "properties": {+        "export": { "$ref": "#/definitions/export_strategy" },+        "aggregate": { "$ref": "#/definitions/aggregation_strategy" }+      },+      "additionalProperties": false+    },+    "trace": {+      "properties": {+        "name": { "type": "string" },+        "description": { "type": "string" },+        "export": { "$ref": "#/definitions/export_strategy" }+      },+      "additionalProperties": false+    },+    "trace_defaults": {+      "description": "An optional 'defaults' block for defining YAML anchors.",+      "properties": {+        "export": { "$ref": "#/definitions/export_strategy" }+      },+      "additionalProperties": false+    },+    "profile": {+      "properties": {+        "name": { "type": "string" },+        "description": { "type": "string" },+        "export": { "$ref": "#/definitions/export_strategy" }+      },+      "additionalProperties": false+    },+    "profile_defaults": {+      "description": "An optional 'defaults' block for defining YAML anchors.",+      "properties": {+        "export": { "$ref": "#/definitions/export_strategy" }+      },+      "additionalProperties": false+    }+  }+}
data/default.yaml view
@@ -1,56 +1,144 @@+# Configuration for the eventlog processors, organised by the type of telemetry+# data they produce. Every processor shares certain properties. Let's look at+# the default configuration for the `blocks_size` metrics processor:+#+#     blocks_size:+#       name: ghc_eventlog_BlocksSize+#       description: The current heap size, calculated by the allocated number of blocks.+#       aggregate: 1x+#       export: 60x+#+# - The `name` property is added to each piece of telemetry data,+#   and determines the name under which the data becomes available in your+#   data source, such as Prometheus.+#+#   The data produced by our example `blocks_size` processor will be available+#   in Prometheus as `ghc_eventlog_BlocksSize`.+#+# - The `description` property is added to each piece of telemetry data,+#   and may be used to describe it. In most cases, this can be safely set+#   to `null` to save bandwidth.+#+#   The data produced by our example `blocks_size` processor will be given the+#   above description.+#+# - The `export` property determines whether or not the telemetry data is+#   exported to the OpenTelemetry Collector and, if so, with what frequency.+#+#   There are three ways to configure the `export` property:+#+#   - A duration in seconds, given as "Ns". For instance, if `export` is "30s"+#     then the processor exports a batch of telemetry data every 30 seconds.+#+#   - A duration in batches, given as "Nx". The length of one batch interval+#     is determined by the `--eventlog-flush-interval=SECONDS` flag, as passed+#     to the RTS of the monitored process and to this program. For instance,+#     if `export` is "2x" and the eventlog flush interval is 30 seconds, then+#     the processor exports a batch of telemetery data every minute.+#+#   - A boolean flag, i.e., `true` or `false`. If `export` is `false`, then the+#     processor will not export data and, if its output is not needed by some+#     other processor, it will not run either. If `export` is `true`, then the+#     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.+#+# - The `aggregate` property determines whether or not the telemetry data is+#   aggregated over time, and, if so, over what interval. Only metrics support+#   aggregation.+#+#   The aggregation interval uses the same format as the export interval (see+#   above). For instance, if `aggregate` is 5s, the processor aggregates data+#   over 5 second intervals.+#+#   Aggregation follows the OpenTelemetry specification. Sums are aggregated by+#   summation. Gauges are aggregated by taking the most recent data point.+#   The `heap_allocated` and `capability_usage` metrics are aggregated as sums.+#   All other metrics are aggregated as gauges.+#+#   The data produced by our example `blocks_size` processor aggregates data+#   over 1 second intervals and only keeps the last metric in each interval.+# processors:+  logs:+    thread_label:+      name: ghc_eventlog_ThreadLabel+      description: A thread label.+      export: 5s+    user_marker:+      name: ghc_eventlog_UserMarker+      description: A user marker.+      export: 5s+    user_message:+      name: ghc_eventlog_UserMessage+      description: A user log message.+      export: 5s+    internal_log_message:+      name: eventlog_live_InternalLogMessage+      description: An internal eventlog-live log message.+      export: 5s   metrics:     blocks_size:-      aggregate: by_batch # one of [null, by_batch]-      description: The current heap size, calculated by the allocated number of blocks.-      enabled: true       name: ghc_eventlog_BlocksSize+      description: The current heap size, calculated by the allocated number of blocks.+      aggregate: 1s+      export: 30s     capability_usage:-      aggregate: by_batch # one of [null, by_batch]-      description: The duration of each capability usage span.-      enabled: true       name: ghc_eventlog_CapabilityUsageDuration+      description: The duration of each capability usage span.+      aggregate: 1s+      export: 30s     heap_allocated:-      aggregate: by_batch # one of [null, by_batch]-      description: The size of a newly allocated chunk of heap.-      enabled: true       name: ghc_eventlog_HeapAllocated+      description: The size of a newly allocated chunk of heap.+      aggregate: 1s+      export: 30s     heap_live:-      aggregate: by_batch-      description: The current heap size, calculated by the allocated number of megablocks.-      enabled: true       name: ghc_eventlog_HeapLive+      description: The current heap size, calculated by the allocated number of megablocks.+      aggregate: 1s+      export: 30s     heap_prof_sample:-      aggregate: by_batch # one of [null, by_batch]-      description: A heap profile sample.-      enabled: true       name: ghc_eventlog_HeapProfSample+      description: A heap profile sample.+      aggregate: 1s+      export: 30s     heap_size:-      aggregate: by_batch # one of [null, by_batch]-      description: The current heap size, calculated by the allocated number of megablocks.-      enabled: true       name: ghc_eventlog_HeapSize+      description: The current heap size, calculated by the allocated number of megablocks.+      aggregate: 1s+      export: 30s     mem_current:-      aggregate: by_batch # one of [null, by_batch]-      description: The number of megablocks currently allocated.-      enabled: true       name: ghc_eventlog_MemCurrent+      description: The number of megablocks currently allocated.+      aggregate: 1s+      export: 30s     mem_needed:-      aggregate: by_batch # one of [null, by_batch]-      description: The number of megablocks currently needed.-      enabled: true       name: ghc_eventlog_MemNeeded+      description: The number of megablocks currently needed.+      aggregate: 1s+      export: 30s     mem_returned:-      aggregate: by_batch # one of [null, by_batch]-      description: The number of megablocks currently being returned to the OS.-      enabled: true       name: ghc_eventlog_MemReturned-  spans:+      description: The number of megablocks currently being returned to the OS.+      aggregate: 1s+      export: 30s+  traces:     capability_usage:-      description: A span of capability usage (either mutator thread or garbage collection).-      enabled: true       name: ghc_eventlog_CapabilityUsage+      description: A trace of capability usage (either mutator thread or garbage collection).+      export: false     thread_state:-      description: A span of thread state changes (either running or stopped).-      enabled: true       name: ghc_eventlog_ThreadState+      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.+      export: 30s+    cost_centre_sample:+      name: ghc_eventlog_CostCentreProfile+      description: A cost centre callstack sample.+      export: 30s
eventlog-live-otelcol.cabal view
@@ -1,119 +1,155 @@-cabal-version:      3.0-name:               eventlog-live-otelcol-version:            0.5.0.0-synopsis:           Stream eventlog data to the OpenTelemetry Collector.+cabal-version: 3.0+name: eventlog-live-otelcol+version: 0.6.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 supports live streaming of eventlog data into the OpenTelemetry Collector. -  > Usage: eventlog-live-otelcol (--eventlog-stdin | --eventlog-file FILE |-  >                                --eventlog-socket SOCKET)-  >                              [--eventlog-socket-timeout NUM]-  >                              [--eventlog-socket-exponent NUM]-  >                              [--batch-interval NUM] [--eventlog-log-file FILE]-  >                              [-h Tcmdyrbi] [--service-name STRING]-  >                              [-v|--verbosity quiet|error|warning|info|debug|0-4]-  >                              [-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]-  >                              [--enable-my-eventlog-socket-unix SOCKET]-  >                              [--enable-my-ghc-debug-socket |-  >                                --enable-my-ghc-debug-socket-unix SOCKET |-  >                                --enable-my-ghc-debug-socket-tcp ADDRESS]+  > Usage: eventlog-live-otelcol (--eventlog-stdin | --eventlog-file FILE | --eventlog-socket SOCKET)+  >                              [--eventlog-socket-timeout SECONDS]+  >                              [--eventlog-socket-exponent NUMBER]+  >                              [--eventlog-flush-interval SECONDS]+  >                              [--eventlog-log-file FILE]+  >                              [-h Tcmdyrbi]+  >                              [--service-name STRING]+  >                              [-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]+  >                              [--control]+  >                              [--control-port PORT]+  >                              [--control-cors-allow-origin ORIGIN]+  >                              [--control-cors-max-age SECONDS]+  >                              [--control-cors-require-origin]+  >                              [--control-cors-ignore-failure]+  >                              [--my-eventlog-socket-unix FILE]+  >                              [--my-ghc-debug-socket | --my-ghc-debug-socket-unix FILE | --my-ghc-debug-socket-tcp ADDRESS]   >                              [--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-timeout NUM-  >                            Eventlog socket connection retry timeout in-  >                            microseconds.-  >   --eventlog-socket-exponent NUM-  >                            Eventlog socket connection retry timeout exponent.-  >   --batch-interval NUM     Batch interval in milliseconds.-  >   --eventlog-log-file FILE Use file to log binary eventlog data.-  >   -h Tcmdyrbi              Heap profile breakdown.-  >   --service-name STRING    The name of the profiled service.-  >   -v,--verbosity quiet|error|warning|info|debug|0-4-  >                            The verbosity threshold for logging.-  >   -s,--stats               Display runtime statistics.-  >   --config FILE            The path to a detailed configuration file.-  >   --print-defaults         Print default configuration options that can be used-  >                            in config.yaml-  >   --help                   Show this help text.-  >   --version                Show version information+  >   --eventlog-stdin                   Read the eventlog from stdin.+  >   --eventlog-file FILE               Read the eventlog from a file.+  >   --eventlog-socket SOCKET           Read the eventlog from a Unix socket.+  >   --eventlog-socket-timeout 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.+  >   -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   >   > OpenTelemetry Collector Server Options-  >   --otelcol-host HOST      Server hostname.-  >   --otelcol-port PORT      Server TCP port.-  >   --otelcol-authority HOST Server authority.-  >   --otelcol-ssl            Use SSL.-  >   --otelcol-certificate-store FILE-  >                            Store for certificate validation.-  >   --otelcol-ssl-key-log FILE-  >                            Use file to log SSL keys.-  >   --otelcol-ssl-key-log-from-env-  >                            Use SSLKEYLOGFILE to log SSL keys.+  >   --otelcol-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.   >+  > Control Server Options+  >   --control                          Start the control server.+  >   --control-port PORT                The port number for the control server.+  >   --control-cors-allow-origin ORIGIN Set the allowed origins for the control server CORS policy.+  >   --control-cors-max-age SECONDS     Set the maximum age of a cached CORS preflight request for the control server CORS policy.+  >   --control-cors-require-origin      If enabled, the control server will not accept requests without an Origin header.+  >   --control-cors-ignore-failure      If enabled, the control server will accept malformed CORS preflight requests.+  >   > Debug Options-  >   --enable-my-eventlog-socket-unix SOCKET-  >                            Enable the eventlog socket for this program on the-  >                            given Unix socket.-  >   --enable-my-ghc-debug-socket-  >                            Enable ghc-debug for this program.-  >   --enable-my-ghc-debug-socket-unix SOCKET-  >                            Enable ghc-debug for this program on the given Unix-  >                            socket.-  >   --enable-my-ghc-debug-socket-tcp ADDRESS-  >                            Enable ghc-debug for this program on the given TCP-  >                            socket specified as 'host:port'.+  >   --my-eventlog-socket-unix FILE     Open an eventlog socket for this program on the given Unix socket.+  >   --my-ghc-debug-socket              Open the default ghc-debug socket for this program.+  >   --my-ghc-debug-socket-unix FILE    Open a ghc-debug Unix domain socket with the given file path.+  >   --my-ghc-debug-socket-tcp ADDRESS  Open a ghc-debug TCP/IP socket with the given address as 'host:port'. -license:            BSD-3-Clause-license-file:       LICENSE-author:             Wen Kokke-maintainer:         wen@well-typed.com-copyright:          (c) 2025 Well-Typed-build-type:         Simple-category:           Debug, Monitoring, System-extra-doc-files:    CHANGELOG.md-extra-source-files: data/default.yaml+license: BSD-3-Clause+license-file: LICENSE+author: Wen Kokke+maintainer: wen@well-typed.com+copyright: (c) 2025 Well-Typed+build-type: Simple+category: Debug, Monitoring, System+extra-doc-files: CHANGELOG.md+extra-source-files:+  data/config.schema.json+  data/default.yaml+ tested-with:-  GHC ==9.2.8 || ==9.4.8 || ==9.6.7 || ==9.8.4 || ==9.10.2+  ghc ==9.2.8 || ==9.4.8 || ==9.6.7 || ==9.8.4 || ==9.10.2  source-repository head-  type:     git+  type: git   location: https://github.com/well-typed/eventlog-live.git-  subdir:   eventlog-live-otelcol+  subdir: eventlog-live-otelcol +source-repository this+  type: git+  location: https://github.com/well-typed/eventlog-live.git+  subdir: eventlog-live-otelcol+  tag: eventlog-live-otelcol-0.6.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+-- eventlog-socket SHOULD also be compiled with +control. If this flag is+-- set at compile-time, the control server can be enabled at runtime.+flag control+  description: Enable control command server.+  default: False+  manual: True++-- 2026-04-02:+-- This flag should enable switching the use of eventlog-socket on+-- and off, since it may cause issues with certain versions of GHC.+flag use-eventlog-socket+  description: Use eventlog-socket+  default: False+  manual: True+ -- 2025-11-10: -- This flag should enable switching the use of ghc-debug-stub on -- and off, since it may cause issues with certain versions of GHC. -- See https://github.com/well-typed/eventlog-live/issues/88 flag use-ghc-debug-stub   description: Use ghc-debug-stub-  default:     False-  manual:      True+  default: False+  manual: True  -- 2025-11-06: -- This flag should enable switching between template-haskell and -- template-haskell-lift, because the latter is not on nixpkgs. flag use-template-haskell-lift   description: Use template-haskell-lift in place of template-haskell-  default:     False-  manual:      False+  default: False+  manual: False  common language   ghc-options:-    -Wall -Wcompat -Widentities -Wprepositive-qualified-module-    -Wredundant-constraints -Wunticked-promoted-constructors+    -Wall+    -Wcompat+    -Widentities+    -Wprepositive-qualified-module+    -Wredundant-constraints+    -Wunticked-promoted-constructors     -Wunused-packages -  default-language:   Haskell2010+  default-language: Haskell2010   default-extensions:     BangPatterns+    ConstraintKinds     DataKinds     DefaultSignatures     DeriveFoldable@@ -139,71 +175,111 @@     RankNTypes     RecordWildCards     ScopedTypeVariables+    StandaloneKindSignatures     TupleSections     TypeApplications     TypeFamilies     TypeOperators  library-  import:          language-  hs-source-dirs:  src-  exposed-modules:-    GHC.Eventlog.Live.Otelcol-    GHC.Eventlog.Live.Otelcol.Config-    GHC.Eventlog.Live.Otelcol.Exporter-    GHC.Eventlog.Live.Otelcol.Stats-+  import: language+  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.Logs+    GHC.Eventlog.Live.Otelcol.Exporter.Metrics+    GHC.Eventlog.Live.Otelcol.Exporter.Profiles+    GHC.Eventlog.Live.Otelcol.Exporter.Traces+    GHC.Eventlog.Live.Otelcol.Options+    GHC.Eventlog.Live.Otelcol.Processor.Common.Core+    GHC.Eventlog.Live.Otelcol.Processor.Common.Logs+    GHC.Eventlog.Live.Otelcol.Processor.Common.Metrics+    GHC.Eventlog.Live.Otelcol.Processor.Common.Profiles+    GHC.Eventlog.Live.Otelcol.Processor.Common.Traces+    GHC.Eventlog.Live.Otelcol.Processor.Heap+    GHC.Eventlog.Live.Otelcol.Processor.Logs+    GHC.Eventlog.Live.Otelcol.Processor.Profiles+    GHC.Eventlog.Live.Otelcol.Processor.Threads+    GHC.Eventlog.Live.Otelcol.Stats+    GHC.Eventlog.Socket.Compat     Language.Haskell.TH.Lift.Compat     Options.Applicative.Compat+    Options.Applicative.Extra.Feature     Paths_eventlog_live_otelcol     System.Random.Compat    autogen-modules: Paths_eventlog_live_otelcol   build-depends:-    , aeson                  >=2.2    && <2.3-    , ansi-terminal          >=1.1    && <1.2-    , base                   >=4.16   && <4.22-    , bytestring             >=0.11   && <0.13-    , data-default           >=0.2    && <0.9-    , dlist                  >=1.0    && <1.1-    , eventlog-live          >=0.4    && <0.5-    , eventlog-socket        >=0.1.0  && <0.2-    , file-embed             >=0.0.16 && <0.1-    , ghc-events             >=0.20   && <0.21-    , grapesy                >=1.0.0  && <1.2-    , hashable               >=1.4    && <1.6-    , hs-opentelemetry-otlp  >=0.1.0  && <0.2-    , lens-family            >=2.1.3  && <2.2-    , machines               >=0.7.4  && <0.8-    , optparse-applicative   >=0.17   && <0.20-    , proto-lens             >=0.7.1  && <0.8-    , random                 >=1.2    && <1.4-    , strict-list            >=0.1    && <0.2-    , table-layout           >=1.0    && <1.1-    , text                   >=1.2    && <2.2-    , unordered-containers   >=0.2.20 && <0.3-    , vector                 >=0.11   && <0.14-    , yaml                   >=0.11   && <0.12+    ansi-terminal >=1.1 && <1.2,+    base >=4.16 && <4.22,+    bytestring >=0.11 && <0.13,+    containers >=0.6 && <0.8,+    data-default >=0.2 && <0.9,+    dlist >=1.0 && <1.1,+    eventlog-live >=0.5 && <0.6,+    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,+    lens-family >=2.1.3 && <2.2,+    machines >=0.7.4 && <0.8,+    optparse-applicative >=0.17 && <0.20,+    proto-lens >=0.7.1 && <0.8,+    random >=1.2 && <1.4,+    stm >=2.5 && <2.6,+    strict-list >=0.1 && <0.2,+    table-layout >=1.0 && <1.1,+    text >=1.2 && <2.2,+    transformers >=0.2 && <0.7,+    unordered-containers >=0.2.20 && <0.3,+    vector >=0.11 && <0.14, +  if flag(control)+    build-depends:+      aeson >=2.2 && <2.3,+      binary >=0.8 && <0.9,+      eventlog-socket-control >=0.1.1 && <0.2,+      fast-logger >=3.0 && <3.3,+      http-api-data >=0.6 && <0.7,+      network >=3.2.7 && <3.3,+      servant >=0.20 && <0.21,+      servant-server >=0.20 && <0.21,+      wai >=3.2 && <3.3,+      wai-cors >=0.2 && <0.3,+      wai-extra >=3.1 && <3.2,+      warp >=3.4 && <3.5,++    cpp-options: -DEVENTLOG_LIVE_OTELCOL_FEATURE_CONTROL++  if flag(use-eventlog-socket)+    build-depends: eventlog-socket >=0.1.2 && <0.2+    cpp-options: -DEVENTLOG_LIVE_OTELCOL_USE_EVENTLOG_SOCKET+   if flag(use-ghc-debug-stub)-    build-depends: ghc-debug-stub >=0.1 && <1.0-    cpp-options:   -DEVENTLOG_LIVE_OTELCOL_USE_GHC_DEBUG_STUB+    build-depends: ghc-debug-stub >=0.1 && <1+    cpp-options: -DEVENTLOG_LIVE_OTELCOL_USE_GHC_DEBUG_STUB    if flag(use-template-haskell-lift)     build-depends: template-haskell-lift >=0.1 && <0.2-    cpp-options:   -DEVENTLOG_LIVE_OTELCOL_USE_TEMPLATE_HASKELL_LIFT-+    cpp-options: -DEVENTLOG_LIVE_OTELCOL_USE_TEMPLATE_HASKELL_LIFT   else-    build-depends: template-haskell >=2.2 && <3.0+    build-depends: template-haskell >=2.2 && <3  executable eventlog-live-otelcol-  import:         language-  main-is:        Main.hs+  import: language+  main-is: Main.hs   hs-source-dirs: app-  build-depends:  eventlog-live-otelcol-  ghc-options:    -threaded -rtsopts -finfo-table-map+  build-depends: eventlog-live-otelcol+  ghc-options:+    -threaded+    -rtsopts+    -finfo-table-map
src/GHC/Debug/Stub/Compat.hs view
@@ -1,4 +1,5 @@ {-# LANGUAGE CPP #-}+{-# LANGUAGE OverloadedStrings #-}  {- | Module      : GHC.Debug.Stub.Compat@@ -13,23 +14,40 @@ ) where  import Control.Applicative (asum)-import Data.Text qualified as T (pack)-import GHC.Eventlog.Live.Logger (logError)-import GHC.Eventlog.Live.Verbosity (Verbosity)+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 -#if defined(EVENTLOG_LIVE_OTELCOL_USE_GHC_DEBUG_STUB)+#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 System.Exit (exitFailure)-import Options.Applicative.Help.Pretty qualified as OP+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@@ -41,141 +59,63 @@ Internal helper. Start @ghc-debug@ on the given `MyGhcDebugSocket`. -}-withMyGhcDebug :: Verbosity -> Maybe MyGhcDebugSocket -> IO a -> IO a-#if defined(EVENTLOG_LIVE_OTELCOL_USE_GHC_DEBUG_STUB)-withMyGhcDebug verbosity maybeMyGhcDebugSocket action =+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 ->+    Just MyGhcDebugSocketDefault -> do+      writeLog logger INFO $+        "Start ghc-debug with default socket."       GHC.Debug.withGhcDebug action-    Just (MyGhcDebugSocketUnix myGhcDebugSocketUnix) ->+    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-          logError verbosity . T.pack $-            "Could not parse ghc-debug TCP address " <> myGhcDebugSocketTcp <> "."+          writeLog logger FATAL $+            T.pack $ "Could not parse ghc-debug TCP address " <> myGhcDebugSocketTcp <> "."           exitFailure         Right portWord16 ->           GHC.Debug.withGhcDebugTCP host portWord16 action #else-withMyGhcDebug verbosity maybeMyGhcDebugSocket action-  | isJust maybeMyGhcDebugSocket = do-    logError verbosity $ T.pack myGhcDebugSocketUnsupportedErrorMessage-    exitFailure-  | otherwise = action+withMyGhcDebug logger maybeMyGhcDebugSocket action = do+  when (isJust maybeMyGhcDebugSocket) $+    OF.exitIfUnsupported useGhcDebugStub logger+  action #endif  -------------------------------------------------------------------------------- -- My GHC Debug  maybeMyGhcDebugSocketParser :: O.Parser (Maybe MyGhcDebugSocket)-#if defined(EVENTLOG_LIVE_OTELCOL_USE_GHC_DEBUG_STUB) maybeMyGhcDebugSocketParser =-  O.optional . asum $+  asum $     [ myGhcDebugSocketDefaultParser     , myGhcDebugSocketUnixParser     , myGhcDebugSocketTcpParser-    ]- where-  myGhcDebugSocketDefaultParser =-    O.flag'-      MyGhcDebugSocketDefault-      ( O.long myGhcDebugSocketDefaultLong-          <> O.help myGhcDebugSocketDefaultHelp-      )-  myGhcDebugSocketUnixParser =-    MyGhcDebugSocketUnix-      <$> O.strOption-        ( O.long myGhcDebugSocketUnixLong-            <> O.metavar myGhcDebugSocketUnixMetavar-            <> O.help myGhcDebugSocketUnixHelp-        )-  myGhcDebugSocketTcpParser =-    MyGhcDebugSocketUnix-      <$> O.strOption-        ( O.long myGhcDebugSocketTcpLong-            <> O.metavar myGhcDebugSocketTcpMetavar-            <> O.help myGhcDebugSocketTcpHelp-        )-#else-maybeMyGhcDebugSocketParser =-  asum . fmap mkUnsupportedParser $-    [ myGhcDebugSocketDefaultMod-    , myGhcDebugSocketUnixMod-    , myGhcDebugSocketTcpMod+    , pure Nothing     ]- where-  mkUnsupportedParser modOptionFields =-    Nothing <$ O.infoOption myGhcDebugSocketUnsupportedErrorMessage modOptionFields-  mkUnsupportedHelpDoc help =-    Just $ OP.vcat [OP.pretty $ "Unsupported. Requires build with -f+" <> myGhcDebugFeatureFlag <> ".", OP.pretty help]-  myGhcDebugSocketDefaultMod =-     O.long myGhcDebugSocketDefaultLong-      <> O.hidden-      <> O.helpDoc (mkUnsupportedHelpDoc myGhcDebugSocketDefaultHelp)-  myGhcDebugSocketUnixMod =-      O.long myGhcDebugSocketUnixLong-      <> O.metavar myGhcDebugSocketUnixMetavar-      <> O.hidden-      <> O.helpDoc (mkUnsupportedHelpDoc myGhcDebugSocketUnixHelp)-  myGhcDebugSocketTcpMod =-      O.long myGhcDebugSocketTcpLong-      <> O.metavar myGhcDebugSocketTcpMetavar-      <> O.hidden-      <> O.helpDoc (mkUnsupportedHelpDoc myGhcDebugSocketTcpHelp)-#endif -#if !defined(EVENTLOG_LIVE_OTELCOL_USE_GHC_DEBUG_STUB)-myGhcDebugFeatureFlag :: String-myGhcDebugFeatureFlag =-  "use-ghc-debug-stub"-#endif--#if !defined(EVENTLOG_LIVE_OTELCOL_USE_GHC_DEBUG_STUB)-myGhcDebugSocketUnsupportedErrorMessage :: String-myGhcDebugSocketUnsupportedErrorMessage =-  "Cannot open ghc-debug socket. This executable was built without -f+" <> myGhcDebugFeatureFlag <> "."-{-# INLINE myGhcDebugSocketUnsupportedErrorMessage #-}-#endif--myGhcDebugSocketDefaultLong :: String-myGhcDebugSocketDefaultLong =-  "enable-my-ghc-debug-socket"-{-# INLINE myGhcDebugSocketDefaultLong #-}--myGhcDebugSocketDefaultHelp :: String-myGhcDebugSocketDefaultHelp =-  "Enable ghc-debug for this program."-{-# INLINE myGhcDebugSocketDefaultHelp #-}--myGhcDebugSocketUnixLong :: String-myGhcDebugSocketUnixLong =-  "enable-my-ghc-debug-socket-unix"-{-# INLINE myGhcDebugSocketUnixLong #-}--myGhcDebugSocketUnixMetavar :: String-myGhcDebugSocketUnixMetavar =-  "SOCKET"-{-# INLINE myGhcDebugSocketUnixMetavar #-}--myGhcDebugSocketUnixHelp :: String-myGhcDebugSocketUnixHelp =-  "Enable ghc-debug for this program on the given Unix socket."-{-# INLINE myGhcDebugSocketUnixHelp #-}--myGhcDebugSocketTcpLong :: String-myGhcDebugSocketTcpLong =-  "enable-my-ghc-debug-socket-tcp"-{-# INLINE myGhcDebugSocketTcpLong #-}+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." -myGhcDebugSocketTcpMetavar :: String-myGhcDebugSocketTcpMetavar =-  "ADDRESS"-{-# INLINE myGhcDebugSocketTcpMetavar #-}+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." -myGhcDebugSocketTcpHelp :: String-myGhcDebugSocketTcpHelp =-  "Enable ghc-debug for this program on the given TCP socket specified as 'host:port'."-{-# INLINE myGhcDebugSocketTcpHelp #-}+myGhcDebugSocketTcpParser :: O.Parser (Maybe MyGhcDebugSocket)+myGhcDebugSocketTcpParser =+  OF.onlyFor useGhcDebugStub (O.option (Just . MyGhcDebugSocketTcp <$> O.str)) (O.metavar "ADDRESS") $+    O.long "my-ghc-debug-socket-tcp"+      <> OF.helpFor useGhcDebugStub "Open a ghc-debug TCP/IP socket with the given address as 'host:port'."
src/GHC/Eventlog/Live/Otelcol.hs view
@@ -1,983 +1,390 @@ {-# LANGUAGE OverloadedStrings #-}-{-# OPTIONS_GHC -Wno-name-shadowing #-}--{- |-Module      : GHC.Eventlog.Live.Otelcol-Description : The implementation of @eventlog-live-otelcol@.-Stability   : experimental-Portability : portable--}-module GHC.Eventlog.Live.Otelcol (-  main,-) where--import Control.Applicative (asum)-import Control.Monad.IO.Class (MonadIO (..))-import Data.ByteString (ByteString)-import Data.Coerce (Coercible, coerce)-import Data.DList (DList)-import Data.DList qualified as D-import Data.Default (Default (..))-import Data.Either (partitionEithers)-import Data.Foldable (for_)-import Data.Functor ((<&>))-import Data.HashMap.Strict qualified as M-import Data.Hashable (Hashable)-import Data.Machine (MachineT, Process, ProcessT, asParts, await, construct, mapping, repeatedly, stopped, yield, (~>))-import Data.Machine.Fanout (fanout)-import Data.Maybe (fromMaybe, mapMaybe)-import Data.ProtoLens (Message (defMessage))-import Data.Proxy (Proxy (..))-import Data.Semigroup (Last (..), Sum (..))-import Data.Text (Text)-import Data.Text qualified as T-import Data.Text.Encoding qualified as TE-import Data.Version (showVersion)-import Data.Word (Word32, Word64)-import Data.Yaml qualified as Y-import GHC.Debug.Stub.Compat (MyGhcDebugSocket (..), maybeMyGhcDebugSocketParser, withMyGhcDebug)-import GHC.Eventlog.Live.Data.Attribute-import GHC.Eventlog.Live.Data.Group (Group, GroupBy, GroupedBy)-import GHC.Eventlog.Live.Data.Group qualified as DG-import GHC.Eventlog.Live.Data.Metric (Metric (..))-import GHC.Eventlog.Live.Logger (logDebug)-import GHC.Eventlog.Live.Machine.Analysis.Capability (CapabilityUsageSpan)-import GHC.Eventlog.Live.Machine.Analysis.Capability qualified as M-import GHC.Eventlog.Live.Machine.Analysis.Heap (MemReturnData (..))-import GHC.Eventlog.Live.Machine.Analysis.Heap qualified as M-import GHC.Eventlog.Live.Machine.Analysis.Thread (ThreadStateSpan (..))-import GHC.Eventlog.Live.Machine.Analysis.Thread qualified as M-import GHC.Eventlog.Live.Machine.Core (Tick)-import GHC.Eventlog.Live.Machine.Core qualified as M-import GHC.Eventlog.Live.Machine.WithStartTime (WithStartTime (..))-import GHC.Eventlog.Live.Machine.WithStartTime qualified as M-import GHC.Eventlog.Live.Options-import GHC.Eventlog.Live.Otelcol.Config (Config)-import GHC.Eventlog.Live.Otelcol.Config qualified as C-import GHC.Eventlog.Live.Otelcol.Config.Default.Raw (defaultConfigString)-import GHC.Eventlog.Live.Otelcol.Exporter (exportResourceMetrics, exportResourceSpans)-import GHC.Eventlog.Live.Otelcol.Stats (Stat (..), eventCountTick, processStats)-import GHC.Eventlog.Live.Socket (runWithEventlogSource)-import GHC.Eventlog.Live.Verbosity (Verbosity)-import GHC.Eventlog.Socket qualified as Eventlog.Socket-import GHC.RTS.Events (Event (..), HeapProfBreakdown (..), ThreadId)-import GHC.Records (HasField)-import Lens.Family2 ((&), (.~))-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 Paths_eventlog_live_otelcol qualified as EventlogLive-import Proto.Opentelemetry.Proto.Collector.Metrics.V1.MetricsService qualified as OMS-import Proto.Opentelemetry.Proto.Collector.Trace.V1.TraceService qualified as OTS-import Proto.Opentelemetry.Proto.Collector.Trace.V1.TraceService_Fields qualified as OTS-import Proto.Opentelemetry.Proto.Common.V1.Common qualified as OC-import Proto.Opentelemetry.Proto.Common.V1.Common_Fields qualified as OC-import Proto.Opentelemetry.Proto.Metrics.V1.Metrics qualified as OM-import Proto.Opentelemetry.Proto.Metrics.V1.Metrics_Fields qualified as OM-import Proto.Opentelemetry.Proto.Trace.V1.Trace qualified as OT-import Proto.Opentelemetry.Proto.Trace.V1.Trace_Fields qualified as OT-import System.Random (StdGen, initStdGen)-import System.Random.Compat (uniformByteString)--{- |-The main function for @eventlog-live-otelcol@.--}-main :: IO ()-main = do-  Options{..} <- O.execParser options--  -- Instument THIS PROGRAM with eventlog-socket and/or ghc-debug.-  let MyDebugOptions{..} = myDebugOptions-  withMyEventlogSocket maybeMyEventlogSocket-  withMyGhcDebug verbosity maybeMyGhcDebugSocket $ do-    ---    -- Read the configuration file.-    config <- flip (maybe (pure def)) maybeConfigFile $ \configFile -> do-      logDebug verbosity $ "Reading configuration file from " <> T.pack configFile-      config <- C.readConfig configFile-      logDebug verbosity $ "Configuration file:\n" <> (TE.decodeUtf8Lenient . Y.encode $ config)-      pure config--    -- Create the service name attribute.-    let attrServiceName = ("service.name", maybe AttrNull (AttrText . (.serviceName)) maybeServiceName)--    -- Open a connection to the OpenTelemetry Collector.-    let OpenTelemetryCollectorOptions{..} = openTelemetryCollectorOptions-    G.withConnection G.def openTelemetryCollectorServer $ \conn -> do-      runWithEventlogSource-        verbosity-        eventlogSocket-        eventlogSocketTimeout-        eventlogSocketTimeoutExponent-        batchInterval-        Nothing-        maybeEventlogLogFile-        $ fanout-          [ M.validateInput verbosity 10-          , eventCountTick ~> mapping (D.singleton . EventCountStat)-          , M.liftTick M.withStartTime-              ~> fanout-                [ processHeapEvents config verbosity maybeHeapProfBreakdown-                    ~> mapping (fmap Left)-                , processThreadEvents config verbosity-                ]-              ~> mapping (partitionEithers . D.toList)-              ~> fanout-                [ runIf (shouldExportMetrics config) $-                    mapping fst-                      ~> fanout-                        [ asScopeMetrics-                            [ OM.scope .~ eventlogLiveScope-                            ]-                            ~> asResourceMetric []-                            ~> asExportMetricServiceRequest-                            ~> exportResourceMetrics conn-                            ~> mapping (D.singleton . ExportMetricsResultStat)-                        ]-                , runIf (shouldExportSpans config) $-                    mapping snd-                      ~> fanout-                        [ asScopeSpans-                            [ OT.scope .~ eventlogLiveScope-                            ]-                            ~> asResourceSpan-                              [ OT.resource-                                  .~ messageWith-                                    [ OM.attributes .~ mapMaybe toMaybeKeyValue [attrServiceName]-                                    ]-                              ]-                            ~> asExportTraceServiceRequest-                            ~> exportResourceSpans conn-                            ~> mapping (D.singleton . ExportTraceResultStat)-                        ]-                ]-          ]-          ~> asParts-          ~> processStats verbosity stats batchInterval 10--{- |-Internal helper.-Determine whether or not any spans should be exported.--}-shouldExportMetrics :: Config -> Bool-shouldExportMetrics config =-  or-    [ C.processorEnabled (.metrics) (.heapAllocated) config-    , C.processorEnabled (.metrics) (.blocksSize) config-    , C.processorEnabled (.metrics) (.heapSize) config-    , C.processorEnabled (.metrics) (.heapLive) config-    , C.processorEnabled (.metrics) (.memCurrent) config-    , C.processorEnabled (.metrics) (.memNeeded) config-    , C.processorEnabled (.metrics) (.memReturned) config-    , C.processorEnabled (.metrics) (.heapProfSample) config-    , C.processorEnabled (.metrics) (.capabilityUsage) config-    ]--{- |-Internal helper.-Determine whether or not any spans should be exported.--}-shouldExportSpans :: Config -> Bool-shouldExportSpans config =-  C.processorEnabled (.spans) (.capabilityUsage) config-    || C.processorEnabled (.spans) (.threadState) config------------------------------------------------------------------------------------- processThreadEvents-----------------------------------------------------------------------------------data OneOf a b c = A !a | B !b | C !c--processThreadEvents ::-  (MonadIO m) =>-  Config ->-  Verbosity ->-  ProcessT m (Tick (WithStartTime Event)) (DList (Either OM.Metric OT.Span))-processThreadEvents config verbosity =-  runIf (shouldProcessThreadEvents config) $-    M.sortByBatchTick (.value.evTime)-      ~> M.liftTick-        ( fanout-            [ M.validateOrder verbosity (.value.evTime)-            , runIf (shouldComputeCapabilityUsageSpan config) $-                M.processGCSpans verbosity-                  ~> mapping (D.singleton . A)-            , runIf (shouldComputeThreadStateSpan config) $-                M.processThreadStateSpans' M.tryGetTimeUnixNano (.value) M.setWithStartTime'value verbosity-                  ~> fanout-                    [ M.asMutatorSpans' (.value) M.setWithStartTime'value-                        ~> mapping (D.singleton . B)-                    , mapping (D.singleton . C)-                    ]-            ]-        )-      ~> M.liftTick-        ( asParts-            ~> mapping repackCapabilityUsageSpanOrThreadStateSpan-        )-      ~> fanout-        [ M.liftTick-            ( mapping leftToMaybe-                ~> asParts-            )-            ~> fanout-              [ runIf (C.processorEnabled (.metrics) (.capabilityUsage) config) $-                  M.liftTick M.processCapabilityUsageMetrics-                    ~> aggregate viaSum (C.processorAggregationStrategy (.metrics) (.capabilityUsage) config)-                    ~> mapping (fmap toNumberDataPoint)-                    ~> asSum-                      [ OM.aggregationTemporality .~ OM.AGGREGATION_TEMPORALITY_DELTA-                      , OM.isMonotonic .~ True-                      ]-                    ~> asMetricWith config (.capabilityUsage) [OM.unit .~ "ns"]-                    ~> mapping (D.singleton . Left)-              , runIf (C.processorEnabled (.spans) (.capabilityUsage) config) $-                  M.liftTick-                    ( M.dropStartTime-                        ~> asSpan config-                        ~> mapping (D.singleton . Right)-                    )-                    ~> M.batchByTick-              ]-        , runIf (C.processorEnabled (.spans) (.threadState) config) $-            M.liftTick-              ( mapping rightToMaybe-                  ~> asParts-                  ~> asSpan config-                  ~> mapping (D.singleton . Right)-              )-              ~> M.batchByTick-        ]- where-  repackCapabilityUsageSpanOrThreadStateSpan = \case-    A i -> Left $ fmap Left i-    B i -> Left $ fmap Right i-    C i -> Right i.value--{- |-Internal helper.-Get the `Left` value, if any.--}-leftToMaybe :: Either a b -> Maybe a-leftToMaybe = either Just (const Nothing)--{- |-Internal helper.-Get the `Right` value, if any.--}-rightToMaybe :: Either a b -> Maybe b-rightToMaybe = either (const Nothing) Just--{- |-Internal helper.-Determine whether or not any thread events should be processed at all.--}-shouldProcessThreadEvents :: Config -> Bool-shouldProcessThreadEvents config =-  C.processorEnabled (.metrics) (.capabilityUsage) config-    || C.processorEnabled (.spans) (.capabilityUsage) config-    || C.processorEnabled (.spans) (.threadState) config--{- |-Internal helper.-Determine whether or not the capability usage spans should be computed.--}-shouldComputeCapabilityUsageSpan :: Config -> Bool-shouldComputeCapabilityUsageSpan config =-  C.processorEnabled (.spans) (.capabilityUsage) config-    || C.processorEnabled (.metrics) (.capabilityUsage) config--{- |-Internal helper.-Determine whether or not the thread state spans should be computed.--}-shouldComputeThreadStateSpan :: Config -> Bool-shouldComputeThreadStateSpan config =-  C.processorEnabled (.spans) (.threadState) config-    || shouldComputeCapabilityUsageSpan config------------------------------------------------------------------------------------- processHeapEvents-----------------------------------------------------------------------------------processHeapEvents ::-  (MonadIO m) =>-  Config ->-  Verbosity ->-  Maybe HeapProfBreakdown ->-  ProcessT m (Tick (WithStartTime Event)) (DList OM.Metric)-processHeapEvents config verbosity maybeHeapProfBreakdown =-  fanout-    [ processHeapAllocated config-    , processBlocksSize config-    , processHeapSize config-    , processHeapLive config-    , processMemReturn config-    , processHeapProfSample config verbosity maybeHeapProfBreakdown-    ]------------------------------------------------------------------------------------- Metric Aggregation--data Aggregators a b = Aggregators-  { nothing :: Process (Tick a) b-  , byBatch :: Process (Tick a) b-  }--{- |-Internal helper.-Aggregate items based on the provided aggregators and aggregation strategy.--}-aggregate :: Aggregators a b -> Maybe C.AggregationStrategy -> Process (Tick a) b-aggregate Aggregators{..} = \case-  Nothing -> nothing-  Just C.AggregationStrategyByBatch -> byBatch--{- |-Internal helper.-Metric aggregators via the `Semigroup` instance for `Sum`.--}-viaSum :: forall a. (Num a) => Aggregators (Metric a) [Metric a]-viaSum =-  Aggregators-    { nothing = M.batchByTickList-    , byBatch =-        -- TODO: emit group sample counts as separate metric-        byBatchVia (Proxy @(Metric (Sum a)))-          ~> mapping (fmap (.representative))-    }--{- |-Internal helper.-Metric aggregators via the `Semigroup` instance for `Last`.--}-viaLast :: forall a. (GroupBy a) => Aggregators a [a]-viaLast =-  Aggregators-    { nothing = M.batchByTickList-    , byBatch =-        -- TODO: emit group sample counts as separate metric-        byBatchVia (Proxy @(Last a))-          ~> mapping (fmap (.representative))-    }--{- |-Internal helper.-This function aggregates items via a `Semigroup` instance and grouped by the `GroupBy` instance.--}-byBatchVia ::-  forall a b.-  (Coercible a b, GroupBy b, Semigroup b) =>-  Proxy b ->-  Process (Tick a) [Group a]-byBatchVia (Proxy :: Proxy b) =-  mapping (fmap DG.singleton . coerce @(Tick a) @(Tick b))-    ~> M.aggregateByTick @(GroupedBy b)-    ~> mapping (coerce @[Group b] @[Group a] . DG.groups)------------------------------------------------------------------------------------- HeapAllocated--processHeapAllocated :: Config -> Process (Tick (WithStartTime Event)) (DList OM.Metric)-processHeapAllocated config =-  runIf (C.processorEnabled (.metrics) (.heapAllocated) config) $-    M.liftTick M.processHeapAllocatedData-      ~> aggregate viaSum (C.processorAggregationStrategy (.metrics) (.heapAllocated) config)-      ~> mapping (fmap toNumberDataPoint)-      ~> asSum-        [ OM.aggregationTemporality .~ OM.AGGREGATION_TEMPORALITY_DELTA-        , OM.isMonotonic .~ True-        ]-      ~> asMetricWith config (.heapAllocated) [OM.unit .~ "By"]-      ~> mapping D.singleton------------------------------------------------------------------------------------- HeapSize--processHeapSize :: Config -> Process (Tick (WithStartTime Event)) (DList OM.Metric)-processHeapSize config =-  runIf (C.processorEnabled (.metrics) (.heapSize) config) $-    M.liftTick M.processHeapSizeData-      ~> aggregate viaLast (C.processorAggregationStrategy (.metrics) (.heapSize) config)-      ~> mapping (fmap toNumberDataPoint)-      ~> asGauge-      ~> asMetricWith config (.heapSize) [OM.unit .~ "By"]-      ~> mapping D.singleton------------------------------------------------------------------------------------- BlocksSize--processBlocksSize :: Config -> Process (Tick (WithStartTime Event)) (DList OM.Metric)-processBlocksSize config =-  runIf (C.processorEnabled (.metrics) (.blocksSize) config) $-    M.liftTick M.processBlocksSizeData-      ~> aggregate viaLast (C.processorAggregationStrategy (.metrics) (.blocksSize) config)-      ~> mapping (fmap toNumberDataPoint)-      ~> asGauge-      ~> asMetricWith config (.blocksSize) [OM.unit .~ "By"]-      ~> mapping D.singleton------------------------------------------------------------------------------------- HeapLive--processHeapLive :: Config -> Process (Tick (WithStartTime Event)) (DList OM.Metric)-processHeapLive config =-  runIf (C.processorEnabled (.metrics) (.heapLive) config) $-    M.liftTick M.processHeapLiveData-      ~> aggregate viaLast (C.processorAggregationStrategy (.metrics) (.heapLive) config)-      ~> mapping (fmap toNumberDataPoint)-      ~> asGauge-      ~> asMetricWith config (.heapLive) [OM.unit .~ "By"]-      ~> mapping D.singleton------------------------------------------------------------------------------------- MemReturn--processMemReturn :: Config -> Process (Tick (WithStartTime Event)) (DList OM.Metric)-processMemReturn config =-  runIf (shouldComputeMemReturn config) $-    M.liftTick M.processMemReturnData-      ~> fanout-        [ runIf (C.processorEnabled (.metrics) (.memCurrent) config) $-            M.liftTick asMemCurrent-              ~> aggregate viaLast (C.processorAggregationStrategy (.metrics) (.memCurrent) config)-              ~> mapping (fmap toNumberDataPoint)-              ~> asGauge-              ~> asMetricWith config (.memCurrent) [OM.unit .~ "{mblock}"]-              ~> mapping D.singleton-        , runIf (C.processorEnabled (.metrics) (.memNeeded) config) $-            M.liftTick asMemNeeded-              ~> aggregate viaLast (C.processorAggregationStrategy (.metrics) (.memNeeded) config)-              ~> mapping (fmap toNumberDataPoint)-              ~> asGauge-              ~> asMetricWith config (.memNeeded) [OM.unit .~ "{mblock}"]-              ~> mapping D.singleton-        , runIf (C.processorEnabled (.metrics) (.memReturned) config) $-            M.liftTick asMemReturned-              ~> aggregate viaLast (C.processorAggregationStrategy (.metrics) (.memReturned) config)-              ~> mapping (fmap toNumberDataPoint)-              ~> asGauge-              ~> asMetricWith config (.memReturned) [OM.unit .~ "{mblock}"]-              ~> mapping D.singleton-        ]--{- |-Internal helper.-Determine whether the MemReturn data should be computed.--}-shouldComputeMemReturn :: Config -> Bool-shouldComputeMemReturn config =-  C.processorEnabled (.metrics) (.memCurrent) config-    || C.processorEnabled (.metrics) (.memNeeded) config-    || C.processorEnabled (.metrics) (.memReturned) config--asMemCurrent :: Process (Metric MemReturnData) (Metric Word32)-asMemCurrent = mapping (fmap (.current))--asMemNeeded :: Process (Metric MemReturnData) (Metric Word32)-asMemNeeded = mapping (fmap (.needed))--asMemReturned :: Process (Metric MemReturnData) (Metric Word32)-asMemReturned = mapping (fmap (.returned))------------------------------------------------------------------------------------- HeapProfSample--heapProfSampleDataAggregators :: Aggregators M.HeapProfSampleData [Metric Word64]-heapProfSampleDataAggregators = Aggregators{..}- where-  nothing :: Process (Tick M.HeapProfSampleData) [Metric Word64]-  nothing =-    M.batchByTickList-      ~> mapping (concatMap M.heapProfSamples)-  byBatch :: Process (Tick M.HeapProfSampleData) [Metric Word64]-  byBatch =-    -- TODO: emit group sample counts as separate metric-    mapping (fmap (DG.singleton . Last))-      ~> M.aggregateByTick @(GroupedBy (Last M.HeapProfSampleData))-      ~> mapping (concatMap (M.heapProfSamples . getLast) . DG.elems)--processHeapProfSample ::-  (MonadIO m) =>-  Config ->-  Verbosity ->-  Maybe HeapProfBreakdown ->-  ProcessT m (Tick (WithStartTime Event)) (DList OM.Metric)-processHeapProfSample config verbosity maybeHeapProfBreakdown =-  runIf (C.processorEnabled (.metrics) (.heapProfSample) config) $-    M.liftTick (M.processHeapProfSampleData verbosity maybeHeapProfBreakdown)-      ~> aggregate heapProfSampleDataAggregators (C.processorAggregationStrategy (.metrics) (.heapProfSample) config)-      ~> mapping (fmap toNumberDataPoint)-      ~> asGauge-      ~> asMetricWith config (.heapProfSample) [OM.unit .~ "By"]-      ~> mapping D.singleton------------------------------------------------------------------------------------- Machines------------------------------------------------------------------------------------- 2025-09-22:--- Once `cabal2nix` supports Cabal 3.12, this can once again use the value from:--- `PackageInfo_eventlog_live_otelcol.name`.-eventlogLiveName :: String-eventlogLiveName = "eventlog-live-otelcol"--eventlogLiveScope :: OC.InstrumentationScope-eventlogLiveScope =-  messageWith-    [ OC.name .~ T.pack eventlogLiveName-    , OC.version .~ T.pack (showVersion EventlogLive.version)-    ]--asExportMetricsServiceRequest :: Process [OM.ResourceMetrics] OMS.ExportMetricsServiceRequest-asExportMetricsServiceRequest = mapping $ (defMessage &) . (OM.resourceMetrics .~)--asExportTracesServiceRequest :: Process [OT.ResourceSpans] OTS.ExportTraceServiceRequest-asExportTracesServiceRequest = mapping $ (defMessage &) . (OTS.resourceSpans .~)--asExportMetricServiceRequest :: Process OM.ResourceMetrics OMS.ExportMetricsServiceRequest-asExportMetricServiceRequest = mapping (: []) ~> asExportMetricsServiceRequest--asExportTraceServiceRequest :: Process OT.ResourceSpans OTS.ExportTraceServiceRequest-asExportTraceServiceRequest = mapping (: []) ~> asExportTracesServiceRequest--asResourceMetrics :: [OM.ResourceMetrics -> OM.ResourceMetrics] -> Process [OM.ScopeMetrics] OM.ResourceMetrics-asResourceMetrics mod = mapping $ \scopeMetrics ->-  messageWith ((OM.scopeMetrics .~ scopeMetrics) : mod)--asResourceSpans :: [OT.ResourceSpans -> OT.ResourceSpans] -> Process [OT.ScopeSpans] OT.ResourceSpans-asResourceSpans mod = mapping $ \scopeSpans ->-  messageWith ((OT.scopeSpans .~ scopeSpans) : mod)--asResourceSpan :: [OT.ResourceSpans -> OT.ResourceSpans] -> Process OT.ScopeSpans OT.ResourceSpans-asResourceSpan mod = mapping (: []) ~> asResourceSpans mod--asResourceMetric :: [OM.ResourceMetrics -> OM.ResourceMetrics] -> Process OM.ScopeMetrics OM.ResourceMetrics-asResourceMetric mod = mapping (: []) ~> asResourceMetrics mod--asScopeSpans :: [OT.ScopeSpans -> OT.ScopeSpans] -> Process [OT.Span] OT.ScopeSpans-asScopeSpans mod = mapping $ \spans ->-  messageWith ((OT.spans .~ spans) : mod)--asScopeMetrics :: [OM.ScopeMetrics -> OM.ScopeMetrics] -> Process [OM.Metric] OM.ScopeMetrics-asScopeMetrics mod = mapping $ \metrics ->-  messageWith ((OM.metrics .~ metrics) : mod)--asMetricWith ::-  (Default a, HasField "description" a (Maybe Text), HasField "name" a Text) =>-  Config ->-  (C.Metrics -> Maybe a) ->-  [OM.Metric -> OM.Metric] ->-  Process OM.Metric'Data OM.Metric-asMetricWith config field mod =-  asMetric $-    [ OM.name .~ C.processorName (.metrics) field config-    , maybe id (OM.description .~) $ C.processorDescription (.metrics) field config-    ]-      <> mod--asMetric :: [OM.Metric -> OM.Metric] -> Process OM.Metric'Data OM.Metric-asMetric mod = mapping $ toMetric mod--toMetric :: [OM.Metric -> OM.Metric] -> OM.Metric'Data -> OM.Metric-toMetric mod metric'data = messageWith ((OM.maybe'data' .~ Just metric'data) : mod)--asGauge :: Process [OM.NumberDataPoint] OM.Metric'Data-asGauge =-  repeatedly $-    await >>= \case-      dataPoints-        | null dataPoints -> pure ()-        | otherwise -> yield . OM.Metric'Gauge . messageWith $ [OM.dataPoints .~ dataPoints]--asSum :: [OM.Sum -> OM.Sum] -> Process [OM.NumberDataPoint] OM.Metric'Data-asSum mod = repeatedly $ await >>= \dataPoints -> for_ (toSum mod dataPoints) yield--toSum :: [OM.Sum -> OM.Sum] -> [OM.NumberDataPoint] -> Maybe OM.Metric'Data-toSum mod dataPoints-  | null dataPoints = Nothing-  | otherwise = Just . OM.Metric'Sum . messageWith $ (OM.dataPoints .~ dataPoints) : mod------------------------------------------------------------------------------------- Interpret data---------------------------------------------------------------------------------------------------------------------------------------------------------------------- Interpret spans--class AsSpan v where-  -- | The `Key` type is used to index a `HashMap` in the default definition of `asSpan`.-  type Key v--  -- | The `toKey` function extracts a `Key` from the input value.-  toKey ::-    -- | The input value.-    v ->-    Key v--  toSpan ::-    -- | The configuration.-    Config ->-    -- | The input value.-    v ->-    -- | The trace ID.-    ByteString ->-    -- | The span ID.-    ByteString ->-    OT.Span--  -- | The `asSpan` machine processes values @v@ into OpenTelemetry spans `OT.Span`.-  asSpan :: (MonadIO m) => Config -> ProcessT m v OT.Span-  default asSpan :: (MonadIO m, Hashable (Key v)) => Config -> ProcessT m v OT.Span-  asSpan config = construct $ go (mempty, Nothing)-   where-    -- go :: (HashMap (Key v) ByteString, Maybe StdGen) -> PlanT (Is v) OT.Span m Void-    go (traceIds, maybeGen) = do-      -- Ensure the StdGen is initialised-      gen0 <- maybe (liftIO initStdGen) pure maybeGen-      -- Receive the next value-      i <- await-      -- Ensure the next value has a trace ID-      let ensureTraceId :: Maybe ByteString -> ((ByteString, StdGen), Maybe ByteString)-          ensureTraceId = wrap . maybe (uniformByteString 16 gen0) (,gen0)-           where-            wrap out@(traceId, _gen) = (out, Just traceId)-      let ((traceId, gen1), traceIds') = M.alterF ensureTraceId (toKey i) traceIds-      -- Ensure the next value has a span ID-      let (spanId, gen2) = uniformByteString 8 gen1-      -- Yield a span-      yield $ toSpan config i traceId spanId-      -- Continue-      go (traceIds', Just gen2)------------------------------------------------------------------------------------- Interpret capability usage spans--instance AsSpan CapabilityUsageSpan where-  type Key CapabilityUsageSpan = Int--  toKey :: CapabilityUsageSpan -> Int-  toKey = (.cap)--  toSpan :: Config -> CapabilityUsageSpan -> ByteString -> ByteString -> OT.Span-  toSpan config i traceId spanId =-    messageWith-      [ OT.traceId .~ traceId-      , OT.spanId .~ spanId-      , OT.name .~ C.processorName (.spans) (.capabilityUsage) config <> " " <> M.showCapabilityUserCategory user-      , OT.kind .~ OT.Span'SPAN_KIND_INTERNAL-      , OT.startTimeUnixNano .~ i.startTimeUnixNano-      , OT.endTimeUnixNano .~ i.endTimeUnixNano-      , OT.attributes-          .~ mapMaybe-            toMaybeKeyValue-            [ "capability" ~= i.cap-            , "user" ~= user-            ]-      , OT.status-          .~ messageWith-            [ OT.code .~ OT.Status'STATUS_CODE_OK-            ]-      ]-   where-    user = M.capabilityUser i------------------------------------------------------------------------------------- Interpret thread state spans--instance AsSpan ThreadStateSpan where-  type Key ThreadStateSpan = ThreadId--  toKey :: ThreadStateSpan -> ThreadId-  toKey = (.thread)--  toSpan :: Config -> ThreadStateSpan -> ByteString -> ByteString -> OT.Span-  toSpan config i traceId spanId =-    messageWith-      [ OT.traceId .~ traceId-      , OT.spanId .~ spanId-      , OT.name .~ C.processorName (.spans) (.threadState) config <> " " <> M.showThreadStateCategory i.threadState-      , OT.kind .~ OT.Span'SPAN_KIND_INTERNAL-      , OT.startTimeUnixNano .~ i.startTimeUnixNano-      , OT.endTimeUnixNano .~ i.endTimeUnixNano-      , OT.attributes-          .~ mapMaybe-            toMaybeKeyValue-            [ "capability" ~= M.threadStateCap i.threadState-            , "status" ~= (show <$> M.threadStateStatus i.threadState)-            ]-      , OT.status-          .~ messageWith-            [ OT.code .~ OT.Status'STATUS_CODE_OK-            ]-      ]------------------------------------------------------------------------------------- Interpret metrics--class IsNumberDataPoint'Value v where-  toNumberDataPoint'Value :: v -> OM.NumberDataPoint'Value--instance IsNumberDataPoint'Value Double where-  toNumberDataPoint'Value :: Double -> OM.NumberDataPoint'Value-  toNumberDataPoint'Value = OM.NumberDataPoint'AsDouble--instance IsNumberDataPoint'Value Word32 where-  toNumberDataPoint'Value :: Word32 -> OM.NumberDataPoint'Value-  toNumberDataPoint'Value = OM.NumberDataPoint'AsInt . fromIntegral---- | __Warning__: This instance may cause overflow.-instance IsNumberDataPoint'Value Word64 where-  toNumberDataPoint'Value :: Word64 -> OM.NumberDataPoint'Value-  toNumberDataPoint'Value = OM.NumberDataPoint'AsInt . fromIntegral--toNumberDataPoint :: (IsNumberDataPoint'Value v) => Metric v -> OM.NumberDataPoint-toNumberDataPoint i =-  messageWith-    [ OM.maybe'value .~ Just (toNumberDataPoint'Value i.value)-    , OM.timeUnixNano .~ fromMaybe 0 i.maybeTimeUnixNano-    , OM.startTimeUnixNano .~ fromMaybe 0 i.maybeStartTimeUnixNano-    , OM.attributes .~ mapMaybe toMaybeKeyValue (toList i.attrs)-    ]--toMaybeKeyValue :: Attr -> Maybe OC.KeyValue-toMaybeKeyValue (k, v) =-  toMaybeAnyValue v <&> \v ->-    messageWith-      [ OC.key .~ k-      , OC.value .~ v-      ]--toMaybeAnyValue :: AttrValue -> Maybe OC.AnyValue-toMaybeAnyValue = \case-  AttrInt v -> Just $ messageWith [OC.intValue .~ fromIntegral v]-  AttrInt8 v -> Just $ messageWith [OC.intValue .~ fromIntegral v]-  AttrInt16 v -> Just $ messageWith [OC.intValue .~ fromIntegral v]-  AttrInt32 v -> Just $ messageWith [OC.intValue .~ fromIntegral v]-  AttrInt64 v -> Just $ messageWith [OC.intValue .~ v]-  AttrWord v -> Just $ messageWith [OC.intValue .~ fromIntegral v]-  AttrWord8 v -> Just $ messageWith [OC.intValue .~ fromIntegral v]-  AttrWord16 v -> Just $ messageWith [OC.intValue .~ fromIntegral v]-  AttrWord32 v -> Just $ messageWith [OC.intValue .~ fromIntegral v]-  AttrWord64 v -> Just $ messageWith [OC.intValue .~ fromIntegral v]-  AttrDouble v -> Just $ messageWith [OC.doubleValue .~ v]-  AttrText v -> Just $ messageWith [OC.stringValue .~ v]-  AttrNull -> Nothing------------------------------------------------------------------------------------- DSL for writing messages---- | Construct a message with a list of modifications applied.-messageWith :: (Message msg) => [msg -> msg] -> msg-messageWith = foldr ($) defMessage------------------------------------------------------------------------------------- Options-----------------------------------------------------------------------------------options :: O.ParserInfo Options-options =-  O.info-    ( optionsParser-        O.<**> defaultsPrinter-        O.<**> OE.helperWith (O.long "help" <> O.help "Show this help text.")-        O.<**> OC.simpleVersioner (showVersion EventlogLive.version)-    )-    O.idm--data Options = Options-  { eventlogSocket :: EventlogSource-  , eventlogSocketTimeout :: Double-  , eventlogSocketTimeoutExponent :: Double-  , batchInterval :: Int-  , maybeEventlogLogFile :: Maybe FilePath-  , maybeHeapProfBreakdown :: Maybe HeapProfBreakdown-  , maybeServiceName :: Maybe ServiceName-  , verbosity :: Verbosity-  , stats :: Bool-  , maybeConfigFile :: Maybe FilePath-  , openTelemetryCollectorOptions :: OpenTelemetryCollectorOptions-  , myDebugOptions :: MyDebugOptions-  }--optionsParser :: O.Parser Options-optionsParser =-  Options-    <$> eventlogSourceParser-    <*> eventlogSocketTimeoutParser-    <*> eventlogSocketTimeoutExponentParser-    <*> batchIntervalParser-    <*> O.optional eventlogLogFileParser-    <*> O.optional heapProfBreakdownParser-    <*> O.optional serviceNameParser-    <*> verbosityParser-    <*> statsParser-    <*> O.optional configFileParser-    <*> openTelemetryCollectorOptionsParser-    <*> myDebugOptionsParser------------------------------------------------------------------------------------- Debug Options--data MyDebugOptions = MyDebugOptions-  { maybeMyEventlogSocket :: Maybe MyEventlogSocket-  , maybeMyGhcDebugSocket :: Maybe MyGhcDebugSocket-  }--myDebugOptionsParser :: O.Parser MyDebugOptions-myDebugOptionsParser =-  OC.parserOptionGroup "Debug Options" $-    MyDebugOptions-      <$> O.optional myEventlogSocketParser-      <*> maybeMyGhcDebugSocketParser------------------------------------------------------------------------------------- My Eventlog Socket--newtype MyEventlogSocket-  = MyEventlogSocketUnix FilePath--myEventlogSocketParser :: O.Parser MyEventlogSocket-myEventlogSocketParser =-  MyEventlogSocketUnix-    <$> O.strOption-      ( O.long "enable-my-eventlog-socket-unix"-          <> O.metavar "SOCKET"-          <> O.help "Enable the eventlog socket for this program on the given Unix socket."-      )--{- |-Set @eventlog-socket@ as the eventlog writer.--}-withMyEventlogSocket :: Maybe MyEventlogSocket -> IO ()-withMyEventlogSocket maybeMyEventlogSocket =-  for_ maybeMyEventlogSocket $ \(MyEventlogSocketUnix myEventlogSocket) ->-    Eventlog.Socket.startWait myEventlogSocket------------------------------------------------------------------------------------- Configuration--configFileParser :: O.Parser FilePath-configFileParser =-  O.strOption-    ( O.long "config"-        <> O.metavar "FILE"-        <> O.help "The path to a detailed configuration file."-    )--defaultsPrinter :: O.Parser (a -> a)-defaultsPrinter =-  O.infoOption defaultConfigString . mconcat $-    [ O.long "print-defaults"-    , O.help "Print default configuration options that can be used in config.yaml"-    ]------------------------------------------------------------------------------------- Service Name--newtype ServiceName = ServiceName {serviceName :: Text}--serviceNameParser :: O.Parser ServiceName-serviceNameParser =-  ServiceName-    <$> O.strOption-      ( O.long "service-name"-          <> O.metavar "STRING"-          <> O.help "The name of the profiled service."-      )------------------------------------------------------------------------------------- OpenTelemetry Collector configuration--newtype OpenTelemetryCollectorOptions = OpenTelemetryCollectorOptions-  { openTelemetryCollectorServer :: G.Server-  }--openTelemetryCollectorOptionsParser :: O.Parser OpenTelemetryCollectorOptions-openTelemetryCollectorOptionsParser =-  OC.parserOptionGroup "OpenTelemetry Collector Server Options" $-    OpenTelemetryCollectorOptions-      <$> otelcolServerParser--otelcolServerParser :: O.Parser G.Server-otelcolServerParser =-  makeServer-    <$> otelcolAddressParser-    <*> O.switch (O.long "otelcol-ssl" <> O.help "Use SSL.")-    <*> otelcolServerValidationParser-    <*> otelcolSslKeyLogParser- where-  makeServer :: G.Address -> Bool -> G.ServerValidation -> G.SslKeyLog -> G.Server-  makeServer address ssl serverValidation sslKeyLog-    | ssl = G.ServerSecure serverValidation sslKeyLog address-    | otherwise = G.ServerInsecure address--otelcolAddressParser :: O.Parser G.Address-otelcolAddressParser =-  G.Address-    <$> O.strOption-      ( O.long "otelcol-host"-          <> O.metavar "HOST"-          <> O.help "Server hostname."-      )-    <*> O.option-      O.auto-      ( O.long "otelcol-port"-          <> O.metavar "PORT"-          <> O.help "Server TCP port."-          <> O.value 4317-      )-    <*> O.optional-      ( O.strOption-          ( O.long "otelcol-authority"-              <> O.metavar "HOST"-              <> O.help "Server authority."-          )-      )--otelcolServerValidationParser :: O.Parser G.ServerValidation-otelcolServerValidationParser =-  asum-    [ G.ValidateServer <$> otelcolCertificateStoreSpecParser-    , pure G.NoServerValidation-    ]- where-  otelcolCertificateStoreSpecParser :: O.Parser G.CertificateStoreSpec-  otelcolCertificateStoreSpecParser =-    makeCertificateStoreSpec-      <$> O.optional-        ( O.strOption-            ( O.long "otelcol-certificate-store"-                <> O.metavar "FILE"-                <> O.help "Store for certificate validation."-            )-        )-   where-    makeCertificateStoreSpec :: Maybe FilePath -> G.CertificateStoreSpec-    makeCertificateStoreSpec = maybe G.certStoreFromSystem G.certStoreFromPath--otelcolSslKeyLogParser :: O.Parser G.SslKeyLog-otelcolSslKeyLogParser =-  asum-    [ G.SslKeyLogPath-        <$> O.strOption-          ( O.long "otelcol-ssl-key-log"-              <> O.metavar "FILE"-              <> O.help "Use file to log SSL keys."-          )-    , O.flag-        G.SslKeyLogNone-        G.SslKeyLogFromEnv-        ( O.long "otelcol-ssl-key-log-from-env"-            <> O.help "Use SSLKEYLOGFILE to log SSL keys."-        )-    ]------------------------------------------------------------------------------------ Internal Helpers----------------------------------------------------------------------------------runIf :: (Monad m) => Bool -> MachineT m k o -> MachineT m k o-runIf b m = if b then m else stopped+{-# OPTIONS_GHC -Wno-name-shadowing -fconstraint-solver-iterations=0 #-}++{- |+Module      : GHC.Eventlog.Live.Otelcol+Description : The implementation of @eventlog-live-otelcol@.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Otelcol (+  main,+) where++import Control.Concurrent.STM.TChan (newTChanIO)+import Control.Exception (bracket_)+import Data.DList (DList)+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.Text (Text)+import Data.Text qualified as T+import Data.Version (showVersion)+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.Logs (exportResourceLogs)+import GHC.Eventlog.Live.Otelcol.Exporter.Metrics (exportResourceMetrics)+import GHC.Eventlog.Live.Otelcol.Exporter.Profiles (exportResourceProfiles)+import GHC.Eventlog.Live.Otelcol.Exporter.Traces (exportResourceSpans)+import GHC.Eventlog.Live.Otelcol.Options+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.Profiles (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.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 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+import Proto.Opentelemetry.Proto.Metrics.V1.Metrics qualified as OM+import Proto.Opentelemetry.Proto.Metrics.V1.Metrics_Fields qualified as OM+import Proto.Opentelemetry.Proto.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++{- |+The main function for @eventlog-live-otelcol@.+-}+main :: IO ()+main = do+  Options{..} <- O.execParser options++  -- Construct the logging action+  myTelemetryDataChan <- newTChanIO+  let logger =+        M.filterBySeverity severityThreshold $+          M.stderrLogger <> M.chanLogger myTelemetryDataChan++  -- Instument THIS PROGRAM with eventlog-socket and/or ghc-debug.+  let MyDebugOptions{..} = myDebugOptions+  startMyEventlogSocket logger maybeMyEventlogSocket+  withMyGhcDebug logger maybeMyGhcDebugSocket $ do+    --+    -- Start the control server.+    controlServerApi <- startControlServer logger controlOptions++    -- Read the configuration file.+    let readConfigFile configFile = do+          writeLog logger DEBUG $+            "Reading configuration file from " <> T.pack configFile+          config <- C.readConfigFile logger configFile+          writeLog logger DEBUG $+            "Configuration file:\n" <> C.prettyConfig config+          pure config++    -- Read the configuration file and add derived settings.+    fullConfig <-+      C.toFullConfig eventlogFlushIntervalS+        <$> maybe (pure def) readConfigFile maybeConfigFile+    writeLog logger DEBUG $+      "Batch interval is " <> T.pack (show fullConfig.batchIntervalMs) <> "ms"+    writeLog logger DEBUG $+      "Eventlog flush interval is " <> T.pack (show fullConfig.eventlogFlushIntervalX) <> "x"++    -- Determine the window size for statistics+    let windowSizeX =+          (10 *) . maximum $+            [ fullConfig.eventlogFlushIntervalX+            , C.maximumAggregationBatches fullConfig+            , C.maximumExportBatches fullConfig+            ]++    -- Resolve the service name (or use a default).+    let serviceName :: ServiceName+        serviceName = fromMaybe (ServiceName "undefined") maybeServiceName++    -- Create a resource to represent the monitored process.+    let eventlogResource :: OR.Resource+        eventlogResource =+          messageWith+            [ OM.attributes+                .~ mapMaybe+                  toMaybeKeyValue+                  [ "service.name" ~= AttrText serviceName.serviceName+                  ]+            ]++    -- Create machine that processes eventlog data into telemetry data+    let processEventlogTelemetry :: ProcessT IO (Tick Event) (Tick ResourceTelemetryData)+        processEventlogTelemetry =+          M.liftTick M.withStartTime+            ~> M.fanoutTick+              [ -- Process the heap events.+                processHeapEvents logger maybeHeapProfBreakdown fullConfig+                  ~> mapping (fmap (fmap TelemetryData'Metric))+              , -- Process the log events.+                processLogEvents fullConfig+                  ~> mapping (fmap (fmap TelemetryData'Log))+              , -- Process the thread events.+                processThreadEvents logger fullConfig+                  ~> mapping (fmap (fmap (either TelemetryData'Metric TelemetryData'Span)))+              , -- Process the profile events.+                processProfileEvents logger fullConfig+                  ~> mapping (fmap (fmap TelemetryData'Profile))+              ]+            ~> M.liftTick (asResourceTelemetryData eventlogResource eventlogLiveScope)++    -- Create a resource to represent the eventlog-live process.+    let internalResource :: OR.Resource+        internalResource =+          messageWith+            [ OM.attributes+                .~ mapMaybe+                  toMaybeKeyValue+                  [ "service.name" ~= AttrText (eventlogLiveName <> "-for-" <> serviceName.serviceName)+                  , "service.version" ~= eventlogLiveVersion+                  ]+            ]++    -- Create the machine that processes internal telemetry data+    --+    -- NOTE: This process only takes a stream of inputs to use their tick.+    let processInternalTelemetry :: ProcessT IO (Tick x) (Tick ResourceTelemetryData)+        processInternalTelemetry =+          M.mergeWithTickCC (M.chanSource myTelemetryDataChan)+            ~> processInternalTelemetryData fullConfig+            ~> M.liftTick (asResourceTelemetryData internalResource eventlogLiveScope)++    -- Create the full machine to process eventlog data.+    let processAndExportTelemetry conn =+          M.fanoutTick+            [ -- Log a warning if no input has been received after 10 ticks.+              M.validateInput logger 10+            , -- Count the number of input events between each tick.+              eventCountTick+                ~> mapping (fmap (D.singleton . EventCountStat))+            , -- Process eventlog and internal telemetry...+              M.fanoutTickCC+                [ processEventlogTelemetry ~> mapping (fmap D.singleton)+                , processInternalTelemetry ~> mapping (fmap D.singleton)+                ]+                ~> M.liftTick asParts+                -- ...and export it.+                ~> exportResourceTelemetryData fullConfig conn+            ]+            -- Process the statistics+            -- TODO: windowSize should be the maximum of all aggregation and export intervals+            ~> M.liftTick (asParts ~> processStats logger stats eventlogFlushIntervalS windowSizeX)+            -- Validate the consistency of the tick+            ~> M.validateTicks logger+            ~> 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+              logger+              eventlogSourceHandle+              fullConfig.batchIntervalMs+              Nothing+              maybeEventlogLogFile+              (processAndExportTelemetry conn)++data TelemetryData+  = TelemetryData'Log OL.LogRecord+  | TelemetryData'Metric OM.Metric+  | TelemetryData'Span OT.Span+  | TelemetryData'Profile M.CallStackData++data ResourceTelemetryData+  = ResourceTelemetryData'Log OL.ResourceLogs+  | ResourceTelemetryData'Metric OM.ResourceMetrics+  | ResourceTelemetryData'Span OT.ResourceSpans+  | ResourceTelemetryData'Profile OP.ProfilesData++{- |+Internal helper.+Export resource telemetry data and yield statistics.+-}+exportResourceTelemetryData ::+  FullConfig ->+  G.Connection ->+  ProcessT IO (Tick ResourceTelemetryData) (Tick (DList Stat))+exportResourceTelemetryData fullConfig connection =+  M.fanoutTick+    [ -- Export logs.+      runIf (C.shouldExportLogs fullConfig) $+        M.liftTick (mapping getResourceLogs ~> asParts ~> mapping D.singleton)+          -- NOTE: This is required to combine different resource telemetry+          --       streams. However, it has the "unfortunate" side-effect of+          --       making it impossible to not batch once per interval.+          ~> M.batchByTick+          ~> M.liftTick (mapping (toExportLogsServiceRequest . D.toList))+          ~> exportResourceLogs connection+          ~> M.liftTick (mapping (D.singleton . ExportLogsResultStat))+    , -- Export metrics.+      runIf (C.shouldExportMetrics fullConfig) $+        M.liftTick (mapping getResourceMetrics ~> asParts ~> mapping D.singleton)+          -- NOTE: See note above.+          ~> M.batchByTick+          ~> M.liftTick (mapping (toExportMetricsServiceRequest . D.toList))+          ~> exportResourceMetrics connection+          ~> M.liftTick (mapping (D.singleton . ExportMetricsResultStat))+    , -- Export spans.+      runIf (C.shouldExportTraces fullConfig) $+        M.liftTick (mapping getResourceSpans ~> asParts ~> mapping D.singleton)+          -- NOTE: See note above.+          ~> M.batchByTick+          ~> M.liftTick (mapping (toExportTracesServiceRequest . D.toList))+          ~> exportResourceSpans connection+          ~> M.liftTick (mapping (D.singleton . ExportTraceResultStat))+    , -- Export profiles.+      runIf (C.shouldExportProfiles fullConfig) $+        M.liftTick (mapping getResourceProfiles ~> asParts ~> mapping toExportProfileServiceRequest)+          ~> exportResourceProfiles connection+          ~> M.liftTick (mapping (D.singleton . ExportProfileResultStat))+    ]++getResourceLogs :: ResourceTelemetryData -> Maybe OL.ResourceLogs+getResourceLogs = \case+  (ResourceTelemetryData'Log resourceLogs) -> Just resourceLogs+  _otherwise -> Nothing++getResourceMetrics :: ResourceTelemetryData -> Maybe OM.ResourceMetrics+getResourceMetrics = \case+  (ResourceTelemetryData'Metric resourceMetrics) -> Just resourceMetrics+  _otherwise -> Nothing++getResourceSpans :: ResourceTelemetryData -> Maybe OT.ResourceSpans+getResourceSpans = \case+  (ResourceTelemetryData'Span resourceSpans) -> Just resourceSpans+  _otherwise -> Nothing++getResourceProfiles :: ResourceTelemetryData -> Maybe OP.ProfilesData+getResourceProfiles = \case+  (ResourceTelemetryData'Profile profilesData) -> Just profilesData+  _otherwise -> Nothing++{- |+Internal helper.+Repack a stream of `TelemetryData` to batched `ResourceTelemetryData`.+-}+asResourceTelemetryData ::+  (Foldable f) =>+  OR.Resource ->+  OC.InstrumentationScope ->+  Process (f TelemetryData) ResourceTelemetryData+asResourceTelemetryData resource instrumentationScope =+  mapping (toResourceTelemetryData . F.toList) ~> asParts+ where+  toResourceTelemetryData ::+    [TelemetryData] ->+    [ResourceTelemetryData]+  toResourceTelemetryData telemetryData =+    catMaybes [maybeResourceLogs, maybeResourceMetrics, maybeResourceSpans, maybeProfiles]+   where+    (logRecords, metrics, spans, profiles) = partitionTelemetryData telemetryData++    maybeResourceLogs = do+      scopeLogs <- toScopeLogs instrumentationScope logRecords+      resourceLogs <- toResourceLogs resource [scopeLogs]+      pure $ ResourceTelemetryData'Log resourceLogs+    maybeResourceMetrics = do+      scopeMetrics <- toScopeMetrics instrumentationScope metrics+      resourceMetrics <- toResourceMetrics resource [scopeMetrics]+      pure $ ResourceTelemetryData'Metric resourceMetrics+    maybeResourceSpans = do+      scopeSpans <- toScopeSpans instrumentationScope spans+      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]+            ]++{- |+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 = 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++{- |+Internal helper.+Process internal telemetry data.+-}+processInternalTelemetryData ::+  FullConfig ->+  Process (Tick MyTelemetryData) (Tick (DList TelemetryData))+processInternalTelemetryData fullConfig =+  M.fanoutTick+    [ -- Process internal log messages.+      M.liftTick (mapping getMyLogRecord ~> asParts ~> mapping (D.singleton . TelemetryData'Log . toLogRecord))+        ~> M.batchByTicks (C.processorExportBatches (.logs) (.internalLogMessage) fullConfig)+        -- TODO: Any internal metrics should be processed below.+    ]++getMyLogRecord :: MyTelemetryData -> Maybe LogRecord+getMyLogRecord = \case+  M.MyTelemetryData'LogRecord{..} -> Just logRecord+  M.MyTelemetryData'Metric{} -> Nothing++--------------------------------------------------------------------------------+-- Instrumentation Scope+--------------------------------------------------------------------------------++-- 2025-09-22:+-- Once `cabal2nix` supports Cabal 3.12, this can once again use the value from:+-- `PackageInfo_eventlog_live_otelcol.name`.+eventlogLiveName :: Text+eventlogLiveName = "eventlog-live-otelcol"++eventlogLiveVersion :: Text+eventlogLiveVersion = T.pack (showVersion EventlogLive.version)++eventlogLiveScope :: OC.InstrumentationScope+eventlogLiveScope =+  messageWith+    [ OC.name .~ eventlogLiveName+    , OC.version .~ eventlogLiveVersion+    ]
src/GHC/Eventlog/Live/Otelcol/Config.hs view
@@ -9,12 +9,34 @@ Portability : portable -} module GHC.Eventlog.Live.Otelcol.Config (-  readConfig,+  -- * Configuration type+  ServiceName (..),   Config (..),+  readConfigFile,+  prettyConfig,+  FullConfig (..),+  toFullConfig,++  -- ** Processor configuration types   Processors (..),+  IsProcessorConfig,+  processorEnabled,+  processorDescription,+  processorName,++  -- *** Log processor configuration types+  Logs (..),+  IsLogProcessorConfig,+  shouldExportLogs,+  ThreadLabel (..),+  UserMarker (..),+  UserMessage (..),+  InternalLogMessage (..),++  -- *** Metric processor configuration types   Metrics (..),-  Spans (..),-  AggregationStrategy (..),+  IsMetricProcessorConfig,+  shouldExportMetrics,   HeapAllocatedMetric (..),   BlocksSizeMetric (..),   HeapSizeMetric (..),@@ -24,30 +46,117 @@   MemReturnedMetric (..),   HeapProfSampleMetric (..),   CapabilityUsageMetric (..),++  -- *** Trace processor configuration types+  Traces (..),+  IsTraceProcessorConfig,+  shouldExportTraces,   CapabilityUsageSpan (..),   ThreadStateSpan (..),-  processorEnabled,-  processorDescription,-  processorName,++  -- *** Profiler processor configuration types+  Profiles (..),+  IsProfileProcessorConfig,+  shouldExportProfiles,+  StackSampleProfile (..),+  CostCentreSampleProfile (..),++  -- ** Property types++  -- *** Aggregation strategy+  AggregationStrategy (..),+  toAggregationBatches,   processorAggregationStrategy,+  processorAggregationBatches,+  maximumAggregationBatches,++  -- *** Export strategy+  ExportStrategy (..),+  toExportBatches,+  processorExportStrategy,+  processorExportBatches,+  maximumExportBatches,++  -- *** Batch interval+  toBatchIntervalMs,+  toBatches, ) where +import Control.Exception (assert)+import Control.Monad ((<=<)) import Control.Monad.IO.Class (MonadIO (..))+import Data.ByteString.Lazy qualified as BSL import Data.Default (Default (..))-import Data.Maybe (fromMaybe)+import Data.Hashable (Hashable)+import Data.List.NonEmpty (NonEmpty (..))+import Data.Maybe (catMaybes, fromMaybe, mapMaybe) import Data.Monoid (Any (..), First (..))+import Data.Semigroup (Semigroup (..)) import Data.Text (Text)-import Data.Yaml qualified as Y+import Data.Text qualified as T+import Data.Text.Encoding qualified as TE+import Data.YAML qualified as YAML+import GHC.Eventlog.Live.Data.Severity (Severity (..))+import GHC.Eventlog.Live.Logger (Logger, writeLog) import GHC.Eventlog.Live.Otelcol.Config.Default (defaultConfig, getDefault) import GHC.Eventlog.Live.Otelcol.Config.Types import GHC.Records (HasField)+import GHC.Stack.Types (HasCallStack)+import System.Exit (exitFailure)  {- |+An OpenTelemetry service name.+-}+newtype ServiceName = ServiceName {serviceName :: Text}+  deriving newtype (Eq, Hashable)++{- | Read a `Config` from a configuration file. -}-readConfig :: (MonadIO m) => FilePath -> m Config-readConfig = Y.decodeFileThrow+readConfigFile ::+  Logger IO ->+  FilePath ->+  IO Config+readConfigFile logger filePath =+  readConfig logger =<< liftIO (BSL.readFile filePath) +{- |+Read a `Config` from a `BSL.ByteString`.+-}+readConfig ::+  Logger IO ->+  BSL.ByteString ->+  IO Config+readConfig logger fileContents = do+  case YAML.decode1 fileContents of+    Left (pos, errorMessage) -> do+      writeLog logger FATAL $+        T.pack $+          YAML.prettyPosWithSource pos fileContents " error" <> errorMessage+      liftIO exitFailure+    Right config -> pure config++{- |+Pretty-print a `Config` to YAML.+-}+prettyConfig :: Config -> Text+prettyConfig = TE.decodeUtf8Lenient . BSL.toStrict . YAML.encode1++{- |+Create a full configuration.+-}+toFullConfig ::+  -- | The @--eventlog-flush-interval@ in seconds.+  Double ->+  -- | The user configuration.+  Config ->+  FullConfig+toFullConfig eventlogFlushIntervalS config =+  FullConfig{..}+ where+  batchIntervalMs = toBatchIntervalMs eventlogFlushIntervalS config+  eventlogFlushIntervalX = toBatches batchIntervalMs eventlogFlushIntervalS+ ------------------------------------------------------------------------------- -- Default Instances -------------------------------------------------------------------------------@@ -60,14 +169,44 @@   def :: Processors   def = $(getDefault @'["processors"] defaultConfig) +instance Default Logs where+  def :: Logs+  def = $(getDefault @'["processors", "logs"] defaultConfig)+ instance Default Metrics where   def :: Metrics   def = $(getDefault @'["processors", "metrics"] defaultConfig) -instance Default Spans where-  def :: Spans-  def = $(getDefault @'["processors", "spans"] defaultConfig)+instance Default Traces where+  def :: Traces+  def = $(getDefault @'["processors", "traces"] defaultConfig) +instance Default Profiles where+  def :: Profiles+  def = $(getDefault @'["processors", "profiles"] defaultConfig)++-- NOTE: This should be kept in sync with the list of logs.+--       Specifically, there should be a `Default` instance for every log.++instance Default ThreadLabel where+  def :: ThreadLabel+  def = $(getDefault @'["processors", "logs", "threadLabel"] defaultConfig)++instance Default UserMarker where+  def :: UserMarker+  def = $(getDefault @'["processors", "logs", "userMarker"] defaultConfig)++instance Default UserMessage where+  def :: UserMessage+  def = $(getDefault @'["processors", "logs", "userMessage"] defaultConfig)++instance Default InternalLogMessage where+  def :: InternalLogMessage+  def = $(getDefault @'["processors", "logs", "internalLogMessage"] defaultConfig)++-- NOTE: This should be kept in sync with the list of metrics.+--       Specifically, there should be a `Default` instance for every metric.+ instance Default HeapAllocatedMetric where   def :: HeapAllocatedMetric   def = $(getDefault @'["processors", "metrics", "heapAllocated"] defaultConfig)@@ -104,14 +243,25 @@   def :: CapabilityUsageMetric   def = $(getDefault @'["processors", "metrics", "capabilityUsage"] defaultConfig) +-- NOTE: This should be kept in sync with the list of traces.+--       Specifically, there should be a `Default` instance for every trace.+ instance Default CapabilityUsageSpan where   def :: CapabilityUsageSpan-  def = $(getDefault @'["processors", "spans", "capabilityUsage"] defaultConfig)+  def = $(getDefault @'["processors", "traces", "capabilityUsage"] defaultConfig)  instance Default ThreadStateSpan where   def :: ThreadStateSpan-  def = $(getDefault @'["processors", "spans", "threadState"] defaultConfig)+  def = $(getDefault @'["processors", "traces", "threadState"] defaultConfig) +instance Default StackSampleProfile where+  def :: StackSampleProfile+  def = $(getDefault @'["processors", "profiles", "stackSample"] defaultConfig)++instance Default CostCentreSampleProfile where+  def :: CostCentreSampleProfile+  def = $(getDefault @'["processors", "profiles", "costCentreSample"] defaultConfig)+ ------------------------------------------------------------------------------- -- Accessors -------------------------------------------------------------------------------@@ -123,10 +273,10 @@   (HasField "enabled" b Bool) =>   (Processors -> Maybe a) ->   (a -> Maybe b) ->-  Config ->+  FullConfig ->   Bool processorEnabled group field =-  getAny . with (.processors) (with group (with field (Any . (.enabled))))+  getAny . with (.processors) (with group (with field (Any . (.enabled)))) . (.config)  {- | Get the description corresponding to a processor.@@ -135,23 +285,34 @@   (Default b, HasField "description" b (Maybe Text)) =>   (Processors -> Maybe a) ->   (a -> Maybe b) ->-  Config ->+  FullConfig ->   Maybe Text processorDescription group field =-  (.description) . fromMaybe def . getFirst . with (.processors) (with group (First . field))+  (.description) . fromMaybe def . getFirst . with (.processors) (with group (First . field)) . (.config)  {- | Get the name corresponding to a processor.++__Warning:__ This assumes the value of @`def`.`name`@ is `Just` some `Text`. -} processorName ::-  (Default b, HasField "name" b Text) =>+  forall a b.+  (HasCallStack, Default b, HasField "name" b (Maybe Text)) =>   (Processors -> Maybe a) ->   (a -> Maybe b) ->-  Config ->+  FullConfig ->   Text processorName group field =-  (.name) . fromMaybe def . getFirst . with (.processors) (with group (First . field))+  fromMaybe defaultName . ((.name) <=< getFirst) . with (.processors) (with group (First . field)) . (.config)+ where+  defaultName :: (HasCallStack) => Text+  defaultName = case (def :: b).name of+    Nothing -> error "The default configuration for this metric has no name."+    Just name -> name +--------------------------------------------------------------------------------+-- Aggregation Strategy+ {- | Get the aggregation strategy corresponding to a metric processor. -}@@ -159,13 +320,351 @@   (Default b, HasField "aggregate" b (Maybe AggregationStrategy)) =>   (Processors -> Maybe a) ->   (a -> Maybe b) ->-  Config ->+  FullConfig ->   Maybe AggregationStrategy processorAggregationStrategy group field =-  (.aggregate) . fromMaybe def . getFirst . with (.processors) (with group (First . field))+  (.aggregate) . fromMaybe def . getFirst . with (.processors) (with group (First . field)) . (.config)  {- |+Convert an `AggregationStrategy` to a number of batches.++__Precondition:__+If the aggregation strategy is defined in seconds,+then the batch interval should divide this duration.+-}+toAggregationBatches ::+  -- | The batch interval in milliseconds.+  Int ->+  -- | The @--eventlog-flush-interval@ in /batches/.+  Int ->+  -- | The aggregation strategy.+  Maybe AggregationStrategy ->+  Int+toAggregationBatches batchIntervalMs eventlogFlushIntervalX = \case+  -- If the setting is '60s' this means 60 seconds.+  Just (AggregationStrategyDuration DurationBySeconds{..}) -> toMilli seconds `div` batchIntervalMs+  -- If the setting is '60x' this means 60 times the /eventlog flush interval/,+  -- not the interal batch interval.+  Just (AggregationStrategyDuration DurationByBatches{..}) -> batches * eventlogFlushIntervalX+  -- If the setting is 'true' this means '1x', i.e., /eventlog flush interval/.+  Just AggregationStrategyBool{..} | isOn -> eventlogFlushIntervalX+  -- If the setting is absent or 'false' this means /do not aggregate/.+  Nothing -> 0+  Just AggregationStrategyBool{..} -> assert (not isOn) 0++{- |+Get the aggregation strategy corresponding to a metric processor.+-}+processorAggregationBatches ::+  (Default b, HasField "aggregate" b (Maybe AggregationStrategy)) =>+  -- | The accessor for the sub-group of processors.+  (Processors -> Maybe a) ->+  -- | The accessor for the individual processor.+  (a -> Maybe b) ->+  -- | The full configuration.+  FullConfig ->+  Int+processorAggregationBatches group field fullConfig =+  toAggregationBatches fullConfig.batchIntervalMs fullConfig.eventlogFlushIntervalX $+    processorAggregationStrategy group field fullConfig++{- |+Get all aggregation strategies.+-}+allAggregationStrategies ::+  Config ->+  [AggregationStrategy]+allAggregationStrategies =+  catMaybes . with (.processors) (with (.metrics) (forEachMetricProcessor (.aggregate)))++{- |+Get the largest aggregation strategy in batches.+-}+maximumAggregationBatches ::+  FullConfig ->+  Int+maximumAggregationBatches fullConfig =+  maximum . fmap (toAggregationBatches fullConfig.batchIntervalMs fullConfig.eventlogFlushIntervalX . Just) $+    allAggregationStrategies fullConfig.config++--------------------------------------------------------------------------------+-- Export Strategy++{- |+Get the export strategy corresponding to a processor.+-}+processorExportStrategy ::+  (Default b, HasField "export" b (Maybe ExportStrategy)) =>+  (Processors -> Maybe a) ->+  (a -> Maybe b) ->+  FullConfig ->+  Maybe ExportStrategy+processorExportStrategy group field =+  (.export) . fromMaybe def . getFirst . with (.processors) (with group (First . field)) . (.config)++{- |+Convert an `ExportStrategy` to a number of batches.++__Precondition:__+If the export strategy is defined in seconds,+then the batch interval should divide this duration.+-}+toExportBatches ::+  -- | The batch interval in milliseconds.+  Int ->+  -- | The @--eventlog-flush-interval@ in /batches/.+  Int ->+  -- | The export strategy.+  Maybe ExportStrategy ->+  Int+toExportBatches batchIntervalMs eventlogFlushIntervalX = \case+  -- If the setting is '60s' this means 60 seconds.+  Just (ExportStrategyDuration DurationBySeconds{..}) -> toMilli seconds `div` batchIntervalMs+  -- If the setting is '60x' this means 60 times the /eventlog flush interval/,+  -- not the interal batch interval.+  Just (ExportStrategyDuration DurationByBatches{..}) -> batches * eventlogFlushIntervalX+  -- If the setting is 'true' this means '1x', i.e., /eventlog flush interval/.+  Just ExportStrategyBool{..} | isOn -> eventlogFlushIntervalX+  -- If the setting is absent or 'false' this means /do not aggregate/.+  Nothing -> 0+  Just ExportStrategyBool{..} -> assert (not isOn) 0++{- |+Get the export strategy corresponding to processor in batches.+-}+processorExportBatches ::+  (Default b, HasField "export" b (Maybe ExportStrategy)) =>+  (Processors -> Maybe a) ->+  (a -> Maybe b) ->+  FullConfig ->+  Int+processorExportBatches group field fullConfig =+  toExportBatches fullConfig.batchIntervalMs fullConfig.eventlogFlushIntervalX $+    processorExportStrategy group field fullConfig++{- |+Get all export strategies.+-}+allExportStrategies ::+  Config ->+  [ExportStrategy]+allExportStrategies =+  catMaybes . with (.processors) (forEachProcessor (.export))++{- |+Get the largest export strategy in batches.+-}+maximumExportBatches ::+  FullConfig ->+  Int+maximumExportBatches fullConfig =+  maximum . fmap (toExportBatches fullConfig.batchIntervalMs fullConfig.eventlogFlushIntervalX . Just) $+    allExportStrategies fullConfig.config++-------------------------------------------------------------------------------+-- Batch Interval++{- |+Get the batch interval such that all user-specified intervals can be respected.+-}+toBatchIntervalMs ::+  -- | The @--eventlog-flush-interval@.+  Double ->+  -- | The configuration.+  Config ->+  Int+toBatchIntervalMs eventlogFlushIntervalS config =+  (.getGCD) . sconcat . fmap GCD $+    eventlogFlushIntervalMs :| aggregationIntervalsMs <> exportIntervalsMs+ where+  -- TODO: Check if any intervals round to 0ms.+  eventlogFlushIntervalMs =+    toMilli eventlogFlushIntervalS+  aggregationIntervalsMs =+    mapMaybe (fmap toMilli . toAggregationSeconds) . allAggregationStrategies $ config+  exportIntervalsMs =+    mapMaybe (fmap toMilli . toExportSeconds) . allExportStrategies $ config++{- |+Get the relevant interval in batches.++__Precondition:__ The batch interval divides the relevant interval.+-}+toBatches ::+  -- | The batch interval in milliseconds.+  Int ->+  -- | The relevant interval in seconds.+  Double ->+  Int+toBatches batchIntervalMs intervalS =+  toMilli intervalS `div` batchIntervalMs++-------------------------------------------------------------------------------+-- Exporters+-------------------------------------------------------------------------------++shouldExportLogs :: FullConfig -> Bool+shouldExportLogs =+  getAny+    . with+      (.processors)+      ( with+          (.logs)+          (mconcat . forEachLogProcessor (Any . isEnabled . (.export)))+      )+    . (.config)++shouldExportMetrics :: FullConfig -> Bool+shouldExportMetrics =+  getAny+    . with+      (.processors)+      ( with+          (.metrics)+          (mconcat . forEachMetricProcessor (Any . isEnabled . (.export)))+      )+    . (.config)++shouldExportTraces :: FullConfig -> Bool+shouldExportTraces =+  getAny+    . with+      (.processors)+      ( with+          (.traces)+          (mconcat . forEachTraceProcessor (Any . isEnabled . (.export)))+      )+    . (.config)++shouldExportProfiles :: FullConfig -> Bool+shouldExportProfiles =+  getAny+    . with+      (.processors)+      ( with+          (.profiles)+          (mconcat . forEachProfileProcessor (Any . isEnabled . (.export)))+      )+    . (.config)++-------------------------------------------------------------------------------+-- Functors for processor configurations+-------------------------------------------------------------------------------++{- |+Apply a function to each processor.+-}+forEachProcessor ::+  ( forall processorConfig.+    (IsProcessorConfig processorConfig) =>+    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)+    ]++{- |+Apply a function to each metric processor.+-}+forEachLogProcessor ::+  ( forall traceProcessorConfig.+    (IsLogProcessorConfig traceProcessorConfig) =>+    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+  ]++{- |+Apply a function to each metric processor.+-}+forEachMetricProcessor ::+  ( forall metricProcessorConfig.+    (IsMetricProcessorConfig metricProcessorConfig) =>+    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+  ]++{- |+Apply a function to each metric processor.+-}+forEachTraceProcessor ::+  ( forall traceProcessorConfig.+    (IsTraceProcessorConfig traceProcessorConfig) =>+    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+  ]++{- |+Apply a function to each metric processor.+-}+forEachProfileProcessor ::+  ( forall profileProcessorConfig.+    (IsProfileProcessorConfig profileProcessorConfig) =>+    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+  ]++-------------------------------------------------------------------------------+-- Internal Helpers+-------------------------------------------------------------------------------++{- | Internal helper. -} with :: (Foldable f, Monoid r) => (s -> f t) -> (t -> r) -> s -> r with = flip ((.) . foldMap)++{- |+Internal helper.+Convert seconds to milliseconds.+-}+toMilli :: Double -> Int+toMilli seconds = round (seconds * 1_000)++{- |+Internal helper.+Wrapper that provides a `Semigroup` instance for `gcd`.+-}+newtype GCD a = GCD {getGCD :: a}++instance (Integral a) => Semigroup (GCD a) where+  (<>) :: GCD a -> GCD a -> GCD a+  x <> y = GCD{getGCD = x.getGCD `gcd` y.getGCD}
src/GHC/Eventlog/Live/Otelcol/Config/Default.hs view
@@ -22,8 +22,7 @@ import Data.Kind (Type) import Data.List (intercalate) import Data.Proxy (Proxy (..))-import Data.Yaml qualified as Y-import GHC.Eventlog.Live.Otelcol.Config.Default.Raw (defaultConfigByteString)+import GHC.Eventlog.Live.Otelcol.Config.Default.Raw (decodeThrow, defaultConfigByteString) import GHC.Eventlog.Live.Otelcol.Config.Types (Config) import GHC.Records (HasField (..)) import GHC.TypeLits (KnownSymbol, Symbol, symbolVal)@@ -34,7 +33,7 @@ The default configuration. -} defaultConfig :: Config-defaultConfig = $(lift =<< Y.decodeThrow @Q @Config defaultConfigByteString)+defaultConfig = $(lift =<< decodeThrow @Q @Config defaultConfigByteString)  {- | __Warning:__ Compile-time only.@@ -60,7 +59,11 @@ Internal helper. Get the default value under the given path of accessors and `lift` it. -}-getDefault :: forall xs a b. (GetDefault xs a b, Lift b) => a -> Q Exp+getDefault ::+  forall xs a b.+  (GetDefault xs a b, Lift b) =>+  a ->+  Q Exp getDefault a = either throw lift (getDefaultEither @xs @a @b a)  {- |@@ -70,7 +73,12 @@ Get the default value under the given accessor after calling a recursor. This function is used to implement the `GetDefault` instances. -}-getDefaultEither' :: forall x a b c. (KnownSymbol x, HasField x a (Maybe b)) => (b -> Either MissingDefaultException c) -> a -> Either MissingDefaultException c+getDefaultEither' ::+  forall x a b c.+  (KnownSymbol x, HasField x a (Maybe b)) =>+  (b -> Either MissingDefaultException c) ->+  a ->+  Either MissingDefaultException c getDefaultEither' rec a =   first addAccessor . maybe (Left $ MissingDefaultException []) rec $ getField @x a  where
src/GHC/Eventlog/Live/Otelcol/Config/Default/Raw.hs view
@@ -9,14 +9,20 @@ module GHC.Eventlog.Live.Otelcol.Config.Default.Raw (   defaultConfigByteString,   defaultConfigString,-  defaultConfigText,+  defaultConfigJSONSchemaByteString,+  defaultConfigJSONSchemaString,++  -- * Compile-time only helper functions+  decodeThrow, ) where  import Data.ByteString (ByteString)+import Data.ByteString.Lazy qualified as BSL import Data.FileEmbed (embedFileRelative)-import Data.Text (Text) import Data.Text qualified as T-import Data.Text.Encoding qualified as TE (decodeUtf8Lenient)+import Data.Text.Encoding qualified as TE+import Data.YAML (FromYAML)+import Data.YAML qualified as YAML  {- | Internal helper.@@ -30,11 +36,39 @@ The default configuration as a `String`. -} defaultConfigString :: String-defaultConfigString = T.unpack defaultConfigText+defaultConfigString = fromByteString defaultConfigByteString  {- | Internal helper.-The default configuration as a `Text`.+The default configuration as a `ByteString`. -}-defaultConfigText :: Text-defaultConfigText = TE.decodeUtf8Lenient defaultConfigByteString+defaultConfigJSONSchemaByteString :: ByteString+defaultConfigJSONSchemaByteString = $(embedFileRelative "data/config.schema.json")++{- |+Internal helper.+The default configuration as a `String`.+-}+defaultConfigJSONSchemaString :: String+defaultConfigJSONSchemaString = fromByteString defaultConfigJSONSchemaByteString++{- |+__Warning:__ Compile-time only.++Internal helper.+Decode a `ByteString` or throw an exception.+-}+decodeThrow :: (MonadFail m, FromYAML a) => ByteString -> m a+decodeThrow byteString =+  either (fail . prettyErrorMessage) pure $ YAML.decode1Strict byteString+ where+  prettyErrorMessage :: (YAML.Pos, String) -> String+  prettyErrorMessage (pos, errorMessage) =+    YAML.prettyPosWithSource pos (BSL.fromStrict byteString) " error" <> errorMessage++{- |+Internal helper.+Decode a `ByteString` to a `String`.+-}+fromByteString :: ByteString -> String+fromByteString = T.unpack . TE.decodeUtf8Lenient
src/GHC/Eventlog/Live/Otelcol/Config/Types.hs view
@@ -1,391 +1,999 @@-{- |-Module      : GHC.Eventlog.Live.Otelcol.Config.Types-Description : The implementation of @eventlog-live-otelcol@.-Stability   : experimental-Portability : portable--}-module GHC.Eventlog.Live.Otelcol.Config.Types (-  Config (..),-  Processors (..),-  Metrics (..),-  Spans (..),-  AggregationStrategy (..),-  HeapAllocatedMetric (..),-  BlocksSizeMetric (..),-  HeapSizeMetric (..),-  HeapLiveMetric (..),-  MemCurrentMetric (..),-  MemNeededMetric (..),-  MemReturnedMetric (..),-  HeapProfSampleMetric (..),-  CapabilityUsageMetric (..),-  CapabilityUsageSpan (..),-  ThreadStateSpan (..),-) where--import Data.Aeson.Types (Encoding, FromJSON (..), Options (..), Parser, SumEncoding (..), ToJSON (..), Value (..), camelTo2, defaultOptions, genericParseJSON, genericToEncoding, genericToJSON)-import Data.Text (Text)-import GHC.Generics (Generic)-import Language.Haskell.TH.Lift.Compat (Lift)--{- |-The configuration for @eventlog-live-otelcol@.--}-newtype Config = Config-  { processors :: Maybe Processors-  }-  deriving (Generic, Lift)--instance FromJSON Config where-  parseJSON :: Value -> Parser Config-  parseJSON = genericParseJSON encodingOptions--instance ToJSON Config where-  toJSON :: Config -> Value-  toJSON = genericToJSON encodingOptions-  toEncoding :: Config -> Encoding-  toEncoding = genericToEncoding encodingOptions--{- |-The configuration options for the processors.--}-data Processors = Processors-  { metrics :: Maybe Metrics-  , spans :: Maybe Spans-  }-  deriving (Generic, Lift)--instance FromJSON Processors where-  parseJSON :: Value -> Parser Processors-  parseJSON = genericParseJSON encodingOptions--instance ToJSON Processors where-  toJSON :: Processors -> Value-  toJSON = genericToJSON encodingOptions-  toEncoding :: Processors -> Encoding-  toEncoding = genericToEncoding encodingOptions--{- |-The configuration options for the metric processors.--}-data Metrics = Metrics-  { heapAllocated :: Maybe HeapAllocatedMetric-  , blocksSize :: Maybe BlocksSizeMetric-  , heapSize :: Maybe HeapSizeMetric-  , heapLive :: Maybe HeapLiveMetric-  , memCurrent :: Maybe MemCurrentMetric-  , memNeeded :: Maybe MemNeededMetric-  , memReturned :: Maybe MemReturnedMetric-  , heapProfSample :: Maybe HeapProfSampleMetric-  , capabilityUsage :: Maybe CapabilityUsageMetric-  }-  deriving (Generic, Lift)--instance FromJSON Metrics where-  parseJSON :: Value -> Parser Metrics-  parseJSON = genericParseJSON encodingOptions--instance ToJSON Metrics where-  toJSON :: Metrics -> Value-  toJSON = genericToJSON encodingOptions-  toEncoding :: Metrics -> Encoding-  toEncoding = genericToEncoding encodingOptions--{- |-The configuration options for the span processors.--}-data Spans = Spans-  { capabilityUsage :: Maybe CapabilityUsageSpan-  , threadState :: Maybe ThreadStateSpan-  }-  deriving (Generic, Lift)--instance FromJSON Spans where-  parseJSON :: Value -> Parser Spans-  parseJSON = genericParseJSON encodingOptions--instance ToJSON Spans where-  toJSON :: Spans -> Value-  toJSON = genericToJSON encodingOptions-  toEncoding :: Spans -> Encoding-  toEncoding = genericToEncoding encodingOptions--{- |-The options for metric aggregation.--}-data AggregationStrategy-  = AggregationStrategyByBatch-  deriving (Generic, Lift)--instance FromJSON AggregationStrategy where-  parseJSON :: Value -> Parser AggregationStrategy-  parseJSON = genericParseJSON aggregationStrategyEncodingOptions--instance ToJSON AggregationStrategy where-  toJSON :: AggregationStrategy -> Value-  toJSON = genericToJSON aggregationStrategyEncodingOptions-  toEncoding :: AggregationStrategy -> Encoding-  toEncoding = genericToEncoding aggregationStrategyEncodingOptions--{- |-The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Heap.processHeapAllocatedData`.--}-data HeapAllocatedMetric = HeapAllocatedMetric-  { description :: Maybe Text-  , enabled :: Bool-  , name :: Text-  , aggregate :: Maybe AggregationStrategy-  }-  deriving (Generic, Lift)--instance FromJSON HeapAllocatedMetric where-  parseJSON :: Value -> Parser HeapAllocatedMetric-  parseJSON = genericParseJSON encodingOptions--instance ToJSON HeapAllocatedMetric where-  toJSON :: HeapAllocatedMetric -> Value-  toJSON = genericToJSON encodingOptions-  toEncoding :: HeapAllocatedMetric -> Encoding-  toEncoding = genericToEncoding encodingOptions--{- |-The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Heap.processHeapSizeData`.--}-data HeapSizeMetric = HeapSizeMetric-  { description :: Maybe Text-  , enabled :: Bool-  , name :: Text-  , aggregate :: Maybe AggregationStrategy-  }-  deriving (Generic, Lift)--instance FromJSON HeapSizeMetric where-  parseJSON :: Value -> Parser HeapSizeMetric-  parseJSON = genericParseJSON encodingOptions--instance ToJSON HeapSizeMetric where-  toJSON :: HeapSizeMetric -> Value-  toJSON = genericToJSON encodingOptions-  toEncoding :: HeapSizeMetric -> Encoding-  toEncoding = genericToEncoding encodingOptions--{- |-The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Heap.processBlocksSizeData`.--}-data BlocksSizeMetric = BlocksSizeMetric-  { description :: Maybe Text-  , enabled :: Bool-  , name :: Text-  , aggregate :: Maybe AggregationStrategy-  }-  deriving (Generic, Lift)--instance FromJSON BlocksSizeMetric where-  parseJSON :: Value -> Parser BlocksSizeMetric-  parseJSON = genericParseJSON encodingOptions--instance ToJSON BlocksSizeMetric where-  toJSON :: BlocksSizeMetric -> Value-  toJSON = genericToJSON encodingOptions-  toEncoding :: BlocksSizeMetric -> Encoding-  toEncoding = genericToEncoding encodingOptions--{- |-The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Heap.processHeapLiveData`.--}-data HeapLiveMetric = HeapLiveMetric-  { description :: Maybe Text-  , enabled :: Bool-  , name :: Text-  , aggregate :: Maybe AggregationStrategy-  }-  deriving (Generic, Lift)--instance FromJSON HeapLiveMetric where-  parseJSON :: Value -> Parser HeapLiveMetric-  parseJSON = genericParseJSON encodingOptions--instance ToJSON HeapLiveMetric where-  toJSON :: HeapLiveMetric -> Value-  toJSON = genericToJSON encodingOptions-  toEncoding :: HeapLiveMetric -> Encoding-  toEncoding = genericToEncoding encodingOptions--{- |-The configuration options for the @memCurrent@ field `GHC.Eventlog.Live.Machine.Analysis.Heap.processMemReturnData`.--}-data MemCurrentMetric = MemCurrentMetric-  { description :: Maybe Text-  , enabled :: Bool-  , name :: Text-  , aggregate :: Maybe AggregationStrategy-  }-  deriving (Generic, Lift)--instance FromJSON MemCurrentMetric where-  parseJSON :: Value -> Parser MemCurrentMetric-  parseJSON = genericParseJSON encodingOptions--instance ToJSON MemCurrentMetric where-  toJSON :: MemCurrentMetric -> Value-  toJSON = genericToJSON encodingOptions-  toEncoding :: MemCurrentMetric -> Encoding-  toEncoding = genericToEncoding encodingOptions--{- |-The configuration options for the @memNeeded@ field `GHC.Eventlog.Live.Machine.Analysis.Heap.processMemReturnData`.--}-data MemNeededMetric = MemNeededMetric-  { description :: Maybe Text-  , enabled :: Bool-  , name :: Text-  , aggregate :: Maybe AggregationStrategy-  }-  deriving (Generic, Lift)--instance FromJSON MemNeededMetric where-  parseJSON :: Value -> Parser MemNeededMetric-  parseJSON = genericParseJSON encodingOptions--instance ToJSON MemNeededMetric where-  toJSON :: MemNeededMetric -> Value-  toJSON = genericToJSON encodingOptions-  toEncoding :: MemNeededMetric -> Encoding-  toEncoding = genericToEncoding encodingOptions--{- |-The configuration options for the @memReturned@ field `GHC.Eventlog.Live.Machine.Analysis.Heap.processMemReturnData`.--}-data MemReturnedMetric = MemReturnedMetric-  { description :: Maybe Text-  , enabled :: Bool-  , name :: Text-  , aggregate :: Maybe AggregationStrategy-  }-  deriving (Generic, Lift)--instance FromJSON MemReturnedMetric where-  parseJSON :: Value -> Parser MemReturnedMetric-  parseJSON = genericParseJSON encodingOptions--instance ToJSON MemReturnedMetric where-  toJSON :: MemReturnedMetric -> Value-  toJSON = genericToJSON encodingOptions-  toEncoding :: MemReturnedMetric -> Encoding-  toEncoding = genericToEncoding encodingOptions--{- |-The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Heap.processHeapProfSampleData`.--}-data HeapProfSampleMetric = HeapProfSampleMetric-  { description :: Maybe Text-  , enabled :: Bool-  , name :: Text-  , aggregate :: Maybe AggregationStrategy-  }-  deriving (Generic, Lift)--instance FromJSON HeapProfSampleMetric where-  parseJSON :: Value -> Parser HeapProfSampleMetric-  parseJSON = genericParseJSON encodingOptions--instance ToJSON HeapProfSampleMetric where-  toJSON :: HeapProfSampleMetric -> Value-  toJSON = genericToJSON encodingOptions-  toEncoding :: HeapProfSampleMetric -> Encoding-  toEncoding = genericToEncoding encodingOptions--{- |-The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Capability.processCapabilityUsageMetrics`.--}-data CapabilityUsageMetric = CapabilityUsageMetric-  { description :: Maybe Text-  , enabled :: Bool-  , name :: Text-  , aggregate :: Maybe AggregationStrategy-  }-  deriving (Generic, Lift)--instance FromJSON CapabilityUsageMetric where-  parseJSON :: Value -> Parser CapabilityUsageMetric-  parseJSON = genericParseJSON encodingOptions--instance ToJSON CapabilityUsageMetric where-  toJSON :: CapabilityUsageMetric -> Value-  toJSON = genericToJSON encodingOptions-  toEncoding :: CapabilityUsageMetric -> Encoding-  toEncoding = genericToEncoding encodingOptions--{- |-The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Capability.processCapabilityUsageSpans`.--}-data CapabilityUsageSpan = CapabilityUsageSpan-  { description :: Maybe Text-  , enabled :: Bool-  , name :: Text-  }-  deriving (Generic, Lift)--instance FromJSON CapabilityUsageSpan where-  parseJSON :: Value -> Parser CapabilityUsageSpan-  parseJSON = genericParseJSON encodingOptions--instance ToJSON CapabilityUsageSpan where-  toJSON :: CapabilityUsageSpan -> Value-  toJSON = genericToJSON encodingOptions-  toEncoding :: CapabilityUsageSpan -> Encoding-  toEncoding = genericToEncoding encodingOptions--{- |-The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Thread.processThreadStateSpans`.--}-data ThreadStateSpan = ThreadStateSpan-  { description :: Maybe Text-  , enabled :: Bool-  , name :: Text-  }-  deriving (Generic, Lift)--instance FromJSON ThreadStateSpan where-  parseJSON :: Value -> Parser ThreadStateSpan-  parseJSON = genericParseJSON encodingOptions--instance ToJSON ThreadStateSpan where-  toJSON :: ThreadStateSpan -> Value-  toJSON = genericToJSON encodingOptions-  toEncoding :: ThreadStateSpan -> Encoding-  toEncoding = genericToEncoding encodingOptions------------------------------------------------------------------------------------ Internal Helpers----------------------------------------------------------------------------------{- |-Internal helper.-The encoding options that should be used by every `FromJSON` and `ToJSON` instance for the configuration.--}-encodingOptions :: Options-encodingOptions =-  defaultOptions-    { fieldLabelModifier = camelTo2 '_'-    , constructorTagModifier = camelTo2 '_'-    , allNullaryToStringTag = True-    , omitNothingFields = False-    , sumEncoding = UntaggedValue-    , tagSingleConstructors = False-    , unwrapUnaryRecords = False-    }--{- |-Internal helper.-The encoding options that should be used by the `FromJSON` and `ToJSON` instances for `AggregationStrategy`.--}-aggregationStrategyEncodingOptions :: Options-aggregationStrategyEncodingOptions =-  encodingOptions-    { constructorTagModifier = camelTo2 '_' . stripAggregationStrategy-    , tagSingleConstructors = True-    }- where-  stripAggregationStrategy :: String -> String-  stripAggregationStrategy = drop (length ("AggregationStrategy" :: String))+{-# LANGUAGE OverloadedStrings #-}+{-# OPTIONS_GHC -Wno-name-shadowing #-}++{- |+Module      : GHC.Eventlog.Live.Otelcol.Config.Types+Description : The implementation of @eventlog-live-otelcol@.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Otelcol.Config.Types (+  -- * Configuration type+  Config (..),+  FullConfig (..),++  -- ** Processor configuration types+  Processors (..),+  IsProcessorConfig,++  -- *** Log processor configuration types+  Logs (..),+  IsLogProcessorConfig,+  ThreadLabel (..),+  UserMarker (..),+  UserMessage (..),+  InternalLogMessage (..),++  -- *** Metric processor configuration types+  Metrics (..),+  IsMetricProcessorConfig,+  HeapAllocatedMetric (..),+  BlocksSizeMetric (..),+  HeapSizeMetric (..),+  HeapLiveMetric (..),+  MemCurrentMetric (..),+  MemNeededMetric (..),+  MemReturnedMetric (..),+  HeapProfSampleMetric (..),+  CapabilityUsageMetric (..),++  -- *** Trace processor configuration types+  Traces (..),+  IsTraceProcessorConfig,+  CapabilityUsageSpan (..),+  ThreadStateSpan (..),++  -- *** Profile processor configuration types+  Profiles (..),+  IsProfileProcessorConfig,+  StackSampleProfile (..),+  CostCentreSampleProfile (..),++  -- ** Property types+  Duration (..),+  AggregationStrategy (..),+  toAggregationSeconds,+  ExportStrategy (..),+  toExportSeconds,+  isEnabled,+) where++import Control.Applicative (asum)+import Data.Char (isDigit)+import Data.Kind (Constraint, Type)+import Data.Text (Text)+import Data.Text qualified as T+import Data.YAML (FromYAML (..), ToYAML, (.:?), (.=))+import Data.YAML qualified as YAML+import GHC.Records (HasField (..))+import Language.Haskell.TH.Lift.Compat (Lift)+import Text.ParserCombinators.ReadP (ReadP)+import Text.ParserCombinators.ReadP qualified as P+import Text.Read (readEither)++{- |+The extended configuration with derived fields.+-}+data FullConfig = FullConfig+  { batchIntervalMs :: !Int+  -- ^ The batch interval in milliseconds.+  , eventlogFlushIntervalX :: !Int+  -- ^ The @--eventlog-flush-interval@ in /batches/.+  , config :: !Config+  }++{- |+The configuration for @eventlog-live-otelcol@.+-}+newtype Config = Config+  { processors :: Maybe Processors+  }+  deriving (Lift)++instance FromYAML Config where+  parseYAML :: YAML.Node YAML.Pos -> YAML.Parser Config+  parseYAML = YAML.withMap "Config" $ \m ->+    Config+      <$> m .:? "processors"++instance ToYAML Config where+  toYAML :: Config -> YAML.Node ()+  toYAML config =+    YAML.mapping+      [ "processors" .= config.processors+      ]++{- |+The configuration options for the processors.+-}+data Processors = Processors+  { logs :: Maybe Logs+  , metrics :: Maybe Metrics+  , traces :: Maybe Traces+  , profiles :: Maybe Profiles+  }+  deriving (Lift)++instance FromYAML Processors where+  parseYAML :: YAML.Node YAML.Pos -> YAML.Parser Processors+  parseYAML = YAML.withMap "Processors" $ \m ->+    Processors+      <$> m .:? "logs"+      <*> m .:? "metrics"+      <*> m .:? "traces"+      <*> m .:? "profiles"++instance ToYAML Processors where+  toYAML :: Processors -> YAML.Node ()+  toYAML processors =+    YAML.mapping+      [ "logs" .= processors.logs+      , "metrics" .= processors.metrics+      , "traces" .= processors.traces+      , "profiles" .= processors.profiles+      ]++{- |+The configuration options for the span processors.+-}++-- NOTE:+-- If you add a new log, search for the string...+--+--   This should be kept in sync with the list of logs.+--+-- ...and update all the relevant locations.+data Logs = Logs+  { threadLabel :: Maybe ThreadLabel+  , userMarker :: Maybe UserMarker+  , userMessage :: Maybe UserMessage+  , internalLogMessage :: Maybe InternalLogMessage+  }+  deriving (Lift)++instance FromYAML Logs where+  parseYAML :: YAML.Node YAML.Pos -> YAML.Parser Logs+  parseYAML =+    -- NOTE: This should be kept in sync with the list of logs.+    YAML.withMap "Logs" $ \m ->+      Logs+        <$> m .:? "thread_label"+        <*> m .:? "user_marker"+        <*> m .:? "user_message"+        <*> m .:? "internal_log_message"++instance ToYAML Logs where+  toYAML :: Logs -> YAML.Node ()+  toYAML logs =+    -- NOTE: This should be kept in sync with the list of logs.+    YAML.mapping+      [ "thread_label" .= logs.threadLabel+      , "user_marker" .= logs.userMarker+      , "user_message" .= logs.userMessage+      , "internal_log_message" .= logs.internalLogMessage+      ]++{- |+The configuration options for the metric processors.+-}++-- NOTE:+-- If you add a new metric, search for the string...+--+--   This should be kept in sync with the list of metrics.+--+-- ...and update all the relevant locations.+data Metrics = Metrics+  { heapAllocated :: Maybe HeapAllocatedMetric+  , blocksSize :: Maybe BlocksSizeMetric+  , heapSize :: Maybe HeapSizeMetric+  , heapLive :: Maybe HeapLiveMetric+  , memCurrent :: Maybe MemCurrentMetric+  , memNeeded :: Maybe MemNeededMetric+  , memReturned :: Maybe MemReturnedMetric+  , heapProfSample :: Maybe HeapProfSampleMetric+  , capabilityUsage :: Maybe CapabilityUsageMetric+  }+  deriving (Lift)++instance FromYAML Metrics where+  parseYAML :: YAML.Node YAML.Pos -> YAML.Parser Metrics+  parseYAML =+    -- NOTE: This should be kept in sync with the list of metrics.+    YAML.withMap "Metrics" $ \m ->+      Metrics+        <$> m .:? "heap_allocated"+        <*> m .:? "blocks_size"+        <*> m .:? "heap_size"+        <*> m .:? "heap_live"+        <*> m .:? "mem_current"+        <*> m .:? "mem_needed"+        <*> m .:? "mem_returned"+        <*> m .:? "heap_prof_sample"+        <*> m .:? "capability_usage"++instance ToYAML Metrics where+  toYAML :: Metrics -> YAML.Node ()+  toYAML metrics =+    -- NOTE: This should be kept in sync with the list of metrics.+    YAML.mapping+      [ "heap_allocated" .= metrics.heapAllocated+      , "blocks_size" .= metrics.blocksSize+      , "heap_size" .= metrics.heapSize+      , "heap_live" .= metrics.heapLive+      , "mem_current" .= metrics.memCurrent+      , "mem_needed" .= metrics.memNeeded+      , "mem_returned" .= metrics.memReturned+      , "heap_prof_sample" .= metrics.heapProfSample+      , "capability_usage" .= metrics.capabilityUsage+      ]++{- |+The configuration options for the span processors.+-}++-- NOTE:+-- If you add a new trace, search for the string...+--+--   This should be kept in sync with the list of traces.+--+-- ...and update all the relevant locations.+data Traces = Traces+  { capabilityUsage :: Maybe CapabilityUsageSpan+  , threadState :: Maybe ThreadStateSpan+  }+  deriving (Lift)++instance FromYAML Traces where+  parseYAML :: YAML.Node YAML.Pos -> YAML.Parser Traces+  parseYAML =+    -- NOTE: This should be kept in sync with the list of traces.+    YAML.withMap "Traces" $ \m ->+      Traces+        <$> m .:? "capability_usage"+        <*> m .:? "thread_state"++instance ToYAML Traces where+  toYAML :: Traces -> YAML.Node ()+  toYAML traces =+    -- NOTE: This should be kept in sync with the list of traces.+    YAML.mapping+      [ "capability_usage" .= traces.capabilityUsage+      , "thread_state" .= traces.threadState+      ]++{- |+The configuration options for the profile processors.+-}+data Profiles = Profiles+  { stackSample :: Maybe StackSampleProfile+  , costCentreSample :: Maybe CostCentreSampleProfile+  }+  deriving (Lift)++instance FromYAML Profiles where+  parseYAML :: YAML.Node YAML.Pos -> YAML.Parser Profiles+  parseYAML =+    -- NOTE: This should be kept in sync with the list of profiles.+    YAML.withMap "Profiles" $ \m ->+      Profiles+        <$> m .:? "stack_sample"+        <*> m .:? "cost_centre_sample"++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+      ]++-------------------------------------------------------------------------------+-- Logs+-------------------------------------------------------------------------------++{- |+The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Thread.processThreadLabelData`.+-}+data ThreadLabel = ThreadLabel+  { name :: Maybe Text+  , description :: Maybe Text+  , export :: Maybe ExportStrategy+  }+  deriving (Lift)++instance FromYAML ThreadLabel where+  parseYAML :: YAML.Node YAML.Pos -> YAML.Parser ThreadLabel+  parseYAML = genericParseYAMLLogProcessorConfig "ThreadLabel" ThreadLabel++instance ToYAML ThreadLabel where+  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`.+-}+data UserMessage = UserMessage+  { name :: Maybe Text+  , description :: Maybe Text+  , export :: Maybe ExportStrategy+  }+  deriving (Lift)++instance FromYAML UserMessage where+  parseYAML :: YAML.Node YAML.Pos -> YAML.Parser UserMessage+  parseYAML = genericParseYAMLLogProcessorConfig "UserMessage" UserMessage++instance ToYAML UserMessage where+  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`.+-}+data UserMarker = UserMarker+  { name :: Maybe Text+  , description :: Maybe Text+  , export :: Maybe ExportStrategy+  }+  deriving (Lift)++instance FromYAML UserMarker where+  parseYAML :: YAML.Node YAML.Pos -> YAML.Parser UserMarker+  parseYAML = genericParseYAMLLogProcessorConfig "UserMarker" UserMarker++instance ToYAML UserMarker where+  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.+-}+data InternalLogMessage = InternalLogMessage+  { name :: Maybe Text+  , description :: Maybe Text+  , export :: Maybe ExportStrategy+  }+  deriving (Lift)++instance FromYAML InternalLogMessage where+  parseYAML :: YAML.Node YAML.Pos -> YAML.Parser InternalLogMessage+  parseYAML = genericParseYAMLLogProcessorConfig "InternalLogMessage" InternalLogMessage++instance ToYAML InternalLogMessage where+  toYAML :: InternalLogMessage -> YAML.Node ()+  toYAML = genericToYAMLLogProcessorConfig++instance HasField "enabled" InternalLogMessage Bool where+  getField :: InternalLogMessage -> Bool+  getField = isEnabled . (.export)++-------------------------------------------------------------------------------+-- Metrics+-------------------------------------------------------------------------------++{- |+The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Heap.processHeapAllocatedData`.+-}+data HeapAllocatedMetric = HeapAllocatedMetric+  { name :: Maybe Text+  , description :: Maybe Text+  , aggregate :: Maybe AggregationStrategy+  , export :: Maybe ExportStrategy+  }+  deriving (Lift)++instance FromYAML HeapAllocatedMetric where+  parseYAML :: YAML.Node YAML.Pos -> YAML.Parser HeapAllocatedMetric+  parseYAML = genericParseYAMLMetricProcessorConfig "HeapAllocatedMetric" HeapAllocatedMetric++instance ToYAML HeapAllocatedMetric where+  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`.+-}+data HeapSizeMetric = HeapSizeMetric+  { name :: Maybe Text+  , description :: Maybe Text+  , aggregate :: Maybe AggregationStrategy+  , export :: Maybe ExportStrategy+  }+  deriving (Lift)++instance FromYAML HeapSizeMetric where+  parseYAML :: YAML.Node YAML.Pos -> YAML.Parser HeapSizeMetric+  parseYAML = genericParseYAMLMetricProcessorConfig "HeapSizeMetric" HeapSizeMetric++instance ToYAML HeapSizeMetric where+  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`.+-}+data BlocksSizeMetric = BlocksSizeMetric+  { name :: Maybe Text+  , description :: Maybe Text+  , aggregate :: Maybe AggregationStrategy+  , export :: Maybe ExportStrategy+  }+  deriving (Lift)++instance FromYAML BlocksSizeMetric where+  parseYAML :: YAML.Node YAML.Pos -> YAML.Parser BlocksSizeMetric+  parseYAML = genericParseYAMLMetricProcessorConfig "BlocksSizeMetric" BlocksSizeMetric++instance ToYAML BlocksSizeMetric where+  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`.+-}+data HeapLiveMetric = HeapLiveMetric+  { name :: Maybe Text+  , description :: Maybe Text+  , aggregate :: Maybe AggregationStrategy+  , export :: Maybe ExportStrategy+  }+  deriving (Lift)++instance FromYAML HeapLiveMetric where+  parseYAML :: YAML.Node YAML.Pos -> YAML.Parser HeapLiveMetric+  parseYAML = genericParseYAMLMetricProcessorConfig "HeapLiveMetric" HeapLiveMetric++instance ToYAML HeapLiveMetric where+  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`.+-}+data MemCurrentMetric = MemCurrentMetric+  { name :: Maybe Text+  , description :: Maybe Text+  , aggregate :: Maybe AggregationStrategy+  , export :: Maybe ExportStrategy+  }+  deriving (Lift)++instance FromYAML MemCurrentMetric where+  parseYAML :: YAML.Node YAML.Pos -> YAML.Parser MemCurrentMetric+  parseYAML = genericParseYAMLMetricProcessorConfig "MemCurrentMetric" MemCurrentMetric++instance ToYAML MemCurrentMetric where+  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`.+-}+data MemNeededMetric = MemNeededMetric+  { name :: Maybe Text+  , description :: Maybe Text+  , aggregate :: Maybe AggregationStrategy+  , export :: Maybe ExportStrategy+  }+  deriving (Lift)++instance FromYAML MemNeededMetric where+  parseYAML :: YAML.Node YAML.Pos -> YAML.Parser MemNeededMetric+  parseYAML = genericParseYAMLMetricProcessorConfig "MemNeededMetric" MemNeededMetric++instance ToYAML MemNeededMetric where+  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`.+-}+data MemReturnedMetric = MemReturnedMetric+  { name :: Maybe Text+  , description :: Maybe Text+  , aggregate :: Maybe AggregationStrategy+  , export :: Maybe ExportStrategy+  }+  deriving (Lift)++instance FromYAML MemReturnedMetric where+  parseYAML :: YAML.Node YAML.Pos -> YAML.Parser MemReturnedMetric+  parseYAML = genericParseYAMLMetricProcessorConfig "MemReturnedMetric" MemReturnedMetric++instance ToYAML MemReturnedMetric where+  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`.+-}+data HeapProfSampleMetric = HeapProfSampleMetric+  { name :: Maybe Text+  , description :: Maybe Text+  , aggregate :: Maybe AggregationStrategy+  , export :: Maybe ExportStrategy+  }+  deriving (Lift)++instance FromYAML HeapProfSampleMetric where+  parseYAML :: YAML.Node YAML.Pos -> YAML.Parser HeapProfSampleMetric+  parseYAML = genericParseYAMLMetricProcessorConfig "HeapProfSampleMetric" HeapProfSampleMetric++instance ToYAML HeapProfSampleMetric where+  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`.+-}+data CapabilityUsageMetric = CapabilityUsageMetric+  { name :: Maybe Text+  , description :: Maybe Text+  , aggregate :: Maybe AggregationStrategy+  , export :: Maybe ExportStrategy+  }+  deriving (Lift)++instance FromYAML CapabilityUsageMetric where+  parseYAML :: YAML.Node YAML.Pos -> YAML.Parser CapabilityUsageMetric+  parseYAML = genericParseYAMLMetricProcessorConfig "CapabilityUsageMetric" CapabilityUsageMetric++instance ToYAML CapabilityUsageMetric where+  toYAML :: CapabilityUsageMetric -> YAML.Node ()+  toYAML = genericToYAMLMetricProcessorConfig++instance HasField "enabled" CapabilityUsageMetric Bool where+  getField :: CapabilityUsageMetric -> Bool+  getField = isEnabled . (.export)++-------------------------------------------------------------------------------+-- Traces+-------------------------------------------------------------------------------++{- |+The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Capability.processCapabilityUsageTraces`.+-}+data CapabilityUsageSpan = CapabilityUsageSpan+  { name :: Maybe Text+  , description :: Maybe Text+  , export :: Maybe ExportStrategy+  }+  deriving (Lift)++instance FromYAML CapabilityUsageSpan where+  parseYAML :: YAML.Node YAML.Pos -> YAML.Parser CapabilityUsageSpan+  parseYAML = genericParseYAMLTraceProcessorConfig "CapabilityUsageSpan" CapabilityUsageSpan++instance ToYAML CapabilityUsageSpan where+  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`.+-}+data ThreadStateSpan = ThreadStateSpan+  { name :: Maybe Text+  , description :: Maybe Text+  , export :: Maybe ExportStrategy+  }+  deriving (Lift)++instance FromYAML ThreadStateSpan where+  parseYAML :: YAML.Node YAML.Pos -> YAML.Parser ThreadStateSpan+  parseYAML = genericParseYAMLTraceProcessorConfig "ThreadStateSpan" ThreadStateSpan++instance ToYAML ThreadStateSpan where+  toYAML :: ThreadStateSpan -> YAML.Node ()+  toYAML = genericToYAMLTraceProcessorConfig++instance HasField "enabled" ThreadStateSpan Bool where+  getField :: ThreadStateSpan -> Bool+  getField = isEnabled . (.export)++{- |+The configuration options for `GHC.Eventlog.Live.Machine.Analysis.Profile.processStackProfSampleData`.+-}+data StackSampleProfile = StackSampleProfile+  { name :: Maybe Text+  , description :: Maybe Text+  , export :: Maybe ExportStrategy+  }+  deriving (Lift)++instance FromYAML StackSampleProfile where+  parseYAML :: YAML.Node YAML.Pos -> YAML.Parser StackSampleProfile+  parseYAML = genericParseYAMLProfilerProcessorConfig "StackSampleProfile" StackSampleProfile++instance ToYAML StackSampleProfile where+  toYAML :: StackSampleProfile -> 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+  { name :: Maybe Text+  , description :: Maybe Text+  , export :: Maybe ExportStrategy+  }+  deriving (Lift)++instance FromYAML CostCentreSampleProfile where+  parseYAML :: YAML.Node YAML.Pos -> YAML.Parser CostCentreSampleProfile+  parseYAML = genericParseYAMLProfilerProcessorConfig "CostCentreSampleProfile" CostCentreSampleProfile++instance ToYAML CostCentreSampleProfile where+  toYAML :: CostCentreSampleProfile -> YAML.Node ()+  toYAML = genericToYAMLProfilerProcessorConfig++instance HasField "enabled" CostCentreSampleProfile Bool where+  getField :: CostCentreSampleProfile -> Bool+  getField = isEnabled . (.export)++-------------------------------------------------------------------------------+-- Configuration supertypes+-------------------------------------------------------------------------------++{- |+The structural type of processor configurations.+-}+type IsProcessorConfig :: Type -> Constraint+type IsProcessorConfig config =+  ( HasField "name" config (Maybe Text)+  , HasField "description" config (Maybe Text)+  , HasField "enabled" config Bool+  , HasField "export" config (Maybe ExportStrategy)+  )++{- |+The structural type of log processor configurations.+-}+type IsLogProcessorConfig :: Type -> Constraint+type IsLogProcessorConfig config =+  (IsProcessorConfig config)++{- |+The structural type of metric processor configurations.+-}+type IsMetricProcessorConfig :: Type -> Constraint+type IsMetricProcessorConfig config =+  ( IsProcessorConfig config+  , HasField "aggregate" config (Maybe AggregationStrategy)+  )++{- |+The structural type of span processor configurations.+-}+type IsTraceProcessorConfig :: Type -> Constraint+type IsTraceProcessorConfig config =+  (IsProcessorConfig config)++{- |+The structural type of span processor configurations.+-}+type IsProfileProcessorConfig :: Type -> Constraint+type IsProfileProcessorConfig config =+  (IsProcessorConfig config)++--------------------------------------------------------------------------------+-- Duration+--------------------------------------------------------------------------------++data Duration+  = DurationByBatches {batches :: !Int}+  | DurationBySeconds {seconds :: !Double}+  deriving (Lift)++{- |+Internal helper.+A `ReadP` style parser for `AggregationStrategy`.+-}+readPDuration :: ReadP Duration+readPDuration = do+  -- Parse the number+  integerPart <- P.munch1 isDigit+  maybeFractionPart <- P.option Nothing (Just <$ P.char '.' <*> P.munch1 isDigit)++  -- Make a duration by batches+  let byBatches =+        case maybeFractionPart of+          Nothing ->+            case readEither integerPart of+              Left errorMsg -> fail $ "Could not parse duration: " <> errorMsg+              Right batches -> pure DurationByBatches{..}+          Just fractionPart -> fail $ "Fractional batches are unsupported; found " <> integerPart <> "." <> fractionPart <> "x"++  -- Make a duration by seconds+  let bySeconds =+        case readEither $ integerPart <> maybe "" ('.' :) maybeFractionPart of+          Left errorMsg -> fail $ "Could not parse duration: " <> errorMsg+          Right seconds -> pure DurationBySeconds{..}++  -- Parse the unit+  asum+    [ P.char 'x' >> byBatches+    , P.char 's' >> bySeconds+    ]++{- |+Internal helper.+Pretty-print a duration.+-}+prettyDuration :: Duration -> String+prettyDuration = \case+  DurationByBatches{..} -> show batches <> "x"+  DurationBySeconds{..} -> show seconds <> "s"++--------------------------------------------------------------------------------+-- Aggregation Strategy+--------------------------------------------------------------------------------++{- |+The options for metric aggregation strategies.+-}+data AggregationStrategy+  = AggregationStrategyBool {isOn :: !Bool}+  | AggregationStrategyDuration {duration :: !Duration}+  deriving (Lift)++{- |+Convert an `AggregationStrategy` to a number of seconds, if specified in seconds.+-}+toAggregationSeconds :: AggregationStrategy -> Maybe Double+toAggregationSeconds aggregationStrategy+  | AggregationStrategyDuration DurationBySeconds{..} <- aggregationStrategy = Just seconds+  | otherwise = Nothing++instance FromYAML AggregationStrategy where+  parseYAML :: YAML.Node YAML.Pos -> YAML.Parser AggregationStrategy+  parseYAML node = YAML.withScalar "AggregationStrategy" parseYAMLScalar node+   where+    parseYAMLScalar = \case+      YAML.SBool isOn -> pure AggregationStrategyBool{..}+      YAML.SStr str+        | [(duration, "")] <- P.readP_to_S readPDuration (T.unpack str) ->+            pure AggregationStrategyDuration{..}+      _otherwise -> YAML.typeMismatch "AggregationStrategy" node++instance ToYAML AggregationStrategy where+  toYAML :: AggregationStrategy -> YAML.Node ()+  toYAML = \case+    AggregationStrategyBool{..} ->+      YAML.Scalar () (YAML.SBool isOn)+    AggregationStrategyDuration{..} ->+      YAML.Scalar () (YAML.SStr . T.pack . prettyDuration $ duration)++--------------------------------------------------------------------------------+-- Export Strategy+--------------------------------------------------------------------------------++{- |+The options for export strategies.+-}+data ExportStrategy+  = ExportStrategyBool {isOn :: !Bool}+  | ExportStrategyDuration {duration :: !Duration}+  deriving (Lift)++{- |+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++{- |+Convert an `ExportStrategy` to a number of seconds, if specified in seconds.+-}+toExportSeconds :: ExportStrategy -> Maybe Double+toExportSeconds exportStrategy+  | ExportStrategyDuration DurationBySeconds{..} <- exportStrategy = Just seconds+  | otherwise = Nothing++instance FromYAML ExportStrategy where+  parseYAML :: YAML.Node YAML.Pos -> YAML.Parser ExportStrategy+  parseYAML node = YAML.withScalar "ExportStrategy" parseYAMLScalar node+   where+    parseYAMLScalar = \case+      YAML.SBool isOn -> pure ExportStrategyBool{..}+      YAML.SStr str+        | [(duration, "")] <- P.readP_to_S readPDuration (T.unpack str) ->+            pure ExportStrategyDuration{..}+      _otherwise -> YAML.typeMismatch "ExportStrategy" node++instance ToYAML ExportStrategy where+  toYAML :: ExportStrategy -> YAML.Node ()+  toYAML = \case+    ExportStrategyBool{..} ->+      YAML.Scalar () (YAML.SBool isOn)+    ExportStrategyDuration{..} ->+      YAML.Scalar () (YAML.SStr . T.pack . prettyDuration $ duration)++-------------------------------------------------------------------------------+-- Internal Helpers+-------------------------------------------------------------------------------++{- |+Internal helper.+Generic parser for log processor configuration.+-}+genericParseYAMLLogProcessorConfig ::+  String ->+  (Maybe Text -> Maybe Text -> Maybe ExportStrategy -> logProcessorConfig) ->+  YAML.Node YAML.Pos ->+  YAML.Parser logProcessorConfig+genericParseYAMLLogProcessorConfig log mkLogProcessorConfig =+  YAML.withMap log $ \m ->+    mkLogProcessorConfig+      <$> m .:? "name"+      <*> m .:? "description"+      <*> m .:? "export"++{- |+Internal helper.+Generic conversion from log processor configuration to YAML mappings.+-}+genericToYAMLLogProcessorConfig ::+  (IsLogProcessorConfig logProcessorConfig) =>+  logProcessorConfig ->+  YAML.Node ()+genericToYAMLLogProcessorConfig logProcessorConfig =+  YAML.mapping+    [ "name" .= logProcessorConfig.name+    , "description" .= logProcessorConfig.description+    , "export" .= logProcessorConfig.export+    ]++{- |+Internal helper.+Generic parser for metric processor configuration.+-}+genericParseYAMLMetricProcessorConfig ::+  String ->+  (Maybe Text -> Maybe Text -> Maybe AggregationStrategy -> Maybe ExportStrategy -> metricProcessorConfig) ->+  YAML.Node YAML.Pos ->+  YAML.Parser metricProcessorConfig+genericParseYAMLMetricProcessorConfig metric mkMetricProcessorConfig =+  YAML.withMap metric $ \m ->+    mkMetricProcessorConfig+      <$> m .:? "name"+      <*> m .:? "description"+      <*> m .:? "aggregate"+      <*> m .:? "export"++{- |+Internal helper.+Generic conversion from metric processor configuration to YAML mappings.+-}+genericToYAMLMetricProcessorConfig ::+  (IsMetricProcessorConfig metricProcessorConfig) =>+  metricProcessorConfig ->+  YAML.Node ()+genericToYAMLMetricProcessorConfig metricProcessorConfig =+  YAML.mapping+    [ "name" .= metricProcessorConfig.name+    , "description" .= metricProcessorConfig.description+    , "aggregate" .= metricProcessorConfig.aggregate+    , "export" .= metricProcessorConfig.export+    ]++{- |+Internal helper.+Generic parser for trace processor configuration.+-}+genericParseYAMLTraceProcessorConfig ::+  String ->+  (Maybe Text -> Maybe Text -> Maybe ExportStrategy -> traceProcessorConfig) ->+  YAML.Node YAML.Pos ->+  YAML.Parser traceProcessorConfig+genericParseYAMLTraceProcessorConfig trace mkTraceProcessorConfig =+  YAML.withMap trace $ \m ->+    mkTraceProcessorConfig+      <$> m .:? "name"+      <*> m .:? "description"+      <*> m .:? "export"++{- |+Internal helper.+Generic conversion from trace processor configuration to YAML mappings.+-}+genericToYAMLTraceProcessorConfig ::+  (IsTraceProcessorConfig traceProcessorConfig) =>+  traceProcessorConfig ->+  YAML.Node ()+genericToYAMLTraceProcessorConfig traceProcessorConfig =+  YAML.mapping+    [ "name" .= traceProcessorConfig.name+    , "description" .= traceProcessorConfig.description+    , "export" .= traceProcessorConfig.export+    ]++{- |+Internal helper.+Generic parser for processor configuration.+-}+genericParseYAMLProfilerProcessorConfig ::+  String ->+  (Maybe Text -> Maybe Text -> Maybe ExportStrategy -> profilerConfig) ->+  YAML.Node YAML.Pos ->+  YAML.Parser profilerConfig+genericParseYAMLProfilerProcessorConfig trace mkProfilerConfig =+  YAML.withMap trace $ \m ->+    mkProfilerConfig+      <$> m .:? "name"+      <*> m .:? "description"+      <*> m .:? "export"++{- |+Internal helper.+Generic parser for metric configuration.+-}+genericToYAMLProfilerProcessorConfig ::+  (IsTraceProcessorConfig profilerConfig) =>+  profilerConfig ->+  YAML.Node ()+genericToYAMLProfilerProcessorConfig profilerConfig =+  YAML.mapping+    [ "name" .= profilerConfig.name+    , "description" .= profilerConfig.description+    , "export" .= profilerConfig.export+    ]
+ src/GHC/Eventlog/Live/Otelcol/Control.hs view
@@ -0,0 +1,713 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE OverloadedStrings #-}++module GHC.Eventlog.Live.Otelcol.Control (+  ControlOptions (..),+  ControlPort (..),+  ControlCors (..),+  ControlCorsAllowOrigin (..),+  ControlServerApi (..),+  controlOptionsParser,+  startControlServer,+) where++import Control.Applicative (asum)+import Data.ByteString.Char8 qualified as BSC+import Data.Char (isSpace)+import Data.Maybe (isJust)+import GHC.Eventlog.Live.Logger (Logger)+import GHC.Eventlog.Live.Otelcol.Config (ServiceName)+import GHC.Eventlog.Live.Source.Core (EventlogSourceHandle (..))+import Options.Applicative qualified as O+import Options.Applicative.Compat qualified as OC+import Options.Applicative.Extra.Feature (Feature (..))+import Options.Applicative.Extra.Feature qualified as OF+import Text.ParserCombinators.ReadP (ReadP)+import Text.ParserCombinators.ReadP qualified as P++#ifdef EVENTLOG_LIVE_OTELCOL_FEATURE_CONTROL+import Control.Concurrent (forkIO, killThread)+import Control.Concurrent.STM (atomically)+import Control.Concurrent.STM.TVar (TVar, modifyTVar', newTVarIO, readTVarIO)+import Control.Exception (Exception (..), catches)+import Control.Exception qualified as E (Handler (..))+import Control.Monad.IO.Class (MonadIO (..))+import Data.Aeson qualified as JSON+import Data.Aeson.Types (FromJSON (..), ToJSON (..), Parser, Value, genericParseJSON, genericToJSON)+import Data.Binary qualified as B+import Data.ByteString.Lazy qualified as BSL+import Data.Char (isLower, isUpper, toLower)+import Data.HashMap.Strict (HashMap)+import Data.HashMap.Strict qualified as M+import Data.Proxy (Proxy (..))+import Data.Text (Text)+import Data.Text qualified as T+import Data.Text.Encoding qualified as TE+import Data.Word (Word8)+import GHC.Eventlog.Live.Data.Severity (Severity (..))+import GHC.Eventlog.Live.Logger (writeLog)+import GHC.Eventlog.Live.Otelcol.Config (ServiceName (..))+import GHC.Eventlog.Socket.Control qualified as C+import GHC.Generics (Generic)+import Network.Socket (Socket)+import Network.Socket.ByteString.Lazy qualified as SBSL+import Network.Wai (Middleware)+import Network.Wai.Handler.Warp qualified as Warp+import Network.Wai.Middleware.Cors (CorsResourcePolicy (..), Origin, cors)+import Network.Wai.Middleware.RequestLogger (Destination (..), DetailedSettings (..), OutputFormat (..), RequestLoggerSettings (..), defaultDetailedSettings, defaultRequestLoggerSettings, mkRequestLogger)+import Servant (PlainText, throwError)+import Servant.API (Capture, FormUrlEncoded, Get, Header, Headers, JSON, NoContent (..), PostAccepted, ReqBody, StdMethod (..), UVerb, Union, WithStatus (..), addHeader, (:>), type (:<|>) (..))+import Servant.Server (Handler, Server, ServerError (..), serve, respond)+import System.Log.FastLogger (fromLogStr)+import Web.FormUrlEncoded (Form, FormOptions, FromForm (..), genericFromForm)+import Web.FormUrlEncoded qualified as Form (FormOptions (fieldLabelModifier), defaultFormOptions)+#else+import Control.Monad (when)+#endif++--------------------------------------------------------------------------------+-- Feature: control+--------------------------------------------------------------------------------++control :: Feature+control = Feature{flag = "control", isOn = isOn, info = "Cannot start control server."}+ where+  isOn :: Bool+#ifdef EVENTLOG_LIVE_OTELCOL_FEATURE_CONTROL+  isOn = True+#else+  isOn = False+#endif++--------------------------------------------------------------------------------+-- Control App+--------------------------------------------------------------------------------++data ControlServerApi = ControlServerApi+  { notifyNewConnection :: ServiceName -> EventlogSourceHandle -> IO ()+  , notifyEndConnection :: ServiceName -> IO ()+  , stop :: IO ()+  }++noopControlServerApi :: ControlServerApi+noopControlServerApi =+  ControlServerApi+    { notifyNewConnection = \_serviceName _eventlogSourceHandle -> pure ()+    , notifyEndConnection = \_serviceName -> pure ()+    , stop = pure ()+    }++{- |+If the binary is built with -f+control and the control server is enabled from+the CLI, this starts the control server and returns the control server API.++If the binary is NOT built with -f+control, but the control server is enabled+from the CLI, this prints an error and exits the process.++If the control server is NOT enabled from the CLI, this returns a no-op API.+-}+startControlServer :: Logger IO -> ControlOptions -> IO ControlServerApi++--------------------------------------------------------------------------------+-- Control App - Disabled+--------------------------------------------------------------------------------+#ifndef EVENTLOG_LIVE_OTELCOL_FEATURE_CONTROL++startControlServer logger controlOptions = do+  when (shouldStart controlOptions) $+    OF.exitIfUnsupported control logger+  pure noopControlServerApi++--------------------------------------------------------------------------------+-- Control App - Enabled+--------------------------------------------------------------------------------+#else++startControlServer logger controlOptions+  | shouldStart controlOptions = do+    -- Determine the control port+    let port = maybe 30179 (.port) controlOptions.controlPort++    writeLog logger INFO $+      "Starting control server on " <> T.pack (show port)+    -- Create middleware that logs all incoming requests.+    requestLogger <- mkLoggerMiddleware logger++    -- Create variable for storing sockets.+    eventlogSourceHandleMap <- newTVarIO M.empty++    -- Create CORS resource policy.+    --+    -- NOTE: The @wai-cors@ package lets the resource policy depend dynamically+    -- on the received request. However, it is difficult to expose this freedom+    -- via command-line arguments.+    --+    -- TODO: We might have to manually find the origin in the request, check it+    -- against the allow-list, and set the @corsOrigins@ field.+    let !corsResourcePolicy = mkCorsResourcePolicy controlOptions.controlCors++    -- Start control server.+    controlServerThreadId <-+      forkIO $+        Warp.run port $+          requestLogger $+            cors (const $ Just corsResourcePolicy) $+              serve (Proxy @(HealthApi :<|> ControlApi)) $+                controlServer logger eventlogSourceHandleMap (corsIgnoreFailures corsResourcePolicy)++    -- When notified of a new connection, update the eventlogSourceHandleMap.+    let notifyNewConnection serviceName eventlogSourceHandle = do+          writeLog logger DEBUG $+            "New connection for service " <> serviceName.serviceName <> "."+          atomically $+            modifyTVar' eventlogSourceHandleMap (M.insert serviceName eventlogSourceHandle)++    -- When notified of the end of a connection, update the eventlogSourceHandleMap.+    let notifyEndConnection serviceName = do+          writeLog logger DEBUG $+            "End connection for service " <> serviceName.serviceName <> "."+          atomically $+            modifyTVar' eventlogSourceHandleMap (M.delete serviceName)++    -- When requested to stop, kill the control server thread.+    let stop = killThread controlServerThreadId++    pure ControlServerApi{..}+  | otherwise =+    pure noopControlServerApi++mkLoggerMiddleware :: Logger IO -> IO Middleware+mkLoggerMiddleware logger =+  mkRequestLogger $+    defaultRequestLoggerSettings+      { destination = Callback $ writeLog logger TRACE2 . TE.decodeUtf8Lenient . fromLogStr+      , outputFormat =+          DetailedWithSettings+            defaultDetailedSettings+              { useColors = False+              }+      }++mkCorsResourcePolicy :: ControlCors -> CorsResourcePolicy+mkCorsResourcePolicy ControlCors{..} =+  CorsResourcePolicy+    { corsOrigins = corsOrigins+      , corsMethods = ["GET", "POST"]+      , corsRequestHeaders = ["Content-Type", "Origin"]+      , corsExposedHeaders = Nothing+      , corsMaxAge = controlCorsMaxAgeS+      , corsVaryOrigin = corsVaryOrigin+      , corsRequireOrigin = controlCorsRequireOrigin+      , corsIgnoreFailures = controlCorsIgnoreFailures+      }+ where+  -- TODO: If we add authentication, this flag needs to be set.+  corsCredentials :: Bool+  corsCredentials = False++  -- The @wai-cors@ package represents wildcard as @Nothing@ and otherwise+  -- accepts a list of origins and a boolean that determines whether or not+  -- credentials are used to access the resource.+  corsOrigins :: Maybe ([Origin], Bool)+  corsOrigins =+    case controlCorsAllowOrigin of+      Just (ControlCorsAllowOriginList origins) -> Just (origins, corsCredentials)+      _otherwise -> Nothing++  -- The @Vary: Origin@ header must be added if @Access-Control-Allow-Origin@+  -- is not set to wildcard and there are multiple origins.+  corsVaryOrigin :: Bool+  corsVaryOrigin =+    case controlCorsAllowOrigin of+      Just (ControlCorsAllowOriginList origins) -> length origins >= 2+      _otherwise -> False++--------------------------------------------------------------------------------+-- Control Server++controlServer :: Logger IO -> TVar (HashMap ServiceName EventlogSourceHandle) -> Bool -> Server (HealthApi :<|> ControlApi)+controlServer logger eventlogSourceHandleMapVar corsIgnoreFailures =+  health :<|> controlApi+ where+  health :: Handler ()+  health = do+    liftIO . writeLog logger DEBUG $+      "Received request on /health."++  controlApi :: Server ControlApi+  controlApi = eventlogSocket :<|> customCommand+   where+    customCommand :: Server CustomCommandApi+    customCommand namespaceText commandId = callCustomCommand :<|> badCORSPreflight+     where+      callCustomCommand :: CustomCommandReq -> Handler (Union '[CustomCommandAccept, CustomCommandReject])+      callCustomCommand req = do+        liftIO . writeLog logger DEBUG $+          "Received request on /control/" <> namespaceText <> " with command ID " <> T.pack (show commandId) <> " for " <> req.serviceName <> "."+        -- Construct the user namespace.+        liftIO (eitherUserNamespace namespaceText) >>= \case+          -- If userNamespace throws an exception, then...+          Left customCommandError ->+            -- ...respond with a custom command error.+            respond . WithStatus @404 $+              customCommandError+          -- Otherwise, ...+          Right namespace -> do+            -- ...construct the user command...+            let command = C.userCommand namespace (C.CommandId commandId)+            -- ...send the user command...+            withSocketFor (ServiceName req.serviceName) $ \socket -> do+              liftIO (SBSL.sendAll socket $ B.encode command)+            -- ...respond with a success status.+            respond . WithStatus @204 $+              NoContent++    eventlogSocket :: Server EventlogSocketApi+    eventlogSocket =+        (startProfiling :<|> badCORSPreflight)+        :<|> (stopProfiling :<|> badCORSPreflight)+        :<|> (startHeapProfiling :<|> badCORSPreflight)+        :<|> (stopHeapProfiling :<|> badCORSPreflight)+        :<|> (requestHeapCensus :<|> badCORSPreflight)+     where+      startProfiling :: StartProfilingReq -> Handler ()+      startProfiling req = do+        liftIO . writeLog logger DEBUG $+          "Received request on /control/eventlog-socket/start-profiling for " <> req.serviceName <> "."+        -- Send the control command over the socket.+        withSocketFor (ServiceName req.serviceName) $ \socket -> do+          liftIO (SBSL.sendAll socket $ B.encode C.startProfiling)++      stopProfiling :: StopProfilingReq -> Handler ()+      stopProfiling req = do+        liftIO . writeLog logger DEBUG $+          "Received request on /control/eventlog-socket/stop-profiling for " <> req.serviceName <> "."+        -- Send the control command over the socket.+        withSocketFor (ServiceName req.serviceName) $ \socket -> do+          liftIO (SBSL.sendAll socket $ B.encode C.stopProfiling)++      startHeapProfiling :: StartHeapProfilingReq -> Handler ()+      startHeapProfiling req = do+        liftIO . writeLog logger DEBUG $+          "Received request on /control/eventlog-socket/start-heap-profiling for " <> req.serviceName <> "."+        -- Send the control command over the socket.+        withSocketFor (ServiceName req.serviceName) $ \socket -> do+          liftIO (SBSL.sendAll socket $ B.encode C.startHeapProfiling)++      stopHeapProfiling :: StopHeapProfilingReq -> Handler ()+      stopHeapProfiling req = do+        liftIO . writeLog logger DEBUG $+          "Received request on /control/eventlog-socket/stop-heap-profiling for " <> req.serviceName <> "."+        -- Send the control command over the socket.+        withSocketFor (ServiceName req.serviceName) $ \socket -> do+          liftIO (SBSL.sendAll socket $ B.encode C.stopHeapProfiling)++      requestHeapCensus :: RequestHeapCensusReq -> Handler ()+      requestHeapCensus req = do+        liftIO . writeLog logger DEBUG $+          "Received request on /control/eventlog-socket/request-heap-census for " <> req.serviceName <> "."+        -- Send the control command over the socket.+        withSocketFor (ServiceName req.serviceName) $ \socket -> do+          liftIO (SBSL.sendAll socket $ B.encode C.requestHeapCensus)++    badCORSPreflight :: Handler (Union '[BadCORSPreflightAccept, BadCORSPreflightReject])+    badCORSPreflight+      | corsIgnoreFailures = do+          -- This accepts malformed CORS preflight requests.+          liftIO . writeLog logger DEBUG $ "Accepted malformed CORS preflight request."+          respond . WithStatus @204 $+            addHeader @"Access-Control-Allow-Origin" @String "*" $+              addHeader @"Access-Control-Allow-Methods" @String "GET, POST" $+                addHeader @"Access-Control-Allow-Headers" @String "*" $+                  NoContent+      | otherwise = do+          liftIO . writeLog logger DEBUG $ "Accepted malformed CORS preflight request."+          respond . WithStatus @400 $+            NoContent++    withSocketFor :: ServiceName -> (Socket -> Handler ()) -> Handler ()+    withSocketFor serviceName action = do+      eventlogSourceHandleMap <- liftIO (readTVarIO eventlogSourceHandleMapVar)+      case M.lookup serviceName eventlogSourceHandleMap of+        -- If the service is not known, return a 404.+        Nothing ->+          throwError+            ServerError+              { errHTTPCode = 404+              , errReasonPhrase = "Not Found"+              , errBody = BSL.fromStrict . TE.encodeUtf8 $ "Could not find eventlog socket for service " <> serviceName.serviceName <> "."+              , errHeaders = []+              }+        -- If the service telemetry is streamed from stdin or a file, return a 400.+        Just EventlogSourceHandleStdin ->+          throwError+            ServerError+              { errHTTPCode = 400+              , errReasonPhrase = "Bad Request"+              , errBody = BSL.fromStrict . TE.encodeUtf8 $ "The telemetry for service " <> serviceName.serviceName <> " is streamed from stdin."+              , errHeaders = []+              }+        Just (EventlogSourceHandleFile _h) ->+          throwError+            ServerError+              { errHTTPCode = 400+              , errReasonPhrase = "Bad Request"+              , errBody = BSL.fromStrict . TE.encodeUtf8 $ "The telemetry for service " <> serviceName.serviceName <> " is streamed from a file."+              , errHeaders = []+              }+        -- If the service telemetry is streamed from a socket, continue.+        Just (EventlogSourceHandleSocketUnix s) -> action s++--------------------------------------------------------------------------------+-- Control API++type ControlApi =+  "control" :> (EventlogSocketApi :<|> CustomCommandApi)++--------------------------------------------------------------------------------+-- Health API++type HealthApi =+  "health" :> Get '[JSON] ()++--------------------------------------------------------------------------------+-- Custom Command API++type CustomCommandApi =+  Capture "namespace" Text+    :> Capture "commandId" Word8+      :> ((ReqBody '[FormUrlEncoded, JSON] CustomCommandReq+        :> UVerb 'POST '[JSON] '[CustomCommandAccept, CustomCommandReject]) :<|> BadCORSPreflight)++type CustomCommandAccept = WithStatus 204 NoContent++type CustomCommandReject = WithStatus 404 CustomCommandError++newtype CustomCommandError = CustomCommandError+  { errorMessage :: String+  }+  deriving (Generic, Show)++eitherUserNamespace :: Text -> IO (Either CustomCommandError C.Namespace)+eitherUserNamespace namespace =+  (pure . Right . C.userNamespace $ namespace) `catches` handlers+ where+  handlers =+    [ E.Handler $ pure . Left . toCustomCommandError @C.NamespaceReservedError+    , E.Handler $ pure . Left . toCustomCommandError @C.NamespaceTooLongError+    ]++toCustomCommandError :: (Exception e) => e -> CustomCommandError+toCustomCommandError = CustomCommandError . displayException++instance ToJSON CustomCommandError where+  toJSON :: CustomCommandError -> Value+  toJSON = genericToJSON myJSONOptions++--------------------------------------------------------------------------------+-- CustomCommandReq++data CustomCommandReq = CustomCommandReq+  { serviceName :: Text+  }+  deriving (Generic, Show)++instance FromForm CustomCommandReq where+  fromForm :: Form -> Either Text CustomCommandReq+  fromForm = genericFromForm myFormOptions++instance FromJSON CustomCommandReq where+  parseJSON :: Value -> Parser CustomCommandReq+  parseJSON = genericParseJSON myJSONOptions++--------------------------------------------------------------------------------+-- Eventlog Socket API++-- 2025-12-11:+-- Ideally, the /heap-profiling API would attach the semantics of /start and+-- /stop to a PUT and DELETE request on /heap-profiling with 204 No Content+-- responses. This would give us exactly the right caching semantics.+-- However, I have not been able to find any Grafana plugins that support+-- sending PUT and DELETE requests, and servant appears to have trouble adding+-- headers to 204 No Content responses, so I'm using POST requests on separate+-- /start and /stop endpoints.++type EventlogSocketApi =+  "eventlog-socket"+    :> ("start-profiling" :> (StartProfilingApi :<|> BadCORSPreflight)+          :<|> "stop-profiling" :> (StopProfilingApi :<|> BadCORSPreflight)+          :<|> "start-heap-profiling" :> (StartHeapProfilingApi :<|> BadCORSPreflight)+          :<|> "stop-heap-profiling" :> (StopHeapProfilingApi :<|> BadCORSPreflight)+          :<|> "request-heap-census" :> (RequestHeapCensusApi :<|> BadCORSPreflight)+       )++type StartProfilingApi =+  ReqBody '[FormUrlEncoded, JSON] StartProfilingReq+    :> PostAccepted '[JSON] ()++type StopProfilingApi =+  ReqBody '[FormUrlEncoded, JSON] StopProfilingReq+    :> PostAccepted '[JSON] ()++type StartHeapProfilingApi =+  ReqBody '[FormUrlEncoded, JSON] StartHeapProfilingReq+    :> PostAccepted '[JSON] ()++type StopHeapProfilingApi =+  ReqBody '[FormUrlEncoded, JSON] StopHeapProfilingReq+    :> PostAccepted '[JSON] ()++type RequestHeapCensusApi =+  ReqBody '[FormUrlEncoded, JSON] RequestHeapCensusReq+    :> PostAccepted '[JSON] ()++type BadCORSPreflight =+  UVerb 'OPTIONS '[PlainText] '[BadCORSPreflightAccept, BadCORSPreflightReject]++type BadCORSPreflightAccept = WithStatus 204 (Headers+    '[ Header "Access-Control-Allow-Origin" String+     , Header "Access-Control-Allow-Methods" String+     , Header "Access-Control-Allow-Headers" String+     ]+    NoContent)++type BadCORSPreflightReject =  WithStatus 400 NoContent++--------------------------------------------------------------------------------+-- StartProfilingReq++newtype StartProfilingReq = StartProfilingReq+  { serviceName :: Text+  }+  deriving (Generic, Show)++instance FromForm StartProfilingReq where+  fromForm :: Form -> Either Text StartProfilingReq+  fromForm = genericFromForm myFormOptions++instance FromJSON StartProfilingReq where+  parseJSON :: Value -> Parser StartProfilingReq+  parseJSON = genericParseJSON myJSONOptions++--------------------------------------------------------------------------------+-- StopProfilingReq++newtype StopProfilingReq = StopProfilingReq+  { serviceName :: Text+  }+  deriving (Generic, Show)++instance FromForm StopProfilingReq where+  fromForm :: Form -> Either Text StopProfilingReq+  fromForm = genericFromForm myFormOptions++instance FromJSON StopProfilingReq where+  parseJSON :: Value -> Parser StopProfilingReq+  parseJSON = genericParseJSON myJSONOptions++--------------------------------------------------------------------------------+-- StartHeapProfilingReq++newtype StartHeapProfilingReq = StartHeapProfilingReq+  { serviceName :: Text+  }+  deriving (Generic, Show)++instance FromForm StartHeapProfilingReq where+  fromForm :: Form -> Either Text StartHeapProfilingReq+  fromForm = genericFromForm myFormOptions++instance FromJSON StartHeapProfilingReq where+  parseJSON :: Value -> Parser StartHeapProfilingReq+  parseJSON = genericParseJSON myJSONOptions++--------------------------------------------------------------------------------+-- StopHeapProfilingReq++newtype StopHeapProfilingReq = StopHeapProfilingReq+  { serviceName :: Text+  }+  deriving (Generic, Show)++instance FromForm StopHeapProfilingReq where+  fromForm :: Form -> Either Text StopHeapProfilingReq+  fromForm = genericFromForm myFormOptions++instance FromJSON StopHeapProfilingReq where+  parseJSON :: Value -> Parser StopHeapProfilingReq+  parseJSON = genericParseJSON myJSONOptions++--------------------------------------------------------------------------------+-- RequestHeapCensusReq++newtype RequestHeapCensusReq = RequestHeapCensusReq+  { serviceName :: Text+  }+  deriving (Generic, Show)++instance FromForm RequestHeapCensusReq where+  fromForm :: Form -> Either Text RequestHeapCensusReq+  fromForm = genericFromForm myFormOptions++instance FromJSON RequestHeapCensusReq where+  parseJSON :: Value -> Parser RequestHeapCensusReq+  parseJSON = genericParseJSON myJSONOptions++--------------------------------------------------------------------------------+-- Internal helpers.++-- | Generic options for `FromForm`.+myFormOptions :: FormOptions+myFormOptions =+  Form.defaultFormOptions+    { Form.fieldLabelModifier = camelTo2 '-'+    }++-- | Generic options for `FromJSON`.+myJSONOptions :: JSON.Options+myJSONOptions =+  JSON.defaultOptions+    { JSON.fieldLabelModifier = camelTo2 '-'+    }++-- | Taken from aeson.+camelTo2 :: Char -> String -> String+camelTo2 c = map toLower . go2 . go1+ where+  go1 "" = ""+  go1 (x : u : l : xs) | isUpper u && isLower l = x : c : u : l : go1 xs+  go1 (x : xs) = x : go1 xs+  go2 "" = ""+  go2 (l : u : xs) | isLower l && isUpper u = l : c : u : go2 xs+  go2 (x : xs) = x : go2 xs++#endif++--------------------------------------------------------------------------------+-- Control Options+--------------------------------------------------------------------------------++data ControlOptions = ControlOptions+  { controlEnabled :: !Bool+  , controlPort :: !(Maybe ControlPort)+  , controlCors :: !ControlCors+  }++{- |+Internal helper.++If the user provides any of the control options, this implies @--control@.+-}+shouldStart :: ControlOptions -> Bool+shouldStart controlOptions =+  controlOptions.controlEnabled+    || isJust controlOptions.controlPort+    || isJust controlOptions.controlCors.controlCorsAllowOrigin+    || isJust controlOptions.controlCors.controlCorsMaxAgeS+    || controlOptions.controlCors.controlCorsRequireOrigin+    || controlOptions.controlCors.controlCorsIgnoreFailures++controlOptionsParser :: O.Parser ControlOptions+controlOptionsParser =+  OC.parserOptionGroup "Control Server Options" $+    ControlOptions+      <$> controlEnabledParser+      <*> controlPortParser+      <*> controlCorsParser++controlEnabledParser :: O.Parser Bool+controlEnabledParser =+  OF.onlyFor control (O.flag False True) mempty $+    O.long "control"+      <> OF.helpFor control "Start the control server."++newtype ControlPort = ControlPort {port :: Int}+  deriving (Eq, Show)++controlPortParser :: O.Parser (Maybe ControlPort)+controlPortParser =+  asum+    [ OF.onlyFor control (O.option (Just . ControlPort <$> O.auto)) (O.metavar "PORT") $+        O.long "control-port"+          <> OF.helpFor control "The port number for the control server."+    , pure Nothing+    ]++data ControlCors = ControlCors+  { controlCorsAllowOrigin :: !(Maybe ControlCorsAllowOrigin)+  , controlCorsMaxAgeS :: !(Maybe Int)+  , controlCorsRequireOrigin :: !Bool+  , controlCorsIgnoreFailures :: !Bool+  }++controlCorsParser :: O.Parser ControlCors+controlCorsParser =+  ControlCors+    <$> controlCorsAllowOriginParser+    <*> controlCorsMaxAgeSParser+    <*> controlCorsRequireOriginParser+    <*> controlCorsIgnoreFailuresParser++controlCorsMaxAgeSParser :: O.Parser (Maybe Int)+controlCorsMaxAgeSParser =+  asum+    [ OF.onlyFor control (O.option (Just <$> O.auto)) (O.metavar "SECONDS") $+        O.long "control-cors-max-age"+          <> OF.helpFor control "Set the maximum age of a cached CORS preflight request for the control server CORS policy."+    , pure Nothing+    ]++controlCorsRequireOriginParser :: O.Parser Bool+controlCorsRequireOriginParser =+  OF.onlyFor control (O.flag False True) mempty $+    O.long "control-cors-require-origin"+      <> OF.helpFor control "If enabled, the control server will not accept requests without an Origin header."++controlCorsIgnoreFailuresParser :: O.Parser Bool+controlCorsIgnoreFailuresParser =+  OF.onlyFor control (O.flag False True) mempty $+    O.long "control-cors-ignore-failure"+      <> OF.helpFor control "If enabled, the control server will accept malformed CORS preflight requests."++type ControlCorsOrigin = BSC.ByteString++data ControlCorsAllowOrigin+  = ControlCorsAllowOriginWildcard+  | ControlCorsAllowOriginList [ControlCorsOrigin]++controlCorsAllowOriginParser :: O.Parser (Maybe ControlCorsAllowOrigin)+controlCorsAllowOriginParser =+  asum+    [ OF.onlyFor control (O.option (Just <$> readPReader pControlCorsAllowOrigin)) (O.metavar "ORIGIN") $+        O.long "control-cors-allow-origin"+          <> OF.helpFor control "Set the allowed origins for the control server CORS policy."+    , pure Nothing+    ]++readPReader :: ReadP a -> O.ReadM a+readPReader readP = readSReader (P.readP_to_S readP)++readSReader :: ReadS a -> O.ReadM a+readSReader readS = O.maybeReader $ \str ->+  case readS str of+    [(a, "")] -> Just a+    _otherwise -> Nothing++pControlCorsAllowOrigin :: ReadP ControlCorsAllowOrigin+pControlCorsAllowOrigin =+  P.skipSpaces+    *> asum+      [ -- Wildcard+        ControlCorsAllowOriginWildcard <$ P.char '*' <* P.skipSpaces+      , -- List of origins+        ControlCorsAllowOriginList <$> P.sepBy1 pOrigin (P.char ',' <* P.skipSpaces)+      ]+ where+  -- TODO: The parser for origin could parse the syntax for origins:+  --+  -- Origin: null+  -- Origin: <scheme>://<hostname>+  -- Origin: <scheme>://<hostname>:<port>+  --+  pOrigin :: ReadP ControlCorsOrigin+  pOrigin = BSC.pack <$> P.munch1 (\c -> not (c == ',' || isSpace c)) <* P.skipSpaces
− src/GHC/Eventlog/Live/Otelcol/Exporter.hs 
@@ -1,276 +0,0 @@-{-# OPTIONS_GHC -Wno-orphans #-}--module GHC.Eventlog.Live.Otelcol.Exporter (-  -- * Metrics-  ExportMetricsResult (..),-  RejectedMetricsError (..),-  exportResourceMetrics,--  -- * Traces-  ExportTraceResult (..),-  RejectedSpansError (..),-  exportResourceSpans,--  -- * Helpers--  -- ** Counting Metrics-  countDataPointsInExportMetricsServiceRequest,-  countDataPointsInResourceMetrics,-  countDataPointsInScopeMetrics,-  countDataPointsInMetric,--  -- ** Counting Spans-  countSpansInExportTraceServiceRequest,-  countSpansInResourceSpans,-  countSpansInScopeSpans,-) where--import Control.Exception (Exception (..), SomeException (..), catch)-import Control.Monad.IO.Class (MonadIO (..))-import Data.Int (Int64)-import Data.Machine (ProcessT, await, repeatedly, yield)-import Data.Semigroup (Sum (..))-import Data.Text (Text)-import Data.Vector qualified as V-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.Collector.Trace.V1.TraceService qualified as OTS-import Proto.Opentelemetry.Proto.Collector.Trace.V1.TraceService_Fields qualified as OTS-import Proto.Opentelemetry.Proto.Metrics.V1.Metrics qualified as OM-import Proto.Opentelemetry.Proto.Metrics.V1.Metrics_Fields qualified as OM-import Proto.Opentelemetry.Proto.Trace.V1.Trace qualified as OT-import Proto.Opentelemetry.Proto.Trace.V1.Trace_Fields qualified as OT-import Text.Printf (printf)------------------------------------------------------------------------------------- OpenTelemetry gRPC Exporters---------------------------------------------------------------------------------------------------------------------------------------------------------------------- OpenTelemetry Exporter Result for Metrics--data ExportMetricsResult-  = ExportMetricsResult-  { exportedDataPoints :: !Int64-  , rejectedDataPoints :: !Int64-  , maybeSomeException :: Maybe SomeException-  }-  deriving (Show)--pattern ExportMetricsSuccess :: Int64 -> ExportMetricsResult-pattern ExportMetricsSuccess exportedDataPoints =-  ExportMetricsResult exportedDataPoints 0 Nothing--pattern ExportMetricsError :: Int64 -> Int64 -> SomeException -> ExportMetricsResult-pattern ExportMetricsError exportedDataPoints rejectedDataPoints someException =-  ExportMetricsResult exportedDataPoints rejectedDataPoints (Just someException)--data RejectedMetricsError-  = RejectedMetricsError-  { rejectedDataPoints :: !Int64-  , errorMessage :: !Text-  }-  deriving (Show)--instance Exception RejectedMetricsError where-  displayException :: RejectedMetricsError -> String-  displayException RejectedMetricsError{..} =-    printf "Error: OpenTelemetry Collector rejected %d data points with message: %s" rejectedDataPoints errorMessage------------------------------------------------------------------------------------- OpenTelemetry gRPC Exporter for Metrics--exportResourceMetrics ::-  G.Connection ->-  ProcessT IO OMS.ExportMetricsServiceRequest ExportMetricsResult-exportResourceMetrics conn =-  repeatedly $-    await >>= \exportMetricsServiceRequest ->-      liftIO (sendResourceMetrics exportMetricsServiceRequest)-        >>= yield- where-  sendResourceMetrics :: OMS.ExportMetricsServiceRequest -> IO ExportMetricsResult-  sendResourceMetrics exportMetricsServiceRequest =-    doGrpc `catch` handleGrpcError-   where-    !totalDataPoints = 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 totalDataPoints-          | otherwise -> do-              let !rejectedDataPoints = resp ^. OMS.partialSuccess . OMS.rejectedDataPoints-              let !exportedDataPoints = totalDataPoints - 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 totalDataPoints (SomeException grpcError)--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------------------------------------------------------------------------------------- OpenTelemetry Exporter Result for Traces--data ExportTraceResult-  = ExportTraceResult-  { exportedSpans :: !Int64-  , rejectedSpans :: !Int64-  , maybeSomeException :: Maybe SomeException-  }-  deriving (Show)--pattern ExportTraceSuccess :: Int64 -> ExportTraceResult-pattern ExportTraceSuccess exportedSpans =-  ExportTraceResult exportedSpans 0 Nothing--pattern ExportTraceError :: Int64 -> Int64 -> SomeException -> ExportTraceResult-pattern ExportTraceError exportedSpans rejectedSpans someException =-  ExportTraceResult exportedSpans rejectedSpans (Just someException)--data RejectedSpansError-  = RejectedSpansError-  { rejectedSpans :: !Int64-  , errorMessage :: !Text-  }-  deriving (Show)--instance Exception RejectedSpansError where-  displayException :: RejectedSpansError -> String-  displayException RejectedSpansError{..} =-    printf "Error: OpenTelemetry Collector rejectedSpans %d data points with message: %s" rejectedSpans errorMessage------------------------------------------------------------------------------------- OpenTelemetry gRPC Exporter for Traces--exportResourceSpans ::-  G.Connection ->-  ProcessT IO OTS.ExportTraceServiceRequest ExportTraceResult-exportResourceSpans conn =-  repeatedly $-    await >>= \exportTraceServiceRequest ->-      liftIO (sendResourceSpans exportTraceServiceRequest)-        >>= yield- where-  sendResourceSpans :: OTS.ExportTraceServiceRequest -> IO ExportTraceResult-  sendResourceSpans exportTraceServiceRequest =-    doGrpc `catch` handleGrpcError-   where-    !totalSpans = countSpansInExportTraceServiceRequest exportTraceServiceRequest--    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 totalSpans-          | otherwise -> do-              let !rejectedSpans = resp ^. OTS.partialSuccess . OTS.rejectedSpans-              let !exportedSpans = totalSpans - rejectedSpans-              let !rejectedMetricsError = RejectedSpansError{errorMessage = resp ^. OTS.partialSuccess . OTS.errorMessage, ..}-              pure $ ExportTraceError exportedSpans rejectedSpans (SomeException rejectedMetricsError)--    handleGrpcError :: G.GrpcError -> IO ExportTraceResult-    handleGrpcError grpcError = pure $ ExportTraceError 0 totalSpans (SomeException grpcError)--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------------------------------------------------------------------------------------- Internal Helpers-----------------------------------------------------------------------------------{- |-Internal helper.-Count the number of `OM.NumberDataPoint` values in an `OMS.ExportMetricsServiceRequest`.--}-{-# SPECIALIZE countDataPointsInExportMetricsServiceRequest :: OMS.ExportMetricsServiceRequest -> Int64 #-}-{-# SPECIALIZE countDataPointsInExportMetricsServiceRequest :: OMS.ExportMetricsServiceRequest -> Word #-}-countDataPointsInExportMetricsServiceRequest :: (Integral i) => OMS.ExportMetricsServiceRequest -> i-countDataPointsInExportMetricsServiceRequest exportMetricsServiceRequest =-  getSum $ foldMap (Sum . countDataPointsInResourceMetrics) (exportMetricsServiceRequest ^. OMS.vec'resourceMetrics)--{- |-Internal helper.-Count the number of `OM.NumberDataPoint` values in an `OM.ResourceMetrics`.--}-{-# SPECIALIZE countDataPointsInResourceMetrics :: OM.ResourceMetrics -> Int64 #-}-{-# SPECIALIZE countDataPointsInResourceMetrics :: OM.ResourceMetrics -> Word #-}-countDataPointsInResourceMetrics :: (Integral i) => OM.ResourceMetrics -> i-countDataPointsInResourceMetrics resourceMetrics =-  getSum $ foldMap (Sum . countDataPointsInScopeMetrics) (resourceMetrics ^. OM.vec'scopeMetrics)--{- |-Internal helper.-Count the number of `OM.NumberDataPoint` values in an `OM.ScopeMetrics`.--}-{-# SPECIALIZE countDataPointsInScopeMetrics :: OM.ScopeMetrics -> Int64 #-}-{-# SPECIALIZE countDataPointsInScopeMetrics :: OM.ScopeMetrics -> Word #-}-countDataPointsInScopeMetrics :: (Integral i) => OM.ScopeMetrics -> i-countDataPointsInScopeMetrics scopeMetrics =-  getSum $ foldMap (Sum . countDataPointsInMetric) (scopeMetrics ^. OM.vec'metrics)--{- |-Internal helper.-Count the number of `OM.NumberDataPoint` values in an `OM.Metric`.--}-{-# SPECIALIZE countDataPointsInMetric :: OM.Metric -> Int64 #-}-{-# SPECIALIZE countDataPointsInMetric :: OM.Metric -> Word #-}-countDataPointsInMetric :: (Integral i) => OM.Metric -> i-countDataPointsInMetric metric =-  fromIntegral $-    case metric ^. OM.maybe'data' of-      Nothing -> 0-      Just (OM.Metric'Gauge gauge) ->-        V.length (gauge ^. OM.vec'dataPoints)-      Just (OM.Metric'Sum sum_) ->-        V.length (sum_ ^. OM.vec'dataPoints)-      Just (OM.Metric'Histogram histogram) ->-        V.length (histogram ^. OM.vec'dataPoints)-      Just (OM.Metric'ExponentialHistogram exponentialHistogram) ->-        V.length (exponentialHistogram ^. OM.vec'dataPoints)-      Just (OM.Metric'Summary summary) ->-        V.length (summary ^. OM.vec'dataPoints)--{- |-Internal helper.-Count the number of `OT.Span` values in an `OTS.ExportTraceServiceRequest`.--}-{-# SPECIALIZE countSpansInExportTraceServiceRequest :: OTS.ExportTraceServiceRequest -> Int64 #-}-{-# SPECIALIZE countSpansInExportTraceServiceRequest :: OTS.ExportTraceServiceRequest -> Word #-}-countSpansInExportTraceServiceRequest :: (Integral i) => OTS.ExportTraceServiceRequest -> i-countSpansInExportTraceServiceRequest exportTraceServiceRequest =-  getSum $ foldMap (Sum . countSpansInResourceSpans) (exportTraceServiceRequest ^. OTS.vec'resourceSpans)--{- |-Internal helper.-Count the number of `OT.Span` values in an `OT.ResourceSpans`.--}-{-# SPECIALIZE countSpansInResourceSpans :: OT.ResourceSpans -> Int64 #-}-{-# SPECIALIZE countSpansInResourceSpans :: OT.ResourceSpans -> Word #-}-countSpansInResourceSpans :: (Integral i) => OT.ResourceSpans -> i-countSpansInResourceSpans resourceSpans =-  getSum $ foldMap (Sum . countSpansInScopeSpans) (resourceSpans ^. OT.vec'scopeSpans)--{- |-Internal helper.-Count the number of `OT.Span` values in an `OT.ScopeSpans`.--}-{-# SPECIALIZE countSpansInScopeSpans :: OT.ScopeSpans -> Int64 #-}-{-# SPECIALIZE countSpansInScopeSpans :: OT.ScopeSpans -> Word #-}-countSpansInScopeSpans :: (Integral i) => OT.ScopeSpans -> i-countSpansInScopeSpans scopeSpans =-  fromIntegral $-    V.length (scopeSpans ^. OT.vec'spans)
+ src/GHC/Eventlog/Live/Otelcol/Exporter/Logs.hs view
@@ -0,0 +1,143 @@+{-# OPTIONS_GHC -Wno-orphans #-}++module GHC.Eventlog.Live.Otelcol.Exporter.Logs (+  -- * Logs+  ExportLogsResult (..),+  RejectedLogsError (..),+  exportResourceLogs,+) where++import Control.Exception (Exception (..), SomeException (..), catch)+import Control.Monad (unless)+import Control.Monad.IO.Class (MonadIO (..))+import Data.Int (Int64)+import Data.Machine (ProcessT, await, construct, yield)+import Data.Semigroup (Sum (..))+import Data.Text (Text)+import Data.Vector qualified as V+import GHC.Eventlog.Live.Machine.Core (Tick (..))+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+import Proto.Opentelemetry.Proto.Logs.V1.Logs_Fields qualified as OL+import Text.Printf (printf)++--------------------------------------------------------------------------------+-- OpenTelemetry gRPC Exporters+--------------------------------------------------------------------------------++--------------------------------------------------------------------------------+-- OpenTelemetry Exporter Result for Logs++data ExportLogsResult+  = ExportLogsResult+  { exportedLogRecords :: !Int64+  , rejectedLogRecords :: !Int64+  , maybeSomeException :: Maybe SomeException+  }+  deriving (Show)++pattern ExportLogsSuccess :: Int64 -> ExportLogsResult+pattern ExportLogsSuccess exportedLogRecords =+  ExportLogsResult exportedLogRecords 0 Nothing++pattern ExportLogsError :: Int64 -> Int64 -> SomeException -> ExportLogsResult+pattern ExportLogsError exportedLogRecords rejectedLogRecords someException =+  ExportLogsResult exportedLogRecords rejectedLogRecords (Just someException)++data RejectedLogsError+  = RejectedLogsError+  { rejectedLogRecords :: !Int64+  , errorMessage :: !Text+  }+  deriving (Show)++instance Exception RejectedLogsError where+  displayException :: RejectedLogsError -> String+  displayException RejectedLogsError{..} =+    printf "Error: OpenTelemetry Collector rejected %d log records with message: %s" rejectedLogRecords errorMessage++--------------------------------------------------------------------------------+-- OpenTelemetry gRPC Exporter for Logs++exportResourceLogs ::+  G.Connection ->+  ProcessT IO (Tick OLS.ExportLogsServiceRequest) (Tick ExportLogsResult)+exportResourceLogs conn = construct $ go False+ where+  go exportedResourceLogs =+    await >>= \case+      Tick -> do+        unless exportedResourceLogs $+          yield (Item $ ExportLogsSuccess 0)+        yield Tick+        go False+      Item exportLogsServiceRequest -> do+        exportLogsResult <- liftIO (sendResourceLogs exportLogsServiceRequest)+        yield (Item exportLogsResult)+        go True++  sendResourceLogs :: OLS.ExportLogsServiceRequest -> IO ExportLogsResult+  sendResourceLogs exportLogsServiceRequest =+    doGrpc `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)++    handleSomeException :: SomeException -> IO ExportLogsResult+    handleSomeException someException = pure $ ExportLogsError 0 exportedLogRecords someException++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++--------------------------------------------------------------------------------+-- Internal Helpers+--------------------------------------------------------------------------------++{- |+Internal helper.+Count the number of `OL.NumberDataPoint` values in an `OLS.ExportLogsServiceRequest`.+-}+{-# SPECIALIZE countLogRecordsInExportLogsServiceRequest :: OLS.ExportLogsServiceRequest -> Int64 #-}+{-# SPECIALIZE countLogRecordsInExportLogsServiceRequest :: OLS.ExportLogsServiceRequest -> Word #-}+countLogRecordsInExportLogsServiceRequest :: (Integral i) => OLS.ExportLogsServiceRequest -> i+countLogRecordsInExportLogsServiceRequest exportLogsServiceRequest =+  getSum $ foldMap (Sum . countLogRecordsInResourceLogs) (exportLogsServiceRequest ^. OLS.vec'resourceLogs)++{- |+Internal helper.+Count the number of `OL.NumberDataPoint` values in an `OL.ResourceLogs`.+-}+{-# SPECIALIZE countLogRecordsInResourceLogs :: OL.ResourceLogs -> Int64 #-}+{-# SPECIALIZE countLogRecordsInResourceLogs :: OL.ResourceLogs -> Word #-}+countLogRecordsInResourceLogs :: (Integral i) => OL.ResourceLogs -> i+countLogRecordsInResourceLogs resourceLogs =+  getSum $ foldMap (Sum . countLogRecordsInScopeLogs) (resourceLogs ^. OL.vec'scopeLogs)++{- |+Internal helper.+Count the number of `OL.NumberDataPoint` values in an `OL.ScopeLogs`.+-}+{-# SPECIALIZE countLogRecordsInScopeLogs :: OL.ScopeLogs -> Int64 #-}+{-# SPECIALIZE countLogRecordsInScopeLogs :: OL.ScopeLogs -> Word #-}+countLogRecordsInScopeLogs :: (Integral i) => OL.ScopeLogs -> i+countLogRecordsInScopeLogs scopeLogs =+  fromIntegral $+    V.length (scopeLogs ^. OL.vec'logRecords)
+ src/GHC/Eventlog/Live/Otelcol/Exporter/Metrics.hs view
@@ -0,0 +1,167 @@+{-# OPTIONS_GHC -Wno-orphans #-}++module GHC.Eventlog.Live.Otelcol.Exporter.Metrics (+  -- * Metrics+  ExportMetricsResult (..),+  RejectedMetricsError (..),+  exportResourceMetrics,+) where++import Control.Exception (Exception (..), SomeException (..), catch)+import Control.Monad (unless)+import Control.Monad.IO.Class (MonadIO (..))+import Data.Int (Int64)+import Data.Machine (ProcessT, await, construct, yield)+import Data.Semigroup (Sum (..))+import Data.Text (Text)+import Data.Vector qualified as V+import GHC.Eventlog.Live.Machine.Core (Tick (..))+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+import Proto.Opentelemetry.Proto.Metrics.V1.Metrics_Fields qualified as OM+import Text.Printf (printf)++--------------------------------------------------------------------------------+-- OpenTelemetry gRPC Exporters+--------------------------------------------------------------------------------++--------------------------------------------------------------------------------+-- OpenTelemetry Exporter Result for Metrics++data ExportMetricsResult+  = ExportMetricsResult+  { exportedDataPoints :: !Int64+  , rejectedDataPoints :: !Int64+  , maybeSomeException :: Maybe SomeException+  }+  deriving (Show)++pattern ExportMetricsSuccess :: Int64 -> ExportMetricsResult+pattern ExportMetricsSuccess exportedDataPoints =+  ExportMetricsResult exportedDataPoints 0 Nothing++pattern ExportMetricsError :: Int64 -> Int64 -> SomeException -> ExportMetricsResult+pattern ExportMetricsError exportedDataPoints rejectedDataPoints someException =+  ExportMetricsResult exportedDataPoints rejectedDataPoints (Just someException)++data RejectedMetricsError+  = RejectedMetricsError+  { rejectedDataPoints :: !Int64+  , errorMessage :: !Text+  }+  deriving (Show)++instance Exception RejectedMetricsError where+  displayException :: RejectedMetricsError -> String+  displayException RejectedMetricsError{..} =+    printf "Error: OpenTelemetry Collector rejected %d data points with message: %s" rejectedDataPoints errorMessage++--------------------------------------------------------------------------------+-- OpenTelemetry gRPC Exporter for Metrics++exportResourceMetrics ::+  G.Connection ->+  ProcessT IO (Tick OMS.ExportMetricsServiceRequest) (Tick ExportMetricsResult)+exportResourceMetrics conn = construct $ go False+ where+  go exportedResourceMetrics =+    await >>= \case+      Tick -> do+        unless exportedResourceMetrics $+          yield (Item $ ExportMetricsSuccess 0)+        yield Tick+        go False+      Item exportMetricsServiceRequest -> do+        exportMetricsResult <- liftIO (sendResourceMetrics exportMetricsServiceRequest)+        yield (Item exportMetricsResult)+        go True++  sendResourceMetrics :: OMS.ExportMetricsServiceRequest -> IO ExportMetricsResult+  sendResourceMetrics exportMetricsServiceRequest =+    doGrpc `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)++    handleSomeException :: SomeException -> IO ExportMetricsResult+    handleSomeException someException = pure $ ExportMetricsError 0 exportedDataPoints someException++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++--------------------------------------------------------------------------------+-- Internal Helpers+--------------------------------------------------------------------------------++{- |+Internal helper.+Count the number of `OM.NumberDataPoint` values in an `OMS.ExportMetricsServiceRequest`.+-}+{-# SPECIALIZE countDataPointsInExportMetricsServiceRequest :: OMS.ExportMetricsServiceRequest -> Int64 #-}+{-# SPECIALIZE countDataPointsInExportMetricsServiceRequest :: OMS.ExportMetricsServiceRequest -> Word #-}+countDataPointsInExportMetricsServiceRequest :: (Integral i) => OMS.ExportMetricsServiceRequest -> i+countDataPointsInExportMetricsServiceRequest exportMetricsServiceRequest =+  getSum $ foldMap (Sum . countDataPointsInResourceMetrics) (exportMetricsServiceRequest ^. OMS.vec'resourceMetrics)++{- |+Internal helper.+Count the number of `OM.NumberDataPoint` values in an `OM.ResourceMetrics`.+-}+{-# SPECIALIZE countDataPointsInResourceMetrics :: OM.ResourceMetrics -> Int64 #-}+{-# SPECIALIZE countDataPointsInResourceMetrics :: OM.ResourceMetrics -> Word #-}+countDataPointsInResourceMetrics :: (Integral i) => OM.ResourceMetrics -> i+countDataPointsInResourceMetrics resourceMetrics =+  getSum $ foldMap (Sum . countDataPointsInScopeMetrics) (resourceMetrics ^. OM.vec'scopeMetrics)++{- |+Internal helper.+Count the number of `OM.NumberDataPoint` values in an `OM.ScopeMetrics`.+-}+{-# SPECIALIZE countDataPointsInScopeMetrics :: OM.ScopeMetrics -> Int64 #-}+{-# SPECIALIZE countDataPointsInScopeMetrics :: OM.ScopeMetrics -> Word #-}+countDataPointsInScopeMetrics :: (Integral i) => OM.ScopeMetrics -> i+countDataPointsInScopeMetrics scopeMetrics =+  getSum $ foldMap (Sum . countDataPointsInMetric) (scopeMetrics ^. OM.vec'metrics)++{- |+Internal helper.+Count the number of `OM.NumberDataPoint` values in an `OM.Metric`.+-}+{-# SPECIALIZE countDataPointsInMetric :: OM.Metric -> Int64 #-}+{-# SPECIALIZE countDataPointsInMetric :: OM.Metric -> Word #-}+countDataPointsInMetric :: (Integral i) => OM.Metric -> i+countDataPointsInMetric metric =+  fromIntegral $+    case metric ^. OM.maybe'data' of+      Nothing -> 0+      Just (OM.Metric'Gauge gauge) ->+        V.length (gauge ^. OM.vec'dataPoints)+      Just (OM.Metric'Sum sum_) ->+        V.length (sum_ ^. OM.vec'dataPoints)+      Just (OM.Metric'Histogram histogram) ->+        V.length (histogram ^. OM.vec'dataPoints)+      Just (OM.Metric'ExponentialHistogram exponentialHistogram) ->+        V.length (exponentialHistogram ^. OM.vec'dataPoints)+      Just (OM.Metric'Summary summary) ->+        V.length (summary ^. OM.vec'dataPoints)
+ src/GHC/Eventlog/Live/Otelcol/Exporter/Profiles.hs view
@@ -0,0 +1,128 @@+{-# OPTIONS_GHC -Wno-orphans #-}++module GHC.Eventlog.Live.Otelcol.Exporter.Profiles (+  -- * Profiles+  ExportProfileResult (..),+  RejectedProfilesError (..),+  exportResourceProfiles,+)+where++import Control.Exception (Exception (..), SomeException (..), catch)+import Control.Monad (unless)+import Control.Monad.IO.Class (MonadIO (..))+import Data.Int (Int64)+import Data.Machine (ProcessT, await, construct, yield)+import Data.Semigroup (Sum (..))+import Data.Text (Text)+import Data.Vector qualified as V+import GHC.Eventlog.Live.Machine.Core (Tick (..))+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+import Proto.Opentelemetry.Proto.Profiles.V1development.Profiles_Fields qualified as OP+import Text.Printf (printf)++data ExportProfileResult+  = ExportProfileResult+  { exportedProfiles :: !Int64+  , rejectedProfiles :: !Int64+  , maybeSomeException :: Maybe SomeException+  }+  deriving (Show)++pattern ExportProfileSuccess :: Int64 -> ExportProfileResult+pattern ExportProfileSuccess exportedProfiles =+  ExportProfileResult exportedProfiles 0 Nothing++pattern ExportProfileError :: Int64 -> Int64 -> SomeException -> ExportProfileResult+pattern ExportProfileError exportedProfiles rejectedProfiles someException =+  ExportProfileResult exportedProfiles rejectedProfiles (Just someException)++data RejectedProfilesError+  = RejectedProfilesError+  { rejectedProfiles :: !Int64+  , errorMessage :: !Text+  }+  deriving (Show)++instance Exception RejectedProfilesError where+  displayException :: RejectedProfilesError -> String+  displayException RejectedProfilesError{..} =+    printf "Error: OpenTelemetry Collector rejectedProfiles %d data points with message: %s" rejectedProfiles errorMessage++--------------------------------------------------------------------------------+-- OpenTelemetry gRPC Exporter for Profiles++exportResourceProfiles ::+  G.Connection ->+  ProcessT IO (Tick OPS.ExportProfilesServiceRequest) (Tick ExportProfileResult)+exportResourceProfiles conn =+  construct $ go False+ where+  go exportedProfiles =+    await >>= \case+      Tick -> do+        unless exportedProfiles $+          yield (Item $ ExportProfileSuccess 0)+        yield Tick+        go False+      Item exportProfilesServiceRequest -> do+        exportTraceResult <- liftIO (sendResourceProfiles exportProfilesServiceRequest)+        yield (Item exportTraceResult)+        go True++  sendResourceProfiles :: OPS.ExportProfilesServiceRequest -> IO ExportProfileResult+  sendResourceProfiles exportProfilesServiceRequest =+    doGrpc `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)++    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++{- |+Internal helper.+Count the number of 'OP.Sample' values in an 'OPS.ExportProfilesServiceRequest'.+-}+{-# SPECIALIZE countSamplesInExportProfileServiceRequest :: OPS.ExportProfilesServiceRequest -> Int64 #-}+{-# SPECIALIZE countSamplesInExportProfileServiceRequest :: OPS.ExportProfilesServiceRequest -> Word #-}+countSamplesInExportProfileServiceRequest :: (Integral i) => OPS.ExportProfilesServiceRequest -> i+countSamplesInExportProfileServiceRequest exportProfileServiceRequest =+  getSum $ foldMap (Sum . countSamplesInResourceProfiles) (exportProfileServiceRequest ^. OPS.vec'resourceProfiles)++countSamplesInResourceProfiles :: (Integral i) => OP.ResourceProfiles -> i+countSamplesInResourceProfiles resourceProfiles =+  getSum $ foldMap (Sum . countSamplesInScopeProfiles) (resourceProfiles ^. OP.vec'scopeProfiles)++countSamplesInScopeProfiles :: (Integral i) => OP.ScopeProfiles -> i+countSamplesInScopeProfiles scopeProfiles =+  getSum $ foldMap (Sum . countSamplesInProfile) (scopeProfiles ^. OP.vec'profiles)++countSamplesInProfile :: (Integral i) => OP.Profile -> i+countSamplesInProfile profile =+  fromIntegral $+    V.length (profile ^. OP.vec'samples)
+ src/GHC/Eventlog/Live/Otelcol/Exporter/Traces.hs view
@@ -0,0 +1,140 @@+{-# OPTIONS_GHC -Wno-orphans #-}++module GHC.Eventlog.Live.Otelcol.Exporter.Traces (+  -- * Traces+  ExportTraceResult (..),+  RejectedSpansError (..),+  exportResourceSpans,+) where++import Control.Exception (Exception (..), SomeException (..), catch)+import Control.Monad (unless)+import Control.Monad.IO.Class (MonadIO (..))+import Data.Int (Int64)+import Data.Machine (ProcessT, await, construct, yield)+import Data.Semigroup (Sum (..))+import Data.Text (Text)+import Data.Vector qualified as V+import GHC.Eventlog.Live.Machine.Core (Tick (..))+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+import Proto.Opentelemetry.Proto.Trace.V1.Trace_Fields qualified as OT+import Text.Printf (printf)++--------------------------------------------------------------------------------+-- OpenTelemetry Exporter Result for Traces++data ExportTraceResult+  = ExportTraceResult+  { exportedSpans :: !Int64+  , rejectedSpans :: !Int64+  , maybeSomeException :: Maybe SomeException+  }+  deriving (Show)++pattern ExportTraceSuccess :: Int64 -> ExportTraceResult+pattern ExportTraceSuccess exportedSpans =+  ExportTraceResult exportedSpans 0 Nothing++pattern ExportTraceError :: Int64 -> Int64 -> SomeException -> ExportTraceResult+pattern ExportTraceError exportedSpans rejectedSpans someException =+  ExportTraceResult exportedSpans rejectedSpans (Just someException)++data RejectedSpansError+  = RejectedSpansError+  { rejectedSpans :: !Int64+  , errorMessage :: !Text+  }+  deriving (Show)++instance Exception RejectedSpansError where+  displayException :: RejectedSpansError -> String+  displayException RejectedSpansError{..} =+    printf "Error: OpenTelemetry Collector rejectedSpans %d spans with message: %s" rejectedSpans errorMessage++--------------------------------------------------------------------------------+-- OpenTelemetry gRPC Exporter for Traces++exportResourceSpans ::+  G.Connection ->+  ProcessT IO (Tick OTS.ExportTraceServiceRequest) (Tick ExportTraceResult)+exportResourceSpans conn =+  construct $ go False+ where+  go exportedResourceSpans =+    await >>= \case+      Tick -> do+        unless exportedResourceSpans $+          yield (Item $ ExportTraceSuccess 0)+        yield Tick+        go False+      Item exportTraceServiceRequest -> do+        exportTraceResult <- liftIO (sendResourceSpans exportTraceServiceRequest)+        yield (Item exportTraceResult)+        go True++  sendResourceSpans :: OTS.ExportTraceServiceRequest -> IO ExportTraceResult+  sendResourceSpans exportTraceServiceRequest =+    doGrpc `catch` handleSomeException+   where+    !exportedSpans = countSpansInExportTraceServiceRequest exportTraceServiceRequest++    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)++    handleSomeException :: SomeException -> IO ExportTraceResult+    handleSomeException someException = pure $ ExportTraceError 0 exportedSpans someException++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++--------------------------------------------------------------------------------+-- Internal Helpers+--------------------------------------------------------------------------------++{- |+Internal helper.+Count the number of `OT.Span` values in an `OTS.ExportTraceServiceRequest`.+-}+{-# SPECIALIZE countSpansInExportTraceServiceRequest :: OTS.ExportTraceServiceRequest -> Int64 #-}+{-# SPECIALIZE countSpansInExportTraceServiceRequest :: OTS.ExportTraceServiceRequest -> Word #-}+countSpansInExportTraceServiceRequest :: (Integral i) => OTS.ExportTraceServiceRequest -> i+countSpansInExportTraceServiceRequest exportTraceServiceRequest =+  getSum $ foldMap (Sum . countSpansInResourceSpans) (exportTraceServiceRequest ^. OTS.vec'resourceSpans)++{- |+Internal helper.+Count the number of `OT.Span` values in an `OT.ResourceSpans`.+-}+{-# SPECIALIZE countSpansInResourceSpans :: OT.ResourceSpans -> Int64 #-}+{-# SPECIALIZE countSpansInResourceSpans :: OT.ResourceSpans -> Word #-}+countSpansInResourceSpans :: (Integral i) => OT.ResourceSpans -> i+countSpansInResourceSpans resourceSpans =+  getSum $ foldMap (Sum . countSpansInScopeSpans) (resourceSpans ^. OT.vec'scopeSpans)++{- |+Internal helper.+Count the number of `OT.Span` values in an `OT.ScopeSpans`.+-}+{-# SPECIALIZE countSpansInScopeSpans :: OT.ScopeSpans -> Int64 #-}+{-# SPECIALIZE countSpansInScopeSpans :: OT.ScopeSpans -> Word #-}+countSpansInScopeSpans :: (Integral i) => OT.ScopeSpans -> i+countSpansInScopeSpans scopeSpans =+  fromIntegral $+    V.length (scopeSpans ^. OT.vec'spans)
+ src/GHC/Eventlog/Live/Otelcol/Options.hs view
@@ -0,0 +1,224 @@+module GHC.Eventlog.Live.Otelcol.Options (+  Options (..),+  MyDebugOptions (..),+  ServiceName (..),+  OpenTelemetryCollectorOptions (..),+  options,+) where++import Control.Applicative (asum)+import Data.Default (Default (..))+import Data.Text qualified as T+import Data.Version (showVersion)+import GHC.Debug.Stub.Compat (MyGhcDebugSocket, maybeMyGhcDebugSocketParser)+import GHC.Eventlog.Live.Data.Severity (Severity (..))+import GHC.Eventlog.Live.Options+import GHC.Eventlog.Live.Otelcol.Config (ServiceName (..))+import GHC.Eventlog.Live.Otelcol.Config qualified as C+import GHC.Eventlog.Live.Otelcol.Config.Default.Raw (defaultConfigJSONSchemaString, defaultConfigString)+import GHC.Eventlog.Live.Otelcol.Config.Types (Config)+import GHC.Eventlog.Live.Otelcol.Control (ControlOptions, controlOptionsParser)+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 Paths_eventlog_live_otelcol qualified as EventlogLive++options :: O.ParserInfo Options+options =+  O.info+    ( optionsParser+        O.<**> defaultsPrinter+        O.<**> debugDefaultsPrinter+        O.<**> configJSONSchemaPrinter+        O.<**> OE.helperWith (O.long "help" <> O.help "Show this help text.")+        O.<**> OC.simpleVersioner (showVersion EventlogLive.version)+    )+    O.idm++data Options = Options+  { eventlogSourceOptions :: EventlogSourceOptions+  , eventlogSocketTimeoutS :: Double+  , eventlogSocketTimeoutExponent :: Double+  , eventlogFlushIntervalS :: Double+  , maybeEventlogLogFile :: Maybe FilePath+  , maybeHeapProfBreakdown :: Maybe HeapProfBreakdown+  , maybeServiceName :: Maybe ServiceName+  , severityThreshold :: Severity+  , stats :: Bool+  , maybeConfigFile :: Maybe FilePath+  , openTelemetryCollectorOptions :: OpenTelemetryCollectorOptions+  , controlOptions :: ControlOptions+  , myDebugOptions :: MyDebugOptions+  }++optionsParser :: O.Parser Options+optionsParser =+  Options+    <$> eventlogSourceOptionsParser+    <*> eventlogSocketTimeoutSParser+    <*> eventlogSocketTimeoutExponentParser+    <*> eventlogFlushIntervalSParser+    <*> O.optional eventlogLogFileParser+    <*> O.optional heapProfBreakdownParser+    <*> O.optional serviceNameParser+    <*> verbosityParser+    <*> statsParser+    <*> O.optional configFileParser+    <*> openTelemetryCollectorOptionsParser+    <*> controlOptionsParser+    <*> myDebugOptionsParser++--------------------------------------------------------------------------------+-- Configuration++configFileParser :: O.Parser FilePath+configFileParser =+  O.strOption+    ( O.long "config"+        <> O.metavar "FILE"+        <> O.help "The path to a detailed configuration file."+    )++defaultsPrinter :: O.Parser (a -> a)+defaultsPrinter =+  O.infoOption defaultConfigString . mconcat $+    [ O.long "print-defaults"+    , O.help "Print default configuration options."+    ]++configJSONSchemaPrinter :: O.Parser (a -> a)+configJSONSchemaPrinter =+  O.infoOption defaultConfigJSONSchemaString . mconcat $+    [ O.long "print-config-json-schema"+    , O.help "Print JSON Schema for configuration format."+    ]++debugDefaultsPrinter :: O.Parser (a -> a)+debugDefaultsPrinter =+  O.infoOption defaultConfigDebugString . mconcat $+    [ O.long "print-defaults-debug"+    , O.help "Print default configuration options using the parsed representation."+    , O.internal+    ]+ where+  defaultConfigDebugString =+    T.unpack . C.prettyConfig $ (def :: Config)++--------------------------------------------------------------------------------+-- Service Name++serviceNameParser :: O.Parser ServiceName+serviceNameParser =+  ServiceName+    <$> O.strOption+      ( O.long "service-name"+          <> O.metavar "STRING"+          <> O.help "The name of the profiled service."+      )++--------------------------------------------------------------------------------+-- OpenTelemetry Collector configuration++newtype OpenTelemetryCollectorOptions = OpenTelemetryCollectorOptions+  { openTelemetryCollectorServer :: G.Server+  }++openTelemetryCollectorOptionsParser :: O.Parser OpenTelemetryCollectorOptions+openTelemetryCollectorOptionsParser =+  OC.parserOptionGroup "OpenTelemetry Collector Server Options" $+    OpenTelemetryCollectorOptions+      <$> otelcolServerParser++otelcolServerParser :: O.Parser G.Server+otelcolServerParser =+  makeServer+    <$> otelcolAddressParser+    <*> O.switch (O.long "otelcol-ssl" <> O.help "Use SSL.")+    <*> otelcolServerValidationParser+    <*> otelcolSslKeyLogParser+ where+  makeServer :: G.Address -> Bool -> G.ServerValidation -> G.SslKeyLog -> G.Server+  makeServer address ssl serverValidation sslKeyLog+    | ssl = G.ServerSecure serverValidation sslKeyLog address+    | otherwise = G.ServerInsecure address++otelcolAddressParser :: O.Parser G.Address+otelcolAddressParser =+  G.Address+    <$> O.strOption+      ( O.long "otelcol-host"+          <> O.metavar "HOST"+          <> O.help "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."+          )+      )++otelcolServerValidationParser :: O.Parser G.ServerValidation+otelcolServerValidationParser =+  asum+    [ G.ValidateServer <$> otelcolCertificateStoreSpecParser+    , pure G.NoServerValidation+    ]+ where+  otelcolCertificateStoreSpecParser :: O.Parser G.CertificateStoreSpec+  otelcolCertificateStoreSpecParser =+    makeCertificateStoreSpec+      <$> O.optional+        ( O.strOption+            ( O.long "otelcol-certificate-store"+                <> O.metavar "FILE"+                <> O.help "Store for certificate validation."+            )+        )+   where+    makeCertificateStoreSpec :: Maybe FilePath -> G.CertificateStoreSpec+    makeCertificateStoreSpec = maybe G.certStoreFromSystem G.certStoreFromPath++otelcolSslKeyLogParser :: O.Parser G.SslKeyLog+otelcolSslKeyLogParser =+  asum+    [ G.SslKeyLogPath+        <$> O.strOption+          ( O.long "otelcol-ssl-key-log"+              <> O.metavar "FILE"+              <> O.help "Use file to log SSL keys."+          )+    , O.flag+        G.SslKeyLogNone+        G.SslKeyLogFromEnv+        ( O.long "otelcol-ssl-key-log-from-env"+            <> O.help "Use SSLKEYLOGFILE to log SSL keys."+        )+    ]++--------------------------------------------------------------------------------+-- Debug Options++data MyDebugOptions = MyDebugOptions+  { maybeMyEventlogSocket :: Maybe MyEventlogSocket+  , maybeMyGhcDebugSocket :: Maybe MyGhcDebugSocket+  }++myDebugOptionsParser :: O.Parser MyDebugOptions+myDebugOptionsParser =+  OC.parserOptionGroup "Debug Options" $+    MyDebugOptions+      <$> maybeMyEventlogSocketParser+      <*> maybeMyGhcDebugSocketParser
+ src/GHC/Eventlog/Live/Otelcol/Processor/Common/Core.hs view
@@ -0,0 +1,68 @@+{- |+Module      : GHC.Eventlog.Live.Otelcol.Processor.Common.Core+Description : Common utilities shared across telemetry data types.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Otelcol.Processor.Common.Core (+  messageWith,+  runIf,+  ifNonEmpty,+  toMaybeKeyValue,+)+where++import Data.Functor ((<&>))+import Data.Machine (MachineT, stopped)+import Data.ProtoLens (Message (..))+import GHC.Eventlog.Live.Data.Attribute (Attr, AttrValue (..))+import Lens.Family2 ((.~))+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++-- | Run a machine if a boolean is @True@, otherwise stop.+runIf :: (Monad m) => Bool -> MachineT m k o -> MachineT m k o+runIf b m = if b then m else stopped++-- | Return the second argument if the first argument is non-empty.+ifNonEmpty :: [a] -> b -> Maybe b+ifNonEmpty xs r = if null xs then Nothing else Just r++--------------------------------------------------------------------------------+-- Interpret attributes++{- |+Convert an `Attr` to an OTLP `OC.KeyValue`.+-}+toMaybeKeyValue :: Attr -> Maybe OC.KeyValue+toMaybeKeyValue (k, v) =+  toMaybeAnyValue v <&> \v' ->+    messageWith+      [ OC.key .~ k+      , OC.value .~ v'+      ]++{- |+Internal helper.+Convert an `AttrValue` to an OTLP `OC.AnyValue`.+-}+toMaybeAnyValue :: AttrValue -> Maybe OC.AnyValue+toMaybeAnyValue = \case+  AttrBool v -> Just $ messageWith [OC.boolValue .~ v]+  AttrInt v -> Just $ messageWith [OC.intValue .~ fromIntegral v]+  AttrInt8 v -> Just $ messageWith [OC.intValue .~ fromIntegral v]+  AttrInt16 v -> Just $ messageWith [OC.intValue .~ fromIntegral v]+  AttrInt32 v -> Just $ messageWith [OC.intValue .~ fromIntegral v]+  AttrInt64 v -> Just $ messageWith [OC.intValue .~ v]+  AttrWord v -> Just $ messageWith [OC.intValue .~ fromIntegral v]+  AttrWord8 v -> Just $ messageWith [OC.intValue .~ fromIntegral v]+  AttrWord16 v -> Just $ messageWith [OC.intValue .~ fromIntegral v]+  AttrWord32 v -> Just $ messageWith [OC.intValue .~ fromIntegral v]+  AttrWord64 v -> Just $ messageWith [OC.intValue .~ fromIntegral v]+  AttrDouble v -> Just $ messageWith [OC.doubleValue .~ v]+  AttrText v -> Just $ messageWith [OC.stringValue .~ v]+  AttrNull -> Nothing
+ src/GHC/Eventlog/Live/Otelcol/Processor/Common/Logs.hs view
@@ -0,0 +1,84 @@+{-# LANGUAGE OverloadedStrings #-}++{- |+Module      : GHC.Eventlog.Live.Otelcol.Processor.Common.Logs+Description : Profile Processors for OTLP.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Otelcol.Processor.Common.Logs (+  ToLogRecord (..),+  toExportLogsServiceRequest,+  toResourceLogs,+  toScopeLogs,+)+where++import Data.Function ((&))+import Data.Maybe (fromMaybe, mapMaybe)+import Data.ProtoLens (Message (..))+import Data.Text (Text)+import GHC.Eventlog.Live.Data.Attribute ((~=))+import GHC.Eventlog.Live.Data.LogRecord (LogRecord (..))+import GHC.Eventlog.Live.Data.Severity (Severity)+import GHC.Eventlog.Live.Data.Severity qualified as DS+import GHC.Eventlog.Live.Machine.Analysis.Thread qualified as M+import GHC.Eventlog.Live.Otelcol.Processor.Common.Core (ifNonEmpty, messageWith, toMaybeKeyValue)+import GHC.IsList (IsList (..))+import Lens.Family2 ((.~))+import Proto.Opentelemetry.Proto.Collector.Logs.V1.LogsService qualified as OLS+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+import Proto.Opentelemetry.Proto.Logs.V1.Logs_Fields qualified as OL+import Proto.Opentelemetry.Proto.Resource.V1.Resource qualified as OR++toExportLogsServiceRequest :: [OL.ResourceLogs] -> OLS.ExportLogsServiceRequest+toExportLogsServiceRequest = (defMessage &) . (OL.resourceLogs .~)++toResourceLogs :: OR.Resource -> [OL.ScopeLogs] -> Maybe OL.ResourceLogs+toResourceLogs resource scopeLogs =+  ifNonEmpty scopeLogs $+    messageWith [OL.resource .~ resource, OL.scopeLogs .~ scopeLogs]++toScopeLogs :: OC.InstrumentationScope -> [OL.LogRecord] -> Maybe OL.ScopeLogs+toScopeLogs instrumentationScope logRecords =+  ifNonEmpty logRecords $+    messageWith [OL.scope .~ instrumentationScope, OL.logRecords .~ logRecords]++--------------------------------------------------------------------------------+-- Interpret logs++class ToLogRecord v where+  toLogRecord :: v -> OL.LogRecord++instance ToLogRecord LogRecord where+  toLogRecord :: LogRecord -> OL.LogRecord+  toLogRecord i =+    messageWith+      [ OL.body .~ messageWith [OC.stringValue .~ i.body]+      , OL.timeUnixNano .~ fromMaybe 0 i.maybeTimeUnixNano+      , -- TODO: this could be set to the actual observed time in the processor.+        OL.observedTimeUnixNano .~ fromMaybe 0 i.maybeTimeUnixNano+      , OL.attributes .~ mapMaybe toMaybeKeyValue (toList i.attrs)+      , OL.severityNumber .~ toSeverityNumber i.maybeSeverity+      ]+   where+    toSeverityNumber :: Maybe Severity -> OL.SeverityNumber+    toSeverityNumber = maybe OL.SEVERITY_NUMBER_UNSPECIFIED (toEnum . (.value) . DS.toSeverityNumber)++instance ToLogRecord M.ThreadLabel where+  toLogRecord :: M.ThreadLabel -> OL.LogRecord+  toLogRecord i =+    messageWith+      [ OL.body .~ messageWith [OC.stringValue .~ i.threadlabel]+      , OL.timeUnixNano .~ i.startTimeUnixNano+      , -- TODO: this could be set to the actual observed time in the processor.+        OL.observedTimeUnixNano .~ i.startTimeUnixNano+      , OL.attributes+          .~ mapMaybe+            toMaybeKeyValue+            [ "kind" ~= ("ThreadLabel" :: Text)+            , "thread" ~= i.thread+            ]+      ]
+ src/GHC/Eventlog/Live/Otelcol/Processor/Common/Metrics.hs view
@@ -0,0 +1,288 @@+{- |+Module      : GHC.Eventlog.Live.Otelcol.Processor.Common.Metrics+Description : Common utilities for metric processors.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Otelcol.Processor.Common.Metrics (+  MetricProcessor (..),+  runMetricProcessor,+  MetricAggregators (..),+  viaSum,+  viaLast,+  asGauge,+  asSum,+  toScopeMetrics,+  toResourceMetrics,+  toExportMetricsServiceRequest,+)+where++import Control.Monad (unless)+import Data.Coerce (Coercible, coerce)+import Data.DList (DList)+import Data.DList qualified as D+import Data.Default (Default)+import Data.Function ((&))+import Data.Int (Int16, Int32, Int64, Int8)+import Data.Kind (Type)+import Data.Machine (Process, ProcessT, asParts, await, echo, mapping, repeatedly, yield, (~>))+import Data.Maybe (fromMaybe, mapMaybe)+import Data.ProtoLens (Message (..))+import Data.Proxy (Proxy (..))+import Data.Semigroup (Last (..), Sum (..))+import Data.Text (Text)+import Data.Word (Word16, Word32, Word64, Word8)+import GHC.Eventlog.Live.Data.Group (Group, GroupBy, GroupedBy)+import GHC.Eventlog.Live.Data.Group qualified as DG+import GHC.Eventlog.Live.Data.Metric (Metric (..))+import GHC.Eventlog.Live.Machine.Core (Tick)+import GHC.Eventlog.Live.Machine.Core 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.Processor.Common.Core (ifNonEmpty, messageWith, runIf, toMaybeKeyValue)+import GHC.IsList (IsList (..))+import GHC.Records (HasField (..))+import GHC.TypeLits (Symbol)+import Lens.Family2 ((.~))+import Proto.Opentelemetry.Proto.Collector.Metrics.V1.MetricsService qualified as OMS+import Proto.Opentelemetry.Proto.Common.V1.Common qualified as OC+import Proto.Opentelemetry.Proto.Metrics.V1.Metrics qualified as OM+import Proto.Opentelemetry.Proto.Metrics.V1.Metrics_Fields qualified as OM+import Proto.Opentelemetry.Proto.Resource.V1.Resource qualified as OR++--------------------------------------------------------------------------------+-- Generic Metric Processor++type MetricProcessor :: Symbol -> Type -> (Type -> Type) -> Type -> Type -> Type -> Type -> Type+data MetricProcessor metricProcessor metricProcessorConfig m a b c d+  = ( Monad m+    , HasField metricProcessor C.Metrics (Maybe metricProcessorConfig)+    , C.IsMetricProcessorConfig metricProcessorConfig+    , IsNumberDataPoint'Value d+    ) =>+  MetricProcessor+  { metricProcessorProxy :: !(Proxy metricProcessor)+  -- ^ The metric's field name in `C.Metrics`.+  , dataProcessor :: !(ProcessT m a b)+  -- ^ The metric's data processor+  , aggregators :: !(MetricAggregators b c)+  -- ^ The metric's aggregator+  , postProcessor :: !(ProcessT m c (Metric d))+  -- ^ The metric's post processor+  , unit :: !Text+  -- ^ The metric's unit (in UCUM format).+  , asMetric'Data :: !(ProcessT m [OM.NumberDataPoint] OM.Metric'Data)+  -- ^ A process to wrap data points as `OM.Metric'Data`.+  }++{- |+Internal helper.+Run a `MetricProcessor`.+-}+runMetricProcessor ::+  forall metricProcessor metricProcessorConfig m a b c d.+  (Default metricProcessorConfig) =>+  MetricProcessor metricProcessor metricProcessorConfig m a b c d ->+  -- | The full configuration.+  FullConfig ->+  ProcessT m (Tick a) (Tick (DList OM.Metric))+runMetricProcessor MetricProcessor{..} fullConfig =+  let metricProcessorConfig :: C.Metrics -> Maybe metricProcessorConfig+      metricProcessorConfig = getField @metricProcessor+   in runIf (C.processorEnabled (.metrics) metricProcessorConfig fullConfig) $+        M.liftTick dataProcessor+          ~> aggregate aggregators (C.processorAggregationBatches (.metrics) metricProcessorConfig fullConfig)+          ~> M.liftTick postProcessor+          ~> mapping (fmap (D.singleton . toNumberDataPoint))+          ~> M.batchByTicks (C.processorExportBatches (.metrics) metricProcessorConfig fullConfig)+          ~> M.liftTick+            ( mapping D.toList+                ~> asMetric'Data+                ~> asMetricWith fullConfig metricProcessorConfig [OM.unit .~ unit]+                ~> mapping D.singleton+            )+{-# INLINE runMetricProcessor #-}++asMetricWith ::+  (Default a, HasField "description" a (Maybe Text), HasField "name" a (Maybe Text)) =>+  FullConfig ->+  (C.Metrics -> Maybe a) ->+  [OM.Metric -> OM.Metric] ->+  Process OM.Metric'Data OM.Metric+asMetricWith fullConfig field f =+  asMetric $+    [ OM.name .~ C.processorName (.metrics) field fullConfig+    , maybe id (OM.description .~) $ C.processorDescription (.metrics) field fullConfig+    ]+      <> f++asMetric :: [OM.Metric -> OM.Metric] -> Process OM.Metric'Data OM.Metric+asMetric f = mapping $ toMetric f++toMetric :: [OM.Metric -> OM.Metric] -> OM.Metric'Data -> OM.Metric+toMetric f metric'data = messageWith ((OM.maybe'data' .~ Just metric'data) : f)++toExportMetricsServiceRequest :: [OM.ResourceMetrics] -> OMS.ExportMetricsServiceRequest+toExportMetricsServiceRequest = (defMessage &) . (OM.resourceMetrics .~)++toResourceMetrics :: OR.Resource -> [OM.ScopeMetrics] -> Maybe OM.ResourceMetrics+toResourceMetrics resource scopeMetrics =+  ifNonEmpty scopeMetrics $+    messageWith [OM.resource .~ resource, OM.scopeMetrics .~ scopeMetrics]++toScopeMetrics :: OC.InstrumentationScope -> [OM.Metric] -> Maybe OM.ScopeMetrics+toScopeMetrics instrumentationScope metrics =+  ifNonEmpty metrics $+    messageWith [OM.scope .~ instrumentationScope, OM.metrics .~ metrics]++asGauge :: Process [OM.NumberDataPoint] OM.Metric'Data+asGauge =+  repeatedly $ do+    await >>= \dataPoints ->+      unless (null dataPoints) $+        yield (toGauge dataPoints)++toGauge :: [OM.NumberDataPoint] -> OM.Metric'Data+toGauge dataPoints = OM.Metric'Gauge . messageWith $ [OM.dataPoints .~ dataPoints]++asSum :: [OM.Sum -> OM.Sum] -> Process [OM.NumberDataPoint] OM.Metric'Data+asSum f =+  repeatedly $+    await >>= \dataPoints ->+      unless (null dataPoints) $+        yield (toSum f dataPoints)++toSum :: [OM.Sum -> OM.Sum] -> [OM.NumberDataPoint] -> OM.Metric'Data+toSum f dataPoints = OM.Metric'Sum . messageWith $ (OM.dataPoints .~ dataPoints) : f++--------------------------------------------------------------------------------+-- Metric Aggregation++data MetricAggregators a b = MetricAggregators+  { nothing :: Process (Tick a) (Tick b)+  , byBatches :: Int -> Process (Tick a) (Tick b)+  }++{- |+Internal helper.+Aggregate items based on the provided aggregators and aggregation strategy.+-}+aggregate :: MetricAggregators a b -> Int -> Process (Tick a) (Tick b)+aggregate MetricAggregators{..} aggregationBatches+  | aggregationBatches >= 1 = byBatches aggregationBatches+  | otherwise = nothing++{- |+Internal helper.+Metric aggregators via the `Semigroup` instance for `Sum`.+-}+viaSum :: forall a. (Num a) => MetricAggregators (Metric a) (Metric a)+viaSum =+  MetricAggregators+    { nothing = echo+    , byBatches = \ticks ->+        -- TODO: Yield group sample counts as separate metric.+        batchByTicksVia ticks (Proxy @(Metric (Sum a)))+          ~> M.liftTick (mapping (fmap (.representative)) ~> asParts)+    }++{- |+Internal helper.+Metric aggregators via the `Semigroup` instance for `Last`.+-}+viaLast :: forall a. (GroupBy a) => MetricAggregators a a+viaLast =+  MetricAggregators+    { nothing = echo+    , byBatches = \ticks ->+        -- TODO: Yield group sample counts as separate metric.+        batchByTicksVia ticks (Proxy @(Last a))+          ~> M.liftTick (mapping (fmap (.representative)) ~> asParts)+    }++{- |+Internal helper.+This function aggregates items via a `Semigroup` instance and grouped by the `GroupBy` instance.+-}+batchByTicksVia ::+  forall a b.+  (Coercible a b, GroupBy b, Semigroup b) =>+  -- | The number of ticks per batch.+  Int ->+  Proxy b ->+  Process (Tick a) (Tick [Group a])+batchByTicksVia ticks (Proxy :: Proxy b) =+  mapping (fmap DG.singleton . coerce @(Tick a) @(Tick b))+    ~> M.batchByTicks @(GroupedBy b) ticks+    ~> M.liftTick (mapping (coerce @[Group b] @[Group a] . DG.groups))++{- |+Internal helper.+Convert a metric datapoint to an `OM.NumberDataPoint`.+-}+toNumberDataPoint :: (IsNumberDataPoint'Value v) => Metric v -> OM.NumberDataPoint+toNumberDataPoint i =+  messageWith+    [ OM.maybe'value .~ Just (toNumberDataPoint'Value i.value)+    , OM.timeUnixNano .~ fromMaybe 0 i.maybeTimeUnixNano+    , OM.startTimeUnixNano .~ fromMaybe 0 i.maybeStartTimeUnixNano+    , OM.attributes .~ mapMaybe toMaybeKeyValue (toList i.attrs)+    ]++{- |+Internal helper.+Class of types that can be converted to `OM.NumberDataPoint'Value` values.+-}+class IsNumberDataPoint'Value v where+  toNumberDataPoint'Value :: v -> OM.NumberDataPoint'Value++instance IsNumberDataPoint'Value Float where+  toNumberDataPoint'Value :: Float -> OM.NumberDataPoint'Value+  toNumberDataPoint'Value = OM.NumberDataPoint'AsDouble . realToFrac++instance IsNumberDataPoint'Value Double where+  toNumberDataPoint'Value :: Double -> OM.NumberDataPoint'Value+  toNumberDataPoint'Value = OM.NumberDataPoint'AsDouble++instance IsNumberDataPoint'Value Word8 where+  toNumberDataPoint'Value :: Word8 -> OM.NumberDataPoint'Value+  toNumberDataPoint'Value = OM.NumberDataPoint'AsInt . fromIntegral++instance IsNumberDataPoint'Value Word16 where+  toNumberDataPoint'Value :: Word16 -> OM.NumberDataPoint'Value+  toNumberDataPoint'Value = OM.NumberDataPoint'AsInt . fromIntegral++instance IsNumberDataPoint'Value Word32 where+  toNumberDataPoint'Value :: Word32 -> OM.NumberDataPoint'Value+  toNumberDataPoint'Value = OM.NumberDataPoint'AsInt . fromIntegral++-- | __Warning__: This instance may cause overflow.+instance IsNumberDataPoint'Value Word64 where+  toNumberDataPoint'Value :: Word64 -> OM.NumberDataPoint'Value+  toNumberDataPoint'Value = OM.NumberDataPoint'AsInt . fromIntegral++-- | __Warning__: This instance may cause overflow.+instance IsNumberDataPoint'Value Word where+  toNumberDataPoint'Value :: Word -> OM.NumberDataPoint'Value+  toNumberDataPoint'Value = OM.NumberDataPoint'AsInt . fromIntegral++instance IsNumberDataPoint'Value Int8 where+  toNumberDataPoint'Value :: Int8 -> OM.NumberDataPoint'Value+  toNumberDataPoint'Value = OM.NumberDataPoint'AsInt . fromIntegral++instance IsNumberDataPoint'Value Int16 where+  toNumberDataPoint'Value :: Int16 -> OM.NumberDataPoint'Value+  toNumberDataPoint'Value = OM.NumberDataPoint'AsInt . fromIntegral++instance IsNumberDataPoint'Value Int32 where+  toNumberDataPoint'Value :: Int32 -> OM.NumberDataPoint'Value+  toNumberDataPoint'Value = OM.NumberDataPoint'AsInt . fromIntegral++instance IsNumberDataPoint'Value Int64 where+  toNumberDataPoint'Value :: Int64 -> OM.NumberDataPoint'Value+  toNumberDataPoint'Value = OM.NumberDataPoint'AsInt++instance IsNumberDataPoint'Value Int where+  toNumberDataPoint'Value :: Int -> OM.NumberDataPoint'Value+  toNumberDataPoint'Value = OM.NumberDataPoint'AsInt . fromIntegral
+ src/GHC/Eventlog/Live/Otelcol/Processor/Common/Profiles.hs view
@@ -0,0 +1,207 @@+{-# LANGUAGE OverloadedStrings #-}++{- |+Module      : GHC.Eventlog.Live.Otelcol.Processor.Common.Profiles+Description : Abstraction over ProfilesDictionary for the OTLP protocol.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Otelcol.Processor.Common.Profiles (+  toExportProfileServiceRequest,++  -- * Dictionary for deduplication logic of common values+  ProfileDictionary,+  emptyProfileDictionary,++  -- * Turn a 'ProfileDictionary' into a 'OP.ProfilesDictionary'+  toProfilesDictionary,++  -- * Retrieve the 'SymbolIndex' for various 'OP.ProfilesData' fields+  SymbolIndex,+  getLocation,+  getFunction,+  getString,+  getMapping,+  getLink,+  getAttribute,+  getStack,+)+where++import Control.Monad.Trans.State.Strict (StateT)+import Control.Monad.Trans.State.Strict qualified as State+import Data.Int (Int32)+import Data.Map.Strict (Map)+import Data.Map.Strict qualified as Map+import Data.ProtoLens (Message (..))+import Data.Text (Text)+import GHC.Eventlog.Live.Otelcol.Processor.Common.Core (messageWith)+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+    ]++type SymbolIndex = Int32++data ProfileDictionary = ProfileDictionary+  { locationTable :: CommonSymbolTable OP.Location+  -- ^ Common 'OP.Location' table, first entry is the 'defMessage'.+  --   This holds for OTLP 1.9.0.+  , functionTable :: CommonSymbolTable OP.Function+  -- ^ Common 'OP.Function' table, first entry is the 'defMessage'.+  --   This holds for OTLP 1.9.0.+  , stringTable :: CommonSymbolTable Text+  -- ^ Common string table, the first entry must be "" per the protobuf+  --   documentation.+  --+  --   @+  --    // A common table for strings referenced by various messages.+  --    // string_table[0] must always be "".+  --    repeated string string_table = 5;+  --   @+  , mappingTable :: CommonSymbolTable OP.Mapping+  -- ^ Common 'OP.Mapping' table, first entry is the 'defMessage'.+  --   This holds for OTLP 1.9.0.+  , linkTable :: CommonSymbolTable OP.Link+  -- ^ Common 'OP.Link' table, first entry is the 'defMessage'.+  --   This holds for OTLP 1.9.0.+  , attributeTable :: CommonSymbolTable OP.KeyValueAndUnit+  -- ^ Common 'OP.KeyValueAndUnit' table, first entry is the 'defMessage'.+  --   This holds for OTLP 1.9.0.+  , stackTable :: CommonSymbolTable OP.Stack+  -- ^ Common 'OP.Stack' table, first entry is the 'defMessage'.+  --   This holds for OTLP 1.9.0.+  }+  deriving (Show, Ord, Eq, Generic)++toProfilesDictionary :: ProfileDictionary -> OP.ProfilesDictionary+toProfilesDictionary st =+  messageWith+    [ OP.locationTable .~ locationTableList st+    , OP.functionTable .~ functionTableList st+    , OP.stringTable .~ stringTableList st+    , OP.mappingTable .~ mappingTableList st+    , OP.linkTable .~ linkTableList st+    , OP.attributeTable .~ attributeTableList st+    , OP.stackTable .~ stackTableList st+    ]++emptyProfileDictionary :: ProfileDictionary+emptyProfileDictionary =+  ProfileDictionary+    { locationTable = commonSymbolTableFromList [defMessage]+    , functionTable = commonSymbolTableFromList [defMessage]+    , stringTable = commonSymbolTableFromList [""]+    , mappingTable = commonSymbolTableFromList [defMessage]+    , linkTable = commonSymbolTableFromList [defMessage]+    , attributeTable = commonSymbolTableFromList [defMessage]+    , stackTable = commonSymbolTableFromList [defMessage]+    }++locationTableList :: ProfileDictionary -> [OP.Location]+locationTableList st = reverse st.locationTable.contents++functionTableList :: ProfileDictionary -> [OP.Function]+functionTableList st = reverse st.functionTable.contents++stringTableList :: ProfileDictionary -> [Text]+stringTableList st = reverse st.stringTable.contents++mappingTableList :: ProfileDictionary -> [OP.Mapping]+mappingTableList st = reverse st.mappingTable.contents++linkTableList :: ProfileDictionary -> [OP.Link]+linkTableList st = reverse st.linkTable.contents++attributeTableList :: ProfileDictionary -> [OP.KeyValueAndUnit]+attributeTableList st = reverse st.attributeTable.contents++stackTableList :: ProfileDictionary -> [OP.Stack]+stackTableList st = reverse st.stackTable.contents++getSymbolIndexFor :: (Ord a, Monad m) => Lens' ProfileDictionary (CommonSymbolTable a) -> a -> StateT ProfileDictionary m SymbolIndex+getSymbolIndexFor accessor a = do+  tbl <- State.gets (^. accessor)+  let (idx, tbl1) = insertCommonSymbolTable a tbl+  State.modify' (\st -> st & accessor .~ tbl1)+  pure idx++getLocation :: (Monad m) => OP.Location -> StateT ProfileDictionary m SymbolIndex+getLocation = getSymbolIndexFor (lens (.locationTable) (\s v -> s{locationTable = v}))++getFunction :: (Monad m) => OP.Function -> StateT ProfileDictionary m SymbolIndex+getFunction = getSymbolIndexFor (lens (.functionTable) (\s v -> s{functionTable = v}))++getString :: (Monad m) => Text -> StateT ProfileDictionary m SymbolIndex+getString = getSymbolIndexFor (lens (.stringTable) (\s v -> s{stringTable = v}))++getMapping :: (Monad m) => OP.Mapping -> StateT ProfileDictionary m SymbolIndex+getMapping = getSymbolIndexFor (lens (.mappingTable) (\s v -> s{mappingTable = v}))++getLink :: (Monad m) => OP.Link -> StateT ProfileDictionary m SymbolIndex+getLink = getSymbolIndexFor (lens (.linkTable) (\s v -> s{linkTable = v}))++getAttribute :: (Monad m) => OP.KeyValueAndUnit -> StateT ProfileDictionary m SymbolIndex+getAttribute = getSymbolIndexFor (lens (.attributeTable) (\s v -> s{attributeTable = v}))++getStack :: (Monad m) => OP.Stack -> StateT ProfileDictionary m SymbolIndex+getStack = getSymbolIndexFor (lens (.stackTable) (\s v -> s{stackTable = v}))++-------------------------------------------------------------------------------+-- Common Symbol Table implementation+-- TODO: share with `ghc-stack-profiler-core` table?++data CommonSymbolTable a+  = CommonSymbolTable+  { counter :: !SymbolIndex+  , table :: Map a SymbolIndex+  , contents :: ![a]+  }+  deriving (Show, Ord, Eq, Generic)++emptyCommonSymbolTable :: CommonSymbolTable a+emptyCommonSymbolTable =+  CommonSymbolTable+    { counter = 0+    , table = Map.empty+    , contents = []+    }++commonSymbolTableFromList :: (Ord a) => [a] -> CommonSymbolTable a+commonSymbolTableFromList = foldr go emptyCommonSymbolTable+ where+  go val tbl0 =+    let (_, tbl1) = insertCommonSymbolTable val tbl0+     in tbl1++nextCounter :: CommonSymbolTable a -> (SymbolIndex, CommonSymbolTable a)+nextCounter tbl = (tbl.counter, tbl{counter = tbl.counter + 1})++insertCommonSymbolTable :: (Ord a) => a -> CommonSymbolTable a -> (SymbolIndex, CommonSymbolTable a)+insertCommonSymbolTable val tbl =+  let updateEntry tbl0 Nothing =+        let (sid, newTbl) = nextCounter tbl0+         in ((sid, True, newTbl), Just sid)+      updateEntry newTbl (Just old) =+        ((old, False, newTbl), Just old)++      ((idx, newEntry, tbl1), newTable) = Map.alterF (updateEntry tbl) val tbl.table+   in ( idx+      , tbl1+          { table = newTable+          , contents =+              if newEntry+                then val : tbl1.contents+                else tbl1.contents+          }+      )
+ src/GHC/Eventlog/Live/Otelcol/Processor/Common/Traces.hs view
@@ -0,0 +1,167 @@+{-# LANGUAGE OverloadedStrings #-}++{- |+Module      : GHC.Eventlog.Live.Otelcol.Processor.Common.Traces+Description : Profile Processors for OTLP.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Otelcol.Processor.Common.Traces (+  asSpan,+  ToSpan (..),+  toExportTracesServiceRequest,+  toResourceSpans,+  toScopeSpans,+)+where++import Control.Monad.IO.Class (MonadIO (..))+import Data.ByteString (ByteString)+import Data.Function ((&))+import Data.HashMap.Strict qualified as HM+import Data.Hashable (Hashable)+import Data.Machine (ProcessT, await, construct, yield)+import Data.Maybe (mapMaybe)+import Data.ProtoLens (Message (..))+import GHC.Eventlog.Live.Data.Attribute ((~=))+import GHC.Eventlog.Live.Machine.Analysis.Capability (CapabilityUsageSpan)+import GHC.Eventlog.Live.Machine.Analysis.Capability qualified as M+import GHC.Eventlog.Live.Machine.Analysis.Thread (ThreadStateSpan (..))+import GHC.Eventlog.Live.Machine.Analysis.Thread 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.Processor.Common.Core (ifNonEmpty, messageWith, toMaybeKeyValue)+import GHC.RTS.Events (ThreadId)+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+import Proto.Opentelemetry.Proto.Trace.V1.Trace_Fields qualified as OTS+import System.Random (StdGen, initStdGen)+import System.Random.Compat (uniformByteString)++toExportTracesServiceRequest :: [OT.ResourceSpans] -> OTS.ExportTraceServiceRequest+toExportTracesServiceRequest = (defMessage &) . (OTS.resourceSpans .~)++toResourceSpans :: OR.Resource -> [OT.ScopeSpans] -> Maybe OT.ResourceSpans+toResourceSpans resource scopeSpans =+  ifNonEmpty scopeSpans $+    messageWith [OL.resource .~ resource, OT.scopeSpans .~ scopeSpans]++toScopeSpans :: OC.InstrumentationScope -> [OT.Span] -> Maybe OT.ScopeSpans+toScopeSpans instrumentationScope spans =+  ifNonEmpty spans $+    messageWith [OT.scope .~ instrumentationScope, OT.spans .~ spans]++--------------------------------------------------------------------------------+-- Interpret spans++-- | The `asSpan` machine processes values @v@ into OpenTelemetry spans `OT.Span`.+asSpan :: (ToSpan v, MonadIO m, Hashable (Key v)) => FullConfig -> ProcessT m v OT.Span+asSpan fullConfig = construct $ go (mempty, Nothing)+ where+  -- go :: (HashMap (Key v) ByteString, Maybe StdGen) -> PlanT (Is v) OT.Span m Void+  go (traceIds, maybeGen) = do+    -- Ensure the StdGen is initialised+    gen0 <- maybe (liftIO initStdGen) pure maybeGen+    -- Receive the next value+    i <- await+    -- Ensure the next value has a trace ID+    let ensureTraceId :: Maybe ByteString -> ((ByteString, StdGen), Maybe ByteString)+        ensureTraceId = wrap . maybe (uniformByteString 16 gen0) (,gen0)+         where+          wrap out@(traceId, _gen) = (out, Just traceId)+    let ((traceId, gen1), traceIds') = HM.alterF ensureTraceId (toKey i) traceIds+    -- Ensure the next value has a span ID+    let (spanId, gen2) = uniformByteString 8 gen1+    -- Yield a span+    yield $ toSpan fullConfig i traceId spanId+    -- Continue+    go (traceIds', Just gen2)++class ToSpan v where+  -- | The `Key` type is used to index a `HashMap` in the default definition of `asSpan`.+  type Key v++  -- | The `toKey` function extracts a `Key` from the input value.+  toKey ::+    -- | The input value.+    v ->+    Key v++  toSpan ::+    -- | The configuration.+    FullConfig ->+    -- | The input value.+    v ->+    -- | The trace ID.+    ByteString ->+    -- | The span ID.+    ByteString ->+    OT.Span++--------------------------------------------------------------------------------+-- Interpret capability usage spans++instance ToSpan CapabilityUsageSpan where+  type Key CapabilityUsageSpan = Int++  toKey :: CapabilityUsageSpan -> Int+  toKey = (.cap)++  toSpan :: FullConfig -> CapabilityUsageSpan -> ByteString -> ByteString -> OT.Span+  toSpan fullConfig i traceId spanId =+    messageWith+      [ OT.traceId .~ traceId+      , OT.spanId .~ spanId+      , OT.name .~ C.processorName (.traces) (.capabilityUsage) fullConfig <> " " <> M.showCapabilityUserCategory user+      , OT.kind .~ OT.Span'SPAN_KIND_INTERNAL+      , OT.startTimeUnixNano .~ i.startTimeUnixNano+      , OT.endTimeUnixNano .~ i.endTimeUnixNano+      , OT.attributes+          .~ mapMaybe+            toMaybeKeyValue+            [ "capability" ~= i.cap+            , "user" ~= user+            ]+      , OT.status+          .~ messageWith+            [ OT.code .~ OT.Status'STATUS_CODE_OK+            ]+      ]+   where+    user = M.capabilityUser i++--------------------------------------------------------------------------------+-- Interpret thread state spans++instance ToSpan ThreadStateSpan where+  type Key ThreadStateSpan = ThreadId++  toKey :: ThreadStateSpan -> ThreadId+  toKey = (.thread)++  toSpan :: FullConfig -> ThreadStateSpan -> ByteString -> ByteString -> OT.Span+  toSpan fullConfig i traceId spanId =+    messageWith+      [ OT.traceId .~ traceId+      , OT.spanId .~ spanId+      , OT.name .~ C.processorName (.traces) (.threadState) fullConfig <> " " <> M.showThreadStateCategory i.threadState+      , OT.kind .~ OT.Span'SPAN_KIND_INTERNAL+      , OT.startTimeUnixNano .~ i.startTimeUnixNano+      , OT.endTimeUnixNano .~ i.endTimeUnixNano+      , OT.attributes+          .~ mapMaybe+            toMaybeKeyValue+            [ "capability" ~= M.threadStateCap i.threadState+            , "thread" ~= show i.thread+            , "status" ~= (show <$> M.threadStateStatus i.threadState)+            ]+      , OT.status+          .~ messageWith+            [ OT.code .~ OT.Status'STATUS_CODE_OK+            ]+      ]
+ src/GHC/Eventlog/Live/Otelcol/Processor/Heap.hs view
@@ -0,0 +1,185 @@+{-# LANGUAGE OverloadedStrings #-}++{- |+Module      : GHC.Eventlog.Live.Otelcol.Processor.Heap+Description : Heap Event Processors for OTLP.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Otelcol.Processor.Heap (+  processHeapEvents,+)+where++import Control.Monad.IO.Class (MonadIO (..))+import Data.DList (DList)+import Data.Machine (Process, ProcessT, asParts, echo, mapping, (~>))+import Data.Proxy (Proxy (..))+import GHC.Eventlog.Live.Logger (Logger)+import GHC.Eventlog.Live.Machine.Analysis.Heap (MemReturnData (..))+import GHC.Eventlog.Live.Machine.Analysis.Heap qualified as M+import GHC.Eventlog.Live.Machine.Core (Tick)+import GHC.Eventlog.Live.Machine.Core qualified as M+import GHC.Eventlog.Live.Machine.WithStartTime (WithStartTime (..))+import GHC.Eventlog.Live.Otelcol.Config qualified as C+import GHC.Eventlog.Live.Otelcol.Config.Types (FullConfig (..))+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 Lens.Family2 ((.~))+import Proto.Opentelemetry.Proto.Metrics.V1.Metrics qualified as OM+import Proto.Opentelemetry.Proto.Metrics.V1.Metrics_Fields qualified as OM++--------------------------------------------------------------------------------+-- processHeapEvents+--------------------------------------------------------------------------------++processHeapEvents ::+  (MonadIO m) =>+  Logger m ->+  Maybe HeapProfBreakdown ->+  FullConfig ->+  ProcessT m (Tick (WithStartTime Event)) (Tick (DList OM.Metric))+processHeapEvents verbosity maybeHeapProfBreakdown fullConfig =+  M.fanoutTick+    [ processHeapAllocated fullConfig+    , processBlocksSize fullConfig+    , processHeapSize fullConfig+    , processHeapLive fullConfig+    , processMemReturn fullConfig+    , processHeapProfSample verbosity maybeHeapProfBreakdown fullConfig+    ]++--------------------------------------------------------------------------------+-- HeapAllocated++processHeapAllocated :: FullConfig -> Process (Tick (WithStartTime Event)) (Tick (DList OM.Metric))+processHeapAllocated =+  runMetricProcessor+    MetricProcessor+      { metricProcessorProxy = Proxy @"heapAllocated"+      , dataProcessor = M.processHeapAllocatedData+      , aggregators = viaSum+      , postProcessor = echo+      , unit = "By"+      , asMetric'Data =+          asSum+            [ OM.aggregationTemporality .~ OM.AGGREGATION_TEMPORALITY_DELTA+            , OM.isMonotonic .~ True+            ]+      }++--------------------------------------------------------------------------------+-- HeapSize++processHeapSize :: FullConfig -> Process (Tick (WithStartTime Event)) (Tick (DList OM.Metric))+processHeapSize =+  runMetricProcessor+    MetricProcessor+      { metricProcessorProxy = Proxy @"heapSize"+      , dataProcessor = M.processHeapSizeData+      , aggregators = viaLast+      , postProcessor = echo+      , unit = "By"+      , asMetric'Data = asGauge+      }++--------------------------------------------------------------------------------+-- BlocksSize++processBlocksSize :: FullConfig -> Process (Tick (WithStartTime Event)) (Tick (DList OM.Metric))+processBlocksSize =+  runMetricProcessor+    MetricProcessor+      { metricProcessorProxy = Proxy @"blocksSize"+      , dataProcessor = M.processBlocksSizeData+      , aggregators = viaLast+      , postProcessor = echo+      , unit = "By"+      , asMetric'Data = asGauge+      }++--------------------------------------------------------------------------------+-- HeapLive++processHeapLive :: FullConfig -> Process (Tick (WithStartTime Event)) (Tick (DList OM.Metric))+processHeapLive =+  runMetricProcessor+    MetricProcessor+      { metricProcessorProxy = Proxy @"heapLive"+      , dataProcessor = M.processHeapLiveData+      , aggregators = viaLast+      , postProcessor = echo+      , unit = "By"+      , asMetric'Data = asGauge+      }++--------------------------------------------------------------------------------+-- MemReturn++processMemReturn :: FullConfig -> Process (Tick (WithStartTime Event)) (Tick (DList OM.Metric))+processMemReturn fullConfig =+  runIf (shouldComputeMemReturn fullConfig) $+    M.liftTick M.processMemReturnData+      ~> M.fanoutTick+        [ runMetricProcessor+            MetricProcessor+              { metricProcessorProxy = Proxy @"memCurrent"+              , dataProcessor = mapping (fmap (.current))+              , aggregators = viaLast+              , postProcessor = echo+              , unit = "{mblock}"+              , asMetric'Data = asGauge+              }+            fullConfig+        , runMetricProcessor+            MetricProcessor+              { metricProcessorProxy = Proxy @"memNeeded"+              , dataProcessor = mapping (fmap (.needed))+              , aggregators = viaLast+              , postProcessor = echo+              , unit = "{mblock}"+              , asMetric'Data = asGauge+              }+            fullConfig+        , runMetricProcessor+            MetricProcessor+              { metricProcessorProxy = Proxy @"memReturned"+              , dataProcessor = mapping (fmap (.returned))+              , aggregators = viaLast+              , postProcessor = echo+              , unit = "{mblock}"+              , asMetric'Data = asGauge+              }+            fullConfig+        ]++{- |+Internal helper.+Determine whether the MemReturn data should be computed.+-}+shouldComputeMemReturn :: FullConfig -> Bool+shouldComputeMemReturn fullConfig =+  C.processorEnabled (.metrics) (.memCurrent) fullConfig+    || C.processorEnabled (.metrics) (.memNeeded) fullConfig+    || C.processorEnabled (.metrics) (.memReturned) fullConfig++--------------------------------------------------------------------------------+-- HeapProfSample++processHeapProfSample ::+  (MonadIO m) =>+  Logger m ->+  Maybe HeapProfBreakdown ->+  FullConfig ->+  ProcessT m (Tick (WithStartTime Event)) (Tick (DList OM.Metric))+processHeapProfSample logger maybeHeapProfBreakdown =+  runMetricProcessor+    MetricProcessor+      { metricProcessorProxy = Proxy @"heapProfSample"+      , dataProcessor = M.processHeapProfSampleData logger maybeHeapProfBreakdown+      , aggregators = viaLast+      , postProcessor = mapping M.heapProfSamples ~> asParts+      , unit = "By"+      , asMetric'Data = asGauge+      }
+ src/GHC/Eventlog/Live/Otelcol/Processor/Logs.hs view
@@ -0,0 +1,71 @@+{- |+Module      : GHC.Eventlog.Live.Otelcol.Processor.Logs+Description : Log Processors for OTLP.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Otelcol.Processor.Logs (+  processLogEvents,+)+where++import Control.Monad.IO.Class (MonadIO (..))+import Data.DList (DList)+import Data.DList qualified as D+import Data.Machine (Process, ProcessT, mapping, (~>))+import GHC.Eventlog.Live.Machine.Analysis.Log qualified as M+import GHC.Eventlog.Live.Machine.Analysis.Thread qualified as M+import GHC.Eventlog.Live.Machine.Core (Tick)+import GHC.Eventlog.Live.Machine.Core qualified as M+import GHC.Eventlog.Live.Machine.WithStartTime (WithStartTime (..))+import GHC.Eventlog.Live.Otelcol.Config qualified as C+import GHC.Eventlog.Live.Otelcol.Config.Types (FullConfig (..))+import GHC.Eventlog.Live.Otelcol.Processor.Common.Core (runIf)+import GHC.Eventlog.Live.Otelcol.Processor.Common.Logs (ToLogRecord (..))+import GHC.RTS.Events (Event (..))+import Proto.Opentelemetry.Proto.Logs.V1.Logs qualified as OL++--------------------------------------------------------------------------------+-- processLogEvents+--------------------------------------------------------------------------------++processLogEvents ::+  (MonadIO m) =>+  FullConfig ->+  ProcessT m (Tick (WithStartTime Event)) (Tick (DList OL.LogRecord))+processLogEvents fullConfig =+  M.fanoutTick+    [ processThreadLabel fullConfig+    , processUserMarker fullConfig+    , processUserMessage fullConfig+    ]++--------------------------------------------------------------------------------+-- UserMessage++processUserMessage :: FullConfig -> Process (Tick (WithStartTime Event)) (Tick (DList OL.LogRecord))+processUserMessage fullConfig =+  runIf (C.processorEnabled (.logs) (.userMessage) fullConfig) $+    M.liftTick M.processUserMessageData+      ~> M.liftTick (mapping (D.singleton . toLogRecord))+      ~> M.batchByTicks (C.processorExportBatches (.logs) (.userMessage) fullConfig)++--------------------------------------------------------------------------------+-- UserMarker++processUserMarker :: FullConfig -> Process (Tick (WithStartTime Event)) (Tick (DList OL.LogRecord))+processUserMarker fullConfig =+  runIf (C.processorEnabled (.logs) (.userMarker) fullConfig) $+    M.liftTick M.processUserMarkerData+      ~> M.liftTick (mapping (D.singleton . toLogRecord))+      ~> M.batchByTicks (C.processorExportBatches (.logs) (.userMarker) fullConfig)++--------------------------------------------------------------------------------+-- ThreadLabel++processThreadLabel :: FullConfig -> Process (Tick (WithStartTime Event)) (Tick (DList OL.LogRecord))+processThreadLabel fullConfig =+  runIf (C.processorEnabled (.logs) (.threadLabel) fullConfig) $+    M.liftTick M.processThreadLabelData+      ~> M.liftTick (mapping (D.singleton . toLogRecord))+      ~> M.batchByTicks (C.processorExportBatches (.logs) (.threadLabel) fullConfig)
+ src/GHC/Eventlog/Live/Otelcol/Processor/Profiles.hs view
@@ -0,0 +1,295 @@+{-# LANGUAGE OverloadedStrings #-}++{- |+Module      : GHC.Eventlog.Live.Otelcol.Processor.Profiles+Description : Profile Processors for OTLP.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Otelcol.Processor.Profiles (+  processProfileEvents,+  processCallStackData,+)+where++import Control.Monad.IO.Class (MonadIO (..))+import Control.Monad.Trans.State.Strict (State, runState)+import Data.DList (DList)+import Data.DList qualified as D+import Data.Machine (ProcessT, asParts, mapping, (~>))+import Data.Text (Text)+import GHC.Eventlog.Live.Data.Metric (Metric (..))+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+import GHC.Eventlog.Live.Machine.WithStartTime (WithStartTime (..))+import GHC.Eventlog.Live.Otelcol.Config qualified as C+import GHC.Eventlog.Live.Otelcol.Config.Types (FullConfig (..))+import GHC.Eventlog.Live.Otelcol.Processor.Common.Core+import GHC.Eventlog.Live.Otelcol.Processor.Common.Profiles (ProfileDictionary, SymbolIndex)+import GHC.Eventlog.Live.Otelcol.Processor.Common.Profiles qualified as ProfDictionary+import GHC.RTS.Events (Event (..))+import GHC.Stack.Profiler.Core.SourceLocation qualified as Profiler+import Lens.Family2 ((.~))+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)++--------------------------------------------------------------------------------+-- processProfileEvents+--------------------------------------------------------------------------------++processProfileEvents ::+  (MonadIO m) =>+  Logger m ->+  FullConfig ->+  ProcessT m (Tick (WithStartTime Event)) (Tick (DList M.CallStackData))+processProfileEvents verbosity config =+  M.fanoutTick+    [ processStackProfSample verbosity config+    , processCostCentreProfSample verbosity config+    ]++--------------------------------------------------------------------------------+-- StackProfSample++processStackProfSample ::+  (MonadIO m) =>+  Logger m ->+  FullConfig ->+  ProcessT m (Tick (WithStartTime Event)) (Tick (DList M.CallStackData))+processStackProfSample logger config =+  runIf (C.processorEnabled (.profiles) (.stackSample) config) $+    M.liftTick+      ( M.processStackProfSampleData logger+          ~> mapping M.stackProfSamples+          ~> asParts+          ~> mapping (D.singleton . (.value))+      )+      ~> M.batchByTicks (C.processorExportBatches (.profiles) (.stackSample) config)++processCostCentreProfSample ::+  (MonadIO m) =>+  Logger m ->+  FullConfig ->+  ProcessT m (Tick (WithStartTime Event)) (Tick (DList M.CallStackData))+processCostCentreProfSample logger config =+  runIf (C.processorEnabled (.profiles) (.costCentreSample) config) $+    M.liftTick+      ( M.processCostCentreProfSampleData logger+          ~> mapping M.stackProfSamples+          ~> asParts+          ~> mapping (D.singleton . (.value))+      )+      ~> M.batchByTicks (C.processorExportBatches (.profiles) (.costCentreSample) config)++processCallStackData :: Resource -> OC.InstrumentationScope -> [M.CallStackData] -> (OP.ResourceProfiles, OP.ProfilesDictionary)+processCallStackData resource instrumentationScope callstacks = (resourceProfiles, profilesDictionary)+ where+  scopedProfiles =+    messageWith+      [ OP.profiles .~ [profile]+      , OP.scope .~ instrumentationScope+      ]++  resourceProfiles =+    messageWith+      [ OP.scopeProfiles .~ [scopedProfiles]+      , OP.resource .~ resource+      ]++  profilesDictionary = ProfDictionary.toProfilesDictionary st++  (profile, st) = flip runState ProfDictionary.emptyProfileDictionary $ do+    sampleNameStrId <- ProfDictionary.getString "__name__"+    sampleTypeStrId <- ProfDictionary.getString "String"+    sampleAttrId <-+      ProfDictionary.getAttribute $+        messageWith @OP.KeyValueAndUnit+          [ OP.keyStrindex .~ sampleNameStrId+          , OP.unitStrindex .~ sampleTypeStrId+          , OP.value .~ messageWith [OC.stringValue .~ "process_cpu"]+          ]++    samples <- traverse (asSample sampleAttrId) callstacks+    cpuId <- ProfDictionary.getString "stack"+    unitId <- ProfDictionary.getString "samples"+    let sampleType :: OP.ValueType+        sampleType =+          messageWith+            [ OP.typeStrindex .~ cpuId+            , OP.unitStrindex .~ unitId+            ]++    pure $+      messageWith+        [ OP.samples .~ samples+        , OP.sampleType .~ sampleType+        ]++asSample :: SymbolIndex -> M.CallStackData -> State ProfileDictionary OP.Sample+asSample six stackData = do+  locIndices <- traverse toIndex stackData.stack+  s <-+    ProfDictionary.getStack $+      messageWith+        [ OP.locationIndices .~ locIndices+        ]++  sampleThreadKeyStrId <- ProfDictionary.getString "thread"+  sampleCapKeyStrId <- ProfDictionary.getString "capability"+  sampleNumberUnitStrId <- ProfDictionary.getString "Number"++  threadAttrId <-+    ProfDictionary.getAttribute $+      messageWith @OP.KeyValueAndUnit+        [ OP.keyStrindex .~ sampleThreadKeyStrId+        , OP.unitStrindex .~ sampleNumberUnitStrId+        , OP.value .~ messageWith [OC.intValue .~ maybe 0 (fromIntegral . (.value)) stackData.threadId]+        ]++  capAttrId <-+    ProfDictionary.getAttribute $+      messageWith @OP.KeyValueAndUnit+        [ OP.keyStrindex .~ sampleCapKeyStrId+        , OP.unitStrindex .~ sampleNumberUnitStrId+        , OP.value .~ messageWith [OC.intValue .~ fromIntegral stackData.capabilityId.value]+        ]++  pure $+    messageWith+      [ OP.values .~ [1]+      , OP.stackIndex .~ s+      , OP.attributeIndices+          .~ [ six+             , threadAttrId+             , capAttrId+             ]+      ]+ where+  toIndex :: M.StackItemData -> State ProfileDictionary SymbolIndex+  toIndex = \case+    M.IpeData infoTable -> getLocationIndexForInfoTable infoTable+    M.UserMessageData message -> getLocationIndexForText message+    M.SourceLocationData srcLoc -> getLocationIndexForSourceLocation srcLoc+    M.CostCentreData costCentre -> getLocationIndexForCostCentre costCentre++getLocationIndexForSourceLocation :: Profiler.SourceLocation -> State ProfileDictionary SymbolIndex+getLocationIndexForSourceLocation srcLoc = do+  functionNameId <- ProfDictionary.getString $ Profiler.functionName srcLoc+  fileNameId <- ProfDictionary.getString $ Profiler.fileName srcLoc+  funcIdx <-+    ProfDictionary.getFunction $+      messageWith+        [ OP.nameStrindex .~ functionNameId+        , OP.systemNameStrindex .~ 0 -- 0 means unset+        , OP.filenameStrindex .~ fileNameId+        , OP.startLine .~ fromIntegral (Profiler.line srcLoc) -- TODO: better casts+        ]++  let line :: OP.Line+      line =+        messageWith+          [ OP.functionIndex .~ funcIdx+          , OP.line .~ fromIntegral (Profiler.line srcLoc)+          , OP.column .~ fromIntegral (Profiler.column srcLoc)+          ]++  ProfDictionary.getLocation $+    messageWith+      [ OP.lines .~ [line]+      , OP.mappingIndex .~ 0 -- 0 means unset+      ]++getLocationIndexForText :: Text -> State ProfileDictionary SymbolIndex+getLocationIndexForText msg = do+  textId <- ProfDictionary.getString msg+  funcIdx <-+    ProfDictionary.getFunction $+      messageWith+        [ OP.nameStrindex .~ textId+        , OP.systemNameStrindex .~ 0 -- 0 means unset+        , OP.filenameStrindex .~ 0 -- 0 means unset+        , OP.startLine .~ 0 -- 0 means unset+        ]++  let line :: OP.Line+      line =+        messageWith+          [ OP.functionIndex .~ funcIdx+          , OP.line .~ 0 -- 0 means unset+          , OP.column .~ 0 -- 0 means unset+          ]++  ProfDictionary.getLocation $+    messageWith+      [ OP.lines .~ [line]+      ]++getLocationIndexForInfoTable :: M.InfoTable -> State ProfileDictionary SymbolIndex+getLocationIndexForInfoTable infoTable = do+  infoTableNameId <- ProfDictionary.getString infoTable.infoTableName+  let label =+        if (infoTable.infoTableLabel) == ""+          then infoTable.infoTableModule <> ":" <> infoTable.infoTableName+          else infoTable.infoTableModule <> ":" <> infoTable.infoTableLabel+  infoTableFuncNameId <- ProfDictionary.getString label+  -- tyDesc <- getText infoTable.infoTableTyDesc+  --+  infoTableSrcLocId <- ProfDictionary.getString infoTable.infoTableSrcLoc+  funcIdx <-+    ProfDictionary.getFunction $+      messageWith+        [ OP.nameStrindex .~ infoTableFuncNameId+        , OP.systemNameStrindex .~ infoTableNameId+        , OP.filenameStrindex .~ infoTableSrcLocId -- 0 means unset+        , OP.startLine .~ 0 -- 0 means unset+        ]++  let line :: OP.Line+      line =+        messageWith+          [ OP.functionIndex .~ funcIdx+          , OP.line .~ 0 -- 0 means unset+          , OP.column .~ 0 -- 0 means unset+          ]++  ProfDictionary.getLocation $+    messageWith+      [ OP.lines .~ [line]+      , OP.address .~ 0 -- 0 means unset+      ]++getLocationIndexForCostCentre :: M.CostCentre -> State ProfileDictionary SymbolIndex+getLocationIndexForCostCentre costCentre = do+  let label = costCentre.costCentreModule <> ":" <> costCentre.costCentreLabel+  costCentreFuncNameId <- ProfDictionary.getString label+  -- tyDesc <- getText infoTable.infoTableTyDesc+  --+  costCentreSrcLocId <- ProfDictionary.getString costCentre.costCentreSrcLoc+  funcIdx <-+    ProfDictionary.getFunction $+      messageWith+        [ OP.nameStrindex .~ costCentreFuncNameId+        , OP.systemNameStrindex .~ costCentreFuncNameId+        , OP.filenameStrindex .~ costCentreSrcLocId -- 0 means unset+        , OP.startLine .~ 0 -- 0 means unset+        ]++  let line :: OP.Line+      line =+        messageWith+          [ OP.functionIndex .~ funcIdx+          , OP.line .~ 0 -- 0 means unset+          , OP.column .~ 0 -- 0 means unset+          ]++  ProfDictionary.getLocation $+    messageWith+      [ OP.lines .~ [line]+      , OP.address .~ 0 -- 0 means unset+      ]
+ src/GHC/Eventlog/Live/Otelcol/Processor/Threads.hs view
@@ -0,0 +1,151 @@+{-# LANGUAGE OverloadedStrings #-}++{- |+Module      : GHC.Eventlog.Live.Otelcol.Processor.Threads+Description : Thread Event Processors for OTLP.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Otelcol.Processor.Threads (+  processThreadEvents,+)+where++import Control.Monad.IO.Class (MonadIO (..))+import Data.DList (DList)+import Data.DList qualified as D+import Data.Machine (ProcessT, asParts, echo, mapping, (~>))+import Data.Machine.Fanout (fanout)+import Data.Proxy (Proxy (..))+import GHC.Eventlog.Live.Logger (Logger)+import GHC.Eventlog.Live.Machine.Analysis.Capability qualified as M+import GHC.Eventlog.Live.Machine.Analysis.Thread qualified as M+import GHC.Eventlog.Live.Machine.Core (Tick)+import GHC.Eventlog.Live.Machine.Core qualified as M+import GHC.Eventlog.Live.Machine.WithStartTime (WithStartTime (..))+import GHC.Eventlog.Live.Machine.WithStartTime qualified as M+import GHC.Eventlog.Live.Otelcol.Config qualified as C+import GHC.Eventlog.Live.Otelcol.Config.Types (FullConfig (..))+import GHC.Eventlog.Live.Otelcol.Processor.Common.Core (runIf)+import GHC.Eventlog.Live.Otelcol.Processor.Common.Metrics (MetricProcessor (..), asSum, runMetricProcessor, viaSum)+import GHC.Eventlog.Live.Otelcol.Processor.Common.Traces (asSpan)+import GHC.RTS.Events (Event (..))+import Lens.Family2 ((.~))+import Proto.Opentelemetry.Proto.Metrics.V1.Metrics qualified as OM+import Proto.Opentelemetry.Proto.Metrics.V1.Metrics_Fields qualified as OM+import Proto.Opentelemetry.Proto.Trace.V1.Trace qualified as OT++data OneOf a b c = A !a | B !b | C !c++processThreadEvents ::+  (MonadIO m) =>+  Logger m ->+  FullConfig ->+  ProcessT m (Tick (WithStartTime Event)) (Tick (DList (Either OM.Metric OT.Span)))+processThreadEvents verbosity fullConfig =+  runIf (shouldProcessThreadEvents fullConfig) $+    M.sortByTicks (.value.evTime) fullConfig.eventlogFlushIntervalX+      ~> M.liftTick+        ( fanout+            [ M.validateOrder verbosity (.value.evTime)+            , runIf (shouldComputeCapabilityUsageSpan fullConfig) $+                M.processGCSpans verbosity+                  ~> mapping (D.singleton . A)+            , runIf (shouldComputeThreadStateSpan fullConfig) $+                M.processThreadStateSpans' M.tryGetTimeUnixNano (.value) M.setWithStartTime'value verbosity+                  ~> fanout+                    [ M.asMutatorSpans' (.value) M.setWithStartTime'value+                        ~> mapping (D.singleton . B)+                    , mapping (D.singleton . C)+                    ]+            ]+        )+      ~> M.liftTick+        ( asParts+            ~> mapping repackCapabilityUsageSpanOrThreadStateSpan+        )+      ~> fanout+        [ M.liftTick+            ( mapping leftToMaybe+                ~> asParts+            )+            ~> M.fanoutTick+              [ runMetricProcessor+                  MetricProcessor+                    { metricProcessorProxy = Proxy @"capabilityUsage"+                    , dataProcessor = M.processCapabilityUsageMetrics+                    , aggregators = viaSum+                    , postProcessor = echo+                    , unit = "ns"+                    , asMetric'Data =+                        asSum+                          [ OM.aggregationTemporality .~ OM.AGGREGATION_TEMPORALITY_DELTA+                          , OM.isMonotonic .~ True+                          ]+                    }+                  fullConfig+                  ~> mapping (fmap (fmap Left))+              , runIf (C.processorEnabled (.traces) (.capabilityUsage) fullConfig) $+                  M.liftTick+                    ( M.dropStartTime+                        ~> asSpan fullConfig+                        ~> mapping (D.singleton . Right)+                    )+                    ~> M.batchByTick+              ]+        , runIf (C.processorEnabled (.traces) (.threadState) fullConfig) $+            M.liftTick+              ( mapping rightToMaybe+                  ~> asParts+                  ~> asSpan fullConfig+                  ~> mapping (D.singleton . Right)+              )+              ~> M.batchByTick+        ]+ where+  repackCapabilityUsageSpanOrThreadStateSpan = \case+    A i -> Left $ fmap Left i+    B i -> Left $ fmap Right i+    C i -> Right i.value++{- |+Internal helper.+Get the `Left` value, if any.+-}+leftToMaybe :: Either a b -> Maybe a+leftToMaybe = either Just (const Nothing)++{- |+Internal helper.+Get the `Right` value, if any.+-}+rightToMaybe :: Either a b -> Maybe b+rightToMaybe = either (const Nothing) Just++{- |+Internal helper.+Determine whether or not any thread events should be processed at all.+-}+shouldProcessThreadEvents :: FullConfig -> Bool+shouldProcessThreadEvents fullConfig =+  C.processorEnabled (.metrics) (.capabilityUsage) fullConfig+    || C.processorEnabled (.traces) (.capabilityUsage) fullConfig+    || C.processorEnabled (.traces) (.threadState) fullConfig++{- |+Internal helper.+Determine whether or not the capability usage spans should be computed.+-}+shouldComputeCapabilityUsageSpan :: FullConfig -> Bool+shouldComputeCapabilityUsageSpan fullConfig =+  C.processorEnabled (.traces) (.capabilityUsage) fullConfig+    || C.processorEnabled (.metrics) (.capabilityUsage) fullConfig++{- |+Internal helper.+Determine whether or not the thread state spans should be computed.+-}+shouldComputeThreadStateSpan :: FullConfig -> Bool+shouldComputeThreadStateSpan fullConfig =+  C.processorEnabled (.traces) (.threadState) fullConfig+    || shouldComputeCapabilityUsageSpan fullConfig
src/GHC/Eventlog/Live/Otelcol/Stats.hs view
@@ -11,7 +11,8 @@   eventCountTick,   Stat (..),   processStats,-) where+)+where  import Control.Exception (Exception (..)) import Control.Monad (when)@@ -19,42 +20,56 @@ import Data.Default (Default (..)) import Data.Foldable (for_) import Data.Int (Int64)-import Data.Machine (ProcessT, await, construct, mapping, repeatedly, (~>))+import Data.Machine (ProcessT, await, construct, repeatedly, yield) import Data.Monoid (First (..)) import Data.Text (Text) import Data.Text qualified as T import Data.Text.IO qualified as TIO import Data.Void (Void)-import GHC.Eventlog.Live.Logger (logDebug, logError)+import GHC.Eventlog.Live.Data.Severity (Severity (..))+import GHC.Eventlog.Live.Logger (Logger, writeLog) import GHC.Eventlog.Live.Machine.Core (Tick) import GHC.Eventlog.Live.Machine.Core qualified as M-import GHC.Eventlog.Live.Otelcol.Exporter (ExportMetricsResult (..), ExportTraceResult (..))-import GHC.Eventlog.Live.Verbosity (Verbosity)+import GHC.Eventlog.Live.Otelcol.Exporter.Logs (ExportLogsResult (..))+import GHC.Eventlog.Live.Otelcol.Exporter.Metrics (ExportMetricsResult (..))+import GHC.Eventlog.Live.Otelcol.Exporter.Profiles (ExportProfileResult (..))+import GHC.Eventlog.Live.Otelcol.Exporter.Traces (ExportTraceResult (..)) import GHC.Records (HasField (..)) import StrictList qualified as Strict import System.Console.ANSI (hNowSupportsANSI) import System.Console.ANSI qualified as ANSI import System.IO qualified as IO import Text.Layout.Table qualified as TBL+import Text.Printf (printf)  {- | This type represents a count of input events. -}-newtype EventCount = EventCount {value :: Int64}+newtype EventCount+  = EventCount {value :: Int64}+  deriving (Show)  {- | Count the number of events seen between each tick. -}-eventCountTick :: (Monad m) => ProcessT m (Tick a) EventCount-eventCountTick = M.batchByTickList ~> mapping (EventCount . fromIntegral . length)+eventCountTick :: (Monad m) => ProcessT m (Tick a) (Tick EventCount)+eventCountTick = construct $ go 0+ where+  go acc =+    await >>= \case+      M.Tick -> yield (M.Item $ EventCount acc) >> yield M.Tick >> go 0+      M.Item{} -> go (acc + 1)  {- | This type represents the various stats produced by the pipeline. -} data Stat   = EventCountStat !EventCount+  | ExportLogsResultStat !ExportLogsResult   | ExportMetricsResultStat !ExportMetricsResult   | ExportTraceResultStat !ExportTraceResult+  | ExportProfileResultStat !ExportProfileResult+  deriving (Show)  {- | Internal helper.@@ -62,10 +77,14 @@ -} data Stats = Stats   { eventCounts :: Row+  , exportedLogRecords :: Row+  , rejectedLogRecords :: Row   , exportedDataPoints :: Row   , rejectedDataPoints :: Row   , exportedSpans :: Row   , rejectedSpans :: Row+  , exportedProfiles :: Row+  , rejectedProfiles :: Row   , errors :: !(Strict.List Text)   , displayedLines :: !(First Int)   }@@ -77,23 +96,23 @@ -} data Row = Row   { total :: !Int64-  , peakRatePerBatch :: !Int64+  , peakRatePerBatch :: !Double   , window :: !(Strict.List Int64)   }   deriving (Show) -instance HasField "ratePerBatch" Row Int64 where-  getField :: Row -> Int64+instance HasField "ratePerBatch" Row Double where+  getField :: Row -> Double   getField row = ratePerBatch row.window  {- | Internal helper. Computes the rate per batch from a window. -}-ratePerBatch :: Strict.List Int64 -> Int64+ratePerBatch :: Strict.List Int64 -> Double ratePerBatch window =   let !n = length window-   in if n <= 0 then 0 else sum window `div` fromIntegral n+   in if n <= 0 then 0 else sum (realToFrac <$> window) / fromIntegral n  {- | Internal helper.@@ -102,7 +121,7 @@ singletonRow :: Int64 -> Row singletonRow total = Row{..}  where-  peakRatePerBatch = total+  peakRatePerBatch = realToFrac total   window = singletonStrictList total  {- |@@ -126,10 +145,14 @@ unionStats windowSize new old = Stats{..}  where   eventCounts = unionRow windowSize new.eventCounts old.eventCounts+  exportedLogRecords = unionRow windowSize new.exportedLogRecords old.exportedLogRecords+  rejectedLogRecords = unionRow windowSize new.rejectedLogRecords old.rejectedLogRecords   exportedDataPoints = unionRow windowSize new.exportedDataPoints old.exportedDataPoints   rejectedDataPoints = unionRow windowSize new.rejectedDataPoints old.rejectedDataPoints   exportedSpans = unionRow windowSize new.exportedSpans old.exportedSpans   rejectedSpans = unionRow windowSize new.rejectedSpans old.rejectedSpans+  exportedProfiles = unionRow windowSize new.exportedProfiles old.exportedProfiles+  rejectedProfiles = unionRow windowSize new.rejectedProfiles old.rejectedProfiles   errors = Strict.take windowSize (new.errors <> old.errors)   displayedLines = new.displayedLines <> old.displayedLines @@ -142,6 +165,18 @@  {- | Internal helper.+Construct a `Stats` object from an `ExportLogsResult`.+-}+fromExportLogsResult :: ExportLogsResult -> Stats+fromExportLogsResult exportLogsResult =+  def+    { exportedLogRecords = singletonRow exportLogsResult.exportedLogRecords+    , rejectedLogRecords = singletonRow exportLogsResult.rejectedLogRecords+    , errors = maybeToStrictList $ T.pack . displayException <$> exportLogsResult.maybeSomeException+    }++{- |+Internal helper. Construct a `Stats` object from an `ExportMetricsResult`. -} fromExportMetricsResult :: ExportMetricsResult -> Stats@@ -157,15 +192,27 @@ Construct a `Stats` object from an `ExportTraceResult`. -} fromExportTraceResult :: ExportTraceResult -> Stats-fromExportTraceResult exportTraceResult =+fromExportTraceResult exportTracesResult =   def-    { exportedSpans = singletonRow exportTraceResult.exportedSpans-    , rejectedSpans = singletonRow exportTraceResult.rejectedSpans-    , errors = maybeToStrictList $ T.pack . displayException <$> exportTraceResult.maybeSomeException+    { exportedSpans = singletonRow exportTracesResult.exportedSpans+    , rejectedSpans = singletonRow exportTracesResult.rejectedSpans+    , errors = maybeToStrictList $ T.pack . displayException <$> exportTracesResult.maybeSomeException     }  {- | Internal helper.+Construct a `Stats` object from an `ExportProfileResult`.+-}+fromExportProfileResult :: ExportProfileResult -> Stats+fromExportProfileResult exportProfileResult =+  def+    { exportedProfiles = singletonRow exportProfileResult.exportedProfiles+    , rejectedProfiles = singletonRow exportProfileResult.rejectedProfiles+    , errors = maybeToStrictList $ T.pack . displayException <$> exportProfileResult.maybeSomeException+    }++{- |+Internal helper. Construct a singleton `Strict.List`. -} singletonStrictList :: a -> Strict.List a@@ -199,10 +246,14 @@   def = Stats{..}    where     eventCounts = def+    exportedLogRecords = def+    rejectedLogRecords = def     exportedDataPoints = def     rejectedDataPoints = def     exportedSpans = def     rejectedSpans = def+    exportedProfiles = def+    rejectedProfiles = def     errors = mempty     displayedLines = First Nothing @@ -212,24 +263,28 @@ __Warning:__ This machine prints to stdout and is intended to be the /only/ function printing to stdout. -} processStats ::-  (MonadIO m) =>-  Verbosity ->+  Logger IO ->   Bool ->-  Int ->+  Double ->   Int ->-  ProcessT m Stat Void-processStats verbosity stats batchIntervalMs windowSize+  ProcessT IO Stat Void+processStats logger stats eventlogFlushIntervalS windowSize   | stats =       -- If --stats is ENABLED, maintain and display `Stats`.       let go stats0 =             await >>= \stat -> do+              -- Log the incoming `Stat` value.+              liftIO $ logStat logger stat+              -- Maintain and display the `Stats`.               let stats1 = updateStats windowSize stats0 stat-              stats2 <- liftIO (displayStats verbosity batchIntervalMs stats1)+              stats2 <- liftIO $ displayStats logger eventlogFlushIntervalS stats1               go stats2        in construct $ go def   | otherwise =       -- If --stats is DISABLED, log all incoming `Stat` values.-      repeatedly $ await >>= logStat verbosity+      repeatedly $+        await >>= \stat ->+          liftIO $ logStat logger stat  {- | Internal helper.@@ -238,37 +293,79 @@ updateStats :: Int -> Stats -> Stat -> Stats updateStats windowSize old = \case   EventCountStat eventCount -> unionStats windowSize (fromEventCount eventCount) old+  ExportLogsResultStat exportLogsResults -> unionStats windowSize (fromExportLogsResult exportLogsResults) old   ExportMetricsResultStat exportMetricsResult -> unionStats windowSize (fromExportMetricsResult exportMetricsResult) old-  ExportTraceResultStat exportTraceResult -> unionStats windowSize (fromExportTraceResult exportTraceResult) old+  ExportTraceResultStat exportTracesResult -> unionStats windowSize (fromExportTraceResult exportTracesResult) old+  ExportProfileResultStat exportProfilesResult -> unionStats windowSize (fromExportProfileResult exportProfilesResult) old  {- | Internal helper. Log a statistic. -}-logStat :: (MonadIO m) => Verbosity -> Stat -> m ()-logStat verbosity = \case+logStat ::+  Logger IO ->+  Stat ->+  IO ()+logStat logger = \case   EventCountStat eventCount ->-    logDebug verbosity $ "received " <> showText eventCount.value <> " events"+    -- Log received events.+    when (eventCount.value > 0) $ do+      writeLog logger DEBUG $+        "Received " <> showText eventCount.value <> " events."+  ExportLogsResultStat exportLogsResult -> do+    -- Log exported events.+    when (exportLogsResult.exportedLogRecords > 0) $ do+      writeLog logger DEBUG $+        "Exported " <> showText exportLogsResult.exportedLogRecords <> " logs."+    -- Log rejected events.+    when (exportLogsResult.rejectedLogRecords > 0) $ do+      writeLog logger ERROR $+        "Rejected " <> showText exportLogsResult.rejectedLogRecords <> " logs."+    -- Log exception.+    for_ exportLogsResult.maybeSomeException $ \someException -> do+      writeLog logger ERROR $+        T.pack $+          displayException someException   ExportMetricsResultStat exportMetricsResult -> do     -- Log exported events.-    logDebug verbosity $ "exported " <> showText exportMetricsResult.exportedDataPoints <> " metrics"+    when (exportMetricsResult.exportedDataPoints > 0) $ do+      writeLog logger DEBUG $+        "Exported " <> showText exportMetricsResult.exportedDataPoints <> " metrics."     -- Log rejected events.-    when (exportMetricsResult.rejectedDataPoints > 0) $-      logError verbosity $-        "rejected " <> showText exportMetricsResult.rejectedDataPoints <> " metrics"+    when (exportMetricsResult.rejectedDataPoints > 0) $ do+      writeLog logger ERROR $+        "Rejected " <> showText exportMetricsResult.rejectedDataPoints <> " metrics."     -- Log exception.     for_ exportMetricsResult.maybeSomeException $ \someException -> do-      logError verbosity . T.pack $ displayException someException-  ExportTraceResultStat exportTraceResult -> do+      writeLog logger ERROR $+        T.pack $+          displayException someException+  ExportTraceResultStat exportTracesResult -> do     -- Log exported events.-    logDebug verbosity $ "exported " <> showText exportTraceResult.exportedSpans <> " metrics"+    when (exportTracesResult.exportedSpans > 0) $ do+      writeLog logger DEBUG $+        "Exported " <> showText exportTracesResult.exportedSpans <> " spans."     -- Log rejected events.-    when (exportTraceResult.rejectedSpans > 0) $-      logError verbosity $-        "rejected " <> showText exportTraceResult.rejectedSpans <> " metrics"+    when (exportTracesResult.rejectedSpans > 0) $ do+      writeLog logger ERROR $+        "Rejected " <> showText exportTracesResult.rejectedSpans <> " spans."     -- Log exception.-    for_ exportTraceResult.maybeSomeException $ \someException -> do-      logError verbosity . T.pack $ displayException someException+    for_ exportTracesResult.maybeSomeException $ \someException -> do+      writeLog logger ERROR $+        T.pack $+          displayException someException+  ExportProfileResultStat exportProfilesResult -> do+    -- Log exported events.+    when (exportProfilesResult.exportedProfiles > 0) $+      writeLog logger DEBUG $+        "Exported " <> showText exportProfilesResult.exportedProfiles <> " profiles."+    -- Log rejected events.+    when (exportProfilesResult.rejectedProfiles > 0) $+      writeLog logger ERROR $+        "Rejected " <> showText exportProfilesResult.rejectedProfiles <> " profiles."+    -- Log exception.+    for_ exportProfilesResult.maybeSomeException $ \someException -> do+      writeLog logger ERROR . T.pack $ displayException someException  {- | Internal helper.@@ -277,59 +374,65 @@  TODO: The stats printer should only overwrite the numbers. -}-displayStats :: Verbosity -> Int -> Stats -> IO Stats-displayStats verbosity intervalMs stats = do+displayStats ::+  Logger IO ->+  Double ->+  Stats ->+  IO Stats+displayStats logger eventlogFlushIntervalS stats = do   -- Check if `displayedLines` is empty...   case stats.displayedLines of     First Nothing ->       -- ...if so, this is the first time this function has been evaluated...       -- ...so we should perform the `warnIfStderrSupportsANSI` check...-      warnIfStderrSupportsANSI verbosity+      warnIfStderrSupportsANSI logger     First (Just numberOfLines) -> do       -- ...if not, we should clear the previous lines of output...-      ANSI.cursorUp numberOfLines-      ANSI.clearFromCursorToScreenEnd+      liftIO $ ANSI.cursorUp numberOfLines+      liftIO $ ANSI.clearFromCursorToScreenEnd    -- Compute the moving average count of items _per second_,   -- by computing the adjusted average of counts over n batches.   let rate :: Row -> Text-      rate row =-        let !r = (row.ratePerBatch * 1_000) `div` fromIntegral intervalMs-         in showText r <> "/s"+      rate row = T.pack . printf "%0.2f" $ row.ratePerBatch / eventlogFlushIntervalS   let peak :: Row -> Text-      peak row =-        let !r = row.peakRatePerBatch-         in showText r <> "/s"+      peak row = T.pack . printf "%0.0f" $ row.peakRatePerBatch    let cSpec :: [TBL.ColSpec]       cSpec = [TBL.defColSpec, TBL.defColSpec, TBL.numCol, TBL.numCol, TBL.numCol]   let hSpec :: TBL.HeaderSpec TBL.LineStyle (Maybe Text)-      hSpec = TBL.titlesH [Just "Item", Just "Action", Just "Total (item)", Just "Rate (item/s)", Just "Peak (item/batch)"]+      hSpec = TBL.titlesH [Just "Item", Just "Action", Just "Total (item)", Just "Rate (item/s)", Just "Peak (item/x)"]   let mkRow :: Maybe Text -> Maybe Text -> Row -> TBL.RowGroup (Maybe Text)       mkRow item result row = TBL.rowG [item, result, Just (showText row.total), Just (rate row), Just (peak row)]   let rSpec :: [TBL.RowGroup (Maybe Text)]       rSpec =-        [ mkRow (Just "Event") (Just "Received") stats.eventCounts-        , mkRow (Just "Metric") (Just "Exported") stats.exportedDataPoints+        [ mkRow (Just "Events") (Just "Received") stats.eventCounts+        , mkRow (Just "Logs") (Just "Exported") stats.exportedLogRecords+        , mkRow Nothing (Just "Rejected") stats.rejectedLogRecords+        , mkRow (Just "Metrics") (Just "Exported") stats.exportedDataPoints         , mkRow Nothing (Just "Rejected") stats.rejectedDataPoints-        , mkRow (Just "Span") (Just "Exported") stats.exportedSpans+        , mkRow (Just "Traces") (Just "Exported") stats.exportedSpans         , mkRow Nothing (Just "Rejected") stats.rejectedSpans+        , mkRow (Just "Profiles") (Just "Exported") stats.exportedProfiles+        , mkRow Nothing (Just "Rejected") stats.rejectedProfiles         ]   let tSpec :: TBL.TableSpec TBL.LineStyle TBL.LineStyle String (Maybe Text) (Maybe Text)       tSpec = TBL.columnHeaderTableS cSpec TBL.unicodeS hSpec rSpec   let table = TBL.tableLinesB tSpec :: [Text]-  for_ table TIO.putStrLn+  for_ table $ \row -> liftIO $ TIO.putStrLn row   pure stats{displayedLines = First (Just $ length table)}  {- | Check if `IO.stderr` supports ANSI codes. If it does, it is likely printed to the same terminal as `IO.stdout`, which causes issues if @--stats@ is enabled. -}-warnIfStderrSupportsANSI :: Verbosity -> IO ()-warnIfStderrSupportsANSI verbosity = do+warnIfStderrSupportsANSI ::+  Logger IO ->+  IO ()+warnIfStderrSupportsANSI logger = do   supportsANSI <- hNowSupportsANSI IO.stderr   when supportsANSI $ do-    logError verbosity $+    writeLog logger WARN $       "When statistics are enabled, stderr should be redirected to a file."  -------------------------------------------------------------------------------
+ src/GHC/Eventlog/Socket/Compat.hs view
@@ -0,0 +1,82 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE OverloadedStrings #-}++{- |+Module      : GHC.Eventlog.Socket.Compat+Description : The implementation of @eventlog-live-otelcol@.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Socket.Compat (+  MyEventlogSocket (..),+  maybeMyEventlogSocketParser,+  startMyEventlogSocket,+) where++import Control.Applicative (asum)+import GHC.Eventlog.Live.Logger (Logger)+import Options.Applicative qualified as O+import Options.Applicative.Extra.Feature (Feature (..))+import Options.Applicative.Extra.Feature qualified as OF++#ifdef EVENTLOG_LIVE_OTELCOL_USE_EVENTLOG_SOCKET+import Data.Foldable (for_)+import Data.Text qualified as T+import GHC.Eventlog.Live.Data.Severity (Severity (..))+import GHC.Eventlog.Live.Logger (writeLog)+import GHC.Eventlog.Socket qualified as Eventlog.Socket+#else+import Control.Monad (when)+import Data.Maybe (isJust)+#endif++--------------------------------------------------------------------------------+-- Feature: use-eventlog-socket+--------------------------------------------------------------------------------++useEventlogSocket :: Feature+useEventlogSocket = Feature{flag = "use-eventlog-socket", isOn = isOn, info = "Cannot open eventlog socket."}+ where+  isOn :: Bool+#ifdef EVENTLOG_LIVE_OTELCOL_USE_EVENTLOG_SOCKET+  isOn = True+#else+  isOn = False+#endif++--------------------------------------------------------------------------------+-- My Eventlog Socket+--------------------------------------------------------------------------------++newtype MyEventlogSocket+  = MyEventlogSocketUnix FilePath++maybeMyEventlogSocketParser :: O.Parser (Maybe MyEventlogSocket)+maybeMyEventlogSocketParser =+  asum $+    [ myEventlogSocketUnixParser+    , pure Nothing+    ]++myEventlogSocketUnixParser :: O.Parser (Maybe MyEventlogSocket)+myEventlogSocketUnixParser =+  OF.onlyFor useEventlogSocket (O.option (Just . MyEventlogSocketUnix <$> O.str)) (O.metavar "FILE") $+    O.long "my-eventlog-socket-unix"+      <> OF.helpFor useEventlogSocket "Open an eventlog socket for this program on the given Unix socket."++{- |+Set @eventlog-socket@ as the eventlog writer.+-}+startMyEventlogSocket :: Logger IO -> Maybe MyEventlogSocket -> IO ()+#ifdef EVENTLOG_LIVE_OTELCOL_USE_EVENTLOG_SOCKET+startMyEventlogSocket logger maybeMyEventlogSocket =+  for_ maybeMyEventlogSocket $ \case+    MyEventlogSocketUnix myEventlogSocketUnix -> do+      writeLog logger INFO $+        "Start eventlog-socket with Unix domain socket at " <> T.pack myEventlogSocketUnix <> "."+      Eventlog.Socket.startWait myEventlogSocketUnix+#else+startMyEventlogSocket logger maybeMyEventlogSocket =+  when (isJust maybeMyEventlogSocket) $+    OF.exitIfUnsupported useEventlogSocket logger+#endif
+ src/Options/Applicative/Extra/Feature.hs view
@@ -0,0 +1,62 @@+{-# LANGUAGE OverloadedStrings #-}++module Options.Applicative.Extra.Feature (+  Feature (..),+  helpFor,+  onlyFor,+  exitIfUnsupported,+) where++import Control.Monad (unless)+import Data.Default (Default (..))+import Data.Text (Text)+import Data.Text qualified as T+import GHC.Eventlog.Live.Data.Severity (Severity (..))+import GHC.Eventlog.Live.Logger (Logger, writeLog)+import Options.Applicative qualified as O+import Options.Applicative.Help.Pretty qualified as OP+import System.Exit (exitFailure)++data Feature = Feature+  { flag :: !String+  , isOn :: !Bool+  , info :: !String+  }++{- |+Create a command-line help document for an option that depend on a feature flag.+-}+helpFor :: Feature -> String -> O.Mod f a+helpFor feature help+  | feature.isOn = O.help help+  | otherwise = O.helpDoc (Just $ OP.vcat [OP.pretty unsupported, OP.pretty help]) <> O.hidden+ where+  unsupported :: Text+  unsupported = T.pack "Unsupported. Requires build with -f+" <> T.pack feature.flag <> "."++{- |+Create a command-line option that depends on a feature flag.+-}+onlyFor ::+  forall f a.+  (O.HasName f, Default a) =>+  Feature ->+  (O.Mod f a -> O.Parser a) ->+  O.Mod f a ->+  (forall g x. (O.HasName g) => O.Mod g x) ->+  O.Parser a+onlyFor feature opt optIfSupported optAlways+  | feature.isOn = opt (optAlways <> optIfSupported)+  | otherwise = def <$ O.infoOption unsupportedInfo optAlways+ where+  unsupportedInfo :: String+  unsupportedInfo = feature.info <> " Requires build with -f+" <> feature.flag <> "."++{- |+Exit with the feature info.+-}+exitIfUnsupported :: Feature -> Logger IO -> IO ()+exitIfUnsupported feature logger =+  unless feature.isOn $ do+    writeLog logger FATAL (T.pack feature.info)+    exitFailure