server/src/main/java/com/cloud/network/NetworkUsageManagerImpl.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.network;

import java.net.URI;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;

import javax.inject.Inject;
import javax.naming.ConfigurationException;

import org.springframework.stereotype.Component;

import org.apache.cloudstack.api.command.admin.usage.AddTrafficMonitorCmd;
import org.apache.cloudstack.api.command.admin.usage.DeleteTrafficMonitorCmd;
import org.apache.cloudstack.api.command.admin.usage.ListTrafficMonitorsCmd;
import org.apache.cloudstack.framework.config.dao.ConfigurationDao;

import com.cloud.agent.AgentManager;
import com.cloud.agent.Listener;
import com.cloud.agent.api.AgentControlAnswer;
import com.cloud.agent.api.AgentControlCommand;
import com.cloud.agent.api.Answer;
import com.cloud.agent.api.Command;
import com.cloud.agent.api.DirectNetworkUsageAnswer;
import com.cloud.agent.api.DirectNetworkUsageCommand;
import com.cloud.agent.api.RecurringNetworkUsageCommand;
import com.cloud.agent.api.StartupCommand;
import com.cloud.agent.api.StartupTrafficMonitorCommand;
import com.cloud.agent.manager.Commands;
import com.cloud.configuration.Config;
import com.cloud.dc.DataCenterVO;
import com.cloud.dc.dao.DataCenterDao;
import com.cloud.event.EventTypes;
import com.cloud.event.UsageEventVO;
import com.cloud.event.dao.UsageEventDao;
import com.cloud.exception.AgentUnavailableException;
import com.cloud.exception.InvalidParameterValueException;
import com.cloud.gpu.dao.HostGpuGroupsDao;
import com.cloud.host.DetailVO;
import com.cloud.host.Host;
import com.cloud.host.HostVO;
import com.cloud.host.Status;
import com.cloud.host.dao.HostDao;
import com.cloud.host.dao.HostDetailsDao;
import com.cloud.network.dao.IPAddressDao;
import com.cloud.network.dao.IPAddressVO;
import com.cloud.network.dao.NetworkDao;
import com.cloud.network.dao.NetworkVO;
import com.cloud.network.resource.TrafficSentinelResource;
import com.cloud.resource.ResourceManager;
import com.cloud.resource.ResourceStateAdapter;
import com.cloud.resource.ServerResource;
import com.cloud.resource.UnableDeleteHostException;
import com.cloud.usage.UsageIPAddressVO;
import com.cloud.user.Account;
import com.cloud.user.AccountManager;
import com.cloud.user.UserStatisticsVO;
import com.cloud.user.dao.UserStatisticsDao;
import com.cloud.utils.NumbersUtil;
import com.cloud.utils.component.ManagerBase;
import com.cloud.utils.db.DB;
import com.cloud.utils.db.GlobalLock;
import com.cloud.utils.db.JoinBuilder;
import com.cloud.utils.db.SearchBuilder;
import com.cloud.utils.db.SearchCriteria;
import com.cloud.utils.db.SearchCriteria.Op;
import com.cloud.utils.db.Transaction;
import com.cloud.utils.db.TransactionCallbackNoReturn;
import com.cloud.utils.db.TransactionStatus;
import com.cloud.utils.exception.CloudRuntimeException;
import com.cloud.utils.net.MacAddress;

@Component
public class NetworkUsageManagerImpl extends ManagerBase implements NetworkUsageService, NetworkUsageManager, ResourceStateAdapter {
    public enum NetworkUsageResourceName {
        TrafficSentinel;
    }

    @Inject
    HostDao _hostDao;
    @Inject
    AgentManager _agentMgr;
    @Inject
    IPAddressDao _ipAddressDao;
    @Inject
    UserStatisticsDao _statsDao;
    @Inject
    ConfigurationDao _configDao;
    @Inject
    UsageEventDao _eventDao;
    @Inject
    DataCenterDao _dcDao;
    @Inject
    HostDetailsDao _detailsDao;
    @Inject
    HostGpuGroupsDao _hostGpuGroupsDao;
    @Inject
    AccountManager _accountMgr;
    @Inject
    NetworkDao _networksDao = null;
    @Inject
    ResourceManager _resourceMgr;
    ScheduledExecutorService _executor;
    int _networkStatsInterval;
    String _TSinclZones;
    String _TSexclZones;
    protected SearchBuilder<IPAddressVO> AllocatedIpSearch;

