]> git.proxmox.com Git - pve-guest-common.git/blobdiff - PVE/Replication.pm
Add config parameter 'source'.
[pve-guest-common.git] / PVE / Replication.pm
index 64432ebe609c1f82bd24641dd495e81e4b0b7073..4d1e071e30bc075c4cbb2e2f2d8e5740580cf86b 100644 (file)
@@ -5,6 +5,7 @@ use strict;
 use Data::Dumper;
 use JSON;
 use Time::HiRes qw(gettimeofday tv_interval);
+use POSIX qw(strftime);
 
 use PVE::INotify;
 use PVE::ProcFSTools;
@@ -19,7 +20,7 @@ use PVE::ReplicationState;
 # regression tests should overwrite this
 sub get_log_time {
 
-    return time();
+    return strftime("%F %H:%M:%S", localtime);
 }
 
 # Find common base replication snapshot, available on local and remote side.
@@ -135,8 +136,22 @@ sub prepare {
                $last_snapshots->{$volid}->{$snap} = 1;
            } elsif ($snap =~ m/^\Q$prefix\E/) {
                $logfunc->("delete stale replication snapshot '$snap' on $volid");
-               PVE::Storage::volume_snapshot_delete($storecfg, $volid, $snap);
-               $cleaned_replicated_volumes->{$volid} = 1;
+
+               eval {
+                   PVE::Storage::volume_snapshot_delete($storecfg, $volid, $snap);
+                   $cleaned_replicated_volumes->{$volid} = 1;
+               };
+
+               # If deleting the snapshot fails, we can not be sure if it was due to an error or a timeout.
+               # The likelihood that the delete has worked out is high at a timeout.
+               # If it really fails, it will try to remove on the next run.
+               if (my $err = $@) {
+                   # warn is for syslog/journal.
+                   warn $err;
+
+                   # logfunc will written in replication log.
+                   $logfunc->("delete stale replication snapshot error: $err");
+               }
            }
        }
     }
@@ -145,13 +160,13 @@ sub prepare {
 }
 
 sub replicate_volume {
-    my ($ssh_info, $storecfg, $volid, $base_snapshot, $sync_snapname, $rate, $insecure) = @_;
+    my ($ssh_info, $storecfg, $volid, $base_snapshot, $sync_snapname, $rate, $insecure, $logfunc) = @_;
 
     my ($storeid, $volname) = PVE::Storage::parse_volume_id($volid);
 
     my $ratelimit_bps = int(1000000*$rate) if $rate;
     PVE::Storage::storage_migrate($storecfg, $volid, $ssh_info, $storeid, $volname,
-                                 $base_snapshot, $sync_snapname, $ratelimit_bps, $insecure);
+                                 $base_snapshot, $sync_snapname, $ratelimit_bps, $insecure, 1, $logfunc);
 }
 
 
