Google Git Google Git
Sign in
apache/cloudstack/refs/heads/ghi14172-lodBalancedDb/./usage/src/main/java/com/cloud/usage/UsageManagerImpl.java
blob: eab371ab353fd6dc575ee7eacda29ecd5a637bba [file]
// 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 static com.cloud.utils.NumbersUtil.toHumanReadableSize;
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 java.util.concurrent.atomic.AtomicBoolean;
import com.cloud.network.Network;
import com.cloud.usage.dao.UsageNetworksDao;
import com.cloud.usage.parser.UsageParser;
import com.cloud.network.vpc.Vpc;
import com.cloud.usage.dao.UsageVpcDao;
import javax.inject.Inject;
import javax.naming.ConfigurationException;
import javax.persistence.EntityExistsException;
import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
import org.apache.cloudstack.managed.context.ManagedContextRunnable;
import org.apache.cloudstack.quota.QuotaAlertManager;
import org.apache.cloudstack.quota.QuotaManager;
import org.apache.cloudstack.quota.QuotaStatement;
import org.apache.cloudstack.usage.UsageTypes;
import org.apache.cloudstack.utils.usage.UsageUtils;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.stereotype.Component;
import com.cloud.alert.AlertManager;
import com.cloud.event.EventTypes;
import com.cloud.event.UsageEventDetailsVO;
import com.cloud.event.UsageEventVO;
import com.cloud.event.dao.UsageEventDao;
import com.cloud.event.dao.UsageEventDetailsDao;
import com.cloud.usage.dao.BucketStatisticsDao;
import com.cloud.usage.dao.UsageBackupDao;
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.UsageVMSnapshotOnPrimaryDao;
import com.cloud.usage.dao.UsageVPNUserDao;
import com.cloud.usage.dao.UsageVmDiskDao;
import com.cloud.usage.dao.UsageVolumeDao;
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.DateUtil;
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.QueryBuilder;
import com.cloud.utils.db.SearchCriteria;
import com.cloud.utils.db.TransactionLegacy;
import com.cloud.utils.exception.CloudRuntimeException;
@Component
public class UsageManagerImpl extends ManagerBase implements UsageManager, Runnable {
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 _accountDao;
@Inject
private UserStatisticsDao _userStatsDao;
@Inject
private UsageDao _usageDao;
@Inject
private UsageVMInstanceDao _usageInstanceDao;
@Inject
private UsageIPAddressDao _usageIPAddressDao;
@Inject
private UsageNetworkDao _usageNetworkDao;
@Inject
private UsageVolumeDao _usageVolumeDao;
@Inject
private UsageStorageDao _usageStorageDao;
@Inject
private UsageLoadBalancerPolicyDao _usageLoadBalancerPolicyDao;
@Inject
private UsagePortForwardingRuleDao _usagePortForwardingRuleDao;
@Inject
private UsageNetworkOfferingDao _usageNetworkOfferingDao;
@Inject
private UsageVPNUserDao _usageVPNUserDao;
@Inject
private UsageSecurityGroupDao _usageSecurityGroupDao;
@Inject
private UsageJobDao _usageJobDao;
@Inject
private VmDiskStatisticsDao _vmDiskStatsDao;
@Inject
private UsageVmDiskDao _usageVmDiskDao;
@Inject
protected AlertManager _alertMgr;
@Inject
protected UsageEventDao _usageEventDao;
@Inject
protected UsageEventDetailsDao _usageEventDetailsDao;
@Inject
ConfigurationDao _configDao;
@Inject
private UsageVMSnapshotDao _usageVMSnapshotDao;
@Inject
private UsageVMSnapshotOnPrimaryDao _usageSnapshotOnPrimaryDao;
@Inject
private UsageBackupDao usageBackupDao;
@Inject
private QuotaManager _quotaManager;
@Inject
private QuotaAlertManager _alertManager;
@Inject
private QuotaStatement _quotaStatement;
@Inject
private UsageNetworksDao usageNetworksDao;
@Inject
private BucketStatisticsDao _bucketStatisticsDao;
@Inject
private UsageVpcDao usageVpcDao;
@Inject
private List<UsageParser> usageParsers;
private String _version = null;
private final Calendar _jobExecTime = Calendar.getInstance();
private int _aggregationDuration = 0;
private int _sanityCheckInterval = 0;
private boolean _runQuota=false;
String _hostname = null;
int _pid = 0;
private final GlobalLock _heartbeatLock = GlobalLock.getInternLock("usage.job.heartbeat.check");
private final List<UsageNetworkVO> usageNetworks = new ArrayList<UsageNetworkVO>();
private final List<UsageVmDiskVO> usageVmDisks = new ArrayList<UsageVmDiskVO>();
private final ScheduledExecutorService _executor = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("Usage-Job"));
private final AtomicBoolean isParsingJobRunning = new AtomicBoolean(false);
private final ScheduledExecutorService _heartbeatExecutor = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("Usage-HB"));
private final ScheduledExecutorService _sanityExecutor = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("Usage-Sanity"));
private Future _scheduledFuture = null;
private Future _heartbeat = null;
private Future _sanity = null;
private boolean usageSnapshotSelection = false;
private static TimeZone usageAggregationTimeZone = TimeZone.getTimeZone("GMT");
private static TimeZone usageExecutionTimeZone = TimeZone.getTimeZone("GMT");
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());
}
}
@Override
public boolean configure(String name, Map<String, Object> params) throws ConfigurationException {
final String run = "usage.vmops.pid";
if (logger.isDebugEnabled()) {
logger.debug("Checking to see if " + run + " exists.");
}
final Class<?> c = UsageServer.class;
_version = c.getPackage().getImplementationVersion();
if (_version == null) _version="unknown";
if (logger.isInfoEnabled()) {
logger.info("Implementation Version is " + _version);
}
Map<String, String> configs;
try {
configs = _configDao.getConfiguration(params);
if (params != null) {
mergeConfigs(configs, params);
logger.info("configs = " + configs);
}
} catch (CloudRuntimeException e) {
logger.error("Unhandled configuration exception: " + e.getMessage());
throw new CloudRuntimeException("Unhandled configuration exception", e);
}
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 aggregationTimeZone = configs.get("usage.aggregation.timezone");
String sanityCheckInterval = configs.get("usage.sanity.check.interval");
String quotaEnable = configs.get("quota.enable.service");
_runQuota = Boolean.valueOf(quotaEnable == null ? "false" : quotaEnable );
usageSnapshotSelection = Boolean.valueOf(configs.get("usage.snapshot.virtualsize.select"));
if (sanityCheckInterval != null) {
_sanityCheckInterval = Integer.parseInt(sanityCheckInterval);
}
if (aggregationTimeZone != null && !aggregationTimeZone.isEmpty()) {
usageAggregationTimeZone = TimeZone.getTimeZone(aggregationTimeZone);
}
if (execTimeZone != null) {
usageExecutionTimeZone = TimeZone.getTimeZone(execTimeZone);
}
try {
if ((execTime == null) || (aggregationRange == null)) {
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);
}
Date nextJobExecTime = UsageUtils.getNextJobExecutionTime(usageExecutionTimeZone, execTime);
if (nextJobExecTime == null) {
throw new ConfigurationException(String.format("Unable to parse configuration 'usage.stats.job.exec.time' value [%s].", execTime));
}
_jobExecTime.setTimeZone(usageExecutionTimeZone);
_jobExecTime.setTime(nextJobExecTime);
logger.info("Usage is configured to execute in time zone [{}], at [{}], each [{}] minutes; the current time in that timezone is [{}] and the " +
"next job is scheduled to execute at [{}]. During its execution, Usage will aggregate stats according to the time zone [{}] defined in global setting [usage.aggregation.timezone].",
usageExecutionTimeZone.getID(), execTime, aggregationRange, DateUtil.displayDateInTimezone(usageExecutionTimeZone, new Date()),
DateUtil.displayDateInTimezone(usageExecutionTimeZone, _jobExecTime.getTime()), usageAggregationTimeZone.getID());
_aggregationDuration = Integer.parseInt(aggregationRange);
if (_aggregationDuration < UsageUtils.USAGE_AGGREGATION_RANGE_MIN) {
logger.warn("Usage stats job aggregation range is to small, using the minimum value of " + UsageUtils.USAGE_AGGREGATION_RANGE_MIN);
_aggregationDuration = UsageUtils.USAGE_AGGREGATION_RANGE_MIN;
}
_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) {
logger.error("Unhandled exception configuring UsageManger", e);
throw new ConfigurationException("Unhandled exception configuring UsageManager " + e.toString());
}
try {
_pid = (int) ProcessHandle.current().pid();
} catch (Exception e) {
String msg = String.format("Unable to get process Id for %s!", e.toString());
logger.debug(msg);
throw new ConfigurationException(msg);
}
return true;
}
public static TimeZone getUsageAggregationTimeZone() {
return usageAggregationTimeZone;
}
@Override
public boolean start() {
if (logger.isInfoEnabled()) {
logger.info("Starting Usage Manager");
}
_usageJobDao.removeLastOpenJobsOwned(_hostname, 0);
Runtime.getRuntime().addShutdownHook(new AbandonJob());
// use the configured exec time and aggregation duration for scheduling the job
_scheduledFuture =
_executor.scheduleAtFixedRate(this, _jobExecTime.getTimeInMillis() - System.currentTimeMillis(), _aggregationDuration * 60 * 1000, TimeUnit.MILLISECONDS);
_heartbeat =
_heartbeatExecutor.scheduleAtFixedRate(new Heartbeat(), /* start in 15 seconds...*/15 * 1000, /* check database every minute*/60 * 1000,
TimeUnit.MILLISECONDS);
if (_sanityCheckInterval > 0) {
_sanity = _sanityExecutor.scheduleAtFixedRate(new SanityCheck(), 1, _sanityCheckInterval, TimeUnit.DAYS);
}
TransactionLegacy usageTxn = TransactionLegacy.open(TransactionLegacy.USAGE_DB);
try {
if (_heartbeatLock.lock(3)) { // 3 second timeout
try {
UsageJobVO job = _usageJobDao.getLastJob();
if (job == null) {
_usageJobDao.createNewJob(_hostname, _pid, UsageJobVO.JOB_TYPE_RECURRING);
}
} finally {
_heartbeatLock.unlock();
}
} else {
if (logger.isTraceEnabled())
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;
}
@Override
public boolean stop() {
_heartbeat.cancel(true);
_scheduledFuture.cancel(true);
if (_sanity != null) {
_sanity.cancel(true);
}
return true;
}
@Override
public void run() {
(new ManagedContextRunnable() {
@Override
protected void runInContext() {
isParsingJobRunning.set(true);
try {
runInContextInternal();
} finally {
isParsingJobRunning.set(false);
}
}
}).run();
}
protected void runInContextInternal() {
if (logger.isInfoEnabled()) {
logger.info("starting usage job...");
}
// how about we update the job exec time when the job starts???
long execTime = _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...
_jobExecTime.add(Calendar.MINUTE, _aggregationDuration);
}
UsageJobVO job = _usageJobDao.isOwner(_hostname, _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(usageAggregationTimeZone);
cal.setTime(new Date());
long startDate = 0;
long endDate = 0;
if (_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 (_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 {
cal.add(Calendar.MINUTE, -1 * _aggregationDuration);
cal.set(Calendar.SECOND, 0);
cal.set(Calendar.MILLISECOND, 0);
startDate = cal.getTime().getTime();
cal.add(Calendar.MINUTE, _aggregationDuration);
cal.add(Calendar.MILLISECOND, -1);
endDate = cal.getTime().getTime();
}
parse(job, startDate, endDate);
if (_runQuota){
try {
_quotaManager.calculateQuotaUsage();
}
catch (Exception e){
logger.error("Exception received while calculating quota", e);
}
try {
_quotaStatement.sendStatement();
} catch (Exception e) {
logger.error("Exception received while sending statements", e);
}
try {
_alertManager.checkAndSendQuotaAlertEmails();
} catch (Exception e) {
logger.error("Exception received while sending alerts", e);
}
}
} else {
if (logger.isDebugEnabled()) {
logger.debug("Not owner of usage job, skipping...");
}
}
if (logger.isInfoEnabled()) {
logger.info("usage job complete");
}
}
@Override
public void scheduleParse() {
if (logger.isDebugEnabled()) {
logger.debug("Scheduling Usage job...");
}
_executor.schedule(this, 0, TimeUnit.MILLISECONDS);
}
@Override
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();
try {
if ((endDateMillis == 0) || (endDateMillis > timeStart)) {
endDateMillis = timeStart;
}
long lastSuccess = _usageJobDao.getLastJobSuccessDateMillis();
if (lastSuccess != 0) {
startDateMillis = lastSuccess + 1; // 1 millisecond after
}
if (startDateMillis >= endDateMillis) {
if (logger.isInfoEnabled()) {
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
_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) {
_usageJobDao.createNewJob(_hostname, _pid, UsageJobVO.JOB_TYPE_RECURRING);
}
jobUpdateTxn.commit();
} finally {
jobUpdateTxn.close();
}
return;
}
Date startDate = new Date(startDateMillis);
Date endDate = new Date(endDateMillis);
logger.info("Parsing usage records between [{}] and [{}].", DateUtil.displayDateInTimezone(usageAggregationTimeZone, startDate),
DateUtil.displayDateInTimezone(usageAggregationTimeZone, endDate));
List<AccountVO> accounts = null;
List<UserStatisticsVO> userStats = null;
Map<String, UsageNetworkVO> networkStats = null;
List<VmDiskStatisticsVO> vmDiskStats = null;
Map<String, UsageVmDiskVO> vmDiskUsages = null;
List<BucketStatisticsVO> bucketStats = null;
TransactionLegacy userTxn = TransactionLegacy.open(TransactionLegacy.CLOUD_DB);
try {
Long limit = Long.valueOf(500);
Long offset = Long.valueOf(0);
Long lastAccountId = _usageDao.getLastAccountId();
if (lastAccountId == null) {
lastAccountId = Long.valueOf(0);
}
do {
Filter filter = new Filter(AccountVO.class, "id", true, offset, limit);
accounts = _accountDao.findActiveAccounts(lastAccountId, filter);
if ((accounts != null) && !accounts.isEmpty()) {
// now update the accounts in the cloud_usage db
_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 = _accountDao.findRecentlyDeletedAccounts(lastAccountId, startDate, filter);
if ((accounts != null) && !accounts.isEmpty()) {
// now update the accounts in the cloud_usage db
_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 = _accountDao.findNewAccounts(lastAccountId, filter);
if ((accounts != null) && !accounts.isEmpty()) {
// now copy the accounts to cloud_usage db
_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 = _usageDao.getLastUserStatsId();
if (lastUserStatsId == null) {
lastUserStatsId = Long.valueOf(0);
}
SearchCriteria<UserStatisticsVO> sc2 = _userStatsDao.createSearchCriteria();
sc2.addAnd("id", SearchCriteria.Op.LTEQ, lastUserStatsId);
do {
Filter filter = new Filter(UserStatisticsVO.class, "id", true, offset, limit);
userStats = _userStatsDao.search(sc2, filter);
if ((userStats != null) && !userStats.isEmpty()) {
// now copy the accounts to cloud_usage db
_usageDao.updateUserStats(userStats);
}
offset = new Long(offset.longValue() + limit.longValue());
} while ((userStats != null) && !userStats.isEmpty());
// reset offset
offset = Long.valueOf(0);
sc2 = _userStatsDao.createSearchCriteria();
sc2.addAnd("id", SearchCriteria.Op.GT, lastUserStatsId);
do {
Filter filter = new Filter(UserStatisticsVO.class, "id", true, offset, limit);
userStats = _userStatsDao.search(sc2, filter);
if ((userStats != null) && !userStats.isEmpty()) {
// now copy the accounts to cloud_usage db
_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 = _usageDao.getLastVmDiskStatsId();
if (lastVmDiskStatsId == null) {
lastVmDiskStatsId = Long.valueOf(0);
}
SearchCriteria<VmDiskStatisticsVO> sc4 = _vmDiskStatsDao.createSearchCriteria();
sc4.addAnd("id", SearchCriteria.Op.LTEQ, lastVmDiskStatsId);
do {
Filter filter = new Filter(VmDiskStatisticsVO.class, "id", true, offset, limit);
vmDiskStats = _vmDiskStatsDao.search(sc4, filter);
if ((vmDiskStats != null) && !vmDiskStats.isEmpty()) {
// now copy the accounts to cloud_usage db
_usageDao.updateVmDiskStats(vmDiskStats);
}
offset = new Long(offset.longValue() + limit.longValue());
} while ((vmDiskStats != null) && !vmDiskStats.isEmpty());
// reset offset
offset = Long.valueOf(0);
sc4 = _vmDiskStatsDao.createSearchCriteria();
sc4.addAnd("id", SearchCriteria.Op.GT, lastVmDiskStatsId);
do {
Filter filter = new Filter(VmDiskStatisticsVO.class, "id", true, offset, limit);
vmDiskStats = _vmDiskStatsDao.search(sc4, filter);
if ((vmDiskStats != null) && !vmDiskStats.isEmpty()) {
// now copy the accounts to cloud_usage db
_usageDao.saveVmDiskStats(vmDiskStats);
}
offset = new Long(offset.longValue() + limit.longValue());
} while ((vmDiskStats != null) && !vmDiskStats.isEmpty());
// reset offset
offset = Long.valueOf(0);
// get all the user stats to create usage records for the bucket usage
Long lastBucketStatsId = _usageDao.getLastBucketStatsId();
if (lastBucketStatsId == null) {
lastBucketStatsId = Long.valueOf(0);
}
SearchCriteria<BucketStatisticsVO> sc5 = _bucketStatisticsDao.createSearchCriteria();
sc5.addAnd("id", SearchCriteria.Op.LTEQ, lastBucketStatsId);
do {
Filter filter = new Filter(BucketStatisticsVO.class, "id", true, offset, limit);
bucketStats = _bucketStatisticsDao.search(sc5, filter);
if ((bucketStats != null) && !bucketStats.isEmpty()) {
// now copy the accounts to cloud_usage db
_usageDao.updateBucketStats(bucketStats);
}
offset = new Long(offset.longValue() + limit.longValue());
} while ((bucketStats != null) && !bucketStats.isEmpty());
// reset offset
offset = Long.valueOf(0);
sc5 = _bucketStatisticsDao.createSearchCriteria();
sc5.addAnd("id", SearchCriteria.Op.GT, lastBucketStatsId);
do {
Filter filter = new Filter(BucketStatisticsVO.class, "id", true, offset, limit);
bucketStats = _bucketStatisticsDao.search(sc5, filter);
if ((bucketStats != null) && !bucketStats.isEmpty()) {
// now copy the accounts to cloud_usage db
_usageDao.saveBucketStats(bucketStats);
}
offset = new Long(offset.longValue() + limit.longValue());
} while ((bucketStats != null) && !bucketStats.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 = _usageNetworkDao.getRecentNetworkStats();
Calendar recentlyDeletedCal = Calendar.getInstance(usageAggregationTimeZone);
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 = _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() + "-Network-" + userStat.getNetworkId();
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++;
}
_usageNetworkDao.saveUsageNetworks(usageNetworks);
if (logger.isDebugEnabled()) {
logger.debug("created network stats helper entries for " + numAcctsProcessed + " accts");
}
// get vm disk stats in order to compute vm disk usage
vmDiskUsages = _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 = _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++;
}
_usageVmDiskDao.saveUsageVmDisks(usageVmDisks);
if (logger.isDebugEnabled()) {
logger.debug("Created Instance 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(usageAggregationTimeZone);
while ((tempDate.after(startDate)) && ((tempDate.getTime() - startDate.getTime()) > 60000)) {
currentEndDate = tempDate;
aggregateCal.setTime(tempDate);
aggregateCal.add(Calendar.MINUTE, -_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 = _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 (logger.isDebugEnabled()) {
logger.debug("Processed Instance/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 = _accountDao.findRecentlyDeletedAccounts(null, recentlyDeletedDate, filter);
if ((accounts != null) && !accounts.isEmpty()) {
for (AccountVO account : accounts) {
parsed = parseHelperTables(account, currentStartDate, currentEndDate);
List<Long> publicTemplates = _usageDao.listPublicTemplatesByAccount(account.getId());
for (Long templateId : publicTemplates) {
//mark public templates owned by deleted accounts as deleted
List<UsageStorageVO> storageVOs = _usageStorageDao.listById(account.getId(), templateId, StorageTypes.TEMPLATE);
if (storageVOs.size() > 1) {
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 (logger.isDebugEnabled()) {
logger.debug("Deleting Template: " + storageVO.getId() + " from Account: " + storageVO.getAccountId());
}
storageVO.setDeleted(account.getRemoved());
_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, _aggregationDuration);
currentEndDate = aggregateCal.getTime();
}
if (logger.isDebugEnabled()) {
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) {
logger.error("Exception in usage manager", ex);
usageTxn.rollback();
} finally {
// everything seemed to work...set endDate as the last success date
_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) {
_usageJobDao.createNewJob(_hostname, _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) {
logger.error("Usage Manager error", e);
}
}
private boolean parseHelperTables(AccountVO account, Date currentStartDate, Date currentEndDate) {
boolean parsed = false;
for (UsageParser parser : usageParsers) {
parsed = parser.doParsing(account, currentStartDate, currentEndDate);
logger.debug("{} usage was {} parsed for [{}].", parser.getParserName(), parsed ? "successfully" : "not successfully", account);
}
return parsed;
}
private void createHelperRecord(UsageEventVO event) {
String eventType = event.getType();
try {
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)) {
handleVpnUserEvent(event);
} else if (isSecurityGroupEvent(eventType)) {
createSecurityGroupEvent(event);
} else if (isVmSnapshotEvent(eventType)) {
handleVMSnapshotEvent(event);
} else if (isVmSnapshotOnPrimaryEvent(eventType)) {
createVmSnapshotOnPrimaryEvent(event);
} else if (isBackupEvent(eventType)) {
createBackupEvent(event);
} else if (EventTypes.isNetworkEvent(eventType)) {
handleNetworkEvent(event);
} else if (EventTypes.isVpcEvent(eventType)) {
handleVpcEvent(event);
}
} catch (EntityExistsException e) {
logger.warn(String.format("Failed to create usage event id: %d type: %s due to %s", event.getId(), eventType, e.getMessage()), e);
}
}
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) ||
eventType.equals(EventTypes.EVENT_VOLUME_UPLOAD) ||
eventType.equals(EventTypes.EVENT_VOLUME_ATTACH) ||
eventType.equals(EventTypes.EVENT_VOLUME_DETACH));
}
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 boolean isVmSnapshotOnPrimaryEvent(String eventType) {
if (eventType == null)
return false;
return (eventType.equals(EventTypes.EVENT_VM_SNAPSHOT_ON_PRIMARY) || eventType.equals(EventTypes.EVENT_VM_SNAPSHOT_OFF_PRIMARY));
}
private boolean isBackupEvent(String eventType) {
return eventType != null && (
eventType.equals(EventTypes.EVENT_VM_BACKUP_OFFERING_ASSIGN) ||
eventType.equals(EventTypes.EVENT_VM_BACKUP_OFFERING_REMOVED_AND_BACKUPS_DELETED) ||
eventType.equals(EventTypes.EVENT_VM_BACKUP_USAGE_METRIC));
}
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 = _usageInstanceDao.createSearchCriteria();
sc.addAnd("vmInstanceId", SearchCriteria.Op.EQ, Long.valueOf(vmId));
sc.addAnd("endDate", SearchCriteria.Op.NULL);
sc.addAnd("usageType", SearchCriteria.Op.EQ, UsageTypes.RUNNING_VM);
List<UsageVMInstanceVO> usageInstances = _usageInstanceDao.search(sc, null);
if (usageInstances != null) {
if (usageInstances.size() > 0) {
logger.error("Found entries for a Instance running with id: " + vmId + ", which are not stopped. Ending them all...");
for (UsageVMInstanceVO usageInstance : usageInstances) {
usageInstance.setEndDate(event.getCreateDate());
_usageInstanceDao.update(usageInstance);
}
}
}
sc = _usageInstanceDao.createSearchCriteria();
sc.addAnd("vmInstanceId", SearchCriteria.Op.EQ, Long.valueOf(vmId));
sc.addAnd("endDate", SearchCriteria.Op.NULL);
sc.addAnd("usageType", SearchCriteria.Op.EQ, UsageTypes.ALLOCATED_VM);
usageInstances = _usageInstanceDao.search(sc, null);
if (usageInstances == null || (usageInstances.size() == 0)) {
logger.error("Cannot find allocated Instance entry for a Instance running with id: " + vmId);
} else if (usageInstances.size() == 1) {
UsageVMInstanceVO usageInstance = usageInstances.get(0);
if (usageInstance.getSerivceOfferingId() != soId) {
//Service Offering changed after Vm creation
//End current Allocated usage and create new Allocated Vm entry with new soId
usageInstance.setEndDate(event.getCreateDate());
_usageInstanceDao.update(usageInstance);
usageInstance.setServiceOfferingId(soId);
usageInstance.setStartDate(event.getCreateDate());
usageInstance.setEndDate(null);
populateDynamicComputeOfferingDetailsAndPersist(usageInstance, event.getId());
}
}
Long templateId = event.getTemplateId();
String hypervisorType = event.getResourceType();
// add this VM to the usage helper table
UsageVMInstanceVO usageInstanceNew =
new UsageVMInstanceVO(UsageTypes.RUNNING_VM, zoneId, event.getAccountId(), vmId, vmName, soId, templateId, hypervisorType, event.getCreateDate(),
null);
populateDynamicComputeOfferingDetailsAndPersist(usageInstanceNew, event.getId());
} catch (Exception ex) {
logger.error("Error saving Instance usage for Instance: " + vmId, ex);
}
} else if (EventTypes.EVENT_VM_STOP.equals(event.getType())) {
// find the latest usage_VM_instance row, update the stop date (should be null) to the event date
// FIXME: search criteria needs to have some kind of type information so we distinguish between START/STOP and CREATE/DESTROY
SearchCriteria<UsageVMInstanceVO> sc = _usageInstanceDao.createSearchCriteria();
sc.addAnd("vmInstanceId", SearchCriteria.Op.EQ, Long.valueOf(vmId));
sc.addAnd("endDate", SearchCriteria.Op.NULL);
sc.addAnd("usageType", SearchCriteria.Op.EQ, UsageTypes.RUNNING_VM);
List<UsageVMInstanceVO> usageInstances = _usageInstanceDao.search(sc, null);
if (usageInstances != null) {
if (usageInstances.size() > 1) {
logger.warn("Found multiple entries for a Instance running with id: " + vmId + ", ending them all...");
}
for (UsageVMInstanceVO usageInstance : usageInstances) {
usageInstance.setEndDate(event.getCreateDate());
// TODO: UsageVMInstanceVO should have an ID field and we should do updates through that field since we are really
// updating one row at a time here
_usageInstanceDao.update(usageInstance);
}
}
} else if (EventTypes.EVENT_VM_CREATE.equals(event.getType())) {
try {
Long templateId = event.getTemplateId();
String hypervisorType = event.getResourceType();
// add this VM to the usage helper table
UsageVMInstanceVO usageInstanceNew = new UsageVMInstanceVO(UsageTypes.ALLOCATED_VM, zoneId, event.getAccountId(), vmId, vmName,
soId, templateId, hypervisorType, event.getCreateDate(), null);
populateDynamicComputeOfferingDetailsAndPersist(usageInstanceNew, event.getId());
} catch (Exception ex) {
logger.error("Error saving usage for Instance: " + vmId, ex);
}
} else if (EventTypes.EVENT_VM_DESTROY.equals(event.getType())) {
SearchCriteria<UsageVMInstanceVO> sc = _usageInstanceDao.createSearchCriteria();
sc.addAnd("vmInstanceId", SearchCriteria.Op.EQ, Long.valueOf(vmId));
sc.addAnd("endDate", SearchCriteria.Op.NULL);
sc.addAnd("usageType", SearchCriteria.Op.EQ, UsageTypes.ALLOCATED_VM);
List<UsageVMInstanceVO> usageInstances = _usageInstanceDao.search(sc, null);
if (usageInstances != null) {
if (usageInstances.size() > 1) {
logger.warn("Found multiple entries for an Instance allocated with id: " + vmId + ", destroying them all...");
}
for (UsageVMInstanceVO usageInstance : usageInstances) {
usageInstance.setEndDate(event.getCreateDate());
_usageInstanceDao.update(usageInstance);
}
}
} else if (EventTypes.EVENT_VM_UPGRADE.equals(event.getType())) {
SearchCriteria<UsageVMInstanceVO> sc = _usageInstanceDao.createSearchCriteria();
sc.addAnd("vmInstanceId", SearchCriteria.Op.EQ, Long.valueOf(vmId));
sc.addAnd("endDate", SearchCriteria.Op.NULL);
sc.addAnd("usageType", SearchCriteria.Op.EQ, UsageTypes.ALLOCATED_VM);
List<UsageVMInstanceVO> usageInstances = _usageInstanceDao.search(sc, null);
if (usageInstances != null) {
if (usageInstances.size() > 1) {
logger.warn("Found multiple entries for an Instance allocated with id: " + vmId + ", updating end_date for all of them...");
}
for (UsageVMInstanceVO usageInstance : usageInstances) {
usageInstance.setEndDate(event.getCreateDate());
_usageInstanceDao.update(usageInstance);
}
}
Long templateId = event.getTemplateId();
String hypervisorType = event.getResourceType();
// add this VM to the usage helper table
UsageVMInstanceVO usageInstanceNew =
new UsageVMInstanceVO(UsageTypes.ALLOCATED_VM, zoneId, event.getAccountId(), vmId, vmName, soId, templateId, hypervisorType, event.getCreateDate(), null);
populateDynamicComputeOfferingDetailsAndPersist(usageInstanceNew, event.getId());
} else if (EventTypes.EVENT_VM_DYNAMIC_SCALE.equals(event.getType())) {
// Ending the running vm event
SearchCriteria<UsageVMInstanceVO> sc = _usageInstanceDao.createSearchCriteria();
sc.addAnd("vmInstanceId", SearchCriteria.Op.EQ, Long.valueOf(vmId));
sc.addAnd("endDate", SearchCriteria.Op.NULL);
sc.addAnd("usageType", SearchCriteria.Op.EQ, UsageTypes.RUNNING_VM);
List<UsageVMInstanceVO> usageInstances = _usageInstanceDao.search(sc, null);
if (usageInstances != null) {
if (usageInstances.size() > 1) {
logger.warn("Found multiple entries for Instance running with ID: " + vmId + ", ending them all...");
}
for (UsageVMInstanceVO usageInstance : usageInstances) {
usageInstance.setEndDate(event.getCreateDate());
_usageInstanceDao.update(usageInstance);
}
}
sc = _usageInstanceDao.createSearchCriteria();
sc.addAnd("vmInstanceId", SearchCriteria.Op.EQ, Long.valueOf(vmId));
sc.addAnd("endDate", SearchCriteria.Op.NULL);
sc.addAnd("usageType", SearchCriteria.Op.EQ, UsageTypes.ALLOCATED_VM);
usageInstances = _usageInstanceDao.search(sc, null);
if (usageInstances == null || (usageInstances.size() == 0)) {
logger.error("Cannot find allocated Instance entry for an Instance running with id: " + vmId);
} else if (usageInstances.size() == 1) {
UsageVMInstanceVO usageInstance = usageInstances.get(0);
if (usageInstance.getSerivceOfferingId() != soId) {
//Service Offering changed after Vm creation
//End current Allocated usage and create new Allocated Vm entry with new soId
usageInstance.setEndDate(event.getCreateDate());
_usageInstanceDao.update(usageInstance);
usageInstance.setServiceOfferingId(soId);
usageInstance.setStartDate(event.getCreateDate());
usageInstance.setEndDate(null);
populateDynamicComputeOfferingDetailsAndPersist(usageInstance, event.getId());
}
}
Long templateId = event.getTemplateId();
String hypervisorType = event.getResourceType();
// add this VM to the usage helper table with new service offering Id
UsageVMInstanceVO usageInstanceNew =
new UsageVMInstanceVO(UsageTypes.RUNNING_VM, zoneId, event.getAccountId(), vmId, vmName, soId, templateId, hypervisorType, event.getCreateDate(), null);
populateDynamicComputeOfferingDetailsAndPersist(usageInstanceNew, event.getId());
}
}
private void populateDynamicComputeOfferingDetailsAndPersist(UsageVMInstanceVO usageInstance, Long eventId) {
//populate the cpu, memory and cpuSpeed of the vm when created from a dynamic offering.
UsageEventDetailsVO cpuNumber = _usageEventDetailsDao.findDetail(eventId, UsageEventVO.DynamicParameters.cpuNumber.name());
if (cpuNumber != null) {
usageInstance.setCpuCores(Long.parseLong(cpuNumber.getValue()));
} else {
usageInstance.setCpuCores(null);
}
UsageEventDetailsVO cpuSpeed = _usageEventDetailsDao.findDetail(eventId, UsageEventVO.DynamicParameters.cpuSpeed.name());
if (cpuSpeed != null) {
usageInstance.setCpuSpeed(Long.parseLong(cpuSpeed.getValue()));
} else {
usageInstance.setCpuSpeed(null);
}
UsageEventDetailsVO memory = _usageEventDetailsDao.findDetail(eventId, UsageEventVO.DynamicParameters.memory.name());
if (memory != null) {
usageInstance.setMemory(Long.parseLong(memory.getValue()));
} else {
usageInstance.setMemory(null);
}
_usageInstanceDao.persist(usageInstance);
}
private void createNetworkHelperEntry(UserStatisticsVO userStat, UsageNetworkVO usageNetworkStats, long timestamp) {
long currentAccountedBytesSent = 0L;
long currentAccountedBytesReceived = 0L;
if (usageNetworkStats != null) {
if (logger.isDebugEnabled()) {
logger.debug("Getting current accounted bytes for... accountId: " + usageNetworkStats.getAccountId() + " in zone: " + userStat.getDataCenterId() +
"; abr: " + toHumanReadableSize(usageNetworkStats.getAggBytesReceived()) + "; abs: " + toHumanReadableSize(usageNetworkStats.getAggBytesSent()));
}
currentAccountedBytesSent = usageNetworkStats.getAggBytesSent();
currentAccountedBytesReceived = usageNetworkStats.getAggBytesReceived();
}
long bytesSent = userStat.getAggBytesSent() - currentAccountedBytesSent;
long bytesReceived = userStat.getAggBytesReceived() - currentAccountedBytesReceived;
if (bytesSent < 0) {
logger.warn("Calculated negative value for bytes sent: " + toHumanReadableSize(bytesSent) + ", user stats say: " + toHumanReadableSize(userStat.getAggBytesSent()) +
", previous network usage was: " + toHumanReadableSize(currentAccountedBytesSent));
bytesSent = 0;
}
if (bytesReceived < 0) {
logger.warn("Calculated negative value for bytes received: " + toHumanReadableSize(bytesReceived) + ", user stats say: " + toHumanReadableSize(userStat.getAggBytesReceived()) +
", previous network usage was: " + toHumanReadableSize(currentAccountedBytesReceived));
bytesReceived = 0;
}
long hostId = 0;
if (userStat.getDeviceId() != null) {
hostId = userStat.getDeviceId();
}
UsageNetworkVO usageNetworkVO =
new UsageNetworkVO(userStat.getAccountId(), userStat.getDataCenterId(), hostId, userStat.getDeviceType(), userStat.getNetworkId(), bytesSent, bytesReceived,
userStat.getAggBytesReceived(), userStat.getAggBytesSent(), timestamp);
if (logger.isDebugEnabled()) {
logger.debug("creating networkHelperEntry... accountId: " + userStat.getAccountId() + " in zone: " + userStat.getDataCenterId() + "; abr: " +
userStat.getAggBytesReceived() + "; abs: " + userStat.getAggBytesSent() + "; curABS: " + currentAccountedBytesSent + "; curABR: " +
currentAccountedBytesReceived + "; ubs: " + bytesSent + "; ubr: " + bytesReceived);
}
usageNetworks.add(usageNetworkVO);
}
private void createVmDiskHelperEntry(VmDiskStatisticsVO vmDiskStat, UsageVmDiskVO usageVmDiskStat, long timestamp) {
long currentAccountedIORead = 0L;
long currentAccountedIOWrite = 0L;
long currentAccountedBytesRead = 0L;
long currentAccountedBytesWrite = 0L;
if (usageVmDiskStat != null) {
if (logger.isDebugEnabled()) {
logger.debug("Getting current accounted bytes for... accountId: " + usageVmDiskStat.getAccountId() + " in zone: " + vmDiskStat.getDataCenterId() +
"; aiw: " + toHumanReadableSize(vmDiskStat.getAggIOWrite()) + "; air: " + toHumanReadableSize(usageVmDiskStat.getAggIORead()) + "; abw: " + toHumanReadableSize(vmDiskStat.getAggBytesWrite()) + "; abr: " +
toHumanReadableSize(usageVmDiskStat.getAggBytesRead()));
}
currentAccountedIORead = usageVmDiskStat.getAggIORead();
currentAccountedIOWrite = usageVmDiskStat.getAggIOWrite();
currentAccountedBytesRead = usageVmDiskStat.getAggBytesRead();
currentAccountedBytesWrite = usageVmDiskStat.getAggBytesWrite();
}
long ioRead = vmDiskStat.getAggIORead() - currentAccountedIORead;
long ioWrite = vmDiskStat.getAggIOWrite() - currentAccountedIOWrite;
long bytesRead = vmDiskStat.getAggBytesRead() - currentAccountedBytesRead;
long bytesWrite = vmDiskStat.getAggBytesWrite() - currentAccountedBytesWrite;
if (ioRead < 0) {
logger.warn("Calculated negative value for IO read: " + toHumanReadableSize(ioRead) + ", Instance disk stats say: " + toHumanReadableSize(vmDiskStat.getAggIORead()) + ", previous Instance disk usage was: " +
toHumanReadableSize(currentAccountedIORead));
ioRead = 0;
}
if (ioWrite < 0) {
logger.warn("Calculated negative value for IO write: " + toHumanReadableSize(ioWrite) + ", Instance disk stats say: " + toHumanReadableSize(vmDiskStat.getAggIOWrite()) + ", previous Instance disk usage was: " +
toHumanReadableSize(currentAccountedIOWrite));
ioWrite = 0;
}
if (bytesRead < 0) {
logger.warn("Calculated negative value for bytes read: " + toHumanReadableSize(bytesRead) + ", vm disk stats say: " + toHumanReadableSize(vmDiskStat.getAggBytesRead()) +
", previous vm disk usage was: " + toHumanReadableSize(currentAccountedBytesRead));
bytesRead = 0;
}
if (bytesWrite < 0) {
logger.warn("Calculated negative value for bytes write: " + toHumanReadableSize(bytesWrite) + ", vm disk stats say: " + toHumanReadableSize(vmDiskStat.getAggBytesWrite()) +
", previous vm disk usage was: " + toHumanReadableSize(currentAccountedBytesWrite));
bytesWrite = 0;
}
long vmId = 0;
if (vmDiskStat.getVmId() != null) {
vmId = vmDiskStat.getVmId();
}
UsageVmDiskVO usageVmDiskVO =
new UsageVmDiskVO(vmDiskStat.getAccountId(), vmDiskStat.getDataCenterId(), vmId, vmDiskStat.getVolumeId(), ioRead, ioWrite, vmDiskStat.getAggIORead(),
vmDiskStat.getAggIOWrite(), bytesRead, bytesWrite, vmDiskStat.getAggBytesRead(), vmDiskStat.getAggBytesWrite(), timestamp);
if (logger.isDebugEnabled()) {
logger.debug("creating vmDiskHelperEntry... accountId: " + vmDiskStat.getAccountId() + " in zone: " + vmDiskStat.getDataCenterId() + "; aiw: " +
toHumanReadableSize(vmDiskStat.getAggIOWrite()) + "; air: " + toHumanReadableSize(vmDiskStat.getAggIORead()) + "; curAIR: " + toHumanReadableSize(currentAccountedIORead) + "; curAIW: " + toHumanReadableSize(currentAccountedIOWrite) +
"; uir: " + toHumanReadableSize(ioRead) + "; uiw: " + toHumanReadableSize(ioWrite) + "; abw: " + toHumanReadableSize(vmDiskStat.getAggBytesWrite()) + "; abr: " + toHumanReadableSize(vmDiskStat.getAggBytesRead()) + "; curABR: " +
toHumanReadableSize(currentAccountedBytesRead) + "; curABW: " + toHumanReadableSize(currentAccountedBytesWrite) + "; ubr: " + toHumanReadableSize(bytesRead) + "; ubw: " + toHumanReadableSize(bytesWrite));
}
usageVmDisks.add(usageVmDiskVO);
}
private void createIPHelperEvent(UsageEventVO event) {
String ipAddress = event.getResourceName();
if (EventTypes.EVENT_NET_IP_ASSIGN.equals(event.getType())) {
if (logger.isDebugEnabled()) {
logger.debug("Assigning IP address: " + ipAddress + " to Account: " + event.getAccountId());
}
Account acct = _accountDao.findByIdIncludingRemoved(event.getAccountId());
long zoneId = event.getZoneId();
long id = event.getResourceId();
long sourceNat = event.getSize();
boolean isSourceNat = (sourceNat == 1) ? true : false;
boolean isSystem = (event.getTemplateId() == null || event.getTemplateId() == 0) ? false : true;
final UsageEventDetailsVO hiddenDetail = _usageEventDetailsDao.findDetail(event.getId(), "hidden");
final boolean isHidden = hiddenDetail != null && "true".equals(hiddenDetail.getValue());
UsageIPAddressVO ipAddressVO =
new UsageIPAddressVO(id, event.getAccountId(), acct.getDomainId(), zoneId, ipAddress, isSourceNat, isSystem, event.getCreateDate(), null, isHidden);
_usageIPAddressDao.persist(ipAddressVO);
} else if (EventTypes.EVENT_NET_IP_RELEASE.equals(event.getType())) {
SearchCriteria<UsageIPAddressVO> sc = _usageIPAddressDao.createSearchCriteria();
sc.addAnd("accountId", SearchCriteria.Op.EQ, event.getAccountId());
sc.addAnd("address", SearchCriteria.Op.EQ, ipAddress);
sc.addAnd("released", SearchCriteria.Op.NULL);
List<UsageIPAddressVO> ipAddressVOs = _usageIPAddressDao.search(sc, null);
if (ipAddressVOs.size() > 1) {
logger.warn("More that one usage entry for IP address: " + ipAddress + " assigned to Account: " + event.getAccountId() +
"; marking them all as released...");
}
for (UsageIPAddressVO ipAddressVO : ipAddressVOs) {
if (logger.isDebugEnabled()) {
logger.debug("Releasing IP address: " + ipAddressVO.getAddress() + " from Account: " + ipAddressVO.getAccountId());
}
ipAddressVO.setReleased(event.getCreateDate()); // there really shouldn't be more than one
_usageIPAddressDao.update(ipAddressVO);
}
}
}
private void deleteExistingSecondaryStorageUsageForVolume(long volId, long accountId, Date deletedDate) {
List<UsageStorageVO> storageVOs = _usageStorageDao.listById(accountId, volId, StorageTypes.VOLUME);
for (UsageStorageVO storageVO : storageVOs) {
logger.debug("Setting the volume with id: {} to 'deleted' in the usage_storage table for account: {}.", volId, accountId);
storageVO.setDeleted(deletedDate);
_usageStorageDao.update(storageVO);
}
}
private void deleteExistingInstanceVolumeUsage(long volId, long accountId, Date deletedDate) {
List<UsageVolumeVO> volumesVOs = _usageVolumeDao.listByVolumeId(volId, accountId);
for (UsageVolumeVO volumesVO : volumesVOs) {
if (volumesVO.getVmId() != null) {
logger.debug("Setting the volume with id: {} for instance id: {} to 'deleted' in the usage_volume table for account {}.",
volumesVO.getVolumeId(), volumesVO.getVmId(), accountId);
volumesVO.setDeleted(deletedDate);
_usageVolumeDao.update(volumesVO.getId(), volumesVO);
}
}
}
private void deleteExistingVolumeUsage(long volId, long accountId, Date deletedDate) {
List<UsageVolumeVO> volumesVOs = _usageVolumeDao.listByVolumeId(volId, accountId);
for (UsageVolumeVO volumesVO : volumesVOs) {
logger.debug("Setting the volume with id: {} to 'deleted' in the usage_volume table for account: {}.", volId, accountId);
volumesVO.setDeleted(deletedDate);
_usageVolumeDao.update(volumesVO.getId(), volumesVO);
}
}
private void createVolumeHelperEvent(UsageEventVO event) {
long volId = event.getResourceId();
Account acct = _accountDao.findByIdIncludingRemoved(event.getAccountId());
List<UsageVolumeVO> volumesVOs;
UsageVolumeVO volumeVO;
switch (event.getType()) {
case EventTypes.EVENT_VOLUME_CREATE:
//For volumes which are 'attached' successfully from uploaded state, set the 'deleted' column in the usage_storage table,
//so that the secondary storage should stop accounting and only primary will be accounted.
deleteExistingSecondaryStorageUsageForVolume(volId, event.getAccountId(), event.getCreateDate());
volumesVOs = _usageVolumeDao.listByVolumeId(volId, event.getAccountId());
if (volumesVOs.size() > 0) {
//This is a safeguard to avoid double counting of volumes.
logger.error("Found duplicate usage entry for volume: {} assigned to Account: ; marking as deleted...", volId, event.getAccountId());
deleteExistingVolumeUsage(volId, event.getAccountId(), event.getCreateDate());
}
logger.debug("Creating a new entry in usage_volume for volume with id: {} for account: {}", volId, event.getAccountId());
volumeVO = new UsageVolumeVO(volId, event.getZoneId(), event.getAccountId(), acct.getDomainId(), event.getOfferingId(), event.getTemplateId(), null, event.getSize(), event.getCreateDate(), null);
_usageVolumeDao.persist(volumeVO);
if (event.getVmId() != null) {
volumeVO = new UsageVolumeVO(volId, event.getZoneId(), event.getAccountId(), acct.getDomainId(), event.getOfferingId(), event.getTemplateId(), event.getVmId(), event.getSize(), event.getCreateDate(), null);
_usageVolumeDao.persist(volumeVO);
}
break;
case EventTypes.EVENT_VOLUME_RESIZE:
volumesVOs = _usageVolumeDao.listByVolumeId(volId, event.getAccountId());
for (UsageVolumeVO volumesVO : volumesVOs) {
String delete_msg = String.format("Setting the volume with id: %s to 'deleted' in the usage_volume table for account: %s.", volId, event.getAccountId());
String create_msg = String.format("Creating a new entry in usage_volume for volume with id: %s after resize for account: %s", volId, event.getAccountId());
Long vmId = volumesVO.getVmId();
if (vmId != null) {
delete_msg = String.format("Setting the volume with id: %s for instance id: %s to 'deleted' in the usage_volume table for account: %s.",
volId, vmId, event.getAccountId());
create_msg = String.format("Creating a new entry in usage_volume for volume with id: %s and instance id: %s after resize for account: %s",
volId, vmId, event.getAccountId());
}
logger.debug(delete_msg);
volumesVO.setDeleted(event.getCreateDate());
_usageVolumeDao.update(volumesVO.getId(), volumesVO);
logger.debug(create_msg);
volumeVO = new UsageVolumeVO(volId, event.getZoneId(), event.getAccountId(), acct.getDomainId(), event.getOfferingId(), event.getTemplateId(), vmId, event.getSize(), event.getCreateDate(), null);
_usageVolumeDao.persist(volumeVO);
}
break;
case EventTypes.EVENT_VOLUME_DELETE:
deleteExistingVolumeUsage(volId, event.getAccountId(), event.getCreateDate());
break;
case EventTypes.EVENT_VOLUME_ATTACH:
deleteExistingInstanceVolumeUsage(event.getResourceId(), event.getAccountId(), event.getCreateDate());
logger.debug("Creating a new entry in usage_volume for volume with id: {}, and instance id: {} for account: {}",
volId, event.getVmId(), event.getAccountId());
volumeVO = new UsageVolumeVO(volId, event.getZoneId(), event.getAccountId(), acct.getDomainId(), event.getOfferingId(), event.getTemplateId(), event.getVmId(), event.getSize(), event.getCreateDate(), null);
_usageVolumeDao.persist(volumeVO);
break;
case EventTypes.EVENT_VOLUME_DETACH:
deleteExistingInstanceVolumeUsage(event.getResourceId(), event.getAccountId(), event.getCreateDate());
break;
case EventTypes.EVENT_VOLUME_UPLOAD:
deleteExistingSecondaryStorageUsageForVolume(volId, event.getAccountId(), event.getCreateDate());
logger.debug("Creating a new entry in usage_storage for volume with id : {} for account: {}", volId, event.getAccountId());
UsageStorageVO storageVO = new UsageStorageVO(volId, event.getZoneId(), event.getAccountId(), acct.getDomainId(), StorageTypes.VOLUME, event.getTemplateId(), event.getSize(), event.getCreateDate(), null);
_usageStorageDao.persist(storageVO);
break;
}
}
private void createTemplateHelperEvent(UsageEventVO event) {
long templateId = -1L;
long zoneId = -1L;
long templateSize = -1L;
templateId = event.getResourceId();
zoneId = event.getZoneId();
if (EventTypes.EVENT_TEMPLATE_CREATE.equals(event.getType()) || EventTypes.EVENT_TEMPLATE_COPY.equals(event.getType())) {
templateSize = event.getSize();
if (templateSize < 1) {
logger.error("Incorrect size for Template with Id " + templateId);
return;
}
if (zoneId == -1L) {
logger.error("Incorrect zoneId for Template with Id " + templateId);
return;
}
}
if (EventTypes.EVENT_TEMPLATE_CREATE.equals(event.getType()) || EventTypes.EVENT_TEMPLATE_COPY.equals(event.getType())) {
if (logger.isDebugEnabled()) {
logger.debug("Create Template with id : " + templateId + " for Account: " + event.getAccountId());
}
List<UsageStorageVO> storageVOs = _usageStorageDao.listByIdAndZone(event.getAccountId(), templateId, StorageTypes.TEMPLATE, zoneId);
if (storageVOs.size() > 0) {
logger.warn("Usage entry for Template: " + templateId + " assigned to Account: " + event.getAccountId() + "already exists in zone " + zoneId);
return;
}
Account acct = _accountDao.findByIdIncludingRemoved(event.getAccountId());
UsageStorageVO storageVO =
new UsageStorageVO(templateId, zoneId, event.getAccountId(), acct.getDomainId(), StorageTypes.TEMPLATE, event.getTemplateId(), templateSize,
event.getVirtualSize(), event.getCreateDate(), null);
_usageStorageDao.persist(storageVO);
} else if (EventTypes.EVENT_TEMPLATE_DELETE.equals(event.getType())) {
List<UsageStorageVO> storageVOs;
if (zoneId != -1L) {
storageVOs = _usageStorageDao.listByIdAndZone(event.getAccountId(), templateId, StorageTypes.TEMPLATE, zoneId);
} else {
storageVOs = _usageStorageDao.listById(event.getAccountId(), templateId, StorageTypes.TEMPLATE);
}
if (storageVOs.size() > 1) {
logger.warn("More that one usage entry for storage: " + templateId + " assigned to Account: " + event.getAccountId() +
"; marking them all as deleted...");
}
for (UsageStorageVO storageVO : storageVOs) {
if (logger.isDebugEnabled()) {
logger.debug("Deleting Template: " + storageVO.getId() + " from account: " + storageVO.getAccountId());
}
storageVO.setDeleted(event.getCreateDate()); // there really shouldn't be more than one
_usageStorageDao.update(storageVO);
}
}
}
private void createISOHelperEvent(UsageEventVO event) {
long isoSize = -1L;
long isoId = event.getResourceId();
long zoneId = event.getZoneId();
if (EventTypes.EVENT_ISO_CREATE.equals(event.getType()) || EventTypes.EVENT_ISO_COPY.equals(event.getType())) {
isoSize = event.getSize();
}
if (EventTypes.EVENT_ISO_CREATE.equals(event.getType()) || EventTypes.EVENT_ISO_COPY.equals(event.getType())) {
if (logger.isDebugEnabled()) {
logger.debug("Create ISO with ID: " + isoId + " for Account: " + event.getAccountId());
}
List<UsageStorageVO> storageVOs = _usageStorageDao.listByIdAndZone(event.getAccountId(), isoId, StorageTypes.ISO, zoneId);
if (storageVOs.size() > 0) {
logger.warn("Usage entry for ISO: " + isoId + " assigned to Account: " + event.getAccountId() + "already exists in zone " + zoneId);
return;
}
Account acct = _accountDao.findByIdIncludingRemoved(event.getAccountId());
UsageStorageVO storageVO =
new UsageStorageVO(isoId, zoneId, event.getAccountId(), acct.getDomainId(), StorageTypes.ISO, null, isoSize, isoSize, event.getCreateDate(), null);
_usageStorageDao.persist(storageVO);
} else if (EventTypes.EVENT_ISO_DELETE.equals(event.getType())) {
List<UsageStorageVO> storageVOs;
if (zoneId != -1L) {
storageVOs = _usageStorageDao.listByIdAndZone(event.getAccountId(), isoId, StorageTypes.ISO, zoneId);
} else {
storageVOs = _usageStorageDao.listById(event.getAccountId(), isoId, StorageTypes.ISO);
}
if (storageVOs.size() > 1) {
logger.warn("More that one usage entry for storage: " + isoId + " assigned to Account: " + event.getAccountId() + "; marking them all as deleted...");
}
for (UsageStorageVO storageVO : storageVOs) {
if (logger.isDebugEnabled()) {
logger.debug("Deleting ISO: " + storageVO.getId() + " from Account: " + storageVO.getAccountId());
}
storageVO.setDeleted(event.getCreateDate()); // there really shouldn't be more than one
_usageStorageDao.update(storageVO);
}
}
}
private void createSnapshotHelperEvent(UsageEventVO event) {
long snapSize = -1L;
long zoneId = -1L;
long snapId = event.getResourceId();
if (EventTypes.EVENT_SNAPSHOT_CREATE.equals(event.getType())) {
if (usageSnapshotSelection){
snapSize = event.getVirtualSize();
}else {
snapSize = event.getSize();
}
zoneId = event.getZoneId();
}
if (EventTypes.EVENT_SNAPSHOT_CREATE.equals(event.getType())) {
if (logger.isDebugEnabled()) {
logger.debug("Create Snapshot with id : " + snapId + " for Account: " + event.getAccountId());
}
Account acct = _accountDao.findByIdIncludingRemoved(event.getAccountId());
UsageStorageVO storageVO =
new UsageStorageVO(snapId, zoneId, event.getAccountId(), acct.getDomainId(), StorageTypes.SNAPSHOT, null, snapSize, event.getCreateDate(), null);
_usageStorageDao.persist(storageVO);
} else if (EventTypes.EVENT_SNAPSHOT_DELETE.equals(event.getType())) {
List<UsageStorageVO> storageVOs = _usageStorageDao.listById(event.getAccountId(), snapId, StorageTypes.SNAPSHOT);
if (storageVOs.size() > 1) {
logger.warn("More that one usage entry for storage: " + snapId + " assigned to Account: " + event.getAccountId() + "; marking them all as deleted...");
}
for (UsageStorageVO storageVO : storageVOs) {
if (logger.isDebugEnabled()) {
logger.debug("Deleting Snapshot: " + storageVO.getId() + " from Account: " + storageVO.getAccountId());
}
storageVO.setDeleted(event.getCreateDate()); // there really shouldn't be more than one
_usageStorageDao.update(storageVO);
}
}
}
private void createLoadBalancerHelperEvent(UsageEventVO event) {
long zoneId = -1L;
long id = event.getResourceId();
if (EventTypes.EVENT_LOAD_BALANCER_CREATE.equals(event.getType())) {
if (logger.isDebugEnabled()) {
logger.debug("Creating load balancer : " + id + " for account: " + event.getAccountId());
}
zoneId = event.getZoneId();
Account acct = _accountDao.findByIdIncludingRemoved(event.getAccountId());
UsageLoadBalancerPolicyVO lbVO = new UsageLoadBalancerPolicyVO(id, zoneId, event.getAccountId(), acct.getDomainId(), event.getCreateDate(), null);
_usageLoadBalancerPolicyDao.persist(lbVO);
} else if (EventTypes.EVENT_LOAD_BALANCER_DELETE.equals(event.getType())) {
SearchCriteria<UsageLoadBalancerPolicyVO> sc = _usageLoadBalancerPolicyDao.createSearchCriteria();
sc.addAnd("accountId", SearchCriteria.Op.EQ, event.getAccountId());
sc.addAnd("lbId", SearchCriteria.Op.EQ, id);
sc.addAnd("deleted", SearchCriteria.Op.NULL);
List<UsageLoadBalancerPolicyVO> lbVOs = _usageLoadBalancerPolicyDao.search(sc, null);
if (lbVOs.size() > 1) {
logger.warn("More that one usage entry for load balancer policy: " + id + " assigned to account: " + event.getAccountId() +
"; marking them all as deleted...");
}
for (UsageLoadBalancerPolicyVO lbVO : lbVOs) {
if (logger.isDebugEnabled()) {
logger.debug("deleting load balancer policy: " + lbVO.getId() + " from account: " + lbVO.getAccountId());
}
lbVO.setDeleted(event.getCreateDate()); // there really shouldn't be more than one
_usageLoadBalancerPolicyDao.update(lbVO);
}
}
}
private void createPortForwardingHelperEvent(UsageEventVO event) {
long zoneId = -1L;
long id = event.getResourceId();
if (EventTypes.EVENT_NET_RULE_ADD.equals(event.getType())) {
if (logger.isDebugEnabled()) {
logger.debug("Creating port forwarding rule : " + id + " for account: " + event.getAccountId());
}
zoneId = event.getZoneId();
Account acct = _accountDao.findByIdIncludingRemoved(event.getAccountId());
UsagePortForwardingRuleVO pfVO = new UsagePortForwardingRuleVO(id, zoneId, event.getAccountId(), acct.getDomainId(), event.getCreateDate(), null);
_usagePortForwardingRuleDao.persist(pfVO);
} else if (EventTypes.EVENT_NET_RULE_DELETE.equals(event.getType())) {
SearchCriteria<UsagePortForwardingRuleVO> sc = _usagePortForwardingRuleDao.createSearchCriteria();
sc.addAnd("accountId", SearchCriteria.Op.EQ, event.getAccountId());
sc.addAnd("pfId", SearchCriteria.Op.EQ, id);
sc.addAnd("deleted", SearchCriteria.Op.NULL);
List<UsagePortForwardingRuleVO> pfVOs = _usagePortForwardingRuleDao.search(sc, null);
if (pfVOs.size() > 1) {
logger.warn("More that one usage entry for port forwarding rule: " + id + " assigned to account: " + event.getAccountId() +
"; marking them all as deleted...");
}
for (UsagePortForwardingRuleVO pfVO : pfVOs) {
if (logger.isDebugEnabled()) {
logger.debug("deleting port forwarding rule: " + pfVO.getId() + " from account: " + pfVO.getAccountId());
}
pfVO.setDeleted(event.getCreateDate()); // there really shouldn't be more than one
_usagePortForwardingRuleDao.update(pfVO);
}
}
}
private void createNetworkOfferingEvent(UsageEventVO event) {
long zoneId = -1L;
long vmId = event.getResourceId();
long networkOfferingId = event.getOfferingId();
long nicId = 0;
try {
nicId = Long.parseLong(event.getResourceName());
} catch (Exception e) {
logger.warn("Failed to get NIC id from resource name, resource name is: " + event.getResourceName());
}
if (EventTypes.EVENT_NETWORK_OFFERING_CREATE.equals(event.getType()) || EventTypes.EVENT_NETWORK_OFFERING_ASSIGN.equals(event.getType())) {
if (logger.isDebugEnabled()) {
logger.debug("Creating networking offering: " + networkOfferingId + " for Instance: " + vmId + " for Account: " + event.getAccountId());
}
zoneId = event.getZoneId();
Account acct = _accountDao.findByIdIncludingRemoved(event.getAccountId());
boolean isDefault = (event.getSize() == 1) ? true : false;
UsageNetworkOfferingVO networkOffering =
new UsageNetworkOfferingVO(zoneId, event.getAccountId(), acct.getDomainId(), vmId, networkOfferingId, nicId, isDefault, event.getCreateDate(), null);
_usageNetworkOfferingDao.persist(networkOffering);
} else if (EventTypes.EVENT_NETWORK_OFFERING_DELETE.equals(event.getType()) || EventTypes.EVENT_NETWORK_OFFERING_REMOVE.equals(event.getType())) {
SearchCriteria<UsageNetworkOfferingVO> sc = _usageNetworkOfferingDao.createSearchCriteria();
sc.addAnd("accountId", SearchCriteria.Op.EQ, event.getAccountId());
sc.addAnd("vmInstanceId", SearchCriteria.Op.EQ, vmId);
sc.addAnd("nicId", SearchCriteria.Op.EQ, nicId);
sc.addAnd("networkOfferingId", SearchCriteria.Op.EQ, networkOfferingId);
sc.addAnd("deleted", SearchCriteria.Op.NULL);
List<UsageNetworkOfferingVO> noVOs = _usageNetworkOfferingDao.search(sc, null);
if (noVOs.size() > 1) {
logger.warn("More that one usage entry for networking offering: " + networkOfferingId + " for Vm: " + vmId + " assigned to account: " +
event.getAccountId() + "; marking them all as deleted...");
}
for (UsageNetworkOfferingVO noVO : noVOs) {
if (logger.isDebugEnabled()) {
logger.debug("deleting network offering: " + noVO.getNetworkOfferingId() + " from Vm: " + noVO.getVmInstanceId());
}
noVO.setDeleted(event.getCreateDate()); // there really shouldn't be more than one
_usageNetworkOfferingDao.update(noVO);
}
}
}
/**
* Handles VPN user create and remove events:
* <ul>
* <li>EventTypes#EVENT_VPN_USER_ADD</li>
* <li>EventTypes#EVENT_VPN_USER_ADD</li>
* </ul>
* if the event received by this method is neither add nor remove, we ignore it.
*/
protected void handleVpnUserEvent(UsageEventVO event) {
long accountId = event.getAccountId();
Account account = _accountDao.findByIdIncludingRemoved(accountId);
long zoneId = event.getZoneId();
long userId = event.getResourceId();
switch (event.getType()) {
case EventTypes.EVENT_VPN_USER_ADD:
createUsageVpnUser(event, account);
break;
case EventTypes.EVENT_VPN_USER_REMOVE:
deleteUsageVpnUser(event, account);
break;
default:
logger.debug("The event [type={}, zoneId={}, accountId={}, userId={}, resourceName={}, createDate={}] is neither of type [{}] nor [{}].",
event.getType(), zoneId, accountId, userId, event.getResourceName(), DateUtil.displayDateInTimezone(usageAggregationTimeZone, event.getCreateDate()),
EventTypes.EVENT_VPN_USER_ADD, EventTypes.EVENT_VPN_USER_REMOVE);
}
}
/**
* Find and delete, if exists, usage VPN user entries
*/
protected void deleteUsageVpnUser(UsageEventVO event, Account account) {
long accountId = account.getId();
long userId = event.getResourceId();
long zoneId = event.getZoneId();
long domainId = account.getDomainId();
List<UsageVPNUserVO> usageVpnUsers = findUsageVpnUsers(accountId, zoneId, userId, domainId);
if (CollectionUtils.isEmpty(usageVpnUsers)) {
logger.warn(String.format("No usage entry for vpn user [%s] assigned to account [%s] domain [%s] and zone [%s] was found.",
userId, accountId, domainId, zoneId));
}
if (usageVpnUsers.size() > 1) {
logger.warn(String.format("More than one usage entry for vpn user [%s] assigned to account [%s] domain [%s] and zone [%s]; marking them all as deleted.", userId,
accountId, domainId, zoneId));
}
for (UsageVPNUserVO vpnUser : usageVpnUsers) {
logger.debug(String.format("Deleting vpn user [%s] assigned to account [%s] domain [%s] and zone [%s] that was created at [%s].", vpnUser.getUserId(),
vpnUser.getAccountId(), vpnUser.getDomainId(), vpnUser.getZoneId(), vpnUser.getCreated()));
vpnUser.setDeleted(new Date());
_usageVPNUserDao.update(vpnUser);
}
}
/**
* Creates an entry for the Usage VPN User.
* If there is already an entry in the database with the same accountId, domainId, userId and zoneId, we do not persist a new entry.
*/
protected void createUsageVpnUser(UsageEventVO event, Account account) {
long accountId = account.getId();
long userId = event.getResourceId();
long zoneId = event.getZoneId();
long domainId = account.getDomainId();
List<UsageVPNUserVO> usageVpnUsers = findUsageVpnUsers(accountId, zoneId, userId, domainId);
if (usageVpnUsers.size() > 0) {
logger.debug(String.format("We do not need to create the usage VPN user [%s] assigned to account [%s] because it already exists.", userId, accountId));
} else {
logger.debug(String.format("Creating VPN user [%s] assigned to account [%s] domain [%s], zone [%s], and created at [%s]", userId, accountId, domainId, zoneId,
event.getCreateDate()));
UsageVPNUserVO vpnUser = new UsageVPNUserVO(zoneId, accountId, domainId, userId, event.getResourceName(), event.getCreateDate(), null);
_usageVPNUserDao.persist(vpnUser);
}
}
protected List<UsageVPNUserVO> findUsageVpnUsers(long accountId, long zoneId, long userId, long domainId) {
SearchCriteria<UsageVPNUserVO> sc = _usageVPNUserDao.createSearchCriteria();
sc.addAnd("zoneId", SearchCriteria.Op.EQ, zoneId);
sc.addAnd("accountId", SearchCriteria.Op.EQ, accountId);
sc.addAnd("domainId", SearchCriteria.Op.EQ, domainId);
sc.addAnd("userId", SearchCriteria.Op.EQ, userId);
sc.addAnd("deleted", SearchCriteria.Op.NULL);
return _usageVPNUserDao.search(sc, null);
}
private void createSecurityGroupEvent(UsageEventVO event) {
long zoneId = -1L;
long vmId = event.getResourceId();
long sgId = event.getOfferingId();
if (EventTypes.EVENT_SECURITY_GROUP_ASSIGN.equals(event.getType())) {
if (logger.isDebugEnabled()) {
logger.debug("Assigning : security group" + sgId + " to Vm: " + vmId + " for account: " + event.getAccountId());
}
zoneId = event.getZoneId();
Account acct = _accountDao.findByIdIncludingRemoved(event.getAccountId());
UsageSecurityGroupVO securityGroup = new UsageSecurityGroupVO(zoneId, event.getAccountId(), acct.getDomainId(), vmId, sgId, event.getCreateDate(), null);
_usageSecurityGroupDao.persist(securityGroup);
} else if (EventTypes.EVENT_SECURITY_GROUP_REMOVE.equals(event.getType())) {
SearchCriteria<UsageSecurityGroupVO> sc = _usageSecurityGroupDao.createSearchCriteria();
sc.addAnd("accountId", SearchCriteria.Op.EQ, event.getAccountId());
sc.addAnd("vmInstanceId", SearchCriteria.Op.EQ, vmId);
sc.addAnd("securityGroupId", SearchCriteria.Op.EQ, sgId);
sc.addAnd("deleted", SearchCriteria.Op.NULL);
List<UsageSecurityGroupVO> sgVOs = _usageSecurityGroupDao.search(sc, null);
if (sgVOs.size() > 1) {
logger.warn("More that one usage entry for security group: " + sgId + " for Vm: " + vmId + " assigned to account: " + event.getAccountId() +
"; marking them all as deleted...");
}
for (UsageSecurityGroupVO sgVO : sgVOs) {
if (logger.isDebugEnabled()) {
logger.debug("deleting security group: " + sgVO.getSecurityGroupId() + " from Vm: " + sgVO.getVmInstanceId());
}
sgVO.setDeleted(event.getCreateDate()); // there really shouldn't be more than one
_usageSecurityGroupDao.update(sgVO);
}
}
}
/**
* Handles Vm Snapshot create and delete events:
* <ul>
* <li>EventTypes#EVENT_VM_SNAPSHOT_CREATE</li>
* <li>EventTypes#EVENT_VM_SNAPSHOT_DELETE</li>
* </ul>
* if the event received by this method is neither add nor remove, we ignore it.
*/
protected void handleVMSnapshotEvent(UsageEventVO event) {
switch (event.getType()) {
case EventTypes.EVENT_VM_SNAPSHOT_CREATE:
createUsageVMSnapshot(event);
break;
case EventTypes.EVENT_VM_SNAPSHOT_DELETE:
deleteUsageVMSnapshot(event);
break;
default:
logger.debug("The event [type={}, zoneId={}, accountId={}, resourceName={}, diskOfferingId={}, createDate={}] is neither of type [{}] nor [{}].",
event.getType(), event.getZoneId(), event.getAccountId(), event.getResourceName(), event.getOfferingId(),
DateUtil.displayDateInTimezone(usageAggregationTimeZone, event.getCreateDate()), EventTypes.EVENT_VM_SNAPSHOT_CREATE,
EventTypes.EVENT_VM_SNAPSHOT_DELETE);
}
}
/**
* Creates an entry for the Usage VM Snapshot.
*/
protected void createUsageVMSnapshot(UsageEventVO event) {
long accountId = event.getAccountId();
Account acct = _accountDao.findByIdIncludingRemoved(event.getAccountId());
long domainId = acct.getDomainId();
Long offeringId = event.getOfferingId();
long vmId = event.getResourceId();
long volumeId = event.getTemplateId();
long zoneId = event.getZoneId();
Date created = event.getCreateDate();
long size = (event.getSize() == null) ? 0 : event.getSize();
UsageEventDetailsVO detailVO = _usageEventDetailsDao.findDetail(event.getId(), UsageEventVO.DynamicParameters.vmSnapshotId.name());
Long vmSnapshotId = null;
if (detailVO != null) {
String snapId = detailVO.getValue();
vmSnapshotId = Long.valueOf(snapId);
}
logger.debug("Creating usage Instance Snapshot for Instance ID [{}] assigned to Account [{}] domain [{}], zone [{}], and created at [{}].",
vmId, accountId, domainId, zoneId, DateUtil.displayDateInTimezone(usageAggregationTimeZone, event.getCreateDate()));
UsageVMSnapshotVO vsVO = new UsageVMSnapshotVO(volumeId, zoneId, accountId, domainId, vmId, offeringId, size, created, null);
vsVO.setVmSnapshotId(vmSnapshotId);
_usageVMSnapshotDao.persist(vsVO);
}
/**
* Find and delete, if exists, usage VM Snapshots entries
*/
protected void deleteUsageVMSnapshot(UsageEventVO event) {
long accountId = event.getAccountId();
Account acct = _accountDao.findByIdIncludingRemoved(event.getAccountId());
Long domainId = acct.getDomainId();
Long diskOfferingId = event.getOfferingId();
long vmId = event.getResourceId();
long zoneId = event.getZoneId();
List<UsageVMSnapshotVO> usageVMSnapshots = findUsageVMSnapshots(accountId, zoneId, domainId, vmId, diskOfferingId);
if (CollectionUtils.isEmpty(usageVMSnapshots)){
logger.warn(String.format("No usage entry for Instance Snapshot for Instance ID [%s] assigned to Account [%s] domain [%s] and zone [%s] was found.",
vmId, accountId, domainId, zoneId));
}
if (usageVMSnapshots.size() > 1) {
logger.warn(String.format("More than one usage entry for Instance Snapshot for VM id [%s] assigned to Account [%s] domain [%s] and zone [%s]; marking them all as deleted.", vmId,
accountId, domainId, zoneId));
}
for (UsageVMSnapshotVO vmSnapshots : usageVMSnapshots) {
logger.debug(String.format("Deleting Instance Snapshot for Instance id [%s] assigned to Account [%s] domain [%s] and zone [%s] that was created at [%s].", vmSnapshots.getVmId(),
vmSnapshots.getAccountId(), vmSnapshots.getDomainId(), vmSnapshots.getZoneId(), vmSnapshots.getCreated()));
vmSnapshots.setProcessed(event.getCreateDate());
_usageVMSnapshotDao.update(vmSnapshots);
}
}
protected List<UsageVMSnapshotVO> findUsageVMSnapshots(long accountId, long zoneId, long domainId, long vmId, Long diskOfferingId) {
SearchCriteria<UsageVMSnapshotVO> sc = _usageVMSnapshotDao.createSearchCriteria();
sc.addAnd("zoneId", SearchCriteria.Op.EQ, zoneId);
sc.addAnd("accountId", SearchCriteria.Op.EQ, accountId);
sc.addAnd("domainId", SearchCriteria.Op.EQ, domainId);
sc.addAnd("vmId", SearchCriteria.Op.EQ, vmId);
sc.addAnd("diskOfferingId", SearchCriteria.Op.EQ, diskOfferingId);
sc.addAnd("processed", SearchCriteria.Op.NULL);
return _usageVMSnapshotDao.search(sc, null);
}
private void createVmSnapshotOnPrimaryEvent(UsageEventVO event) {
Long vmId = event.getResourceId();
String name = event.getResourceName();
if (EventTypes.EVENT_VM_SNAPSHOT_ON_PRIMARY.equals(event.getType())) {
Long zoneId = event.getZoneId();
Long accountId = event.getAccountId();
long physicalsize = (event.getSize() == null) ? 0 : event.getSize();
long virtualsize = (event.getVirtualSize() == null) ? 0 : event.getVirtualSize();
Date created = event.getCreateDate();
Account acct = _accountDao.findByIdIncludingRemoved(event.getAccountId());
Long domainId = acct.getDomainId();
UsageEventDetailsVO detailVO = _usageEventDetailsDao.findDetail(event.getId(), UsageEventVO.DynamicParameters.vmSnapshotId.name());
Long vmSnapshotId = null;
if (detailVO != null) {
String snapId = detailVO.getValue();
vmSnapshotId = Long.valueOf(snapId);
}
UsageSnapshotOnPrimaryVO vsVO = new UsageSnapshotOnPrimaryVO(vmId, zoneId, accountId, domainId, vmId, name, 0, virtualsize, physicalsize, created, null);
vsVO.setVmSnapshotId(vmSnapshotId);
if (logger.isDebugEnabled()) {
logger.debug("createSnapshotOnPrimaryEvent UsageSnapshotOnPrimaryVO " + vsVO);
}
_usageSnapshotOnPrimaryDao.persist(vsVO);
} else if (EventTypes.EVENT_VM_SNAPSHOT_OFF_PRIMARY.equals(event.getType())) {
QueryBuilder<UsageSnapshotOnPrimaryVO> sc = QueryBuilder.create(UsageSnapshotOnPrimaryVO.class);
sc.and(sc.entity().getAccountId(), SearchCriteria.Op.EQ, event.getAccountId());
sc.and(sc.entity().getVmId(), SearchCriteria.Op.EQ, vmId);
sc.and(sc.entity().getName(), SearchCriteria.Op.EQ, name);
sc.and(sc.entity().getDeleted(), SearchCriteria.Op.NULL);
List<UsageSnapshotOnPrimaryVO> vmsnaps = sc.list();
if (vmsnaps.size() > 1) {
logger.warn("More that one usage entry for Instance Snapshot: " + name + " for Instance id:" + vmId + " assigned to Account: " + event.getAccountId()
+ "; marking them all as deleted...");
}
for (UsageSnapshotOnPrimaryVO vmsnap : vmsnaps) {
if (logger.isDebugEnabled()) {
logger.debug("Deleting Instance Snapshot name: " + vmsnap.getName() + " from Account: " + vmsnap.getAccountId());
}
vmsnap.setDeleted(event.getCreateDate()); // there really shouldn't be more than one
_usageSnapshotOnPrimaryDao.updateDeleted(vmsnap);
}
}
}
private void createBackupEvent(final UsageEventVO event) {
Long vmId = event.getResourceId();
Long zoneId = event.getZoneId();
Long accountId = event.getAccountId();
Long backupOfferingId = event.getOfferingId();
Account account = _accountDao.findByIdIncludingRemoved(event.getAccountId());
Long domainId = account.getDomainId();
Date created = event.getCreateDate();
if (EventTypes.EVENT_VM_BACKUP_OFFERING_ASSIGN.equals(event.getType())) {
final UsageBackupVO backupVO = new UsageBackupVO(zoneId, accountId, domainId, vmId, backupOfferingId, created);
usageBackupDao.persist(backupVO);
} else if (EventTypes.EVENT_VM_BACKUP_OFFERING_REMOVED_AND_BACKUPS_DELETED.equals(event.getType())) {
usageBackupDao.removeUsage(accountId, vmId, backupOfferingId, event.getCreateDate());
} else if (EventTypes.EVENT_VM_BACKUP_USAGE_METRIC.equals(event.getType())) {
usageBackupDao.updateMetrics(vmId, backupOfferingId, event.getSize(), event.getVirtualSize());
}
}
private void handleNetworkEvent(UsageEventVO event) {
String eventType = event.getType();
if (EventTypes.EVENT_NETWORK_DELETE.equals(eventType)) {
removeNetworkHelperEntry(event);
} else if (EventTypes.EVENT_NETWORK_CREATE.equals(eventType)) {
createNetworkHelperEntry(event);
} else if (EventTypes.EVENT_NETWORK_UPDATE.equals(eventType)) {
updateNetworkHelperEntry(event);
} else {
logger.error(String.format("Unknown event type [%s] in Networks event parser. Skipping it.", eventType));
}
}
private void removeNetworkHelperEntry(UsageEventVO event) {
long networkId = event.getResourceId();
logger.debug(String.format("Removing helper entries of network [%s].", networkId));
usageNetworksDao.remove(networkId, event.getCreateDate());
}
private void createNetworkHelperEntry(UsageEventVO event) {
long networkId = event.getResourceId();
Account account = _accountDao.findByIdIncludingRemoved(event.getAccountId());
long domainId = account.getDomainId();
List<UsageNetworksVO> entries = usageNetworksDao.listAll(networkId);
if (!entries.isEmpty()) {
logger.warn(String.format("Received a NETWORK.CREATE event for a network [%s] that already has helper entries; " +
"therefore, we will not create a new one.", networkId));
return;
}
logger.debug(String.format("Creating a helper entry for network [%s].", networkId));
UsageNetworksVO usageNetworksVO = new UsageNetworksVO(networkId, event.getOfferingId(), event.getZoneId(),
event.getAccountId(), domainId, Network.State.Allocated.name(), event.getCreateDate(), null);
usageNetworksDao.persist(usageNetworksVO);
}
private void updateNetworkHelperEntry(UsageEventVO event) {
long networkId = event.getResourceId();
Account account = _accountDao.findByIdIncludingRemoved(event.getAccountId());
long domainId = account.getDomainId();
logger.debug(String.format("Marking previous helper entries of network [%s] as removed.", networkId));
usageNetworksDao.remove(networkId, event.getCreateDate());
logger.debug(String.format("Creating an updated helper entry for network [%s].", networkId));
UsageNetworksVO usageNetworksVO = new UsageNetworksVO(networkId, event.getOfferingId(), event.getZoneId(),
event.getAccountId(), domainId, event.getResourceType(), event.getCreateDate(), null);
usageNetworksDao.persist(usageNetworksVO);
}
private void handleVpcEvent(UsageEventVO event) {
String eventType = event.getType();
if (EventTypes.EVENT_VPC_DELETE.equals(eventType)) {
removeVpcHelperEntry(event);
} else if (EventTypes.EVENT_VPC_CREATE.equals(eventType)) {
createVpcHelperEntry(event);
} else {
logger.error(String.format("Unknown event type [%s] in VPC event parser. Skipping it.", eventType));
}
}
private void removeVpcHelperEntry(UsageEventVO event) {
long vpcId = event.getResourceId();
logger.debug(String.format("Removing helper entries of VPC [%s].", vpcId));
usageVpcDao.remove(vpcId, event.getCreateDate());
}
private void createVpcHelperEntry(UsageEventVO event) {
long vpcId = event.getResourceId();
Account account = _accountDao.findByIdIncludingRemoved(event.getAccountId());
long domainId = account.getDomainId();
List<UsageVpcVO> entries = usageVpcDao.listAll(vpcId);
if (!entries.isEmpty()) {
logger.warn(String.format("Active helper entries already exist for VPC [%s]; therefore, we will not create a new one.",
vpcId));
return;
}
logger.debug(String.format("Creating a helper entry for VPC [%s].", vpcId));
UsageVpcVO usageVPCVO = new UsageVpcVO(vpcId, event.getZoneId(), event.getAccountId(), domainId, Vpc.State.Enabled.name(), event.getCreateDate(), null);
usageVpcDao.persist(usageVPCVO);
}
private class Heartbeat extends ManagedContextRunnable {
@Override
protected void runInContext() {
TransactionLegacy usageTxn = TransactionLegacy.open(TransactionLegacy.USAGE_DB);
try {
if (!_heartbeatLock.lock(3)) { // 3 second timeout
if (logger.isTraceEnabled())
logger.trace("Heartbeat lock is in use by others, returning true as someone else will take over the job if required");
return;
}
try {
// check for one-off jobs
UsageJobVO nextJob = _usageJobDao.getNextImmediateJob();
if (nextJob != null) {
if (_hostname.equals(nextJob.getHost()) && (_pid == nextJob.getPid().intValue())) {
updateJob(nextJob.getId(), null, null, null, UsageJobVO.JOB_SCHEDULED);
scheduleParse();
}
}
if (_usageJobDao.getNextRecurringJob() == null) {
// Own the usage processing immediately if no other node is owning it
_usageJobDao.createNewJob(_hostname, _pid, UsageJobVO.JOB_TYPE_RECURRING);
}
Long jobId = _usageJobDao.checkHeartbeat(_hostname, _pid, _aggregationDuration);
if (jobId != null) {
// if I'm taking over the job...see how long it's been since the last job, and if it's more than the
// aggregation range...do a one off job to catch up. However, only do this if we are more than half
// the aggregation range away from executing the next job
long now = System.currentTimeMillis();
long timeToJob = _jobExecTime.getTimeInMillis() - now;
long timeSinceLastSuccessJob = 0;
long aggregationDurationMillis = _aggregationDuration * 60L * 1000L;
long lastSuccess = _usageJobDao.getLastJobSuccessDateMillis();
if (lastSuccess > 0) {
timeSinceLastSuccessJob = now - lastSuccess;
}
if ((timeSinceLastSuccessJob > 0) && (timeSinceLastSuccessJob > (aggregationDurationMillis - 100))) {
if (timeToJob > (aggregationDurationMillis / 2)) {
logger.debug("Heartbeat: it's been {} ms since last finished usage job and {} ms until next job (aggregation duration is {} minutes)",
timeSinceLastSuccessJob, timeToJob, _aggregationDuration);
if (isParsingJobRunning.get()) {
logger.debug("Heartbeat: A parsing job is already running");
} else {
logger.debug("Heartbeat: Scheduling an immediate job to catch up");
scheduleParse();
}
}
}
boolean changeOwner = updateJob(jobId, _hostname, Integer.valueOf(_pid), new Date(), UsageJobVO.JOB_NOT_SCHEDULED);
if (changeOwner) {
deleteOneOffJobs(_hostname, _pid);
}
}
} finally {
_heartbeatLock.unlock();
}
} catch (Exception ex) {
logger.error("error in heartbeat", ex);
} finally {
usageTxn.close();
}
}
@DB
protected boolean updateJob(Long jobId, String hostname, Integer pid, Date heartbeat, int scheduled) {
boolean changeOwner = false;
TransactionLegacy txn = TransactionLegacy.currentTxn();
try {
txn.start();
// take over the job, setting our hostname/pid/heartbeat time
UsageJobVO job = _usageJobDao.lockRow(jobId, Boolean.TRUE);
if (!job.getHost().equals(hostname) || !job.getPid().equals(pid)) {
changeOwner = true;
}
UsageJobVO jobForUpdate = _usageJobDao.createForUpdate();
if (hostname != null) {
jobForUpdate.setHost(hostname);
}
if (pid != null) {
jobForUpdate.setPid(pid);
}
if (heartbeat != null) {
jobForUpdate.setHeartbeat(heartbeat);
}
jobForUpdate.setScheduled(scheduled);
_usageJobDao.update(job.getId(), jobForUpdate);
txn.commit();
} catch (Exception dbEx) {
txn.rollback();
logger.error("error updating usage job", dbEx);
}
return changeOwner;
}
@DB
protected void deleteOneOffJobs(String hostname, int pid) {
SearchCriteria<UsageJobVO> sc = _usageJobDao.createSearchCriteria();
SearchCriteria<UsageJobVO> ssc = _usageJobDao.createSearchCriteria();
ssc.addOr("host", SearchCriteria.Op.NEQ, hostname);
ssc.addOr("pid", SearchCriteria.Op.NEQ, pid);
sc.addAnd("host", SearchCriteria.Op.SC, ssc);
sc.addAnd("endMillis", SearchCriteria.Op.EQ, Long.valueOf(0));
sc.addAnd("jobType", SearchCriteria.Op.EQ, Integer.valueOf(UsageJobVO.JOB_TYPE_SINGLE));
sc.addAnd("scheduled", SearchCriteria.Op.EQ, Integer.valueOf(0));
_usageJobDao.expunge(sc);
}
}
private class SanityCheck extends ManagedContextRunnable {
@Override
protected void runInContext() {
logger.info("running sanity check");
UsageSanityChecker usc = new UsageSanityChecker();
try {
String errors = usc.runSanityCheck();
if (errors.length() > 0) {
_alertMgr.sendAlert(AlertManager.AlertType.ALERT_TYPE_USAGE_SANITY_RESULT, 0, new Long(0), "Usage Sanity Check failed", errors);
} else {
_alertMgr.clearAlert(AlertManager.AlertType.ALERT_TYPE_USAGE_SANITY_RESULT, 0, 0);
}
} catch (SQLException e) {
logger.error("Error in sanity check", e);
}
}
}
private class AbandonJob extends Thread {
@Override
public void run() {
logger.info("exiting Usage Manager");
_usageJobDao.removeLastOpenJobsOwned(_hostname, _pid);
}
}
}