31 #define RECONNECT_SLEEP_TIME 5 49 static void usage(
void);
55 bool segment_finished);
65 #define IsCompressXLogFileName(fname) \ 66 (strlen(fname) == XLOG_FNAME_LEN + strlen(".gz") && \ 67 strspn(fname, "0123456789ABCDEF") == XLOG_FNAME_LEN && \ 68 strcmp((fname) + XLOG_FNAME_LEN, ".gz") == 0) 69 #define IsPartialCompressXLogFileName(fname) \ 70 (strlen(fname) == XLOG_FNAME_LEN + strlen(".gz.partial") && \ 71 strspn(fname, "0123456789ABCDEF") == XLOG_FNAME_LEN && \ 72 strcmp((fname) + XLOG_FNAME_LEN, ".gz.partial") == 0) 77 printf(
_(
"%s receives PostgreSQL streaming write-ahead logs.\n\n"),
82 printf(
_(
" -D, --directory=DIR receive write-ahead log files into this directory\n"));
83 printf(
_(
" -E, --endpos=LSN exit after receiving the specified LSN\n"));
84 printf(
_(
" --if-not-exists do not error if slot already exists when creating a slot\n"));
85 printf(
_(
" -n, --no-loop do not loop on connection lost\n"));
86 printf(
_(
" --no-sync do not wait for changes to be written safely to disk\n"));
87 printf(
_(
" -s, --status-interval=SECS\n" 89 printf(
_(
" -S, --slot=SLOTNAME replication slot to use\n"));
90 printf(
_(
" --synchronous flush write-ahead log immediately after writing\n"));
91 printf(
_(
" -v, --verbose output verbose messages\n"));
92 printf(
_(
" -V, --version output version information, then exit\n"));
93 printf(
_(
" -Z, --compress=0-9 compress logs with given compression level\n"));
94 printf(
_(
" -?, --help show this help, then exit\n"));
95 printf(
_(
"\nConnection options:\n"));
96 printf(
_(
" -d, --dbname=CONNSTR connection string\n"));
97 printf(
_(
" -h, --host=HOSTNAME database server host or socket directory\n"));
98 printf(
_(
" -p, --port=PORT database server port number\n"));
99 printf(
_(
" -U, --username=NAME connect as specified database user\n"));
100 printf(
_(
" -w, --no-password never prompt for password\n"));
101 printf(
_(
" -W, --password force password prompt (should happen automatically)\n"));
102 printf(
_(
"\nOptional actions:\n"));
103 printf(
_(
" --create-slot create a new replication slot (for the slot's name see --slot)\n"));
104 printf(
_(
" --drop-slot drop the replication slot (for the slot's name see --slot)\n"));
105 printf(
_(
"\nReport bugs to <pgsql-bugs@lists.postgresql.org>.\n"));
111 static uint32 prevtimeline = 0;
115 if (
verbose && segment_finished)
116 pg_log_info(
"finished segment at %X/%X (timeline %u)",
123 pg_log_info(
"stopped log streaming at %X/%X (timeline %u)",
138 if (
verbose && prevtimeline != 0 && prevtimeline != timeline)
143 prevtimeline = timeline;
164 Assert(dest_folder != NULL);
182 Assert(dest_dir != NULL && dest_folder != NULL);
185 pg_log_error(
"could not close directory \"%s\": %m", dest_folder);
205 bool high_ispartial =
false;
209 while (errno = 0, (dirent =
readdir(dir)) != NULL)
257 if (!ispartial && !iscompress)
263 if (
stat(fullpath, &statbuf) != 0)
265 pg_log_error(
"could not stat file \"%s\": %m", fullpath);
271 pg_log_warning(
"segment file \"%s\" has incorrect size %d, skipping",
272 dirent->
d_name, (
int) statbuf.st_size);
276 else if (!ispartial && iscompress)
286 fd = open(fullpath, O_RDONLY |
PG_BINARY, 0);
289 pg_log_error(
"could not open compressed file \"%s\": %m",
293 if (lseek(fd, (off_t) (-4), SEEK_END) < 0)
295 pg_log_error(
"could not seek in compressed file \"%s\": %m",
299 r =
read(fd, (
char *) buf,
sizeof(buf));
300 if (r !=
sizeof(buf))
303 pg_log_error(
"could not read compressed file \"%s\": %m",
306 pg_log_error(
"could not read compressed file \"%s\": read %d of %zu",
307 fullpath, r,
sizeof(buf));
312 bytes_out = (buf[3] << 24) | (buf[2] << 16) |
313 (buf[1] << 8) | buf[0];
317 pg_log_warning(
"compressed segment file \"%s\" has incorrect uncompressed size %d, skipping",
318 dirent->
d_name, bytes_out);
324 if ((segno > high_segno) ||
325 (segno == high_segno && tli > high_tli) ||
326 (segno == high_segno && tli == high_tli && high_ispartial && !ispartial))
330 high_ispartial = ispartial;
373 MemSet(&stream, 0,
sizeof(stream));
421 pg_log_info(
"starting log streaming at %X/%X (timeline %u)",
440 pg_log_info(
"could not finish writing WAL files: %m");
469 static struct option long_options[] = {
506 if (strcmp(argv[1],
"--help") == 0 || strcmp(argv[1],
"-?") == 0)
511 else if (strcmp(argv[1],
"-V") == 0 ||
512 strcmp(argv[1],
"--version") == 0)
514 puts(
"pg_receivewal (PostgreSQL) " PG_VERSION);
519 while ((c =
getopt_long(argc, argv,
"D:d:E:h:p:U:s:S:nwWvZ:",
520 long_options, &option_index)) != -1)
562 if (sscanf(
optarg,
"%X/%X", &hi, &lo) != 2)
567 endpos = ((uint64) hi) << 32 | lo;
577 if (compresslevel < 0 || compresslevel > 9)
604 fprintf(stderr,
_(
"Try \"%s --help\" for more information.\n"),
615 pg_log_error(
"too many command-line arguments (first is \"%s\")",
617 fprintf(stderr,
_(
"Try \"%s --help\" for more information.\n"),
624 pg_log_error(
"cannot use --create-slot together with --drop-slot");
625 fprintf(stderr,
_(
"Try \"%s --help\" for more information.\n"),
633 pg_log_error(
"%s needs a slot to be specified using --slot",
635 fprintf(stderr,
_(
"Try \"%s --help\" for more information.\n"),
642 pg_log_error(
"cannot use --synchronous together with --no-sync");
643 fprintf(stderr,
_(
"Try \"%s --help\" for more information.\n"),
654 fprintf(stderr,
_(
"Try \"%s --help\" for more information.\n"),
662 pg_log_error(
"this build does not support compression");
718 pg_log_error(
"replication connection using slot \"%s\" is unexpectedly database specific",
772 pg_log_info(
"disconnected; waiting %d seconds to try again",
static char * replication_slot
#define InvalidXLogRecPtr
int main(int argc, char **argv)
static int standby_message_timeout
const char * get_progname(const char *argv0)
static void DropReplicationSlot(DropReplicationSlotCmd *cmd)
#define pg_log_error(...)
int getopt_long(int argc, char *const argv[], const char *optstring, const struct option *longopts, int *longindex)
void pg_logging_init(const char *argv0)
#define IsPartialCompressXLogFileName(fname)
bool RunIdentifySystem(PGconn *conn, char **sysid, TimeLineID *starttli, XLogRecPtr *startpos, char **db_name)
void PQfinish(PGconn *conn)
#define MemSet(start, val, len)
static bool stop_streaming(XLogRecPtr segendpos, uint32 timeline, bool segment_finished)
static void close_destination_dir(DIR *dest_dir, char *dest_folder)
static int fd(const char *x, int i)
WalWriteMethod * CreateWalDirectoryMethod(const char *basedir, int compression, bool sync)
static void sigint_handler(int signum)
static bool slot_exists_ok
static void StreamLog(void)
void pg_usleep(long microsec)
#define required_argument
#define IsXLogFileName(fname)
bool RetrieveWalSegSize(PGconn *conn)
DIR * opendir(const char *)
#define XLogFromFileName(fname, tli, logSegNo, wal_segsz_bytes)
static XLogRecPtr FindStreamingStart(uint32 *tli)
char * pg_strdup(const char *in)
static volatile bool time_to_stop
stream_stop_callback stream_stop
WalWriteMethod * walmethod
#define XLogRecPtrIsInvalid(r)
#define PG_TEXTDOMAIN(domain)
#define XLogSegmentOffset(xlogptr, wal_segsz_bytes)
static void disconnect_atexit(void)
pqsigfunc pqsignal(int signum, pqsigfunc handler)
static DIR * get_destination_dir(char *dest_folder)
PGconn * GetConnection(UserMapping *user, bool will_prep_stmt)
#define XLogSegNoOffsetToRecPtr(segno, offset, wal_segsz_bytes, dest)
#define Assert(condition)
bool ReceiveXlogStream(PGconn *conn, StreamCtl *stream)
#define RECONNECT_SLEEP_TIME
struct dirent * readdir(DIR *)
static bool do_create_slot
static void CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
#define IsPartialXLogFileName(fname)
void set_pglocale_pgservice(const char *argv0, const char *app)
int standby_message_timeout
#define pg_log_warning(...)
#define IsCompressXLogFileName(fname)
void FreeWalDirectoryMethod(void)
bool CheckServerVersionForStreaming(PGconn *conn)