diff --git a/cli/Main.hs b/cli/Main.hs --- a/cli/Main.hs +++ b/cli/Main.hs @@ -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" diff --git a/lightstep-haskell.cabal b/lightstep-haskell.cabal --- a/lightstep-haskell.cabal +++ b/lightstep-haskell.cabal @@ -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 diff --git a/src/LightStep.hs b/src/LightStep.hs deleted file mode 100644 --- a/src/LightStep.hs +++ /dev/null @@ -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 diff --git a/src/LightStep/GlobalSharedMutableSingleton.hs b/src/LightStep/GlobalSharedMutableSingleton.hs deleted file mode 100644 --- a/src/LightStep/GlobalSharedMutableSingleton.hs +++ /dev/null @@ -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 diff --git a/src/LightStep/HighLevel/IO.hs b/src/LightStep/HighLevel/IO.hs new file mode 100644 --- /dev/null +++ b/src/LightStep/HighLevel/IO.hs @@ -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" + ) diff --git a/src/LightStep/Internal/Debug.hs b/src/LightStep/Internal/Debug.hs --- a/src/LightStep/Internal/Debug.hs +++ b/src/LightStep/Internal/Debug.hs @@ -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) diff --git a/src/LightStep/LowLevel.hs b/src/LightStep/LowLevel.hs --- a/src/LightStep/LowLevel.hs +++ b/src/LightStep/LowLevel.hs @@ -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" ] diff --git a/src/LightStep/WithSpan.hs b/src/LightStep/WithSpan.hs deleted file mode 100644 --- a/src/LightStep/WithSpan.hs +++ /dev/null @@ -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]))