diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterEventTypes.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterEventTypes.java index 486a093e4ad4..0f80743189c5 100755 --- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterEventTypes.java +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterEventTypes.java @@ -23,6 +23,7 @@ public class KubernetesClusterEventTypes { public static final String EVENT_KUBERNETES_CLUSTER_STOP = "KUBERNETES.CLUSTER.STOP"; public static final String EVENT_KUBERNETES_CLUSTER_SCALE = "KUBERNETES.CLUSTER.SCALE"; public static final String EVENT_KUBERNETES_CLUSTER_UPGRADE = "KUBERNETES.CLUSTER.UPGRADE"; + public static final String EVENT_KUBERNETES_CLUSTER_VM_MAP_CHECK = "KUBERNETES.CLUSTER.VM.MAP.CHECK"; public static final String EVENT_KUBERNETES_CLUSTER_NODES_ADD = "KUBERNETES.CLUSTER.NODES.ADD"; public static final String EVENT_KUBERNETES_CLUSTER_NODES_REMOVE = "KUBERNETES.CLUSTER.NODES.REMOVE"; } diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImpl.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImpl.java index 035554e38237..3d42f5b18630 100644 --- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImpl.java +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImpl.java @@ -304,6 +304,7 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne private static final String DEFAULT_NETRIS_VPC_NETWORK_OFFERING_FOR_KUBERNETES_SERVICE_DISPLAY_TEXT = "Network Offering for Netris CloudStack Kubernetes Service on VPC"; protected StateMachine2 _stateMachine = KubernetesCluster.State.getStateMachine(); + private KubernetesClusterVmMapChecker vmMapChecker; protected final static List CLUSTER_NODES_TYPES_LIST = Arrays.asList(WORKER.name(), CONTROL.name(), ETCD.name()); @@ -2563,6 +2564,20 @@ public void reallyRun() { } } + /** + * Reports the clusters whose VM map does not match their VMs or their Kubernetes nodes. Changes nothing. + */ + protected void checkKubernetesClusterVmMaps() { + try { + if (vmMapChecker == null) { + vmMapChecker = ComponentContext.inject(new KubernetesClusterVmMapChecker(this)); + } + vmMapChecker.checkClustersIfDue(); + } catch (Exception e) { + logger.warn("Caught exception while checking the VM maps of Kubernetes clusters", e); + } + } + /* Kubernetes cluster scanner checks if the Kubernetes cluster is in desired state. If it detects Kubernetes cluster is not in desired state, it will trigger an event and marks the Kubernetes cluster to be 'Alert' state. For e.g a Kubernetes cluster in 'Running' state should mean all the cluster of node VM's in the custer should be running and @@ -2680,6 +2695,8 @@ public void reallyRun() { } catch (Exception e) { logger.warn("Caught exception while running Kubernetes cluster state scanner", e); } + // starts the check on its own thread, as it needs SSH to each cluster + checkKubernetesClusterVmMaps(); firstRun = false; } } @@ -2803,6 +2820,7 @@ public String getConfigComponentName() { public ConfigKey[] getConfigKeys() { return new ConfigKey[] { KubernetesServiceEnabled, + KubernetesClusterVmMapChecker.KubernetesClusterVmMapCheckEnabled, KubernetesClusterNetworkOffering, KubernetesClusterStartTimeout, KubernetesClusterScaleTimeout, diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterVmMapChecker.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterVmMapChecker.java new file mode 100644 index 000000000000..af45416ef67b --- /dev/null +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterVmMapChecker.java @@ -0,0 +1,486 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package com.cloud.kubernetes.cluster; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Date; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Set; +import java.util.TreeSet; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.regex.Pattern; +import java.util.stream.Collectors; + +import javax.inject.Inject; + +import org.apache.cloudstack.alert.AlertService; +import org.apache.cloudstack.api.ApiCommandResourceType; +import org.apache.cloudstack.framework.config.ConfigKey; +import org.apache.cloudstack.managed.context.ManagedContextRunnable; +import org.apache.commons.codec.digest.DigestUtils; +import org.apache.commons.lang3.StringUtils; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import com.cloud.alert.AlertManager; +import com.cloud.event.ActionEventUtils; +import com.cloud.event.EventVO; +import com.cloud.kubernetes.cluster.actionworkers.KubernetesClusterVmMapCheckWorker; +import com.cloud.kubernetes.cluster.dao.KubernetesClusterDao; +import com.cloud.kubernetes.cluster.dao.KubernetesClusterDetailsDao; +import com.cloud.kubernetes.cluster.dao.KubernetesClusterVmMapDao; +import com.cloud.user.User; +import com.cloud.utils.component.ComponentContext; +import com.cloud.utils.concurrency.NamedThreadFactory; +import com.cloud.utils.db.GlobalLock; +import com.cloud.vm.VMInstanceVO; +import com.cloud.vm.VirtualMachine; +import com.cloud.vm.dao.VMInstanceDao; + +/** + * Compares, for each CloudManaged Kubernetes cluster in Running or Alert state, the VMs CloudStack records for the + * cluster (kubernetes_cluster_vm_map) with the VMs of the cluster in CloudStack and with the nodes Kubernetes knows + * about, and raises an event and an alert for each mismatch found. + *

+ * It only reports: the map, the VMs and the cluster are left as they are. An alert is raised once for a mismatch, + * and again only once it has gone and come back. The mismatches already reported are kept as details of the cluster, + * so that they are not reported again after a restart or by another management server, and are removed once the + * cluster matches again. + */ +public class KubernetesClusterVmMapChecker { + protected Logger logger = LogManager.getLogger(getClass()); + + public static final ConfigKey KubernetesClusterVmMapCheckEnabled = new ConfigKey<>("Advanced", Boolean.class, + "cloud.kubernetes.cluster.vm.map.check.enabled", + "true", + "Indicates whether the Kubernetes cluster status scanner checks, every few minutes, that the VMs CloudStack records " + + "for each Running or Alert CloudManaged Kubernetes cluster match the cluster's VMs and its Kubernetes nodes, " + + "and raises an alert and an event for each mismatch. The check only reports, it changes nothing", + true, + KubernetesClusterService.KubernetesServiceEnabled.key()); + + protected static final long CHECK_INTERVAL_MS = TimeUnit.MINUTES.toMillis(5); + /** + * A running VM that is not a Kubernetes node is reported only once it has been in its state for this long, and a + * VM named as a node of the cluster but not mapped only once it has existed for this long. + */ + protected static final long GRACE_PERIOD_MS = TimeUnit.MINUTES.toMillis(15); + /** The mismatches found by comparing the VM map with the VMs in CloudStack that were reported. */ + protected static final String REPORTED_VM_DRIFT_DETAIL = "vmMapCheckReportedVmDrift"; + /** The mismatches found by comparing the VM map with the Kubernetes nodes that were reported. */ + protected static final String REPORTED_NODE_DRIFT_DETAIL = "vmMapCheckReportedNodeDrift"; + protected static final int MAX_REPORTED_DRIFT_LENGTH = 10000; + protected static final String CHECK_LOCK_NAME = "KubernetesCluster.VmMap.Check.Lock"; + /** Beyond this many new mismatches in a cluster, one alert sums up the rest; each still gets its event. */ + protected static final int MAX_ALERTS_PER_REPORT = 5; + private static final String KEY_SEPARATOR = ","; + private static final String HASHED_KEYS_PREFIX = "sha256:"; + private static final List CHECKED_STATES = Arrays.asList(KubernetesCluster.State.Running, KubernetesCluster.State.Alert); + private static final List GONE_VM_STATES = Arrays.asList(VirtualMachine.State.Destroyed, + VirtualMachine.State.Expunging, VirtualMachine.State.Error); + + protected enum DriftType { + /** Keyed by the VM alone, so a VM going from destroyed to expunged to gone is reported once. */ + MAPPED_VM, + UNMAPPED_VM, + NODE_WITHOUT_VM, + VM_NOT_A_NODE + } + + protected static class Drift { + private final DriftType type; + private final String id; + private final String description; + + protected Drift(DriftType type, String id, String description) { + this.type = type; + this.id = id; + this.description = description; + } + + public DriftType getType() { + return type; + } + + public String getKey() { + return type.name() + ":" + id; + } + + public String getDescription() { + return description; + } + } + + // from the manager, as the plugin's own beans cannot be injected here, as in the action workers + protected KubernetesClusterDao kubernetesClusterDao; + protected KubernetesClusterDetailsDao kubernetesClusterDetailsDao; + protected KubernetesClusterVmMapDao kubernetesClusterVmMapDao; + @Inject + protected VMInstanceDao vmInstanceDao; + @Inject + protected AlertManager alertManager; + + private final KubernetesClusterManagerImpl clusterManager; + private final AtomicBoolean checkRunning = new AtomicBoolean(false); + private ExecutorService executor; + private Long lastCheckNanoTime; + + public KubernetesClusterVmMapChecker(KubernetesClusterManagerImpl clusterManager) { + this.clusterManager = clusterManager; + if (clusterManager != null) { + this.kubernetesClusterDao = clusterManager.kubernetesClusterDao; + this.kubernetesClusterDetailsDao = clusterManager.kubernetesClusterDetailsDao; + this.kubernetesClusterVmMapDao = clusterManager.kubernetesClusterVmMapDao; + } + } + + /** + * Starts a check of the clusters if it is enabled, the last one was long enough ago and it is not still running. + * Called on each run of the cluster status scanner, which runs far more often than the check needs to. The check + * needs SSH to each cluster, so it runs on its own thread, not holding up the scanner and its lock. + */ + public void checkClustersIfDue() { + if (!isCheckEnabled()) { + return; + } + // monotonic, so that a step of the clock does not hold up the checks + long nanoTime = System.nanoTime(); + if ((lastCheckNanoTime != null && nanoTime - lastCheckNanoTime < TimeUnit.MILLISECONDS.toNanos(CHECK_INTERVAL_MS)) || + !checkRunning.compareAndSet(false, true)) { + return; + } + lastCheckNanoTime = nanoTime; + long now = System.currentTimeMillis(); + try { + runInBackground(new ManagedContextRunnable() { + @Override + public void run() { + // here, as the managed context may fail before runInContext is called + try { + super.run(); + } finally { + checkRunning.set(false); + } + } + + @Override + protected void runInContext() { + checkClustersLocked(new Date(now)); + } + }); + } catch (RuntimeException e) { + checkRunning.set(false); + throw e; + } + } + + protected synchronized void runInBackground(Runnable check) { + if (executor == null) { + NamedThreadFactory threadFactory = new NamedThreadFactory("Kubernetes-Cluster-VM-Map-Checker"); + executor = Executors.newSingleThreadExecutor(runnable -> { + Thread thread = threadFactory.newThread(runnable); + thread.setDaemon(true); + return thread; + }); + } + executor.execute(check); + } + + /** Only one management server checks at a time. */ + protected void checkClustersLocked(Date now) { + GlobalLock lock = GlobalLock.getInternLock(CHECK_LOCK_NAME); + try { + if (lock.lock(0)) { + try { + checkClusters(now); + } finally { + lock.unlock(); + } + } + } catch (Exception e) { + logger.warn("Caught exception while checking the VM maps of Kubernetes clusters", e); + } finally { + lock.releaseRef(); + } + } + + protected boolean isCheckEnabled() { + return Boolean.TRUE.equals(KubernetesClusterService.KubernetesServiceEnabled.value()) && + Boolean.TRUE.equals(KubernetesClusterVmMapCheckEnabled.value()); + } + + protected void checkClusters(Date now) { + // by ID, as a cluster can move from one state to the other between the two listings + Map clusters = new LinkedHashMap<>(); + for (KubernetesCluster.State state : CHECKED_STATES) { + kubernetesClusterDao.findManagedKubernetesClustersInState(state).forEach(cluster -> clusters.putIfAbsent(cluster.getId(), cluster)); + } + for (KubernetesClusterVO cluster : clusters.values()) { + try { + checkCluster(cluster.getId(), now); + } catch (Exception e) { + logger.warn("Failed to check the VM map of Kubernetes cluster {}", cluster, e); + } + } + } + + protected boolean isCheckable(KubernetesCluster cluster) { + return cluster != null && + KubernetesCluster.ClusterType.CloudManaged.equals(cluster.getClusterType()) && + CHECKED_STATES.contains(cluster.getState()); + } + + protected KubernetesClusterVmMapCheckWorker createWorker(KubernetesCluster cluster) { + return ComponentContext.inject(new KubernetesClusterVmMapCheckWorker(cluster, clusterManager)); + } + + protected void checkCluster(long clusterId, Date now) { + // a cluster listed a while ago may since have been taken by an operation, during which its map changes + KubernetesClusterVO cluster = kubernetesClusterDao.findById(clusterId); + if (!isCheckable(cluster)) { + return; + } + KubernetesClusterVmMapCheckWorker worker = createWorker(cluster); + + List vmDrift = new ArrayList<>(); + // the VMs of the cluster that are there, mapped or not, by lower case host name + Map clusterVmsByHostName = new HashMap<>(); + List vmMaps = kubernetesClusterVmMapDao.listByClusterId(clusterId); + Set mappedVmIds = new HashSet<>(); + Map liveMappedVms = new LinkedHashMap<>(); + for (KubernetesClusterVmMapVO vmMap : vmMaps) { + mappedVmIds.add(vmMap.getVmId()); + VMInstanceVO vm = vmInstanceDao.findByIdIncludingRemoved(vmMap.getVmId()); + Drift drift = getMappedVmDrift(cluster, vmMap, vm); + if (drift != null) { + vmDrift.add(drift); + // its Kubernetes node, if still there, is already accounted for by this report + if (vm != null) { + addByHostName(clusterVmsByHostName, vm); + } + continue; + } + liveMappedVms.put(vmMap, vm); + addByHostName(clusterVmsByHostName, vm); + } + + Pattern hostNamePattern = worker.getClusterVmHostNamePattern(); + for (VMInstanceVO vm : vmInstanceDao.listByAccountId(cluster.getAccountId())) { + if (mappedVmIds.contains(vm.getId()) || !VirtualMachine.Type.User.equals(vm.getType()) || vm.getRemoved() != null || + GONE_VM_STATES.contains(vm.getState()) || StringUtils.isEmpty(vm.getHostName()) || + !hostNamePattern.matcher(vm.getHostName()).matches()) { + continue; + } + // a VM of another cluster of the same owner whose name happens to match + if (kubernetesClusterVmMapDao.getClusterMapFromVmId(vm.getId()) != null) { + continue; + } + addByHostName(clusterVmsByHostName, vm); + // or one just created by an operation, on this cluster or another with the same name, that has not + // mapped it yet + if (isWithinGracePeriod(vm.getCreated(), now)) { + continue; + } + vmDrift.add(new Drift(DriftType.UNMAPPED_VM, String.valueOf(vm.getId()), + String.format("VM %s (%s, %s) is named as a node of Kubernetes cluster %s but is not one of its VMs in CloudStack", + vm.getHostName(), vm.getUuid(), vm.getState(), cluster.getName()))); + } + + List nodeDrift = null; + List nodeNames = worker.listKubernetesNodeNames(); + if (nodeNames == null) { + logger.debug("Could not list the Kubernetes nodes of cluster {}, comparing its VM map with the VMs in CloudStack only", cluster); + } else { + nodeDrift = getNodeDrift(cluster, nodeNames, liveMappedVms, clusterVmsByHostName, getReportedKeys(clusterId, REPORTED_NODE_DRIFT_DETAIL), now); + } + + // nor report what an operation that ran meanwhile may have changed + if (!isCheckable(kubernetesClusterDao.findById(clusterId)) || !mappedVmIds.equals(getMappedVmIds(clusterId))) { + logger.debug("Kubernetes cluster {} changed during its VM map check, not reporting it", cluster); + return; + } + reportDrift(cluster, REPORTED_VM_DRIFT_DETAIL, vmDrift); + if (nodeDrift != null) { + reportDrift(cluster, REPORTED_NODE_DRIFT_DETAIL, nodeDrift); + } + } + + private Set getReportedKeys(long clusterId, String reportedDetail) { + KubernetesClusterDetailsVO detail = kubernetesClusterDetailsDao.findDetail(clusterId, reportedDetail); + return parseKeys(detail == null ? null : detail.getValue()); + } + + private Set getMappedVmIds(long clusterId) { + return kubernetesClusterVmMapDao.listByClusterId(clusterId).stream().map(KubernetesClusterVmMapVO::getVmId).collect(Collectors.toSet()); + } + + private static void addByHostName(Map vmsByHostName, VMInstanceVO vm) { + if (StringUtils.isNotEmpty(vm.getHostName())) { + vmsByHostName.put(vm.getHostName().toLowerCase(Locale.ROOT), vm); + } + } + + protected Drift getMappedVmDrift(KubernetesCluster cluster, KubernetesClusterVmMapVO vmMap, VMInstanceVO vm) { + String problem; + if (vm == null) { + problem = "does not exist"; + } else if (vm.getRemoved() != null || VirtualMachine.State.Expunging.equals(vm.getState())) { + problem = "has been or is being expunged"; + } else if (VirtualMachine.State.Destroyed.equals(vm.getState())) { + problem = "has been destroyed"; + } else if (VirtualMachine.State.Error.equals(vm.getState())) { + problem = "is in Error state"; + } else { + return null; + } + String vmName = vm == null ? String.format("with ID %d", vmMap.getVmId()) : String.format("%s (%s)", vm.getHostName(), vm.getUuid()); + return new Drift(DriftType.MAPPED_VM, String.valueOf(vmMap.getVmId()), + String.format("VM %s is one of the VMs of Kubernetes cluster %s in CloudStack, but %s", vmName, cluster.getName(), problem)); + } + + protected List getNodeDrift(KubernetesCluster cluster, List nodeNames, Map liveMappedVms, + Map clusterVmsByHostName, Set reportedKeys, Date now) { + List drift = new ArrayList<>(); + Set nodes = new HashSet<>(nodeNames); + // a node may be registered under the fully qualified name of its VM + Set shortNodeNames = nodes.stream().map(n -> StringUtils.substringBefore(n, ".")).collect(Collectors.toSet()); + for (String nodeName : new TreeSet<>(nodes)) { + if (!clusterVmsByHostName.containsKey(nodeName) && !clusterVmsByHostName.containsKey(StringUtils.substringBefore(nodeName, "."))) { + drift.add(new Drift(DriftType.NODE_WITHOUT_VM, nodeName, + String.format("Kubernetes node %s of Kubernetes cluster %s matches none of the cluster's VMs in CloudStack, " + + "neither the VMs in its VM map nor the VMs named as its nodes", nodeName, cluster.getName()))); + } + } + for (Map.Entry entry : liveMappedVms.entrySet()) { + KubernetesClusterVmMapVO vmMap = entry.getKey(); + VMInstanceVO vm = entry.getValue(); + // etcd nodes run etcd only and are not Kubernetes nodes + if (vmMap.isEtcdNode() || !VirtualMachine.State.Running.equals(vm.getState()) || StringUtils.isEmpty(vm.getHostName()) || + nodes.contains(vm.getHostName().toLowerCase(Locale.ROOT)) || shortNodeNames.contains(vm.getHostName().toLowerCase(Locale.ROOT))) { + continue; + } + Drift notANode = new Drift(DriftType.VM_NOT_A_NODE, String.valueOf(vm.getId()), + String.format("VM %s (%s) of Kubernetes cluster %s is running but is not a Kubernetes node of the cluster", + vm.getHostName(), vm.getUuid(), cluster.getName())); + // a node VM that was just created or started may still be joining, unless it was already reported, as + // a restart or a migration does not make it one + if (isWithinNodeJoinGracePeriod(vm, now) && !reportedKeys.contains(notANode.getKey())) { + continue; + } + drift.add(notANode); + } + return drift; + } + + protected static boolean isWithinNodeJoinGracePeriod(VMInstanceVO vm, Date now) { + return isWithinGracePeriod(vm.getCreated(), now) || isWithinGracePeriod(vm.getUpdateTime(), now); + } + + private static boolean isWithinGracePeriod(Date since, Date now) { + return since != null && now.getTime() - since.getTime() < GRACE_PERIOD_MS; + } + + /** + * Raises an event and an alert for each mismatch not reported yet, and records the mismatches reported. A + * mismatch that went away is reported again if it comes back. + */ + protected void reportDrift(KubernetesCluster cluster, String reportedDetail, List drift) { + KubernetesClusterDetailsVO detail = kubernetesClusterDetailsDao.findDetail(cluster.getId(), reportedDetail); + String reported = detail == null ? null : detail.getValue(); + Set keys = drift.stream().map(Drift::getKey).collect(Collectors.toCollection(TreeSet::new)); + String toReport = keys.isEmpty() ? null : serializeKeys(keys); + if (StringUtils.equals(reported, toReport)) { + return; + } + if (toReport == null) { + kubernetesClusterDetailsDao.removeDetail(cluster.getId(), reportedDetail); + String message = String.format("The VM map of Kubernetes cluster %s matches %s again", cluster.getName(), + REPORTED_VM_DRIFT_DETAIL.equals(reportedDetail) ? "the VMs of the cluster in CloudStack" : "the Kubernetes nodes of the cluster"); + logger.info(message); + recordEvent(cluster, EventVO.LEVEL_INFO, message); + return; + } + // recorded first, so that a failure to record does not report the same mismatches on every check + kubernetesClusterDetailsDao.addDetail(cluster.getId(), reportedDetail, toReport, false); + Set reportedKeys = parseKeys(reported); + // distinct, as the map may hold the same VM twice + Set seen = new HashSet<>(reportedKeys); + List newDrift = drift.stream().filter(item -> seen.add(item.getKey())).collect(Collectors.toList()); + for (int i = 0; i < newDrift.size(); i++) { + Drift item = newDrift.get(i); + logger.warn(item.getDescription()); + recordEvent(cluster, EventVO.LEVEL_WARN, item.getDescription()); + if (i < MAX_ALERTS_PER_REPORT) { + sendAlert(cluster, item.getDescription(), String.format("%s. This is only reported, nothing was changed. Check " + + "the VMs of the Kubernetes cluster in CloudStack against its VMs and its Kubernetes nodes.", item.getDescription())); + } + } + if (newDrift.size() > MAX_ALERTS_PER_REPORT) { + String subject = String.format("%d more mismatches between Kubernetes cluster %s and its VMs in CloudStack", + newDrift.size() - MAX_ALERTS_PER_REPORT, cluster.getName()); + sendAlert(cluster, subject, String.format("%s, listed in the events of the cluster.", subject)); + } + } + + /** + * The keys, sorted, or a hash of them if they are too many to be stored. Once hashed, a change to the mismatches + * reports them all again. + */ + protected static String serializeKeys(Set sortedKeys) { + String joined = String.join(KEY_SEPARATOR, sortedKeys); + if (joined.length() <= MAX_REPORTED_DRIFT_LENGTH) { + return joined; + } + return HASHED_KEYS_PREFIX + DigestUtils.sha256Hex(joined); + } + + protected static Set parseKeys(String reported) { + if (StringUtils.isEmpty(reported) || reported.startsWith(HASHED_KEYS_PREFIX)) { + return new HashSet<>(); + } + return new LinkedHashSet<>(Arrays.asList(reported.split(KEY_SEPARATOR))); + } + + protected void recordEvent(KubernetesCluster cluster, String level, String description) { + try { + ActionEventUtils.onCompletedActionEvent(User.UID_SYSTEM, cluster.getAccountId(), level, + KubernetesClusterEventTypes.EVENT_KUBERNETES_CLUSTER_VM_MAP_CHECK, false, description, + cluster.getId(), ApiCommandResourceType.KubernetesCluster.toString(), 0); + } catch (Exception e) { + logger.warn("Unable to record the event \"{}\" for Kubernetes cluster {}", description, cluster, e); + } + } + + protected void sendAlert(KubernetesCluster cluster, String subject, String body) { + try { + alertManager.sendAlert(AlertService.AlertType.ALERT_TYPE_USERVM, cluster.getZoneId(), null, subject, body); + } catch (Exception e) { + logger.warn("Unable to send the alert \"{}\" for Kubernetes cluster {}", subject, cluster, e); + } + } +} diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterVmMapCheckWorker.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterVmMapCheckWorker.java new file mode 100644 index 000000000000..428e51a5e58c --- /dev/null +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterVmMapCheckWorker.java @@ -0,0 +1,119 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package com.cloud.kubernetes.cluster.actionworkers; + +import java.util.ArrayList; +import java.util.List; +import java.util.Locale; +import java.util.regex.Pattern; + +import org.apache.commons.lang3.StringUtils; + +import com.cloud.kubernetes.cluster.KubernetesCluster; +import com.cloud.kubernetes.cluster.KubernetesClusterManagerImpl; +import com.cloud.utils.Pair; +import com.cloud.utils.exception.CloudRuntimeException; +import com.cloud.utils.ssh.SshHelper; + +/** + * Gathers what the VM map check of the Kubernetes cluster status scanner needs from the cluster: the pattern of the + * host names CloudStack gives the VMs it creates for the cluster, and the names of the nodes Kubernetes knows about. + * It only reads, and changes nothing on the cluster. + */ +public class KubernetesClusterVmMapCheckWorker extends KubernetesClusterResourceModifierActionWorker { + + protected static final String LIST_NODES_COMMAND = "sudo /opt/bin/kubectl get nodes -o name"; + private static final String NODE_RESOURCE_PREFIX = "node/"; + private static final int LEGACY_MAX_CLUSTER_PREFIX_LENGTH = 40; + /** The names end in the creation time in milliseconds in hex, 11 digits since 2004, so not one a user picks by chance. */ + private static final int MIN_NAME_SUFFIX_LENGTH = 10; + + public KubernetesClusterVmMapCheckWorker(final KubernetesCluster kubernetesCluster, final KubernetesClusterManagerImpl clusterManager) { + super(kubernetesCluster, clusterManager); + } + + /** + * @return the pattern matching the host names of the control, worker and etcd node VMs that CloudStack creates + * for the cluster: {@code -control-}, {@code -node-} and + * {@code --}, with the prefix truncated to 40 characters as well, as it was before + * it was allowed 43. + */ + public Pattern getClusterVmHostNamePattern() { + String prefix = getKubernetesClusterNodeNamePrefix(); + String nodePrefix = Pattern.quote(prefix); + if (prefix.length() > LEGACY_MAX_CLUSTER_PREFIX_LENGTH) { + nodePrefix = String.format("(?:%s|%s)", nodePrefix, Pattern.quote(prefix.substring(0, LEGACY_MAX_CLUSTER_PREFIX_LENGTH))); + } + String etcdPrefix = Pattern.quote(getEtcdNodeNameForCluster()); + return Pattern.compile(String.format("^(?:%s-(?:control|node)|%s-\\d+)-[0-9a-f]{%d,}$", + nodePrefix, etcdPrefix, MIN_NAME_SUFFIX_LENGTH), Pattern.CASE_INSENSITIVE); + } + + /** + * Lists the nodes of the cluster with kubectl on its control node, over SSH as the other operations do. + * + * @return the lower case names of the nodes, or null if they could not be listed + */ + public List listKubernetesNodeNames() { + try { + Pair sshIpPort = getKubernetesClusterServerIpSshPort(null); + if (StringUtils.isEmpty(sshIpPort.first())) { + logger.debug("No address to reach the control node of Kubernetes cluster {} over SSH", kubernetesCluster); + return null; + } + Pair result = SshHelper.sshExecute(sshIpPort.first(), sshIpPort.second(), getLoginUser(), + getManagementServerSshPublicKeyFile(), null, LIST_NODES_COMMAND, 10000, 10000, 20000); + if (!Boolean.TRUE.equals(result.first())) { + logger.debug("Failed to list the nodes of Kubernetes cluster {}: {}", kubernetesCluster, result.second()); + return null; + } + // kubectl succeeded, so no node listed means the cluster has none + return parseNodeNames(result.second()); + } catch (Exception e) { + logger.debug("Failed to list the nodes of Kubernetes cluster {}", kubernetesCluster, e); + return null; + } + } + + private String getLoginUser() { + try { + return getControlNodeLoginUser(); + } catch (CloudRuntimeException e) { + // the VM of the first map entry may itself be gone, which is what the check is looking for + return CLUSTER_NODE_VM_USER; + } + } + + /** + * Parses the output of {@code kubectl get nodes -o name}, one {@code node/} per line. Anything else, such as + * a warning, is ignored. + */ + protected static List parseNodeNames(String output) { + List nodeNames = new ArrayList<>(); + if (StringUtils.isBlank(output)) { + return nodeNames; + } + for (String line : output.split("\\R")) { + String trimmed = line.trim(); + if (trimmed.startsWith(NODE_RESOURCE_PREFIX) && trimmed.length() > NODE_RESOURCE_PREFIX.length()) { + nodeNames.add(trimmed.substring(NODE_RESOURCE_PREFIX.length()).toLowerCase(Locale.ROOT)); + } + } + return nodeNames; + } +} diff --git a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/KubernetesClusterVmMapCheckerTest.java b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/KubernetesClusterVmMapCheckerTest.java new file mode 100644 index 000000000000..966b81a62b1b --- /dev/null +++ b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/KubernetesClusterVmMapCheckerTest.java @@ -0,0 +1,591 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package com.cloud.kubernetes.cluster; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.Date; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.TreeSet; +import java.util.concurrent.TimeUnit; +import java.util.regex.Pattern; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.ArgumentCaptor; +import org.mockito.ArgumentMatchers; +import org.mockito.Mock; +import org.mockito.Mockito; +import org.mockito.junit.MockitoJUnitRunner; + +import com.cloud.alert.AlertManager; +import com.cloud.event.EventVO; +import com.cloud.kubernetes.cluster.actionworkers.KubernetesClusterVmMapCheckWorker; +import com.cloud.kubernetes.cluster.dao.KubernetesClusterDao; +import com.cloud.kubernetes.cluster.dao.KubernetesClusterDetailsDao; +import com.cloud.kubernetes.cluster.dao.KubernetesClusterVmMapDao; +import com.cloud.vm.VMInstanceVO; +import com.cloud.vm.VirtualMachine; +import com.cloud.vm.dao.VMInstanceDao; + +@RunWith(MockitoJUnitRunner.Silent.class) +public class KubernetesClusterVmMapCheckerTest { + + private static final long CLUSTER_ID = 1L; + private static final long ACCOUNT_ID = 2L; + private static final long OTHER_CLUSTER_ID = 3L; + private static final Pattern HOST_NAME_PATTERN = Pattern.compile("^k8s-(?:control|node)-[0-9a-f]+$"); + + @Mock + private KubernetesClusterDao kubernetesClusterDao; + @Mock + private KubernetesClusterDetailsDao kubernetesClusterDetailsDao; + @Mock + private KubernetesClusterVmMapDao kubernetesClusterVmMapDao; + @Mock + private VMInstanceDao vmInstanceDao; + @Mock + private AlertManager alertManager; + @Mock + private KubernetesClusterVmMapCheckWorker worker; + @Mock + private KubernetesClusterVO cluster; + + private KubernetesClusterVmMapChecker checker; + private final Map details = new HashMap<>(); + private final List vmMaps = new ArrayList<>(); + private final Map vms = new HashMap<>(); + private final List accountVms = new ArrayList<>(); + private final List events = new ArrayList<>(); + private final List alerts = new ArrayList<>(); + private List nodeNames; + private final Date now = new Date(); + + @Before + public void setUp() { + checker = Mockito.spy(new KubernetesClusterVmMapChecker(null)); + checker.kubernetesClusterDao = kubernetesClusterDao; + checker.kubernetesClusterDetailsDao = kubernetesClusterDetailsDao; + checker.kubernetesClusterVmMapDao = kubernetesClusterVmMapDao; + checker.vmInstanceDao = vmInstanceDao; + checker.alertManager = alertManager; + Mockito.doReturn(worker).when(checker).createWorker(ArgumentMatchers.any()); + Mockito.doAnswer(i -> events.add(i.getArgument(1) + " " + i.getArgument(2))).when(checker) + .recordEvent(ArgumentMatchers.any(), ArgumentMatchers.anyString(), ArgumentMatchers.anyString()); + Mockito.doAnswer(i -> alerts.add(i.getArgument(1))).when(checker) + .sendAlert(ArgumentMatchers.any(), ArgumentMatchers.anyString(), ArgumentMatchers.anyString()); + + Mockito.when(cluster.getId()).thenReturn(CLUSTER_ID); + Mockito.when(cluster.getName()).thenReturn("k8s"); + Mockito.when(cluster.getAccountId()).thenReturn(ACCOUNT_ID); + Mockito.when(cluster.getZoneId()).thenReturn(1L); + Mockito.when(cluster.getClusterType()).thenReturn(KubernetesCluster.ClusterType.CloudManaged); + Mockito.when(cluster.getState()).thenReturn(KubernetesCluster.State.Running); + Mockito.when(kubernetesClusterDao.findById(CLUSTER_ID)).thenReturn(cluster); + + Mockito.when(kubernetesClusterDetailsDao.findDetail(ArgumentMatchers.eq(CLUSTER_ID), ArgumentMatchers.anyString())).thenAnswer(i -> { + String value = details.get(i.getArgument(1)); + return value == null ? null : new KubernetesClusterDetailsVO(CLUSTER_ID, i.getArgument(1), value, false); + }); + Mockito.doAnswer(i -> details.put(i.getArgument(1), i.getArgument(2))).when(kubernetesClusterDetailsDao) + .addDetail(ArgumentMatchers.eq(CLUSTER_ID), ArgumentMatchers.anyString(), ArgumentMatchers.anyString(), ArgumentMatchers.eq(false)); + Mockito.doAnswer(i -> details.remove(i.getArgument(1))).when(kubernetesClusterDetailsDao) + .removeDetail(ArgumentMatchers.eq(CLUSTER_ID), ArgumentMatchers.anyString()); + + Mockito.when(kubernetesClusterVmMapDao.listByClusterId(CLUSTER_ID)).thenAnswer(i -> + vmMaps.stream().filter(m -> m.getClusterId() == CLUSTER_ID).collect(Collectors.toList())); + Mockito.when(kubernetesClusterVmMapDao.getClusterMapFromVmId(ArgumentMatchers.anyLong())).thenAnswer(i -> + vmMaps.stream().filter(m -> m.getVmId() == i.getArgument(0)).findFirst().orElse(null)); + Mockito.when(vmInstanceDao.findByIdIncludingRemoved(ArgumentMatchers.anyLong())).thenAnswer(i -> vms.get(i.getArgument(0))); + Mockito.when(vmInstanceDao.listByAccountId(ACCOUNT_ID)).thenReturn(accountVms); + Mockito.when(worker.getClusterVmHostNamePattern()).thenReturn(HOST_NAME_PATTERN); + Mockito.when(worker.listKubernetesNodeNames()).thenAnswer(i -> nodeNames); + } + + private VMInstanceVO addVm(long id, String hostName, VirtualMachine.State state, boolean mapped, boolean etcd) { + VMInstanceVO vm = Mockito.mock(VMInstanceVO.class); + Mockito.when(vm.getId()).thenReturn(id); + Mockito.when(vm.getUuid()).thenReturn("uuid-" + id); + Mockito.when(vm.getHostName()).thenReturn(hostName); + Mockito.when(vm.getState()).thenReturn(state); + Mockito.when(vm.getType()).thenReturn(VirtualMachine.Type.User); + Date longAgo = new Date(now.getTime() - TimeUnit.HOURS.toMillis(1)); + Mockito.when(vm.getCreated()).thenReturn(longAgo); + Mockito.when(vm.getUpdateTime()).thenReturn(longAgo); + vms.put(id, vm); + if (!VirtualMachine.State.Expunging.equals(state)) { + accountVms.add(vm); + } + if (mapped) { + addVmMap(CLUSTER_ID, id, etcd); + } + return vm; + } + + private void addVmMap(long clusterId, long vmId, boolean etcd) { + KubernetesClusterVmMapVO vmMap = new KubernetesClusterVmMapVO(clusterId, vmId, false); + vmMap.setEtcdNode(etcd); + vmMaps.add(vmMap); + } + + /** A healthy cluster: one control and two worker nodes, all of them Kubernetes nodes. */ + private void addHealthyCluster() { + addVm(10L, "k8s-control-1a", VirtualMachine.State.Running, true, false); + addVm(11L, "k8s-node-2b", VirtualMachine.State.Running, true, false); + addVm(12L, "k8s-node-3c", VirtualMachine.State.Running, true, false); + nodeNames = new ArrayList<>(Arrays.asList("k8s-control-1a", "k8s-node-2b", "k8s-node-3c")); + } + + private void check() { + events.clear(); + alerts.clear(); + checker.checkCluster(CLUSTER_ID, now); + } + + private Set reportedKeys(String detail) { + return KubernetesClusterVmMapChecker.parseKeys(details.get(detail)).stream().collect(Collectors.toCollection(TreeSet::new)); + } + + @Test + public void pluginDaosAreTakenFromTheManager() { + KubernetesClusterManagerImpl manager = Mockito.mock(KubernetesClusterManagerImpl.class); + manager.kubernetesClusterDao = kubernetesClusterDao; + manager.kubernetesClusterDetailsDao = kubernetesClusterDetailsDao; + manager.kubernetesClusterVmMapDao = kubernetesClusterVmMapDao; + KubernetesClusterVmMapChecker fromManager = new KubernetesClusterVmMapChecker(manager); + Assert.assertSame(kubernetesClusterDao, fromManager.kubernetesClusterDao); + Assert.assertSame(kubernetesClusterDetailsDao, fromManager.kubernetesClusterDetailsDao); + Assert.assertSame(kubernetesClusterVmMapDao, fromManager.kubernetesClusterVmMapDao); + } + + @Test + public void healthyClusterReportsNothing() { + addHealthyCluster(); + check(); + Assert.assertTrue(events.isEmpty()); + Assert.assertTrue(alerts.isEmpty()); + Assert.assertTrue(details.isEmpty()); + } + + @Test + public void mappedVmThatDoesNotExistIsReported() { + addHealthyCluster(); + addVmMap(CLUSTER_ID, 99L, false); + check(); + Assert.assertEquals(1, alerts.size()); + Assert.assertTrue(alerts.get(0).contains("VM with ID 99") && alerts.get(0).contains("k8s")); + Assert.assertEquals(1, events.size()); + Assert.assertTrue(events.get(0).startsWith(EventVO.LEVEL_WARN)); + Assert.assertEquals(Set.of("MAPPED_VM:99"), reportedKeys(KubernetesClusterVmMapChecker.REPORTED_VM_DRIFT_DETAIL)); + } + + @Test + public void mappedVmThatIsDestroyedIsReported() { + addHealthyCluster(); + addVm(13L, "k8s-node-4d", VirtualMachine.State.Destroyed, true, false); + check(); + Assert.assertEquals(1, alerts.size()); + Assert.assertTrue(alerts.get(0).contains("k8s-node-4d") && alerts.get(0).contains("destroyed")); + Assert.assertEquals(Set.of("MAPPED_VM:13"), reportedKeys(KubernetesClusterVmMapChecker.REPORTED_VM_DRIFT_DETAIL)); + } + + @Test + public void mappedVmThatIsExpungedIsReported() { + addHealthyCluster(); + VMInstanceVO vm = addVm(13L, "k8s-node-4d", VirtualMachine.State.Expunging, true, false); + Mockito.when(vm.getRemoved()).thenReturn(now); + check(); + Assert.assertEquals(1, alerts.size()); + Assert.assertTrue(alerts.get(0).contains("k8s-node-4d") && alerts.get(0).contains("expunged")); + Assert.assertEquals(Set.of("MAPPED_VM:13"), reportedKeys(KubernetesClusterVmMapChecker.REPORTED_VM_DRIFT_DETAIL)); + } + + @Test + public void mappedVmInErrorStateIsReported() { + addHealthyCluster(); + addVm(13L, "k8s-node-4d", VirtualMachine.State.Error, true, false); + check(); + Assert.assertEquals(1, alerts.size()); + Assert.assertTrue(alerts.get(0).contains("k8s-node-4d") && alerts.get(0).contains("Error")); + Assert.assertEquals(Set.of("MAPPED_VM:13"), reportedKeys(KubernetesClusterVmMapChecker.REPORTED_VM_DRIFT_DETAIL)); + } + + @Test + public void vmMappedTwiceIsReportedOnce() { + addHealthyCluster(); + addVmMap(CLUSTER_ID, 99L, false); + addVmMap(CLUSTER_ID, 99L, false); + check(); + Assert.assertEquals(1, alerts.size()); + Assert.assertEquals(1, events.size()); + } + + @Test + public void unmappedVmNamedAsANodeOfTheClusterIsReported() { + addHealthyCluster(); + addVm(13L, "k8s-node-4d", VirtualMachine.State.Running, false, false); + nodeNames.add("k8s-node-4d"); + // not reported: names that do not match, VMs on their way out and VMs of another cluster + addVm(14L, "k8s-node-other", VirtualMachine.State.Running, false, false); + addVm(15L, "k8s-node-5e", VirtualMachine.State.Destroyed, false, false); + addVm(16L, "k8s-node-6f", VirtualMachine.State.Error, false, false); + addVm(17L, "k8s-node-7a", VirtualMachine.State.Running, false, false); + addVmMap(OTHER_CLUSTER_ID, 17L, false); + nodeNames.add("k8s-node-7a"); + VMInstanceVO router = addVm(18L, "k8s-node-8b", VirtualMachine.State.Running, false, false); + Mockito.when(router.getType()).thenReturn(VirtualMachine.Type.DomainRouter); + check(); + Assert.assertEquals(Set.of("UNMAPPED_VM:13"), reportedKeys(KubernetesClusterVmMapChecker.REPORTED_VM_DRIFT_DETAIL)); + // the node of the VM of the other cluster has no VM in this one + Assert.assertEquals(Set.of("NODE_WITHOUT_VM:k8s-node-7a"), reportedKeys(KubernetesClusterVmMapChecker.REPORTED_NODE_DRIFT_DETAIL)); + Assert.assertEquals(2, alerts.size()); + } + + @Test + public void unmappedVmJustCreatedIsNotReported() { + addHealthyCluster(); + VMInstanceVO vm = addVm(13L, "k8s-node-4d", VirtualMachine.State.Running, false, false); + Mockito.when(vm.getCreated()).thenReturn(new Date(now.getTime() - TimeUnit.MINUTES.toMillis(1))); + // nor is its node, if it already joined + nodeNames.add("k8s-node-4d"); + check(); + Assert.assertTrue(alerts.isEmpty()); + Assert.assertTrue(details.isEmpty()); + } + + @Test + public void kubernetesNodeWithoutVmIsReported() { + addHealthyCluster(); + nodeNames.add("k8s-node-0ld"); + check(); + Assert.assertEquals(1, alerts.size()); + Assert.assertTrue(alerts.get(0).contains("k8s-node-0ld")); + Assert.assertTrue(details.get(KubernetesClusterVmMapChecker.REPORTED_VM_DRIFT_DETAIL) == null); + Assert.assertEquals(Set.of("NODE_WITHOUT_VM:k8s-node-0ld"), reportedKeys(KubernetesClusterVmMapChecker.REPORTED_NODE_DRIFT_DETAIL)); + } + + @Test + public void nodeOfAMappedVmThatIsDestroyedIsReportedOnce() { + addHealthyCluster(); + addVm(13L, "k8s-node-4d", VirtualMachine.State.Destroyed, true, false); + nodeNames.add("k8s-node-4d"); + check(); + Assert.assertEquals(1, alerts.size()); + Assert.assertEquals(Set.of("MAPPED_VM:13"), reportedKeys(KubernetesClusterVmMapChecker.REPORTED_VM_DRIFT_DETAIL)); + Assert.assertNull(details.get(KubernetesClusterVmMapChecker.REPORTED_NODE_DRIFT_DETAIL)); + } + + @Test + public void nodesRegisteredUnderAFullyQualifiedNameMatchTheirVm() { + addHealthyCluster(); + nodeNames.remove("k8s-node-3c"); + nodeNames.add("k8s-node-3c.example.com"); + check(); + Assert.assertTrue(alerts.isEmpty()); + Assert.assertTrue(details.isEmpty()); + } + + @Test + public void allRunningVmsAreReportedWhenTheClusterHasNoNodes() { + addHealthyCluster(); + nodeNames = new ArrayList<>(); + check(); + Assert.assertEquals(Set.of("VM_NOT_A_NODE:10", "VM_NOT_A_NODE:11", "VM_NOT_A_NODE:12"), + reportedKeys(KubernetesClusterVmMapChecker.REPORTED_NODE_DRIFT_DETAIL)); + } + + @Test + public void alertsAreCappedButEachMismatchGetsAnEvent() { + addHealthyCluster(); + int count = KubernetesClusterVmMapChecker.MAX_ALERTS_PER_REPORT + 3; + for (int i = 0; i < count; i++) { + addVmMap(CLUSTER_ID, 100L + i, false); + } + check(); + Assert.assertEquals(count, events.size()); + Assert.assertEquals(KubernetesClusterVmMapChecker.MAX_ALERTS_PER_REPORT + 1, alerts.size()); + Assert.assertTrue(alerts.get(alerts.size() - 1).startsWith("3 more mismatches")); + check(); + Assert.assertTrue(alerts.isEmpty()); + Assert.assertTrue(events.isEmpty()); + } + + @Test + public void clusterWhoseMapChangedDuringTheCheckIsNotReported() { + addHealthyCluster(); + nodeNames = null; + Mockito.when(worker.listKubernetesNodeNames()).thenAnswer(i -> { + // e.g. a scale down that started and finished while kubectl ran + vmMaps.removeIf(m -> m.getVmId() == 12L); + return null; + }); + check(); + Assert.assertTrue(alerts.isEmpty()); + Assert.assertTrue(details.isEmpty()); + } + + @Test + public void nodeNamesAreMatchedIgnoringCase() { + addHealthyCluster(); + addVm(13L, "K8S-Node-4D", VirtualMachine.State.Running, true, false); + nodeNames.add("k8s-node-4d"); + check(); + Assert.assertTrue(alerts.isEmpty()); + } + + @Test + public void runningMappedVmThatIsNotANodeIsReported() { + addHealthyCluster(); + nodeNames.remove("k8s-node-3c"); + check(); + Assert.assertEquals(1, alerts.size()); + Assert.assertTrue(alerts.get(0).contains("k8s-node-3c")); + Assert.assertEquals(Set.of("VM_NOT_A_NODE:12"), reportedKeys(KubernetesClusterVmMapChecker.REPORTED_NODE_DRIFT_DETAIL)); + } + + @Test + public void mappedVmThatIsNotANodeIsNotReportedWithinTheGracePeriod() { + addHealthyCluster(); + nodeNames.remove("k8s-node-3c"); + Mockito.when(vms.get(12L).getUpdateTime()).thenReturn(new Date(now.getTime() - TimeUnit.MINUTES.toMillis(5))); + check(); + Assert.assertTrue(alerts.isEmpty()); + Assert.assertTrue(details.isEmpty()); + } + + @Test + public void reportedVmThatIsNotANodeStaysReportedThroughTheGracePeriod() { + addHealthyCluster(); + nodeNames.remove("k8s-node-3c"); + check(); + Assert.assertEquals(1, alerts.size()); + // e.g. migrated or restarted: still not a node, and neither cleared nor reported again + Mockito.when(vms.get(12L).getUpdateTime()).thenReturn(new Date(now.getTime() - TimeUnit.MINUTES.toMillis(1))); + check(); + Assert.assertTrue(alerts.isEmpty()); + Assert.assertTrue(events.isEmpty()); + Assert.assertEquals(Set.of("VM_NOT_A_NODE:12"), reportedKeys(KubernetesClusterVmMapChecker.REPORTED_NODE_DRIFT_DETAIL)); + } + + @Test + public void mappedVmIsReportedOnceAsItGoesFromDestroyedToExpungedToGone() { + addHealthyCluster(); + VMInstanceVO vm = addVm(13L, "k8s-node-4d", VirtualMachine.State.Destroyed, true, false); + check(); + Assert.assertEquals(1, alerts.size()); + Mockito.when(vm.getState()).thenReturn(VirtualMachine.State.Expunging); + Mockito.when(vm.getRemoved()).thenReturn(now); + check(); + Assert.assertTrue(alerts.isEmpty()); + vms.remove(13L); + check(); + Assert.assertTrue(alerts.isEmpty()); + Assert.assertTrue(events.isEmpty()); + Assert.assertEquals(Set.of("MAPPED_VM:13"), reportedKeys(KubernetesClusterVmMapChecker.REPORTED_VM_DRIFT_DETAIL)); + } + + @Test + public void mappedVmThatIsNotANodeIsNotReportedWhenStoppedOrEtcd() { + addHealthyCluster(); + addVm(13L, "k8s-node-4d", VirtualMachine.State.Stopped, true, false); + addVm(14L, "k8s-etcd-0-5e", VirtualMachine.State.Running, true, true); + check(); + Assert.assertTrue(alerts.isEmpty()); + } + + @Test + public void sameDriftIsReportedOnce() { + addHealthyCluster(); + addVmMap(CLUSTER_ID, 99L, false); + nodeNames.add("k8s-node-0ld"); + check(); + Assert.assertEquals(2, alerts.size()); + Assert.assertEquals(2, events.size()); + check(); + check(); + Assert.assertTrue(alerts.isEmpty()); + Assert.assertTrue(events.isEmpty()); + } + + @Test + public void onlyNewDriftIsReportedWhenTheDriftChanges() { + addHealthyCluster(); + addVmMap(CLUSTER_ID, 99L, false); + check(); + Assert.assertEquals(1, alerts.size()); + addVmMap(CLUSTER_ID, 98L, false); + check(); + Assert.assertEquals(1, alerts.size()); + Assert.assertTrue(alerts.get(0).contains("VM with ID 98")); + Assert.assertEquals(Set.of("MAPPED_VM:98", "MAPPED_VM:99"), + reportedKeys(KubernetesClusterVmMapChecker.REPORTED_VM_DRIFT_DETAIL)); + // one of them going away changes what is recorded but raises no alert + vmMaps.removeIf(m -> m.getVmId() == 98L); + check(); + Assert.assertTrue(alerts.isEmpty()); + Assert.assertEquals(Set.of("MAPPED_VM:99"), reportedKeys(KubernetesClusterVmMapChecker.REPORTED_VM_DRIFT_DETAIL)); + // and is reported again if it comes back + addVmMap(CLUSTER_ID, 98L, false); + check(); + Assert.assertEquals(1, alerts.size()); + Assert.assertTrue(alerts.get(0).contains("VM with ID 98")); + } + + @Test + public void resolvedDriftIsClearedAndReportedAgainIfItReturns() { + addHealthyCluster(); + addVmMap(CLUSTER_ID, 99L, false); + check(); + Assert.assertEquals(1, alerts.size()); + vmMaps.removeIf(m -> m.getVmId() == 99L); + check(); + Assert.assertTrue(alerts.isEmpty()); + Assert.assertEquals(1, events.size()); + Assert.assertTrue(events.get(0).startsWith(EventVO.LEVEL_INFO)); + Assert.assertTrue(details.isEmpty()); + check(); + Assert.assertTrue(events.isEmpty()); + addVmMap(CLUSTER_ID, 99L, false); + check(); + Assert.assertEquals(1, alerts.size()); + } + + @Test + public void onlyTheVmComparisonRunsWhenTheNodesCannotBeListed() { + addHealthyCluster(); + nodeNames.add("k8s-node-0ld"); + check(); + Assert.assertEquals(1, alerts.size()); + // SSH fails: the VMs are still compared, and what was found on the nodes is neither cleared nor reported again + nodeNames = null; + addVmMap(CLUSTER_ID, 99L, false); + check(); + Assert.assertEquals(1, alerts.size()); + Assert.assertTrue(alerts.get(0).contains("VM with ID 99")); + Assert.assertEquals(1, events.size()); + Assert.assertEquals(Set.of("NODE_WITHOUT_VM:k8s-node-0ld"), reportedKeys(KubernetesClusterVmMapChecker.REPORTED_NODE_DRIFT_DETAIL)); + // and is not reported again once SSH works again + nodeNames = new ArrayList<>(Arrays.asList("k8s-control-1a", "k8s-node-2b", "k8s-node-3c", "k8s-node-0ld")); + check(); + Assert.assertTrue(alerts.isEmpty()); + } + + @Test + public void runningVmsAreNotReportedAsNotNodesWhenTheNodesCannotBeListed() { + addHealthyCluster(); + nodeNames = null; + check(); + Assert.assertTrue(alerts.isEmpty()); + Assert.assertTrue(details.isEmpty()); + } + + @Test + public void clustersInOtherStatesAreNotChecked() { + addHealthyCluster(); + addVmMap(CLUSTER_ID, 99L, false); + for (KubernetesCluster.State state : KubernetesCluster.State.values()) { + if (KubernetesCluster.State.Running.equals(state) || KubernetesCluster.State.Alert.equals(state)) { + continue; + } + Mockito.when(cluster.getState()).thenReturn(state); + check(); + Assert.assertTrue(state.toString(), alerts.isEmpty()); + } + Mockito.verify(checker, Mockito.never()).createWorker(ArgumentMatchers.any()); + Mockito.when(cluster.getState()).thenReturn(KubernetesCluster.State.Alert); + check(); + Assert.assertEquals(1, alerts.size()); + } + + @Test + public void externalManagedClustersAreNotChecked() { + addHealthyCluster(); + addVmMap(CLUSTER_ID, 99L, false); + Mockito.when(cluster.getClusterType()).thenReturn(KubernetesCluster.ClusterType.ExternalManaged); + check(); + Assert.assertTrue(alerts.isEmpty()); + Mockito.verify(checker, Mockito.never()).createWorker(ArgumentMatchers.any()); + } + + @Test + public void clusterTakenByAnOperationDuringTheCheckIsNotReported() { + addHealthyCluster(); + addVmMap(CLUSTER_ID, 99L, false); + KubernetesClusterVO scaling = Mockito.mock(KubernetesClusterVO.class); + Mockito.when(scaling.getClusterType()).thenReturn(KubernetesCluster.ClusterType.CloudManaged); + Mockito.when(scaling.getState()).thenReturn(KubernetesCluster.State.Scaling); + Mockito.when(kubernetesClusterDao.findById(CLUSTER_ID)).thenReturn(cluster, scaling); + check(); + Assert.assertTrue(alerts.isEmpty()); + Assert.assertTrue(details.isEmpty()); + } + + @Test + public void checkClustersChecksRunningAndAlertClustersOnly() { + KubernetesClusterVO running = Mockito.mock(KubernetesClusterVO.class); + Mockito.when(running.getId()).thenReturn(5L); + KubernetesClusterVO alert = Mockito.mock(KubernetesClusterVO.class); + Mockito.when(alert.getId()).thenReturn(6L); + Mockito.when(kubernetesClusterDao.findManagedKubernetesClustersInState(ArgumentMatchers.any())).thenReturn(Collections.emptyList()); + Mockito.when(kubernetesClusterDao.findManagedKubernetesClustersInState(KubernetesCluster.State.Running)).thenReturn(List.of(running)); + // the running cluster moved to Alert between the two listings: checked once + Mockito.when(kubernetesClusterDao.findManagedKubernetesClustersInState(KubernetesCluster.State.Alert)).thenReturn(List.of(running, alert)); + // a failure on one cluster does not stop the others being checked + Mockito.doThrow(new RuntimeException("test")).when(checker).checkCluster(ArgumentMatchers.eq(5L), ArgumentMatchers.any()); + Mockito.doNothing().when(checker).checkCluster(ArgumentMatchers.eq(6L), ArgumentMatchers.any()); + checker.checkClusters(now); + ArgumentCaptor ids = ArgumentCaptor.forClass(Long.class); + Mockito.verify(checker, Mockito.times(2)).checkCluster(ids.capture(), ArgumentMatchers.any()); + Assert.assertEquals(List.of(5L, 6L), ids.getAllValues()); + } + + @Test + public void checkClustersIfDueRunsOnlyWhenEnabledAndOncePerInterval() { + Mockito.doNothing().when(checker).checkClustersLocked(ArgumentMatchers.any()); + Mockito.doAnswer(i -> { + i.getArgument(0).run(); + return null; + }).when(checker).runInBackground(ArgumentMatchers.any()); + Mockito.doReturn(false).when(checker).isCheckEnabled(); + checker.checkClustersIfDue(); + Mockito.verify(checker, Mockito.never()).runInBackground(ArgumentMatchers.any()); + Mockito.doReturn(true).when(checker).isCheckEnabled(); + checker.checkClustersIfDue(); + checker.checkClustersIfDue(); + Mockito.verify(checker, Mockito.times(1)).checkClustersLocked(ArgumentMatchers.any()); + } + + @Test + public void tooManyKeysAreStoredAsAHash() { + Set keys = IntStream.range(0, 2000).mapToObj(i -> "MAPPED_VM:" + i).collect(Collectors.toCollection(TreeSet::new)); + String serialized = KubernetesClusterVmMapChecker.serializeKeys(keys); + Assert.assertTrue(serialized.length() <= KubernetesClusterVmMapChecker.MAX_REPORTED_DRIFT_LENGTH); + Assert.assertTrue(KubernetesClusterVmMapChecker.parseKeys(serialized).isEmpty()); + Assert.assertEquals(serialized, KubernetesClusterVmMapChecker.serializeKeys(new TreeSet<>(keys))); + Set few = new TreeSet<>(List.of("UNMAPPED_VM:1", "NODE_WITHOUT_VM:a")); + Assert.assertEquals(few, new TreeSet<>(KubernetesClusterVmMapChecker.parseKeys(KubernetesClusterVmMapChecker.serializeKeys(few)))); + } +} diff --git a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterVmMapCheckWorkerTest.java b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterVmMapCheckWorkerTest.java new file mode 100644 index 000000000000..8664761461ae --- /dev/null +++ b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterVmMapCheckWorkerTest.java @@ -0,0 +1,141 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package com.cloud.kubernetes.cluster.actionworkers; + +import java.util.List; +import java.util.regex.Pattern; + +import org.junit.Assert; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.ArgumentMatchers; +import org.mockito.Mock; +import org.mockito.MockedStatic; +import org.mockito.Mockito; +import org.mockito.junit.MockitoJUnitRunner; + +import com.cloud.kubernetes.cluster.KubernetesCluster; +import com.cloud.kubernetes.cluster.KubernetesClusterManagerImpl; +import com.cloud.utils.Pair; +import com.cloud.utils.exception.CloudRuntimeException; +import com.cloud.utils.ssh.SshHelper; + +@RunWith(MockitoJUnitRunner.class) +public class KubernetesClusterVmMapCheckWorkerTest { + + @Mock + private KubernetesCluster kubernetesCluster; + @Mock + private KubernetesClusterManagerImpl clusterManager; + + @Test + public void parseNodeNamesKeepsNodesOnly() { + Assert.assertEquals(List.of("k8s-control-1a", "k8s-node-2b"), KubernetesClusterVmMapCheckWorker.parseNodeNames( + "node/k8s-control-1a\r\nWarning: something\n node/K8S-Node-2B \nnode/\n")); + Assert.assertTrue(KubernetesClusterVmMapCheckWorker.parseNodeNames("").isEmpty()); + Assert.assertTrue(KubernetesClusterVmMapCheckWorker.parseNodeNames(null).isEmpty()); + } + + @Test + public void clusterVmHostNamePatternMatchesTheNamesOfTheClusterVmsOnly() { + Mockito.when(kubernetesCluster.getName()).thenReturn("Prod"); + Pattern pattern = new KubernetesClusterVmMapCheckWorker(kubernetesCluster, clusterManager).getClusterVmHostNamePattern(); + for (String name : List.of("prod-control-19a8f2c3d4e", "prod-node-19a8f2c3d4e", "PROD-NODE-19A8F2C3D4E", + "Prod-etcd-0-19a8f2c3d4e")) { + Assert.assertTrue(name, pattern.matcher(name).matches()); + } + // e.g. the nodes of a cluster named prod-node or prod2, or a VM named by hand + for (String name : List.of("prod-node-node-19a8f2c3d4e", "prod2-node-19a8f2c3d4e", "prod-node-x1", + "prod-control", "prod-etcd-19a8f2c3d4e", "other-prod-node-1a", + "prod-node-1a", "prod-node-02", "prod-master", "prod-control-beef")) { + Assert.assertFalse(name, pattern.matcher(name).matches()); + } + } + + @Test + public void clusterVmHostNamePatternMatchesNamesWithThePrefixTruncatedTheOldWay() { + String name = "a234567890b234567890c234567890d234567890e2345"; + Mockito.when(kubernetesCluster.getName()).thenReturn(name); + Pattern pattern = new KubernetesClusterVmMapCheckWorker(kubernetesCluster, clusterManager).getClusterVmHostNamePattern(); + Assert.assertTrue(pattern.matcher(name.substring(0, 43) + "-node-19a8f2c3d4e").matches()); + Assert.assertTrue(pattern.matcher(name.substring(0, 40) + "-node-19a8f2c3d4e").matches()); + Assert.assertFalse(pattern.matcher(name.substring(0, 41) + "-node-19a8f2c3d4e").matches()); + } + + @Test + public void listKubernetesNodeNamesReturnsNullWithoutAnAddress() { + KubernetesClusterVmMapCheckWorker worker = Mockito.spy(new KubernetesClusterVmMapCheckWorker(kubernetesCluster, clusterManager)); + Mockito.doReturn(new Pair<>(null, 22)).when(worker).getKubernetesClusterServerIpSshPort(null); + Assert.assertNull(worker.listKubernetesNodeNames()); + } + + @Test + public void listKubernetesNodeNamesReturnsNullOnFailure() { + KubernetesClusterVmMapCheckWorker worker = Mockito.spy(new KubernetesClusterVmMapCheckWorker(kubernetesCluster, clusterManager)); + Mockito.doThrow(new RuntimeException("test")).when(worker).getKubernetesClusterServerIpSshPort(null); + Assert.assertNull(worker.listKubernetesNodeNames()); + } + + private KubernetesClusterVmMapCheckWorker mockSshWorker() { + KubernetesClusterVmMapCheckWorker worker = Mockito.spy(new KubernetesClusterVmMapCheckWorker(kubernetesCluster, clusterManager)); + Mockito.doReturn(new Pair<>("192.0.2.1", 2222)).when(worker).getKubernetesClusterServerIpSshPort(null); + Mockito.doReturn(null).when(worker).getManagementServerSshPublicKeyFile(); + return worker; + } + + private MockedStatic mockSsh(boolean success, String output) { + MockedStatic sshHelper = Mockito.mockStatic(SshHelper.class); + sshHelper.when(() -> SshHelper.sshExecute(ArgumentMatchers.anyString(), ArgumentMatchers.anyInt(), ArgumentMatchers.anyString(), + ArgumentMatchers.any(), ArgumentMatchers.any(), ArgumentMatchers.eq(KubernetesClusterVmMapCheckWorker.LIST_NODES_COMMAND), + ArgumentMatchers.anyInt(), ArgumentMatchers.anyInt(), ArgumentMatchers.anyInt())).thenReturn(new Pair<>(success, output)); + return sshHelper; + } + + @Test + public void listKubernetesNodeNamesListsTheNodes() { + KubernetesClusterVmMapCheckWorker worker = mockSshWorker(); + Mockito.doReturn("core").when(worker).getControlNodeLoginUser(); + try (MockedStatic ignored = mockSsh(true, "node/k8s-control-1a\nnode/k8s-node-2b\n")) { + Assert.assertEquals(List.of("k8s-control-1a", "k8s-node-2b"), worker.listKubernetesNodeNames()); + } + } + + @Test + public void listKubernetesNodeNamesReturnsNullOnlyWhenKubectlFails() { + KubernetesClusterVmMapCheckWorker worker = mockSshWorker(); + Mockito.doReturn("core").when(worker).getControlNodeLoginUser(); + try (MockedStatic ignored = mockSsh(false, "connection refused")) { + Assert.assertNull(worker.listKubernetesNodeNames()); + } + try (MockedStatic ignored = mockSsh(true, "")) { + Assert.assertEquals(List.of(), worker.listKubernetesNodeNames()); + } + } + + @Test + public void listKubernetesNodeNamesUsesTheDefaultUserWhenTheFirstVmIsGone() { + KubernetesClusterVmMapCheckWorker worker = mockSshWorker(); + Mockito.doThrow(new CloudRuntimeException("gone")).when(worker).getControlNodeLoginUser(); + try (MockedStatic sshHelper = mockSsh(true, "node/k8s-control-1a")) { + Assert.assertEquals(List.of("k8s-control-1a"), worker.listKubernetesNodeNames()); + sshHelper.verify(() -> SshHelper.sshExecute(ArgumentMatchers.anyString(), ArgumentMatchers.anyInt(), + ArgumentMatchers.eq(KubernetesClusterActionWorker.CLUSTER_NODE_VM_USER), ArgumentMatchers.any(), ArgumentMatchers.any(), + ArgumentMatchers.anyString(), ArgumentMatchers.anyInt(), ArgumentMatchers.anyInt(), ArgumentMatchers.anyInt())); + } + } +}