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