Add monitoring to node service (#4607)

## Motivation

Currently, if we start a node service we get no metrics or memory
profiling.

## Proposal

Add support for metrics and memory profiling from node service. Also fix
the fact that it doesn't terminate with Ctrl+C (the faucet had the same
thing, so I fixed it there too).

## Test Plan

Start a node service locally, see that memory profile is enabled and
working, and that killing it with Ctrl+C actually kills it

## Release Plan

- Nothing to do / These changes follow the usual release cycle.

Author: ndr-ds

linera-faucet/server/src/lib.rs

@@ -702,7 +702,9 @@ where
         let batch_processor_task = batch_processor.run(cancellation_token.clone());
         let tcp_listener =
             tokio::net::TcpListener::bind(SocketAddr::from(([0, 0, 0, 0], port))).await?;
-        let server = axum::serve(tcp_listener, app).into_future();
+        let server = axum::serve(tcp_listener, app)
+            .with_graceful_shutdown(cancellation_token.cancelled_owned())
+            .into_future();
         futures::select! {
             result = Box::pin(chain_listener).fuse() => result?,
             _ = Box::pin(batch_processor_task).fuse() => {},

linera-metrics/src/monitoring_server.rs

@@ -15,8 +15,6 @@ pub fn start_metrics(
     address: impl ToSocketAddrs + Debug + Send + 'static,
     shutdown_signal: CancellationToken,
 ) {
-    info!("Starting to serve metrics on {:?}", address);
-
     #[cfg(feature = "memory-profiling")]
     let app = {
         // Try to add memory profiling endpoint
@@ -42,7 +40,13 @@ pub fn start_metrics(
     let app = Router::new().route("/metrics", get(serve_metrics));
 
     tokio::spawn(async move {
-        if let Err(e) = axum::serve(tokio::net::TcpListener::bind(address).await.unwrap(), app)
+        let listener = tokio::net::TcpListener::bind(address)
+            .await
+            .expect("Failed to bind to address");
+        let address = listener.local_addr().expect("Failed to get local address");
+
+        info!("Starting to serve metrics on {:?}", address);
+        if let Err(e) = axum::serve(listener, app)
             .with_graceful_shutdown(shutdown_signal.cancelled_owned())
             .await
         {

linera-service/src/cli/command.rs

@@ -742,6 +742,11 @@ pub enum ClientCommand {
         /// The port on which to run the server
         #[arg(long)]
         port: NonZeroU16,
+
+        /// The port to expose metrics on.
+        #[cfg(with_metrics)]
+        #[arg(long)]
+        metrics_port: NonZeroU16,
     },
 
     /// Run a GraphQL service that exposes a faucet where users can claim tokens.

linera-service/src/cli/main.rs

@@ -1307,7 +1307,12 @@ impl Runnable for Job {
                 info!("Notification stream ended.");
             }
 
-            Service { config, port } => {
+            Service {
+                config,
+                port,
+                #[cfg(with_metrics)]
+                metrics_port,
+            } => {
                 let context = ClientContext::new(
                     storage,
                     options.context_options.clone(),
@@ -1316,11 +1321,17 @@ impl Runnable for Job {
                 );
 
                 let default_chain = context.wallet().default_chain();
-                let service = NodeService::new(config, port, default_chain, context);
+                let service = NodeService::new(
+                    config,
+                    port,
+                    #[cfg(with_metrics)]
+                    metrics_port,
+                    default_chain,
+                    context,
+                );
                 let cancellation_token = CancellationToken::new();
-                let child_token = cancellation_token.child_token();
-                tokio::spawn(listen_for_shutdown_signals(cancellation_token));
-                service.run(child_token).await?;
+                tokio::spawn(listen_for_shutdown_signals(cancellation_token.clone()));
+                service.run(cancellation_token).await?;
             }
 
             Faucet {

linera-service/src/node_service.rs

@@ -36,6 +36,8 @@ use linera_execution::{
     committee::Committee, system::AdminOperation, Operation, Query, QueryOutcome, QueryResponse,
     SystemOperation,
 };
+#[cfg(with_metrics)]
+use linera_metrics::monitoring_server;
 use linera_sdk::linera_base_types::BlobContent;
 use serde::{Deserialize, Serialize};
 use serde_json::json;
@@ -816,6 +818,8 @@ where
 {
     config: ChainListenerConfig,
     port: NonZeroU16,
+    #[cfg(with_metrics)]
+    metrics_port: NonZeroU16,
     default_chain: Option<ChainId>,
     context: Arc<Mutex<C>>,
 }
@@ -828,6 +832,8 @@ where
         Self {
             config: self.config.clone(),
             port: self.port,
+            #[cfg(with_metrics)]
+            metrics_port: self.metrics_port,
             default_chain: self.default_chain,
             context: Arc::clone(&self.context),
         }
@@ -842,17 +848,25 @@ where
     pub fn new(
         config: ChainListenerConfig,
         port: NonZeroU16,
+        #[cfg(with_metrics)] metrics_port: NonZeroU16,
         default_chain: Option<ChainId>,
         context: C,
     ) -> Self {
         Self {
             config,
             port,
+            #[cfg(with_metrics)]
+            metrics_port,
             default_chain,
             context: Arc::new(Mutex::new(context)),
         }
     }
 
+    #[cfg(with_metrics)]
+    pub fn metrics_address(&self) -> SocketAddr {
+        SocketAddr::from(([0, 0, 0, 0], self.metrics_port.get()))
+    }
+
     pub fn schema(&self) -> Schema<QueryRoot<C>, MutationRoot<C>, SubscriptionRoot<C>> {
         Schema::build(
             QueryRoot {
@@ -878,6 +892,9 @@ where
         let application_handler =
             axum::routing::get(util::graphiql).post(Self::application_handler);
 
+        #[cfg(with_metrics)]
+        monitoring_server::start_metrics(self.metrics_address(), cancellation_token.clone());
+
         let app = Router::new()
             .route("/", index_handler)
             .route(
@@ -894,14 +911,20 @@ where
 
         let storage = self.context.lock().await.storage().clone();
 
-        let chain_listener =
-            ChainListener::new(self.config, self.context, storage, cancellation_token)
-                .run()
-                .await?;
+        let chain_listener = ChainListener::new(
+            self.config,
+            self.context,
+            storage,
+            cancellation_token.clone(),
+        )
+        .run()
+        .await?;
         let mut chain_listener = Box::pin(chain_listener).fuse();
         let tcp_listener =
             tokio::net::TcpListener::bind(SocketAddr::from(([0, 0, 0, 0], port))).await?;
-        let server = axum::serve(tcp_listener, app).into_future();
+        let server = axum::serve(tcp_listener, app)
+            .with_graceful_shutdown(cancellation_token.cancelled_owned())
+            .into_future();
         futures::select! {
             result = chain_listener => result?,
             result = Box::pin(server).fuse() => result?,

linera-service/src/schema_export.rs

@@ -228,6 +228,8 @@ async fn main() -> std::io::Result<()> {
     let service = NodeService::new(
         ChainListenerConfig::default(),
         std::num::NonZeroU16::new(8080).unwrap(),
+        #[cfg(with_metrics)]
+        std::num::NonZeroU16::new(8081).unwrap(),
         None,
         DummyContext,
     );