package org.zstack.storage.primary;

import com.google.gson.JsonSyntaxException;
import org.apache.commons.lang.StringUtils;
import com.google.common.collect.Maps;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.transaction.annotation.Transactional;
import org.zstack.configuration.DiskOfferingSystemTags;
import org.zstack.configuration.InstanceOfferingSystemTags;
import org.zstack.configuration.OfferingUserConfigUtils;
import org.zstack.core.Platform;
import org.zstack.core.cloudbus.*;
import org.zstack.core.componentloader.PluginRegistry;
import org.zstack.core.config.GlobalConfigException;
import org.zstack.core.config.GlobalConfigValidatorExtensionPoint;
import org.zstack.core.db.DatabaseFacade;
import org.zstack.core.db.DbEntityLister;
import org.zstack.core.db.SQL;
import org.zstack.core.db.SimpleQuery;
import org.zstack.core.db.SimpleQuery.Op;
import org.zstack.core.defer.Deferred;
import org.zstack.core.errorcode.ErrorFacade;
import org.zstack.core.thread.AsyncThread;
import org.zstack.header.AbstractService;
import org.zstack.header.apimediator.ApiMessageInterceptionException;
import org.zstack.header.configuration.userconfig.DiskOfferingUserConfig;
import org.zstack.header.configuration.userconfig.DiskOfferingUserConfigValidator;
import org.zstack.header.configuration.userconfig.InstanceOfferingUserConfig;
import org.zstack.header.configuration.userconfig.InstanceOfferingUserConfigValidator;
import org.zstack.header.core.ReturnValueCompletion;
import org.zstack.header.errorcode.ErrorCode;
import org.zstack.header.errorcode.OperationFailureException;
import org.zstack.header.errorcode.SysErrors;
import org.zstack.header.exception.CloudRuntimeException;
import org.zstack.header.managementnode.ManagementNodeChangeListener;
import org.zstack.header.managementnode.ManagementNodeInventory;
import org.zstack.header.managementnode.ManagementNodeReadyExtensionPoint;
import org.zstack.header.message.APICreateMessage;
import org.zstack.header.message.APIMessage;
import org.zstack.header.message.Message;
import org.zstack.header.message.MessageReply;
import org.zstack.header.storage.primary.*;
import org.zstack.header.tag.SystemTagCreateMessageValidator;
import org.zstack.header.tag.SystemTagValidator;
import org.zstack.header.vm.*;
import org.zstack.header.volume.*;
import org.zstack.tag.TagManager;
import org.zstack.utils.*;
import org.zstack.utils.function.Function;
import org.zstack.utils.logging.CLogger;
import org.zstack.utils.network.NetworkUtils;

import javax.persistence.Tuple;
import javax.persistence.TypedQuery;
import java.util.*;
import java.util.concurrent.Callable;

import static org.zstack.core.Platform.*;

