Line data    Source code

       1              : use std::sync::Arc;
       2              : 
       3              : use anyhow::Context;
       4              : use aws_sdk_s3::Client;
       5              : use either::Either;
       6              : use tokio::sync::mpsc::UnboundedSender;
       7              : use tracing::info;
       8              : 
       9              : use crate::cloud_admin_api::{CloudAdminApiClient, ProjectData};
      10              : use crate::delete_batch_producer::FetchResult;
      11              : use crate::{RootTarget, TraversingDepth};
      12              : use utils::id::{TenantId, TenantTimelineId};
      13              : 
      14              : use super::ProcessedS3List;
      15              : 
      16            0 : pub async fn schedule_cleanup_deleted_tenants(
      17            0 :     s3_root_target: &RootTarget,
      18            0 :     s3_client: &Arc<Client>,
      19            0 :     admin_client: &Arc<CloudAdminApiClient>,
      20            0 :     projects_to_check_sender: UnboundedSender<ProjectData>,
      21            0 :     delete_sender: Arc<UnboundedSender<Either<TenantId, TenantTimelineId>>>,
      22            0 :     traversing_depth: TraversingDepth,
      23            0 : ) -> anyhow::Result<ProcessedS3List<TenantId, ProjectData>> {
      24            0 :     info!(
      25            0 :         "Starting to list the bucket from root {}",
      26            0 :         s3_root_target.bucket_name()
      27            0 :     );
      28            0 :     s3_client
      29            0 :         .head_bucket()
      30            0 :         .bucket(s3_root_target.bucket_name())
      31            0 :         .send()
      32            0 :         .await
      33            0 :         .with_context(|| format!("bucket {} was not found", s3_root_target.bucket_name()))?;
      34              : 
      35            0 :     let check_client = Arc::clone(admin_client);
      36            0 :     let tenant_stats = super::process_s3_target_recursively(
      37            0 :         s3_client,
      38            0 :         s3_root_target.tenants_root(),
      39            0 :         |s3_tenants| async move {
      40            0 :             let another_client = Arc::clone(&check_client);
      41            0 :             super::split_to_active_and_deleted_entries(s3_tenants, move |tenant_id| async move {
      42            0 :                 let project_data = another_client
      43            0 :                     .find_tenant_project(tenant_id)
      44            0 :                     .await
      45            0 :                     .with_context(|| format!("Tenant {tenant_id} project admin check"))?;
      46              : 
      47            0 :                 Ok(if let Some(console_project) = project_data {
      48            0 :                     if console_project.deleted {
      49            0 :                         delete_sender.send(Either::Left(tenant_id)).ok();
      50            0 :                         FetchResult::Deleted
      51              :                     } else {
      52            0 :                         if traversing_depth == TraversingDepth::Timeline {
      53            0 :                             projects_to_check_sender.send(console_project.clone()).ok();
      54            0 :                         }
      55            0 :                         FetchResult::Found(console_project)
      56              :                     }
      57              :                 } else {
      58            0 :                     delete_sender.send(Either::Left(tenant_id)).ok();
      59            0 :                     FetchResult::Absent
      60              :                 })
      61            0 :             })
      62            0 :             .await
      63            0 :         },
      64            0 :     )
      65            0 :     .await
      66            0 :     .context("tenant batch processing")?;
      67              : 
      68            0 :     info!(
      69            0 :         "Among {} tenants, found {} tenants to delete and {} active ones",
      70            0 :         tenant_stats.entries_total,
      71            0 :         tenant_stats.entries_to_delete.len(),
      72            0 :         tenant_stats.active_entries.len(),
      73            0 :     );
      74              : 
      75            0 :     let tenant_stats = match traversing_depth {
      76              :         TraversingDepth::Tenant => {
      77            0 :             info!("Finished listing the bucket for tenants only");
      78            0 :             tenant_stats
      79              :         }
      80              :         TraversingDepth::Timeline => {
      81            0 :             info!("Finished listing the bucket for tenants and sent {} active tenants to check for timelines", tenant_stats.active_entries.len());
      82            0 :             tenant_stats
      83              :         }
      84              :     };
      85              : 
      86            0 :     Ok(tenant_stats)
      87            0 : }