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