hcq/oceanbase
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity
Files
generic
oceanbase/unittest/storage/mockcontainer/mock_ob_server.cpp
ly0 68cd558747 patch tair code removing from 22x to 31CE
2022-04-06 10:56:12 +08:00

472 lines
15 KiB
C++
Raw Permalink Blame History

/**
* Copyright (c) 2021 OceanBase
* OceanBase CE is licensed under Mulan PubL v2.
* You can use this software according to the terms and conditions of the Mulan PubL v2.
* You may obtain a copy of Mulan PubL v2 at:
* http://license.coscl.org.cn/MulanPubL-2.0
* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
* EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
* MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
* See the Mulan PubL v2 for more details.
*/
#define USING_LOG_PREFIX STORAGE
#include "observer/ob_server.h"
#define private public
#include "mock_ob_server.h"
#include "mock_ob_location_cache.h"
#include "lib/string/ob_sql_string.h"
#include "lib/io/ob_io_manager.h"
#include "lib/net/ob_net_util.h"
#include "rpc/obrpc/ob_rpc_proxy.h"
#include "share/schema/ob_multi_version_schema_service.h"
#include "share/partition_table/ob_partition_info.h"
#include "storage/blocksstable/ob_block_sstable_struct.h"
#include "storage/transaction/ob_ts_mgr.h"
#include "share/ob_tenant_mgr.h"
namespace oceanbase {
using namespace common;
using namespace lib;
using namespace blocksstable;
namespace unittest {
using namespace common;
using namespace election;
const char* ELECT_ASYNC_LOG_FILE_NAME = "election.log";
int MockObServer::init(
const char* schema_file, MockObLocationCache* location_cache, int64_t data_file_size, int64_t macro_block_size)
{
int ret = OB_SUCCESS;
int tmp_ret = 0;
ObStorageEnv env;
char* logdir = NULL;
char* clogdir = NULL;
char* ilogdir = NULL;
if (is_inited_) {
STORAGE_LOG(WARN, "ob server inited twice");
ret = OB_INIT_TWICE;
} else if (NULL == schema_file) {
STORAGE_LOG(ERROR, "invalid argument", "schema_file", OB_P(schema_file));
ret = OB_INVALID_ARGUMENT;
} else {
}
// init config
if (OB_SUCC(ret)) {
config_.datafile_size = data_file_size;
if (opts_.rpc_port_) {
config_.rpc_port = opts_.rpc_port_;
}
if (opts_.mysql_port_) {
config_.mysql_port = opts_.mysql_port_;
}
if (opts_.devname_) {
config_.devname.set_value(opts_.devname_);
}
if (opts_.rs_list_) {
config_.rootservice_list.set_value(opts_.rs_list_);
}
if (opts_.optstr_) {
config_.add_extra_config(opts_.optstr_);
}
if (opts_.devname_ && strlen(opts_.devname_) > 0) {
config_.devname.set_value(opts_.devname_);
} else {
const char* devname = get_default_if();
if (devname && devname[0] != '\0') {
LOG_INFO("guess interface name", K(devname));
config_.devname.set_value(devname);
} else {
LOG_INFO("can't guess interface name, use default bond0");
}
}
if (OB_SUCCESS != (ret = ASYNC_LOG_INIT(ELECT_ASYNC_LOG_FILE_NAME, OB_LOG_LEVEL_INFO, true))) {
LOG_ERROR("async log init error.");
ret = OB_ELECTION_ASYNC_LOG_WARN_INIT;
}
int32_t local_ip = ntohl(obsys::ObNetUtil::get_local_addr_ipv4(config_.devname));
int32_t local_port = static_cast<int32_t>(config_.rpc_port);
config_.print();
// initialize self address
self_addr_.set_ipv4_addr(local_ip, local_port);
}
// init env
if (OB_SUCC(ret)) {
if (NULL == (logdir = new char[MAX_PATH_SIZE])) {
STORAGE_LOG(ERROR, "new log dir error");
ret = OB_ERR_UNEXPECTED;
} else if (NULL == (clogdir = new char[MAX_PATH_SIZE])) {
STORAGE_LOG(ERROR, "new clog dir error");
ret = OB_ERR_UNEXPECTED;
} else if (NULL == (ilogdir = new char[MAX_PATH_SIZE])) {
STORAGE_LOG(ERROR, "new clog dir error");
ret = OB_ERR_UNEXPECTED;
} else if (0 > (tmp_ret = snprintf(logdir, MAX_PATH_SIZE, "%s/slog", opts_.data_dir_ /*, opts_.appname_*/))) {
STORAGE_LOG(ERROR, "concate log path fail", "ret", tmp_ret);
ret = OB_ERR_UNEXPECTED;
} else if (0 > (tmp_ret = snprintf(clogdir, MAX_PATH_SIZE, "%s/clog", opts_.data_dir_ /*, opts_.appname_*/))) {
STORAGE_LOG(ERROR, "concate log path fail", "ret", tmp_ret);
ret = OB_ERR_UNEXPECTED;
} else if (0 > (tmp_ret = snprintf(ilogdir, MAX_PATH_SIZE, "%s/ilog", opts_.data_dir_ /*, opts_.appname_*/))) {
STORAGE_LOG(ERROR, "concate log path fail", "ret", tmp_ret);
ret = OB_ERR_UNEXPECTED;
} else {
env.data_dir_ = opts_.data_dir_;
env.default_block_size_ = macro_block_size;
env.log_spec_.log_dir_ = logdir;
env.log_spec_.max_log_size_ = 256 << 20;
env.clog_dir_ = clogdir;
env.ilog_dir_ = ilogdir;
env.index_cache_priority_ = 10;
env.user_block_cache_priority_ = 1;
env.user_row_cache_priority_ = 1;
env.fuse_row_cache_priority_ = 1;
env.bf_cache_priority_ = 1;
env.clog_cache_priority_ = 1;
env.index_clog_cache_priority_ = 1;
}
}
// init schema service
if (OB_SUCC(ret)) {
ObSchemaGetterGuard* schema_guard = NULL;
if (OB_FAIL(restore_schema_.init())) {
STORAGE_LOG(ERROR, "restore_schema init fail", K(ret));
} else if (OB_FAIL(restore_schema_.parse_from_file(schema_file, schema_guard))) {
STORAGE_LOG(ERROR, "parse_from_file fail", K(ret));
} else {
schema_service_ = restore_schema_.schema_service_;
}
// if (OB_SUCCESS != (ret = schema_service_.init(schema_file))) {
// STORAGE_LOG(ERROR, "schema service init error", K(ret));
//} else {
// STORAGE_LOG(INFO, "schema service init success");
//}
}
// init global context
if (OB_SUCC(ret)) {
// gctx_.root_service_ = &root_service_;
// gctx_.ob_service_ = &ob_service_;
gctx_.schema_service_ = schema_service_;
gctx_.config_ = &config_;
gctx_.config_mgr_ = &config_mgr_;
gctx_.pt_operator_ = &pt_operator_;
// gctx_.srv_rpc_proxy_ = &srv_rpc_proxy_;
// gctx_.rs_rpc_proxy_ = &rs_rpc_proxy_;
GCTX.sql_proxy_ = &sql_proxy_;
// gctx_.timer_ = &timer_;
gctx_.self_addr_ = self_addr_;
// gctx_.rs_mgr_ = &rs_mgr_;
gctx_.par_ser_ = &partition_service_;
gctx_.omt_ = &multi_tenant_;
gctx_.session_mgr_ = &session_mgr_;
// gctx_.sql_engine_ = &sql_engine_;
gctx_.warm_up_start_time_ = &warm_up_start_time_;
gctx_.global_last_merged_version_ = &global_last_merged_version_;
}
// init net frame
if (OB_SUCC(ret)) {
if (OB_SUCCESS != (ret = net_frame_.init())) {
STORAGE_LOG(ERROR, "net frame init error", K(ret));
} else if (OB_FAIL(batch_rpc_.init(
net_frame_.get_batch_rpc_req_transport(), net_frame_.get_high_prio_req_transport(), self_addr_))) {
LOG_WARN("init batch rpc failed", K(ret));
} else {
STORAGE_LOG(INFO, "net frame init success");
}
}
// init location cache
if (OB_SUCC(ret)) {
if (OB_SUCCESS != (ret = location_cache_def_.init())) {
STORAGE_LOG(ERROR, "location cache init error", K(ret));
} else if (NULL != location_cache) {
location_cache_ = location_cache;
STORAGE_LOG(INFO, "use extern location cache");
} else {
STORAGE_LOG(INFO, "location cache init success");
}
}
if (OB_SUCC(ret)) {
if (OB_SUCCESS != (ret = bandwidth_throttle_.init(1024 * 1024 * 60))) {
STORAGE_LOG(ERROR, "failed to init bandwidth_throttle_", K(ret));
}
}
if (OB_SUCC(ret)) {
gctx_.rs_server_status_ = RSS_IS_WORKING;
}
// init global kv cache
if (OB_SUCC(ret)) {
ret = ObKVGlobalCache::get_instance().init();
}
if (OB_SUCC(ret)) {
if (OB_SUCCESS != (ret = init_tenant_mgr())) {
STORAGE_LOG(WARN, "init tenant mgr failed", K(ret));
} else {
STORAGE_LOG(INFO, "init tenant mgr success");
}
}
// init io
if (OB_SUCC(ret)) {
if (OB_FAIL(ObIOManager::get_instance().init())) {
STORAGE_LOG(WARN, "io manager init failead", K(ret));
}
}
// init partition service
if (OB_SUCC(ret)) {
// make log directory for election
system("rm -rf log");
system("mkdir log");
if (OB_SUCCESS != (ret = partition_service_.init(env,
self_addr_,
&partition_cfy_,
schema_service_,
location_cache_,
&rs_mgr_,
&rs_cb_,
net_frame_.get_req_transport(),
&batch_rpc_,
rs_rpc_proxy_,
srv_rpc_proxy_,
sql_proxy_,
remote_sql_proxy_,
server_tracer_,
bandwidth_throttle_))) {
STORAGE_LOG(WARN, "failed to init partitin service", K(ret));
} else {
STORAGE_LOG(INFO, "partition service init success");
}
}
// init gts response rpc
if (OB_SUCC(ret)) {
if (OB_FAIL(gts_response_rpc_.init(net_frame_.get_req_transport(), self_addr_))) {
LOG_ERROR("gts response rpc init failed", K(ret));
} else {
LOG_INFO("gts response rpc init success");
}
}
// init gtx
if (OB_SUCC(ret)) {
if (OB_FAIL(gts_.init(partition_service_.get_election_mgr(), &gts_response_rpc_, self_addr_))) {
LOG_ERROR("gts init failed", K(ret));
} else {
LOG_INFO("gts init success");
}
}
// init gts local cache mgr
if (OB_SUCC(ret)) {
ObILocationAdapter* location_adapter =
static_cast<ObTransService*>(partition_service_.get_trans_service())->get_location_adapter();
if (OB_SUCCESS != (ret = OB_TS_MGR.init(self_addr_, location_adapter, net_frame_.get_req_transport(), &gts_))) {
STORAGE_LOG(WARN, "init gts local cache mgr failed", K(ret));
} else {
STORAGE_LOG(INFO, "init gts local cache mgr success");
}
}
// init multi tenant
GCTX = gctx_;
if (OB_SUCC(ret)) {
if (OB_SUCCESS != (ret = init_multi_tenant())) {
STORAGE_LOG(WARN, "init multi tenant failed", K(ret));
} else {
STORAGE_LOG(INFO, "init multi tenant success");
}
}
if (OB_SUCC(ret)) {
is_inited_ = true;
STORAGE_LOG(INFO, "ob server inited success");
}
return ret;
}
int MockObServer::init_multi_tenant()
{
int ret = OB_SUCCESS;
if (OB_SUCCESS != (ret = multi_tenant_.init(self_addr_, 6))) {
STORAGE_LOG(WARN, "init multi_tenant failed", K(ret));
} else if (OB_SUCCESS != (ret = multi_tenant_.add_tenant(OB_SYS_TENANT_ID, 3, 3))) {
STORAGE_LOG(WARN, "add sys tenant failed", K(ret));
} else if (OB_SUCCESS != (ret = multi_tenant_.add_tenant(OB_ELECT_TENANT_ID, 5, 5))) {
STORAGE_LOG(WARN, "add election tenant failed", K(ret));
} else if (OB_SUCCESS != (ret = multi_tenant_.add_tenant(OB_SERVER_TENANT_ID, 5, 5))) {
STORAGE_LOG(WARN, "add election tenant failed", K(ret));
} else {
multi_tenant_.start();
}
return ret;
}
int MockObServer::init_tenant_mgr()
{
int ret = OB_SUCCESS;
static const int64_t SYS_MEM_MIN = 8LL << 30;
static const int64_t SYS_MEM_MAX = 16LL << 30;
static const int64_t SERVER_TENANT_MEM_MIN = 8LL << 30;
static const int64_t SERVER_TENANT_MEM_MAX = 16LL << 30;
static const int64_t SERVER_MEM_MAX = 16LL << 30;
ObTenantManager& omti = ObTenantManager::get_instance();
obrpc::ObCommonRpcProxy common_rpc;
share::ObRsMgr rs_mgr;
if (OB_SUCC(ret)) {
if (OB_FAIL(ObTenantManager::get_instance().init(
self_addr_, srv_rpc_proxy_, rs_rpc_proxy_, rs_mgr_, net_frame_.get_req_transport(), &config_))) {
LOG_ERROR("Fail to init ObTenantManager, ", K(ret));
} else if (OB_FAIL(omti.add_tenant(common::OB_SYS_TENANT_ID))) {
LOG_ERROR("Fail to add sys tenant to tenant manager, ", K(ret));
} else if (OB_SUCCESS != (ret = omti.set_tenant_mem_limit(OB_SYS_TENANT_ID, SYS_MEM_MIN, SYS_MEM_MAX))) {
LOG_ERROR("Fail to set tenant mem limit", K(ret));
} else if (OB_FAIL(omti.add_tenant(common::OB_SERVER_TENANT_ID))) {
LOG_ERROR("Fail to add sys tenant to tenant manager, ", K(ret));
} else if (OB_SUCCESS !=
(ret = omti.set_tenant_mem_limit(OB_SERVER_TENANT_ID, SERVER_TENANT_MEM_MIN, SERVER_TENANT_MEM_MAX))) {
LOG_ERROR("Fail to set tenant mem limit", K(ret));
} else {
lib::set_memory_limit(SERVER_MEM_MAX);
}
}
return ret;
}
void MockObServer::destroy()
{
partition_service_.destroy();
ObKVGlobalCache::get_instance().destroy();
STORAGE_LOG(INFO, "MockObServer::destroy(). destroy gloabal_cache\n");
net_frame_.destroy();
STORAGE_LOG(INFO, "MockObServer::destroy(). destroy net_frame\n");
ObTenantManager& omti = ObTenantManager::get_instance();
omti.destroy();
multi_tenant_.destroy();
ObIOManager::get_instance().destroy();
ASYNC_LOG_DESTROY();
}
int MockObServer::start()
{
int ret = OB_SUCCESS;
if (!is_inited_) {
STORAGE_LOG(WARN, "ob server not inited");
ret = OB_NOT_INIT;
} else if (OB_SUCCESS != (ret = net_frame_.start())) {
STORAGE_LOG(WARN, "net frame start error", K(ret));
} else if (OB_SUCCESS != partition_service_.start()) {
STORAGE_LOG(WARN, "start partition service fail", K(ret));
} else if (OB_FAIL(OB_TS_MGR.start())) {
STORAGE_LOG(WARN, "start gts cache mgr error", K(ret));
} else {
STORAGE_LOG(INFO, "net frame start success");
}
// wait clog replay over
if (OB_SUCC(ret)) {
ret = partition_service_.wait_start_finish();
}
if (OB_SUCC(ret)) {
STORAGE_LOG(INFO, "ob server start success");
}
return ret;
}
int MockObServer::stop()
{
int ret = OB_SUCCESS;
if (!is_inited_) {
STORAGE_LOG(WARN, "ob server not inited");
ret = OB_NOT_INIT;
} else if (OB_SUCCESS != (partition_service_.stop())) {
STORAGE_LOG(WARN, "partition service stop error", K(ret));
} else if (OB_SUCCESS != (ret = net_frame_.stop())) {
STORAGE_LOG(WARN, "net frame stop error", K(ret));
} else {
OB_TS_MGR.stop();
STORAGE_LOG(INFO, "net frame stop success");
}
multi_tenant_.stop();
if (OB_SUCC(ret)) {
STORAGE_LOG(INFO, "ob server stop success");
}
return ret;
}
int MockObServer::wait()
{
int ret = OB_SUCCESS;
partition_service_.wait();
multi_tenant_.wait();
net_frame_.wait();
(void)OB_TS_MGR.wait();
if (OB_SUCC(ret)) {
STORAGE_LOG(INFO, "ob server wait success");
}
return ret;
}
ObPartitionService* MockObServer::get_partition_service()
{
ObPartitionService* ps = NULL;
if (!is_inited_) {
STORAGE_LOG(WARN, "ob server not inited");
} else {
ps = &partition_service_;
}
return ps;
}
MockObLocationCache* MockObServer::get_location_cache()
{
MockObLocationCache* lc = NULL;
if (!is_inited_) {
STORAGE_LOG(WARN, "ob server not inited");
} else {
lc = location_cache_;
}
return lc;
}
MockSchemaService* MockObServer::get_schema_service()
{
MockSchemaService* ss = NULL;
if (!is_inited_) {
STORAGE_LOG(WARN, "ob server not inited");
} else {
ss = schema_service_;
}
return ss;
}
} // namespace unittest
} // namespace oceanbase
Reference in New Issue View Git Blame Copy Permalink