content: // 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.jobs.JobInfo;
  import org.apache.cloudstack.managed.context.ManagedContextTimerTask;
  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.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.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 {
  
      @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);
              logger.debug("Current time is {}. NextScheduledTime of policy {} is {}", currentTime, policy, 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 {
                      scheduleNextSnapshotJobsIfNecessary();
                  } 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) {
              logger.warn("Error in expiring vm snapshots", e);
          }
      }
  
      private void scheduleNextSnapshotJobsIfNecessary() {
          List<SnapshotScheduleVO> snapshotSchedules = _snapshotScheduleDao.getSchedulesAssignedWithAsyncJob();
          logger.info("Verifying the current state of [{}] snapshot schedules and scheduling next jobs, if necessary.", snapshotSchedules.size());
          for (SnapshotScheduleVO snapshotSchedule : snapshotSchedules) {
              scheduleNextSnapshotJobIfNecessary(snapshotSchedule);
          }
      }
  
      protected void scheduleNextSnapshotJobIfNecessary(SnapshotScheduleVO snapshotSchedule) {
          Long asyncJobId = snapshotSchedule.getAsyncJobId();
          AsyncJobVO asyncJob = _asyncJobDao.findByIdIncludingRemoved(asyncJobId);
  
          if (asyncJob == null) {
              logger.debug("The async job [{}] of snapshot schedule [{}] does not exist anymore. Considering it as finished and scheduling the next snapshot job.",
                      asyncJobId, snapshotSchedule);
              scheduleNextSnapshotJob(snapshotSchedule);
              return;
          }
  
          JobInfo.Status status = asyncJob.getStatus();
  
          if (JobInfo.Status.SUCCEEDED.equals(status)) {
              logger.debug("Last job of schedule [{}] succeeded; scheduling the next snapshot job.", snapshotSchedule);
          } else if (JobInfo.Status.FAILED.equals(status)) {
              logger.debug("Last job of schedule [{}] failed with [{}]; scheduling a new snapshot job.", snapshotSchedule, asyncJob.getResult());
          } else {
              logger.debug("Schedule [{}] is still in progress, skipping next job scheduling.", snapshotSchedule);
              return;
          }
  
          scheduleNextSnapshotJob(snapshotSchedule);
      }
  
      @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 (logger.isDebugEnabled()){
                      logger.debug("Deleting expired VM snapshot: {}", vmSnapshot);
                  }
                  _vmSnaphostManager.deleteVMSnapshot(vmSnapshot.getId());
              }
          }
      }
  
      @DB
      protected void scheduleSnapshots() {
          String displayTime = DateUtil.displayDateInTimezone(DateUtil.GMT_TIMEZONE, _currentTimestamp);
          logger.debug(String.format("Snapshot scheduler is being called at [%s].", displayTime));
  
          final List<SnapshotScheduleVO> snapshotsToBeExecuted = _snapshotScheduleDao.getSchedulesToExecute(_currentTimestamp);
          logger.debug(String.format("There are [%s] scheduled snapshots to be executed at [%s].", snapshotsToBeExecuted.size(), displayTime));
  
          for (final SnapshotScheduleVO snapshotToBeExecuted : snapshotsToBeExecuted) {
              SnapshotScheduleVO tmpSnapshotScheduleVO = null;
              final long snapshotScheId = snapshotToBeExecuted.getId();
              final long policyId = snapshotToBeExecuted.getPolicyId();
              final long volumeId = snapshotToBeExecuted.getVolumeId();
              final VolumeVO volume = _volsDao.findByIdIncludingRemoved(snapshotToBeExecuted.getVolumeId());
              try {
                  if (!canSnapshotBeScheduled(snapshotToBeExecuted, volume)) {
                      continue;
                  }
  
                  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);
  
                  logger.trace("Mapping parameters required to generate a CreateSnapshotCmd for snapshot [{}].", snapshotToBeExecuted);
                  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++;
                      }
                  }
  
                  logger.trace("Generating a CreateSnapshotCmd for snapshot [{}] with parameters: [{}].", snapshotToBeExecuted, params.toString());
                  final CreateSnapshotCmd cmd = new CreateSnapshotCmd();
                  ComponentContext.inject(cmd);
                  _dispatcher.dispatchCreateCmd(cmd, params);
                  params.put("id", "" + cmd.getEntityId());
                  params.put("ctxStartEventId", "1");
  
                  final Date scheduledTimestamp = snapshotToBeExecuted.getScheduledTimestamp();
                  displayTime = DateUtil.displayDateInTimezone(DateUtil.GMT_TIMEZONE, scheduledTimestamp);
                  logger.debug("Scheduling snapshot [{}] for volume [{}] at [{}].", snapshotToBeExecuted, volume, displayTime);
                  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);
                  logger.debug("Scheduled snapshot [{}] for volume [{}] as job [{}].", snapshotToBeExecuted, volume, job);
  
                  tmpSnapshotScheduleVO.setAsyncJobId(jobId);
                  _snapshotScheduleDao.update(snapshotScheId, tmpSnapshotScheduleVO);
              } catch (final Exception e) {
                  logger.error("The scheduling of snapshot [{}] for volume [{}] failed due to [{}].", snapshotToBeExecuted, volume, e.toString(), e);
              } finally {
                  if (tmpSnapshotScheduleVO != null) {
                      _snapshotScheduleDao.releaseFromLockTable(snapshotScheId);
                  }
              }
          }
      }
  
      /**
       * Verifies if a snapshot for a volume can be scheduled or not based on volume and account status, and removes it from the snapshot scheduler if its policy was removed.
       *
       * @param snapshotToBeScheduled the snapshot to be scheduled
       * @param volume the volume associated with the snapshot to be scheduled
       * @return <code>true</code> if the snapshot can be scheduled, and <code>false</code> otherwise.
       */
      protected boolean canSnapshotBeScheduled(final SnapshotScheduleVO snapshotToBeScheduled, final VolumeVO volume) {
          if (volume.getRemoved() != null) {
              logger.warn("Skipping snapshot [{}] for volume [{}] because it has been removed. Having a snapshot scheduled for a volume that has been "
                      + "removed is an inconsistency; please, check your database.", snapshotToBeScheduled, volume);
              return false;
          }
  
          if (volume.getPoolId() == null) {
              logger.debug("Skipping snapshot [{}] for volume [{}] because it is not attached to any storage pool.", snapshotToBeScheduled, volume);
              return false;
          }
  
          if (isAccountRemovedOrDisabled(snapshotToBeScheduled, volume)) {
              return false;
          }
  
          if (_snapshotPolicyDao.findById(snapshotToBeScheduled.getPolicyId()) == null) {
              logger.debug("Snapshot's policy [{}] for volume [{}] has been removed; " +
                      "therefore, this snapshot will be removed from the snapshot scheduler.",
                      snapshotToBeScheduled.getPolicyId(), volume);
              _snapshotScheduleDao.remove(snapshotToBeScheduled.getId());
          }
  
          logger.debug("Snapshot [{}] for volume [{}] can be executed.", snapshotToBeScheduled, volume);
          return true;
      }
  
      protected boolean isAccountRemovedOrDisabled(final SnapshotScheduleVO snapshotToBeExecuted, final VolumeVO volume) {
          Account volAcct = _acctDao.findById(volume.getAccountId());
  
          if (volAcct == null) {
              logger.debug(String.format("Skipping snapshot [%s] for volume [%s] because its account [%s] has been removed.",
                      snapshotToBeExecuted, volume, volume.getAccountId()));
              return true;
          }
  
          if (volAcct.getState() == Account.State.DISABLED) {
              logger.debug("Skipping snapshot [{}] for volume [{}] because its account [{}] is disabled.", snapshotToBeExecuted, volume, volAcct);
              return true;
          }
  
          return false;
      }
  
      protected 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) {
              logger.warn("Found snapshot policy: {} for volume ID: {} that does not exist or has been removed", policy, policy.getVolumeId());
              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) {
              logger.debug("Error while deleting Snapshot schedule: " + schedule);
          }
          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();
  
          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) {
                          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;
      }
  }
  