TLA  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::{UnboundedReceiver, UnboundedSender};
       7                 : use tracing::{info, info_span, Instrument};
       8                 : 
       9                 : use crate::cloud_admin_api::{BranchData, CloudAdminApiClient, ProjectData};
      10                 : use crate::delete_batch_producer::{FetchResult, ProcessedS3List};
      11                 : use crate::RootTarget;
      12                 : use utils::id::{TenantId, TenantTimelineId};
      13                 : 
      14 UBC           0 : pub async fn schedule_cleanup_deleted_timelines(
      15               0 :     s3_root_target: &RootTarget,
      16               0 :     s3_client: &Arc<Client>,
      17               0 :     admin_client: &Arc<CloudAdminApiClient>,
      18               0 :     projects_to_check_receiver: &mut UnboundedReceiver<ProjectData>,
      19               0 :     delete_elements_sender: Arc<UnboundedSender<Either<TenantId, TenantTimelineId>>>,
      20               0 : ) -> anyhow::Result<ProcessedS3List<TenantTimelineId, BranchData>> {
      21               0 :     info!(
      22               0 :         "Starting to list the bucket from root {}",
      23               0 :         s3_root_target.bucket_name()
      24               0 :     );
      25               0 :     s3_client
      26               0 :         .head_bucket()
      27               0 :         .bucket(s3_root_target.bucket_name())
      28               0 :         .send()
      29               0 :         .await
      30               0 :         .with_context(|| format!("bucket {} was not found", s3_root_target.bucket_name()))?;
      31                 : 
      32               0 :     let mut timeline_stats = ProcessedS3List::default();
      33               0 :     while let Some(project_to_check) = projects_to_check_receiver.recv().await {
      34               0 :         let check_client = Arc::clone(admin_client);
      35               0 : 
      36               0 :         let check_s3_client = Arc::clone(s3_client);
      37               0 : 
      38               0 :         let check_delete_sender = Arc::clone(&delete_elements_sender);
      39               0 : 
      40               0 :         let check_root = s3_root_target.clone();
      41                 : 
      42               0 :         let new_stats = async move {
      43               0 :             let tenant_id_to_check = project_to_check.tenant;
      44               0 :             let check_target = check_root.timelines_root(&tenant_id_to_check);
      45               0 :             let stats = super::process_s3_target_recursively(
      46               0 :                 &check_s3_client,
      47               0 :                 &check_target,
      48               0 :                 |s3_timelines| async move {
      49               0 :                     let another_client = check_client.clone();
      50               0 :                     super::split_to_active_and_deleted_entries(
      51               0 :                         s3_timelines,
      52               0 :                         move |timeline_id| async move {
      53               0 :                             let console_branch = another_client
      54               0 :                                 .find_timeline_branch(timeline_id)
      55               0 :                                 .await
      56               0 :                                 .map_err(|e| {
      57               0 :                                     anyhow::anyhow!(
      58               0 :                                         "Timeline {timeline_id} branch admin check: {e}"
      59               0 :                                     )
      60               0 :                                 })?;
      61                 : 
      62               0 :                             let id = TenantTimelineId::new(tenant_id_to_check, timeline_id);
      63               0 :                             Ok(match console_branch {
      64               0 :                                 Some(console_branch) => {
      65               0 :                                     if console_branch.deleted {
      66               0 :                                         check_delete_sender.send(Either::Right(id)).ok();
      67               0 :                                         FetchResult::Deleted
      68                 :                                     } else {
      69               0 :                                         FetchResult::Found(console_branch)
      70                 :                                     }
      71                 :                                 }
      72                 :                                 None => {
      73               0 :                                     check_delete_sender.send(Either::Right(id)).ok();
      74               0 :                                     FetchResult::Absent
      75                 :                                 }
      76                 :                             })
      77               0 :                         },
      78               0 :                     )
      79               0 :                     .await
      80               0 :                 },
      81               0 :             )
      82               0 :             .await
      83               0 :             .with_context(|| format!("tenant {tenant_id_to_check} timeline batch processing"))?
      84               0 :             .change_ids(|timeline_id| TenantTimelineId::new(tenant_id_to_check, timeline_id));
      85               0 : 
      86               0 :             Ok::<_, anyhow::Error>(stats)
      87               0 :         }
      88               0 :         .instrument(info_span!("delete_timelines_sender", tenant = %project_to_check.tenant))
      89               0 :         .await?;
      90                 : 
      91               0 :         timeline_stats.merge(new_stats);
      92                 :     }
      93                 : 
      94               0 :     info!(
      95               0 :         "Among {} timelines, found {} timelines to delete and {} active ones",
      96               0 :         timeline_stats.entries_total,
      97               0 :         timeline_stats.entries_to_delete.len(),
      98               0 :         timeline_stats.active_entries.len(),
      99               0 :     );
     100                 : 
     101               0 :     Ok(timeline_stats)
     102               0 : }