Introduced the Index mode for Parseable server

Author: parmesant

src/catalog/mod.rs

@@ -340,6 +340,7 @@ pub async fn remove_manifest_from_snapshot(
             Ok(get_first_event(storage.clone(), stream_name, Vec::new()).await?)
         }
         Mode::Query => Ok(get_first_event(storage, stream_name, dates).await?),
+        Mode::Index => unimplemented!(),
     }
 }
 
@@ -350,6 +351,7 @@ pub async fn get_first_event(
 ) -> Result<Option<String>, ObjectStorageError> {
     let mut first_event_at: String = String::default();
     match PARSEABLE.options.mode {
+        Mode::Index => unimplemented!(),
         Mode::All | Mode::Ingest => {
             // get current snapshot
             let stream_first_event = PARSEABLE.get_stream(stream_name)?.get_first_event();

src/handlers/http/middleware.rs

@@ -357,6 +357,25 @@ where
                     Ok(res)
                 })
             }
+
+            Mode::Index => {
+                let accessable_endpoints = ["create", "delete"];
+                let cond = path.split('/').any(|x| accessable_endpoints.contains(&x));
+                if !cond {
+                    Box::pin(async {
+                        Err(actix_web::error::ErrorUnauthorized(
+                            "Only Index API can be accessed in Index Mode",
+                        ))
+                    })
+                } else {
+                    let fut = self.service.call(req);
+
+                    Box::pin(async move {
+                        let res = fut.await?;
+                        Ok(res)
+                    })
+                }
+            }
         }
     }
 }

src/main.rs

@@ -1,3 +1,5 @@
+use std::process::exit;
+
 /*
  * Parseable Server (C) 2022 - 2024 Parseable, Inc.
  *
@@ -37,6 +39,10 @@ async fn main() -> anyhow::Result<()> {
     let server: Box<dyn ParseableServer> = match &PARSEABLE.options.mode {
         Mode::Query => Box::new(QueryServer),
         Mode::Ingest => Box::new(IngestServer),
+        Mode::Index => {
+            println!("Indexing is an enterprise feature. Check out https://www.parseable.com/pricing to know more!");
+            exit(0)
+        }
         Mode::All => Box::new(Server),
     };
 

src/option.rs

@@ -22,6 +22,7 @@ use serde::{Deserialize, Serialize};
 pub enum Mode {
     Query,
     Ingest,
+    Index,
     #[default]
     All,
 }
@@ -128,6 +129,7 @@ pub mod validation {
             "query" => Ok(Mode::Query),
             "ingest" => Ok(Mode::Ingest),
             "all" => Ok(Mode::All),
+            "index" => Ok(Mode::Index),
             _ => Err("Invalid MODE provided".to_string()),
         }
     }

src/parseable/mod.rs

@@ -243,6 +243,7 @@ impl Parseable {
         match self.options.mode {
             Mode::Query => "Distributed (Query)",
             Mode::Ingest => "Distributed (Ingest)",
+            Mode::Index => "Distributed (Index)",
             Mode::All => "Standalone",
         }
     }

src/storage/azure_blob.rs

@@ -428,10 +428,20 @@ impl ObjectStorage for BlobStore {
         &self,
         _path: &RelativePath,
     ) -> Result<BufReader, ObjectStorageError> {
-        unimplemented!()
+        Err(ObjectStorageError::UnhandledError(Box::new(
+            std::io::Error::new(
+                std::io::ErrorKind::Unsupported,
+                "Buffered reader not implemented for Blob Storage yet",
+            ),
+        )))
     }
     async fn head(&self, _path: &RelativePath) -> Result<ObjectMeta, ObjectStorageError> {
-        unimplemented!()
+        Err(ObjectStorageError::UnhandledError(Box::new(
+            std::io::Error::new(
+                std::io::ErrorKind::Unsupported,
+                "Head operation not implemented for Blob Storage yet",
+            ),
+        )))
     }
 
     async fn get_object(&self, path: &RelativePath) -> Result<Bytes, ObjectStorageError> {

src/storage/localfs.rs

@@ -108,10 +108,20 @@ impl ObjectStorage for LocalFS {
         &self,
         _path: &RelativePath,
     ) -> Result<BufReader, ObjectStorageError> {
-        unimplemented!()
+        Err(ObjectStorageError::UnhandledError(Box::new(
+            std::io::Error::new(
+                std::io::ErrorKind::Unsupported,
+                "Buffered reader not implemented for LocalFS yet",
+            ),
+        )))
     }
     async fn head(&self, _path: &RelativePath) -> Result<ObjectMeta, ObjectStorageError> {
-        unimplemented!()
+        Err(ObjectStorageError::UnhandledError(Box::new(
+            std::io::Error::new(
+                std::io::ErrorKind::Unsupported,
+                "Head operation not implemented for LocalFS yet",
+            ),
+        )))
     }
     async fn get_object(&self, path: &RelativePath) -> Result<Bytes, ObjectStorageError> {
         let time = Instant::now();

src/storage/object_storage.rs

@@ -820,7 +820,7 @@ pub fn schema_path(stream_name: &str) -> RelativePathBuf {
 
             RelativePathBuf::from_iter([stream_name, STREAM_ROOT_DIRECTORY, &file_name])
         }
-        Mode::All | Mode::Query => {
+        Mode::All | Mode::Query | Mode::Index => {
             RelativePathBuf::from_iter([stream_name, STREAM_ROOT_DIRECTORY, SCHEMA_FILE_NAME])
         }
     }
@@ -838,7 +838,7 @@ pub fn stream_json_path(stream_name: &str) -> RelativePathBuf {
             let file_name = format!(".ingestor.{id}{STREAM_METADATA_FILE_NAME}",);
             RelativePathBuf::from_iter([stream_name, STREAM_ROOT_DIRECTORY, &file_name])
         }
-        Mode::Query | Mode::All => RelativePathBuf::from_iter([
+        Mode::Query | Mode::All | Mode::Index => RelativePathBuf::from_iter([
             stream_name,
             STREAM_ROOT_DIRECTORY,
             STREAM_METADATA_FILE_NAME,

src/storage/s3.rs

@@ -560,14 +560,14 @@ impl ObjectStorage for S3 {
         path: &RelativePath,
     ) -> Result<BufReader, ObjectStorageError> {
         let path = &to_object_store_path(path);
-        let meta = self.client.head(path).await.unwrap();
+        let meta = self.client.head(path).await?;
 
         let store: Arc<dyn ObjectStore> = Arc::new(self.client.clone());
         let buf = object_store::buffered::BufReader::new(store, &meta);
         Ok(buf)
     }
     async fn head(&self, path: &RelativePath) -> Result<ObjectMeta, ObjectStorageError> {
-        Ok(self.client.head(&to_object_store_path(path)).await.unwrap())
+        Ok(self.client.head(&to_object_store_path(path)).await?)
     }
 
     async fn get_object(&self, path: &RelativePath) -> Result<Bytes, ObjectStorageError> {

src/storage/store_metadata.rs

@@ -159,6 +159,12 @@ pub async fn resolve_parseable_metadata(
                         metadata.server_mode = PARSEABLE.options.mode;
                         metadata.staging = PARSEABLE.options.staging_dir().to_path_buf();
                       },
+                    Mode::Index => {
+                        // if index server is started fetch the metadata from remote
+                        // update the server mode for local metadata
+                        metadata.server_mode = PARSEABLE.options.mode;
+                        metadata.staging = PARSEABLE.options.staging_dir().to_path_buf();
+                    }
                 }
                 Ok(metadata)
             }