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 org.apache.cloudstack.vm.schedule;
  
  import com.cloud.api.ApiGsonHelper;
  import com.cloud.event.ActionEventUtils;
  import com.cloud.event.EventTypes;
  import com.cloud.user.User;
  import com.cloud.utils.DateUtil;
  import com.cloud.utils.component.ComponentContext;
  import com.cloud.utils.component.ManagerBase;
  import com.cloud.utils.db.GlobalLock;
  import com.cloud.vm.UserVmManager;
  import com.cloud.vm.VirtualMachine;
  import com.google.common.primitives.Longs;
  import org.apache.cloudstack.api.ApiCommandResourceType;
  import org.apache.cloudstack.api.ApiConstants;
  import org.apache.cloudstack.api.command.user.vm.RebootVMCmd;
  import org.apache.cloudstack.api.command.user.vm.StartVMCmd;
  import org.apache.cloudstack.api.command.user.vm.StopVMCmd;
  import org.apache.cloudstack.framework.config.ConfigKey;
  import org.apache.cloudstack.framework.config.Configurable;
  import org.apache.cloudstack.framework.jobs.AsyncJobDispatcher;
  import org.apache.cloudstack.framework.jobs.AsyncJobManager;
  import org.apache.cloudstack.framework.jobs.impl.AsyncJobVO;
  import org.apache.cloudstack.managed.context.ManagedContextTimerTask;
  import org.apache.cloudstack.vm.schedule.dao.VMScheduleDao;
  import org.apache.cloudstack.vm.schedule.dao.VMScheduledJobDao;
  import org.apache.commons.lang.time.DateUtils;
  import org.springframework.scheduling.support.CronExpression;
  
  import javax.inject.Inject;
  import javax.persistence.EntityExistsException;
  import java.time.ZonedDateTime;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.Calendar;
  import java.util.Date;
  import java.util.EnumMap;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.Timer;
  import java.util.TimerTask;
  
  public class VMSchedulerImpl extends ManagerBase implements VMScheduler, Configurable {
      @Inject
      private VMScheduledJobDao vmScheduledJobDao;
      @Inject
      private VMScheduleDao vmScheduleDao;
      @Inject
      private UserVmManager userVmManager;
      @Inject
      private AsyncJobManager asyncJobManager;
      private AsyncJobDispatcher asyncJobDispatcher;
      private Timer vmSchedulerTimer;
      private Date currentTimestamp;
  
      private EnumMap<VMSchedule.Action, String> actionEventMap = new EnumMap<>(VMSchedule.Action.class);
  
      public AsyncJobDispatcher getAsyncJobDispatcher() {
          return asyncJobDispatcher;
      }
  
      public void setAsyncJobDispatcher(final AsyncJobDispatcher dispatcher) {
          asyncJobDispatcher = dispatcher;
      }
  
      @Override
      public ConfigKey<?>[] getConfigKeys() {
          return new ConfigKey<?>[]{VMScheduledJobExpireInterval};
      }
  
      @Override
      public String getConfigComponentName() {
          return VMScheduler.class.getSimpleName();
      }
  
      @Override
      public void removeScheduledJobs(List<Long> vmScheduleIds) {
          if (vmScheduleIds == null || vmScheduleIds.isEmpty()) {
              logger.debug("Removed 0 scheduled jobs");
              return;
          }
          Date now = new Date();
          int rowsRemoved = vmScheduledJobDao.expungeJobsForSchedules(vmScheduleIds, now);
          logger.debug(String.format("Removed %s VM scheduled jobs", rowsRemoved));
      }
  
      @Override
      public void updateScheduledJob(VMScheduleVO vmSchedule) {
          removeScheduledJobs(Longs.asList(vmSchedule.getId()));
          scheduleNextJob(vmSchedule, new Date());
      }
  
      @Override
      public Date scheduleNextJob(VMScheduleVO vmSchedule, Date timestamp) {
          if (!vmSchedule.getEnabled()) {
              logger.debug("VM Schedule {} for VM {} with id {} is disabled. Not scheduling next job.",
                      vmSchedule::toString, () -> userVmManager.getUserVm(vmSchedule.getVmId()), vmSchedule::getVmId);
              return null;
          }
  
          CronExpression cron = DateUtil.parseSchedule(vmSchedule.getSchedule());
          Date startDate = vmSchedule.getStartDate();
          Date endDate = vmSchedule.getEndDate();
          VirtualMachine vm = userVmManager.getUserVm(vmSchedule.getVmId());
  
          if (vm == null) {
              logger.info("VM id={} is removed. Disabling VM schedule {}.", vmSchedule.getVmId(), vmSchedule);
              vmSchedule.setEnabled(false);
              vmScheduleDao.persist(vmSchedule);
              return null;
          }
  
          ZonedDateTime now;
          if (timestamp != null) {
              now = ZonedDateTime.ofInstant(timestamp.toInstant(), vmSchedule.getTimeZoneId());
          } else {
              now = ZonedDateTime.now(vmSchedule.getTimeZoneId());
          }
          ZonedDateTime zonedStartDate = ZonedDateTime.ofInstant(startDate.toInstant(), vmSchedule.getTimeZoneId());
          ZonedDateTime zonedEndDate = null;
          if (endDate != null) {
              zonedEndDate = ZonedDateTime.ofInstant(endDate.toInstant(), vmSchedule.getTimeZoneId());
          }
          if (zonedEndDate != null && now.isAfter(zonedEndDate)) {
              logger.info("End time is less than current time. Disabling VM schedule {} for VM {}.", vmSchedule, vm);
              vmSchedule.setEnabled(false);
              vmScheduleDao.persist(vmSchedule);
              return null;
          }
  
          ZonedDateTime ts = null;
          if (zonedStartDate.isAfter(now)) {
              ts = cron.next(zonedStartDate);
          } else {
              ts = cron.next(now);
          }
  
          if (ts == null) {
              logger.info("No next schedule found. Disabling VM schedule {} for VM {}.", vmSchedule, vm);
              vmSchedule.setEnabled(false);
              vmScheduleDao.persist(vmSchedule);
              return null;
          }
  
          Date scheduledDateTime = Date.from(ts.toInstant());
          VMScheduledJobVO scheduledJob = new VMScheduledJobVO(vmSchedule.getVmId(), vmSchedule.getId(), vmSchedule.getAction(), scheduledDateTime);
          try {
              vmScheduledJobDao.persist(scheduledJob);
              ActionEventUtils.onScheduledActionEvent(User.UID_SYSTEM, vm.getAccountId(), actionEventMap.get(vmSchedule.getAction()),
                      String.format("Scheduled action (%s) [vm: %s, schedule: %s] at %s", vmSchedule.getAction(), vm, vmSchedule, scheduledDateTime),
                      vm.getId(), ApiCommandResourceType.VirtualMachine.toString(), true, 0);
          } catch (EntityExistsException exception) {
              logger.debug("Job is already scheduled.");
          }
          return scheduledDateTime;
      }
  
      @Override
      public boolean start() {
          actionEventMap.put(VMSchedule.Action.START, EventTypes.EVENT_VM_SCHEDULE_START);
          actionEventMap.put(VMSchedule.Action.STOP, EventTypes.EVENT_VM_SCHEDULE_STOP);
          actionEventMap.put(VMSchedule.Action.REBOOT, EventTypes.EVENT_VM_SCHEDULE_REBOOT);
          actionEventMap.put(VMSchedule.Action.FORCE_STOP, EventTypes.EVENT_VM_SCHEDULE_FORCE_STOP);
          actionEventMap.put(VMSchedule.Action.FORCE_REBOOT, EventTypes.EVENT_VM_SCHEDULE_FORCE_REBOOT);
  
          // Adding 1 minute to currentTimestamp to ensure that
          // jobs which were to be run at current time, doesn't cause issues
          currentTimestamp = DateUtils.addMinutes(new Date(), 1);
          scheduleNextJobs(currentTimestamp);
  
          final TimerTask schedulerPollTask = new ManagedContextTimerTask() {
              @Override
              protected void runInContext() {
                  try {
                      poll(new Date());
                  } catch (final Throwable t) {
                      logger.warn("Catch throwable in VM scheduler ", t);
                  }
              }
          };
  
          vmSchedulerTimer = new Timer("VMSchedulerPollTask");
          vmSchedulerTimer.scheduleAtFixedRate(schedulerPollTask, 5000L, 60 * 1000L);
          return true;
      }
  
      @Override
      public void poll(Date timestamp) {
          currentTimestamp = DateUtils.round(timestamp, Calendar.MINUTE);
          String displayTime = DateUtil.displayDateInTimezone(DateUtil.GMT_TIMEZONE, currentTimestamp);
          logger.debug(String.format("VM scheduler.poll is being called at %s", displayTime));
  
          GlobalLock scanLock = GlobalLock.getInternLock("vmScheduler.poll");
          try {
              if (scanLock.lock(30)) {
                  try {
                      scheduleNextJobs(currentTimestamp);
                  } finally {
                      scanLock.unlock();
                  }
              }
          } finally {
              scanLock.releaseRef();
          }
  
          scanLock = GlobalLock.getInternLock("vmScheduler.poll");
          try {
              if (scanLock.lock(30)) {
                  try {
                      startJobs(); // Create async job and update scheduled job
                  } finally {
                      scanLock.unlock();
                  }
              }
          } finally {
              scanLock.releaseRef();
          }
  
          try {
              cleanupVMScheduledJobs();
          } catch (Exception e) {
              logger.warn("Error in cleaning up vm scheduled jobs", e);
          }
      }
  
      private void scheduleNextJobs(Date timestamp) {
          for (final VMScheduleVO schedule : vmScheduleDao.listAllActiveSchedules()) {
              try {
                  scheduleNextJob(schedule, timestamp);
              } catch (Exception e) {
                  logger.warn("Error in scheduling next job for schedule {}", schedule, e);
              }
          }
      }
  
      /**
       * Delete scheduled jobs before vm.scheduler.expire.interval days
       */
      private void cleanupVMScheduledJobs() {
          Date deleteBeforeDate = DateUtils.addDays(currentTimestamp, -1 * VMScheduledJobExpireInterval.value());
          int rowsRemoved = vmScheduledJobDao.expungeJobsBefore(deleteBeforeDate);
          logger.info(String.format("Cleaned up %d VM scheduled job entries", rowsRemoved));
      }
  
      void executeJobs(Map<Long, VMScheduledJob> jobsToExecute) {
          String displayTime = DateUtil.displayDateInTimezone(DateUtil.GMT_TIMEZONE, currentTimestamp);
  
          for (Map.Entry<Long, VMScheduledJob> entry : jobsToExecute.entrySet()) {
              VMScheduledJob vmScheduledJob = entry.getValue();
              VirtualMachine vm = userVmManager.getUserVm(vmScheduledJob.getVmId());
  
              VMScheduledJobVO tmpVMScheduleJob = null;
              try {
                  if (logger.isDebugEnabled()) {
                      final Date scheduledTimestamp = vmScheduledJob.getScheduledTime();
                      displayTime = DateUtil.displayDateInTimezone(DateUtil.GMT_TIMEZONE, scheduledTimestamp);
                      logger.debug("Executing {} for VM {} for scheduled job: {} at {}",
                              vmScheduledJob.getAction(), vm, vmScheduledJob, displayTime);
                  }
  
                  tmpVMScheduleJob = vmScheduledJobDao.acquireInLockTable(vmScheduledJob.getId());
                  Long jobId = processJob(vmScheduledJob, vm);
                  if (jobId != null) {
                      tmpVMScheduleJob.setAsyncJobId(jobId);
                      vmScheduledJobDao.update(vmScheduledJob.getId(), tmpVMScheduleJob);
                  }
              } catch (final Exception e) {
                  logger.warn("Executing scheduled job {} failed due to {}", vmScheduledJob, e);
              } finally {
                  if (tmpVMScheduleJob != null) {
                      vmScheduledJobDao.releaseFromLockTable(vmScheduledJob.getId());
                  }
              }
          }
      }
  
      Long processJob(VMScheduledJob vmScheduledJob, VirtualMachine vm) {
          if (!Arrays.asList(VirtualMachine.State.Running, VirtualMachine.State.Stopped).contains(vm.getState())) {
              logger.info("Skipping action ({}) for [vm: {}, scheduled job: {}] because VM is invalid state: {}",
                      vmScheduledJob.getAction(), vm, vmScheduledJob, vm.getState());
              return null;
          }
  
          final Long eventId = ActionEventUtils.onCompletedActionEvent(User.UID_SYSTEM, vm.getAccountId(), null,
                  actionEventMap.get(vmScheduledJob.getAction()), true,
                  String.format("Executing action (%s) for VM: %s", vmScheduledJob.getAction(), vm),
                  vm.getId(), ApiCommandResourceType.VirtualMachine.toString(), 0);
  
          if (vm.getState() == VirtualMachine.State.Running) {
              switch (vmScheduledJob.getAction()) {
                  case STOP:
                      return executeStopVMJob(vm, false, eventId);
                  case FORCE_STOP:
                      return executeStopVMJob(vm, true, eventId);
                  case REBOOT:
                      return executeRebootVMJob(vm, false, eventId);
                  case FORCE_REBOOT:
                      return executeRebootVMJob(vm, true, eventId);
              }
          } else if (vm.getState() == VirtualMachine.State.Stopped && vmScheduledJob.getAction() == VMSchedule.Action.START) {
              return executeStartVMJob(vm, eventId);
          }
  
          logger.warn("Skipping action ({}) for [vm: {}, scheduled job: {}] because VM is in state: {}",
                  vmScheduledJob.getAction(), vm, vmScheduledJob, vm.getState());
          return null;
      }
  
      private void skipJobs(Map<Long, VMScheduledJob> jobsToExecute, Map<Long, List<VMScheduledJob>> jobsNotToExecute) {
          for (Map.Entry<Long, List<VMScheduledJob>> entry : jobsNotToExecute.entrySet()) {
              Long vmId = entry.getKey();
              List<VMScheduledJob> skippedVmScheduledJobVOS = entry.getValue();
              VirtualMachine vm = userVmManager.getUserVm(vmId);
              for (final VMScheduledJob skippedVmScheduledJobVO : skippedVmScheduledJobVOS) {
                  VMScheduledJob scheduledJob = jobsToExecute.get(vmId);
                  logger.info("Skipping scheduled job {} for vm {} because of conflict with another scheduled job {}",
                          skippedVmScheduledJobVO, vm, scheduledJob);
              }
          }
      }
  
      /**
       * Create async jobs for VM scheduled jobs
       */
      private void startJobs() {
          String displayTime = DateUtil.displayDateInTimezone(DateUtil.GMT_TIMEZONE, currentTimestamp);
  
          final List<VMScheduledJobVO> vmScheduledJobs = vmScheduledJobDao.listJobsToStart(currentTimestamp);
          logger.debug(String.format("Got %d scheduled jobs to be executed at %s", vmScheduledJobs.size(), displayTime));
  
          Map<Long, VMScheduledJob> jobsToExecute = new HashMap<>();
          Map<Long, List<VMScheduledJob>> jobsNotToExecute = new HashMap<>();
          for (final VMScheduledJobVO vmScheduledJobVO : vmScheduledJobs) {
              long vmId = vmScheduledJobVO.getVmId();
              if (jobsToExecute.get(vmId) == null) {
                  jobsToExecute.put(vmId, vmScheduledJobVO);
              } else {
                  jobsNotToExecute.computeIfAbsent(vmId, k -> new ArrayList<>()).add(vmScheduledJobVO);
              }
          }
  
          executeJobs(jobsToExecute);
          skipJobs(jobsToExecute, jobsNotToExecute);
      }
  
      long executeStartVMJob(VirtualMachine vm, long eventId) {
          final Map<String, String> params = new HashMap<>();
          params.put(ApiConstants.ID, String.valueOf(vm.getId()));
          params.put("ctxUserId", "1");
          params.put("ctxAccountId", String.valueOf(vm.getAccountId()));
          params.put(ApiConstants.CTX_START_EVENT_ID, String.valueOf(eventId));
  
          final StartVMCmd cmd = new StartVMCmd();
          ComponentContext.inject(cmd);
  
          AsyncJobVO job = new AsyncJobVO("", User.UID_SYSTEM, vm.getAccountId(), StartVMCmd.class.getName(), ApiGsonHelper.getBuilder().create().toJson(params), vm.getId(), cmd.getApiResourceType() != null ? cmd.getApiResourceType().toString() : null, null);
          job.setDispatcher(asyncJobDispatcher.getName());
  
          return asyncJobManager.submitAsyncJob(job);
      }
  
      long executeStopVMJob(VirtualMachine vm, boolean isForced, long eventId) {
          final Map<String, String> params = new HashMap<>();
          params.put(ApiConstants.ID, String.valueOf(vm.getId()));
          params.put("ctxUserId", "1");
          params.put("ctxAccountId", String.valueOf(vm.getAccountId()));
          params.put(ApiConstants.CTX_START_EVENT_ID, String.valueOf(eventId));
          params.put(ApiConstants.FORCED, String.valueOf(isForced));
  
          final StopVMCmd cmd = new StopVMCmd();
          ComponentContext.inject(cmd);
  
          AsyncJobVO job = new AsyncJobVO("", User.UID_SYSTEM, vm.getAccountId(), StopVMCmd.class.getName(), ApiGsonHelper.getBuilder().create().toJson(params), vm.getId(), cmd.getApiResourceType() != null ? cmd.getApiResourceType().toString() : null, null);
          job.setDispatcher(asyncJobDispatcher.getName());
  
          return asyncJobManager.submitAsyncJob(job);
      }
  
      long executeRebootVMJob(VirtualMachine vm, boolean isForced, long eventId) {
          final Map<String, String> params = new HashMap<>();
          params.put(ApiConstants.ID, String.valueOf(vm.getId()));
          params.put("ctxUserId", "1");
          params.put("ctxAccountId", String.valueOf(vm.getAccountId()));
          params.put(ApiConstants.CTX_START_EVENT_ID, String.valueOf(eventId));
          params.put(ApiConstants.FORCED, String.valueOf(isForced));
  
          final RebootVMCmd cmd = new RebootVMCmd();
          ComponentContext.inject(cmd);
  
          AsyncJobVO job = new AsyncJobVO("", User.UID_SYSTEM, vm.getAccountId(), RebootVMCmd.class.getName(), ApiGsonHelper.getBuilder().create().toJson(params), vm.getId(), cmd.getApiResourceType() != null ? cmd.getApiResourceType().toString() : null, null);
          job.setDispatcher(asyncJobDispatcher.getName());
  
          return asyncJobManager.submitAsyncJob(job);
      }
  }
  