Compare commits

...

1 Commits

Author SHA1 Message Date
Tristan Partin
3b56c3eecd Remove dead spec parameters
Pageserver and safekeeper connection strings are passed as Postgres
GUCs by the production control plane, so remove the dead spec parameters
and teach neon_local to behave like the production control plane.

Signed-off-by: Tristan Partin <tristan@neon.tech>
2025-05-01 13:23:33 -05:00
5 changed files with 83 additions and 91 deletions

View File

@@ -210,24 +210,20 @@ impl TryFrom<ComputeSpec> for ParsedSpec {
// may be empty. In that case, we need to dig them from the GUCs in the
// cluster.settings field.
let pageserver_connstr = spec
.pageserver_connstring
.clone()
.or_else(|| spec.cluster.settings.find("neon.pageserver_connstring"))
.ok_or("pageserver connstr should be provided")?;
let safekeeper_connstrings = if spec.safekeeper_connstrings.is_empty() {
if matches!(spec.mode, ComputeMode::Primary) {
spec.cluster
.settings
.find("neon.safekeepers")
.ok_or("safekeeper connstrings should be provided")?
.split(',')
.map(|str| str.to_string())
.collect()
} else {
vec![]
}
.cluster
.settings
.find("neon.pageserver_connstring")
.expect("pageserver connstr should be provided");
let safekeeper_connstrings = if matches!(spec.mode, ComputeMode::Primary) {
spec.cluster
.settings
.find("neon.safekeepers")
.ok_or("safekeeper connstrings should be provided")?
.split(',')
.map(|str| str.to_string())
.collect()
} else {
spec.safekeeper_connstrings.clone()
vec![]
};
let storage_auth_token = spec.storage_auth_token.clone();
let tenant_id: TenantId = if let Some(tenant_id) = spec.tenant_id {

View File

@@ -1,5 +1,4 @@
use anyhow::Result;
use std::fmt::Write as FmtWrite;
use std::fs::{File, OpenOptions};
use std::io;
use std::io::Write;
@@ -56,29 +55,9 @@ pub fn write_postgres_conf(
// Add options for connecting to storage
writeln!(file, "# Neon storage settings")?;
if let Some(s) = &spec.pageserver_connstring {
writeln!(file, "neon.pageserver_connstring={}", escape_conf_value(s))?;
}
if let Some(stripe_size) = spec.shard_stripe_size {
writeln!(file, "neon.stripe_size={stripe_size}")?;
}
if !spec.safekeeper_connstrings.is_empty() {
let mut neon_safekeepers_value = String::new();
tracing::info!(
"safekeepers_connstrings is not zero, gen: {:?}",
spec.safekeepers_generation
);
// If generation is given, prepend sk list with g#number:
if let Some(generation) = spec.safekeepers_generation {
write!(neon_safekeepers_value, "g#{}:", generation)?;
}
neon_safekeepers_value.push_str(&spec.safekeeper_connstrings.join(","));
writeln!(
file,
"neon.safekeepers={}",
escape_conf_value(&neon_safekeepers_value)
)?;
}
if let Some(s) = &spec.tenant_id {
writeln!(file, "neon.tenant_id={}", escape_conf_value(&s.to_string()))?;
}
@@ -165,6 +144,21 @@ pub fn write_postgres_conf(
if spec.cluster.settings.is_some() {
writeln!(file, "# Managed by compute_ctl: begin")?;
write!(file, "{}", spec.cluster.settings.as_pg_settings())?;
// If generation is given, prepend sk list with g#number:
if let Some(generation) = spec.safekeepers_generation {
let neon_safekeepers = spec.cluster.settings.find("neon.safekeepers").unwrap();
// Don't try to add it if it already exists
if !neon_safekeepers.starts_with("g#") {
writeln!(
file,
"# Overwriting original neon.safekeepers value to include generation"
)?;
writeln!(
file,
"neon.safekeepers = 'g#{generation}:{neon_safekeepers}'"
)?;
}
}
writeln!(file, "# Managed by compute_ctl: end")?;
}

View File

@@ -366,47 +366,24 @@ impl ComputeNode {
// and can thus use all `max_connections` connection slots. However, that's generally not
// very efficient, so we generally still limit it to a smaller number.
if compute_state.status == ComputeStatus::Init {
// If the settings contain 'max_connections', use that as template
if let Some(config) = spec.cluster.settings.find("max_connections") {
config.parse::<usize>().ok()
} else {
// Otherwise, try to find the setting in the postgresql_conf string
spec.cluster
.postgresql_conf
.iter()
.flat_map(|conf| conf.split("\n"))
.filter_map(|line| {
if !line.contains("max_connections") {
return None;
}
// If the settings contain 'max_connections', use that as a
// template. Otherwise, if we didn't find max_connections, default
// to 10 concurrent connections.
spec.cluster
.settings
.find("max_connections")
.map_or(10, |guc| {
let max_connections = guc.parse::<usize>().unwrap();
let (key, value) = line.split_once("=")?;
let key = key
.trim_start_matches(char::is_whitespace)
.trim_end_matches(char::is_whitespace);
let value = value
.trim_start_matches(char::is_whitespace)
.trim_end_matches(char::is_whitespace);
if key != "max_connections" {
return None;
}
value.parse::<usize>().ok()
})
.next()
}
// If max_connections is present, use at most 1/3rd of that.
// When max_connections is lower than 30, try to use at least 10 connections, but
// never more than max_connections.
.map(|limit| match limit {
0..10 => limit,
10..30 => 10,
30.. => limit / 3,
})
// If we didn't find max_connections, default to 10 concurrent connections.
.unwrap_or(10)
// If max_connections is present, use at most 1/3rd of that.
// When max_connections is lower than 30, try to use at least 10 connections, but
// never more than max_connections.
match max_connections {
0..10 => max_connections,
10..30 => 10,
30.. => max_connections / 3,
}
})
} else {
// state == Running
// Because the cluster is already in the Running state, we should assume users are

View File

@@ -50,8 +50,8 @@ use compute_api::responses::{
ComputeConfig, ComputeCtlConfig, ComputeStatus, ComputeStatusResponse, TlsConfig,
};
use compute_api::spec::{
Cluster, ComputeAudit, ComputeFeature, ComputeMode, ComputeSpec, Database, PgIdent,
RemoteExtSpec, Role,
Cluster, ComputeAudit, ComputeFeature, ComputeMode, ComputeSpec, Database, GenericOption,
PgIdent, RemoteExtSpec, Role,
};
use jsonwebtoken::jwk::{
AlgorithmParameters, CommonParameters, EllipticCurve, Jwk, JwkSet, KeyAlgorithm, KeyOperations,
@@ -711,7 +711,18 @@ impl Endpoint {
} else {
Vec::new()
},
settings: None,
settings: Some(vec![
GenericOption {
name: "neon.pageserver_connstring".to_string(),
value: Some(pageserver_connstring),
vartype: "string".to_string(),
},
GenericOption {
name: "neon.safekeepers".to_string(),
value: Some(safekeeper_connstrings.join(",")),
vartype: "string".to_string(),
},
]),
postgresql_conf: Some(postgresql_conf.clone()),
},
delta_operations: None,
@@ -721,9 +732,7 @@ impl Endpoint {
branch_id: None,
endpoint_id: Some(self.endpoint_id.clone()),
mode: self.mode,
pageserver_connstring: Some(pageserver_connstring),
safekeepers_generation: safekeepers_generation.map(|g| g.into_inner()),
safekeeper_connstrings,
storage_auth_token: auth_token.clone(),
remote_extensions,
pgbouncer_settings: None,
@@ -928,6 +937,8 @@ impl Endpoint {
stripe_size: Option<ShardStripeSize>,
safekeepers: Option<Vec<NodeId>>,
) -> Result<()> {
let mut pg_settings: Vec<GenericOption> = vec![];
let (mut spec, compute_ctl_config) = {
let config_path = self.endpoint_path().join("config.json");
let file = std::fs::File::open(config_path)?;
@@ -958,7 +969,12 @@ impl Endpoint {
let pageserver_connstr = Self::build_pageserver_connstr(&pageservers);
assert!(!pageserver_connstr.is_empty());
spec.pageserver_connstring = Some(pageserver_connstr);
pg_settings.push(GenericOption {
name: "neon.pageserver_connstring".to_string(),
value: Some(pageserver_connstr),
vartype: "string".to_string(),
});
if stripe_size.is_some() {
spec.shard_stripe_size = stripe_size.map(|s| s.0 as usize);
}
@@ -966,9 +982,22 @@ impl Endpoint {
// If safekeepers are not specified, don't change them.
if let Some(safekeepers) = safekeepers {
let safekeeper_connstrings = self.build_safekeepers_connstrs(safekeepers)?;
spec.safekeeper_connstrings = safekeeper_connstrings;
pg_settings.push(GenericOption {
name: "neon.safekeepers".to_string(),
value: Some(safekeeper_connstrings.join(",")),
vartype: "string".to_string(),
});
}
spec.cluster.settings = match spec.cluster.settings {
// Append the new settings to the configured settings
Some(mut settings) => {
settings.append(&mut pg_settings);
Some(settings)
}
None => Some(pg_settings),
};
let client = reqwest::Client::builder()
.timeout(Duration::from_secs(120))
.build()

View File

@@ -90,19 +90,17 @@ pub struct ComputeSpec {
// Information needed to connect to the storage layer.
//
// `tenant_id`, `timeline_id` and `pageserver_connstring` are always needed.
// `tenant_id` and `timeline_id` are always needed.
//
// Depending on `mode`, this can be a primary read-write node, a read-only
// replica, or a read-only node pinned at an older LSN.
// `safekeeper_connstrings` must be set for a primary.
//
// For backwards compatibility, the control plane may leave out all of
// For backwards compatibility, the control plane may leave out both of
// these, and instead set the "neon.tenant_id", "neon.timeline_id",
// etc. GUCs in cluster.settings. TODO: Once the control plane has been
// updated to fill these fields, we can make these non optional.
pub tenant_id: Option<TenantId>,
pub timeline_id: Option<TimelineId>,
pub pageserver_connstring: Option<String>,
// More neon ids that we expose to the compute_ctl
// and to postgres as neon extension GUCs.
@@ -121,8 +119,6 @@ pub struct ComputeSpec {
/// compute_ctl with postgres_ffi.
#[serde(default)]
pub safekeepers_generation: Option<u32>,
#[serde(default)]
pub safekeeper_connstrings: Vec<String>,
#[serde(default)]
pub mode: ComputeMode,