LCOV - code coverage report
Current view: top level - pageserver/src/bin - pageserver.rs (source / functions) Coverage Total Hit
Test: c639aa5f7ab62b43d647b10f40d15a15686ce8a9.info Lines: 95.0 % 577 548
Test Date: 2024-02-12 20:26:03 Functions: 64.1 % 64 41

            Line data    Source code
       1              : //! Main entry point for the Page Server executable.
       2              : 
       3              : use std::env::{var, VarError};
       4              : use std::sync::Arc;
       5              : use std::time::Duration;
       6              : use std::{env, ops::ControlFlow, str::FromStr};
       7              : 
       8              : use anyhow::{anyhow, Context};
       9              : use camino::Utf8Path;
      10              : use clap::{Arg, ArgAction, Command};
      11              : 
      12              : use metrics::launch_timestamp::{set_launch_timestamp_metric, LaunchTimestamp};
      13              : use pageserver::control_plane_client::ControlPlaneClient;
      14              : use pageserver::disk_usage_eviction_task::{self, launch_disk_usage_global_eviction_task};
      15              : use pageserver::metrics::{STARTUP_DURATION, STARTUP_IS_LOADING};
      16              : use pageserver::task_mgr::WALRECEIVER_RUNTIME;
      17              : use pageserver::tenant::{secondary, TenantSharedResources};
      18              : use remote_storage::GenericRemoteStorage;
      19              : use tokio::time::Instant;
      20              : use tracing::*;
      21              : 
      22              : use metrics::set_build_info_metric;
      23              : use pageserver::{
      24              :     config::{defaults::*, PageServerConf},
      25              :     context::{DownloadBehavior, RequestContext},
      26              :     deletion_queue::DeletionQueue,
      27              :     http, page_cache, page_service, task_mgr,
      28              :     task_mgr::TaskKind,
      29              :     task_mgr::{BACKGROUND_RUNTIME, COMPUTE_REQUEST_RUNTIME, MGMT_REQUEST_RUNTIME},
      30              :     tenant::mgr,
      31              :     virtual_file,
      32              : };
      33              : use postgres_backend::AuthType;
      34              : use utils::failpoint_support;
      35              : use utils::logging::TracingErrorLayerEnablement;
      36              : use utils::{
      37              :     auth::{JwtAuth, SwappableJwtAuth},
      38              :     logging, project_build_tag, project_git_version,
      39              :     sentry_init::init_sentry,
      40              :     tcp_listener,
      41              : };
      42              : 
      43              : project_git_version!(GIT_VERSION);
      44              : project_build_tag!(BUILD_TAG);
      45              : 
      46              : const PID_FILE_NAME: &str = "pageserver.pid";
      47              : 
      48              : const FEATURES: &[&str] = &[
      49              :     #[cfg(feature = "testing")]
      50              :     "testing",
      51              : ];
      52              : 
      53         2052 : fn version() -> String {
      54         2052 :     format!(
      55         2052 :         "{GIT_VERSION} failpoints: {}, features: {:?}",
      56         2052 :         fail::has_failpoints(),
      57         2052 :         FEATURES,
      58         2052 :     )
      59         2052 : }
      60              : 
      61         1426 : fn main() -> anyhow::Result<()> {
      62         1426 :     let launch_ts = Box::leak(Box::new(LaunchTimestamp::generate()));
      63         1426 : 
      64         1426 :     let arg_matches = cli().get_matches();
      65         1426 : 
      66         1426 :     if arg_matches.get_flag("enabled-features") {
      67            0 :         println!("{{\"features\": {FEATURES:?} }}");
      68            0 :         return Ok(());
      69         1426 :     }
      70         1426 : 
      71         1426 :     let workdir = arg_matches
      72         1426 :         .get_one::<String>("workdir")
      73         1426 :         .map(Utf8Path::new)
      74         1426 :         .unwrap_or_else(|| Utf8Path::new(".neon"));
      75         1426 :     let workdir = workdir
      76         1426 :         .canonicalize_utf8()
      77         1426 :         .with_context(|| format!("Error opening workdir '{workdir}'"))?;
      78              : 
      79         1426 :     let cfg_file_path = workdir.join("pageserver.toml");
      80         1426 : 
      81         1426 :     // Set CWD to workdir for non-daemon modes
      82         1426 :     env::set_current_dir(&workdir)
      83         1426 :         .with_context(|| format!("Failed to set application's current dir to '{workdir}'"))?;
      84              : 
      85         1426 :     let conf = match initialize_config(&cfg_file_path, arg_matches, &workdir)? {
      86          624 :         ControlFlow::Continue(conf) => conf,
      87              :         ControlFlow::Break(()) => {
      88          400 :             info!("Pageserver config init successful");
      89          400 :             return Ok(());
      90              :         }
      91              :     };
      92              : 
      93              :     // Initialize logging.
      94              :     //
      95              :     // It must be initialized before the custom panic hook is installed below.
      96              :     //
      97              :     // Regarding tracing_error enablement: at this time, we only use the
      98              :     // tracing_error crate to debug_assert that log spans contain tenant and timeline ids.
      99              :     // See `debug_assert_current_span_has_tenant_and_timeline_id` in the timeline module
     100          624 :     let tracing_error_layer_enablement = if cfg!(debug_assertions) {
     101          624 :         TracingErrorLayerEnablement::EnableWithRustLogFilter
     102              :     } else {
     103            0 :         TracingErrorLayerEnablement::Disabled
     104              :     };
     105          624 :     logging::init(
     106          624 :         conf.log_format,
     107          624 :         tracing_error_layer_enablement,
     108          624 :         logging::Output::Stdout,
     109          624 :     )?;
     110              : 
     111              :     // mind the order required here: 1. logging, 2. panic_hook, 3. sentry.
     112              :     // disarming this hook on pageserver, because we never tear down tracing.
     113          624 :     logging::replace_panic_hook_with_tracing_panic_hook().forget();
     114          624 : 
     115          624 :     // initialize sentry if SENTRY_DSN is provided
     116          624 :     let _sentry_guard = init_sentry(
     117          624 :         Some(GIT_VERSION.into()),
     118          624 :         &[("node_id", &conf.id.to_string())],
     119          624 :     );
     120          624 : 
     121          624 :     let tenants_path = conf.tenants_path();
     122          624 :     if !tenants_path.exists() {
     123          398 :         utils::crashsafe::create_dir_all(conf.tenants_path())
     124          398 :             .with_context(|| format!("Failed to create tenants root dir at '{tenants_path}'"))?;
     125          226 :     }
     126              : 
     127              :     // Initialize up failpoints support
     128          624 :     let scenario = failpoint_support::init();
     129          624 : 
     130          624 :     // Basic initialization of things that don't change after startup
     131          624 :     virtual_file::init(conf.max_file_descriptors, conf.virtual_file_io_engine);
     132          624 :     page_cache::init(conf.page_cache_size);
     133          624 : 
     134          624 :     start_pageserver(launch_ts, conf).context("Failed to start pageserver")?;
     135              : 
     136            0 :     scenario.teardown();
     137            0 :     Ok(())
     138          403 : }
     139              : 
     140         1027 : fn initialize_config(
     141         1027 :     cfg_file_path: &Utf8Path,
     142         1027 :     arg_matches: clap::ArgMatches,
     143         1027 :     workdir: &Utf8Path,
     144         1027 : ) -> anyhow::Result<ControlFlow<(), &'static PageServerConf>> {
     145         1027 :     let init = arg_matches.get_flag("init");
     146         1027 :     let update_config = init || arg_matches.get_flag("update-config");
     147              : 
     148         1027 :     let (mut toml, config_file_exists) = if cfg_file_path.is_file() {
     149          626 :         if init {
     150            1 :             anyhow::bail!(
     151            1 :                 "Config file '{cfg_file_path}' already exists, cannot init it, use --update-config to update it",
     152            1 :             );
     153          625 :         }
     154              :         // Supplement the CLI arguments with the config file
     155          625 :         let cfg_file_contents = std::fs::read_to_string(cfg_file_path)
     156          625 :             .with_context(|| format!("Failed to read pageserver config at '{cfg_file_path}'"))?;
     157              :         (
     158          625 :             cfg_file_contents
     159          625 :                 .parse::<toml_edit::Document>()
     160          625 :                 .with_context(|| {
     161            0 :                     format!("Failed to parse '{cfg_file_path}' as pageserver config")
     162          625 :                 })?,
     163              :             true,
     164              :         )
     165          401 :     } else if cfg_file_path.exists() {
     166            0 :         anyhow::bail!("Config file '{cfg_file_path}' exists but is not a regular file");
     167              :     } else {
     168              :         // We're initializing the tenant, so there's no config file yet
     169              :         (
     170          401 :             DEFAULT_CONFIG_FILE
     171          401 :                 .parse::<toml_edit::Document>()
     172          401 :                 .context("could not parse built-in config file")?,
     173              :             false,
     174              :         )
     175              :     };
     176              : 
     177         1026 :     if let Some(values) = arg_matches.get_many::<String>("config-override") {
     178        10446 :         for option_line in values {
     179         9421 :             let doc = toml_edit::Document::from_str(option_line).with_context(|| {
     180            0 :                 format!("Option '{option_line}' could not be parsed as a toml document")
     181         9421 :             })?;
     182              : 
     183         9447 :             for (key, item) in doc.iter() {
     184         9447 :                 if config_file_exists && update_config && key == "id" && toml.contains_key(key) {
     185            1 :                     anyhow::bail!("Pageserver config file exists at '{cfg_file_path}' and has node id already, it cannot be overridden");
     186         9446 :                 }
     187         9446 :                 toml.insert(key, item.clone());
     188              :             }
     189              :         }
     190            0 :     }
     191              : 
     192         1025 :     debug!("Resulting toml: {toml}");
     193         1025 :     let conf = PageServerConf::parse_and_validate(&toml, workdir)
     194         1025 :         .context("Failed to parse pageserver configuration")?;
     195              : 
     196         1024 :     if update_config {
     197          400 :         info!("Writing pageserver config to '{cfg_file_path}'");
     198              : 
     199          400 :         std::fs::write(cfg_file_path, toml.to_string())
     200          400 :             .with_context(|| format!("Failed to write pageserver config to '{cfg_file_path}'"))?;
     201          400 :         info!("Config successfully written to '{cfg_file_path}'")
     202          624 :     }
     203              : 
     204         1024 :     Ok(if init {
     205          400 :         ControlFlow::Break(())
     206              :     } else {
     207          624 :         ControlFlow::Continue(Box::leak(Box::new(conf)))
     208              :     })
     209         1027 : }
     210              : 
     211              : struct WaitForPhaseResult<F: std::future::Future + Unpin> {
     212              :     timeout_remaining: Duration,
     213              :     skipped: Option<F>,
     214              : }
     215              : 
     216              : /// During startup, we apply a timeout to our waits for readiness, to avoid
     217              : /// stalling the whole service if one Tenant experiences some problem.  Each
     218              : /// phase may consume some of the timeout: this function returns the updated
     219              : /// timeout for use in the next call.
     220         1241 : async fn wait_for_phase<F>(phase: &str, mut fut: F, timeout: Duration) -> WaitForPhaseResult<F>
     221         1241 : where
     222         1241 :     F: std::future::Future + Unpin,
     223         1241 : {
     224         1241 :     let initial_t = Instant::now();
     225         1241 :     let skipped = match tokio::time::timeout(timeout, &mut fut).await {
     226         1226 :         Ok(_) => None,
     227              :         Err(_) => {
     228            8 :             tracing::info!(
     229            8 :                 timeout_millis = timeout.as_millis(),
     230            8 :                 %phase,
     231            8 :                 "Startup phase timed out, proceeding anyway"
     232            8 :             );
     233            8 :             Some(fut)
     234              :         }
     235              :     };
     236              : 
     237         1234 :     WaitForPhaseResult {
     238         1234 :         timeout_remaining: timeout
     239         1234 :             .checked_sub(Instant::now().duration_since(initial_t))
     240         1234 :             .unwrap_or(Duration::ZERO),
     241         1234 :         skipped,
     242         1234 :     }
     243         1234 : }
     244              : 
     245         3082 : fn startup_checkpoint(started_at: Instant, phase: &str, human_phase: &str) {
     246         3082 :     let elapsed = started_at.elapsed();
     247         3082 :     let secs = elapsed.as_secs_f64();
     248         3082 :     STARTUP_DURATION.with_label_values(&[phase]).set(secs);
     249         3082 : 
     250         3082 :     info!(
     251         3082 :         elapsed_ms = elapsed.as_millis(),
     252         3082 :         "{human_phase} ({secs:.3}s since start)"
     253         3082 :     )
     254         3082 : }
     255              : 
     256          624 : fn start_pageserver(
     257          624 :     launch_ts: &'static LaunchTimestamp,
     258          624 :     conf: &'static PageServerConf,
     259          624 : ) -> anyhow::Result<()> {
     260          624 :     // Monotonic time for later calculating startup duration
     261          624 :     let started_startup_at = Instant::now();
     262          624 : 
     263          624 :     // Print version and launch timestamp to the log,
     264          624 :     // and expose them as prometheus metrics.
     265          624 :     // A changed version string indicates changed software.
     266          624 :     // A changed launch timestamp indicates a pageserver restart.
     267          624 :     info!(
     268          624 :         "version: {} launch_timestamp: {} build_tag: {}",
     269          624 :         version(),
     270          624 :         launch_ts.to_string(),
     271          624 :         BUILD_TAG,
     272          624 :     );
     273          624 :     set_build_info_metric(GIT_VERSION, BUILD_TAG);
     274          624 :     set_launch_timestamp_metric(launch_ts);
     275          624 :     #[cfg(target_os = "linux")]
     276          624 :     metrics::register_internal(Box::new(metrics::more_process_metrics::Collector::new())).unwrap();
     277          624 :     metrics::register_internal(Box::new(
     278          624 :         pageserver::metrics::tokio_epoll_uring::Collector::new(),
     279          624 :     ))
     280          624 :     .unwrap();
     281          624 :     pageserver::preinitialize_metrics();
     282          624 : 
     283          624 :     // If any failpoints were set from FAILPOINTS environment variable,
     284          624 :     // print them to the log for debugging purposes
     285          624 :     let failpoints = fail::list();
     286          624 :     if !failpoints.is_empty() {
     287            8 :         info!(
     288            8 :             "started with failpoints: {}",
     289            8 :             failpoints
     290            8 :                 .iter()
     291            9 :                 .map(|(name, actions)| format!("{name}={actions}"))
     292            8 :                 .collect::<Vec<String>>()
     293            8 :                 .join(";")
     294            8 :         )
     295          616 :     }
     296              : 
     297              :     // Create and lock PID file. This ensures that there cannot be more than one
     298              :     // pageserver process running at the same time.
     299          624 :     let lock_file_path = conf.workdir.join(PID_FILE_NAME);
     300          624 :     let lock_file =
     301          624 :         utils::pid_file::claim_for_current_process(&lock_file_path).context("claim pid file")?;
     302          624 :     info!("Claimed pid file at {lock_file_path:?}");
     303              : 
     304              :     // Ensure that the lock file is held even if the main thread of the process panics.
     305              :     // We need to release the lock file only when the process exits.
     306          624 :     std::mem::forget(lock_file);
     307          624 : 
     308          624 :     // Bind the HTTP and libpq ports early, so that if they are in use by some other
     309          624 :     // process, we error out early.
     310          624 :     let http_addr = &conf.listen_http_addr;
     311          624 :     info!("Starting pageserver http handler on {http_addr}");
     312          624 :     let http_listener = tcp_listener::bind(http_addr)?;
     313              : 
     314          624 :     let pg_addr = &conf.listen_pg_addr;
     315          624 :     info!("Starting pageserver pg protocol handler on {pg_addr}");
     316          624 :     let pageserver_listener = tcp_listener::bind(pg_addr)?;
     317              : 
     318              :     // Launch broker client
     319              :     // The storage_broker::connect call needs to happen inside a tokio runtime thread.
     320          624 :     let broker_client = WALRECEIVER_RUNTIME
     321          624 :         .block_on(async {
     322          624 :             // Note: we do not attempt connecting here (but validate endpoints sanity).
     323          624 :             storage_broker::connect(conf.broker_endpoint.clone(), conf.broker_keepalive_interval)
     324          624 :         })
     325          624 :         .with_context(|| {
     326            0 :             format!(
     327            0 :                 "create broker client for uri={:?} keepalive_interval={:?}",
     328            0 :                 &conf.broker_endpoint, conf.broker_keepalive_interval,
     329            0 :             )
     330          624 :         })?;
     331              : 
     332              :     // Initialize authentication for incoming connections
     333              :     let http_auth;
     334              :     let pg_auth;
     335          624 :     if conf.http_auth_type == AuthType::NeonJWT || conf.pg_auth_type == AuthType::NeonJWT {
     336              :         // unwrap is ok because check is performed when creating config, so path is set and exists
     337           11 :         let key_path = conf.auth_validation_public_key_path.as_ref().unwrap();
     338           11 :         info!("Loading public key(s) for verifying JWT tokens from {key_path:?}");
     339              : 
     340           11 :         let jwt_auth = JwtAuth::from_key_path(key_path)?;
     341           11 :         let auth: Arc<SwappableJwtAuth> = Arc::new(SwappableJwtAuth::new(jwt_auth));
     342              : 
     343           11 :         http_auth = match &conf.http_auth_type {
     344            0 :             AuthType::Trust => None,
     345           11 :             AuthType::NeonJWT => Some(auth.clone()),
     346              :         };
     347           11 :         pg_auth = match &conf.pg_auth_type {
     348            0 :             AuthType::Trust => None,
     349           11 :             AuthType::NeonJWT => Some(auth),
     350              :         };
     351          613 :     } else {
     352          613 :         http_auth = None;
     353          613 :         pg_auth = None;
     354          613 :     }
     355          624 :     info!("Using auth for http API: {:#?}", conf.http_auth_type);
     356          624 :     info!("Using auth for pg connections: {:#?}", conf.pg_auth_type);
     357              : 
     358          613 :     match var("NEON_AUTH_TOKEN") {
     359           11 :         Ok(v) => {
     360           11 :             info!("Loaded JWT token for authentication with Safekeeper");
     361           11 :             pageserver::config::SAFEKEEPER_AUTH_TOKEN
     362           11 :                 .set(Arc::new(v))
     363           11 :                 .map_err(|_| anyhow!("Could not initialize SAFEKEEPER_AUTH_TOKEN"))?;
     364              :         }
     365              :         Err(VarError::NotPresent) => {
     366          613 :             info!("No JWT token for authentication with Safekeeper detected");
     367              :         }
     368            0 :         Err(e) => {
     369            0 :             return Err(e).with_context(|| {
     370            0 :                 "Failed to either load to detect non-present NEON_AUTH_TOKEN environment variable"
     371            0 :             })
     372              :         }
     373              :     };
     374              : 
     375              :     // Top-level cancellation token for the process
     376          624 :     let shutdown_pageserver = tokio_util::sync::CancellationToken::new();
     377              : 
     378              :     // Set up remote storage client
     379          624 :     let remote_storage = create_remote_storage_client(conf)?;
     380              : 
     381              :     // Set up deletion queue
     382          624 :     let (deletion_queue, deletion_workers) = DeletionQueue::new(
     383          624 :         remote_storage.clone(),
     384          624 :         ControlPlaneClient::new(conf, &shutdown_pageserver),
     385          624 :         conf,
     386          624 :     );
     387          624 :     if let Some(deletion_workers) = deletion_workers {
     388          624 :         deletion_workers.spawn_with(BACKGROUND_RUNTIME.handle());
     389          624 :     }
     390              : 
     391              :     // Up to this point no significant I/O has been done: this should have been fast.  Record
     392              :     // duration prior to starting I/O intensive phase of startup.
     393          624 :     startup_checkpoint(started_startup_at, "initial", "Starting loading tenants");
     394          624 :     STARTUP_IS_LOADING.set(1);
     395          624 : 
     396          624 :     // Startup staging or optimizing:
     397          624 :     //
     398          624 :     // We want to minimize downtime for `page_service` connections, and trying not to overload
     399          624 :     // BACKGROUND_RUNTIME by doing initial compactions and initial logical sizes at the same time.
     400          624 :     //
     401          624 :     // init_done_rx will notify when all initial load operations have completed.
     402          624 :     //
     403          624 :     // background_jobs_can_start (same name used to hold off background jobs from starting at
     404          624 :     // consumer side) will be dropped once we can start the background jobs. Currently it is behind
     405          624 :     // completing all initial logical size calculations (init_logical_size_done_rx) and a timeout
     406          624 :     // (background_task_maximum_delay).
     407          624 :     let (init_remote_done_tx, init_remote_done_rx) = utils::completion::channel();
     408          624 :     let (init_done_tx, init_done_rx) = utils::completion::channel();
     409          624 : 
     410          624 :     let (background_jobs_can_start, background_jobs_barrier) = utils::completion::channel();
     411          624 : 
     412          624 :     let order = pageserver::InitializationOrder {
     413          624 :         initial_tenant_load_remote: Some(init_done_tx),
     414          624 :         initial_tenant_load: Some(init_remote_done_tx),
     415          624 :         background_jobs_can_start: background_jobs_barrier.clone(),
     416          624 :     };
     417          624 : 
     418          624 :     // Scan the local 'tenants/' directory and start loading the tenants
     419          624 :     let deletion_queue_client = deletion_queue.new_client();
     420          624 :     let tenant_manager = BACKGROUND_RUNTIME.block_on(mgr::init_tenant_mgr(
     421          624 :         conf,
     422          624 :         TenantSharedResources {
     423          624 :             broker_client: broker_client.clone(),
     424          624 :             remote_storage: remote_storage.clone(),
     425          624 :             deletion_queue_client,
     426          624 :         },
     427          624 :         order,
     428          624 :         shutdown_pageserver.clone(),
     429          624 :     ))?;
     430          624 :     let tenant_manager = Arc::new(tenant_manager);
     431          624 : 
     432          624 :     BACKGROUND_RUNTIME.spawn({
     433          624 :         let shutdown_pageserver = shutdown_pageserver.clone();
     434          624 :         let drive_init = async move {
     435          624 :             // NOTE: unlike many futures in pageserver, this one is cancellation-safe
     436          624 :             let guard = scopeguard::guard_on_success((), |_| {
     437            0 :                 tracing::info!("Cancelled before initial load completed")
     438          624 :             });
     439          624 : 
     440          624 :             let timeout = conf.background_task_maximum_delay;
     441          624 : 
     442          624 :             let init_remote_done = std::pin::pin!(async {
     443          624 :                 init_remote_done_rx.wait().await;
     444          612 :                 startup_checkpoint(
     445          612 :                     started_startup_at,
     446          612 :                     "initial_tenant_load_remote",
     447          612 :                     "Remote part of initial load completed",
     448          612 :                 );
     449          624 :             });
     450              : 
     451              :             let WaitForPhaseResult {
     452          617 :                 timeout_remaining: timeout,
     453          617 :                 skipped: init_remote_skipped,
     454          624 :             } = wait_for_phase("initial_tenant_load_remote", init_remote_done, timeout).await;
     455              : 
     456          617 :             let init_load_done = std::pin::pin!(async {
     457          617 :                 init_done_rx.wait().await;
     458          617 :                 startup_checkpoint(
     459          617 :                     started_startup_at,
     460          617 :                     "initial_tenant_load",
     461          617 :                     "Initial load completed",
     462          617 :                 );
     463          617 :                 STARTUP_IS_LOADING.set(0);
     464          617 :             });
     465              : 
     466              :             let WaitForPhaseResult {
     467          617 :                 timeout_remaining: _timeout,
     468          617 :                 skipped: init_load_skipped,
     469          617 :             } = wait_for_phase("initial_tenant_load", init_load_done, timeout).await;
     470              : 
     471              :             // initial logical sizes can now start, as they were waiting on init_done_rx.
     472              : 
     473          617 :             scopeguard::ScopeGuard::into_inner(guard);
     474          617 : 
     475          617 :             // allow background jobs to start: we either completed prior stages, or they reached timeout
     476          617 :             // and were skipped.  It is important that we do not let them block background jobs indefinitely,
     477          617 :             // because things like consumption metrics for billing are blocked by this barrier.
     478          617 :             drop(background_jobs_can_start);
     479          617 :             startup_checkpoint(
     480          617 :                 started_startup_at,
     481          617 :                 "background_jobs_can_start",
     482          617 :                 "Starting background jobs",
     483          617 :             );
     484          617 : 
     485          617 :             // We are done. If we skipped any phases due to timeout, run them to completion here so that
     486          617 :             // they will eventually update their startup_checkpoint, and so that we do not declare the
     487          617 :             // 'complete' stage until all the other stages are really done.
     488          617 :             let guard = scopeguard::guard_on_success((), |_| {
     489            0 :                 tracing::info!("Cancelled before waiting for skipped phases done")
     490          617 :             });
     491          617 :             if let Some(f) = init_remote_skipped {
     492            8 :                 f.await;
     493          609 :             }
     494          612 :             if let Some(f) = init_load_skipped {
     495            0 :                 f.await;
     496          612 :             }
     497          612 :             scopeguard::ScopeGuard::into_inner(guard);
     498          612 : 
     499          612 :             startup_checkpoint(started_startup_at, "complete", "Startup complete");
     500          624 :         };
     501          624 : 
     502          624 :         async move {
     503          624 :             let mut drive_init = std::pin::pin!(drive_init);
     504          624 :             // just race these tasks
     505          800 :             tokio::select! {
     506          800 :                 _ = shutdown_pageserver.cancelled() => {},
     507          800 :                 _ = &mut drive_init => {},
     508          800 :             }
     509          624 :         }
     510          624 :     });
     511              : 
     512          624 :     let secondary_controller = if let Some(remote_storage) = &remote_storage {
     513          624 :         secondary::spawn_tasks(
     514          624 :             tenant_manager.clone(),
     515          624 :             remote_storage.clone(),
     516          624 :             background_jobs_barrier.clone(),
     517          624 :             shutdown_pageserver.clone(),
     518          624 :         )
     519              :     } else {
     520            0 :         secondary::null_controller()
     521              :     };
     522              : 
     523              :     // shared state between the disk-usage backed eviction background task and the http endpoint
     524              :     // that allows triggering disk-usage based eviction manually. note that the http endpoint
     525              :     // is still accessible even if background task is not configured as long as remote storage has
     526              :     // been configured.
     527          624 :     let disk_usage_eviction_state: Arc<disk_usage_eviction_task::State> = Arc::default();
     528              : 
     529          624 :     if let Some(remote_storage) = &remote_storage {
     530          624 :         launch_disk_usage_global_eviction_task(
     531          624 :             conf,
     532          624 :             remote_storage.clone(),
     533          624 :             disk_usage_eviction_state.clone(),
     534          624 :             tenant_manager.clone(),
     535          624 :             background_jobs_barrier.clone(),
     536          624 :         )?;
     537            0 :     }
     538              : 
     539              :     // Start up the service to handle HTTP mgmt API request. We created the
     540              :     // listener earlier already.
     541              :     {
     542          624 :         let _rt_guard = MGMT_REQUEST_RUNTIME.enter();
     543              : 
     544          624 :         let router_state = Arc::new(
     545          624 :             http::routes::State::new(
     546          624 :                 conf,
     547          624 :                 tenant_manager,
     548          624 :                 http_auth.clone(),
     549          624 :                 remote_storage.clone(),
     550          624 :                 broker_client.clone(),
     551          624 :                 disk_usage_eviction_state,
     552          624 :                 deletion_queue.new_client(),
     553          624 :                 secondary_controller,
     554          624 :             )
     555          624 :             .context("Failed to initialize router state")?,
     556              :         );
     557          624 :         let router = http::make_router(router_state, launch_ts, http_auth.clone())?
     558          624 :             .build()
     559          624 :             .map_err(|err| anyhow!(err))?;
     560          624 :         let service = utils::http::RouterService::new(router).unwrap();
     561          624 :         let server = hyper::Server::from_tcp(http_listener)?
     562          624 :             .serve(service)
     563          624 :             .with_graceful_shutdown(task_mgr::shutdown_watcher());
     564          624 : 
     565          624 :         task_mgr::spawn(
     566          624 :             MGMT_REQUEST_RUNTIME.handle(),
     567          624 :             TaskKind::HttpEndpointListener,
     568          624 :             None,
     569          624 :             None,
     570          624 :             "http endpoint listener",
     571          624 :             true,
     572          624 :             async {
     573         4719 :                 server.await?;
     574          182 :                 Ok(())
     575          624 :             },
     576          624 :         );
     577              :     }
     578              : 
     579          624 :     if let Some(metric_collection_endpoint) = &conf.metric_collection_endpoint {
     580            6 :         let metrics_ctx = RequestContext::todo_child(
     581            6 :             TaskKind::MetricsCollection,
     582            6 :             // This task itself shouldn't download anything.
     583            6 :             // The actual size calculation does need downloads, and
     584            6 :             // creates a child context with the right DownloadBehavior.
     585            6 :             DownloadBehavior::Error,
     586            6 :         );
     587            6 : 
     588            6 :         let local_disk_storage = conf.workdir.join("last_consumption_metrics.json");
     589            6 : 
     590            6 :         task_mgr::spawn(
     591            6 :             crate::BACKGROUND_RUNTIME.handle(),
     592            6 :             TaskKind::MetricsCollection,
     593            6 :             None,
     594            6 :             None,
     595            6 :             "consumption metrics collection",
     596            6 :             true,
     597            6 :             async move {
     598            6 :                 // first wait until background jobs are cleared to launch.
     599            6 :                 //
     600            6 :                 // this is because we only process active tenants and timelines, and the
     601            6 :                 // Timeline::get_current_logical_size will spawn the logical size calculation,
     602            6 :                 // which will not be rate-limited.
     603            6 :                 let cancel = task_mgr::shutdown_token();
     604            6 : 
     605            9 :                 tokio::select! {
     606            9 :                     _ = cancel.cancelled() => { return Ok(()); },
     607            9 :                     _ = background_jobs_barrier.wait() => {}
     608            9 :                 };
     609              : 
     610            6 :                 pageserver::consumption_metrics::collect_metrics(
     611            6 :                     metric_collection_endpoint,
     612            6 :                     conf.metric_collection_interval,
     613            6 :                     conf.cached_metric_collection_interval,
     614            6 :                     conf.synthetic_size_calculation_interval,
     615            6 :                     conf.id,
     616            6 :                     local_disk_storage,
     617            6 :                     cancel,
     618            6 :                     metrics_ctx,
     619            6 :                 )
     620            6 :                 .instrument(info_span!("metrics_collection"))
     621          215 :                 .await?;
     622            3 :                 Ok(())
     623            6 :             },
     624            6 :         );
     625          618 :     }
     626              : 
     627              :     // Spawn a task to listen for libpq connections. It will spawn further tasks
     628              :     // for each connection. We created the listener earlier already.
     629          624 :     {
     630          624 :         let libpq_ctx = RequestContext::todo_child(
     631          624 :             TaskKind::LibpqEndpointListener,
     632          624 :             // listener task shouldn't need to download anything. (We will
     633          624 :             // create a separate sub-contexts for each connection, with their
     634          624 :             // own download behavior. This context is used only to listen and
     635          624 :             // accept connections.)
     636          624 :             DownloadBehavior::Error,
     637          624 :         );
     638          624 :         task_mgr::spawn(
     639          624 :             COMPUTE_REQUEST_RUNTIME.handle(),
     640          624 :             TaskKind::LibpqEndpointListener,
     641          624 :             None,
     642          624 :             None,
     643          624 :             "libpq endpoint listener",
     644          624 :             true,
     645          624 :             async move {
     646          624 :                 page_service::libpq_listener_main(
     647          624 :                     conf,
     648          624 :                     broker_client,
     649          624 :                     pg_auth,
     650          624 :                     pageserver_listener,
     651          624 :                     conf.pg_auth_type,
     652          624 :                     libpq_ctx,
     653          624 :                     task_mgr::shutdown_token(),
     654          624 :                 )
     655        10496 :                 .await
     656          624 :             },
     657          624 :         );
     658          624 :     }
     659          624 : 
     660          624 :     let mut shutdown_pageserver = Some(shutdown_pageserver.drop_guard());
     661          624 : 
     662          624 :     // All started up! Now just sit and wait for shutdown signal.
     663          624 :     {
     664          624 :         use signal_hook::consts::*;
     665          624 :         let signal_handler = BACKGROUND_RUNTIME.spawn_blocking(move || {
     666          624 :             let mut signals =
     667          624 :                 signal_hook::iterator::Signals::new([SIGINT, SIGTERM, SIGQUIT]).unwrap();
     668          624 :             return signals
     669          624 :                 .forever()
     670          624 :                 .next()
     671          624 :                 .expect("forever() never returns None unless explicitly closed");
     672          624 :         });
     673          624 :         let signal = BACKGROUND_RUNTIME
     674          624 :             .block_on(signal_handler)
     675          624 :             .expect("join error");
     676          624 :         match signal {
     677              :             SIGQUIT => {
     678          442 :                 info!("Got signal {signal}. Terminating in immediate shutdown mode",);
     679          438 :                 std::process::exit(111);
     680              :             }
     681              :             SIGINT | SIGTERM => {
     682          182 :                 info!("Got signal {signal}. Terminating gracefully in fast shutdown mode",);
     683              : 
     684              :                 // This cancels the `shutdown_pageserver` cancellation tree.
     685              :                 // Right now that tree doesn't reach very far, and `task_mgr` is used instead.
     686              :                 // The plan is to change that over time.
     687          182 :                 shutdown_pageserver.take();
     688          182 :                 let bg_remote_storage = remote_storage.clone();
     689          182 :                 let bg_deletion_queue = deletion_queue.clone();
     690          182 :                 BACKGROUND_RUNTIME.block_on(pageserver::shutdown_pageserver(
     691          182 :                     bg_remote_storage.map(|_| bg_deletion_queue),
     692          182 :                     0,
     693          182 :                 ));
     694          182 :                 unreachable!()
     695              :             }
     696            0 :             _ => unreachable!(),
     697              :         }
     698              :     }
     699            0 : }
     700              : 
     701          624 : fn create_remote_storage_client(
     702          624 :     conf: &'static PageServerConf,
     703          624 : ) -> anyhow::Result<Option<GenericRemoteStorage>> {
     704          624 :     let config = if let Some(config) = &conf.remote_storage_config {
     705          624 :         config
     706              :     } else {
     707            0 :         tracing::warn!("no remote storage configured, this is a deprecated configuration");
     708            0 :         return Ok(None);
     709              :     };
     710              : 
     711              :     // Create the client
     712          624 :     let mut remote_storage = GenericRemoteStorage::from_config(config)?;
     713              : 
     714              :     // If `test_remote_failures` is non-zero, wrap the client with a
     715              :     // wrapper that simulates failures.
     716          624 :     if conf.test_remote_failures > 0 {
     717           63 :         if !cfg!(feature = "testing") {
     718            0 :             anyhow::bail!("test_remote_failures option is not available because pageserver was compiled without the 'testing' feature");
     719           63 :         }
     720           63 :         info!(
     721           63 :             "Simulating remote failures for first {} attempts of each op",
     722           63 :             conf.test_remote_failures
     723           63 :         );
     724           63 :         remote_storage =
     725           63 :             GenericRemoteStorage::unreliable_wrapper(remote_storage, conf.test_remote_failures);
     726          561 :     }
     727              : 
     728          624 :     Ok(Some(remote_storage))
     729          624 : }
     730              : 
     731         1428 : fn cli() -> Command {
     732         1428 :     Command::new("Neon page server")
     733         1428 :         .about("Materializes WAL stream to pages and serves them to the postgres")
     734         1428 :         .version(version())
     735         1428 :         .arg(
     736         1428 :             Arg::new("init")
     737         1428 :                 .long("init")
     738         1428 :                 .action(ArgAction::SetTrue)
     739         1428 :                 .help("Initialize pageserver with all given config overrides"),
     740         1428 :         )
     741         1428 :         .arg(
     742         1428 :             Arg::new("workdir")
     743         1428 :                 .short('D')
     744         1428 :                 .long("workdir")
     745         1428 :                 .help("Working directory for the pageserver"),
     746         1428 :         )
     747         1428 :         // See `settings.md` for more details on the extra configuration patameters pageserver can process
     748         1428 :         .arg(
     749         1428 :             Arg::new("config-override")
     750         1428 :                 .short('c')
     751         1428 :                 .num_args(1)
     752         1428 :                 .action(ArgAction::Append)
     753         1428 :                 .help("Additional configuration overrides of the ones from the toml config file (or new ones to add there). \
     754         1428 :                 Any option has to be a valid toml document, example: `-c=\"foo='hey'\"` `-c=\"foo={value=1}\"`"),
     755         1428 :         )
     756         1428 :         .arg(
     757         1428 :             Arg::new("update-config")
     758         1428 :                 .long("update-config")
     759         1428 :                 .action(ArgAction::SetTrue)
     760         1428 :                 .help("Update the config file when started"),
     761         1428 :         )
     762         1428 :         .arg(
     763         1428 :             Arg::new("enabled-features")
     764         1428 :                 .long("enabled-features")
     765         1428 :                 .action(ArgAction::SetTrue)
     766         1428 :                 .help("Show enabled compile time features"),
     767         1428 :         )
     768         1428 : }
     769              : 
     770            2 : #[test]
     771            2 : fn verify_cli() {
     772            2 :     cli().debug_assert();
     773            2 : }
        

Generated by: LCOV version 2.1-beta