add(metrics): expose db pool flushes counter

Emit a dedicated PoolFlushed observation when the DB pool is released during schema cache reload and expose it as pgrst_db_pool_flushes_total.

Author: mkleczek

CHANGELOG.md

@@ -12,6 +12,7 @@ All notable changes to this project will be documented in this file. From versio
 - Log host, port and pg version of listener database connection by @mkleczek in #4617 #4618
 - Optimize requests with `Prefer: count=exact` that do not use ranges or `db-max-rows` by @laurenceisla in #3957
   + Removed unnecessary double count when building the `Content-Range`.
+- Expose db pool flushes counter by @mkleczek in #4658
 
 ### Changed
 

docs/references/observability.rst

@@ -177,6 +177,15 @@ pgrst_db_pool_timeouts_total
 
 The total number of pool connection timeouts.
 
+pgrst_db_pool_flushes_total
+~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+======== =======
+**Type** Counter
+======== =======
+
+The total number of times the pool was flushed.
+
 pgrst_db_pool_available
 ~~~~~~~~~~~~~~~~~~~~~~~
 

postgrest.cabal

@@ -218,6 +218,7 @@ test-suite spec
                       Feature.ConcurrentSpec
                       Feature.CorsSpec
                       Feature.ExtraSearchPathSpec
+                      Feature.MetricsSpec
                       Feature.NoSuperuserSpec
                       Feature.ObservabilitySpec
                       Feature.OpenApi.DisabledOpenApiSpec

src/PostgREST/AppState.hs

@@ -271,7 +271,9 @@ 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.
 flushPool :: AppState -> IO ()
-flushPool AppState{..} = SQL.release statePool
+flushPool AppState{..} = do
+  SQL.release statePool
+  stateObserver PoolFlushed
 
 -- | Destroy the pool on shutdown.
 destroyPool :: AppState -> IO ()

src/PostgREST/Logger.hs

@@ -102,6 +102,9 @@ observationLogger loggerState logLevel obs = case obs of
   o@PoolRequestFullfilled ->
     when (logLevel >= LogDebug) $ do
       logWithZTime loggerState $ observationMessage o
+  o@PoolFlushed ->
+    when (logLevel >= LogDebug) $ do
+      logWithZTime loggerState $ observationMessage o
   o@JwtCacheEviction ->
     when (logLevel >= LogDebug) $ do
       logWithZTime loggerState $ observationMessage o

src/PostgREST/Metrics.hs

