Add a server to provide information about data sources and destinations (#48)

lazamar · web-flow · commit 731d118f5588 · 2025-04-23T11:57:08.000+01:00
diff --git a/README.md b/README.md
@@ -114,6 +114,10 @@ data_destinations:
       - file_source
 ```
 
+## Projections info
+
+The emulator starts a server that provides information about projections progress on the `/projections` endpoint.
+
 ## Running the program
 
 You only need to provide the address of the configuration file and the emulator
@@ -128,6 +132,8 @@ will start streaming your data.
 You can see all commands available with the `--help` flag
 
 ``` bash
+$ emulator --help
+
 Ambar Emulator v0.0.1 - alpha release
 
   A local version of Ambar <https://ambar.cloud>
@@ -144,3 +150,29 @@ Available commands:
 
 More info at <https://github.com/ambarltd/emulator>
 ```
+
+And these are the options for the `run` command.
+
+```
+$ emulator run --help
+
+Usage: emulator run [--partitions-per-topic INT] [--port INT]
+                    [--override-polling-interval SECONDS] [--data-path PATH]
+                    --config FILE [--verbose]
+
+  run the emulator
+
+Available options:
+  -h,--help                Show this help text
+  --partitions-per-topic INT
+                           How many partitions should newly created topics have.
+  --port INT               Port to attach projections info server to.
+  --override-polling-interval SECONDS
+                           Override the polling interval for all polled data
+                           sources.
+  --data-path PATH         Where to put emulation data including file queues.
+                           Defaults to $XDG_DATA_HOME/ambar-emulator.
+  --config FILE            Yaml file with environment configuration. Spec at at
+                           <https://github.com/ambarltd/emulator>.
+  --verbose                Enable verbose logging.
+```
diff --git a/build/Dockerfile.dynamic b/build/Dockerfile.dynamic
@@ -0,0 +1,57 @@
+# Builds a dynamically linked binary.
+# Quicker for rebuilds. Works in OSX.
+
+FROM alpine:3.19
+
+RUN apk update
+RUN apk add libffi-dev
+RUN apk add ncurses-dev
+RUN apk add libpq-dev
+RUN apk add gmp-dev
+RUN apk add zlib-dev
+RUN apk add libc-dev
+RUN apk add musl-dev
+RUN apk add mariadb-connector-c-dev
+RUN apk add pcre-dev
+
+RUN apk add git
+RUN apk add curl
+RUN apk add binutils-gold
+RUN apk add gcc
+RUN apk add g++
+RUN apk add make
+RUN apk add perl
+RUN apk add pkgconfig
+RUN apk add tar
+RUN apk add xz
+RUN apk add bash
+
+RUN curl --proto '=https' --tlsv1.3 -sSf https://get-ghcup.haskell.org | \
+    BOOTSTRAP_HASKELL_NONINTERACTIVE=1 \
+    BOOTSTRAP_HASKELL_GHC_VERSION="9.10.1" \
+    BOOTSTRAP_HASKELL_INSTALL_NO_STACK=1 \
+    sh
+
+ENV PATH="/root/.ghcup/bin:$PATH"
+ENV PATH="/root/.cabal/bin:$PATH"
+
+WORKDIR /opt/emulator
+
+# Add only the build files and build only the dependencies. Docker will cache
+# this layer, freeing us up to modify source code without re-building the
+# dependencies (unless the .cabal file changes)
+COPY ./emulator.cabal       /opt/emulator
+COPY ./cabal.project        /opt/emulator
+COPY ./deps                 /opt/emulator/deps
+RUN cabal build --dependencies-only +RTS -N
+
+COPY ./src          /opt/emulator/src
+COPY ./tests        /opt/emulator/tests
+COPY ./benchmarks   /opt/emulator/benchmarks
+RUN cabal build exe:emulator
+RUN cp $(cabal list-bin emulator) /opt/emulator/emulator
+RUN mkdir -p /opt/emulator/config
+
+ENTRYPOINT ["sh", "-c", "rm -Rf /root/.local/share/ambar-emulator/queues/*.lock /root/.local/share/ambar-emulator/queues/**/*.lock && /opt/emulator/emulator $0 $@"]
+
+CMD ["run", "--config", "/opt/emulator/config/config.yaml"]
diff --git a/build/Dockerfile.static b/build/Dockerfile.static
diff --git a/emulator.cabal b/emulator.cabal
@@ -54,6 +54,7 @@ library
         Ambar.Emulator.Queue.Partition
         Ambar.Emulator.Queue.Partition.File
         Ambar.Emulator.Queue.Partition.STMReader
+        Ambar.Emulator.Server
         Ambar.Transport
         Ambar.Transport.Http
         Ambar.Transport.File
@@ -95,6 +96,8 @@ library
         , time
         , unordered-containers
         , yaml
+        , warp
+        , wai
 
 test-suite emulator-tests
     import: common
diff --git a/src/Ambar/Emulator.hs b/src/Ambar/Emulator.hs
@@ -21,6 +21,7 @@ import Ambar.Emulator.Connector.MicrosoftSQLServer (SQLServerState)
 import Ambar.Emulator.Connector.MySQL (MySQLState)
 import Ambar.Emulator.Connector.Postgres (PostgreSQLState)
 
+import qualified Ambar.Emulator.Server as Server
 import qualified Ambar.Emulator.Projector as Projector
 import Ambar.Emulator.Projector (Projection(..))
 import qualified Ambar.Transport.File as FileTransport
@@ -37,6 +38,7 @@ import Ambar.Emulator.Config
   , DataDestination(..)
   , Destination(..)
   )
+import Util.Async (withAsyncThrow)
 import Util.Delay (every, seconds)
 import Util.Directory (writeAtomically)
 import Util.Logger (SimpleLogger, annotate, logInfo, logDebugAction)
@@ -46,12 +48,18 @@ import Util.STM (atomicallyNamed)
 emulate :: SimpleLogger -> EmulatorConfig -> EnvironmentConfig -> IO ()
 emulate logger_ config env = do
   Queue.withQueue queuePath pcount $ \queue ->
+    withServer queue $
     concurrently_ (connectAll queue) (projectAll queue)
   where
   queuePath = c_dataPath config </> "queues"
   statePath = c_dataPath config </> "state.json"
   pcount = Topic.PartitionCount $ c_partitionsPerTopic config
 
+  withServer queue act =
+    case c_port config of
+      Nothing -> act
+      Just port -> withAsyncThrow (Server.run port queue) act
+
   connectAll queue = do
     EmulatorState connectorStates <- load
     let getState source =
@@ -180,7 +188,7 @@ toConnectorConfig source sstate =
       "Incompatible state for source: " <> show (s_id source)
 
 topicName :: Id DataSource -> TopicName
-topicName sid = TopicName $ "t-" <> unId sid
+topicName sid = TopicName $ unId sid
 
 projectionId :: Id DataDestination -> Id Projection
-projectionId (Id dst) = Id ("p-" <> dst)
+projectionId (Id dst) = Id dst
diff --git a/src/Ambar/Emulator/Config.hs b/src/Ambar/Emulator/Config.hs
@@ -8,6 +8,7 @@ module Ambar.Emulator.Config
   , Source(..)
   , DataDestination(..)
   , Destination(..)
+  , Port(..)
   )
   where
 
@@ -28,6 +29,7 @@ import qualified Data.Map.Strict as Map
 import qualified Data.Yaml as Yaml
 
 import Ambar.Emulator.Connector.Poll (PollingInterval(..))
+import Ambar.Emulator.Server (Port(..))
 import Util.Delay (millis)
 import Ambar.Emulator.Connector.MicrosoftSQLServer (SQLServer(..))
 import Ambar.Emulator.Connector.Postgres (PostgreSQL(..))
@@ -42,6 +44,7 @@ newtype Id a = Id { unId :: Text }
 -- | Configures the internals of the emulator
 data EmulatorConfig = EmulatorConfig
   { c_partitionsPerTopic :: Int
+  , c_port :: Maybe Port
   , c_dataPath :: FilePath
   }
 
diff --git a/src/Ambar/Emulator/Queue.hs b/src/Ambar/Emulator/Queue.hs
@@ -31,8 +31,10 @@ import Ambar.Emulator.Queue.Topic
   , TopicState(..)
   , PartitionNumber(..)
   , PartitionCount(..)
+  , ConsumerGroupName
   )
 import qualified Ambar.Emulator.Queue.Topic as T
+import Ambar.Emulator.Queue.Partition (Count, Offset)
 import qualified Ambar.Emulator.Queue.Partition.File as FilePartition
 import Ambar.Emulator.Queue.Partition.File (FilePartition)
 import Util.Async (withAsyncThrow)
@@ -203,8 +205,18 @@ inventoryRelease (Store path) = do
   let lock = path </> "inventory.lock"
   removeFile lock
 
-getInfo :: Queue -> IO (HashMap TopicName PartitionCount)
+type QueueInfo =
+  HashMap TopicName
+    ( HashMap PartitionNumber Count
+    , HashMap ConsumerGroupName (HashMap PartitionNumber Offset)
+    )
+
+getInfo :: Queue -> IO QueueInfo
 getInfo queue = do
   topics <- readMVar (q_topics queue)
-  return $ flip fmap topics $ \tdata ->
-    PartitionCount $ HashMap.size (d_partitions tdata)
+  forM topics $ \(TopicData topic partitions) -> do
+    consumers <- T.s_consumers <$> T.getState topic
+    counts <- traverse FilePartition.getRecordCount partitions
+    return (counts, consumers)
+
+
diff --git a/src/Ambar/Emulator/Queue/Partition.hs b/src/Ambar/Emulator/Queue/Partition.hs
@@ -25,7 +25,7 @@ newtype Offset = Offset { unOffset :: Int }
 -- The total number of records in a partition.
 newtype Count = Count Int
   deriving (Show)
-  deriving newtype (Eq, Ord, Enum, Integral, Real, Num)
+  deriving newtype (Eq, Ord, Enum, Integral, Real, Num, ToJSON)
 
 newtype Record = Record { unRecord :: ByteString }
   deriving newtype (Eq, Ord, Show)
diff --git a/src/Ambar/Emulator/Queue/Partition/File.hs b/src/Ambar/Emulator/Queue/Partition/File.hs
@@ -10,6 +10,7 @@ module Ambar.Emulator.Queue.Partition.File
   , openNonLockingWritableFD
   , closeNonLockingWritableFD
   , writeFD
+  , getRecordCount
   ) where
 
 import Control.Concurrent (MVar, newMVar, withMVarMasked, modifyMVar_)
@@ -199,6 +200,11 @@ openNonLockingWritableFD path = do
 closeNonLockingWritableFD :: FD -> IO ()
 closeNonLockingWritableFD = FD.close
 
+getRecordCount :: FilePartition -> IO Count
+getRecordCount (FilePartition{..}) = do
+  (count, _) <- STM.readTVarIO p_info
+  return count
+
 -- A single-threaded file readed.
 -- If you use it from multiple threads you will have problems.
 data FileReader = FileReader PartitionName (TMVar ReaderInfo)
diff --git a/src/Ambar/Emulator/Queue/Topic.hs b/src/Ambar/Emulator/Queue/Topic.hs
@@ -489,6 +489,7 @@ commit (Consumer cid _ var) (Meta pnumber offset) =
     STM.modifyTVar (r_committed r) $ \previous -> max previous (offset + 1)
 
 newtype PartitionCount = PartitionCount Int
+  deriving newtype (ToJSON)
 
 partitionCount :: Topic -> PartitionCount
 partitionCount topic = PartitionCount $ HashMap.size (t_partitions topic)
diff --git a/src/Ambar/Emulator/Server.hs b/src/Ambar/Emulator/Server.hs
@@ -0,0 +1,45 @@
+module Ambar.Emulator.Server
+  ( run
+  , Port(..)
+  ) where
+
+import Data.Aeson (toJSON, encode)
+import qualified Data.HashMap.Strict as HashMap
+import Data.Void (Void)
+import qualified Network.Wai.Handler.Warp as Warp
+import Network.Wai (Application, Response, responseLBS, rawPathInfo)
+import Network.HTTP.Types (status200, status404, hContentType)
+import System.IO (hPutStrLn, stderr)
+
+import Ambar.Emulator.Queue (Queue, getInfo)
+
+newtype Port = Port Int
+    deriving (Show, Eq, Read)
+
+run :: Port -> Queue -> IO Void
+run (Port port) queue = do
+  hPutStrLn stderr $ "Running server on port " <> show port
+  Warp.run port $ app queue
+  error "server terminated unexpectedly"
+
+app :: Queue -> Application
+app queue request respond = do
+  response <- case rawPathInfo request of
+    "/projections" -> projections queue
+    _   -> return pageNotFound
+  respond response
+
+pageNotFound :: Response
+pageNotFound = responseLBS
+    status404
+    [("Content-Type", "text/plain")]
+    "404 - Not Found"
+
+projections :: Queue -> IO Response
+projections queue = do
+  info <- getInfo queue
+  let response = fmap (\(partitions, consumers) -> HashMap.fromList
+        [ ("partitions" :: String, toJSON partitions)
+        , ("projections", toJSON consumers)
+        ]) info
+  return $ responseLBS status200 [(hContentType, "application/json")] (encode response)
diff --git a/src/Main.hs b/src/Main.hs
@@ -16,7 +16,7 @@ import System.Posix.Signals (installHandler, sigINT, sigTERM, Handler(Catch))
 #endif
 
 import Ambar.Emulator (emulate)
-import Ambar.Emulator.Config (parseEnvConfigFile, EmulatorConfig(..))
+import Ambar.Emulator.Config (parseEnvConfigFile, EmulatorConfig(..), Port(..))
 import Ambar.Emulator.Connector.Poll (PollingInterval(..))
 import Util.Logger (plainLogger, Severity(..), logInfo)
 import Util.Delay (fromDiffTime)
@@ -25,6 +25,9 @@ import Util.Delay (fromDiffTime)
 _DEFAULT_PARTITIONS_PER_TOPIC :: Int
 _DEFAULT_PARTITIONS_PER_TOPIC = 10
 
+_DEFAULT_PORT :: Port
+_DEFAULT_PORT = Port 8080
+
 -- | Version of the binary file
 _VERSION :: String
 _VERSION = "v0.0.1 - alpha release"
@@ -44,6 +47,7 @@ main = do
       queue <- maybe defaultStatePath return o_statePath
       let config = EmulatorConfig
             { c_partitionsPerTopic = fromMaybe _DEFAULT_PARTITIONS_PER_TOPIC o_partitionsPerTopic
+            , c_port = o_port
             , c_dataPath = queue
             }
 
@@ -75,6 +79,7 @@ data Command
     , o_statePath :: Maybe FilePath
     , o_configPath :: FilePath
     , o_verbose :: Bool
+    , o_port :: Maybe Port
     , o_overridePollingInterval :: Maybe PollingInterval
     }
   | CmdVersion
@@ -107,6 +112,11 @@ cliOptions = O.info (O.simpleVersioner _VERSION <*> O.helper <*> parser) $ mconc
           , O.metavar "INT"
           , O.help "How many partitions should newly created topics have."
           ]
+      o_port <- fmap (fmap Port) $ optional $ O.option O.auto $ mconcat
+          [ O.long "port"
+          , O.metavar "INT"
+          , O.help "Port to attach projections info server to."
+          ]
       o_overridePollingInterval <-
         fmap (fmap $ PollingInterval . fromDiffTime . realToFrac @Double) $ optional $ O.option O.auto $ mconcat
           [ O.long "override-polling-interval"
diff --git a/tests/Test/Emulator.hs b/tests/Test/Emulator.hs
@@ -118,6 +118,7 @@ testEmulator p = describe "emulator" $ do
       f EmulatorConfig
         { c_partitionsPerTopic = 5
         , c_dataPath = path
+        , c_port = Nothing
         }
 
     withPostgresSource ::
diff --git a/utils.sh b/utils.sh
@@ -17,7 +17,7 @@ function bench {
 }
 
 function build-docker {
-  docker build --file "./build/Dockerfile.aarch64.linux" -t emulator .
+  docker build --file "./build/Dockerfile.static" -t emulator .
 }
 
 # Save results of benchmarking
@@ -35,7 +35,7 @@ function typecheck {
 function typecheck-executable {
   # Start fast type-checking of the executable. (Just Main.hs)
   # Watches your files and type-checks on save
-  ghcid -c 'cabal v2-repl' emulator "${@}"
+  ghcid -c 'cabal v2-repl' exe:emulator "${@}"
 }
 
 function typecheck-tests {
