The problem:

You've got 100+ servers in your web server cluster, and would like to keep them all synchronized from a centralized location. Furthermore, you'd like that replication to be instantaneous.

The solution: FREP

FREP was born out of a need I had to keep a cluster of server synchronized to a central location with instant file replication (one-way replication).

I realized that polling for file changes was going to be out of the question since the application would get progressively slower the more files were added.

I wanted something that wasn't going to be a ressource hog, and since I was looking at replicating over 200,000 files I wanted something that would scale too. When inotify was released in the vanilla linux kernel, I knew I had found the tool I needed to accomplish this job.

The next part of the puzzle that I needed to figure out what the communication mechanism that I was going to need to have in place to communicate changes to a large number of clients. With IP broadcast and multicast capabilities, the Spread cluster message toolkit was a natrual fit. Using Spread's multicasting capabilities, FREP should be able to scale to hundreds (if not thousands) of nodes.

The final piece of the puzzle was figuring out a bandwidth efficient mechanism for file transfers. I initially wanted to implement bits of the rsync protocol, but instead settled on a combination of Zlib compression on file chunks combined with using the generic diff format specification for partial data transfers. This combination should minimize the amount of bandwidth necessary to have file changes replicated to the cluster.

The system consists of several components:

• FREP_Server: The file monitor daemon
• FREP_Resync: The resynchronization daemon
• FREP_FileServer: A generic file server
• FREP_Client: The client installed on the nodes
• FREP_ResyncClient: A resynchronization client installed on the nodes as a failsafe mechanism
• FREP_Mon: Testing application
  

Commands to run use the Courier New font and are highlighted in orange:

/path/to/a/command/to/run --options

File contents are always displayed within a text area field:

#!/bin/bash echo "Mark rocks"

(sorry firefox folks, copy-paste button only works with IE)

Explanations use Arial – This text.

• Working Linux server
• You need to be root to install these packages
• Linux kernel with inotify compiled into the kernel (2.6.13+)
• Perl 5.6+ (tested with 5.8)
• Libevent installed and working properly. I highly recommend that you ensure that libevent can use the epoll() syscall mechanism.
• Daemontools installed
• Working Spread network. The Spread configuration depends on the number of nodes you'll have in the cluster and the data throughput (filechange bandwidth) you are expecting. More on this later.
• You'll need to have an idea of how many files you are gong to want to watch. More on this later.

13/02/2007 - Version 1.0 Initial public release

15/02/2007 - Version 1.1 Bug fixes, configuration management

This application was designed with the following goals in mind:

• Speed
• Reliability/fault tolerance
• Efficiency

These goals have been achieved by implementing the following features:

• Intelligent file transfers using a generic binary difference algorithm
• Compressed data transfers.
• Reliable message transport mechanism which can leverage broadcast and multicast IP using the Spread cluster messaging toolkit.
• Automatic resynchronization with a stream-based approach. File lists are sent to clients by traversing the filesystem hierarchy incrementally.
• Kernel based file change notification using Inotify.
• Edge-level event notification mechanisms using libevent (which in turn uses the epoll() kernel call).
• Client automatically drops root privileges on start up, all files created by the client owned by an unprivileged user.
• Several levels of fault-tolerance using application retries and automatic resynchronization.
• When unicast traffic is necessary, the system uses the kernel sendfile() functionality for zero-copy file transfer operations.
• Messages are split between several synchronization groups. (server roles dictate which files will be synchronized).
• When possible, expensive operations are cached.

Here's a visio diagram of the client and server operation

Synchronization application: FREP_Server

The following messages are sent by the synchronization application using the spread toolkit.

Each message is sent to a specific synchronization group as defined by the configuration hash and the file path.

File creation

Command:CREATE\n
Chunk:<data chunk count>\n
LastChucnk:<set to 1 when current chunk is the last chunk>\n
Cookie:<cookie value>\n
FilePath:<filepath>\n
FileSig:<file signature>\n
__DATA__\n
<Compressed data>__DATAEND__

File modification

Command:MODIFY\n
Chunk:<data chunk count>\n
LastChucnk:<set to 1 when current chunk is the last chunk>\n
Cookie:<cookie value>\n
FilePath:<filepath>\n
FileSig:<file signature>\n
OriginFileSig:<shadow copy file signature>\n
__DATA__\n
<Compressed delta data>__DATAEND__

File deletion

Command:DELETE\n
FilePath:<filepath>\n

File move

Command:MOVE\n
FilePath:<old filepath>\n
MovedTo:<new filepath>\n

Folder move

Command:MOVEDIR\n
FilePath:<old filepath>\n
MovedTo:<new filepath>\n

Heartbeat

Command: HEARTBEAT\n

Resynchronization application : FREP_Resync

The resynchronization application sends out the following messages using the Spread toolkit to specific synchronization groups.

Re-synchronization

Command:Sync\n
Chunk:<data chunk count>\n
LastChucnk:<set to 1 when current chunk is the last chunk>\n
Cookie:<cookie value>\n
__DATA__\n
<Compressed synchronization data>__DATAEND__

The synchronization data format is:

DIR:/file/path/1\n
Fileentry1\n
Fileentry2\n
FileentryN\n
DIR:/file/path/2\n
FileEntry1\n
DIR:/file/path/N\n
FileEntryN\n
…

Note: The file entries are relative to the last directory entry.

Synchronization client : FREP_Client

The synchronization client uses the following commands when communicating with the FREP_FileServer.

Retrieve file

GET <filepath>\n

Retrieve file signature

SIG <filepath>\n

File server: FREP_FileServer

Signature format  (response)

<md5 hex hash>|<filesize in bytes>|<file modification time>\n

File Transfer

The files are transferred as binary data with no encapsulation or header information.

The following instructions are for the installation of the server side code (the server that will be the main file repository).

Download and install prerequisite perl modules.

perl -MCPAN -e shell
cpan> install Bundle::CPAN IO::Stringy Compress::Zlib
cpan> install Sys::Syscall Digest::Adler32 Digest::MD5
cpan> install Linux::Inotify2 File::Path Event::Lib Spread
cpan> install Algorithm::GDiffDelta BerkeleyDB MLDBM Storable

Create some run scripts for daemontools

mkdir /etc/FREP_Server /etc/FREP_FileServer /etc/FREP_Resync
cat <<EOF >/etc/FREP_Resync/run
#!/bin/sh
exec > /dev/null
exec 2>&1
exec /usr/sbin/FREP_Resync
EOF
chmod +x /etc/FREP_Resync/run
cat <<EOF >/etc/FREP_Server/run
#!/bin/sh
exec > /dev/null
exec 2>&1
exec /usr/sbin/FREP_Server
EOF
chmod +x /etc/FREP_Server/run
cat <<EOF >/etc/FREP_FileServer/run
#!/bin/sh
exec > /dev/null
exec 2>&1
exec /usr/sbin/FREP_FileServer
EOF
chmod +x /etc/FREP_FileServer/run

Next up, the applications.

Edit /usr/sbin/FREP_Server:

