content: /*
   * Copyright (C) 2016-2023 ActionTech.
   * License: http://www.gnu.org/licenses/gpl.html GPL version 2 or higher.
   */
  
  package com.actiontech.dble.services.manager.response;
  
  import com.actiontech.dble.DbleServer;
  import com.actiontech.dble.backend.datasource.PhysicalDbGroup;
  import com.actiontech.dble.backend.datasource.PhysicalDbInstance;
  import com.actiontech.dble.backend.datasource.ShardingNode;
  import com.actiontech.dble.backend.mysql.PacketUtil;
  import com.actiontech.dble.cluster.ClusterHelper;
  import com.actiontech.dble.cluster.DistributeLock;
  import com.actiontech.dble.cluster.logic.ClusterLogic;
  import com.actiontech.dble.cluster.logic.ClusterOperation;
  import com.actiontech.dble.cluster.path.ClusterMetaUtil;
  import com.actiontech.dble.cluster.path.PathMeta;
  import com.actiontech.dble.cluster.values.Empty;
  import com.actiontech.dble.cluster.values.FeedBackType;
  import com.actiontech.dble.cluster.values.OnlineType;
  import com.actiontech.dble.config.ErrorCode;
  import com.actiontech.dble.config.Fields;
  import com.actiontech.dble.config.model.ClusterConfig;
  import com.actiontech.dble.net.IOProcessor;
  import com.actiontech.dble.net.connection.FrontendConnection;
  import com.actiontech.dble.net.mysql.*;
  import com.actiontech.dble.server.NonBlockingSession;
  import com.actiontech.dble.services.manager.ManagerService;
  import com.actiontech.dble.services.mysqlsharding.ShardingService;
  import com.actiontech.dble.services.rwsplit.RWSplitService;
  import com.actiontech.dble.sqlengine.OneRawSQLQueryResultHandler;
  import com.actiontech.dble.sqlengine.SQLJob;
  import com.actiontech.dble.sqlengine.SQLQueryResult;
  import com.actiontech.dble.sqlengine.SQLQueryResultListener;
  import com.actiontech.dble.util.StringUtil;
  import com.actiontech.dble.util.TimeUtil;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.nio.ByteBuffer;
  import java.util.ArrayList;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  import java.util.concurrent.CopyOnWriteArrayList;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.atomic.AtomicInteger;
  import java.util.concurrent.locks.LockSupport;
  import java.util.stream.Collectors;
  
  public final class ShowBinlogStatus {
      private ShowBinlogStatus() {
      }
  
      private static final int FIELD_COUNT = 6;
      private static final ResultSetHeaderPacket HEADER = PacketUtil.getHeader(FIELD_COUNT);
      private static final FieldPacket[] FIELDS_PACKET = new FieldPacket[FIELD_COUNT];
      private static final EOFPacket EOF = new EOFPacket();
      private static final String[] FIELDS = new String[]{"File", "Position", "Binlog_Do_DB", "Binlog_Ignore_DB", "Executed_Gtid_Set"};
  
      static {
          int i = 0;
          byte packetId = 0;
          HEADER.setPacketId(++packetId);
          FIELDS_PACKET[i] = PacketUtil.getField("Url", Fields.FIELD_TYPE_VAR_STRING);
          FIELDS_PACKET[i++].setPacketId(++packetId);
          for (String field : FIELDS) {
              FIELDS_PACKET[i] = PacketUtil.getField(field, Fields.FIELD_TYPE_VAR_STRING);
              FIELDS_PACKET[i++].setPacketId(++packetId);
          }
          EOF.setPacketId(++packetId);
      }
  
      private static final String SHOW_BINLOG_QUERY = "SHOW MASTER STATUS";
      private static Logger logger = LoggerFactory.getLogger(ShowBinlogStatus.class);
      private static AtomicInteger sourceCount;
      private static List<RowDataPacket> rows;
      private static volatile String errMsg = null;
  
      public static void execute(ManagerService service) {
          long timeout = ClusterConfig.getInstance().getShowBinlogStatusTimeout();
          if (ClusterConfig.getInstance().isClusterEnable()) {
              showBinlogWithCluster(service, timeout);
          } else {
              if (!DbleServer.getInstance().getBackupLocked().compareAndSet(false, true)) {
                  service.writeErrMessage(ErrorCode.ER_UNKNOWN_ERROR, "There is another command is showing BinlogStatus");
              } else {
                  try {
                      errMsg = null;
                      if (waitAllSession(service, timeout, TimeUtil.currentTimeMillis())) {
                          getQueryResult(service.getCharset().getResults());
                      }
                      writeResponse(service);
                  } finally {
                      DbleServer.getInstance().getBackupLocked().compareAndSet(true, false);
                  }
              }
          }
  
      }
  
  
      private static void showBinlogWithCluster(ManagerService service, long timeout) {
          //step 1 get the distributeLock
          final ClusterHelper clusterHelper = ClusterHelper.getInstance(ClusterOperation.BINGLOG);
          DistributeLock distributeLock = clusterHelper.createDistributeLock(ClusterMetaUtil.getBinlogPauseLockPath());
          try {
              if (!distributeLock.acquire()) {
                  errMsg = "There is another command is showing BinlogStatus";
                  return;
              }
              try {
                  //step 2 try to lock all the commit flag in server
                  if (!DbleServer.getInstance().getBackupLocked().compareAndSet(false, true)) {
                      errMsg = "There is another command is showing BinlogStatus";
                  } else {
                      //step 3 wait til other dbles to feedback the ucore flag
                      errMsg = null;
                      long beginTime = TimeUtil.currentTimeMillis();
                      boolean isPaused = waitAllSession(service, timeout, beginTime);
                      if (!isPaused) {
                          return;
                      }
                      //step 4 notify other dble to stop the commit & set self status
                      PathMeta<Empty> binlogStatusPath = ClusterMetaUtil.getBinlogPauseStatusPath();
                      clusterHelper.setKV(binlogStatusPath, new Empty());
                      clusterHelper.createSelfTempNode(binlogStatusPath.getPath(), FeedBackType.SUCCESS);
  
                      Map<String, OnlineType> expectedMap = ClusterHelper.getOnlineMap();
                      while (true) {
                          StringBuilder errorStringBuf = new StringBuilder();
                          if (ClusterLogic.forBinlog().checkResponseForOneTime(binlogStatusPath.getPath(), expectedMap, errorStringBuf)) {
                              errMsg = errorStringBuf.length() <= 0 ? null : errorStringBuf.toString();
                              break;
                          } else if (TimeUtil.currentTimeMillis() > beginTime + 2 * timeout) {
                              errMsg = "timeout while waiting for unfinished distributed transactions.";
                              logger.info(errMsg);
                              break;
                          }
                      }
  
                      // step 6 query for the GTID and write back to frontend connections
                      if (errMsg == null) {
                          getQueryResult(service.getCharset().getResults());
                      }
                      //step 7 delete the KVtree and notify the cluster
                      ClusterHelper.cleanPath(binlogStatusPath);
                  }
              } catch (Exception e) {
                  errMsg = e.getMessage();
                  logger.warn("catch Exception", e);
              } finally {
                  DbleServer.getInstance().getBackupLocked().compareAndSet(true, false);
                  distributeLock.release();
              }
          } catch (Exception e) {
              errMsg = e.getMessage();
          } finally {
              writeResponse(service);
          }
      }
  
  
      private static void writeResponse(ManagerService service) {
          if (errMsg == null) {
              ByteBuffer buffer = service.allocate();
              buffer = HEADER.write(buffer, service, true);
              for (FieldPacket field : FIELDS_PACKET) {
                  buffer = field.write(buffer, service, true);
              }
              buffer = EOF.write(buffer, service, true);
              byte packetId = EOF.getPacketId();
              for (RowDataPacket row : rows) {
                  row.setPacketId(++packetId);
                  buffer = row.write(buffer, service, true);
              }
              rows.clear();
              EOFRowPacket lastEof = new EOFRowPacket();
              lastEof.setPacketId(++packetId);
  
              lastEof.write(buffer, service);
          } else {
              service.writeErrMessage(ErrorCode.ER_UNKNOWN_ERROR, errMsg);
              errMsg = null;
          }
      }
  
      public static boolean waitAllSession() {
          logger.info("waiting all sessions of distributed transaction which are not finished.");
          long timeout = ClusterConfig.getInstance().getShowBinlogStatusTimeout();
          long beginTime = TimeUtil.currentTimeMillis();
          List<NonBlockingSession> fcList = getNeedWaitSession();
          while (!fcList.isEmpty()) {
              LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(10));
              fcList.removeIf(session -> !session.isNeedWaitFinished());
              if ((TimeUtil.currentTimeMillis() > beginTime + timeout)) {
                  logger.info("wait session finished timeout");
                  return false;
              }
          }
          logger.info("all sessions of distributed transaction  are paused.");
          return true;
      }
  
      private static boolean waitAllSession(ManagerService service, long timeout, long beginTime) {
          logger.info("waiting all sessions of distributed transaction which are not finished.");
          List<NonBlockingSession> fcList = getNeedWaitSession();
          while (!fcList.isEmpty()) {
              LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(10));
              fcList.removeIf(session -> !session.isNeedWaitFinished());
              if (service.getConnection().isClosed()) {
                  errMsg = "client closed while waiting for unfinished distributed transactions.";
                  logger.info(errMsg);
                  return false;
              }
              if (TimeUtil.currentTimeMillis() > beginTime + timeout) {
                  errMsg = "timeout while waiting for unfinished distributed transactions.";
                  logger.info(errMsg);
                  return false;
              }
          }
          logger.info("all sessions of distributed transaction  are paused.");
          return true;
      }
  
      private static List<NonBlockingSession> getNeedWaitSession() {
          List<NonBlockingSession> fcList = new ArrayList<>();
          for (IOProcessor process : DbleServer.getInstance().getFrontProcessors()) {
              for (FrontendConnection front : process.getFrontends().values()) {
                  if (front.isManager() || front.getService() instanceof RWSplitService) {
                      continue;
                  }
                  NonBlockingSession session = ((ShardingService) front.getService()).getSession2();
                  if (session.isNeedWaitFinished()) {
                      fcList.add(session);
                  }
              }
          }
          return fcList;
      }
  
      /**
       * getQueryResult: show master status
       *
       * @param charset
       */
      private static void getQueryResult(final String charset) {
          Map<String, ShardingNode> shardingNodes = DbleServer.getInstance().getConfig().getShardingNodes();
          Set<PhysicalDbGroup> dbGroupSet = shardingNodes.values().stream().map(ShardingNode::getDbGroup).collect(Collectors.toSet());
          sourceCount = new AtomicInteger(dbGroupSet.size());
          rows = new CopyOnWriteArrayList<>();
          for (PhysicalDbGroup pool : dbGroupSet) {
              //if WRITE_RANDOM_NODE ,may the binlog is not ready.
              final PhysicalDbInstance source = pool.getWriteDbInstance();
              OneRawSQLQueryResultHandler resultHandler = new OneRawSQLQueryResultHandler(FIELDS,
                      new SQLQueryResultListener<SQLQueryResult<Map<String, String>>>() {
                          @Override
                          public void onResult(SQLQueryResult<Map<String, String>> result) {
                              String url = source.getConfig().getUrl();
                              if (!result.isSuccess()) {
                                  errMsg = "Getting binlog status from this instance[" + url + "] is failed";
                              } else {
                                  rows.add(getRow(url, result.getResult(), charset));
                              }
                              sourceCount.decrementAndGet();
                          }
  
                      });
              SQLJob sqlJob = new SQLJob(SHOW_BINLOG_QUERY, null, resultHandler, source);
              sqlJob.run();
          }
          while (sourceCount.get() > 0) {
              LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(10));
          }
      }
  
      private static RowDataPacket getRow(String url, Map<String, String> result, String charset) {
          RowDataPacket row = new RowDataPacket(FIELD_COUNT);
          row.add(StringUtil.encode(url, charset));
          for (String field : FIELDS) {
              row.add(StringUtil.encode(result.get(field), charset));
          }
          return row;
      }
  }
  