]>
Commit | Line | Data |
---|---|---|
3f7cacff DM |
1 | package PVE::ReplicationState; |
2 | ||
3 | use warnings; | |
4 | use strict; | |
5 | use JSON; | |
6 | ||
c292c8e9 | 7 | use PVE::INotify; |
c17dcb3e | 8 | use PVE::ProcFSTools; |
3f7cacff | 9 | use PVE::Tools; |
c292c8e9 DM |
10 | use PVE::CalendarEvent; |
11 | use PVE::Cluster; | |
90c07bf7 | 12 | use PVE::GuestHelpers; |
3f7cacff DM |
13 | use PVE::ReplicationConfig; |
14 | ||
3f7cacff DM |
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"; | |
c292c8e9 DM |
18 | our $replicate_logdir = "/var/log/pve/replicate"; |
19 | ||
20 | # regression tests should overwrite this | |
21 | sub job_logfile_name { | |
22 | my ($jobid) = @_; | |
23 | ||
24 | return "${replicate_logdir}/$jobid"; | |
25 | } | |
3f7cacff DM |
26 | |
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). | |
29 | ||
30 | sub read_state { | |
31 | ||
32 | return {} if ! -e $state_path; | |
33 | ||
34 | my $raw = PVE::Tools::file_get_contents($state_path); | |
35 | ||
36 | return {} if $raw eq ''; | |
37 | ||
38 | # untaint $raw | |
39 | if ($raw =~ m/^({.*})$/) { | |
40 | return decode_json($1); | |
41 | } | |
42 | ||
43 | die "invalid json data in '$state_path'\n"; | |
44 | } | |
45 | ||
46 | sub extract_job_state { | |
47 | my ($stateobj, $jobcfg) = @_; | |
48 | ||
49 | my $plugin = PVE::ReplicationConfig->lookup($jobcfg->{type}); | |
50 | ||
51 | my $vmid = $jobcfg->{guest}; | |
52 | my $tid = $plugin->get_unique_target_id($jobcfg); | |
53 | my $state = $stateobj->{$vmid}->{$tid}; | |
54 | ||
55 | $state = {} if !$state; | |
56 | ||
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; | |
61 | ||
62 | return $state; | |
63 | } | |
64 | ||
210a5f79 DM |
65 | sub extract_vmid_tranfer_state { |
66 | my ($stateobj, $vmid, $old_target, $new_target) = @_; | |
67 | ||
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 }); | |
70 | ||
71 | if (defined(my $vmstate = $stateobj->{$vmid})) { | |
72 | $vmstate->{$newid} = delete($vmstate->{$oldid}) if defined($vmstate->{$oldid}); | |
73 | return $vmstate; | |
74 | } | |
75 | ||
76 | return {}; | |
77 | } | |
78 | ||
3f7cacff DM |
79 | sub read_job_state { |
80 | my ($jobcfg) = @_; | |
81 | ||
82 | my $stateobj = read_state(); | |
83 | return extract_job_state($stateobj, $jobcfg); | |
84 | } | |
85 | ||
55222f37 | 86 | # update state for a single job |
2c508173 | 87 | # pass $state = undef to delete the job state completely |
3f7cacff DM |
88 | sub write_job_state { |
89 | my ($jobcfg, $state) = @_; | |
90 | ||
91 | my $plugin = PVE::ReplicationConfig->lookup($jobcfg->{type}); | |
92 | ||
93 | my $vmid = $jobcfg->{guest}; | |
94 | my $tid = $plugin->get_unique_target_id($jobcfg); | |
95 | ||
90c07bf7 | 96 | my $update = sub { |
3f7cacff DM |
97 | |
98 | my $stateobj = read_state(); | |
99 | # Note: tuple ($vmid, $tid) is unique | |
2c508173 DM |
100 | if (defined($state)) { |
101 | $stateobj->{$vmid}->{$tid} = $state; | |
102 | } else { | |
103 | delete $stateobj->{$vmid}->{$tid}; | |
104 | delete $stateobj->{$vmid} if !%{$stateobj->{$vmid}}; | |
105 | } | |
3f7cacff DM |
106 | PVE::Tools::file_set_contents($state_path, encode_json($stateobj)); |
107 | }; | |
108 | ||
90c07bf7 DM |
109 | my $code = sub { |
110 | PVE::Tools::lock_file($state_lock, 10, $update); | |
111 | die $@ if $@; | |
112 | }; | |
113 | ||
114 | # make sure we have guest_migration_lock during update | |
115 | PVE::GuestHelpers::guest_migration_lock($vmid, undef, $code); | |
3f7cacff DM |
116 | } |
117 | ||
55222f37 DM |
118 | # update all job states related to a specific $vmid |
119 | sub write_vmid_job_states { | |
120 | my ($vmid_state, $vmid) = @_; | |
121 | ||
122 | my $update = sub { | |
123 | my $stateobj = read_state(); | |
124 | $stateobj->{$vmid} = $vmid_state; | |
125 | PVE::Tools::file_set_contents($state_path, encode_json($stateobj)); | |
126 | }; | |
127 | ||
128 | my $code = sub { | |
129 | PVE::Tools::lock_file($state_lock, 10, $update); | |
130 | die $@ if $@; | |
131 | }; | |
132 | ||
133 | # make sure we have guest_migration_lock during update | |
134 | PVE::GuestHelpers::guest_migration_lock($vmid, undef, $code); | |
135 | } | |
136 | ||
c17dcb3e DM |
137 | sub record_job_start { |
138 | my ($jobcfg, $state, $start_time, $iteration) = @_; | |
139 | ||
140 | $state->{pid} = $$; | |
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} //= []; | |
146 | ||
147 | write_job_state($jobcfg, $state); | |
148 | } | |
149 | ||
150 | sub record_job_end { | |
151 | my ($jobcfg, $state, $start_time, $duration, $err) = @_; | |
152 | ||
153 | $state->{duration} = $duration; | |
154 | delete $state->{pid}; | |
155 | delete $state->{ptime}; | |
156 | ||
157 | if ($err) { | |
158 | chomp $err; | |
159 | $state->{fail_count}++; | |
160 | $state->{error} = "$err"; | |
92a243e9 | 161 | write_job_state($jobcfg, $state); |
c17dcb3e | 162 | } else { |
92a243e9 DM |
163 | if ($jobcfg->{remove_job}) { |
164 | write_job_state($jobcfg, undef); | |
165 | } else { | |
166 | $state->{last_sync} = $start_time; | |
167 | $state->{fail_count} = 0; | |
168 | delete $state->{error}; | |
169 | write_job_state($jobcfg, $state); | |
170 | } | |
c17dcb3e | 171 | } |
c17dcb3e DM |
172 | } |
173 | ||
52dcecfc DM |
174 | sub replication_snapshot_name { |
175 | my ($jobid, $last_sync) = @_; | |
176 | ||
177 | my $prefix = "__replicate_${jobid}_"; | |
178 | my $snapname = "${prefix}${last_sync}__"; | |
179 | ||
180 | wantarray ? ($prefix, $snapname) : $snapname; | |
181 | } | |
182 | ||
44972014 DM |
183 | sub purge_old_states { |
184 | ||
185 | my $local_node = PVE::INotify::nodename(); | |
186 | ||
187 | my $cfg = PVE::ReplicationConfig->new(); | |
188 | my $vms = PVE::Cluster::get_vmlist(); | |
189 | ||
190 | my $used_tids = {}; | |
191 | ||
192 | foreach my $jobid (sort keys %{$cfg->{ids}}) { | |
193 | my $jobcfg = $cfg->{ids}->{$jobid}; | |
194 | my $plugin = PVE::ReplicationConfig->lookup($jobcfg->{type}); | |
195 | my $tid = $plugin->get_unique_target_id($jobcfg); | |
196 | my $vmid = $jobcfg->{guest}; | |
197 | $used_tids->{$vmid}->{$tid} = 1 | |
198 | if defined($vms->{ids}->{$vmid}); # && $vms->{ids}->{$vmid}->{node} eq $local_node; | |
199 | } | |
200 | ||
201 | my $purge_state = sub { | |
202 | my $stateobj = read_state(); | |
203 | my $next_stateobj = {}; | |
204 | ||
205 | foreach my $vmid (keys %$stateobj) { | |
206 | foreach my $tid (keys %{$stateobj->{$vmid}}) { | |
207 | $next_stateobj->{$vmid}->{$tid} = $stateobj->{$vmid}->{$tid} if $used_tids->{$vmid}->{$tid}; | |
208 | } | |
209 | } | |
210 | PVE::Tools::file_set_contents($state_path, encode_json($next_stateobj)); | |
211 | }; | |
212 | ||
213 | PVE::Tools::lock_file($state_lock, 10, $purge_state); | |
214 | die $@ if $@; | |
215 | } | |
216 | ||
c292c8e9 DM |
217 | sub job_status { |
218 | ||
219 | my $local_node = PVE::INotify::nodename(); | |
220 | ||
221 | my $jobs = {}; | |
222 | ||
223 | my $stateobj = read_state(); | |
224 | ||
225 | my $cfg = PVE::ReplicationConfig->new(); | |
226 | ||
227 | my $vms = PVE::Cluster::get_vmlist(); | |
228 | ||
229 | foreach my $jobid (sort keys %{$cfg->{ids}}) { | |
230 | my $jobcfg = $cfg->{ids}->{$jobid}; | |
231 | my $vmid = $jobcfg->{guest}; | |
232 | ||
233 | die "internal error - not implemented" if $jobcfg->{type} ne 'local'; | |
234 | ||
235 | # skip non existing vms | |
236 | next if !$vms->{ids}->{$vmid}; | |
237 | ||
238 | # only consider guest on local node | |
239 | next if $vms->{ids}->{$vmid}->{node} ne $local_node; | |
240 | ||
241 | if (!$jobcfg->{remove_job}) { | |
242 | # never sync to local node | |
243 | next if $jobcfg->{target} eq $local_node; | |
244 | ||
245 | next if $jobcfg->{disable}; | |
246 | } | |
247 | ||
248 | my $state = extract_job_state($stateobj, $jobcfg); | |
249 | $jobcfg->{state} = $state; | |
250 | $jobcfg->{id} = $jobid; | |
251 | $jobcfg->{vmtype} = $vms->{ids}->{$vmid}->{type}; | |
252 | ||
253 | my $next_sync = 0; | |
254 | ||
255 | if ($jobcfg->{remove_job}) { | |
256 | $next_sync = 1; # lowest possible value | |
257 | # todo: consider fail_count? How many retries? | |
258 | } else { | |
259 | if (my $fail_count = $state->{fail_count}) { | |
33853993 | 260 | $next_sync = $state->{last_try} + 60*($fail_count < 3 ? 5*$fail_count : 30); |
c292c8e9 DM |
261 | } else { |
262 | my $schedule = $jobcfg->{schedule} || '*/15'; | |
263 | my $calspec = PVE::CalendarEvent::parse_calendar_event($schedule); | |
264 | $next_sync = PVE::CalendarEvent::compute_next_event($calspec, $state->{last_try}) // 0; | |
265 | } | |
266 | } | |
267 | ||
268 | $jobcfg->{next_sync} = $next_sync; | |
269 | ||
270 | $jobs->{$jobid} = $jobcfg; | |
271 | } | |
272 | ||
273 | return $jobs; | |
274 | } | |
275 | ||
276 | sub get_next_job { | |
277 | my ($iteration, $start_time) = @_; | |
278 | ||
279 | my $jobs = job_status(); | |
280 | ||
281 | my $sort_func = sub { | |
282 | my $joba = $jobs->{$a}; | |
283 | my $jobb = $jobs->{$b}; | |
284 | my $sa = $joba->{state}; | |
285 | my $sb = $jobb->{state}; | |
286 | my $res = $sa->{last_iteration} cmp $sb->{last_iteration}; | |
287 | return $res if $res != 0; | |
288 | $res = $joba->{next_sync} <=> $jobb->{next_sync}; | |
289 | return $res if $res != 0; | |
290 | return $joba->{guest} <=> $jobb->{guest}; | |
291 | }; | |
292 | ||
293 | foreach my $jobid (sort $sort_func keys %$jobs) { | |
294 | my $jobcfg = $jobs->{$jobid}; | |
295 | next if $jobcfg->{state}->{last_iteration} >= $iteration; | |
296 | if ($jobcfg->{next_sync} && ($start_time >= $jobcfg->{next_sync})) { | |
297 | return $jobcfg; | |
298 | } | |
299 | } | |
300 | ||
301 | return undef; | |
302 | } | |
303 | ||
3f7cacff | 304 | 1; |