Use an enum rather than JSON value for storing source params (#1058)

* Use an enum rather than json value for storing source params

* Fix backward compatibility tests and use absolute path in FileSourceParams.

* Renamed SourceType -> SourceParams

* Added a SourceParams::validate function

Co-authored-by: François Massot <[email protected]>
Co-authored-by: Paul Masurel <[email protected]>

Author: 3 people

quickwit-cli/src/index.rs

@@ -31,12 +31,12 @@ use itertools::Itertools;
 use quickwit_actors::{ActorHandle, ObservationType};
 use quickwit_common::uri::Uri;
 use quickwit_common::{run_checklist, GREEN_COLOR};
-use quickwit_config::{IndexConfig, IndexerConfig, SourceConfig};
+use quickwit_config::{IndexConfig, IndexerConfig, SourceConfig, SourceParams};
 use quickwit_core::{create_index, delete_index, garbage_collect_index, reset_index};
 use quickwit_doc_mapper::tag_pruning::match_tag_field_name;
 use quickwit_indexing::actors::{IndexingPipeline, IndexingServer};
 use quickwit_indexing::models::IndexingStatistics;
-use quickwit_indexing::source::{FileSourceParams, INGEST_SOURCE_ID};
+use quickwit_indexing::source::INGEST_SOURCE_ID;
 use quickwit_metastore::{quickwit_metastore_uri_resolver, IndexMetadata, Split, SplitState};
 use quickwit_proto::{SearchRequest, SearchResponse};
 use quickwit_search::{single_node_search, SearchResponseRest};
@@ -625,16 +625,14 @@ pub async fn ingest_docs_cli(args: IngestDocsArgs) -> anyhow::Result<()> {
 
     let config = load_quickwit_config(args.config_uri, args.data_dir).await?;
 
-    let file_source_params = if let Some(filepath) = args.input_path_opt.as_ref() {
-        FileSourceParams::for_file(filepath.clone())
+    let source_params = if let Some(filepath) = args.input_path_opt.as_ref() {
+        SourceParams::file(filepath)
     } else {
-        FileSourceParams::stdin()
+        SourceParams::stdin()
     };
-    let params = serde_json::to_value(file_source_params)?;
     let source = SourceConfig {
         source_id: INGEST_SOURCE_ID.to_string(),
-        source_type: "file".to_string(),
-        params,
+        source_params,
     };
     run_index_checklist(&config.metastore_uri, &args.index_id, Some(&source)).await?;
     let metastore_uri_resolver = quickwit_metastore_uri_resolver();

quickwit-cli/src/source.rs

@@ -21,12 +21,12 @@ use anyhow::{bail, Context};
 use clap::ArgMatches;
 use itertools::Itertools;
 use quickwit_common::uri::Uri;
-use quickwit_config::SourceConfig;
+use quickwit_config::{SourceConfig, SourceParams};
 use quickwit_indexing::check_source_connectivity;
 use quickwit_metastore::checkpoint::SourceCheckpoint;
 use quickwit_metastore::{quickwit_metastore_uri_resolver, IndexMetadata};
 use quickwit_storage::load_file;
-use serde_json::Value;
+use serde_json::{Map, Value};
 use tabled::{Table, Tabled};
 
 use crate::{load_quickwit_config, make_table};
@@ -184,12 +184,17 @@ async fn add_source_cli(args: AddSourceArgs) -> anyhow::Result<()> {
         .resolve(&config.metastore_uri)
         .await?;
     let params = sniff_params(&args.params).await?;
+    let mut source_params_json: Map<String, Value> = Map::new();
+    source_params_json.insert("source_type".to_string(), Value::String(args.source_type));
+    source_params_json.insert("params".to_string(), Value::Object(params));
+    let source_params: SourceParams = serde_json::from_value(Value::Object(source_params_json))?;
     let source = SourceConfig {
         source_id: args.source_id.clone(),
-        source_type: args.source_type,
-        params,
+        source_params,
     };
+    source.validate()?;
     check_source_connectivity(&source).await?;
+
     metastore.add_source(&args.index_id, source).await?;
     println!(
         "Source `{}` successfully created for index `{}`.",
@@ -244,12 +249,12 @@ where
         .with_context(|| format!("Source `{}` does not exist.", source_id))?;
 
     let source_rows = vec![SourceRow {
-        source_id: source.source_id,
-        source_type: source.source_type,
+        source_id: source.source_id.clone(),
+        source_type: source.source_type().to_string(),
     }];
     let source_table = make_table("Source", source_rows);
 
-    let params_rows = flatten_json(source.params)
+    let params_rows = flatten_json(source.params())
         .into_iter()
         .map(|(key, value)| ParamsRow { key, value })
         .sorted_by(|left, right| left.key.cmp(&right.key));
@@ -279,8 +284,8 @@ where I: IntoIterator<Item = SourceConfig> {
     let rows = sources
         .into_iter()
         .map(|source| SourceRow {
+            source_type: source.source_type().to_string(),
             source_id: source.source_id,
-            source_type: source.source_type,
         })
         .sorted_by(|left, right| left.source_id.cmp(&right.source_id));
     make_table("Sources", rows)
@@ -346,14 +351,14 @@ fn flatten_json(value: Value) -> Vec<(String, Value)> {
 
 /// Tries to read a JSON object from a string, assuming the string is an inline JSON object or a
 /// path to a file holding a JSON object.
-async fn sniff_params(params: &str) -> anyhow::Result<Value> {
-    if let Ok(object @ Value::Object(_)) = serde_json::from_str(params) {
-        return Ok(object);
+async fn sniff_params(params: &str) -> anyhow::Result<Map<String, Value>> {
+    if let Ok(Value::Object(values)) = serde_json::from_str(params) {
+        return Ok(values);
     }
     let params_uri = Uri::try_new(params)?;
     let params_bytes = load_file(&params_uri).await?;
-    if let Ok(object @ Value::Object(_)) = serde_json::from_slice(params_bytes.as_slice()) {
-        return Ok(object);
+    if let Ok(Value::Object(values)) = serde_json::from_slice(params_bytes.as_slice()) {
+        return Ok(values);
     }
     bail!("Failed to parse JSON object from `{}`.", params)
 }
@@ -402,10 +407,10 @@ mod tests {
         sniff_params("0").await.unwrap_err();
         sniff_params("[]").await.unwrap_err();
 
-        assert!(matches!(
-            sniff_params(r#"{"foo": 0}"#).await.unwrap(),
-            Value::Object(map) if map.contains_key("foo")
-        ));
+        assert!(sniff_params(r#"{"foo": 0}"#)
+            .await
+            .unwrap()
+            .contains_key("foo"));
 
         let storage = quickwit_storage_uri_resolver()
             .resolve("ram:///tmp")
@@ -416,10 +421,10 @@ mod tests {
             .await
             .unwrap();
 
-        assert!(matches!(
-            sniff_params("ram:///tmp/params.json").await.unwrap(),
-            Value::Object(map) if map.contains_key("bar")
-        ));
+        assert!(sniff_params("ram:///tmp/params.json")
+            .await
+            .unwrap()
+            .contains_key("bar"));
     }
 
     #[test]
@@ -520,8 +525,7 @@ mod tests {
             .collect();
         let sources = vec![SourceConfig {
             source_id: "foo-source".to_string(),
-            source_type: "file".to_string(),
-            params: json!({"filepath": "path/to/file"}),
+            source_params: SourceParams::file("path/to/file"),
         }];
         let expected_source = vec![SourceRow {
             source_id: "foo-source".to_string(),
@@ -584,13 +588,11 @@ mod tests {
         let sources = [
             SourceConfig {
                 source_id: "foo-source".to_string(),
-                source_type: "file".to_string(),
-                params: json!({}),
+                source_params: SourceParams::stdin(),
             },
             SourceConfig {
                 source_id: "bar-source".to_string(),
-                source_type: "file".to_string(),
-                params: json!({}),
+                source_params: SourceParams::stdin(),
             },
         ];
         let expected_sources = [

quickwit-cli/tests/cli.rs

@@ -214,6 +214,7 @@ fn test_cmd_ingest_simple() -> Result<()> {
     .stdout(predicate::str::contains("Indexed"))
     .stdout(predicate::str::contains("documents in"))
     .stdout(predicate::str::contains("Now, you can query the index"));
+    println!("piped input");
     Ok(())
 }
 

quickwit-cli/update_on_config.md

@@ -0,0 +1 @@
+This does not directly reflect the overall memory usage of `quickwit index ingest`, but doubling this value should give a fair approximation.

quickwit-config/resources/tests/index_config/hdfs-logs.json

@@ -66,8 +66,10 @@
             "source_id": "hdfs-logs-kafka-source",
             "source_type": "kafka",
             "params": {
-                "bootstrap.servers": "host:9092",
-                "topic": "cloudera-cluster-logs"
+                "topic": "cloudera-cluster-logs",
+                "client_params": {
+                    "bootstrap.servers": "host:9092"
+                }
             }
         },
         {

quickwit-config/resources/tests/index_config/hdfs-logs.toml

@@ -36,7 +36,7 @@ default_search_fields = [ "severity_text", "body" ]
 [[sources]]
 source_id = "hdfs-logs-kafka-source"
 source_type = "kafka"
-params = { "bootstrap.servers" = "host:9092", "topic" = "cloudera-cluster-logs" }
+params = { "topic" = "cloudera-cluster-logs", "client_params" = { "bootstrap.servers" = "host:9092" } }
 
 [[sources]]
 source_id = "hdfs-logs-kinesis-source"

quickwit-config/resources/tests/index_config/hdfs-logs.yaml

@@ -48,8 +48,9 @@ sources:
   - source_id: hdfs-logs-kafka-source
     source_type: kafka
     params:
-      bootstrap.servers: host:9092
       topic: cloudera-cluster-logs
+      client_params:
+        bootstrap.servers: host:9092
 
   - source_id: hdfs-logs-kinesis-source
     source_type: kinesis

quickwit-config/src/index_config.rs

@@ -32,12 +32,15 @@ use quickwit_doc_mapper::{
 };
 use serde::{Deserialize, Serialize};
 
+use crate::source_config::SourceConfig;
+
 // Note(fmassot): `DocMapping` is a struct only used for
 // serialization/deserialization of `DocMapper` parameters.
 // This is partly a duplicate of the `DocMapper` and can
 // be viewed as a temporary hack for 0.2 release before
 // refactoring.
 #[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
 pub struct DocMapping {
     pub field_mappings: Vec<FieldMappingEntry>,
     #[serde(default)]
@@ -47,6 +50,7 @@ pub struct DocMapping {
 }
 
 #[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
+#[serde(deny_unknown_fields)]
 pub struct IndexingResources {
     #[serde(default = "IndexingResources::default_num_threads")]
     pub num_threads: usize,
@@ -81,6 +85,7 @@ impl Default for IndexingResources {
 }
 
 #[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
+#[serde(deny_unknown_fields)]
 pub struct MergePolicy {
     #[serde(default = "MergePolicy::default_demux_factor")]
     pub demux_factor: usize,
@@ -119,6 +124,7 @@ fn is_false(val: &bool) -> bool {
 }
 
 #[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
+#[serde(deny_unknown_fields)]
 pub struct IndexingSettings {
     #[serde(default, skip_serializing_if = "is_false")]
     pub demux_enabled: bool,
@@ -195,19 +201,14 @@ impl Default for IndexingSettings {
 }
 
 #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Default)]
+#[serde(deny_unknown_fields)]
 pub struct SearchSettings {
     #[serde(default)]
     pub default_search_fields: Vec<String>,
 }
 
-#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
-pub struct SourceConfig {
-    pub source_id: String,
-    pub source_type: String,
-    pub params: serde_json::Value,
-}
-
 #[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
 pub struct IndexConfig {
     pub version: usize,
     pub index_id: String,
@@ -275,6 +276,10 @@ impl IndexConfig {
             bail!("Index config contains duplicate sources.")
         }
 
+        for source in self.sources.iter() {
+            source.validate()?;
+        }
+
         // Validation is made by building the doc mapper.
         // Note: this needs a deep refactoring to separate the doc mapping configuration,
         // and doc mapper implementations.
@@ -319,9 +324,9 @@ pub fn build_doc_mapper(
 
 #[cfg(test)]
 mod tests {
-    use serde_json::json;
 
     use super::*;
+    use crate::SourceParams;
 
     fn get_resource_path(resource_filename: &str) -> String {
         format!(
@@ -415,12 +420,12 @@ mod tests {
                 {
                     let source = &index_config.sources[0];
                     assert_eq!(source.source_id, "hdfs-logs-kafka-source");
-                    assert_eq!(source.source_type, "kafka");
+                    assert!(matches!(source.source_params, SourceParams::Kafka(_)));
                 }
                 {
                     let source = &index_config.sources[1];
                     assert_eq!(source.source_id, "hdfs-logs-kinesis-source");
-                    assert_eq!(source.source_type, "kinesis");
+                    assert!(matches!(source.source_params, SourceParams::Kinesis(_)));
                 }
                 Ok(())
             }
@@ -537,13 +542,11 @@ mod tests {
             invalid_index_config.sources = vec![
                 SourceConfig {
                     source_id: "void_1".to_string(),
-                    source_type: "void".to_string(),
-                    params: json!(null),
+                    source_params: SourceParams::void(),
                 },
                 SourceConfig {
                     source_id: "void_1".to_string(),
-                    source_type: "void".to_string(),
-                    params: json!(null),
+                    source_params: SourceParams::void(),
                 },
             ];
             assert!(invalid_index_config.validate().is_err());
@@ -553,6 +556,20 @@ mod tests {
                 .to_string()
                 .contains("Index config contains duplicate sources."));
         }
+        {
+            // Add source file params with no filepath.
+            let mut invalid_index_config = index_config.clone();
+            invalid_index_config.sources = vec![SourceConfig {
+                source_id: "file_params_1".to_string(),
+                source_params: SourceParams::stdin(),
+            }];
+            assert!(invalid_index_config.validate().is_err());
+            assert!(invalid_index_config
+                .validate()
+                .unwrap_err()
+                .to_string()
+                .contains("must contain a `filepath`"));
+        }
         {
             // Add a demux field not declared in the mapping.
             let mut invalid_index_config = index_config;

quickwit-config/src/lib.rs

@@ -19,12 +19,17 @@
 
 mod config;
 mod index_config;
+mod source_config;
 
 pub use config::{
     get_searcher_config_instance, IndexerConfig, QuickwitConfig, SearcherConfig,
     SEARCHER_CONFIG_INSTANCE,
 };
 pub use index_config::{
     build_doc_mapper, DocMapping, IndexConfig, IndexingResources, IndexingSettings, MergePolicy,
-    SearchSettings, SourceConfig,
+    SearchSettings,
+};
+pub use source_config::{
+    FileSourceParams, KafkaSourceParams, SourceConfig, SourceParams, VecSourceParams,
+    VoidSourceParams,
 };