001/*
002 * VM-Operator
003 * Copyright (C) 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 io.kubernetes.client.openapi.ApiException;
022import io.kubernetes.client.openapi.models.V1Pod;
023import io.kubernetes.client.openapi.models.V1PodList;
024import io.kubernetes.client.util.Watch.Response;
025import io.kubernetes.client.util.generic.options.ListOptions;
026import java.io.IOException;
027import java.time.Duration;
028import java.time.Instant;
029import java.util.Map;
030import java.util.Optional;
031import java.util.concurrent.ConcurrentHashMap;
032import java.util.logging.Level;
033import static org.jdrupes.vmoperator.common.Constants.APP_NAME;
034import static org.jdrupes.vmoperator.common.Constants.VM_OP_NAME;
035import org.jdrupes.vmoperator.common.K8sClient;
036import org.jdrupes.vmoperator.common.K8sObserver.ResponseType;
037import org.jdrupes.vmoperator.common.K8sV1PodStub;
038import org.jdrupes.vmoperator.manager.events.ChannelDictionary;
039import org.jdrupes.vmoperator.manager.events.PodChanged;
040import org.jdrupes.vmoperator.manager.events.VmChannel;
041import org.jdrupes.vmoperator.manager.events.VmResourceChanged;
042import org.jgrapes.core.Channel;
043import org.jgrapes.core.annotation.Handler;
044
045/**
046 * Watches for changes of pods that run VMs.
047 */
048public class PodMonitor extends AbstractMonitor<V1Pod, V1PodList, VmChannel> {
049
050    private final ChannelDictionary<String, VmChannel, ?> channelDictionary;
051
052    private final Map<String, PendingChange> pendingChanges
053        = new ConcurrentHashMap<>();
054
055    /**
056     * Instantiates a new pod monitor.
057     *
058     * @param componentChannel the component channel
059     * @param channelDictionary the channel dictionary
060     */
061    @SuppressWarnings("PMD.ConstructorCallsOverridableMethod")
062    public PodMonitor(Channel componentChannel,
063            ChannelDictionary<String, VmChannel, ?> channelDictionary) {
064        super(componentChannel, V1Pod.class, V1PodList.class);
065        this.channelDictionary = channelDictionary;
066        context(K8sV1PodStub.CONTEXT);
067        ListOptions options = new ListOptions();
068        options.setLabelSelector("app.kubernetes.io/name=" + APP_NAME + ","
069            + "app.kubernetes.io/component=" + APP_NAME + ","
070            + "app.kubernetes.io/managed-by=" + VM_OP_NAME);
071        options(options);
072    }
073
074    @Override
075    protected void prepareMonitoring() throws IOException, ApiException {
076        client(new K8sClient());
077    }
078
079    @Override
080    protected void handleChange(K8sClient client, Response<V1Pod> change) {
081        String vmName = change.object.getMetadata().getLabels()
082            .get("app.kubernetes.io/instance");
083        if (vmName == null) {
084            return;
085        }
086        var channel = channelDictionary.channel(vmName).orElse(null);
087        var responseType = ResponseType.valueOf(change.type);
088        if (channel != null && channel.vmDefinition() != null) {
089            pendingChanges.remove(vmName);
090            channel.fire(new PodChanged(change.object, responseType));
091            return;
092        }
093
094        // VM definition not available yet, may happen during startup
095        if (responseType == ResponseType.DELETED) {
096            return;
097        }
098        purgePendingChanges();
099        logger.finer(() -> "Add pending pod change for " + vmName);
100        pendingChanges.put(vmName, new PendingChange(Instant.now(), change));
101    }
102
103    private void purgePendingChanges() {
104        Instant tooOld = Instant.now().minus(Duration.ofMinutes(15));
105        for (var itr = pendingChanges.entrySet().iterator(); itr.hasNext();) {
106            var change = itr.next();
107            if (change.getValue().from().isBefore(tooOld)) {
108                itr.remove();
109                logger.finer(
110                    () -> "Cleaned pending pod change for " + change.getKey());
111            }
112        }
113    }
114
115    /**
116     * Check for pending changes.
117     *
118     * @param event the event
119     * @param channel the channel
120     */
121    @Handler
122    public void onVmResourceChanged(VmResourceChanged event,
123            VmChannel channel) {
124        Optional.ofNullable(pendingChanges.remove(event.vmDefinition().name()))
125            .map(PendingChange::change).ifPresent(change -> {
126                logger.finer(() -> "Firing pending pod change for "
127                    + event.vmDefinition().name());
128                channel.fire(new PodChanged(change.object,
129                    ResponseType.valueOf(change.type)));
130                if (logger.isLoggable(Level.FINER)
131                    && pendingChanges.isEmpty()) {
132                    logger.finer("No pending pod changes left.");
133                }
134            });
135    }
136
137    private record PendingChange(Instant from, Response<V1Pod> change) {
138    }
139
140}