diff --git a/examples/readme/Main.hs b/examples/readme/Main.hs --- a/examples/readme/Main.hs +++ b/examples/readme/Main.hs @@ -2,11 +2,7 @@ import Control.Concurrent import Control.Concurrent.Async -import Data.Maybe -import qualified Data.Text as T -import LightStep.HighLevel.IO (LightStepConfig (..), LogEntryKey (..), addLog, setTag, withSingletonLightStep, withSpan) -import System.Environment -import System.Exit +import LightStep.HighLevel.IO (LogEntryKey (..), addLog, getEnvConfig, setTag, withSingletonLightStep, withSpan) seriousBusinessMain :: IO () seriousBusinessMain = concurrently_ frontend backend @@ -50,14 +46,11 @@ main :: IO () main = do - token <- - lookupEnv "LIGHTSTEP_TOKEN" >>= \case - Just t -> pure $ T.pack t - Nothing -> do - putStrLn "LIGHTSTEP_TOKEN environment variable not defined" - exitFailure - host <- fromMaybe "ingest.lightstep.com" <$> lookupEnv "LIGHTSTEP_HOST" - port <- maybe 443 read <$> lookupEnv "LIGHTSTEP_PORT" - let lsConfig = LightStepConfig host port token "helloworld" 5 + -- Construct a config from env variables + -- - LIGHTSTEP_ACCESS_TOKEN + -- - LIGHTSTEP_HOST (optional) + -- - LIGHTSTEP_PORT (optional) + -- - LIGHTSTEP_SERVICE (optional) + Just lsConfig <- getEnvConfig withSingletonLightStep lsConfig seriousBusinessMain putStrLn "All done" diff --git a/examples/req/Main.hs b/examples/req/Main.hs --- a/examples/req/Main.hs +++ b/examples/req/Main.hs @@ -1,13 +1,9 @@ -{-# language OverloadedStrings #-} +{-# LANGUAGE OverloadedStrings #-} import Control.Concurrent -import Data.Maybe import LightStep.HighLevel.IO import LightStep.Propagation (b3headersForSpanContext) import Network.HTTP.Req -import qualified Data.Text as T -import System.Environment -import System.Exit clientMain :: IO () clientMain = do @@ -17,18 +13,11 @@ Just ctx <- currentSpanContext let opts = port 8736 <> foldMap (\(k, v) -> header k v) (b3headersForSpanContext ctx) url = http "127.0.0.1" /: "test" - runReq defaultHttpConfig $ req GET url NoReqBody ignoreResponse opts + _ <- runReq defaultHttpConfig $ req GET url NoReqBody ignoreResponse opts + pure () threadDelay 1_000_000 main :: IO () main = do - token <- - lookupEnv "LIGHTSTEP_TOKEN" >>= \case - Just t -> pure $ T.pack t - Nothing -> do - putStrLn "LIGHTSTEP_TOKEN environment variable not defined" - exitFailure - host <- fromMaybe "ingest.lightstep.com" <$> lookupEnv "LIGHTSTEP_HOST" - lsport <- maybe 443 read <$> lookupEnv "LIGHTSTEP_PORT" - let lsConfig = LightStepConfig host lsport token "example-req-client" 5 + Just lsConfig <- getEnvConfig withSingletonLightStep lsConfig clientMain diff --git a/examples/wai/Main.hs b/examples/wai/Main.hs --- a/examples/wai/Main.hs +++ b/examples/wai/Main.hs @@ -1,34 +1,37 @@ {-# LANGUAGE OverloadedStrings #-} -import Data.Maybe (fromMaybe) -import System.Environment -import System.Exit -import LightStep.HighLevel.IO (LightStepConfig (..), withSingletonLightStep) -import Network.Wai.Middleware.LightStep -import Network.Wai +import qualified Data.ByteString.Lazy.Char8 as LBS +import GHC.Stats +import LightStep.Diagnostics +import LightStep.HighLevel.IO (getEnvConfig, withSingletonLightStep) import Network.HTTP.Types +import Network.Wai import Network.Wai.Handler.Warp (run) -import qualified Data.Text as T +import Network.Wai.Middleware.LightStep robustScalableProductionReadyMicroservice :: Application robustScalableProductionReadyMicroservice = \_req respond -> do - putStrLn "Disrupting the industry" - respond $ responseLBS - status200 - [("Content-Type", "text/plain")] - "It's all data!" + rtsStats <- getRTSStats + diagnostics <- getDiagnostics + respond $ + responseLBS + status200 + [("Content-Type", "text/plain")] + ( LBS.pack $ + unlines + [ "RAM usage: " <> show (max_live_bytes rtsStats `div` 1_000) <> " KB", + "Reconnect count: " <> show (diagReconnectCount diagnostics), + "Started span count: " <> show (diagStartedSpanCount diagnostics), + "Finished span count: " <> show (diagFinishedSpanCount diagnostics), + "Reported span count: " <> show (diagReportedSpanCount diagnostics), + "Rejected span count: " <> show (diagRejectedSpanCount diagnostics), + "Dropped span count: " <> show (diagDroppedSpanCount diagnostics) + ] + ) main :: IO () main = do - token <- - lookupEnv "LIGHTSTEP_TOKEN" >>= \case - Just t -> pure $ T.pack t - Nothing -> do - putStrLn "LIGHTSTEP_TOKEN environment variable not defined" - exitFailure - host <- fromMaybe "ingest.lightstep.com" <$> lookupEnv "LIGHTSTEP_HOST" - port <- maybe 443 read <$> lookupEnv "LIGHTSTEP_PORT" - let lsConfig = LightStepConfig host port token "example-wai-service" 5 - microserviceWithTracing = tracingMiddleware robustScalableProductionReadyMicroservice + Just lsConfig <- getEnvConfig + let microserviceWithTracing = tracingMiddleware robustScalableProductionReadyMicroservice withSingletonLightStep lsConfig $ run 8736 microserviceWithTracing 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.4.5 +version: 0.5.0 synopsis: LightStep OpenTracing client library description: LightStep OpenTracing client library. Uses GRPC transport via proto-lens. category: Tools @@ -48,9 +48,11 @@ library import: options exposed-modules: - LightStep.LowLevel + LightStep.Config LightStep.HighLevel.IO LightStep.Internal.Debug + LightStep.LowLevel + LightStep.Diagnostics LightStep.Propagation Network.Wai.Middleware.LightStep other-modules: @@ -119,7 +121,7 @@ ghc-options: -threaded -rtsopts - -with-rtsopts=-N + "-with-rtsopts=-N -T -s" main-is: Main.hs hs-source-dirs: examples/wai @@ -130,6 +132,7 @@ , text , wai , warp + , bytestring executable lightstep-haskell-req-example import: options, example @@ -146,3 +149,18 @@ , http-types , text , req >= 3.0.0 + +executable lightstep-haskell-stress-test + import: options + ghc-options: + -threaded + -rtsopts + "-with-rtsopts=-N -T" + main-is: Main.hs + hs-source-dirs: + stress-test + build-depends: + base >=4.9 && <5 + , lightstep-haskell + , text + , async diff --git a/src/LightStep/Config.hs b/src/LightStep/Config.hs new file mode 100644 --- /dev/null +++ b/src/LightStep/Config.hs @@ -0,0 +1,45 @@ +{-# LANGUAGE OverloadedStrings #-} + +module LightStep.Config where + +import Control.Monad.IO.Class +import Data.Foldable +import Data.Maybe +import qualified Data.Text as T +import Network.HTTP2.Client +import System.Environment +import System.IO + +data LightStepConfig + = LightStepConfig + { lsHostName :: HostName, + lsPort :: PortNumber, + lsToken :: T.Text, + lsServiceName :: T.Text, + lsGlobalTags :: [(T.Text, T.Text)], + lsGracefulShutdownTimeoutSeconds :: Int + } + +lookupOneOfEnvs :: [String] -> IO (Maybe String) +lookupOneOfEnvs names = asum <$> traverse lookupEnv names + +getEnvTagsWithPrefix :: T.Text -> IO [(T.Text, T.Text)] +getEnvTagsWithPrefix prefix = + mapMaybe unprefix <$> getEnvironment + where + unprefix ((T.stripPrefix prefix . T.pack) -> Just k, v) = Just (k, T.pack v) + unprefix _ = Nothing + +getEnvConfig :: MonadIO m => m (Maybe LightStepConfig) +getEnvConfig = liftIO $ do + maybe_token_from_env <- lookupOneOfEnvs ["LIGHTSTEP_TOKEN", "LIGHTSTEP_ACCESS_TOKEN", "OPENTRACING_LIGHTSTEP_ACCESS_TOKEN"] + global_tags <- getEnvTagsWithPrefix "OPENTRACING_TAG_" + case maybe_token_from_env of + Just t -> do + host <- fromMaybe "ingest.lightstep.com" <$> lookupOneOfEnvs ["LIGHTSTEP_HOST", "OPENTRACING_LIGHTSTEP_COLLECTOR_HOST"] + port <- maybe 443 read <$> lookupOneOfEnvs ["LIGHTSTEP_PORT", "OPENTRACING_LIGHTSTEP_COLLECTOR_PORT"] + service <- fromMaybe "example-haskell-service" <$> lookupOneOfEnvs ["LIGHTSTEP_SERVICE", "OPENTRACING_LIGHTSTEP_COMPONENT_NAME"] + pure $ Just $ LightStepConfig host port (T.pack t) (T.pack service) global_tags 5 + Nothing -> do + hPutStrLn stderr "LIGHTSTEP_ACCESS_TOKEN environment variable not defined" + pure Nothing diff --git a/src/LightStep/Diagnostics.hs b/src/LightStep/Diagnostics.hs new file mode 100644 --- /dev/null +++ b/src/LightStep/Diagnostics.hs @@ -0,0 +1,54 @@ +module LightStep.Diagnostics where + +import Control.Concurrent.STM +import System.IO.Unsafe + +inc :: Int -> TVar Int -> IO () +inc n var = atomically $ do + value <- readTVar var + writeTVar var $! value + n + +reconnectCountVar :: TVar Int +reconnectCountVar = unsafePerformIO $ newTVarIO 0 +{-# NOINLINE reconnectCountVar #-} + +startedSpanCountVar :: TVar Int +startedSpanCountVar = unsafePerformIO $ newTVarIO 0 +{-# NOINLINE startedSpanCountVar #-} + +finishedSpanCountVar :: TVar Int +finishedSpanCountVar = unsafePerformIO $ newTVarIO 0 +{-# NOINLINE finishedSpanCountVar #-} + +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 #-} + +data Diagnostics + = Diagnostics + { diagReconnectCount :: !Int, + diagStartedSpanCount :: !Int, + diagFinishedSpanCount :: !Int, + diagDroppedSpanCount :: !Int, + diagReportedSpanCount :: !Int, + diagRejectedSpanCount :: !Int + } + +getDiagnostics :: IO Diagnostics +getDiagnostics = + atomically $ + Diagnostics + <$> readTVar reconnectCountVar + <*> readTVar startedSpanCountVar + <*> readTVar finishedSpanCountVar + <*> readTVar droppedSpanCountVar + <*> readTVar reportedSpanCountVar + <*> readTVar rejectedSpanCountVar diff --git a/src/LightStep/HighLevel/IO.hs b/src/LightStep/HighLevel/IO.hs --- a/src/LightStep/HighLevel/IO.hs +++ b/src/LightStep/HighLevel/IO.hs @@ -3,6 +3,7 @@ module LightStep.HighLevel.IO ( module LightStep.HighLevel.IO, module LightStep.LowLevel, + module LightStep.Config, ) where @@ -18,11 +19,13 @@ import qualified Data.Text as T import GHC.Conc import LightStep.Internal.Debug +import LightStep.Diagnostics import LightStep.LowLevel import Proto.Collector import Proto.Collector_Fields import System.IO.Unsafe import System.Timeout +import LightStep.Config {-# NOINLINE globalSharedMutableSpanStacks #-} globalSharedMutableSpanStacks :: MVar (HM.HashMap ThreadId [Span]) @@ -37,12 +40,16 @@ showLogEntryKey Stack = T.pack "stack" showLogEntryKey (Custom x) = x -withSpan :: MonadIO m => MonadMask m => T.Text -> m a -> m a +withSpan :: forall m a. MonadIO m => MonadMask m => T.Text -> m a -> m a withSpan opName action = - bracket + let onExc :: SomeException -> m () + onExc ex = do + setTag "error" "true" + setTag "error.message" (T.pack $ displayException ex) + in bracket (pushSpan opName) popSpan - (const (onException action (setTag "error" "true"))) + (const (withException action onExc)) pushSpan :: MonadIO m => T.Text -> m () pushSpan opName = liftIO $ do @@ -51,7 +58,9 @@ modifyMVar_ globalSharedMutableSpanStacks $ \stacks -> case fromMaybe [] (HM.lookup tId stacks) of [] -> do - pure $! HM.insert tId [sp] stacks + let !sp' = sp + & tags %~ (<> [defMessage & key .~ "thread" & stringValue .~ T.pack (show tId)]) + pure $! HM.insert tId [sp'] stacks (psp : _) -> let !sp' = sp @@ -158,6 +167,7 @@ written <- atomically $ tryWriteTBQueue globalSharedMutableSingletonState sp when (not written) $ do d_ "internal span queue is full, dropping the span" + inc 1 droppedSpanCountVar tryWriteTBQueue :: TBQueue a -> a -> STM Bool tryWriteTBQueue q a = isFullTBQueue q >>= \case diff --git a/src/LightStep/LowLevel.hs b/src/LightStep/LowLevel.hs --- a/src/LightStep/LowLevel.hs +++ b/src/LightStep/LowLevel.hs @@ -12,17 +12,19 @@ import Control.Lens hiding (op) import Data.ProtoLens.Message (defMessage) import qualified Data.Text as T +import Data.Version (showVersion) +import LightStep.Config +import LightStep.Diagnostics import LightStep.Internal.Debug import Network.GRPC.Client import Network.GRPC.Client.Helpers import Network.GRPC.HTTP2.ProtoLens import Network.HTTP2.Client +import Paths_lightstep_haskell (version) import Proto.Collector import Proto.Collector_Fields import Proto.Google.Protobuf.Timestamp_Fields import System.Timeout -import Data.Version (showVersion) -import Paths_lightstep_haskell (version) data LightStepClient = LightStepClient @@ -32,15 +34,6 @@ lscConfig :: LightStepConfig } -data LightStepConfig - = LightStepConfig - { lsHostName :: HostName, - lsPort :: PortNumber, - lsToken :: T.Text, - lsServiceName :: T.Text, - lsGracefulShutdownTimeoutSeconds :: Int - } - reportSpans :: LightStepClient -> [Span] -> IO () reportSpans client@LightStepClient {..} sps = do let tryOnce = do @@ -56,6 +49,7 @@ & reporter .~ lscReporter ) req `withException` (\err -> d_ $ "reportSpans failed: " <> show (err :: SomeException)) + inc (length sps) reportedSpanCountVar ret <- tryOnce ret2 <- case ret of Nothing -> do @@ -64,6 +58,10 @@ -- one retry after reconnect tryOnce _ -> pure ret + case ret2 of + Nothing -> + inc (length sps) rejectedSpanCountVar + _ -> pure () d_ $ show ret2 pure () @@ -79,10 +77,14 @@ defMessage & reporterId .~ 2 & tags - .~ [ defMessage & key .~ "lightstep.component_name" & stringValue .~ lsServiceName, - defMessage & key .~ "lightstep.tracer_platform" & stringValue .~ "haskell", - defMessage & key .~ "lightstep.tracer_version" & stringValue .~ (T.pack $ showVersion version) - ] + .~ ( [ defMessage & key .~ "lightstep.component_name" & stringValue .~ lsServiceName, + defMessage & key .~ "lightstep.tracer_platform" & stringValue .~ "haskell", + defMessage & key .~ "lightstep.tracer_version" & stringValue .~ (T.pack $ showVersion version) + ] + <> [ defMessage & key .~ k & stringValue .~ v + | (k, v) <- lsGlobalTags + ] + ) makeGrpcClient :: LightStepClient -> IO GrpcClient makeGrpcClient client = do @@ -105,19 +107,21 @@ reconnectClient :: LightStepClient -> IO () reconnectClient client@LightStepClient {lscGrpcVar} = do d_ "reconnectClient begin" + inc 1 reconnectCountVar newClient <- makeGrpcClient client oldClient <- swapMVar lscGrpcVar newClient - runExceptT $ close oldClient + _ <- runExceptT $ close oldClient d_ "reconnectClient end" closeClient :: LightStepClient -> IO () closeClient LightStepClient {lscGrpcVar} = do grpc <- readMVar lscGrpcVar - runExceptT $ close grpc + _ <- runExceptT $ close grpc pure () startSpan :: T.Text -> IO Span startSpan op = do + inc 1 startedSpanCountVar nanosSinceEpoch <- getTime <$> now -- FIXME: make those ids randomer let sid = fromIntegral nanosSinceEpoch @@ -135,6 +139,7 @@ finishSpan :: Span -> IO Span finishSpan sp = do + inc 1 finishedSpanCountVar nanosSinceEpoch <- getTime <$> now let dur = (nanosSinceEpoch - (sp ^. startTimestamp . seconds) * 1_000_000_000 - fromIntegral (sp ^. startTimestamp . nanos)) diff --git a/stress-test/Main.hs b/stress-test/Main.hs new file mode 100644 --- /dev/null +++ b/stress-test/Main.hs @@ -0,0 +1,70 @@ +{-# LANGUAGE OverloadedStrings #-} + +import Control.Concurrent +import Control.Concurrent.Async +import Control.Monad +import GHC.Stats +import LightStep.HighLevel.IO (LogEntryKey (..), addLog, getEnvConfig, setTag, withSingletonLightStep, withSpan) +import System.Exit +import Text.Printf + +seriousBusinessMain :: IO () +seriousBusinessMain = concurrently_ frontend backend + where + frontend = + withSpan "RESTful API" $ do + threadDelay 10000 + setTag "foo" "bar" + withSpan "Kafka" $ do + threadDelay 20000 + setTag "foo" "baz" + threadDelay 30000 + withSpan "GraphQL" $ do + threadDelay 40000 + setTag "foo" "quux" + addLog Event "monkey-job" + addLog (Custom "foo") "bar" + withSpan "Mongodb" $ do + 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 + threadDelay 10000 + withSpan "Tensorflow" $ do + threadDelay 100000 + setTag "learning" "deep" + withSpan "Torch" $ do + threadDelay 100000 + setTag "learning" "very_deep" + withSpan "Hadoop" $ do + threadDelay 100000 + setTag "learning" "super_deep" + +reportMemoryUsage :: IO () +reportMemoryUsage = do + RTSStats {..} <- getRTSStats + let GCDetails {..} = gc + printf "max_live_bytes %d\n" max_live_bytes + printf "gcdetails_live_bytes %d\n" gcdetails_live_bytes + when (max_live_bytes > 10_000_000) $ do + putStrLn "Ate too much memory" + exitFailure + +main :: IO () +main = do + Just lsConfig <- getEnvConfig + withSingletonLightStep lsConfig $ do + replicateM_ 10 $ do + reportMemoryUsage + replicateM_ 1000 $ do + _ <- async seriousBusinessMain + threadDelay 10000 + putStrLn "All done"