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