diff --git a/.github/workflows/run-tests-tiered.yml b/.github/workflows/run-tests-tiered.yml index 5f826f6a2..dcb2777b7 100644 --- a/.github/workflows/run-tests-tiered.yml +++ b/.github/workflows/run-tests-tiered.yml @@ -145,6 +145,7 @@ jobs: - skip-large-objects - cdc-low-level - cdc-test-decoding + - cdc-prune - cdc-endpos-between-transaction - cdc-filtering - cdc-wal2json diff --git a/docs/include/clone.rst b/docs/include/clone.rst index 8c01f053f..478dd1028 100644 --- a/docs/include/clone.rst +++ b/docs/include/clone.rst @@ -49,4 +49,6 @@ --defer-analyze Defer ANALYZE until after post-data restore --defer-validate-fks Create FK constraints as NOT VALID, skipping validation scan --use-copy-binary Use the COPY BINARY format for COPY operations + --prune-threshold Max size of applied CDC files to retain (e.g. 10GB, 0 to disable) + --prune-min-age Min age before applied CDC files can be deleted (e.g. 15m, 2h) diff --git a/docs/include/follow.rst b/docs/include/follow.rst index 37f2a359d..253cc1741 100644 --- a/docs/include/follow.rst +++ b/docs/include/follow.rst @@ -17,4 +17,6 @@ --create-slot Create the replication slot --origin Use this Postgres replication origin node name --endpos Stop replaying changes when reaching this LSN + --prune-threshold Max size of applied CDC files to retain (e.g. 10GB, 0 to disable) + --prune-min-age Min age before applied CDC files can be deleted (e.g. 15m, 2h) diff --git a/src/bin/pgcopydb/cli_clone_follow.c b/src/bin/pgcopydb/cli_clone_follow.c index 0fb5e0ef6..b0ded2b26 100644 --- a/src/bin/pgcopydb/cli_clone_follow.c +++ b/src/bin/pgcopydb/cli_clone_follow.c @@ -72,6 +72,8 @@ " --defer-analyze Defer ANALYZE until after post-data restore\n" \ " --defer-validate-fks Create FK constraints as NOT VALID, skipping validation scan\n" \ " --use-copy-binary Use the COPY BINARY format for COPY operations\n" \ + " --prune-threshold Max size of applied CDC files to retain (e.g. 10GB, 0 to disable)\n" \ + " --prune-min-age Min age before applied CDC files can be deleted (e.g. 15m, 2h)\n" \ CommandLine clone_command = make_command( @@ -110,7 +112,9 @@ CommandLine follow_command = " --slot-name Use this Postgres replication slot name\n" " --create-slot Create the replication slot\n" " --origin Use this Postgres replication origin node name\n" - " --endpos Stop replaying changes when reaching this LSN\n", + " --endpos Stop replaying changes when reaching this LSN\n" + " --prune-threshold Max size of applied CDC files to retain (e.g. 10GB, 0 to disable)\n" + " --prune-min-age Min age before applied CDC files can be deleted (e.g. 15m, 2h)\n", cli_copy_db_getopts, cli_follow); @@ -225,7 +229,9 @@ clone_and_follow(CopyDataSpec *copySpecs) &(copySpecs->filters), copyDBoptions.stdIn, copyDBoptions.stdOut, - logSQL)) + logSQL, + copyDBoptions.pruneThresholdBytes, + copyDBoptions.pruneMinAgeSeconds)) { /* errors have already been logged */ exit(EXIT_CODE_INTERNAL_ERROR); @@ -561,7 +567,9 @@ cli_follow(int argc, char **argv) &(copySpecs.filters), copyDBoptions.stdIn, copyDBoptions.stdOut, - logSQL)) + logSQL, + copyDBoptions.pruneThresholdBytes, + copyDBoptions.pruneMinAgeSeconds)) { /* errors have already been logged */ exit(EXIT_CODE_INTERNAL_ERROR); diff --git a/src/bin/pgcopydb/cli_common.c b/src/bin/pgcopydb/cli_common.c index 0e6a79e66..bc5f75362 100644 --- a/src/bin/pgcopydb/cli_common.c +++ b/src/bin/pgcopydb/cli_common.c @@ -657,6 +657,8 @@ cli_copy_db_getopts(int argc, char **argv) { "defer-indexes", no_argument, NULL, 257 }, { "defer-analyze", no_argument, NULL, 258 }, { "defer-validate-fks", no_argument, NULL, 259 }, + { "prune-threshold", required_argument, NULL, 260 }, + { "prune-min-age", required_argument, NULL, 261 }, { "help", no_argument, NULL, 'h' }, { NULL, 0, NULL, 0 } }; @@ -1159,6 +1161,45 @@ cli_copy_db_getopts(int argc, char **argv) break; } + case 260: + { + if (!cli_parse_bytes_pretty( + optarg, + &(options.pruneThresholdBytes), + (char *) &(options.pruneThresholdPretty), + sizeof(options.pruneThresholdPretty))) + { + log_fatal("Failed to parse --prune-threshold: \"%s\"", + optarg); + ++errors; + } + + log_trace("--prune-threshold %s (%lld)", + options.pruneThresholdPretty, + (long long) options.pruneThresholdBytes); + break; + } + + case 261: + { + if (!cli_parse_duration( + optarg, + &(options.pruneMinAgeSeconds))) + { + log_fatal("Failed to parse --prune-min-age: \"%s\"", + optarg); + ++errors; + } + + strlcpy(options.pruneMinAgePretty, optarg, + sizeof(options.pruneMinAgePretty)); + + log_trace("--prune-min-age %s (%d seconds)", + options.pruneMinAgePretty, + options.pruneMinAgeSeconds); + break; + } + case '?': default: { @@ -1210,6 +1251,23 @@ cli_copy_db_getopts(int argc, char **argv) exit(EXIT_CODE_BAD_ARGS); } + if (options.pruneThresholdBytes == 0 && options.pruneMinAgeSeconds > 0) + { + log_warn("--prune-min-age has no effect without --prune-threshold"); + } + + /* + * When prune threshold is set but min-age wasn't explicitly provided, + * default to 15 minutes (900 seconds) for safety. + */ + if (options.pruneThresholdBytes > 0 && options.pruneMinAgeSeconds == 0 && + options.pruneMinAgePretty[0] == '\0') + { + options.pruneMinAgeSeconds = 900; + strlcpy(options.pruneMinAgePretty, "15m", + sizeof(options.pruneMinAgePretty)); + } + if (errors > 0) { commandline_help(stderr); diff --git a/src/bin/pgcopydb/cli_common.h b/src/bin/pgcopydb/cli_common.h index fefc34369..18867c75e 100644 --- a/src/bin/pgcopydb/cli_common.h +++ b/src/bin/pgcopydb/cli_common.h @@ -100,6 +100,12 @@ typedef struct CopyDBOptions char filterFileName[MAXPGPATH]; char requirementsFileName[MAXPGPATH]; + + /* CDC file prune configuration */ + uint64_t pruneThresholdBytes; + char pruneThresholdPretty[BUFSIZE]; + int pruneMinAgeSeconds; + char pruneMinAgePretty[BUFSIZE]; } CopyDBOptions; extern bool outputJSON; diff --git a/src/bin/pgcopydb/cli_snapshot.c b/src/bin/pgcopydb/cli_snapshot.c index f80dd3ebc..7868ad3ba 100644 --- a/src/bin/pgcopydb/cli_snapshot.c +++ b/src/bin/pgcopydb/cli_snapshot.c @@ -326,7 +326,9 @@ cli_create_snapshot(int argc, char **argv) &(copySpecs.filters), createSNoptions.stdIn, createSNoptions.stdOut, - logSQL)) + logSQL, + 0, + 0)) { /* errors have already been logged */ exit(EXIT_CODE_INTERNAL_ERROR); diff --git a/src/bin/pgcopydb/cli_stream.c b/src/bin/pgcopydb/cli_stream.c index 35fd4aedc..c973a6382 100644 --- a/src/bin/pgcopydb/cli_stream.c +++ b/src/bin/pgcopydb/cli_stream.c @@ -222,6 +222,8 @@ cli_stream_getopts(int argc, char **argv) { "debug", no_argument, NULL, 'd' }, { "trace", no_argument, NULL, 'z' }, { "quiet", no_argument, NULL, 'q' }, + { "prune-threshold", required_argument, NULL, 256 }, + { "prune-min-age", required_argument, NULL, 257 }, { "help", no_argument, NULL, 'h' }, { NULL, 0, NULL, 0 } }; @@ -434,6 +436,45 @@ cli_stream_getopts(int argc, char **argv) break; } + case 256: + { + if (!cli_parse_bytes_pretty( + optarg, + &(options.pruneThresholdBytes), + (char *) &(options.pruneThresholdPretty), + sizeof(options.pruneThresholdPretty))) + { + log_fatal("Failed to parse --prune-threshold: \"%s\"", + optarg); + ++errors; + } + + log_trace("--prune-threshold %s (%lld)", + options.pruneThresholdPretty, + (long long) options.pruneThresholdBytes); + break; + } + + case 257: + { + if (!cli_parse_duration( + optarg, + &(options.pruneMinAgeSeconds))) + { + log_fatal("Failed to parse --prune-min-age: \"%s\"", + optarg); + ++errors; + } + + strlcpy(options.pruneMinAgePretty, optarg, + sizeof(options.pruneMinAgePretty)); + + log_trace("--prune-min-age %s (%d seconds)", + options.pruneMinAgePretty, + options.pruneMinAgeSeconds); + break; + } + case '?': default: { @@ -472,6 +513,23 @@ cli_stream_getopts(int argc, char **argv) exit(EXIT_CODE_BAD_ARGS); } + if (options.pruneThresholdBytes == 0 && options.pruneMinAgeSeconds > 0) + { + log_warn("--prune-min-age has no effect without --prune-threshold"); + } + + /* + * When prune threshold is set but min-age wasn't explicitly provided, + * default to 15 minutes (900 seconds) for safety. + */ + if (options.pruneThresholdBytes > 0 && options.pruneMinAgeSeconds == 0 && + options.pruneMinAgePretty[0] == '\0') + { + options.pruneMinAgeSeconds = 900; + strlcpy(options.pruneMinAgePretty, "15m", + sizeof(options.pruneMinAgePretty)); + } + if (errors > 0) { commandline_help(stderr); @@ -585,7 +643,9 @@ cli_stream_setup(int argc, char **argv) &(copySpecs.filters), streamDBoptions.stdIn, streamDBoptions.stdOut, - logSQL)) + logSQL, + streamDBoptions.pruneThresholdBytes, + streamDBoptions.pruneMinAgeSeconds)) { /* errors have already been logged */ exit(EXIT_CODE_INTERNAL_ERROR); @@ -729,7 +789,9 @@ cli_stream_catchup(int argc, char **argv) &(copySpecs.filters), streamDBoptions.stdIn, streamDBoptions.stdOut, - logSQL)) + logSQL, + streamDBoptions.pruneThresholdBytes, + streamDBoptions.pruneMinAgeSeconds)) { /* errors have already been logged */ exit(EXIT_CODE_INTERNAL_ERROR); @@ -813,7 +875,9 @@ cli_stream_replay(int argc, char **argv) &(copySpecs.filters), true, /* stdin */ true, /* stdout */ - logSQL)) + logSQL, + streamDBoptions.pruneThresholdBytes, + streamDBoptions.pruneMinAgeSeconds)) { /* errors have already been logged */ exit(EXIT_CODE_INTERNAL_ERROR); @@ -939,7 +1003,9 @@ cli_stream_transform(int argc, char **argv) &(copySpecs.filters), streamDBoptions.stdIn, streamDBoptions.stdOut, - logSQL)) + logSQL, + streamDBoptions.pruneThresholdBytes, + streamDBoptions.pruneMinAgeSeconds)) { /* errors have already been logged */ exit(EXIT_CODE_INTERNAL_ERROR); @@ -1102,7 +1168,9 @@ cli_stream_apply(int argc, char **argv) &(copySpecs.filters), true, /* streamDBoptions.stdIn */ false, /* streamDBoptions.stdOut */ - logSQL)) + logSQL, + streamDBoptions.pruneThresholdBytes, + streamDBoptions.pruneMinAgeSeconds)) { /* errors have already been logged */ exit(EXIT_CODE_INTERNAL_ERROR); @@ -1215,7 +1283,9 @@ stream_start_in_mode(LogicalStreamMode mode) &(copySpecs.filters), streamDBoptions.stdIn, streamDBoptions.stdOut, - logSQL)) + logSQL, + streamDBoptions.pruneThresholdBytes, + streamDBoptions.pruneMinAgeSeconds)) { /* errors have already been logged */ exit(EXIT_CODE_INTERNAL_ERROR); diff --git a/src/bin/pgcopydb/follow.c b/src/bin/pgcopydb/follow.c index eb9aa17bb..ff9342539 100644 --- a/src/bin/pgcopydb/follow.c +++ b/src/bin/pgcopydb/follow.c @@ -12,6 +12,7 @@ #include "cli_common.h" #include "cli_root.h" +#include "ld_prune.h" #include "ld_stream.h" #include "log.h" #include "progress.h" @@ -607,6 +608,23 @@ followDB(CopyDataSpec *copySpecs, StreamSpecs *streamSpecs) } } + /* + * When prune threshold is configured, start the prune watchdog + * to periodically remove old applied CDC files. + */ + if (streamSpecs->pruneThresholdBytes > 0) + { + FollowSubProcess *prune = &(streamSpecs->prune); + + if (!follow_start_subprocess(streamSpecs, prune)) + { + log_error("Failed to start the %s process", prune->name); + + (void) follow_exit_early(streamSpecs); + return false; + } + } + /* * Close pipe ends which follow is not using. Otherwise the processes * like transform and apply which reads from the pipe during replay @@ -830,6 +848,18 @@ follow_start_catchup(StreamSpecs *specs) } +/* + * follow_start_prune starts a sub-process that prunes old CDC files. + * The catalog is already opened by follow_start_subprocess before this is + * called. + */ +bool +follow_start_prune(StreamSpecs *specs) +{ + return cdc_prune_loop(specs); +} + + /* * follow_start_subprocess forks a subprocess and calls the given function. */ @@ -946,7 +976,8 @@ follow_wait_subprocesses(StreamSpecs *specs) FollowSubProcess *processArray[] = { &(specs->prefetch), &(specs->transform), - &(specs->catchup) + &(specs->catchup), + &(specs->prune) }; int count = sizeof(processArray) / sizeof(processArray[0]); @@ -1184,7 +1215,8 @@ follow_terminate_subprocesses(StreamSpecs *specs) FollowSubProcess *processArray[] = { &(specs->prefetch), &(specs->transform), - &(specs->catchup) + &(specs->catchup), + &(specs->prune) }; int count = sizeof(processArray) / sizeof(processArray[0]); diff --git a/src/bin/pgcopydb/ld_prune.c b/src/bin/pgcopydb/ld_prune.c new file mode 100644 index 000000000..c04a64e17 --- /dev/null +++ b/src/bin/pgcopydb/ld_prune.c @@ -0,0 +1,422 @@ +/* + * src/bin/pgcopydb/ld_prune.c + * CDC file prune watchdog for pgcopydb. + * + * Periodically scans the CDC directory and removes .json and .sql files + * that have already been applied (fileLSN < replayLSN) once total applied + * file bytes exceed the configured threshold. + */ + +#include +#include +#include +#include +#include +#include + +#include "postgres.h" +#include "postgres_fe.h" +#include "access/xlog_internal.h" +#include "access/xlogdefs.h" + +#include "copydb.h" +#include "env_utils.h" +#include "file_utils.h" +#include "ld_prune.h" +#include "ld_stream.h" +#include "log.h" +#include "signals.h" +#include "string_utils.h" + + +#define CDC_PRUNE_CYCLE_SECONDS 30 +#define CDC_PRUNE_MAX_FILES 16384 + + +typedef struct CDCFileEntry +{ + char path[MAXPGPATH]; + uint64_t lsn; + off_t size; + time_t mtime; +} CDCFileEntry; + + +/* + * cdc_file_is_eligible returns true when a CDC file is eligible for pruning: + * its LSN is behind the replay position and it is at least minAgeSeconds old. + */ +bool +cdc_file_is_eligible(uint64_t fileLSN, + uint64_t replayLSN, + time_t fileMtime, + time_t now, + int minAgeSeconds) +{ + return fileLSN < replayLSN && + difftime(now, fileMtime) >= minAgeSeconds; +} + + +/* + * find_oldest_entry scans entries[0..count) for the entry with the smallest + * mtime whose path has not been cleared (i.e. not yet deleted). When + * eligibleOnly is true, only entries that pass cdc_file_is_eligible are + * considered. Returns the index of the oldest match, or -1 if none. + */ +static int +find_oldest_entry(CDCFileEntry *entries, int count, + bool eligibleOnly, uint64_t replayLSN, + time_t now, int minAgeSeconds) +{ + int oldest = -1; + + for (int i = 0; i < count; i++) + { + if (entries[i].path[0] == '\0') + { + continue; + } + + if (eligibleOnly && + !cdc_file_is_eligible(entries[i].lsn, replayLSN, + entries[i].mtime, now, minAgeSeconds)) + { + continue; + } + + if (oldest == -1 || entries[i].mtime < entries[oldest].mtime) + { + oldest = i; + } + } + + return oldest; +} + + +/* + * cdc_prune_loop is the main watchdog loop that runs in a forked subprocess. + * It periodically scans the CDC directory and removes old applied files when + * the total size of applied files exceeds the configured threshold. + */ +bool +cdc_prune_loop(struct StreamSpecs *specs) +{ + uint64_t thresholdBytes = specs->pruneThresholdBytes; + int minAgeSeconds = specs->pruneMinAgeSeconds; + uint32_t WalSegSz = specs->WalSegSz; + char *cdcDir = specs->paths.dir; + + /* + * The watchdog wakes every CDC_PRUNE_CYCLE_SECONDS. Allow shortening the + * cycle via an environment variable, which the integration test relies on + * to observe pruning deterministically (and which is useful for operators + * reacting to disk pressure). + */ + int cycleSeconds = CDC_PRUNE_CYCLE_SECONDS; + + if (env_exists("PGCOPYDB_CDC_PRUNE_CYCLE_SECONDS")) + { + char envval[32] = { 0 }; + int parsed = 0; + + if (get_env_copy("PGCOPYDB_CDC_PRUNE_CYCLE_SECONDS", + envval, sizeof(envval)) && + stringToInt(envval, &parsed) && parsed >= 1) + { + cycleSeconds = parsed; + } + } + + log_info("CDC prune watchdog started: threshold %llu bytes, " + "min age %d seconds, cycle %d seconds, dir %s", + (unsigned long long) thresholdBytes, + minAgeSeconds, + cycleSeconds, + cdcDir); + + while (true) + { + /* + * Sleep in 1-second increments for cycleSeconds, checking signal + * flags each second. + */ + for (int i = 0; i < cycleSeconds; i++) + { + if (asked_to_stop || asked_to_stop_fast || asked_to_quit) + { + log_info("CDC prune watchdog received shutdown signal"); + return true; + } + + pg_usleep(1000000L); /* 1 second */ + } + + if (asked_to_stop || asked_to_stop_fast || asked_to_quit) + { + log_info("CDC prune watchdog received shutdown signal"); + return true; + } + + /* + * If WalSegSz hasn't been populated yet (the receive process + * writes context files on first connect), try to read it now. + */ + if (WalSegSz == 0) + { + if (!file_exists(specs->paths.walsegsizefile)) + { + log_debug("CDC prune: context files not ready yet, " + "will retry next cycle"); + continue; + } + + if (!stream_read_context(specs)) + { + log_warn("CDC prune: failed to read context, " + "will retry next cycle"); + continue; + } + + WalSegSz = specs->WalSegSz; + + if (WalSegSz == 0) + { + log_debug("CDC prune: WalSegSz still unknown, " + "will retry next cycle"); + continue; + } + } + + /* Read the current replay_lsn from the sentinel */ + CopyDBSentinel sentinel = { 0 }; + + if (!sentinel_get(specs->sourceDB, &sentinel)) + { + log_warn("CDC prune: failed to read sentinel, " + "will retry next cycle"); + continue; + } + + uint64_t replayLSN = sentinel.replay_lsn; + + if (replayLSN == 0) + { + log_debug("CDC prune: replay_lsn is 0, nothing to prune"); + continue; + } + + /* Scan the CDC directory */ + DIR *dir = opendir(cdcDir); + + if (dir == NULL) + { + log_warn("CDC prune: failed to open directory %s: %m", cdcDir); + continue; + } + + CDCFileEntry *entries = (CDCFileEntry *) calloc(CDC_PRUNE_MAX_FILES, + sizeof(CDCFileEntry)); + + if (entries == NULL) + { + log_error("CDC prune: failed to allocate file entry array"); + closedir(dir); + continue; + } + + int entryCount = 0; + uint64_t totalAppliedBytes = 0; + struct dirent *de; + + while ((de = readdir(dir)) != NULL) + { + char *name = de->d_name; + size_t nameLen = strlen(name); + + /* only consider .json and .sql files */ + bool isJson = (nameLen > 5 && + strcmp(name + nameLen - 5, ".json") == 0); + + bool isSql = (nameLen > 4 && + strcmp(name + nameLen - 4, ".sql") == 0); + + if (!isJson && !isSql) + { + continue; + } + + /* strip the suffix to get the bare WAL name */ + char barename[MAXPGPATH]; + strlcpy(barename, name, MAXPGPATH); + char *dot = strrchr(barename, '.'); + if (dot != NULL) + { + *dot = '\0'; + } + + if (!IsXLogFileName(barename)) + { + log_debug("CDC prune: skipping non-WAL file %s", name); + continue; + } + + TimeLineID tli; + XLogSegNo segno; + XLogFromFileName(barename, &tli, &segno, WalSegSz); + + uint64_t fileLSN = 0; + XLogSegNoOffsetToRecPtr(segno, 0, WalSegSz, fileLSN); + + /* only consider files whose LSN is behind the replay position */ + if (fileLSN >= replayLSN) + { + continue; + } + + /* stat for size and mtime */ + char fullpath[MAXPGPATH] = { 0 }; + + sformat(fullpath, sizeof(fullpath), "%s/%s", cdcDir, name); + + struct stat st; + + if (stat(fullpath, &st) != 0) + { + log_debug("CDC prune: stat failed for %s: %m", fullpath); + continue; + } + + if (entryCount < CDC_PRUNE_MAX_FILES) + { + totalAppliedBytes += st.st_size; + CDCFileEntry *entry = &entries[entryCount++]; + + strlcpy(entry->path, fullpath, MAXPGPATH); + entry->lsn = fileLSN; + entry->size = st.st_size; + entry->mtime = st.st_mtime; + } + } + + if (entryCount >= CDC_PRUNE_MAX_FILES) + { + log_warn("CDC prune: more than %d applied files found; " + "excess files are not tracked for deletion", + CDC_PRUNE_MAX_FILES); + } + + closedir(dir); + + log_debug("CDC prune: found %d applied files, " + "total %llu bytes (threshold %llu)", + entryCount, + (unsigned long long) totalAppliedBytes, + (unsigned long long) thresholdBytes); + + /* if under threshold, nothing to do */ + if (totalAppliedBytes <= thresholdBytes) + { + free(entries); + continue; + } + + time_t now = time(NULL); + uint64_t bytesToFree = totalAppliedBytes - thresholdBytes; + uint64_t freedBytes = 0; + int deletedCount = 0; + + /* + * First pass: repeatedly find and delete the oldest eligible + * file (age >= minAgeSeconds) until we are under threshold. + */ + for (;;) + { + if (freedBytes >= bytesToFree) + { + break; + } + + int idx = find_oldest_entry(entries, entryCount, + true, replayLSN, + now, minAgeSeconds); + + if (idx == -1) + { + break; + } + + CDCFileEntry *entry = &entries[idx]; + + if (unlink(entry->path) != 0) + { + log_warn("CDC prune: failed to delete %s: %m", entry->path); + entry->path[0] = '\0'; + continue; + } + + freedBytes += entry->size; + deletedCount++; + + log_debug("CDC prune: deleted %s (%lld bytes, age %.0fs)", + entry->path, + (long long) entry->size, + difftime(now, entry->mtime)); + + entry->path[0] = '\0'; + } + + /* + * Second pass: if old-enough files alone couldn't bring us under + * threshold, override the age floor (disk pressure) and delete + * the oldest remaining files regardless of age. + */ + for (;;) + { + if (freedBytes >= bytesToFree) + { + break; + } + + int idx = find_oldest_entry(entries, entryCount, + false, replayLSN, + now, minAgeSeconds); + + if (idx == -1) + { + break; + } + + CDCFileEntry *entry = &entries[idx]; + + log_notice("CDC prune: disk pressure override, " + "deleting young file %s (age %.0fs)", + entry->path, + difftime(now, entry->mtime)); + + if (unlink(entry->path) != 0) + { + log_warn("CDC prune: failed to delete %s: %m", + entry->path); + entry->path[0] = '\0'; + continue; + } + + freedBytes += entry->size; + deletedCount++; + entry->path[0] = '\0'; + } + + if (deletedCount > 0) + { + log_info("CDC prune: deleted %d files, freed %llu bytes", + deletedCount, + (unsigned long long) freedBytes); + } + + free(entries); + } + + return true; +} diff --git a/src/bin/pgcopydb/ld_prune.h b/src/bin/pgcopydb/ld_prune.h new file mode 100644 index 000000000..8892707b4 --- /dev/null +++ b/src/bin/pgcopydb/ld_prune.h @@ -0,0 +1,24 @@ +/* + * src/bin/pgcopydb/ld_prune.h + * CDC file prune watchdog for pgcopydb + */ + +#ifndef LD_PRUNE_H +#define LD_PRUNE_H + +#include +#include +#include + +/* Forward declaration -- full definition in ld_stream.h */ +struct StreamSpecs; + +bool cdc_file_is_eligible(uint64_t fileLSN, + uint64_t replayLSN, + time_t fileMtime, + time_t now, + int minAgeSeconds); + +bool cdc_prune_loop(struct StreamSpecs *specs); + +#endif /* LD_PRUNE_H */ diff --git a/src/bin/pgcopydb/ld_stream.c b/src/bin/pgcopydb/ld_stream.c index c9ce934bf..5d1a67053 100644 --- a/src/bin/pgcopydb/ld_stream.c +++ b/src/bin/pgcopydb/ld_stream.c @@ -52,7 +52,9 @@ stream_init_specs(StreamSpecs *specs, SourceFilters *filters, bool stdin, bool stdout, - bool logSQL) + bool logSQL, + uint64_t pruneThresholdBytes, + int pruneMinAgeSeconds) { /* just copy into StreamSpecs what's been initialized in copySpecs */ specs->mode = mode; @@ -150,6 +152,9 @@ stream_init_specs(StreamSpecs *specs, return false; } + specs->pruneThresholdBytes = pruneThresholdBytes; + specs->pruneMinAgeSeconds = pruneMinAgeSeconds; + log_trace("stream_init_specs: %s(%d)", OutputPluginToString(slot->plugin), specs->pluginOptions.count); @@ -177,9 +182,16 @@ stream_init_specs(StreamSpecs *specs, .pid = -1 }; + FollowSubProcess prune = { + .name = "prune", + .command = &follow_start_prune, + .pid = -1 + }; + specs->prefetch = prefetch; specs->transform = transform; specs->catchup = catchup; + specs->prune = prune; switch (specs->mode) { diff --git a/src/bin/pgcopydb/ld_stream.h b/src/bin/pgcopydb/ld_stream.h index d0c91f0d4..4c6067c59 100644 --- a/src/bin/pgcopydb/ld_stream.h +++ b/src/bin/pgcopydb/ld_stream.h @@ -552,6 +552,11 @@ struct StreamSpecs FollowSubProcess prefetch; FollowSubProcess transform; FollowSubProcess catchup; + FollowSubProcess prune; + + /* CDC file prune configuration */ + uint64_t pruneThresholdBytes; + int pruneMinAgeSeconds; /* transform needs some catalog lookups (pkey, type oid) */ DatabaseCatalog *sourceDB; @@ -590,7 +595,9 @@ bool stream_init_specs(StreamSpecs *specs, SourceFilters *filters, bool stdIn, bool stdOut, - bool logSQL); + bool logSQL, + uint64_t pruneThresholdBytes, + int pruneMinAgeSeconds); bool stream_init_for_mode(StreamSpecs *specs, LogicalStreamMode mode); @@ -806,6 +813,7 @@ bool follow_start_subprocess(StreamSpecs *specs, FollowSubProcess *subprocess); bool follow_start_prefetch(StreamSpecs *specs); bool follow_start_transform(StreamSpecs *specs); bool follow_start_catchup(StreamSpecs *specs); +bool follow_start_prune(StreamSpecs *specs); void follow_exit_early(StreamSpecs *specs); bool follow_wait_subprocesses(StreamSpecs *specs); diff --git a/src/bin/pgcopydb/string_utils.c b/src/bin/pgcopydb/string_utils.c index 71aa40a42..485e05dc5 100644 --- a/src/bin/pgcopydb/string_utils.c +++ b/src/bin/pgcopydb/string_utils.c @@ -362,6 +362,91 @@ IntervalToString(uint64_t millisecs, char *buffer, size_t size) } +/* + * cli_parse_duration parses a duration string like "30s", "15m", "2h" into + * seconds. A bare number (no suffix) is treated as seconds. Returns false on + * parse error. + */ +bool +cli_parse_duration(const char *str, int *seconds) +{ + if (str == NULL || str[0] == '\0') + { + return false; + } + + char *end = NULL; + + errno = 0; + + long value = strtol(str, &end, 10); + + if (end == str || value < 0) + { + return false; + } + else if (errno != 0) + { + return false; + } + + if (*end == '\0') + { + /* bare number, treat as seconds */ + if (value > INT_MAX) + { + return false; + } + *seconds = (int) value; + return true; + } + + if (*(end + 1) != '\0') + { + /* trailing characters after suffix */ + return false; + } + + switch (*end) + { + case 's': + { + if (value > INT_MAX) + { + return false; + } + *seconds = (int) value; + return true; + } + + case 'm': + { + if (value > INT_MAX / 60) + { + return false; + } + *seconds = (int) (value * 60); + return true; + } + + case 'h': + { + if (value > INT_MAX / 3600) + { + return false; + } + *seconds = (int) (value * 3600); + return true; + } + + default: + { + return false; + } + } +} + + /* * countLines returns how many line separators (\n) are found in the given * string. diff --git a/src/bin/pgcopydb/string_utils.h b/src/bin/pgcopydb/string_utils.h index 9629e6168..3b8d61e7c 100644 --- a/src/bin/pgcopydb/string_utils.h +++ b/src/bin/pgcopydb/string_utils.h @@ -38,6 +38,8 @@ bool hexStringToUInt32(const char *str, uint32_t *number); bool IntervalToString(uint64_t millisecs, char *buffer, size_t size); +bool cli_parse_duration(const char *str, int *seconds); + typedef struct LinesBuffer { char *buffer; diff --git a/tests/Makefile b/tests/Makefile index 62b476f8b..0ae21c313 100644 --- a/tests/Makefile +++ b/tests/Makefile @@ -12,6 +12,7 @@ BUILD_ARGS = --build-arg PGVERSION=$(PGVERSION) all: pagila pagila-multi-steps blobs unit filtering filtering-standby extensions \ cdc-wal2json cdc-test-decoding cdc-endpos-between-transaction cdc-low-level \ + cdc-prune \ follow-wal2json follow-standby follow-9.6 follow-data-only follow-target-reconnect \ endpos-in-multi-wal-txn exclude-extension \ blob-snapshot-release follow-defer-indexes fk-not-valid defer-validate-fks \ @@ -47,6 +48,9 @@ cdc-wal2json: build cdc-test-decoding: build $(MAKE) -C $@ +cdc-prune: build + $(MAKE) -C $@ + cdc-endpos-between-transaction: build $(MAKE) -C $@ @@ -104,7 +108,7 @@ build: .PHONY: all build .PHONY: pagila pagila-multi-steps blobs unit filtering filtering-standby extensions -.PHONY: cdc-wal2json cdc-test-decoding cdc-low-level +.PHONY: cdc-wal2json cdc-test-decoding cdc-low-level cdc-prune .PHONY: follow-wal2json follow-standby follow-9.6 follow-target-reconnect .PHONY: endpos-in-multi-wal-txn exclude-extension .PHONY: blob-snapshot-release follow-defer-indexes fk-not-valid defer-validate-fks diff --git a/tests/cdc-prune/Dockerfile b/tests/cdc-prune/Dockerfile new file mode 100644 index 000000000..f0cf93ccb --- /dev/null +++ b/tests/cdc-prune/Dockerfile @@ -0,0 +1,7 @@ +FROM pagila + +WORKDIR /usr/src/pgcopydb +COPY ./copydb.sh copydb.sh + +USER docker +CMD ["/usr/src/pgcopydb/copydb.sh"] diff --git a/tests/cdc-prune/Dockerfile.pg b/tests/cdc-prune/Dockerfile.pg new file mode 100644 index 000000000..52bcaa3a2 --- /dev/null +++ b/tests/cdc-prune/Dockerfile.pg @@ -0,0 +1,9 @@ +ARG PGVERSION=16 +FROM postgres:${PGVERSION} + +ARG PGVERSION=16 +USER root +RUN apt-get update \ + && apt-get install -y --no-install-recommends postgresql-${PGVERSION}-wal2json \ + && rm -rf /var/lib/apt/lists/* +USER postgres diff --git a/tests/cdc-prune/Makefile b/tests/cdc-prune/Makefile new file mode 100644 index 000000000..5daf0cb69 --- /dev/null +++ b/tests/cdc-prune/Makefile @@ -0,0 +1,20 @@ +# Copyright (c) 2021 The PostgreSQL Global Development Group. +# Licensed under the PostgreSQL License. + +COMPOSE_EXIT = --exit-code-from=test --abort-on-container-exit + +test: down run down ; + +up: down build + $(DOCKER) compose up $(COMPOSE_EXIT) + +run: build + $(DOCKER) compose run test + +down: + $(DOCKER) compose down + +build: + $(DOCKER) compose build + +.PHONY: run down build test diff --git a/tests/cdc-prune/compose.yaml b/tests/cdc-prune/compose.yaml new file mode 100644 index 000000000..781ff2aa0 --- /dev/null +++ b/tests/cdc-prune/compose.yaml @@ -0,0 +1,34 @@ +services: + source: + build: + context: . + dockerfile: Dockerfile.pg + expose: + - 5432 + environment: + POSTGRES_USER: postgres + POSTGRES_PASSWORD: h4ckm3 + POSTGRES_HOST_AUTH_METHOD: trust + command: > + -c wal_level=logical + target: + image: postgres:${PGVERSION:-16} + expose: + - 5432 + environment: + POSTGRES_USER: postgres + POSTGRES_PASSWORD: h4ckm3 + POSTGRES_HOST_AUTH_METHOD: trust + test: + build: + context: . + dockerfile: Dockerfile + environment: + PGCOPYDB_SOURCE_PGURI: postgres://postgres:h4ckm3@source/postgres + PGCOPYDB_TARGET_PGURI: postgres://postgres:h4ckm3@target/postgres + PGCOPYDB_TABLE_JOBS: 4 + PGCOPYDB_INDEX_JOBS: 2 + PGCOPYDB_OUTPUT_PLUGIN: wal2json + depends_on: + - source + - target diff --git a/tests/cdc-prune/copydb.sh b/tests/cdc-prune/copydb.sh new file mode 100755 index 000000000..aabe3b1eb --- /dev/null +++ b/tests/cdc-prune/copydb.sh @@ -0,0 +1,107 @@ +#! /bin/bash + +set -x +set -e + +# Disable pager for psql to avoid hanging in non-interactive environments +export PAGER=cat + +# This script expects the following environment variables to be set: +# +# - PGCOPYDB_SOURCE_PGURI +# - PGCOPYDB_TARGET_PGURI +# - PGCOPYDB_TABLE_JOBS +# - PGCOPYDB_INDEX_JOBS + +# make sure source and target databases are ready +pgcopydb ping + +# create a simple test table +psql -d ${PGCOPYDB_SOURCE_PGURI} -c "CREATE TABLE test_data (id serial primary key, val text)" + +# create the replication slot that captures all the changes +coproc ( pgcopydb snapshot --follow ) + +sleep 1 + +# now setup the replication origin (target) and the pgcopydb.sentinel (source) +pgcopydb stream setup + +# pgcopydb clone uses the environment variables +pgcopydb clone + +kill -TERM ${COPROC_PID} +wait ${COPROC_PID} || true + +# inject enough data to produce multiple WAL segments worth of CDC files, +# comfortably larger than the prune threshold used below +for i in $(seq 1 200); do + psql -d ${PGCOPYDB_SOURCE_PGURI} -c \ + "INSERT INTO test_data (val) SELECT md5(random()::text) FROM generate_series(1, 200)" +done + +# allow applying/replaying changes; the follow below streams live (no endpos) +pgcopydb stream sentinel set apply + +CDCDIR=/tmp/pgcopydb/cdc +followlog=/tmp/prune-follow.log + +# Run follow in the background with a small prune threshold, a short min-age, +# and a short prune cycle. --not-consistent lets follow proceed without the +# snapshot exported (and since released) during the setup phase. As apply +# advances the replay LSN past the injected WAL, the watchdog (every 2s) should +# delete the applied .json/.sql files once their total exceeds the threshold. +PGCOPYDB_CDC_PRUNE_CYCLE_SECONDS=2 \ +pgcopydb follow --resume --not-consistent \ + --prune-threshold 64kB \ + --prune-min-age 1s \ + -vv > "${followlog}" 2>&1 & +follow_pid=$! + +# Poll (up to ~90s) for the watchdog to report deleting applied CDC files. +pruned=0 +for i in $(seq 1 45); do + if grep -qE "CDC prune: deleted [1-9][0-9]* files, freed [1-9]" "${followlog}"; then + pruned=1 + break + fi + + # stop early if the follow process exited on its own + kill -0 ${follow_pid} 2>/dev/null || break + + sleep 2 +done + +# Stop the background follow (and its subprocesses) before asserting / tearing +# down. follow does not reliably exit on a single SIGTERM, so escalate to +# SIGKILL and do not block on wait (which would hang). +kill -TERM ${follow_pid} 2>/dev/null || true +for i in $(seq 1 5); do + kill -0 ${follow_pid} 2>/dev/null || break + sleep 1 +done +pkill -KILL -P ${follow_pid} 2>/dev/null || true +kill -KILL ${follow_pid} 2>/dev/null || true + +# let the replication slot go inactive so stream cleanup can drop it +sleep 3 + +echo "=== prune watchdog log lines ===" +grep "CDC prune" "${followlog}" || true + +remaining=$(find ${CDCDIR} -name '*.json' -o -name '*.sql' 2>/dev/null | wc -l | tr -d ' ') +echo "Remaining CDC files after pruning: ${remaining}" + +# Assert the watchdog actually deleted applied CDC files, freeing a non-zero +# number of files/bytes (not just that follow ran with the flags enabled). +if [ "${pruned}" != "1" ]; then + echo "ERROR: prune watchdog did not delete any applied CDC files" + echo "--- follow log tail ---" + tail -60 "${followlog}" || true + exit 1 +fi + +echo "CDC prune integration test passed" + +# verify the stream cleanup teardown command still works +pgcopydb stream cleanup