Avoid primary fallback for read replica routing

Author: shuheiktgw

config/quickwit.yaml

@@ -105,8 +105,10 @@ version: 0.8
 #
 # Optional PostgreSQL read replica URI. Nodes started with the
 # `metastore_read_replica` service connect to it over a read-only connection and
-# serve stale-tolerant read-only metastore requests (e.g. from searchers). When
-# unset, those requests are served by the primary metastore. Defaults to unset.
+# serve stale-tolerant read-only metastore requests (e.g. from searchers).
+# When set, nodes that issue stale-tolerant read-only metastore requests require
+# a `metastore_read_replica` node at startup and do not fall back to the primary
+# metastore. Defaults to unset.
 #
 # metastore_read_replica_uri: postgres://username:password@read-replica-host:port/db
 #

docs/configuration/node-config.md

@@ -30,7 +30,7 @@ A commented example is available here: [quickwit.yaml](https://github.com/quickw
 | `peer_seeds` | List of IP addresses or hostnames used to bootstrap the cluster and discover the complete set of nodes. This list may contain the current node address and does not need to be exhaustive. If the list of peer seeds contains a host name, Quickwit will resolve it by querying the DNS every minute. On kubernetes for instance, it is a good practise to set it to a [headless service](https://kubernetes.io/docs/concepts/services-networking/service/#headless-services). | `QW_PEER_SEEDS` | |
 | `data_dir` | Path to directory where data (tmp data, splits kept for caching purpose) is persisted. This is mostly used in indexing. | `QW_DATA_DIR` | `./qwdata` |
 | `metastore_uri` | Metastore URI. Can be a local directory or `s3://my-bucket/indexes` or `postgres://username:password@localhost:5432/metastore`. [Learn more about the metastore configuration](metastore-config.md). | `QW_METASTORE_URI` | `{data_dir}/indexes` |
-| `metastore_read_replica_uri` | Optional PostgreSQL read replica URI. Nodes running the `metastore_read_replica` service connect to it over a read-only connection and serve stale-tolerant read-only metastore requests (e.g. from searchers). | `QW_METASTORE_READ_REPLICA_URI` | |
+| `metastore_read_replica_uri` | Optional PostgreSQL read replica URI. Nodes running the `metastore_read_replica` service connect to it over a read-only connection and serve stale-tolerant read-only metastore requests (e.g. from searchers). When set, nodes that issue those requests require a `metastore_read_replica` node at startup and do not fall back to the primary metastore. | `QW_METASTORE_READ_REPLICA_URI` | |
 | `default_index_root_uri` | Default index root URI that defines the location where index data (splits) is stored. The index URI is built following the scheme: `{default_index_root_uri}/{index-id}` | `QW_DEFAULT_INDEX_ROOT_URI` | `{data_dir}/indexes` |
 | environment variable only | Log level of Quickwit. Can be a direct log level, or a comma separated list of `module_name=level` | `RUST_LOG` | `info` |
 

quickwit/quickwit-config/src/node_config/serialize.rs

@@ -374,9 +374,17 @@ fn validate(node_config: &NodeConfig) -> anyhow::Result<()> {
 
 /// Validates the configuration of the [`QuickwitService::MetastoreReadReplica`] role.
 ///
-/// The role serves the same gRPC service as [`QuickwitService::Metastore`], so the two cannot run
-/// on the same node, and it requires a PostgreSQL `metastore_read_replica_uri` to connect to.
+/// When configured, `metastore_read_replica_uri` must be a PostgreSQL URI. The
+/// [`QuickwitService::MetastoreReadReplica`] role serves the same gRPC service as
+/// [`QuickwitService::Metastore`], so the two cannot run on the same node, and the read-replica
+/// role requires `metastore_read_replica_uri` to connect to.
 fn validate_metastore_read_replica(node_config: &NodeConfig) -> anyhow::Result<()> {
+    if let Some(uri) = &node_config.metastore_read_replica_uri {
+        if !uri.protocol().is_database() {
+            bail!("`metastore_read_replica_uri` must point to a PostgreSQL database, got `{uri}`");
+        }
+    }
+
     let read_replica_enabled =
         node_config.is_service_enabled(QuickwitService::MetastoreReadReplica);
     if !read_replica_enabled {
@@ -389,10 +397,7 @@ fn validate_metastore_read_replica(node_config: &NodeConfig) -> anyhow::Result<(
         );
     }
     match &node_config.metastore_read_replica_uri {
-        Some(uri) if uri.protocol().is_database() => Ok(()),
-        Some(uri) => {
-            bail!("`metastore_read_replica_uri` must point to a PostgreSQL database, got `{uri}`")
-        }
+        Some(_) => Ok(()),
         None => {
             bail!(
                 "the `metastore_read_replica` service requires `metastore_read_replica_uri` to be \
@@ -1141,6 +1146,23 @@ mod tests {
         assert!(error.to_string().contains("must point to a PostgreSQL"));
     }
 
+    #[tokio::test]
+    async fn test_metastore_read_replica_uri_must_be_a_database_even_without_role() {
+        let config_yaml = r#"
+            version: 0.8
+            node_id: test-node
+            metastore_read_replica_uri: s3://not-a-database
+        "#;
+        let error = load_node_config_with_env(
+            ConfigFormat::Yaml,
+            config_yaml.as_bytes(),
+            &Default::default(),
+        )
+        .await
+        .unwrap_err();
+        assert!(error.to_string().contains("must point to a PostgreSQL"));
+    }
+
     #[tokio::test]
     async fn test_metastore_and_read_replica_roles_are_mutually_exclusive() {
         let config_yaml = r#"

quickwit/quickwit-metastore/src/lib.rs

@@ -42,7 +42,6 @@ pub use metastore::file_backed::FileBackedMetastore;
 pub(crate) use metastore::index_metadata::serialize::{IndexMetadataV0_8, VersionedIndexMetadata};
 #[cfg(feature = "postgres")]
 pub use metastore::postgres::PostgresqlMetastore;
-pub use metastore::read_replica_routing_metastore::ReadReplicaRoutingMetastore;
 pub use metastore::read_service::{MetastoreReadService, MetastoreReadServiceClient};
 pub use metastore::{
     AddSourceRequestExt, CreateIndexRequestExt, CreateIndexResponseExt, IndexMetadata,

quickwit/quickwit-metastore/src/metastore/mod.rs

@@ -18,7 +18,6 @@ pub(crate) mod index_metadata;
 pub mod postgres;
 
 pub mod control_plane_metastore;
-pub mod read_replica_routing_metastore;
 pub mod read_service;
 
 use std::cmp::Ordering;

quickwit/quickwit-metastore/src/metastore/read_replica_routing_metastore.rs

@@ -93,7 +93,7 @@ use quickwit_jaeger::JaegerService;
 use quickwit_janitor::{JanitorService, start_janitor_service};
 use quickwit_metastore::{
     ControlPlaneMetastore, ListIndexesMetadataResponseExt, MetastoreReadServiceClient,
-    MetastoreResolver, ReadReplicaRoutingMetastore,
+    MetastoreResolver,
 };
 use quickwit_opentelemetry::otlp::{OtlpGrpcLogsService, OtlpGrpcTracesService};
 use quickwit_proto::control_plane::ControlPlaneServiceClient;
@@ -245,31 +245,31 @@ fn metastore_max_in_flight_requests(node_config: &NodeConfig, uri: &Uri) -> usiz
     }
 }
 
-/// Builds the read-only metastore client used by the search and analytics paths.
-///
-/// Stale-tolerant reads are routed to `metastore_read_replica` nodes when any are present in the
-/// cluster, falling back to `primary` otherwise.
-async fn build_search_metastore_client(
+async fn build_metastore_client(
     cluster: &Cluster,
+    service: QuickwitService,
     max_message_size: ByteSize,
-    primary: MetastoreServiceClient,
-) -> MetastoreReadServiceClient {
-    let read_replica_balance_channel =
-        balance_channel_for_service(cluster, QuickwitService::MetastoreReadReplica).await;
-    let read_replica_connections = read_replica_balance_channel.connection_keys_watcher();
-    let read_replica_metastore = MetastoreServiceClient::tower()
+) -> anyhow::Result<MetastoreServiceClient> {
+    info!(%service, "connecting to metastore service");
+
+    let balance_channel = balance_channel_for_service(cluster, service).await;
+
+    if !balance_channel
+        .wait_for(Duration::from_secs(300), |connections| {
+            !connections.is_empty()
+        })
+        .await
+    {
+        bail!("could not find any `{service}` node in the cluster");
+    }
+    Ok(MetastoreServiceClient::tower()
         .stack_layer(RetryLayer::new(RetryPolicy::from(RetryParams::standard())))
         .stack_layer(TimeoutLayer::new(GRPC_METASTORE_SERVICE_TIMEOUT))
         .stack_layer(METASTORE_GRPC_CLIENT_METRICS_LAYER.clone())
         .stack_layer(tower::limit::GlobalConcurrencyLimitLayer::new(
             get_metastore_client_max_concurrency(),
         ))
-        .build_from_balance_channel(read_replica_balance_channel, max_message_size, None);
-    Arc::new(ReadReplicaRoutingMetastore::new(
-        primary,
-        read_replica_metastore,
-        read_replica_connections,
-    ))
+        .build_from_balance_channel(balance_channel, max_message_size, None))
 }
 
 async fn balance_channel_for_service(
@@ -569,27 +569,12 @@ pub async fn serve_quickwit(
         if let Some(metastore_server) = &metastore_server_opt {
             metastore_server.clone()
         } else {
-            info!("connecting to metastore");
-
-            let balance_channel =
-                balance_channel_for_service(&cluster, QuickwitService::Metastore).await;
-
-            if !balance_channel
-                .wait_for(Duration::from_secs(300), |connections| {
-                    !connections.is_empty()
-                })
-                .await
-            {
-                bail!("could not find any metastore node in the cluster");
-            }
-            MetastoreServiceClient::tower()
-                .stack_layer(RetryLayer::new(RetryPolicy::from(RetryParams::standard())))
-                .stack_layer(TimeoutLayer::new(GRPC_METASTORE_SERVICE_TIMEOUT))
-                .stack_layer(METASTORE_GRPC_CLIENT_METRICS_LAYER.clone())
-                .stack_layer(tower::limit::GlobalConcurrencyLimitLayer::new(
-                    get_metastore_client_max_concurrency(),
-                ))
-                .build_from_balance_channel(balance_channel, grpc_config.max_message_size, None)
+            build_metastore_client(
+                &cluster,
+                QuickwitService::Metastore,
+                grpc_config.max_message_size,
+            )
+            .await?
         };
     // Instantiate a control plane server if the `control-plane` role is enabled on the node.
     // Otherwise, instantiate a control plane client.
@@ -742,15 +727,21 @@ pub async fn serve_quickwit(
         ))
     };
 
-    // Searchers and the DataFusion analytics path route their stale-tolerant, read-only metastore
-    // requests to `metastore_read_replica` nodes when any are present in the cluster, and fall back
-    // to the primary metastore otherwise.
-    let search_metastore_client = build_search_metastore_client(
-        &cluster,
-        grpc_config.max_message_size,
-        metastore_through_control_plane.clone(),
-    )
-    .await;
+    // Searchers and the DataFusion analytics path use the primary metastore by default. When
+    // `metastore_read_replica_uri` is configured, they require `metastore_read_replica` nodes and
+    // do not fall back to the primary if none are available.
+    let search_metastore_client: MetastoreReadServiceClient =
+        if node_config.metastore_read_replica_uri.is_some() {
+            let read_replica_metastore = build_metastore_client(
+                &cluster,
+                QuickwitService::MetastoreReadReplica,
+                grpc_config.max_message_size,
+            )
+            .await?;
+            Arc::new(read_replica_metastore)
+        } else {
+            Arc::new(metastore_through_control_plane.clone())
+        };
 
     let (search_job_placer, search_service, searcher_pool) = setup_searcher(
         &node_config,