]> git.proxmox.com Git - pve-zsync.git/blob - pve-zsync
Improve error handling in parse_disk.
[pve-zsync.git] / pve-zsync
1 #!/usr/bin/perl
2
3 use strict;
4 use warnings;
5 use Data::Dumper qw(Dumper);
6 use Fcntl qw(:flock SEEK_END);
7 use Getopt::Long qw(GetOptionsFromArray);
8 use File::Copy qw(move);
9 use File::Path qw(make_path);
10 use JSON;
11 use IO::File;
12 use String::ShellQuote 'shell_quote';
13
14 my $PROGNAME = "pve-zsync";
15 my $CONFIG_PATH = "/var/lib/${PROGNAME}";
16 my $STATE = "${CONFIG_PATH}/sync_state";
17 my $CRONJOBS = "/etc/cron.d/$PROGNAME";
18 my $PATH = "/usr/sbin";
19 my $PVE_DIR = "/etc/pve/local";
20 my $QEMU_CONF = "${PVE_DIR}/qemu-server";
21 my $LXC_CONF = "${PVE_DIR}/lxc";
22 my $LOCKFILE = "$CONFIG_PATH/${PROGNAME}.lock";
23 my $PROG_PATH = "$PATH/${PROGNAME}";
24 my $INTERVAL = 15;
25 my $DEBUG = 0;
26
27 my $IPV4OCTET = "(?:25[0-5]|(?:[1-9]|1[0-9]|2[0-4])?[0-9])";
28 my $IPV4RE = "(?:(?:$IPV4OCTET\\.){3}$IPV4OCTET)";
29 my $IPV6H16 = "(?:[0-9a-fA-F]{1,4})";
30 my $IPV6LS32 = "(?:(?:$IPV4RE|$IPV6H16:$IPV6H16))";
31
32 my $IPV6RE = "(?:" .
33 "(?:(?:" . "(?:$IPV6H16:){6})$IPV6LS32)|" .
34 "(?:(?:" . "::(?:$IPV6H16:){5})$IPV6LS32)|" .
35 "(?:(?:(?:" . "$IPV6H16)?::(?:$IPV6H16:){4})$IPV6LS32)|" .
36 "(?:(?:(?:(?:$IPV6H16:){0,1}$IPV6H16)?::(?:$IPV6H16:){3})$IPV6LS32)|" .
37 "(?:(?:(?:(?:$IPV6H16:){0,2}$IPV6H16)?::(?:$IPV6H16:){2})$IPV6LS32)|" .
38 "(?:(?:(?:(?:$IPV6H16:){0,3}$IPV6H16)?::(?:$IPV6H16:){1})$IPV6LS32)|" .
39 "(?:(?:(?:(?:$IPV6H16:){0,4}$IPV6H16)?::" . ")$IPV6LS32)|" .
40 "(?:(?:(?:(?:$IPV6H16:){0,5}$IPV6H16)?::" . ")$IPV6H16)|" .
41 "(?:(?:(?:(?:$IPV6H16:){0,6}$IPV6H16)?::" . ")))";
42
43 my $HOSTv4RE0 = "(?:[\\w\\.\\-_]+|$IPV4RE)"; # hostname or ipv4 address
44 my $HOSTv4RE1 = "(?:$HOSTv4RE0|\\[$HOSTv4RE0\\])"; # these may be in brackets, too
45 my $HOSTRE = "(?:$HOSTv4RE1|\\[$IPV6RE\\])"; # ipv6 must always be in brackets
46 # targets are either a VMID, or a 'host:zpool/path' with 'host:' being optional
47 my $TARGETRE = qr!^(?:($HOSTRE):)?(\d+|(?:[\w\-_]+)(/.+)?)$!;
48
49 check_bin ('cstream');
50 check_bin ('zfs');
51 check_bin ('ssh');
52 check_bin ('scp');
53
54 $SIG{TERM} = $SIG{QUIT} = $SIG{PIPE} = $SIG{HUP} = $SIG{KILL} = $SIG{INT} =
55 sub {
56 die "Signal aborting sync\n";
57 };
58
59 sub check_bin {
60 my ($bin) = @_;
61
62 foreach my $p (split (/:/, $ENV{PATH})) {
63 my $fn = "$p/$bin";
64 if (-x $fn) {
65 return $fn;
66 }
67 }
68
69 die "unable to find command '$bin'\n";
70 }
71
72 sub cut_target_width {
73 my ($path, $maxlen) = @_;
74 $path =~ s@/+@/@g;
75
76 return $path if length($path) <= $maxlen;
77
78 return '..'.substr($path, -$maxlen+2) if $path !~ m@/@;
79
80 $path =~ s@/([^/]+/?)$@@;
81 my $tail = $1;
82
83 if (length($tail)+3 == $maxlen) {
84 return "../$tail";
85 } elsif (length($tail)+2 >= $maxlen) {
86 return '..'.substr($tail, -$maxlen+2)
87 }
88
89 $path =~ s@(/[^/]+)(?:/|$)@@;
90 my $head = $1;
91 my $both = length($head) + length($tail);
92 my $remaining = $maxlen-$both-4; # -4 for "/../"
93
94 if ($remaining < 0) {
95 return substr($head, 0, $maxlen - length($tail) - 3) . "../$tail"; # -3 for "../"
96 }
97
98 substr($path, ($remaining/2), (length($path)-$remaining), '..');
99 return "$head/" . $path . "/$tail";
100 }
101
102 sub lock {
103 my ($fh) = @_;
104 flock($fh, LOCK_EX) || die "Can't lock config - $!\n";
105 }
106
107 sub unlock {
108 my ($fh) = @_;
109 flock($fh, LOCK_UN) || die "Can't unlock config- $!\n";
110 }
111
112 sub get_status {
113 my ($source, $name, $status) = @_;
114
115 if ($status->{$source->{all}}->{$name}->{status}) {
116 return $status;
117 }
118
119 return undef;
120 }
121
122 sub check_pool_exists {
123 my ($target) = @_;
124
125 my $cmd = [];
126
127 if ($target->{ip}) {
128 push @$cmd, 'ssh', "root\@$target->{ip}", '--';
129 }
130 push @$cmd, 'zfs', 'list', '-H', '--', $target->{all};
131 eval {
132 run_cmd($cmd);
133 };
134
135 if ($@) {
136 return 0;
137 }
138 return 1;
139 }
140
141 sub parse_target {
142 my ($text) = @_;
143
144 my $errstr = "$text : is not a valid input! Use [IP:]<VMID> or [IP:]<ZFSPool>[/Path]";
145 my $target = {};
146
147 if ($text !~ $TARGETRE) {
148 die "$errstr\n";
149 }
150 $target->{all} = $2;
151 $target->{ip} = $1 if $1;
152 my @parts = split('/', $2);
153
154 $target->{ip} =~ s/^\[(.*)\]$/$1/ if $target->{ip};
155
156 my $pool = $target->{pool} = shift(@parts);
157 die "$errstr\n" if !$pool;
158
159 if ($pool =~ m/^\d+$/) {
160 $target->{vmid} = $pool;
161 delete $target->{pool};
162 }
163
164 return $target if (@parts == 0);
165 $target->{last_part} = pop(@parts);
166
167 if ($target->{ip}) {
168 pop(@parts);
169 }
170 if (@parts > 0) {
171 $target->{path} = join('/', @parts);
172 }
173
174 return $target;
175 }
176
177 sub read_cron {
178
179 #This is for the first use to init file;
180 if (!-e $CRONJOBS) {
181 my $new_fh = IO::File->new("> $CRONJOBS");
182 die "Could not create $CRONJOBS: $!\n" if !$new_fh;
183 close($new_fh);
184 return undef;
185 }
186
187 my $fh = IO::File->new("< $CRONJOBS");
188 die "Could not open file $CRONJOBS: $!\n" if !$fh;
189
190 my @text = <$fh>;
191
192 close($fh);
193
194 return encode_cron(@text);
195 }
196
197 sub parse_argv {
198 my (@arg) = @_;
199
200 my $param = {};
201 $param->{dest} = undef;
202 $param->{source} = undef;
203 $param->{verbose} = undef;
204 $param->{limit} = undef;
205 $param->{maxsnap} = undef;
206 $param->{name} = undef;
207 $param->{skip} = undef;
208 $param->{method} = undef;
209
210 my ($ret, $ar) = GetOptionsFromArray(\@arg,
211 'dest=s' => \$param->{dest},
212 'source=s' => \$param->{source},
213 'verbose' => \$param->{verbose},
214 'limit=i' => \$param->{limit},
215 'maxsnap=i' => \$param->{maxsnap},
216 'name=s' => \$param->{name},
217 'skip' => \$param->{skip},
218 'method=s' => \$param->{method});
219
220 if ($ret == 0) {
221 die "can't parse options\n";
222 }
223
224 $param->{name} = "default" if !$param->{name};
225 $param->{maxsnap} = 1 if !$param->{maxsnap};
226 $param->{method} = "ssh" if !$param->{method};
227
228 return $param;
229 }
230
231 sub add_state_to_job {
232 my ($job) = @_;
233
234 my $states = read_state();
235 my $state = $states->{$job->{source}}->{$job->{name}};
236
237 $job->{state} = $state->{state};
238 $job->{lsync} = $state->{lsync};
239 $job->{vm_type} = $state->{vm_type};
240
241 for (my $i = 0; $state->{"snap$i"}; $i++) {
242 $job->{"snap$i"} = $state->{"snap$i"};
243 }
244
245 return $job;
246 }
247
248 sub encode_cron {
249 my (@text) = @_;
250
251 my $cfg = {};
252
253 while (my $line = shift(@text)) {
254
255 my @arg = split('\s', $line);
256 my $param = parse_argv(@arg);
257
258 if ($param->{source} && $param->{dest}) {
259 $cfg->{$param->{source}}->{$param->{name}}->{dest} = $param->{dest};
260 $cfg->{$param->{source}}->{$param->{name}}->{verbose} = $param->{verbose};
261 $cfg->{$param->{source}}->{$param->{name}}->{limit} = $param->{limit};
262 $cfg->{$param->{source}}->{$param->{name}}->{maxsnap} = $param->{maxsnap};
263 $cfg->{$param->{source}}->{$param->{name}}->{skip} = $param->{skip};
264 $cfg->{$param->{source}}->{$param->{name}}->{method} = $param->{method};
265 }
266 }
267
268 return $cfg;
269 }
270
271 sub param_to_job {
272 my ($param) = @_;
273
274 my $job = {};
275
276 my $source = parse_target($param->{source});
277 my $dest = parse_target($param->{dest}) if $param->{dest};
278
279 $job->{name} = !$param->{name} ? "default" : $param->{name};
280 $job->{dest} = $param->{dest} if $param->{dest};
281 $job->{method} = "local" if !$dest->{ip} && !$source->{ip};
282 $job->{method} = "ssh" if !$job->{method};
283 $job->{limit} = $param->{limit};
284 $job->{maxsnap} = $param->{maxsnap} if $param->{maxsnap};
285 $job->{source} = $param->{source};
286
287 return $job;
288 }
289
290 sub read_state {
291
292 if (!-e $STATE) {
293 make_path $CONFIG_PATH;
294 my $new_fh = IO::File->new("> $STATE");
295 die "Could not create $STATE: $!\n" if !$new_fh;
296 print $new_fh "{}";
297 close($new_fh);
298 return undef;
299 }
300
301 my $fh = IO::File->new("< $STATE");
302 die "Could not open file $STATE: $!\n" if !$fh;
303
304 my $text = <$fh>;
305 my $states = decode_json($text);
306
307 close($fh);
308
309 return $states;
310 }
311
312 sub update_state {
313 my ($job) = @_;
314 my $text;
315 my $in_fh;
316
317 eval {
318
319 $in_fh = IO::File->new("< $STATE");
320 die "Could not open file $STATE: $!\n" if !$in_fh;
321 lock($in_fh);
322 $text = <$in_fh>;
323 };
324
325 my $out_fh = IO::File->new("> $STATE.new");
326 die "Could not open file ${STATE}.new: $!\n" if !$out_fh;
327
328 my $states = {};
329 my $state = {};
330 if ($text){
331 $states = decode_json($text);
332 $state = $states->{$job->{source}}->{$job->{name}};
333 }
334
335 if ($job->{state} ne "del") {
336 $state->{state} = $job->{state};
337 $state->{lsync} = $job->{lsync};
338 $state->{vm_type} = $job->{vm_type};
339
340 for (my $i = 0; $job->{"snap$i"} ; $i++) {
341 $state->{"snap$i"} = $job->{"snap$i"};
342 }
343 $states->{$job->{source}}->{$job->{name}} = $state;
344 } else {
345
346 delete $states->{$job->{source}}->{$job->{name}};
347 delete $states->{$job->{source}} if !keys %{$states->{$job->{source}}};
348 }
349
350 $text = encode_json($states);
351 print $out_fh $text;
352
353 close($out_fh);
354 move("$STATE.new", $STATE);
355 eval {
356 close($in_fh);
357 };
358
359 return $states;
360 }
361
362 sub update_cron {
363 my ($job) = @_;
364
365 my $updated;
366 my $has_header;
367 my $line_no = 0;
368 my $text = "";
369 my $header = "SHELL=/bin/sh\n";
370 $header .= "PATH=/usr/local/sbin:/usr/local/bin:/sbin:/bin:/usr/sbin:/usr/bin\n\n";
371
372 my $fh = IO::File->new("< $CRONJOBS");
373 die "Could not open file $CRONJOBS: $!\n" if !$fh;
374 lock($fh);
375
376 my @test = <$fh>;
377
378 while (my $line = shift(@test)) {
379 chomp($line);
380 if ($line =~ m/source $job->{source} .*name $job->{name} /) {
381 $updated = 1;
382 next if $job->{state} eq "del";
383 $text .= format_job($job, $line);
384 } else {
385 if (($line_no < 3) && ($line =~ /^(PATH|SHELL)/ )) {
386 $has_header = 1;
387 }
388 $text .= "$line\n";
389 }
390 $line_no++;
391 }
392
393 if (!$has_header) {
394 $text = "$header$text";
395 }
396
397 if (!$updated) {
398 $text .= format_job($job);
399 }
400 my $new_fh = IO::File->new("> ${CRONJOBS}.new");
401 die "Could not open file ${CRONJOBS}.new: $!\n" if !$new_fh;
402
403 die "can't write to $CRONJOBS.new\n" if !print($new_fh $text);
404 close ($new_fh);
405
406 die "can't move $CRONJOBS.new: $!\n" if !move("${CRONJOBS}.new", "$CRONJOBS");
407 close ($fh);
408 }
409
410 sub format_job {
411 my ($job, $line) = @_;
412 my $text = "";
413
414 if ($job->{state} eq "stopped") {
415 $text = "#";
416 }
417 if ($line) {
418 $line =~ /^#*(.+) root/;
419 $text .= $1;
420 } else {
421 $text .= "*/$INTERVAL * * * *";
422 }
423 $text .= " root";
424 $text .= " $PROGNAME sync --source $job->{source} --dest $job->{dest}";
425 $text .= " --name $job->{name} --maxsnap $job->{maxsnap}";
426 $text .= " --limit $job->{limit}" if $job->{limit};
427 $text .= " --method $job->{method}";
428 $text .= " --verbose" if $job->{verbose};
429 $text .= "\n";
430
431 return $text;
432 }
433
434 sub list {
435
436 my $cfg = read_cron();
437
438 my $list = sprintf("%-25s%-25s%-10s%-20s%-6s%-5s\n" , "SOURCE", "NAME", "STATE", "LAST SYNC", "TYPE", "CON");
439
440 my $states = read_state();
441 foreach my $source (sort keys%{$cfg}) {
442 foreach my $name (sort keys%{$cfg->{$source}}) {
443 $list .= sprintf("%-25s", cut_target_width($source, 25));
444 $list .= sprintf("%-25s", cut_target_width($name, 25));
445 $list .= sprintf("%-10s", $states->{$source}->{$name}->{state});
446 $list .= sprintf("%-20s", $states->{$source}->{$name}->{lsync});
447 $list .= sprintf("%-6s", defined($states->{$source}->{$name}->{vm_type}) ? $states->{$source}->{$name}->{vm_type} : "undef");
448 $list .= sprintf("%-5s\n", $cfg->{$source}->{$name}->{method});
449 }
450 }
451
452 return $list;
453 }
454
455 sub vm_exists {
456 my ($target) = @_;
457
458 my @cmd = ('ssh', "root\@$target->{ip}", '--') if $target->{ip};
459
460 my $res = undef;
461
462 return undef if !defined($target->{vmid});
463
464 eval { $res = run_cmd([@cmd, 'ls', "$QEMU_CONF/$target->{vmid}.conf"]) };
465
466 return "qemu" if $res;
467
468 eval { $res = run_cmd([@cmd, 'ls', "$LXC_CONF/$target->{vmid}.conf"]) };
469
470 return "lxc" if $res;
471
472 return undef;
473 }
474
475 sub init {
476 my ($param) = @_;
477
478 my $cfg = read_cron();
479
480 my $job = param_to_job($param);
481
482 $job->{state} = "ok";
483 $job->{lsync} = 0;
484
485 my $source = parse_target($param->{source});
486 my $dest = parse_target($param->{dest});
487
488 if (my $ip = $dest->{ip}) {
489 run_cmd(['ssh-copy-id', '-i', '/root/.ssh/id_rsa.pub', "root\@$ip"]);
490 }
491
492 if (my $ip = $source->{ip}) {
493 run_cmd(['ssh-copy-id', '-i', '/root/.ssh/id_rsa.pub', "root\@$ip"]);
494 }
495
496 die "Pool $dest->{all} does not exists\n" if !check_pool_exists($dest);
497
498 if (!defined($source->{vmid})) {
499 die "Pool $source->{all} does not exists\n" if !check_pool_exists($source);
500 }
501
502 my $vm_type = vm_exists($source);
503 $job->{vm_type} = $vm_type;
504 $source->{vm_type} = $vm_type;
505
506 die "VM $source->{vmid} doesn't exist\n" if $source->{vmid} && !$vm_type;
507
508 die "Config already exists\n" if $cfg->{$job->{source}}->{$job->{name}};
509
510 #check if vm has zfs disks if not die;
511 get_disks($source) if $source->{vmid};
512
513 update_cron($job);
514 update_state($job);
515
516 eval {
517 sync($param) if !$param->{skip};
518 };
519 if(my $err = $@) {
520 destroy_job($param);
521 print $err;
522 }
523 }
524
525 sub get_job {
526 my ($param) = @_;
527
528 my $cfg = read_cron();
529
530 if (!$cfg->{$param->{source}}->{$param->{name}}) {
531 die "Job with source $param->{source} and name $param->{name} does not exist\n" ;
532 }
533 my $job = $cfg->{$param->{source}}->{$param->{name}};
534 $job->{name} = $param->{name};
535 $job->{source} = $param->{source};
536 $job = add_state_to_job($job);
537
538 return $job;
539 }
540
541 sub destroy_job {
542 my ($param) = @_;
543
544 my $job = get_job($param);
545 $job->{state} = "del";
546
547 update_cron($job);
548 update_state($job);
549 }
550
551 sub sync {
552 my ($param) = @_;
553
554 my $lock_fh = IO::File->new("> $LOCKFILE");
555 die "Can't open Lock File: $LOCKFILE $!\n" if !$lock_fh;
556 lock($lock_fh);
557
558 my $date = get_date();
559 my $job;
560 eval {
561 $job = get_job($param);
562 };
563
564 if ($job && $job->{state} eq "syncing") {
565 die "Job --source $param->{source} --name $param->{name} is syncing at the moment";
566 }
567
568 my $dest = parse_target($param->{dest});
569 my $source = parse_target($param->{source});
570
571 my $sync_path = sub {
572 my ($source, $dest, $job, $param, $date) = @_;
573
574 ($source->{old_snap}, $source->{last_snap}) = snapshot_get($source, $dest, $param->{maxsnap}, $param->{name});
575
576 snapshot_add($source, $dest, $param->{name}, $date);
577
578 send_image($source, $dest, $param);
579
580 snapshot_destroy($source, $dest, $param->{method}, $source->{old_snap}) if ($source->{destroy} && $source->{old_snap});
581
582 };
583
584 my $vm_type = vm_exists($source);
585 $source->{vm_type} = $vm_type;
586
587 if ($job) {
588 $job->{state} = "syncing";
589 $job->{vm_type} = $vm_type if !$job->{vm_type};
590 update_state($job);
591 }
592
593 eval{
594 if ($source->{vmid}) {
595 die "VM $source->{vmid} doesn't exist\n" if !$vm_type;
596 my $disks = get_disks($source);
597
598 foreach my $disk (sort keys %{$disks}) {
599 $source->{all} = $disks->{$disk}->{all};
600 $source->{pool} = $disks->{$disk}->{pool};
601 $source->{path} = $disks->{$disk}->{path} if $disks->{$disk}->{path};
602 $source->{last_part} = $disks->{$disk}->{last_part};
603 &$sync_path($source, $dest, $job, $param, $date);
604 }
605 if ($param->{method} eq "ssh" && ($source->{ip} || $dest->{ip})) {
606 send_config($source, $dest,'ssh');
607 } else {
608 send_config($source, $dest,'local');
609 }
610 } else {
611 &$sync_path($source, $dest, $job, $param, $date);
612 }
613 };
614 if(my $err = $@) {
615 if ($job) {
616 $job->{state} = "error";
617 update_state($job);
618 unlock($lock_fh);
619 close($lock_fh);
620 print "Job --source $param->{source} --name $param->{name} got an ERROR!!!\nERROR Message:\n";
621 }
622 die "$err\n";
623 }
624
625 if ($job) {
626 $job->{state} = "ok";
627 $job->{lsync} = $date;
628 update_state($job);
629 }
630
631 unlock($lock_fh);
632 close($lock_fh);
633 }
634
635 sub snapshot_get{
636 my ($source, $dest, $max_snap, $name) = @_;
637
638 my $cmd = [];
639 push @$cmd, 'ssh', "root\@$source->{ip}", '--', if $source->{ip};
640 push @$cmd, 'zfs', 'list', '-r', '-t', 'snapshot', '-Ho', 'name', '-S', 'creation';
641 push @$cmd, $source->{all};
642
643 my $raw = run_cmd($cmd);
644 my $index = 0;
645 my $line = "";
646 my $last_snap = undef;
647 my $old_snap;
648
649 while ($raw && $raw =~ s/^(.*?)(\n|$)//) {
650 $line = $1;
651 if ($line =~ m/(rep_\Q${name}\E_\d{4}-\d{2}-\d{2}_\d{2}:\d{2}:\d{2})$/) {
652
653 $last_snap = $1 if (!$last_snap);
654 $old_snap = $1;
655 $index++;
656 if ($index == $max_snap) {
657 $source->{destroy} = 1;
658 last;
659 };
660 }
661 }
662
663 return ($old_snap, $last_snap) if $last_snap;
664
665 return undef;
666 }
667
668 sub snapshot_add {
669 my ($source, $dest, $name, $date) = @_;
670
671 my $snap_name = "rep_$name\_".$date;
672
673 $source->{new_snap} = $snap_name;
674
675 my $path = "$source->{all}\@$snap_name";
676
677 my $cmd = [];
678 push @$cmd, 'ssh', "root\@$source->{ip}", '--', if $source->{ip};
679 push @$cmd, 'zfs', 'snapshot', $path;
680 eval{
681 run_cmd($cmd);
682 };
683
684 if (my $err = $@) {
685 snapshot_destroy($source, $dest, 'ssh', $snap_name);
686 die "$err\n";
687 }
688 }
689
690 sub write_cron {
691 my ($cfg) = @_;
692
693 my $text = "SHELL=/bin/sh\n";
694 $text .= "PATH=/usr/local/sbin:/usr/local/bin:/sbin:/bin:/usr/sbin:/usr/bin\n";
695
696 my $fh = IO::File->new("> $CRONJOBS");
697 die "Could not open file: $!\n" if !$fh;
698
699 foreach my $source (sort keys%{$cfg}) {
700 foreach my $sync_name (sort keys%{$cfg->{$source}}) {
701 next if $cfg->{$source}->{$sync_name}->{status} ne 'ok';
702 $text .= "$PROG_PATH sync";
703 $text .= " -source ";
704 if ($cfg->{$source}->{$sync_name}->{vmid}) {
705 $text .= "$cfg->{$source}->{$sync_name}->{source_ip}:" if $cfg->{$source}->{$sync_name}->{source_ip};
706 $text .= "$cfg->{$source}->{$sync_name}->{vmid} ";
707 } else {
708 $text .= "$cfg->{$source}->{$sync_name}->{source_ip}:" if $cfg->{$source}->{$sync_name}->{source_ip};
709 $text .= "$cfg->{$source}->{$sync_name}->{source_pool}";
710 $text .= "$cfg->{$source}->{$sync_name}->{source_path}" if $cfg->{$source}->{$sync_name}->{source_path};
711 }
712 $text .= " -dest ";
713 $text .= "$cfg->{$source}->{$sync_name}->{dest_ip}:" if $cfg->{$source}->{$sync_name}->{dest_ip};
714 $text .= "$cfg->{$source}->{$sync_name}->{dest_pool}";
715 $text .= "$cfg->{$source}->{$sync_name}->{dest_path}" if $cfg->{$source}->{$sync_name}->{dest_path};
716 $text .= " -name $sync_name ";
717 $text .= " -limit $cfg->{$source}->{$sync_name}->{limit}" if $cfg->{$source}->{$sync_name}->{limit};
718 $text .= " -maxsnap $cfg->{$source}->{$sync_name}->{maxsnap}" if $cfg->{$source}->{$sync_name}->{maxsnap};
719 $text .= "\n";
720 }
721 }
722 die "Can't write to cron\n" if (!print($fh $text));
723 close($fh);
724 }
725
726 sub get_disks {
727 my ($target) = @_;
728
729 my $cmd = [];
730 push @$cmd, 'ssh', "root\@$target->{ip}", '--', if $target->{ip};
731
732 if ($target->{vm_type} eq 'qemu') {
733 push @$cmd, 'qm', 'config', $target->{vmid};
734 } elsif ($target->{vm_type} eq 'lxc') {
735 push @$cmd, 'pct', 'config', $target->{vmid};
736 } else {
737 die "VM Type unknown\n";
738 }
739
740 my $res = run_cmd($cmd);
741
742 my $disks = parse_disks($res, $target->{ip}, $target->{vm_type});
743
744 return $disks;
745 }
746
747 sub run_cmd {
748 my ($cmd) = @_;
749 print "Start CMD\n" if $DEBUG;
750 print Dumper $cmd if $DEBUG;
751 if (ref($cmd) eq 'ARRAY') {
752 $cmd = join(' ', map { ref($_) ? $$_ : shell_quote($_) } @$cmd);
753 }
754 my $output = `$cmd 2>&1`;
755
756 die "COMMAND:\n\t$cmd\nGET ERROR:\n\t$output" if 0 != $?;
757
758 chomp($output);
759 print Dumper $output if $DEBUG;
760 print "END CMD\n" if $DEBUG;
761 return $output;
762 }
763
764 sub parse_disks {
765 my ($text, $ip, $vm_type) = @_;
766
767 my $disks;
768
769 my $num = 0;
770 while ($text && $text =~ s/^(.*?)(\n|$)//) {
771 my $line = $1;
772
773 next if $line =~ /cdrom|none/;
774 next if $line !~ m/^(?:((?:virtio|ide|scsi|sata|mp)\d+)|rootfs): /;
775
776 #QEMU if backup is not set include in sync
777 next if $vm_type eq 'qemu' && ($line =~ m/backup=(?i:0|no|off|false)/);
778
779 #LXC if backup is not set do no in sync
780 next if $vm_type eq 'lxc' && ($line =~ m/^mp\d:/) && ($line !~ m/backup=(?i:1|yes|on|true)/);
781
782 my $disk = undef;
783 my $stor = undef;
784 if($line =~ m/^(?:(?:(?:virtio|ide|scsi|sata|mp)\d+)|rootfs): (.*)$/) {
785 my @parameter = split(/,/,$1);
786
787 foreach my $opt (@parameter) {
788 if ($opt =~ m/^(?:file=|volume=)?([^:]+:)([A-Za-z0-9\-]+)$/){
789 $disk = $2;
790 $stor = $1;
791 last;
792 }
793 }
794 if (!defined($disk) || !defined($stor)) {
795 print "Disk: \"$line\" has no valid zfs dataset format and will not include in pve-sync\n";
796 next;
797 }
798 } else {
799 print "Disk: \"$line\" has no valid zfs dataset format and will not include in pve-sync\n";
800 next;
801 }
802
803 my $cmd = [];
804 push @$cmd, 'ssh', "root\@$ip", '--' if $ip;
805 push @$cmd, 'pvesm', 'path', "$stor$disk";
806 my $path = run_cmd($cmd);
807
808 die "Get no path from pvesm path $stor$disk\n" if !$path;
809
810 if ($vm_type eq 'qemu' && $path =~ m/^\/dev\/zvol\/(\w+.*)(\/$disk)$/) {
811
812 my @array = split('/', $1);
813 $disks->{$num}->{pool} = shift(@array);
814 $disks->{$num}->{all} = $disks->{$num}->{pool};
815 if (0 < @array) {
816 $disks->{$num}->{path} = join('/', @array);
817 $disks->{$num}->{all} .= "\/$disks->{$num}->{path}";
818 }
819 $disks->{$num}->{last_part} = $disk;
820 $disks->{$num}->{all} .= "\/$disk";
821
822 $num++;
823 } elsif ($vm_type eq 'lxc' && $path =~ m/^\/(\w+.+)(\/(\w+.*))*(\/$disk)$/) {
824
825 $disks->{$num}->{pool} = $1;
826 $disks->{$num}->{all} = $disks->{$num}->{pool};
827
828 if ($2) {
829 $disks->{$num}->{path} = $3;
830 $disks->{$num}->{all} .= "\/$disks->{$num}->{path}";
831 }
832
833 $disks->{$num}->{last_part} = $disk;
834 $disks->{$num}->{all} .= "\/$disk";
835
836 $num++;
837
838 } else {
839 die "ERROR: in path\n";
840 }
841 }
842
843 die "Vm include no disk on zfs.\n" if !$disks->{0};
844 return $disks;
845 }
846
847 sub snapshot_destroy {
848 my ($source, $dest, $method, $snap) = @_;
849
850 my @zfscmd = ('zfs', 'destroy');
851 my $snapshot = "$source->{all}\@$snap";
852
853 eval {
854 if($source->{ip} && $method eq 'ssh'){
855 run_cmd(['ssh', "root\@$source->{ip}", '--', @zfscmd, $snapshot]);
856 } else {
857 run_cmd([@zfscmd, $snapshot]);
858 }
859 };
860 if (my $erro = $@) {
861 warn "WARN: $erro";
862 }
863 if ($dest) {
864 my @ssh = $dest->{ip} ? ('ssh', "root\@$dest->{ip}", '--') : ();
865
866 my $path = "$dest->{all}\/$source->{last_part}";
867
868 eval {
869 run_cmd([@ssh, @zfscmd, "$path\@$snap"]);
870 };
871 if (my $erro = $@) {
872 warn "WARN: $erro";
873 }
874 }
875 }
876
877 sub snapshot_exist {
878 my ($source , $dest, $method) = @_;
879
880 my $cmd = [];
881 push @$cmd, 'ssh', "root\@$dest->{ip}", '--' if $dest->{ip};
882 push @$cmd, 'zfs', 'list', '-rt', 'snapshot', '-Ho', 'name';
883 push @$cmd, "$dest->{all}/$source->{last_part}\@$source->{old_snap}";
884
885 my $text = "";
886 eval {$text =run_cmd($cmd);};
887 if (my $erro =$@) {
888 warn "WARN: $erro";
889 return undef;
890 }
891
892 while ($text && $text =~ s/^(.*?)(\n|$)//) {
893 my $line =$1;
894 return 1 if $line =~ m/^.*$source->{old_snap}$/;
895 }
896 }
897
898 sub send_image {
899 my ($source, $dest, $param) = @_;
900
901 my $cmd = [];
902
903 push @$cmd, 'ssh', '-o', 'BatchMode=yes', "root\@$source->{ip}", '--' if $source->{ip};
904 push @$cmd, 'zfs', 'send';
905 push @$cmd, '-v' if $param->{verbose};
906
907 if($source->{last_snap} && snapshot_exist($source , $dest, $param->{method})) {
908 push @$cmd, '-i', "$source->{all}\@$source->{last_snap}";
909 }
910 push @$cmd, '--', "$source->{all}\@$source->{new_snap}";
911
912 if ($param->{limit}){
913 my $bwl = $param->{limit}*1024;
914 push @$cmd, \'|', 'cstream', '-t', $bwl;
915 }
916 my $target = "$dest->{all}/$source->{last_part}";
917 $target =~ s!/+!/!g;
918
919 push @$cmd, \'|';
920 push @$cmd, 'ssh', '-o', 'BatchMode=yes', "root\@$dest->{ip}", '--' if $dest->{ip};
921 push @$cmd, 'zfs', 'recv', '-F', '--';
922 push @$cmd, "$target";
923
924 eval {
925 run_cmd($cmd)
926 };
927
928 if (my $erro = $@) {
929 snapshot_destroy($source, undef, $param->{method}, $source->{new_snap});
930 die $erro;
931 };
932 }
933
934
935 sub send_config{
936 my ($source, $dest, $method) = @_;
937
938 my $source_target = $source->{vm_type} eq 'qemu' ? "$QEMU_CONF/$source->{vmid}.conf": "$LXC_CONF/$source->{vmid}.conf";
939 my $dest_target_new ="$source->{vmid}.conf.$source->{vm_type}.$source->{new_snap}";
940
941 my $config_dir = $dest->{last_part} ? "${CONFIG_PATH}/$dest->{last_part}" : $CONFIG_PATH;
942
943 $dest_target_new = $config_dir.'/'.$dest_target_new;
944
945 if ($method eq 'ssh'){
946 if ($dest->{ip} && $source->{ip}) {
947 run_cmd(['ssh', "root\@$dest->{ip}", '--', 'mkdir', '-p', '--', $config_dir]);
948 run_cmd(['scp', '--', "root\@[$source->{ip}]:$source_target", "root\@[$dest->{ip}]:$dest_target_new"]);
949 } elsif ($dest->{ip}) {
950 run_cmd(['ssh', "root\@$dest->{ip}", '--', 'mkdir', '-p', '--', $config_dir]);
951 run_cmd(['scp', '--', $source_target, "root\@[$dest->{ip}]:$dest_target_new"]);
952 } elsif ($source->{ip}) {
953 run_cmd(['mkdir', '-p', '--', $config_dir]);
954 run_cmd(['scp', '--', "root\@$source->{ip}:$source_target", $dest_target_new]);
955 }
956
957 if ($source->{destroy}){
958 my $dest_target_old ="${config_dir}/$source->{vmid}.conf.$source->{vm_type}.$source->{old_snap}";
959 if($dest->{ip}){
960 run_cmd(['ssh', "root\@$dest->{ip}", '--', 'rm', '-f', '--', $dest_target_old]);
961 } else {
962 run_cmd(['rm', '-f', '--', $dest_target_old]);
963 }
964 }
965 } elsif ($method eq 'local') {
966 run_cmd(['mkdir', '-p', '--', $config_dir]);
967 run_cmd(['cp', $source_target, $dest_target_new]);
968 }
969 }
970
971 sub get_date {
972 my ($sec, $min, $hour, $mday, $mon, $year, $wday, $yday, $isdst) = localtime(time);
973 my $datestamp = sprintf ("%04d-%02d-%02d_%02d:%02d:%02d", $year+1900, $mon+1, $mday, $hour, $min, $sec);
974
975 return $datestamp;
976 }
977
978 sub status {
979 my $cfg = read_cron();
980
981 my $status_list = sprintf("%-25s%-25s%-10s\n", "SOURCE", "NAME", "STATUS");
982
983 my $states = read_state();
984
985 foreach my $source (sort keys%{$cfg}) {
986 foreach my $sync_name (sort keys%{$cfg->{$source}}) {
987 $status_list .= sprintf("%-25s", cut_target_width($source, 25));
988 $status_list .= sprintf("%-25s", cut_target_width($sync_name, 25));
989 $status_list .= "$states->{$source}->{$sync_name}->{state}\n";
990 }
991 }
992
993 return $status_list;
994 }
995
996 sub enable_job {
997 my ($param) = @_;
998
999 my $job = get_job($param);
1000 $job->{state} = "ok";
1001 update_state($job);
1002 update_cron($job);
1003 }
1004
1005 sub disable_job {
1006 my ($param) = @_;
1007
1008 my $job = get_job($param);
1009 $job->{state} = "stopped";
1010 update_state($job);
1011 update_cron($job);
1012 }
1013
1014 my $command = $ARGV[0];
1015
1016 my $commands = {'destroy' => 1,
1017 'create' => 1,
1018 'sync' => 1,
1019 'list' => 1,
1020 'status' => 1,
1021 'help' => 1,
1022 'enable' => 1,
1023 'disable' => 1};
1024
1025 if (!$command || !$commands->{$command}) {
1026 usage();
1027 die "\n";
1028 }
1029
1030 my $help_sync = "$PROGNAME sync -dest <string> -source <string> [OPTIONS]\n
1031 \twill sync one time\n
1032 \t-dest\tstring\n
1033 \t\tthe destination target is like [IP:]<Pool>[/Path]\n
1034 \t-limit\tinteger\n
1035 \t\tmax sync speed in kBytes/s, default unlimited\n
1036 \t-maxsnap\tinteger\n
1037 \t\thow much snapshots will be kept before get erased, default 1/n
1038 \t-name\tstring\n
1039 \t\tname of the sync job, if not set it is default.
1040 \tIt is only necessary if scheduler allready contains this source.\n
1041 \t-source\tstring\n
1042 \t\tthe source can be an <VMID> or [IP:]<ZFSPool>[/Path]\n";
1043
1044 my $help_create = "$PROGNAME create -dest <string> -source <string> [OPTIONS]/n
1045 \tCreate a sync Job\n
1046 \t-dest\tstring\n
1047 \t\tthe destination target is like [IP]:<Pool>[/Path]\n
1048 \t-limit\tinteger\n
1049 \t\tmax sync speed in kBytes/s, default unlimited\n
1050 \t-maxsnap\tstring\n
1051 \t\thow much snapshots will be kept before get erased, default 1\n
1052 \t-name\tstring\n
1053 \t\tname of the sync job, if not set it is default\n
1054 \t-skip\tboolean\n
1055 \t\tif this flag is set it will skip the first sync\n
1056 \t-source\tstring\n
1057 \t\tthe source can be an <VMID> or [IP:]<ZFSPool>[/Path]\n";
1058
1059 my $help_destroy = "$PROGNAME destroy -source <string> [OPTIONS]\n
1060 \tremove a sync Job from the scheduler\n
1061 \t-name\tstring\n
1062 \t\tname of the sync job, if not set it is default\n
1063 \t-source\tstring\n
1064 \t\tthe source can be an <VMID> or [IP:]<ZFSPool>[/Path]\n";
1065
1066 my $help_help = "$PROGNAME help <cmd> [OPTIONS]\n
1067 \tGet help about specified command.\n
1068 \t<cmd>\tstring\n
1069 \t\tCommand name\n
1070 \t-verbose\tboolean\n
1071 \t\tVerbose output format.\n";
1072
1073 my $help_list = "$PROGNAME list\n
1074 \tGet a List of all scheduled Sync Jobs\n";
1075
1076 my $help_status = "$PROGNAME status\n
1077 \tGet the status of all scheduled Sync Jobs\n";
1078
1079 my $help_enable = "$PROGNAME enable -source <string> [OPTIONS]\n
1080 \tenable a syncjob and reset error\n
1081 \t-name\tstring\n
1082 \t\tname of the sync job, if not set it is default\n
1083 \t-source\tstring\n
1084 \t\tthe source can be an <VMID> or [IP:]<ZFSPool>[/Path]\n";
1085
1086 my $help_disable = "$PROGNAME disable -source <string> [OPTIONS]\n
1087 \tpause a syncjob\n
1088 \t-name\tstring\n
1089 \t\tname of the sync job, if not set it is default\n
1090 \t-source\tstring\n
1091 \t\tthe source can be an <VMID> or [IP:]<ZFSPool>[/Path]\n";
1092
1093 sub help {
1094 my ($command) = @_;
1095
1096 if ($command eq 'help') {
1097 die "$help_help\n";
1098
1099 } elsif ($command eq 'sync') {
1100 die "$help_sync\n";
1101
1102 } elsif ($command eq 'destroy') {
1103 die "$help_destroy\n";
1104
1105 } elsif ($command eq 'create') {
1106 die "$help_create\n";
1107
1108 } elsif ($command eq 'list') {
1109 die "$help_list\n";
1110
1111 } elsif ($command eq 'status') {
1112 die "$help_status\n";
1113
1114 } elsif ($command eq 'enable') {
1115 die "$help_enable\n";
1116
1117 } elsif ($command eq 'disable') {
1118 die "$help_disable\n";
1119
1120 }
1121
1122 }
1123
1124 my @arg = @ARGV;
1125 my $param = parse_argv(@arg);
1126
1127 if ($command eq 'destroy') {
1128 die "$help_destroy\n" if !$param->{source};
1129
1130 check_target($param->{source});
1131 destroy_job($param);
1132
1133 } elsif ($command eq 'sync') {
1134 die "$help_sync\n" if !$param->{source} || !$param->{dest};
1135
1136 check_target($param->{source});
1137 check_target($param->{dest});
1138 sync($param);
1139
1140 } elsif ($command eq 'create') {
1141 die "$help_create\n" if !$param->{source} || !$param->{dest};
1142
1143 check_target($param->{source});
1144 check_target($param->{dest});
1145 init($param);
1146
1147 } elsif ($command eq 'status') {
1148 print status();
1149
1150 } elsif ($command eq 'list') {
1151 print list();
1152
1153 } elsif ($command eq 'help') {
1154 my $help_command = $ARGV[1];
1155
1156 if ($help_command && $commands->{$help_command}) {
1157 print help($help_command);
1158
1159 }
1160 if ($param->{verbose} == 1){
1161 exec("man $PROGNAME");
1162
1163 } else {
1164 usage(1);
1165
1166 }
1167
1168 } elsif ($command eq 'enable') {
1169 die "$help_enable\n" if !$param->{source};
1170
1171 check_target($param->{source});
1172 enable_job($param);
1173
1174 } elsif ($command eq 'disable') {
1175 die "$help_disable\n" if !$param->{source};
1176
1177 check_target($param->{source});
1178 disable_job($param);
1179
1180 }
1181
1182 sub usage {
1183 my ($help) = @_;
1184
1185 print("ERROR:\tno command specified\n") if !$help;
1186 print("USAGE:\t$PROGNAME <COMMAND> [ARGS] [OPTIONS]\n");
1187 print("\t$PROGNAME help [<cmd>] [OPTIONS]\n\n");
1188 print("\t$PROGNAME create -dest <string> -source <string> [OPTIONS]\n");
1189 print("\t$PROGNAME destroy -source <string> [OPTIONS]\n");
1190 print("\t$PROGNAME disable -source <string> [OPTIONS]\n");
1191 print("\t$PROGNAME enable -source <string> [OPTIONS]\n");
1192 print("\t$PROGNAME list\n");
1193 print("\t$PROGNAME status\n");
1194 print("\t$PROGNAME sync -dest <string> -source <string> [OPTIONS]\n");
1195 }
1196
1197 sub check_target {
1198 my ($target) = @_;
1199 parse_target($target);
1200 }
1201
1202 __END__
1203
1204 =head1 NAME
1205
1206 pve-zsync - PVE ZFS Replication Manager
1207
1208 =head1 SYNOPSIS
1209
1210 pve-zsync <COMMAND> [ARGS] [OPTIONS]
1211
1212 pve-zsync help <cmd> [OPTIONS]
1213
1214 Get help about specified command.
1215
1216 <cmd> string
1217
1218 Command name
1219
1220 -verbose boolean
1221
1222 Verbose output format.
1223
1224 pve-zsync create -dest <string> -source <string> [OPTIONS]
1225
1226 Create a sync Job
1227
1228 -dest string
1229
1230 the destination target is like [IP]:<Pool>[/Path]
1231
1232 -limit integer
1233
1234 max sync speed in kBytes/s, default unlimited
1235
1236 -maxsnap string
1237
1238 how much snapshots will be kept before get erased, default 1
1239
1240 -name string
1241
1242 name of the sync job, if not set it is default
1243
1244 -skip boolean
1245
1246 if this flag is set it will skip the first sync
1247
1248 -source string
1249
1250 the source can be an <VMID> or [IP:]<ZFSPool>[/Path]
1251
1252 pve-zsync destroy -source <string> [OPTIONS]
1253
1254 remove a sync Job from the scheduler
1255
1256 -name string
1257
1258 name of the sync job, if not set it is default
1259
1260 -source string
1261
1262 the source can be an <VMID> or [IP:]<ZFSPool>[/Path]
1263
1264 pve-zsync disable -source <string> [OPTIONS]
1265
1266 pause a sync job
1267
1268 -name string
1269
1270 name of the sync job, if not set it is default
1271
1272 -source string
1273
1274 the source can be an <VMID> or [IP:]<ZFSPool>[/Path]
1275
1276 pve-zsync enable -source <string> [OPTIONS]
1277
1278 enable a syncjob and reset error
1279
1280 -name string
1281
1282 name of the sync job, if not set it is default
1283
1284 -source string
1285
1286 the source can be an <VMID> or [IP:]<ZFSPool>[/Path]
1287 pve-zsync list
1288
1289 Get a List of all scheduled Sync Jobs
1290
1291 pve-zsync status
1292
1293 Get the status of all scheduled Sync Jobs
1294
1295 pve-zsync sync -dest <string> -source <string> [OPTIONS]
1296
1297 will sync one time
1298
1299 -dest string
1300
1301 the destination target is like [IP:]<Pool>[/Path]
1302
1303 -limit integer
1304
1305 max sync speed in kBytes/s, default unlimited
1306
1307 -maxsnap integer
1308
1309 how much snapshots will be kept before get erased, default 1
1310
1311 -name string
1312
1313 name of the sync job, if not set it is default.
1314 It is only necessary if scheduler allready contains this source.
1315
1316 -source string
1317
1318 the source can be an <VMID> or [IP:]<ZFSPool>[/Path]
1319
1320 =head1 DESCRIPTION
1321
1322 This Tool helps you to sync your VM or directory which stored on ZFS between 2 servers.
1323 This tool also has the capability to add jobs to cron so the sync will be automatically done.
1324 The default syncing interval is set to 15 min, if you want to change this value you can do this in /etc/cron.d/pve-zsync.
1325 To config cron see man crontab.
1326
1327 =head2 PVE ZFS Storage sync Tool
1328
1329 This Tool can get remote pool on other PVE or send Pool to others ZFS machines
1330
1331 =head1 EXAMPLES
1332
1333 add sync job from local VM to remote ZFS Server
1334 pve-zsync create -source=100 -dest=192.168.1.2:zfspool
1335
1336 =head1 IMPORTANT FILES
1337
1338 Cron jobs and config are stored at /etc/cron.d/pve-zsync
1339
1340 The VM config get copied on the destination machine to /var/lib/pve-zsync/
1341
1342 =head1 COPYRIGHT AND DISCLAIMER
1343
1344 Copyright (C) 2007-2015 Proxmox Server Solutions GmbH
1345
1346 This program is free software: you can redistribute it and/or modify it
1347 under the terms of the GNU Affero General Public License as published
1348 by the Free Software Foundation, either version 3 of the License, or
1349 (at your option) any later version.
1350
1351 This program is distributed in the hope that it will be useful, but
1352 WITHOUT ANY WARRANTY; without even the implied warranty of
1353 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
1354 Affero General Public License for more details.
1355
1356 You should have received a copy of the GNU Affero General Public
1357 License along with this program. If not, see
1358 <http://www.gnu.org/licenses/>.