1 package PVE
::ReplicationState
;
10 use PVE
::CalendarEvent
;
12 use PVE
::GuestHelpers
;
13 use PVE
::ReplicationConfig
;
15 # Note: regression tests can overwrite $state_path for testing
16 our $state_path = "/var/lib/pve-manager/pve-replication-state.json";
17 our $state_lock = "/var/lib/pve-manager/pve-replication-state.lck";
18 our $replicate_logdir = "/var/log/pve/replicate";
20 # regression tests should overwrite this
21 sub job_logfile_name
{
24 return "${replicate_logdir}/$jobid";
27 # Note: We use PVE::Tools::file_set_contents to write state file atomically,
28 # so read_state() always returns an consistent copy (even when not locked).
32 return {} if ! -e
$state_path;
34 my $raw = PVE
::Tools
::file_get_contents
($state_path, (1024 * 1024));
36 return {} if $raw eq '';
39 if ($raw =~ m/^(\{.*\})$/) {
40 return decode_json
($1);
43 die "invalid json data in '$state_path'\n";
46 sub extract_job_state
{
47 my ($stateobj, $jobcfg) = @_;
49 my $plugin = PVE
::ReplicationConfig-
>lookup($jobcfg->{type
});
51 my $vmid = $jobcfg->{guest
};
52 my $tid = $plugin->get_unique_target_id($jobcfg);
53 my $state = $stateobj->{$vmid}->{$tid};
55 $state = {} if !$state;
57 $state->{last_iteration
} //= 0;
58 $state->{last_try
} //= 0; # last sync start time
59 $state->{last_sync
} //= 0; # last successful sync start time
60 $state->{fail_count
} //= 0;
65 sub extract_vmid_tranfer_state
{
66 my ($stateobj, $vmid, $old_target, $new_target) = @_;
68 my $oldid = PVE
::ReplicationConfig
::Cluster-
>get_unique_target_id({ target
=> $old_target });
69 my $newid = PVE
::ReplicationConfig
::Cluster-
>get_unique_target_id({ target
=> $new_target });
71 if (defined(my $vmstate = $stateobj->{$vmid})) {
72 $vmstate->{$newid} = delete($vmstate->{$oldid}) if defined($vmstate->{$oldid});
82 my $stateobj = read_state
();
83 return extract_job_state
($stateobj, $jobcfg);
86 # update state for a single job
87 # pass $state = undef to delete the job state completely
89 my ($jobcfg, $state) = @_;
91 my $plugin = PVE
::ReplicationConfig-
>lookup($jobcfg->{type
});
93 my $vmid = $jobcfg->{guest
};
94 my $tid = $plugin->get_unique_target_id($jobcfg);
98 my $stateobj = read_state
();
99 # Note: tuple ($vmid, $tid) is unique
100 if (defined($state)) {
101 $stateobj->{$vmid}->{$tid} = $state;
103 delete $stateobj->{$vmid}->{$tid};
104 delete $stateobj->{$vmid} if !%{$stateobj->{$vmid}};
106 PVE
::Tools
::file_set_contents
($state_path, encode_json
($stateobj));
110 PVE
::Tools
::lock_file
($state_lock, 10, $update);
114 # make sure we have guest_migration_lock during update
115 PVE
::GuestHelpers
::guest_migration_lock
($vmid, undef, $code);
118 # update all job states related to a specific $vmid
119 sub write_vmid_job_states
{
120 my ($vmid_state, $vmid) = @_;
123 my $stateobj = read_state
();
124 $stateobj->{$vmid} = $vmid_state;
125 PVE
::Tools
::file_set_contents
($state_path, encode_json
($stateobj));
129 PVE
::Tools
::lock_file
($state_lock, 10, $update);
133 # make sure we have guest_migration_lock during update
134 PVE
::GuestHelpers
::guest_migration_lock
($vmid, undef, $code);
137 sub record_job_start
{
138 my ($jobcfg, $state, $start_time, $iteration) = @_;
141 $state->{ptime
} = PVE
::ProcFSTools
::read_proc_starttime
($state->{pid
});
142 $state->{last_node
} = PVE
::INotify
::nodename
();
143 $state->{last_try
} = $start_time;
144 $state->{last_iteration
} = $iteration;
145 $state->{storeid_list
} //= [];
147 write_job_state
($jobcfg, $state);
150 sub delete_guest_states
{
154 my $stateobj = read_state
();
155 delete $stateobj->{$vmid};
156 PVE
::Tools
::file_set_contents
($state_path, encode_json
($stateobj));
159 PVE
::Tools
::lock_file
($state_lock, 10, $code);
163 my ($jobcfg, $state, $start_time, $duration, $err) = @_;
165 $state->{duration
} = $duration;
166 delete $state->{pid
};
167 delete $state->{ptime
};
171 $state->{fail_count
}++;
172 $state->{error
} = "$err";
173 write_job_state
($jobcfg, $state);
175 if ($jobcfg->{remove_job
}) {
176 write_job_state
($jobcfg, undef);
178 $state->{last_sync
} = $start_time;
179 $state->{fail_count
} = 0;
180 delete $state->{error
};
181 write_job_state
($jobcfg, $state);
186 sub replication_snapshot_name
{
187 my ($jobid, $last_sync) = @_;
189 my $prefix = "__replicate_${jobid}_";
190 my $snapname = "${prefix}${last_sync}__";
192 wantarray ?
($prefix, $snapname) : $snapname;
195 sub purge_old_states
{
197 my $local_node = PVE
::INotify
::nodename
();
199 my $cfg = PVE
::ReplicationConfig-
>new();
200 PVE
::Cluster
::cfs_update
(1); # fail if we cannot query the vm list
201 my $vms = PVE
::Cluster
::get_vmlist
();
205 foreach my $jobid (sort keys %{$cfg->{ids
}}) {
206 my $jobcfg = $cfg->{ids
}->{$jobid};
207 my $plugin = PVE
::ReplicationConfig-
>lookup($jobcfg->{type
});
208 my $tid = $plugin->get_unique_target_id($jobcfg);
209 my $vmid = $jobcfg->{guest
};
210 $used_tids->{$vmid}->{$tid} = 1
211 if defined($vms->{ids
}->{$vmid}); # && $vms->{ids}->{$vmid}->{node} eq $local_node;
214 my $purge_state = sub {
215 my $stateobj = read_state
();
216 my $next_stateobj = {};
218 foreach my $vmid (keys %$stateobj) {
219 foreach my $tid (keys %{$stateobj->{$vmid}}) {
220 $next_stateobj->{$vmid}->{$tid} = $stateobj->{$vmid}->{$tid} if $used_tids->{$vmid}->{$tid};
223 PVE
::Tools
::file_set_contents
($state_path, encode_json
($next_stateobj));
226 PVE
::Tools
::lock_file
($state_lock, 10, $purge_state);
231 my ($get_disabled) = @_;
233 my $local_node = PVE
::INotify
::nodename
();
237 my $stateobj = read_state
();
239 my $cfg = PVE
::ReplicationConfig-
>new();
241 my $vms = PVE
::Cluster
::get_vmlist
();
244 foreach my $jobid (sort keys %{$cfg->{ids
}}) {
245 my $jobcfg = $cfg->{ids
}->{$jobid};
246 my $vmid = $jobcfg->{guest
};
248 die "internal error - not implemented" if $jobcfg->{type
} ne 'local';
250 # skip non existing vms
251 next if !$vms->{ids
}->{$vmid};
253 # only consider guest on local node
254 next if $vms->{ids
}->{$vmid}->{node
} ne $local_node;
256 my $target = $jobcfg->{target
};
257 if (!$jobcfg->{remove_job
}) {
258 # check if vm was stolen (swapped source target)
259 if ($target eq $local_node) {
260 my $source = $jobcfg->{source
};
261 if (defined($source) && $source ne $target) {
262 $jobcfg = PVE
::ReplicationConfig
::swap_source_target_nolock
($jobid);
263 $cfg->{ids
}->{$jobid} = $jobcfg;
265 # never sync to local node
270 next if !$get_disabled && $jobcfg->{disable
};
273 my $state = extract_job_state
($stateobj, $jobcfg);
274 $jobcfg->{state} = $state;
275 $jobcfg->{id
} = $jobid;
276 $jobcfg->{vmtype
} = $vms->{ids
}->{$vmid}->{type
};
280 if ($jobcfg->{remove_job
}) {
281 $next_sync = 1; # lowest possible value
282 # todo: consider fail_count? How many retries?
284 if (my $fail_count = $state->{fail_count
}) {
285 my $members = PVE
::Cluster
::get_members
();
286 if (!$fail_count || ($members->{$target} && $members->{$target}->{online
})) {
287 $next_sync = $state->{last_try
} + 60*($fail_count < 3 ?
5*$fail_count : 30);
290 my $schedule = $jobcfg->{schedule
} || '*/15';
291 my $calspec = PVE
::CalendarEvent
::parse_calendar_event
($schedule);
292 $next_sync = PVE
::CalendarEvent
::compute_next_event
($calspec, $state->{last_try
}) // 0;
296 $jobcfg->{next_sync
} = $next_sync;
298 if (!defined($jobcfg->{source
}) || $jobcfg->{source
} ne $local_node) {
299 $jobcfg->{source
} = $cfg->{ids
}->{$jobid}->{source
} = $local_node;
300 PVE
::ReplicationConfig
::write($cfg);
303 $jobs->{$jobid} = $jobcfg;
307 PVE
::ReplicationConfig
::lock($func);
313 my ($iteration, $start_time) = @_;
315 my $jobs = job_status
();
317 my $sort_func = sub {
318 my $joba = $jobs->{$a};
319 my $jobb = $jobs->{$b};
320 my $sa = $joba->{state};
321 my $sb = $jobb->{state};
322 my $res = $sa->{last_iteration
} <=> $sb->{last_iteration
};
323 return $res if $res != 0;
324 $res = $joba->{next_sync
} <=> $jobb->{next_sync
};
325 return $res if $res != 0;
326 return $joba->{guest
} <=> $jobb->{guest
};
329 foreach my $jobid (sort $sort_func keys %$jobs) {
330 my $jobcfg = $jobs->{$jobid};
331 next if $jobcfg->{state}->{last_iteration
} >= $iteration;
332 if ($jobcfg->{next_sync
} && ($start_time >= $jobcfg->{next_sync
})) {
340 sub schedule_job_now
{
343 PVE
::GuestHelpers
::guest_migration_lock
($jobcfg->{guest
}, undef, sub {
344 PVE
::Tools
::lock_file
($state_lock, 10, sub {
345 my $stateobj = read_state
();
346 my $vmid = $jobcfg->{guest
};
347 my $plugin = PVE
::ReplicationConfig-
>lookup($jobcfg->{type
});
348 my $tid = $plugin->get_unique_target_id($jobcfg);
349 # no not modify anything if there is no state
350 return if !defined($stateobj->{$vmid}->{$tid});
352 my $state = read_job_state
($jobcfg);
353 $state->{last_try
} = 0;
354 write_job_state
($jobcfg, $state);