mirror of
https://github.com/quickwit-oss/quickwit.git
synced 2026-10-28 10:45:05 +08:00
Fixing spelling mistakes. (#2927)
This commit is contained in:
1 parent
81af21ff37
commit
b17062b831
81 files changed
+148
-142
No files matched your search
@@ -44,6 +44,7 @@ You will be notified by email from the CI system if any issues are discovered, b
|
||||
* `make -k test-all docker-compose-down` - the same as above, but tears down the Docker services after running all the tests.
|
||||
* `make fmt` - runs formatter, this command requires the nightly toolchain to be installed by running `rustup toolchain install nightly`.
|
||||
* `make fix` - runs formatter and clippy checks.
|
||||
* `make typos` - runs the spellcheck tool over the codebase. (Install by running `cargo install typos`)
|
||||
* `make build-docs` - builds docs.
|
||||
* `make docker-compose-up` - starts Docker services.
|
||||
* `make docker-compose-down` - stops Docker services.
|
||||
|
||||
@@ -41,6 +41,9 @@ fmt:
|
||||
fix:
|
||||
@$(MAKE) -C $(QUICKWIT_SRC) fix
|
||||
|
||||
typos:
|
||||
typos
|
||||
|
||||
# Usage:
|
||||
# `make test-all` starts the Docker services and runs all the tests.
|
||||
# `make -k test-all docker-compose-down`, tears down the Docker services after running all the tests.
|
||||
|
||||
@@ -0,0 +1,2 @@
|
||||
[files]
|
||||
extend-exclude = ["**/*.json"]
|
||||
@@ -4,7 +4,7 @@ ARG PBC_URL="https://github.com/protocolbuffers/protobuf/releases/download/v21.5
|
||||
|
||||
#TODO:
|
||||
# We can switch to static linking (remove `libsasl2-dev:arm64`) using
|
||||
# `rdkafka/gssapi-vendored` feature when there is a release incuding:
|
||||
# `rdkafka/gssapi-vendored` feature when there is a release including:
|
||||
# https://github.com/MaterializeInc/rust-sasl/pull/48
|
||||
|
||||
RUN dpkg --add-architecture arm64 && \
|
||||
|
||||
@@ -65,7 +65,7 @@ a few faulty documents.
|
||||
Quickwit allows you to configure how document are routed with a simple DSL. Here are some sample expression with a short description of their result:
|
||||
|
||||
- `tenant_id`: create one partition per tenant\_id
|
||||
- `tenant_id,app_id`: create one partiton per unique combination of tenant\_id and app\_id
|
||||
- `tenant_id,app_id`: create one partition per unique combination of tenant\_id and app\_id
|
||||
- `tenant_id,hash_mod(app_id, 8)`: for each tenant, create up to 8 partitions containing each data related to some applications
|
||||
- `hash_mod((tenant_id,app_id), 50)`: create 50 partition in total, containing some combination of tenant and apps.
|
||||
|
||||
|
||||
@@ -9,7 +9,7 @@ Quickwit lets you place the cursor on how strict you would like your schema to b
|
||||
|
||||
:::note
|
||||
|
||||
To execute the CLI commands thoughout this guide, [install](/docs/get-started/installation.md) Quickwit and start a server in a terminal with the following command:
|
||||
To execute the CLI commands throughout this guide, [install](/docs/get-started/installation.md) Quickwit and start a server in a terminal with the following command:
|
||||
|
||||
```bash
|
||||
./quickwit run
|
||||
|
||||
@@ -15,7 +15,7 @@ Stored in metastore only:
|
||||
- FileBackedIndex
|
||||
- SplitMetadata.
|
||||
|
||||
Quickwit currently manages backward compatibiltiy of all of these resources but QuickwitConfig.
|
||||
Quickwit currently manages backward compatibility of all of these resources but QuickwitConfig.
|
||||
This document describes how to handle a change, and how to make test such a change,
|
||||
and spot eventual regression.
|
||||
|
||||
|
||||
@@ -106,7 +106,7 @@ See our [Scaleway Setup Guide](../guides/storage-setup/scaleway-setup.md) for th
|
||||
|
||||
### Garage
|
||||
|
||||
[Garage](https://garagehq.deuxfleurs.fr/) is an Open-Souce lightweight and efficient object storage.
|
||||
[Garage](https://garagehq.deuxfleurs.fr/) is an Open-Source lightweight and efficient object storage.
|
||||
|
||||
To use it with Quickwit, you will need to setup the region, as mentioned in [Garage documentation](https://garagehq.deuxfleurs.fr/documentation/connect/), it's often set just to `garage`.
|
||||
|
||||
|
||||
+2
-2
@@ -4,7 +4,7 @@
|
||||
#
|
||||
# This is just a little script that can be downloaded from the internet to
|
||||
# install Quickwit.
|
||||
# It just does platform detection, fetches the lastest appropriate release version from github
|
||||
# It just does platform detection, fetches the latest appropriate release version from github
|
||||
# and execute the appropriate commands to download the binary.
|
||||
#
|
||||
# Heavily inspired by the Vector & Meilisearch installation scripts
|
||||
@@ -199,7 +199,7 @@ get_latest_version() {
|
||||
latest=""
|
||||
current_tag=""
|
||||
for release_info in $releases; do
|
||||
if [ $i -eq 0 ]; then # Cheking tag_name
|
||||
if [ $i -eq 0 ]; then # Checking tag_name
|
||||
if echo "$release_info" | grep -q "$GREP_SEMVER_REGEXP"; then # If it's not an alpha or beta release
|
||||
current_tag=$release_info
|
||||
else
|
||||
|
||||
+2
-2
@@ -66,7 +66,7 @@ ignore = [
|
||||
[licenses]
|
||||
# The lint level for crates which do not have a detectable license
|
||||
unlicensed = "deny"
|
||||
# List of explictly allowed licenses
|
||||
# List of explicitly allowed licenses
|
||||
# See https://spdx.org/licenses/ for list of possible licenses
|
||||
# [possible values: any SPDX 3.11 short identifier (+ optional exception)].
|
||||
allow = [
|
||||
@@ -85,7 +85,7 @@ allow = [
|
||||
"Zlib",
|
||||
"zlib-acknowledgement",
|
||||
]
|
||||
# List of explictly disallowed licenses
|
||||
# List of explicitly disallowed licenses
|
||||
# See https://spdx.org/licenses/ for list of possible licenses
|
||||
# [possible values: any SPDX 3.11 short identifier (+ optional exception)].
|
||||
deny = [
|
||||
|
||||
@@ -226,7 +226,7 @@ impl<A: Actor> ActorContext<A> {
|
||||
/// more indicated.
|
||||
///
|
||||
/// Droppping the receiver channel will not cancel the
|
||||
/// processing of the messsage. It is a very common usage.
|
||||
/// processing of the message. It is a very common usage.
|
||||
/// In fact most actors are expected to send message in a
|
||||
/// fire-and-forget fashion.
|
||||
///
|
||||
|
||||
@@ -434,7 +434,7 @@ mod tests {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_try_recv_prority_corner_case() -> anyhow::Result<()> {
|
||||
async fn test_try_recv_priority_corner_case() -> anyhow::Result<()> {
|
||||
let (sender, receiver) = super::channel::<usize>(QueueCapacity::Unbounded);
|
||||
tokio::task::spawn(async move {
|
||||
tokio::time::sleep(Duration::from_millis(10)).await;
|
||||
|
||||
@@ -113,7 +113,7 @@ impl<A: Actor> Handler<Command> for A {
|
||||
|
||||
/// Asks the actor to update its ObservableState.
|
||||
///
|
||||
/// The observation is then available using the `ActorHander::last_observation()`
|
||||
/// The observation is then available using the `ActorHandler::last_observation()`
|
||||
/// method.
|
||||
#[derive(Debug)]
|
||||
pub(crate) struct Observe;
|
||||
|
||||
@@ -89,7 +89,7 @@ pub const HEARTBEAT: Duration = if cfg!(any(test, feature = "testsuite")) {
|
||||
/// Once this time is elapsed, we just return the last observation.
|
||||
const OBSERVE_TIMEOUT: Duration = Duration::from_secs(3);
|
||||
|
||||
/// Error that occured while calling `ActorContext::ask(..)` or `Universe::ask`
|
||||
/// Error that occurred while calling `ActorContext::ask(..)` or `Universe::ask`
|
||||
#[derive(Error, Debug)]
|
||||
pub enum AskError<E: fmt::Debug> {
|
||||
#[error("Message could not be delivered")]
|
||||
|
||||
@@ -555,7 +555,7 @@ mod tests {
|
||||
.await
|
||||
.unwrap();
|
||||
// That second message will present some backpressure, since the capacity is 0 and
|
||||
// the first message willl take 1000 micros to be processed.
|
||||
// the first message will take 1000 micros to be processed.
|
||||
mailbox
|
||||
.send_message_with_backpressure_counter(
|
||||
Duration::default(),
|
||||
|
||||
@@ -248,7 +248,7 @@ impl ActorJoinHandle {
|
||||
})
|
||||
}
|
||||
|
||||
/// Joins the actor and returns its exit status on the frist invocation.
|
||||
/// Joins the actor and returns its exit status on the first invocation.
|
||||
/// Returns None afterwards.
|
||||
pub(crate) async fn join(&self) -> ActorExitStatus {
|
||||
self.holder.clone().await
|
||||
|
||||
@@ -216,7 +216,7 @@ mod tests {
|
||||
async fn test_schedule_for_actor() {
|
||||
let universe = Universe::with_accelerated_time();
|
||||
let actor_with_schedule = CountingMinutesActor::default();
|
||||
let (_maibox, handler) = universe.spawn_builder().spawn(actor_with_schedule);
|
||||
let (_mailbox, handler) = universe.spawn_builder().spawn(actor_with_schedule);
|
||||
let count_after_initialization = handler.process_pending_and_observe().await.state;
|
||||
assert_eq!(count_after_initialization, 1);
|
||||
universe.sleep(Duration::from_secs(200)).await;
|
||||
@@ -229,7 +229,7 @@ mod tests {
|
||||
async fn test_actor_quit_after_universe_quit() {
|
||||
let universe = Universe::with_accelerated_time();
|
||||
let actor_with_schedule = CountingMinutesActor::default();
|
||||
let (_maibox, handler) = universe.spawn_builder().spawn(actor_with_schedule);
|
||||
let (_mailbox, handler) = universe.spawn_builder().spawn(actor_with_schedule);
|
||||
universe.sleep(Duration::from_secs(200)).await;
|
||||
let res = universe.quit().await;
|
||||
assert_eq!(res.len(), 1);
|
||||
@@ -241,7 +241,7 @@ mod tests {
|
||||
async fn test_universe_join_after_actor_quit() {
|
||||
let universe = Universe::default();
|
||||
let actor_with_schedule = CountingMinutesActor::default();
|
||||
let (_maibox, handler) = universe.spawn_builder().spawn(actor_with_schedule);
|
||||
let (_mailbox, handler) = universe.spawn_builder().spawn(actor_with_schedule);
|
||||
assert!(matches!(handler.quit().await, (ActorExitStatus::Quit, 1)));
|
||||
assert!(!universe
|
||||
.quit()
|
||||
@@ -255,8 +255,8 @@ mod tests {
|
||||
let universe = Universe::default();
|
||||
let panicking_actor = ExitPanickingActor::default();
|
||||
let actor_with_schedule = CountingMinutesActor::default();
|
||||
let (_maibox, _handler) = universe.spawn_builder().spawn(panicking_actor);
|
||||
let (_maibox, _handler) = universe.spawn_builder().spawn(actor_with_schedule);
|
||||
let (_mailbox, _handler) = universe.spawn_builder().spawn(panicking_actor);
|
||||
let (_mailbox, _handler) = universe.spawn_builder().spawn(actor_with_schedule);
|
||||
assert!(universe
|
||||
.quit()
|
||||
.await
|
||||
|
||||
@@ -76,7 +76,7 @@ pub fn region_from_str(region_str: &str) -> anyhow::Result<Region> {
|
||||
return Ok(region);
|
||||
}
|
||||
// Maybe this is a custom endpoint (for MinIO for instance).
|
||||
// We require custom endpoints to explicitely state the http/https protocol`.
|
||||
// We require custom endpoints to explicitly state the http/https protocol`.
|
||||
if !region_str.starts_with("http") {
|
||||
anyhow::bail!(
|
||||
"Invalid AWS region. Quickwit expects an AWS region code like `us-east-1` or a \
|
||||
|
||||
@@ -888,7 +888,7 @@ pub async fn start_statistics_reporting_loop(
|
||||
|
||||
loop {
|
||||
// TODO fixme. The way we wait today is a bit lame: if the indexing pipeline exits, we will
|
||||
// stil wait up to an entire heartbeat... Ideally we should select between two
|
||||
// still wait up to an entire heartbeat... Ideally we should select between two
|
||||
// futures.
|
||||
report_interval.tick().await;
|
||||
// Try to receive with a timeout of 1 second.
|
||||
|
||||
@@ -126,7 +126,7 @@ async fn load_quickwit_config(config_uri: &Uri) -> anyhow::Result<QuickwitConfig
|
||||
}
|
||||
|
||||
/// Runs connectivity checks for a given `metastore_uri` and `index_id`.
|
||||
/// Optionaly, it takes a `SourceConfig` that will be checked instead
|
||||
/// Optionally, it takes a `SourceConfig` that will be checked instead
|
||||
/// of the index's sources.
|
||||
pub async fn run_index_checklist(
|
||||
metastore_uri: &Uri,
|
||||
|
||||
@@ -580,7 +580,7 @@ pub async fn start_statistics_reporting_loop(
|
||||
|
||||
loop {
|
||||
// TODO fixme. The way we wait today is a bit lame: if the indexing pipeline exits, we will
|
||||
// stil wait up to an entire heartbeat... Ideally we should select between two
|
||||
// still wait up to an entire heartbeat... Ideally we should select between two
|
||||
// futures.
|
||||
report_interval.tick().await;
|
||||
// Try to receive with a timeout of 1 second.
|
||||
|
||||
@@ -119,7 +119,7 @@ pub async fn wait_port_ready(port: u16) -> anyhow::Result<()> {
|
||||
}
|
||||
}
|
||||
|
||||
/// A struct to hold few info about the test environement.
|
||||
/// A struct to hold few info about the test environment.
|
||||
pub struct TestEnv {
|
||||
/// The temporary directory of the test.
|
||||
_tempdir: TempDir,
|
||||
@@ -182,7 +182,7 @@ pub enum TestStorageType {
|
||||
LocalFileSystem,
|
||||
}
|
||||
|
||||
/// Creates all necessary artifacts in a test environement.
|
||||
/// Creates all necessary artifacts in a test environment.
|
||||
pub fn create_test_env(index_id: String, storage_type: TestStorageType) -> anyhow::Result<TestEnv> {
|
||||
let tempdir = tempdir()?;
|
||||
let data_dir_path = tempdir.path().join("data");
|
||||
|
||||
@@ -278,7 +278,7 @@ impl Cluster {
|
||||
info!(self_addr = ?self.gossip_listen_addr, "Shutting down chitchat.");
|
||||
let result = self.chitchat_handle.shutdown().await;
|
||||
if let Err(error) = result {
|
||||
error!(self_addr = ?self.gossip_listen_addr, error = ?error, "Error while shuting down chitchat.");
|
||||
error!(self_addr = ?self.gossip_listen_addr, error = ?error, "Error while shutting down chitchat.");
|
||||
}
|
||||
|
||||
self.stop.store(true, Ordering::Relaxed);
|
||||
@@ -304,7 +304,7 @@ impl Cluster {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Set self readyness value.
|
||||
/// Set self readiness value.
|
||||
pub async fn set_self_node_ready(&self, ready: bool) {
|
||||
let health_value = if ready {
|
||||
HEALTH_VALUE_READY
|
||||
@@ -499,7 +499,7 @@ mod tests {
|
||||
use super::*;
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_cluster_single_node_readyness() -> anyhow::Result<()> {
|
||||
async fn test_cluster_single_node_readiness() -> anyhow::Result<()> {
|
||||
let transport = ChannelTransport::default();
|
||||
let cluster = create_cluster_for_test(Vec::new(), &[], &transport, false).await?;
|
||||
let members: Vec<SocketAddr> = cluster
|
||||
|
||||
@@ -83,7 +83,7 @@ pub fn print_checklist(check_list_results: &[(&str, anyhow::Result<()>)]) {
|
||||
|
||||
/// Run a checklist and print out its successes and failures on stdout.
|
||||
///
|
||||
/// If an error is encountered, the proccess will exit with exit code 1.
|
||||
/// If an error is encountered, the process will exit with exit code 1.
|
||||
pub fn run_checklist(checks: Vec<(&str, anyhow::Result<()>)>) -> Result<(), ChecklistError> {
|
||||
print_checklist(&checks);
|
||||
if !checks
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
version: 0.4
|
||||
searcher:
|
||||
fast_field_cache_capacity: 10G
|
||||
# Typo here. It is supposed to be searchs.
|
||||
max_num_concurrent_split_searchs: 150
|
||||
# Typo here. It is supposed to be searches.
|
||||
max_num_concurrent_split_searchs_with_typo: 150
|
||||
@@ -486,7 +486,7 @@ mod tests {
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(format!("{parsing_error:?}")
|
||||
.contains("unknown field `max_num_concurrent_split_searchs`"));
|
||||
.contains("unknown field `max_num_concurrent_split_searches_with_typo`"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
@@ -50,7 +50,7 @@ pub struct SourceConfig {
|
||||
|
||||
/// Maximum number of indexing pipelines spawned for the source on a given indexer.
|
||||
/// The maximum is reached only if there is enough `desired_num_pipelines` to run.
|
||||
/// The value is only used by sources that Quickwit knows how to distribute accross
|
||||
/// The value is only used by sources that Quickwit knows how to distribute across
|
||||
/// pipelines/nodes, that is for Kafka sources only.
|
||||
/// Example:
|
||||
/// - `max_num_pipelines_per_indexer=2`
|
||||
@@ -59,7 +59,7 @@ pub struct SourceConfig {
|
||||
pub max_num_pipelines_per_indexer: usize,
|
||||
/// Number of desired indexing pipelines to run on a cluster for the source.
|
||||
/// This number could not be reach if there is not enough indexers.
|
||||
/// The value is only used by sources that Quickwit knows how to distribute accross
|
||||
/// The value is only used by sources that Quickwit knows how to distribute across
|
||||
/// pipelines/nodes, that is for Kafka sources only.
|
||||
/// Example:
|
||||
/// - `max_num_pipelines_per_indexer=1`
|
||||
|
||||
@@ -126,9 +126,9 @@ impl PhysicalIndexingPlan {
|
||||
/// Builds a [`PhysicalIndexingPlan`] by assigning each indexing tasks to a node ID.
|
||||
/// The algorithm first sort indexing tasks by (index_id, source_id).
|
||||
/// Then for each indexing tasks, it performs the following steps:
|
||||
/// 1. Sort node by rendez-vous hashing to make the assigment stable (it makes it
|
||||
/// 1. Sort node by rendez-vous hashing to make the assignment stable (it makes it
|
||||
/// deterministic too). This is not bullet proof as the node score has an impact
|
||||
/// on the assigment too.
|
||||
/// on the assignment too.
|
||||
/// 2. Select node candidates that can run the task, see [`select_node_candidates`]
|
||||
/// function.
|
||||
/// 3. For each node, compute a score for this task, the higher, the better, see
|
||||
@@ -162,7 +162,7 @@ pub(crate) fn build_physical_indexing_plan(
|
||||
.expect("SourceConfig should always be present.");
|
||||
let candidates = select_node_candidates(&node_ids, &plan, source_config, &indexing_task);
|
||||
|
||||
// It's theoritically possible to have no candidate as all indexers can already
|
||||
// It's theoretically possible to have no candidate as all indexers can already
|
||||
// have more than `max_num_pipelines_per_indexer` assigned for a given source.
|
||||
// But, when building the list of indexing tasks to run on the cluster in
|
||||
// `build_indexing_plan`, we make sure to always respect the constraint
|
||||
|
||||
@@ -65,12 +65,12 @@ pub struct IndexingSchedulerState {
|
||||
/// The scheduling executes the following steps:
|
||||
/// 1. Fetches all indexes metadata.
|
||||
/// 2. Builds an indexing plan = `[Vec<IndexingTask>]`, from the indexes metadatas.
|
||||
/// See [`build_indexing_plan`] for the implemention details.
|
||||
/// See [`build_indexing_plan`] for the implementation details.
|
||||
/// 3. Builds a [`PhysicalIndexingPlan`] from the list of indexing tasks.
|
||||
/// See [`build_physical_indexing_plan`] for the implemention details.
|
||||
/// See [`build_physical_indexing_plan`] for the implementation details.
|
||||
/// 4. Apply the [`PhysicalIndexingPlan`]: for each indexer, the scheduler send the indexing tasks
|
||||
/// by gRPC. An indexer immediately returns an Ok and apply asynchronously the received plan.
|
||||
/// Any errors (network) happening in this step are ignored. The schduler runs a control loop
|
||||
/// Any errors (network) happening in this step are ignored. The scheduler runs a control loop
|
||||
/// that regularly checks if indexers are effectively running their plans (more details in the
|
||||
/// next section).
|
||||
///
|
||||
@@ -78,14 +78,14 @@ pub struct IndexingSchedulerState {
|
||||
/// certains conditions. The following events possibly trigger a scheduling:
|
||||
/// - [`NotifyIndexChangeRequest`]: this gRPC event is sent by a metastore node and will trigger a
|
||||
/// scheduling on each event. TODO(fmassot): this can be refined by adding some relevant info to
|
||||
/// the event, example: the creation of a source of type `void` should not trigger a sheduling.
|
||||
/// the event, example: the creation of a source of type `void` should not trigger a scheduling.
|
||||
/// - [`RefreshPlanLoop`]: this event is scheduled every [`REFRESH_PLAN_LOOP_INTERVAL`] and triggers
|
||||
/// a scheduling. Due to network issues, a control plane will not always receive the gRPC events
|
||||
/// [`NotifyIndexChangeRequest`] and thus will not be aware of index changes in the metastore.
|
||||
/// TODO(fmassot): to avoid a scheduling on each [`RefreshPlanLoop`], we can store in the
|
||||
/// scheduler state a metastore version number that will be compared to the number stored in the
|
||||
/// metastore itself.
|
||||
/// - [`ControlPlanLoop`]: this event is sheduled every [`HEARTBEAT`] and control if the `desired
|
||||
/// - [`ControlPlanLoop`]: this event is scheduled every [`HEARTBEAT`] and control if the `desired
|
||||
/// plan`, that is the last applied [`PhysicalIndexingPlan`] by the scheduler, and the `running
|
||||
/// plan`, that is the indexing tasks running on all indexers and retrieved from the chitchat
|
||||
/// state, are the same. If not, the scheduler will trigger a scheduling.
|
||||
@@ -186,7 +186,7 @@ impl IndexingScheduler {
|
||||
};
|
||||
|
||||
// TODO(fmassot): once the actor framekwork can correctly simulate time advance, change
|
||||
// `MIN_DURATION_BETWEEN_SCHEDULING_SECS` value for test and correclty test it. Currently, I
|
||||
// `MIN_DURATION_BETWEEN_SCHEDULING_SECS` value for test and correctly test it. Currently, I
|
||||
// put its value 0 to make test simple.
|
||||
if let Some(last_applied_plan_timestamp) = self.state.last_applied_plan_timestamp.as_ref() {
|
||||
// If the last applied plan is empty, the node is probably starting and did not find
|
||||
@@ -246,7 +246,7 @@ impl IndexingScheduler {
|
||||
})
|
||||
.await
|
||||
{
|
||||
error!(indexer_node_id=%indexer.node_id, err=?error, "Error occured when appling indexing plan to indexer.");
|
||||
error!(indexer_node_id=%indexer.node_id, err=?error, "Error occurred when appling indexing plan to indexer.");
|
||||
}
|
||||
}
|
||||
None => {
|
||||
@@ -271,7 +271,7 @@ impl Handler<NotifyIndexChangeRequest> for IndexingScheduler {
|
||||
_: NotifyIndexChangeRequest,
|
||||
_: &ActorContext<Self>,
|
||||
) -> Result<(), ActorExitStatus> {
|
||||
info!("Index change notification: shedule indexing plan.");
|
||||
info!("Index change notification: schedule indexing plan.");
|
||||
self.schedule_indexing_plan()
|
||||
.await
|
||||
.context("Error when scheduling indexing plan")?;
|
||||
|
||||
@@ -281,7 +281,7 @@ impl IndexService {
|
||||
) -> Result<SourceConfig, IndexServiceError> {
|
||||
let source_id = source_config.source_id.clone();
|
||||
// This is a bit redundant, as SourceConfig deserialization also checks
|
||||
// that the indentifier is valid. However it authorizes the special
|
||||
// that the identifier is valid. However it authorizes the special
|
||||
// private names internal to quickwit, so we do an extra check.
|
||||
validate_identifier("Source ID", &source_id).map_err(|_| {
|
||||
IndexServiceError::InvalidIdentifier(format!("Invalid source ID: `{source_id}`"))
|
||||
|
||||
@@ -19,7 +19,7 @@
|
||||
|
||||
//! This crate contains all of the building pieces that make quickwit's IO possible.
|
||||
//!
|
||||
//! - The `StorageDirectory` justs wraps a `Storage` trait to make it compatible with tantivy's
|
||||
//! - The `StorageDirectory` just wraps a `Storage` trait to make it compatible with tantivy's
|
||||
//! Directory API.
|
||||
//! - The `BundleDirectory` bundles multiple files into a single file.
|
||||
//! - The `HotDirectory` wraps another directory with a static cache.
|
||||
|
||||
@@ -843,7 +843,7 @@ mod tests {
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_deserialize_i64_mappping_singlevalued() -> anyhow::Result<()> {
|
||||
fn test_deserialize_i64_mapping_singlevalued() -> anyhow::Result<()> {
|
||||
let result = serde_json::from_str::<FieldMappingEntry>(
|
||||
r#"
|
||||
{
|
||||
|
||||
@@ -138,11 +138,11 @@ pub struct NamedField {
|
||||
|
||||
clone_trait_object!(DocMapper);
|
||||
|
||||
/// Informations about what a DocMapper think should be warmed up before
|
||||
/// Information about what a DocMapper think should be warmed up before
|
||||
/// running the query.
|
||||
#[derive(Debug, Default, Clone, PartialEq, Eq)]
|
||||
pub struct WarmupInfo {
|
||||
/// Name of fields from the term dictionnary which needs to be entirely
|
||||
/// Name of fields from the term dictionary which needs to be entirely
|
||||
/// loaded
|
||||
pub term_dict_field_names: HashSet<String>,
|
||||
/// Name of fields from the posting lists which needs to be entirely
|
||||
|
||||
@@ -186,7 +186,7 @@ mod tests {
|
||||
res.push(tok.clone());
|
||||
}
|
||||
|
||||
// latin alphabet splited on white spaces, Han splitted on each char
|
||||
// latin alphabet splited on white spaces, Han split on each char
|
||||
let expected = [
|
||||
Token {
|
||||
offset_from: 0,
|
||||
|
||||
@@ -27,7 +27,7 @@
|
||||
//! They rely on a global variable, which requires them to be executed in a single
|
||||
//! thread.
|
||||
//! For this reason, we isolate them from the other unit tests and define an
|
||||
//! independant binary target.
|
||||
//! independent binary target.
|
||||
//!
|
||||
//! They are not executed by default.
|
||||
//! They are executed in CI and can be executed locally
|
||||
@@ -138,7 +138,7 @@ async fn test_failpoint_uploader_panics_right_away() -> anyhow::Result<()> {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_failpoint_uploader_panics_after_one_sucess() -> anyhow::Result<()> {
|
||||
async fn test_failpoint_uploader_panics_after_one_success() -> anyhow::Result<()> {
|
||||
let scenario = FailScenario::setup();
|
||||
fail::cfg_callback("uploader:before", deterministic_panic_sequence(vec![true])).unwrap();
|
||||
aux_test_failpoints().await?;
|
||||
|
||||
@@ -956,7 +956,7 @@ mod tests {
|
||||
}
|
||||
universe.sleep(Duration::from_millis(100)).await;
|
||||
}
|
||||
panic!("Pipeline not exited succesfully.");
|
||||
panic!("Pipeline not exited successfully.");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
@@ -185,9 +185,9 @@ mod tests {
|
||||
merge_split_downloader_handler
|
||||
.process_pending_and_observe()
|
||||
.await;
|
||||
let merge_scratchs = merge_executor_inbox.drain_for_test();
|
||||
assert_eq!(merge_scratchs.len(), 1);
|
||||
let merge_scratch = merge_scratchs
|
||||
let merge_scratches = merge_executor_inbox.drain_for_test();
|
||||
assert_eq!(merge_scratches.len(), 1);
|
||||
let merge_scratch = merge_scratches
|
||||
.into_iter()
|
||||
.next()
|
||||
.unwrap()
|
||||
|
||||
@@ -53,7 +53,7 @@ use crate::models::{
|
||||
///
|
||||
/// This includes the following steps:
|
||||
/// - commit: this step is CPU heavy
|
||||
/// - indentifying the list of tags for the splits, and labelling it accordingly
|
||||
/// - identifying the list of tags for the splits, and labelling it accordingly
|
||||
/// - creating a bundle file
|
||||
/// - computing the hotcache
|
||||
/// - appending it to the split file.
|
||||
|
||||
@@ -31,7 +31,7 @@ use tokio::sync::oneshot;
|
||||
/// queue of message by sending `oneshot::Receiver<Message>` to the `Sequencer`.
|
||||
///
|
||||
/// The Sequencer then simply resolves these messages and forwards them to the
|
||||
/// targetted actor.
|
||||
/// targeted actor.
|
||||
///
|
||||
/// It is used by the uploader actor, to run uploads concurrently and yet
|
||||
/// ensures that publish message are send in the right order.
|
||||
|
||||
@@ -244,7 +244,7 @@ impl Actor for Uploader {
|
||||
// a queue capacity.
|
||||
//
|
||||
// Having a large queue is costly too, because each message is a handle over
|
||||
// a split directory. We DO need agressive backpressure here.
|
||||
// a split directory. We DO need aggressive backpressure here.
|
||||
QueueCapacity::Bounded(0)
|
||||
}
|
||||
|
||||
@@ -275,7 +275,7 @@ impl Handler<PackagedSplitBatch> for Uploader {
|
||||
// This is not a valid usage of protected zone here.
|
||||
//
|
||||
// Protected zone are supposed to be used when the cause for blocking is
|
||||
// outside of the responsability of the current actor.
|
||||
// outside of the responsibility of the current actor.
|
||||
// For instance, when sending a message on a downstream actor with a saturated
|
||||
// mailbox.
|
||||
// This is meant to be fixed with ParallelActors.
|
||||
|
||||
@@ -602,7 +602,7 @@ fn spawn_consumer_poll_loop(
|
||||
// consumer might not call `poll()` for a duration that exceeds
|
||||
// `max.poll.interval.ms`. When that happens the consumer is kicked out of the group
|
||||
// and the source fails. This should not happen in practice with a
|
||||
// sufficiently large value for `max.poll.interval.ms`. The defaut value is 5
|
||||
// sufficiently large value for `max.poll.interval.ms`. The default value is 5
|
||||
// minutes.
|
||||
if events_tx.blocking_send(event).is_err() {
|
||||
break;
|
||||
@@ -1191,7 +1191,7 @@ mod kafka_broker_tests {
|
||||
create_topic(&admin_client, &topic, 1).await.unwrap();
|
||||
|
||||
let metastore = metastore_for_test();
|
||||
let index_id = append_random_suffix("test-kafka-source--process-partiton-eof--index");
|
||||
let index_id = append_random_suffix("test-kafka-source--process-partition-eof--index");
|
||||
let (_source_id, source_config) = get_source_config(&topic);
|
||||
let params = if let SourceParams::Kafka(params) = source_config.clone().source_params {
|
||||
params
|
||||
|
||||
@@ -191,7 +191,7 @@ impl IndexingSplitStore {
|
||||
/// # Implementation detail:
|
||||
///
|
||||
/// Depending on whether the split was obtained from the `Storage`
|
||||
/// or the cache, it could consist in a direclty or a proper split file.
|
||||
/// or the cache, it could consist in a directly or a proper split file.
|
||||
/// This method takes care of the dealing with opening the split correctly.
|
||||
///
|
||||
/// As we fetch the split, we optimistically assume that this is for a merge
|
||||
|
||||
@@ -210,7 +210,7 @@ impl InnerLocalSplitStore {
|
||||
|
||||
/// Tries to move a `split_folder` file into the cache.
|
||||
///
|
||||
/// Move is not an image here. We are litterally moving the directory.
|
||||
/// Move is not an image here. We are literally moving the directory.
|
||||
///
|
||||
/// If the cache capacity does not allow it, this function
|
||||
/// just logs a warning and returns Ok(false).
|
||||
@@ -396,7 +396,7 @@ impl LocalSplitStore {
|
||||
|
||||
/// Tries to move a `split_folder` file into the cache.
|
||||
///
|
||||
/// Move is not an image here. We are litterally moving the directory.
|
||||
/// Move is not an image here. We are literally moving the directory.
|
||||
///
|
||||
/// If the cache capacity does not allow it, this function
|
||||
/// just logs a warning and returns Ok(false).
|
||||
|
||||
@@ -524,7 +524,7 @@ pub mod ingest_service_grpc_client {
|
||||
/// /
|
||||
/// / Fetching does not necessarily return all of the available records.
|
||||
/// / If returning all records would exceed `FETCH_PAYLOAD_LIMIT` (2MB),
|
||||
/// / the reponse will be partial.
|
||||
/// / the response will be partial.
|
||||
pub async fn fetch(
|
||||
&mut self,
|
||||
request: impl tonic::IntoRequest<super::FetchRequest>,
|
||||
@@ -597,7 +597,7 @@ pub mod ingest_service_grpc_server {
|
||||
/// /
|
||||
/// / Fetching does not necessarily return all of the available records.
|
||||
/// / If returning all records would exceed `FETCH_PAYLOAD_LIMIT` (2MB),
|
||||
/// / the reponse will be partial.
|
||||
/// / the response will be partial.
|
||||
async fn fetch(
|
||||
&self,
|
||||
request: tonic::Request<super::FetchRequest>,
|
||||
|
||||
@@ -111,7 +111,7 @@ impl IngestApiService {
|
||||
});
|
||||
}
|
||||
|
||||
let (memory, disk) = self.queues.ressource_usage();
|
||||
let (memory, disk) = self.queues.resource_usage();
|
||||
if memory > self.memory_limit {
|
||||
info!("Ingestion rejected due to memory limits");
|
||||
return Err(IngestServiceError::RateLimited);
|
||||
|
||||
@@ -41,7 +41,7 @@ service IngestService {
|
||||
///
|
||||
/// Fetching does not necessarily return all of the available records.
|
||||
/// If returning all records would exceed `FETCH_PAYLOAD_LIMIT` (2MB),
|
||||
/// the reponse will be partial.
|
||||
/// the response will be partial.
|
||||
rpc Fetch(FetchRequest) returns (FetchResponse);
|
||||
|
||||
/// Returns a batch containing the last records.
|
||||
|
||||
@@ -207,10 +207,10 @@ impl Queues {
|
||||
})
|
||||
}
|
||||
|
||||
/// Get ressource used by the queue.
|
||||
/// Get resource used by the queue.
|
||||
///
|
||||
/// Returns the in-memory size, and the on disk size of the queue.
|
||||
pub fn ressource_usage(&self) -> (usize, usize) {
|
||||
pub fn resource_usage(&self) -> (usize, usize) {
|
||||
(
|
||||
self.record_log.in_memory_size(),
|
||||
self.record_log.on_disk_size(),
|
||||
|
||||
@@ -271,7 +271,7 @@ impl DeleteTaskPlanner {
|
||||
Ok((splits_with_deletes, splits_without_deletes))
|
||||
}
|
||||
|
||||
/// Executes a `LeafSearchRequet` on the split and returns true
|
||||
/// Executes a `LeafSearchRequest` on the split and returns true
|
||||
/// if it matches documents.
|
||||
async fn has_split_docs_to_delete(
|
||||
&self,
|
||||
|
||||
@@ -173,7 +173,7 @@ impl Handler<SuperviseLoop> for DeleteTaskService {
|
||||
) -> Result<(), ActorExitStatus> {
|
||||
let result = self.update_pipeline_handles(ctx).await;
|
||||
if let Err(error) = result {
|
||||
error!("Delete task pipelines udpate failed: {}", error);
|
||||
error!("Delete task pipelines update failed: {}", error);
|
||||
}
|
||||
ctx.schedule_self_msg(HEARTBEAT, SuperviseLoop).await;
|
||||
Ok(())
|
||||
|
||||
@@ -376,7 +376,7 @@ mod tests {
|
||||
let garbage_collect_actor =
|
||||
GarbageCollector::new(Arc::new(mock_metastore), storage_resolver);
|
||||
let universe = Universe::with_accelerated_time();
|
||||
let (_maibox, handler) = universe.spawn_builder().spawn(garbage_collect_actor);
|
||||
let (_mailbox, handler) = universe.spawn_builder().spawn(garbage_collect_actor);
|
||||
|
||||
let state_after_initialization = handler.process_pending_and_observe().await.state;
|
||||
assert_eq!(state_after_initialization.num_passes, 1);
|
||||
@@ -437,7 +437,7 @@ mod tests {
|
||||
let garbage_collect_actor =
|
||||
GarbageCollector::new(Arc::new(mock_metastore), storage_resolver);
|
||||
let universe = Universe::with_accelerated_time();
|
||||
let (_maibox, handle) = universe.spawn_builder().spawn(garbage_collect_actor);
|
||||
let (_mailbox, handle) = universe.spawn_builder().spawn(garbage_collect_actor);
|
||||
|
||||
let counters = handle.process_pending_and_observe().await.state;
|
||||
assert_eq!(counters.num_passes, 1);
|
||||
@@ -488,7 +488,7 @@ mod tests {
|
||||
let garbage_collect_actor =
|
||||
GarbageCollector::new(Arc::new(mock_metastore), storage_resolver);
|
||||
let universe = Universe::with_accelerated_time();
|
||||
let (_maibox, handle) = universe.spawn_builder().spawn(garbage_collect_actor);
|
||||
let (_mailbox, handle) = universe.spawn_builder().spawn(garbage_collect_actor);
|
||||
|
||||
let counters = handle.process_pending_and_observe().await.state;
|
||||
assert_eq!(counters.num_passes, 1);
|
||||
@@ -520,7 +520,7 @@ mod tests {
|
||||
let garbage_collect_actor =
|
||||
GarbageCollector::new(Arc::new(mock_metastore), storage_resolver);
|
||||
let universe = Universe::with_accelerated_time();
|
||||
let (_maibox, handle) = universe.spawn_builder().spawn(garbage_collect_actor);
|
||||
let (_mailbox, handle) = universe.spawn_builder().spawn(garbage_collect_actor);
|
||||
|
||||
let counters = handle.process_pending_and_observe().await.state;
|
||||
assert_eq!(counters.num_passes, 1);
|
||||
@@ -589,7 +589,7 @@ mod tests {
|
||||
let garbage_collect_actor =
|
||||
GarbageCollector::new(Arc::new(mock_metastore), storage_resolver);
|
||||
let universe = Universe::with_accelerated_time();
|
||||
let (_maibox, handle) = universe.spawn_builder().spawn(garbage_collect_actor);
|
||||
let (_mailbox, handle) = universe.spawn_builder().spawn(garbage_collect_actor);
|
||||
|
||||
let counters = handle.process_pending_and_observe().await.state;
|
||||
assert_eq!(counters.num_passes, 1);
|
||||
@@ -661,7 +661,7 @@ mod tests {
|
||||
let garbage_collect_actor =
|
||||
GarbageCollector::new(Arc::new(mock_metastore), storage_resolver);
|
||||
let universe = Universe::with_accelerated_time();
|
||||
let (_maibox, handle) = universe.spawn_builder().spawn(garbage_collect_actor);
|
||||
let (_mailbox, handle) = universe.spawn_builder().spawn(garbage_collect_actor);
|
||||
|
||||
let counters = handle.process_pending_and_observe().await.state;
|
||||
assert_eq!(counters.num_passes, 1);
|
||||
|
||||
@@ -200,7 +200,7 @@ impl IndexCheckpoint {
|
||||
|
||||
/// A source checkpoint is a map of the last processed position for every partition.
|
||||
///
|
||||
/// If a partition is missing, it implicitely means that none of its message
|
||||
/// If a partition is missing, it implicitly means that none of its message
|
||||
/// has been processed.
|
||||
#[derive(Default, Clone, Eq, PartialEq)]
|
||||
pub struct SourceCheckpoint {
|
||||
|
||||
+1
-1
@@ -29,7 +29,7 @@ use super::store_operations::fetch_index;
|
||||
use crate::MetastoreResult;
|
||||
|
||||
/// Lazy [`FileBackedIndex`]. It loads a `FileBackedIndex`
|
||||
/// on demand and optionaly spawns a task to poll
|
||||
/// on demand and optionally spawns a task to poll
|
||||
/// regularly the storage and update the index.
|
||||
pub(crate) struct LazyFileBackedIndex {
|
||||
index_id: String,
|
||||
|
||||
@@ -65,7 +65,7 @@ pub(crate) enum IndexState {
|
||||
/// Metastore that stores all of the metadata associated to each index
|
||||
/// into as many files and stores a map of indexes
|
||||
/// (index_id, index_state) in a dedicated file `indexes_states.json`.
|
||||
/// An `IndexState` describes the lifecyle of an index: `Creating` and
|
||||
/// An `IndexState` describes the lifecycle of an index: `Creating` and
|
||||
/// `Deleting` are transitioning states that indicates that index is not
|
||||
/// yet available. On the contrary, `Alive` state indicates the index is ready
|
||||
/// to be retrieved / updated.
|
||||
@@ -240,7 +240,7 @@ impl FileBackedMetastore {
|
||||
return get_index_mutex(index_id, index_state).await;
|
||||
}
|
||||
|
||||
// We need to instanciate a `LazyFileBackedIndex` that will hold the mutex
|
||||
// We need to instantiate a `LazyFileBackedIndex` that will hold the mutex
|
||||
// and take care of spawning the polling if needed.
|
||||
let index = index_result?;
|
||||
let lazy_index = LazyFileBackedIndex::new(
|
||||
|
||||
@@ -635,8 +635,8 @@ mod tests {
|
||||
metastore: Arc<dyn Metastore>,
|
||||
) -> anyhow::Result<()> {
|
||||
tokio::spawn(async move {
|
||||
let grpc_adpater = GrpcMetastoreAdapter::from(metastore);
|
||||
let service = MetastoreApiServiceServer::new(grpc_adpater);
|
||||
let grpc_adapter = GrpcMetastoreAdapter::from(metastore);
|
||||
let service = MetastoreApiServiceServer::new(grpc_adapter);
|
||||
Server::builder()
|
||||
.add_service(service)
|
||||
.serve(address)
|
||||
|
||||
@@ -75,7 +75,7 @@ use crate::{MetastoreError, MetastoreResult, Split, SplitMetadata, SplitState};
|
||||
/// applied to all the splits of the index.
|
||||
///
|
||||
/// Quickwit needs a way to track that a delete task has been applied to a split. This is ensured
|
||||
/// by two mecanisms:
|
||||
/// by two mechanisms:
|
||||
/// - On creation of a delete task, we give to the task a monotically increasing opstamp (uniqueness
|
||||
/// and monotonically increasing must be true at the index level).
|
||||
/// - When a delete task is executed on a split, that is when the documents matched by the search
|
||||
@@ -85,7 +85,7 @@ use crate::{MetastoreError, MetastoreResult, Split, SplitMetadata, SplitState};
|
||||
/// `delete_optstamp` will be inferior to the `opstamp` of the new tasks.
|
||||
///
|
||||
/// For splits created after a given delete task, Quickwit's indexing ensures that these splits
|
||||
/// are created with a `delete_optstamp` equal the lastest opstamp of the tasks of the
|
||||
/// are created with a `delete_optstamp` equal the latest opstamp of the tasks of the
|
||||
/// corresponding index.
|
||||
#[cfg_attr(any(test, feature = "testsuite"), mockall::automock)]
|
||||
#[async_trait]
|
||||
|
||||
@@ -719,7 +719,7 @@ pub mod test_suite {
|
||||
cleanup_index(&metastore, &index_id).await;
|
||||
}
|
||||
|
||||
// Publish a published split and non-existant split on an index
|
||||
// Publish a published split and non-existent split on an index
|
||||
{
|
||||
metastore.create_index(index_config.clone()).await.unwrap();
|
||||
|
||||
@@ -760,7 +760,7 @@ pub mod test_suite {
|
||||
cleanup_index(&metastore, &index_id).await;
|
||||
}
|
||||
|
||||
// Publish a non-staged split and non-existant split on an index
|
||||
// Publish a non-staged split and non-existent split on an index
|
||||
{
|
||||
metastore.create_index(index_config.clone()).await.unwrap();
|
||||
|
||||
|
||||
@@ -121,7 +121,7 @@ message SearchResponse {
|
||||
// server-side and expressed in microseconds.
|
||||
uint64 elapsed_time_micros = 3;
|
||||
|
||||
// The searcherrors that occured formatted as string.
|
||||
// The searcherrors that occurred formatted as string.
|
||||
repeated string errors = 4;
|
||||
|
||||
// Serialized aggregation response
|
||||
@@ -130,7 +130,7 @@ message SearchResponse {
|
||||
}
|
||||
|
||||
message SplitSearchError {
|
||||
// The searcherror that occured formatted as string.
|
||||
// The searcherror that occurred formatted as string.
|
||||
string error = 1;
|
||||
|
||||
// Split id that failed.
|
||||
@@ -164,7 +164,7 @@ message SplitIdAndFooterOffsets {
|
||||
string split_id = 1;
|
||||
// The offset of the start of footer in the split bundle. The footer contains the file bundle metadata and the hotcache.
|
||||
uint64 split_footer_start = 2;
|
||||
// The offset of the end of the footer in split bundle. The footer contains the file bundle metada and the hotcache.
|
||||
// The offset of the end of the footer in split bundle. The footer contains the file bundle metadata and the hotcache.
|
||||
uint64 split_footer_end = 3;
|
||||
|
||||
}
|
||||
@@ -302,7 +302,7 @@ message ListTermsResponse {
|
||||
// server-side and expressed in microseconds.
|
||||
uint64 elapsed_time_micros = 3;
|
||||
|
||||
// The searcherrors that occured formatted as string.
|
||||
// The searcherrors that occurred formatted as string.
|
||||
repeated string errors = 4;
|
||||
}
|
||||
|
||||
|
||||
@@ -55,7 +55,7 @@ pub struct SearchResponse {
|
||||
/// server-side and expressed in microseconds.
|
||||
#[prost(uint64, tag = "3")]
|
||||
pub elapsed_time_micros: u64,
|
||||
/// The searcherrors that occured formatted as string.
|
||||
/// The searcherrors that occurred formatted as string.
|
||||
#[prost(string, repeated, tag = "4")]
|
||||
pub errors: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
|
||||
/// Serialized aggregation response
|
||||
@@ -66,7 +66,7 @@ pub struct SearchResponse {
|
||||
#[allow(clippy::derive_partial_eq_without_eq)]
|
||||
#[derive(Clone, PartialEq, ::prost::Message)]
|
||||
pub struct SplitSearchError {
|
||||
/// The searcherror that occured formatted as string.
|
||||
/// The searcherror that occurred formatted as string.
|
||||
#[prost(string, tag = "1")]
|
||||
pub error: ::prost::alloc::string::String,
|
||||
/// Split id that failed.
|
||||
@@ -107,7 +107,7 @@ pub struct SplitIdAndFooterOffsets {
|
||||
/// The offset of the start of footer in the split bundle. The footer contains the file bundle metadata and the hotcache.
|
||||
#[prost(uint64, tag = "2")]
|
||||
pub split_footer_start: u64,
|
||||
/// The offset of the end of the footer in split bundle. The footer contains the file bundle metada and the hotcache.
|
||||
/// The offset of the end of the footer in split bundle. The footer contains the file bundle metadata and the hotcache.
|
||||
#[prost(uint64, tag = "3")]
|
||||
pub split_footer_end: u64,
|
||||
}
|
||||
@@ -278,7 +278,7 @@ pub struct ListTermsResponse {
|
||||
/// server-side and expressed in microseconds.
|
||||
#[prost(uint64, tag = "3")]
|
||||
pub elapsed_time_micros: u64,
|
||||
/// The searcherrors that occured formatted as string.
|
||||
/// The searcherrors that occurred formatted as string.
|
||||
#[prost(string, repeated, tag = "4")]
|
||||
pub errors: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
|
||||
}
|
||||
|
||||
@@ -54,7 +54,7 @@ impl Transport {
|
||||
pub fn new(endpoint: Url) -> Self {
|
||||
let base_url = endpoint
|
||||
.join("api/v1/")
|
||||
.expect("Endpoint shoud not be malformed.");
|
||||
.expect("Endpoint should not be malformed.");
|
||||
Self {
|
||||
base_url,
|
||||
client: Client::new(),
|
||||
|
||||
@@ -29,7 +29,7 @@ use tokio_stream::wrappers::UnboundedReceiverStream;
|
||||
use tracing::debug;
|
||||
|
||||
use crate::retry::search::LeafSearchRetryPolicy;
|
||||
use crate::retry::search_stream::{LeafSearchStreamRetryPolicy, SuccessfullSplitIds};
|
||||
use crate::retry::search_stream::{LeafSearchStreamRetryPolicy, SuccessfulSplitIds};
|
||||
use crate::retry::{retry_client, DefaultRetryPolicy, RetryPolicy};
|
||||
use crate::{SearchError, SearchJobPlacer, SearchServiceClient};
|
||||
|
||||
@@ -207,7 +207,7 @@ async fn forward_leaf_search_stream(
|
||||
mut stream: UnboundedReceiverStream<crate::Result<LeafSearchStreamResponse>>,
|
||||
sender: UnboundedSender<crate::Result<LeafSearchStreamResponse>>,
|
||||
send_error: bool,
|
||||
) -> Result<SuccessfullSplitIds, SendError<crate::Result<LeafSearchStreamResponse>>> {
|
||||
) -> Result<SuccessfulSplitIds, SendError<crate::Result<LeafSearchStreamResponse>>> {
|
||||
let mut successful_split_ids: Vec<String> = Vec::new();
|
||||
while let Some(result) = stream.next().await {
|
||||
match result {
|
||||
@@ -222,7 +222,7 @@ async fn forward_leaf_search_stream(
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(SuccessfullSplitIds(successful_split_ids))
|
||||
Ok(SuccessfulSplitIds(successful_split_ids))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
|
||||
@@ -23,7 +23,7 @@ use tracing::warn;
|
||||
|
||||
use super::RetryPolicy;
|
||||
|
||||
pub struct SuccessfullSplitIds(pub Vec<String>);
|
||||
pub struct SuccessfulSplitIds(pub Vec<String>);
|
||||
|
||||
/// Retry policy for consuming the result stream of a LeafSearchStreamRequest.
|
||||
/// A retry is only made if there are some missing splits.
|
||||
@@ -33,7 +33,7 @@ pub struct LeafSearchStreamRetryPolicy {}
|
||||
impl
|
||||
RetryPolicy<
|
||||
LeafSearchStreamRequest,
|
||||
SuccessfullSplitIds,
|
||||
SuccessfulSplitIds,
|
||||
SendError<crate::Result<LeafSearchStreamResponse>>,
|
||||
> for LeafSearchStreamRetryPolicy
|
||||
{
|
||||
@@ -44,12 +44,12 @@ impl
|
||||
&self,
|
||||
mut request: LeafSearchStreamRequest,
|
||||
response_res: &Result<
|
||||
SuccessfullSplitIds,
|
||||
SuccessfulSplitIds,
|
||||
SendError<crate::Result<LeafSearchStreamResponse>>,
|
||||
>,
|
||||
) -> Option<LeafSearchStreamRequest> {
|
||||
match response_res {
|
||||
Ok(SuccessfullSplitIds(successful_split_ids)) => {
|
||||
Ok(SuccessfulSplitIds(successful_split_ids)) => {
|
||||
if successful_split_ids.len() == request.split_offsets.len() {
|
||||
// All splits were successful!
|
||||
return None;
|
||||
@@ -80,7 +80,7 @@ mod tests {
|
||||
};
|
||||
use tokio::sync::mpsc::error::SendError;
|
||||
|
||||
use crate::retry::search_stream::{LeafSearchStreamRetryPolicy, SuccessfullSplitIds};
|
||||
use crate::retry::search_stream::{LeafSearchStreamRetryPolicy, SuccessfulSplitIds};
|
||||
use crate::retry::RetryPolicy;
|
||||
|
||||
#[tokio::test]
|
||||
@@ -97,7 +97,7 @@ mod tests {
|
||||
async fn test_retry_policy_search_stream_should_not_retry_on_successful_response() {
|
||||
let retry_policy = LeafSearchStreamRetryPolicy {};
|
||||
let request = LeafSearchStreamRequest::default();
|
||||
let response_res = Ok(SuccessfullSplitIds(Vec::new()));
|
||||
let response_res = Ok(SuccessfulSplitIds(Vec::new()));
|
||||
let retry_req_opt = retry_policy.retry_request(request, &response_res);
|
||||
assert!(retry_req_opt.is_none());
|
||||
}
|
||||
@@ -119,7 +119,7 @@ mod tests {
|
||||
split_offsets: vec![split_1, split_2],
|
||||
..Default::default()
|
||||
};
|
||||
let response_res = Ok(SuccessfullSplitIds(vec!["split_1".to_string()]));
|
||||
let response_res = Ok(SuccessfulSplitIds(vec!["split_1".to_string()]));
|
||||
let retry_req = retry_policy.retry_request(request, &response_res).unwrap();
|
||||
assert_eq!(retry_req.split_offsets.len(), 1);
|
||||
assert_eq!(retry_req.split_offsets[0].split_id, "split_2");
|
||||
|
||||
@@ -35,7 +35,7 @@ use crate::SearchServiceClient;
|
||||
/// The `split_id` is used to define an affinity between a leaf nodes and a job.
|
||||
/// The `cost` is used to spread the work evenly amongst nodes.
|
||||
pub trait Job {
|
||||
/// SplitId of the split that is targetted.
|
||||
/// SplitId of the split that is targeted.
|
||||
fn split_id(&self) -> &str;
|
||||
/// Estimation of the load associated with running a given job.
|
||||
///
|
||||
@@ -102,7 +102,7 @@ impl SearchJobPlacer {
|
||||
// TODO optimize the case where there are few jobs and many clients.
|
||||
let clients = self.clients();
|
||||
|
||||
// when exclude_addresses excludes all adresses we discard it
|
||||
// when exclude_addresses excludes all addresses we discard it
|
||||
let empty_set = HashSet::default();
|
||||
let exclude_addresses_if_not_saturated = if exclude_addresses.len() == clients.len() {
|
||||
&empty_set
|
||||
|
||||
@@ -47,7 +47,7 @@ use crate::{Result, SearchError};
|
||||
// Note: we return a stream of a result with a tonic::Status error
|
||||
// to be compatible with the stream coming from the grpc client.
|
||||
// It would be better to have a SearchError but we need then
|
||||
// to process stream in grpc_adapater.rs to change SearchError
|
||||
// to process stream in grpc_adapter.rs to change SearchError
|
||||
// to tonic::Status as tonic::Status is required by the stream result
|
||||
// signature defined by proto generated code.
|
||||
pub async fn leaf_search_stream(
|
||||
|
||||
@@ -281,7 +281,7 @@ impl SearchService for SearchServiceImpl {
|
||||
}
|
||||
|
||||
/// [`SearcherContext`] provides a common set of variables
|
||||
/// shared by a searcher instance (which instanciates a
|
||||
/// shared by a searcher instance (which instantiates a
|
||||
/// [`SearchServiceImpl`]).
|
||||
pub struct SearcherContext {
|
||||
/// Searcher config.
|
||||
|
||||
@@ -14,7 +14,7 @@ The API is split into:
|
||||
The APIs are usually accessible both via gRPC and REST.
|
||||
This is done consistently using the following pattern.
|
||||
|
||||
A service async trait mimicks the tonic service api, but without the `tonic`
|
||||
A service async trait mimics the tonic service api, but without the `tonic`
|
||||
wrapping of the request and with a rich and specific error type instead of tonic::Status.
|
||||
The argument and the response on the other hand are typically using protobuf object
|
||||
directly whenever sensible.
|
||||
@@ -31,9 +31,9 @@ The rest API then relies on calling this service.
|
||||
|
||||
```mermaid
|
||||
graph TD
|
||||
grpc_ser[grpc::BlopService] --> |wraps| ser(Go shopping)
|
||||
rest[blop_handler] --> |calls| ser(Go shopping)
|
||||
ser[BlopService]
|
||||
grpc_service[grpc::BlopService] --> |wraps| service(Go shopping)
|
||||
rest[blop_handler] --> |calls| service(Go shopping)
|
||||
service[BlopService]
|
||||
```
|
||||
|
||||
## UI
|
||||
|
||||
@@ -45,7 +45,7 @@ const GENERATED_FILE_HEADER: &str = r#"
|
||||
// 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/>.
|
||||
|
||||
/// This file is auto-generated, any change can be overriden.
|
||||
/// This file is auto-generated, any change can be overridden.
|
||||
"#;
|
||||
|
||||
pub static ROOT_DIR: Lazy<path::PathBuf> = Lazy::new(|| {
|
||||
|
||||
@@ -20,7 +20,7 @@
|
||||
use std::str::FromStr;
|
||||
|
||||
use quickwit_common::simple_list::{from_simple_list, to_simple_list, SimpleList};
|
||||
/// This file is auto-generated, any change can be overriden.
|
||||
/// This file is auto-generated, any change can be overridden.
|
||||
use serde::{Deserialize, Serialize};
|
||||
use warp::{Filter, Rejection};
|
||||
|
||||
|
||||
@@ -1103,7 +1103,7 @@ mod tests {
|
||||
.await;
|
||||
assert_eq!(resp.status(), 405);
|
||||
|
||||
// Check get a non exising source returns 404.
|
||||
// Check get a non existing source returns 404.
|
||||
let resp = warp::test::request()
|
||||
.path("/indexes/hdfs-logs/sources/file-source")
|
||||
.method("GET")
|
||||
|
||||
@@ -232,7 +232,7 @@ pub fn elastic_bulk_handler(
|
||||
post,
|
||||
tag = "Ingest",
|
||||
path = "/_bulk",
|
||||
request_body(content = String, description = "Elasticsearch compatible bulk resquest body limited to 10MB", content_type = "application/json"),
|
||||
request_body(content = String, description = "Elasticsearch compatible bulk request body limited to 10MB", content_type = "application/json"),
|
||||
responses(
|
||||
(status = 200, description = "Successfully ingested documents.", body = IngestResponse)
|
||||
),
|
||||
|
||||
@@ -82,7 +82,7 @@ pub use crate::metrics::SERVE_METRICS;
|
||||
use crate::rest::recover_fn;
|
||||
pub use crate::search_api::{SearchRequestQueryString, SortByField};
|
||||
|
||||
const READYNESS_REPORTING_INTERVAL: Duration = if cfg!(any(test, feature = "testsuite")) {
|
||||
const READINESS_REPORTING_INTERVAL: Duration = if cfg!(any(test, feature = "testsuite")) {
|
||||
Duration::from_millis(25)
|
||||
} else {
|
||||
Duration::from_secs(10)
|
||||
@@ -118,8 +118,8 @@ pub async fn serve_quickwit(config: QuickwitConfig) -> anyhow::Result<()> {
|
||||
let cluster =
|
||||
quickwit_cluster::start_cluster_service(&config, &config.enabled_services).await?;
|
||||
|
||||
// Instanciate either a file-backed or postgresql [`Metastore`] if the node runs a `Metastore`
|
||||
// service, else instanciate a [`MetastoreGrpcClient`].
|
||||
// Instantiate either a file-backed or postgresql [`Metastore`] if the node runs a `Metastore`
|
||||
// service, else instantiate a [`MetastoreGrpcClient`].
|
||||
let metastore: Arc<dyn Metastore> = if config
|
||||
.enabled_services
|
||||
.contains(&QuickwitService::Metastore)
|
||||
@@ -298,7 +298,7 @@ fn with_arg<T: Clone + Send>(arg: T) -> impl Filter<Extract = (T,), Error = Infa
|
||||
|
||||
/// Reports node readiness to chitchat cluster every 10 seconds (25 ms for tests).
|
||||
async fn node_readiness_reporting_task(cluster: Arc<Cluster>, metastore: Arc<dyn Metastore>) {
|
||||
let mut interval = tokio::time::interval(READYNESS_REPORTING_INTERVAL);
|
||||
let mut interval = tokio::time::interval(READINESS_REPORTING_INTERVAL);
|
||||
loop {
|
||||
interval.tick().await;
|
||||
let node_ready = metastore.check_connectivity().await.is_ok();
|
||||
|
||||
@@ -50,10 +50,10 @@ mod tests {
|
||||
address: SocketAddr,
|
||||
search_service: Arc<dyn SearchService>,
|
||||
) -> anyhow::Result<()> {
|
||||
let search_grpc_adpater = GrpcSearchAdapter::from(search_service);
|
||||
let search_grpc_adapter = GrpcSearchAdapter::from(search_service);
|
||||
tokio::spawn(async move {
|
||||
Server::builder()
|
||||
.add_service(SearchServiceServer::new(search_grpc_adpater))
|
||||
.add_service(SearchServiceServer::new(search_grpc_adapter))
|
||||
.serve(address)
|
||||
.await?;
|
||||
Result::<_, anyhow::Error>::Ok(())
|
||||
|
||||
@@ -93,7 +93,7 @@ impl QuickwitRestClient {
|
||||
|
||||
async fn parse_body<T: DeserializeOwned>(mut response: Response<Body>) -> anyhow::Result<T> {
|
||||
if response.status() != StatusCode::OK {
|
||||
anyhow::bail!("Unexepected status {}", response.status());
|
||||
anyhow::bail!("Unexpected status {}", response.status());
|
||||
}
|
||||
let mut body = Vec::new();
|
||||
while let Some(chunk) = response.body_mut().next().await {
|
||||
|
||||
+1
-1
@@ -258,7 +258,7 @@ pub struct ByteRangeCache {
|
||||
}
|
||||
|
||||
impl ByteRangeCache {
|
||||
/// Creates a slice cache that nevers removes any entry.
|
||||
/// Creates a slice cache that never removes any entry.
|
||||
pub fn with_infinite_capacity(cache_counters: &'static CacheMetrics) -> Self {
|
||||
ByteRangeCache {
|
||||
inner: Mutex::new(NeedMutByteRangeCache::with_infinite_capacity(
|
||||
|
||||
@@ -195,7 +195,7 @@ impl<K: Hash + Eq> MemorySizedCache<K> {
|
||||
}
|
||||
}
|
||||
|
||||
/// Creates a slice cache that nevers removes any entry.
|
||||
/// Creates a slice cache that never removes any entry.
|
||||
pub fn with_infinite_capacity(cache_counters: &'static CacheMetrics) -> Self {
|
||||
MemorySizedCache {
|
||||
inner: Mutex::new(NeedMutMemorySizedCache::with_capacity(
|
||||
|
||||
@@ -129,7 +129,7 @@ fn delete_all_dirs_if_empty<'a>(
|
||||
let full_path = root.join(path);
|
||||
let path_entries_result = full_path.read_dir();
|
||||
if let Err(err) = &path_entries_result {
|
||||
// Ignore `ErrorKind::NotFound` as this could be deleted by another concurent task.
|
||||
// Ignore `ErrorKind::NotFound` as this could be deleted by another concurrent task.
|
||||
if err.kind() == ErrorKind::NotFound {
|
||||
return Ok(());
|
||||
}
|
||||
@@ -142,7 +142,7 @@ fn delete_all_dirs_if_empty<'a>(
|
||||
|
||||
let delete_result = fs::remove_dir(full_path).await;
|
||||
if let Err(err) = &delete_result {
|
||||
// Ignore `ErrorKind::NotFound` as this could be deleted by another concurent task.
|
||||
// Ignore `ErrorKind::NotFound` as this could be deleted by another concurrent task.
|
||||
if err.kind() == ErrorKind::NotFound {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
@@ -28,7 +28,7 @@ pub struct MultiPartPolicy {
|
||||
pub target_part_num_bytes: usize,
|
||||
/// Maximum number of parts allowed.
|
||||
pub max_num_parts: usize,
|
||||
/// Threshold above which multipart is trigged.
|
||||
/// Threshold above which multipart is triggered.
|
||||
pub multipart_threshold_num_bytes: u64,
|
||||
/// Maximum size allowed for an object.
|
||||
pub max_object_num_bytes: u64,
|
||||
@@ -45,7 +45,7 @@ impl MultiPartPolicy {
|
||||
pub fn part_num_bytes(&self, len: u64) -> u64 {
|
||||
assert!(
|
||||
len < self.max_object_num_bytes,
|
||||
"This objet storage does not support object of that size {}",
|
||||
"This object storage does not support object of that size {}",
|
||||
self.max_object_num_bytes
|
||||
);
|
||||
assert!(
|
||||
|
||||
@@ -121,7 +121,7 @@ impl SplitPayloadBuilder {
|
||||
/// Adds the file to the bundle file.
|
||||
///
|
||||
/// The hotcache needs to be the last file that is added, in order to be able to read
|
||||
/// the hotcache and the metadata in one continous read.
|
||||
/// the hotcache and the metadata in one continuous read.
|
||||
pub fn add_file(&mut self, path: &Path) -> io::Result<()> {
|
||||
let file = std::fs::metadata(path)?;
|
||||
let file_range = self.current_offset as u64..self.current_offset as u64 + file.len();
|
||||
@@ -387,7 +387,7 @@ mod tests {
|
||||
// border 2 case skip and take cross adjacent blocks
|
||||
assert_eq!(fetch_data(&split_streamer, 1..3).await?, vec![76, 99]);
|
||||
|
||||
// border 3 case skip and take in seperate blocks with full block between
|
||||
// border 3 case skip and take in separate blocks with full block between
|
||||
assert_eq!(
|
||||
fetch_data(&split_streamer, 1..6).await?,
|
||||
vec![76, 99, 55, 44, 123]
|
||||
|
||||
@@ -142,7 +142,7 @@ pub trait Storage: Send + Sync + 'static {
|
||||
#[must_use]
|
||||
#[derive(Debug, Default, thiserror::Error)]
|
||||
pub struct BulkDeleteError {
|
||||
/// Error that ocurred for a whole batch and caused the entire deletion operation to be
|
||||
/// Error that occurred for a whole batch and caused the entire deletion operation to be
|
||||
/// aborted.
|
||||
pub error: Option<StorageError>,
|
||||
/// List of files that were successfully deleted, including non-existing files.
|
||||
|
||||
@@ -17,7 +17,7 @@
|
||||
// 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/>.
|
||||
|
||||
// This file is an integration test that assumes that the environement
|
||||
// This file is an integration test that assumes that the environment
|
||||
// makes it possible to connect to Amazon S3's quickwit-integration-test bucket.
|
||||
|
||||
use std::path::Path;
|
||||
|
||||
@@ -39,7 +39,7 @@ function IndexAutocomplete(props: IndexMetadataProps) {
|
||||
const [value, setValue] = React.useState<IndexMetadata | null>(null);
|
||||
const [loading, setLoading] = React.useState(false);
|
||||
// We want to show the circular progress only if we are loading some results and
|
||||
// when there is no option avaiable.
|
||||
// when there is no option available.
|
||||
const showLoading = loading && options.length === 0;
|
||||
const quickwitClient = useMemo(() => new Client(), []);
|
||||
|
||||
|
||||
@@ -83,7 +83,7 @@ export function LanguageFeatures(): any {
|
||||
|
||||
// @ annotations.
|
||||
// As an example, we emit a debugging log message on these tokens.
|
||||
// Note: message are supressed during the first load -- change some lines to see them.
|
||||
// Note: message are suppressed during the first load -- change some lines to see them.
|
||||
[/@\s*[a-zA-Z_$][\w$]*/, { token: 'annotation', log: 'annotation token: $0' }],
|
||||
|
||||
// numbers
|
||||
@@ -95,7 +95,7 @@ export function LanguageFeatures(): any {
|
||||
[/[;,.]/, 'delimiter'],
|
||||
|
||||
// strings
|
||||
[/"([^"\\]|\\.)*$/, 'string.invalid' ], // non-teminated string
|
||||
[/"([^"\\]|\\.)*$/, 'string.invalid' ], // non-terminated string
|
||||
[/"/, { token: 'string.quote', bracket: '@open', next: '@string' } ],
|
||||
|
||||
// characters
|
||||
|
||||
Reference in new issue
Block a user