opentelemetry-extra 0.6.0 → 0.6.1
raw patch · 8 files changed
+181/−143 lines, 8 filesdep ~opentelemetry
Dependency ranges changed: opentelemetry
Files
- exe/eventlog-summary/Main.hs +19/−15
- exe/eventlog-to-zipkin/Main.hs +1/−1
- opentelemetry-extra.cabal +15/−14
- src/OpenTelemetry/ChromeExporter.hs +20/−17
- src/OpenTelemetry/Common.hs +33/−29
- src/OpenTelemetry/EventlogStreaming_Internal.hs +73/−57
- unit-tests/Arbitrary.hs +11/−10
- unit-tests/Main.hs +9/−0
exe/eventlog-summary/Main.hs view
@@ -1,13 +1,10 @@+{-# LANGUAGE CPP #-} {-# LANGUAGE OverloadedStrings #-} module Main where import Control.Monad-import qualified Data.Text as T import qualified Data.ByteString.Char8 as B8-import OpenTelemetry.Common-import OpenTelemetry.EventlogStreaming_Internal-import System.Environment import Data.Char (isDigit) import Data.Function import qualified Data.HashTable.IO as H@@ -15,8 +12,12 @@ import Data.IntMap.Strict (IntMap) import qualified Data.IntMap.Strict as IntMap import Data.List (sortOn)+import qualified Data.Text as T import Data.Word import Graphics.Vega.VegaLite hiding (name)+import OpenTelemetry.Common+import OpenTelemetry.EventlogStreaming_Internal+import System.Environment import Text.Printf type HashTable k v = H.BasicHashTable k v@@ -68,17 +69,19 @@ pure ExportSuccess ) (pure ())- metric_exporter <- aggregated $ Exporter- ( \metrics -> do- forM_ metrics $ \(AggregatedMetric (CaptureInstrument _ name) (MetricDatapoint _ value)) ->- modifyIORef metricStats $ \s -> case splitCapability (B8.unpack name) of- (_, "threads") -> s { max_threads = max value (max_threads s) }- (Just cap, "heap_alloc_bytes") -> s { total_alloc_bytes = IntMap.insert cap value (total_alloc_bytes s) }- (_, "heap_live_bytes") -> s { max_live_bytes = max value (max_live_bytes s) }- _ -> s- pure ExportSuccess- )- (pure ())+ metric_exporter <-+ aggregated $+ Exporter+ ( \metrics -> do+ forM_ metrics $ \(AggregatedMetric (CaptureInstrument _ name) (MetricDatapoint _ value)) ->+ modifyIORef metricStats $ \s -> case splitCapability (B8.unpack name) of+ (_, "threads") -> s {max_threads = max value (max_threads s)}+ (Just cap, "heap_alloc_bytes") -> s {total_alloc_bytes = IntMap.insert cap value (total_alloc_bytes s)}+ (_, "heap_live_bytes") -> s {max_live_bytes = max value (max_live_bytes s)}+ _ -> s+ pure ExportSuccess+ )+ (pure ()) exportEventlog span_exporter metric_exporter path leaderboard <- sortOn (total_ns . snd) <$> H.toList opCounts printf "Count\tTot ms\tMin ms\tMax ms\tOperation\n"@@ -115,6 +118,7 @@ mark Bar [], vega_enc ]+ vega_enc = [] & position Y [PName "op", PmType Nominal, PAxis [AxTitle "operation"]]
exe/eventlog-to-zipkin/Main.hs view
@@ -3,8 +3,8 @@ module Main where import qualified Data.Text as T-import OpenTelemetry.EventlogStreaming_Internal import OpenTelemetry.Common+import OpenTelemetry.EventlogStreaming_Internal import OpenTelemetry.ZipkinExporter import System.Environment (getArgs) import System.FilePath
opentelemetry-extra.cabal view
@@ -2,7 +2,7 @@ name: opentelemetry-extra description: The OpenTelemetry Haskell Client https://opentelemetry.io category: OpenTelemetry-version: 0.6.0+version: 0.6.1 license-file: LICENSE license: Apache-2.0 author: Dmitry Ivanov@@ -63,7 +63,7 @@ http-client, http-client-tls, http-types,- opentelemetry >= 0.6.0,+ opentelemetry >= 0.6.1, random >= 1.1, scientific, text-show,@@ -84,12 +84,13 @@ import: options type: exitcode-stdio-1.0 main-is: Main.hs- other-modules:- Arbitrary- TestBinary- TestPropagation- LogEventSerializer- TestEventlogStreaming+ if impl(ghc >= 8.8)+ other-modules:+ Arbitrary+ TestBinary+ TestPropagation+ LogEventSerializer+ TestEventlogStreaming hs-source-dirs: unit-tests build-depends: base,@@ -97,7 +98,7 @@ bytestring, ghc-events, hashable,- opentelemetry >= 0.6.0,+ opentelemetry >= 0.6.1, opentelemetry-extra, splitmix, tasty,@@ -122,7 +123,7 @@ filepath, http-client, http-client-tls,- opentelemetry >= 0.6.0,+ opentelemetry >= 0.6.1, opentelemetry-extra, text, typed-process@@ -141,7 +142,7 @@ build-depends: base, gauge >= 0.2.4,- opentelemetry >= 0.6.0,+ opentelemetry >= 0.6.1, executable eventlog-to-chrome import: options@@ -151,7 +152,7 @@ base, exceptions, clock,- opentelemetry >= 0.6.0,+ opentelemetry >= 0.6.1, opentelemetry-extra, executable eventlog-to-tracy@@ -162,7 +163,7 @@ base, clock, directory,- opentelemetry >= 0.6.0,+ opentelemetry >= 0.6.1, opentelemetry-extra, process, @@ -174,7 +175,7 @@ base, hashtables, containers,- opentelemetry >= 0.6.0,+ opentelemetry >= 0.6.1, opentelemetry-extra, hvega, text,
src/OpenTelemetry/ChromeExporter.hs view
@@ -5,14 +5,14 @@ import Control.Monad import Data.Aeson import qualified Data.ByteString.Lazy as LBS-import qualified Data.Text.Encoding as TE import Data.Function import Data.HashMap.Strict as HM import Data.List (sortOn)+import qualified Data.Text.Encoding as TE import Data.Word import OpenTelemetry.Common-import System.IO import OpenTelemetry.EventlogStreaming_Internal+import System.IO newtype ChromeBeginSpan = ChromeBegin Span @@ -101,21 +101,24 @@ hPutStrLn f "\n]" hClose f )- metric_exporter <- aggregated $ Exporter- ( \metrics -> do- -- forM_ metrics $ \(AggregatedMetric (SomeInstrument (TE.decodeUtf8 . instrumentName -> name)) (MetricDatapoint ts value)) -> do- forM_ metrics $ \(AggregatedMetric (CaptureInstrument _ (TE.decodeUtf8 -> name)) (MetricDatapoint ts value)) -> do- LBS.hPutStr f $ encode $- object- [ "ph" .= ("C" :: String),- "name" .= name,- "ts" .= (div ts 1000),- "args" .= object [name .= Number (fromIntegral value)]- ]- LBS.hPutStr f ",\n"- pure ExportSuccess- )- (pure ())+ metric_exporter <-+ aggregated $+ Exporter+ ( \metrics -> do+ -- forM_ metrics $ \(AggregatedMetric (SomeInstrument (TE.decodeUtf8 . instrumentName -> name)) (MetricDatapoint ts value)) -> do+ forM_ metrics $ \(AggregatedMetric (CaptureInstrument _ (TE.decodeUtf8 -> name)) (MetricDatapoint ts value)) -> do+ LBS.hPutStr f $+ encode $+ object+ [ "ph" .= ("C" :: String),+ "name" .= name,+ "ts" .= (div ts 1000),+ "args" .= object [name .= Number (fromIntegral value)]+ ]+ LBS.hPutStr f ",\n"+ pure ExportSuccess+ )+ (pure ()) pure (span_exporter, metric_exporter) data DoWeCollapseThreads = CollapseThreads | SplitThreads
src/OpenTelemetry/Common.hs view
@@ -1,25 +1,25 @@-{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE DeriveFunctor #-}-{-# LANGUAGE LambdaCase #-} {-# LANGUAGE DuplicateRecordFields #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-} module OpenTelemetry.Common where +import Control.Monad import Data.Aeson+import qualified Data.ByteString as BS import qualified Data.HashMap.Strict as HM import Data.Hashable+import Data.IORef (modifyIORef, newIORef, readIORef)+import Data.List (sortOn) import Data.String import qualified Data.Text as T import Data.Word import GHC.Generics+import GHC.Int (Int8) import OpenTelemetry.SpanContext import System.Clock-import Data.IORef (readIORef, modifyIORef, newIORef)-import Control.Monad-import Data.List (sortOn)-import qualified Data.ByteString as BS-import GHC.Int (Int8) type Timestamp = Word64 @@ -84,6 +84,7 @@ | UpDownSumObserverType | ValueObserverType deriving (Show, Eq, Enum, Generic)+ instance Hashable InstrumentType data CaptureInstrument = CaptureInstrument@@ -91,6 +92,7 @@ instrumentName :: !BS.ByteString } deriving (Show, Eq, Generic)+ instance Hashable CaptureInstrument -- | Based on https://github.com/open-telemetry/opentelemetry-proto/blob/1a931b4b57c34e7fd8f7dddcaa9b7587840e9c08/opentelemetry/proto/metrics/v1/metrics.proto#L96-L107@@ -147,11 +149,10 @@ | ExportFailedNotRetryable deriving (Show, Eq) -data Exporter thing- = Exporter- { export :: [thing] -> IO ExportResult,- shutdown :: IO ()- }+data Exporter thing = Exporter+ { export :: [thing] -> IO ExportResult,+ shutdown :: IO ()+ } readInstrumentTag :: Int8 -> Maybe InstrumentType readInstrumentTag 1 = Just CounterType@@ -179,24 +180,27 @@ -- in, it either replaces or gets added to the current value, based on whether -- the instrument is additive. currentValuesRef <- newIORef HM.empty- return $ Exporter- { export = \metrics -> do- forM_ metrics $ \(Metric instrument datapoints) -> do- forM_ (sortOn timestamp datapoints) $ \dp@(MetricDatapoint ts value) ->- modifyIORef currentValuesRef $- if additive (instrumentType instrument)- then HM.alter- (\case- Nothing -> Just dp- Just (MetricDatapoint _ oldValue) -> Just (MetricDatapoint ts $ oldValue+value))- instrument- else HM.insert instrument dp+ return $+ Exporter+ { export = \metrics -> do+ forM_ metrics $ \(Metric instrument datapoints) -> do+ forM_ (sortOn timestamp datapoints) $ \dp@(MetricDatapoint ts value) ->+ modifyIORef currentValuesRef $+ if additive (instrumentType instrument)+ then+ HM.alter+ ( \case+ Nothing -> Just dp+ Just (MetricDatapoint _ oldValue) -> Just (MetricDatapoint ts $ oldValue + value)+ )+ instrument+ else HM.insert instrument dp - -- Read the latest value for each instrument- currentValues <- readIORef currentValuesRef- export [AggregatedMetric i (currentValues HM.! i) | Metric i _ <- metrics]- , shutdown- }+ -- Read the latest value for each instrument+ currentValues <- readIORef currentValuesRef+ export [AggregatedMetric i (currentValues HM.! i) | Metric i _ <- metrics],+ shutdown+ } now64 :: IO Timestamp now64 = do
src/OpenTelemetry/EventlogStreaming_Internal.hs view
@@ -6,11 +6,11 @@ import Control.Concurrent (threadDelay) import qualified Data.Binary.Get as DBG import Data.Bits-import Data.Int import qualified Data.ByteString as B import qualified Data.ByteString.Char8 as B8 import qualified Data.ByteString.Lazy as LBS import qualified Data.HashMap.Strict as HM+import Data.Int import qualified Data.IntMap as IM import Data.List (isSuffixOf) import Data.Maybe@@ -23,14 +23,13 @@ import GHC.Stack import OpenTelemetry.Common hiding (Event, Timestamp) import OpenTelemetry.Debug-import OpenTelemetry.SpanContext import OpenTelemetry.Eventlog (InstrumentId, InstrumentName)-import Text.Printf-import qualified System.Random.SplitMix as R- import OpenTelemetry.Eventlog_Internal+import OpenTelemetry.SpanContext import System.Clock import System.IO+import qualified System.Random.SplitMix as R+import Text.Printf data WatDoOnEOF = StopOnEOF | SleepAndRetryOnEOF @@ -153,19 +152,21 @@ let trace_id = case m_trace_id of Just t -> t Nothing -> TId originTimestamp -- TODO: something more random- in (st { traceMap = HM.insert new_tid trace_id traceMap }- , []- , [Metric threadsI [MetricDatapoint now 1]])+ in ( st {traceMap = HM.insert new_tid trace_id traceMap},+ [],+ [Metric threadsI [MetricDatapoint now 1]]+ ) (RunThread tid, Just cap, _) -> (st {threadMap = IM.insert cap tid threadMap}, [], []) (StopThread tid tstatus, Just cap, _) | isTerminalThreadStatus tstatus -> ( st- { threadMap = IM.delete cap threadMap- , traceMap = HM.delete tid traceMap+ { threadMap = IM.delete cap threadMap,+ traceMap = HM.delete tid traceMap },- []- , [Metric threadsI [MetricDatapoint now (-1)]])+ [],+ [Metric threadsI [MetricDatapoint now (-1)]]+ ) (StartGC, _, _) -> (st {gcStartedAt = now}, [], []) (HeapLive {liveBytes}, _, _) -> (st, [], [Metric heapLiveBytesI [MetricDatapoint now $ fromIntegral liveBytes]])@@ -205,7 +206,6 @@ heapAllocBytesI :: Int -> CaptureInstrument heapAllocBytesI cap = CaptureInstrument SumObserverType ("cap_" <> B8.pack (show cap) <> "_heap_alloc_bytes") - isTerminalThreadStatus :: ThreadStopStatus -> Bool isTerminalThreadStatus ThreadFinished = True isTerminalThreadStatus _ = False@@ -306,7 +306,7 @@ let (st', sp) = emitSpan serial span_id st in (st', [sp {spanOperation = operation, spanStartedAt = now, spanThreadId = tid}], []) DeclareInstrumentEv iType iId iName ->- (st { instrumentMap = HM.insert iId (CaptureInstrument iType iName) (instrumentMap st) }, [], [])+ (st {instrumentMap = HM.insert iId (CaptureInstrument iType iName) (instrumentMap st)}, [], []) MetricCaptureEv instrumentId val -> case HM.lookup instrumentId (instrumentMap st) of Just instrument -> (st, [], [Metric instrument [MetricDatapoint now val]]) Nothing -> error $ "Undeclared instrument id: " ++ show instrumentId@@ -379,43 +379,57 @@ parseText :: [T.Text] -> Maybe OpenTelemetryEventlogEvent parseText =- \case- ("ot2" : "begin" : "span" : serial_text : name) ->- let serial = read (T.unpack serial_text)- operation = T.intercalate " " name- in Just $ BeginSpanEv (SpanInFlight serial) (SpanName operation)- ["ot2", "end", "span", serial_text] ->- let serial = read (T.unpack serial_text)- in Just $ EndSpanEv (SpanInFlight serial)- ("ot2" : "set" : "tag" : serial_text : k : v) ->- let serial = read (T.unpack serial_text)- in Just $ TagEv (SpanInFlight serial) (TagName k) (TagVal $ T.unwords v)- ["ot2", "set", "traceid", serial_text, trace_id_text] ->- let serial = read (T.unpack serial_text)- trace_id = TId (read ("0x" <> T.unpack trace_id_text))- in Just $ SetTraceEv (SpanInFlight serial) trace_id- ["ot2", "set", "spanid", serial_text, new_span_id_text] ->- let serial = read (T.unpack serial_text)- span_id = (SId (read ("0x" <> T.unpack new_span_id_text)))- in Just $ SetSpanEv (SpanInFlight serial) span_id- ["ot2", "set", "parent", serial_text, trace_id_text, parent_span_id_text] ->- let trace_id = TId (read ("0x" <> T.unpack trace_id_text))- serial = read (T.unpack serial_text)- psid = SId (read ("0x" <> T.unpack parent_span_id_text))- in Just $- SetParentEv- (SpanInFlight serial)- (SpanContext psid trace_id)- ("ot2" : "add" : "event" : serial_text : k : v) ->- let serial = read (T.unpack serial_text)- in Just . EventEv (SpanInFlight serial) (EventName k) $ EventVal $ T.unwords v- ("ot2" : "metric" : "capture" : instrumentIdText : valStr) ->+ \case+ ("ot2" : "begin" : "span" : serial_text : name) ->+ let serial = read (T.unpack serial_text)+ operation = T.intercalate " " name+ in Just $ BeginSpanEv (SpanInFlight serial) (SpanName operation)+ ["ot2", "end", "span", serial_text] ->+ let serial = read (T.unpack serial_text)+ in Just $ EndSpanEv (SpanInFlight serial)+ ("ot2" : "set" : "tag" : serial_text : k : v) ->+ let serial = read (T.unpack serial_text)+ in Just $ TagEv (SpanInFlight serial) (TagName k) (TagVal $ T.unwords v)+ ["ot2", "set", "traceid", serial_text, trace_id_text] ->+ let serial = read (T.unpack serial_text)+ trace_id = TId (read ("0x" <> T.unpack trace_id_text))+ in Just $ SetTraceEv (SpanInFlight serial) trace_id+ ["ot2", "set", "spanid", serial_text, new_span_id_text] ->+ let serial = read (T.unpack serial_text)+ span_id = (SId (read ("0x" <> T.unpack new_span_id_text)))+ in Just $ SetSpanEv (SpanInFlight serial) span_id+ ["ot2", "set", "parent", serial_text, trace_id_text, parent_span_id_text] ->+ let trace_id = TId (read ("0x" <> T.unpack trace_id_text))+ serial = read (T.unpack serial_text)+ psid = SId (read ("0x" <> T.unpack parent_span_id_text))+ in Just $+ SetParentEv+ (SpanInFlight serial)+ (SpanContext psid trace_id)+ ("ot2" : "add" : "event" : serial_text : k : v) ->+ let serial = read (T.unpack serial_text)+ in Just . EventEv (SpanInFlight serial) (EventName k) $ EventVal $ T.unwords v+ ("ot2" : "metric" : "create" : instrumentTypeTag : instrumentIdText : instrumentNameStrs) ->+ instrumentStringTagP instrumentTypeTag >>= \instrumentType -> let instrumentId = read ("0x" ++ T.unpack instrumentIdText)- val = read (T.unpack $ T.intercalate " " valStr)- in Just $ MetricCaptureEv instrumentId val- ("ot2" : rest) -> error $ printf "Unrecognized %s" (show rest)- _ -> Nothing+ instrumentName = TE.encodeUtf8 (T.intercalate " " instrumentNameStrs)+ in Just $ DeclareInstrumentEv instrumentType instrumentId instrumentName+ ("ot2" : "metric" : "capture" : instrumentIdText : valStr) ->+ let instrumentId = read ("0x" ++ T.unpack instrumentIdText)+ val = read (T.unpack $ T.intercalate " " valStr)+ in Just $ MetricCaptureEv instrumentId val+ ("ot2" : rest) -> error $ printf "Unrecognized %s" (show rest)+ _ -> Nothing +instrumentStringTagP :: T.Text -> Maybe InstrumentType+instrumentStringTagP "Counter" = Just CounterType+instrumentStringTagP "UpDownCounter" = Just UpDownCounterType+instrumentStringTagP "ValueRecorder" = Just ValueRecorderType+instrumentStringTagP "SumObserver" = Just SumObserverType+instrumentStringTagP "UpDownSumObserver" = Just UpDownSumObserverType+instrumentStringTagP "ValueObserver" = Just ValueObserverType+instrumentStringTagP _ = Nothing+ headerP :: DBG.Get (Maybe MsgType) headerP = do h <- DBG.getWord32le@@ -463,13 +477,15 @@ SET_SPAN_ID -> SetSpanEv <$> (SpanInFlight <$> DBG.getWord64le) <*> (SId <$> DBG.getWord64le)- DECLARE_INSTRUMENT -> DeclareInstrumentEv- <$> (instrumentTagP =<< DBG.getInt8)- <*> DBG.getWord64le- <*> (LBS.toStrict <$> DBG.getRemainingLazyByteString)- METRIC_CAPTURE -> MetricCaptureEv- <$> DBG.getWord64le- <*> (fromIntegral <$> DBG.getInt64le)+ DECLARE_INSTRUMENT ->+ DeclareInstrumentEv+ <$> (instrumentTagP =<< DBG.getInt8)+ <*> DBG.getWord64le+ <*> (LBS.toStrict <$> DBG.getRemainingLazyByteString)+ METRIC_CAPTURE ->+ MetricCaptureEv+ <$> DBG.getWord64le+ <*> (fromIntegral <$> DBG.getInt64le) MsgType mti -> fail $ "Log event of type " ++ show mti ++ " is not supported" @@ -479,7 +495,7 @@ Nothing -> return Nothing Just msgType -> logEventBodyP msgType >>= return . Just -instrumentTagP :: MonadFail m => Int8 -> m InstrumentType+instrumentTagP :: Int8 -> DBG.Get InstrumentType instrumentTagP 1 = return CounterType instrumentTagP 2 = return UpDownCounterType instrumentTagP 3 = return ValueRecorderType
unit-tests/Arbitrary.hs view
@@ -10,12 +10,12 @@ import qualified Data.Text as T import OpenTelemetry.Common import OpenTelemetry.Eventlog-import OpenTelemetry.Metrics_Internal import OpenTelemetry.EventlogStreaming_Internal+import OpenTelemetry.Metrics_Internal import OpenTelemetry.SpanContext import Test.QuickCheck import Test.QuickCheck.Arbitrary.Generic-import Test.QuickCheck.Instances.ByteString()+import Test.QuickCheck.Instances.ByteString () import TextShow newtype TextWithout0 = TextWithout0 T.Text@@ -47,14 +47,15 @@ deriving instance Arbitrary TraceId instance Arbitrary SomeInstrument where- arbitrary = oneof- [ SomeInstrument <$> (Counter <$> arbitrary <*> arbitrary)- , SomeInstrument <$> (UpDownCounter <$> arbitrary <*> arbitrary)- , SomeInstrument <$> (ValueRecorder <$> arbitrary <*> arbitrary)- , SomeInstrument <$> (SumObserver <$> arbitrary <*> arbitrary)- , SomeInstrument <$> (UpDownSumObserver <$> arbitrary <*> arbitrary)- , SomeInstrument <$> (ValueObserver <$> arbitrary <*> arbitrary)- ]+ arbitrary =+ oneof+ [ SomeInstrument <$> (Counter <$> arbitrary <*> arbitrary),+ SomeInstrument <$> (UpDownCounter <$> arbitrary <*> arbitrary),+ SomeInstrument <$> (ValueRecorder <$> arbitrary <*> arbitrary),+ SomeInstrument <$> (SumObserver <$> arbitrary <*> arbitrary),+ SomeInstrument <$> (UpDownSumObserver <$> arbitrary <*> arbitrary),+ SomeInstrument <$> (ValueObserver <$> arbitrary <*> arbitrary)+ ] instance Arbitrary SpanContext where arbitrary = genericArbitrary
unit-tests/Main.hs view
@@ -1,1 +1,10 @@+{-# LANGUAGE CPP #-}++#if __GLASGOW_HASKELL__ >= 808 {-# OPTIONS_GHC -F -pgmF tasty-discover #-}+#else+main :: IO ()+main = do+ putStrLn "Tests are not available below ghc 8.8"+ putStrLn "Use ghc 8.8 or above to run the tests"+#endif