Merge remote-tracking branch 'hstream/main' into improve_kafka_configs

docker/Dockerfile

@@ -143,8 +143,6 @@ COPY --from=ld_image /logdevice/common/test/ssl_certs/ /logdevice/common/test/ss
 
 COPY --from=builder /root/.cabal/bin/hstream-server \
                     /root/.cabal/bin/hstream-server-old \
-                    /root/.cabal/bin/hstream-kafka-server \
-                    /root/.cabal/bin/hstream-kafka-server-old \
                     /root/.cabal/bin/hstream-kafka-cli \
                     /root/.cabal/bin/hstream \
                     /root/.cabal/bin/hadmin \

hstream-kafka/HStream/Kafka/Server/Config.hs

@@ -1,6 +1,10 @@
 module HStream.Kafka.Server.Config
   ( ServerOpts (..)
+  , CliOptions (..)
+  , cliOptionsParser
+
   , runServerConfig
+  , runServerFromCliOpts
 
   , MetaStoreAddr (..)
   , AdvertisedListeners
@@ -31,6 +35,9 @@ runServerConfig args f = do
     ShowVersion -> showVersion
     Cli cliOpts -> getConfig cliOpts >>= f
 
+runServerFromCliOpts :: CliOptions -> (ServerOpts -> IO ()) -> IO ()
+runServerFromCliOpts cliOpts f = getConfig cliOpts >>= f
+
 getConfig :: CliOptions -> IO ServerOpts
 getConfig opts@CliOptions{..} = do
   path <- makeAbsolute cliConfigPath

hstream-kafka/HStream/Kafka/Server/Config/FromCli.hs

@@ -6,6 +6,8 @@
 module HStream.Kafka.Server.Config.FromCli
   ( serverCliParser
   , runServerCli
+
+  , CliOptions (..)
   , cliOptionsParser
 
     -- * Helpers

hstream/app/kafka-server.hs → hstream/app/lib/KafkaServer.hs

@@ -9,6 +9,12 @@
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TypeApplications    #-}
 
+-- FIXME: How about merging this module into hstream/app/server.hs?
+module KafkaServer
+  ( app
+  , runApp
+  ) where
+
 import           Control.Concurrent                (forkIO)
 import qualified Control.Concurrent.Async          as Async
 import           Control.Concurrent.MVar           (MVar, newEmptyMVar, putMVar,
@@ -67,8 +73,8 @@ import           Text.RawString.QQ                 (r)
 
 -------------------------------------------------------------------------------
 
-main :: IO ()
-main = do
+runApp :: IO ()
+runApp = do
   args <- getArgs
   runServerConfig args app
 

hstream/app/server.hs

@@ -48,6 +48,7 @@ import           HStream.Gossip                   (GossipContext (..),
                                                    startGossip, waitGossipBoot)
 import           HStream.Gossip.Types             (Epoch, InitType (Gossip))
 import           HStream.Gossip.Utils             (getMemberListWithEpochSTM)
+import qualified HStream.Kafka.Server.Config      as Ka
 import qualified HStream.Logger                   as Log
 import           HStream.MetaStore.Types          as M (MetaHandle (..),
                                                         MetaStore (..),
@@ -78,6 +79,7 @@ import           HStream.Server.Types             (ServerContext (..))
 import qualified HStream.Store.Logger             as Log
 import qualified HStream.ThirdParty.Protobuf      as Proto
 import           HStream.Utils                    (getProtoTimestamp)
+import qualified KafkaServer                      as Ka
 
 #ifdef HStreamUseGrpcHaskell
 import           HStream.Server.Handler           (handlers)
@@ -98,8 +100,12 @@ main = do
   args <- getArgs
   serverCli <- runServerCli args
   case serverCli of
-    ShowVersion -> showVersion
-    Cli cliOpts -> getConfig cliOpts >>= app
+    ShowVersion      -> showVersion
+    Cli cliOpts      -> getConfig cliOpts >>= app
+    KafkaCli cliOpts -> Ka.runServerFromCliOpts cliOpts Ka.app
+
+-------------------------------------------------------------------------------
+-- HStream Server
 
 app :: ServerOpts -> IO ()
 app config@ServerOpts{..} = do
@@ -291,8 +297,6 @@ serveListeners sc grpcOpts
        else HsGrpc.runServer grpcOpts' (HsGrpcHandler.handlers sc')
 #endif
 
--------------------------------------------------------------------------------
-
 -- default grpc options
 #ifdef HStreamUseGrpcHaskell
 defGrpcOpts
@@ -328,6 +332,9 @@ defGrpcOpts host port tlsConfig chanArgs = do
       }
 #endif
 
+-------------------------------------------------------------------------------
+-- Misc
+
 showVersion :: IO ()
 showVersion = do
   API.HStreamVersion{..} <- getHStreamVersion

hstream/hstream.cabal

@@ -156,6 +156,7 @@ library
     , hstream-common-stats
     , hstream-gossip
     , hstream-io
+    , hstream-kafka
     , hstream-processing
     , hstream-sql
     , hstream-store
@@ -209,14 +210,11 @@ library
     TypeApplications
     UnliftedFFITypes
 
-executable hstream-server
-  if flag(hstream_enable_asan)
-    import: shared-properties, link-asan
-
-  if !flag(hstream_enable_asan)
-    import: shared-properties
-
-  main-is:          app/server.hs
+-- Local app library
+library hstream-app
+  import:           shared-properties
+  hs-source-dirs:   app/lib
+  exposed-modules:  KafkaServer
   build-depends:
     , async
     , base
@@ -227,18 +225,19 @@ executable hstream-server
     , grpc-haskell
     , grpc-haskell-core
     , hs-grpc-server
-    , hstream
     , hstream-admin-server
     , hstream-api-hs
     , hstream-common
     , hstream-common-base
     , hstream-common-server
     , hstream-common-stats
     , hstream-gossip
+    , hstream-kafka
     , hstream-store
     , http-client
     , memory
     , optparse-applicative
+    , prometheus-client
     , proto3-suite
     , proto3-wire
     , raw-strings-qq
@@ -247,6 +246,8 @@ executable hstream-server
     , text
     , timers
     , vector
+    , wai-middleware-prometheus
+    , warp
     , Z-Data
     , zoovisitor
 
@@ -257,52 +258,15 @@ executable hstream-server
     cpp-options: -DHSTREAM_ENABLE_ASAN
 
   default-language: Haskell2010
-  ghc-options:
-    -threaded -rtsopts "-with-rtsopts=-N -A64m -n4m -qg -qn1"
-
-executable hstream
-  import:           shared-properties
-  main-is:          app/client.hs
-  build-depends:
-    , aeson
-    , ansi-terminal
-    , base
-    , bytestring
-    , containers
-    , grpc-haskell
-    , grpc-haskell-core
-    , haskeline
-    , hstream
-    , hstream-admin-server
-    , hstream-api-hs
-    , hstream-common
-    , hstream-common-base
-    , hstream-common-stats
-    , hstream-sql
-    , hstream-store
-    , network
-    , optparse-applicative
-    , proto3-suite
-    , random
-    , raw-strings-qq
-    , split
-    , text
-    , unix
-    , unordered-containers
-    , vector
-    , Z-Data
-
-  default-language: Haskell2010
-  ghc-options:      -threaded -rtsopts -with-rtsopts=-N
 
-executable hstream-kafka-server
+executable hstream-server
   if flag(hstream_enable_asan)
     import: shared-properties, link-asan
 
   if !flag(hstream_enable_asan)
     import: shared-properties
 
-  main-is:          app/kafka-server.hs
+  main-is:          app/server.hs
   build-depends:
     , async
     , base
@@ -322,10 +286,10 @@ executable hstream-kafka-server
     , hstream-gossip
     , hstream-kafka
     , hstream-store
+    , hstream:{hstream, hstream-app}
     , http-client
     , memory
     , optparse-applicative
-    , prometheus-client
     , proto3-suite
     , proto3-wire
     , raw-strings-qq
@@ -334,8 +298,6 @@ executable hstream-kafka-server
     , text
     , timers
     , vector
-    , wai-middleware-prometheus
-    , warp
     , Z-Data
     , zoovisitor
 
@@ -349,6 +311,41 @@ executable hstream-kafka-server
   ghc-options:
     -threaded -rtsopts "-with-rtsopts=-N -A64m -n4m -qg -qn1"
 
+executable hstream
+  import:           shared-properties
+  main-is:          app/client.hs
+  build-depends:
+    , aeson
+    , ansi-terminal
+    , base
+    , bytestring
+    , containers
+    , grpc-haskell
+    , grpc-haskell-core
+    , haskeline
+    , hstream
+    , hstream-admin-server
+    , hstream-api-hs
+    , hstream-common
+    , hstream-common-base
+    , hstream-common-stats
+    , hstream-sql
+    , hstream-store
+    , network
+    , optparse-applicative
+    , proto3-suite
+    , random
+    , raw-strings-qq
+    , split
+    , text
+    , unix
+    , unordered-containers
+    , vector
+    , Z-Data
+
+  default-language: Haskell2010
+  ghc-options:      -threaded -rtsopts -with-rtsopts=-N
+
 executable hstream-kafka-cli
   if flag(hstream_enable_asan)
     import: shared-properties, link-asan

hstream/src/HStream/Server/Configuration/Cli.hs

@@ -46,6 +46,7 @@ import           System.Exit                    (exitSuccess)
 import qualified Z.Data.CBytes                  as CB
 import           Z.Data.CBytes                  (CBytes)
 
+import qualified HStream.Kafka.Server.Config    as Ka
 import qualified HStream.Logger                 as Log
 import qualified HStream.Server.HStreamInternal as SAI
 import           HStream.Store                  (Compression (..))
@@ -55,11 +56,15 @@ import qualified HStream.Store.Logger           as Log
 
 data ServerCli
   = Cli CliOptions
+  | KafkaCli Ka.CliOptions
   | ShowVersion
   deriving (Show)
 
 serverCliParser :: O.Parser ServerCli
-serverCliParser = (Cli <$> cliOptionsParser) <|> showVersionParser
+serverCliParser =
+      (Cli <$> cliOptionsParser)
+  <|> kafkaCliParser
+  <|> showVersionParser
 
 runServerCli :: [String] -> IO ServerCli
 runServerCli args =
@@ -81,6 +86,11 @@ runServerCli args =
      putStr msg
      exitSuccess
 
+kafkaCliParser :: O.Parser ServerCli
+kafkaCliParser = O.hsubparser
+  (O.command "kafka" (O.info (KafkaCli <$> Ka.cliOptionsParser)
+                             (O.progDesc "Run as kafka broker")))
+
 showVersionParser :: O.Parser ServerCli
 showVersionParser = O.flag' ShowVersion
   ( O.long "version" <> O.short 'v' <> O.help "Show server version." )

script/dev-tools

@@ -1003,10 +1003,10 @@ class QuickBuildDevImageCommand:
             exe_renames = {}
 
         build_targets = ["hstream", "hstream-admin"]
-        exe_targets = ["hstream-server", "hstream", "hadmin", "hstream-kafka-server"]
+        exe_targets = ["hstream-server", "hstream", "hadmin"]
         if only_hstream:
             build_targets = ["hstream"]
-            exe_targets = ["hstream-server", "hstream", "hstream-kafka-server"]
+            exe_targets = ["hstream-server", "hstream"]
 
         tmp_container_name = f"tmp_build_dev_image_{gen_randoms(20)}"
         try: