VirtualMachinePowerStateSyncImpl.java

// 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.vm;

import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

import javax.inject.Inject;

import org.apache.cloudstack.framework.messagebus.MessageBus;
import org.apache.cloudstack.framework.messagebus.PublishScope;
import org.apache.cloudstack.utils.cache.LazyCache;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import com.cloud.agent.api.HostVmStateReportEntry;
import com.cloud.alert.AlertManager;
import com.cloud.configuration.ManagementServiceConfiguration;
import com.cloud.host.Host;
import com.cloud.host.HostVO;
import com.cloud.host.dao.HostDao;
import com.cloud.utils.DateUtil;
import com.cloud.vm.dao.VMInstanceDao;

public class VirtualMachinePowerStateSyncImpl implements VirtualMachinePowerStateSync {
    protected Logger logger = LogManager.getLogger(getClass());

    @Inject MessageBus _messageBus;
    @Inject VMInstanceDao _instanceDao;
    @Inject HostDao hostDao;
    @Inject ManagementServiceConfiguration mgmtServiceConf;
    @Inject AlertManager _alertMgr;

    protected static final String UNKNOWN_INSTANCES_ALERT_SUBJECT = "Instances running on a host that CloudStack has no record of";

    /**
     * The names CloudStack gives the instances it creates: i-[account]-[id]-[suffix] for user instances and
     * [prefix]-[id]-[suffix] for system VMs (r router, s secondary storage, v console proxy, b internal LB,
     * l elastic LB, NS NetScaler VPX). See VirtualMachineName.
     */
    private static final Pattern CLOUDSTACK_INSTANCE_NAME = Pattern.compile("^(i-\\d+-\\d+|(r|s|v|b|l|NS)-\\d+)-.+$");

    private final Map<Long, Set<String>> unknownInstancesByHost = new ConcurrentHashMap<>();

    private LazyCache<Long, VMInstanceVO> vmCache;
    private LazyCache<Long, HostVO> hostCache;

    public VirtualMachinePowerStateSyncImpl() {
        vmCache = new LazyCache<>(16, 10, this::getVmFromId);
        hostCache = new LazyCache<>(16, 10, this::getHostFromId);
    }

    @Override
    public void resetHostSyncState(Host host) {
        logger.info("Reset VM power state sync for host: {}", host);
        _instanceDao.resetHostPowerStateTracking(host.getId());
    }

    @Override
    public void processHostVmStateReport(long hostId, Map<String, HostVmStateReportEntry> report) {
        logger.debug("Process host VM state report. host: {}", hostCache.get(hostId));
        Map<Long, VirtualMachine.PowerState> translatedInfo = convertVmStateReport(hostId, report);
        processReport(hostId, translatedInfo, false);
    }

    @Override
    public void processHostVmStatePingReport(long hostId, Map<String, HostVmStateReportEntry> report, boolean force) {
        logger.debug("Process host VM state report from ping process. host: {}", hostCache.get(hostId));
        Map<Long, VirtualMachine.PowerState> translatedInfo = convertVmStateReport(hostId, report);
        processReport(hostId, translatedInfo, force);
    }

    protected void updateAndPublishVmPowerStates(long hostId, Map<Long, VirtualMachine.PowerState> instancePowerStates,
               Date updateTime) {
        if (instancePowerStates.isEmpty()) {
            return;
        }
        Set<Long> vmIds = instancePowerStates.keySet();
        Map<Long, VirtualMachine.PowerState> notUpdated =
                _instanceDao.updatePowerState(instancePowerStates, hostId, updateTime);
        if (notUpdated.size() > vmIds.size()) {
            return;
        }
        for (Long vmId : vmIds) {
            if (MapUtils.isEmpty(notUpdated) || !notUpdated.containsKey(vmId)) {
                logger.debug("VM state report is updated. {}, {}, power state: {}",
                        () -> hostCache.get(hostId), () -> vmCache.get(vmId), () -> instancePowerStates.get(vmId));
                _messageBus.publish(null, VirtualMachineManager.Topics.VM_POWER_STATE,
                        PublishScope.GLOBAL, vmId);
                continue;
            }
            logger.trace("VM power state does not change, skip DB writing. {}", () -> vmCache.get(vmId));
        }
    }

    private List<VMInstanceVO> filterOutdatedFromMissingVmReport(List<VMInstanceVO> vmsThatAreMissingReport) {
        List<Long> outdatedVms = vmsThatAreMissingReport.stream()
                .filter(v -> !_instanceDao.isPowerStateUpToDate(v))
                .map(VMInstanceVO::getId)
                .collect(Collectors.toList());
        if (CollectionUtils.isEmpty(outdatedVms)) {
            return vmsThatAreMissingReport;
        }
        _instanceDao.resetVmPowerStateTracking(outdatedVms);
        return vmsThatAreMissingReport.stream()
                .filter(v -> !outdatedVms.contains(v.getId()))
                .collect(Collectors.toList());
    }

    /**
     * A host report only lists the instances that were running on the host when the report was collected. When the
     * management server changes an instance's state around that moment, an in-flight report may have been collected
     * before the change and therefore says nothing about it. Treating such an instance as missing would undo the
     * change that just happened, so instances whose state changed within the graceful period are left alone and
     * judged by a later report instead.
     */
    protected boolean hasRecentStateChange(VMInstanceVO instance, Date currentTime, long milliSecondsGracefulPeriod) {
        Date lastStateChange = instance.getUpdateTime();
        if (lastStateChange == null) {
            return false;
        }
        return currentTime.getTime() - lastStateChange.getTime() < milliSecondsGracefulPeriod;
    }

    private void processMissingVmReport(long hostId, Set<Long> vmIds, boolean force) {
        // any state outdates should be checked against the time before this list was retrieved
        Date startTime = DateUtil.currentGMTTime();
        // for all running/stopping VMs, we provide monitoring of missing report
        List<VMInstanceVO> vmsThatAreMissingReport = _instanceDao.findByHostInStatesExcluding(hostId, vmIds,
                VirtualMachine.State.Running, VirtualMachine.State.Stopping, VirtualMachine.State.Starting);
        // here we need to be wary of out of band migration as opposed to other, more unexpected state changes
        if (vmsThatAreMissingReport.isEmpty()) {
            return;
        }
        Date currentTime = DateUtil.currentGMTTime();
        logger.debug("Run missing VM report. current time: {}", currentTime.getTime());
        if (!force) {
            vmsThatAreMissingReport = filterOutdatedFromMissingVmReport(vmsThatAreMissingReport);
        }

        // 2 times of sync-update interval for graceful period
        long milliSecondsGracefulPeriod = mgmtServiceConf.getPingInterval() * 2000L;
        Map<Long, VirtualMachine.PowerState> instancePowerStates = new HashMap<>();
        for (VMInstanceVO instance : vmsThatAreMissingReport) {
            Date vmStateUpdateTime = instance.getPowerStateUpdateTime();
            if (vmStateUpdateTime == null) {
                logger.warn("VM power state update time is null, falling back to update time for {}", instance);
                vmStateUpdateTime = instance.getUpdateTime();
                if (vmStateUpdateTime == null) {
                    logger.warn("VM update time is null, falling back to creation time for {}", instance);
                    vmStateUpdateTime = instance.getCreated();
                }
            }
            if (hasRecentStateChange(instance, currentTime, milliSecondsGracefulPeriod)) {
                logger.debug("vm id: {} - state changed at {}, which is within the graceful period ({} ms); " +
                                "the report may have been collected before that change, skipping missing report",
                        instance.getId(), DateUtil.getOutputString(instance.getUpdateTime()), milliSecondsGracefulPeriod);
                continue;
            }
            logger.debug("Detected missing VM. host: {}, vm id: {}({}), power state: {}, last state update: {}",
                    hostId,
                    instance.getId(),
                    instance.getUuid(),
                    VirtualMachine.PowerState.PowerReportMissing,
                    DateUtil.getOutputString(vmStateUpdateTime));
            long milliSecondsSinceLastStateUpdate = currentTime.getTime() - vmStateUpdateTime.getTime();
            if (force || (milliSecondsSinceLastStateUpdate > milliSecondsGracefulPeriod)) {
                logger.debug("vm id: {} - reporting missing (time since last state update: {} ms, graceful period: {} ms, forced: {})",
                        instance.getId(), milliSecondsSinceLastStateUpdate, milliSecondsGracefulPeriod, force);
                // this is where a race condition might have happened if we don't re-fetch the instance;
                // between the startime of this job and the currentTime of this missing-branch
                // an update might have occurred that we should not override in case of out of band migration
                instancePowerStates.put(instance.getId(), VirtualMachine.PowerState.PowerReportMissing);
            } else {
                logger.debug("vm id: {} - time since last state update({} ms) has not passed graceful period ({} ms) yet",
                        instance.getId(), milliSecondsSinceLastStateUpdate, milliSecondsGracefulPeriod);
            }
        }
        updateAndPublishVmPowerStates(hostId, instancePowerStates, startTime);
    }

    private void processReport(long hostId, Map<Long, VirtualMachine.PowerState> translatedInfo, boolean force) {
        logger.debug("Process VM state report. {}, number of records in report: {}. VMs: [{}]",
                () -> hostCache.get(hostId),
                translatedInfo::size,
                () -> translatedInfo.entrySet().stream().map(entry -> entry.getKey() + ":" + entry.getValue())
                        .collect(Collectors.joining(", ")) + "]");
        updateAndPublishVmPowerStates(hostId, translatedInfo, DateUtil.currentGMTTime());

        processMissingVmReport(hostId, translatedInfo.keySet(), force);

        logger.debug("Done with process of VM state report. host: {}", () -> hostCache.get(hostId));
    }

    public Map<Long, VirtualMachine.PowerState> convertVmStateReport(long hostId, Map<String, HostVmStateReportEntry> states) {
        final HashMap<Long, VirtualMachine.PowerState> map = new HashMap<>();
        if (MapUtils.isEmpty(states)) {
            reportUnknownInstances(hostId, new HashSet<>());
            return map;
        }
        Set<String> unknownInstanceNames = new HashSet<>();
        Map<String, Long> nameIdMap = _instanceDao.getNameIdMapForVmInstanceNames(states.keySet());
        for (Map.Entry<String, HostVmStateReportEntry> entry : states.entrySet()) {
            Long id = nameIdMap.get(entry.getKey());
            if (id != null) {
                map.put(id, entry.getValue().getState());
            } else {
                logger.debug("Unable to find matched VM in CloudStack DB. name: {} powerstate: {}", entry.getKey(), entry.getValue());
                if (isRunningCloudStackInstance(entry.getKey(), entry.getValue())) {
                    unknownInstanceNames.add(entry.getKey());
                }
            }
        }
        reportUnknownInstances(hostId, unknownInstanceNames);
        return map;
    }

    /**
     * Whether an entry the database has no record of is worth raising. A host report is not limited to CloudStack's
     * own instances: VMware lists every VM on the host in any power state, including the worker VMs CloudStack creates
     * and destroys for volume operations (named by a bare UUID) and anything else the vCenter places there, and a KVM
     * host may run domains of its own. Only a powered-on instance carrying a CloudStack instance name can be one that
     * CloudStack left running, everything else stays at debug level as before.
     */
    protected static boolean isRunningCloudStackInstance(String name, HostVmStateReportEntry entry) {
        return name != null && entry != null && VirtualMachine.PowerState.PowerOn.equals(entry.getState())
                && CLOUDSTACK_INSTANCE_NAME.matcher(name).matches();
    }

    /**
     * A host reporting an instance that CloudStack has no record of means something is running there unmanaged,
     * usually left behind by a deploy or an expunge that did not reach the host. Previously this produced one debug
     * line per instance per report and nothing else, so it could go unnoticed indefinitely.
     *
     * Logged at warn level, and only when the set of unknown instances on a host changes, so a standing condition
     * does not repeat on every report.
     *
     * @return true when the set changed and was reported, false when there was nothing new to say.
     */
    protected boolean reportUnknownInstances(long hostId, Set<String> unknownInstanceNames) {
        Set<String> previous = unknownInstancesByHost.get(hostId);
        if (unknownInstanceNames.equals(previous) || (CollectionUtils.isEmpty(unknownInstanceNames) && previous == null)) {
            return false;
        }
        if (unknownInstanceNames.isEmpty()) {
            unknownInstancesByHost.remove(hostId);
            logger.info("No unknown instances are reported anymore. host: {}", () -> hostCache.get(hostId));
            return true;
        }
        unknownInstancesByHost.put(hostId, unknownInstanceNames);
        HostVO host = hostCache.get(hostId);
        String names = String.join(", ", unknownInstanceNames);
        logger.warn("Host reports {} instance(s) that do not exist in CloudStack DB, they are running unmanaged. " +
                        "host: {}, instances: [{}]",
                unknownInstanceNames.size(), host, names);
        if (host != null) {
            _alertMgr.sendAlert(AlertManager.AlertType.ALERT_TYPE_SYNC, host.getDataCenterId(), host.getPodId(),
                    UNKNOWN_INSTANCES_ALERT_SUBJECT,
                    String.format("Host %s reports %d instance(s) that do not exist in CloudStack: %s. They are "
                                    + "running unmanaged and may still be holding addresses and storage.",
                            host.getName(), unknownInstanceNames.size(), names));
        }
        return true;
    }

    protected VMInstanceVO getVmFromId(long vmId) {
        return _instanceDao.findById(vmId);
    }

    protected HostVO getHostFromId(long hostId) {
        return hostDao.findById(hostId);
    }
}