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