/*
 * Copyright (C) 2016-2022 ActionTech.
 * License: http://www.gnu.org/licenses/gpl.html GPL version 2 or higher.
 */

package com.actiontech.dble.backend.datasource;

import com.actiontech.dble.DbleServer;
import com.actiontech.dble.alarm.AlarmCode;
import com.actiontech.dble.alarm.Alert;
import com.actiontech.dble.alarm.AlertUtil;
import com.actiontech.dble.backend.heartbeat.MySQLHeartbeat;
import com.actiontech.dble.backend.mysql.nio.MySQLInstance;
import com.actiontech.dble.cluster.JsonFactory;
import com.actiontech.dble.cluster.values.DbInstanceStatus;
import com.actiontech.dble.cluster.values.JsonObjectWriter;
import com.actiontech.dble.cluster.values.RawJson;
import com.actiontech.dble.cluster.zkprocess.parse.JsonProcessBase;
import com.actiontech.dble.config.helper.GetAndSyncDbInstanceKeyVariables;
import com.actiontech.dble.config.helper.KeyVariables;
import com.actiontech.dble.config.model.db.DbGroupConfig;
import com.actiontech.dble.config.model.db.DbInstanceConfig;
import com.actiontech.dble.meta.ReloadLogHelper;
import com.actiontech.dble.net.IOProcessor;
import com.actiontech.dble.net.Session;
import com.actiontech.dble.net.connection.BackendConnection;
import com.actiontech.dble.net.connection.PooledConnection;
import com.actiontech.dble.rwsplit.RWSplitNonBlockingSession;
import com.actiontech.dble.singleton.HaConfigManager;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.google.gson.Gson;
import com.google.gson.JsonObject;
import com.google.gson.reflect.TypeToken;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.lang.reflect.Type;
import java.util.*;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class PhysicalDbGroup {
    private static final Logger LOGGER = LoggerFactory.getLogger(PhysicalDbGroup.class);
    public static final String JSON_NAME = "dbGroup";
    public static final String JSON_LIST = "dbInstance";
    // rw split
    public static final int RW_SPLIT_OFF = 0;
    public static final int RW_SPLIT_ALL = 2;
    public static final int RW_SPLIT_ALL_SLAVES_MAY_MASTER = 3;
    private List<PhysicalDbInstance> writeInstanceList;

    private String groupName;
    private DbGroupConfig dbGroupConfig;
    private volatile PhysicalDbInstance writeDbInstance;
    private Map<String, PhysicalDbInstance> allSourceMap = new HashMap<>();

    private int rwSplitMode;
    protected List<String> schemas = Lists.newArrayList();
    private final LoadBalancer loadBalancer = new RandomLoadBalancer();
    private final LocalReadLoadBalancer localReadLoadBalancer = new LocalReadLoadBalancer();
    private final ReentrantReadWriteLock adjustLock = new ReentrantReadWriteLock();

    private boolean shardingUseless = true;
    private boolean rwSplitUseless = true;
    private boolean analysisUseless = true;
    private Set<Session> rwSplitSessionSet = Sets.newConcurrentHashSet();
    private volatile Integer state = Integer.valueOf(INITIAL);


    public static final int STATE_DELETING = 2;
    public static final int STATE_ABANDONED = 1;
    public static final int INITIAL = 0;

    public PhysicalDbGroup(String name, DbGroupConfig config, PhysicalDbInstance writeDbInstances, PhysicalDbInstance[] readDbInstances, int rwSplitMode) {
        this.groupName = name;
        this.rwSplitMode = rwSplitMode;
        this.dbGroupConfig = config;

        writeDbInstances.setDbGroup(this);
        this.writeDbInstance = writeDbInstances;
        this.writeInstanceList = Collections.singletonList(writeDbInstance);
        allSourceMap.put(writeDbInstances.getName(), writeDbInstances);

        for (PhysicalDbInstance readDbInstance : readDbInstances) {
            readDbInstance.setDbGroup(this);
            allSourceMap.put(readDbInstance.getName(), readDbInstance);
        }
    }

    public PhysicalDbGroup(PhysicalDbGroup org) {
        this.groupName = org.groupName;
        this.rwSplitMode = org.rwSplitMode;
        this.dbGroupConfig = org.dbGroupConfig;
        this.allSourceMap = new HashMap<>();
        for (Map.Entry<String, PhysicalDbInstance> entry : org.allSourceMap.entrySet()) {
            MySQLInstance newSource = new MySQLInstance((MySQLInstance) entry.getValue());
            this.allSourceMap.put(entry.getKey(), newSource);
            if (entry.getValue() == org.writeDbInstance) {
                this.writeDbInstance = newSource;
            }
        }
        writeInstanceList = Collections.singletonList(writeDbInstance);
    }

    public String getGroupName() {
        return groupName;
    }

    public List<String> getSchemas() {
        return schemas;
    }

    public void setSchemas(List<String> mySchemas) {
        this.schemas = mySchemas;
    }

    public void addSchema(String schema) {
        this.schemas.add(schema);
    }

    public void removeSchema(String schema) {
        this.schemas.remove(schema);
    }

    public DbGroupConfig getDbGroupConfig() {
        return dbGroupConfig;
    }

    public boolean isAllFakeNode() {
        for (PhysicalDbInstance source : allSourceMap.values()) {
            if (!source.isFakeNode()) {
                return false;
            }
        }
        return true;
    }

    PhysicalDbInstance findDbInstance(BackendConnection exitsCon) {
        PhysicalDbInstance source = (PhysicalDbInstance) exitsCon.getPoolRelated().getInstance();
        PhysicalDbInstance target = allSourceMap.get(source.getName());
        if (source == target) {
            return source;
        }
        LOGGER.info("can't find connection in pool " + this.groupName + " con:" + exitsCon);
        return null;
    }

    boolean isSlave(PhysicalDbInstance ds) {
        return !(writeDbInstance == ds);
    }

    public int getRwSplitMode() {
        return rwSplitMode;
    }

    public boolean isUseless() {
        return shardingUseless && rwSplitUseless && analysisUseless;
    }

    public boolean isShardingUseless() {
        return shardingUseless;
    }

    public void setShardingUseless(boolean shardingUseless) {
        this.shardingUseless = shardingUseless;
    }

    public boolean isRwSplitUseless() {
        return rwSplitUseless;
    }

    public void setRwSplitUseless(boolean rwSplitUseless) {
        this.rwSplitUseless = rwSplitUseless;
    }

    public boolean isAnalysisUseless() {
        return analysisUseless;
    }

    public void setAnalysisUseless(boolean analysisUseless) {
        this.analysisUseless = analysisUseless;
    }

    private boolean checkSlaveSynStatus() {
        return (dbGroupConfig.getDelayThreshold() != -1) &&
                (dbGroupConfig.isShowSlaveSql());
    }

    public PhysicalDbInstance getWriteDbInstance() {
        return writeDbInstance;
    }

    public void init(String reason) {
        if (LOGGER.isDebugEnabled()) {
            ReloadLogHelper.debug("init new group :{},reason:{}", LOGGER, this.toString(), reason);
        }
        for (Map.Entry<String, PhysicalDbInstance> entry : allSourceMap.entrySet()) {
            entry.getValue().init(reason, true);
        }
    }

    public void startOfFresh(List<String> sourceNames, String reason) {
        for (String sourceName : sourceNames) {
            if (allSourceMap.containsKey(sourceName)) {
                allSourceMap.get(sourceName).init(reason, false);
            }
        }
    }

    private boolean checkState() {
        if (isStop()) {
            return false;
        }
        if (getBindingCount() != 0) {
            state = STATE_DELETING;
            IOProcessor.BACKENDS_OLD_GROUP.add(this);
            return false;
        }
        state = STATE_ABANDONED;
        return true;
    }

    public void stop(String reason) {
        stop(reason, false);
    }

    public void stop(String reason, boolean closeFront) {
        if (LOGGER.isDebugEnabled()) {
            ReloadLogHelper.debug("recycle old group :{},reason:{},is close front:{}", LOGGER, this.toString(), reason, closeFront);
        }
        boolean flag = checkState();
        if (!flag) {
            return;
        }
        for (PhysicalDbInstance dbInstance : allSourceMap.values()) {
            dbInstance.stopDirectly(reason, closeFront, false);
        }
    }

    public void stopOfFresh(List<String> sourceNames, String reason, boolean closeFront) {
        for (String sourceName : sourceNames) {
            if (allSourceMap.containsKey(sourceName)) {
                PhysicalDbInstance dbInstance = allSourceMap.get(sourceName);
                dbInstance.stop(reason, closeFront, false, dbGroupConfig.getRwSplitMode() != RW_SPLIT_OFF || writeDbInstance == dbInstance);
            }
        }

        if (closeFront) {
            Iterator<PooledConnection> iterator = IOProcessor.BACKENDS_OLD.iterator();
            while (iterator.hasNext()) {
                PooledConnection con = iterator.next();
                if (con instanceof BackendConnection) {
                    BackendConnection backendCon = (BackendConnection) con;
                    if (backendCon.getPoolDestroyedTime() != 0 && sourceNames.contains(backendCon.getInstance().getConfig().getInstanceName())) {
                        backendCon.closeWithFront("old active backend conn will be forced closed by closing front conn");
                        iterator.remove();
                    }
                }
            }
        }
    }


    public boolean stopOfBackground(String reason) {
        if (state.intValue() == STATE_DELETING && getBindingCount() == 0) {
            for (PhysicalDbInstance dbInstance : allSourceMap.values()) {
                dbInstance.stopDirectly(reason, false, false);
            }
            return true;
        }
        return false;
    }


    public boolean isStop() {
        return state.intValue() != INITIAL;
    }

    public void stopPool(String reason, boolean closeFront, boolean closeWrite) {
        for (PhysicalDbInstance dbInstance : allSourceMap.values()) {
            if (!closeWrite && writeDbInstance == dbInstance) {
                continue;
            }
            dbInstance.stopPool(reason, closeFront);
        }
    }

    public void startPool(String reason, boolean startWrite) {
        for (PhysicalDbInstance dbInstance : allSourceMap.values()) {
            if (!startWrite && writeDbInstance == dbInstance) {
                continue;
            }
            dbInstance.startPool(reason);
        }
    }

    public void stopHeartbeat(String reason) {
        for (PhysicalDbInstance dbInstance : allSourceMap.values()) {
            dbInstance.stopHeartbeat(reason);
        }
    }

    public void startHeartbeat() {
        for (PhysicalDbInstance dbInstance : allSourceMap.values()) {
            if (dbInstance.heartbeat.isStop()) {
                dbInstance.heartbeat = new MySQLHeartbeat(dbInstance);
            }
            dbInstance.startHeartbeat();
        }
    }


    public Collection<PhysicalDbInstance> getDbInstances(boolean isAll) {
        if (!isAll && rwSplitMode == RW_SPLIT_OFF) {
            return writeInstanceList;
        }
        return allSourceMap.values();
    }

    public Map<String, PhysicalDbInstance> getAllDbInstanceMap() {
        return allSourceMap;
    }

    public PhysicalDbInstance[] getReadDbInstances() {
        PhysicalDbInstance[] readSources = new PhysicalDbInstance[allSourceMap.size() - 1];
        int i = 0;
        for (PhysicalDbInstance source : allSourceMap.values()) {
            if (source.getName().equals(writeDbInstance.getName())) {
                continue;
            }
            readSources[i++] = source;
        }
        return readSources;
    }

    /**
     * rwsplit user
     *
     * @param master
     * @param writeStatistical
     * @return
     * @throws IOException
     */
    public PhysicalDbInstance rwSelect(Boolean master, Boolean writeStatistical) throws IOException {
        return rwSelect(master, writeStatistical, false);
    }

    /**
     * rwsplit user
     *
     * @param master
     * @param writeStatistical
     * @param localRead        only the SELECT and show statements attempt to localRead
     * @return
     * @throws IOException
     */
    public PhysicalDbInstance rwSelect(Boolean master, Boolean writeStatistical, boolean localRead) throws IOException {
        if (Objects.nonNull(writeStatistical)) {
            return select(master, false, writeStatistical, localRead);
        }
        return select(master, false, Objects.nonNull(master) ? master : false, localRead);
    }

    /**
     * Sharding user
     *
     * @param master
     * @param isForUpdate
     * @return
     * @throws IOException
     */
    public PhysicalDbInstance select(Boolean master, boolean isForUpdate) throws IOException {
        if (Objects.nonNull(master)) {
            return select(master, isForUpdate, master);
        }
        return select(master, isForUpdate, false);
    }

    public PhysicalDbInstance select(Boolean master, boolean isForUpdate, boolean writeStatistical) throws IOException {
        return select(master, isForUpdate, writeStatistical, false);
    }

    /**
     * Select an instance
     *
     * @param master
     * @param isForUpdate
     * @param writeStatistical
     * @param localRead        only the SELECT and show statements attempt to localRead
     * @return
     * @throws IOException
     */
    public PhysicalDbInstance select(Boolean master, boolean isForUpdate, boolean writeStatistical, boolean localRead) throws IOException {

        if (rwSplitMode == RW_SPLIT_OFF && (master != null && !master)) {
            LOGGER.warn("force slave,but the dbGroup[{}] doesn't contains active slave dbInstance", groupName);
            throw new IOException("force slave,but the dbGroup[" + groupName + "] doesn't contain active slave dbInstance");
        }

        if (rwSplitMode == RW_SPLIT_OFF || allSourceMap.size() == 1 || (master != null && master) || isForUpdate) {
            if (writeDbInstance.isAlive()) {
                if (LOGGER.isDebugEnabled()) {
                    LOGGER.debug("select write {}", writeDbInstance);
                }
                if (writeStatistical) {
                    writeDbInstance.incrementWriteCount();
                } else {
                    writeDbInstance.incrementReadCount();
                }
                return writeDbInstance;
            } else {
                reportError(writeDbInstance);
            }
        }

        List<PhysicalDbInstance> instances = getRWDbInstances(master == null);
        if (instances.size() == 0) {
            throw new IOException("the dbGroup[" + groupName + "] doesn't contain active dbInstance.");
        }

        if (localRead) {
            PhysicalDbInstance selectInstance = localReadLoadBalancer.select(instances);
            selectInstance.incrementReadCount();
            if (selectInstance.isAlive()) {
                if (LOGGER.isDebugEnabled()) {
                    LOGGER.debug("select {}", selectInstance);
                }
                return selectInstance;
            } else {
                reportError(selectInstance);
                return selectInstance;
            }
        }
        PhysicalDbInstance selectInstance = loadBalancer.select(instances);
        selectInstance.incrementReadCount();
        if (selectInstance.isAlive()) {
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("select {}", selectInstance);
            }
            return selectInstance;
        } else {
            reportError(selectInstance);
            return selectInstance;
        }
    }

    private List<PhysicalDbInstance> getRWDbInstances(boolean includeWrite) {
        ArrayList<PhysicalDbInstance> okSources = new ArrayList<>(allSourceMap.values().size());
        for (PhysicalDbInstance ds : allSourceMap.values()) {
            if (ds == writeDbInstance) {
                if (includeWrite && rwSplitMode == RW_SPLIT_ALL && writeDbInstance.isAlive()) {
                    okSources.add(ds);
                }
                continue;
            }

            if (ds.isAlive() && (!checkSlaveSynStatus() || ds.canSelectAsReadNode())) {
                if (ds.getLogCount() != 0) {
                    ds.setLogCount(0);
                }
                okSources.add(ds);
            } else {
                if (ds.isAlive()) {
                    if (ds.getLogCount() != 0) {
                        ds.setLogCount(0);
                    }
                    LOGGER.warn("can't select dbInstance[{}] as read node, please check delay with primary", ds);
                } else {
                    if (ds.getLogCount() < 10) {
                        ds.setLogCount(ds.getLogCount() + 1);
                        LOGGER.warn("can't select dbInstance[{}] as read node, please check the disabled and heartbeat status", ds);
                    }
                }
            }
        }
        if (okSources.size() == 0 && rwSplitMode == RW_SPLIT_ALL_SLAVES_MAY_MASTER && includeWrite) {
            if (writeDbInstance.isAlive()) {
                okSources.add(writeDbInstance);
            } else {
                LOGGER.warn("can't select dbInstance[{}] as read node, please check delay with primary", writeDbInstance);
            }
        }
        return okSources;
    }

    public RawJson disableHosts(String hostNames, boolean syncWriteConf) {
        String[] nameList = hostNames == null ? Arrays.copyOf(allSourceMap.keySet().toArray(), allSourceMap.keySet().toArray().length, String[].class) : hostNames.split(",");
        final ReentrantReadWriteLock lock = DbleServer.getInstance().getConfig().getLock();
        lock.readLock().lock();
        adjustLock.writeLock().lock();
        try {
            HaConfigManager.getInstance().updateDbGroupConf(createDisableSnapshot(this, nameList), syncWriteConf);
            for (String dsName : nameList) {
                allSourceMap.get(dsName).disable("ha command disable dbInstance");
            }
            return this.getClusterHaJson();
        } finally {
            lock.readLock().unlock();
            adjustLock.writeLock().unlock();
        }
    }

    private PhysicalDbGroup createDisableSnapshot(PhysicalDbGroup org, String[] nameList) {
        PhysicalDbGroup snapshot = new PhysicalDbGroup(org);
        for (String dsName : nameList) {
            PhysicalDbInstance dbInstance = snapshot.allSourceMap.get(dsName);
            dbInstance.setDisabled(true);
        }
        return snapshot;
    }

    public RawJson enableHosts(String hostNames, boolean syncWriteConf) {
        String[] nameList = hostNames == null ? Arrays.copyOf(allSourceMap.keySet().toArray(), allSourceMap.keySet().toArray().length, String[].class) : hostNames.split(",");
        final ReentrantReadWriteLock lock = DbleServer.getInstance().getConfig().getLock();
        lock.readLock().lock();
        adjustLock.writeLock().lock();
        try {

            HaConfigManager.getInstance().updateDbGroupConf(createEnableSnapshot(this, nameList), syncWriteConf);

            for (String dsName : nameList) {
                allSourceMap.get(dsName).enable();
            }
            return this.getClusterHaJson();
        } finally {
            lock.readLock().unlock();
            adjustLock.writeLock().unlock();
        }
    }

    private PhysicalDbGroup createEnableSnapshot(PhysicalDbGroup org, String[] nameList) {
        PhysicalDbGroup snapshot = new PhysicalDbGroup(org);
        for (String dsName : nameList) {
            PhysicalDbInstance dbInstance = snapshot.allSourceMap.get(dsName);
            dbInstance.setDisabled(false);
        }
        return snapshot;
    }

    public RawJson switchMaster(String writeHost, boolean syncWriteConf) {
        final ReentrantReadWriteLock lock = DbleServer.getInstance().getConfig().getLock();
        lock.readLock().lock();
        adjustLock.writeLock().lock();
        try {
            HaConfigManager.getInstance().updateDbGroupConf(createSwitchSnapshot(writeHost), syncWriteConf);

            PhysicalDbInstance newWriteHost = allSourceMap.get(writeHost);
            writeDbInstance.setReadInstance(true);
            //close all old master connection ,so that new writeDirectly query would not put into the old writeHost
            writeDbInstance.closeAllConnection("ha command switch dbInstance");
            if (!newWriteHost.isDisabled()) {
                GetAndSyncDbInstanceKeyVariables task = new GetAndSyncDbInstanceKeyVariables(newWriteHost, true);
                KeyVariables variables = task.call();
                if (variables != null) {
                    newWriteHost.setReadOnly(variables.isReadOnly());
                } else {
                    LOGGER.warn(" GetAndSyncDbInstanceKeyVariables failed, set new Primary dbInstance ReadOnly");
                    newWriteHost.setReadOnly(true);
                }
            }
            newWriteHost.setReadInstance(false);
            String oldWriteInstance = writeDbInstance.getName();
            writeDbInstance = newWriteHost;
            newWriteHost.start("switch master from " + oldWriteInstance + " to the instance", false);
            return this.getClusterHaJson();
        } catch (Exception e) {
            LOGGER.warn("switchMaster Exception ", e);
            throw e;
        } finally {
            lock.readLock().unlock();
            adjustLock.writeLock().unlock();
        }
    }

    private PhysicalDbGroup createSwitchSnapshot(String writeHost) {
        PhysicalDbGroup snapshot = new PhysicalDbGroup(this);
        PhysicalDbInstance newWriteHost = snapshot.allSourceMap.get(writeHost);
        snapshot.writeDbInstance.setReadInstance(true);
        newWriteHost.setReadInstance(false);
        snapshot.writeDbInstance = newWriteHost;
        return snapshot;
    }


    public void changeIntoLatestStatus(RawJson jsonStatus) {
        final ReentrantReadWriteLock lock = DbleServer.getInstance().getConfig().getLock();
        lock.readLock().lock();
        adjustLock.writeLock().lock();
        try {
            JsonObject jsonObj = jsonStatus.getJsonObject();
            JsonProcessBase base = new JsonProcessBase();
            Type parseType = new TypeToken<List<DbInstanceStatus>>() {
            }.getType();
            List<DbInstanceStatus> list = base.toBeanformJson(jsonObj.get(JSON_LIST).toString(), parseType);
            for (DbInstanceStatus status : list) {
                PhysicalDbInstance dbInstance = allSourceMap.get(status.getName());
                if (dbInstance != null) {
                    if (status.isDisable()) {
                        //clear old resource
                        dbInstance.disable("ha command disable dbInstance");
                    } else {
                        //change dbInstance from disable to enable ,start heartbeat
                        dbInstance.enable();
                    }
                    if (status.isPrimary() && dbInstance != writeDbInstance) {
                        writeDbInstance.setReadInstance(true);
                        writeDbInstance.closeAllConnection("ha command switch dbInstance");
                        dbInstance.setReadInstance(false);
                        writeDbInstance = dbInstance;
                    }
                } else {
                    LOGGER.warn("Can match dbInstance " + status.getName() + ".Check for the config file please");
                }
            }
            HaConfigManager.getInstance().updateDbGroupConf(this, false);
        } catch (Exception e) {
            LOGGER.warn("changeIntoLatestStatus Exception ", e);
            throw e;
        } finally {
            lock.readLock().unlock();
            adjustLock.writeLock().unlock();
        }
    }

    public RawJson getClusterHaJson() {
        JsonObjectWriter jsonObject = new JsonObjectWriter();
        jsonObject.addProperty(JSON_NAME, this.getGroupName());
        List<DbInstanceStatus> list = new ArrayList<>();
        for (PhysicalDbInstance phys : allSourceMap.values()) {
            list.add(new DbInstanceStatus(phys.getName(), phys.isDisabled(), !phys.isReadInstance()));
        }
        Gson gson = JsonFactory.getJson();
        jsonObject.add(JSON_LIST, gson.toJsonTree(list));
        return RawJson.of(jsonObject);
    }

    public boolean checkInstanceExist(String instanceName) {
        //add check for subHostName
        if (instanceName != null) {
            for (String dn : instanceName.split(",")) {
                boolean find = false;
                for (PhysicalDbInstance pds : this.allSourceMap.values()) {
                    if (pds.getName().equals(dn)) {
                        find = true;
                        break;
                    }
                }
                if (!find) {
                    return false;
                }
            }
        }
        return true;
    }

    private void reportDisableError(PhysicalDbInstance ins) throws IOException {
        final DbInstanceConfig config = ins.getConfig();
        String disableError = "the dbInstance[" + config.getUrl() + "] is disable. Please check the dbInstance disable status";
        LOGGER.warn(disableError);
        Map<String, String> labels = AlertUtil.genSingleLabel("dbInstance", dbGroupConfig.getName() + "-" + config.getInstanceName());
        AlertUtil.alert(AlarmCode.DB_INSTANCE_CAN_NOT_REACH, Alert.AlertLevel.WARN, disableError, "mysql", config.getId(), labels);
        throw new IOException(disableError);
    }

    private void reportFakeNodeError(PhysicalDbInstance ins) throws IOException {
        final DbInstanceConfig config = ins.getConfig();
        String fakeNodeError = "the dbInstance[" + config.getUrl() + "] is fake node. Please check the dbInstance whether or not it is used";
        LOGGER.warn(fakeNodeError);
        Map<String, String> labels = AlertUtil.genSingleLabel("dbInstance", dbGroupConfig.getName() + "-" + config.getInstanceName());
        AlertUtil.alert(AlarmCode.DB_INSTANCE_CAN_NOT_REACH, Alert.AlertLevel.WARN, fakeNodeError, "mysql", config.getId(), labels);
        throw new IOException(fakeNodeError);
    }

    private void reportHeartbeatError(PhysicalDbInstance ins) throws IOException {
        final DbInstanceConfig config = ins.getConfig();
        String heartbeatError = "the dbInstance[" + config.getUrl() + "] can't reach. Please check the dbInstance is accessible";
        if (dbGroupConfig.isShowSlaveSql()) {
            heartbeatError += " and the privileges of user is sufficient (NOTE:heartbeat[show slave status] need grant the SUPER or REPLICATION CLIENT privilege(s) to db user,and then restart the dble or fresh conn).";
        }
        LOGGER.warn(heartbeatError);
        Map<String, String> labels = AlertUtil.genSingleLabel("dbInstance", dbGroupConfig.getName() + "-" + config.getInstanceName());
        AlertUtil.alert(AlarmCode.DB_INSTANCE_CAN_NOT_REACH, Alert.AlertLevel.WARN, heartbeatError, "mysql", config.getId(), labels);
        throw new IOException(heartbeatError);
    }

    private void reportError(PhysicalDbInstance dbInstance) throws IOException {
        if (dbInstance.isFakeNode()) {
            reportFakeNodeError(dbInstance);
        } else if (dbInstance.isDisabled()) {
            reportDisableError(dbInstance);
        } else {
            reportHeartbeatError(dbInstance);
        }
    }

    public void setState(Integer state) {
        this.state = state;
    }

    public Integer getState() {
        return state;
    }

    public int getBindingCount() {
        return rwSplitSessionSet.size();
    }

    public boolean bindRwSplitSession(RWSplitNonBlockingSession session) {
        return this.rwSplitSessionSet.add(session);
    }

    public boolean unBindRwSplitSession(RWSplitNonBlockingSession session) {
        return this.rwSplitSessionSet.remove(session);
    }

    public void setWriteDbInstance(PhysicalDbInstance writeDbInstance) {
        this.writeDbInstance = writeDbInstance;
        if (writeDbInstance == null) {
            this.writeInstanceList = Lists.newArrayList();
        } else {
            this.writeInstanceList = Collections.singletonList(writeDbInstance);
        }
    }

    public void setDbInstance(PhysicalDbInstance dbInstance) {
        dbInstance.setDbGroup(this);
        if (dbInstance.getConfig().isPrimary()) {
            setWriteDbInstance(dbInstance);
            this.dbGroupConfig.setWriteInstanceConfig(dbInstance.getConfig());
        } else {
            this.dbGroupConfig.addReadInstance(dbInstance.getConfig());
        }
        this.allSourceMap.put(dbInstance.getName(), dbInstance);
    }

    public boolean equalsBaseInfo(PhysicalDbGroup pool) {
        return pool.dbGroupConfig.equalsBaseInfo(this.dbGroupConfig) &&
                pool.rwSplitMode == this.rwSplitMode &&
                pool.getGroupName().equals(this.groupName) &&
                pool.isUseless() == this.isUseless();
    }

    public boolean equalsForConnectionPool(PhysicalDbGroup pool) {
        boolean rwSplitModeFlag1 = pool.getDbGroupConfig().getRwSplitMode() != 0 && this.dbGroupConfig.getRwSplitMode() != 0;
        boolean rwSplitModeFlag2 = pool.getDbGroupConfig().getRwSplitMode() == 0 && this.dbGroupConfig.getRwSplitMode() == 0;
        return (rwSplitModeFlag1 || rwSplitModeFlag2) && pool.isUseless() == this.isUseless();
    }

    public boolean equalsForHeartbeat(PhysicalDbGroup pool) {
        return pool.getDbGroupConfig().getHeartbeatSQL().equals(this.dbGroupConfig.getHeartbeatSQL()) &&
                pool.getDbGroupConfig().getHeartbeatTimeout() == this.dbGroupConfig.getHeartbeatTimeout() &&
                pool.getDbGroupConfig().getErrorRetryCount() == this.dbGroupConfig.getErrorRetryCount() &&
                pool.getDbGroupConfig().getKeepAlive() == this.getDbGroupConfig().getKeepAlive();
    }


    public void copyBaseInfo(PhysicalDbGroup physicalDbGroup) {
        this.dbGroupConfig = physicalDbGroup.dbGroupConfig;
        this.groupName = physicalDbGroup.groupName;
        this.rwSplitMode = physicalDbGroup.rwSplitMode;
        this.schemas = physicalDbGroup.schemas;
        this.rwSplitUseless = physicalDbGroup.rwSplitUseless;
        this.shardingUseless = physicalDbGroup.shardingUseless;
        for (PhysicalDbInstance dbInstance : this.allSourceMap.values()) {
            dbInstance.setDbGroupConfig(physicalDbGroup.dbGroupConfig);
        }
    }

    @Override
    public String toString() {
        return "PhysicalDbGroup{" +
                "groupName='" + groupName + '\'' +
                ", dbGroupConfig=" + dbGroupConfig +
                ", writeDbInstance=" + writeDbInstance +
                ", allSourceMap=" + allSourceMap +
                ", rwSplitMode=" + rwSplitMode +
                ", schemas=" + schemas +
                ", shardingUseless=" + shardingUseless +
                ", rwSplitUseless=" + rwSplitUseless +
                '}';
    }
}
