[Hackathon] otel-instrumentation

deploy/dockerephemeral/docker-compose.yaml

@@ -277,6 +277,17 @@ services:
     networks:
       - demo_wire
 
+  grafana-lgtm:
+    container_name: demo_wire_grafana_lgtm
+    image: grafana/otel-lgtm:latest
+    ports:
+      - "3000:3000"
+      - "3100:3100"
+      - "4317:4317"
+      - "4318:4318"
+    networks:
+      - demo_wire
+
   postgres:
     container_name: postgres
     image: 'postgres:17-alpine'

integration/test/Testlib/ModService.hs

@@ -59,8 +59,10 @@ import qualified Data.Text.IO as Text
 import Data.Traversable
 import Data.Word
 import qualified Data.Yaml as Yaml
+import Data.Time.Clock.POSIX (getPOSIXTime)
 import GHC.Stack
 import qualified Network.HTTP.Client as HTTP
+import Network.HTTP.Client (RequestBody(..))
 import System.Directory (copyFile, createDirectoryIfMissing, doesDirectoryExist, doesFileExist, listDirectory, removeDirectoryRecursive, removeFile)
 import System.Exit
 import System.FilePath
@@ -509,8 +511,12 @@ withProcess artifacts resource service = do
           createServiceInstance params = do
             (_, Just stdoutHdl, Just stderrHdl, ph) <- createProcess (proc exe params) {cwd = cwd, std_out = CreatePipe, std_err = CreatePipe}
             let colorize = fromMaybe id (lookup execName processColors)
-            void $ forkIO $ logToConsoleDebug (Just stdOut) colorize prefix stdoutHdl
-            void $ forkIO $ logToConsoleDebug (Just stdErr) colorize prefix stderrHdl
+            mgr <- HTTP.newManager HTTP.defaultManagerSettings
+            let lokiUrl = "http://localhost:3100"
+                lokiStdoutPusher = Just $ makeLokiPusher mgr lokiUrl execName domain "stdout"
+                lokiStderrPusher = Just $ makeLokiPusher mgr lokiUrl execName domain "stderr"
+            void $ forkIO $ logToConsoleDebug (Just stdOut) colorize prefix stdoutHdl lokiStdoutPusher
+            void $ forkIO $ logToConsoleDebug (Just stdErr) colorize prefix stderrHdl lokiStderrPusher
             writeIORef phRef (Just ph)
             mkProcessInstance service ph
 
@@ -584,11 +590,38 @@ lookupServiceConfig service artifacts =
   fromMaybe (error $ "missing backend artifact for service " <> show service) $
     Map.lookup service artifacts.serviceConfigs
 
