[opt](blacklist) Backend should not be added to blacklist easily

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -1954,6 +1954,20 @@ public class Config extends ConfigBase {
  @ConfField(mutable = true, masterOnly = false)
  public static boolean disable_backend_black_list = false;
 
+ @ConfField(mutable = true, masterOnly = false, description = {
+ "If a backend is tried to be added to black list do_add_backend_black_list_threshold_count times "
+ + "in do_add_backend_black_list_threshold_seconds, it will be added to black list."})
+ public static long do_add_backend_black_list_threshold_count = 10;
+
+ @ConfField(mutable = true, masterOnly = false, description = {
+ "If a backend is tried to be added to black list do_add_backend_black_list_threshold_count times "
+ + "in do_add_backend_black_list_threshold_seconds, it will be added to black list."})
+ public static long do_add_backend_black_list_threshold_seconds = 30;
+
+ @ConfField(mutable = true, masterOnly = false, description = {
+ "Backend will stay in black list for this time after it is added to black list."})
+ public static long stay_in_backend_black_list_threshold_seconds = 60;
+
  /**
  * Maximum backend heartbeat failure tolerance count.
  * Default is 1, which means if 1 heart failed, the backend will be marked as dead.

fe/fe-core/src/main/java/org/apache/doris/qe/SimpleScheduler.java

@@ -19,37 +19,185 @@
 
 import org.apache.doris.catalog.Env;
 import org.apache.doris.common.Config;
-import org.apache.doris.common.Pair;
 import org.apache.doris.common.Reference;
 import org.apache.doris.common.UserException;
 import org.apache.doris.system.Backend;
 import org.apache.doris.system.SystemInfoService;
 import org.apache.doris.thrift.TNetworkAddress;
 import org.apache.doris.thrift.TScanRangeLocation;
 
+import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.stream.Collectors;
 
 public class SimpleScheduler {
  private static final Logger LOG = LogManager.getLogger(SimpleScheduler.class);
 
+ public static class BlackListInfo {
+ public BlackListInfo() {}
+
+ private Lock lock = new ReentrantLock();
+ // Record the reason why this backend is added to black list, will be updated only once.
+ private String reasonForBlackList = "";
+ // Record the first time this backend is tried to be added to black list.
+ private Long firstRecordBlackTimestampMs = 0L;
+ // Record the last time this backend is tried to be added to black list.
+ private Long lastRecordBlackTimestampMs = 0L;
+ // Record the timestamp this backend is really regarded as blacked.
+ private Long lastBlackTimestampMs = 0L;
+ // Record the count of this backend is tried to be added to black list.
+ private Long recordBlackListCount = 0L;
+ // Record the backend id
+ private Long backendID = 0L;
+
+ // Try to add this backend to black list, backend is not really added to black list until
+ // condition in shouldBeBlackListed is met.
+ public void tryAddBlackList(String reason) {
+ lock.lock();
+ try {
+ recordAddBlackList(reason);
+ if (shouldBeBlackListed()) {
+ doAddBlackList();
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ // Just update the fileds.
+ private void recordAddBlackList(String reason) {
+ if (firstRecordBlackTimestampMs <= 0) {
+ firstRecordBlackTimestampMs = System.currentTimeMillis();
+ }
+
+ lastRecordBlackTimestampMs = System.currentTimeMillis();
+
+ // Restart the counter if the time interval is too long
+ if (lastRecordBlackTimestampMs - firstRecordBlackTimestampMs
+ >= Config.do_add_backend_black_list_threshold_seconds * 1000) {
+ firstRecordBlackTimestampMs = lastRecordBlackTimestampMs;
+ recordBlackListCount = 0L;
+ }
+
+ recordBlackListCount++;
+
+ if (Strings.isNullOrEmpty(reasonForBlackList) && !Strings.isNullOrEmpty(reason)) {
+ reasonForBlackList = reason;
+ }
+ }
+
+ private boolean shouldBeBlackListed() {
+ if (lastRecordBlackTimestampMs <= 0 || firstRecordBlackTimestampMs <= 0) {
+ return false;
+ }
+
+ if (recordBlackListCount < Config.do_add_backend_black_list_threshold_count) {
+ return false;
+ }
+
+ if (lastRecordBlackTimestampMs - firstRecordBlackTimestampMs
+ >= Config.do_add_backend_black_list_threshold_seconds * 1000) {
+ return false;
+ }
+
+ return true;
+ }
+
+ private void doAddBlackList() {
+ lastBlackTimestampMs = System.currentTimeMillis();
+ Exception e = new Exception();
+ String stack = org.apache.commons.lang3.exception.ExceptionUtils.getStackTrace(e);
+ LOG.warn("Backend is added to black list.\nInformation:\n{}\nStack:\n{}", toString(), stack);
+ }
+
+ public boolean shouldBeRemoved() {
+ lock.lock();
+ try {
+ if (lastBlackTimestampMs <= 0) {
+ return false;
+ }
+
+ Long currentTimeStamp = System.currentTimeMillis();
+ // If this backend has not been recorded as black for more than 10 secs, then regard it as normal
+ if (currentTimeStamp - lastBlackTimestampMs
+ >= Config.stay_in_backend_black_list_threshold_seconds * 1000) {
+ return true;
+ } else {
+ return false;
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ public boolean isBlacked() {
+ lock.lock();
+ try {
+ return lastBlackTimestampMs > 0;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ public String getReasonForBlackList() {
+ lock.lock();
+ try {
+ return reasonForBlackList;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ public String toString() {
+ StringBuilder sb = new StringBuilder();
+ lock.lock();
+ // Convert to human readable time
+ LocalDateTime firstRecordBlackTimes = LocalDateTime.ofInstant(
+ Instant.ofEpochMilli(firstRecordBlackTimestampMs), ZoneId.systemDefault());
+ LocalDateTime lastRecordBlackTimes = LocalDateTime.ofInstant(
+ Instant.ofEpochMilli(lastRecordBlackTimestampMs), ZoneId.systemDefault());
+ LocalDateTime lastBlackTimes = LocalDateTime.ofInstant(
+ Instant.ofEpochMilli(lastBlackTimestampMs), ZoneId.systemDefault());
+ try {
+ sb.append("\nbackendID: ").append(backendID).append("\n");
+ sb.append("reasonForBlackList: ").append(reasonForBlackList).append("\n");
+ sb.append("firstRecordBlackTimestampMs: ").append(firstRecordBlackTimes).append("\n");
+ sb.append("lastRecordBlackTimestampMs: ").append(lastRecordBlackTimes).append("\n");
+ sb.append("lastBlackTimestampMs: ").append(lastBlackTimes).append("\n");
+ sb.append("recordBlackListCount: ").append(recordBlackListCount).append("\n");
+ sb.append("Config.do_add_backend_black_list_threshold_seconds: ")
+ .append(Config.do_add_backend_black_list_threshold_seconds).append("\n");
+ sb.append("Config.stay_in_backend_black_list_threshold_seconds: ")
+ .append(Config.stay_in_backend_black_list_threshold_seconds).append("\n");
+ return sb.toString();
+ } finally {
+ lock.unlock();
+ }
+ }
+ }
+
  private static AtomicLong nextId = new AtomicLong(0);
 
  // backend id -> (try time, reason)
  // There will be multi threads to read and modify this map.
  // But only one thread (UpdateBlacklistThread) will modify the `Pair`.
  // So using concurrent map is enough
- private static Map<Long, Pair<Integer, String>> blacklistBackends = Maps.newConcurrentMap();
+ private static Map<Long, BlackListInfo> blacklistBackends = Maps.newConcurrentMap();
  private static UpdateBlacklistThread updateBlacklistThread;
 
  public static void init() {
@@ -68,6 +216,7 @@ public static TNetworkAddress getHost(long backendId,
  if (LOG.isDebugEnabled()) {
  LOG.debug("getHost backendID={}, backendSize={}", backendId, backends.size());
  }
+
  Backend backend = backends.get(backendId);
 
  if (isAvailable(backend)) {
@@ -154,13 +303,13 @@ private static String getBackendErrorMsg(List<Long> backendIds, ImmutableMap<Lon
  for (int i = 0; i < backendIds.size() && i < limit; i++) {
  long beId = backendIds.get(i);
  Backend be = backends.get(beId);
+ BlackListInfo blackListInfo = blacklistBackends.get(beId);
  if (be == null) {
  res.add(beId + ": not exist");
  } else if (!be.isAlive()) {
  res.add(beId + ": not alive");
- } else if (blacklistBackends.containsKey(beId)) {
- Pair<Integer, String> pair = blacklistBackends.get(beId);
- res.add(beId + ": in black list(" + (pair == null ? "unknown" : pair.second) + ")");
+ } else if (blackListInfo != null && blackListInfo.isBlacked()) {
+ res.add(beId + ": in black list(" + blackListInfo.getReasonForBlackList() + ")");
  } else if (!be.isQueryAvailable()) {
  res.add(beId + ": disable query");
  } else {
@@ -177,12 +326,26 @@ public static void addToBlacklist(Long backendID, String reason) {
  return;
  }
 
- blacklistBackends.put(backendID, Pair.of(Config.blacklist_duration_second + 1, reason));
- LOG.warn("add backend {} to black list. reason: {}", backendID, reason);
+ BlackListInfo blackListInfo = blacklistBackends.putIfAbsent(backendID, new BlackListInfo());
+ if (blackListInfo == null) {
+ blackListInfo = blacklistBackends.get(backendID);
+ }
+
+ blackListInfo.tryAddBlackList(reason);
  }
 
  public static boolean isAvailable(Backend backend) {
- return (backend != null && backend.isQueryAvailable() && !blacklistBackends.containsKey(backend.getId()));
+ if (backend == null) {
+ return false;
+ }
+ if (!backend.isQueryAvailable()) {
+ return false;
+ }
+ BlackListInfo blackListInfo = blacklistBackends.get(backend.getId());
+ if (blackListInfo != null && blackListInfo.isBlacked()) {
+ return false;
+ }
+ return true;
  }
 
  private static class UpdateBlacklistThread implements Runnable {
@@ -208,25 +371,26 @@ public void run() {
  Thread.sleep(1000L);
  SystemInfoService clusterInfoService = Env.getCurrentSystemInfo();
 
- Iterator<Map.Entry<Long, Pair<Integer, String>>> iterator = blacklistBackends.entrySet().iterator();
+ Iterator<Map.Entry<Long, SimpleScheduler.BlackListInfo>>
+ iterator = blacklistBackends.entrySet().iterator();
+
  while (iterator.hasNext()) {
- Map.Entry<Long, Pair<Integer, String>> entry = iterator.next();
+ Map.Entry<Long, SimpleScheduler.BlackListInfo> entry = iterator.next();
  Long backendId = entry.getKey();
-
+ Backend backend = clusterInfoService.getBackend(backendId);
  // remove from blacklist if backend does not exist anymore
- if (clusterInfoService.getBackend(backendId) == null) {
+ if (backend == null) {
  iterator.remove();
  LOG.info("remove backend {} from black list because it does not exist", backendId);
  } else {
- // 3. max try time is reach
- entry.getValue().first = entry.getValue().first - 1;
- if (entry.getValue().first <= 0) {
+ BlackListInfo blackListInfo = entry.getValue();
+ if (backend.isAlive() || blackListInfo.shouldBeRemoved()) {
  iterator.remove();
- LOG.warn("remove backend {} from black list. reach max try time", backendId);
+ LOG.info("remove backend {} from black list. backend is alive: {}",
+ backendId, backend.isAlive());
  } else {
  if (LOG.isDebugEnabled()) {
- LOG.debug("blacklistBackends backendID={} retryTimes={}",
- backendId, entry.getValue().first);
+ LOG.debug("blacklistBackends {}", blackListInfo.toString());
  }
  }
  }

fe/fe-core/src/test/java/org/apache/doris/qe/SimpleSchedulerTest.java

@@ -17,10 +17,12 @@
 
 package org.apache.doris.qe;
 
+import org.apache.doris.catalog.Env;
 import org.apache.doris.common.Config;
 import org.apache.doris.common.Reference;
 import org.apache.doris.common.UserException;
 import org.apache.doris.system.Backend;
+import org.apache.doris.system.SystemInfoService;
 import org.apache.doris.thrift.TNetworkAddress;
 import org.apache.doris.thrift.TScanRangeLocation;
 
@@ -108,8 +110,8 @@ public void run() {
  @Override
  public void run() {
  try {
- Set<String> resBackends = Sets.newHashSet();
  long start = System.currentTimeMillis();
+ Set<String> resBackends = Sets.newHashSet();
  for (int i = 0; i < 1000; i++) {
  TNetworkAddress address = SimpleScheduler.getHost(backends, ref);
  Assert.assertNotNull(address);
@@ -126,20 +128,29 @@ public void run() {
  Thread t3 = new Thread(new Runnable() {
  @Override
  public void run() {
- SimpleScheduler.addToBlacklist(be1.getId(), "test");
+ for (int i = 0; i < 100; i++) {
+ SimpleScheduler.addToBlacklist(be1.getId(), "test");
+ }
  }
  });
 
+ SystemInfoService clusterInfoService = Env.getCurrentSystemInfo();
+ be1.setAlive(false);
+ clusterInfoService.addBackend(be1);
  t3.start();
+ t3.join();
+
+ // When t1 and t2 starts, be1 has already been in blacklist.
  t1.start();
  t2.start();
 
  t1.join();
  t2.join();
- t3.join();
 
  Assert.assertFalse(SimpleScheduler.isAvailable(be1));
- Thread.sleep((Config.heartbeat_interval_second + 5) * 1000);
+ be1.setAlive(true);
+ // Sleep 5s so that UpdateBlacklistThread will remove be1 from blacklist
+ Thread.sleep(1000 * 5L);
  Assert.assertTrue(SimpleScheduler.isAvailable(be1));
  }
 
@@ -182,11 +193,14 @@ public void testGetHostAbnormal() throws UserException, InterruptedException {
  scanRangeLocation5.setBackendId(be5.getId());
  locations.add(scanRangeLocation5);
 
- SimpleScheduler.addToBlacklist(be1.getId(), "test");
- SimpleScheduler.addToBlacklist(be2.getId(), "test");
- SimpleScheduler.addToBlacklist(be3.getId(), "test");
- SimpleScheduler.addToBlacklist(be4.getId(), "test");
- SimpleScheduler.addToBlacklist(be5.getId(), "test");
+ for (int i = 0; i <= Config.do_add_backend_black_list_threshold_count; i++) {
+ SimpleScheduler.addToBlacklist(be1.getId(), "test");
+ SimpleScheduler.addToBlacklist(be2.getId(), "test");
+ SimpleScheduler.addToBlacklist(be3.getId(), "test");
+ SimpleScheduler.addToBlacklist(be4.getId(), "test");
+ SimpleScheduler.addToBlacklist(be5.getId(), "test");
+ }
+
  try {
  SimpleScheduler.getHost(locations.get(0).backend_id, locations, backends, ref);
  Assert.fail();