#!/usr/bin/perl # FREP_Server - file replication/synchronization daemon. # File monitoring done with inotify/epoll, file differences calculated using gdiff # algorithm (see http://www.w3.org/TR/NOTE-gdiff-19970901), Zlib compression used # to minimize file transfers. # # This product uses software developed by Spread Concepts LLC for use in the Spread toolkit. # For more information about Spread see http://www.spread.org # # Copyright (C) 2007 Mark Steele http://www.control-alt-del.org/code # # This program is free software; you can redistribute it and/or # modify it under the terms of the GNU General Public License # as published by the Free Software Foundation; either version 2 # of the License, or (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. # use Linux::Inotify2; use File::Find; use POSIX qw(nice SIGCHLD SIGTERM SIGHUP); use strict; use File::Path; use File::Basename; use Event::Lib; use Spread qw(:ERROR :MESS); use Digest::MD5 qw(md5_hex); use Algorithm::GDiffDelta qw(gdiff_delta gdiff_apply); use File::Temp qw(:seekable :POSIX); use Sys::Syslog; use Compress::Zlib; use File::Copy; use XML::Simple; use Fcntl qw(F_GETFL F_SETFL O_NONBLOCK); no strict 'subs'; $| = 1; $0 = 'FREP_Server => starting up'; openlog($0, 'cons,pid', 'user'); my $CONFIG; eval {$CONFIG = XMLin("/etc/FREP.conf",ForceArray => ['Dir','Pool']);}; die "Couldn't parse config: $!" if ($@); ## Configuration sanity checks for (qw(ShadowRoot HeartbeatTimer MaxWatches QueueTimer)) { die("Missing configuration parameter: $_") if (!$CONFIG->{Sync}->{$_}); } die "No pools defined" if (!scalar(keys %{$CONFIG->{Pool}})); ## Go through each pool and make sure it contains an array for (keys %{$CONFIG->{Pool}}) { die "Configuration error, pools seem misconfigured" if (ref($CONFIG->{Pool}->{$_}->{Dir}) ne 'ARRAY'); } ## End sanity checking open(F,">/proc/sys/fs/inotify/max_user_watches") || die "couldn't open kernel max_user_watches: $!"; print F $CONFIG->{Sync}->{MaxWatches}; close(F); my $EVENT_PID; nice(-19); my $inotify = new Linux::Inotify2; die if (!$inotify); if (!-d $CONFIG->{Sync}->{ShadowRoot}) { eval {mkpath($CONFIG->{Sync}->{ShadowRoot},0755);}; if ($@) { die "error creating path: $!"; } } else { opendir(DIR,$CONFIG->{Sync}->{ShadowRoot}); while(readdir(DIR)) { next if (/^\.{1,2}$/); if (-d "$CONFIG->{Sync}->{ShadowRoot}/$_") { eval {rmtree("$CONFIG->{Sync}->{ShadowRoot}/$_")}; die "couldn't delete tree $CONFIG->{Sync}->{ShadowRoot}/$_: $!" if ($@); } else { unlink("$CONFIG->{Sync}->{ShadowRoot}/$_") || die "couldn't unlink file $CONFIG->{Sync}->{ShadowRoot}/$_: $!"; } } closedir(DIR); } ## Build fast regular expression lookup tables to determine if an item is part of ## a specific watch group or ignore list. my (%WatchRegex,$ignore_regex); if (defined($CONFIG->{IgnoreList}->{Dir}) && ref($CONFIG->{IgnoreList}->{Dir}) eq 'ARRAY') { my $expr = join('||', map { "m#^@{$CONFIG->{IgnoreList}->{Dir}}[$_]#o" } 0..$#{$CONFIG->{IgnoreList}->{Dir}}); $ignore_regex = eval "sub { $expr }"; die "couldn't build regexes" if $@; } else { $ignore_regex = eval "sub { return 0 }"; ## Avoid errors if no ignore paths defined } ########## $WatchRegex{'Admin'}->($itm) to test... foreach my $itm (keys %{$CONFIG->{Pool}}) { my $expr = join('||',map {"\$a =~ m#^@{$CONFIG->{Pool}->{$itm}->{Dir}}[$_]#o" } 0..$#{$CONFIG->{Pool}->{$itm}->{Dir}}); $WatchRegex{$itm} = eval "sub { \$a = shift; $expr }"; die "couldn't build regexes" if $@; } ## Unique folder list my %dirs; foreach my $group (keys %{$CONFIG->{Pool}}) { for (@{$CONFIG->{Pool}->{$group}->{Dir}}) { if (!-d) { eval {mkpath($_,0755);}; if ($@) { die "error creating path: $!"; } } $dirs{$_} = 1; } } ## Shadow copy of all files/folders will allow us to create differences between files debug_log("Creating shadow copy"); ## There may be race conditions here, however this shouldn't be too much of a problem as ## the copy will converge to the live filesystem over time since files are copied on change/modify for (keys %dirs) { find({ wanted => \&init_process, follow => 1, no_chdir => 1}, $_); ## On start-up, we want to sync the watchgroups and shadowroot folders. ## Eventually, should replace this with a pure perl copy just to keep this clean. `cp -a --parent $_ $CONFIG->{Sync}->{ShadowRoot}`; } undef %dirs; my @DataQueue; my %OpsQueue; my $timer_clear_data_queue = timer_new(\&clear_data_queue); $timer_clear_data_queue->add($CONFIG->{Sync}->{QueueTimer}); my $send_heartbeat = timer_new(\&send_heartbeat); $send_heartbeat->add($CONFIG->{Sync}->{HeartbeatTimer}); ## Manage signals for process management my $sig_chld = signal_new(SIGCHLD, \&reap); $sig_chld->add; my $sig_term = signal_new(SIGTERM,\&ASSASSIN); $sig_term->add; ## Open a filehandle to inotify's filedescriptor open(INOTIFY_FH,"<&=",$inotify->fileno) || die "couldn't open fd: $!"; my $flags = fcntl(INOTIFY_FH, F_GETFL, 0) or die "Can't get flags for the inotify FH: $!\n"; fcntl(INOTIFY_FH, F_SETFL, $flags | O_NONBLOCK) or die "Can't set O_NONBLOCK flag for the inotify FH: $!\n"; my $ev_inotify_read = event_new(*INOTIFY_FH, EV_READ|EV_PERSIST, \&process_inotify_events); $ev_inotify_read->add(); debug_log("Starting main loop"); $0 = "FREP_Server => Watching files"; event_mainloop(); sub process_inotify_events { ## NOTE: The last event received for a file in the same batch of events will be the one that 'wins'. my @events = $inotify->read; ## Shouldn't block as we've been triggered by epoll my %move_queue; foreach my $e (@events) { my $file = $e->fullname; if ($e->IN_CLOSE_WRITE) { debug_log("Detected CLOSE_WRITE on $file"); if (!defined($OpsQueue{$file})) { if (-e "$CONFIG->{Sync}->{ShadowRoot}$file") { push(@DataQueue,{'Command' => 'MODIFY', 'FilePath' => $file}); } else { debug_log("Origin file $CONFIG->{Sync}->{ShadowRoot}$file doesn't exist"); push(@DataQueue,{'Command' => 'CREATE', 'FilePath' => $file}); } $OpsQueue{$file} = $#DataQueue; ## This contains the array index of the last operation that occurred on this file in the data queue } else { # The file is already in the queue. if (-e "$CONFIG->{Sync}->{ShadowRoot}$file") { ## It's not a new file $DataQueue[$OpsQueue{$file}]->{Command} = 'MODIFY'; } else { $DataQueue[$OpsQueue{$file}]->{Command} = 'CREATE'; } } } if ($e->IN_CREATE) { # We'll catch this when data is written and the file is closed with close_write events debug_log("Detected CREATE on $file"); init_process($file); find({ wanted => \&init_process, follow => 1, no_chdir => 1}, $file) if ($e->IN_ISDIR); } if ($e->IN_DELETE) { debug_log("Detected DELETE on $file"); if (!defined($OpsQueue{$file})) { push(@DataQueue, {'Command' => 'DELETE','FilePath' => $file}); $OpsQueue{$file} = $#DataQueue; } else { $DataQueue[$OpsQueue{$file}]->{Command} = 'DELETE'; } } ### THIS CODE PROBABLY NEEDS SOME REVIEW... Moves are more complicated... Not dealing with possibility of having the file ### already in the data queue. Need to fix this. if ($e->IN_MOVED_FROM) { ## Use the cookie to store the name of the file being moved (from) debug_log("Received MOVE from, setting cookie"); $move_queue{$e->cookie} = $file if ($file ne ''); } if ($e->IN_MOVED_TO) { if ($move_queue{$e->cookie}) { ## Use the cookie to figure out what the original file name was if ($e->IN_ISDIR) { ## Ack, folder move. debug_log("Received MOVETO with cookie for DIR"); find({ wanted => \&init_process, follow => 1, no_chdir => 1}, $file); push(@DataQueue,{'Command' => 'MOVEDIR','FilePath' => $move_queue{$e->cookie},'FileTo' => $file}); } else { debug_log("Received MOVETO with cookie"); push(@DataQueue,{'Command' => 'MOVE','FilePath' => $move_queue{$e->cookie},'FileTo' => $file}); } delete($move_queue{$e->cookie}); ## don't need cookie anymore } else { ## No cookie?? debug_log("Move to with no cookie :("); if ($e->IN_ISDIR) { ## Maybe we weren't watching this folder before the move. Add it, and resync will take care of files in that folder. find({ wanted => \&init_process, follow => 1, no_chdir => 1}, $file); } else { ## This is a file that's been moved. Treat it as a new file. push(@DataQueue,{'Command' => 'CREATE', 'FilePath' => $file}); $OpsQueue{$file} = $#DataQueue; } } init_process($file); } if ($e->IN_Q_OVERFLOW) { $0 = "FREP_Server => Watching files (got some overflowed events)"; debug_log("Lost events!"); } } } sub reap { my $p = wait; if ($p == $EVENT_PID) { $EVENT_PID = 0; } } sub ASSASSIN { if ($EVENT_PID) { kill(9,$EVENT_PID); } exit; } sub clear_data_queue { my $e = shift; $e->add($CONFIG->{Sync}->{QueueTimer}); ## We have received events. if (scalar @DataQueue <= 0 || $EVENT_PID) { ## Nothing in queue or already have spawned a child. return; } if ($EVENT_PID = fork()) { @DataQueue = (); %OpsQueue = (); return; } else { ## Child nice(19); $0 = "FREP_Server => child clearing queue"; ## Connect to spread my ($mbox, $private_group) = Spread::connect({ spread_name => $CONFIG->{Spread}->{Port}.'@'.$CONFIG->{Sperad}->{Address},private_name => "$$-".int(rand(100)) }); if (!$mbox) { debug_log("Ack no mailbox, bailing (losing messages that were queued)"); exit; } for (@DataQueue) { if ($_->{Command} eq 'CREATE') { send_new_file($mbox,$_->{FilePath}); } elsif ($_->{Command} eq 'MODIFY') { send_modified_file($mbox,$_->{FilePath}); } elsif ($_->{Command} eq 'DELETE') { send_delete_file($mbox,$_->{FilePath}); } elsif ($_->{Command} eq 'MOVE') { send_move_file($mbox,$_->{FilePath},$_->{FileTo}); } elsif ($_->{Command} eq 'MOVEDIR') { send_move_dir($mbox,$_->{FilePath},$_->{FileTo}); } else { debug_log("ACK: Unknown command"); next; } } Spread::disconnect($mbox); exit(0); } } sub init_process { $_ ||= shift; ## $_ is set when using File::Find, otherwise we grab arg passed to function if ($ignore_regex && &$ignore_regex) { undef $_; return; } $inotify->watch($_,IN_CLOSE_WRITE|IN_CREATE|IN_DELETE|IN_IGNORED|IN_MOVED_TO|IN_MOVED_FROM) || debug_log("Couldn't add watch for $_: $!"); undef $_; } sub md5_file { my $file = shift; if (!open(FILE, $file)) { debug_log("couldn't open file for MD5 checksum"); return undef; } binmode(FILE); my $digest = Digest::MD5->new->addfile(*FILE)->hexdigest; close(FILE); return $digest; } sub find_watchgroup { my @groups; my $itm = shift; for (keys %{$CONFIG->{Pool}}) { push(@groups,$_) if ($WatchRegex{$_}->($itm)); } return @groups; } sub debug_log { return if (!$CONFIG->{Sync}->{Debug}); syslog('debug', '%s', shift); } sub crit_log { openlog($0, 'cons,pid', 'user'); syslog('debug', '%s', shift); } sub send_new_file { my ($mbox,$file) = @_; my ($filesize,$modtime) = ((stat($file))[7,9]); my $sig = md5_file($file) . '|' . $modtime; my @groups = find_watchgroup($file); my $cookie = time . int(rand(1000)); debug_log("Sending new file"); my ($chunk,$lastchunk,$totalread); if (!open(FILE_SEND,"<$file")) { crit_log("Unable to open file in send_new_file: $!"); return; } binmode(FILE_SEND); while (my $t = sysread(FILE_SEND, my $buf,90000)) { if (!defined($t)) { next if $! =~ /^Interrupted/; debug_log("ACK: system read error: $!"); return; } $totalread += $t; if ($totalread == $filesize) { debug_log("Setting lastchunk bit"); $lastchunk = "LastChunk:1\n"; } my $sendbuf = compress($buf); if (!$sendbuf) { debug_log("Ack error compressing buffer"); return; } $chunk++; foreach my $group (@groups) { debug_log("Sending $chunk to $group length:" . length($sendbuf)); my $r = Spread::multicast($mbox,SAFE_MESS|SELF_DISCARD,$group,0,"Command:CREATE\nChunk:$chunk\n".$lastchunk."Cookie:$cookie\nFilePath:$file\nFileSig:$sig\n__DATA__\n$sendbuf"."__DATAEND__"); if ($r < 0) { if ($r == ILLEGAL_SESSION) { debug_log("Ack: mailbox no good! (maybe disconnected?)"); } elsif ($r == ILLEGAL_MESSAGE) { debug_log("Ack: the message had an illegal structure! (oops)"); } elsif ($r == CONNECTION_CLOSED) { debug_log("Ack: spread closed!"); } else { debug_log("Ack: dunno whats wrong"); } } } } close(FILE_SEND); ## We want to move the file to it's new location. if (!-d File::Basename::dirname("$CONFIG->{Sync}->{ShadowRoot}$file")) { eval {mkpath(File::Basename::dirname("$CONFIG->{Sync}->{ShadowRoot}$file"),0755);}; if ($@) { crit_log("error creating path: $!"); return; } } if (!copy($file,"$CONFIG->{Sync}->{ShadowRoot}$file") || !chmod(0644,"$CONFIG->{Sync}->{ShadowRoot}$file") || !utime($modtime,$modtime,"$CONFIG->{Sync}->{ShadowRoot}$file")) { crit_log("error copying file to shadowcopy: $!"); } } sub send_modified_file { my ($mbox,$file) = @_; my $oldsig = md5_file("$CONFIG->{Sync}->{ShadowRoot}$file") . '|' . (stat("$CONFIG->{Sync}->{ShadowRoot}$file"))[9]; my ($filesize,$modtime) = (stat($file))[7,9]; my $sig = md5_file($file) . '|' . $modtime; my @groups = find_watchgroup($file); my $cookie = time . int(rand(1000)); debug_log("Sending diff file"); my $fh_diff = new File::Temp(DIR => '/www/synctemp'); if (!$fh_diff) { debug_log("Couldn't create tempfile: $!"); return; } binmode($fh_diff); if (!open(FH_ORIG,"<$CONFIG->{Sync}->{ShadowRoot}$file")) { debug_log("couldn't open original: $!"); return; } binmode(FH_ORIG); if (!open(FH_NEW,"<$file")) { debug_log("couldn't open new: $!"); return; } binmode(FH_NEW); eval {gdiff_delta(FH_ORIG, FH_NEW, $fh_diff);}; if ($@) { debug_log("Error creating patch file"); close(FH_NEW); close(FH_ORIG); return; } close(FH_NEW); close(FH_ORIG); $fh_diff->seek(0,0); my ($chunk,$lastchunk); while (my $t = read($fh_diff, my $buf,90000)) { if (!defined($t)) { next if $! =~ /^Interrupted/; debug_log("ACK: system read error: $!"); return; } if (eof($fh_diff)) { debug_log("Setting lastchunk bit"); $lastchunk = "LastChunk:1\n"; } my $sendbuf = compress($buf); if (!$sendbuf) { debug_log("Ack error compressing buffer"); return; } $chunk++; foreach my $group (@groups) { my $r = Spread::multicast($mbox,SAFE_MESS|SELF_DISCARD,$group,0,"Command:MODIFY\nChunk:$chunk\n".$lastchunk."Cookie:$cookie\nFilePath:$file\nFileSig:$sig\nOriginFileSig:$oldsig\n__DATA__\n$sendbuf"."__DATAEND__"); if ($r < 0) { if ($r == ILLEGAL_SESSION) { debug_log("Ack: mailbox no good! (maybe disconnected?)"); } elsif ($r == ILLEGAL_MESSAGE) { debug_log("Ack: the message had an illegal structure! (oops)"); } elsif ($r == CONNECTION_CLOSED) { debug_log("Ack: spread closed!"); } else { debug_log("Ack: dunno whats wrong"); } } } } close($fh_diff); ## We want to move the file to it's new location. if (!-d File::Basename::dirname("$CONFIG->{Sync}->{ShadowRoot}$file")) { eval {mkpath(File::Basename::dirname("$CONFIG->{Sync}->{ShadowRoot}$file"),0755);}; if ($@) { crit_log("error creating path: $!"); return; } } if (!copy($file,"$CONFIG->{Sync}->{ShadowRoot}$file") || !chmod(0644,"$CONFIG->{Sync}->{ShadowRoot}$file") || !utime($modtime,$modtime,"$CONFIG->{Sync}->{ShadowRoot}$file")) { crit_log("error copying file to shadowcopy: $!"); } } sub send_delete_file { my ($mbox, $file) = @_; my @groups = find_watchgroup($file); my $cookie = time; foreach my $group (@groups) { my $r = Spread::multicast($mbox,SAFE_MESS|SELF_DISCARD,$group,0,"Command:DELETE\nFilePath:$file\n"); if ($r < 0) { if ($r == ILLEGAL_SESSION) { debug_log("Ack: mailbox no good! (maybe disconnected?)"); } elsif ($r == ILLEGAL_MESSAGE) { debug_log("Ack: the message had an illegal structure! (oops)"); } elsif ($r == CONNECTION_CLOSED) { debug_log("Ack: spread closed!"); } else { debug_log("Ack: dunno whats wrong"); } } } if (!-d "$CONFIG->{Sync}->{ShadowRoot}$file" && -f "$CONFIG->{Sync}->{ShadowRoot}$file") { ## File if (!unlink("$CONFIG->{Sync}->{ShadowRoot}$file")) { crit_log("Couldn't unlink shadow file $file: $!"); } else { debug_log("Deleted shadow $file"); } } else { ## Folder eval {rmtree("$CONFIG->{Sync}->{ShadowRoot}$file");}; if ($@) { debug_log("Couldn't remove shadow file/folder $file: $@"); } else { debug_log("Deleted shadow $file"); } } } sub send_move_file { my ($mbox, $filefrom, $fileto) = @_; my @groups = find_watchgroup($filefrom); my $cookie = time; foreach my $group (@groups) { my $r = Spread::multicast($mbox,SAFE_MESS|SELF_DISCARD,$group,0,"Command:MOVE\nFilePath:$filefrom\nMovedTo:$fileto\n"); if ($r < 0) { if ($r == ILLEGAL_SESSION) { debug_log("Ack: mailbox no good! (maybe disconnected?)"); } elsif ($r == ILLEGAL_MESSAGE) { debug_log("Ack: the message had an illegal structure! (oops)"); } elsif ($r == CONNECTION_CLOSED) { debug_log("Ack: spread closed!"); } else { debug_log("Ack: dunno whats wrong ($r)"); } } } if (!-d File::Basename::dirname("$CONFIG->{Sync}->{ShadowRoot}$fileto")) { eval {mkpath(File::Basename::dirname("$CONFIG->{Sync}->{ShadowRoot}$fileto"),0755);}; if ($@) { crit_log("error creating path during file move: $!"); return; } } if (!-f "$CONFIG->{Sync}->{ShadowRoot}$filefrom") { crit_log("ACK origin file doesn't exist for move"); if (!copy($fileto,"$CONFIG->{Sync}->{ShadowRoot}$fileto") || !chmod(0644,"$CONFIG->{Sync}->{ShadowRoot}$fileto")) { crit_log("error copying file to shadowcopy: $!"); } } else { if (!rename("$CONFIG->{Sync}->{ShadowRoot}$filefrom","$CONFIG->{Sync}->{ShadowRoot}$fileto")) { crit_log("error renaming file in move file: $!"); if (!copy($fileto,"$CONFIG->{Sync}->{ShadowRoot}$fileto") || !chmod(0644,"$CONFIG->{Sync}->{ShadowRoot}$fileto")) { crit_log("error copying file to shadowcopy after rename failutre: $!"); } unlink("$CONFIG->{Sync}->{ShadowRoot}$filefrom"); } } } sub send_move_dir { my ($mbox, $filefrom, $fileto) = @_; my @groups = find_watchgroup($filefrom); my $cookie = time; foreach my $group (@groups) { my $r = Spread::multicast($mbox,SAFE_MESS|SELF_DISCARD,$group,0,"Command:MOVEDIR\nFilePath:$filefrom\nMovedTo:$fileto\n"); if ($r < 0) { if ($r == ILLEGAL_SESSION) { debug_log("Ack: mailbox no good! (maybe disconnected?)"); } elsif ($r == ILLEGAL_MESSAGE) { debug_log("Ack: the message had an illegal structure! (oops)"); } elsif ($r == CONNECTION_CLOSED) { debug_log("Ack: spread closed!"); } else { debug_log("Ack: dunno whats wrong ($r)"); } } } eval {mkpath(File::Basename::dirname("$CONFIG->{Sync}->{ShadowRoot}$fileto"),0755);}; if ($@) { debug_log("error creating path: $!"); return; } rename("$CONFIG->{Sync}->{ShadowRoot}$filefrom","$CONFIG->{Sync}->{ShadowRoot}$fileto"); } sub send_heartbeat { my $e = shift; $e->add($CONFIG->{Sync}->{HeartbeatTimer}); #debug_log("Sending heartbeat"); ## Connect to spread my ($mbox, $private_group) = Spread::connect({ spread_name => $CONFIG->{Spread}->{Port}.'@'.$CONFIG->{Sperad}->{Address},private_name => "$$-".int(rand(100)) }); if (!$mbox) { debug_log("Ack no mailbox, bailing"); return; } foreach my $group (keys %{$CONFIG->{Pool}}) { my $r = Spread::multicast($mbox,SAFE_MESS|SELF_DISCARD,$group,0,"Command:HEARTBEAT\n"); if ($r < 0) { if ($r == ILLEGAL_SESSION) { debug_log("Ack: mailbox no good! (maybe disconnected?)"); } elsif ($r == ILLEGAL_MESSAGE) { debug_log("Ack: the message had an illegal structure! (oops)"); } elsif ($r == CONNECTION_CLOSED) { debug_log("Ack: spread closed!"); } else { debug_log("Ack: dunno whats wrong"); } } } Spread::disconnect($mbox); } __END__

