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     PVE::Storage::storage_migrate($storecfg, $volid, $ssh_info, $storeid, $volname,
 191                                   $base_snapshot, $sync_snapname, $ratelimit_bps, $insecure, 1, $logfunc);
 192 }
 193
 194
 195 sub replicate {
 196     my ($guest_class, $jobcfg, $state, $start_time, $logfunc) = @_;
 197
 198     my $local_node = PVE::INotify::nodename();
 199
 200     die "not implemented - internal error" if $jobcfg->{type} ne 'local';
 201
 202     my $dc_conf = PVE::Cluster::cfs_read_file('datacenter.cfg');
 203
 204     my $migration_network;
 205     my $migration_type = 'secure';
 206     if (my $mc = $dc_conf->{migration}) {
 207         $migration_network = $mc->{network};
 208         $migration_type = $mc->{type} if defined($mc->{type});
 209     }
 210
 211     my $jobid = $jobcfg->{id};
 212     my $storecfg = PVE::Storage::config();
 213     my $last_sync = $state->{last_sync};
 214
 215     die "start time before last sync ($start_time <= $last_sync) - abort sync\n"
 216         if $start_time <= $last_sync;
 217
 218     my $vmid = $jobcfg->{guest};
 219
 220     my $conf = $guest_class->load_config($vmid);
 221     my ($running, $freezefs) = $guest_class->__snapshot_check_freeze_needed($vmid, $conf, 0);
 222     my $volumes = $guest_class->get_replicatable_volumes($storecfg, $vmid, $conf, defined($jobcfg->{remove_job}));
 223
 224     my $sorted_volids = [ sort keys %$volumes ];
 225
 226     $running //= 0;  # to avoid undef warnings from logfunc
 227
 228     my $guest_name = $guest_class->guest_type() . ' ' . $vmid;
 229
 230     $logfunc->("guest => $guest_name, running => $running");
 231     $logfunc->("volumes => " . join(',', @$sorted_volids));
 232
 233     if (my $remove_job = $jobcfg->{remove_job}) {
 234
 235         $logfunc->("start job removal - mode '${remove_job}'");
 236
 237         if ($remove_job eq 'full' && $jobcfg->{target} ne $local_node) {
 238             # remove all remote volumes
 239             my @store_list = map { (PVE::Storage::parse_volume_id($_))[0] } @$sorted_volids;
 240
 241             my %hash = map { $_ => 1 } @store_list;
 242
 243             my $ssh_info = PVE::SSHInfo::get_ssh_info($jobcfg->{target});
 244             remote_prepare_local_job($ssh_info, $jobid, $vmid, [], [ keys %hash ], 1, undef, 1, $logfunc);
 245
 246         }
 247         # remove all local replication snapshots (lastsync => 0)
 248         prepare($storecfg, $sorted_volids, $jobid, 1, undef, $logfunc);
 249
 250         PVE::ReplicationConfig::delete_job($jobid); # update config
 251         $logfunc->("job removed");
 252
 253         return undef;
 254     }
 255
 256     my $ssh_info = PVE::SSHInfo::get_ssh_info($jobcfg->{target}, $migration_network);
 257
 258     my $parent_snapname = $conf->{parent};
 259
 260     my ($base_snapshots, $last_snapshots, $last_sync_snapname) = find_common_replication_snapshot(
 261         $ssh_info, $jobid, $vmid, $storecfg, $sorted_volids, $state->{storeid_list}, $last_sync, $parent_snapname, $logfunc);
 262
 263     my $storeid_hash = {};
 264     foreach my $volid (@$sorted_volids) {
 265         my ($storeid) = PVE::Storage::parse_volume_id($volid);
 266         $storeid_hash->{$storeid} = 1;
 267     }
 268     $state->{storeid_list} = [ sort keys %$storeid_hash ];
 269
 270     # freeze filesystem for data consistency
 271     if ($freezefs) {
 272         $logfunc->("freeze guest filesystem");
 273         $guest_class->__snapshot_freeze($vmid, 0);
 274     }
 275
 276     # make snapshot of all volumes
 277     my $sync_snapname =
 278         PVE::ReplicationState::replication_snapshot_name($jobid, $start_time);
 279
 280     my $replicate_snapshots = {};
 281     eval {
 282         foreach my $volid (@$sorted_volids) {
 283             $logfunc->("create snapshot '${sync_snapname}' on $volid");
 284             PVE::Storage::volume_snapshot($storecfg, $volid, $sync_snapname);
 285             $replicate_snapshots->{$volid} = 1;
 286         }
 287     };
 288     my $err = $@;
 289
 290     # thaw immediately
 291     if ($freezefs) {
 292         $logfunc->("thaw guest filesystem");
 293         $guest_class->__snapshot_freeze($vmid, 1);
 294     }
 295
 296     my $cleanup_local_snapshots = sub {
 297         my ($volid_hash, $snapname) = @_;
 298         foreach my $volid (sort keys %$volid_hash) {
 299             $logfunc->("delete previous replication snapshot '$snapname' on $volid");
 300             eval { PVE::Storage::volume_snapshot_delete($storecfg, $volid, $snapname); };
 301             warn $@ if $@;
 302         }
 303     };
 304
 305     if ($err) {
 306         $cleanup_local_snapshots->($replicate_snapshots, $sync_snapname); # try to cleanup
 307         die $err;
 308     }
 309
 310     eval {
 311
 312         my $rate = $jobcfg->{rate};
 313         my $insecure = $migration_type eq 'insecure';
 314
 315         foreach my $volid (@$sorted_volids) {
 316             my $base_snapname;
 317
 318             if (defined($base_snapname = $base_snapshots->{$volid})) {
 319                 $logfunc->("incremental sync '$volid' ($base_snapname => $sync_snapname)");
 320             } else {
 321                 $logfunc->("full sync '$volid' ($sync_snapname)");
 322             }
 323
 324             replicate_volume($ssh_info, $storecfg, $volid, $base_snapname, $sync_snapname, $rate, $insecure, $logfunc);
 325         }
 326     };
 327
 328     if ($err = $@) {
 329         $cleanup_local_snapshots->($replicate_snapshots, $sync_snapname); # try to cleanup
 330         # we do not cleanup the remote side here - this is done in
 331         # next run of prepare_local_job
 332         die $err;
 333     }
 334
 335     # remove old snapshots because they are no longer needed
 336     $cleanup_local_snapshots->($last_snapshots, $last_sync_snapname);
 337
 338     eval {
 339         remote_finalize_local_job($ssh_info, $jobid, $vmid, $sorted_volids, $start_time, $logfunc);
 340     };
 341
 342     # old snapshots will removed by next run from prepare_local_job.
 343     if ($err = $@) {
 344         # warn is for syslog/journal.
 345         warn $err;
 346
 347         # logfunc will written in replication log.
 348         $logfunc->("delete stale replication snapshot error: $err");
 349     }
 350
 351     return $volumes;
 352 }
 353
 354 my $run_replication_nolock = sub {
 355     my ($guest_class, $jobcfg, $iteration, $start_time, $logfunc, $verbose) = @_;
 356
 357     my $jobid = $jobcfg->{id};
 358
 359     my $volumes;
 360
 361     # we normally write errors into the state file,
 362     # but we also catch unexpected errors and log them to syslog
 363     # (for examply when there are problems writing the state file)
 364
 365     my $state = PVE::ReplicationState::read_job_state($jobcfg);
 366
 367     PVE::ReplicationState::record_job_start($jobcfg, $state, $start_time, $iteration);
 368
 369     my $t0 = [gettimeofday];
 370
 371     mkdir $PVE::ReplicationState::replicate_logdir;
 372     my $logfile = PVE::ReplicationState::job_logfile_name($jobid);
 373     open(my $logfd, '>', $logfile) ||
 374         die "unable to open replication log '$logfile' - $!\n";
 375
 376     my $logfunc_wrapper = sub {
 377         my ($msg) = @_;
 378
 379         my $ctime = get_log_time();
 380         print $logfd "$ctime $jobid: $msg\n";
 381         if ($logfunc) {
 382             if ($verbose) {
 383                 $logfunc->("$ctime $jobid: $msg");
 384             } else {
 385                 $logfunc->($msg);
 386             }
 387         }
 388     };
 389
 390     $logfunc_wrapper->("start replication job");
 391
 392     eval {
 393         $volumes = replicate($guest_class, $jobcfg, $state, $start_time, $logfunc_wrapper);
 394     };
 395     my $err = $@;
 396
 397     if ($err) {
 398         my $msg = "end replication job with error: $err";
 399         chomp $msg;
 400         $logfunc_wrapper->($msg);
 401     } else {
 402         $logfunc_wrapper->("end replication job");
 403     }
 404
 405     PVE::ReplicationState::record_job_end($jobcfg, $state, $start_time, tv_interval($t0), $err);
 406
 407     close($logfd);
 408
 409     die $err if $err;
 410
 411     return $volumes;
 412 };
 413
 414 sub run_replication {
 415     my ($guest_class, $jobcfg, $iteration, $start_time, $logfunc, $verbose) = @_;
 416
 417     my $volumes;
 418
 419     my $timeout = 2; # do not wait too long - we repeat periodically anyways
 420     $volumes = PVE::GuestHelpers::guest_migration_lock(
 421         $jobcfg->{guest}, $timeout, $run_replication_nolock,
 422         $guest_class, $jobcfg, $iteration, $start_time, $logfunc, $verbose);
 423
 424     return $volumes;
 425 }
 426
 427 1;