// 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

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

import java.net.InetAddress;

import java.sql.SQLException;

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.TimeZone;

import java.util.concurrent.Executors;

import java.util.concurrent.Future;

import java.util.concurrent.ScheduledExecutorService;

import java.util.concurrent.TimeUnit;

import javax.ejb.Local;

import javax.inject.Inject;

import javax.naming.ConfigurationException;

import com.cloud.event.UsageEventDetailsVO;

import com.cloud.service.ServiceOfferingVO;

import org.apache.log4j.Logger;

import org.apache.cloudstack.framework.config.dao.ConfigurationDao;

import org.apache.cloudstack.managed.context.ManagedContext;

import org.apache.cloudstack.managed.context.ManagedContextRunnable;

import org.apache.cloudstack.usage.UsageTypes;

import org.apache.cloudstack.utils.usage.UsageUtils;

import org.springframework.stereotype.Component;

import com.cloud.alert.AlertManager;

import com.cloud.event.EventTypes;

import com.cloud.event.UsageEventVO;

import com.cloud.event.dao.UsageEventDao;

import com.cloud.usage.dao.UsageDao;

import com.cloud.usage.dao.UsageIPAddressDao;

import com.cloud.usage.dao.UsageJobDao;

import com.cloud.usage.dao.UsageLoadBalancerPolicyDao;

import com.cloud.usage.dao.UsageNetworkDao;

import com.cloud.usage.dao.UsageNetworkOfferingDao;

import com.cloud.usage.dao.UsagePortForwardingRuleDao;

import com.cloud.usage.dao.UsageSecurityGroupDao;

import com.cloud.usage.dao.UsageStorageDao;

import com.cloud.usage.dao.UsageVMInstanceDao;

import com.cloud.usage.dao.UsageVMSnapshotDao;

import com.cloud.usage.dao.UsageVPNUserDao;

import com.cloud.usage.dao.UsageVmDiskDao;

import com.cloud.usage.dao.UsageVolumeDao;

import com.cloud.usage.parser.IPAddressUsageParser;

import com.cloud.usage.parser.LoadBalancerUsageParser;

import com.cloud.usage.parser.NetworkOfferingUsageParser;

import com.cloud.usage.parser.NetworkUsageParser;

import com.cloud.usage.parser.PortForwardingUsageParser;

import com.cloud.usage.parser.SecurityGroupUsageParser;

import com.cloud.usage.parser.StorageUsageParser;

import com.cloud.usage.parser.VMInstanceUsageParser;

import com.cloud.usage.parser.VMSnapshotUsageParser;

import com.cloud.usage.parser.VPNUserUsageParser;

import com.cloud.usage.parser.VmDiskUsageParser;

import com.cloud.usage.parser.VolumeUsageParser;

import com.cloud.user.Account;

import com.cloud.user.AccountVO;

import com.cloud.user.UserStatisticsVO;

import com.cloud.user.VmDiskStatisticsVO;

import com.cloud.user.dao.AccountDao;

import com.cloud.user.dao.UserStatisticsDao;

import com.cloud.user.dao.VmDiskStatisticsDao;

import com.cloud.utils.component.ManagerBase;

import com.cloud.utils.concurrency.NamedThreadFactory;

import com.cloud.utils.db.DB;

import com.cloud.utils.db.Filter;

import com.cloud.utils.db.GlobalLock;

import com.cloud.utils.db.SearchCriteria;

import com.cloud.utils.db.TransactionLegacy;

import com.cloud.event.dao.UsageEventDetailsDao;

@Component

@Local(value={UsageManager.class})

public class UsageManagerImpl extends ManagerBase implements UsageManager, Runnable {

public static final Logger s_logger = Logger.getLogger(UsageManagerImpl.class.getName());

protected static final String DAILY = "DAILY";

protected static final String WEEKLY = "WEEKLY";

protected static final String MONTHLY = "MONTHLY";

private static final int HOURLY_TIME = 60;

private static final int DAILY_TIME = 60 * 24;

private static final int THREE_DAYS_IN_MINUTES = 60 * 24 * 3;

@Inject private AccountDao m_accountDao;

@Inject private UserStatisticsDao m_userStatsDao;

@Inject private UsageDao m_usageDao;

@Inject private UsageVMInstanceDao m_usageInstanceDao;

@Inject private UsageIPAddressDao m_usageIPAddressDao;

@Inject private UsageNetworkDao m_usageNetworkDao;

@Inject private UsageVolumeDao m_usageVolumeDao;

@Inject private UsageStorageDao m_usageStorageDao;

@Inject private UsageLoadBalancerPolicyDao m_usageLoadBalancerPolicyDao;

@Inject private UsagePortForwardingRuleDao m_usagePortForwardingRuleDao;

@Inject private UsageNetworkOfferingDao m_usageNetworkOfferingDao;

@Inject private UsageVPNUserDao m_usageVPNUserDao;

@Inject private UsageSecurityGroupDao m_usageSecurityGroupDao;

@Inject private UsageJobDao m_usageJobDao;

@Inject private VmDiskStatisticsDao m_vmDiskStatsDao;

@Inject private UsageVmDiskDao m_usageVmDiskDao;

@Inject protected AlertManager _alertMgr;

@Inject protected UsageEventDao _usageEventDao;

@Inject protected UsageEventDetailsDao _usageEventDetailsDao;

@Inject ConfigurationDao _configDao;

@Inject private UsageVMSnapshotDao m_usageVMSnapshotDao;

private String m_version = null;

private final Calendar m_jobExecTime = Calendar.getInstance();

private int m_aggregationDuration = 0;

private int m_sanityCheckInterval = 0;

String m_hostname = null;

int m_pid = 0;

TimeZone m_usageTimezone = TimeZone.getTimeZone("GMT");;

private final GlobalLock m_heartbeatLock = GlobalLock.getInternLock("usage.job.heartbeat.check");

private List<UsageNetworkVO> usageNetworks = new ArrayList<UsageNetworkVO>();

private List<UsageVmDiskVO> usageVmDisks = new ArrayList<UsageVmDiskVO>();

private final ScheduledExecutorService m_executor = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("Usage-Job"));

