rust/neon
1
0
Fork 0
mirror of https://github.com/neondatabase/neon.git synced 2026-07-06 05:30:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

Implement layer eviction and download

Browse Source
This commit is contained in:
Kirill Bulatov
2023-01-30 15:31:13 +02:00
parent aae4382613
commit d446befb51
2 changed files with 141 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
pageserver/src/http/routes.rs
View File

@@ -7,6 +7,7 @@ use pageserver_api::models::DownloadRemoteLayersTaskSpawnRequest;
use remote_storage::GenericRemoteStorage;
use tokio_util::sync::CancellationToken;
use tracing::*;
use utils::http::request::get_request_param;
use super::models::{
StatusResponse, TenantConfigRequest, TenantCreateRequest, TenantCreateResponse, TenantInfo,
@@ -556,9 +557,7 @@ async fn tenant_size_handler(request: Request<Body>) -> Result<Response<Body>, A
)
}
async fn timeline_layer_map_info_handler(
request: Request<Body>,
) -> Result<Response<Body>, ApiError> {
async fn layer_map_info_handler(request: Request<Body>) -> Result<Response<Body>, ApiError> {
let tenant_id: TenantId = parse_request_param(&request, "tenant_id")?;
let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
check_permission(&request, Some(tenant_id))?;
@@ -575,6 +574,58 @@ async fn timeline_layer_map_info_handler(
json_response(StatusCode::OK, layer_map_info)
}
async fn layer_download_handler(request: Request<Body>) -> Result<Response<Body>, ApiError> {
let tenant_id: TenantId = parse_request_param(&request, "tenant_id")?;
let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
let layer_file_name = get_request_param(&request, "layer_file_name")?;
check_permission(&request, Some(tenant_id))?;
let tenant = mgr::get_tenant(tenant_id, true)
.await
.map_err(ApiError::NotFound)?;
let timeline = tenant
.get_timeline(timeline_id, true)
.map_err(ApiError::NotFound)?;
let downloaded = timeline
.download_layer(layer_file_name)
.await
.map_err(ApiError::InternalServerError)?;
let status = if downloaded {
StatusCode::OK
} else {
StatusCode::NOT_MODIFIED
};
json_response(status, ())
}
async fn evict_timeline_layer_handler(request: Request<Body>) -> Result<Response<Body>, ApiError> {
let tenant_id: TenantId = parse_request_param(&request, "tenant_id")?;
let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
let layer_file_name = get_request_param(&request, "layer_file_name")?;
check_permission(&request, Some(tenant_id))?;
let tenant = mgr::get_tenant(tenant_id, true)
.await
.map_err(ApiError::NotFound)?;
let timeline = tenant
.get_timeline(timeline_id, true)
.map_err(ApiError::NotFound)?;
let evicted = timeline
.evict_layer(layer_file_name)
.await
.map_err(ApiError::InternalServerError)?;
let status = if evicted {
StatusCode::OK
} else {
StatusCode::NOT_MODIFIED
};
json_response(status, ())
}
// Helper function to standardize the error messages we produce on bad durations
//
// Intended to be used with anyhow's `with_context`, e.g.:
@@ -1007,7 +1058,15 @@ pub fn make_router(
)
.get(
"/v1/tenant/:tenant_id/timeline/:timeline_id/layer",
timeline_layer_map_info_handler,
layer_map_info_handler,
)
.get(
"/v1/tenant/:tenant_id/timeline/:timeline_id/layer/:layer_file_name",
layer_download_handler,
)
.delete(
"/v1/tenant/:tenant_id/timeline/:timeline_id/layer/:layer_file_name",
evict_timeline_layer_handler,
)
.any(handler_404))
}

78
pageserver/src/tenant/timeline.rs
View File

@@ -839,6 +839,73 @@ impl Timeline {
historic_layers,
}
}
pub async fn download_layer(&self, layer_file_name: &str) -> anyhow::Result<bool> {
if self.remote_client.is_none() {
return Ok(false);
}
let Some(remote_layer) = self.find_layer(layer_file_name)
.and_then(PersistentLayer::downcast_remote_layer)
else {
return Ok(false);
};
self.download_remote_layer(remote_layer)
.await
.map(|()| true)
}
pub async fn evict_layer(&self, layer_file_name: &str) -> anyhow::Result<bool> {
let Some(remote_client) = &self.remote_client
else {
return Ok(false);
};
let Some(local_layer) = self.find_layer(layer_file_name)
else {
return Ok(false);
};
if local_layer.is_remote_layer() {
return Ok(false);
}
// ensure the current layer is uploaded for sure
remote_client
.wait_completion()
.await
.context("wait for layer upload ops to complete")?;
let layer_metadata = LayerFileMetadata::new(
local_layer
.file_size()
.expect("Local layer should have a file size"),
);
let new_remote_layer = Arc::new(match local_layer.filename() {
LayerFileName::Image(image_name) => RemoteLayer::new_img(
self.tenant_id,
self.timeline_id,
&image_name,
&layer_metadata,
),
LayerFileName::Delta(delta_name) => RemoteLayer::new_delta(
self.tenant_id,
self.timeline_id,
&delta_name,
&layer_metadata,
),
#[cfg(test)]
LayerFileName::Test(_) => unreachable!(),
});
let mut layers = self.layers.write().unwrap();
let mut updates = layers.batch_update();
local_layer.delete()?;
updates.remove_historic(local_layer);
updates.insert_historic(new_remote_layer);
updates.flush();
drop(layers);
Ok(true)
}
}
// Private functions
@@ -1628,6 +1695,17 @@ impl Timeline {
Err(e) => error!("Failed to compute current logical size for metrics update: {e:?}"),
}
}
fn find_layer(&self, layer_file_name: &str) -> Option<Arc<dyn PersistentLayer>> {
for historic_layer in self.layers.read().unwrap().iter_historic_layers() {
let historic_layer_name = historic_layer.filename().file_name();
if layer_file_name == historic_layer_name {
return Some(historic_layer);
}
}
None
}
}
type TraversalId = String;