@@ -22,6 +22,7 @@ import Protolude
 data MetricsState =
   MetricsState {
     poolTimeouts         :: Counter,
+    poolFlushes          :: Counter,
     poolAvailable        :: Gauge,
     poolWaiting          :: Gauge,
     poolMaxSize          :: Gauge,
@@ -36,6 +37,7 @@ init :: Int -> IO MetricsState
 init configDbPoolSize = do
   metricState <- MetricsState <$>
     register (counter (Info "pgrst_db_pool_timeouts_total" "The total number of pool connection timeouts")) <*>
+    register (counter (Info "pgrst_db_pool_flushes_total" "The total number of times the pool was flushed")) <*>
     register (gauge (Info "pgrst_db_pool_available" "Available connections in the pool")) <*>
     register (gauge (Info "pgrst_db_pool_waiting" "Requests waiting to acquire a pool connection")) <*>
     register (gauge (Info "pgrst_db_pool_max" "Max pool connections")) <*>
@@ -64,6 +66,8 @@ observationMetrics MetricsState{..} obs = case obs of
     incGauge poolWaiting
   PoolRequestFullfilled ->
     decGauge poolWaiting
+  PoolFlushed ->
+    incCounter poolFlushes
   SchemaCacheLoadedObs resTime -> do
     withLabel schemaCacheLoads "SUCCESS" incCounter
     setGauge schemaCacheQueryTime resTime

src/PostgREST/Observation.hs

@@ -63,6 +63,7 @@ data Observation
   | HasqlPoolObs SQL.Observation
   | PoolRequest
   | PoolRequestFullfilled
+  | PoolFlushed
   | JwtCacheLookup Bool
   | JwtCacheEviction
   | WarpErrorObs Text
@@ -157,6 +158,8 @@ observationMessage = \case
     "Trying to borrow a connection from pool"
   PoolRequestFullfilled ->
     "Borrowed a connection from the pool"
+  PoolFlushed ->
+    "Database connection pool flushed"
   JwtCacheLookup _ ->
     "Looked up a JWT in JWT cache"
   JwtCacheEviction ->

test/spec/Feature/MetricsSpec.hs

@@ -0,0 +1,47 @@
+{-# LANGUAGE DataKinds        #-}
+{-# LANGUAGE LambdaCase       #-}
+{-# LANGUAGE TypeApplications #-}
+module Feature.MetricsSpec
+
+where
+
+import Network.Wai (Application)
+
+import qualified PostgREST.AppState    as AppState
+import           PostgREST.Metrics     (MetricsState (..))
+import           PostgREST.Observation
+import           Protolude
+import           SpecHelper
+import           Test.Hspec            (SpecWith, describe,
+                                        expectationFailure, it)
+import           Test.Hspec.Wai        (getState)
+
+untilM :: Int -> (a -> Bool) -> IO a -> IO (Maybe a)
+untilM timeout cond act = rightToMaybe <$> race (threadDelay timeout) (fix $
+  \loop -> do
+    value <- act
+    if cond value then
+      pure value
+    else
+      loop)
+
+waitForSchemaReload :: Int -> IO Observation -> IO (Maybe Observation)
+waitForSchemaReload timeout = untilM timeout $ \case
+  SchemaCacheLoadedObs _ -> True
+  _ -> False
+
+spec :: SpecWith ((MetricsState, AppState.AppState, IO Observation), Application)
+spec = describe "Server started with metrics enabled" $
+  it "Should increase pool flushes metric when schema cache is reloaded" $ do
+    (metrics, appState, readObs) <- getState
+
+    checkState' metrics
+
+      [
+        expectCounter @"poolFlushes" (+1)
+      ] $
+
+      liftIO $ do
+        AppState.schemaCacheLoader appState
+        waitForSchemaReload 1000000 readObs >>=
+          maybe (expectationFailure "Timeout waiting for schema reloading") mempty

test/spec/Main.hs

@@ -29,6 +29,7 @@ import qualified Feature.Auth.NoJwtSecretSpec
 import qualified Feature.ConcurrentSpec
 import qualified Feature.CorsSpec
 import qualified Feature.ExtraSearchPathSpec
+import qualified Feature.MetricsSpec
 import qualified Feature.NoSuperuserSpec
 import qualified Feature.ObservabilitySpec
 import qualified Feature.OpenApi.DisabledOpenApiSpec
@@ -87,6 +88,7 @@ main = do
   sockets <- AppState.initSockets testCfg
   loggerState <- Logger.init
   metricsState <- Metrics.init (configDbPoolSize testCfg)
+  obsChan <- newChan
 
   let
     initApp sCache st config = do
@@ -95,6 +97,12 @@ main = do
       AppState.putSchemaCache appState (Just sCache)
       return (st, postgrest (configLogLevel config) appState (pure ()))
 
+    initMetricsApp sCache config = do
+      appState <- AppState.initWithPool sockets pool config loggerState metricsState (Metrics.observationMetrics metricsState <> writeChan obsChan)
+      AppState.putPgVersion appState actualPgVersion
+      AppState.putSchemaCache appState (Just sCache)
+      return ((metricsState, appState, readChan obsChan), postgrest (configLogLevel config) appState (pure ()))
+
     -- For tests that run with the same schema cache
     app = initApp baseSchemaCache ()
 
@@ -123,6 +131,7 @@ main = do
       obsApp               = app testObservabilityCfg
       serverTiming         = app testCfgServerTiming
       aggregatesEnabled    = app testCfgAggregatesEnabled
+      metricsApp           = initMetricsApp baseSchemaCache testCfg
 
       extraSearchPathApp   = appDbs testCfgExtraSearchPath
       unicodeApp           = appDbs testUnicodeCfg
@@ -278,6 +287,9 @@ main = do
     before (initApp baseSchemaCache metricsState testCfgJwtCache) $
       describe "Feature.Auth.JwtCacheSpec" Feature.Auth.JwtCacheSpec.spec
 
+    before metricsApp $
+      describe "Feature.MetricsSpec" Feature.MetricsSpec.spec
+
   where
     loadSCache pool conf =
       either (panic.show) id <$> P.use pool (HT.transaction HT.ReadCommitted HT.Read $ querySchemaCache conf)