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