LCOV - code coverage report
Current view: top level - pageserver/src/bin - pageserver.rs (source / functions) Coverage Total Hit
Test: 1b0a6a0c05cee5a7de360813c8034804e105ce1c.info Lines: 4.7 % 575 27
Test Date: 2025-03-12 00:01:28 Functions: 8.1 % 37 3

            Line data    Source code
       1              : #![recursion_limit = "300"]
       2              : 
       3              : //! Main entry point for the Page Server executable.
       4              : 
       5              : use std::env;
       6              : use std::env::{VarError, var};
       7              : use std::io::Read;
       8              : use std::str::FromStr;
       9              : use std::sync::Arc;
      10              : use std::time::Duration;
      11              : 
      12              : use anyhow::{Context, anyhow};
      13              : use camino::Utf8Path;
      14              : use clap::{Arg, ArgAction, Command};
      15              : use metrics::launch_timestamp::{LaunchTimestamp, set_launch_timestamp_metric};
      16              : use metrics::set_build_info_metric;
      17              : use nix::sys::socket::{setsockopt, sockopt};
      18              : use pageserver::config::{PageServerConf, PageserverIdentity};
      19              : use pageserver::controller_upcall_client::ControllerUpcallClient;
      20              : use pageserver::deletion_queue::DeletionQueue;
      21              : use pageserver::disk_usage_eviction_task::{self, launch_disk_usage_global_eviction_task};
      22              : use pageserver::metrics::{STARTUP_DURATION, STARTUP_IS_LOADING};
      23              : use pageserver::task_mgr::{
      24              :     BACKGROUND_RUNTIME, COMPUTE_REQUEST_RUNTIME, MGMT_REQUEST_RUNTIME, WALRECEIVER_RUNTIME,
      25              : };
      26              : use pageserver::tenant::{TenantSharedResources, mgr, secondary};
      27              : use pageserver::{
      28              :     CancellableTask, ConsumptionMetricsTasks, HttpEndpointListener, HttpsEndpointListener, http,
      29              :     page_cache, page_service, task_mgr, virtual_file,
      30              : };
      31              : use postgres_backend::AuthType;
      32              : use remote_storage::GenericRemoteStorage;
      33              : use rustls_pki_types::{CertificateDer, PrivateKeyDer};
      34              : use tokio::signal::unix::SignalKind;
      35              : use tokio::time::Instant;
      36              : use tokio_util::sync::CancellationToken;
      37              : use tracing::*;
      38              : use utils::auth::{JwtAuth, SwappableJwtAuth};
      39              : use utils::crashsafe::syncfs;
      40              : use utils::logging::TracingErrorLayerEnablement;
      41              : use utils::sentry_init::init_sentry;
      42              : use utils::{failpoint_support, logging, project_build_tag, project_git_version, tcp_listener};
      43              : 
      44              : project_git_version!(GIT_VERSION);
      45              : project_build_tag!(BUILD_TAG);
      46              : 
      47              : #[global_allocator]
      48              : static GLOBAL: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
      49              : 
      50              : /// Configure jemalloc to profile heap allocations by sampling stack traces every 2 MB (1 << 21).
      51              : /// This adds roughly 3% overhead for allocations on average, which is acceptable considering
      52              : /// performance-sensitive code will avoid allocations as far as possible anyway.
      53              : #[allow(non_upper_case_globals)]
      54              : #[unsafe(export_name = "malloc_conf")]
      55              : pub static malloc_conf: &[u8] = b"prof:true,prof_active:true,lg_prof_sample:21\0";
      56              : 
      57              : const PID_FILE_NAME: &str = "pageserver.pid";
      58              : 
      59              : const FEATURES: &[&str] = &[
      60              :     #[cfg(feature = "testing")]
      61              :     "testing",
      62              : ];
      63              : 
      64            4 : fn version() -> String {
      65            4 :     format!(
      66            4 :         "{GIT_VERSION} failpoints: {}, features: {:?}",
      67            4 :         fail::has_failpoints(),
      68            4 :         FEATURES,
      69            4 :     )
      70            4 : }
      71              : 
      72            0 : fn main() -> anyhow::Result<()> {
      73            0 :     let launch_ts = Box::leak(Box::new(LaunchTimestamp::generate()));
      74            0 : 
      75            0 :     let arg_matches = cli().get_matches();
      76            0 : 
      77            0 :     if arg_matches.get_flag("enabled-features") {
      78            0 :         println!("{{\"features\": {FEATURES:?} }}");
      79            0 :         return Ok(());
      80            0 :     }
      81            0 : 
      82            0 :     // Initialize up failpoints support
      83            0 :     let scenario = failpoint_support::init();
      84            0 : 
      85            0 :     let workdir = arg_matches
      86            0 :         .get_one::<String>("workdir")
      87            0 :         .map(Utf8Path::new)
      88            0 :         .unwrap_or_else(|| Utf8Path::new(".neon"));
      89            0 :     let workdir = workdir
      90            0 :         .canonicalize_utf8()
      91            0 :         .with_context(|| format!("Error opening workdir '{workdir}'"))?;
      92              : 
      93            0 :     let cfg_file_path = workdir.join("pageserver.toml");
      94            0 :     let identity_file_path = workdir.join("identity.toml");
      95            0 : 
      96            0 :     // Set CWD to workdir for non-daemon modes
      97            0 :     env::set_current_dir(&workdir)
      98            0 :         .with_context(|| format!("Failed to set application's current dir to '{workdir}'"))?;
      99              : 
     100            0 :     let conf = initialize_config(&identity_file_path, &cfg_file_path, &workdir)?;
     101              : 
     102              :     // Initialize logging.
     103              :     //
     104              :     // It must be initialized before the custom panic hook is installed below.
     105              :     //
     106              :     // Regarding tracing_error enablement: at this time, we only use the
     107              :     // tracing_error crate to debug_assert that log spans contain tenant and timeline ids.
     108              :     // See `debug_assert_current_span_has_tenant_and_timeline_id` in the timeline module
     109            0 :     let tracing_error_layer_enablement = if cfg!(debug_assertions) {
     110            0 :         TracingErrorLayerEnablement::EnableWithRustLogFilter
     111              :     } else {
     112            0 :         TracingErrorLayerEnablement::Disabled
     113              :     };
     114            0 :     logging::init(
     115            0 :         conf.log_format,
     116            0 :         tracing_error_layer_enablement,
     117            0 :         logging::Output::Stdout,
     118            0 :     )?;
     119              : 
     120              :     // mind the order required here: 1. logging, 2. panic_hook, 3. sentry.
     121              :     // disarming this hook on pageserver, because we never tear down tracing.
     122            0 :     logging::replace_panic_hook_with_tracing_panic_hook().forget();
     123            0 : 
     124            0 :     // initialize sentry if SENTRY_DSN is provided
     125            0 :     let _sentry_guard = init_sentry(
     126            0 :         Some(GIT_VERSION.into()),
     127            0 :         &[("node_id", &conf.id.to_string())],
     128            0 :     );
     129            0 : 
     130            0 :     // after setting up logging, log the effective IO engine choice and read path implementations
     131            0 :     info!(?conf.virtual_file_io_engine, "starting with virtual_file IO engine");
     132            0 :     info!(?conf.virtual_file_io_mode, "starting with virtual_file IO mode");
     133            0 :     info!(?conf.wal_receiver_protocol, "starting with WAL receiver protocol");
     134            0 :     info!(?conf.validate_wal_contiguity, "starting with WAL contiguity validation");
     135            0 :     info!(?conf.page_service_pipelining, "starting with page service pipelining config");
     136            0 :     info!(?conf.get_vectored_concurrent_io, "starting with get_vectored IO concurrency config");
     137              : 
     138              :     // The tenants directory contains all the pageserver local disk state.
     139              :     // Create if not exists and make sure all the contents are durable before proceeding.
     140              :     // Ensuring durability eliminates a whole bug class where we come up after an unclean shutdown.
     141              :     // After unclea shutdown, we don't know if all the filesystem content we can read via syscalls is actually durable or not.
     142              :     // Examples for that: OOM kill, systemd killing us during shutdown, self abort due to unrecoverable IO error.
     143            0 :     let tenants_path = conf.tenants_path();
     144            0 :     {
     145            0 :         let open = || {
     146            0 :             nix::dir::Dir::open(
     147            0 :                 tenants_path.as_std_path(),
     148            0 :                 nix::fcntl::OFlag::O_DIRECTORY | nix::fcntl::OFlag::O_RDONLY,
     149            0 :                 nix::sys::stat::Mode::empty(),
     150            0 :             )
     151            0 :         };
     152            0 :         let dirfd = match open() {
     153            0 :             Ok(dirfd) => dirfd,
     154            0 :             Err(e) => match e {
     155              :                 nix::errno::Errno::ENOENT => {
     156            0 :                     utils::crashsafe::create_dir_all(&tenants_path).with_context(|| {
     157            0 :                         format!("Failed to create tenants root dir at '{tenants_path}'")
     158            0 :                     })?;
     159            0 :                     open().context("open tenants dir after creating it")?
     160              :                 }
     161            0 :                 e => anyhow::bail!(e),
     162              :             },
     163              :         };
     164              : 
     165            0 :         if conf.no_sync {
     166            0 :             info!("Skipping syncfs on startup");
     167              :         } else {
     168            0 :             let started = Instant::now();
     169            0 :             syncfs(dirfd)?;
     170            0 :             let elapsed = started.elapsed();
     171            0 :             info!(
     172            0 :                 elapsed_ms = elapsed.as_millis(),
     173            0 :                 "made tenant directory contents durable"
     174              :             );
     175              :         }
     176              :     }
     177              : 
     178              :     // Basic initialization of things that don't change after startup
     179            0 :     tracing::info!("Initializing virtual_file...");
     180              :     virtual_file::init(
     181            0 :         conf.max_file_descriptors,
     182            0 :         conf.virtual_file_io_engine,
     183            0 :         conf.virtual_file_io_mode,
     184            0 :         if conf.no_sync {
     185            0 :             virtual_file::SyncMode::UnsafeNoSync
     186              :         } else {
     187            0 :             virtual_file::SyncMode::Sync
     188              :         },
     189              :     );
     190            0 :     tracing::info!("Initializing page_cache...");
     191            0 :     page_cache::init(conf.page_cache_size);
     192            0 : 
     193            0 :     start_pageserver(launch_ts, conf).context("Failed to start pageserver")?;
     194              : 
     195            0 :     scenario.teardown();
     196            0 :     Ok(())
     197            0 : }
     198              : 
     199            0 : fn initialize_config(
     200            0 :     identity_file_path: &Utf8Path,
     201            0 :     cfg_file_path: &Utf8Path,
     202            0 :     workdir: &Utf8Path,
     203            0 : ) -> anyhow::Result<&'static PageServerConf> {
     204              :     // The deployment orchestrator writes out an indentity file containing the node id
     205              :     // for all pageservers. This file is the source of truth for the node id. In order
     206              :     // to allow for rolling back pageserver releases, the node id is also included in
     207              :     // the pageserver config that the deployment orchestrator writes to disk for the pageserver.
     208              :     // A rolled back version of the pageserver will get the node id from the pageserver.toml
     209              :     // config file.
     210            0 :     let identity = match std::fs::File::open(identity_file_path) {
     211            0 :         Ok(mut f) => {
     212            0 :             let md = f.metadata().context("stat config file")?;
     213            0 :             if !md.is_file() {
     214            0 :                 anyhow::bail!(
     215            0 :                     "Pageserver found identity file but it is a dir entry: {identity_file_path}. Aborting start up ..."
     216            0 :                 );
     217            0 :             }
     218            0 : 
     219            0 :             let mut s = String::new();
     220            0 :             f.read_to_string(&mut s).context("read identity file")?;
     221            0 :             toml_edit::de::from_str::<PageserverIdentity>(&s)?
     222              :         }
     223            0 :         Err(e) => {
     224            0 :             anyhow::bail!(
     225            0 :                 "Pageserver could not read identity file: {identity_file_path}: {e}. Aborting start up ..."
     226            0 :             );
     227              :         }
     228              :     };
     229              : 
     230            0 :     let config_file_contents =
     231            0 :         std::fs::read_to_string(cfg_file_path).context("read config file from filesystem")?;
     232            0 :     let config_toml = serde_path_to_error::deserialize(
     233            0 :         toml_edit::de::Deserializer::from_str(&config_file_contents)
     234            0 :             .context("build toml deserializer")?,
     235              :     )
     236            0 :     .context("deserialize config toml")?;
     237            0 :     let conf = PageServerConf::parse_and_validate(identity.id, config_toml, workdir)
     238            0 :         .context("runtime-validation of config toml")?;
     239              : 
     240            0 :     Ok(Box::leak(Box::new(conf)))
     241            0 : }
     242              : 
     243              : struct WaitForPhaseResult<F: std::future::Future + Unpin> {
     244              :     timeout_remaining: Duration,
     245              :     skipped: Option<F>,
     246              : }
     247              : 
     248              : /// During startup, we apply a timeout to our waits for readiness, to avoid
     249              : /// stalling the whole service if one Tenant experiences some problem.  Each
     250              : /// phase may consume some of the timeout: this function returns the updated
     251              : /// timeout for use in the next call.
     252            0 : async fn wait_for_phase<F>(phase: &str, mut fut: F, timeout: Duration) -> WaitForPhaseResult<F>
     253            0 : where
     254            0 :     F: std::future::Future + Unpin,
     255            0 : {
     256            0 :     let initial_t = Instant::now();
     257            0 :     let skipped = match tokio::time::timeout(timeout, &mut fut).await {
     258            0 :         Ok(_) => None,
     259              :         Err(_) => {
     260            0 :             tracing::info!(
     261            0 :                 timeout_millis = timeout.as_millis(),
     262            0 :                 %phase,
     263            0 :                 "Startup phase timed out, proceeding anyway"
     264              :             );
     265            0 :             Some(fut)
     266              :         }
     267              :     };
     268              : 
     269            0 :     WaitForPhaseResult {
     270            0 :         timeout_remaining: timeout
     271            0 :             .checked_sub(Instant::now().duration_since(initial_t))
     272            0 :             .unwrap_or(Duration::ZERO),
     273            0 :         skipped,
     274            0 :     }
     275            0 : }
     276              : 
     277            0 : fn startup_checkpoint(started_at: Instant, phase: &str, human_phase: &str) {
     278            0 :     let elapsed = started_at.elapsed();
     279            0 :     let secs = elapsed.as_secs_f64();
     280            0 :     STARTUP_DURATION.with_label_values(&[phase]).set(secs);
     281            0 : 
     282            0 :     info!(
     283            0 :         elapsed_ms = elapsed.as_millis(),
     284            0 :         "{human_phase} ({secs:.3}s since start)"
     285              :     )
     286            0 : }
     287              : 
     288            0 : fn start_pageserver(
     289            0 :     launch_ts: &'static LaunchTimestamp,
     290            0 :     conf: &'static PageServerConf,
     291            0 : ) -> anyhow::Result<()> {
     292            0 :     // Monotonic time for later calculating startup duration
     293            0 :     let started_startup_at = Instant::now();
     294            0 : 
     295            0 :     // Print version and launch timestamp to the log,
     296            0 :     // and expose them as prometheus metrics.
     297            0 :     // A changed version string indicates changed software.
     298            0 :     // A changed launch timestamp indicates a pageserver restart.
     299            0 :     info!(
     300            0 :         "version: {} launch_timestamp: {} build_tag: {}",
     301            0 :         version(),
     302            0 :         launch_ts.to_string(),
     303              :         BUILD_TAG,
     304              :     );
     305            0 :     set_build_info_metric(GIT_VERSION, BUILD_TAG);
     306            0 :     set_launch_timestamp_metric(launch_ts);
     307            0 :     #[cfg(target_os = "linux")]
     308            0 :     metrics::register_internal(Box::new(metrics::more_process_metrics::Collector::new())).unwrap();
     309            0 :     metrics::register_internal(Box::new(
     310            0 :         pageserver::metrics::tokio_epoll_uring::Collector::new(),
     311            0 :     ))
     312            0 :     .unwrap();
     313            0 :     pageserver::preinitialize_metrics(conf);
     314            0 : 
     315            0 :     // If any failpoints were set from FAILPOINTS environment variable,
     316            0 :     // print them to the log for debugging purposes
     317            0 :     let failpoints = fail::list();
     318            0 :     if !failpoints.is_empty() {
     319            0 :         info!(
     320            0 :             "started with failpoints: {}",
     321            0 :             failpoints
     322            0 :                 .iter()
     323            0 :                 .map(|(name, actions)| format!("{name}={actions}"))
     324            0 :                 .collect::<Vec<String>>()
     325            0 :                 .join(";")
     326              :         )
     327            0 :     }
     328              : 
     329              :     // Create and lock PID file. This ensures that there cannot be more than one
     330              :     // pageserver process running at the same time.
     331            0 :     let lock_file_path = conf.workdir.join(PID_FILE_NAME);
     332            0 :     info!("Claiming pid file at {lock_file_path:?}...");
     333            0 :     let lock_file =
     334            0 :         utils::pid_file::claim_for_current_process(&lock_file_path).context("claim pid file")?;
     335            0 :     info!("Claimed pid file at {lock_file_path:?}");
     336              : 
     337              :     // Ensure that the lock file is held even if the main thread of the process panics.
     338              :     // We need to release the lock file only when the process exits.
     339            0 :     std::mem::forget(lock_file);
     340            0 : 
     341            0 :     // Bind the HTTP and libpq ports early, so that if they are in use by some other
     342            0 :     // process, we error out early.
     343            0 :     let http_addr = &conf.listen_http_addr;
     344            0 :     info!("Starting pageserver http handler on {http_addr}");
     345            0 :     let http_listener = tcp_listener::bind(http_addr)?;
     346              : 
     347            0 :     let https_listener = match conf.listen_https_addr.as_ref() {
     348            0 :         Some(https_addr) => {
     349            0 :             info!("Starting pageserver https handler on {https_addr}");
     350            0 :             Some(tcp_listener::bind(https_addr)?)
     351              :         }
     352            0 :         None => None,
     353              :     };
     354              : 
     355            0 :     let pg_addr = &conf.listen_pg_addr;
     356            0 :     info!("Starting pageserver pg protocol handler on {pg_addr}");
     357            0 :     let pageserver_listener = tcp_listener::bind(pg_addr)?;
     358              : 
     359              :     // Enable SO_KEEPALIVE on the socket, to detect dead connections faster.
     360              :     // These are configured via net.ipv4.tcp_keepalive_* sysctls.
     361              :     //
     362              :     // TODO: also set this on the walreceiver socket, but tokio-postgres doesn't
     363              :     // support enabling keepalives while using the default OS sysctls.
     364            0 :     setsockopt(&pageserver_listener, sockopt::KeepAlive, &true)?;
     365              : 
     366              :     // Launch broker client
     367              :     // The storage_broker::connect call needs to happen inside a tokio runtime thread.
     368            0 :     let broker_client = WALRECEIVER_RUNTIME
     369            0 :         .block_on(async {
     370            0 :             // Note: we do not attempt connecting here (but validate endpoints sanity).
     371            0 :             storage_broker::connect(conf.broker_endpoint.clone(), conf.broker_keepalive_interval)
     372            0 :         })
     373            0 :         .with_context(|| {
     374            0 :             format!(
     375            0 :                 "create broker client for uri={:?} keepalive_interval={:?}",
     376            0 :                 &conf.broker_endpoint, conf.broker_keepalive_interval,
     377            0 :             )
     378            0 :         })?;
     379              : 
     380              :     // Initialize authentication for incoming connections
     381              :     let http_auth;
     382              :     let pg_auth;
     383            0 :     if conf.http_auth_type == AuthType::NeonJWT || conf.pg_auth_type == AuthType::NeonJWT {
     384              :         // unwrap is ok because check is performed when creating config, so path is set and exists
     385            0 :         let key_path = conf.auth_validation_public_key_path.as_ref().unwrap();
     386            0 :         info!("Loading public key(s) for verifying JWT tokens from {key_path:?}");
     387              : 
     388            0 :         let jwt_auth = JwtAuth::from_key_path(key_path)?;
     389            0 :         let auth: Arc<SwappableJwtAuth> = Arc::new(SwappableJwtAuth::new(jwt_auth));
     390              : 
     391            0 :         http_auth = match &conf.http_auth_type {
     392            0 :             AuthType::Trust => None,
     393            0 :             AuthType::NeonJWT => Some(auth.clone()),
     394              :         };
     395            0 :         pg_auth = match &conf.pg_auth_type {
     396            0 :             AuthType::Trust => None,
     397            0 :             AuthType::NeonJWT => Some(auth),
     398              :         };
     399            0 :     } else {
     400            0 :         http_auth = None;
     401            0 :         pg_auth = None;
     402            0 :     }
     403            0 :     info!("Using auth for http API: {:#?}", conf.http_auth_type);
     404            0 :     info!("Using auth for pg connections: {:#?}", conf.pg_auth_type);
     405              : 
     406            0 :     match var("NEON_AUTH_TOKEN") {
     407            0 :         Ok(v) => {
     408            0 :             info!("Loaded JWT token for authentication with Safekeeper");
     409            0 :             pageserver::config::SAFEKEEPER_AUTH_TOKEN
     410            0 :                 .set(Arc::new(v))
     411            0 :                 .map_err(|_| anyhow!("Could not initialize SAFEKEEPER_AUTH_TOKEN"))?;
     412              :         }
     413              :         Err(VarError::NotPresent) => {
     414            0 :             info!("No JWT token for authentication with Safekeeper detected");
     415              :         }
     416            0 :         Err(e) => return Err(e).with_context(
     417            0 :             || "Failed to either load to detect non-present NEON_AUTH_TOKEN environment variable",
     418            0 :         ),
     419              :     };
     420              : 
     421              :     // Top-level cancellation token for the process
     422            0 :     let shutdown_pageserver = tokio_util::sync::CancellationToken::new();
     423              : 
     424              :     // Set up remote storage client
     425            0 :     let remote_storage = BACKGROUND_RUNTIME.block_on(create_remote_storage_client(conf))?;
     426              : 
     427              :     // Set up deletion queue
     428            0 :     let (deletion_queue, deletion_workers) = DeletionQueue::new(
     429            0 :         remote_storage.clone(),
     430            0 :         ControllerUpcallClient::new(conf, &shutdown_pageserver),
     431            0 :         conf,
     432            0 :     );
     433            0 :     deletion_workers.spawn_with(BACKGROUND_RUNTIME.handle());
     434            0 : 
     435            0 :     // Up to this point no significant I/O has been done: this should have been fast.  Record
     436            0 :     // duration prior to starting I/O intensive phase of startup.
     437            0 :     startup_checkpoint(started_startup_at, "initial", "Starting loading tenants");
     438            0 :     STARTUP_IS_LOADING.set(1);
     439            0 : 
     440            0 :     // Startup staging or optimizing:
     441            0 :     //
     442            0 :     // We want to minimize downtime for `page_service` connections, and trying not to overload
     443            0 :     // BACKGROUND_RUNTIME by doing initial compactions and initial logical sizes at the same time.
     444            0 :     //
     445            0 :     // init_done_rx will notify when all initial load operations have completed.
     446            0 :     //
     447            0 :     // background_jobs_can_start (same name used to hold off background jobs from starting at
     448            0 :     // consumer side) will be dropped once we can start the background jobs. Currently it is behind
     449            0 :     // completing all initial logical size calculations (init_logical_size_done_rx) and a timeout
     450            0 :     // (background_task_maximum_delay).
     451            0 :     let (init_remote_done_tx, init_remote_done_rx) = utils::completion::channel();
     452            0 :     let (init_done_tx, init_done_rx) = utils::completion::channel();
     453            0 : 
     454            0 :     let (background_jobs_can_start, background_jobs_barrier) = utils::completion::channel();
     455            0 : 
     456            0 :     let order = pageserver::InitializationOrder {
     457            0 :         initial_tenant_load_remote: Some(init_done_tx),
     458            0 :         initial_tenant_load: Some(init_remote_done_tx),
     459            0 :         background_jobs_can_start: background_jobs_barrier.clone(),
     460            0 :     };
     461            0 : 
     462            0 :     info!(config=?conf.l0_flush, "using l0_flush config");
     463            0 :     let l0_flush_global_state =
     464            0 :         pageserver::l0_flush::L0FlushGlobalState::new(conf.l0_flush.clone());
     465            0 : 
     466            0 :     // Scan the local 'tenants/' directory and start loading the tenants
     467            0 :     let deletion_queue_client = deletion_queue.new_client();
     468            0 :     let background_purges = mgr::BackgroundPurges::default();
     469            0 :     let tenant_manager = BACKGROUND_RUNTIME.block_on(mgr::init_tenant_mgr(
     470            0 :         conf,
     471            0 :         background_purges.clone(),
     472            0 :         TenantSharedResources {
     473            0 :             broker_client: broker_client.clone(),
     474            0 :             remote_storage: remote_storage.clone(),
     475            0 :             deletion_queue_client,
     476            0 :             l0_flush_global_state,
     477            0 :         },
     478            0 :         order,
     479            0 :         shutdown_pageserver.clone(),
     480            0 :     ))?;
     481            0 :     let tenant_manager = Arc::new(tenant_manager);
     482            0 : 
     483            0 :     BACKGROUND_RUNTIME.spawn({
     484            0 :         let shutdown_pageserver = shutdown_pageserver.clone();
     485            0 :         let drive_init = async move {
     486            0 :             // NOTE: unlike many futures in pageserver, this one is cancellation-safe
     487            0 :             let guard = scopeguard::guard_on_success((), |_| {
     488            0 :                 tracing::info!("Cancelled before initial load completed")
     489            0 :             });
     490            0 : 
     491            0 :             let timeout = conf.background_task_maximum_delay;
     492            0 : 
     493            0 :             let init_remote_done = std::pin::pin!(async {
     494            0 :                 init_remote_done_rx.wait().await;
     495            0 :                 startup_checkpoint(
     496            0 :                     started_startup_at,
     497            0 :                     "initial_tenant_load_remote",
     498            0 :                     "Remote part of initial load completed",
     499            0 :                 );
     500            0 :             });
     501              : 
     502              :             let WaitForPhaseResult {
     503            0 :                 timeout_remaining: timeout,
     504            0 :                 skipped: init_remote_skipped,
     505            0 :             } = wait_for_phase("initial_tenant_load_remote", init_remote_done, timeout).await;
     506              : 
     507            0 :             let init_load_done = std::pin::pin!(async {
     508            0 :                 init_done_rx.wait().await;
     509            0 :                 startup_checkpoint(
     510            0 :                     started_startup_at,
     511            0 :                     "initial_tenant_load",
     512            0 :                     "Initial load completed",
     513            0 :                 );
     514            0 :                 STARTUP_IS_LOADING.set(0);
     515            0 :             });
     516              : 
     517              :             let WaitForPhaseResult {
     518            0 :                 timeout_remaining: _timeout,
     519            0 :                 skipped: init_load_skipped,
     520            0 :             } = wait_for_phase("initial_tenant_load", init_load_done, timeout).await;
     521              : 
     522              :             // initial logical sizes can now start, as they were waiting on init_done_rx.
     523              : 
     524            0 :             scopeguard::ScopeGuard::into_inner(guard);
     525            0 : 
     526            0 :             // allow background jobs to start: we either completed prior stages, or they reached timeout
     527            0 :             // and were skipped.  It is important that we do not let them block background jobs indefinitely,
     528            0 :             // because things like consumption metrics for billing are blocked by this barrier.
     529            0 :             drop(background_jobs_can_start);
     530            0 :             startup_checkpoint(
     531            0 :                 started_startup_at,
     532            0 :                 "background_jobs_can_start",
     533            0 :                 "Starting background jobs",
     534            0 :             );
     535            0 : 
     536            0 :             // We are done. If we skipped any phases due to timeout, run them to completion here so that
     537            0 :             // they will eventually update their startup_checkpoint, and so that we do not declare the
     538            0 :             // 'complete' stage until all the other stages are really done.
     539            0 :             let guard = scopeguard::guard_on_success((), |_| {
     540            0 :                 tracing::info!("Cancelled before waiting for skipped phases done")
     541            0 :             });
     542            0 :             if let Some(f) = init_remote_skipped {
     543            0 :                 f.await;
     544            0 :             }
     545            0 :             if let Some(f) = init_load_skipped {
     546            0 :                 f.await;
     547            0 :             }
     548            0 :             scopeguard::ScopeGuard::into_inner(guard);
     549            0 : 
     550            0 :             startup_checkpoint(started_startup_at, "complete", "Startup complete");
     551            0 :         };
     552            0 : 
     553            0 :         async move {
     554            0 :             let mut drive_init = std::pin::pin!(drive_init);
     555            0 :             // just race these tasks
     556            0 :             tokio::select! {
     557            0 :                 _ = shutdown_pageserver.cancelled() => {},
     558            0 :                 _ = &mut drive_init => {},
     559              :             }
     560            0 :         }
     561            0 :     });
     562            0 : 
     563            0 :     let (secondary_controller, secondary_controller_tasks) = secondary::spawn_tasks(
     564            0 :         tenant_manager.clone(),
     565            0 :         remote_storage.clone(),
     566            0 :         background_jobs_barrier.clone(),
     567            0 :         shutdown_pageserver.clone(),
     568            0 :     );
     569            0 : 
     570            0 :     // shared state between the disk-usage backed eviction background task and the http endpoint
     571            0 :     // that allows triggering disk-usage based eviction manually. note that the http endpoint
     572            0 :     // is still accessible even if background task is not configured as long as remote storage has
     573            0 :     // been configured.
     574            0 :     let disk_usage_eviction_state: Arc<disk_usage_eviction_task::State> = Arc::default();
     575            0 : 
     576            0 :     let disk_usage_eviction_task = launch_disk_usage_global_eviction_task(
     577            0 :         conf,
     578            0 :         remote_storage.clone(),
     579            0 :         disk_usage_eviction_state.clone(),
     580            0 :         tenant_manager.clone(),
     581            0 :         background_jobs_barrier.clone(),
     582            0 :     );
     583              : 
     584              :     // Start up the service to handle HTTP mgmt API request. We created the
     585              :     // listener earlier already.
     586            0 :     let (http_endpoint_listener, https_endpoint_listener) = {
     587            0 :         let _rt_guard = MGMT_REQUEST_RUNTIME.enter(); // for hyper
     588              : 
     589            0 :         let router_state = Arc::new(
     590            0 :             http::routes::State::new(
     591            0 :                 conf,
     592            0 :                 tenant_manager.clone(),
     593            0 :                 http_auth.clone(),
     594            0 :                 remote_storage.clone(),
     595            0 :                 broker_client.clone(),
     596            0 :                 disk_usage_eviction_state,
     597            0 :                 deletion_queue.new_client(),
     598            0 :                 secondary_controller,
     599            0 :             )
     600            0 :             .context("Failed to initialize router state")?,
     601              :         );
     602              : 
     603            0 :         let router = http::make_router(router_state, launch_ts, http_auth.clone())?
     604            0 :             .build()
     605            0 :             .map_err(|err| anyhow!(err))?;
     606              : 
     607            0 :         let service =
     608            0 :             Arc::new(http_utils::RequestServiceBuilder::new(router).map_err(|err| anyhow!(err))?);
     609              : 
     610            0 :         let http_task = {
     611            0 :             let server =
     612            0 :                 http_utils::server::Server::new(Arc::clone(&service), http_listener, None)?;
     613            0 :             let cancel = CancellationToken::new();
     614            0 : 
     615            0 :             let task = MGMT_REQUEST_RUNTIME.spawn(task_mgr::exit_on_panic_or_error(
     616            0 :                 "http endpoint listener",
     617            0 :                 server.serve(cancel.clone()),
     618            0 :             ));
     619            0 :             HttpEndpointListener(CancellableTask { task, cancel })
     620              :         };
     621              : 
     622            0 :         let https_task = match https_listener {
     623            0 :             Some(https_listener) => {
     624            0 :                 let certs = load_certs(&conf.ssl_cert_file)?;
     625            0 :                 let key = load_private_key(&conf.ssl_key_file)?;
     626              : 
     627            0 :                 let server_config = rustls::ServerConfig::builder()
     628            0 :                     .with_no_client_auth()
     629            0 :                     .with_single_cert(certs, key)?;
     630              : 
     631            0 :                 let tls_acceptor = tokio_rustls::TlsAcceptor::from(Arc::new(server_config));
     632              : 
     633            0 :                 let server =
     634            0 :                     http_utils::server::Server::new(service, https_listener, Some(tls_acceptor))?;
     635            0 :                 let cancel = CancellationToken::new();
     636            0 : 
     637            0 :                 let task = MGMT_REQUEST_RUNTIME.spawn(task_mgr::exit_on_panic_or_error(
     638            0 :                     "https endpoint listener",
     639            0 :                     server.serve(cancel.clone()),
     640            0 :                 ));
     641            0 :                 Some(HttpsEndpointListener(CancellableTask { task, cancel }))
     642              :             }
     643            0 :             None => None,
     644              :         };
     645              : 
     646            0 :         (http_task, https_task)
     647            0 :     };
     648            0 : 
     649            0 :     let consumption_metrics_tasks = {
     650            0 :         let cancel = shutdown_pageserver.child_token();
     651            0 :         let task = crate::BACKGROUND_RUNTIME.spawn({
     652            0 :             let tenant_manager = tenant_manager.clone();
     653            0 :             let cancel = cancel.clone();
     654            0 :             async move {
     655            0 :                 // first wait until background jobs are cleared to launch.
     656            0 :                 //
     657            0 :                 // this is because we only process active tenants and timelines, and the
     658            0 :                 // Timeline::get_current_logical_size will spawn the logical size calculation,
     659            0 :                 // which will not be rate-limited.
     660            0 :                 tokio::select! {
     661            0 :                     _ = cancel.cancelled() => { return; },
     662            0 :                     _ = background_jobs_barrier.wait() => {}
     663            0 :                 };
     664            0 : 
     665            0 :                 pageserver::consumption_metrics::run(conf, tenant_manager, cancel).await;
     666            0 :             }
     667            0 :         });
     668            0 :         ConsumptionMetricsTasks(CancellableTask { task, cancel })
     669              :     };
     670              : 
     671              :     // Spawn a task to listen for libpq connections. It will spawn further tasks
     672              :     // for each connection. We created the listener earlier already.
     673            0 :     let page_service = page_service::spawn(conf, tenant_manager.clone(), pg_auth, {
     674            0 :         let _entered = COMPUTE_REQUEST_RUNTIME.enter(); // TcpListener::from_std requires it
     675            0 :         pageserver_listener
     676            0 :             .set_nonblocking(true)
     677            0 :             .context("set listener to nonblocking")?;
     678            0 :         tokio::net::TcpListener::from_std(pageserver_listener).context("create tokio listener")?
     679              :     });
     680              : 
     681              :     // All started up! Now just sit and wait for shutdown signal.
     682            0 :     BACKGROUND_RUNTIME.block_on(async move {
     683            0 :         let signal_token = CancellationToken::new();
     684            0 :         let signal_cancel = signal_token.child_token();
     685            0 : 
     686            0 :         // Spawn signal handlers. Runs in a loop since we want to be responsive to multiple signals
     687            0 :         // even after triggering shutdown (e.g. a SIGQUIT after a slow SIGTERM shutdown). See:
     688            0 :         // https://github.com/neondatabase/neon/issues/9740.
     689            0 :         tokio::spawn(async move {
     690            0 :             let mut sigint = tokio::signal::unix::signal(SignalKind::interrupt()).unwrap();
     691            0 :             let mut sigterm = tokio::signal::unix::signal(SignalKind::terminate()).unwrap();
     692            0 :             let mut sigquit = tokio::signal::unix::signal(SignalKind::quit()).unwrap();
     693              : 
     694              :             loop {
     695            0 :                 let signal = tokio::select! {
     696            0 :                     _ = sigquit.recv() => {
     697            0 :                         info!("Got signal SIGQUIT. Terminating in immediate shutdown mode.");
     698            0 :                         std::process::exit(111);
     699              :                     }
     700            0 :                     _ = sigint.recv() => "SIGINT",
     701            0 :                     _ = sigterm.recv() => "SIGTERM",
     702              :                 };
     703              : 
     704            0 :                 if !signal_token.is_cancelled() {
     705            0 :                     info!("Got signal {signal}. Terminating gracefully in fast shutdown mode.");
     706            0 :                     signal_token.cancel();
     707              :                 } else {
     708            0 :                     info!("Got signal {signal}. Already shutting down.");
     709              :                 }
     710              :             }
     711            0 :         });
     712            0 : 
     713            0 :         // Wait for cancellation signal and shut down the pageserver.
     714            0 :         //
     715            0 :         // This cancels the `shutdown_pageserver` cancellation tree. Right now that tree doesn't
     716            0 :         // reach very far, and `task_mgr` is used instead. The plan is to change that over time.
     717            0 :         signal_cancel.cancelled().await;
     718              : 
     719            0 :         shutdown_pageserver.cancel();
     720            0 :         pageserver::shutdown_pageserver(
     721            0 :             http_endpoint_listener,
     722            0 :             https_endpoint_listener,
     723            0 :             page_service,
     724            0 :             consumption_metrics_tasks,
     725            0 :             disk_usage_eviction_task,
     726            0 :             &tenant_manager,
     727            0 :             background_purges,
     728            0 :             deletion_queue.clone(),
     729            0 :             secondary_controller_tasks,
     730            0 :             0,
     731            0 :         )
     732            0 :         .await;
     733            0 :         unreachable!();
     734            0 :     })
     735            0 : }
     736              : 
     737            0 : fn load_certs(filename: &Utf8Path) -> std::io::Result<Vec<CertificateDer<'static>>> {
     738            0 :     let file = std::fs::File::open(filename)?;
     739            0 :     let mut reader = std::io::BufReader::new(file);
     740            0 : 
     741            0 :     rustls_pemfile::certs(&mut reader).collect()
     742            0 : }
     743              : 
     744            0 : fn load_private_key(filename: &Utf8Path) -> anyhow::Result<PrivateKeyDer<'static>> {
     745            0 :     let file = std::fs::File::open(filename)?;
     746            0 :     let mut reader = std::io::BufReader::new(file);
     747              : 
     748            0 :     let key = rustls_pemfile::private_key(&mut reader)?;
     749              : 
     750            0 :     key.ok_or(anyhow::anyhow!(
     751            0 :         "no private key found in {}",
     752            0 :         filename.as_str(),
     753            0 :     ))
     754            0 : }
     755              : 
     756            0 : async fn create_remote_storage_client(
     757            0 :     conf: &'static PageServerConf,
     758            0 : ) -> anyhow::Result<GenericRemoteStorage> {
     759            0 :     let config = if let Some(config) = &conf.remote_storage_config {
     760            0 :         config
     761              :     } else {
     762            0 :         anyhow::bail!("no remote storage configured, this is a deprecated configuration");
     763              :     };
     764              : 
     765              :     // Create the client
     766            0 :     let mut remote_storage = GenericRemoteStorage::from_config(config).await?;
     767              : 
     768              :     // If `test_remote_failures` is non-zero, wrap the client with a
     769              :     // wrapper that simulates failures.
     770            0 :     if conf.test_remote_failures > 0 {
     771            0 :         if !cfg!(feature = "testing") {
     772            0 :             anyhow::bail!(
     773            0 :                 "test_remote_failures option is not available because pageserver was compiled without the 'testing' feature"
     774            0 :             );
     775            0 :         }
     776            0 :         info!(
     777            0 :             "Simulating remote failures for first {} attempts of each op",
     778              :             conf.test_remote_failures
     779              :         );
     780            0 :         remote_storage =
     781            0 :             GenericRemoteStorage::unreliable_wrapper(remote_storage, conf.test_remote_failures);
     782            0 :     }
     783              : 
     784            0 :     Ok(remote_storage)
     785            0 : }
     786              : 
     787            4 : fn cli() -> Command {
     788            4 :     Command::new("Neon page server")
     789            4 :         .about("Materializes WAL stream to pages and serves them to the postgres")
     790            4 :         .version(version())
     791            4 :         .arg(
     792            4 :             Arg::new("workdir")
     793            4 :                 .short('D')
     794            4 :                 .long("workdir")
     795            4 :                 .help("Working directory for the pageserver"),
     796            4 :         )
     797            4 :         .arg(
     798            4 :             Arg::new("enabled-features")
     799            4 :                 .long("enabled-features")
     800            4 :                 .action(ArgAction::SetTrue)
     801            4 :                 .help("Show enabled compile time features"),
     802            4 :         )
     803            4 : }
     804              : 
     805              : #[test]
     806            4 : fn verify_cli() {
     807            4 :     cli().debug_assert();
     808            4 : }
        

Generated by: LCOV version 2.1-beta