Add http source.

quickwit-oss · Jul 12, 2023 · e50f38b · e50f38b
1 parent 36cba7e
commit e50f38b
Show file tree

Hide file tree

Showing 17 changed files with 789 additions and 55 deletions.
diff --git a/quickwit/Cargo.lock b/quickwit/Cargo.lock
diff --git a/quickwit/Cargo.toml b/quickwit/Cargo.toml
@@ -44,6 +44,7 @@ arc-swap = "1.6"
 assert-json-diff = "2"
 async-speed-limit = "0.4"
 async-trait = "0.1"
+async-compression = { version = "0.4.1", features = ["gzip", "futures-io"] }
 atty = "0.2"
 backoff = { version = "0.4", features = ["tokio"] }
 base64 = "0.21"
@@ -133,6 +134,7 @@ regex = "1.9.1"
 reqwest = { version = "0.11", default-features = false, features = [
   "json",
   "rustls-tls",
+  "stream",
 ] }
 rust-embed = "6.8.1"
 serde = { version = "1.0", features = ["derive", "rc"] }

diff --git a/quickwit/quickwit-common/src/uri.rs b/quickwit/quickwit-common/src/uri.rs
@@ -37,6 +37,8 @@ pub enum Protocol {
     Azure,
     File,
     Grpc,
+    Https,
+    Http,
     PostgreSQL,
     Ram,
     S3,
@@ -48,6 +50,8 @@ impl Protocol {
             Protocol::Azure => "azure",
             Protocol::File => "file",
             Protocol::Grpc => "grpc",
+            Protocol::Https => "https",
+            Protocol::Http => "http",
             Protocol::PostgreSQL => "postgresql",
             Protocol::Ram => "ram",
             Protocol::S3 => "s3",
@@ -105,6 +109,8 @@ impl FromStr for Protocol {
             "azure" => Ok(Protocol::Azure),
             "file" => Ok(Protocol::File),
             "grpc" => Ok(Protocol::Grpc),
+            "https" => Ok(Protocol::Https),
+            "http" => Ok(Protocol::Http),
             "pg" | "postgres" | "postgresql" => Ok(Protocol::PostgreSQL),
             "ram" => Ok(Protocol::Ram),
             "s3" => Ok(Protocol::S3),

diff --git a/quickwit/quickwit-config/src/lib.rs b/quickwit/quickwit-config/src/lib.rs
@@ -50,10 +50,10 @@ use serde::de::DeserializeOwned;
 use serde::Serialize;
 use serde_json::Value as JsonValue;
 pub use source_config::{
-    load_source_config_from_user_config, FileSourceParams, KafkaSourceParams, KinesisSourceParams,
-    PulsarSourceAuth, PulsarSourceParams, RegionOrEndpoint, SourceConfig, SourceInputFormat,
-    SourceParams, TransformConfig, VecSourceParams, VoidSourceParams, CLI_INGEST_SOURCE_ID,
-    INGEST_API_SOURCE_ID,
+    load_source_config_from_user_config, FileSourceParams, HttpSourceParams, KafkaSourceParams,
+    KinesisSourceParams, PulsarSourceAuth, PulsarSourceParams, RegionOrEndpoint, SourceConfig,
+    SourceInputFormat, SourceParams, TransformConfig, VecSourceParams, VoidSourceParams,
+    CLI_INGEST_SOURCE_ID, INGEST_API_SOURCE_ID,
 };
 use tracing::warn;
 
@@ -88,6 +88,7 @@ pub use crate::storage_config::{
     SourceInputFormat,
     SourceParams,
     FileSourceParams,
+    HttpSourceParams,
     KafkaSourceParams,
     KinesisSourceParams,
     PulsarSourceParams,

diff --git a/quickwit/quickwit-config/src/source_config/mod.rs b/quickwit/quickwit-config/src/source_config/mod.rs
@@ -89,6 +89,7 @@ impl SourceConfig {
     pub fn source_type(&self) -> &str {
         match self.source_params {
             SourceParams::File(_) => "file",
+            SourceParams::Http(_) => "http",
             SourceParams::Kafka(_) => "kafka",
             SourceParams::Kinesis(_) => "kinesis",
             SourceParams::Vec(_) => "vec",
@@ -103,6 +104,7 @@ impl SourceConfig {
     pub fn params(&self) -> JsonValue {
         match &self.source_params {
             SourceParams::File(params) => serde_json::to_value(params),
+            SourceParams::Http(params) => serde_json::to_value(params),
             SourceParams::Kafka(params) => serde_json::to_value(params),
             SourceParams::Kinesis(params) => serde_json::to_value(params),
             SourceParams::Vec(params) => serde_json::to_value(params),
@@ -206,6 +208,8 @@ impl FromStr for SourceInputFormat {
 pub enum SourceParams {
     #[serde(rename = "file")]
     File(FileSourceParams),
+    #[serde(rename = "http")]
+    Http(HttpSourceParams),
     #[serde(rename = "kafka")]
     Kafka(KafkaSourceParams),
     #[serde(rename = "kinesis")]
@@ -476,6 +480,34 @@ impl TransformConfig {
     }
 }
 
+#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize, utoipa::ToSchema)]
+#[serde(deny_unknown_fields)]
+pub struct HttpSourceParams {
+    #[schema(value_type = Vec<String>)]
+    #[serde(default)]
+    pub uris: Vec<Uri>,
+    #[schema(value_type = Option<String>)]
+    #[serde(default)]
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub uri_pattern: Option<Uri>,
+}
+
+impl HttpSourceParams {
+    pub fn from_pattern(uri: Uri) -> Self {
+        Self {
+            uris: vec![],
+            uri_pattern: Some(uri),
+        }
+    }
+
+    pub fn from_list(uris: Vec<Uri>) -> Self {
+        Self {
+            uris,
+            uri_pattern: None,
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::str::FromStr;
@@ -1136,4 +1168,16 @@ mod tests {
                 .unwrap();
         assert_eq!(source_config.input_format, SourceInputFormat::PlainText);
     }
+
+    #[tokio::test]
+    async fn test_http_source_config() {
+        let file_content = r#"{"uris":["http://localhost:9000/file.json"], "uri_pattern": "http://localhost:9000/a.json"}"#;
+        assert_eq!(
+            serde_json::from_str::<HttpSourceParams>(file_content).unwrap(),
+            HttpSourceParams {
+                uris: vec![Uri::from_well_formed("http://localhost:9000/file.json")],
+                uri_pattern: Some(Uri::from_well_formed("http://localhost:9000/a.json")),
+            }
+        );
+    }
 }
diff --git a/quickwit/quickwit-config/src/source_config/serialize.rs b/quickwit/quickwit-config/src/source_config/serialize.rs
@@ -91,6 +91,7 @@ impl SourceConfigForSerialization {
                     )
                 }
             }
+            SourceParams::Http(_) => {}
             SourceParams::Kafka(_) | SourceParams::Kinesis(_) | SourceParams::Pulsar(_) => {
                 // TODO consider any validation opportunity
             }
@@ -100,7 +101,7 @@ impl SourceConfigForSerialization {
             | SourceParams::IngestCli => {}
         }
         match &self.source_params {
-            SourceParams::Kafka(_) => {}
+            SourceParams::Kafka(_) | SourceParams::Http(_) => {}
             _ => {
                 if self.desired_num_pipelines > 1 || self.max_num_pipelines_per_indexer > 1 {
                     bail!("Quickwit currently supports multiple pipelines only for Kafka sources. Open an issue https://github.com/quickwit-oss/quickwit/issues if you need the feature for other source types.");

diff --git a/quickwit/quickwit-indexing/Cargo.toml b/quickwit/quickwit-indexing/Cargo.toml
@@ -16,6 +16,7 @@ aws-sdk-kinesis = { workspace = true, optional = true }
 
 anyhow = { workspace = true }
 arc-swap = { workspace = true }
+async-compression = { workspace = true }
 async-trait = { workspace = true }
 backoff = { workspace = true, optional = true }
 byte-unit = { workspace = true }
@@ -25,6 +26,7 @@ fail = { workspace = true }
 flume = { workspace = true }
 fnv = { workspace = true }
 futures = { workspace = true }
+futures-util = {workspace = true }
 itertools = { workspace = true }
 libz-sys = { workspace = true, optional = true }
 once_cell = { workspace = true }
@@ -33,6 +35,8 @@ openssl = { workspace = true, optional = true }
 pulsar = { workspace = true, optional = true }
 quickwit-query = { workspace = true }
 rdkafka = { workspace = true, optional = true }
+regex = { workspace = true }
+reqwest = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 tantivy = { workspace = true }

diff --git a/quickwit/quickwit-indexing/src/actors/indexing_pipeline.rs b/quickwit/quickwit-indexing/src/actors/indexing_pipeline.rs
@@ -380,7 +380,7 @@ impl IndexingPipeline {
             .protect_future(quickwit_supported_sources().load_source(
                 Arc::new(SourceExecutionContext {
                     metastore: self.params.metastore.clone(),
-                    index_uid: self.params.pipeline_id.index_uid.clone(),
+                    pipeline_id: self.params.pipeline_id.clone(),
                     queues_dir_path: self.params.queues_dir_path.clone(),
                     source_config: self.params.source_config.clone(),
                 }),

diff --git a/quickwit/quickwit-indexing/src/source/file_source.rs b/quickwit/quickwit-indexing/src/source/file_source.rs
@@ -181,6 +181,7 @@ mod tests {
     use quickwit_proto::IndexUid;
 
     use super::*;
+    use crate::models::IndexingPipelineId;
     use crate::source::SourceActor;
 
     #[tokio::test]
@@ -190,10 +191,16 @@ mod tests {
         let params = FileSourceParams::file("data/test_corpus.json");
 
         let metastore = metastore_for_test();
+        let pipeline_id = IndexingPipelineId {
+            index_uid: IndexUid::new("test-index"),
+            source_id: "test-file-source".to_string(),
+            node_id: "test-node".to_string(),
+            pipeline_ord: 0,
+        };
         let file_source = FileSourceFactory::typed_create_source(
             SourceExecutionContext::for_test(
                 metastore,
-                IndexUid::new("test-index"),
+                pipeline_id,
                 PathBuf::from("./queues"),
                 SourceConfig {
                     source_id: "test-file-source".to_string(),
@@ -256,10 +263,16 @@ mod tests {
             .to_string();
 
         let metastore = metastore_for_test();
+        let pipeline_id = IndexingPipelineId {
+            index_uid: IndexUid::new("test-index"),
+            source_id: "test-file-source".to_string(),
+            node_id: "test-node".to_string(),
+            pipeline_ord: 0,
+        };
         let source = FileSourceFactory::typed_create_source(
             SourceExecutionContext::for_test(
                 metastore,
-                IndexUid::new("test-index"),
+                pipeline_id,
                 PathBuf::from("./queues"),
                 SourceConfig {
                     source_id: "test-file-source".to_string(),
@@ -346,10 +359,16 @@ mod tests {
         checkpoint.try_apply_delta(checkpoint_delta).unwrap();
 
         let metastore = metastore_for_test();
+        let pipeline_id = IndexingPipelineId {
+            index_uid: IndexUid::new("test-index"),
+            source_id: "test-file-source".to_string(),
+            node_id: "test-node".to_string(),
+            pipeline_ord: 0,
+        };
         let source = FileSourceFactory::typed_create_source(
             SourceExecutionContext::for_test(
                 metastore,
-                IndexUid::new("test-index"),
+                pipeline_id,
                 PathBuf::from("./queues"),
                 SourceConfig {
                     source_id: "test-file-source".to_string(),