You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cloudstack.apache.org by ah...@apache.org on 2013/08/02 20:08:43 UTC
[4/9] Switched over to use the new jobs framework
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/8930cfa9/server/src/com/cloud/async/AsyncJobMBeanImpl.java
----------------------------------------------------------------------
diff --git a/server/src/com/cloud/async/AsyncJobMBeanImpl.java b/server/src/com/cloud/async/AsyncJobMBeanImpl.java
deleted file mode 100644
index 282dace..0000000
--- a/server/src/com/cloud/async/AsyncJobMBeanImpl.java
+++ /dev/null
@@ -1,144 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-package com.cloud.async;
-
-import java.util.Date;
-import java.util.TimeZone;
-
-import javax.management.StandardMBean;
-
-import com.cloud.utils.DateUtil;
-
-public class AsyncJobMBeanImpl extends StandardMBean implements AsyncJobMBean {
- private AsyncJobVO _jobVo;
-
- public AsyncJobMBeanImpl(AsyncJobVO jobVo) {
- super(AsyncJobMBean.class, false);
-
- _jobVo = jobVo;
- }
-
- public long getAccountId() {
- return _jobVo.getAccountId();
- }
-
- public long getUserId() {
- return _jobVo.getUserId();
- }
-
- public String getCmd() {
- return _jobVo.getCmd();
- }
-
- public String getCmdInfo() {
- return _jobVo.getCmdInfo();
- }
-
- public String getStatus() {
- int jobStatus = _jobVo.getStatus();
- switch(jobStatus) {
- case AsyncJobResult.STATUS_SUCCEEDED :
- return "Completed";
-
- case AsyncJobResult.STATUS_IN_PROGRESS:
- return "In preogress";
-
- case AsyncJobResult.STATUS_FAILED:
- return "failed";
- }
-
- return "Unknow";
- }
-
- public int getProcessStatus() {
- return _jobVo.getProcessStatus();
- }
-
- public int getResultCode() {
- return _jobVo.getResultCode();
- }
-
- public String getResult() {
- return _jobVo.getResult();
- }
-
- public String getInstanceType() {
- if(_jobVo.getInstanceType() != null)
- return _jobVo.getInstanceType().toString();
- return "N/A";
- }
-
- public String getInstanceId() {
- if(_jobVo.getInstanceId() != null)
- return String.valueOf(_jobVo.getInstanceId());
- return "N/A";
- }
-
- public String getInitMsid() {
- if(_jobVo.getInitMsid() != null) {
- return String.valueOf(_jobVo.getInitMsid());
- }
- return "N/A";
- }
-
- public String getCreateTime() {
- Date time = _jobVo.getCreated();
- if(time != null)
- return DateUtil.getDateDisplayString(TimeZone.getDefault(), time);
- return "N/A";
- }
-
- public String getLastUpdateTime() {
- Date time = _jobVo.getLastUpdated();
- if(time != null)
- return DateUtil.getDateDisplayString(TimeZone.getDefault(), time);
- return "N/A";
- }
-
- public String getLastPollTime() {
- Date time = _jobVo.getLastPolled();
-
- if(time != null)
- return DateUtil.getDateDisplayString(TimeZone.getDefault(), time);
- return "N/A";
- }
-
- public String getSyncQueueId() {
- SyncQueueItemVO item = _jobVo.getSyncSource();
- if(item != null && item.getQueueId() != null) {
- return String.valueOf(item.getQueueId());
- }
- return "N/A";
- }
-
- public String getSyncQueueContentType() {
- SyncQueueItemVO item = _jobVo.getSyncSource();
- if(item != null) {
- return item.getContentType();
- }
- return "N/A";
- }
-
- public String getSyncQueueContentId() {
- SyncQueueItemVO item = _jobVo.getSyncSource();
- if(item != null && item.getContentId() != null) {
- return String.valueOf(item.getContentId());
- }
- return "N/A";
- }
-
-}
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/8930cfa9/server/src/com/cloud/async/AsyncJobManager.java
----------------------------------------------------------------------
diff --git a/server/src/com/cloud/async/AsyncJobManager.java b/server/src/com/cloud/async/AsyncJobManager.java
deleted file mode 100644
index c9eee40..0000000
--- a/server/src/com/cloud/async/AsyncJobManager.java
+++ /dev/null
@@ -1,52 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-package com.cloud.async;
-
-import java.util.List;
-
-import org.apache.cloudstack.api.ApiCommandJobType;
-import org.apache.cloudstack.api.command.user.job.QueryAsyncJobResultCmd;
-import com.cloud.utils.component.Manager;
-
-public interface AsyncJobManager extends Manager {
-
- public AsyncJobExecutorContext getExecutorContext();
-
- public AsyncJobVO getAsyncJob(long jobId);
- public AsyncJobVO findInstancePendingAsyncJob(String instanceType, long instanceId);
-
- public List<? extends AsyncJob> findInstancePendingAsyncJobs(ApiCommandJobType instanceType, Long accountId);
-
- public long submitAsyncJob(AsyncJobVO job);
- public long submitAsyncJob(AsyncJobVO job, boolean scheduleJobExecutionInContext);
- public AsyncJobResult queryAsyncJobResult(long jobId);
-
- public void completeAsyncJob(long jobId, int jobStatus, int resultCode, Object resultObject);
- public void updateAsyncJobStatus(long jobId, int processStatus, Object resultObject);
- public void updateAsyncJobAttachment(long jobId, String instanceType, Long instanceId);
-
- public void releaseSyncSource(AsyncJobExecutor executor);
-
- public void syncAsyncJobExecution(AsyncJob job, String syncObjType, long syncObjId, long queueSizeLimit);
-
- /**
- * Queries for the status or final result of an async job.
- * @param cmd the command that specifies the job id
- * @return an async-call result object
- */
- public AsyncJob queryAsyncJobResult(QueryAsyncJobResultCmd cmd);
-}
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/8930cfa9/server/src/com/cloud/async/AsyncJobManagerImpl.java
----------------------------------------------------------------------
diff --git a/server/src/com/cloud/async/AsyncJobManagerImpl.java b/server/src/com/cloud/async/AsyncJobManagerImpl.java
deleted file mode 100644
index 3aceec4..0000000
--- a/server/src/com/cloud/async/AsyncJobManagerImpl.java
+++ /dev/null
@@ -1,887 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-package com.cloud.async;
-
-import java.io.File;
-import java.io.FileInputStream;
-import java.lang.reflect.Type;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.Random;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-
-import javax.ejb.Local;
-import javax.inject.Inject;
-import javax.naming.ConfigurationException;
-
-import org.apache.log4j.Logger;
-import org.apache.log4j.NDC;
-import org.springframework.beans.factory.NoSuchBeanDefinitionException;
-import org.springframework.stereotype.Component;
-
-import com.google.gson.Gson;
-import com.google.gson.reflect.TypeToken;
-
-import org.apache.cloudstack.api.ApiCommandJobType;
-import org.apache.cloudstack.api.ApiErrorCode;
-import org.apache.cloudstack.api.BaseAsyncCmd;
-import org.apache.cloudstack.api.ServerApiException;
-import org.apache.cloudstack.api.command.user.job.QueryAsyncJobResultCmd;
-import org.apache.cloudstack.api.response.ExceptionResponse;
-import org.apache.cloudstack.context.CallContext;
-import org.apache.cloudstack.framework.events.EventBus;
-import org.apache.cloudstack.framework.events.EventBusException;
-import org.apache.cloudstack.utils.identity.ManagementServerNode;
-
-import com.cloud.api.ApiDBUtils;
-import com.cloud.api.ApiDispatcher;
-import com.cloud.api.ApiGsonHelper;
-import com.cloud.api.ApiSerializerHelper;
-import com.cloud.async.dao.AsyncJobDao;
-import com.cloud.cluster.ClusterManagerListener;
-import com.cloud.cluster.ManagementServerHost;
-import com.cloud.configuration.Config;
-import com.cloud.configuration.dao.ConfigurationDao;
-import com.cloud.domain.DomainVO;
-import com.cloud.domain.dao.DomainDao;
-import com.cloud.event.EventCategory;
-import com.cloud.exception.InvalidParameterValueException;
-import com.cloud.exception.PermissionDeniedException;
-import com.cloud.server.ManagementServer;
-import com.cloud.user.Account;
-import com.cloud.user.AccountManager;
-import com.cloud.user.User;
-import com.cloud.user.dao.AccountDao;
-import com.cloud.utils.DateUtil;
-import com.cloud.utils.NumbersUtil;
-import com.cloud.utils.PropertiesUtil;
-import com.cloud.utils.component.ComponentContext;
-import com.cloud.utils.component.ManagerBase;
-import com.cloud.utils.concurrency.NamedThreadFactory;
-import com.cloud.utils.db.DB;
-import com.cloud.utils.db.EntityManager;
-import com.cloud.utils.db.GlobalLock;
-import com.cloud.utils.db.Transaction;
-import com.cloud.utils.exception.CloudRuntimeException;
-import com.cloud.utils.exception.ExceptionUtil;
-import com.cloud.utils.mgmt.JmxUtil;
-
-@Component
-@Local(value={AsyncJobManager.class})
-public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager, ClusterManagerListener {
- public static final Logger s_logger = Logger.getLogger(AsyncJobManagerImpl.class.getName());
- private static final int ACQUIRE_GLOBAL_LOCK_TIMEOUT_FOR_COOPERATION = 3; // 3 seconds
-
- private static final int MAX_ONETIME_SCHEDULE_SIZE = 50;
- private static final int HEARTBEAT_INTERVAL = 2000;
- private static final int GC_INTERVAL = 10000; // 10 seconds
-
- @Inject private AsyncJobExecutorContext _context;
- @Inject private SyncQueueManager _queueMgr;
- @Inject private AccountManager _accountMgr;
- @Inject private AccountDao _accountDao;
- @Inject private AsyncJobDao _jobDao;
- @Inject private ConfigurationDao _configDao;
- @Inject private DomainDao _domainDao;
- private long _jobExpireSeconds = 86400; // 1 day
- private long _jobCancelThresholdSeconds = 3600; // 1 hour (for cancelling the jobs blocking other jobs)
-
- @Inject
- private EntityManager _entityMgr;
-
- @Inject private ApiDispatcher _dispatcher;
-
- private final ScheduledExecutorService _heartbeatScheduler =
- Executors.newScheduledThreadPool(1, new NamedThreadFactory("AsyncJobMgr-Heartbeat"));
- private ExecutorService _executor;
-
- @Override
- public AsyncJobExecutorContext getExecutorContext() {
- return _context;
- }
-
- @Override
- public AsyncJobVO getAsyncJob(long jobId) {
- return _jobDao.findById(jobId);
- }
-
- @Override
- public AsyncJobVO findInstancePendingAsyncJob(String instanceType, long instanceId) {
- return _jobDao.findInstancePendingAsyncJob(instanceType, instanceId);
- }
-
- @Override
- public List<AsyncJobVO> findInstancePendingAsyncJobs(ApiCommandJobType instanceType, Long accountId) {
- return _jobDao.findInstancePendingAsyncJobs(instanceType, accountId);
- }
-
- private void publishOnEventBus(AsyncJobVO job, String jobEvent) {
- EventBus eventBus = null;
- try {
- eventBus = ComponentContext.getComponent(EventBus.class);
- } catch(NoSuchBeanDefinitionException nbe) {
- return; // no provider is configured to provide events bus, so just return
- }
-
- // Get the event type from the cmdInfo json string
- String info = job.getCmdInfo();
- String cmdEventType;
- if ( info == null ) {
- cmdEventType = "unknown";
- } else {
- String marker = "\"cmdEventType\"";
- int begin = info.indexOf(marker);
- cmdEventType = info.substring(begin + marker.length() + 2, info.indexOf(",", begin) - 1);
- }
-
- // For some reason, the instanceType / instanceId are not abstract, which means we may get null values.
- org.apache.cloudstack.framework.events.Event event = new org.apache.cloudstack.framework.events.Event(
- ManagementServer.Name,
- EventCategory.ASYNC_JOB_CHANGE_EVENT.getName(),
- jobEvent,
- ( job.getInstanceType() != null ? job.getInstanceType().toString() : "unknown" ), null);
-
- User userJobOwner = _accountMgr.getUserIncludingRemoved(job.getUserId());
- Account jobOwner = _accountMgr.getAccount(userJobOwner.getAccountId());
-
- Map<String, String> eventDescription = new HashMap<String, String>();
- eventDescription.put("command", job.getCmd());
- eventDescription.put("user", userJobOwner.getUuid());
- eventDescription.put("account", jobOwner.getUuid());
- eventDescription.put("processStatus", "" + job.getProcessStatus());
- eventDescription.put("resultCode", "" + job.getResultCode());
- eventDescription.put("instanceUuid", ApiDBUtils.findJobInstanceUuid(job));
- eventDescription.put("instanceType", ( job.getInstanceType() != null ? job.getInstanceType().toString() : "unknown" ) );
- eventDescription.put("commandEventType", cmdEventType);
- eventDescription.put("jobId", job.getUuid());
-
- // If the event.accountinfo boolean value is set, get the human readable value for the username / domainname
- Map<String, String> configs = _configDao.getConfiguration("management-server", new HashMap<String, String>());
- if ( Boolean.valueOf(configs.get("event.accountinfo")) ) {
- DomainVO domain = _domainDao.findById(jobOwner.getDomainId());
- eventDescription.put("username", userJobOwner.getUsername());
- eventDescription.put("domainname", domain.getName());
- }
-
- event.setDescription(eventDescription);
-
- try {
- eventBus.publish(event);
- } catch (EventBusException evx) {
- String errMsg = "Failed to publish async job event on the the event bus.";
- s_logger.warn(errMsg, evx);
- throw new CloudRuntimeException(errMsg);
- }
- }
-
- @Override
- public long submitAsyncJob(AsyncJobVO job) {
- return submitAsyncJob(job, false);
- }
-
- @Override @DB
- public long submitAsyncJob(AsyncJobVO job, boolean scheduleJobExecutionInContext) {
- Transaction txt = Transaction.currentTxn();
- try {
- txt.start();
- job.setInitMsid(getMsid());
- _jobDao.persist(job);
- txt.commit();
-
- // no sync source originally
- job.setSyncSource(null);
- scheduleExecution(job, scheduleJobExecutionInContext);
- if(s_logger.isDebugEnabled()) {
- s_logger.debug("submit async job-" + job.getId() + " = [ " + job.getUuid() + " ], details: " + job.toString());
- }
- publishOnEventBus(job, "submit");
- return job.getId();
- } catch(Exception e) {
- txt.rollback();
- String errMsg = "Unable to schedule async job for command " + job.getCmd() + ", unexpected exception.";
- s_logger.warn(errMsg, e);
- throw new CloudRuntimeException(errMsg);
- }
- }
-
- @Override @DB
- public void completeAsyncJob(long jobId, int jobStatus, int resultCode, Object resultObject) {
- AsyncJobVO job = _jobDao.findById(jobId);
- String jobUuid = null;
- if (job != null) {
- jobUuid = job.getUuid();
- if(s_logger.isDebugEnabled()) {
- s_logger.debug("Complete async job-" + jobId + " = [ " + jobUuid + " ], jobStatus: " + jobStatus +
- ", resultCode: " + resultCode + ", result: " + resultObject);
- }
- } else {
- if(s_logger.isDebugEnabled()) {
- s_logger.debug("job-" + jobId + " no longer exists, we just log completion info here. " + jobStatus +
- ", resultCode: " + resultCode + ", result: " + resultObject);
- }
- return;
- }
-
- Transaction txt = Transaction.currentTxn();
- try {
- txt.start();
- job.setCompleteMsid(getMsid());
- job.setStatus(jobStatus);
- job.setResultCode(resultCode);
-
- publishOnEventBus(job, "complete"); // publish before the instance type and ID are wiped out
-
- // reset attached object
- job.setInstanceType(null);
- job.setInstanceId(null);
-
- if (resultObject != null) {
- job.setResult(ApiSerializerHelper.toSerializedStringOld(resultObject));
- }
-
- job.setLastUpdated(DateUtil.currentGMTTime());
- _jobDao.update(jobId, job);
- txt.commit();
- } catch(Exception e) {
- s_logger.error("Unexpected exception while completing async job-" + jobId + " = [ " + jobUuid + " ]", e);
- txt.rollback();
- }
- }
-
- @Override @DB
- public void updateAsyncJobStatus(long jobId, int processStatus, Object resultObject) {
- AsyncJobVO job = _jobDao.findById(jobId);
- String jobUuid = null;
- if (job != null) {
- jobUuid = job.getUuid();
- if(s_logger.isDebugEnabled()) {
- s_logger.debug("Update async-job progress, job-" + jobId + " = [ " + jobUuid + " ], processStatus: " + processStatus +
- ", result: " + resultObject);
- }
- } else {
- if(s_logger.isDebugEnabled()) {
- s_logger.debug("job-" + jobId + " no longer exists, we just log progress info here. progress status: " + processStatus);
- }
- return;
- }
-
- Transaction txt = Transaction.currentTxn();
- try {
- txt.start();
- job.setProcessStatus(processStatus);
- if(resultObject != null) {
- job.setResult(ApiSerializerHelper.toSerializedStringOld(resultObject));
- }
- job.setLastUpdated(DateUtil.currentGMTTime());
- _jobDao.update(jobId, job);
- publishOnEventBus(job, "update");
- txt.commit();
- } catch(Exception e) {
- s_logger.error("Unexpected exception while updating async job-" + jobId + " = [ " + jobUuid + " ] status: ", e);
- txt.rollback();
- }
- }
-
- @Override @DB
- public void updateAsyncJobAttachment(long jobId, String instanceType, Long instanceId) {
- AsyncJobVO job = _jobDao.findById(jobId);
- String jobUuid = null;
- if (job != null) {
- jobUuid = job.getUuid();
- if(s_logger.isDebugEnabled()) {
- s_logger.debug("Update async-job attachment, job-" + jobId + " = [ " + jobUuid + " ], instanceType: "
- + instanceType + ", instanceId: " + instanceId);
- }
- } else {
- if(s_logger.isDebugEnabled()) {
- s_logger.debug("job-" + jobId + " no longer exists, instanceType: " + instanceType + ", instanceId: "
- + instanceId);
- }
- return;
- }
-
- Transaction txt = Transaction.currentTxn();
- try {
- txt.start();
- //job.setInstanceType(instanceType);
- job.setInstanceId(instanceId);
- job.setLastUpdated(DateUtil.currentGMTTime());
- _jobDao.update(jobId, job);
-
- txt.commit();
- } catch(Exception e) {
- s_logger.error("Unexpected exception while updating async job-" + jobId + " = [ " + jobUuid + " ] attachment: ", e);
- txt.rollback();
- }
- }
-
- @Override
- public void syncAsyncJobExecution(AsyncJob job, String syncObjType, long syncObjId, long queueSizeLimit) {
- // This method is re-entrant. If an API developer wants to synchronized on an object, e.g. the router,
- // when executing business logic, they will call this method (actually a method in BaseAsyncCmd that calls this).
- // This method will get called every time their business logic executes. The first time it exectues for a job
- // there will be no sync source, but on subsequent execution there will be a sync souce. If this is the first
- // time the job executes we queue the job, otherwise we just return so that the business logic can execute.
- if (job.getSyncSource() != null) {
- return;
- }
-
- if(s_logger.isDebugEnabled()) {
- s_logger.debug("Sync job-" + job.getId() + " = [ " + job.getUuid() + " ] execution on object " + syncObjType + "." + syncObjId);
- }
-
- SyncQueueVO queue = null;
-
- // to deal with temporary DB exceptions like DB deadlock/Lock-wait time out cased rollbacks
- // we retry five times until we throw an exception
- Random random = new Random();
-
- for(int i = 0; i < 5; i++) {
- queue = _queueMgr.queue(syncObjType, syncObjId, SyncQueueItem.AsyncJobContentType, job.getId(), queueSizeLimit);
- if(queue != null) {
- break;
- }
-
- try {
- Thread.sleep(1000 + random.nextInt(5000));
- } catch (InterruptedException e) {
- }
- }
-
- if (queue == null) {
- throw new CloudRuntimeException("Unable to insert queue item into database, DB is full?");
- } else {
- throw new AsyncCommandQueued(queue, "job-" + job.getId() + " = [ " + job.getUuid() + " ] queued");
- }
- }
-
- @Override
- public AsyncJob queryAsyncJobResult(QueryAsyncJobResultCmd cmd) {
- Account caller = CallContext.current().getCallingAccount();
-
- AsyncJobVO job = _jobDao.findById(cmd.getId());
- if (job == null) {
- throw new InvalidParameterValueException("Unable to find a job by id " + cmd.getId());
- }
-
- User userJobOwner = _accountMgr.getUserIncludingRemoved(job.getUserId());
- Account jobOwner = _accountMgr.getAccount(userJobOwner.getAccountId());
-
- //check permissions
- if (caller.getType() == Account.ACCOUNT_TYPE_NORMAL) {
- //regular user can see only jobs he owns
- if (caller.getId() != jobOwner.getId()) {
- throw new PermissionDeniedException("Account " + caller + " is not authorized to see job-" + job.getId() + " = [ " + job.getUuid() + " ]");
- }
- } else if (caller.getType() == Account.ACCOUNT_TYPE_DOMAIN_ADMIN) {
- _accountMgr.checkAccess(caller, null, true, jobOwner);
- }
-
- //poll the job
- queryAsyncJobResult(cmd.getId());
- return _jobDao.findById(cmd.getId());
- }
-
- @Override @DB
- public AsyncJobResult queryAsyncJobResult(long jobId) {
- AsyncJobVO job = _jobDao.findById(jobId);
- String jobUuid = null;
- if (job != null) {
- jobUuid = job.getUuid();
- if(s_logger.isTraceEnabled()) {
- s_logger.trace("Query async-job status, job-" + jobId + " = [ " + jobUuid + " ]");
- }
- } else {
- if(s_logger.isDebugEnabled()) {
- s_logger.debug("Async job-" + jobId + " does not exist, invalid job id?");
- }
- }
-
- Transaction txt = Transaction.currentTxn();
- AsyncJobResult jobResult = new AsyncJobResult(jobId);
-
- try {
- txt.start();
- if(job != null) {
- jobResult.setCmdOriginator(job.getCmdOriginator());
- jobResult.setJobStatus(job.getStatus());
- jobResult.setProcessStatus(job.getProcessStatus());
- jobResult.setResult(job.getResult());
- jobResult.setResultCode(job.getResultCode());
- jobResult.setUuid(job.getUuid());
-
- if(job.getStatus() == AsyncJobResult.STATUS_SUCCEEDED ||
- job.getStatus() == AsyncJobResult.STATUS_FAILED) {
-
- if(s_logger.isDebugEnabled()) {
- s_logger.debug("Async job-" + jobId + " = [ " + jobUuid + " ] completed");
- }
- } else {
- job.setLastPolled(DateUtil.currentGMTTime());
- _jobDao.update(jobId, job);
- }
- } else {
- jobResult.setJobStatus(AsyncJobResult.STATUS_FAILED);
- jobResult.setResult("job-" + jobId + " does not exist");
- }
- txt.commit();
- } catch(Exception e) {
- if (jobUuid == null) {
- s_logger.error("Unexpected exception while querying async job-" + jobId + " status: ", e);
- } else {
- s_logger.error("Unexpected exception while querying async job-" + jobId + " = [ " + jobUuid + " ] status: ", e);
- }
-
- jobResult.setJobStatus(AsyncJobResult.STATUS_FAILED);
- jobResult.setResult("Exception: " + e.toString());
- txt.rollback();
- }
-
- if(s_logger.isTraceEnabled()) {
- s_logger.trace("Job status: " + jobResult.toString());
- }
-
- return jobResult;
- }
-
- private void scheduleExecution(final AsyncJobVO job) {
- scheduleExecution(job, false);
- }
-
- private void scheduleExecution(final AsyncJobVO job, boolean executeInContext) {
- Runnable runnable = getExecutorRunnable(this, job);
- if (executeInContext) {
- runnable.run();
- } else {
- _executor.submit(runnable);
- }
- }
-
- private Runnable getExecutorRunnable(final AsyncJobManager mgr, final AsyncJobVO job) {
- return new Runnable() {
- @Override
- public void run() {
- try {
- long jobId = 0;
-
- try {
- JmxUtil.registerMBean("AsyncJobManager", "Active Job " + job.getId(), new AsyncJobMBeanImpl(job));
- } catch(Exception e) {
- s_logger.warn("Unable to register active job [ " + job.getId() + " ] = [ " + job.getUuid() + " ] to JMX monitoring due to exception " + ExceptionUtil.toString(e));
- }
-
- BaseAsyncCmd cmdObj = null;
- Transaction txn = Transaction.open(Transaction.CLOUD_DB);
- try {
- jobId = job.getId();
- String jobUuid = job.getUuid();
- NDC.push("job-" + jobId + " = [ " + jobUuid + " ]");
-
- if(s_logger.isDebugEnabled()) {
- s_logger.debug("Executing " + job.getCmd() + " for job-" + jobId + " = [ " + jobUuid + " ]");
- }
-
- Class<?> cmdClass = Class.forName(job.getCmd());
- cmdObj = (BaseAsyncCmd)cmdClass.newInstance();
- cmdObj = ComponentContext.inject(cmdObj);
- cmdObj.configure();
- cmdObj.setJob(job);
-
- Type mapType = new TypeToken<Map<String, String>>() {}.getType();
- Gson gson = ApiGsonHelper.getBuilder().create();
- Map<String, String> params = gson.fromJson(job.getCmdInfo(), mapType);
-
- // whenever we deserialize, the UserContext needs to be updated
- String userIdStr = params.get("ctxUserId");
- String acctIdStr = params.get("ctxAccountId");
- Long userId = null;
- Account accountObject = null;
- User user = null;
-
- if (userIdStr != null) {
- userId = Long.parseLong(userIdStr);
- user = _entityMgr.findById(User.class, userId);
- }
-
- if (acctIdStr != null) {
- accountObject = _accountDao.findById(Long.parseLong(acctIdStr));
- }
-
-
- CallContext.register(user, accountObject);
- try {
- // dispatch could ultimately queue the job
- _dispatcher.dispatch(cmdObj, params);
-
- // serialize this to the async job table
- completeAsyncJob(jobId, AsyncJobResult.STATUS_SUCCEEDED, 0, cmdObj.getResponseObject());
- } finally {
- CallContext.unregister();
- }
-
- // commands might need to be queued as part of synchronization here, so they just have to be re-dispatched from the queue mechanism...
- if (job.getSyncSource() != null) {
- _queueMgr.purgeItem(job.getSyncSource().getId());
- checkQueue(job.getSyncSource().getQueueId());
- }
-
- if (s_logger.isDebugEnabled()) {
- s_logger.debug("Done executing " + job.getCmd() + " for job-" + job.getId() + " = [ " + jobUuid + " ]");
- }
-
- } catch(Throwable e) {
- if (e instanceof AsyncCommandQueued) {
- if (s_logger.isDebugEnabled()) {
- s_logger.debug("job " + job.getCmd() + " for job-" + jobId + " = [ " + job.getUuid() + " ] was queued, processing the queue.");
- }
- checkQueue(((AsyncCommandQueued)e).getQueue().getId());
- } else {
- String errorMsg = null;
- int errorCode = ApiErrorCode.INTERNAL_ERROR.getHttpCode();
- if (!(e instanceof ServerApiException)) {
- s_logger.error("Unexpected exception while executing " + job.getCmd(), e);
- errorMsg = e.getMessage();
- } else {
- ServerApiException sApiEx = (ServerApiException)e;
- errorMsg = sApiEx.getDescription();
- errorCode = sApiEx.getErrorCode().getHttpCode();
- }
-
- ExceptionResponse response = new ExceptionResponse();
- response.setErrorCode(errorCode);
- response.setErrorText(errorMsg);
- response.setResponseName((cmdObj == null) ? "unknowncommandresponse" : cmdObj.getCommandName());
-
- // FIXME: setting resultCode to ApiErrorCode.INTERNAL_ERROR is not right, usually executors have their exception handling
- // and we need to preserve that as much as possible here
- completeAsyncJob(jobId, AsyncJobResult.STATUS_FAILED, ApiErrorCode.INTERNAL_ERROR.getHttpCode(), response);
-
- // need to clean up any queue that happened as part of the dispatching and move on to the next item in the queue
- try {
- if (job.getSyncSource() != null) {
- _queueMgr.purgeItem(job.getSyncSource().getId());
- checkQueue(job.getSyncSource().getQueueId());
- }
- } catch(Throwable ex) {
- s_logger.fatal("Exception on exception, log it for record", ex);
- }
- }
- } finally {
-
- try {
- JmxUtil.unregisterMBean("AsyncJobManager", "Active Job " + job.getId());
- } catch(Exception e) {
- s_logger.warn("Unable to unregister active job [ " + job.getId() + " ] = [ " + job.getUuid() + " ] from JMX monitoring");
- }
-
- txn.close();
- NDC.pop();
- }
- } catch (Throwable th) {
- try {
- s_logger.error("Caught: " + th);
- } catch (Throwable th2) {
- }
- }
- }
- };
- }
-
- private void executeQueueItem(SyncQueueItemVO item, boolean fromPreviousSession) {
- long jobId = item.getContentId();
- AsyncJobVO job = _jobDao.findById(item.getContentId());
- if (job != null) {
- String jobUuid = job.getUuid();
- if(s_logger.isDebugEnabled()) {
- s_logger.debug("Schedule queued job-" + jobId + " = [ " + jobUuid + " ]");
- }
-
- job.setFromPreviousSession(fromPreviousSession);
- job.setSyncSource(item);
-
- job.setCompleteMsid(getMsid());
- _jobDao.update(job.getId(), job);
-
- try {
- scheduleExecution(job);
- } catch(RejectedExecutionException e) {
- s_logger.warn("Execution for job-" + jobId + " = [ " + jobUuid + " ] is rejected, return it to the queue for next turn");
- _queueMgr.returnItem(item.getId());
- }
-
- } else {
- if(s_logger.isDebugEnabled()) {
- s_logger.debug("Unable to find related job for queue item: " + item.toString());
- }
-
- _queueMgr.purgeItem(item.getId());
- }
- }
-
- @Override
- public void releaseSyncSource(AsyncJobExecutor executor) {
- if(executor.getSyncSource() != null) {
- if(s_logger.isDebugEnabled()) {
- s_logger.debug("Release sync source for job-" + executor.getJob().getId() + " = [ " + executor.getJob().getUuid() + " ] sync source: "
- + executor.getSyncSource().getContentType() + "-"
- + executor.getSyncSource().getContentId());
- }
-
- _queueMgr.purgeItem(executor.getSyncSource().getId());
- checkQueue(executor.getSyncSource().getQueueId());
- }
- }
-
- private void checkQueue(long queueId) {
- while(true) {
- try {
- SyncQueueItemVO item = _queueMgr.dequeueFromOne(queueId, getMsid());
- if(item != null) {
- if(s_logger.isDebugEnabled()) {
- s_logger.debug("Executing sync queue item: " + item.toString());
- }
-
- executeQueueItem(item, false);
- } else {
- break;
- }
- } catch(Throwable e) {
- s_logger.error("Unexpected exception when kicking sync queue-" + queueId, e);
- break;
- }
- }
- }
-
- private Runnable getHeartbeatTask() {
- return new Runnable() {
- @Override
- public void run() {
- try {
- List<SyncQueueItemVO> l = _queueMgr.dequeueFromAny(getMsid(), MAX_ONETIME_SCHEDULE_SIZE);
- if(l != null && l.size() > 0) {
- for(SyncQueueItemVO item: l) {
- if(s_logger.isDebugEnabled()) {
- s_logger.debug("Execute sync-queue item: " + item.toString());
- }
- executeQueueItem(item, false);
- }
- }
- } catch(Throwable e) {
- s_logger.error("Unexpected exception when trying to execute queue item, ", e);
- }
- }
- };
- }
-
- @DB
- private Runnable getGCTask() {
- return new Runnable() {
- @Override
- public void run() {
- GlobalLock scanLock = GlobalLock.getInternLock("AsyncJobManagerGC");
- try {
- if(scanLock.lock(ACQUIRE_GLOBAL_LOCK_TIMEOUT_FOR_COOPERATION)) {
- try {
- reallyRun();
- } finally {
- scanLock.unlock();
- }
- }
- } finally {
- scanLock.releaseRef();
- }
- }
-
- public void reallyRun() {
- try {
- s_logger.trace("Begin cleanup expired async-jobs");
-
- Date cutTime = new Date(DateUtil.currentGMTTime().getTime() - _jobExpireSeconds*1000);
-
- // limit to 100 jobs per turn, this gives cleanup throughput as 600 jobs per minute
- // hopefully this will be fast enough to balance potential growth of job table
- //1) Expire unfinished jobs that weren't processed yet
- List<AsyncJobVO> l = _jobDao.getExpiredUnfinishedJobs(cutTime, 100);
- for(AsyncJobVO job : l) {
- s_logger.trace("Expunging unfinished job " + job);
- expungeAsyncJob(job);
- }
-
- //2) Expunge finished jobs
- List<AsyncJobVO> completedJobs = _jobDao.getExpiredCompletedJobs(cutTime, 100);
- for(AsyncJobVO job : completedJobs) {
- s_logger.trace("Expunging completed job " + job);
- expungeAsyncJob(job);
- }
-
- // forcefully cancel blocking queue items if they've been staying there for too long
- List<SyncQueueItemVO> blockItems = _queueMgr.getBlockedQueueItems(_jobCancelThresholdSeconds*1000, false);
- if(blockItems != null && blockItems.size() > 0) {
- for(SyncQueueItemVO item : blockItems) {
- if(item.getContentType().equalsIgnoreCase(SyncQueueItem.AsyncJobContentType)) {
- completeAsyncJob(item.getContentId(), AsyncJobResult.STATUS_FAILED, 0,
- getResetResultResponse("Job is cancelled as it has been blocking others for too long"));
- }
-
- // purge the item and resume queue processing
- _queueMgr.purgeItem(item.getId());
- }
- }
-
- s_logger.trace("End cleanup expired async-jobs");
- } catch(Throwable e) {
- s_logger.error("Unexpected exception when trying to execute queue item, ", e);
- }
- }
-
-
- };
- }
-
- @DB
- protected void expungeAsyncJob(AsyncJobVO job) {
- Transaction txn = Transaction.currentTxn();
- txn.start();
- _jobDao.expunge(job.getId());
- //purge corresponding sync queue item
- _queueMgr.purgeAsyncJobQueueItemId(job.getId());
- txn.commit();
- }
-
- private long getMsid() {
- return ManagementServerNode.getManagementServerId();
- }
-
- private void cleanupPendingJobs(List<SyncQueueItemVO> l) {
- if(l != null && l.size() > 0) {
- for(SyncQueueItemVO item: l) {
- if(s_logger.isInfoEnabled()) {
- s_logger.info("Discard left-over queue item: " + item.toString());
- }
-
- String contentType = item.getContentType();
- if(contentType != null && contentType.equalsIgnoreCase(SyncQueueItem.AsyncJobContentType)) {
- Long jobId = item.getContentId();
- if(jobId != null) {
- s_logger.warn("Mark job as failed as its correspoding queue-item has been discarded. job id: " + jobId);
- completeAsyncJob(jobId, AsyncJobResult.STATUS_FAILED, 0, getResetResultResponse("Execution was cancelled because of server shutdown"));
- }
- }
- _queueMgr.purgeItem(item.getId());
- }
- }
- }
-
- @Override
- public boolean configure(String name, Map<String, Object> params) throws ConfigurationException {
- int expireMinutes = NumbersUtil.parseInt(
- _configDao.getValue(Config.JobExpireMinutes.key()), 24*60);
- _jobExpireSeconds = (long)expireMinutes*60;
-
- _jobCancelThresholdSeconds = NumbersUtil.parseInt(
- _configDao.getValue(Config.JobCancelThresholdMinutes.key()), 60);
- _jobCancelThresholdSeconds *= 60;
-
- try {
- final File dbPropsFile = PropertiesUtil.findConfigFile("db.properties");
- final Properties dbProps = new Properties();
- dbProps.load(new FileInputStream(dbPropsFile));
-
- final int cloudMaxActive = Integer.parseInt(dbProps.getProperty("db.cloud.maxActive"));
-
- int poolSize = (cloudMaxActive * 2) / 3;
-
- s_logger.info("Start AsyncJobManager thread pool in size " + poolSize);
- _executor = Executors.newFixedThreadPool(poolSize, new NamedThreadFactory("Job-Executor"));
- } catch (final Exception e) {
- throw new ConfigurationException("Unable to load db.properties to configure AsyncJobManagerImpl");
- }
-
- return true;
- }
-
- @Override
- public void onManagementNodeJoined(List<? extends ManagementServerHost> nodeList, long selfNodeId) {
- }
-
- @Override
- public void onManagementNodeLeft(List<? extends ManagementServerHost> nodeList, long selfNodeId) {
- for (ManagementServerHost msHost : nodeList) {
- Transaction txn = Transaction.open(Transaction.CLOUD_DB);
- try {
- txn.start();
- List<SyncQueueItemVO> items = _queueMgr.getActiveQueueItems(msHost.getId(), true);
- cleanupPendingJobs(items);
- _jobDao.resetJobProcess(msHost.getId(), ApiErrorCode.INTERNAL_ERROR.getHttpCode(), getSerializedErrorMessage("job cancelled because of management server restart"));
- txn.commit();
- } catch(Throwable e) {
- s_logger.warn("Unexpected exception ", e);
- txn.rollback();
- } finally {
- txn.close();
- }
- }
- }
-
- @Override
- public void onManagementNodeIsolated() {
- }
-
- @Override
- public boolean start() {
- try {
- List<SyncQueueItemVO> l = _queueMgr.getActiveQueueItems(getMsid(), false);
- cleanupPendingJobs(l);
- _jobDao.resetJobProcess(getMsid(), ApiErrorCode.INTERNAL_ERROR.getHttpCode(), getSerializedErrorMessage("job cancelled because of management server restart"));
- } catch(Throwable e) {
- s_logger.error("Unexpected exception " + e.getMessage(), e);
- }
-
- _heartbeatScheduler.scheduleAtFixedRate(getHeartbeatTask(), HEARTBEAT_INTERVAL,
- HEARTBEAT_INTERVAL, TimeUnit.MILLISECONDS);
- _heartbeatScheduler.scheduleAtFixedRate(getGCTask(), GC_INTERVAL,
- GC_INTERVAL, TimeUnit.MILLISECONDS);
-
- return true;
- }
-
- private static ExceptionResponse getResetResultResponse(String errorMessage) {
- ExceptionResponse resultObject = new ExceptionResponse();
- resultObject.setErrorCode(ApiErrorCode.INTERNAL_ERROR.getHttpCode());
- resultObject.setErrorText(errorMessage);
- return resultObject;
- }
-
- private static String getSerializedErrorMessage(String errorMessage) {
- return ApiSerializerHelper.toSerializedStringOld(getResetResultResponse(errorMessage));
- }
-
- @Override
- public boolean stop() {
- _heartbeatScheduler.shutdown();
- _executor.shutdown();
- return true;
- }
-}
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/8930cfa9/server/src/com/cloud/async/AsyncJobResult.java
----------------------------------------------------------------------
diff --git a/server/src/com/cloud/async/AsyncJobResult.java b/server/src/com/cloud/async/AsyncJobResult.java
index cf343ea..d71e64b 100644
--- a/server/src/com/cloud/async/AsyncJobResult.java
+++ b/server/src/com/cloud/async/AsyncJobResult.java
@@ -16,16 +16,14 @@
// under the License.
package com.cloud.async;
+import org.apache.cloudstack.jobs.JobInfo;
+
import com.cloud.api.ApiSerializerHelper;
public class AsyncJobResult {
- public static final int STATUS_IN_PROGRESS = 0;
- public static final int STATUS_SUCCEEDED = 1;
- public static final int STATUS_FAILED = 2;
- private String cmdOriginator;
private long jobId;
- private int jobStatus;
+ private JobInfo.Status jobStatus;
private int processStatus;
private int resultCode;
private String result;
@@ -33,20 +31,12 @@ public class AsyncJobResult {
public AsyncJobResult(long jobId) {
this.jobId = jobId;
- jobStatus = STATUS_IN_PROGRESS;
+ jobStatus = JobInfo.Status.IN_PROGRESS;
processStatus = 0;
resultCode = 0;
result = "";
}
- public String getCmdOriginator() {
- return cmdOriginator;
- }
-
- public void setCmdOriginator(String cmdOriginator) {
- this.cmdOriginator = cmdOriginator;
- }
-
public long getJobId() {
return jobId;
}
@@ -56,18 +46,18 @@ public class AsyncJobResult {
}
public String getUuid() {
- return this.uuid;
+ return uuid;
}
public void setUuid(String uuid) {
this.uuid = uuid;
}
- public int getJobStatus() {
+ public JobInfo.Status getJobStatus() {
return jobStatus;
}
- public void setJobStatus(int jobStatus) {
+ public void setJobStatus(JobInfo.Status jobStatus) {
this.jobStatus = jobStatus;
}
@@ -100,14 +90,14 @@ public class AsyncJobResult {
}
public void setResultObject(Object result) {
- this.result = ApiSerializerHelper.toSerializedStringOld(result);
+ this.result = ApiSerializerHelper.toSerializedString(result);
}
@Override
public String toString() {
StringBuffer sb = new StringBuffer();
sb.append("AsyncJobResult {jobId:").append(getJobId());
- sb.append(", jobStatus: ").append(getJobStatus());
+ sb.append(", jobStatus: ").append(getJobStatus().ordinal());
sb.append(", processStatus: ").append(getProcessStatus());
sb.append(", resultCode: ").append(getResultCode());
sb.append(", result: ").append(result);
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/8930cfa9/server/src/com/cloud/async/AsyncJobVO.java
----------------------------------------------------------------------
diff --git a/server/src/com/cloud/async/AsyncJobVO.java b/server/src/com/cloud/async/AsyncJobVO.java
deleted file mode 100644
index 41eccb4..0000000
--- a/server/src/com/cloud/async/AsyncJobVO.java
+++ /dev/null
@@ -1,403 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-package com.cloud.async;
-
-import java.util.Date;
-import java.util.UUID;
-
-import javax.persistence.Column;
-import javax.persistence.Entity;
-import javax.persistence.EnumType;
-import javax.persistence.Enumerated;
-import javax.persistence.GeneratedValue;
-import javax.persistence.GenerationType;
-import javax.persistence.Id;
-import javax.persistence.Table;
-import javax.persistence.Temporal;
-import javax.persistence.TemporalType;
-import javax.persistence.Transient;
-
-import com.cloud.utils.db.GenericDao;
-
-import org.apache.cloudstack.api.ApiCommandJobType;
-import org.apache.cloudstack.api.InternalIdentity;
-
-@Entity
-@Table(name="async_job")
-public class AsyncJobVO implements AsyncJob {
- public static final int CALLBACK_POLLING = 0;
- public static final int CALLBACK_EMAIL = 1;
-
- @Id
- @GeneratedValue(strategy=GenerationType.IDENTITY)
- @Column(name="id")
- private Long id = null;
-
- @Column(name="user_id")
- private long userId;
-
- @Column(name="account_id")
- private long accountId;
-
- @Column(name="session_key")
- private String sessionKey;
-
- @Column(name="job_cmd")
- private String cmd;
-
- @Column(name="job_cmd_originator")
- private String cmdOriginator;
-
- @Column(name="job_cmd_ver")
- private int cmdVersion;
-
- @Column(name="job_cmd_info", length=65535)
- private String cmdInfo;
-
- @Column(name="callback_type")
- private int callbackType;
-
- @Column(name="callback_address")
- private String callbackAddress;
-
- @Column(name="job_status")
- private int status;
-
- @Column(name="job_process_status")
- private int processStatus;
-
- @Column(name="job_result_code")
- private int resultCode;
-
- @Column(name="job_result", length=65535)
- private String result;
-
- @Enumerated(value=EnumType.STRING)
- @Column(name="instance_type", length=64)
- private ApiCommandJobType instanceType;
-
- @Column(name="instance_id", length=64)
- private Long instanceId;
-
- @Column(name="job_init_msid")
- private Long initMsid;
-
- @Column(name="job_complete_msid")
- private Long completeMsid;
-
- @Column(name=GenericDao.CREATED_COLUMN)
- private Date created;
-
- @Column(name="last_updated")
- @Temporal(TemporalType.TIMESTAMP)
- private Date lastUpdated;
-
- @Column(name="last_polled")
- @Temporal(TemporalType.TIMESTAMP)
- private Date lastPolled;
-
- @Column(name=GenericDao.REMOVED_COLUMN)
- private Date removed;
-
- @Column(name="uuid")
- private String uuid;
-
- @Transient
- private SyncQueueItemVO syncSource = null;
-
- @Transient
- private boolean fromPreviousSession = false;
-
-
- public AsyncJobVO() {
- this.uuid = UUID.randomUUID().toString();
- }
-
- public AsyncJobVO(long userId, long accountId, String cmd, String cmdInfo, Long instanceId, ApiCommandJobType instanceType) {
- this.userId = userId;
- this.accountId = accountId;
- this.cmd = cmd;
- this.cmdInfo = cmdInfo;
- this.callbackType = CALLBACK_POLLING;
- this.uuid = UUID.randomUUID().toString();
- this.instanceId = instanceId;
- this.instanceType = instanceType;
- }
-
- public AsyncJobVO(long userId, long accountId, String cmd, String cmdInfo,
- int callbackType, String callbackAddress, Long instanceId, ApiCommandJobType instanceType) {
-
- this(userId, accountId, cmd, cmdInfo, instanceId, instanceType);
- this.callbackType = callbackType;
- this.callbackAddress = callbackAddress;
- this.uuid = UUID.randomUUID().toString();
- }
-
-
- @Override
- public long getId() {
- return id;
- }
-
- public void setId(Long id) {
- this.id = id;
- }
-
- @Override
- public long getUserId() {
- return userId;
- }
-
- public void setUserId(long userId) {
- this.userId = userId;
- }
-
- @Override
- public long getAccountId() {
- return accountId;
- }
-
- public void setAccountId(long accountId) {
- this.accountId = accountId;
- }
-
- @Override
- public String getCmd() {
- return cmd;
- }
-
- public void setCmd(String cmd) {
- this.cmd = cmd;
- }
-
- @Override
- public int getCmdVersion() {
- return cmdVersion;
- }
-
- public void setCmdVersion(int version) {
- cmdVersion = version;
- }
-
- @Override
- public String getCmdInfo() {
- return cmdInfo;
- }
-
- public void setCmdInfo(String cmdInfo) {
- this.cmdInfo = cmdInfo;
- }
-
- @Override
- public int getCallbackType() {
- return callbackType;
- }
-
- public void setCallbackType(int callbackType) {
- this.callbackType = callbackType;
- }
-
- @Override
- public String getCallbackAddress() {
- return callbackAddress;
- }
-
- public void setCallbackAddress(String callbackAddress) {
- this.callbackAddress = callbackAddress;
- }
-
- @Override
- public int getStatus() {
- return status;
- }
-
- public void setStatus(int status) {
- this.status = status;
- }
-
- @Override
- public int getProcessStatus() {
- return processStatus;
- }
-
- public void setProcessStatus(int status) {
- processStatus = status;
- }
-
- @Override
- public int getResultCode() {
- return resultCode;
- }
-
- public void setResultCode(int resultCode) {
- this.resultCode = resultCode;
- }
-
- @Override
- public String getResult() {
- return result;
- }
-
- public void setResult(String result) {
- this.result = result;
- }
-
- @Override
- public Long getInitMsid() {
- return initMsid;
- }
-
- public void setInitMsid(Long initMsid) {
- this.initMsid = initMsid;
- }
-
- @Override
- public Long getCompleteMsid() {
- return completeMsid;
- }
-
- public void setCompleteMsid(Long completeMsid) {
- this.completeMsid = completeMsid;
- }
-
- @Override
- public Date getCreated() {
- return created;
- }
-
- public void setCreated(Date created) {
- this.created = created;
- }
-
- @Override
- public Date getLastUpdated() {
- return lastUpdated;
- }
-
- public void setLastUpdated(Date lastUpdated) {
- this.lastUpdated = lastUpdated;
- }
-
- @Override
- public Date getLastPolled() {
- return lastPolled;
- }
-
- public void setLastPolled(Date lastPolled) {
- this.lastPolled = lastPolled;
- }
-
- @Override
- public Date getRemoved() {
- return removed;
- }
-
- public void setRemoved(Date removed) {
- this.removed = removed;
- }
-
- @Override
- public ApiCommandJobType getInstanceType() {
- return instanceType;
- }
-
- public void setInstanceType(ApiCommandJobType instanceType) {
- this.instanceType = instanceType;
- }
-
- @Override
- public Long getInstanceId() {
- return instanceId;
- }
-
- public void setInstanceId(Long instanceId) {
- this.instanceId = instanceId;
- }
-
- @Override
- public String getSessionKey() {
- return sessionKey;
- }
-
- public void setSessionKey(String sessionKey) {
- this.sessionKey = sessionKey;
- }
-
- @Override
- public String getCmdOriginator() {
- return cmdOriginator;
- }
-
- public void setCmdOriginator(String cmdOriginator) {
- this.cmdOriginator = cmdOriginator;
- }
-
- @Override
- public SyncQueueItemVO getSyncSource() {
- return syncSource;
- }
-
- public void setSyncSource(SyncQueueItemVO syncSource) {
- this.syncSource = syncSource;
- }
-
- @Override
- public boolean isFromPreviousSession() {
- return fromPreviousSession;
- }
-
- public void setFromPreviousSession(boolean fromPreviousSession) {
- this.fromPreviousSession = fromPreviousSession;
- }
-
- @Override
- public String getUuid() {
- return this.uuid;
- }
-
- public void setUuid(String uuid) {
- this.uuid = uuid;
- }
-
- @Override
- public String toString() {
- StringBuffer sb = new StringBuffer();
- sb.append("AsyncJobVO {id:").append(getId());
- sb.append(", userId: ").append(getUserId());
- sb.append(", accountId: ").append(getAccountId());
- sb.append(", sessionKey: ").append(getSessionKey());
- sb.append(", instanceType: ").append(getInstanceType());
- sb.append(", instanceId: ").append(getInstanceId());
- sb.append(", cmd: ").append(getCmd());
- sb.append(", cmdOriginator: ").append(getCmdOriginator());
- sb.append(", cmdInfo: ").append(getCmdInfo());
- sb.append(", cmdVersion: ").append(getCmdVersion());
- sb.append(", callbackType: ").append(getCallbackType());
- sb.append(", callbackAddress: ").append(getCallbackAddress());
- sb.append(", status: ").append(getStatus());
- sb.append(", processStatus: ").append(getProcessStatus());
- sb.append(", resultCode: ").append(getResultCode());
- sb.append(", result: ").append(getResult());
- sb.append(", initMsid: ").append(getInitMsid());
- sb.append(", completeMsid: ").append(getCompleteMsid());
- sb.append(", lastUpdated: ").append(getLastUpdated());
- sb.append(", lastPolled: ").append(getLastPolled());
- sb.append(", created: ").append(getCreated());
- sb.append("}");
- return sb.toString();
- }
-}
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/8930cfa9/server/src/com/cloud/async/BaseAsyncJobExecutor.java
----------------------------------------------------------------------
diff --git a/server/src/com/cloud/async/BaseAsyncJobExecutor.java b/server/src/com/cloud/async/BaseAsyncJobExecutor.java
deleted file mode 100644
index 122b34b..0000000
--- a/server/src/com/cloud/async/BaseAsyncJobExecutor.java
+++ /dev/null
@@ -1,69 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-package com.cloud.async;
-
-
-public abstract class BaseAsyncJobExecutor implements AsyncJobExecutor {
- private SyncQueueItemVO _syncSource;
- private AsyncJobVO _job;
- private boolean _fromPreviousSession;
- private AsyncJobManager _asyncJobMgr;
-
- private static ThreadLocal<AsyncJobExecutor> s_currentExector = new ThreadLocal<AsyncJobExecutor>();
-
- public AsyncJobManager getAsyncJobMgr() {
- return _asyncJobMgr;
- }
-
- public void setAsyncJobMgr(AsyncJobManager asyncMgr) {
- _asyncJobMgr = asyncMgr;
- }
-
- public SyncQueueItemVO getSyncSource() {
- return _syncSource;
- }
-
- public void setSyncSource(SyncQueueItemVO syncSource) {
- _syncSource = syncSource;
- }
-
- public AsyncJobVO getJob() {
- return _job;
- }
-
- public void setJob(AsyncJobVO job) {
- _job = job;
- }
-
- public void setFromPreviousSession(boolean value) {
- _fromPreviousSession = value;
- }
-
- public boolean isFromPreviousSession() {
- return _fromPreviousSession;
- }
-
- public abstract boolean execute();
-
- public static AsyncJobExecutor getCurrentExecutor() {
- return s_currentExector.get();
- }
-
- public static void setCurrentExecutor(AsyncJobExecutor currentExecutor) {
- s_currentExector.set(currentExecutor);
- }
-}
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/8930cfa9/server/src/com/cloud/async/SyncQueueItemVO.java
----------------------------------------------------------------------
diff --git a/server/src/com/cloud/async/SyncQueueItemVO.java b/server/src/com/cloud/async/SyncQueueItemVO.java
deleted file mode 100644
index b0546a7..0000000
--- a/server/src/com/cloud/async/SyncQueueItemVO.java
+++ /dev/null
@@ -1,141 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-package com.cloud.async;
-
-import org.apache.cloudstack.api.InternalIdentity;
-
-import java.util.Date;
-
-import javax.persistence.Column;
-import javax.persistence.Entity;
-import javax.persistence.GeneratedValue;
-import javax.persistence.GenerationType;
-import javax.persistence.Id;
-import javax.persistence.Table;
-import javax.persistence.Temporal;
-import javax.persistence.TemporalType;
-
-@Entity
-@Table(name="sync_queue_item")
-public class SyncQueueItemVO implements SyncQueueItem, InternalIdentity {
-
- @Id
- @GeneratedValue(strategy=GenerationType.IDENTITY)
- @Column(name="id")
- private Long id = null;
-
- @Column(name="queue_id")
- private Long queueId;
-
- @Column(name="content_type")
- private String contentType;
-
- @Column(name="content_id")
- private Long contentId;
-
- @Column(name="queue_proc_msid")
- private Long lastProcessMsid;
-
- @Column(name="queue_proc_number")
- private Long lastProcessNumber;
-
- @Column(name="queue_proc_time")
- @Temporal(TemporalType.TIMESTAMP)
- private Date lastProcessTime;
-
- @Column(name="created")
- private Date created;
-
- public long getId() {
- return id;
- }
-
- public void setId(Long id) {
- this.id = id;
- }
-
- public Long getQueueId() {
- return queueId;
- }
-
- public void setQueueId(Long queueId) {
- this.queueId = queueId;
- }
-
- @Override
- public String getContentType() {
- return contentType;
- }
-
- public void setContentType(String contentType) {
- this.contentType = contentType;
- }
-
- @Override
- public Long getContentId() {
- return contentId;
- }
-
- public void setContentId(Long contentId) {
- this.contentId = contentId;
- }
-
- public Long getLastProcessMsid() {
- return lastProcessMsid;
- }
-
- public void setLastProcessMsid(Long lastProcessMsid) {
- this.lastProcessMsid = lastProcessMsid;
- }
-
- public Long getLastProcessNumber() {
- return lastProcessNumber;
- }
-
- public void setLastProcessNumber(Long lastProcessNumber) {
- this.lastProcessNumber = lastProcessNumber;
- }
-
- public Date getCreated() {
- return created;
- }
-
- public void setCreated(Date created) {
- this.created = created;
- }
-
- public String toString() {
- StringBuffer sb = new StringBuffer();
- sb.append("SyncQueueItemVO {id:").append(getId()).append(", queueId: ").append(getQueueId());
- sb.append(", contentType: ").append(getContentType());
- sb.append(", contentId: ").append(getContentId());
- sb.append(", lastProcessMsid: ").append(getLastProcessMsid());
- sb.append(", lastprocessNumber: ").append(getLastProcessNumber());
- sb.append(", lastProcessTime: ").append(getLastProcessTime());
- sb.append(", created: ").append(getCreated());
- sb.append("}");
- return sb.toString();
- }
-
- public Date getLastProcessTime() {
- return lastProcessTime;
- }
-
- public void setLastProcessTime(Date lastProcessTime) {
- this.lastProcessTime = lastProcessTime;
- }
-}
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/8930cfa9/server/src/com/cloud/async/SyncQueueManager.java
----------------------------------------------------------------------
diff --git a/server/src/com/cloud/async/SyncQueueManager.java b/server/src/com/cloud/async/SyncQueueManager.java
deleted file mode 100644
index 2641a10..0000000
--- a/server/src/com/cloud/async/SyncQueueManager.java
+++ /dev/null
@@ -1,35 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-package com.cloud.async;
-
-import java.util.List;
-
-import com.cloud.utils.component.Manager;
-
-
-public interface SyncQueueManager extends Manager {
- public SyncQueueVO queue(String syncObjType, long syncObjId, String itemType, long itemId, long queueSizeLimit);
- public SyncQueueItemVO dequeueFromOne(long queueId, Long msid);
- public List<SyncQueueItemVO> dequeueFromAny(Long msid, int maxItems);
- public void purgeItem(long queueItemId);
- public void returnItem(long queueItemId);
-
- public List<SyncQueueItemVO> getActiveQueueItems(Long msid, boolean exclusive);
- public List<SyncQueueItemVO> getBlockedQueueItems(long thresholdMs, boolean exclusive);
-
- void purgeAsyncJobQueueItemId(long asyncJobId);
-}
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/8930cfa9/server/src/com/cloud/async/SyncQueueManagerImpl.java
----------------------------------------------------------------------
diff --git a/server/src/com/cloud/async/SyncQueueManagerImpl.java b/server/src/com/cloud/async/SyncQueueManagerImpl.java
deleted file mode 100644
index 14e24b1..0000000
--- a/server/src/com/cloud/async/SyncQueueManagerImpl.java
+++ /dev/null
@@ -1,253 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-package com.cloud.async;
-
-import java.util.ArrayList;
-import java.util.Date;
-import java.util.List;
-import java.util.Map;
-
-import javax.ejb.Local;
-import javax.inject.Inject;
-import javax.naming.ConfigurationException;
-
-import org.apache.log4j.Logger;
-import org.springframework.stereotype.Component;
-
-import com.cloud.async.dao.SyncQueueDao;
-import com.cloud.async.dao.SyncQueueItemDao;
-import com.cloud.utils.DateUtil;
-import com.cloud.utils.component.ManagerBase;
-import com.cloud.utils.db.DB;
-import com.cloud.utils.db.Transaction;
-import com.cloud.utils.exception.CloudRuntimeException;
-
-@Component
-@Local(value={SyncQueueManager.class})
-public class SyncQueueManagerImpl extends ManagerBase implements SyncQueueManager {
- public static final Logger s_logger = Logger.getLogger(SyncQueueManagerImpl.class.getName());
-
- @Inject private SyncQueueDao _syncQueueDao;
- @Inject private SyncQueueItemDao _syncQueueItemDao;
-
- @Override
- @DB
- public SyncQueueVO queue(String syncObjType, long syncObjId, String itemType, long itemId, long queueSizeLimit) {
- Transaction txn = Transaction.currentTxn();
- try {
- txn.start();
-
- _syncQueueDao.ensureQueue(syncObjType, syncObjId);
- SyncQueueVO queueVO = _syncQueueDao.find(syncObjType, syncObjId);
- if(queueVO == null)
- throw new CloudRuntimeException("Unable to queue item into DB, DB is full?");
-
- queueVO.setQueueSizeLimit(queueSizeLimit);
- _syncQueueDao.update(queueVO.getId(), queueVO);
-
- Date dt = DateUtil.currentGMTTime();
- SyncQueueItemVO item = new SyncQueueItemVO();
- item.setQueueId(queueVO.getId());
- item.setContentType(itemType);
- item.setContentId(itemId);
- item.setCreated(dt);
-
- _syncQueueItemDao.persist(item);
- txn.commit();
-
- return queueVO;
- } catch(Exception e) {
- s_logger.error("Unexpected exception: ", e);
- txn.rollback();
- }
- return null;
- }
-
- @Override
- @DB
- public SyncQueueItemVO dequeueFromOne(long queueId, Long msid) {
- Transaction txt = Transaction.currentTxn();
- try {
- txt.start();
-
- SyncQueueVO queueVO = _syncQueueDao.lockRow(queueId, true);
- if(queueVO == null) {
- s_logger.error("Sync queue(id: " + queueId + ") does not exist");
- txt.commit();
- return null;
- }
-
- if(queueReadyToProcess(queueVO)) {
- SyncQueueItemVO itemVO = _syncQueueItemDao.getNextQueueItem(queueVO.getId());
- if(itemVO != null) {
- Long processNumber = queueVO.getLastProcessNumber();
- if(processNumber == null)
- processNumber = new Long(1);
- else
- processNumber = processNumber + 1;
- Date dt = DateUtil.currentGMTTime();
- queueVO.setLastProcessNumber(processNumber);
- queueVO.setLastUpdated(dt);
- queueVO.setQueueSize(queueVO.getQueueSize() + 1);
- _syncQueueDao.update(queueVO.getId(), queueVO);
-
- itemVO.setLastProcessMsid(msid);
- itemVO.setLastProcessNumber(processNumber);
- itemVO.setLastProcessTime(dt);
- _syncQueueItemDao.update(itemVO.getId(), itemVO);
-
- txt.commit();
- return itemVO;
- } else {
- if(s_logger.isDebugEnabled())
- s_logger.debug("Sync queue (" + queueId + ") is currently empty");
- }
- } else {
- if(s_logger.isDebugEnabled())
- s_logger.debug("There is a pending process in sync queue(id: " + queueId + ")");
- }
- txt.commit();
- } catch(Exception e) {
- s_logger.error("Unexpected exception: ", e);
- txt.rollback();
- }
-
- return null;
- }
-
- @Override
- @DB
- public List<SyncQueueItemVO> dequeueFromAny(Long msid, int maxItems) {
-
- List<SyncQueueItemVO> resultList = new ArrayList<SyncQueueItemVO>();
- Transaction txt = Transaction.currentTxn();
- try {
- txt.start();
-
- List<SyncQueueItemVO> l = _syncQueueItemDao.getNextQueueItems(maxItems);
- if(l != null && l.size() > 0) {
- for(SyncQueueItemVO item : l) {
- SyncQueueVO queueVO = _syncQueueDao.lockRow(item.getQueueId(), true);
- SyncQueueItemVO itemVO = _syncQueueItemDao.lockRow(item.getId(), true);
- if(queueReadyToProcess(queueVO) && itemVO.getLastProcessNumber() == null) {
- Long processNumber = queueVO.getLastProcessNumber();
- if(processNumber == null)
- processNumber = new Long(1);
- else
- processNumber = processNumber + 1;
-
- Date dt = DateUtil.currentGMTTime();
- queueVO.setLastProcessNumber(processNumber);
- queueVO.setLastUpdated(dt);
- queueVO.setQueueSize(queueVO.getQueueSize() + 1);
- _syncQueueDao.update(queueVO.getId(), queueVO);
-
- itemVO.setLastProcessMsid(msid);
- itemVO.setLastProcessNumber(processNumber);
- itemVO.setLastProcessTime(dt);
- _syncQueueItemDao.update(item.getId(), itemVO);
-
- resultList.add(item);
- }
- }
- }
- txt.commit();
- return resultList;
- } catch(Exception e) {
- s_logger.error("Unexpected exception: ", e);
- txt.rollback();
- }
- return null;
- }
-
- @Override
- @DB
- public void purgeItem(long queueItemId) {
- Transaction txt = Transaction.currentTxn();
- try {
- txt.start();
-
- SyncQueueItemVO itemVO = _syncQueueItemDao.findById(queueItemId);
- if(itemVO != null) {
- SyncQueueVO queueVO = _syncQueueDao.lockRow(itemVO.getQueueId(), true);
-
- _syncQueueItemDao.expunge(itemVO.getId());
-
- //if item is active, reset queue information
- if (itemVO.getLastProcessMsid() != null) {
- queueVO.setLastUpdated(DateUtil.currentGMTTime());
- //decrement the count
- assert (queueVO.getQueueSize() > 0) : "Count reduce happens when it's already <= 0!";
- queueVO.setQueueSize(queueVO.getQueueSize() - 1);
- _syncQueueDao.update(queueVO.getId(), queueVO);
- }
- }
- txt.commit();
- } catch(Exception e) {
- s_logger.error("Unexpected exception: ", e);
- txt.rollback();
- }
- }
-
- @Override
- @DB
- public void returnItem(long queueItemId) {
- Transaction txt = Transaction.currentTxn();
- try {
- txt.start();
-
- SyncQueueItemVO itemVO = _syncQueueItemDao.findById(queueItemId);
- if(itemVO != null) {
- SyncQueueVO queueVO = _syncQueueDao.lockRow(itemVO.getQueueId(), true);
-
- itemVO.setLastProcessMsid(null);
- itemVO.setLastProcessNumber(null);
- itemVO.setLastProcessTime(null);
- _syncQueueItemDao.update(queueItemId, itemVO);
-
- queueVO.setLastUpdated(DateUtil.currentGMTTime());
- _syncQueueDao.update(queueVO.getId(), queueVO);
- }
- txt.commit();
- } catch(Exception e) {
- s_logger.error("Unexpected exception: ", e);
- txt.rollback();
- }
- }
-
- @Override
- public List<SyncQueueItemVO> getActiveQueueItems(Long msid, boolean exclusive) {
- return _syncQueueItemDao.getActiveQueueItems(msid, exclusive);
- }
-
- @Override
- public List<SyncQueueItemVO> getBlockedQueueItems(long thresholdMs, boolean exclusive) {
- return _syncQueueItemDao.getBlockedQueueItems(thresholdMs, exclusive);
- }
-
- private boolean queueReadyToProcess(SyncQueueVO queueVO) {
- return queueVO.getQueueSize() < queueVO.getQueueSizeLimit();
- }
-
- @Override
- public void purgeAsyncJobQueueItemId(long asyncJobId) {
- Long itemId = _syncQueueItemDao.getQueueItemIdByContentIdAndType(asyncJobId, SyncQueueItem.AsyncJobContentType);
- if (itemId != null) {
- purgeItem(itemId);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/8930cfa9/server/src/com/cloud/async/SyncQueueVO.java
----------------------------------------------------------------------
diff --git a/server/src/com/cloud/async/SyncQueueVO.java b/server/src/com/cloud/async/SyncQueueVO.java
deleted file mode 100644
index 4f2bc4f..0000000
--- a/server/src/com/cloud/async/SyncQueueVO.java
+++ /dev/null
@@ -1,137 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-package com.cloud.async;
-
-import org.apache.cloudstack.api.InternalIdentity;
-
-import java.util.Date;
-import javax.persistence.Column;
-import javax.persistence.Entity;
-import javax.persistence.GeneratedValue;
-import javax.persistence.GenerationType;
-import javax.persistence.Id;
-import javax.persistence.Table;
-import javax.persistence.Temporal;
-import javax.persistence.TemporalType;
-
-@Entity
-@Table(name="sync_queue")
-public class SyncQueueVO implements InternalIdentity {
-
- @Id
- @GeneratedValue(strategy=GenerationType.IDENTITY)
- @Column(name="id")
- private Long id;
-
- @Column(name="sync_objtype")
-
- private String syncObjType;
-
- @Column(name="sync_objid")
- private Long syncObjId;
-
- @Column(name="queue_proc_number")
- private Long lastProcessNumber;
-
- @Column(name="created")
- @Temporal(TemporalType.TIMESTAMP)
- private Date created;
-
- @Column(name="last_updated")
- @Temporal(TemporalType.TIMESTAMP)
- private Date lastUpdated;
-
- @Column(name="queue_size")
- private long queueSize = 0;
-
- @Column(name="queue_size_limit")
- private long queueSizeLimit = 0;
-
- public long getId() {
- return id;
- }
-
- public String getSyncObjType() {
- return syncObjType;
- }
-
- public void setSyncObjType(String syncObjType) {
- this.syncObjType = syncObjType;
- }
-
- public Long getSyncObjId() {
- return syncObjId;
- }
-
- public void setSyncObjId(Long syncObjId) {
- this.syncObjId = syncObjId;
- }
-
- public Long getLastProcessNumber() {
- return lastProcessNumber;
- }
-
- public void setLastProcessNumber(Long number) {
- lastProcessNumber = number;
- }
-
- public Date getCreated() {
- return created;
- }
-
- public void setCreated(Date created) {
- this.created = created;
- }
-
- public Date getLastUpdated() {
- return lastUpdated;
- }
-
- public void setLastUpdated(Date lastUpdated) {
- this.lastUpdated = lastUpdated;
- }
-
- public String toString() {
- StringBuffer sb = new StringBuffer();
- sb.append("SyncQueueVO {id:").append(getId());
- sb.append(", syncObjType: ").append(getSyncObjType());
- sb.append(", syncObjId: ").append(getSyncObjId());
- sb.append(", lastProcessNumber: ").append(getLastProcessNumber());
- sb.append(", lastUpdated: ").append(getLastUpdated());
- sb.append(", created: ").append(getCreated());
- sb.append(", count: ").append(getQueueSize());
- sb.append("}");
- return sb.toString();
- }
-
- public long getQueueSize() {
- return queueSize;
- }
-
- public void setQueueSize(long queueSize) {
- this.queueSize = queueSize;
- }
-
- public long getQueueSizeLimit() {
- return queueSizeLimit;
- }
-
- public void setQueueSizeLimit(long queueSizeLimit) {
- this.queueSizeLimit = queueSizeLimit;
- }
-}
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/8930cfa9/server/src/com/cloud/async/dao/AsyncJobDao.java
----------------------------------------------------------------------
diff --git a/server/src/com/cloud/async/dao/AsyncJobDao.java b/server/src/com/cloud/async/dao/AsyncJobDao.java
deleted file mode 100644
index 104bd90..0000000
--- a/server/src/com/cloud/async/dao/AsyncJobDao.java
+++ /dev/null
@@ -1,33 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-package com.cloud.async.dao;
-
-import java.util.Date;
-import java.util.List;
-
-import org.apache.cloudstack.api.ApiCommandJobType;
-
-import com.cloud.async.AsyncJobVO;
-import com.cloud.utils.db.GenericDao;
-
-public interface AsyncJobDao extends GenericDao<AsyncJobVO, Long> {
- AsyncJobVO findInstancePendingAsyncJob(String instanceType, long instanceId);
- List<AsyncJobVO> findInstancePendingAsyncJobs(ApiCommandJobType instanceType, Long accountId);
- List<AsyncJobVO> getExpiredUnfinishedJobs(Date cutTime, int limit);
- void resetJobProcess(long msid, int jobResultCode, String jobResultMessage);
- List<AsyncJobVO> getExpiredCompletedJobs(Date cutTime, int limit);
-}
\ No newline at end of file