rust/neon
1
0
Fork 0
mirror of https://github.com/neondatabase/neon.git synced 2026-05-28 02:20:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

storcon: add per-tenant rate limiting for API requests (#10924)

Browse Source 
## Problem

Incoming requests often take the service lock, and sometimes even do
database transactions. That creates a risk that a rogue client can
starve the controller of the ability to do its primary job of
reconciling tenants to an available state.

## Summary of changes

* Use the `governor` crate to rate limit tenant requests at 10 requests
per second. This is ~10-100x lower than the worst "attack" we've seen
from a client bug. Admin APIs are not rate limited.
* Add a `storage_controller_http_request_rate_limited` histogram for
rate limited requests.
* Log a warning every 10 seconds for rate limited tenants.

The rate limiter is parametrized on TenantId, because the kinds of
client bug we're protecting against generally happen within tenant
scope, and the rates should be somewhat stable: we expect the global
rate of requests to increase as we do more work, but we do not expect
the rate of requests to one tenant to increase.

---------

Co-authored-by: John Spray <john@neon.tech>
This commit is contained in:
Erik Grinaker
2025-03-03 23:04:59 +01:00
committed by GitHub
parent 6d0976dad5
commit 65addfc524
8 changed files with 156 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
storage_controller/Cargo.toml
View File

@@ -21,6 +21,7 @@ clap.workspace = true
cron.workspace = true
fail.workspace = true
futures.workspace = true
governor.workspace = true
hex.workspace = true
hyper0.workspace = true
humantime.workspace = true

69
storage_controller/src/http.rs
View File

@@ -1,5 +1,5 @@
use std::str::FromStr;
use std::sync::Arc;
use std::sync::{Arc, LazyLock};
use std::time::{Duration, Instant};
use anyhow::Context;
@@ -33,6 +33,7 @@ use pageserver_api::upcall_api::{ReAttachRequest, ValidateRequest};
use pageserver_client::{BlockUnblock, mgmt_api};
use routerify::Middleware;
use tokio_util::sync::CancellationToken;
use tracing::warn;
use utils::auth::{Scope, SwappableJwtAuth};
use utils::id::{NodeId, TenantId, TimelineId};
@@ -49,6 +50,7 @@ use crate::service::{LeadershipStatus, RECONCILE_TIMEOUT, STARTUP_RECONCILE_TIME
pub struct HttpState {
service: Arc<crate::service::Service>,
auth: Option<Arc<SwappableJwtAuth>>,
rate_limiter: governor::DefaultKeyedRateLimiter<TenantId>,
neon_metrics: NeonMetrics,
allowlist_routes: &'static [&'static str],
}
@@ -59,9 +61,11 @@ impl HttpState {
auth: Option<Arc<SwappableJwtAuth>>,
build_info: BuildInfo,
) -> Self {
let quota = governor::Quota::per_second(service.get_config().tenant_rate_limit);
Self {
service,
auth,
rate_limiter: governor::RateLimiter::keyed(quota),
neon_metrics: NeonMetrics::new(build_info),
allowlist_routes: &[
"/status",
@@ -82,6 +86,40 @@ fn get_state(request: &Request<Body>) -> &HttpState {
.as_ref()
}
/// Rate limits tenant requests.
///
/// TODO: this should be a request middleware, but requires us to extract the tenant ID from
/// different URLs in a systematic way.
///
/// TODO: consider returning a 429 response if these start piling up.
async fn maybe_rate_limit(request: &Request<Body>, tenant_id: TenantId) {
// Check if the tenant should be rate-limited.
let rate_limiter = &get_state(request).rate_limiter;
if rate_limiter.check_key(&tenant_id).is_ok() {
return;
}
// Measure the rate limiting delay.
let _timer = METRICS_REGISTRY
.metrics_group
.storage_controller_http_request_rate_limited
.start_timer();
// Log rate limited tenants once every 10 seconds.
static LOG_RATE_LIMITER: LazyLock<governor::DefaultKeyedRateLimiter<TenantId>> =
LazyLock::new(|| {
let quota = governor::Quota::with_period(Duration::from_secs(10)).unwrap();
governor::RateLimiter::keyed(quota)
});
if LOG_RATE_LIMITER.check_key(&tenant_id).is_ok() {
warn!("tenant {tenant_id} is rate limited")
}
// Wait for quota.
rate_limiter.until_key_ready(&tenant_id).await;
}
/// Pageserver calls into this on startup, to learn which tenants it should attach
async fn handle_re_attach(req: Request<Body>) -> Result<Response<Body>, ApiError> {
check_permissions(&req, Scope::GenerationsApi)?;
@@ -247,6 +285,7 @@ async fn handle_tenant_config_get(
) -> Result<Response<Body>, ApiError> {
let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
check_permissions(&req, Scope::PageServerApi)?;
maybe_rate_limit(&req, tenant_id).await;
match maybe_forward(req).await {
ForwardOutcome::Forwarded(res) => {
@@ -264,6 +303,7 @@ async fn handle_tenant_time_travel_remote_storage(
) -> Result<Response<Body>, ApiError> {
let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
check_permissions(&req, Scope::PageServerApi)?;
maybe_rate_limit(&req, tenant_id).await;
let mut req = match maybe_forward(req).await {
ForwardOutcome::Forwarded(res) => {
@@ -311,6 +351,7 @@ async fn handle_tenant_secondary_download(
) -> Result<Response<Body>, ApiError> {
let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
let wait = parse_query_param(&req, "wait_ms")?.map(Duration::from_millis);
maybe_rate_limit(&req, tenant_id).await;
match maybe_forward(req).await {
ForwardOutcome::Forwarded(res) => {
@@ -329,6 +370,7 @@ async fn handle_tenant_delete(
) -> Result<Response<Body>, ApiError> {
let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
check_permissions(&req, Scope::PageServerApi)?;
maybe_rate_limit(&req, tenant_id).await;
match maybe_forward(req).await {
ForwardOutcome::Forwarded(res) => {
@@ -356,6 +398,7 @@ async fn handle_tenant_timeline_create(
) -> Result<Response<Body>, ApiError> {
let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
check_permissions(&req, Scope::PageServerApi)?;
maybe_rate_limit(&req, tenant_id).await;
let mut req = match maybe_forward(req).await {
ForwardOutcome::Forwarded(res) => {
@@ -381,6 +424,7 @@ async fn handle_tenant_timeline_delete(
let timeline_id: TimelineId = parse_request_param(&req, "timeline_id")?;
check_permissions(&req, Scope::PageServerApi)?;
maybe_rate_limit(&req, tenant_id).await;
match maybe_forward(req).await {
ForwardOutcome::Forwarded(res) => {
@@ -457,6 +501,7 @@ async fn handle_tenant_timeline_archival_config(
let timeline_id: TimelineId = parse_request_param(&req, "timeline_id")?;
check_permissions(&req, Scope::PageServerApi)?;
maybe_rate_limit(&req, tenant_id).await;
let mut req = match maybe_forward(req).await {
ForwardOutcome::Forwarded(res) => {
@@ -482,6 +527,7 @@ async fn handle_tenant_timeline_detach_ancestor(
let timeline_id: TimelineId = parse_request_param(&req, "timeline_id")?;
check_permissions(&req, Scope::PageServerApi)?;
maybe_rate_limit(&req, tenant_id).await;
match maybe_forward(req).await {
ForwardOutcome::Forwarded(res) => {
@@ -504,6 +550,7 @@ async fn handle_tenant_timeline_block_unblock_gc(
) -> Result<Response<Body>, ApiError> {
let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
check_permissions(&req, Scope::PageServerApi)?;
maybe_rate_limit(&req, tenant_id).await;
let timeline_id: TimelineId = parse_request_param(&req, "timeline_id")?;
@@ -521,6 +568,7 @@ async fn handle_tenant_timeline_download_heatmap_layers(
let tenant_shard_id: TenantShardId = parse_request_param(&req, "tenant_shard_id")?;
check_permissions(&req, Scope::PageServerApi)?;
maybe_rate_limit(&req, tenant_shard_id.tenant_id).await;
let timeline_id: TimelineId = parse_request_param(&req, "timeline_id")?;
let concurrency: Option<usize> = parse_query_param(&req, "concurrency")?;
@@ -550,6 +598,7 @@ async fn handle_tenant_timeline_passthrough(
) -> Result<Response<Body>, ApiError> {
let tenant_or_shard_id: TenantShardId = parse_request_param(&req, "tenant_id")?;
check_permissions(&req, Scope::PageServerApi)?;
maybe_rate_limit(&req, tenant_or_shard_id.tenant_id).await;
let req = match maybe_forward(req).await {
ForwardOutcome::Forwarded(res) => {
@@ -654,6 +703,7 @@ async fn handle_tenant_locate(
let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
check_permissions(&req, Scope::Admin)?;
// NB: don't rate limit: admin operation.
match maybe_forward(req).await {
ForwardOutcome::Forwarded(res) => {
@@ -669,9 +719,9 @@ async fn handle_tenant_describe(
service: Arc<Service>,
req: Request<Body>,
) -> Result<Response<Body>, ApiError> {
check_permissions(&req, Scope::Scrubber)?;
let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
check_permissions(&req, Scope::Scrubber)?;
// NB: don't rate limit: scrubber operation.
match maybe_forward(req).await {
ForwardOutcome::Forwarded(res) => {
@@ -1006,6 +1056,7 @@ async fn handle_tenant_shard_split(
req: Request<Body>,
) -> Result<Response<Body>, ApiError> {
check_permissions(&req, Scope::Admin)?;
// NB: don't rate limit: admin operation.
let mut req = match maybe_forward(req).await {
ForwardOutcome::Forwarded(res) => {
@@ -1028,6 +1079,7 @@ async fn handle_tenant_shard_migrate(
req: Request<Body>,
) -> Result<Response<Body>, ApiError> {
check_permissions(&req, Scope::Admin)?;
// NB: don't rate limit: admin operation.
let mut req = match maybe_forward(req).await {
ForwardOutcome::Forwarded(res) => {
@@ -1051,6 +1103,7 @@ async fn handle_tenant_shard_migrate_secondary(
req: Request<Body>,
) -> Result<Response<Body>, ApiError> {
check_permissions(&req, Scope::Admin)?;
// NB: don't rate limit: admin operation.
let mut req = match maybe_forward(req).await {
ForwardOutcome::Forwarded(res) => {
@@ -1074,6 +1127,7 @@ async fn handle_tenant_shard_cancel_reconcile(
req: Request<Body>,
) -> Result<Response<Body>, ApiError> {
check_permissions(&req, Scope::Admin)?;
// NB: don't rate limit: admin operation.
let req = match maybe_forward(req).await {
ForwardOutcome::Forwarded(res) => {
@@ -1093,6 +1147,7 @@ async fn handle_tenant_shard_cancel_reconcile(
async fn handle_tenant_update_policy(req: Request<Body>) -> Result<Response<Body>, ApiError> {
check_permissions(&req, Scope::Admin)?;
// NB: don't rate limit: admin operation.
let mut req = match maybe_forward(req).await {
ForwardOutcome::Forwarded(res) => {
@@ -1148,9 +1203,9 @@ async fn handle_step_down(req: Request<Body>) -> Result<Response<Body>, ApiError
}
async fn handle_tenant_drop(req: Request<Body>) -> Result<Response<Body>, ApiError> {
check_permissions(&req, Scope::PageServerApi)?;
let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
check_permissions(&req, Scope::PageServerApi)?;
maybe_rate_limit(&req, tenant_id).await;
let req = match maybe_forward(req).await {
ForwardOutcome::Forwarded(res) => {
@@ -1165,9 +1220,9 @@ async fn handle_tenant_drop(req: Request<Body>) -> Result<Response<Body>, ApiErr
}
async fn handle_tenant_import(req: Request<Body>) -> Result<Response<Body>, ApiError> {
check_permissions(&req, Scope::PageServerApi)?;
let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
check_permissions(&req, Scope::PageServerApi)?;
maybe_rate_limit(&req, tenant_id).await;
let req = match maybe_forward(req).await {
ForwardOutcome::Forwarded(res) => {

6
storage_controller/src/main.rs
View File

@@ -1,3 +1,4 @@
use std::num::NonZeroU32;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
@@ -98,6 +99,10 @@ struct Cli {
#[arg(long)]
priority_reconciler_concurrency: Option<usize>,
/// Tenant API rate limit, as requests per second per tenant.
#[arg(long, default_value = "10")]
tenant_rate_limit: NonZeroU32,
/// How long to wait for the initial database connection to be available.
#[arg(long, default_value = "5s")]
db_connect_timeout: humantime::Duration,
@@ -339,6 +344,7 @@ async fn async_main() -> anyhow::Result<()> {
priority_reconciler_concurrency: args
.priority_reconciler_concurrency
.unwrap_or(PRIORITY_RECONCILER_CONCURRENCY_DEFAULT),
tenant_rate_limit: args.tenant_rate_limit,
split_threshold: args.split_threshold,
neon_local_repo_dir: args.neon_local_repo_dir,
max_secondary_lag_bytes: args.max_secondary_lag_bytes,

4
storage_controller/src/metrics.rs
View File

@@ -76,6 +76,10 @@ pub(crate) struct StorageControllerMetricGroup {
pub(crate) storage_controller_http_request_latency:
measured::HistogramVec<HttpRequestLatencyLabelGroupSet, 5>,
/// HTTP rate limiting latency across all tenants and endpoints
#[metric(metadata = histogram::Thresholds::exponential_buckets(0.1, 10.0))]
pub(crate) storage_controller_http_request_rate_limited: measured::Histogram<10>,
/// Count of HTTP requests to the pageserver that resulted in an error,
/// broken down by the pageserver node id, request name and method
pub(crate) storage_controller_pageserver_request_error:

5
storage_controller/src/service.rs
View File

@@ -5,6 +5,7 @@ use std::borrow::Cow;
use std::cmp::Ordering;
use std::collections::{BTreeMap, HashMap, HashSet};
use std::error::Error;
use std::num::NonZeroU32;
use std::ops::Deref;
use std::path::PathBuf;
use std::str::FromStr;
@@ -365,6 +366,10 @@ pub struct Config {
/// How many high-priority Reconcilers may be spawned concurrently
pub priority_reconciler_concurrency: usize,
/// How many API requests per second to allow per tenant, across all
/// tenant-scoped API endpoints. Further API requests queue until ready.
pub tenant_rate_limit: NonZeroU32,
/// How large must a shard grow in bytes before we split it?
/// None disables auto-splitting.
pub split_threshold: Option<u64>,