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