+makeLokiPusher :: HTTP.Manager -> String -> String -> String -> String -> (String -> IO ())
+makeLokiPusher mgr lokiUrl service domain streamName line =
+  E.catch sendToLoki (\(_ :: E.SomeException) -> pure ())
+  where
+    sendToLoki = do
+      now <- getPOSIXTime
+      let ns = show (round (now * 1e9) :: Integer)
+          streamObj = object
+            [ "service" .= service
+            , "domain"  .= domain
+            , "source"  .= streamName
+            ]
+          bodyObj = object
+            [ "streams" .= [ object
+                [ "stream" .= streamObj
+                , "values" .= [[ns, line]]
+                ]
+            ]
+            ]
+      req <- HTTP.parseRequest (lokiUrl <> "/loki/api/v1/push")
+      let req' = req
+            { HTTP.method = "POST"
+            , HTTP.requestBody = RequestBodyLBS (encode bodyObj)
+            , HTTP.requestHeaders = [("Content-Type", "application/json")]
+            }
+      void (HTTP.httpLbs req' mgr)
+
 logToConsole :: (String -> String) -> String -> Handle -> IO ()
-logToConsole = logToConsoleDebug Nothing
+logToConsole colorize prefix hdl = logToConsoleDebug Nothing colorize prefix hdl Nothing
 
-logToConsoleDebug :: Maybe (IORef [String]) -> (String -> String) -> String -> Handle -> IO ()
-logToConsoleDebug mOutput colorize prefix hdl = do
+logToConsoleDebug :: Maybe (IORef [String]) -> (String -> String) -> String -> Handle -> Maybe (String -> IO ()) -> IO ()
+logToConsoleDebug mOutput colorize prefix hdl mSink = do
   let go =
         do
           line <- hGetLine hdl
@@ -597,6 +630,9 @@ logToConsoleDebug mOutput colorize prefix hdl = do
             Nothing -> pure ()
             Just output -> do
               modifyIORef output (<> [line])
+          case mSink of
+            Nothing -> pure ()
+            Just sink -> sink line
           go
           `E.catch` (\(_ :: E.IOException) -> pure ())
   go

libs/cassandra-util/cassandra-util.cabal

@@ -82,6 +82,7 @@ library
     , cql-io                >=0.14
     , cql-io-tinylog
     , exceptions            >=0.6
+    , hs-opentelemetry-sdk
     , HsOpenSSL
     , imports
     , lens                  >=4.4
@@ -93,6 +94,7 @@ library
     , text                  >=0.11
     , time                  >=1.4
     , tinylog               >=0.7
+    , unordered-containers
     , uuid
     , wreq                  >=0.2
 

libs/cassandra-util/default.nix

@@ -11,6 +11,7 @@
 , cql-io
 , cql-io-tinylog
 , exceptions
+, hs-opentelemetry-sdk
 , HsOpenSSL
 , imports
 , lens
@@ -39,6 +40,7 @@ mkDerivation {
     cql-io
     cql-io-tinylog
     exceptions
+    hs-opentelemetry-sdk
     HsOpenSSL
     imports
     lens

libs/cassandra-util/src/Cassandra.hs

@@ -58,6 +58,7 @@ import Cassandra.Exec as C
   ( BatchM,
     Client,
     ClientState,
+    runClientTraced,
     GeneralPaginationState (..),
     MonadClient,
     Page (..),
@@ -83,6 +84,8 @@ import Cassandra.Exec as C
     pwsHasMore,
     query,
     query1,
+    query1Traced,
+    queryTraced,
     result,
     retry,
     runClient,
@@ -91,6 +94,7 @@ import Cassandra.Exec as C
     shutdown,
     trans,
     write,
+    writeTraced,
     x1,
     x5,
   )

libs/cassandra-util/src/Cassandra/Exec.hs

@@ -33,11 +33,19 @@ module Cassandra.Exec
     paginateWithStateC,
     paramsPagingState,
     pwsHasMore,
+    runClientTraced,
+    queryTraced,
+    query1Traced,
+    writeTraced,
     module C,
   )
 where
 
-import Cassandra.CQL (Consistency, R)
+import Cassandra.CQL (Consistency, R, W)
+import qualified Data.HashMap.Strict as HashMap
+import qualified Data.Text.Lazy as LT
+import OpenTelemetry.Trace (SpanArguments (..), defaultSpanArguments, getGlobalTracerProvider, inSpan, makeTracer, toAttribute, tracerOptions)
+import Database.CQL.Protocol (QueryString (..))
 import Control.Monad.Catch
 import Data.Conduit
 -- We only use these locally.
@@ -48,6 +56,7 @@ import Database.CQL.Protocol (Error, QueryParams (QueryParams), Tuple, pagingSta
 import Database.CQL.Protocol qualified as Protocol
 import Imports hiding (init)
 
+
 params :: Consistency -> a -> QueryParams a
 params c p = QueryParams c False p Nothing Nothing Nothing Nothing
 {-# INLINE params #-}
@@ -162,3 +171,43 @@ paramsPagingState c p n state = QueryParams c False p (Just n) state Nothing Not
 
 pwsHasMore :: PageWithState a b -> Bool
 pwsHasMore = isJust . pwsState
+
+-- | Like 'runClient' but wraps the call in an OpenTelemetry span.
+-- This creates a 'cassandra.query' span for tracing and performance monitoring.
+-- The span includes timing information for the entire Client operation.
+--
+-- Example:
+--   result <- runClientTraced cassandraState $ do
+--     retry x1 $ query1 cql $ params LocalQuorum args
+runClientTraced :: ClientState -> Client a -> IO a
+runClientTraced st action = do
+  tp <- getGlobalTracerProvider
+  let tracer = makeTracer tp "cassandra" tracerOptions
+  inSpan tracer "cassandra.query" defaultSpanArguments $ runClient st action
+
+cqlSpanArgs :: PrepQuery k a b -> SpanArguments
+cqlSpanArgs q =
+  let QueryString cql = queryString q
+   in defaultSpanArguments
+        { attributes = HashMap.fromList [("db.statement", toAttribute (LT.toStrict cql))] }
+
+-- | Like 'query' but adds a 'db.statement' OTel span attribute with the CQL template.
+queryTraced :: (MonadClient m, Tuple a, Tuple b) => PrepQuery R a b -> QueryParams a -> m [b]
+queryTraced q p = liftClient $ do
+  tp <- liftIO getGlobalTracerProvider
+  let tracer = makeTracer tp "cassandra" tracerOptions
+  inSpan tracer "cassandra.query" (cqlSpanArgs q) $ query q p
+
+-- | Like 'query1' but adds a 'db.statement' OTel span attribute with the CQL template.
+query1Traced :: (MonadClient m, Tuple a, Tuple b) => PrepQuery R a b -> QueryParams a -> m (Maybe b)
+query1Traced q p = liftClient $ do
+  tp <- liftIO getGlobalTracerProvider
+  let tracer = makeTracer tp "cassandra" tracerOptions
+  inSpan tracer "cassandra.query" (cqlSpanArgs q) $ query1 q p
+
+-- | Like 'write' but adds a 'db.statement' OTel span attribute with the CQL template.
+writeTraced :: (MonadClient m, Tuple a) => PrepQuery W a () -> QueryParams a -> m ()
+writeTraced q p = liftClient $ do
+  tp <- liftIO getGlobalTracerProvider
+  let tracer = makeTracer tp "cassandra" tracerOptions
+  inSpan tracer "cassandra.query" (cqlSpanArgs q) $ write q p

libs/wire-otel/src/Wire/OpenTelemetry.hs

@@ -2,6 +2,7 @@
 -- instrument http/2 request similarly to how it was done for http-client here:
 -- https://github.com/iand675/hs-opentelemetry/blob/0b3c854a88113fc18df8561202a76357e593a294/instrumentation/http-client/src/OpenTelemetry/Instrumentation/HttpClient/Raw.hs#L60
 -- This is non-trivial because http/2 forgets the structure on the out objs.
+{-# LANGUAGE DataKinds #-}
 module Wire.OpenTelemetry
   ( -- * instrumentation helpers
     withTracer,
@@ -21,18 +22,26 @@ import OpenTelemetry.Context.ThreadLocal (getContext)
 import OpenTelemetry.Instrumentation.HttpClient.Raw
 import OpenTelemetry.Trace
 import UnliftIO (MonadUnliftIO, bracket, liftIO)
+import OpenTelemetry.Resource (Resource, mkResource, (.=))
+
 
 -- | a tracer for a service like brig, galley, etc.
-withTracer :: (MonadUnliftIO m) => (Tracer -> m r) -> m r
-withTracer k =
+withTracer :: Text -> (Tracer -> IO r) -> IO r
+withTracer serviceName k =
   bracket
-    (liftIO initializeGlobalTracerProvider)
+    (liftIO $ do
+      let svcResource = mkResource ["service.name" .= serviceName] :: Resource 'Nothing
+      (processors, opts) <- getTracerProviderInitializationOptions' svcResource
+      tp <- createTracerProvider processors opts
+      setGlobalTracerProvider tp
+      pure tp
+    )
     shutdownTracerProvider
     \tp -> k $ makeTracer tp "wire-otel" tracerOptions
 
 -- | like 'withTracer' but in 'Codensity'
-withTracerC :: Codensity IO Tracer
-withTracerC = Codensity withTracer
+withTracerC :: Text -> Codensity IO Tracer
+withTracerC serviceName = Codensity (withTracer serviceName)
 
 -- | instrument a http client
 withClientInstrumentation ::

libs/wire-subsystems/default.nix

@@ -57,6 +57,7 @@
 , hasql-th
 , hasql-transaction
 , hex
+, hs-opentelemetry-sdk
 , hscim
 , HsOpenSSL
 , hspec
@@ -196,6 +197,7 @@ mkDerivation {
     hasql-th
     hasql-transaction
     hex
+    hs-opentelemetry-sdk
     hscim
     HsOpenSSL
     hspec
@@ -326,6 +328,7 @@ mkDerivation {
     hasql-th
     hasql-transaction
     hex
+    hs-opentelemetry-sdk
     hscim
     HsOpenSSL
     hspec

libs/wire-subsystems/src/Wire/ConversationStore/Postgres.hs

@@ -198,8 +198,12 @@ upsertConversationImpl lcnv nc = do
 
 deleteConversationImpl :: (PGConstraints r) => ConvId -> Sem r ()
 deleteConversationImpl cid =
-  runStatement cid delete
+  runStatementTraced "db.mutation.delete_conversation" deleteQuery cid delete
   where
+    deleteQuery =
+      "DELETE FROM conversation \
+      \WHERE id = ($1 :: uuid)"
+
     delete :: Hasql.Statement ConvId ()
     delete =
       -- cascades to shadow convs, subconvs, local and remote members
@@ -594,8 +598,15 @@ deleteTeamConversationImpl :: (PGConstraints r) => TeamId -> ConvId -> Sem r ()
 deleteTeamConversationImpl _ = deleteConversationImpl
 
 getTeamConversationImpl :: (PGConstraints r) => TeamId -> ConvId -> Sem r (Maybe ConvId)
-getTeamConversationImpl tid cid = runStatement (tid, cid) select
+getTeamConversationImpl tid cid =
+  runStatementTraced "db.query.get_team_conversation" selectQuery (tid, cid) select
   where
+    selectQuery =
+      "SELECT (id :: uuid) \
+      \FROM conversation \
+      \WHERE team = ($1 :: uuid) \
+      \AND id = ($2 :: uuid)"
+
     select :: Hasql.Statement (TeamId, ConvId) (Maybe ConvId)
     select =
       dimapPG
@@ -607,8 +618,13 @@ getTeamConversationImpl tid cid = runStatement (tid, cid) select
 
 getTeamConversationsImpl :: (PGConstraints r) => TeamId -> Sem r [ConvId]
 getTeamConversationsImpl tid =
-  runStatement tid select
+  runStatementTraced "db.query.get_team_conversations" selectQuery tid select
   where
+    selectQuery =
+      "SELECT (id :: uuid) \
+      \FROM conversation \
+      \WHERE team = ($1 :: uuid)"
+
     select :: Hasql.Statement TeamId [ConvId]
     select =
       dimapPG
@@ -619,8 +635,12 @@ getTeamConversationsImpl tid =
 
 deleteTeamConversationsImpl :: (PGConstraints r) => TeamId -> Sem r ()
 deleteTeamConversationsImpl tid =
-  runStatement tid delete
+  runStatementTraced "db.mutation.delete_team_conversations" deleteQuery tid delete
   where
+    deleteQuery =
+      "DELETE FROM conversation \
+      \WHERE team = ($1 :: uuid)"
+
     delete :: Hasql.Statement TeamId ()
     delete =
       lmapPG