diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 36af98f96e..3d6fc75e08 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -2,7 +2,7 @@ name: Create Release Branch on: schedule: - - cron: '0 7 * * 2' + - cron: '0 7 * * 5' workflow_dispatch: jobs: diff --git a/Cargo.lock b/Cargo.lock index f7598a79cf..39c33304ae 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1609,16 +1609,6 @@ dependencies = [ "subtle", ] -[[package]] -name = "ctor" -version = "0.1.26" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6d2301688392eb071b0bf1a37be05c469d3cc4dbbd95df672fe28ab021e6a096" -dependencies = [ - "quote", - "syn 1.0.109", -] - [[package]] name = "ctr" version = "0.6.0" @@ -2714,11 +2704,10 @@ dependencies = [ [[package]] name = "log" -version = "0.4.17" +version = "0.4.20" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "abb12e687cfb44aa40f41fc3978ef76448f9b6038cad6aef4259d3c095a2382e" +checksum = "b5e6163cb8c49088c2c36f57875e58ccd8c87c7427f7fbd50ea6710b2f3f2e8f" dependencies = [ - "cfg-if", "value-bag", ] @@ -6011,13 +6000,9 @@ checksum = "830b7e5d4d90034032940e4ace0d9a9a057e7a45cd94e6c007832e39edb82f6d" [[package]] name = "value-bag" -version = "1.0.0-alpha.9" +version = "1.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2209b78d1249f7e6f3293657c9779fe31ced465df091bbd433a1cf88e916ec55" -dependencies = [ - "ctor", - "version_check", -] +checksum = "4a72e1902dde2bd6441347de2b70b7f5d59bf157c6c62f0c44572607a1d55bbe" [[package]] name = "vcpkg" diff --git a/compute_tools/src/pg_helpers.rs b/compute_tools/src/pg_helpers.rs index b79e516650..8722822f5e 100644 --- a/compute_tools/src/pg_helpers.rs +++ b/compute_tools/src/pg_helpers.rs @@ -193,11 +193,16 @@ impl Escaping for PgIdent { /// Build a list of existing Postgres roles pub fn get_existing_roles(xact: &mut Transaction<'_>) -> Result> { let postgres_roles = xact - .query("SELECT rolname, rolpassword FROM pg_catalog.pg_authid", &[])? + .query( + "SELECT rolname, rolpassword, rolreplication, rolbypassrls FROM pg_catalog.pg_authid", + &[], + )? .iter() .map(|row| Role { name: row.get("rolname"), encrypted_password: row.get("rolpassword"), + replication: Some(row.get("rolreplication")), + bypassrls: Some(row.get("rolbypassrls")), options: None, }) .collect(); diff --git a/compute_tools/src/spec.rs b/compute_tools/src/spec.rs index f7ca2eb33c..6e4d4ccf49 100644 --- a/compute_tools/src/spec.rs +++ b/compute_tools/src/spec.rs @@ -265,6 +265,8 @@ pub fn handle_roles(spec: &ComputeSpec, client: &mut Client) -> Result<()> { let action = if let Some(r) = pg_role { if (r.encrypted_password.is_none() && role.encrypted_password.is_some()) || (r.encrypted_password.is_some() && role.encrypted_password.is_none()) + || !r.bypassrls.unwrap_or(false) + || !r.replication.unwrap_or(false) { RoleAction::Update } else if let Some(pg_pwd) = &r.encrypted_password { @@ -296,7 +298,8 @@ pub fn handle_roles(spec: &ComputeSpec, client: &mut Client) -> Result<()> { match action { RoleAction::None => {} RoleAction::Update => { - let mut query: String = format!("ALTER ROLE {} ", name.pg_quote()); + let mut query: String = + format!("ALTER ROLE {} BYPASSRLS REPLICATION", name.pg_quote()); query.push_str(&role.to_pg_options()); xact.execute(query.as_str(), &[])?; } diff --git a/control_plane/src/attachment_service.rs b/control_plane/src/attachment_service.rs index d7828cdba7..ca632c5eb6 100644 --- a/control_plane/src/attachment_service.rs +++ b/control_plane/src/attachment_service.rs @@ -19,7 +19,7 @@ const COMMAND: &str = "attachment_service"; pub struct AttachHookRequest { #[serde_as(as = "DisplayFromStr")] pub tenant_id: TenantId, - pub pageserver_id: Option, + pub node_id: Option, } #[derive(Serialize, Deserialize)] @@ -85,7 +85,7 @@ impl AttachmentService { .control_plane_api .clone() .unwrap() - .join("attach_hook") + .join("attach-hook") .unwrap(); let client = reqwest::blocking::ClientBuilder::new() .build() @@ -93,7 +93,7 @@ impl AttachmentService { let request = AttachHookRequest { tenant_id, - pageserver_id: Some(pageserver_id), + node_id: Some(pageserver_id), }; let response = client.post(url).json(&request).send()?; diff --git a/control_plane/src/bin/attachment_service.rs b/control_plane/src/bin/attachment_service.rs index 65ed78d20b..0ccd9e75fc 100644 --- a/control_plane/src/bin/attachment_service.rs +++ b/control_plane/src/bin/attachment_service.rs @@ -171,7 +171,7 @@ async fn handle_re_attach(mut req: Request) -> Result, ApiE state.generation += 1; response.tenants.push(ReAttachResponseTenant { id: *t, - generation: state.generation, + gen: state.generation, }); } } @@ -217,14 +217,14 @@ async fn handle_attach_hook(mut req: Request) -> Result, Ap .tenants .entry(attach_req.tenant_id) .or_insert_with(|| TenantState { - pageserver: attach_req.pageserver_id, + pageserver: attach_req.node_id, generation: 0, }); - if attach_req.pageserver_id.is_some() { + if attach_req.node_id.is_some() { tenant_state.generation += 1; } - tenant_state.pageserver = attach_req.pageserver_id; + tenant_state.pageserver = attach_req.node_id; let generation = tenant_state.generation; locked.save().await.map_err(ApiError::InternalServerError)?; @@ -232,7 +232,7 @@ async fn handle_attach_hook(mut req: Request) -> Result, Ap json_response( StatusCode::OK, AttachHookResponse { - gen: attach_req.pageserver_id.map(|_| generation), + gen: attach_req.node_id.map(|_| generation), }, ) } @@ -242,7 +242,7 @@ fn make_router(persistent_state: PersistentState) -> RouterBuilder, + pub replication: Option, + pub bypassrls: Option, pub options: GenericOptions, } diff --git a/libs/metrics/src/lib.rs b/libs/metrics/src/lib.rs index b4026d93e1..0b3b24c18e 100644 --- a/libs/metrics/src/lib.rs +++ b/libs/metrics/src/lib.rs @@ -89,14 +89,14 @@ pub const DISK_WRITE_SECONDS_BUCKETS: &[f64] = &[ 0.000_050, 0.000_100, 0.000_500, 0.001, 0.003, 0.005, 0.01, 0.05, 0.1, 0.3, 0.5, ]; -pub fn set_build_info_metric(revision: &str) { +pub fn set_build_info_metric(revision: &str, build_tag: &str) { let metric = register_int_gauge_vec!( "libmetrics_build_info", "Build/version information", - &["revision"] + &["revision", "build_tag"] ) .expect("Failed to register build info metric"); - metric.with_label_values(&[revision]).set(1); + metric.with_label_values(&[revision, build_tag]).set(1); } // Records I/O stats in a "cross-platform" way. diff --git a/libs/pageserver_api/src/control_api.rs b/libs/pageserver_api/src/control_api.rs index a54fee47a5..3819111a5b 100644 --- a/libs/pageserver_api/src/control_api.rs +++ b/libs/pageserver_api/src/control_api.rs @@ -17,7 +17,7 @@ pub struct ReAttachRequest { pub struct ReAttachResponseTenant { #[serde_as(as = "DisplayFromStr")] pub id: TenantId, - pub generation: u32, + pub gen: u32, } #[derive(Serialize, Deserialize)] diff --git a/libs/postgres_backend/src/lib.rs b/libs/postgres_backend/src/lib.rs index 8ce7f85481..ad3af4e794 100644 --- a/libs/postgres_backend/src/lib.rs +++ b/libs/postgres_backend/src/lib.rs @@ -242,6 +242,7 @@ impl MaybeWriteOnly { } } + /// Cancellation safe as long as the underlying IO is cancellation safe. async fn shutdown(&mut self) -> io::Result<()> { match self { MaybeWriteOnly::Full(framed) => framed.shutdown().await, @@ -393,13 +394,23 @@ impl PostgresBackend { shutdown_watcher: F, ) -> Result<(), QueryError> where - F: Fn() -> S, + F: Fn() -> S + Clone, S: Future, { - let ret = self.run_message_loop(handler, shutdown_watcher).await; - // socket might be already closed, e.g. if previously received error, - // so ignore result. - self.framed.shutdown().await.ok(); + let ret = self + .run_message_loop(handler, shutdown_watcher.clone()) + .await; + + tokio::select! { + _ = shutdown_watcher() => { + // do nothing; we most likely got already stopped by shutdown and will log it next. + } + _ = self.framed.shutdown() => { + // socket might be already closed, e.g. if previously received error, + // so ignore result. + }, + } + match ret { Ok(()) => Ok(()), Err(QueryError::Shutdown) => { diff --git a/libs/postgres_ffi/wal_craft/src/lib.rs b/libs/postgres_ffi/wal_craft/src/lib.rs index 75ffd3f055..281a180e3b 100644 --- a/libs/postgres_ffi/wal_craft/src/lib.rs +++ b/libs/postgres_ffi/wal_craft/src/lib.rs @@ -14,6 +14,7 @@ macro_rules! xlog_utils_test { ($version:ident) => { #[path = "."] mod $version { + #[allow(unused_imports)] pub use postgres_ffi::$version::wal_craft_test_export::*; #[allow(clippy::duplicate_mod)] #[cfg(test)] diff --git a/libs/pq_proto/src/framed.rs b/libs/pq_proto/src/framed.rs index c12898a05c..6e97b8c2a0 100644 --- a/libs/pq_proto/src/framed.rs +++ b/libs/pq_proto/src/framed.rs @@ -214,27 +214,24 @@ where } } +/// Cancellation safe as long as the AsyncWrite is cancellation safe. async fn flush( stream: &mut S, write_buf: &mut BytesMut, ) -> Result<(), io::Error> { while write_buf.has_remaining() { - let bytes_written = stream.write(write_buf.chunk()).await?; + let bytes_written = stream.write_buf(write_buf).await?; if bytes_written == 0 { return Err(io::Error::new( ErrorKind::WriteZero, "failed to write message", )); } - // The advanced part will be garbage collected, likely during shifting - // data left on next attempt to write to buffer when free space is not - // enough. - write_buf.advance(bytes_written); } - write_buf.clear(); stream.flush().await } +/// Cancellation safe as long as the AsyncWrite is cancellation safe. async fn shutdown( stream: &mut S, write_buf: &mut BytesMut, diff --git a/libs/utils/src/lib.rs b/libs/utils/src/lib.rs index d3de960b65..2d86b0e254 100644 --- a/libs/utils/src/lib.rs +++ b/libs/utils/src/lib.rs @@ -132,6 +132,21 @@ macro_rules! project_git_version { }; } +/// This is a shortcut to embed build tag into binaries and avoid copying the same build script to all packages +#[macro_export] +macro_rules! project_build_tag { + ($const_identifier:ident) => { + const $const_identifier: &::core::primitive::str = { + const __ARG: &[&::core::primitive::str; 2] = &match ::core::option_env!("BUILD_TAG") { + ::core::option::Option::Some(x) => ["build_tag-env:", x], + ::core::option::Option::None => ["build_tag:", ""], + }; + + $crate::__const_format::concatcp!(__ARG[0], __ARG[1]) + }; + }; +} + /// Re-export for `project_git_version` macro #[doc(hidden)] pub use const_format as __const_format; diff --git a/pageserver/src/bin/pageserver.rs b/pageserver/src/bin/pageserver.rs index 798b9f258b..df2db25ec6 100644 --- a/pageserver/src/bin/pageserver.rs +++ b/pageserver/src/bin/pageserver.rs @@ -34,11 +34,12 @@ use postgres_backend::AuthType; use utils::logging::TracingErrorLayerEnablement; use utils::signals::ShutdownSignals; use utils::{ - auth::JwtAuth, logging, project_git_version, sentry_init::init_sentry, signals::Signal, - tcp_listener, + auth::JwtAuth, logging, project_build_tag, project_git_version, sentry_init::init_sentry, + signals::Signal, tcp_listener, }; project_git_version!(GIT_VERSION); +project_build_tag!(BUILD_TAG); const PID_FILE_NAME: &str = "pageserver.pid"; @@ -258,11 +259,12 @@ fn start_pageserver( // A changed version string indicates changed software. // A changed launch timestamp indicates a pageserver restart. info!( - "version: {} launch_timestamp: {}", + "version: {} launch_timestamp: {} build_tag: {}", version(), - launch_ts.to_string() + launch_ts.to_string(), + BUILD_TAG, ); - set_build_info_metric(GIT_VERSION); + set_build_info_metric(GIT_VERSION, BUILD_TAG); set_launch_timestamp_metric(launch_ts); pageserver::preinitialize_metrics(); diff --git a/pageserver/src/control_plane_client.rs b/pageserver/src/control_plane_client.rs index d37dc694dd..be4fcab75e 100644 --- a/pageserver/src/control_plane_client.rs +++ b/pageserver/src/control_plane_client.rs @@ -144,7 +144,7 @@ impl ControlPlaneGenerationsApi for ControlPlaneClient { Ok(response .tenants .into_iter() - .map(|t| (t.id, Generation::new(t.generation))) + .map(|t| (t.id, Generation::new(t.gen))) .collect::>()) } diff --git a/pageserver/src/deletion_queue/list_writer.rs b/pageserver/src/deletion_queue/list_writer.rs index 21b8c356cd..6727957b2a 100644 --- a/pageserver/src/deletion_queue/list_writer.rs +++ b/pageserver/src/deletion_queue/list_writer.rs @@ -349,7 +349,7 @@ impl ListWriter { info!("Started deletion frontend worker"); // Synchronous, but we only do it once per process lifetime so it's tolerable - if let Err(e) = create_dir_all(&self.conf.deletion_prefix()) { + if let Err(e) = create_dir_all(self.conf.deletion_prefix()) { tracing::error!( "Failed to create deletion list directory {}, deletions will not be executed ({e})", self.conf.deletion_prefix(), diff --git a/pageserver/src/http/openapi_spec.yml b/pageserver/src/http/openapi_spec.yml index e8421d619d..2d954f6823 100644 --- a/pageserver/src/http/openapi_spec.yml +++ b/pageserver/src/http/openapi_spec.yml @@ -713,6 +713,12 @@ paths: Errors if the tenant is absent on disk, already present in memory or fails to schedule its load. Scheduling a load does not mean that the tenant would load successfully, check tenant status to ensure load correctness. + requestBody: + required: false + content: + application/json: + schema: + $ref: "#/components/schemas/TenantLoadRequest" responses: "202": description: Tenant scheduled to load successfully @@ -1203,6 +1209,15 @@ components: new_tenant_id: type: string format: hex + generation: + type: integer + description: Attachment generation number. + TenantLoadRequest: + type: object + properties: + generation: + type: integer + description: Attachment generation number. TenantAttachRequest: type: object required: diff --git a/pageserver/src/tenant/delete.rs b/pageserver/src/tenant/delete.rs index 3e2d6ca7bb..87b48e4bee 100644 --- a/pageserver/src/tenant/delete.rs +++ b/pageserver/src/tenant/delete.rs @@ -148,7 +148,8 @@ async fn ensure_timelines_dir_empty(timelines_path: &Utf8Path) -> Result<(), Del // Assert timelines dir is empty. if !fs_ext::is_directory_empty(timelines_path).await? { // Display first 10 items in directory - let list = &fs_ext::list_dir(timelines_path).await.context("list_dir")?[..10]; + let list = fs_ext::list_dir(timelines_path).await.context("list_dir")?; + let list = &list.into_iter().take(10).collect::>(); return Err(DeleteTenantError::Other(anyhow::anyhow!( "Timelines directory is not empty after all timelines deletion: {list:?}" ))); diff --git a/pageserver/src/tenant/remote_timeline_client.rs b/pageserver/src/tenant/remote_timeline_client.rs index d91970a1c0..76e75253fb 100644 --- a/pageserver/src/tenant/remote_timeline_client.rs +++ b/pageserver/src/tenant/remote_timeline_client.rs @@ -721,6 +721,17 @@ impl RemoteTimelineClient { }) .collect(); + #[cfg(feature = "testing")] + for (name, gen) in &with_generations { + if let Some(unexpected) = upload_queue.dangling_files.insert(name.to_owned(), *gen) { + if &unexpected == gen { + tracing::error!("{name} was unlinked twice with same generation"); + } else { + tracing::error!("{name} was unlinked twice with different generations {gen:?} and {unexpected:?}"); + } + } + } + // after unlinking files from the upload_queue.latest_files we must always schedule an // index_part update, because that needs to be uploaded before we can actually delete the // files. @@ -754,6 +765,19 @@ impl RemoteTimelineClient { info!("scheduling deletion of layer {}{}", name, gen.get_suffix()); } + #[cfg(feature = "testing")] + for (name, gen) in &with_generations { + match upload_queue.dangling_files.remove(name) { + Some(same) if &same == gen => { /* expected */ } + Some(other) => { + tracing::error!("{name} was unlinked with {other:?} but deleted with {gen:?}"); + } + None => { + tracing::error!("{name} was unlinked but was not dangling"); + } + } + } + // schedule the actual deletions let op = UploadOp::Delete(Delete { layers: with_generations, @@ -779,9 +803,7 @@ impl RemoteTimelineClient { let names = compacted_from.iter().map(|x| x.layer_desc().filename()); - let with_generations = - self.schedule_unlinking_of_layers_from_index_part0(upload_queue, names); - self.schedule_deletion_of_unlinked0(upload_queue, with_generations); + self.schedule_unlinking_of_layers_from_index_part0(upload_queue, names); self.launch_queued_tasks(upload_queue); Ok(()) @@ -1434,6 +1456,8 @@ impl RemoteTimelineClient { num_inprogress_deletions: 0, inprogress_tasks: HashMap::default(), queued_operations: VecDeque::default(), + #[cfg(feature = "testing")] + dangling_files: HashMap::default(), }; let upload_queue = std::mem::replace( diff --git a/pageserver/src/tenant/remote_timeline_client/index.rs b/pageserver/src/tenant/remote_timeline_client/index.rs index 05f9f5dcd2..7cf963ca9d 100644 --- a/pageserver/src/tenant/remote_timeline_client/index.rs +++ b/pageserver/src/tenant/remote_timeline_client/index.rs @@ -98,7 +98,7 @@ impl IndexPart { const LATEST_VERSION: usize = 4; // Versions we may see when reading from a bucket. - pub const KNOWN_VERSIONS: &[usize] = &[1, 2, 3, 4]; + pub const KNOWN_VERSIONS: &'static [usize] = &[1, 2, 3, 4]; pub const FILE_NAME: &'static str = "index_part.json"; diff --git a/pageserver/src/tenant/storage_layer/layer.rs b/pageserver/src/tenant/storage_layer/layer.rs index 759f6d664c..5a26421b5b 100644 --- a/pageserver/src/tenant/storage_layer/layer.rs +++ b/pageserver/src/tenant/storage_layer/layer.rs @@ -47,7 +47,16 @@ pub(crate) struct Layer(Arc); impl std::fmt::Display for Layer { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "{}", self.layer_desc().short_id()) + if matches!(self.0.generation, Generation::Broken) { + write!(f, "{}-broken", self.layer_desc().short_id()) + } else { + write!( + f, + "{}{}", + self.layer_desc().short_id(), + self.0.generation.get_suffix() + ) + } } } @@ -433,7 +442,7 @@ impl Drop for LayerInner { return; } - let span = tracing::info_span!(parent: None, "layer_gc", tenant_id = %self.layer_desc().tenant_id, timeline_id = %self.layer_desc().timeline_id, layer = %self); + let span = tracing::info_span!(parent: None, "layer_gc", tenant_id = %self.layer_desc().tenant_id, timeline_id = %self.layer_desc().timeline_id); let path = std::mem::take(&mut self.path); let file_name = self.layer_desc().filename(); diff --git a/pageserver/src/tenant/tasks.rs b/pageserver/src/tenant/tasks.rs index f419d2e73a..eb77f7c83a 100644 --- a/pageserver/src/tenant/tasks.rs +++ b/pageserver/src/tenant/tasks.rs @@ -162,8 +162,11 @@ async fn compaction_loop(tenant: Arc, cancel: CancellationToken) { // TODO: we shouldn't need to await to find tenant and this could be moved outside of // loop, #3501. There are also additional "allowed_errors" in tests. - if first && random_init_delay(period, &cancel).await.is_err() { - break; + if first { + first = false; + if random_init_delay(period, &cancel).await.is_err() { + break; + } } let started_at = Instant::now(); @@ -193,16 +196,7 @@ async fn compaction_loop(tenant: Arc, cancel: CancellationToken) { } }; - if !first { - // The first iteration is typically much slower, because all tenants compete for the - // compaction sempahore to run, and because of concurrent startup work like initializing - // logical sizes. To avoid routinely spamming warnings, we suppress this log on first iteration. - warn_when_period_overrun( - started_at.elapsed(), - period, - BackgroundLoopKind::Compaction, - ); - } + warn_when_period_overrun(started_at.elapsed(), period, BackgroundLoopKind::Compaction); // Sleep if tokio::time::timeout(sleep_duration, cancel.cancelled()) @@ -211,8 +205,6 @@ async fn compaction_loop(tenant: Arc, cancel: CancellationToken) { { break; } - - first = false; } } .await; @@ -247,8 +239,11 @@ async fn gc_loop(tenant: Arc, cancel: CancellationToken) { let period = tenant.get_gc_period(); - if first && random_init_delay(period, &cancel).await.is_err() { - break; + if first { + first = false; + if random_init_delay(period, &cancel).await.is_err() { + break; + } } let started_at = Instant::now(); @@ -282,12 +277,7 @@ async fn gc_loop(tenant: Arc, cancel: CancellationToken) { } }; - if !first { - // The first iteration is typically much slower, because all tenants compete for the - // compaction sempahore to run, and because of concurrent startup work like initializing - // logical sizes. To avoid routinely spamming warnings, we suppress this log on first iteration. - warn_when_period_overrun(started_at.elapsed(), period, BackgroundLoopKind::Gc); - } + warn_when_period_overrun(started_at.elapsed(), period, BackgroundLoopKind::Gc); // Sleep if tokio::time::timeout(sleep_duration, cancel.cancelled()) @@ -296,8 +286,6 @@ async fn gc_loop(tenant: Arc, cancel: CancellationToken) { { break; } - - first = false; } } .await; @@ -373,7 +361,7 @@ pub(crate) fn warn_when_period_overrun( // humantime does no significant digits clamping whereas Duration's debug is a bit more // intelligent. however it makes sense to keep the "configuration format" for period, even // though there's no way to output the actual config value. - warn!( + info!( ?elapsed, period = %humantime::format_duration(period), ?task, diff --git a/pageserver/src/tenant/timeline.rs b/pageserver/src/tenant/timeline.rs index 278290eef4..4608683c4a 100644 --- a/pageserver/src/tenant/timeline.rs +++ b/pageserver/src/tenant/timeline.rs @@ -1859,19 +1859,17 @@ impl Timeline { "aborted because task_mgr shutdown requested".to_string() }; - loop { - tokio::select! { - res = &mut calculation => { return res } - reason = timeline_state_cancellation => { - debug!(reason = reason, "cancelling calculation"); - cancel.cancel(); - return calculation.await; - } - reason = taskmgr_shutdown_cancellation => { - debug!(reason = reason, "cancelling calculation"); - cancel.cancel(); - return calculation.await; - } + tokio::select! { + res = &mut calculation => { res } + reason = timeline_state_cancellation => { + debug!(reason = reason, "cancelling calculation"); + cancel.cancel(); + calculation.await + } + reason = taskmgr_shutdown_cancellation => { + debug!(reason = reason, "cancelling calculation"); + cancel.cancel(); + calculation.await } } } diff --git a/pageserver/src/tenant/timeline/delete.rs b/pageserver/src/tenant/timeline/delete.rs index 8cead5ebf2..6d30664515 100644 --- a/pageserver/src/tenant/timeline/delete.rs +++ b/pageserver/src/tenant/timeline/delete.rs @@ -38,6 +38,14 @@ async fn stop_tasks(timeline: &Timeline) -> Result<(), DeleteTimelineError> { } debug!("wal receiver shutdown confirmed"); + // Shut down the layer flush task before the remote client, as one depends on the other + task_mgr::shutdown_tasks( + Some(TaskKind::LayerFlushTask), + Some(timeline.tenant_id), + Some(timeline.timeline_id), + ) + .await; + // Prevent new uploads from starting. if let Some(remote_client) = timeline.remote_client.as_ref() { let res = remote_client.stop(); diff --git a/pageserver/src/tenant/timeline/walreceiver/walreceiver_connection.rs b/pageserver/src/tenant/timeline/walreceiver/walreceiver_connection.rs index 0831b9ceda..3e56753ad4 100644 --- a/pageserver/src/tenant/timeline/walreceiver/walreceiver_connection.rs +++ b/pageserver/src/tenant/timeline/walreceiver/walreceiver_connection.rs @@ -122,7 +122,7 @@ pub(super) async fn handle_walreceiver_connection( // Connect to the database in replication mode. info!("connecting to {wal_source_connconf:?}"); - let (mut replication_client, connection) = { + let (replication_client, connection) = { let mut config = wal_source_connconf.to_tokio_postgres_config(); config.application_name("pageserver"); config.replication_mode(tokio_postgres::config::ReplicationMode::Physical); @@ -205,7 +205,7 @@ pub(super) async fn handle_walreceiver_connection( gauge.dec(); } - let identify = identify_system(&mut replication_client).await?; + let identify = identify_system(&replication_client).await?; info!("{identify:?}"); let end_of_wal = Lsn::from(u64::from(identify.xlogpos)); @@ -444,7 +444,7 @@ struct IdentifySystem { struct IdentifyError; /// Run the postgres `IDENTIFY_SYSTEM` command -async fn identify_system(client: &mut Client) -> anyhow::Result { +async fn identify_system(client: &Client) -> anyhow::Result { let query_str = "IDENTIFY_SYSTEM"; let response = client.simple_query(query_str).await?; @@ -459,7 +459,7 @@ async fn identify_system(client: &mut Client) -> anyhow::Result // extract the row contents into an IdentifySystem struct. // written as a closure so I can use ? for Option here. - if let Some(SimpleQueryMessage::Row(first_row)) = response.get(0) { + if let Some(SimpleQueryMessage::Row(first_row)) = response.first() { Ok(IdentifySystem { systemid: get_parse(first_row, 0)?, timeline: get_parse(first_row, 1)?, diff --git a/pageserver/src/tenant/upload_queue.rs b/pageserver/src/tenant/upload_queue.rs index 5959819548..b47878aacc 100644 --- a/pageserver/src/tenant/upload_queue.rs +++ b/pageserver/src/tenant/upload_queue.rs @@ -80,6 +80,14 @@ pub(crate) struct UploadQueueInitialized { /// tasks to finish. For example, metadata upload cannot be performed before all /// preceding layer file uploads have completed. pub(crate) queued_operations: VecDeque, + + /// Files which have been unlinked but not yet had scheduled a deletion for. Only kept around + /// for error logging. + /// + /// Putting this behind a testing feature to catch problems in tests, but assuming we could have a + /// bug causing leaks, then it's better to not leave this enabled for production builds. + #[cfg(feature = "testing")] + pub(crate) dangling_files: HashMap, } impl UploadQueueInitialized { @@ -136,6 +144,8 @@ impl UploadQueue { num_inprogress_deletions: 0, inprogress_tasks: HashMap::new(), queued_operations: VecDeque::new(), + #[cfg(feature = "testing")] + dangling_files: HashMap::new(), }; *self = UploadQueue::Initialized(state); @@ -181,6 +191,8 @@ impl UploadQueue { num_inprogress_deletions: 0, inprogress_tasks: HashMap::new(), queued_operations: VecDeque::new(), + #[cfg(feature = "testing")] + dangling_files: HashMap::new(), }; *self = UploadQueue::Initialized(state); diff --git a/pageserver/src/walingest.rs b/pageserver/src/walingest.rs index fb1dbcd6ba..23367928d3 100644 --- a/pageserver/src/walingest.rs +++ b/pageserver/src/walingest.rs @@ -443,7 +443,7 @@ impl<'a> WalIngest<'a> { &mut self, buf: &mut Bytes, modification: &mut DatadirModification<'_>, - decoded: &mut DecodedWALRecord, + decoded: &DecodedWALRecord, ctx: &RequestContext, ) -> anyhow::Result<()> { // Handle VM bit updates that are implicitly part of heap records. @@ -749,7 +749,7 @@ impl<'a> WalIngest<'a> { &mut self, buf: &mut Bytes, modification: &mut DatadirModification<'_>, - decoded: &mut DecodedWALRecord, + decoded: &DecodedWALRecord, ctx: &RequestContext, ) -> anyhow::Result<()> { // Handle VM bit updates that are implicitly part of heap records. diff --git a/pgxn/neon_walredo/walredoproc.c b/pgxn/neon_walredo/walredoproc.c index e0c6c6e30a..bdc50b0aa9 100644 --- a/pgxn/neon_walredo/walredoproc.c +++ b/pgxn/neon_walredo/walredoproc.c @@ -895,7 +895,12 @@ apply_error_callback(void *arg) StringInfoData buf; initStringInfo(&buf); - xlog_outdesc(&buf, record); +#if PG_VERSION_NUM >= 150000 + if (record->record) +#else + if (record->decoded_record) +#endif + xlog_outdesc(&buf, record); /* translator: %s is a WAL record description */ errcontext("WAL redo at %X/%X for %s", diff --git a/proxy/src/auth/backend.rs b/proxy/src/auth/backend.rs index 850d3a1837..9cf45c0eec 100644 --- a/proxy/src/auth/backend.rs +++ b/proxy/src/auth/backend.rs @@ -5,7 +5,7 @@ mod link; pub use link::LinkAuthError; use tokio_postgres::config::AuthKeys; -use crate::proxy::{handle_try_wake, retry_after}; +use crate::proxy::{handle_try_wake, retry_after, LatencyTimer}; use crate::{ auth::{self, ClientCredentials}, config::AuthenticationConfig, @@ -134,13 +134,14 @@ async fn auth_quirks_creds( client: &mut stream::PqStream, allow_cleartext: bool, config: &'static AuthenticationConfig, + latency_timer: &mut LatencyTimer, ) -> auth::Result> { // If there's no project so far, that entails that client doesn't // support SNI or other means of passing the endpoint (project) name. // We now expect to see a very specific payload in the place of password. if creds.project.is_none() { // Password will be checked by the compute node later. - return hacks::password_hack(creds, client).await; + return hacks::password_hack(creds, client, latency_timer).await; } // Password hack should set the project name. @@ -151,11 +152,11 @@ async fn auth_quirks_creds( // Currently, we use it for websocket connections (latency). if allow_cleartext { // Password will be checked by the compute node later. - return hacks::cleartext_hack(client).await; + return hacks::cleartext_hack(client, latency_timer).await; } // Finally, proceed with the main auth flow (SCRAM-based). - classic::authenticate(api, extra, creds, client, config).await + classic::authenticate(api, extra, creds, client, config, latency_timer).await } /// True to its name, this function encapsulates our current auth trade-offs. @@ -167,8 +168,18 @@ async fn auth_quirks( client: &mut stream::PqStream, allow_cleartext: bool, config: &'static AuthenticationConfig, + latency_timer: &mut LatencyTimer, ) -> auth::Result> { - let auth_stuff = auth_quirks_creds(api, extra, creds, client, allow_cleartext, config).await?; + let auth_stuff = auth_quirks_creds( + api, + extra, + creds, + client, + allow_cleartext, + config, + latency_timer, + ) + .await?; let mut num_retries = 0; let mut node = loop { @@ -233,6 +244,7 @@ impl BackendType<'_, ClientCredentials<'_>> { client: &mut stream::PqStream, allow_cleartext: bool, config: &'static AuthenticationConfig, + latency_timer: &mut LatencyTimer, ) -> auth::Result> { use BackendType::*; @@ -245,7 +257,16 @@ impl BackendType<'_, ClientCredentials<'_>> { ); let api = api.as_ref(); - auth_quirks(api, extra, creds, client, allow_cleartext, config).await? + auth_quirks( + api, + extra, + creds, + client, + allow_cleartext, + config, + latency_timer, + ) + .await? } Postgres(api, creds) => { info!( @@ -255,7 +276,16 @@ impl BackendType<'_, ClientCredentials<'_>> { ); let api = api.as_ref(); - auth_quirks(api, extra, creds, client, allow_cleartext, config).await? + auth_quirks( + api, + extra, + creds, + client, + allow_cleartext, + config, + latency_timer, + ) + .await? } // NOTE: this auth backend doesn't use client credentials. Link(url) => { diff --git a/proxy/src/auth/backend/classic.rs b/proxy/src/auth/backend/classic.rs index e7013a4cac..aee0057606 100644 --- a/proxy/src/auth/backend/classic.rs +++ b/proxy/src/auth/backend/classic.rs @@ -4,6 +4,7 @@ use crate::{ compute, config::AuthenticationConfig, console::{self, AuthInfo, ConsoleReqExtra}, + proxy::LatencyTimer, sasl, scram, stream::PqStream, }; @@ -16,6 +17,7 @@ pub(super) async fn authenticate( creds: &ClientCredentials<'_>, client: &mut PqStream, config: &'static AuthenticationConfig, + latency_timer: &mut LatencyTimer, ) -> auth::Result> { info!("fetching user's authentication info"); let info = api.get_auth_info(extra, creds).await?.unwrap_or_else(|| { @@ -36,24 +38,26 @@ pub(super) async fn authenticate( info!("auth endpoint chooses SCRAM"); let scram = auth::Scram(&secret); - let auth_flow = flow.begin(scram).await.map_err(|error| { - warn!(?error, "error sending scram acknowledgement"); - error - })?; - let auth_outcome = tokio::time::timeout( config.scram_protocol_timeout, - auth_flow.authenticate(), + async { + // pause the timer while we communicate with the client + let _paused = latency_timer.pause(); + + flow.begin(scram).await.map_err(|error| { + warn!(?error, "error sending scram acknowledgement"); + error + })?.authenticate().await.map_err(|error| { + warn!(?error, "error processing scram messages"); + error + }) + } ) .await .map_err(|error| { warn!("error processing scram messages error = authentication timed out, execution time exeeded {} seconds", config.scram_protocol_timeout.as_secs()); auth::io::Error::new(auth::io::ErrorKind::TimedOut, error) - })? - .map_err(|error| { - warn!(?error, "error processing scram messages"); - error - })?; + })??; let client_key = match auth_outcome { sasl::Outcome::Success(key) => key, diff --git a/proxy/src/auth/backend/hacks.rs b/proxy/src/auth/backend/hacks.rs index 53958bf337..895683af1b 100644 --- a/proxy/src/auth/backend/hacks.rs +++ b/proxy/src/auth/backend/hacks.rs @@ -1,6 +1,7 @@ use super::{AuthSuccess, ComputeCredentials}; use crate::{ auth::{self, AuthFlow, ClientCredentials}, + proxy::LatencyTimer, stream, }; use tokio::io::{AsyncRead, AsyncWrite}; @@ -12,8 +13,13 @@ use tracing::{info, warn}; /// use this mechanism for websocket connections. pub async fn cleartext_hack( client: &mut stream::PqStream, + latency_timer: &mut LatencyTimer, ) -> auth::Result> { warn!("cleartext auth flow override is enabled, proceeding"); + + // pause the timer while we communicate with the client + let _paused = latency_timer.pause(); + let password = AuthFlow::new(client) .begin(auth::CleartextPassword) .await? @@ -32,8 +38,13 @@ pub async fn cleartext_hack( pub async fn password_hack( creds: &mut ClientCredentials<'_>, client: &mut stream::PqStream, + latency_timer: &mut LatencyTimer, ) -> auth::Result> { warn!("project not specified, resorting to the password hack auth flow"); + + // pause the timer while we communicate with the client + let _paused = latency_timer.pause(); + let payload = AuthFlow::new(client) .begin(auth::PasswordHack) .await? diff --git a/proxy/src/bin/proxy.rs b/proxy/src/bin/proxy.rs index a9ca308797..28d0d95c5b 100644 --- a/proxy/src/bin/proxy.rs +++ b/proxy/src/bin/proxy.rs @@ -16,9 +16,10 @@ use tokio::task::JoinSet; use tokio_util::sync::CancellationToken; use tracing::info; use tracing::warn; -use utils::{project_git_version, sentry_init::init_sentry}; +use utils::{project_build_tag, project_git_version, sentry_init::init_sentry}; project_git_version!(GIT_VERSION); +project_build_tag!(BUILD_TAG); use clap::{Parser, ValueEnum}; @@ -100,7 +101,8 @@ async fn main() -> anyhow::Result<()> { let _sentry_guard = init_sentry(Some(GIT_VERSION.into()), &[]); info!("Version: {GIT_VERSION}"); - ::metrics::set_build_info_metric(GIT_VERSION); + info!("Build_tag: {BUILD_TAG}"); + ::metrics::set_build_info_metric(GIT_VERSION, BUILD_TAG); let args = ProxyCliArgs::parse(); let config = build_config(&args)?; diff --git a/proxy/src/config.rs b/proxy/src/config.rs index 96856ba051..9607ecd153 100644 --- a/proxy/src/config.rs +++ b/proxy/src/config.rs @@ -223,7 +223,7 @@ pub struct CacheOptions { impl CacheOptions { /// Default options for [`crate::console::provider::NodeInfoCache`]. - pub const DEFAULT_OPTIONS_NODE_INFO: &str = "size=4000,ttl=4m"; + pub const DEFAULT_OPTIONS_NODE_INFO: &'static str = "size=4000,ttl=4m"; /// Parse cache options passed via cmdline. /// Example: [`Self::DEFAULT_OPTIONS_NODE_INFO`]. diff --git a/proxy/src/console/provider.rs b/proxy/src/console/provider.rs index 86fd810ac3..32c3467092 100644 --- a/proxy/src/console/provider.rs +++ b/proxy/src/console/provider.rs @@ -97,8 +97,6 @@ pub mod errors { !text.contains("quota exceeded") && !text.contains("the limit for current plan reached") } - // retry server errors - Self::Console { status, .. } if status.is_server_error() => true, _ => false, } } diff --git a/proxy/src/console/provider/mock.rs b/proxy/src/console/provider/mock.rs index 282567269d..750a2d141e 100644 --- a/proxy/src/console/provider/mock.rs +++ b/proxy/src/console/provider/mock.rs @@ -59,7 +59,7 @@ impl Api { let rows = client.query(query, &[&creds.user]).await?; // We can get at most one row, because `rolname` is unique. - let row = match rows.get(0) { + let row = match rows.first() { Some(row) => row, // This means that the user doesn't exist, so there can be no secret. // However, this is still a *valid* outcome which is very similar diff --git a/proxy/src/proxy.rs b/proxy/src/proxy.rs index 9a01a27fc4..884aae1651 100644 --- a/proxy/src/proxy.rs +++ b/proxy/src/proxy.rs @@ -106,17 +106,26 @@ static COMPUTE_CONNECTION_LATENCY: Lazy = Lazy::new(|| { }); pub struct LatencyTimer { - start: Instant, + // time since the stopwatch was started + start: Option, + // accumulated time on the stopwatch + accumulated: std::time::Duration, + // label data protocol: &'static str, cache_miss: bool, pool_miss: bool, outcome: &'static str, } +pub struct LatencyTimerPause<'a> { + timer: &'a mut LatencyTimer, +} + impl LatencyTimer { pub fn new(protocol: &'static str) -> Self { Self { - start: Instant::now(), + start: Some(Instant::now()), + accumulated: std::time::Duration::ZERO, protocol, cache_miss: false, // by default we don't do pooling @@ -126,6 +135,13 @@ impl LatencyTimer { } } + pub fn pause(&mut self) -> LatencyTimerPause<'_> { + // stop the stopwatch and record the time that we have accumulated + let start = self.start.take().expect("latency timer should be started"); + self.accumulated += start.elapsed(); + LatencyTimerPause { timer: self } + } + pub fn cache_miss(&mut self) { self.cache_miss = true; } @@ -139,9 +155,17 @@ impl LatencyTimer { } } +impl Drop for LatencyTimerPause<'_> { + fn drop(&mut self) { + // start the stopwatch again + self.timer.start = Some(Instant::now()); + } +} + impl Drop for LatencyTimer { fn drop(&mut self) { - let duration = self.start.elapsed().as_secs_f64(); + let duration = + self.start.map(|start| start.elapsed()).unwrap_or_default() + self.accumulated; COMPUTE_CONNECTION_LATENCY .with_label_values(&[ self.protocol, @@ -149,7 +173,7 @@ impl Drop for LatencyTimer { bool_to_str(self.pool_miss), self.outcome, ]) - .observe(duration) + .observe(duration.as_secs_f64()) } } @@ -862,10 +886,16 @@ impl Client<'_, S> { application_name: params.get("application_name"), }; - let latency_timer = LatencyTimer::new(mode.protocol_label()); + let mut latency_timer = LatencyTimer::new(mode.protocol_label()); let auth_result = match creds - .authenticate(&extra, &mut stream, mode.allow_cleartext(), config) + .authenticate( + &extra, + &mut stream, + mode.allow_cleartext(), + config, + &mut latency_timer, + ) .await { Ok(auth_result) => auth_result, diff --git a/proxy/src/proxy/tests.rs b/proxy/src/proxy/tests.rs index 8f7f461bef..142c32fb84 100644 --- a/proxy/src/proxy/tests.rs +++ b/proxy/src/proxy/tests.rs @@ -409,7 +409,7 @@ impl TestBackend for TestConnectMechanism { } ConnectAction::WakeRetry => { let err = console::errors::ApiError::Console { - status: http::StatusCode::INTERNAL_SERVER_ERROR, + status: http::StatusCode::BAD_REQUEST, text: "TEST".into(), }; assert!(err.could_retry()); diff --git a/proxy/src/scram.rs b/proxy/src/scram.rs index 07822e8da5..2de26af96b 100644 --- a/proxy/src/scram.rs +++ b/proxy/src/scram.rs @@ -18,7 +18,6 @@ mod password; pub use exchange::Exchange; pub use key::ScramKey; pub use secret::ServerSecret; -pub use secret::*; use hmac::{Hmac, Mac}; use sha2::{Digest, Sha256}; diff --git a/safekeeper/src/bin/safekeeper.rs b/safekeeper/src/bin/safekeeper.rs index 202343fca1..d476cf33ae 100644 --- a/safekeeper/src/bin/safekeeper.rs +++ b/safekeeper/src/bin/safekeeper.rs @@ -42,7 +42,7 @@ use utils::auth::{JwtAuth, Scope}; use utils::{ id::NodeId, logging::{self, LogFormat}, - project_git_version, + project_build_tag, project_git_version, sentry_init::init_sentry, tcp_listener, }; @@ -51,6 +51,7 @@ const PID_FILE_NAME: &str = "safekeeper.pid"; const ID_FILE_NAME: &str = "safekeeper.id"; project_git_version!(GIT_VERSION); +project_build_tag!(BUILD_TAG); const ABOUT: &str = r#" A fleet of safekeepers is responsible for reliably storing WAL received from @@ -204,6 +205,7 @@ async fn main() -> anyhow::Result<()> { )?; logging::replace_panic_hook_with_tracing_panic_hook().forget(); info!("version: {GIT_VERSION}"); + info!("buld_tag: {BUILD_TAG}"); let args_workdir = &args.datadir; let workdir = args_workdir.canonicalize_utf8().with_context(|| { @@ -423,7 +425,7 @@ async fn start_safekeeper(conf: SafeKeeperConf) -> Result<()> { .map(|res| ("WAL remover".to_owned(), res)); tasks_handles.push(Box::pin(wal_remover_handle)); - set_build_info_metric(GIT_VERSION); + set_build_info_metric(GIT_VERSION, BUILD_TAG); // TODO: update tokio-stream, convert to real async Stream with // SignalStream, map it to obtain missing signal name, combine streams into diff --git a/safekeeper/src/handler.rs b/safekeeper/src/handler.rs index 71b99ab1d8..6d0ed8650d 100644 --- a/safekeeper/src/handler.rs +++ b/safekeeper/src/handler.rs @@ -140,6 +140,12 @@ impl postgres_backend::Handler } } + let ttid = TenantTimelineId::new( + self.tenant_id.unwrap_or(TenantId::from([0u8; 16])), + self.timeline_id.unwrap_or(TimelineId::from([0u8; 16])), + ); + tracing::Span::current().record("ttid", tracing::field::display(ttid)); + Ok(()) } else { Err(QueryError::Other(anyhow::anyhow!( @@ -208,26 +214,22 @@ impl postgres_backend::Handler PG_QUERIES_FINISHED.with_label_values(&[cmd_str]).inc(); } - info!( - "got query {:?} in timeline {:?}", - query_string, self.timeline_id - ); + info!("got query {:?}", query_string); let tenant_id = self.tenant_id.context("tenantid is required")?; let timeline_id = self.timeline_id.context("timelineid is required")?; self.check_permission(Some(tenant_id))?; self.ttid = TenantTimelineId::new(tenant_id, timeline_id); - let span_ttid = self.ttid; // satisfy borrow checker match cmd { SafekeeperPostgresCommand::StartWalPush => { self.handle_start_wal_push(pgb) - .instrument(info_span!("WAL receiver", ttid = %span_ttid)) + .instrument(info_span!("WAL receiver")) .await } SafekeeperPostgresCommand::StartReplication { start_lsn, term } => { self.handle_start_replication(pgb, start_lsn, term) - .instrument(info_span!("WAL sender", ttid = %span_ttid)) + .instrument(info_span!("WAL sender")) .await } SafekeeperPostgresCommand::IdentifySystem => self.handle_identify_system(pgb).await, diff --git a/safekeeper/src/remove_wal.rs b/safekeeper/src/remove_wal.rs index 3306f0b63a..d96eedf401 100644 --- a/safekeeper/src/remove_wal.rs +++ b/safekeeper/src/remove_wal.rs @@ -16,20 +16,16 @@ pub async fn task_main(conf: SafeKeeperConf) -> anyhow::Result<()> { continue; } let ttid = tli.ttid; - if let Err(e) = tli - .maybe_persist_control_file() - .instrument(info_span!("", tenant = %ttid.tenant_id, timeline = %ttid.timeline_id)) - .await - { - warn!("failed to persist control file: {e}"); - } - if let Err(e) = tli - .remove_old_wal(conf.wal_backup_enabled) - .instrument(info_span!("", tenant = %ttid.tenant_id, timeline = %ttid.timeline_id)) - .await - { - error!("failed to remove WAL: {}", e); + async { + if let Err(e) = tli.maybe_persist_control_file().await { + warn!("failed to persist control file: {e}"); + } + if let Err(e) = tli.remove_old_wal(conf.wal_backup_enabled).await { + error!("failed to remove WAL: {}", e); + } } + .instrument(info_span!("WAL removal", ttid = %ttid)) + .await; } sleep(wal_removal_interval).await; } diff --git a/safekeeper/src/timelines_global_map.rs b/safekeeper/src/timelines_global_map.rs index 1434644123..cbb3342e40 100644 --- a/safekeeper/src/timelines_global_map.rs +++ b/safekeeper/src/timelines_global_map.rs @@ -296,8 +296,8 @@ impl GlobalTimelines { global_lock .timelines .values() - .cloned() .filter(|t| !t.is_cancelled()) + .cloned() .collect() } diff --git a/safekeeper/src/wal_backup.rs b/safekeeper/src/wal_backup.rs index da8c197411..22c68ce3c9 100644 --- a/safekeeper/src/wal_backup.rs +++ b/safekeeper/src/wal_backup.rs @@ -136,7 +136,7 @@ async fn update_task( if elected_me != (entry.handle.is_some()) { if elected_me { - info!("elected for backup {}: {}", ttid, election_dbg_str); + info!("elected for backup: {}", election_dbg_str); let (shutdown_tx, shutdown_rx) = mpsc::channel(1); let timeline_dir = conf.timeline_dir(&ttid); @@ -149,7 +149,7 @@ async fn update_task( conf.backup_parallel_jobs, shutdown_rx, ) - .instrument(info_span!("WAL backup task", ttid = %ttid)), + .in_current_span(), ); entry.handle = Some(WalBackupTaskHandle { @@ -157,7 +157,7 @@ async fn update_task( handle, }); } else { - info!("stepping down from backup {}: {}", ttid, election_dbg_str); + info!("stepping down from backup: {}", election_dbg_str); shut_down_task(ttid, entry).await; } } @@ -199,29 +199,33 @@ pub async fn wal_backup_launcher_task_main( if conf.remote_storage.is_none() || !conf.wal_backup_enabled { continue; /* just drain the channel and do nothing */ } - let timeline = is_wal_backup_required(ttid).await; - // do we need to do anything at all? - if timeline.is_some() != tasks.contains_key(&ttid) { - if let Some(timeline) = timeline { - // need to start the task - let entry = tasks.entry(ttid).or_insert(WalBackupTimelineEntry { - timeline, - handle: None, - }); - update_task(&conf, ttid, entry).await; - } else { - // need to stop the task - info!("stopping WAL backup task for {}", ttid); - let mut entry = tasks.remove(&ttid).unwrap(); - shut_down_task(ttid, &mut entry).await; + async { + let timeline = is_wal_backup_required(ttid).await; + // do we need to do anything at all? + if timeline.is_some() != tasks.contains_key(&ttid) { + if let Some(timeline) = timeline { + // need to start the task + let entry = tasks.entry(ttid).or_insert(WalBackupTimelineEntry { + timeline, + handle: None, + }); + update_task(&conf, ttid, entry).await; + } else { + // need to stop the task + info!("stopping WAL backup task"); + let mut entry = tasks.remove(&ttid).unwrap(); + shut_down_task(ttid, &mut entry).await; + } } - } + }.instrument(info_span!("WAL backup", ttid = %ttid)).await; } // For each timeline needing offloading, check if this safekeeper // should do the job and start/stop the task accordingly. _ = ticker.tick() => { for (ttid, entry) in tasks.iter_mut() { - update_task(&conf, *ttid, entry).await; + update_task(&conf, *ttid, entry) + .instrument(info_span!("WAL backup", ttid = %ttid)) + .await; } } } @@ -248,7 +252,7 @@ async fn backup_task_main( info!("started"); let res = GlobalTimelines::get(ttid); if let Err(e) = res { - error!("backup error for timeline {}: {}", ttid, e); + error!("backup error: {}", e); return; } let tli = res.unwrap(); @@ -346,7 +350,7 @@ impl WalBackupTask { } } -pub async fn backup_lsn_range( +async fn backup_lsn_range( timeline: &Arc, backup_lsn: &mut Lsn, end_lsn: Lsn, diff --git a/safekeeper/src/wal_service.rs b/safekeeper/src/wal_service.rs index 9fabaa79fb..bceaad1e16 100644 --- a/safekeeper/src/wal_service.rs +++ b/safekeeper/src/wal_service.rs @@ -36,14 +36,14 @@ pub async fn task_main( let conf = conf.clone(); let conn_id = issue_connection_id(&mut connection_count); - tokio::spawn(async move { - if let Err(err) = handle_socket(socket, conf, conn_id, allowed_auth_scope) - .instrument(info_span!("", cid = %conn_id)) - .await - { - error!("connection handler exited: {}", err); + tokio::spawn( + async move { + if let Err(err) = handle_socket(socket, conf, conn_id, allowed_auth_scope).await { + error!("connection handler exited: {}", err); + } } - }); + .instrument(info_span!("", cid = %conn_id, ttid = field::Empty)), + ); } } diff --git a/safekeeper/src/wal_storage.rs b/safekeeper/src/wal_storage.rs index 2070122e8e..fa44b24258 100644 --- a/safekeeper/src/wal_storage.rs +++ b/safekeeper/src/wal_storage.rs @@ -188,7 +188,7 @@ impl PhysicalStorage { } /// Call fdatasync if config requires so. - async fn fdatasync_file(&mut self, file: &mut File) -> Result<()> { + async fn fdatasync_file(&mut self, file: &File) -> Result<()> { if !self.conf.no_sync { self.metrics .observe_flush_seconds(time_io_closure(file.sync_data()).await?); @@ -197,7 +197,7 @@ impl PhysicalStorage { } /// Call fsync if config requires so. - async fn fsync_file(&mut self, file: &mut File) -> Result<()> { + async fn fsync_file(&mut self, file: &File) -> Result<()> { if !self.conf.no_sync { self.metrics .observe_flush_seconds(time_io_closure(file.sync_all()).await?); @@ -231,7 +231,7 @@ impl PhysicalStorage { .with_context(|| format!("Failed to open log file {:?}", &wal_file_path))?; write_zeroes(&mut file, self.wal_seg_size).await?; - self.fsync_file(&mut file).await?; + self.fsync_file(&file).await?; Ok((file, true)) } } @@ -255,7 +255,7 @@ impl PhysicalStorage { if xlogoff + buf.len() == self.wal_seg_size { // If we reached the end of a WAL segment, flush and close it. - self.fdatasync_file(&mut file).await?; + self.fdatasync_file(&file).await?; // Rename partial file to completed file let (wal_file_path, wal_file_partial_path) = @@ -277,8 +277,8 @@ impl PhysicalStorage { async fn write_exact(&mut self, pos: Lsn, mut buf: &[u8]) -> Result<()> { if self.write_lsn != pos { // need to flush the file before discarding it - if let Some(mut file) = self.file.take() { - self.fdatasync_file(&mut file).await?; + if let Some(file) = self.file.take() { + self.fdatasync_file(&file).await?; } self.write_lsn = pos; @@ -367,8 +367,8 @@ impl Storage for PhysicalStorage { return Ok(()); } - if let Some(mut unflushed_file) = self.file.take() { - self.fdatasync_file(&mut unflushed_file).await?; + if let Some(unflushed_file) = self.file.take() { + self.fdatasync_file(&unflushed_file).await?; self.file = Some(unflushed_file); } else { // We have unflushed data (write_lsn != flush_lsn), but no file. @@ -410,8 +410,8 @@ impl Storage for PhysicalStorage { } // Close previously opened file, if any - if let Some(mut unflushed_file) = self.file.take() { - self.fdatasync_file(&mut unflushed_file).await?; + if let Some(unflushed_file) = self.file.take() { + self.fdatasync_file(&unflushed_file).await?; } let xlogoff = end_pos.segment_offset(self.wal_seg_size); @@ -425,7 +425,7 @@ impl Storage for PhysicalStorage { // Fill end with zeroes file.seek(SeekFrom::Start(xlogoff as u64)).await?; write_zeroes(&mut file, self.wal_seg_size - xlogoff).await?; - self.fdatasync_file(&mut file).await?; + self.fdatasync_file(&file).await?; if !is_partial { // Make segment partial once again diff --git a/storage_broker/src/bin/storage_broker.rs b/storage_broker/src/bin/storage_broker.rs index 23d1bee354..5ce24a6a42 100644 --- a/storage_broker/src/bin/storage_broker.rs +++ b/storage_broker/src/bin/storage_broker.rs @@ -44,10 +44,11 @@ use storage_broker::{ }; use utils::id::TenantTimelineId; use utils::logging::{self, LogFormat}; -use utils::project_git_version; use utils::sentry_init::init_sentry; +use utils::{project_build_tag, project_git_version}; project_git_version!(GIT_VERSION); +project_build_tag!(BUILD_TAG); const DEFAULT_CHAN_SIZE: usize = 32; const DEFAULT_ALL_KEYS_CHAN_SIZE: usize = 16384; @@ -438,7 +439,8 @@ async fn main() -> Result<(), Box> { // initialize sentry if SENTRY_DSN is provided let _sentry_guard = init_sentry(Some(GIT_VERSION.into()), &[]); info!("version: {GIT_VERSION}"); - ::metrics::set_build_info_metric(GIT_VERSION); + info!("build_tag: {BUILD_TAG}"); + metrics::set_build_info_metric(GIT_VERSION, BUILD_TAG); // On any shutdown signal, log receival and exit. std::thread::spawn(move || { diff --git a/test_runner/fixtures/neon_fixtures.py b/test_runner/fixtures/neon_fixtures.py index 7a3e0a9efc..81a7b0750d 100644 --- a/test_runner/fixtures/neon_fixtures.py +++ b/test_runner/fixtures/neon_fixtures.py @@ -1554,8 +1554,8 @@ class NeonAttachmentService: def attach_hook(self, tenant_id: TenantId, pageserver_id: int) -> int: response = requests.post( - f"{self.env.control_plane_api}/attach_hook", - json={"tenant_id": str(tenant_id), "pageserver_id": pageserver_id}, + f"{self.env.control_plane_api}/attach-hook", + json={"tenant_id": str(tenant_id), "node_id": pageserver_id}, ) response.raise_for_status() gen = response.json()["gen"] diff --git a/test_runner/performance/test_bulk_insert.py b/test_runner/performance/test_bulk_insert.py index aafc315576..62301f3919 100644 --- a/test_runner/performance/test_bulk_insert.py +++ b/test_runner/performance/test_bulk_insert.py @@ -1,7 +1,7 @@ -import shutil from contextlib import closing from fixtures.compare_fixtures import NeonCompare, PgCompare +from fixtures.pageserver.utils import wait_tenant_status_404 from fixtures.pg_version import PgVersion @@ -42,17 +42,14 @@ def measure_recovery_time(env: NeonCompare): client = env.env.pageserver.http_client() pg_version = PgVersion(client.timeline_detail(env.tenant, env.timeline)["pg_version"]) - # Stop pageserver and remove tenant data - env.env.pageserver.stop() - timeline_dir = env.env.pageserver.timeline_dir(env.tenant, env.timeline) - shutil.rmtree(timeline_dir) - - # Start pageserver - env.env.pageserver.start() + # Delete the Tenant in the pageserver: this will drop local and remote layers, such that + # when we "create" the Tenant again, we will replay the WAL from the beginning. + client.tenant_delete(env.tenant) + wait_tenant_status_404(client, env.tenant, iterations=60, interval=0.5) + client.tenant_create(new_tenant_id=env.tenant) # Measure recovery time with env.record_duration("wal_recovery"): - # Create the tenant, which will start walingest client.timeline_create(pg_version, env.tenant, env.timeline) # Flush, which will also wait for lsn to catch up diff --git a/test_runner/regress/test_build_info_metric.py b/test_runner/regress/test_build_info_metric.py index 4e53928d14..8f714dae67 100644 --- a/test_runner/regress/test_build_info_metric.py +++ b/test_runner/regress/test_build_info_metric.py @@ -17,3 +17,6 @@ def test_build_info_metric(neon_env_builder: NeonEnvBuilder, link_proxy: NeonPro assert "revision" in sample.labels assert len(sample.labels["revision"]) > 0 + + assert "build_tag" in sample.labels + assert len(sample.labels["build_tag"]) > 0 diff --git a/test_runner/regress/test_remote_storage.py b/test_runner/regress/test_remote_storage.py index f201c735e1..31bc97703e 100644 --- a/test_runner/regress/test_remote_storage.py +++ b/test_runner/regress/test_remote_storage.py @@ -585,9 +585,11 @@ def test_timeline_deletion_with_files_stuck_in_upload_queue( # checkpoint operations. Hence, checkpoint is allowed to fail now. log.info("sending delete request") checkpoint_allowed_to_fail.set() - env.pageserver.allowed_errors.append( - ".* ERROR .*Error processing HTTP request: InternalServerError\\(The timeline or pageserver is shutting down" - ".* ERROR .*[Cc]ould not flush frozen layer.*" + env.pageserver.allowed_errors.extend( + [ + ".* ERROR .*Error processing HTTP request: InternalServerError\\(The timeline or pageserver is shutting down", + ".* ERROR .*[Cc]ould not flush frozen layer.*", + ] ) # Generous timeout, because currently deletions can get blocked waiting for compaction