[ADP-3305] Enable http service in local cluster executable (#4579)

- [x] Add an http service to the local-cluster to check it's readiness
and to send assets

ADP-3305

.buildkite/pipeline.yml

@@ -55,6 +55,19 @@ steps:
     env:
       TMPDIR: "/cache"
 
+  - label: Run local-cluster tests
+    key: local-cluster-tests
+    depends_on: linux-nix
+    command: |
+      mkdir local-cluster-logs
+      nix shell "nixpkgs#just" -c just test-local-cluster
+    agents:
+      system: ${linux}
+    artifact_paths: [ "./local-cluster-logs/**" ]
+    env:
+      TMPDIR: "/cache"
+      CLUSTER_LOGS_DIR_PATH: local-cluster-logs
+
   - label: "Babbage integration tests (linux)"
     key: linux-tests-integration-babbage
     depends_on: linux-nix

flake.nix

@@ -239,6 +239,7 @@
                 inherit (project.hsPkgs.cardano-wallet.components.exes) mock-token-metadata-server;
                 inherit (project.hsPkgs.local-cluster.components.exes) local-cluster;
                 inherit (project.hsPkgs.cardano-wallet-integration.components.exes) integration-exe;
+                inherit (project.hsPkgs.local-cluster.components.exes) test-local-cluster-exe;
 
                 # Adrestia tool belt
                 inherit (project.hsPkgs.bech32.components.exes) bech32;

justfile

@@ -46,6 +46,9 @@ unit-tests-cabal-match match:
     -O0 -v0 \
     --test-options '--match="{{match}}"'
 
+unit-tests-local-cluster-match match:
+    nix shell '.#local-cluster' 'nixpkgs#just' \
+    -c just unit-tests-cabal-match {{match}}
 # run unit tests
 unit-tests-cabal:
     just unit-tests-cabal-match ""
@@ -138,3 +141,13 @@ conway-integration-tests:
 latency-bench:
    cabal run -O2 -v0 cardano-wallet-benchmarks:latency -- \
    --cluster-configs lib/local-cluster/test/data/cluster-configs
+
+test-local-cluster:
+    LOCAL_CLUSTER_CONFIGS=lib/local-cluster/test/data/cluster-configs \
+    nix shell \
+        '.#local-cluster' \
+        '.#test-local-cluster-exe' \
+        '.#cardano-cli' \
+        '.#cardano-node' \
+        '.#cardano-wallet' \
+        -c test-local-cluster-exe

lib/iohk-monitoring-extra/iohk-monitoring-extra.cabal

@@ -11,8 +11,8 @@ build-type:    Simple
 library
   default-language:   Haskell2010
   default-extensions:
-    NoImplicitPrelude
     DerivingStrategies
+    NoImplicitPrelude
     OverloadedStrings
 
   ghc-options:
@@ -25,7 +25,10 @@ library
     -freverse-errors
 
   hs-source-dirs:     src
-  exposed-modules:    Cardano.BM.Extra
+  exposed-modules:
+    Cardano.BM.Extra
+    Cardano.BM.ToTextTracer
+
   build-depends:
     , aeson
     , base
@@ -34,9 +37,10 @@ library
     , deepseq
     , exceptions
     , fmt
-    , iohk-monitoring      ^>=0.1.11.3
+    , iohk-monitoring
+    , stm
     , text
-    , text-class           ^>=2024.5.5
+    , text-class
     , time
     , tracer-transformers
     , transformers

lib/iohk-monitoring-extra/src/Cardano/BM/ToTextTracer.hs

@@ -0,0 +1,97 @@
+{-# LANGUAGE RankNTypes #-}
+
+module Cardano.BM.ToTextTracer
+    ( ToTextTracer (..)
+    , newToTextTracer
+    )
+where
+
+import Prelude
+
+import Cardano.BM.Data.Tracer
+    ( HasSeverityAnnotation (..)
+    , Tracer (Tracer)
+    )
+import Cardano.BM.Tracing
+    ( Severity
+    )
+import Control.Monad
+    ( forever
+    , unless
+    , (>=>)
+    )
+import Control.Monad.STM
+    ( retry
+    )
+import Control.Monad.Trans.Cont
+    ( ContT (..)
+    )
+import Data.Text.Class
+    ( ToText (..)
+    )
+import Data.Time
+    ( getCurrentTime
+    )
+import Data.Time.Format.ISO8601
+    ( iso8601Show
+    )
+import UnliftIO
+    ( BufferMode (NoBuffering)
+    , IOMode (WriteMode)
+    , MonadIO (liftIO)
+    , async
+    , atomically
+    , hSetBuffering
+    , isEmptyTChan
+    , link
+    , newTChanIO
+    , readTChan
+    , stdout
+    , withFile
+    , writeTChan
+    )
+
+import qualified Data.Text as T
+import qualified Data.Text.IO as T
+
+-- | A thread-safe tracer that logs messages to a file or stdout for anything
+-- that has ToText instance
+newtype ToTextTracer
+    = ToTextTracer
+        (forall a. (HasSeverityAnnotation a, ToText a) => Tracer IO a)
+
+-- | Create a new `ToTextTracer`
+newToTextTracer
+    :: Maybe FilePath
+    -- ^ If provided, logs will be written to this file, otherwise to stdout
+    -> Maybe Severity
+    -- ^ Minimum severity level to log
+    -> (ToTextTracer -> IO r)
+    -- ^ Action to run with the new tracer
+    -> IO r
+newToTextTracer clusterLogs minSeverity = runContT $ do
+    ch <- newTChanIO
+    h <- case clusterLogs of
+        Nothing -> pure stdout
+        Just logFile -> do
+            ContT $ withFile logFile WriteMode
+    liftIO $ hSetBuffering h NoBuffering
+    liftIO $ async >=> link $ forever $ do
+        (x, s, t) <- atomically $ readTChan ch
+        T.hPutStrLn h
+            $ T.pack (iso8601Show t)
+                <> " ["
+                <> T.pack (show s)
+                <> "] "
+                <> x
+    ContT $ \k -> do
+        r <- k $ ToTextTracer $ Tracer $ \msg -> do
+            let severity = getSeverityAnnotation msg
+            unless (Just severity < minSeverity) $ do
+                t <- getCurrentTime
+                atomically $ writeTChan ch (toText msg, severity, t)
+        -- wait until the channel is empty
+        atomically $ do
+            empty <- isEmptyTChan ch
+            unless empty retry
+        pure r

lib/launcher/src/Cardano/Launcher.hs

@@ -21,6 +21,8 @@ module Cardano.Launcher
     , ProcessHasExited(..)
     , withBackendProcess
     , withBackendCreateProcess
+    , IfToSendSigINT (..)
+    , TimeoutInSecs (..)
 
     -- * Logging
     , LauncherLog(..)
@@ -36,7 +38,8 @@ import Cardano.BM.Data.Tracer
     , HasSeverityAnnotation (..)
     )
 import Cardano.Startup
-    ( killProcess
+    ( interruptProcess
+    , killProcess
     )
 import Control.Monad
     ( join
@@ -65,6 +68,9 @@ import Data.Text
 import Data.Text.Class
     ( ToText (..)
     )
+import Data.Text.Lazy.Builder
+    ( toLazyText
+    )
 import Fmt
     ( Buildable (..)
     , Builder
@@ -124,6 +130,7 @@ import UnliftIO.Process
     )
 
 import qualified Data.Text as T
+import qualified Data.Text.Lazy as TL
 
 -- | Represent a command to execute. Args are provided as a list where options
 -- are expected to be prefixed with `--` or `-`. For example:
@@ -149,7 +156,7 @@ data Command = Command
     } deriving (Generic)
 
 instance Show Command where
-    show = show . build
+    show = TL.unpack . toLazyText . build
 
 instance Eq Command where
     a == b = build a == build b
@@ -187,6 +194,14 @@ data ProcessHasExited
 
 instance Exception ProcessHasExited
 
+-- | Whether to send a SIGINT to the process before cleanup
+data IfToSendSigINT = SendSigINT | DoNotSendSigINT
+
+-- | Timeout in seconds to wait before killing the process, after termination
+-- Do not use NoTimeout if you are not sure the process will terminate with
+-- SIGTERM / SIGINT
+data TimeoutInSecs = NoTimeout | TimeoutInSecs Int
+
 data ProcessHandles = ProcessHandles
     { inputHandle :: Maybe Handle
     , outputHandle :: Maybe Handle
@@ -205,13 +220,23 @@ withBackendProcess
     -- ^ Logging
     -> Command
     -- ^ 'Command' description
+    -> TimeoutInSecs
+    -- ^ Seconds to wait before killing the process, after termination
+    -> IfToSendSigINT
+    -- ^ Whether to send a sigINT to the process before clenup
     -> (ProcessHandles -> m a)
     -- ^ Action to execute while process is running.
     -> m a
-withBackendProcess tr (Command name args before std_in std_out) action =
-    liftIO before >> withBackendCreateProcess tr process action
-  where
-    process = (proc name args) { std_in, std_out, std_err = std_out }
+withBackendProcess
+    tr
+    (Command name args before std_in std_out)
+    mTimeoutSecs
+    ifToSendSigINT
+    action = do
+        liftIO before
+        withBackendCreateProcess tr process mTimeoutSecs ifToSendSigINT action
+      where
+        process = (proc name args){std_in, std_out, std_err = std_out}
 
 -- | A variant of 'withBackendProcess' which accepts a general 'CreateProcess'
 -- object. This version also has nicer async properties than
@@ -233,27 +258,35 @@ withBackendProcess tr (Command name args before std_in std_out) action =
 -- 'System.Process.Typed.withProcessWait' (except for wait timeout). The
 -- launcher code should be converted to use @typed-process@.
 withBackendCreateProcess
-    :: forall m a. (MonadUnliftIO m)
+    :: forall m a
+     . MonadUnliftIO m
     => Tracer m LauncherLog
     -- ^ Logging
     -> CreateProcess
     -- ^ 'Command' description
+    -> TimeoutInSecs
+    -- ^ Seconds to wait before killing the process, after termination
+    -> IfToSendSigINT
+    -- ^ Whether to send a sigINT to the process before clenup
     -> (ProcessHandles -> m a)
     -- ^ Action to execute while process is running.
     -> m a
-withBackendCreateProcess tr process action = do
+withBackendCreateProcess tr process mTimeoutSecs ifToSendSigINT action = do
     traceWith tr $ MsgLauncherStart name args
     exitVar <- newEmptyMVar
-    res <- fmap join $ tryJust spawnPredicate $ bracket
-        (createProcess process)
-        (cleanupProcessAndWait (readMVar exitVar)) $
-            \(mstdin, mstdout, mstderr, ph) -> do
-                pid <- maybe "-" (T.pack . show) <$> liftIO (getPid ph)
-                let tr' = contramap (WithProcessInfo name pid) tr
-                let tr'' = contramap MsgLauncherWait tr'
-                traceWith tr' MsgLauncherStarted
-                interruptibleWaitForProcess tr'' ph (putMVar exitVar)
-                race (ProcessHasExited name <$> readMVar exitVar) $ bracket_
+    res <- fmap join
+        $ tryJust spawnPredicate
+        $ bracket
+            (createProcess process)
+            (cleanupProcessAndWait (readMVar exitVar))
+        $ \(mstdin, mstdout, mstderr, ph) -> do
+            pid <- maybe "-" (T.pack . show) <$> liftIO (getPid ph)
+            let tr' = contramap (WithProcessInfo name pid) tr
+            let tr'' = contramap MsgLauncherWait tr'
+            traceWith tr' MsgLauncherStarted
+            interruptibleWaitForProcess tr'' ph (putMVar exitVar)
+            race (ProcessHasExited name <$> readMVar exitVar)
+                $ bracket_
                     (traceWith tr' MsgLauncherAction)
                     (traceWith tr' MsgLauncherActionDone)
                     (action $ ProcessHandles mstdin mstdout mstderr ph)
@@ -275,16 +308,22 @@ withBackendCreateProcess tr process action = do
      -- doesn't exit after timeout, kill it, to avoid blocking indefinitely.
     cleanupProcessAndWait getExitStatus ps@(_, _, _, ph) = do
         traceWith tr MsgLauncherCleanup
+        -- we also send a sigINT to the process to make sure it terminates
+        case ifToSendSigINT of
+            SendSigINT -> liftIO $ getPid ph >>= mapM_ interruptProcess
+            DoNotSendSigINT -> pure ()
         liftIO $ cleanupProcess ps
-        let timeoutSecs = 5
         -- Async exceptions are currently masked because this is running in a
         -- bracket cleanup handler. We fork a thread and unmask so that the
         -- timeout can be cancelled.
-        tid <- forkIOWithUnmask $ \unmask -> unmask $ do
-            threadDelay (timeoutSecs * 1000 * 1000)
-            traceWith tr (MsgLauncherCleanupTimedOut timeoutSecs)
-            liftIO (getPid ph >>= mapM_ killProcess)
-        void getExitStatus `finally` killThread tid
+        void $ case mTimeoutSecs of
+            NoTimeout -> getExitStatus
+            TimeoutInSecs timeoutSecs -> do
+                tid <- forkIOWithUnmask $ \unmask -> unmask $ do
+                    threadDelay (timeoutSecs * 1000 * 1000)
+                    traceWith tr (MsgLauncherCleanupTimedOut timeoutSecs)
+                    liftIO (getPid ph >>= mapM_ killProcess)
+                getExitStatus `finally` killThread tid
         traceWith tr MsgLauncherCleanupFinished
 
     -- Wraps 'waitForProcess' in another thread. This works around the unwanted