public class PrimaryStorageManagerImpl extends AbstractService implements PrimaryStorageManager,
        ManagementNodeChangeListener, ManagementNodeReadyExtensionPoint, VmInstanceStartExtensionPoint,
        VmInstanceCreateExtensionPoint, CreateDataVolumeExtensionPoint, PrimaryStorageDeleteExtensionPoint, InstanceOfferingUserConfigValidator, DiskOfferingUserConfigValidator {
    private static final CLogger logger = Utils.getLogger(PrimaryStorageManager.class);

    @Autowired
    private CloudBus bus;
    @Autowired
    private PluginRegistry pluginRgty;
    @Autowired
    private DatabaseFacade dbf;
    @Autowired
    private DbEntityLister dl;
    @Autowired
    private ErrorFacade errf;
    @Autowired
    private EventFacade evtf;
    @Autowired
    private TagManager tagMgr;
    @Autowired
    private ResourceDestinationMaker destMaker;
    @Autowired
    private PrimaryStorageOverProvisioningManager ratioMgr;
    @Autowired
    private PrimaryStoragePhysicalCapacityManager physicalCapacityMgr;

    private Map<String, PrimaryStorageFactory> primaryStorageFactories = Collections.synchronizedMap(new HashMap<>());
    private Map<String, PrimaryStorageAllocatorStrategyFactory> allocatorFactories = Collections.synchronizedMap(new HashMap<>());
    private Map<String, PrimaryStorageLicenseInfoFactory> primaryStorageLicenseInfoFactories = Collections.synchronizedMap(new HashMap<>());
    private static final Set<Class> allowedMessageAfterSoftDeletion = new HashSet<>();
    private Map<String, PrimaryStorageLicenseInfo> primaryStorageLicenseInfoMap = Maps.newConcurrentMap();

    static {
        allowedMessageAfterSoftDeletion.add(PrimaryStorageDeletionMsg.class);
    }

    void init() {
        PrimaryStorageSystemTags.PRIMARY_STORAGE_ALLOCATOR_UUID_TAG.installValidator(new SystemTagValidator() {
            @Override
            public void validateSystemTag(String resourceUuid, Class resourceType, String systemTag) {
                String uuid = PrimaryStorageSystemTags.PRIMARY_STORAGE_ALLOCATOR_UUID_TAG.getTokenByTag(
                        systemTag, PrimaryStorageSystemTags.PRIMARY_STORAGE_ALLOCATOR_UUID_TAG_TOKEN);
                if (!StringDSL.isZStackUuid(uuid)) {
                    throw new ApiMessageInterceptionException(argerr("%s is invalid. %s is not a valid zstack uuid", systemTag, uuid));
                }

                if (!dbf.isExist(uuid, PrimaryStorageVO.class)) {
                    throw new ApiMessageInterceptionException(argerr("no primary storage[uuid:%s] found", resourceUuid));
                }
            }
        });
    }

    private void handleApiMessage(APIMessage msg) {
        if (msg instanceof APIAddPrimaryStorageMsg) {
            handle((APIAddPrimaryStorageMsg) msg);
        } else if (msg instanceof PrimaryStorageMessage) {
            passThrough((PrimaryStorageMessage) msg);
        } else if (msg instanceof APIGetPrimaryStorageTypesMsg) {
            handle((APIGetPrimaryStorageTypesMsg) msg);
        } else if (msg instanceof APIGetPrimaryStorageAllocatorStrategiesMsg) {
            handle((APIGetPrimaryStorageAllocatorStrategiesMsg) msg);
        } else if (msg instanceof APIGetPrimaryStorageCapacityMsg) {
            handle((APIGetPrimaryStorageCapacityMsg) msg);
        } else if (msg instanceof APIGetPrimaryStorageLicenseInfoMsg) {
            handle((APIGetPrimaryStorageLicenseInfoMsg) msg);
        } else {
            bus.dealWithUnknownMessage(msg);
        }
    }

    private void handle(APIGetPrimaryStorageLicenseInfoMsg msg) {
        APIGetPrimaryStorageLicenseInfoReply reply = new APIGetPrimaryStorageLicenseInfoReply();
        PrimaryStorageVO primaryStorageVO = dbf.findByUuid(msg.getUuid(), PrimaryStorageVO.class);
        if (primaryStorageVO == null) {
            reply.setError(operr("primaryStorage[uuid=%s] does not exist", msg.getUuid()));
            bus.reply(msg, reply);
            return;
        }

        if (primaryStorageLicenseInfoMap.containsKey(msg.getUuid())) {
            PrimaryStorageLicenseInfo primaryStorageLicenseInfo = primaryStorageLicenseInfoMap.get(msg.getUuid());
            reply.setUuid(primaryStorageLicenseInfo.getUuid());
            reply.setExpireTime(primaryStorageLicenseInfo.getExpireTime());
            reply.setName(primaryStorageVO.getName());
            bus.reply(msg, reply);
            return;
        }

        GetPrimaryStorageLicenseInfoMsg gmsg = new GetPrimaryStorageLicenseInfoMsg();
        gmsg.setPrimaryStorageUuid(msg.getUuid());
        bus.makeTargetServiceIdByResourceUuid(gmsg, PrimaryStorageConstant.SERVICE_ID, gmsg.getPrimaryStorageUuid());
        bus.send(gmsg, new CloudBusCallBack(msg) {
            @Override
            public void run(MessageReply r) {
                if (r.isSuccess()) {
                    GetPrimaryStorageLicenseInfoReply reply1 = r.castReply();
                    if (reply1.getPrimaryStorageLicenseInfo() != null) {
                        reply.setUuid(reply1.getPrimaryStorageLicenseInfo().getUuid());
                        reply.setExpireTime(reply1.getPrimaryStorageLicenseInfo().getExpireTime());
                        reply.setName(primaryStorageVO.getName());
                    }
                } else {
                    reply.setError(r.getError());
                }
                bus.reply(msg, reply);
            }
        });
    }

    private void handle(final APIGetPrimaryStorageCapacityMsg msg) {
        APIGetPrimaryStorageCapacityReply reply = new APIGetPrimaryStorageCapacityReply();

        Tuple ret = new Callable<Tuple>() {
            @Override
            @Transactional(readOnly = true)
            public Tuple call() {
                if (msg.getPrimaryStorageUuids() != null && !msg.getPrimaryStorageUuids().isEmpty()) {
                    String sql = "select sum(psc.totalCapacity)," +
                            " sum(psc.availableCapacity)," +
                            " sum(psc.totalPhysicalCapacity)," +
                            " sum(psc.availablePhysicalCapacity)" +
                            " from PrimaryStorageCapacityVO psc" +
                            " where psc.uuid in (:psUuids)";
                    TypedQuery<Tuple> q = dbf.getEntityManager().createQuery(sql, Tuple.class);
                    q.setParameter("psUuids", msg.getPrimaryStorageUuids());
                    return q.getSingleResult();
                } else if (msg.getClusterUuids() != null && !msg.getClusterUuids().isEmpty()) {
                    String sql = "select sum(psc.totalCapacity)," +
                            " sum(psc.availableCapacity)," +
                            " sum(psc.totalPhysicalCapacity)," +
                            " sum(psc.availablePhysicalCapacity)" +
                            " from PrimaryStorageCapacityVO psc" +
                            " where psc.uuid in" +
                            " (" +
                            " select distinct ref.primaryStorageUuid from PrimaryStorageClusterRefVO ref" +
                            " where ref.clusterUuid in (:clusterUuids)" +
                            " )";
                    TypedQuery<Tuple> q = dbf.getEntityManager().createQuery(sql, Tuple.class);
                    q.setParameter("clusterUuids", msg.getClusterUuids());
                    return q.getSingleResult();
                } else if (msg.getZoneUuids() != null && !msg.getZoneUuids().isEmpty()) {
                    String sql = "select sum(psc.totalCapacity)," +
                            " sum(psc.availableCapacity)," +
                            " sum(psc.totalPhysicalCapacity)," +
                            " sum(psc.availablePhysicalCapacity)" +
                            " from PrimaryStorageCapacityVO psc, PrimaryStorageVO ps" +
                            " where ps.uuid = psc.uuid" +
                            " and ps.zoneUuid in (:zoneUuids)";
                    TypedQuery<Tuple> q = dbf.getEntityManager().createQuery(sql, Tuple.class);
                    q.setParameter("zoneUuids", msg.getZoneUuids());
                    return q.getSingleResult();
                }

                throw new CloudRuntimeException("should not be here");
            }
        }.call();

        Long total = ret.get(0, Long.class);
        Long avail = ret.get(1, Long.class);
        Long ptotal = ret.get(2, Long.class);
        Long pavail = ret.get(3, Long.class);
        reply.setTotalCapacity(total == null ? 0 : total);
        reply.setAvailableCapacity(avail == null ? 0 : avail);
        reply.setTotalPhysicalCapacity(ptotal == null ? 0 : ptotal);
        reply.setAvailablePhysicalCapacity(pavail == null ? 0 : pavail);
        bus.reply(msg, reply);
    }

    private void handle(APIGetPrimaryStorageAllocatorStrategiesMsg msg) {
        APIGetPrimaryStorageAllocatorStrategiesReply reply = new APIGetPrimaryStorageAllocatorStrategiesReply();
        reply.setPrimaryStorageAllocatorStrategies(PrimaryStorageAllocatorStrategyType.getAllExposedTypeNames());
        bus.reply(msg, reply);
    }

    private void handle(APIGetPrimaryStorageTypesMsg msg) {
        APIGetPrimaryStorageTypesReply reply = new APIGetPrimaryStorageTypesReply();
        List<String> ret = new ArrayList<>(PrimaryStorageType.getAllTypeNames());
        reply.setPrimaryStorageTypes(ret);
        bus.reply(msg, reply);
    }

    private void passThrough(PrimaryStorageMessage pmsg) {
        PrimaryStorageVO vo = dbf.findByUuid(pmsg.getPrimaryStorageUuid(), PrimaryStorageVO.class);
        if (vo == null && allowedMessageAfterSoftDeletion.contains(pmsg.getClass())) {
            PrimaryStorageEO eo = dbf.findByUuid(pmsg.getPrimaryStorageUuid(), PrimaryStorageEO.class);
            vo = ObjectUtils.newAndCopy(eo, PrimaryStorageVO.class);
        }

        Message msg = (Message) pmsg;
        if (vo == null) {
            bus.replyErrorByMessageType(msg, err(SysErrors.RESOURCE_NOT_FOUND, "Cannot find primary storage[uuid:%s], it may have been deleted", pmsg.getPrimaryStorageUuid()));
            return;
        }

        PrimaryStorageFactory factory = getPrimaryStorageFactory(PrimaryStorageType.valueOf(vo.getType()));
        PrimaryStorageVO finalVo = vo;
        PrimaryStorage ps = Platform.New(()-> factory.getPrimaryStorage(finalVo));
        ps.handleMessage(msg);
    }

    @Deferred
    private void handle(final APIAddPrimaryStorageMsg msg) {
        PrimaryStorageType type = PrimaryStorageType.valueOf(msg.getType());
        final PrimaryStorageFactory factory = getPrimaryStorageFactory(type);

        PrimaryStorageVO primaryStorageVO = new PrimaryStorageVO();
        if (msg.getResourceUuid() != null) {
            primaryStorageVO.setUuid(msg.getResourceUuid());
        } else {
            primaryStorageVO.setUuid(Platform.getUuid());
        }
        primaryStorageVO.setUrl(msg.getUrl());
        primaryStorageVO.setType(type.toString());
        primaryStorageVO.setName(msg.getName());
        primaryStorageVO.setDescription(msg.getDescription());
        primaryStorageVO.setState(PrimaryStorageState.Enabled);
        primaryStorageVO.setStatus(PrimaryStorageStatus.Connecting);
        primaryStorageVO.setZoneUuid(msg.getZoneUuid());

        final APIAddPrimaryStorageEvent evt = new APIAddPrimaryStorageEvent(msg.getId());
        final PrimaryStorageInventory inv = factory.createPrimaryStorage(primaryStorageVO, msg);
        primaryStorageVO = dbf.findByUuid(primaryStorageVO.getUuid(), PrimaryStorageVO.class);

        tagMgr.createTagsFromAPICreateMessage(msg, inv.getUuid(), PrimaryStorageVO.class.getSimpleName());

        PrimaryStorageCapacityVO primaryStorageCapacityVO = dbf.findByUuid(primaryStorageVO.getUuid(), PrimaryStorageCapacityVO.class);
        if (primaryStorageCapacityVO == null) {
            primaryStorageCapacityVO = new PrimaryStorageCapacityVO();
            primaryStorageCapacityVO.setUuid(primaryStorageVO.getUuid());
            dbf.persist(primaryStorageCapacityVO);
        }

        final ConnectPrimaryStorageMsg cmsg = new ConnectPrimaryStorageMsg();
        cmsg.setPrimaryStorageUuid(inv.getUuid());
        cmsg.setNewAdded(true);
        bus.makeTargetServiceIdByResourceUuid(cmsg, PrimaryStorageConstant.SERVICE_ID, inv.getUuid());
        final PrimaryStorageVO finalVo = primaryStorageVO;
        bus.send(cmsg, new CloudBusCallBack(msg) {
            @Override
            public void run(MessageReply reply) {
                if (!reply.isSuccess() && !reply.getError().isError(PrimaryStorageErrors.DISCONNECTED)) {
                    evt.setError(reply.getError());
                    logger.warn(String.format("failed to connect primary storage[uuid:%s, name:%s, url:%s]",
                            finalVo.getUuid(), finalVo.getName(), finalVo.getUrl()));
                    dbf.remove(finalVo);
                } else {
                    PrimaryStorageInventory pinv = factory.getInventory(finalVo.getUuid());
                    logger.debug(String.format("successfully add primary storage[uuid:%s, name:%s, url: %s]",
                            finalVo.getUuid(), finalVo.getName(), finalVo.getUrl()));
                    evt.setInventory(pinv);
                }

                bus.publish(evt);
            }
        });

    }

    @Override
    @MessageSafe
    public void handleMessage(Message msg) {
        if (msg instanceof APIMessage) {
            handleApiMessage((APIMessage) msg);
        } else {
            handleLocalMessage(msg);
        }
    }

    private void handleLocalMessage(Message msg) {
        if (msg instanceof AllocatePrimaryStorageMsg) {
            handle((AllocatePrimaryStorageMsg) msg);
        } else if (msg instanceof IncreasePrimaryStorageCapacityMsg) {
            handle((IncreasePrimaryStorageCapacityMsg) msg);
        } else if (msg instanceof DecreasePrimaryStorageCapacityMsg) {
            handle((DecreasePrimaryStorageCapacityMsg) msg);
        } else if (msg instanceof RecalculatePrimaryStorageCapacityMsg) {
            handle((RecalculatePrimaryStorageCapacityMsg) msg);
        } else if (msg instanceof PrimaryStorageMessage) {
            passThrough((PrimaryStorageMessage) msg);
        } else if (msg instanceof GetPrimaryStorageLicenseInfoMsg) {
            handle((GetPrimaryStorageLicenseInfoMsg) msg);
        } else {
            bus.dealWithUnknownMessage(msg);
        }
    }

    private void handle(GetPrimaryStorageLicenseInfoMsg msg) {
        GetPrimaryStorageLicenseInfoReply reply = new GetPrimaryStorageLicenseInfoReply();

        if (!PrimaryStorageSystemTags.PRIMARY_STORAGE_VENDOR.hasTag(msg.getPrimaryStorageUuid())) {
            bus.reply(msg, reply);
            return;
        }
        String vendor = PrimaryStorageSystemTags.PRIMARY_STORAGE_VENDOR.getTokenByResourceUuid(msg.getPrimaryStorageUuid(), PrimaryStorageSystemTags.PRIMARY_STORAGE_VENDOR_TOKEN);
        PrimaryStorageVendor primaryStorageVendor = PrimaryStorageVendor.valueOf(vendor);
        final PrimaryStorageLicenseInfoFactory factory = getPrimaryStorageLicenseInfoFactory(primaryStorageVendor);
        factory.getPrimaryStorageLicenseInfo(msg.getPrimaryStorageUuid(), new ReturnValueCompletion<PrimaryStorageLicenseInfo>(msg) {

            @Override
            public void success(PrimaryStorageLicenseInfo primaryStorageLicenseInfo) {
                primaryStorageLicenseInfoMap.put(msg.getPrimaryStorageUuid(), primaryStorageLicenseInfo);
                reply.setPrimaryStorageLicenseInfo(primaryStorageLicenseInfo);
                bus.reply(msg, reply);
            }

            @Override
            public void fail(ErrorCode errorCode) {
                reply.setError(errorCode);
                bus.reply(msg, reply);
            }
        });
    }

    private void handle(final RecalculatePrimaryStorageCapacityMsg msg) {
        RecalculatePrimaryStorageCapacityReply reply = new RecalculatePrimaryStorageCapacityReply();

        final List<String> psUuids = new ArrayList<>();

        if (msg.getPrimaryStorageUuid() != null) {
            passThrough(msg);
        } else if (msg.getZoneUuid() != null) {
            SimpleQuery<PrimaryStorageVO> q = dbf.createQuery(PrimaryStorageVO.class);
            q.select(PrimaryStorageVO_.uuid);
            q.add(PrimaryStorageVO_.zoneUuid, Op.EQ, msg.getZoneUuid());
            List<String> uuids = q.listValue();
            psUuids.addAll(uuids);

            PrimaryStorageCapacityRecalculator psRecal = new PrimaryStorageCapacityRecalculator();
            psRecal.psUuids = psUuids;
            psRecal.recalculate();
            bus.reply(msg, reply);
        }
    }

    private void handle(IncreasePrimaryStorageCapacityMsg msg) {
        long diskSize = msg.isNoOverProvisioning() ? msg.getDiskSize() : ratioMgr.calculateByRatio(msg.getPrimaryStorageUuid(), msg.getDiskSize());
        PrimaryStorageCapacityUpdater updater = new PrimaryStorageCapacityUpdater(msg.getPrimaryStorageUuid());
        if (updater.increaseAvailableCapacity(diskSize)) {
            if (logger.isTraceEnabled()) {
                logger.trace(String.format("Successfully return %s bytes to primary storage[uuid:%s]",
                        diskSize, msg.getPrimaryStorageUuid()));
            }
        }
    }

    private void handle(DecreasePrimaryStorageCapacityMsg msg) {
        long diskSize = msg.isNoOverProvisioning() ? msg.getDiskSize() : ratioMgr.calculateByRatio(msg.getPrimaryStorageUuid(), msg.getDiskSize());
        PrimaryStorageCapacityUpdater updater = new PrimaryStorageCapacityUpdater(msg.getPrimaryStorageUuid());
        if (updater.decreaseAvailableCapacity(diskSize)) {
            if (logger.isTraceEnabled()) {
                logger.trace(String.format("Successfully return %s bytes to primary storage[uuid:%s]",
                        diskSize, msg.getPrimaryStorageUuid()));
            }
        }
    }

    /**
     * Supported allocation strategy：
     * DefaultPrimaryStorageAllocationStrategy (only work for non-local primary storage)
     * LocalPrimaryStorageStrategy (only work for local primary storage)
     *
     * Note：
     * If the allocation strategy is not specified
     * If the cluster is mounted with local storage, the default is LocalPrimaryStorageStrategy。
     * Otherwise, it is DefaultPrimaryStorageAllocationStrategy
     */
    private void handle(AllocatePrimaryStorageMsg msg) {
        AllocatePrimaryStorageReply reply = new AllocatePrimaryStorageReply(null);

        String allocatorStrategyType = null;
        for (PrimaryStorageAllocatorStrategyExtensionPoint ext : pluginRgty.getExtensionList(PrimaryStorageAllocatorStrategyExtensionPoint.class)) {
            allocatorStrategyType = ext.getPrimaryStorageAllocatorStrategyName(msg);
            if (allocatorStrategyType != null) {
                break;
            }
        }

        if (allocatorStrategyType == null) {
            allocatorStrategyType = msg.getAllocationStrategy() == null ?
                    PrimaryStorageConstant.DEFAULT_PRIMARY_STORAGE_ALLOCATION_STRATEGY_TYPE
                    : msg.getAllocationStrategy();
        }

        if (msg.getExcludeAllocatorStrategies() != null && msg.getExcludeAllocatorStrategies().contains(allocatorStrategyType)) {
            throw new CloudRuntimeException(
                    String.format("%s is set as excluded, there is no available primary storage allocator strategy",
                            allocatorStrategyType));
        }

        PrimaryStorageAllocatorStrategyFactory factory = getPrimaryStorageAllocatorStrategyFactory(
                PrimaryStorageAllocatorStrategyType.valueOf(allocatorStrategyType));
        PrimaryStorageAllocatorStrategy strategy = factory.getPrimaryStorageAllocatorStrategy();
        //
        PrimaryStorageAllocationSpec spec = new PrimaryStorageAllocationSpec();
        spec.setPossiblePrimaryStorageTypes(msg.getPossiblePrimaryStorageTypes());
        spec.setExcludePrimaryStorageTypes(msg.getExcludePrimaryStorageTypes());
        spec.setImageUuid(msg.getImageUuid());
        spec.setDiskOfferingUuid(msg.getDiskOfferingUuid());
        spec.setVmInstanceUuid(msg.getVmInstanceUuid());
        spec.setPurpose(msg.getPurpose());
        spec.setSize(msg.getSize());
        spec.setTotalSize(msg.getTotalSize());
        spec.setNoOverProvisioning(msg.isNoOverProvisioning());
        spec.setRequiredClusterUuids(msg.getRequiredClusterUuids());
        spec.setRequiredHostUuid(msg.getRequiredHostUuid());
        spec.setRequiredZoneUuid(msg.getRequiredZoneUuid());
        spec.setBackupStorageUuid(msg.getBackupStorageUuid());
        spec.setRequiredPrimaryStorageUuid(msg.getRequiredPrimaryStorageUuid());
        spec.setTags(msg.getTags());
        spec.setAllocationMessage(msg);
        spec.setAvoidPrimaryStorageUuids(msg.getExcludePrimaryStorageUuids());
        List<PrimaryStorageInventory> ret = strategy.allocateAllCandidates(spec);

        if (msg.isDryRun()) {
            // check capacity has been done before
            AllocatePrimaryStorageDryRunReply r = new AllocatePrimaryStorageDryRunReply();
            r.setPrimaryStorageInventories(ret);
            bus.reply(msg, r);
            return;
        }
        Iterator<PrimaryStorageInventory> it = ret.iterator();
        List<String> errs = new ArrayList<>();
        PrimaryStorageInventory target = null;
        while (it.hasNext()) {
            PrimaryStorageInventory psInv = it.next();

            if (!physicalCapacityMgr.checkCapacityByRatio(psInv.getUuid(), psInv.getTotalPhysicalCapacity(), psInv.getAvailablePhysicalCapacity())) {
                errs.add(String.format("primary storage[uuid:%s]'s physical capacity usage has exceeded the threshold[%s]",
                        psInv.getUuid(), physicalCapacityMgr.getRatio(psInv.getUuid())));
                continue;
            }

            long requiredSize = spec.getSize();
            if (!msg.isNoOverProvisioning()) {
                requiredSize = ratioMgr.calculateByRatio(psInv.getUuid(), requiredSize);
            }

            if (reserve(psInv, requiredSize)) {
                target = psInv;
                break;
            } else {
                errs.add(String.format("unable to reserve capacity on the primary storage[uuid:%s], it has no space", psInv.getUuid()));
                logger.debug(String.format("concurrent reservation on the primary storage[uuid:%s], try next one", psInv.getUuid()));
            }
        }

        if (target == null) {
            throw new OperationFailureException(operr("cannot find any qualified primary storage, errors are %s", errs));
        }

        reply.setPrimaryStorageInventory(target);
        reply.setSize(msg.getSize());
        bus.reply(msg, reply);
    }

    private boolean reserve(final PrimaryStorageInventory inv, final long size) {
        PrimaryStorageCapacityUpdater updater = new PrimaryStorageCapacityUpdater(inv.getUuid());
        return updater.run(new PrimaryStorageCapacityUpdaterRunnable() {
            @Override
            public PrimaryStorageCapacityVO call(PrimaryStorageCapacityVO cap) {
                long avail = cap.getAvailableCapacity() - size;
                if (avail < 0) {
                    logger.warn(String.format("[Primary Storage Allocation] reserved capacity on primary storage[uuid:%s] failed," +
                            " no available capacity on it", inv.getUuid()));
                    return null;
                }

                long origin = cap.getAvailableCapacity();
                cap.setAvailableCapacity(avail);

                if (logger.isTraceEnabled()) {
                    logger.trace(String.format("[Primary Storage Allocation] reserved %s bytes on primary storage[uuid:%s," +
                            " available before:%s, available now:%s]", size, inv.getUuid(), origin, avail));
                }

                return cap;
            }
        });
    }

    @Override
    public String getId() {
        return bus.makeLocalServiceId(PrimaryStorageConstant.SERVICE_ID);
    }


    @Override
    public boolean start() {
        populateExtensions();
        installGlobalConfigValidator();
        installPrimaryStorageCidrValidator();
        return true;
    }


    private void installPrimaryStorageCidrValidator() {
        class PrimaryStorageCidrValidator implements SystemTagCreateMessageValidator, SystemTagValidator {

            @Override
            public void validateSystemTag(String resourceUuid, Class resourceType, String systemTag) {
                if (PrimaryStorageSystemTags.PRIMARY_STORAGE_GATEWAY.isMatch(systemTag)) {
                    String cidr = PrimaryStorageSystemTags.PRIMARY_STORAGE_GATEWAY.getTokenByTag(systemTag, PrimaryStorageSystemTags.PRIMARY_STORAGE_GATEWAY_TOKEN);
                    if (!NetworkUtils.isCidr(cidr)) {
                        throw new ApiMessageInterceptionException(argerr(
                                "cidr[%s] Input Format Error", cidr));
                    }
                }
            }

            @Override
            public void validateSystemTagInCreateMessage(APICreateMessage msg) {
                if (msg.getSystemTags() == null || msg.getSystemTags().isEmpty()) {
                    return;
                }

                int cidrCount = 0;
                for (String systemTag : msg.getSystemTags()) {
                    if (PrimaryStorageSystemTags.PRIMARY_STORAGE_GATEWAY.isMatch(systemTag)) {
                        if (++cidrCount > 1) {
                            throw new ApiMessageInterceptionException(argerr("only one primaryStorage cidr system tag is allowed, but %d got", cidrCount));
                        }
                        String cidr = PrimaryStorageSystemTags.PRIMARY_STORAGE_GATEWAY.getTokenByTag(systemTag, PrimaryStorageSystemTags.PRIMARY_STORAGE_GATEWAY_TOKEN);
                        if (!NetworkUtils.isCidr(cidr)) {
                            throw new ApiMessageInterceptionException(argerr(
                                    "cidr[%s] Input Format Error", cidr));
                        }
                    }
                }
            }
        }

        PrimaryStorageCidrValidator primaryStorageCidrValidator = new PrimaryStorageCidrValidator();
        tagMgr.installCreateMessageValidator(PrimaryStorageVO.class.getSimpleName(), primaryStorageCidrValidator);
        PrimaryStorageSystemTags.PRIMARY_STORAGE_GATEWAY.installValidator(primaryStorageCidrValidator);
    }


    private void installGlobalConfigValidator() {
        PrimaryStorageGlobalConfig.RESERVED_CAPACITY.installValidateExtension(new GlobalConfigValidatorExtensionPoint() {
            @Override
            public void validateGlobalConfig(String category, String name, String oldValue, String newValue) throws GlobalConfigException {
                if (!SizeUtils.isSizeString(newValue)) {
                    throw new GlobalConfigException(String.format("%s is not a size string;" +
                            " a size string consists of a number ending with suffix B/K/M/G/T or without suffix;" +
                            " for example, 512M, 1G", newValue));
                }
            }
        });
    }

    @Override
    public boolean stop() {
        return true;
    }

    private void populateExtensions() {
        for (PrimaryStorageAllocatorStrategyFactory f : pluginRgty.getExtensionList(PrimaryStorageAllocatorStrategyFactory.class)) {
            PrimaryStorageAllocatorStrategyFactory old = allocatorFactories.get(f.getPrimaryStorageAllocatorStrategyType().toString());
            if (old != null) {
                throw new CloudRuntimeException(String.format("duplicate PrimaryStorageAllocatorStrategyFactory[%s, %s] for type[%s]",
                        f.getClass().getName(), old.getClass().getName(), f.getPrimaryStorageAllocatorStrategyType()));
            }
            allocatorFactories.put(f.getPrimaryStorageAllocatorStrategyType().toString(), f);
        }

        for (PrimaryStorageFactory f : pluginRgty.getExtensionList(PrimaryStorageFactory.class)) {
            PrimaryStorageFactory old = primaryStorageFactories.get(f.getPrimaryStorageType().toString());
            if (old != null) {
                throw new CloudRuntimeException(String.format("duplicate PrimaryStorageFactory[%s, %s] for type[%s]",
                        f.getClass().getName(), old.getClass().getName(), old.getPrimaryStorageType()));
            }
            primaryStorageFactories.put(f.getPrimaryStorageType().toString(), f);
        }

        for (PrimaryStorageLicenseInfoFactory f : pluginRgty.getExtensionList(PrimaryStorageLicenseInfoFactory.class)) {
            PrimaryStorageLicenseInfoFactory old = primaryStorageLicenseInfoFactories.get(f.getPrimaryStorageVendor().toString());
            if (old != null) {
                throw new CloudRuntimeException(String.format("duplicate PrimaryStorageLicenseInfoFactory[%s, %s] for type[%s]",
                        f.getClass().getName(), old.getClass().getName(), old.getPrimaryStorageVendor()));
            }
            primaryStorageLicenseInfoFactories.put(f.getPrimaryStorageVendor().toString(), f);
        }
    }


    private PrimaryStorageAllocatorStrategyFactory getPrimaryStorageAllocatorStrategyFactory(PrimaryStorageAllocatorStrategyType type) {
        PrimaryStorageAllocatorStrategyFactory factory = allocatorFactories.get(type.toString());
        if (factory == null) {
            throw new CloudRuntimeException(String.format("No PrimaryStorageAllocatorStrategyFactory for type: %s found", type));
        }
        return factory;
    }

    @Override
    public PrimaryStorageFactory getPrimaryStorageFactory(PrimaryStorageType type) {
        PrimaryStorageFactory factory = primaryStorageFactories.get(type.toString());
        if (factory == null) {
            throw new CloudRuntimeException(String.format("No PrimaryStorageFactory for type: %s found", type));
        }
        return factory;
    }

    public PrimaryStorageLicenseInfoFactory getPrimaryStorageLicenseInfoFactory(PrimaryStorageVendor type) {
        PrimaryStorageLicenseInfoFactory factory = primaryStorageLicenseInfoFactories.get(type.toString());
        if (factory == null) {
            throw new CloudRuntimeException(String.format("No PrimaryStorageLicenseInfoFactory for type: %s found", type));
        }
        return factory;
    }

    @Override
    public void nodeJoin(ManagementNodeInventory inv) {
    }

    @Override
    public void nodeLeft(ManagementNodeInventory inv) {
        logger.debug(String.format("management node[uuid:%s] left, node[uuid:%s] starts taking over primary storage...",
                inv.getUuid(), Platform.getManagementServerId()));
        loadPrimaryStorage(true);
    }

    private List<String> getPrimaryStorageManagedByUs(boolean skipConnected) {
        List<String> ret = new ArrayList<>();
        SimpleQuery<PrimaryStorageVO> q = dbf.createQuery(PrimaryStorageVO.class);
        q.select(PrimaryStorageVO_.uuid);

        if (skipConnected) {
            // treat connecting as disconnected
            q.add(PrimaryStorageVO_.status, Op.NOT_EQ, PrimaryStorageStatus.Connected);
        }

        List<String> uuids = q.listValue();
        for (String uuid : uuids) {
            if (destMaker.isManagedByUs(uuid)) {
                ret.add(uuid);
            }
        }

        return ret;
    }

    private void loadPrimaryStorage(boolean skipConnected) {
        List<String> uuids = getPrimaryStorageManagedByUs(skipConnected);
        if (uuids.isEmpty()) {
            return;
        }

        List<ConnectPrimaryStorageMsg> msgs = CollectionUtils.transformToList(uuids, new Function<ConnectPrimaryStorageMsg, String>() {
            @Override
            public ConnectPrimaryStorageMsg call(String arg) {
                ConnectPrimaryStorageMsg msg = new ConnectPrimaryStorageMsg();
                msg.setPrimaryStorageUuid(arg);
                bus.makeTargetServiceIdByResourceUuid(msg, PrimaryStorageConstant.SERVICE_ID, arg);
                return msg;
            }
        });

        bus.send(msgs);
    }

    @Override
    public void iAmDead(ManagementNodeInventory inv) {

    }

    @Override
    public void iJoin(ManagementNodeInventory inv) {
    }

    @AsyncThread
    @Override
    public void managementNodeReady() {
        logger.debug(String.format("management node[uuid:%s] joins, starts load primary storage ...",
                Platform.getManagementServerId()));
        loadPrimaryStorage(false);
    }

    private void checkVmAllVolumePrimaryStorageState(String vmUuid) {
        String sql = "select uuid from PrimaryStorageVO where uuid in (" +
                " select distinct(primaryStorageUuid) from VolumeVO" +
                " where vmInstanceUuid = :vmUuid and primaryStorageUuid is not null)" +
                " and state = :psState";
        List<String> result = SQL.New(sql, String.class)
                .param("vmUuid", vmUuid)
                .param("psState", PrimaryStorageState.Maintenance)
                .list();
        if (result != null && !result.isEmpty()) {
            throw new OperationFailureException(argerr("the VM[uuid:%s] volume stored location primary storage is in a state of maintenance", vmUuid));
        }
    }

    @Override
    public String preStartVm(VmInstanceInventory inv) {
        try{
            checkVmAllVolumePrimaryStorageState(inv.getUuid());
            return null;
        }catch (Exception e){
            return e.getMessage();
        }
    }

    @Override
    public void beforeStartVm(VmInstanceInventory inv) {
        // do nothing
    }

    @Override
    public void afterStartVm(VmInstanceInventory inv) {
        // do nothing
    }

    @Override
    public void failedToStartVm(VmInstanceInventory inv, ErrorCode reason) {
        // do nothing
    }

    @Override
    public void preCreateVmInstance(CreateVmInstanceMsg msg) {
        settingRootVolume(msg);
        settingDataVolume(msg);
    }

    private void settingRootVolume(CreateVmInstanceMsg msg) {
        String instanceOffering = msg.getInstanceOfferingUuid();

        if (InstanceOfferingSystemTags.INSTANCE_OFFERING_USER_CONFIG.hasTag(instanceOffering)) {
            InstanceOfferingUserConfig config = OfferingUserConfigUtils.getInstanceOfferingConfig(instanceOffering, InstanceOfferingUserConfig.class);
            if (config.getAllocate() != null && config.getAllocate().getPrimaryStorage() != null) {
                String psUuid = config.getAllocate().getPrimaryStorage().getUuid();
                if (msg.getPrimaryStorageUuidForRootVolume() != null && !msg.getPrimaryStorageUuidForRootVolume().equals(psUuid)) {
                    throw new OperationFailureException(operr("primaryStorageUuid conflict, the primary storage specified by the instance offering is %s, and the primary storage specified in the creation parameter is %s"
                            , psUuid, msg.getPrimaryStorageUuidForRootVolume()));
                }
                msg.setPrimaryStorageUuidForRootVolume(psUuid);
            }
        }

        String rootDiskOffering = msg.getRootDiskOfferingUuid();
        if (rootDiskOffering == null) {
            return;
        }

        if (DiskOfferingSystemTags.DISK_OFFERING_USER_CONFIG.hasTag(rootDiskOffering)) {
            DiskOfferingUserConfig config = OfferingUserConfigUtils.getDiskOfferingConfig(rootDiskOffering, DiskOfferingUserConfig.class);

            if (config.getAllocate() == null) {
                return;
            }

            if (config.getAllocate().getPrimaryStorage() == null) {
                return;
            }

            String psUuid = config.getAllocate().getPrimaryStorage().getUuid();
            if (msg.getPrimaryStorageUuidForRootVolume() != null && !msg.getPrimaryStorageUuidForRootVolume().equals(psUuid)) {
                throw new OperationFailureException(operr("primaryStorageUuid conflict, the primary storage specified by the disk offering is %s, and the primary storage specified in the creation parameter is %s",
                        psUuid, msg.getPrimaryStorageUuidForRootVolume()));
            }
            msg.setPrimaryStorageUuidForRootVolume(psUuid);
        }
    }

    private void settingDataVolume(CreateVmInstanceMsg msg) {
        if (msg.getDataDiskOfferingUuids() == null || msg.getDataDiskOfferingUuids().isEmpty()) {
            return;
        }

        String diskOffering = msg.getDataDiskOfferingUuids().get(0);
        if (diskOffering == null) {
            return;
        }

        if (DiskOfferingSystemTags.DISK_OFFERING_USER_CONFIG.hasTag(diskOffering)) {
            DiskOfferingUserConfig config = OfferingUserConfigUtils.getDiskOfferingConfig(diskOffering, DiskOfferingUserConfig.class);

            if (config.getAllocate() == null) {
                return;
            }

            if (config.getAllocate().getPrimaryStorage() == null) {
                return;
            }

            String psUuid = config.getAllocate().getPrimaryStorage().getUuid();
            if (msg.getPrimaryStorageUuidForDataVolume() != null && !msg.getPrimaryStorageUuidForDataVolume().equals(psUuid)) {
                throw new OperationFailureException(operr("primaryStorageUuid conflict, the primary storage specified by the disk offering is %s, and the primary storage specified in the creation parameter is %s",
                        psUuid, msg.getPrimaryStorageUuidForDataVolume()));
            }
            msg.setPrimaryStorageUuidForDataVolume(psUuid);
        }
    }

    @Override
    public void preCreateVolume(APICreateDataVolumeMsg msg) {
        String diskOffering = msg.getDiskOfferingUuid();
        if (diskOffering == null) {
            return;
        }

        if (DiskOfferingSystemTags.DISK_OFFERING_USER_CONFIG.hasTag(diskOffering)) {
            DiskOfferingUserConfig config = OfferingUserConfigUtils.getDiskOfferingConfig(diskOffering, DiskOfferingUserConfig.class);

            if (config.getAllocate() == null) {
                return;
            }

            if (config.getAllocate().getPrimaryStorage() == null) {
                return;
            }

            String psUuid = config.getAllocate().getPrimaryStorage().getUuid();
            if (msg.getPrimaryStorageUuid() != null && !msg.getPrimaryStorageUuid().equals(psUuid)) {
                throw new OperationFailureException(operr("primaryStorageUuid conflict, the primary storage specified by the disk offering is %s, and the primary storage specified in the creation parameter is %s",
                        psUuid, msg.getPrimaryStorageUuid()));
            }
            msg.setPrimaryStorageUuid(psUuid);
        }
    }

    @Override
    public void afterCreateVolume(VolumeVO vo) {
        return;
    }

    @Override
    public void beforeCreateVolume(VolumeInventory volume) {
        return;
    }

    @Override
    public void validateInstanceOfferingUserConfig(String userConfig, String instanceOfferingUuid) {
        if (StringUtils.isBlank(userConfig)) {
            return;
        }

        InstanceOfferingUserConfig config;

        try {
            config = OfferingUserConfigUtils.toObject(userConfig, InstanceOfferingUserConfig.class);
        } catch (JsonSyntaxException e) {
            throw new IllegalArgumentException("Syntax error(s) in billing instance offering user configuration, user configuration should write in json format.", e);
        }

        if (config.getAllocate() == null) {
            return;
        }

        PrimaryStorageAllocateConfig primaryStorageAllocateConfig = config.getAllocate().getPrimaryStorage();
        if (primaryStorageAllocateConfig == null) {
            return;
        }

        if (primaryStorageAllocateConfig.getType() == null) {
            throw new IllegalArgumentException("primaryStorage type cannot be empty");
        }

        if (primaryStorageAllocateConfig.getUuid() == null) {
            throw new IllegalArgumentException("primaryStorage uuid cannot be empty");
        }

        String psUuid = primaryStorageAllocateConfig.getUuid();
        PrimaryStorageVO primaryStorageVO = dbf.findByUuid(psUuid, PrimaryStorageVO.class);
        if (primaryStorageVO == null) {
            throw new IllegalArgumentException(String.format("primaryStorage[uuid=%s] does not exist", psUuid));
        }

        if (!primaryStorageVO.getType().equalsIgnoreCase(primaryStorageAllocateConfig.getType())) {
            throw new IllegalArgumentException(String.format("primaryStorage[uuid=%s] type is %s", psUuid, primaryStorageVO.getType()));
        }
    }

    @Override
    public void validateDiskOfferingUserConfig(String userConfig, String diskOfferingUuid) {
        if (StringUtils.isBlank(userConfig)) {
            return;
        }

        DiskOfferingUserConfig config;

        try {
            config = OfferingUserConfigUtils.toObject(userConfig, DiskOfferingUserConfig.class);
        } catch (JsonSyntaxException e) {
            throw new IllegalArgumentException("Syntax error(s) in disk offering user configuration, user configuration should write in json format.", e);
        }

        if (config.getAllocate() == null) {
            return;
        }

        PrimaryStorageAllocateConfig primaryStorageAllocateConfig = config.getAllocate().getPrimaryStorage();
        if (primaryStorageAllocateConfig == null) {
            return;
        }

        if (primaryStorageAllocateConfig.getType() == null) {
            throw new IllegalArgumentException("primaryStorage type cannot be empty");
        }

        if (primaryStorageAllocateConfig.getUuid() == null) {
            throw new IllegalArgumentException("primaryStorage uuid cannot be empty");
        }

        String psUuid = primaryStorageAllocateConfig.getUuid();
        PrimaryStorageVO primaryStorageVO = dbf.findByUuid(psUuid, PrimaryStorageVO.class);
        if (primaryStorageVO == null) {
            throw new IllegalArgumentException(String.format("primaryStorage[uuid=%s] does not exist", psUuid));
        }

        if (!primaryStorageVO.getType().equalsIgnoreCase(primaryStorageAllocateConfig.getType())) {
            throw new IllegalArgumentException(String.format("primaryStorage[uuid=%s] type is %s", psUuid, primaryStorageVO.getType()));
        }
    }

    @Override
    public void preDeletePrimaryStorage(PrimaryStorageInventory inv) throws PrimaryStorageException {

    }

    @Override
    public void beforeDeletePrimaryStorage(PrimaryStorageInventory inv) {

    }

    @Override
    public void afterDeletePrimaryStorage(PrimaryStorageInventory inv) {
        primaryStorageLicenseInfoMap.remove(inv.getUuid());
    }
}
