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