Merge pull request #1 from samoii/fix-s3-sse

add sse enum variables

.github/workflows/publish_docker_images.yml

@@ -91,7 +91,7 @@ jobs:
     needs: [docker]
     steps:
       - name: Download digests
-        uses: actions/download-artifact@v3
+        uses: actions/download-artifact@v4.1.8
         with:
           name: digest
           path: /tmp/digests

quickwit/Cargo.toml

@@ -215,6 +215,7 @@ serde_json_borrow = "0.5"
 serde_qs = { version = "0.12", features = ["warp"] }
 serde_with = "3.9.0"
 serde_yaml = "0.9"
+serial_test = { version = "3.1.1", features = ["file_locks"] }
 siphasher = "0.3"
 smallvec = "1"
 sqlx = { version = "0.7", features = [

quickwit/quickwit-actors/src/actor.rs

@@ -36,11 +36,10 @@ pub enum ActorExitStatus {
     /// The actor successfully exited.
     ///
     /// It happens either because:
-    /// - all of the existing mailboxes were dropped and the actor message queue was exhausted.
-    /// No new message could ever arrive to the actor. (This exit is triggered by the framework.)
-    /// or
-    /// - the actor `process_message` method returned `Err(ExitStatusCode::Success)`.
-    /// (This exit is triggered by the actor implementer.)
+    /// - all of the existing mailboxes were dropped and the actor message queue was exhausted. No
+    ///   new message could ever arrive to the actor. (This exit is triggered by the framework.) or
+    /// - the actor `process_message` method returned `Err(ExitStatusCode::Success)`. (This exit is
+    ///   triggered by the actor implementer.)
     ///
     /// (This is equivalent to exit status code 0.)
     /// Note that this is not really an error.

quickwit/quickwit-actors/src/scheduler.rs

@@ -323,10 +323,9 @@ impl Scheduler {
     /// Updates the simulated time shift, if appropriate.
     ///
     /// We advance time if:
-    /// - someone is actually requesting for a simulated fast forward in time.
-    /// (if Universe::simulate_time_shift(..) has been called).
-    /// - no message is queued for processing, no initialize or no finalize
-    /// is being processed.
+    /// - someone is actually requesting for a simulated fast forward in time. (if
+    ///   Universe::simulate_time_shift(..) has been called).
+    /// - no message is queued for processing, no initialize or no finalize is being processed.
     fn advance_time_if_necessary(&mut self) {
         let Some(scheduler_client) = self.scheduler_client() else {
             return;

quickwit/quickwit-cli/tests/helpers.rs

@@ -114,7 +114,6 @@ pub struct TestResourceFiles {
     pub index_config_without_uri: Uri,
     pub index_config_with_retention: Uri,
     pub log_docs: Uri,
-    pub wikipedia_docs: Uri,
 }
 
 /// A struct to hold few info about the test environment.
@@ -130,7 +129,6 @@ pub struct TestEnv {
     /// The metastore URI.
     pub metastore_uri: Uri,
     pub metastore_resolver: MetastoreResolver,
-    pub metastore: MetastoreServiceClient,
 
     pub cluster_endpoint: Url,
 
@@ -219,7 +217,6 @@ pub async fn create_test_env(
     let storage_resolver = StorageResolver::unconfigured();
     let storage = storage_resolver.resolve(&metastore_uri).await?;
     let metastore_resolver = MetastoreResolver::unconfigured();
-    let metastore = metastore_resolver.resolve(&metastore_uri).await?;
     let index_uri = metastore_uri.join(&index_id).unwrap();
     let index_config_path = resources_dir_path.join("index_config.yaml");
     fs::write(
@@ -258,7 +255,7 @@ pub async fn create_test_env(
     let log_docs_path = resources_dir_path.join("logs.json");
     fs::write(&log_docs_path, LOGS_JSON_DOCS)?;
     let wikipedia_docs_path = resources_dir_path.join("wikis.json");
-    fs::write(&wikipedia_docs_path, WIKI_JSON_DOCS)?;
+    fs::write(wikipedia_docs_path, WIKI_JSON_DOCS)?;
 
     let cluster_endpoint = Url::parse(&format!("http://localhost:{rest_listen_port}"))
         .context("failed to parse cluster endpoint")?;
@@ -269,7 +266,6 @@ pub async fn create_test_env(
         index_config_without_uri: uri_from_path(&index_config_without_uri_path),
         index_config_with_retention: uri_from_path(&index_config_with_retention_path),
         log_docs: uri_from_path(&log_docs_path),
-        wikipedia_docs: uri_from_path(&wikipedia_docs_path),
     };
 
     Ok(TestEnv {
@@ -279,7 +275,6 @@ pub async fn create_test_env(
         resource_files,
         metastore_uri,
         metastore_resolver,
-        metastore,
         cluster_endpoint,
         index_id,
         index_uri,

quickwit/quickwit-cluster/src/cluster.rs

@@ -416,6 +416,7 @@ impl Cluster {
     /// Tasks are grouped by (index_id, source_id), each group is stored in a key as follows:
     /// - key: `{INDEXING_TASK_PREFIX}{index_id}{INDEXING_TASK_SEPARATOR}{source_id}`
     /// - value: Number of indexing tasks in the group.
+    ///
     /// Keys present in chitchat state but not in the given `indexing_tasks` are marked for
     /// deletion.
     pub async fn update_self_node_indexing_tasks(&self, indexing_tasks: &[IndexingTask]) {

quickwit/quickwit-codegen/example/Cargo.toml

@@ -36,3 +36,6 @@ quickwit-actors = { workspace = true, features = ["testsuite"] }
 
 [build-dependencies]
 quickwit-codegen = { workspace = true }
+
+[features]
+testsuite = []

quickwit/quickwit-common/build.rs

@@ -0,0 +1,22 @@
+// Copyright (C) 2024 Quickwit, Inc.
+//
+// Quickwit is offered under the AGPL v3.0 and as commercial software.
+// For commercial licensing, contact us at [email protected].
+//
+// AGPL:
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as
+// published by the Free Software Foundation, either version 3 of the
+// License, or (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+fn main() {
+    println!("cargo::rustc-check-cfg=cfg(tokio_unstable)");
+}

quickwit/quickwit-common/src/pubsub.rs

@@ -54,6 +54,7 @@ type EventSubscriptions<E> = HashMap<usize, EventSubscription<E>>;
 /// The event broker makes it possible to
 /// - emit specific local events
 /// - subscribe to these local events
+///
 /// The event broker is not distributed in itself. Only events emitted
 /// locally will be received by the subscribers.
 ///

quickwit/quickwit-common/src/thread_pool.rs

@@ -69,12 +69,11 @@ impl ThreadPool {
     ///
     /// Here are two important differences however:
     ///
-    /// 1) The task runs on a rayon thread pool managed by Quickwit.
-    /// This pool is specifically used only to run CPU-intensive work
-    /// and is configured to contain `num_cpus` cores.
+    /// 1) The task runs on a rayon thread pool managed by Quickwit. This pool is specifically used
+    ///    only to run CPU-intensive work and is configured to contain `num_cpus` cores.
     ///
-    /// 2) Before the task is effectively scheduled, we check that
-    /// the spawner is still interested in its result.
+    /// 2) Before the task is effectively scheduled, we check that the spawner is still interested
+    ///    in its result.
     ///
     /// It is therefore required to `await` the result of this
     /// function to get any work done.

quickwit/quickwit-config/src/lib.rs

@@ -80,7 +80,7 @@ pub use crate::node_config::{
 use crate::source_config::serialize::{SourceConfigV0_7, SourceConfigV0_8, VersionedSourceConfig};
 pub use crate::storage_config::{
     AzureStorageConfig, FileStorageConfig, GoogleCloudStorageConfig, RamStorageConfig,
-    S3StorageConfig, StorageBackend, StorageBackendFlavor, StorageConfig, StorageConfigs,
+    S3StorageConfig, StorageBackend, StorageBackendFlavor, StorageConfig, StorageConfigs, S3ServerSideEncryption,
 };
 
 /// Returns true if the ingest API v2 is enabled.

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

@@ -68,8 +68,8 @@ impl SourceConfigForSerialization {
     /// Checks the validity of the `SourceConfig` as a "deserializable source".
     ///
     /// Two remarks:
-    /// - This does not check connectivity, it just validate configuration,
-    /// without performing any IO. See `check_connectivity(..)`.
+    /// - This does not check connectivity, it just validate configuration, without performing any
+    ///   IO. See `check_connectivity(..)`.
     /// - This is used each time the `SourceConfig` is deserialized (at creation but also during
     ///   communications with the metastore). When ingesting from stdin, we programmatically create
     ///   an invalid `SourceConfig` and only use it locally.

quickwit/quickwit-config/src/storage_config.rs

@@ -18,6 +18,7 @@
 // along with this program. If not, see <http://www.gnu.org/licenses/>.
 
 use std::ops::Deref;
+use std::sync::OnceLock;
 use std::{env, fmt};
 
 use anyhow::ensure;
@@ -311,7 +312,13 @@ impl fmt::Debug for AzureStorageConfig {
             .finish()
     }
 }
-
+#[derive(Debug, Clone, Copy, Eq, PartialEq, Ord, PartialOrd, Hash, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum S3ServerSideEncryption {
+    Aes256,
+    AwsKms,
+    AwsKmsDsse,
+}
 #[derive(Clone, Default, Eq, PartialEq, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct S3StorageConfig {
@@ -334,7 +341,7 @@ pub struct S3StorageConfig {
     #[serde(default)]
     pub disable_multipart_upload: bool,
     #[serde(default)]
-    pub server_side_encryption: Option<String>,
+    pub server_side_encryption: Option<S3ServerSideEncryption>,
 }
 
 impl S3StorageConfig {
@@ -372,11 +379,14 @@ impl S3StorageConfig {
     }
 
     pub fn force_path_style_access(&self) -> Option<bool> {
-        let force_path_style_access = get_bool_from_env(
-            "QW_S3_FORCE_PATH_STYLE_ACCESS",
-            self.force_path_style_access,
-        );
-        Some(force_path_style_access)
+        static FORCE_PATH_STYLE: OnceLock<Option<bool>> = OnceLock::new();
+        *FORCE_PATH_STYLE.get_or_init(|| {
+            let force_path_style_access = get_bool_from_env(
+                "QW_S3_FORCE_PATH_STYLE_ACCESS",
+                self.force_path_style_access,
+            );
+            Some(force_path_style_access)
+        })
     }
 }
 

quickwit/quickwit-config/src/templating.rs

@@ -30,7 +30,7 @@ use tracing::debug;
 // `ENV_VAR` or `ENV_VAR:DEFAULT`
 // Ignores whitespaces in curly braces
 static TEMPLATE_ENV_VAR_CAPTURE: Lazy<Regex> = Lazy::new(|| {
-    Regex::new(r"\$\{\s*([A-Za-z0-9_]+)\s*(?::\-\s*([\S]+)\s*)?}")
+    Regex::new(r"\$\{\s*([A-Za-z0-9_]+)\s*(?::\-\s*([^\s\}]+)\s*)?}")
         .expect("regular expression should compile")
 });
 
@@ -158,6 +158,23 @@ mod test {
         assert_eq!(rendered, "metastore_uri: s3://test-bucket/metastore");
     }
 
+    #[test]
+    fn test_template_render_with_multiple_vars_per_line() {
+        let config_content =
+            b"metastore_uri: s3://${RENDER_MULTIPLE_BUCKET}/${RENDER_MULTIPLE_PREFIX:-index}#polling_interval=${RENDER_MULTIPLE_INTERVAL}s";
+        env::set_var("RENDER_MULTIPLE_BUCKET", "test-bucket");
+        env::set_var("RENDER_MULTIPLE_PREFIX", "metastore");
+        env::set_var("RENDER_MULTIPLE_INTERVAL", "30");
+        let rendered = render_config(config_content).unwrap();
+        std::env::remove_var("RENDER_MULTIPLE_BUCKET");
+        std::env::remove_var("RENDER_MULTIPLE_PREFIX");
+        std::env::remove_var("RENDER_MULTIPLE_INTERVAL");
+        assert_eq!(
+            rendered,
+            "metastore_uri: s3://test-bucket/metastore#polling_interval=30s"
+        );
+    }
+
     #[test]
     fn test_template_render_ignores_commented_lines() {
         {

quickwit/quickwit-control-plane/src/indexing_scheduler/scheduling/mod.rs

@@ -633,14 +633,13 @@ fn inflate_node_capacities_if_necessary(problem: &mut SchedulingProblem) {
 /// to transform scheduling into a math problem.
 ///
 /// This function implementation therefore goes
-/// - 1) transform our problem into a scheduling problem. Something closer to a well-defined
-/// optimization problem. In particular this step removes:
-///   - the notion of shard ids, and only considers a number of shards being allocated.
-///   - node_ids and shard ids. These are replaced by integers.
-/// - 2) convert the current situation of the cluster into something a previous scheduling
-/// solution.
-/// - 3) compute the new scheduling solution.
-/// - 4) convert the new scheduling solution back to the real world by reallocating the shard ids.
+/// 1) transform our problem into a scheduling problem. Something closer to a well-defined
+///    optimization problem. In particular this step removes:
+///    - the notion of shard ids, and only considers a number of shards being allocated.
+///    - node_ids and shard ids. These are replaced by integers.
+/// 2) convert the current situation of the cluster into something a previous scheduling solution.
+/// 3) compute the new scheduling solution.
+/// 4) convert the new scheduling solution back to the real world by reallocating the shard ids.
 ///
 /// TODO cut into pipelines.
 /// Panics if any sources has no shards.

quickwit/quickwit-directories/src/debug_proxy_directory.rs

@@ -111,8 +111,8 @@ impl ReadOperationBuilder {
 /// recording all of its read operations.
 ///
 /// It has two purpose
-/// - It is used when building our hotcache, to identify the file sections that
-/// should be in the hotcache.
+/// - It is used when building our hotcache, to identify the file sections that should be in the
+///   hotcache.
 /// - It is used in the search-api to provide debugging/performance information.
 #[derive(Debug)]
 pub struct DebugProxyDirectory<D: Directory> {

quickwit/quickwit-doc-mapper/src/default_doc_mapper/mapping_tree.rs

@@ -1735,7 +1735,7 @@ mod tests {
     #[test]
     fn test_parse_i64_too_large() {
         let leaf = LeafType::I64(QuickwitNumericOptions::default());
-        let err = leaf.value_from_json(json!(u64::max_value())).err().unwrap();
+        let err = leaf.value_from_json(json!(u64::MAX)).err().unwrap();
         assert_eq!(
             err,
             "expected i64, got inconvertible JSON number `18446744073709551615`"