Edit /usr/sbin/FREP_FileServer :

#!/usr/bin/perl # FREP_FileServer - file server used by FREP. # Calculate file signatures as well as sending files (uses sendfile() for zero-copy # file transfers. # # Copyright (C) 2007 Mark Steele http://www.control-alt-del.org/code # # This program is free software; you can redistribute it and/or # modify it under the terms of the GNU General Public License # as published by the Free Software Foundation; either version 2 # of the License, or (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. # use Event::Lib; use POSIX qw/SIGHUP SIGTERM/; use IO::Socket::INET; use Digest::MD5; use Sys::Syscall qw(sendfile); use Number::Bytes::Human qw(format_bytes); use BerkeleyDB; use MLDBM qw(BerkeleyDB::Hash Storable) ; use Fcntl; use XML::Simple; use strict; use Sys::Syslog; $| = 1; my %stats; my $CONFIG; eval {$CONFIG = XMLin("/etc/FMon.conf",ForceArray => ['Dir','Pool']);}; die "Couldn't parse config: $!" if ($@); ## Configuration sanity checks for (qw(Port Address ValidPathRegex ConnectionTimeout SignatureCache)) { die("Missing configuration parameter: $_") if (!$CONFIG->{FileServer}->{$_}); } ## End sanity checking # Create a listening socket my $server = IO::Socket::INET->new(LocalAddr => $CONFIG->{FileServer}->{Address}, LocalPort => $CONFIG->{FileServer}->{Port}, Proto => 'tcp', ReuseAddr => SO_REUSEADDR, Listen => 1, Blocking => 0,) or die $!; my %filecache; tie %filecache, 'MLDBM', -Filename => $CONFIG->{FileServer}->{SignatureCache}, -Flags=> DB_CREATE or die "Cannot open file: $!\n" ; my $main = event_new($server, EV_READ|EV_PERSIST, \&handle_incoming); $main->add; my $signal_hup = signal_new(SIGHUP, \&sig_hup_stats); $signal_hup->add; my $signal_term = signal_new(SIGTERM, \&sig_term); $signal_term->add; my $timer_stats = timer_new(\&sig_hup_stats); $timer_stats->add(5); $0 = "FREP_FileServer:"; event_mainloop(); sub handle_incoming { my $e = shift; my $h = $e->fh; my $client = $h->accept or return; $client->blocking(0); my $event = event_new($client, EV_READ, \&handle_client); $event->add($CONFIG->{FileServer}->{ConnectionTimeout}); } sub handle_client { my $e = shift; my $h = $e->fh; my $type = shift; if ($type == EV_TIMEOUT) { $e->remove; undef($e); close($h); $stats{"errors"}++; return; } my $cmd = shift; sysread($h,my $buf, 8192); if (!$buf) { close($h); $e->remove; undef($e); $stats{"errors"}++; return; } $cmd .= $buf; if (length($cmd) > 8192) { close($h); $e->remove; undef($e); $stats{"errors"}++; return; } if ($cmd !~ /(.*)\r?\n/) { my $event = event_new($h, EV_READ, \&handle_client,$cmd); $event->add($CONFIG->{FileServer}->{ConnectionTimeout}); return; } my $file = $1; $file =~ s/(?:\r|\n)//g; if ($file !~ /^(GET|SIG) /) { $stats{"errors"}++; close($h); $e->remove; undef($e); return; } my $cmd = $1; $file =~ s/^(?:GET|SIG) //; if ($file !~ /$CONFIG->{FileServer}->{ValidPathRegex}/) { $stats{"errors"}++; close($h); $e->remove; undef($e); return; } my ($filesize,$mtime) = (stat($file))[7,9]; if (!-f $file || -d _ || -z _) { $stats{"errors"}++; close($h); $e->remove; undef($e); return; } if ($cmd eq 'GET') { my $event = event_new($h, EV_WRITE, \&send_file, $file, 0,$filesize,open_file($file)); if ($event) { $event->add($CONFIG->{FileServer}->{ConnectionTimeout}); } else { close($h); $e->remove; undef($e); return; } } elsif ($cmd eq 'SIG') { my $sig; debug_log("SIG requested for $file"); if (my $cdata = $filecache{$file}) { debug_log("retrieved cached data"); if ($cdata->{MTIME} == $mtime) { debug_log("cached data matches on mtime"); $stats{'cachehit'}++; $sig = $cdata->{SIG} . '|' . $cdata->{FSIZE} . '|' . $cdata->{MTIME}; my $event = event_new($h, EV_WRITE, \&send_sig, $sig,0); $event->add($CONFIG->{FileServer}->{ConnectionTimeout}); return; } else { $stats{'cachemiss'}++; if (my $md5 = md5_file($file)) { my $cdata = { SIG => $md5, FSIZE => $filesize, MTIME => $mtime}; $sig = $cdata->{SIG} . '|' . $cdata->{FSIZE} . '|' . $cdata->{MTIME}; $filecache{$file} = $cdata; my $event = event_new($h, EV_WRITE, \&send_sig, $sig,0); $event->add($CONFIG->{FileServer}->{ConnectionTimeout}); return; } else { ## Barf debug_log("couldn't get MD5 checksum to add to cache. Barfing."); $stats{"errors"}++; close($h); $e->remove; undef($e); return; } } } else { debug_log("cache miss"); $stats{'cachemiss'}++; if (my $md5 = md5_file($file)) { my $cdata = { SIG => $md5, FSIZE => $filesize, MTIME => $mtime}; $sig = $cdata->{SIG} . '|' . $cdata->{FSIZE} . '|' . $cdata->{MTIME}; $filecache{$file} = $cdata; my $event = event_new($h, EV_WRITE, \&send_sig, $sig,0); $event->add($CONFIG->{FileServer}->{ConnectionTimeout}); return; } else { ## Barf debug_log("couldn't get MD5 checksum to add to cache. Barfing."); $stats{"errors"}++; close($h); $e->remove; undef($e); return; } } } else { $stats{"errors"}++; close($h); $e->remove; undef($e); return; } } sub send_file { my $e = shift; my $type = shift; my $h = $e->fh; my $file = shift; my $offset = shift; my $filesize = shift; my $fh = shift; if ($type == EV_TIMEOUT || !$file || !$fh) { $stats{"errors"}++; close($h); $e->remove; undef $e; close($fh); return; } my $sent = sendfile(fileno($h),fileno($fh),$filesize); if (!$sent) { ## Socket was ready for writing but couldn't write any data $stats{"errors"}++; close($h); close($fh); $e->remove; undef $e; return; } $stats{"bytes"} += $sent; if (($sent+$offset) == $filesize) { $stats{"sent"}++; close($h); close($fh); $e->remove; undef($e); } else { my $event = event_new($h, EV_WRITE, \&send_file, $file, $sent+$offset,$filesize,$fh); $event->add($CONFIG->{FileServer}->{ConnectionTimeout}); } } sub md5_file { my $file = shift; if (!open(FILE,$file)) { return undef; } binmode(FILE); my $digest = Digest::MD5->new->addfile(*FILE)->hexdigest; close(FILE); return $digest; } sub open_file { my $file = shift; local *F; if (!sysopen(F,$file,O_RDONLY)) { return undef; } return *F; } sub sig_hup_stats { my $e = shift; my $type = shift; if ($type == EV_TIMEOUT) { $e->add($CONFIG->{FileServer}->{ConnectionTimeout}); } $0 = qq#FREP_FileServer => chits: $stats{cachehit} cmiss: $stats{cachemiss} err: $stats{errors} files: $stats{'sent'} data: # . format_bytes($stats{'bytes'}); } sub send_sig { my $e = shift; my $h = $e->fh; my $type = shift; my $signature = shift; my $offset = shift; if ($type == EV_TIMEOUT) { $e->remove; undef($e); close($h); $stats{"errors"}++; return; } my $ret = 0; eval { $ret = syswrite($h,$signature,512,$offset); }; if ($@) { $e->remove; undef($e); close($h); return; } if ($ret == 0) { ## ERROR: couldn't write anything debug_log("couldn't write whole signature to client"); $e->remove; undef($e); close($h); return; } if ($ret != (length($signature)-$offset)) { ## Couldn't write everything, reschedule my $event = event_new($h, EV_WRITE, \&send_sig, $signature,$offset+$ret); $event->add($CONFIG->{FileServer}->{ConnectionTimeout}); } else { $e->remove; undef($e); close($h); } } sub sig_term { untie(%filecache); exit; } sub debug_log { return if (!$CONFIG->{FileServer}->{Debug}); openlog("FMon_FileServer", 'cons,pid', 'user'); syslog('debug', '%s', shift); closelog(); }

