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