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