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