private final ScheduledExecutorService m_heartbeatExecutor = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("Usage-HB"));

private final ScheduledExecutorService m_sanityExecutor = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("Usage-Sanity"));

private Future m_scheduledFuture = null;

private Future m_heartbeat = null;

private Future m_sanity = null;

public UsageManagerImpl() {

}

private void mergeConfigs(Map<String, String> dbParams, Map<String, Object> xmlParams) {

for (Map.Entry<String, Object> param : xmlParams.entrySet()) {

dbParams.put(param.getKey(), (String)param.getValue());

}

}

public boolean configure(String name, Map<String, Object> params) throws ConfigurationException {

final String run = "usage.vmops.pid";

if (s_logger.isDebugEnabled()) {

s_logger.debug("Checking to see if " + run + " exists.");

}

final Class<?> c = UsageServer.class;

m_version = c.getPackage().getImplementationVersion();

if (m_version == null) {

// TODO

// throw new CloudRuntimeException("Unable to find the implementation version of this usage server");

}

if (s_logger.isInfoEnabled()) {

s_logger.info("Implementation Version is " + m_version);

}

Map<String, String> configs = _configDao.getConfiguration(params);

if (params != null) {

mergeConfigs(configs, params);

}

String execTime = configs.get("usage.stats.job.exec.time");

String aggregationRange = configs.get("usage.stats.job.aggregation.range");

String execTimeZone = configs.get("usage.execution.timezone");

String aggreagationTimeZone = configs.get("usage.aggregation.timezone");

String sanityCheckInterval = configs.get("usage.sanity.check.interval");

if(sanityCheckInterval != null){

m_sanityCheckInterval = Integer.parseInt(sanityCheckInterval);

}

if(aggreagationTimeZone != null && !aggreagationTimeZone.isEmpty()){

m_usageTimezone = TimeZone.getTimeZone(aggreagationTimeZone);

}

s_logger.debug("Usage stats aggregation time zone: "+aggreagationTimeZone);

try {

if ((execTime == null) || (aggregationRange == null)) {

s_logger.error("missing configuration values for usage job, usage.stats.job.exec.time = " + execTime + ", usage.stats.job.aggregation.range = " + aggregationRange);

throw new ConfigurationException("Missing configuration values for usage job, usage.stats.job.exec.time = " + execTime + ", usage.stats.job.aggregation.range = " + aggregationRange);

}

String[] execTimeSegments = execTime.split(":");

if (execTimeSegments.length != 2) {

s_logger.error("Unable to parse usage.stats.job.exec.time");

throw new ConfigurationException("Unable to parse usage.stats.job.exec.time '" + execTime + "'");

}

int hourOfDay = Integer.parseInt(execTimeSegments[0]);

int minutes = Integer.parseInt(execTimeSegments[1]);

m_jobExecTime.setTime(new Date());

m_jobExecTime.set(Calendar.HOUR_OF_DAY, hourOfDay);

m_jobExecTime.set(Calendar.MINUTE, minutes);

m_jobExecTime.set(Calendar.SECOND, 0);

m_jobExecTime.set(Calendar.MILLISECOND, 0);

if(execTimeZone != null && !execTimeZone.isEmpty()){

m_jobExecTime.setTimeZone(TimeZone.getTimeZone(execTimeZone));

}

// if the hour to execute the job has already passed, roll the day forward to the next day

Date execDate = m_jobExecTime.getTime();

if (execDate.before(new Date())) {

m_jobExecTime.roll(Calendar.DAY_OF_YEAR, true);

}

s_logger.debug("Execution Time: "+execDate.toString());

Date currentDate = new Date(System.currentTimeMillis());

s_logger.debug("Current Time: "+currentDate.toString());

m_aggregationDuration = Integer.parseInt(aggregationRange);

if (m_aggregationDuration < UsageUtils.USAGE_AGGREGATION_RANGE_MIN) {

s_logger.warn("Usage stats job aggregation range is to small, using the minimum value of " + UsageUtils.USAGE_AGGREGATION_RANGE_MIN);

m_aggregationDuration = UsageUtils.USAGE_AGGREGATION_RANGE_MIN;

}

m_hostname = InetAddress.getLocalHost().getHostName() + "/" + InetAddress.getLocalHost().getHostAddress();

} catch (NumberFormatException ex) {

throw new ConfigurationException("Unable to parse usage.stats.job.exec.time '" + execTime + "' or usage.stats.job.aggregation.range '" + aggregationRange + "', please check configuration values");

} catch (Exception e) {

s_logger.error("Unhandled exception configuring UsageManger", e);

throw new ConfigurationException("Unhandled exception configuring UsageManager " + e.toString());

}

m_pid = Integer.parseInt(System.getProperty("pid"));

return true;

}

