Skip to content

Commit dc882f1

Browse files
develop7develop7
committed
otel: add opentelemetry traces
1 parent b22bb74 commit dc882f1

8 files changed

Lines changed: 188 additions & 71 deletions

File tree

‎postgrest.cabal‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,7 @@ library
5858
PostgREST.Query.QueryBuilder
5959
PostgREST.Query.SqlFragment
6060
PostgREST.Query.Statements
61+
PostgREST.OpenTelemetry
6162
PostgREST.Plan
6263
PostgREST.Plan.CallPlan
6364
PostgREST.Plan.MutatePlan
@@ -102,6 +103,9 @@ library
102103
, hasql-transaction >= 1.0.1 && < 1.1
103104
, heredoc >= 0.2 && < 0.3
104105
, http-types >= 0.12.2 && < 0.13
106+
, hs-opentelemetry-sdk >= 0.0.3.6 && < 0.0.4
107+
, hs-opentelemetry-instrumentation-wai
108+
, hs-opentelemetry-utils-exceptions
105109
, insert-ordered-containers >= 0.2.2 && < 0.3
106110
, interpolatedstring-perl6 >= 1 && < 1.1
107111
, jose >= 0.8.5.1 && < 0.12

‎src/PostgREST/App.hs‎

Lines changed: 62 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -45,24 +45,27 @@ import qualified PostgREST.Unix as Unix (installSignalHandlers)
4545

4646
import PostgREST.ApiRequest (Action (..), ApiRequest (..),
4747
Mutation (..), Target (..))
48-
import PostgREST.AppState (AppState)
48+
import PostgREST.AppState (AppState, getOTelTracer)
4949
import PostgREST.Auth (AuthResult (..))
5050
import PostgREST.Config (AppConfig (..))
5151
import PostgREST.Config.PgVersion (PgVersion (..))
52-
import PostgREST.Error (Error)
52+
import PostgREST.Error (Error (..))
5353
import PostgREST.Query (DbHandler)
5454
import PostgREST.Response.Performance (ServerTiming (..),
5555
serverTimingHeader)
5656
import PostgREST.SchemaCache (SchemaCache (..))
5757
import PostgREST.SchemaCache.Routine (Routine (..))
5858
import PostgREST.Version (docsVersion, prettyVersion)
5959

60-
import qualified Data.ByteString.Char8 as BS
61-
import qualified Data.List as L
62-
import qualified Network.HTTP.Types as HTTP
63-
import qualified Network.Socket as NS
64-
import Protolude hiding (Handler)
65-
import System.TimeIt (timeItT)
60+
import qualified Data.ByteString.Char8 as BS
61+
import qualified Data.List as L
62+
import qualified Network.HTTP.Types as HTTP
63+
import qualified Network.Socket as NS
64+
import OpenTelemetry.Instrumentation.Wai (newOpenTelemetryWaiMiddleware)
65+
import OpenTelemetry.Trace (defaultSpanArguments)
66+
import OpenTelemetry.Utils.Exceptions (inSpanM)
67+
import Protolude hiding (Handler)
68+
import System.TimeIt (timeItT)
6669

6770
type Handler = ExceptT Error
6871

@@ -87,7 +90,9 @@ run appState = do
8790
pure $ "port " <> show port
8891
AppState.logWithZTime appState $ "Listening on " <> what
8992

90-
Warp.runSettingsSocket (serverSettings conf) (AppState.getSocketREST appState) app
93+
oTelMWare <- newOpenTelemetryWaiMiddleware
94+
95+
Warp.runSettingsSocket (serverSettings conf) (AppState.getSocketREST appState) (oTelMWare app)
9196

