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     push @$cmd, 'ssh', "root\@$target->{ip}", '--', if $target->{ip};
 109     push @$cmd, 'zfs', 'list', '-H', '--', $target->{all};
 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     push @$cmd, 'ssh', "root\@$target->{ip}", '--', if $target->{ip};
 435     push @$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 = [];
 594     push @$cmd, 'ssh', "root\@$source->{ip}", '--', if $source->{ip};
 595     push @$cmd, 'zfs', 'list', '-r', '-t', 'snapshot', '-Ho', 'name', '-S', 'creation';
 596     push @$cmd, $source->{all};
 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 = [];
 632     push @$cmd, 'ssh', "root\@$source->{ip}", '--', if $source->{ip};
 633     push @$cmd, 'zfs', 'snapshot', $path;
 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     push @$cmd, 'ssh', "root\@$target->{ip}", '--', if $target->{ip};
 685     push @$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         push @$cmd, 'ssh', "root\@$ip", '--' if $ip;
 734         push @$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     push @$cmd, 'ssh', "root\@$dest->{ip}", '--' if $dest->{ip};
 794     push @$cmd, 'zfs', 'list', '-rt', 'snapshot', '-Ho', 'name';
 795     push @$cmd, "$dest->{all}/$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     push @$cmd, 'ssh', "root\@$source->{ip}", '--' if $source->{ip};
 816     push @$cmd, 'zfs', 'send';
 817     push @$cmd, '-v' if $param->{verbose};
 818
 819     if($source->{last_snap} && snapshot_exist($source ,$dest, $param->{method})) {
 820         push @$cmd, '-i', "$source->{all}\@$source->{last_snap}";
 821     }
 822     push @$cmd, '--', "$source->{all}\@$source->{new_snap}";
 823
 824     if ($param->{limit}){
 825         my $bwl = $param->{limit}*1024;
 826         push @$cmd, \'|', 'cstream', '-t', $bwl;
 827     }
 828     push @$cmd, \'|';
 829     push @$cmd, 'ssh', "root\@$dest->{ip}", '--' if $dest->{ip};
 830     push @$cmd, 'zfs', 'recv', '--';
 831     push @$cmd, "$dest->{all}/$source->{last_part}\@$source->{new_snap}";
 832
 833     eval {
 834         run_cmd($cmd)
 835     };
 836
 837     if (my $erro = $@) {
 838         snapshot_destroy($source, undef, $param->{method}, $source->{new_snap});
 839         die $erro;
 840     };
 841 }
 842
 843
 844 sub send_config{
 845     my ($source, $dest, $method) = @_;
 846
 847     my $source_target ="$QEMU_CONF$source->{vmid}.conf";
 848     my $dest_target_new ="$CONFIG_PATH$source->{vmid}.conf.$source->{new_snap}";
 849
 850     if ($method eq 'ssh'){
 851         if ($dest->{ip} && $source->{ip}) {
 852             run_cmd(['ssh', "root\@$dest->{ip}", '--', 'mkdir', '-p', '--', $CONFIG_PATH]);
 853             run_cmd(['scp', '--', "root\@[$source->{ip}]:$source_target", "root\@[$dest->{ip}]:$dest_target_new"]);
 854         } elsif ($dest->{ip}) {
 855             run_cmd(['ssh', "root\@$dest->{ip}", '--', 'mkdir', '-p', '--', $CONFIG_PATH]);
 856             run_cmd(['scp', '--', $source_target, "root\@[$dest->{ip}]:$dest_target_new"]);
 857         } elsif ($source->{ip}) {
 858             run_cmd(['mkdir', '-p', '--', $CONFIG_PATH]);
 859             run_cmd(['scp', '--', "root\@$source->{ip}:$source_target", $dest_target_new]);
 860         }
 861
 862         if ($source->{destroy}){
 863             my $dest_target_old ="$CONFIG_PATH$source->{vmid}.conf.$source->{old_snap}";
 864             if($dest->{ip}){
 865                 run_cmd(['ssh', "root\@$dest->{ip}", '--', 'rm', '-f', '--', $dest_target_old]);
 866             } else {
 867                 run_cmd(['rm', '-f', '--', $dest_target_old]);
 868             }
 869         }
 870     }
 871 }
 872
 873 sub get_date {
 874     my ($sec, $min, $hour, $mday, $mon, $year, $wday, $yday, $isdst) = localtime(time);
 875     my $datestamp = sprintf ("%04d-%02d-%02d_%02d:%02d:%02d", $year+1900, $mon+1, $mday, $hour, $min, $sec);
 876
 877     return $datestamp;
 878 }
 879
 880 sub status {
 881     my $cfg = read_cron();
 882
 883     my $status_list = sprintf("%-25s%-15s%-10s\n", "SOURCE", "NAME", "STATUS");
 884
 885     my $states = read_state();
 886
 887     foreach my $source (sort keys%{$cfg}) {
 888         foreach my $sync_name (sort keys%{$cfg->{$source}}) {
 889             $status_list .= sprintf("%-25s", cut_target_width($source, 25));
 890             $status_list .= sprintf("%-15s", cut_target_width($sync_name, 25));
 891             $status_list .= "$states->{$source}->{$sync_name}->{state}\n";
 892         }
 893     }
 894
 895     return $status_list;
 896 }
 897
 898 sub enable_job {
 899     my ($param) = @_;
 900
 901     my $job = get_job($param);
 902     $job->{state} = "ok";
 903     update_state($job);
 904     update_cron($job);
 905 }
 906
 907 sub disable_job {
 908     my ($param) = @_;
 909
 910     my $job = get_job($param);
 911     $job->{state} = "stopped";
 912     update_state($job);
 913     update_cron($job);
 914 }
 915
 916 my $command = $ARGV[0];
 917
 918 my $commands = {'destroy' => 1,
 919                'create' => 1,
 920                'sync' => 1,
 921                'list' => 1,
 922                'status' => 1,
 923                'help' => 1,
 924                'enable' => 1,
 925                'disable' => 1};
 926
 927 if (!$command || !$commands->{$command}) {
 928     usage();
 929     die "\n";
 930 }
 931
 932 my $help_sync = "$PROGNAME sync -dest <string> -source <string> [OPTIONS]\n
 933 \twill sync one time\n
 934 \t-dest\tstring\n
 935 \t\tthe destination target is like [IP:]<Pool>[/Path]\n
 936 \t-limit\tinteger\n
 937 \t\tmax sync speed in kBytes/s, default unlimited\n
 938 \t-maxsnap\tinteger\n
 939 \t\thow much snapshots will be kept before get erased, default 1/n
 940 \t-name\tstring\n
 941 \t\tname of the sync job, if not set it is default.
 942 \tIt is only necessary if scheduler allready contains this source.\n
 943 \t-source\tstring\n
 944 \t\tthe source can be an <VMID> or [IP:]<ZFSPool>[/Path]\n";
 945
 946 my $help_create = "$PROGNAME create -dest <string> -source <string> [OPTIONS]/n
 947 \tCreate a sync Job\n
 948 \t-dest\tstringn\n
 949 \t\tthe destination target is like [IP]:<Pool>[/Path]\n
 950 \t-limit\tinteger\n
 951 \t\tmax sync speed in kBytes/s, default unlimited\n
 952 \t-maxsnap\tstring\n
 953 \t\thow much snapshots will be kept before get erased, default 1\n
 954 \t-name\tstring\n
 955 \t\tname of the sync job, if not set it is default\n
 956 \t-skip\tboolean\n
 957 \t\tif this flag is set it will skip the first sync\n
 958 \t-source\tstring\n
 959 \t\tthe source can be an <VMID> or [IP:]<ZFSPool>[/Path]\n";
 960
 961 my $help_destroy = "$PROGNAME destroy -source <string> [OPTIONS]\n
 962 \tremove a sync Job from the scheduler\n
 963 \t-name\tstring\n
 964 \t\tname of the sync job, if not set it is default\n
 965 \t-source\tstring\n
 966 \t\tthe source can be an  <VMID> or [IP:]<ZFSPool>[/Path]\n";
 967
 968 my $help_help = "$PROGNAME help <cmd> [OPTIONS]\n
 969 \tGet help about specified command.\n
 970 \t<cmd>\tstring\n
 971 \t\tCommand name\n
 972 \t-verbose\tboolean\n
 973 \t\tVerbose output format.\n";
 974
 975 my $help_list = "$PROGNAME list\n
 976 \tGet a List of all scheduled Sync Jobs\n";
 977
 978 my $help_status = "$PROGNAME status\n
 979 \tGet the status of all scheduled Sync Jobs\n";
 980
 981 my $help_enable = "$PROGNAME enable -source <string> [OPTIONS]\n
 982 \tenable a syncjob and reset error\n
 983 \t-name\tstring\n
 984 \t\tname of the sync job, if not set it is default\n
 985 \t-source\tstring\n
 986 \t\tthe source can be an  <VMID> or [IP:]<ZFSPool>[/Path]\n";
 987
 988 my $help_disable = "$PROGNAME disable -source <string> [OPTIONS]\n
 989 \tpause a syncjob\n
 990 \t-name\tstring\n
 991 \t\tname of the sync job, if not set it is default\n
 992 \t-source\tstring\n
 993 \t\tthe source can be an  <VMID> or [IP:]<ZFSPool>[/Path]\n";
 994
 995 sub help {
 996     my ($command) = @_;
 997
 998     switch($command){
 999         case 'help'
1000         {
1001             die "$help_help\n";
1002         }
1003         case 'sync'
1004         {
1005             die "$help_sync\n";
1006         }
1007         case 'destroy'
1008         {
1009             die "$help_destroy\n";
1010         }
1011         case 'create'
1012         {
1013             die "$help_create\n";
1014         }
1015         case 'list'
1016         {
1017             die "$help_list\n";
1018         }
1019         case 'status'
1020         {
1021             die "$help_status\n";
1022         }
1023         case 'enable'
1024         {
1025              die "$help_enable\n";
1026         }
1027         case 'disable'
1028         {
1029             die "$help_enable\n";
1030         }
1031     }
1032
1033 }
1034
1035 my @arg = @ARGV;
1036 my $param = parse_argv(@arg);
1037
1038
1039 switch($command) {
1040     case "destroy"
1041     {
1042         die "$help_destroy\n" if !$param->{source};
1043         check_target($param->{source});
1044         destroy_job($param);
1045     }
1046     case "sync"
1047     {
1048         die "$help_sync\n" if !$param->{source} || !$param->{dest};
1049         check_target($param->{source});
1050         check_target($param->{dest});
1051         sync($param);
1052     }
1053     case "create"
1054     {
1055         die "$help_create\n" if !$param->{source} || !$param->{dest};
1056         check_target($param->{source});
1057         check_target($param->{dest});
1058         init($param);
1059     }
1060     case "status"
1061     {
1062         print status();
1063     }
1064     case "list"
1065     {
1066         print list();
1067     }
1068     case "help"
1069     {
1070         my $help_command = $ARGV[1];
1071         if ($help_command && $commands->{$help_command}) {
1072             print help($help_command);
1073         }
1074         if ($param->{verbose} == 1){
1075             exec("man $PROGNAME");
1076         } else {
1077             usage(1);
1078         }
1079     }
1080     case "enable"
1081     {
1082         die "$help_enable\n" if !$param->{source};
1083         check_target($param->{source});
1084         enable_job($param);
1085     }
1086     case "disable"
1087     {
1088         die "$help_disable\n" if !$param->{source};
1089         check_target($param->{source});
1090         disable_job($param);
1091     }
1092 }
1093
1094 sub usage {
1095     my ($help) = @_;
1096
1097     print("ERROR:\tno command specified\n") if !$help;
1098     print("USAGE:\t$PROGNAME <COMMAND> [ARGS] [OPTIONS]\n");
1099     print("\t$PROGNAME help [<cmd>] [OPTIONS]\n\n");
1100     print("\t$PROGNAME create -dest <string> -source <string> [OPTIONS]\n");
1101     print("\t$PROGNAME destroy -source <string> [OPTIONS]\n");
1102     print("\t$PROGNAME disable -source <string> [OPTIONS]\n");
1103     print("\t$PROGNAME enable -source <string> [OPTIONS]\n");
1104     print("\t$PROGNAME list\n");
1105     print("\t$PROGNAME status\n");
1106     print("\t$PROGNAME sync -dest <string> -source <string> [OPTIONS]\n");
1107 }
1108
1109 sub check_target {
1110     my ($target) = @_;
1111     parse_target($target);
1112 }
1113
1114 __END__
1115
1116 =head1 NAME
1117
1118 pve-zsync - PVE ZFS Replication Manager
1119
1120 =head1 SYNOPSIS
1121
1122 pve-zsync <COMMAND> [ARGS] [OPTIONS]
1123
1124 pve-zsync help <cmd> [OPTIONS]
1125
1126         Get help about specified command.
1127
1128         <cmd>      string
1129
1130                 Command name
1131
1132         -verbose   boolean
1133
1134                 Verbose output format.
1135
1136 pve-zsync create -dest <string> -source <string> [OPTIONS]
1137
1138         Create a sync Job
1139
1140         -dest      string
1141
1142                 the destination target is like [IP]:<Pool>[/Path]
1143
1144         -limit     integer
1145
1146                 max sync speed in kBytes/s, default unlimited
1147
1148         -maxsnap   string
1149
1150                 how much snapshots will be kept before get erased, default 1
1151
1152         -name      string
1153
1154                 name of the sync job, if not set it is default
1155
1156         -skip      boolean
1157
1158                 if this flag is set it will skip the first sync
1159
1160         -source    string
1161
1162                 the source can be an <VMID> or [IP:]<ZFSPool>[/Path]
1163
1164 pve-zsync destroy -source <string> [OPTIONS]
1165
1166         remove a sync Job from the scheduler
1167
1168         -name      string
1169
1170                 name of the sync job, if not set it is default
1171
1172         -source    string
1173
1174                 the source can be an  <VMID> or [IP:]<ZFSPool>[/Path]
1175
1176 pve-zsync disable -source <string> [OPTIONS]
1177
1178         pause a sync job
1179
1180         -name      string
1181
1182                 name of the sync job, if not set it is default
1183
1184         -source    string
1185
1186                 the source can be an  <VMID> or [IP:]<ZFSPool>[/Path]
1187
1188 pve-zsync enable -source <string> [OPTIONS]
1189
1190         enable a syncjob and reset error
1191
1192         -name      string
1193
1194                 name of the sync job, if not set it is default
1195
1196         -source    string
1197
1198                 the source can be an  <VMID> or [IP:]<ZFSPool>[/Path]
1199 pve-zsync list
1200
1201         Get a List of all scheduled Sync Jobs
1202
1203 pve-zsync status
1204
1205         Get the status of all scheduled Sync Jobs
1206
1207 pve-zsync sync -dest <string> -source <string> [OPTIONS]
1208
1209         will sync one time
1210
1211         -dest      string
1212
1213                 the destination target is like [IP:]<Pool>[/Path]
1214
1215         -limit     integer
1216
1217                 max sync speed in kBytes/s, default unlimited
1218
1219         -maxsnap   integer
1220
1221                 how much snapshots will be kept before get erased, default 1
1222
1223         -name      string
1224
1225                 name of the sync job, if not set it is default.
1226                 It is only necessary if scheduler allready contains this source.
1227
1228         -source    string
1229
1230                 the source can be an <VMID> or [IP:]<ZFSPool>[/Path]
1231
1232 =head1 DESCRIPTION
1233
1234 This Tool helps you to sync your VM or directory which stored on ZFS between 2 servers.
1235 This tool also has the capability to add jobs to cron so the sync will be automatically done.
1236 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.
1237 To config cron see man crontab.
1238
1239 =head2 PVE ZFS Storage sync Tool
1240
1241 This Tool can get remote pool on other PVE or send Pool to others ZFS machines
1242
1243 =head1 EXAMPLES
1244
1245 add sync job from local VM to remote ZFS Server
1246 pve-zsync create -source=100 -dest=192.168.1.2:zfspool
1247
1248 =head1 IMPORTANT FILES
1249
1250 Cron jobs and config are stored at                      /etc/cron.d/pve-zsync
1251
1252 The VM config get copied on the destination machine to  /var/lib/pve-zsync/
1253
1254 =head1 COPYRIGHT AND DISCLAIMER
1255
1256 Copyright (C) 2007-2015 Proxmox Server Solutions GmbH
1257
1258 This program is free software: you can redistribute it and/or modify it
1259 under the terms of the GNU Affero General Public License as published
1260 by the Free Software Foundation, either version 3 of the License, or
1261 (at your option) any later version.
1262
1263 This program is distributed in the hope that it will be useful, but
1264 WITHOUT ANY WARRANTY; without even the implied warranty of
1265 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
1266 Affero General Public License for more details.
1267
1268 You should have received a copy of the GNU Affero General Public
1269 License along with this program.  If not, see
1270 <http://www.gnu.org/licenses/>.