rust/neon
1
0
Fork 0
mirror of https://github.com/neondatabase/neon.git synced 2026-10-17 01:03:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

Hook deletion queue into generations

Browse Source
This commit is contained in:
John Spray
2023-08-30 15:35:51 +01:00
parent 584c0d3c7b
commit 35e4b43531
4 changed files with 47 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pageserver/src/deletion_queue.rs
+27 -35
View File
@@ -13,6 +13,7 @@ use thiserror::Error;
use tokio;
use tokio_util::sync::CancellationToken;
use tracing::{self, debug, error};
use utils::generation::Generation;
use utils::id::{TenantId, TenantTimelineId, TimelineId};
pub(crate) use self::backend::BackendQueueWorker;
@@ -204,7 +205,7 @@ impl DeletionQueueClient {
&self,
tenant_id: TenantId,
timeline_id: TimelineId,
layers: Vec<LayerFileName>,
layers: Vec<(LayerFileName, Generation)>,
) -> Result<(), DeletionQueueError> {
DELETION_QUEUE_SUBMITTED.inc_by(layers.len() as u64);
self.do_push(FrontendQueueMessage::Delete(DeletionOp {
@@ -363,7 +364,7 @@ mod test {
use remote_storage::{RemoteStorageConfig, RemoteStorageKind};
use tokio::{runtime::EnterGuard, task::JoinHandle};
use crate::tenant::harness::TenantHarness;
use crate::tenant::{harness::TenantHarness, remote_timeline_client::remote_timeline_path};
use super::*;
pub const TIMELINE_ID: TimelineId =
@@ -538,38 +539,37 @@ mod test {
let tenant_id = ctx.harness.tenant_id;
let content: Vec<u8> = "victim1 contents".into();
let relative_remote_path = ctx
.harness
.conf
.remote_path(&ctx.harness.timeline_path(&TIMELINE_ID))
.expect("Failed to construct remote path");
let relative_remote_path = remote_timeline_path(&tenant_id, &TIMELINE_ID);
let remote_timeline_path = ctx.remote_fs_dir.join(relative_remote_path.get_path());
let deletion_prefix = ctx.harness.conf.deletion_prefix();
let generation = Generation::new(0xdeadbeef);
let remote_layer_file_name_1 = format!("{}{}", layer_file_name_1, generation.get_suffix());
// Inject a victim file to remote storage
info!("Writing");
std::fs::create_dir_all(&remote_timeline_path)?;
std::fs::write(
remote_timeline_path.join(layer_file_name_1.to_string()),
remote_timeline_path.join(remote_layer_file_name_1.clone()),
content,
)?;
assert_remote_files(&[&layer_file_name_1.file_name()], &remote_timeline_path);
assert_remote_files(&[&remote_layer_file_name_1], &remote_timeline_path);
// File should still be there after we push it to the queue (we haven't pushed enough to flush anything)
info!("Pushing");
ctx.runtime.block_on(client.push_layers(
tenant_id,
TIMELINE_ID,
[layer_file_name_1.clone()].to_vec(),
[(layer_file_name_1.clone(), generation)].to_vec(),
))?;
assert_remote_files(&[&layer_file_name_1.file_name()], &remote_timeline_path);
assert_remote_files(&[&remote_layer_file_name_1], &remote_timeline_path);
assert_local_files(&[], &deletion_prefix);
// File should still be there after we write a deletion list (we haven't pushed enough to execute anything)
info!("Flushing");
ctx.runtime.block_on(client.flush())?;
assert_remote_files(&[&layer_file_name_1.file_name()], &remote_timeline_path);
assert_remote_files(&[&remote_layer_file_name_1], &remote_timeline_path);
assert_local_files(&["0000000000000001-01.list"], &deletion_prefix);
// File should go away when we execute
@@ -596,24 +596,22 @@ mod test {
let tenant_id = ctx.harness.tenant_id;
let content: Vec<u8> = "victim1 contents".into();
let relative_remote_path = ctx
.harness
.conf
.remote_path(&ctx.harness.timeline_path(&TIMELINE_ID))
.expect("Failed to construct remote path");
let relative_remote_path = remote_timeline_path(&tenant_id, &TIMELINE_ID);
let remote_timeline_path = ctx.remote_fs_dir.join(relative_remote_path.get_path());
let deletion_prefix = ctx.harness.conf.deletion_prefix();
let generation = Generation::new(0xdeadbeef);
let remote_layer_file_name_1 = format!("{}{}", layer_file_name_1, generation.get_suffix());
// Inject a file, delete it, and flush to a deletion list
std::fs::create_dir_all(&remote_timeline_path)?;
std::fs::write(
remote_timeline_path.join(layer_file_name_1.to_string()),
remote_timeline_path.join(remote_layer_file_name_1.clone()),
content,
)?;
ctx.runtime.block_on(client.push_layers(
tenant_id,
TIMELINE_ID,
[layer_file_name_1.clone()].to_vec(),
[(layer_file_name_1.clone(), generation)].to_vec(),
))?;
ctx.runtime.block_on(client.flush())?;
assert_local_files(&["0000000000000001-01.list"], &deletion_prefix);
@@ -638,6 +636,8 @@ mod test {
pub mod mock {
use tracing::info;
use crate::tenant::remote_timeline_client::remote_layer_path;
use super::*;
use std::sync::{
atomic::{AtomicUsize, Ordering},
@@ -652,10 +652,7 @@ pub mod mock {
}
impl MockDeletionQueue {
pub fn new(
remote_storage: Option<GenericRemoteStorage>,
conf: &'static PageServerConf,
) -> Self {
pub fn new(remote_storage: Option<GenericRemoteStorage>) -> Self {
let (tx, mut rx) = tokio::sync::mpsc::channel(16384);
let (tx_pump, mut rx_pump) = tokio::sync::mpsc::channel::<FlushOp>(1);
let (executor_tx, mut executor_rx) = tokio::sync::mpsc::channel(16384);
@@ -703,19 +700,14 @@ pub mod mock {
while let Ok(msg) = rx.try_recv() {
match msg {
FrontendQueueMessage::Delete(op) => {
let timeline_path =
conf.timeline_path(&op.tenant_id, &op.timeline_id);
let mut objects = op.objects;
for layer in op.layers {
let local_path = timeline_path.join(layer.file_name());
let path = match conf.remote_path(&local_path) {
Ok(p) => p,
Err(e) => {
panic!("Can't make a timeline path! {e}");
}
};
objects.push(path);
for (layer, generation) in op.layers {
objects.push(remote_layer_path(
&op.tenant_id,
&op.timeline_id,
&layer,
generation,
));
}
for path in objects {
pageserver/src/deletion_queue/frontend.rs
+10 -12
View File
@@ -12,12 +12,14 @@ use tokio_util::sync::CancellationToken;
use tracing::debug;
use tracing::info;
use tracing::warn;
use utils::generation::Generation;
use utils::id::TenantId;
use utils::id::TimelineId;
use crate::config::PageServerConf;
use crate::metrics::DELETION_QUEUE_ERRORS;
use crate::metrics::DELETION_QUEUE_SUBMITTED;
use crate::tenant::remote_timeline_client::remote_layer_path;
use crate::tenant::storage_layer::LayerFileName;
// The number of keys in a DeletionList before we will proactively persist it
@@ -41,7 +43,7 @@ pub(super) struct DeletionOp {
// `layers` and `objects` are both just lists of objects. `layers` is used if you do not
// have a config object handy to project it to a remote key, and need the consuming worker
// to do it for you.
pub(super) layers: Vec<LayerFileName>,
pub(super) layers: Vec<(LayerFileName, Generation)>,
pub(super) objects: Vec<RemotePath>,
}
@@ -306,18 +308,14 @@ impl FrontendQueueWorker {
op.objects.len()
);
let timeline_path = self.conf.timeline_path(&op.tenant_id, &op.timeline_id);
let mut layer_paths = Vec::new();
for layer in op.layers {
// TODO go directly to remote path without composing local path
let local_path = timeline_path.join(layer.file_name());
let path = match self.conf.remote_path(&local_path) {
Ok(p) => p,
Err(e) => {
panic!("Can't make a timeline path! {e}");
}
};
layer_paths.push(path);
for (layer, generation) in op.layers {
layer_paths.push(remote_layer_path(
&op.tenant_id,
&op.timeline_id,
&layer,
generation,
));
}
self.pending
pageserver/src/tenant.rs
+1 -1
View File
@@ -121,7 +121,7 @@ mod span;
pub mod metadata;
mod par_fsync;
mod remote_timeline_client;
pub mod remote_timeline_client;
pub mod storage_layer;
pub mod config;
pageserver/src/tenant/remote_timeline_client.rs
+9 -12
View File
@@ -646,7 +646,7 @@ impl RemoteTimelineClient {
deletion_queue_client: &DeletionQueueClient,
) -> anyhow::Result<()> {
// Synchronous update of upload queues under mutex
let _with_generations = {
let with_generations = {
let mut guard = self.upload_queue.lock().unwrap();
let upload_queue = guard.initialized_mut()?;
@@ -657,14 +657,14 @@ impl RemoteTimelineClient {
// Decorate our list of names with each name's generation, dropping
// makes that are unexpectedly missing from our metadata.
let with_generations: Vec<_> = names
.iter()
.into_iter()
.filter_map(|name| {
// Remove from latest_files, learning the file's remote generation in the process
let meta = upload_queue.latest_files.remove(name);
if let Some(meta) = meta {
upload_queue.latest_files_changes_since_metadata_upload_scheduled += 1;
Some((name, meta.generation))
Some((name.clone(), meta.generation))
} else {
// This is unexpected: latest_files is meant to be kept up to
// date. We can't delete the layer if we have forgotten what
@@ -693,7 +693,7 @@ impl RemoteTimelineClient {
// Enqueue deletions
deletion_queue_client
.push_layers(self.tenant_id, self.timeline_id, names.to_vec())
.push_layers(self.tenant_id, self.timeline_id, with_generations)
.await?;
Ok(())
}
@@ -826,7 +826,7 @@ impl RemoteTimelineClient {
) -> anyhow::Result<()> {
debug_assert_current_span_has_tenant_and_timeline_id();
let layers: Vec<LayerFileName> = {
let layers: Vec<_> = {
let mut locked = self.upload_queue.lock().unwrap();
let stopped = locked.stopped_mut()?;
@@ -840,20 +840,16 @@ impl RemoteTimelineClient {
.upload_queue_for_deletion
.latest_files
.drain()
.map(|kv| kv.0)
.map(|kv| (kv.0, kv.1.generation))
.collect()
};
let layer_deletion_count = layers.len();
let timeline_path = self.conf.timeline_path(&self.tenant_id, &self.timeline_id);
let layer_paths = layers
.into_iter()
.map(|l| {
let local_path = timeline_path.join(l.file_name());
self.conf
.remote_path(&local_path)
.expect("Timeline path should always convert to remote")
.map(|(layer, generation)| {
remote_layer_path(&self.tenant_id, &self.timeline_id, &layer, generation)
})
.collect();
deletion_queue.push_immediate(layer_paths).await?;
@@ -1711,6 +1707,7 @@ mod tests {
"index_part.json",
],
&remote_timeline_dir,
generation,
);
// Finish uploads and deletions