9297
serverSettings :: AppConfig -> Warp.Settings
9398
serverSettings AppConfig{..} =
@@ -105,27 +110,28 @@ postgrest conf appState connWorker =
105110
Logger.middleware (configLogLevel conf) $
106111
-- fromJust can be used, because the auth middleware will **always** add
107112
-- some AuthResult to the vault.
108-
\req respond -> case fromJust $ Auth.getResult req of
109-
Left err -> respond $ Error.errorResponseFor err
110-
Right authResult -> do
111-
appConf <- AppState.getConfig appState -- the config must be read again because it can reload
112-
maybeSchemaCache <- AppState.getSchemaCache appState
113-
pgVer <- AppState.getPgVersion appState
114-
115-
let
116-
eitherResponse :: IO (Either Error Wai.Response)
117-
eitherResponse =
118-
runExceptT $ postgrestResponse appState appConf maybeSchemaCache pgVer authResult req
119-
120-
response <- either Error.errorResponseFor identity <$> eitherResponse
121-
-- Launch the connWorker when the connection is down. The postgrest
122-
-- function can respond successfully (with a stale schema cache) before
123-
-- the connWorker is done.
124-
when (isServiceUnavailable response) connWorker
125-
resp <- do
126-
delay <- AppState.getRetryNextIn appState
127-
return $ addRetryHint delay response
128-
respond resp
113+
\req respond -> inSpanM (getOTelTracer appState) "respond" defaultSpanArguments $
114+
case fromJust $ Auth.getResult req of
115+
Left err -> respond $ Error.errorResponseFor err
116+
Right authResult -> do
117+
appConf <- AppState.getConfig appState -- the config must be read again because it can reload
118+
maybeSchemaCache <- AppState.getSchemaCache appState
119+
pgVer <- AppState.getPgVersion appState
120+
121+
let
122+
eitherResponse :: IO (Either Error Wai.Response)
123+
eitherResponse = inSpanM (getOTelTracer appState) "eitherResponse" defaultSpanArguments $
124+
runExceptT $ postgrestResponse appState appConf maybeSchemaCache pgVer authResult req
125+
126+
response <- either Error.errorResponseFor identity <$> eitherResponse
127+
-- Launch the connWorker when the connection is down. The postgrest
128+
-- function can respond successfully (with a stale schema cache) before
129+
-- the connWorker is done.
130+
when (isServiceUnavailable response) connWorker
131+
resp <- do
132+
delay <- AppState.getRetryNextIn appState
133+
return $ addRetryHint delay response
134+
respond resp
129135

