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