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