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