Line data Source code
1 : //! Code to manage compute endpoints
2 : //!
3 : //! In the local test environment, the data for each endpoint is stored in
4 : //!
5 : //! ```text
6 : //! .neon/endpoints/<endpoint id>
7 : //! ```
8 : //!
9 : //! Some basic information about the endpoint, like the tenant and timeline IDs,
10 : //! are stored in the `endpoint.json` file. The `endpoint.json` file is created
11 : //! when the endpoint is created, and doesn't change afterwards.
12 : //!
13 : //! The endpoint is managed by the `compute_ctl` binary. When an endpoint is
14 : //! started, we launch `compute_ctl` It synchronizes the safekeepers, downloads
15 : //! the basebackup from the pageserver to initialize the data directory, and
16 : //! finally launches the PostgreSQL process. It watches the PostgreSQL process
17 : //! until it exits.
18 : //!
19 : //! When an endpoint is created, a `postgresql.conf` file is also created in
20 : //! the endpoint's directory. The file can be modified before starting PostgreSQL.
21 : //! However, the `postgresql.conf` file in the endpoint directory is not used directly
22 : //! by PostgreSQL. It is passed to `compute_ctl`, and `compute_ctl` writes another
23 : //! copy of it in the data directory.
24 : //!
25 : //! Directory contents:
26 : //!
27 : //! ```text
28 : //! .neon/endpoints/main/
29 : //! compute.log - log output of `compute_ctl` and `postgres`
30 : //! endpoint.json - serialized `EndpointConf` struct
31 : //! postgresql.conf - postgresql settings
32 : //! spec.json - passed to `compute_ctl`
33 : //! pgdata/
34 : //! postgresql.conf - copy of postgresql.conf created by `compute_ctl`
35 : //! zenith.signal
36 : //! <other PostgreSQL files>
37 : //! ```
38 : //!
39 : use std::collections::BTreeMap;
40 : use std::net::{IpAddr, Ipv4Addr, SocketAddr, TcpStream};
41 : use std::path::PathBuf;
42 : use std::process::Command;
43 : use std::str::FromStr;
44 : use std::sync::Arc;
45 : use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
46 :
47 : use anyhow::{Context, Result, anyhow, bail};
48 : use compute_api::requests::ConfigurationRequest;
49 : use compute_api::responses::{ComputeCtlConfig, ComputeStatus, ComputeStatusResponse};
50 : use compute_api::spec::{
51 : Cluster, ComputeAudit, ComputeFeature, ComputeMode, ComputeSpec, Database, PgIdent,
52 : RemoteExtSpec, Role,
53 : };
54 : use nix::sys::signal::{Signal, kill};
55 : use pageserver_api::shard::ShardStripeSize;
56 : use reqwest::header::CONTENT_TYPE;
57 : use safekeeper_api::membership::SafekeeperGeneration;
58 : use serde::{Deserialize, Serialize};
59 : use tracing::debug;
60 : use url::Host;
61 : use utils::id::{NodeId, TenantId, TimelineId};
62 :
63 : use crate::local_env::LocalEnv;
64 : use crate::postgresql_conf::PostgresConf;
65 : use crate::storage_controller::StorageController;
66 :
67 : // contents of a endpoint.json file
68 0 : #[derive(Serialize, Deserialize, PartialEq, Eq, Clone, Debug)]
69 : pub struct EndpointConf {
70 : endpoint_id: String,
71 : tenant_id: TenantId,
72 : timeline_id: TimelineId,
73 : mode: ComputeMode,
74 : pg_port: u16,
75 : external_http_port: u16,
76 : internal_http_port: u16,
77 : pg_version: u32,
78 : skip_pg_catalog_updates: bool,
79 : reconfigure_concurrency: usize,
80 : drop_subscriptions_before_start: bool,
81 : features: Vec<ComputeFeature>,
82 : cluster: Option<Cluster>,
83 : }
84 :
85 : //
86 : // ComputeControlPlane
87 : //
88 : pub struct ComputeControlPlane {
89 : base_port: u16,
90 :
91 : // endpoint ID is the key
92 : pub endpoints: BTreeMap<String, Arc<Endpoint>>,
93 :
94 : env: LocalEnv,
95 : }
96 :
97 : impl ComputeControlPlane {
98 : // Load current endpoints from the endpoints/ subdirectories
99 0 : pub fn load(env: LocalEnv) -> Result<ComputeControlPlane> {
100 0 : let mut endpoints = BTreeMap::default();
101 0 : for endpoint_dir in std::fs::read_dir(env.endpoints_path())
102 0 : .with_context(|| format!("failed to list {}", env.endpoints_path().display()))?
103 : {
104 0 : let ep_res = Endpoint::from_dir_entry(endpoint_dir?, &env);
105 0 : let ep = match ep_res {
106 0 : Ok(ep) => ep,
107 0 : Err(e) => match e.downcast::<std::io::Error>() {
108 0 : Ok(e) => {
109 0 : // A parallel task could delete an endpoint while we have just scanned the directory
110 0 : if e.kind() == std::io::ErrorKind::NotFound {
111 0 : continue;
112 : } else {
113 0 : Err(e)?
114 : }
115 : }
116 0 : Err(e) => Err(e)?,
117 : },
118 : };
119 0 : endpoints.insert(ep.endpoint_id.clone(), Arc::new(ep));
120 : }
121 :
122 0 : Ok(ComputeControlPlane {
123 0 : base_port: 55431,
124 0 : endpoints,
125 0 : env,
126 0 : })
127 0 : }
128 :
129 0 : fn get_port(&mut self) -> u16 {
130 0 : 1 + self
131 0 : .endpoints
132 0 : .values()
133 0 : .map(|ep| std::cmp::max(ep.pg_address.port(), ep.external_http_address.port()))
134 0 : .max()
135 0 : .unwrap_or(self.base_port)
136 0 : }
137 :
138 : #[allow(clippy::too_many_arguments)]
139 0 : pub fn new_endpoint(
140 0 : &mut self,
141 0 : endpoint_id: &str,
142 0 : tenant_id: TenantId,
143 0 : timeline_id: TimelineId,
144 0 : pg_port: Option<u16>,
145 0 : external_http_port: Option<u16>,
146 0 : internal_http_port: Option<u16>,
147 0 : pg_version: u32,
148 0 : mode: ComputeMode,
149 0 : skip_pg_catalog_updates: bool,
150 0 : drop_subscriptions_before_start: bool,
151 0 : ) -> Result<Arc<Endpoint>> {
152 0 : let pg_port = pg_port.unwrap_or_else(|| self.get_port());
153 0 : let external_http_port = external_http_port.unwrap_or_else(|| self.get_port() + 1);
154 0 : let internal_http_port = internal_http_port.unwrap_or_else(|| external_http_port + 1);
155 0 : let ep = Arc::new(Endpoint {
156 0 : endpoint_id: endpoint_id.to_owned(),
157 0 : pg_address: SocketAddr::new(IpAddr::from(Ipv4Addr::LOCALHOST), pg_port),
158 0 : external_http_address: SocketAddr::new(
159 0 : IpAddr::from(Ipv4Addr::UNSPECIFIED),
160 0 : external_http_port,
161 0 : ),
162 0 : internal_http_address: SocketAddr::new(
163 0 : IpAddr::from(Ipv4Addr::LOCALHOST),
164 0 : internal_http_port,
165 0 : ),
166 0 : env: self.env.clone(),
167 0 : timeline_id,
168 0 : mode,
169 0 : tenant_id,
170 0 : pg_version,
171 0 : // We don't setup roles and databases in the spec locally, so we don't need to
172 0 : // do catalog updates. Catalog updates also include check availability
173 0 : // data creation. Yet, we have tests that check that size and db dump
174 0 : // before and after start are the same. So, skip catalog updates,
175 0 : // with this we basically test a case of waking up an idle compute, where
176 0 : // we also skip catalog updates in the cloud.
177 0 : skip_pg_catalog_updates,
178 0 : drop_subscriptions_before_start,
179 0 : reconfigure_concurrency: 1,
180 0 : features: vec![],
181 0 : cluster: None,
182 0 : });
183 0 :
184 0 : ep.create_endpoint_dir()?;
185 : std::fs::write(
186 0 : ep.endpoint_path().join("endpoint.json"),
187 0 : serde_json::to_string_pretty(&EndpointConf {
188 0 : endpoint_id: endpoint_id.to_string(),
189 0 : tenant_id,
190 0 : timeline_id,
191 0 : mode,
192 0 : external_http_port,
193 0 : internal_http_port,
194 0 : pg_port,
195 0 : pg_version,
196 0 : skip_pg_catalog_updates,
197 0 : drop_subscriptions_before_start,
198 0 : reconfigure_concurrency: 1,
199 0 : features: vec![],
200 0 : cluster: None,
201 0 : })?,
202 0 : )?;
203 : std::fs::write(
204 0 : ep.endpoint_path().join("postgresql.conf"),
205 0 : ep.setup_pg_conf()?.to_string(),
206 0 : )?;
207 :
208 0 : self.endpoints
209 0 : .insert(ep.endpoint_id.clone(), Arc::clone(&ep));
210 0 :
211 0 : Ok(ep)
212 0 : }
213 :
214 0 : pub fn check_conflicting_endpoints(
215 0 : &self,
216 0 : mode: ComputeMode,
217 0 : tenant_id: TenantId,
218 0 : timeline_id: TimelineId,
219 0 : ) -> Result<()> {
220 0 : if matches!(mode, ComputeMode::Primary) {
221 : // this check is not complete, as you could have a concurrent attempt at
222 : // creating another primary, both reading the state before checking it here,
223 : // but it's better than nothing.
224 0 : let mut duplicates = self.endpoints.iter().filter(|(_k, v)| {
225 0 : v.tenant_id == tenant_id
226 0 : && v.timeline_id == timeline_id
227 0 : && v.mode == mode
228 0 : && v.status() != EndpointStatus::Stopped
229 0 : });
230 :
231 0 : if let Some((key, _)) = duplicates.next() {
232 0 : bail!(
233 0 : "attempting to create a duplicate primary endpoint on tenant {tenant_id}, timeline {timeline_id}: endpoint {key:?} exists already. please don't do this, it is not supported."
234 0 : );
235 0 : }
236 0 : }
237 0 : Ok(())
238 0 : }
239 : }
240 :
241 : ///////////////////////////////////////////////////////////////////////////////
242 :
243 : #[derive(Debug)]
244 : pub struct Endpoint {
245 : /// used as the directory name
246 : endpoint_id: String,
247 : pub tenant_id: TenantId,
248 : pub timeline_id: TimelineId,
249 : pub mode: ComputeMode,
250 :
251 : // port and address of the Postgres server and `compute_ctl`'s HTTP APIs
252 : pub pg_address: SocketAddr,
253 : pub external_http_address: SocketAddr,
254 : pub internal_http_address: SocketAddr,
255 :
256 : // postgres major version in the format: 14, 15, etc.
257 : pg_version: u32,
258 :
259 : // These are not part of the endpoint as such, but the environment
260 : // the endpoint runs in.
261 : pub env: LocalEnv,
262 :
263 : // Optimizations
264 : skip_pg_catalog_updates: bool,
265 :
266 : drop_subscriptions_before_start: bool,
267 : reconfigure_concurrency: usize,
268 : // Feature flags
269 : features: Vec<ComputeFeature>,
270 : // Cluster settings
271 : cluster: Option<Cluster>,
272 : }
273 :
274 : #[derive(PartialEq, Eq)]
275 : pub enum EndpointStatus {
276 : Running,
277 : Stopped,
278 : Crashed,
279 : RunningNoPidfile,
280 : }
281 :
282 : impl std::fmt::Display for EndpointStatus {
283 0 : fn fmt(&self, writer: &mut std::fmt::Formatter) -> std::fmt::Result {
284 0 : let s = match self {
285 0 : Self::Running => "running",
286 0 : Self::Stopped => "stopped",
287 0 : Self::Crashed => "crashed",
288 0 : Self::RunningNoPidfile => "running, no pidfile",
289 : };
290 0 : write!(writer, "{}", s)
291 0 : }
292 : }
293 :
294 : impl Endpoint {
295 0 : fn from_dir_entry(entry: std::fs::DirEntry, env: &LocalEnv) -> Result<Endpoint> {
296 0 : if !entry.file_type()?.is_dir() {
297 0 : anyhow::bail!(
298 0 : "Endpoint::from_dir_entry failed: '{}' is not a directory",
299 0 : entry.path().display()
300 0 : );
301 0 : }
302 0 :
303 0 : // parse data directory name
304 0 : let fname = entry.file_name();
305 0 : let endpoint_id = fname.to_str().unwrap().to_string();
306 :
307 : // Read the endpoint.json file
308 0 : let conf: EndpointConf =
309 0 : serde_json::from_slice(&std::fs::read(entry.path().join("endpoint.json"))?)?;
310 :
311 0 : debug!("serialized endpoint conf: {:?}", conf);
312 :
313 0 : Ok(Endpoint {
314 0 : pg_address: SocketAddr::new(IpAddr::from(Ipv4Addr::LOCALHOST), conf.pg_port),
315 0 : external_http_address: SocketAddr::new(
316 0 : IpAddr::from(Ipv4Addr::UNSPECIFIED),
317 0 : conf.external_http_port,
318 0 : ),
319 0 : internal_http_address: SocketAddr::new(
320 0 : IpAddr::from(Ipv4Addr::LOCALHOST),
321 0 : conf.internal_http_port,
322 0 : ),
323 0 : endpoint_id,
324 0 : env: env.clone(),
325 0 : timeline_id: conf.timeline_id,
326 0 : mode: conf.mode,
327 0 : tenant_id: conf.tenant_id,
328 0 : pg_version: conf.pg_version,
329 0 : skip_pg_catalog_updates: conf.skip_pg_catalog_updates,
330 0 : reconfigure_concurrency: conf.reconfigure_concurrency,
331 0 : drop_subscriptions_before_start: conf.drop_subscriptions_before_start,
332 0 : features: conf.features,
333 0 : cluster: conf.cluster,
334 0 : })
335 0 : }
336 :
337 0 : fn create_endpoint_dir(&self) -> Result<()> {
338 0 : std::fs::create_dir_all(self.endpoint_path()).with_context(|| {
339 0 : format!(
340 0 : "could not create endpoint directory {}",
341 0 : self.endpoint_path().display()
342 0 : )
343 0 : })
344 0 : }
345 :
346 : // Generate postgresql.conf with default configuration
347 0 : fn setup_pg_conf(&self) -> Result<PostgresConf> {
348 0 : let mut conf = PostgresConf::new();
349 0 : conf.append("max_wal_senders", "10");
350 0 : conf.append("wal_log_hints", "off");
351 0 : conf.append("max_replication_slots", "10");
352 0 : conf.append("hot_standby", "on");
353 0 : // Set to 1MB to both exercise getPage requests/LFC, and still have enough room for
354 0 : // Postgres to operate. Everything smaller might be not enough for Postgres under load,
355 0 : // and can cause errors like 'no unpinned buffers available', see
356 0 : // <https://github.com/neondatabase/neon/issues/9956>
357 0 : conf.append("shared_buffers", "1MB");
358 0 : // Postgres defaults to effective_io_concurrency=1, which does not exercise the pageserver's
359 0 : // batching logic. Set this to 2 so that we exercise the code a bit without letting
360 0 : // individual tests do a lot of concurrent work on underpowered test machines
361 0 : conf.append("effective_io_concurrency", "2");
362 0 : conf.append("fsync", "off");
363 0 : conf.append("max_connections", "100");
364 0 : conf.append("wal_level", "logical");
365 0 : // wal_sender_timeout is the maximum time to wait for WAL replication.
366 0 : // It also defines how often the walreciever will send a feedback message to the wal sender.
367 0 : conf.append("wal_sender_timeout", "5s");
368 0 : conf.append("listen_addresses", &self.pg_address.ip().to_string());
369 0 : conf.append("port", &self.pg_address.port().to_string());
370 0 : conf.append("wal_keep_size", "0");
371 0 : // walproposer panics when basebackup is invalid, it is pointless to restart in this case.
372 0 : conf.append("restart_after_crash", "off");
373 0 :
374 0 : // Load the 'neon' extension
375 0 : conf.append("shared_preload_libraries", "neon");
376 0 :
377 0 : conf.append_line("");
378 0 : // Replication-related configurations, such as WAL sending
379 0 : match &self.mode {
380 : ComputeMode::Primary => {
381 : // Configure backpressure
382 : // - Replication write lag depends on how fast the walreceiver can process incoming WAL.
383 : // This lag determines latency of get_page_at_lsn. Speed of applying WAL is about 10MB/sec,
384 : // so to avoid expiration of 1 minute timeout, this lag should not be larger than 600MB.
385 : // Actually latency should be much smaller (better if < 1sec). But we assume that recently
386 : // updates pages are not requested from pageserver.
387 : // - Replication flush lag depends on speed of persisting data by checkpointer (creation of
388 : // delta/image layers) and advancing disk_consistent_lsn. Safekeepers are able to
389 : // remove/archive WAL only beyond disk_consistent_lsn. Too large a lag can cause long
390 : // recovery time (in case of pageserver crash) and disk space overflow at safekeepers.
391 : // - Replication apply lag depends on speed of uploading changes to S3 by uploader thread.
392 : // To be able to restore database in case of pageserver node crash, safekeeper should not
393 : // remove WAL beyond this point. Too large lag can cause space exhaustion in safekeepers
394 : // (if they are not able to upload WAL to S3).
395 0 : conf.append("max_replication_write_lag", "15MB");
396 0 : conf.append("max_replication_flush_lag", "10GB");
397 0 :
398 0 : if !self.env.safekeepers.is_empty() {
399 0 : // Configure Postgres to connect to the safekeepers
400 0 : conf.append("synchronous_standby_names", "walproposer");
401 0 :
402 0 : let safekeepers = self
403 0 : .env
404 0 : .safekeepers
405 0 : .iter()
406 0 : .map(|sk| format!("localhost:{}", sk.get_compute_port()))
407 0 : .collect::<Vec<String>>()
408 0 : .join(",");
409 0 : conf.append("neon.safekeepers", &safekeepers);
410 0 : } else {
411 0 : // We only use setup without safekeepers for tests,
412 0 : // and don't care about data durability on pageserver,
413 0 : // so set more relaxed synchronous_commit.
414 0 : conf.append("synchronous_commit", "remote_write");
415 0 :
416 0 : // Configure the node to stream WAL directly to the pageserver
417 0 : // This isn't really a supported configuration, but can be useful for
418 0 : // testing.
419 0 : conf.append("synchronous_standby_names", "pageserver");
420 0 : }
421 : }
422 0 : ComputeMode::Static(lsn) => {
423 0 : conf.append("recovery_target_lsn", &lsn.to_string());
424 0 : }
425 : ComputeMode::Replica => {
426 0 : assert!(!self.env.safekeepers.is_empty());
427 :
428 : // TODO: use future host field from safekeeper spec
429 : // Pass the list of safekeepers to the replica so that it can connect to any of them,
430 : // whichever is available.
431 0 : let sk_ports = self
432 0 : .env
433 0 : .safekeepers
434 0 : .iter()
435 0 : .map(|x| x.get_compute_port().to_string())
436 0 : .collect::<Vec<_>>()
437 0 : .join(",");
438 0 : let sk_hosts = vec!["localhost"; self.env.safekeepers.len()].join(",");
439 0 :
440 0 : let connstr = format!(
441 0 : "host={} port={} options='-c timeline_id={} tenant_id={}' application_name=replica replication=true",
442 0 : sk_hosts,
443 0 : sk_ports,
444 0 : &self.timeline_id.to_string(),
445 0 : &self.tenant_id.to_string(),
446 0 : );
447 0 :
448 0 : let slot_name = format!("repl_{}_", self.timeline_id);
449 0 : conf.append("primary_conninfo", connstr.as_str());
450 0 : conf.append("primary_slot_name", slot_name.as_str());
451 0 : conf.append("hot_standby", "on");
452 0 : // prefetching of blocks referenced in WAL doesn't make sense for us
453 0 : // Neon hot standby ignores pages that are not in the shared_buffers
454 0 : if self.pg_version >= 15 {
455 0 : conf.append("recovery_prefetch", "off");
456 0 : }
457 : }
458 : }
459 :
460 0 : Ok(conf)
461 0 : }
462 :
463 0 : pub fn endpoint_path(&self) -> PathBuf {
464 0 : self.env.endpoints_path().join(&self.endpoint_id)
465 0 : }
466 :
467 0 : pub fn pgdata(&self) -> PathBuf {
468 0 : self.endpoint_path().join("pgdata")
469 0 : }
470 :
471 0 : pub fn status(&self) -> EndpointStatus {
472 0 : let timeout = Duration::from_millis(300);
473 0 : let has_pidfile = self.pgdata().join("postmaster.pid").exists();
474 0 : let can_connect = TcpStream::connect_timeout(&self.pg_address, timeout).is_ok();
475 0 :
476 0 : match (has_pidfile, can_connect) {
477 0 : (true, true) => EndpointStatus::Running,
478 0 : (false, false) => EndpointStatus::Stopped,
479 0 : (true, false) => EndpointStatus::Crashed,
480 0 : (false, true) => EndpointStatus::RunningNoPidfile,
481 : }
482 0 : }
483 :
484 0 : fn pg_ctl(&self, args: &[&str], auth_token: &Option<String>) -> Result<()> {
485 0 : let pg_ctl_path = self.env.pg_bin_dir(self.pg_version)?.join("pg_ctl");
486 0 : let mut cmd = Command::new(&pg_ctl_path);
487 0 : cmd.args(
488 0 : [
489 0 : &[
490 0 : "-D",
491 0 : self.pgdata().to_str().unwrap(),
492 0 : "-w", //wait till pg_ctl actually does what was asked
493 0 : ],
494 0 : args,
495 0 : ]
496 0 : .concat(),
497 0 : )
498 0 : .env_clear()
499 0 : .env(
500 0 : "LD_LIBRARY_PATH",
501 0 : self.env.pg_lib_dir(self.pg_version)?.to_str().unwrap(),
502 0 : )
503 0 : .env(
504 0 : "DYLD_LIBRARY_PATH",
505 0 : self.env.pg_lib_dir(self.pg_version)?.to_str().unwrap(),
506 : );
507 :
508 : // Pass authentication token used for the connections to pageserver and safekeepers
509 0 : if let Some(token) = auth_token {
510 0 : cmd.env("NEON_AUTH_TOKEN", token);
511 0 : }
512 :
513 0 : let pg_ctl = cmd
514 0 : .output()
515 0 : .context(format!("{} failed", pg_ctl_path.display()))?;
516 0 : if !pg_ctl.status.success() {
517 0 : anyhow::bail!(
518 0 : "pg_ctl failed, exit code: {}, stdout: {}, stderr: {}",
519 0 : pg_ctl.status,
520 0 : String::from_utf8_lossy(&pg_ctl.stdout),
521 0 : String::from_utf8_lossy(&pg_ctl.stderr),
522 0 : );
523 0 : }
524 0 :
525 0 : Ok(())
526 0 : }
527 :
528 0 : fn wait_for_compute_ctl_to_exit(&self, send_sigterm: bool) -> Result<()> {
529 0 : // TODO use background_process::stop_process instead: https://github.com/neondatabase/neon/pull/6482
530 0 : let pidfile_path = self.endpoint_path().join("compute_ctl.pid");
531 0 : let pid: u32 = std::fs::read_to_string(pidfile_path)?.parse()?;
532 0 : let pid = nix::unistd::Pid::from_raw(pid as i32);
533 0 : if send_sigterm {
534 0 : kill(pid, Signal::SIGTERM).ok();
535 0 : }
536 0 : crate::background_process::wait_until_stopped("compute_ctl", pid)?;
537 0 : Ok(())
538 0 : }
539 :
540 0 : fn read_postgresql_conf(&self) -> Result<String> {
541 0 : // Slurp the endpoints/<endpoint id>/postgresql.conf file into
542 0 : // memory. We will include it in the spec file that we pass to
543 0 : // `compute_ctl`, and `compute_ctl` will write it to the postgresql.conf
544 0 : // in the data directory.
545 0 : let postgresql_conf_path = self.endpoint_path().join("postgresql.conf");
546 0 : match std::fs::read(&postgresql_conf_path) {
547 0 : Ok(content) => Ok(String::from_utf8(content)?),
548 0 : Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok("".to_string()),
549 0 : Err(e) => Err(anyhow::Error::new(e).context(format!(
550 0 : "failed to read config file in {}",
551 0 : postgresql_conf_path.to_str().unwrap()
552 0 : ))),
553 : }
554 0 : }
555 :
556 0 : fn build_pageserver_connstr(pageservers: &[(Host, u16)]) -> String {
557 0 : pageservers
558 0 : .iter()
559 0 : .map(|(host, port)| format!("postgresql://no_user@{host}:{port}"))
560 0 : .collect::<Vec<_>>()
561 0 : .join(",")
562 0 : }
563 :
564 : /// Map safekeepers ids to the actual connection strings.
565 0 : fn build_safekeepers_connstrs(&self, sk_ids: Vec<NodeId>) -> Result<Vec<String>> {
566 0 : let mut safekeeper_connstrings = Vec::new();
567 0 : if self.mode == ComputeMode::Primary {
568 0 : for sk_id in sk_ids {
569 0 : let sk = self
570 0 : .env
571 0 : .safekeepers
572 0 : .iter()
573 0 : .find(|node| node.id == sk_id)
574 0 : .ok_or_else(|| anyhow!("safekeeper {sk_id} does not exist"))?;
575 0 : safekeeper_connstrings.push(format!("127.0.0.1:{}", sk.get_compute_port()));
576 : }
577 0 : }
578 0 : Ok(safekeeper_connstrings)
579 0 : }
580 :
581 : #[allow(clippy::too_many_arguments)]
582 0 : pub async fn start(
583 0 : &self,
584 0 : auth_token: &Option<String>,
585 0 : safekeepers_generation: Option<SafekeeperGeneration>,
586 0 : safekeepers: Vec<NodeId>,
587 0 : pageservers: Vec<(Host, u16)>,
588 0 : remote_ext_config: Option<&String>,
589 0 : shard_stripe_size: usize,
590 0 : create_test_user: bool,
591 0 : start_timeout: Duration,
592 0 : ) -> Result<()> {
593 0 : if self.status() == EndpointStatus::Running {
594 0 : anyhow::bail!("The endpoint is already running");
595 0 : }
596 :
597 0 : let postgresql_conf = self.read_postgresql_conf()?;
598 :
599 : // We always start the compute node from scratch, so if the Postgres
600 : // data dir exists from a previous launch, remove it first.
601 0 : if self.pgdata().exists() {
602 0 : std::fs::remove_dir_all(self.pgdata())?;
603 0 : }
604 :
605 0 : let pageserver_connstring = Self::build_pageserver_connstr(&pageservers);
606 0 : assert!(!pageserver_connstring.is_empty());
607 :
608 0 : let safekeeper_connstrings = self.build_safekeepers_connstrs(safekeepers)?;
609 :
610 : // check for file remote_extensions_spec.json
611 : // if it is present, read it and pass to compute_ctl
612 0 : let remote_extensions_spec_path = self.endpoint_path().join("remote_extensions_spec.json");
613 0 : let remote_extensions_spec = std::fs::File::open(remote_extensions_spec_path);
614 : let remote_extensions: Option<RemoteExtSpec>;
615 :
616 0 : if let Ok(spec_file) = remote_extensions_spec {
617 0 : remote_extensions = serde_json::from_reader(spec_file).ok();
618 0 : } else {
619 0 : remote_extensions = None;
620 0 : };
621 :
622 : // Create spec file
623 0 : let mut spec = ComputeSpec {
624 0 : skip_pg_catalog_updates: self.skip_pg_catalog_updates,
625 0 : format_version: 1.0,
626 0 : operation_uuid: None,
627 0 : features: self.features.clone(),
628 0 : swap_size_bytes: None,
629 0 : disk_quota_bytes: None,
630 0 : disable_lfc_resizing: None,
631 0 : cluster: Cluster {
632 0 : cluster_id: None, // project ID: not used
633 0 : name: None, // project name: not used
634 0 : state: None,
635 0 : roles: if create_test_user {
636 0 : vec![Role {
637 0 : name: PgIdent::from_str("test").unwrap(),
638 0 : encrypted_password: None,
639 0 : options: None,
640 0 : }]
641 : } else {
642 0 : Vec::new()
643 : },
644 0 : databases: if create_test_user {
645 0 : vec![Database {
646 0 : name: PgIdent::from_str("neondb").unwrap(),
647 0 : owner: PgIdent::from_str("test").unwrap(),
648 0 : options: None,
649 0 : restrict_conn: false,
650 0 : invalid: false,
651 0 : }]
652 : } else {
653 0 : Vec::new()
654 : },
655 0 : settings: None,
656 0 : postgresql_conf: Some(postgresql_conf.clone()),
657 0 : },
658 0 : delta_operations: None,
659 0 : tenant_id: Some(self.tenant_id),
660 0 : timeline_id: Some(self.timeline_id),
661 0 : mode: self.mode,
662 0 : pageserver_connstring: Some(pageserver_connstring),
663 0 : safekeepers_generation: safekeepers_generation.map(|g| g.into_inner()),
664 0 : safekeeper_connstrings,
665 0 : storage_auth_token: auth_token.clone(),
666 0 : remote_extensions,
667 0 : pgbouncer_settings: None,
668 0 : shard_stripe_size: Some(shard_stripe_size),
669 0 : local_proxy_config: None,
670 0 : reconfigure_concurrency: self.reconfigure_concurrency,
671 0 : drop_subscriptions_before_start: self.drop_subscriptions_before_start,
672 0 : audit_log_level: ComputeAudit::Disabled,
673 0 : };
674 0 :
675 0 : // this strange code is needed to support respec() in tests
676 0 : if self.cluster.is_some() {
677 0 : debug!("Cluster is already set in the endpoint spec, using it");
678 0 : spec.cluster = self.cluster.clone().unwrap();
679 0 :
680 0 : debug!("spec.cluster {:?}", spec.cluster);
681 :
682 : // fill missing fields again
683 0 : if create_test_user {
684 0 : spec.cluster.roles.push(Role {
685 0 : name: PgIdent::from_str("test").unwrap(),
686 0 : encrypted_password: None,
687 0 : options: None,
688 0 : });
689 0 : spec.cluster.databases.push(Database {
690 0 : name: PgIdent::from_str("neondb").unwrap(),
691 0 : owner: PgIdent::from_str("test").unwrap(),
692 0 : options: None,
693 0 : restrict_conn: false,
694 0 : invalid: false,
695 0 : });
696 0 : }
697 0 : spec.cluster.postgresql_conf = Some(postgresql_conf);
698 0 : }
699 :
700 0 : let spec_path = self.endpoint_path().join("spec.json");
701 0 : std::fs::write(spec_path, serde_json::to_string_pretty(&spec)?)?;
702 :
703 : // Open log file. We'll redirect the stdout and stderr of `compute_ctl` to it.
704 0 : let logfile = std::fs::OpenOptions::new()
705 0 : .create(true)
706 0 : .append(true)
707 0 : .open(self.endpoint_path().join("compute.log"))?;
708 :
709 : // Launch compute_ctl
710 0 : let conn_str = self.connstr("cloud_admin", "postgres");
711 0 : println!("Starting postgres node at '{}'", conn_str);
712 0 : if create_test_user {
713 0 : let conn_str = self.connstr("test", "neondb");
714 0 : println!("Also at '{}'", conn_str);
715 0 : }
716 0 : let mut cmd = Command::new(self.env.neon_distrib_dir.join("compute_ctl"));
717 0 : cmd.args([
718 0 : "--external-http-port",
719 0 : &self.external_http_address.port().to_string(),
720 0 : ])
721 0 : .args([
722 0 : "--internal-http-port",
723 0 : &self.internal_http_address.port().to_string(),
724 0 : ])
725 0 : .args(["--pgdata", self.pgdata().to_str().unwrap()])
726 0 : .args(["--connstr", &conn_str])
727 0 : .args([
728 0 : "--spec-path",
729 0 : self.endpoint_path().join("spec.json").to_str().unwrap(),
730 0 : ])
731 0 : .args([
732 0 : "--pgbin",
733 0 : self.env
734 0 : .pg_bin_dir(self.pg_version)?
735 0 : .join("postgres")
736 0 : .to_str()
737 0 : .unwrap(),
738 0 : ])
739 0 : // TODO: It would be nice if we generated compute IDs with the same
740 0 : // algorithm as the real control plane.
741 0 : .args([
742 0 : "--compute-id",
743 0 : &format!(
744 0 : "compute-{}",
745 0 : SystemTime::now()
746 0 : .duration_since(UNIX_EPOCH)
747 0 : .unwrap()
748 0 : .as_secs()
749 0 : ),
750 0 : ])
751 0 : .stdin(std::process::Stdio::null())
752 0 : .stderr(logfile.try_clone()?)
753 0 : .stdout(logfile);
754 :
755 0 : if let Some(remote_ext_config) = remote_ext_config {
756 0 : cmd.args(["--remote-ext-config", remote_ext_config]);
757 0 : }
758 :
759 0 : let child = cmd.spawn()?;
760 : // set up a scopeguard to kill & wait for the child in case we panic or bail below
761 0 : let child = scopeguard::guard(child, |mut child| {
762 0 : println!("SIGKILL & wait the started process");
763 0 : (|| {
764 0 : // TODO: use another signal that can be caught by the child so it can clean up any children it spawned
765 0 : child.kill().context("SIGKILL child")?;
766 0 : child.wait().context("wait() for child process")?;
767 0 : anyhow::Ok(())
768 0 : })()
769 0 : .with_context(|| format!("scopeguard kill&wait child {child:?}"))
770 0 : .unwrap();
771 0 : });
772 0 :
773 0 : // Write down the pid so we can wait for it when we want to stop
774 0 : // TODO use background_process::start_process instead: https://github.com/neondatabase/neon/pull/6482
775 0 : let pid = child.id();
776 0 : let pidfile_path = self.endpoint_path().join("compute_ctl.pid");
777 0 : std::fs::write(pidfile_path, pid.to_string())?;
778 :
779 : // Wait for it to start
780 : const ATTEMPT_INTERVAL: Duration = Duration::from_millis(100);
781 0 : let start_at = Instant::now();
782 : loop {
783 0 : match self.get_status().await {
784 0 : Ok(state) => {
785 0 : match state.status {
786 : ComputeStatus::Init => {
787 0 : if Instant::now().duration_since(start_at) > start_timeout {
788 0 : bail!(
789 0 : "compute startup timed out {:?}; still in Init state",
790 0 : start_timeout
791 0 : );
792 0 : }
793 : // keep retrying
794 : }
795 : ComputeStatus::Running => {
796 : // All good!
797 0 : break;
798 : }
799 : ComputeStatus::Failed => {
800 0 : bail!(
801 0 : "compute startup failed: {}",
802 0 : state
803 0 : .error
804 0 : .as_deref()
805 0 : .unwrap_or("<no error from compute_ctl>")
806 0 : );
807 : }
808 : ComputeStatus::Empty
809 : | ComputeStatus::ConfigurationPending
810 : | ComputeStatus::Configuration
811 : | ComputeStatus::TerminationPending
812 : | ComputeStatus::Terminated => {
813 0 : bail!("unexpected compute status: {:?}", state.status)
814 : }
815 : }
816 : }
817 0 : Err(e) => {
818 0 : if Instant::now().duration_since(start_at) > start_timeout {
819 0 : return Err(e).context(format!(
820 0 : "timed out {:?} waiting to connect to compute_ctl HTTP",
821 0 : start_timeout,
822 0 : ));
823 0 : }
824 : }
825 : }
826 0 : tokio::time::sleep(ATTEMPT_INTERVAL).await;
827 : }
828 :
829 : // disarm the scopeguard, let the child outlive this function (and neon_local invoction)
830 0 : drop(scopeguard::ScopeGuard::into_inner(child));
831 0 :
832 0 : Ok(())
833 0 : }
834 :
835 : // Call the /status HTTP API
836 0 : pub async fn get_status(&self) -> Result<ComputeStatusResponse> {
837 0 : let client = reqwest::Client::new();
838 :
839 0 : let response = client
840 0 : .request(
841 0 : reqwest::Method::GET,
842 0 : format!(
843 0 : "http://{}:{}/status",
844 0 : self.external_http_address.ip(),
845 0 : self.external_http_address.port()
846 0 : ),
847 0 : )
848 0 : .send()
849 0 : .await?;
850 :
851 : // Interpret the response
852 0 : let status = response.status();
853 0 : if !(status.is_client_error() || status.is_server_error()) {
854 0 : Ok(response.json().await?)
855 : } else {
856 : // reqwest does not export its error construction utility functions, so let's craft the message ourselves
857 0 : let url = response.url().to_owned();
858 0 : let msg = match response.text().await {
859 0 : Ok(err_body) => format!("Error: {}", err_body),
860 0 : Err(_) => format!("Http error ({}) at {}.", status.as_u16(), url),
861 : };
862 0 : Err(anyhow::anyhow!(msg))
863 : }
864 0 : }
865 :
866 0 : pub async fn reconfigure(
867 0 : &self,
868 0 : mut pageservers: Vec<(Host, u16)>,
869 0 : stripe_size: Option<ShardStripeSize>,
870 0 : safekeepers: Option<Vec<NodeId>>,
871 0 : ) -> Result<()> {
872 0 : let mut spec: ComputeSpec = {
873 0 : let spec_path = self.endpoint_path().join("spec.json");
874 0 : let file = std::fs::File::open(spec_path)?;
875 0 : serde_json::from_reader(file)?
876 : };
877 :
878 0 : let postgresql_conf = self.read_postgresql_conf()?;
879 0 : spec.cluster.postgresql_conf = Some(postgresql_conf);
880 0 :
881 0 : // If we weren't given explicit pageservers, query the storage controller
882 0 : if pageservers.is_empty() {
883 0 : let storage_controller = StorageController::from_env(&self.env);
884 0 : let locate_result = storage_controller.tenant_locate(self.tenant_id).await?;
885 0 : pageservers = locate_result
886 0 : .shards
887 0 : .into_iter()
888 0 : .map(|shard| {
889 0 : (
890 0 : Host::parse(&shard.listen_pg_addr)
891 0 : .expect("Storage controller reported bad hostname"),
892 0 : shard.listen_pg_port,
893 0 : )
894 0 : })
895 0 : .collect::<Vec<_>>();
896 0 : }
897 :
898 0 : let pageserver_connstr = Self::build_pageserver_connstr(&pageservers);
899 0 : assert!(!pageserver_connstr.is_empty());
900 0 : spec.pageserver_connstring = Some(pageserver_connstr);
901 0 : if stripe_size.is_some() {
902 0 : spec.shard_stripe_size = stripe_size.map(|s| s.0 as usize);
903 0 : }
904 :
905 : // If safekeepers are not specified, don't change them.
906 0 : if let Some(safekeepers) = safekeepers {
907 0 : let safekeeper_connstrings = self.build_safekeepers_connstrs(safekeepers)?;
908 0 : spec.safekeeper_connstrings = safekeeper_connstrings;
909 0 : }
910 :
911 0 : let client = reqwest::Client::builder()
912 0 : .timeout(Duration::from_secs(120))
913 0 : .build()
914 0 : .unwrap();
915 0 : let response = client
916 0 : .post(format!(
917 0 : "http://{}:{}/configure",
918 0 : self.external_http_address.ip(),
919 0 : self.external_http_address.port()
920 0 : ))
921 0 : .header(CONTENT_TYPE.as_str(), "application/json")
922 0 : .body(
923 0 : serde_json::to_string(&ConfigurationRequest {
924 0 : spec,
925 0 : compute_ctl_config: ComputeCtlConfig::default(),
926 0 : })
927 0 : .unwrap(),
928 0 : )
929 0 : .send()
930 0 : .await?;
931 :
932 0 : let status = response.status();
933 0 : if !(status.is_client_error() || status.is_server_error()) {
934 0 : Ok(())
935 : } else {
936 0 : let url = response.url().to_owned();
937 0 : let msg = match response.text().await {
938 0 : Ok(err_body) => format!("Error: {}", err_body),
939 0 : Err(_) => format!("Http error ({}) at {}.", status.as_u16(), url),
940 : };
941 0 : Err(anyhow::anyhow!(msg))
942 : }
943 0 : }
944 :
945 0 : pub fn stop(&self, mode: &str, destroy: bool) -> Result<()> {
946 0 : self.pg_ctl(&["-m", mode, "stop"], &None)?;
947 :
948 : // Also wait for the compute_ctl process to die. It might have some
949 : // cleanup work to do after postgres stops, like syncing safekeepers,
950 : // etc.
951 : //
952 : // If destroying or stop mode is immediate, send it SIGTERM before
953 : // waiting. Sometimes we do *not* want this cleanup: tests intentionally
954 : // do stop when majority of safekeepers is down, so sync-safekeepers
955 : // would hang otherwise. This could be a separate flag though.
956 0 : let send_sigterm = destroy || mode == "immediate";
957 0 : self.wait_for_compute_ctl_to_exit(send_sigterm)?;
958 0 : if destroy {
959 0 : println!(
960 0 : "Destroying postgres data directory '{}'",
961 0 : self.pgdata().to_str().unwrap()
962 0 : );
963 0 : std::fs::remove_dir_all(self.endpoint_path())?;
964 0 : }
965 0 : Ok(())
966 0 : }
967 :
968 0 : pub fn connstr(&self, user: &str, db_name: &str) -> String {
969 0 : format!(
970 0 : "postgresql://{}@{}:{}/{}",
971 0 : user,
972 0 : self.pg_address.ip(),
973 0 : self.pg_address.port(),
974 0 : db_name
975 0 : )
976 0 : }
977 : }
|
