diff --git a/deploy/dockerephemeral/docker-compose.yaml b/deploy/dockerephemeral/docker-compose.yaml
index f96f794b019..bf745fbfcb0 100644
--- a/deploy/dockerephemeral/docker-compose.yaml
+++ b/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'
diff --git a/integration/test/Testlib/ModService.hs b/integration/test/Testlib/ModService.hs
index 399aa950ee2..04cf69502ad 100644
--- a/integration/test/Testlib/ModService.hs
+++ b/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
diff --git a/libs/cassandra-util/cassandra-util.cabal b/libs/cassandra-util/cassandra-util.cabal
index dc3e3abe855..c0457d68b47 100644
--- a/libs/cassandra-util/cassandra-util.cabal
+++ b/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
diff --git a/libs/cassandra-util/default.nix b/libs/cassandra-util/default.nix
index 5f3ee2a1a3a..2a2ad3ba94d 100644
--- a/libs/cassandra-util/default.nix
+++ b/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
diff --git a/libs/cassandra-util/src/Cassandra.hs b/libs/cassandra-util/src/Cassandra.hs
index 24406334813..fc82cc614c7 100644
--- a/libs/cassandra-util/src/Cassandra.hs
+++ b/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,
)
diff --git a/libs/cassandra-util/src/Cassandra/Exec.hs b/libs/cassandra-util/src/Cassandra/Exec.hs
index 9d5c7a03d52..e06c1487974 100644
--- a/libs/cassandra-util/src/Cassandra/Exec.hs
+++ b/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
diff --git a/libs/wire-otel/src/Wire/OpenTelemetry.hs b/libs/wire-otel/src/Wire/OpenTelemetry.hs
index aa76dc500b4..fe487d1d008 100644
--- a/libs/wire-otel/src/Wire/OpenTelemetry.hs
+++ b/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 ::
diff --git a/libs/wire-subsystems/default.nix b/libs/wire-subsystems/default.nix
index 292b9a59947..d5a21bb20c3 100644
--- a/libs/wire-subsystems/default.nix
+++ b/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
diff --git a/libs/wire-subsystems/src/Wire/ConversationStore/Postgres.hs b/libs/wire-subsystems/src/Wire/ConversationStore/Postgres.hs
index ea5f3c82a3e..1a754069696 100644
--- a/libs/wire-subsystems/src/Wire/ConversationStore/Postgres.hs
+++ b/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
diff --git a/libs/wire-subsystems/src/Wire/Postgres.hs b/libs/wire-subsystems/src/Wire/Postgres.hs
index d18de5f10a8..667a08dd772 100644
--- a/libs/wire-subsystems/src/Wire/Postgres.hs
+++ b/libs/wire-subsystems/src/Wire/Postgres.hs
@@ -15,6 +15,8 @@
-- You should have received a copy of the GNU Affero General Public License along
-- with this program. If not, see .
+{-# LANGUAGE OverloadedStrings #-}
+
module Wire.Postgres
( -- | This module provides a composable DSL for constructing postgres
-- statements. Queries are assembled from smaller 'QueryFragment's that
@@ -37,12 +39,15 @@ module Wire.Postgres
-- resulting statement can be run with something like @runStatement ()@.
-- * Runners
+ dbStatementSpanArguments,
runStatement,
+ runStatementTraced,
runSession,
runSessionWithRetry,
runTransaction,
runTransactionWithRetry,
runPipeline,
+ runPipelineTraced,
parseCount,
PGConstraints,
@@ -69,6 +74,7 @@ where
import Control.Monad.Trans.State
import Data.Functor.Contravariant
+import Data.HashMap.Strict qualified as HashMap
import Data.Id
import Data.Text qualified as Text
import Data.Text.Encoding qualified as Text
@@ -85,6 +91,7 @@ import Hasql.Transaction (Transaction)
import Hasql.Transaction.Sessions
import Hasql.Transaction.Sessions qualified as Transaction
import Imports
+import OpenTelemetry.Trace (SpanArguments (..), defaultSpanArguments, getGlobalTracerProvider, inSpan, makeTracer, toAttribute, tracerOptions)
import Polysemy
import Polysemy.Error (Error, throw)
import Polysemy.Input
@@ -154,6 +161,39 @@ runStatement ::
runStatement a stmt =
runSessionWithRetry $ statement a stmt
+dbStatementSpanArguments :: Text -> SpanArguments
+dbStatementSpanArguments queryTemplate =
+ defaultSpanArguments
+ { attributes = HashMap.fromList [("db.statement", toAttribute queryTemplate)]
+ }
+
+-- | Runs a 'Statement' using the 'Hasql.Pool' with OpenTelemetry tracing.
+-- Always retries on server errors due to admin intervention.
+--
+-- Span name: query operation name (e.g., "db.query.select_users")
+runStatementTraced ::
+ (PGConstraints r) =>
+ Text ->
+ Text ->
+ a ->
+ Statement a b ->
+ Sem r b
+runStatementTraced queryName queryTemplate a stmt = do
+ pool <- input
+ let sess = statement a stmt
+ result <- liftIO $ useWithResetAndRetryTraced queryName queryTemplate pool sess
+ either throw pure result
+
+-- | Internal: Wraps useWithResetAndRetry with OpenTelemetry tracing
+useWithResetAndRetryTraced :: Text -> Text -> Pool -> Session a -> IO (Either UsageError a)
+useWithResetAndRetryTraced queryName queryTemplate pool sess = do
+ tp <- getGlobalTracerProvider
+ let tracer = makeTracer tp "postgres" tracerOptions
+ inSpan tracer queryName (dbStatementSpanArguments queryTemplate) $
+ useWithResetAndRetry pool sess
+
+
+
-- | Runs a 'Transaction' using the 'Hasql.Pool'. Unlike
-- 'runTransactionWithRetry' it doesn't retry on server errors due to admin
-- intervention. Things like server restart.
@@ -189,6 +229,14 @@ runPipeline ::
runPipeline p =
runSession $ pipeline p
+-- | Runs a 'Pipeline' using the 'Hasql.Pool' with OpenTelemetry tracing.
+-- Always retries on server errors due to admin intervention.
+runPipelineTraced :: (PGConstraints r) => Text -> Text -> Pipeline a -> Sem r a
+runPipelineTraced queryName queryTemplate p = do
+ pool <- input
+ result <- liftIO $ useWithResetAndRetryTraced queryName queryTemplate pool (pipeline p)
+ either throw pure result
+
class PostgresValue a where
postgresType :: Text
postgresValue :: a -> Enc.Value ()
diff --git a/libs/wire-subsystems/src/Wire/UserStore/Cassandra.hs b/libs/wire-subsystems/src/Wire/UserStore/Cassandra.hs
index bc919cd854d..739275e6e55 100644
--- a/libs/wire-subsystems/src/Wire/UserStore/Cassandra.hs
+++ b/libs/wire-subsystems/src/Wire/UserStore/Cassandra.hs
@@ -120,7 +120,7 @@ getUserAuthenticationInfoImpl uid = fmap f <$> retry x1 (query1 authSelect (para
getUsersImpl :: [UserId] -> Client [StoredUser]
getUsersImpl usrs =
map asRecord
- <$> retry x1 (query selectUsers (params LocalQuorum (Identity usrs)))
+ <$> retry x1 (queryTraced selectUsers (params LocalQuorum (Identity usrs)))
doesUserExistImpl :: UserId -> Client Bool
doesUserExistImpl uid =
@@ -329,7 +329,7 @@ freeHandleImpl uid h = do
lookupHandleImpl :: Consistency -> Handle -> Client (Maybe UserId)
lookupHandleImpl consistencyLevel h = do
(runIdentity =<<)
- <$> retry x1 (query1 handleSelect (params consistencyLevel (Identity h)))
+ <$> retry x1 (query1Traced handleSelect (params consistencyLevel (Identity h)))
deleteUserImpl :: User -> Client ()
deleteUserImpl user = do
diff --git a/libs/wire-subsystems/src/Wire/UserStore/Postgres.hs b/libs/wire-subsystems/src/Wire/UserStore/Postgres.hs
index 9192317b407..0bf243febc2 100644
--- a/libs/wire-subsystems/src/Wire/UserStore/Postgres.hs
+++ b/libs/wire-subsystems/src/Wire/UserStore/Postgres.hs
@@ -30,7 +30,7 @@ import Wire.API.Team.Feature (FeatureStatus)
import Wire.API.User hiding (DeleteUser)
import Wire.API.User.RichInfo
import Wire.API.User.Search
-import Wire.Postgres
+import Wire.Postgres (PGConstraints, runPipeline, runPipelineTraced, runStatement, runStatementTraced, runTransaction)
import Wire.Sem.Logger
import Wire.StoredUser
import Wire.UserStore
@@ -248,7 +248,7 @@ deleteAssetsStatement =
getUsersImpl :: (PGConstraints r, Member TinyLog r) => [UserId] -> Sem r [StoredUser]
getUsersImpl uids = do
(userRows, deletedUserIds, assetRows) <-
- runPipeline $
+ runPipelineTraced "db.query.get_users" "pipeline: select_users, select_deleted_users, select_assets" $
(,,)
<$> Pipeline.statement uids selectUsers
<*> Pipeline.statement uids selectDeletedUsers
@@ -547,7 +547,7 @@ lookupNameImpl uid = runStatement uid select
|]
lookupHandleImpl :: (PGConstraints r) => Handle -> Sem r (Maybe UserId)
-lookupHandleImpl h = runStatement h select
+lookupHandleImpl h = runStatementTraced "db.query.lookup_handle" "SELECT id :: uuid FROM wire_user WHERE handle = $1 :: text" h select
where
select :: Hasql.Statement Handle (Maybe UserId)
select =
diff --git a/libs/wire-subsystems/test/unit/Wire/PostgresSpec.hs b/libs/wire-subsystems/test/unit/Wire/PostgresSpec.hs
new file mode 100644
index 00000000000..71fe2e8368c
--- /dev/null
+++ b/libs/wire-subsystems/test/unit/Wire/PostgresSpec.hs
@@ -0,0 +1,23 @@
+-- This file is part of the Wire Server implementation.
+--
+-- Copyright (C) 2026 Wire Swiss GmbH
+--
+-- This program is free software: you can redistribute it and/or modify it under
+-- the terms of the GNU Affero General Public License as published by the Free
+-- Software Foundation, either version 3 of the License, or (at your option) any
+-- later version.
+
+module Wire.PostgresSpec where
+
+import Data.HashMap.Strict qualified as HashMap
+import Imports
+import OpenTelemetry.Trace (SpanArguments (..), toAttribute)
+import Test.Hspec
+import Wire.Postgres (dbStatementSpanArguments)
+
+spec :: Spec
+spec =
+ describe "dbStatementSpanArguments" $ do
+ it "stores the query template in db.statement" $ do
+ attributes (dbStatementSpanArguments "SELECT 1")
+ `shouldBe` HashMap.fromList @Text [("db.statement", toAttribute ("SELECT 1" :: Text))]
diff --git a/libs/wire-subsystems/wire-subsystems.cabal b/libs/wire-subsystems/wire-subsystems.cabal
index 881b3965068..729bde21b9a 100644
--- a/libs/wire-subsystems/wire-subsystems.cabal
+++ b/libs/wire-subsystems/wire-subsystems.cabal
@@ -133,6 +133,7 @@ common common-all
, hasql-th
, hasql-transaction
, hex
+ , hs-opentelemetry-sdk
, hscim
, HsOpenSSL
, hspec
@@ -507,6 +508,7 @@ library
, HaskellNet
, HaskellNet-SSL
, hex
+ , hs-opentelemetry-sdk
, HsOpenSSL
, hspec
, html-entities
@@ -631,6 +633,7 @@ test-suite wire-subsystems-tests
Wire.MockInterpreters.UserSubsystem
Wire.MockInterpreters.VerificationCodeStore
Wire.NotificationSubsystem.InterpreterSpec
+ Wire.PostgresSpec
Wire.PropertySubsystem.InterpreterSpec
Wire.RateLimited.InterpreterSpec
Wire.SAMLEmailSubsystem.InterpreterSpec
diff --git a/services/brig/src/Brig/API/OAuth.hs b/services/brig/src/Brig/API/OAuth.hs
index 8738656b71b..2afcb635c9c 100644
--- a/services/brig/src/Brig/API/OAuth.hs
+++ b/services/brig/src/Brig/API/OAuth.hs
@@ -398,26 +398,26 @@ deleteOAuthRefreshTokenById (tUnqualified -> uid) cid tokenId req = do
-- DB
deleteOAuthClient' :: (MonadClient m) => OAuthClientId -> m ()
-deleteOAuthClient' cid = retry x5 . write q $ params LocalQuorum (Identity cid)
+deleteOAuthClient' cid = retry x5 . writeTraced q $ params LocalQuorum (Identity cid)
where
q :: PrepQuery W (Identity OAuthClientId) ()
q = "DELETE FROM oauth_client WHERE id = ?"
updateOAuthClient' :: (MonadClient m) => OAuthClientId -> OAuthApplicationName -> RedirectUrl -> m ()
-updateOAuthClient' cid name uri = retry x5 . write q $ params LocalQuorum (name, uri, cid)
+updateOAuthClient' cid name uri = retry x5 . writeTraced q $ params LocalQuorum (name, uri, cid)
where
q :: PrepQuery W (OAuthApplicationName, RedirectUrl, OAuthClientId) ()
q = {- `IF EXISTS`, but that requires benchmarking -} "UPDATE oauth_client SET name = ?, redirect_uri = ? WHERE id = ?"
insertOAuthClient :: (MonadClient m) => OAuthClientId -> OAuthApplicationName -> RedirectUrl -> Password -> m ()
-insertOAuthClient cid name uri pw = retry x5 . write q $ params LocalQuorum (cid, name, uri, pw)
+insertOAuthClient cid name uri pw = retry x5 . writeTraced q $ params LocalQuorum (cid, name, uri, pw)
where
q :: PrepQuery W (OAuthClientId, OAuthApplicationName, RedirectUrl, Password) ()
q = "INSERT INTO oauth_client (id, name, redirect_uri, secret) VALUES (?, ?, ?, ?)"
lookupOauthClient :: (MonadClient m) => OAuthClientId -> m (Maybe OAuthClient)
lookupOauthClient cid = do
- mNameUrl <- retry x5 . query1 q $ params LocalQuorum (Identity cid)
+ mNameUrl <- retry x5 . query1Traced q $ params LocalQuorum (Identity cid)
pure $ mNameUrl <&> uncurry (OAuthClient cid)
where
q :: PrepQuery R (Identity OAuthClientId) (OAuthApplicationName, RedirectUrl)
@@ -426,7 +426,7 @@ lookupOauthClient cid = do
insertOAuthAuthorizationCode :: (MonadClient m) => Word64 -> OAuthAuthorizationCode -> OAuthClientId -> UserId -> OAuthScopes -> RedirectUrl -> OAuthCodeChallenge -> m ()
insertOAuthAuthorizationCode ttl code cid uid scope uri chal = do
let cqlScope = C.Set (Set.toList (unOAuthScopes scope))
- retry x5 . write q $ params LocalQuorum (code, cid, uid, cqlScope, uri, chal, fromIntegral ttl)
+ retry x5 . writeTraced q $ params LocalQuorum (code, cid, uid, cqlScope, uri, chal, fromIntegral ttl)
where
q :: PrepQuery W (OAuthAuthorizationCode, OAuthClientId, UserId, C.Set OAuthScope, RedirectUrl, OAuthCodeChallenge, Int32) ()
q = fromString $ "INSERT INTO oauth_auth_code (code, client, user, scope, redirect_uri, code_challenge) VALUES (?, ?, ?, ?, ?, ?) USING TTL ?"
@@ -436,14 +436,14 @@ lookupAndDeleteByOAuthAuthorizationCode code = lookupOAuthAuthorizationCode <* d
where
lookupOAuthAuthorizationCode :: (MonadClient m) => m (Maybe (OAuthClientId, UserId, OAuthScopes, RedirectUrl, Maybe OAuthCodeChallenge))
lookupOAuthAuthorizationCode = do
- mTuple <- retry x5 . query1 q $ params LocalQuorum (Identity code)
+ mTuple <- retry x5 . query1Traced q $ params LocalQuorum (Identity code)
pure $ mTuple <&> \(cid, uid, C.Set scope, uri, mChal) -> (cid, uid, OAuthScopes (Set.fromList scope), uri, mChal)
where
q :: PrepQuery R (Identity OAuthAuthorizationCode) (OAuthClientId, UserId, C.Set OAuthScope, RedirectUrl, Maybe OAuthCodeChallenge)
q = "SELECT client, user, scope, redirect_uri, code_challenge FROM oauth_auth_code WHERE code = ?"
deleteOAuthAuthorizationCode :: (MonadClient m) => m ()
- deleteOAuthAuthorizationCode = retry x5 . write q $ params LocalQuorum (Identity code)
+ deleteOAuthAuthorizationCode = retry x5 . writeTraced q $ params LocalQuorum (Identity code)
where
q :: PrepQuery W (Identity OAuthAuthorizationCode) ()
q = "DELETE FROM oauth_auth_code WHERE code = ?"
@@ -453,8 +453,8 @@ insertOAuthRefreshToken maxActiveTokens ttl info = do
let rid = info.refreshTokenId
oldTokes <- determineOldestTokensToBeDeleted <$> lookupOAuthRefreshTokens info.userId
for_ oldTokes (\t -> deleteOAuthRefreshToken t.userId t.refreshTokenId)
- retry x5 . write qInsertId $ params LocalQuorum (info.userId, rid, fromIntegral ttl)
- retry x5 . write qInsertInfo $ params LocalQuorum (rid, info.clientId, info.userId, C.Set (Set.toList (unOAuthScopes info.scopes)), info.createdAt, fromIntegral ttl)
+ retry x5 . writeTraced qInsertId $ params LocalQuorum (info.userId, rid, fromIntegral ttl)
+ retry x5 . writeTraced qInsertInfo $ params LocalQuorum (rid, info.clientId, info.userId, C.Set (Set.toList (unOAuthScopes info.scopes)), info.createdAt, fromIntegral ttl)
where
qInsertInfo :: PrepQuery W (OAuthRefreshTokenId, OAuthClientId, UserId, C.Set OAuthScope, UTCTime, Int32) ()
qInsertInfo = fromString $ "INSERT INTO oauth_refresh_token (id, client, user, scope, created_at) VALUES (?, ?, ?, ?, ?) USING TTL ?"
@@ -470,7 +470,7 @@ insertOAuthRefreshToken maxActiveTokens ttl info = do
lookupOAuthRefreshTokens :: (MonadClient m) => UserId -> m [OAuthRefreshTokenInfo]
lookupOAuthRefreshTokens uid = do
- ids <- runIdentity <$$> (retry x5 . query q $ params LocalQuorum (Identity uid))
+ ids <- runIdentity <$$> (retry x5 . queryTraced q $ params LocalQuorum (Identity uid))
catMaybes <$> for ids lookupOAuthRefreshTokenInfo
where
q :: PrepQuery R (Identity UserId) (Identity OAuthRefreshTokenId)
@@ -478,7 +478,7 @@ lookupOAuthRefreshTokens uid = do
lookupOAuthRefreshTokenInfo :: (MonadClient m) => OAuthRefreshTokenId -> m (Maybe OAuthRefreshTokenInfo)
lookupOAuthRefreshTokenInfo rid = do
- mTuple <- retry x5 . query1 q $ params LocalQuorum (Identity rid)
+ mTuple <- retry x5 . query1Traced q $ params LocalQuorum (Identity rid)
pure $ mTuple <&> \(cid, uid, C.Set scope, createdAt) -> OAuthRefreshTokenInfo rid cid uid (OAuthScopes (Set.fromList scope)) createdAt
where
q :: PrepQuery R (Identity OAuthRefreshTokenId) (OAuthClientId, UserId, C.Set OAuthScope, UTCTime)
@@ -486,8 +486,8 @@ lookupOAuthRefreshTokenInfo rid = do
deleteOAuthRefreshToken :: (MonadClient m) => UserId -> OAuthRefreshTokenId -> m ()
deleteOAuthRefreshToken uid rid = do
- retry x5 . write qDeleteId $ params LocalQuorum (uid, rid)
- retry x5 . write qDeleteInfo $ params LocalQuorum (Identity rid)
+ retry x5 . writeTraced qDeleteId $ params LocalQuorum (uid, rid)
+ retry x5 . writeTraced qDeleteInfo $ params LocalQuorum (Identity rid)
where
qDeleteId :: PrepQuery W (UserId, OAuthRefreshTokenId) ()
qDeleteId = "DELETE FROM oauth_user_refresh_token WHERE user = ? AND token_id = ?"
diff --git a/services/brig/src/Brig/App.hs b/services/brig/src/Brig/App.hs
index 1e50c5179ed..c5ead7e8684 100644
--- a/services/brig/src/Brig/App.hs
+++ b/services/brig/src/Brig/App.hs
@@ -117,7 +117,7 @@ import Brig.Team.Template
import Brig.Template (InvitationUrlTemplates (..), genTemplateBranding, genTemplateBrandingMap)
import Brig.User.Search.Index (IndexEnv (..), MonadIndexIO (..), runIndexIO)
import Brig.User.Template
-import Cassandra (runClient)
+import Cassandra (runClientTraced)
import Cassandra qualified as Cas
import Cassandra.Util (initCassandraForService)
import Control.AutoUpdate
@@ -555,7 +555,7 @@ instance MonadHttp (AppT r) where
wrapClient :: ReaderT Env Cas.Client a -> AppT r a
wrapClient m = do
env <- ask
- runClient env.casClient $ runReaderT m env
+ liftIO $ runClientTraced env.casClient $ runReaderT m env
wrapClientE :: ExceptT e (ReaderT Env Cas.Client) a -> ExceptT e (AppT r) a
wrapClientE = mapExceptT wrapClient
@@ -590,7 +590,8 @@ newtype HttpClientIO a = HttpClientIO
runHttpClientIO :: (MonadIO m) => Env -> HttpClientIO a -> m a
runHttpClientIO env =
- runClient (env.casClient)
+ liftIO
+ . runClientTraced (env.casClient)
. runHttpT (env.httpManager)
. flip runReaderT env
. unHttpClientIO
@@ -601,7 +602,7 @@ instance HasRequestId HttpClientIO where
instance Cas.MonadClient HttpClientIO where
liftClient cl = do
env <- ask
- liftIO $ runClient (asks (.casClient) env) cl
+ liftIO $ runClientTraced (asks (.casClient) env) cl
localState f = local (casClientLens %~ f)
instance MonadMonitor HttpClientIO where
diff --git a/services/brig/src/Brig/Data/Activation.hs b/services/brig/src/Brig/Data/Activation.hs
index 9c9eb1446e1..82c1e031f49 100644
--- a/services/brig/src/Brig/Data/Activation.hs
+++ b/services/brig/src/Brig/Data/Activation.hs
@@ -143,7 +143,7 @@ verifyCode ::
ActivationCode ->
ExceptT ActivationError m (EmailKey, Maybe UserId)
verifyCode key code = do
- s <- lift . retry x1 . query1 keySelect $ params LocalQuorum (Identity key)
+ s <- lift . retry x1 . query1Traced keySelect $ params LocalQuorum (Identity key)
case s of
Just (ttl, Ascii t, k, c, u, r) ->
if
@@ -156,7 +156,7 @@ verifyCode key code = do
Just e -> pure (mkEmailKey e, u)
Nothing -> throwE invalidCode
mkScope _ _ _ = throwE invalidCode
- countdown = lift . retry x5 . write keyInsert . params LocalQuorum
+ countdown = lift . retry x5 . writeTraced keyInsert . params LocalQuorum
revoke = lift $ deleteActivationPair key
keyInsert :: PrepQuery W (ActivationKey, Text, Text, ActivationCode, Maybe UserId, Int32, Int32) ()
keyInsert =
@@ -172,7 +172,7 @@ mkActivationKey k = do
pure . ActivationKey $ Ascii.encodeBase64Url bs
deleteActivationPair :: (MonadClient m) => ActivationKey -> m ()
-deleteActivationPair = write keyDelete . params LocalQuorum . Identity
+deleteActivationPair = writeTraced keyDelete . params LocalQuorum . Identity
invalidUser :: ActivationError
invalidUser = InvalidActivationCodeWrongUser -- "User does not exist."
diff --git a/services/brig/src/Brig/Data/Connection.hs b/services/brig/src/Brig/Data/Connection.hs
index 41a6f9a8275..c1dab5292ac 100644
--- a/services/brig/src/Brig/Data/Connection.hs
+++ b/services/brig/src/Brig/Data/Connection.hs
@@ -75,10 +75,10 @@ insertConnection ::
insertConnection self target rel qcnv@(Qualified cnv cdomain) = do
now <- toUTCTimeMillis <$> liftIO getCurrentTime
let local (tUnqualified -> ltarget) =
- write connectionInsert $
+ writeTraced connectionInsert $
params LocalQuorum (tUnqualified self, ltarget, rel, now, cnv)
let remote (tUntagged -> Qualified rtarget domain) =
- write remoteConnectionInsert $
+ writeTraced remoteConnectionInsert $
params LocalQuorum (tUnqualified self, domain, rtarget, rel, now, cdomain, cnv)
retry x5 $ foldQualified self local remote target
pure $
@@ -104,10 +104,10 @@ updateConnectionStatus :: (MonadClient m) => Local UserId -> Qualified UserId ->
updateConnectionStatus self target status = do
now <- toUTCTimeMillis <$> liftIO getCurrentTime
let local (tUnqualified -> ltarget) =
- write connectionUpdate $
+ writeTraced connectionUpdate $
params LocalQuorum (status, now, tUnqualified self, ltarget)
let remote (tUntagged -> Qualified rtarget domain) =
- write remoteConnectionUpdate $
+ writeTraced remoteConnectionUpdate $
params LocalQuorum (status, now, tUnqualified self, domain, rtarget)
retry x5 $ foldQualified self local remote target
pure now
@@ -117,12 +117,12 @@ lookupConnection :: (MonadClient m) => Local UserId -> Qualified UserId -> m (Ma
lookupConnection self target = runMaybeT $ do
let local (tUnqualified -> ltarget) = do
(_, _, rel, time, mcnv) <-
- MaybeT . query1 connectionSelect $
+ MaybeT . query1Traced connectionSelect $
params LocalQuorum (tUnqualified self, ltarget)
pure (rel, time, fmap (tUntagged . qualifyAs self) mcnv)
let remote (tUntagged -> Qualified rtarget domain) = do
(rel, time, cdomain, cnv) <-
- MaybeT . query1 remoteConnectionSelectFrom $
+ MaybeT . query1Traced remoteConnectionSelectFrom $
params LocalQuorum (tUnqualified self, domain, rtarget)
pure (rel, time, Just (Qualified cnv cdomain))
(rel, time, mqcnv) <- hoist (retry x1) $ foldQualified self local remote target
@@ -145,9 +145,9 @@ lookupRelationWithHistory ::
m (Maybe RelationWithHistory)
lookupRelationWithHistory self target = do
let local (tUnqualified -> ltarget) =
- query1 relationSelect (params LocalQuorum (tUnqualified self, ltarget))
+ query1Traced relationSelect (params LocalQuorum (tUnqualified self, ltarget))
let remote (tUntagged -> Qualified rtarget domain) =
- query1 remoteRelationSelect (params LocalQuorum (tUnqualified self, domain, rtarget))
+ query1Traced remoteRelationSelect (params LocalQuorum (tUnqualified self, domain, rtarget))
runIdentity <$$> retry x1 (foldQualified self local remote target)
-- | For a given user 'A', lookup their outgoing connections (A -> X) to other users.
@@ -198,13 +198,13 @@ lookupRemoteConnectionsPage self pagingState size =
lookupConnectionStatus :: (MonadClient m) => [UserId] -> [UserId] -> m [ConnectionStatus]
lookupConnectionStatus from to =
map toConnectionStatus
- <$> retry x1 (query connectionStatusSelect (params LocalQuorum (from, to)))
+ <$> retry x1 (queryTraced connectionStatusSelect (params LocalQuorum (from, to)))
-- | Lookup all relations between two sets of users (cartesian product).
lookupConnectionStatus' :: (MonadClient m) => [UserId] -> m [ConnectionStatus]
lookupConnectionStatus' from =
map toConnectionStatus
- <$> retry x1 (query connectionStatusSelect' (params LocalQuorum (Identity from)))
+ <$> retry x1 (queryTraced connectionStatusSelect' (params LocalQuorum (Identity from)))
lookupLocalConnectionStatuses ::
( MonadClient m,
@@ -219,7 +219,7 @@ lookupLocalConnectionStatuses froms tos = do
lookupStatuses :: (MonadClient m) => UserId -> m [ConnectionStatusV2]
lookupStatuses from =
map (uncurry $ toConnectionStatusV2 from (tDomain tos))
- <$> retry x1 (query relationsSelect (params LocalQuorum (from, tUnqualified tos)))
+ <$> retry x1 (queryTraced relationsSelect (params LocalQuorum (from, tUnqualified tos)))
lookupRemoteConnectionStatuses ::
( MonadClient m,
@@ -234,7 +234,7 @@ lookupRemoteConnectionStatuses froms tos = do
lookupStatuses :: (MonadClient m) => UserId -> m [ConnectionStatusV2]
lookupStatuses from =
map (uncurry $ toConnectionStatusV2 from (tDomain tos))
- <$> retry x1 (query remoteRelationsSelect (params LocalQuorum (from, tDomain tos, tUnqualified tos)))
+ <$> retry x1 (queryTraced remoteRelationsSelect (params LocalQuorum (from, tDomain tos, tUnqualified tos)))
lookupAllStatuses ::
( MonadClient m,
@@ -252,11 +252,11 @@ lookupAllStatuses lfroms = do
lookupLocalStatuses :: (MonadClient m) => UserId -> m [ConnectionStatusV2]
lookupLocalStatuses from =
map (uncurry $ toConnectionStatusV2 from (tDomain lfroms))
- <$> retry x1 (query relationsSelectAll (params LocalQuorum (Identity from)))
+ <$> retry x1 (queryTraced relationsSelectAll (params LocalQuorum (Identity from)))
lookupRemoteStatuses :: (MonadClient m) => UserId -> m [ConnectionStatusV2]
lookupRemoteStatuses from =
map (\(d, u, r) -> toConnectionStatusV2 from d u r)
- <$> retry x1 (query remoteRelationsSelectAll (params LocalQuorum (Identity from)))
+ <$> retry x1 (queryTraced remoteRelationsSelectAll (params LocalQuorum (Identity from)))
lookupRemoteConnectedUsersPaginated :: (MonadClient m) => Local UserId -> Int32 -> m (Page (Remote UserConnection))
lookupRemoteConnectedUsersPaginated u maxResults = do
@@ -271,7 +271,7 @@ lookupContactList u =
-- i.e. the users to whom 'A' has an outgoing 'Accepted' relation (A -> B).
lookupContactListWithRelation :: (MonadClient m) => UserId -> m [(UserId, RelationWithHistory)]
lookupContactListWithRelation u =
- retry x1 (query contactsSelect (params LocalQuorum (Identity u)))
+ retry x1 (queryTraced contactsSelect (params LocalQuorum (Identity u)))
-- | Count the number of connections a user has in a specific relation status.
-- (If you want to distinguish 'RelationWithHistory', write a new function.)
@@ -299,10 +299,10 @@ deleteConnections u = do
.| C.mapM_
(pooledMapConcurrentlyN_ 16 delete)
do
- retry x1 . write connectionClear $ params LocalQuorum (Identity u)
- retry x1 . write remoteConnectionClear $ params LocalQuorum (Identity u)
+ retry x1 . writeTraced connectionClear $ params LocalQuorum (Identity u)
+ retry x1 . writeTraced remoteConnectionClear $ params LocalQuorum (Identity u)
where
- delete (other, _status) = write connectionDelete $ params LocalQuorum (other, u)
+ delete (other, _status) = writeTraced connectionDelete $ params LocalQuorum (other, u)
deleteRemoteConnections ::
( MonadClient m,
@@ -313,7 +313,7 @@ deleteRemoteConnections ::
m ()
deleteRemoteConnections (tUntagged -> Qualified remoteUser remoteDomain) (fromRange -> locals) =
pooledForConcurrentlyN_ 16 locals $ \u ->
- write remoteConnectionDelete $ params LocalQuorum (u, remoteDomain, remoteUser)
+ writeTraced remoteConnectionDelete $ params LocalQuorum (u, remoteDomain, remoteUser)
-- Queries
diff --git a/services/brig/src/Brig/Data/MLS/KeyPackage.hs b/services/brig/src/Brig/Data/MLS/KeyPackage.hs
index 20f95a40afa..a7b11f7ec5f 100644
--- a/services/brig/src/Brig/Data/MLS/KeyPackage.hs
+++ b/services/brig/src/Brig/Data/MLS/KeyPackage.hs
@@ -75,7 +75,7 @@ claimKeyPackage u c suite = do
kps <- getNonClaimedKeyPackages u c suite
mk <- liftIO (pick kps)
for mk $ \(ref, kpd) -> do
- retry x5 $ write delete1Query (params LocalQuorum (tUnqualified u, c, suite, ref))
+ retry x5 $ writeTraced delete1Query (params LocalQuorum (tUnqualified u, c, suite, ref))
pure (ref, kpd)
pure (ref, kpd)
where
@@ -93,7 +93,7 @@ getNonClaimedKeyPackages ::
CipherSuiteTag ->
m [(KeyPackageRef, KeyPackageData)]
getNonClaimedKeyPackages u c suite = do
- kps <- retry x1 $ query lookupQuery (params LocalQuorum (tUnqualified u, c, suite))
+ kps <- retry x1 $ queryTraced lookupQuery (params LocalQuorum (tUnqualified u, c, suite))
let decodedKps = foldMap (keepDecoded . (decodeKp &&& id)) kps
now <- liftIO getPOSIXTime
@@ -136,7 +136,7 @@ countKeyPackages u c suite = fromIntegral . length <$> getNonClaimedKeyPackages
deleteKeyPackages :: (MonadClient m) => UserId -> ClientId -> CipherSuiteTag -> [KeyPackageRef] -> m ()
deleteKeyPackages u c suite refs =
retry x5 $
- write
+ writeTraced
deleteQuery
(params LocalQuorum (u, c, suite, refs))
where
@@ -152,7 +152,7 @@ deleteAllKeyPackages ::
deleteAllKeyPackages u c suites =
pooledForConcurrentlyN_ 16 suites $ \suite ->
retry x5 $
- write
+ writeTraced
deleteQuery
(params LocalQuorum (u, c, suite))
where
diff --git a/services/brig/src/Brig/Data/Nonce.hs b/services/brig/src/Brig/Data/Nonce.hs
index 00fda3fbcd5..ff80f6e1d97 100644
--- a/services/brig/src/Brig/Data/Nonce.hs
+++ b/services/brig/src/Brig/Data/Nonce.hs
@@ -34,7 +34,7 @@ insertNonce ::
Text ->
Nonce ->
m ()
-insertNonce ttl uid key nonce = retry x5 . write insert $ params LocalQuorum (uid, key, nonce, ttl)
+insertNonce ttl uid key nonce = retry x5 . writeTraced insert $ params LocalQuorum (uid, key, nonce, ttl)
where
insert :: PrepQuery W (UserId, Text, Nonce, NonceTtlSecs) ()
insert = "INSERT INTO nonce (user, key, nonce) VALUES (?, ?, ?) USING TTL ?"
@@ -51,7 +51,7 @@ lookupNonce ::
UserId ->
Text ->
m (Maybe Nonce)
-lookupNonce uid key = (runIdentity <$$>) . retry x5 . query1 get $ params LocalQuorum (uid, key)
+lookupNonce uid key = (runIdentity <$$>) . retry x5 . query1Traced get $ params LocalQuorum (uid, key)
where
get :: PrepQuery R (UserId, Text) (Identity Nonce)
get = "SELECT nonce FROM nonce WHERE user = ? AND key = ?"
@@ -61,7 +61,7 @@ deleteNonce ::
UserId ->
Text ->
m ()
-deleteNonce uid key = retry x5 . write delete $ params LocalQuorum (uid, key)
+deleteNonce uid key = retry x5 . writeTraced delete $ params LocalQuorum (uid, key)
where
delete :: PrepQuery W (UserId, Text) ()
delete = "DELETE FROM nonce WHERE user = ? AND key = ?"
diff --git a/services/brig/src/Brig/Provider/DB.hs b/services/brig/src/Brig/Provider/DB.hs
index fc1a828c169..f7b100872c1 100644
--- a/services/brig/src/Brig/Provider/DB.hs
+++ b/services/brig/src/Brig/Provider/DB.hs
@@ -50,7 +50,7 @@ insertAccount ::
m ProviderId
insertAccount name pass url descr = do
pid <- randomId
- retry x5 $ write cql $ params LocalQuorum (pid, name, pass, url, descr)
+ retry x5 $ writeTraced cql $ params LocalQuorum (pid, name, pass, url, descr)
pure pid
where
cql :: PrepQuery W (ProviderId, Name, Password, HttpsUrl, Text) ()
@@ -82,7 +82,7 @@ lookupAccountData ::
(MonadClient m) =>
ProviderId ->
m (Maybe (Name, Maybe EmailAddress, HttpsUrl, Text))
-lookupAccountData p = retry x1 $ query1 cql $ params LocalQuorum (Identity p)
+lookupAccountData p = retry x1 $ query1Traced cql $ params LocalQuorum (Identity p)
where
cql :: PrepQuery R (Identity ProviderId) (Name, Maybe EmailAddress, HttpsUrl, Text)
cql = "SELECT name, email, url, descr FROM provider WHERE id = ?"
@@ -107,7 +107,7 @@ deleteAccount ::
(MonadClient m) =>
ProviderId ->
m ()
-deleteAccount pid = retry x5 $ write cql $ params LocalQuorum (Identity pid)
+deleteAccount pid = retry x5 $ writeTraced cql $ params LocalQuorum (Identity pid)
where
cql :: PrepQuery W (Identity ProviderId) ()
cql = "DELETE FROM provider WHERE id = ?"
@@ -118,7 +118,7 @@ updateAccountPassword ::
Password ->
m ()
updateAccountPassword pid pwd = do
- retry x5 $ write cql $ params LocalQuorum (pwd, pid)
+ retry x5 $ writeTraced cql $ params LocalQuorum (pwd, pid)
where
cql :: PrepQuery W (Password, ProviderId) ()
cql = "UPDATE provider SET password = ? where id = ?"
@@ -155,14 +155,14 @@ lookupKey ::
lookupKey k =
fmap (fmap runIdentity) $
retry x1 $
- query1 cql $
+ query1Traced cql $
params LocalQuorum (Identity (emailKeyUniq k))
where
cql :: PrepQuery R (Identity Text) (Identity ProviderId)
cql = "SELECT provider FROM provider_keys WHERE key = ?"
deleteKey :: (MonadClient m) => EmailKey -> m ()
-deleteKey k = retry x5 $ write cql $ params LocalQuorum (Identity (emailKeyUniq k))
+deleteKey k = retry x5 $ writeTraced cql $ params LocalQuorum (Identity (emailKeyUniq k))
where
cql :: PrepQuery W (Identity Text) ()
cql = "DELETE FROM provider_keys WHERE key = ?"
@@ -187,7 +187,7 @@ insertService pid name summary descr url token key fprint assets tags = do
sid <- randomId
let tagSet = C.Set (Set.toList tags)
retry x5 $
- write cql $
+ writeTraced cql $
params
LocalQuorum
(pid, sid, name, summary, descr, url, [token], [key], [fprint], assets, tagSet, False)
@@ -223,7 +223,7 @@ lookupService ::
lookupService pid sid =
fmap (fmap mk) $
retry x1 $
- query1 cql $
+ query1Traced cql $
params LocalQuorum (pid, sid)
where
cql ::
@@ -244,7 +244,7 @@ listServices ::
listServices p =
fmap (map mk) $
retry x1 $
- query cql $
+ queryTraced cql $
params LocalQuorum (Identity p)
where
cql ::
@@ -339,7 +339,7 @@ lookupServiceProfile ::
lookupServiceProfile p s =
fmap (fmap mk) $
retry x1 $
- query1 cql $
+ query1Traced cql $
params One (p, s)
where
cql :: PrepQuery R (ProviderId, ServiceId) (Name, Maybe Text, Text, [Asset], C.Set ServiceTag, Bool)
@@ -358,7 +358,7 @@ listServiceProfiles ::
listServiceProfiles p =
fmap (map mk) $
retry x1 $
- query cql $
+ queryTraced cql $
params One (Identity p)
where
cql ::
@@ -394,7 +394,7 @@ lookupServiceConn ::
lookupServiceConn pid sid =
fmap (fmap mk) $
retry x1 $
- query1 cql $
+ query1Traced cql $
params LocalQuorum (pid, sid)
where
cql :: PrepQuery R (ProviderId, ServiceId) (HttpsUrl, CqlNonEmpty ServiceToken, CqlNonEmpty (Fingerprint Rsa), Bool)
@@ -728,7 +728,7 @@ insertServiceWhitelist tid pid sid =
deleteServiceWhitelist :: (MonadClient m) => Maybe TeamId -> ProviderId -> ServiceId -> m ()
deleteServiceWhitelist mbTid pid sid = case mbTid of
Nothing -> do
- teams <- retry x5 $ query lookupRev $ params LocalQuorum (pid, sid)
+ teams <- retry x5 $ queryTraced lookupRev $ params LocalQuorum (pid, sid)
retry x5 . batch $ do
setType BatchLogged
setConsistency LocalQuorum
@@ -776,7 +776,7 @@ paginateServiceWhitelist tid mbPrefix filterDisabled size = liftClient $ do
-- services, regardless of 'size'. This is because otherwise we would
-- have to go through multiple passes of query->filter->query a bit
-- more->filter->... if we get unlucky.
- p <- retry x1 $ query cql $ params One (Identity tid)
+ p <- retry x1 $ queryTraced cql $ params One (Identity tid)
r <-
maybeFilterPrefix
. sortOn (Text.toLower . fromName . serviceProfileName)
@@ -809,7 +809,7 @@ getServiceWhitelistStatus ::
ServiceId ->
m Bool
getServiceWhitelistStatus tid pid sid = liftClient $ do
- fmap isJust $ retry x1 $ query1 cql $ params One (tid, pid, sid)
+ fmap isJust $ retry x1 $ query1Traced cql $ params One (tid, pid, sid)
where
cql :: PrepQuery R (TeamId, ProviderId, ServiceId) (Identity TeamId)
cql =
diff --git a/services/brig/src/Brig/Run.hs b/services/brig/src/Brig/Run.hs
index 5b5d0f96818..782da354252 100644
--- a/services/brig/src/Brig/Run.hs
+++ b/services/brig/src/Brig/Run.hs
@@ -80,7 +80,7 @@ import Wire.UserStore
-- thread terminates for any reason.
-- https://github.com/zinfra/backend-issues/issues/1647
run :: Opts -> IO ()
-run opts = withTracer \tracer -> do
+run opts = withTracer "brig" \tracer -> do
(app, e) <- mkApp opts
runAllMigrations e.hasqlPool e.appLogger
let s = Server.newSettings (server e)
diff --git a/services/cannon/src/Cannon/RabbitMqConsumerApp.hs b/services/cannon/src/Cannon/RabbitMqConsumerApp.hs
index 13f7f1950cb..7ae9cfc7a84 100644
--- a/services/cannon/src/Cannon/RabbitMqConsumerApp.hs
+++ b/services/cannon/src/Cannon/RabbitMqConsumerApp.hs
@@ -298,7 +298,7 @@ sendFullSyncMessageIfNeeded ::
IO ()
sendFullSyncMessageIfNeeded wsConn uid env cid = do
row <- C.runClient env.cassandra do
- retry x5 $ query1 q (params LocalQuorum (uid, cid))
+ retry x5 $ query1Traced q (params LocalQuorum (uid, cid))
for_ row $ \_ -> sendFullSyncMessage uid cid wsConn env
where
q :: PrepQuery R (UserId, ClientId) (Identity (Maybe UserId))
@@ -320,7 +320,7 @@ sendFullSyncMessage uid cid wsConn env = do
AckMessage _ -> throwIO UnexpectedAck
AckFullSync ->
C.runClient env.cassandra do
- retry x1 $ write delete (params LocalQuorum (uid, cid))
+ retry x1 $ writeTraced delete (params LocalQuorum (uid, cid))
where
delete :: PrepQuery W (UserId, ClientId) ()
delete =
diff --git a/services/cannon/src/Cannon/Run.hs b/services/cannon/src/Cannon/Run.hs
index cf2073f212f..6d2443c870e 100644
--- a/services/cannon/src/Cannon/Run.hs
+++ b/services/cannon/src/Cannon/Run.hs
@@ -71,7 +71,7 @@ type CombinedAPI = CannonAPI :<|> Internal.API
run :: Opts -> IO ()
run o = lowerCodensity $ do
lift $ validateOpts o
- tracer <- Codensity withTracer
+ tracer <- Codensity (withTracer "cannon")
when (o ^. drainOpts . millisecondsBetweenBatches == 0) $
error "drainOpts.millisecondsBetweenBatches must not be set to 0."
when (o ^. drainOpts . gracePeriodSeconds == 0) $
diff --git a/services/cargohold/cargohold.cabal b/services/cargohold/cargohold.cabal
index ecc35c1f4a5..29ca6d69ce2 100644
--- a/services/cargohold/cargohold.cabal
+++ b/services/cargohold/cargohold.cabal
@@ -85,62 +85,65 @@ library
-Wredundant-constraints -Wunused-packages
build-depends:
- aeson >=2.0.1.0
- , amazonka >=1.3.7
- , amazonka-s3 >=1.3.7
- , amazonka-s3-streaming >=1.3.7
- , attoparsec >=0.12
- , auto-update >=0.1.4
- , base >=4 && <5
- , base64-bytestring >=1.0
- , bilge >=0.21
- , bytestring >=0.10
- , bytestring-conversion >=0.2
- , cargohold-types >=0.5
- , case-insensitive >=1.0
- , conduit >=1.2
- , conduit-extra >=1.1.5
+ aeson >=2.0.1.0
+ , amazonka >=1.3.7
+ , amazonka-s3 >=1.3.7
+ , amazonka-s3-streaming >=1.3.7
+ , attoparsec >=0.12
+ , auto-update >=0.1.4
+ , base >=4 && <5
+ , base64-bytestring >=1.0
+ , bilge >=0.21
+ , bytestring >=0.10
+ , bytestring-conversion >=0.2
+ , cargohold-types >=0.5
+ , case-insensitive >=1.0
+ , conduit >=1.2
+ , conduit-extra >=1.1.5
, containers
- , crypton >=0.20
+ , crypton >=0.20
, data-default
- , errors >=1.4
- , exceptions >=0.6
+ , errors >=1.4
+ , exceptions >=0.6
, extended
- , HsOpenSSL >=0.11
- , http-client >=0.7
- , http-client-openssl >=0.2
- , http-types >=0.8
+ , hs-opentelemetry-instrumentation-wai
+ , hs-opentelemetry-sdk
+ , HsOpenSSL >=0.11
+ , http-client >=0.7
+ , http-client-openssl >=0.2
+ , http-types >=0.8
, http2-manager
, imports
, kan-extensions
- , lens >=4.1
+ , lens >=4.1
, metrics-core
- , metrics-wai >=0.4
- , mime >=0.4
+ , metrics-wai >=0.4
+ , mime >=0.4
, prometheus-client
- , QuickCheck >=2.14
- , resourcet >=1.1
- , retry >=0.5
+ , QuickCheck >=2.14
+ , resourcet >=1.1
+ , retry >=0.5
, schema-profunctor
, servant
, servant-client
, servant-server
- , text >=1.1
- , time >=1.4
- , tinylog >=0.10
+ , text >=1.1
+ , time >=1.4
+ , tinylog >=0.10
, transformers
- , types-common >=0.16
+ , types-common >=0.16
, types-common-aws
, unliftio
- , unordered-containers >=0.2
- , uri-bytestring >=0.2
- , uuid >=1.3.5
- , wai >=3.0
+ , unordered-containers >=0.2
+ , uri-bytestring >=0.2
+ , uuid >=1.3.5
+ , wai >=3.0
, wai-extra
- , wai-utilities >=0.16.1
+ , wai-utilities >=0.16.1
, wire-api
, wire-api-federation
- , yaml >=0.8
+ , wire-otel
+ , yaml >=0.8
default-language: Haskell2010
diff --git a/services/cargohold/default.nix b/services/cargohold/default.nix
index 0ee6682349a..b4332b8601d 100644
--- a/services/cargohold/default.nix
+++ b/services/cargohold/default.nix
@@ -25,6 +25,8 @@
, exceptions
, extended
, federator
+, hs-opentelemetry-instrumentation-wai
+, hs-opentelemetry-sdk
, HsOpenSSL
, http-api-data
, http-client
@@ -72,6 +74,7 @@
, wai-utilities
, wire-api
, wire-api-federation
+, wire-otel
, yaml
}:
mkDerivation {
@@ -102,6 +105,8 @@ mkDerivation {
errors
exceptions
extended
+ hs-opentelemetry-instrumentation-wai
+ hs-opentelemetry-sdk
HsOpenSSL
http-client
http-client-openssl
@@ -136,6 +141,7 @@ mkDerivation {
wai-utilities
wire-api
wire-api-federation
+ wire-otel
yaml
];
executableHaskellDepends = [
diff --git a/services/cargohold/src/CargoHold/Run.hs b/services/cargohold/src/CargoHold/Run.hs
index 77eb3628cf4..083d3b12c38 100644
--- a/services/cargohold/src/CargoHold/Run.hs
+++ b/services/cargohold/src/CargoHold/Run.hs
@@ -59,6 +59,8 @@ import qualified Network.Wai.Middleware.Gzip as GZip
import Network.Wai.Utilities.Request
import Network.Wai.Utilities.Server
import qualified Network.Wai.Utilities.Server as Server
+import qualified OpenTelemetry.Instrumentation.Wai as OtelWai
+import OpenTelemetry.Trace as Otel
import qualified Servant
import Servant.API
import Servant.Server hiding (Handler, runHandler)
@@ -69,11 +71,13 @@ import Wire.API.Routes.Internal.Cargohold
import Wire.API.Routes.Public.Cargohold
import Wire.API.Routes.Version
import Wire.API.Routes.Version.Wai
+import Wire.OpenTelemetry (withTracerC)
type CombinedAPI = FederationAPI :<|> CargoholdAPI :<|> InternalAPI
run :: Opts -> IO ()
run o = lowerCodensity $ do
+ tracer <- withTracerC "cargohold"
(app, e) <- mkApp o
void $ Codensity $ Async.withAsync (collectAuthMetrics e.aws.amazonkaEnv)
let s =
@@ -82,17 +86,19 @@ run o = lowerCodensity $ do
(unpack . host $ o.cargohold)
(port o.cargohold)
e.appLogger
- liftIO $ runSettingsWithShutdown s app Nothing
+ lift $ inSpan tracer "cargohold" defaultSpanArguments {kind = Otel.Server} (runSettingsWithShutdown s app Nothing)
mkApp :: Opts -> Codensity IO (Application, Env)
mkApp o = Codensity $ \k ->
- bracket (newEnv o) closeEnv $ \e ->
- k (middleware e (servantApp e), e)
+ bracket (newEnv o) closeEnv $ \e -> do
+ otelMiddleware <- OtelWai.newOpenTelemetryWaiMiddleware
+ k (middleware otelMiddleware e (servantApp e), e)
where
- middleware :: Env -> Wai.Middleware
- middleware e =
+ middleware :: Wai.Middleware -> Env -> Wai.Middleware
+ middleware otelMiddleware e =
versionMiddleware (foldMap expandVersionExp o.settings.disabledAPIVersions)
. requestIdMiddleware e.appLogger defaultRequestIdHeaderName
+ . otelMiddleware
. servantPrometheusMiddleware (Proxy @CombinedAPI)
. GZip.gzip GZip.defaultGzipSettings
. catchErrors e.appLogger defaultRequestIdHeaderName
diff --git a/services/federator/default.nix b/services/federator/default.nix
index 5c6a9f429d2..a3a6527d882 100644
--- a/services/federator/default.nix
+++ b/services/federator/default.nix
@@ -23,6 +23,8 @@
, extended
, filepath
, fsnotify
+, hs-opentelemetry-instrumentation-wai
+, hs-opentelemetry-sdk
, HsOpenSSL
, hspec
, hspec-junit-formatter
@@ -71,6 +73,7 @@
, warp-tls
, wire-api
, wire-api-federation
+, wire-otel
, yaml
}:
mkDerivation {
@@ -97,6 +100,8 @@ mkDerivation {
extended
filepath
fsnotify
+ hs-opentelemetry-instrumentation-wai
+ hs-opentelemetry-sdk
HsOpenSSL
http-client
http-media
@@ -129,6 +134,7 @@ mkDerivation {
warp
wire-api
wire-api-federation
+ wire-otel
];
executableHaskellDepends = [
aeson
diff --git a/services/federator/federator.cabal b/services/federator/federator.cabal
index 6194c6020b1..406ad63b20f 100644
--- a/services/federator/federator.cabal
+++ b/services/federator/federator.cabal
@@ -60,6 +60,7 @@ library
default-extensions:
AllowAmbiguousTypes
BangPatterns
+ BlockArguments
ConstraintKinds
DataKinds
DefaultSignatures
@@ -122,6 +123,8 @@ library
, extended
, filepath
, fsnotify
+ , hs-opentelemetry-instrumentation-wai
+ , hs-opentelemetry-sdk
, HsOpenSSL
, http-client
, http-media
@@ -154,6 +157,7 @@ library
, warp
, wire-api
, wire-api-federation
+ , wire-otel
default-language: GHC2021
diff --git a/services/federator/src/Federator/Interpreter.hs b/services/federator/src/Federator/Interpreter.hs
index 879503ae30f..72bf2fbfc79 100644
--- a/services/federator/src/Federator/Interpreter.hs
+++ b/services/federator/src/Federator/Interpreter.hs
@@ -46,6 +46,7 @@ import Network.HTTP.Types qualified as HTTP
import Network.Wai qualified as Wai
import Network.Wai.Handler.Warp qualified as Warp
import Network.Wai.Utilities (getRequestId)
+import OpenTelemetry.Instrumentation.Wai qualified as OtelWai
import Network.Wai.Utilities.Error qualified as Wai
import Network.Wai.Utilities.Server (defaultRequestIdHeaderName, federationRequestIdHeaderName, requestIdMiddleware)
import Network.Wai.Utilities.Server qualified as Wai
@@ -135,8 +136,10 @@ serveServant env port cleanupsRef server = do
Warp.setPort port
. Warp.setInstallShutdownHandler registerCleanupAction
$ Warp.defaultSettings
+ otelMiddleware <- OtelWai.newOpenTelemetryWaiMiddleware
Warp.runSettings settings
. requestIdMiddleware env._applog federationRequestIdHeaderName
+ . otelMiddleware
. Wai.catchErrors (view applog env) federationRequestIdHeaderName
. Metrics.servantPrometheusMiddleware (Proxy @api)
$ app hoistApp
diff --git a/services/federator/src/Federator/Run.hs b/services/federator/src/Federator/Run.hs
index 87fa12710b8..2cef59a9fb0 100644
--- a/services/federator/src/Federator/Run.hs
+++ b/services/federator/src/Federator/Run.hs
@@ -52,16 +52,18 @@ import System.Logger qualified as Log
import System.Logger.Extended qualified as LogExt
import System.Posix (installHandler)
import System.Posix.Signals qualified as Sig
+import OpenTelemetry.Trace as Otel
import Util.Options
import Wire.API.Federation.Component
import Wire.Network.DNS.Helper qualified as DNS
+import Wire.OpenTelemetry (withTracer)
------------------------------------------------------------------------------
-- run/app
-- FUTUREWORK(federation): Add metrics and status endpoints
run :: Opts -> IO ()
-run opts = do
+run opts = withTracer "federator" \tracer -> do
spawnGCMetricsCollector
let resolvConf = mkResolvConf (optSettings opts) DNS.defaultResolvConf
DNS.withCachingResolver resolvConf $ \res -> do
@@ -70,12 +72,13 @@ run opts = do
bracket (newEnv opts res logger) closeEnv $ \env -> do
let externalServer = serveInward env portExternal cleanupActionsRef
internalServer = serveOutward env portInternal cleanupActionsRef
- withMonitor logger (onNewSSLContext env) (optSettings opts) $ do
- _ <- installHandler Sig.sigINT (Sig.CatchOnce $ cleanup cleanupActionsRef) Nothing
- _ <- installHandler Sig.sigTERM (Sig.CatchOnce $ cleanup cleanupActionsRef) Nothing
- internalServerThread <- async internalServer
- externalServerThread <- async externalServer
- void $ waitAnyCancel [internalServerThread, externalServerThread]
+ withMonitor logger (onNewSSLContext env) (optSettings opts) $
+ inSpan tracer "federator" defaultSpanArguments {kind = Otel.Server} $ do
+ _ <- installHandler Sig.sigINT (Sig.CatchOnce $ cleanup cleanupActionsRef) Nothing
+ _ <- installHandler Sig.sigTERM (Sig.CatchOnce $ cleanup cleanupActionsRef) Nothing
+ internalServerThread <- async internalServer
+ externalServerThread <- async externalServer
+ void $ waitAnyCancel [internalServerThread, externalServerThread]
where
endpointInternal = federatorInternal opts
portInternal = fromIntegral $ endpointInternal.port
diff --git a/services/galley/src/Galley/Monad.hs b/services/galley/src/Galley/Monad.hs
index 43af4e1dc20..964e82c87cb 100644
--- a/services/galley/src/Galley/Monad.hs
+++ b/services/galley/src/Galley/Monad.hs
@@ -60,7 +60,7 @@ instance MonadHttp App where
instance MonadClient App where
liftClient m = do
cs <- view cstate
- liftIO $ runClient cs m
+ liftIO $ runClientTraced cs m
localState f = locally cstate f
instance LC.MonadLogger App where
diff --git a/services/galley/src/Galley/Run.hs b/services/galley/src/Galley/Run.hs
index be55ed36d83..afc979477eb 100644
--- a/services/galley/src/Galley/Run.hs
+++ b/services/galley/src/Galley/Run.hs
@@ -72,7 +72,7 @@ import Wire.PostgresMigrations (runAllMigrations)
run :: Opts -> IO ()
run opts = lowerCodensity do
- tracer <- withTracerC
+ tracer <- withTracerC "galley"
(app, env) <- mkApp opts
lift $ runAllMigrations env._hasqlPool env._applog
let settings' =
diff --git a/services/gundeck/src/Gundeck/Run.hs b/services/gundeck/src/Gundeck/Run.hs
index 89e4c9f8ef2..4f01c64e0f6 100644
--- a/services/gundeck/src/Gundeck/Run.hs
+++ b/services/gundeck/src/Gundeck/Run.hs
@@ -83,7 +83,7 @@ import Wire.API.Routes.Version.Wai
import Wire.OpenTelemetry
run :: Opts -> IO ()
-run opts = withTracer \tracer -> do
+run opts = withTracer "gundeck" \tracer -> do
(rThreads, env) <- createEnv opts
let logger = env ^. applog
diff --git a/services/proxy/default.nix b/services/proxy/default.nix
index 9158b0b395b..e98d0cb18c8 100644
--- a/services/proxy/default.nix
+++ b/services/proxy/default.nix
@@ -13,6 +13,8 @@
, errors
, exceptions
, extended
+, hs-opentelemetry-instrumentation-wai
+, hs-opentelemetry-sdk
, http-client
, http-client-tls
, http-reverse-proxy
@@ -32,6 +34,7 @@
, wai-middleware-gunzip
, wai-utilities
, wire-api
+, wire-otel
}:
mkDerivation {
pname = "proxy";
@@ -50,6 +53,8 @@ mkDerivation {
errors
exceptions
extended
+ hs-opentelemetry-instrumentation-wai
+ hs-opentelemetry-sdk
http-client
http-client-tls
http-reverse-proxy
@@ -68,6 +73,7 @@ mkDerivation {
wai-middleware-gunzip
wai-utilities
wire-api
+ wire-otel
];
executableHaskellDepends = [ base imports types-common ];
license = lib.licenses.agpl3Only;
diff --git a/services/proxy/proxy.cabal b/services/proxy/proxy.cabal
index 440e1f6b1e5..355131c82d2 100644
--- a/services/proxy/proxy.cabal
+++ b/services/proxy/proxy.cabal
@@ -74,34 +74,37 @@ library
-funbox-strict-fields -Wredundant-constraints -Wunused-packages
build-depends:
- aeson >=2.0.1.0
- , base >=4.6 && <5
- , bilge >=0.21
- , bytestring >=0.10
- , case-insensitive >=1.2
+ aeson >=2.0.1.0
+ , base >=4.6 && <5
+ , bilge >=0.21
+ , bytestring >=0.10
+ , case-insensitive >=1.2
, cassandra-util
- , configurator >=0.3
+ , configurator >=0.3
, errors
- , exceptions >=0.8
+ , exceptions >=0.8
, extended
- , http-client >=0.7
- , http-client-tls >=0.3
- , http-reverse-proxy >=0.4
- , http-types >=0.9
+ , hs-opentelemetry-instrumentation-wai
+ , hs-opentelemetry-sdk
+ , http-client >=0.7
+ , http-client-tls >=0.3
+ , http-reverse-proxy >=0.4
+ , http-types >=0.9
, imports
- , lens >=4.11
- , metrics-wai >=0.5
- , retry >=0.7
+ , lens >=4.11
+ , metrics-wai >=0.5
+ , retry >=0.7
, servant-server
- , text >=1.2
- , tinylog >=0.12
- , types-common >=0.8
+ , text >=1.2
+ , tinylog >=0.12
+ , types-common >=0.8
, unliftio-core
, uuid
- , wai >=3.2
+ , wai >=3.2
, wai-middleware-gunzip
- , wai-utilities >=0.14.3
+ , wai-utilities >=0.14.3
, wire-api
+ , wire-otel
default-language: GHC2021
diff --git a/services/proxy/src/Proxy/Run.hs b/services/proxy/src/Proxy/Run.hs
index 3a7b33eb3d0..6090ca32442 100644
--- a/services/proxy/src/Proxy/Run.hs
+++ b/services/proxy/src/Proxy/Run.hs
@@ -32,6 +32,8 @@ import Imports hiding (head)
import Network.Wai (Middleware, Request, requestHeaders)
import Network.Wai.Middleware.Gunzip qualified as GZip
import Network.Wai.Utilities.Server
+import OpenTelemetry.Instrumentation.Wai qualified as OtelWai
+import OpenTelemetry.Trace as Otel
import Proxy.API.Internal as I
import Proxy.API.Public as P
import Proxy.Env
@@ -41,6 +43,7 @@ import Servant qualified
import Wire.API.Routes.Public.Proxy
import Wire.API.Routes.Version
import Wire.API.Routes.Version.Wai
+import Wire.OpenTelemetry (withTracer)
type CombinedAPI = InternalAPI Servant.:<|> ProxyAPI
@@ -48,22 +51,24 @@ combinedSitemap :: Env -> Servant.ServerT CombinedAPI Proxy
combinedSitemap env = I.servantSitemap Servant.:<|> P.servantSitemap env
run :: Opts -> IO ()
-run o = do
+run o = withTracer "proxy" $ \tracer -> do
e <- createEnv o
let s = newSettings $ defaultServer (o ^. proxy . to (T.unpack . host)) (o ^. proxy . to port) (e ^. applog)
let metricsMW :: Middleware
metricsMW = waiPrometheusMiddlewarePaths (routesToPaths @ProxyAPI <> routesToPaths @InternalAPI)
- middleware :: Middleware
+ otelMiddleware <- OtelWai.newOpenTelemetryWaiMiddleware
+ let middleware :: Middleware
middleware =
versionMiddleware (foldMap expandVersionExp (o ^. disabledAPIVersions))
. requestIdMiddleware (e ^. applog) defaultRequestIdHeaderName
+ . otelMiddleware
. metricsMW
. GZip.gunzip
. catchErrors (e ^. applog) defaultRequestIdHeaderName
- runSettingsWithShutdown s (middleware (mkApp e)) Nothing `finally` destroyEnv e
+ inSpan tracer "proxy" defaultSpanArguments {kind = Otel.Server} (runSettingsWithShutdown s (middleware (mkApp e)) Nothing) `finally` destroyEnv e
mkApp :: Env -> Servant.Application
mkApp env req = Servant.serve (Servant.Proxy @CombinedAPI) toServantSitemap req
diff --git a/services/spar/default.nix b/services/spar/default.nix
index a63782d1423..0e8a56a3330 100644
--- a/services/spar/default.nix
+++ b/services/spar/default.nix
@@ -23,6 +23,9 @@
, exceptions
, extended
, filepath
+, hs-opentelemetry-api
+, hs-opentelemetry-instrumentation-wai
+, hs-opentelemetry-sdk
, hscim
, HsOpenSSL
, hspec
@@ -75,6 +78,7 @@
, wai-middleware-gunzip
, wai-utilities
, wire-api
+, wire-otel
, wire-subsystems
, xml-conduit
, yaml
@@ -101,6 +105,9 @@ mkDerivation {
crypton-x509
exceptions
extended
+ hs-opentelemetry-api
+ hs-opentelemetry-instrumentation-wai
+ hs-opentelemetry-sdk
hscim
hspec
http-types
@@ -133,6 +140,7 @@ mkDerivation {
wai-middleware-gunzip
wai-utilities
wire-api
+ wire-otel
wire-subsystems
yaml
];
diff --git a/services/spar/spar.cabal b/services/spar/spar.cabal
index 6512673b66b..b14d36610d5 100644
--- a/services/spar/spar.cabal
+++ b/services/spar/spar.cabal
@@ -167,6 +167,9 @@ library
, crypton-x509
, exceptions
, extended
+ , hs-opentelemetry-api
+ , hs-opentelemetry-instrumentation-wai
+ , hs-opentelemetry-sdk
, hscim
, hspec
, http-types
@@ -199,6 +202,7 @@ library
, wai-middleware-gunzip
, wai-utilities
, wire-api
+ , wire-otel
, wire-subsystems
, yaml
diff --git a/services/spar/src/Spar/API.hs b/services/spar/src/Spar/API.hs
index d1e255da57c..f76509cdd9d 100644
--- a/services/spar/src/Spar/API.hs
+++ b/services/spar/src/Spar/API.hs
@@ -78,6 +78,10 @@ import Network.Wai (Request, requestHeaders)
import qualified Network.Wai.Utilities.Error as Wai
import Network.Wai.Utilities.Request
import Network.Wai.Utilities.Server (defaultRequestIdHeaderName)
+import qualified OpenTelemetry.Context as Otel (lookupSpan)
+import qualified OpenTelemetry.Context.ThreadLocal as Otel
+import OpenTelemetry.Trace as Otel
+import OpenTelemetry.Trace.Core as Otel
import Polysemy
import Polysemy.Error
import Polysemy.Input
@@ -142,7 +146,16 @@ import qualified Wire.Sem.Random as Random
app :: Env -> Application
app ctx0 req cont = do
let rid = getRequestId defaultRequestIdHeaderName req
- let ctx = ctx0 {sparCtxRequestId = rid}
+ mbSpan <- Otel.lookupSpan <$> Otel.getContext
+ immSpan <- case mbSpan of
+ Nothing -> pure Nothing
+ Just span' -> either (const Nothing) Just <$> Otel.toImmutableSpan span'
+
+ let ctx =
+ ctx0
+ { sparCtxRequestId = rid,
+ sparCtxOtelLocalRootSpanContext = Otel.spanContext <$> immSpan
+ }
SAML.setHttpCachePolicy
( serveWithContext
(Proxy @SparAPI)
diff --git a/services/spar/src/Spar/App.hs b/services/spar/src/Spar/App.hs
index 8ec7c658523..076719f9348 100644
--- a/services/spar/src/Spar/App.hs
+++ b/services/spar/src/Spar/App.hs
@@ -57,6 +57,7 @@ import Data.These
import Imports hiding (MonadReader, asks, log)
import qualified Network.HTTP.Types.Status as Http
import qualified Network.Wai.Utilities.Error as Wai
+import qualified OpenTelemetry.Trace as Otel
import Polysemy
import Polysemy.Error
import SAML2.Util (renderURI)
@@ -117,7 +118,8 @@ data Env = Env
sparCtxHttpBrig :: Bilge.Request,
sparCtxHttpGalley :: Bilge.Request,
sparCtxRequestId :: RequestId,
- sparCtxScimSubsystemConfig :: ScimSubsystemConfig
+ sparCtxScimSubsystemConfig :: ScimSubsystemConfig,
+ sparCtxOtelLocalRootSpanContext :: Maybe Otel.SpanContext
}
-- | Get a user by UserRef, no matter what the team.
diff --git a/services/spar/src/Spar/CanonicalInterpreter.hs b/services/spar/src/Spar/CanonicalInterpreter.hs
index 235a7f78392..849e805b3f5 100644
--- a/services/spar/src/Spar/CanonicalInterpreter.hs
+++ b/services/spar/src/Spar/CanonicalInterpreter.hs
@@ -26,7 +26,10 @@ where
import qualified Cassandra as Cas
import Control.Monad.Except hiding (mapError)
+import Data.Id (RequestId, unRequestId)
import Imports
+import qualified OpenTelemetry.Trace as Otel
+import qualified OpenTelemetry.Trace.Id as Otel
import Polysemy
import Polysemy.Error
import Polysemy.Input (Input, runInputConst)
@@ -62,6 +65,7 @@ import Spar.Sem.ScimUserTimesStore.Cassandra (scimUserTimesStoreToCassandra)
import Spar.Sem.Utils
import Spar.Sem.VerdictFormatStore (VerdictFormatStore)
import Spar.Sem.VerdictFormatStore.Cassandra (verdictFormatStoreToCassandra)
+import qualified System.Logger as Log
import qualified System.Logger as TinyLog
import Wire.API.Routes.Version (expandVersionExp)
import Wire.API.User.Saml (TTLError)
@@ -79,6 +83,7 @@ import Wire.Rpc (Rpc, runRpcWithHttp)
import Wire.RpcException
import Wire.ScimSubsystem
import Wire.ScimSubsystem.Interpreter
+import Wire.Sem.Logger
import Wire.Sem.Logger.TinyLog (loggerToTinyLog, stringLoggerToTinyLog)
import Wire.Sem.Now (Now)
import Wire.Sem.Now.IO (nowToIO)
@@ -134,7 +139,7 @@ runSparToIO ctx =
. randomToIO
. runInputConst (sparCtxLogger ctx)
. runInputConst (sparCtxOpts ctx)
- . loggerToTinyLog (sparCtxLogger ctx)
+ . loggerToTinyOtel ctx.sparCtxRequestId (sparCtxOtelLocalRootSpanContext ctx) (sparCtxLogger ctx)
. stringLoggerToTinyLog
. reporterToTinyLogWai
. runError @SparError
@@ -167,6 +172,29 @@ runSparToIO ctx =
. interpretScimSubsystem
. interpretIdPSubsystem (enableIdPByEmailDiscovery . sparCtxOpts $ ctx)
+-- | Log the request ID along with the message
+loggerToTinyOtel ::
+ (Member (Embed IO) r) =>
+ RequestId ->
+ Maybe Otel.SpanContext ->
+ Log.Logger ->
+ Sem (TinyLog ': r) a ->
+ Sem r a
+loggerToTinyOtel r Nothing tinylog =
+ loggerToTinyLog tinylog
+ . mapLogger (Log.field "request" (unRequestId r) .)
+ . raiseUnder @TinyLog
+loggerToTinyOtel r (Just spanContext) tinylog =
+ loggerToTinyLog tinylog
+ . mapLogger
+ ( \m ->
+ Log.field "request" (unRequestId r)
+ . Log.field "trace_id" (Otel.traceIdBaseEncodedByteString Otel.Base16 $ Otel.traceId spanContext)
+ . Log.field "span_id" (Otel.spanIdBaseEncodedByteString Otel.Base16 $ Otel.spanId spanContext)
+ . m
+ )
+ . raiseUnder @TinyLog
+
iParseException :: (Member (Error SparError) r) => InterpreterFor (Error ParseException) r
iParseException = Polysemy.Error.mapError (httpErrorToSparError . parseExceptionToHttpError)
@@ -177,4 +205,4 @@ runSparToHandler :: Env -> Sem CanonicalEffs a -> Handler a
runSparToHandler ctx spar = do
liftIO (runSparToIO ctx spar) >>= \case
Right val -> pure val
- Left err -> sparToServerErrorWithLogging (sparCtxLogger ctx) err >>= throwError
+ Left sparErr -> sparToServerErrorWithLogging (sparCtxLogger ctx) sparErr >>= throwError
diff --git a/services/spar/src/Spar/Run.hs b/services/spar/src/Spar/Run.hs
index 36482a10abf..3809a3a1461 100644
--- a/services/spar/src/Spar/Run.hs
+++ b/services/spar/src/Spar/Run.hs
@@ -33,6 +33,7 @@ import Cassandra as Cas
import Cassandra.Util (initCassandraForService)
import Control.Exception (ErrorCall (ErrorCall), throwIO)
import Control.Lens (to, (^.))
+import Control.Monad.Catch (finally)
import qualified Data.ByteString.UTF8 as UTF8
import Data.Id
import Data.Metrics.Servant (servantPrometheusMiddleware)
@@ -45,6 +46,8 @@ import qualified Network.Wai as Wai
import qualified Network.Wai.Middleware.Gunzip as GZip
import Network.Wai.Utilities.Server
import qualified Network.Wai.Utilities.Server as WU
+import qualified OpenTelemetry.Instrumentation.Wai as OtelWai
+import OpenTelemetry.Trace as Otel
import qualified SAML2.WebSSO as SAML
import Spar.API (SparAPI, app)
import Spar.App
@@ -59,6 +62,7 @@ import Util.Options
import qualified Web.Scim.Schema.Common as Scim
import Wire.API.Routes.Version (expandVersionExp)
import Wire.API.Routes.Version.Wai
+import Wire.OpenTelemetry (withTracer)
import Wire.ScimSubsystem.Interpreter
----------------------------------------------------------------------
@@ -77,13 +81,13 @@ initCassandra opts lgr =
-- servant / wai / warp
runServer :: Opts -> IO ()
-runServer sparCtxOpts = do
+runServer sparCtxOpts = withTracer "spar" \tracer -> do
let shost :: String = sparCtxOpts ^. to saml . SAML.cfgSPHost
sport :: Int = sparCtxOpts ^. to saml . SAML.cfgSPPort
(wrappedApp, ctxOpts) <- mkApp sparCtxOpts
let logger = sparCtxLogger ctxOpts
let settings = newSettings $ defaultServer shost (fromIntegral sport) logger
- WU.runSettingsWithShutdown settings wrappedApp Nothing
+ inSpan tracer "spar" defaultSpanArguments {kind = Otel.Server} (WU.runSettingsWithShutdown settings wrappedApp Nothing) `finally` pure ()
mkApp :: Opts -> IO (Application, Env)
mkApp sparCtxOpts = do
@@ -100,6 +104,7 @@ mkApp sparCtxOpts = do
. Bilge.port (sparCtxOpts ^. to galley . to port)
$ Bilge.empty
let sparCtxRequestId = RequestId defRequestId
+ sparCtxOtelLocalRootSpanContext = Nothing
sparCtxScimSubsystemConfig <- do
let bsUri :: URI.URI
bsUri = sparCtxOpts.scimBaseUri
@@ -122,9 +127,11 @@ mkApp sparCtxOpts = do
if Wai.requestMethod req == "POST" && Wai.pathInfo req == ["sso", "finalize-login"]
then Just out
else Nothing
+ otelMiddleware <- OtelWai.newOpenTelemetryWaiMiddleware
let middleware =
versionMiddleware (foldMap expandVersionExp (disabledAPIVersions sparCtxOpts))
. requestIdMiddleware (ctx0.sparCtxLogger) defaultRequestIdHeaderName
+ . otelMiddleware
. WU.heavyDebugLogging heavyLogOnly logLevel sparCtxLogger defaultRequestIdHeaderName
. servantPrometheusMiddleware (Proxy @SparAPI)
. GZip.gunzip
diff --git a/services/spar/test-integration/Util/Core.hs b/services/spar/test-integration/Util/Core.hs
index b172bac74b1..f269cfb0cad 100644
--- a/services/spar/test-integration/Util/Core.hs
+++ b/services/spar/test-integration/Util/Core.hs
@@ -276,6 +276,7 @@ mkEnv tstOpts opts = do
sparCtxHttpBrig = brig empty
sparCtxHttpGalley = galley empty
sparCtxRequestId = RequestId ""
+ sparCtxOtelLocalRootSpanContext = Nothing
sparCtxScimSubsystemConfig = error "mkEnv: implement sparCtxScimSubsystemConfig when needed"
pure $
TestEnv
diff --git a/services/wire-server-enterprise b/services/wire-server-enterprise
index 0cbae9136e4..c39f49ede13 160000
--- a/services/wire-server-enterprise
+++ b/services/wire-server-enterprise
@@ -1 +1 @@
-Subproject commit 0cbae9136e4e0ea54e621ddef45b86966223a135
+Subproject commit c39f49ede131e21da8e4fade9999d103763d7e47