lustre/utils/lustre_rsync.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2009, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2012, 2014, Intel Corporation.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * lustre/utils/lustre_rsync.c
  37  *
  38  * Author: Kalpak Shah <Kalpak.Shah@Sun.COM>
  39  * Author: Manoj Joseph <Manoj.Joseph@Sun.COM>
  40  */
  41
  42 /*
  43  * - lustre_rsync is a tool for replicating a lustre filesystem.
  44  *
  45  * - The source-fs is a live lustre filesystem. It is not a
  46  * snapshot. It is mounted and undergoing changes
  47  *
  48  * - The target-fs is a copy of the source-fs from the past. Let's
  49  * call this point, the 'sync point'.
  50  *
  51  * - There is a changelog of all metadata operations that happened on
  52  * the filesystem since the 'sync point'.
  53  *
  54  * - lustre_rsync replicates all the operations saved in the changelog
  55  * on to the target filesystem to make it identical to the source.
  56  *
  57  * To facilitate replication, the lustre filesystem provides
  58  *    a) a way to get the current filesystem path of a given FID
  59  *    b) a way to open files by specifying its FID
  60  *
  61  * The changelog only has a limited amount of information.
  62  *  tfid - The FID of the target file
  63  *  pfid - The FID of the parent of the target file (at the time of
  64  *         the operation)
  65  *  sfid - The FID of the source file
  66  *  spfid - The FID of the parent of the source file
  67  *  name - The name of the target file (at the time of the operation), the name
  68  *         of the source file is appended (delimited with '\0') if this
  69  *         operation involves a source
  70  *
  71  * With just this information, it is not alwasy possible to determine
  72  * the file paths for each operation. For instance, if pfid does not
  73  * exist on the source-fs (due to a subsequent deletion), its path
  74  * cannot be queried. In such cases, lustre_rsync keeps the files in a
  75  * special directory ("/.lustrerepl"). Once all the operations in a
  76  * changelog are replayed, all the files in this special directory
  77  * will get moved to the location as in the source-fs.
  78  *
  79  * Shorthand used: f2p(fid) = fid2path(fid)
  80  *
  81  * The following are the metadata operations of interest.
  82  * 1. creat
  83  *    If tfid is absent on the source-fs, ignore this operation
  84  *    If pfid is absent on the source-fs [or]
  85  *    if f2p(pfid) is not present on target-fs [or]
  86  *    if f2p(pfid)+name != f2p(tfid)
  87  *      creat .lustrerepl/tfid
  88  *      track [pfid,tfid,name]
  89  *    Else
  90  *      creat f2p[tfid]
  91  *
  92  * 2. remove
  93  *    If .lustrerepl/[tfid] is present on the target
  94  *      rm .lustrerepl/[tfid]
  95  *    Else if pfid is present on the source-fs,
  96  *      if f2p(pfid)+name is present,
  97  *        rm f2p(pfid)+name
  98  *
  99  * 3. move (spfid,sname) to (pfid,name)
 100  *    If pfid is present
 101  *      if spfid is also present, mv (spfid,sname) to (pfid,name)
 102  *      else mv .lustrerepl/[sfid] to (pfid,name)
 103  *    Else if pfid is not present,
 104  *      if spfid is present, mv (spfid,sname) .lustrerepl/[sfid]
 105  *    If moving out of .lustrerepl
 106  *      move out all its children in .lustrerepl.
 107  *      [pfid,tfid,name] tracked from (1) is used for this.
 108  */
 109
 110 #include <stdio.h>
 111 #include <stdlib.h>
 112 #include <string.h>
 113 #include <unistd.h>
 114 #include <getopt.h>
 115 #include <stdarg.h>
 116 #include <fcntl.h>
 117 #include <signal.h>
 118 #include <sys/stat.h>
 119 #include <sys/types.h>
 120 #include <errno.h>
 121 #include <limits.h>
 122 #include <utime.h>
 123 #include <sys/xattr.h>
 124
 125 #include <libcfs/util/string.h>
 126 #include <libcfs/util/parser.h>
 127 #include <lustre/lustreapi.h>
 128 #include <lustre/lustre_idl.h>
 129 #include "lustre_rsync.h"
 130
 131 #define REPLICATE_STATUS_VER 1
 132 #define CLEAR_INTERVAL 100
 133 #define DEFAULT_RSYNC_THRESHOLD 0xA00000 /* 10 MB */
 134
 135 #define TYPE_STR_LEN 16
 136
 137 #define DEFAULT_MDT "-MDT0000"
 138 #define SPECIAL_DIR ".lustrerepl"
 139 #define RSYNC "rsync"
 140 #define TYPE "type"
 141
 142 /* Debug flags */
 143 #define DINFO 1
 144 #define DTRACE 2
 145
 146 /* Not used; declared for fulfilling obd.c's dependency. */
 147 command_t cmdlist[0];
 148 extern int obd_initialize(int argc, char **argv);
 149
 150 /* Information for processing a changelog record. This structure is
 151    allocated on the heap instead of allocating large variables on the
 152    stack. */
 153 struct lr_info {
 154         long long recno;
 155         int target_no;
 156         unsigned int is_extended:1;
 157         enum changelog_rec_type type;
 158         char tfid[LR_FID_STR_LEN];
 159         char pfid[LR_FID_STR_LEN];
 160         char sfid[LR_FID_STR_LEN];
 161         char spfid[LR_FID_STR_LEN];
 162         char sname[NAME_MAX + 1];
 163         char name[NAME_MAX + 1];
 164         char src[PATH_MAX + 1];
 165         char dest[PATH_MAX + 1];
 166         char path[PATH_MAX + 1];
 167         char savedpath[PATH_MAX + 1];
 168         char link[PATH_MAX + 1];
 169         char linktmp[PATH_MAX + 1];
 170         char cmd[PATH_MAX];
 171         int bufsize;
 172         char *buf;
 173
 174         /* Variables for querying the xattributes */
 175         char *xlist;
 176         size_t xsize;
 177         char *xvalue;
 178         size_t xvsize;
 179 };
 180
 181 struct lr_parent_child_list {
 182         struct lr_parent_child_log pc_log;
 183         struct lr_parent_child_list *pc_next;
 184 };
 185
 186 struct lustre_rsync_status *status;
 187 char *statuslog;  /* Name of the status log file */
 188 int logbackedup;
 189 int noxattr;    /* Flag to turn off replicating xattrs */
 190 int noclear;    /* Flag to turn off clearing changelogs */
 191 int debug;      /* Flag to turn debugging information on and off */
 192 int verbose;    /* Verbose output */
 193 long long rec_count; /* No of changelog records that were processed */
 194 int errors;
 195 int dryrun;
 196 int use_rsync;  /* Flag to turn on use of rsync to copy data */
 197 long long rsync_threshold = DEFAULT_RSYNC_THRESHOLD;
 198 int quit;       /* Flag to stop processing the changelog; set on the
 199                    receipt of a signal */
 200 int abort_on_err = 0;
 201
 202 char rsync[PATH_MAX];
 203 char rsync_ver[PATH_MAX];
 204 struct lr_parent_child_list *parents;
 205
 206 FILE *debug_log;
 207
 208 /* Command line options */
 209 struct option long_opts[] = {
 210         {"source",      required_argument, 0, 's'},
 211         {"target",      required_argument, 0, 't'},
 212         {"mdt",         required_argument, 0, 'm'},
 213         {"user",        required_argument, 0, 'u'},
 214         {"statuslog",   required_argument, 0, 'l'},
 215         {"verbose",     no_argument,       0, 'v'},
 216         {"xattr",       required_argument, 0, 'x'},
 217         {"dry-run",     no_argument,       0, 'z'},
 218         /* Undocumented options follow */
 219         {"cl-clear",    required_argument, 0, 'c'},
 220         {"use-rsync",   no_argument,       0, 'r'},
 221         {"rsync-threshold", required_argument, 0, 'y'},
 222         {"start-recno", required_argument, 0, 'n'},
 223         {"abort-on-err",no_argument,       0, 'a'},
 224         {"debug",       required_argument, 0, 'd'},
 225         {"debuglog",    required_argument, 0, 'D'},
 226         {0, 0, 0, 0}
 227 };
 228
 229 /* Command line usage */
 230 void lr_usage()
 231 {
 232         fprintf(stderr, "\tlustre_rsync -s <lustre_root_path> -t <target_path> "
 233                 "-m <mdt> -r <user id> -l <status log>\n"
 234                 "lustre_rsync can also pick up parameters from a "
 235                 "status log created earlier.\n"
 236                 "\tlustre_rsync -l <log_file>\n"
 237                 "options:\n"
 238                 "\t--xattr <yes|no> replicate EAs\n"
 239                 "\t--abort-on-err   abort at first err\n"
 240                 "\t--verbose\n"
 241                 "\t--dry-run        don't write anything\n");
 242 }
 243
 244 #define DEBUG_ENTRY(info)                                                      \
 245         lr_debug(D_TRACE, "***** Start %lld %s (%d) %s %s %s *****\n",         \
 246                  (info)->recno, changelog_type2str((info)->type),              \
 247                  (info)->type, (info)->tfid, (info)->pfid, (info)->name);
 248
 249 #define DEBUG_EXIT(info, rc)                                                   \
 250         lr_debug(D_TRACE, "##### End %lld %s (%d) %s %s %s rc=%d #####\n",     \
 251                  (info)->recno, changelog_type2str((info)->type),              \
 252                  (info)->type, (info)->tfid, (info)->pfid, (info)->name, rc);
 253
 254 /* Print debug information. This is controlled by the value of the
 255    global variable 'debug' */
 256 void lr_debug(int level, const char *fmt, ...)
 257 {
 258         va_list ap;
 259
 260         if (level > debug)
 261                 return;
 262
 263         va_start(ap, fmt);
 264         if (debug_log != NULL)
 265                 vfprintf(debug_log, fmt, ap);
 266         else
 267                 vfprintf(stdout, fmt, ap);
 268         va_end(ap);
 269 }
 270
 271
 272 void * lr_grow_buf(void *buf, int size)
 273 {
 274         void *ptr;
 275
 276         ptr = realloc(buf, size);
 277         if (ptr == NULL)
 278                 free(buf);
 279         return ptr;
 280 }
 281
 282
 283 /* Use rsync to replicate file data */
 284 int lr_rsync_data(struct lr_info *info)
 285 {
 286         int rc;
 287         struct stat st_src, st_dest;
 288         char cmd[PATH_MAX];
 289
 290         lr_debug(DTRACE, "Syncing data%s\n", info->tfid);
 291
 292         rc = stat(info->src, &st_src);
 293         if (rc == -1) {
 294                 fprintf(stderr, "Error: Unable to stat src=%s %s\n",
 295                         info->src, info->name);
 296                 if (errno == ENOENT)
 297                         return 0;
 298                 else
 299                         return -errno;
 300         }
 301         rc = stat(info->dest, &st_dest);
 302         if (rc == -1) {
 303                 fprintf(stderr, "Error: Unable to stat dest=%s\n",
 304                         info->dest);
 305                 return -errno;
 306         }
 307
 308         if (st_src.st_mtime != st_dest.st_mtime ||
 309             st_src.st_size != st_dest.st_size) {
 310                 /* XXX spawning off an rsync for every data sync and
 311                  * waiting synchronously is bad for performance.
 312                  * librsync could possibly used here. But it does not
 313                  * seem to be of production grade. Multi-threaded
 314                  * replication is also to be considered.
 315                  */
 316                 int status;
 317                 snprintf(cmd, PATH_MAX, "%s --inplace %s %s", rsync, info->src,
 318                         info->dest);
 319                 lr_debug(DTRACE, "\t%s %s\n", cmd, info->tfid);
 320                 status = system(cmd);
 321                 if (status == -1) {
 322                         rc = -errno;
 323                 } else if (WIFEXITED(status)) {
 324                         status = WEXITSTATUS(status);
 325                         if (!status)
 326                                 rc = 0;
 327                         else if (status == 23 || status == 24)
 328                                 /* Error due to vanished source files;
 329                                    Ignore this error*/
 330                                 rc = 0;
 331                         else
 332                                 rc = -EINVAL;
 333                         if (status)
 334                                 lr_debug(DINFO, "rsync %s exited with %d %d\n",
 335                                          info->src, status, rc);
 336                 } else {
 337                         rc = -EINTR;
 338                 }
 339         } else {
 340                 lr_debug(DTRACE, "Not syncing %s and %s %s\n", info->src,
 341                          info->dest, info->tfid);
 342         }
 343
 344         return rc;
 345 }
 346
 347 int lr_copy_data(struct lr_info *info)
 348 {
 349         int fd_src = -1;
 350         int fd_dest = -1;
 351         int bufsize;
 352         int rsize;
 353         int rc = 0;
 354         struct stat st_src;
 355         struct stat st_dest;
 356
 357         fd_src = open(info->src, O_RDONLY);
 358         if (fd_src == -1)
 359                 return -errno;
 360         if (fstat(fd_src, &st_src) == -1 ||
 361             stat(info->dest, &st_dest) == -1)
 362                 goto out;
 363
 364         if (st_src.st_mtime == st_dest.st_mtime &&
 365             st_src.st_size == st_dest.st_size)
 366                 goto out;
 367
 368         if (st_src.st_size > rsync_threshold && rsync[0] != '\0') {
 369                 /* It is more efficient to use rsync to replicate
 370                    large files. Any file larger than rsync_threshold
 371                    is handed off to rsync. */
 372                 lr_debug(DTRACE, "Using rsync to replicate %s\n", info->tfid);
 373                 rc = lr_rsync_data(info);
 374                 goto out;
 375         }
 376
 377         fd_dest = open(info->dest, O_WRONLY | O_TRUNC, st_src.st_mode);
 378         if (fd_dest == -1) {
 379                 rc = -errno;
 380                 goto out;
 381         }
 382         bufsize = st_dest.st_blksize;
 383
 384         if (info->bufsize < bufsize) {
 385                 /* Grow buffer */
 386                 info->buf = lr_grow_buf(info->buf, bufsize);
 387                 if (info->buf == NULL) {
 388                         rc = -ENOMEM;
 389                         goto out;
 390                 }
 391                 info->bufsize = bufsize;
 392         }
 393
 394         while (1) {
 395                 rsize = read(fd_src, info->buf, bufsize);
 396                 if (rsize == 0) {
 397                         break;
 398                 } else if (rsize < 0) {
 399                         rc = -errno;
 400                         goto out;
 401                 }
 402                 errno = 0;
 403                 if (write(fd_dest, info->buf, rsize) != rsize) {
 404                         if (errno != 0)
 405                                 rc = -errno;
 406                         else
 407                                 rc = -EINTR;
 408                 }
 409         }
 410         fsync(fd_dest);
 411
 412 out:
 413         if (fd_src != -1)
 414                 close(fd_src);
 415         if (fd_dest != -1)
 416                 close(fd_dest);
 417
 418         return rc;
 419 }
 420
 421 /* Copy data from source to destination */
 422 int lr_sync_data(struct lr_info *info)
 423 {
 424         if (use_rsync)
 425                 return lr_rsync_data(info);
 426         else
 427                 return lr_copy_data(info);
 428 }
 429
 430 /* Copy all attributes from file src to file dest */
 431 int lr_copy_attr(char *src, char *dest)
 432 {
 433         struct stat st;
 434         struct utimbuf time;
 435
 436         if (stat(src, &st) == -1 ||
 437             chmod(dest, st.st_mode) == -1 ||
 438             chown(dest, st.st_uid, st.st_gid) == -1)
 439                 return -errno;
 440
 441         time.actime = st.st_atime;
 442         time.modtime = st.st_mtime;
 443         if (utime(dest, &time) == -1)
 444                 return -errno;
 445         return 0;
 446 }
 447
 448 /* Copy all xattrs from file info->src to info->dest */
 449 int lr_copy_xattr(struct lr_info *info)
 450 {
 451         size_t size = info->xsize;
 452         int start;
 453         int len;
 454         int rc;
 455
 456         if (noxattr)
 457                 return 0;
 458
 459         errno = 0;
 460         rc = llistxattr(info->src, info->xlist, size);
 461         lr_debug(DTRACE, "llistxattr(%s,%p) returned %d, errno=%d\n",
 462                  info->src, info->xlist, rc, errno);
 463         if ((rc > 0 && info->xlist == NULL) || errno == ERANGE) {
 464                 size = rc > PATH_MAX ? rc : PATH_MAX;
 465                 info->xlist = lr_grow_buf(info->xlist, size);
 466                 if (info->xlist == NULL)
 467                         return -ENOMEM;
 468                 info->xsize = size;
 469                 rc = llistxattr(info->src, info->xlist, size);
 470                 lr_debug(DTRACE, "llistxattr %s returned %d, errno=%d\n",
 471                          info->src, rc, errno);
 472         }
 473         if (rc < 0)
 474                 return rc;
 475
 476         len = rc;
 477         start = 0;
 478         while (start < len) {
 479                 size = info->xvsize;
 480                 rc = lgetxattr(info->src, info->xlist + start,
 481                                info->xvalue, size);
 482                 if (info->xvalue == NULL || errno == ERANGE) {
 483                         size = rc > PATH_MAX ? rc : PATH_MAX;
 484                         info->xvalue = lr_grow_buf(info->xvalue, size);
 485                         if (info->xvalue == NULL)
 486                                 return -ENOMEM;
 487                         info->xvsize = size;
 488                         rc = lgetxattr(info->src, info->xlist + start,
 489                                        info->xvalue, size);
 490                 }
 491                 lr_debug(DTRACE, "\t(%s,%d) rc=%p\n", info->xlist + start,
 492                          info->xvalue, rc);
 493                 if (rc > 0) {
 494                         size = rc;
 495                         rc = lsetxattr(info->dest, info->xlist + start,
 496                                        info->xvalue, size, 0);
 497                         lr_debug(DTRACE, "\tlsetxattr(), rc=%d, errno=%d\n",
 498                                  rc, errno);
 499                         if (rc == -1) {
 500                                 if (errno != ENOTSUP) {
 501                                         fprintf(stderr, "Error replicating "
 502                                                 " xattr for %s: %d\n",
 503                                                 info->dest, errno);
 504                                         errors++;
 505                                 }
 506                                 rc = 0;
 507                         }
 508                 }
 509                 start += strlen(info->xlist + start) + 1;
 510         }
 511
 512         lr_debug(DINFO, "setxattr: %s %s\n", info->src, info->dest);
 513
 514         return rc;
 515 }
 516
 517 /* Retrieve the filesystem path for a given FID and a given
 518    linkno. The path is returned in info->path */
 519 int lr_get_path_ln(struct lr_info *info, char *fidstr, int linkno)
 520 {
 521         long long recno = -1;
 522         int rc;
 523
 524         rc = llapi_fid2path(status->ls_source, fidstr, info->path,
 525                             PATH_MAX, &recno, &linkno);
 526         if (rc < 0 && rc != -ENOENT) {
 527                 fprintf(stderr, "fid2path error: (%s, %s) %d %s\n",
 528                         status->ls_source, fidstr, -rc, strerror(errno = -rc));
 529         }
 530
 531         return rc;
 532 }
 533
 534 /* Retrieve the filesystem path for a given FID. The path is returned
 535    in info->path */
 536 int lr_get_path(struct lr_info *info, char *fidstr)
 537 {
 538         return lr_get_path_ln(info, fidstr, 0);
 539 }
 540
 541 /* Generate the path for opening by FID */
 542 void lr_get_FID_PATH(char *mntpt, char *fidstr, char *buf, int bufsize)
 543 {
 544         /* Open-by-FID path is <mntpt>/.lustre/fid/[SEQ:OID:VER] */
 545         snprintf(buf, bufsize, "%s/%s/fid/%s", mntpt, dot_lustre_name,
 546                  fidstr);
 547         return;
 548 }
 549
 550 /* Read the symlink information into 'info->link' */
 551 int lr_get_symlink(struct lr_info *info)
 552 {
 553         int rc;
 554         char *link;
 555
 556         lr_get_FID_PATH(status->ls_source, info->tfid, info->src, PATH_MAX);
 557         rc = readlink(info->src, info->linktmp, PATH_MAX);
 558         if (rc > 0)
 559                 info->linktmp[rc] = '\0';
 560         else
 561                 return rc;
 562         lr_debug(DTRACE, "symlink: readlink returned %s\n", info->linktmp);
 563
 564         if (strncmp(info->linktmp, status->ls_source,
 565                     strlen(status->ls_source)) == 0) {
 566                 /* Strip source fs path and replace with target fs path. */
 567                 link = info->linktmp + strlen(status->ls_source);
 568                 snprintf(info->src, PATH_MAX, "%s%s",
 569                          status->ls_targets[info->target_no], link);
 570                 link = info->src;
 571         } else {
 572                 link = info->linktmp;
 573         }
 574         strlcpy(info->link, link, sizeof(info->link));
 575
 576         return rc;
 577 }
 578
 579 /* Create file/directory/device file/symlink. */
 580 int lr_mkfile(struct lr_info *info)
 581 {
 582         struct stat st;
 583         int rc = 0;
 584
 585         errno = 0;
 586         lr_debug(DINFO, "mkfile(%d) %s \n", info->type, info->dest);
 587         if (info->type == CL_MKDIR) {
 588                 rc = mkdir(info->dest, 0777);
 589         } else if (info->type == CL_SOFTLINK) {
 590                 lr_get_symlink(info);
 591                 rc = symlink(info->link, info->dest);
 592         } else if (info->type == CL_MKNOD) {
 593                 lr_get_FID_PATH(status->ls_source, info->tfid,
 594                                     info->src, PATH_MAX);
 595                 rc = stat(info->src, &st);
 596                 if (rc == -1) {
 597                         if (errno == ENOENT)
 598                                 return 0;
 599                         else
 600                                 return -errno;
 601                 }
 602                 rc = mknod(info->dest, st.st_mode, st.st_rdev);
 603         } else {
 604                 rc = mknod(info->dest, S_IFREG | 0777, 0);
 605         }
 606         if (rc)
 607                 return -errno;
 608
 609         /* Sync data and attributes */
 610         if (info->type == CL_CREATE || info->type == CL_MKDIR) {
 611                 lr_debug(DTRACE, "Syncing data and attributes %s\n",
 612                          info->tfid);
 613                 (void) lr_copy_xattr(info);
 614                 if (info->type == CL_CREATE)
 615                         rc = lr_sync_data(info);
 616                 if (!rc)
 617                         rc = lr_copy_attr(info->src, info->dest);
 618
 619                 if (rc == -ENOENT)
 620                         /* Source file has disappeared. Not an error. */
 621                         rc = 0;
 622         } else {
 623                 lr_debug(DTRACE, "Not syncing data and attributes %s\n",
 624                          info->tfid);
 625         }
 626
 627         return rc;
 628 }
 629
 630 int lr_add_pc(const char *pfid, const char *tfid, const char *name)
 631 {
 632         struct lr_parent_child_list *p;
 633         size_t len;
 634
 635         p = calloc(1, sizeof(*p));
 636         if (p == NULL)
 637                 return -ENOMEM;
 638         len = strlcpy(p->pc_log.pcl_pfid, pfid, sizeof(p->pc_log.pcl_pfid));
 639         if (len >= sizeof(p->pc_log.pcl_pfid))
 640                 goto out_err;
 641         len = strlcpy(p->pc_log.pcl_tfid, tfid, sizeof(p->pc_log.pcl_tfid));
 642         if (len >= sizeof(p->pc_log.pcl_tfid))
 643                 goto out_err;
 644         len = strlcpy(p->pc_log.pcl_name, name, sizeof(p->pc_log.pcl_name));
 645         if (len >= sizeof(p->pc_log.pcl_name))
 646                 goto out_err;
 647
 648         p->pc_next = parents;
 649         parents = p;
 650         return 0;
 651
 652 out_err:
 653         free(p);
 654         return -E2BIG;
 655 }
 656
 657 void lr_cascade_move(const char *fid, const char *dest, struct lr_info *info)
 658 {
 659         struct lr_parent_child_list *curr, *prev;
 660         char *d;
 661         int rc;
 662
 663         d = calloc(1, PATH_MAX + 1);
 664         prev = curr = parents;
 665         while (curr) {
 666                 if (strcmp(curr->pc_log.pcl_pfid, fid) == 0) {
 667                         snprintf(d, PATH_MAX, "%s/%s", dest,
 668                                  curr->pc_log.pcl_name);
 669                         snprintf(info->src, PATH_MAX, "%s/%s/%s",
 670                                 status->ls_targets[info->target_no],
 671                                 SPECIAL_DIR, curr->pc_log.pcl_tfid);
 672                         rc = rename(info->src, d);
 673                         if (rc == -1) {
 674                                 fprintf(stderr, "Error renaming file "
 675                                         " %s to %s: %d\n",
 676                                         info->src, d, errno);
 677                                 errors++;
 678                         }
 679                         if (curr == parents)
 680                                 parents = curr->pc_next;
 681                         else
 682                                 prev->pc_next = curr->pc_next;
 683                         lr_cascade_move(curr->pc_log.pcl_tfid, d, info);
 684                         free(curr);
 685                         prev = curr = parents;
 686
 687                 } else {
 688                         prev = curr;
 689                         curr = curr->pc_next;
 690                 }
 691         }
 692
 693         free(d);
 694 }
 695
 696 /* remove [info->spfid, info->sfid] from parents */
 697 int lr_remove_pc(const char *pfid, const char *tfid)
 698 {
 699         struct lr_parent_child_list *curr, *prev;
 700
 701         for (prev = curr = parents; curr; prev = curr, curr = curr->pc_next) {
 702                 if (strcmp(curr->pc_log.pcl_pfid, pfid) == 0 &&
 703                     strcmp(curr->pc_log.pcl_tfid, tfid) == 0) {
 704                         if (curr == parents)
 705                                 parents = curr->pc_next;
 706                         else
 707                                 prev->pc_next = curr->pc_next;
 708                         free(curr);
 709                         break;
 710                 }
 711         }
 712         return 0;
 713 }
 714
 715 /* Create file under SPECIAL_DIR with its tfid as its name. */
 716 int lr_mk_special(struct lr_info *info)
 717 {
 718         int rc;
 719
 720         snprintf(info->dest, PATH_MAX, "%s/%s/%s",
 721                 status->ls_targets[info->target_no], SPECIAL_DIR,
 722                 info->tfid);
 723
 724         rc = lr_mkfile(info);
 725         if (rc)
 726                 return rc;
 727
 728         rc = lr_add_pc(info->pfid, info->tfid, info->name);
 729         return rc;
 730 }
 731
 732 /* Remove a file or directory */
 733 int lr_rmfile(struct lr_info *info)
 734 {
 735         int rc;
 736
 737         if (info->type == CL_RMDIR)
 738                 rc = rmdir(info->dest);
 739         else
 740                 rc = unlink(info->dest);
 741         if (rc == -1)
 742                 rc = -errno;
 743         return rc;
 744 }
 745
 746 /* Recursively remove directory and its contents */
 747 int lr_rm_recursive(struct lr_info *info)
 748 {
 749         int rc;
 750
 751         snprintf(info->cmd, PATH_MAX, "rm -rf %s", info->dest);
 752         rc = system(info->cmd);
 753         if (rc == -1)
 754                 rc = -errno;
 755
 756         return rc;
 757 }
 758
 759 /* Remove a file under SPECIAL_DIR with its tfid as its name. */
 760 int lr_rm_special(struct lr_info *info)
 761 {
 762         int rc;
 763
 764         snprintf(info->dest, PATH_MAX, "%s/%s/%s",
 765                  status->ls_targets[info->target_no], SPECIAL_DIR,
 766                  info->tfid);
 767         rc = lr_rmfile(info);
 768
 769         if (rc)
 770                 lr_debug(DINFO, "remove: %s; rc=%d, errno=%d\n",
 771                          info->dest, rc, errno);
 772         return rc;
 773 }
 774
 775 /* Replicate file and directory create events */
 776 int lr_create(struct lr_info *info)
 777 {
 778         int len;
 779         int rc1 = 0;
 780         int rc;
 781         int mkspecial = 0;
 782
 783         /* Is target FID present on the source? */
 784         rc = lr_get_path(info, info->tfid);
 785         if (rc == -ENOENT) {
 786                 /* Source file has disappeared. Not an error. */
 787                 lr_debug(DINFO, "create: tfid %s not found on"
 788                          "source-fs\n", info->tfid);
 789                 return 0;
 790         } else if (rc) {
 791                 return rc;
 792         }
 793         strcpy(info->savedpath, info->path);
 794
 795         /* Is parent FID present on the source */
 796         rc = lr_get_path(info, info->pfid);
 797         if (rc == -ENOENT) {
 798                 lr_debug(DINFO, "create: pfid %s not found on source-fs\n",
 799                          info->tfid);
 800                 mkspecial = 1;
 801         } else if (rc < 0) {
 802                 return rc;
 803         }
 804
 805         /* Is f2p(pfid)+name != f2p(tfid)? If not the file has moved. */
 806         len = strlen(info->path);
 807         if (len == 1 && info->path[0] == '/')
 808                 snprintf(info->dest, PATH_MAX, "%s", info->name);
 809         else if (len - 1 > 0 && info->path[len - 1] == '/')
 810                 snprintf(info->dest, PATH_MAX, "%s%s", info->path, info->name);
 811         else
 812                 snprintf(info->dest, PATH_MAX, "%s/%s", info->path, info->name);
 813
 814         lr_debug(DTRACE, "dest = %s; savedpath = %s\n", info->dest,
 815                  info->savedpath);
 816         if (strncmp(info->dest, info->savedpath, PATH_MAX) != 0) {
 817                 lr_debug(DTRACE, "create: file moved (%s). %s != %s\n",
 818                          info->tfid, info->dest, info->savedpath);
 819                 mkspecial = 1;
 820         }
 821
 822         /* Is f2p(pfid) present on the target? If not, the parent has
 823            moved */
 824         if (!mkspecial) {
 825                 snprintf(info->dest, PATH_MAX, "%s/%s", status->ls_targets[0],
 826                         info->path);
 827                 if (access(info->dest, F_OK) != 0) {
 828                         lr_debug(DTRACE, "create: parent %s not found\n",
 829                                 info->dest);
 830                         mkspecial = 1;
 831                 }
 832         }
 833         for (info->target_no = 0; info->target_no < status->ls_num_targets;
 834              info->target_no++) {
 835                 snprintf(info->dest, PATH_MAX, "%s/%s",
 836                         status->ls_targets[info->target_no], info->savedpath);
 837                 lr_get_FID_PATH(status->ls_source, info->tfid, info->src,
 838                                     PATH_MAX);
 839
 840                 if (!mkspecial)
 841                         rc1 = lr_mkfile(info);
 842                 if (mkspecial || rc1 == -ENOENT) {
 843                         rc1 = lr_mk_special(info);
 844                 }
 845                 if (rc1)
 846                         rc = rc1;
 847         }
 848         return rc;
 849 }
 850
 851 /* Replicate a file remove (rmdir/unlink) operation */
 852 int lr_remove(struct lr_info *info)
 853 {
 854         int rc = 0;
 855         int rc1;
 856
 857         for (info->target_no = 0; info->target_no < status->ls_num_targets;
 858              info->target_no++) {
 859
 860                 rc1 = lr_rm_special(info);
 861                 if (!rc1)
 862                         continue;
 863
 864                 rc1 = lr_get_path(info, info->pfid);
 865                 if (rc1 == -ENOENT) {
 866                         lr_debug(DINFO, "remove: pfid %s not found\n",
 867                                  info->pfid);
 868                         continue;
 869                 }
 870                 if (rc1) {
 871                         rc = rc1;
 872                         continue;
 873                 }
 874                 snprintf(info->dest, PATH_MAX, "%s/%s/%s",
 875                         status->ls_targets[info->target_no], info->path,
 876                         info->name);
 877
 878                 rc1 = lr_rmfile(info);
 879                 lr_debug(DINFO, "remove: %s; rc1=%d, errno=%d\n",
 880                          info->dest, rc1, errno);
 881                 if (rc1 == -ENOTEMPTY)
 882                         rc1 = lr_rm_recursive(info);
 883
 884                 if (rc1) {
 885                         rc = rc1;
 886                         continue;
 887                 }
 888         }
 889         return rc;
 890 }
 891
 892 /* Replicate a rename/move operation. */
 893 int lr_move(struct lr_info *info)
 894 {
 895         int rc = 0;
 896         int rc1;
 897         int rc_dest, rc_src;
 898         int special_src = 0;
 899         int special_dest = 0;
 900         char srcpath[PATH_MAX + 1] = "";
 901
 902         LASSERT(info->is_extended);
 903
 904         rc_src = lr_get_path(info, info->spfid);
 905         if (rc_src < 0 && rc_src != -ENOENT)
 906                 return rc_src;
 907         memcpy(srcpath, info->path, strlen(info->path));
 908
 909         rc_dest = lr_get_path(info, info->pfid);
 910         if (rc_dest < 0 && rc_dest != -ENOENT)
 911                 return rc_dest;
 912
 913         for (info->target_no = 0; info->target_no < status->ls_num_targets;
 914              info->target_no++) {
 915
 916                 if (!rc_dest) {
 917                         snprintf(info->dest, PATH_MAX, "%s/%s",
 918                                 status->ls_targets[info->target_no],
 919                                 info->path);
 920                         if (access(info->dest, F_OK) != 0) {
 921                                 rc_dest = -errno;
 922                         } else {
 923                                 snprintf(info->dest, PATH_MAX, "%s/%s/%s",
 924                                         status->ls_targets[info->target_no],
 925                                         info->path, info->name);
 926                         }
 927                         lr_debug(DINFO, "dest path %s rc_dest=%d\n", info->dest,
 928                                  rc_dest);
 929                 }
 930                 if (rc_dest == -ENOENT) {
 931                         snprintf(info->dest, PATH_MAX, "%s/%s/%s",
 932                                 status->ls_targets[info->target_no],
 933                                 SPECIAL_DIR, info->sfid);
 934                         special_dest = 1;
 935                         lr_debug(DINFO, "special dest %s\n", info->dest);
 936                 }
 937
 938                 if (!rc_src) {
 939                         snprintf(info->src, PATH_MAX, "%s/%s/%s",
 940                                 status->ls_targets[info->target_no],
 941                                 srcpath, info->sname);
 942                         lr_debug(DINFO, "src path %s rc_src=%d\n", info->src,
 943                                  rc_src);
 944                 }
 945                 if (rc_src == -ENOENT || (access(info->src, F_OK) != 0 &&
 946                                           errno == ENOENT)) {
 947                         snprintf(info->src, PATH_MAX, "%s/%s/%s",
 948                                 status->ls_targets[info->target_no],
 949                                 SPECIAL_DIR, info->sfid);
 950                         special_src = 1;
 951                         lr_debug(DINFO, "special src %s\n", info->src);
 952                 }
 953
 954                 rc1 = 0;
 955                 errno = 0;
 956                 if (strcmp(info->src, info->dest) != 0) {
 957                         rc1 = rename(info->src, info->dest);
 958                         if (rc1 == -1)
 959                                 rc1 = -errno;
 960                         lr_debug(DINFO, "rename returns %d\n", rc1);
 961                 }
 962
 963                 if (special_src)
 964                         rc1 = lr_remove_pc(info->spfid, info->sfid);
 965
 966                 if (!special_dest)
 967                         lr_cascade_move(info->sfid, info->dest, info);
 968                 else
 969                         rc1 = lr_add_pc(info->pfid, info->sfid, info->name);
 970
 971                 lr_debug(DINFO, "move: %s [to] %s rc1=%d, errno=%d\n",
 972                          info->src, info->dest, rc1, errno);
 973                 if (rc1)
 974                         rc = rc1;
 975         }
 976         return rc;
 977 }
 978
 979 /* Replicate a hard link */
 980 int lr_link(struct lr_info *info)
 981 {
 982         int i;
 983         int len;
 984         int rc;
 985         int rc1;
 986         struct stat st;
 987
 988         lr_get_FID_PATH(status->ls_source, info->tfid, info->src, PATH_MAX);
 989         rc = stat(info->src, &st);
 990         if (rc == -1)
 991                 return -errno;
 992
 993         for (info->target_no = 0; info->target_no < status->ls_num_targets;
 994              info->target_no++) {
 995
 996                 info->src[0] = 0;
 997                 info->dest[0] = 0;
 998                 rc1 = 0;
 999
1000                 /* Search through the hardlinks to get the src and dest */
1001                 for (i = 0; i < st.st_nlink && (info->src[0] == 0 ||
1002                                                 info->dest[0] == 0); i++) {
1003                         rc1 = lr_get_path_ln(info, info->tfid, i);
1004                         lr_debug(rc1 ? 0:DTRACE, "\tfid2path %s, %s, %d rc=%d\n",
1005                                  info->path, info->name, i, rc1);
1006                         if (rc1)
1007                                 break;
1008
1009                         len = strlen(info->path) - strlen(info->name);
1010                         if (len >= 0 && strcmp(info->path + len,
1011                                               info->name) == 0)
1012                                 snprintf(info->dest, PATH_MAX, "%s/%s",
1013                                         status->ls_targets[info->target_no],
1014                                         info->path);
1015                         else if (info->src[0] == 0)
1016                                 snprintf(info->src, PATH_MAX, "%s/%s",
1017                                         status->ls_targets[info->target_no],
1018                                         info->path);
1019                 }
1020
1021                 if (rc1) {
1022                         rc = rc1;
1023                         continue;
1024                 }
1025
1026                 if (info->src[0] == 0 || info->dest[0] == 0)
1027                         /* Could not find the source or destination.
1028                          This can happen when some links don't exist
1029                          anymore. */
1030                         return -EINVAL;
1031
1032                 if (info->src[0] == 0)
1033                         snprintf(info->src, PATH_MAX, "%s/%s/%s",
1034                                 status->ls_targets[info->target_no],
1035                                 SPECIAL_DIR, info->tfid);
1036                 else if (info->dest[0] == 0)
1037                         snprintf(info->dest, PATH_MAX, "%s/%s/%s",
1038                                 status->ls_targets[info->target_no],
1039                                 SPECIAL_DIR, info->tfid);
1040
1041                 rc1 = link(info->src, info->dest);
1042                 lr_debug(rc1?0:DINFO, "link: %s [to] %s; rc1=%d %s\n",
1043                          info->src, info->dest, rc1, strerror(errno));
1044
1045                 if (rc1)
1046                         rc = rc1;
1047         }
1048         return rc;
1049 }
1050
1051 /* Replicate file attributes */
1052 int lr_setattr(struct lr_info *info)
1053 {
1054         int rc1;
1055         int rc;
1056
1057         lr_get_FID_PATH(status->ls_source, info->tfid, info->src, PATH_MAX);
1058
1059         rc = lr_get_path(info, info->tfid);
1060         if (rc == -ENOENT)
1061                 lr_debug(DINFO, "setattr: %s not present on source-fs\n",
1062                          info->src);
1063         if (rc)
1064                 return rc;
1065
1066         for (info->target_no = 0; info->target_no < status->ls_num_targets;
1067              info->target_no++) {
1068
1069                 snprintf(info->dest, PATH_MAX, "%s/%s",
1070                          status->ls_targets[info->target_no], info->path);
1071                 lr_debug(DINFO, "setattr: %s %s %s", info->src, info->dest,
1072                          info->tfid);
1073
1074                 rc1 = lr_sync_data(info);
1075                 if (!rc1)
1076                         rc1 = lr_copy_attr(info->src, info->dest);
1077                 if (rc1)
1078                         rc = rc1;
1079         }
1080         return rc;
1081 }
1082
1083 /* Replicate xattrs */
1084 int lr_setxattr(struct lr_info *info)
1085 {
1086         int rc, rc1;
1087
1088         lr_get_FID_PATH(status->ls_source, info->tfid, info->src, PATH_MAX);
1089
1090         rc = lr_get_path(info, info->tfid);
1091         if (rc == -ENOENT)
1092                 lr_debug(DINFO, "setxattr: %s not present on source-fs\n",
1093                          info->src);
1094         if (rc)
1095                 return rc;
1096
1097         for (info->target_no = 0; info->target_no < status->ls_num_targets;
1098              info->target_no++) {
1099
1100                 snprintf(info->dest, PATH_MAX, "%s/%s",
1101                         status->ls_targets[info->target_no], info->path);
1102                 lr_debug(DINFO, "setxattr: %s %s %s\n", info->src, info->dest,
1103                          info->tfid);
1104
1105                 rc1 = lr_copy_xattr(info);
1106                 if (rc1)
1107                         rc = rc1;
1108         }
1109
1110         return rc;
1111 }
1112
1113 /* Parse a line of changelog entry */
1114 int lr_parse_line(void *priv, struct lr_info *info)
1115 {
1116         struct changelog_rec            *rec;
1117         struct changelog_ext_rename     *rnm;
1118         size_t                           namelen;
1119         size_t                           copylen = sizeof(info->name);
1120
1121         if (llapi_changelog_recv(priv, &rec) != 0)
1122                 return -1;
1123
1124         info->is_extended = !!(rec->cr_flags & CLF_RENAME);
1125         info->recno = rec->cr_index;
1126         info->type = rec->cr_type;
1127         snprintf(info->tfid, sizeof(info->tfid), DFID, PFID(&rec->cr_tfid));
1128         snprintf(info->pfid, sizeof(info->pfid), DFID, PFID(&rec->cr_pfid));
1129
1130         namelen = strnlen(changelog_rec_name(rec), rec->cr_namelen);
1131         if (copylen > namelen + 1)
1132                 copylen = namelen + 1;
1133         strlcpy(info->name, changelog_rec_name(rec), copylen);
1134
1135         /* Don't use rnm if CLF_RENAME isn't set */
1136         rnm = changelog_rec_rename(rec);
1137         if (rec->cr_flags & CLF_RENAME && !fid_is_zero(&rnm->cr_sfid)) {
1138                 copylen = sizeof(info->sname);
1139
1140                 snprintf(info->sfid, sizeof(info->sfid), DFID,
1141                          PFID(&rnm->cr_sfid));
1142                 snprintf(info->spfid, sizeof(info->spfid), DFID,
1143                          PFID(&rnm->cr_spfid));
1144                 namelen = changelog_rec_snamelen(rec);
1145                 if (copylen > namelen + 1)
1146                         copylen = namelen + 1;
1147                 strlcpy(info->sname, changelog_rec_sname(rec), copylen);
1148
1149                 if (verbose > 1)
1150                         printf("Rec %lld: %d %s %s\n", info->recno, info->type,
1151                                 info->name, info->sname);
1152         } else {
1153                 if (verbose > 1)
1154                         printf("Rec %lld: %d %s\n", info->recno, info->type,
1155                                 info->name);
1156         }
1157
1158         llapi_changelog_free(&rec);
1159
1160         rec_count++;
1161         return 0;
1162 }
1163
1164 /* Initialize the replication parameters */
1165 int lr_init_status()
1166 {
1167         size_t size = sizeof(struct lustre_rsync_status) + PATH_MAX + 1;
1168
1169         if (status != NULL)
1170                 return 0;
1171         status = calloc(size, 1);
1172         if (status == NULL)
1173                 return -ENOMEM;
1174         status->ls_version = REPLICATE_STATUS_VER;
1175         status->ls_size = size;
1176         status->ls_last_recno = -1;
1177         return 0;
1178 }
1179
1180 /* Make a backup of the statuslog */
1181 void lr_backup_log()
1182 {
1183         char backupfile[PATH_MAX];
1184
1185         if (logbackedup)
1186                 return;
1187         snprintf(backupfile, PATH_MAX, "%s.old", statuslog);
1188         (void) rename(statuslog, backupfile);
1189         logbackedup = 1;
1190
1191         return;
1192 }
1193
1194 /* Save replication parameters to a statuslog. */
1195 int lr_write_log()
1196 {
1197         int fd;
1198         size_t size;
1199         size_t write_size = status->ls_size;
1200         struct lr_parent_child_list *curr;
1201         int rc = 0;
1202
1203         if (statuslog == NULL)
1204                 return 0;
1205
1206         lr_backup_log();
1207
1208         fd = open(statuslog, O_WRONLY | O_CREAT | O_SYNC,
1209                              S_IRUSR | S_IWUSR);
1210         if (fd == -1) {
1211                 fprintf(stderr, "Error opening log file for writing (%s)\n",
1212                         statuslog);
1213                 return -1;
1214         }
1215         errno = 0;
1216         size = write(fd, status, write_size);
1217         if (size != write_size) {
1218                 fprintf(stderr, "Error writing to log file (%s) %d\n",
1219                         statuslog, errno);
1220                 close(fd);
1221                 return -1;
1222         }
1223
1224         for (curr = parents; curr; curr = curr->pc_next) {
1225                 size = write(fd, &curr->pc_log, sizeof(curr->pc_log));
1226                 if (size != sizeof(curr->pc_log)) {
1227                         fprintf(stderr, "Error writing to log file (%s) %d\n",
1228                                 statuslog, errno);
1229                         rc = -1;
1230                         break;
1231                 }
1232         }
1233         close(fd);
1234         return rc;
1235 }
1236
1237 /* Read statuslog and populate the replication parameters.  Command
1238  * line parameters take precedence over parameters in the log file.*/
1239 int lr_read_log()
1240 {
1241         struct lr_parent_child_list *tmp;
1242         struct lr_parent_child_log rec;
1243         struct lustre_rsync_status *s;
1244         int fd = -1;
1245         size_t size;
1246         size_t read_size = sizeof(struct lustre_rsync_status) + PATH_MAX + 1;
1247         int rc = 0;
1248
1249         if (statuslog == NULL)
1250                 return 0;
1251
1252         s = calloc(1, read_size);
1253         if (s == NULL) {
1254                 rc = -ENOMEM;
1255                 goto out;
1256         }
1257
1258         fd = open(statuslog, O_RDONLY);
1259         if (fd == -1) {
1260                 rc = -errno;
1261                 goto out;
1262         }
1263
1264         size = read(fd, s, read_size);
1265         if (size != read_size) {
1266                 rc = -EINVAL;
1267                 goto out;
1268         }
1269
1270         if (read_size < s->ls_size) {
1271                 read_size = s->ls_size;
1272                 s = lr_grow_buf(s, read_size);
1273                 if (s == NULL) {
1274                         rc = -ENOMEM;
1275                         goto out;
1276                 }
1277
1278                 if (lseek(fd, 0, SEEK_SET) == -1) {
1279                         rc = -ENOMEM;
1280                         goto out;
1281                 }
1282
1283                 size = read(fd, s, read_size);
1284                 if (size != read_size) {
1285                         rc = -EINVAL;
1286                         goto out;
1287                 }
1288         }
1289
1290         while (read(fd, &rec, sizeof(rec)) != 0) {
1291                 tmp = calloc(1, sizeof(*tmp));
1292                 if (!tmp) {
1293                         rc = -ENOMEM;
1294                         goto out;
1295                 }
1296
1297                 tmp->pc_log = rec;
1298                 tmp->pc_next = parents;
1299                 parents = tmp;
1300         }
1301
1302         /* copy uninitialized fields to status */
1303         if (status->ls_num_targets == 0) {
1304                 if (status->ls_size != s->ls_size) {
1305                         status = lr_grow_buf(status, s->ls_size);
1306                         if (status == NULL) {
1307                                 rc = -ENOMEM;
1308                                 goto out;
1309                         }
1310
1311                         status->ls_size = s->ls_size;
1312                 }
1313                 status->ls_num_targets = s->ls_num_targets;
1314                 memcpy(status->ls_targets, s->ls_targets,
1315                        (PATH_MAX + 1) * s->ls_num_targets);
1316         }
1317         if (status->ls_last_recno == -1)
1318                 status->ls_last_recno = s->ls_last_recno;
1319
1320         if (status->ls_registration[0] == '\0')
1321                 strlcpy(status->ls_registration, s->ls_registration,
1322                         sizeof(status->ls_registration));
1323
1324         if (status->ls_mdt_device[0] == '\0')
1325                 strlcpy(status->ls_mdt_device, s->ls_mdt_device,
1326                         sizeof(status->ls_mdt_device));
1327
1328         if (status->ls_source_fs[0] == '\0')
1329                 strlcpy(status->ls_source_fs, s->ls_source_fs,
1330                         sizeof(status->ls_source_fs));
1331
1332         if (status->ls_source[0] == '\0')
1333                 strlcpy(status->ls_source, s->ls_source,
1334                         sizeof(status->ls_source));
1335
1336  out:
1337         if (fd != -1)
1338                 close(fd);
1339         if (s)
1340                 free(s);
1341         return rc;
1342 }
1343
1344 /* Clear changelogs every CLEAR_INTERVAL records or at the end of
1345    processing. */
1346 int lr_clear_cl(struct lr_info *info, int force)
1347 {
1348         char            mdt_device[LR_NAME_MAXLEN + 1];
1349         long long       rec;
1350         int             rc = 0;
1351
1352         if (force || info->recno > status->ls_last_recno + CLEAR_INTERVAL) {
1353                 if (info->type == CL_RENAME)
1354                         rec = info->recno + 1;
1355                 else
1356                         rec = info->recno;
1357                 if (!noclear && !dryrun) {
1358                         /* llapi_changelog_clear modifies the mdt
1359                          * device name so make a copy of it until this
1360                          * is fixed.
1361                         */
1362                         strlcpy(mdt_device, status->ls_mdt_device,
1363                                 sizeof(mdt_device));
1364                         rc = llapi_changelog_clear(mdt_device,
1365                                                    status->ls_registration,
1366                                                    rec);
1367                         if (rc)
1368                                 printf("Changelog clear (%s, %s, %lld) "
1369                                        "returned %d\n", status->ls_mdt_device,
1370                                        status->ls_registration, rec, rc);
1371                 }
1372                 if (!rc && !dryrun) {
1373                         status->ls_last_recno = rec;
1374                         lr_write_log();
1375
1376                 }
1377         }
1378
1379         return rc;
1380 }
1381
1382 /* Locate a usable version of rsync. At this point we'll use any
1383    version. */
1384 int lr_locate_rsync()
1385 {
1386         FILE *fp;
1387         int len;
1388
1389         /* Locate rsync */
1390         snprintf(rsync, PATH_MAX, "%s -p %s", TYPE, RSYNC);
1391         fp = popen(rsync, "r");
1392         if (fp == NULL)
1393                 return -1;
1394
1395         if (fgets(rsync, PATH_MAX, fp) == NULL) {
1396                 fclose(fp);
1397                 return -1;
1398         }
1399
1400         len = strlen(rsync);
1401         if (len > 0 && rsync[len - 1] == '\n')
1402                 rsync[len - 1] = '\0';
1403         fclose(fp);
1404
1405         /* Determine the version of rsync */
1406         snprintf(rsync_ver, PATH_MAX, "%s --version", rsync);
1407         fp = popen(rsync_ver, "r");
1408         if (fp == NULL)
1409                 return -1;
1410
1411         if (fgets(rsync_ver, PATH_MAX, fp) == NULL) {
1412                 fclose(fp);
1413                 return -1;
1414         }
1415         len = strlen(rsync_ver);
1416         if (len > 0 && rsync_ver[len - 1] == '\n')
1417                 rsync_ver[len - 1] = '\0';
1418         fclose(fp);
1419
1420         return 0;
1421
1422 }
1423
1424 /* Print the replication parameters */
1425 void lr_print_status(struct lr_info *info)
1426 {
1427         int i;
1428
1429         if (!verbose)
1430                 return;
1431
1432         printf("Lustre filesystem: %s\n", status->ls_source_fs);
1433         printf("MDT device: %s\n", status->ls_mdt_device);
1434         printf("Source: %s\n", status->ls_source);
1435         for (i = 0; i < status->ls_num_targets; i++)
1436                 printf("Target: %s\n", status->ls_targets[i]);
1437         if (statuslog != NULL)
1438                 printf("Statuslog: %s\n", statuslog);
1439         printf("Changelog registration: %s\n", status->ls_registration);
1440         printf("Starting changelog record: "LPD64"\n", status->ls_last_recno);
1441         if (noxattr)
1442                 printf("Replicate xattrs: no\n");
1443         if (noclear)
1444                 printf("Clear changelog after use: no\n");
1445         if (use_rsync)
1446                 printf("Using rsync: %s (%s)\n", rsync, rsync_ver);
1447 }
1448
1449 void lr_print_failure(struct lr_info *info, int rc)
1450 {
1451         fprintf(stderr, "Replication of operation failed(%d):"
1452                 " %lld %s (%d) %s %s %s\n", rc, info->recno,
1453                 changelog_type2str(info->type), info->type, info->tfid,
1454                 info->pfid, info->name);
1455 }
1456
1457 /* Replicate filesystem operations from src_path to target_path */
1458 int lr_replicate()
1459 {
1460         void *changelog_priv;
1461         struct lr_info *info;
1462         struct lr_info *ext = NULL;
1463         time_t start;
1464         int xattr_not_supp;
1465         int i;
1466         int rc;
1467
1468         start = time(NULL);
1469
1470         info = calloc(1, sizeof(struct lr_info));
1471         if (info == NULL)
1472                 return -ENOMEM;
1473
1474         rc = llapi_search_fsname(status->ls_source, status->ls_source_fs);
1475         if (rc) {
1476                 fprintf(stderr, "Source path is not a valid Lustre client "
1477                         "mountpoint.\n");
1478                 goto out;
1479         }
1480         if (status->ls_mdt_device[0] == '\0')
1481                 snprintf(status->ls_mdt_device, LR_NAME_MAXLEN, "%s%s",
1482                         status->ls_source_fs, DEFAULT_MDT);
1483
1484         ext = calloc(1, sizeof(struct lr_info));
1485         if (ext == NULL) {
1486                 rc = -ENOMEM;
1487                 goto out;
1488         }
1489
1490         for (i = 0, xattr_not_supp = 0; i < status->ls_num_targets; i++) {
1491                 snprintf(info->dest, PATH_MAX, "%s/%s", status->ls_targets[i],
1492                         SPECIAL_DIR);
1493                 rc = mkdir(info->dest, 0777);
1494                 if (rc == -1 && errno != EEXIST) {
1495                         fprintf(stderr, "Error writing to target path %s.\n",
1496                                 status->ls_targets[i]);
1497                         rc = -errno;
1498                         goto out;
1499                 }
1500                 rc = llistxattr(info->src, info->xlist, info->xsize);
1501                 if (rc == -1 && errno == ENOTSUP) {
1502                         fprintf(stderr, "xattrs not supported on %s\n",
1503                                 status->ls_targets[i]);
1504                         xattr_not_supp++;
1505                 }
1506         }
1507         if (xattr_not_supp == status->ls_num_targets)
1508                 /* None of the targets support xattrs. */
1509                 noxattr = 1;
1510
1511         lr_print_status(info);
1512
1513         /* Open changelogs for consumption*/
1514         rc = llapi_changelog_start(&changelog_priv,
1515                                    CHANGELOG_FLAG_BLOCK | CHANGELOG_FLAG_JOBID,
1516                                    status->ls_source_fs, status->ls_last_recno);
1517         if (rc < 0) {
1518                 fprintf(stderr, "Error opening changelog file for fs %s.\n",
1519                         status->ls_source_fs);
1520                 goto out;
1521         }
1522
1523         while (!quit && lr_parse_line(changelog_priv, info) == 0) {
1524                 rc = 0;
1525                 if (info->type == CL_RENAME && !info->is_extended) {
1526                         /* Newer rename operations extends changelog to store
1527                          * source file information, but old changelog has
1528                          * another record.
1529                          */
1530                         if (lr_parse_line(changelog_priv, ext) != 0)
1531                                 break;
1532                         memcpy(info->sfid, info->tfid, sizeof(info->sfid));
1533                         memcpy(info->spfid, info->pfid, sizeof(info->spfid));
1534                         memcpy(info->tfid, ext->tfid, sizeof(info->tfid));
1535                         memcpy(info->pfid, ext->pfid, sizeof(info->pfid));
1536                         strlcpy(info->sname, info->name, sizeof(info->sname));
1537                         strlcpy(info->name, ext->name, sizeof(info->name));
1538                         info->is_extended = 1;
1539                 }
1540
1541                 if (dryrun)
1542                         continue;
1543
1544                 DEBUG_ENTRY(info);
1545
1546                 switch(info->type) {
1547                 case CL_CREATE:
1548                 case CL_MKDIR:
1549                 case CL_MKNOD:
1550                 case CL_SOFTLINK:
1551                         rc = lr_create(info);
1552                         break;
1553                 case CL_RMDIR:
1554                 case CL_UNLINK:
1555                         rc = lr_remove(info);
1556                         break;
1557                 case CL_RENAME:
1558                         rc = lr_move(info);
1559                         break;
1560                 case CL_HARDLINK:
1561                         rc = lr_link(info);
1562                         break;
1563                 case CL_TRUNC:
1564                 case CL_SETATTR:
1565                         rc = lr_setattr(info);
1566                         break;
1567                 case CL_XATTR:
1568                         rc = lr_setxattr(info);
1569                         break;
1570                 case CL_CLOSE:
1571                 case CL_EXT:
1572                 case CL_OPEN:
1573                 case CL_LAYOUT:
1574                 case CL_MARK:
1575                         /* Nothing needs to be done for these entries */
1576                         /* fallthrough */
1577                 default:
1578                         break;
1579                 }
1580                 DEBUG_EXIT(info, rc);
1581                 if (rc && rc != -ENOENT) {
1582                         lr_print_failure(info, rc);
1583                         errors++;
1584                         if (abort_on_err)
1585                                 break;
1586                 }
1587                 lr_clear_cl(info, 0);
1588                 if (debug) {
1589                         bzero(info, sizeof(struct lr_info));
1590                         bzero(ext, sizeof(struct lr_info));
1591                 }
1592         }
1593
1594         llapi_changelog_fini(&changelog_priv);
1595
1596         if (errors || verbose)
1597                 printf("Errors: %d\n", errors);
1598
1599         /* Clear changelog records used so far */
1600         lr_clear_cl(info, 1);
1601
1602         if (verbose) {
1603                 printf("lustre_rsync took %ld seconds\n", time(NULL) - start);
1604                 printf("Changelog records consumed: %lld\n", rec_count);
1605         }
1606
1607         rc = 0;
1608
1609 out:
1610         if (info != NULL)
1611                 free(info);
1612         if (ext != NULL)
1613                 free(ext);
1614
1615         return rc;
1616 }
1617
1618 void
1619 termination_handler (int signum)
1620 {
1621         /* Set a flag for the replicator to gracefully shutdown */
1622         quit = 1;
1623         printf("lustre_rsync halting.\n");
1624 }
1625
1626 int main(int argc, char *argv[])
1627 {
1628         int newsize;
1629         int numtargets = 0;
1630         int rc = 0;
1631
1632         if ((rc = lr_init_status()) != 0)
1633                 return rc;
1634
1635         while ((rc = getopt_long(argc, argv, "as:t:m:u:l:vx:zc:ry:n:d:D:",
1636                                  long_opts, NULL)) >= 0) {
1637                 switch (rc) {
1638                 case 'a':
1639                         /* Assume absolute paths */
1640                         abort_on_err++;
1641                         break;
1642                 case 's':
1643                         /* Assume absolute paths */
1644                         strlcpy(status->ls_source, optarg,
1645                                 sizeof(status->ls_source));
1646                         break;
1647                 case 't':
1648                         status->ls_num_targets++;
1649                         numtargets++;
1650                         if (numtargets != status->ls_num_targets) {
1651                                 /* Targets were read from a log
1652                                    file. The ones specified on the
1653                                    command line take precedence. The
1654                                    ones from the log file will be
1655                                    ignored. */
1656                                 status->ls_num_targets = numtargets;
1657                         }
1658                         newsize = sizeof (struct lustre_rsync_status) +
1659                                 (status->ls_num_targets * (PATH_MAX + 1));
1660                         if (status->ls_size != newsize) {
1661                                 status->ls_size = newsize;
1662                                 status = lr_grow_buf(status, newsize);
1663                                 if (status == NULL)
1664                                         return -ENOMEM;
1665                         }
1666                         strlcpy(status->ls_targets[status->ls_num_targets - 1],
1667                                 optarg, sizeof(status->ls_targets[0]));
1668                         break;
1669                 case 'm':
1670                         strlcpy(status->ls_mdt_device, optarg,
1671                                 sizeof(status->ls_mdt_device));
1672                         break;
1673                 case 'u':
1674                         strlcpy(status->ls_registration, optarg,
1675                                 sizeof(status->ls_registration));
1676                         break;
1677                 case 'l':
1678                         statuslog = optarg;
1679                         (void) lr_read_log();
1680                         break;
1681                 case 'v':
1682                         verbose++;
1683                         break;
1684                 case 'x':
1685                         if (strcmp("no", optarg) == 0) {
1686                                 noxattr = 1;
1687                         } else if (strcmp("yes", optarg) != 0) {
1688                                 printf("Invalid parameter %s. "
1689                                        "Specify --xattr=no or --xattr=yes\n",
1690                                        optarg);
1691                                 return -1;
1692                         }
1693                         break;
1694                 case 'z':
1695                         dryrun = 1;
1696                         break;
1697                 case 'c':
1698                         /* Undocumented option cl-clear */
1699                         if (strcmp("no", optarg) == 0) {
1700                                 noclear = 1;
1701                         } else if (strcmp("yes", optarg) != 0) {
1702                                 printf("Invalid parameter %s. "
1703                                        "Specify --cl-clear=no "
1704                                        "or --cl-clear=yes\n",
1705                                        optarg);
1706                                 return -1;
1707                         }
1708                         break;
1709                 case 'r':
1710                         /* Undocumented option use-rsync */
1711                         use_rsync = 1;
1712                         break;
1713                 case 'y':
1714                         /* Undocumented option rsync-threshold */
1715                         rsync_threshold = atol(optarg);
1716                         break;
1717                 case 'n':
1718                         /* Undocumented option start-recno */
1719                         status->ls_last_recno = atol(optarg);
1720                         break;
1721                 case 'd':
1722                         /* Undocumented option debug */
1723                         debug = atoi(optarg);
1724                         if (debug < 0 || debug > 2)
1725                                 debug = 0;
1726                         break;
1727                 case 'D':
1728                         /* Undocumented option debug log file */
1729                         if (debug_log != NULL)
1730                                 fclose(debug_log);
1731                         debug_log = fopen(optarg, "a");
1732                         if (debug_log == NULL) {
1733                                 printf("Cannot open %s for debug log\n",
1734                                        optarg);
1735                                 return -1;
1736                         }
1737                         break;
1738                 default:
1739                         fprintf(stderr, "error: %s: option '%s' "
1740                                 "unrecognized.\n", argv[0], argv[optind - 1]);
1741                         lr_usage();
1742                         return -1;
1743                 }
1744         }
1745
1746         if (status->ls_last_recno == -1)
1747                 status->ls_last_recno = 0;
1748         if (strnlen(status->ls_registration, LR_NAME_MAXLEN) == 0) {
1749                 /* No registration ID was passed in. */
1750                 printf("Please specify changelog consumer registration id.\n");
1751                 lr_usage();
1752                 return -1;
1753         }
1754         if (strnlen(status->ls_source, PATH_MAX) == 0) {
1755                 fprintf(stderr, "Please specify the source path.\n");
1756                 lr_usage();
1757                 return -1;
1758         }
1759         if (strnlen(status->ls_targets[0], PATH_MAX) == 0) {
1760                 fprintf(stderr, "Please specify the target path.\n");
1761                 lr_usage();
1762                 return -1;
1763         }
1764
1765         /* This plumbing is needed for some of the ioctls behind
1766            llapi calls to work. */
1767         if (obd_initialize(argc, argv) < 0) {
1768                 fprintf(stderr, "obd_initialize failed.\n");
1769                 exit(-1);
1770         }
1771
1772         rc = lr_locate_rsync();
1773         if (use_rsync && rc != 0) {
1774                 fprintf(stderr, "Error: unable to locate %s.\n", RSYNC);
1775                 exit(-1);
1776         }
1777
1778         signal(SIGINT, termination_handler);
1779         signal(SIGHUP, termination_handler);
1780         signal(SIGTERM, termination_handler);
1781
1782         rc = lr_replicate();
1783
1784         if (debug_log != NULL)
1785                 fclose(debug_log);
1786         return rc;
1787 }