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