130136
postgrestResponse
131137
:: AppState.AppState
@@ -169,54 +175,54 @@ handleRequest :: AuthResult -> AppConfig -> AppState.AppState -> Bool -> Bool ->
169175
handleRequest AuthResult{..} conf appState authenticated prepared pgVer apiReq@ApiRequest{..} sCache jwtTime parseTime =
170176
case (iAction, iTarget) of
171177
(ActionRead headersOnly, TargetIdent identifier) -> do
172-
(planTime', wrPlan) <- withTiming $ liftEither $ Plan.wrappedReadPlan identifier conf sCache apiReq
173-
(txTime', resultSet) <- withTiming $ runQuery roleIsoLvl mempty (Plan.wrTxMode wrPlan) $ Query.readQuery wrPlan conf apiReq
174-
(respTime', pgrst) <- withTiming $ liftEither $ Response.readResponse wrPlan headersOnly identifier apiReq resultSet
178+
(planTime', wrPlan) <- withOTel "plan" $ withTiming $ liftEither $ Plan.wrappedReadPlan identifier conf sCache apiReq
179+
(txTime', resultSet) <- withOTel "query" $ withTiming $ runQuery roleIsoLvl [] (Plan.wrTxMode wrPlan) $ Query.readQuery wrPlan conf apiReq
180+
(respTime', pgrst) <- withOTel "response" $ withTiming $ liftEither $ Response.readResponse wrPlan headersOnly identifier apiReq resultSet
175181
return $ pgrstResponse (ServerTiming jwtTime parseTime planTime' txTime' respTime') pgrst
176182

177183
(ActionMutate MutationCreate, TargetIdent identifier) -> do
178-
(planTime', mrPlan) <- withTiming $ liftEither $ Plan.mutateReadPlan MutationCreate apiReq identifier conf sCache
179-
(txTime', resultSet) <- withTiming $ runQuery roleIsoLvl mempty (Plan.mrTxMode mrPlan) $ Query.createQuery mrPlan apiReq conf
180-
(respTime', pgrst) <- withTiming $ liftEither $ Response.createResponse identifier mrPlan apiReq resultSet
184+
(planTime', mrPlan) <- withOTel "plan" $ withTiming $ liftEither $ Plan.mutateReadPlan MutationCreate apiReq identifier conf sCache
185+
(txTime', resultSet) <- withOTel "query" $ withTiming $ runQuery roleIsoLvl [] (Plan.mrTxMode mrPlan) $ Query.createQuery mrPlan apiReq conf
186+
(respTime', pgrst) <- withOTel "response" $ withTiming $ liftEither $ Response.createResponse identifier mrPlan apiReq resultSet
181187
return $ pgrstResponse (ServerTiming jwtTime parseTime planTime' txTime' respTime') pgrst
182188

183189
(ActionMutate MutationUpdate, TargetIdent identifier) -> do
184-
(planTime', mrPlan) <- withTiming $ liftEither $ Plan.mutateReadPlan MutationUpdate apiReq identifier conf sCache
185-
(txTime', resultSet) <- withTiming $ runQuery roleIsoLvl mempty (Plan.mrTxMode mrPlan) $ Query.updateQuery mrPlan apiReq conf
186-
(respTime', pgrst) <- withTiming $ liftEither $ Response.updateResponse mrPlan apiReq resultSet
190+
(planTime', mrPlan) <- withOTel "plan" $ withTiming $ liftEither $ Plan.mutateReadPlan MutationUpdate apiReq identifier conf sCache
191+
(txTime', resultSet) <- withOTel "query" $ withTiming $ runQuery roleIsoLvl [] (Plan.mrTxMode mrPlan) $ Query.updateQuery mrPlan apiReq conf
192+
(respTime', pgrst) <- withOTel "response" $ withTiming $ liftEither $ Response.updateResponse mrPlan apiReq resultSet
187193
return $ pgrstResponse (ServerTiming jwtTime parseTime planTime' txTime' respTime') pgrst
188194

189195
(ActionMutate MutationSingleUpsert, TargetIdent identifier) -> do
190-
(planTime', mrPlan) <- withTiming $ liftEither $ Plan.mutateReadPlan MutationSingleUpsert apiReq identifier conf sCache
191-
(txTime', resultSet) <- withTiming $ runQuery roleIsoLvl mempty (Plan.mrTxMode mrPlan) $ Query.singleUpsertQuery mrPlan apiReq conf
192-
(respTime', pgrst) <- withTiming $ liftEither $ Response.singleUpsertResponse mrPlan apiReq resultSet
196+
(planTime', mrPlan) <- withOTel "plan" $ withTiming $ liftEither $ Plan.mutateReadPlan MutationSingleUpsert apiReq identifier conf sCache
197+
(txTime', resultSet) <- withOTel "query" $ withTiming $ runQuery roleIsoLvl [] (Plan.mrTxMode mrPlan) $ Query.singleUpsertQuery mrPlan apiReq conf
198+
(respTime', pgrst) <- withOTel "response" $ withTiming $ liftEither $ Response.singleUpsertResponse mrPlan apiReq resultSet
193199
return $ pgrstResponse (ServerTiming jwtTime parseTime planTime' txTime' respTime') pgrst
194200

195201
(ActionMutate MutationDelete, TargetIdent identifier) -> do
196-
(planTime', mrPlan) <- withTiming $ liftEither $ Plan.mutateReadPlan MutationDelete apiReq identifier conf sCache
197-
(txTime', resultSet) <- withTiming $ runQuery roleIsoLvl mempty (Plan.mrTxMode mrPlan) $ Query.deleteQuery mrPlan apiReq conf
198-
(respTime', pgrst) <- withTiming $ liftEither $ Response.deleteResponse mrPlan apiReq resultSet
202+
(planTime', mrPlan) <- withOTel "plan" $ withTiming $ liftEither $ Plan.mutateReadPlan MutationDelete apiReq identifier conf sCache
203+
(txTime', resultSet) <- withOTel "query" $ withTiming $ runQuery roleIsoLvl [] (Plan.mrTxMode mrPlan) $ Query.deleteQuery mrPlan apiReq conf
204+
(respTime', pgrst) <- withOTel "response" $ withTiming $ liftEither $ Response.deleteResponse mrPlan apiReq resultSet
199205
return $ pgrstResponse (ServerTiming jwtTime parseTime planTime' txTime' respTime') pgrst
200206

201207
(ActionInvoke invMethod, TargetProc identifier _) -> do
202-
(planTime', cPlan) <- withTiming $ liftEither $ Plan.callReadPlan identifier conf sCache apiReq invMethod
203-
(txTime', resultSet) <- withTiming $ runQuery (fromMaybe roleIsoLvl $ pdIsoLvl (Plan.crProc cPlan)) (pdFuncSettings $ Plan.crProc cPlan) (Plan.crTxMode cPlan) $ Query.invokeQuery (Plan.crProc cPlan) cPlan apiReq conf pgVer
204-
(respTime', pgrst) <- withTiming $ liftEither $ Response.invokeResponse cPlan invMethod (Plan.crProc cPlan) apiReq resultSet
208+
(planTime', cPlan) <- withOTel "plan" $ withTiming $ liftEither $ Plan.callReadPlan identifier conf sCache apiReq invMethod
209+
(txTime', resultSet) <- withOTel "query" $ withTiming $ runQuery (fromMaybe roleIsoLvl $ pdIsoLvl (Plan.crProc cPlan)) (pdFuncSettings $ Plan.crProc cPlan) (Plan.crTxMode cPlan) $ Query.invokeQuery (Plan.crProc cPlan) cPlan apiReq conf pgVer
210+
(respTime', pgrst) <- withOTel "response" $ withTiming $ liftEither $ Response.invokeResponse cPlan invMethod (Plan.crProc cPlan) apiReq resultSet
205211
return $ pgrstResponse (ServerTiming jwtTime parseTime planTime' txTime' respTime') pgrst
206212

207213
(ActionInspect headersOnly, TargetDefaultSpec tSchema) -> do
208-
(planTime', iPlan) <- withTiming $ liftEither $ Plan.inspectPlan apiReq
209-
(txTime', oaiResult) <- withTiming $ runQuery roleIsoLvl mempty (Plan.ipTxmode iPlan) $ Query.openApiQuery sCache pgVer conf tSchema
210-
(respTime', pgrst) <- withTiming $ liftEither $ Response.openApiResponse (T.decodeUtf8 prettyVersion, docsVersion) headersOnly oaiResult conf sCache iSchema iNegotiatedByProfile
214+
(planTime', iPlan) <- withOTel "plan" $ withTiming $ liftEither $ Plan.inspectPlan apiReq
215+
(txTime', oaiResult) <- withOTel "query" $ withTiming $ runQuery roleIsoLvl [] (Plan.ipTxmode iPlan) $ Query.openApiQuery sCache pgVer conf tSchema
216+
(respTime', pgrst) <- withOTel "response" $ withTiming $ liftEither $ Response.openApiResponse (T.decodeUtf8 prettyVersion, docsVersion) headersOnly oaiResult conf sCache iSchema iNegotiatedByProfile
211217
return $ pgrstResponse (ServerTiming jwtTime parseTime planTime' txTime' respTime') pgrst
212218

213219
(ActionInfo, TargetIdent identifier) -> do
214-
(respTime', pgrst) <- withTiming $ liftEither $ Response.infoIdentResponse identifier sCache
220+
(respTime', pgrst) <- withOTel "response" $ withTiming $ liftEither $ Response.infoIdentResponse identifier sCache
215221
return $ pgrstResponse (ServerTiming jwtTime parseTime Nothing Nothing respTime') pgrst
216222

217223
(ActionInfo, TargetProc identifier _) -> do
218-
(planTime', cPlan) <- withTiming $ liftEither $ Plan.callReadPlan identifier conf sCache apiReq ApiRequest.InvHead
219-
(respTime', pgrst) <- withTiming $ liftEither $ Response.infoProcResponse (Plan.crProc cPlan)
224+
(planTime', cPlan) <- withOTel "plan" $ withTiming $ liftEither $ Plan.callReadPlan identifier conf sCache apiReq ApiRequest.InvHead
225+
(respTime', pgrst) <- withOTel "response" $ withTiming $ liftEither $ Response.infoProcResponse (Plan.crProc cPlan)
220226
return $ pgrstResponse (ServerTiming jwtTime parseTime planTime' Nothing respTime') pgrst
221227

222228
(ActionInfo, TargetDefaultSpec _) -> do
@@ -241,6 +247,8 @@ handleRequest AuthResult{..} conf appState authenticated prepared pgVer apiReq@A
241247

242248
withTiming = calcTiming $ configServerTimingEnabled conf
243249

250+
withOTel label = inSpanM (getOTelTracer appState) label defaultSpanArguments
251+
244252
calcTiming :: Bool -> Handler IO a -> Handler IO (Maybe Double, a)
245253
calcTiming timingEnabled f = if timingEnabled
246254
then do

‎src/PostgREST/AppState.hs‎

Lines changed: 14 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@ module PostgREST.AppState
1616
, getJwtCache
1717
, getSocketREST
1818
, getSocketAdmin
19+
, getOTelTracer
1920
, init
2021
, initSockets
2122
, initWithPool
@@ -78,6 +79,7 @@ import PostgREST.Unix (createAndBindDomainSocket)
7879

7980
import Data.Streaming.Network (bindPortTCP, bindRandomPortTCP)
8081
import Data.String (IsString (..))
82+
import OpenTelemetry.Trace (Tracer)
8183
import Protolude
8284

8385
data AuthResult = AuthResult
@@ -116,19 +118,21 @@ data AppState = AppState
116118
, stateSocketREST :: NS.Socket
117119
-- | Network socket for the admin UI
118120
, stateSocketAdmin :: Maybe NS.Socket
121+
-- | OpenTelemetry tracer
122+
, oTelTracer :: Tracer
119123
}
120124

121125
type AppSockets = (NS.Socket, Maybe NS.Socket)
122126

123-
init :: AppConfig -> IO AppState
124-
init conf = do
127+
init :: AppConfig -> Tracer -> IO AppState
128+
init conf tracer = do
125129
pool <- initPool conf
126130
(sock, adminSock) <- initSockets conf
127-
state' <- initWithPool (sock, adminSock) pool conf
128-
pure state' { stateSocketREST = sock, stateSocketAdmin = adminSock }
131+
state' <- initWithPool (sock, adminSock) pool tracer conf
132+
pure state' { stateSocketREST = sock, stateSocketAdmin = adminSock}
129133

130-
initWithPool :: AppSockets -> SQL.Pool -> AppConfig -> IO AppState
131-
initWithPool (sock, adminSock) pool conf = do
134+
initWithPool :: AppSockets -> SQL.Pool -> Tracer -> AppConfig -> IO AppState
135+
initWithPool (sock, adminSock) pool tracer conf = do
132136
appState <- AppState pool
133137
<$> newIORef minimumPgVersion -- assume we're in a supported version when starting, this will be corrected on a later step
134138
<*> newIORef Nothing
@@ -144,6 +148,7 @@ initWithPool (sock, adminSock) pool conf = do
144148
<*> C.newCache Nothing
145149
<*> pure sock
146150
<*> pure adminSock
151+
<*> pure tracer
147152

148153

149154
debLogTimeout <-
@@ -272,6 +277,9 @@ getSocketREST = stateSocketREST
272277
getSocketAdmin :: AppState -> Maybe NS.Socket
273278
getSocketAdmin = stateSocketAdmin
274279

280+
getOTelTracer :: AppState -> Tracer
281+
getOTelTracer = oTelTracer
282+
275283
-- | Log to stderr with local time
276284
logWithZTime :: AppState -> Text -> IO ()
277285
logWithZTime appState txt = do

‎src/PostgREST/CLI.hs‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -17,28 +17,28 @@ import qualified Options.Applicative as O
1717
import Data.Text.IO (hPutStrLn)
1818
import Text.Heredoc (str)
1919

20-
import PostgREST.AppState (AppState)
21-
import PostgREST.Config (AppConfig (..))
22-
import PostgREST.SchemaCache (querySchemaCache)
23-
import PostgREST.Version (prettyVersion)
20+
import PostgREST.AppState (AppState)
21+
import PostgREST.Config (AppConfig (..))
22+
import PostgREST.OpenTelemetry (withTracer)
23+
import PostgREST.SchemaCache (querySchemaCache)
24+
import PostgREST.Version (prettyVersion)
2425

2526
import qualified PostgREST.App as App
2627
import qualified PostgREST.AppState as AppState
2728
import qualified PostgREST.Config as Config
2829

2930
import Protolude hiding (hPutStrLn)
3031

31-
3232
main :: CLI -> IO ()
33-
main CLI{cliCommand, cliPath} = do
33+
main CLI{cliCommand, cliPath} = withTracer "PostgREST" $ \tracer -> do
3434
conf@AppConfig{..} <-
3535
either panic identity <$> Config.readAppConfig mempty cliPath Nothing mempty mempty
3636

3737
-- Per https://github.com/PostgREST/postgrest/issues/268, we want to
3838
-- explicitly close the connections to PostgreSQL on shutdown.
3939
-- 'AppState.destroy' takes care of that.
4040
bracket
41-
(AppState.init conf)
41+
(AppState.init conf tracer)
4242
AppState.destroy
4343
(\appState -> case cliCommand of
4444
CmdDumpConfig -> do

‎src/PostgREST/OpenTelemetry.hs‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
1+
module PostgREST.OpenTelemetry (withTracer) where
2+
3+
import OpenTelemetry.Trace (InstrumentationLibrary (..), Tracer,
4+
initializeGlobalTracerProvider,
5+
makeTracer, shutdownTracerProvider,
6+
tracerOptions)
7+
import PostgREST.Version (prettyVersion)
8+
import Protolude
9+
10+
withTracer :: Text -> (Tracer -> IO c) -> IO c
11+
withTracer label f = bracket
12+
initializeGlobalTracerProvider
13+
shutdownTracerProvider
14+
(\tracerProvider -> f $ makeTracer tracerProvider instrumentationLibrary tracerOptions)
15+
where
16+
instrumentationLibrary = InstrumentationLibrary {libraryName = label, libraryVersion = decodeUtf8 prettyVersion}

‎stack.yaml‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,3 +16,13 @@ extra-deps:
1616
- hasql-pool-0.10
1717
- megaparsec-9.2.2
1818
- postgresql-libpq-0.10.0.0
19+
- hs-opentelemetry-sdk-0.0.3.6@sha256:6776705a4e0c06c6a4bfa16a9bed3ba353901f52d214ac737f57ea7f8e1ed465,3746
20+
- hs-opentelemetry-api-0.1.0.0@sha256:8af01d0c81dd1af6d3293b105178fd9bfa0057c9eb88ac24d3c440bff660abe3,3705
21+
- hs-opentelemetry-propagator-b3-0.0.1.1@sha256:f0e9da77a888b89f81e5f5186788d5ace2a665e8914f6b446712a1c2edf17743,1854
22+
- hs-opentelemetry-propagator-w3c-0.0.1.3@sha256:5dc2dbdd6b0a4e434ca5fd949e9ebe5611a5d513ef58009b935e9e810cc85d1b,1852
23+
- hs-opentelemetry-exporter-otlp-0.0.1.5@sha256:89b0a6481096a338fa6383fbdf08ccaa0eb7bb009c4cbb340894eac33e55c5de,2214
24+
- hs-opentelemetry-utils-exceptions-0.2.0.0@sha256:b0fe38a18034a2e264719104e288d648eba5e27d5e0e1dd8df6583024f1e3b8c,1579
25+
- hs-opentelemetry-instrumentation-wai-0.1.0.0@sha256:6019cf031b3edec6ff0ace0df4c2e41358b9e5d939e6c326e4e1df50726348ee,1852
26+
- hs-opentelemetry-otlp-0.0.1.0@sha256:88bb6b68f172a336f78018b0823f47363fb7408eb19f7301489f81ad4d5c0f33,2307
27+
- thread-utils-context-0.3.0.4@sha256:e763da1c6cab3b6d378fb670ca74aa9bf03c9b61b6fcf7628c56363fb0e3e71e,1671
28+
- thread-utils-finalizers-0.1.1.0@sha256:24944b71d9f1d01695a5908b4a3b44838fab870883114a323336d537995e0a5b,1381

0 commit comments

Comments
 (0)