Edit /usr/sbin/FREP_Resync:

#!/usr/bin/perl # FREP_Resync - Resynchronization daemon used as a failsafe. # Walks the defines paths, builds a file list and sends it to the clients. # # This product uses software developed by Spread Concepts LLC for use in the Spread toolkit. # For more information about Spread see http://www.spread.org # # Copyright (C) 2007 Mark Steele http://www.control-alt-del.org/code # # This program is free software; you can redistribute it and/or # modify it under the terms of the GNU General Public License # as published by the Free Software Foundation; either version 2 # of the License, or (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. # use File::Find; use Spread qw(:ERROR :MESS); use strict; use File::Path; use Event::Lib; use File::Temp qw(:seekable); use POSIX qw(nice SIGCHLD SIGTERM SIGHUP); use Digest::MD5 qw(md5_hex); use Compress::Zlib; use File::Basename; use Sys::Syslog; use XML::Simple; $| = 1; nice(19); my $CONFIG; eval {$CONFIG = XMLin("/etc/FMon.conf",ForceArray => ['Dir','Pool']);}; die "Couldn't parse config: $!" if ($@); ## Configuration sanity checks for (qw(ResyncTimer)) { die("Missing configuration parameter: $_") if (!$CONFIG->{Resync}->{$_}); } die "No pools defined" if (!scalar(keys %{$CONFIG->{Pool}})); ## Go through each pool and make sure it contains an array for (keys %{$CONFIG->{Pool}}) { die "Configuration error, pools seem misconfigured" if (ref($CONFIG->{Pool}->{$_}->{Dir}) ne 'ARRAY'); } ## End sanity checking ## Build fast regular expression lookup tables to determine if an item is part of ## a specific watch group or ignore list. my %WatchRegex; my $ignore_regex; if (defined($CONFIG->{IgnoreList}->{Dir}) && ref($CONFIG->{IgnoreList}->{Dir}) eq 'ARRAY') { my $expr = join('||', map { "m#^@{$CONFIG->{IgnoreList}->{Dir}}[$_]#o" } 0..$#{$CONFIG->{IgnoreList}->{Dir}}); $ignore_regex = eval "sub { $expr }"; die "couldn't build regexes" if $@; } else { $ignore_regex = eval "sub { return 0 }"; } ########## $WatchRegex{'Admin'}->($itm) to test... foreach my $itm (keys %{$CONFIG->{Pool}}) { my $expr = join('||',map {"\$a =~ m#^@{$CONFIG->{Pool}->{$itm}->{Dir}}[$_]#o" } 0..$#{$CONFIG->{Pool}->{$itm}->{Dir}}); $WatchRegex{$itm} = eval "sub { \$a = shift; $expr }"; die "couldn't build regexes" if $@; } ## Unique folder list my (%DIRS,%FH); foreach my $group (keys %{$CONFIG->{Pool}}) { for (@{$CONFIG->{Pool}->{$group}->{Dir}}) { if (!-d) { eval {mkpath($_,0755);}; if ($@) { die "error creating path: $!"; } } $DIRS{$_} = 1; } } ## Setup signal handler for SIGHUP to start resync my $sig_hup = signal_new(SIGHUP,\&resync); $sig_hup->add(); ## Resync every 300 seconds my $timer_resync = timer_new(\&resync); $timer_resync->add($CONFIG->{Resync}->{ResyncTimer}); ## Update time left in process list my $timer_stats = timer_new(\&display_stats); $timer_stats->add(2); debug_log("Entering main loop"); event_mainloop(); sub find_watchgroup { my @groups; my $itm = shift; for (keys %{$CONFIG->{Pool}}) { push(@groups,$_) if ($WatchRegex{$_}->($itm)); } return @groups; } sub build_list { $_ ||= shift; if (&$ignore_regex) { undef $_; return; } if (-d) { ## Got directory foreach my $group (find_watchgroup($_ . '/')) { print {$FH{$group}} "DIR:$_\n"; } } elsif (-f && !-z) { foreach my $group (find_watchgroup($_)) { print {$FH{$group}} join(":",(scalar(fileparse($_)),(stat($_))[7,9])),"\n"; } } } sub resync { my $e = shift; my $type = shift; $0 = "FREP_Resync => Resync in progress"; debug_log("Starting resync"); if ($type == EV_TIMEOUT) { ## Timer caused us debug_log("Resync caused by timeout"); $e->add($CONFIG->{Resync}->{ResyncTimer}); } else { ## From a sighup, reset the timer debug_log("Resync caused by SIGHUP"); $timer_resync->remove; $timer_resync->add($CONFIG->{Resync}->{ResyncTimer}); } ## Remove old resync files if any `rm /tmp/*.resync`; ## Cleanup file handles undef %FH; File::Temp::cleanup(); ## Open up a filehandle for each watch group. for (keys %{$CONFIG->{Pool}}) { $FH{$_} = new File::Temp(SUFFIX => '.resync'); if (!$FH{$_}) { debug_log("Error creating temp file: $!"); return; } } debug_log("Filehandles open for each group"); ## Loop through unique directory listing and create a list of all files for each group. debug_log("Walking file tree"); for (keys %DIRS) { find({ wanted => \&build_list, follow => 1, no_chdir => 1}, $_); } debug_log("Finished walking tree"); ## Connect to spread my ($mbox, $private_group) = Spread::connect({ spread_name => $CONFIG->{Spread}->{Port}.'@'.$CONFIG->{Sperad}->{Address},private_name => "$$-".int(rand(100)) }); if (!defined($mbox)) { debug_log("Ack no mailbox, bailing"); return; } for (keys %{$CONFIG->{Pool}}) { send_sync_file($mbox,$FH{$_},$_); } Spread::disconnect($mbox); debug_log("Resync finished"); } sub debug_log { return if (!$CONFIG->{Resync}->{Debug}); openlog($0, 'cons,pid', 'user'); syslog('debug', '%s', shift); closelog(); } sub md5_file { my $file = shift; my $digest = Digest::MD5->new->addfile($file)->hexdigest; close(FILE); return $digest; } sub send_sync_file { my ($mbox,$fh,$group) = @_; my $filesize = (stat($fh->filename))[7]; $fh->seek(0,0); my $sig = md5_file($fh); $fh->seek(0,0); my $cookie = time . int(rand(1000)); my ($chunk,$lastchunk); debug_log("Sending to $group ".$fh->filename.": $sig"); while (my $t = read($fh, my $buf,90000)) { if (!defined($t)) { return if $! =~ /^Interrupted/; debug_log("ACK: system read error: $!"); return; } if (eof($fh)) { $lastchunk = "LastChunk:1\n"; } my $sendbuf = compress($buf); if (!$sendbuf) { debug_log("Ack error compressing buffer"); return; } $chunk++; my $r = Spread::multicast($mbox,SAFE_MESS|SELF_DISCARD,"Resync-$group",0,"Command:Sync\nChunk:$chunk\n".$lastchunk."Cookie:$cookie\nFileSig:$sig\n__DATA__\n$sendbuf"."__DATAEND__"); if ($r < 0) { if ($r == ILLEGAL_SESSION) { debug_log("Ack: mailbox no good! (maybe disconnected?)"); return; } elsif ($r == ILLEGAL_MESSAGE) { debug_log("Ack: the message had an illegal structure! (oops)"); return; } elsif ($r == CONNECTION_CLOSED) { debug_log("Ack: spread closed!"); return; } else { debug_log("Ack: dunno whats wrong"); return; } } } } sub display_stats { my $e = shift; $e->add(2); $0 = "FREP_Resync => next resync: " . int($timer_resync->pending) . " seconds"; }

Make them execuatble:

chmod +x /usr/sbin/FREP_FileServer
chmod +x /usr/sbin/FREP_Server
chmod +x /usr/sbin/FREP_Resync

Before going any further...

BIG SECURITY NOTICE:

***************************************************

