Zookeeper ha

Cargo.toml

@@ -37,23 +37,5 @@ opt-level = 3
 [profile.dev.package.roaring]
 opt-level = 3
 
-[profile.dev.package.lindera-ipadic-builder]
-opt-level = 3
-[profile.dev.package.encoding]
-opt-level = 3
-[profile.dev.package.yada]
-opt-level = 3
-
-[profile.release.package.lindera-ipadic-builder]
-opt-level = 3
-[profile.release.package.encoding]
-opt-level = 3
-[profile.release.package.yada]
-opt-level = 3
-
-[profile.bench.package.lindera-ipadic-builder]
-opt-level = 3
-[profile.bench.package.encoding]
-opt-level = 3
-[profile.bench.package.yada]
-opt-level = 3
+[patch.crates-io]
+strois = { git = "https://github.com/meilisearch/strois", rev = "eeb945c" }

docker-compose.yml

@@ -0,0 +1,59 @@
+version: "3.9"
+services:
+  zk1:
+    container_name: zk1
+    hostname: zk1
+    image: bitnami/zookeeper:3.7.1
+    ports:
+      - 21811:2181
+    environment:
+      - ALLOW_ANONYMOUS_LOGIN=yes
+      - ZOO_SERVER_ID=1
+      - ZOO_SERVERS=0.0.0.0:2888:3888,zk2:2888:3888,zk3:2888:3888
+  zk2:
+    container_name: zk2
+    hostname: zk2
+    image: bitnami/zookeeper:3.7.1
+    ports:
+      - 21812:2181
+    environment:
+      - ALLOW_ANONYMOUS_LOGIN=yes
+      - ZOO_SERVER_ID=2
+      - ZOO_SERVERS=zk1:2888:3888,0.0.0.0:2888:3888,zk3:2888:3888
+  zk3:
+    container_name: zk3
+    hostname: zk3
+    image: bitnami/zookeeper:3.7.1
+    ports:
+      - 21813:2181
+    environment:
+      - ALLOW_ANONYMOUS_LOGIN=yes
+      - ZOO_SERVER_ID=3
+      - ZOO_SERVERS=zk1:2888:3888,zk2:2888:3888,0.0.0.0:2888:3888
+  zoonavigator:
+    container_name: zoonavigator
+    image: elkozmon/zoonavigator
+    ports:
+      - 9000:9000
+
+# Meilisearch instances
+  # m1:
+  #   container_name: m1
+  #   hostname: m1
+  #   image: getmeili/meilisearch:prototype-zookeeper-ha-0
+  #   ports:
+  #     - 7700:7700
+  #   environment:
+  #     - MEILI_ZK_URL=zk1:2181
+  #     - MEILI_MASTER_KEY=masterkey
+  #   restart: always
+  # m2:
+  #   container_name: m2
+  #   hostname: m2
+  #   image: getmeili/meilisearch:prototype-zookeeper-ha-0
+  #   ports:
+  #     - 7701:7700
+  #   environment:
+  #     - MEILI_ZK_URL=zk2:2181
+  #     - MEILI_MASTER_KEY=masterkey
+  #   restart: always

file-store/src/lib.rs

