fix: Multiple fixes (#1582)

- remove invalid tenant dir
- log sync errors
- add tenant to filter

Author: parmesant

src/handlers/http/modal/query/querier_logstream.rs

@@ -157,7 +157,9 @@ pub async fn put_stream(
             actix_web::http::header::HeaderValue::from_str(&userid).unwrap(),
         );
     }
-    sync_streams_with_ingestors(headers, body, &stream_name, &tenant_id).await?;
+    if let Err(e) = sync_streams_with_ingestors(headers, body, &stream_name, &tenant_id).await {
+        tracing::error!("{e}");
+    };
 
     if is_update {
         Ok(("Log stream updated", StatusCode::OK))

src/handlers/http/modal/query/querier_rbac.rs

@@ -92,7 +92,9 @@ pub async fn post_user(
     // let created_role = user_roles.clone();
     Users.put_user(user.clone());
 
-    sync_user_creation(&req, user, &None, &tenant_id, &caller_userid).await?;
+    if let Err(e) = sync_user_creation(&req, user, &None, &tenant_id, &caller_userid).await {
+        tracing::error!("{e}");
+    };
 
     Ok(password)
 }
@@ -161,7 +163,11 @@ pub async fn delete_user(
     }
     put_metadata(&metadata, &tenant_id).await?;
 
-    sync_user_deletion_with_ingestors(&req, &userid, &tenant_id, &caller_userid).await?;
+    if let Err(e) =
+        sync_user_deletion_with_ingestors(&req, &userid, &tenant_id, &caller_userid).await
+    {
+        tracing::error!("{e}");
+    };
 
     // update in mem table
     Users.delete_user(&userid, &tenant_id);
@@ -380,7 +386,9 @@ pub async fn post_gen_password(
     put_metadata(&metadata, &tenant_id).await?;
     Users.change_password_hash(&username, &new_hash, &tenant_id);
 
-    sync_password_reset_with_ingestors(req, &username, &caller_userid).await?;
+    if let Err(e) = sync_password_reset_with_ingestors(req, &username, &caller_userid).await {
+        tracing::error!("{e}");
+    };
 
     Ok(new_password)
 }

src/handlers/http/users/filters.rs

@@ -71,6 +71,7 @@ pub async fn post(
     filter.filter_id = Some(filter_id.clone());
     filter.user_id = Some(user_id.clone());
     filter.version = Some(CURRENT_FILTER_VERSION.to_string());
+    filter.tenant_id.clone_from(&tenant_id);
     PARSEABLE.metastore.put_filter(&filter, &tenant_id).await?;
     FILTERS.update(&filter, &tenant_id).await;
 
@@ -99,6 +100,7 @@ pub async fn update(
     filter.filter_id = Some(filter_id.clone());
     filter.user_id = Some(user_id.clone());
     filter.version = Some(CURRENT_FILTER_VERSION.to_string());
+    filter.tenant_id.clone_from(&tenant_id);
 
     PARSEABLE.metastore.put_filter(&filter, &tenant_id).await?;
     FILTERS.update(&filter, &tenant_id).await;

src/parseable/mod.rs

@@ -1148,8 +1148,10 @@ impl Parseable {
             .filter(|d| !d.starts_with("."))
             .collect();
 
+        let mut recognized_tenants = vec![];
+
         // validate the possible presence of tenant storage metadata
-        for tenant_id in dirs.iter() {
+        for tenant_id in dirs.into_iter() {
             if let Some(meta) = PARSEABLE
                 .metastore
                 .get_parseable_metadata(&Some(tenant_id.clone()))
@@ -1159,11 +1161,23 @@ impl Parseable {
                 let metadata: StorageMetadata = serde_json::from_slice(&meta)?;
 
                 TENANT_METADATA.insert_tenant(tenant_id.clone(), metadata.clone());
+                recognized_tenants.push(tenant_id);
             } else if !is_multi_tenant {
             } else {
-                return Err(anyhow::Error::msg(format!(
-                    "Found invalid tenant directory with multi-tenant mode- {tenant_id}.\nExiting."
-                )));
+                // found an invalid multi-tenant directory, delete
+                tracing::warn!(
+                    "Found invalid tenant directory with multi-tenant mode- {tenant_id}.\nDeleting."
+                );
+
+                // delete from storage
+                if let Err(e) = obj_store
+                    .delete_prefix(&RelativePathBuf::from_iter([&tenant_id]), &None)
+                    .await
+                {
+                    tracing::error!(
+                        "Unable to delete invalid directory- {tenant_id} with error-\n{e}"
+                    );
+                };
             }
         }
 
@@ -1172,7 +1186,7 @@ impl Parseable {
                 .tenants
                 .write()
                 .expect("Unable to write to in-memory tenant map");
-            t.extend(dirs);
+            t.extend(recognized_tenants);
             Ok(())
         } else {
             Ok(())