    @Override
    public Host addTrafficMonitor(AddTrafficMonitorCmd cmd) {

        long zoneId = cmd.getZoneId();

        DataCenterVO zone = _dcDao.findById(zoneId);
        if (zone == null) {
            throw new InvalidParameterValueException("Could not find zone with ID: " + zoneId);
        }

        List<HostVO> trafficMonitorsInZone = _resourceMgr.listAllHostsInOneZoneByType(Host.Type.TrafficMonitor, zoneId);
        if (trafficMonitorsInZone.size() != 0) {
            throw new InvalidParameterValueException(String.format("Already added an traffic monitor in zone: %s", zone));
        }

        URI uri;
        try {
            uri = new URI(cmd.getUrl());
        } catch (Exception e) {
            logger.debug(e);
            throw new InvalidParameterValueException(e.getMessage());
        }

        String ipAddress = uri.getHost();
        //String numRetries = params.get("numretries");
        //String timeout = params.get("timeout");

        TrafficSentinelResource resource = new TrafficSentinelResource();
        String guid = getTrafficMonitorGuid(zoneId, NetworkUsageResourceName.TrafficSentinel, ipAddress);

        Map<String, Object> hostParams = new HashMap<String, Object>();
        hostParams.put("zone", String.valueOf(zoneId));
        hostParams.put("ipaddress", ipAddress);
        hostParams.put("url", cmd.getUrl());
        hostParams.put("inclZones", (cmd.getInclZones() != null) ? cmd.getInclZones() : _TSinclZones);
        hostParams.put("exclZones", (cmd.getExclZones() != null) ? cmd.getExclZones() : _TSexclZones);
        hostParams.put("guid", guid);
        hostParams.put("name", guid);

        try {
            resource.configure(guid, hostParams);
        } catch (ConfigurationException e) {
            throw new CloudRuntimeException(e.getMessage());
        }

        Map<String, String> hostDetails = new HashMap<String, String>();
        hostDetails.put("url", cmd.getUrl());
        hostDetails.put("last_collection", "" + System.currentTimeMillis());
        if (cmd.getInclZones() != null) {
            hostDetails.put("inclZones", cmd.getInclZones());
        }
        if (cmd.getExclZones() != null) {
            hostDetails.put("exclZones", cmd.getExclZones());
        }

        Host trafficMonitor = _resourceMgr.addHost(zoneId, resource, Host.Type.TrafficMonitor, hostDetails);
        return trafficMonitor;
    }

    public String getTrafficMonitorGuid(long zoneId, NetworkUsageResourceName name, String ip) {
        return zoneId + "-" + name + "-" + ip;
    }

    @Override
    public boolean deleteTrafficMonitor(DeleteTrafficMonitorCmd cmd) {
        long hostId = cmd.getId();
        HostVO trafficMonitor = _hostDao.findById(hostId);
        if (trafficMonitor == null) {
            throw new InvalidParameterValueException("Could not find an traffic monitor with ID: " + hostId);
        }

        if (_resourceMgr.deleteHost(hostId, false, false)) {
            return true;
        } else {
            return false;
        }
    }

    @Override
    public List<HostVO> listTrafficMonitors(ListTrafficMonitorsCmd cmd) {
        long zoneId = cmd.getZoneId();
        return _resourceMgr.listAllHostsInOneZoneByType(Host.Type.TrafficMonitor, zoneId);
    }

    @Override
    public boolean configure(String name, Map<String, Object> params) throws ConfigurationException {
        AllocatedIpSearch = _ipAddressDao.createSearchBuilder();
        AllocatedIpSearch.and("allocated", AllocatedIpSearch.entity().getAllocatedTime(), Op.NNULL);
        AllocatedIpSearch.and("dc", AllocatedIpSearch.entity().getDataCenterId(), Op.EQ);
        SearchBuilder<NetworkVO> networkJoin = _networksDao.createSearchBuilder();
        networkJoin.and("guestType", networkJoin.entity().getGuestType(), Op.EQ);
        AllocatedIpSearch.join("network", networkJoin, AllocatedIpSearch.entity().getSourceNetworkId(), networkJoin.entity().getId(), JoinBuilder.JoinType.INNER);
        AllocatedIpSearch.done();

        _networkStatsInterval = NumbersUtil.parseInt(_configDao.getValue(Config.DirectNetworkStatsInterval.key()), 86400);
        _TSinclZones = _configDao.getValue(Config.TrafficSentinelIncludeZones.key());
        _TSexclZones = _configDao.getValue(Config.TrafficSentinelExcludeZones.key());
        _agentMgr.registerForHostEvents(new DirectNetworkStatsListener(_networkStatsInterval), true, false, false);
        _resourceMgr.registerResourceStateAdapter(this.getClass().getSimpleName(), this);
        return true;
    }

