continusync

   1 #!/usr/bin/env perl
   2
   3 # Continuous mirroring script around inotifywait and rsync, as suggested by
   4 # Buck Huppmann.  Supports local and remote pushing.
   5 # EXPERIMENTAL!  THERE IS ABSOLUTELY NO WARRANTY!
   6 # -- Matt McCutchen <hashproduct@gmail.com>
   7 # See: http://www.kepreon.com/~matt/utils/#continusync
   8
   9 # Usage:
  10 #     continusync path/to/srcdir/ path/to/destdir/
  11 #     continusync path/to/srcdir/ [user@]host:path/to/destdir/
  12
  13 # It seems to work, but it runs rsync once per event, which is ridiculous.
  14 # TODO: Event batching!!!
  15 # TODO: Do the recursive deletion in perl instead of calling rm(1).
  16
  17 use warnings;
  18 use strict;
  19
  20 # Configuration.  TODO: Add options for these.
  21
  22 # Let the rsyncs we invoke piggyback on the main ssh connection.
  23 # For this to work, you have to set a ControlPath in your ~/.ssh/config ;
  24 # see the ssh_config(5) man page.
  25 our @rshArgs = ('ssh', '-o', 'ControlMaster auto');
  26
  27 our $csPath = 'continusync';
  28
  29 # Don't put -r or --delete here.
  30 # cp2 :)
  31 our @rsyncArgs = ('rsync', '-lE', '--chmod=ugo=rwX', '-i');
  32
  33 use IPC::Open2;
  34 use IO::Handle;
  35
  36 # readFully(fh, length) -> data
  37 sub readFully(*$) {
  38         my ($fh, $bytesLeft) = @_;
  39         my ($buf, $off, $rv) = ('', 0);
  40         while ($bytesLeft > 0) {
  41                 $rv = sysread($fh, $buf, $bytesLeft, $off);
  42                 return undef if $rv == 0; # HMMM: May lose partial read
  43                 die "Read error" unless $rv > 0;
  44                 $bytesLeft -= $rv;
  45                 $off += $rv;
  46         }
  47         return $buf;
  48 }
  49
  50 # writeFully(fh, data)
  51 sub writeFully(*$) {
  52         my ($fh, $buf) = @_;
  53         my ($bytesLeft, $off, $rv) = (length($buf), 0);
  54         while ($bytesLeft > 0) {
  55                 $rv = syswrite($fh, $buf, $bytesLeft, $off);
  56                 die "Write error" unless $rv > 0;
  57                 $bytesLeft -= $rv;
  58                 $off += $rv;
  59         }
  60 }
  61
  62 # readMsg(fh) -> (type, body)
  63 sub readMsg(*) {
  64         my ($fh) = @_;
  65         my $head = readFully($fh, 8);
  66         return (undef, undef) unless defined($head);
  67         my ($type, $bodyLen) = unpack('NN', $head);
  68         return ($type, readFully($fh, $bodyLen));
  69 }
  70
  71 # writeMsg(fh, type, body)
  72 sub writeMsg(*$$) {
  73         my ($fh, $type, $body) = @_;
  74         writeFully($fh, pack('NN/a*', $type, $body));
  75 }
  76
  77 # Message types
  78 #sub MSG_EXIT { 0; }
  79 sub MSG_REMOTE_PATH { 1; }
  80 sub MSG_PERFORMED { 2; }
  81 sub MSG_RENAME { 3; }
  82 sub MSG_DELREC { 4; }
  83 sub MSG_DELETED { 5; }
  84
  85 sub doServer($) {
  86         my ($dest) = @_;
  87
  88         chdir($dest);
  89
  90         my ($type, $body);
  91         while (($type, $body) = readMsg(STDIN), defined($type)) {
  92                 if ($type == MSG_RENAME) {
  93                         my ($src, $dest) = unpack('N/a*N/a*', $body);
  94                         rename($src, $dest);
  95                         writeMsg(STDOUT, MSG_PERFORMED, '');
  96                 } elsif ($type == MSG_DELREC) {
  97                         my $victim = $body;
  98                         my ($rmPid, $fromRm);
  99                         $rmPid = open($fromRm, '-|', 'rm', '-rf', '-v', $victim);
 100                         my $rmLine;
 101                         while (defined($rmLine = <$fromRm>)) {
 102                                 chomp($rmLine);
 103                                 if ($rmLine =~ /^[^`]*`(.*)'[^']*$/) {
 104                                         writeMsg(STDOUT, MSG_DELETED, $1);
 105                                 }
 106                         }
 107                         close($fromRm);
 108                         waitpid($rmPid, 0);
 109                         writeMsg(STDOUT, MSG_PERFORMED, '');
 110                 }
 111         }
 112 }
 113
 114 # The stuff below applies only to the client.
 115
 116 our ($src, $dest);
 117 our $localDestFH;
 118
 119 our ($fromServer, $toServer, $serverPid);
 120 our ($fromInwt, $inwtPid);
 121
 122 sub clientQuit() {
 123         print "Caught a signal.  Shutting down.\n";
 124
 125         #print STDOUT "serverPid is $serverPid\n";
 126         close($fromServer);
 127         close($toServer);
 128         waitpid($serverPid, 0);
 129
 130         #print STDOUT "inwtPid is $inwtPid\n";
 131         kill(2, $inwtPid);
 132         close($fromInwt);
 133         waitpid($inwtPid, 0);
 134
 135         exit(0);
 136 }
 137
 138 sub doRsync($$@) {
 139         my ($isRecursive, $isDelete, @paths) = @_;
 140
 141         my ($rsyncPid, $toRsync);
 142         $rsyncPid = open($toRsync, '|-', @rsyncArgs,
 143                 ($isRecursive ? '-r' : '-d'), ($isDelete ? '--del' : ()),
 144                 '--no-implied-dirs', '-t', '--from0', '--files-from=-', '.', $dest);
 145         foreach my $p (@paths) {
 146                 print $toRsync $p, "\0";
 147         }
 148         close($toRsync);
 149         waitpid($rsyncPid, 0);
 150 }
 151
 152 sub doRename($$) {
 153         my ($src, $dest) = @_;
 154         writeMsg($toServer, MSG_RENAME, pack('N/a*N/a*', $src, $dest));
 155         readMsg($fromServer); # MSG_PERFORMED
 156         print "*movefrom   $src\n",
 157                 "*moveto     $dest\n";
 158 }
 159
 160 sub doDelete($) {
 161         my ($path) = @_;
 162         writeMsg($toServer, MSG_DELREC, $path);
 163         my ($type, $body);
 164         while (($type, $body) = readMsg($fromServer), $type == MSG_DELETED) {
 165                 print "*deleting   $body\n";
 166         }
 167         # Also reads the final MSG_PERFORMED.
 168 }
 169
 170 # move_self so we can reliably detect moves out
 171 our @interestingEvents = ('modify', 'attrib', 'move', 'move_self', 'create', 'delete');
 172
 173 sub doClient($$) {
 174         ($src, $dest) = @_;
 175
 176         print "Continusync starting up.\n",
 177                 "This software is EXPERIMENTAL.  There is ABSOLUTELY NO WARRANTY.\n";
 178
 179         # Get a server process.
 180         # Echoes of rsync...
 181         if ($dest =~ /^([^:]*):(.*)$/) {
 182                 # Invoke over remote shell
 183                 my ($uhost, $rdest) = ($1, $2);
 184                 $serverPid = open2($fromServer, $toServer, @rshArgs, $uhost, $csPath, '--server');
 185                 # Pass path on stdin to stop the shell from messing with it.
 186                 # Echoes of rsync daemon protocol...
 187                 writeMsg($toServer, MSG_REMOTE_PATH, $rdest);
 188         } else {
 189                 # Fork locally
 190                 my ($fromClient, $toClient);
 191                 pipe($fromServer, $toClient);
 192                 pipe($fromClient, $toServer);
 193                 $serverPid = fork();
 194                 if ($serverPid == 0) {
 195                         # Child server
 196                         close($fromServer);
 197                         close($toServer);
 198                         open(STDIN, "<&", $fromClient);
 199                         open(STDOUT, ">&", $toClient);
 200                         doServer($dest);
 201                         exit(0);
 202                 }
 203                 # Parent client
 204                 close($fromClient);
 205                 close($toClient);
 206                 # Get a dest path that we can pass to rsync even after we chdir into the source.
 207                 {
 208                         local $^F = 100000;
 209                         open($localDestFH, '<', $dest);
 210                 }
 211                 $dest = "/proc/self/fd/" . fileno($localDestFH);
 212         }
 213
 214         chdir($src);
 215
 216         # Get inotifywait.
 217         $inwtPid = open($fromInwt, '-|');
 218         if ($inwtPid == 0) {
 219                 # Parent wants all our output on the single filehandle $fromInwt.
 220                 open(STDERR, ">&", STDOUT);
 221                 my @args = ('inotifywait', '-r', '-m', '--format', "%e\n%w\n%f", map(('-e', $_), @interestingEvents), '.');
 222                 exec(@args);
 223         }
 224
 225         <$fromInwt>; # `Setting up watches'
 226         <$fromInwt>; # `Watches established'
 227         $SIG{INT} = \&clientQuit;
 228         print "Continuously mirroring.  Give me a SIGINT when you want me to quit.\n";
 229
 230         # Now we can do the initial copy without danger of losing events.
 231         doRsync(1, 1, '.');
 232
 233         # Consecutive MOVED_FROM and MOVED_TO events constitute an internal
 234         # move.  A move-out followed by a move-in gives an intervening
 235         # MOVED_SELF, so we aren't fooled.
 236         my $movedFrom = undef;
 237
 238         for (;;) {
 239                 my ($e, $w, $f);
 240                 chomp($e = <$fromInwt>);
 241                 chomp($w = <$fromInwt>);
 242                 chomp($f = <$fromInwt>);
 243                 my $path = $w . $f;
 244                 $path =~ s,^\./(.),$1,; # Remove initial ./ if it isn't all
 245                 my $isDir = ($e =~ s/,ISDIR$//);
 246                 #print "Got event: ($e,$isDir,$w,$f)\n";
 247
 248                 if (defined($movedFrom)) {
 249                         if ($e eq 'MOVED_TO') {
 250                                 # Complete the move.
 251                                 doRename($movedFrom, $path);
 252                                 next;
 253                         } else {
 254                                 # Moved out.
 255                                 doDelete($movedFrom);
 256                         }
 257                         $movedFrom = undef;
 258                 }
 259
 260                 if ($e eq 'MODIFY') {
 261                         doRsync(0, 0, $path);
 262                 } elsif ($e eq 'ATTRIB') {
 263                         doRsync(0, 0, $path);
 264                 } elsif ($e eq 'MOVED_FROM') {
 265                         $movedFrom = $path;
 266                 } elsif ($e eq 'MOVED_TO') {
 267                         # Moved in.
 268                         # Must be recursive in case it was an entire directory.
 269                         doRsync(1, 0, $path);
 270                 } elsif ($e eq 'CREATE') {
 271                         doRsync(0, 0, $path);
 272                 } elsif ($e eq 'DELETE') {
 273                         doDelete($path);
 274                 }
 275         }
 276         # not reached
 277 }
 278
 279 if ($ARGV[0] eq '--server') {
 280         #STDOUT->autoflush(1);
 281         my ($type, $dest) = readMsg(STDIN);
 282         doServer($dest);
 283 } else {
 284         doClient($ARGV[0], $ARGV[1]);
 285 }
 286