Apply node config overrides before validation

Author: shuheiktgw

quickwit/quickwit-cli/src/lib.rs

@@ -26,8 +26,8 @@ use quickwit_common::runtimes::RuntimesConfig;
 use quickwit_common::uri::Uri;
 use quickwit_config::service::QuickwitService;
 use quickwit_config::{
-    ConfigFormat, DEFAULT_QW_CONFIG_PATH, MetastoreConfigs, NodeConfig, SourceConfig,
-    StorageConfigs,
+    ConfigFormat, DEFAULT_QW_CONFIG_PATH, MetastoreConfigs, NodeConfig, NodeConfigOverrides,
+    SourceConfig, StorageConfigs,
 };
 use quickwit_indexing::check_source_connectivity;
 use quickwit_metastore::{IndexMetadataResponseExt, MetastoreResolver};
@@ -229,12 +229,15 @@ 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<NodeConfig> {
+async fn load_node_config(
+    config_uri: &Uri,
+    overrides: NodeConfigOverrides,
+) -> anyhow::Result<NodeConfig> {
     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())
+    let config = NodeConfig::load(config_format, config_content.as_slice(), overrides)
         .await
         .with_context(|| format!("failed to parse node config `{config_uri}`"))?;
     info!(config_uri=%config_uri, config=?config, "loaded node config");

quickwit/quickwit-cli/src/service.rs

@@ -22,6 +22,7 @@ use futures::future::select;
 use itertools::Itertools;
 use quickwit_common::runtimes::RuntimesConfig;
 use quickwit_common::uri::Uri;
+use quickwit_config::NodeConfigOverrides;
 use quickwit_config::service::QuickwitService;
 use quickwit_serve::tcp_listener::DefaultTcpListenerResolver;
 use quickwit_serve::{BuildInfo, EnvFilterReloadFn, reload_tls_cert, serve_quickwit};
@@ -117,14 +118,19 @@ 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 node_config = load_node_config(
+            &self.config_uri,
+            NodeConfigOverrides {
+                enabled_services: self.services.clone(),
+            },
+        )
+        .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);
         }
         // TODO move in serve quickwit?
         let runtimes_config = RuntimesConfig::default();

quickwit/quickwit-cli/src/source.rs

@@ -614,6 +614,7 @@ fn flatten_json(value: JsonValue) -> Vec<(String, JsonValue)> {
         }
         acc.push((root, value))
     }
+    acc.sort_by(|left, right| left.0.cmp(&right.0));
     acc
 }
 

quickwit/quickwit-cli/src/tool.rs

@@ -31,8 +31,8 @@ use quickwit_common::runtimes::RuntimesConfig;
 use quickwit_common::uri::Uri;
 use quickwit_config::service::QuickwitService;
 use quickwit_config::{
-    CLI_SOURCE_ID, IndexerConfig, NodeConfig, SourceConfig, SourceInputFormat, SourceParams,
-    TransformConfig, VecSourceParams,
+    CLI_SOURCE_ID, IndexerConfig, NodeConfig, NodeConfigOverrides, SourceConfig, SourceInputFormat,
+    SourceParams, TransformConfig, VecSourceParams,
 };
 use quickwit_index_management::{IndexService, clear_cache_directory};
 use quickwit_indexing::BoxedPipelineHandle;
@@ -400,7 +400,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, NodeConfigOverrides::default()).await?;
     let (storage_resolver, metastore_resolver) =
         get_resolvers(&config.storage_configs, &config.metastore_configs);
     let mut metastore = metastore_resolver.resolve(&config.metastore_uri).await?;
@@ -529,7 +529,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, NodeConfigOverrides::default()).await?;
     let (storage_resolver, metastore_resolver) =
         get_resolvers(&config.storage_configs, &config.metastore_configs);
     let metastore: MetastoreServiceClient =
@@ -567,7 +567,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, NodeConfigOverrides::default()).await?;
     let (storage_resolver, metastore_resolver) =
         get_resolvers(&config.storage_configs, &config.metastore_configs);
     let mut metastore = metastore_resolver.resolve(&config.metastore_uri).await?;
@@ -654,7 +654,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, NodeConfigOverrides::default()).await?;
     let (storage_resolver, metastore_resolver) =
         get_resolvers(&config.storage_configs, &config.metastore_configs);
     let metastore = metastore_resolver.resolve(&config.metastore_uri).await?;
@@ -784,7 +784,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, NodeConfigOverrides::default()).await?;
     let (storage_resolver, metastore_resolver) =
         get_resolvers(&config.storage_configs, &config.metastore_configs);
     let metastore = metastore_resolver.resolve(&config.metastore_uri).await?;

quickwit/quickwit-config/src/lib.rs

@@ -76,7 +76,8 @@ pub use crate::metastore_config::{
 pub use crate::node_config::{
     CacheConfig, CachePolicy, DEFAULT_QW_CONFIG_PATH, GrpcConfig, HealthConfig, IndexerConfig,
     IngestApiConfig, JaegerConfig, KeepAliveConfig, LambdaConfig, LambdaDeployConfig, NodeConfig,
-    RestConfig, SearcherConfig, SplitCacheLimits, StorageTimeoutPolicy, TlsConfig,
+    NodeConfigOverrides, RestConfig, SearcherConfig, SplitCacheLimits, StorageTimeoutPolicy,
+    TlsConfig,
 };
 pub use crate::serde_utils::HumanDuration;
 use crate::source_config::serialize::{SourceConfigV0_7, SourceConfigV0_8, VersionedSourceConfig};

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

@@ -833,15 +833,26 @@ pub struct NodeConfig {
     pub jaeger_config: JaegerConfig,
 }
 
+/// Values supplied outside the config file and environment, applied before validation.
+#[derive(Clone, Debug, Default, Eq, PartialEq)]
+pub struct NodeConfigOverrides {
+    pub enabled_services: Option<HashSet<QuickwitService>>,
+}
+
 impl NodeConfig {
     pub fn is_service_enabled(&self, service: QuickwitService) -> bool {
         self.enabled_services.contains(&service)
     }
 
     /// 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> {
+    pub async fn load(
+        config_format: ConfigFormat,
+        config_content: &[u8],
+        overrides: NodeConfigOverrides,
+    ) -> anyhow::Result<Self> {
         let env_vars = env::vars().collect::<HashMap<_, _>>();
-        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, overrides).await?;
         if !config.data_dir_path.try_exists()? {
             bail!(
                 "data dir `{}` does not exist",