PVE/Replication.pm

   1 package PVE::Replication;
   2
   3 use warnings;
   4 use strict;
   5 use Data::Dumper;
   6 use JSON;
   7 use Time::HiRes qw(gettimeofday tv_interval);
   8 use POSIX qw(strftime);
   9
  10 use PVE::INotify;
  11 use PVE::ProcFSTools;
  12 use PVE::Tools;
  13 use PVE::Cluster;
  14 use PVE::DataCenterConfig;
  15 use PVE::Storage;
  16 use PVE::GuestHelpers;
  17 use PVE::ReplicationConfig;
  18 use PVE::ReplicationState;
  19 use PVE::SSHInfo;
  20
  21
  22 # regression tests should overwrite this
  23 sub get_log_time {
  24
  25     return strftime("%F %H:%M:%S", localtime);
  26 }
  27
  28 # Find common base replication snapshot, available on local and remote side.
  29 # Note: this also removes stale replication snapshots
  30 sub find_common_replication_snapshot {
  31     my ($ssh_info, $jobid, $vmid, $storecfg, $volumes, $storeid_list, $last_sync, $parent_snapname, $logfunc) = @_;
  32
  33     my $last_sync_snapname =
  34         PVE::ReplicationState::replication_snapshot_name($jobid, $last_sync);
  35
  36     # test if we have a replication_ snapshot from last sync
  37     # and remove all other/stale replication snapshots
  38
  39     my $last_snapshots = prepare(
  40         $storecfg, $volumes, $jobid, $last_sync, $parent_snapname, $logfunc);
  41
  42     # prepare remote side
  43     my $remote_snapshots = remote_prepare_local_job(
  44         $ssh_info, $jobid, $vmid, $volumes, $storeid_list, $last_sync, $parent_snapname, 0, $logfunc);
  45
  46     my $base_snapshots = {};
  47
  48     foreach my $volid (@$volumes) {
  49         my $base_snapname;
  50
  51         if (defined($last_snapshots->{$volid}) && defined($remote_snapshots->{$volid})) {
  52             if ($last_snapshots->{$volid}->{$last_sync_snapname} &&
  53                 $remote_snapshots->{$volid}->{$last_sync_snapname}) {
  54                 $base_snapshots->{$volid} = $last_sync_snapname;
  55             } elsif (defined($parent_snapname) &&
  56                      ($last_snapshots->{$volid}->{$parent_snapname} &&
  57                       $remote_snapshots->{$volid}->{$parent_snapname})) {
  58                 $base_snapshots->{$volid} = $parent_snapname;
  59             } elsif ($last_sync == 0) {
  60                 my @desc_sorted_snap =
  61                     map { $_->[1] } sort { $b->[0] <=> $a->[0] }
  62                     map { [ ($_ =~ /__replicate_\Q$jobid\E_(\d+)_/)[0] || 0, $_ ] }
  63                     keys %{$remote_snapshots->{$volid}};
  64
  65                 foreach my $remote_snap (@desc_sorted_snap) {
  66                     if (defined($last_snapshots->{$volid}->{$remote_snap})) {
  67                         $base_snapshots->{$volid} = $remote_snap;
  68                         last;
  69                     }
  70                 }
  71                 die "No common base to restore the job state\n".
  72                     "please delete jobid: $jobid and create the job again\n"
  73                     if !defined($base_snapshots->{$volid});
  74             }
  75         }
  76     }
  77
  78     return ($base_snapshots, $last_snapshots, $last_sync_snapname);
  79 }
  80
  81 sub remote_prepare_local_job {
  82     my ($ssh_info, $jobid, $vmid, $volumes, $storeid_list, $last_sync, $parent_snapname, $force, $logfunc) = @_;
  83
  84     my $ssh_cmd = PVE::SSHInfo::ssh_info_to_command($ssh_info);
  85     my $cmd = [@$ssh_cmd, '--', 'pvesr', 'prepare-local-job', $jobid];
  86     push @$cmd, '--scan', join(',', @$storeid_list) if scalar(@$storeid_list);
  87     push @$cmd, @$volumes if scalar(@$volumes);
  88
  89     push @$cmd, '--last_sync', $last_sync;
  90     push @$cmd, '--parent_snapname', $parent_snapname
  91         if $parent_snapname;
  92     push @$cmd, '--force' if $force;
  93
  94     my $remote_snapshots;
  95
  96     my $parser = sub {
  97         my $line = shift;
  98         $remote_snapshots = JSON::decode_json($line);
  99     };
 100
 101     my $logger = sub {
 102         my $line = shift;
 103         chomp $line;
 104         $logfunc->("(remote_prepare_local_job) $line");
 105     };
 106
 107     PVE::Tools::run_command($cmd, outfunc => $parser, errfunc => $logger);
 108
 109     die "prepare remote node failed - no result\n"
 110         if !defined($remote_snapshots);
 111
 112     return $remote_snapshots;
 113 }
 114
 115 sub remote_finalize_local_job {
 116     my ($ssh_info, $jobid, $vmid, $volumes, $last_sync, $logfunc) = @_;
 117
 118     my $ssh_cmd = PVE::SSHInfo::ssh_info_to_command($ssh_info);
 119     my $cmd = [@$ssh_cmd, '--', 'pvesr', 'finalize-local-job', $jobid,
 120                @$volumes, '--last_sync', $last_sync];
 121
 122     my $logger = sub {
 123         my $line = shift;
 124         chomp $line;
 125         $logfunc->("(remote_finalize_local_job) $line");
 126     };
 127
 128     PVE::Tools::run_command($cmd, outfunc => $logger, errfunc => $logger);
 129 }
 130
 131 # finds local replication snapshots from $last_sync
 132 # and removes all replication snapshots with other time stamps
 133 sub prepare {
 134     my ($storecfg, $volids, $jobid, $last_sync, $parent_snapname, $logfunc) = @_;
 135
 136     $last_sync //= 0;
 137
 138     my ($prefix, $snapname);
 139
 140     if (defined($jobid)) {
 141         ($prefix, $snapname) = PVE::ReplicationState::replication_snapshot_name($jobid, $last_sync);
 142     } else {
 143         $prefix = '__replicate_';
 144     }
 145
 146     my $last_snapshots = {};
 147     my $cleaned_replicated_volumes = {};
 148     foreach my $volid (@$volids) {
 149         my $list = PVE::Storage::volume_snapshot_list($storecfg, $volid);
 150         foreach my $snap (@$list) {
 151             if ((defined($snapname) && ($snap eq $snapname)) ||
 152                 (defined($parent_snapname) && ($snap eq $parent_snapname))) {
 153                 $last_snapshots->{$volid}->{$snap} = 1;
 154             } elsif ($snap =~ m/^\Q$prefix\E/) {
 155                 if ($last_sync != 0) {
 156                     $logfunc->("delete stale replication snapshot '$snap' on $volid");
 157                     eval {
 158                         PVE::Storage::volume_snapshot_delete($storecfg, $volid, $snap);
 159                         $cleaned_replicated_volumes->{$volid} = 1;
 160                     };
 161
 162                     # If deleting the snapshot fails, we can not be sure if it was due to an error or a timeout.
 163                     # The likelihood that the delete has worked out is high at a timeout.
 164                     # If it really fails, it will try to remove on the next run.
 165                     if (my $err = $@) {
 166                         # warn is for syslog/journal.
 167                         warn $err;
 168
 169                         # logfunc will written in replication log.
 170                         $logfunc->("delete stale replication snapshot error: $err");
 171                     }
 172                 # Last_sync=0 and a replication snapshot only occur, if the VM was stolen
 173                 } else {
 174                     $last_snapshots->{$volid}->{$snap} = 1;
 175                 }
 176             }
 177         }
 178     }
 179
 180     return wantarray ? ($last_snapshots, $cleaned_replicated_volumes) : $last_snapshots;
 181 }
 182
 183 sub replicate_volume {
 184     my ($ssh_info, $storecfg, $volid, $base_snapshot, $sync_snapname, $rate, $insecure, $logfunc) = @_;
 185
 186     my ($storeid, $volname) = PVE::Storage::parse_volume_id($volid);
 187
 188     my $ratelimit_bps = $rate ? int(1000000 * $rate) : undef;
 189
 190     my $opts = {
 191         'target_volname' => $volname,
 192         'base_snapshot' => $base_snapshot,
 193         'snapshot' => $sync_snapname,
 194         'ratelimit_bps' => $ratelimit_bps,
 195         'insecure' => $insecure,
 196         'with_snapshots' => 1,
 197     };
 198
 199     PVE::Storage::storage_migrate($storecfg, $volid, $ssh_info, $storeid, $opts, $logfunc);
 200 }
 201
 202
 203 sub replicate {
 204     my ($guest_class, $jobcfg, $state, $start_time, $logfunc) = @_;
 205
 206     my $local_node = PVE::INotify::nodename();
 207
 208     die "not implemented - internal error" if $jobcfg->{type} ne 'local';
 209
 210     my $dc_conf = PVE::Cluster::cfs_read_file('datacenter.cfg');
 211
 212     my $migration_network;
 213     my $migration_type = 'secure';
 214     if (my $mc = $dc_conf->{migration}) {
 215         $migration_network = $mc->{network};
 216         $migration_type = $mc->{type} if defined($mc->{type});
 217     }
 218
 219     my $jobid = $jobcfg->{id};
 220     my $storecfg = PVE::Storage::config();
 221     my $last_sync = $state->{last_sync};
 222
 223     die "start time before last sync ($start_time <= $last_sync) - abort sync\n"
 224         if $start_time <= $last_sync;
 225
 226     my $vmid = $jobcfg->{guest};
 227
 228     my $conf = $guest_class->load_config($vmid);
 229     my ($running, $freezefs) = $guest_class->__snapshot_check_freeze_needed($vmid, $conf, 0);
 230     my $volumes = $guest_class->get_replicatable_volumes($storecfg, $vmid, $conf, defined($jobcfg->{remove_job}));
 231
 232     my $sorted_volids = [ sort keys %$volumes ];
 233
 234     $running //= 0;  # to avoid undef warnings from logfunc
 235
 236     my $guest_name = $guest_class->guest_type() . ' ' . $vmid;
 237
 238     $logfunc->("guest => $guest_name, running => $running");
 239     $logfunc->("volumes => " . join(',', @$sorted_volids));
 240
 241     if (my $remove_job = $jobcfg->{remove_job}) {
 242
 243         $logfunc->("start job removal - mode '${remove_job}'");
 244
 245         if ($remove_job eq 'full' && $jobcfg->{target} ne $local_node) {
 246             # remove all remote volumes
 247             my @store_list = map { (PVE::Storage::parse_volume_id($_))[0] } @$sorted_volids;
 248
 249             my %hash = map { $_ => 1 } @store_list;
 250
 251             my $ssh_info = PVE::SSHInfo::get_ssh_info($jobcfg->{target});
 252             remote_prepare_local_job($ssh_info, $jobid, $vmid, [], [ keys %hash ], 1, undef, 1, $logfunc);
 253
 254         }
 255         # remove all local replication snapshots (lastsync => 0)
 256         prepare($storecfg, $sorted_volids, $jobid, 1, undef, $logfunc);
 257
 258         PVE::ReplicationConfig::delete_job($jobid); # update config
 259         $logfunc->("job removed");
 260
 261         return undef;
 262     }
 263
 264     my $ssh_info = PVE::SSHInfo::get_ssh_info($jobcfg->{target}, $migration_network);
 265
 266     my $parent_snapname = $conf->{parent};
 267
 268     my ($base_snapshots, $last_snapshots, $last_sync_snapname) = find_common_replication_snapshot(
 269         $ssh_info, $jobid, $vmid, $storecfg, $sorted_volids, $state->{storeid_list}, $last_sync, $parent_snapname, $logfunc);
 270
 271     my $storeid_hash = {};
 272     foreach my $volid (@$sorted_volids) {
 273         my ($storeid) = PVE::Storage::parse_volume_id($volid);
 274         $storeid_hash->{$storeid} = 1;
 275     }
 276     $state->{storeid_list} = [ sort keys %$storeid_hash ];
 277
 278     # freeze filesystem for data consistency
 279     if ($freezefs) {
 280         $logfunc->("freeze guest filesystem");
 281         $guest_class->__snapshot_freeze($vmid, 0);
 282     }
 283
 284     # make snapshot of all volumes
 285     my $sync_snapname =
 286         PVE::ReplicationState::replication_snapshot_name($jobid, $start_time);
 287
 288     my $replicate_snapshots = {};
 289     eval {
 290         foreach my $volid (@$sorted_volids) {
 291             $logfunc->("create snapshot '${sync_snapname}' on $volid");
 292             PVE::Storage::volume_snapshot($storecfg, $volid, $sync_snapname);
 293             $replicate_snapshots->{$volid} = 1;
 294         }
 295     };
 296     my $err = $@;
 297
 298     # thaw immediately
 299     if ($freezefs) {
 300         $logfunc->("thaw guest filesystem");
 301         $guest_class->__snapshot_freeze($vmid, 1);
 302     }
 303
 304     my $cleanup_local_snapshots = sub {
 305         my ($volid_hash, $snapname) = @_;
 306         foreach my $volid (sort keys %$volid_hash) {
 307             $logfunc->("delete previous replication snapshot '$snapname' on $volid");
 308             eval { PVE::Storage::volume_snapshot_delete($storecfg, $volid, $snapname); };
 309             warn $@ if $@;
 310         }
 311     };
 312
 313     if ($err) {
 314         $cleanup_local_snapshots->($replicate_snapshots, $sync_snapname); # try to cleanup
 315         die $err;
 316     }
 317
 318     eval {
 319
 320         my $rate = $jobcfg->{rate};
 321         my $insecure = $migration_type eq 'insecure';
 322
 323         $logfunc->("using $migration_type transmission, rate limit: "
 324             . ($rate ? "$rate MByte/s" : "none"));
 325
 326         foreach my $volid (@$sorted_volids) {
 327             my $base_snapname;
 328
 329             if (defined($base_snapname = $base_snapshots->{$volid})) {
 330                 $logfunc->("incremental sync '$volid' ($base_snapname => $sync_snapname)");
 331             } else {
 332                 $logfunc->("full sync '$volid' ($sync_snapname)");
 333             }
 334
 335             replicate_volume($ssh_info, $storecfg, $volid, $base_snapname, $sync_snapname, $rate, $insecure, $logfunc);
 336         }
 337     };
 338
 339     if ($err = $@) {
 340         $cleanup_local_snapshots->($replicate_snapshots, $sync_snapname); # try to cleanup
 341         # we do not cleanup the remote side here - this is done in
 342         # next run of prepare_local_job
 343         die $err;
 344     }
 345
 346     # remove old snapshots because they are no longer needed
 347     $cleanup_local_snapshots->($last_snapshots, $last_sync_snapname);
 348
 349     eval {
 350         remote_finalize_local_job($ssh_info, $jobid, $vmid, $sorted_volids, $start_time, $logfunc);
 351     };
 352
 353     # old snapshots will removed by next run from prepare_local_job.
 354     if ($err = $@) {
 355         # warn is for syslog/journal.
 356         warn $err;
 357
 358         # logfunc will written in replication log.
 359         $logfunc->("delete stale replication snapshot error: $err");
 360     }
 361
 362     return $volumes;
 363 }
 364
 365 my $run_replication_nolock = sub {
 366     my ($guest_class, $jobcfg, $iteration, $start_time, $logfunc, $verbose) = @_;
 367
 368     my $jobid = $jobcfg->{id};
 369
 370     my $volumes;
 371
 372     # we normally write errors into the state file,
 373     # but we also catch unexpected errors and log them to syslog
 374     # (for examply when there are problems writing the state file)
 375
 376     my $state = PVE::ReplicationState::read_job_state($jobcfg);
 377
 378     PVE::ReplicationState::record_job_start($jobcfg, $state, $start_time, $iteration);
 379
 380     my $t0 = [gettimeofday];
 381
 382     mkdir $PVE::ReplicationState::replicate_logdir;
 383     my $logfile = PVE::ReplicationState::job_logfile_name($jobid);
 384     open(my $logfd, '>', $logfile) ||
 385         die "unable to open replication log '$logfile' - $!\n";
 386
 387     my $logfunc_wrapper = sub {
 388         my ($msg) = @_;
 389
 390         my $ctime = get_log_time();
 391         print $logfd "$ctime $jobid: $msg\n";
 392         if ($logfunc) {
 393             if ($verbose) {
 394                 $logfunc->("$ctime $jobid: $msg");
 395             } else {
 396                 $logfunc->($msg);
 397             }
 398         }
 399     };
 400
 401     $logfunc_wrapper->("start replication job");
 402
 403     eval {
 404         $volumes = replicate($guest_class, $jobcfg, $state, $start_time, $logfunc_wrapper);
 405     };
 406     my $err = $@;
 407
 408     if ($err) {
 409         my $msg = "end replication job with error: $err";
 410         chomp $msg;
 411         $logfunc_wrapper->($msg);
 412     } else {
 413         $logfunc_wrapper->("end replication job");
 414     }
 415
 416     PVE::ReplicationState::record_job_end($jobcfg, $state, $start_time, tv_interval($t0), $err);
 417
 418     close($logfd);
 419
 420     die $err if $err;
 421
 422     return $volumes;
 423 };
 424
 425 sub run_replication {
 426     my ($guest_class, $jobcfg, $iteration, $start_time, $logfunc, $verbose) = @_;
 427
 428     my $volumes;
 429
 430     my $timeout = 2; # do not wait too long - we repeat periodically anyways
 431     $volumes = PVE::GuestHelpers::guest_migration_lock(
 432         $jobcfg->{guest}, $timeout, $run_replication_nolock,
 433         $guest_class, $jobcfg, $iteration, $start_time, $logfunc, $verbose);
 434
 435     return $volumes;
 436 }
 437
 438 1;