    @Override
    public boolean start() {
        return true;
    }

    @Override
    public boolean stop() {
        _resourceMgr.unregisterResourceStateAdapter(this.getClass().getSimpleName());
        return true;
    }

    @Override
    public List<IPAddressVO> listAllocatedDirectIps(long zoneId) {
        SearchCriteria<IPAddressVO> sc = AllocatedIpSearch.create();
        sc.setParameters("dc", zoneId);
        sc.setJoinParameters("network", "guestType", Network.GuestType.Shared);
        return _ipAddressDao.search(sc, null);
    }

    protected class DirectNetworkStatsListener implements Listener {

        private int _interval;

        private final long mgmtSrvrId = MacAddress.getMacAddress().toLong();

        protected DirectNetworkStatsListener(int interval) {
            _interval = interval;
        }

        @Override
        public boolean isRecurring() {
            return true;
        }

        @Override
        public boolean processAnswers(long agentId, long seq, Answer[] answers) {
            return processAnswers(agentId, null, null, seq, answers);
        }

        @Override
        @DB
        public boolean processAnswers(long agentId, String uuid, String name, long seq, Answer[] answers) {
            /*
             * Do not collect Direct Network usage stats if the Traffic Monitor is not owned by this mgmt server
             */
            HostVO host = _hostDao.findById(agentId);
            if (host != null) {
                if ((host.getManagementServerId() == null) || (mgmtSrvrId != host.getManagementServerId())) {
                    logger.warn("Not the owner. Not collecting Direct Network usage from TrafficMonitor : {}", host);
                    return false;
                }
            } else {
                logger.warn("Agent not found. Not collecting Direct Network usage from  TrafficMonitor [id: {}, uuid: {}, name: {}", agentId, uuid, name);
                return false;
            }

            GlobalLock scanLock = GlobalLock.getInternLock("direct.network.usage.collect" + host.getDataCenterId());
            try {
                if (scanLock.lock(10)) {
                    try {
                        return collectDirectNetworkUsage(host);
                    } finally {
                        scanLock.unlock();
                    }
                }
            } finally {
                scanLock.releaseRef();
            }
            return false;
        }

