granicus.if.org Git - postgresql/blob - src/bin/pg_basebackup/pg_receivexlog.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * pg_receivexlog.c - receive streaming transaction log data and write it
   4  *                                        to a local file.
   5  *
   6  * Author: Magnus Hagander <magnus@hagander.net>
   7  *
   8  * Portions Copyright (c) 1996-2014, PostgreSQL Global Development Group
   9  *
  10  * IDENTIFICATION
  11  *                src/bin/pg_basebackup/pg_receivexlog.c
  12  *-------------------------------------------------------------------------
  13  */
  14
  15 #include "postgres_fe.h"
  16
  17 #include <dirent.h>
  18 #include <signal.h>
  19 #include <sys/stat.h>
  20 #include <sys/types.h>
  21 #include <unistd.h>
  22
  23 #include "libpq-fe.h"
  24 #include "access/xlog_internal.h"
  25 #include "getopt_long.h"
  26
  27 #include "receivelog.h"
  28 #include "streamutil.h"
  29
  30
  31 /* Time to sleep between reconnection attempts */
  32 #define RECONNECT_SLEEP_TIME 5
  33
  34 /* Global options */
  35 static char *basedir = NULL;
  36 static int      verbose = 0;
  37 static int      noloop = 0;
  38 static int      standby_message_timeout = 10 * 1000;            /* 10 sec = default */
  39 static volatile bool time_to_abort = false;
  40
  41
  42 static void usage(void);
  43 static XLogRecPtr FindStreamingStart(uint32 *tli);
  44 static void StreamLog();
  45 static bool stop_streaming(XLogRecPtr segendpos, uint32 timeline,
  46                            bool segment_finished);
  47
  48 #define disconnect_and_exit(code)                               \
  49         {                                                                                       \
  50         if (conn != NULL) PQfinish(conn);                       \
  51         exit(code);                                                                     \
  52         }
  53
  54
  55 static void
  56 usage(void)
  57 {
  58         printf(_("%s receives PostgreSQL streaming transaction logs.\n\n"),
  59                    progname);
  60         printf(_("Usage:\n"));
  61         printf(_("  %s [OPTION]...\n"), progname);
  62         printf(_("\nOptions:\n"));
  63         printf(_("  -D, --directory=DIR    receive transaction log files into this directory\n"));
  64         printf(_("  -n, --no-loop          do not loop on connection lost\n"));
  65         printf(_("  -v, --verbose          output verbose messages\n"));
  66         printf(_("  -V, --version          output version information, then exit\n"));
  67         printf(_("  -?, --help             show this help, then exit\n"));
  68         printf(_("\nConnection options:\n"));
  69         printf(_("  -d, --dbname=CONNSTR   connection string\n"));
  70         printf(_("  -h, --host=HOSTNAME    database server host or socket directory\n"));
  71         printf(_("  -p, --port=PORT        database server port number\n"));
  72         printf(_("  -s, --status-interval=INTERVAL\n"
  73                          "                         time between status packets sent to server (in seconds)\n"));
  74         printf(_("  -U, --username=NAME    connect as specified database user\n"));
  75         printf(_("  -w, --no-password      never prompt for password\n"));
  76         printf(_("  -W, --password         force password prompt (should happen automatically)\n"));
  77         printf(_("      --slot=SLOTNAME    replication slot to use\n"));
  78         printf(_("\nReport bugs to <pgsql-bugs@postgresql.org>.\n"));
  79 }
  80
  81 static bool
  82 stop_streaming(XLogRecPtr xlogpos, uint32 timeline, bool segment_finished)
  83 {
  84         static uint32 prevtimeline = 0;
  85         static XLogRecPtr prevpos = InvalidXLogRecPtr;
  86
  87         /* we assume that we get called once at the end of each segment */
  88         if (verbose && segment_finished)
  89                 fprintf(stderr, _("%s: finished segment at %X/%X (timeline %u)\n"),
  90                                 progname, (uint32) (xlogpos >> 32), (uint32) xlogpos,
  91                                 timeline);
  92
  93         /*
  94          * Note that we report the previous, not current, position here. After a
  95          * timeline switch, xlogpos points to the beginning of the segment because
  96          * that's where we always begin streaming. Reporting the end of previous
  97          * timeline isn't totally accurate, because the next timeline can begin
  98          * slightly before the end of the WAL that we received on the previous
  99          * timeline, but it's close enough for reporting purposes.
 100          */
 101         if (prevtimeline != 0 && prevtimeline != timeline)
 102                 fprintf(stderr, _("%s: switched to timeline %u at %X/%X\n"),
 103                                 progname, timeline,
 104                                 (uint32) (prevpos >> 32), (uint32) prevpos);
 105
 106         prevtimeline = timeline;
 107         prevpos = xlogpos;
 108
 109         if (time_to_abort)
 110         {
 111                 fprintf(stderr, _("%s: received interrupt signal, exiting\n"),
 112                                 progname);
 113                 return true;
 114         }
 115         return false;
 116 }
 117
 118 /*
 119  * Determine starting location for streaming, based on any existing xlog
 120  * segments in the directory. We start at the end of the last one that is
 121  * complete (size matches XLogSegSize), on the timeline with highest ID.
 122  *
 123  * If there are no WAL files in the directory, returns InvalidXLogRecPtr.
 124  */
 125 static XLogRecPtr
 126 FindStreamingStart(uint32 *tli)
 127 {
 128         DIR                *dir;
 129         struct dirent *dirent;
 130         XLogSegNo       high_segno = 0;
 131         uint32          high_tli = 0;
 132         bool            high_ispartial = false;
 133
 134         dir = opendir(basedir);
 135         if (dir == NULL)
 136         {
 137                 fprintf(stderr, _("%s: could not open directory \"%s\": %s\n"),
 138                                 progname, basedir, strerror(errno));
 139                 disconnect_and_exit(1);
 140         }
 141
 142         while (errno = 0, (dirent = readdir(dir)) != NULL)
 143         {
 144                 uint32          tli;
 145                 XLogSegNo       segno;
 146                 bool            ispartial;
 147
 148                 /*
 149                  * Check if the filename looks like an xlog file, or a .partial file.
 150                  * Xlog files are always 24 characters, and .partial files are 32
 151                  * characters.
 152                  */
 153                 if (strlen(dirent->d_name) == 24)
 154                 {
 155                         if (strspn(dirent->d_name, "0123456789ABCDEF") != 24)
 156                                 continue;
 157                         ispartial = false;
 158                 }
 159                 else if (strlen(dirent->d_name) == 32)
 160                 {
 161                         if (strspn(dirent->d_name, "0123456789ABCDEF") != 24)
 162                                 continue;
 163                         if (strcmp(&dirent->d_name[24], ".partial") != 0)
 164                                 continue;
 165                         ispartial = true;
 166                 }
 167                 else
 168                         continue;
 169
 170                 /*
 171                  * Looks like an xlog file. Parse its position.
 172                  */
 173                 XLogFromFileName(dirent->d_name, &tli, &segno);
 174
 175                 /*
 176                  * Check that the segment has the right size, if it's supposed to be
 177                  * completed.
 178                  */
 179                 if (!ispartial)
 180                 {
 181                         struct stat statbuf;
 182                         char            fullpath[MAXPGPATH];
 183
 184                         snprintf(fullpath, sizeof(fullpath), "%s/%s", basedir, dirent->d_name);
 185                         if (stat(fullpath, &statbuf) != 0)
 186                         {
 187                                 fprintf(stderr, _("%s: could not stat file \"%s\": %s\n"),
 188                                                 progname, fullpath, strerror(errno));
 189                                 disconnect_and_exit(1);
 190                         }
 191
 192                         if (statbuf.st_size != XLOG_SEG_SIZE)
 193                         {
 194                                 fprintf(stderr,
 195                                                 _("%s: segment file \"%s\" has incorrect size %d, skipping\n"),
 196                                                 progname, dirent->d_name, (int) statbuf.st_size);
 197                                 continue;
 198                         }
 199                 }
 200
 201                 /* Looks like a valid segment. Remember that we saw it. */
 202                 if ((segno > high_segno) ||
 203                         (segno == high_segno && tli > high_tli) ||
 204                         (segno == high_segno && tli == high_tli && high_ispartial && !ispartial))
 205                 {
 206                         high_segno = segno;
 207                         high_tli = tli;
 208                         high_ispartial = ispartial;
 209                 }
 210         }
 211
 212 #ifdef WIN32
 213         /* Bug in old Mingw dirent.c;  fixed in mingw-runtime-3.2, 2003-10-10 */
 214         if (GetLastError() == ERROR_NO_MORE_FILES)
 215                 errno = 0;
 216 #endif
 217
 218         if (errno)
 219         {
 220                 fprintf(stderr, _("%s: could not read directory \"%s\": %s\n"),
 221                                 progname, basedir, strerror(errno));
 222                 disconnect_and_exit(1);
 223         }
 224
 225         if (closedir(dir))
 226         {
 227                 fprintf(stderr, _("%s: could not close directory \"%s\": %s\n"),
 228                                 progname, basedir, strerror(errno));
 229                 disconnect_and_exit(1);
 230         }
 231
 232         if (high_segno > 0)
 233         {
 234                 XLogRecPtr      high_ptr;
 235
 236                 /*
 237                  * Move the starting pointer to the start of the next segment, if
 238                  * the highest one we saw was completed. Otherwise start streaming
 239                  * from the beginning of the .partial segment.
 240                  */
 241                 if (!high_ispartial)
 242                         high_segno++;
 243
 244                 XLogSegNoOffsetToRecPtr(high_segno, 0, high_ptr);
 245
 246                 *tli = high_tli;
 247                 return high_ptr;
 248         }
 249         else
 250                 return InvalidXLogRecPtr;
 251 }
 252
 253 /*
 254  * Start the log streaming
 255  */
 256 static void
 257 StreamLog(void)
 258 {
 259         PGresult   *res;
 260         XLogRecPtr      startpos;
 261         uint32          starttli;
 262         XLogRecPtr      serverpos;
 263         uint32          servertli;
 264         uint32          hi,
 265                                 lo;
 266
 267         /*
 268          * Connect in replication mode to the server
 269          */
 270         conn = GetConnection();
 271         if (!conn)
 272                 /* Error message already written in GetConnection() */
 273                 return;
 274
 275         if (!CheckServerVersionForStreaming(conn))
 276         {
 277                 /*
 278                  * Error message already written in CheckServerVersionForStreaming().
 279                  * There's no hope of recovering from a version mismatch, so don't
 280                  * retry.
 281                  */
 282                 disconnect_and_exit(1);
 283         }
 284
 285         /*
 286          * Run IDENTIFY_SYSTEM so we can get the timeline and current xlog
 287          * position.
 288          */
 289         res = PQexec(conn, "IDENTIFY_SYSTEM");
 290         if (PQresultStatus(res) != PGRES_TUPLES_OK)
 291         {
 292                 fprintf(stderr, _("%s: could not send replication command \"%s\": %s"),
 293                                 progname, "IDENTIFY_SYSTEM", PQerrorMessage(conn));
 294                 disconnect_and_exit(1);
 295         }
 296         if (PQntuples(res) != 1 || PQnfields(res) < 3)
 297         {
 298                 fprintf(stderr,
 299                                 _("%s: could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields\n"),
 300                                 progname, PQntuples(res), PQnfields(res), 1, 3);
 301                 disconnect_and_exit(1);
 302         }
 303         servertli = atoi(PQgetvalue(res, 0, 1));
 304         if (sscanf(PQgetvalue(res, 0, 2), "%X/%X", &hi, &lo) != 2)
 305         {
 306                 fprintf(stderr,
 307                                 _("%s: could not parse transaction log location \"%s\"\n"),
 308                                 progname, PQgetvalue(res, 0, 2));
 309                 disconnect_and_exit(1);
 310         }
 311         serverpos = ((uint64) hi) << 32 | lo;
 312         PQclear(res);
 313
 314         /*
 315          * Figure out where to start streaming.
 316          */
 317         startpos = FindStreamingStart(&starttli);
 318         if (startpos == InvalidXLogRecPtr)
 319         {
 320                 startpos = serverpos;
 321                 starttli = servertli;
 322         }
 323
 324         /*
 325          * Always start streaming at the beginning of a segment
 326          */
 327         startpos -= startpos % XLOG_SEG_SIZE;
 328
 329         /*
 330          * Start the replication
 331          */
 332         if (verbose)
 333                 fprintf(stderr,
 334                                 _("%s: starting log streaming at %X/%X (timeline %u)\n"),
 335                                 progname, (uint32) (startpos >> 32), (uint32) startpos,
 336                                 starttli);
 337
 338         ReceiveXlogStream(conn, startpos, starttli, NULL, basedir,
 339                                           stop_streaming, standby_message_timeout, ".partial");
 340
 341         PQfinish(conn);
 342 }
 343
 344 /*
 345  * When sigint is called, just tell the system to exit at the next possible
 346  * moment.
 347  */
 348 #ifndef WIN32
 349
 350 static void
 351 sigint_handler(int signum)
 352 {
 353         time_to_abort = true;
 354 }
 355 #endif
 356
 357 int
 358 main(int argc, char **argv)
 359 {
 360         static struct option long_options[] = {
 361                 {"help", no_argument, NULL, '?'},
 362                 {"version", no_argument, NULL, 'V'},
 363                 {"directory", required_argument, NULL, 'D'},
 364                 {"dbname", required_argument, NULL, 'd'},
 365                 {"host", required_argument, NULL, 'h'},
 366                 {"port", required_argument, NULL, 'p'},
 367                 {"username", required_argument, NULL, 'U'},
 368                 {"no-loop", no_argument, NULL, 'n'},
 369                 {"no-password", no_argument, NULL, 'w'},
 370                 {"password", no_argument, NULL, 'W'},
 371                 {"status-interval", required_argument, NULL, 's'},
 372                 {"slot", required_argument, NULL, 'S'},
 373                 {"verbose", no_argument, NULL, 'v'},
 374                 {NULL, 0, NULL, 0}
 375         };
 376
 377         int                     c;
 378         int                     option_index;
 379
 380         progname = get_progname(argv[0]);
 381         set_pglocale_pgservice(argv[0], PG_TEXTDOMAIN("pg_receivexlog"));
 382
 383         if (argc > 1)
 384         {
 385                 if (strcmp(argv[1], "--help") == 0 || strcmp(argv[1], "-?") == 0)
 386                 {
 387                         usage();
 388                         exit(0);
 389                 }
 390                 else if (strcmp(argv[1], "-V") == 0 ||
 391                                  strcmp(argv[1], "--version") == 0)
 392                 {
 393                         puts("pg_receivexlog (PostgreSQL) " PG_VERSION);
 394                         exit(0);
 395                 }
 396         }
 397
 398         while ((c = getopt_long(argc, argv, "D:d:h:p:U:s:nwWv",
 399                                                         long_options, &option_index)) != -1)
 400         {
 401                 switch (c)
 402                 {
 403                         case 'D':
 404                                 basedir = pg_strdup(optarg);
 405                                 break;
 406                         case 'd':
 407                                 connection_string = pg_strdup(optarg);
 408                                 break;
 409                         case 'h':
 410                                 dbhost = pg_strdup(optarg);
 411                                 break;
 412                         case 'p':
 413                                 if (atoi(optarg) <= 0)
 414                                 {
 415                                         fprintf(stderr, _("%s: invalid port number \"%s\"\n"),
 416                                                         progname, optarg);
 417                                         exit(1);
 418                                 }
 419                                 dbport = pg_strdup(optarg);
 420                                 break;
 421                         case 'U':
 422                                 dbuser = pg_strdup(optarg);
 423                                 break;
 424                         case 'w':
 425                                 dbgetpassword = -1;
 426                                 break;
 427                         case 'W':
 428                                 dbgetpassword = 1;
 429                                 break;
 430                         case 's':
 431                                 standby_message_timeout = atoi(optarg) * 1000;
 432                                 if (standby_message_timeout < 0)
 433                                 {
 434                                         fprintf(stderr, _("%s: invalid status interval \"%s\"\n"),
 435                                                         progname, optarg);
 436                                         exit(1);
 437                                 }
 438                                 break;
 439                         case 'S':
 440                                 replication_slot = pg_strdup(optarg);
 441                                 break;
 442                         case 'n':
 443                                 noloop = 1;
 444                                 break;
 445                         case 'v':
 446                                 verbose++;
 447                                 break;
 448                         default:
 449
 450                                 /*
 451                                  * getopt_long already emitted a complaint
 452                                  */
 453                                 fprintf(stderr, _("Try \"%s --help\" for more information.\n"),
 454                                                 progname);
 455                                 exit(1);
 456                 }
 457         }
 458
 459         /*
 460          * Any non-option arguments?
 461          */
 462         if (optind < argc)
 463         {
 464                 fprintf(stderr,
 465                                 _("%s: too many command-line arguments (first is \"%s\")\n"),
 466                                 progname, argv[optind]);
 467                 fprintf(stderr, _("Try \"%s --help\" for more information.\n"),
 468                                 progname);
 469                 exit(1);
 470         }
 471
 472         /*
 473          * Required arguments
 474          */
 475         if (basedir == NULL)
 476         {
 477                 fprintf(stderr, _("%s: no target directory specified\n"), progname);
 478                 fprintf(stderr, _("Try \"%s --help\" for more information.\n"),
 479                                 progname);
 480                 exit(1);
 481         }
 482
 483 #ifndef WIN32
 484         pqsignal(SIGINT, sigint_handler);
 485 #endif
 486
 487         while (true)
 488         {
 489                 StreamLog();
 490                 if (time_to_abort)
 491                 {
 492                         /*
 493                          * We've been Ctrl-C'ed. That's not an error, so exit without an
 494                          * errorcode.
 495                          */
 496                         exit(0);
 497                 }
 498                 else if (noloop)
 499                 {
 500                         fprintf(stderr, _("%s: disconnected\n"), progname);
 501                         exit(1);
 502                 }
 503                 else
 504                 {
 505                         fprintf(stderr,
 506                         /* translator: check source for value for %d */
 507                                         _("%s: disconnected; waiting %d seconds to try again\n"),
 508                                         progname, RECONNECT_SLEEP_TIME);
 509                         pg_usleep(RECONNECT_SLEEP_TIME * 1000000);
 510                 }
 511         }
 512 }