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