Line data Source code
1 : //! `ComputeSpec` represents the contents of the spec.json file.
2 : //!
3 : //! The spec.json file is used to pass information to 'compute_ctl'. It contains
4 : //! all the information needed to start up the right version of PostgreSQL,
5 : //! and connect it to the storage nodes.
6 : use std::collections::HashMap;
7 :
8 : use indexmap::IndexMap;
9 : use regex::Regex;
10 : use remote_storage::RemotePath;
11 : use serde::{Deserialize, Serialize};
12 : use utils::id::{TenantId, TimelineId};
13 : use utils::lsn::Lsn;
14 :
15 : use crate::responses::TlsConfig;
16 :
17 : /// String type alias representing Postgres identifier and
18 : /// intended to be used for DB / role names.
19 : pub type PgIdent = String;
20 :
21 : /// String type alias representing Postgres extension version
22 : pub type ExtVersion = String;
23 :
24 6 : fn default_reconfigure_concurrency() -> usize {
25 6 : 1
26 6 : }
27 :
28 : /// Cluster spec or configuration represented as an optional number of
29 : /// delta operations + final cluster state description.
30 45 : #[derive(Clone, Debug, Default, Deserialize, Serialize)]
31 : pub struct ComputeSpec {
32 : pub format_version: f32,
33 :
34 : // The control plane also includes a 'timestamp' field in the JSON document,
35 : // but we don't use it for anything. Serde will ignore missing fields when
36 : // deserializing it.
37 : pub operation_uuid: Option<String>,
38 :
39 : /// Compute features to enable. These feature flags are provided, when we
40 : /// know all the details about client's compute, so they cannot be used
41 : /// to change `Empty` compute behavior.
42 : #[serde(default)]
43 : pub features: Vec<ComputeFeature>,
44 :
45 : /// If compute_ctl was passed `--resize-swap-on-bind`, a value of `Some(_)` instructs
46 : /// compute_ctl to `/neonvm/bin/resize-swap` with the given size, when the spec is first
47 : /// received.
48 : ///
49 : /// Both this field and `--resize-swap-on-bind` are required, so that the control plane's
50 : /// spec generation doesn't need to be aware of the actual compute it's running on, while
51 : /// guaranteeing gradual rollout of swap. Otherwise, without `--resize-swap-on-bind`, we could
52 : /// end up trying to resize swap in VMs without it -- or end up *not* resizing swap, thus
53 : /// giving every VM much more swap than it should have (32GiB).
54 : ///
55 : /// Eventually we may remove `--resize-swap-on-bind` and exclusively use `swap_size_bytes` for
56 : /// enabling the swap resizing behavior once rollout is complete.
57 : ///
58 : /// See neondatabase/cloud#12047 for more.
59 : #[serde(default)]
60 : pub swap_size_bytes: Option<u64>,
61 :
62 : /// If compute_ctl was passed `--set-disk-quota-for-fs`, a value of `Some(_)` instructs
63 : /// compute_ctl to run `/neonvm/bin/set-disk-quota` with the given size and fs, when the
64 : /// spec is first received.
65 : ///
66 : /// Both this field and `--set-disk-quota-for-fs` are required, so that the control plane's
67 : /// spec generation doesn't need to be aware of the actual compute it's running on, while
68 : /// guaranteeing gradual rollout of disk quota.
69 : #[serde(default)]
70 : pub disk_quota_bytes: Option<u64>,
71 :
72 : /// Disables the vm-monitor behavior that resizes LFC on upscale/downscale, instead relying on
73 : /// the initial size of LFC.
74 : ///
75 : /// This is intended for use when the LFC size is being overridden from the default but
76 : /// autoscaling is still enabled, and we don't want the vm-monitor to interfere with the custom
77 : /// LFC sizing.
78 : #[serde(default)]
79 : pub disable_lfc_resizing: Option<bool>,
80 :
81 : /// Expected cluster state at the end of transition process.
82 : pub cluster: Cluster,
83 : pub delta_operations: Option<Vec<DeltaOp>>,
84 :
85 : /// An optional hint that can be passed to speed up startup time if we know
86 : /// that no pg catalog mutations (like role creation, database creation,
87 : /// extension creation) need to be done on the actual database to start.
88 : #[serde(default)] // Default false
89 : pub skip_pg_catalog_updates: bool,
90 :
91 : // Information needed to connect to the storage layer.
92 : //
93 : // `tenant_id`, `timeline_id` and `pageserver_connstring` are always needed.
94 : //
95 : // Depending on `mode`, this can be a primary read-write node, a read-only
96 : // replica, or a read-only node pinned at an older LSN.
97 : // `safekeeper_connstrings` must be set for a primary.
98 : //
99 : // For backwards compatibility, the control plane may leave out all of
100 : // these, and instead set the "neon.tenant_id", "neon.timeline_id",
101 : // etc. GUCs in cluster.settings. TODO: Once the control plane has been
102 : // updated to fill these fields, we can make these non optional.
103 : pub tenant_id: Option<TenantId>,
104 : pub timeline_id: Option<TimelineId>,
105 : pub pageserver_connstring: Option<String>,
106 :
107 : /// Safekeeper membership config generation. It is put in
108 : /// neon.safekeepers GUC and serves two purposes:
109 : /// 1) Non zero value forces walproposer to use membership configurations.
110 : /// 2) If walproposer wants to update list of safekeepers to connect to
111 : /// taking them from some safekeeper mconf, it should check what value
112 : /// is newer by comparing the generation.
113 : ///
114 : /// Note: it could be SafekeeperGeneration, but this needs linking
115 : /// compute_ctl with postgres_ffi.
116 : #[serde(default)]
117 : pub safekeepers_generation: Option<u32>,
118 : #[serde(default)]
119 : pub safekeeper_connstrings: Vec<String>,
120 :
121 : #[serde(default)]
122 : pub mode: ComputeMode,
123 :
124 : /// If set, 'storage_auth_token' is used as the password to authenticate to
125 : /// the pageserver and safekeepers.
126 : pub storage_auth_token: Option<String>,
127 :
128 : // information about available remote extensions
129 : pub remote_extensions: Option<RemoteExtSpec>,
130 :
131 : pub pgbouncer_settings: Option<IndexMap<String, String>>,
132 :
133 : // Stripe size for pageserver sharding, in pages
134 : #[serde(default)]
135 : pub shard_stripe_size: Option<usize>,
136 :
137 : /// Local Proxy configuration used for JWT authentication
138 : #[serde(default)]
139 : pub local_proxy_config: Option<LocalProxySpec>,
140 :
141 : /// Number of concurrent connections during the parallel RunInEachDatabase
142 : /// phase of the apply config process.
143 : ///
144 : /// We need a higher concurrency during reconfiguration in case of many DBs,
145 : /// but instance is already running and used by client. We can easily get out of
146 : /// `max_connections` limit, and the current code won't handle that.
147 : ///
148 : /// Default is 1, but also allow control plane to override this value for specific
149 : /// projects. It's also recommended to bump `superuser_reserved_connections` +=
150 : /// `reconfigure_concurrency` for such projects to ensure that we always have
151 : /// enough spare connections for reconfiguration process to succeed.
152 : #[serde(default = "default_reconfigure_concurrency")]
153 : pub reconfigure_concurrency: usize,
154 :
155 : /// If set to true, the compute_ctl will drop all subscriptions before starting the
156 : /// compute. This is needed when we start an endpoint on a branch, so that child
157 : /// would not compete with parent branch subscriptions
158 : /// over the same replication content from publisher.
159 : #[serde(default)] // Default false
160 : pub drop_subscriptions_before_start: bool,
161 :
162 : /// Log level for audit logging:
163 : ///
164 : /// Disabled - no audit logging. This is the default.
165 : /// log - log masked statements to the postgres log using pgaudit extension
166 : /// hipaa - log unmasked statements to the file using pgaudit and pgauditlogtofile extension
167 : ///
168 : /// Extensions should be present in shared_preload_libraries
169 : #[serde(default)]
170 : pub audit_log_level: ComputeAudit,
171 : }
172 :
173 : /// Feature flag to signal `compute_ctl` to enable certain experimental functionality.
174 3 : #[derive(Serialize, Clone, Copy, Debug, Deserialize, PartialEq, Eq)]
175 : #[serde(rename_all = "snake_case")]
176 : pub enum ComputeFeature {
177 : // XXX: Add more feature flags here.
178 : /// Enable the experimental activity monitor logic, which uses `pg_stat_database` to
179 : /// track short-lived connections as user activity.
180 : ActivityMonitorExperimental,
181 :
182 : /// Pre-install and initialize anon extension for every database in the cluster
183 : AnonExtension,
184 :
185 : /// Allow to configure rsyslog for Postgres logs export
186 : PostgresLogsExport,
187 :
188 : /// This is a special feature flag that is used to represent unknown feature flags.
189 : /// Basically all unknown to enum flags are represented as this one. See unit test
190 : /// `parse_unknown_features()` for more details.
191 : #[serde(other)]
192 : UnknownFeature,
193 : }
194 :
195 44 : #[derive(Clone, Debug, Default, Deserialize, Serialize)]
196 : pub struct RemoteExtSpec {
197 : pub public_extensions: Option<Vec<String>>,
198 : pub custom_extensions: Option<Vec<String>>,
199 : pub library_index: HashMap<String, String>,
200 : pub extension_data: HashMap<String, ExtensionData>,
201 : }
202 :
203 30 : #[derive(Clone, Debug, Serialize, Deserialize)]
204 : pub struct ExtensionData {
205 : pub control_data: HashMap<String, String>,
206 : pub archive_path: String,
207 : }
208 :
209 : impl RemoteExtSpec {
210 6 : pub fn get_ext(
211 6 : &self,
212 6 : ext_name: &str,
213 6 : is_library: bool,
214 6 : build_tag: &str,
215 6 : pg_major_version: &str,
216 6 : ) -> anyhow::Result<(String, RemotePath)> {
217 6 : let mut real_ext_name = ext_name;
218 6 : if is_library {
219 : // sometimes library names might have a suffix like
220 : // library.so or library.so.3. We strip this off
221 : // because library_index is based on the name without the file extension
222 1 : let strip_lib_suffix = Regex::new(r"\.so.*").unwrap();
223 1 : let lib_raw_name = strip_lib_suffix.replace(real_ext_name, "").to_string();
224 1 :
225 1 : real_ext_name = self
226 1 : .library_index
227 1 : .get(&lib_raw_name)
228 1 : .ok_or(anyhow::anyhow!("library {} is not found", lib_raw_name))?;
229 5 : }
230 :
231 : // Check if extension is present in public or custom.
232 : // If not, then it is not allowed to be used by this compute.
233 6 : if !self
234 6 : .public_extensions
235 6 : .as_ref()
236 6 : .is_some_and(|exts| exts.iter().any(|e| e == real_ext_name))
237 4 : && !self
238 4 : .custom_extensions
239 4 : .as_ref()
240 4 : .is_some_and(|exts| exts.iter().any(|e| e == real_ext_name))
241 : {
242 3 : return Err(anyhow::anyhow!("extension {} is not found", real_ext_name));
243 3 : }
244 3 :
245 3 : match self.extension_data.get(real_ext_name) {
246 3 : Some(_ext_data) => {
247 3 : // Construct the path to the extension archive
248 3 : // BUILD_TAG/PG_MAJOR_VERSION/extensions/EXTENSION_NAME.tar.zst
249 3 : //
250 3 : // Keep it in sync with path generation in
251 3 : // https://github.com/neondatabase/build-custom-extensions/tree/main
252 3 : let archive_path_str =
253 3 : format!("{build_tag}/{pg_major_version}/extensions/{real_ext_name}.tar.zst");
254 3 : Ok((
255 3 : real_ext_name.to_string(),
256 3 : RemotePath::from_string(&archive_path_str)?,
257 : ))
258 : }
259 0 : None => Err(anyhow::anyhow!(
260 0 : "real_ext_name {} is not found",
261 0 : real_ext_name
262 0 : )),
263 : }
264 6 : }
265 : }
266 :
267 0 : #[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Deserialize, Serialize)]
268 : pub enum ComputeMode {
269 : /// A read-write node
270 : #[default]
271 : Primary,
272 : /// A read-only node, pinned at a particular LSN
273 : Static(Lsn),
274 : /// A read-only node that follows the tip of the branch in hot standby mode
275 : ///
276 : /// Future versions may want to distinguish between replicas with hot standby
277 : /// feedback and other kinds of replication configurations.
278 : Replica,
279 : }
280 :
281 : /// Log level for audit logging
282 : /// Disabled, log, hipaa
283 : /// Default is Disabled
284 0 : #[derive(Clone, Debug, Default, Eq, PartialEq, Deserialize, Serialize)]
285 : pub enum ComputeAudit {
286 : #[default]
287 : Disabled,
288 : Log,
289 : Hipaa,
290 : }
291 :
292 36 : #[derive(Clone, Debug, Default, Deserialize, Serialize, PartialEq, Eq)]
293 : pub struct Cluster {
294 : pub cluster_id: Option<String>,
295 : pub name: Option<String>,
296 : pub state: Option<String>,
297 : pub roles: Vec<Role>,
298 : pub databases: Vec<Database>,
299 :
300 : /// Desired contents of 'postgresql.conf' file. (The 'compute_ctl'
301 : /// tool may add additional settings to the final file.)
302 : pub postgresql_conf: Option<String>,
303 :
304 : /// Additional settings that will be appended to the 'postgresql.conf' file.
305 : pub settings: GenericOptions,
306 : }
307 :
308 : /// Single cluster state changing operation that could not be represented as
309 : /// a static `Cluster` structure. For example:
310 : /// - DROP DATABASE
311 : /// - DROP ROLE
312 : /// - ALTER ROLE name RENAME TO new_name
313 : /// - ALTER DATABASE name RENAME TO new_name
314 60 : #[derive(Clone, Debug, Deserialize, Serialize)]
315 : pub struct DeltaOp {
316 : pub action: String,
317 : pub name: PgIdent,
318 : pub new_name: Option<PgIdent>,
319 : }
320 :
321 : /// Rust representation of Postgres role info with only those fields
322 : /// that matter for us.
323 90 : #[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
324 : pub struct Role {
325 : pub name: PgIdent,
326 : pub encrypted_password: Option<String>,
327 : pub options: GenericOptions,
328 : }
329 :
330 : /// Rust representation of Postgres database info with only those fields
331 : /// that matter for us.
332 42 : #[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
333 : pub struct Database {
334 : pub name: PgIdent,
335 : pub owner: PgIdent,
336 : pub options: GenericOptions,
337 : // These are derived flags, not present in the spec file.
338 : // They are never set by the control plane.
339 : #[serde(skip_deserializing, default)]
340 : pub restrict_conn: bool,
341 : #[serde(skip_deserializing, default)]
342 : pub invalid: bool,
343 : }
344 :
345 : /// Common type representing both SQL statement params with or without value,
346 : /// like `LOGIN` or `OWNER username` in the `CREATE/ALTER ROLE`, and config
347 : /// options like `wal_level = logical`.
348 468 : #[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
349 : pub struct GenericOption {
350 : pub name: String,
351 : pub value: Option<String>,
352 : pub vartype: String,
353 : }
354 :
355 : /// Optional collection of `GenericOption`'s. Type alias allows us to
356 : /// declare a `trait` on it.
357 : pub type GenericOptions = Option<Vec<GenericOption>>;
358 :
359 : /// Configured the local_proxy application with the relevant JWKS and roles it should
360 : /// use for authorizing connect requests using JWT.
361 0 : #[derive(Clone, Debug, Deserialize, Serialize)]
362 : pub struct LocalProxySpec {
363 : #[serde(default)]
364 : #[serde(skip_serializing_if = "Option::is_none")]
365 : pub jwks: Option<Vec<JwksSettings>>,
366 : #[serde(default)]
367 : #[serde(skip_serializing_if = "Option::is_none")]
368 : pub tls: Option<TlsConfig>,
369 : }
370 :
371 0 : #[derive(Clone, Debug, Deserialize, Serialize)]
372 : pub struct JwksSettings {
373 : pub id: String,
374 : pub role_names: Vec<String>,
375 : pub jwks_url: String,
376 : pub provider_name: String,
377 : pub jwt_audience: Option<String>,
378 : }
379 :
380 : #[cfg(test)]
381 : mod tests {
382 : use std::fs::File;
383 :
384 : use super::*;
385 :
386 : #[test]
387 1 : fn allow_installing_remote_extensions() {
388 1 : let rspec: RemoteExtSpec = serde_json::from_value(serde_json::json!({
389 1 : "public_extensions": null,
390 1 : "custom_extensions": null,
391 1 : "library_index": {},
392 1 : "extension_data": {},
393 1 : }))
394 1 : .unwrap();
395 1 :
396 1 : rspec
397 1 : .get_ext("ext", false, "latest", "v17")
398 1 : .expect_err("Extension should not be found");
399 1 :
400 1 : let rspec: RemoteExtSpec = serde_json::from_value(serde_json::json!({
401 1 : "public_extensions": [],
402 1 : "custom_extensions": null,
403 1 : "library_index": {},
404 1 : "extension_data": {},
405 1 : }))
406 1 : .unwrap();
407 1 :
408 1 : rspec
409 1 : .get_ext("ext", false, "latest", "v17")
410 1 : .expect_err("Extension should not be found");
411 1 :
412 1 : let rspec: RemoteExtSpec = serde_json::from_value(serde_json::json!({
413 1 : "public_extensions": [],
414 1 : "custom_extensions": [],
415 1 : "library_index": {
416 1 : "ext": "ext"
417 1 : },
418 1 : "extension_data": {
419 1 : "ext": {
420 1 : "control_data": {
421 1 : "ext.control": ""
422 1 : },
423 1 : "archive_path": ""
424 1 : }
425 1 : },
426 1 : }))
427 1 : .unwrap();
428 1 :
429 1 : rspec
430 1 : .get_ext("ext", false, "latest", "v17")
431 1 : .expect_err("Extension should not be found");
432 1 :
433 1 : let rspec: RemoteExtSpec = serde_json::from_value(serde_json::json!({
434 1 : "public_extensions": [],
435 1 : "custom_extensions": ["ext"],
436 1 : "library_index": {
437 1 : "ext": "ext"
438 1 : },
439 1 : "extension_data": {
440 1 : "ext": {
441 1 : "control_data": {
442 1 : "ext.control": ""
443 1 : },
444 1 : "archive_path": ""
445 1 : }
446 1 : },
447 1 : }))
448 1 : .unwrap();
449 1 :
450 1 : rspec
451 1 : .get_ext("ext", false, "latest", "v17")
452 1 : .expect("Extension should be found");
453 1 :
454 1 : let rspec: RemoteExtSpec = serde_json::from_value(serde_json::json!({
455 1 : "public_extensions": ["ext"],
456 1 : "custom_extensions": [],
457 1 : "library_index": {
458 1 : "extlib": "ext",
459 1 : },
460 1 : "extension_data": {
461 1 : "ext": {
462 1 : "control_data": {
463 1 : "ext.control": ""
464 1 : },
465 1 : "archive_path": ""
466 1 : }
467 1 : },
468 1 : }))
469 1 : .unwrap();
470 1 :
471 1 : rspec
472 1 : .get_ext("ext", false, "latest", "v17")
473 1 : .expect("Extension should be found");
474 1 :
475 1 : // test library index for the case when library name
476 1 : // doesn't match the extension name
477 1 : rspec
478 1 : .get_ext("extlib", true, "latest", "v17")
479 1 : .expect("Library should be found");
480 1 : }
481 :
482 : #[test]
483 1 : fn parse_spec_file() {
484 1 : let file = File::open("tests/cluster_spec.json").unwrap();
485 1 : let spec: ComputeSpec = serde_json::from_reader(file).unwrap();
486 1 :
487 1 : // Features list defaults to empty vector.
488 1 : assert!(spec.features.is_empty());
489 :
490 : // Reconfigure concurrency defaults to 1.
491 1 : assert_eq!(spec.reconfigure_concurrency, 1);
492 1 : }
493 :
494 : #[test]
495 1 : fn parse_unknown_fields() {
496 1 : // Forward compatibility test
497 1 : let file = File::open("tests/cluster_spec.json").unwrap();
498 1 : let mut json: serde_json::Value = serde_json::from_reader(file).unwrap();
499 1 : let ob = json.as_object_mut().unwrap();
500 1 : ob.insert("unknown_field_123123123".into(), "hello".into());
501 1 : let _spec: ComputeSpec = serde_json::from_value(json).unwrap();
502 1 : }
503 :
504 : #[test]
505 1 : fn parse_unknown_features() {
506 1 : // Test that unknown feature flags do not cause any errors.
507 1 : let file = File::open("tests/cluster_spec.json").unwrap();
508 1 : let mut json: serde_json::Value = serde_json::from_reader(file).unwrap();
509 1 : let ob = json.as_object_mut().unwrap();
510 1 :
511 1 : // Add unknown feature flags.
512 1 : let features = vec!["foo_bar_feature", "baz_feature"];
513 1 : ob.insert("features".into(), features.into());
514 1 :
515 1 : let spec: ComputeSpec = serde_json::from_value(json).unwrap();
516 1 :
517 1 : assert!(spec.features.len() == 2);
518 1 : assert!(spec.features.contains(&ComputeFeature::UnknownFeature));
519 1 : assert_eq!(spec.features, vec![ComputeFeature::UnknownFeature; 2]);
520 1 : }
521 :
522 : #[test]
523 1 : fn parse_known_features() {
524 1 : // Test that we can properly parse known feature flags.
525 1 : let file = File::open("tests/cluster_spec.json").unwrap();
526 1 : let mut json: serde_json::Value = serde_json::from_reader(file).unwrap();
527 1 : let ob = json.as_object_mut().unwrap();
528 1 :
529 1 : // Add known feature flags.
530 1 : let features = vec!["activity_monitor_experimental"];
531 1 : ob.insert("features".into(), features.into());
532 1 :
533 1 : let spec: ComputeSpec = serde_json::from_value(json).unwrap();
534 1 :
535 1 : assert_eq!(
536 1 : spec.features,
537 1 : vec![ComputeFeature::ActivityMonitorExperimental]
538 1 : );
539 1 : }
540 : }
|
