Source code

001/*
002 * VM-Operator
003 * Copyright (C) 2023,2025 Michael N. Lipp
004 * 
005 * This program is free software: you can redistribute it and/or modify
006 * it under the terms of the GNU Affero General Public License as
007 * published by the Free Software Foundation, either version 3 of the
008 * License, or (at your option) any later version.
009 *
010 * This program is distributed in the hope that it will be useful,
011 * but WITHOUT ANY WARRANTY; without even the implied warranty of
012 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
013 * GNU Affero General Public License for more details.
014 *
015 * You should have received a copy of the GNU Affero General Public License
016 * along with this program.  If not, see <https://www.gnu.org/licenses/>.
017 */
018
019package org.jdrupes.vmoperator.runner.qemu;
020
021import com.fasterxml.jackson.databind.ObjectMapper;
022import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
023import com.google.gson.Gson;
024import com.google.gson.JsonObject;
025import io.kubernetes.client.apimachinery.GroupVersionKind;
026import io.kubernetes.client.custom.Quantity;
027import io.kubernetes.client.custom.Quantity.Format;
028import io.kubernetes.client.custom.V1Patch;
029import io.kubernetes.client.openapi.ApiException;
030import io.kubernetes.client.openapi.JSON;
031import io.kubernetes.client.openapi.models.EventsV1Event;
032import java.io.IOException;
033import java.math.BigDecimal;
034import java.math.BigInteger;
035import java.time.Instant;
036import java.util.Optional;
037import java.util.logging.Level;
038import static org.jdrupes.vmoperator.common.Constants.APP_NAME;
039import org.jdrupes.vmoperator.common.Constants.Crd;
040import org.jdrupes.vmoperator.common.Constants.Status;
041import org.jdrupes.vmoperator.common.Constants.Status.Condition;
042import org.jdrupes.vmoperator.common.Constants.Status.Condition.Reason;
043import org.jdrupes.vmoperator.common.K8s;
044import org.jdrupes.vmoperator.common.VmDefinition;
045import org.jdrupes.vmoperator.common.VmDefinitionStub;
046import org.jdrupes.vmoperator.runner.qemu.events.BalloonChangeEvent;
047import org.jdrupes.vmoperator.runner.qemu.events.ConfigureQemu;
048import org.jdrupes.vmoperator.runner.qemu.events.DisplayPasswordChanged;
049import org.jdrupes.vmoperator.runner.qemu.events.Exit;
050import org.jdrupes.vmoperator.runner.qemu.events.HotpluggableCpuStatus;
051import org.jdrupes.vmoperator.runner.qemu.events.OsinfoEvent;
052import org.jdrupes.vmoperator.runner.qemu.events.RunnerStateChange;
053import org.jdrupes.vmoperator.runner.qemu.events.RunnerStateChange.RunState;
054import org.jdrupes.vmoperator.runner.qemu.events.ShutdownEvent;
055import org.jdrupes.vmoperator.runner.qemu.events.VmopAgentConnected;
056import org.jdrupes.vmoperator.runner.qemu.events.VmopAgentLoggedIn;
057import org.jdrupes.vmoperator.runner.qemu.events.VmopAgentLoggedOut;
058import org.jdrupes.vmoperator.util.GsonPtr;
059import org.jgrapes.core.Channel;
060import org.jgrapes.core.Components;
061import org.jgrapes.core.Components.Timer;
062import org.jgrapes.core.annotation.Handler;
063import org.jgrapes.core.events.HandlingError;
064import org.jgrapes.core.events.Start;
065
066/**
067 * Updates the CR status.
068 */
069@SuppressWarnings({ "PMD.CouplingBetweenObjects" })
070public class StatusUpdater extends VmDefUpdater {
071
072    @SuppressWarnings("PMD.FieldNamingConventions")
073    private static final Gson gson = new JSON().getGson();
074    @SuppressWarnings("PMD.FieldNamingConventions")
075    private static final ObjectMapper objectMapper
076        = new ObjectMapper().registerModule(new JavaTimeModule());
077
078    private boolean guestShutdownStops;
079    private boolean shutdownByGuest;
080    private VmDefinitionStub vmStub;
081    private String loggedInUser;
082    private BigInteger lastRamValue;
083    private Instant lastRamChange;
084    private Timer balloonTimer;
085    private BigInteger targetRamValue;
086
087    /**
088     * Instantiates a new status updater.
089     *
090     * @param componentChannel the component channel
091     */
092    @SuppressWarnings("PMD.ConstructorCallsOverridableMethod")
093    public StatusUpdater(Channel componentChannel) {
094        super(componentChannel);
095        attach(new ConsoleTracker(componentChannel));
096    }
097
098    /**
099     * On handling error.
100     *
101     * @param event the event
102     */
103    @Handler(channels = Channel.class)
104    public void onHandlingError(HandlingError event) {
105        if (event.throwable() instanceof ApiException exc) {
106            logger.log(Level.WARNING, exc,
107                () -> "Problem accessing kubernetes: " + exc.getResponseBody());
108            event.stop();
109        }
110    }
111
112    /**
113     * Handle the start event.
114     *
115     * @param event the event
116     * @throws IOException 
117     * @throws ApiException 
118     */
119    @Handler
120    public void onStart(Start event) {
121        if (namespace == null) {
122            return;
123        }
124        try {
125            vmStub = VmDefinitionStub.get(apiClient,
126                new GroupVersionKind(Crd.GROUP, "", Crd.KIND_VM),
127                namespace, vmName);
128            var vmDef = vmStub.model().orElse(null);
129            if (vmDef == null) {
130                return;
131            }
132            vmStub.updateStatus(from -> {
133                JsonObject status = from.statusJson();
134                status.addProperty(Status.RUNNER_VERSION, Optional.ofNullable(
135                    Runner.class.getPackage().getImplementationVersion())
136                    .orElse("(unknown)"));
137                status.remove(Status.LOGGED_IN_USER);
138                return status;
139            });
140        } catch (ApiException e) {
141            logger.log(Level.SEVERE, e,
142                () -> "Cannot access VM object, terminating.");
143            event.cancel(true);
144            fire(new Exit(1));
145        }
146    }
147
148    /**
149     * On runner configuration update.
150     *
151     * @param event the event
152     * @throws ApiException 
153     */
154    @Handler
155    public void onConfigureQemu(ConfigureQemu event)
156            throws ApiException {
157        guestShutdownStops = event.configuration().guestShutdownStops;
158        loggedInUser = event.configuration().vm.display.loggedInUser;
159        targetRamValue = event.configuration().vm.currentRam;
160
161        // Remainder applies only if we have a connection to k8s.
162        if (vmStub == null) {
163            return;
164        }
165        vmStub.updateStatus(from -> {
166            JsonObject status = from.statusJson();
167            if (!event.configuration().hasDisplayPassword) {
168                status.addProperty(Status.DISPLAY_PASSWORD_SERIAL, -1);
169            }
170            status.getAsJsonArray("conditions").asList().stream()
171                .map(cond -> (JsonObject) cond)
172                .filter(cond -> Condition.RUNNING
173                    .equals(cond.get("type").getAsString()))
174                .forEach(cond -> cond.addProperty("observedGeneration",
175                    from.getMetadata().getGeneration()));
176            updateUserLoggedIn(from);
177            return status;
178        });
179    }
180
181    /**
182     * On runner state changed.
183     *
184     * @param event the event
185     * @throws ApiException 
186     */
187    @Handler
188    @SuppressWarnings({ "PMD.AssignmentInOperand" })
189    public void onRunnerStateChanged(RunnerStateChange event)
190            throws ApiException {
191        VmDefinition vmDef;
192        if (vmStub == null || (vmDef = vmStub.model().orElse(null)) == null) {
193            return;
194        }
195        vmStub.updateStatus(from -> {
196            boolean running = event.runState().vmRunning();
197            updateCondition(vmDef, Condition.RUNNING, running, event.reason(),
198                event.message());
199            JsonObject status = updateCondition(vmDef, Condition.BOOTED,
200                event.runState() == RunState.BOOTED, event.reason(),
201                event.message());
202            if (event.runState() == RunState.STARTING) {
203                status.addProperty(Status.RAM, GsonPtr.to(from.data())
204                    .getAsString("spec", "vm", "maximumRam").orElse("0"));
205                status.addProperty(Status.CPUS, 1);
206            } else if (event.runState() == RunState.STOPPED) {
207                status.addProperty(Status.RAM, "0");
208                status.addProperty(Status.CPUS, 0);
209                status.remove(Status.LOGGED_IN_USER);
210            }
211
212            if (!running) {
213                // In case console connection was still present
214                status.addProperty(Status.CONSOLE_CLIENT, "");
215                updateCondition(from, Condition.CONSOLE_CONNECTED, false,
216                    "VmStopped",
217                    "The VM is not running");
218
219                // In case we had an irregular shutdown
220                updateCondition(from, Condition.USER_LOGGED_IN, false,
221                    "VmStopped", "The VM is not running");
222                status.remove(Status.OSINFO);
223                updateCondition(vmDef, "VmopAgentConnected", false, "VmStopped",
224                    "The VM is not running");
225            }
226            return status;
227        }, vmDef);
228
229        // Maybe stop VM
230        if (event.runState() == RunState.TERMINATING && !event.failed()
231            && guestShutdownStops && shutdownByGuest) {
232            logger.info(() -> "Stopping VM because of shutdown by guest.");
233            var res = vmStub.patch(V1Patch.PATCH_FORMAT_JSON_PATCH,
234                new V1Patch("[{\"op\": \"replace\", \"path\": \"/spec/vm/state"
235                    + "\", \"value\": \"Stopped\"}]"),
236                apiClient.defaultPatchOptions());
237            if (!res.isPresent()) {
238                logger.warning(
239                    () -> "Cannot patch pod annotations for: " + vmStub.name());
240            }
241        }
242
243        // Log event
244        var evt = new EventsV1Event()
245            .reportingController(Crd.GROUP + "/" + APP_NAME)
246            .action("StatusUpdate").reason(event.reason())
247            .note(event.message());
248        K8s.createEvent(apiClient, vmDef, evt);
249    }
250
251    private void updateUserLoggedIn(VmDefinition from) {
252        if (loggedInUser == null) {
253            updateCondition(from, Condition.USER_LOGGED_IN, false,
254                Reason.NOT_REQUESTED, "No user to be logged in");
255            return;
256        }
257        if (!from.conditionStatus(Condition.VMOP_AGENT).orElse(false)) {
258            updateCondition(from, Condition.USER_LOGGED_IN, false,
259                "VmopAgentDisconnected", "Waiting for VMOP agent to connect");
260            return;
261        }
262        if (!from.fromStatus(Status.LOGGED_IN_USER).map(loggedInUser::equals)
263            .orElse(false)) {
264            updateCondition(from, Condition.USER_LOGGED_IN, false,
265                "Processing", "Waiting for user to be logged in");
266        }
267        updateCondition(from, Condition.USER_LOGGED_IN, true,
268            Reason.LOGGED_IN, "User is logged in");
269    }
270
271    /**
272     * Update the current RAM size in the status. Balloon changes happen
273     * more than once every second during changes. While this is nice
274     * to watch, this puts a heavy load on the system. Therefore we
275     * only update the status once every 15 seconds or when the target
276     * value is reached.
277     *
278     * @param event the event
279     * @throws ApiException 
280     */
281    @Handler
282    public void onBallonChange(BalloonChangeEvent event) throws ApiException {
283        if (vmStub == null) {
284            return;
285        }
286        Instant now = Instant.now();
287        if (lastRamChange == null
288            || lastRamChange.isBefore(now.minusSeconds(15))
289            || event.size().equals(targetRamValue)) {
290            if (balloonTimer != null) {
291                balloonTimer.cancel();
292                balloonTimer = null;
293            }
294            lastRamChange = now;
295            lastRamValue = event.size();
296            updateRam();
297            return;
298        }
299
300        // Save for later processing and maybe start timer
301        lastRamChange = now;
302        lastRamValue = event.size();
303        if (balloonTimer != null) {
304            return;
305        }
306        final var pipeline = activeEventPipeline();
307        balloonTimer = Components.schedule(t -> {
308            pipeline.submit("Update RAM size", () -> {
309                try {
310                    updateRam();
311                } catch (ApiException e) {
312                    logger.log(Level.WARNING, e,
313                        () -> "Failed to update ram size: " + e.getMessage());
314                }
315                balloonTimer = null;
316            });
317        }, now.plusSeconds(15));
318    }
319
320    private void updateRam() throws ApiException {
321        vmStub.updateStatus(from -> {
322            JsonObject status = from.statusJson();
323            status.addProperty(Status.RAM,
324                new Quantity(new BigDecimal(lastRamValue), Format.BINARY_SI)
325                    .toSuffixedString());
326            return status;
327        });
328    }
329
330    /**
331     * On ballon change.
332     *
333     * @param event the event
334     * @throws ApiException 
335     */
336    @Handler
337    public void onCpuChange(HotpluggableCpuStatus event) throws ApiException {
338        if (vmStub == null) {
339            return;
340        }
341        vmStub.updateStatus(from -> {
342            JsonObject status = from.statusJson();
343            status.addProperty(Status.CPUS, event.usedCpus().size());
344            return status;
345        });
346    }
347
348    /**
349     * On ballon change.
350     *
351     * @param event the event
352     * @throws ApiException 
353     */
354    @Handler
355    public void onDisplayPasswordChanged(DisplayPasswordChanged event)
356            throws ApiException {
357        if (vmStub == null) {
358            return;
359        }
360        vmStub.updateStatus(from -> {
361            JsonObject status = from.statusJson();
362            status.addProperty(Status.DISPLAY_PASSWORD_SERIAL,
363                status.get(Status.DISPLAY_PASSWORD_SERIAL).getAsLong() + 1);
364            return status;
365        });
366    }
367
368    /**
369     * On shutdown.
370     *
371     * @param event the event
372     * @throws ApiException the api exception
373     */
374    @Handler
375    public void onShutdown(ShutdownEvent event) throws ApiException {
376        shutdownByGuest = event.byGuest();
377    }
378
379    /**
380     * On osinfo.
381     *
382     * @param event the event
383     * @throws ApiException 
384     */
385    @Handler
386    public void onOsinfo(OsinfoEvent event) throws ApiException {
387        if (vmStub == null) {
388            return;
389        }
390        var asGson = gson.toJsonTree(
391            objectMapper.convertValue(event.osinfo(), Object.class));
392        vmStub.updateStatus(from -> {
393            JsonObject status = from.statusJson();
394            status.add(Status.OSINFO, asGson);
395            return status;
396        });
397
398    }
399
400    /**
401     * @param event the event
402     * @throws ApiException 
403     */
404    @Handler
405    @SuppressWarnings("PMD.AssignmentInOperand")
406    public void onVmopAgentConnected(VmopAgentConnected event)
407            throws ApiException {
408        VmDefinition vmDef;
409        if (vmStub == null || (vmDef = vmStub.model().orElse(null)) == null) {
410            return;
411        }
412        vmStub.updateStatus(from -> {
413            var status = updateCondition(vmDef, "VmopAgentConnected",
414                true, "VmopAgentStarted", "The VM operator agent is running");
415            updateUserLoggedIn(from);
416            return status;
417        }, vmDef);
418    }
419
420    /**
421     * @param event the event
422     * @throws ApiException 
423     */
424    @Handler
425    public void onVmopAgentLoggedIn(VmopAgentLoggedIn event)
426            throws ApiException {
427        vmStub.updateStatus(from -> {
428            JsonObject status = from.statusJson();
429            status.addProperty(Status.LOGGED_IN_USER,
430                event.triggering().user());
431            updateUserLoggedIn(from);
432            return status;
433        });
434    }
435
436    /**
437     * @param event the event
438     * @throws ApiException 
439     */
440    @Handler
441    public void onVmopAgentLoggedOut(VmopAgentLoggedOut event)
442            throws ApiException {
443        vmStub.updateStatus(from -> {
444            JsonObject status = from.statusJson();
445            status.remove(Status.LOGGED_IN_USER);
446            updateUserLoggedIn(from);
447            return status;
448        });
449    }
450}