@@ -22,20 +22,6 @@ pub enum Error {
 
 pub type Result<T> = std::result::Result<T, Error>;
 
-impl Deref for File {
-    type Target = NamedTempFile;
-
-    fn deref(&self) -> &Self::Target {
-        &self.file
-    }
-}
-
-impl DerefMut for File {
-    fn deref_mut(&mut self) -> &mut Self::Target {
-        &mut self.file
-    }
-}
-
 #[derive(Clone, Debug)]
 pub struct FileStore {
     path: PathBuf,
@@ -146,6 +132,20 @@ impl File {
     }
 }
 
+impl Deref for File {
+    type Target = NamedTempFile;
+
+    fn deref(&self) -> &Self::Target {
+        &self.file
+    }
+}
+
+impl DerefMut for File {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.file
+    }
+}
+
 #[cfg(test)]
 mod test {
     use std::io::Write;

ha_test/run.sh

@@ -0,0 +1,61 @@
+#!/bin/bash
+
+function is_everything_installed {
+  everything_ok=yes
+
+  if hash zkli 2>/dev/null; then
+    echo "✅ zkli installed"
+  else
+    everything_ok=no
+    echo "🥺 zkli is missing, please run \`cargo install zkli\`"
+  fi
+
+  if hash s3cmd 2>/dev/null; then
+    echo "✅ s3cmd installed"
+  else
+    everything_ok=no
+    echo "🥺 s3cmd is missing, see how to install it here https://s3tools.org/s3cmd"
+  fi
+
+  if [ $everything_ok = "no" ]; then
+    echo "Exiting..."
+    exit 1
+  fi
+}
+
+# param: addr of zookeeper
+function connect_to_zookeeper {
+  if ! zkli -a "$1" ls > /dev/null; then
+    echo "zkli can't connect"
+    return 1
+  fi
+}
+
+# param: addr of the s3 bucket
+function connect_to_s3 {
+  # S3_SECRET_KEY
+  # S3_ACCESS_KEY
+  # S3_HOST
+  # S3_BUCKET
+
+  s3cmd --host="$S3_HOST" --host-bucket="$S3_BUCKET" --access_key="$ACCESS_KEY" --secret_key="$S3_SECRET_KEY" ls
+
+  if $?; then
+    echo "s3cmd can't connect"
+    return 1
+  fi
+}
+
+is_everything_installed
+
+ZOOKEEPER_ADDR="localhost:2181"
+if ! connect_to_zookeeper $ZOOKEEPER_ADDR; then
+  ZOOKEEPER_ADDR="localhost:21811"
+  if ! connect_to_zookeeper $ZOOKEEPER_ADDR; then
+    echo "Can't connect to zkli"
+    exit 1
+  fi
+fi
+
+
+connect_to_s3

index-scheduler/Cargo.toml

@@ -31,6 +31,10 @@ tempfile = "3.5.0"
 thiserror = "1.0.40"
 time = { version = "0.3.20", features = ["serde-well-known", "formatting", "parsing", "macros"] }
 uuid = { version = "1.3.1", features = ["serde", "v4"] }
+tokio = { version = "1.27.0", features = ["full"] }
+zookeeper-client-sync = { git = "https://github.com/meilisearch/zookeeper-client-sync.git" }
+parking_lot = "0.12.1"
+strois = "0.0.4"
 
 [dev-dependencies]
 big_s = "1.0.2"

index-scheduler/src/batch.rs

@@ -43,7 +43,7 @@ use uuid::Uuid;
 
 use crate::autobatcher::{self, BatchKind};
 use crate::utils::{self, swap_index_uid_in_task};
-use crate::{Error, IndexScheduler, ProcessingTasks, Result, TaskId};
+use crate::{Error, IndexSchedulerInner, ProcessingTasks, Result, TaskId};
 
 /// Represents a combination of tasks that can all be processed at the same time.
 ///
@@ -198,6 +198,35 @@ impl Batch {
             | IndexDocumentDeletionByFilter { index_uid, .. } => Some(index_uid),
         }
     }
+
+    /// Return the content fields uuids associated with this batch.
+    pub fn content_uuids(&self) -> Vec<Uuid> {
+        match self {
+            Batch::TaskCancelation { .. }
+            | Batch::TaskDeletion(_)
+            | Batch::Dump(_)
+            | Batch::IndexCreation { .. }
+            | Batch::IndexDocumentDeletionByFilter { .. }
+            | Batch::IndexUpdate { .. }
+            | Batch::SnapshotCreation(_)
+            | Batch::IndexDeletion { .. }
+            | Batch::IndexSwap { .. } => vec![],
+            Batch::IndexOperation { op, .. } => match op {
+                IndexOperation::DocumentOperation { operations, .. } => operations
+                    .iter()
+                    .flat_map(|op| match op {
+                        DocumentOperation::Add(uuid) => Some(*uuid),
+                        DocumentOperation::Delete(_) => None,
+                    })
+                    .collect(),
+                IndexOperation::DocumentDeletion { .. }
+                | IndexOperation::Settings { .. }
+                | IndexOperation::DocumentClear { .. }
+                | IndexOperation::SettingsAndDocumentOperation { .. }
+                | IndexOperation::DocumentClearAndSetting { .. } => vec![],
+            },
+        }
+    }
 }
 
 impl IndexOperation {
@@ -213,7 +242,7 @@ impl IndexOperation {
     }
 }
 