public boolean start() {

if (s_logger.isInfoEnabled()) {

s_logger.info("Starting Usage Manager");

}

// use the configured exec time and aggregation duration for scheduling the job

m_scheduledFuture = m_executor.scheduleAtFixedRate(this, m_jobExecTime.getTimeInMillis() - System.currentTimeMillis(), m_aggregationDuration * 60 * 1000, TimeUnit.MILLISECONDS);

m_heartbeat = m_heartbeatExecutor.scheduleAtFixedRate(new Heartbeat(), /* start in 15 seconds...*/15*1000, /* check database every minute*/60*1000, TimeUnit.MILLISECONDS);

if(m_sanityCheckInterval > 0){

m_sanity = m_sanityExecutor.scheduleAtFixedRate(new SanityCheck(), 1, m_sanityCheckInterval, TimeUnit.DAYS);

}

TransactionLegacy usageTxn = TransactionLegacy.open(TransactionLegacy.USAGE_DB);

try {

if(m_heartbeatLock.lock(3)) { // 3 second timeout

try {

UsageJobVO job = m_usageJobDao.getLastJob();

if (job == null) {

m_usageJobDao.createNewJob(m_hostname, m_pid, UsageJobVO.JOB_TYPE_RECURRING);

}

} finally {

m_heartbeatLock.unlock();

}

} else {

if(s_logger.isTraceEnabled())

s_logger.trace("Heartbeat lock is in use by others, returning true as someone else will take over the job if required");

}

} finally {

usageTxn.close();

}

return true;

}

public boolean stop() {

m_heartbeat.cancel(true);

m_scheduledFuture.cancel(true);

if(m_sanity != null){

m_sanity.cancel(true);

}

return true;

}

@Override

public void run() {

(new ManagedContextRunnable() {

@Override

protected void runInContext() {

runInContextInternal();

}

}).run();

}

protected void runInContextInternal() {

if (s_logger.isInfoEnabled()) {

s_logger.info("starting usage job...");

}

// how about we update the job exec time when the job starts???

long execTime = m_jobExecTime.getTimeInMillis();

long now = System.currentTimeMillis() + 2000; // 2 second buffer since jobs can run a little early (though usually just by milliseconds)

if (execTime < now) {

// if exec time is in the past, calculate the next time the job will execute...if this is a one-off job that is a result

// of scheduleParse() then don't update the next exec time...

m_jobExecTime.add(Calendar.MINUTE, m_aggregationDuration);

}

UsageJobVO job = m_usageJobDao.isOwner(m_hostname, m_pid);

if (job != null) {

// FIXME: we really need to do a better job of not missing any events...so we should some how

// keep track of the last time usage was run, then go from there...

// For executing the job, we treat hourly and daily as special time ranges, using the previous full hour or the previous

// full day. Otherwise we just subtract off the aggregation range from the current time and use that as start date with

// current time as end date.

Calendar cal = Calendar.getInstance(m_usageTimezone);

cal.setTime(new Date());

long startDate = 0;

long endDate = 0;

if (m_aggregationDuration == DAILY_TIME) {

cal.roll(Calendar.DAY_OF_YEAR, false);

cal.set(Calendar.HOUR_OF_DAY, 0);

cal.set(Calendar.MINUTE, 0);

cal.set(Calendar.SECOND, 0);

cal.set(Calendar.MILLISECOND, 0);

startDate = cal.getTime().getTime();

cal.roll(Calendar.DAY_OF_YEAR, true);

cal.add(Calendar.MILLISECOND, -1);

endDate = cal.getTime().getTime();

} else if (m_aggregationDuration == HOURLY_TIME) {

cal.roll(Calendar.HOUR_OF_DAY, false);

cal.set(Calendar.MINUTE, 0);

cal.set(Calendar.SECOND, 0);

cal.set(Calendar.MILLISECOND, 0);

startDate = cal.getTime().getTime();

cal.roll(Calendar.HOUR_OF_DAY, true);

cal.add(Calendar.MILLISECOND, -1);

endDate = cal.getTime().getTime();

} else {

endDate = cal.getTime().getTime(); // current time

cal.add(Calendar.MINUTE, -1*m_aggregationDuration);

startDate = cal.getTime().getTime();

}

parse(job, startDate, endDate);

} else {

if (s_logger.isDebugEnabled()) {

s_logger.debug("Not owner of usage job, skipping...");

}

}

if (s_logger.isInfoEnabled()) {

s_logger.info("usage job complete");

}

}

public void scheduleParse() {

if (s_logger.isDebugEnabled()) {

s_logger.debug("Scheduling Usage job...");

}

m_executor.schedule(this, 0, TimeUnit.MILLISECONDS);

}

