Merge branch 'main' into add-stream-creation-time

Author: gurjotkaur20

README.md

@@ -11,8 +11,8 @@
 <div align="center">
 
 [![Docker Pulls](https://img.shields.io/docker/pulls/parseable/parseable?logo=docker&label=Docker%20Pulls)](https://hub.docker.com/r/parseable/parseable)
-[![Slack](https://img.shields.io/badge/slack-brightgreen.svg?logo=slack&label=Community&style=flat&color=%2373DC8C&)](https://launchpass.com/parseable)
-[![Docs](https://img.shields.io/badge/stable%20docs-parseable.io%2Fdocs-brightgreen?style=flat&color=%2373DC8C&label=Docs)](https://www.parseable.io/docs)
+[![Slack](https://img.shields.io/badge/slack-brightgreen.svg?logo=slack&label=Community&style=flat&color=%2373DC8C&)](https://logg.ing/community)
+[![Docs](https://img.shields.io/badge/stable%20docs-parseable.io%2Fdocs-brightgreen?style=flat&color=%2373DC8C&label=Docs)](https://logg.ing/docs)
 [![Build](https://img.shields.io/github/checks-status/parseablehq/parseable/main?style=flat&color=%2373DC8C&label=Checks)](https://github.com/parseablehq/parseable/actions)
 
 [Key Concepts](https://www.parseable.io/docs/concepts) | [Features](https://github.com/parseablehq/parseable#rocket-highlights) | [Documentation](https://www.parseable.io/docs) | [Demo](https://demo.parseable.com/login?q=eyJ1c2VybmFtZSI6ImFkbWluIiwicGFzc3dvcmQiOiJhZG1pbiJ9) | [Integrations](https://www.parseable.io/docs/category/integrations) | [FAQ](https://www.parseable.io/docs/faq)

server/src/about.rs

@@ -104,7 +104,7 @@ pub fn print_about(
     eprintln!(
         "
         Commit:             \"{commit_hash}\"
-        Docs:               \"https://www.parseable.io/docs\""
+        Docs:               \"https://logg.ing/docs\""
     );
 }
 

server/src/main.rs

@@ -55,9 +55,9 @@ use crate::localcache::LocalCacheManager;
 async fn main() -> anyhow::Result<()> {
     env_logger::init();
     let storage = CONFIG.storage().get_object_store();
+    CONFIG.validate().await?;
     migration::run_metadata_migration(&CONFIG).await?;
     let metadata = storage::resolve_parseable_metadata().await?;
-    CONFIG.validate_staging()?;
     banner::print(&CONFIG, &metadata).await;
     rbac::map::init(&metadata);
     metadata.set_global();

server/src/migration.rs

@@ -32,6 +32,8 @@ use crate::{
     storage::{ObjectStorage, ObjectStorageError},
 };
 
+/// Migrate the metdata from v1 or v2 to v3
+/// This is a one time migration
 pub async fn run_metadata_migration(config: &Config) -> anyhow::Result<()> {
     let object_store = config.storage().get_object_store();
     let storage_metadata = get_storage_metadata(&*object_store).await?;

server/src/option.rs

@@ -27,11 +27,11 @@ use std::sync::Arc;
 use url::Url;
 
 use crate::oidc::{self, OpenidConfig};
-use crate::storage::{FSConfig, ObjectStorageProvider, S3Config};
-use crate::utils::validate_path_is_writeable;
+use crate::storage::{FSConfig, ObjectStorageError, ObjectStorageProvider, S3Config};
 
 pub const MIN_CACHE_SIZE_BYTES: u64 = 1000u64.pow(3); // 1 GiB
-
+pub const JOIN_COMMUNITY: &str =
+    "Join us on Parseable Slack community for questions : https://logg.ing/community";
 pub static CONFIG: Lazy<Arc<Config>> = Lazy::new(|| Arc::new(Config::new()));
 
 #[derive(Debug)]
@@ -99,9 +99,31 @@ impl Config {
         }
     }
 
-    pub fn validate_staging(&self) -> anyhow::Result<()> {
-        let staging_path = self.staging_dir();
-        validate_path_is_writeable(staging_path)
+    pub async fn validate(&self) -> Result<(), ObjectStorageError> {
+        let obj_store = self.storage.get_object_store();
+        let rel_path = relative_path::RelativePathBuf::from(".parseable.json");
+
+        let has_parseable_json = obj_store.get_object(&rel_path).await.is_ok();
+
+        // Lists all the directories in the root of the bucket/directory
+        // can be a stream (if it contains .stream.json file) or not
+        let has_dirs = match obj_store.list_dirs().await {
+            Ok(dirs) => !dirs.is_empty(),
+            Err(_) => false,
+        };
+
+        let has_streams = obj_store.list_streams().await.is_ok();
+
+        if has_streams || !has_dirs && !has_parseable_json {
+            return Ok(());
+        }
+
+        if self.mode_string() == "Local drive" {
+            return Err(ObjectStorageError::Custom(format!("Could not start the server because directory '{}' contains stale data, please use an empty directory, and restart the server.\n{}", self.storage.get_endpoint(), JOIN_COMMUNITY)));
+        }
+
+        // S3 bucket mode
+        Err(ObjectStorageError::Custom(format!("Could not start the server because bucket '{}' contains stale data, please use an empty bucket and restart the server.\n{}", self.storage.get_endpoint(), JOIN_COMMUNITY)))
     }
 
     pub fn storage(&self) -> Arc<dyn ObjectStorageProvider + Send + Sync> {
@@ -159,7 +181,7 @@ fn parseable_cli_command() -> Command {
         .next_line_help(false)
         .help_template(
             r#"
-{about} Join the community at https://launchpass.com/parseable.
+{about} Join the community at https://logg.ing/community.
 
 {all-args}
         "#,

server/src/storage/localfs.rs

@@ -32,7 +32,7 @@ use tokio::fs::{self, DirEntry};
 use tokio_stream::wrappers::ReadDirStream;
 
 use crate::metrics::storage::{localfs::REQUEST_RESPONSE_TIME, StorageMetrics};
-use crate::{option::validation, utils::validate_path_is_writeable};
+use crate::option::validation;
 
 use super::{object_storage, LogStream, ObjectStorage, ObjectStorageError, ObjectStorageProvider};
 
@@ -139,8 +139,8 @@ impl ObjectStorage for LocalFS {
     }
 
     async fn check(&self) -> Result<(), ObjectStorageError> {
-        fs::create_dir_all(&self.root).await?;
-        validate_path_is_writeable(&self.root)
+        fs::create_dir_all(&self.root)
+            .await
             .map_err(|e| ObjectStorageError::UnhandledError(e.into()))
     }
 
@@ -169,6 +169,23 @@ impl ObjectStorage for LocalFS {
         Ok(logstreams)
     }
 
+    async fn list_dirs(&self) -> Result<Vec<String>, ObjectStorageError> {
+        let dirs = ReadDirStream::new(fs::read_dir(&self.root).await?)
+            .try_collect::<Vec<DirEntry>>()
+            .await?
+            .into_iter()
+            .map(dir_name);
+
+        let dirs = FuturesUnordered::from_iter(dirs)
+            .try_collect::<Vec<_>>()
+            .await?
+            .into_iter()
+            .flatten()
+            .collect::<Vec<_>>();
+
+        Ok(dirs)
+    }
+
     async fn list_dates(&self, stream_name: &str) -> Result<Vec<String>, ObjectStorageError> {
         let path = self.root.join(stream_name);
         let directories = ReadDirStream::new(fs::read_dir(&path).await?);

server/src/storage/object_storage.rs

@@ -75,6 +75,7 @@ pub trait ObjectStorage: Sync + 'static {
     async fn check(&self) -> Result<(), ObjectStorageError>;
     async fn delete_stream(&self, stream_name: &str) -> Result<(), ObjectStorageError>;
     async fn list_streams(&self) -> Result<Vec<LogStream>, ObjectStorageError>;
+    async fn list_dirs(&self) -> Result<Vec<String>, ObjectStorageError>;
     async fn list_dates(&self, stream_name: &str) -> Result<Vec<String>, ObjectStorageError>;
     async fn upload_file(&self, key: &str, path: &Path) -> Result<(), ObjectStorageError>;
 

server/src/storage/s3.rs

@@ -470,6 +470,18 @@ impl ObjectStorage for S3 {
     fn store_url(&self) -> url::Url {
         url::Url::parse(&format!("s3://{}", self.bucket)).unwrap()
     }
+
+    async fn list_dirs(&self) -> Result<Vec<String>, ObjectStorageError> {
+        let pre = object_store::path::Path::from("/");
+        let resp = self.client.list_with_delimiter(Some(&pre)).await?;
+
+        Ok(resp
+            .common_prefixes
+            .iter()
+            .flat_map(|path| path.parts())
+            .map(|name| name.as_ref().to_string())
+            .collect::<Vec<_>>())
+    }
 }
 
 impl From<object_store::Error> for ObjectStorageError {

server/src/storage/store_metadata.rs

@@ -26,7 +26,7 @@ use once_cell::sync::OnceCell;
 use std::io;
 
 use crate::{
-    option::CONFIG,
+    option::{CONFIG, JOIN_COMMUNITY},
     rbac::{role::model::DefaultPrivilege, user::User},
     storage::ObjectStorageError,
     utils::uid,
@@ -92,8 +92,8 @@ impl StorageMetadata {
     }
 }
 
-// always returns remote metadata as it is source of truth
-// overwrites staging metadata while updating storage info
+/// always returns remote metadata as it is source of truth
+/// overwrites staging metadata while updating storage info
 pub async fn resolve_parseable_metadata() -> Result<StorageMetadata, ObjectStorageError> {
     let staging_metadata = get_staging_metadata()?;
     let storage = CONFIG.storage().get_object_store();
@@ -145,11 +145,7 @@ pub async fn resolve_parseable_metadata() -> Result<StorageMetadata, ObjectStora
     };
 
     let metadata = res.map_err(|err| {
-        let err = format!(
-            "{}. {}",
-            err,
-            "Join us on Parseable Slack to report this incident : https://launchpass.com/parseable"
-        );
+        let err = format!("{}. {}", err, JOIN_COMMUNITY);
         let err: Box<dyn std::error::Error + Send + Sync + 'static> = err.into();
         ObjectStorageError::UnhandledError(err)
     })?;
@@ -168,7 +164,7 @@ pub async fn resolve_parseable_metadata() -> Result<StorageMetadata, ObjectStora
 // variant contain remote metadata
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum EnvChange {
-    /// No change in env i.e both staging and remote have same id  
+    /// No change in env i.e both staging and remote have same id
     /// or deployment id of staging is not matching with that of remote
     None(StorageMetadata),
     /// Metadata not found in storage. Treated as possible misconfiguration on user side.

server/src/utils.rs

@@ -23,8 +23,6 @@ pub mod json;
 pub mod uid;
 pub mod update;
 
-use std::path::Path;
-
 use chrono::{DateTime, NaiveDate, Timelike, Utc};
 
 #[allow(dead_code)]
@@ -43,17 +41,6 @@ pub fn capitalize_ascii(s: &str) -> String {
     s[0..1].to_uppercase() + &s[1..]
 }
 
-pub fn validate_path_is_writeable(path: &Path) -> anyhow::Result<()> {
-    let Ok(md) = std::fs::metadata(path) else {
-        anyhow::bail!("Could not read metadata for staging dir")
-    };
-    let permissions = md.permissions();
-    if permissions.readonly() {
-        anyhow::bail!("Staging directory {} is not writable", path.display())
-    }
-    Ok(())
-}
-
 /// Convert minutes to a slot range
 /// e.g. given minute = 15 and OBJECT_STORE_DATA_GRANULARITY = 10 returns "10-19"
 pub fn minute_to_slot(minute: u32, data_granularity: u32) -> Option<String> {
@@ -263,7 +250,7 @@ mod tests {
         ]
     )]
     #[case::same_hour_with_00_to_59_minute_block(
-        "2022-06-11T16:00:00+00:00", "2022-06-11T16:59:59+00:00",   
+        "2022-06-11T16:00:00+00:00", "2022-06-11T16:59:59+00:00",
         &["date=2022-06-11/hour=16/"]
     )]
     #[case::same_date_different_hours_coherent_minute(
@@ -274,37 +261,37 @@ mod tests {
         ]
     )]
     #[case::same_date_different_hours_incoherent_minutes(
-        "2022-06-11T15:59:00+00:00", "2022-06-11T16:01:00+00:00", 
+        "2022-06-11T15:59:00+00:00", "2022-06-11T16:01:00+00:00",
         &[
             "date=2022-06-11/hour=15/minute=59/",
             "date=2022-06-11/hour=16/minute=00/"
         ]
     )]
     #[case::same_date_different_hours_whole_hours_between_incoherent_minutes(
-        "2022-06-11T15:59:00+00:00", "2022-06-11T17:01:00+00:00", 
+        "2022-06-11T15:59:00+00:00", "2022-06-11T17:01:00+00:00",
         &[
             "date=2022-06-11/hour=15/minute=59/",
             "date=2022-06-11/hour=16/",
             "date=2022-06-11/hour=17/minute=00/"
         ]
     )]
     #[case::different_date_coherent_hours_and_minutes(
-        "2022-06-11T00:00:00+00:00", "2022-06-13T00:00:00+00:00", 
+        "2022-06-11T00:00:00+00:00", "2022-06-13T00:00:00+00:00",
         &[
             "date=2022-06-11/",
             "date=2022-06-12/"
         ]
     )]
     #[case::different_date_incoherent_hours_coherent_minutes(
-        "2022-06-11T23:00:01+00:00", "2022-06-12T01:59:59+00:00", 
+        "2022-06-11T23:00:01+00:00", "2022-06-12T01:59:59+00:00",
         &[
             "date=2022-06-11/hour=23/",
             "date=2022-06-12/hour=00/",
             "date=2022-06-12/hour=01/"
         ]
     )]
     #[case::different_date_incoherent_hours_incoherent_minutes(
-        "2022-06-11T23:59:59+00:00", "2022-06-12T00:01:00+00:00", 
+        "2022-06-11T23:59:59+00:00", "2022-06-12T00:01:00+00:00",
         &[
             "date=2022-06-11/hour=23/minute=59/",
             "date=2022-06-12/hour=00/minute=00/"