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