public void parse(UsageJobVO job, long startDateMillis, long endDateMillis) {

// TODO: Shouldn't we also allow parsing by the type of usage?

boolean success = false;

long timeStart = System.currentTimeMillis();

long deleteOldStatsTimeMillis = 0L;

try {

if ((endDateMillis == 0) || (endDateMillis > timeStart)) {

endDateMillis = timeStart;

}

long lastSuccess = m_usageJobDao.getLastJobSuccessDateMillis();

if (lastSuccess != 0) {

startDateMillis = lastSuccess+1; // 1 millisecond after

}

if (startDateMillis >= endDateMillis) {

if (s_logger.isInfoEnabled()) {

s_logger.info("not parsing usage records since start time mills (" + startDateMillis + ") is on or after end time millis (" + endDateMillis + ")");

}

TransactionLegacy jobUpdateTxn = TransactionLegacy.open(TransactionLegacy.USAGE_DB);

try {

jobUpdateTxn.start();

// everything seemed to work...set endDate as the last success date

m_usageJobDao.updateJobSuccess(job.getId(), startDateMillis, endDateMillis, System.currentTimeMillis() - timeStart, success);

// create a new job if this is a recurring job

if (job.getJobType() == UsageJobVO.JOB_TYPE_RECURRING) {

m_usageJobDao.createNewJob(m_hostname, m_pid, UsageJobVO.JOB_TYPE_RECURRING);

}

jobUpdateTxn.commit();

} finally {

jobUpdateTxn.close();

}

return;

}

deleteOldStatsTimeMillis = startDateMillis;

Date startDate = new Date(startDateMillis);

Date endDate = new Date(endDateMillis);

if (s_logger.isInfoEnabled()) {

s_logger.info("Parsing usage records between " + startDate + " and " + endDate);

}

List<AccountVO> accounts = null;

List<UserStatisticsVO> userStats = null;

Map<String, UsageNetworkVO> networkStats = null;

List<VmDiskStatisticsVO> vmDiskStats = null;

Map<String, UsageVmDiskVO> vmDiskUsages = null;

TransactionLegacy userTxn = TransactionLegacy.open(TransactionLegacy.CLOUD_DB);

try {

Long limit = Long.valueOf(500);

Long offset = Long.valueOf(0);

Long lastAccountId = m_usageDao.getLastAccountId();

if (lastAccountId == null) {

lastAccountId = Long.valueOf(0);

}

do {

Filter filter = new Filter(AccountVO.class, "id", true, offset, limit);

accounts = m_accountDao.findActiveAccounts(lastAccountId, filter);

if ((accounts != null) && !accounts.isEmpty()) {

// now update the accounts in the cloud_usage db

m_usageDao.updateAccounts(accounts);

}

offset = new Long(offset.longValue() + limit.longValue());

} while ((accounts != null) && !accounts.isEmpty());

// reset offset

offset = Long.valueOf(0);

do {

Filter filter = new Filter(AccountVO.class, "id", true, offset, limit);

accounts = m_accountDao.findRecentlyDeletedAccounts(lastAccountId, startDate, filter);

if ((accounts != null) && !accounts.isEmpty()) {

// now update the accounts in the cloud_usage db

m_usageDao.updateAccounts(accounts);

}

offset = new Long(offset.longValue() + limit.longValue());

} while ((accounts != null) && !accounts.isEmpty());

// reset offset

offset = Long.valueOf(0);

do {

Filter filter = new Filter(AccountVO.class, "id", true, offset, limit);

accounts = m_accountDao.findNewAccounts(lastAccountId, filter);

if ((accounts != null) && !accounts.isEmpty()) {

// now copy the accounts to cloud_usage db

m_usageDao.saveAccounts(accounts);

}

offset = new Long(offset.longValue() + limit.longValue());

} while ((accounts != null) && !accounts.isEmpty());

// reset offset

offset = Long.valueOf(0);

// get all the user stats to create usage records for the network usage

Long lastUserStatsId = m_usageDao.getLastUserStatsId();

if (lastUserStatsId == null) {

lastUserStatsId = Long.valueOf(0);

}

SearchCriteria<UserStatisticsVO> sc2 = m_userStatsDao.createSearchCriteria();

sc2.addAnd("id", SearchCriteria.Op.LTEQ, lastUserStatsId);

do {

Filter filter = new Filter(UserStatisticsVO.class, "id", true, offset, limit);

userStats = m_userStatsDao.search(sc2, filter);

if ((userStats != null) && !userStats.isEmpty()) {

// now copy the accounts to cloud_usage db

m_usageDao.updateUserStats(userStats);

}

offset = new Long(offset.longValue() + limit.longValue());

} while ((userStats != null) && !userStats.isEmpty());

// reset offset

offset = Long.valueOf(0);

sc2 = m_userStatsDao.createSearchCriteria();

sc2.addAnd("id", SearchCriteria.Op.GT, lastUserStatsId);

do {

Filter filter = new Filter(UserStatisticsVO.class, "id", true, offset, limit);

userStats = m_userStatsDao.search(sc2, filter);

if ((userStats != null) && !userStats.isEmpty()) {

// now copy the accounts to cloud_usage db

m_usageDao.saveUserStats(userStats);

}

offset = new Long(offset.longValue() + limit.longValue());

} while ((userStats != null) && !userStats.isEmpty());

// reset offset

offset = Long.valueOf(0);

// get all the vm network stats to create usage_vm_network records for the vm network usage

Long lastVmDiskStatsId = m_usageDao.getLastVmDiskStatsId();

if (lastVmDiskStatsId == null) {

lastVmDiskStatsId = Long.valueOf(0);

}

SearchCriteria<VmDiskStatisticsVO> sc4 = m_vmDiskStatsDao.createSearchCriteria();

sc4.addAnd("id", SearchCriteria.Op.LTEQ, lastVmDiskStatsId);

do {

Filter filter = new Filter(VmDiskStatisticsVO.class, "id", true, offset, limit);

vmDiskStats = m_vmDiskStatsDao.search(sc4, filter);

if ((vmDiskStats != null) && !vmDiskStats.isEmpty()) {

// now copy the accounts to cloud_usage db

m_usageDao.updateVmDiskStats(vmDiskStats);

}

offset = new Long(offset.longValue() + limit.longValue());

} while ((vmDiskStats != null) && !vmDiskStats.isEmpty());

// reset offset

offset = Long.valueOf(0);

sc4 = m_vmDiskStatsDao.createSearchCriteria();

sc4.addAnd("id", SearchCriteria.Op.GT, lastVmDiskStatsId);

do {

Filter filter = new Filter(VmDiskStatisticsVO.class, "id", true, offset, limit);

vmDiskStats = m_vmDiskStatsDao.search(sc4, filter);

if ((vmDiskStats != null) && !vmDiskStats.isEmpty()) {

// now copy the accounts to cloud_usage db

m_usageDao.saveVmDiskStats(vmDiskStats);

}

offset = new Long(offset.longValue() + limit.longValue());

} while ((vmDiskStats != null) && !vmDiskStats.isEmpty());

} finally {

userTxn.close();

}

// TODO: Fetch a maximum number of events and process them before moving on to the next range of events

// - get a list of the latest events

// - insert the latest events into the usage.events table

List<UsageEventVO> events = _usageEventDao.getRecentEvents(new Date(endDateMillis));

TransactionLegacy usageTxn = TransactionLegacy.open(TransactionLegacy.USAGE_DB);

try {

usageTxn.start();

// make sure start date is before all of our un-processed events (the events are ordered oldest

// to newest, so just test against the first event)

if ((events != null) && (events.size() > 0)) {

Date oldestEventDate = events.get(0).getCreateDate();

if (oldestEventDate.getTime() < startDateMillis) {

startDateMillis = oldestEventDate.getTime();

startDate = new Date(startDateMillis);

}

// - loop over the list of events and create entries in the helper tables

// - create the usage records using the parse methods below

for (UsageEventVO event : events) {

event.setProcessed(true);

_usageEventDao.update(event.getId(), event);

createHelperRecord(event);

}

}

// TODO: Fetch a maximum number of user stats and process them before moving on to the next range of user stats

// get user stats in order to compute network usage

networkStats = m_usageNetworkDao.getRecentNetworkStats();

Calendar recentlyDeletedCal = Calendar.getInstance(m_usageTimezone);

recentlyDeletedCal.setTimeInMillis(startDateMillis);

recentlyDeletedCal.add(Calendar.MINUTE, -1*THREE_DAYS_IN_MINUTES);

Date recentlyDeletedDate = recentlyDeletedCal.getTime();

// Keep track of user stats for an account, across all of its public IPs

Map<String, UserStatisticsVO> aggregatedStats = new HashMap<String, UserStatisticsVO>();

int startIndex = 0;

do {

userStats = m_userStatsDao.listActiveAndRecentlyDeleted(recentlyDeletedDate, startIndex, 500);

if (userStats != null) {

for (UserStatisticsVO userStat : userStats) {

if(userStat.getDeviceId() != null){

String hostKey = userStat.getDataCenterId() + "-" + userStat.getAccountId()+"-Host-" + userStat.getDeviceId();

UserStatisticsVO hostAggregatedStat = aggregatedStats.get(hostKey);

if (hostAggregatedStat == null) {

hostAggregatedStat = new UserStatisticsVO(userStat.getAccountId(), userStat.getDataCenterId(), userStat.getPublicIpAddress(),

userStat.getDeviceId(), userStat.getDeviceType(), userStat.getNetworkId());

}

hostAggregatedStat.setAggBytesSent(hostAggregatedStat.getAggBytesSent() + userStat.getAggBytesSent());

hostAggregatedStat.setAggBytesReceived(hostAggregatedStat.getAggBytesReceived() + userStat.getAggBytesReceived());

aggregatedStats.put(hostKey, hostAggregatedStat);

}

}

}

startIndex += 500;

} while ((userStats != null) && !userStats.isEmpty());

// loop over the user stats, create delta entries in the usage_network helper table

int numAcctsProcessed = 0;

usageNetworks.clear();

for (String key : aggregatedStats.keySet()) {

UsageNetworkVO currentNetworkStats = null;

if (networkStats != null) {

currentNetworkStats = networkStats.get(key);

}

createNetworkHelperEntry(aggregatedStats.get(key), currentNetworkStats, endDateMillis);

numAcctsProcessed++;

}

m_usageNetworkDao.saveUsageNetworks(usageNetworks);

if (s_logger.isDebugEnabled()) {

s_logger.debug("created network stats helper entries for " + numAcctsProcessed + " accts");

}

// get vm disk stats in order to compute vm disk usage

vmDiskUsages = m_usageVmDiskDao.getRecentVmDiskStats();

// Keep track of user stats for an account, across all of its public IPs

Map<String, VmDiskStatisticsVO> aggregatedDiskStats = new HashMap<String, VmDiskStatisticsVO>();

startIndex = 0;

do {

vmDiskStats = m_vmDiskStatsDao.listActiveAndRecentlyDeleted(recentlyDeletedDate, startIndex, 500);

if (vmDiskUsages != null) {

for (VmDiskStatisticsVO vmDiskStat : vmDiskStats) {

if(vmDiskStat.getVmId() != null){

String hostKey = vmDiskStat.getDataCenterId() + "-" + vmDiskStat.getAccountId()+"-Vm-" + vmDiskStat.getVmId()+"-Disk-" + vmDiskStat.getVolumeId();

VmDiskStatisticsVO hostAggregatedStat = aggregatedDiskStats.get(hostKey);

if (hostAggregatedStat == null) {

hostAggregatedStat = new VmDiskStatisticsVO(vmDiskStat.getAccountId(), vmDiskStat.getDataCenterId(), vmDiskStat.getVmId(),vmDiskStat.getVolumeId());

}

hostAggregatedStat.setAggIORead(hostAggregatedStat.getAggIORead() + vmDiskStat.getAggIORead());

hostAggregatedStat.setAggIOWrite(hostAggregatedStat.getAggIOWrite() + vmDiskStat.getAggIOWrite());

hostAggregatedStat.setAggBytesRead(hostAggregatedStat.getAggBytesRead() + vmDiskStat.getAggBytesRead());

hostAggregatedStat.setAggBytesWrite(hostAggregatedStat.getAggBytesWrite() + vmDiskStat.getAggBytesWrite());

aggregatedDiskStats.put(hostKey, hostAggregatedStat);

}

}

}

startIndex += 500;

} while ((userStats != null) && !userStats.isEmpty());

// loop over the user stats, create delta entries in the usage_disk helper table

numAcctsProcessed = 0;

usageVmDisks.clear();

for (String key : aggregatedDiskStats.keySet()) {

UsageVmDiskVO currentVmDiskStats = null;

if (vmDiskStats != null) {

currentVmDiskStats = vmDiskUsages.get(key);

}

createVmDiskHelperEntry(aggregatedDiskStats.get(key), currentVmDiskStats, endDateMillis);

numAcctsProcessed++;

}

m_usageVmDiskDao.saveUsageVmDisks(usageVmDisks);

if (s_logger.isDebugEnabled()) {

s_logger.debug("created vm disk stats helper entries for " + numAcctsProcessed + " accts");

}

// commit the helper records, then start a new transaction

usageTxn.commit();

usageTxn.start();

boolean parsed = false;

numAcctsProcessed = 0;

Date currentStartDate = startDate;

Date currentEndDate = endDate;

Date tempDate = endDate;

Calendar aggregateCal = Calendar.getInstance(m_usageTimezone);

while ((tempDate.after(startDate)) && ((tempDate.getTime() - startDate.getTime()) > 60000)){

currentEndDate = tempDate;

aggregateCal.setTime(tempDate);

aggregateCal.add(Calendar.MINUTE, -m_aggregationDuration);

tempDate = aggregateCal.getTime();

}

while (!currentEndDate.after(endDate) || (currentEndDate.getTime() -endDate.getTime() < 60000)){

Long offset = Long.valueOf(0);

Long limit = Long.valueOf(500);

do {

Filter filter = new Filter(AccountVO.class, "id", true, offset, limit);

accounts = m_accountDao.listAll(filter);

if ((accounts != null) && !accounts.isEmpty()) {

for (AccountVO account : accounts) {

parsed = parseHelperTables(account, currentStartDate, currentEndDate);

numAcctsProcessed++;

}

}

offset = new Long(offset.longValue() + limit.longValue());

} while ((accounts != null) && !accounts.isEmpty());

if (s_logger.isDebugEnabled()) {

s_logger.debug("processed VM/Network Usage for " + numAcctsProcessed + " ACTIVE accts");

}

numAcctsProcessed = 0;

// reset offset

offset = Long.valueOf(0);

do {

Filter filter = new Filter(AccountVO.class, "id", true, offset, limit);

accounts = m_accountDao.findRecentlyDeletedAccounts(null, recentlyDeletedDate, filter);

if ((accounts != null) && !accounts.isEmpty()) {

for (AccountVO account : accounts) {

parsed = parseHelperTables(account, currentStartDate, currentEndDate);

List<Long> publicTemplates = m_usageDao.listPublicTemplatesByAccount(account.getId());

for(Long templateId : publicTemplates){

//mark public templates owned by deleted accounts as deleted

List<UsageStorageVO> storageVOs = m_usageStorageDao.listById(account.getId(), templateId, StorageTypes.TEMPLATE);

if (storageVOs.size() > 1) {

s_logger.warn("More that one usage entry for storage: " + templateId + " assigned to account: " + account.getId() + "; marking them all as deleted...");

}

for (UsageStorageVO storageVO : storageVOs) {

if (s_logger.isDebugEnabled()) {

s_logger.debug("deleting template: " + storageVO.getId() + " from account: " + storageVO.getAccountId());

}

storageVO.setDeleted(account.getRemoved());

m_usageStorageDao.update(storageVO);

}

}

numAcctsProcessed++;

}

}

offset = new Long(offset.longValue() + limit.longValue());

} while ((accounts != null) && !accounts.isEmpty());

currentStartDate = new Date(currentEndDate.getTime() + 1);

aggregateCal.setTime(currentEndDate);

aggregateCal.add(Calendar.MINUTE, m_aggregationDuration);

currentEndDate = aggregateCal.getTime();

}

if (s_logger.isDebugEnabled()) {

s_logger.debug("processed Usage for " + numAcctsProcessed + " RECENTLY DELETED accts");

}

// FIXME: we don't break the above loop if something fails to parse, so it gets reset every account,

// do we want to break out of processing accounts and rollback if there are errors?

if (!parsed) {

usageTxn.rollback();

} else {

success = true;

}

} catch (Exception ex) {

s_logger.error("Exception in usage manager", ex);

usageTxn.rollback();

} finally {

// everything seemed to work...set endDate as the last success date

m_usageJobDao.updateJobSuccess(job.getId(), startDateMillis, endDateMillis, System.currentTimeMillis() - timeStart, success);

// create a new job if this is a recurring job

if (job.getJobType() == UsageJobVO.JOB_TYPE_RECURRING) {

m_usageJobDao.createNewJob(m_hostname, m_pid, UsageJobVO.JOB_TYPE_RECURRING);

}

usageTxn.commit();

usageTxn.close();

// switch back to CLOUD_DB

TransactionLegacy swap = TransactionLegacy.open(TransactionLegacy.CLOUD_DB);

if(!success){

_alertMgr.sendAlert(AlertManager.AlertType.ALERT_TYPE_USAGE_SERVER_RESULT, 0, new Long(0), "Usage job failed. Job id: "+job.getId(), "Usage job failed. Job id: "+job.getId());

} else {

_alertMgr.clearAlert(AlertManager.AlertType.ALERT_TYPE_USAGE_SERVER_RESULT, 0, 0);

}

swap.close();

}

} catch (Exception e) {

s_logger.error("Usage Manager error", e);

}

}

