diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -2,6 +2,40 @@ ## Unreleased +## 1.0.0.0 - 2026-05-29 + +- **Spec: `Retry-After` now supports HTTP-date format in addition to delay-seconds.** + Previous implementation only parsed integer seconds. Now handles both RFC 7231 + Section 7.1.3 formats. Applied to all three signal exporters (Span, Metric, LogRecord). + Spec: +- **`LogRecordExporter.forceFlush` now returns `FlushSuccess` instead of `()`.** + Aligns with the updated `LogRecordExporter` type in `hs-opentelemetry-api`. +- **gRPC transport wired for metrics and logs.** + When the `grpc` Cabal flag is enabled and `OTEL_EXPORTER_OTLP_PROTOCOL=grpc`, + all three signals (traces, metrics, logs) now use gRPC transport. + Previously only traces supported gRPC. +- **Concurrent export configuration.** + New `OTEL_EXPORTER_OTLP_CONCURRENT_EXPORTS` env var and + `otlpConcurrentExports` config field (default 1). +- **Fix: `droppedAttributesCount` was reporting stored count, not dropped count.** + All three OTLP exporters (Span, Metric, LogRecord) used `getCount` instead of + `getDropped` for span attributes, event attributes, link attributes, resource + attributes, and scope attributes. Now uses `getDropped` everywhere. +- **Fix: `Accept-Encoding` header set to protobuf MIME type.** Changed to `Accept` + header, which is the correct HTTP header for content-type negotiation. +- **Fix: `Unknown n` severity number could crash with `toEnum` on out-of-range values.** + Now falls back to `SEVERITY_NUMBER_UNSPECIFIED` for values outside 0–24. +- **Fix: `Span.flags` and `Link.flags` proto fields never set.** The sampled + bit in W3C trace flags was always 0 in exported OTLP spans and links, even + when the span was sampled. Now populates `flags` from `traceFlagsValue` on + both `Span` and `Span.Link` messages (matching the existing log exporter). +- OTLP span exporter: set `schema_url` on `ResourceSpans` and `ScopeSpans`, set scope `attributes` and `droppedAttributesCount` +- `OpenTelemetry.Exporter.OTLP` barrel module now re-exports all three signals (Span, Metric, LogRecord) +- Implement `OpenTelemetry.Exporter.OTLP.LogRecord`: full OTLP HTTP/Protobuf log exporter with retry, compression, and severity/AnyValue/tracing-context serialization +- Use `startTimeUnixNano` from data points instead of hardcoded 0 +- Comprehensive protobuf round-trip tests for all metric types +- OTLP metrics: exemplars on number and histogram data points; exponential histogram messages; aggregation temporality from export model. + ## 0.1.1.0 - Complete `loadExporterEnvironmentVariables` implementation diff --git a/LICENSE b/LICENSE --- a/LICENSE +++ b/LICENSE @@ -1,4 +1,4 @@ -Copyright Ian Duncan (c) 2021 +Copyright Ian Duncan (c) 2021-2026 All rights reserved. diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -1,1 +1,3 @@ # hs-opentelemetry-exporter-otlp + +[![hs-opentelemetry-exporter-otlp](https://img.shields.io/hackage/v/hs-opentelemetry-exporter-otlp?style=flat-square&logo=haskell&label=hs-opentelemetry-exporter-otlp&labelColor=5D4F85)](https://hackage.haskell.org/package/hs-opentelemetry-exporter-otlp) diff --git a/hs-opentelemetry-exporter-otlp.cabal b/hs-opentelemetry-exporter-otlp.cabal --- a/hs-opentelemetry-exporter-otlp.cabal +++ b/hs-opentelemetry-exporter-otlp.cabal @@ -1,11 +1,11 @@ cabal-version: 1.12 --- This file has been generated from package.yaml by hpack version 0.37.0. +-- This file has been generated from package.yaml by hpack version 0.38.3. -- -- see: https://github.com/sol/hpack name: hs-opentelemetry-exporter-otlp -version: 0.1.1.0 +version: 1.0.0.0 synopsis: OpenTelemetry exporter supporting the standard OTLP protocol description: Please see the README on GitHub at category: OpenTelemetry, Telemetry, Monitoring, Observability, Metrics @@ -25,11 +25,18 @@ type: git location: https://github.com/iand675/hs-opentelemetry +flag grpc + description: Enable gRPC export support (requires grapesy) + manual: True + default: False + library exposed-modules: OpenTelemetry.Exporter.OTLP OpenTelemetry.Exporter.OTLP.LogRecord OpenTelemetry.Exporter.OTLP.Span + OpenTelemetry.Exporter.OTLP.Metric + OpenTelemetry.Exporter.OTLP.Internal.Config other-modules: Paths_hs_opentelemetry_exporter_otlp hs-source-dirs: @@ -39,16 +46,46 @@ base >=4.7 && <5 , bytestring , case-insensitive - , hs-opentelemetry-api ==0.3.* - , hs-opentelemetry-otlp ==0.2.* - , hs-opentelemetry-propagator-w3c + , hs-opentelemetry-api ==1.0.* + , hs-opentelemetry-otlp ==1.0.* + , hs-opentelemetry-propagator-w3c ==1.0.* , http-client , http-conduit , http-types , microlens , proto-lens >=0.7.1.0 + , random , text + , time , unordered-containers , vector , zlib + default-language: Haskell2010 + if flag(grpc) + exposed-modules: + OpenTelemetry.Exporter.OTLP.GRPC + build-depends: + grapesy + +test-suite hs-opentelemetry-exporter-otlp-test + type: exitcode-stdio-1.0 + main-is: Spec.hs + other-modules: + Paths_hs_opentelemetry_exporter_otlp + hs-source-dirs: + test + ghc-options: -threaded -rtsopts -with-rtsopts=-N + build-depends: + base >=4.7 && <5 + , bytestring + , hs-opentelemetry-api ==1.0.* + , hs-opentelemetry-exporter-otlp ==1.0.* + , hs-opentelemetry-otlp ==1.0.* + , hspec + , http-types + , microlens + , proto-lens + , text + , unordered-containers + , vector default-language: Haskell2010 diff --git a/src/OpenTelemetry/Exporter/OTLP.hs b/src/OpenTelemetry/Exporter/OTLP.hs --- a/src/OpenTelemetry/Exporter/OTLP.hs +++ b/src/OpenTelemetry/Exporter/OTLP.hs @@ -1,7 +1,15 @@ -module OpenTelemetry.Exporter.OTLP - {-# DEPRECATED "use OpenTelemetry.Exporter.OTLP.Span instead" #-} ( +{- | +Module : OpenTelemetry.Exporter.OTLP +Description : Re-exports for OTLP (OpenTelemetry Protocol) exporters. +Stability : experimental +-} +module OpenTelemetry.Exporter.OTLP ( module OpenTelemetry.Exporter.OTLP.Span, + module OpenTelemetry.Exporter.OTLP.Metric, + module OpenTelemetry.Exporter.OTLP.LogRecord, ) where +import OpenTelemetry.Exporter.OTLP.LogRecord +import OpenTelemetry.Exporter.OTLP.Metric import OpenTelemetry.Exporter.OTLP.Span diff --git a/src/OpenTelemetry/Exporter/OTLP/GRPC.hs b/src/OpenTelemetry/Exporter/OTLP/GRPC.hs new file mode 100644 --- /dev/null +++ b/src/OpenTelemetry/Exporter/OTLP/GRPC.hs @@ -0,0 +1,269 @@ +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE NumericUnderscores #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TypeApplications #-} +{-# LANGUAGE TypeFamilies #-} +{-# OPTIONS_GHC -Wno-orphans #-} + +{- | gRPC transport for OTLP export via grapesy. + +Provides gRPC-based span, metric, and log exporters that speak the native +OTLP\/gRPC protocol (default port 4317). + +Requires the @grpc@ cabal flag to be enabled. + +@since 0.2.0.0 +-} +module OpenTelemetry.Exporter.OTLP.GRPC ( + grpcOtlpSpanExporter, + grpcOtlpMetricExporter, + grpcOtlpLogRecordExporter, + grpcEndpoint, + grpcTracesEndpoint, + grpcMetricsEndpoint, + grpcLogsEndpoint, +) where + +import Control.Applicative ((<|>)) +import Control.Exception (SomeException (..), catch) +import Control.Monad (void) +import Control.Monad.IO.Class (MonadIO, liftIO) +import Data.HashMap.Strict (HashMap) +import Data.IORef (atomicModifyIORef', newIORef, readIORef) +import Data.List (isPrefixOf) +import Data.Vector (Vector) +import Network.GRPC.Client ( + Address (..), + CertificateStoreSpec, + Server (..), + ServerValidation (..), + certStoreFromPath, + certStoreFromSystem, + closeConnection, + openConnection, + rpc, + ) +import Network.GRPC.Client.StreamType.IO (nonStreaming) +import Network.GRPC.Common ( + NoMetadata, + RequestMetadata, + ResponseInitialMetadata, + ResponseTrailingMetadata, + def, + ) +import Network.GRPC.Common.Protobuf (Proto (..), Protobuf) +import OpenTelemetry.Exporter.Metric (MetricExporter (..), ResourceMetricsExport) +import OpenTelemetry.Exporter.OTLP.Internal.Config ( + OTLPExporterConfig (..), + grpcEndpoint, + grpcLogsEndpoint, + grpcMetricsEndpoint, + grpcTracesEndpoint, + ) +import OpenTelemetry.Exporter.Span (SpanExporter (..)) +import OpenTelemetry.Internal.Common.Types (ExportResult (..), FlushResult (..), InstrumentationLibrary, ShutdownResult (..)) +import OpenTelemetry.Internal.Log.Types (LogRecordExporter, LogRecordExporterArguments (..), ReadableLogRecord, mkLogRecordExporter) +import OpenTelemetry.Internal.Logging (otelLogWarning) +import qualified OpenTelemetry.Trace.Core as OT +import Proto.Opentelemetry.Proto.Collector.Logs.V1.LogsService (ExportLogsServiceRequest, LogsService) +import Proto.Opentelemetry.Proto.Collector.Metrics.V1.MetricsService (ExportMetricsServiceRequest, MetricsService) +import Proto.Opentelemetry.Proto.Collector.Trace.V1.TraceService (ExportTraceServiceRequest, TraceService) +import System.Timeout (timeout) + + +type ExportTracesRPC = Protobuf TraceService "export" + + +type ExportMetricsRPC = Protobuf MetricsService "export" + + +type ExportLogsRPC = Protobuf LogsService "export" + + +type instance RequestMetadata (Protobuf TraceService meth) = NoMetadata + + +type instance ResponseInitialMetadata (Protobuf TraceService meth) = NoMetadata + + +type instance ResponseTrailingMetadata (Protobuf TraceService meth) = NoMetadata + + +type instance RequestMetadata (Protobuf MetricsService meth) = NoMetadata + + +type instance ResponseInitialMetadata (Protobuf MetricsService meth) = NoMetadata + + +type instance ResponseTrailingMetadata (Protobuf MetricsService meth) = NoMetadata + + +type instance RequestMetadata (Protobuf LogsService meth) = NoMetadata + + +type instance ResponseInitialMetadata (Protobuf LogsService meth) = NoMetadata + + +type instance ResponseTrailingMetadata (Protobuf LogsService meth) = NoMetadata + + +parseGrpcAddress :: String -> Address +parseGrpcAddress url = + let stripped = dropScheme url + (host, portPart) = break (== ':') stripped + port = case portPart of + ':' : rest -> case reads rest of + [(p, "")] -> p + [(p, "/")] -> p + _ -> 4317 + _ -> 4317 + in Address + { addressHost = host + , addressPort = port + , addressAuthority = Nothing + } + where + dropScheme s = case break (== '/') (drop 1 $ dropWhile (/= ':') s) of + (_, '/' : rest) -> rest + _ -> s + + +resolveGrpcServer :: Bool -> Maybe FilePath -> String -> Server +resolveGrpcServer insecure mCert endpoint + | insecure = ServerInsecure addr + | "https://" `isPrefixOf` endpoint = ServerSecure validation def addr + | otherwise = ServerInsecure addr + where + addr = parseGrpcAddress endpoint + validation = case mCert of + Just certPath -> ValidateServer (certStoreFromPath certPath) + Nothing -> ValidateServer certStoreFromSystem + + +{- | Create a gRPC-based span exporter. + +The serialization function converts the SDK's span map into the OTLP +protobuf request. Pass 'OpenTelemetry.Exporter.OTLP.Span.immutableSpansToProtobuf'. +-} +grpcOtlpSpanExporter + :: (MonadIO m) + => OTLPExporterConfig + -> (HashMap InstrumentationLibrary (Vector OT.ImmutableSpan) -> IO ExportTraceServiceRequest) + -> m SpanExporter +grpcOtlpSpanExporter conf toProto = liftIO $ do + let endpoint = grpcTracesEndpoint conf + server = resolveGrpcServer (otlpTracesInsecure conf || otlpInsecure conf) (otlpTracesCertificate conf <|> otlpCertificate conf) endpoint + conn <- openConnection def server + shutdownRef <- newIORef False + pure $ + SpanExporter + { spanExporterExport = \spans_ -> do + isShut <- readIORef shutdownRef + if isShut + then pure $ Failure Nothing + else do + req <- toProto spans_ + let timeoutUs = maybe 10_000_000 (* 1_000) (otlpTracesTimeout conf <|> otlpTimeout conf) + ( do + result <- timeout timeoutUs $ nonStreaming conn (rpc @ExportTracesRPC) (Proto req) + case result of + Nothing -> do + otelLogWarning "gRPC trace export timed out" + pure $ Failure Nothing + Just _ -> pure Success + ) + `catch` \(e :: SomeException) -> do + otelLogWarning $ "gRPC trace export failed: " <> show e + pure $ Failure (Just e) + , spanExporterShutdown = do + void $ atomicModifyIORef' shutdownRef $ \s -> (True, s) + closeConnection conn + pure ShutdownSuccess + , spanExporterForceFlush = pure FlushSuccess + } + + +{- | Create a gRPC-based metric exporter. + +The serialization function converts metric batches into the OTLP +protobuf request. Pass 'OpenTelemetry.Exporter.OTLP.Metric.resourceMetricsToExportRequest'. +-} +grpcOtlpMetricExporter + :: (MonadIO m) + => OTLPExporterConfig + -> (Vector ResourceMetricsExport -> ExportMetricsServiceRequest) + -> m MetricExporter +grpcOtlpMetricExporter conf toProto = liftIO $ do + let endpoint = grpcMetricsEndpoint conf + server = resolveGrpcServer (otlpMetricsInsecure conf || otlpInsecure conf) (otlpMetricsCertificate conf <|> otlpCertificate conf) endpoint + conn <- openConnection def server + shutdownRef <- newIORef False + pure $ + MetricExporter + { metricExporterExport = \rmes -> do + isShut <- readIORef shutdownRef + if isShut + then pure $ Failure Nothing + else do + let req = toProto rmes + timeoutUs = maybe 10_000_000 (* 1_000) (otlpMetricsTimeout conf <|> otlpTimeout conf) + ( do + result <- timeout timeoutUs $ nonStreaming conn (rpc @ExportMetricsRPC) (Proto req) + case result of + Nothing -> do + otelLogWarning "gRPC metric export timed out" + pure $ Failure Nothing + Just _ -> pure Success + ) + `catch` \(e :: SomeException) -> do + otelLogWarning $ "gRPC metric export failed: " <> show e + pure $ Failure (Just e) + , metricExporterShutdown = do + void $ atomicModifyIORef' shutdownRef $ \s -> (True, s) + closeConnection conn + pure ShutdownSuccess + , metricExporterForceFlush = pure FlushSuccess + } + + +{- | Create a gRPC-based log record exporter. + +The serialization function converts log records into the OTLP +protobuf request. +-} +grpcOtlpLogRecordExporter + :: (MonadIO m) + => OTLPExporterConfig + -> (Vector ReadableLogRecord -> IO ExportLogsServiceRequest) + -> m LogRecordExporter +grpcOtlpLogRecordExporter conf toProto = liftIO $ do + let endpoint = grpcLogsEndpoint conf + server = resolveGrpcServer (otlpLogsInsecure conf || otlpInsecure conf) (otlpLogsCertificate conf <|> otlpCertificate conf) endpoint + conn <- openConnection def server + shutdownRef <- newIORef False + mkLogRecordExporter + LogRecordExporterArguments + { logRecordExporterArgumentsExport = \lrs -> do + isShut <- readIORef shutdownRef + if isShut + then pure $ Failure Nothing + else do + req <- toProto lrs + let timeoutUs = maybe 10_000_000 (* 1_000) (otlpLogsTimeout conf <|> otlpTimeout conf) + ( do + result <- timeout timeoutUs $ nonStreaming conn (rpc @ExportLogsRPC) (Proto req) + case result of + Nothing -> do + otelLogWarning "gRPC log export timed out" + pure $ Failure Nothing + Just _ -> pure Success + ) + `catch` \(e :: SomeException) -> do + otelLogWarning $ "gRPC log export failed: " <> show e + pure $ Failure (Just e) + , logRecordExporterArgumentsForceFlush = pure FlushSuccess + , logRecordExporterArgumentsShutdown = do + void $ atomicModifyIORef' shutdownRef $ \s -> (True, s) + closeConnection conn + } diff --git a/src/OpenTelemetry/Exporter/OTLP/Internal/Config.hs b/src/OpenTelemetry/Exporter/OTLP/Internal/Config.hs new file mode 100644 --- /dev/null +++ b/src/OpenTelemetry/Exporter/OTLP/Internal/Config.hs @@ -0,0 +1,395 @@ +{- FOURMOLU_DISABLE -} +{-# LANGUAGE CPP #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE NumericUnderscores #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} + +{- | Internal module containing shared OTLP exporter configuration types. + +Not intended for direct use — import via "OpenTelemetry.Exporter.OTLP.Span" instead. + += Environment variables + +'loadExporterEnvironmentVariables' reads the following @OTEL_EXPORTER_OTLP_*@ +variables. Each generic variable has per-signal overrides for traces, metrics, +and logs (e.g. @OTEL_EXPORTER_OTLP_TRACES_ENDPOINT@). Per-signal values take +precedence over generic ones. + +=== Endpoint + +* @OTEL_EXPORTER_OTLP_ENDPOINT@ (default: @http:\/\/localhost:4318@ HTTP, + @http:\/\/localhost:4317@ gRPC) +* @OTEL_EXPORTER_OTLP_TRACES_ENDPOINT@, @OTEL_EXPORTER_OTLP_METRICS_ENDPOINT@, + @OTEL_EXPORTER_OTLP_LOGS_ENDPOINT@ + +=== Security + +* @OTEL_EXPORTER_OTLP_INSECURE@ (default: @false@) +* @OTEL_EXPORTER_OTLP_TRACES_INSECURE@ (fallback: @OTEL_EXPORTER_OTLP_SPAN_INSECURE@) +* @OTEL_EXPORTER_OTLP_METRICS_INSECURE@ (fallback: @OTEL_EXPORTER_OTLP_METRIC_INSECURE@) +* @OTEL_EXPORTER_OTLP_LOGS_INSECURE@ +* @OTEL_EXPORTER_OTLP_CERTIFICATE@, @OTEL_EXPORTER_OTLP_TRACES_CERTIFICATE@, + @OTEL_EXPORTER_OTLP_METRICS_CERTIFICATE@, @OTEL_EXPORTER_OTLP_LOGS_CERTIFICATE@ + +=== Headers + +* @OTEL_EXPORTER_OTLP_HEADERS@, @OTEL_EXPORTER_OTLP_TRACES_HEADERS@, + @OTEL_EXPORTER_OTLP_METRICS_HEADERS@, @OTEL_EXPORTER_OTLP_LOGS_HEADERS@ + +Format: @key1=value1,key2=value2@ (W3C Baggage encoding). + +=== Compression + +* @OTEL_EXPORTER_OTLP_COMPRESSION@, @OTEL_EXPORTER_OTLP_TRACES_COMPRESSION@, + @OTEL_EXPORTER_OTLP_METRICS_COMPRESSION@, @OTEL_EXPORTER_OTLP_LOGS_COMPRESSION@ + +Values: @gzip@, @none@ (default: @none@). + +=== Timeout + +* @OTEL_EXPORTER_OTLP_TIMEOUT@, @OTEL_EXPORTER_OTLP_TRACES_TIMEOUT@, + @OTEL_EXPORTER_OTLP_METRICS_TIMEOUT@, @OTEL_EXPORTER_OTLP_LOGS_TIMEOUT@ + +Milliseconds (default: @10000@). + +=== Protocol + +* @OTEL_EXPORTER_OTLP_PROTOCOL@, @OTEL_EXPORTER_OTLP_TRACES_PROTOCOL@, + @OTEL_EXPORTER_OTLP_METRICS_PROTOCOL@, @OTEL_EXPORTER_OTLP_LOGS_PROTOCOL@ + +Values: @http\/protobuf@ (default), @grpc@. +-} +module OpenTelemetry.Exporter.OTLP.Internal.Config ( + OTLPExporterConfig (..), + CompressionFormat (..), + Protocol (..), + loadExporterEnvironmentVariables, + otlpExporterHttpEndpoint, + otlpExporterGRpcEndpoint, + defaultExporterTimeout, + readCompressionFormat, + readProtocol, + readTimeout, + putWarningLn, + + -- * Shared HTTP transport helpers + otlpUserAgent, + httpProtobufContentType, + httpSignalEndpointUrl, + isRetryableHttpStatus, + + -- * Retry-After parsing + parseRetryAfterMicros, + + -- * gRPC endpoint resolution (same env vars as HTTP; default port 4317) + grpcEndpoint, + grpcTracesEndpoint, + grpcMetricsEndpoint, + grpcLogsEndpoint, +) where + +import Control.Monad.IO.Class (MonadIO, liftIO) +import Data.Char (toLower) +import Data.Function ((&)) +import Data.Maybe (fromMaybe) +import Data.Version (showVersion) +import qualified Data.ByteString.Char8 as C +import qualified Data.CaseInsensitive as CI +import qualified Data.HashMap.Strict as H +import qualified Data.Text.Encoding as T +import Data.Time.Clock (getCurrentTime, diffUTCTime) +import Data.Time.Format (parseTimeM, defaultTimeLocale, rfc822DateFormat) +import Network.HTTP.Types.Header (Header) +import Network.HTTP.Types.Status (Status, status429, status502, status503, status504) +import qualified OpenTelemetry.Baggage as Baggage +import OpenTelemetry.Environment (lookupBooleanEnv) +import OpenTelemetry.Internal.Logging (otelLogWarning) +import Paths_hs_opentelemetry_exporter_otlp (version) +import System.Environment (lookupEnv) +import Text.Read (readMaybe) + + +data OTLPExporterConfig = OTLPExporterConfig + { otlpEndpoint :: Maybe String + , otlpTracesEndpoint :: Maybe String + , otlpMetricsEndpoint :: Maybe String + , otlpLogsEndpoint :: Maybe String + , otlpInsecure :: Bool + , otlpTracesInsecure :: Bool + , otlpMetricsInsecure :: Bool + , otlpLogsInsecure :: Bool + , otlpCertificate :: Maybe FilePath + , otlpTracesCertificate :: Maybe FilePath + , otlpMetricsCertificate :: Maybe FilePath + , otlpLogsCertificate :: Maybe FilePath + , otlpHeaders :: Maybe [Header] + , otlpTracesHeaders :: Maybe [Header] + , otlpMetricsHeaders :: Maybe [Header] + , otlpLogsHeaders :: Maybe [Header] + , otlpCompression :: Maybe CompressionFormat + , otlpTracesCompression :: Maybe CompressionFormat + , otlpMetricsCompression :: Maybe CompressionFormat + , otlpLogsCompression :: Maybe CompressionFormat + , otlpTimeout :: Maybe Int + -- ^ Measured in milliseconds. + , otlpTracesTimeout :: Maybe Int + -- ^ Measured in milliseconds. + , otlpMetricsTimeout :: Maybe Int + -- ^ Measured in milliseconds. + , otlpLogsTimeout :: Maybe Int + -- ^ Measured in milliseconds. + , otlpProtocol :: Maybe Protocol + , otlpTracesProtocol :: Maybe Protocol + , otlpMetricsProtocol :: Maybe Protocol + , otlpLogsProtocol :: Maybe Protocol + , otlpConcurrentExports :: !Int + -- ^ Maximum number of concurrent export requests per signal. + -- Default: 1 (sequential). Set higher for pipelined export. + -- Spec: + } + + +loadExporterEnvironmentVariables :: (MonadIO m) => m OTLPExporterConfig +loadExporterEnvironmentVariables = liftIO $ do + hdrs <- lookupDecodeHeaders "OTEL_EXPORTER_OTLP_HEADERS" + tracesHdrs <- lookupDecodeHeaders "OTEL_EXPORTER_OTLP_TRACES_HEADERS" + metricsHdrs <- lookupDecodeHeaders "OTEL_EXPORTER_OTLP_METRICS_HEADERS" + logsHdrs <- lookupDecodeHeaders "OTEL_EXPORTER_OTLP_LOGS_HEADERS" + OTLPExporterConfig + <$> lookupEnv "OTEL_EXPORTER_OTLP_ENDPOINT" + <*> lookupEnv "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT" + <*> lookupEnv "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT" + <*> lookupEnv "OTEL_EXPORTER_OTLP_LOGS_ENDPOINT" + <*> lookupBooleanEnv "OTEL_EXPORTER_OTLP_INSECURE" + <*> lookupBooleanEnvFallback "OTEL_EXPORTER_OTLP_TRACES_INSECURE" "OTEL_EXPORTER_OTLP_SPAN_INSECURE" + <*> lookupBooleanEnvFallback "OTEL_EXPORTER_OTLP_METRICS_INSECURE" "OTEL_EXPORTER_OTLP_METRIC_INSECURE" + <*> lookupBooleanEnv "OTEL_EXPORTER_OTLP_LOGS_INSECURE" + <*> lookupEnv "OTEL_EXPORTER_OTLP_CERTIFICATE" + <*> lookupEnv "OTEL_EXPORTER_OTLP_TRACES_CERTIFICATE" + <*> lookupEnv "OTEL_EXPORTER_OTLP_METRICS_CERTIFICATE" + <*> lookupEnv "OTEL_EXPORTER_OTLP_LOGS_CERTIFICATE" + <*> pure hdrs + <*> pure tracesHdrs + <*> pure metricsHdrs + <*> pure logsHdrs + <*> (traverse readCompressionFormat =<< lookupEnv "OTEL_EXPORTER_OTLP_COMPRESSION") + <*> (traverse readCompressionFormat =<< lookupEnv "OTEL_EXPORTER_OTLP_TRACES_COMPRESSION") + <*> (traverse readCompressionFormat =<< lookupEnv "OTEL_EXPORTER_OTLP_METRICS_COMPRESSION") + <*> (traverse readCompressionFormat =<< lookupEnv "OTEL_EXPORTER_OTLP_LOGS_COMPRESSION") + <*> (traverse readTimeout =<< lookupEnv "OTEL_EXPORTER_OTLP_TIMEOUT") + <*> (traverse readTimeout =<< lookupEnv "OTEL_EXPORTER_OTLP_TRACES_TIMEOUT") + <*> (traverse readTimeout =<< lookupEnv "OTEL_EXPORTER_OTLP_METRICS_TIMEOUT") + <*> (traverse readTimeout =<< lookupEnv "OTEL_EXPORTER_OTLP_LOGS_TIMEOUT") + <*> (traverse readProtocol =<< lookupEnv "OTEL_EXPORTER_OTLP_PROTOCOL") + <*> (traverse readProtocol =<< lookupEnv "OTEL_EXPORTER_OTLP_TRACES_PROTOCOL") + <*> (traverse readProtocol =<< lookupEnv "OTEL_EXPORTER_OTLP_METRICS_PROTOCOL") + <*> (traverse readProtocol =<< lookupEnv "OTEL_EXPORTER_OTLP_LOGS_PROTOCOL") + <*> (maybe 1 (\s -> max 1 (fromMaybe 1 (readMaybe s))) <$> lookupEnv "OTEL_EXPORTER_OTLP_CONCURRENT_EXPORTS") + where + lookupDecodeHeaders :: String -> IO (Maybe [Header]) + lookupDecodeHeaders envName = do + m <- lookupEnv envName + case m of + Nothing -> pure Nothing + Just hsString -> case Baggage.decodeBaggageHeader $ C.pack hsString of + Left _err -> do + otelLogWarning ("Failed to parse " <> envName <> ", ignoring") + pure (Just []) + Right baggageFmt -> + pure $ + Just $ + (\(k, v) -> (CI.mk $ Baggage.tokenValue k, T.encodeUtf8 $ Baggage.value v)) + <$> H.toList (Baggage.values baggageFmt) + + lookupBooleanEnvFallback primary fallback = do + p <- lookupBooleanEnv primary + if p then pure True else lookupBooleanEnv fallback + + +data CompressionFormat + = None + | GZip + + +{- | +The OpenTelemetry Protocol. + +@http\/protobuf@ is always available. When the @grpc@ cabal flag is enabled, +@grpc@ is also supported (native HTTP\/2 gRPC via grapesy). +-} +data Protocol + = HttpProtobuf +#ifdef GRPC_ENABLED + | GRpc +#endif + + +readCompressionFormat :: (MonadIO m) => String -> m CompressionFormat +readCompressionFormat compressionFormat = + compressionFormat & fmap toLower & \case + "gzip" -> pure GZip + "none" -> pure None + _ -> do + putWarningLn $ "Unsupported compression format '" <> compressionFormat <> "'" + pure None + + +readProtocol :: (MonadIO m) => String -> m Protocol +readProtocol protocol = + protocol & fmap toLower & \case + "http/protobuf" -> pure HttpProtobuf +#ifdef GRPC_ENABLED + "grpc" -> pure GRpc +#endif + _ -> do + putWarningLn $ "Unsupported protocol '" <> protocol <> "'" + pure HttpProtobuf + + +readTimeout :: (MonadIO m) => String -> m Int +readTimeout timeout = + case readMaybe timeout of + Just timeoutInt | timeoutInt >= 0 -> pure timeoutInt + _otherwise -> do + putWarningLn $ "Unsupported timeout value '" <> timeout <> "'" + pure defaultExporterTimeout + + +defaultExporterTimeout :: Int +defaultExporterTimeout = 10_000 + + +otlpExporterHttpEndpoint :: C.ByteString +otlpExporterHttpEndpoint = "http://localhost:4318" + + +otlpExporterGRpcEndpoint :: C.ByteString +otlpExporterGRpcEndpoint = "http://localhost:4317" + + +{- | Base OTLP gRPC URL from @OTEL_EXPORTER_OTLP_ENDPOINT@. + +Defaults to @http:\/\/localhost:4317@. Per-signal +@OTEL_EXPORTER_OTLP_TRACES_ENDPOINT@ (and metrics\/logs variants) override via +'grpcTracesEndpoint' and siblings. +-} +grpcEndpoint :: OTLPExporterConfig -> String +grpcEndpoint conf = + fromMaybe "http://localhost:4317" (otlpEndpoint conf) + + +{- | Effective gRPC URL for traces: per-signal env overrides the generic endpoint. + +See . +-} +grpcTracesEndpoint :: OTLPExporterConfig -> String +grpcTracesEndpoint conf = + fromMaybe (grpcEndpoint conf) (otlpTracesEndpoint conf) + + +{- | Effective gRPC URL for metrics (per-signal override or generic base). +-} +grpcMetricsEndpoint :: OTLPExporterConfig -> String +grpcMetricsEndpoint conf = + fromMaybe (grpcEndpoint conf) (otlpMetricsEndpoint conf) + + +{- | Effective gRPC URL for logs (per-signal override or generic base). +-} +grpcLogsEndpoint :: OTLPExporterConfig -> String +grpcLogsEndpoint conf = + fromMaybe (grpcEndpoint conf) (otlpLogsEndpoint conf) + + +putWarningLn :: (MonadIO m) => String -> m () +putWarningLn = liftIO . otelLogWarning + + +-- | @OTel-OTLP-Exporter-Haskell\/\@ per the OTLP spec User-Agent convention. +otlpUserAgent :: C.ByteString +otlpUserAgent = C.pack $ "OTel-OTLP-Exporter-Haskell/" <> showVersion version + + +httpProtobufContentType :: C.ByteString +httpProtobufContentType = "application/x-protobuf" + + +{- | Build the effective endpoint URL for a signal. + +Per the OTLP exporter spec: + +* If a per-signal endpoint is set, use it as-is. +* Otherwise, append the signal path (@\/v1\/traces@, @\/v1\/metrics@, @\/v1\/logs@) + to the base endpoint. +-} +httpSignalEndpointUrl + :: Maybe String + -- ^ Per-signal endpoint (e.g. 'otlpTracesEndpoint') + -> OTLPExporterConfig + -> String + -- ^ Signal path, e.g. @\"\/v1\/traces\"@ + -> String +httpSignalEndpointUrl perSignalEndpoint conf signalPath = + case perSignalEndpoint of + Just url -> url + Nothing -> + let base = maybe "http://localhost:4318" id (otlpEndpoint conf) + in ensureTrailingSlash base <> dropWhile (== '/') signalPath + + +ensureTrailingSlash :: String -> String +ensureTrailingSlash [] = "/" +ensureTrailingSlash s + | last s == '/' = s + | otherwise = s <> "/" + + +{- | Whether an HTTP status code is retryable per the OTLP spec. + +Only 429 (Too Many Requests), 502 (Bad Gateway), 503 (Service Unavailable), +and 504 (Gateway Timeout) are retryable. All other status codes, including +other 4xx\/5xx codes, MUST NOT be retried. + +Spec: +-} +isRetryableHttpStatus :: Status -> Bool +isRetryableHttpStatus s = + s == status429 || s == status502 || s == status503 || s == status504 + + +{- | Parse an HTTP @Retry-After@ header value into microseconds. + +Supports both formats defined in RFC 7231 Section 7.1.3: + +* @delay-seconds@ — a non-negative integer (e.g. @\"120\"@) +* @HTTP-date@ — RFC 822 / RFC 1123 format (e.g. @\"Fri, 31 Dec 1999 23:59:59 GMT\"@) + +Returns 'Nothing' when the value cannot be parsed in either format or +the resulting delay is non-positive. + +Spec: + +@since 0.1.2.0 +-} +parseRetryAfterMicros :: C.ByteString -> IO (Maybe Int) +parseRetryAfterMicros bs = + case readMaybe (C.unpack bs) of + Just (seconds :: Int) + | seconds > 0 -> pure $ Just (seconds * 1_000_000) + _ -> do + let dateStr = C.unpack bs + case parseTimeM True defaultTimeLocale rfc822DateFormat dateStr of + Just targetTime -> do + now <- getCurrentTime + let diffSecs = diffUTCTime targetTime now + micros = ceiling (diffSecs * 1_000_000) :: Int + pure $ if micros > 0 then Just micros else Nothing + Nothing -> + case parseTimeM True defaultTimeLocale "%a, %d %b %Y %H:%M:%S GMT" dateStr of + Just targetTime -> do + now <- getCurrentTime + let diffSecs = diffUTCTime targetTime now + micros = ceiling (diffSecs * 1_000_000) :: Int + pure $ if micros > 0 then Just micros else Nothing + Nothing -> pure Nothing diff --git a/src/OpenTelemetry/Exporter/OTLP/LogRecord.hs b/src/OpenTelemetry/Exporter/OTLP/LogRecord.hs --- a/src/OpenTelemetry/Exporter/OTLP/LogRecord.hs +++ b/src/OpenTelemetry/Exporter/OTLP/LogRecord.hs @@ -1,2 +1,362 @@ -module OpenTelemetry.Exporter.OTLP.LogRecord () where +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE NumericUnderscores #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} +module OpenTelemetry.Exporter.OTLP.LogRecord ( + otlpLogRecordExporter, + immutableLogRecordToProto, +) where + +import Codec.Compression.GZip (compress) +import Control.Concurrent (threadDelay) +import Control.Exception (SomeAsyncException (..), SomeException (..), fromException, throwIO, try) +import Control.Monad.IO.Class (MonadIO, liftIO) +import Data.Bits (shiftL) +import qualified Data.ByteString as BS +import qualified Data.ByteString.Char8 as C +import qualified Data.ByteString.Lazy as L +import qualified Data.HashMap.Strict as H +import Data.List (isInfixOf) +import Data.Maybe (fromMaybe) +import Data.ProtoLens (defMessage, encodeMessage) +import Data.Text (Text) +import qualified Data.Text as T +import qualified Data.Vector as V +import Data.Word (Word64) +import Lens.Micro ((&), (.~)) +import Network.HTTP.Client +import Network.HTTP.Simple (httpBS) +import Network.HTTP.Types.Header +import Network.HTTP.Types.Status +import qualified OpenTelemetry.Attributes as A +import OpenTelemetry.Common (Timestamp (..)) +import OpenTelemetry.Exporter.OTLP.Span (CompressionFormat (..), OTLPExporterConfig (..)) +import OpenTelemetry.Internal.Common.Types +import OpenTelemetry.Internal.Log.Types +import OpenTelemetry.Internal.Trace.Id (spanIdBytes, traceIdBytes) +import OpenTelemetry.LogAttributes (AnyValue (..), LogAttributes (..)) +import OpenTelemetry.Resource (MaterializedResources, emptyMaterializedResources, getMaterializedResourcesAttributes, getMaterializedResourcesSchema) +import OpenTelemetry.Trace.Core (timestampNanoseconds, traceFlagsValue) +import Proto.Opentelemetry.Proto.Collector.Logs.V1.LogsService (ExportLogsServiceRequest) +import qualified Proto.Opentelemetry.Proto.Collector.Logs.V1.LogsService_Fields as LSF +import Proto.Opentelemetry.Proto.Common.V1.Common (KeyValue) +import qualified Proto.Opentelemetry.Proto.Common.V1.Common as Common +import qualified Proto.Opentelemetry.Proto.Common.V1.Common_Fields as CF +import Proto.Opentelemetry.Proto.Logs.V1.Logs (ResourceLogs, ScopeLogs) +import qualified Proto.Opentelemetry.Proto.Logs.V1.Logs as PL +import qualified Proto.Opentelemetry.Proto.Logs.V1.Logs_Fields as LF +import qualified Proto.Opentelemetry.Proto.Resource.V1.Resource as Res +import qualified Proto.Opentelemetry.Proto.Resource.V1.Resource_Fields as RF +import Text.Read (readMaybe) + + +defaultExporterTimeout :: Int +defaultExporterTimeout = 10_000 + + +httpProtobufMimeType :: C.ByteString +httpProtobufMimeType = "application/x-protobuf" + + +otlpLogRecordExporter :: (MonadIO m) => OTLPExporterConfig -> m LogRecordExporter +otlpLogRecordExporter conf = liftIO $ do + req <- parseRequest (logsEndpointUrl conf) + let (encodingHeaders, encoder) = httpLogsCompression conf + let baseReq = + req + { method = "POST" + , requestHeaders = encodingHeaders <> httpLogsBaseHeaders conf req + , responseTimeout = httpLogsResponseTimeout conf + } + mkLogRecordExporter + LogRecordExporterArguments + { logRecordExporterArgumentsExport = \lrs -> do + if V.null lrs + then pure Success + else do + result <- try $ exporterExportCall encoder baseReq lrs + case result of + Left err -> case fromException err of + Just (SomeAsyncException _) -> throwIO err + Nothing -> pure $ Failure $ Just err + Right ok -> pure ok + , logRecordExporterArgumentsForceFlush = pure FlushSuccess + , logRecordExporterArgumentsShutdown = pure () + } + where + retryDelay = 100_000 + maxRetryCount = 5 + isRetryableStatusCode status_ = + status_ == status408 || status_ == status429 || (statusCode status_ >= 500 && statusCode status_ < 600) + isRetryableException = \case + ResponseTimeout -> True + ConnectionTimeout -> True + ConnectionFailure _ -> True + ConnectionClosed -> True + _ -> False + + exporterExportCall encoder baseReq lrs = do + rl <- buildResourceLogsFromBatch lrs + let exportReq :: ExportLogsServiceRequest + exportReq = + defMessage + & LSF.vec'resourceLogs + .~ V.singleton rl + let msg = encodeMessage exportReq + let req = + baseReq + { requestBody = RequestBodyLBS $ encoder $ L.fromStrict msg + } + sendReq req 0 + + sendReq req backoffCount = do + eResp <- try $ httpBS req + let exponentialBackoff = + if backoffCount == maxRetryCount + then pure $ Failure Nothing + else do + threadDelay (retryDelay `shiftL` backoffCount) + sendReq req (backoffCount + 1) + case eResp of + Left (HttpExceptionRequest _req' e) + | isRetryableException e -> exponentialBackoff + Left err -> pure $ Failure $ Just $ SomeException err + Right resp -> + if isRetryableStatusCode (responseStatus resp) + then case lookup hRetryAfter $ responseHeaders resp of + Nothing -> exponentialBackoff + Just retryAfter -> case readMaybe $ C.unpack retryAfter of + Nothing -> exponentialBackoff + Just seconds -> do + threadDelay (seconds * 1_000_000) + sendReq req (backoffCount + 1) + else + if statusCode (responseStatus resp) >= 300 + then pure $ Failure Nothing + else pure Success + + +groupByScope + :: H.HashMap InstrumentationLibrary [ReadableLogRecord] + -> ReadableLogRecord + -> IO (H.HashMap InstrumentationLibrary [ReadableLogRecord]) +groupByScope acc lr = do + let scope = readLogRecordInstrumentationScope lr + pure $ H.insertWith (++) scope [lr] acc + + +buildResourceLogsFromBatch :: V.Vector ReadableLogRecord -> IO ResourceLogs +buildResourceLogsFromBatch lrs = do + grouped <- V.foldM' groupByScope H.empty lrs + scopeLogsList <- mapM (uncurry buildScopeLogs) (H.toList grouped) + let res = if V.null lrs then emptyMaterializedResources else readLogRecordResource (V.head lrs) + pure $ + defMessage + & LF.resource + .~ materializedResourceToProto res + & LF.vec'scopeLogs + .~ V.fromList scopeLogsList + & LF.schemaUrl + .~ maybe T.empty T.pack (getMaterializedResourcesSchema res) + + +buildScopeLogs :: InstrumentationLibrary -> [ReadableLogRecord] -> IO ScopeLogs +buildScopeLogs scope lrs = do + protoRecords <- mapM readableLogRecordToProtoIO lrs + pure $ + defMessage + & LF.scope + .~ instrumentationLibraryToProto scope + & LF.vec'logRecords + .~ V.fromList protoRecords + & LF.schemaUrl + .~ librarySchemaUrl scope + + +readableLogRecordToProtoIO :: ReadableLogRecord -> IO PL.LogRecord +readableLogRecordToProtoIO rlr = do + ilr <- readLogRecord rlr + pure $ immutableLogRecordToProto ilr + + +immutableLogRecordToProto :: ImmutableLogRecord -> PL.LogRecord +immutableLogRecordToProto ImmutableLogRecord {..} = + defMessage + & LF.timeUnixNano + .~ maybe 0 tsToNanos (toBaseMaybe logRecordTimestamp) + & LF.observedTimeUnixNano + .~ tsToNanos logRecordObservedTimestamp + & LF.severityNumber + .~ maybe PL.SEVERITY_NUMBER_UNSPECIFIED severityToProto (toBaseMaybe logRecordSeverityNumber) + & LF.severityText + .~ fromMaybe "" (toBaseMaybe logRecordSeverityText) + & LF.maybe'body + .~ Just (anyValueToProto logRecordBody) + & LF.vec'attributes + .~ logAttributesToProto logRecordAttributes + & LF.droppedAttributesCount + .~ fromIntegral (attributesDropped logRecordAttributes) + & LF.traceId + .~ case logRecordTracingDetails of + TracingDetails tid _ _ -> traceIdBytes tid + NoTracingDetails -> BS.empty + & LF.spanId + .~ case logRecordTracingDetails of + TracingDetails _ sid _ -> spanIdBytes sid + NoTracingDetails -> BS.empty + & LF.flags + .~ case logRecordTracingDetails of + TracingDetails _ _ fl -> fromIntegral (traceFlagsValue fl) + NoTracingDetails -> 0 + & LF.eventName + .~ fromMaybe "" (toBaseMaybe logRecordEventName) + + +tsToNanos :: Timestamp -> Word64 +tsToNanos = fromIntegral . timestampNanoseconds + + +severityToProto :: SeverityNumber -> PL.SeverityNumber +severityToProto = \case + Trace -> PL.SEVERITY_NUMBER_TRACE + Trace2 -> PL.SEVERITY_NUMBER_TRACE2 + Trace3 -> PL.SEVERITY_NUMBER_TRACE3 + Trace4 -> PL.SEVERITY_NUMBER_TRACE4 + Debug -> PL.SEVERITY_NUMBER_DEBUG + Debug2 -> PL.SEVERITY_NUMBER_DEBUG2 + Debug3 -> PL.SEVERITY_NUMBER_DEBUG3 + Debug4 -> PL.SEVERITY_NUMBER_DEBUG4 + Info -> PL.SEVERITY_NUMBER_INFO + Info2 -> PL.SEVERITY_NUMBER_INFO2 + Info3 -> PL.SEVERITY_NUMBER_INFO3 + Info4 -> PL.SEVERITY_NUMBER_INFO4 + Warn -> PL.SEVERITY_NUMBER_WARN + Warn2 -> PL.SEVERITY_NUMBER_WARN2 + Warn3 -> PL.SEVERITY_NUMBER_WARN3 + Warn4 -> PL.SEVERITY_NUMBER_WARN4 + Error -> PL.SEVERITY_NUMBER_ERROR + Error2 -> PL.SEVERITY_NUMBER_ERROR2 + Error3 -> PL.SEVERITY_NUMBER_ERROR3 + Error4 -> PL.SEVERITY_NUMBER_ERROR4 + Fatal -> PL.SEVERITY_NUMBER_FATAL + Fatal2 -> PL.SEVERITY_NUMBER_FATAL2 + Fatal3 -> PL.SEVERITY_NUMBER_FATAL3 + Fatal4 -> PL.SEVERITY_NUMBER_FATAL4 + Unknown _ -> PL.SEVERITY_NUMBER_UNSPECIFIED + + +logAttributesToProto :: LogAttributes -> V.Vector KeyValue +logAttributesToProto LogAttributes {..} = + V.fromList $ fmap anyValueToKeyValue $ H.toList attributes + where + anyValueToKeyValue :: (Text, OpenTelemetry.LogAttributes.AnyValue) -> KeyValue + anyValueToKeyValue (k, v) = + defMessage + & CF.key .~ k + & CF.value .~ anyValueToProto v + + +anyValueToProto :: OpenTelemetry.LogAttributes.AnyValue -> Common.AnyValue +anyValueToProto = \case + TextValue t -> defMessage & CF.stringValue .~ t + BoolValue b -> defMessage & CF.boolValue .~ b + DoubleValue d -> defMessage & CF.doubleValue .~ d + IntValue i -> defMessage & CF.intValue .~ fromIntegral i + ByteStringValue bs -> defMessage & CF.bytesValue .~ bs + ArrayValue arr -> + defMessage + & CF.arrayValue + .~ (defMessage & CF.values .~ fmap anyValueToProto arr) + HashMapValue hm -> + defMessage + & CF.kvlistValue + .~ (defMessage & CF.values .~ fmap (\(k, v) -> defMessage & CF.key .~ k & CF.value .~ anyValueToProto v) (H.toList hm)) + NullValue -> defMessage + + +materializedResourceToProto :: MaterializedResources -> Res.Resource +materializedResourceToProto r = + let attrs = getMaterializedResourcesAttributes r + in defMessage + & RF.vec'attributes + .~ attrsToProto attrs + & RF.droppedAttributesCount + .~ fromIntegral (A.getDropped attrs) + + +instrumentationLibraryToProto :: InstrumentationLibrary -> Common.InstrumentationScope +instrumentationLibraryToProto InstrumentationLibrary {..} = + defMessage + & CF.name .~ libraryName + & CF.version .~ libraryVersion + & CF.vec'attributes .~ attrsToProto libraryAttributes + & CF.droppedAttributesCount .~ fromIntegral (A.getDropped libraryAttributes) + + +attrsToProto :: A.Attributes -> V.Vector KeyValue +attrsToProto = + V.fromList + . fmap attrToKeyValue + . H.toList + . A.getAttributeMap + where + primToAnyValue = \case + A.TextAttribute t -> defMessage & CF.stringValue .~ t + A.BoolAttribute b -> defMessage & CF.boolValue .~ b + A.DoubleAttribute d -> defMessage & CF.doubleValue .~ d + A.IntAttribute i -> defMessage & CF.intValue .~ i + attrToKeyValue :: (Text, A.Attribute) -> KeyValue + attrToKeyValue (k, v) = + defMessage + & CF.key .~ k + & CF.value + .~ ( case v of + A.AttributeValue a -> primToAnyValue a + A.AttributeArray a -> + defMessage + & CF.arrayValue + .~ (defMessage & CF.values .~ fmap primToAnyValue a) + ) + + +type Encoder = L.ByteString -> L.ByteString + + +httpLogsCompression :: OTLPExporterConfig -> ([(HeaderName, C.ByteString)], Encoder) +httpLogsCompression conf = + case otlpCompression conf of + Just GZip -> ([(hContentEncoding, "gzip")], compress) + _ -> ([], id) + + +httpLogsResponseTimeout :: OTLPExporterConfig -> ResponseTimeout +httpLogsResponseTimeout conf = case otlpTimeout conf of + Just timeoutMilli + | timeoutMilli == 0 -> responseTimeoutNone + | timeoutMilli >= 1 -> responseTimeoutMicro (timeoutMilli * 1_000) + _ -> responseTimeoutMicro (defaultExporterTimeout * 1_000) + + +httpLogsBaseHeaders :: OTLPExporterConfig -> Request -> RequestHeaders +httpLogsBaseHeaders conf req = + concat + [ [(hContentType, httpProtobufMimeType)] + , [(hAccept, httpProtobufMimeType)] + , fromMaybe [] (otlpHeaders conf) + , requestHeaders req + ] + + +logsEndpointUrl :: OTLPExporterConfig -> String +logsEndpointUrl conf = + case otlpEndpoint conf of + Nothing -> "http://localhost:4318/v1/logs" + Just e -> + if "/v1/" `isInfixOf` e + then e + else trimTrailingSlash e <> "/v1/logs" + + +trimTrailingSlash :: String -> String +trimTrailingSlash = reverse . dropWhile (== '/') . reverse diff --git a/src/OpenTelemetry/Exporter/OTLP/Metric.hs b/src/OpenTelemetry/Exporter/OTLP/Metric.hs new file mode 100644 --- /dev/null +++ b/src/OpenTelemetry/Exporter/OTLP/Metric.hs @@ -0,0 +1,478 @@ +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE NumericUnderscores #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} + +{- | OTLP HTTP\/Protobuf metrics exporter (@\/v1\/metrics@). + +See "OpenTelemetry.Exporter.OTLP.Span" for shared configuration ('OTLPExporterConfig'). +-} +module OpenTelemetry.Exporter.OTLP.Metric ( + otlpMetricExporter, + resourceMetricsToExportRequest, +) where + +import Codec.Compression.GZip (compress) +import Control.Applicative ((<|>)) +import Control.Concurrent (threadDelay) +import Control.Exception (SomeAsyncException (..), SomeException (..), fromException, throwIO, try) +import Control.Monad.IO.Class (MonadIO, liftIO) +import Data.Bits (shiftL) +import qualified Data.ByteString.Char8 as C +import qualified Data.ByteString.Lazy as L +import qualified Data.HashMap.Strict as H +import Data.List (isInfixOf) +import Data.Maybe (fromMaybe) +import Data.ProtoLens (defMessage, encodeMessage) +import Data.Text (Text) +import qualified Data.Text as T +import Data.Vector (Vector) +import qualified Data.Vector as V +import qualified Data.Vector.Generic as VG +import Lens.Micro ((&), (.~)) +import Network.HTTP.Client +import Network.HTTP.Simple (httpBS) +import Network.HTTP.Types.Header +import Network.HTTP.Types.Status +import OpenTelemetry.Attributes +import OpenTelemetry.Exporter.Metric ( + AggregationTemporality (..), + ExponentialHistogramDataPoint (..), + GaugeDataPoint (..), + HistogramDataPoint (..), + MetricExemplar (..), + MetricExport (..), + MetricExporter (..), + NumberValue (..), + ResourceMetricsExport (..), + ScopeMetricsExport (..), + SumDataPoint (..), + ) +import OpenTelemetry.Exporter.OTLP.Span (CompressionFormat (..), OTLPExporterConfig (..)) +import OpenTelemetry.Internal.Common.Types (ExportResult (..), FlushResult (..), InstrumentationLibrary (..), ShutdownResult (..)) +import OpenTelemetry.Resource (MaterializedResources, getMaterializedResourcesAttributes, getMaterializedResourcesSchema) +import Proto.Opentelemetry.Proto.Collector.Metrics.V1.MetricsService (ExportMetricsServiceRequest) +import qualified Proto.Opentelemetry.Proto.Collector.Metrics.V1.MetricsService_Fields as MSF +import Proto.Opentelemetry.Proto.Common.V1.Common (InstrumentationScope, KeyValue) +import qualified Proto.Opentelemetry.Proto.Common.V1.Common_Fields as Common_Fields +import qualified Proto.Opentelemetry.Proto.Metrics.V1.Metrics as PM +import qualified Proto.Opentelemetry.Proto.Metrics.V1.Metrics_Fields as Mf +import qualified Proto.Opentelemetry.Proto.Resource.V1.Resource as Res +import qualified Proto.Opentelemetry.Proto.Resource.V1.Resource_Fields as Rf +import Text.Read (readMaybe) + + +-- | Default OTLP timeout (milliseconds), aligned with "OpenTelemetry.Exporter.OTLP.Span". +defaultExporterTimeout :: Int +defaultExporterTimeout = 10_000 + + +httpHost :: OTLPExporterConfig -> String +httpHost conf = fromMaybe defaultHost (otlpEndpoint conf) + where + defaultHost = "http://localhost:4318" + + +httpProtobufMimeType :: C.ByteString +httpProtobufMimeType = "application/x-protobuf" + + +-- | Encode metric batches to an OTLP 'ExportMetricsServiceRequest'. +resourceMetricsToExportRequest :: Vector ResourceMetricsExport -> ExportMetricsServiceRequest +resourceMetricsToExportRequest rms = + defMessage + & MSF.vec'resourceMetrics + .~ V.map resourceMetricsExportToProto rms + + +-- | OTLP 'MetricExporter' using HTTP\/Protobuf (same transport as 'OpenTelemetry.Exporter.OTLP.Span.otlpExporter'). +otlpMetricExporter :: (MonadIO m) => OTLPExporterConfig -> m MetricExporter +otlpMetricExporter conf = liftIO $ do + req <- parseRequest (metricsEndpointUrl conf) + let (encodingHeaders, encoder) = httpMetricsCompression conf + let baseReq = + req + { method = "POST" + , requestHeaders = encodingHeaders <> httpMetricsBaseHeaders conf req + , responseTimeout = httpMetricsResponseTimeout conf + } + pure $ + MetricExporter + { metricExporterExport = \batches -> do + if not (anyMetricsToExport batches) + then pure Success + else do + result <- try $ exporterExportCall encoder baseReq batches + case result of + Left err -> case fromException err of + Just (SomeAsyncException _) -> throwIO err + Nothing -> pure $ Failure $ Just err + Right ok -> pure ok + , metricExporterShutdown = pure ShutdownSuccess + , metricExporterForceFlush = pure FlushSuccess + } + where + retryDelay = 100_000 + maxRetryCount = 5 + isRetryableStatusCode status_ = + status_ == status408 || status_ == status429 || (statusCode status_ >= 500 && statusCode status_ < 600) + isRetryableException = \case + ResponseTimeout -> True + ConnectionTimeout -> True + ConnectionFailure _ -> True + ConnectionClosed -> True + _ -> False + + exporterExportCall encoder baseReq batches = do + let msg = encodeMessage (resourceMetricsToExportRequest batches) + let req = + baseReq + { requestBody = + RequestBodyLBS $ encoder $ L.fromStrict msg + } + sendReq req 0 + sendReq req backoffCount = do + eResp <- try $ httpBS req + let exponentialBackoff = + if backoffCount == maxRetryCount + then pure $ Failure Nothing + else do + threadDelay (retryDelay `shiftL` backoffCount) + sendReq req (backoffCount + 1) + case eResp of + Left (HttpExceptionRequest _req' e) + | isRetryableException e -> exponentialBackoff + Left err -> pure $ Failure $ Just $ SomeException err + Right resp -> + if isRetryableStatusCode (responseStatus resp) + then case lookup hRetryAfter $ responseHeaders resp of + Nothing -> exponentialBackoff + Just retryAfter -> case readMaybe $ C.unpack retryAfter of + Nothing -> exponentialBackoff + Just seconds -> do + threadDelay (seconds * 1_000_000) + sendReq req (backoffCount + 1) + else + if statusCode (responseStatus resp) >= 300 + then pure $ Failure Nothing + else pure Success + + +type Encoder = L.ByteString -> L.ByteString + + +httpMetricsCompression :: OTLPExporterConfig -> ([(HeaderName, C.ByteString)], Encoder) +httpMetricsCompression conf = + case otlpMetricsCompression conf <|> otlpCompression conf of + Just GZip -> ([(hContentEncoding, "gzip")], compress) + _ -> ([], id) + + +httpMetricsResponseTimeout :: OTLPExporterConfig -> ResponseTimeout +httpMetricsResponseTimeout conf = case otlpMetricsTimeout conf <|> otlpTimeout conf of + Just timeoutMilli + | timeoutMilli == 0 -> responseTimeoutNone + | timeoutMilli >= 1 -> responseTimeoutMicro (timeoutMilli * 1_000) + _ -> responseTimeoutMicro (defaultExporterTimeout * 1_000) + + +httpMetricsBaseHeaders :: OTLPExporterConfig -> Request -> RequestHeaders +httpMetricsBaseHeaders conf req = + concat + [ [(hContentType, httpProtobufMimeType)] + , [(hAccept, httpProtobufMimeType)] + , fromMaybe [] (otlpHeaders conf) + , fromMaybe [] (otlpMetricsHeaders conf) + , requestHeaders req + ] + + +-- | Like traces: default @http:\/\/localhost:4318\/v1\/metrics@, or 'otlpMetricsEndpoint' if set (path appended when missing). +metricsEndpointUrl :: OTLPExporterConfig -> String +metricsEndpointUrl conf = + case otlpMetricsEndpoint conf of + Nothing -> httpHost conf <> "/v1/metrics" + Just e -> + if "/v1/" `isInfixOf` e + then e + else trimTrailingSlash e <> "/v1/metrics" + + +trimTrailingSlash :: String -> String +trimTrailingSlash = reverse . dropWhile (== '/') . reverse + + +anyMetricsToExport :: Vector ResourceMetricsExport -> Bool +anyMetricsToExport batches = + V.any (\rm -> V.any (not . V.null . scopeMetricsExports) (resourceMetricsScopes rm)) batches + + +resourceMetricsExportToProto :: ResourceMetricsExport -> PM.ResourceMetrics +resourceMetricsExportToProto ResourceMetricsExport {..} = + defMessage + & Mf.resource + .~ materializedResourceToProto resourceMetricsResource + & Mf.vec'scopeMetrics + .~ V.map scopeMetricsExportToProto resourceMetricsScopes + & Mf.schemaUrl + .~ maybe T.empty T.pack (getMaterializedResourcesSchema resourceMetricsResource) + + +materializedResourceToProto :: MaterializedResources -> Res.Resource +materializedResourceToProto r = + let attrs = getMaterializedResourcesAttributes r + in defMessage + & Rf.vec'attributes + .~ attributesToProto attrs + & Rf.droppedAttributesCount + .~ fromIntegral (getDropped attrs) + + +scopeMetricsExportToProto :: ScopeMetricsExport -> PM.ScopeMetrics +scopeMetricsExportToProto ScopeMetricsExport {..} = + defMessage + & Mf.scope + .~ instrumentationLibraryToProto scopeMetricsScope + & Mf.vec'metrics + .~ V.map metricExportToProto scopeMetricsExports + & Mf.schemaUrl + .~ librarySchemaUrl scopeMetricsScope + + +instrumentationLibraryToProto :: InstrumentationLibrary -> InstrumentationScope +instrumentationLibraryToProto InstrumentationLibrary {..} = + defMessage + & Common_Fields.name + .~ libraryName + & Common_Fields.version + .~ libraryVersion + & Common_Fields.vec'attributes + .~ attributesToProto libraryAttributes + & Common_Fields.droppedAttributesCount + .~ fromIntegral (getDropped libraryAttributes) + + +temporalityToProto :: AggregationTemporality -> PM.AggregationTemporality +temporalityToProto = \case + AggregationDelta -> PM.AGGREGATION_TEMPORALITY_DELTA + AggregationCumulative -> PM.AGGREGATION_TEMPORALITY_CUMULATIVE + + +metricExportToProto :: MetricExport -> PM.Metric +metricExportToProto = \case + MetricExportSum name desc unit_ _scope monotonic _isInt temp pts -> + defMessage + & Mf.name + .~ name + & Mf.description + .~ desc + & Mf.unit + .~ unit_ + & Mf.sum + .~ ( defMessage + & Mf.aggregationTemporality + .~ temporalityToProto temp + & Mf.isMonotonic + .~ monotonic + & Mf.vec'dataPoints + .~ V.map sumPointToProto pts + ) + MetricExportHistogram name desc unit_ _scope temp pts -> + defMessage + & Mf.name + .~ name + & Mf.description + .~ desc + & Mf.unit + .~ unit_ + & Mf.histogram + .~ ( defMessage + & Mf.aggregationTemporality + .~ temporalityToProto temp + & Mf.vec'dataPoints + .~ V.map histogramPointToProto pts + ) + MetricExportExponentialHistogram name desc unit_ _scope temp pts -> + defMessage + & Mf.name + .~ name + & Mf.description + .~ desc + & Mf.unit + .~ unit_ + & Mf.exponentialHistogram + .~ ( defMessage + & Mf.aggregationTemporality + .~ temporalityToProto temp + & Mf.vec'dataPoints + .~ V.map exponentialHistogramPointToProto pts + ) + MetricExportGauge name desc unit_ _scope _isInt pts -> + defMessage + & Mf.name + .~ name + & Mf.description + .~ desc + & Mf.unit + .~ unit_ + & Mf.gauge + .~ ( defMessage + & Mf.vec'dataPoints + .~ V.map gaugePointToProto pts + ) + + +sumPointToProto :: SumDataPoint -> PM.NumberDataPoint +sumPointToProto SumDataPoint {..} = + numberDataPointFromValue sumDataPointValue $ + defMessage + & Mf.vec'attributes + .~ attributesToProto sumDataPointAttributes + & Mf.startTimeUnixNano + .~ sumDataPointStartTimeUnixNano + & Mf.timeUnixNano + .~ sumDataPointTimeUnixNano + & Mf.vec'exemplars + .~ V.map metricExemplarToProto sumDataPointExemplars + + +gaugePointToProto :: GaugeDataPoint -> PM.NumberDataPoint +gaugePointToProto GaugeDataPoint {..} = + numberDataPointFromValue gaugeDataPointValue $ + defMessage + & Mf.vec'attributes + .~ attributesToProto gaugeDataPointAttributes + & Mf.startTimeUnixNano + .~ gaugeDataPointStartTimeUnixNano + & Mf.timeUnixNano + .~ gaugeDataPointTimeUnixNano + & Mf.vec'exemplars + .~ V.map metricExemplarToProto gaugeDataPointExemplars + + +numberDataPointFromValue :: NumberValue -> PM.NumberDataPoint -> PM.NumberDataPoint +numberDataPointFromValue val dp = case val of + IntNumber i -> dp & Mf.asInt .~ i + DoubleNumber d -> dp & Mf.asDouble .~ d + + +histogramPointToProto :: HistogramDataPoint -> PM.HistogramDataPoint +histogramPointToProto HistogramDataPoint {..} = + defMessage + & Mf.vec'attributes + .~ attributesToProto histogramDataPointAttributes + & Mf.startTimeUnixNano + .~ histogramDataPointStartTimeUnixNano + & Mf.timeUnixNano + .~ histogramDataPointTimeUnixNano + & Mf.count + .~ histogramDataPointCount + & Mf.maybe'sum + .~ Just histogramDataPointSum + & Mf.vec'bucketCounts + .~ VG.convert histogramDataPointBucketCounts + & Mf.vec'explicitBounds + .~ VG.convert histogramDataPointExplicitBounds + & Mf.maybe'min + .~ histogramDataPointMin + & Mf.maybe'max + .~ histogramDataPointMax + & Mf.vec'exemplars + .~ V.map metricExemplarToProto histogramDataPointExemplars + + +metricExemplarToProto :: MetricExemplar -> PM.Exemplar +metricExemplarToProto MetricExemplar {..} = + defMessage + & Mf.traceId + .~ metricExemplarTraceId + & Mf.spanId + .~ metricExemplarSpanId + & Mf.timeUnixNano + .~ metricExemplarTimeUnixNano + & Mf.vec'filteredAttributes + .~ attributesToProto metricExemplarFilteredAttributes + & Mf.maybe'value + .~ fmap + ( \case + IntNumber i -> PM.Exemplar'AsInt i + DoubleNumber d -> PM.Exemplar'AsDouble d + ) + metricExemplarValue + + +exponentialHistogramPointToProto :: ExponentialHistogramDataPoint -> PM.ExponentialHistogramDataPoint +exponentialHistogramPointToProto ExponentialHistogramDataPoint {..} = + defMessage + & Mf.vec'attributes + .~ attributesToProto exponentialHistogramDataPointAttributes + & Mf.startTimeUnixNano + .~ exponentialHistogramDataPointStartTimeUnixNano + & Mf.timeUnixNano + .~ exponentialHistogramDataPointTimeUnixNano + & Mf.count + .~ exponentialHistogramDataPointCount + & Mf.maybe'sum + .~ exponentialHistogramDataPointSum + & Mf.scale + .~ exponentialHistogramDataPointScale + & Mf.zeroCount + .~ exponentialHistogramDataPointZeroCount + & Mf.maybe'positive + .~ ( if V.null exponentialHistogramDataPointPositiveBucketCounts + then Nothing + else + Just $ + defMessage + & Mf.offset + .~ exponentialHistogramDataPointPositiveOffset + & Mf.vec'bucketCounts + .~ VG.convert exponentialHistogramDataPointPositiveBucketCounts + ) + & Mf.maybe'negative + .~ ( if V.null exponentialHistogramDataPointNegativeBucketCounts + then Nothing + else + Just $ + defMessage + & Mf.offset + .~ exponentialHistogramDataPointNegativeOffset + & Mf.vec'bucketCounts + .~ VG.convert exponentialHistogramDataPointNegativeBucketCounts + ) + & Mf.maybe'min + .~ exponentialHistogramDataPointMin + & Mf.maybe'max + .~ exponentialHistogramDataPointMax + & Mf.vec'exemplars + .~ V.map metricExemplarToProto exponentialHistogramDataPointExemplars + & Mf.zeroThreshold + .~ exponentialHistogramDataPointZeroThreshold + + +attributesToProto :: Attributes -> Vector KeyValue +attributesToProto = + V.fromList + . fmap attributeToKeyValue + . H.toList + . snd + . ((,) <$> getCount <*> getAttributeMap) + where + primAttributeToAnyValue = \case + TextAttribute t -> defMessage & Common_Fields.stringValue .~ t + BoolAttribute b -> defMessage & Common_Fields.boolValue .~ b + DoubleAttribute d -> defMessage & Common_Fields.doubleValue .~ d + IntAttribute i -> defMessage & Common_Fields.intValue .~ i + attributeToKeyValue :: (Text, Attribute) -> KeyValue + attributeToKeyValue (k, v) = + defMessage + & Common_Fields.key + .~ k + & Common_Fields.value + .~ ( case v of + AttributeValue a -> primAttributeToAnyValue a + AttributeArray a -> + defMessage + & Common_Fields.arrayValue + .~ (defMessage & Common_Fields.values .~ fmap primAttributeToAnyValue a) + ) diff --git a/src/OpenTelemetry/Exporter/OTLP/Span.hs b/src/OpenTelemetry/Exporter/OTLP/Span.hs --- a/src/OpenTelemetry/Exporter/OTLP/Span.hs +++ b/src/OpenTelemetry/Exporter/OTLP/Span.hs @@ -43,6 +43,9 @@ -- * Default local endpoints otlpExporterHttpEndpoint, otlpExporterGRpcEndpoint, + + -- * Protobuf conversion (testing) + immutableSpansToProtobuf, ) where import Codec.Compression.GZip @@ -50,13 +53,14 @@ import Control.Concurrent (threadDelay) import Control.Exception (SomeAsyncException (..), SomeException (..), fromException, throwIO, try) import Control.Monad.IO.Class -import Data.Bits (shiftL) +import Data.Bits (shiftL, (.|.)) import qualified Data.ByteString.Char8 as C import qualified Data.ByteString.Lazy as L import qualified Data.CaseInsensitive as CI import Data.Char (toLower) import Data.HashMap.Strict (HashMap) import qualified Data.HashMap.Strict as H +import Data.IORef (readIORef) import Data.Maybe import Data.ProtoLens.Encoding import Data.ProtoLens.Message @@ -65,16 +69,18 @@ import Data.Vector (Vector) import qualified Data.Vector as V import qualified Data.Vector as Vector +import qualified Data.Word import Lens.Micro import Network.HTTP.Client -import qualified Network.HTTP.Client as HTTPClient import Network.HTTP.Simple (httpBS) import Network.HTTP.Types.Header import Network.HTTP.Types.Status import OpenTelemetry.Attributes import qualified OpenTelemetry.Baggage as Baggage +import OpenTelemetry.Common (optionalTimestampToMaybe) import OpenTelemetry.Environment import OpenTelemetry.Exporter.Span +import OpenTelemetry.Internal.Common.Types (FlushResult (..), ShutdownResult (..)) import OpenTelemetry.Propagator.W3CTraceContext (encodeTraceStateFull) import OpenTelemetry.Resource import OpenTelemetry.Trace.Core (timestampNanoseconds) @@ -140,9 +146,9 @@ <*> lookupEnv "OTEL_EXPORTER_OTLP_CERTIFICATE" <*> lookupEnv "OTEL_EXPORTER_OTLP_TRACES_CERTIFICATE" <*> lookupEnv "OTEL_EXPORTER_OTLP_METRICS_CERTIFICATE" - <*> (fmap decodeHeaders <$> lookupEnv "OTEL_EXPORTER_OTLP_HEADERS") - <*> (fmap decodeHeaders <$> lookupEnv "OTEL_EXPORTER_OTLP_TRACES_HEADERS") - <*> (fmap decodeHeaders <$> lookupEnv "OTEL_EXPORTER_OTLP_METRICS_HEADERS") + <*> (traverse decodeHeaders =<< lookupEnv "OTEL_EXPORTER_OTLP_HEADERS") + <*> (traverse decodeHeaders =<< lookupEnv "OTEL_EXPORTER_OTLP_TRACES_HEADERS") + <*> (traverse decodeHeaders =<< lookupEnv "OTEL_EXPORTER_OTLP_METRICS_HEADERS") <*> (traverse readCompressionFormat =<< lookupEnv "OTEL_EXPORTER_OTLP_COMPRESSION") <*> (traverse readCompressionFormat =<< lookupEnv "OTEL_EXPORTER_OTLP_TRACES_COMPRESSION") <*> (traverse readCompressionFormat =<< lookupEnv "OTEL_EXPORTER_OTLP_METRICS_COMPRESSION") @@ -154,9 +160,11 @@ <*> (traverse readProtocol =<< lookupEnv "OTEL_EXPORTER_OTLP_METRICS_PROTOCOL") where decodeHeaders hsString = case Baggage.decodeBaggageHeader $ C.pack hsString of - Left _ -> mempty + Left err -> do + putWarningLn $ "Warning: failed to parse OTEL_EXPORTER_OTLP_HEADERS: " <> show err + pure mempty Right baggageFmt -> - (\(k, v) -> (CI.mk $ Baggage.tokenValue k, T.encodeUtf8 $ Baggage.value v)) <$> H.toList (Baggage.values baggageFmt) + pure $ (\(k, v) -> (CI.mk $ Baggage.tokenValue k, T.encodeUtf8 $ Baggage.value v)) <$> H.toList (Baggage.values baggageFmt) {- | @@ -288,7 +296,8 @@ pure $ Failure $ Just err Right ok -> pure ok else pure Success - , spanExporterShutdown = pure () + , spanExporterShutdown = pure ShutdownSuccess + , spanExporterForceFlush = pure FlushSuccess } where retryDelay = 100_000 -- 100ms @@ -322,17 +331,9 @@ sendReq req (backoffCount + 1) case eResp of - Left err@(HttpExceptionRequest req' e) - | HTTPClient.host req' == "localhost" - , HTTPClient.port req' == 4317 || HTTPClient.port req' == 4318 - , ConnectionFailure _someExn <- e -> - do - pure $ Failure Nothing - | otherwise -> - if isRetryableException e - then exponentialBackoff - else pure $ Failure $ Just $ SomeException err - Left err -> do + Left (HttpExceptionRequest _req' e) + | isRetryableException e -> exponentialBackoff + Left err -> pure $ Failure $ Just $ SomeException err Right resp -> if isRetryableStatusCode (responseStatus resp) @@ -347,9 +348,7 @@ sendReq req (backoffCount + 1) else if statusCode (responseStatus resp) >= 300 - then do - print resp - pure $ Failure Nothing + then pure $ Failure Nothing else pure Success @@ -373,9 +372,8 @@ Get the HTTP host from the `OTLPExporterConfig`. -} httpHost :: OTLPExporterConfig -> String -httpHost conf = fromMaybe defaultHost $ otlpEndpoint conf +httpHost conf = fromMaybe defaultHost $ otlpTracesEndpoint conf <|> otlpEndpoint conf where - -- TODO shouldn't this be http://localhost:4317 ? defaultHost = "http://localhost:4318" @@ -413,7 +411,7 @@ httpBaseHeaders conf req = concat [ [(hContentType, httpProtobufMimeType)] - , [(hAcceptEncoding, httpProtobufMimeType)] + , [(hAccept, httpProtobufMimeType)] , fromMaybe [] (otlpHeaders conf) , fromMaybe [] (otlpTracesHeaders conf) , requestHeaders req @@ -437,11 +435,11 @@ ( defMessage & Trace_Fields.resource .~ ( defMessage - & Trace_Fields.vec'attributes - .~ attributesToProto (getMaterializedResourcesAttributes someResourceGroup) - -- TODO - & Trace_Fields.droppedAttributesCount - .~ 0 + & Trace_Fields.vec'attributes + .~ attributesToProto (getMaterializedResourcesAttributes someResourceGroup) + -- TODO + & Trace_Fields.droppedAttributesCount + .~ 0 ) -- TODO, seems like spans need to be emitted via an API -- that lets us keep them grouped by instrumentation originator @@ -470,10 +468,10 @@ defMessage & Trace_Fields.scope .~ ( defMessage - & Trace_Fields.name - .~ OT.libraryName library - & Common_Fields.version - .~ OT.libraryVersion library + & Trace_Fields.name + .~ OT.libraryName library + & Common_Fields.version + .~ OT.libraryVersion library ) & Trace_Fields.vec'spans .~ spans_ @@ -487,6 +485,7 @@ -} makeSpan :: (MonadIO m) => OT.ImmutableSpan -> m Span makeSpan completedSpan = do + hot <- liftIO $ readIORef (OT.spanHot completedSpan) let startTime = timestampNanoseconds (OT.spanStart completedSpan) parentSpanF <- do case OT.spanParent completedSpan of @@ -503,48 +502,50 @@ .~ spanIdBytes (OT.spanId $ OT.spanContext completedSpan) & Trace_Fields.traceState .~ T.decodeUtf8 (encodeTraceStateFull $ OT.traceState $ OT.spanContext completedSpan) + & Trace_Fields.flags + .~ fromIntegral (OT.traceFlagsValue $ OT.traceFlags $ OT.spanContext completedSpan) & Trace_Fields.name - .~ OT.spanName completedSpan + .~ OT.hotName hot & Trace_Fields.kind .~ ( case OT.spanKind completedSpan of - OT.Server -> Span'SPAN_KIND_SERVER - OT.Client -> Span'SPAN_KIND_CLIENT - OT.Producer -> Span'SPAN_KIND_PRODUCER - OT.Consumer -> Span'SPAN_KIND_CONSUMER - OT.Internal -> Span'SPAN_KIND_INTERNAL + OT.Server -> Span'SPAN_KIND_SERVER + OT.Client -> Span'SPAN_KIND_CLIENT + OT.Producer -> Span'SPAN_KIND_PRODUCER + OT.Consumer -> Span'SPAN_KIND_CONSUMER + OT.Internal -> Span'SPAN_KIND_INTERNAL ) & Trace_Fields.startTimeUnixNano .~ startTime & Trace_Fields.endTimeUnixNano - .~ maybe startTime timestampNanoseconds (OT.spanEnd completedSpan) + .~ maybe startTime timestampNanoseconds (optionalTimestampToMaybe (OT.hotEnd hot)) & Trace_Fields.vec'attributes - .~ attributesToProto (OT.spanAttributes completedSpan) + .~ attributesToProto (OT.hotAttributes hot) & Trace_Fields.droppedAttributesCount - .~ fromIntegral (getCount $ OT.spanAttributes completedSpan) + .~ fromIntegral (getDropped $ OT.hotAttributes hot) & Trace_Fields.vec'events - .~ fmap makeEvent (appendOnlyBoundedCollectionValues $ OT.spanEvents completedSpan) + .~ fmap makeEvent (appendOnlyBoundedCollectionValues $ OT.hotEvents hot) & Trace_Fields.droppedEventsCount - .~ fromIntegral (appendOnlyBoundedCollectionDroppedElementCount (OT.spanEvents completedSpan)) + .~ fromIntegral (appendOnlyBoundedCollectionDroppedElementCount (OT.hotEvents hot)) & Trace_Fields.vec'links - .~ fmap makeLink (appendOnlyBoundedCollectionValues $ OT.spanLinks completedSpan) + .~ fmap makeLink (appendOnlyBoundedCollectionValues $ OT.hotLinks hot) & Trace_Fields.droppedLinksCount - .~ fromIntegral (appendOnlyBoundedCollectionDroppedElementCount (OT.spanLinks completedSpan)) + .~ fromIntegral (appendOnlyBoundedCollectionDroppedElementCount (OT.hotLinks hot)) & Trace_Fields.status - .~ ( case OT.spanStatus completedSpan of - OT.Unset -> - defMessage - & Trace_Fields.code - .~ Status'STATUS_CODE_UNSET - OT.Ok -> - defMessage - & Trace_Fields.code - .~ Status'STATUS_CODE_OK - (OT.Error e) -> - defMessage - & Trace_Fields.code - .~ Status'STATUS_CODE_ERROR - & Trace_Fields.message - .~ e + .~ ( case OT.hotStatus hot of + OT.Unset -> + defMessage + & Trace_Fields.code + .~ Status'STATUS_CODE_UNSET + OT.Ok -> + defMessage + & Trace_Fields.code + .~ Status'STATUS_CODE_OK + (OT.Error e) -> + defMessage + & Trace_Fields.code + .~ Status'STATUS_CODE_ERROR + & Trace_Fields.message + .~ e ) & parentSpanF @@ -563,7 +564,7 @@ & Trace_Fields.vec'attributes .~ attributesToProto (OT.eventAttributes e) & Trace_Fields.droppedAttributesCount - .~ fromIntegral (getCount $ OT.eventAttributes e) + .~ fromIntegral (getDropped $ OT.eventAttributes e) {- | @@ -574,21 +575,35 @@ makeLink l = defMessage & Trace_Fields.traceId - .~ traceIdBytes (OT.traceId $ OT.frozenLinkContext l) + .~ traceIdBytes (OT.traceId ctx) & Trace_Fields.spanId - .~ spanIdBytes (OT.spanId $ OT.frozenLinkContext l) + .~ spanIdBytes (OT.spanId ctx) & Trace_Fields.traceState - .~ T.decodeUtf8 (encodeTraceStateFull $ OT.traceState $ OT.frozenLinkContext l) + .~ T.decodeUtf8 (encodeTraceStateFull $ OT.traceState ctx) + & Trace_Fields.flags + .~ linkFlags ctx & Trace_Fields.vec'attributes .~ attributesToProto (OT.frozenLinkAttributes l) & Trace_Fields.droppedAttributesCount - .~ fromIntegral (getCount $ OT.frozenLinkAttributes l) + .~ fromIntegral (getDropped $ OT.frozenLinkAttributes l) + where + ctx = OT.frozenLinkContext l {- | Internal helper. Translate a collection of `Attributes` to a vector of OTLP `KeyValue` pairs. -} + +-- | OTLP link flags: bits 0-7 = W3C trace flags, bit 8 = HAS_IS_REMOTE, bit 9 = IS_REMOTE. +linkFlags :: OT.SpanContext -> Data.Word.Word32 +linkFlags ctx = + let w8flags = fromIntegral (OT.traceFlagsValue $ OT.traceFlags ctx) :: Data.Word.Word32 + hasIsRemote = 1 `shiftL` 8 + isRemoteBit = if OT.isRemote ctx then 1 `shiftL` 9 else 0 + in w8flags .|. hasIsRemote .|. isRemoteBit + + attributesToProto :: Attributes -> Vector KeyValue attributesToProto = V.fromList @@ -609,9 +624,9 @@ .~ k & Common_Fields.value .~ ( case v of - AttributeValue a -> primAttributeToAnyValue a - AttributeArray a -> - defMessage - & Common_Fields.arrayValue - .~ (defMessage & Common_Fields.values .~ fmap primAttributeToAnyValue a) + AttributeValue a -> primAttributeToAnyValue a + AttributeArray a -> + defMessage + & Common_Fields.arrayValue + .~ (defMessage & Common_Fields.values .~ fmap primAttributeToAnyValue a) ) diff --git a/test/Spec.hs b/test/Spec.hs new file mode 100644 --- /dev/null +++ b/test/Spec.hs @@ -0,0 +1,346 @@ +{-# LANGUAGE OverloadedStrings #-} + +import Data.Int (Int64) +import Data.ProtoLens (decodeMessage, encodeMessage) +import Data.Text (Text) +import qualified Data.Vector as V +import qualified Data.Vector.Unboxed as U +import Data.Word (Word64) +import Lens.Micro ((^.)) +import OpenTelemetry.Attributes (emptyAttributes) +import OpenTelemetry.Exporter.Metric ( + AggregationTemporality (..), + ExponentialHistogramDataPoint (..), + GaugeDataPoint (..), + HistogramDataPoint (..), + MetricExport (..), + NumberValue (..), + ResourceMetricsExport (..), + ScopeMetricsExport (..), + SumDataPoint (..), + ) +import OpenTelemetry.Exporter.OTLP.Metric (resourceMetricsToExportRequest) +import OpenTelemetry.Internal.Common.Types (InstrumentationLibrary (..)) +import OpenTelemetry.Resource (emptyMaterializedResources) +import Proto.Opentelemetry.Proto.Collector.Metrics.V1.MetricsService (ExportMetricsServiceRequest) +import qualified Proto.Opentelemetry.Proto.Collector.Metrics.V1.MetricsService_Fields as MSF +import qualified Proto.Opentelemetry.Proto.Metrics.V1.Metrics as PM +import qualified Proto.Opentelemetry.Proto.Metrics.V1.Metrics_Fields as Mf +import Test.Hspec + + +decodeExport :: [ResourceMetricsExport] -> Either String ExportMetricsServiceRequest +decodeExport rms = + decodeMessage (encodeMessage (resourceMetricsToExportRequest (V.fromList rms))) + + +main :: IO () +main = hspec $ do + describe "resourceMetricsToExportRequest" $ do + it "round-trips SumDataPoint through protobuf decode" $ do + let lib = "lib" :: InstrumentationLibrary + pt = + SumDataPoint + { sumDataPointStartTimeUnixNano = 0 + , sumDataPointTimeUnixNano = 1 + , sumDataPointValue = DoubleNumber 3 + , sumDataPointAttributes = emptyAttributes + , sumDataPointExemplars = V.empty + } + exp = + MetricExportSum "c" "d" "By" lib True False AggregationCumulative $ + V.singleton pt + rm = + ResourceMetricsExport emptyMaterializedResources $ + V.singleton $ + ScopeMetricsExport lib (V.singleton exp) + decoded = decodeExport [rm] + case decoded of + Left e -> expectationFailure e + Right r -> do + (r ^. MSF.vec'resourceMetrics) `shouldNotSatisfy` V.null + case firstNumberDataPoint r of + Nothing -> expectationFailure "expected a NumberDataPoint" + Just ndp -> do + ndp ^. Mf.startTimeUnixNano `shouldBe` 0 + ndp ^. Mf.timeUnixNano `shouldBe` 1 + ndp ^. Mf.asDouble `shouldBe` 3 + + it "round-trips HistogramDataPoint with non-empty buckets" $ do + let lib = "lib" :: InstrumentationLibrary + bounds = V.fromList [0, 5, 10] :: V.Vector Double + counts = V.fromList [1, 2, 3, 4] :: V.Vector Word64 + hdp = + HistogramDataPoint + 100 + 200 + 10 + 99.5 + counts + bounds + emptyAttributes + (Just 0.25) + (Just 98.0) + V.empty + exp = + MetricExportHistogram "hist" "desc" "1" lib AggregationCumulative $ + V.singleton hdp + rm = + ResourceMetricsExport emptyMaterializedResources $ + V.singleton $ + ScopeMetricsExport lib (V.singleton exp) + case decodeExport [rm] of + Left e -> expectationFailure e + Right r -> case firstHistogramDataPoint r of + Nothing -> expectationFailure "expected a HistogramDataPoint" + Just dp -> do + dp ^. Mf.startTimeUnixNano `shouldBe` 100 + dp ^. Mf.timeUnixNano `shouldBe` 200 + dp ^. Mf.count `shouldBe` 10 + dp ^. Mf.maybe'sum `shouldBe` Just 99.5 + U.toList (dp ^. Mf.vec'bucketCounts) `shouldBe` [1, 2, 3, 4] + U.toList (dp ^. Mf.vec'explicitBounds) `shouldBe` [0, 5, 10] + dp ^. Mf.maybe'min `shouldBe` Just 0.25 + dp ^. Mf.maybe'max `shouldBe` Just 98.0 + + it "round-trips GaugeDataPoint (Int)" $ do + let lib = "lib" :: InstrumentationLibrary + gdp = + GaugeDataPoint + { gaugeDataPointStartTimeUnixNano = 5 + , gaugeDataPointTimeUnixNano = 6 + , gaugeDataPointValue = IntNumber 42 + , gaugeDataPointAttributes = emptyAttributes + , gaugeDataPointExemplars = V.empty + } + exp = MetricExportGauge "g" "" "" lib True $ V.singleton gdp + rm = + ResourceMetricsExport emptyMaterializedResources $ + V.singleton $ + ScopeMetricsExport lib (V.singleton exp) + case decodeExport [rm] of + Left e -> expectationFailure e + Right r -> case firstNumberDataPoint r of + Nothing -> expectationFailure "expected a NumberDataPoint" + Just ndp -> do + ndp ^. Mf.startTimeUnixNano `shouldBe` 5 + ndp ^. Mf.timeUnixNano `shouldBe` 6 + ndp ^. Mf.asInt `shouldBe` 42 + + it "round-trips GaugeDataPoint (Double)" $ do + let lib = "lib" :: InstrumentationLibrary + gdp = + GaugeDataPoint + { gaugeDataPointStartTimeUnixNano = 7 + , gaugeDataPointTimeUnixNano = 8 + , gaugeDataPointValue = DoubleNumber 2.718 + , gaugeDataPointAttributes = emptyAttributes + , gaugeDataPointExemplars = V.empty + } + exp = MetricExportGauge "g2" "" "" lib False $ V.singleton gdp + rm = + ResourceMetricsExport emptyMaterializedResources $ + V.singleton $ + ScopeMetricsExport lib (V.singleton exp) + case decodeExport [rm] of + Left e -> expectationFailure e + Right r -> case firstNumberDataPoint r of + Nothing -> expectationFailure "expected a NumberDataPoint" + Just ndp -> do + ndp ^. Mf.startTimeUnixNano `shouldBe` 7 + ndp ^. Mf.timeUnixNano `shouldBe` 8 + ndp ^. Mf.asDouble `shouldBe` 2.718 + + it "round-trips ExponentialHistogramDataPoint" $ do + let lib = "lib" :: InstrumentationLibrary + edp = + ExponentialHistogramDataPoint + 1000 + 2000 + 50 + (Just 123.4) + 3 + 7 + 2 + (V.fromList [5, 6]) + (-1) + (V.fromList [1, 2]) + emptyAttributes + (Just 0.1) + (Just 99.9) + V.empty + 0.5 + exp = + MetricExportExponentialHistogram "eh" "" "" lib AggregationDelta $ + V.singleton edp + rm = + ResourceMetricsExport emptyMaterializedResources $ + V.singleton $ + ScopeMetricsExport lib (V.singleton exp) + case decodeExport [rm] of + Left e -> expectationFailure e + Right r -> case firstExponentialHistogramDataPoint r of + Nothing -> expectationFailure "expected an ExponentialHistogramDataPoint" + Just dp -> do + dp ^. Mf.startTimeUnixNano `shouldBe` 1000 + dp ^. Mf.timeUnixNano `shouldBe` 2000 + dp ^. Mf.count `shouldBe` 50 + dp ^. Mf.maybe'sum `shouldBe` Just 123.4 + dp ^. Mf.scale `shouldBe` 3 + dp ^. Mf.zeroCount `shouldBe` 7 + dp ^. Mf.zeroThreshold `shouldBe` 0.5 + dp ^. Mf.maybe'min `shouldBe` Just 0.1 + dp ^. Mf.maybe'max `shouldBe` Just 99.9 + case dp ^. Mf.maybe'positive of + Nothing -> expectationFailure "expected positive buckets" + Just pos -> do + pos ^. Mf.offset `shouldBe` 2 + U.toList (pos ^. Mf.vec'bucketCounts) `shouldBe` [5, 6] + case dp ^. Mf.maybe'negative of + Nothing -> expectationFailure "expected negative buckets" + Just neg -> do + neg ^. Mf.offset `shouldBe` (-1) + U.toList (neg ^. Mf.vec'bucketCounts) `shouldBe` [1, 2] + + it "serializes AggregationDelta on Sum" $ do + let lib = "lib" :: InstrumentationLibrary + pt = + SumDataPoint + { sumDataPointStartTimeUnixNano = 0 + , sumDataPointTimeUnixNano = 0 + , sumDataPointValue = IntNumber 0 + , sumDataPointAttributes = emptyAttributes + , sumDataPointExemplars = V.empty + } + exp = + MetricExportSum "s" "" "" lib False False AggregationDelta $ + V.singleton pt + rm = + ResourceMetricsExport emptyMaterializedResources $ + V.singleton $ + ScopeMetricsExport lib (V.singleton exp) + case decodeExport [rm] of + Left e -> expectationFailure e + Right r -> case firstSum r of + Nothing -> expectationFailure "expected Sum" + Just s -> + s ^. Mf.aggregationTemporality `shouldBe` PM.AGGREGATION_TEMPORALITY_DELTA + + it "serializes AggregationCumulative on Histogram" $ do + let lib = "lib" :: InstrumentationLibrary + hdp = + HistogramDataPoint + 0 + 0 + 0 + 0 + V.empty + V.empty + emptyAttributes + Nothing + Nothing + V.empty + exp = + MetricExportHistogram "h" "" "" lib AggregationCumulative $ + V.singleton hdp + rm = + ResourceMetricsExport emptyMaterializedResources $ + V.singleton $ + ScopeMetricsExport lib (V.singleton exp) + case decodeExport [rm] of + Left e -> expectationFailure e + Right r -> case firstHistogram r of + Nothing -> expectationFailure "expected Histogram" + Just h -> + h ^. Mf.aggregationTemporality `shouldBe` PM.AGGREGATION_TEMPORALITY_CUMULATIVE + + it "produces valid empty protobuf for an empty batch" $ do + case decodeExport [] of + Left e -> expectationFailure e + Right r -> r ^. MSF.vec'resourceMetrics `shouldSatisfy` V.null + + it "encodes multiple ResourceMetricsExport batches" $ do + let lib = "lib" :: InstrumentationLibrary + mkRm name = + ResourceMetricsExport emptyMaterializedResources $ + V.singleton $ + ScopeMetricsExport lib $ + V.singleton $ + MetricExportSum name "" "" lib True False AggregationCumulative $ + V.singleton + SumDataPoint + { sumDataPointStartTimeUnixNano = 0 + , sumDataPointTimeUnixNano = 0 + , sumDataPointValue = DoubleNumber 0 + , sumDataPointAttributes = emptyAttributes + , sumDataPointExemplars = V.empty + } + rmA = mkRm "metric-a" + rmB = mkRm "metric-b" + case decodeExport [rmA, rmB] of + Left e -> expectationFailure e + Right r -> do + let v = r ^. MSF.vec'resourceMetrics + V.length v `shouldBe` 2 + metricNameAt v 0 `shouldBe` Just "metric-a" + metricNameAt v 1 `shouldBe` Just "metric-b" + + +firstMetric :: ExportMetricsServiceRequest -> Maybe PM.Metric +firstMetric r = do + rm <- (r ^. MSF.vec'resourceMetrics) V.!? 0 + sm <- (rm ^. Mf.vec'scopeMetrics) V.!? 0 + (sm ^. Mf.vec'metrics) V.!? 0 + + +metricNameAt :: V.Vector PM.ResourceMetrics -> Int -> Maybe Text +metricNameAt v i = do + rm <- v V.!? i + sm <- (rm ^. Mf.vec'scopeMetrics) V.!? 0 + m <- (sm ^. Mf.vec'metrics) V.!? 0 + pure (m ^. Mf.name) + + +firstNumberDataPoint :: ExportMetricsServiceRequest -> Maybe PM.NumberDataPoint +firstNumberDataPoint r = do + m <- firstMetric r + case m ^. Mf.maybe'data' of + Just (PM.Metric'Sum s) -> (s ^. Mf.vec'dataPoints) V.!? 0 + Just (PM.Metric'Gauge g) -> (g ^. Mf.vec'dataPoints) V.!? 0 + _ -> Nothing + + +firstSum :: ExportMetricsServiceRequest -> Maybe PM.Sum +firstSum r = do + m <- firstMetric r + case m ^. Mf.maybe'data' of + Just (PM.Metric'Sum s) -> Just s + _ -> Nothing + + +firstHistogram :: ExportMetricsServiceRequest -> Maybe PM.Histogram +firstHistogram r = do + m <- firstMetric r + case m ^. Mf.maybe'data' of + Just (PM.Metric'Histogram h) -> Just h + _ -> Nothing + + +firstHistogramDataPoint :: ExportMetricsServiceRequest -> Maybe PM.HistogramDataPoint +firstHistogramDataPoint r = do + h <- firstHistogram r + (h ^. Mf.vec'dataPoints) V.!? 0 + + +firstExponentialHistogram :: ExportMetricsServiceRequest -> Maybe PM.ExponentialHistogram +firstExponentialHistogram r = do + m <- firstMetric r + case m ^. Mf.maybe'data' of + Just (PM.Metric'ExponentialHistogram eh) -> Just eh + _ -> Nothing + + +firstExponentialHistogramDataPoint :: ExportMetricsServiceRequest -> Maybe PM.ExponentialHistogramDataPoint +firstExponentialHistogramDataPoint r = do + eh <- firstExponentialHistogram r + (eh ^. Mf.vec'dataPoints) V.!? 0