| // 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.storage.snapshot; |
| |
| import java.util.Date; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Timer; |
| import java.util.TimerTask; |
| import java.util.concurrent.TimeUnit; |
| |
| import javax.inject.Inject; |
| import javax.naming.ConfigurationException; |
| |
| import org.apache.cloudstack.api.ApiCommandResourceType; |
| import org.apache.cloudstack.api.ApiConstants; |
| import org.apache.cloudstack.api.command.user.snapshot.CreateSnapshotCmd; |
| import org.apache.cloudstack.framework.config.dao.ConfigurationDao; |
| import org.apache.cloudstack.framework.jobs.AsyncJobDispatcher; |
| import org.apache.cloudstack.framework.jobs.AsyncJobManager; |
| import org.apache.cloudstack.framework.jobs.dao.AsyncJobDao; |
| import org.apache.cloudstack.framework.jobs.impl.AsyncJobVO; |
| import org.apache.cloudstack.managed.context.ManagedContextTimerTask; |
| import org.apache.log4j.Logger; |
| import org.springframework.stereotype.Component; |
| |
| import com.cloud.api.ApiDispatcher; |
| import com.cloud.api.ApiGsonHelper; |
| import com.cloud.event.ActionEventUtils; |
| import com.cloud.event.EventTypes; |
| import com.cloud.server.ResourceTag; |
| import com.cloud.server.TaggedResourceService; |
| import com.cloud.storage.Snapshot; |
| import com.cloud.storage.SnapshotPolicyVO; |
| import com.cloud.storage.SnapshotScheduleVO; |
| import com.cloud.storage.SnapshotVO; |
| import com.cloud.storage.VolumeVO; |
| import com.cloud.storage.dao.SnapshotDao; |
| import com.cloud.storage.dao.SnapshotPolicyDao; |
| import com.cloud.storage.dao.SnapshotScheduleDao; |
| import com.cloud.storage.dao.VolumeDao; |
| import com.cloud.user.Account; |
| import com.cloud.user.User; |
| import com.cloud.user.dao.AccountDao; |
| import com.cloud.utils.DateUtil; |
| import com.cloud.utils.DateUtil.IntervalType; |
| import com.cloud.utils.NumbersUtil; |
| import com.cloud.utils.component.ComponentContext; |
| import com.cloud.utils.component.ManagerBase; |
| import com.cloud.utils.concurrency.TestClock; |
| import com.cloud.utils.db.DB; |
| import com.cloud.utils.db.GlobalLock; |
| import com.cloud.utils.db.SearchCriteria; |
| import com.cloud.utils.db.TransactionLegacy; |
| import com.cloud.vm.snapshot.VMSnapshotManager; |
| import com.cloud.vm.snapshot.VMSnapshotVO; |
| import com.cloud.vm.snapshot.dao.VMSnapshotDao; |
| |
| @Component |
| public class SnapshotSchedulerImpl extends ManagerBase implements SnapshotScheduler { |
| private static final Logger s_logger = Logger.getLogger(SnapshotSchedulerImpl.class); |
| |
| @Inject |
| protected AsyncJobDao _asyncJobDao; |
| @Inject |
| protected SnapshotDao _snapshotDao; |
| @Inject |
| protected SnapshotScheduleDao _snapshotScheduleDao; |
| @Inject |
| protected SnapshotPolicyDao _snapshotPolicyDao; |
| @Inject |
| protected AsyncJobManager _asyncMgr; |
| @Inject |
| protected VolumeDao _volsDao; |
| @Inject |
| protected ConfigurationDao _configDao; |
| @Inject |
| protected ApiDispatcher _dispatcher; |
| @Inject |
| protected AccountDao _acctDao; |
| @Inject |
| protected SnapshotApiService _snapshotService; |
| @Inject |
| protected VMSnapshotDao _vmSnapshotDao; |
| @Inject |
| protected VMSnapshotManager _vmSnaphostManager; |
| @Inject |
| public TaggedResourceService taggedResourceService; |
| |
| protected AsyncJobDispatcher _asyncDispatcher; |
| |
| private static final int ACQUIRE_GLOBAL_LOCK_TIMEOUT_FOR_COOPERATION = 5; // 5 seconds |
| private int _snapshotPollInterval; |
| private Timer _testClockTimer; |
| private Date _currentTimestamp; |
| private TestClock _testTimerTask; |
| |
| public AsyncJobDispatcher getAsyncJobDispatcher() { |
| return _asyncDispatcher; |
| } |
| |
| public void setAsyncJobDispatcher(final AsyncJobDispatcher dispatcher) { |
| _asyncDispatcher = dispatcher; |
| } |
| |
| private Date getNextScheduledTime(final long policyId, final Date currentTimestamp) { |
| final SnapshotPolicyVO policy = _snapshotPolicyDao.findById(policyId); |
| Date nextTimestamp = null; |
| if (policy != null) { |
| final short intervalType = policy.getInterval(); |
| final IntervalType type = DateUtil.getIntervalType(intervalType); |
| final String schedule = policy.getSchedule(); |
| final String timezone = policy.getTimezone(); |
| nextTimestamp = DateUtil.getNextRunTime(type, schedule, timezone, currentTimestamp); |
| final String currentTime = DateUtil.displayDateInTimezone(DateUtil.GMT_TIMEZONE, currentTimestamp); |
| final String nextScheduledTime = DateUtil.displayDateInTimezone(DateUtil.GMT_TIMEZONE, nextTimestamp); |
| s_logger.debug("Current time is " + currentTime + ". NextScheduledTime of policyId " + policyId + " is " + nextScheduledTime); |
| } |
| return nextTimestamp; |
| } |
| |
| /** |
| * {@inheritDoc} |
| */ |
| @Override |
| public void poll(final Date currentTimestamp) { |
| // We don't maintain the time. The timer task does. |
| _currentTimestamp = currentTimestamp; |
| |
| GlobalLock scanLock = GlobalLock.getInternLock("snapshot.poll"); |
| try { |
| if (scanLock.lock(ACQUIRE_GLOBAL_LOCK_TIMEOUT_FOR_COOPERATION)) { |
| try { |
| checkStatusOfCurrentlyExecutingSnapshots(); |
| } finally { |
| scanLock.unlock(); |
| } |
| } |
| } finally { |
| scanLock.releaseRef(); |
| } |
| |
| scanLock = GlobalLock.getInternLock("snapshot.poll"); |
| try { |
| if (scanLock.lock(ACQUIRE_GLOBAL_LOCK_TIMEOUT_FOR_COOPERATION)) { |
| try { |
| scheduleSnapshots(); |
| } finally { |
| scanLock.unlock(); |
| } |
| } |
| } finally { |
| scanLock.releaseRef(); |
| } |
| |
| try { |
| deleteExpiredVMSnapshots(); |
| } |
| catch (Exception e) { |
| s_logger.warn("Error in expiring vm snapshots", e); |
| } |
| } |
| |
| private void checkStatusOfCurrentlyExecutingSnapshots() { |
| final SearchCriteria<SnapshotScheduleVO> sc = _snapshotScheduleDao.createSearchCriteria(); |
| sc.addAnd("asyncJobId", SearchCriteria.Op.NNULL); |
| final List<SnapshotScheduleVO> snapshotSchedules = _snapshotScheduleDao.search(sc, null); |
| for (final SnapshotScheduleVO snapshotSchedule : snapshotSchedules) { |
| final Long asyncJobId = snapshotSchedule.getAsyncJobId(); |
| final AsyncJobVO asyncJob = _asyncJobDao.findByIdIncludingRemoved(asyncJobId); |
| switch (asyncJob.getStatus()) { |
| case SUCCEEDED: |
| // The snapshot has been successfully backed up. |
| // The snapshot state has also been cleaned up. |
| // We can schedule the next job for this snapshot. |
| // Remove the existing entry in the snapshot_schedule table. |
| scheduleNextSnapshotJob(snapshotSchedule); |
| break; |
| case FAILED: |
| // Check the snapshot status. |
| final Long snapshotId = snapshotSchedule.getSnapshotId(); |
| if (snapshotId == null) { |
| // createSnapshotAsync exited, successfully or unsuccessfully, |
| // even before creating a snapshot record |
| // No cleanup needs to be done. |
| // Schedule the next snapshot. |
| scheduleNextSnapshotJob(snapshotSchedule); |
| } else { |
| final SnapshotVO snapshot = _snapshotDao.findById(snapshotId); |
| if (snapshot == null || snapshot.getRemoved() != null) { |
| // This snapshot has been deleted successfully from the primary storage |
| // Again no cleanup needs to be done. |
| // Schedule the next snapshot. |
| // There's very little probability that the code reaches this point. |
| // The snapshotId is a foreign key for the snapshot_schedule table |
| // set to ON DELETE CASCADE. So if the snapshot entry is deleted, the snapshot_schedule entry will be too. |
| // But what if it has only been marked as removed? |
| scheduleNextSnapshotJob(snapshotSchedule); |
| } else { |
| // The management server executing this snapshot job appears to have crashed |
| // while creating the snapshot on primary storage/or backing it up. |
| // We have no idea whether the snapshot was successfully taken on the primary or not. |
| // Schedule the next snapshot job. |
| // The ValidatePreviousSnapshotCommand will take appropriate action on this snapshot |
| // If the snapshot was taken successfully on primary, it will retry backing it up. |
| // and cleanup the previous snapshot |
| // Set the userId to that of system. |
| //_snapshotManager.validateSnapshot(1L, snapshot); |
| // In all cases, schedule the next snapshot job |
| scheduleNextSnapshotJob(snapshotSchedule); |
| } |
| } |
| |
| break; |
| case IN_PROGRESS: |
| // There is no way of knowing from here whether |
| // 1) Another management server is processing this snapshot job |
| // 2) The management server has crashed and this snapshot is lying |
| // around in an inconsistent state. |
| // Hopefully, this can be resolved at the backend when the current snapshot gets executed. |
| // But if it remains in this state, the current snapshot will not get executed. |
| // And it will remain in stasis. |
| break; |
| } |
| } |
| } |
| |
| @DB |
| protected void deleteExpiredVMSnapshots() { |
| Date now = new Date(); |
| List<VMSnapshotVO> vmSnapshots = _vmSnapshotDao.listAll(); |
| for (VMSnapshotVO vmSnapshot : vmSnapshots) { |
| long accountId = vmSnapshot.getAccountId(); |
| int expiration_interval_hours = VMSnapshotManager.VMSnapshotExpireInterval.valueIn(accountId); |
| if (expiration_interval_hours < 0 ) { |
| continue; |
| } |
| Date creationTime = vmSnapshot.getCreated(); |
| long diffInHours = TimeUnit.MILLISECONDS.toHours(now.getTime() - creationTime.getTime()); |
| if (diffInHours >= expiration_interval_hours) { |
| if (s_logger.isDebugEnabled()){ |
| s_logger.debug("Deleting expired VM snapshot id: " + vmSnapshot.getId()); |
| } |
| _vmSnaphostManager.deleteVMSnapshot(vmSnapshot.getId()); |
| } |
| } |
| } |
| |
| @DB |
| protected void scheduleSnapshots() { |
| String displayTime = DateUtil.displayDateInTimezone(DateUtil.GMT_TIMEZONE, _currentTimestamp); |
| s_logger.debug("Snapshot scheduler.poll is being called at " + displayTime); |
| |
| final List<SnapshotScheduleVO> snapshotsToBeExecuted = _snapshotScheduleDao.getSchedulesToExecute(_currentTimestamp); |
| s_logger.debug("Got " + snapshotsToBeExecuted.size() + " snapshots to be executed at " + displayTime); |
| |
| for (final SnapshotScheduleVO snapshotToBeExecuted : snapshotsToBeExecuted) { |
| SnapshotScheduleVO tmpSnapshotScheduleVO = null; |
| final long snapshotScheId = snapshotToBeExecuted.getId(); |
| final long policyId = snapshotToBeExecuted.getPolicyId(); |
| final long volumeId = snapshotToBeExecuted.getVolumeId(); |
| try { |
| final VolumeVO volume = _volsDao.findById(volumeId); |
| if (volume.getPoolId() == null) { |
| // this volume is not attached |
| continue; |
| } |
| Account volAcct = _acctDao.findById(volume.getAccountId()); |
| if (volAcct == null || volAcct.getState() == Account.State.DISABLED) { |
| // this account has been removed, so don't trigger recurring snapshot |
| if (s_logger.isDebugEnabled()) { |
| s_logger.debug("Skip snapshot for volume " + volume.getUuid() + " since its account has been removed or disabled"); |
| } |
| continue; |
| } |
| if (_snapshotPolicyDao.findById(policyId) == null) { |
| _snapshotScheduleDao.remove(snapshotToBeExecuted.getId()); |
| } |
| if (s_logger.isDebugEnabled()) { |
| final Date scheduledTimestamp = snapshotToBeExecuted.getScheduledTimestamp(); |
| displayTime = DateUtil.displayDateInTimezone(DateUtil.GMT_TIMEZONE, scheduledTimestamp); |
| s_logger.debug("Scheduling 1 snapshot for volume id " + volumeId + " (volume name:" + |
| volume.getName() + ") for schedule id: " + snapshotToBeExecuted.getId() + " at " + displayTime); |
| } |
| |
| tmpSnapshotScheduleVO = _snapshotScheduleDao.acquireInLockTable(snapshotScheId); |
| final Long eventId = |
| ActionEventUtils.onScheduledActionEvent(User.UID_SYSTEM, volume.getAccountId(), EventTypes.EVENT_SNAPSHOT_CREATE, "creating snapshot for volume Id:" + |
| volume.getUuid(), volumeId, ApiCommandResourceType.Volume.toString(), true, 0); |
| |
| final Map<String, String> params = new HashMap<String, String>(); |
| params.put(ApiConstants.VOLUME_ID, "" + volumeId); |
| params.put(ApiConstants.POLICY_ID, "" + policyId); |
| params.put("ctxUserId", "1"); |
| params.put("ctxAccountId", "" + volume.getAccountId()); |
| params.put("ctxStartEventId", String.valueOf(eventId)); |
| List<? extends ResourceTag> resourceTags = taggedResourceService.listByResourceTypeAndId(ResourceTag.ResourceObjectType.SnapshotPolicy, policyId); |
| if (resourceTags != null && !resourceTags.isEmpty()) { |
| int tagNumber = 0; |
| for (ResourceTag resourceTag : resourceTags) { |
| params.put("tags[" + tagNumber + "].key", resourceTag.getKey()); |
| params.put("tags[" + tagNumber + "].value", resourceTag.getValue()); |
| tagNumber++; |
| } |
| } |
| |
| final CreateSnapshotCmd cmd = new CreateSnapshotCmd(); |
| ComponentContext.inject(cmd); |
| _dispatcher.dispatchCreateCmd(cmd, params); |
| params.put("id", "" + cmd.getEntityId()); |
| params.put("ctxStartEventId", "1"); |
| |
| AsyncJobVO job = new AsyncJobVO("", User.UID_SYSTEM, volume.getAccountId(), CreateSnapshotCmd.class.getName(), |
| ApiGsonHelper.getBuilder().create().toJson(params), cmd.getEntityId(), |
| cmd.getApiResourceType() != null ? cmd.getApiResourceType().toString() : null, null); |
| job.setDispatcher(_asyncDispatcher.getName()); |
| |
| final long jobId = _asyncMgr.submitAsyncJob(job); |
| |
| tmpSnapshotScheduleVO.setAsyncJobId(jobId); |
| _snapshotScheduleDao.update(snapshotScheId, tmpSnapshotScheduleVO); |
| } catch (final Exception e) { |
| // TODO Logging this exception is enough? |
| s_logger.warn("Scheduling snapshot failed due to " + e.toString()); |
| } finally { |
| if (tmpSnapshotScheduleVO != null) { |
| _snapshotScheduleDao.releaseFromLockTable(snapshotScheId); |
| } |
| } |
| } |
| } |
| |
| private Date scheduleNextSnapshotJob(final SnapshotScheduleVO snapshotSchedule) { |
| if (snapshotSchedule == null) { |
| return null; |
| } |
| final Long policyId = snapshotSchedule.getPolicyId(); |
| if (policyId.longValue() == Snapshot.MANUAL_POLICY_ID) { |
| // Don't need to schedule the next job for this. |
| return null; |
| } |
| final SnapshotPolicyVO snapshotPolicy = _snapshotPolicyDao.findById(policyId); |
| if (snapshotPolicy == null) { |
| _snapshotScheduleDao.expunge(snapshotSchedule.getId()); |
| } |
| return scheduleNextSnapshotJob(snapshotPolicy); |
| } |
| |
| @Override |
| @DB |
| public Date scheduleNextSnapshotJob(final SnapshotPolicyVO policy) { |
| if (policy == null) { |
| return null; |
| } |
| |
| // If display attribute is false then remove schedules if any and return. |
| if(!policy.isDisplay()){ |
| removeSchedule(policy.getVolumeId(), policy.getId()); |
| return null; |
| } |
| |
| final long policyId = policy.getId(); |
| if (policyId == Snapshot.MANUAL_POLICY_ID) { |
| return null; |
| } |
| |
| if (_volsDao.findById(policy.getVolumeId()) == null) { |
| s_logger.warn("Found snapshot policy ID: " + policyId + " for volume ID: " + policy.getVolumeId() + " that does not exist or has been removed"); |
| removeSchedule(policy.getVolumeId(), policy.getId()); |
| return null; |
| } |
| |
| final Date nextSnapshotTimestamp = getNextScheduledTime(policyId, _currentTimestamp); |
| SnapshotScheduleVO spstSchedVO = _snapshotScheduleDao.findOneByVolumePolicy(policy.getVolumeId(), policy.getId()); |
| if (spstSchedVO == null) { |
| spstSchedVO = new SnapshotScheduleVO(policy.getVolumeId(), policyId, nextSnapshotTimestamp); |
| _snapshotScheduleDao.persist(spstSchedVO); |
| } else { |
| TransactionLegacy txn = TransactionLegacy.open(TransactionLegacy.CLOUD_DB); |
| |
| try { |
| spstSchedVO = _snapshotScheduleDao.acquireInLockTable(spstSchedVO.getId()); |
| spstSchedVO.setPolicyId(policyId); |
| spstSchedVO.setScheduledTimestamp(nextSnapshotTimestamp); |
| spstSchedVO.setAsyncJobId(null); |
| spstSchedVO.setSnapshotId(null); |
| _snapshotScheduleDao.update(spstSchedVO.getId(), spstSchedVO); |
| txn.commit(); |
| } finally { |
| if (spstSchedVO != null) { |
| _snapshotScheduleDao.releaseFromLockTable(spstSchedVO.getId()); |
| } |
| txn.close(); |
| } |
| } |
| return nextSnapshotTimestamp; |
| } |
| |
| @Override |
| public void scheduleOrCancelNextSnapshotJobOnDisplayChange(final SnapshotPolicyVO policy, boolean previousDisplay) { |
| |
| // Take action only if display changed |
| if(policy.isDisplay() != previousDisplay ){ |
| if(policy.isDisplay()){ |
| scheduleNextSnapshotJob(policy); |
| }else{ |
| removeSchedule(policy.getVolumeId(), policy.getId()); |
| } |
| } |
| } |
| |
| |
| @Override |
| @DB |
| public boolean removeSchedule(final Long volumeId, final Long policyId) { |
| // We can only remove schedules which are in the future. Not which are already executed in the past. |
| final SnapshotScheduleVO schedule = _snapshotScheduleDao.getCurrentSchedule(volumeId, policyId, false); |
| boolean success = true; |
| if (schedule != null) { |
| success = _snapshotScheduleDao.remove(schedule.getId()); |
| } |
| if (!success) { |
| s_logger.debug("Error while deleting Snapshot schedule with Id: " + schedule.getId()); |
| } |
| return success; |
| } |
| |
| @Override |
| public boolean configure(final String name, final Map<String, Object> params) throws ConfigurationException { |
| |
| _snapshotPollInterval = NumbersUtil.parseInt(_configDao.getValue("snapshot.poll.interval"), 300); |
| final boolean snapshotsRecurringTest = Boolean.parseBoolean(_configDao.getValue("snapshot.recurring.test")); |
| if (snapshotsRecurringTest) { |
| // look for some test values in the configuration table so that snapshots can be taken more frequently (QA test code) |
| final int minutesPerHour = NumbersUtil.parseInt(_configDao.getValue("snapshot.test.minutes.per.hour"), 60); |
| final int hoursPerDay = NumbersUtil.parseInt(_configDao.getValue("snapshot.test.hours.per.day"), 24); |
| final int daysPerWeek = NumbersUtil.parseInt(_configDao.getValue("snapshot.test.days.per.week"), 7); |
| final int daysPerMonth = NumbersUtil.parseInt(_configDao.getValue("snapshot.test.days.per.month"), 30); |
| final int weeksPerMonth = NumbersUtil.parseInt(_configDao.getValue("snapshot.test.weeks.per.month"), 4); |
| final int monthsPerYear = NumbersUtil.parseInt(_configDao.getValue("snapshot.test.months.per.year"), 12); |
| |
| _testTimerTask = new TestClock(this, minutesPerHour, hoursPerDay, daysPerWeek, daysPerMonth, weeksPerMonth, monthsPerYear); |
| } |
| _currentTimestamp = new Date(); |
| |
| s_logger.info("Snapshot Scheduler is configured."); |
| |
| return true; |
| } |
| |
| @Override |
| @DB |
| public boolean start() { |
| // reschedule all policies after management restart |
| final List<SnapshotPolicyVO> policyInstances = _snapshotPolicyDao.listAll(); |
| for (final SnapshotPolicyVO policyInstance : policyInstances) { |
| if (policyInstance.getId() != Snapshot.MANUAL_POLICY_ID) { |
| scheduleNextSnapshotJob(policyInstance); |
| } |
| } |
| if (_testTimerTask != null) { |
| _testClockTimer = new Timer("TestClock"); |
| // Run the test clock every 60s. Because every tick is counted as 1 minute. |
| // Else it becomes too confusing. |
| _testClockTimer.schedule(_testTimerTask, 100 * 1000L, 60 * 1000L); |
| } else { |
| final TimerTask timerTask = new ManagedContextTimerTask() { |
| @Override |
| protected void runInContext() { |
| try { |
| final Date currentTimestamp = new Date(); |
| poll(currentTimestamp); |
| } catch (final Throwable t) { |
| s_logger.warn("Catch throwable in snapshot scheduler ", t); |
| } |
| } |
| }; |
| _testClockTimer = new Timer("SnapshotPollTask"); |
| _testClockTimer.schedule(timerTask, _snapshotPollInterval * 1000L, _snapshotPollInterval * 1000L); |
| } |
| |
| return true; |
| } |
| |
| @Override |
| public boolean stop() { |
| return true; |
| } |
| } |