private boolean parseHelperTables(AccountVO account, Date currentStartDate, Date currentEndDate){

boolean parsed = false;

parsed = VMInstanceUsageParser.parse(account, currentStartDate, currentEndDate);

if (s_logger.isDebugEnabled()) {

if (!parsed) {

s_logger.debug("vm usage instances successfully parsed? " + parsed + " (for account: " + account.getAccountName() + ", id: " + account.getId() + ")");

}

}

parsed = NetworkUsageParser.parse(account, currentStartDate, currentEndDate);

if (s_logger.isDebugEnabled()) {

if (!parsed) {

s_logger.debug("network usage successfully parsed? " + parsed + " (for account: " + account.getAccountName() + ", id: " + account.getId() + ")");

}

}

parsed = VmDiskUsageParser.parse(account, currentStartDate, currentEndDate);

if (s_logger.isDebugEnabled()) {

if (!parsed) {

s_logger.debug("vm disk usage successfully parsed? " + parsed + " (for account: " + account.getAccountName() + ", id: " + account.getId() + ")");

}

}

parsed = VolumeUsageParser.parse(account, currentStartDate, currentEndDate);

if (s_logger.isDebugEnabled()) {

if (!parsed) {

s_logger.debug("volume usage successfully parsed? " + parsed + " (for account: " + account.getAccountName() + ", id: " + account.getId() + ")");

}

}

parsed = StorageUsageParser.parse(account, currentStartDate, currentEndDate);

if (s_logger.isDebugEnabled()) {

if (!parsed) {

s_logger.debug("storage usage successfully parsed? " + parsed + " (for account: " + account.getAccountName() + ", id: " + account.getId() + ")");

}

}

parsed = SecurityGroupUsageParser.parse(account, currentStartDate, currentEndDate);

if (s_logger.isDebugEnabled()) {

if (!parsed) {

s_logger.debug("Security Group usage successfully parsed? " + parsed + " (for account: " + account.getAccountName() + ", id: " + account.getId() + ")");

}

}

parsed = LoadBalancerUsageParser.parse(account, currentStartDate, currentEndDate);

if (s_logger.isDebugEnabled()) {

if (!parsed) {

s_logger.debug("load balancer usage successfully parsed? " + parsed + " (for account: " + account.getAccountName() + ", id: " + account.getId() + ")");

}

}

parsed = PortForwardingUsageParser.parse(account, currentStartDate, currentEndDate);

if (s_logger.isDebugEnabled()) {

if (!parsed) {

s_logger.debug("port forwarding usage successfully parsed? " + parsed + " (for account: " + account.getAccountName() + ", id: " + account.getId() + ")");

}

}

parsed = NetworkOfferingUsageParser.parse(account, currentStartDate, currentEndDate);

if (s_logger.isDebugEnabled()) {

if (!parsed) {

s_logger.debug("network offering usage successfully parsed? " + parsed + " (for account: " + account.getAccountName() + ", id: " + account.getId() + ")");

}

}

parsed = IPAddressUsageParser.parse(account, currentStartDate, currentEndDate);

if (s_logger.isDebugEnabled()) {

if (!parsed) {

s_logger.debug("IPAddress usage successfully parsed? " + parsed + " (for account: " + account.getAccountName() + ", id: " + account.getId() + ")");

}

}

parsed = VPNUserUsageParser.parse(account, currentStartDate, currentEndDate);

if (s_logger.isDebugEnabled()) {

if (!parsed) {

s_logger.debug("VPN user usage successfully parsed? " + parsed + " (for account: " + account.getAccountName() + ", id: " + account.getId() + ")");

}

}

parsed = VMSnapshotUsageParser.parse(account, currentStartDate, currentEndDate);

if (s_logger.isDebugEnabled()) {

if (!parsed) {

s_logger.debug("VM Snapshot usage successfully parsed? " + parsed + " (for account: " + account.getAccountName() + ", id: " + account.getId() + ")");

}

}

return parsed;

}

