LCOV - code coverage report
Current view: top level - storage_scrubber/src - main.rs (source / functions) Coverage Total Hit
Test: 5fe7fa8d483b39476409aee736d6d5e32728bfac.info Lines: 0.0 % 317 0
Test Date: 2025-03-12 16:10:49 Functions: 0.0 % 89 0

            Line data    Source code
       1              : use anyhow::{Context, anyhow, bail};
       2              : use camino::Utf8PathBuf;
       3              : use clap::{Parser, Subcommand};
       4              : use pageserver_api::controller_api::{MetadataHealthUpdateRequest, MetadataHealthUpdateResponse};
       5              : use pageserver_api::shard::TenantShardId;
       6              : use reqwest::{Method, Url};
       7              : use storage_controller_client::control_api;
       8              : use storage_scrubber::garbage::{PurgeMode, find_garbage, purge_garbage};
       9              : use storage_scrubber::pageserver_physical_gc::{GcMode, pageserver_physical_gc};
      10              : use storage_scrubber::scan_pageserver_metadata::scan_pageserver_metadata;
      11              : use storage_scrubber::scan_safekeeper_metadata::{DatabaseOrList, scan_safekeeper_metadata};
      12              : use storage_scrubber::tenant_snapshot::SnapshotDownloader;
      13              : use storage_scrubber::{
      14              :     BucketConfig, ConsoleConfig, ControllerClientConfig, NodeKind, TraversingDepth,
      15              :     find_large_objects, init_logging,
      16              : };
      17              : use utils::id::TenantId;
      18              : use utils::{project_build_tag, project_git_version};
      19              : 
      20              : project_git_version!(GIT_VERSION);
      21              : project_build_tag!(BUILD_TAG);
      22              : 
      23              : #[derive(Parser)]
      24              : #[command(author, version, about, long_about = None)]
      25              : #[command(arg_required_else_help(true))]
      26              : struct Cli {
      27              :     #[command(subcommand)]
      28              :     command: Command,
      29              : 
      30            0 :     #[arg(short, long, default_value_t = false)]
      31            0 :     delete: bool,
      32              : 
      33              :     #[arg(long)]
      34              :     /// URL to storage controller.  e.g. http://127.0.0.1:1234 when using `neon_local`
      35              :     controller_api: Option<Url>,
      36              : 
      37              :     #[arg(long)]
      38              :     /// JWT token for authenticating with storage controller.  Requires scope 'scrubber' or 'admin'.
      39              :     controller_jwt: Option<String>,
      40              : 
      41              :     /// If set to true, the scrubber will exit with error code on fatal error.
      42            0 :     #[arg(long, default_value_t = false)]
      43            0 :     exit_code: bool,
      44              : }
      45              : 
      46              : #[derive(Subcommand, Debug)]
      47              : enum Command {
      48              :     FindGarbage {
      49              :         #[arg(short, long)]
      50            0 :         node_kind: NodeKind,
      51            0 :         #[arg(short, long, default_value_t=TraversingDepth::Tenant)]
      52            0 :         depth: TraversingDepth,
      53              :         #[arg(short, long, default_value=None)]
      54              :         tenant_id_prefix: Option<String>,
      55            0 :         #[arg(short, long, default_value_t = String::from("garbage.json"))]
      56            0 :         output_path: String,
      57              :     },
      58              :     PurgeGarbage {
      59              :         #[arg(short, long)]
      60            0 :         input_path: String,
      61            0 :         #[arg(short, long, default_value_t = PurgeMode::DeletedOnly)]
      62            0 :         mode: PurgeMode,
      63              :         #[arg(long = "min-age")]
      64            0 :         min_age: humantime::Duration,
      65              :     },
      66              :     #[command(verbatim_doc_comment)]
      67              :     ScanMetadata {
      68              :         #[arg(short, long)]
      69            0 :         node_kind: NodeKind,
      70            0 :         #[arg(short, long, default_value_t = false)]
      71            0 :         json: bool,
      72              :         #[arg(long = "tenant-id", num_args = 0..)]
      73            0 :         tenant_ids: Vec<TenantShardId>,
      74            0 :         #[arg(long = "post", default_value_t = false)]
      75            0 :         post_to_storcon: bool,
      76              :         #[arg(long, default_value = None)]
      77              :         /// For safekeeper node_kind only, points to db with debug dump
      78              :         dump_db_connstr: Option<String>,
      79              :         /// For safekeeper node_kind only, table in the db with debug dump
      80              :         #[arg(long, default_value = None)]
      81              :         dump_db_table: Option<String>,
      82              :         /// For safekeeper node_kind only, json list of timelines and their lsn info
      83              :         #[arg(long, default_value = None)]
      84              :         timeline_lsns: Option<String>,
      85            0 :         #[arg(long, default_value_t = false)]
      86            0 :         verbose: bool,
      87              :     },
      88              :     TenantSnapshot {
      89              :         #[arg(long = "tenant-id")]
      90            0 :         tenant_id: TenantId,
      91            0 :         #[arg(long = "concurrency", short = 'j', default_value_t = 8)]
      92            0 :         concurrency: usize,
      93              :         #[arg(short, long)]
      94            0 :         output_path: Utf8PathBuf,
      95              :     },
      96              :     PageserverPhysicalGc {
      97              :         #[arg(long = "tenant-id", num_args = 0..)]
      98            0 :         tenant_ids: Vec<TenantShardId>,
      99              :         #[arg(long = "min-age")]
     100            0 :         min_age: humantime::Duration,
     101            0 :         #[arg(short, long, default_value_t = GcMode::IndicesOnly)]
     102            0 :         mode: GcMode,
     103              :     },
     104              :     FindLargeObjects {
     105              :         #[arg(long = "min-size")]
     106            0 :         min_size: u64,
     107            0 :         #[arg(short, long, default_value_t = false)]
     108            0 :         ignore_deltas: bool,
     109            0 :         #[arg(long = "concurrency", short = 'j', default_value_t = 64)]
     110            0 :         concurrency: usize,
     111              :     },
     112              :     CronJob {
     113              :         // PageserverPhysicalGc
     114              :         #[arg(long = "min-age")]
     115            0 :         gc_min_age: humantime::Duration,
     116            0 :         #[arg(short, long, default_value_t = GcMode::IndicesOnly)]
     117            0 :         gc_mode: GcMode,
     118              :         // ScanMetadata
     119            0 :         #[arg(long = "post", default_value_t = false)]
     120            0 :         post_to_storcon: bool,
     121              :     },
     122              : }
     123              : 
     124              : #[tokio::main]
     125            0 : async fn main() -> anyhow::Result<()> {
     126            0 :     let cli = Cli::parse();
     127            0 : 
     128            0 :     let bucket_config = BucketConfig::from_env()?;
     129            0 : 
     130            0 :     let command_log_name = match &cli.command {
     131            0 :         Command::ScanMetadata { .. } => "scan",
     132            0 :         Command::FindGarbage { .. } => "find-garbage",
     133            0 :         Command::PurgeGarbage { .. } => "purge-garbage",
     134            0 :         Command::TenantSnapshot { .. } => "tenant-snapshot",
     135            0 :         Command::PageserverPhysicalGc { .. } => "pageserver-physical-gc",
     136            0 :         Command::FindLargeObjects { .. } => "find-large-objects",
     137            0 :         Command::CronJob { .. } => "cron-job",
     138            0 :     };
     139            0 :     let _guard = init_logging(&format!(
     140            0 :         "{}_{}_{}_{}.log",
     141            0 :         std::env::args().next().unwrap(),
     142            0 :         command_log_name,
     143            0 :         bucket_config.bucket_name().unwrap_or("nobucket"),
     144            0 :         chrono::Utc::now().format("%Y_%m_%d__%H_%M_%S")
     145            0 :     ));
     146            0 : 
     147            0 :     tracing::info!("version: {}, build_tag {}", GIT_VERSION, BUILD_TAG);
     148            0 : 
     149            0 :     let controller_client = cli.controller_api.map(|controller_api| {
     150            0 :         ControllerClientConfig {
     151            0 :             controller_api,
     152            0 :             // Default to no key: this is a convenience when working in a development environment
     153            0 :             controller_jwt: cli.controller_jwt.unwrap_or("".to_owned()),
     154            0 :         }
     155            0 :         .build_client()
     156            0 :     });
     157            0 : 
     158            0 :     match cli.command {
     159            0 :         Command::ScanMetadata {
     160            0 :             json,
     161            0 :             tenant_ids,
     162            0 :             node_kind,
     163            0 :             post_to_storcon,
     164            0 :             dump_db_connstr,
     165            0 :             dump_db_table,
     166            0 :             timeline_lsns,
     167            0 :             verbose,
     168            0 :         } => {
     169            0 :             if let NodeKind::Safekeeper = node_kind {
     170            0 :                 let db_or_list = match (timeline_lsns, dump_db_connstr) {
     171            0 :                     (Some(timeline_lsns), _) => {
     172            0 :                         let timeline_lsns = serde_json::from_str(&timeline_lsns)
     173            0 :                             .context("parsing timeline_lsns")?;
     174            0 :                         DatabaseOrList::List(timeline_lsns)
     175            0 :                     }
     176            0 :                     (None, Some(dump_db_connstr)) => {
     177            0 :                         let dump_db_table = dump_db_table
     178            0 :                             .ok_or_else(|| anyhow::anyhow!("dump_db_table not specified"))?;
     179            0 :                         let tenant_ids = tenant_ids.iter().map(|tshid| tshid.tenant_id).collect();
     180            0 :                         DatabaseOrList::Database {
     181            0 :                             tenant_ids,
     182            0 :                             connstr: dump_db_connstr,
     183            0 :                             table: dump_db_table,
     184            0 :                         }
     185            0 :                     }
     186            0 :                     (None, None) => anyhow::bail!(
     187            0 :                         "neither `timeline_lsns` specified, nor `dump_db_connstr` and `dump_db_table`"
     188            0 :                     ),
     189            0 :                 };
     190            0 :                 let summary = scan_safekeeper_metadata(bucket_config.clone(), db_or_list).await?;
     191            0 :                 if json {
     192            0 :                     println!("{}", serde_json::to_string(&summary).unwrap())
     193            0 :                 } else {
     194            0 :                     println!("{}", summary.summary_string());
     195            0 :                 }
     196            0 :                 if summary.is_fatal() {
     197            0 :                     bail!("Fatal scrub errors detected");
     198            0 :                 }
     199            0 :                 if summary.is_empty() {
     200            0 :                     // Strictly speaking an empty bucket is a valid bucket, but if someone ran the
     201            0 :                     // scrubber they were likely expecting to scan something, and if we see no timelines
     202            0 :                     // at all then it's likely due to some configuration issues like a bad prefix
     203            0 :                     bail!("No timelines found in {}", bucket_config.desc_str());
     204            0 :                 }
     205            0 :                 Ok(())
     206            0 :             } else {
     207            0 :                 scan_pageserver_metadata_cmd(
     208            0 :                     bucket_config,
     209            0 :                     controller_client.as_ref(),
     210            0 :                     tenant_ids,
     211            0 :                     json,
     212            0 :                     post_to_storcon,
     213            0 :                     verbose,
     214            0 :                     cli.exit_code,
     215            0 :                 )
     216            0 :                 .await
     217            0 :             }
     218            0 :         }
     219            0 :         Command::FindGarbage {
     220            0 :             node_kind,
     221            0 :             depth,
     222            0 :             tenant_id_prefix,
     223            0 :             output_path,
     224            0 :         } => {
     225            0 :             let console_config = ConsoleConfig::from_env()?;
     226            0 :             find_garbage(
     227            0 :                 bucket_config,
     228            0 :                 console_config,
     229            0 :                 depth,
     230            0 :                 node_kind,
     231            0 :                 tenant_id_prefix,
     232            0 :                 output_path,
     233            0 :             )
     234            0 :             .await
     235            0 :         }
     236            0 :         Command::PurgeGarbage {
     237            0 :             input_path,
     238            0 :             mode,
     239            0 :             min_age,
     240            0 :         } => purge_garbage(input_path, mode, min_age.into(), !cli.delete).await,
     241            0 :         Command::TenantSnapshot {
     242            0 :             tenant_id,
     243            0 :             output_path,
     244            0 :             concurrency,
     245            0 :         } => {
     246            0 :             let downloader =
     247            0 :                 SnapshotDownloader::new(bucket_config, tenant_id, output_path, concurrency).await?;
     248            0 :             downloader.download().await
     249            0 :         }
     250            0 :         Command::PageserverPhysicalGc {
     251            0 :             tenant_ids,
     252            0 :             min_age,
     253            0 :             mode,
     254            0 :         } => {
     255            0 :             pageserver_physical_gc_cmd(
     256            0 :                 &bucket_config,
     257            0 :                 controller_client.as_ref(),
     258            0 :                 tenant_ids,
     259            0 :                 min_age,
     260            0 :                 mode,
     261            0 :             )
     262            0 :             .await
     263            0 :         }
     264            0 :         Command::FindLargeObjects {
     265            0 :             min_size,
     266            0 :             ignore_deltas,
     267            0 :             concurrency,
     268            0 :         } => {
     269            0 :             let summary = find_large_objects::find_large_objects(
     270            0 :                 bucket_config,
     271            0 :                 min_size,
     272            0 :                 ignore_deltas,
     273            0 :                 concurrency,
     274            0 :             )
     275            0 :             .await?;
     276            0 :             println!("{}", serde_json::to_string(&summary).unwrap());
     277            0 :             Ok(())
     278            0 :         }
     279            0 :         Command::CronJob {
     280            0 :             gc_min_age,
     281            0 :             gc_mode,
     282            0 :             post_to_storcon,
     283            0 :         } => {
     284            0 :             run_cron_job(
     285            0 :                 bucket_config,
     286            0 :                 controller_client.as_ref(),
     287            0 :                 gc_min_age,
     288            0 :                 gc_mode,
     289            0 :                 post_to_storcon,
     290            0 :                 cli.exit_code,
     291            0 :             )
     292            0 :             .await
     293            0 :         }
     294            0 :     }
     295            0 : }
     296              : 
     297              : /// Runs the scrubber cron job.
     298              : /// 1. Do pageserver physical gc
     299              : /// 2. Scan pageserver metadata
     300            0 : pub async fn run_cron_job(
     301            0 :     bucket_config: BucketConfig,
     302            0 :     controller_client: Option<&control_api::Client>,
     303            0 :     gc_min_age: humantime::Duration,
     304            0 :     gc_mode: GcMode,
     305            0 :     post_to_storcon: bool,
     306            0 :     exit_code: bool,
     307            0 : ) -> anyhow::Result<()> {
     308            0 :     tracing::info!(%gc_min_age, %gc_mode, "Running pageserver-physical-gc");
     309            0 :     pageserver_physical_gc_cmd(
     310            0 :         &bucket_config,
     311            0 :         controller_client,
     312            0 :         Vec::new(),
     313            0 :         gc_min_age,
     314            0 :         gc_mode,
     315            0 :     )
     316            0 :     .await?;
     317            0 :     tracing::info!(%post_to_storcon, node_kind = %NodeKind::Pageserver, "Running scan-metadata");
     318            0 :     scan_pageserver_metadata_cmd(
     319            0 :         bucket_config,
     320            0 :         controller_client,
     321            0 :         Vec::new(),
     322            0 :         true,
     323            0 :         post_to_storcon,
     324            0 :         false, // default to non-verbose mode
     325            0 :         exit_code,
     326            0 :     )
     327            0 :     .await?;
     328              : 
     329            0 :     Ok(())
     330            0 : }
     331              : 
     332            0 : pub async fn pageserver_physical_gc_cmd(
     333            0 :     bucket_config: &BucketConfig,
     334            0 :     controller_client: Option<&control_api::Client>,
     335            0 :     tenant_shard_ids: Vec<TenantShardId>,
     336            0 :     min_age: humantime::Duration,
     337            0 :     mode: GcMode,
     338            0 : ) -> anyhow::Result<()> {
     339            0 :     match (controller_client, mode) {
     340            0 :         (Some(_), _) => {
     341            0 :             // Any mode may run when controller API is set
     342            0 :         }
     343              :         (None, GcMode::Full) => {
     344              :             // The part of physical GC where we erase ancestor layers cannot be done safely without
     345              :             // confirming the most recent complete shard split with the controller.  Refuse to run, rather
     346              :             // than doing it unsafely.
     347            0 :             return Err(anyhow!(
     348            0 :                 "Full physical GC requires `--controller-api` and `--controller-jwt` to run"
     349            0 :             ));
     350              :         }
     351            0 :         (None, GcMode::DryRun | GcMode::IndicesOnly) => {
     352            0 :             // These GcModes do not require the controller to run.
     353            0 :         }
     354              :     }
     355              : 
     356            0 :     let summary = pageserver_physical_gc(
     357            0 :         bucket_config,
     358            0 :         controller_client,
     359            0 :         tenant_shard_ids,
     360            0 :         min_age.into(),
     361            0 :         mode,
     362            0 :     )
     363            0 :     .await?;
     364            0 :     println!("{}", serde_json::to_string(&summary).unwrap());
     365            0 :     Ok(())
     366            0 : }
     367              : 
     368            0 : pub async fn scan_pageserver_metadata_cmd(
     369            0 :     bucket_config: BucketConfig,
     370            0 :     controller_client: Option<&control_api::Client>,
     371            0 :     tenant_shard_ids: Vec<TenantShardId>,
     372            0 :     json: bool,
     373            0 :     post_to_storcon: bool,
     374            0 :     verbose: bool,
     375            0 :     exit_code: bool,
     376            0 : ) -> anyhow::Result<()> {
     377            0 :     if controller_client.is_none() && post_to_storcon {
     378            0 :         return Err(anyhow!(
     379            0 :             "Posting pageserver scan health status to storage controller requires `--controller-api` and `--controller-jwt` to run"
     380            0 :         ));
     381            0 :     }
     382            0 :     match scan_pageserver_metadata(bucket_config.clone(), tenant_shard_ids, verbose).await {
     383            0 :         Err(e) => {
     384            0 :             tracing::error!("Failed: {e}");
     385            0 :             Err(e)
     386              :         }
     387            0 :         Ok(summary) => {
     388            0 :             if json {
     389            0 :                 println!("{}", serde_json::to_string(&summary).unwrap())
     390            0 :             } else {
     391            0 :                 println!("{}", summary.summary_string());
     392            0 :             }
     393              : 
     394            0 :             if post_to_storcon {
     395            0 :                 if let Some(client) = controller_client {
     396            0 :                     let body = summary.build_health_update_request();
     397            0 :                     client
     398            0 :                         .dispatch::<MetadataHealthUpdateRequest, MetadataHealthUpdateResponse>(
     399            0 :                             Method::POST,
     400            0 :                             "control/v1/metadata_health/update".to_string(),
     401            0 :                             Some(body),
     402            0 :                         )
     403            0 :                         .await?;
     404            0 :                 }
     405            0 :             }
     406              : 
     407            0 :             if summary.is_fatal() {
     408            0 :                 tracing::error!("Fatal scrub errors detected");
     409            0 :                 if exit_code {
     410            0 :                     std::process::exit(1);
     411            0 :                 }
     412            0 :             } else if summary.is_empty() {
     413              :                 // Strictly speaking an empty bucket is a valid bucket, but if someone ran the
     414              :                 // scrubber they were likely expecting to scan something, and if we see no timelines
     415              :                 // at all then it's likely due to some configuration issues like a bad prefix
     416            0 :                 tracing::error!("No timelines found in {}", bucket_config.desc_str());
     417            0 :                 if exit_code {
     418            0 :                     std::process::exit(1);
     419            0 :                 }
     420            0 :             }
     421              : 
     422            0 :             Ok(())
     423              :         }
     424              :     }
     425            0 : }
        

Generated by: LCOV version 2.1-beta