You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by da...@apache.org on 2024/09/13 10:41:53 UTC
(doris) 03/03: [improvement](cloud) manage node via sql like non cloud mode (#40264)
This is an automated email from the ASF dual-hosted git repository.
dataroaring pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/doris.git
commit a0588eb212a684b2f3dc1395b2fa2d56bc292e04
Author: Yongqiang YANG <98...@users.noreply.github.com>
AuthorDate: Thu Sep 12 17:22:57 2024 +0800
[improvement](cloud) manage node via sql like non cloud mode (#40264)
1. add a config deploy_mode to enable disaggregated mode. In this mode
users have to config cloud_instance_id and meta_service_endpoint.
2. When a fe starts from empty, it would try to create instance.
3. If the instance does not exists, it creates and starts from master
role. If the instance exists, then it stats with its role got from ms.
4. Frontends are added via sql alter system add frontend. Backends are
added via sql alter system add backend.
5. Users do not need config cloud_instance_id and meta_service_endpoint
in be.conf, because fe sends them to be via heartbeat.
6. Builtin vault is not needed any more, internal tabels are stored in
default vault.
TODO:
1. decomission fe and be via sql
2. change cloud_instance_id to cluster id.
doc pr https://github.com/apache/doris-website/pull/1072
---
be/src/agent/heartbeat_server.cpp | 32 ++
be/src/cloud/cloud_meta_mgr.cpp | 2 +-
be/src/cloud/cloud_storage_engine.cpp | 24 +-
be/src/cloud/cloud_storage_engine.h | 21 +-
be/src/cloud/config.cpp | 14 +-
be/src/cloud/config.h | 11 +-
be/src/common/config.cpp | 3 +
cloud/src/meta-service/meta_service_resource.cpp | 5 +-
cloud/src/resource-manager/resource_manager.cpp | 4 +-
docker/runtime/doris-compose/cluster.py | 80 ++++-
docker/runtime/doris-compose/command.py | 147 +++++++-
docker/runtime/doris-compose/database.py | 72 +++-
docker/runtime/doris-compose/resource/init_be.sh | 6 +
.../runtime/doris-compose/resource/init_cloud.sh | 7 +
docker/runtime/doris-compose/resource/init_fe.sh | 15 +
.../fe/resource/fe_disaggregated_entrypoint.sh | 2 +
.../main/java/org/apache/doris/common/Config.java | 19 +-
.../main/java/org/apache/doris/alter/Alter.java | 14 +-
.../java/org/apache/doris/alter/SystemHandler.java | 10 +-
.../main/java/org/apache/doris/catalog/Env.java | 44 ++-
.../doris/catalog/InternalSchemaInitializer.java | 19 +-
.../org/apache/doris/cloud/catalog/CloudEnv.java | 78 ++++-
.../apache/doris/cloud/rpc/MetaServiceClient.java | 5 +
.../apache/doris/cloud/rpc/MetaServiceProxy.java | 10 +
.../doris/cloud/system/CloudSystemInfoService.java | 238 ++++++++++++-
.../doris/common/RandomIdentifierGenerator.java | 51 +++
.../apache/doris/common/util/PropertyAnalyzer.java | 2 +-
.../apache/doris/datasource/InternalCatalog.java | 8 +-
.../org/apache/doris/deploy/DeployManager.java | 4 +-
.../doris/httpv2/rest/manager/NodeAction.java | 2 +-
.../main/java/org/apache/doris/qe/DdlExecutor.java | 6 +-
.../main/java/org/apache/doris/resource/Tag.java | 2 +
.../doris/statistics/util/StatisticsUtil.java | 3 +
.../java/org/apache/doris/system/HeartbeatMgr.java | 8 +
.../org/apache/doris/system/SystemInfoService.java | 9 +
.../org/apache/doris/clone/DecommissionTest.java | 2 +-
.../doris/cluster/DecommissionBackendTest.java | 8 +-
gensrc/thrift/HeartbeatService.thrift | 2 +
.../org/apache/doris/regression/suite/Suite.groovy | 3 +-
.../doris/regression/suite/SuiteCluster.groovy | 31 +-
.../node_mgr/test_sql_mode_node_mgr.groovy | 388 +++++++++++++++++++++
41 files changed, 1269 insertions(+), 142 deletions(-)
diff --git a/be/src/agent/heartbeat_server.cpp b/be/src/agent/heartbeat_server.cpp
index 02c9bda41b6..e981e9165b5 100644
--- a/be/src/agent/heartbeat_server.cpp
+++ b/be/src/agent/heartbeat_server.cpp
@@ -26,6 +26,7 @@
#include <ostream>
#include <string>
+#include "cloud/config.h"
#include "common/config.h"
#include "common/status.h"
#include "olap/storage_engine.h"
@@ -244,6 +245,37 @@ Status HeartbeatServer::_heartbeat(const TMasterInfo& master_info) {
_engine.notify_listeners();
}
+ if (master_info.__isset.meta_service_endpoint != config::is_cloud_mode()) {
+ return Status::InvalidArgument(
+ "fe and be do not work in same mode, fe cloud mode: {},"
+ " be cloud mode: {}",
+ master_info.__isset.meta_service_endpoint, config::is_cloud_mode());
+ }
+
+ if (master_info.__isset.meta_service_endpoint && config::meta_service_endpoint.empty() &&
+ !master_info.meta_service_endpoint.empty()) {
+ auto st = config::set_config("meta_service_endpoint", master_info.meta_service_endpoint,
+ true);
+ LOG(INFO) << "set config meta_service_endpoing " << master_info.meta_service_endpoint << " "
+ << st;
+ }
+
+ if (master_info.__isset.cloud_instance_id) {
+ if (!config::cloud_instance_id.empty() &&
+ config::cloud_instance_id != master_info.cloud_instance_id) {
+ return Status::InvalidArgument(
+ "cloud_instance_id in fe.conf and be.conf are not same, fe: {}, be: {}",
+ master_info.cloud_instance_id, config::cloud_instance_id);
+ }
+
+ if (config::cloud_instance_id.empty() && !master_info.cloud_instance_id.empty()) {
+ auto st = config::set_config("cloud_instance_id", master_info.cloud_instance_id, true);
+ config::set_cloud_unique_id(master_info.cloud_instance_id);
+ LOG(INFO) << "set config cloud_instance_id " << master_info.cloud_instance_id << " "
+ << st;
+ }
+ }
+
return Status::OK();
}
diff --git a/be/src/cloud/cloud_meta_mgr.cpp b/be/src/cloud/cloud_meta_mgr.cpp
index 816f1108299..36217ddb61d 100644
--- a/be/src/cloud/cloud_meta_mgr.cpp
+++ b/be/src/cloud/cloud_meta_mgr.cpp
@@ -943,7 +943,7 @@ Status CloudMetaMgr::get_storage_vault_info(StorageVaultInfos* vault_infos, bool
j->mutable_obj_info()->set_sk(j->obj_info().sk().substr(0, 2) + "xxx");
}
- LOG(INFO) << "get storage vault, enable_storage_vault=" << is_vault_mode
+ LOG(INFO) << "get storage vault, enable_storage_vault=" << *is_vault_mode
<< " response=" << resp.ShortDebugString();
return Status::OK();
}
diff --git a/be/src/cloud/cloud_storage_engine.cpp b/be/src/cloud/cloud_storage_engine.cpp
index b98b2e3d0ef..1cf6e2ac039 100644
--- a/be/src/cloud/cloud_storage_engine.cpp
+++ b/be/src/cloud/cloud_storage_engine.cpp
@@ -160,29 +160,7 @@ struct RefreshFSVaultVisitor {
};
Status CloudStorageEngine::open() {
- cloud::StorageVaultInfos vault_infos;
- bool enable_storage_vault = false;
- do {
- auto st = _meta_mgr->get_storage_vault_info(&vault_infos, &enable_storage_vault);
- if (st.ok()) {
- break;
- }
-
- LOG(WARNING) << "failed to get vault info, retry after 5s, err=" << st;
- std::this_thread::sleep_for(5s);
- } while (vault_infos.empty());
-
- for (auto& [id, vault_info, path_format] : vault_infos) {
- if (auto st = std::visit(VaultCreateFSVisitor {id, path_format}, vault_info); !st.ok())
- [[unlikely]] {
- return vault_process_error(id, vault_info, std::move(st));
- }
- }
-
- // vault mode should not support latest_fs to get rid of unexpected storage backends choosen
- if (!enable_storage_vault) {
- set_latest_fs(get_filesystem(std::get<0>(vault_infos.back())));
- }
+ sync_storage_vault();
// TODO(plat1ko): DeleteBitmapTxnManager
diff --git a/be/src/cloud/cloud_storage_engine.h b/be/src/cloud/cloud_storage_engine.h
index 337b3194fd2..d3a55c3c377 100644
--- a/be/src/cloud/cloud_storage_engine.h
+++ b/be/src/cloud/cloud_storage_engine.h
@@ -74,14 +74,27 @@ public:
}
void _check_file_cache_ttl_block_valid();
- std::optional<StorageResource> get_storage_resource(const std::string& vault_id) const {
+ std::optional<StorageResource> get_storage_resource(const std::string& vault_id) {
+ LOG(INFO) << "Getting storage resource for vault_id: " << vault_id;
if (vault_id.empty()) {
+ if (latest_fs() == nullptr) {
+ LOG(INFO) << "there is not latest fs";
+ return std::nullopt;
+ }
return StorageResource {latest_fs()};
}
- if (auto storage_resource = doris::get_storage_resource(vault_id); storage_resource) {
- return storage_resource->first;
- }
+ bool synced = false;
+ do {
+ if (auto storage_resource = doris::get_storage_resource(vault_id); storage_resource) {
+ return storage_resource->first;
+ }
+ if (synced) {
+ break;
+ }
+ sync_storage_vault();
+ synced = true;
+ } while (true);
return std::nullopt;
}
diff --git a/be/src/cloud/config.cpp b/be/src/cloud/config.cpp
index 1b8256bf932..0f59b51059b 100644
--- a/be/src/cloud/config.cpp
+++ b/be/src/cloud/config.cpp
@@ -17,10 +17,14 @@
#include "cloud/config.h"
+#include "common/status.h"
+
namespace doris::config {
-DEFINE_String(cloud_unique_id, "");
-DEFINE_String(meta_service_endpoint, "");
+DEFINE_String(deploy_mode, "");
+DEFINE_mString(cloud_instance_id, "");
+DEFINE_mString(cloud_unique_id, "");
+DEFINE_mString(meta_service_endpoint, "");
DEFINE_Bool(meta_service_use_load_balancer, "false");
DEFINE_mInt32(meta_service_rpc_timeout_ms, "10000");
DEFINE_Bool(meta_service_connection_pooled, "true");
@@ -64,4 +68,10 @@ DEFINE_mBool(enable_new_tablet_do_compaction, "false");
DEFINE_Bool(enable_cloud_txn_lazy_commit, "false");
+void set_cloud_unique_id(std::string instance_id) {
+ if (cloud_unique_id.empty() && !instance_id.empty()) {
+ static_cast<void>(set_config("cloud_unique_id", "1:" + instance_id + ":compute", true));
+ }
+}
+
} // namespace doris::config
diff --git a/be/src/cloud/config.h b/be/src/cloud/config.h
index 104ead04996..57f6348df70 100644
--- a/be/src/cloud/config.h
+++ b/be/src/cloud/config.h
@@ -21,12 +21,17 @@
namespace doris::config {
-DECLARE_String(cloud_unique_id);
+DECLARE_String(deploy_mode);
+// deprecated do not configure directly
+DECLARE_mString(cloud_instance_id);
+DECLARE_mString(cloud_unique_id);
static inline bool is_cloud_mode() {
- return !cloud_unique_id.empty();
+ return deploy_mode == "cloud" || !cloud_unique_id.empty();
}
+void set_cloud_unique_id(std::string instance_id);
+
// Set the endpoint of meta service.
//
// If meta services are deployed behind a load balancer, set this config to "host:port" of the load balancer.
@@ -40,7 +45,7 @@ static inline bool is_cloud_mode() {
// If you want to access a group of meta services directly, set the addresses of meta services to this config,
// separated by a comma, like "host:port,host:port,host:port", then BE will choose a server to connect in randomly.
// In this mode, The config meta_service_connection_pooled is still useful, but the other two configs will be ignored.
-DECLARE_String(meta_service_endpoint);
+DECLARE_mString(meta_service_endpoint);
// Set the underlying connection type to pooled.
DECLARE_Bool(meta_service_connection_pooled);
DECLARE_mInt64(meta_service_connection_pool_size);
diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp
index c310f6b3e15..75f4bb7c4ea 100644
--- a/be/src/common/config.cpp
+++ b/be/src/common/config.cpp
@@ -37,6 +37,7 @@
#include <utility>
#include <vector>
+#include "cloud/config.h"
#include "common/config.h"
#include "common/logging.h"
#include "common/status.h"
@@ -1673,6 +1674,8 @@ bool init(const char* conf_file, bool fill_conf_map, bool must_exist, bool set_t
SET_FIELD(it.second, std::vector<std::string>, fill_conf_map, set_to_default);
}
+ set_cloud_unique_id(cloud_instance_id);
+
return true;
}
diff --git a/cloud/src/meta-service/meta_service_resource.cpp b/cloud/src/meta-service/meta_service_resource.cpp
index 8a25a73771c..b8bef65c91b 100644
--- a/cloud/src/meta-service/meta_service_resource.cpp
+++ b/cloud/src/meta-service/meta_service_resource.cpp
@@ -2338,9 +2338,8 @@ void MetaServiceImpl::get_cluster(google::protobuf::RpcController* controller,
std::find_if(instance.storage_vault_names().begin(), instance.storage_vault_names().end(),
[](const std::string& name) { return name == BUILT_IN_STORAGE_VAULT_NAME; }) ==
instance.storage_vault_names().end()) {
- code = MetaServiceCode::STORAGE_VAULT_NOT_FOUND;
- msg = "instance has no built in storage vault";
- return;
+ LOG_EVERY_N(INFO, 100) << "There is no builtin vault in instance "
+ << instance.instance_id();
}
auto get_cluster_mysql_user = [](const ClusterPB& c, std::set<std::string>* mysql_users) {
diff --git a/cloud/src/resource-manager/resource_manager.cpp b/cloud/src/resource-manager/resource_manager.cpp
index 58122ac3f7c..43f0a7368d8 100644
--- a/cloud/src/resource-manager/resource_manager.cpp
+++ b/cloud/src/resource-manager/resource_manager.cpp
@@ -171,7 +171,9 @@ bool ResourceManager::check_cluster_params_valid(const ClusterPB& cluster, std::
n.heartbeat_port()) {
continue;
}
- ss << "check cluster params failed, node : " << proto_to_json(n);
+ ss << "check cluster params failed, edit_log_port is required for frontends while "
+ "heatbeat_port is required for banckens, node : "
+ << proto_to_json(n);
*err = ss.str();
no_err = false;
break;
diff --git a/docker/runtime/doris-compose/cluster.py b/docker/runtime/doris-compose/cluster.py
index fcddd7f4c93..0ce12f3c7d4 100644
--- a/docker/runtime/doris-compose/cluster.py
+++ b/docker/runtime/doris-compose/cluster.py
@@ -291,15 +291,27 @@ class Node(object):
enable_coverage = self.cluster.coverage_dir
envs = {
- "MY_IP": self.get_ip(),
- "MY_ID": self.id,
- "MY_TYPE": self.node_type(),
- "FE_QUERY_PORT": FE_QUERY_PORT,
- "FE_EDITLOG_PORT": FE_EDITLOG_PORT,
- "BE_HEARTBEAT_PORT": BE_HEARTBEAT_PORT,
- "DORIS_HOME": os.path.join(self.docker_home_dir()),
- "STOP_GRACE": 1 if enable_coverage else 0,
- "IS_CLOUD": 1 if self.cluster.is_cloud else 0,
+ "MY_IP":
+ self.get_ip(),
+ "MY_ID":
+ self.id,
+ "MY_TYPE":
+ self.node_type(),
+ "FE_QUERY_PORT":
+ FE_QUERY_PORT,
+ "FE_EDITLOG_PORT":
+ FE_EDITLOG_PORT,
+ "BE_HEARTBEAT_PORT":
+ BE_HEARTBEAT_PORT,
+ "DORIS_HOME":
+ os.path.join(self.docker_home_dir()),
+ "STOP_GRACE":
+ 1 if enable_coverage else 0,
+ "IS_CLOUD":
+ 1 if self.cluster.is_cloud else 0,
+ "SQL_MODE_NODE_MGR":
+ 1 if hasattr(self.cluster, 'sql_mode_node_mgr')
+ and self.cluster.sql_mode_node_mgr else 0
}
if self.cluster.is_cloud:
@@ -390,15 +402,22 @@ class FE(Node):
cfg += self.cluster.fe_config
if self.cluster.is_cloud:
cfg += [
- "cloud_unique_id = " + self.cloud_unique_id(),
"meta_service_endpoint = {}".format(
- self.cluster.get_meta_server_addr()),
- "",
+ self.cluster.get_meta_server_addr()), "",
"# For regression-test",
"ignore_unsupported_properties_in_cloud_mode = true",
"merge_on_write_forced_to_false = true",
+ "deploy_mode = cloud"
]
+ if self.cluster.sql_mode_node_mgr:
+ cfg += [
+ "cloud_instance_id = " + self.cloud_instance_id(),
+ ]
+ else:
+ cfg += [
+ "cloud_unique_id = " + self.cloud_unique_id(),
+ ]
return cfg
def init_is_follower(self):
@@ -420,6 +439,9 @@ class FE(Node):
def cloud_unique_id(self):
return "sql_server_{}".format(self.id)
+ def cloud_instance_id(self):
+ return "reg_cloud_instance"
+
def entrypoint(self):
return ["bash", os.path.join(DOCKER_RESOURCE_PATH, "init_fe.sh")]
@@ -450,14 +472,26 @@ class BE(Node):
cfg += self.cluster.be_config
if self.cluster.is_cloud:
cfg += [
- "cloud_unique_id = " + self.cloud_unique_id(),
- "meta_service_endpoint = {}".format(
- self.cluster.get_meta_server_addr()),
'tmp_file_dirs = [ {"path":"./storage/tmp","max_cache_bytes":10240000, "max_upload_bytes":10240000}]',
'enable_file_cache = true',
'file_cache_path = [ {{"path": "{}/storage/file_cache", "total_size":53687091200, "query_limit": 10737418240}}]'
.format(self.docker_home_dir()),
+ "deploy_mode = cloud",
]
+
+ if self.cluster.be_metaservice_endpoint:
+ cfg += [
+ "meta_service_endpoint = {}".format(
+ self.cluster.get_meta_server_addr()),
+ ]
+ if self.cluster.be_cloud_instanceid:
+ cfg += [
+ "cloud_instance_id = " + self.cloud_instance_id(),
+ ]
+ if not self.cluster.sql_mode_node_mgr:
+ cfg += [
+ "cloud_unique_id = " + self.cloud_unique_id(),
+ ]
return cfg
def init_cluster_name(self):
@@ -519,6 +553,9 @@ class BE(Node):
def cloud_unique_id(self):
return "compute_node_{}".format(self.id)
+ def cloud_instance_id(self):
+ return "reg_cloud_instance"
+
def docker_home_dir(self):
return os.path.join(DOCKER_DORIS_PATH, "be")
@@ -628,7 +665,8 @@ class Cluster(object):
def __init__(self, name, subnet, image, is_cloud, fe_config, be_config,
ms_config, recycle_config, fe_follower, be_disks, be_cluster,
- reg_be, coverage_dir, cloud_store_config):
+ reg_be, coverage_dir, cloud_store_config, sql_mode_node_mgr,
+ be_metaservice_endpoint, be_cloud_instanceid):
self.name = name
self.subnet = subnet
self.image = image
@@ -647,11 +685,15 @@ class Cluster(object):
node_type: Group(node_type)
for node_type in Node.TYPE_ALL
}
+ self.sql_mode_node_mgr = sql_mode_node_mgr
+ self.be_metaservice_endpoint = be_metaservice_endpoint
+ self.be_cloud_instanceid = be_cloud_instanceid
@staticmethod
def new(name, image, is_cloud, fe_config, be_config, ms_config,
recycle_config, fe_follower, be_disks, be_cluster, reg_be,
- coverage_dir, cloud_store_config):
+ coverage_dir, cloud_store_config, sql_mode_node_mgr,
+ be_metaservice_endpoint, be_cloud_instanceid):
if not os.path.exists(LOCAL_DORIS_PATH):
os.makedirs(LOCAL_DORIS_PATH, exist_ok=True)
os.chmod(LOCAL_DORIS_PATH, 0o777)
@@ -663,7 +705,9 @@ class Cluster(object):
cluster = Cluster(name, subnet, image, is_cloud, fe_config,
be_config, ms_config, recycle_config,
fe_follower, be_disks, be_cluster, reg_be,
- coverage_dir, cloud_store_config)
+ coverage_dir, cloud_store_config,
+ sql_mode_node_mgr, be_metaservice_endpoint,
+ be_cloud_instanceid)
os.makedirs(cluster.get_path(), exist_ok=True)
os.makedirs(get_status_path(name), exist_ok=True)
cluster._save_meta()
diff --git a/docker/runtime/doris-compose/command.py b/docker/runtime/doris-compose/command.py
index ed88dd03f4d..de818c4f6c4 100644
--- a/docker/runtime/doris-compose/command.py
+++ b/docker/runtime/doris-compose/command.py
@@ -297,6 +297,45 @@ class UpCommand(Command):
default="",
help="Set code coverage output directory")
+ parser.add_argument("--sql-mode-node-mgr",
+ default=False,
+ action=self._get_parser_bool_action(True),
+ help="Manager fe be via sql instead of http")
+
+ if self._support_boolean_action():
+ parser.add_argument(
+ "--be-metaservice-endpoint",
+ default=True,
+ action=self._get_parser_bool_action(False),
+ help=
+ "Do not set BE meta service endpoint in conf. Default is False."
+ )
+ else:
+ parser.add_argument(
+ "--no-be-metaservice-endpoint",
+ dest='be_metaservice_endpoint',
+ default=True,
+ action=self._get_parser_bool_action(False),
+ help=
+ "Do not set BE meta service endpoint in conf. Default is False."
+ )
+
+ if self._support_boolean_action():
+ parser.add_argument(
+ "--be-cloud-instanceid",
+ default=True,
+ action=self._get_parser_bool_action(False),
+ help=
+ "Do not set BE cloud instance ID in conf. Default is False.")
+ else:
+ parser.add_argument(
+ "--no-be-cloud-instanceid",
+ dest='be_cloud_instanceid',
+ default=True,
+ action=self._get_parser_bool_action(False),
+ help=
+ "Do not set BE cloud instance ID in conf. Default is False.")
+
parser.add_argument(
"--fdb-version",
type=str,
@@ -394,7 +433,8 @@ class UpCommand(Command):
args.NAME, args.IMAGE, args.cloud, args.fe_config,
args.be_config, args.ms_config, args.recycle_config,
args.fe_follower, args.be_disks, args.be_cluster, args.reg_be,
- args.coverage_dir, cloud_store_config)
+ args.coverage_dir, cloud_store_config, args.sql_mode_node_mgr,
+ args.be_metaservice_endpoint, args.be_cloud_instanceid)
LOG.info("Create new cluster {} succ, cluster path is {}".format(
args.NAME, cluster.get_path()))
@@ -483,6 +523,49 @@ class UpCommand(Command):
"Not up cluster cause specific --no-start, related node num {}"
.format(related_node_num)))
else:
+ LOG.info("Using SQL mode for node management ? {}".format(
+ args.sql_mode_node_mgr))
+
+ # Wait for FE master to be elected
+ LOG.info("Waiting for FE master to be elected...")
+ expire_ts = time.time() + 30
+ while expire_ts > time.time():
+ db_mgr = database.get_db_mgr(args.NAME, False)
+ for id in add_fe_ids:
+ fe_state = db_mgr.get_fe(id)
+ if fe_state is not None and fe_state.alive:
+ break
+ LOG.info("there is no fe ready")
+ time.sleep(5)
+
+ if cluster.is_cloud and args.sql_mode_node_mgr:
+ db_mgr = database.get_db_mgr(args.NAME, False)
+ master_fe_endpoint = CLUSTER.get_master_fe_endpoint(
+ cluster.name)
+ # Add FEs except master_fe
+ for fe in cluster.get_all_nodes(CLUSTER.Node.TYPE_FE):
+ fe_endpoint = f"{fe.get_ip()}:{CLUSTER.FE_EDITLOG_PORT}"
+ if fe_endpoint != master_fe_endpoint:
+ try:
+ db_mgr.add_fe(fe_endpoint)
+ LOG.info(f"Added FE {fe_endpoint} successfully.")
+ except Exception as e:
+ LOG.error(
+ f"Failed to add FE {fe_endpoint}: {str(e)}")
+
+ # Add BEs
+ for be in cluster.get_all_nodes(CLUSTER.Node.TYPE_BE):
+ be_endpoint = f"{be.get_ip()}:{CLUSTER.BE_HEARTBEAT_PORT}"
+ try:
+ db_mgr.add_be(be_endpoint)
+ LOG.info(f"Added BE {be_endpoint} successfully.")
+ except Exception as e:
+ LOG.error(f"Failed to add BE {be_endpoint}: {str(e)}")
+
+ cloud_store_config = self._get_cloud_store_config()
+
+ db_mgr.create_default_storage_vault(cloud_store_config)
+
if args.wait_timeout != 0:
if args.wait_timeout == -1:
args.wait_timeout = 1000000000
@@ -509,6 +592,7 @@ class UpCommand(Command):
err += "dead be: " + str(dead_backends) + ". "
raise Exception(err)
time.sleep(1)
+
LOG.info(
utils.render_green(
"Up cluster {} succ, related node num {}".format(
@@ -615,6 +699,8 @@ class DownCommand(Command):
args.fdb_id,
ignore_not_exists=True)
+ LOG.info("down cluster " + args.NAME + " for all " + str(for_all))
+
if for_all:
if os.path.exists(cluster.get_compose_file()):
try:
@@ -700,6 +786,8 @@ class ListNode(object):
self.tablet_num = ""
self.last_heartbeat = ""
self.err_msg = ""
+ self.edit_log_port = 0
+ self.heartbeat_port = 0
def info(self, detail):
result = [
@@ -711,21 +799,23 @@ class ListNode(object):
if detail:
query_port = ""
http_port = ""
+ heartbeat_port = ""
+ edit_log_port = ""
node_path = CLUSTER.get_node_path(self.cluster_name,
self.node_type, self.id)
if self.node_type == CLUSTER.Node.TYPE_FE:
query_port = CLUSTER.FE_QUERY_PORT
http_port = CLUSTER.FE_HTTP_PORT
+ edit_log_port = CLUSTER.FE_EDITLOG_PORT
elif self.node_type == CLUSTER.Node.TYPE_BE:
http_port = CLUSTER.BE_WEBSVR_PORT
+ heartbeat_port = CLUSTER.BE_HEARTBEAT_PORT
elif self.node_type == CLUSTER.Node.TYPE_MS or self.node_type == CLUSTER.Node.TYPE_RECYCLE:
http_port = CLUSTER.MS_PORT
else:
pass
result += [
- query_port,
- http_port,
- node_path,
+ query_port, http_port, node_path, edit_log_port, heartbeat_port
]
return result
@@ -738,6 +828,7 @@ class ListNode(object):
self.query_port = fe.query_port
self.last_heartbeat = fe.last_heartbeat
self.err_msg = fe.err_msg
+ self.edit_log_port = fe.edit_log_port
elif self.node_type == CLUSTER.Node.TYPE_BE:
self.backend_id = -1
be = db_mgr.get_be(self.id)
@@ -747,6 +838,7 @@ class ListNode(object):
self.tablet_num = be.tablet_num
self.last_heartbeat = be.last_heartbeat
self.err_msg = be.err_msg
+ self.heartbeat_port = be.heartbeat_port
class GenConfCommand(Command):
@@ -903,8 +995,7 @@ class ListCommand(Command):
if services is None:
return COMPOSE_BAD, {}
return COMPOSE_GOOD, {
- service:
- ComposeService(
+ service: ComposeService(
service,
list(service_conf["networks"].values())[0]
["ipv4_address"], service_conf["image"])
@@ -980,6 +1071,8 @@ class ListCommand(Command):
"query_port",
"http_port",
"path",
+ "edit_log_port",
+ "heartbeat_port",
]
rows = []
@@ -1056,6 +1149,47 @@ class ListCommand(Command):
return self._handle_data(header, rows)
+class GetCloudIniCommand(Command):
+
+ def add_parser(self, args_parsers):
+ parser = args_parsers.add_parser("get-cloud-ini",
+ help="Get cloud.init")
+ parser.add_argument(
+ "NAME",
+ nargs="*",
+ help=
+ "Specify multiple clusters, if specific, show all their containers."
+ )
+ self._add_parser_output_json(parser)
+
+ def _handle_data(self, header, datas):
+ if utils.is_enable_log():
+ table = prettytable.PrettyTable(
+ [utils.render_green(field) for field in header])
+ for row in datas:
+ table.add_row(row)
+ print(table)
+ return ""
+ else:
+ datas.insert(0, header)
+ return datas
+
+ def run(self, args):
+
+ header = ["key", "value"]
+
+ rows = []
+
+ with open(CLUSTER.CLOUD_CFG_FILE, "r") as f:
+ for line in f:
+ line = line.strip()
+ if line and not line.startswith('#'):
+ key, value = line.split('=', 1)
+ rows.append([key.strip(), value.strip()])
+
+ return self._handle_data(header, rows)
+
+
ALL_COMMANDS = [
UpCommand("up"),
DownCommand("down"),
@@ -1064,6 +1198,7 @@ ALL_COMMANDS = [
SimpleCommand("restart", "Restart the doris containers. "),
SimpleCommand("pause", "Pause the doris containers. "),
SimpleCommand("unpause", "Unpause the doris containers. "),
+ GetCloudIniCommand("get-cloud-ini"),
GenConfCommand("config"),
ListCommand("ls"),
]
diff --git a/docker/runtime/doris-compose/database.py b/docker/runtime/doris-compose/database.py
index d29905e94a9..bbf6fb4fbeb 100644
--- a/docker/runtime/doris-compose/database.py
+++ b/docker/runtime/doris-compose/database.py
@@ -20,6 +20,7 @@ import os.path
import pymysql
import time
import utils
+import uuid
LOG = utils.get_logger()
@@ -27,19 +28,20 @@ LOG = utils.get_logger()
class FEState(object):
def __init__(self, id, query_port, is_master, alive, last_heartbeat,
- err_msg):
+ err_msg, edit_log_port):
self.id = id
self.query_port = query_port
self.is_master = is_master
self.alive = alive
self.last_heartbeat = last_heartbeat
self.err_msg = err_msg
+ self.edit_log_port = edit_log_port
class BEState(object):
def __init__(self, id, backend_id, decommissioned, alive, tablet_num,
- last_heartbeat, err_msg):
+ last_heartbeat, err_msg, heartbeat_port):
self.id = id
self.backend_id = backend_id
self.decommissioned = decommissioned
@@ -47,6 +49,7 @@ class BEState(object):
self.tablet_num = tablet_num
self.last_heartbeat = last_heartbeat
self.err_msg = err_msg
+ self.heartbeat_port = heartbeat_port
class DBManager(object):
@@ -70,6 +73,15 @@ class DBManager(object):
self._load_fe_states(query_ports)
self._load_be_states()
+ def add_fe(self, fe_endpoint):
+ try:
+ sql = f"ALTER SYSTEM ADD FOLLOWER '{fe_endpoint}'"
+ self._exec_query(sql)
+ LOG.info(f"Added FE {fe_endpoint} via SQL successfully.")
+ except Exception as e:
+ LOG.error(f"Failed to add FE {fe_endpoint} via SQL: {str(e)}")
+ raise
+
def drop_fe(self, fe_endpoint):
id = CLUSTER.Node.get_id_from_ip(fe_endpoint[:fe_endpoint.find(":")])
try:
@@ -85,6 +97,15 @@ class DBManager(object):
return
raise e
+ def add_be(self, be_endpoint):
+ try:
+ sql = f"ALTER SYSTEM ADD BACKEND '{be_endpoint}'"
+ self._exec_query(sql)
+ LOG.info(f"Added BE {be_endpoint} via SQL successfully.")
+ except Exception as e:
+ LOG.error(f"Failed to add BE {be_endpoint} via SQL: {str(e)}")
+ raise
+
def drop_be(self, be_endpoint):
id = CLUSTER.Node.get_id_from_ip(be_endpoint[:be_endpoint.find(":")])
try:
@@ -140,23 +161,55 @@ class DBManager(object):
time.sleep(5)
+ def create_default_storage_vault(self, cloud_store_config):
+ try:
+ # Create storage vault
+ create_vault_sql = f"""
+ CREATE STORAGE VAULT IF NOT EXISTS default_vault
+ PROPERTIES (
+ "type" = "S3",
+ "s3.access_key" = "{cloud_store_config['DORIS_CLOUD_AK']}",
+ "s3.secret_key" = "{cloud_store_config['DORIS_CLOUD_SK']}",
+ "s3.endpoint" = "{cloud_store_config['DORIS_CLOUD_ENDPOINT']}",
+ "s3.bucket" = "{cloud_store_config['DORIS_CLOUD_BUCKET']}",
+ "s3.region" = "{cloud_store_config['DORIS_CLOUD_REGION']}",
+ "s3.root.path" = "{str(uuid.uuid4())}",
+ "provider" = "{cloud_store_config['DORIS_CLOUD_PROVIDER']}"
+ );
+ """
+ self._exec_query(create_vault_sql)
+ LOG.info("Created storage vault 'default_vault'")
+
+ # Set as default storage vault
+ set_default_vault_sql = "SET default_vault as DEFAULT STORAGE VAULT;"
+ self._exec_query(set_default_vault_sql)
+ LOG.info("Set 'default_vault' as the default storage vault")
+
+ except Exception as e:
+ LOG.error(f"Failed to create default storage vault: {str(e)}")
+ raise
+
def _load_fe_states(self, query_ports):
fe_states = {}
alive_master_fe_port = None
for record in self._exec_query('''
- select Host, IsMaster, Alive, LastHeartbeat, ErrMsg
- from frontends()'''):
- ip, is_master, alive, last_heartbeat, err_msg = record
+ show frontends '''):
+ # Unpack the record into individual columns
+ name, ip, edit_log_port, _, query_port, _, _, role, is_master, cluster_id, _, alive, _, _, last_heartbeat, _, err_msg, _, _ = record
is_master = utils.is_true(is_master)
alive = utils.is_true(alive)
id = CLUSTER.Node.get_id_from_ip(ip)
query_port = query_ports.get(id, "")
last_heartbeat = utils.escape_null(last_heartbeat)
fe = FEState(id, query_port, is_master, alive, last_heartbeat,
- err_msg)
+ err_msg, edit_log_port)
fe_states[id] = fe
if is_master and alive and query_port:
alive_master_fe_port = query_port
+ LOG.info(
+ "record of show frontends, name {}, ip {}, alive {}, is_master {}, role {}"
+ .format(name, ip, alive, is_master, role))
+
self.fe_states = fe_states
if alive_master_fe_port and alive_master_fe_port != self.query_port:
self.query_port = alive_master_fe_port
@@ -165,17 +218,18 @@ class DBManager(object):
def _load_be_states(self):
be_states = {}
for record in self._exec_query('''
- select BackendId, Host, LastHeartbeat, Alive, SystemDecommissioned, TabletNum, ErrMsg
+ select BackendId, Host, LastHeartbeat, Alive, SystemDecommissioned, TabletNum, ErrMsg, HeartbeatPort
from backends()'''):
- backend_id, ip, last_heartbeat, alive, decommissioned, tablet_num, err_msg = record
+ backend_id, ip, last_heartbeat, alive, decommissioned, tablet_num, err_msg, heartbeat_port = record
backend_id = int(backend_id)
alive = utils.is_true(alive)
decommissioned = utils.is_true(decommissioned)
tablet_num = int(tablet_num)
id = CLUSTER.Node.get_id_from_ip(ip)
last_heartbeat = utils.escape_null(last_heartbeat)
+ heartbeat_port = utils.escape_null(heartbeat_port)
be = BEState(id, backend_id, decommissioned, alive, tablet_num,
- last_heartbeat, err_msg)
+ last_heartbeat, err_msg, heartbeat_port)
be_states[id] = be
self.be_states = be_states
diff --git a/docker/runtime/doris-compose/resource/init_be.sh b/docker/runtime/doris-compose/resource/init_be.sh
index d9b7953b534..e0ed5b92cb8 100755
--- a/docker/runtime/doris-compose/resource/init_be.sh
+++ b/docker/runtime/doris-compose/resource/init_be.sh
@@ -48,6 +48,12 @@ add_cloud_be() {
return
fi
+ # Check if SQL_MODE_NODE_MGR is set to 1
+ if [ "$SQL_MODE_NODE_MGR" -eq 1 ]; then
+ health_log "SQL_MODE_NODE_MGR is set to 1, skipping cluster creation"
+ return
+ fi
+
cluster_file_name="${DORIS_HOME}/conf/CLUSTER_NAME"
cluster_name=$(cat $cluster_file_name)
if [ -z $cluster_name ]; then
diff --git a/docker/runtime/doris-compose/resource/init_cloud.sh b/docker/runtime/doris-compose/resource/init_cloud.sh
index 78152b5330b..5740335ace3 100644
--- a/docker/runtime/doris-compose/resource/init_cloud.sh
+++ b/docker/runtime/doris-compose/resource/init_cloud.sh
@@ -50,6 +50,13 @@ check_init_cloud() {
lock_cluster
+ # Check if SQL_MODE_NODE_MGR is set
+ if [[ "$SQL_MODE_NODE_MGR" -eq 1 ]]; then
+ health_log "SQL_MODE_NODE_MGR is set, skipping create_instance"
+ touch $HAS_CREATE_INSTANCE_FILE
+ return
+ fi
+
output=$(curl -s "${META_SERVICE_ENDPOINT}/MetaService/http/create_instance?token=greedisgood9999" \
-d '{"instance_id":"default_instance_id",
"name": "default_instance",
diff --git a/docker/runtime/doris-compose/resource/init_fe.sh b/docker/runtime/doris-compose/resource/init_fe.sh
index e46a6cac9b2..d4aad29b0e5 100755
--- a/docker/runtime/doris-compose/resource/init_fe.sh
+++ b/docker/runtime/doris-compose/resource/init_fe.sh
@@ -88,6 +88,21 @@ start_cloud_fe() {
return
fi
+ # Check if SQL_MODE_NODE_MGR is set to 1
+ if [ "${SQL_MODE_NODE_MGR}" = "1" ]; then
+ health_log "SQL_MODE_NODE_MGR is set to 1. Skipping add FE."
+
+ touch $REGISTER_FILE
+
+ fe_daemon &
+ bash $DORIS_HOME/bin/start_fe.sh --daemon
+
+ if [ "$MY_ID" == "1" ]; then
+ echo $MY_IP >$MASTER_FE_IP_FILE
+ fi
+ return
+ fi
+
wait_create_instance
action=add_cluster
diff --git a/docker/runtime/fe/resource/fe_disaggregated_entrypoint.sh b/docker/runtime/fe/resource/fe_disaggregated_entrypoint.sh
index d9af8ab612a..876a3f9ace3 100755
--- a/docker/runtime/fe/resource/fe_disaggregated_entrypoint.sh
+++ b/docker/runtime/fe/resource/fe_disaggregated_entrypoint.sh
@@ -202,6 +202,8 @@ add_cluster_info_to_conf
check_and_modify_fqdn_config
link_config_files
variables_inital
+check_or_register_in_ms
+
check_or_register_in_ms
/opt/apache-doris/fe/bin/start_fe.sh --console
diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
index 42affa36daf..968a307c842 100644
--- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
+++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
@@ -1823,7 +1823,7 @@ public class Config extends ConfigBase {
public static boolean enable_date_conversion = true;
@ConfField(mutable = false, masterOnly = true)
- public static boolean enable_multi_tags = false;
+ public static boolean enable_multi_tags = true;
/**
* If set to TRUE, FE will convert DecimalV2 to DecimalV3 automatically.
@@ -2783,15 +2783,24 @@ public class Config extends ConfigBase {
@ConfField public static int warn_sys_accumulated_file_size = 2;
@ConfField public static int audit_sys_accumulated_file_size = 4;
+ @ConfField
+ public static String deploy_mode = "";
+
+ // compatibily with elder version.
+ // cloud_unique_id is introduced before cloud_instance_id, so it has higher priority.
@ConfField
public static String cloud_unique_id = "";
+ // If cloud_unique_id is empty, cloud_instance_id works, otherwise cloud_unique_id works.
+ @ConfField
+ public static String cloud_instance_id = "";
+
public static boolean isCloudMode() {
- return !cloud_unique_id.isEmpty();
+ return deploy_mode.equals("cloud") || !cloud_unique_id.isEmpty() || !cloud_instance_id.isEmpty();
}
public static boolean isNotCloudMode() {
- return cloud_unique_id.isEmpty();
+ return !isCloudMode();
}
/**
@@ -3023,4 +3032,8 @@ public class Config extends ConfigBase {
//==========================================================================
// end of cloud config
//==========================================================================
+
+ @ConfField(description = {"检查资源就绪的周期，单位秒",
+ "Interval checking if resource is ready"})
+ public static long resource_not_ready_sleep_seconds = 5;
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/alter/Alter.java b/fe/fe-core/src/main/java/org/apache/doris/alter/Alter.java
index 95ad5ae824b..51ed811cc06 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/alter/Alter.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/alter/Alter.java
@@ -99,18 +99,18 @@ public class Alter {
private AlterHandler schemaChangeHandler;
private AlterHandler materializedViewHandler;
- private SystemHandler clusterHandler;
+ private SystemHandler systemHandler;
public Alter() {
schemaChangeHandler = Config.isCloudMode() ? new CloudSchemaChangeHandler() : new SchemaChangeHandler();
materializedViewHandler = new MaterializedViewHandler();
- clusterHandler = new SystemHandler();
+ systemHandler = new SystemHandler();
}
public void start() {
schemaChangeHandler.start();
materializedViewHandler.start();
- clusterHandler.start();
+ systemHandler.start();
}
public void processCreateMaterializedView(CreateMaterializedViewStmt stmt)
@@ -710,8 +710,8 @@ public class Alter {
}
}
- public void processAlterCluster(AlterSystemStmt stmt) throws UserException {
- clusterHandler.process(Collections.singletonList(stmt.getAlterClause()), null, null);
+ public void processAlterSystem(AlterSystemStmt stmt) throws UserException {
+ systemHandler.process(Collections.singletonList(stmt.getAlterClause()), null, null);
}
private void processRename(Database db, OlapTable table, List<AlterClause> alterClauses) throws DdlException {
@@ -958,8 +958,8 @@ public class Alter {
return materializedViewHandler;
}
- public AlterHandler getClusterHandler() {
- return clusterHandler;
+ public AlterHandler getSystemHandler() {
+ return systemHandler;
}
public void processAlterMTMV(AlterMTMV alterMTMV, boolean isReplay) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/alter/SystemHandler.java b/fe/fe-core/src/main/java/org/apache/doris/alter/SystemHandler.java
index 28b3684ed04..78b3b76ea78 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/alter/SystemHandler.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/alter/SystemHandler.java
@@ -72,7 +72,7 @@ public class SystemHandler extends AlterHandler {
private static final Logger LOG = LogManager.getLogger(SystemHandler.class);
public SystemHandler() {
- super("cluster");
+ super("system");
}
@Override
@@ -156,15 +156,13 @@ public class SystemHandler extends AlterHandler {
// for decommission operation, here is no decommission job. the system handler will check
// all backend in decommission state
for (Backend backend : decommissionBackends) {
- backend.setDecommissioned(true);
- Env.getCurrentEnv().getEditLog().logBackendStateChange(backend);
- LOG.info("set backend {} to decommission", backend.getId());
+ Env.getCurrentSystemInfo().decommissionBackend(backend);
}
} else if (alterClause instanceof AddObserverClause) {
AddObserverClause clause = (AddObserverClause) alterClause;
Env.getCurrentEnv().addFrontend(FrontendNodeType.OBSERVER, clause.getHost(),
- clause.getPort(), "");
+ clause.getPort());
} else if (alterClause instanceof DropObserverClause) {
DropObserverClause clause = (DropObserverClause) alterClause;
Env.getCurrentEnv().dropFrontend(FrontendNodeType.OBSERVER, clause.getHost(),
@@ -172,7 +170,7 @@ public class SystemHandler extends AlterHandler {
} else if (alterClause instanceof AddFollowerClause) {
AddFollowerClause clause = (AddFollowerClause) alterClause;
Env.getCurrentEnv().addFrontend(FrontendNodeType.FOLLOWER, clause.getHost(),
- clause.getPort(), "");
+ clause.getPort());
} else if (alterClause instanceof DropFollowerClause) {
DropFollowerClause clause = (DropFollowerClause) alterClause;
Env.getCurrentEnv().dropFrontend(FrontendNodeType.FOLLOWER, clause.getHost(),
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
index eb8b10bbfe4..a4bc55eeeec 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
@@ -423,7 +423,7 @@ public class Env {
// node name is used for bdbje NodeName.
protected String nodeName;
protected FrontendNodeType role;
- private FrontendNodeType feType;
+ protected FrontendNodeType feType;
// replica and observer use this value to decide provide read service or not
private long synchronizedTimeMs;
private MasterInfo masterInfo;
@@ -1142,6 +1142,13 @@ public class Env {
this.httpReady.set(httpReady);
}
+ protected boolean isStartFromEmpty() {
+ File roleFile = new File(this.imageDir, Storage.ROLE_FILE);
+ File versionFile = new File(this.imageDir, Storage.VERSION_FILE);
+
+ return !roleFile.exists() && !versionFile.exists();
+ }
+
protected void getClusterIdAndRole() throws IOException {
File roleFile = new File(this.imageDir, Storage.ROLE_FILE);
File versionFile = new File(this.imageDir, Storage.VERSION_FILE);
@@ -2982,11 +2989,19 @@ public class Env {
};
}
+ public void addFrontend(FrontendNodeType role, String host, int editLogPort) throws DdlException {
+ addFrontend(role, host, editLogPort, "");
+ }
+
public void addFrontend(FrontendNodeType role, String host, int editLogPort, String nodeName) throws DdlException {
if (!tryLock(false)) {
throw new DdlException("Failed to acquire env lock. Try again");
}
try {
+ if (Strings.isNullOrEmpty(nodeName)) {
+ nodeName = genFeNodeName(host, editLogPort, false /* new name style */);
+ }
+
Frontend fe = checkFeExist(host, editLogPort);
if (fe != null) {
throw new DdlException("frontend already exists " + fe);
@@ -2995,10 +3010,6 @@ public class Env {
throw new DdlException("frontend's hostName should not be empty while enable_fqdn_mode is true");
}
- if (Strings.isNullOrEmpty(nodeName)) {
- nodeName = genFeNodeName(host, editLogPort, false /* new name style */);
- }
-
if (removedFrontends.contains(nodeName)) {
throw new DdlException("frontend name already exists " + nodeName + ". Try again");
}
@@ -3031,7 +3042,7 @@ public class Env {
modifyFrontendHost(fe.getNodeName(), destHost);
}
- public void modifyFrontendHost(String nodeName, String destHost) throws DdlException {
+ private void modifyFrontendHost(String nodeName, String destHost) throws DdlException {
if (!tryLock(false)) {
throw new DdlException("Failed to acquire env lock. Try again");
}
@@ -3059,6 +3070,11 @@ public class Env {
}
public void dropFrontend(FrontendNodeType role, String host, int port) throws DdlException {
+ dropFrontendFromBDBJE(role, host, port);
+ }
+
+ public void dropFrontendFromBDBJE(FrontendNodeType role, String host, int port) throws DdlException {
+
if (port == selfNode.getPort() && feType == FrontendNodeType.MASTER
&& selfNode.getHost().equals(host)) {
throw new DdlException("can not drop current master node.");
@@ -4308,8 +4324,8 @@ public class Env {
return cooldownConfHandler;
}
- public SystemHandler getClusterHandler() {
- return (SystemHandler) this.alter.getClusterHandler();
+ public SystemHandler getSystemHandler() {
+ return (SystemHandler) this.alter.getSystemHandler();
}
public BackupHandler getBackupHandler() {
@@ -5430,15 +5446,15 @@ public class Env {
}
/*
- * used for handling AlterClusterStmt
- * (for client is the ALTER CLUSTER command).
+ * used for handling AlterSystemStmt
+ * (for client is the ALTER SYSTEM command).
*/
- public void alterCluster(AlterSystemStmt stmt) throws DdlException, UserException {
- this.alter.processAlterCluster(stmt);
+ public void alterSystem(AlterSystemStmt stmt) throws DdlException, UserException {
+ this.alter.processAlterSystem(stmt);
}
- public void cancelAlterCluster(CancelAlterSystemStmt stmt) throws DdlException {
- this.alter.getClusterHandler().cancel(stmt);
+ public void cancelAlterSystem(CancelAlterSystemStmt stmt) throws DdlException {
+ this.alter.getSystemHandler().cancel(stmt);
}
// Switch catalog of this session
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/InternalSchemaInitializer.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/InternalSchemaInitializer.java
index 87e8a0fc3b0..cd33ee9980c 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/InternalSchemaInitializer.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/InternalSchemaInitializer.java
@@ -30,7 +30,6 @@ import org.apache.doris.analysis.ModifyPartitionClause;
import org.apache.doris.analysis.PartitionDesc;
import org.apache.doris.analysis.RangePartitionDesc;
import org.apache.doris.analysis.TableName;
-import org.apache.doris.cloud.catalog.CloudEnv;
import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.FeConstants;
@@ -55,8 +54,6 @@ import java.util.Optional;
public class InternalSchemaInitializer extends Thread {
- public static final int TABLE_CREATION_RETRY_INTERVAL_IN_SECONDS = 5;
-
private static final Logger LOG = LogManager.getLogger(InternalSchemaInitializer.class);
public InternalSchemaInitializer() {
@@ -72,17 +69,17 @@ public class InternalSchemaInitializer extends Thread {
FrontendNodeType feType = Env.getCurrentEnv().getFeType();
if (feType.equals(FrontendNodeType.INIT) || feType.equals(FrontendNodeType.UNKNOWN)) {
LOG.warn("FE is not ready");
- Thread.sleep(5000);
+ Thread.sleep(Config.resource_not_ready_sleep_seconds * 1000);
continue;
}
Thread.currentThread()
- .join(TABLE_CREATION_RETRY_INTERVAL_IN_SECONDS * 1000L);
+ .join(Config.resource_not_ready_sleep_seconds * 1000L);
createDb();
createTbl();
} catch (Throwable e) {
LOG.warn("Statistics storage initiated failed, will try again later", e);
try {
- Thread.sleep(5000);
+ Thread.sleep(Config.resource_not_ready_sleep_seconds * 1000);
} catch (InterruptedException ex) {
LOG.info("Sleep interrupted. {}", ex.getMessage());
}
@@ -154,7 +151,7 @@ public class InternalSchemaInitializer extends Thread {
}
}
try {
- Thread.sleep(5000);
+ Thread.sleep(Config.resource_not_ready_sleep_seconds * 1000);
} catch (InterruptedException t) {
// IGNORE
}
@@ -200,10 +197,6 @@ public class InternalSchemaInitializer extends Thread {
}
};
- if (Config.isCloudMode() && ((CloudEnv) Env.getCurrentEnv()).getEnableStorageVault()) {
- properties.put(PropertyAnalyzer.PROPERTIES_STORAGE_VAULT_NAME, FeConstants.BUILT_IN_STORAGE_VAULT_NAME);
- }
-
PropertyAnalyzer.getInstance().rewriteForceProperties(properties);
CreateTableStmt createTableStmt = new CreateTableStmt(true, false,
tableName, InternalSchema.getCopiedSchema(statsTableName),
@@ -238,10 +231,6 @@ public class InternalSchemaInitializer extends Thread {
}
};
- if (Config.isCloudMode() && ((CloudEnv) Env.getCurrentEnv()).getEnableStorageVault()) {
- properties.put(PropertyAnalyzer.PROPERTIES_STORAGE_VAULT_NAME, FeConstants.BUILT_IN_STORAGE_VAULT_NAME);
- }
-
PropertyAnalyzer.getInstance().rewriteForceProperties(properties);
CreateTableStmt createTableStmt = new CreateTableStmt(true, false,
tableName, InternalSchema.getCopiedSchema(AuditLoaderPlugin.AUDIT_LOG_TABLE),
diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudEnv.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudEnv.java
index 1ebda4dc8aa..e212a7f948e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudEnv.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudEnv.java
@@ -38,11 +38,13 @@ import org.apache.doris.common.ErrorCode;
import org.apache.doris.common.MetaNotFoundException;
import org.apache.doris.common.UserException;
import org.apache.doris.common.io.CountingDataOutputStream;
+import org.apache.doris.ha.FrontendNodeType;
import org.apache.doris.mysql.privilege.PrivPredicate;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.system.SystemInfoService.HostInfo;
import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -89,6 +91,21 @@ public class CloudEnv extends Env {
return this.upgradeMgr;
}
+ @Override
+ public void initialize(String[] args) throws Exception {
+ if (Strings.isNullOrEmpty(Config.cloud_unique_id)) {
+ if (Strings.isNullOrEmpty(Config.cloud_instance_id)) {
+ throw new UserException("cloud_instance_id must be specified if deployed in dissaggregated");
+ }
+ LOG.info("cloud_unique_id is not set, setting it using instance_id");
+ Config.cloud_unique_id = "1:" + Config.cloud_instance_id + ":sql_server00";
+ }
+
+ LOG.info("Initializing CloudEnv with cloud_unique_id: {}", Config.cloud_unique_id);
+
+ super.initialize(args);
+ }
+
@Override
protected void startMasterOnlyDaemonThreads() {
LOG.info("start cloud Master only daemon threads");
@@ -117,9 +134,13 @@ public class CloudEnv extends Env {
return cacheHotspotMgr;
}
+ private CloudSystemInfoService getCloudSystemInfoService() {
+ return (CloudSystemInfoService) systemInfo;
+ }
+
private Cloud.NodeInfoPB getLocalTypeFromMetaService() {
// get helperNodes from ms
- Cloud.GetClusterResponse response = ((CloudSystemInfoService) Env.getCurrentSystemInfo())
+ Cloud.GetClusterResponse response = getCloudSystemInfoService()
.getCloudCluster(Config.cloud_sql_server_cluster_name, Config.cloud_sql_server_cluster_id, "");
if (!response.hasStatus() || !response.getStatus().hasCode()
|| response.getStatus().getCode() != Cloud.MetaServiceCode.OK) {
@@ -166,6 +187,22 @@ public class CloudEnv extends Env {
return local.orElse(null);
}
+ private void tryAddMyselToMS() {
+ try {
+ try {
+ if (Strings.isNullOrEmpty(Config.cloud_instance_id)) {
+ throw new DdlException("unable to create instance due to empty cloud_instance_id");
+ }
+ getCloudSystemInfoService().tryCreateInstance(Config.cloud_instance_id,
+ Config.cloud_instance_id, false);
+ } catch (Exception e) {
+ return;
+ }
+ addFrontend(FrontendNodeType.MASTER, selfNode.getHost(), selfNode.getPort());
+ } catch (DdlException e) {
+ LOG.warn("get ddl exception ", e);
+ }
+ }
protected void getClusterIdAndRole() throws IOException {
NodeInfoPB.NodeType type = NodeInfoPB.NodeType.UNKNOWN;
@@ -175,14 +212,19 @@ public class CloudEnv extends Env {
try {
nodeInfoPB = getLocalTypeFromMetaService();
} catch (Exception e) {
- LOG.warn("failed to get local fe's type, sleep 5 s, try again. exception: {}", e.getMessage());
+ LOG.warn("failed to get local fe's type, sleep {} s, try again. exception: {}",
+ Config.resource_not_ready_sleep_seconds, e.getMessage());
}
if (nodeInfoPB == null) {
- LOG.warn("failed to get local fe's type, sleep 5 s, try again.");
+ LOG.warn("failed to get local fe's type, sleep {} s, try again.",
+ Config.resource_not_ready_sleep_seconds);
+ if (isStartFromEmpty()) {
+ tryAddMyselToMS();
+ }
try {
- Thread.sleep(5000);
+ Thread.sleep(Config.resource_not_ready_sleep_seconds * 1000);
} catch (InterruptedException e) {
- LOG.warn("thread sleep Exception", e);
+ LOG.info("interrupted by {}", e);
}
continue;
}
@@ -221,7 +263,7 @@ public class CloudEnv extends Env {
+ "' for cloud cluster '" + clusterName + "'", ErrorCode.ERR_CLUSTER_NO_PERMISSIONS);
}
- if (!((CloudSystemInfoService) Env.getCurrentSystemInfo()).getCloudClusterNames().contains(clusterName)) {
+ if (!getCloudSystemInfoService().getCloudClusterNames().contains(clusterName)) {
if (LOG.isDebugEnabled()) {
LOG.debug("current instance does not have a cluster name :{}", clusterName);
}
@@ -232,9 +274,9 @@ public class CloudEnv extends Env {
public void changeCloudCluster(String clusterName, ConnectContext ctx) throws DdlException {
checkCloudClusterPriv(clusterName);
- ((CloudSystemInfoService) Env.getCurrentSystemInfo()).waitForAutoStart(clusterName);
+ getCloudSystemInfoService().waitForAutoStart(clusterName);
try {
- ((CloudSystemInfoService) Env.getCurrentSystemInfo()).addCloudCluster(clusterName, "");
+ getCloudSystemInfoService().addCloudCluster(clusterName, "");
} catch (UserException e) {
throw new DdlException(e.getMessage(), e.getMysqlErrorCode());
}
@@ -344,4 +386,24 @@ public class CloudEnv extends Env {
public void cancelCloudWarmUp(CancelCloudWarmUpStmt stmt) throws DdlException {
getCacheHotspotMgr().cancel(stmt);
}
+
+ @Override
+ public void addFrontend(FrontendNodeType role, String host, int editLogPort) throws DdlException {
+ getCloudSystemInfoService().addFrontend(role, host, editLogPort);
+ }
+
+ @Override
+ public void dropFrontend(FrontendNodeType role, String host, int port) throws DdlException {
+ if (port == selfNode.getPort() && feType == FrontendNodeType.MASTER
+ && selfNode.getHost().equals(host)) {
+ throw new DdlException("can not drop current master node.");
+ }
+
+ getCloudSystemInfoService().dropFrontend(role, host, port);
+ }
+
+ @Override
+ public void modifyFrontendHostName(String srcHost, int srcPort, String destHost) throws DdlException {
+ throw new DdlException("modify frontend host name is not supported in cloud mode");
+ }
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceClient.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceClient.java
index c9c799687ab..47108b147f5 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceClient.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceClient.java
@@ -403,4 +403,9 @@ public class MetaServiceClient {
}
return blockingStub.finishTabletJob(request);
}
+
+ public Cloud.CreateInstanceResponse
+ createInstance(Cloud.CreateInstanceRequest request) {
+ return blockingStub.createInstance(request);
+ }
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceProxy.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceProxy.java
index 33a4d81feaa..918f0c34189 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceProxy.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceProxy.java
@@ -106,6 +106,7 @@ public class MetaServiceProxy {
}
String address = Config.meta_service_endpoint;
+ address = address.replaceAll("^[\"']|[\"']$", "");
MetaServiceClient service = serviceMap.get(address);
if (service != null && service.isNormalState() && !service.isConnectionAgeExpired()) {
return service;
@@ -547,4 +548,13 @@ public class MetaServiceProxy {
throw new RpcException("", e.getMessage(), e);
}
}
+
+ public Cloud.CreateInstanceResponse createInstance(Cloud.CreateInstanceRequest request) throws RpcException {
+ try {
+ final MetaServiceClient client = getProxy();
+ return client.createInstance(request);
+ } catch (Exception e) {
+ throw new RpcException("", e.getMessage(), e);
+ }
+ }
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java
index 03cbbfe814a..606f52369e5 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java
@@ -29,7 +29,9 @@ import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.FeConstants;
import org.apache.doris.common.Pair;
+import org.apache.doris.common.RandomIdentifierGenerator;
import org.apache.doris.common.UserException;
+import org.apache.doris.common.util.NetUtils;
import org.apache.doris.ha.FrontendNodeType;
import org.apache.doris.metric.MetricRepo;
import org.apache.doris.qe.ConnectContext;
@@ -287,7 +289,7 @@ public class CloudSystemInfoService extends SystemInfoService {
continue;
}
try {
- Env.getCurrentEnv().dropFrontend(fe.getRole(), fe.getHost(), fe.getEditLogPort());
+ Env.getCurrentEnv().dropFrontendFromBDBJE(fe.getRole(), fe.getHost(), fe.getEditLogPort());
LOG.info("dropped cloud frontend={} ", fe);
} catch (DdlException e) {
LOG.warn("failed to drop cloud frontend={} ", fe);
@@ -308,6 +310,103 @@ public class CloudSystemInfoService extends SystemInfoService {
}
}
+ private void alterBackendCluster(List<HostInfo> hostInfos, String clusterId,
+ Cloud.AlterClusterRequest.Operation operation) throws DdlException {
+ if (Strings.isNullOrEmpty(Config.cloud_instance_id)) {
+ throw new DdlException("unable to alter backends due to empty cloud_instance_id");
+ }
+ // Issue rpc to meta to alter node, then fe master would add this node to its frontends
+ Cloud.ClusterPB clusterPB = Cloud.ClusterPB.newBuilder()
+ .setClusterId(clusterId)
+ .setType(Cloud.ClusterPB.Type.COMPUTE)
+ .build();
+
+ for (HostInfo hostInfo : hostInfos) {
+ Cloud.NodeInfoPB nodeInfoPB = Cloud.NodeInfoPB.newBuilder()
+ .setIp(hostInfo.getHost())
+ .setHost(hostInfo.getHost())
+ .setHeartbeatPort(hostInfo.getPort())
+ .setCtime(System.currentTimeMillis() / 1000)
+ .build();
+ clusterPB = clusterPB.toBuilder().addNodes(nodeInfoPB).build();
+ }
+
+ Cloud.AlterClusterRequest request = Cloud.AlterClusterRequest.newBuilder()
+ .setInstanceId(Config.cloud_instance_id)
+ .setOp(operation)
+ .setCluster(clusterPB)
+ .build();
+
+ Cloud.AlterClusterResponse response;
+ try {
+ response = MetaServiceProxy.getInstance().alterCluster(request);
+ LOG.info("alter cluster, request: {}, response: {}", request, response);
+ if (response.getStatus().getCode() != Cloud.MetaServiceCode.OK) {
+ LOG.warn("alter backends not ok, response: {}", response);
+ throw new DdlException("failed to alter backends errorCode: " + response.getStatus().getCode()
+ + " msg: " + response.getStatus().getMsg());
+ }
+ } catch (RpcException e) {
+ throw new DdlException("failed to alter backends", e);
+ }
+ }
+
+ /**
+ * @param hostInfos : backend's ip, hostName and port
+ * @throws DdlException
+ */
+ @Override
+ public void addBackends(List<HostInfo> hostInfos, Map<String, String> tagMap) throws UserException {
+ // issue rpc to meta to add this node, then fe master would add this node to its backends
+
+ String clusterName = tagMap.getOrDefault(Tag.CLOUD_CLUSTER_NAME, Tag.VALUE_DEFAULT_CLOUD_CLUSTER_NAME);
+ if (clusterName.isEmpty()) {
+ throw new UserException("clusterName empty");
+ }
+
+ String clusterId = tryCreateCluster(clusterName, RandomIdentifierGenerator.generateRandomIdentifier(8));
+ alterBackendCluster(hostInfos, clusterId, Cloud.AlterClusterRequest.Operation.ADD_NODE);
+ }
+
+ // final entry of dropping backend
+ // We can not do it in batch, because be may belong to different clusters.
+ // Maybe we can opt it in the future.
+ @Override
+ public void dropBackend(String host, int heartbeatPort) throws DdlException {
+ Backend droppedBackend = getBackendWithHeartbeatPort(host, heartbeatPort);
+ if (droppedBackend == null) {
+ throw new DdlException("backend does not exists[" + NetUtils
+ .getHostPortInAccessibleFormat(host, heartbeatPort) + "]");
+ }
+ String clusterId = droppedBackend.getTagMap().get(Tag.CLOUD_CLUSTER_ID);
+ if (clusterId == null || clusterId.isEmpty()) {
+ throw new DdlException("Failed to get cluster ID for backend: " + droppedBackend.getId());
+ }
+
+ List<HostInfo> hostInfos = new ArrayList<>();
+ hostInfos.add(new HostInfo(host, heartbeatPort));
+
+ alterBackendCluster(hostInfos, clusterId, Cloud.AlterClusterRequest.Operation.DROP_NODE);
+ }
+
+ @Override
+ public void decommissionBackend(Backend backend) throws UserException {
+ String clusterId = backend.getTagMap().get(Tag.CLOUD_CLUSTER_ID);
+ if (clusterId == null || clusterId.isEmpty()) {
+ throw new UserException("Failed to get cluster ID for backend: " + backend.getId());
+ }
+
+ List<HostInfo> hostInfos = new ArrayList<>();
+ hostInfos.add(new HostInfo(backend.getHost(), backend.getHeartbeatPort()));
+ try {
+ alterBackendCluster(hostInfos, clusterId, Cloud.AlterClusterRequest.Operation.DECOMMISSION_NODE);
+ } catch (DdlException e) {
+ String errorMessage = e.getMessage();
+ LOG.warn("Failed to decommission backend: {}", errorMessage);
+ throw new UserException(errorMessage);
+ }
+ }
+
@Override
public void replayAddBackend(Backend newBackend) {
super.replayAddBackend(newBackend);
@@ -663,6 +762,115 @@ public class CloudSystemInfoService extends SystemInfoService {
}
}
+ // FrontendCluster = SqlServerCluster
+ private void alterFrontendCluster(FrontendNodeType role, String host, int editLogPort,
+ Cloud.AlterClusterRequest.Operation op) throws DdlException {
+ if (Strings.isNullOrEmpty(Config.cloud_instance_id)) {
+ throw new DdlException("unable to alter frontend due to empty cloud_instance_id");
+ }
+
+ // Issue rpc to meta to add this node, then fe master would add this node to its frontends
+ Cloud.NodeInfoPB nodeInfoPB = Cloud.NodeInfoPB.newBuilder()
+ .setIp(host)
+ .setHost(host)
+ .setEditLogPort(editLogPort)
+ .setNodeType(role == FrontendNodeType.MASTER ? Cloud.NodeInfoPB.NodeType.FE_MASTER
+ : Cloud.NodeInfoPB.NodeType.FE_OBSERVER)
+ .setCtime(System.currentTimeMillis() / 1000)
+ .build();
+
+ Cloud.ClusterPB clusterPB = Cloud.ClusterPB.newBuilder()
+ .setClusterId(Config.cloud_sql_server_cluster_id)
+ .setClusterName(Config.cloud_sql_server_cluster_name)
+ .setType(Cloud.ClusterPB.Type.SQL)
+ .addNodes(nodeInfoPB)
+ .build();
+
+ Cloud.AlterClusterRequest request = Cloud.AlterClusterRequest.newBuilder()
+ .setInstanceId(Config.cloud_instance_id)
+ .setOp(op)
+ .setCluster(clusterPB)
+ .build();
+
+ Cloud.AlterClusterResponse response;
+ try {
+ response = MetaServiceProxy.getInstance().alterCluster(request);
+ LOG.info("alter cluster, request: {}, response: {}", request, response);
+ if (response.getStatus().getCode() != Cloud.MetaServiceCode.OK) {
+ LOG.warn("alter frontend not ok, response: {}", response);
+ throw new DdlException("failed to alter frontend errorCode: " + response.getStatus().getCode()
+ + " msg: " + response.getStatus().getMsg());
+ }
+ } catch (RpcException e) {
+ throw new DdlException("failed to alter frontend", e);
+ }
+ }
+
+ public void addFrontend(FrontendNodeType role, String host, int editLogPort) throws DdlException {
+ Cloud.AlterClusterRequest.Operation op;
+ op = role == FrontendNodeType.MASTER ? Cloud.AlterClusterRequest.Operation.ADD_CLUSTER
+ : Cloud.AlterClusterRequest.Operation.ADD_NODE;
+ alterFrontendCluster(role, host, editLogPort, op);
+ }
+
+ public void dropFrontend(FrontendNodeType role, String host, int editLogPort) throws DdlException {
+ alterFrontendCluster(role, host, editLogPort, Cloud.AlterClusterRequest.Operation.DROP_NODE);
+ }
+
+ private String tryCreateCluster(String clusterName, String clusterId) throws UserException {
+ if (Strings.isNullOrEmpty(Config.cloud_instance_id)) {
+ throw new DdlException("unable to create cluster due to empty cloud_instance_id");
+ }
+
+ Cloud.ClusterPB clusterPB = Cloud.ClusterPB.newBuilder()
+ .setClusterId(clusterId)
+ .setClusterName(clusterName)
+ .setType(Cloud.ClusterPB.Type.COMPUTE)
+ .build();
+
+ Cloud.AlterClusterRequest request = Cloud.AlterClusterRequest.newBuilder()
+ .setCloudUniqueId(Config.cloud_unique_id)
+ .setOp(Cloud.AlterClusterRequest.Operation.ADD_CLUSTER)
+ .setCluster(clusterPB)
+ .build();
+
+ Cloud.AlterClusterResponse response;
+ try {
+ response = MetaServiceProxy.getInstance().alterCluster(request);
+ LOG.info("alter cluster, request: {}, response: {}", request, response);
+ if (response.getStatus().getCode() != Cloud.MetaServiceCode.OK
+ && response.getStatus().getCode() != Cloud.MetaServiceCode.ALREADY_EXISTED) {
+ LOG.warn("create cluster not ok, response: {}", response);
+ throw new UserException("failed to create cluster errorCode: " + response.getStatus().getCode()
+ + " msg: " + response.getStatus().getMsg());
+ }
+
+ if (response.getStatus().getCode() == Cloud.MetaServiceCode.OK) {
+ return clusterId;
+ } else if (response.getStatus().getCode() == Cloud.MetaServiceCode.ALREADY_EXISTED) {
+ Cloud.GetClusterResponse clusterResponse = getCloudCluster(clusterName, "", "");
+ if (clusterResponse.getStatus().getCode() == Cloud.MetaServiceCode.OK) {
+ if (clusterResponse.getClusterCount() > 0) {
+ Cloud.ClusterPB cluster = clusterResponse.getCluster(0);
+ return cluster.getClusterId();
+ } else {
+ throw new UserException("Cluster information not found in the response");
+ }
+ } else {
+ throw new UserException("Failed to get cluster. Error code: "
+ + clusterResponse.getStatus().getCode() + ", message: "
+ + clusterResponse.getStatus().getMsg());
+ }
+ } else {
+ throw new UserException("Failed to create or get cluster. Error code: "
+ + response.getStatus().getCode() + ", message: " + response.getStatus().getMsg());
+ }
+ } catch (RpcException e) {
+ throw new UserException("failed to create cluster", e);
+ }
+
+ }
+
public List<Pair<String, Integer>> getCurrentObFrontends() {
List<Frontend> frontends = Env.getCurrentEnv().getFrontends(FrontendNodeType.OBSERVER);
List<Pair<String, Integer>> frontendsPair = new ArrayList<>();
@@ -778,7 +986,9 @@ public class CloudSystemInfoService extends SystemInfoService {
Cloud.AlterClusterResponse response;
try {
- response = MetaServiceProxy.getInstance().alterCluster(builder.build());
+ Cloud.AlterClusterRequest request = builder.build();
+ response = MetaServiceProxy.getInstance().alterCluster(request);
+ LOG.info("alter cluster, request: {}, response: {}", request, response);
if (response.getStatus().getCode() != Cloud.MetaServiceCode.OK) {
LOG.warn("notify to resume cluster not ok, cluster {}, response: {}", clusterName, response);
}
@@ -829,4 +1039,28 @@ public class CloudSystemInfoService extends SystemInfoService {
}
return clusterName;
}
+
+ public void tryCreateInstance(String instanceId, String name, boolean sseEnabled) throws DdlException {
+ Cloud.CreateInstanceRequest.Builder builder = Cloud.CreateInstanceRequest.newBuilder();
+ builder.setInstanceId(instanceId);
+ builder.setName(name);
+ builder.setSseEnabled(sseEnabled);
+
+ Cloud.CreateInstanceResponse response;
+ try {
+
+ Cloud.CreateInstanceRequest request = builder.build();
+ response = MetaServiceProxy.getInstance().createInstance(request);
+ LOG.info("create instance, request: {}, response: {}", request, response);
+ if (response.getStatus().getCode() != Cloud.MetaServiceCode.OK
+ && response.getStatus().getCode() != Cloud.MetaServiceCode.ALREADY_EXISTED) {
+ LOG.warn("Failed to create instance {}, response: {}", instanceId, response);
+ throw new DdlException("Failed to create instance");
+ }
+ LOG.info("Successfully created instance {}, response: {}", instanceId, response);
+ } catch (RpcException e) {
+ LOG.warn("Failed to create instance {}", instanceId, e);
+ throw new DdlException("Failed to create instance");
+ }
+ }
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/RandomIdentifierGenerator.java b/fe/fe-core/src/main/java/org/apache/doris/common/RandomIdentifierGenerator.java
new file mode 100644
index 00000000000..38fee894eb3
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/common/RandomIdentifierGenerator.java
@@ -0,0 +1,51 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.common;
+
+import java.security.SecureRandom;
+
+public class RandomIdentifierGenerator {
+
+ private static final String ALPHABET = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ";
+ private static final String ALPHANUMERIC = ALPHABET + "0123456789_";
+ private static final SecureRandom RANDOM = new SecureRandom();
+
+ /**
+ * Generates a random identifier matching the pattern "^[a-zA-Z][a-zA-Z0-9_]*$".
+ *
+ * @param length The desired length of the identifier (must be at least 1)
+ * @return A randomly generated identifier
+ */
+ public static String generateRandomIdentifier(int length) {
+ if (length < 1) {
+ throw new IllegalArgumentException("Length must be at least 1");
+ }
+
+ StringBuilder sb = new StringBuilder(length);
+
+ // First character must be a letter
+ sb.append(ALPHABET.charAt(RANDOM.nextInt(ALPHABET.length())));
+
+ // Remaining characters can be alphanumeric or underscore
+ for (int i = 1; i < length; i++) {
+ sb.append(ALPHANUMERIC.charAt(RANDOM.nextInt(ALPHANUMERIC.length())));
+ }
+
+ return sb.toString();
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/util/PropertyAnalyzer.java b/fe/fe-core/src/main/java/org/apache/doris/common/util/PropertyAnalyzer.java
index 760ee10542a..27fd06cd6cf 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/common/util/PropertyAnalyzer.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/common/util/PropertyAnalyzer.java
@@ -1235,7 +1235,7 @@ public class PropertyAnalyzer {
tagMap.put(tag.type, tag.value);
iter.remove();
}
- if (tagMap.isEmpty() && defaultValue != null) {
+ if (defaultValue != null && !tagMap.containsKey(defaultValue.type)) {
tagMap.put(defaultValue.type, defaultValue.value);
}
return tagMap;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java
index 2b7f6090bb9..033af006dd8 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java
@@ -2681,6 +2681,7 @@ public class InternalCatalog implements CatalogIf<Database> {
if (info != null) {
storageVaultName = info.first;
storageVaultId = info.second;
+ LOG.info("Using default storage vault: name={}, id={}", storageVaultName, storageVaultId);
} else {
throw new DdlException("No default storage vault."
+ " You can use `SHOW STORAGE VAULT` to get all available vaults,"
@@ -2704,9 +2705,12 @@ public class InternalCatalog implements CatalogIf<Database> {
olapTable.setStorageVaultName(storageVaultName);
storageVaultId = env.getStorageVaultMgr().getVaultIdByName(storageVaultName);
- if (storageVaultId != null && !storageVaultId.isEmpty()) {
- olapTable.setStorageVaultId(storageVaultId);
+ if (Strings.isNullOrEmpty(storageVaultId)) {
+ throw new DdlException("Storage vault '" + storageVaultName + "' does not exist. "
+ + "You can use `SHOW STORAGE VAULT` to get all available vaults, "
+ + "or create a new one with `CREATE STORAGE VAULT`.");
}
+ olapTable.setStorageVaultId(storageVaultId);
}
// check `update on current_timestamp`
diff --git a/fe/fe-core/src/main/java/org/apache/doris/deploy/DeployManager.java b/fe/fe-core/src/main/java/org/apache/doris/deploy/DeployManager.java
index 52d0b814b00..11230daf973 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/deploy/DeployManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/deploy/DeployManager.java
@@ -549,10 +549,10 @@ public class DeployManager extends MasterDaemon {
try {
switch (nodeType) {
case ELECTABLE:
- env.addFrontend(FrontendNodeType.FOLLOWER, remoteHost, remotePort, "");
+ env.addFrontend(FrontendNodeType.FOLLOWER, remoteHost, remotePort);
break;
case OBSERVER:
- env.addFrontend(FrontendNodeType.OBSERVER, remoteHost, remotePort, "");
+ env.addFrontend(FrontendNodeType.OBSERVER, remoteHost, remotePort);
break;
case BACKEND:
case BACKEND_CN:
diff --git a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/manager/NodeAction.java b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/manager/NodeAction.java
index 814b00b49ac..4658f936db5 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/manager/NodeAction.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/manager/NodeAction.java
@@ -661,7 +661,7 @@ public class NodeAction extends RestBaseController {
}
HostInfo info = SystemInfoService.getHostAndPort(reqInfo.getHostPort());
if ("ADD".equals(action)) {
- currentEnv.addFrontend(frontendNodeType, info.getHost(), info.getPort(), "");
+ currentEnv.addFrontend(frontendNodeType, info.getHost(), info.getPort());
} else if ("DROP".equals(action)) {
currentEnv.dropFrontend(frontendNodeType, info.getHost(), info.getPort());
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/DdlExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/DdlExecutor.java
index bdc7b765101..8bdbc57f999 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/DdlExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/DdlExecutor.java
@@ -263,10 +263,10 @@ public class DdlExecutor {
env.getAuth().updateUserProperty((SetUserPropertyStmt) ddlStmt);
} else if (ddlStmt instanceof AlterSystemStmt) {
AlterSystemStmt stmt = (AlterSystemStmt) ddlStmt;
- env.alterCluster(stmt);
+ env.alterSystem(stmt);
} else if (ddlStmt instanceof CancelAlterSystemStmt) {
CancelAlterSystemStmt stmt = (CancelAlterSystemStmt) ddlStmt;
- env.cancelAlterCluster(stmt);
+ env.cancelAlterSystem(stmt);
} else if (ddlStmt instanceof AlterDatabaseQuotaStmt) {
env.alterDatabaseQuota((AlterDatabaseQuotaStmt) ddlStmt);
} else if (ddlStmt instanceof AlterDatabaseRename) {
@@ -565,7 +565,7 @@ public class DdlExecutor {
|| ddlStmt instanceof AdminCancelRebalanceDiskStmt
|| ddlStmt instanceof AlterResourceStmt
|| ddlStmt instanceof AlterPolicyStmt
- || ddlStmt instanceof AlterSystemStmt) {
+ || ddlStmt instanceof CancelAlterSystemStmt) {
LOG.info("stmt={}, not supported in cloud mode", ddlStmt.toString());
throw new DdlException("Unsupported operation");
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/resource/Tag.java b/fe/fe-core/src/main/java/org/apache/doris/resource/Tag.java
index a51755412b9..b88cf84282f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/resource/Tag.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/resource/Tag.java
@@ -73,6 +73,8 @@ public class Tag implements Writable {
public static final String CLOUD_CLUSTER_PRIVATE_ENDPOINT = "cloud_cluster_private_endpoint";
public static final String CLOUD_CLUSTER_STATUS = "cloud_cluster_status";
+ public static final String VALUE_DEFAULT_CLOUD_CLUSTER_NAME = "default_cluster";
+
public static final String WORKLOAD_GROUP = "workload_group";
public static final ImmutableSet<String> RESERVED_TAG_TYPE = ImmutableSet.of(
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/util/StatisticsUtil.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/util/StatisticsUtil.java
index 0f5c81b1cf0..f01c1a13d1b 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/util/StatisticsUtil.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/util/StatisticsUtil.java
@@ -480,10 +480,13 @@ public class StatisticsUtil {
// dbName,
// StatisticConstants.HISTOGRAM_TBL_NAME));
} catch (Throwable t) {
+ LOG.info("stat table does not exist, db_name: {}, table_name: {}", dbName,
+ StatisticConstants.TABLE_STATISTIC_TBL_NAME);
return false;
}
if (Config.isCloudMode()) {
if (!((CloudSystemInfoService) Env.getCurrentSystemInfo()).availableBackendsExists()) {
+ LOG.info("there are no available backends");
return false;
}
try (AutoCloseConnectContext r = buildConnectContext()) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/system/HeartbeatMgr.java b/fe/fe-core/src/main/java/org/apache/doris/system/HeartbeatMgr.java
index 515db76b096..5dd8dd9fca1 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/system/HeartbeatMgr.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/system/HeartbeatMgr.java
@@ -95,6 +95,14 @@ public class HeartbeatMgr extends MasterDaemon {
tMasterInfo.setHttpPort(Config.http_port);
long flags = heartbeatFlags.getHeartbeatFlags();
tMasterInfo.setHeartbeatFlags(flags);
+ if (Config.isCloudMode()) {
+ // Set cloud_instance_id and meta_service_endpoint even if there are empty
+ // Be can knowns that fe is working in cloud mode.
+ // Set the cloud instance ID for cloud deployment identification
+ tMasterInfo.setCloudInstanceId(Config.cloud_instance_id);
+ // Set the endpoint for the metadata service in cloud mode
+ tMasterInfo.setMetaServiceEndpoint(Config.meta_service_endpoint);
+ }
masterInfo.set(tMasterInfo);
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java b/fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java
index f81d8b4d7b0..6126cc7a702 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java
@@ -290,6 +290,15 @@ public class SystemInfoService {
MetricRepo.generateBackendsTabletMetrics();
}
+ public void decommissionBackend(Backend backend) throws UserException {
+ // set backend's state as 'decommissioned'
+ // for decommission operation, here is no decommission job. the system handler will check
+ // all backend in decommission state
+ backend.setDecommissioned(true);
+ Env.getCurrentEnv().getEditLog().logBackendStateChange(backend);
+ LOG.info("set backend {} to decommission", backend.getId());
+ }
+
// only for test
public void dropAllBackend() {
// update idToBackend
diff --git a/fe/fe-core/src/test/java/org/apache/doris/clone/DecommissionTest.java b/fe/fe-core/src/test/java/org/apache/doris/clone/DecommissionTest.java
index 81e29d3abf3..6901d3b604c 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/clone/DecommissionTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/clone/DecommissionTest.java
@@ -144,7 +144,7 @@ public class DecommissionTest {
+ ":" + backend.getHeartbeatPort() + "\"";
AlterSystemStmt decommissionStmt =
(AlterSystemStmt) UtFrameUtils.parseAndAnalyzeStmt(decommissionStmtStr, connectContext);
- Env.getCurrentEnv().getAlterInstance().processAlterCluster(decommissionStmt);
+ Env.getCurrentEnv().getAlterInstance().processAlterSystem(decommissionStmt);
Assert.assertEquals(true, backend.isDecommissioned());
diff --git a/fe/fe-core/src/test/java/org/apache/doris/cluster/DecommissionBackendTest.java b/fe/fe-core/src/test/java/org/apache/doris/cluster/DecommissionBackendTest.java
index fc80ecbd97d..53355f6faa4 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/cluster/DecommissionBackendTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/cluster/DecommissionBackendTest.java
@@ -100,7 +100,7 @@ public class DecommissionBackendTest extends TestWithFeService {
Assertions.assertNotNull(srcBackend);
String decommissionStmtStr = "alter system decommission backend \"" + srcBackend.getAddress() + "\"";
AlterSystemStmt decommissionStmt = (AlterSystemStmt) parseAndAnalyzeStmt(decommissionStmtStr);
- Env.getCurrentEnv().getAlterInstance().processAlterCluster(decommissionStmt);
+ Env.getCurrentEnv().getAlterInstance().processAlterSystem(decommissionStmt);
Assertions.assertTrue(srcBackend.isDecommissioned());
long startTimestamp = System.currentTimeMillis();
@@ -153,7 +153,7 @@ public class DecommissionBackendTest extends TestWithFeService {
// decommission backend by id
String decommissionByIdStmtStr = "alter system decommission backend \"" + srcBackend.getId() + "\"";
AlterSystemStmt decommissionByIdStmt = (AlterSystemStmt) parseAndAnalyzeStmt(decommissionByIdStmtStr);
- Env.getCurrentEnv().getAlterInstance().processAlterCluster(decommissionByIdStmt);
+ Env.getCurrentEnv().getAlterInstance().processAlterSystem(decommissionByIdStmt);
Assertions.assertTrue(srcBackend.isDecommissioned());
long startTimestamp = System.currentTimeMillis();
@@ -215,7 +215,7 @@ public class DecommissionBackendTest extends TestWithFeService {
// 6. execute decommission
String decommissionStmtStr = "alter system decommission backend \"" + srcBackend.getAddress() + "\"";
AlterSystemStmt decommissionStmt = (AlterSystemStmt) parseAndAnalyzeStmt(decommissionStmtStr);
- Env.getCurrentEnv().getAlterInstance().processAlterCluster(decommissionStmt);
+ Env.getCurrentEnv().getAlterInstance().processAlterSystem(decommissionStmt);
Assertions.assertTrue(srcBackend.isDecommissioned());
long startTimestamp = System.currentTimeMillis();
@@ -315,7 +315,7 @@ public class DecommissionBackendTest extends TestWithFeService {
String decommissionStmtStr = "alter system decommission backend \"" + srcBackend.getAddress() + "\"";
AlterSystemStmt decommissionStmt = (AlterSystemStmt) parseAndAnalyzeStmt(decommissionStmtStr);
- Env.getCurrentEnv().getAlterInstance().processAlterCluster(decommissionStmt);
+ Env.getCurrentEnv().getAlterInstance().processAlterSystem(decommissionStmt);
Assertions.assertTrue(srcBackend.isDecommissioned());
long startTimestamp = System.currentTimeMillis();
diff --git a/gensrc/thrift/HeartbeatService.thrift b/gensrc/thrift/HeartbeatService.thrift
index 4daea779735..7c94bc9ce0c 100644
--- a/gensrc/thrift/HeartbeatService.thrift
+++ b/gensrc/thrift/HeartbeatService.thrift
@@ -39,6 +39,8 @@ struct TMasterInfo {
7: optional i64 heartbeat_flags
8: optional i64 backend_id
9: optional list<TFrontendInfo> frontend_infos
+ 10: optional string meta_service_endpoint;
+ 11: optional string cloud_instance_id;
}
struct TBackendInfo {
diff --git a/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy b/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy
index 9272c818199..96020f1a35b 100644
--- a/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy
+++ b/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy
@@ -303,6 +303,7 @@ class Suite implements GroovyInterceptable {
Thread.sleep(1000)
}
+ logger.info("get fe {}", fe)
assertNotNull(fe)
if (!dockerIsCloud) {
for (def be : cluster.getAllBackends()) {
@@ -319,8 +320,8 @@ class Suite implements GroovyInterceptable {
def sql = "CREATE DATABASE IF NOT EXISTS " + context.dbName
logger.info("try create database if not exists {}", context.dbName)
JdbcUtils.executeToList(conn, sql)
- url = Config.buildUrlWithDb(url, context.dbName)
+ url = Config.buildUrlWithDb(url, context.dbName)
logger.info("connect to docker cluster: suite={}, url={}", name, url)
connect(user, password, url, actionSupplier)
} finally {
diff --git a/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/SuiteCluster.groovy b/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/SuiteCluster.groovy
index 44220500d1b..fad99702068 100644
--- a/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/SuiteCluster.groovy
+++ b/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/SuiteCluster.groovy
@@ -17,8 +17,9 @@
package org.apache.doris.regression.suite
import org.apache.doris.regression.Config
-import org.apache.doris.regression.util.Http
import org.apache.doris.regression.util.DebugPoint
+import org.apache.doris.regression.util.Http
+import org.apache.doris.regression.util.JdbcUtils
import org.apache.doris.regression.util.NodeType
import com.google.common.collect.Maps
@@ -29,12 +30,19 @@ import groovy.json.JsonSlurper
import groovy.transform.CompileStatic
import groovy.util.logging.Slf4j
import java.util.stream.Collectors
+import java.sql.Connection
class ClusterOptions {
int feNum = 1
int beNum = 3
+ Boolean sqlModeNodeMgr = false
+ Boolean beMetaServiceEndpoint = true
+ Boolean beCloudInstanceId = false
+
+ int waitTimeout = 180
+
List<String> feConfigs = [
'heartbeat_interval_second=5',
]
@@ -150,6 +158,7 @@ class ServerNode {
class Frontend extends ServerNode {
+ int editLogPort
int queryPort
boolean isMaster
@@ -157,6 +166,7 @@ class Frontend extends ServerNode {
Frontend fe = new Frontend()
ServerNode.fromCompose(fe, header, index, fields)
fe.queryPort = (Integer) fields.get(header.indexOf('query_port'))
+ fe.editLogPort = (Integer) fields.get(header.indexOf('edit_log_port'))
fe.isMaster = fields.get(header.indexOf('is_master')) == 'true'
return fe
}
@@ -177,14 +187,17 @@ class Frontend extends ServerNode {
class Backend extends ServerNode {
+ int heartbeatPort
long backendId
int tabletNum
static Backend fromCompose(ListHeader header, int index, List<Object> fields) {
Backend be = new Backend()
ServerNode.fromCompose(be, header, index, fields)
+ be.heartbeatPort = (Integer) fields.get(header.indexOf('heartbeat_port'))
be.backendId = toLongOrDefault(fields.get(header.indexOf('backend_id')), -1L)
be.tabletNum = (int) toLongOrDefault(fields.get(header.indexOf('tablet_num')), 0L)
+
return be
}
@@ -255,6 +268,7 @@ class SuiteCluster {
final String name
final Config config
private boolean running
+ private boolean sqlModeNodeMgr = false;
SuiteCluster(String name, Config config) {
this.name = name
@@ -301,7 +315,19 @@ class SuiteCluster {
cmd += ['--fe-follower']
}
- cmd += ['--wait-timeout', String.valueOf(180)]
+ if (options.sqlModeNodeMgr) {
+ cmd += ['--sql-mode-node-mgr']
+ }
+ if (!options.beMetaServiceEndpoint) {
+ cmd += ['--no-be-metaservice-endpoint']
+ }
+ if (!options.beCloudInstanceId) {
+ cmd += ['--no-be-cloud-instanceid']
+ }
+
+ cmd += ['--wait-timeout', String.valueOf(options.waitTimeout)]
+
+ sqlModeNodeMgr = options.sqlModeNodeMgr;
runCmd(cmd.join(' '), -1)
@@ -405,6 +431,7 @@ class SuiteCluster {
def data = runCmd(cmd)
assert data instanceof List
def rows = (List<List<Object>>) data
+ logger.info("get all nodes {}", rows);
def header = new ListHeader(rows.get(0))
for (int i = 1; i < rows.size(); i++) {
def row = (List<Object>) rows.get(i)
diff --git a/regression-test/suites/cloud_p0/node_mgr/test_sql_mode_node_mgr.groovy b/regression-test/suites/cloud_p0/node_mgr/test_sql_mode_node_mgr.groovy
new file mode 100644
index 00000000000..c79219aeac2
--- /dev/null
+++ b/regression-test/suites/cloud_p0/node_mgr/test_sql_mode_node_mgr.groovy
@@ -0,0 +1,388 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import org.apache.doris.regression.suite.ClusterOptions
+import groovy.json.JsonSlurper
+
+suite('test_sql_mode_node_mgr', 'docker,p1') {
+ if (!isCloudMode()) {
+ return;
+ }
+
+ def clusterOptions = [
+ new ClusterOptions(),
+ new ClusterOptions(),
+ ]
+
+ for (options in clusterOptions) {
+ options.feConfigs += [
+ 'cloud_cluster_check_interval_second=1',
+ ]
+ options.cloudMode = true
+ options.sqlModeNodeMgr = true
+ options.waitTimeout = 0
+ options.feNum = 3
+ options.feConfigs += ["resource_not_ready_sleep_seconds=1",
+ "heartbeat_interval_second=1",]
+ }
+
+ clusterOptions[0].beCloudInstanceId = true;
+ clusterOptions[0].beMetaServiceEndpoint = true;
+
+ clusterOptions[1].beCloudInstanceId = false;
+ clusterOptions[1].beMetaServiceEndpoint = false;
+
+ for (options in clusterOptions) {
+ docker(options) {
+ logger.info("docker started");
+
+ def checkFrontendsAndBackends = {
+ // Check frontends
+ def frontendResult = sql_return_maparray """show frontends;"""
+ logger.info("show frontends result {}", frontendResult)
+ // Check that we have the expected number of frontends
+ assert frontendResult.size() == 3, "Expected 3 frontends, but got ${frontendResult.size()}"
+
+ // Check that all required columns are present
+ def requiredColumns = ['Name', 'Host', 'EditLogPort', 'HttpPort', 'QueryPort', 'RpcPort', 'Role', 'IsMaster', 'ClusterId', 'Join', 'Alive', 'ReplayedJournalId', 'LastHeartbeat', 'IsHelper', 'ErrMsg']
+ def actualColumns = frontendResult[0].keySet()
+ assert actualColumns.containsAll(requiredColumns), "Missing required columns. Expected: ${requiredColumns}, Actual: ${actualColumns}"
+
+ // Check that we have one master and two followers
+ def masterCount = frontendResult.count { it['IsMaster'] == 'true' }
+ assert masterCount == 1, "Expected 1 master, but got ${masterCount}"
+
+ def followerCount = frontendResult.count { it['IsMaster'] == 'false' }
+ assert followerCount == 2, "Expected 2 followers, but got ${followerCount}"
+
+ // Check that all frontends are alive
+ def aliveCount = frontendResult.count { it['Alive'] == 'true' }
+ assert aliveCount == 3, "Expected all 3 frontends to be alive, but only ${aliveCount} are alive"
+
+ // Check backends
+ def backendResult = sql_return_maparray """show backends;"""
+ logger.info("show backends result {}", backendResult)
+ // Check that we have the expected number of backends
+ assert backendResult.size() == 3, "Expected 3 backends, but got ${backendResult.size()}"
+
+ // Check that all required columns are present
+ def requiredBackendColumns = ['Host', 'HeartbeatPort', 'BePort', 'HttpPort', 'BrpcPort', 'LastStartTime', 'LastHeartbeat', 'Alive', 'SystemDecommissioned', 'TabletNum', 'DataUsedCapacity', 'AvailCapacity', 'TotalCapacity', 'UsedPct', 'MaxDiskUsedPct', 'Tag', 'ErrMsg', 'Version', 'Status']
+ def actualBackendColumns = backendResult[0].keySet()
+ assert actualBackendColumns.containsAll(requiredBackendColumns), "Missing required backend columns. Expected: ${requiredBackendColumns}, Actual: ${actualBackendColumns}"
+
+ // Check that all backends are alive
+ def aliveBackendCount = backendResult.count { it['Alive'] == 'true' }
+ assert aliveBackendCount == 3, "Expected all 3 backends to be alive, but only ${aliveBackendCount} are alive"
+
+ // Check that no backends are decommissioned
+ def decommissionedCount = backendResult.count { it['SystemDecommissioned'] == 'true' || it['ClusterDecommissioned'] == 'true' }
+ assert decommissionedCount == 0, "Expected no decommissioned backends, but found ${decommissionedCount}"
+
+ // Check that all backends have valid capacities
+ backendResult.each { backend ->
+ assert backend['DataUsedCapacity'] != null && backend['AvailCapacity'] != null && backend['TotalCapacity'] != null, "Backend ${backend['BackendId']} has invalid capacity values"
+ assert backend['UsedPct'] != null && backend['MaxDiskUsedPct'] != null, "Backend ${backend['BackendId']} has invalid disk usage percentages"
+ }
+
+ logger.info("All backend checks passed successfully")
+ }
+
+ // Call the function to check frontends and backends
+ checkFrontendsAndBackends()
+
+ def checkClusterStatus = { int expectedFeNum, int expectedBeNum, int existsRows ->
+ logger.info("Checking cluster status...")
+
+ // Check FE number
+ def frontendResult = sql_return_maparray """SHOW FRONTENDS;"""
+ assert frontendResult.size() == expectedFeNum, "Expected ${expectedFeNum} frontends, but got ${frontendResult.size()}"
+ logger.info("FE number check passed: ${frontendResult.size()} FEs found")
+
+ // Check BE number
+ def backendResult = sql_return_maparray """SHOW BACKENDS;"""
+ assert backendResult.size() == expectedBeNum, "Expected ${expectedBeNum} backends, but got ${backendResult.size()}"
+ logger.info("BE number check passed: ${backendResult.size()} BEs found")
+
+ // Create table if not exists
+ sql """
+ CREATE TABLE IF NOT EXISTS test_table (
+ id INT,
+ name VARCHAR(50)
+ )
+ DISTRIBUTED BY HASH(id) BUCKETS 3
+ PROPERTIES("replication_num" = "3");
+ """
+
+ logger.info("Table 'test_table' created or already exists")
+ sql """ INSERT INTO test_table VALUES (1, 'test1') """
+
+ def result = sql """ SELECT * FROM test_table ORDER BY id """
+ assert result.size() == existsRows + 1, "Expected ${existsRows + 1} rows, but got ${result.size()}"
+ logger.info("Read/write check passed: ${result.size()} rows found")
+
+ logger.info("All cluster status checks passed successfully")
+ }
+
+ // Call the function to check cluster status
+ checkClusterStatus(3, 3, 0)
+
+ // CASE 1 . check restarting fe and be work.
+ logger.info("Restarting frontends and backends...")
+ cluster.restartFrontends();
+ cluster.restartBackends();
+
+ sleep(30000)
+ context.reconnectFe()
+
+ checkClusterStatus(3, 3, 1)
+
+ // CASE 2. If a be is dropped, query and writing also work.
+ // Get the list of backends
+ def backends = sql_return_maparray("SHOW BACKENDS")
+ logger.info("Current backends: {}", backends)
+
+ // Find a backend to drop
+ def backendToDrop = backends[0]
+ def backendHost = backendToDrop['Host']
+ def backendHeartbeatPort = backendToDrop['HeartbeatPort']
+
+ logger.info("Dropping backend: {}:{}", backendHost, backendHeartbeatPort)
+
+ // Drop the selected backend
+ sql """ ALTER SYSTEM DROPP BACKEND "${backendHost}:${backendHeartbeatPort}"; """
+
+ // Wait for the backend to be fully dropped
+ def maxWaitSeconds = 300
+ def waited = 0
+ while (waited < maxWaitSeconds) {
+ def currentBackends = sql_return_maparray("SHOW BACKENDS")
+ if (currentBackends.size() == 2) {
+ logger.info("Backend successfully dropped")
+ break
+ }
+ sleep(10000)
+ waited += 10
+ }
+
+ if (waited >= maxWaitSeconds) {
+ throw new Exception("Timeout waiting for backend to be dropped")
+ }
+
+ checkClusterStatus(3, 2, 2)
+
+ // CASE 3. Add the dropped backend back
+ logger.info("Adding back the dropped backend: {}:{}", backendHost, backendHeartbeatPort)
+ sql """ ALTER SYSTEM ADD BACKEND "${backendHost}:${backendHeartbeatPort}"; """
+
+ // Wait for the backend to be fully added back
+ maxWaitSeconds = 300
+ waited = 0
+ while (waited < maxWaitSeconds) {
+ def currentBackends = sql_return_maparray("SHOW BACKENDS")
+ if (currentBackends.size() == 3) {
+ logger.info("Backend successfully added back")
+ break
+ }
+ sleep(10000)
+ waited += 10
+ }
+
+ if (waited >= maxWaitSeconds) {
+ throw new Exception("Timeout waiting for backend to be added back")
+ }
+
+ checkClusterStatus(3, 3, 3)
+
+ // CASE 4. If a fe is dropped, query and writing also work.
+ // Get the list of frontends
+ def frontends = sql_return_maparray("SHOW FRONTENDS")
+ logger.info("Current frontends: {}", frontends)
+
+ // Find a non-master frontend to drop
+ def feToDropMap = frontends.find { it['IsMaster'] == "false" }
+ assert feToDropMap != null, "No non-master frontend found to drop"
+
+ def feHost = feToDropMap['Host']
+ def feEditLogPort = feToDropMap['EditLogPort']
+
+ logger.info("Dropping non-master frontend: {}:{}", feHost, feEditLogPort)
+
+ // Drop the selected non-master frontend
+ sql """ ALTER SYSTEM DROP FOLLOWER "${feHost}:${feEditLogPort}"; """
+
+ // Wait for the frontend to be fully dropped
+ maxWaitSeconds = 300
+ waited = 0
+ while (waited < maxWaitSeconds) {
+ def currentFrontends = sql_return_maparray("SHOW FRONTENDS")
+ if (currentFrontends.size() == frontends.size() - 1) {
+ logger.info("Non-master frontend successfully dropped")
+ break
+ }
+ sleep(10000)
+ waited += 10
+ }
+
+ if (waited >= maxWaitSeconds) {
+ throw new Exception("Timeout waiting for non-master frontend to be dropped")
+ }
+
+ checkClusterStatus(2, 3, 4)
+
+ // CASE 5. Add dropped frontend back
+ logger.info("Adding dropped frontend back")
+
+ // Get the current list of frontends
+ def currentFrontends = sql_return_maparray("SHOW FRONTENDS")
+
+ // Find the dropped frontend by comparing with the original list
+ def droppedFE = frontends.find { fe ->
+ !currentFrontends.any { it['Host'] == fe['Host'] && it['EditLogPort'] == fe['EditLogPort'] }
+ }
+
+ assert droppedFE != null, "Could not find the dropped frontend"
+
+ feHost = droppedFE['Host']
+ feEditLogPort = droppedFE['EditLogPort']
+
+ logger.info("Adding back frontend: {}:{}", feHost, feEditLogPort)
+
+ // Add the frontend back
+ sql """ ALTER SYSTEM ADD FOLLOWER "${feHost}:${feEditLogPort}"; """
+
+ // Wait for the frontend to be fully added back
+ maxWaitSeconds = 300
+ waited = 0
+ while (waited < maxWaitSeconds) {
+ def updatedFrontends = sql_return_maparray("SHOW FRONTENDS")
+ if (updatedFrontends.size() == frontends.size()) {
+ logger.info("Frontend successfully added back")
+ break
+ }
+ sleep(10000)
+ waited += 10
+ }
+
+ if (waited >= maxWaitSeconds) {
+ throw new Exception("Timeout waiting for frontend to be added back")
+ }
+
+ // Verify cluster status after adding the frontend back
+ checkClusterStatus(3, 3, 5)
+
+ logger.info("Frontend successfully added back and cluster status verified")
+
+ // CASE 6. If fe can not drop itself.
+ // 6. Attempt to drop the master FE and expect an exception
+ logger.info("Attempting to drop the master frontend")
+
+ // Get the master frontend information
+ def masterFE = frontends.find { it['IsMaster'] == "true" }
+ assert masterFE != null, "No master frontend found"
+
+ def masterHost = masterFE['Host']
+ def masterEditLogPort = masterFE['EditLogPort']
+
+ logger.info("Attempting to drop master frontend: {}:{}", masterHost, masterEditLogPort)
+
+ try {
+ sql """ ALTER SYSTEM DROP FOLLOWER "${masterHost}:${masterEditLogPort}"; """
+ throw new Exception("Expected an exception when trying to drop master frontend, but no exception was thrown")
+ } catch (Exception e) {
+ logger.info("Received expected exception when trying to drop master frontend: {}", e.getMessage())
+ assert e.getMessage().contains("can not drop current master node."), "Unexpected exception message when trying to drop master frontend"
+ }
+
+ // Verify that the master frontend is still present
+ currentFrontends = sql_return_maparray("SHOW FRONTENDS")
+ assert currentFrontends.find { it['IsMaster'] == "true" && it['Host'] == masterHost && it['EditLogPort'] == masterEditLogPort } != null, "Master frontend should still be present"
+ logger.info("Successfully verified that the master frontend cannot be dropped")
+
+
+ // CASE 7. Attempt to drop a non-existent backend
+ logger.info("Attempting to drop a non-existent backend")
+
+ // Generate a non-existent host and port
+ def nonExistentHost = "non.existent.host"
+ def nonExistentPort = 12345
+
+ try {
+ sql """ ALTER SYSTEM DROPP BACKEND "${nonExistentHost}:${nonExistentPort}"; """
+ throw new Exception("Expected an exception when trying to drop non-existent backend, but no exception was thrown")
+ } catch (Exception e) {
+ logger.info("Received expected exception when trying to drop non-existent backend: {}", e.getMessage())
+ assert e.getMessage().contains("backend does not exists"), "Unexpected exception message when trying to drop non-existent backend"
+ }
+
+ // Verify that the number of backends remains unchanged
+ def currentBackends = sql_return_maparray("SHOW BACKENDS")
+ def originalBackendCount = 3 // As per the initial setup in this test
+ assert currentBackends.size() == originalBackendCount, "Number of backends should remain unchanged after attempting to drop a non-existent backend"
+
+ checkClusterStatus(3, 3, 6)
+
+ // CASE 8. Decommission a backend and verify the process
+ logger.info("Attempting to decommission a backend")
+
+ // Get the list of current backends
+ backends = sql_return_maparray("SHOW BACKENDS")
+ assert backends.size() >= 1, "Not enough backends to perform decommission test"
+
+ // Select a backend to decommission (not the first one, as it might be the master)
+ def backendToDecommission = backends[1]
+ def decommissionHost = backendToDecommission['Host']
+ def decommissionPort = backendToDecommission['HeartbeatPort']
+
+ logger.info("Decommissioning backend: {}:{}", decommissionHost, decommissionPort)
+
+ // Decommission the selected backend
+ sql """ ALTER SYSTEM DECOMMISSION BACKEND "${decommissionHost}:${decommissionPort}"; """
+
+ // Wait for the decommission process to complete (this may take some time in a real environment)
+ int maxAttempts = 30
+ int attempts = 0
+ boolean decommissionComplete = false
+
+ while (attempts < maxAttempts && !decommissionComplete) {
+ currentBackends = sql_return_maparray("SHOW BACKENDS")
+ def decommissionedBackend = currentBackends.find { it['Host'] == decommissionHost && it['HeartbeatPort'] == decommissionPort }
+
+ logger.info("decomissionedBackend {}", decommissionedBackend)
+ // TODO: decomisssion is alive?
+ if (decommissionedBackend && decommissionedBackend['Alive'] == "true" && decommissionedBackend['SystemDecommissioned'] == "true") {
+ decommissionComplete = true
+ } else {
+ attempts++
+ sleep(1000) // Wait for 1 second before checking again
+ }
+ }
+
+ assert decommissionComplete, "Backend decommission did not complete within the expected time"
+
+ // Verify that the decommissioned backend is no longer active
+ def finalBackends = sql_return_maparray("SHOW BACKENDS")
+ def decommissionedBackend = finalBackends.find { it['Host'] == decommissionHost && it['HeartbeatPort'] == decommissionPort }
+ assert decommissionedBackend['Alive'] == "true", "Decommissioned backend should not be alive"
+ assert decommissionedBackend['SystemDecommissioned'] == "true", "Decommissioned backend should have SystemDecommissioned set to true"
+
+ logger.info("Successfully decommissioned backend and verified its status")
+
+ checkClusterStatus(3, 3, 7)
+ }
+ }
+
+}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org