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