diff --git a/exe/eventlog-summary/Main.hs b/exe/eventlog-summary/Main.hs --- a/exe/eventlog-summary/Main.hs +++ b/exe/eventlog-summary/Main.hs @@ -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"]] diff --git a/exe/eventlog-to-zipkin/Main.hs b/exe/eventlog-to-zipkin/Main.hs --- a/exe/eventlog-to-zipkin/Main.hs +++ b/exe/eventlog-to-zipkin/Main.hs @@ -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 diff --git a/opentelemetry-extra.cabal b/opentelemetry-extra.cabal --- a/opentelemetry-extra.cabal +++ b/opentelemetry-extra.cabal @@ -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, diff --git a/src/OpenTelemetry/ChromeExporter.hs b/src/OpenTelemetry/ChromeExporter.hs --- a/src/OpenTelemetry/ChromeExporter.hs +++ b/src/OpenTelemetry/ChromeExporter.hs @@ -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 diff --git a/src/OpenTelemetry/Common.hs b/src/OpenTelemetry/Common.hs --- a/src/OpenTelemetry/Common.hs +++ b/src/OpenTelemetry/Common.hs @@ -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 diff --git a/src/OpenTelemetry/EventlogStreaming_Internal.hs b/src/OpenTelemetry/EventlogStreaming_Internal.hs --- a/src/OpenTelemetry/EventlogStreaming_Internal.hs +++ b/src/OpenTelemetry/EventlogStreaming_Internal.hs @@ -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 diff --git a/unit-tests/Arbitrary.hs b/unit-tests/Arbitrary.hs --- a/unit-tests/Arbitrary.hs +++ b/unit-tests/Arbitrary.hs @@ -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 diff --git a/unit-tests/Main.hs b/unit-tests/Main.hs --- a/unit-tests/Main.hs +++ b/unit-tests/Main.hs @@ -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