001/*
002 * VM-Operator
003 * Copyright (C) 2023 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.core.JsonProcessingException;
022import com.fasterxml.jackson.databind.node.ObjectNode;
023import java.io.IOException;
024import java.nio.file.Path;
025import java.time.Duration;
026import java.time.Instant;
027import java.util.LinkedList;
028import java.util.Queue;
029import java.util.logging.Level;
030import org.jdrupes.vmoperator.runner.qemu.Constants.ProcessName;
031import org.jdrupes.vmoperator.runner.qemu.commands.QmpCapabilities;
032import org.jdrupes.vmoperator.runner.qemu.commands.QmpCommand;
033import org.jdrupes.vmoperator.runner.qemu.commands.QmpPowerdown;
034import org.jdrupes.vmoperator.runner.qemu.events.ConfigureQemu;
035import org.jdrupes.vmoperator.runner.qemu.events.MonitorCommand;
036import org.jdrupes.vmoperator.runner.qemu.events.MonitorEvent;
037import org.jdrupes.vmoperator.runner.qemu.events.MonitorReady;
038import org.jdrupes.vmoperator.runner.qemu.events.MonitorResult;
039import org.jdrupes.vmoperator.runner.qemu.events.PowerdownEvent;
040import org.jgrapes.core.Channel;
041import org.jgrapes.core.Components;
042import org.jgrapes.core.Components.Timer;
043import org.jgrapes.core.annotation.Handler;
044import org.jgrapes.core.events.Stop;
045import org.jgrapes.io.events.Closed;
046import org.jgrapes.io.events.ProcessExited;
047import org.jgrapes.net.SocketIOChannel;
048import org.jgrapes.util.events.ConfigurationUpdate;
049
050/**
051 * A component that handles the communication over the Qemu monitor
052 * socket.
053 * 
054 * If the log level for this class is set to fine, the messages 
055 * exchanged on the monitor socket are logged.
056 */
057public class QemuMonitor extends QemuConnector {
058
059    private int powerdownTimeout;
060    private final Queue<QmpCommand> executing = new LinkedList<>();
061    private Instant powerdownStartedAt;
062    private Stop suspendedStop;
063    private Timer powerdownTimer;
064    private boolean powerdownConfirmed;
065    private boolean monitorReady;
066
067    /**
068     * Instantiates a new QEMU monitor.
069     *
070     * @param componentChannel the component channel
071     * @param configDir the config dir
072     * @throws IOException Signals that an I/O exception has occurred.
073     */
074    @SuppressWarnings("PMD.ConstructorCallsOverridableMethod")
075    public QemuMonitor(Channel componentChannel, Path configDir)
076            throws IOException {
077        super(componentChannel);
078        attach(new RamController(channel()));
079        attach(new CpuController(channel()));
080        attach(new DisplayController(channel(), configDir));
081        attach(new CdMediaController(channel()));
082    }
083
084    /**
085     * As the initial configuration of this component depends on the 
086     * configuration of the {@link Runner}, it doesn't have a handler 
087     * for the {@link ConfigurationUpdate} event. The values are 
088     * forwarded from the {@link Runner} instead.
089     *
090     * @param socketPath the socket path
091     * @param powerdownTimeout 
092     */
093    /* default */ void configure(Path socketPath, int powerdownTimeout) {
094        super.configure(socketPath);
095        this.powerdownTimeout = powerdownTimeout;
096    }
097
098    /**
099     * When the socket is connected, send the capabilities command.
100     */
101    @Override
102    protected void socketConnected() {
103        rep().fire(new MonitorCommand(new QmpCapabilities()));
104    }
105
106    @Override
107    protected void processInput(String line)
108            throws IOException {
109        logger.finer(() -> "monitor(in): " + line);
110        try {
111            var response = mapper.readValue(line, ObjectNode.class);
112            if (response.has("QMP")) {
113                monitorReady = true;
114                logger.fine(() -> "QMP connection ready");
115                rep().fire(new MonitorReady());
116                return;
117            }
118            if (response.has("return") || response.has("error")) {
119                QmpCommand executed = executing.poll();
120                logger.finer(
121                    () -> String.format("(Previous \"monitor(in)\" is result "
122                        + "from executing %s)", executed));
123                var monRes = MonitorResult.from(executed, response);
124                logger.fine(() -> "QMP triggers: " + monRes);
125                rep().fire(monRes);
126                return;
127            }
128            if (response.has("event")) {
129                MonitorEvent.from(response).ifPresent(me -> {
130                    logger.fine(() -> "QMP triggers: " + me);
131                    rep().fire(me);
132                });
133            }
134        } catch (JsonProcessingException e) {
135            throw new IOException(e);
136        }
137    }
138
139    /**
140     * On closed.
141     *
142     * @param event the event
143     */
144    @Handler
145    @Override
146    public void onClosed(Closed<?> event, SocketIOChannel channel) {
147        channel.associated(this, getClass()).ifPresent(qm -> {
148            super.onClosed(event, channel);
149            logger.fine(() -> "QMP connection closed.");
150            monitorReady = false;
151        });
152    }
153
154    /**
155     * On monitor command.
156     *
157     * @param event the event
158     * @throws IOException 
159     */
160    @Handler
161    @SuppressWarnings({ "PMD.AvoidSynchronizedStatement",
162        "PMD.AvoidDuplicateLiterals" })
163    public void onMonitorCommand(MonitorCommand event) throws IOException {
164        // Check prerequisites
165        if (!monitorReady && !(event.command() instanceof QmpCapabilities)) {
166            logger.severe(() -> "Premature QMP command (not ready): "
167                + event.command());
168            rep().fire(new Stop());
169            return;
170        }
171
172        // Send the command
173        var command = event.command();
174        logger.fine(() -> "QMP handles: " + event.toString());
175        String asText;
176        try {
177            asText = command.asText();
178            logger.finer(() -> "monitor(out): " + asText);
179        } catch (JsonProcessingException e) {
180            logger.log(Level.SEVERE, e,
181                () -> "Cannot serialize Json: " + e.getMessage());
182            return;
183        }
184        synchronized (executing) {
185            if (writer().isPresent()) {
186                executing.add(command);
187                sendCommand(asText);
188            }
189        }
190    }
191
192    /**
193     * Shutdown the VM.
194     *
195     * @param event the event
196     */
197    @Handler(priority = 100)
198    @SuppressWarnings("PMD.AvoidSynchronizedStatement")
199    public void onStop(Stop event) {
200        if (!monitorReady) {
201            logger.fine(() -> "Not sending QMP powerdown command"
202                + " because QMP connection is closed");
203            return;
204        }
205
206        // We have a connection to Qemu, attempt ACPI shutdown if time left
207        powerdownStartedAt = event.associated(Instant.class).orElseGet(() -> {
208            var now = Instant.now();
209            event.setAssociated(Instant.class, now);
210            return now;
211        });
212        if (powerdownStartedAt.plusSeconds(powerdownTimeout)
213            .isBefore(Instant.now())) {
214            return;
215        }
216        event.suspendHandling();
217        suspendedStop = event;
218
219        // Send command. If not confirmed, assume "hanging" qemu process.
220        powerdownTimer = Components.schedule(t -> {
221            logger.fine(() -> "QMP powerdown command not confirmed");
222            synchronized (this) {
223                powerdownTimer = null;
224                if (suspendedStop != null) {
225                    suspendedStop.resumeHandling();
226                    suspendedStop = null;
227                }
228            }
229        }, Duration.ofSeconds(5));
230        logger.fine(() -> "Attempting QMP (ACPI) powerdown.");
231        rep().fire(new MonitorCommand(new QmpPowerdown()));
232    }
233
234    /**
235     * When the powerdown event is confirmed, wait for termination
236     * or timeout. Termination is detected by the qemu process exiting
237     * (see {@link #onProcessExited(ProcessExited)}).
238     *
239     * @param event the event
240     */
241    @Handler
242    @SuppressWarnings("PMD.AvoidSynchronizedStatement")
243    public void onPowerdownEvent(PowerdownEvent event) {
244        synchronized (this) {
245            // Cancel confirmation timeout
246            if (powerdownTimer != null) {
247                powerdownTimer.cancel();
248            }
249
250            // (Re-)schedule timer as fallback
251            var waitUntil = powerdownStartedAt.plusSeconds(powerdownTimeout);
252            logger.fine(() -> "QMP powerdown confirmed, waiting for"
253                + " termination until " + waitUntil);
254            powerdownTimer = Components.schedule(t -> {
255                logger.fine(() -> "Powerdown timeout reached.");
256                synchronized (this) {
257                    powerdownTimer = null;
258                    if (suspendedStop != null) {
259                        suspendedStop.resumeHandling();
260                        suspendedStop = null;
261                    }
262                }
263            }, waitUntil);
264            powerdownConfirmed = true;
265        }
266    }
267
268    /**
269     * On process exited.
270     *
271     * @param event the event
272     */
273    @Handler
274    @SuppressWarnings("PMD.AvoidSynchronizedStatement")
275    public void onProcessExited(ProcessExited event) {
276        if (!event.startedBy().associated(CommandDefinition.class)
277            .map(cd -> ProcessName.QEMU.equals(cd.name())).orElse(false)) {
278            return;
279        }
280        synchronized (this) {
281            if (powerdownTimer != null) {
282                powerdownTimer.cancel();
283            }
284            if (suspendedStop != null) {
285                suspendedStop.resumeHandling();
286                suspendedStop = null;
287            }
288        }
289    }
290
291    /**
292     * On configure qemu.
293     *
294     * @param event the event
295     */
296    @Handler
297    @SuppressWarnings("PMD.AvoidSynchronizedStatement")
298    public void onConfigureQemu(ConfigureQemu event) {
299        int newTimeout = event.configuration().vm.powerdownTimeout;
300        if (powerdownTimeout != newTimeout) {
301            powerdownTimeout = newTimeout;
302            synchronized (this) {
303                if (powerdownTimer != null && powerdownConfirmed) {
304                    powerdownTimer
305                        .reschedule(powerdownStartedAt.plusSeconds(newTimeout));
306                }
307
308            }
309        }
310    }
311
312}