Line data Source code
1 : pub(crate) mod split_state;
2 : use std::collections::HashMap;
3 : use std::str::FromStr;
4 : use std::time::Duration;
5 : use std::time::Instant;
6 :
7 : use self::split_state::SplitState;
8 : use diesel::pg::PgConnection;
9 : use diesel::prelude::*;
10 : use diesel::Connection;
11 : use itertools::Itertools;
12 : use pageserver_api::controller_api::AvailabilityZone;
13 : use pageserver_api::controller_api::MetadataHealthRecord;
14 : use pageserver_api::controller_api::ShardSchedulingPolicy;
15 : use pageserver_api::controller_api::{NodeSchedulingPolicy, PlacementPolicy};
16 : use pageserver_api::models::TenantConfig;
17 : use pageserver_api::shard::ShardConfigError;
18 : use pageserver_api::shard::ShardIdentity;
19 : use pageserver_api::shard::ShardStripeSize;
20 : use pageserver_api::shard::{ShardCount, ShardNumber, TenantShardId};
21 : use serde::{Deserialize, Serialize};
22 : use utils::generation::Generation;
23 : use utils::id::{NodeId, TenantId};
24 :
25 : use crate::metrics::{
26 : DatabaseQueryErrorLabelGroup, DatabaseQueryLatencyLabelGroup, METRICS_REGISTRY,
27 : };
28 : use crate::node::Node;
29 :
30 : use diesel_migrations::{embed_migrations, EmbeddedMigrations};
31 : const MIGRATIONS: EmbeddedMigrations = embed_migrations!("./migrations");
32 :
33 : /// ## What do we store?
34 : ///
35 : /// The storage controller service does not store most of its state durably.
36 : ///
37 : /// The essential things to store durably are:
38 : /// - generation numbers, as these must always advance monotonically to ensure data safety.
39 : /// - Tenant's PlacementPolicy and TenantConfig, as the source of truth for these is something external.
40 : /// - Node's scheduling policies, as the source of truth for these is something external.
41 : ///
42 : /// Other things we store durably as an implementation detail:
43 : /// - Node's host/port: this could be avoided it we made nodes emit a self-registering heartbeat,
44 : /// but it is operationally simpler to make this service the authority for which nodes
45 : /// it talks to.
46 : ///
47 : /// ## Performance/efficiency
48 : ///
49 : /// The storage controller service does not go via the database for most things: there are
50 : /// a couple of places where we must, and where efficiency matters:
51 : /// - Incrementing generation numbers: the Reconciler has to wait for this to complete
52 : /// before it can attach a tenant, so this acts as a bound on how fast things like
53 : /// failover can happen.
54 : /// - Pageserver re-attach: we will increment many shards' generations when this happens,
55 : /// so it is important to avoid e.g. issuing O(N) queries.
56 : ///
57 : /// Database calls relating to nodes have low performance requirements, as they are very rarely
58 : /// updated, and reads of nodes are always from memory, not the database. We only require that
59 : /// we can UPDATE a node's scheduling mode reasonably quickly to mark a bad node offline.
60 : pub struct Persistence {
61 : connection_pool: diesel::r2d2::Pool<diesel::r2d2::ConnectionManager<PgConnection>>,
62 : }
63 :
64 : /// Legacy format, for use in JSON compat objects in test environment
65 0 : #[derive(Serialize, Deserialize)]
66 : struct JsonPersistence {
67 : tenants: HashMap<TenantShardId, TenantShardPersistence>,
68 : }
69 :
70 0 : #[derive(thiserror::Error, Debug)]
71 : pub(crate) enum DatabaseError {
72 : #[error(transparent)]
73 : Query(#[from] diesel::result::Error),
74 : #[error(transparent)]
75 : Connection(#[from] diesel::result::ConnectionError),
76 : #[error(transparent)]
77 : ConnectionPool(#[from] r2d2::Error),
78 : #[error("Logical error: {0}")]
79 : Logical(String),
80 : #[error("Migration error: {0}")]
81 : Migration(String),
82 : }
83 :
84 : #[derive(measured::FixedCardinalityLabel, Copy, Clone)]
85 : pub(crate) enum DatabaseOperation {
86 : InsertNode,
87 : UpdateNode,
88 : DeleteNode,
89 : ListNodes,
90 : BeginShardSplit,
91 : CompleteShardSplit,
92 : AbortShardSplit,
93 : Detach,
94 : ReAttach,
95 : IncrementGeneration,
96 : TenantGenerations,
97 : ShardGenerations,
98 : ListTenantShards,
99 : InsertTenantShards,
100 : UpdateTenantShard,
101 : DeleteTenant,
102 : UpdateTenantConfig,
103 : UpdateMetadataHealth,
104 : ListMetadataHealth,
105 : ListMetadataHealthUnhealthy,
106 : ListMetadataHealthOutdated,
107 : GetLeader,
108 : UpdateLeader,
109 : SetPreferredAzs,
110 : }
111 :
112 : #[must_use]
113 : pub(crate) enum AbortShardSplitStatus {
114 : /// We aborted the split in the database by reverting to the parent shards
115 : Aborted,
116 : /// The split had already been persisted.
117 : Complete,
118 : }
119 :
120 : pub(crate) type DatabaseResult<T> = Result<T, DatabaseError>;
121 :
122 : /// Some methods can operate on either a whole tenant or a single shard
123 : pub(crate) enum TenantFilter {
124 : Tenant(TenantId),
125 : Shard(TenantShardId),
126 : }
127 :
128 : /// Represents the results of looking up generation+pageserver for the shards of a tenant
129 : pub(crate) struct ShardGenerationState {
130 : pub(crate) tenant_shard_id: TenantShardId,
131 : pub(crate) generation: Option<Generation>,
132 : pub(crate) generation_pageserver: Option<NodeId>,
133 : }
134 :
135 : impl Persistence {
136 : // The default postgres connection limit is 100. We use up to 99, to leave one free for a human admin under
137 : // normal circumstances. This assumes we have exclusive use of the database cluster to which we connect.
138 : pub const MAX_CONNECTIONS: u32 = 99;
139 :
140 : // We don't want to keep a lot of connections alive: close them down promptly if they aren't being used.
141 : const IDLE_CONNECTION_TIMEOUT: Duration = Duration::from_secs(10);
142 : const MAX_CONNECTION_LIFETIME: Duration = Duration::from_secs(60);
143 :
144 0 : pub fn new(database_url: String) -> Self {
145 0 : let manager = diesel::r2d2::ConnectionManager::<PgConnection>::new(database_url);
146 0 :
147 0 : // We will use a connection pool: this is primarily to _limit_ our connection count, rather than to optimize time
148 0 : // to execute queries (database queries are not generally on latency-sensitive paths).
149 0 : let connection_pool = diesel::r2d2::Pool::builder()
150 0 : .max_size(Self::MAX_CONNECTIONS)
151 0 : .max_lifetime(Some(Self::MAX_CONNECTION_LIFETIME))
152 0 : .idle_timeout(Some(Self::IDLE_CONNECTION_TIMEOUT))
153 0 : // Always keep at least one connection ready to go
154 0 : .min_idle(Some(1))
155 0 : .test_on_check_out(true)
156 0 : .build(manager)
157 0 : .expect("Could not build connection pool");
158 0 :
159 0 : Self { connection_pool }
160 0 : }
161 :
162 : /// A helper for use during startup, where we would like to tolerate concurrent restarts of the
163 : /// database and the storage controller, therefore the database might not be available right away
164 0 : pub async fn await_connection(
165 0 : database_url: &str,
166 0 : timeout: Duration,
167 0 : ) -> Result<(), diesel::ConnectionError> {
168 0 : let started_at = Instant::now();
169 : loop {
170 0 : match PgConnection::establish(database_url) {
171 : Ok(_) => {
172 0 : tracing::info!("Connected to database.");
173 0 : return Ok(());
174 : }
175 0 : Err(e) => {
176 0 : if started_at.elapsed() > timeout {
177 0 : return Err(e);
178 : } else {
179 0 : tracing::info!("Database not yet available, waiting... ({e})");
180 0 : tokio::time::sleep(Duration::from_millis(100)).await;
181 : }
182 : }
183 : }
184 : }
185 0 : }
186 :
187 : /// Execute the diesel migrations that are built into this binary
188 0 : pub(crate) async fn migration_run(&self) -> DatabaseResult<()> {
189 : use diesel_migrations::{HarnessWithOutput, MigrationHarness};
190 :
191 0 : self.with_conn(move |conn| -> DatabaseResult<()> {
192 0 : HarnessWithOutput::write_to_stdout(conn)
193 0 : .run_pending_migrations(MIGRATIONS)
194 0 : .map(|_| ())
195 0 : .map_err(|e| DatabaseError::Migration(e.to_string()))
196 0 : })
197 0 : .await
198 0 : }
199 :
200 : /// Wraps `with_conn` in order to collect latency and error metrics
201 0 : async fn with_measured_conn<F, R>(&self, op: DatabaseOperation, func: F) -> DatabaseResult<R>
202 0 : where
203 0 : F: Fn(&mut PgConnection) -> DatabaseResult<R> + Send + 'static,
204 0 : R: Send + 'static,
205 0 : {
206 0 : let latency = &METRICS_REGISTRY
207 0 : .metrics_group
208 0 : .storage_controller_database_query_latency;
209 0 : let _timer = latency.start_timer(DatabaseQueryLatencyLabelGroup { operation: op });
210 :
211 0 : let res = self.with_conn(func).await;
212 :
213 0 : if let Err(err) = &res {
214 0 : let error_counter = &METRICS_REGISTRY
215 0 : .metrics_group
216 0 : .storage_controller_database_query_error;
217 0 : error_counter.inc(DatabaseQueryErrorLabelGroup {
218 0 : error_type: err.error_label(),
219 0 : operation: op,
220 0 : })
221 0 : }
222 :
223 0 : res
224 0 : }
225 :
226 : /// Call the provided function in a tokio blocking thread, with a Diesel database connection.
227 0 : async fn with_conn<F, R>(&self, func: F) -> DatabaseResult<R>
228 0 : where
229 0 : F: Fn(&mut PgConnection) -> DatabaseResult<R> + Send + 'static,
230 0 : R: Send + 'static,
231 0 : {
232 : // A generous allowance for how many times we may retry serializable transactions
233 : // before giving up. This is not expected to be hit: it is a defensive measure in case we
234 : // somehow engineer a situation where duelling transactions might otherwise live-lock.
235 : const MAX_RETRIES: usize = 128;
236 :
237 0 : let mut conn = self.connection_pool.get()?;
238 0 : tokio::task::spawn_blocking(move || -> DatabaseResult<R> {
239 0 : let mut retry_count = 0;
240 : loop {
241 0 : match conn.build_transaction().serializable().run(|c| func(c)) {
242 0 : Ok(r) => break Ok(r),
243 : Err(
244 0 : err @ DatabaseError::Query(diesel::result::Error::DatabaseError(
245 0 : diesel::result::DatabaseErrorKind::SerializationFailure,
246 0 : _,
247 0 : )),
248 0 : ) => {
249 0 : retry_count += 1;
250 0 : if retry_count > MAX_RETRIES {
251 0 : tracing::error!(
252 0 : "Exceeded max retries on SerializationFailure errors: {err:?}"
253 : );
254 0 : break Err(err);
255 : } else {
256 : // Retry on serialization errors: these are expected, because even though our
257 : // transactions don't fight for the same rows, they will occasionally collide
258 : // on index pages (e.g. increment_generation for unrelated shards can collide)
259 0 : tracing::debug!(
260 0 : "Retrying transaction on serialization failure {err:?}"
261 : );
262 0 : continue;
263 : }
264 : }
265 0 : Err(e) => break Err(e),
266 : }
267 : }
268 0 : })
269 0 : .await
270 0 : .expect("Task panic")
271 0 : }
272 :
273 : /// When a node is first registered, persist it before using it for anything
274 0 : pub(crate) async fn insert_node(&self, node: &Node) -> DatabaseResult<()> {
275 0 : let np = node.to_persistent();
276 0 : self.with_measured_conn(
277 0 : DatabaseOperation::InsertNode,
278 0 : move |conn| -> DatabaseResult<()> {
279 0 : diesel::insert_into(crate::schema::nodes::table)
280 0 : .values(&np)
281 0 : .execute(conn)?;
282 0 : Ok(())
283 0 : },
284 0 : )
285 0 : .await
286 0 : }
287 :
288 : /// At startup, populate the list of nodes which our shards may be placed on
289 0 : pub(crate) async fn list_nodes(&self) -> DatabaseResult<Vec<NodePersistence>> {
290 0 : let nodes: Vec<NodePersistence> = self
291 0 : .with_measured_conn(
292 0 : DatabaseOperation::ListNodes,
293 0 : move |conn| -> DatabaseResult<_> {
294 0 : Ok(crate::schema::nodes::table.load::<NodePersistence>(conn)?)
295 0 : },
296 0 : )
297 0 : .await?;
298 :
299 0 : tracing::info!("list_nodes: loaded {} nodes", nodes.len());
300 :
301 0 : Ok(nodes)
302 0 : }
303 :
304 0 : pub(crate) async fn update_node(
305 0 : &self,
306 0 : input_node_id: NodeId,
307 0 : input_scheduling: NodeSchedulingPolicy,
308 0 : ) -> DatabaseResult<()> {
309 : use crate::schema::nodes::dsl::*;
310 0 : let updated = self
311 0 : .with_measured_conn(DatabaseOperation::UpdateNode, move |conn| {
312 0 : let updated = diesel::update(nodes)
313 0 : .filter(node_id.eq(input_node_id.0 as i64))
314 0 : .set((scheduling_policy.eq(String::from(input_scheduling)),))
315 0 : .execute(conn)?;
316 0 : Ok(updated)
317 0 : })
318 0 : .await?;
319 :
320 0 : if updated != 1 {
321 0 : Err(DatabaseError::Logical(format!(
322 0 : "Node {node_id:?} not found for update",
323 0 : )))
324 : } else {
325 0 : Ok(())
326 : }
327 0 : }
328 :
329 : /// At startup, load the high level state for shards, such as their config + policy. This will
330 : /// be enriched at runtime with state discovered on pageservers.
331 0 : pub(crate) async fn list_tenant_shards(&self) -> DatabaseResult<Vec<TenantShardPersistence>> {
332 0 : self.with_measured_conn(
333 0 : DatabaseOperation::ListTenantShards,
334 0 : move |conn| -> DatabaseResult<_> {
335 0 : Ok(crate::schema::tenant_shards::table.load::<TenantShardPersistence>(conn)?)
336 0 : },
337 0 : )
338 0 : .await
339 0 : }
340 :
341 : /// Tenants must be persisted before we schedule them for the first time. This enables us
342 : /// to correctly retain generation monotonicity, and the externally provided placement policy & config.
343 0 : pub(crate) async fn insert_tenant_shards(
344 0 : &self,
345 0 : shards: Vec<TenantShardPersistence>,
346 0 : ) -> DatabaseResult<()> {
347 : use crate::schema::metadata_health;
348 : use crate::schema::tenant_shards;
349 :
350 0 : let now = chrono::Utc::now();
351 0 :
352 0 : let metadata_health_records = shards
353 0 : .iter()
354 0 : .map(|t| MetadataHealthPersistence {
355 0 : tenant_id: t.tenant_id.clone(),
356 0 : shard_number: t.shard_number,
357 0 : shard_count: t.shard_count,
358 0 : healthy: true,
359 0 : last_scrubbed_at: now,
360 0 : })
361 0 : .collect::<Vec<_>>();
362 0 :
363 0 : self.with_measured_conn(
364 0 : DatabaseOperation::InsertTenantShards,
365 0 : move |conn| -> DatabaseResult<()> {
366 0 : diesel::insert_into(tenant_shards::table)
367 0 : .values(&shards)
368 0 : .execute(conn)?;
369 :
370 0 : diesel::insert_into(metadata_health::table)
371 0 : .values(&metadata_health_records)
372 0 : .execute(conn)?;
373 0 : Ok(())
374 0 : },
375 0 : )
376 0 : .await
377 0 : }
378 :
379 : /// Ordering: call this _after_ deleting the tenant on pageservers, but _before_ dropping state for
380 : /// the tenant from memory on this server.
381 0 : pub(crate) async fn delete_tenant(&self, del_tenant_id: TenantId) -> DatabaseResult<()> {
382 : use crate::schema::tenant_shards::dsl::*;
383 0 : self.with_measured_conn(
384 0 : DatabaseOperation::DeleteTenant,
385 0 : move |conn| -> DatabaseResult<()> {
386 0 : // `metadata_health` status (if exists) is also deleted based on the cascade behavior.
387 0 : diesel::delete(tenant_shards)
388 0 : .filter(tenant_id.eq(del_tenant_id.to_string()))
389 0 : .execute(conn)?;
390 0 : Ok(())
391 0 : },
392 0 : )
393 0 : .await
394 0 : }
395 :
396 0 : pub(crate) async fn delete_node(&self, del_node_id: NodeId) -> DatabaseResult<()> {
397 : use crate::schema::nodes::dsl::*;
398 0 : self.with_measured_conn(
399 0 : DatabaseOperation::DeleteNode,
400 0 : move |conn| -> DatabaseResult<()> {
401 0 : diesel::delete(nodes)
402 0 : .filter(node_id.eq(del_node_id.0 as i64))
403 0 : .execute(conn)?;
404 :
405 0 : Ok(())
406 0 : },
407 0 : )
408 0 : .await
409 0 : }
410 :
411 : /// When a tenant invokes the /re-attach API, this function is responsible for doing an efficient
412 : /// batched increment of the generations of all tenants whose generation_pageserver is equal to
413 : /// the node that called /re-attach.
414 0 : #[tracing::instrument(skip_all, fields(node_id))]
415 : pub(crate) async fn re_attach(
416 : &self,
417 : input_node_id: NodeId,
418 : ) -> DatabaseResult<HashMap<TenantShardId, Generation>> {
419 : use crate::schema::nodes::dsl::scheduling_policy;
420 : use crate::schema::nodes::dsl::*;
421 : use crate::schema::tenant_shards::dsl::*;
422 : let updated = self
423 0 : .with_measured_conn(DatabaseOperation::ReAttach, move |conn| {
424 0 : let rows_updated = diesel::update(tenant_shards)
425 0 : .filter(generation_pageserver.eq(input_node_id.0 as i64))
426 0 : .set(generation.eq(generation + 1))
427 0 : .execute(conn)?;
428 :
429 0 : tracing::info!("Incremented {} tenants' generations", rows_updated);
430 :
431 : // TODO: UPDATE+SELECT in one query
432 :
433 0 : let updated = tenant_shards
434 0 : .filter(generation_pageserver.eq(input_node_id.0 as i64))
435 0 : .select(TenantShardPersistence::as_select())
436 0 : .load(conn)?;
437 :
438 : // If the node went through a drain and restart phase before re-attaching,
439 : // then reset it's node scheduling policy to active.
440 0 : diesel::update(nodes)
441 0 : .filter(node_id.eq(input_node_id.0 as i64))
442 0 : .filter(
443 0 : scheduling_policy
444 0 : .eq(String::from(NodeSchedulingPolicy::PauseForRestart))
445 0 : .or(scheduling_policy.eq(String::from(NodeSchedulingPolicy::Draining)))
446 0 : .or(scheduling_policy.eq(String::from(NodeSchedulingPolicy::Filling))),
447 0 : )
448 0 : .set(scheduling_policy.eq(String::from(NodeSchedulingPolicy::Active)))
449 0 : .execute(conn)?;
450 :
451 0 : Ok(updated)
452 0 : })
453 : .await?;
454 :
455 : let mut result = HashMap::new();
456 : for tsp in updated {
457 : let tenant_shard_id = TenantShardId {
458 : tenant_id: TenantId::from_str(tsp.tenant_id.as_str())
459 0 : .map_err(|e| DatabaseError::Logical(format!("Malformed tenant id: {e}")))?,
460 : shard_number: ShardNumber(tsp.shard_number as u8),
461 : shard_count: ShardCount::new(tsp.shard_count as u8),
462 : };
463 :
464 : let Some(g) = tsp.generation else {
465 : // If the generation_pageserver column was non-NULL, then the generation column should also be non-NULL:
466 : // we only set generation_pageserver when setting generation.
467 : return Err(DatabaseError::Logical(
468 : "Generation should always be set after incrementing".to_string(),
469 : ));
470 : };
471 : result.insert(tenant_shard_id, Generation::new(g as u32));
472 : }
473 :
474 : Ok(result)
475 : }
476 :
477 : /// Reconciler calls this immediately before attaching to a new pageserver, to acquire a unique, monotonically
478 : /// advancing generation number. We also store the NodeId for which the generation was issued, so that in
479 : /// [`Self::re_attach`] we can do a bulk UPDATE on the generations for that node.
480 0 : pub(crate) async fn increment_generation(
481 0 : &self,
482 0 : tenant_shard_id: TenantShardId,
483 0 : node_id: NodeId,
484 0 : ) -> anyhow::Result<Generation> {
485 : use crate::schema::tenant_shards::dsl::*;
486 0 : let updated = self
487 0 : .with_measured_conn(DatabaseOperation::IncrementGeneration, move |conn| {
488 0 : let updated = diesel::update(tenant_shards)
489 0 : .filter(tenant_id.eq(tenant_shard_id.tenant_id.to_string()))
490 0 : .filter(shard_number.eq(tenant_shard_id.shard_number.0 as i32))
491 0 : .filter(shard_count.eq(tenant_shard_id.shard_count.literal() as i32))
492 0 : .set((
493 0 : generation.eq(generation + 1),
494 0 : generation_pageserver.eq(node_id.0 as i64),
495 0 : ))
496 0 : // TODO: only returning() the generation column
497 0 : .returning(TenantShardPersistence::as_returning())
498 0 : .get_result(conn)?;
499 :
500 0 : Ok(updated)
501 0 : })
502 0 : .await?;
503 :
504 : // Generation is always non-null in the rseult: if the generation column had been NULL, then we
505 : // should have experienced an SQL Confilict error while executing a query that tries to increment it.
506 0 : debug_assert!(updated.generation.is_some());
507 0 : let Some(g) = updated.generation else {
508 0 : return Err(DatabaseError::Logical(
509 0 : "Generation should always be set after incrementing".to_string(),
510 0 : )
511 0 : .into());
512 : };
513 :
514 0 : Ok(Generation::new(g as u32))
515 0 : }
516 :
517 : /// When we want to call out to the running shards for a tenant, e.g. during timeline CRUD operations,
518 : /// we need to know where the shard is attached, _and_ the generation, so that we can re-check the generation
519 : /// afterwards to confirm that our timeline CRUD operation is truly persistent (it must have happened in the
520 : /// latest generation)
521 : ///
522 : /// If the tenant doesn't exist, an empty vector is returned.
523 : ///
524 : /// Output is sorted by shard number
525 0 : pub(crate) async fn tenant_generations(
526 0 : &self,
527 0 : filter_tenant_id: TenantId,
528 0 : ) -> Result<Vec<ShardGenerationState>, DatabaseError> {
529 : use crate::schema::tenant_shards::dsl::*;
530 0 : let rows = self
531 0 : .with_measured_conn(DatabaseOperation::TenantGenerations, move |conn| {
532 0 : let result = tenant_shards
533 0 : .filter(tenant_id.eq(filter_tenant_id.to_string()))
534 0 : .select(TenantShardPersistence::as_select())
535 0 : .order(shard_number)
536 0 : .load(conn)?;
537 0 : Ok(result)
538 0 : })
539 0 : .await?;
540 :
541 0 : Ok(rows
542 0 : .into_iter()
543 0 : .map(|p| ShardGenerationState {
544 0 : tenant_shard_id: p
545 0 : .get_tenant_shard_id()
546 0 : .expect("Corrupt tenant shard id in database"),
547 0 : generation: p.generation.map(|g| Generation::new(g as u32)),
548 0 : generation_pageserver: p.generation_pageserver.map(|n| NodeId(n as u64)),
549 0 : })
550 0 : .collect())
551 0 : }
552 :
553 : /// Read the generation number of specific tenant shards
554 : ///
555 : /// Output is unsorted. Output may not include values for all inputs, if they are missing in the database.
556 0 : pub(crate) async fn shard_generations(
557 0 : &self,
558 0 : mut tenant_shard_ids: impl Iterator<Item = &TenantShardId>,
559 0 : ) -> Result<Vec<(TenantShardId, Option<Generation>)>, DatabaseError> {
560 0 : let mut rows = Vec::with_capacity(tenant_shard_ids.size_hint().0);
561 :
562 : // We will chunk our input to avoid composing arbitrarily long `IN` clauses. Typically we are
563 : // called with a single digit number of IDs, but in principle we could be called with tens
564 : // of thousands (all the shards on one pageserver) from the generation validation API.
565 0 : loop {
566 0 : // A modest hardcoded chunk size to handle typical cases in a single query but never generate particularly
567 0 : // large query strings.
568 0 : let chunk_ids = tenant_shard_ids.by_ref().take(32);
569 0 :
570 0 : // Compose a comma separated list of tuples for matching on (tenant_id, shard_number, shard_count)
571 0 : let in_clause = chunk_ids
572 0 : .map(|tsid| {
573 0 : format!(
574 0 : "('{}', {}, {})",
575 0 : tsid.tenant_id, tsid.shard_number.0, tsid.shard_count.0
576 0 : )
577 0 : })
578 0 : .join(",");
579 0 :
580 0 : // We are done when our iterator gives us nothing to filter on
581 0 : if in_clause.is_empty() {
582 0 : break;
583 0 : }
584 :
585 0 : let chunk_rows = self
586 0 : .with_measured_conn(DatabaseOperation::ShardGenerations, move |conn| {
587 : // diesel doesn't support multi-column IN queries, so we compose raw SQL. No escaping is required because
588 : // the inputs are strongly typed and cannot carry any user-supplied raw string content.
589 0 : let result : Vec<TenantShardPersistence> = diesel::sql_query(
590 0 : format!("SELECT * from tenant_shards where (tenant_id, shard_number, shard_count) in ({in_clause});").as_str()
591 0 : ).load(conn)?;
592 :
593 0 : Ok(result)
594 0 : })
595 0 : .await?;
596 0 : rows.extend(chunk_rows.into_iter())
597 : }
598 :
599 0 : Ok(rows
600 0 : .into_iter()
601 0 : .map(|tsp| {
602 0 : (
603 0 : tsp.get_tenant_shard_id()
604 0 : .expect("Bad tenant ID in database"),
605 0 : tsp.generation.map(|g| Generation::new(g as u32)),
606 0 : )
607 0 : })
608 0 : .collect())
609 0 : }
610 :
611 : #[allow(non_local_definitions)]
612 : /// For use when updating a persistent property of a tenant, such as its config or placement_policy.
613 : ///
614 : /// Do not use this for settting generation, unless in the special onboarding code path (/location_config)
615 : /// API: use [`Self::increment_generation`] instead. Setting the generation via this route is a one-time thing
616 : /// that we only do the first time a tenant is set to an attached policy via /location_config.
617 0 : pub(crate) async fn update_tenant_shard(
618 0 : &self,
619 0 : tenant: TenantFilter,
620 0 : input_placement_policy: Option<PlacementPolicy>,
621 0 : input_config: Option<TenantConfig>,
622 0 : input_generation: Option<Generation>,
623 0 : input_scheduling_policy: Option<ShardSchedulingPolicy>,
624 0 : ) -> DatabaseResult<()> {
625 : use crate::schema::tenant_shards::dsl::*;
626 :
627 0 : self.with_measured_conn(DatabaseOperation::UpdateTenantShard, move |conn| {
628 0 : let query = match tenant {
629 0 : TenantFilter::Shard(tenant_shard_id) => diesel::update(tenant_shards)
630 0 : .filter(tenant_id.eq(tenant_shard_id.tenant_id.to_string()))
631 0 : .filter(shard_number.eq(tenant_shard_id.shard_number.0 as i32))
632 0 : .filter(shard_count.eq(tenant_shard_id.shard_count.literal() as i32))
633 0 : .into_boxed(),
634 0 : TenantFilter::Tenant(input_tenant_id) => diesel::update(tenant_shards)
635 0 : .filter(tenant_id.eq(input_tenant_id.to_string()))
636 0 : .into_boxed(),
637 : };
638 :
639 0 : #[derive(AsChangeset)]
640 : #[diesel(table_name = crate::schema::tenant_shards)]
641 : struct ShardUpdate {
642 : generation: Option<i32>,
643 : placement_policy: Option<String>,
644 : config: Option<String>,
645 : scheduling_policy: Option<String>,
646 : }
647 :
648 0 : let update = ShardUpdate {
649 0 : generation: input_generation.map(|g| g.into().unwrap() as i32),
650 0 : placement_policy: input_placement_policy
651 0 : .as_ref()
652 0 : .map(|p| serde_json::to_string(&p).unwrap()),
653 0 : config: input_config
654 0 : .as_ref()
655 0 : .map(|c| serde_json::to_string(&c).unwrap()),
656 0 : scheduling_policy: input_scheduling_policy
657 0 : .map(|p| serde_json::to_string(&p).unwrap()),
658 0 : };
659 0 :
660 0 : query.set(update).execute(conn)?;
661 :
662 0 : Ok(())
663 0 : })
664 0 : .await?;
665 :
666 0 : Ok(())
667 0 : }
668 :
669 0 : pub(crate) async fn set_tenant_shard_preferred_azs(
670 0 : &self,
671 0 : preferred_azs: Vec<(TenantShardId, AvailabilityZone)>,
672 0 : ) -> DatabaseResult<Vec<(TenantShardId, AvailabilityZone)>> {
673 : use crate::schema::tenant_shards::dsl::*;
674 :
675 0 : self.with_measured_conn(DatabaseOperation::SetPreferredAzs, move |conn| {
676 0 : let mut shards_updated = Vec::default();
677 :
678 0 : for (tenant_shard_id, preferred_az) in preferred_azs.iter() {
679 0 : let updated = diesel::update(tenant_shards)
680 0 : .filter(tenant_id.eq(tenant_shard_id.tenant_id.to_string()))
681 0 : .filter(shard_number.eq(tenant_shard_id.shard_number.0 as i32))
682 0 : .filter(shard_count.eq(tenant_shard_id.shard_count.literal() as i32))
683 0 : .set(preferred_az_id.eq(preferred_az.0.clone()))
684 0 : .execute(conn)?;
685 :
686 0 : if updated == 1 {
687 0 : shards_updated.push((*tenant_shard_id, preferred_az.clone()));
688 0 : }
689 : }
690 :
691 0 : Ok(shards_updated)
692 0 : })
693 0 : .await
694 0 : }
695 :
696 0 : pub(crate) async fn detach(&self, tenant_shard_id: TenantShardId) -> anyhow::Result<()> {
697 : use crate::schema::tenant_shards::dsl::*;
698 0 : self.with_measured_conn(DatabaseOperation::Detach, move |conn| {
699 0 : let updated = diesel::update(tenant_shards)
700 0 : .filter(tenant_id.eq(tenant_shard_id.tenant_id.to_string()))
701 0 : .filter(shard_number.eq(tenant_shard_id.shard_number.0 as i32))
702 0 : .filter(shard_count.eq(tenant_shard_id.shard_count.literal() as i32))
703 0 : .set((
704 0 : generation_pageserver.eq(Option::<i64>::None),
705 0 : placement_policy.eq(serde_json::to_string(&PlacementPolicy::Detached).unwrap()),
706 0 : ))
707 0 : .execute(conn)?;
708 :
709 0 : Ok(updated)
710 0 : })
711 0 : .await?;
712 :
713 0 : Ok(())
714 0 : }
715 :
716 : // When we start shard splitting, we must durably mark the tenant so that
717 : // on restart, we know that we must go through recovery.
718 : //
719 : // We create the child shards here, so that they will be available for increment_generation calls
720 : // if some pageserver holding a child shard needs to restart before the overall tenant split is complete.
721 0 : pub(crate) async fn begin_shard_split(
722 0 : &self,
723 0 : old_shard_count: ShardCount,
724 0 : split_tenant_id: TenantId,
725 0 : parent_to_children: Vec<(TenantShardId, Vec<TenantShardPersistence>)>,
726 0 : ) -> DatabaseResult<()> {
727 : use crate::schema::tenant_shards::dsl::*;
728 0 : self.with_measured_conn(DatabaseOperation::BeginShardSplit, move |conn| -> DatabaseResult<()> {
729 : // Mark parent shards as splitting
730 :
731 0 : let updated = diesel::update(tenant_shards)
732 0 : .filter(tenant_id.eq(split_tenant_id.to_string()))
733 0 : .filter(shard_count.eq(old_shard_count.literal() as i32))
734 0 : .set((splitting.eq(1),))
735 0 : .execute(conn)?;
736 0 : if u8::try_from(updated)
737 0 : .map_err(|_| DatabaseError::Logical(
738 0 : format!("Overflow existing shard count {} while splitting", updated))
739 0 : )? != old_shard_count.count() {
740 : // Perhaps a deletion or another split raced with this attempt to split, mutating
741 : // the parent shards that we intend to split. In this case the split request should fail.
742 0 : return Err(DatabaseError::Logical(
743 0 : format!("Unexpected existing shard count {updated} when preparing tenant for split (expected {})", old_shard_count.count())
744 0 : ));
745 0 : }
746 0 :
747 0 : // FIXME: spurious clone to sidestep closure move rules
748 0 : let parent_to_children = parent_to_children.clone();
749 :
750 : // Insert child shards
751 0 : for (parent_shard_id, children) in parent_to_children {
752 0 : let mut parent = crate::schema::tenant_shards::table
753 0 : .filter(tenant_id.eq(parent_shard_id.tenant_id.to_string()))
754 0 : .filter(shard_number.eq(parent_shard_id.shard_number.0 as i32))
755 0 : .filter(shard_count.eq(parent_shard_id.shard_count.literal() as i32))
756 0 : .load::<TenantShardPersistence>(conn)?;
757 0 : let parent = if parent.len() != 1 {
758 0 : return Err(DatabaseError::Logical(format!(
759 0 : "Parent shard {parent_shard_id} not found"
760 0 : )));
761 : } else {
762 0 : parent.pop().unwrap()
763 : };
764 0 : for mut shard in children {
765 : // Carry the parent's generation into the child
766 0 : shard.generation = parent.generation;
767 0 :
768 0 : debug_assert!(shard.splitting == SplitState::Splitting);
769 0 : diesel::insert_into(tenant_shards)
770 0 : .values(shard)
771 0 : .execute(conn)?;
772 : }
773 : }
774 :
775 0 : Ok(())
776 0 : })
777 0 : .await
778 0 : }
779 :
780 : // When we finish shard splitting, we must atomically clean up the old shards
781 : // and insert the new shards, and clear the splitting marker.
782 0 : pub(crate) async fn complete_shard_split(
783 0 : &self,
784 0 : split_tenant_id: TenantId,
785 0 : old_shard_count: ShardCount,
786 0 : ) -> DatabaseResult<()> {
787 : use crate::schema::tenant_shards::dsl::*;
788 0 : self.with_measured_conn(
789 0 : DatabaseOperation::CompleteShardSplit,
790 0 : move |conn| -> DatabaseResult<()> {
791 0 : // Drop parent shards
792 0 : diesel::delete(tenant_shards)
793 0 : .filter(tenant_id.eq(split_tenant_id.to_string()))
794 0 : .filter(shard_count.eq(old_shard_count.literal() as i32))
795 0 : .execute(conn)?;
796 :
797 : // Clear sharding flag
798 0 : let updated = diesel::update(tenant_shards)
799 0 : .filter(tenant_id.eq(split_tenant_id.to_string()))
800 0 : .set((splitting.eq(0),))
801 0 : .execute(conn)?;
802 0 : debug_assert!(updated > 0);
803 :
804 0 : Ok(())
805 0 : },
806 0 : )
807 0 : .await
808 0 : }
809 :
810 : /// Used when the remote part of a shard split failed: we will revert the database state to have only
811 : /// the parent shards, with SplitState::Idle.
812 0 : pub(crate) async fn abort_shard_split(
813 0 : &self,
814 0 : split_tenant_id: TenantId,
815 0 : new_shard_count: ShardCount,
816 0 : ) -> DatabaseResult<AbortShardSplitStatus> {
817 : use crate::schema::tenant_shards::dsl::*;
818 0 : self.with_measured_conn(
819 0 : DatabaseOperation::AbortShardSplit,
820 0 : move |conn| -> DatabaseResult<AbortShardSplitStatus> {
821 : // Clear the splitting state on parent shards
822 0 : let updated = diesel::update(tenant_shards)
823 0 : .filter(tenant_id.eq(split_tenant_id.to_string()))
824 0 : .filter(shard_count.ne(new_shard_count.literal() as i32))
825 0 : .set((splitting.eq(0),))
826 0 : .execute(conn)?;
827 :
828 : // Parent shards are already gone: we cannot abort.
829 0 : if updated == 0 {
830 0 : return Ok(AbortShardSplitStatus::Complete);
831 0 : }
832 0 :
833 0 : // Sanity check: if parent shards were present, their cardinality should
834 0 : // be less than the number of child shards.
835 0 : if updated >= new_shard_count.count() as usize {
836 0 : return Err(DatabaseError::Logical(format!(
837 0 : "Unexpected parent shard count {updated} while aborting split to \
838 0 : count {new_shard_count:?} on tenant {split_tenant_id}"
839 0 : )));
840 0 : }
841 0 :
842 0 : // Erase child shards
843 0 : diesel::delete(tenant_shards)
844 0 : .filter(tenant_id.eq(split_tenant_id.to_string()))
845 0 : .filter(shard_count.eq(new_shard_count.literal() as i32))
846 0 : .execute(conn)?;
847 :
848 0 : Ok(AbortShardSplitStatus::Aborted)
849 0 : },
850 0 : )
851 0 : .await
852 0 : }
853 :
854 : /// Stores all the latest metadata health updates durably. Updates existing entry on conflict.
855 : ///
856 : /// **Correctness:** `metadata_health_updates` should all belong the tenant shards managed by the storage controller.
857 : #[allow(dead_code)]
858 0 : pub(crate) async fn update_metadata_health_records(
859 0 : &self,
860 0 : healthy_records: Vec<MetadataHealthPersistence>,
861 0 : unhealthy_records: Vec<MetadataHealthPersistence>,
862 0 : now: chrono::DateTime<chrono::Utc>,
863 0 : ) -> DatabaseResult<()> {
864 : use crate::schema::metadata_health::dsl::*;
865 :
866 0 : self.with_measured_conn(
867 0 : DatabaseOperation::UpdateMetadataHealth,
868 0 : move |conn| -> DatabaseResult<_> {
869 0 : diesel::insert_into(metadata_health)
870 0 : .values(&healthy_records)
871 0 : .on_conflict((tenant_id, shard_number, shard_count))
872 0 : .do_update()
873 0 : .set((healthy.eq(true), last_scrubbed_at.eq(now)))
874 0 : .execute(conn)?;
875 :
876 0 : diesel::insert_into(metadata_health)
877 0 : .values(&unhealthy_records)
878 0 : .on_conflict((tenant_id, shard_number, shard_count))
879 0 : .do_update()
880 0 : .set((healthy.eq(false), last_scrubbed_at.eq(now)))
881 0 : .execute(conn)?;
882 0 : Ok(())
883 0 : },
884 0 : )
885 0 : .await
886 0 : }
887 :
888 : /// Lists all the metadata health records.
889 : #[allow(dead_code)]
890 0 : pub(crate) async fn list_metadata_health_records(
891 0 : &self,
892 0 : ) -> DatabaseResult<Vec<MetadataHealthPersistence>> {
893 0 : self.with_measured_conn(
894 0 : DatabaseOperation::ListMetadataHealth,
895 0 : move |conn| -> DatabaseResult<_> {
896 0 : Ok(
897 0 : crate::schema::metadata_health::table
898 0 : .load::<MetadataHealthPersistence>(conn)?,
899 : )
900 0 : },
901 0 : )
902 0 : .await
903 0 : }
904 :
905 : /// Lists all the metadata health records that is unhealthy.
906 : #[allow(dead_code)]
907 0 : pub(crate) async fn list_unhealthy_metadata_health_records(
908 0 : &self,
909 0 : ) -> DatabaseResult<Vec<MetadataHealthPersistence>> {
910 : use crate::schema::metadata_health::dsl::*;
911 0 : self.with_measured_conn(
912 0 : DatabaseOperation::ListMetadataHealthUnhealthy,
913 0 : move |conn| -> DatabaseResult<_> {
914 0 : Ok(crate::schema::metadata_health::table
915 0 : .filter(healthy.eq(false))
916 0 : .load::<MetadataHealthPersistence>(conn)?)
917 0 : },
918 0 : )
919 0 : .await
920 0 : }
921 :
922 : /// Lists all the metadata health records that have not been updated since an `earlier` time.
923 : #[allow(dead_code)]
924 0 : pub(crate) async fn list_outdated_metadata_health_records(
925 0 : &self,
926 0 : earlier: chrono::DateTime<chrono::Utc>,
927 0 : ) -> DatabaseResult<Vec<MetadataHealthPersistence>> {
928 : use crate::schema::metadata_health::dsl::*;
929 :
930 0 : self.with_measured_conn(
931 0 : DatabaseOperation::ListMetadataHealthOutdated,
932 0 : move |conn| -> DatabaseResult<_> {
933 0 : let query = metadata_health.filter(last_scrubbed_at.lt(earlier));
934 0 : let res = query.load::<MetadataHealthPersistence>(conn)?;
935 :
936 0 : Ok(res)
937 0 : },
938 0 : )
939 0 : .await
940 0 : }
941 :
942 : /// Get the current entry from the `leader` table if one exists.
943 : /// It is an error for the table to contain more than one entry.
944 0 : pub(crate) async fn get_leader(&self) -> DatabaseResult<Option<ControllerPersistence>> {
945 0 : let mut leader: Vec<ControllerPersistence> = self
946 0 : .with_measured_conn(
947 0 : DatabaseOperation::GetLeader,
948 0 : move |conn| -> DatabaseResult<_> {
949 0 : Ok(crate::schema::controllers::table.load::<ControllerPersistence>(conn)?)
950 0 : },
951 0 : )
952 0 : .await?;
953 :
954 0 : if leader.len() > 1 {
955 0 : return Err(DatabaseError::Logical(format!(
956 0 : "More than one entry present in the leader table: {leader:?}"
957 0 : )));
958 0 : }
959 0 :
960 0 : Ok(leader.pop())
961 0 : }
962 :
963 : /// Update the new leader with compare-exchange semantics. If `prev` does not
964 : /// match the current leader entry, then the update is treated as a failure.
965 : /// When `prev` is not specified, the update is forced.
966 0 : pub(crate) async fn update_leader(
967 0 : &self,
968 0 : prev: Option<ControllerPersistence>,
969 0 : new: ControllerPersistence,
970 0 : ) -> DatabaseResult<()> {
971 : use crate::schema::controllers::dsl::*;
972 :
973 0 : let updated = self
974 0 : .with_measured_conn(
975 0 : DatabaseOperation::UpdateLeader,
976 0 : move |conn| -> DatabaseResult<usize> {
977 0 : let updated = match &prev {
978 0 : Some(prev) => diesel::update(controllers)
979 0 : .filter(address.eq(prev.address.clone()))
980 0 : .filter(started_at.eq(prev.started_at))
981 0 : .set((
982 0 : address.eq(new.address.clone()),
983 0 : started_at.eq(new.started_at),
984 0 : ))
985 0 : .execute(conn)?,
986 0 : None => diesel::insert_into(controllers)
987 0 : .values(new.clone())
988 0 : .execute(conn)?,
989 : };
990 :
991 0 : Ok(updated)
992 0 : },
993 0 : )
994 0 : .await?;
995 :
996 0 : if updated == 0 {
997 0 : return Err(DatabaseError::Logical(
998 0 : "Leader table update failed".to_string(),
999 0 : ));
1000 0 : }
1001 0 :
1002 0 : Ok(())
1003 0 : }
1004 :
1005 0 : pub(crate) async fn safekeeper_get(
1006 0 : &self,
1007 0 : id: i64,
1008 0 : ) -> Result<SafekeeperPersistence, DatabaseError> {
1009 : use crate::schema::safekeepers::dsl::{id as id_column, safekeepers};
1010 0 : self.with_conn(move |conn| -> DatabaseResult<SafekeeperPersistence> {
1011 0 : Ok(safekeepers
1012 0 : .filter(id_column.eq(&id))
1013 0 : .select(SafekeeperPersistence::as_select())
1014 0 : .get_result(conn)?)
1015 0 : })
1016 0 : .await
1017 0 : }
1018 :
1019 0 : pub(crate) async fn safekeeper_upsert(
1020 0 : &self,
1021 0 : record: SafekeeperPersistence,
1022 0 : ) -> Result<(), DatabaseError> {
1023 : use crate::schema::safekeepers::dsl::*;
1024 :
1025 0 : self.with_conn(move |conn| -> DatabaseResult<()> {
1026 0 : let bind = record.as_insert_or_update();
1027 :
1028 0 : let inserted_updated = diesel::insert_into(safekeepers)
1029 0 : .values(&bind)
1030 0 : .on_conflict(id)
1031 0 : .do_update()
1032 0 : .set(&bind)
1033 0 : .execute(conn)?;
1034 :
1035 0 : if inserted_updated != 1 {
1036 0 : return Err(DatabaseError::Logical(format!(
1037 0 : "unexpected number of rows ({})",
1038 0 : inserted_updated
1039 0 : )));
1040 0 : }
1041 0 :
1042 0 : Ok(())
1043 0 : })
1044 0 : .await
1045 0 : }
1046 : }
1047 :
1048 : /// Parts of [`crate::tenant_shard::TenantShard`] that are stored durably
1049 : #[derive(
1050 0 : QueryableByName, Queryable, Selectable, Insertable, Serialize, Deserialize, Clone, Eq, PartialEq,
1051 : )]
1052 : #[diesel(table_name = crate::schema::tenant_shards)]
1053 : pub(crate) struct TenantShardPersistence {
1054 : #[serde(default)]
1055 : pub(crate) tenant_id: String,
1056 : #[serde(default)]
1057 : pub(crate) shard_number: i32,
1058 : #[serde(default)]
1059 : pub(crate) shard_count: i32,
1060 : #[serde(default)]
1061 : pub(crate) shard_stripe_size: i32,
1062 :
1063 : // Latest generation number: next time we attach, increment this
1064 : // and use the incremented number when attaching.
1065 : //
1066 : // Generation is only None when first onboarding a tenant, where it may
1067 : // be in PlacementPolicy::Secondary and therefore have no valid generation state.
1068 : pub(crate) generation: Option<i32>,
1069 :
1070 : // Currently attached pageserver
1071 : #[serde(rename = "pageserver")]
1072 : pub(crate) generation_pageserver: Option<i64>,
1073 :
1074 : #[serde(default)]
1075 : pub(crate) placement_policy: String,
1076 : #[serde(default)]
1077 : pub(crate) splitting: SplitState,
1078 : #[serde(default)]
1079 : pub(crate) config: String,
1080 : #[serde(default)]
1081 : pub(crate) scheduling_policy: String,
1082 :
1083 : // Hint that we should attempt to schedule this tenant shard the given
1084 : // availability zone in order to minimise the chances of cross-AZ communication
1085 : // with compute.
1086 : pub(crate) preferred_az_id: Option<String>,
1087 : }
1088 :
1089 : impl TenantShardPersistence {
1090 0 : pub(crate) fn get_shard_identity(&self) -> Result<ShardIdentity, ShardConfigError> {
1091 0 : if self.shard_count == 0 {
1092 0 : Ok(ShardIdentity::unsharded())
1093 : } else {
1094 0 : Ok(ShardIdentity::new(
1095 0 : ShardNumber(self.shard_number as u8),
1096 0 : ShardCount::new(self.shard_count as u8),
1097 0 : ShardStripeSize(self.shard_stripe_size as u32),
1098 0 : )?)
1099 : }
1100 0 : }
1101 :
1102 0 : pub(crate) fn get_tenant_shard_id(&self) -> Result<TenantShardId, hex::FromHexError> {
1103 0 : Ok(TenantShardId {
1104 0 : tenant_id: TenantId::from_str(self.tenant_id.as_str())?,
1105 0 : shard_number: ShardNumber(self.shard_number as u8),
1106 0 : shard_count: ShardCount::new(self.shard_count as u8),
1107 : })
1108 0 : }
1109 : }
1110 :
1111 : /// Parts of [`crate::node::Node`] that are stored durably
1112 0 : #[derive(Serialize, Deserialize, Queryable, Selectable, Insertable, Eq, PartialEq)]
1113 : #[diesel(table_name = crate::schema::nodes)]
1114 : pub(crate) struct NodePersistence {
1115 : pub(crate) node_id: i64,
1116 : pub(crate) scheduling_policy: String,
1117 : pub(crate) listen_http_addr: String,
1118 : pub(crate) listen_http_port: i32,
1119 : pub(crate) listen_pg_addr: String,
1120 : pub(crate) listen_pg_port: i32,
1121 : pub(crate) availability_zone_id: String,
1122 : }
1123 :
1124 : /// Tenant metadata health status that are stored durably.
1125 0 : #[derive(Queryable, Selectable, Insertable, Serialize, Deserialize, Clone, Eq, PartialEq)]
1126 : #[diesel(table_name = crate::schema::metadata_health)]
1127 : pub(crate) struct MetadataHealthPersistence {
1128 : #[serde(default)]
1129 : pub(crate) tenant_id: String,
1130 : #[serde(default)]
1131 : pub(crate) shard_number: i32,
1132 : #[serde(default)]
1133 : pub(crate) shard_count: i32,
1134 :
1135 : pub(crate) healthy: bool,
1136 : pub(crate) last_scrubbed_at: chrono::DateTime<chrono::Utc>,
1137 : }
1138 :
1139 : impl MetadataHealthPersistence {
1140 0 : pub fn new(
1141 0 : tenant_shard_id: TenantShardId,
1142 0 : healthy: bool,
1143 0 : last_scrubbed_at: chrono::DateTime<chrono::Utc>,
1144 0 : ) -> Self {
1145 0 : let tenant_id = tenant_shard_id.tenant_id.to_string();
1146 0 : let shard_number = tenant_shard_id.shard_number.0 as i32;
1147 0 : let shard_count = tenant_shard_id.shard_count.literal() as i32;
1148 0 :
1149 0 : MetadataHealthPersistence {
1150 0 : tenant_id,
1151 0 : shard_number,
1152 0 : shard_count,
1153 0 : healthy,
1154 0 : last_scrubbed_at,
1155 0 : }
1156 0 : }
1157 :
1158 : #[allow(dead_code)]
1159 0 : pub(crate) fn get_tenant_shard_id(&self) -> Result<TenantShardId, hex::FromHexError> {
1160 0 : Ok(TenantShardId {
1161 0 : tenant_id: TenantId::from_str(self.tenant_id.as_str())?,
1162 0 : shard_number: ShardNumber(self.shard_number as u8),
1163 0 : shard_count: ShardCount::new(self.shard_count as u8),
1164 : })
1165 0 : }
1166 : }
1167 :
1168 : impl From<MetadataHealthPersistence> for MetadataHealthRecord {
1169 0 : fn from(value: MetadataHealthPersistence) -> Self {
1170 0 : MetadataHealthRecord {
1171 0 : tenant_shard_id: value
1172 0 : .get_tenant_shard_id()
1173 0 : .expect("stored tenant id should be valid"),
1174 0 : healthy: value.healthy,
1175 0 : last_scrubbed_at: value.last_scrubbed_at,
1176 0 : }
1177 0 : }
1178 : }
1179 :
1180 : #[derive(
1181 0 : Serialize, Deserialize, Queryable, Selectable, Insertable, Eq, PartialEq, Debug, Clone,
1182 : )]
1183 : #[diesel(table_name = crate::schema::controllers)]
1184 : pub(crate) struct ControllerPersistence {
1185 : pub(crate) address: String,
1186 : pub(crate) started_at: chrono::DateTime<chrono::Utc>,
1187 : }
1188 :
1189 0 : #[derive(Serialize, Deserialize, Queryable, Selectable, Eq, PartialEq, Debug, Clone)]
1190 : #[diesel(table_name = crate::schema::safekeepers)]
1191 : pub(crate) struct SafekeeperPersistence {
1192 : pub(crate) id: i64,
1193 : pub(crate) region_id: String,
1194 : /// 1 is special, it means just created (not currently posted to storcon).
1195 : /// Zero or negative is not really expected.
1196 : /// Otherwise the number from `release-$(number_of_commits_on_branch)` tag.
1197 : pub(crate) version: i64,
1198 : pub(crate) host: String,
1199 : pub(crate) port: i32,
1200 : pub(crate) active: bool,
1201 : pub(crate) http_port: i32,
1202 : pub(crate) availability_zone_id: String,
1203 : }
1204 :
1205 : impl SafekeeperPersistence {
1206 0 : fn as_insert_or_update(&self) -> InsertUpdateSafekeeper<'_> {
1207 0 : InsertUpdateSafekeeper {
1208 0 : id: self.id,
1209 0 : region_id: &self.region_id,
1210 0 : version: self.version,
1211 0 : host: &self.host,
1212 0 : port: self.port,
1213 0 : active: self.active,
1214 0 : http_port: self.http_port,
1215 0 : availability_zone_id: &self.availability_zone_id,
1216 0 : }
1217 0 : }
1218 : }
1219 :
1220 0 : #[derive(Insertable, AsChangeset)]
1221 : #[diesel(table_name = crate::schema::safekeepers)]
1222 : struct InsertUpdateSafekeeper<'a> {
1223 : id: i64,
1224 : region_id: &'a str,
1225 : version: i64,
1226 : host: &'a str,
1227 : port: i32,
1228 : active: bool,
1229 : http_port: i32,
1230 : availability_zone_id: &'a str,
1231 : }
|