private void createHelperRecord(UsageEventVO event) {

String eventType = event.getType();

if (isVMEvent(eventType)) {

createVMHelperEvent(event);

} else if (isIPEvent(eventType)) {

createIPHelperEvent(event);

} else if (isVolumeEvent(eventType)) {

createVolumeHelperEvent(event);

} else if (isTemplateEvent(eventType)) {

createTemplateHelperEvent(event);

} else if (isISOEvent(eventType)) {

createISOHelperEvent(event);

} else if (isSnapshotEvent(eventType)) {

createSnapshotHelperEvent(event);

} else if (isLoadBalancerEvent(eventType)) {

createLoadBalancerHelperEvent(event);

} else if (isPortForwardingEvent(eventType)) {

createPortForwardingHelperEvent(event);

} else if (isNetworkOfferingEvent(eventType)) {

createNetworkOfferingEvent(event);

} else if (isVPNUserEvent(eventType)) {

createVPNUserEvent(event);

} else if (isSecurityGroupEvent(eventType)) {

createSecurityGroupEvent(event);

} else if (isVmSnapshotEvent(eventType)){

createVMSnapshotEvent(event);

}

}

private boolean isVMEvent(String eventType) {

if (eventType == null) return false;

return eventType.startsWith("VM.");

}

private boolean isIPEvent(String eventType) {

if (eventType == null) return false;

return eventType.startsWith("NET.IP");

}

