1 package PVE
::Replication
;
7 use Time
::HiRes
qw(gettimeofday tv_interval);
8 use POSIX
qw(strftime);
15 use PVE
::GuestHelpers
;
16 use PVE
::ReplicationConfig
;
17 use PVE
::ReplicationState
;
20 # regression tests should overwrite this
23 return strftime
("%F %H:%M:%S", localtime);
26 # Find common base replication snapshot, available on local and remote side.
27 # Note: this also removes stale replication snapshots
28 sub find_common_replication_snapshot
{
29 my ($ssh_info, $jobid, $vmid, $storecfg, $volumes, $storeid_list, $last_sync, $parent_snapname, $logfunc) = @_;
31 my $last_sync_snapname =
32 PVE
::ReplicationState
::replication_snapshot_name
($jobid, $last_sync);
34 # test if we have a replication_ snapshot from last sync
35 # and remove all other/stale replication snapshots
37 my $last_snapshots = prepare
(
38 $storecfg, $volumes, $jobid, $last_sync, $parent_snapname, $logfunc);
41 my $remote_snapshots = remote_prepare_local_job
(
42 $ssh_info, $jobid, $vmid, $volumes, $storeid_list, $last_sync, $parent_snapname, 0, $logfunc);
44 my $base_snapshots = {};
46 foreach my $volid (@$volumes) {
49 if (defined($last_snapshots->{$volid}) && defined($remote_snapshots->{$volid})) {
50 if ($last_snapshots->{$volid}->{$last_sync_snapname} &&
51 $remote_snapshots->{$volid}->{$last_sync_snapname}) {
52 $base_snapshots->{$volid} = $last_sync_snapname;
53 } elsif (defined($parent_snapname) &&
54 ($last_snapshots->{$volid}->{$parent_snapname} &&
55 $remote_snapshots->{$volid}->{$parent_snapname})) {
56 $base_snapshots->{$volid} = $parent_snapname;
57 } elsif ($last_sync == 0) {
58 my @desc_sorted_snap =
59 map { $_->[1] } sort { $b->[0] <=> $a->[0] }
60 map { [ ($_ =~ /__replicate_\Q$jobid\E_(\d+)_/)[0] || 0, $_ ] }
61 keys %{$remote_snapshots->{$volid}};
63 foreach my $remote_snap (@desc_sorted_snap) {
64 if (defined($last_snapshots->{$volid}->{$remote_snap})) {
65 $base_snapshots->{$volid} = $remote_snap;
69 die "No common base to restore the job state\n".
70 "please delete jobid: $jobid and create the job again\n"
71 if !defined($base_snapshots->{$volid});
76 return ($base_snapshots, $last_snapshots, $last_sync_snapname);
79 sub remote_prepare_local_job
{
80 my ($ssh_info, $jobid, $vmid, $volumes, $storeid_list, $last_sync, $parent_snapname, $force, $logfunc) = @_;
82 my $ssh_cmd = PVE
::Cluster
::ssh_info_to_command
($ssh_info);
83 my $cmd = [@$ssh_cmd, '--', 'pvesr', 'prepare-local-job', $jobid];
84 push @$cmd, '--scan', join(',', @$storeid_list) if scalar(@$storeid_list);
85 push @$cmd, @$volumes if scalar(@$volumes);
87 push @$cmd, '--last_sync', $last_sync;
88 push @$cmd, '--parent_snapname', $parent_snapname
90 push @$cmd, '--force' if $force;
96 $remote_snapshots = JSON
::decode_json
($line);
102 $logfunc->("(remote_prepare_local_job) $line");
105 PVE
::Tools
::run_command
($cmd, outfunc
=> $parser, errfunc
=> $logger);
107 die "prepare remote node failed - no result\n"
108 if !defined($remote_snapshots);
110 return $remote_snapshots;
113 sub remote_finalize_local_job
{
114 my ($ssh_info, $jobid, $vmid, $volumes, $last_sync, $logfunc) = @_;
116 my $ssh_cmd = PVE
::Cluster
::ssh_info_to_command
($ssh_info);
117 my $cmd = [@$ssh_cmd, '--', 'pvesr', 'finalize-local-job', $jobid,
118 @$volumes, '--last_sync', $last_sync];
123 $logfunc->("(remote_finalize_local_job) $line");
126 PVE
::Tools
::run_command
($cmd, outfunc
=> $logger, errfunc
=> $logger);
129 # finds local replication snapshots from $last_sync
130 # and removes all replication snapshots with other time stamps
132 my ($storecfg, $volids, $jobid, $last_sync, $parent_snapname, $logfunc) = @_;
136 my ($prefix, $snapname);
138 if (defined($jobid)) {
139 ($prefix, $snapname) = PVE
::ReplicationState
::replication_snapshot_name
($jobid, $last_sync);
141 $prefix = '__replicate_';
144 my $last_snapshots = {};
145 my $cleaned_replicated_volumes = {};
146 foreach my $volid (@$volids) {
147 my $list = PVE
::Storage
::volume_snapshot_list
($storecfg, $volid);
148 foreach my $snap (@$list) {
149 if ((defined($snapname) && ($snap eq $snapname)) ||
150 (defined($parent_snapname) && ($snap eq $parent_snapname))) {
151 $last_snapshots->{$volid}->{$snap} = 1;
152 } elsif ($snap =~ m/^\Q$prefix\E/) {
153 if ($last_sync != 0) {
154 $logfunc->("delete stale replication snapshot '$snap' on $volid");
156 PVE
::Storage
::volume_snapshot_delete
($storecfg, $volid, $snap);
157 $cleaned_replicated_volumes->{$volid} = 1;
160 # If deleting the snapshot fails, we can not be sure if it was due to an error or a timeout.
161 # The likelihood that the delete has worked out is high at a timeout.
162 # If it really fails, it will try to remove on the next run.
164 # warn is for syslog/journal.
167 # logfunc will written in replication log.
168 $logfunc->("delete stale replication snapshot error: $err");
170 # Last_sync=0 and a replication snapshot only occur, if the VM was stolen
172 $last_snapshots->{$volid}->{$snap} = 1;
178 return wantarray ?
($last_snapshots, $cleaned_replicated_volumes) : $last_snapshots;
181 sub replicate_volume
{
182 my ($ssh_info, $storecfg, $volid, $base_snapshot, $sync_snapname, $rate, $insecure, $logfunc) = @_;
184 my ($storeid, $volname) = PVE
::Storage
::parse_volume_id
($volid);
186 my $ratelimit_bps = int(1000000*$rate) if $rate;
187 PVE
::Storage
::storage_migrate
($storecfg, $volid, $ssh_info, $storeid, $volname,
188 $base_snapshot, $sync_snapname, $ratelimit_bps, $insecure, 1, $logfunc);
193 my ($guest_class, $jobcfg, $state, $start_time, $logfunc) = @_;
195 my $local_node = PVE
::INotify
::nodename
();
197 die "not implemented - internal error" if $jobcfg->{type
} ne 'local';
199 my $dc_conf = PVE
::Cluster
::cfs_read_file
('datacenter.cfg');
201 my $migration_network;
202 my $migration_type = 'secure';
203 if (my $mc = $dc_conf->{migration
}) {
204 $migration_network = $mc->{network
};
205 $migration_type = $mc->{type
} if defined($mc->{type
});
208 my $jobid = $jobcfg->{id
};
209 my $storecfg = PVE
::Storage
::config
();
210 my $last_sync = $state->{last_sync
};
212 die "start time before last sync ($start_time <= $last_sync) - abort sync\n"
213 if $start_time <= $last_sync;
215 my $vmid = $jobcfg->{guest
};
217 my $conf = $guest_class->load_config($vmid);
218 my ($running, $freezefs) = $guest_class->__snapshot_check_freeze_needed($vmid, $conf, 0);
219 my $volumes = $guest_class->get_replicatable_volumes($storecfg, $vmid, $conf, defined($jobcfg->{remove_job
}));
221 my $sorted_volids = [ sort keys %$volumes ];
223 $running //= 0; # to avoid undef warnings from logfunc
225 my $guest_name = $guest_class->guest_type() . ' ' . $vmid;
227 $logfunc->("guest => $guest_name, running => $running");
228 $logfunc->("volumes => " . join(',', @$sorted_volids));
230 if (my $remove_job = $jobcfg->{remove_job
}) {
232 $logfunc->("start job removal - mode '${remove_job}'");
234 if ($remove_job eq 'full' && $jobcfg->{target
} ne $local_node) {
235 # remove all remote volumes
236 my @store_list = map { (PVE
::Storage
::parse_volume_id
($_))[0] } @$sorted_volids;
238 my %hash = map { $_ => 1 } @store_list;
240 my $ssh_info = PVE
::Cluster
::get_ssh_info
($jobcfg->{target
});
241 remote_prepare_local_job
($ssh_info, $jobid, $vmid, [], [ keys %hash ], 1, undef, 1, $logfunc);
244 # remove all local replication snapshots (lastsync => 0)
245 prepare
($storecfg, $sorted_volids, $jobid, 1, undef, $logfunc);
247 PVE
::ReplicationConfig
::delete_job
($jobid); # update config
248 $logfunc->("job removed");
253 my $ssh_info = PVE
::Cluster
::get_ssh_info
($jobcfg->{target
}, $migration_network);
255 my $parent_snapname = $conf->{parent
};
257 my ($base_snapshots, $last_snapshots, $last_sync_snapname) = find_common_replication_snapshot
(
258 $ssh_info, $jobid, $vmid, $storecfg, $sorted_volids, $state->{storeid_list
}, $last_sync, $parent_snapname, $logfunc);
260 my $storeid_hash = {};
261 foreach my $volid (@$sorted_volids) {
262 my ($storeid) = PVE
::Storage
::parse_volume_id
($volid);
263 $storeid_hash->{$storeid} = 1;
265 $state->{storeid_list
} = [ sort keys %$storeid_hash ];
267 # freeze filesystem for data consistency
269 $logfunc->("freeze guest filesystem");
270 $guest_class->__snapshot_freeze($vmid, 0);
273 # make snapshot of all volumes
275 PVE
::ReplicationState
::replication_snapshot_name
($jobid, $start_time);
277 my $replicate_snapshots = {};
279 foreach my $volid (@$sorted_volids) {
280 $logfunc->("create snapshot '${sync_snapname}' on $volid");
281 PVE
::Storage
::volume_snapshot
($storecfg, $volid, $sync_snapname);
282 $replicate_snapshots->{$volid} = 1;
289 $logfunc->("thaw guest filesystem");
290 $guest_class->__snapshot_freeze($vmid, 1);
293 my $cleanup_local_snapshots = sub {
294 my ($volid_hash, $snapname) = @_;
295 foreach my $volid (sort keys %$volid_hash) {
296 $logfunc->("delete previous replication snapshot '$snapname' on $volid");
297 eval { PVE
::Storage
::volume_snapshot_delete
($storecfg, $volid, $snapname); };
303 $cleanup_local_snapshots->($replicate_snapshots, $sync_snapname); # try to cleanup
309 my $rate = $jobcfg->{rate
};
310 my $insecure = $migration_type eq 'insecure';
312 foreach my $volid (@$sorted_volids) {
315 if (defined($base_snapname = $base_snapshots->{$volid})) {
316 $logfunc->("incremental sync '$volid' ($base_snapname => $sync_snapname)");
318 $logfunc->("full sync '$volid' ($sync_snapname)");
321 replicate_volume
($ssh_info, $storecfg, $volid, $base_snapname, $sync_snapname, $rate, $insecure, $logfunc);
326 $cleanup_local_snapshots->($replicate_snapshots, $sync_snapname); # try to cleanup
327 # we do not cleanup the remote side here - this is done in
328 # next run of prepare_local_job
332 # remove old snapshots because they are no longer needed
333 $cleanup_local_snapshots->($last_snapshots, $last_sync_snapname);
336 remote_finalize_local_job
($ssh_info, $jobid, $vmid, $sorted_volids, $start_time, $logfunc);
339 # old snapshots will removed by next run from prepare_local_job.
341 # warn is for syslog/journal.
344 # logfunc will written in replication log.
345 $logfunc->("delete stale replication snapshot error: $err");
351 my $run_replication_nolock = sub {
352 my ($guest_class, $jobcfg, $iteration, $start_time, $logfunc, $verbose) = @_;
354 my $jobid = $jobcfg->{id
};
358 # we normaly write errors into the state file,
359 # but we also catch unexpected errors and log them to syslog
360 # (for examply when there are problems writing the state file)
362 my $state = PVE
::ReplicationState
::read_job_state
($jobcfg);
364 PVE
::ReplicationState
::record_job_start
($jobcfg, $state, $start_time, $iteration);
366 my $t0 = [gettimeofday
];
368 mkdir $PVE::ReplicationState
::replicate_logdir
;
369 my $logfile = PVE
::ReplicationState
::job_logfile_name
($jobid);
370 open(my $logfd, '>', $logfile) ||
371 die "unable to open replication log '$logfile' - $!\n";
373 my $logfunc_wrapper = sub {
376 my $ctime = get_log_time
();
377 print $logfd "$ctime $jobid: $msg\n";
380 $logfunc->("$ctime $jobid: $msg");
387 $logfunc_wrapper->("start replication job");
390 $volumes = replicate
($guest_class, $jobcfg, $state, $start_time, $logfunc_wrapper);
395 my $msg = "end replication job with error: $err";
397 $logfunc_wrapper->($msg);
399 $logfunc_wrapper->("end replication job");
402 PVE
::ReplicationState
::record_job_end
($jobcfg, $state, $start_time, tv_interval
($t0), $err);
411 sub run_replication
{
412 my ($guest_class, $jobcfg, $iteration, $start_time, $logfunc, $verbose) = @_;
416 my $timeout = 2; # do not wait too long - we repeat periodically anyways
417 $volumes = PVE
::GuestHelpers
::guest_migration_lock
(
418 $jobcfg->{guest
}, $timeout, $run_replication_nolock,
419 $guest_class, $jobcfg, $iteration, $start_time, $logfunc, $verbose);