rsync server now creates user objects
[cloudstore.git] / rsync-piper.pl
1 #!/usr/bin/perl
2 use warnings;
3 use strict;
4
5 use autodie;
6 use POSIX;
7 use File::Slurp;
8 use IO::Select;
9 use Time::HiRes;
10 use Data::Dump qw(dump);
11 use English;
12 use Module::Refresh;
13
14 use lib 'lib';
15 use CloudStore::Couchbase;
16
17 my $dir   = $ENV{RSYNC_DIR}  || '/srv/cloudstore';
18 my $port  = $ENV{RSYNC_PORT} || 6501;
19 my $users = "users";
20
21 my $log_fifo = "$dir/var/$port.log";
22 my $pid_file = "$dir/var/$port.pid";
23 my $cfg_file = "$dir/var/$port.conf";
24
25 my $rsync = 'rsync';
26 $rsync = 'bin/rsync' if -x 'bin/rsync'; # use 3.1dev version!
27
28 my @transfer = qw(
29 timestamp:%t:timestamp
30 login:%m:text
31 port:$port:int
32 auth:%u:text
33 host:%h:text
34 pid:%p:int
35 perms:%B:text
36 itemize:%i:text
37 mtime:%M:timestamp
38 md5:%C:text
39 op:%o:text
40 size:%l:int
41 transfered:%b:int
42 file:%f:text
43 );
44
45 $transfer[2] = "port:$port:int"; # expand $port
46
47 my @transfer_names =          map { ( split(/:/,$_,3) )[0] } @transfer;
48 my $transfer_log   = join('|',map { ( split(/:/,$_,3) )[1] } @transfer );
49
50 if ( $ENV{SQL} ) {
51         print "CREATE TABLE rsync_transfer (\n\t",
52         join(",\n\t", map { my @m = split(/:/,$_,3); "$m[0] $m[2]" } @transfer),
53         "\n);\n";
54         exit 1;
55 }
56
57 my $store = CloudStore::Couchbase->new;
58
59 mkdir "$dir/var" if ! -e "$dir/var";
60
61 unlink $log_fifo if -f $log_fifo;
62 mkfifo $log_fifo, 0700 unless -p $log_fifo;
63
64 my $rsync_config = qq{
65
66 #uid = nobody
67 #gid = nogroup
68 #use chroot = yes
69 use chroot = no
70
71 #max connections = 4
72 lock file = $dir/var/$port.lock
73
74 #syslog facility = local5
75 log file  = $log_fifo
76
77 transfer logging = yes
78 log format = transfer-log:$transfer_log
79 max verbosity = 5
80
81 pid file  = $pid_file
82
83 # don't check secrets file permission (uid)
84 strict modes = no
85
86 #pre-xfer exec = /srv/cloudstore/pre-xfer.sh
87 #post-xfer exec = /srv/cloudstore/post-xfer.sh
88
89 };
90
91 foreach my $path ( glob "$users/*" ) {
92
93         my $login = $path;
94         $login =~ s{^.+/([^/]+)$}{$1}; 
95
96         if ( -d $path && -d "$path/blob" && -f "$path/secrets" ) {
97                 my @secrets = map { chomp; $_ } read_file "$path/secrets";
98                 my $auth_users = join(', ', map { s/:.+$//; $_ } @secrets );
99
100                 $rsync_config .= <<__RSYNC_MODULE__;
101
102 [$login]
103         path = $dir/users/$login/blob
104         auth users = $auth_users
105         secrets file = $dir/users/$login/secrets
106         read only = false
107
108 __RSYNC_MODULE__
109
110                 print "INFO: added $login = $auth_users\n";
111
112                 my $quota = read_file "$path/quota" if -e "$path/quota";
113                 chomp $quota && $quota * 1;
114                 $quota ||= 100 * 1024; # Kb - XXX default quota
115
116                 $store->user_set({
117                         login => $login,
118                         path => $path,
119                         secrets => [ @secrets ],
120                         quota => $quota,
121                         port => $port,
122                 });
123
124         } else {
125                 warn "skipped $login: $!";
126         }
127
128 }
129
130 write_file $cfg_file, $rsync_config;
131 warn "created $cfg_file ", -s $cfg_file, " bytes\n";
132
133 if ( -e $pid_file ) {
134         my $pid = read_file $pid_file;
135         chomp($pid);
136         if ( kill 0, $pid ) {
137                 warn "found rsync pid $pid\n";
138                 kill 2, $pid;
139                 while ( -e $pid_file ) {
140                         warn "waiting for rsync to die...\n";
141                         sleep 1;
142                 }
143                 
144                 open(my $fifo, '<', $log_fifo);
145                 while ( kill 0, $pid ) {
146                         my $line = <$fifo>;
147                         warn "<<< $line\n";
148                 }
149
150                 kill 0, $pid && die "can't kill it!";
151         } else {
152                 unlink $pid_file;
153         }
154 }
155
156 use POSIX ":sys_wait_h";
157 sub REAPER {
158         my $child;
159         while ((my $waitedpid = waitpid(-1,WNOHANG)) > 0) {
160                 warn "reaped $waitedpid" . ($? ? " with exit $?" : '');
161         }
162         $SIG{CHLD} = \&REAPER;  # loathe SysV
163 }
164
165 $SIG{CHLD} = \&REAPER;
166
167
168 if ( ! -e $pid_file ) {
169         my $exec = "$rsync --daemon --config $cfg_file --no-detach --port=$port";
170         warn "START $exec\n";
171
172         die "could not fork\n" unless defined(my $pid = fork);
173         unless ($pid) {
174                 warn "start server with $exec\n";
175                 exec $exec || die $!;
176         }
177
178         warn "wait for pid file";
179         while ( ! -e $pid_file ) {
180                 sleep 1;
181         }
182 }
183
184 =for gearman
185 use Gearman::Client;
186 my $gearman = Gearman::Client->new;
187 $gearman->job_servers('127.0.0.1:4730');
188 =cut
189
190 while(1) {
191         warn "# reading log output from $log_fifo\n";
192         open(my $fifo, '<', $log_fifo);
193         while( my $line = <$fifo> ) {
194                 Module::Refresh->refresh;
195                 die "ERROR: $line" if $line =~ /rsync error:/;
196                 chomp $line;
197                 print $line, $/;
198
199                 if ( $line =~ /transfer-log:(.+)/ ) {
200                         my $transfer = $1;
201                         $transfer =~ s|(\d\d\d\d)/(\d\d)/(\d\d)[-\s](\d\d:\d\d:\d\d)|$1-$2-$3T$4|g;
202                         my ( $yyyy,$mm,$dd,undef,$login,undef ) = split( /[\-T\|]/, $transfer, 6 );
203                         my $host = $1 if $login =~ s/\+(.+)//;
204
205                         my $path = "users/$login/log";
206                         mkdir $path unless -d $path;
207                         $path .= "/$yyyy-$mm-$dd";
208                         my $new_log = ! -e $path;
209                         open( my $log, '>>', $path );
210                         print $log join('|',@transfer),"\n" if $new_log; # store header
211                         print $log "$transfer\n";
212                         close $log;
213
214                         my @v = split(/\|/,$transfer,$#transfer + 1);
215                         my %data;
216                         @data{@transfer_names} = @v ; # FIXME validate?
217
218                         print ">>> data ",dump( \%data ) if $ENV{DEBUG};
219
220                         $store->transfer( \%data );
221 =for gearman
222                         $gearman->dispatch_background( 'rsync_transfer' => $json );
223 =cut
224
225                 }
226         }
227         close($fifo);
228         sleep 1;
229 }
230