diff --git a/config/quickwit.yaml b/config/quickwit.yaml index 6ad7cd69323..2cacb3bca4a 100644 --- a/config/quickwit.yaml +++ b/config/quickwit.yaml @@ -115,6 +115,13 @@ version: 0.8 # metastore_uri: s3://your-bucket/indexes # metastore_uri: postgres://username:password@host:port/db # +# 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. Searchers use those nodes +# only when `searcher.use_metastore_read_replica` is enabled. Defaults to unset. +# +# metastore_read_replica_uri: postgres://username:password@read-replica-host:port/db +# # When using a file-backed metastore, the state of the metastore will be cached forever. # If you are indexing and searching from different processes, it is possible to periodically # refresh the state of the metastore on the searcher using the `polling_interval` hashtag. @@ -180,6 +187,11 @@ indexer: # https://quickwit.io/docs/configuration/node-config#searcher-configuration # # searcher: +# # If true, routes read-only metastore requests from searchers, including +# # DataFusion when enabled, to nodes running the `metastore_read_replica` +# # service. Searchers require at least one `metastore_read_replica` node at +# # startup and do not fall back to the primary metastore. +# use_metastore_read_replica: false # fast_field_cache_capacity: 1G # split_footer_cache_capacity: 500M # partial_request_cache_capacity: 64M diff --git a/docs/configuration/node-config.md b/docs/configuration/node-config.md index 66da698665b..bba20e1f3f4 100644 --- a/docs/configuration/node-config.md +++ b/docs/configuration/node-config.md @@ -22,7 +22,7 @@ A commented example is available here: [quickwit.yaml](https://github.com/quickw | `version` | Config file version. `0.7` is the only available value with a retro compatibility on `0.5` and `0.4`. | | | | `cluster_id` | Unique identifier of the cluster the node will be joining. Clusters sharing the same network should use distinct cluster IDs.| `QW_CLUSTER_ID` | `quickwit-default-cluster` | | `node_id` | Unique identifier of the node. It must be distinct from the node IDs of its cluster peers. Defaults to the instance's short hostname if not set. | `QW_NODE_ID` | short hostname | -| `enabled_services` | Enabled services (control_plane, indexer, janitor, metastore, searcher) | `QW_ENABLED_SERVICES` | all services | +| `enabled_services` | Enabled services (control_plane, indexer, janitor, metastore, metastore_read_replica, searcher) | `QW_ENABLED_SERVICES` | all services except metastore_read_replica | | `listen_address` | The IP address or hostname that Quickwit service binds to for starting REST and GRPC server and connecting this node to other nodes. By default, Quickwit binds itself to 127.0.0.1 (localhost). This default is not valid when trying to form a cluster. | `QW_LISTEN_ADDRESS` | `127.0.0.1` | | `advertise_address` | IP address advertised by the node, i.e. the IP address that peer nodes should use to connect to the node for RPCs. | `QW_ADVERTISE_ADDRESS` | `listen_address` | | `gossip_listen_port` | The port which to listen for the Gossip cluster membership service (UDP). | `QW_GOSSIP_LISTEN_PORT` | `rest.listen_port` | @@ -30,6 +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. Searchers use those nodes only when `searcher.use_metastore_read_replica` is enabled. | `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` | @@ -285,6 +286,7 @@ This section contains the configuration options for a Searcher. | `max_num_concurrent_split_searches` | Maximum number of concurrent split search requests running on a Searcher. | `100` | | `split_cache` | Searcher split cache configuration options defined in the section below. Cache disabled if unspecified. | | | `request_timeout_secs` | The time before a search request is cancelled. This should match the timeout of the stack calling into quickwit if there is one set. | `30` | +| `use_metastore_read_replica` | If true, routes read-only metastore requests from searchers, including DataFusion when enabled, to nodes running the `metastore_read_replica` service. Searchers require at least one `metastore_read_replica` node at startup and do not fall back to the primary metastore. | `false` | ### Searcher split cache configuration @@ -301,6 +303,7 @@ Example: ```yaml searcher: + use_metastore_read_replica: false fast_field_cache_capacity: 1G split_footer_cache_capacity: 500M partial_request_cache_capacity: 64M diff --git a/quickwit/quickwit-cli/src/lib.rs b/quickwit/quickwit-cli/src/lib.rs index 1e3afba14cd..c3dfd952cc3 100644 --- a/quickwit/quickwit-cli/src/lib.rs +++ b/quickwit/quickwit-cli/src/lib.rs @@ -230,14 +230,21 @@ pub fn start_actor_runtimes( } /// Loads a node config located at `config_uri` with the default storage configuration. -async fn load_node_config(config_uri: &Uri) -> anyhow::Result { +async fn load_node_config( + config_uri: &Uri, + enabled_services: Option<&HashSet>, +) -> anyhow::Result { let config_content = load_file(&StorageResolver::unconfigured(), config_uri) .await .context("failed to load node config")?; let config_format = ConfigFormat::sniff_from_uri(config_uri)?; - let config = NodeConfig::load(config_format, config_content.as_slice()) - .await - .with_context(|| format!("failed to parse node config `{config_uri}`"))?; + let config = NodeConfig::load_with_enabled_services( + config_format, + config_content.as_slice(), + enabled_services, + ) + .await + .with_context(|| format!("failed to parse node config `{config_uri}`"))?; info!(config_uri=%config_uri, config=?config, "loaded node config"); Ok(config) } diff --git a/quickwit/quickwit-cli/src/service.rs b/quickwit/quickwit-cli/src/service.rs index 2b63e1617ac..04b5762db54 100644 --- a/quickwit/quickwit-cli/src/service.rs +++ b/quickwit/quickwit-cli/src/service.rs @@ -34,10 +34,10 @@ use crate::{config_cli_arg, get_resolvers, load_node_config, start_actor_runtime pub fn build_run_command() -> Command { Command::new("run") .about("Starts a Quickwit node.") - .long_about("Starts a Quickwit node with all services enabled by default: `indexer`, `searcher`, `metastore`, `control-plane`, and `janitor`.") + .long_about("Starts a Quickwit node with the default services enabled: `indexer`, `searcher`, `metastore`, `control-plane`, and `janitor`.") .arg(config_cli_arg()) .args(&[ - arg!(--"service" "Services (`indexer`, `searcher`, `metastore`, `control-plane`, or `janitor`) to run. If unspecified, all the supported services are started.") + arg!(--"service" "Services (`indexer`, `searcher`, `metastore`, `metastore-read-replica`, `control-plane`, or `janitor`) to run. If unspecified, services from the config are used.") .action(ArgAction::Append) .required(false), ]) @@ -117,15 +117,13 @@ impl RunCliCommand { debug!(args = ?self, "run-service"); let version_text = BuildInfo::get_version_text(); info!("quickwit version: {version_text}"); - let mut node_config = load_node_config(&self.config_uri).await?; - let (storage_resolver, metastore_resolver) = - get_resolvers(&node_config.storage_configs, &node_config.metastore_configs); - crate::busy_detector::set_enabled(true); - if let Some(services) = &self.services { info!(services = %services.iter().join(", "), "setting services from override"); - node_config.enabled_services.clone_from(services); } + let node_config = load_node_config(&self.config_uri, self.services.as_ref()).await?; + let (storage_resolver, metastore_resolver) = + get_resolvers(&node_config.storage_configs, &node_config.metastore_configs); + crate::busy_detector::set_enabled(true); // TODO move in serve quickwit? let runtimes_config = RuntimesConfig::default(); start_actor_runtimes(runtimes_config, &node_config.enabled_services)?; @@ -152,9 +150,35 @@ impl RunCliCommand { #[cfg(test)] mod tests { + use tempfile::tempdir; + use super::*; use crate::cli::{CliCommand, build_cli}; + #[tokio::test] + async fn test_service_override_is_applied_before_config_validation() -> anyhow::Result<()> { + let temp_dir = tempdir()?; + let config_path = temp_dir.path().join("quickwit.yaml"); + let config = format!( + r#" + version: 0.8 + data_dir: {} + searcher: + use_metastore_read_replica: true + "#, + temp_dir.path().display() + ); + std::fs::write(&config_path, config)?; + let config_uri = Uri::from_str(&format!("file://{}", config_path.display()))?; + let enabled_services = HashSet::from([QuickwitService::Searcher]); + + let node_config = load_node_config(&config_uri, Some(&enabled_services)).await?; + + assert_eq!(node_config.enabled_services, enabled_services); + assert!(node_config.searcher_config.use_metastore_read_replica); + Ok(()) + } + #[test] fn test_parse_service_run_args_all_services() -> anyhow::Result<()> { let command = build_cli().no_binary_name(true); diff --git a/quickwit/quickwit-cli/src/tool.rs b/quickwit/quickwit-cli/src/tool.rs index d64d9ca3e02..dbbc264c8b8 100644 --- a/quickwit/quickwit-cli/src/tool.rs +++ b/quickwit/quickwit-cli/src/tool.rs @@ -402,7 +402,7 @@ pub async fn local_ingest_docs_cli(args: LocalIngestDocsArgs) -> anyhow::Result< debug!(args=?args, "local-ingest-docs"); println!("❯ Ingesting documents locally..."); - let config = load_node_config(&args.config_uri).await?; + let config = load_node_config(&args.config_uri, None).await?; let (storage_resolver, metastore_resolver) = get_resolvers(&config.storage_configs, &config.metastore_configs); let mut metastore = metastore_resolver.resolve(&config.metastore_uri).await?; @@ -536,7 +536,7 @@ pub async fn local_ingest_docs_cli(args: LocalIngestDocsArgs) -> anyhow::Result< pub async fn local_search_cli(args: LocalSearchArgs) -> anyhow::Result<()> { debug!(args=?args, "local-search"); println!("❯ Searching directly on the index storage (without calling REST API)..."); - let config = load_node_config(&args.config_uri).await?; + let config = load_node_config(&args.config_uri, None).await?; let (storage_resolver, metastore_resolver) = get_resolvers(&config.storage_configs, &config.metastore_configs); let metastore: MetastoreServiceClient = @@ -574,7 +574,7 @@ pub async fn local_search_cli(args: LocalSearchArgs) -> anyhow::Result<()> { pub async fn merge_cli(args: MergeArgs) -> anyhow::Result<()> { debug!(args=?args, "run-merge-operations"); println!("❯ Merging splits locally..."); - let config = load_node_config(&args.config_uri).await?; + let config = load_node_config(&args.config_uri, None).await?; let (storage_resolver, metastore_resolver) = get_resolvers(&config.storage_configs, &config.metastore_configs); let mut metastore = metastore_resolver.resolve(&config.metastore_uri).await?; @@ -662,7 +662,7 @@ pub async fn garbage_collect_index_cli(args: GarbageCollectIndexArgs) -> anyhow: debug!(args=?args, "garbage-collect-index"); println!("❯ Garbage collecting index..."); - let config = load_node_config(&args.config_uri).await?; + let config = load_node_config(&args.config_uri, None).await?; let (storage_resolver, metastore_resolver) = get_resolvers(&config.storage_configs, &config.metastore_configs); let metastore = metastore_resolver.resolve(&config.metastore_uri).await?; @@ -792,7 +792,7 @@ async fn extract_split_cli(args: ExtractSplitArgs) -> anyhow::Result<()> { debug!(args=?args, "extract-split"); println!("❯ Extracting split..."); - let config = load_node_config(&args.config_uri).await?; + let config = load_node_config(&args.config_uri, None).await?; let (storage_resolver, metastore_resolver) = get_resolvers(&config.storage_configs, &config.metastore_configs); let metastore = metastore_resolver.resolve(&config.metastore_uri).await?; diff --git a/quickwit/quickwit-cli/tests/helpers.rs b/quickwit/quickwit-cli/tests/helpers.rs index b9b8207aa1a..6bbcbdfbe37 100644 --- a/quickwit/quickwit-cli/tests/helpers.rs +++ b/quickwit/quickwit-cli/tests/helpers.rs @@ -157,7 +157,7 @@ impl TestEnv { pub async fn start_server(&self) -> anyhow::Result<()> { let run_command = RunCliCommand { config_uri: self.resource_files.config.clone(), - services: Some(QuickwitService::supported_services()), + services: Some(QuickwitService::default_services()), }; let server_handle = tokio::spawn(async move { if let Err(error) = run_command diff --git a/quickwit/quickwit-config/Cargo.toml b/quickwit/quickwit-config/Cargo.toml index aebea5e9c31..44cae30260c 100644 --- a/quickwit/quickwit-config/Cargo.toml +++ b/quickwit/quickwit-config/Cargo.toml @@ -46,5 +46,5 @@ quickwit-proto = { workspace = true, features = ["testsuite"] } [features] metrics = [] -testsuite = [] +testsuite = ["quickwit-proto/testsuite"] vrl = ["dep:vrl"] diff --git a/quickwit/quickwit-config/resources/tests/node_config/quickwit.json b/quickwit/quickwit-config/resources/tests/node_config/quickwit.json index a60becaee7e..5871cf972ee 100644 --- a/quickwit/quickwit-config/resources/tests/node_config/quickwit.json +++ b/quickwit/quickwit-config/resources/tests/node_config/quickwit.json @@ -18,6 +18,7 @@ ], "data_dir": "/opt/quickwit/data", "metastore_uri": "postgres://username:password@host:port/db", + "metastore_read_replica_uri": "postgres://username:replica-password@replica-host:port/db", "default_index_root_uri": "s3://quickwit-indexes", "rest": { "listen_port": 1111, @@ -81,6 +82,7 @@ "replication_factor": 2 }, "searcher": { + "use_metastore_read_replica": true, "aggregation_memory_limit": "1G", "aggregation_bucket_limit": 500000, "fast_field_cache_capacity": "10G", diff --git a/quickwit/quickwit-config/resources/tests/node_config/quickwit.toml b/quickwit/quickwit-config/resources/tests/node_config/quickwit.toml index 665d0f2a624..685a70e355f 100644 --- a/quickwit/quickwit-config/resources/tests/node_config/quickwit.toml +++ b/quickwit/quickwit-config/resources/tests/node_config/quickwit.toml @@ -11,6 +11,7 @@ grpc_listen_port = 3333 peer_seeds = [ "quickwit-searcher-0.local", "quickwit-searcher-1.local" ] data_dir = "/opt/quickwit/data" metastore_uri = "postgres://username:password@host:port/db" +metastore_read_replica_uri = "postgres://username:replica-password@replica-host:port/db" default_index_root_uri = "s3://quickwit-indexes" [rest] @@ -71,6 +72,7 @@ parquet_merge_use_streaming_engine = true replication_factor = 2 [searcher] +use_metastore_read_replica = true aggregation_memory_limit = "1G" aggregation_bucket_limit = 500_000 fast_field_cache_capacity = "10G" diff --git a/quickwit/quickwit-config/resources/tests/node_config/quickwit.yaml b/quickwit/quickwit-config/resources/tests/node_config/quickwit.yaml index 46f988e447c..7b22a93184a 100644 --- a/quickwit/quickwit-config/resources/tests/node_config/quickwit.yaml +++ b/quickwit/quickwit-config/resources/tests/node_config/quickwit.yaml @@ -15,6 +15,7 @@ peer_seeds: - quickwit-searcher-1.local data_dir: /opt/quickwit/data metastore_uri: postgres://username:password@host:port/db +metastore_read_replica_uri: postgres://username:replica-password@replica-host:port/db default_index_root_uri: s3://quickwit-indexes rest: @@ -73,6 +74,7 @@ ingest_api: replication_factor: 2 searcher: + use_metastore_read_replica: true aggregation_memory_limit: 1G aggregation_bucket_limit: 500000 fast_field_cache_capacity: 10G diff --git a/quickwit/quickwit-config/src/node_config/mod.rs b/quickwit/quickwit-config/src/node_config/mod.rs index 678e2fb2e03..f3c771eec4c 100644 --- a/quickwit/quickwit-config/src/node_config/mod.rs +++ b/quickwit/quickwit-config/src/node_config/mod.rs @@ -419,6 +419,10 @@ pub struct SearcherConfig { #[serde(default)] #[serde(skip_serializing_if = "Option::is_none")] pub storage_timeout_policy: Option, + /// Routes read-only metastore requests from searchers, including DataFusion when enabled, to + /// nodes running the `metastore_read_replica` service. + #[serde(default)] + pub use_metastore_read_replica: bool, pub warmup_memory_budget: ByteSize, pub warmup_single_split_initial_allocation: ByteSize, /// Lambda configuration for serverless leaf search execution. @@ -642,6 +646,7 @@ impl Default for SearcherConfig { request_timeout_secs: Self::default_request_timeout_secs(), leaf_request_timeout_secs: Self::default_request_timeout_secs(), storage_timeout_policy: None, + use_metastore_read_replica: false, warmup_memory_budget: ByteSize::gb(100), warmup_single_split_initial_allocation: ByteSize::mb(300), lambda: None, @@ -888,6 +893,10 @@ pub struct NodeConfig { pub peer_seeds: Vec, pub data_dir_path: PathBuf, pub metastore_uri: Uri, + /// Optional PostgreSQL read replica URI. It is used as the connection URI by nodes running the + /// [`QuickwitService::MetastoreReadReplica`] role. + #[serde(skip_serializing_if = "Option::is_none")] + pub metastore_read_replica_uri: Option, pub default_index_root_uri: Uri, pub rest_config: RestConfig, #[serde(skip_serializing_if = "Option::is_none")] @@ -911,8 +920,22 @@ impl NodeConfig { /// Parses and validates a [`NodeConfig`] from a given URI and config content. pub async fn load(config_format: ConfigFormat, config_content: &[u8]) -> anyhow::Result { + Self::load_with_enabled_services(config_format, config_content, None).await + } + + /// Parses and validates a [`NodeConfig`] after overriding its enabled services. + /// + /// The override takes precedence over both the config file and `QW_ENABLED_SERVICES` and is + /// applied before service-dependent validation. + pub async fn load_with_enabled_services( + config_format: ConfigFormat, + config_content: &[u8], + enabled_services: Option<&HashSet>, + ) -> anyhow::Result { let env_vars = env::vars().collect::>(); - let config = load_node_config_with_env(config_format, config_content, &env_vars).await?; + let config = + load_node_config_with_env(config_format, config_content, &env_vars, enabled_services) + .await?; if !config.data_dir_path.try_exists()? { bail!( "data dir `{}` does not exist", @@ -951,6 +974,9 @@ impl NodeConfig { pub fn redact(&mut self) { self.metastore_configs.redact(); self.metastore_uri.redact(); + if let Some(metastore_read_replica_uri) = &mut self.metastore_read_replica_uri { + metastore_read_replica_uri.redact(); + } self.storage_configs.redact(); } @@ -983,6 +1009,26 @@ mod tests { use super::*; use crate::IndexerConfig; + #[test] + fn test_node_config_redact_metastore_uris() { + let mut config = NodeConfig::for_test(); + config.metastore_uri = Uri::for_test("postgresql://username:password@host:5432/db"); + config.metastore_read_replica_uri = Some(Uri::for_test( + "postgresql://replica-user:replica-password@replica-host:5432/db", + )); + + config.redact(); + + assert_eq!( + config.metastore_uri, + "postgresql://username:***redacted***@host:5432/db" + ); + assert_eq!( + config.metastore_read_replica_uri.unwrap(), + "postgresql://replica-user:***redacted***@replica-host:5432/db" + ); + } + #[test] fn test_indexer_config_serialization() { { diff --git a/quickwit/quickwit-config/src/node_config/serialize.rs b/quickwit/quickwit-config/src/node_config/serialize.rs index f026a07fa96..1656f7f5713 100644 --- a/quickwit/quickwit-config/src/node_config/serialize.rs +++ b/quickwit/quickwit-config/src/node_config/serialize.rs @@ -88,12 +88,10 @@ impl FromStr for List { } fn default_enabled_services() -> ConfigValue { - // The compactor is excluded by default — it only runs in standalone mode, - // which an operator must explicitly opt into via `enable_standalone_compactors`. + // Opt-in standalone services are excluded from all-in-one nodes by default. ConfigValue::with_default(List( - QuickwitService::supported_services() + QuickwitService::default_services() .into_iter() - .filter(|service| *service != QuickwitService::Compactor) .map(|service| service.to_string()) .collect(), )) @@ -134,6 +132,10 @@ fn default_metastore_uri(data_dir_uri: &Uri) -> Uri { data_dir_uri.join("indexes#polling_interval=30s").expect("Failed to create default metastore URI. This should never happen! Please, report on https://github.com/quickwit-oss/quickwit/issues.") } +fn default_metastore_read_replica_uri() -> ConfigValue { + ConfigValue::none() +} + // See comment above. fn default_index_root_uri(data_dir_uri: &Uri) -> Uri { data_dir_uri.join("indexes").expect("Failed to create default index root URI. This should never happen! Please, report on https://github.com/quickwit-oss/quickwit/issues.") @@ -143,12 +145,15 @@ pub async fn load_node_config_with_env( config_format: ConfigFormat, config_content: &[u8], env_vars: &HashMap, + enabled_services: Option<&HashSet>, ) -> anyhow::Result { let rendered_config_content = render_config(config_content)?; let versioned_node_config: VersionedNodeConfig = config_format.parse(rendered_config_content.as_bytes())?; let node_config_builder: NodeConfigBuilder = versioned_node_config.into(); - let config = node_config_builder.build_and_validate(env_vars).await?; + let config = node_config_builder + .build_and_validate(env_vars, enabled_services) + .await?; Ok(config) } @@ -195,6 +200,8 @@ struct NodeConfigBuilder { #[serde(default = "default_data_dir_uri")] data_dir_uri: ConfigValue, metastore_uri: ConfigValue, + #[serde(default = "default_metastore_read_replica_uri")] + metastore_read_replica_uri: ConfigValue, default_index_root_uri: ConfigValue, #[serde(default)] enable_standalone_compactors: ConfigValue, @@ -234,6 +241,7 @@ impl NodeConfigBuilder { pub async fn build_and_validate( mut self, env_vars: &HashMap, + enabled_services: Option<&HashSet>, ) -> anyhow::Result { let node_id = self .node_id @@ -243,13 +251,17 @@ impl NodeConfigBuilder { let enable_standalone_compactors = self.enable_standalone_compactors.resolve(env_vars)?; - let enabled_services: HashSet = self - .enabled_services - .resolve(env_vars)? - .0 - .into_iter() - .map(|service| service.parse()) - .collect::>()?; + let resolved_enabled_services: HashSet = + if let Some(enabled_services) = enabled_services { + enabled_services.clone() + } else { + self.enabled_services + .resolve(env_vars)? + .0 + .into_iter() + .map(|service| service.parse()) + .collect::>()? + }; let listen_address = self.listen_address.resolve(env_vars)?; let listen_host = listen_address.parse::()?; @@ -312,6 +324,9 @@ impl NodeConfigBuilder { .resolve_optional(env_vars)? .unwrap_or_else(|| default_metastore_uri(&data_dir_uri)); + let metastore_read_replica_uri = + self.metastore_read_replica_uri.resolve_optional(env_vars)?; + let default_index_root_uri = self .default_index_root_uri .resolve_optional(env_vars)? @@ -332,7 +347,7 @@ impl NodeConfigBuilder { cluster_id: self.cluster_id.resolve(env_vars)?, node_id, availability_zone, - enabled_services, + enabled_services: resolved_enabled_services, gossip_listen_addr, grpc_listen_addr, gossip_advertise_addr, @@ -341,6 +356,7 @@ impl NodeConfigBuilder { peer_seeds: self.peer_seeds.resolve(env_vars)?.0, data_dir_path, metastore_uri, + metastore_read_replica_uri, default_index_root_uri, rest_config, health_config, @@ -370,6 +386,7 @@ fn validate(node_config: &NodeConfig) -> anyhow::Result<()> { if node_config.peer_seeds.is_empty() { warn!("peer seeds are empty"); } + validate_metastore_read_replica(node_config)?; if node_config.is_service_enabled(QuickwitService::Compactor) && !node_config.enable_standalone_compactors { @@ -383,6 +400,37 @@ fn validate(node_config: &NodeConfig) -> anyhow::Result<()> { Ok(()) } +/// Validates the configuration of the [`QuickwitService::MetastoreReadReplica`] role and searcher +/// read-replica routing. +/// +/// The [`QuickwitService::MetastoreReadReplica`] role serves the same gRPC service as a read-only +/// metastore, so it must run standalone and requires `metastore_read_replica_uri` to connect to. +fn validate_metastore_read_replica(node_config: &NodeConfig) -> anyhow::Result<()> { + let read_replica_enabled = + node_config.is_service_enabled(QuickwitService::MetastoreReadReplica); + if read_replica_enabled { + ensure!( + node_config.enabled_services.len() == 1, + "the `metastore_read_replica` service must run standalone and cannot be combined with \ + any other service" + ); + ensure!( + node_config.metastore_read_replica_uri.is_some(), + "the `metastore_read_replica` service requires `metastore_read_replica_uri` to be set" + ); + } + let searcher_uses_read_replica = node_config.is_service_enabled(QuickwitService::Searcher) + && node_config.searcher_config.use_metastore_read_replica; + // Avoid a deadlock where the searcher waits for a READY read replica, while the read + // replica waits for this primary metastore to become READY. + ensure!( + !(searcher_uses_read_replica && node_config.is_service_enabled(QuickwitService::Metastore)), + "`searcher.use_metastore_read_replica` cannot be enabled on a node running the \ + `metastore` service" + ); + Ok(()) +} + /// A list of all the known disk budgets /// /// External disk usage and unbounded disk usages, e.g the indexing workbench @@ -451,6 +499,7 @@ impl Default for NodeConfigBuilder { peer_seeds: ConfigValue::with_default(List::default()), data_dir_uri: default_data_dir_uri(), metastore_uri: ConfigValue::none(), + metastore_read_replica_uri: default_metastore_read_replica_uri(), default_index_root_uri: ConfigValue::none(), enable_standalone_compactors: Default::default(), rest_config_builder: RestConfigBuilder::default(), @@ -556,7 +605,7 @@ pub fn node_config_for_tests_from_ports( grpc_listen_port: u16, ) -> NodeConfig { let node_id = NodeId::from_str(&default_node_id().unwrap()); - let enabled_services = QuickwitService::supported_services(); + let enabled_services = QuickwitService::default_services(); let availability_zone = Some(String::from("az-1")); let listen_address = Host::default(); let rest_listen_addr = listen_address @@ -600,6 +649,7 @@ pub fn node_config_for_tests_from_ports( peer_seeds: Vec::new(), data_dir_path, metastore_uri, + metastore_read_replica_uri: None, default_index_root_uri, rest_config, health_config: None, @@ -642,7 +692,8 @@ mod tests { get_config_filepath(&format!("quickwit.{config_format:?}").to_lowercase()); let file = std::fs::read_to_string(&config_filepath).unwrap(); let env_vars = HashMap::default(); - let config = load_node_config_with_env(config_format, file.as_bytes(), &env_vars).await?; + let config = + load_node_config_with_env(config_format, file.as_bytes(), &env_vars, None).await?; assert_eq!(config.cluster_id, "quickwit-cluster"); assert_eq!(config.enabled_services.len(), 2); @@ -738,6 +789,10 @@ mod tests { config.metastore_uri, "postgresql://username:password@host:port/db" ); + assert_eq!( + config.metastore_read_replica_uri.unwrap(), + "postgresql://username:replica-password@replica-host:port/db" + ); assert_eq!(config.default_index_root_uri, "s3://quickwit-indexes"); let azure_storage_config = config.storage_configs.find_azure().unwrap(); @@ -817,6 +872,7 @@ mod tests { timeout_millis: 2_000, max_num_retries: 2 }), + use_metastore_read_replica: true, warmup_memory_budget: ByteSize::gb(100), warmup_single_split_initial_allocation: ByteSize::mb(300), lambda: Some(LambdaConfig { @@ -873,6 +929,7 @@ mod tests { ConfigFormat::Yaml, config_str.as_bytes(), &Default::default(), + None, ) .await .unwrap_err(); @@ -889,19 +946,14 @@ mod tests { ConfigFormat::Yaml, config_yaml.as_bytes(), &Default::default(), + None, ) .await .unwrap(); assert_eq!(config.cluster_id, DEFAULT_CLUSTER_ID); assert_eq!(config.node_id.as_str(), get_short_hostname().unwrap()); assert_eq!(config.availability_zone, None); - assert_eq!( - config.enabled_services, - QuickwitService::supported_services() - .into_iter() - .filter(|service| *service != QuickwitService::Compactor) - .collect::>(), - ); + assert_eq!(config.enabled_services, QuickwitService::default_services()); assert_eq!( config.rest_config.listen_addr, SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 7280) @@ -926,6 +978,7 @@ mod tests { env::current_dir().unwrap().display() ) ); + assert!(config.metastore_read_replica_uri.is_none()); assert_eq!( config.default_index_root_uri, format!( @@ -970,7 +1023,7 @@ mod tests { "s3://quickwit-indexes/prod".to_string(), ); let config = - load_node_config_with_env(ConfigFormat::Yaml, config_yaml.as_bytes(), &env_vars) + load_node_config_with_env(ConfigFormat::Yaml, config_yaml.as_bytes(), &env_vars, None) .await .unwrap(); assert_eq!(config.cluster_id, "test-cluster"); @@ -1041,6 +1094,7 @@ mod tests { ConfigFormat::Yaml, config_yaml.as_bytes(), &Default::default(), + None, ) .await .unwrap(); @@ -1063,6 +1117,7 @@ mod tests { ConfigFormat::Yaml, config_yaml.as_bytes(), &Default::default(), + None, ) .await .unwrap(); @@ -1087,11 +1142,115 @@ mod tests { "QW_DATA_DIR".to_string(), data_dir_path.to_string_lossy().to_string(), ); - load_node_config_with_env(ConfigFormat::Toml, file_content.as_bytes(), &env_vars) + load_node_config_with_env(ConfigFormat::Toml, file_content.as_bytes(), &env_vars, None) .await .unwrap(); } + #[tokio::test] + async fn test_metastore_read_replica_role_without_uri_is_rejected() { + let config_yaml = r#" + version: 0.8 + node_id: test-node + enabled_services: + - metastore_read_replica + "#; + let error = load_node_config_with_env( + ConfigFormat::Yaml, + config_yaml.as_bytes(), + &Default::default(), + None, + ) + .await + .unwrap_err(); + assert!( + error + .to_string() + .contains("requires `metastore_read_replica_uri`") + ); + } + + #[tokio::test] + async fn test_metastore_read_replica_role_must_run_standalone() { + let config_yaml = r#" + version: 0.8 + node_id: test-node + enabled_services: + - metastore_read_replica + - searcher + metastore_read_replica_uri: postgres://user:pass@host:5432/db + "#; + let error = load_node_config_with_env( + ConfigFormat::Yaml, + config_yaml.as_bytes(), + &Default::default(), + None, + ) + .await + .unwrap_err(); + assert!(error.to_string().contains("must run standalone")); + } + + #[tokio::test] + async fn test_searcher_read_replica_cannot_run_with_metastore() { + let error = NodeConfigBuilder { + enabled_services: ConfigValue::for_test(List(vec![ + "metastore".to_string(), + "searcher".to_string(), + ])), + searcher_config: SearcherConfig { + use_metastore_read_replica: true, + ..Default::default() + }, + ..Default::default() + } + .build_and_validate(&HashMap::new(), None) + .await + .unwrap_err(); + assert!( + error + .to_string() + .contains("cannot be enabled on a node running the `metastore` service") + ); + } + + #[tokio::test] + async fn test_searcher_read_replica_is_allowed_on_dedicated_searcher() { + let node_config = NodeConfigBuilder { + enabled_services: ConfigValue::for_test(List(vec!["searcher".to_string()])), + searcher_config: SearcherConfig { + use_metastore_read_replica: true, + ..Default::default() + }, + ..Default::default() + } + .build_and_validate(&HashMap::new(), None) + .await + .unwrap(); + assert!(node_config.searcher_config.use_metastore_read_replica); + } + + #[tokio::test] + async fn test_enabled_services_validates_metastore_read_replica_uri() { + let config_yaml = "version: 0.8"; + let enabled_services = HashSet::from([QuickwitService::MetastoreReadReplica]); + + let error = load_node_config_with_env( + ConfigFormat::Yaml, + config_yaml.as_bytes(), + &HashMap::new(), + Some(&enabled_services), + ) + .await + .unwrap_err(); + + assert!( + error + .to_string() + .contains("requires `metastore_read_replica_uri`") + ); + } + #[tokio::test] async fn test_compactor_service_requires_standalone_flag() { // Compactor service enabled while `enable_standalone_compactors` is off is an @@ -1104,7 +1263,7 @@ mod tests { ])), ..Default::default() } - .build_and_validate(&HashMap::new()) + .build_and_validate(&HashMap::new(), None) .await .unwrap_err(); assert!( @@ -1129,7 +1288,7 @@ mod tests { enable_standalone_compactors: ConfigValue::for_test(true), ..Default::default() } - .build_and_validate(&HashMap::new()) + .build_and_validate(&HashMap::new(), None) .await .unwrap(); assert!(node_config.is_service_enabled(QuickwitService::Compactor)); @@ -1141,7 +1300,7 @@ mod tests { let node_config = NodeConfigBuilder { ..Default::default() } - .build_and_validate(&HashMap::new()) + .build_and_validate(&HashMap::new(), None) .await .unwrap(); assert!(node_config.peer_seed_addrs().await.unwrap().is_empty()); @@ -1161,7 +1320,7 @@ mod tests { ])), ..Default::default() } - .build_and_validate(&HashMap::new()) + .build_and_validate(&HashMap::new(), None) .await .unwrap(); assert_eq!( @@ -1184,7 +1343,7 @@ mod tests { listen_address: default_listen_address(), ..Default::default() } - .build_and_validate(&HashMap::new()) + .build_and_validate(&HashMap::new(), None) .await .unwrap(); assert_eq!( @@ -1203,7 +1362,7 @@ mod tests { }, ..Default::default() } - .build_and_validate(&HashMap::new()) + .build_and_validate(&HashMap::new(), None) .await .unwrap(); assert_eq!( @@ -1224,7 +1383,7 @@ mod tests { }, ..Default::default() } - .build_and_validate(&HashMap::new()) + .build_and_validate(&HashMap::new(), None) .await .unwrap(); assert_eq!( @@ -1248,7 +1407,7 @@ mod tests { }, ..Default::default() } - .build_and_validate(&HashMap::new()) + .build_and_validate(&HashMap::new(), None) .await .unwrap(); assert_eq!( @@ -1281,7 +1440,8 @@ mod tests { load_node_config_with_env( ConfigFormat::Yaml, config_yaml.as_bytes(), - &Default::default() + &Default::default(), + None, ) .await .is_err() @@ -1298,7 +1458,8 @@ mod tests { load_node_config_with_env( ConfigFormat::Yaml, config_yaml.as_bytes(), - &Default::default() + &Default::default(), + None, ) .await .is_err() @@ -1317,6 +1478,7 @@ mod tests { ConfigFormat::Yaml, config_yaml.as_bytes(), &HashMap::default(), + None, ) .await .unwrap(); @@ -1331,6 +1493,7 @@ mod tests { ConfigFormat::Yaml, config_yaml.as_bytes(), &HashMap::default(), + None, ) .await .unwrap(); @@ -1345,6 +1508,7 @@ mod tests { ConfigFormat::Yaml, config_yaml.as_bytes(), &HashMap::default(), + None, ) .await .unwrap_err(); @@ -1364,6 +1528,7 @@ mod tests { ConfigFormat::Yaml, config_yaml.as_bytes(), &HashMap::default(), + None, ) .await .unwrap_err(); @@ -1400,6 +1565,7 @@ mod tests { ConfigFormat::Yaml, rest_config_yaml.as_bytes(), &Default::default(), + None, ) .await .expect("Deserialize rest config"); @@ -1418,6 +1584,7 @@ mod tests { ConfigFormat::Yaml, rest_config_yaml.as_bytes(), &Default::default(), + None, ) .await .expect("Deserialize rest config"); @@ -1435,6 +1602,7 @@ mod tests { ConfigFormat::Yaml, rest_config_yaml.as_bytes(), &Default::default(), + None, ) .await .expect("Deserialize rest config"); @@ -1456,6 +1624,7 @@ mod tests { ConfigFormat::Yaml, rest_config_yaml.as_bytes(), &Default::default(), + None, ) .await .expect("Deserialize rest config"); @@ -1475,6 +1644,7 @@ mod tests { ConfigFormat::Yaml, rest_config_yaml.as_bytes(), &Default::default(), + None, ) .await .expect("Deserialize rest config"); @@ -1495,6 +1665,7 @@ mod tests { ConfigFormat::Yaml, rest_config_yaml.as_bytes(), &Default::default(), + None, ) .await .expect_err("Config should not allow empty origins."); @@ -1509,6 +1680,7 @@ mod tests { ConfigFormat::Yaml, rest_config_yaml.as_bytes(), &Default::default(), + None, ) .await .expect_err("Config should not allow empty origins."); @@ -1539,6 +1711,7 @@ mod tests { ConfigFormat::Yaml, node_config_yaml.as_bytes(), &Default::default(), + None, ) .await .unwrap_err() @@ -1557,7 +1730,7 @@ mod tests { "true".to_string(), )]); let config = - load_node_config_with_env(ConfigFormat::Yaml, config_yaml.as_bytes(), &env_vars) + load_node_config_with_env(ConfigFormat::Yaml, config_yaml.as_bytes(), &env_vars, None) .await .unwrap(); assert!(config.enabled_services.contains(&QuickwitService::Indexer)); @@ -1579,7 +1752,7 @@ mod tests { enabled_services: [indexer] "#; let config = - load_node_config_with_env(ConfigFormat::Yaml, config_yaml.as_bytes(), &env_vars) + load_node_config_with_env(ConfigFormat::Yaml, config_yaml.as_bytes(), &env_vars, None) .await .unwrap(); assert!(config.enabled_services.contains(&QuickwitService::Indexer)); diff --git a/quickwit/quickwit-config/src/qw_env_vars.rs b/quickwit/quickwit-config/src/qw_env_vars.rs index 7ec5bfd360c..00d83636309 100644 --- a/quickwit/quickwit-config/src/qw_env_vars.rs +++ b/quickwit/quickwit-config/src/qw_env_vars.rs @@ -55,6 +55,7 @@ qw_env_vars!( QW_HEALTH_LISTEN_PORT, QW_LISTEN_ADDRESS, QW_METASTORE_URI, + QW_METASTORE_READ_REPLICA_URI, QW_NODE_ID, QW_PEER_SEEDS, QW_REST_LISTEN_PORT @@ -72,7 +73,13 @@ mod tests { assert_eq!(QW_CLUSTER_ID, 3); assert_eq!(QW_ENV_VARS.get(&QW_CLUSTER_ID).unwrap(), &"QW_CLUSTER_ID"); + assert_eq!( + QW_ENV_VARS.get(&QW_METASTORE_READ_REPLICA_URI).unwrap(), + &"QW_METASTORE_READ_REPLICA_URI" + ); + assert_eq!(QW_METASTORE_READ_REPLICA_URI, 14); + assert_eq!(QW_ENV_VARS.get(&QW_NODE_ID).unwrap(), &"QW_NODE_ID"); - assert_eq!(QW_NODE_ID, 14); + assert_eq!(QW_NODE_ID, 15); } } diff --git a/quickwit/quickwit-config/src/service.rs b/quickwit/quickwit-config/src/service.rs index 006d48cd7fc..ba82df06668 100644 --- a/quickwit/quickwit-config/src/service.rs +++ b/quickwit/quickwit-config/src/service.rs @@ -29,6 +29,10 @@ pub enum QuickwitService { Indexer, Janitor, Metastore, + /// A read-only metastore node backed by a PostgreSQL read replica. It serves the same gRPC + /// metastore service as [`QuickwitService::Metastore`] but over a read-only connection, and is + /// discovered separately so that read-only callers (e.g. searchers) can route reads to it. + MetastoreReadReplica, Searcher, } @@ -47,13 +51,33 @@ impl QuickwitService { QuickwitService::Indexer => "indexer", QuickwitService::Janitor => "janitor", QuickwitService::Metastore => "metastore", + QuickwitService::MetastoreReadReplica => "metastore_read_replica", QuickwitService::Searcher => "searcher", } } + /// Returns every service the binary knows how to run. pub fn supported_services() -> HashSet { all::().collect() } + + /// Returns the services enabled on a node when none are explicitly configured. + /// + /// This is every supported service except the opt-in standalone services. + /// + /// [`QuickwitService::MetastoreReadReplica`] requires a dedicated read replica URI and is meant + /// to be deployed as its own set of nodes. [`QuickwitService::Compactor`] requires an explicit + /// standalone compactor opt-in. + pub fn default_services() -> HashSet { + all::() + .filter(|service| { + !matches!( + *service, + QuickwitService::Compactor | QuickwitService::MetastoreReadReplica + ) + }) + .collect() + } } impl Display for QuickwitService { @@ -72,6 +96,9 @@ impl FromStr for QuickwitService { "indexer" => Ok(QuickwitService::Indexer), "janitor" => Ok(QuickwitService::Janitor), "metastore" => Ok(QuickwitService::Metastore), + "metastore-read-replica" | "metastore_read_replica" => { + Ok(QuickwitService::MetastoreReadReplica) + } "searcher" => Ok(QuickwitService::Searcher), _ => { bail!( diff --git a/quickwit/quickwit-datafusion/src/sources/metrics/index_resolver.rs b/quickwit/quickwit-datafusion/src/sources/metrics/index_resolver.rs index a8f96fec60b..81ef3471856 100644 --- a/quickwit/quickwit-datafusion/src/sources/metrics/index_resolver.rs +++ b/quickwit/quickwit-datafusion/src/sources/metrics/index_resolver.rs @@ -85,7 +85,6 @@ impl MetricsIndexResolver for MetastoreIndexResolver { let response = self .metastore - .clone() .index_metadata(IndexMetadataRequest::for_index_id(index_name.to_string())) .await .map_err(|err| datafusion::error::DataFusionError::External(Box::new(err)))?; @@ -111,7 +110,6 @@ impl MetricsIndexResolver for MetastoreIndexResolver { async fn list_index_names(&self) -> DFResult> { let response = self .metastore - .clone() .list_indexes_metadata(ListIndexesMetadataRequest::all()) .await .map_err(|err| datafusion::error::DataFusionError::External(Box::new(err)))?; diff --git a/quickwit/quickwit-datafusion/src/sources/metrics/metastore_provider.rs b/quickwit/quickwit-datafusion/src/sources/metrics/metastore_provider.rs index 4e0927554e4..13accb6aa47 100644 --- a/quickwit/quickwit-datafusion/src/sources/metrics/metastore_provider.rs +++ b/quickwit/quickwit-datafusion/src/sources/metrics/metastore_provider.rs @@ -65,7 +65,7 @@ impl MetricsSplitProvider for MetastoreSplitProvider { async fn list_splits(&self, query: &MetricsSplitQuery) -> DFResult> { let metastore_query = to_metastore_query(&self.index_uid, query); let records = - list_parquet_splits_paginated(self.metastore.clone(), self.split_kind, metastore_query) + list_parquet_splits_paginated(&self.metastore, self.split_kind, metastore_query) .await .map_err(|err| datafusion::error::DataFusionError::External(Box::new(err)))?; let splits: Vec = diff --git a/quickwit/quickwit-index-management/src/parquet_garbage_collection.rs b/quickwit/quickwit-index-management/src/parquet_garbage_collection.rs index 6517123c43c..5700b286f04 100644 --- a/quickwit/quickwit-index-management/src/parquet_garbage_collection.rs +++ b/quickwit/quickwit-index-management/src/parquet_garbage_collection.rs @@ -177,7 +177,7 @@ async fn list_parquet_splits( let kind = parquet_split_kind_for_index(index_uid); protect_future( progress_opt, - list_parquet_splits_paginated(metastore.clone(), kind, query), + list_parquet_splits_paginated(metastore, kind, query), ) .await .context("failed to list parquet splits") diff --git a/quickwit/quickwit-indexing/src/actors/parquet_pipeline/parquet_merge_pipeline.rs b/quickwit/quickwit-indexing/src/actors/parquet_pipeline/parquet_merge_pipeline.rs index 6de3f5f5983..8ee01056e01 100644 --- a/quickwit/quickwit-indexing/src/actors/parquet_pipeline/parquet_merge_pipeline.rs +++ b/quickwit/quickwit-indexing/src/actors/parquet_pipeline/parquet_merge_pipeline.rs @@ -78,7 +78,7 @@ async fn fetch_published_parquet_splits_paginated( }; let query = quickwit_metastore::ListParquetSplitsQuery::for_index(index_uid.clone()) .retain_immature(OffsetDateTime::now_utc()); - let records = list_parquet_splits_paginated(metastore, kind, query).await?; + let records = list_parquet_splits_paginated(&metastore, kind, query).await?; debug!( num_splits = records.len(), "fetched published parquet splits for merge planning" diff --git a/quickwit/quickwit-integration-tests/src/test_utils/cluster_sandbox.rs b/quickwit/quickwit-integration-tests/src/test_utils/cluster_sandbox.rs index b79c113c481..7b4980f428d 100644 --- a/quickwit/quickwit-integration-tests/src/test_utils/cluster_sandbox.rs +++ b/quickwit/quickwit-integration-tests/src/test_utils/cluster_sandbox.rs @@ -211,7 +211,7 @@ impl ClusterSandboxBuilder { pub async fn build_and_start_standalone() -> ClusterSandbox { ClusterSandboxBuilder::default() - .add_node(QuickwitService::supported_services()) + .add_node(QuickwitService::default_services()) .build_config() .await .start() diff --git a/quickwit/quickwit-integration-tests/src/tests/metrics_distributed_tests.rs b/quickwit/quickwit-integration-tests/src/tests/metrics_distributed_tests.rs index 533c5960997..bb7e5e02293 100644 --- a/quickwit/quickwit-integration-tests/src/tests/metrics_distributed_tests.rs +++ b/quickwit/quickwit-integration-tests/src/tests/metrics_distributed_tests.rs @@ -201,7 +201,7 @@ async fn test_distributed_tasks_not_shuffles() { quickwit_common::setup_logging_for_tests(); let sandbox = ClusterSandboxBuilder::default() - .add_node(QuickwitService::supported_services()) + .add_node(QuickwitService::default_services()) .add_node([QuickwitService::Searcher]) .build_and_start() .await; diff --git a/quickwit/quickwit-integration-tests/src/tests/tls_tests.rs b/quickwit/quickwit-integration-tests/src/tests/tls_tests.rs index 8ba283be630..11a074a6010 100644 --- a/quickwit/quickwit-integration-tests/src/tests/tls_tests.rs +++ b/quickwit/quickwit-integration-tests/src/tests/tls_tests.rs @@ -83,7 +83,7 @@ async fn fetch_served_leaf_cert(addr: SocketAddr, ca_path: &str) -> Vec { async fn test_tls_rest() { quickwit_common::setup_logging_for_tests(); let mut sandbox_config = ClusterSandboxBuilder::default() - .add_node(QuickwitService::supported_services()) + .add_node(QuickwitService::default_services()) .build_config() .await; sandbox_config.node_configs[0].0.rest_config.tls_config = Some(fixture_tls_config()); @@ -195,7 +195,7 @@ async fn test_tls_grpc() { async fn test_mtls_rest() { quickwit_common::setup_logging_for_tests(); let mut sandbox_config = ClusterSandboxBuilder::default() - .add_node(QuickwitService::supported_services()) + .add_node(QuickwitService::default_services()) .build_config() .await; // Reuse the server cert/key as the client identity — it is signed by the same CA, so the @@ -247,7 +247,7 @@ async fn test_mtls_rest() { async fn test_health_check_server_plaintext_with_mtls_rest() { quickwit_common::setup_logging_for_tests(); let mut sandbox_config = ClusterSandboxBuilder::default() - .add_node(QuickwitService::supported_services()) + .add_node(QuickwitService::default_services()) .enable_health_check() .build_config() .await; @@ -344,7 +344,7 @@ async fn test_tls_rest_max_connection_age() { let ca_path = format!("{TLS_FIXTURES_DIR}/ca.crt"); let mut sandbox_config = ClusterSandboxBuilder::default() - .add_node(QuickwitService::supported_services()) + .add_node(QuickwitService::default_services()) .build_config() .await; sandbox_config.node_configs[0].0.rest_config.tls_config = Some(TlsConfig { @@ -410,7 +410,7 @@ async fn test_tls_grpc_max_connection_age() { let ca_path = format!("{TLS_FIXTURES_DIR}/ca.crt"); let mut sandbox_config = ClusterSandboxBuilder::default() - .add_node(QuickwitService::supported_services()) + .add_node(QuickwitService::default_services()) .build_config() .await; // One-way TLS (no client-cert verification) so the bare HTTP/2 probe below, which presents no @@ -459,7 +459,7 @@ async fn test_tls_rest_cert_hot_reload() { let ca_path = format!("{TLS_FIXTURES_DIR}/ca.crt"); let mut sandbox_config = ClusterSandboxBuilder::default() - .add_node(QuickwitService::supported_services()) + .add_node(QuickwitService::default_services()) .build_config() .await; sandbox_config.node_configs[0].0.rest_config.tls_config = Some(TlsConfig { diff --git a/quickwit/quickwit-janitor/src/retention_policy_execution.rs b/quickwit/quickwit-janitor/src/retention_policy_execution.rs index 97109cec427..99aa2a5caf9 100644 --- a/quickwit/quickwit-janitor/src/retention_policy_execution.rs +++ b/quickwit/quickwit-janitor/src/retention_policy_execution.rs @@ -117,11 +117,7 @@ pub async fn run_execute_parquet_retention_policy( ParquetSplitKind::Metrics }; let expired_splits: Vec = ctx - .protect_future(list_parquet_splits_paginated( - metastore.clone(), - kind, - query, - )) + .protect_future(list_parquet_splits_paginated(&metastore, kind, query)) .await?; if expired_splits.is_empty() { diff --git a/quickwit/quickwit-metastore/migrations/postgresql_deferred/README.md b/quickwit/quickwit-metastore/migrations/postgresql_deferred/README.md index 8d354c8ac09..eccdc277ea8 100644 --- a/quickwit/quickwit-metastore/migrations/postgresql_deferred/README.md +++ b/quickwit/quickwit-metastore/migrations/postgresql_deferred/README.md @@ -17,5 +17,5 @@ Long, degrade-gracefully-only migrations (e.g. `CREATE INDEX CONCURRENTLY`) run DROP INDEX CONCURRENTLY IF EXISTS foo_idx; ``` - If any migrations failed, they'll need to be manually cleaned up before attempting to retry as failed index creation leaves a broken index, without adding an entry into the migrations table. -- We tried to clean up invalid indexes inside the migration SQL, but because it's two separate statements, Postgres implicitly runs them inside a transaction block, overriding `-- no-transaction`, which can't work with `concurrently`. -- Never depend on an unshipped required migration; required migrations must never depend on a deferred one. \ No newline at end of file +- We tried to clean up invalid indexes inside the migration SQL, but because it's two separate statements, Postgres implicitly runs them inside a transaction block, overriding `-- no-transaction`, which can't work with `concurrently`. +- Never depend on an unshipped required migration; required migrations must never depend on a deferred one. diff --git a/quickwit/quickwit-metastore/src/lib.rs b/quickwit/quickwit-metastore/src/lib.rs index 62d46b4af49..31bec51a292 100644 --- a/quickwit/quickwit-metastore/src/lib.rs +++ b/quickwit/quickwit-metastore/src/lib.rs @@ -52,7 +52,7 @@ pub use metastore::{ StageParquetSplitsRequestExt, StageSplitsRequestExt, UpdateIndexRequestExt, UpdateSourceRequestExt, file_backed, list_parquet_splits_page, list_parquet_splits_paginated, }; -pub use metastore_factory::{MetastoreFactory, UnsupportedMetastore}; +pub use metastore_factory::{MetastoreFactory, MetastoreFactoryOptions, UnsupportedMetastore}; pub use metastore_resolver::MetastoreResolver; use quickwit_common::is_disjoint; use quickwit_doc_mapper::tag_pruning::TagFilterAst; diff --git a/quickwit/quickwit-metastore/src/metastore/file_backed/file_backed_metastore_factory.rs b/quickwit/quickwit-metastore/src/metastore/file_backed/file_backed_metastore_factory.rs index 2c489eb4dd5..96823851221 100644 --- a/quickwit/quickwit-metastore/src/metastore/file_backed/file_backed_metastore_factory.rs +++ b/quickwit/quickwit-metastore/src/metastore/file_backed/file_backed_metastore_factory.rs @@ -26,7 +26,9 @@ use regex::Regex; use tokio::sync::Mutex; use tracing::debug; -use crate::{FileBackedMetastore, MetastoreFactory, MetastoreResolverError}; +use crate::{ + FileBackedMetastore, MetastoreFactory, MetastoreFactoryOptions, MetastoreResolverError, +}; /// A file-backed metastore factory. /// @@ -101,6 +103,7 @@ impl MetastoreFactory for FileBackedMetastoreFactory { &self, _metastore_config: &MetastoreConfig, uri: &Uri, + _options: MetastoreFactoryOptions, ) -> Result { let (uri_stripped, polling_interval_opt) = extract_polling_interval_from_uri(uri.as_str()); let uri = Uri::from_str(&uri_stripped).map_err(|_| { diff --git a/quickwit/quickwit-metastore/src/metastore/mod.rs b/quickwit/quickwit-metastore/src/metastore/mod.rs index c93d381cdc7..84011bcb4d2 100644 --- a/quickwit/quickwit-metastore/src/metastore/mod.rs +++ b/quickwit/quickwit-metastore/src/metastore/mod.rs @@ -1012,7 +1012,7 @@ pub async fn list_parquet_splits_page( /// `page_size`; `after_split_id` is used as the starting cursor when already /// set and is advanced internally after each full page. pub async fn list_parquet_splits_paginated( - metastore: MetastoreServiceClient, + metastore: &MetastoreServiceClient, kind: ParquetSplitKind, mut query: ListParquetSplitsQuery, ) -> MetastoreResult> { @@ -1020,7 +1020,7 @@ pub async fn list_parquet_splits_paginated( let mut splits = Vec::new(); loop { - let mut page = list_parquet_splits_page(&metastore, kind, &mut query).await?; + let mut page = list_parquet_splits_page(metastore, kind, &mut query).await?; splits.append(&mut page.splits); if !page.has_next_page { break; diff --git a/quickwit/quickwit-metastore/src/metastore/postgres/error.rs b/quickwit/quickwit-metastore/src/metastore/postgres/error.rs index 4f4fe7a2205..611d6a6822f 100644 --- a/quickwit/quickwit-metastore/src/metastore/postgres/error.rs +++ b/quickwit/quickwit-metastore/src/metastore/postgres/error.rs @@ -13,40 +13,46 @@ // limitations under the License. use quickwit_proto::metastore::{EntityKind, MetastoreError}; -use sqlx::postgres::PgDatabaseError; use tracing::error; // https://www.postgresql.org/docs/current/errcodes-appendix.html mod pg_error_codes { pub const FOREIGN_KEY_VIOLATION: &str = "23503"; pub const UNIQUE_VIOLATION: &str = "23505"; + pub const READ_ONLY_SQL_TRANSACTION: &str = "25006"; } pub(super) fn convert_sqlx_err(index_id: &str, sqlx_error: sqlx::Error) -> MetastoreError { match &sqlx_error { sqlx::Error::Database(boxed_db_error) => { - let pg_db_error = boxed_db_error.downcast_ref::(); - let pg_error_code = pg_db_error.code(); - let pg_error_table = pg_db_error.table(); + let pg_error_code = boxed_db_error.code(); + let pg_error_table = boxed_db_error.table(); - match (pg_error_code, pg_error_table) { - (pg_error_codes::FOREIGN_KEY_VIOLATION, _) => { + match (pg_error_code.as_deref(), pg_error_table) { + (Some(pg_error_codes::FOREIGN_KEY_VIOLATION), _) => { MetastoreError::NotFound(EntityKind::Index { index_id: index_id.to_string(), }) } - (pg_error_codes::UNIQUE_VIOLATION, Some(table)) if table.starts_with("indexes") => { + (Some(pg_error_codes::UNIQUE_VIOLATION), Some(table)) + if table.starts_with("indexes") => + { MetastoreError::AlreadyExists(EntityKind::Index { index_id: index_id.to_string(), }) } - (pg_error_codes::UNIQUE_VIOLATION, _) => { + (Some(pg_error_codes::UNIQUE_VIOLATION), _) => { error!(error=?boxed_db_error, "postgresql-error"); MetastoreError::Internal { message: "unique key violation".to_string(), cause: format!("DB error {boxed_db_error:?}"), } } + (Some(pg_error_codes::READ_ONLY_SQL_TRANSACTION), _) => MetastoreError::Forbidden { + message: format!( + "cannot issue write request on a read-only replica: {boxed_db_error}" + ), + }, _ => { error!(error=?boxed_db_error, "postgresql-error"); MetastoreError::Db { diff --git a/quickwit/quickwit-metastore/src/metastore/postgres/factory.rs b/quickwit/quickwit-metastore/src/metastore/postgres/factory.rs index 97aded689b1..5f872a611fa 100644 --- a/quickwit/quickwit-metastore/src/metastore/postgres/factory.rs +++ b/quickwit/quickwit-metastore/src/metastore/postgres/factory.rs @@ -22,7 +22,9 @@ use quickwit_proto::metastore::MetastoreServiceClient; use tokio::sync::Mutex; use tracing::debug; -use crate::{MetastoreFactory, MetastoreResolverError, PostgresqlMetastore}; +use crate::{ + MetastoreFactory, MetastoreFactoryOptions, MetastoreResolverError, PostgresqlMetastore, +}; #[derive(Clone, Default)] pub struct PostgresqlMetastoreFactory { @@ -31,30 +33,39 @@ pub struct PostgresqlMetastoreFactory { // In contrast to the file-backed metastore, we use a strong pointer here, so that the // `Metastore` doesn't get dropped. This is done in order to keep the underlying connection // pool to Postgres alive. - cache: Arc>>, + // + // The cache is keyed on the resolution options as well as the URI, so that the read-write and + // read-only clients for the same URI get their own connection pool. + cache: Arc>>, } impl PostgresqlMetastoreFactory { - async fn get_from_cache(&self, uri: &Uri) -> Option { + async fn get_from_cache( + &self, + uri: &Uri, + options: MetastoreFactoryOptions, + ) -> Option { let cache_lock = self.cache.lock().await; - cache_lock.get(uri).cloned() + cache_lock.get(&(uri.clone(), options)).cloned() } /// If there is a valid entry in the cache to begin with, we trash the new /// one and return the old one. /// /// This way we make sure that we keep only one instance associated - /// to the key `uri` outside of this struct. + /// to the cache key outside of this struct. async fn cache_metastore( &self, uri: Uri, + options: MetastoreFactoryOptions, metastore: MetastoreServiceClient, ) -> MetastoreServiceClient { let mut cache_lock = self.cache.lock().await; - if let Some(metastore) = cache_lock.get(&uri) { + let cache_key = (uri, options); + if let Some(metastore) = cache_lock.get(&cache_key) { return metastore.clone(); } - cache_lock.insert(uri, metastore.clone()); + cache_lock.insert(cache_key, metastore.clone()); metastore } } @@ -69,8 +80,9 @@ impl MetastoreFactory for PostgresqlMetastoreFactory { &self, metastore_config: &MetastoreConfig, uri: &Uri, + options: MetastoreFactoryOptions, ) -> Result { - if let Some(metastore) = self.get_from_cache(uri).await { + if let Some(metastore) = self.get_from_cache(uri, options).await { debug!("using metastore from cache"); return Ok(metastore); } @@ -82,12 +94,15 @@ impl MetastoreFactory for PostgresqlMetastoreFactory { ); MetastoreResolverError::InvalidConfig(message) })?; - let postgresql_metastore = PostgresqlMetastore::new(postgresql_metastore_config, uri) - .await - .map(MetastoreServiceClient::new) - .map_err(MetastoreResolverError::Initialization)?; + let postgresql_metastore = if options.read_only { + PostgresqlMetastore::new_read_only(postgresql_metastore_config, uri).await + } else { + PostgresqlMetastore::new(postgresql_metastore_config, uri).await + } + .map(MetastoreServiceClient::new) + .map_err(MetastoreResolverError::Initialization)?; let unique_metastore_for_uri = self - .cache_metastore(uri.clone(), postgresql_metastore) + .cache_metastore(uri.clone(), options, postgresql_metastore) .await; Ok(unique_metastore_for_uri) } diff --git a/quickwit/quickwit-metastore/src/metastore/postgres/metastore.rs b/quickwit/quickwit-metastore/src/metastore/postgres/metastore.rs index 28bf78315ed..a236fe91630 100644 --- a/quickwit/quickwit-metastore/src/metastore/postgres/metastore.rs +++ b/quickwit/quickwit-metastore/src/metastore/postgres/metastore.rs @@ -61,7 +61,6 @@ use time::OffsetDateTime; use tracing::{debug, info, instrument, warn}; use uuid::Uuid; -use super::QW_POSTGRES_READ_ONLY_ENV_KEY; use super::error::convert_sqlx_err; use super::migrator::Migrations; use super::model::{PgDeleteTask, PgIndex, PgIndexTemplate, PgShard, PgSplit, Splits}; @@ -69,6 +68,10 @@ use super::parquet_model::{InsertableParquetSplit, ParquetSplitRecord, PgParquet use super::pool::TrackedPool; use super::split_stream::SplitStream; use super::utils::{append_query_filters_and_order_by, establish_connection}; +use super::{ + QW_POSTGRES_READ_ONLY_ENV_KEY, QW_POSTGRES_SKIP_DEFERRED_MIGRATIONS_ENV_KEY, + QW_POSTGRES_SKIP_MIGRATION_LOCKING_ENV_KEY, QW_POSTGRES_SKIP_MIGRATIONS_ENV_KEY, +}; use crate::checkpoint::{ IndexCheckpointDelta, PartitionId, SourceCheckpoint, SourceCheckpointDelta, }; @@ -101,11 +104,78 @@ impl fmt::Debug for PostgresqlMetastore { } } +/// Connection/migration options for a [`PostgresqlMetastore`]. +#[derive(Clone, Copy, Debug)] +struct PostgresqlMetastoreOptions { + read_only: bool, + skip_migrations: bool, + skip_locking: bool, + skip_deferred: bool, +} + +impl PostgresqlMetastoreOptions { + /// Default options, with the read-only and migration flags read from the environment. + fn from_env() -> Self { + Self { + // These environment variables are process-global and don't change over a process's + // lifetime, so they're read (and logged) once rather than on every metastore + // construction, which can recur frequently. + read_only: get_bool_from_env_cached!(QW_POSTGRES_READ_ONLY_ENV_KEY, false), + skip_migrations: get_bool_from_env_cached!(QW_POSTGRES_SKIP_MIGRATIONS_ENV_KEY, false), + skip_locking: get_bool_from_env_cached!( + QW_POSTGRES_SKIP_MIGRATION_LOCKING_ENV_KEY, + false + ), + skip_deferred: get_bool_from_env_cached!( + QW_POSTGRES_SKIP_DEFERRED_MIGRATIONS_ENV_KEY, + false + ), + } + } + + /// Options for a read-only metastore: the connection is read-only and migrations are skipped, + /// since the read replica cannot be written to and is migrated by the primary. + fn read_only() -> Self { + Self { + read_only: true, + skip_migrations: true, + skip_locking: false, + skip_deferred: true, + } + } +} + impl PostgresqlMetastore { /// Creates a metastore given a database URI. pub async fn new( postgres_metastore_config: &PostgresMetastoreConfig, connection_uri: &Uri, + ) -> MetastoreResult { + Self::new_with_options( + postgres_metastore_config, + connection_uri, + PostgresqlMetastoreOptions::from_env(), + ) + .await + } + + /// Creates a read-only metastore given a database URI. + pub async fn new_read_only( + postgres_metastore_config: &PostgresMetastoreConfig, + connection_uri: &Uri, + ) -> MetastoreResult { + Self::new_with_options( + postgres_metastore_config, + connection_uri, + PostgresqlMetastoreOptions::read_only(), + ) + .await + } + + async fn new_with_options( + postgres_metastore_config: &PostgresMetastoreConfig, + connection_uri: &Uri, + options: PostgresqlMetastoreOptions, ) -> MetastoreResult { let min_connections = postgres_metastore_config.min_connections; let max_connections = postgres_metastore_config.max_connections.get(); @@ -119,11 +189,6 @@ impl PostgresqlMetastore { .max_connection_lifetime_opt() .expect("PostgreSQL metastore config should have been validated"); - // This environment variable is process-global and doesn't change over a process's - // lifetime, so it's read (and logged) once rather than on every metastore - // construction, which can recur frequently. - let read_only = get_bool_from_env_cached!(QW_POSTGRES_READ_ONLY_ENV_KEY, false); - let connection_pool = establish_connection( connection_uri, min_connections, @@ -131,11 +196,18 @@ impl PostgresqlMetastore { acquire_timeout, idle_timeout_opt, max_lifetime_opt, - read_only, + options.read_only, ) .await?; - Migrations::from_env(connection_pool.clone()).run().await?; + Migrations::new( + connection_pool.clone(), + options.skip_migrations, + options.skip_locking, + options.skip_deferred, + ) + .run() + .await?; let metastore = PostgresqlMetastore { uri: connection_uri.clone(), diff --git a/quickwit/quickwit-metastore/src/metastore/postgres/migrator.rs b/quickwit/quickwit-metastore/src/metastore/postgres/migrator.rs index 342973483ef..4e96dca7645 100644 --- a/quickwit/quickwit-metastore/src/metastore/postgres/migrator.rs +++ b/quickwit/quickwit-metastore/src/metastore/postgres/migrator.rs @@ -14,7 +14,6 @@ use std::collections::BTreeMap; -use quickwit_common::get_bool_from_env; use quickwit_metrics::{counter, labels}; use quickwit_proto::metastore::{MetastoreError, MetastoreResult}; use sqlx::migrate::{Migrate, Migrator}; @@ -23,10 +22,6 @@ use tracing::{error, info}; use super::metrics::DEFERRED_MIGRATIONS_APPLY; use super::pool::TrackedPool; -use super::{ - QW_POSTGRES_SKIP_DEFERRED_MIGRATIONS_ENV_KEY, QW_POSTGRES_SKIP_MIGRATION_LOCKING_ENV_KEY, - QW_POSTGRES_SKIP_MIGRATIONS_ENV_KEY, -}; // The deferred migrations should be attempted by only one metastore pod. We do that by having // metastore pods try to acquire a lock to run the deferred migrations. Pods that don't get the lock @@ -60,12 +55,18 @@ pub(super) struct Migrations { } impl Migrations { - pub(super) fn from_env(connection_pool: TrackedPool) -> Self { + /// Builds a migration runner with explicit skip settings. + pub(super) fn new( + connection_pool: TrackedPool, + skip_migrations: bool, + skip_locking: bool, + skip_deferred: bool, + ) -> Self { Self { connection_pool, - skip_migrations: get_bool_from_env(QW_POSTGRES_SKIP_MIGRATIONS_ENV_KEY, false), - skip_locking: get_bool_from_env(QW_POSTGRES_SKIP_MIGRATION_LOCKING_ENV_KEY, false), - skip_deferred: get_bool_from_env(QW_POSTGRES_SKIP_DEFERRED_MIGRATIONS_ENV_KEY, false), + skip_migrations, + skip_locking, + skip_deferred, } } @@ -265,12 +266,7 @@ mod tests { } fn migrations(connection_pool: &TrackedPool, skip_migrations: bool) -> Migrations { - Migrations { - connection_pool: connection_pool.clone(), - skip_migrations, - skip_locking: false, - skip_deferred: false, - } + Migrations::new(connection_pool.clone(), skip_migrations, false, false) } // A single controlled migration so these tests don't depend on the production migration set. diff --git a/quickwit/quickwit-metastore/src/metastore_factory.rs b/quickwit/quickwit-metastore/src/metastore_factory.rs index 872b16d3071..89eb0334090 100644 --- a/quickwit/quickwit-metastore/src/metastore_factory.rs +++ b/quickwit/quickwit-metastore/src/metastore_factory.rs @@ -19,6 +19,14 @@ use quickwit_proto::metastore::MetastoreServiceClient; use crate::MetastoreResolverError; +/// Options controlling how a metastore client is resolved. +#[derive(Clone, Copy, Debug, Default, Eq, Hash, PartialEq)] +pub struct MetastoreFactoryOptions { + /// Whether the resolved metastore client should use a read-only connection. Only supported by + /// the PostgreSQL backend. + pub read_only: bool, +} + /// A metastore factory builds a [`MetastoreServiceClient`] object for a target [`MetastoreBackend`] /// from a [`MetastoreConfig`] and a [`Uri`]. #[cfg_attr(any(test, feature = "testsuite"), mockall::automock)] @@ -27,11 +35,13 @@ pub trait MetastoreFactory: Send + Sync + 'static { /// Returns the metastore backend targeted by the factory. fn backend(&self) -> MetastoreBackend; - /// Returns the appropriate [`MetastoreServiceClient`] object for the `uri`. + /// Returns the appropriate [`MetastoreServiceClient`] object for the `uri` and resolution + /// `options`. async fn resolve( &self, metastore_config: &MetastoreConfig, uri: &Uri, + options: MetastoreFactoryOptions, ) -> Result; } @@ -59,6 +69,7 @@ impl MetastoreFactory for UnsupportedMetastore { &self, _metastore_config: &MetastoreConfig, _uri: &Uri, + _options: MetastoreFactoryOptions, ) -> Result { Err(MetastoreResolverError::UnsupportedBackend( self.message.to_string(), diff --git a/quickwit/quickwit-metastore/src/metastore_resolver.rs b/quickwit/quickwit-metastore/src/metastore_resolver.rs index bc4d2cac6e7..cf7fc435faa 100644 --- a/quickwit/quickwit-metastore/src/metastore_resolver.rs +++ b/quickwit/quickwit-metastore/src/metastore_resolver.rs @@ -25,7 +25,7 @@ use quickwit_storage::StorageResolver; use crate::metastore::file_backed::FileBackedMetastoreFactory; #[cfg(feature = "postgres")] use crate::metastore::postgres::PostgresqlMetastoreFactory; -use crate::{MetastoreFactory, MetastoreResolverError}; +use crate::{MetastoreFactory, MetastoreFactoryOptions, MetastoreResolverError}; type FactoryAndConfig = (Box, MetastoreConfig); @@ -53,6 +53,24 @@ impl MetastoreResolver { pub async fn resolve( &self, uri: &Uri, + ) -> Result { + self.resolve_inner(uri, MetastoreFactoryOptions::default()) + .await + } + + /// Resolves the given `uri` as a read-only metastore client. Only supported for PostgreSQL. + pub async fn resolve_read_only( + &self, + uri: &Uri, + ) -> Result { + self.resolve_inner(uri, MetastoreFactoryOptions { read_only: true }) + .await + } + + async fn resolve_inner( + &self, + uri: &Uri, + options: MetastoreFactoryOptions, ) -> Result { let backend = match uri.protocol() { Protocol::Azure => MetastoreBackend::File, @@ -67,13 +85,20 @@ impl MetastoreResolver { )); } }; + if options.read_only && backend != MetastoreBackend::PostgreSQL { + return Err(MetastoreResolverError::UnsupportedBackend( + "read-only option is only supported for the PostgreSQL backend".to_string(), + )); + } let (metastore_factory, metastore_config) = self .per_backend_factories .get(&backend) .ok_or(MetastoreResolverError::UnsupportedBackend( "no metastore factory is registered for this backend".to_string(), ))?; - let metastore = metastore_factory.resolve(metastore_config, uri).await?; + let metastore = metastore_factory + .resolve(metastore_config, uri, options) + .await?; Ok(metastore) } @@ -184,6 +209,21 @@ mod tests { metastore_resolver.resolve(&metastore_uri).await.unwrap(); } + #[tokio::test] + async fn test_metastore_resolver_should_reject_read_only_file() { + let metastore_resolver = MetastoreResolver::unconfigured(); + let metastore_uri = Uri::from_str("ram:///metastore").unwrap(); + let error = metastore_resolver + .resolve_read_only(&metastore_uri) + .await + .unwrap_err(); + assert!(matches!( + error, + MetastoreResolverError::UnsupportedBackend(message) + if message == "read-only option is only supported for the PostgreSQL backend" + )); + } + #[cfg(feature = "postgres")] #[tokio::test] async fn test_postgres_and_postgresql_protocol_accepted() { @@ -200,4 +240,48 @@ mod tests { metastore_resolver.resolve(&postgres_uri).await.unwrap(); } } + + #[cfg(feature = "postgres")] + #[tokio::test] + async fn test_metastore_resolver_resolve_read_only_postgres() { + use std::env; + + use quickwit_common::rand::append_random_suffix; + use quickwit_config::IndexConfig; + use quickwit_proto::metastore::{ + CreateIndexRequest, ListIndexesMetadataRequest, MetastoreError, MetastoreService, + }; + + use crate::CreateIndexRequestExt; + + let metastore_resolver = MetastoreResolver::unconfigured(); + let test_database_url = env::var("QW_TEST_DATABASE_URL").unwrap_or_else(|_| { + "postgres://quickwit-dev:quickwit-dev@localhost/quickwit-metastore-dev".to_string() + }); + let postgres_uri = Uri::from_str(&test_database_url).unwrap(); + // The read replica skips migrations, so ensure the schema exists by resolving the + // read-write metastore first. + metastore_resolver.resolve(&postgres_uri).await.unwrap(); + + let read_only_metastore = metastore_resolver + .resolve_read_only(&postgres_uri) + .await + .unwrap(); + // A read-only connection must still serve read RPCs. + read_only_metastore + .list_indexes_metadata(ListIndexesMetadataRequest::all()) + .await + .unwrap(); + + let index_id = append_random_suffix("test-read-only-postgres"); + let index_uri = format!("ram:///indexes/{index_id}"); + let index_config = IndexConfig::for_test(&index_id, &index_uri); + let create_index_request = + CreateIndexRequest::try_from_index_config(&index_config).unwrap(); + let error = read_only_metastore + .create_index(create_index_request) + .await + .unwrap_err(); + assert!(matches!(error, MetastoreError::Forbidden { .. })); + } } diff --git a/quickwit/quickwit-search/src/lib.rs b/quickwit/quickwit-search/src/lib.rs index cde7597b186..5c3b8ff30a0 100644 --- a/quickwit/quickwit-search/src/lib.rs +++ b/quickwit/quickwit-search/src/lib.rs @@ -50,7 +50,7 @@ use quickwit_common::thread_pool::with_priority::ThreadPoolWithPriority; use quickwit_common::tower::Pool; use quickwit_doc_mapper::DocMapper; use quickwit_proto::metastore::{ - ListIndexesMetadataRequest, ListSplitsRequest, MetastoreService, MetastoreServiceClient, + ListIndexesMetadataRequest, ListSplitsRequest, MetastoreServiceClient, }; use tantivy::schema::NamedFieldDocument; @@ -67,6 +67,7 @@ use quickwit_metastore::{ IndexMetadata, ListIndexesMetadataResponseExt, ListSplitsQuery, ListSplitsRequestExt, MetastoreServiceStreamSplitsExt, SplitMetadata, SplitState, }; +use quickwit_proto::metastore::MetastoreService; use quickwit_proto::search::{ LeafResourceStats, PartialHit, SearchRequest, SearchResponse, SplitIdAndFooterOffsets, SplitResourceStats, @@ -182,7 +183,7 @@ fn extract_split_and_footer_offsets(split_metadata: &SplitMetadata) -> SplitIdAn /// Get all splits of given index ids pub async fn list_all_splits( index_uids: Vec, - metastore: &mut MetastoreServiceClient, + metastore: &MetastoreServiceClient, ) -> crate::Result> { list_relevant_splits(index_uids, None, None, None, metastore).await } @@ -193,7 +194,7 @@ pub async fn list_relevant_splits( start_timestamp: Option, end_timestamp: Option, tags_filter_opt: Option, - metastore: &mut MetastoreServiceClient, + metastore: &MetastoreServiceClient, ) -> crate::Result> { let Some(mut query) = ListSplitsQuery::try_from_index_uids(index_uids) else { return Ok(Vec::new()); @@ -222,7 +223,7 @@ pub async fn list_relevant_splits( /// Patterns follow the elastic search patterns. pub async fn resolve_index_patterns( index_id_patterns: &[String], - metastore: &mut MetastoreServiceClient, + metastore: &MetastoreServiceClient, ) -> crate::Result> { let list_indexes_metadata_request = if index_id_patterns.is_empty() { ListIndexesMetadataRequest::all() @@ -300,7 +301,7 @@ pub async fn single_node_search( root_search( &searcher_context, search_request, - metastore, + &metastore, &cluster_client, ) .await diff --git a/quickwit/quickwit-search/src/list_fields/root.rs b/quickwit/quickwit-search/src/list_fields/root.rs index 66d5947d123..6e0382f9874 100644 --- a/quickwit/quickwit-search/src/list_fields/root.rs +++ b/quickwit/quickwit-search/src/list_fields/root.rs @@ -54,10 +54,10 @@ struct IndexMetasForLeafSearch { pub async fn root_list_fields( list_fields_req: ListFieldsRequest, cluster_client: &ClusterClient, - mut metastore: MetastoreServiceClient, + metastore: &MetastoreServiceClient, ) -> crate::Result { let indexes_metadata = - resolve_index_patterns(&list_fields_req.index_id_patterns[..], &mut metastore).await?; + resolve_index_patterns(&list_fields_req.index_id_patterns[..], metastore).await?; // The request contains a wildcard, but couldn't find any index. if indexes_metadata.is_empty() { @@ -115,7 +115,7 @@ pub async fn root_list_fields( start_timestamp, end_timestamp, tags_filter_opt, - &mut metastore, + metastore, ) .await?; diff --git a/quickwit/quickwit-search/src/list_terms.rs b/quickwit/quickwit-search/src/list_terms.rs index 0c575793f41..906574df19c 100644 --- a/quickwit/quickwit-search/src/list_terms.rs +++ b/quickwit/quickwit-search/src/list_terms.rs @@ -48,12 +48,12 @@ use crate::{ClusterClient, SearchError, SearchJob, SearcherContext, resolve_inde #[instrument(skip(list_terms_request, cluster_client, metastore))] pub async fn root_list_terms( list_terms_request: &ListTermsRequest, - mut metastore: MetastoreServiceClient, + metastore: &MetastoreServiceClient, cluster_client: &ClusterClient, ) -> crate::Result { let start_instant = tokio::time::Instant::now(); let indexes_metadata = - resolve_index_patterns(&list_terms_request.index_id_patterns, &mut metastore).await?; + resolve_index_patterns(&list_terms_request.index_id_patterns, metastore).await?; // The request contains a wildcard, but couldn't find any index. if indexes_metadata.is_empty() { return Ok(ListTermsResponse { @@ -113,7 +113,6 @@ pub async fn root_list_terms( .collect(); let list_splits_request = ListSplitsRequest::try_from_list_splits_query(&query)?; let split_metadatas: Vec = metastore - .clone() .list_splits(list_splits_request) .await? .collect_splits_metadata() diff --git a/quickwit/quickwit-search/src/root.rs b/quickwit/quickwit-search/src/root.rs index aaf29b7c813..0839ce3ef3c 100644 --- a/quickwit/quickwit-search/src/root.rs +++ b/quickwit/quickwit-search/src/root.rs @@ -1208,7 +1208,7 @@ pub fn ensure_all_indexes_found( } async fn refine_and_list_matches( - metastore: &mut MetastoreServiceClient, + metastore: &MetastoreServiceClient, search_request: &mut SearchRequest, indexes_metadata: Vec, query_ast_resolved: QueryAst, @@ -1251,7 +1251,7 @@ async fn refine_and_list_matches( /// Fetches the list of splits and their metadata from the metastore async fn plan_splits_for_root_search( search_request: &mut SearchRequest, - metastore: &mut MetastoreServiceClient, + metastore: &MetastoreServiceClient, ) -> crate::Result<(Vec, IndexesMetasForLeafSearch)> { let list_indexes_metadatas_request = ListIndexesMetadataRequest { index_id_patterns: search_request.index_id_patterns.clone(), @@ -1295,14 +1295,14 @@ async fn plan_splits_for_root_search( pub async fn root_search( searcher_context: &SearcherContext, mut search_request: SearchRequest, - mut metastore: MetastoreServiceClient, + metastore: &MetastoreServiceClient, cluster_client: &ClusterClient, ) -> crate::Result { let start_instant = Instant::now(); let (split_metadatas, indexes_meta_for_leaf_search) = RootSearchMetricsFuture { start: start_instant, - tracked: plan_splits_for_root_search(&mut search_request, &mut metastore), + tracked: plan_splits_for_root_search(&mut search_request, metastore), is_success: None, step: RootSearchMetricsStep::Plan, } @@ -1364,7 +1364,7 @@ pub async fn root_search( /// Returns details on how a query would be executed pub async fn search_plan( mut search_request: SearchRequest, - mut metastore: MetastoreServiceClient, + metastore: &MetastoreServiceClient, ) -> crate::Result { let list_indexes_metadatas_request = ListIndexesMetadataRequest { index_id_patterns: search_request.index_id_patterns.clone(), @@ -1396,7 +1396,7 @@ pub async fn search_plan( let request_metadata = validate_request_and_build_metadata(&indexes_metadata, &search_request)?; let split_metadatas = refine_and_list_matches( - &mut metastore, + metastore, &mut search_request, indexes_metadata, request_metadata.query_ast_resolved.clone(), @@ -1902,7 +1902,8 @@ mod tests { use quickwit_indexing::MockSplitBuilder; use quickwit_metastore::{IndexMetadata, ListSplitsRequestExt, ListSplitsResponseExt}; use quickwit_proto::metastore::{ - ListIndexesMetadataResponse, ListSplitsResponse, MockMetastoreService, + ListIndexesMetadataResponse, ListSplitsResponse, MetastoreServiceClient, + MockMetastoreService, }; use quickwit_proto::search::{ ScrollRequest, SortByValue, SortOrder, SortValue, SplitSearchError, @@ -2758,7 +2759,7 @@ mod tests { let search_response = root_search( &SearcherContext::for_test(), search_request, - MetastoreServiceClient::from_mock(mock_metastore), + &MetastoreServiceClient::from_mock(mock_metastore), &cluster_client, ) .await @@ -2828,7 +2829,7 @@ mod tests { let search_response = root_search( &searcher_context, search_request, - MetastoreServiceClient::from_mock(mock_metastore), + &MetastoreServiceClient::from_mock(mock_metastore), &cluster_client, ) .await @@ -2920,7 +2921,7 @@ mod tests { let search_response = root_search( &SearcherContext::for_test(), search_request, - MetastoreServiceClient::from_mock(mock_metastore), + &MetastoreServiceClient::from_mock(mock_metastore), &cluster_client, ) .await @@ -3215,7 +3216,7 @@ mod tests { let search_response = root_search( &SearcherContext::for_test(), search_request, - MetastoreServiceClient::from_mock(mock_metastore), + &MetastoreServiceClient::from_mock(mock_metastore), &cluster_client, ) .await?; @@ -3333,7 +3334,7 @@ mod tests { let search_response = root_search( &SearcherContext::for_test(), search_request, - MetastoreServiceClient::from_mock(mock_metastore), + &MetastoreServiceClient::from_mock(mock_metastore), &cluster_client, ) .await @@ -3465,7 +3466,7 @@ mod tests { let search_response = root_search( &SearcherContext::for_test(), search_request.clone(), - MetastoreServiceClient::from_mock(mock_metastore), + &MetastoreServiceClient::from_mock(mock_metastore), &cluster_client, ) .await?; @@ -3647,7 +3648,7 @@ mod tests { let search_response = root_search( &SearcherContext::for_test(), search_request.clone(), - MetastoreServiceClient::from_mock(mock_metastore), + &MetastoreServiceClient::from_mock(mock_metastore), &cluster_client, ) .await?; @@ -3771,7 +3772,7 @@ mod tests { let search_response = root_search( &searcher_context, search_request, - mock_metastore_client.clone(), + &mock_metastore_client, &cluster_client, ) .await @@ -3790,7 +3791,7 @@ mod tests { let search_error = root_search( &searcher_context, search_request, - mock_metastore_client, + &mock_metastore_client, &cluster_client, ) .await @@ -3915,7 +3916,7 @@ mod tests { let search_response = root_search( &SearcherContext::for_test(), search_request, - MetastoreServiceClient::from_mock(mock_metastore), + &MetastoreServiceClient::from_mock(mock_metastore), &cluster_client, ) .await @@ -4055,7 +4056,7 @@ mod tests { let search_response = root_search( &SearcherContext::for_test(), search_request, - MetastoreServiceClient::from_mock(mock_metastore), + &MetastoreServiceClient::from_mock(mock_metastore), &cluster_client, ) .await @@ -4136,7 +4137,7 @@ mod tests { let search_response = root_search( &SearcherContext::for_test(), search_request, - MetastoreServiceClient::from_mock(mock_metastore), + &MetastoreServiceClient::from_mock(mock_metastore), &cluster_client, ) .await @@ -4203,7 +4204,7 @@ mod tests { let search_response = root_search( &SearcherContext::for_test(), search_request, - MetastoreServiceClient::from_mock(mock_metastore), + &MetastoreServiceClient::from_mock(mock_metastore), &cluster_client, ) .await @@ -4293,7 +4294,7 @@ mod tests { let search_response = root_search( &SearcherContext::for_test(), search_request, - MetastoreServiceClient::from_mock(mock_metastore), + &MetastoreServiceClient::from_mock(mock_metastore), &cluster_client, ) .await @@ -4375,7 +4376,7 @@ mod tests { let search_response = root_search( &SearcherContext::for_test(), search_request, - MetastoreServiceClient::from_mock(mock_metastore), + &MetastoreServiceClient::from_mock(mock_metastore), &cluster_client, ) .await @@ -4424,7 +4425,7 @@ mod tests { max_hits: 10, ..Default::default() }, - metastore.clone(), + &metastore, &cluster_client, ) .await @@ -4440,7 +4441,7 @@ mod tests { max_hits: 10, ..Default::default() }, - metastore, + &metastore, &cluster_client, ) .await @@ -4505,7 +4506,7 @@ mod tests { let search_response = root_search( &SearcherContext::for_test(), search_request, - MetastoreServiceClient::from_mock(mock_metastore), + &MetastoreServiceClient::from_mock(mock_metastore), &cluster_client, ) .await; @@ -4555,7 +4556,7 @@ mod tests { let search_response = root_search( &SearcherContext::for_test(), search_request, - metastore.clone(), + &metastore, &cluster_client, ) .await; @@ -4575,7 +4576,7 @@ mod tests { let search_response = root_search( &SearcherContext::for_test(), search_request, - metastore, + &metastore, &cluster_client, ) .await; @@ -4625,7 +4626,7 @@ mod tests { }); let search_response = search_plan( search_request, - MetastoreServiceClient::from_mock(mock_metastore), + &MetastoreServiceClient::from_mock(mock_metastore), ) .await .unwrap(); @@ -4712,7 +4713,7 @@ mod tests { ignore_missing_indexes: true, ..Default::default() }, - mock_metastore_service.clone(), + &mock_metastore_service, ) .await .unwrap(); @@ -4726,7 +4727,7 @@ mod tests { ignore_missing_indexes: false, ..Default::default() }, - mock_metastore_service.clone(), + &mock_metastore_service, ) .await .unwrap_err(); @@ -5093,7 +5094,7 @@ mod tests { let search_response = root_search( &searcher_context, search_request, - MetastoreServiceClient::from_mock(mock_metastore), + &MetastoreServiceClient::from_mock(mock_metastore), &cluster_client, ) .await @@ -5359,7 +5360,7 @@ mod tests { let search_response = root_search( &searcher_context, search_request, - MetastoreServiceClient::from_mock(mock_metastore), + &MetastoreServiceClient::from_mock(mock_metastore), &cluster_client, ) .await @@ -5541,7 +5542,7 @@ mod tests { let search_response = root_search( &SearcherContext::for_test(), search_request, - MetastoreServiceClient::from_mock(mock_metastore), + &MetastoreServiceClient::from_mock(mock_metastore), &cluster_client, ) .await @@ -5662,7 +5663,7 @@ mod tests { let search_response = root_search( &SearcherContext::for_test(), search_request, - MetastoreServiceClient::from_mock(mock_metastore), + &MetastoreServiceClient::from_mock(mock_metastore), &cluster_client, ) .await @@ -5715,7 +5716,7 @@ mod tests { let search_error = root_search( &searcher_context, search_request, - MetastoreServiceClient::from_mock(mock_metastore), + &MetastoreServiceClient::from_mock(mock_metastore), &cluster_client, ) .await diff --git a/quickwit/quickwit-search/src/service.rs b/quickwit/quickwit-search/src/service.rs index 1f89e9d0b21..52bfc846696 100644 --- a/quickwit/quickwit-search/src/service.rs +++ b/quickwit/quickwit-search/src/service.rs @@ -170,7 +170,7 @@ impl SearchService for SearchServiceImpl { let search_result = root_search( &self.searcher_context, search_request, - self.metastore.clone(), + &self.metastore, &self.cluster_client, ) .await?; @@ -231,12 +231,8 @@ impl SearchService for SearchServiceImpl { &self, list_terms_request: ListTermsRequest, ) -> crate::Result { - let search_result = root_list_terms( - &list_terms_request, - self.metastore.clone(), - &self.cluster_client, - ) - .await?; + let search_result = + root_list_terms(&list_terms_request, &self.metastore, &self.cluster_client).await?; Ok(search_result) } @@ -290,12 +286,7 @@ impl SearchService for SearchServiceImpl { &self, list_fields_req: ListFieldsRequest, ) -> crate::Result { - root_list_fields( - list_fields_req, - &self.cluster_client, - self.metastore.clone(), - ) - .await + root_list_fields(list_fields_req, &self.cluster_client, &self.metastore).await } async fn leaf_list_fields( @@ -320,7 +311,7 @@ impl SearchService for SearchServiceImpl { &self, search_request: SearchRequest, ) -> crate::Result { - let search_plan = search_plan(search_request, self.metastore.clone()).await?; + let search_plan = search_plan(search_request, &self.metastore).await?; Ok(search_plan) } diff --git a/quickwit/quickwit-search/src/tests.rs b/quickwit/quickwit-search/src/tests.rs index cba72ec3d34..94183d16c2a 100644 --- a/quickwit/quickwit-search/src/tests.rs +++ b/quickwit/quickwit-search/src/tests.rs @@ -966,7 +966,7 @@ async fn test_single_node_split_pruning_by_tags() -> anyhow::Result<()> { None, None, extract_tags_from_query(query_ast), - &mut test_sandbox.metastore(), + &test_sandbox.metastore(), ) .await?; assert!(selected_splits.is_empty()); @@ -978,7 +978,7 @@ async fn test_single_node_split_pruning_by_tags() -> anyhow::Result<()> { None, None, extract_tags_from_query(query_ast), - &mut test_sandbox.metastore(), + &test_sandbox.metastore(), ) .await?; assert_eq!(selected_splits.len(), 2); @@ -990,7 +990,7 @@ async fn test_single_node_split_pruning_by_tags() -> anyhow::Result<()> { None, None, extract_tags_from_query(query_ast), - &mut test_sandbox.metastore(), + &test_sandbox.metastore(), ) .await?; assert_eq!(selected_splits.len(), 2); @@ -1899,8 +1899,8 @@ async fn negative_cache_test_setup() -> ( .add_documents(vec![json!({"body": "hello world"})]) .await .unwrap(); - let mut metastore = test_sandbox.metastore(); - let splits_metadata = list_all_splits(vec![test_sandbox.index_uid()], &mut metastore) + let metastore = test_sandbox.metastore(); + let splits_metadata = list_all_splits(vec![test_sandbox.index_uid()], &metastore) .await .unwrap(); let splits: Vec = splits_metadata @@ -2115,8 +2115,8 @@ async fn negative_cache_ts_test_setup() -> ( docs.push(json!({"body": format!("info {i}"), "ts": start_timestamp + i})); } test_sandbox.add_documents(docs).await.unwrap(); - let mut metastore = test_sandbox.metastore(); - let splits_metadata = list_all_splits(vec![test_sandbox.index_uid()], &mut metastore) + let metastore = test_sandbox.metastore(); + let splits_metadata = list_all_splits(vec![test_sandbox.index_uid()], &metastore) .await .unwrap(); let splits: Vec = splits_metadata diff --git a/quickwit/quickwit-serve/src/elasticsearch_api/rest_handler.rs b/quickwit/quickwit-serve/src/elasticsearch_api/rest_handler.rs index 0e733081f6b..a5795de6b4e 100644 --- a/quickwit/quickwit-serve/src/elasticsearch_api/rest_handler.rs +++ b/quickwit/quickwit-serve/src/elasticsearch_api/rest_handler.rs @@ -192,8 +192,7 @@ async fn get_index_metadata( metastore: MetastoreServiceClient, ) -> Result { let index_metadata_request = IndexMetadataRequest::for_index_id(index_id); - let index_metadata = metastore - .index_metadata(index_metadata_request) + let index_metadata = MetastoreService::index_metadata(&metastore, index_metadata_request) .await? .deserialize_index_metadata()?; Ok(index_metadata) @@ -205,12 +204,12 @@ async fn get_index_metadata( pub(crate) async fn es_compat_index_mapping( index_id: String, params: IndexMappingQueryParams, - mut metastore: MetastoreServiceClient, + metastore: MetastoreServiceClient, search_service: Arc, ) -> Result { let indexes_metadata = if index_id.contains('*') || index_id.contains(',') { let patterns: Vec = index_id.split(',').map(|s| s.trim().to_string()).collect(); - resolve_index_patterns(&patterns, &mut metastore).await? + resolve_index_patterns(&patterns, &metastore).await? } else { vec![get_index_metadata(index_id.clone(), metastore).await?] }; @@ -736,9 +735,9 @@ async fn es_compat_stats( pub(crate) async fn es_compat_index_stats( index_id_patterns: Vec, - mut metastore: MetastoreServiceClient, + metastore: MetastoreServiceClient, ) -> Result { - let indexes_metadata = resolve_index_patterns(&index_id_patterns, &mut metastore).await?; + let indexes_metadata = resolve_index_patterns(&index_id_patterns, &metastore).await?; // Index uid to index id mapping let index_uid_to_index_id: HashMap = indexes_metadata @@ -751,7 +750,7 @@ pub(crate) async fn es_compat_index_stats( .map(|index_metadata| index_metadata.index_uid) .collect_vec(); // calling into the search module is not necessary, but reuses established patterns - let splits_metadata = list_all_splits(index_uids, &mut metastore).await?; + let splits_metadata = list_all_splits(index_uids, &metastore).await?; let search_response_rest: ElasticsearchStatsResponse = convert_to_es_stats_response(index_uid_to_index_id, splits_metadata); @@ -769,10 +768,10 @@ pub(crate) async fn es_compat_cat_indices( pub(crate) async fn es_compat_index_cat_indices( index_id_patterns: Vec, query_params: CatIndexQueryParams, - mut metastore: MetastoreServiceClient, + metastore: MetastoreServiceClient, ) -> Result, ElasticsearchError> { query_params.validate()?; - let indexes_metadata = resolve_index_patterns(&index_id_patterns, &mut metastore).await?; + let indexes_metadata = resolve_index_patterns(&index_id_patterns, &metastore).await?; let mut index_id_to_resp: HashMap = indexes_metadata .iter() .map(|metadata| (metadata.index_uid.to_owned(), metadata.clone().into())) @@ -785,7 +784,7 @@ pub(crate) async fn es_compat_index_cat_indices( .collect_vec(); // calling into the search module is not necessary, but reuses established patterns - list_all_splits(index_uids, &mut metastore).await? + list_all_splits(index_uids, &metastore).await? }; let search_response_rest: Vec = @@ -815,9 +814,9 @@ pub(crate) async fn es_compat_index_cat_indices( pub(crate) async fn es_compat_resolve_index( index_id_patterns: Vec, - mut metastore: MetastoreServiceClient, + metastore: MetastoreServiceClient, ) -> Result { - let indexes_metadata = resolve_index_patterns(&index_id_patterns, &mut metastore).await?; + let indexes_metadata = resolve_index_patterns(&index_id_patterns, &metastore).await?; let mut indices: Vec = indexes_metadata .into_iter() .map(|metadata| metadata.into()) diff --git a/quickwit/quickwit-serve/src/grpc.rs b/quickwit/quickwit-serve/src/grpc.rs index a2048e54f03..0eeb98ad10c 100644 --- a/quickwit/quickwit-serve/src/grpc.rs +++ b/quickwit/quickwit-serve/src/grpc.rs @@ -69,7 +69,8 @@ pub(crate) async fn start_grpc_server( let cluster_grpc_service = cluster_grpc_server(services.cluster.clone()); file_descriptor_sets.push(quickwit_proto::cluster::CLUSTER_PLANE_FILE_DESCRIPTOR_SET); - // Mount gRPC metastore service if `QuickwitService::Metastore` is enabled on node. + // Mount gRPC metastore service if this node serves either the primary metastore or a + // read-only metastore replica. let metastore_grpc_service = if let Some(metastore_server) = &services.metastore_server_opt { enabled_grpc_services.insert("metastore"); file_descriptor_sets.push(quickwit_proto::metastore::METASTORE_FILE_DESCRIPTOR_SET); diff --git a/quickwit/quickwit-serve/src/lib.rs b/quickwit/quickwit-serve/src/lib.rs index 16387366be3..ab9d67f0d15 100644 --- a/quickwit/quickwit-serve/src/lib.rs +++ b/quickwit/quickwit-serve/src/lib.rs @@ -30,6 +30,7 @@ mod indexing_api; mod ingest_api; mod jaeger_api; mod load_shield; +mod metastore; mod metrics; mod metrics_api; mod node_info_handler; @@ -65,12 +66,11 @@ use quickwit_cluster::{ }; use quickwit_common::pubsub::{EventBroker, EventSubscriptionHandle}; use quickwit_common::rate_limiter::RateLimiterSettings; -use quickwit_common::retry::RetryParams; use quickwit_common::runtimes::RuntimesConfig; use quickwit_common::tower::{ BalanceChannel, BoxFutureInfaillible, BufferLayer, Change, CircuitBreakerEvaluator, - ConstantRate, EstimateRateLayer, EventListenerLayer, GrpcMetricsLayer, LoadShedLayer, - RateLimitLayer, RetryLayer, RetryPolicy, SmaRateEstimator, TimeoutLayer, + ConstantRate, EstimateRateLayer, GrpcMetricsLayer, LoadShedLayer, RateLimitLayer, + SmaRateEstimator, TimeoutLayer, }; use quickwit_common::uri::Uri; use quickwit_common::{get_bool_from_env, spawn_named_task}; @@ -135,6 +135,7 @@ use warp::{Filter, Rejection}; pub use crate::build_info::{BuildInfo, RuntimeInfo}; pub use crate::index_api::{ListSplitsQueryParams, ListSplitsResponse}; pub use crate::ingest_api::{RestIngestResponse, RestParseFailure}; +use crate::metastore::start_metastore_service_if_needed; use crate::metrics::CIRCUIT_BREAK_TOTAL; use crate::rate_modulator::RateModulator; #[cfg(test)] @@ -154,18 +155,8 @@ const COMPACTION_SERVICE_DISCOVERY_TIMEOUT: Duration = if cfg!(any(test, feature Duration::from_secs(300) }; -const METASTORE_CLIENT_MAX_CONCURRENCY_ENV_KEY: &str = "QW_METASTORE_CLIENT_MAX_CONCURRENCY"; -const DEFAULT_METASTORE_CLIENT_MAX_CONCURRENCY: usize = 6; const DISABLE_DELETE_TASK_SERVICE_ENV_KEY: &str = "QW_DISABLE_DELETE_TASK_SERVICE"; -fn get_metastore_client_max_concurrency() -> usize { - quickwit_common::get_from_env( - METASTORE_CLIENT_MAX_CONCURRENCY_ENV_KEY, - DEFAULT_METASTORE_CLIENT_MAX_CONCURRENCY, - false, - ) -} - static CP_GRPC_CLIENT_METRICS_LAYER: LazyLock = LazyLock::new(|| GrpcMetricsLayer::new("control_plane", "client")); static CP_GRPC_SERVER_METRICS_LAYER: LazyLock = @@ -181,18 +172,13 @@ static INGEST_GRPC_CLIENT_METRICS_LAYER: LazyLock = static INGEST_GRPC_SERVER_METRICS_LAYER: LazyLock = LazyLock::new(|| GrpcMetricsLayer::new("ingest", "server")); -static METASTORE_GRPC_CLIENT_METRICS_LAYER: LazyLock = - LazyLock::new(|| GrpcMetricsLayer::new("metastore", "client")); -static METASTORE_GRPC_SERVER_METRICS_LAYER: LazyLock = - LazyLock::new(|| GrpcMetricsLayer::new("metastore", "server")); - static GRPC_INGESTER_SERVICE_TIMEOUT: Duration = Duration::from_secs(30); static GRPC_INDEXING_SERVICE_TIMEOUT: Duration = Duration::from_secs(30); -static GRPC_METASTORE_SERVICE_TIMEOUT: Duration = Duration::from_secs(10); struct QuickwitServices { pub node_config: Arc, pub cluster: Cluster, + /// Locally served metastore gRPC service, either the primary metastore or a read-only replica. pub metastore_server_opt: Option, pub metastore_client: MetastoreServiceClient, pub control_plane_server_opt: Option>, @@ -424,7 +410,7 @@ async fn start_control_plane_if_needed( node_config: &NodeConfig, cluster: &Cluster, event_broker: &EventBroker, - metastore_client: &MetastoreServiceClient, + primary_metastore_client: &MetastoreServiceClient, universe: &Universe, indexer_pool: &IndexerPool, ingester_pool: &IngesterPool, @@ -433,7 +419,7 @@ async fn start_control_plane_if_needed( check_cluster_configuration( &node_config.enabled_services, &node_config.peer_seeds, - metastore_client.clone(), + primary_metastore_client.clone(), ) .await?; @@ -446,7 +432,7 @@ async fn start_control_plane_if_needed( cluster.clone(), indexer_pool.clone(), ingester_pool.clone(), - metastore_client.clone(), + primary_metastore_client.clone(), node_config.default_index_root_uri.clone(), &node_config.ingest_api_config, ) @@ -463,8 +449,11 @@ async fn start_control_plane_if_needed( balance_channel_for_service(cluster, QuickwitService::ControlPlane).await; // If the node is a metastore, we skip this check in order to avoid a deadlock. + // A read-replica metastore node is skipped for the same reason: it only serves read-only + // metastore traffic and does not need the control plane. // If the node is a searcher, we skip this check because the searcher does not need to. if !node_config.is_service_enabled(QuickwitService::Metastore) + && !node_config.is_service_enabled(QuickwitService::MetastoreReadReplica) && node_config.enabled_services != HashSet::from([QuickwitService::Searcher]) { info!("connecting to control plane"); @@ -584,80 +573,25 @@ pub async fn serve_quickwit( let universe = Universe::new(); let grpc_config = node_config.grpc_config.clone(); - // Instantiate a metastore "server" if the `metastore` role is enabled on the node. - let metastore_server_opt: Option = - if node_config.is_service_enabled(QuickwitService::Metastore) { - let metastore: MetastoreServiceClient = metastore_resolver - .resolve(&node_config.metastore_uri) - .await - .with_context(|| { - format!( - "failed to resolve metastore uri `{}`", - node_config.metastore_uri - ) - })?; - let max_in_flight_requests = if node_config.metastore_uri.protocol().is_database() { - node_config - .metastore_configs - .find_postgres() - .map(|config| config.max_connections.get() * 2) - .unwrap_or_default() - .max(100) - } else { - 100 - }; - // These layers apply to all the RPCs of the metastore. - let shared_layer = ServiceBuilder::new() - .layer(METASTORE_GRPC_SERVER_METRICS_LAYER.clone()) - .layer(LoadShedLayer::new(max_in_flight_requests)) - .into_inner(); - let broker_layer = EventListenerLayer::new(event_broker.clone()); - let metastore = MetastoreServiceClient::tower() - .stack_layer(shared_layer) - .stack_create_index_layer(broker_layer.clone()) - .stack_delete_index_layer(broker_layer.clone()) - .stack_add_source_layer(broker_layer.clone()) - .stack_delete_source_layer(broker_layer.clone()) - .stack_toggle_source_layer(broker_layer) - .build(metastore); - Some(metastore) - } else { - None - }; - // Instantiate a metastore client, either local if available or remote otherwise. - let metastore_client: MetastoreServiceClient = - if let Some(metastore_server) = &metastore_server_opt { - metastore_server.clone() - } else { - info!("connecting to metastore"); + // Instantiate the local metastore gRPC server for this node (the primary metastore, a + // read-only replica, or none; the `metastore` and `metastore_read_replica` roles are mutually + // exclusive). + let local_metastore_server = + start_metastore_service_if_needed(&node_config, &metastore_resolver, &event_broker) + .await + .context("failed to start metastore service")?; - let balance_channel = - balance_channel_for_service(&cluster, QuickwitService::Metastore).await; + let primary_metastore_client = local_metastore_server + .resolve_primary_client(&cluster, &node_config) + .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) - }; // Instantiate a control plane server if the `control-plane` role is enabled on the node. // Otherwise, instantiate a control plane client. let (control_plane_server_opt, control_plane_client) = start_control_plane_if_needed( &node_config, &cluster, &event_broker, - &metastore_client, + &primary_metastore_client, &universe, &indexer_pool, &ingester_pool, @@ -665,11 +599,12 @@ pub async fn serve_quickwit( .await .context("failed to start control plane service")?; - // Set up the "control plane proxy" for the metastore. - let metastore_through_control_plane = MetastoreServiceClient::new(ControlPlaneMetastore::new( - control_plane_client.clone(), - metastore_client.clone(), - )); + // Set up the "control plane proxy" for the primary metastore. + let primary_metastore_through_control_plane = + MetastoreServiceClient::new(ControlPlaneMetastore::new( + control_plane_client.clone(), + primary_metastore_client.clone(), + )); // Setup ingest service v1. let ingest_service = start_ingest_client_if_needed(&node_config, &universe, &cluster) @@ -681,7 +616,7 @@ pub async fn serve_quickwit( &node_config, &cluster, &universe, - &metastore_client, + &primary_metastore_client, ) .await .context("failed to initialize compaction service client")?; @@ -703,7 +638,7 @@ pub async fn serve_quickwit( &node_config, runtimes_config.num_threads_blocking, cluster.clone(), - metastore_through_control_plane.clone(), + primary_metastore_through_control_plane.clone(), ingester_pool.clone(), storage_resolver.clone(), event_broker.clone(), @@ -750,7 +685,7 @@ pub async fn serve_quickwit( // Any node can serve index management requests (create/update/delete index, add/remove source, // etc.), so we always instantiate an index manager. let mut index_manager = IndexManager::new( - metastore_through_control_plane.clone(), + primary_metastore_through_control_plane.clone(), storage_resolver.clone(), ); @@ -825,12 +760,30 @@ pub async fn serve_quickwit( )) }; + let read_replica_metastore_client_opt = local_metastore_server + .resolve_read_only_client(&cluster, &node_config) + .await?; + + // Search uses the read replica when configured, and the primary otherwise. + let search_metastore_kind = if read_replica_metastore_client_opt.is_some() { + "read_replica" + } else { + "primary" + }; + let search_metastore_client = read_replica_metastore_client_opt + .clone() + .unwrap_or_else(|| primary_metastore_through_control_plane.clone()); + info!( + metastore_kind = search_metastore_kind, + "configured search metastore client" + ); + let (search_job_placer, search_service, searcher_pool) = setup_searcher( &node_config, cluster.change_stream(), // search remains available without a control plane because not all // metastore RPCs are proxied - metastore_through_control_plane.clone(), + search_metastore_client.clone(), storage_resolver.clone(), searcher_context, ) @@ -847,7 +800,7 @@ pub async fn serve_quickwit( let datafusion_session_builder = datafusion_api::setup::build_datafusion_session_builder( &node_config, cluster.change_stream(), - metastore_through_control_plane.clone(), + search_metastore_client, storage_resolver.clone(), )?; // The search job placer owns a clone of this pool; the local binding is not @@ -878,7 +831,7 @@ pub async fn serve_quickwit( let janitor_service = start_janitor_service( &universe, &node_config, - metastore_through_control_plane.clone(), + primary_metastore_through_control_plane.clone(), search_job_placer, storage_resolver.clone(), event_broker.clone(), @@ -909,7 +862,7 @@ pub async fn serve_quickwit( cluster.self_node_id(), compaction_client, &node_config.compactor_config, - metastore_client.clone(), + primary_metastore_client.clone(), storage_resolver.clone(), split_cache, event_broker.clone(), @@ -958,8 +911,8 @@ pub async fn serve_quickwit( let quickwit_services: Arc = Arc::new(QuickwitServices { node_config: Arc::new(node_config), cluster: cluster.clone(), - metastore_server_opt, - metastore_client: metastore_through_control_plane.clone(), + metastore_server_opt: local_metastore_server.client().cloned(), + metastore_client: primary_metastore_through_control_plane.clone(), control_plane_server_opt, control_plane_client, _local_shards_update_listener_handle_opt: local_shards_update_listener_handle_opt, @@ -1063,7 +1016,8 @@ pub async fn serve_quickwit( spawn_named_task( node_readiness_reporting_task( cluster.clone(), - metastore_through_control_plane, + primary_metastore_through_control_plane, + read_replica_metastore_client_opt, ingester_opt.clone(), grpc_readiness_signal_rx, rest_readiness_signal_rx, @@ -1624,12 +1578,23 @@ fn with_arg(arg: T) -> impl Filter, ingester_opt: Option, grpc_readiness_signal_rx: oneshot::Receiver<()>, rest_readiness_signal_rx: oneshot::Receiver<()>, health_reporter: HealthReporter, ) { + // When a read replica metastore is configured, node readiness follows the replica only. + // This keeps search/read traffic available if the primary metastore is down. Write-capable + // index-management APIs are still routed to the primary metastore and may fail while this node + // remains ready. + let (metastore, metastore_kind) = match read_replica_metastore_opt { + Some(read_replica_metastore) => (read_replica_metastore, "read_replica"), + None => (primary_metastore, "primary"), + }; + info!(metastore_kind, "configured metastore readiness dependency"); + let mut node_ready = false; cluster.set_self_node_readiness(node_ready).await; // Set the initial health status to `NotServing` with "" meaning all services, as per @@ -1655,13 +1620,13 @@ async fn node_readiness_reporting_task( loop { interval.tick().await; - let metastore_is_available = match metastore.check_connectivity().await { + let metastore_available = match metastore.check_connectivity().await { Ok(()) => { debug!(metastore_endpoints=?metastore.endpoints(), "metastore service is available"); true } Err(error) => { - warn!(metastore_endpoints=?metastore.endpoints(), error=?error, "metastore service is unavailable"); + debug!(metastore_endpoints=?metastore.endpoints(), error=?error, "metastore service is unavailable"); false } }; @@ -1680,7 +1645,7 @@ async fn node_readiness_reporting_task( } else { true }; - let is_ready = metastore_is_available && ingester_is_available; + let is_ready = metastore_available && ingester_is_available; let new_node_ready = if is_ready { consecutive_readiness_failures = 0; true @@ -1693,6 +1658,13 @@ async fn node_readiness_reporting_task( if new_node_ready != node_ready { node_ready = new_node_ready; + if !node_ready && !metastore_available { + warn!( + metastore_kind, + metastore_endpoints = ?metastore.endpoints(), + "metastore unavailability caused node readiness to decrease" + ); + } cluster.set_self_node_readiness(node_ready).await; let serving_status = if node_ready { @@ -1755,7 +1727,7 @@ async fn check_cluster_configuration( mod tests { use std::sync::{Arc, Mutex}; - use anyhow::bail; + use anyhow::{bail, ensure}; use quickwit_cluster::{ChitchatTransport, ClusterNode, create_cluster_for_test}; use quickwit_common::uri::Uri; use quickwit_common::{ServiceStream, assert_eventually}; @@ -1774,6 +1746,23 @@ mod tests { use super::*; + fn metastore_readiness_client( + readiness_rx: watch::Receiver, + uri: &'static str, + ) -> MetastoreServiceClient { + let mut mock_metastore = MockMetastoreService::new(); + mock_metastore + .expect_check_connectivity() + .returning(move || { + ensure!(*readiness_rx.borrow(), "metastore `{uri}` not ready"); + Ok(()) + }); + mock_metastore + .expect_endpoints() + .return_const(vec![Uri::for_test(uri)]); + MetastoreServiceClient::from_mock(mock_metastore) + } + #[tokio::test] async fn test_check_cluster_configuration() { let services = HashSet::from_iter([QuickwitService::Metastore]); @@ -1816,16 +1805,20 @@ mod tests { let mut failures_to_inject = metastore_failures_to_inject_clone.lock().unwrap(); if *failures_to_inject > 0 { *failures_to_inject -= 1; - bail!("metastore transiently not ready"); + bail!("metastore `ram:///metastore` transiently not ready"); } drop(failures_to_inject); - if *metastore_readiness_rx.borrow() { - Ok(()) - } else { - bail!("metastore not ready") - } + ensure!( + *metastore_readiness_rx.borrow(), + "metastore `ram:///metastore` not ready" + ); + Ok(()) }); + mock_metastore + .expect_endpoints() + .return_const(vec![Uri::for_test("ram:///metastore")]); + let mock_metastore = MetastoreServiceClient::from_mock(mock_metastore); let (ingester_status_tx, ingester_status_rx) = watch::channel(IngesterStatus::Initializing); let mut mock_ingester = MockIngesterService::new(); mock_ingester @@ -1867,7 +1860,8 @@ mod tests { tokio::spawn(node_readiness_reporting_task( cluster.clone(), - MetastoreServiceClient::from_mock(mock_metastore), + mock_metastore, + None, Some(mock_ingester), grpc_readiness_signal_rx, rest_readiness_signal_rx, @@ -1905,6 +1899,44 @@ mod tests { assert_eq!(response.status(), ServingStatus::NotServing.into()); } + #[tokio::test] + async fn test_readiness_uses_read_replica_without_requiring_primary() { + let transport = ChitchatTransport::default(); + let cluster = create_cluster_for_test(Vec::new(), &[], &transport, false) + .await + .unwrap(); + let (replica_readiness_tx, replica_readiness_rx) = watch::channel(false); + let mut primary_metastore = MockMetastoreService::new(); + primary_metastore.expect_check_connectivity().times(0); + let primary_metastore = MetastoreServiceClient::from_mock(primary_metastore); + let replica_metastore = + metastore_readiness_client(replica_readiness_rx, "ram:///replica-metastore"); + let (grpc_readiness_trigger_tx, grpc_readiness_signal_rx) = oneshot::channel(); + let (rest_readiness_trigger_tx, rest_readiness_signal_rx) = oneshot::channel(); + let (health_reporter, _health_service) = health_reporter(); + + tokio::spawn(node_readiness_reporting_task( + cluster.clone(), + primary_metastore, + Some(replica_metastore), + None::, + grpc_readiness_signal_rx, + rest_readiness_signal_rx, + health_reporter, + )); + grpc_readiness_trigger_tx.send(()).unwrap(); + rest_readiness_trigger_tx.send(()).unwrap(); + + tokio::time::sleep(READINESS_REPORTING_INTERVAL * 3).await; + assert!(!cluster.is_self_node_ready().await); + + replica_readiness_tx.send(true).unwrap(); + assert_eventually!(cluster.is_self_node_ready().await); + + replica_readiness_tx.send(false).unwrap(); + assert_eventually!(!cluster.is_self_node_ready().await); + } + #[tokio::test] async fn test_setup_indexer_pool() { let universe = Universe::with_accelerated_time(); diff --git a/quickwit/quickwit-serve/src/metastore.rs b/quickwit/quickwit-serve/src/metastore.rs new file mode 100644 index 00000000000..995a06c0d8c --- /dev/null +++ b/quickwit/quickwit-serve/src/metastore.rs @@ -0,0 +1,242 @@ +// Copyright 2021-Present Datadog, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::sync::LazyLock; +use std::time::Duration; + +use anyhow::{Context, bail, ensure}; +use bytesize::ByteSize; +use quickwit_cluster::Cluster; +use quickwit_common::pubsub::EventBroker; +use quickwit_common::retry::RetryParams; +use quickwit_common::tower::{ + EventListenerLayer, GrpcMetricsLayer, LoadShedLayer, RetryLayer, RetryPolicy, TimeoutLayer, +}; +use quickwit_common::uri::Uri; +use quickwit_config::NodeConfig; +use quickwit_config::service::QuickwitService; +use quickwit_metastore::MetastoreResolver; +use quickwit_proto::metastore::MetastoreServiceClient; +use tower::ServiceBuilder; +use tracing::info; + +use crate::balance_channel_for_service; + +const METASTORE_CLIENT_MAX_CONCURRENCY_ENV_KEY: &str = "QW_METASTORE_CLIENT_MAX_CONCURRENCY"; +const DEFAULT_METASTORE_CLIENT_MAX_CONCURRENCY: usize = 6; +const GRPC_METASTORE_SERVICE_TIMEOUT: Duration = Duration::from_secs(10); + +static METASTORE_GRPC_CLIENT_METRICS_LAYER: LazyLock = + LazyLock::new(|| GrpcMetricsLayer::new("metastore", "client")); +static METASTORE_GRPC_SERVER_METRICS_LAYER: LazyLock = + LazyLock::new(|| GrpcMetricsLayer::new("metastore", "server")); + +fn get_metastore_client_max_concurrency() -> usize { + quickwit_common::get_from_env( + METASTORE_CLIENT_MAX_CONCURRENCY_ENV_KEY, + DEFAULT_METASTORE_CLIENT_MAX_CONCURRENCY, + false, + ) +} + +/// The metastore gRPC server this node serves locally — or `NotServed` if it serves none. +/// +/// A node serves *either* a writable primary metastore (`metastore` role) *or* a read-only +/// replica (`metastore_read_replica` role), never both: the two roles are mutually exclusive, as +/// enforced by `validate_metastore_read_replica` at config load. +pub(super) enum LocalMetastoreServer { + /// Writable primary metastore. + Primary(MetastoreServiceClient), + /// Read-only metastore replica. + ReadReplica(MetastoreServiceClient), + /// This node does not serve a metastore locally (it may still be a client of a remote one). + NotServed, +} + +impl LocalMetastoreServer { + /// The locally served gRPC metastore client, if this node serves one. + pub(super) fn client(&self) -> Option<&MetastoreServiceClient> { + match self { + LocalMetastoreServer::Primary(client) | LocalMetastoreServer::ReadReplica(client) => { + Some(client) + } + LocalMetastoreServer::NotServed => None, + } + } + + /// Resolves the primary (writable) metastore client for this node. + pub(super) async fn resolve_primary_client( + &self, + cluster: &Cluster, + node_config: &NodeConfig, + ) -> anyhow::Result { + match self { + LocalMetastoreServer::Primary(metastore_server) => Ok(metastore_server.clone()), + LocalMetastoreServer::ReadReplica(_) | LocalMetastoreServer::NotServed => { + Self::build_metastore_client( + cluster, + QuickwitService::Metastore, + node_config.grpc_config.max_message_size, + ) + .await + } + } + } + + /// Resolves the read-only metastore client for this node. + pub(super) async fn resolve_read_only_client( + &self, + cluster: &Cluster, + node_config: &NodeConfig, + ) -> anyhow::Result> { + let should_use_remote_read_replica = node_config + .is_service_enabled(QuickwitService::Searcher) + && node_config.searcher_config.use_metastore_read_replica; + + match self { + LocalMetastoreServer::ReadReplica(metastore_server) => { + Ok(Some(metastore_server.clone())) + } + LocalMetastoreServer::Primary(_) | LocalMetastoreServer::NotServed + if should_use_remote_read_replica => + { + let read_replica_client = Self::build_metastore_client( + cluster, + QuickwitService::MetastoreReadReplica, + node_config.grpc_config.max_message_size, + ) + .await?; + Ok(Some(read_replica_client)) + } + LocalMetastoreServer::Primary(_) | LocalMetastoreServer::NotServed => Ok(None), + } + } + + async fn build_metastore_client( + cluster: &Cluster, + service: QuickwitService, + max_message_size: ByteSize, + ) -> anyhow::Result { + info!(%service, "connecting to {service} service"); + + let balance_channel = balance_channel_for_service(cluster, service).await; + + ensure!( + balance_channel + .wait_for(Duration::from_secs(300), |connections| { + !connections.is_empty() + }) + .await, + "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(balance_channel, max_message_size, None)) + } + + fn metastore_max_in_flight_requests(node_config: &NodeConfig, uri: &Uri) -> usize { + if uri.protocol().is_database() { + node_config + .metastore_configs + .find_postgres() + .map(|config| config.max_connections.get() * 2) + .unwrap_or_default() + .max(100) + } else { + 100 + } + } +} + +pub(super) async fn start_metastore_service_if_needed( + node_config: &NodeConfig, + metastore_resolver: &MetastoreResolver, + event_broker: &EventBroker, +) -> anyhow::Result { + // Instantiate a primary metastore server if the `metastore` role is enabled on the node. + if node_config.is_service_enabled(QuickwitService::Metastore) { + info!( + metastore_kind = "primary", + "starting local metastore service" + ); + let metastore: MetastoreServiceClient = metastore_resolver + .resolve(&node_config.metastore_uri) + .await + .with_context(|| { + format!( + "failed to resolve metastore uri `{}`", + node_config.metastore_uri + ) + })?; + // These layers apply to all the RPCs of the metastore. + let shared_layer = ServiceBuilder::new() + .layer(METASTORE_GRPC_SERVER_METRICS_LAYER.clone()) + .layer(LoadShedLayer::new( + LocalMetastoreServer::metastore_max_in_flight_requests( + node_config, + &node_config.metastore_uri, + ), + )) + .into_inner(); + let broker_layer = EventListenerLayer::new(event_broker.clone()); + let metastore = MetastoreServiceClient::tower() + .stack_layer(shared_layer) + .stack_create_index_layer(broker_layer.clone()) + .stack_delete_index_layer(broker_layer.clone()) + .stack_add_source_layer(broker_layer.clone()) + .stack_delete_source_layer(broker_layer.clone()) + .stack_toggle_source_layer(broker_layer) + .build(metastore); + return Ok(LocalMetastoreServer::Primary(metastore)); + } + // Instantiate a read-only metastore replica server if the `metastore_read_replica` role is + // enabled on the node. + if node_config.is_service_enabled(QuickwitService::MetastoreReadReplica) { + info!( + metastore_kind = "read_replica", + "starting local metastore service" + ); + let Some(read_replica_uri) = &node_config.metastore_read_replica_uri else { + bail!( + "`metastore_read_replica_uri` must be set when the `metastore_read_replica` role \ + is enabled" + ); + }; + let metastore: MetastoreServiceClient = metastore_resolver + .resolve_read_only(read_replica_uri) + .await + .with_context(|| { + format!("failed to resolve metastore read replica uri `{read_replica_uri}`") + })?; + let shared_layer = ServiceBuilder::new() + .layer(METASTORE_GRPC_SERVER_METRICS_LAYER.clone()) + .layer(LoadShedLayer::new( + LocalMetastoreServer::metastore_max_in_flight_requests( + node_config, + read_replica_uri, + ), + )) + .into_inner(); + let metastore = MetastoreServiceClient::tower() + .stack_layer(shared_layer) + .build(metastore); + return Ok(LocalMetastoreServer::ReadReplica(metastore)); + } + Ok(LocalMetastoreServer::NotServed) +}