lightstep-haskell 0.1.0 → 0.1.1
raw patch · 8 files changed
+152/−142 lines, 8 files
Files
- cli/Main.hs +9/−2
- lightstep-haskell.cabal +2/−4
- src/LightStep.hs +0/−10
- src/LightStep/GlobalSharedMutableSingleton.hs +0/−57
- src/LightStep/HighLevel/IO.hs +131/−0
- src/LightStep/Internal/Debug.hs +4/−1
- src/LightStep/LowLevel.hs +6/−5
- src/LightStep/WithSpan.hs +0/−63
cli/Main.hs view
@@ -4,7 +4,7 @@ import Control.Concurrent.Async import Data.Maybe import qualified Data.Text as T-import LightStep+import LightStep.HighLevel.IO (LightStepConfig (..), setTag, withSingletonLightStep, withSpan) import System.Environment import System.Exit @@ -26,6 +26,12 @@ threadDelay 50000 setTag "lorem" "ipsum" threadDelay 60000+ withSpan "data->json" $ pure ()+ withSpan "json->yaml" $ pure ()+ withSpan "yaml->xml" $ pure ()+ withSpan "xml->protobuf" $ pure ()+ withSpan "protobuf->thrift" $ pure ()+ withSpan "thrift->base64" $ pure () threadDelay 70000 backend = withSpan "Background Data Science" $ do@@ -50,5 +56,6 @@ exitFailure host <- fromMaybe "ingest.lightstep.com" <$> lookupEnv "LIGHTSTEP_HOST" port <- maybe 443 read <$> lookupEnv "LIGHTSTEP_PORT"- let lsConfig = LightStepConfig host port token "helloworld"+ let lsConfig = LightStepConfig host port token "helloworld" 5 withSingletonLightStep lsConfig seriousBusinessMain+ putStrLn "All done"
lightstep-haskell.cabal view
@@ -1,7 +1,7 @@ cabal-version: 2.4 name: lightstep-haskell-version: 0.1.0+version: 0.1.1 synopsis: LightStep OpenTracing client library description: LightStep OpenTracing client library. Uses GRPC transport via proto-lens. category: Tools@@ -46,10 +46,8 @@ library import: options exposed-modules:- LightStep LightStep.LowLevel- LightStep.GlobalSharedMutableSingleton- LightStep.WithSpan+ LightStep.HighLevel.IO LightStep.Internal.Debug other-modules: Paths_lightstep_haskell
− src/LightStep.hs
@@ -1,10 +0,0 @@-{-# LANGUAGE OverloadedStrings #-}--module LightStep- ( module M,- )-where--import LightStep.GlobalSharedMutableSingleton as M-import LightStep.LowLevel as M-import LightStep.WithSpan as M
@@ -1,57 +0,0 @@-module LightStep.GlobalSharedMutableSingleton where--import Control.Concurrent.Async-import Control.Concurrent.STM-import Control.Monad.Catch-import Control.Monad.Except-import LightStep.Internal.Debug-import LightStep.LowLevel-import System.IO.Unsafe--{-# NOINLINE globalSharedMutableSingletonState #-}-globalSharedMutableSingletonState :: TBQueue Span-globalSharedMutableSingletonState = unsafePerformIO $ newTBQueueIO 100--withSingletonLightStep :: LightStepConfig -> IO () -> IO ()-withSingletonLightStep cfg action =- race_ (action >> waitUntilQueueIsEmpty) $ do- let work client = do- d_ "Getting more spans"- spans <- liftIO $ atomically $ do- x <- readTBQueue globalSharedMutableSingletonState- xs <- flushTBQueue globalSharedMutableSingletonState- pure (x : xs)- d_ $ "Got " <> show (length spans) <> " spans"- reportSpans client spans- d_ $ "Reported " <> show (length spans) <> " spans"- shutdown client = do- d_ "Getting the last spans before shutdown"- spans <- liftIO $ atomically $ flushTBQueue globalSharedMutableSingletonState- d_ $ "Got " <> show (length spans) <> " spans"- when (not $ null spans) $- reportSpans client spans- d_ $ "Reported " <> show (length spans) <> " spans"- closeClient client- d_ "Client closed"- runExceptT- $ bracket- (mkClient cfg)- shutdown- $ \client -> do- let loop = do- work client- loop- loop- pure ()--submitSpan :: Span -> IO ()-submitSpan = atomically . writeTBQueue globalSharedMutableSingletonState---- TODO: handle span batches larger that queue size-submitSpans :: Foldable f => f Span -> IO ()-submitSpans = atomically . mapM_ (writeTBQueue globalSharedMutableSingletonState)--waitUntilQueueIsEmpty :: IO ()-waitUntilQueueIsEmpty = atomically $ do- e <- isEmptyTBQueue globalSharedMutableSingletonState- when (not e) retry
+ src/LightStep/HighLevel/IO.hs view
@@ -0,0 +1,131 @@+module LightStep.HighLevel.IO + ( module LightStep.HighLevel.IO+ , module LightStep.LowLevel+ ) where++import Control.Concurrent+import Control.Concurrent.Async+import Control.Concurrent.STM+import Control.Lens+import Control.Monad.Catch+import Control.Monad.Except+import Data.Maybe+import Data.ProtoLens.Message (defMessage)+import LightStep.Internal.Debug+import LightStep.LowLevel+import Proto.Collector+import Proto.Collector_Fields+import System.IO.Unsafe+import qualified Data.HashMap.Strict as HM+import qualified Data.Text as T++{-# noinline globalSharedMutableSpanStacks #-}+globalSharedMutableSpanStacks :: MVar (HM.HashMap ThreadId [Span])+globalSharedMutableSpanStacks = unsafePerformIO (newMVar mempty)++withSpan :: T.Text -> IO a -> IO a+withSpan opName action =+ bracket+ (pushSpan opName)+ popSpan + (const action)++pushSpan :: T.Text -> IO ()+pushSpan opName = do+ sp <- startSpan opName+ tId <- myThreadId+ modifyMVar_ globalSharedMutableSpanStacks $ \stacks ->+ case fromMaybe [] (HM.lookup tId stacks) of+ [] -> do+ pure $! HM.insert tId [sp] stacks+ (psp : _) ->+ let !sp' = sp+ & references .~ [defMessage & relationship .~ Reference'CHILD_OF & spanContext .~ (psp ^. spanContext)]+ & spanContext.traceId .~ (psp ^. spanContext.traceId)+ in pure $! HM.update (Just . (sp' :)) tId stacks++popSpan :: () -> IO ()+popSpan () = do+ tId <- myThreadId+ sp <- modifyMVar globalSharedMutableSpanStacks+ (\stacks ->+ let (sp : sps) = stacks HM.! tId+ !stacks' = HM.insert tId sps stacks+ in pure (stacks', sp))+ sp' <- finishSpan sp+ submitSpan sp'++modifyCurrentSpan :: (Span -> Span) -> IO ()+modifyCurrentSpan f = do+ tId <- myThreadId+ modifyMVar_ globalSharedMutableSpanStacks+ (\stacks ->+ let (sp : sps) = stacks HM.! tId+ !stacks' = HM.insert tId (f sp : sps) stacks+ in pure stacks')++setTag :: T.Text -> T.Text -> IO ()+setTag k v =+ modifyCurrentSpan (tags %~ (<> [defMessage & key .~ k & stringValue .~ v]))+++{-# NOINLINE globalSharedMutableSingletonState #-}+globalSharedMutableSingletonState :: TBQueue Span+globalSharedMutableSingletonState = unsafePerformIO $ newTBQueueIO 100++withSingletonLightStep :: LightStepConfig -> IO () -> IO ()+withSingletonLightStep cfg action = do+ doneVar <- newEmptyMVar+ race_ (action >> waitUntilDone (lsGracefulShutdownTimeoutSeconds cfg) doneVar) $ do+ let work client = do+ d_ "Getting more spans"+ spans <- liftIO $ atomically $ do+ x <- readTBQueue globalSharedMutableSingletonState+ xs <- flushTBQueue globalSharedMutableSingletonState+ pure (x : xs)+ d_ $ "Got " <> show (length spans) <> " spans"+ reportSpansRes <- try (reportSpans client spans)+ case reportSpansRes of+ Right () ->+ d_ $ "Reported " <> show (length spans) <> " spans"+ Left (err :: SomeException) ->+ d_ $ "Error while reporting spans: " <> show err+ shutdown client = do+ d_ "Getting the last spans before shutdown"+ spans <- liftIO $ atomically $ flushTBQueue globalSharedMutableSingletonState+ d_ $ "Got " <> show (length spans) <> " spans"+ when (not $ null spans) $+ reportSpans client spans+ d_ $ "Reported " <> show (length spans) <> " spans"+ closeClient client+ d_ "Client closed"+ liftIO $ putMVar doneVar ()+ runExceptT+ $ bracket+ (mkClient cfg)+ shutdown+ $ \client -> do+ let loop = do+ work client+ loop+ loop+ pure ()++submitSpan :: Span -> IO ()+submitSpan = atomically . writeTBQueue globalSharedMutableSingletonState++-- TODO: handle span batches larger that queue size+submitSpans :: Foldable f => f Span -> IO ()+submitSpans = atomically . mapM_ (writeTBQueue globalSharedMutableSingletonState)++waitUntilDone :: Int -> MVar () -> IO ()+waitUntilDone timeoutSeconds doneVar =+ race_+ ( do+ threadDelay $ 1_000_000 * timeoutSeconds+ d_ "waitUntilDone: timeout"+ )+ ( do+ takeMVar doneVar+ d_ "waitUntilDone: done"+ )
src/LightStep/Internal/Debug.hs view
@@ -1,5 +1,6 @@ module LightStep.Internal.Debug where +import Control.Concurrent import Control.Monad.IO.Class import System.Environment import System.IO@@ -12,4 +13,6 @@ Nothing -> pure $ const (pure ()) Just "0" -> pure $ const (pure ()) Just "false" -> pure $ const (pure ())- _ -> pure (liftIO . hPutStrLn stderr)+ _ -> pure $ \s -> liftIO $ do+ tid <- myThreadId+ hPutStrLn stderr (show tid <> ": " <> s)
src/LightStep/LowLevel.hs view
@@ -25,7 +25,8 @@ { lsHostName :: HostName, lsPort :: PortNumber, lsToken :: T.Text,- lsServiceName :: T.Text+ lsServiceName :: T.Text,+ lsGracefulShutdownTimeoutSeconds :: Int } reportSpans :: LightStepClient -> [Span] -> ExceptT ClientError IO ()@@ -44,9 +45,9 @@ pure () mkClient :: LightStepConfig -> ClientIO LightStepClient-mkClient (LightStepConfig host port token serviceName) = do- grpc <- setupGrpcClient ((grpcClientConfigSimple host port True) {_grpcClientConfigCompression = compression})- pure $ LightStepClient grpc token rep+mkClient LightStepConfig {..} = do+ grpc <- setupGrpcClient ((grpcClientConfigSimple lsHostName lsPort True) {_grpcClientConfigCompression = compression})+ pure $ LightStepClient grpc lsToken rep where compression = if False then gzip else uncompressed rep =@@ -54,7 +55,7 @@ & reporterId .~ 2 & tags .~ [ defMessage & key .~ "span.kind" & stringValue .~ "server",- defMessage & key .~ "lightstep.component_name" & stringValue .~ serviceName,+ defMessage & key .~ "lightstep.component_name" & stringValue .~ lsServiceName, defMessage & key .~ "lightstep.tracer_platform" & stringValue .~ "haskell" ]
− src/LightStep/WithSpan.hs
@@ -1,63 +0,0 @@-module LightStep.WithSpan where--import Data.Maybe-import Data.ProtoLens.Message (defMessage)-import Control.Lens-import Control.Concurrent-import Control.Monad.Catch-import qualified Data.Text as T-import LightStep.GlobalSharedMutableSingleton-import LightStep.LowLevel-import System.IO.Unsafe-import Proto.Collector-import Proto.Collector_Fields-import qualified Data.HashMap.Strict as HM--{-# noinline globalSharedMutableSpanStacks #-}-globalSharedMutableSpanStacks :: MVar (HM.HashMap ThreadId [Span])-globalSharedMutableSpanStacks = unsafePerformIO (newMVar mempty)--withSpan :: T.Text -> IO a -> IO a-withSpan opName action =- bracket- (pushSpan opName)- popSpan - (const action)--pushSpan :: T.Text -> IO ()-pushSpan opName = do- sp <- startSpan opName- tId <- myThreadId- modifyMVar_ globalSharedMutableSpanStacks $ \stacks ->- case fromMaybe [] (HM.lookup tId stacks) of- [] -> do- pure $! HM.insert tId [sp] stacks- (psp : _) ->- let !sp' = sp- & references .~ [defMessage & relationship .~ Reference'CHILD_OF & spanContext .~ (psp ^. spanContext)]- & spanContext.traceId .~ (psp ^. spanContext.traceId)- in pure $! HM.update (Just . (sp' :)) tId stacks--popSpan :: () -> IO ()-popSpan () = do- tId <- myThreadId- sp <- modifyMVar globalSharedMutableSpanStacks- (\stacks ->- let (sp : sps) = stacks HM.! tId- !stacks' = HM.insert tId sps stacks- in pure (stacks', sp))- sp' <- finishSpan sp- submitSpan sp'--modifyCurrentSpan :: (Span -> Span) -> IO ()-modifyCurrentSpan f = do- tId <- myThreadId- modifyMVar_ globalSharedMutableSpanStacks- (\stacks ->- let (sp : sps) = stacks HM.! tId- !stacks' = HM.insert tId (f sp : sps) stacks- in pure stacks')--setTag :: T.Text -> T.Text -> IO ()-setTag k v =- modifyCurrentSpan (tags %~ (<> [defMessage & key .~ k & stringValue .~ v]))