server: changes catalog initialization and logging for pro customization (hasura#5139)

ecthiender · codingkarthik · commit 08d6ea455f76 · 2020-07-27T21:16:26.000+05:30
* new typeclass to abstract the logic of QueryLog-ing

* abstract the logic of logging websocket-server logs

  introduce a MonadWSLog typeclass

* move catalog initialization to init step

  expose a helper function to migrate catalog
  create schema cache in initialiseCtx

* expose various modules and functions for pro
diff --git a/server/commit_diff.txt b/server/commit_diff.txt
@@ -207,25 +207,6 @@ Date:   Tue Jun 23 20:51:34 2020 +0530
 
     Co-authored-by: Tirumarai Selvan <tiru@hasura.io>
 
-commit a7a60c2dfe42f9257d57d748a231bd47b164337b
-Author: Anon Ray <ecthiender@users.noreply.github.com>
-Date:   Fri Jun 19 12:12:32 2020 +0530
-
-    server: changes catalog initialization and logging for pro customization (#5139)
-
-    * new typeclass to abstract the logic of QueryLog-ing
-
-    * abstract the logic of logging websocket-server logs
-
-      introduce a MonadWSLog typeclass
-
-    * move catalog initialization to init step
-
-      expose a helper function to migrate catalog
-      create schema cache in initialiseCtx
-
-    * expose various modules and functions for pro
-
 (Done, but we should re-visit this, if we do query plan caching)
 commit 20cbe9cfd3e90b91d3f4faf370b081fc3859cbde
 Author: Auke Booij <auke@hasura.io>
diff --git a/server/graphql-engine.cabal b/server/graphql-engine.cabal
@@ -300,11 +300,13 @@ library
 
                      -- exposed for Pro
                      , Hasura.GraphQL.Execute
+                     , Hasura.GraphQL.Execute.Query
                      , Hasura.GraphQL.Execute.LiveQuery
 --                     , Hasura.GraphQL.Validate
                      , Hasura.GraphQL.Transport.HTTP
                      , Hasura.GraphQL.Transport.WebSocket.Protocol
                      , Hasura.GraphQL.Transport.WebSocket.Server
+                     , Hasura.GraphQL.Logging
 
                      , Hasura.RQL.Types
                      , Hasura.RQL.Types.Run
@@ -320,6 +322,9 @@ library
                      , Data.Aeson.Ordered
                      , Data.TByteString
 
+                     , Network.Wai.Extended
+                     , Control.Concurrent.Extended
+
   other-modules:       Hasura.Incremental.Select
                      , Hasura.Incremental.Internal.Cache
                      , Hasura.Incremental.Internal.Dependency
@@ -411,7 +416,6 @@ library
                      , Hasura.GraphQL.Explain
                      , Hasura.GraphQL.Execute.Inline
                      , Hasura.GraphQL.Execute.Plan
-                     , Hasura.GraphQL.Execute.Query
                      , Hasura.GraphQL.Execute.Types
                      , Hasura.GraphQL.Execute.Mutation
                      , Hasura.GraphQL.Execute.Resolve
@@ -433,7 +437,6 @@ library
                      -- , Hasura.GraphQL.Resolve.Select
                      , Hasura.GraphQL.RemoteServer
                      , Hasura.GraphQL.Context
-                     , Hasura.GraphQL.Logging
 
                      , Hasura.GraphQL.Parser
                      , Hasura.GraphQL.Parser.Class
@@ -460,8 +463,6 @@ library
                      , Hasura.Eventing.ScheduledTrigger
                      , Hasura.Eventing.Common
 
-                     , Network.Wai.Extended
-                     , Control.Concurrent.Extended
                      , Control.Lens.Extended
                      , Data.Aeson.Extended
                      , Data.GADT.Compare.Extended
@@ -482,7 +483,6 @@ library
                      , Hasura.SQL.Value
 
                      , Network.URI.Extended
-                     , Network.Wai.Extended
                      , Network.Wai.Handler.WebSockets.Custom
 
 executable graphql-engine
diff --git a/server/src-lib/Hasura/GraphQL/Execute.hs b/server/src-lib/Hasura/GraphQL/Execute.hs
@@ -56,6 +56,7 @@ import           Hasura.EncJSON
 import           Hasura.GraphQL.Logging
 import           Hasura.GraphQL.Transport.HTTP.Protocol
 import           Hasura.HTTP
+import           Hasura.GraphQL.RemoteServer            (execRemoteGQ')
 import           Hasura.RQL.DDL.Headers
 import           Hasura.RQL.Types
 import           Hasura.Server.Utils                    (RequestId, mkClientHeadersForward,
@@ -348,6 +349,7 @@ execRemoteGQ
      , MonadIO m
      , MonadError QErr m
      , MonadReader ExecutionCtx m
+     , MonadQueryLog m
      )
   => RequestId
   -> UserInfo
@@ -361,40 +363,8 @@ execRemoteGQ reqId userInfo reqHdrs q rsi opDef = do
   execCtx <- ask
   let logger  = _ecxLogger execCtx
       manager = _ecxHttpManager execCtx
-      opTy    = G._todType opDef
-  when (opTy == G.OperationTypeSubscription) $
-    throw400 NotSupported "subscription to remote server is not supported"
-  confHdrs <- makeHeadersFromConf hdrConf
-  let clientHdrs = bool [] (mkClientHeadersForward reqHdrs) fwdClientHdrs
-      -- filter out duplicate headers
-      -- priority: conf headers > resolved userinfo vars > client headers
-      hdrMaps    = [ Map.fromList confHdrs
-                   , Map.fromList userInfoToHdrs
-                   , Map.fromList clientHdrs
-                   ]
-      headers  = Map.toList $ foldr Map.union Map.empty hdrMaps
-      finalHeaders = addDefaultHeaders headers
-  initReqE <- liftIO $ try $ HTTP.parseRequest (show url)
-  initReq <- either httpThrow pure initReqE
-  let req = initReq
-           { HTTP.method = "POST"
-           , HTTP.requestHeaders = finalHeaders
-           , HTTP.requestBody = HTTP.RequestBodyLBS (J.encode q)
-           , HTTP.responseTimeout = HTTP.responseTimeoutMicro (timeout * 1000000)
-           }
-
-  L.unLogger logger $ QueryLog q Nothing reqId
-  (time, res)  <- withElapsedTime $ liftIO $ try $ HTTP.httpLbs req manager
-  resp <- either httpThrow return res
-  let !httpResp = HttpResponse (encJFromLBS $ resp ^. Wreq.responseBody) $ mkSetCookieHeaders resp
+      opType  = G._todType opDef
+  logQueryLog logger q Nothing reqId
+  (time, respHdrs, resp) <- execRemoteGQ' manager userInfo reqHdrs q rsi opType
+  let !httpResp = HttpResponse (encJFromLBS resp) respHdrs
   return (time, httpResp)
-
-  where
-    RemoteSchemaInfo url hdrConf fwdClientHdrs timeout = rsi
-    httpThrow :: (MonadError QErr m) => HTTP.HttpException -> m a
-    httpThrow = \case
-      HTTP.HttpExceptionRequest _req content -> throw500 $ T.pack . show $ content
-      HTTP.InvalidUrlException _url reason -> throw500 $ T.pack . show $ reason
-
-    userInfoToHdrs = userInfoToList userInfo
-      & map (CI.mk . unUTF8 . fromText *** unUTF8 . fromText)
diff --git a/server/src-lib/Hasura/GraphQL/Transport/HTTP.hs b/server/src-lib/Hasura/GraphQL/Transport/HTTP.hs
@@ -1,4 +1,5 @@
 -- | Execution of GraphQL queries over HTTP transport
+{-# LANGUAGE RecordWildCards #-}
 module Hasura.GraphQL.Transport.HTTP
   ( runGQ
   , runGQBatched
@@ -12,7 +13,7 @@ module Hasura.GraphQL.Transport.HTTP
   ) where
 
 import           Hasura.EncJSON
-import           Hasura.GraphQL.Logging
+import           Hasura.GraphQL.Logging                 (MonadQueryLog (..))
 import           Hasura.GraphQL.Transport.HTTP.Protocol
 import           Hasura.HTTP
 import           Hasura.Prelude
@@ -40,6 +41,7 @@ runGQ
      , MonadError QErr m
      , MonadReader E.ExecutionCtx m
      , E.MonadGQLExecutionCheck m
+     , MonadQueryLog m
      )
   => RequestId
   -> UserInfo
@@ -114,6 +116,7 @@ runGQBatched
      , MonadError QErr m
      , MonadReader E.ExecutionCtx m
      , E.MonadGQLExecutionCheck m
+     , MonadQueryLog m
      )
   => RequestId
   -> ResponseInternalErrorsConfig
@@ -147,6 +150,7 @@ runQueryDB
   :: ( MonadIO m
      , MonadError QErr m
      , MonadReader E.ExecutionCtx m
+     , MonadQueryLog m
      )
   => RequestId
   -> GQLReqUnparsed
@@ -156,10 +160,10 @@ runQueryDB
   -- ^ Also return 'Mutation' when the operation was a mutation, and the time
   -- spent in the PG query; for telemetry.
 runQueryDB reqId query _userInfo (tx, genSql) = do
+  -- log the generated SQL and the graphql query
   E.ExecutionCtx logger _ pgExecCtx _ _ _ _ _ <- ask
+  logQueryLog logger query (Just genSql) reqId
   (telemTimeIO, respE) <- withElapsedTime $ liftIO $ runExceptT $ do
-    -- log the generated SQL and the graphql query
-    L.unLogger logger $ QueryLog query (Just genSql) reqId
     runQueryTx pgExecCtx tx
   resp <- liftEither respE
   let !json = encodeGQResp $ GQSuccess $ encJToLBS resp
@@ -170,6 +174,7 @@ runMutationDB
   :: ( MonadIO m
      , MonadError QErr m
      , MonadReader E.ExecutionCtx m
+     , MonadQueryLog m
      )
   => RequestId
   -> GQLReqUnparsed
@@ -180,9 +185,9 @@ runMutationDB
   -- spent in the PG query; for telemetry.
 runMutationDB reqId query userInfo tx = do
   E.ExecutionCtx logger _ pgExecCtx _ _ _ _ _ <- ask
+  -- log the graphql query
+  logQueryLog logger query Nothing reqId
   (telemTimeIO, respE) <- withElapsedTime $ liftIO $ runExceptT $ do
-    -- log the graphql query
-    L.unLogger logger $ QueryLog query Nothing reqId
     runLazyTx pgExecCtx Q.ReadWrite $ withUserInfo userInfo tx
   resp <- liftEither respE
   let !json = encodeGQResp $ GQSuccess $ encJToLBS resp
@@ -194,6 +199,7 @@ runHasuraGQ
   :: ( MonadIO m
      , MonadError QErr m
      , MonadReader E.ExecutionCtx m
+     , MonadQueryLog m
      )
   => RequestId
   -> GQLReqUnparsed
@@ -203,19 +209,19 @@ runHasuraGQ
   -- ^ Also return 'Mutation' when the operation was a mutation, and the time
   -- spent in the PG query; for telemetry.
 runHasuraGQ reqId query userInfo resolvedOp = do
-  E.ExecutionCtx logger _ pgExecCtx _ _ _ _ _ <- ask
+  (E.ExecutionCtx logger _ pgExecCtx _ _ _ _ _) <- ask
+  logQuery' logger
   (telemTimeIO, respE) <- withElapsedTime $ liftIO $ runExceptT $ case resolvedOp of
-    E.ExOpQuery tx genSql  -> do
+    E.ExOpQuery tx _genSql -> do
       -- log the generated SQL and the graphql query
-      L.unLogger logger $ QueryLog query genSql reqId
+      -- L.unLogger logger $ QueryLog query genSql reqId
       ([],) <$> runQueryTx pgExecCtx tx
     E.ExOpMutation respHeaders tx -> do
-      -- log the graphql query
-      L.unLogger logger $ QueryLog query Nothing reqId
       (respHeaders,) <$> runLazyTx pgExecCtx Q.ReadWrite (withUserInfo userInfo tx)
     E.ExOpSubs _ ->
       throw400 UnexpectedPayload
       "subscriptions are not supported over HTTP, use websockets instead"
+
   (respHdrs, resp) <- liftEither respE
   let !json = encodeGQResp $ GQSuccess $ encJToLBS resp
       telemQueryType = case resolvedOp of E.ExOpMutation{} -> Telem.Mutation ; _ -> Telem.Query
diff --git a/server/src-lib/Hasura/GraphQL/Transport/WebSocket.hs b/server/src-lib/Hasura/GraphQL/Transport/WebSocket.hs
@@ -27,6 +27,7 @@ import qualified Data.Text.Encoding                          as TE
 import qualified Data.Time.Clock                             as TC
 import qualified Database.PG.Query                           as Q
 import qualified Language.GraphQL.Draft.Syntax               as G
+import qualified ListT
 import qualified Network.HTTP.Client                         as H
 import qualified Network.HTTP.Types                          as H
 import qualified Network.Wai.Extended                        as Wai
@@ -36,11 +37,10 @@ import qualified StmContainers.Map                           as STMMap
 import           Control.Concurrent.Extended                 (sleep)
 import           Control.Exception.Lifted
 import           Data.String
-import           GHC.AssertNF
-import qualified ListT
+-- import           GHC.AssertNF
 
 import           Hasura.EncJSON
-import           Hasura.GraphQL.Logging
+import           Hasura.GraphQL.Logging                      (MonadQueryLog (..))
 import           Hasura.GraphQL.Transport.HTTP.Protocol
 import           Hasura.GraphQL.Transport.WebSocket.Protocol
 import           Hasura.HTTP
@@ -121,8 +121,13 @@ sendMsgWithMetadata wsConn msg (LQ.LiveQueryMetadata execTime) =
   liftIO $ WS.sendMsg wsConn $ WS.WSQueueResponse bs wsInfo
   where
     bs = encodeServerMsg msg
+    (msgType, operationId) = case msg of
+      (SMData (DataMsg opId _)) -> (Just SMT_GQL_DATA, Just opId)
+      _                         -> (Nothing, Nothing)
     wsInfo = Just $! WS.WSEventInfo
-      { WS._wseiQueryExecutionTime = Just $! realToFrac execTime
+      { WS._wseiEventType = msgType
+      , WS._wseiOperationId = operationId
+      , WS._wseiQueryExecutionTime = Just $! realToFrac execTime
       , WS._wseiResponseSize = Just $! BL.length bs
       }
 
@@ -326,8 +331,7 @@ onStart serverEnv wsConn (StartMsg opId q) = catchAndIgnore $ do
                planCache userInfo sqlGenCtx sc scVer queryType httpMgr reqHdrs (q, reqParsed)
 
   (telemCacheHit, execPlan) <- either (withComplete . preExecErr requestId) return execPlanE
-  let execCtx = E.ExecutionCtx logger sqlGenCtx pgExecCtx
-                planCache sc scVer httpMgr enableAL
+  let execCtx = E.ExecutionCtx logger sqlGenCtx pgExecCtx planCache sc scVer httpMgr enableAL
 
   case execPlan of
     E.QueryExecutionPlan queryPlan -> do
@@ -354,15 +358,15 @@ onStart serverEnv wsConn (StartMsg opId q) = catchAndIgnore $ do
       case subscriptionPlan of
         E.ExecStepDB lqOp -> do
           -- log the graphql query
-          L.unLogger logger $ QueryLog q Nothing requestId
+          logQueryLog logger q Nothing requestId
           let subscriberMetadata = LQ.mkSubscriberMetadata $ J.object
                                    [ "websocket_id" J..= WS.getWSId wsConn
                                    , "operation_id" J..= opId
                                    ]
           -- NOTE!: we mask async exceptions higher in the call stack, but it's
           -- crucial we don't lose lqId after addLiveQuery returns successfully.
           !lqId <- liftIO $ LQ.addLiveQuery logger subscriberMetadata lqMap lqOp liveQOnChange
-          let !opName = _grOperationName q
+          -- let !opName = _grOperationName q -- TODO: uncomment this
           -- liftIO $ $assertNFHere $! (lqId, opName)  -- so we don't write thunks to mutable vars
           liftIO $ STM.atomically $
             -- NOTE: see crucial `lookup` check above, ensuring this doesn't clobber:
@@ -381,7 +385,7 @@ onStart serverEnv wsConn (StartMsg opId q) = catchAndIgnore $ do
     execQueryOrMut timerTot telemQueryType telemCacheHit telemLocality genSql requestId q' action = do
       logOpEv ODStarted (Just requestId)
       -- log the generated SQL and the graphql query
-      L.unLogger logger $ QueryLog q' genSql requestId
+      logQueryLog logger q' genSql requestId
       (withElapsedTime $ liftIO $ runExceptT action) >>= \case
         (_,      Left err) -> postExecErr requestId err
         (telemTimeIO_DT, Right encJson) -> do
@@ -405,7 +409,8 @@ onStart serverEnv wsConn (StartMsg opId q) = catchAndIgnore $ do
           runLazyTx pgExecCtx Q.ReadWrite $ withUserInfo userInfo opTx
       E.ExOpSubs lqOp -> do
         -- log the graphql query
-        L.unLogger logger $ QueryLog query Nothing reqId
+        -- L.unLogger logger $ QueryLog query Nothing reqId
+        logQueryLog logger query Nothing reqId
         let subscriberMetadata = LQ.mkSubscriberMetadata $ J.object
                                  [ "websocket_id" J..= WS.getWSId wsConn
                                  , "operation_id" J..= opId
@@ -426,8 +431,8 @@ onStart serverEnv wsConn (StartMsg opId q) = catchAndIgnore $ do
         execQueryOrMut telemQueryType genSql action = do
           logOpEv ODStarted (Just reqId)
           -- log the generated SQL and the graphql query
-          L.unLogger logger $ QueryLog query genSql reqId
-          (withElapsedTime $ liftIO $ runExceptT action) >>= \case
+          logQueryLog logger query genSql reqId
+          (withElapsedTime $ runExceptT action) >>= \case
             (_,      Left err) -> postExecErr reqId err
             (telemTimeIO_DT, Right encJson) -> do
               -- Telemetry. NOTE: don't time network IO:
diff --git a/server/src-lib/Hasura/GraphQL/Transport/WebSocket/Protocol.hs b/server/src-lib/Hasura/GraphQL/Transport/WebSocket/Protocol.hs
@@ -6,6 +6,7 @@ module Hasura.GraphQL.Transport.WebSocket.Protocol
   , StopMsg(..)
   , ClientMsg(..)
   , ServerMsg(..)
+  , ServerMsgType(..)
   , encodeServerMsg
   , DataMsg(..)
   , ErrorMsg(..)
diff --git a/server/src-lib/Hasura/GraphQL/Transport/WebSocket/Server.hs b/server/src-lib/Hasura/GraphQL/Transport/WebSocket/Server.hs
diff --git a/server/src-lib/Hasura/Server/App.hs b/server/src-lib/Hasura/Server/App.hs
diff --git a/server/src-lib/Hasura/Server/Logging.hs b/server/src-lib/Hasura/Server/Logging.hs
