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