From 0b2ed17f8629624167f287878d85629146edf8b8 Mon Sep 17 00:00:00 2001 From: Heikki Linnakangas Date: Tue, 13 Jul 2021 14:04:52 +0300 Subject: [PATCH] Implement garbage collection. The counters returned by garbage collection, in GcResult, don't make much sense with the snapshot files implemention, so I added new counters. That broke the test_gc test, so I made a copy of it as test_snapfiles_gc based on the new counters. Handling relation drops is still not implemented. --- pageserver/src/layered_repository.rs | 100 +++++++++++++++- .../src/layered_repository/snapshot_layer.rs | 92 ++++++++++++++- pageserver/src/lib.rs | 2 +- pageserver/src/page_service.rs | 70 ++++------- pageserver/src/repository.rs | 30 +++++ test_runner/batch_others/test_snapfiles_gc.py | 111 ++++++++++++++++++ zenith_utils/src/pq_proto.rs | 15 +++ 7 files changed, 364 insertions(+), 56 deletions(-) create mode 100644 test_runner/batch_others/test_snapfiles_gc.py diff --git a/pageserver/src/layered_repository.rs b/pageserver/src/layered_repository.rs index a86f3041d1..4f23cb0f07 100644 --- a/pageserver/src/layered_repository.rs +++ b/pageserver/src/layered_repository.rs @@ -22,13 +22,14 @@ use bytes::Bytes; use log::*; use serde::{Deserialize, Serialize}; -use std::collections::HashSet; +use std::collections::{HashSet, BTreeSet}; use std::collections::{BTreeMap, HashMap}; +use std::fs; use std::fs::File; use std::io::Write; use std::ops::Bound::Included; use std::sync::{Arc, Mutex}; -use std::time::Duration; +use std::time::{Duration, Instant}; use crate::repository::{GcResult, History, RelTag, Repository, Timeline, WALRecord}; use crate::restore_local_repo::import_timeline_wal; @@ -201,6 +202,83 @@ impl LayeredRepository { Ok(TimelineMetadata::des(&data)?) } + + // + // How garbage collection works + // -------- + // + // +--bar-------------> + // / + // +----+-----foo----------------> + // / + // ----main--+--------------------------> + // \ + // +-----baz--------> + // + // + // 1. Grab a mutex to prevent new timelines from being created + // 2. Scan all timelines, and on each timeline, make note of the + // all the points where other timelines have been branched off. + // We will refrain from removing page versions at those LSNs. + // 3. For each timeline, scan all snapshot files on the timeline. + // Remove all files for which a newer file exists and which + // don't cover any branch point LSNs. + // + // TODO: + // - if a relation has been modified on a child branch, then we + // we don't need to keep that in the parent anymore. + // + // - Currently, this is only triggered manually by the 'do_gc' command. + // There is no background thread to do it automatically. + fn gc_iteration(conf: &'static PageServerConf, horizon: u64) -> Result { + + let mut totals: GcResult = Default::default(); + let now = Instant::now(); + + // TODO: grab mutex to prevent new timelines from being created here. + + // Scan all timelines for the branch points. + let mut all_branchpoints: BTreeSet<(ZTimelineId, Lsn)> = BTreeSet::new(); + + // Remember timelineid and its last_record_lsn for each timeline + let mut timelines: Vec<(ZTimelineId, Lsn)> = Vec::new(); + + let timelines_path = conf.workdir.join("timelines"); + for direntry in fs::read_dir(timelines_path)? { + let direntry = direntry?; + if let Some(fname) = direntry.file_name().to_str() { + if let Ok(timelineid) = fname.parse::() { + + // Read the metadata of this timeline to get its parent timeline. + let metadata = Self::load_metadata(conf, timelineid)?; + + timelines.push((timelineid, metadata.last_record_lsn)); + + if let Some(ancestor_timeline) = metadata.ancestor_timeline { + all_branchpoints.insert((ancestor_timeline, metadata.ancestor_lsn)); + } + } + } + } + + // Ok, we now know all the branch points. Iterate through them. + for (timelineid, last_lsn) in timelines { + let branchpoints: Vec = all_branchpoints.range( + (Included((timelineid, Lsn(0))), + Included((timelineid, Lsn(u64::MAX))))) + .map(|&x| x.1) + .collect(); + + if let Some(cutoff) = last_lsn.checked_sub(horizon) { + let result = SnapshotLayer::gc_timeline(conf, timelineid, branchpoints, cutoff)?; + + totals += result; + } + } + + totals.elapsed = now.elapsed(); + Ok(totals) + } } /// LayerMap is a BTreeMap keyed by RelTag and the snapshot file's start LSN. @@ -368,9 +446,21 @@ impl Timeline for LayeredTimeline { todo!(); } - fn gc_iteration(&self, _horizon: u64) -> Result { - //TODO - Ok(Default::default()) + fn gc_iteration(&self, horizon: u64) -> Result { + // In the layered repository, event to GC a single timeline, + // we have to scan all the timelines to determine what child + // timelines there are, so that we know to retain snapshot + // files that are still needed by the children. So we just do + // GC on the whole repository. + // + // FIXME: This makes writing repeatable tests harder, if + // activity on other timelines can affect the counters that + // we return + + // But do flush the in-memory layers to disk first. + self.checkpoint()?; + + LayeredRepository::gc_iteration(self.conf, horizon) } fn put_wal_record(&self, tag: BufferTag, rec: WALRecord) -> Result<()> { diff --git a/pageserver/src/layered_repository/snapshot_layer.rs b/pageserver/src/layered_repository/snapshot_layer.rs index 96325938dc..fad888d743 100644 --- a/pageserver/src/layered_repository/snapshot_layer.rs +++ b/pageserver/src/layered_repository/snapshot_layer.rs @@ -31,20 +31,21 @@ //! use crate::layered_repository::storage_layer::Layer; use crate::layered_repository::storage_layer::PageVersion; -use crate::repository::{RelTag, WALRecord}; +use crate::repository::{GcResult, RelTag, WALRecord}; use crate::walredo::WalRedoManager; use crate::PageServerConf; use crate::ZTimelineId; use anyhow::{anyhow, bail, Result}; use bytes::Bytes; use log::*; -use std::collections::{BTreeMap, HashSet}; +use std::collections::{BTreeMap, HashSet, BTreeSet}; use std::fs; use std::fs::File; use std::io::Write; use std::ops::Bound::Included; use std::path::PathBuf; use std::sync::Mutex; +use std::time::Instant; use bookfile::{Book, BookWriter}; @@ -498,4 +499,91 @@ impl SnapshotLayer { Some((reltag, start_lsn, end_lsn)) } + + + /// + /// Garbage collect snapshot files on a timeline that are no longer needed. + /// + /// The caller specifies how much history is needed with the two arguments: + /// + /// retain_lsns: keep page a version of each page at these LSNs + /// cutoff: also keep everything newer than this LSN + /// + /// The 'retain_lsns' lists is currently used to prevent removing files that + /// are needed by child timelines. In the future, the user might be able to + /// name additional points in time to retain. The caller is responsible for + /// collecting that information. + /// + /// The 'cutoff' point is used to retain recent versions that might still be + /// needed by read-only nodes. (As of this writing, the caller just passes + /// the latest LSN subtracted by a constant, and doesn't do anything smart + /// to figure out what read-only nodes might actually need.) + /// + /// Currently, we don't make any attempt at removing unneeded page versions + /// within a snapshot file. We can only remove the whole file if it's fully + /// obsolete. + /// + pub fn gc_timeline(conf: &'static PageServerConf, + timelineid: ZTimelineId, + retain_lsns: Vec, + cutoff: Lsn) -> Result { + + let now = Instant::now(); + let mut result: GcResult = Default::default(); + + // Scan all snapshot files in the directory. For each file, if a newer file + // exists, we can remove the old one. + + // For convenience and speed, slurp the list of files in the directoy into memory first. + let mut snapfiles: BTreeSet<(RelTag, Lsn, Lsn)> = BTreeSet::new(); + + let timeline_path = conf.timeline_path(timelineid); + for direntry in fs::read_dir(timeline_path)? { + let direntry = direntry?; + let fname = direntry.file_name(); + let fname = fname.to_str().unwrap(); + + if let Some((reltag, start_lsn, end_lsn)) = Self::fname_to_tag(fname) { + snapfiles.insert((reltag, start_lsn, end_lsn)); + } + result.snapshot_files_total += 1; + } + + // Now determine for each file if it needs to be retained + 'outer: for (reltag, start_lsn, end_lsn) in &snapfiles { + + // Is it newer than cutoff point? + if *end_lsn >= cutoff { + result.snapshot_files_needed_by_cutoff += 1; + continue 'outer; + } + + // Is it needed by a child branch? + for retain_lsn in &retain_lsns { + // FIXME: are the bounds inclusive or exclusive? + if *start_lsn <= *retain_lsn && *retain_lsn <= *end_lsn { + result.snapshot_files_needed_by_branches += 1; + continue 'outer; + } + } + + // Is there a later snapshot file for this relation? + if snapfiles.range( + (Included((*reltag, *end_lsn, Lsn(0))), + Included((*reltag, Lsn(u64::MAX), Lsn(0))))).next().is_none() { + // there is no later file, so keep it + result.snapshot_files_not_updated += 1; + continue 'outer; + } + + // We didn't find any reason to keep this file, so remove it. + let path = Self::path_for(conf, timelineid, *reltag, *start_lsn, *end_lsn); + info!("garbage collecting {}", path.display()); + fs::remove_file(path)?; + result.snapshot_files_removed += 1; + } + + result.elapsed = now.elapsed(); + Ok(result) + } } diff --git a/pageserver/src/lib.rs b/pageserver/src/lib.rs index 7dc0d1926b..cac0877733 100644 --- a/pageserver/src/lib.rs +++ b/pageserver/src/lib.rs @@ -192,7 +192,7 @@ impl fmt::Display for ZId { /// is separate from PostgreSQL timelines, and doesn't have those /// limitations. A zenith timeline is identified by a 128-bit ID, which /// is usually printed out as a hex string. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] pub struct ZTimelineId(ZId); impl FromStr for ZTimelineId { diff --git a/pageserver/src/page_service.rs b/pageserver/src/page_service.rs index 98fe7e6838..8b92fa298e 100644 --- a/pageserver/src/page_service.rs +++ b/pageserver/src/page_service.rs @@ -543,54 +543,21 @@ impl postgres_backend::Handler for PageServerHandler { let result = timeline.gc_iteration(gc_horizon, true)?; pgb.write_message_noflush(&BeMessage::RowDescription(&[ - RowDescriptor { - name: b"n_relations", - typoid: 20, - typlen: 8, - ..Default::default() - }, - RowDescriptor { - name: b"truncated", - typoid: 20, - typlen: 8, - ..Default::default() - }, - RowDescriptor { - name: b"deleted", - typoid: 20, - typlen: 8, - ..Default::default() - }, - RowDescriptor { - name: b"prep_deleted", - typoid: 20, - typlen: 8, - ..Default::default() - }, - RowDescriptor { - name: b"slru_deleted", - typoid: 20, - typlen: 8, - ..Default::default() - }, - RowDescriptor { - name: b"chkp_deleted", - typoid: 20, - typlen: 8, - ..Default::default() - }, - RowDescriptor { - name: b"dropped", - typoid: 20, - typlen: 8, - ..Default::default() - }, - RowDescriptor { - name: b"elapsed", - typoid: 20, - typlen: 8, - ..Default::default() - }, + RowDescriptor::int8_col(b"n_relations"), + RowDescriptor::int8_col(b"truncated"), + RowDescriptor::int8_col(b"deleted"), + RowDescriptor::int8_col(b"prep_deleted"), + RowDescriptor::int8_col(b"slru_deleted"), + RowDescriptor::int8_col(b"chkp_deleted"), + RowDescriptor::int8_col(b"dropped"), + + RowDescriptor::int8_col(b"snapshot_files_total"), + RowDescriptor::int8_col(b"snapshot_files_needed_by_cutoff"), + RowDescriptor::int8_col(b"snapshot_files_needed_by_branches"), + RowDescriptor::int8_col(b"snapshot_files_not_updated"), + RowDescriptor::int8_col(b"snapshot_files_removed"), + + RowDescriptor::int8_col(b"elapsed"), ]))? .write_message_noflush(&BeMessage::DataRow(&[ Some(&result.n_relations.to_string().as_bytes()), @@ -600,6 +567,13 @@ impl postgres_backend::Handler for PageServerHandler { Some(&result.slru_deleted.to_string().as_bytes()), Some(&result.chkp_deleted.to_string().as_bytes()), Some(&result.dropped.to_string().as_bytes()), + + Some(&result.snapshot_files_total.to_string().as_bytes()), + Some(&result.snapshot_files_needed_by_cutoff.to_string().as_bytes()), + Some(&result.snapshot_files_needed_by_branches.to_string().as_bytes()), + Some(&result.snapshot_files_not_updated.to_string().as_bytes()), + Some(&result.snapshot_files_removed.to_string().as_bytes()), + Some(&result.elapsed.as_millis().to_string().as_bytes()), ]))? .write_message(&BeMessage::CommandComplete(b"SELECT 1"))?; diff --git a/pageserver/src/repository.rs b/pageserver/src/repository.rs index 3d82b2ff50..24c8670267 100644 --- a/pageserver/src/repository.rs +++ b/pageserver/src/repository.rs @@ -9,6 +9,8 @@ use postgres_ffi::TransactionId; use serde::{Deserialize, Serialize}; use std::collections::HashSet; use std::iter::Iterator; +use std::fmt; +use std::ops::AddAssign; use std::sync::Arc; use std::time::Duration; use zenith_utils::lsn::Lsn; @@ -39,6 +41,8 @@ pub trait Repository: Send + Sync { /// #[derive(Default)] pub struct GcResult { + // FIXME: These counters make sense for the ObjectRepository. They are not used + // by the LayeredRepository. pub n_relations: u64, pub inspected: u64, pub truncated: u64, @@ -47,9 +51,35 @@ pub struct GcResult { pub slru_deleted: u64, // SLRU (clog, multixact) pub chkp_deleted: u64, // Checkpoints pub dropped: u64, + + // These are used for the LayeredRepository instead + pub snapshot_files_total: u64, + pub snapshot_files_needed_by_cutoff: u64, + pub snapshot_files_needed_by_branches: u64, + pub snapshot_files_not_updated: u64, + pub snapshot_files_removed: u64, + pub elapsed: Duration, } + +impl AddAssign for GcResult { + fn add_assign(&mut self, other: Self) { + self.n_relations += other.n_relations; + self.truncated += other.truncated; + self.deleted += other.deleted; + self.dropped += other.dropped; + + self.snapshot_files_total += other.snapshot_files_total; + self.snapshot_files_needed_by_cutoff += other.snapshot_files_needed_by_cutoff; + self.snapshot_files_needed_by_branches += other.snapshot_files_needed_by_branches; + self.snapshot_files_not_updated += other.snapshot_files_not_updated; + self.snapshot_files_removed += other.snapshot_files_removed; + + self.elapsed += other.elapsed; + } +} + pub trait Timeline: Send + Sync { //------------------------------------------------------------------------------ // Public GET functions diff --git a/test_runner/batch_others/test_snapfiles_gc.py b/test_runner/batch_others/test_snapfiles_gc.py new file mode 100644 index 0000000000..cd9a2fd3ab --- /dev/null +++ b/test_runner/batch_others/test_snapfiles_gc.py @@ -0,0 +1,111 @@ +from contextlib import closing +import psycopg2.extras +import time; + +pytest_plugins = ("fixtures.zenith_fixtures") + +def print_gc_result(row): + print("GC duration {elapsed} ms, total: {snapshot_files_total}, needed_by_cutoff {snapshot_files_needed_by_cutoff}, needed_by_branches: {snapshot_files_needed_by_branches}, not_updated: {snapshot_files_not_updated}, removed: {snapshot_files_removed}".format_map(row)) + + +# +# Test Garbage Collection of old snapshot files +# +# This test is pretty tightly coupled with the current implementation of layered +# storage, in layered_repository.rs. +# +def test_snapfiles_gc(zenith_cli, pageserver, postgres, pg_bin): + zenith_cli.run(["branch", "test_snapfiles_gc", "empty"]) + pg = postgres.create_start('test_snapfiles_gc') + + with closing(pg.connect()) as conn: + with conn.cursor() as cur: + with closing(pageserver.connect()) as psconn: + with psconn.cursor(cursor_factory = psycopg2.extras.DictCursor) as pscur: + + # Get the timeline ID of our branch. We need it for the 'do_gc' command + cur.execute("SHOW zenith.zenith_timeline") + timeline = cur.fetchone()[0] + + # Create a test table + cur.execute("CREATE TABLE foo(x integer)") + + # Run GC, to clear out any garbage left behind in the catalogs by + # the CREATE TABLE command. We want to have a clean slate with no garbage + # before running the actual tests below, otherwise the counts won't match + # what we expect. + # + # Also run vacuum first to make it less likely that autovacuum or pruning + # kicks in and confuses our numbers. + cur.execute("VACUUM") + + print("Running GC before test") + pscur.execute(f"do_gc {timeline} 0") + row = pscur.fetchone() + print_gc_result(row); + # remember the number of files + snapshot_files_total = row['snapshot_files_total'] + assert snapshot_files_total > 0 + + # Insert a row. The first insert will also create a metadata entry for the + # relation, with size == 1 block. Hence, bump up the expected relation count. + snapshot_files_total += 1; + print("Inserting one row and running GC") + cur.execute("INSERT INTO foo VALUES (1)") + pscur.execute(f"do_gc {timeline} 0") + row = pscur.fetchone() + print_gc_result(row); + assert row['snapshot_files_total'] == snapshot_files_total + assert row['snapshot_files_removed'] == 0 + + # Insert two more rows and run GC. + # This should create a new snapshot file with the new contents, and + # remove the old one. + print("Inserting two more rows and running GC") + cur.execute("INSERT INTO foo VALUES (2)") + cur.execute("INSERT INTO foo VALUES (3)") + + pscur.execute(f"do_gc {timeline} 0") + row = pscur.fetchone() + print_gc_result(row); + assert row['snapshot_files_total'] == snapshot_files_total + 1 + assert row['snapshot_files_removed'] == 1 + + # Do it again. Should again create a new snapshot file and remove old one. + print("Inserting two more rows and running GC") + cur.execute("INSERT INTO foo VALUES (2)") + cur.execute("INSERT INTO foo VALUES (3)") + + pscur.execute(f"do_gc {timeline} 0") + row = pscur.fetchone() + print_gc_result(row); + assert row['snapshot_files_total'] == snapshot_files_total + 1 + assert row['snapshot_files_removed'] == 1 + + # Run GC again, with no changes in the database. Should not remove anything. + print("Run GC again, with nothing to do") + pscur.execute(f"do_gc {timeline} 0") + row = pscur.fetchone() + print_gc_result(row); + assert row['snapshot_files_total'] == snapshot_files_total + assert row['snapshot_files_removed'] == 0 + + # + # Test DROP TABLE checks that relation data and metadata was deleted by GC from object storage + # + print("Drop table and run GC again"); + cur.execute("DROP TABLE foo") + + pscur.execute(f"do_gc {timeline} 0") + row = pscur.fetchone() + print_gc_result(row); + + # Each relation fork is counted separately, hence 3. But the catalog + # updates also create new snapshot files of the catalogs. + + # TODO: perhaps we should count catalog and user relations separately, + # to make this kind of testing more robust + + # FIXME: Unlinking relations hasn't been implemented yet + #assert row['snapshot_files_removed'] == 3 + #assert row['snapshot_files_removed'] == 5 diff --git a/zenith_utils/src/pq_proto.rs b/zenith_utils/src/pq_proto.rs index 3a9b41859e..7aaf86c86a 100644 --- a/zenith_utils/src/pq_proto.rs +++ b/zenith_utils/src/pq_proto.rs @@ -379,6 +379,21 @@ impl Default for RowDescriptor<'_> { } } +impl RowDescriptor<'_> { + /// Convenience function to create a RowDescriptor message for an int8 column + pub const fn int8_col(name: &[u8]) -> RowDescriptor { + RowDescriptor { + name, + tableoid: 0, + attnum: 0, + typoid: 20, + typlen: 8, + typmod: 0, + formatcode: 0, + } + } +} + #[derive(Debug)] pub struct XLogDataBody<'a> { pub wal_start: u64,