-impl IndexScheduler {
+impl IndexSchedulerInner {
     /// Convert an [`BatchKind`](crate::autobatcher::BatchKind) into a [`Batch`].
     ///
     /// ## Arguments
@@ -480,8 +509,7 @@ impl IndexScheduler {
         if let Some(task_id) = to_cancel.max() {
             // We retrieve the tasks that were processing before this tasks cancelation started.
             // We must *not* reset the processing tasks before calling this method.
-            let ProcessingTasks { started_at, processing } =
-                &*self.processing_tasks.read().unwrap();
+            let ProcessingTasks { started_at, processing, .. } = &*self.processing_tasks.read();
             return Ok(Some(Batch::TaskCancelation {
                 task: self.get_task(rtxn, task_id)?.ok_or(Error::CorruptedTaskQueue)?,
                 previous_started_at: *started_at,
@@ -1392,7 +1420,7 @@ impl IndexScheduler {
     fn delete_matched_tasks(&self, wtxn: &mut RwTxn, matched_tasks: &RoaringBitmap) -> Result<u64> {
         // 1. Remove from this list the tasks that we are not allowed to delete
         let enqueued_tasks = self.get_status(wtxn, Status::Enqueued)?;
-        let processing_tasks = &self.processing_tasks.read().unwrap().processing.clone();
+        let processing_tasks = &self.processing_tasks.read().processing.clone();
 
         let all_task_ids = self.all_task_ids(wtxn)?;
         let mut to_delete_tasks = all_task_ids & matched_tasks;

index-scheduler/src/index_mapper/index_map.rs

@@ -295,6 +295,11 @@ impl IndexMap {
             "Attempt to finish deletion of an index that was being closed"
         );
     }
+
+    /// Returns the indexes that were opened by the `IndexMap`.
+    pub fn clear(&mut self) -> Vec<Index> {
+        self.available.clear().into_iter().map(|(_, (_, index))| index).collect()
+    }
 }
 
 /// Create or open an index in the specified path.
@@ -335,7 +340,8 @@ mod tests {
     impl IndexMapper {
         fn test() -> (Self, Env, IndexSchedulerHandle) {
             let (index_scheduler, handle) = IndexScheduler::test(true, vec![]);
-            (index_scheduler.index_mapper, index_scheduler.env, handle)
+            let index_scheduler = index_scheduler.inner();
+            (index_scheduler.index_mapper.clone(), index_scheduler.env.clone(), handle)
         }
     }
 

index-scheduler/src/index_mapper/mod.rs

@@ -61,7 +61,7 @@ pub struct IndexMapper {
     pub(crate) index_stats: Database<UuidCodec, SerdeJson<IndexStats>>,
 
     /// Path to the folder where the LMDB environments of each index are.
-    base_path: PathBuf,
+    pub(crate) base_path: PathBuf,
     /// The map size an index is opened with on the first time.
     index_base_map_size: usize,
     /// The quantity by which the map size of an index is incremented upon reopening, in bytes.
@@ -135,7 +135,7 @@ impl IndexMapper {
         index_growth_amount: usize,
         index_count: usize,
         enable_mdb_writemap: bool,
-        indexer_config: IndexerConfig,
+        indexer_config: Arc<IndexerConfig>,
     ) -> Result<Self> {
         let mut wtxn = env.write_txn()?;
         let index_mapping = env.create_database(&mut wtxn, Some(INDEX_MAPPING))?;
@@ -150,7 +150,7 @@ impl IndexMapper {
             index_base_map_size,
             index_growth_amount,
             enable_mdb_writemap,
-            indexer_config: Arc::new(indexer_config),
+            indexer_config,
         })
     }
 
@@ -428,6 +428,11 @@ impl IndexMapper {
         Ok(())
     }
 
+    /// Returns the indexes that were opened by the `IndexMapper`.
+    pub fn clear(&mut self) -> Vec<Index> {
+        self.index_map.write().unwrap().clear()
+    }
+
     /// The stats of an index.
     ///
     /// If available in the cache, they are directly returned.

index-scheduler/src/insta_snapshot.rs

@@ -1,5 +1,6 @@
 use std::collections::BTreeSet;
 use std::fmt::Write;
+use std::ops::Deref;
 
 use meilisearch_types::heed::types::{OwnedType, SerdeBincode, SerdeJson, Str};
 use meilisearch_types::heed::{Database, RoTxn};
@@ -8,12 +9,13 @@ use meilisearch_types::tasks::{Details, Task};
 use roaring::RoaringBitmap;
 
 use crate::index_mapper::IndexMapper;
-use crate::{IndexScheduler, Kind, Status, BEI128};
+use crate::{IndexScheduler, IndexSchedulerInner, Kind, Status, BEI128};
 
 pub fn snapshot_index_scheduler(scheduler: &IndexScheduler) -> String {
     scheduler.assert_internally_consistent();
 
-    let IndexScheduler {
+    let inner = scheduler.inner();
+    let IndexSchedulerInner {
         autobatching_enabled,
         must_stop_processing: _,
         processing_tasks,
@@ -38,13 +40,15 @@ pub fn snapshot_index_scheduler(scheduler: &IndexScheduler) -> String {
         test_breakpoint_sdr: _,
         planned_failures: _,
         run_loop_iteration: _,
-    } = scheduler;
+        zookeeper: _,
+        options: _,
+    } = inner.deref();
 
     let rtxn = env.read_txn().unwrap();
 
     let mut snap = String::new();
 
-    let processing_tasks = processing_tasks.read().unwrap().processing.clone();
+    let processing_tasks = processing_tasks.read().processing.clone();
     snap.push_str(&format!("### Autobatching Enabled = {autobatching_enabled}\n"));
     snap.push_str("### Processing Tasks:\n");
     snap.push_str(&snapshot_bitmap(&processing_tasks));