chore: remove unwanted error and warning logs

1. on fresh deployment - server startup - when alerts and targets load
2. on delete stream - at query node - delete stats
3. on delete stream - at query node - delete staging
4. get_objects in object store

Author: nikhilsinhaparseable

src/handlers/http/modal/ingest/ingestor_logstream.rs

@@ -26,6 +26,7 @@ use actix_web::{
 use bytes::Bytes;
 use tracing::warn;
 
+use crate::option::Mode;
 use crate::{
     catalog::remove_manifest_from_snapshot,
     handlers::http::logstream::error::StreamError,
@@ -77,7 +78,12 @@ pub async fn delete(
     let tenant_id = get_tenant_id_from_request(&req);
     // Delete from staging
     let stream_dir = PARSEABLE.get_stream(&stream_name, &tenant_id)?;
-    if let Err(err) = fs::remove_dir_all(&stream_dir.data_path) {
+
+    // delete staging only for ingest server or standalone server
+    // else skip
+    if (PARSEABLE.options.mode == Mode::Ingest || PARSEABLE.options.mode == Mode::All)
+        && let Err(err) = fs::remove_dir_all(&stream_dir.data_path)
+    {
         warn!(
             "failed to delete local data for stream {} with error {err}. Clean {} manually",
             stream_name,

src/handlers/http/modal/mod.rs

@@ -130,7 +130,7 @@ pub trait ParseableServer {
             .backlog(PARSEABLE.options.connection_backlog)
             .max_connections(PARSEABLE.options.max_connections)
             .shutdown_timeout(60);
-        tracing::warn!(
+        tracing::info!(
             "Starting server with-\nNum workers: {}\nKeep Alive: {}\nRequest timeout: {}\nConnection backlog: {}\nMax connections: {}",
             PARSEABLE.options.num_workers,
             PARSEABLE.options.keep_alive,

src/metastore/metastores/object_store_metastore.rs

@@ -65,6 +65,14 @@ pub struct ObjectStoreMetastore {
     pub storage: Arc<dyn ObjectStorage>,
 }
 
+fn is_missing_optional_dir(err: &ObjectStorageError) -> bool {
+    match err {
+        ObjectStorageError::NoSuchKey(_) => true,
+        ObjectStorageError::IoError(err) => err.kind() == std::io::ErrorKind::NotFound,
+        _ => false,
+    }
+}
+
 #[async_trait]
 impl Metastore for ObjectStoreMetastore {
     /// Since Parseable already starts with a connection to an object store, no need to implement this
@@ -263,7 +271,7 @@ impl Metastore for ObjectStoreMetastore {
         let mut all_alerts = HashMap::new();
         for mut tenant in base_paths {
             let alerts_path = RelativePathBuf::from_iter([&tenant, ALERTS_ROOT_DIRECTORY]);
-            let alerts = self
+            let alerts = match self
                 .storage
                 .get_objects(
                     Some(&alerts_path),
@@ -272,7 +280,12 @@ impl Metastore for ObjectStoreMetastore {
                     }),
                     &Some(tenant.clone()),
                 )
-                .await?;
+                .await
+            {
+                Ok(alerts) => alerts,
+                Err(err) if is_missing_optional_dir(&err) => Vec::new(),
+                Err(err) => return Err(MetastoreError::ObjectStorageError(err)),
+            };
             if tenant.is_empty() {
                 tenant.clone_from(&DEFAULT_TENANT.to_string());
             }
@@ -1122,14 +1135,20 @@ impl Metastore for ObjectStoreMetastore {
                 SETTINGS_ROOT_DIRECTORY,
                 TARGETS_ROOT_DIRECTORY,
             ]);
-            let targets = self
+            let target_bytes = match self
                 .storage
                 .get_objects(
                     Some(&targets_path),
                     Box::new(|file_name| file_name.ends_with(".json")),
                     &Some(tenant.clone()),
                 )
-                .await?
+                .await
+            {
+                Ok(targets) => targets,
+                Err(err) if is_missing_optional_dir(&err) => Vec::new(),
+                Err(err) => return Err(MetastoreError::ObjectStorageError(err)),
+            };
+            let targets = target_bytes
                 .iter()
                 .filter_map(|bytes| {
                     serde_json::from_slice(bytes)

src/stats.rs

@@ -31,7 +31,8 @@ use crate::metrics::{
     EVENTS_STORAGE_SIZE_DATE, LIFETIME_EVENTS_INGESTED, LIFETIME_EVENTS_INGESTED_SIZE,
     LIFETIME_EVENTS_STORAGE_SIZE, STORAGE_SIZE,
 };
-use crate::parseable::DEFAULT_TENANT;
+use crate::option::Mode;
+use crate::parseable::{DEFAULT_TENANT, PARSEABLE};
 use crate::storage::{ObjectStorage, ObjectStorageError, ObjectStoreFormat};
 
 /// Helper struct type created by copying stats values from metadata
@@ -187,6 +188,11 @@ pub fn delete_stats(
     format: &'static str,
     tenant_id: &Option<String>,
 ) -> prometheus::Result<()> {
+    // delete stats only for ingest server or standalone server
+    // else skip
+    if PARSEABLE.options.mode != Mode::Ingest && PARSEABLE.options.mode != Mode::All {
+        return Ok(());
+    }
     let event_labels = event_labels(stream_name, format, tenant_id);
     let storage_size_labels = storage_size_labels(stream_name, tenant_id);
 

src/storage/azure_blob.rs

@@ -320,8 +320,7 @@ impl BlobStore {
     #[tracing::instrument(
         name = "azure.get_object",
         skip(self),
-        fields(path = %path, tenant = ?tenant_id, bytes = tracing::field::Empty),
-        err
+        fields(path = %path, tenant = ?tenant_id, bytes = tracing::field::Empty)
     )]
     async fn _get_object(
         &self,
@@ -446,8 +445,7 @@ impl BlobStore {
     #[tracing::instrument(
         name = "azure.list_dates",
         skip(self),
-        fields(stream = %stream, tenant = ?tenant_id, dates = tracing::field::Empty),
-        err
+        fields(stream = %stream, tenant = ?tenant_id, dates = tracing::field::Empty)
     )]
     async fn _list_dates(
         &self,
@@ -651,8 +649,7 @@ impl ObjectStorage for BlobStore {
     #[tracing::instrument(
         name = "azure.head",
         skip(self),
-        fields(path = %path, tenant = ?tenant_id),
-        err
+        fields(path = %path, tenant = ?tenant_id)
     )]
     async fn head(
         &self,
@@ -827,8 +824,7 @@ impl ObjectStorage for BlobStore {
     #[tracing::instrument(
         name = "azure.check",
         skip(self),
-        fields(tenant = ?tenant_id, ok = tracing::field::Empty),
-        err
+        fields(tenant = ?tenant_id, ok = tracing::field::Empty)
     )]
     async fn check(&self, tenant_id: &Option<String>) -> Result<(), ObjectStorageError> {
         let result = self

src/storage/field_stats.rs

@@ -453,7 +453,7 @@ impl FieldCountState {
 
             if !self.approximate {
                 self.approximate = true;
-                warn!(
+                debug!(
                     "Field stats cardinality cap reached for stream {} field {}. Tracking bounded top-value candidates with max_tracked_values={}",
                     self.stream_name, self.field_name, self.max_tracked_values
                 );

src/storage/gcs.rs

@@ -287,8 +287,7 @@ impl Gcs {
     #[tracing::instrument(
         name = "gcs.get_object",
         skip(self),
-        fields(path = %path, tenant = ?tenant_id, bytes = tracing::field::Empty),
-        err
+        fields(path = %path, tenant = ?tenant_id, bytes = tracing::field::Empty)
     )]
     async fn _get_object(
         &self,
@@ -412,8 +411,7 @@ impl Gcs {
     #[tracing::instrument(
         name = "gcs.list_dates",
         skip(self),
-        fields(stream = %stream, tenant = ?tenant_id, dates = tracing::field::Empty),
-        err
+        fields(stream = %stream, tenant = ?tenant_id, dates = tracing::field::Empty)
     )]
     async fn _list_dates(
         &self,
@@ -633,8 +631,7 @@ impl ObjectStorage for Gcs {
     #[tracing::instrument(
         name = "gcs.head",
         skip(self),
-        fields(path = %path, tenant = ?tenant_id),
-        err
+        fields(path = %path, tenant = ?tenant_id)
     )]
     async fn head(
         &self,
@@ -809,8 +806,7 @@ impl ObjectStorage for Gcs {
     #[tracing::instrument(
         name = "gcs.check",
         skip(self),
-        fields(tenant = ?tenant_id, ok = tracing::field::Empty),
-        err
+        fields(tenant = ?tenant_id, ok = tracing::field::Empty)
     )]
     async fn check(&self, tenant_id: &Option<String>) -> Result<(), ObjectStorageError> {
         let result = self

src/storage/s3.rs

@@ -477,8 +477,7 @@ impl S3 {
     #[tracing::instrument(
         name = "s3.get_object",
         skip(self),
-        fields(path = %path, tenant = ?tenant_id, bytes = tracing::field::Empty),
-        err
+        fields(path = %path, tenant = ?tenant_id, bytes = tracing::field::Empty)
     )]
     async fn _get_object(
         &self,
@@ -615,8 +614,7 @@ impl S3 {
     #[tracing::instrument(
         name = "s3.list_dates",
         skip(self),
-        fields(stream = %stream, tenant = ?tenant_id, dates = tracing::field::Empty),
-        err
+        fields(stream = %stream, tenant = ?tenant_id, dates = tracing::field::Empty)
     )]
     async fn _list_dates(
         &self,
@@ -848,8 +846,7 @@ impl ObjectStorage for S3 {
     #[tracing::instrument(
         name = "s3.head",
         skip(self),
-        fields(path = %path, tenant = ?tenant_id),
-        err
+        fields(path = %path, tenant = ?tenant_id)
     )]
     async fn head(
         &self,
@@ -1046,8 +1043,7 @@ impl ObjectStorage for S3 {
     #[tracing::instrument(
         name = "s3.check",
         skip(self),
-        fields(tenant = ?tenant_id, ok = tracing::field::Empty),
-        err
+        fields(tenant = ?tenant_id, ok = tracing::field::Empty)
     )]
     async fn check(&self, tenant_id: &Option<String>) -> Result<(), ObjectStorageError> {
         let tenant_str = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);