        private boolean collectDirectNetworkUsage(final HostVO host) {
            logger.debug("Direct Network Usage stats collector is running...");

            final long zoneId = host.getDataCenterId();
            final DetailVO lastCollectDetail = _detailsDao.findDetail(host.getId(), "last_collection");
            if (lastCollectDetail == null) {
                logger.warn("Last collection time not available. Skipping direct usage collection for Traffic Monitor: {}", host);
                return false;
            }
            Date lastCollection = new Date(Long.parseLong(lastCollectDetail.getValue()));

            //Get list of IPs currently allocated
            List<IPAddressVO> allocatedIps = listAllocatedDirectIps(zoneId);
            Calendar rightNow = Calendar.getInstance();

            // Allow 2 hours for traffic sentinel to populate historical traffic
            // This coule be made configurable

            rightNow.add(Calendar.HOUR_OF_DAY, -2);
            final Date now = rightNow.getTime();

            if (lastCollection.after(now)) {
                logger.debug("Current time is less than 2 hours after last collection time : " + lastCollection.toString() +
                    ". Skipping direct network usage collection");
                return false;
            }

            //Get IP Assign/Release events from lastCollection time till now
            List<UsageEventVO> IpEvents = _eventDao.listDirectIpEvents(lastCollection, now, zoneId);

            Map<String, Date> ipAssigment = new HashMap<String, Date>();
            List<UsageIPAddressVO> IpPartialUsage = new ArrayList<UsageIPAddressVO>(); //Ips which were allocated only for the part of collection duration
            List<UsageIPAddressVO> fullDurationIpUsage = new ArrayList<UsageIPAddressVO>(); //Ips which were allocated only for the entire collection duration

            // Use UsageEvents to track the IP assignment
            // Add them to IpUsage list with account_id , ip_address, alloc_date, release_date

            for (UsageEventVO IpEvent : IpEvents) {
                String address = IpEvent.getResourceName();
                if (EventTypes.EVENT_NET_IP_ASSIGN.equals(IpEvent.getType())) {
                    ipAssigment.put(address, IpEvent.getCreateDate());
                } else if (EventTypes.EVENT_NET_IP_RELEASE.equals(IpEvent.getType())) {
                    if (ipAssigment.containsKey(address)) {
                        Date assigned = ipAssigment.get(address);
                        ipAssigment.remove(address);
                        IpPartialUsage.add(new UsageIPAddressVO(IpEvent.getAccountId(), address, assigned, IpEvent.getCreateDate()));
                    } else {
                        // Ip was assigned prior to lastCollection Date
                        IpPartialUsage.add(new UsageIPAddressVO(IpEvent.getAccountId(), address, lastCollection, IpEvent.getCreateDate()));
                    }
                }
            }

            List<String> IpList = new ArrayList<String>();
            for (IPAddressVO ip : allocatedIps) {
                if (ip.getAllocatedToAccountId() == Account.ACCOUNT_ID_SYSTEM) {
                    //Ignore usage for system account
                    continue;
                }
                String address = (ip.getAddress()).toString();
                if (ipAssigment.containsKey(address)) {
                    // Ip was assigned during the current period but not release till Date now
                    IpPartialUsage.add(new UsageIPAddressVO(ip.getAllocatedToAccountId(), address, ipAssigment.get(address), now));
                } else {
                    // Ip was not assigned or released during current period. Consider entire duration for usage calculation (lastCollection to now)
                    fullDurationIpUsage.add(new UsageIPAddressVO(ip.getAllocatedToAccountId(), address, lastCollection, now));
                    //Store just the Ips to send the list as part of DirectNetworkUsageCommand
                    IpList.add(address);
                }

            }

            final List<UserStatisticsVO> collectedStats = new ArrayList<UserStatisticsVO>();

            //Get usage for Ips which were assigned for the entire duration
            if (fullDurationIpUsage.size() > 0) {
                DirectNetworkUsageCommand cmd = new DirectNetworkUsageCommand(IpList, lastCollection, now, _TSinclZones, _TSexclZones);
                DirectNetworkUsageAnswer answer = (DirectNetworkUsageAnswer)_agentMgr.easySend(host.getId(), cmd);
                if (answer == null || !answer.getResult()) {
                    String details = (answer != null) ? answer.getDetails() : "details unavailable";
                    String msg = String.format("Unable to get network usage stats from %s due to: %s.", host, details);
                    logger.error(msg);
                    return false;
                } else {
                    for (UsageIPAddressVO usageIp : fullDurationIpUsage) {
                        String publicIp = usageIp.getAddress();
                        long[] bytesSentRcvd = answer.get(publicIp);
                        Long bytesSent = bytesSentRcvd[0];
                        Long bytesRcvd = bytesSentRcvd[1];
                        if (bytesSent == null || bytesRcvd == null) {
                            logger.debug("Incorrect bytes for IP: " + publicIp);
                            continue;
                        }
                        if (bytesSent == 0L && bytesRcvd == 0L) {
                            logger.trace("Ignore zero bytes for IP: " + publicIp);
                            continue;
                        }
                        UserStatisticsVO stats = new UserStatisticsVO(usageIp.getAccountId(), zoneId, null, null, null, null);
                        stats.setCurrentBytesSent(bytesSent);
                        stats.setCurrentBytesReceived(bytesRcvd);
                        collectedStats.add(stats);
                    }
                }
            }

            //Get usage for Ips which were assigned for part of the duration period
            for (UsageIPAddressVO usageIp : IpPartialUsage) {
                IpList = new ArrayList<String>();
                IpList.add(usageIp.getAddress());
                DirectNetworkUsageCommand cmd = new DirectNetworkUsageCommand(IpList, usageIp.getAssigned(), usageIp.getReleased(), _TSinclZones, _TSexclZones);
                DirectNetworkUsageAnswer answer = (DirectNetworkUsageAnswer)_agentMgr.easySend(host.getId(), cmd);
                if (answer == null || !answer.getResult()) {
                    String details = (answer != null) ? answer.getDetails() : "details unavailable";
                    String msg = String.format("Unable to get network usage stats from %s due to: %s.", host, details);
                    logger.error(msg);
                    return false;
                } else {
                    String publicIp = usageIp.getAddress();
                    long[] bytesSentRcvd = answer.get(publicIp);
                    Long bytesSent = bytesSentRcvd[0];
                    Long bytesRcvd = bytesSentRcvd[1];
                    if (bytesSent == null || bytesRcvd == null) {
                        logger.debug("Incorrect bytes for IP: " + publicIp);
                        continue;
                    }
                    if (bytesSent == 0L && bytesRcvd == 0L) {
                        logger.trace("Ignore zero bytes for IP: " + publicIp);
                        continue;
                    }
                    UserStatisticsVO stats = new UserStatisticsVO(usageIp.getAccountId(), zoneId, null, null, null, null);
                    stats.setCurrentBytesSent(bytesSent);
                    stats.setCurrentBytesReceived(bytesRcvd);
                    collectedStats.add(stats);

                }
            }

            if (collectedStats.size() == 0) {
                logger.debug("No new direct network stats. No need to persist");
                return false;
            }
            //Persist all the stats and last_collection time in a single transaction
            Transaction.execute(new TransactionCallbackNoReturn() {
                @Override
                public void doInTransactionWithoutResult(TransactionStatus status) {
                    for (UserStatisticsVO stat : collectedStats) {
                        UserStatisticsVO stats = _statsDao.lock(stat.getAccountId(), stat.getDataCenterId(), 0L, null, host.getId(), "DirectNetwork");
                        if (stats == null) {
                            stats = new UserStatisticsVO(stat.getAccountId(), zoneId, null, host.getId(), "DirectNetwork", 0L);
                            stats.setCurrentBytesSent(stat.getCurrentBytesSent());
                            stats.setCurrentBytesReceived(stat.getCurrentBytesReceived());
                            _statsDao.persist(stats);
                        } else {
                            stats.setCurrentBytesSent(stats.getCurrentBytesSent() + stat.getCurrentBytesSent());
                            stats.setCurrentBytesReceived(stats.getCurrentBytesReceived() + stat.getCurrentBytesReceived());
                            _statsDao.update(stats.getId(), stats);
                        }
                    }
                    lastCollectDetail.setValue("" + now.getTime());
                    _detailsDao.update(lastCollectDetail.getId(), lastCollectDetail);
                }
            });

            return true;
        }

