[Feature] Support disk disable/decommission (part1) (#37134)

Signed-off-by: gengjun-git <[email protected]>

Author: gengjun-git

be/src/agent/heartbeat_server.cpp

@@ -109,6 +109,14 @@ void HeartbeatServer::heartbeat(THeartbeatResult& heartbeat_result, const TMaste
         // nothing to do
     }
 
+    if (master_info.__isset.disabled_disks) {
+        _olap_engine->disable_disks(master_info.disabled_disks);
+    }
+
+    if (master_info.__isset.decommissioned_disks) {
+        _olap_engine->decommission_disks(master_info.decommissioned_disks);
+    }
+
     static auto num_hardware_cores = static_cast<int32_t>(CpuInfo::num_cores());
     if (res.ok()) {
         heartbeat_result.backend_info.__set_be_port(config::be_port);

be/src/storage/data_dir.cpp

@@ -72,7 +72,6 @@ DataDir::DataDir(const std::string& path, TStorageMedium::type storage_medium, T
           _available_bytes(0),
           _disk_capacity_bytes(0),
           _storage_medium(storage_medium),
-          _is_used(false),
           _tablet_manager(tablet_manager),
           _txn_manager(txn_manager),
           _cluster_id_mgr(std::make_shared<ClusterIdMgr>(path)),
@@ -99,7 +98,7 @@ Status DataDir::init(bool read_only) {
     RETURN_IF_ERROR_WITH_WARN(_init_meta(read_only), "_init_meta failed");
     RETURN_IF_ERROR_WITH_WARN(init_persistent_index_dir(), "_init_persistent_index_dir failed");
 
-    _is_used = true;
+    _state = DiskState::ONLINE;
     return Status::OK();
 }
 
@@ -150,15 +149,26 @@ Status DataDir::set_cluster_id(int32_t cluster_id) {
 }
 
 void DataDir::health_check() {
+    const int retry_times = 10;
     // check disk
-    if (_is_used) {
-        Status res = _read_and_write_test_file();
-        if (!res.ok()) {
-            LOG(WARNING) << "store read/write test file occur IO Error. path=" << _path << ", res=" << res.to_string();
-            if (is_io_error(res)) {
-                _is_used = false;
+    if (_state != DiskState::DECOMMISSIONED && _state != DiskState::DISABLED) {
+        bool all_failed = true;
+        for (int i = 0; i < retry_times; i++) {
+            Status res = _read_and_write_test_file();
+            if (res.ok() || !is_io_error(res)) {
+                all_failed = false;
+                break;
+            } else {
+                LOG(WARNING) << "store read/write test file occur IO Error. path=" << _path
+                             << ", res=" << res.to_string();
             }
         }
+        if (all_failed) {
+            LOG(WARNING) << "store test failed " << retry_times << " times, set _state to OFFLINE. path=" << _path;
+            _state = DiskState::OFFLINE;
+        } else {
+            _state = DiskState::ONLINE;
+        }
     }
 }
 

be/src/storage/data_dir.h

@@ -56,6 +56,13 @@ class Tablet;
 class TabletManager;
 class TxnManager;
 
+enum DiskState {
+    ONLINE,
+    OFFLINE,       // detected by health_check, tablets on OFFLINE disk will be dropped.
+    DISABLED,      // set by user, tablets on DISABLED disk will be dropped.
+    DECOMMISSIONED // set by user, tablets on DECOMMISSIONED disk will be migrated to other disks.
+};
+
 // A DataDir used to manage data in same path.
 // Now, After DataDir was created, it will never be deleted for easy implementation.
 class DataDir {
@@ -72,8 +79,9 @@ class DataDir {
 
     const std::string& path() const { return _path; }
     int64_t path_hash() const { return _path_hash; }
-    bool is_used() const { return _is_used; }
-    void set_is_used(bool is_used) { _is_used = is_used; }
+    bool is_used() const { return _state == DiskState::ONLINE || _state == DiskState::DECOMMISSIONED; }
+    DiskState get_state() const { return _state; }
+    void set_state(DiskState state) { _state = state; }
     int32_t cluster_id() const { return _cluster_id_mgr->cluster_id(); }
 
     DataDirInfo get_dir_info() {
@@ -82,7 +90,7 @@ class DataDir {
         info.path_hash = _path_hash;
         info.disk_capacity = _disk_capacity_bytes;
         info.available = _available_bytes;
-        info.is_used = _is_used;
+        info.is_used = is_used();
         info.storage_medium = _storage_medium;
         return info;
     }
@@ -170,7 +178,7 @@ class DataDir {
     // the actual capacity of the disk of this data dir
     int64_t _disk_capacity_bytes;
     TStorageMedium::type _storage_medium;
-    bool _is_used;
+    DiskState _state;
 
     TabletManager* _tablet_manager;
     TxnManager* _txn_manager;

be/src/storage/storage_engine.cpp

@@ -473,7 +473,7 @@ std::vector<DataDir*> StorageEngine::get_stores_for_create_tablet(TStorageMedium
     {
         std::lock_guard<std::mutex> l(_store_lock);
         for (auto& it : _store_map) {
-            if (it.second->is_used()) {
+            if (it.second->get_state() == DiskState::ONLINE) {
                 if (_available_storage_medium_type_count == 1 || it.second->storage_medium() == storage_medium) {
                     if (!it.second->capacity_limit_reached(0)) {
                         stores.push_back(it.second);
@@ -1541,4 +1541,38 @@ void StorageEngine::clear_rowset_delta_column_group_cache(const Rowset& rowset)
     }());
 }
 
+void StorageEngine::disable_disks(const std::vector<string>& disabled_disks) {
+    std::lock_guard<std::mutex> l(_store_lock);
+
+    for (auto& it : _store_map) {
+        if (std::find(disabled_disks.begin(), disabled_disks.end(), it.second->path()) != disabled_disks.end()) {
+            it.second->set_state(DiskState::DISABLED);
+            LOG(INFO) << "disk " << it.second->path() << " is set to DISABLED";
+        } else {
+            if (it.second->get_state() == DiskState::DISABLED) {
+                it.second->set_state(DiskState::ONLINE);
+                LOG(INFO) << "disk " << it.second->path() << " is recovered to ONLINE, previous state is DISABLED";
+            }
+        }
+    }
+}
+
+void StorageEngine::decommission_disks(const std::vector<string>& decommission_disks) {
+    std::lock_guard<std::mutex> l(_store_lock);
+
+    for (auto& it : _store_map) {
+        if (std::find(decommission_disks.begin(), decommission_disks.end(), it.second->path()) !=
+            decommission_disks.end()) {
+            it.second->set_state(DiskState::DECOMMISSIONED);
+            LOG(INFO) << "disk " << it.second->path() << " is set to DECOMMISSIONED";
+        } else {
+            if (it.second->get_state() == DiskState::DECOMMISSIONED) {
+                it.second->set_state(DiskState::ONLINE);
+                LOG(INFO) << "disk " << it.second->path()
+                          << " is recovered to ONLINE, previous state is DECOMMISSIONED";
+            }
+        }
+    }
+}
+
 } // namespace starrocks

be/src/storage/storage_engine.h

@@ -282,6 +282,10 @@ class StorageEngine {
 
     void clear_rowset_delta_column_group_cache(const Rowset& rowset);
 
+    void disable_disks(const std::vector<string>& disabled_disks);
+
+    void decommission_disks(const std::vector<string>& decommissioned_disks);
+
     void wake_finish_publish_vesion_thread() {
         std::unique_lock<std::mutex> wl(_finish_publish_version_mutex);
         _finish_publish_version_cv.notify_one();

fe/fe-core/src/main/java/com/starrocks/catalog/DataProperty.java

@@ -91,7 +91,7 @@ public static DataProperty getInferredDefaultDataProperty() {
         for (Backend backend : backends) {
             if (backend.hasPathHash()) {
                 mediumSet.addAll(backend.getDisks().values().stream()
-                        .filter(v -> v.getState() == DiskInfo.DiskState.ONLINE)
+                        .filter(DiskInfo::canCreateTablet)
                         .map(DiskInfo::getStorageMedium).collect(Collectors.toSet()));
             }
         }

fe/fe-core/src/main/java/com/starrocks/catalog/DiskInfo.java

@@ -49,7 +49,14 @@
 public class DiskInfo implements Writable {
     public enum DiskState {
         ONLINE,
-        OFFLINE
+        // Reported from BE, if disk is detected as unavailable, state will be OFFLINE.
+        // Tablets on OFFLINE disk will be dropped.
+        OFFLINE,
+        // Set by user, tablets on DISABLED disk will be dropped.
+        DISABLED,
+        // Set by user, tablets on DECOMMISSIONED disk will be cloned to other backends.
+        // Before the decommission finish, the disk is still usable, i.e. can provide read and write capability.
+        DECOMMISSIONED
     }
 
     private static final Logger LOG = LogManager.getLogger(DiskInfo.class);
@@ -129,6 +136,14 @@ public DiskState getState() {
         return state;
     }
 
+    public boolean canReadWrite() {
+        return state == DiskState.ONLINE || state == DiskState.DECOMMISSIONED;
+    }
+
+    public boolean canCreateTablet() {
+        return state == DiskState.ONLINE;
+    }
+
     // return true if changed
     public boolean setState(DiskState state) {
         if (this.state != state) {

fe/fe-core/src/main/java/com/starrocks/catalog/LocalTablet.java

@@ -83,6 +83,7 @@ public enum TabletStatus {
         COLOCATE_MISMATCH, // replicas do not all locate in right colocate backends set.
         COLOCATE_REDUNDANT, // replicas match the colocate backends set, but redundant.
         NEED_FURTHER_REPAIR, // one of replicas need a definite repair.
+        DISK_MIGRATION, // The disk where the replica is located is decommissioned.
     }
 
     // Most read only accesses to replicas should acquire db lock, to prevent
@@ -586,7 +587,8 @@ private Pair<TabletStatus, TabletSchedCtx.Priority> getHealthStatusWithPriorityU
 
         int alive = 0;
         int aliveAndVersionComplete = 0;
-        int stable = 0;
+        int backendStable = 0;
+        int diskStable = 0;
 
         Replica needFurtherRepairReplica = null;
         Set<String> hosts = Sets.newHashSet();
@@ -617,7 +619,13 @@ private Pair<TabletStatus, TabletSchedCtx.Priority> getHealthStatusWithPriorityU
                 // this replica is alive, version complete, but backend is not available
                 continue;
             }
-            stable++;
+            backendStable++;
+
+            if (backend.isDiskDecommissioned(replica.getPathHash())) {
+                // disk in decommission state
+                continue;
+            }
+            diskStable++;
         }
 
         // 1. alive replicas are not enough
@@ -628,7 +636,7 @@ private Pair<TabletStatus, TabletSchedCtx.Priority> getHealthStatusWithPriorityU
         if (needRecoverWithEmptyTablet(systemInfoService)) {
             LOG.info("need to forcefully recover with empty tablet for {}, replica info:{}",
                     id, getReplicaInfos());
-            return createRedundantSchedCtx(TabletStatus.FORCE_REDUNDANT, TabletSchedCtx.Priority.VERY_HIGH,
+            return createRedundantSchedCtx(TabletStatus.FORCE_REDUNDANT, Priority.VERY_HIGH,
                     needFurtherRepairReplica);
         }
 
@@ -644,7 +652,7 @@ private Pair<TabletStatus, TabletSchedCtx.Priority> getHealthStatusWithPriorityU
             //    at least one backend for new replica.
             // 4. replicationNum > 1: if replication num is set to 1, do not delete any replica, for safety reason
             // For example: 3 replica, 3 be, one set bad, we need to forcefully delete one first
-            return createRedundantSchedCtx(TabletStatus.FORCE_REDUNDANT, TabletSchedCtx.Priority.VERY_HIGH,
+            return createRedundantSchedCtx(TabletStatus.FORCE_REDUNDANT, Priority.VERY_HIGH,
                     needFurtherRepairReplica);
         } else {
             List<Long> availableBEs = systemInfoService.getAvailableBackendIds();
@@ -654,26 +662,26 @@ private Pair<TabletStatus, TabletSchedCtx.Priority> getHealthStatusWithPriorityU
             // of load task won't be blocked either.
             if (availableBEs.size() > alive) {
                 if (alive < (replicationNum / 2) + 1) {
-                    return Pair.create(TabletStatus.REPLICA_MISSING, TabletSchedCtx.Priority.HIGH);
+                    return Pair.create(TabletStatus.REPLICA_MISSING, Priority.HIGH);
                 } else if (alive < replicationNum) {
-                    return Pair.create(TabletStatus.REPLICA_MISSING, TabletSchedCtx.Priority.NORMAL);
+                    return Pair.create(TabletStatus.REPLICA_MISSING, Priority.NORMAL);
                 }
             }
         }
 
         // 2. version complete replicas are not enough
         if (aliveAndVersionComplete < (replicationNum / 2) + 1) {
-            return Pair.create(TabletStatus.VERSION_INCOMPLETE, TabletSchedCtx.Priority.HIGH);
+            return Pair.create(TabletStatus.VERSION_INCOMPLETE, Priority.HIGH);
         } else if (aliveAndVersionComplete < replicationNum) {
-            return Pair.create(TabletStatus.VERSION_INCOMPLETE, TabletSchedCtx.Priority.NORMAL);
+            return Pair.create(TabletStatus.VERSION_INCOMPLETE, Priority.NORMAL);
         } else if (aliveAndVersionComplete > replicationNum) {
             // we set REDUNDANT as VERY_HIGH, because delete redundant replicas can free the space quickly.
-            return createRedundantSchedCtx(TabletStatus.REDUNDANT, TabletSchedCtx.Priority.VERY_HIGH,
+            return createRedundantSchedCtx(TabletStatus.REDUNDANT, Priority.VERY_HIGH,
                     needFurtherRepairReplica);
         }
 
         // 3. replica is under relocating
-        if (stable < replicationNum) {
+        if (backendStable < replicationNum) {
             List<Long> replicaBeIds = replicas.stream()
                     .map(Replica::getBackendId).collect(Collectors.toList());
             List<Long> availableBeIds = aliveBeIdsInCluster.stream()
@@ -683,25 +691,30 @@ private Pair<TabletStatus, TabletSchedCtx.Priority> getHealthStatusWithPriorityU
                     && availableBeIds.size() >= replicationNum
                     && replicationNum > 1) { // Doesn't have any BE that can be chosen to create a new replica
                 return createRedundantSchedCtx(TabletStatus.FORCE_REDUNDANT,
-                        stable < (replicationNum / 2) + 1 ? TabletSchedCtx.Priority.NORMAL :
-                                TabletSchedCtx.Priority.LOW, needFurtherRepairReplica);
+                        backendStable < (replicationNum / 2) + 1 ? Priority.NORMAL :
+                                Priority.LOW, needFurtherRepairReplica);
             }
-            if (stable < (replicationNum / 2) + 1) {
-                return Pair.create(TabletStatus.REPLICA_RELOCATING, TabletSchedCtx.Priority.NORMAL);
+            if (backendStable < (replicationNum / 2) + 1) {
+                return Pair.create(TabletStatus.REPLICA_RELOCATING, Priority.NORMAL);
             } else {
                 return Pair.create(TabletStatus.REPLICA_RELOCATING, Priority.LOW);
             }
         }
 
-        // 4. replica redundant
+        // 4. disk decommission
+        if (diskStable < replicationNum) {
+            return Pair.create(TabletStatus.DISK_MIGRATION, Priority.NORMAL);
+        }
+
+        // 5. replica redundant
         if (replicas.size() > replicationNum) {
             // we set REDUNDANT as VERY_HIGH, because delete redundant replicas can free the space quickly.
-            return createRedundantSchedCtx(TabletStatus.REDUNDANT, TabletSchedCtx.Priority.VERY_HIGH,
+            return createRedundantSchedCtx(TabletStatus.REDUNDANT, Priority.VERY_HIGH,
                     needFurtherRepairReplica);
         }
 
-        // 5. healthy
-        return Pair.create(TabletStatus.HEALTHY, TabletSchedCtx.Priority.NORMAL);
+        // 6. healthy
+        return Pair.create(TabletStatus.HEALTHY, Priority.NORMAL);
     }
 
     public TabletStatus getColocateHealthStatus(long visibleVersion,
@@ -752,6 +765,7 @@ && containsAnyHighPrioBackend(replicaBackendIds, Config.tablet_sched_colocate_ba
             }
         }
 
+        int diskStableCnt = 0;
         // 2. check version completeness
         for (Replica replica : replicas) {
             // do not check the replica that is not in the colocate backend set,
@@ -771,9 +785,19 @@ && containsAnyHighPrioBackend(replicaBackendIds, Config.tablet_sched_colocate_ba
                 // this replica is alive but version incomplete
                 return TabletStatus.VERSION_INCOMPLETE;
             }
+
+            Backend backend = GlobalStateMgr.getCurrentSystemInfo().getBackend(replica.getBackendId());
+            if (backend != null && !backend.isDiskDecommissioned(replica.getPathHash())) {
+                diskStableCnt++;
+            }
+        }
+
+        // 3. check disk decommission
+        if (diskStableCnt < replicationNum) {
+            return TabletStatus.DISK_MIGRATION;
         }
 
-        // 3. check redundant
+        // 4. check redundant
         if (replicas.size() > replicationNum) {
             return TabletStatus.COLOCATE_REDUNDANT;
         }