@@ -179,7 +194,6 @@ sub replicate {
        if $start_time <= $last_sync;
 
     my $vmid = $jobcfg->{guest};
-    my $vmtype = $jobcfg->{vmtype};
 
     my $conf = $guest_class->load_config($vmid);
     my ($running, $freezefs) = $guest_class->__snapshot_check_freeze_needed($vmid, $conf, 0);
@@ -189,7 +203,9 @@ sub replicate {
 
     $running //= 0;  # to avoid undef warnings from logfunc
 
-    $logfunc->("guest => $vmid, type => $vmtype, running => $running");
+    my $guest_name = $guest_class->guest_type() . ' ' . $vmid;
+
+    $logfunc->("guest => $guest_name, running => $running");
     $logfunc->("volumes => " . join(',', @$sorted_volids));
 
     if (my $remove_job = $jobcfg->{remove_job}) {
@@ -198,8 +214,13 @@ sub replicate {
 
        if ($remove_job eq 'full' && $jobcfg->{target} ne $local_node) {
            # remove all remote volumes
+           my @store_list = map { (PVE::Storage::parse_volume_id($_))[0] } @$sorted_volids;
+
+           my %hash = map { $_ => 1 } @store_list;
+
            my $ssh_info = PVE::Cluster::get_ssh_info($jobcfg->{target});
-           remote_prepare_local_job($ssh_info, $jobid, $vmid, [], $state->{storeid_list}, 0, undef, 1, $logfunc);
+
+           remote_prepare_local_job($ssh_info, $jobid, $vmid, [], [ keys %hash ], 1, undef, 1, $logfunc);
 
        }
        # remove all local replication snapshots (lastsync => 0)
@@ -245,8 +266,9 @@ sub replicate {
     };
     my $err = $@;
 
-    # unfreeze immediately
+    # thaw immediately
     if ($freezefs) {
+       $logfunc->("thaw guest filesystem");
        $guest_class->__snapshot_freeze($vmid, 1);
     }
 
@@ -278,12 +300,11 @@ sub replicate {
                $logfunc->("full sync '$volid' ($sync_snapname)");
            }
 
-           replicate_volume($ssh_info, $storecfg, $volid, $base_snapname, $sync_snapname, $rate, $insecure);
+           replicate_volume($ssh_info, $storecfg, $volid, $base_snapname, $sync_snapname, $rate, $insecure, $logfunc);
        }
     };
-    $err = $@;
 
-    if ($err) {
+    if ($err = $@) {
        $cleanup_local_snapshots->($replicate_snapshots, $sync_snapname); # try to cleanup
        # we do not cleanup the remote side here - this is done in
        # next run of prepare_local_job
@@ -293,15 +314,24 @@ sub replicate {
     # remove old snapshots because they are no longer needed
     $cleanup_local_snapshots->($last_snapshots, $last_sync_snapname);
 
-    remote_finalize_local_job($ssh_info, $jobid, $vmid, $sorted_volids, $start_time, $logfunc);
+    eval {
+       remote_finalize_local_job($ssh_info, $jobid, $vmid, $sorted_volids, $start_time, $logfunc);
+    };
+
+    # old snapshots will removed by next run from prepare_local_job.
+    if ($err = $@) {
+       # warn is for syslog/journal.
+       warn $err;
 
-    die $err if $err;
+       # logfunc will written in replication log.
+       $logfunc->("delete stale replication snapshot error: $err");
+    }
 
     return $volumes;
 }
 
 my $run_replication_nolock = sub {
-    my ($guest_class, $jobcfg, $iteration, $start_time, $logfunc) = @_;
+    my ($guest_class, $jobcfg, $iteration, $start_time, $logfunc, $verbose) = @_;
 
     my $jobid = $jobcfg->{id};
 
@@ -310,66 +340,66 @@ my $run_replication_nolock = sub {
     # we normaly write errors into the state file,
     # but we also catch unexpected errors and log them to syslog
     # (for examply when there are problems writing the state file)
-    eval {
-       my $state = PVE::ReplicationState::read_job_state($jobcfg);
 
-       PVE::ReplicationState::record_job_start($jobcfg, $state, $start_time, $iteration);
+    my $state = PVE::ReplicationState::read_job_state($jobcfg);
 
-       my $t0 = [gettimeofday];
+    PVE::ReplicationState::record_job_start($jobcfg, $state, $start_time, $iteration);
 
-       mkdir $PVE::ReplicationState::replicate_logdir;
-       my $logfile = PVE::ReplicationState::job_logfile_name($jobid);
-       open(my $logfd, '>', $logfile) ||
-           die "unable to open replication log '$logfile' - $!\n";
+    my $t0 = [gettimeofday];
 
-       my $logfunc_wrapper = sub {
-           my ($msg) = @_;
+    mkdir $PVE::ReplicationState::replicate_logdir;
+    my $logfile = PVE::ReplicationState::job_logfile_name($jobid);
+    open(my $logfd, '>', $logfile) ||
+       die "unable to open replication log '$logfile' - $!\n";
 
-           my $ctime = get_log_time();
-           print $logfd "$ctime $jobid: $msg\n";
-           $logfunc->("$ctime $jobid: $msg") if $logfunc;
-       };
+    my $logfunc_wrapper = sub {
+       my ($msg) = @_;
 
-       $logfunc_wrapper->("start replication job");
-
-       eval {
-           $volumes = replicate($guest_class, $jobcfg, $state, $start_time, $logfunc_wrapper);
-       };
-       my $err = $@;
-
-       if ($err) {
-           chomp $err;
-           $logfunc_wrapper->("end replication job with error: $err");
-       } else {
-           $logfunc_wrapper->("end replication job");
+       my $ctime = get_log_time();
+       print $logfd "$ctime $jobid: $msg\n";
+       if ($logfunc) {
+           if ($verbose) {
+               $logfunc->("$ctime $jobid: $msg");
+           } else {
+               $logfunc->($msg);
+           }
        }
+    };
 
-       PVE::ReplicationState::record_job_end($jobcfg, $state, $start_time, tv_interval($t0), $err);
+    $logfunc_wrapper->("start replication job");
 
-       close($logfd);
+    eval {
+       $volumes = replicate($guest_class, $jobcfg, $state, $start_time, $logfunc_wrapper);
     };
-    if (my $err = $@) {
-       warn "$jobid: got unexpected replication job error - $err";
+    my $err = $@;
+
+    if ($err) {
+       my $msg = "end replication job with error: $err";
+       chomp $msg;
+       $logfunc_wrapper->($msg);
+    } else {
+       $logfunc_wrapper->("end replication job");
     }
 
+    PVE::ReplicationState::record_job_end($jobcfg, $state, $start_time, tv_interval($t0), $err);
+
+    close($logfd);
+
+    die $err if $err;
+
     return $volumes;
 };
 
 sub run_replication {
-    my ($guest_class, $jobcfg, $iteration, $start_time, $logfunc, $noerr) = @_;
+    my ($guest_class, $jobcfg, $iteration, $start_time, $logfunc, $verbose) = @_;
 
     my $volumes;
 
-    eval {
-       my $timeout = 2; # do not wait too long - we repeat periodically anyways
-       $volumes = PVE::GuestHelpers::guest_migration_lock(
-           $jobcfg->{guest}, $timeout, $run_replication_nolock,
-           $guest_class, $jobcfg, $iteration, $start_time, $logfunc);
-    };
-    if (my $err = $@) {
-       return undef if $noerr;
-       die $err;
-    }
+    my $timeout = 2; # do not wait too long - we repeat periodically anyways
+    $volumes = PVE::GuestHelpers::guest_migration_lock(
+       $jobcfg->{guest}, $timeout, $run_replication_nolock,
+       $guest_class, $jobcfg, $iteration, $start_time, $logfunc, $verbose);
+
     return $volumes;
 }