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.manager;
020
021import com.google.gson.JsonObject;
022import io.kubernetes.client.apimachinery.GroupVersionKind;
023import io.kubernetes.client.custom.V1Patch;
024import io.kubernetes.client.openapi.ApiException;
025import io.kubernetes.client.openapi.models.V1PodIP;
026import io.kubernetes.client.util.Watch;
027import io.kubernetes.client.util.generic.options.ListOptions;
028import java.io.IOException;
029import java.net.HttpURLConnection;
030import java.time.Instant;
031import java.util.ArrayList;
032import java.util.Collections;
033import java.util.Optional;
034import java.util.Set;
035import java.util.stream.Collectors;
036import org.jdrupes.vmoperator.common.Constants.Crd;
037import org.jdrupes.vmoperator.common.Constants.Status;
038import org.jdrupes.vmoperator.common.K8s;
039import org.jdrupes.vmoperator.common.K8sClient;
040import org.jdrupes.vmoperator.common.K8sDynamicStub;
041import org.jdrupes.vmoperator.common.K8sGenericStub;
042import org.jdrupes.vmoperator.common.K8sObserver.ResponseType;
043import org.jdrupes.vmoperator.common.K8sV1ConfigMapStub;
044import org.jdrupes.vmoperator.common.K8sV1StatefulSetStub;
045import org.jdrupes.vmoperator.common.VmDefinition;
046import org.jdrupes.vmoperator.common.VmDefinitionStub;
047import org.jdrupes.vmoperator.common.VmDefinitions;
048import org.jdrupes.vmoperator.common.VmExtraData;
049import static org.jdrupes.vmoperator.manager.Constants.APP_NAME;
050import static org.jdrupes.vmoperator.manager.Constants.VM_OP_NAME;
051import org.jdrupes.vmoperator.manager.events.ChannelManager;
052import org.jdrupes.vmoperator.manager.events.ModifyVm;
053import org.jdrupes.vmoperator.manager.events.PodChanged;
054import org.jdrupes.vmoperator.manager.events.UpdateAssignment;
055import org.jdrupes.vmoperator.manager.events.VmChannel;
056import org.jdrupes.vmoperator.manager.events.VmResourceChanged;
057import org.jdrupes.vmoperator.util.GsonPtr;
058import org.jgrapes.core.Channel;
059import org.jgrapes.core.Event;
060import org.jgrapes.core.EventPipeline;
061import org.jgrapes.core.annotation.Handler;
062
063/**
064 * Watches for changes of VM definitions. When a VM definition (CR)
065 * becomes known, is is registered with a {@link ChannelManager} and thus
066 * gets an associated {@link VmChannel} and an associated
067 * {@link EventPipeline}.
068 * 
069 * The {@link EventPipeline} is used for submitting an action that processes
070 * the change data from kubernetes, eventually transforming it to a
071 * {@link VmResourceChanged} event that is handled by another
072 * {@link EventPipeline} associated with the {@link VmChannel}. This
073 * event pipeline should be used for all events related to changes of
074 * a particular VM.
075 */
076public class VmMonitor extends
077        AbstractMonitor<VmDefinition, VmDefinitions, VmChannel> {
078
079    private final ChannelManager<String, VmChannel,
080            EventPipeline> channelManager;
081
082    /**
083     * Instantiates a new VM definition watcher.
084     *
085     * @param componentChannel the component channel
086     * @param channelManager the channel manager
087     */
088    public VmMonitor(Channel componentChannel,
089            ChannelManager<String, VmChannel, EventPipeline> channelManager) {
090        super(componentChannel, VmDefinition.class,
091            VmDefinitions.class);
092        this.channelManager = channelManager;
093    }
094
095    @Override
096    protected void prepareMonitoring() throws IOException, ApiException {
097        client(new K8sClient());
098
099        // Get all our API versions
100        var ctx = K8s.context(client(), Crd.GROUP, "", Crd.KIND_VM);
101        if (ctx.isEmpty()) {
102            logger.severe(() -> "Cannot get CRD context.");
103            return;
104        }
105        context(ctx.get());
106
107        // Remove left over resources
108        purge();
109    }
110
111    private void purge() throws ApiException {
112        // Get existing CRs (VMs)
113        var known = K8sDynamicStub.list(client(), context(), namespace())
114            .stream().map(K8sGenericStub::name).collect(Collectors.toSet());
115        ListOptions opts = new ListOptions();
116        opts.setLabelSelector(
117            "app.kubernetes.io/managed-by=" + VM_OP_NAME + ","
118                + "app.kubernetes.io/name=" + APP_NAME);
119        for (var context : Set.of(K8sV1StatefulSetStub.CONTEXT,
120            K8sV1ConfigMapStub.CONTEXT)) {
121            for (var resStub : K8sDynamicStub.list(client(), context,
122                namespace(), opts)) {
123                String instance = resStub.model()
124                    .map(m -> m.metadata().getName()).orElse("(unknown)");
125                if (!known.contains(instance)) {
126                    resStub.delete();
127                }
128            }
129        }
130    }
131
132    @Override
133    protected void handleChange(K8sClient client,
134            Watch.Response<VmDefinition> response) {
135        var name = response.object.getMetadata().getName();
136
137        // Process the response data on a VM specific pipeline to
138        // increase concurrency when e.g. starting many VMs.
139        var preparing = channelManager.associated(name)
140            .orElseGet(this::newEventPipeline);
141        preparing.submit("VmChange[" + name + "]",
142            () -> processChange(client, response, preparing));
143    }
144
145    private void processChange(K8sClient client,
146            Watch.Response<VmDefinition> response, EventPipeline preparing) {
147        // Get full definition and associate with channel as backup
148        var vmDef = response.object;
149        if (vmDef.data() == null) {
150            // ADDED event does not provide data, see
151            // https://github.com/kubernetes-client/java/issues/3215
152            vmDef = getModel(client, vmDef);
153        }
154        var name = response.object.getMetadata().getName();
155        var channel = channelManager.channel(name)
156            .orElseGet(() -> channelManager.createChannel(name));
157        if (vmDef.data() != null) {
158            // New data, augment and save
159            addExtraData(vmDef, channel.vmDefinition());
160            channel.setVmDefinition(vmDef);
161        } else {
162            // Reuse cached (e.g. if deleted)
163            vmDef = channel.vmDefinition();
164        }
165        if (vmDef == null) {
166            logger.warning(() -> "Cannot get defintion for "
167                + response.object.getMetadata());
168            return;
169        }
170        channelManager.put(name, channel, preparing);
171
172        // Create and fire changed event. Remove channel from channel
173        // manager on completion.
174        VmResourceChanged chgEvt
175            = new VmResourceChanged(ResponseType.valueOf(response.type), vmDef,
176                channel.setGeneration(response.object.getMetadata()
177                    .getGeneration()),
178                false);
179        if (ResponseType.valueOf(response.type) == ResponseType.DELETED) {
180            chgEvt = Event.onCompletion(chgEvt,
181                e -> channelManager.remove(e.vmDefinition().name()));
182        }
183        channel.fire(chgEvt);
184    }
185
186    private VmDefinition getModel(K8sClient client, VmDefinition vmDef) {
187        try {
188            return VmDefinitionStub.get(client, context(), namespace(),
189                vmDef.metadata().getName()).model().orElse(null);
190        } catch (ApiException e) {
191            return null;
192        }
193    }
194
195    private void addExtraData(VmDefinition vmDef, VmDefinition prevState) {
196        var extra = new VmExtraData(vmDef);
197        var prevExtra = Optional.ofNullable(prevState).map(VmDefinition::extra);
198
199        // Maintain (or initialize) the resetCount
200        extra.resetCount(prevExtra.map(VmExtraData::resetCount).orElse(0L));
201
202        // Maintain node info
203        prevExtra
204            .ifPresent(e -> extra.nodeInfo(e.nodeName(), e.nodeAddresses()));
205    }
206
207    /**
208     * On pod changed.
209     *
210     * @param event the event
211     * @param channel the channel
212     */
213    @Handler
214    public void onPodChanged(PodChanged event, VmChannel channel) {
215        var vmDef = channel.vmDefinition();
216
217        // Make sure that this is properly sync'd with VM CR changes.
218        channelManager.associated(vmDef.name())
219            .orElseGet(this::activeEventPipeline)
220            .submit("NodeInfo[" + vmDef.name() + "]",
221                () -> {
222                    updateNodeInfo(event, vmDef);
223                    channel.fire(new VmResourceChanged(ResponseType.MODIFIED,
224                        vmDef, false, true));
225                });
226    }
227
228    private void updateNodeInfo(PodChanged event, VmDefinition vmDef) {
229        var extra = vmDef.extra();
230        if (event.type() == ResponseType.DELETED) {
231            // The status of a deleted pod is the status before deletion,
232            // i.e. the node info is still cached and must be removed.
233            extra.nodeInfo("", Collections.emptyList());
234            return;
235        }
236
237        // Get current node info from pod
238        var pod = event.pod();
239        var nodeName = Optional
240            .ofNullable(pod.getSpec().getNodeName()).orElse("");
241        logger.finer(() -> "Adding node name " + nodeName
242            + " to VM info for " + vmDef.name());
243        var addrs = new ArrayList<String>();
244        Optional.ofNullable(pod.getStatus().getPodIPs())
245            .orElse(Collections.emptyList()).stream()
246            .map(V1PodIP::getIp).forEach(addrs::add);
247        logger.finer(() -> "Adding node addresses " + addrs
248            + " to VM info for " + vmDef.name());
249        extra.nodeInfo(nodeName, addrs);
250    }
251
252    /**
253     * On modify vm.
254     *
255     * @param event the event
256     * @throws ApiException the api exception
257     * @throws IOException Signals that an I/O exception has occurred.
258     */
259    @Handler
260    public void onModifyVm(ModifyVm event, VmChannel channel)
261            throws ApiException, IOException {
262        patchVmDef(channel.client(), event.name(), "spec/vm/" + event.path(),
263            event.value());
264    }
265
266    private void patchVmDef(K8sClient client, String name, String path,
267            Object value) throws ApiException, IOException {
268        var vmStub = K8sDynamicStub.get(client,
269            new GroupVersionKind(Crd.GROUP, "", Crd.KIND_VM), namespace(),
270            name);
271
272        // Patch running
273        String valueAsText = value instanceof String
274            ? "\"" + value + "\""
275            : value.toString();
276        var res = vmStub.patch(V1Patch.PATCH_FORMAT_JSON_PATCH,
277            new V1Patch("[{\"op\": \"replace\", \"path\": \"/"
278                + path + "\", \"value\": " + valueAsText + "}]"),
279            client.defaultPatchOptions());
280        if (!res.isPresent()) {
281            logger.warning(
282                () -> "Cannot patch definition for Vm " + vmStub.name());
283        }
284    }
285
286    /**
287     * Attempt to Update the assignment information in the status of the
288     * VM CR. Returns true if successful. The handler does not attempt
289     * retries, because in case of failure it will be necessary to
290     * re-evaluate the chosen VM.
291     *
292     * @param event the event
293     * @param channel the channel
294     * @throws ApiException the api exception
295     */
296    @Handler
297    public void onUpdatedAssignment(UpdateAssignment event, VmChannel channel)
298            throws ApiException {
299        try {
300            var vmDef = channel.vmDefinition();
301            var vmStub = VmDefinitionStub.get(channel.client(),
302                new GroupVersionKind(Crd.GROUP, "", Crd.KIND_VM),
303                vmDef.namespace(), vmDef.name());
304            if (vmStub.updateStatus(vmDef, from -> {
305                JsonObject status = from.statusJson();
306                if (event.toUser() == null) {
307                    ((JsonObject) GsonPtr.to(status).get())
308                        .remove(Status.ASSIGNMENT);
309                } else {
310                    var assignment = GsonPtr.to(status).to(Status.ASSIGNMENT);
311                    assignment.set("pool", event.fromPool().name());
312                    assignment.set("user", event.toUser());
313                    assignment.set("lastUsed", Instant.now().toString());
314                }
315                return status;
316            }).isPresent()) {
317                event.setResult(true);
318            }
319        } catch (ApiException e) {
320            // Log exceptions except for conflict, which can be expected
321            if (HttpURLConnection.HTTP_CONFLICT != e.getCode()) {
322                throw e;
323            }
324        }
325        event.setResult(false);
326    }
327
328}