sync metadata cache across multiple instances connected to same db (closes #1182) (#1574)

1. Haskel library `pg-client-hs` has been updated to expose a function that helps listen to `postgres` notifications over a `channel` in this [PR](hasura/pg-client-hs#5)
2. The server records an event in a table `hdb_catalog.hdb_cache_update_event` whenever any `/v1/query` (that changes metadata) is requested. A trigger notifies a `cache update` event via `hasura_cache_update` channel
3. The server runs two concurrent threads namely `listener` and `processor`. The `listener` thread listens to events on `hasura_cache_update` channel and pushed into a `Queue`. The `processor` thread fetches events from that `Queue` and processes it. Thus server rebuilds schema cache from database and updates.

Author: rakeshkky

.circleci/config.yml

@@ -101,14 +101,17 @@ refs:
         keys:
         - server-deps-cache-{{ checksum "server/graphql-engine.cabal" }}-{{ checksum "server/stack.yaml" }}
     - *wait_for_postgres
+    - run:
+        name: Install deps
+        command: |
+          apt-get update
+          apt install --yes pgbouncer jq curl postgresql-client
     - run:
         name: Run Python tests
         environment:
           HASURA_GRAPHQL_DATABASE_URL: 'postgres://gql_test:@localhost:5432/gql_test'
           GRAPHQL_ENGINE: '/build/_server_output/graphql-engine'
         command: |
-          apt-get update
-          apt install --yes jq curl
           OUTPUT_FOLDER=/build/_server_test_output/$PG_VERSION .circleci/test-server.sh
     - run:
         name: Generate coverage report

.circleci/pgbouncer/pgbouncer.ini

@@ -0,0 +1,11 @@
+[databases]
+hs_hge_test = host=localhost port=5432 dbname=hs_hge_test user=gql_test
+
+[pgbouncer]
+listen_port = 6543
+listen_addr = 127.0.0.1
+logfile = pgbouncer/pgbouncer.log
+pidfile = pgbouncer/pgbouncer.pid
+auth_type = md5
+auth_file = pgbouncer/users.txt
+admin_users = postgres

.circleci/pgbouncer/users.txt

@@ -0,0 +1 @@
+"postgres" "postgres"

.circleci/test-server.sh

@@ -128,6 +128,8 @@ export HASURA_GRAPHQL_STRINGIFY_NUMERIC_TYPES=true
 
 PID=""
 WH_PID=""
+HS_PID=""
+
 trap stop_services ERR
 trap stop_services INT
 
@@ -356,4 +358,66 @@ if [ "$RUN_WEBHOOK_TESTS" == "true" ] ; then
 
 fi
 
+# horizontal scale test
+unset HASURA_GRAPHQL_AUTH_HOOK
+unset HASURA_GRAPHQL_AUTH_HOOK_MODE
+unset HASURA_GRAPHQL_ADMIN_SECRET
+
+echo -e "\n<########## TEST GRAPHQL-ENGINE WITH HORIZONTAL SCALING ########>\n"
+
+HASURA_HS_TEST_DB='postgres://postgres:postgres@localhost:6543/hs_hge_test'
+psql "$HASURA_GRAPHQL_DATABASE_URL" -c "create database hs_hge_test;"
+
+# create pgbouncer user
+useradd pgbouncer
+cd $CIRCLECI_FOLDER
+chown -R pgbouncer:pgbouncer pgbouncer
+
+# start pgbouncer
+pgbouncer -u pgbouncer -d pgbouncer/pgbouncer.ini
+
+cd $PYTEST_ROOT
+
+# start 1st server
+"$GRAPHQL_ENGINE" --database-url "$HASURA_HS_TEST_DB" serve >> "$OUTPUT_FOLDER/graphql-engine.log" 2>&1 & PID=$!
+wait_for_port 8080
+
+# start 2nd server
+"$GRAPHQL_ENGINE" --database-url "$HASURA_HS_TEST_DB" serve \
+                  --server-port 8081 \
+                  >> "$OUTPUT_FOLDER/hs-graphql-engine.log" 2>&1 & HS_PID=$!
+wait_for_port 8081
+
+# run test
+pytest -vv --hge-url="$HGE_URL" --pg-url="$HASURA_GRAPHQL_DATABASE_URL" --test-hge-scale-url="http://localhost:8081" test_horizontal_scale.py
+
+# Shutdown pgbouncer
+psql "postgres://postgres:postgres@localhost:6543/pgbouncer" -c "SHUTDOWN;" || true
+
+cd $CIRCLECI_FOLDER
+
+# start pgbouncer again
+pgbouncer -u pgbouncer -d pgbouncer/pgbouncer.ini
+
+cd $PYTEST_ROOT
+
+# sleep for 30 seconds
+sleep 30
+
+# run test
+pytest -vv --hge-url="$HGE_URL" --pg-url="$HASURA_GRAPHQL_DATABASE_URL" --test-hge-scale-url="http://localhost:8081" test_horizontal_scale.py
+
+# Shutdown pgbouncer
+psql "postgres://postgres:postgres@localhost:6543/pgbouncer" -c "SHUTDOWN;" || true
+
+kill $PID
+kill $HS_PID
+psql "$HASURA_GRAPHQL_DATABASE_URL" -c "drop database hs_hge_test;"
+sleep 4
+combine_hpc_reports
+unset HASURA_HS_TEST_DB
+
+
+# end horizontal scale test
+
 mv graphql-engine-combined.tix "$OUTPUT_FOLDER/graphql-engine.tix" || true

server/graphql-engine.cabal

@@ -150,6 +150,7 @@ library
                      , Hasura.Server.Version
                      , Hasura.Server.CheckUpdates
                      , Hasura.Server.Telemetry
+                     , Hasura.Server.SchemaUpdate
                      , Hasura.RQL.Types
                      , Hasura.RQL.Instances
                      , Hasura.RQL.Types.SchemaCache
@@ -322,6 +323,7 @@ executable graphql-engine
                      , wreq
                      , connection
                      , string-conversions
+                     , uuid
 
   other-modules:       Ops
                      , Migrate

server/src-exec/Main.hs

@@ -9,6 +9,7 @@ import           Options.Applicative
 import           System.Environment         (getEnvironment, lookupEnv)
 import           System.Exit                (exitFailure)
 
+
 import qualified Control.Concurrent         as C
 import qualified Data.Aeson                 as A
 import qualified Data.ByteString.Char8      as BC
@@ -21,17 +22,17 @@ import qualified Network.HTTP.Client.TLS    as HTTP
 import qualified Network.Wai.Handler.Warp   as Warp
 
 import           Hasura.Events.Lib
-import           Hasura.Logging             (Logger (..), defaultLoggerSettings,
-                                             mkLogger, mkLoggerCtx)
+import           Hasura.Logging
 import           Hasura.Prelude
 import           Hasura.RQL.DDL.Metadata    (fetchMetadata)
-import           Hasura.RQL.Types           (QErr, adminUserInfo,
-                                             emptySchemaCache)
-import           Hasura.Server.App          (mkWaiApp)
+import           Hasura.RQL.Types           (adminUserInfo, emptySchemaCache)
+import           Hasura.Server.App          (SchemaCacheRef (..), mkWaiApp)
 import           Hasura.Server.Auth
 import           Hasura.Server.CheckUpdates (checkForUpdates)
 import           Hasura.Server.Init
+import           Hasura.Server.Logging
 import           Hasura.Server.Query        (peelRun)
+import           Hasura.Server.SchemaUpdate
 import           Hasura.Server.Telemetry
 import           Hasura.Server.Version      (currentVersion)
 
@@ -97,13 +98,19 @@ printJSON = BLC.putStrLn . A.encode
 printYaml :: (A.ToJSON a) => a -> IO ()
 printYaml = BC.putStrLn . Y.encode
 
+mkPGLogger :: Logger -> Q.PGLogger
+mkPGLogger (Logger logger) (Q.PLERetryMsg msg) =
+  logger $ PGLog LevelWarn msg
+
 main :: IO ()
 main =  do
   (HGEOptionsG rci hgeCmd) <- parseArgs
   -- global http manager
   httpManager <- HTTP.newManager HTTP.tlsManagerSettings
   loggerCtx   <- mkLoggerCtx $ defaultLoggerSettings True
+  instanceId <- mkInstanceId
   let logger = mkLogger loggerCtx
+      pgLogger = mkPGLogger logger
   case hgeCmd of
     HCServe so@(ServeOptions port host cp isoL mAdminSecret mAuthHook mJwtSecret
                 mUnAuthRole corsCfg enableConsole enableTelemetry strfyNum enabledAPIs) -> do
@@ -120,15 +127,21 @@ main =  do
       -- log postgres connection info
       unLogger logger $ connInfoToLog ci
 
+      pool <- Q.initPGPool ci cp pgLogger
+
       -- safe init catalog
       initRes <- initialise logger ci httpManager
 
       -- prepare event triggers data
       prepareEvents logger ci
 
-      pool <- Q.initPGPool ci cp
-      (app, cacheRef) <- mkWaiApp isoL loggerCtx pool httpManager
-                         strfyNum am corsCfg enableConsole enableTelemetry enabledAPIs
+      (app, cacheRef, cacheInitTime) <-
+        mkWaiApp isoL loggerCtx strfyNum pool httpManager am
+          corsCfg enableConsole enableTelemetry instanceId enabledAPIs
+
+      -- start a background thread for schema sync
+      startSchemaSync strfyNum pool logger httpManager
+                      cacheRef instanceId cacheInitTime
 
       let warpSettings = Warp.setPort port $ Warp.setHost host Warp.defaultSettings
 
@@ -138,48 +151,48 @@ main =  do
 
       eventEngineCtx <- atomically $ initEventEngineCtx maxEvThrds evFetchMilliSec
 
+      let scRef = _scrCache cacheRef
       unLogger logger $
         mkGenericStrLog "event_triggers" "starting workers"
-      void $ C.forkIO $ processEventQueue hloggerCtx logEnvHeaders httpManager pool cacheRef eventEngineCtx
+      void $ C.forkIO $ processEventQueue hloggerCtx logEnvHeaders httpManager pool scRef eventEngineCtx
 
       -- start a background thread to check for updates
       void $ C.forkIO $ checkForUpdates loggerCtx httpManager
 
       -- start a background thread for telemetry
       when enableTelemetry $ do
         unLogger logger $ mkGenericStrLog "telemetry" telemetryNotice
-        void $ C.forkIO $ runTelemetry logger httpManager cacheRef initRes
+        void $ C.forkIO $ runTelemetry logger httpManager scRef initRes
 
       unLogger logger $
         mkGenericStrLog "server" "starting API server"
       Warp.runSettings warpSettings app
 
     HCExport -> do
       ci <- procConnInfo rci
-      res <- runTx ci fetchMetadata
+      res <- runTx pgLogger ci fetchMetadata
       either printErrJExit printJSON res
 
     HCClean -> do
       ci <- procConnInfo rci
-      res <- runTx ci cleanCatalog
+      res <- runTx pgLogger ci cleanCatalog
       either printErrJExit (const cleanSuccess) res
 
     HCExecute -> do
       queryBs <- BL.getContents
       ci <- procConnInfo rci
-      res <- runAsAdmin ci httpManager $ execQuery queryBs
+      res <- runAsAdmin pgLogger ci httpManager $ execQuery queryBs
       either printErrJExit BLC.putStrLn res
 
     HCVersion -> putStrLn $ "Hasura GraphQL Engine: " ++ T.unpack currentVersion
   where
 
-    runTx :: Q.ConnInfo -> Q.TxE QErr a -> IO (Either QErr a)
-    runTx ci tx = do
-      pool <- getMinimalPool ci
+    runTx pgLogger ci tx = do
+      pool <- getMinimalPool pgLogger ci
       runExceptT $ Q.runTx pool (Q.Serializable, Nothing) tx
 
-    runAsAdmin ci httpManager m = do
-      pool <- getMinimalPool ci
+    runAsAdmin pgLogger ci httpManager m = do
+      pool <- getMinimalPool pgLogger ci
       res  <- runExceptT $ peelRun emptySchemaCache adminUserInfo
               httpManager False pool Q.Serializable m
       return $ fmap fst res
@@ -188,31 +201,32 @@ main =  do
       either (printErrExit . connInfoErrModifier) return $
         mkConnInfo rci
 
-    getMinimalPool ci = do
+    getMinimalPool pgLogger ci = do
       let connParams = Q.defaultConnParams { Q.cpConns = 1 }
-      Q.initPGPool ci connParams
+      Q.initPGPool ci connParams pgLogger
 
     initialise (Logger logger) ci httpMgr = do
       currentTime <- getCurrentTime
-
+      let pgLogger = mkPGLogger $ Logger logger
       -- initialise the catalog
-      initRes <- runAsAdmin ci httpMgr $ initCatalogSafe currentTime
+      initRes <- runAsAdmin pgLogger ci httpMgr $ initCatalogSafe currentTime
       either printErrJExit (logger . mkGenericStrLog "db_init") initRes
 
       -- migrate catalog if necessary
-      migRes <- runAsAdmin ci httpMgr $ migrateCatalog currentTime
+      migRes <- runAsAdmin pgLogger ci httpMgr $ migrateCatalog currentTime
       either printErrJExit (logger . mkGenericStrLog "db_migrate") migRes
 
       -- generate and retrieve uuids
-      getUniqIds ci
+      getUniqIds pgLogger ci
 
     prepareEvents (Logger logger) ci = do
+      let pgLogger = mkPGLogger $ Logger logger
       logger $ mkGenericStrLog "event_triggers" "preparing data"
-      res <- runTx ci unlockAllEvents
+      res <- runTx pgLogger ci unlockAllEvents
       either printErrJExit return res
 
-    getUniqIds ci = do
-      eDbId <- runTx ci getDbId
+    getUniqIds pgLogger ci = do
+      eDbId <- runTx pgLogger ci getDbId
       dbId <- either printErrJExit return eDbId
       fp <- liftIO generateFingerprint
       return (dbId, fp)

server/src-exec/Migrate.hs

@@ -19,7 +19,7 @@ import qualified Data.Yaml.TH                as Y
 import qualified Database.PG.Query           as Q
 
 curCatalogVer :: T.Text
-curCatalogVer = "10"
+curCatalogVer = "11"
 
 migrateMetadata
   :: ( MonadTx m
@@ -251,6 +251,13 @@ from9To10 = liftTx $ do
     $(Q.sqlFromFile "src-rsr/migrate_from_9_to_10.sql")
   return ()
 
+from10To11 :: (MonadTx m) => m ()
+from10To11 = liftTx $ do
+  -- Migrate database
+  Q.Discard () <- Q.multiQE defaultTxErrorHandler
+    $(Q.sqlFromFile "src-rsr/migrate_from_10_to_11.sql")
+  return ()
+
 migrateCatalog
   :: ( MonadTx m
      , CacheRWM m
@@ -274,10 +281,13 @@ migrateCatalog migrationTime = do
      | preVer == "7"   -> from7ToCurrent
      | preVer == "8"   -> from8ToCurrent
      | preVer == "9"   -> from9ToCurrent
+     | preVer == "10"   -> from10ToCurrent
      | otherwise -> throw400 NotSupported $
                     "unsupported version : " <> preVer
   where
-    from9ToCurrent = from9To10 >> postMigrate
+    from10ToCurrent = from10To11 >> postMigrate
+
+    from9ToCurrent = from9To10 >> from10ToCurrent
 
     from8ToCurrent = from8To9 >> from9ToCurrent