add(observation): db pool flushes observation

DISCLAIMER:
This commit was authored entirely by a human without the assistance of LLMs.

Emit a dedicated PoolFlushed observation when the DB pool is released during schema cache reload.

Author: mkleczek

postgrest.cabal

@@ -308,6 +308,7 @@ test-suite observability
   other-modules:      ObsHelper
                       Observation.JwtCache
                       Observation.MetricsSpec
+                      Observation.SchemaCacheSpec
   build-depends:      base              >= 4.9 && < 4.20
                     , base64-bytestring >= 1 && < 1.3
                     , bytestring        >= 0.10.8 && < 0.13

src/PostgREST/AppState.hs

@@ -219,10 +219,14 @@ usePool AppState{stateObserver=observer, stateMainThreadId=mainThreadId, ..} ses
 
 -- | Flush the connection pool so that any future use of the pool will
 -- use connections freshly established after this call.
+-- | Emits PoolFlushed observation
 flushPool :: AppState -> IO ()
-flushPool AppState{..} = SQL.release statePool
+flushPool AppState{..} = do
+  SQL.release statePool
+  stateObserver PoolFlushed
 
 -- | Destroy the pool on shutdown.
+-- | Differs from flushPool in not emiting PoolFlushed observation.
 destroyPool :: AppState -> IO ()
 destroyPool AppState{..} = SQL.release statePool
 

src/PostgREST/Logger.hs

@@ -111,6 +111,9 @@ observationLogger loggerState logLevel obs = case obs of
   o@PoolRequestFullfilled ->
     when (logLevel >= LogDebug) $ do
       logWithZTime loggerState $ observationMessages o
+  o@PoolFlushed ->
+    when (logLevel >= LogDebug) $ do
+      logWithZTime loggerState $ observationMessages o
   o@JwtCacheEviction ->
     when (logLevel >= LogDebug) $ do
       logWithZTime loggerState $ observationMessages o
@@ -224,6 +227,8 @@ observationMessages = \case
     pure "Trying to borrow a connection from pool"
   PoolRequestFullfilled ->
     pure "Borrowed a connection from the pool"
+  PoolFlushed ->
+    pure "Database connection pool flushed"
   JwtCacheLookup _ ->
     pure "Looked up a JWT in JWT cache"
   JwtCacheEviction ->

src/PostgREST/Observation.hs

@@ -53,6 +53,7 @@ data Observation
   | HasqlPoolObs SQL.Observation
   | PoolRequest
   | PoolRequestFullfilled
+  | PoolFlushed
   | JwtCacheLookup Bool
   | JwtCacheEviction
   | TerminationUnixSignalObs Text

test/observability/Main.hs

@@ -17,10 +17,11 @@ import           PostgREST.SchemaCache     (querySchemaCache)
 import qualified Observation.JwtCache
 import qualified Observation.MetricsSpec
 
-import ObsHelper
-import PostgREST.Observation (Observation (HasqlPoolObs))
-import Protolude             hiding (toList, toS)
-import Test.Hspec
+import qualified Observation.SchemaCacheSpec
+import           ObsHelper
+import           PostgREST.Observation       (Observation (HasqlPoolObs))
+import           Protolude                   hiding (toList, toS)
+import           Test.Hspec
 
 main :: IO ()
 main = do
@@ -63,6 +64,8 @@ main = do
       describe "Observation.JwtCacheObs" Observation.JwtCache.spec
     before (initApp baseSchemaCache testCfg) $
       describe "Feature.MetricsSpec" Observation.MetricsSpec.spec
+    before (initApp baseSchemaCache testCfg) $
+      describe "Feature.SchemaCacheSpec" Observation.SchemaCacheSpec.spec
 
   where
     loadSCache pool conf =

test/observability/Observation/SchemaCacheSpec.hs

@@ -0,0 +1,51 @@
+{-# LANGUAGE DataKinds           #-}
+{-# LANGUAGE MonadComprehensions #-}
+{-# LANGUAGE NamedFieldPuns      #-}
+module Observation.SchemaCacheSpec where
+
+import           Network.Wai           (Application)
+import           ObsHelper
+import qualified PostgREST.AppState    as AppState
+import           PostgREST.Config      (configDbSchemas)
+import           PostgREST.Observation
+import           Protolude
+import           Test.Hspec            (SpecWith, describe, it)
+import           Test.Hspec.Wai        (getState)
+
+spec :: SpecWith (SpecState, Application)
+spec = describe "Server started with metrics enabled" $ do
+
+  it "Should emit PoolFlushed, SchemaCacheQueriedObs and SchemaCacheLoadedObs when schema cache is reloaded" $ do
+    SpecState{specAppState = appState, specObsChan} <- getState
+    let waitFor = waitForObs specObsChan
+
+    liftIO $ do
+      AppState.schemaCacheLoader appState
+
+      waitFor (1 * sec) "PoolFlushed" $ \x -> [ o | o@PoolFlushed <- pure x ]
+      waitFor (1 * sec) "SchemaCacheQueriedObs" $ \x -> [ o | o@SchemaCacheQueriedObs{} <- pure x ]
+      waitFor (1 * sec) "SchemaCacheLoadedObs" $ \x -> [ o | o@SchemaCacheLoadedObs{} <- pure x ]
+
+
+  it "Should flush pool multiple times when schema reloading retries" $ do
+    SpecState{specAppState = appState, specObsChan} <- getState
+    let waitFor = waitForObs specObsChan
+
+    liftIO $ do
+        AppState.getConfig appState >>= \cfg -> do
+          AppState.putConfig appState $ cfg { configDbSchemas = pure "bad_schema" }
+          AppState.schemaCacheLoader appState
+
+          waitFor (1 * sec) "PoolFlushed 1" $ \x -> [ o | o@PoolFlushed <- pure x ]
+          waitFor (1 * sec) "SchemaCacheErrorObs" $ \x -> [ o | o@SchemaCacheErrorObs{} <- pure x ]
+
+          -- Restore configuration
+          AppState.putConfig appState cfg
+
+        -- Wait for 2 seconds so that retry can happen
+        waitFor (2 * sec) "PoolFlushed 2" $ \x -> [ o | o@PoolFlushed <- pure x ]
+        waitFor (1 * sec) "SchemaCacheQueriedObs" $ \x -> [ o | o@SchemaCacheQueriedObs{} <- pure x ]
+        waitFor (1 * sec) "SchemaCacheLoadedObs" $ \x -> [ o | o@SchemaCacheLoadedObs{} <- pure x ]
+
+  where
+    sec = 1000000