private boolean isVolumeEvent(String eventType) {

return eventType != null &&

(eventType.equals(EventTypes.EVENT_VOLUME_CREATE) || eventType.equals(EventTypes.EVENT_VOLUME_DELETE) || eventType.equals(EventTypes.EVENT_VOLUME_RESIZE));

}

private boolean isTemplateEvent(String eventType) {

if (eventType == null) return false;

return (eventType.equals(EventTypes.EVENT_TEMPLATE_CREATE) ||

eventType.equals(EventTypes.EVENT_TEMPLATE_COPY) ||

eventType.equals(EventTypes.EVENT_TEMPLATE_DELETE));

}

private boolean isISOEvent(String eventType) {

if (eventType == null) return false;

return (eventType.equals(EventTypes.EVENT_ISO_CREATE) ||

eventType.equals(EventTypes.EVENT_ISO_COPY) ||

eventType.equals(EventTypes.EVENT_ISO_DELETE));

}

private boolean isSnapshotEvent(String eventType) {

if (eventType == null) return false;

return (eventType.equals(EventTypes.EVENT_SNAPSHOT_CREATE) ||

eventType.equals(EventTypes.EVENT_SNAPSHOT_DELETE));

}

private boolean isLoadBalancerEvent(String eventType) {

if (eventType == null) return false;

return eventType.startsWith("LB.");

}

