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