Add misc. fixes for Kubernetes support

.dockerignore

@@ -1,7 +1,10 @@
 **/*.md
 **/*.txt
 **/.*
+**/build
 **/Dockerfile
+**/node_modules
+**/qwdata
 **/target
 docs
 examples

.github/actions/build-docker/action.yml

@@ -2,10 +2,10 @@ name: 'Build Quickwit docker image'
 description: 'Build & push docker image'
 inputs:
   username:
-    description: 'Dockerhub username'
+    description: 'Docker Hub username'
     required: true
   token:
-    description: 'Dockerhub access token'
+    description: 'Docker Hub access token'
     required: true
   version:
     description: 'Image version suffix'

Dockerfile

@@ -9,7 +9,7 @@ LABEL maintainer="Quickwit, Inc. <hello@quickwit.io>"
 LABEL org.opencontainers.image.vendor="Quickwit, Inc."
 LABEL org.opencontainers.image.licenses="AGPL-3.0"
 
-RUN echo "Add nodejs ppa" \
+RUN echo "Adding Node.js PPA" \
     && curl -s https://deb.nodesource.com/setup_16.x | bash
 
 RUN apt-get -y update \
@@ -26,14 +26,21 @@ RUN apt-get -y update \
 # Required by tonic
 RUN rustup component add rustfmt
 
+# Build UI
+COPY quickwit-ui /quickwit/quickwit-ui
+
+WORKDIR /quickwit/quickwit-ui
+
+RUN echo "Building Quickwit UI" \
+    && npm install --location=global yarn \
+    && yarn install \
+    && yarn build
+
+# Build workspace
 COPY . /quickwit
 
 WORKDIR /quickwit
 
-RUN echo "Building quickwit ui" \
-    && npm install --global yarn \
-    && make build-ui
-
 RUN echo "Building workspace with feature(s) '$CARGO_FEATURES' and profile '$CARGO_PROFILE'" \
     && cargo build \
         --features $CARGO_FEATURES \
@@ -42,8 +49,8 @@ RUN echo "Building workspace with feature(s) '$CARGO_FEATURES' and profile '$CAR
     && mkdir -p /quickwit/bin \
     && find target/$CARGO_PROFILE -maxdepth 1 -perm /a+x -type f -exec mv {} /quickwit/bin \;
 
-# Change the default configuration file in order to make the rest,
-# grpc servers and gossip accessible from outside of docker.
+# Change the default configuration file in order to make the REST,
+# gRPC, and gossip services accessible outside of Docker container.
 COPY ./config/quickwit.yaml ./config/quickwit.yaml
 RUN sed -i 's/#[ ]*listen_address: 127.0.0.1/listen_address: 0.0.0.0/g' ./config/quickwit.yaml
 

config/quickwit.yaml

@@ -27,7 +27,7 @@ version: 0
 # listen_address: 127.0.0.1
 # rest_listen_port: 7280
 #
-# In order to connect to a cluster, one needs to specify a list of
+# In order to join a cluster, one needs to specify a list of
 # seeds to connect to. If no port is specified, Quickwit will assume
 # the seeds are using the same port as the current node gossip port.
 # By default, the peer seed list is empty.

distribution/docker/alpine/Dockerfile

