manage schema cache when horizontally scaled (closes #1182)

server/graphql-engine.cabal

@@ -149,6 +149,7 @@ library
                      , Hasura.Server.Version
                      , Hasura.Server.CheckUpdates
                      , Hasura.Server.Telemetry
+                     , Hasura.Server.CacheUpdate
                      , Hasura.RQL.Types
                      , Hasura.RQL.Instances
                      , Hasura.RQL.Types.SchemaCache
@@ -316,6 +317,7 @@ executable graphql-engine
                      , wreq
                      , connection
                      , string-conversions
+                     , uuid
 
   other-modules:     Ops
 

server/src-exec/Main.hs

@@ -8,28 +8,32 @@ import           Options.Applicative
 import           System.Environment         (getEnvironment, lookupEnv)
 import           System.Exit                (exitFailure)
 
+
 import qualified Control.Concurrent         as C
+import qualified Control.Concurrent.STM     as STM
 import qualified Data.Aeson                 as A
 import qualified Data.ByteString.Char8      as BC
 import qualified Data.ByteString.Lazy       as BL
 import qualified Data.ByteString.Lazy.Char8 as BLC
 import qualified Data.Text                  as T
+import qualified Data.UUID.V4               as UUID
 import qualified Data.Yaml                  as Y
 import qualified Network.HTTP.Client        as HTTP
 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.Server.Auth
+import           Hasura.Server.CacheUpdate
 import           Hasura.Server.CheckUpdates (checkForUpdates)
 import           Hasura.Server.Init
+import           Hasura.Server.Logging
 import           Hasura.Server.Query        (peelRun)
 import           Hasura.Server.Telemetry
 import           Hasura.Server.Version      (currentVersion)
@@ -100,6 +104,7 @@ main =  do
   -- global http manager
   httpManager <- HTTP.newManager HTTP.tlsManagerSettings
   loggerCtx   <- mkLoggerCtx $ defaultLoggerSettings True
+  serverId <- UUID.nextRandom
   let logger = mkLogger loggerCtx
   case hgeCmd of
     HCServe so@(ServeOptions port host cp isoL mAccessKey mAuthHook mJwtSecret
@@ -117,15 +122,30 @@ main =  do
       -- log postgres connection info
       unLogger logger $ connInfoToLog ci
 
+      -- create empty cache update events queue
+      eventsQueue <- STM.newTQueueIO
+
+      -- start postgres cache update events listener thread in background
+      listenerPool <- getMinimalPool ci
+      listenerTId <- C.forkIO $ cacheUpdateEventListener listenerPool logger eventsQueue
+      unLogger logger $ mkThreadLog listenerTId serverId TTListener
+
       -- 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
-                         am corsCfg enableConsole enableTelemetry
+
+      (app, cacheRef, cacheLk, cacheBuiltTime) <-
+        mkWaiApp isoL loggerCtx pool httpManager am corsCfg enableConsole
+          enableTelemetry serverId
+
+      -- start cache update events processor thread in background
+      procTId <- C.forkIO $ cacheUpdateEventProcessor pool logger httpManager
+                              eventsQueue cacheRef cacheLk serverId cacheBuiltTime
+      unLogger logger $ mkThreadLog procTId serverId TTProcessor
 
       let warpSettings = Warp.setPort port $ Warp.setHost host Warp.defaultSettings
 
@@ -209,6 +229,14 @@ main =  do
       res <- runTx ci unlockAllEvents
       either printErrJExit return res
 
+    mkThreadLog threadId serverId threadType =
+      let msg = T.pack (show threadType) <> " thread started"
+      in StartupLog LevelInfo "threads" $
+           A.object [ "server_id" A..= serverId
+                    , "thread_id" A..= show threadId
+                    , "message" A..= msg
+                    ]
+
     getUniqIds ci = do
       eDbId <- runTx ci getDbId
       dbId <- either printErrJExit return eDbId

server/src-exec/Ops.hs

@@ -23,7 +23,7 @@ import qualified Database.PG.Query            as Q
 import qualified Database.PG.Query.Connection as Q
 
 curCatalogVer :: T.Text
-curCatalogVer = "9"
+curCatalogVer = "10"
 
 initCatalogSafe
   :: (QErrM m, UserInfoM m, CacheRWM m, MonadTx m, MonadIO m, HasHttpManager m)
@@ -318,6 +318,12 @@ from8To9 = do
     migrateMetadataFrom8 =
       $(unTypeQ (Y.decodeFile "src-rsr/migrate_metadata_from_8_to_9.yaml" :: Q (TExp RQLQuery)))
 
+from9To10 :: MonadTx m => m ()
+from9To10 = do
+  -- migrate database
+  Q.Discard () <- liftTx $ Q.multiQE defaultTxErrorHandler
+    $(Q.sqlFromFile "src-rsr/migrate_from_9_to_10.sql")
+  return ()
 
 migrateCatalog
   :: (MonadTx m, CacheRWM m, MonadIO m, UserInfoM m, HasHttpManager m)
@@ -335,12 +341,17 @@ migrateCatalog migrationTime = do
      | preVer == "6"   -> from6ToCurrent
      | preVer == "7"   -> from7ToCurrent
      | preVer == "8"   -> from8ToCurrent
+     | preVer == "9"   -> from9ToCurrent
      | otherwise -> throw400 NotSupported $
                     "unsupported version : " <> preVer
   where
+    from9ToCurrent = do
+      from9To10
+      postMigrate
+
     from8ToCurrent = do
       from8To9
-      postMigrate
+      from9ToCurrent
 
     from7ToCurrent = do
       from7To8

server/src-lib/Hasura/Server/App.hs

@@ -10,6 +10,7 @@ import           Data.Aeson                             hiding (json)
 import           Data.IORef
 import           Data.Time.Clock                        (UTCTime,
                                                          getCurrentTime)
+import           Data.UUID                              (UUID)
 import           Network.Wai                            (requestHeaders,
                                                          strictRequestBody)
 import           Web.Spock.Core
@@ -96,6 +97,7 @@ data ServerCtx
   , scCacheLock :: MVar ()
   , scAuthMode  :: AuthMode
   , scManager   :: HTTP.Manager
+  , scServerId  :: UUID
   }
 
 data HandlerCtx
@@ -214,7 +216,8 @@ v1QueryHandler query = do
       httpMgr <- scManager . hcServerCtx <$> ask
       pool <- scPGPool . hcServerCtx <$> ask
       isoL <- scIsolation . hcServerCtx <$> ask
-      runQuery pool isoL userInfo schemaCache httpMgr query
+      serverId <- scServerId . hcServerCtx <$> ask
+      runQuery pool isoL serverId userInfo schemaCache httpMgr query
 
     -- Also update the schema cache
     dbActionReload = do
@@ -289,18 +292,24 @@ mkWaiApp
   -> CorsConfig
   -> Bool
   -> Bool
-  -> IO (Wai.Application, IORef SchemaCache)
-mkWaiApp isoLevel loggerCtx pool httpManager mode corsCfg enableConsole enableTelemetry = do
-    cacheRef <- do
+  -> UUID
+  -> IO (Wai.Application, IORef SchemaCache, MVar (), UTCTime)
+mkWaiApp isoLevel loggerCtx pool httpManager mode corsCfg
+         enableConsole enableTelemetry serverId = do
+    (cacheRef, cacheBuiltTime) <- do
       pgResp <- runExceptT $ peelRun emptySchemaCache adminUserInfo
-                httpManager pool Q.Serializable buildSchemaCache
-      either initErrExit return pgResp >>= newIORef . snd
+                httpManager pool Q.Serializable $ do
+                  buildSchemaCache
+                  liftIO getCurrentTime
+      (time, sc) <- either initErrExit return pgResp
+      scRef <- newIORef sc
+      return (scRef, time)
 
     cacheLock <- newMVar ()
 
     let serverCtx =
           ServerCtx isoLevel pool (L.mkLogger loggerCtx) cacheRef
-          cacheLock mode httpManager
+          cacheLock mode httpManager serverId
 
     spockApp <- spockAsApp $ spockT id $
                 httpApp corsCfg serverCtx enableConsole enableTelemetry
@@ -309,7 +318,11 @@ mkWaiApp isoLevel loggerCtx pool httpManager mode corsCfg enableConsole enableTe
 
     wsServerEnv <- WS.createWSServerEnv (scLogger serverCtx) httpManager cacheRef runTx
     let wsServerApp = WS.createWSServerApp mode wsServerEnv
-    return (WS.websocketsOr WS.defaultConnectionOptions wsServerApp spockApp, cacheRef)
+    return ( WS.websocketsOr WS.defaultConnectionOptions wsServerApp spockApp
+           , cacheRef
+           , cacheLock
+           , cacheBuiltTime
+           )
 
 httpApp :: CorsConfig -> ServerCtx -> Bool -> Bool -> SpockT IO ()
 httpApp corsCfg serverCtx enableConsole enableTelemetry = do

server/src-lib/Hasura/Server/CacheUpdate.hs

@@ -0,0 +1,178 @@
+module Hasura.Server.CacheUpdate
+  ( ThreadType(..)
+  , cacheUpdateEventListener
+  , cacheUpdateEventProcessor
+  )
+where
+
+import           Hasura.Prelude
+
+import           Hasura.Logging
+import           Hasura.RQL.DDL.Schema.Table (buildSchemaCache)
+import           Hasura.RQL.Types
+import           Hasura.Server.App           (withLock)
+import           Hasura.Server.Query
+import           Hasura.Server.Utils         (bsToTxt)
+
+import           Control.Concurrent.MVar
+import           Data.Aeson
+import           Data.Aeson.Casing
+import           Data.Aeson.TH
+import           Data.UUID
+import           Database.PG.Query
+
+import qualified Control.Concurrent          as C
+import qualified Control.Concurrent.STM      as STM
+import qualified Data.IORef                  as IORef
+import qualified Data.Text                   as T
+import qualified Data.Time                   as UTC
+import qualified Database.PostgreSQL.LibPQ   as PQ
+import qualified Network.HTTP.Client         as HTTP
+
+pgChannel :: PGChannel
+pgChannel = "hasura_cache_update"
+
+data ThreadType
+  = TTListener
+  | TTProcessor
+  deriving (Eq)
+
+instance Show ThreadType where
+  show TTListener  = "listener"
+  show TTProcessor = "processor"
+
+
+data CacheUpdateEventLog
+  = CacheUpdateEventLog
+  { cuelLogLevel   :: !LogLevel
+  , cuelThreadType :: !ThreadType
+  , cuelInfo       :: !Value
+  } deriving (Show, Eq)
+
+instance ToJSON CacheUpdateEventLog where
+  toJSON (CacheUpdateEventLog _ t info) =
+    object [ "thread_type" .= show t
+           , "info" .= info
+           ]
+
+instance ToEngineLog CacheUpdateEventLog where
+  toEngineLog threadLog =
+    (cuelLogLevel threadLog, "cache_update_event", toJSON threadLog)
+
+data EventPayload
+  = EventPayload
+  { _epServerId   :: !UUID
+  , _epOccurredAt :: !UTC.UTCTime
+  } deriving (Show, Eq)
+$(deriveJSON (aesonDrop 3 snakeCase) ''EventPayload)
+
+data CacheUpdateEvent
+  = CUEListenSuccess !EventPayload
+  | CUEListenFail
+  deriving (Show, Eq)
+
+instance ToJSON CacheUpdateEvent where
+  toJSON (CUEListenSuccess payload) = toJSON payload
+  toJSON CUEListenFail              = String "event listening failed"
+
+data ThreadError
+  = TEJsonParse !T.Text
+  | TEQueryError !QErr
+$(deriveToJSON
+  defaultOptions { constructorTagModifier = snakeCase . drop 2
+                 , sumEncoding = TaggedObject "type" "info"
+                 }
+ ''ThreadError)
+
+-- | An IO action that listens to postgres for events and pushes them to a Queue
+cacheUpdateEventListener
+  :: PGPool
+  -> Logger
+  -> STM.TQueue CacheUpdateEvent
+  -> IO ()
+cacheUpdateEventListener pool logger eventsQueue =
+  -- Never exits
+  forever $ do
+    listenResE <- liftIO $ runExceptT $ listen pool pgChannel notifyHandler
+    either onError return listenResE
+  where
+    threadType = TTListener
+
+    onError e = do
+      logError logger threadType $ TEQueryError e
+      -- Push a listen failed event to queue
+      STM.atomically $ STM.writeTQueue eventsQueue CUEListenFail
+      logInfo logger threadType $
+        object ["message" .= ("retrying in 10 seconds" :: T.Text)]
+      C.threadDelay $ 10 * 1000 * 1000
+
+    -- Postgres notification handler
+    notifyHandler notif = do
+      let eventChannel = PGChannel $ bsToTxt $ PQ.notifyRelname notif
+      when (eventChannel == pgChannel) $
+        case eitherDecodeStrict $ PQ.notifyExtra notif of
+          Left e -> logError logger threadType $ TEJsonParse $ T.pack e
+          Right payload -> do
+            let newEvent = CUEListenSuccess payload
+            logInfo logger threadType $ object [ "received_event" .= newEvent]
+            -- Push a success event to Queue along with event payload
+            STM.atomically $ STM.writeTQueue eventsQueue newEvent
+
+-- | An IO action that processes events from Queue
+cacheUpdateEventProcessor
+  :: PGPool
+  -> Logger
+  -> HTTP.Manager
+  -> STM.TQueue CacheUpdateEvent
+  -> IORef.IORef SchemaCache
+  -> MVar ()
+  -> UUID
+  -> UTC.UTCTime
+  -> IO ()
+cacheUpdateEventProcessor pool logger httpManager eventsQueue
+                       cacheRef lk serverId cacheInit = do
+  -- Initiate previous event IO reference
+  prevEventRef <- IORef.newIORef Nothing
+
+  -- Never exits
+  forever $ do
+    event <- STM.atomically $ STM.readTQueue eventsQueue
+    prevEvent <- IORef.readIORef prevEventRef
+    logInfo logger threadType $
+      object [ "previous_event" .= prevEvent
+             , "processed_event" .= event
+             ]
+    when (shouldReload prevEvent event) $ do
+      -- Build schema cache from catalog
+      scE <- liftIO $ runExceptT $ withLock lk $
+             snd <$> peelRun emptySchemaCache adminUserInfo
+                     httpManager pool Serializable buildSchemaCache
+      case scE of
+        Left e -> logError logger threadType $ TEQueryError e
+        Right sc -> do
+          -- Updata cache reference with newly built cache
+          IORef.writeIORef cacheRef sc
+          logInfo logger threadType $
+            object ["message" .= ("schema cache reloaded" :: T.Text)]
+
+    IORef.writeIORef prevEventRef $ Just event
+  where
+    threadType = TTProcessor
+
+    -- If previous event is failed and current event is success,
+    -- reload irrespective current event payload
+    shouldReload (Just CUEListenFail) (CUEListenSuccess _) = True
+    -- If current event is failed, do not reload
+    shouldReload _                    CUEListenFail        = False
+    shouldReload _                    (CUEListenSuccess payload) =
+      -- If event is from another server and occurred after
+      -- init schema cache built then reload
+      (_epServerId payload /= serverId) && (_epOccurredAt payload > cacheInit)
+
+logInfo :: Logger -> ThreadType -> Value -> IO ()
+logInfo logger threadType val = unLogger logger $
+  CacheUpdateEventLog LevelInfo threadType val
+
+logError :: ToJSON a => Logger -> ThreadType -> a -> IO ()
+logError logger threadType err = unLogger logger $
+  CacheUpdateEventLog LevelError threadType $ object ["error" .= toJSON err]