THESE SERVICES ARE NOT DESIGNED TO FACE THE INTERNET. YOU SHOULD HAVE A FIREWALL BETWEEN THIS APPLICATION AND THE INTERNET.

***************************************************

That said, on to the configuration.

Edit /etc/FREP.conf:

<Config> <!-- Pool configuration --> <!-- Note: all watched dirs must be located on the same device --> <Pool name="Dynamic"> <Dir>/www/mysite.com/ad/</Dir> <Dir>/www/mysite.com/images/</Dir> <Dir>/www/mysite.com/php/</Dir> <Dir>/www/Smarty/</Dir> </Pool> <Pool name="Static"> <Dir>/www/mysite.com/images/</Dir> <Dir>/www/mysite.com/html/</Dir> </Pool> <!-- Spread configuration --> <Spread> <Port>4803</Port> <Address>192.168.1.100</Address> </Spread> <!-- FileServer configuration --> <FileServer> <Port>9000</Port> <Address>192.168.1.100</Address> <ValidPathRegex>^/www/</ValidPathRegex> <Debug>0</Debug> <SignatureCache>/tmp/FileCache.db</SignatureCache> <ConnectionTimeout>5</ConnectionTimeout> </FileServer> <!-- Ignore list --> <IgnoreList> <Dir>/asdfasdf</Dir> </IgnoreList> <!-- Resynchronization configuration --> <Resync> <ResyncTimer>300</ResyncTimer> <Debug>0</Debug> </Resync> <!-- synchronization configuration --> <Sync> <MaxWatches>524288</MaxWatches> <ShadowRoot>/www/ShadowRoot</ShadowRoot> <!-- MUST BE ON SAME DEVICE AS WATCHED FILES --> <HeartbeatTimer>3</HeartbeatTimer> <QueueTimer>5</QueueTimer> <Debug>0</Debug> </Sync> </Config>

There's basically six different things you need to configure to get FREP up and running. They are...

Spread configuration:

<!-- Spread configuration -->
<Spread>
<Port>4803</Port>
<Address>192.168.1.100</Address>
</Spread>

This defines the address and port of the Spread daemon. This should be localhost, but you could run the spread daemon on another box if you wanted to.

Pools:

<!-- Pool configuration -->
<!-- Note: all watched dirs must be located on the same device -->
<Pool name="Dynamic">
<Dir>/www/mysite.com/ad/</Dir>
<Dir>/www/mysite.com/images/</Dir>
<Dir>/www/mysite.com/php/</Dir>
<Dir>/www/Smarty/</Dir>
</Pool>
<Pool name="Static">
<Dir>/www/mysite.com/images/</Dir>
<Dir>/www/mysite.com/html/</Dir>
</Pool>

Each pool represents a group of severs that serve a specific function. A pool has to have a name, and a list of folders that we'll want to watch.

Ignore list:

<!-- Ignore list -->
<IgnoreList>
<Dir>/www/mysite.com/ad/ignoreme</Dir>

<Dir>/www/mysite.com/html/topsecret</Dir>
</IgnoreList>

The ignore list is a set of paths that we _DON'T_ want to synchronize or replicate.

Synchronization configuration:

<!-- synchronization configuration -->
<Sync>
<MaxWatches>524288</MaxWatches>
<ShadowRoot>/www/ShadowRoot</ShadowRoot> <!-- MUST BE ON SAME DEVICE AS WATCHED FILES -->
<HeartbeatTimer>3</HeartbeatTimer>
<QueueTimer>5</QueueTimer>
<Debug>0</Debug>
</Sync>

This section defines the maximun number of of files+folders (MaxWatches), if you try to watch more than this number of files+folders, it'll barf.

