From 83567f9e4e95f79c5298a30c4bc57125c87548fd Mon Sep 17 00:00:00 2001 From: John Spray Date: Fri, 27 Oct 2023 09:23:49 +0100 Subject: [PATCH 01/19] tests: revise perf test that interfered with local disk state (#5682) This benchmark started failing after #5580 merged. It was manually deleting some local content on a pageserver, and expecting the behavior that the pageserver would "forget" about the timeline on startup as a result. That is no longer our behavior: pageservers use the remote storage as the source of truth. Rather than having the test go manually delete things at all, we can just delete the whole tenant via the pageserver API, and thereby start from a clean situation. --- test_runner/performance/test_bulk_insert.py | 15 ++++++--------- 1 file changed, 6 insertions(+), 9 deletions(-) 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 From 39f8fd69456ba59e11fec9e6a52beb58f011bef3 Mon Sep 17 00:00:00 2001 From: duguorong009 <80258679+duguorong009@users.noreply.github.com> Date: Fri, 27 Oct 2023 05:47:11 -0400 Subject: [PATCH 02/19] feat: add `build_tag` env support for `set_build_info_metric` (#5576) - Add a new util `project_build_tag` macro, similar to `project_git_version` - Update the `set_build_info_metric` to accept and make use of `build_tag` info - Update all codes which use the `set_build_info_metric` --- libs/metrics/src/lib.rs | 6 +++--- libs/utils/src/lib.rs | 15 +++++++++++++++ pageserver/src/bin/pageserver.rs | 12 +++++++----- proxy/src/bin/proxy.rs | 6 ++++-- safekeeper/src/bin/safekeeper.rs | 6 ++++-- storage_broker/src/bin/storage_broker.rs | 6 ++++-- test_runner/regress/test_build_info_metric.py | 3 +++ 7 files changed, 40 insertions(+), 14 deletions(-) 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/utils/src/lib.rs b/libs/utils/src/lib.rs index b3e38cd9e1..88cefd516d 100644 --- a/libs/utils/src/lib.rs +++ b/libs/utils/src/lib.rs @@ -130,6 +130,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/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/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/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/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 From 68f15cf967d36936341477c9ccdae38ebab68e8d Mon Sep 17 00:00:00 2001 From: Joonas Koivunen Date: Fri, 27 Oct 2023 13:16:01 +0300 Subject: [PATCH 03/19] fix: schedule_compaction_update must only unlink (#5675) #5649 added the concept of dangling layers which #4938 uses but only partially. I forgot to change `schedule_compaction_update` to not schedule deletions to uphold the "have a layer, you can read it". With the now remembered fix, I don't think these checks should ever fail except for a mistake I already did. These changes might be useful for protecting future changes, even though the Layer carrying the generation AND the `schedule_(gc|compaction)_update` require strong arcs. Rationale for keeping the `#[cfg(feature = "testing")]` is worsening any leak situation which might come up. --- .../src/tenant/remote_timeline_client.rs | 30 +++++++++++++++++-- pageserver/src/tenant/upload_queue.rs | 12 ++++++++ 2 files changed, 39 insertions(+), 3 deletions(-) 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/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); From 262348e41b04dab7643c2854f4af0cd4a8a20a58 Mon Sep 17 00:00:00 2001 From: Arthur Petukhovsky Date: Fri, 27 Oct 2023 12:09:02 +0100 Subject: [PATCH 04/19] Fix safekeeper log spans (#5643) We were missing spans with ttid in "WAL backup" and several other places, this commit should fix it. Here are the examples of logs before and after: https://gist.github.com/petuhovskiy/711a4a4e7ddde3cab3fa6419b2f70fb9 --- safekeeper/src/handler.rs | 16 +++++++----- safekeeper/src/remove_wal.rs | 22 +++++++--------- safekeeper/src/wal_backup.rs | 48 +++++++++++++++++++---------------- safekeeper/src/wal_service.rs | 14 +++++----- 4 files changed, 51 insertions(+), 49 deletions(-) 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/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)), + ); } } From a5292f7e67727307341c1d4a372ee7f69efbb626 Mon Sep 17 00:00:00 2001 From: Gleb Novikov Date: Fri, 27 Oct 2023 14:36:34 +0300 Subject: [PATCH 05/19] Some minor renames in attachment service API (#5687) ## Problem ## Summary of changes ## Checklist before requesting a review - [x] I have performed a self-review of my code. - [ ] ~~If it is a core feature, I have added thorough tests.~~ - [ ] ~~Do we need to implement analytics? if so did you add the relevant metrics to the dashboard?~~ - [ ] ~~If this PR requires public announcement, mark it with /release-notes label and add several sentences in this section.~~ ## Checklist before merging - [ ] Do not forget to reformat commit message to not include the above checklist --- control_plane/src/attachment_service.rs | 6 +++--- control_plane/src/bin/attachment_service.rs | 12 ++++++------ libs/pageserver_api/src/control_api.rs | 2 +- pageserver/src/control_plane_client.rs | 2 +- test_runner/fixtures/neon_fixtures.py | 4 ++-- 5 files changed, 13 insertions(+), 13 deletions(-) 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>()) } 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"] From c13e932c3bd8096669619326ab8decdfd2ffca20 Mon Sep 17 00:00:00 2001 From: John Spray Date: Fri, 27 Oct 2023 14:20:04 +0100 Subject: [PATCH 06/19] pageserver: add generation fields in openapi spec (#5690) These optional fields have existed for as while, but weren't mentioned in `openapi_spec.yaml` yet. --- pageserver/src/http/openapi_spec.yml | 15 +++++++++++++++ 1 file changed, 15 insertions(+) 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: From 493b47e1da97b028423396c37ee3ec936b714323 Mon Sep 17 00:00:00 2001 From: Conrad Ludgate Date: Fri, 27 Oct 2023 18:17:39 +0100 Subject: [PATCH 07/19] proxy: exclude client latencies in metrics (#5688) ## Problem In #5539, I moved the connect_to_compute latency to start counting before authentication - this is because authentication will perform some calls to the control plane in order to get credentials and to eagerly wake a compute server. It felt important to include these times in the latency metric as these are times we should definitely care about reducing. What is not interesting to record in this metric is the roundtrip time during authentication when we wait for the client to respond. ## Summary of changes Implement a mechanism to pause the latency timer, resuming on drop of the pause struct. We pause the timer right before we send the authentication message to the client, and we resume the timer right after we complete the authentication flow. --- proxy/src/auth/backend.rs | 44 ++++++++++++++++++++++++++----- proxy/src/auth/backend/classic.rs | 26 ++++++++++-------- proxy/src/auth/backend/hacks.rs | 11 ++++++++ proxy/src/proxy.rs | 42 ++++++++++++++++++++++++----- 4 files changed, 99 insertions(+), 24 deletions(-) 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/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, From 2bd79906d944fa1e670727196e613192fc994c58 Mon Sep 17 00:00:00 2001 From: Joonas Koivunen Date: Fri, 27 Oct 2023 21:09:34 +0300 Subject: [PATCH 08/19] fix: possible page_service hang on cancel (#5696) Fixes #5341, one more suspected case, see: https://github.com/neondatabase/neon/issues/5341#issuecomment-1783052379 - races `MaybeWriteOnly::shutdown` with cancellation - switches to using `AsyncWriteExt::write_buf` - notes cancellation safety for shutdown --- libs/postgres_backend/src/lib.rs | 21 ++++++++++++++++----- libs/pq_proto/src/framed.rs | 9 +++------ 2 files changed, 19 insertions(+), 11 deletions(-) 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/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, From bd59349af373cbd4965c3f8e49facb7623ce5f88 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Arpad=20M=C3=BCller?= Date: Sat, 28 Oct 2023 03:47:26 +0200 Subject: [PATCH 09/19] Fix Rust 1.74 warnings (#5702) Fixes new warnings and clippy changes introduced by version 1.74 of the rust compiler toolchain. --- pageserver/src/deletion_queue/list_writer.rs | 2 +- .../tenant/remote_timeline_client/index.rs | 2 +- pageserver/src/tenant/timeline.rs | 24 +++++++++---------- .../walreceiver/walreceiver_connection.rs | 6 ++--- pageserver/src/walingest.rs | 4 ++-- proxy/src/config.rs | 2 +- safekeeper/src/timelines_global_map.rs | 2 +- safekeeper/src/wal_storage.rs | 22 ++++++++--------- 8 files changed, 31 insertions(+), 33 deletions(-) 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/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/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/walreceiver/walreceiver_connection.rs b/pageserver/src/tenant/timeline/walreceiver/walreceiver_connection.rs index 0831b9ceda..3a6599ec55 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?; 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/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/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_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 From 964c5c56b75ac98344af6c364a2764864f480cfd Mon Sep 17 00:00:00 2001 From: Conrad Ludgate Date: Mon, 30 Oct 2023 08:38:56 +0000 Subject: [PATCH 10/19] proxy: dont retry server errors (#5694) ## Problem accidental spam ## Summary of changes don't spam control plane if control plane is down :) ## Checklist before requesting a review - [x] I have performed a self-review of my code. - [ ] If it is a core feature, I have added thorough tests. - [ ] Do we need to implement analytics? if so did you add the relevant metrics to the dashboard? - [ ] If this PR requires public announcement, mark it with /release-notes label and add several sentences in this section. ## Checklist before merging - [ ] Do not forget to reformat commit message to not include the above checklist --- proxy/src/console/provider.rs | 2 -- proxy/src/proxy/tests.rs | 2 +- 2 files changed, 1 insertion(+), 3 deletions(-) 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/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()); From 87db4b441c31fb704772beddf46307bea444bbe9 Mon Sep 17 00:00:00 2001 From: John Spray Date: Mon, 30 Oct 2023 09:18:40 +0000 Subject: [PATCH 11/19] pageserver: cleaner shutdown in timeline delete (#5701) The flush task logs a backtrace if it tries to upload and remote timeline client is already in stopped state. Therefore we cannot shut them down concurrently: flush task must be shut down first. This wasn't more obvious because: - Timeline deletions IRL usually happen when not much is being written - In tests, there is a global allow-list for this log It's not obvious whether removing the global log allow list is safe, this PR was prompted by how the log spam got in my way when testing deletion changes. --- pageserver/src/tenant/timeline/delete.rs | 8 ++++++++ 1 file changed, 8 insertions(+) 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(); From 9cdffd164a8b3d486360fd306bb37ac8fb5ebed3 Mon Sep 17 00:00:00 2001 From: Konstantin Knizhnik Date: Mon, 30 Oct 2023 12:06:08 +0200 Subject: [PATCH 12/19] Prevent SIGSEGV in apply_error_callback when record was not decoded (#5703) ## Problem See https://neondb.slack.com/archives/C036U0GRMRB/p1698652221399419?thread_ts=1698438997.903919&cid=C036U0GRMRB ## Summary of changes Check if record pointer is not NULL before trying to print record descriptor ## Checklist before requesting a review - [ ] I have performed a self-review of my code. - [ ] If it is a core feature, I have added thorough tests. - [ ] Do we need to implement analytics? if so did you add the relevant metrics to the dashboard? - [ ] If this PR requires public announcement, mark it with /release-notes label and add several sentences in this section. ## Checklist before merging - [ ] Do not forget to reformat commit message to not include the above checklist Co-authored-by: Konstantin Knizhnik --- pgxn/neon_walredo/walredoproc.c | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) 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", From 07c2b29895321faeb86b0aa8e73c40381ce68adf Mon Sep 17 00:00:00 2001 From: John Spray Date: Mon, 30 Oct 2023 13:24:52 +0000 Subject: [PATCH 13/19] pageserver: fix error logging on stray timeline files (#5712) ## Problem If there were stray files in the timelines/ dir after tenant deletion, pageserver could panic on out of range. ## Summary of changes Use iterator `take()`, which doesn't care if the number of elements available is less than requested. --- pageserver/src/tenant/delete.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) 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:?}" ))); From 4db8efb2cf16f67bd8ecf23dd71f54de506cde50 Mon Sep 17 00:00:00 2001 From: Joonas Koivunen Date: Mon, 30 Oct 2023 16:21:30 +0200 Subject: [PATCH 14/19] Layer: logging fixes (#5676) - include Layer generation in the default display, with Generation::Broken as `-broken` - omit layer from `layer_gc` span because the api it works with needs to support N layers, so the api needs to log each layer --- pageserver/src/tenant/storage_layer/layer.rs | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) 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(); From e675f4cec85b9341448230af72ef732fb0dae02e Mon Sep 17 00:00:00 2001 From: John Spray Date: Mon, 30 Oct 2023 15:18:32 +0000 Subject: [PATCH 15/19] =?UTF-8?q?tests:=20fix=20missing=20comma=20=20in=20?= =?UTF-8?q?test=5Ftimeline=5Fdeletion=5Fwith=5Ffiles=5Fstuck=5F=E2=80=A6?= =?UTF-8?q?=20(#5713)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit …in_upload_queue This was a syntax mistake in https://github.com/neondatabase/neon/pull/5149 We didn't notice because the situation the log allow list covers is a relative rare race. --- test_runner/regress/test_remote_storage.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) 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 From ad99fa5f0393e2679e5323df653c508ffa0ac072 Mon Sep 17 00:00:00 2001 From: Konstantin Knizhnik Date: Mon, 30 Oct 2023 17:29:25 +0200 Subject: [PATCH 16/19] Grant BYPASSRLS and REPLICATION to exited roles (#5657) ## Problem Role need to have REPLICATION privilege to be able to used for logical replication. New roles are created with this option. This PR tries to update existed roles. ## Summary of changes Update roles in `handle_roles` method ## Checklist before requesting a review - [ ] I have performed a self-review of my code. - [ ] If it is a core feature, I have added thorough tests. - [ ] Do we need to implement analytics? if so did you add the relevant metrics to the dashboard? - [ ] If this PR requires public announcement, mark it with /release-notes label and add several sentences in this section. ## Checklist before merging - [ ] Do not forget to reformat commit message to not include the above checklist --------- Co-authored-by: Konstantin Knizhnik --- compute_tools/src/pg_helpers.rs | 7 ++++++- compute_tools/src/spec.rs | 5 ++++- libs/compute_api/src/spec.rs | 2 ++ 3 files changed, 12 insertions(+), 2 deletions(-) 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/libs/compute_api/src/spec.rs b/libs/compute_api/src/spec.rs index cfbd50d38a..c16deceebb 100644 --- a/libs/compute_api/src/spec.rs +++ b/libs/compute_api/src/spec.rs @@ -190,6 +190,8 @@ pub struct DeltaOp { pub struct Role { pub name: PgIdent, pub encrypted_password: Option, + pub replication: Option, + pub bypassrls: Option, pub options: GenericOptions, } From d8c21ec70d60f5e4a4675a16bc596cbf60eefc8f Mon Sep 17 00:00:00 2001 From: Conrad Ludgate Date: Mon, 30 Oct 2023 16:43:06 +0000 Subject: [PATCH 17/19] fix nightly 1.75 (#5719) ## Problem Neon doesn't compile on nightly and had numerous clippy complaints. ## Summary of changes 1. Fixed troublesome dependency 2. Fixed or ignored the lints where appropriate --- Cargo.lock | 23 ++++--------------- libs/postgres_ffi/wal_craft/src/lib.rs | 1 + .../walreceiver/walreceiver_connection.rs | 2 +- proxy/src/console/provider/mock.rs | 2 +- proxy/src/scram.rs | 1 - 5 files changed, 7 insertions(+), 22 deletions(-) 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/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/pageserver/src/tenant/timeline/walreceiver/walreceiver_connection.rs b/pageserver/src/tenant/timeline/walreceiver/walreceiver_connection.rs index 3a6599ec55..3e56753ad4 100644 --- a/pageserver/src/tenant/timeline/walreceiver/walreceiver_connection.rs +++ b/pageserver/src/tenant/timeline/walreceiver/walreceiver_connection.rs @@ -459,7 +459,7 @@ async fn identify_system(client: &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/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/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}; From 9c35e1e6e5609780a5bc55b5e98a4a711f6e6465 Mon Sep 17 00:00:00 2001 From: John Spray Date: Mon, 30 Oct 2023 18:32:30 +0000 Subject: [PATCH 18/19] pageserver: downgrade slow task warnings from warn to info (#5724) ## Problem In #5658 we suppressed the first-iteration output from these logs, but the volume of warnings is still problematic. ## Summary of changes - Downgrade all slow task warnings to INFO. The information is still there if we actively want to know about which tasks are running slowly, without polluting the overall stream of warnings with situations that are unsurprising to us. - Revert the previous change so that we output on the first iteration as we used to do. There is no reason to suppress these, now that the severity is just info. --- pageserver/src/tenant/tasks.rs | 38 ++++++++++++---------------------- 1 file changed, 13 insertions(+), 25 deletions(-) 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, From 1588601503a0dc6a92972fd9946d924d5eb1747a Mon Sep 17 00:00:00 2001 From: Shany Pozin Date: Mon, 30 Oct 2023 23:10:24 +0200 Subject: [PATCH 19/19] Move release PR creation to Friday (#5721) Prepare for a new release workflow * Release PR is created on Fridays * The discussion/approval happens during Friday * Sunday morning the deployment will be done in central-il and perf tests will be run * On Monday early IST morning gradually start rolling (starting from US regions as they are still in weekend time) See slack for discussion: https://neondb.slack.com/archives/C04P81J55LK/p1698565305607839?thread_ts=1698428241.031979&cid=C04P81J55LK --- .github/workflows/release.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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: