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 2056 : fn version() -> String {
54 2056 : format!(
55 2056 : "{GIT_VERSION} failpoints: {}, features: {:?}",
56 2056 : fail::has_failpoints(),
57 2056 : FEATURES,
58 2056 : )
59 2056 : }
60 :
61 1429 : fn main() -> anyhow::Result<()> {
62 1429 : let launch_ts = Box::leak(Box::new(LaunchTimestamp::generate()));
63 1429 :
64 1429 : let arg_matches = cli().get_matches();
65 1429 :
66 1429 : if arg_matches.get_flag("enabled-features") {
67 0 : println!("{{\"features\": {FEATURES:?} }}");
68 0 : return Ok(());
69 1429 : }
70 1429 :
71 1429 : let workdir = arg_matches
72 1429 : .get_one::<String>("workdir")
73 1429 : .map(Utf8Path::new)
74 1429 : .unwrap_or_else(|| Utf8Path::new(".neon"));
75 1429 : let workdir = workdir
76 1429 : .canonicalize_utf8()
77 1429 : .with_context(|| format!("Error opening workdir '{workdir}'"))?;
78 :
79 1429 : let cfg_file_path = workdir.join("pageserver.toml");
80 1429 :
81 1429 : // Set CWD to workdir for non-daemon modes
82 1429 : env::set_current_dir(&workdir)
83 1429 : .with_context(|| format!("Failed to set application's current dir to '{workdir}'"))?;
84 :
85 1429 : let conf = match initialize_config(&cfg_file_path, arg_matches, &workdir)? {
86 625 : ControlFlow::Continue(conf) => conf,
87 : ControlFlow::Break(()) => {
88 401 : info!("Pageserver config init successful");
89 401 : 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 625 : let tracing_error_layer_enablement = if cfg!(debug_assertions) {
101 625 : TracingErrorLayerEnablement::EnableWithRustLogFilter
102 : } else {
103 0 : TracingErrorLayerEnablement::Disabled
104 : };
105 625 : logging::init(
106 625 : conf.log_format,
107 625 : tracing_error_layer_enablement,
108 625 : logging::Output::Stdout,
109 625 : )?;
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 625 : logging::replace_panic_hook_with_tracing_panic_hook().forget();
114 625 :
115 625 : // initialize sentry if SENTRY_DSN is provided
116 625 : let _sentry_guard = init_sentry(
117 625 : Some(GIT_VERSION.into()),
118 625 : &[("node_id", &conf.id.to_string())],
119 625 : );
120 625 :
121 625 : let tenants_path = conf.tenants_path();
122 625 : if !tenants_path.exists() {
123 399 : utils::crashsafe::create_dir_all(conf.tenants_path())
124 399 : .with_context(|| format!("Failed to create tenants root dir at '{tenants_path}'"))?;
125 226 : }
126 :
127 : // Initialize up failpoints support
128 625 : let scenario = failpoint_support::init();
129 625 :
130 625 : // Basic initialization of things that don't change after startup
131 625 : virtual_file::init(conf.max_file_descriptors, conf.virtual_file_io_engine);
132 625 : page_cache::init(conf.page_cache_size);
133 625 :
134 625 : start_pageserver(launch_ts, conf).context("Failed to start pageserver")?;
135 :
136 0 : scenario.teardown();
137 0 : Ok(())
138 404 : }
139 :
140 1029 : fn initialize_config(
141 1029 : cfg_file_path: &Utf8Path,
142 1029 : arg_matches: clap::ArgMatches,
143 1029 : workdir: &Utf8Path,
144 1029 : ) -> anyhow::Result<ControlFlow<(), &'static PageServerConf>> {
145 1029 : let init = arg_matches.get_flag("init");
146 1029 : let update_config = init || arg_matches.get_flag("update-config");
147 :
148 1029 : let (mut toml, config_file_exists) = if cfg_file_path.is_file() {
149 627 : 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 626 : }
154 : // Supplement the CLI arguments with the config file
155 626 : let cfg_file_contents = std::fs::read_to_string(cfg_file_path)
156 626 : .with_context(|| format!("Failed to read pageserver config at '{cfg_file_path}'"))?;
157 : (
158 626 : cfg_file_contents
159 626 : .parse::<toml_edit::Document>()
160 626 : .with_context(|| {
161 0 : format!("Failed to parse '{cfg_file_path}' as pageserver config")
162 626 : })?,
163 : true,
164 : )
165 402 : } 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 402 : DEFAULT_CONFIG_FILE
171 402 : .parse::<toml_edit::Document>()
172 402 : .context("could not parse built-in config file")?,
173 : false,
174 : )
175 : };
176 :
177 1028 : if let Some(values) = arg_matches.get_many::<String>("config-override") {
178 10464 : for option_line in values {
179 9437 : 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 9437 : })?;
182 :
183 9463 : for (key, item) in doc.iter() {
184 9463 : 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 9462 : }
187 9462 : toml.insert(key, item.clone());
188 : }
189 : }
190 0 : }
191 :
192 1027 : debug!("Resulting toml: {toml}");
193 1027 : let conf = PageServerConf::parse_and_validate(&toml, workdir)
194 1027 : .context("Failed to parse pageserver configuration")?;
195 :
196 1026 : if update_config {
197 401 : info!("Writing pageserver config to '{cfg_file_path}'");
198 :
199 401 : std::fs::write(cfg_file_path, toml.to_string())
200 401 : .with_context(|| format!("Failed to write pageserver config to '{cfg_file_path}'"))?;
201 401 : info!("Config successfully written to '{cfg_file_path}'")
202 625 : }
203 :
204 1026 : Ok(if init {
205 401 : ControlFlow::Break(())
206 : } else {
207 625 : ControlFlow::Continue(Box::leak(Box::new(conf)))
208 : })
209 1029 : }
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 1224 : 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 1232 : WaitForPhaseResult {
238 1232 : timeout_remaining: timeout
239 1232 : .checked_sub(Instant::now().duration_since(initial_t))
240 1232 : .unwrap_or(Duration::ZERO),
241 1232 : skipped,
242 1232 : }
243 1232 : }
244 :
245 3079 : fn startup_checkpoint(started_at: Instant, phase: &str, human_phase: &str) {
246 3079 : let elapsed = started_at.elapsed();
247 3079 : let secs = elapsed.as_secs_f64();
248 3079 : STARTUP_DURATION.with_label_values(&[phase]).set(secs);
249 3079 :
250 3079 : info!(
251 3079 : elapsed_ms = elapsed.as_millis(),
252 3079 : "{human_phase} ({secs:.3}s since start)"
253 3079 : )
254 3079 : }
255 :
256 625 : fn start_pageserver(
257 625 : launch_ts: &'static LaunchTimestamp,
258 625 : conf: &'static PageServerConf,
259 625 : ) -> anyhow::Result<()> {
260 625 : // Monotonic time for later calculating startup duration
261 625 : let started_startup_at = Instant::now();
262 625 :
263 625 : // Print version and launch timestamp to the log,
264 625 : // and expose them as prometheus metrics.
265 625 : // A changed version string indicates changed software.
266 625 : // A changed launch timestamp indicates a pageserver restart.
267 625 : info!(
268 625 : "version: {} launch_timestamp: {} build_tag: {}",
269 625 : version(),
270 625 : launch_ts.to_string(),
271 625 : BUILD_TAG,
272 625 : );
273 625 : set_build_info_metric(GIT_VERSION, BUILD_TAG);
274 625 : set_launch_timestamp_metric(launch_ts);
275 625 : #[cfg(target_os = "linux")]
276 625 : metrics::register_internal(Box::new(metrics::more_process_metrics::Collector::new())).unwrap();
277 625 : metrics::register_internal(Box::new(
278 625 : pageserver::metrics::tokio_epoll_uring::Collector::new(),
279 625 : ))
280 625 : .unwrap();
281 625 : pageserver::preinitialize_metrics();
282 625 :
283 625 : // If any failpoints were set from FAILPOINTS environment variable,
284 625 : // print them to the log for debugging purposes
285 625 : let failpoints = fail::list();
286 625 : 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 617 : }
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 625 : let lock_file_path = conf.workdir.join(PID_FILE_NAME);
300 625 : let lock_file =
301 625 : utils::pid_file::claim_for_current_process(&lock_file_path).context("claim pid file")?;
302 625 : 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 625 : std::mem::forget(lock_file);
307 625 :
308 625 : // Bind the HTTP and libpq ports early, so that if they are in use by some other
309 625 : // process, we error out early.
310 625 : let http_addr = &conf.listen_http_addr;
311 625 : info!("Starting pageserver http handler on {http_addr}");
312 625 : let http_listener = tcp_listener::bind(http_addr)?;
313 :
314 625 : let pg_addr = &conf.listen_pg_addr;
315 625 : info!("Starting pageserver pg protocol handler on {pg_addr}");
316 625 : 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 625 : let broker_client = WALRECEIVER_RUNTIME
321 625 : .block_on(async {
322 625 : // Note: we do not attempt connecting here (but validate endpoints sanity).
323 625 : storage_broker::connect(conf.broker_endpoint.clone(), conf.broker_keepalive_interval)
324 625 : })
325 625 : .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 625 : })?;
331 :
332 : // Initialize authentication for incoming connections
333 : let http_auth;
334 : let pg_auth;
335 625 : 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 614 : } else {
352 614 : http_auth = None;
353 614 : pg_auth = None;
354 614 : }
355 625 : info!("Using auth for http API: {:#?}", conf.http_auth_type);
356 625 : info!("Using auth for pg connections: {:#?}", conf.pg_auth_type);
357 :
358 614 : 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 614 : 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 625 : let shutdown_pageserver = tokio_util::sync::CancellationToken::new();
377 :
378 : // Set up remote storage client
379 625 : let remote_storage = create_remote_storage_client(conf)?;
380 :
381 : // Set up deletion queue
382 625 : let (deletion_queue, deletion_workers) = DeletionQueue::new(
383 625 : remote_storage.clone(),
384 625 : ControlPlaneClient::new(conf, &shutdown_pageserver),
385 625 : conf,
386 625 : );
387 625 : if let Some(deletion_workers) = deletion_workers {
388 625 : deletion_workers.spawn_with(BACKGROUND_RUNTIME.handle());
389 625 : }
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 625 : startup_checkpoint(started_startup_at, "initial", "Starting loading tenants");
394 625 : STARTUP_IS_LOADING.set(1);
395 625 :
396 625 : // Startup staging or optimizing:
397 625 : //
398 625 : // We want to minimize downtime for `page_service` connections, and trying not to overload
399 625 : // BACKGROUND_RUNTIME by doing initial compactions and initial logical sizes at the same time.
400 625 : //
401 625 : // init_done_rx will notify when all initial load operations have completed.
402 625 : //
403 625 : // background_jobs_can_start (same name used to hold off background jobs from starting at
404 625 : // consumer side) will be dropped once we can start the background jobs. Currently it is behind
405 625 : // completing all initial logical size calculations (init_logical_size_done_rx) and a timeout
406 625 : // (background_task_maximum_delay).
407 625 : let (init_remote_done_tx, init_remote_done_rx) = utils::completion::channel();
408 625 : let (init_done_tx, init_done_rx) = utils::completion::channel();
409 625 :
410 625 : let (background_jobs_can_start, background_jobs_barrier) = utils::completion::channel();
411 625 :
412 625 : let order = pageserver::InitializationOrder {
413 625 : initial_tenant_load_remote: Some(init_done_tx),
414 625 : initial_tenant_load: Some(init_remote_done_tx),
415 625 : background_jobs_can_start: background_jobs_barrier.clone(),
416 625 : };
417 625 :
418 625 : // Scan the local 'tenants/' directory and start loading the tenants
419 625 : let deletion_queue_client = deletion_queue.new_client();
420 625 : let tenant_manager = BACKGROUND_RUNTIME.block_on(mgr::init_tenant_mgr(
421 625 : conf,
422 625 : TenantSharedResources {
423 625 : broker_client: broker_client.clone(),
424 625 : remote_storage: remote_storage.clone(),
425 625 : deletion_queue_client,
426 625 : },
427 625 : order,
428 625 : shutdown_pageserver.clone(),
429 625 : ))?;
430 625 : let tenant_manager = Arc::new(tenant_manager);
431 625 :
432 625 : BACKGROUND_RUNTIME.spawn({
433 625 : let shutdown_pageserver = shutdown_pageserver.clone();
434 625 : let drive_init = async move {
435 625 : // NOTE: unlike many futures in pageserver, this one is cancellation-safe
436 625 : let guard = scopeguard::guard_on_success((), |_| {
437 0 : tracing::info!("Cancelled before initial load completed")
438 625 : });
439 625 :
440 625 : let timeout = conf.background_task_maximum_delay;
441 625 :
442 625 : let init_remote_done = std::pin::pin!(async {
443 625 : init_remote_done_rx.wait().await;
444 611 : startup_checkpoint(
445 611 : started_startup_at,
446 611 : "initial_tenant_load_remote",
447 611 : "Remote part of initial load completed",
448 611 : );
449 625 : });
450 :
451 : let WaitForPhaseResult {
452 616 : timeout_remaining: timeout,
453 616 : skipped: init_remote_skipped,
454 625 : } = wait_for_phase("initial_tenant_load_remote", init_remote_done, timeout).await;
455 :
456 616 : let init_load_done = std::pin::pin!(async {
457 616 : init_done_rx.wait().await;
458 616 : startup_checkpoint(
459 616 : started_startup_at,
460 616 : "initial_tenant_load",
461 616 : "Initial load completed",
462 616 : );
463 616 : STARTUP_IS_LOADING.set(0);
464 616 : });
465 :
466 : let WaitForPhaseResult {
467 616 : timeout_remaining: _timeout,
468 616 : skipped: init_load_skipped,
469 616 : } = 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 616 : scopeguard::ScopeGuard::into_inner(guard);
474 616 :
475 616 : // allow background jobs to start: we either completed prior stages, or they reached timeout
476 616 : // and were skipped. It is important that we do not let them block background jobs indefinitely,
477 616 : // because things like consumption metrics for billing are blocked by this barrier.
478 616 : drop(background_jobs_can_start);
479 616 : startup_checkpoint(
480 616 : started_startup_at,
481 616 : "background_jobs_can_start",
482 616 : "Starting background jobs",
483 616 : );
484 616 :
485 616 : // We are done. If we skipped any phases due to timeout, run them to completion here so that
486 616 : // they will eventually update their startup_checkpoint, and so that we do not declare the
487 616 : // 'complete' stage until all the other stages are really done.
488 616 : let guard = scopeguard::guard_on_success((), |_| {
489 0 : tracing::info!("Cancelled before waiting for skipped phases done")
490 616 : });
491 616 : if let Some(f) = init_remote_skipped {
492 8 : f.await;
493 608 : }
494 611 : if let Some(f) = init_load_skipped {
495 0 : f.await;
496 611 : }
497 611 : scopeguard::ScopeGuard::into_inner(guard);
498 611 :
499 611 : startup_checkpoint(started_startup_at, "complete", "Startup complete");
500 625 : };
501 625 :
502 625 : async move {
503 625 : let mut drive_init = std::pin::pin!(drive_init);
504 625 : // just race these tasks
505 800 : tokio::select! {
506 800 : _ = shutdown_pageserver.cancelled() => {},
507 800 : _ = &mut drive_init => {},
508 800 : }
509 625 : }
510 625 : });
511 :
512 625 : let secondary_controller = if let Some(remote_storage) = &remote_storage {
513 625 : secondary::spawn_tasks(
514 625 : tenant_manager.clone(),
515 625 : remote_storage.clone(),
516 625 : background_jobs_barrier.clone(),
517 625 : shutdown_pageserver.clone(),
518 625 : )
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 625 : let disk_usage_eviction_state: Arc<disk_usage_eviction_task::State> = Arc::default();
528 :
529 625 : if let Some(remote_storage) = &remote_storage {
530 625 : launch_disk_usage_global_eviction_task(
531 625 : conf,
532 625 : remote_storage.clone(),
533 625 : disk_usage_eviction_state.clone(),
534 625 : tenant_manager.clone(),
535 625 : background_jobs_barrier.clone(),
536 625 : )?;
537 0 : }
538 :
539 : // Start up the service to handle HTTP mgmt API request. We created the
540 : // listener earlier already.
541 : {
542 625 : let _rt_guard = MGMT_REQUEST_RUNTIME.enter();
543 :
544 625 : let router_state = Arc::new(
545 625 : http::routes::State::new(
546 625 : conf,
547 625 : tenant_manager,
548 625 : http_auth.clone(),
549 625 : remote_storage.clone(),
550 625 : broker_client.clone(),
551 625 : disk_usage_eviction_state,
552 625 : deletion_queue.new_client(),
553 625 : secondary_controller,
554 625 : )
555 625 : .context("Failed to initialize router state")?,
556 : );
557 625 : let router = http::make_router(router_state, launch_ts, http_auth.clone())?
558 625 : .build()
559 625 : .map_err(|err| anyhow!(err))?;
560 625 : let service = utils::http::RouterService::new(router).unwrap();
561 625 : let server = hyper::Server::from_tcp(http_listener)?
562 625 : .serve(service)
563 625 : .with_graceful_shutdown(task_mgr::shutdown_watcher());
564 625 :
565 625 : task_mgr::spawn(
566 625 : MGMT_REQUEST_RUNTIME.handle(),
567 625 : TaskKind::HttpEndpointListener,
568 625 : None,
569 625 : None,
570 625 : "http endpoint listener",
571 625 : true,
572 625 : async {
573 4612 : server.await?;
574 183 : Ok(())
575 625 : },
576 625 : );
577 : }
578 :
579 625 : 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 263 : .await?;
622 3 : Ok(())
623 6 : },
624 6 : );
625 619 : }
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 625 : {
630 625 : let libpq_ctx = RequestContext::todo_child(
631 625 : TaskKind::LibpqEndpointListener,
632 625 : // listener task shouldn't need to download anything. (We will
633 625 : // create a separate sub-contexts for each connection, with their
634 625 : // own download behavior. This context is used only to listen and
635 625 : // accept connections.)
636 625 : DownloadBehavior::Error,
637 625 : );
638 625 : task_mgr::spawn(
639 625 : COMPUTE_REQUEST_RUNTIME.handle(),
640 625 : TaskKind::LibpqEndpointListener,
641 625 : None,
642 625 : None,
643 625 : "libpq endpoint listener",
644 625 : true,
645 625 : async move {
646 625 : page_service::libpq_listener_main(
647 625 : conf,
648 625 : broker_client,
649 625 : pg_auth,
650 625 : pageserver_listener,
651 625 : conf.pg_auth_type,
652 625 : libpq_ctx,
653 625 : task_mgr::shutdown_token(),
654 625 : )
655 10514 : .await
656 625 : },
657 625 : );
658 625 : }
659 625 :
660 625 : let mut shutdown_pageserver = Some(shutdown_pageserver.drop_guard());
661 625 :
662 625 : // All started up! Now just sit and wait for shutdown signal.
663 625 : {
664 625 : use signal_hook::consts::*;
665 625 : let signal_handler = BACKGROUND_RUNTIME.spawn_blocking(move || {
666 625 : let mut signals =
667 625 : signal_hook::iterator::Signals::new([SIGINT, SIGTERM, SIGQUIT]).unwrap();
668 625 : return signals
669 625 : .forever()
670 625 : .next()
671 625 : .expect("forever() never returns None unless explicitly closed");
672 625 : });
673 625 : let signal = BACKGROUND_RUNTIME
674 625 : .block_on(signal_handler)
675 625 : .expect("join error");
676 625 : 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 183 : 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 183 : shutdown_pageserver.take();
688 183 : let bg_remote_storage = remote_storage.clone();
689 183 : let bg_deletion_queue = deletion_queue.clone();
690 183 : BACKGROUND_RUNTIME.block_on(pageserver::shutdown_pageserver(
691 183 : bg_remote_storage.map(|_| bg_deletion_queue),
692 183 : 0,
693 183 : ));
694 183 : unreachable!()
695 : }
696 0 : _ => unreachable!(),
697 : }
698 : }
699 0 : }
700 :
701 625 : fn create_remote_storage_client(
702 625 : conf: &'static PageServerConf,
703 625 : ) -> anyhow::Result<Option<GenericRemoteStorage>> {
704 625 : let config = if let Some(config) = &conf.remote_storage_config {
705 625 : 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 625 : 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 625 : 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 562 : }
727 :
728 625 : Ok(Some(remote_storage))
729 625 : }
730 :
731 1431 : fn cli() -> Command {
732 1431 : Command::new("Neon page server")
733 1431 : .about("Materializes WAL stream to pages and serves them to the postgres")
734 1431 : .version(version())
735 1431 : .arg(
736 1431 : Arg::new("init")
737 1431 : .long("init")
738 1431 : .action(ArgAction::SetTrue)
739 1431 : .help("Initialize pageserver with all given config overrides"),
740 1431 : )
741 1431 : .arg(
742 1431 : Arg::new("workdir")
743 1431 : .short('D')
744 1431 : .long("workdir")
745 1431 : .help("Working directory for the pageserver"),
746 1431 : )
747 1431 : // See `settings.md` for more details on the extra configuration patameters pageserver can process
748 1431 : .arg(
749 1431 : Arg::new("config-override")
750 1431 : .short('c')
751 1431 : .num_args(1)
752 1431 : .action(ArgAction::Append)
753 1431 : .help("Additional configuration overrides of the ones from the toml config file (or new ones to add there). \
754 1431 : Any option has to be a valid toml document, example: `-c=\"foo='hey'\"` `-c=\"foo={value=1}\"`"),
755 1431 : )
756 1431 : .arg(
757 1431 : Arg::new("update-config")
758 1431 : .long("update-config")
759 1431 : .action(ArgAction::SetTrue)
760 1431 : .help("Update the config file when started"),
761 1431 : )
762 1431 : .arg(
763 1431 : Arg::new("enabled-features")
764 1431 : .long("enabled-features")
765 1431 : .action(ArgAction::SetTrue)
766 1431 : .help("Show enabled compile time features"),
767 1431 : )
768 1431 : }
769 :
770 2 : #[test]
771 2 : fn verify_cli() {
772 2 : cli().debug_assert();
773 2 : }
|