@@ -1,15 +1,17 @@
-FROM alpine:3.15 AS builder
+FROM alpine:3 AS builder
 
 COPY quickwit-*-unknown-linux-musl.tar.gz ./
 RUN tar -xzf quickwit-*-$(cat /etc/apk/arch)-unknown-linux-musl.tar.gz
 RUN mv ./quickwit-*/* ./
 RUN chmod 744 ./quickwit
 
-# Change the default configuration file in order to make the rest,
-# grpc servers and gossip accessible from outside of docker.
+# Change the default configuration file in order to make the REST,
+# gRPC and gossip services accessible outside of Docker.
 RUN sed -i 's/#listen_address: 127.0.0.1/listen_address: 0.0.0.0/g' ./config/quickwit.yaml
 
-FROM alpine:3.15
+
+FROM alpine:3
+
 LABEL org.opencontainers.image.title="Quickwit"
 LABEL maintainer="Quickwit, Inc. <hello@quickwit.io>"
 LABEL org.opencontainers.image.vendor="Quickwit, Inc."

quickwit-cli/src/index.rs

@@ -63,7 +63,7 @@ pub fn build_index_command<'a>() -> Command<'a> {
                 .about("List indexes.")
                 .alias("ls")
                 .args(&[
-                    arg!(--"metastore-uri" <METASTORE_URI> "Metastore URI. Override the `metastore_uri` parameter defined in the config file. Default to file-backed, but could be Amazon S3 or PostgreSQL.")
+                    arg!(--"metastore-uri" <METASTORE_URI> "Metastore URI. Override the `metastore_uri` parameter defined in the config file. Defaults to file-backed, but could be Amazon S3 or PostgreSQL.")
                         .required(false)
                 ])
             )
@@ -982,13 +982,11 @@ pub async fn delete_index_cli(args: DeleteIndexArgs) -> anyhow::Result<()> {
         }
         return Ok(());
     }
-
     if let Err(error) =
         remove_indexing_directory(&quickwit_config.data_dir_path, args.index_id.clone()).await
     {
         warn!(error= ?error, "Failed to remove indexing directory.");
     }
-
     println!("Index `{}` successfully deleted.", args.index_id);
     Ok(())
 }

quickwit-cli/src/service.rs

@@ -36,7 +36,19 @@ pub fn build_run_command<'a>() -> Command<'a> {
         .about("Runs quickwit services. By default, `indexer` and `searcher` are started.")
         .args(&[
             arg!(--"data-dir" <DATA_DIR> "Where data is persisted. Override data-dir defined in config file, default is `./qwdata`.").env("QW_DATA_DIR").required(false),
-            arg!(--"service" <SERVICE> "Services (searcher|indexer) to run. If unspecified run both `searcher` and `indexer`.").required(false)
+            arg!(--"service" <SERVICE> "Services (searcher|indexer) to run. If unspecified run both `searcher` and `indexer`.").required(false),
+            arg!(--"metastore-uri" <METASTORE_URI> "Metastore URI. Override the `metastore_uri` parameter defined in the config file. Defaults to file-backed, but could be Amazon S3 or PostgreSQL.")
+                .env("QW_METASTORE_URI")
+                .required(false),
+            arg!(--"cluster-id" <CLUSTER_ID> "ID of the cluster to connect to.")
+                .env("QW_CLUSTER_ID")
+                .required(false),
+            arg!(--"node-id" <NODE_ID> "Node ID identifying uniquely the node in the cluster.")
+                .env("QW_NODE_ID")
+                .required(false),
+            arg!(--"peer-seeds" <PEER_SEEDS> "Comma-separated list of peer seeds to connect to in order to join a cluster.")
+                .env("QW_PEER_SEEDS")
+                .required(false),
         ])
 }
 
@@ -45,6 +57,10 @@ pub struct RunCliCommand {
     pub config_uri: Uri,
     pub data_dir_path: Option<PathBuf>,
     pub services: HashSet<QuickwitService>,
+    pub metastore_uri: Option<Uri>,
+    pub cluster_id: Option<String>,
+    pub node_id: Option<String>,
+    pub peer_seeds: Option<Vec<String>>,
 }
 
 impl RunCliCommand {
@@ -63,10 +79,23 @@ impl RunCliCommand {
                     .into_iter()
                     .collect()
             };
+        let metastore_uri = matches
+            .value_of("metastore-uri")
+            .map(Uri::try_new)
+            .transpose()?;
+        let cluster_id = matches.value_of("cluster-id").map(String::from);
+        let node_id = matches.value_of("node-id").map(String::from);
+        let peer_seeds = matches
+            .value_of("peer-seeds")
+            .map(|peer_seeds_str| peer_seeds_str.split(',').map(String::from).collect());
         Ok(RunCliCommand {
             config_uri,
             data_dir_path,
             services,
+            metastore_uri,
+            cluster_id,
+            node_id,
+            peer_seeds,
         })
     }
 
@@ -80,7 +109,28 @@ impl RunCliCommand {
         let telemetry_event = TelemetryEvent::RunService(service_str);
         quickwit_telemetry::send_telemetry_event(telemetry_event).await;
 
-        let config = load_quickwit_config(&self.config_uri, self.data_dir_path.clone()).await?;
+        let mut config = load_quickwit_config(&self.config_uri, self.data_dir_path.clone()).await?;
+
+        // TODO: Remove these overrides when #1011 lands.
+        if let Some(metastore_uri) = &self.metastore_uri {
+            tracing::info!(metastore_uri = %metastore_uri, "Setting metastore URI from override.");
+            config.metastore_uri = Some(metastore_uri.clone());
+        }
+        if let Some(cluster_id) = &self.cluster_id {
+            tracing::info!(cluster_id = %cluster_id, "Setting cluster ID from override.");
+            config.cluster_id = cluster_id.clone();
+        }
+        if let Some(node_id) = &self.node_id {
+            tracing::info!(node_id = %node_id, "Setting node ID from override.");
+            config.node_id = node_id.clone();
+        }
+        if let Some(peer_seeds) = &self.peer_seeds {
+            tracing::info!(peer_seeds = %peer_seeds.join(", "), "Setting peer seeds from override.");
+            config.peer_seeds = peer_seeds.clone();
+        }
+        // Revalidate config because of overrides.
+        config.validate()?;
+
         serve_quickwit(&config, &self.services).await?;
         Ok(())
     }
@@ -103,7 +153,8 @@ mod tests {
             CliCommand::Run(RunCliCommand {
                 config_uri,
                 data_dir_path: None,
-                services
+                services,
+                ..
             })
             if config_uri == expected_config_uri && services.len() == 2
         ));
@@ -127,7 +178,8 @@ mod tests {
             CliCommand::Run(RunCliCommand {
                 config_uri,
                 data_dir_path: None,
-                services
+                services,
+                ..
             })
             if config_uri == expected_config_uri && services.len() == 1 && services.iter().cloned().next().unwrap() == QuickwitService::Indexer
         ));
@@ -150,7 +202,8 @@ mod tests {
             CliCommand::Run(RunCliCommand {
                 config_uri,
                 data_dir_path: None,
-                services
+                services,
+                ..
             })
             if config_uri == expected_config_uri && services.len() == 1 && services.contains(&QuickwitService::Indexer)
         ));

quickwit-cluster/src/cluster.rs

@@ -29,7 +29,6 @@ use chitchat::{
     NodeId,
 };
 use itertools::Itertools;
-use quickwit_common::net::get_socket_addr;
 use serde::{Deserialize, Serialize};
 use tokio::sync::watch;
 use tokio::time::timeout;
@@ -109,7 +108,7 @@ pub struct Cluster {
     /// Node ID.
     pub node_id: String,
     /// A socket address that represents itself.
-    pub listen_addr: SocketAddr,
+    pub gossip_listen_addr: SocketAddr,
 
     /// A handle to command Chitchat.
     /// If it is dropped, the chitchat server will stop.
@@ -133,32 +132,34 @@ impl Cluster {
     pub async fn join(
         me: Member,
         services: &HashSet<QuickwitService>,
-        listen_addr: SocketAddr,
+        gossip_listen_addr: SocketAddr,
         cluster_id: String,
-        grpc_addr: SocketAddr,
-        seed_nodes: Vec<String>,
+        grpc_public_addr: SocketAddr,
+        peer_seed_addrs: Vec<String>,
         failure_detector_config: FailureDetectorConfig,
         transport: &dyn Transport,
     ) -> ClusterResult<Self> {
         info!(
-            cluster_id=%cluster_id,
-            node_id=%me.node_unique_id,
-            listen_address=%listen_addr,
-            gossip_address=%me.gossip_public_address,
+            cluster_id = %cluster_id,
+            node_id = %me.node_unique_id,
+            grpc_public_addr = %grpc_public_addr,
+            gossip_listen_addr = %gossip_listen_addr,
+            gossip_public_addr = %me.gossip_public_address,
+            peer_seed_addrs = %peer_seed_addrs.join(", "),
             "Joining cluster."
         );
         let chitchat_config = ChitchatConfig {
             node_id: NodeId::from(me.clone()),
             cluster_id: cluster_id.clone(),
             gossip_interval: GOSSIP_INTERVAL,
-            listen_addr,
-            seed_nodes,
+            listen_addr: gossip_listen_addr,
+            seed_nodes: peer_seed_addrs,
             failure_detector_config,
         };
         let chitchat_handle = spawn_chitchat(
             chitchat_config,
             vec![
-                (GRPC_ADDRESS_KEY.to_string(), grpc_addr.to_string()),
+                (GRPC_ADDRESS_KEY.to_string(), grpc_public_addr.to_string()),
                 (
                     AVAILABLE_SERVICES_KEY.to_string(),
                     services.iter().map(|service| service.as_str()).join(","),
@@ -168,7 +169,7 @@ impl Cluster {
         )
         .await
         .map_err(|cause| ClusterError::UDPPortBindingError {
-            listen_addr,
+            listen_addr: gossip_listen_addr,
             cause: cause.to_string(),
         })?;
         let chitchat = chitchat_handle.chitchat();
@@ -179,7 +180,7 @@ impl Cluster {
         let cluster = Cluster {
             cluster_id,
             node_id: me.internal_id(),
-            listen_addr,
+            gossip_listen_addr,
             chitchat_handle,
             members: members_receiver,
             stop: Arc::new(AtomicBool::new(false)),
@@ -204,13 +205,13 @@ impl Cluster {
                 members.push(me.clone());
 
                 if task_stop.load(Ordering::Relaxed) {
-                    debug!("receive a stop signal");
+                    debug!("Received a stop signal. Stopping.");
                     break;
                 }
 
                 if members_sender.send(members).is_err() {
                     // Somehow the cluster has been dropped.
-                    error!("Failed to send a member list.");
+                    error!("Failed to send members list. Stopping.");
                     break;
                 }
             }
@@ -268,7 +269,7 @@ impl Cluster {
                         member.internal_id()
                     )
                 })
-                .map(|addr_str| get_socket_addr(&addr_str))??;
+                .map(|addr_str| addr_str.parse::<SocketAddr>())??;
             grpc_addresses.push(grpc_address);
         }
 
@@ -284,7 +285,7 @@ impl Cluster {
 
     /// Leave the cluster.
     pub async fn leave(&self) {
-        info!(self_addr = ?self.listen_addr, "Leaving the cluster.");
+        info!(self_addr = ?self.gossip_listen_addr, "Leaving the cluster.");
         // TODO: implements leave/join on Chitchat
         // https://github.com/quickwit-oss/chitchat/issues/30
         self.stop.store(true, Ordering::Relaxed);
@@ -306,10 +307,10 @@ impl Cluster {
 
     /// Leave the cluster.
     pub async fn shutdown(self) {
-        info!(self_addr = ?self.listen_addr, "Shutting down the cluster.");
+        info!(self_addr = ?self.gossip_listen_addr, "Shutting down the cluster.");
         let result = self.chitchat_handle.shutdown().await;
         if let Err(error) = result {
-            error!(self_addr = ?self.listen_addr, error = ?error, "Error while shuting down.");
+            error!(self_addr = ?self.gossip_listen_addr, error = ?error, "Error while shuting down.");
         }
 
         self.stop.store(true, Ordering::Relaxed);
@@ -443,7 +444,7 @@ mod tests {
             .iter()
             .map(|member| member.gossip_public_address)
             .collect();
-        let expected_members = vec![cluster.listen_addr];
+        let expected_members = vec![cluster.gossip_listen_addr];
         assert_eq!(members, expected_members);
         cluster.shutdown().await;
         Ok(())
@@ -454,14 +455,14 @@ mod tests {
         quickwit_common::setup_logging_for_tests();
         let transport = ChannelTransport::default();
         let cluster1 = create_cluster_for_test(Vec::new(), &["searcher"], &transport).await?;
-        let node_1 = cluster1.listen_addr.to_string();
+        let node_1 = cluster1.gossip_listen_addr.to_string();
         let cluster2 =
             create_cluster_for_test(vec![node_1.clone()], &["searcher"], &transport).await?;
         let cluster3 = create_cluster_for_test(vec![node_1], &["indexer"], &transport).await?;
 
         let mut expected_searchers = vec![
-            grpc_addr_from_listen_addr_for_test(cluster1.listen_addr),
-            grpc_addr_from_listen_addr_for_test(cluster2.listen_addr),
+            grpc_addr_from_listen_addr_for_test(cluster1.gossip_listen_addr),
+            grpc_addr_from_listen_addr_for_test(cluster2.gossip_listen_addr),
         ];
         expected_searchers.sort();
 
@@ -487,16 +488,16 @@ mod tests {
         let transport = ChannelTransport::default();
         let cluster1 =
             create_cluster_for_test(Vec::new(), &["searcher", "indexer"], &transport).await?;
-        let node_1 = cluster1.listen_addr.to_string();
+        let node_1 = cluster1.gossip_listen_addr.to_string();
         let cluster2 =
             create_cluster_for_test(vec![node_1.clone()], &["indexer"], &transport).await?;
         let cluster3 =
             create_cluster_for_test(vec![node_1], &["indexer", "searcher"], &transport).await?;
 
         let mut expected_indexers = vec![
-            grpc_addr_from_listen_addr_for_test(cluster1.listen_addr),
-            grpc_addr_from_listen_addr_for_test(cluster2.listen_addr),
-            grpc_addr_from_listen_addr_for_test(cluster3.listen_addr),
+            grpc_addr_from_listen_addr_for_test(cluster1.gossip_listen_addr),
+            grpc_addr_from_listen_addr_for_test(cluster2.gossip_listen_addr),
+            grpc_addr_from_listen_addr_for_test(cluster3.gossip_listen_addr),
         ];
         expected_indexers.sort();
 
@@ -523,7 +524,7 @@ mod tests {
         quickwit_common::setup_logging_for_tests();
         let transport = ChannelTransport::default();
         let cluster1 = create_cluster_for_test(Vec::new(), &[], &transport).await?;
-        let node_1 = cluster1.listen_addr.to_string();
+        let node_1 = cluster1.gossip_listen_addr.to_string();
         let cluster2 = create_cluster_for_test(vec![node_1.clone()], &[], &transport).await?;
         let cluster3 = create_cluster_for_test(vec![node_1], &[], &transport).await?;
 
@@ -542,9 +543,9 @@ mod tests {
             .sorted()
             .collect();
         let mut expected_members = vec![
-            cluster1.listen_addr,
-            cluster2.listen_addr,
-            cluster3.listen_addr,
+            cluster1.gossip_listen_addr,
+            cluster2.gossip_listen_addr,
+            cluster3.gossip_listen_addr,
         ];
         expected_members.sort();
         assert_eq!(members, expected_members);
@@ -579,7 +580,7 @@ mod tests {
         let cluster2a = create_cluster_for_test_with_id(
             21u16,
             "cluster2".to_string(),
-            vec![cluster1a.listen_addr.to_string()],
+            vec![cluster1a.gossip_listen_addr.to_string()],
             &HashSet::default(),
             &transport,
         )
@@ -588,8 +589,8 @@ mod tests {
             12,
             "cluster1".to_string(),
             vec![
-                cluster1a.listen_addr.to_string(),
-                cluster2a.listen_addr.to_string(),
+                cluster1a.gossip_listen_addr.to_string(),
+                cluster2a.gossip_listen_addr.to_string(),
             ],
             &HashSet::default(),
             &transport,
@@ -599,8 +600,8 @@ mod tests {
             22,
             "cluster2".to_string(),
             vec![
-                cluster1a.listen_addr.to_string(),
-                cluster2a.listen_addr.to_string(),
+                cluster1a.gossip_listen_addr.to_string(),
+                cluster2a.gossip_listen_addr.to_string(),
             ],
             &HashSet::default(),
             &transport,
@@ -622,7 +623,8 @@ mod tests {
             .map(|member| member.gossip_public_address)
             .sorted()
             .collect();
-        let mut expected_members_a = vec![cluster1a.listen_addr, cluster1b.listen_addr];
+        let mut expected_members_a =
+            vec![cluster1a.gossip_listen_addr, cluster1b.gossip_listen_addr];
         expected_members_a.sort();
         assert_eq!(members_a, expected_members_a);
 
@@ -632,7 +634,8 @@ mod tests {
             .map(|member| member.gossip_public_address)
             .sorted()
             .collect();
-        let mut expected_members_b = vec![cluster2a.listen_addr, cluster2b.listen_addr];
+        let mut expected_members_b =
+            vec![cluster2a.gossip_listen_addr, cluster2b.gossip_listen_addr];
         expected_members_b.sort();
         assert_eq!(members_b, expected_members_b);
 
@@ -652,7 +655,7 @@ mod tests {
             &transport,
         )
         .await?;
-        let node_1 = cluster1.listen_addr.to_string();
+        let node_1 = cluster1.gossip_listen_addr.to_string();
         let cluster2 = create_cluster_for_test_with_id(
             2u16,
             cluster_id.to_string(),
@@ -676,11 +679,11 @@ mod tests {
             .map(|member| member.gossip_public_address)
             .sorted()
             .collect();
-        let mut expected_members = vec![cluster1.listen_addr, cluster2.listen_addr];
+        let mut expected_members = vec![cluster1.gossip_listen_addr, cluster2.gossip_listen_addr];
         expected_members.sort();
         assert_eq!(members, expected_members);
 
-        let cluster2_listen_addr = cluster2.listen_addr;
+        let cluster2_listen_addr = cluster2.gossip_listen_addr;
         cluster2.shutdown().await;
         cluster1
             .wait_for_members(|members| members.len() == 1, wait_secs)
@@ -733,7 +736,7 @@ mod tests {
             &transport,
         )
         .await?;
-        let node_1 = cluster1.listen_addr.to_string();
+        let node_1 = cluster1.gossip_listen_addr.to_string();
         let cluster2 = create_cluster_for_test_with_id(
             2u16,
             cluster_id.to_string(),
@@ -742,7 +745,7 @@ mod tests {
             &transport,
         )
         .await?;
-        let node_2 = cluster2.listen_addr.to_string();
+        let node_2 = cluster2.gossip_listen_addr.to_string();
         let cluster3 = create_cluster_for_test_with_id(
             3u16,
             cluster_id.to_string(),
@@ -767,15 +770,15 @@ mod tests {
             .sorted()
             .collect();
         let mut expected_members = vec![
-            cluster1.listen_addr,
-            cluster2.listen_addr,
-            cluster3.listen_addr,
+            cluster1.gossip_listen_addr,
+            cluster2.gossip_listen_addr,
+            cluster3.gossip_listen_addr,
         ];
         expected_members.sort();
         assert_eq!(members, expected_members);
 
-        let cluster2_listen_addr = cluster2.listen_addr;
-        let cluster3_listen_addr = cluster3.listen_addr;
+        let cluster2_listen_addr = cluster2.gossip_listen_addr;
+        let cluster3_listen_addr = cluster3.gossip_listen_addr;
         cluster2.shutdown().await;
         cluster3.shutdown().await;
         cluster1
@@ -796,7 +799,7 @@ mod tests {
             &transport,
         )
         .await?;
-        let node_2 = cluster2.listen_addr.to_string();
+        let node_2 = cluster2.gossip_listen_addr.to_string();
 
         let cluster3 = Cluster::join(
             Member::new("new_node_3".to_string(), 1, cluster3_listen_addr),

quickwit-cluster/src/lib.rs

@@ -75,25 +75,19 @@ pub async fn start_cluster_service(
     quickwit_config: &QuickwitConfig,
     services: &HashSet<QuickwitService>,
 ) -> anyhow::Result<Arc<Cluster>> {
-    let seed_nodes = quickwit_config
-        .seed_socket_addrs()?
-        .iter()
-        .map(|addr| addr.to_string())
-        .collect::<Vec<_>>();
-
     let member = Member::new(
         quickwit_config.node_id.clone(),
         unix_timestamp(),
-        quickwit_config.gossip_public_addr()?,
+        quickwit_config.gossip_public_addr().await?,
     );
 
     let cluster = Cluster::join(
         member,
         services,
-        quickwit_config.gossip_socket_addr()?,
+        quickwit_config.gossip_listen_addr().await?,
         quickwit_config.cluster_id.clone(),
-        quickwit_config.grpc_socket_addr()?,
-        seed_nodes,
+        quickwit_config.grpc_public_addr().await?,
+        quickwit_config.peer_seed_addrs().await?,
         FailureDetectorConfig::default(),
         &UdpTransport,
     )

quickwit-common/src/net.rs

@@ -17,7 +17,96 @@
 // You should have received a copy of the GNU Affero General Public License
 // along with this program. If not, see <http://www.gnu.org/licenses/>.
 
-use std::net::{SocketAddr, TcpListener, ToSocketAddrs};
+use std::fmt::Display;
+use std::net::{IpAddr, SocketAddr, TcpListener};
+
+use anyhow::{bail, Context};
+use tokio::net::{lookup_host, ToSocketAddrs};
+
+/// Represents a host, i.e. an IP address (`127.0.0.1`) or a hostname (`localhost`).
+#[derive(Clone, Debug)]
+pub enum Host {
+    Hostname(String),
+    IpAddr(IpAddr),
+}
+
+impl Display for Host {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {
+        match self {
+            Host::Hostname(hostname) => hostname.fmt(formatter),
+            Host::IpAddr(ip_addr) => ip_addr.fmt(formatter),
+        }
+    }
+}
+
+/// Represents an address `<host>:<port>` where `host` can be an IP address or a hostname.
+#[derive(Clone, Debug)]
+pub struct HostAddr {
+    host: Host,
+    port: u16,
+}
+
+impl HostAddr {
+    /// Attempts to parse a `host_addr`.
+    /// If no port is defined, it just accepts the host and uses the given default port.
+    ///
+    /// This function supports:
+    /// - IPv4
+    /// - IPv4:port
+    /// - IPv6
+    /// - \[IPv6\]:port -- IpV6 contains colon. It is customary to require bracket for this reason.
+    /// - hostname
+    /// - hostname:port
+    pub fn parse_with_default_port(host_addr: &str, default_port: u16) -> anyhow::Result<Self> {
+        if let Ok(socket_addr) = host_addr.parse::<SocketAddr>() {
+            return Ok(Self {
+                host: Host::IpAddr(socket_addr.ip()),
+                port: socket_addr.port(),
+            });
+        }
+        if let Ok(ip_addr) = host_addr.parse::<IpAddr>() {
+            return Ok(Self {
+                host: Host::IpAddr(ip_addr),
+                port: default_port,
+            });
+        }
+        let (hostname, port) = if let Some((hostname_str, port_str)) = host_addr.split_once(':') {
+            let port_u16 = port_str.parse::<u16>().with_context(|| {
+                format!("Failed to parse address `{}`: port is invalid.", host_addr)
+            })?;
+            (hostname_str, port_u16)
+        } else {
+            (host_addr, default_port)
+        };
+        if !is_valid_hostname(hostname) {
+            bail!(
+                "Failed to parse address `{}`: hostname is invalid.",
+                host_addr
+            )
+        }
+        Ok(Self {
+            host: Host::Hostname(hostname.to_string()),
+            port,
+        })
+    }
+
+    /// Resolves the host if necessary and returns a `SocketAddr`.
+    pub async fn to_socket_addr(&self) -> anyhow::Result<SocketAddr> {
+        match &self.host {
+            Host::IpAddr(ip_addr) => Ok(SocketAddr::new(*ip_addr, self.port)),
+            Host::Hostname(hostname) => get_socket_addr(&(hostname.as_str(), self.port)).await,
+        }
+    }
+}
+
+impl Display for HostAddr {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {
+        match self.host {
+            Host::IpAddr(IpAddr::V6(_)) => write!(formatter, "[{}]:{}", self.host, self.port),
+            _ => write!(formatter, "{}:{}", self.host, self.port),
+        }
+    }
+}
 
 /// Finds a random available TCP port.
 pub fn find_available_tcp_port() -> anyhow::Result<u16> {
@@ -28,100 +117,133 @@ pub fn find_available_tcp_port() -> anyhow::Result<u16> {
 }
 
 /// Converts an object into a resolved `SocketAddr`.
-pub fn get_socket_addr<T: ToSocketAddrs + std::fmt::Debug>(addr: &T) -> anyhow::Result<SocketAddr> {
-    addr.to_socket_addrs()?
-        .next()
-        .ok_or_else(|| anyhow::anyhow!("Failed to resolve address `{:?}`.", addr))
-}
-
-/// Returns true if the socket addr is a valid socket address containing a port.
-///
-/// If the socket adddress looks invalid to begin with, we may return false or true.
-fn contains_port(addr: &str) -> bool {
-    // [IPv6]:port
-    if let Some((_, colon_port)) = addr[1..].rsplit_once(']') {
-        return colon_port.starts_with(':');
-    }
-    if let Some((host, _port)) = addr[1..].rsplit_once(':') {
-        // if host contains a ":" then is thi is probably a IPv6 address.
-        return !host.contains(':');
-    }
-    false
-}
-
-/// Attempts to parse a `socket_addr`.
-/// If no port is defined, it just accepts the address and uses the given default port.
-///
-/// This function supports
-/// - IPv4
-/// - IPv4:port
-/// - IPv6
-/// - \[IPv6\]:port -- IpV6 contains colon. It is customary to require bracket for this reason.
-/// - hostname
-/// - hostname:port
-/// with or without a port.
-///
-/// Note that this function returns a SocketAddr, so that if a hostname
-/// is given, DNS resolution will happen once and for all.
-pub fn parse_socket_addr_with_default_port(
-    addr: &str,
-    default_port: u16,
+pub async fn get_socket_addr<T: ToSocketAddrs + std::fmt::Debug>(
+    addr: &T,
 ) -> anyhow::Result<SocketAddr> {
-    if contains_port(addr) {
-        get_socket_addr(&addr)
-    } else {
-        get_socket_addr(&(addr, default_port))
+    lookup_host(addr)
+        .await
+        .with_context(|| format!("Failed to parse address or resolve hostname `{:?}`.", addr))?
+        .next()
+        .ok_or_else(|| {
+            anyhow::anyhow!(
+                "DNS resolution did not yield any record for hostname `{:?}`.",
+                addr
+            )
+        })
+}
+
+/// Returns whether a hostname is valid according to [IETF RFC 1123](https://tools.ietf.org/html/rfc1123).
+///
+/// A hostname is valid if the following conditions are met:
+///
+/// - It does not start or end with `-` or `.`.
+/// - It does not contain any characters outside of the alphanumeric range, except for `-` and `.`.
+/// - It is not empty.
+/// - It is 253 or fewer characters.
+/// - Its labels (characters separated by `.`) are not empty.
+/// - Its labels are 63 or fewer characters.
+/// - Its labels do not start or end with '-' or '.'.
+fn is_valid_hostname(hostname: &str) -> bool {
+    if hostname.is_empty() || hostname.len() > 253 {
+        return false;
     }
+    if !hostname
+        .chars()
+        .all(|ch| ch.is_ascii_alphanumeric() || ch == '-' || ch == '.')
+    {
+        return false;
+    }
+    if hostname.split('.').any(|label| {
+        label.is_empty() || label.len() > 63 || label.starts_with('-') || label.ends_with('-')
+    }) {
+        return false;
+    }
+    true
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
 
-    fn test_parse_socket_addr_helper(addr: &str, expected_opt: Option<&str>) {
-        let socket_addr_res = parse_socket_addr_with_default_port(addr, 1337);
-        if let Some(expected) = expected_opt {
+    fn test_parse_addr_helper(addr: &str, expected_addr_opt: Option<&str>) {
+        let addr_res = HostAddr::parse_with_default_port(addr, 1337);
+        if let Some(expected_addr) = expected_addr_opt {
             assert!(
-                socket_addr_res.is_ok(),
-                "Parsing `{}` was expected to succeed",
+                addr_res.is_ok(),
+                "Parsing `{}` was expected to succeed.",
                 addr
             );
-            let socket_addr = socket_addr_res.unwrap();
-            let expected_socket_addr: SocketAddr = expected.parse().unwrap();
-            assert_eq!(socket_addr, expected_socket_addr);
+            assert_eq!(addr_res.unwrap().to_string(), expected_addr);
         } else {
             assert!(
-                socket_addr_res.is_err(),
-                "Parsing `{}` was expected to fail",
-                addr
+                addr_res.is_err(),
+                "Parsing `{}` was expected to fail, got `{}`",
+                addr,
+                addr_res.unwrap()
             );
         }
     }
 
-    #[test]
-    fn test_parse_socket_addr_with_ips() {
-        test_parse_socket_addr_helper("127.0.0.1", Some("127.0.0.1:1337"));
-        test_parse_socket_addr_helper("127.0.0.1:100", Some("127.0.0.1:100"));
-        test_parse_socket_addr_helper("127.0..1:100", None);
-        test_parse_socket_addr_helper(
+    #[tokio::test]
+    async fn test_parse_addr_with_ips() {
+        // IPv4
+        test_parse_addr_helper("127.0.0.1", Some("127.0.0.1:1337"));
+        test_parse_addr_helper("127.0.0.1:100", Some("127.0.0.1:100"));
+        test_parse_addr_helper("127.0..1:100", None);
+
+        // IPv6
+        test_parse_addr_helper(
             "2001:0db8:85a3:0000:0000:8a2e:0370:7334",
-            Some("[2001:0db8:85a3:0000:0000:8a2e:0370:7334]:1337"),
+            Some("[2001:db8:85a3::8a2e:370:7334]:1337"),
         );
-        test_parse_socket_addr_helper("2001:0db8:85a3:0000:0000:8a2e:0370:7334:1000", None);
-        test_parse_socket_addr_helper(
+        test_parse_addr_helper("2001:0db8:85a3:0000:0000:8a2e:0370:7334:1000", None);
+        test_parse_addr_helper(
             "[2001:0db8:85a3:0000:0000:8a2e:0370:7334]:1000",
-            Some("[2001:0db8:85a3:0000:0000:8a2e:0370:7334]:1000"),
+            Some("[2001:db8:85a3::8a2e:370:7334]:1000"),
         );
-        test_parse_socket_addr_helper("[2001:0db8:1000", None);
-        test_parse_socket_addr_helper("2001:0db8:85a3:0000:0000:8a2e:0370:7334]:1000", None);
+        test_parse_addr_helper("[2001:0db8:1000", None);
+        test_parse_addr_helper("2001:0db8:85a3:0000:0000:8a2e:0370:7334]:1000", None);
+
+        // Hostname
+        test_parse_addr_helper("google.com", Some("google.com:1337"));
+        test_parse_addr_helper("google.com:1000", Some("google.com:1000"));
     }
 
-    // This test require DNS.
     #[test]
-    fn test_parse_socket_addr_with_resolution() {
-        let socket_addr = parse_socket_addr_with_default_port("google.com:1000", 1337).unwrap();
-        assert_eq!(socket_addr.port(), 1000);
-        let socket_addr = parse_socket_addr_with_default_port("google.com", 1337).unwrap();
-        assert_eq!(socket_addr.port(), 1337);
+    fn test_is_valid_hostname() {
+        for hostname in &[
+            "VaLiD-HoStNaMe",
+            "50-name",
+            "235235",
+            "example.com",
+            "VaLid.HoStNaMe",
+            "123.456",
+        ] {
+            assert!(
+                is_valid_hostname(hostname),
+                "Hostname `{}` is valid.",
+                hostname
+            );
+        }
+
+        for hostname in &[
+            "-invalid-name",
+            "also-invalid-",
+            "asdf@fasd",
+            "@asdfl",
+            "asd f@",
+            ".invalid",
+            "invalid.name.",
+            "foo.label-is-way-to-longgggggggggggggggggggggggggggggggggggggggggggg.org",
+            "invalid.-starting.char",
+            "invalid.ending-.char",
+            "empty..label",
+        ] {
+            assert!(
+                !is_valid_hostname(hostname),
+                "Hostname `{}` is invalid.",
+                hostname
+            );
+        }
     }
 }

quickwit-common/src/uri.rs

@@ -203,6 +203,7 @@ impl PartialEq<&str> for Uri {
         &self.uri == other
     }
 }
+
 impl PartialEq<String> for Uri {
     fn eq(&self, other: &String) -> bool {
         &self.uri == other

quickwit-config/src/config.rs

@@ -25,7 +25,7 @@ use anyhow::{bail, Context};
 use byte_unit::Byte;
 use json_comments::StripComments;
 use once_cell::sync::OnceCell;
-use quickwit_common::net::{get_socket_addr, parse_socket_addr_with_default_port};
+use quickwit_common::net::{get_socket_addr, HostAddr};
 use quickwit_common::new_coolid;
 use quickwit_common::uri::{Extension, Uri, FILE_PROTOCOL};
 use serde::de::Error;
@@ -183,7 +183,7 @@ pub struct QuickwitConfig {
     pub peer_seeds: Vec<String>,
     #[serde(default)]
     #[serde(deserialize_with = "deser_valid_uri")]
-    metastore_uri: Option<Uri>,
+    pub metastore_uri: Option<Uri>,
     #[serde(default)]
     #[serde(deserialize_with = "deser_valid_uri")]
     default_index_root_uri: Option<Uri>,
@@ -280,69 +280,77 @@ impl QuickwitConfig {
         Ok(())
     }
 
-    pub fn rest_socket_addr(&self) -> anyhow::Result<SocketAddr> {
-        get_socket_addr(&(self.listen_address.as_str(), self.rest_listen_port))
+    /// Returns the REST listen address of the node, i.e. the socket address on which the REST API
+    /// service listens for TCP connections.
+    pub async fn rest_listen_addr(&self) -> anyhow::Result<SocketAddr> {
+        get_socket_addr(&(self.listen_address.as_str(), self.rest_listen_port)).await
     }
 
-    /// Returns the grpc listen port
-    /// extracted from config if specified or computed from `rest_listen_port + 1`.
+    /// Returns the gRPC listen port of the node.
+    /// Extracted from the config if specified or computed as `rest_listen_port + 1` otherwise.
     fn grpc_listen_port(&self) -> u16 {
-        match self.grpc_listen_port {
-            Some(grpc_listen_port) => grpc_listen_port,
-            None => self.rest_listen_port + 1,
+        self.grpc_listen_port.unwrap_or(self.rest_listen_port + 1)
+    }
+
+    /// Returns the gRPC listen address of the node, i.e. the socket address on which the gRPC
+    /// service listens for TCP connections.
+    pub async fn grpc_listen_addr(&self) -> anyhow::Result<SocketAddr> {
+        get_socket_addr(&(self.listen_address.as_str(), self.grpc_listen_port())).await
+    }
+
+    /// Returns the gRPC public address of the node, i.e. the socket address to connect to in order
+    /// to send gRPC requests to the node.
+    pub async fn grpc_public_addr(&self) -> anyhow::Result<SocketAddr> {
+        match env::var("QW_PUBLIC_IP") {
+            Ok(ip_addr) => get_socket_addr(&(ip_addr, self.grpc_listen_port())).await,
+            Err(_) => self.grpc_listen_addr().await,
         }
     }
 
-    /// Returns the gossip listen port
-    /// extracted from config if specified or the same as `rest_listen_port`.
+    /// Returns the gossip listen port of the node (UDP).
+    /// Extracted from the config if specified or same as `rest_listen_port` otherwise.
     fn gossip_listen_port(&self) -> u16 {
-        match self.gossip_listen_port {
-            Some(grpc_listen_port) => grpc_listen_port,
-            // By default, we use the same port number as the rest port but this is UDP.
-            None => self.rest_listen_port,
+        // By default, we use the same port number as the REST port but UDP this time.
+        self.gossip_listen_port.unwrap_or(self.rest_listen_port)
+    }
+
+    /// Returns the gossip listen address of the node, i.e. the UDP socket address on which the node
+    /// receives gossip messages.
+    pub async fn gossip_listen_addr(&self) -> anyhow::Result<SocketAddr> {
+        get_socket_addr(&(self.listen_address.as_str(), self.gossip_listen_port())).await
+    }
+
+    /// Returns the gossip public address of the node, i.e. the socket address to send UDP packets
+    /// to in order to gossip with the node.
+    pub async fn gossip_public_addr(&self) -> anyhow::Result<SocketAddr> {
+        match env::var("QW_PUBLIC_IP") {
+            Ok(ip_addr) => get_socket_addr(&(ip_addr, self.gossip_listen_port())).await,
+            Err(_) => self.gossip_listen_addr().await,
         }
     }
 
-    /// Returns the grpc socket address.
-    pub fn grpc_socket_addr(&self) -> anyhow::Result<SocketAddr> {
-        get_socket_addr(&(self.listen_address.as_str(), self.grpc_listen_port()))
-    }
+    /// Returns the list of peer seed addresses. The addresses MUST NOT be resolved. Otherwise, the
+    /// DNS-based discovery mechanism implemented in Chitchat will not work correctly.
+    pub async fn peer_seed_addrs(&self) -> anyhow::Result<Vec<String>> {
+        let mut peer_seed_addrs = Vec::new();
+        let default_gossip_port = self.gossip_listen_port();
 
-    /// Returns the gossip socket address.
-    pub fn gossip_socket_addr(&self) -> anyhow::Result<SocketAddr> {
-        get_socket_addr(&(self.listen_address.as_str(), self.gossip_listen_port()))
-    }
-
-    /// The node gossip_public_address should ideally be specified via config;
-    /// environment variable interpolation if necessary.
-    /// right now we just try to fetch from environment variable otherwise fallback
-    /// to listen_address.
-    pub fn gossip_public_addr(&self) -> anyhow::Result<SocketAddr> {
-        match env::var("QW_GOSSIP_PUBLIC_ADDRESS") {
-            Ok(addr) => Ok(addr.parse::<SocketAddr>()?),
-            Err(_) => self.gossip_socket_addr(),
+        for peer_seed in &self.peer_seeds {
+            let peer_seed_addr =
+                HostAddr::parse_with_default_port(peer_seed.as_str(), default_gossip_port)?;
+            if let Err(error) = peer_seed_addr.to_socket_addr().await {
+                warn!(peer_seed = %peer_seed_addr, error = ?error, "Failed to resolve peer seed address.");
+                continue;
+            }
+            peer_seed_addrs.push(peer_seed_addr.to_string())
         }
-    }
-
-    pub fn seed_socket_addrs(&self) -> anyhow::Result<Vec<SocketAddr>> {
-        // If no port is given, we assume a seed is using the same port as ourself.
-        let default_gossip_port = self.gossip_socket_addr()?.port();
-        let seed_socket_addrs: Vec<SocketAddr> = self
-            .peer_seeds
-            .iter()
-            .flat_map(|seed| {
-                parse_socket_addr_with_default_port(seed, default_gossip_port)
-                    .map_err(|_| warn!(address = %seed, "Failed to resolve seed address."))
-            })
-            .collect();
-
-        if !self.peer_seeds.is_empty() && seed_socket_addrs.is_empty() {
+        if !self.peer_seeds.is_empty() && peer_seed_addrs.is_empty() {
             bail!(
-                "Failed to resolve any of the seed addresses: `{}`",
+                "Failed to resolve any of the peer seed addresses: `{}`",
                 self.peer_seeds.join(", ")
             )
         }
-        Ok(seed_socket_addrs)
+        Ok(peer_seed_addrs)
     }
 
     #[doc(hidden)]
@@ -574,14 +582,14 @@ mod tests {
         assert!(quickwit_config.validate().is_ok());
     }
 
-    #[test]
-    fn test_peer_socket_addrs() {
+    #[tokio::test]
+    async fn test_peer_socket_addrs() {
         {
             let quickwit_config = QuickwitConfig {
                 rest_listen_port: 1789,
                 ..Default::default()
             };
-            assert!(quickwit_config.seed_socket_addrs().unwrap().is_empty());
+            assert!(quickwit_config.peer_seed_addrs().await.unwrap().is_empty());
         }
         {
             let quickwit_config = QuickwitConfig {
@@ -589,38 +597,61 @@ mod tests {
                 peer_seeds: vec!["unresolvable-host".to_string()],
                 ..Default::default()
             };
-            assert!(quickwit_config.seed_socket_addrs().is_err());
+            assert!(quickwit_config.peer_seed_addrs().await.is_err());
         }
         {
             let quickwit_config = QuickwitConfig {
                 rest_listen_port: 1789,
-                peer_seeds: vec!["unresolvable-host".to_string(), "127.0.0.1".to_string()],
+                peer_seeds: vec![
+                    "unresolvable-host".to_string(),
+                    "localhost".to_string(),
+                    "localhost:1337".to_string(),
+                    "127.0.0.1".to_string(),
+                    "127.0.0.1:1337".to_string(),
+                ],
                 ..Default::default()
             };
             assert_eq!(
-                quickwit_config.seed_socket_addrs().unwrap(),
-                vec!["127.0.0.1:1789".parse().unwrap()]
+                quickwit_config.peer_seed_addrs().await.unwrap(),
+                vec![
+                    "localhost:1789".to_string(),
+                    "localhost:1337".to_string(),
+                    "127.0.0.1:1789".to_string(),
+                    "127.0.0.1:1337".to_string()
+                ]
             );
         }
     }
 
-    #[test]
-    fn test_socket_addr_ports() {
+    #[tokio::test]
+    async fn test_socket_addr_ports() {
         {
             let quickwit_config = QuickwitConfig {
                 listen_address: "127.0.0.1".to_string(),
                 ..Default::default()
             };
             assert_eq!(
-                quickwit_config.rest_socket_addr().unwrap().to_string(),
+                quickwit_config
+                    .rest_listen_addr()
+                    .await
+                    .unwrap()
+                    .to_string(),
                 "127.0.0.1:7280"
             );
             assert_eq!(
-                quickwit_config.gossip_socket_addr().unwrap().to_string(),
+                quickwit_config
+                    .gossip_listen_addr()
+                    .await
+                    .unwrap()
+                    .to_string(),
                 "127.0.0.1:7280"
             );
             assert_eq!(
-                quickwit_config.grpc_socket_addr().unwrap().to_string(),
+                quickwit_config
+                    .grpc_listen_addr()
+                    .await
+                    .unwrap()
+                    .to_string(),
                 "127.0.0.1:7281"
             );
         }
@@ -631,15 +662,27 @@ mod tests {
                 ..Default::default()
             };
             assert_eq!(
-                quickwit_config.rest_socket_addr().unwrap().to_string(),
+                quickwit_config
+                    .rest_listen_addr()
+                    .await
+                    .unwrap()
+                    .to_string(),
                 "127.0.0.1:1789"
             );
             assert_eq!(
-                quickwit_config.gossip_socket_addr().unwrap().to_string(),
+                quickwit_config
+                    .gossip_listen_addr()
+                    .await
+                    .unwrap()
+                    .to_string(),
                 "127.0.0.1:1789"
             );
             assert_eq!(
-                quickwit_config.grpc_socket_addr().unwrap().to_string(),
+                quickwit_config
+                    .grpc_listen_addr()
+                    .await
+                    .unwrap()
+                    .to_string(),
                 "127.0.0.1:1790"
             );
         }
@@ -652,15 +695,27 @@ mod tests {
                 ..Default::default()
             };
             assert_eq!(
-                quickwit_config.rest_socket_addr().unwrap().to_string(),
+                quickwit_config
+                    .rest_listen_addr()
+                    .await
+                    .unwrap()
+                    .to_string(),
                 "127.0.0.1:1789"
             );
             assert_eq!(
-                quickwit_config.gossip_socket_addr().unwrap().to_string(),
+                quickwit_config
+                    .gossip_listen_addr()
+                    .await
+                    .unwrap()
+                    .to_string(),
                 "127.0.0.1:1889"
             );
             assert_eq!(
-                quickwit_config.grpc_socket_addr().unwrap().to_string(),
+                quickwit_config
+                    .grpc_listen_addr()
+                    .await
+                    .unwrap()
+                    .to_string(),
                 "127.0.0.1:1989"
             );
         }

quickwit-core/src/index.rs

@@ -17,6 +17,7 @@
 // You should have received a copy of the GNU Affero General Public License
 // along with this program. If not, see <http://www.gnu.org/licenses/>.
 
+use std::io;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use std::time::Duration;
@@ -311,12 +312,12 @@ pub async fn clear_cache_directory(
 /// * `data_dir_path` - Path to directory where data (tmp data, splits kept for caching purpose) is
 ///   persisted.
 /// * `index_id` - The target index ID.
-pub async fn remove_indexing_directory(
-    data_dir_path: &Path,
-    index_id: String,
-) -> anyhow::Result<()> {
+pub async fn remove_indexing_directory(data_dir_path: &Path, index_id: String) -> io::Result<()> {
     let indexing_directory_path = data_dir_path.join(INDEXING_DIR_NAME).join(index_id);
     info!(path = %indexing_directory_path.display(), "Clearing indexing directory.");
-    tokio::fs::remove_dir_all(indexing_directory_path.as_path()).await?;
-    Ok(())
+    match tokio::fs::remove_dir_all(indexing_directory_path.as_path()).await {
+        Ok(_) => Ok(()),
+        Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()),
+        Err(error) => Err(error),
+    }
 }

quickwit-indexing/src/lib.rs

@@ -19,6 +19,7 @@
 
 use std::sync::Arc;
 
+use itertools::Itertools;
 use quickwit_actors::{Mailbox, Universe};
 use quickwit_config::QuickwitConfig;
 use quickwit_ingest_api::IngestApiService;
@@ -64,8 +65,7 @@ pub async fn start_indexer_service(
     storage_uri_resolver: StorageUriResolver,
     ingest_api_service: Option<Mailbox<IngestApiService>>,
 ) -> anyhow::Result<Mailbox<IndexingService>> {
-    info!("start-indexer-service");
-    let index_metadatas = metastore.list_indexes_metadatas().await?;
+    info!("Starting indexer service.");
     let indexing_server = IndexingService::new(
         config.data_dir_path.to_path_buf(),
         config.indexer_config.clone(),
@@ -74,8 +74,11 @@ pub async fn start_indexer_service(
         ingest_api_service.clone(),
     );
     let (indexer_service_mailbox, _) = universe.spawn_actor(indexing_server).spawn();
+
+    let index_metadatas = metastore.list_indexes_metadatas().await?;
+    info!(index_ids = %index_metadatas.iter().map(|im| &im.index_id).join(", "), "Spawning indexing pipeline(s).");
+
     for index_metadata in index_metadatas {
-        info!(index_id=%index_metadata.index_id, "spawn-indexing-pipeline");
         indexer_service_mailbox
             .ask_for_res(SpawnPipelinesForIndex {
                 index_id: index_metadata.index_id,

quickwit-metastore/src/metastore/postgresql_metastore.rs

@@ -93,19 +93,22 @@ fn initialize_db(pool: &Pool<ConnectionManager<PgConnection>>) -> anyhow::Result
     match embedded_migrations::run_with_output(&*db_conn, &mut migrations_log_buffer) {
         Ok(_) => {
             let migrations_log = String::from_utf8_lossy(&migrations_log_buffer);
-            info!(
-                migrations_log = migrations_log.as_ref(),
-                "Database migrations succeeded"
-            );
+            if !migrations_log.is_empty() {
+                info!(
+                    migrations_log = migrations_log.as_ref(),
+                    "Database migrations succeeded."
+                );
+            }
             Ok(())
         }
-        Err(err) => {
+        Err(error) => {
             let migrations_log = String::from_utf8_lossy(&migrations_log_buffer);
             error!(
                 migrations_log = migrations_log.as_ref(),
-                "Database migrations failed"
+                error = ?error,
+                "Database migrations failed."
             );
-            Err(anyhow::anyhow!(err))
+            Err(anyhow::anyhow!(error))
         }
     }
 }
@@ -141,13 +144,13 @@ impl PostgresqlMetastore {
             );
             match connection_pool.get() {
                 Ok(_conn) => {
-                    info!("The connection pool works fine");
+                    info!("Connection to PostgreSQL database successfully established.");
                     is_status_ok = true;
                     break;
                 }
                 Err(err) => {
-                    warn!(err=?err, "Failed to get connection from the connection pool. Trying again");
                     retry_cnt += 1;
+                    warn!(num_attempts = %retry_cnt, err = ?err, "Failed to get connection from the connection pool. Trying again");
                     tokio::time::sleep(CONNECTION_STATUS_CHECK_INTERVAL).await;
                 }
             }
@@ -158,10 +161,9 @@ impl PostgresqlMetastore {
                 CONNECTION_STATUS_CHECK_MAX_RETRY_COUNT
             );
             return Err(MetastoreError::ConnectionError {
-                message: "The connection pool does not work fine".to_string(),
+                message: "Failed to establish connection to PostgreSQL database.".to_string(),
             });
         }
-
         initialize_db(&connection_pool).map_err(|err| MetastoreError::InternalError {
             message: "Failed to initialize database".to_string(),
             cause: anyhow::anyhow!(err),

quickwit-search/src/lib.rs

@@ -190,7 +190,7 @@ pub async fn single_node_search(
 ) -> crate::Result<SearchResponse> {
     let start_instant = tokio::time::Instant::now();
     let index_metadata = metastore.index_metadata(&search_request.index_id).await?;
-    let index_storage = storage_resolver.resolve(&index_metadata.index_uri)?;
+    let index_storage = storage_resolver.resolve(index_metadata.index_uri.as_ref())?;
     let metas = list_relevant_splits(search_request, metastore).await?;
     let split_metadata: Vec<SplitIdAndFooterOffsets> =
         metas.iter().map(extract_split_and_footer_offsets).collect();

quickwit-search/src/search_client_pool.rs

@@ -345,7 +345,9 @@ mod tests {
         let client_pool = SearchClientPool::create_and_keep_updated(cluster.clone()).await?;
         let clients = client_pool.clients();
         let addrs: Vec<SocketAddr> = clients.into_keys().collect();
-        let expected_addrs = vec![grpc_addr_from_listen_addr_for_test(cluster.listen_addr)];
+        let expected_addrs = vec![grpc_addr_from_listen_addr_for_test(
+            cluster.gossip_listen_addr,
+        )];
         assert_eq!(addrs, expected_addrs);
         Ok(())
     }
@@ -354,7 +356,7 @@ mod tests {
     async fn test_search_client_pool_multiple_nodes() -> anyhow::Result<()> {
         let transport = ChannelTransport::default();
         let cluster1 = create_cluster_simple_for_test(&transport).await?;
-        let node_1 = cluster1.listen_addr.to_string();
+        let node_1 = cluster1.gossip_listen_addr.to_string();
         let cluster2 = create_cluster_for_test(vec![node_1], &["searcher"], &transport).await?;
 
         cluster1
@@ -366,8 +368,8 @@ mod tests {
 
         let addrs: Vec<SocketAddr> = clients.into_keys().sorted().collect();
         let mut expected_addrs = vec![
-            grpc_addr_from_listen_addr_for_test(cluster1.listen_addr),
-            grpc_addr_from_listen_addr_for_test(cluster2.listen_addr),
+            grpc_addr_from_listen_addr_for_test(cluster1.gossip_listen_addr),
+            grpc_addr_from_listen_addr_for_test(cluster2.gossip_listen_addr),
         ];
         expected_addrs.sort();
         assert_eq!(addrs, expected_addrs);
@@ -388,8 +390,10 @@ mod tests {
 
         let assigned_jobs = client_pool.assign_jobs(jobs, &HashSet::default())?;
         let expected_assigned_jobs = vec![(
-            create_search_service_client(grpc_addr_from_listen_addr_for_test(cluster.listen_addr))
-                .await?,
+            create_search_service_client(grpc_addr_from_listen_addr_for_test(
+                cluster.gossip_listen_addr,
+            ))
+            .await?,
             vec![
                 SearchJob::for_test("split4", 4),
                 SearchJob::for_test("split3", 3),

quickwit-serve/src/grpc.rs

@@ -30,12 +30,12 @@ use crate::cluster_api::GrpcClusterAdapter;
 use crate::search_api::GrpcSearchAdapter;
 use crate::QuickwitServices;
 
-/// Start gRPC service given a gRPC address and a search service and cluster service.
+/// Starts gRPC service given a gRPC address and a search service and cluster service.
 pub(crate) async fn start_grpc_server(
-    grpc_addr: SocketAddr,
+    grpc_listen_addr: SocketAddr,
     quickwit_services: &QuickwitServices,
 ) -> anyhow::Result<()> {
-    info!(grpc_addr=?grpc_addr, "Start gRPC service.");
+    info!(grpc_listen_addr = ?grpc_listen_addr, "Starting gRPC server.");
 
     let mut server = Server::builder();
 
@@ -52,6 +52,6 @@ pub(crate) async fn start_grpc_server(
         server_router = server_router.add_service(SearchServiceServer::new(grpc_search_service));
     }
 
-    server_router.serve(grpc_addr).await?;
+    server_router.serve(grpc_listen_addr).await?;
     Ok(())
 }

quickwit-serve/src/lib.rs

@@ -139,7 +139,6 @@ pub async fn serve_quickwit(
         storage_resolver,
         config.default_index_root_uri(),
     ));
-
     let quickwit_services = QuickwitServices {
         ingest_api_service,
         cluster_service,
@@ -148,15 +147,13 @@ pub async fn serve_quickwit(
         index_service,
         services: services.clone(),
     };
+    let grpc_listen_addr = config.grpc_listen_addr().await?;
+    let grpc_server = grpc::start_grpc_server(grpc_listen_addr, &quickwit_services);
 
-    let rest_addr = config.rest_socket_addr()?;
-    let grpc_addr = config.grpc_socket_addr()?;
-
-    let grpc_server = grpc::start_grpc_server(grpc_addr, &quickwit_services);
-    let rest_server = rest::start_rest_server(rest_addr, &quickwit_services);
-
-    tokio::try_join!(rest_server, grpc_server)?;
+    let rest_listen_addr = config.rest_listen_addr().await?;
+    let rest_server = rest::start_rest_server(rest_listen_addr, &quickwit_services);
 
+    tokio::try_join!(grpc_server, rest_server)?;
     Ok(())
 }
 
@@ -171,19 +168,21 @@ async fn check_is_configured_for_cluster(
     if config.peer_seeds.is_empty() {
         return Ok(());
     }
-
     let metastore_uri = Uri::try_new(&metastore.uri())?;
     if metastore_uri.protocol() == FILE_PROTOCOL || metastore_uri.protocol() == S3_PROTOCOL {
         anyhow::bail!(
-            "Cluster feature cannot be used in conjunction with a file backed metastore."
+            "Quickwit cannot run in cluster mode with a file-backed metastore. Please, use a \
+             PostgreSQL metastore instead."
         );
     }
-
     for index_metadata in metastore.list_indexes_metadatas().await? {
         let index_uri = Uri::try_new(&index_metadata.index_uri)?;
         if index_uri.protocol() == FILE_PROTOCOL {
             anyhow::bail!(
-                "Quickwit cluster feature cannot be used in conjunction with a file storage."
+                "Quickwit cannot run in cluster mode with an index whose data is stored on a \
+                 local file system. Index URI for index `{}` is `{}`.",
+                index_metadata.index_id,
+                index_uri
             );
         }
     }

quickwit-serve/src/rest.rs

@@ -35,12 +35,12 @@ use crate::search_api::{search_get_handler, search_post_handler, search_stream_h
 use crate::ui_handler::ui_handler;
 use crate::{Format, QuickwitServices};
 
-/// Start REST service given a HTTP address and a search service.
+/// Starts REST service given a HTTP address and a search service.
 pub(crate) async fn start_rest_server(
-    rest_addr: SocketAddr,
+    rest_listen_addr: SocketAddr,
     quickwit_services: &QuickwitServices,
 ) -> anyhow::Result<()> {
-    info!(rest_addr=?rest_addr, "Starting REST service.");
+    info!(rest_listen_addr = ?rest_listen_addr, "Starting REST server.");
     let request_counter = warp::log::custom(|_| {
         crate::COUNTERS.num_requests.inc();
     });
@@ -78,8 +78,8 @@ pub(crate) async fn start_rest_server(
         .with(request_counter)
         .recover(recover_fn);
 
-    info!("Searcher ready to accept requests at http://{rest_addr}/");
-    warp::serve(rest_routes).run(rest_addr).await;
+    info!("Searcher ready to accept requests at http://{rest_listen_addr}/");
+    warp::serve(rest_routes).run(rest_listen_addr).await;
     Ok(())
 }
 

quickwit-storage/src/storage_resolver.rs

@@ -113,7 +113,7 @@ impl StorageUriResolver {
                 kind: storage_error.kind(),
                 message: storage_error
                     .source()
-                    .map(|err| format!("{}", err))
+                    .map(|err| format!("{:?}", err))
                     .unwrap_or_else(String::new),
             }
         })?;