We also have the ShadowRoot configuration directive, which defines where FREP will create a copy of all the watched files/folders. This is used to be able to do partial file transfers (send only the changed parts of a file when it's modified). The ShadowRoot file path MUST be on the same device as the watched files.

The HeartBeatTimer directive tells FREP how often it should be sending out heartbeats to the client nodes. The clients will use this variable to determine if they need to reconnect to the spread network.

The QueueTimer defines how long FREP_Server will keep the received events in the queue before starting to process them. This is to avoid duplicate events triggering needless IO. For example, if you are writing a file in 5 chunks, inotify will be sending multiple file modification events. The queue is designed to schlep these all together. It also handles sequential operations. For example, suppose you open a file, write to it, close it, then delete it FREP_Server will only send 1 event (delete).

File server configuration:

<!-- FileServer configuration -->
<FileServer>
<Port>9000</Port>
<Address>192.168.1.100</Address>
<ValidPathRegex>^/www/</ValidPathRegex>
<Debug>0</Debug>
<SignatureCache>/tmp/FileCache.db</SignatureCache>
<ConnectionTimeout>5</ConnectionTimeout>
</FileServer>

The fileserver's address should be the IP address of the server that the FREP_Server sync daemon is running on. The port should be any unused port.

The ValidPathRegex is a safety measure that defines a perl-compatible regular expression that will be matched to see if the file signature or transfer will be allowed. As an example, if you're synchronization folders are /www/mysite.com and /files/syncme, you'd set the ValidPathRegex to: (?:/www/mysite.com|/files/syncme).

The FREP_FileServer keeps a BerkeleyDB cache of file signatures to speed things up. the SignatureCache directive tells the file server where to store it.

The ConnectionTimeout is the amount of time the fileserver will wait for data on a socket (for reading or writing). 5 seconds should be fine.

Resynchronization configuration:

<!-- Resynchronization configuration -->
<Resync>
<ResyncTimer>300</ResyncTimer>
<Debug>0</Debug>
</Resync>

The FREP_Resync daemon only has one tweakable setting, the ResyncTimer, which is the frequency at which it'll broadcast the full file list to all the clients.

Note:

The debug flag controls if debugging to syslog is enabled. These daemons are pretty verbose when the debug flag is on, so leave it off unless you're debugging :)

Start up the daemons:

cd /service
ln -s /etc/FREP_Server
ln -s /etc/FREP_FileServer
ln -s /etc/FREP_Resync

Wait a few seconds, daemontools should fire them all up. Check to make sure they are running with:

svstat /service/FREP*

That's all there is to do for the server side applications. Now onto the clients.

These are the instructions for the installation of the client applications.

First, the perl modules

perl -MCPAN -e shell
cpan> install Bundle::CPAN IO::Stringy Compress::Zlib
cpan> install Digest::Adler32 Digest::MD5
cpan> install File::Path Event::Lib Spread
cpan> install Algorithm::GDiffDelta Getopt::Long

Create some run scripts for daemontools

mkdir /etc/FREP_Client /etc/FREP_ResyncClient
cat <<EOF >/etc/FREP_Client/run
#!/bin/sh
exec > /dev/null
exec 2>&1
exec /usr/sbin/FREP_Client --pool=Static \
--userid=1500 --groupid=100 --spread-port=4803 \
--spread-address=192.168.1.100 --fs-port=9000 \
--fs-address=192.168.1.100
EOF
chmod +x /etc/FREP_Client/run
cat <<EOF >/etc/FREP_ResyncClient/run
#!/bin/sh
exec > /dev/null
exec 2>&1
exec /usr/sbin/FREP_ResyncClient --pool=Static \
--userid=1500 --groupid=100 --spread-port=4803 \
--spread-address=192.168.1.100 --fs-port=9000 \
--fs-address=192.168.1.100
EOF
chmod +x /etc/FREP_ResyncClient/run

Next, the file replication client. Edit /usr/sbin/FREP_Client:

#!/usr/bin/perl # FREP_Client - File replication client # # This product uses software developed by Spread Concepts LLC for use in the Spread toolkit. # For more information about Spread see http://www.spread.org # # Copyright (C) 2007 Mark Steele http://www.control-alt-del.org # # This program is free software; you can redistribute it and/or # modify it under the terms of the GNU General Public License # as published by the Free Software Foundation; either version 2 # of the License, or (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. # use Spread qw(:ERROR :MESS); use Event::Lib; use strict; use POSIX qw(nice SIGCHLD SIGTERM setuid setgid :sys_wait_h); use Digest::MD5 qw(md5_hex); use Sys::Syslog; use File::Path; use File::Basename; use File::Temp qw(:POSIX); use Algorithm::GDiffDelta qw(gdiff_apply); use IO::Socket; use Digest::MD5; use Compress::Zlib; use Getopt::Long; $| = 1; my ($DEBUG,$Pool,$USERID,$GROUPID,$SPREAD_PORT,$SPREAD_ADDRESS,$FILESERVER_PORT,$FILESERVER_ADDRESS, $HEARTBEAT_TIMEOUT,$QUEUE_TIMEOUT); GetOptions ("debug" => \$DEBUG, "pool=s" => \$Pool, "userid|u=i" => \$USERID, "groupid|g=i" => \$GROUPID, "spread-port=i" => \$SPREAD_PORT, "spread-address=s" => \$SPREAD_ADDRESS, "fs-port=i" => \$FILESERVER_PORT, "fs-address=s" => \$FILESERVER_ADDRESS, "heartbeat=i" => \$HEARTBEAT_TIMEOUT, "queue-timer=i" => \$QUEUE_TIMEOUT) || show_help(); $HEARTBEAT_TIMEOUT ||= 20; $QUEUE_TIMEOUT ||= 5; if (!$Pool || !$USERID || !$GROUPID || !$SPREAD_PORT || !$SPREAD_ADDRESS || !$FILESERVER_PORT || !$FILESERVER_ADDRESS) { show_help(); } if ($SPREAD_ADDRESS !~ /^(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)/ || $FILESERVER_ADDRESS !~ /^(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)/) { print "Invalid IP address\n"; show_help(); } my $CHILD; nice(-19); $0 = "FREP_Client($Pool)"; openlog($0, 'cons,pid', 'user'); setgid($GROUPID) || die; setuid($USERID) || die; umask(0022) || die; ## Just to be on safe side my @DataQueue; my ($mbox, $private_group) = Spread::connect({ spread_name => $SPREAD_PORT.'@'.$SPREAD_ADDRESS,private_name => "$$-".int(rand(100)) }); if (!$mbox) { crit_log("couldn't connect to spread $!"); die $!; } if (Spread::join($mbox, $Pool) < 0) { crit_log("couldn't join group"); die; } open(SPREAD_FH,"<&=$mbox") || die "couldn't open fd: $!"; my $HEARTBEAT = time; my %MSGDATA; my $timer_clear_data_queue = timer_new(\&clear_data_queue); $timer_clear_data_queue->add($QUEUE_TIMEOUT); my $timer_check_heartbeat = timer_new(\&check_heartbeat); $timer_check_heartbeat->add($HEARTBEAT_TIMEOUT); my $event = event_new(*SPREAD_FH, EV_READ|EV_PERSIST, \&get_spread_messages); $event->add; my $sig_chld = signal_new(SIGCHLD, \&REAPER); $sig_chld->add; my $sig_term = signal_new(SIGTERM,\&ASSASSIN); $sig_term->add; event_mainloop(); sub REAPER { if (waitpid($CHILD,0) != $CHILD) { debug_log("Ack! Child waited pid not the same!"); } $CHILD = 0; } sub ASSASSIN { kill(15, $CHILD) if ($CHILD); # 15 = SIGTERM exit; } sub clear_data_queue { my $e = shift; if (scalar @DataQueue) { if ($CHILD) { ## Don't spawn a child if there's already one running if (waitpid($CHILD, WNOHANG) != -1) { # Child already exists, wait for it to finish before spawning a new one $e->add($QUEUE_TIMEOUT); return; } else { # Child isn't there??? $CHILD = 0; } } if ($CHILD = fork()) { ## Parent undef @DataQueue; } else { ## Child nice(19); process_queue(); exit; } } $e->add($QUEUE_TIMEOUT); } sub get_spread_messages { while(my ($st, $s, $g, $mt, $e, $mess) = Spread::receive($mbox,0)) { return if (!$mess); my ($headers,$body) = parse_message($mess); process_message($headers,$body) if ($headers); } } sub process_queue { for (@DataQueue) { if ($_->{Headers}->{Command} eq 'DELETE') { delete_file($_->{Headers}->{FilePath}); } elsif ($_->{Headers}->{Command} eq 'FETCH') { fetch_file($_->{Headers}->{FilePath},0); } elsif ($_->{Headers}->{Command} eq 'CREATE') { create_file($_->{Headers}->{FilePath},$_->{Headers}->{FileSig},$_->{Body}), } elsif ($_->{Headers}->{Command} eq 'MODIFY') { modify_file($_->{Headers}->{FilePath},$_->{Headers}->{FileSig},$_->{Headers}->{OriginFileSig},$_->{Body}); } elsif ($_->{Headers}->{Command} eq 'MOVE') { move_file($_->{Headers}->{FilePath},$_->{Headers}->{MovedTo}); } elsif ($_->{Headers}->{Command} eq 'MOVEDIR') { move_dir($_->{Headers}->{FilePath},$_->{Headers}->{MovedTo}); } else { ## Invalid debug_log("Invalid command"); } } } sub md5_file { my $file = shift; if (!open(FILE,"$file")) { debug_log("couldn't open file for MD5: $!"); return undef; } binmode(FILE); my $digest = Digest::MD5->new->addfile(*FILE)->hexdigest; close(FILE); return $digest; } sub debug_log { return if (!$DEBUG); syslog('debug', '%s', shift); } sub crit_log { syslog('debug', '%s', shift); } sub parse_message { my $mess = shift; return if ($mess !~ /^Command:/); my ($headers,$body,%headers); if ($mess =~ /^(.*?)__DATA__\n(.+?)__DATAEND__$/s) { ## This is a message that contains a data segment ($headers,$body) = ($1,$2); } else { $headers = $mess; } ## Parse headers my @headers = split(/\n/,$headers); for (@headers) { my ($key,$val) = split(/:/,$_); $headers{$key} = $val; } return(\%headers,$body); } sub process_message { my ($headers,$body) = @_; ## If this is a heartbeat, update time stamp. if ($headers->{Command} eq 'HEARTBEAT') { $HEARTBEAT = time; return; } if (defined($MSGDATA{'Cookie'})) { debug_log("Continuation of multi-chunk"); if ($headers->{'Cookie'} != $MSGDATA{'Cookie'}) { debug_log("Cookie doesn't match. Retrieve file from fileserver, process this new message"); push(@DataQueue,{ Headers => { Command => 'FETCH', FilePath => $MSGDATA{'FilePath'}}}); unlink($MSGDATA{'FILENAME'}); %MSGDATA = (); $timer_clear_data_queue->add(1); process_message($headers,$body); return; } if ($MSGDATA{'Chunk'} !~ /^(\d+)$/) { ## Odd, chunks doesn't contain what we expect. Abort. debug_log("Chunks seem messed up, couldn't parse current chunk data"); push(@DataQueue,{ Headers => { Command => 'FETCH', FilePath => $MSGDATA{'FilePath'}}}); unlink($MSGDATA{'FILENAME'}); %MSGDATA = (); $timer_clear_data_queue->add(1); return; } my $curchunk = $1; if ($headers->{Chunk} !~ /^(\d+)$/) { ## Odd, chunks doens't contain what we expect. Abort. debug_log("Chunks weird. Last chunk processed: $curchunk. This chunk couldn't be parsed"); push(@DataQueue,{ Headers => { Command => 'FETCH', FilePath => $MSGDATA{'FilePath'}}}); unlink($MSGDATA{'FILENAME'}); %MSGDATA = (); $timer_clear_data_queue->add(1); return; } my $newchunk = $1; if ($newchunk != ($curchunk + 1)) { ## Out of order. debug_log("Chunks out of order or missing. Retrieving file from fileserver, scrapping this message"); push(@DataQueue,{ Headers => { Command => 'FETCH', FilePath => $MSGDATA{'FilePath'}}}); unlink($MSGDATA{'FILENAME'}); %MSGDATA = (); $timer_clear_data_queue->add(1); return; } ## Step 3. Write data to tmpfile. if (!open(FH,">>".$MSGDATA{'FILENAME'})) { debug_log("Error opening temp file: $!"); push(@DataQueue,{ Headers => { Command => 'FETCH', FilePath => $MSGDATA{'FilePath'}}}); %MSGDATA = (); $timer_clear_data_queue->add(1); return; } binmode(FH); my $output = uncompress($body); if (!$output) { debug_log("Couldn't uncompress data"); push(@DataQueue,{ Headers => { Command => 'FETCH', FilePath => $headers->{'FilePath'}}}); close(FH); unlink($MSGDATA{'FILENAME'}); return; } if (syswrite(FH,$output) != length($output)) { debug_log("Couldn't write all data to file for some reason"); push(@DataQueue,{ Headers => { Command => 'FETCH', FilePath => $MSGDATA{'FilePath'}}}); close(FH); unlink($MSGDATA{'FILENAME'}); %MSGDATA = (); $timer_clear_data_queue->add(1); return; } close(FH); ## Step 4. Last chunk? if ($headers->{LastChunk}) { debug_log("Got last chunk"); push(@DataQueue,{Headers => $headers, Body => $MSGDATA{'FILENAME'}}); $timer_clear_data_queue->add(1); %MSGDATA = (); return; } else { ## Update chunk count debug_log("Not last chunk"); $MSGDATA{'Chunk'} = $headers->{'Chunk'}; } return; } elsif (!$headers->{Chunk}) { ## Single chunk message if ($body) { debug_log("Single chunk message with body"); my $file = tmpnam(); if (!open(FH,">$file")) { debug_log("Ack couldn't open tmp file: $!"); push(@DataQueue,{ Headers => { Command => 'FETCH', FilePath => $headers->{'FilePath'}}}); return; } binmode(FH); my $output = uncompress($body); if (!$output) { debug_log("Couldn't uncompress data"); push(@DataQueue,{ Headers => { Command => 'FETCH', FilePath => $headers->{'FilePath'}}}); close(FH); unlink($file); return; } if (syswrite(FH,$output) != length($output)) { debug_log("Couldn't write all data to file for some reason"); push(@DataQueue,{ Headers => { Command => 'FETCH', FilePath => $headers->{'FilePath'}}}); close(FH); unlink($file); return; } close(FH); push(@DataQueue,{Headers => $headers, Body => $file}); } else { debug_log("single chunk with no body"); push(@DataQueue,{Headers => $headers}); } return; } else { ## First chunk of a multi chunk message, or bad chunk if ($headers->{Chunk} == 1) { #debug_log("Start of multichunk"); $timer_clear_data_queue->remove; ## Stop processing queue while we are getting the contents of a multi-chunk message. for (keys %{$headers}) { $MSGDATA{$_} = $headers->{$_}; } $MSGDATA{'FILENAME'} = tmpnam(); if (!open(FH,">".$MSGDATA{'FILENAME'})) { debug_log("Error creating temp file"); push(@DataQueue,{ Headers => { Command => 'FETCH', FilePath => $MSGDATA{'FilePath'}}}); %MSGDATA = (); $timer_clear_data_queue->add(1); return; } binmode(FH); my $output = uncompress($body); if (!$output) { debug_log("Couldn't uncompress data"); push(@DataQueue,{ Headers => { Command => 'FETCH', FilePath => $headers->{'FilePath'}}}); close(FH); unlink($MSGDATA{'FILENAME'}); return; } if (syswrite(FH,$output) != length($output)) { debug_log("Couldn't write all data to file for some reason"); push(@DataQueue,{ Headers => { Command => 'FETCH', FilePath => $MSGDATA{'FilePath'}}}); close(FH); unlink($MSGDATA{'FILENAME'}); %MSGDATA = (); $timer_clear_data_queue->add(1); close(FH); return; } close(FH); if ($headers->{LastChunk}) { #debug_log("Got last chunk"); push(@DataQueue,{Headers => $headers, Body => $MSGDATA{'FILENAME'}}); $timer_clear_data_queue->add(1); %MSGDATA = (); return; } } else { ## Bad chunk? Scrap whatever chunked file we might have, request files directly. debug_log("Chunks out of order or missing. "); push(@DataQueue,{ Headers => { Command => 'FETCH', FilePath => $headers->{'FilePath'}}}); push(@DataQueue,{ Headers => { Command => 'FETCH', FilePath => $MSGDATA{'FilePath'}}}) if ($MSGDATA{'FilePath'} && $MSGDATA{'FilePath'} ne $headers->{'FilePath'}); unlink($MSGDATA{'FILENAME'}); %MSGDATA = (); return; } } } sub delete_file { my $file = shift; if (!-d $file && -f $file) { ## File debug_log("attempting to delete $file"); if (!unlink($file)) { debug_log("Couldn't unlink file $file: $!"); } else { debug_log("Deleted $file"); } } else { ## Folder eval {File::Path::rmtree($file);}; if ($@) { debug_log("Couldn't remove file/folder $file: $@"); } else { debug_log("Deleted $file"); } } } sub move_file { my ($file,$moveto) = @_; return if (!-f $file); if (!-d File::Basename::dirname($moveto)) { eval {File::Path::mkpath(File::Basename::dirname($moveto,0755));}; if ($@) { debug_log("error creating file path for $moveto: $!"); fetch_file($moveto,0); return; } } if (!rename($file,$moveto)) { debug_log("error renaming file path for $moveto: $!"); fetch_file($moveto,0); return; } debug_log("$file moved to $moveto"); } sub move_dir { my ($file,$moveto) = @_; return if (!-d $file); return if (-d $moveto); ## this _should not_ happen if (!rename($file,$moveto)) { debug_log("error moving folder: $!"); return; } debug_log("folder $file moved to $moveto"); } sub create_file { my ($file,$filesig,$tmp) = @_; my ($sig,$mtime) = split(/\|/,$filesig); ## Does our temp file's signature match the origin's signature? if (!-e $tmp || (md5_file($tmp) ne $sig)) { debug_log("something weird happened to our tmp file $tmp"); unlink($tmp); fetch_file($file,0); return; } else { ## Copy tmp file over if (!-d File::Basename::dirname($file)) { eval {File::Path::mkpath(File::Basename::dirname($file,0755));}; if ($@) { debug_log("couldn't create path: $!"); unlink($tmp); fetch_file($file,0); return; } } if (!rename($tmp,$file) || !utime($mtime,$mtime,$file)) { debug_log("error renaming or setting mtime: $!"); fetch_file($file,0); return; } debug_log("Successfully received new $file"); } } sub modify_file { my ($file,$filesig,$originfilesig,$tmp) = @_; my ($sig,$mtime) = split(/\|/,$filesig); my ($oldsig,$oldmtime) = split(/\|/,$originfilesig); if ((stat($file))[9] != $oldmtime) { ## No need to check MD5, not the same mtime debug_log("Current mtime " . (stat($file))[9] . " doesn't match on old server file mtime $oldmtime"); unlink($tmp); ## Grab the file fetch_file($file,0); return; } elsif (md5_file($file) ne $oldsig) { debug_log("MD5 doesn't match on old file"); unlink($tmp); ## Grab the file fetch_file($file,0); return; } else { ## Old signatures match, apply the patch. no strict 'subs'; open(FH_ORIG,$file); open(FH_DELTA,$tmp); my $fh_new = new File::Temp(); eval { gdiff_apply(FH_ORIG, FH_DELTA, $fh_new); }; if ($@) { ## Error applying patch close(FH_ORIG); close(FH_DELTA); close($fh_new); unlink($tmp); fetch_file($file,0); return; } ## Patch seems to have worked. Check new checksum. close(FH_ORIG); close(FH_DELTA); close($fh_new); unlink($tmp); if (md5_file($fh_new->filename) ne $sig) { ## MD5 signatures don't match, fetch file debug_log("MD5 doesnt' match on patched file"); fetch_file($file,0); return; } else { ## Signatures do match, rename file debug_log("Successfully received mod $file"); if (!-d File::Basename::dirname($file)) { eval {File::Path::mkpath(File::Basename::dirname($file,0755));}; if ($@) { debug_log("couldn't create path: $!"); unlink($tmp); fetch_file($file,0); return; } } if (!rename($fh_new->filename,$file) || !utime($mtime,$mtime,$file)) { debug_log("error renaming or setting mtime: $!"); fetch_file($file,0); return; } } } } sub fetch_file { my ($file,$retries) = @_; my ($md5,$length,$mtime) = fetch_file_sig($file); if (!$md5 || !$length || !$mtime) { debug_log("Missing info while attempting to retrieve $file"); if ($retries > 10) { debug_log("Too many retries, info missing aborting."); return; } fetch_file($file,++$retries); return; } my $sock = IO::Socket::INET->new(PeerAddr => $FILESERVER_ADDRESS, PeerPort => $FILESERVER_PORT, Proto => "tcp") or fetch_file($file,++$retries); return if (!$sock); my $fh = new File::Temp(SUFFIX=> '.resyncclient'); binmode($fh); syswrite($sock, "GET $file\n",length("GET $file\n")); while(sysread($sock, my $buf, 32768) > 0) { syswrite($fh,$buf,length($buf)); } close($fh); close($sock); if (md5_file($fh->filename) ne $md5) { if ($retries > 10) { debug_log("FAIL: signatures don't match for $file... Too many retries"); debug_log(md5_file($fh->filename) . "!= ($md5)"); return 0; } debug_log("FAIL: signatures didn't match, trying again ($retries)"); debug_log(md5_file($fh->filename) . "!= ($md5)"); fetch_file($file,++$retries); return 0; } if (!-d File::Basename::dirname($file)) { eval {File::Path::mkpath(File::Basename::dirname($file,0755));}; if ($@) { return; } } rename($fh->filename,$file); chmod(0644,$file); utime($mtime,$mtime,$file) || warn "couldn't set utime on $file: $!"; #debug_log("Fetched $file"); } sub fetch_file_sig { my $file = shift; ## First get the signature/filesize/mtime my $sock; for (1..5) { if ($sock = IO::Socket::INET->new(PeerAddr => $FILESERVER_ADDRESS,PeerPort => $FILESERVER_PORT,Proto => "tcp")) { last; } } if (!$sock) { debug_log("Too many connection attempts to retrieve file signature"); return undef; } my $r = syswrite($sock, "SIG $file\n",length("SIG $file\n")); if ($r != length("SIG $file\n")) { ## shouldn't happen, we're blocking debug_log("Partial write for some reason to retrieve file signature"); } my $sig; while(sysread($sock, my $buf, 32768) > 0) { $sig .= $buf; } if ($sig !~ /^([a-zA-Z0-9]{32,32})\|(\d+)\|(\d+)/) { debug_log("Invalid signature received: $sig"); return(undef) } return($1,$2,$3); } sub check_heartbeat { my $e = shift; $e->add($HEARTBEAT_TIMEOUT); if ((time() - $HEARTBEAT) > $HEARTBEAT_TIMEOUT) { debug_log("ACK HEARTBEAT TIMEOUT, RECONNECTING"); Spread::disconnect($mbox); my ($mbox, $private_group) = Spread::connect({ spread_name => $SPREAD_PORT.'@'.$SPREAD_ADDRESS,private_name => "$$-".int(rand(100)) }); if (!$mbox) { crit_log("Couldn't reconnect after heartbeat failure $!"); die; } if (Spread::join($mbox, $Pool) < 0) { crit_log("Couldn't re-join group after heartbeat failure $!"); die; } $event->remove; close(SPREAD_FH); open(SPREAD_FH,"<&=$mbox") || die "couldn't open fd: $!"; $event = event_new(*SPREAD_FH, EV_READ|EV_PERSIST, \&get_spread_messages); $event->add; } } sub show_help { print <<"EOF"; $0 <options> Options: --debug Turn on debugging --pool=<Pool name> Name of server pool --userid=<uid> UserID to change to on start-up --groupid=<gid> Group id to change to on start-up --spread-port=<port> Spread port (usually 4803) --spread-address=<ip> Spread server IP address --fs-port=<port> Port that the file server is running on --fs-address=<ip> IP address of the file server --heartbeat=<seconds> Number of seconds to wait for a heartbeat before reconnecting (default 20) --queue-timer=<seconds> Number of seconds between queue runs (default 5) EOF exit; }

Edit /usr/sbin/FREP_ResyncClient:

#!/usr/bin/perl # FREP_ResyncClient - File replication resynchronization client # # This product uses software developed by Spread Concepts LLC for use in the Spread toolkit. # For more information about Spread see http://www.spread.org # # Copyright (C) 2007 Mark Steele http://www.control-alt-del.org/code # # This program is free software; you can redistribute it and/or # modify it under the terms of the GNU General Public License # as published by the Free Software Foundation; either version 2 # of the License, or (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. # use Spread qw(:ERROR :MESS); use Event::Lib; use strict; use POSIX qw(nice setuid setgid); use Digest::MD5 qw(md5_hex); use Sys::Syslog; use File::Path; use File::Basename; use File::Temp qw(:POSIX); use IO::Socket; use Compress::Zlib; use File::Find; use Getopt::Long; $| = 1; my ($DEBUG,$Pool,$USERID,$GROUPID,$SPREAD_PORT,$SPREAD_ADDRESS,$FILESERVER_PORT,$FILESERVER_ADDRESS, $HEARTBEAT_TIMEOUT,$QUEUE_TIMEOUT); GetOptions ("debug" => \$DEBUG, "pool=s" => \$Pool, "userid|u=i" => \$USERID, "groupid|g=i" => \$GROUPID, "spread-port=i" => \$SPREAD_PORT, "spread-address=s" => \$SPREAD_ADDRESS, "fs-port=i" => \$FILESERVER_PORT, "fs-address=s" => \$FILESERVER_ADDRESS) || show_help(); if (!$Pool || !$USERID || !$GROUPID || !$SPREAD_PORT || !$SPREAD_ADDRESS || !$FILESERVER_PORT || !$FILESERVER_ADDRESS) { show_help(); } if ($SPREAD_ADDRESS !~ /^(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)/ || $FILESERVER_ADDRESS !~ /^(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)/) { print "Invalid IP address\n"; show_help(); } setgid($GROUPID) || die; setuid($USERID) || die; umask(0022) || die; ## Just to be on safe side nice(19); $0 = "FREP_ResyncClient($Pool)"; openlog($0, 'cons,pid', 'user'); my ($mbox, $private_group) = Spread::connect({ spread_name => $SPREAD_PORT.'@'.$SPREAD_ADDRESS,private_name => "$$-".int(rand(100)) }); if (!$mbox) { debug_log("couldn't connect to spread"); exit; } else { debug_log("Connected to spread"); } if (Spread::join($mbox, "Resync-$Pool") < 0) { debug_log("couldn't connect to spread"); } else { debug_log("Joined Resync-$Pool"); } open(SPREAD_FH,"<&=$mbox") || die "couldn't open fd: $!"; my $event = event_new(*SPREAD_FH, EV_READ|EV_PERSIST, \&get_spread_messages); $event->add; my %MSGDATA; event_mainloop(); sub get_spread_messages { while(my ($st, $s, $g, $mt, $e, $mess) = Spread::receive($mbox,0)) { return if (!$mess); my ($headers,$body) = parse_message($mess); process_message($headers,$body) if ($headers); } } sub md5_file { my $file = shift; if (!open(FILE,"$file")) { debug_log("couldn't open file for MD5: $!"); return undef; } binmode(FILE); my $digest = Digest::MD5->new->addfile(*FILE)->hexdigest; close(FILE); return $digest; } sub debug_log { return if (!$DEBUG); syslog('debug', '%s', shift); } sub crit_log { syslog('debug', '%s', shift); } sub parse_message { my $mess = shift; return if ($mess !~ /^Command:/); my ($headers,$body,%headers); if ($mess =~ /^(.*?)__DATA__\n(.+?)__DATAEND__$/s) { ## This is a message that contains a data segment ($headers,$body) = ($1,$2); } else { $headers = $mess; } ## Parse headers my @headers = split(/\n/,$headers); for (@headers) { my ($key,$val) = split(/:/,$_); $headers{$key} = $val; } return(\%headers,$body); } sub process_message { my ($headers,$body) = @_; if (defined($MSGDATA{'Cookie'})) { if ($headers->{'Cookie'} != $MSGDATA{'Cookie'}) { debug_log("Cookie doesn't match."); unlink($MSGDATA{'FILENAME'}); %MSGDATA = (); return; } if ($MSGDATA{'Chunk'} !~ /^(\d+)$/) { ## Odd, chunks doesn't contain what we expect. Abort. debug_log("Chunks seem messed up, couldn't parse current chunk data"); unlink($MSGDATA{'FILENAME'}); %MSGDATA = (); return; } my $curchunk = $1; if ($headers->{Chunk} !~ /^(\d+)$/) { ## Odd, chunks doens't contain what we expect. Abort. debug_log("Chunks weird. Last chunk processed: $curchunk. This chunk couldn't be parsed"); unlink($MSGDATA{'FILENAME'}); %MSGDATA = (); return; } my $newchunk = $1; if ($newchunk != ($curchunk + 1)) { ## Out of order. debug_log("Chunks out of order or missing."); unlink($MSGDATA{'FILENAME'}); %MSGDATA = (); return; } ## Step 3. Write data to tmpfile. if (!open(FH,">>".$MSGDATA{'FILENAME'})) { debug_log("Error opening temp file: $!"); %MSGDATA = (); return; } binmode(FH); my $output = uncompress($body); if (!$output) { debug_log("Couldn't uncompress data"); close(FH); unlink($MSGDATA{'FILENAME'}); %MSGDATA = (); return; } if (syswrite(FH,$output) != length($output)) { debug_log("Couldn't write all data to file for some reason"); close(FH); unlink($MSGDATA{'FILENAME'}); %MSGDATA = (); return; } close(FH); ## Step 4. Last chunk? if ($headers->{LastChunk}) { process_resync($MSGDATA{'FILENAME'},$headers->{FileSig}); %MSGDATA = (); return; } else { ## Update chunk count $MSGDATA{'Chunk'} = $headers->{'Chunk'}; } return; } elsif (!$headers->{Chunk}) { ## Single chunk message if ($body) { my $file = tmpnam() . ".resyncclient"; if (!open(FH,">$file")) { debug_log("Ack couldn't open tmp file: $!"); return; } binmode(FH); my $output = uncompress($body); if (!$output) { debug_log("Couldn't uncompress data"); close(FH); unlink($file); return; } if (syswrite(FH,$output) != length($output)) { debug_log("Couldn't write all data to file for some reason"); close(FH); unlink($file); return; } close(FH); process_resync($file,$headers->{FileSig}); } return; } else { ## First chunk of a multi chunk message, or bad chunk if ($headers->{Chunk} == 1) { for (keys %{$headers}) { $MSGDATA{$_} = $headers->{$_}; } $MSGDATA{'FILENAME'} = tmpnam() . ".resyncclient"; if (!open(FH,">".$MSGDATA{'FILENAME'})) { debug_log("Error creating temp file"); %MSGDATA = (); return; } binmode(FH); my $output = uncompress($body); if (!$output) { debug_log("Couldn't uncompress data"); close(FH); unlink($MSGDATA{'FILENAME'}); %MSGDATA = (); return; } if (syswrite(FH,$output) != length($output)) { debug_log("Couldn't write all data to file for some reason"); close(FH); unlink($MSGDATA{'FILENAME'}); %MSGDATA = (); return; } close(FH); if ($headers->{LastChunk}) { process_resync($MSGDATA{'FILENAME'},$headers->{FileSig}); %MSGDATA = (); return; } } else { ## Bad chunk? Scrap whatever chunked file we might have, request files directly. unlink($MSGDATA{'FILENAME'}); %MSGDATA = (); return; } } } sub fetch_file { my ($file,$retries) = @_; if ($retries > 10) { crit_log("Too many retries retrieving $file, info missing aborting."); return; } my ($md5,$length,$mtime) = fetch_file_sig($file); if (!$md5 || !$length || !$mtime) { debug_log("Missing info while attempting to retrieve $file"); fetch_file($file,++$retries); return; } my $sock = IO::Socket::INET->new(PeerAddr => $FILESERVER_ADDRESS, PeerPort => $FILESERVER_PORT, Proto => "tcp"); if (!$sock) { crit_log("Failed to connect to fileserver: $!"); fetch_file($file,++$retries); return; } my $fh = new File::Temp(SUFFIX=> '.resyncclient'); binmode($fh); syswrite($sock, "GET $file\n",length("GET $file\n")); while(sysread($sock, my $buf, 32768) > 0) { syswrite($fh,$buf,length($buf)); } close($fh); close($sock); if (md5_file($fh->filename) ne $md5) { if ($retries > 10) { crit_log("FAIL: signatures don't match for $file... Too many retries"); return; } debug_log("FAIL: signatures didn't match, trying again ($retries)"); fetch_file($file,++$retries); return; } if (!-d File::Basename::dirname($file)) { eval {File::Path::mkpath(File::Basename::dirname($file,0755));}; if ($@) { crit_log("couldn't create path for $file: $!"); return; } } rename($fh->filename,$file); chmod(0644,$file); utime($mtime,$mtime,$file) || warn "couldn't set utime on $file: $!"; debug_log("Fetched $file"); } sub fetch_file_sig { my $file = shift; ## First get the signature/filesize/mtime my $sock; for (1..5) { if ($sock = IO::Socket::INET->new(PeerAddr => $FILESERVER_ADDRESS,PeerPort => $FILESERVER_PORT,Proto => "tcp")) { last; } } if (!$sock) { crit_log("Too many connection attempts to retrieve file signature"); return undef; } my $r = syswrite($sock, "SIG $file\n",length("SIG $file\n")); if ($r != length("SIG $file\n")) { debug_log("Partial write for some reason to retrieve file signature"); ## shouldn't happen, this socket is blocking return undef; } my $sig; while(sysread($sock, my $buf, 32768) > 0) { $sig .= $buf; } if ($sig !~ /^([a-zA-Z0-9]{32,32})\|(\d+)\|(\d+)/) { debug_log("Invalid signature received: $sig"); return(undef) } return($1,$2,$3); } sub process_resync { my ($file,$sig) = @_; debug_log("Starting resync"); if (md5_file($file) ne $sig) { debug_log("File ($file) sig ($sig) doesn't match: " . md5_file($file)); return; } my @fetch; ## Read tempfile and start comparing local filesystem. if (!open(F,"<$file")) { debug_log("Couldn't open tempfile $file: $!"); unlink($file) if (-e $file); return; } my ($CURDIR,%DIRFILES); while(<F>) { chomp; if (/^DIR:(.+)$/) { my $newdir = $1; #debug_log("Examining dir $newdir"); if ($CURDIR) { ## We should have something in %DIRFILES. Compare with current folder. if (scalar(keys %DIRFILES) && opendir(DIR,$CURDIR)) { while(my $itm = readdir(DIR)) { next if ($itm =~ /^\.{1,2}$/ || -d "$CURDIR/$itm"); if (!defined($DIRFILES{$itm})) { ## We've got a file locally that the server didn't. Remove local file. debug_log("Deleting $CURDIR/$itm"); unlink("$CURDIR/$itm"); } } closedir(DIR); } } $CURDIR = $newdir; File::Find::finddepth(sub{ rmdir },$CURDIR); ## Remove empty folders %DIRFILES = (); } else { ## This is supposed to be a file, and $CURDIR is supposed to be defined. if (!$CURDIR) { debug_log("ACK NO CURDIR: >>$_<<"); return; } my ($file,$size,$mtime) = split(/:/); $DIRFILES{$file} = 1; if (-s "$CURDIR/$file" != $size || (stat(_))[9] != $mtime) { ## File isn't there, or size doesn't match, or mtime doesn't match. Get the file. push(@fetch,"$CURDIR/$file"); } } } close(F); unlink($file); if (!$CURDIR) { crit_log("ACK NO CURDIR: $CURDIR after traversal"); return; } ## We should have something in %DIRFILES. Compare with current folder. if (opendir(DIR,"$CURDIR")) { while(my $itm = readdir(DIR)) { next if ($itm =~ /^\.{1,2}$/ || -d "$CURDIR/$itm"); if (!defined($DIRFILES{$itm})) { ## We've got a file locally that the server didn't. Remove local file. debug_log("Deleting $CURDIR/$itm"); unlink("$CURDIR/$itm"); } } closedir(DIR); } %DIRFILES = (); File::Find::finddepth(sub{ rmdir },$CURDIR); ## Will remove empty folders for (@fetch) { debug_log("fetching $_ for resync"); fetch_file($_,0); } unlink(</tmp/*.resyncclient>); crit_log("Resync finished (".scalar @fetch.")"); } sub show_help { print <<"EOF"; $0 <options> Options: --debug Turn on debugging --pool=<Pool name> Name of server pool --userid=<uid> UserID to change to on start-up --groupid=<gid> Group id to change to on start-up --spread-port=<port> Spread port (usually 4803) --spread-address=<ip> Spread server IP address --fs-port=<port> Port that the file server is running on --fs-address=<ip> IP address of the file server EOF exit; } __END__

Edit /usr/sbin/FREP_Mon:

#!/usr/bin/perl # FREP_Mon - Monitoring client for FREP # # This product uses software developed by Spread Concepts LLC for use in the Spread toolkit. # For more information about Spread see http://www.spread.org # # Copyright (C) 2007 Mark Steele http://www.control-alt-del.org/code # # This program is free software; you can redistribute it and/or # modify it under the terms of the GNU General Public License # as published by the Free Software Foundation; either version 2 # of the License, or (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. # use Spread qw(:ERROR :MESS); use Event::Lib; use strict; use Getopt::Long; $| = 1; my ($POOL,$SPREAD_SERVER,$SPREAD_PORT); GetOptions ("pool=s" => \$POOL, "spread-port=i" => \$SPREAD_PORT, "spread-address=s" =>\$SPREAD_SERVER) || show_help(); if (!$SPREAD_SERVER || !$POOL || !$SPREAD_PORT) { show_help(); } my ($mbox, $private_group) = Spread::connect({ spread_name => $SPREAD_PORT.'@'.$SPREAD_SERVER,private_name => "$$-".int(rand(100)) }); if (!$mbox) { die $!; } if (Spread::join($mbox, $POOL) < 0) { die; } open(SPREAD_FH,"<&=$mbox") || die "couldn't open fd: $!"; my $event = event_new(*SPREAD_FH, EV_READ|EV_PERSIST, \&get_spread_messages); $event->add; event_mainloop(); sub get_spread_messages { while(my ($st, $s, $g, $mt, $e, $mess) = Spread::receive($mbox,0)) { return if (!$mess); my ($headers,$body) = parse_message($mess); process_message($headers,$body) if ($headers); } } sub parse_message { my $mess = shift; return if ($mess !~ /^Command:/); my ($headers,$body,%headers); if ($mess =~ /^(.*?)__DATA__\n(.+?)__DATAEND__$/s) { ## This is a message that contains a data segment ($headers,$body) = ($1,$2); } else { $headers = $mess; } ## Parse headers my @headers = split(/\n/,$headers); for (@headers) { my ($key,$val) = split(/:/,$_); $headers{$key} = $val; } return(\%headers,$body); } sub process_message { my ($headers,$body) = @_; ## If this is a heartbeat, update time stamp. if ($headers->{Command} eq 'HEARTBEAT') { debug_log("Recieved heartbeat"); return; } if ($headers->{'Chunk'} > 1) { return; } debug_log("CMD: $headers->{Command} FP: $headers->{FilePath}"); } sub debug_log { print scalar localtime, " ",shift,"\n"; } sub show_help { print <<"EOF"; Usage: $0 --pool=<Pool name> --spread-address=<spread server ip> --spread-port=<spread server port> EOF exit; }

Make the files executable:

chmod +x /usr/sbin/FREP_Client
chmod +x /usr/sbin/FREP_ResyncClient
chmod +x /usr/sbin/FREP_Mon

Configuration options for FREP_Client:

Turn on debugging: --debug
Name of server pool: --pool=<Pool name>

UserID to run as: --userid=<uid>

GroupID run as: --groupid=<gid>
Spread port (usually 4803): --spread-port=<port>
Spread server IP address: --spread-address=<ip>
Port that the file server is running on: --fs-port=<port>
IP address of the file server: --fs-address=<ip>
Heartbeat timer (default 20): --heartbeat=<seconds>
Queue timer (default 5): --queue-timer=<seconds>

Configuration options for FREP_ResyncClient:

Turn on debugging: --debug
Name of server pool: --pool=<Pool name>

UserID to run as: --userid=<uid>

GroupID run as: --groupid=<gid>
Spread port (usually 4803): --spread-port=<port>
Spread server IP address: --spread-address=<ip>
Port that the file server is running on: --fs-port=<port>
IP address of the file server: --fs-address=<ip>

Configuration options for FREP_Mon:

Name of server pool: --pool=<Pool name>

Spread port (usually 4803): --spread-port=<port>
Spread server IP address: --spread-address=<ip>


Note: The userid and groupid's should match on the FREP_Client and FREP_ResyncClient.

Start the services

cd /service
ln -s /etc/FREP_Client
ln -s /etc/FREP_ResyncClient

Wait a few seconds, daemontools should fire them all up. Check to make sure they are running with:

svstat /service/FREP*

Some of the things I'll get around to eventually:

• Add flag in client to specify temporary folder (rename()'s across devices will fail, see bugs below).
• More testing maybe.
  

The beauty of using Spread as a messaging toolkit is it's ability to use IP broadcast or mutlicast for sending out it's messages.

If you've got a modest number of nodes, I'd recommend making every client node part of the Spread network (and have the client connect to localhost).

For larger setups, you probably want to configure one Spread server every 20 clients or so.

In theory, this solution should scale to hundreds or thousands of clients since we can configure the Spread network to multicast on several VLANs. It may also even be possible to use this replication mechanism across WAN links by tunnelling multicast traffic.

The Spread Toolkit

Generic diff format specification

Algorithm::GdiffDelta perl module

Zlib compression library

Libevent

This system works well with relatively small file sizes, I haven't done very extensive tests on large files.

I am almost 100% sure that trying to keep large (>1 meg) files synched up won't be a problem UNLESS you are modifying them frequently.

There is a 2 gig limitation on the perl Algorithm::GdiffDelta module. Also the bigger the file, the longer it takes to calculate the differences when the files are modified.

The environment I've deployed FREP in involves many (300,000+ files/folders) that are changed at a rate of about 2 files per second. The average file size is probably around 15kb.

I haven't tested this out with Spread 4.X, your mileage may vary.

There's a bug in the client code, I don't check if the folder I'm synching to is on the same device as the system's default temporary folder path (/tmp on linux). This means that if your files are on a different device, the rename() call will fail, and nothing will work. I'll fix this eventually if anyone asks.