hserver: catch exception for stats handler and gossip nodeChangeEvent (…

…#1760)
hstreamdb · Feb 7, 2024 · 5360678 · 5360678
1 parent 78d505d
commit 5360678
Show file tree

Hide file tree

Showing 3 changed files with 23 additions and 8 deletions.
diff --git a/hstream/app/server.hs b/hstream/app/server.hs
@@ -11,10 +11,11 @@
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TypeApplications    #-}
 
-import           Control.Concurrent               (forkIO, newEmptyMVar,
+import           Control.Concurrent               (MVar, forkIO, newEmptyMVar,
                                                    putMVar, readMVar)
 import qualified Control.Concurrent.Async         as Async
-import           Control.Exception                (bracket, handle)
+import           Control.Exception                (SomeException, bracket,
+                                                   catch, handle)
 import           Control.Monad                    (forM, forM_, join, void,
                                                    when)
 import           Data.ByteString                  (ByteString)
@@ -91,6 +92,8 @@ import qualified Network.GRPC.HighLevel.Generated as GRPC
 import           Text.RawString.QQ                (r)
 #endif
 
+import qualified HStream.Gossip.Types             as Gossip
+
 -------------------------------------------------------------------------------
 
 main :: IO ()
@@ -143,7 +146,7 @@ app config@ServerOpts{..} = do
                         }
 
       scMVar <- newEmptyMVar
-      gossipContext <- initGossipContext defaultGossipOpts mempty (Just $ Cluster.nodeChangeEventHandler scMVar) serverNode _seedNodes
+      gossipContext <- initGossipContext defaultGossipOpts mempty (Just $ nodeChangeEventHd scMVar) serverNode _seedNodes
 
       serverContext <- initializeServer config gossipContext h db_m
       putMVar scMVar serverContext
@@ -342,3 +345,15 @@ showVersion = do
   API.HStreamVersion{..} <- getHStreamVersion
   putStrLn $ "version: " <> T.unpack hstreamVersionVersion
           <> " (" <> T.unpack hstreamVersionCommit <> ")"
+
+-- TODO: here, we just catch `SomeException` and then show some error messages.
+--
+-- 1. There may need to be a uniform way to handle all gossip handles(threads).
+-- 2. Retry retriable actions
+nodeChangeEventHd
+  :: MVar ServerContext
+  -> Gossip.ServerState
+  -> I.ServerNode
+  -> IO ()
+nodeChangeEventHd m s n = catch (Cluster.nodeChangeEventHandler m s n) $
+  \(e :: SomeException) -> Log.fatal $ "handle node change event error: " <> Log.buildString' e
diff --git a/hstream/src/HStream/Server/Handler/ShardReader.hs b/hstream/src/HStream/Server/Handler/ShardReader.hs
@@ -60,15 +60,15 @@ createShardReaderHandler
   -> ServerRequest 'Normal CreateShardReaderRequest CreateShardReaderResponse
   -> IO (ServerResponse 'Normal CreateShardReaderResponse)
 createShardReaderHandler sc (ServerNormalRequest _metadata request) = defaultExceptionHandle $ do
-  Log.debug $ "Receive Create ShardReader Request" <> Log.buildString' (show request)
+  Log.info $ "Receive Create ShardReader Request" <> Log.buildString' (show request)
   validateCreateShardReader request
   C.createShardReader sc request >>= returnResp
 
 handleCreateShardReader
   :: ServerContext
   -> G.UnaryHandler CreateShardReaderRequest CreateShardReaderResponse
 handleCreateShardReader sc _ req = catchDefaultEx $ do
-  Log.debug $ "Receive Create ShardReader Request" <> Log.buildString' (show req)
+  Log.info $ "Receive Create ShardReader Request" <> Log.buildString' (show req)
   validateCreateShardReader req
   C.createShardReader sc req
 
@@ -77,7 +77,7 @@ deleteShardReaderHandler
   -> ServerRequest 'Normal DeleteShardReaderRequest Empty
   -> IO (ServerResponse 'Normal Empty)
 deleteShardReaderHandler sc@ServerContext{..} (ServerNormalRequest _metadata request@DeleteShardReaderRequest{..}) = defaultExceptionHandle $ do
-  Log.debug $ "Receive Delete ShardReader Request" <> Log.buildString' (show request)
+  Log.info $ "Receive Delete ShardReader Request" <> Log.buildString' (show request)
   validateNameAndThrow ResShardReader deleteShardReaderRequestReaderId
   ServerNode{..} <- lookupResource sc ResShardReader deleteShardReaderRequestReaderId
   unless (serverNodeId == serverID) $
@@ -88,7 +88,7 @@ handleDeleteShardReader
   :: ServerContext
   -> G.UnaryHandler DeleteShardReaderRequest Empty
 handleDeleteShardReader sc@ServerContext{..} _ req@DeleteShardReaderRequest{..} = catchDefaultEx $ do
-  Log.debug $ "Receive Delete ShardReader Request" <> Log.buildString' (show req)
+  Log.info $ "Receive Delete ShardReader Request" <> Log.buildString' (show req)
   validateNameAndThrow ResShardReader deleteShardReaderRequestReaderId
   ServerNode{..} <- lookupResource sc ResShardReader deleteShardReaderRequestReaderId
   unless (serverNodeId == serverID) $

diff --git a/hstream/src/HStream/Server/Handler/Stats.hs b/hstream/src/HStream/Server/Handler/Stats.hs
@@ -86,7 +86,7 @@ getStatsHandler sc (ServerNormalRequest _ (API.GetStatsRequest mstats)) = defaul
 handleGetStats
   :: ServerContext
   -> G.UnaryHandler API.GetStatsRequest API.GetStatsResponse
-handleGetStats sc _ (API.GetStatsRequest mstats) = do
+handleGetStats sc _ (API.GetStatsRequest mstats) = catchDefaultEx $ do
   (failed, suc) <- getStats mstats sc
   pure $ API.GetStatsResponse {getStatsResponseStatsValues = V.fromList suc, getStatsResponseErrors = V.fromList failed}