private boolean isPortForwardingEvent(String eventType) {

if (eventType == null) return false;

return eventType.startsWith("NET.RULE");

}

private boolean isNetworkOfferingEvent(String eventType) {

if (eventType == null) return false;

return (eventType.equals(EventTypes.EVENT_NETWORK_OFFERING_CREATE) ||

eventType.equals(EventTypes.EVENT_NETWORK_OFFERING_DELETE) ||

eventType.equals(EventTypes.EVENT_NETWORK_OFFERING_ASSIGN) ||

eventType.equals(EventTypes.EVENT_NETWORK_OFFERING_REMOVE));

}

private boolean isVPNUserEvent(String eventType) {

if (eventType == null) return false;

return eventType.startsWith("VPN.USER");

}

private boolean isSecurityGroupEvent(String eventType) {

if (eventType == null) return false;

return (eventType.equals(EventTypes.EVENT_SECURITY_GROUP_ASSIGN) ||

eventType.equals(EventTypes.EVENT_SECURITY_GROUP_REMOVE));

}

private boolean isVmSnapshotEvent(String eventType){

if (eventType == null) return false;

return (eventType.equals(EventTypes.EVENT_VM_SNAPSHOT_CREATE) ||

eventType.equals(EventTypes.EVENT_VM_SNAPSHOT_DELETE));

}

private void createVMHelperEvent(UsageEventVO event) {

// One record for handling VM.START and VM.STOP

// One record for handling VM.CREATE and VM.DESTROY

// VM events have the parameter "id=<virtualMachineId>"

long vmId = event.getResourceId();

Long soId = event.getOfferingId();; // service offering id

long zoneId = event.getZoneId();

String vmName = event.getResourceName();

if (EventTypes.EVENT_VM_START.equals(event.getType())) {

// create a new usage_vm_instance row for this VM

try {

SearchCriteria<UsageVMInstanceVO> sc = m_usageInstanceDao.createSearchCriteria();

sc.addAnd("vmInstanceId", SearchCriteria.Op.EQ, Long.valueOf(vmId));