Add metastore refresh poll

Author: rdettai

quickwit/quickwit-lambda/src/indexer/environment.rs

@@ -21,13 +21,14 @@ use std::env::var;
 
 use once_cell::sync::Lazy;
 
-pub const CONFIGURATION_TEMPLATE: &str = "version: 0.8
+pub const CONFIGURATION_TEMPLATE: &str = r#"
+version: 0.8
 node_id: lambda-indexer
 cluster_id: lambda-ephemeral
 metastore_uri: s3://${QW_LAMBDA_METASTORE_BUCKET}/index
 default_index_root_uri: s3://${QW_LAMBDA_INDEX_BUCKET}/index
 data_dir: /tmp
-";
+"#;
 
 pub static INDEX_CONFIG_URI: Lazy<String> = Lazy::new(|| {
     var("QW_LAMBDA_INDEX_CONFIG_URI").expect("QW_LAMBDA_INDEX_CONFIG_URI must be set")

quickwit/quickwit-lambda/src/searcher/environment.rs

@@ -17,11 +17,12 @@
 // 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/>.
 
-pub(crate) const CONFIGURATION_TEMPLATE: &str = "version: 0.8
+pub(crate) const CONFIGURATION_TEMPLATE: &str = r#"
+version: 0.8
 node_id: lambda-searcher
-metastore_uri: s3://${QW_LAMBDA_METASTORE_BUCKET}/index
+metastore_uri: s3://${QW_LAMBDA_METASTORE_BUCKET}/index#polling_interval=${QW_LAMBDA_SEARCHER_METASTORE_POLLING_INTERVAL:-1m}
 default_index_root_uri: s3://${QW_LAMBDA_INDEX_BUCKET}/index
 data_dir: /tmp
 searcher:
   partial_request_cache_capacity: ${QW_LAMBDA_PARTIAL_REQUEST_CACHE_CAPACITY:-64M}
-";
+"#;

quickwit/quickwit-metastore/src/metastore/file_backed/lazy_file_backed_index.rs

@@ -29,9 +29,8 @@ use tracing::error;
 use super::file_backed_index::FileBackedIndex;
 use super::store_operations::{load_index, METASTORE_FILE_NAME};
 
-/// Lazy [`FileBackedIndex`]. It loads a `FileBackedIndex`
-/// on demand and optionally spawns a task to poll
-/// regularly the storage and update the index.
+/// Lazy [`FileBackedIndex`]. It loads a `FileBackedIndex` on demand. When the index is first
+/// loaded, it optionally spawns a task to regularly poll the storage and update the index.
 pub(crate) struct LazyFileBackedIndex {
     index_id: IndexId,
     storage: Arc<dyn Storage>,
@@ -48,8 +47,8 @@ impl LazyFileBackedIndex {
         file_backed_index: Option<FileBackedIndex>,
     ) -> Self {
         let index_mutex_opt = file_backed_index.map(|index| Arc::new(Mutex::new(index)));
-        // If an index is given and a polling interval is given,
-        // spawn immediately the polling task.
+        // If the polling interval is configured and the index is already loaded,
+        // spawn immediately the polling task
         if let Some(index_mutex) = &index_mutex_opt {
             if let Some(polling_interval) = polling_interval_opt {
                 spawn_index_metadata_polling_task(
@@ -68,15 +67,24 @@ impl LazyFileBackedIndex {
         }
     }
 
-    /// Get `FileBackedIndex`.
+    /// Get a syncronized `FileBackedIndex`. If the index wasn't provided on creation, we load it
+    /// lazily on the first call of this method.
     pub async fn get(&self) -> MetastoreResult<Arc<Mutex<FileBackedIndex>>> {
         self.lazy_index
-            .get_or_try_init(|| {
-                load_file_backed_index(
-                    self.storage.clone(),
-                    self.index_id.clone(),
-                    self.polling_interval_opt,
-                )
+            .get_or_try_init(|| async move {
+                let index = load_index(&*self.storage, &self.index_id).await?;
+                let index_mutex = Arc::new(Mutex::new(index));
+                // When the index is loaded lazily, the polling task is not started in the
+                // constructor so we do it here when the index is actually loaded.
+                if let Some(polling_interval) = self.polling_interval_opt {
+                    spawn_index_metadata_polling_task(
+                        self.storage.clone(),
+                        self.index_id.clone(),
+                        Arc::downgrade(&index_mutex),
+                        polling_interval,
+                    );
+                }
+                Ok(index_mutex)
             })
             .await
             .cloned()
@@ -122,6 +130,7 @@ fn spawn_index_metadata_polling_task(
 ) {
     tokio::task::spawn(async move {
         let mut interval = tokio::time::interval(polling_interval);
+        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
         interval.tick().await; //< this is to prevent fetch right after the first population of the data.
 
         while let Some(metadata_mutex) = metastore_weak.upgrade() {
@@ -130,21 +139,3 @@ fn spawn_index_metadata_polling_task(
         }
     });
 }
-
-async fn load_file_backed_index(
-    storage: Arc<dyn Storage>,
-    index_id: IndexId,
-    polling_interval_opt: Option<Duration>,
-) -> MetastoreResult<Arc<Mutex<FileBackedIndex>>> {
-    let index = load_index(&*storage, &index_id).await?;
-    let index_mutex = Arc::new(Mutex::new(index));
-    if let Some(polling_interval) = polling_interval_opt {
-        spawn_index_metadata_polling_task(
-            storage.clone(),
-            index_id.clone(),
-            Arc::downgrade(&index_mutex),
-            polling_interval,
-        );
-    }
-    Ok(index_mutex)
-}