        @Override
        public boolean processCommands(long agentId, long seq, Command[] commands) {
            return false;
        }

        @Override
        public AgentControlAnswer processControlCommand(long agentId, AgentControlCommand cmd) {
            return null;
        }

        @Override
        public boolean processDisconnect(long agentId, Status state) {
            return processDisconnect(agentId, null, null, state);
        }

        @Override
        public boolean processDisconnect(long agentId, String uuid, String name, Status state) {
            if (logger.isDebugEnabled()) {
                logger.debug("Disconnected called on [id: {}, uuid: {}, name: {}] with status {}", agentId, uuid, name, state.toString());
            }
            return true;
        }

        @Override
        public void processHostAdded(long hostId) {
        }

        @Override
        public void processConnect(Host agent, StartupCommand cmd, boolean forRebalance) {
            if (cmd instanceof StartupTrafficMonitorCommand) {
                logger.debug("Sending RecurringNetworkUsageCommand to {}", agent);
                RecurringNetworkUsageCommand watch = new RecurringNetworkUsageCommand(_interval);
                try {
                    _agentMgr.send(agent.getId(), new Commands(watch), this);
                } catch (AgentUnavailableException e) {
                    logger.debug("Can not process connect for host {}", agent, e);
                }
            }
            return;
        }

        @Override
        public void processHostAboutToBeRemoved(long hostId) {
        }

        @Override
        public void processHostRemoved(long hostId, long clusterId) {
        }

        @Override
        public boolean processTimeout(long agentId, long seq) {
            return true;
        }

        @Override
        public int getTimeout() {
            return -1;
        }

        protected DirectNetworkStatsListener() {
        }

    }

    @Override
    public HostVO createHostVOForConnectedAgent(HostVO host, StartupCommand[] cmd) {
        // TODO Auto-generated method stub
        return null;
    }

    @Override
    public HostVO createHostVOForDirectConnectAgent(HostVO host, StartupCommand[] startup, ServerResource resource, Map<String, String> details, List<String> hostTags) {
        if (!(startup[0] instanceof StartupTrafficMonitorCommand)) {
            return null;
        }

        host.setType(Host.Type.TrafficMonitor);
        return host;
    }

    @Override
    public DeleteHostAnswer deleteHost(HostVO host, boolean isForced, boolean isForceDeleteStorage) throws UnableDeleteHostException {
        if (host.getType() != Host.Type.TrafficMonitor) {
            return null;
        }

        long hostId = host.getId();
        _agentMgr.disconnectWithoutInvestigation(hostId, Status.Event.Remove);
        _detailsDao.deleteDetails(hostId);
        _hostGpuGroupsDao.deleteGpuEntries(hostId);
        host.setGuid(null);
        _hostDao.update(hostId, host);
        _hostDao.remove(hostId);
        return new DeleteHostAnswer(false);

    }

}