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