[enhancement](cloud) support param to drop fe cluster not in safe time

cloud/src/common/config.h

@@ -94,6 +94,7 @@ CONF_mInt64(delete_bitmap_storage_optimize_check_version_gap, "1000");
 CONF_mInt32(scan_instances_interval_seconds, "60"); // 1min
 // interval for check object
 CONF_mInt32(check_object_interval_seconds, "43200"); // 12hours
+CONF_mInt32(sql_node_safe_drop_time_seconds, "300"); // 5min
 
 CONF_mInt64(check_recycle_task_interval_seconds, "600"); // 10min
 CONF_mInt64(recycler_sleep_before_scheduling_seconds, "60");

cloud/src/meta-service/meta_service_resource.cpp

@@ -2072,7 +2072,10 @@ void MetaServiceImpl::alter_cluster(google::protobuf::RpcController* controller,
         msg = r.second;
     } break;
     case AlterClusterRequest::DROP_CLUSTER: {
-        auto r = resource_mgr_->drop_cluster(instance_id, cluster);
+        bool safe_drop_on_sql_cluster = request->has_safe_drop_on_sql_cluster()
+                                                ? request->safe_drop_on_sql_cluster()
+                                                : true;
+        auto r = resource_mgr_->drop_cluster(instance_id, cluster, safe_drop_on_sql_cluster);
         code = r.first;
         msg = r.second;
     } break;

cloud/src/resource-manager/resource_manager.cpp

@@ -461,7 +461,7 @@ std::pair<MetaServiceCode, std::string> ResourceManager::add_cluster(const std::
 static bool is_sql_node_exceeded_safe_drop_time(const NodeInfoPB& node) {
     int64_t ctime = node.ctime();
     // protect time 5mins
-    int64_t exceed_time = 5 * 60;
+    int64_t exceed_time = config::sql_node_safe_drop_time_seconds;
     TEST_SYNC_POINT_CALLBACK("resource_manager::set_safe_drop_time", &exceed_time);
     exceed_time = ctime + exceed_time;
     auto now_time = std::chrono::system_clock::now();
@@ -471,7 +471,7 @@ static bool is_sql_node_exceeded_safe_drop_time(const NodeInfoPB& node) {
 }
 
 std::pair<MetaServiceCode, std::string> ResourceManager::drop_cluster(
-        const std::string& instance_id, const ClusterInfo& cluster) {
+        const std::string& instance_id, const ClusterInfo& cluster, bool safe_drop_on_sql_cluster) {
     std::stringstream ss;
     std::string msg;
 
@@ -521,7 +521,7 @@ std::pair<MetaServiceCode, std::string> ResourceManager::drop_cluster(
     for (auto& i : instance.clusters()) {
         ++idx;
         if (i.cluster_id() == cluster.cluster.cluster_id()) {
-            if (i.type() == ClusterPB::SQL) {
+            if (i.type() == ClusterPB::SQL && safe_drop_on_sql_cluster) {
                 for (auto& fe_node : i.nodes()) {
                     // check drop fe cluster
                     if (!is_sql_node_exceeded_safe_drop_time(fe_node)) {

cloud/src/resource-manager/resource_manager.h

@@ -78,10 +78,12 @@ class ResourceManager {
      * Drops a cluster
      *
      * @param cluster cluster to drop, only cluster name and cluster id are concered
+     * @param safe_drop_on_sql_cluster if true, check if sql-cluster is drop in safe time
      * @return empty string for success, otherwise failure reason returned
      */
-    virtual std::pair<MetaServiceCode, std::string> drop_cluster(const std::string& instance_id,
-                                                                 const ClusterInfo& cluster);
+    virtual std::pair<MetaServiceCode, std::string> drop_cluster(
+            const std::string& instance_id, const ClusterInfo& cluster,
+            bool safe_drop_on_sql_cluster = true);
 
     /**
      * Update a cluster

cloud/test/mock_resource_manager.h

@@ -46,8 +46,9 @@ class MockResourceManager : public ResourceManager {
         return std::make_pair(MetaServiceCode::OK, "");
     }
 
-    std::pair<MetaServiceCode, std::string> drop_cluster(const std::string& instance_id,
-                                                         const ClusterInfo& cluster) override {
+    std::pair<MetaServiceCode, std::string> drop_cluster(
+            const std::string& instance_id, const ClusterInfo& cluster,
+            bool safe_drop_on_sql_cluster = true) override {
         return std::make_pair(MetaServiceCode::OK, "");
     }
 
@@ -74,4 +75,4 @@ class MockResourceManager : public ResourceManager {
                              const std::vector<NodeInfo>& to_del) override {
         return "";
     }
-};
+};

gensrc/proto/cloud.proto

@@ -1119,6 +1119,8 @@ message AlterClusterRequest {
     optional Operation op = 4;
     // for SQL mode rename cluster, rename to cluster name eq instance empty cluster name, need drop empty cluster
     optional bool replace_if_existing_empty_target_cluster = 5;
+    // if true, check if drop_cluster op on sql type cluster is in safe time
+    optional bool safe_drop_on_sql_cluster = 6;
 }
 
 message AlterClusterResponse {