hs-opentelemetry-exporter-otlp 0.0.1.5 → 0.1.0.0
raw patch · 6 files changed
+505/−488 lines, 6 filesdep −clockdep −hs-opentelemetry-exporter-otlpdep −mtldep ~hs-opentelemetry-apidep ~hs-opentelemetry-otlpnew-uploader
Dependencies removed: clock, hs-opentelemetry-exporter-otlp, mtl, vector-builder
Dependency ranges changed: hs-opentelemetry-api, hs-opentelemetry-otlp
Files
- ChangeLog.md +2/−3
- hs-opentelemetry-exporter-otlp.cabal +18/−48
- src/OpenTelemetry/Exporter/OTLP.hs +4/−434
- src/OpenTelemetry/Exporter/OTLP/LogRecord.hs +2/−0
- src/OpenTelemetry/Exporter/OTLP/Span.hs +479/−0
- test/Spec.hs +0/−3
ChangeLog.md view
@@ -1,12 +1,11 @@ # Changelog for hs-opentelemetry-exporter-otlp -## 0.0.2.0+## 0.1.0.0 - Export dropped span, link, event, and attribute counts-- Upgrade to account for interfaces changed in hs-opentelemetry-api-0.0.2.0 - Add gzip compression support+- Swallow errors from sending data to localhost ## 0.0.1.0 - Initial release-## Unreleased changes
hs-opentelemetry-exporter-otlp.cabal view
@@ -1,22 +1,22 @@ cabal-version: 1.12 --- This file has been generated from package.yaml by hpack version 0.35.2.+-- This file has been generated from package.yaml by hpack version 0.36.0. -- -- see: https://github.com/sol/hpack -name: hs-opentelemetry-exporter-otlp-version: 0.0.1.5-synopsis: OpenTelemetry exporter supporting the standard OTLP protocol-description: Please see the README on GitHub at <https://github.com/iand675/hs-opentelemetry/tree/main/exporters/otlp#readme>-category: OpenTelemetry, Telemetry, Monitoring, Observability, Metrics-homepage: https://github.com/iand675/hs-opentelemetry#readme-bug-reports: https://github.com/iand675/hs-opentelemetry/issues-author: Ian Duncan-maintainer: ian@iankduncan.com-copyright: 2021 Ian Duncan-license: BSD3-license-file: LICENSE-build-type: Simple+name: hs-opentelemetry-exporter-otlp+version: 0.1.0.0+synopsis: OpenTelemetry exporter supporting the standard OTLP protocol+description: Please see the README on GitHub at <https://github.com/iand675/hs-opentelemetry/tree/main/exporters/otlp#readme>+category: OpenTelemetry, Telemetry, Monitoring, Observability, Metrics+homepage: https://github.com/iand675/hs-opentelemetry#readme+bug-reports: https://github.com/iand675/hs-opentelemetry/issues+author: Ian Duncan+maintainer: ian@iankduncan.com+copyright: 2021 Ian Duncan+license: BSD3+license-file: LICENSE+build-type: Simple extra-source-files: README.md ChangeLog.md@@ -28,6 +28,8 @@ library exposed-modules: OpenTelemetry.Exporter.OTLP+ OpenTelemetry.Exporter.OTLP.LogRecord+ OpenTelemetry.Exporter.OTLP.Span other-modules: Paths_hs_opentelemetry_exporter_otlp hs-source-dirs:@@ -37,47 +39,15 @@ base >=4.7 && <5 , bytestring , case-insensitive- , clock- , hs-opentelemetry-api >=0.0.3 && <0.2- , hs-opentelemetry-otlp ==0.0.1.*+ , hs-opentelemetry-api ==0.2.*+ , hs-opentelemetry-otlp ==0.1.* , http-client , http-conduit , http-types , microlens- , mtl , proto-lens >=0.7.1.0 , text , unordered-containers , vector- , vector-builder- , zlib- default-language: Haskell2010--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- , case-insensitive- , clock- , hs-opentelemetry-api >=0.0.3 && <0.2- , hs-opentelemetry-exporter-otlp- , hs-opentelemetry-otlp ==0.0.1.*- , http-client- , http-conduit- , http-types- , microlens- , mtl- , proto-lens >=0.7.1.0- , text- , unordered-containers- , vector- , vector-builder , zlib default-language: Haskell2010
src/OpenTelemetry/Exporter/OTLP.hs view
@@ -1,437 +1,7 @@-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE LambdaCase #-}-{-# LANGUAGE NumericUnderscores #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE RecordWildCards #-}----------------------------------------------------------------------------------------------------------------------------------------------------------------{- |- Module : OpenTelemetry.Exporter.OTLP- Copyright : (c) Ian Duncan, 2021- License : BSD-3- Description : OTLP Exporter- Maintainer : Ian Duncan- Stability : experimental- Portability : non-portable (GHC extensions)-- The OTLP Exporter is the recommend exporter format to use where possible.-- A number of vendors offer support for exporting traces, logs, and metrics using the vendor-agnostic OTLP protocol.-- Additionally, the OTLP format is supported by the <https://opentelemetry.io/docs/collector/ OpenTelemetry Collector>.-- The OpenTelemetry Collector offers a vendor-agnostic implementation of how to receive, process and export telemetry data.- It removes the need to run, operate, and maintain multiple agents/collectors.- This works with improved scalability and supports open-source observability data formats (e.g. Jaeger, Prometheus, Fluent Bit, etc.) sending to- one or more open-source or commercial back-ends. The local Collector agent is the default location to which instrumentation libraries export- their telemetry data.--}-module OpenTelemetry.Exporter.OTLP (- -- * Initializing the exporter- otlpExporter,-- -- * Configuring the exporter- OTLPExporterConfig (..),- CompressionFormat (..),- Protocol (..),- loadExporterEnvironmentVariables,-- -- * Default local endpoints- otlpExporterHttpEndpoint,- otlpExporterGRpcEndpoint,+module OpenTelemetry.Exporter.OTLP+ {-# DEPRECATED "use OpenTelemetry.Exporter.OTLP.Span instead" #-} (+ module OpenTelemetry.Exporter.OTLP.Span, ) where -import Codec.Compression.GZip-import Control.Applicative ((<|>))-import Control.Concurrent (threadDelay)-import Control.Exception (SomeAsyncException (..), SomeException (..), fromException, throwIO, try)-import Control.Monad.IO.Class-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.HashMap.Strict (HashMap)-import qualified Data.HashMap.Strict as H-import Data.Maybe-import Data.ProtoLens.Encoding-import Data.ProtoLens.Message-import Data.Text (Text)-import qualified Data.Text.Encoding as T-import Data.Vector (Vector)-import qualified Data.Vector as V-import qualified Data.Vector as Vector-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 qualified OpenTelemetry.Baggage as Baggage-import OpenTelemetry.Exporter-import OpenTelemetry.Resource-import OpenTelemetry.Trace.Core (timestampNanoseconds)-import qualified OpenTelemetry.Trace.Core as OT-import OpenTelemetry.Trace.Id (spanIdBytes, traceIdBytes)-import OpenTelemetry.Util-import Proto.Opentelemetry.Proto.Collector.Trace.V1.TraceService (ExportTraceServiceRequest)-import Proto.Opentelemetry.Proto.Common.V1.Common-import Proto.Opentelemetry.Proto.Common.V1.Common_Fields-import Proto.Opentelemetry.Proto.Trace.V1.Trace (InstrumentationLibrarySpans, Span, Span'Event, Span'Link, Span'SpanKind (Span'SPAN_KIND_CLIENT, Span'SPAN_KIND_CONSUMER, Span'SPAN_KIND_INTERNAL, Span'SPAN_KIND_PRODUCER, Span'SPAN_KIND_SERVER), Status'StatusCode (Status'STATUS_CODE_ERROR, Status'STATUS_CODE_OK, Status'STATUS_CODE_UNSET))-import Proto.Opentelemetry.Proto.Trace.V1.Trace_Fields-import System.Environment-import Text.Read (readMaybe)---data CompressionFormat = None | GZip---data Protocol {- GRpc | HttpJson | -}- = -- | Note: grpc and http/json will likely be supported eventually,- -- but not yet.- HttpProtobuf---otlpExporterHttpEndpoint :: C.ByteString-otlpExporterHttpEndpoint = "http://localhost:4318"---otlpExporterGRpcEndpoint :: C.ByteString-otlpExporterGRpcEndpoint = "http://localhost:4317"---data OTLPExporterConfig = OTLPExporterConfig- { otlpEndpoint :: Maybe String- , otlpTracesEndpoint :: Maybe String- , otlpMetricsEndpoint :: Maybe String- , otlpInsecure :: Maybe Bool- , otlpSpanInsecure :: Maybe Bool- , otlpMetricInsecure :: Maybe Bool- , otlpCertificate :: Maybe FilePath- , otlpTracesCertificate :: Maybe FilePath- , otlpMetricCertificate :: Maybe FilePath- , otlpHeaders :: Maybe [Header]- , otlpTracesHeaders :: Maybe [Header]- , otlpMetricsHeaders :: Maybe [Header]- , otlpCompression :: Maybe CompressionFormat- , otlpTracesCompression :: Maybe CompressionFormat- , otlpMetricsCompression :: Maybe CompressionFormat- , otlpTimeout :: Maybe Int- -- ^ Measured in seconds- , otlpTracesTimeout :: Maybe Int- , otlpMetricsTimeout :: Maybe Int- , otlpProtocol :: Maybe Protocol- , otlpTracesProtocol :: Maybe Protocol- , otlpMetricsProtocol :: Maybe Protocol- }---loadExporterEnvironmentVariables :: (MonadIO m) => m OTLPExporterConfig-loadExporterEnvironmentVariables = liftIO $ do- OTLPExporterConfig- <$> lookupEnv "OTEL_EXPORTER_OTLP_ENDPOINT"- <*> lookupEnv "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT"- <*> lookupEnv "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT"- <*> (fmap (== "true") <$> lookupEnv "OTEL_EXPORTER_OTLP_INSECURE")- <*> (fmap (== "true") <$> lookupEnv "OTEL_EXPORTER_OTLP_SPAN_INSECURE")- <*> (fmap (== "true") <$> lookupEnv "OTEL_EXPORTER_OTLP_METRIC_INSECURE")- <*> lookupEnv "OTEL_EXPORTER_OTLP_CERTIFICATE"- <*> lookupEnv "OTEL_EXPORTER_OTLP_TRACES_CERTIFICATE"- <*> lookupEnv "OTEL_EXPORTER_OTLP_METRICS_CERTIFICATE"- <*> (fmap decodeHeaders <$> lookupEnv "OTEL_EXPORTER_OTLP_HEADERS")- <*>- -- TODO lookupEnv "OTEL_EXPORTER_OTLP_TRACES_HEADERS" <*>- pure Nothing- <*>- -- TODO lookupEnv "OTEL_EXPORTER_OTLP_METRICS_HEADERS" <*>- pure Nothing- <*>- -- TODO lookupEnv <*>- ( fmap- ( \case- "gzip" -> GZip- "none" -> None- -- TODO- _ -> None- )- <$> lookupEnv "OTEL_EXPORTER_OTLP_COMPRESSION"- )- <*>- -- TODO lookupEnv "OTEL_EXPORTER_OTLP_TRACES_COMPRESSION" <*>- pure Nothing- <*>- -- TODO lookupEnv "OTEL_EXPORTER_OTLP_METRICS_COMPRESSION" <*>- pure Nothing- <*>- -- TODO lookupEnv "OTEL_EXPORTER_OTLP_TIMEOUT" <*>- pure Nothing- <*>- -- TODO lookupEnv "OTEL_EXPORTER_OTLP_TRACES_TIMEOUT" <*>- pure Nothing- <*>- -- TODO lookupEnv "OTEL_EXPORTER_OTLP_METRICS_TIMEOUT" <*>- pure Nothing- <*>- -- TODO lookupEnv "OTEL_EXPORTER_OTLP_PROTOCOL" <*>- pure Nothing- <*>- -- TODO lookupEnv "OTEL_EXPORTER_OTLP_TRACES_PROTOCOL" <*>- pure Nothing- <*>- -- TODO lookupEnv "OTEL_EXPORTER_OTLP_METRICS_PROTOCOL"- pure Nothing- where- decodeHeaders hsString = case Baggage.decodeBaggageHeader $ C.pack hsString of- Left _ -> mempty- Right baggageFmt ->- (\(k, v) -> (CI.mk $ Baggage.tokenValue k, T.encodeUtf8 $ Baggage.value v)) <$> H.toList (Baggage.values baggageFmt)---protobufMimeType :: C.ByteString-protobufMimeType = "application/x-protobuf"----- | Initial the OTLP 'Exporter'-otlpExporter :: (MonadIO m) => OTLPExporterConfig -> m (Exporter OT.ImmutableSpan)-otlpExporter conf = do- -- TODO, url parsing is janky- -- TODO configurable retryDelay, maximum retry counts- req <- liftIO $ parseRequest (maybe "http://localhost:4318/v1/traces" (<> "/v1/traces") (otlpEndpoint conf))-- let (encodingHeader, encoder) =- maybe- (id, id)- ( \case- None -> (id, id)- GZip -> (((hContentEncoding, "gzip") :), compress)- )- (otlpTracesCompression conf <|> otlpCompression conf)-- baseReqHeaders =- encodingHeader $- (hContentType, protobufMimeType)- : (hAcceptEncoding, protobufMimeType)- : fromMaybe [] (otlpHeaders conf)- ++ fromMaybe [] (otlpTracesHeaders conf)- ++ requestHeaders req- baseReq =- req- { method = "POST"- , requestHeaders = baseReqHeaders- }- pure $- Exporter- { exporterExport = \spans_ -> do- let anySpansToExport = H.size spans_ /= 0 && not (all V.null $ H.elems spans_)- if anySpansToExport- then do- result <- try $ exporterExportCall encoder baseReq spans_- case result of- Left err -> do- -- If the exception is async, then we need to rethrow it- -- here. Otherwise, there's a good chance that the- -- calling code will swallow the exception and cause- -- a problem.- case fromException err of- Just (SomeAsyncException _) ->- throwIO err- Nothing ->- pure $ Failure $ Just err- Right ok -> pure ok- else pure Success- , exporterShutdown = pure ()- }- where- retryDelay = 100_000 -- 100ms- 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 spans_ = do- msg <- encodeMessage <$> immutableSpansToProtobuf spans_- -- TODO handle server disconnect- let req =- baseReq- { requestBody =- RequestBodyLBS $ encoder $ L.fromStrict msg- }- sendReq req 0 -- TODO =<< getTime for maximum cutoff- 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)-- either print (\_ -> pure ()) eResp-- case eResp of- Left err@(HttpExceptionRequest _ e) ->- if isRetryableException e- then exponentialBackoff- else pure $ Failure $ Just $ SomeException err- Left err -> do- pure $ Failure $ Just $ SomeException err- Right resp ->- if isRetryableStatusCode (responseStatus resp)- then case lookup hRetryAfter $ responseHeaders resp of- Nothing -> exponentialBackoff- Just retryAfter -> do- -- TODO support date in retry-after header- 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 do- print resp- pure $ Failure Nothing- else pure Success---attributesToProto :: Attributes -> Vector KeyValue-attributesToProto =- V.fromList- . fmap attributeToKeyValue- . H.toList- . snd- . getAttributes- where- primAttributeToAnyValue = \case- TextAttribute t -> defMessage & stringValue .~ t- BoolAttribute b -> defMessage & boolValue .~ b- DoubleAttribute d -> defMessage & doubleValue .~ d- IntAttribute i -> defMessage & intValue .~ i- attributeToKeyValue :: (Text, Attribute) -> KeyValue- attributeToKeyValue (k, v) =- defMessage- & key .~ k- & value- .~ ( case v of- AttributeValue a -> primAttributeToAnyValue a- AttributeArray a ->- defMessage- & arrayValue .~ (defMessage & values .~ fmap primAttributeToAnyValue a)- )---immutableSpansToProtobuf :: (MonadIO m) => HashMap OT.InstrumentationLibrary (Vector OT.ImmutableSpan) -> m ExportTraceServiceRequest-immutableSpansToProtobuf completedSpans = do- spansByLibrary <- mapM makeInstrumentationLibrarySpans spanGroupList- pure $- defMessage- & vec'resourceSpans- .~ Vector.singleton- ( defMessage- & resource- .~ ( defMessage- & vec'attributes .~ attributesToProto (getMaterializedResourcesAttributes someResourceGroup)- -- TODO- & droppedAttributesCount .~ 0- )- -- TODO, seems like spans need to be emitted via an API- -- that lets us keep them grouped by instrumentation originator- & instrumentationLibrarySpans .~ spansByLibrary- )- where- -- TODO this won't work right if multiple TracerProviders are exporting to a single OTLP exporter with different resources- someResourceGroup = case spanGroupList of- [] -> emptyMaterializedResources- ((_, r) : _) -> case r V.!? 0 of- Nothing -> emptyMaterializedResources- Just s -> OT.getTracerProviderResources $ OT.getTracerTracerProvider $ OT.spanTracer s-- spanGroupList = H.toList completedSpans-- makeInstrumentationLibrarySpans :: (MonadIO m) => (OT.InstrumentationLibrary, Vector OT.ImmutableSpan) -> m InstrumentationLibrarySpans- makeInstrumentationLibrarySpans (library, completedSpans_) = do- spans_ <- mapM makeSpan completedSpans_- pure $- defMessage- & instrumentationLibrary- .~ ( defMessage- & Proto.Opentelemetry.Proto.Trace.V1.Trace_Fields.name .~ OT.libraryName library- & version .~ OT.libraryVersion library- )- & vec'spans .~ spans_----- & schemaUrl .~ "" -- TODO--makeSpan :: (MonadIO m) => OT.ImmutableSpan -> m Span-makeSpan completedSpan = do- let startTime = timestampNanoseconds (OT.spanStart completedSpan)- parentSpanF <- do- case OT.spanParent completedSpan of- Nothing -> pure id- Just s -> do- spanCtxt <- OT.spanId <$> OT.getSpanContext s- pure (\otlpSpan -> otlpSpan & parentSpanId .~ spanIdBytes spanCtxt)-- pure $- defMessage- & traceId .~ traceIdBytes (OT.traceId $ OT.spanContext completedSpan)- & spanId .~ spanIdBytes (OT.spanId $ OT.spanContext completedSpan)- & traceState .~ "" -- TODO (_ $ OT.traceState $ OT.spanContext completedSpan)- & Proto.Opentelemetry.Proto.Trace.V1.Trace_Fields.name .~ OT.spanName completedSpan- & 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- )- & startTimeUnixNano .~ startTime- & endTimeUnixNano .~ maybe startTime timestampNanoseconds (OT.spanEnd completedSpan)- & vec'attributes .~ attributesToProto (OT.spanAttributes completedSpan)- & droppedAttributesCount .~ fromIntegral (fst (getAttributes $ OT.spanAttributes completedSpan))- & vec'events .~ fmap makeEvent (appendOnlyBoundedCollectionValues $ OT.spanEvents completedSpan)- & droppedEventsCount .~ fromIntegral (appendOnlyBoundedCollectionDroppedElementCount (OT.spanEvents completedSpan))- & vec'links .~ fmap makeLink (frozenBoundedCollectionValues $ OT.spanLinks completedSpan)- & droppedLinksCount .~ fromIntegral (frozenBoundedCollectionDroppedElementCount (OT.spanLinks completedSpan))- & status- .~ ( case OT.spanStatus completedSpan of- OT.Unset ->- defMessage- & code .~ Status'STATUS_CODE_UNSET- OT.Ok ->- defMessage- & code .~ Status'STATUS_CODE_OK- (OT.Error e) ->- defMessage- & code .~ Status'STATUS_CODE_ERROR- & message .~ e- )- & parentSpanF---makeEvent :: OT.Event -> Span'Event-makeEvent e =- defMessage- & timeUnixNano .~ timestampNanoseconds (OT.eventTimestamp e)- & Proto.Opentelemetry.Proto.Trace.V1.Trace_Fields.name .~ OT.eventName e- & vec'attributes .~ attributesToProto (OT.eventAttributes e)- & droppedAttributesCount .~ fromIntegral (fst (getAttributes $ OT.eventAttributes e))-+import OpenTelemetry.Exporter.OTLP.Span -makeLink :: OT.Link -> Span'Link-makeLink l =- defMessage- & traceId .~ traceIdBytes (OT.traceId $ OT.frozenLinkContext l)- & spanId .~ spanIdBytes (OT.spanId $ OT.frozenLinkContext l)- & vec'attributes .~ attributesToProto (OT.frozenLinkAttributes l)- & droppedAttributesCount .~ fromIntegral (fst (getAttributes $ OT.frozenLinkAttributes l))
+ src/OpenTelemetry/Exporter/OTLP/LogRecord.hs view
@@ -0,0 +1,2 @@+module OpenTelemetry.Exporter.OTLP.LogRecord () where+
+ src/OpenTelemetry/Exporter/OTLP/Span.hs view
@@ -0,0 +1,479 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE NumericUnderscores #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}++-----------------------------------------------------------------------------++-----------------------------------------------------------------------------++{- |+ Module : OpenTelemetry.Exporter.OTLP.Span+ Copyright : (c) Ian Duncan, 2021+ License : BSD-3+ Description : OTLP Exporter+ Maintainer : Ian Duncan+ Stability : experimental+ Portability : non-portable (GHC extensions)++ The OTLP Exporter is the recommend exporter format to use where possible.++ A number of vendors offer support for exporting traces, logs, and metrics using the vendor-agnostic OTLP protocol.++ Additionally, the OTLP format is supported by the <https://opentelemetry.io/docs/collector/ OpenTelemetry Collector>.++ The OpenTelemetry Collector offers a vendor-agnostic implementation of how to receive, process and export telemetry data.+ It removes the need to run, operate, and maintain multiple agents/collectors.+ This works with improved scalability and supports open-source observability data formats (e.g. Jaeger, Prometheus, Fluent Bit, etc.) sending to+ one or more open-source or commercial back-ends. The local Collector agent is the default location to which instrumentation libraries export+ their telemetry data.+-}+module OpenTelemetry.Exporter.OTLP.Span (+ -- * Initializing the exporter+ otlpExporter,++ -- * Configuring the exporter+ OTLPExporterConfig (..),+ CompressionFormat (..),+ Protocol (..),+ loadExporterEnvironmentVariables,++ -- * Default local endpoints+ otlpExporterHttpEndpoint,+ otlpExporterGRpcEndpoint,+) where++import Codec.Compression.GZip+import Control.Applicative ((<|>))+import Control.Concurrent (threadDelay)+import Control.Exception (SomeAsyncException (..), SomeException (..), fromException, throwIO, try)+import Control.Monad.IO.Class+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.HashMap.Strict (HashMap)+import qualified Data.HashMap.Strict as H+import Data.Maybe+import Data.ProtoLens.Encoding+import Data.ProtoLens.Message+import Data.Text (Text)+import qualified Data.Text.Encoding as T+import Data.Vector (Vector)+import qualified Data.Vector as V+import qualified Data.Vector as Vector+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.Environment+import OpenTelemetry.Exporter.Span+import OpenTelemetry.Resource+import OpenTelemetry.Trace.Core (timestampNanoseconds)+import qualified OpenTelemetry.Trace.Core as OT+import OpenTelemetry.Trace.Id (spanIdBytes, traceIdBytes)+import OpenTelemetry.Util+import Proto.Opentelemetry.Proto.Collector.Trace.V1.TraceService (ExportTraceServiceRequest)+import Proto.Opentelemetry.Proto.Common.V1.Common+import qualified Proto.Opentelemetry.Proto.Common.V1.Common_Fields as Common_Fields+import Proto.Opentelemetry.Proto.Trace.V1.Trace+import qualified Proto.Opentelemetry.Proto.Trace.V1.Trace_Fields as Trace_Fields+import System.Environment+import Text.Read (readMaybe)+++data CompressionFormat = None | GZip+++data Protocol {- GRpc | HttpJson | -}+ = -- | Note: grpc and http/json will likely be supported eventually,+ -- but not yet.+ HttpProtobuf+++otlpExporterHttpEndpoint :: C.ByteString+otlpExporterHttpEndpoint = "http://localhost:4318"+++otlpExporterGRpcEndpoint :: C.ByteString+otlpExporterGRpcEndpoint = "http://localhost:4317"+++data OTLPExporterConfig = OTLPExporterConfig+ { otlpEndpoint :: Maybe String+ , otlpTracesEndpoint :: Maybe String+ , otlpMetricsEndpoint :: Maybe String+ , otlpInsecure :: Bool+ , otlpSpanInsecure :: Bool+ , otlpMetricInsecure :: Bool+ , otlpCertificate :: Maybe FilePath+ , otlpTracesCertificate :: Maybe FilePath+ , otlpMetricCertificate :: Maybe FilePath+ , otlpHeaders :: Maybe [Header]+ , otlpTracesHeaders :: Maybe [Header]+ , otlpMetricsHeaders :: Maybe [Header]+ , otlpCompression :: Maybe CompressionFormat+ , otlpTracesCompression :: Maybe CompressionFormat+ , otlpMetricsCompression :: Maybe CompressionFormat+ , otlpTimeout :: Maybe Int+ -- ^ Measured in seconds+ , otlpTracesTimeout :: Maybe Int+ , otlpMetricsTimeout :: Maybe Int+ , otlpProtocol :: Maybe Protocol+ , otlpTracesProtocol :: Maybe Protocol+ , otlpMetricsProtocol :: Maybe Protocol+ }+++loadExporterEnvironmentVariables :: (MonadIO m) => m OTLPExporterConfig+loadExporterEnvironmentVariables = liftIO $ do+ OTLPExporterConfig+ <$> lookupEnv "OTEL_EXPORTER_OTLP_ENDPOINT"+ <*> lookupEnv "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT"+ <*> lookupEnv "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT"+ <*> lookupBooleanEnv "OTEL_EXPORTER_OTLP_INSECURE"+ <*> lookupBooleanEnv "OTEL_EXPORTER_OTLP_SPAN_INSECURE"+ <*> lookupBooleanEnv "OTEL_EXPORTER_OTLP_METRIC_INSECURE"+ <*> lookupEnv "OTEL_EXPORTER_OTLP_CERTIFICATE"+ <*> lookupEnv "OTEL_EXPORTER_OTLP_TRACES_CERTIFICATE"+ <*> lookupEnv "OTEL_EXPORTER_OTLP_METRICS_CERTIFICATE"+ <*> (fmap decodeHeaders <$> lookupEnv "OTEL_EXPORTER_OTLP_HEADERS")+ <*>+ -- TODO lookupEnv "OTEL_EXPORTER_OTLP_TRACES_HEADERS" <*>+ pure Nothing+ <*>+ -- TODO lookupEnv "OTEL_EXPORTER_OTLP_METRICS_HEADERS" <*>+ pure Nothing+ <*>+ -- TODO lookupEnv <*>+ ( fmap+ ( \case+ "gzip" -> GZip+ "none" -> None+ -- TODO+ _ -> None+ )+ <$> lookupEnv "OTEL_EXPORTER_OTLP_COMPRESSION"+ )+ <*>+ -- TODO lookupEnv "OTEL_EXPORTER_OTLP_TRACES_COMPRESSION" <*>+ pure Nothing+ <*>+ -- TODO lookupEnv "OTEL_EXPORTER_OTLP_METRICS_COMPRESSION" <*>+ pure Nothing+ <*>+ -- TODO lookupEnv "OTEL_EXPORTER_OTLP_TIMEOUT" <*>+ pure Nothing+ <*>+ -- TODO lookupEnv "OTEL_EXPORTER_OTLP_TRACES_TIMEOUT" <*>+ pure Nothing+ <*>+ -- TODO lookupEnv "OTEL_EXPORTER_OTLP_METRICS_TIMEOUT" <*>+ pure Nothing+ <*>+ -- TODO lookupEnv "OTEL_EXPORTER_OTLP_PROTOCOL" <*>+ pure Nothing+ <*>+ -- TODO lookupEnv "OTEL_EXPORTER_OTLP_TRACES_PROTOCOL" <*>+ pure Nothing+ <*>+ -- TODO lookupEnv "OTEL_EXPORTER_OTLP_METRICS_PROTOCOL"+ pure Nothing+ where+ decodeHeaders hsString = case Baggage.decodeBaggageHeader $ C.pack hsString of+ Left _ -> mempty+ Right baggageFmt ->+ (\(k, v) -> (CI.mk $ Baggage.tokenValue k, T.encodeUtf8 $ Baggage.value v)) <$> H.toList (Baggage.values baggageFmt)+++protobufMimeType :: C.ByteString+protobufMimeType = "application/x-protobuf"+++-- | Initial the OTLP 'Exporter'+otlpExporter :: (MonadIO m) => OTLPExporterConfig -> m SpanExporter+otlpExporter conf = do+ -- TODO, url parsing is janky+ -- TODO configurable retryDelay, maximum retry counts+ let+ defaultHost = "http://localhost:4318"+ host = fromMaybe defaultHost $ otlpEndpoint conf+ req <- liftIO $ parseRequest (host <> "/v1/traces")++ let (encodingHeader, encoder) =+ maybe+ (id, id)+ ( \case+ None -> (id, id)+ GZip -> (((hContentEncoding, "gzip") :), compress)+ )+ (otlpTracesCompression conf <|> otlpCompression conf)++ baseReqHeaders =+ encodingHeader $+ (hContentType, protobufMimeType)+ : (hAcceptEncoding, protobufMimeType)+ : fromMaybe [] (otlpHeaders conf)+ ++ fromMaybe [] (otlpTracesHeaders conf)+ ++ requestHeaders req+ baseReq =+ req+ { method = "POST"+ , requestHeaders = baseReqHeaders+ }+ pure $+ SpanExporter+ { spanExporterExport = \spans_ -> do+ let anySpansToExport = H.size spans_ /= 0 && not (all V.null $ H.elems spans_)+ if anySpansToExport+ then do+ result <- try $ exporterExportCall encoder baseReq spans_+ case result of+ Left err -> do+ -- If the exception is async, then we need to rethrow it+ -- here. Otherwise, there's a good chance that the+ -- calling code will swallow the exception and cause+ -- a problem.+ case fromException err of+ Just (SomeAsyncException _) -> do+ throwIO err+ Nothing ->+ pure $ Failure $ Just err+ Right ok -> pure ok+ else pure Success+ , spanExporterShutdown = pure ()+ }+ where+ retryDelay = 100_000 -- 100ms+ 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 spans_ = do+ msg <- encodeMessage <$> immutableSpansToProtobuf spans_+ -- TODO handle server disconnect+ let req =+ baseReq+ { requestBody =+ RequestBodyLBS $ encoder $ L.fromStrict msg+ }+ sendReq req 0 -- TODO =<< getTime for maximum cutoff+ 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 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+ pure $ Failure $ Just $ SomeException err+ Right resp ->+ if isRetryableStatusCode (responseStatus resp)+ then case lookup hRetryAfter $ responseHeaders resp of+ Nothing -> exponentialBackoff+ Just retryAfter -> do+ -- TODO support date in retry-after header+ 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 do+ print resp+ pure $ Failure Nothing+ else pure Success+++attributesToProto :: Attributes -> Vector KeyValue+attributesToProto =+ V.fromList+ . fmap attributeToKeyValue+ . H.toList+ . snd+ . getAttributes+ 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)+ )+++immutableSpansToProtobuf :: (MonadIO m) => HashMap OT.InstrumentationLibrary (Vector OT.ImmutableSpan) -> m ExportTraceServiceRequest+immutableSpansToProtobuf completedSpans = do+ spansByLibrary <- mapM makeScopeSpans spanGroupList+ pure $+ defMessage+ & Trace_Fields.vec'resourceSpans+ .~ Vector.singleton+ ( defMessage+ & Trace_Fields.resource+ .~ ( defMessage+ & 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+ & Trace_Fields.scopeSpans+ .~ spansByLibrary+ )+ where+ -- TODO this won't work right if multiple TracerProviders are exporting to a single OTLP exporter with different resources+ someResourceGroup = case spanGroupList of+ [] -> emptyMaterializedResources+ ((_, r) : _) -> case r V.!? 0 of+ Nothing -> emptyMaterializedResources+ Just s -> OT.getTracerProviderResources $ OT.getTracerTracerProvider $ OT.spanTracer s++ spanGroupList = H.toList completedSpans++ makeScopeSpans :: (MonadIO m) => (OT.InstrumentationLibrary, Vector OT.ImmutableSpan) -> m ScopeSpans+ makeScopeSpans (library, completedSpans_) = do+ spans_ <- mapM makeSpan completedSpans_+ pure $+ defMessage+ & Trace_Fields.scope+ .~ ( defMessage+ & Trace_Fields.name+ .~ OT.libraryName library+ & Common_Fields.version+ .~ OT.libraryVersion library+ )+ & Trace_Fields.vec'spans+ .~ spans_+++-- & schemaUrl .~ "" -- TODO++makeSpan :: (MonadIO m) => OT.ImmutableSpan -> m Span+makeSpan completedSpan = do+ let startTime = timestampNanoseconds (OT.spanStart completedSpan)+ parentSpanF <- do+ case OT.spanParent completedSpan of+ Nothing -> pure id+ Just s -> do+ spanCtxt <- OT.spanId <$> OT.getSpanContext s+ pure (\otlpSpan -> otlpSpan & Trace_Fields.parentSpanId .~ spanIdBytes spanCtxt)++ pure $+ defMessage+ & Trace_Fields.traceId+ .~ traceIdBytes (OT.traceId $ OT.spanContext completedSpan)+ & Trace_Fields.spanId+ .~ spanIdBytes (OT.spanId $ OT.spanContext completedSpan)+ & Trace_Fields.traceState+ .~ "" -- TODO (_ $ OT.traceState $ OT.spanContext completedSpan)+ & Trace_Fields.name+ .~ OT.spanName completedSpan+ & 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+ )+ & Trace_Fields.startTimeUnixNano+ .~ startTime+ & Trace_Fields.endTimeUnixNano+ .~ maybe startTime timestampNanoseconds (OT.spanEnd completedSpan)+ & Trace_Fields.vec'attributes+ .~ attributesToProto (OT.spanAttributes completedSpan)+ & Trace_Fields.droppedAttributesCount+ .~ fromIntegral (fst (getAttributes $ OT.spanAttributes completedSpan))+ & Trace_Fields.vec'events+ .~ fmap makeEvent (appendOnlyBoundedCollectionValues $ OT.spanEvents completedSpan)+ & Trace_Fields.droppedEventsCount+ .~ fromIntegral (appendOnlyBoundedCollectionDroppedElementCount (OT.spanEvents completedSpan))+ & Trace_Fields.vec'links+ .~ fmap makeLink (appendOnlyBoundedCollectionValues $ OT.spanLinks completedSpan)+ & Trace_Fields.droppedLinksCount+ .~ fromIntegral (appendOnlyBoundedCollectionDroppedElementCount (OT.spanLinks completedSpan))+ & 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+ )+ & parentSpanF+++makeEvent :: OT.Event -> Span'Event+makeEvent e =+ defMessage+ & Trace_Fields.timeUnixNano+ .~ timestampNanoseconds (OT.eventTimestamp e)+ & Trace_Fields.name+ .~ OT.eventName e+ & Trace_Fields.vec'attributes+ .~ attributesToProto (OT.eventAttributes e)+ & Trace_Fields.droppedAttributesCount+ .~ fromIntegral (fst (getAttributes $ OT.eventAttributes e))+++makeLink :: OT.Link -> Span'Link+makeLink l =+ defMessage+ & Trace_Fields.traceId+ .~ traceIdBytes (OT.traceId $ OT.frozenLinkContext l)+ & Trace_Fields.spanId+ .~ spanIdBytes (OT.spanId $ OT.frozenLinkContext l)+ & Trace_Fields.vec'attributes+ .~ attributesToProto (OT.frozenLinkAttributes l)+ & Trace_Fields.droppedAttributesCount+ .~ fromIntegral (fst (getAttributes $ OT.frozenLinkAttributes l))
− test/Spec.hs
@@ -1,3 +0,0 @@--main :: IO ()-main = putStrLn "Test suite not yet implemented"