packages feed

opentelemetry-extra (empty) → 0.3.0

raw patch · 13 files changed

+903/−0 lines, 13 filesdep +aesondep +asyncdep +base

Dependencies added: aeson, async, base, bytestring, clock, containers, directory, exceptions, filepath, ghc-events, hashable, http-client, http-client-tls, http-types, opentelemetry, opentelemetry-extra, random, scientific, splitmix, stm, tasty, tasty-discover, tasty-hunit, tasty-quickcheck, text, unordered-containers

Files

+ LICENSE view
@@ -0,0 +1,7 @@+Copyright 2020-present Dmitry Ivanov++Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at++http://www.apache.org/licenses/LICENSE-2.0++Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.
+ exe/eventlog-to-chrome/Main.hs view
@@ -0,0 +1,24 @@+{-# LANGUAGE OverloadedStrings #-}++module Main where++import OpenTelemetry.ChromeExporter+import OpenTelemetry.EventlogStreaming_Internal+import OpenTelemetry.Exporter+import System.Clock+import System.Environment (getArgs)+import System.IO+import Text.Printf++main :: IO ()+main = do+  args <- getArgs+  case args of+    ["read", path] -> do+      let target_path = (path <> ".trace.json")+      printf "Converting %s to %s...\n" path target_path+      exporter <- createChromeSpanExporter target_path+      origin_timestamp <- fromIntegral . toNanoSecs <$> getTime Realtime+      withFile path ReadMode (work origin_timestamp exporter)+      shutdown exporter+      putStrLn "\nAll done."
+ exe/eventlog-to-zipkin/Main.hs view
@@ -0,0 +1,30 @@+{-# LANGUAGE OverloadedStrings #-}++module Main where++import qualified Data.Text as T+import OpenTelemetry.EventlogStreaming_Internal+import OpenTelemetry.Exporter+import OpenTelemetry.ZipkinExporter+import System.Clock+import System.Environment (getArgs)+import System.FilePath+import System.IO+import Text.Printf++main :: IO ()+main = do+  args <- getArgs+  case args of+    ["read", path] -> do+      printf "Sending %s to Zipkin...\n" path+      let service_name = T.pack $ takeBaseName path+      exporter <- createZipkinSpanExporter $ localhostZipkinConfig service_name+      origin_timestamp <- fromIntegral . toNanoSecs <$> getTime Realtime+      withFile path ReadMode (work origin_timestamp exporter)+      shutdown exporter+      putStrLn "\nAll done."+    _ -> do+      putStrLn "Usage:"+      putStrLn ""+      putStrLn "  eventlog-to-lightstep read <program.eventlog>"
+ opentelemetry-extra.cabal view
@@ -0,0 +1,125 @@+cabal-version:       2.4+name:                opentelemetry-extra+description:         The OpenTelemetry Haskell Client https://opentelemetry.io+category:            OpenTelemetry+version:             0.3.0+license-file:        LICENSE+license:             Apache-2.0+author:              Dmitry Ivanov+maintainer:          ethercrow@gmail.com+build-type:          Simple++source-repository head+  type: git+  location: https://github.com/ethercrow/opentelemetry-haskell++common options+  default-language: Haskell2010+  default-extensions:+    BangPatterns+    BlockArguments+    DataKinds+    DeriveGeneric+    DerivingStrategies+    DerivingVia+    FlexibleInstances+    InstanceSigs+    LambdaCase+    MultiParamTypeClasses+    MultiWayIf+    NamedFieldPuns+    NumericUnderscores+    RecordWildCards+    ScopedTypeVariables+    TupleSections+    TypeApplications+    ViewPatterns+  ghc-options:+    -Wall+    -Wcompat+    -Widentities+    -Wincomplete-record-updates+    -Wincomplete-uni-patterns+    -Wpartial-fields+    -fhide-source-paths+    -ferror-spans+    -freverse-errors++library+  import: options+  build-depends:+    base >= 4.12 && < 5,+    aeson,+    async,+    bytestring,+    clock >= 0.8,+    containers,+    directory,+    exceptions,+    filepath,+    ghc-events,+    hashable,+    http-client,+    http-client-tls,+    http-types,+    opentelemetry >= 0.3.0,+    random >= 1.1,+    scientific,+    splitmix,+    stm,+    text,+    unordered-containers++  hs-source-dirs: src+  exposed-modules:+    OpenTelemetry.Common+    OpenTelemetry.Tracer+    OpenTelemetry.Debug+    OpenTelemetry.ChromeExporter+    OpenTelemetry.EventlogStreaming_Internal+    OpenTelemetry.ZipkinExporter++test-suite unit-tests+  import: options+  type:       exitcode-stdio-1.0+  main-is:    Main.hs+  other-modules:+    TestCommon+    TestPropagation+  hs-source-dirs: unit-tests+  build-depends:+    base,+    bytestring,+    tasty,+    tasty-quickcheck,+    tasty-hunit,+    tasty-discover,+    opentelemetry >= 0.3.0,+    opentelemetry-extra++executable eventlog-to-zipkin+  import: options+  main-is:    Main.hs+  hs-source-dirs: exe/eventlog-to-zipkin+  build-depends:+    base,+    async,+    exceptions,+    clock,+    filepath,+    http-client,+    http-client-tls,+    opentelemetry >= 0.3.0,+    opentelemetry-extra,+    text++executable eventlog-to-chrome+  import: options+  main-is:    Main.hs+  hs-source-dirs: exe/eventlog-to-chrome+  build-depends:+    base,+    exceptions,+    clock,+    opentelemetry >= 0.3.0,+    opentelemetry-extra,
+ src/OpenTelemetry/ChromeExporter.hs view
@@ -0,0 +1,61 @@+{-# LANGUAGE OverloadedStrings #-}++module OpenTelemetry.ChromeExporter where++import Data.Function+import qualified Data.HashMap.Strict as HM+import Data.List (intersperse)+import qualified Data.Text as T+import OpenTelemetry.Common+import OpenTelemetry.Exporter+import OpenTelemetry.SpanContext+import System.IO+import Text.Printf+import Text.Read++showValue :: TagValue -> String+showValue (StringTagValue s) = show s+showValue (IntTagValue i) = show i+showValue _ = "\"unknown\""++showSpan :: Span -> String+showSpan s@(Span {..}) =+  let (TId tid) = spanTraceId s+      threadId = case HM.lookup "thread_id" spanTags of+        Just (StringTagValue (T.stripPrefix "ThreadId " -> Just (readMaybe . T.unpack -> Just t))) -> t+        Just (IntTagValue t) -> t+        _ -> 1+      meta :: String+      meta =+        spanTags+          & HM.toList+          & map (\(k, v) -> ["\"", T.unpack k, "\":", showValue v])+          & ([printf "\"traceId\":\"%x\"" tid] :)+          & intersperse [","]+          & concat+          & concat+   in printf+        "{\"ph\":\"B\",\"name\":\"%s\",\"pid\":1,\"ts\":%d,\"tid\":%d,\"args\":{%s}},{\"ph\":\"E\",\"name\":\"%s\",\"pid\":1,\"ts\":%d,\"tid\":%d},"+        spanOperation+        (div spanStartedAt 1000)+        threadId+        meta+        spanOperation+        (div spanFinishedAt 1000)+        threadId++createChromeSpanExporter :: FilePath -> IO (Exporter Span)+createChromeSpanExporter path = do+  f <- openFile path WriteMode+  hPutStrLn f "["+  pure+    $! Exporter+      ( \sps -> do+          mapM_ (hPutStrLn f . showSpan) sps+          pure ExportSuccess+      )+      ( do+          hSeek f RelativeSeek (-2) -- overwrite the last comma+          hPutStrLn f "\n]"+          hClose f+      )
+ src/OpenTelemetry/Common.hs view
@@ -0,0 +1,84 @@+{-# LANGUAGE OverloadedStrings #-}++module OpenTelemetry.Common where++import qualified Data.HashMap.Strict as HM+import qualified Data.Text as T+import Data.Word+import OpenTelemetry.Exporter+import OpenTelemetry.SpanContext+import System.Clock++type Timestamp = Word64++data TagValue+  = StringTagValue !T.Text+  | BoolTagValue !Bool+  | IntTagValue !Int+  | DoubleTagValue !Double+  deriving (Eq, Show)++class ToTagValue a where+  toTagValue :: a -> TagValue++instance ToTagValue String where+  toTagValue = StringTagValue . T.pack++instance ToTagValue T.Text where+  toTagValue = StringTagValue++instance ToTagValue Bool where+  toTagValue = BoolTagValue++instance ToTagValue Int where+  toTagValue = IntTagValue++data Span+  = Span+      { spanContext :: {-# UNPACK #-} !SpanContext,+        spanOperation :: T.Text,+        spanStartedAt :: !Timestamp,+        spanFinishedAt :: !Timestamp,+        spanTags :: !(HM.HashMap T.Text TagValue),+        spanEvents :: [SpanEvent],+        spanStatus :: !SpanStatus,+        spanParentId :: Maybe SpanId+      }+  deriving (Show, Eq)++spanTraceId :: Span -> TraceId+spanTraceId Span {spanContext = SpanContext _ tid} = tid++spanId :: Span -> SpanId+spanId Span {spanContext = SpanContext sid _} = sid++data SpanEvent+  = SpanEvent+      { spanEventTimestamp :: !Timestamp,+        spanEventKey :: !T.Text,+        spanEventValue :: !T.Text+      }+  deriving (Show, Eq)++data SpanStatus = OK+  deriving (Show, Eq)++data Event+  = Event T.Text Timestamp+  deriving (Show, Eq)++data SpanProcessor+  = SpanProcessor+      { onStart :: Span -> IO (),+        onEnd :: Span -> IO ()+      }++data OpenTelemetryConfig+  = OpenTelemetryConfig+      { otcSpanExporter :: Exporter Span+      }++now64 :: IO Timestamp+now64 = do+  TimeSpec secs nsecs <- getTime Realtime+  pure $! fromIntegral secs * 1_000_000_000 + fromIntegral nsecs
+ src/OpenTelemetry/Debug.hs view
@@ -0,0 +1,31 @@+module OpenTelemetry.Debug where++import Control.Concurrent.STM+import Control.Monad.IO.Class+import System.Environment+import System.IO+import System.IO.Unsafe++dd_ :: Show a => String -> a -> IO ()+dd_ = unsafePerformIO $+  lookupEnv "OPENTELEMETRY_DEBUG" >>= \case+    Nothing -> pure $ \_ _ -> pure ()+    Just "0" -> pure $ \_ _ -> pure ()+    Just "false" -> pure $ \_ _ -> pure ()+    _ -> pure $ \s thing -> liftIO $ do+      hPutStr stderr (s <> ": ")+      hPutStrLn stderr (show thing)+{-# NOINLINE dd_ #-}++d_ :: String -> IO ()+d_ = unsafePerformIO $+  lookupEnv "OPENTELEMETRY_DEBUG" >>= \case+    Nothing -> pure $ \_ -> pure ()+    Just "0" -> pure $ \_ -> pure ()+    Just "false" -> pure $ \_ -> pure ()+    _ -> pure $ \s -> liftIO $ do+      hPutStrLn stderr s+{-# NOINLINE d_ #-}++inc :: Int -> TVar Int -> IO ()+inc amount counterVar = atomically $ modifyTVar counterVar (+ amount)
+ src/OpenTelemetry/EventlogStreaming_Internal.hs view
@@ -0,0 +1,231 @@+{-# LANGUAGE OverloadedStrings #-}++module OpenTelemetry.EventlogStreaming_Internal where++import Control.Concurrent (threadDelay)+import qualified Data.ByteString as B+import qualified Data.HashMap.Strict as HM+import qualified Data.IntMap as IM+import Data.List.NonEmpty as NE+import Data.Maybe+import qualified Data.Text as T+import Data.Word+import GHC.RTS.Events+import GHC.RTS.Events.Incremental+import GHC.Stack+import OpenTelemetry.Common hiding (Event, Timestamp)+import qualified OpenTelemetry.Common as OTel+import OpenTelemetry.Debug+import OpenTelemetry.Exporter+import OpenTelemetry.SpanContext+import System.IO+import qualified System.Random.SplitMix as R+import Text.Printf++work :: Timestamp -> Exporter Span -> Handle -> IO ()+work origin_timestamp exporter input = do+  d_ "Starting the eventlog reader"+  smgen <- R.initSMGen -- TODO(divanov): seed the random generator with something more random than current time+  go (initialState origin_timestamp smgen) decodeEventLog+  d_ "no more work"+  where+    go s (Produce event next) = do+      case evSpec event of+        Shutdown {} -> do+          d_ "Shutdown-like event detected"+        CapDelete {} -> do+          d_ "Shutdown-like event detected"+        CapsetDelete {} -> do+          d_ "Shutdown-like event detected"+        _ -> do+          -- d_ "go Produce"+          -- print (evTime event, evCap event, evSpec event)+          let (s', sps) = processEvent event s+          _ <- export exporter sps+          -- print s'+          mapM_ (d_ . ("emit " <>) . show) sps+          go s' next+    go s d@(Consume consume) = do+      -- d_ "go Consume"+      eof <- hIsEOF input+      case eof of+        False -> do+          chunk <- B.hGetSome input 4096+          -- printf "chunk = %d bytes\n" (B.length chunk)+          if B.null chunk+            then do+              -- d_ "chunk is null"+              threadDelay 1000 -- TODO(divanov): remove the sleep by replacing the hGetSome with something that blocks until data is available+              go s d+            else do+              -- d_ "chunk is not null"+              go s $ consume chunk+        True -> do+          d_ "EOF"+    go _ (Done _) = do+      d_ "go Done"+      pure ()+    go _ (Error _leftover err) = do+      d_ "go Error"+      d_ err++data State = S+  { originTimestamp :: Timestamp,+    threadMap :: IM.IntMap ThreadId,+    spanStacks :: HM.HashMap ThreadId (NonEmpty Span),+    traceMap :: HM.HashMap ThreadId TraceId,+    randomGen :: R.SMGen+  }+  deriving (Show)++initialState :: Word64 -> R.SMGen -> State+initialState timestamp = S timestamp mempty mempty mempty++processEvent :: Event -> State -> (State, [Span])+processEvent (Event ts ev m_cap) st@(S {..}) =+  let now = originTimestamp + ts+      m_thread_id = m_cap >>= flip IM.lookup threadMap+      m_trace_id = m_thread_id >>= flip HM.lookup traceMap+   in case (ev, m_cap, m_thread_id) of+        (WallClockTime {sec, nsec}, _, _) -> (st {originTimestamp = sec * 1_000_000_000 + fromIntegral nsec - ts}, [])+        (CreateThread new_tid, _, _) ->+          case m_trace_id of+            Just trace_id -> (st {traceMap = HM.insert new_tid trace_id traceMap}, [])+            _ -> (st, [])+        (RunThread tid, Just cap, _) ->+          (st {threadMap = IM.insert cap tid threadMap}, [])+        (StopThread _ tstatus, Just cap, _)+          | isTerminalThreadStatus tstatus -> (st {threadMap = IM.delete cap threadMap}, [])+        (StartGC, _, _) -> (pushGCSpans st now, [])+        (GCStatsGHC {gen}, _, _) -> (modifyAllSpans (setTag "gen" gen) st, [])+        (EndGC, _, _) -> popSpansAcrossAllThreads now st+        (HeapAllocated {allocBytes}, _, Just tid) ->+          (modifySpan tid (addEvent now "heap_alloc_bytes" (showT allocBytes)) st, [])+        (UserMessage {msg}, _, fromMaybe 1 -> tid) -> case T.words msg of+          ("ot1" : "begin" : "span" : name) ->+            (pushSpan tid (T.intercalate " " name) now st, [])+          ("ot1" : "end" : "span" : _) -> popSpan tid now st+          ("ot1" : "set" : "tag" : k : v) -> (modifySpan tid (setTag k (T.unwords v)) st, [])+          ["ot1", "set", "traceid", trace_id_text] ->+            let trace_id = TId (read ("0x" <> T.unpack trace_id_text))+             in ( (modifySpan tid (setTraceId trace_id) st)+                    { traceMap = HM.insert tid trace_id traceMap+                    },+                  []+                )+          ["ot1", "set", "spanid", span_id] ->+            (modifySpan tid (setSpanId (SId (read ("0x" <> T.unpack span_id)))) st, [])+          ["ot1", "set", "parent", trace_id_text, span_id_text] ->+            let trace_id = TId (read ("0x" <> T.unpack trace_id_text))+                sid = SId (read ("0x" <> T.unpack span_id_text))+             in ( (modifySpan tid (setParent trace_id sid) st)+                    { traceMap = HM.insert tid trace_id traceMap+                    },+                  []+                )+          ("ot1" : "add" : "event" : k : v) -> (modifySpan tid (addEvent now k (T.unwords v)) st, [])+          ("ot1" : rest) -> error $ printf "Unrecognized %s" (show rest)+          _ -> (st, [])+        _ -> (st, [])++setTag :: ToTagValue v => T.Text -> v -> Span -> Span+setTag k v sp =+  sp+    { spanTags = HM.insert k (toTagValue v) (spanTags sp)+    }++setSpanId :: SpanId -> Span -> Span+setSpanId sid sp =+  sp+    { spanContext = SpanContext sid (spanTraceId sp)+    }++setTraceId :: TraceId -> Span -> Span+setTraceId tid sp =+  sp+    { spanContext = SpanContext (spanId sp) tid+    }++setParent :: TraceId -> SpanId -> Span -> Span+setParent ptid psid sp =+  sp+    { spanParentId = Just psid,+      spanContext = SpanContext (spanId sp) ptid+    }++addEvent :: Timestamp -> T.Text -> T.Text -> Span -> Span+addEvent ts k v sp = sp {spanEvents = new_events}+  where+    new_events = ev : spanEvents sp+    ev = SpanEvent ts k v++modifyAllSpans :: (Span -> Span) -> State -> State+modifyAllSpans f st =+  st+    { spanStacks =+        fmap+          (\(sp :| sps) -> (f sp :| sps))+          (spanStacks st)+    }++modifySpan :: HasCallStack => ThreadId -> (Span -> Span) -> State -> State+modifySpan tid f st =+  st+    { spanStacks =+        HM.update (\(sp :| sps) -> Just (f sp :| sps)) tid (spanStacks st)+    }++pushSpan :: HasCallStack => ThreadId -> T.Text -> OTel.Timestamp -> State -> State+pushSpan tid name timestamp st = st {spanStacks = new_stacks, randomGen = new_randomGen, traceMap = new_traceMap}+  where+    maybe_parent = NE.head <$> HM.lookup tid (spanStacks st)+    new_stacks = HM.alter f tid (spanStacks st)+    f Nothing = Just $ sp :| []+    f (Just sps) = Just $ cons sp sps+    (sid, new_randomGen) = R.nextWord64 (randomGen st)+    (new_traceMap, trace_id) = case (maybe_parent, HM.lookup tid (traceMap st)) of+      (Just parent, _) -> (traceMap st, spanTraceId parent)+      (_, Just trace_id') -> (traceMap st, trace_id')+      _ -> let new_trace_id = TId sid in (HM.insert tid new_trace_id (traceMap st), new_trace_id)+    sp =+      Span+        { spanContext = SpanContext (SId sid) trace_id,+          spanOperation = name,+          spanStartedAt = timestamp,+          spanFinishedAt = 0,+          spanTags = HM.singleton "tid" (IntTagValue $ fromIntegral tid),+          spanEvents = mempty,+          spanStatus = OK,+          spanParentId = spanId <$> maybe_parent+        }++popSpan :: HasCallStack => ThreadId -> OTel.Timestamp -> State -> (State, [Span])+popSpan tid timestamp st = (st {spanStacks = new_stacks, traceMap = new_traceMap}, [sp {spanFinishedAt = timestamp}])+  where+    sp :| new_stack = fromMaybe (error $ printf "popSpan: missing span stack for thread %d" tid) $ HM.lookup tid (spanStacks st)+    (new_traceMap, new_stacks) = case new_stack of+      [] -> (HM.delete tid (traceMap st), HM.delete tid (spanStacks st))+      x : xs -> (traceMap st, HM.insert tid (x :| xs) (spanStacks st))++pushGCSpans :: HasCallStack => State -> OTel.Timestamp -> State+pushGCSpans st timestamp = foldr go st tids+  where+    tids = HM.keys (spanStacks st)+    go tid = pushSpan tid "gc" timestamp++popSpansAcrossAllThreads :: HasCallStack => OTel.Timestamp -> State -> (State, [Span])+popSpansAcrossAllThreads timestamp st = foldr go (st, []) tids+  where+    tids = HM.keys (spanStacks st)+    go tid (st', sps) =+      let (st'', sps') = popSpan tid timestamp st'+       in (st'', sps' <> sps)++isTerminalThreadStatus :: ThreadStopStatus -> Bool+isTerminalThreadStatus HeapOverflow = True+isTerminalThreadStatus StackOverflow = True+isTerminalThreadStatus ThreadFinished = True+isTerminalThreadStatus _ = False++showT :: Show a => a -> T.Text+showT = T.pack . show
+ src/OpenTelemetry/Tracer.hs view
@@ -0,0 +1,43 @@+module OpenTelemetry.Tracer where++import OpenTelemetry.Common+import qualified Data.HashMap.Strict as HM+import Data.List.NonEmpty as NE+import Data.Hashable++data Tracer threadId+  = Tracer+      { tracerSpanStacks :: !(HM.HashMap threadId (NE.NonEmpty Span))+      }+      deriving (Eq, Show)++tracerPushSpan :: (Eq tid, Hashable tid) => Tracer tid -> tid -> Span -> Tracer tid+tracerPushSpan t@(Tracer {..}) tid sp =+  case HM.lookup tid tracerSpanStacks of+    Nothing ->+      let !stacks = HM.insert tid (sp :| []) tracerSpanStacks+      in Tracer stacks+    Just sps ->+      let !stacks = HM.insert tid (sp <| sps) tracerSpanStacks+      in t { tracerSpanStacks = stacks }++tracerPopSpan :: (Eq tid, Hashable tid) => Tracer tid -> tid -> (Maybe Span, Tracer tid)+tracerPopSpan t@(Tracer {..}) tid =+  case HM.lookup tid tracerSpanStacks of+    Nothing -> (Nothing, t)+    Just (sp :| sps) ->+      let stacks =+            case NE.nonEmpty sps of+              Nothing -> HM.delete tid tracerSpanStacks+              Just sps' -> HM.insert tid sps' tracerSpanStacks+       in (Just sp, Tracer stacks)++tracerGetCurrentActiveSpan :: (Hashable tid, Eq tid) => Tracer tid -> tid -> Maybe Span+tracerGetCurrentActiveSpan (Tracer stacks) tid =+  case HM.lookup tid stacks of+    Nothing -> Nothing+    Just (sp NE.:| _) -> Just sp++createTracer :: (Hashable tid, Eq tid) => IO (Tracer tid)+createTracer = pure $ Tracer mempty+
+ src/OpenTelemetry/ZipkinExporter.hs view
@@ -0,0 +1,193 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TypeApplications #-}++module OpenTelemetry.ZipkinExporter where++-- Zipkin V2 protocol spec: https://github.com/openzipkin/zipkin-api/blob/master/zipkin2-api.yaml++import Control.Concurrent.Async+import Control.Concurrent.STM+import Control.Monad.IO.Class+import Data.Aeson+import qualified Data.HashMap.Strict as HM+import Data.Scientific+import qualified Data.Text as T+import Network.HTTP.Client+import Network.HTTP.Client.TLS+import Network.HTTP.Types+import OpenTelemetry.Common+import OpenTelemetry.Debug+import OpenTelemetry.Exporter+import OpenTelemetry.SpanContext+import System.IO.Unsafe+import Text.Printf++data ZipkinSpan = ZipkinSpan+  { zsConfig :: ZipkinConfig,+    zsSpan :: Span+  }++tagValue2text :: TagValue -> T.Text+tagValue2text tv = case tv of+  (StringTagValue s) -> s+  (BoolTagValue b) -> if b then "true" else "false"+  (IntTagValue i) -> T.pack $ show i+  (DoubleTagValue d) -> T.pack $ show (fromFloatDigits d)++instance ToJSON ZipkinSpan where+  -- FIXME(divanov): deduplicate+  toJSON (ZipkinSpan ZipkinConfig {..} s@(Span {..})) =+    let TId tid = spanTraceId s+        SId sid = spanId s+        ts = spanStartedAt `div` 1000+        duration = (spanFinishedAt - spanStartedAt) `div` 1000+     in object $+          [ "name" .= spanOperation,+            "traceId" .= T.pack (printf "%016x" tid),+            "id" .= T.pack (printf "%016x" sid),+            "timestamp" .= ts,+            "duration" .= duration,+            "localEndpoint" .= object ["serviceName" .= zServiceName],+            "tags"+              .= object+                ( [k .= v | (k, v) <- zGlobalTags]+                    <> [k .= tagValue2text v | (k, v) <- HM.toList spanTags]+                ),+            "annotations"+              .= [ object ["timestamp" .= (t `div` 1000), "value" .= v]+                   | SpanEvent t _ v <- spanEvents+                 ]+          ]+            <> (maybe [] (\(SId psid) -> ["parentId" .= psid]) spanParentId)++  toEncoding (ZipkinSpan ZipkinConfig {..} s@(Span {..})) =+    let TId tid = spanTraceId s+        SId sid = spanId s+        ts = spanStartedAt `div` 1000+        duration = (spanFinishedAt - spanStartedAt) `div` 1000+     in pairs+          ( "name" .= spanOperation+              <> "traceId" .= T.pack (printf "%016x" tid)+              <> "id" .= T.pack (printf "%016x" sid)+              <> "timestamp" .= ts+              <> "duration" .= duration+              <> "localEndpoint" .= object ["serviceName" .= zServiceName]+              <> "tags"+                .= object+                  ( [k .= v | (k, v) <- zGlobalTags]+                      <> [k .= tagValue2text v | (k, v) <- HM.toList spanTags]+                  )+              <> ( maybe+                     mempty+                     (\(SId psid) -> "parentId" .= T.pack (printf "%016x" psid))+                     spanParentId+                 )+              <> "annotations"+                .= [ object ["timestamp" .= (t `div` 1000), "value" .= v]+                     | SpanEvent t _ v <- spanEvents+                   ]+          )++data ZipkinConfig = ZipkinConfig+  { zEndpoint :: String,+    zServiceName :: T.Text,+    zGlobalTags :: [(T.Text, T.Text)],+    zGracefulShutdownTimeoutSeconds :: Word,+    zSpanQueueSize :: Word+  }++localhostZipkinConfig :: T.Text -> ZipkinConfig+localhostZipkinConfig service =+  ZipkinConfig+    { zEndpoint = "http://localhost:9411/api/v2/spans",+      zServiceName = service,+      zGlobalTags = mempty,+      zGracefulShutdownTimeoutSeconds = 5,+      zSpanQueueSize = 2048+    }++data ZipkinClient = ZipkinClient+  { zcConfig :: ZipkinConfig,+    zcSenderThread :: Async (),+    zcSenderQueue :: TBQueue Span,+    zcShutdownVar :: TVar Bool+  }++createZipkinSpanExporter :: MonadIO m => ZipkinConfig -> m (Exporter Span)+createZipkinSpanExporter cfg = liftIO do+  client <- mkClient cfg+  pure+    $! Exporter+      ( \sps -> do+          let q = zcSenderQueue client+          atomically $ do+            q_population <- fromIntegral <$> lengthTBQueue q+            let q_vacancy = fromIntegral (zSpanQueueSize (zcConfig client) - q_population)+            modifyTVar droppedSpanCountVar (\x -> x + length sps - q_vacancy)+            mapM_+              (writeTBQueue q)+              (take q_vacancy sps)+          pure ExportSuccess+      )+      ( do+          atomically $+            writeTVar (zcShutdownVar client) True+          wait (zcSenderThread client)+      )++mkClient :: ZipkinConfig -> IO ZipkinClient+mkClient cfg@(ZipkinConfig {..}) = do+  manager <- newManager tlsManagerSettings+  q <- newTBQueueIO (fromIntegral zSpanQueueSize)+  shutdown_var <- newTVarIO False+  sender <- async $ do+    let loop = do+          (must_shutdown, sps) <- atomically $ do+            must_shutdown <- readTVar shutdown_var+            sps <- flushTBQueue q+            case (must_shutdown, sps) of+              (False, []) -> retry+              _ -> pure (must_shutdown, sps)+          case sps of+            [] -> pure ()+            _ -> reportSpans zEndpoint manager cfg sps+          dd_ "must_shutdown" must_shutdown+          case must_shutdown of+            True -> pure ()+            False -> loop+    loop+  pure $! ZipkinClient cfg sender q shutdown_var++reportSpans :: String -> Manager -> ZipkinConfig -> [Span] -> IO ()+reportSpans endpoint httpManager cfg sps = do+  dd_ "reportSpans" sps+  let body = encode (map (ZipkinSpan cfg) sps)+      request =+        (parseRequest_ endpoint)+          { method = "POST",+            requestBody = RequestBodyLBS body,+            requestHeaders = [("Content-Type", "application/json")]+          }+  resp <- httpLbs request httpManager+  case statusCode (responseStatus resp) of+    s | s `elem` [200, 202] -> do+      inc 1 reportedSpanCountVar+      pure ()+    _ -> do+      -- TODO(divanov): handle failures+      inc 1 rejectedSpanCountVar+      dd_ "body" body+      dd_ "resp status" $ responseStatus resp+      dd_ "resp" $ responseBody resp++droppedSpanCountVar :: TVar Int+droppedSpanCountVar = unsafePerformIO $ newTVarIO 0+{-# NOINLINE droppedSpanCountVar #-}++reportedSpanCountVar :: TVar Int+reportedSpanCountVar = unsafePerformIO $ newTVarIO 0+{-# NOINLINE reportedSpanCountVar #-}++rejectedSpanCountVar :: TVar Int+rejectedSpanCountVar = unsafePerformIO $ newTVarIO 0+{-# NOINLINE rejectedSpanCountVar #-}
+ unit-tests/Main.hs view
@@ -0,0 +1,1 @@+{-# OPTIONS_GHC -F -pgmF tasty-discover #-}
+ unit-tests/TestCommon.hs view
@@ -0,0 +1,36 @@+{-# LANGUAGE OverloadedStrings #-}++module TestCommon where++import Data.Word+import OpenTelemetry.Common+import OpenTelemetry.SpanContext+import OpenTelemetry.Tracer++mkTestSpan :: Word64 -> Word64 -> Span+mkTestSpan sid tid =+  Span+    { spanContext = SpanContext (SId sid) (TId tid),+      spanStartedAt = 0,+      spanFinishedAt = 10,+      spanTags = mempty,+      spanEvents = mempty,+      spanStatus = OK,+      spanOperation = "foo",+      spanParentId = Nothing+    }++prop_tracer_push_does_something :: Word64 -> Word64 -> Int -> Bool+prop_tracer_push_does_something sid tid threadid =+  let tracer0 = Tracer mempty+      sp0 = mkTestSpan sid tid+      tracer1 = tracerPushSpan tracer0 threadid sp0+   in tracer0 /= tracer1++prop_tracer_push_pop :: Word64 -> Word64 -> Int -> Bool+prop_tracer_push_pop sid tid threadid =+  let tracer0 = Tracer mempty+      sp0 = mkTestSpan sid tid+      tracer1 = tracerPushSpan tracer0 threadid sp0+      (Just sp1, tracer2) = tracerPopSpan tracer1 threadid+   in tracer0 == tracer2 && sp0 == sp1
+ unit-tests/TestPropagation.hs view
@@ -0,0 +1,37 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TypeApplications #-}++module TestPropagation where++import qualified Data.ByteString.Char8 as BS+import Data.Word+import OpenTelemetry.Propagation+import OpenTelemetry.SpanContext+import Test.Tasty.HUnit+import Text.Printf++prop_render_parse_roundtrip_w3c :: Word64 -> Word64 -> Bool+prop_render_parse_roundtrip_w3c sid tid =+  let c = SpanContext (SId sid) (TId tid)+   in Just c == propagateFromHeaders w3cTraceContext @String (propagateToHeaders w3cTraceContext c)++prop_render_parse_roundtrip_b3 :: Word64 -> Word64 -> Bool+prop_render_parse_roundtrip_b3 sid tid =+  let c = SpanContext (SId sid) (TId tid)+   in Just c == propagateFromHeaders b3 @String (propagateToHeaders b3 c)++unit_render_smoke :: Assertion+unit_render_smoke =+  propagateToHeaders w3cTraceContext (SpanContext (SId 1) (TId 2)) @?= [("traceparent", "00-2-1-00")]++unit_parse_smoke :: Assertion+unit_parse_smoke =+  propagateFromHeaders w3cTraceContext @String [("traceparent", "00-2-1-00")] @?= Just (SpanContext (SId 1) (TId 2))++prop_fromHex :: Word64 -> Bool+prop_fromHex x =+  Just x == fromHex (BS.pack $ printf "%x" x)++unit_fromHex_smoke :: Assertion+unit_fromHex_smoke =+  Just 1 @?= fromHex "1"