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.snapshot;
  
  import com.cloud.api.query.dao.SnapshotJoinDao;
  import com.cloud.api.query.vo.SnapshotJoinVO;
  import com.cloud.exception.InvalidParameterValueException;
  import com.cloud.hypervisor.Hypervisor;
  import com.cloud.hypervisor.Hypervisor.HypervisorType;
  import com.cloud.storage.DataStoreRole;
  import com.cloud.storage.Snapshot;
  import com.cloud.storage.SnapshotVO;
  import com.cloud.storage.Storage.StoragePoolType;
  import com.cloud.storage.VolumeVO;
  import com.cloud.storage.dao.SnapshotDao;
  import com.cloud.storage.snapshot.SnapshotManager;
  import com.cloud.utils.exception.CloudRuntimeException;
  
  import org.apache.cloudstack.engine.subsystem.api.storage.DataStore;
  import org.apache.cloudstack.engine.subsystem.api.storage.DataStoreCapabilities;
  import org.apache.cloudstack.engine.subsystem.api.storage.DataStoreManager;
  import org.apache.cloudstack.engine.subsystem.api.storage.SnapshotDataFactory;
  import org.apache.cloudstack.engine.subsystem.api.storage.SnapshotInfo;
  import org.apache.cloudstack.engine.subsystem.api.storage.SnapshotService;
  import org.apache.cloudstack.engine.subsystem.api.storage.SnapshotStrategy;
  import org.apache.cloudstack.engine.subsystem.api.storage.StorageStrategyFactory;
  import org.apache.cloudstack.engine.subsystem.api.storage.VolumeInfo;
  import org.apache.cloudstack.storage.datastore.db.PrimaryDataStoreDao;
  import org.apache.cloudstack.storage.datastore.db.SnapshotDataStoreDao;
  import org.apache.cloudstack.storage.datastore.db.SnapshotDataStoreVO;
  import org.apache.cloudstack.storage.datastore.db.StoragePoolVO;
  
  import org.apache.commons.collections.CollectionUtils;
  import org.apache.commons.collections.MapUtils;
  
  import org.apache.commons.lang3.BooleanUtils;
  import org.apache.commons.lang3.builder.ToStringBuilder;
  import org.apache.commons.lang3.builder.ToStringStyle;
  
  import org.apache.logging.log4j.LogManager;
  import org.apache.logging.log4j.Logger;
  
  import javax.inject.Inject;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.Collections;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  import java.util.stream.Collectors;
  
  public class SnapshotHelper {
      protected Logger logger = LogManager.getLogger(getClass());
  
      @Inject
      protected SnapshotDataStoreDao snapshotDataStoreDao;
  
      @Inject
      protected SnapshotDataFactory snapshotFactory;
  
      @Inject
      protected SnapshotService snapshotService;
  
      @Inject
      protected StorageStrategyFactory storageStrategyFactory;
  
      @Inject
      protected DataStoreManager dataStorageManager;
  
      @Inject
      protected SnapshotDao snapshotDao;
  
      @Inject
      protected PrimaryDataStoreDao primaryDataStoreDao;
  
      @Inject
      protected SnapshotJoinDao snapshotJoinDao;
  
      protected boolean backupSnapshotAfterTakingSnapshot = SnapshotInfo.BackupSnapshotAfterTakingSnapshot.value();
  
      protected final Set<StoragePoolType> storagePoolTypesToValidateWithBackupSnapshotAfterTakingSnapshot = new HashSet<>(Arrays.asList(StoragePoolType.RBD,
              StoragePoolType.PowerFlex));
  
       /**
       * If the snapshot is a backup from a KVM snapshot that should be kept only in primary storage, expunges it from secondary storage.
       * @param snapInfo the snapshot info to delete.
       */
      public void expungeTemporarySnapshot(boolean kvmSnapshotOnlyInPrimaryStorage, SnapshotInfo snapInfo) {
          long storeId = snapInfo.getDataStore().getId();
          long zoneId = dataStorageManager.getStoreZoneId(storeId, snapInfo.getDataStore().getRole());
  
          if (isStorageSupportSnapshotToTemplate(snapInfo)) {
              logger.debug("The primary storage does not delete the snapshots even if there is a backup on secondary");
              return;
          }
  
          List<SnapshotJoinVO> snapshots = snapshotJoinDao.listBySnapshotIdAndZoneId(zoneId, snapInfo.getSnapshotId());
          if (snapshots.size() <= 1) {
              if (snapInfo != null) {
                  logger.trace(String.format("Snapshot [{}] is not a temporary backup to create a volume from snapshot. Not expunging it.", snapInfo.getId()));
              }
              return;
          }
  
          if (snapInfo == null) {
              logger.warn("Unable to expunge snapshot due to its info is null.");
              return;
          }
  
          if (!kvmSnapshotOnlyInPrimaryStorage) {
              logger.trace("Snapshot [{}] is not a temporary backup to create a volume from snapshot. Not expunging it.", snapInfo.getSnapshotVO());
              return;
          }
  
          if (!DataStoreRole.Image.equals(snapInfo.getDataStore().getRole())) {
              logger.debug("Expunge template for Snapshot [{}] is called for primary storage role. Not expunging it, " +
                      "but we will still expunge the database reference of the snapshot for image storage role if any", snapInfo.getSnapshotVO());
          } else {
              logger.debug("Expunging snapshot [{}] due to it is a temporary backup to create a volume from snapshot." +
                              " It is occurring because the global setting [{}] has the value [{}].",
                      snapInfo.getSnapshotVO(), SnapshotInfo.BackupSnapshotAfterTakingSnapshot.key(), backupSnapshotAfterTakingSnapshot);
  
              try {
                  snapshotService.deleteSnapshot(snapInfo);
              } catch (CloudRuntimeException ex) {
                  logger.warn("Unable to delete the temporary snapshot [{}] on secondary storage due to [{}]. We still will expunge the database reference, consider"
                          + " manually deleting the file [{}].", snapInfo, ex.getMessage(), snapInfo.getPath(), ex);
              }
          }
  
          if (!DataStoreRole.Image.equals(snapInfo.getDataStore().getRole())) {
              SnapshotInfo imageStoreSnapInfo = snapshotFactory.getSnapshotWithRoleAndZone(snapInfo.getId(), DataStoreRole.Image, zoneId);
              storeId = imageStoreSnapInfo.getDataStore().getId();
          }
          snapshotDataStoreDao.expungeReferenceBySnapshotIdAndDataStoreRole(snapInfo.getId(), storeId, DataStoreRole.Image);
      }
  
      public boolean isStorageSupportSnapshotToTemplate(SnapshotInfo snapInfo) {
          if (DataStoreRole.Primary.equals(snapInfo.getDataStore().getRole())) {
              Map<String, String> capabilities = snapInfo.getDataStore().getDriver().getCapabilities();
              return org.apache.commons.collections4.MapUtils.isNotEmpty(capabilities)
                      && Boolean.parseBoolean(capabilities.get(DataStoreCapabilities.CAN_CREATE_TEMPLATE_FROM_SNAPSHOT.toString()));
          }
          return false;
      }
      /**
       * Backup the snapshot to secondary storage if it should be backed up and was not yet or it is a temporary backup to create a volume.
       * @return The parameter snapInfo if the snapshot is not backupable, else backs up the snapshot to secondary storage and returns its info.
       * @throws CloudRuntimeException
       */
      public SnapshotInfo backupSnapshotToSecondaryStorageIfNotExists(SnapshotInfo snapInfo, DataStoreRole dataStoreRole, Snapshot snapshot, boolean kvmSnapshotOnlyInPrimaryStorage) throws CloudRuntimeException {
          if (!isSnapshotBackupable(snapInfo, dataStoreRole, kvmSnapshotOnlyInPrimaryStorage)) {
              logger.trace(String.format("Snapshot [%s] is already on secondary storage or is not a KVM snapshot that is only kept in primary storage. Therefore, we do not back it up."
                + " up.", snapInfo.getSnapshotVO()));
  
              return snapInfo;
          }
  
          snapInfo = getSnapshotInfoByIdAndRole(snapshot.getId(), DataStoreRole.Primary, null);
  
          SnapshotStrategy snapshotStrategy = storageStrategyFactory.getSnapshotStrategy(snapshot, SnapshotStrategy.SnapshotOperation.BACKUP);
          snapshotStrategy.backupSnapshot(snapInfo);
  
          return getSnapshotInfoByIdAndRole(snapshot.getId(), kvmSnapshotOnlyInPrimaryStorage ? DataStoreRole.Image : dataStoreRole, dataStorageManager.getStoreZoneId(snapInfo.getDataStore().getId(), snapInfo.getDataStore().getRole()));
      }
  
      /**
       * Search for the snapshot info by the snapshot id and {@link DataStoreRole}.
       * @return The snapshot info if it exists, else throws an exception.
       * @throws CloudRuntimeException
       */
      protected SnapshotInfo getSnapshotInfoByIdAndRole(long snapshotId, DataStoreRole dataStoreRole, Long zoneId) throws CloudRuntimeException {
          SnapshotInfo snapInfo = null;
          if (DataStoreRole.Primary.equals(dataStoreRole)) {
              snapInfo = snapshotFactory.getSnapshotOnPrimaryStore(snapshotId);
          } else {
              snapInfo = snapshotFactory.getSnapshotWithRoleAndZone(snapshotId, dataStoreRole, zoneId);
          }
  
          if (snapInfo != null) {
              return snapInfo;
          }
  
          throw new CloudRuntimeException(String.format("Could not find snapshot [%s] in %s storage. Therefore, we do not back it up.", snapshotId, dataStoreRole));
      }
  
      /**
       * Verifies if the snapshot is backupable.
       * @return true if snapInfo is null and dataStoreRole is {@link DataStoreRole#Image} or is a KVM snapshot that is only kept in primary storage, else false.
       */
      protected boolean isSnapshotBackupable(SnapshotInfo snapInfo, DataStoreRole dataStoreRole, boolean kvmSnapshotOnlyInPrimaryStorage) {
          return (snapInfo == null && dataStoreRole == DataStoreRole.Image) || kvmSnapshotOnlyInPrimaryStorage;
      }
  
      /**
       * Verifies if the snapshot was took on KVM and is kept in primary storage.
       * @return true if hypervisor is {@link  HypervisorType#KVM} and data store role is {@link  DataStoreRole#Primary} and global setting "snapshot.backup.to.secondary" is false,
       * else false.
       */
      public boolean isKvmSnapshotOnlyInPrimaryStorage(Snapshot snapshot, DataStoreRole dataStoreRole) {
          return snapshot.getHypervisorType() == Hypervisor.HypervisorType.KVM && dataStoreRole == DataStoreRole.Primary && !backupSnapshotAfterTakingSnapshot;
      }
  
      public boolean isKvmSnapshotOnlyInPrimaryStorage(Snapshot snapshot, DataStoreRole dataStoreRole, Long zoneId){
          List<SnapshotJoinVO> snapshots = snapshotJoinDao.listBySnapshotIdAndZoneId(zoneId, snapshot.getSnapshotId());
          boolean isKvmSnapshotOnlyInPrimaryStorage = snapshots.stream().noneMatch(s -> s.getStoreRole().equals(DataStoreRole.Image));
  
          return snapshot.getHypervisorType() == Hypervisor.HypervisorType.KVM && dataStoreRole == DataStoreRole.Primary && isKvmSnapshotOnlyInPrimaryStorage;
      }
  
      public DataStoreRole getDataStoreRole(Snapshot snapshot) {
          SnapshotDataStoreVO snapshotStore = snapshotDataStoreDao.findOneBySnapshotAndDatastoreRole(snapshot.getId(), DataStoreRole.Primary);
          if (snapshotStore == null) {
              return DataStoreRole.Image;
          }
  
          long storagePoolId = snapshotStore.getDataStoreId();
          StoragePoolVO storagePoolVO = primaryDataStoreDao.findById(storagePoolId);
          if ((storagePoolTypesToValidateWithBackupSnapshotAfterTakingSnapshot.contains(storagePoolVO.getPoolType()) || snapshot.getHypervisorType() == HypervisorType.KVM)
                  && !backupSnapshotAfterTakingSnapshot) {
              return DataStoreRole.Primary;
          }
  
          DataStore dataStore = dataStorageManager.getDataStore(storagePoolId, DataStoreRole.Primary);
          if (dataStore == null) {
              return DataStoreRole.Image;
          }
  
          Map<String, String> mapCapabilities = dataStore.getDriver().getCapabilities();
          if (MapUtils.isNotEmpty(mapCapabilities) && BooleanUtils.toBoolean(mapCapabilities.get(DataStoreCapabilities.STORAGE_SYSTEM_SNAPSHOT.toString()))) {
              return DataStoreRole.Primary;
          }
  
          return DataStoreRole.Image;
      }
  
      public DataStoreRole getDataStoreRole(Snapshot snapshot, Long zoneId) {
          if (zoneId == null) {
              getDataStoreRole(snapshot);
          }
  
          List<SnapshotJoinVO> snapshots = snapshotJoinDao.listBySnapshotIdAndZoneId(zoneId, snapshot.getId());
          boolean snapshotOnPrimary = snapshots.stream().anyMatch(s -> s.getStoreRole().equals(DataStoreRole.Primary));
          if (snapshotOnPrimary) {
              return DataStoreRole.Primary;
          }
  
          return DataStoreRole.Image;
      }
          /**
           * Verifies if it is a KVM volume that has snapshots only in primary storage.
           * @throws CloudRuntimeException If it is a KVM volume and has at least one snapshot only in primary storage.
           */
      public void checkKvmVolumeSnapshotsOnlyInPrimaryStorage(VolumeVO volumeVo, HypervisorType hypervisorType) throws CloudRuntimeException {
          if (HypervisorType.KVM != hypervisorType) {
              logger.trace(String.format("The %s hypervisor [%s] is not KVM, therefore we will not check if the snapshots are only in primary storage.", volumeVo, hypervisorType));
              return;
          }
  
          Set<Long> snapshotIdsOnlyInPrimaryStorage = getSnapshotIdsOnlyInPrimaryStorage(volumeVo.getId());
  
          if (CollectionUtils.isEmpty(snapshotIdsOnlyInPrimaryStorage)) {
              logger.trace(String.format("%s is a KVM volume and all its snapshots exists in the secondary storage, therefore this volume is able for migration.", volumeVo));
              return;
          }
  
          throwCloudRuntimeExceptionOfSnapshotsOnlyInPrimaryStorage(volumeVo, snapshotIdsOnlyInPrimaryStorage);
      }
  
      /**
       * Throws a CloudRuntimeException with the volume and the snapshots only in primary storage.
       */
      protected void throwCloudRuntimeExceptionOfSnapshotsOnlyInPrimaryStorage(VolumeVO volumeVo, Set<Long> snapshotIdsOnlyInPrimaryStorage) throws CloudRuntimeException {
          List<SnapshotVO> snapshots = snapshotDao.listByIds(snapshotIdsOnlyInPrimaryStorage.toArray());
  
          String message = String.format("%s is a KVM volume and has snapshots only in primary storage. Snapshots [%s].%s", volumeVo,
                  snapshots.stream().map(snapshot -> new ToStringBuilder(snapshot, ToStringStyle.JSON_STYLE).append("uuid", snapshot.getUuid()).append("name", snapshot.getName())
                          .build()).collect(Collectors.joining(", ")), backupSnapshotAfterTakingSnapshot ? "" : " Consider excluding them to migrate the volume to another storage.");
  
          logger.error(message);
          throw new CloudRuntimeException(message);
      }
  
      /**
       * Retrieves the ids of the ready snapshots of the volume that only exists in primary storage.
       * @param volumeId volume id to retrieve the snapshots.
       * @return The ids of the ready snapshots of the volume that only exists in primary storage
       */
      protected Set<Long> getSnapshotIdsOnlyInPrimaryStorage(long volumeId) {
          List<SnapshotDataStoreVO> snapshotsReferences = snapshotDataStoreDao.listReadyByVolumeId(volumeId);
          Map<Long, List<SnapshotDataStoreVO>> referencesGroupBySnapshotId = snapshotsReferences.stream().collect(Collectors.groupingBy(reference -> reference.getSnapshotId()));
  
          Set<Long> snapshotIdsOnlyInPrimaryStorage = new HashSet<>();
          for (var reference: referencesGroupBySnapshotId.entrySet()) {
              List<SnapshotDataStoreVO> listReferencesBySnapshotId = reference.getValue();
  
              if  (!listReferencesBySnapshotId.stream().anyMatch(ref -> DataStoreRole.Image == ref.getRole())) {
                  snapshotIdsOnlyInPrimaryStorage.add(reference.getKey());
              }
          }
  
          return snapshotIdsOnlyInPrimaryStorage;
      }
  
      public SnapshotInfo convertSnapshotIfNeeded(SnapshotInfo snapshotInfo) {
  
          if (snapshotInfo.getParent() == null || !HypervisorType.KVM.equals(snapshotInfo.getHypervisorType())) {
              return snapshotInfo;
          }
  
          return snapshotService.convertSnapshot(snapshotInfo);
      }
  
      public void checkIfThereAreMoreThanOnePoolInTheZone(List<Long> poolIds) {
          List<Long> poolsInOneZone = new ArrayList<>();
          for (Long poolId : poolIds) {
              StoragePoolVO pool = primaryDataStoreDao.findById(poolId);
              if (pool != null) {
                  poolsInOneZone.add(pool.getDataCenterId());
              }
          }
          boolean moreThanOnePoolForZone = poolsInOneZone.stream().filter(itr -> Collections.frequency(poolsInOneZone, itr) > 1).count() > 1;
          if (moreThanOnePoolForZone) {
              throw new CloudRuntimeException("Cannot copy the snapshot on multiple storage pools in one zone");
          }
      }
  
      public List<Long> addStoragePoolsForCopyToPrimary(VolumeInfo volume, List<Long> destZoneIds, List<Long> storagePoolIds, Boolean useStorageReplication) {
          if (useStorageReplication) {
              if (volume == null) {
                  throw new InvalidParameterValueException("Could not find volume of a snapshot");
              } else if (!doesStorageSupportCopyBetweenZones(volume.getPoolId())){
                  throw new InvalidParameterValueException("The storage pool does not support copy between zones");
              }
              if (CollectionUtils.isEmpty(destZoneIds)) {
                  throw new InvalidParameterValueException("There is no destination zone provided");
              }
              if (CollectionUtils.isEmpty(storagePoolIds)) {
                  storagePoolIds = new ArrayList<>();
                  for (Long destZone : destZoneIds) {
                      List<StoragePoolVO> pools = primaryDataStoreDao.findPoolsByStorageTypeAndZone(volume.getStoragePoolType(), destZone);
                      if (CollectionUtils.isNotEmpty(pools)) {
                          StoragePoolVO storagePoolVO = pools.stream().filter(pool -> SnapshotManager.UseStorageReplication.valueIn(pool.getId()) == true).findFirst().get();
                          storagePoolIds.add(storagePoolVO.getId());
                      }
                  }
                  if (CollectionUtils.isEmpty(storagePoolIds)) {
                      throw new InvalidParameterValueException("Cannot copy snapshot to primary storage. There aren't storage pools that support this operation");
                  }
              }
              destZoneIds.clear();
          }
          return storagePoolIds;
      }
  
      public boolean doesStorageSupportCopyBetweenZones(Long poolId) {
          DataStore dataStore = dataStorageManager.getDataStore(poolId, DataStoreRole.Primary);
          if (dataStore != null
                  && dataStore.getDriver().getCapabilities().containsKey(DataStoreCapabilities.CAN_COPY_SNAPSHOT_BETWEEN_ZONES_AND_SAME_POOL_TYPE.toString())) {
              return true;
          }
          return false;
      }
  }
  