You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pegasus.apache.org by wa...@apache.org on 2022/10/25 07:44:01 UTC
[incubator-pegasus] branch master updated: refactor(log): adjust log level and rename for d*_f macros (#1201)
This is an automated email from the ASF dual-hosted git repository.
wangdan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pegasus.git
The following commit(s) were added to refs/heads/master by this push:
new 88fa55ce5 refactor(log): adjust log level and rename for d*_f macros (#1201)
88fa55ce5 is described below
commit 88fa55ce58681702ab6762fa0cc3955964d9106e
Author: Yingchun Lai <la...@apache.org>
AuthorDate: Tue Oct 25 15:43:55 2022 +0800
refactor(log): adjust log level and rename for d*_f macros (#1201)
---
src/aio/native_linux_aio_provider.cpp | 15 +-
src/base/pegasus_value_schema.h | 4 +-
src/block_service/block_service_manager.cpp | 30 +-
src/block_service/directio_writable_file.cpp | 9 +-
src/block_service/fds/fds_service.cpp | 43 +-
src/block_service/hdfs/hdfs_service.cpp | 87 +--
src/block_service/local/local_service.cpp | 2 +-
src/client/partition_resolver_simple.cpp | 34 +-
src/common/fs_manager.cpp | 49 +-
src/common/replication_common.cpp | 12 +-
src/geo/lib/geo_client.cpp | 127 ++--
src/geo/lib/latlng_codec.cpp | 2 +-
src/http/pprof_http_service.cpp | 6 +-
src/meta/app_balance_policy.cpp | 26 +-
src/meta/app_env_validator.cpp | 2 +-
src/meta/backup_engine.cpp | 80 +--
src/meta/cluster_balance_policy.cpp | 48 +-
src/meta/duplication/duplication_info.cpp | 6 +-
src/meta/duplication/duplication_info.h | 18 +-
src/meta/duplication/meta_duplication_service.cpp | 61 +-
src/meta/load_balance_policy.cpp | 41 +-
src/meta/meta_backup_service.cpp | 129 ++--
src/meta/meta_bulk_load_ingestion_context.cpp | 18 +-
src/meta/meta_bulk_load_service.cpp | 680 +++++++++++----------
src/meta/meta_service.cpp | 40 +-
src/meta/meta_split_service.cpp | 176 +++---
src/meta/meta_state_service_utils_impl.h | 6 +-
src/meta/server_state.cpp | 419 ++++++-------
src/meta/server_state_restore.cpp | 6 +-
src/meta/test/meta_bulk_load_service_test.cpp | 32 +-
src/meta/test/meta_split_service_test.cpp | 6 +-
src/meta/test/meta_test_base.cpp | 8 +-
src/redis_protocol/proxy_lib/redis_parser.cpp | 80 +--
src/replica/backup/replica_backup_manager.cpp | 2 +-
src/replica/backup/replica_backup_server.cpp | 6 +-
src/replica/disk_cleaner.cpp | 24 +-
src/replica/duplication/duplication_sync_timer.cpp | 13 +-
src/replica/duplication/load_from_private_log.cpp | 2 +-
.../test/load_from_private_log_test.cpp | 2 +-
src/replica/mutation_log.cpp | 56 +-
src/replica/mutation_log_replay.cpp | 2 +-
src/replica/replica_2pc.cpp | 8 +-
src/replica/replica_context.cpp | 6 +-
src/replica/replica_init.cpp | 23 +-
src/replica/replica_restore.cpp | 42 +-
src/replica/replica_stub.cpp | 82 +--
src/replica/replication_app_base.cpp | 10 +-
src/replica/test/mutation_log_test.cpp | 2 +-
src/replica/test/replica_disk_test.cpp | 2 +-
src/reporter/pegasus_counter_reporter.cpp | 2 +-
src/runtime/security/client_negotiation.cpp | 39 +-
src/runtime/security/init.cpp | 8 +-
src/runtime/security/kinit_context.cpp | 20 +-
src/runtime/security/negotiation.cpp | 8 +-
src/runtime/security/negotiation_manager.cpp | 6 +-
src/runtime/security/replica_access_controller.cpp | 2 +-
src/runtime/security/server_negotiation.cpp | 30 +-
src/runtime/task/task_engine.cpp | 8 +-
src/server/compaction_filter_rule.cpp | 2 +-
src/server/hotspot_partition_calculator.cpp | 38 +-
src/server/info_collector.cpp | 8 +-
src/server/logging_utils.h | 14 +-
src/server/pegasus_server_write.cpp | 4 +-
src/test/function_test/base_api_test/test_copy.cpp | 6 +-
src/test/function_test/base_api_test/test_scan.cpp | 4 +-
src/utils/filesystem.cpp | 46 +-
src/utils/fmt_logging.h | 25 +-
src/utils/latency_tracer.cpp | 4 +-
68 files changed, 1452 insertions(+), 1406 deletions(-)
diff --git a/src/aio/native_linux_aio_provider.cpp b/src/aio/native_linux_aio_provider.cpp
index 8a9d206d4..3ff8dda22 100644
--- a/src/aio/native_linux_aio_provider.cpp
+++ b/src/aio/native_linux_aio_provider.cpp
@@ -84,11 +84,11 @@ error_code native_linux_aio_provider::write(const aio_context &aio_ctx,
aio_ctx.file_offset + buffer_offset);
if (dsn_unlikely(ret < 0)) {
if (errno == EINTR) {
- dwarn_f("write failed with errno={} and will retry it.", strerror(errno));
+ LOG_WARNING_F("write failed with errno={} and will retry it.", strerror(errno));
continue;
}
resp = ERR_FILE_OPERATION_FAILED;
- derror_f("write failed with errno={}, return {}.", strerror(errno), resp);
+ LOG_ERROR_F("write failed with errno={}, return {}.", strerror(errno), resp);
return resp;
}
@@ -101,11 +101,12 @@ error_code native_linux_aio_provider::write(const aio_context &aio_ctx,
buffer_offset += ret;
if (dsn_unlikely(buffer_offset != aio_ctx.buffer_size)) {
- dwarn_f("write incomplete, request_size={}, total_write_size={}, this_write_size={}, "
- "and will retry it.",
- aio_ctx.buffer_size,
- buffer_offset,
- ret);
+ LOG_WARNING_F(
+ "write incomplete, request_size={}, total_write_size={}, this_write_size={}, "
+ "and will retry it.",
+ aio_ctx.buffer_size,
+ buffer_offset,
+ ret);
}
} while (dsn_unlikely(buffer_offset < aio_ctx.buffer_size));
diff --git a/src/base/pegasus_value_schema.h b/src/base/pegasus_value_schema.h
index 7fa9997ec..b1283563e 100644
--- a/src/base/pegasus_value_schema.h
+++ b/src/base/pegasus_value_schema.h
@@ -115,7 +115,7 @@ inline void pegasus_update_expire_ts(uint32_t version, std::string &value, uint3
new_expire_ts = dsn::endian::hton(new_expire_ts);
memcpy(const_cast<char *>(value.data()), &new_expire_ts, sizeof(uint32_t));
} else {
- dfatal_f("unsupported value schema version: {}", version);
+ LOG_FATAL_F("unsupported value schema version: {}", version);
__builtin_unreachable();
}
}
@@ -156,7 +156,7 @@ public:
} else if (value_schema_version == 1) {
return generate_value_v1(expire_ts, timetag, user_data);
} else {
- dfatal_f("unsupported value schema version: {}", value_schema_version);
+ LOG_FATAL_F("unsupported value schema version: {}", value_schema_version);
__builtin_unreachable();
}
}
diff --git a/src/block_service/block_service_manager.cpp b/src/block_service/block_service_manager.cpp
index 1e84dd5ce..1ea204bdc 100644
--- a/src/block_service/block_service_manager.cpp
+++ b/src/block_service/block_service_manager.cpp
@@ -72,9 +72,9 @@ block_filesystem *block_service_manager::get_or_create_block_filesystem(const st
block_filesystem *fs =
utils::factory_store<block_filesystem>::create(provider_type, PROVIDER_TYPE_MAIN);
if (fs == nullptr) {
- derror_f("acquire block filesystem failed, provider = {}, provider_type = {}",
- provider,
- std::string(provider_type));
+ LOG_ERROR_F("acquire block filesystem failed, provider = {}, provider_type = {}",
+ provider,
+ std::string(provider_type));
return nullptr;
}
@@ -86,12 +86,12 @@ block_filesystem *block_service_manager::get_or_create_block_filesystem(const st
dsn::error_code err = fs->initialize(args);
if (dsn::ERR_OK == err) {
- ddebug_f("create block filesystem ok for provider {}", provider);
+ LOG_INFO_F("create block filesystem ok for provider {}", provider);
_fs_map.emplace(provider, std::unique_ptr<block_filesystem>(fs));
} else {
- derror_f("create block file system err {} for provider {}",
- std::string(err.to_string()),
- provider);
+ LOG_ERROR_F("create block file system err {} for provider {}",
+ std::string(err.to_string()),
+ provider);
delete fs;
fs = nullptr;
}
@@ -145,7 +145,7 @@ error_code block_service_manager::download_file(const std::string &remote_dir,
// local file exists
const std::string local_file_name = utils::filesystem::path_combine(local_dir, file_name);
if (utils::filesystem::file_exists(local_file_name)) {
- ddebug_f("local file({}) exists", local_file_name);
+ LOG_INFO_F("local file({}) exists", local_file_name);
return ERR_PATH_ALREADY_EXIST;
}
@@ -157,7 +157,7 @@ error_code block_service_manager::download_file(const std::string &remote_dir,
create_block_file_sync(remote_file_name, false /*ignore file meta*/, fs, &tracker);
error_code err = create_resp.err;
if (err != ERR_OK) {
- derror_f("create file({}) failed with error({})", remote_file_name, err.to_string());
+ LOG_ERROR_F("create file({}) failed with error({})", remote_file_name, err.to_string());
return err;
}
block_file_ptr bf = create_resp.file_handle;
@@ -168,17 +168,17 @@ error_code block_service_manager::download_file(const std::string &remote_dir,
// error, however, if file damaged on remote file provider, bulk load should stop,
// return ERR_CORRUPTION instead
if (resp.err == ERR_OBJECT_NOT_FOUND) {
- derror_f("download file({}) failed, file on remote file provider is damaged",
- local_file_name);
+ LOG_ERROR_F("download file({}) failed, file on remote file provider is damaged",
+ local_file_name);
return ERR_CORRUPTION;
}
return resp.err;
}
- ddebug_f("download file({}) succeed, file_size = {}, md5 = {}",
- local_file_name.c_str(),
- resp.downloaded_size,
- resp.file_md5);
+ LOG_INFO_F("download file({}) succeed, file_size = {}, md5 = {}",
+ local_file_name.c_str(),
+ resp.downloaded_size,
+ resp.file_md5);
download_file_size = resp.downloaded_size;
download_file_md5 = resp.file_md5;
return ERR_OK;
diff --git a/src/block_service/directio_writable_file.cpp b/src/block_service/directio_writable_file.cpp
index c2e9a165b..ca7b179f4 100644
--- a/src/block_service/directio_writable_file.cpp
+++ b/src/block_service/directio_writable_file.cpp
@@ -72,7 +72,7 @@ direct_io_writable_file::~direct_io_writable_file()
bool direct_io_writable_file::initialize()
{
if (posix_memalign(&_buffer, g_page_size, _buffer_size) != 0) {
- derror_f("Allocate memaligned buffer failed, errno = {}", errno);
+ LOG_ERROR_F("Allocate memaligned buffer failed, errno = {}", errno);
return false;
}
@@ -82,7 +82,7 @@ bool direct_io_writable_file::initialize()
#endif
_fd = open(_file_path.c_str(), flag, S_IRUSR | S_IWUSR | S_IRGRP);
if (_fd < 0) {
- derror_f("Failed to open {} with flag {}, errno = {}", _file_path, flag, errno);
+ LOG_ERROR_F("Failed to open {} with flag {}, errno = {}", _file_path, flag, errno);
free(_buffer);
_buffer = nullptr;
return false;
@@ -96,7 +96,8 @@ bool direct_io_writable_file::finalize()
if (_offset > 0) {
if (::write(_fd, _buffer, _buffer_size) != _buffer_size) {
- derror_f("Failed to write last chunk, filie_path = {}, errno = {}", _file_path, errno);
+ LOG_ERROR_F(
+ "Failed to write last chunk, filie_path = {}, errno = {}", _file_path, errno);
return false;
}
_offset = 0;
@@ -119,7 +120,7 @@ bool direct_io_writable_file::write(const char *s, size_t n)
// buffer is full, flush to file
if (_offset == _buffer_size) {
if (::write(_fd, _buffer, _buffer_size) != _buffer_size) {
- derror_f("Failed to write to direct_io_writable_file, errno = {}", errno);
+ LOG_ERROR_F("Failed to write to direct_io_writable_file, errno = {}", errno);
return false;
}
// reset offset
diff --git a/src/block_service/fds/fds_service.cpp b/src/block_service/fds/fds_service.cpp
index c458b4e31..67064c331 100644
--- a/src/block_service/fds/fds_service.cpp
+++ b/src/block_service/fds/fds_service.cpp
@@ -427,10 +427,10 @@ error_code fds_file_object::get_file_meta()
if (ex.code() == Poco::Net::HTTPResponse::HTTP_NOT_FOUND) {
err = ERR_OBJECT_NOT_FOUND;
} else {
- derror_f("fds getObjectMetadata failed: parameter({}), code({}), msg({})",
- _name.c_str(),
- ex.code(),
- ex.what());
+ LOG_ERROR_F("fds getObjectMetadata failed: parameter({}), code({}), msg({})",
+ _name.c_str(),
+ ex.code(),
+ ex.what());
err = ERR_FS_INTERNAL;
}
}
@@ -538,10 +538,10 @@ error_code fds_file_object::put_content(/*in-out*/ std::istream &is,
FLAGS_fds_write_limit_rate << 20,
FLAGS_fds_write_burst_size
<< 20)) {
- ddebug_f("the transfer count({}B) is greater than burst size({}MB), so it is rejected by "
- "token bucket",
- to_transfer_bytes,
- FLAGS_fds_write_burst_size);
+ LOG_INFO_F("the transfer count({}B) is greater than burst size({}MB), so it is rejected by "
+ "token bucket",
+ to_transfer_bytes,
+ FLAGS_fds_write_burst_size);
return ERR_BUSY;
}
@@ -614,10 +614,11 @@ dsn::task_ptr fds_file_object::upload(const upload_request &req,
std::ifstream is(local_file, std::ios::binary | std::ios::in);
if (!is.is_open()) {
- derror_f("fds upload failed: open local file({}) failed when upload to({}), error({})",
- local_file,
- file_name(),
- ::dsn::utils::safe_strerror(errno));
+ LOG_ERROR_F(
+ "fds upload failed: open local file({}) failed when upload to({}), error({})",
+ local_file,
+ file_name(),
+ ::dsn::utils::safe_strerror(errno));
resp.err = dsn::ERR_FILE_OPERATION_FAILED;
} else {
resp.err = put_content(is, file_sz, resp.uploaded_size);
@@ -673,10 +674,10 @@ dsn::task_ptr fds_file_object::download(const download_request &req,
std::shared_ptr<std::ofstream> handle(new std::ofstream(
req.output_local_name, std::ios::binary | std::ios::out | std::ios::trunc));
if (!handle->is_open()) {
- derror_f("fds download failed: fail to open localfile({}) when download({}), error({})",
- req.output_local_name,
- _fds_path,
- ::dsn::utils::safe_strerror(errno));
+ LOG_ERROR_F("fds download failed: fail to open localfile({}) when download({}), error({})",
+ req.output_local_name,
+ _fds_path,
+ ::dsn::utils::safe_strerror(errno));
resp.err = ERR_FILE_OPERATION_FAILED;
resp.downloaded_size = 0;
t->enqueue_with(resp);
@@ -695,14 +696,14 @@ dsn::task_ptr fds_file_object::download(const download_request &req,
}
handle->close();
if (resp.err != ERR_OK && dsn::utils::filesystem::file_exists(req.output_local_name)) {
- derror_f("fail to download file {} from fds, remove localfile {}",
- _fds_path,
- req.output_local_name);
+ LOG_ERROR_F("fail to download file {} from fds, remove localfile {}",
+ _fds_path,
+ req.output_local_name);
dsn::utils::filesystem::remove_path(req.output_local_name);
} else if ((resp.err = dsn::utils::filesystem::md5sum(req.output_local_name,
resp.file_md5)) != ERR_OK) {
- derror_f("download failed when calculate the md5sum of local file {}",
- req.output_local_name);
+ LOG_ERROR_F("download failed when calculate the md5sum of local file {}",
+ req.output_local_name);
}
t->enqueue_with(resp);
release_ref();
diff --git a/src/block_service/hdfs/hdfs_service.cpp b/src/block_service/hdfs/hdfs_service.cpp
index 6b8dfb2d3..c4951f923 100644
--- a/src/block_service/hdfs/hdfs_service.cpp
+++ b/src/block_service/hdfs/hdfs_service.cpp
@@ -84,7 +84,7 @@ error_code hdfs_service::initialize(const std::vector<std::string> &args)
// If no path was configured, just use "/" as default root path.
_hdfs_name_node = args[0];
_hdfs_path = args.size() >= 2 ? args[1] : "/";
- ddebug_f("hdfs backup root path is initialized to {}.", _hdfs_path);
+ LOG_INFO_F("hdfs backup root path is initialized to {}.", _hdfs_path);
return create_fs();
}
@@ -93,18 +93,18 @@ error_code hdfs_service::create_fs()
{
hdfsBuilder *builder = hdfsNewBuilder();
if (!builder) {
- derror_f("Fail to create an HDFS builder, error: {}.", utils::safe_strerror(errno));
+ LOG_ERROR_F("Fail to create an HDFS builder, error: {}.", utils::safe_strerror(errno));
return ERR_FS_INTERNAL;
}
hdfsBuilderSetNameNode(builder, _hdfs_name_node.c_str());
_fs = hdfsBuilderConnect(builder);
if (!_fs) {
- derror_f("Fail to connect hdfs name node {}, error: {}.",
- _hdfs_name_node,
- utils::safe_strerror(errno));
+ LOG_ERROR_F("Fail to connect hdfs name node {}, error: {}.",
+ _hdfs_name_node,
+ utils::safe_strerror(errno));
return ERR_FS_INTERNAL;
}
- ddebug_f("Succeed to connect hdfs name node {}.", _hdfs_name_node);
+ LOG_INFO_F("Succeed to connect hdfs name node {}.", _hdfs_name_node);
return ERR_OK;
}
@@ -116,8 +116,8 @@ void hdfs_service::close()
LOG_INFO("Try to disconnect hdfs.");
int result = hdfsDisconnect(_fs);
if (result == -1) {
- derror_f("Fail to disconnect from the hdfs file system, error: {}.",
- utils::safe_strerror(errno));
+ LOG_ERROR_F("Fail to disconnect from the hdfs file system, error: {}.",
+ utils::safe_strerror(errno));
}
// Even if there is an error, the resources associated with the hdfsFS will be freed.
_fs = nullptr;
@@ -143,7 +143,7 @@ dsn::task_ptr hdfs_service::list_dir(const ls_request &req,
ls_response resp;
if (hdfsExists(_fs, path.c_str()) == -1) {
- derror_f("HDFS list directory failed: path {} not found.", path);
+ LOG_ERROR_F("HDFS list directory failed: path {} not found.", path);
resp.err = ERR_OBJECT_NOT_FOUND;
tsk->enqueue_with(resp);
return;
@@ -151,20 +151,20 @@ dsn::task_ptr hdfs_service::list_dir(const ls_request &req,
hdfsFileInfo *dir_info = hdfsGetPathInfo(_fs, path.c_str());
if (dir_info == nullptr) {
- derror_f("HDFS get path {} failed.", path);
+ LOG_ERROR_F("HDFS get path {} failed.", path);
resp.err = ERR_FS_INTERNAL;
tsk->enqueue_with(resp);
return;
}
if (dir_info->mKind == kObjectKindFile) {
- derror_f("HDFS list directory failed, {} is not a directory", path);
+ LOG_ERROR_F("HDFS list directory failed, {} is not a directory", path);
resp.err = ERR_INVALID_PARAMETERS;
} else {
int entries = 0;
hdfsFileInfo *info = hdfsListDirectory(_fs, path.c_str(), &entries);
if (info == nullptr) {
- derror_f("HDFS list directory {} failed.", path);
+ LOG_ERROR_F("HDFS list directory {} failed.", path);
resp.err = ERR_FS_INTERNAL;
} else {
for (int i = 0; i < entries; i++) {
@@ -211,7 +211,7 @@ dsn::task_ptr hdfs_service::create_file(const create_file_request &req,
// immediately after this call.
resp.err = ERR_OK;
resp.file_handle = f;
- ddebug_f("create remote file {} succeed", hdfs_file);
+ LOG_INFO_F("create remote file {} succeed", hdfs_file);
}
tsk->enqueue_with(resp);
};
@@ -234,7 +234,7 @@ dsn::task_ptr hdfs_service::remove_path(const remove_path_request &req,
// Check if path exists.
if (hdfsExists(_fs, path.c_str()) == -1) {
- derror_f("HDFS remove_path failed: path {} not found.", path);
+ LOG_ERROR_F("HDFS remove_path failed: path {} not found.", path);
resp.err = ERR_OBJECT_NOT_FOUND;
tsk->enqueue_with(resp);
return;
@@ -244,7 +244,7 @@ dsn::task_ptr hdfs_service::remove_path(const remove_path_request &req,
hdfsFileInfo *info = hdfsListDirectory(_fs, path.c_str(), &entries);
hdfsFreeFileInfo(info, entries);
if (entries > 0 && !req.recursive) {
- derror_f("HDFS remove_path failed: directory {} is not empty.", path);
+ LOG_ERROR_F("HDFS remove_path failed: directory {} is not empty.", path);
resp.err = ERR_DIR_NOT_EMPTY;
tsk->enqueue_with(resp);
return;
@@ -252,7 +252,7 @@ dsn::task_ptr hdfs_service::remove_path(const remove_path_request &req,
// Remove directory now.
if (hdfsDelete(_fs, path.c_str(), req.recursive) == -1) {
- derror_f("HDFS remove_path {} failed.", path);
+ LOG_ERROR_F("HDFS remove_path {} failed.", path);
resp.err = ERR_FS_INTERNAL;
} else {
resp.err = ERR_OK;
@@ -272,12 +272,12 @@ hdfs_file_object::hdfs_file_object(hdfs_service *s, const std::string &name)
error_code hdfs_file_object::get_file_meta()
{
if (hdfsExists(_service->get_fs(), file_name().c_str()) == -1) {
- dwarn_f("HDFS file {} does not exist.", file_name());
+ LOG_WARNING_F("HDFS file {} does not exist.", file_name());
return ERR_OBJECT_NOT_FOUND;
}
hdfsFileInfo *info = hdfsGetPathInfo(_service->get_fs(), file_name().c_str());
if (info == nullptr) {
- derror_f("HDFS get file info failed, file: {}.", file_name());
+ LOG_ERROR_F("HDFS get file info failed, file: {}.", file_name());
return ERR_FS_INTERNAL;
}
_size = info->mSize;
@@ -296,9 +296,9 @@ error_code hdfs_file_object::write_data_in_batches(const char *data,
hdfsFile write_file =
hdfsOpenFile(_service->get_fs(), file_name().c_str(), O_WRONLY | O_CREAT, 0, 0, 0);
if (!write_file) {
- derror_f("Failed to open hdfs file {} for writting, error: {}.",
- file_name(),
- utils::safe_strerror(errno));
+ LOG_ERROR_F("Failed to open hdfs file {} for writting, error: {}.",
+ file_name(),
+ utils::safe_strerror(errno));
return ERR_FS_INTERNAL;
}
uint64_t cur_pos = 0;
@@ -314,23 +314,23 @@ error_code hdfs_file_object::write_data_in_batches(const char *data,
(void *)(data + cur_pos),
static_cast<tSize>(write_len));
if (num_written_bytes == -1) {
- derror_f("Failed to write hdfs file {}, error: {}.",
- file_name(),
- utils::safe_strerror(errno));
+ LOG_ERROR_F("Failed to write hdfs file {}, error: {}.",
+ file_name(),
+ utils::safe_strerror(errno));
hdfsCloseFile(_service->get_fs(), write_file);
return ERR_FS_INTERNAL;
}
cur_pos += num_written_bytes;
}
if (hdfsHFlush(_service->get_fs(), write_file) != 0) {
- derror_f(
+ LOG_ERROR_F(
"Failed to flush hdfs file {}, error: {}.", file_name(), utils::safe_strerror(errno));
hdfsCloseFile(_service->get_fs(), write_file);
return ERR_FS_INTERNAL;
}
written_size = cur_pos;
if (hdfsCloseFile(_service->get_fs(), write_file) != 0) {
- derror_f(
+ LOG_ERROR_F(
"Failed to close hdfs file {}, error: {}", file_name(), utils::safe_strerror(errno));
return ERR_FS_INTERNAL;
}
@@ -378,10 +378,11 @@ dsn::task_ptr hdfs_file_object::upload(const upload_request &req,
is.close();
resp.err = write_data_in_batches(buffer.get(), file_sz, resp.uploaded_size);
} else {
- derror_f("HDFS upload failed: open local file {} failed when upload to {}, error: {}",
- req.input_local_name,
- file_name(),
- utils::safe_strerror(errno));
+ LOG_ERROR_F(
+ "HDFS upload failed: open local file {} failed when upload to {}, error: {}",
+ req.input_local_name,
+ file_name(),
+ utils::safe_strerror(errno));
resp.err = dsn::ERR_FILE_OPERATION_FAILED;
}
t->enqueue_with(resp);
@@ -401,16 +402,16 @@ error_code hdfs_file_object::read_data_in_batches(uint64_t start_pos,
if (!_has_meta_synced) {
error_code err = get_file_meta();
if (err != ERR_OK) {
- derror_f("Failed to read remote file {}", file_name());
+ LOG_ERROR_F("Failed to read remote file {}", file_name());
return err;
}
}
hdfsFile read_file = hdfsOpenFile(_service->get_fs(), file_name().c_str(), O_RDONLY, 0, 0, 0);
if (!read_file) {
- derror_f("Failed to open hdfs file {} for reading, error: {}.",
- file_name(),
- utils::safe_strerror(errno));
+ LOG_ERROR_F("Failed to open hdfs file {} for reading, error: {}.",
+ file_name(),
+ utils::safe_strerror(errno));
return ERR_FS_INTERNAL;
}
std::unique_ptr<char[]> raw_buf(new char[_size]);
@@ -437,15 +438,15 @@ error_code hdfs_file_object::read_data_in_batches(uint64_t start_pos,
cur_pos += num_read_bytes;
dst_buf += num_read_bytes;
} else if (num_read_bytes == -1) {
- derror_f("Failed to read hdfs file {}, error: {}.",
- file_name(),
- utils::safe_strerror(errno));
+ LOG_ERROR_F("Failed to read hdfs file {}, error: {}.",
+ file_name(),
+ utils::safe_strerror(errno));
read_success = false;
break;
}
}
if (hdfsCloseFile(_service->get_fs(), read_file) != 0) {
- derror_f(
+ LOG_ERROR_F(
"Failed to close hdfs file {}, error: {}.", file_name(), utils::safe_strerror(errno));
return ERR_FS_INTERNAL;
}
@@ -529,11 +530,11 @@ dsn::task_ptr hdfs_file_object::download(const download_request &req,
}
}
if (!write_succ) {
- derror_f("HDFS download failed: fail to open localfile {} when download {}, "
- "error: {}",
- req.output_local_name,
- file_name(),
- utils::safe_strerror(errno));
+ LOG_ERROR_F("HDFS download failed: fail to open localfile {} when download {}, "
+ "error: {}",
+ req.output_local_name,
+ file_name(),
+ utils::safe_strerror(errno));
resp.err = ERR_FILE_OPERATION_FAILED;
resp.downloaded_size = 0;
}
diff --git a/src/block_service/local/local_service.cpp b/src/block_service/local/local_service.cpp
index fc4502f02..e226edd2c 100644
--- a/src/block_service/local/local_service.cpp
+++ b/src/block_service/local/local_service.cpp
@@ -53,7 +53,7 @@ bool file_metadata_from_json(std::ifstream &fin, file_metadata &fmeta) noexcept
nlohmann::json::parse(data).get_to(fmeta);
return true;
} catch (nlohmann::json::exception &exp) {
- dwarn_f("decode meta data from json failed: {} [{}]", exp.what(), data);
+ LOG_WARNING_F("decode meta data from json failed: {} [{}]", exp.what(), data);
return false;
}
}
diff --git a/src/client/partition_resolver_simple.cpp b/src/client/partition_resolver_simple.cpp
index fd428eef6..2d39e4fc2 100644
--- a/src/client/partition_resolver_simple.cpp
+++ b/src/client/partition_resolver_simple.cpp
@@ -83,16 +83,16 @@ void partition_resolver_simple::on_access_failure(int partition_index, error_cod
zauto_write_lock l(_config_lock);
if (err == ERR_PARENT_PARTITION_MISUSED) {
- ddebug_f("clear all partition configuration cache due to access failure {} at {}.{}",
- err,
- _app_id,
- partition_index);
+ LOG_INFO_F("clear all partition configuration cache due to access failure {} at {}.{}",
+ err,
+ _app_id,
+ partition_index);
_app_partition_count = -1;
} else {
- ddebug_f("clear partition configuration cache {}.{} due to access failure {}",
- _app_id,
- partition_index,
- err);
+ LOG_INFO_F("clear partition configuration cache {}.{} due to access failure {}",
+ _app_id,
+ partition_index,
+ err);
_config_cache.erase(partition_index);
}
}
@@ -270,18 +270,20 @@ void partition_resolver_simple::query_config_reply(error_code err,
zauto_write_lock l(_config_lock);
if (_app_id != -1 && _app_id != resp.app_id) {
- dwarn_f("app id is changed (mostly the app was removed and created with the same "
- "name), local Vs remote: {} vs {} ",
- _app_id,
- resp.app_id);
+ LOG_WARNING_F(
+ "app id is changed (mostly the app was removed and created with the same "
+ "name), local Vs remote: {} vs {} ",
+ _app_id,
+ resp.app_id);
}
if (_app_partition_count != -1 && _app_partition_count != resp.partition_count &&
_app_partition_count * 2 != resp.partition_count &&
_app_partition_count != resp.partition_count * 2) {
- dwarn_f("partition count is changed (mostly the app was removed and created with "
- "the same name), local Vs remote: %u vs %u ",
- _app_partition_count,
- resp.partition_count);
+ LOG_WARNING_F(
+ "partition count is changed (mostly the app was removed and created with "
+ "the same name), local Vs remote: %u vs %u ",
+ _app_partition_count,
+ resp.partition_count);
}
_app_id = resp.app_id;
_app_partition_count = resp.partition_count;
diff --git a/src/common/fs_manager.cpp b/src/common/fs_manager.cpp
index 9c13de2b6..2e554df5d 100644
--- a/src/common/fs_manager.cpp
+++ b/src/common/fs_manager.cpp
@@ -89,7 +89,7 @@ bool dir_node::update_disk_stat(const bool update_disk_status)
FAIL_POINT_INJECT_F("update_disk_stat", [](string_view) { return false; });
dsn::utils::filesystem::disk_space_info info;
if (!dsn::utils::filesystem::get_disk_space_info(full_dir, info)) {
- derror_f("update disk space failed: dir = {}", full_dir);
+ LOG_ERROR_F("update disk space failed: dir = {}", full_dir);
return false;
}
// update disk space info
@@ -99,12 +99,12 @@ bool dir_node::update_disk_stat(const bool update_disk_status)
disk_capacity_mb == 0 ? 0 : std::round(disk_available_mb * 100.0 / disk_capacity_mb));
if (!update_disk_status) {
- ddebug_f("update disk space succeed: dir = {}, capacity_mb = {}, available_mb = {}, "
- "available_ratio = {}%",
- full_dir,
- disk_capacity_mb,
- disk_available_mb,
- disk_available_ratio);
+ LOG_INFO_F("update disk space succeed: dir = {}, capacity_mb = {}, available_mb = {}, "
+ "available_ratio = {}%",
+ full_dir,
+ disk_capacity_mb,
+ disk_available_mb,
+ disk_available_ratio);
return false;
}
auto old_status = status;
@@ -114,13 +114,13 @@ bool dir_node::update_disk_stat(const bool update_disk_status)
if (old_status != new_status) {
status = new_status;
}
- ddebug_f("update disk space succeed: dir = {}, capacity_mb = {}, available_mb = {}, "
- "available_ratio = {}%, disk_status = {}",
- full_dir,
- disk_capacity_mb,
- disk_available_mb,
- disk_available_ratio,
- enum_to_string(status));
+ LOG_INFO_F("update disk space succeed: dir = {}, capacity_mb = {}, available_mb = {}, "
+ "available_ratio = {}%, disk_status = {}",
+ full_dir,
+ disk_capacity_mb,
+ disk_available_mb,
+ disk_available_ratio,
+ enum_to_string(status));
return (old_status != new_status);
}
@@ -322,15 +322,15 @@ void fs_manager::update_disk_stat(bool check_status_changed)
_total_available_ratio = static_cast<int>(
_total_capacity_mb == 0 ? 0 : std::round(_total_available_mb * 100.0 / _total_capacity_mb));
- ddebug_f("update disk space succeed: disk_count = {}, total_capacity_mb = {}, "
- "total_available_mb = {}, total_available_ratio = {}%, min_available_ratio = {}%, "
- "max_available_ratio = {}%",
- _dir_nodes.size(),
- _total_capacity_mb,
- _total_available_mb,
- _total_available_ratio,
- _min_available_ratio,
- _max_available_ratio);
+ LOG_INFO_F("update disk space succeed: disk_count = {}, total_capacity_mb = {}, "
+ "total_available_mb = {}, total_available_ratio = {}%, min_available_ratio = {}%, "
+ "max_available_ratio = {}%",
+ _dir_nodes.size(),
+ _total_capacity_mb,
+ _total_available_mb,
+ _total_available_ratio,
+ _min_available_ratio,
+ _max_available_ratio);
_counter_total_capacity_mb->set(_total_capacity_mb);
_counter_total_available_mb->set(_total_available_mb);
_counter_total_available_ratio->set(_total_available_ratio);
@@ -346,7 +346,8 @@ void fs_manager::add_new_dir_node(const std::string &data_dir, const std::string
dir_node *n = new dir_node(tag, norm_path);
_dir_nodes.emplace_back(n);
_available_data_dirs.emplace_back(data_dir);
- ddebug_f("{}: mark data dir({}) as tag({})", dsn_primary_address().to_string(), norm_path, tag);
+ LOG_INFO_F(
+ "{}: mark data dir({}) as tag({})", dsn_primary_address().to_string(), norm_path, tag);
}
bool fs_manager::is_dir_node_available(const std::string &data_dir, const std::string &tag) const
diff --git a/src/common/replication_common.cpp b/src/common/replication_common.cpp
index e4c0fa930..51bf51b41 100644
--- a/src/common/replication_common.cpp
+++ b/src/common/replication_common.cpp
@@ -494,13 +494,13 @@ bool replica_helper::load_meta_servers(/*out*/ std::vector<dsn::rpc_address> &se
if (0 != (ip = ::dsn::rpc_address::ipv4_from_host(hostname_port[0].c_str()))) {
addr.assign_ipv4(ip, static_cast<uint16_t>(port_num));
} else if (!addr.from_string_ipv4(s.c_str())) {
- derror_f("invalid address '{}' specified in config [{}].{}", s, section, key);
+ LOG_ERROR_F("invalid address '{}' specified in config [{}].{}", s, section, key);
return false;
}
servers.push_back(addr);
}
if (servers.empty()) {
- derror_f("no meta server specified in config [{}].{}", section, key);
+ LOG_ERROR_F("no meta server specified in config [{}].{}", section, key);
return false;
}
return true;
@@ -562,7 +562,7 @@ replication_options::get_data_dir_and_tag(const std::string &config_dirs_str,
for (unsigned i = 0; i < dirs.size(); ++i) {
const std::string &dir = dirs[i];
- ddebug_f("data_dirs[{}] = {}, tag = {}", i + 1, dir, dir_tags[i]);
+ LOG_INFO_F("data_dirs[{}] = {}, tag = {}", i + 1, dir, dir_tags[i]);
data_dirs.push_back(utils::filesystem::path_combine(dir, "reps"));
data_dir_tags.push_back(dir_tags[i]);
}
@@ -574,11 +574,11 @@ replication_options::get_data_dirs_in_black_list(const std::string &fname,
/*out*/ std::vector<std::string> &dirs)
{
if (fname.empty() || !utils::filesystem::file_exists(fname)) {
- ddebug_f("data_dirs_black_list_file[{}] not found, ignore it", fname);
+ LOG_INFO_F("data_dirs_black_list_file[{}] not found, ignore it", fname);
return;
}
- ddebug_f("data_dirs_black_list_file[{}] found, apply it", fname);
+ LOG_INFO_F("data_dirs_black_list_file[{}] found, apply it", fname);
std::ifstream file(fname);
if (!file) {
dassert_f(false, "open data_dirs_black_list_file failed: {}", fname);
@@ -596,7 +596,7 @@ replication_options::get_data_dirs_in_black_list(const std::string &fname,
}
dirs.push_back(str2);
count++;
- ddebug_f("black_list[{}] = [{}]", count, str2);
+ LOG_INFO_F("black_list[{}] = [{}]", count, str2);
}
}
diff --git a/src/geo/lib/geo_client.cpp b/src/geo/lib/geo_client.cpp
index cee267043..bdb979721 100644
--- a/src/geo/lib/geo_client.cpp
+++ b/src/geo/lib/geo_client.cpp
@@ -120,10 +120,10 @@ int geo_client::set(const std::string &hash_key,
dsn::utils::notify_event set_completed;
auto async_set_callback = [&](int ec_, pegasus_client::internal_info &&info_) {
if (ec_ != PERR_OK) {
- derror_f("set data failed. hash_key={}, sort_key={}, error={}",
- hash_key,
- sort_key,
- get_error_string(ec_));
+ LOG_ERROR_F("set data failed. hash_key={}, sort_key={}, error={}",
+ hash_key,
+ sort_key,
+ get_error_string(ec_));
ret = ec_;
}
if (info != nullptr) {
@@ -167,11 +167,11 @@ void geo_client::async_set(const std::string &hash_key,
}
if (ec_ != PERR_OK) {
- derror_f("set {} data failed. hash_key={}, sort_key={}, error={}",
- data_type_ == DataType::common ? "common" : "geo",
- hash_key,
- sort_key,
- get_error_string(ec_));
+ LOG_ERROR_F("set {} data failed. hash_key={}, sort_key={}, error={}",
+ data_type_ == DataType::common ? "common" : "geo",
+ hash_key,
+ sort_key,
+ get_error_string(ec_));
*ret = ec_;
}
@@ -220,10 +220,10 @@ int geo_client::get(const std::string &hash_key,
lat_degrees = lat_degrees_;
lng_degrees = lng_degrees_;
} else {
- dwarn_f("get data failed. hash_key={}, sort_key={}, error={}",
- hash_key,
- sort_key,
- get_error_string(ec_));
+ LOG_WARNING_F("get data failed. hash_key={}, sort_key={}, error={}",
+ hash_key,
+ sort_key,
+ get_error_string(ec_));
}
ret = ec_;
get_completed.notify();
@@ -251,10 +251,10 @@ void geo_client::async_get(const std::string &hash_key,
}
S2LatLng latlng;
if (!_codec.decode_from_value(value_, latlng)) {
- derror_f("decode_from_value failed. hash_key={}, sort_key={}, value={}",
- hash_key,
- sort_key,
- value_);
+ LOG_ERROR_F("decode_from_value failed. hash_key={}, sort_key={}, value={}",
+ hash_key,
+ sort_key,
+ value_);
cb(PERR_GEO_DECODE_VALUE_ERROR, id, 0, 0);
return;
}
@@ -272,10 +272,10 @@ int geo_client::del(const std::string &hash_key,
dsn::utils::notify_event del_completed;
auto async_del_callback = [&](int ec_, pegasus_client::internal_info &&info_) {
if (ec_ != PERR_OK) {
- derror_f("del data failed. hash_key={}, sort_key={}, error={}",
- hash_key,
- sort_key,
- get_error_string(ec_));
+ LOG_ERROR_F("del data failed. hash_key={}, sort_key={}, error={}",
+ hash_key,
+ sort_key,
+ get_error_string(ec_));
ret = ec_;
}
if (info != nullptr) {
@@ -319,7 +319,7 @@ void geo_client::async_del(const std::string &hash_key,
std::string geo_sort_key;
if (!generate_geo_keys(hash_key, sort_key, value_, geo_hash_key, geo_sort_key)) {
keep_geo_data = true;
- dwarn_f("generate_geo_keys failed");
+ LOG_WARNING_F("generate_geo_keys failed");
}
std::shared_ptr<int> ret = std::make_shared<int>(PERR_OK);
@@ -339,11 +339,11 @@ void geo_client::async_del(const std::string &hash_key,
auto async_del_callback =
[=](int ec__, pegasus_client::internal_info &&, DataType data_type_) mutable {
if (ec__ != PERR_OK) {
- derror_f("del {} data failed. hash_key={}, sort_key={}, error={}",
- data_type_ == DataType::common ? "common" : "geo",
- hash_key,
- sort_key,
- get_error_string(ec_));
+ LOG_ERROR_F("del {} data failed. hash_key={}, sort_key={}, error={}",
+ data_type_ == DataType::common ? "common" : "geo",
+ hash_key,
+ sort_key,
+ get_error_string(ec_));
*ret = ec__;
}
@@ -373,10 +373,10 @@ int geo_client::set_geo_data(const std::string &hash_key,
auto async_set_callback = [&](int ec_, pegasus_client::internal_info &&info_) {
if (ec_ != PERR_OK) {
ret = ec_;
- derror_f("set geo data failed. hash_key={}, sort_key={}, error={}",
- hash_key,
- sort_key,
- get_error_string(ec_));
+ LOG_ERROR_F("set geo data failed. hash_key={}, sort_key={}, error={}",
+ hash_key,
+ sort_key,
+ get_error_string(ec_));
}
set_completed.notify();
};
@@ -416,7 +416,7 @@ int geo_client::search_radial(double lat_degrees,
int ret = PERR_OK;
S2LatLng latlng = S2LatLng::FromDegrees(lat_degrees, lng_degrees);
if (!latlng.is_valid()) {
- derror_f("latlng is invalid. lat_degrees={}, lng_degrees={}", lat_degrees, lng_degrees);
+ LOG_ERROR_F("latlng is invalid. lat_degrees={}, lng_degrees={}", lat_degrees, lng_degrees);
return PERR_GEO_INVALID_LATLNG_ERROR;
}
dsn::utils::notify_event search_completed;
@@ -446,7 +446,7 @@ void geo_client::async_search_radial(double lat_degrees,
{
S2LatLng latlng = S2LatLng::FromDegrees(lat_degrees, lng_degrees);
if (!latlng.is_valid()) {
- derror_f("latlng is invalid. lat_degrees={}, lng_degrees={}", lat_degrees, lng_degrees);
+ LOG_ERROR_F("latlng is invalid. lat_degrees={}, lng_degrees={}", lat_degrees, lng_degrees);
callback(PERR_GEO_INVALID_LATLNG_ERROR, {});
}
@@ -503,20 +503,20 @@ void geo_client::async_search_radial(const std::string &hash_key,
cb = std::move(callback)
](int ec_, std::string &&value_, pegasus_client::internal_info &&) mutable {
if (ec_ != PERR_OK) {
- derror_f("get failed. hash_key={}, sort_key={}, error={}",
- hash_key,
- sort_key,
- get_error_string(ec_));
+ LOG_ERROR_F("get failed. hash_key={}, sort_key={}, error={}",
+ hash_key,
+ sort_key,
+ get_error_string(ec_));
cb(ec_, {});
return;
}
S2LatLng latlng;
if (!_codec.decode_from_value(value_, latlng)) {
- derror_f("decode_from_value failed. hash_key={}, sort_key={}, value={}",
- hash_key,
- sort_key,
- value_);
+ LOG_ERROR_F("decode_from_value failed. hash_key={}, sort_key={}, value={}",
+ hash_key,
+ sort_key,
+ value_);
cb(ec_, {});
return;
}
@@ -705,10 +705,10 @@ bool geo_client::generate_geo_keys(const std::string &hash_key,
// extract latitude and longitude from value
S2LatLng latlng;
if (!_codec.decode_from_value(value, latlng)) {
- derror_f("decode_from_value failed. hash_key={}, sort_key={}, value={}",
- hash_key,
- sort_key,
- value);
+ LOG_ERROR_F("decode_from_value failed. hash_key={}, sort_key={}, value={}",
+ hash_key,
+ sort_key,
+ value);
return false;
}
@@ -883,14 +883,14 @@ void geo_client::do_scan(pegasus_client::pegasus_scanner_wrapper scanner_wrapper
}
if (ret != PERR_OK) {
- derror_f("async_next failed. error={}", get_error_string(ret));
+ LOG_ERROR_F("async_next failed. error={}", get_error_string(ret));
cb();
return;
}
S2LatLng latlng;
if (!_codec.decode_from_value(value, latlng)) {
- derror_f("decode_from_value failed. value={}", value);
+ LOG_ERROR_F("decode_from_value failed. value={}", value);
cb();
return;
}
@@ -899,7 +899,7 @@ void geo_client::do_scan(pegasus_client::pegasus_scanner_wrapper scanner_wrapper
if (distance <= S2Earth::ToMeters(cap_ptr->radius())) {
std::string origin_hash_key, origin_sort_key;
if (!restore_origin_keys(geo_sort_key, origin_hash_key, origin_sort_key)) {
- derror_f("restore_origin_keys failed. geo_sort_key={}", geo_sort_key);
+ LOG_ERROR_F("restore_origin_keys failed. geo_sort_key={}", geo_sort_key);
cb();
return;
}
@@ -932,13 +932,14 @@ int geo_client::distance(const std::string &hash_key1,
dsn::utils::notify_event get_completed;
auto async_calculate_callback = [&](int ec_, double &&distance_) {
if (ec_ != PERR_OK) {
- derror_f("get distance failed. hash_key1={}, sort_key1={}, hash_key2={}, sort_key2={}, "
- "error={}",
- hash_key1,
- sort_key1,
- hash_key2,
- sort_key2,
- get_error_string(ec_));
+ LOG_ERROR_F(
+ "get distance failed. hash_key1={}, sort_key1={}, hash_key2={}, sort_key2={}, "
+ "error={}",
+ hash_key1,
+ sort_key1,
+ hash_key2,
+ sort_key2,
+ get_error_string(ec_));
ret = ec_;
}
distance = distance_;
@@ -965,19 +966,19 @@ void geo_client::async_distance(const std::string &hash_key1,
int ec_, std::string &&value_, pegasus_client::internal_info &&)
{
if (ec_ != PERR_OK) {
- derror_f("get data failed. hash_key1={}, sort_key1={}, hash_key2={}, sort_key2={}, "
- "error={}",
- hash_key1,
- sort_key1,
- hash_key2,
- sort_key2,
- get_error_string(ec_));
+ LOG_ERROR_F("get data failed. hash_key1={}, sort_key1={}, hash_key2={}, sort_key2={}, "
+ "error={}",
+ hash_key1,
+ sort_key1,
+ hash_key2,
+ sort_key2,
+ get_error_string(ec_));
*ret = ec_;
}
S2LatLng latlng;
if (!_codec.decode_from_value(value_, latlng)) {
- derror_f("decode_from_value failed. value={}", value_);
+ LOG_ERROR_F("decode_from_value failed. value={}", value_);
*ret = PERR_GEO_DECODE_VALUE_ERROR;
}
diff --git a/src/geo/lib/latlng_codec.cpp b/src/geo/lib/latlng_codec.cpp
index bc52240e1..2f26d6577 100644
--- a/src/geo/lib/latlng_codec.cpp
+++ b/src/geo/lib/latlng_codec.cpp
@@ -95,7 +95,7 @@ bool latlng_codec::encode_to_value(double lat_degrees, double lng_degrees, std::
dcheck_eq(_sorted_indices.size(), 2);
S2LatLng latlng = S2LatLng::FromDegrees(lat_degrees, lng_degrees);
if (!latlng.is_valid()) {
- derror_f("latlng is invalid. lat_degrees={}, lng_degrees={}", lat_degrees, lng_degrees);
+ LOG_ERROR_F("latlng is invalid. lat_degrees={}, lng_degrees={}", lat_degrees, lng_degrees);
return false;
}
diff --git a/src/http/pprof_http_service.cpp b/src/http/pprof_http_service.cpp
index 8e6758fd0..4b9d99003 100644
--- a/src/http/pprof_http_service.cpp
+++ b/src/http/pprof_http_service.cpp
@@ -328,7 +328,7 @@ void pprof_http_service::heap_handler(const http_request &req, http_response &re
{
bool in_pprof = false;
if (!_in_pprof_action.compare_exchange_strong(in_pprof, true)) {
- dwarn_f("node is already exectuting pprof action, please wait and retry");
+ LOG_WARNING_F("node is already exectuting pprof action, please wait and retry");
resp.status_code = http_status_code::internal_server_error;
return;
}
@@ -425,7 +425,7 @@ void pprof_http_service::growth_handler(const http_request &req, http_response &
{
bool in_pprof = false;
if (!_in_pprof_action.compare_exchange_strong(in_pprof, true)) {
- dwarn_f("node is already exectuting pprof action, please wait and retry");
+ LOG_WARNING_F("node is already exectuting pprof action, please wait and retry");
resp.status_code = http_status_code::internal_server_error;
return;
}
@@ -468,7 +468,7 @@ void pprof_http_service::profile_handler(const http_request &req, http_response
{
bool in_pprof = false;
if (!_in_pprof_action.compare_exchange_strong(in_pprof, true)) {
- dwarn_f("node is already exectuting pprof action, please wait and retry");
+ LOG_WARNING_F("node is already exectuting pprof action, please wait and retry");
resp.status_code = http_status_code::internal_server_error;
return;
}
diff --git a/src/meta/app_balance_policy.cpp b/src/meta/app_balance_policy.cpp
index b404f6d48..9775e74ac 100644
--- a/src/meta/app_balance_policy.cpp
+++ b/src/meta/app_balance_policy.cpp
@@ -156,7 +156,7 @@ bool copy_secondary_operation::can_continue()
int id_max = *_ordered_address_ids.rbegin();
if (_partition_counts[id_max] <= _replicas_low ||
_partition_counts[id_max] - _partition_counts[id_min] <= 1) {
- ddebug_f("{}: stop copy secondary coz it will be balanced later", _app->get_logname());
+ LOG_INFO_F("{}: stop copy secondary coz it will be balanced later", _app->get_logname());
return false;
}
return true;
@@ -172,29 +172,29 @@ bool copy_secondary_operation::can_select(gpid pid, migration_list *result)
int id_max = *_ordered_address_ids.rbegin();
const node_state &max_ns = _nodes.at(_address_vec[id_max]);
if (max_ns.served_as(pid) == partition_status::PS_PRIMARY) {
- dinfo_f("{}: skip gpid({}.{}) coz it is primary",
- _app->get_logname(),
- pid.get_app_id(),
- pid.get_partition_index());
+ LOG_DEBUG_F("{}: skip gpid({}.{}) coz it is primary",
+ _app->get_logname(),
+ pid.get_app_id(),
+ pid.get_partition_index());
return false;
}
// if the pid have been used
if (result->find(pid) != result->end()) {
- dinfo_f("{}: skip gpid({}.{}) coz it is already copyed",
- _app->get_logname(),
- pid.get_app_id(),
- pid.get_partition_index());
+ LOG_DEBUG_F("{}: skip gpid({}.{}) coz it is already copyed",
+ _app->get_logname(),
+ pid.get_app_id(),
+ pid.get_partition_index());
return false;
}
int id_min = *_ordered_address_ids.begin();
const node_state &min_ns = _nodes.at(_address_vec[id_min]);
if (min_ns.served_as(pid) != partition_status::PS_INACTIVE) {
- dinfo_f("{}: skip gpid({}.{}) coz it is already a member on the target node",
- _app->get_logname(),
- pid.get_app_id(),
- pid.get_partition_index());
+ LOG_DEBUG_F("{}: skip gpid({}.{}) coz it is already a member on the target node",
+ _app->get_logname(),
+ pid.get_app_id(),
+ pid.get_partition_index());
return false;
}
return true;
diff --git a/src/meta/app_env_validator.cpp b/src/meta/app_env_validator.cpp
index ed112683b..45edf8b4c 100644
--- a/src/meta/app_env_validator.cpp
+++ b/src/meta/app_env_validator.cpp
@@ -154,7 +154,7 @@ bool app_env_validator::validate_app_env(const std::string &env_name,
if (func_iter != _validator_funcs.end()) {
// check function == nullptr means no check
if (nullptr != func_iter->second && !func_iter->second(env_value, hint_message)) {
- dwarn_f("{}={} is invalid.", env_name, env_value);
+ LOG_WARNING_F("{}={} is invalid.", env_name, env_value);
return false;
}
diff --git a/src/meta/backup_engine.cpp b/src/meta/backup_engine.cpp
index 69c316016..a66a48ae0 100644
--- a/src/meta/backup_engine.cpp
+++ b/src/meta/backup_engine.cpp
@@ -41,7 +41,7 @@ error_code backup_engine::init_backup(int32_t app_id)
_backup_service->get_state()->lock_read(l);
std::shared_ptr<app_state> app = _backup_service->get_state()->get_app(app_id);
if (app == nullptr || app->status != app_status::AS_AVAILABLE) {
- derror_f("app {} is not available, couldn't do backup now.", app_id);
+ LOG_ERROR_F("app {} is not available, couldn't do backup now.", app_id);
return ERR_INVALID_STATE;
}
app_name = app->app_name;
@@ -77,7 +77,7 @@ error_code backup_engine::set_backup_path(const std::string &path)
if (_block_service && _block_service->is_root_path_set()) {
return ERR_INVALID_PARAMETERS;
}
- ddebug_f("backup path is set to {}.", path);
+ LOG_INFO_F("backup path is set to {}.", path);
_backup_path = path;
return ERR_OK;
}
@@ -100,7 +100,7 @@ error_code backup_engine::write_backup_file(const std::string &file_name,
})
->wait();
if (err != dsn::ERR_OK) {
- ddebug_f("create file {} failed", file_name);
+ LOG_INFO_F("create file {} failed", file_name);
return err;
}
dassert_f(remote_file != nullptr,
@@ -122,7 +122,7 @@ error_code backup_engine::backup_app_meta()
_backup_service->get_state()->lock_read(l);
std::shared_ptr<app_state> app = _backup_service->get_state()->get_app(_cur_backup.app_id);
if (app == nullptr || app->status != app_status::AS_AVAILABLE) {
- derror_f("app {} is not available, couldn't do backup now.", _cur_backup.app_id);
+ LOG_ERROR_F("app {} is not available, couldn't do backup now.", _cur_backup.app_id);
return ERR_INVALID_STATE;
}
app_state tmp = *app;
@@ -147,7 +147,7 @@ void backup_engine::backup_app_partition(const gpid &pid)
_backup_service->get_state()->lock_read(l);
std::shared_ptr<app_state> app = _backup_service->get_state()->get_app(pid.get_app_id());
if (app == nullptr || app->status != app_status::AS_AVAILABLE) {
- derror_f("app {} is not available, couldn't do backup now.", pid.get_app_id());
+ LOG_ERROR_F("app {} is not available, couldn't do backup now.", pid.get_app_id());
zauto_lock lock(_lock);
_is_backup_failed = true;
@@ -157,9 +157,10 @@ void backup_engine::backup_app_partition(const gpid &pid)
}
if (partition_primary.is_invalid()) {
- dwarn_f("backup_id({}): partition {} doesn't have a primary now, retry to backup it later.",
- _cur_backup.backup_id,
- pid.to_string());
+ LOG_WARNING_F(
+ "backup_id({}): partition {} doesn't have a primary now, retry to backup it later.",
+ _cur_backup.backup_id,
+ pid.to_string());
tasking::enqueue(LPC_DEFAULT_CALLBACK,
&_tracker,
[this, pid]() { backup_app_partition(pid); },
@@ -180,10 +181,10 @@ void backup_engine::backup_app_partition(const gpid &pid)
req->__set_backup_path(_backup_path);
}
- ddebug_f("backup_id({}): send backup request to partition {}, target_addr = {}",
- _cur_backup.backup_id,
- pid.to_string(),
- partition_primary.to_string());
+ LOG_INFO_F("backup_id({}): send backup request to partition {}, target_addr = {}",
+ _cur_backup.backup_id,
+ pid.to_string(),
+ partition_primary.to_string());
backup_rpc rpc(std::move(req), RPC_COLD_BACKUP, 10000_ms, 0, pid.thread_hash());
rpc.call(
partition_primary, &_tracker, [this, rpc, pid, partition_primary](error_code err) mutable {
@@ -200,10 +201,10 @@ inline void backup_engine::handle_replica_backup_failed(const backup_response &r
dcheck_eq(response.pid, pid);
dcheck_eq(response.backup_id, _cur_backup.backup_id);
- derror_f("backup_id({}): backup for partition {} failed, response.err: {}",
- _cur_backup.backup_id,
- pid.to_string(),
- response.err.to_string());
+ LOG_ERROR_F("backup_id({}): backup for partition {} failed, response.err: {}",
+ _cur_backup.backup_id,
+ pid.to_string(),
+ response.err.to_string());
zauto_lock l(_lock);
// if one partition fail, the whole backup plan fail.
_is_backup_failed = true;
@@ -246,11 +247,11 @@ void backup_engine::on_backup_reply(const error_code err,
}
if (rep_error != ERR_OK) {
- derror_f("backup_id({}): backup request to server {} failed, error: {}, retry to "
- "send backup request.",
- _cur_backup.backup_id,
- primary.to_string(),
- rep_error.to_string());
+ LOG_ERROR_F("backup_id({}): backup request to server {} failed, error: {}, retry to "
+ "send backup request.",
+ _cur_backup.backup_id,
+ primary.to_string(),
+ rep_error.to_string());
retry_backup(pid);
return;
};
@@ -258,9 +259,9 @@ void backup_engine::on_backup_reply(const error_code err,
if (response.progress == cold_backup_constant::PROGRESS_FINISHED) {
dcheck_eq(response.pid, pid);
dcheck_eq(response.backup_id, _cur_backup.backup_id);
- ddebug_f("backup_id({}): backup for partition {} completed.",
- _cur_backup.backup_id,
- pid.to_string());
+ LOG_INFO_F("backup_id({}): backup for partition {} completed.",
+ _cur_backup.backup_id,
+ pid.to_string());
{
zauto_lock l(_lock);
_backup_status[pid.get_partition_index()] = backup_status::COMPLETED;
@@ -270,12 +271,12 @@ void backup_engine::on_backup_reply(const error_code err,
}
// backup is not finished, meta polling to send request
- ddebug_f("backup_id({}): receive backup response for partition {} from server {}, now "
- "progress {}, retry to send backup request.",
- _cur_backup.backup_id,
- pid.to_string(),
- primary.to_string(),
- response.progress);
+ LOG_INFO_F("backup_id({}): receive backup response for partition {} from server {}, now "
+ "progress {}, retry to send backup request.",
+ _cur_backup.backup_id,
+ pid.to_string(),
+ primary.to_string(),
+ response.progress);
retry_backup(pid);
}
@@ -288,7 +289,7 @@ void backup_engine::write_backup_info()
blob buf = dsn::json::json_forwarder<app_backup_info>::encode(_cur_backup);
error_code err = write_backup_file(file_name, buf);
if (err == ERR_FS_INTERNAL) {
- derror_f(
+ LOG_ERROR_F(
"backup_id({}): write backup info failed, error {}, do not try again for this error.",
_cur_backup.backup_id,
err.to_string());
@@ -297,7 +298,8 @@ void backup_engine::write_backup_info()
return;
}
if (err != ERR_OK) {
- dwarn_f("backup_id({}): write backup info failed, retry it later.", _cur_backup.backup_id);
+ LOG_WARNING_F("backup_id({}): write backup info failed, retry it later.",
+ _cur_backup.backup_id);
tasking::enqueue(LPC_DEFAULT_CALLBACK,
&_tracker,
[this]() { write_backup_info(); },
@@ -305,9 +307,9 @@ void backup_engine::write_backup_info()
std::chrono::seconds(1));
return;
}
- ddebug_f("backup_id({}): successfully wrote backup info, backup for app {} completed.",
- _cur_backup.backup_id,
- _cur_backup.app_id);
+ LOG_INFO_F("backup_id({}): successfully wrote backup info, backup for app {} completed.",
+ _cur_backup.backup_id,
+ _cur_backup.app_id);
zauto_lock l(_lock);
_cur_backup.end_time_ms = dsn_now_ms();
}
@@ -331,10 +333,10 @@ error_code backup_engine::start()
{
error_code err = backup_app_meta();
if (err != ERR_OK) {
- derror_f("backup_id({}): backup meta data for app {} failed, error {}",
- _cur_backup.backup_id,
- _cur_backup.app_id,
- err.to_string());
+ LOG_ERROR_F("backup_id({}): backup meta data for app {} failed, error {}",
+ _cur_backup.backup_id,
+ _cur_backup.app_id,
+ err.to_string());
return err;
}
for (int i = 0; i < _backup_status.size(); ++i) {
diff --git a/src/meta/cluster_balance_policy.cpp b/src/meta/cluster_balance_policy.cpp
index f8a5e5d70..ee5ef998d 100644
--- a/src/meta/cluster_balance_policy.cpp
+++ b/src/meta/cluster_balance_policy.cpp
@@ -120,7 +120,7 @@ bool cluster_balance_policy::cluster_replica_balance(const meta_view *global_vie
return false;
}
if (!list.empty()) {
- ddebug_f("migration count of {} = {}", enum_to_string(type), list.size());
+ LOG_INFO_F("migration count of {} = {}", enum_to_string(type), list.size());
return false;
}
return true;
@@ -165,11 +165,11 @@ bool cluster_balance_policy::get_cluster_migration_info(
const std::shared_ptr<app_state> &app = kv.second;
auto ignored = is_ignored_app(app->app_id);
if (ignored || app->is_bulk_loading || app->splitting()) {
- ddebug_f("skip to balance app({}), ignored={}, bulk loading={}, splitting={}",
- app->app_name,
- ignored,
- app->is_bulk_loading,
- app->splitting());
+ LOG_INFO_F("skip to balance app({}), ignored={}, bulk loading={}, splitting={}",
+ app->app_name,
+ ignored,
+ app->is_bulk_loading,
+ app->splitting());
continue;
}
if (app->status == app_status::AS_AVAILABLE) {
@@ -263,13 +263,13 @@ bool cluster_balance_policy::get_next_move(const cluster_migration_info &cluster
std::multimap<uint32_t, int32_t> app_skew_multimap = utils::flip_map(cluster_info.apps_skew);
auto max_app_skew = app_skew_multimap.rbegin()->first;
if (max_app_skew == 0) {
- ddebug_f("every app is balanced and any move will unbalance a app");
+ LOG_INFO_F("every app is balanced and any move will unbalance a app");
return false;
}
auto server_skew = get_skew(cluster_info.replicas_count);
if (max_app_skew <= 1 && server_skew <= 1) {
- ddebug_f("every app is balanced and the cluster as a whole is balanced");
+ LOG_INFO_F("every app is balanced and the cluster as a whole is balanced");
return false;
}
@@ -314,10 +314,10 @@ bool cluster_balance_policy::get_next_move(const cluster_migration_info &cluster
std::multimap<uint32_t, rpc_address> app_count_multimap = utils::flip_map(app_map);
if (app_count_multimap.rbegin()->first <= app_count_multimap.begin()->first + 1 &&
(app_cluster_min_set.empty() || app_cluster_max_set.empty())) {
- ddebug_f("do not move replicas of a balanced app({}) if the least (most) loaded "
- "servers overall do not intersect the servers hosting the least (most) "
- "replicas of the app",
- app_id);
+ LOG_INFO_F("do not move replicas of a balanced app({}) if the least (most) loaded "
+ "servers overall do not intersect the servers hosting the least (most) "
+ "replicas of the app",
+ app_id);
continue;
}
@@ -357,10 +357,10 @@ bool cluster_balance_policy::pick_up_move(const cluster_migration_info &cluster_
}
auto index = rand() % max_load_disk_set.size();
auto max_load_disk = *select_random(max_load_disk_set, index);
- ddebug_f("most load disk({}) on node({}) is picked, has {} partition",
- max_load_disk.node.to_string(),
- max_load_disk.disk_tag,
- max_load_disk.partitions.size());
+ LOG_INFO_F("most load disk({}) on node({}) is picked, has {} partition",
+ max_load_disk.node.to_string(),
+ max_load_disk.disk_tag,
+ max_load_disk.partitions.size());
for (const auto &node_addr : min_nodes) {
gpid picked_pid;
if (pick_up_partition(
@@ -370,17 +370,17 @@ bool cluster_balance_policy::pick_up_move(const cluster_migration_info &cluster_
move_info.source_disk_tag = max_load_disk.disk_tag;
move_info.target_node = node_addr;
move_info.type = cluster_info.type;
- ddebug_f("partition[{}] will migrate from {} to {}",
- picked_pid,
- max_load_disk.node.to_string(),
- node_addr.to_string());
+ LOG_INFO_F("partition[{}] will migrate from {} to {}",
+ picked_pid,
+ max_load_disk.node.to_string(),
+ node_addr.to_string());
return true;
}
}
- ddebug_f("can not find a partition(app_id={}) from random max load disk(node={}, disk={})",
- app_id,
- max_load_disk.node.to_string(),
- max_load_disk.disk_tag);
+ LOG_INFO_F("can not find a partition(app_id={}) from random max load disk(node={}, disk={})",
+ app_id,
+ max_load_disk.node.to_string(),
+ max_load_disk.disk_tag);
return false;
}
diff --git a/src/meta/duplication/duplication_info.cpp b/src/meta/duplication/duplication_info.cpp
index 373aea2c6..ad8e7dcfc 100644
--- a/src/meta/duplication/duplication_info.cpp
+++ b/src/meta/duplication/duplication_info.cpp
@@ -48,7 +48,7 @@ namespace replication {
s = it->second;
return true;
}
- derror_f("unexpected duplication_status name: {}", name);
+ LOG_ERROR_F("unexpected duplication_status name: {}", name);
// for forward compatibility issue, duplication of unexpected status
// will be marked as invisible.
@@ -77,7 +77,7 @@ namespace replication {
fmode = it->second;
return true;
}
- derror_f("unexpected duplication_fail_mode name: {}", name);
+ LOG_ERROR_F("unexpected duplication_fail_mode name: {}", name);
// marked as default value.
fmode = duplication_fail_mode::FAIL_SLOW;
return false;
@@ -197,7 +197,7 @@ void duplication_info::report_progress_if_time_up()
// progress report is not supposed to be too frequent.
if (dsn_now_ms() > _last_progress_report_ms + PROGRESS_REPORT_PERIOD_MS) {
_last_progress_report_ms = dsn_now_ms();
- ddebug_f("duplication report: {}", to_string());
+ LOG_INFO_F("duplication report: {}", to_string());
}
}
diff --git a/src/meta/duplication/duplication_info.h b/src/meta/duplication/duplication_info.h
index b4d08704f..dbf007e8c 100644
--- a/src/meta/duplication/duplication_info.h
+++ b/src/meta/duplication/duplication_info.h
@@ -74,10 +74,10 @@ public:
if (is_duplicating_checkpoint) {
return alter_status(duplication_status::DS_PREPARE);
}
- dwarn_f("you now create duplication[{}[{}.{}]] without duplicating checkpoint",
- id,
- follower_cluster_name,
- app_name);
+ LOG_WARNING_F("you now create duplication[{}[{}.{}]] without duplicating checkpoint",
+ id,
+ follower_cluster_name,
+ app_name);
return alter_status(duplication_status::DS_LOG);
}
@@ -168,7 +168,7 @@ public:
return item.second.checkpoint_prepared;
});
if (!completed) {
- dwarn_f("replica checkpoint still running: {}/{}", prepared, _progress.size());
+ LOG_WARNING_F("replica checkpoint still running: {}/{}", prepared, _progress.size());
}
return completed;
}
@@ -249,13 +249,13 @@ extern bool json_decode(const dsn::json::JsonObject &in, duplication_fail_mode::
// Macros for writing log message prefixed by appid and dupid.
#define ddebug_dup(_dup_, ...) \
- ddebug_f("[a{}d{}] {}", _dup_->app_id, _dup_->id, fmt::format(__VA_ARGS__));
+ LOG_INFO_F("[a{}d{}] {}", _dup_->app_id, _dup_->id, fmt::format(__VA_ARGS__));
#define dwarn_dup(_dup_, ...) \
- dwarn_f("[a{}d{}] {}", _dup_->app_id, _dup_->id, fmt::format(__VA_ARGS__));
+ LOG_WARNING_F("[a{}d{}] {}", _dup_->app_id, _dup_->id, fmt::format(__VA_ARGS__));
#define derror_dup(_dup_, ...) \
- derror_f("[a{}d{}] {}", _dup_->app_id, _dup_->id, fmt::format(__VA_ARGS__));
+ LOG_ERROR_F("[a{}d{}] {}", _dup_->app_id, _dup_->id, fmt::format(__VA_ARGS__));
#define dfatal_dup(_dup_, ...) \
- dfatal_f("[a{}d{}] {}", _dup_->app_id, _dup_->id, fmt::format(__VA_ARGS__));
+ LOG_FATAL_F("[a{}d{}] {}", _dup_->app_id, _dup_->id, fmt::format(__VA_ARGS__));
#define dassert_dup(_pred_, _dup_, ...) \
dassert_f(_pred_, "[a{}d{}] {}", _dup_->app_id, _dup_->id, fmt::format(__VA_ARGS__));
diff --git a/src/meta/duplication/meta_duplication_service.cpp b/src/meta/duplication/meta_duplication_service.cpp
index fe36975c9..54b021580 100644
--- a/src/meta/duplication/meta_duplication_service.cpp
+++ b/src/meta/duplication/meta_duplication_service.cpp
@@ -35,7 +35,7 @@ using namespace literals::chrono_literals;
void meta_duplication_service::query_duplication_info(const duplication_query_request &request,
duplication_query_response &response)
{
- ddebug_f("query duplication info for app: {}", request.app_name);
+ LOG_INFO_F("query duplication info for app: {}", request.app_name);
response.err = ERR_OK;
{
@@ -59,11 +59,12 @@ void meta_duplication_service::modify_duplication(duplication_modify_rpc rpc)
const auto &request = rpc.request();
auto &response = rpc.response();
- ddebug_f("modify duplication({}) to [status={},fail_mode={}] for app({})",
- request.dupid,
- request.__isset.status ? duplication_status_to_string(request.status) : "nil",
- request.__isset.fail_mode ? duplication_fail_mode_to_string(request.fail_mode) : "nil",
- request.app_name);
+ LOG_INFO_F("modify duplication({}) to [status={},fail_mode={}] for app({})",
+ request.dupid,
+ request.__isset.status ? duplication_status_to_string(request.status) : "nil",
+ request.__isset.fail_mode ? duplication_fail_mode_to_string(request.fail_mode)
+ : "nil",
+ request.app_name);
dupid_t dupid = request.dupid;
@@ -132,9 +133,9 @@ void meta_duplication_service::add_duplication(duplication_add_rpc rpc)
const auto &request = rpc.request();
auto &response = rpc.response();
- ddebug_f("add duplication for app({}), remote cluster name is {}",
- request.app_name,
- request.remote_cluster_name);
+ LOG_INFO_F("add duplication for app({}), remote cluster name is {}",
+ request.app_name,
+ request.remote_cluster_name);
response.err = ERR_OK;
@@ -190,7 +191,7 @@ void meta_duplication_service::do_add_duplication(std::shared_ptr<app_state> &ap
{
const auto err = dup->start(rpc.request().is_duplicating_checkpoint);
if (dsn_unlikely(err != ERR_OK)) {
- derror_f("start dup[{}({})] failed: err = {}", app->app_name, dup->id, err.to_string());
+ LOG_ERROR_F("start dup[{}({})] failed: err = {}", app->app_name, dup->id, err.to_string());
return;
}
blob value = dup->to_json_blob();
@@ -251,7 +252,7 @@ void meta_duplication_service::duplication_sync(duplication_sync_rpc rpc)
node_state *ns = get_node_state(_state->_nodes, request.node, false);
if (ns == nullptr) {
- dwarn_f("node({}) is not found in meta server", request.node.to_string());
+ LOG_WARNING_F("node({}) is not found in meta server", request.node.to_string());
response.err = ERR_OBJECT_NOT_FOUND;
return;
}
@@ -367,13 +368,13 @@ void meta_duplication_service::create_follower_app_for_duplication(
std::move(value),
[=]() { dup->persist_status(); });
} else {
- derror_f("created follower app[{}.{}] to trigger duplicate checkpoint failed: "
- "duplication_status = {}, create_err = {}, update_err = {}",
- dup->follower_cluster_name,
- dup->app_name,
- duplication_status_to_string(dup->status()),
- create_err.to_string(),
- update_err.to_string());
+ LOG_ERROR_F("created follower app[{}.{}] to trigger duplicate checkpoint failed: "
+ "duplication_status = {}, create_err = {}, update_err = {}",
+ dup->follower_cluster_name,
+ dup->app_name,
+ duplication_status_to_string(dup->status()),
+ create_err.to_string(),
+ update_err.to_string());
}
});
}
@@ -450,13 +451,14 @@ void meta_duplication_service::check_follower_app_if_create_completed(
std::move(value),
[dup]() { dup->persist_status(); });
} else {
- derror_f("query follower app[{}.{}] replica configuration completed, result: "
- "duplication_status = {}, query_err = {}, update_err = {}",
- dup->follower_cluster_name,
- dup->app_name,
- duplication_status_to_string(dup->status()),
- query_err.to_string(),
- update_err);
+ LOG_ERROR_F(
+ "query follower app[{}.{}] replica configuration completed, result: "
+ "duplication_status = {}, query_err = {}, update_err = {}",
+ dup->follower_cluster_name,
+ dup->app_name,
+ duplication_status_to_string(dup->status()),
+ query_err.to_string(),
+ update_err);
}
});
}
@@ -532,7 +534,7 @@ meta_duplication_service::new_dup_from_init(const std::string &follower_cluster_
// ThreadPool(WRITE): THREAD_POOL_META_STATE
void meta_duplication_service::recover_from_meta_state()
{
- ddebug_f("recovering duplication states from meta storage");
+ LOG_INFO_F("recovering duplication states from meta storage");
// /<app>/duplication/<dupid>/<partition_idx>
// | |-> confirmed_decree
@@ -556,8 +558,8 @@ void meta_duplication_service::recover_from_meta_state()
dupid_t dup_id;
if (!buf2int32(raw_dup_id, dup_id)) {
// unlikely
- derror_f("invalid duplication path: {}",
- get_duplication_path(*app, raw_dup_id));
+ LOG_ERROR_F("invalid duplication path: {}",
+ get_duplication_path(*app, raw_dup_id));
return;
}
do_restore_duplication(dup_id, app);
@@ -620,7 +622,8 @@ void meta_duplication_service::do_restore_duplication(dupid_t dup_id,
auto dup = duplication_info::decode_from_blob(
dup_id, app->app_id, app->app_name, app->partition_count, store_path, json);
if (nullptr == dup) {
- derror_f("failed to decode json \"{}\" on path {}", json.to_string(), store_path);
+ LOG_ERROR_F(
+ "failed to decode json \"{}\" on path {}", json.to_string(), store_path);
return; // fail fast
}
if (!dup->is_invalid_status()) {
diff --git a/src/meta/load_balance_policy.cpp b/src/meta/load_balance_policy.cpp
index b83f029ea..c805528c3 100644
--- a/src/meta/load_balance_policy.cpp
+++ b/src/meta/load_balance_policy.cpp
@@ -95,9 +95,10 @@ get_node_loads(const std::shared_ptr<app_state> &app,
for (auto iter = nodes.begin(); iter != nodes.end(); ++iter) {
if (!calc_disk_load(
nodes, apps, app->app_id, iter->first, only_primary, node_loads[iter->first])) {
- dwarn_f("stop the balancer as some replica infos aren't collected, node({}), app({})",
- iter->first.to_string(),
- app->get_logname());
+ LOG_WARNING_F(
+ "stop the balancer as some replica infos aren't collected, node({}), app({})",
+ iter->first.to_string(),
+ app->get_logname());
return node_loads;
}
}
@@ -192,24 +193,25 @@ bool load_balance_policy::primary_balance(const std::shared_ptr<app_state> &app,
{
dassert(_alive_nodes >= FLAGS_min_live_node_count_for_unfreeze,
"too few alive nodes will lead to freeze");
- ddebug_f("primary balancer for app({}:{})", app->app_name, app->app_id);
+ LOG_INFO_F("primary balancer for app({}:{})", app->app_name, app->app_id);
auto graph = ford_fulkerson::builder(app, *_global_view->nodes, address_id).build();
if (nullptr == graph) {
- dinfo_f("the primaries are balanced for app({}:{})", app->app_name, app->app_id);
+ LOG_DEBUG_F("the primaries are balanced for app({}:{})", app->app_name, app->app_id);
return true;
}
auto path = graph->find_shortest_path();
if (path != nullptr) {
- dinfo_f("{} primaries are flew", path->_flow.back());
+ LOG_DEBUG_F("{} primaries are flew", path->_flow.back());
return move_primary(std::move(path));
} else {
- ddebug_f("we can't make the server load more balanced by moving primaries to secondaries");
+ LOG_INFO_F(
+ "we can't make the server load more balanced by moving primaries to secondaries");
if (!only_move_primary) {
return copy_primary(app, graph->have_less_than_average());
} else {
- ddebug_f("stop to copy primary for app({}) coz it is disabled", app->get_logname());
+ LOG_INFO_F("stop to copy primary for app({}) coz it is disabled", app->get_logname());
return true;
}
}
@@ -241,9 +243,9 @@ bool load_balance_policy::move_primary(std::unique_ptr<flow_path> path)
int current = path->_prev.back();
if (!calc_disk_load(
nodes, apps, path->_app->app_id, address_vec[current], true, *current_load)) {
- dwarn_f("stop move primary as some replica infos aren't collected, node({}), app({})",
- address_vec[current].to_string(),
- path->_app->get_logname());
+ LOG_WARNING_F("stop move primary as some replica infos aren't collected, node({}), app({})",
+ address_vec[current].to_string(),
+ path->_app->get_logname());
return false;
}
@@ -252,9 +254,10 @@ bool load_balance_policy::move_primary(std::unique_ptr<flow_path> path)
rpc_address from = address_vec[path->_prev[current]];
rpc_address to = address_vec[current];
if (!calc_disk_load(nodes, apps, path->_app->app_id, from, true, *prev_load)) {
- dwarn_f("stop move primary as some replica infos aren't collected, node({}), app({})",
- from.to_string(),
- path->_app->get_logname());
+ LOG_WARNING_F(
+ "stop move primary as some replica infos aren't collected, node({}), app({})",
+ from.to_string(),
+ path->_app->get_logname());
return false;
}
@@ -349,7 +352,7 @@ bool load_balance_policy::execute_balance(
for (const auto &kv : apps) {
const std::shared_ptr<app_state> &app = kv.second;
if (is_ignored_app(kv.first)) {
- ddebug_f("skip to do balance for the ignored app[{}]", app->get_logname());
+ LOG_INFO_F("skip to do balance for the ignored app[{}]", app->get_logname());
continue;
}
if (app->status != app_status::AS_AVAILABLE || app->is_bulk_loading || app->splitting())
@@ -770,15 +773,15 @@ bool copy_primary_operation::can_continue()
{
int id_min = *_ordered_address_ids.begin();
if (_have_lower_than_average && _partition_counts[id_min] >= _replicas_low) {
- ddebug_f("{}: stop the copy due to primaries on all nodes will reach low later.",
- _app->get_logname());
+ LOG_INFO_F("{}: stop the copy due to primaries on all nodes will reach low later.",
+ _app->get_logname());
return false;
}
int id_max = *_ordered_address_ids.rbegin();
if (!_have_lower_than_average && _partition_counts[id_max] - _partition_counts[id_min] <= 1) {
- ddebug_f("{}: stop the copy due to the primary will be balanced later.",
- _app->get_logname());
+ LOG_INFO_F("{}: stop the copy due to the primary will be balanced later.",
+ _app->get_logname());
return false;
}
return true;
diff --git a/src/meta/meta_backup_service.cpp b/src/meta/meta_backup_service.cpp
index 15ab9a4a4..4113d9758 100644
--- a/src/meta/meta_backup_service.cpp
+++ b/src/meta/meta_backup_service.cpp
@@ -129,9 +129,9 @@ void policy_context::start_backup_app_meta_unlocked(int32_t app_id)
} else if (resp.err == ERR_FS_INTERNAL) {
zauto_lock l(_lock);
_is_backup_failed = true;
- derror_f("write {} failed, err = {}, don't try again when got this error.",
- remote_file->file_name(),
- resp.err.to_string());
+ LOG_ERROR_F("write {} failed, err = {}, don't try again when got this error.",
+ remote_file->file_name(),
+ resp.err.to_string());
return;
} else {
LOG_WARNING("write %s failed, reason(%s), try it later",
@@ -245,9 +245,9 @@ void policy_context::write_backup_app_finish_flag_unlocked(int32_t app_id,
} else if (resp.err == ERR_FS_INTERNAL) {
zauto_lock l(_lock);
_is_backup_failed = true;
- derror_f("write {} failed, err = {}, don't try again when got this error.",
- remote_file->file_name(),
- resp.err.to_string());
+ LOG_ERROR_F("write {} failed, err = {}, don't try again when got this error.",
+ remote_file->file_name(),
+ resp.err.to_string());
return;
} else {
LOG_WARNING("write %s failed, reason(%s), try it later",
@@ -353,9 +353,9 @@ void policy_context::write_backup_info_unlocked(const backup_info &b_info,
} else if (resp.err == ERR_FS_INTERNAL) {
zauto_lock l(_lock);
_is_backup_failed = true;
- derror_f("write {} failed, err = {}, don't try again when got this error.",
- remote_file->file_name(),
- resp.err.to_string());
+ LOG_ERROR_F("write {} failed, err = {}, don't try again when got this error.",
+ remote_file->file_name(),
+ resp.err.to_string());
return;
} else {
LOG_WARNING("write %s failed, reason(%s), try it later",
@@ -379,32 +379,33 @@ bool policy_context::update_partition_progress_unlocked(gpid pid,
{
int32_t &local_progress = _progress.partition_progress[pid];
if (local_progress == cold_backup_constant::PROGRESS_FINISHED) {
- dwarn_f("{}: backup of partition {} has been finished, ignore the backup response from {} ",
- _backup_sig,
- pid.to_string(),
- source.to_string());
+ LOG_WARNING_F(
+ "{}: backup of partition {} has been finished, ignore the backup response from {} ",
+ _backup_sig,
+ pid.to_string(),
+ source.to_string());
return true;
}
if (progress < local_progress) {
- dwarn_f("{}: local backup progress {} is larger than progress {} from server {} for "
- "partition {}, perhaps it's primary has changed",
- _backup_sig,
- local_progress,
- progress,
- source.to_string(),
- pid.to_string());
+ LOG_WARNING_F("{}: local backup progress {} is larger than progress {} from server {} for "
+ "partition {}, perhaps it's primary has changed",
+ _backup_sig,
+ local_progress,
+ progress,
+ source.to_string(),
+ pid.to_string());
}
local_progress = progress;
- dinfo_f(
+ LOG_DEBUG_F(
"{}: update partition {} backup progress to {}.", _backup_sig, pid.to_string(), progress);
if (local_progress == cold_backup_constant::PROGRESS_FINISHED) {
- ddebug_f("{}: finish backup for partition {}, the app has {} unfinished backup "
- "partition now.",
- _backup_sig,
- pid.to_string(),
- _progress.unfinished_partitions_per_app[pid.get_app_id()]);
+ LOG_INFO_F("{}: finish backup for partition {}, the app has {} unfinished backup "
+ "partition now.",
+ _backup_sig,
+ pid.to_string(),
+ _progress.unfinished_partitions_per_app[pid.get_app_id()]);
// update the progress-chain: partition => app => current_backup_instance
if (--_progress.unfinished_partitions_per_app[pid.get_app_id()] == 0) {
@@ -434,7 +435,7 @@ void policy_context::start_backup_partition_unlocked(gpid pid)
const app_state *app = _backup_service->get_state()->get_app(pid.get_app_id()).get();
if (app == nullptr || app->status == app_status::AS_DROPPED) {
- dwarn_f(
+ LOG_WARNING_F(
"{}: app {} is not available, skip to backup it.", _backup_sig, pid.get_app_id());
_progress.is_app_skipped[pid.get_app_id()] = true;
update_partition_progress_unlocked(
@@ -444,9 +445,9 @@ void policy_context::start_backup_partition_unlocked(gpid pid)
partition_primary = app->partitions[pid.get_partition_index()].primary;
}
if (partition_primary.is_invalid()) {
- dwarn_f("{}: partition {} doesn't have a primary now, retry to backup it later",
- _backup_sig,
- pid.to_string());
+ LOG_WARNING_F("{}: partition {} doesn't have a primary now, retry to backup it later",
+ _backup_sig,
+ pid.to_string());
tasking::enqueue(LPC_DEFAULT_CALLBACK,
&_tracker,
[this, pid]() {
@@ -472,10 +473,10 @@ void policy_context::start_backup_partition_unlocked(gpid pid)
[this, pid, partition_primary](error_code err, backup_response &&response) {
on_backup_reply(err, std::move(response), pid, partition_primary);
});
- ddebug_f("{}: send backup command to partition {}, target_addr = {}",
- _backup_sig,
- pid.to_string(),
- partition_primary.to_string());
+ LOG_INFO_F("{}: send backup command to partition {}, target_addr = {}",
+ _backup_sig,
+ pid.to_string(),
+ partition_primary.to_string());
_backup_service->get_meta_service()->send_request(request, partition_primary, rpc_callback);
}
@@ -484,10 +485,10 @@ void policy_context::on_backup_reply(error_code err,
gpid pid,
const rpc_address &primary)
{
- ddebug_f("{}: receive backup response for partition {} from server {}.",
- _backup_sig,
- pid.to_string(),
- primary.to_string());
+ LOG_INFO_F("{}: receive backup response for partition {} from server {}.",
+ _backup_sig,
+ pid.to_string(),
+ primary.to_string());
if (err == dsn::ERR_OK && response.err == dsn::ERR_OK) {
dassert(response.policy_name == _policy.policy_name,
"policy name(%s vs %s) don't match, pid(%d.%d), replica_server(%s)",
@@ -513,13 +514,14 @@ void policy_context::on_backup_reply(error_code err,
pid.get_partition_index());
if (response.backup_id < _cur_backup.backup_id) {
- dwarn_f("{}: got a backup response of partition {} from server {}, whose backup id "
- "{} is smaller than current backup id {}, maybe it is a stale message",
- _backup_sig,
- pid.to_string(),
- primary.to_string(),
- response.backup_id,
- _cur_backup.backup_id);
+ LOG_WARNING_F(
+ "{}: got a backup response of partition {} from server {}, whose backup id "
+ "{} is smaller than current backup id {}, maybe it is a stale message",
+ _backup_sig,
+ pid.to_string(),
+ primary.to_string(),
+ response.backup_id,
+ _cur_backup.backup_id);
} else {
zauto_lock l(_lock);
record_partition_checkpoint_size_unlock(pid, response.checkpoint_total_size);
@@ -531,20 +533,21 @@ void policy_context::on_backup_reply(error_code err,
} else if (response.err == dsn::ERR_LOCAL_APP_FAILURE) {
zauto_lock l(_lock);
_is_backup_failed = true;
- derror_f("{}: backup got error {} for partition {} from {}, don't try again when got "
- "this error.",
- _backup_sig.c_str(),
- response.err.to_string(),
- pid.to_string(),
- primary.to_string());
+ LOG_ERROR_F("{}: backup got error {} for partition {} from {}, don't try again when got "
+ "this error.",
+ _backup_sig.c_str(),
+ response.err.to_string(),
+ pid.to_string(),
+ primary.to_string());
return;
} else {
- dwarn_f("{}: backup got error for partition {} from {}, rpc error {}, response error {}",
- _backup_sig.c_str(),
- pid.to_string(),
- primary.to_string(),
- err.to_string(),
- response.err.to_string());
+ LOG_WARNING_F(
+ "{}: backup got error for partition {} from {}, rpc error {}, response error {}",
+ _backup_sig.c_str(),
+ pid.to_string(),
+ primary.to_string(),
+ err.to_string(),
+ response.err.to_string());
}
// retry to backup the partition.
@@ -660,8 +663,8 @@ void policy_context::sync_backup_to_remote_storage_unlocked(const backup_info &b
void policy_context::continue_current_backup_unlocked()
{
if (_policy.is_disable) {
- ddebug_f("{}: policy is disabled, ignore this backup and try it later",
- _policy.policy_name);
+ LOG_INFO_F("{}: policy is disabled, ignore this backup and try it later",
+ _policy.policy_name);
tasking::enqueue(LPC_DEFAULT_CALLBACK,
&_tracker,
[this]() {
@@ -1262,9 +1265,9 @@ void backup_service::add_backup_policy(dsn::message_ex *msg)
for (auto &app_id : request.app_ids) {
const std::shared_ptr<app_state> &app = _state->get_app(app_id);
if (app == nullptr) {
- derror_f("app {} doesn't exist, policy {} shouldn't be added.",
- app_id,
- request.policy_name);
+ LOG_ERROR_F("app {} doesn't exist, policy {} shouldn't be added.",
+ app_id,
+ request.policy_name);
response.err = ERR_INVALID_PARAMETERS;
response.hint_message = "invalid app " + std::to_string(app_id);
_meta_svc->reply_data(msg, response);
@@ -1298,7 +1301,7 @@ void backup_service::add_backup_policy(dsn::message_ex *msg)
return;
}
- ddebug_f("start to add backup polciy {}.", request.policy_name);
+ LOG_INFO_F("start to add backup polciy {}.", request.policy_name);
std::shared_ptr<policy_context> policy_context_ptr = _factory(this);
dassert(policy_context_ptr != nullptr, "invalid policy_context");
policy p;
diff --git a/src/meta/meta_bulk_load_ingestion_context.cpp b/src/meta/meta_bulk_load_ingestion_context.cpp
index bfeda7685..effe6c266 100644
--- a/src/meta/meta_bulk_load_ingestion_context.cpp
+++ b/src/meta/meta_bulk_load_ingestion_context.cpp
@@ -80,20 +80,20 @@ bool ingestion_context::node_context::check_if_add(const std::string &disk_tag)
{
auto max_node_ingestion_count = FLAGS_bulk_load_node_max_ingesting_count;
if (node_ingesting_count >= max_node_ingestion_count) {
- dwarn_f("node[{}] has {} partition executing ingestion, max_count = {}",
- address.to_string(),
- node_ingesting_count,
- max_node_ingestion_count);
+ LOG_WARNING_F("node[{}] has {} partition executing ingestion, max_count = {}",
+ address.to_string(),
+ node_ingesting_count,
+ max_node_ingestion_count);
return false;
}
auto max_disk_ingestion_count = get_max_disk_ingestion_count(max_node_ingestion_count);
if (disk_ingesting_counts[disk_tag] >= max_disk_ingestion_count) {
- dwarn_f("node[{}] disk[{}] has {} partition executing ingestion, max_count = {}",
- address.to_string(),
- disk_tag,
- disk_ingesting_counts[disk_tag],
- max_disk_ingestion_count);
+ LOG_WARNING_F("node[{}] disk[{}] has {} partition executing ingestion, max_count = {}",
+ address.to_string(),
+ disk_tag,
+ disk_ingesting_counts[disk_tag],
+ max_disk_ingestion_count);
return false;
}
return true;
diff --git a/src/meta/meta_bulk_load_service.cpp b/src/meta/meta_bulk_load_service.cpp
index 9ae19679e..6dff9c0c6 100644
--- a/src/meta/meta_bulk_load_service.cpp
+++ b/src/meta/meta_bulk_load_service.cpp
@@ -77,7 +77,7 @@ void bulk_load_service::on_start_bulk_load(start_bulk_load_rpc rpc)
if (!FLAGS_enable_concurrent_bulk_load &&
!_meta_svc->try_lock_meta_op_status(meta_op_status::BULKLOAD)) {
response.hint_msg = "meta server is busy now, please wait";
- derror_f("{}", response.hint_msg);
+ LOG_ERROR_F("{}", response.hint_msg);
response.err = ERR_BUSY;
return;
}
@@ -87,14 +87,14 @@ void bulk_load_service::on_start_bulk_load(start_bulk_load_rpc rpc)
response.err = app == nullptr ? ERR_APP_NOT_EXIST : ERR_APP_DROPPED;
response.hint_msg = fmt::format(
"app {} is ", response.err == ERR_APP_NOT_EXIST ? "not existed" : "not available");
- derror_f("{}", response.hint_msg);
+ LOG_ERROR_F("{}", response.hint_msg);
_meta_svc->unlock_meta_op_status();
return;
}
if (app->is_bulk_loading) {
response.err = ERR_BUSY;
response.hint_msg = fmt::format("app({}) is already executing bulk load", app->app_name);
- derror_f("{}", response.hint_msg);
+ LOG_ERROR_F("{}", response.hint_msg);
_meta_svc->unlock_meta_op_status();
return;
}
@@ -109,13 +109,13 @@ void bulk_load_service::on_start_bulk_load(start_bulk_load_rpc rpc)
return;
}
- ddebug_f("app({}) start bulk load, cluster_name = {}, provider = {}, remote root path = {}, "
- "ingest_behind = {}",
- request.app_name,
- request.cluster_name,
- request.file_provider_type,
- request.remote_root_path,
- request.ingest_behind);
+ LOG_INFO_F("app({}) start bulk load, cluster_name = {}, provider = {}, remote root path = {}, "
+ "ingest_behind = {}",
+ request.app_name,
+ request.cluster_name,
+ request.file_provider_type,
+ request.remote_root_path,
+ request.ingest_behind);
// clear old bulk load result
reset_local_bulk_load_states(app->app_id, app->app_name, true);
@@ -141,7 +141,7 @@ bulk_load_service::check_bulk_load_request_params(const start_bulk_load_request
if (!validate_ingest_behind(envs, request.ingest_behind)) {
hint_msg = fmt::format("inconsistent ingestion behind option");
- derror_f("{}", hint_msg);
+ LOG_ERROR_F("{}", hint_msg);
return ERR_INCONSISTENT_STATE;
}
@@ -150,7 +150,7 @@ bulk_load_service::check_bulk_load_request_params(const start_bulk_load_request
dsn::dist::block_service::block_filesystem *blk_fs =
_meta_svc->get_block_service_manager().get_or_create_block_filesystem(file_provider);
if (blk_fs == nullptr) {
- derror_f("invalid remote file provider type: {}", file_provider);
+ LOG_ERROR_F("invalid remote file provider type: {}", file_provider);
hint_msg = "invalid file_provider";
return ERR_INVALID_PARAMETERS;
}
@@ -173,7 +173,7 @@ bulk_load_service::check_bulk_load_request_params(const start_bulk_load_request
})
->wait();
if (err != ERR_OK || file_handler == nullptr) {
- derror_f(
+ LOG_ERROR_F(
"failed to get file({}) handler on remote provider({})", remote_path, file_provider);
hint_msg = "file_provider error";
return ERR_FILE_OPERATION_FAILED;
@@ -187,29 +187,29 @@ bulk_load_service::check_bulk_load_request_params(const start_bulk_load_request
[&r_resp](const dsn::dist::block_service::read_response &resp) { r_resp = resp; })
->wait();
if (r_resp.err != ERR_OK) {
- derror_f("failed to read file({}) on remote provider({}), error = {}",
- remote_path,
- file_provider,
- r_resp.err.to_string());
+ LOG_ERROR_F("failed to read file({}) on remote provider({}), error = {}",
+ remote_path,
+ file_provider,
+ r_resp.err.to_string());
hint_msg = "read bulk_load_info failed";
return r_resp.err;
}
bulk_load_info bl_info;
if (!::dsn::json::json_forwarder<bulk_load_info>::decode(r_resp.buffer, bl_info)) {
- derror_f("file({}) is damaged on remote file provider({})", remote_path, file_provider);
+ LOG_ERROR_F("file({}) is damaged on remote file provider({})", remote_path, file_provider);
hint_msg = "bulk_load_info damaged";
return ERR_CORRUPTION;
}
if (bl_info.app_id != app_id || bl_info.partition_count != partition_count) {
- derror_f("app({}) information is inconsistent, local app_id({}) VS remote app_id({}), "
- "local partition_count({}) VS remote partition_count({})",
- request.app_name,
- app_id,
- bl_info.app_id,
- partition_count,
- bl_info.partition_count);
+ LOG_ERROR_F("app({}) information is inconsistent, local app_id({}) VS remote app_id({}), "
+ "local partition_count({}) VS remote partition_count({})",
+ request.app_name,
+ app_id,
+ bl_info.app_id,
+ partition_count,
+ bl_info.partition_count);
hint_msg = "app_id or partition_count is inconsistent";
return ERR_INCONSISTENT_STATE;
}
@@ -264,12 +264,12 @@ void bulk_load_service::create_app_bulk_load_dir(const std::string &app_name,
_meta_svc->get_meta_storage()->delete_node_recursively(
get_app_bulk_load_path(app_id), [this, rpc, ainfo]() {
std::string bulk_load_path = get_app_bulk_load_path(ainfo.app_id);
- ddebug_f("remove app({}) bulk load dir {} succeed", ainfo.app_name, bulk_load_path);
+ LOG_INFO_F("remove app({}) bulk load dir {} succeed", ainfo.app_name, bulk_load_path);
blob value = dsn::json::json_forwarder<app_bulk_load_info>::encode(ainfo);
_meta_svc->get_meta_storage()->create_node(
std::move(bulk_load_path), std::move(value), [this, rpc, ainfo]() {
- dinfo_f("create app({}) bulk load dir", ainfo.app_name);
+ LOG_DEBUG_F("create app({}) bulk load dir", ainfo.app_name);
{
zauto_write_lock l(_lock);
_app_bulk_load_info[ainfo.app_id] = ainfo;
@@ -301,13 +301,13 @@ void bulk_load_service::create_partition_bulk_load_dir(const std::string &app_na
get_partition_bulk_load_path(pid),
std::move(value),
[app_name, pid, partition_count, rpc, pinfo, this]() {
- dinfo_f("app({}) create partition({}) bulk_load_info", app_name, pid.to_string());
+ LOG_DEBUG_F("app({}) create partition({}) bulk_load_info", app_name, pid.to_string());
{
zauto_write_lock l(_lock);
_partition_bulk_load_info[pid] = pinfo;
_partitions_pending_sync_flag[pid] = false;
if (--_apps_in_progress_count[pid.get_app_id()] == 0) {
- ddebug_f("app({}) start bulk load succeed", app_name);
+ LOG_INFO_F("app({}) start bulk load succeed", app_name);
_apps_in_progress_count[pid.get_app_id()] = partition_count;
rpc.response().err = ERR_OK;
}
@@ -327,7 +327,7 @@ bool bulk_load_service::check_partition_status(
{
std::shared_ptr<app_state> app = get_app(pid.get_app_id());
if (app == nullptr || app->status != app_status::AS_AVAILABLE) {
- dwarn_f(
+ LOG_WARNING_F(
"app(name={}, id={}, status={}) is not existed or not available, set bulk load failed",
app_name,
pid.get_app_id(),
@@ -338,7 +338,7 @@ bool bulk_load_service::check_partition_status(
pconfig = app->partitions[pid.get_partition_index()];
if (pconfig.primary.is_invalid()) {
- dwarn_f("app({}) partition({}) primary is invalid, try it later", app_name, pid);
+ LOG_WARNING_F("app({}) partition({}) primary is invalid, try it later", app_name, pid);
tasking::enqueue(LPC_META_STATE_NORMAL,
_meta_svc->tracker(),
[retry_function, app_name, pid]() { retry_function(app_name, pid); },
@@ -361,10 +361,10 @@ bool bulk_load_service::check_partition_status(
p_status == bulk_load_status::BLS_FAILED)) {
return true;
}
- dwarn_f("app({}) partition({}) is unhealthy, status({}), try it later",
- app_name,
- pid,
- dsn::enum_to_string(p_status));
+ LOG_WARNING_F("app({}) partition({}) is unhealthy, status({}), try it later",
+ app_name,
+ pid,
+ dsn::enum_to_string(p_status));
tasking::enqueue(LPC_META_STATE_NORMAL,
_meta_svc->tracker(),
[retry_function, app_name, pid]() { retry_function(app_name, pid); },
@@ -408,15 +408,15 @@ void bulk_load_service::partition_bulk_load(const std::string &app_name, const g
req->remote_root_path = ainfo.remote_root_path;
}
- ddebug_f("send bulk load request to node({}), app({}), partition({}), partition "
- "status = {}, remote provider = {}, cluster_name = {}, remote_root_path = {}",
- primary_addr.to_string(),
- app_name,
- pid,
- dsn::enum_to_string(req->meta_bulk_load_status),
- req->remote_provider_name,
- req->cluster_name,
- req->remote_root_path);
+ LOG_INFO_F("send bulk load request to node({}), app({}), partition({}), partition "
+ "status = {}, remote provider = {}, cluster_name = {}, remote_root_path = {}",
+ primary_addr.to_string(),
+ app_name,
+ pid,
+ dsn::enum_to_string(req->meta_bulk_load_status),
+ req->remote_provider_name,
+ req->cluster_name,
+ req->remote_root_path);
bulk_load_rpc rpc(std::move(req), RPC_BULK_LOAD, 0_ms, 0, pid.thread_hash());
rpc.call(primary_addr, _meta_svc->tracker(), [this, rpc](error_code err) mutable {
@@ -434,7 +434,7 @@ void bulk_load_service::on_partition_bulk_load_reply(error_code err,
const rpc_address &primary_addr = request.primary_addr;
if (err != ERR_OK) {
- derror_f(
+ LOG_ERROR_F(
"app({}), partition({}) failed to receive bulk load response from node({}), error = {}",
app_name,
pid,
@@ -446,7 +446,7 @@ void bulk_load_service::on_partition_bulk_load_reply(error_code err,
}
if (response.err == ERR_OBJECT_NOT_FOUND || response.err == ERR_INVALID_STATE) {
- derror_f(
+ LOG_ERROR_F(
"app({}), partition({}) doesn't exist or has invalid state on node({}), error = {}",
app_name,
pid,
@@ -458,23 +458,25 @@ void bulk_load_service::on_partition_bulk_load_reply(error_code err,
}
if (response.err == ERR_BUSY) {
- dwarn_f("node({}) has enough replicas downloading, wait for next round to send bulk load "
- "request for app({}), partition({})",
- primary_addr.to_string(),
- app_name,
- pid);
+ LOG_WARNING_F(
+ "node({}) has enough replicas downloading, wait for next round to send bulk load "
+ "request for app({}), partition({})",
+ primary_addr.to_string(),
+ app_name,
+ pid);
try_resend_bulk_load_request(app_name, pid);
return;
}
if (response.err != ERR_OK) {
- derror_f("app({}), partition({}) from node({}) handle bulk load response failed, error = "
- "{}, primary status = {}",
- app_name,
- pid,
- primary_addr.to_string(),
- response.err.to_string(),
- dsn::enum_to_string(response.primary_bulk_load_status));
+ LOG_ERROR_F(
+ "app({}), partition({}) from node({}) handle bulk load response failed, error = "
+ "{}, primary status = {}",
+ app_name,
+ pid,
+ primary_addr.to_string(),
+ response.err.to_string(),
+ dsn::enum_to_string(response.primary_bulk_load_status));
handle_bulk_load_failed(pid.get_app_id(), response.err);
try_resend_bulk_load_request(app_name, pid);
return;
@@ -483,20 +485,21 @@ void bulk_load_service::on_partition_bulk_load_reply(error_code err,
// response.err = ERR_OK
std::shared_ptr<app_state> app = get_app(pid.get_app_id());
if (app == nullptr || app->status != app_status::AS_AVAILABLE) {
- dwarn_f(
+ LOG_WARNING_F(
"app(name={}, id={}) is not existed, set bulk load failed", app_name, pid.get_app_id());
handle_app_unavailable(pid.get_app_id(), app_name);
return;
}
ballot current_ballot = app->partitions[pid.get_partition_index()].ballot;
if (request.ballot < current_ballot) {
- dwarn_f("receive out-date response from node({}), app({}), partition({}), request ballot = "
- "{}, current ballot= {}",
- primary_addr.to_string(),
- app_name,
- pid,
- request.ballot,
- current_ballot);
+ LOG_WARNING_F(
+ "receive out-date response from node({}), app({}), partition({}), request ballot = "
+ "{}, current ballot= {}",
+ primary_addr.to_string(),
+ app_name,
+ pid,
+ request.ballot,
+ current_ballot);
try_rollback_to_downloading(app_name, pid);
try_resend_bulk_load_request(app_name, pid);
return;
@@ -557,7 +560,7 @@ void bulk_load_service::handle_app_downloading(const bulk_load_response &respons
const gpid &pid = response.pid;
if (!response.__isset.total_download_progress) {
- dwarn_f(
+ LOG_WARNING_F(
"receive bulk load response from node({}) app({}), partition({}), primary_status({}), "
"but total_download_progress is not set",
primary_addr.to_string(),
@@ -571,23 +574,23 @@ void bulk_load_service::handle_app_downloading(const bulk_load_response &respons
const auto &bulk_load_states = kv.second;
if (!bulk_load_states.__isset.download_progress ||
!bulk_load_states.__isset.download_status) {
- dwarn_f("receive bulk load response from node({}) app({}), partition({}), "
- "primary_status({}), but node({}) progress or status is not set",
- primary_addr.to_string(),
- app_name,
- pid,
- dsn::enum_to_string(response.primary_bulk_load_status),
- kv.first.to_string());
+ LOG_WARNING_F("receive bulk load response from node({}) app({}), partition({}), "
+ "primary_status({}), but node({}) progress or status is not set",
+ primary_addr.to_string(),
+ app_name,
+ pid,
+ dsn::enum_to_string(response.primary_bulk_load_status),
+ kv.first.to_string());
return;
}
// check partition download status
if (bulk_load_states.download_status != ERR_OK) {
- derror_f("app({}) partition({}) on node({}) meet unrecoverable error during "
- "downloading files, error = {}",
- app_name,
- pid,
- kv.first.to_string(),
- bulk_load_states.download_status);
+ LOG_ERROR_F("app({}) partition({}) on node({}) meet unrecoverable error during "
+ "downloading files, error = {}",
+ app_name,
+ pid,
+ kv.first.to_string(),
+ bulk_load_states.download_status);
error_code err = ERR_UNKNOWN;
// ERR_FILE_OPERATION_FAILED: local file system error
@@ -610,13 +613,14 @@ void bulk_load_service::handle_app_downloading(const bulk_load_response &respons
// update download progress
int32_t total_progress = response.total_download_progress;
- ddebug_f("receive bulk load response from node({}) app({}) partition({}), primary_status({}), "
- "total_download_progress = {}",
- primary_addr.to_string(),
- app_name,
- pid,
- dsn::enum_to_string(response.primary_bulk_load_status),
- total_progress);
+ LOG_INFO_F(
+ "receive bulk load response from node({}) app({}) partition({}), primary_status({}), "
+ "total_download_progress = {}",
+ primary_addr.to_string(),
+ app_name,
+ pid,
+ dsn::enum_to_string(response.primary_bulk_load_status),
+ total_progress);
{
zauto_write_lock l(_lock);
_partitions_total_download_progress[pid] = total_progress;
@@ -625,7 +629,7 @@ void bulk_load_service::handle_app_downloading(const bulk_load_response &respons
// update partition status to `downloaded` if all replica downloaded
if (total_progress >= bulk_load_constant::PROGRESS_FINISHED) {
- ddebug_f(
+ LOG_INFO_F(
"app({}) partirion({}) download all files from remote provider succeed", app_name, pid);
update_partition_info_on_remote_storage(app_name, pid, bulk_load_status::BLS_DOWNLOADED);
}
@@ -639,53 +643,54 @@ void bulk_load_service::handle_app_ingestion(const bulk_load_response &response,
const gpid &pid = response.pid;
if (!response.__isset.is_group_ingestion_finished) {
- dwarn_f("receive bulk load response from node({}) app({}) partition({}), "
- "primary_status({}), but is_group_ingestion_finished is not set",
- primary_addr.to_string(),
- app_name,
- pid,
- dsn::enum_to_string(response.primary_bulk_load_status));
+ LOG_WARNING_F("receive bulk load response from node({}) app({}) partition({}), "
+ "primary_status({}), but is_group_ingestion_finished is not set",
+ primary_addr.to_string(),
+ app_name,
+ pid,
+ dsn::enum_to_string(response.primary_bulk_load_status));
return;
}
for (const auto &kv : response.group_bulk_load_state) {
const auto &bulk_load_states = kv.second;
if (!bulk_load_states.__isset.ingest_status) {
- dwarn_f("receive bulk load response from node({}) app({}) partition({}), "
- "primary_status({}), but node({}) ingestion_status is not set",
- primary_addr.to_string(),
- app_name,
- pid,
- dsn::enum_to_string(response.primary_bulk_load_status),
- kv.first.to_string());
+ LOG_WARNING_F("receive bulk load response from node({}) app({}) partition({}), "
+ "primary_status({}), but node({}) ingestion_status is not set",
+ primary_addr.to_string(),
+ app_name,
+ pid,
+ dsn::enum_to_string(response.primary_bulk_load_status),
+ kv.first.to_string());
return;
}
if (bulk_load_states.ingest_status == ingestion_status::IS_FAILED) {
- derror_f("app({}) partition({}) on node({}) ingestion failed",
- app_name,
- pid,
- kv.first.to_string());
+ LOG_ERROR_F("app({}) partition({}) on node({}) ingestion failed",
+ app_name,
+ pid,
+ kv.first.to_string());
finish_ingestion(pid);
handle_bulk_load_failed(pid.get_app_id(), ERR_INGESTION_FAILED);
return;
}
}
- ddebug_f("receive bulk load response from node({}) app({}) partition({}), primary_status({}), "
- "is_group_ingestion_finished = {}",
- primary_addr.to_string(),
- app_name,
- pid,
- dsn::enum_to_string(response.primary_bulk_load_status),
- response.is_group_ingestion_finished);
+ LOG_INFO_F(
+ "receive bulk load response from node({}) app({}) partition({}), primary_status({}), "
+ "is_group_ingestion_finished = {}",
+ primary_addr.to_string(),
+ app_name,
+ pid,
+ dsn::enum_to_string(response.primary_bulk_load_status),
+ response.is_group_ingestion_finished);
{
zauto_write_lock l(_lock);
_partitions_bulk_load_state[pid] = response.group_bulk_load_state;
}
if (response.is_group_ingestion_finished) {
- ddebug_f("app({}) partition({}) ingestion files succeed", app_name, pid);
+ LOG_INFO_F("app({}) partition({}) ingestion files succeed", app_name, pid);
finish_ingestion(pid);
update_partition_info_on_remote_storage(app_name, pid, bulk_load_status::BLS_SUCCEED);
}
@@ -699,24 +704,24 @@ void bulk_load_service::handle_bulk_load_finish(const bulk_load_response &respon
const gpid &pid = response.pid;
if (!response.__isset.is_group_bulk_load_context_cleaned_up) {
- dwarn_f("receive bulk load response from node({}) app({}) partition({}), "
- "primary_status({}), but is_group_bulk_load_context_cleaned_up is not set",
- primary_addr.to_string(),
- app_name,
- pid,
- dsn::enum_to_string(response.primary_bulk_load_status));
+ LOG_WARNING_F("receive bulk load response from node({}) app({}) partition({}), "
+ "primary_status({}), but is_group_bulk_load_context_cleaned_up is not set",
+ primary_addr.to_string(),
+ app_name,
+ pid,
+ dsn::enum_to_string(response.primary_bulk_load_status));
return;
}
for (const auto &kv : response.group_bulk_load_state) {
if (!kv.second.__isset.is_cleaned_up) {
- dwarn_f("receive bulk load response from node({}) app({}), partition({}), "
- "primary_status({}), but node({}) is_cleaned_up is not set",
- primary_addr.to_string(),
- app_name,
- pid,
- dsn::enum_to_string(response.primary_bulk_load_status),
- kv.first.to_string());
+ LOG_WARNING_F("receive bulk load response from node({}) app({}), partition({}), "
+ "primary_status({}), but node({}) is_cleaned_up is not set",
+ primary_addr.to_string(),
+ app_name,
+ pid,
+ dsn::enum_to_string(response.primary_bulk_load_status),
+ kv.first.to_string());
return;
}
}
@@ -724,7 +729,7 @@ void bulk_load_service::handle_bulk_load_finish(const bulk_load_response &respon
{
zauto_read_lock l(_lock);
if (_partitions_cleaned_up[pid]) {
- dwarn_f(
+ LOG_WARNING_F(
"receive bulk load response from node({}) app({}) partition({}), current partition "
"has already been cleaned up",
primary_addr.to_string(),
@@ -736,13 +741,14 @@ void bulk_load_service::handle_bulk_load_finish(const bulk_load_response &respon
// The replicas have cleaned up their bulk load states and removed temporary sst files
bool group_cleaned_up = response.is_group_bulk_load_context_cleaned_up;
- ddebug_f("receive bulk load response from node({}) app({}) partition({}), primary status = {}, "
- "is_group_bulk_load_context_cleaned_up = {}",
- primary_addr.to_string(),
- app_name,
- pid,
- dsn::enum_to_string(response.primary_bulk_load_status),
- group_cleaned_up);
+ LOG_INFO_F(
+ "receive bulk load response from node({}) app({}) partition({}), primary status = {}, "
+ "is_group_bulk_load_context_cleaned_up = {}",
+ primary_addr.to_string(),
+ app_name,
+ pid,
+ dsn::enum_to_string(response.primary_bulk_load_status),
+ group_cleaned_up);
{
zauto_write_lock l(_lock);
_partitions_cleaned_up[pid] = group_cleaned_up;
@@ -758,14 +764,14 @@ void bulk_load_service::handle_bulk_load_finish(const bulk_load_response &respon
if (count == 0) {
std::shared_ptr<app_state> app = get_app(pid.get_app_id());
if (app == nullptr || app->status != app_status::AS_AVAILABLE) {
- dwarn_f("app(name={}, id={}) is not existed, remove bulk load dir on remote "
- "storage",
- app_name,
- pid.get_app_id());
+ LOG_WARNING_F("app(name={}, id={}) is not existed, remove bulk load dir on remote "
+ "storage",
+ app_name,
+ pid.get_app_id());
remove_bulk_load_dir_on_remote_storage(pid.get_app_id(), app_name);
return;
}
- ddebug_f("app({}) update app to not bulk loading", app_name);
+ LOG_INFO_F("app({}) update app to not bulk loading", app_name);
update_app_not_bulk_loading_on_remote_storage(std::move(app));
reset_local_bulk_load_states(pid.get_app_id(), app_name, false);
}
@@ -780,43 +786,44 @@ void bulk_load_service::handle_app_pausing(const bulk_load_response &response,
const gpid &pid = response.pid;
if (!response.__isset.is_group_bulk_load_paused) {
- dwarn_f("receive bulk load response from node({}) app({}) partition({}), "
- "primary_status({}), but is_group_bulk_load_paused is not set",
- primary_addr.to_string(),
- app_name,
- pid,
- dsn::enum_to_string(response.primary_bulk_load_status));
+ LOG_WARNING_F("receive bulk load response from node({}) app({}) partition({}), "
+ "primary_status({}), but is_group_bulk_load_paused is not set",
+ primary_addr.to_string(),
+ app_name,
+ pid,
+ dsn::enum_to_string(response.primary_bulk_load_status));
return;
}
for (const auto &kv : response.group_bulk_load_state) {
if (!kv.second.__isset.is_paused) {
- dwarn_f("receive bulk load response from node({}) app({}), partition({}), "
- "primary_status({}), but node({}) is_paused is not set",
- primary_addr.to_string(),
- app_name,
- pid,
- dsn::enum_to_string(response.primary_bulk_load_status),
- kv.first.to_string());
+ LOG_WARNING_F("receive bulk load response from node({}) app({}), partition({}), "
+ "primary_status({}), but node({}) is_paused is not set",
+ primary_addr.to_string(),
+ app_name,
+ pid,
+ dsn::enum_to_string(response.primary_bulk_load_status),
+ kv.first.to_string());
return;
}
}
bool is_group_paused = response.is_group_bulk_load_paused;
- ddebug_f("receive bulk load response from node({}) app({}) partition({}), primary status = {}, "
- "is_group_bulk_load_paused = {}",
- primary_addr.to_string(),
- app_name,
- pid,
- dsn::enum_to_string(response.primary_bulk_load_status),
- is_group_paused);
+ LOG_INFO_F(
+ "receive bulk load response from node({}) app({}) partition({}), primary status = {}, "
+ "is_group_bulk_load_paused = {}",
+ primary_addr.to_string(),
+ app_name,
+ pid,
+ dsn::enum_to_string(response.primary_bulk_load_status),
+ is_group_paused);
{
zauto_write_lock l(_lock);
_partitions_bulk_load_state[pid] = response.group_bulk_load_state;
}
if (is_group_paused) {
- ddebug_f("app({}) partirion({}) pause bulk load succeed", response.app_name, pid);
+ LOG_INFO_F("app({}) partirion({}) pause bulk load succeed", response.app_name, pid);
update_partition_info_on_remote_storage(
response.app_name, pid, bulk_load_status::BLS_PAUSED);
}
@@ -831,18 +838,18 @@ void bulk_load_service::try_rollback_to_downloading(const std::string &app_name,
if (app_status != bulk_load_status::BLS_DOWNLOADING &&
app_status != bulk_load_status::BLS_DOWNLOADED &&
app_status != bulk_load_status::BLS_INGESTING) {
- ddebug_f("app({}) status={}, no need to rollback to downloading, wait for next round",
- app_name,
- dsn::enum_to_string(app_status));
+ LOG_INFO_F("app({}) status={}, no need to rollback to downloading, wait for next round",
+ app_name,
+ dsn::enum_to_string(app_status));
return;
}
if (_apps_rolling_back[pid.get_app_id()]) {
- dwarn_f("app({}) is rolling back to downloading, ignore this request", app_name);
+ LOG_WARNING_F("app({}) is rolling back to downloading, ignore this request", app_name);
return;
}
if (_apps_rollback_count[pid.get_app_id()] >= FLAGS_bulk_load_max_rollback_times) {
- dwarn_f(
+ LOG_WARNING_F(
"app({}) has been rollback to downloading for {} times, make bulk load process failed",
app_name,
_apps_rollback_count[pid.get_app_id()]);
@@ -854,11 +861,11 @@ void bulk_load_service::try_rollback_to_downloading(const std::string &app_name,
: ERR_RETRY_EXHAUSTED);
return;
}
- ddebug_f("app({}) will rolling back from {} to {}, current rollback_count = {}",
- app_name,
- dsn::enum_to_string(app_status),
- dsn::enum_to_string(bulk_load_status::BLS_DOWNLOADING),
- _apps_rollback_count[pid.get_app_id()]);
+ LOG_INFO_F("app({}) will rolling back from {} to {}, current rollback_count = {}",
+ app_name,
+ dsn::enum_to_string(app_status),
+ dsn::enum_to_string(bulk_load_status::BLS_DOWNLOADING),
+ _apps_rollback_count[pid.get_app_id()]);
_apps_rolling_back[pid.get_app_id()] = true;
_apps_rollback_count[pid.get_app_id()]++;
update_app_status_on_remote_storage_unlocked(pid.get_app_id(),
@@ -898,7 +905,7 @@ void bulk_load_service::update_partition_metadata_on_remote_storage(
get_partition_bulk_load_path(pid), std::move(value), [this, app_name, pid, pinfo]() {
zauto_write_lock l(_lock);
_partition_bulk_load_info[pid] = pinfo;
- ddebug_f(
+ LOG_INFO_F(
"app({}) update partition({}) bulk load metadata, file count = {}, file size = {}",
app_name,
pid,
@@ -916,23 +923,24 @@ void bulk_load_service::update_partition_info_on_remote_storage(const std::strin
zauto_write_lock l(_lock);
partition_bulk_load_info pinfo = _partition_bulk_load_info[pid];
if (pinfo.status == new_status && new_status != bulk_load_status::BLS_DOWNLOADING) {
- dwarn_f("app({}) partition({}) old status:{} VS new status:{}, ignore it",
- app_name,
- pid,
- dsn::enum_to_string(pinfo.status),
- dsn::enum_to_string(new_status));
+ LOG_WARNING_F("app({}) partition({}) old status:{} VS new status:{}, ignore it",
+ app_name,
+ pid,
+ dsn::enum_to_string(pinfo.status),
+ dsn::enum_to_string(new_status));
return;
}
if (_partitions_pending_sync_flag[pid]) {
if (_apps_rolling_back[pid.get_app_id()] &&
new_status == bulk_load_status::BLS_DOWNLOADING) {
- dwarn_f("app({}) partition({}) has already sync bulk load status, current_status = {}, "
- "wait and retry to set status as {}",
- app_name,
- pid,
- dsn::enum_to_string(pinfo.status),
- dsn::enum_to_string(new_status));
+ LOG_WARNING_F(
+ "app({}) partition({}) has already sync bulk load status, current_status = {}, "
+ "wait and retry to set status as {}",
+ app_name,
+ pid,
+ dsn::enum_to_string(pinfo.status),
+ dsn::enum_to_string(new_status));
tasking::enqueue(LPC_META_STATE_NORMAL,
_meta_svc->tracker(),
std::bind(&bulk_load_service::update_partition_info_on_remote_storage,
@@ -944,12 +952,12 @@ void bulk_load_service::update_partition_info_on_remote_storage(const std::strin
0,
std::chrono::seconds(1));
} else {
- ddebug_f("app({}) partition({}) has already sync bulk load status, current_status = "
- "{}, new_status = {}, wait for next round",
- app_name,
- pid,
- dsn::enum_to_string(pinfo.status),
- dsn::enum_to_string(new_status));
+ LOG_INFO_F("app({}) partition({}) has already sync bulk load status, current_status = "
+ "{}, new_status = {}, wait for next round",
+ app_name,
+ pid,
+ dsn::enum_to_string(pinfo.status),
+ dsn::enum_to_string(new_status));
}
return;
}
@@ -1004,11 +1012,11 @@ void bulk_load_service::update_partition_info_on_remote_storage_reply(
_partition_bulk_load_info[pid] = new_info;
_partitions_pending_sync_flag[pid] = false;
- ddebug_f("app({}) update partition({}) status from {} to {}",
- app_name,
- pid,
- dsn::enum_to_string(old_status),
- dsn::enum_to_string(new_status));
+ LOG_INFO_F("app({}) update partition({}) status from {} to {}",
+ app_name,
+ pid,
+ dsn::enum_to_string(old_status),
+ dsn::enum_to_string(new_status));
switch (new_status) {
case bulk_load_status::BLS_DOWNLOADED:
@@ -1029,7 +1037,7 @@ void bulk_load_service::update_partition_info_on_remote_storage_reply(
_apps_in_progress_count[pid.get_app_id()] =
_app_bulk_load_info[pid.get_app_id()].partition_count;
_apps_rolling_back[pid.get_app_id()] = false;
- ddebug_f("app({}) restart to bulk load", app_name);
+ LOG_INFO_F("app({}) restart to bulk load", app_name);
}
} break;
default:
@@ -1053,19 +1061,20 @@ void bulk_load_service::update_app_status_on_remote_storage_unlocked(
auto old_status = ainfo.status;
if (old_status == new_status && new_status != bulk_load_status::BLS_DOWNLOADING) {
- dwarn_f("app({}) old status:{} VS new status:{}, ignore it",
- ainfo.app_name,
- dsn::enum_to_string(old_status),
- dsn::enum_to_string(new_status));
+ LOG_WARNING_F("app({}) old status:{} VS new status:{}, ignore it",
+ ainfo.app_name,
+ dsn::enum_to_string(old_status),
+ dsn::enum_to_string(new_status));
return;
}
if (_apps_pending_sync_flag[app_id]) {
- ddebug_f("app({}) has already sync bulk load status, wait and retry, current status = {}, "
- "new status = {}",
- ainfo.app_name,
- dsn::enum_to_string(old_status),
- dsn::enum_to_string(new_status));
+ LOG_INFO_F(
+ "app({}) has already sync bulk load status, wait and retry, current status = {}, "
+ "new status = {}",
+ ainfo.app_name,
+ dsn::enum_to_string(old_status),
+ dsn::enum_to_string(new_status));
tasking::enqueue(LPC_META_STATE_NORMAL,
_meta_svc->tracker(),
std::bind(&bulk_load_service::update_app_status_on_remote_storage_unlocked,
@@ -1119,10 +1128,10 @@ void bulk_load_service::update_app_status_on_remote_storage_reply(const app_bulk
}
}
- ddebug_f("update app({}) status from {} to {}",
- ainfo.app_name,
- dsn::enum_to_string(old_status),
- dsn::enum_to_string(new_status));
+ LOG_INFO_F("update app({}) status from {} to {}",
+ ainfo.app_name,
+ dsn::enum_to_string(old_status),
+ dsn::enum_to_string(new_status));
if (new_status == bulk_load_status::BLS_INGESTING) {
for (auto i = 0; i < partition_count; ++i) {
@@ -1165,14 +1174,14 @@ bool bulk_load_service::check_ever_ingestion_succeed(const partition_configurati
std::sort(pinfo.addresses.begin(), pinfo.addresses.end());
std::sort(current_nodes.begin(), current_nodes.end());
if (current_nodes == pinfo.addresses) {
- ddebug_f("app({}) partition({}) has already executed ingestion succeed", app_name, pid);
+ LOG_INFO_F("app({}) partition({}) has already executed ingestion succeed", app_name, pid);
update_partition_info_on_remote_storage(app_name, pid, bulk_load_status::BLS_SUCCEED);
return true;
}
- dwarn_f("app({}) partition({}) configuration changed, should executed ingestion again",
- app_name,
- pid);
+ LOG_WARNING_F("app({}) partition({}) configuration changed, should executed ingestion again",
+ app_name,
+ pid);
return false;
}
@@ -1183,17 +1192,17 @@ void bulk_load_service::partition_ingestion(const std::string &app_name, const g
auto app_status = get_app_bulk_load_status(pid.get_app_id());
if (app_status != bulk_load_status::BLS_INGESTING) {
- dwarn_f("app({}) current status is {}, partition({}), ignore it",
- app_name,
- dsn::enum_to_string(app_status),
- pid);
+ LOG_WARNING_F("app({}) current status is {}, partition({}), ignore it",
+ app_name,
+ dsn::enum_to_string(app_status),
+ pid);
return;
}
if (is_partition_metadata_not_updated(pid)) {
- derror_f("app({}) partition({}) doesn't have bulk load metadata, set bulk load failed",
- app_name,
- pid);
+ LOG_ERROR_F("app({}) partition({}) doesn't have bulk load metadata, set bulk load failed",
+ app_name,
+ pid);
handle_bulk_load_failed(pid.get_app_id(), ERR_CORRUPTION);
return;
}
@@ -1216,7 +1225,7 @@ void bulk_load_service::partition_ingestion(const std::string &app_name, const g
auto app = get_app(pid.get_app_id());
if (!try_partition_ingestion(pconfig, app->helpers->contexts[pid.get_partition_index()])) {
- dwarn_f(
+ LOG_WARNING_F(
"app({}) partition({}) couldn't execute ingestion, wait and try later", app_name, pid);
tasking::enqueue(LPC_META_STATE_NORMAL,
_meta_svc->tracker(),
@@ -1270,10 +1279,10 @@ void bulk_load_service::send_ingestion_request(const std::string &app_name,
on_partition_ingestion_reply(err, std::move(resp), app_name, pid, primary_addr);
});
_meta_svc->send_request(msg, primary_addr, rpc_callback);
- ddebug_f("send ingest_request to node({}), app({}) partition({})",
- primary_addr.to_string(),
- app_name,
- pid);
+ LOG_INFO_F("send ingest_request to node({}), app({}) partition({})",
+ primary_addr.to_string(),
+ app_name,
+ pid);
}
// ThreadPool: THREAD_POOL_DEFAULT
@@ -1288,21 +1297,22 @@ void bulk_load_service::on_partition_ingestion_reply(error_code err,
}
if (err == ERR_NO_NEED_OPERATE) {
- dwarn_f("app({}) partition({}) on node({}) has already executing ingestion, ignore this "
- "repeated request",
- app_name,
- pid,
- primary_addr.to_string());
+ LOG_WARNING_F(
+ "app({}) partition({}) on node({}) has already executing ingestion, ignore this "
+ "repeated request",
+ app_name,
+ pid,
+ primary_addr.to_string());
return;
}
// if meet 2pc error, ingesting will rollback to downloading, no need to retry here
if (err != ERR_OK) {
- derror_f("app({}) partition({}) on node({}) ingestion files failed, error = {}",
- app_name,
- pid,
- primary_addr.to_string(),
- err);
+ LOG_ERROR_F("app({}) partition({}) on node({}) ingestion files failed, error = {}",
+ app_name,
+ pid,
+ primary_addr.to_string(),
+ err);
tasking::enqueue(
LPC_META_STATE_NORMAL,
_meta_svc->tracker(),
@@ -1311,13 +1321,13 @@ void bulk_load_service::on_partition_ingestion_reply(error_code err,
}
if (resp.err == ERR_TRY_AGAIN && resp.rocksdb_error != 0) {
- derror_f("app({}) partition({}) on node({}) ingestion files failed while empty write, "
- "rocksdb error = "
- "{}, retry it later",
- app_name,
- pid,
- primary_addr.to_string(),
- resp.rocksdb_error);
+ LOG_ERROR_F("app({}) partition({}) on node({}) ingestion files failed while empty write, "
+ "rocksdb error = "
+ "{}, retry it later",
+ app_name,
+ pid,
+ primary_addr.to_string(),
+ resp.rocksdb_error);
tasking::enqueue(LPC_BULK_LOAD_INGESTION,
_meta_svc->tracker(),
std::bind(&bulk_load_service::partition_ingestion, this, app_name, pid),
@@ -1329,13 +1339,14 @@ void bulk_load_service::on_partition_ingestion_reply(error_code err,
// some unexpected errors happened, such as write empty write failed but rocksdb_error is ok
// stop bulk load process with failed
if (resp.err != ERR_OK || resp.rocksdb_error != 0) {
- derror_f("app({}) partition({}) on node({}) failed to ingestion files, error = {}, rocksdb "
- "error = {}",
- app_name,
- pid,
- primary_addr.to_string(),
- resp.err,
- resp.rocksdb_error);
+ LOG_ERROR_F(
+ "app({}) partition({}) on node({}) failed to ingestion files, error = {}, rocksdb "
+ "error = {}",
+ app_name,
+ pid,
+ primary_addr.to_string(),
+ resp.err,
+ resp.rocksdb_error);
tasking::enqueue(LPC_META_STATE_NORMAL,
_meta_svc->tracker(),
@@ -1346,10 +1357,10 @@ void bulk_load_service::on_partition_ingestion_reply(error_code err,
return;
}
- ddebug_f("app({}) partition({}) receive ingestion response from node({}) succeed",
- app_name,
- pid,
- primary_addr.to_string());
+ LOG_INFO_F("app({}) partition({}) receive ingestion response from node({}) succeed",
+ app_name,
+ pid,
+ primary_addr.to_string());
}
// ThreadPool: THREAD_POOL_META_STATE
@@ -1359,7 +1370,7 @@ void bulk_load_service::remove_bulk_load_dir_on_remote_storage(int32_t app_id,
std::string bulk_load_path = get_app_bulk_load_path(app_id);
_meta_svc->get_meta_storage()->delete_node_recursively(
std::move(bulk_load_path), [this, app_id, app_name, bulk_load_path]() {
- ddebug_f("remove app({}) bulk load dir {} succeed", app_name, bulk_load_path);
+ LOG_INFO_F("remove app({}) bulk load dir {} succeed", app_name, bulk_load_path);
reset_local_bulk_load_states(app_id, app_name, true);
});
}
@@ -1371,7 +1382,7 @@ void bulk_load_service::remove_bulk_load_dir_on_remote_storage(std::shared_ptr<a
std::string bulk_load_path = get_app_bulk_load_path(app->app_id);
_meta_svc->get_meta_storage()->delete_node_recursively(
std::move(bulk_load_path), [this, app, set_app_not_bulk_loading, bulk_load_path]() {
- ddebug_f("remove app({}) bulk load dir {} succeed", app->app_name, bulk_load_path);
+ LOG_INFO_F("remove app({}) bulk load dir {} succeed", app->app_name, bulk_load_path);
reset_local_bulk_load_states(app->app_id, app->app_name, true);
if (set_app_not_bulk_loading) {
update_app_not_bulk_loading_on_remote_storage(std::move(app));
@@ -1414,7 +1425,7 @@ void bulk_load_service::reset_local_bulk_load_states_unlocked(int32_t app_id,
_apps_cleaning_up.erase(app_id);
}
- ddebug_f(
+ LOG_INFO_F(
"reset local app({}) bulk load context, is_reset_result({})", app_name, is_reset_result);
}
@@ -1439,7 +1450,7 @@ void bulk_load_service::update_app_not_bulk_loading_on_remote_storage(
_state->get_app_path(*app), std::move(value), [app, this]() {
zauto_write_lock l(app_lock());
app->is_bulk_loading = false;
- ddebug_f("app({}) update app is_bulk_loading to false", app->app_name);
+ LOG_INFO_F("app({}) update app is_bulk_loading to false", app->app_name);
_meta_svc->unlock_meta_op_status();
});
}
@@ -1454,14 +1465,14 @@ void bulk_load_service::on_control_bulk_load(control_bulk_load_rpc rpc)
std::shared_ptr<app_state> app = get_app(app_name);
if (app == nullptr || app->status != app_status::AS_AVAILABLE) {
- derror_f("app({}) is not existed or not available", app_name);
+ LOG_ERROR_F("app({}) is not existed or not available", app_name);
response.err = app == nullptr ? ERR_APP_NOT_EXIST : ERR_APP_DROPPED;
response.__set_hint_msg(fmt::format("app({}) is not existed or not available", app_name));
return;
}
if (!app->is_bulk_loading) {
- derror_f("app({}) is not executing bulk load", app_name);
+ LOG_ERROR_F("app({}) is not executing bulk load", app_name);
response.err = ERR_INACTIVE_STATE;
response.__set_hint_msg(fmt::format("app({}) is not executing bulk load", app_name));
return;
@@ -1476,12 +1487,12 @@ void bulk_load_service::on_control_bulk_load(control_bulk_load_rpc rpc)
auto hint_msg = fmt::format("can not pause bulk load for app({}) with status({})",
app_name,
dsn::enum_to_string(app_status));
- derror_f("{}", hint_msg);
+ LOG_ERROR_F("{}", hint_msg);
response.err = ERR_INVALID_STATE;
response.__set_hint_msg(hint_msg);
return;
}
- ddebug_f("app({}) start to pause bulk load", app_name);
+ LOG_INFO_F("app({}) start to pause bulk load", app_name);
update_app_status_on_remote_storage_unlocked(app_id, bulk_load_status::BLS_PAUSING);
} break;
case bulk_load_control_type::BLC_RESTART: {
@@ -1489,12 +1500,12 @@ void bulk_load_service::on_control_bulk_load(control_bulk_load_rpc rpc)
auto hint_msg = fmt::format("can not restart bulk load for app({}) with status({})",
app_name,
dsn::enum_to_string(app_status));
- derror_f("{}", hint_msg);
+ LOG_ERROR_F("{}", hint_msg);
response.err = ERR_INVALID_STATE;
response.__set_hint_msg(hint_msg);
return;
}
- ddebug_f("app({}) restart bulk load", app_name);
+ LOG_INFO_F("app({}) restart bulk load", app_name);
update_app_status_on_remote_storage_unlocked(
app_id, bulk_load_status::BLS_DOWNLOADING, ERR_OK, true);
} break;
@@ -1504,16 +1515,16 @@ void bulk_load_service::on_control_bulk_load(control_bulk_load_rpc rpc)
auto hint_msg = fmt::format("can not cancel bulk load for app({}) with status({})",
app_name,
dsn::enum_to_string(app_status));
- derror_f("{}", hint_msg);
+ LOG_ERROR_F("{}", hint_msg);
response.err = ERR_INVALID_STATE;
response.__set_hint_msg(hint_msg);
return;
}
case bulk_load_control_type::BLC_FORCE_CANCEL: {
- ddebug_f("app({}) start to {} cancel bulk load, original status = {}",
- app_name,
- control_type == bulk_load_control_type::BLC_FORCE_CANCEL ? "force" : "",
- dsn::enum_to_string(app_status));
+ LOG_INFO_F("app({}) start to {} cancel bulk load, original status = {}",
+ app_name,
+ control_type == bulk_load_control_type::BLC_FORCE_CANCEL ? "force" : "",
+ dsn::enum_to_string(app_status));
update_app_status_on_remote_storage_unlocked(app_id,
bulk_load_status::BLS_CANCELED,
ERR_OK,
@@ -1537,7 +1548,7 @@ void bulk_load_service::on_query_bulk_load_status(query_bulk_load_rpc rpc)
std::shared_ptr<app_state> app = get_app(app_name);
if (app == nullptr || app->status != app_status::AS_AVAILABLE) {
auto hint_msg = fmt::format("app({}) is not existed or not available", app_name);
- derror_f("{}", hint_msg);
+ LOG_ERROR_F("{}", hint_msg);
response.err = (app == nullptr) ? ERR_APP_NOT_EXIST : ERR_APP_DROPPED;
response.__set_hint_msg(hint_msg);
return;
@@ -1546,7 +1557,7 @@ void bulk_load_service::on_query_bulk_load_status(query_bulk_load_rpc rpc)
if (!app->is_bulk_loading) {
auto hint_msg =
fmt::format("app({}) is not during bulk load, return last time result", app_name);
- dwarn_f("{}", hint_msg);
+ LOG_WARNING_F("{}", hint_msg);
response.__set_hint_msg(hint_msg);
}
@@ -1577,9 +1588,9 @@ void bulk_load_service::on_query_bulk_load_status(query_bulk_load_rpc rpc)
response.err = get_app_bulk_load_err_unlocked(app_id);
}
- ddebug_f("query app({}) bulk_load_status({}) succeed",
- app_name,
- dsn::enum_to_string(response.app_status));
+ LOG_INFO_F("query app({}) bulk_load_status({}) succeed",
+ app_name,
+ dsn::enum_to_string(response.app_status));
}
void bulk_load_service::on_clear_bulk_load(clear_bulk_load_rpc rpc)
@@ -1592,14 +1603,14 @@ void bulk_load_service::on_clear_bulk_load(clear_bulk_load_rpc rpc)
if (app == nullptr || app->status != app_status::AS_AVAILABLE) {
response.err = (app == nullptr) ? ERR_APP_NOT_EXIST : ERR_APP_DROPPED;
response.hint_msg = fmt::format("app({}) is not existed or not available", app_name);
- derror_f("{}", response.hint_msg);
+ LOG_ERROR_F("{}", response.hint_msg);
return;
}
if (app->is_bulk_loading) {
response.err = ERR_INVALID_STATE;
response.hint_msg = fmt::format("app({}) is executing bulk load", app_name);
- derror_f("{}", response.hint_msg);
+ LOG_ERROR_F("{}", response.hint_msg);
return;
}
@@ -1619,7 +1630,7 @@ void bulk_load_service::do_clear_app_bulk_load_result(int32_t app_id, clear_bulk
fmt::format("clear app({}) bulk load result succeed, remove bulk load dir succeed",
rpc.request().app_name);
reset_local_bulk_load_states(app_id, rpc.request().app_name, true);
- ddebug_f("{}", response.hint_msg);
+ LOG_INFO_F("{}", response.hint_msg);
});
}
@@ -1629,7 +1640,7 @@ void bulk_load_service::create_bulk_load_root_dir()
blob value = blob();
std::string path = _bulk_load_root;
_sync_bulk_load_storage->create_node(std::move(path), std::move(value), [this]() {
- ddebug_f("create bulk load root({}) succeed", _bulk_load_root);
+ LOG_INFO_F("create bulk load root({}) succeed", _bulk_load_root);
sync_apps_from_remote_storage();
});
}
@@ -1641,10 +1652,10 @@ void bulk_load_service::sync_apps_from_remote_storage()
_sync_bulk_load_storage->get_children(
std::move(path), [this](bool flag, const std::vector<std::string> &children) {
if (flag && children.size() > 0) {
- ddebug_f("There are {} apps need to sync bulk load status", children.size());
+ LOG_INFO_F("There are {} apps need to sync bulk load status", children.size());
for (const auto &elem : children) {
int32_t app_id = boost::lexical_cast<int32_t>(elem);
- ddebug_f("start to sync app({}) bulk load status", app_id);
+ LOG_INFO_F("start to sync app({}) bulk load status", app_id);
do_sync_app(app_id);
}
}
@@ -1675,10 +1686,10 @@ void bulk_load_service::sync_partitions_from_remote_storage(int32_t app_id,
_sync_bulk_load_storage->get_children(
std::move(app_path),
[this, app_path, app_id, app_name](bool flag, const std::vector<std::string> &children) {
- ddebug_f("app(name={},app_id={}) has {} partition bulk load info to be synced",
- app_name,
- app_id,
- children.size());
+ LOG_INFO_F("app(name={},app_id={}) has {} partition bulk load info to be synced",
+ app_name,
+ app_id,
+ children.size());
for (const auto &child_pidx : children) {
int32_t pidx = boost::lexical_cast<int32_t>(child_pidx);
std::string partition_path = get_partition_bulk_load_path(app_path, pidx);
@@ -1726,7 +1737,7 @@ void bulk_load_service::try_to_continue_app_bulk_load(
std::shared_ptr<app_state> app = get_app(ainfo.app_name);
// if app is not available, remove bulk load dir
if (app == nullptr || app->status != app_status::AS_AVAILABLE) {
- derror_f(
+ LOG_ERROR_F(
"app(name={},app_id={}) is not existed or not available", ainfo.app_name, ainfo.app_id);
if (app == nullptr) {
remove_bulk_load_dir_on_remote_storage(ainfo.app_id, ainfo.app_name);
@@ -1774,7 +1785,7 @@ bulk_load_service::validate_ingest_behind(const std::map<std::string, std::strin
const auto &iter = envs.find(replica_envs::ROCKSDB_ALLOW_INGEST_BEHIND);
if (iter != envs.end()) {
if (!buf2bool(iter->second, app_allow_ingest_behind)) {
- dwarn_f("can not convert {} to bool", iter->second);
+ LOG_WARNING_F("can not convert {} to bool", iter->second);
app_allow_ingest_behind = false;
}
}
@@ -1793,23 +1804,23 @@ bulk_load_service::validate_ingest_behind(const std::map<std::string, std::strin
{
// app id and partition from `app_bulk_load_info` is inconsistent with current app_info
if (app_id != ainfo.app_id || partition_count != ainfo.partition_count) {
- derror_f("app({}) has different app_id or partition_count, bulk load app_id = {}, "
- "partition_count = {}, current app_id = {}, partition_count = {}",
- ainfo.app_name,
- ainfo.app_id,
- ainfo.partition_count,
- app_id,
- partition_count);
+ LOG_ERROR_F("app({}) has different app_id or partition_count, bulk load app_id = {}, "
+ "partition_count = {}, current app_id = {}, partition_count = {}",
+ ainfo.app_name,
+ ainfo.app_id,
+ ainfo.partition_count,
+ app_id,
+ partition_count);
return false;
}
// partition_bulk_load_info count should not be greater than partition_count
if (partition_count < pinfo_count) {
- derror_f("app({}) has invalid count, app partition_count = {}, remote "
- "partition_bulk_load_info count = {}",
- ainfo.app_name,
- partition_count,
- pinfo_count);
+ LOG_ERROR_F("app({}) has invalid count, app partition_count = {}, remote "
+ "partition_bulk_load_info count = {}",
+ ainfo.app_name,
+ partition_count,
+ pinfo_count);
return false;
}
@@ -1820,16 +1831,16 @@ bulk_load_service::validate_ingest_behind(const std::map<std::string, std::strin
// however, meta server crash when create app directory and part of partition directory
// when meta server recover, partition directory count is less than partition_count
if (pinfo_count != partition_count && ainfo.status != bulk_load_status::BLS_DOWNLOADING) {
- derror_f("app({}) bulk_load_status = {}, but there are {} partitions lack "
- "partition_bulk_load dir",
- ainfo.app_name,
- dsn::enum_to_string(ainfo.status),
- partition_count - pinfo_count);
+ LOG_ERROR_F("app({}) bulk_load_status = {}, but there are {} partitions lack "
+ "partition_bulk_load dir",
+ ainfo.app_name,
+ dsn::enum_to_string(ainfo.status),
+ partition_count - pinfo_count);
return false;
}
if (!validate_ingest_behind(envs, ainfo.ingest_behind)) {
- derror_f("app({}) has inconsistent ingest_behind option", ainfo.app_name);
+ LOG_ERROR_F("app({}) has inconsistent ingest_behind option", ainfo.app_name);
return false;
}
@@ -1853,12 +1864,13 @@ bulk_load_service::validate_ingest_behind(const std::map<std::string, std::strin
// some partition bulk load status is not downloading and some partition directroy is
// missing on remote storage
if (ainfo.partition_count - pinfo_map.size() > 0 && different_status_count > 0) {
- derror_f("app({}) bulk_load_status = {}, there are {} partitions status is different "
- "from app, and {} partitions not existed, this is invalid",
- ainfo.app_name,
- dsn::enum_to_string(app_status),
- different_status_count,
- ainfo.partition_count - pinfo_map.size());
+ LOG_ERROR_F(
+ "app({}) bulk_load_status = {}, there are {} partitions status is different "
+ "from app, and {} partitions not existed, this is invalid",
+ ainfo.app_name,
+ dsn::enum_to_string(app_status),
+ different_status_count,
+ ainfo.partition_count - pinfo_map.size());
is_valid = false;
}
break;
@@ -1871,14 +1883,15 @@ bulk_load_service::validate_ingest_behind(const std::map<std::string, std::strin
: bulk_load_status::BLS_SUCCEED;
for (const auto &kv : pinfo_map) {
if (kv.second.status != app_status && kv.second.status != other_valid_status) {
- derror_f("app({}) bulk_load_status = {}, but partition[{}] bulk_load_status = {}, "
- "only {} and {} is valid",
- ainfo.app_name,
- app_status,
- kv.first,
- dsn::enum_to_string(kv.second.status),
- dsn::enum_to_string(app_status),
- dsn::enum_to_string(other_valid_status));
+ LOG_ERROR_F(
+ "app({}) bulk_load_status = {}, but partition[{}] bulk_load_status = {}, "
+ "only {} and {} is valid",
+ ainfo.app_name,
+ app_status,
+ kv.first,
+ dsn::enum_to_string(kv.second.status),
+ dsn::enum_to_string(app_status),
+ dsn::enum_to_string(other_valid_status));
is_valid = false;
break;
}
@@ -1889,11 +1902,12 @@ bulk_load_service::validate_ingest_behind(const std::map<std::string, std::strin
// if app status is succeed or paused, all partitions' status should not be different from
// app's
if (different_status_count > 0) {
- derror_f("app({}) bulk_load_status = {}, {} partitions status is different from app, "
- "this is invalid",
- ainfo.app_name,
- dsn::enum_to_string(app_status),
- different_status_count);
+ LOG_ERROR_F(
+ "app({}) bulk_load_status = {}, {} partitions status is different from app, "
+ "this is invalid",
+ ainfo.app_name,
+ dsn::enum_to_string(app_status),
+ different_status_count);
is_valid = false;
}
break;
@@ -1920,10 +1934,10 @@ void bulk_load_service::do_continue_app_bulk_load(
if (!FLAGS_enable_concurrent_bulk_load &&
!_meta_svc->try_lock_meta_op_status(meta_op_status::BULKLOAD)) {
- derror_f("fatal, the op status of meta server must be meta_op_status::FREE");
+ LOG_ERROR_F("fatal, the op status of meta server must be meta_op_status::FREE");
return;
}
- ddebug_f(
+ LOG_INFO_F(
"app({}) continue bulk load, app_id = {}, partition_count = {}, status = {}, there are {} "
"partitions have bulk_load_info, {} partitions have same status with app, {} "
"partitions different",
@@ -2017,7 +2031,7 @@ void bulk_load_service::create_missing_partition_dir(const std::string &app_name
[app_name, pid, partition_count, pinfo, this]() {
const int32_t app_id = pid.get_app_id();
bool send_request = false;
- ddebug_f("app({}) create partition({}) bulk_load_info", app_name, pid);
+ LOG_INFO_F("app({}) create partition({}) bulk_load_info", app_name, pid);
{
zauto_write_lock l(_lock);
_partition_bulk_load_info[pid] = pinfo;
@@ -2028,7 +2042,7 @@ void bulk_load_service::create_missing_partition_dir(const std::string &app_name
}
}
if (send_request) {
- ddebug_f("app({}) start to bulk load", app_name);
+ LOG_INFO_F("app({}) start to bulk load", app_name);
for (auto i = 0; i < partition_count; ++i) {
partition_bulk_load(app_name, gpid(app_id, i));
}
@@ -2044,10 +2058,10 @@ void bulk_load_service::check_app_bulk_load_states(std::shared_ptr<app_state> ap
_meta_svc->get_remote_storage()->node_exist(
app_path, LPC_META_CALLBACK, [this, app_path, app, is_app_bulk_loading](error_code err) {
if (err != ERR_OK && err != ERR_OBJECT_NOT_FOUND) {
- dwarn_f("check app({}) bulk load dir({}) failed, error = {}, try later",
- app->app_name,
- app_path,
- err);
+ LOG_WARNING_F("check app({}) bulk load dir({}) failed, error = {}, try later",
+ app->app_name,
+ app_path,
+ err);
tasking::enqueue(LPC_META_CALLBACK,
nullptr,
std::bind(&bulk_load_service::check_app_bulk_load_states,
@@ -2060,11 +2074,11 @@ void bulk_load_service::check_app_bulk_load_states(std::shared_ptr<app_state> ap
}
if (err == ERR_OBJECT_NOT_FOUND && is_app_bulk_loading) {
- derror_f("app({}): bulk load dir({}) not exist, but is_bulk_loading = {}, reset "
- "app is_bulk_loading flag",
- app->app_name,
- app_path,
- is_app_bulk_loading);
+ LOG_ERROR_F("app({}): bulk load dir({}) not exist, but is_bulk_loading = {}, reset "
+ "app is_bulk_loading flag",
+ app->app_name,
+ app_path,
+ is_app_bulk_loading);
update_app_not_bulk_loading_on_remote_storage(std::move(app));
return;
}
diff --git a/src/meta/meta_service.cpp b/src/meta/meta_service.cpp
index bed803b63..ffdf440eb 100644
--- a/src/meta/meta_service.cpp
+++ b/src/meta/meta_service.cpp
@@ -191,19 +191,19 @@ bool meta_service::try_lock_meta_op_status(meta_op_status op_status)
{
meta_op_status expected = meta_op_status::FREE;
if (!_meta_op_status.compare_exchange_strong(expected, op_status)) {
- derror_f("LOCK meta op status failed, meta "
- "server is busy, current op status is {}",
- enum_to_string(expected));
+ LOG_ERROR_F("LOCK meta op status failed, meta "
+ "server is busy, current op status is {}",
+ enum_to_string(expected));
return false;
}
- ddebug_f("LOCK meta op status to {}", enum_to_string(op_status));
+ LOG_INFO_F("LOCK meta op status to {}", enum_to_string(op_status));
return true;
}
void meta_service::unlock_meta_op_status()
{
- ddebug_f("UNLOCK meta op status from {}", enum_to_string(_meta_op_status.load()));
+ LOG_INFO_F("UNLOCK meta op status from {}", enum_to_string(_meta_op_status.load()));
_meta_op_status.store(meta_op_status::FREE);
}
@@ -647,10 +647,10 @@ void meta_service::on_query_configuration_by_index(configuration_query_by_index_
_state->query_configuration_by_index(rpc.request(), response);
if (ERR_OK == response.err) {
- ddebug_f("client {} queried an available app {} with appid {}",
- rpc.dsn_request()->header->from_address.to_string(),
- rpc.request().app_name,
- response.app_id);
+ LOG_INFO_F("client {} queried an available app {} with appid {}",
+ rpc.dsn_request()->header->from_address.to_string(),
+ rpc.request().app_name,
+ response.app_id);
}
}
@@ -1001,7 +1001,7 @@ void meta_service::on_start_partition_split(start_split_rpc rpc)
return;
}
if (_split_svc == nullptr) {
- derror_f("meta doesn't support partition split");
+ LOG_ERROR_F("meta doesn't support partition split");
rpc.response().err = ERR_SERVICE_NOT_ACTIVE;
return;
}
@@ -1018,7 +1018,7 @@ void meta_service::on_control_partition_split(control_split_rpc rpc)
}
if (_split_svc == nullptr) {
- derror_f("meta doesn't support partition split");
+ LOG_ERROR_F("meta doesn't support partition split");
rpc.response().err = ERR_SERVICE_NOT_ACTIVE;
return;
}
@@ -1035,7 +1035,7 @@ void meta_service::on_query_partition_split(query_split_rpc rpc)
}
if (_split_svc == nullptr) {
- derror_f("meta doesn't support partition split");
+ LOG_ERROR_F("meta doesn't support partition split");
rpc.response().err = ERR_SERVICE_NOT_ACTIVE;
return;
}
@@ -1060,7 +1060,7 @@ void meta_service::on_notify_stop_split(notify_stop_split_rpc rpc)
return;
}
if (_split_svc == nullptr) {
- derror_f("meta doesn't support partition split");
+ LOG_ERROR_F("meta doesn't support partition split");
rpc.response().err = ERR_SERVICE_NOT_ACTIVE;
return;
}
@@ -1076,7 +1076,7 @@ void meta_service::on_query_child_state(query_child_state_rpc rpc)
return;
}
if (_split_svc == nullptr) {
- derror_f("meta doesn't support partition split");
+ LOG_ERROR_F("meta doesn't support partition split");
rpc.response().err = ERR_SERVICE_NOT_ACTIVE;
return;
}
@@ -1090,7 +1090,7 @@ void meta_service::on_start_bulk_load(start_bulk_load_rpc rpc)
}
if (_bulk_load_svc == nullptr) {
- derror_f("meta doesn't support bulk load");
+ LOG_ERROR_F("meta doesn't support bulk load");
rpc.response().err = ERR_SERVICE_NOT_ACTIVE;
return;
}
@@ -1104,7 +1104,7 @@ void meta_service::on_control_bulk_load(control_bulk_load_rpc rpc)
}
if (_bulk_load_svc == nullptr) {
- derror_f("meta doesn't support bulk load");
+ LOG_ERROR_F("meta doesn't support bulk load");
rpc.response().err = ERR_SERVICE_NOT_ACTIVE;
return;
}
@@ -1121,7 +1121,7 @@ void meta_service::on_query_bulk_load_status(query_bulk_load_rpc rpc)
}
if (_bulk_load_svc == nullptr) {
- derror_f("meta doesn't support bulk load");
+ LOG_ERROR_F("meta doesn't support bulk load");
rpc.response().err = ERR_SERVICE_NOT_ACTIVE;
return;
}
@@ -1135,7 +1135,7 @@ void meta_service::on_clear_bulk_load(clear_bulk_load_rpc rpc)
}
if (_bulk_load_svc == nullptr) {
- derror_f("meta doesn't support bulk load");
+ LOG_ERROR_F("meta doesn't support bulk load");
rpc.response().err = ERR_SERVICE_NOT_ACTIVE;
return;
}
@@ -1151,7 +1151,7 @@ void meta_service::on_start_backup_app(start_backup_app_rpc rpc)
return;
}
if (_backup_handler == nullptr) {
- derror_f("meta doesn't enable backup service");
+ LOG_ERROR_F("meta doesn't enable backup service");
rpc.response().err = ERR_SERVICE_NOT_ACTIVE;
return;
}
@@ -1164,7 +1164,7 @@ void meta_service::on_query_backup_status(query_backup_status_rpc rpc)
return;
}
if (_backup_handler == nullptr) {
- derror_f("meta doesn't enable backup service");
+ LOG_ERROR_F("meta doesn't enable backup service");
rpc.response().err = ERR_SERVICE_NOT_ACTIVE;
return;
}
diff --git a/src/meta/meta_split_service.cpp b/src/meta/meta_split_service.cpp
index 1d6265487..0e90ff3fc 100644
--- a/src/meta/meta_split_service.cpp
+++ b/src/meta/meta_split_service.cpp
@@ -43,7 +43,7 @@ void meta_split_service::start_partition_split(start_split_rpc rpc)
app = _state->get_app(request.app_name);
if (app == nullptr || app->status != app_status::AS_AVAILABLE) {
- derror_f("app({}) is not existed or not available", request.app_name);
+ LOG_ERROR_F("app({}) is not existed or not available", request.app_name);
response.err = app == nullptr ? ERR_APP_NOT_EXIST : ERR_APP_DROPPED;
response.hint_msg = fmt::format(
"app {}", response.err == ERR_APP_NOT_EXIST ? "not existed" : "dropped");
@@ -53,10 +53,11 @@ void meta_split_service::start_partition_split(start_split_rpc rpc)
// new_partition_count != old_partition_count*2
if (request.new_partition_count != app->partition_count * 2) {
response.err = ERR_INVALID_PARAMETERS;
- derror_f("wrong partition count: app({}), partition count({}), new_partition_count({})",
- request.app_name,
- app->partition_count,
- request.new_partition_count);
+ LOG_ERROR_F(
+ "wrong partition count: app({}), partition count({}), new_partition_count({})",
+ request.app_name,
+ app->partition_count,
+ request.new_partition_count);
response.hint_msg =
fmt::format("wrong partition_count, should be {}", app->partition_count * 2);
return;
@@ -66,15 +67,15 @@ void meta_split_service::start_partition_split(start_split_rpc rpc)
response.err = ERR_BUSY;
auto err_msg =
fmt::format("app({}) is already executing partition split", request.app_name);
- derror_f("{}", err_msg);
+ LOG_ERROR_F("{}", err_msg);
response.hint_msg = err_msg;
return;
}
}
- ddebug_f("app({}) start to partition split, new_partition_count={}",
- request.app_name,
- request.new_partition_count);
+ LOG_INFO_F("app({}) start to partition split, new_partition_count={}",
+ request.app_name,
+ request.new_partition_count);
do_start_partition_split(std::move(app), std::move(rpc));
}
@@ -83,9 +84,9 @@ void meta_split_service::do_start_partition_split(std::shared_ptr<app_state> app
start_split_rpc rpc)
{
auto on_write_storage_complete = [app, rpc, this]() {
- ddebug_f("app({}) update partition count on remote storage, new partition_count = {}",
- app->app_name,
- app->partition_count * 2);
+ LOG_INFO_F("app({}) update partition count on remote storage, new partition_count = {}",
+ app->app_name,
+ app->partition_count * 2);
zauto_write_lock l(app_lock());
app->helpers->split_states.splitting_count = app->partition_count;
@@ -135,13 +136,13 @@ void meta_split_service::register_child_on_meta(register_child_rpc rpc)
const gpid &child_gpid = request.child_config.pid;
const auto &parent_config = app->partitions[parent_gpid.get_partition_index()];
if (request.parent_config.ballot != parent_config.ballot) {
- derror_f("app({}) partition({}) register child({}) failed, request is outdated, request "
- "parent ballot = {}, local parent ballot = {}",
- app_name,
- parent_gpid,
- child_gpid,
- request.parent_config.ballot,
- parent_config.ballot);
+ LOG_ERROR_F("app({}) partition({}) register child({}) failed, request is outdated, request "
+ "parent ballot = {}, local parent ballot = {}",
+ app_name,
+ parent_gpid,
+ child_gpid,
+ request.parent_config.ballot,
+ parent_config.ballot);
response.err = ERR_INVALID_VERSION;
response.parent_config = parent_config;
return;
@@ -149,15 +150,16 @@ void meta_split_service::register_child_on_meta(register_child_rpc rpc)
config_context &parent_context = app->helpers->contexts[parent_gpid.get_partition_index()];
if (parent_context.stage == config_status::pending_remote_sync) {
- dwarn_f("app({}) partition({}): another request is syncing with remote storage, ignore "
- "this request",
- app_name,
- parent_gpid);
+ LOG_WARNING_F(
+ "app({}) partition({}): another request is syncing with remote storage, ignore "
+ "this request",
+ app_name,
+ parent_gpid);
return;
}
if (child_gpid.get_partition_index() >= app->partition_count) {
- derror_f(
+ LOG_ERROR_F(
"app({}) partition({}) register child({}) failed, partition split has been canceled",
app_name,
parent_gpid,
@@ -169,7 +171,7 @@ void meta_split_service::register_child_on_meta(register_child_rpc rpc)
auto iter = app->helpers->split_states.status.find(parent_gpid.get_partition_index());
if (iter == app->helpers->split_states.status.end()) {
- derror_f(
+ LOG_ERROR_F(
"duplicated register request, app({}) child partition({}) has already been registered",
app_name,
child_gpid);
@@ -184,7 +186,7 @@ void meta_split_service::register_child_on_meta(register_child_rpc rpc)
}
if (iter->second != split_status::SPLITTING) {
- derror_f(
+ LOG_ERROR_F(
"app({}) partition({}) register child({}) failed, current partition split_status = {}",
app_name,
parent_gpid,
@@ -196,7 +198,7 @@ void meta_split_service::register_child_on_meta(register_child_rpc rpc)
app->helpers->split_states.status.erase(parent_gpid.get_partition_index());
app->helpers->split_states.splitting_count--;
- ddebug_f("app({}) parent({}) will register child({})", app_name, parent_gpid, child_gpid);
+ LOG_INFO_F("app({}) parent({}) will register child({})", app_name, parent_gpid, child_gpid);
parent_context.stage = config_status::pending_remote_sync;
parent_context.msg = rpc.dsn_request();
@@ -267,7 +269,7 @@ void meta_split_service::on_add_child_on_remote_storage_reply(error_code ec,
}
dassert_f(ec == ERR_OK, "we can't handle this right now, err = {}", ec);
- ddebug_f("parent({}) resgiter child({}) on remote storage succeed", parent_gpid, child_gpid);
+ LOG_INFO_F("parent({}) resgiter child({}) on remote storage succeed", parent_gpid, child_gpid);
// update local child partition configuration
std::shared_ptr<configuration_update_request> update_child_request =
@@ -304,23 +306,23 @@ void meta_split_service::query_partition_split(query_split_rpc rpc) const
response.err = app == nullptr ? ERR_APP_NOT_EXIST : ERR_APP_DROPPED;
response.__set_hint_msg(fmt::format(
"app({}) {}", app_name, response.err == ERR_APP_NOT_EXIST ? "not existed" : "dropped"));
- derror_f("query partition split failed, {}", response.hint_msg);
+ LOG_ERROR_F("query partition split failed, {}", response.hint_msg);
return;
}
if (!app->splitting()) {
response.err = ERR_INVALID_STATE;
response.__set_hint_msg(fmt::format("app({}) is not splitting", app_name));
- derror_f("query partition split failed, {}", response.hint_msg);
+ LOG_ERROR_F("query partition split failed, {}", response.hint_msg);
return;
}
response.new_partition_count = app->partition_count;
response.status = app->helpers->split_states.status;
- ddebug_f("query partition split succeed, app({}), partition_count({}), splitting_count({})",
- app->app_name,
- response.new_partition_count,
- response.status.size());
+ LOG_INFO_F("query partition split succeed, app({}), partition_count({}), splitting_count({})",
+ app->app_name,
+ response.new_partition_count,
+ response.status.size());
}
void meta_split_service::control_partition_split(control_split_rpc rpc)
@@ -335,14 +337,14 @@ void meta_split_service::control_partition_split(control_split_rpc rpc)
response.err = app == nullptr ? ERR_APP_NOT_EXIST : ERR_APP_DROPPED;
response.__set_hint_msg(fmt::format(
"app {}", response.err == ERR_APP_NOT_EXIST ? "not existed" : "dropped", req.app_name));
- derror_f("{} split failed, {}", control_type_str(control_type), response.hint_msg);
+ LOG_ERROR_F("{} split failed, {}", control_type_str(control_type), response.hint_msg);
return;
}
if (!app->splitting()) {
response.err = ERR_INVALID_STATE;
response.__set_hint_msg(fmt::format("app({}) is not splitting", req.app_name));
- derror_f("{} split failed, {}", control_type_str(control_type), response.hint_msg);
+ LOG_ERROR_F("{} split failed, {}", control_type_str(control_type), response.hint_msg);
return;
}
@@ -365,10 +367,10 @@ void meta_split_service::do_control_single(std::shared_ptr<app_state> app, contr
if (parent_pidx >= app->partition_count / 2) {
response.err = ERR_INVALID_PARAMETERS;
response.__set_hint_msg(fmt::format("invalid parent partition index({})", parent_pidx));
- derror_f("{} split for app({}) failed, {}",
- control_type_str(control_type),
- app_name,
- response.hint_msg);
+ LOG_ERROR_F("{} split for app({}) failed, {}",
+ control_type_str(control_type),
+ app_name,
+ response.hint_msg);
return;
}
@@ -377,10 +379,10 @@ void meta_split_service::do_control_single(std::shared_ptr<app_state> app, contr
response.err =
control_type == split_control_type::PAUSE ? ERR_CHILD_REGISTERED : ERR_INVALID_STATE;
response.__set_hint_msg(fmt::format("partition[{}] is not splitting", parent_pidx));
- derror_f("{} split for app({}) failed, {}",
- control_type_str(control_type),
- app_name,
- response.hint_msg);
+ LOG_ERROR_F("{} split for app({}) failed, {}",
+ control_type_str(control_type),
+ app_name,
+ response.hint_msg);
return;
}
@@ -391,19 +393,19 @@ void meta_split_service::do_control_single(std::shared_ptr<app_state> app, contr
if (iter->second == old_status) {
iter->second = target_status;
response.err = ERR_OK;
- ddebug_f("app({}) partition[{}] {} split succeed",
- app_name,
- parent_pidx,
- control_type_str(control_type));
+ LOG_INFO_F("app({}) partition[{}] {} split succeed",
+ app_name,
+ parent_pidx,
+ control_type_str(control_type));
} else {
response.err = ERR_INVALID_STATE;
response.__set_hint_msg(fmt::format("partition[{}] wrong split_status({})",
parent_pidx,
dsn::enum_to_string(iter->second)));
- derror_f("{} split for app({}) failed, {}",
- control_type_str(control_type),
- app_name,
- response.hint_msg);
+ LOG_ERROR_F("{} split for app({}) failed, {}",
+ control_type_str(control_type),
+ app_name,
+ response.hint_msg);
}
}
@@ -418,26 +420,27 @@ void meta_split_service::do_control_all(std::shared_ptr<app_state> app, control_
response.err = ERR_INVALID_PARAMETERS;
response.__set_hint_msg(
fmt::format("wrong partition_count, should be {}", app->partition_count / 2));
- derror_f("cancel split for app({}) failed, wrong partition count: partition count({}) "
- "VS req partition_count({})",
- app->app_name,
- app->partition_count,
- req.old_partition_count);
+ LOG_ERROR_F(
+ "cancel split for app({}) failed, wrong partition count: partition count({}) "
+ "VS req partition_count({})",
+ app->app_name,
+ app->partition_count,
+ req.old_partition_count);
return;
}
if (app->helpers->split_states.splitting_count != req.old_partition_count) {
response.err = ERR_CHILD_REGISTERED;
response.__set_hint_msg("some partitions have already finished split");
- derror_f("cancel split for app({}) failed, {}", app->app_name, response.hint_msg);
+ LOG_ERROR_F("cancel split for app({}) failed, {}", app->app_name, response.hint_msg);
return;
}
for (auto &kv : app->helpers->split_states.status) {
- ddebug_f("app({}) partition({}) cancel split, old status = {}",
- app->app_name,
- kv.first,
- dsn::enum_to_string(kv.second));
+ LOG_INFO_F("app({}) partition({}) cancel split, old status = {}",
+ app->app_name,
+ kv.first,
+ dsn::enum_to_string(kv.second));
kv.second = split_status::CANCELING;
}
return;
@@ -450,10 +453,10 @@ void meta_split_service::do_control_all(std::shared_ptr<app_state> app, control_
for (auto &kv : app->helpers->split_states.status) {
if (kv.second == old_status) {
kv.second = target_status;
- ddebug_f("app({}) partition[{}] {} split succeed",
- app->app_name,
- kv.first,
- control_type_str(control_type));
+ LOG_INFO_F("app({}) partition[{}] {} split succeed",
+ app->app_name,
+ kv.first,
+ control_type_str(control_type));
}
}
response.err = ERR_OK;
@@ -477,29 +480,30 @@ void meta_split_service::notify_stop_split(notify_stop_split_rpc rpc)
const auto iter =
app->helpers->split_states.status.find(request.parent_gpid.get_partition_index());
if (iter == app->helpers->split_states.status.end()) {
- dwarn_f("app({}) partition({}) is not executing partition split, ignore out-dated {} split "
- "request",
- app->app_name,
- request.parent_gpid,
- stop_type);
+ LOG_WARNING_F(
+ "app({}) partition({}) is not executing partition split, ignore out-dated {} split "
+ "request",
+ app->app_name,
+ request.parent_gpid,
+ stop_type);
response.err = ERR_INVALID_VERSION;
return;
}
if (iter->second != request.meta_split_status) {
- dwarn_f("app({}) partition({}) split_status = {}, ignore out-dated {} split request",
- app->app_name,
- request.parent_gpid,
- dsn::enum_to_string(iter->second),
- stop_type);
+ LOG_WARNING_F("app({}) partition({}) split_status = {}, ignore out-dated {} split request",
+ app->app_name,
+ request.parent_gpid,
+ dsn::enum_to_string(iter->second),
+ stop_type);
response.err = ERR_INVALID_VERSION;
return;
}
- ddebug_f("app({}) partition({}) notify {} split succeed",
- app->app_name,
- request.parent_gpid,
- stop_type);
+ LOG_INFO_F("app({}) partition({}) notify {} split succeed",
+ app->app_name,
+ request.parent_gpid,
+ stop_type);
// pausing split
if (iter->second == split_status::PAUSING) {
@@ -525,9 +529,9 @@ void meta_split_service::do_cancel_partition_split(std::shared_ptr<app_state> ap
notify_stop_split_rpc rpc)
{
auto on_write_storage_complete = [app, rpc, this]() {
- ddebug_f("app({}) update partition count on remote storage, new partition count is {}",
- app->app_name,
- app->partition_count / 2);
+ LOG_INFO_F("app({}) update partition count on remote storage, new partition count is {}",
+ app->app_name,
+ app->partition_count / 2);
zauto_write_lock l(app_lock());
app->partition_count /= 2;
app->helpers->contexts.resize(app->partition_count);
@@ -555,7 +559,7 @@ void meta_split_service::query_child_state(query_child_state_rpc rpc)
if (app->partition_count == request.partition_count) {
response.err = ERR_INVALID_STATE;
- derror_f("app({}) is not executing partition split", app_name);
+ LOG_ERROR_F("app({}) is not executing partition split", app_name);
return;
}
@@ -566,10 +570,10 @@ void meta_split_service::query_child_state(query_child_state_rpc rpc)
auto child_pidx = parent_pid.get_partition_index() + request.partition_count;
if (app->partitions[child_pidx].ballot == invalid_ballot) {
response.err = ERR_INVALID_STATE;
- derror_f("app({}) parent partition({}) split has been canceled", app_name, parent_pid);
+ LOG_ERROR_F("app({}) parent partition({}) split has been canceled", app_name, parent_pid);
return;
}
- ddebug_f(
+ LOG_INFO_F(
"app({}) child partition({}.{}) is ready", app_name, parent_pid.get_app_id(), child_pidx);
response.err = ERR_OK;
response.__set_partition_count(app->partition_count);
diff --git a/src/meta/meta_state_service_utils_impl.h b/src/meta/meta_state_service_utils_impl.h
index c1d355bbd..910bed66b 100644
--- a/src/meta/meta_state_service_utils_impl.h
+++ b/src/meta/meta_state_service_utils_impl.h
@@ -75,9 +75,9 @@ struct operation : pipeline::environment
void on_error(T *this_instance, op_type::type type, error_code ec, const std::string &path)
{
if (ec == ERR_TIMEOUT) {
- dwarn_f("request({}) on path({}) was timeout, retry after 1 second",
- op_type::to_string(type),
- path);
+ LOG_WARNING_F("request({}) on path({}) was timeout, retry after 1 second",
+ op_type::to_string(type),
+ path);
pipeline::repeat(std::move(*this_instance), 1_s);
return;
}
diff --git a/src/meta/server_state.cpp b/src/meta/server_state.cpp
index 65d9f0475..7621b7fc6 100644
--- a/src/meta/server_state.cpp
+++ b/src/meta/server_state.cpp
@@ -623,12 +623,12 @@ dsn::error_code server_state::sync_apps_from_remote_storage()
? app->init_partition_count
: app->partition_count;
if (partition_id < init_partition_count) {
- dwarn_f(
+ LOG_WARNING_F(
"partition node {} not exist on remote storage, may half create before",
partition_path);
init_app_partition_node(app, partition_id, nullptr);
} else if (partition_id >= app->partition_count / 2) {
- dwarn_f(
+ LOG_WARNING_F(
"partition node {} not exist on remote storage, may half split before",
partition_path);
zauto_write_lock l(_lock);
@@ -883,10 +883,11 @@ void server_state::on_config_sync(configuration_query_by_node_rpc rpc)
rep.pid.get_partition_index() < app->partition_count * 2 &&
rep.status == partition_status::PS_ERROR) {
response.gc_replicas.push_back(rep);
- dwarn_f("notify node({}) to gc replica({}) because it is useless partition "
- "which is caused by cancel split",
- request.node.to_string(),
- rep.pid);
+ LOG_WARNING_F(
+ "notify node({}) to gc replica({}) because it is useless partition "
+ "which is caused by cancel split",
+ request.node.to_string(),
+ rep.pid);
} else {
// app is not recognized or partition is not recognized
dassert(false,
@@ -957,12 +958,12 @@ void server_state::on_config_sync(configuration_query_by_node_rpc rpc)
response.err = ERR_BUSY;
response.partitions.clear();
}
- ddebug_f("send config sync response to {}, err({}), partitions_count({}), "
- "gc_replicas_count({})",
- request.node.to_string(),
- response.err,
- response.partitions.size(),
- response.gc_replicas.size());
+ LOG_INFO_F("send config sync response to {}, err({}), partitions_count({}), "
+ "gc_replicas_count({})",
+ request.node.to_string(),
+ response.err,
+ response.partitions.size(),
+ response.gc_replicas.size());
}
bool server_state::query_configuration_by_gpid(dsn::gpid id,
@@ -1099,18 +1100,19 @@ void server_state::create_app(dsn::message_ex *msg)
const auto &duplication_env_iterator =
request.options.envs.find(duplication_constants::kDuplicationEnvMasterClusterKey);
- ddebug_f("create app request, name({}), type({}), partition_count({}), replica_count({}), "
- "duplication({})",
- request.app_name,
- request.options.app_type,
- request.options.partition_count,
- request.options.replica_count,
- duplication_env_iterator == request.options.envs.end()
- ? "false"
- : fmt::format(
- "{}.{}",
- request.options.envs[duplication_constants::kDuplicationEnvMasterClusterKey],
- request.app_name));
+ LOG_INFO_F(
+ "create app request, name({}), type({}), partition_count({}), replica_count({}), "
+ "duplication({})",
+ request.app_name,
+ request.options.app_type,
+ request.options.partition_count,
+ request.options.replica_count,
+ duplication_env_iterator == request.options.envs.end()
+ ? "false"
+ : fmt::format(
+ "{}.{}",
+ request.options.envs[duplication_constants::kDuplicationEnvMasterClusterKey],
+ request.app_name));
auto option_match_check = [](const create_app_options &opt, const app_state &exist_app) {
return opt.partition_count == exist_app.partition_count &&
@@ -2979,7 +2981,7 @@ bool server_state::validate_target_max_replica_count(int32_t max_replica_count)
std::string hint_message;
const auto valid = validate_target_max_replica_count(max_replica_count, hint_message);
if (!valid) {
- derror_f("target max replica count is invalid: message={}", hint_message);
+ LOG_ERROR_F("target max replica count is invalid: message={}", hint_message);
}
return valid;
@@ -3000,7 +3002,7 @@ void server_state::on_start_manual_compact(start_manual_compact_rpc rpc)
fmt::format("app {} is {}",
app_name,
response.err == ERR_APP_NOT_EXIST ? "not existed" : "not available");
- derror_f("{}", response.hint_msg);
+ LOG_ERROR_F("{}", response.hint_msg);
return;
}
envs = app->envs;
@@ -3010,7 +3012,7 @@ void server_state::on_start_manual_compact(start_manual_compact_rpc rpc)
if (iter != envs.end() && iter->second == "true") {
response.err = ERR_OPERATION_DISABLED;
response.hint_msg = fmt::format("app {} disable manual compaction", app_name);
- derror_f("{}", response.hint_msg);
+ LOG_ERROR_F("{}", response.hint_msg);
return;
}
@@ -3044,7 +3046,7 @@ bool server_state::parse_compaction_envs(start_manual_compact_rpc rpc,
response.err = ERR_INVALID_PARAMETERS;
response.hint_msg = fmt::format(
"invalid target_level({}), should in range of [-1, num_levels]", target_level);
- derror_f("{}", response.hint_msg);
+ LOG_ERROR_F("{}", response.hint_msg);
return false;
}
}
@@ -3057,7 +3059,7 @@ bool server_state::parse_compaction_envs(start_manual_compact_rpc rpc,
response.hint_msg =
fmt::format("invalid max_running_count({}), should be greater than 0",
request.max_running_count);
- derror_f("{}", response.hint_msg);
+ LOG_ERROR_F("{}", response.hint_msg);
return false;
}
if (request.max_running_count > 0) {
@@ -3109,9 +3111,9 @@ void server_state::update_compaction_envs_on_remote_storage(start_manual_compact
app->envs[keys[idx]] = values[idx];
}
std::string new_envs = dsn::utils::kv_map_to_string(app->envs, ',', '=');
- ddebug_f("update manual compaction envs succeed: old_envs = {}, new_envs = {}",
- old_envs,
- new_envs);
+ LOG_INFO_F("update manual compaction envs succeed: old_envs = {}, new_envs = {}",
+ old_envs,
+ new_envs);
rpc.response().err = ERR_OK;
rpc.response().hint_msg = "succeed";
@@ -3135,7 +3137,7 @@ void server_state::on_query_manual_compact_status(query_manual_compact_rpc rpc)
fmt::format("app {} is {}",
app_name,
response.err == ERR_APP_NOT_EXIST ? "not existed" : "not available");
- derror_f("{}", response.hint_msg);
+ LOG_ERROR_F("{}", response.hint_msg);
return;
}
@@ -3143,11 +3145,12 @@ void server_state::on_query_manual_compact_status(query_manual_compact_rpc rpc)
if (!app->helpers->get_manual_compact_progress(total_progress)) {
response.err = ERR_INVALID_STATE;
response.hint_msg = fmt::format("app {} is not manual compaction", app_name);
- dwarn_f("{}", response.hint_msg);
+ LOG_WARNING_F("{}", response.hint_msg);
return;
}
- ddebug_f("query app {} manual compact succeed, total_progress = {}", app_name, total_progress);
+ LOG_INFO_F(
+ "query app {} manual compact succeed, total_progress = {}", app_name, total_progress);
response.err = ERR_OK;
response.hint_msg = "succeed";
response.__set_progress(total_progress);
@@ -3199,32 +3202,33 @@ void server_state::get_max_replica_count(configuration_get_max_replica_count_rpc
auto app = get_app_and_check_exist(app_name, response);
if (app == nullptr) {
response.max_replica_count = 0;
- dwarn_f("failed to get max_replica_count: app_name={}, error_code={}, hint_message={}",
- app_name,
- response.err.to_string(),
- response.hint_message);
+ LOG_WARNING_F(
+ "failed to get max_replica_count: app_name={}, error_code={}, hint_message={}",
+ app_name,
+ response.err.to_string(),
+ response.hint_message);
return;
}
if (!check_max_replica_count_consistent(app, response)) {
response.max_replica_count = 0;
- derror_f("failed to get max_replica_count: app_name={}, app_id={}, error_code={}, "
- "hint_message={}",
- app_name,
- app->app_id,
- response.err.to_string(),
- response.hint_message);
+ LOG_ERROR_F("failed to get max_replica_count: app_name={}, app_id={}, error_code={}, "
+ "hint_message={}",
+ app_name,
+ app->app_id,
+ response.err.to_string(),
+ response.hint_message);
return;
}
response.err = ERR_OK;
response.max_replica_count = app->max_replica_count;
- ddebug_f("get max_replica_count successfully: app_name={}, app_id={}, "
- "max_replica_count={}",
- app_name,
- app->app_id,
- response.max_replica_count);
+ LOG_INFO_F("get max_replica_count successfully: app_name={}, app_id={}, "
+ "max_replica_count={}",
+ app_name,
+ app->app_id,
+ response.max_replica_count);
}
// ThreadPool: THREAD_POOL_META_STATE
@@ -3243,10 +3247,11 @@ void server_state::set_max_replica_count(configuration_set_max_replica_count_rpc
app = get_app_and_check_exist(app_name, response);
if (app == nullptr) {
response.old_max_replica_count = 0;
- dwarn_f("failed to set max_replica_count: app_name={}, error_code={}, hint_message={}",
- app_name,
- response.err.to_string(),
- response.hint_message);
+ LOG_WARNING_F(
+ "failed to set max_replica_count: app_name={}, error_code={}, hint_message={}",
+ app_name,
+ response.err.to_string(),
+ response.hint_message);
return;
}
@@ -3254,12 +3259,12 @@ void server_state::set_max_replica_count(configuration_set_max_replica_count_rpc
if (!check_max_replica_count_consistent(app, response)) {
response.old_max_replica_count = 0;
- derror_f("failed to set max_replica_count: app_name={}, app_id={}, error_code={}, "
- "hint_message={}",
- app_name,
- app_id,
- response.err.to_string(),
- response.hint_message);
+ LOG_ERROR_F("failed to set max_replica_count: app_name={}, app_id={}, error_code={}, "
+ "hint_message={}",
+ app_name,
+ app_id,
+ response.err.to_string(),
+ response.hint_message);
return;
}
@@ -3268,12 +3273,12 @@ void server_state::set_max_replica_count(configuration_set_max_replica_count_rpc
if (app->status != app_status::AS_AVAILABLE) {
response.err = ERR_INVALID_PARAMETERS;
response.hint_message = fmt::format("app({}) is not in available status", app_name);
- derror_f("failed to set max_replica_count: app_name={}, app_id={}, error_code={}, "
- "hint_message={}",
- app_name,
- app_id,
- response.err.to_string(),
- response.hint_message);
+ LOG_ERROR_F("failed to set max_replica_count: app_name={}, app_id={}, error_code={}, "
+ "hint_message={}",
+ app_name,
+ app_id,
+ response.err.to_string(),
+ response.hint_message);
return;
}
}
@@ -3283,7 +3288,7 @@ void server_state::set_max_replica_count(configuration_set_max_replica_count_rpc
response.err = ERR_STATE_FREEZED;
response.hint_message =
"current meta function level is freezed, since there are too few alive nodes";
- derror_f(
+ LOG_ERROR_F(
"failed to set max_replica_count: app_name={}, app_id={}, error_code={}, message={}",
app_name,
app_id,
@@ -3294,7 +3299,7 @@ void server_state::set_max_replica_count(configuration_set_max_replica_count_rpc
if (!validate_target_max_replica_count(new_max_replica_count, response.hint_message)) {
response.err = ERR_INVALID_PARAMETERS;
- dwarn_f(
+ LOG_WARNING_F(
"failed to set max_replica_count: app_name={}, app_id={}, error_code={}, message={}",
app_name,
app_id,
@@ -3306,17 +3311,17 @@ void server_state::set_max_replica_count(configuration_set_max_replica_count_rpc
if (new_max_replica_count == response.old_max_replica_count) {
response.err = ERR_OK;
response.hint_message = "no need to update max_replica_count since it's not changed";
- dwarn_f("{}: app_name={}, app_id={}", response.hint_message, app_name, app_id);
+ LOG_WARNING_F("{}: app_name={}, app_id={}", response.hint_message, app_name, app_id);
return;
}
- ddebug_f("request for {} max_replica_count: app_name={}, app_id={}, "
- "old_max_replica_count={}, new_max_replica_count={}",
- new_max_replica_count > response.old_max_replica_count ? "increasing" : "decreasing",
- app_name,
- app_id,
- response.old_max_replica_count,
- new_max_replica_count);
+ LOG_INFO_F("request for {} max_replica_count: app_name={}, app_id={}, "
+ "old_max_replica_count={}, new_max_replica_count={}",
+ new_max_replica_count > response.old_max_replica_count ? "increasing" : "decreasing",
+ app_name,
+ app_id,
+ response.old_max_replica_count,
+ new_max_replica_count);
set_max_replica_count_env_updating(app, rpc);
}
@@ -3337,12 +3342,12 @@ void server_state::set_max_replica_count_env_updating(std::shared_ptr<app_state>
response.hint_message = fmt::format("max_replica_count of app({}) is being updated, "
"thus this request would be rejected",
app->app_name);
- derror_f("failed to set max_replica_count: app_name={}, app_id={}, error_code={}, "
- "hint_message={}",
- app->app_name,
- app->app_id,
- response.err.to_string(),
- response.hint_message);
+ LOG_ERROR_F("failed to set max_replica_count: app_name={}, app_id={}, error_code={}, "
+ "hint_message={}",
+ app->app_name,
+ app->app_id,
+ response.err.to_string(),
+ response.hint_message);
return;
}
}
@@ -3350,14 +3355,14 @@ void server_state::set_max_replica_count_env_updating(std::shared_ptr<app_state>
const auto new_max_replica_count = rpc.request().max_replica_count;
const auto old_max_replica_count = rpc.response().old_max_replica_count;
- ddebug_f("ready to update remote env of max_replica_count: app_name={}, app_id={}, "
- "old_max_replica_count={}, new_max_replica_count={}, {}={}",
- app->app_name,
- app->app_id,
- old_max_replica_count,
- new_max_replica_count,
- replica_envs::UPDATE_MAX_REPLICA_COUNT,
- app->envs[replica_envs::UPDATE_MAX_REPLICA_COUNT]);
+ LOG_INFO_F("ready to update remote env of max_replica_count: app_name={}, app_id={}, "
+ "old_max_replica_count={}, new_max_replica_count={}, {}={}",
+ app->app_name,
+ app->app_id,
+ old_max_replica_count,
+ new_max_replica_count,
+ replica_envs::UPDATE_MAX_REPLICA_COUNT,
+ app->envs[replica_envs::UPDATE_MAX_REPLICA_COUNT]);
auto ainfo = *(reinterpret_cast<app_info *>(app.get()));
ainfo.envs[replica_envs::UPDATE_MAX_REPLICA_COUNT] =
@@ -3382,13 +3387,13 @@ void server_state::set_max_replica_count_env_updating(std::shared_ptr<app_state>
app->envs[replica_envs::UPDATE_MAX_REPLICA_COUNT] =
fmt::format("updating;{}", new_max_replica_count);
- ddebug_f("both remote and local env of max_replica_count have been updated "
- "successfully: app_name={}, app_id={}, new_max_replica_count={}, {}={}",
- app->app_name,
- app->app_id,
- new_max_replica_count,
- replica_envs::UPDATE_MAX_REPLICA_COUNT,
- app->envs[replica_envs::UPDATE_MAX_REPLICA_COUNT]);
+ LOG_INFO_F("both remote and local env of max_replica_count have been updated "
+ "successfully: app_name={}, app_id={}, new_max_replica_count={}, {}={}",
+ app->app_name,
+ app->app_id,
+ new_max_replica_count,
+ replica_envs::UPDATE_MAX_REPLICA_COUNT,
+ app->envs[replica_envs::UPDATE_MAX_REPLICA_COUNT]);
}
do_update_max_replica_count(app, rpc);
@@ -3447,13 +3452,13 @@ void server_state::do_update_max_replica_count(std::shared_ptr<app_state> &app,
new_max_replica_count);
}
- ddebug_f("all partitions have been changed to the new max_replica_count, ready to update "
- "the app-level max_replica_count: app_name={}, app_id={}, partition_count={}, "
- "new_max_replica_count={}",
- app_name,
- app->app_id,
- app->partition_count,
- new_max_replica_count);
+ LOG_INFO_F("all partitions have been changed to the new max_replica_count, ready to update "
+ "the app-level max_replica_count: app_name={}, app_id={}, partition_count={}, "
+ "new_max_replica_count={}",
+ app_name,
+ app->app_id,
+ app->partition_count,
+ new_max_replica_count);
update_app_max_replica_count(app, rpc);
};
@@ -3475,14 +3480,14 @@ void server_state::update_app_max_replica_count(std::shared_ptr<app_state> &app,
const auto new_max_replica_count = rpc.request().max_replica_count;
const auto old_max_replica_count = rpc.response().old_max_replica_count;
- ddebug_f("ready to update remote app-level max_replica_count: app_name={}, app_id={}, "
- "old_max_replica_count={}, new_max_replica_count={}, {}={}",
- app->app_name,
- app->app_id,
- old_max_replica_count,
- new_max_replica_count,
- replica_envs::UPDATE_MAX_REPLICA_COUNT,
- app->envs[replica_envs::UPDATE_MAX_REPLICA_COUNT]);
+ LOG_INFO_F("ready to update remote app-level max_replica_count: app_name={}, app_id={}, "
+ "old_max_replica_count={}, new_max_replica_count={}, {}={}",
+ app->app_name,
+ app->app_id,
+ old_max_replica_count,
+ new_max_replica_count,
+ replica_envs::UPDATE_MAX_REPLICA_COUNT,
+ app->envs[replica_envs::UPDATE_MAX_REPLICA_COUNT]);
auto ainfo = *(reinterpret_cast<app_info *>(app.get()));
ainfo.max_replica_count = new_max_replica_count;
@@ -3519,13 +3524,13 @@ void server_state::update_app_max_replica_count(std::shared_ptr<app_state> &app,
app->max_replica_count = new_max_replica_count;
app->envs.erase(replica_envs::UPDATE_MAX_REPLICA_COUNT);
- ddebug_f("both remote and local app-level max_replica_count have been updated "
- "successfully: app_name={}, app_id={}, old_max_replica_count={}, "
- "new_max_replica_count={}",
- app->app_name,
- app->app_id,
- old_max_replica_count,
- new_max_replica_count);
+ LOG_INFO_F("both remote and local app-level max_replica_count have been updated "
+ "successfully: app_name={}, app_id={}, old_max_replica_count={}, "
+ "new_max_replica_count={}",
+ app->app_name,
+ app->app_id,
+ old_max_replica_count,
+ new_max_replica_count);
auto &response = rpc.response();
response.err = ERR_OK;
@@ -3551,24 +3556,24 @@ void server_state::update_partition_max_replica_count(std::shared_ptr<app_state>
const auto old_max_replica_count = old_partition_config.max_replica_count;
if (new_max_replica_count == old_max_replica_count) {
- dwarn_f("partition-level max_replica_count has been updated: app_name={}, "
- "app_id={}, partition_index={}, new_max_replica_count={}",
- app->app_name,
- app->app_id,
- partition_index,
- new_max_replica_count);
+ LOG_WARNING_F("partition-level max_replica_count has been updated: app_name={}, "
+ "app_id={}, partition_index={}, new_max_replica_count={}",
+ app->app_name,
+ app->app_id,
+ partition_index,
+ new_max_replica_count);
return;
}
auto &context = app->helpers->contexts[partition_index];
if (context.stage == config_status::pending_remote_sync) {
- ddebug_f("have to wait until another request which is syncing with remote storage "
- "is finished, then process the current request of updating max_replica_count: "
- "app_name={}, app_id={}, partition_index={}, new_max_replica_count={}",
- app->app_name,
- app->app_id,
- partition_index,
- new_max_replica_count);
+ LOG_INFO_F("have to wait until another request which is syncing with remote storage "
+ "is finished, then process the current request of updating max_replica_count: "
+ "app_name={}, app_id={}, partition_index={}, new_max_replica_count={}",
+ app->app_name,
+ app->app_id,
+ partition_index,
+ new_max_replica_count);
tasking::enqueue(
LPC_META_STATE_HIGH,
@@ -3615,16 +3620,16 @@ task_ptr server_state::update_partition_max_replica_count_on_remote(
const auto level = _meta_svc->get_function_level();
if (level <= meta_function_level::fl_blind) {
- dwarn_f("have to wait until meta level becomes more than fl_blind, then process the "
- "current request of updating max_replica_count: current_meta_level={}, "
- "app_name={}, app_id={}, partition_index={}, new_max_replica_count={}, "
- "new_ballot={}",
- _meta_function_level_VALUES_TO_NAMES.find(level)->second,
- app->app_name,
- app->app_id,
- partition_index,
- new_max_replica_count,
- new_ballot);
+ LOG_WARNING_F("have to wait until meta level becomes more than fl_blind, then process the "
+ "current request of updating max_replica_count: current_meta_level={}, "
+ "app_name={}, app_id={}, partition_index={}, new_max_replica_count={}, "
+ "new_ballot={}",
+ _meta_function_level_VALUES_TO_NAMES.find(level)->second,
+ app->app_name,
+ app->app_id,
+ partition_index,
+ new_max_replica_count,
+ new_ballot);
// NOTICE: pending_sync_task should be reassigned
return tasking::enqueue(LPC_META_STATE_HIGH,
@@ -3644,13 +3649,13 @@ task_ptr server_state::update_partition_max_replica_count_on_remote(
std::chrono::seconds(1));
}
- ddebug_f("request for updating partition-level max_replica_count on remote storage: "
- "app_name={}, app_id={}, partition_id={}, new_max_replica_count={}, new_ballot={}",
- app->app_name,
- app->app_id,
- partition_index,
- new_max_replica_count,
- new_ballot);
+ LOG_INFO_F("request for updating partition-level max_replica_count on remote storage: "
+ "app_name={}, app_id={}, partition_id={}, new_max_replica_count={}, new_ballot={}",
+ app->app_name,
+ app->app_id,
+ partition_index,
+ new_max_replica_count,
+ new_ballot);
auto partition_path = get_partition_path(gpid);
auto json_config =
@@ -3682,15 +3687,15 @@ void server_state::on_update_partition_max_replica_count_on_remote_reply(
zauto_write_lock l(_lock);
- ddebug_f("reply for updating partition-level max_replica_count on remote storage: "
- "error_code={}, app_name={}, app_id={}, partition_id={}, new_max_replica_count={}, "
- "new_ballot={}",
- ec.to_string(),
- app->app_name,
- app->app_id,
- partition_index,
- new_max_replica_count,
- new_ballot);
+ LOG_INFO_F("reply for updating partition-level max_replica_count on remote storage: "
+ "error_code={}, app_name={}, app_id={}, partition_id={}, new_max_replica_count={}, "
+ "new_ballot={}",
+ ec.to_string(),
+ app->app_name,
+ app->app_id,
+ partition_index,
+ new_max_replica_count,
+ new_ballot);
auto &context = app->helpers->contexts[partition_index];
if (ec == ERR_TIMEOUT) {
@@ -3759,14 +3764,14 @@ void server_state::update_partition_max_replica_count_locally(
old_partition_config = new_partition_config;
- ddebug_f("local partition-level max_replica_count has been changed successfully: ",
- "app_name={}, app_id={}, partition_id={}, old_partition_config={}, "
- "new_partition_config={}",
- app->app_name,
- app->app_id,
- partition_index,
- old_config_str,
- new_config_str);
+ LOG_INFO_F("local partition-level max_replica_count has been changed successfully: ",
+ "app_name={}, app_id={}, partition_id={}, old_partition_config={}, "
+ "new_partition_config={}",
+ app->app_name,
+ app->app_id,
+ partition_index,
+ old_config_str,
+ new_config_str);
}
// ThreadPool: THREAD_POOL_META_SERVER
@@ -3832,26 +3837,26 @@ void server_state::recover_all_partitions_max_replica_count(std::shared_ptr<app_
auto new_pc = app->partitions[i];
if (new_pc.max_replica_count == new_max_replica_count) {
- dwarn_f("no need to recover partition-level max_replica_count since it has been "
- "updated before: app_name={}, app_id={}, partition_index={}, "
- "partition_count={}, new_max_replica_count={}",
- app->app_name,
- app->app_id,
- i,
- app->partition_count,
- new_max_replica_count);
+ LOG_WARNING_F("no need to recover partition-level max_replica_count since it has been "
+ "updated before: app_name={}, app_id={}, partition_index={}, "
+ "partition_count={}, new_max_replica_count={}",
+ app->app_name,
+ app->app_id,
+ i,
+ app->partition_count,
+ new_max_replica_count);
continue;
}
- ddebug_f("ready to recover partition-level max_replica_count: app_name={}, app_id={}, "
- "partition_index={}, partition_count={}, old_max_replica_count={}, "
- "new_max_replica_count={}",
- app->app_name,
- app->app_id,
- i,
- app->partition_count,
- app->max_replica_count,
- new_max_replica_count);
+ LOG_INFO_F("ready to recover partition-level max_replica_count: app_name={}, app_id={}, "
+ "partition_index={}, partition_count={}, old_max_replica_count={}, "
+ "new_max_replica_count={}",
+ app->app_name,
+ app->app_id,
+ i,
+ app->partition_count,
+ app->max_replica_count,
+ new_max_replica_count);
new_pc.max_replica_count = new_max_replica_count;
++(new_pc.ballot);
@@ -3894,15 +3899,15 @@ void server_state::recover_all_partitions_max_replica_count(std::shared_ptr<app_
old_pc = new_pc;
- ddebug_f("partition-level max_replica_count has been recovered successfully: "
- "app_name={}, app_id={}, partition_index={}, partition_count={}, "
- "old_partition_config={}, new_partition_config={}",
- app->app_name,
- app->app_id,
- i,
- app->partition_count,
- old_pc_str,
- new_pc_str);
+ LOG_INFO_F("partition-level max_replica_count has been recovered successfully: "
+ "app_name={}, app_id={}, partition_index={}, partition_count={}, "
+ "old_partition_config={}, new_partition_config={}",
+ app->app_name,
+ app->app_id,
+ i,
+ app->partition_count,
+ old_pc_str,
+ new_pc_str);
},
&tracker);
}
@@ -3915,14 +3920,14 @@ void server_state::recover_app_max_replica_count(std::shared_ptr<app_state> &app
{
zauto_read_lock l(_lock);
- ddebug_f("ready to recover app-level max_replica_count: app_name={}, app_id={}, "
- "old_max_replica_count={}, new_max_replica_count={}, {}={}",
- app->app_name,
- app->app_id,
- app->max_replica_count,
- new_max_replica_count,
- replica_envs::UPDATE_MAX_REPLICA_COUNT,
- app->envs[replica_envs::UPDATE_MAX_REPLICA_COUNT]);
+ LOG_INFO_F("ready to recover app-level max_replica_count: app_name={}, app_id={}, "
+ "old_max_replica_count={}, new_max_replica_count={}, {}={}",
+ app->app_name,
+ app->app_id,
+ app->max_replica_count,
+ new_max_replica_count,
+ replica_envs::UPDATE_MAX_REPLICA_COUNT,
+ app->envs[replica_envs::UPDATE_MAX_REPLICA_COUNT]);
auto ainfo = *(reinterpret_cast<app_info *>(app.get()));
ainfo.max_replica_count = new_max_replica_count;
@@ -3950,13 +3955,13 @@ void server_state::recover_app_max_replica_count(std::shared_ptr<app_state> &app
app->max_replica_count = new_max_replica_count;
app->envs.erase(replica_envs::UPDATE_MAX_REPLICA_COUNT);
- ddebug_f("app-level max_replica_count has been recovered successfully: "
- "app_name={}, app_id={}, old_max_replica_count={}, "
- "new_max_replica_count={}",
- app->app_name,
- app->app_id,
- old_max_replica_count,
- app->max_replica_count);
+ LOG_INFO_F("app-level max_replica_count has been recovered successfully: "
+ "app_name={}, app_id={}, old_max_replica_count={}, "
+ "new_max_replica_count={}",
+ app->app_name,
+ app->app_id,
+ old_max_replica_count,
+ app->max_replica_count);
},
&tracker);
}
diff --git a/src/meta/server_state_restore.cpp b/src/meta/server_state_restore.cpp
index 8fc6b43b3..58fc62205 100644
--- a/src/meta/server_state_restore.cpp
+++ b/src/meta/server_state_restore.cpp
@@ -59,7 +59,7 @@ void server_state::sync_app_from_backup_media(
error_code err = ERR_OK;
block_file_ptr file_handle = nullptr;
- ddebug_f("start to create metadata file {}", app_metadata);
+ LOG_INFO_F("start to create metadata file {}", app_metadata);
blk_fs
->create_file(create_file_request{app_metadata, true},
TASK_CODE_EXEC_INLINED,
@@ -70,7 +70,7 @@ void server_state::sync_app_from_backup_media(
->wait();
if (err != ERR_OK) {
- derror_f("create metadata file {} failed.", app_metadata);
+ LOG_ERROR_F("create metadata file {} failed.", app_metadata);
callback_tsk->enqueue_with(err, dsn::blob());
return;
}
@@ -89,7 +89,7 @@ std::pair<dsn::error_code, std::shared_ptr<app_state>> server_state::restore_app
dsn::app_info info;
if (!::dsn::json::json_forwarder<dsn::app_info>::decode(app_info, info)) {
std::string b_str(app_info.data(), app_info.length());
- derror_f("decode app_info '{}' failed", b_str);
+ LOG_ERROR_F("decode app_info '{}' failed", b_str);
// NOTICE : maybe find a better error_code to replace err_corruption
res.first = ERR_CORRUPTION;
return res;
diff --git a/src/meta/test/meta_bulk_load_service_test.cpp b/src/meta/test/meta_bulk_load_service_test.cpp
index 047e272ed..ceb3c51f3 100644
--- a/src/meta/test/meta_bulk_load_service_test.cpp
+++ b/src/meta/test/meta_bulk_load_service_test.cpp
@@ -134,7 +134,7 @@ public:
void mock_partition_bulk_load(const std::string &app_name, const gpid &pid)
{
- ddebug_f("mock function, app({}), pid({})", app_name, pid);
+ LOG_INFO_F("mock function, app({}), pid({})", app_name, pid);
}
gpid before_check_partition_status(bulk_load_status::type status)
@@ -373,11 +373,11 @@ public:
std::move(app_path),
std::move(value),
[this, app_path, &ainfo, &partition_bulk_load_info_map]() {
- ddebug_f("create app({}) app_id={} bulk load dir({}), bulk_load_status={}",
- ainfo.app_name,
- ainfo.app_id,
- app_path,
- dsn::enum_to_string(ainfo.status));
+ LOG_INFO_F("create app({}) app_id={} bulk load dir({}), bulk_load_status={}",
+ ainfo.app_name,
+ ainfo.app_id,
+ app_path,
+ dsn::enum_to_string(ainfo.status));
for (const auto &kv : partition_bulk_load_info_map) {
mock_partition_bulk_load_info_on_remote_storage(gpid(ainfo.app_id, kv.first),
kv.second);
@@ -392,10 +392,10 @@ public:
blob value = json::json_forwarder<partition_bulk_load_info>::encode(pinfo);
_ms->get_meta_storage()->create_node(
std::move(partition_path), std::move(value), [partition_path, pid, &pinfo]() {
- ddebug_f("create partition[{}] bulk load dir({}), bulk_load_status={}",
- pid,
- partition_path,
- dsn::enum_to_string(pinfo.status));
+ LOG_INFO_F("create partition[{}] bulk load dir({}), bulk_load_status={}",
+ pid,
+ partition_path,
+ dsn::enum_to_string(pinfo.status));
});
}
@@ -407,7 +407,7 @@ public:
_ms->get_meta_storage()->create_node(
std::move(path), blob(lock_state, 0, strlen(lock_state)), [this]() {
- ddebug_f("create app root {}", _app_root);
+ LOG_INFO_F("create app root {}", _app_root);
});
wait_all();
@@ -416,7 +416,7 @@ public:
_app_root + "/" + boost::lexical_cast<std::string>(info.app_id),
std::move(value),
[this, &info]() {
- ddebug_f("create app({}) app_id={}, dir succeed", info.app_name, info.app_id);
+ LOG_INFO_F("create app({}) app_id={}, dir succeed", info.app_name, info.app_id);
for (int i = 0; i < info.partition_count; ++i) {
partition_configuration config;
config.max_replica_count = 3;
@@ -428,10 +428,10 @@ public:
boost::lexical_cast<std::string>(i),
std::move(v),
[info, i]() {
- ddebug_f("create app({}), partition({}.{}) dir succeed",
- info.app_name,
- info.app_id,
- i);
+ LOG_INFO_F("create app({}), partition({}.{}) dir succeed",
+ info.app_name,
+ info.app_id,
+ i);
});
}
});
diff --git a/src/meta/test/meta_split_service_test.cpp b/src/meta/test/meta_split_service_test.cpp
index e6a6f955f..cbc78a494 100644
--- a/src/meta/test/meta_split_service_test.cpp
+++ b/src/meta/test/meta_split_service_test.cpp
@@ -289,7 +289,7 @@ public:
_ms->get_meta_storage()->create_node(
std::move(path), blob(lock_state, 0, strlen(lock_state)), [&app_root]() {
- ddebug_f("create app root {}", app_root);
+ LOG_INFO_F("create app root {}", app_root);
});
wait_all();
@@ -309,7 +309,7 @@ public:
app_root + "/" + boost::lexical_cast<std::string>(ainfo.app_id),
std::move(value),
[this, &app_root, &ainfo]() {
- ddebug_f("create app({}) app_id={}, dir succeed", ainfo.app_name, ainfo.app_id);
+ LOG_INFO_F("create app({}) app_id={}, dir succeed", ainfo.app_name, ainfo.app_id);
for (int i = 0; i < ainfo.init_partition_count; ++i) {
create_partition_configuration_on_remote_storage(app_root, ainfo.app_id, i);
}
@@ -338,7 +338,7 @@ public:
boost::lexical_cast<std::string>(pidx),
std::move(value),
[app_id, pidx, this]() {
- ddebug_f("create app({}), partition({}.{}) dir succeed", NAME, app_id, pidx);
+ LOG_INFO_F("create app({}), partition({}.{}) dir succeed", NAME, app_id, pidx);
});
}
diff --git a/src/meta/test/meta_test_base.cpp b/src/meta/test/meta_test_base.cpp
index b70e649b4..438c8e824 100644
--- a/src/meta/test/meta_test_base.cpp
+++ b/src/meta/test/meta_test_base.cpp
@@ -127,9 +127,9 @@ std::vector<rpc_address> meta_test_base::ensure_enough_alive_nodes(int min_node_
min_node_count,
node_count);
- dinfo_f("already exists {} alive nodes: ", nodes.size());
+ LOG_DEBUG_F("already exists {} alive nodes: ", nodes.size());
for (const auto &node : nodes) {
- dinfo_f(" {}", node.to_string());
+ LOG_DEBUG_F(" {}", node.to_string());
}
// ensure that _ms->_alive_set is identical with _ss->_nodes
@@ -151,9 +151,9 @@ std::vector<rpc_address> meta_test_base::ensure_enough_alive_nodes(int min_node_
std::this_thread::sleep_for(std::chrono::milliseconds(5));
}
- dinfo_f("created {} alive nodes: ", nodes.size());
+ LOG_DEBUG_F("created {} alive nodes: ", nodes.size());
for (const auto &node : nodes) {
- dinfo_f(" {}", node.to_string());
+ LOG_DEBUG_F(" {}", node.to_string());
}
return nodes;
}
diff --git a/src/redis_protocol/proxy_lib/redis_parser.cpp b/src/redis_protocol/proxy_lib/redis_parser.cpp
index cb297c9c6..f4607d4c4 100644
--- a/src/redis_protocol/proxy_lib/redis_parser.cpp
+++ b/src/redis_protocol/proxy_lib/redis_parser.cpp
@@ -163,7 +163,7 @@ bool redis_parser::eat(char c)
--_total_length;
return true;
} else {
- derror_f("{}: expect token: {}, got {}", _remote_address.to_string(), c, peek());
+ LOG_ERROR_F("{}: expect token: {}, got {}", _remote_address.to_string(), c, peek());
return false;
}
}
@@ -189,14 +189,14 @@ bool redis_parser::end_array_size()
{
int32_t count = 0;
if (dsn_unlikely(!dsn::buf2int32(dsn::string_view(_current_size), count))) {
- derror_f(
+ LOG_ERROR_F(
"{}: invalid size string \"{}\"", _remote_address.to_string(), _current_size.c_str());
return false;
}
if (dsn_unlikely(count <= 0)) {
- derror_f("{}: array size should be positive in redis request, but got {}",
- _remote_address.to_string(),
- count);
+ LOG_ERROR_F("{}: array size should be positive in redis request, but got {}",
+ _remote_address.to_string(),
+ count);
return false;
}
@@ -228,7 +228,7 @@ bool redis_parser::end_bulk_string_size()
int32_t length = 0;
if (dsn_unlikely(!dsn::buf2int32(
dsn::string_view(_current_size.c_str(), _current_size.length()), length))) {
- derror_f(
+ LOG_ERROR_F(
"{}: invalid size string \"{}\"", _remote_address.to_string(), _current_size.c_str());
return false;
}
@@ -247,7 +247,7 @@ bool redis_parser::end_bulk_string_size()
return true;
}
- derror_f(
+ LOG_ERROR_F(
"{}: invalid bulk string length: {}", _remote_address.to_string(), _current_str.length);
return false;
}
@@ -257,7 +257,7 @@ void redis_parser::append_message(dsn::message_ex *msg)
msg->add_ref();
_recv_buffers.push(msg);
_total_length += msg->body_size();
- dinfo_f(
+ LOG_DEBUG_F(
"{}: recv message, currently total length: {}", _remote_address.to_string(), _total_length);
}
@@ -408,7 +408,7 @@ void redis_parser::default_handler(redis_parser::message_entry &entry)
{
::dsn::blob &cmd = entry.request.sub_requests[0].data;
std::string message = "unknown command '" + std::string(cmd.data(), cmd.length()) + "'";
- ddebug_f("{}: {} with seqid {}", _remote_address.to_string(), message, entry.sequence_id);
+ LOG_INFO_F("{}: {} with seqid {}", _remote_address.to_string(), message, entry.sequence_id);
simple_error_reply(entry, message);
}
@@ -833,12 +833,12 @@ void redis_parser::geo_radius(message_entry &entry)
double lng_degrees = 0.0;
const std::string &str_lng_degrees = redis_request.sub_requests[2].data.to_string();
if (!dsn::buf2double(str_lng_degrees, lng_degrees)) {
- dwarn_f("longitude parameter '{}' is error, use {}", str_lng_degrees, lng_degrees);
+ LOG_WARNING_F("longitude parameter '{}' is error, use {}", str_lng_degrees, lng_degrees);
}
double lat_degrees = 0.0;
const std::string &str_lat_degrees = redis_request.sub_requests[3].data.to_string();
if (!dsn::buf2double(str_lat_degrees, lat_degrees)) {
- dwarn_f("latitude parameter '{}' is error, use {}", str_lat_degrees, lat_degrees);
+ LOG_WARNING_F("latitude parameter '{}' is error, use {}", str_lat_degrees, lat_degrees);
}
// radius m|km|ft|mi [WITHCOORD] [WITHDIST] [COUNT count] [ASC|DESC]
@@ -940,36 +940,36 @@ void redis_parser::counter_internal(message_entry &entry)
int64_t increment = 1;
if (strcasecmp(command, "INCR") == 0 || strcasecmp(command, "DECR") == 0) {
if (entry.request.sub_requests.size() != 2) {
- dwarn_f("{}: command {} seqid({}) with invalid arguments count: {}",
- _remote_address.to_string(),
- command,
- entry.sequence_id,
- entry.request.sub_requests.size());
+ LOG_WARNING_F("{}: command {} seqid({}) with invalid arguments count: {}",
+ _remote_address.to_string(),
+ command,
+ entry.sequence_id,
+ entry.request.sub_requests.size());
simple_error_reply(entry, fmt::format("wrong number of arguments for '{}'", command));
return;
}
} else if (strcasecmp(command, "INCRBY") == 0 || strcasecmp(command, "DECRBY") == 0) {
if (entry.request.sub_requests.size() != 3) {
- dwarn_f("{}: command {} seqid({}) with invalid arguments count: {}",
- _remote_address.to_string(),
- command,
- entry.sequence_id,
- entry.request.sub_requests.size());
+ LOG_WARNING_F("{}: command {} seqid({}) with invalid arguments count: {}",
+ _remote_address.to_string(),
+ command,
+ entry.sequence_id,
+ entry.request.sub_requests.size());
simple_error_reply(entry, fmt::format("wrong number of arguments for '{}'", command));
return;
}
if (!dsn::buf2int64(entry.request.sub_requests[2].data, increment)) {
- dwarn_f("{}: command {} seqid({}) with invalid 'increment': {}",
- _remote_address.to_string(),
- command,
- entry.sequence_id,
- entry.request.sub_requests[2].data.to_string());
+ LOG_WARNING_F("{}: command {} seqid({}) with invalid 'increment': {}",
+ _remote_address.to_string(),
+ command,
+ entry.sequence_id,
+ entry.request.sub_requests[2].data.to_string());
simple_error_reply(entry,
fmt::format("wrong type of argument 'increment 'for '{}'", command));
return;
}
} else {
- dfatal_f("command not support: {}", command);
+ LOG_FATAL_F("command not support: {}", command);
}
if (strncasecmp(command, "DECR", 4) == 0) {
increment = -increment;
@@ -979,19 +979,19 @@ void redis_parser::counter_internal(message_entry &entry)
auto on_incr_reply = [ref_this, this, command, &entry](
::dsn::error_code ec, dsn::message_ex *, dsn::message_ex *response) {
if (_is_session_reset.load(std::memory_order_acquire)) {
- dwarn_f("{}: command {} seqid({}) got reply, but session has reset",
- _remote_address.to_string(),
- command,
- entry.sequence_id);
+ LOG_WARNING_F("{}: command {} seqid({}) got reply, but session has reset",
+ _remote_address.to_string(),
+ command,
+ entry.sequence_id);
return;
}
if (::dsn::ERR_OK != ec) {
- dwarn_f("{}: command {} seqid({}) got reply with error = {}",
- _remote_address.to_string(),
- command,
- entry.sequence_id,
- ec.to_string());
+ LOG_WARNING_F("{}: command {} seqid({}) got reply with error = {}",
+ _remote_address.to_string(),
+ command,
+ entry.sequence_id,
+ ec.to_string());
simple_error_reply(entry, ec.to_string());
} else {
::dsn::apps::incr_response incr_resp;
@@ -1019,10 +1019,10 @@ void redis_parser::parse_set_parameters(const std::vector<redis_bulk_string> &op
if (strcasecmp(opt.c_str(), "EX") == 0 && i + 1 < opts.size()) {
const std::string &str_ttl_seconds = opts[i + 1].data.to_string();
if (!dsn::buf2int32(str_ttl_seconds, ttl_seconds)) {
- dwarn_f("'EX {}' option is error, use {}", str_ttl_seconds, ttl_seconds);
+ LOG_WARNING_F("'EX {}' option is error, use {}", str_ttl_seconds, ttl_seconds);
}
} else {
- dwarn_f("only 'EX' option is supported");
+ LOG_WARNING_F("only 'EX' option is supported");
}
}
}
@@ -1043,7 +1043,7 @@ void redis_parser::parse_geo_radius_parameters(const std::vector<redis_bulk_stri
}
const std::string &str_radius = opts[base_index++].data.to_string();
if (!dsn::buf2double(str_radius, radius_m)) {
- dwarn_f("radius parameter '{}' is error, use {}", str_radius, radius_m);
+ LOG_WARNING_F("radius parameter '{}' is error, use {}", str_radius, radius_m);
}
// m|km|ft|mi
@@ -1075,7 +1075,7 @@ void redis_parser::parse_geo_radius_parameters(const std::vector<redis_bulk_stri
} else if (strcasecmp(opt.c_str(), "COUNT") == 0 && base_index + 1 < opts.size()) {
const std::string &str_count = opts[base_index + 1].data.to_string();
if (!dsn::buf2int32(str_count, count)) {
- derror_f("'COUNT {}' option is error, use {}", str_count, count);
+ LOG_ERROR_F("'COUNT {}' option is error, use {}", str_count, count);
}
} else if (strcasecmp(opt.c_str(), "ASC") == 0) {
sort_type = geo::geo_client::SortType::asc;
diff --git a/src/replica/backup/replica_backup_manager.cpp b/src/replica/backup/replica_backup_manager.cpp
index ebbf48cdd..1573975ce 100644
--- a/src/replica/backup/replica_backup_manager.cpp
+++ b/src/replica/backup/replica_backup_manager.cpp
@@ -46,7 +46,7 @@ static bool get_policy_checkpoint_dirs(const std::string &dir,
// list sub dirs
std::vector<std::string> sub_dirs;
if (!utils::filesystem::get_subdirectories(dir, sub_dirs, false)) {
- derror_f("list sub dirs of dir {} failed", dir.c_str());
+ LOG_ERROR_F("list sub dirs of dir {} failed", dir.c_str());
return false;
}
diff --git a/src/replica/backup/replica_backup_server.cpp b/src/replica/backup/replica_backup_server.cpp
index c16f97c7e..6edb01a14 100644
--- a/src/replica/backup/replica_backup_server.cpp
+++ b/src/replica/backup/replica_backup_server.cpp
@@ -72,9 +72,9 @@ void replica_backup_server::on_cold_backup(backup_rpc rpc)
void replica_backup_server::on_clear_cold_backup(const backup_clear_request &request)
{
- ddebug_f("receive clear cold backup request: backup({}.{})",
- request.pid.to_string(),
- request.policy_name.c_str());
+ LOG_INFO_F("receive clear cold backup request: backup({}.{})",
+ request.pid.to_string(),
+ request.policy_name.c_str());
replica_ptr rep = _stub->get_replica(request.pid);
if (rep != nullptr) {
diff --git a/src/replica/disk_cleaner.cpp b/src/replica/disk_cleaner.cpp
index f54b22f51..5b2d9227a 100644
--- a/src/replica/disk_cleaner.cpp
+++ b/src/replica/disk_cleaner.cpp
@@ -68,7 +68,7 @@ error_s disk_remove_useless_dirs(const std::vector<std::string> &data_dirs,
for (auto &dir : data_dirs) {
std::vector<std::string> tmp_list;
if (!dsn::utils::filesystem::get_subdirectories(dir, tmp_list, false)) {
- dwarn_f("gc_disk: failed to get subdirectories in {}", dir);
+ LOG_WARNING_F("gc_disk: failed to get subdirectories in {}", dir);
return error_s::make(ERR_OBJECT_NOT_FOUND, "failed to get subdirectories");
}
sub_list.insert(sub_list.end(), tmp_list.begin(), tmp_list.end());
@@ -82,7 +82,7 @@ error_s disk_remove_useless_dirs(const std::vector<std::string> &data_dirs,
time_t mt;
if (!dsn::utils::filesystem::last_write_time(fpath, mt)) {
- dwarn_f("gc_disk: failed to get last write time of {}", fpath);
+ LOG_WARNING_F("gc_disk: failed to get last write time of {}", fpath);
continue;
}
@@ -107,20 +107,20 @@ error_s disk_remove_useless_dirs(const std::vector<std::string> &data_dirs,
if (last_write_time + remove_interval_seconds <= current_time_ms / 1000) {
if (!dsn::utils::filesystem::remove_path(fpath)) {
- dwarn_f("gc_disk: failed to delete directory '{}', time_used_ms = {}",
- fpath,
- dsn_now_ms() - current_time_ms);
+ LOG_WARNING_F("gc_disk: failed to delete directory '{}', time_used_ms = {}",
+ fpath,
+ dsn_now_ms() - current_time_ms);
} else {
- dwarn_f("gc_disk: replica_dir_op succeed to delete directory '{}'"
- ", time_used_ms = {}",
- fpath,
- dsn_now_ms() - current_time_ms);
+ LOG_WARNING_F("gc_disk: replica_dir_op succeed to delete directory '{}'"
+ ", time_used_ms = {}",
+ fpath,
+ dsn_now_ms() - current_time_ms);
report.remove_dir_count++;
}
} else {
- ddebug_f("gc_disk: reserve directory '{}', wait_seconds = {}",
- fpath,
- last_write_time + remove_interval_seconds - current_time_ms / 1000);
+ LOG_INFO_F("gc_disk: reserve directory '{}', wait_seconds = {}",
+ fpath,
+ last_write_time + remove_interval_seconds - current_time_ms / 1000);
}
}
return error_s::ok();
diff --git a/src/replica/duplication/duplication_sync_timer.cpp b/src/replica/duplication/duplication_sync_timer.cpp
index 6dde87cc3..a3520403d 100644
--- a/src/replica/duplication/duplication_sync_timer.cpp
+++ b/src/replica/duplication/duplication_sync_timer.cpp
@@ -35,15 +35,16 @@ void duplication_sync_timer::run()
{
// ensure duplication sync never be concurrent
if (_rpc_task) {
- ddebug_f("a duplication sync is already ongoing");
+ LOG_INFO_F("a duplication sync is already ongoing");
return;
}
{
zauto_lock l(_stub->_state_lock);
if (_stub->_state == replica_stub::NS_Disconnected) {
- ddebug_f("stop this round of duplication sync because this server is disconnected from "
- "meta server");
+ LOG_INFO_F(
+ "stop this round of duplication sync because this server is disconnected from "
+ "meta server");
return;
}
}
@@ -64,7 +65,7 @@ void duplication_sync_timer::run()
duplication_sync_rpc rpc(std::move(req), RPC_CM_DUPLICATION_SYNC, 3_s);
rpc_address meta_server_address(_stub->get_meta_server_address());
- ddebug_f("duplication_sync to meta({})", meta_server_address.to_string());
+ LOG_INFO_F("duplication_sync to meta({})", meta_server_address.to_string());
zauto_lock l(_lock);
_rpc_task =
@@ -80,7 +81,7 @@ void duplication_sync_timer::on_duplication_sync_reply(error_code err,
err = resp.err;
}
if (err != ERR_OK) {
- derror_f("on_duplication_sync_reply: err({})", err.to_string());
+ LOG_ERROR_F("on_duplication_sync_reply: err({})", err.to_string());
} else {
update_duplication_map(resp.dup_map);
}
@@ -156,7 +157,7 @@ void duplication_sync_timer::close()
void duplication_sync_timer::start()
{
- ddebug_f("run duplication sync periodically in {}s", DUPLICATION_SYNC_PERIOD_SECOND);
+ LOG_INFO_F("run duplication sync periodically in {}s", DUPLICATION_SYNC_PERIOD_SECOND);
_timer_task = tasking::enqueue_timer(LPC_DUPLICATION_SYNC_TIMER,
&_stub->_tracker,
diff --git a/src/replica/duplication/load_from_private_log.cpp b/src/replica/duplication/load_from_private_log.cpp
index 8af6145a1..cd66b41fc 100644
--- a/src/replica/duplication/load_from_private_log.cpp
+++ b/src/replica/duplication/load_from_private_log.cpp
@@ -59,7 +59,7 @@ bool load_from_private_log::switch_to_next_log_file()
start_from_log_file(file);
return true;
} else {
- ddebug_f("no next log file (log.{}) is found", _current->index() + 1);
+ LOG_INFO_F("no next log file (log.{}) is found", _current->index() + 1);
_current = nullptr;
return false;
}
diff --git a/src/replica/duplication/test/load_from_private_log_test.cpp b/src/replica/duplication/test/load_from_private_log_test.cpp
index 788e97902..bf5f09684 100644
--- a/src/replica/duplication/test/load_from_private_log_test.cpp
+++ b/src/replica/duplication/test/load_from_private_log_test.cpp
@@ -223,7 +223,7 @@ public:
if (err == ERR_OK) {
break;
}
- derror_f("mlog open failed, encountered error: {}", err);
+ LOG_ERROR_F("mlog open failed, encountered error: {}", err);
}
return mlog;
}
diff --git a/src/replica/mutation_log.cpp b/src/replica/mutation_log.cpp
index 8bc7dff00..d32899547 100644
--- a/src/replica/mutation_log.cpp
+++ b/src/replica/mutation_log.cpp
@@ -947,9 +947,9 @@ error_code mutation_log::reset_from(const std::string &dir,
// make sure logs in `dir` (such as /learn) are valid.
error_s es = log_utils::check_log_files_continuity(dir);
if (!es.is_ok()) {
- derror_f("the log of source dir {} is invalid:{}, will remove it.", dir, es);
+ LOG_ERROR_F("the log of source dir {} is invalid:{}, will remove it.", dir, es);
if (!utils::filesystem::remove_path(dir)) {
- derror_f("remove {} failed", dir);
+ LOG_ERROR_F("remove {} failed", dir);
return err;
}
return es.code();
@@ -957,10 +957,10 @@ error_code mutation_log::reset_from(const std::string &dir,
std::string temp_dir = _dir + '.' + std::to_string(dsn_now_ns());
if (!utils::filesystem::rename_path(_dir, temp_dir)) {
- derror_f("rename {} to {} failed", _dir, temp_dir);
+ LOG_ERROR_F("rename {} to {} failed", _dir, temp_dir);
return err;
}
- ddebug_f("moved current log dir {} to tmp_dir {}", _dir, temp_dir);
+ LOG_INFO_F("moved current log dir {} to tmp_dir {}", _dir, temp_dir);
// define `defer` for rollback temp_dir when failed or remove temp_dir when success
auto temp_dir_resolve = dsn::defer([this, err, temp_dir]() {
if (err != ERR_OK) {
@@ -972,24 +972,24 @@ error_code mutation_log::reset_from(const std::string &dir,
} else {
if (!dsn::utils::filesystem::remove_path(temp_dir)) {
// temp dir allow delete failed, it's only garbage
- derror_f("remove temp dir {} failed", temp_dir);
+ LOG_ERROR_F("remove temp dir {} failed", temp_dir);
}
}
});
// move source dir to target dir
if (!utils::filesystem::rename_path(dir, _dir)) {
- derror_f("rename {} to {} failed", dir, _dir);
+ LOG_ERROR_F("rename {} to {} failed", dir, _dir);
return err;
}
- ddebug_f("move {} to {} as our new log directory", dir, _dir);
+ LOG_INFO_F("move {} to {} as our new log directory", dir, _dir);
// - make sure logs in moved dir(such as /plog) are valid and can be opened successfully.
// - re-open new log files for loading the new log file and register the files into replica,
// please make sure the old log files has been closed
err = open(replay_error_callback, write_error_callback);
if (err != ERR_OK) {
- derror_f("the logs of moved dir {} are invalid and open failed:{}", _dir, err);
+ LOG_ERROR_F("the logs of moved dir {} are invalid and open failed:{}", _dir, err);
}
return err;
}
@@ -1106,11 +1106,11 @@ bool mutation_log::get_learn_state(gpid gpid, decree start, /*out*/ learn_state
if (state.meta.length() == 0 && start > _private_log_info.max_decree) {
// no memory data and no disk data
- ddebug_f("gpid({}) get_learn_state returns false"
- "learn_start_decree={}, max_decree_in_private_log={}",
- gpid,
- start,
- _private_log_info.max_decree);
+ LOG_INFO_F("gpid({}) get_learn_state returns false"
+ "learn_start_decree={}, max_decree_in_private_log={}",
+ gpid,
+ start,
+ _private_log_info.max_decree);
return false;
}
@@ -1339,25 +1339,25 @@ int mutation_log::garbage_collection(gpid gpid,
// log is invalid, ok to delete
else if (valid_start_offset >= log->end_offset()) {
- ddebug_f("gc_private @ {}: will remove files {} ~ log.{} because "
- "valid_start_offset={} outdates log_end_offset={}",
- _private_gpid,
- files.begin()->second->path(),
- log->index(),
- valid_start_offset,
- log->end_offset());
+ LOG_INFO_F("gc_private @ {}: will remove files {} ~ log.{} because "
+ "valid_start_offset={} outdates log_end_offset={}",
+ _private_gpid,
+ files.begin()->second->path(),
+ log->index(),
+ valid_start_offset,
+ log->end_offset());
break;
}
// all mutations are cleanable, ok to delete
else if (cleanable_decree >= max_decree) {
- ddebug_f("gc_private @ {}: will remove files {} ~ log.{} because "
- "cleanable_decree={} outdates max_decree={}",
- _private_gpid,
- files.begin()->second->path(),
- log->index(),
- cleanable_decree,
- max_decree);
+ LOG_INFO_F("gc_private @ {}: will remove files {} ~ log.{} because "
+ "cleanable_decree={} outdates max_decree={}",
+ _private_gpid,
+ files.begin()->second->path(),
+ log->index(),
+ cleanable_decree,
+ max_decree);
break;
}
@@ -1397,7 +1397,7 @@ int mutation_log::garbage_collection(gpid gpid,
}
// delete succeed
- ddebug_f("gc_private @ {}: log file {} is removed", _private_gpid, fpath);
+ LOG_INFO_F("gc_private @ {}: log file {} is removed", _private_gpid, fpath);
deleted++;
// erase from _log_files
diff --git a/src/replica/mutation_log_replay.cpp b/src/replica/mutation_log_replay.cpp
index b34944e20..cc1552cd8 100644
--- a/src/replica/mutation_log_replay.cpp
+++ b/src/replica/mutation_log_replay.cpp
@@ -155,7 +155,7 @@ namespace replication {
error_s error = log_utils::check_log_files_continuity(logs);
if (!error.is_ok()) {
- derror_f("check_log_files_continuity failed: {}", error);
+ LOG_ERROR_F("check_log_files_continuity failed: {}", error);
return error.code();
}
diff --git a/src/replica/replica_2pc.cpp b/src/replica/replica_2pc.cpp
index e40ec6d3d..cb22edefa 100644
--- a/src/replica/replica_2pc.cpp
+++ b/src/replica/replica_2pc.cpp
@@ -82,10 +82,10 @@ void replica::on_client_write(dsn::message_ex *request, bool ignore_throttling)
task_spec *spec = task_spec::get(request->rpc_code());
if (dsn_unlikely(nullptr == spec || request->rpc_code() == TASK_CODE_INVALID)) {
- derror_f("recv message with unhandled rpc name {} from {}, trace_id = {}",
- request->rpc_code().to_string(),
- request->header->from_address.to_string(),
- request->header->trace_id);
+ LOG_ERROR_F("recv message with unhandled rpc name {} from {}, trace_id = {}",
+ request->rpc_code().to_string(),
+ request->header->from_address.to_string(),
+ request->header->trace_id);
response_client_write(request, ERR_HANDLER_NOT_FOUND);
return;
}
diff --git a/src/replica/replica_context.cpp b/src/replica/replica_context.cpp
index 9d7cec2d0..36bb48498 100644
--- a/src/replica/replica_context.cpp
+++ b/src/replica/replica_context.cpp
@@ -182,9 +182,9 @@ bool primary_context::secondary_disk_space_insufficient() const
{
for (const auto &kv : secondary_disk_status) {
if (kv.second == disk_status::SPACE_INSUFFICIENT) {
- ddebug_f("partition[{}] secondary[{}] disk space is insufficient",
- membership.pid,
- kv.first.to_string());
+ LOG_INFO_F("partition[{}] secondary[{}] disk space is insufficient",
+ membership.pid,
+ kv.first.to_string());
return true;
}
}
diff --git a/src/replica/replica_init.cpp b/src/replica/replica_init.cpp
index 024371002..f9b57d916 100644
--- a/src/replica/replica_init.cpp
+++ b/src/replica/replica_init.cpp
@@ -83,25 +83,25 @@ error_code replica::initialize_on_new()
new replica(stub, gpid, app, dir.c_str(), restore_if_necessary, is_duplication_follower);
error_code err;
if (restore_if_necessary && (err = rep->restore_checkpoint()) != dsn::ERR_OK) {
- derror_f("{}: try to restore replica failed, error({})", rep->name(), err.to_string());
+ LOG_ERROR_F("{}: try to restore replica failed, error({})", rep->name(), err.to_string());
return clear_on_failure(stub, rep, dir, gpid);
}
if (is_duplication_follower &&
(err = rep->get_replica_follower()->duplicate_checkpoint()) != dsn::ERR_OK) {
- derror_f("{}: try to duplicate replica checkpoint failed, error({}) and please check "
- "previous detail error log",
- rep->name(),
- err.to_string());
+ LOG_ERROR_F("{}: try to duplicate replica checkpoint failed, error({}) and please check "
+ "previous detail error log",
+ rep->name(),
+ err.to_string());
return clear_on_failure(stub, rep, dir, gpid);
}
err = rep->initialize_on_new();
if (err == ERR_OK) {
- dinfo_f("{}: new replica succeed", rep->name());
+ LOG_DEBUG_F("{}: new replica succeed", rep->name());
return rep;
} else {
- derror_f("{}: new replica failed, err = {}", rep->name(), err.to_string());
+ LOG_ERROR_F("{}: new replica failed, err = {}", rep->name(), err.to_string());
return clear_on_failure(stub, rep, dir, gpid);
}
}
@@ -172,10 +172,11 @@ error_code replica::initialize_on_load()
}
if (info.partition_count < pidx) {
- derror_f("partition[{}], count={}, this replica may be partition split garbage partition, "
- "ignore it",
- pid,
- info.partition_count);
+ LOG_ERROR_F(
+ "partition[{}], count={}, this replica may be partition split garbage partition, "
+ "ignore it",
+ pid,
+ info.partition_count);
return nullptr;
}
diff --git a/src/replica/replica_restore.cpp b/src/replica/replica_restore.cpp
index 690d220c9..a34973686 100644
--- a/src/replica/replica_restore.cpp
+++ b/src/replica/replica_restore.cpp
@@ -235,7 +235,7 @@ void replica::clear_restore_useless_files(const std::string &local_chkpt_dir,
dsn::error_code replica::find_valid_checkpoint(const configuration_restore_request &req,
std::string &remote_chkpt_dir)
{
- ddebug_f("{}: start to find valid checkpoint of backup_id {}", name(), req.time_stamp);
+ LOG_INFO_F("{}: start to find valid checkpoint of backup_id {}", name(), req.time_stamp);
// we should base on old gpid to combine the path on cold backup media
dsn::gpid old_gpid;
@@ -255,9 +255,9 @@ dsn::error_code replica::find_valid_checkpoint(const configuration_restore_reque
block_filesystem *fs =
_stub->_block_service_manager.get_or_create_block_filesystem(req.backup_provider_name);
if (fs == nullptr) {
- derror_f("{}: get block filesystem by provider {} failed",
- std::string(name()),
- req.backup_provider_name);
+ LOG_ERROR_F("{}: get block filesystem by provider {} failed",
+ std::string(name()),
+ req.backup_provider_name);
return ERR_CORRUPTION;
}
@@ -270,9 +270,9 @@ dsn::error_code replica::find_valid_checkpoint(const configuration_restore_reque
->wait();
if (create_response.err != dsn::ERR_OK) {
- derror_f("{}: create file of block_service failed, reason {}",
- name(),
- create_response.err.to_string());
+ LOG_ERROR_F("{}: create file of block_service failed, reason {}",
+ name(),
+ create_response.err.to_string());
return create_response.err;
}
@@ -286,15 +286,15 @@ dsn::error_code replica::find_valid_checkpoint(const configuration_restore_reque
->wait();
if (r.err != dsn::ERR_OK) {
- derror_f("{}: read file {} failed, reason {}",
- name(),
- create_response.file_handle->file_name(),
- r.err.to_string());
+ LOG_ERROR_F("{}: read file {} failed, reason {}",
+ name(),
+ create_response.file_handle->file_name(),
+ r.err.to_string());
return r.err;
}
std::string valid_chkpt_entry(r.buffer.data(), r.buffer.length());
- ddebug_f("{}: got a valid chkpt {}", name(), valid_chkpt_entry);
+ LOG_INFO_F("{}: got a valid chkpt {}", name(), valid_chkpt_entry);
remote_chkpt_dir = ::dsn::utils::filesystem::path_combine(
cold_backup::get_replica_backup_path(backup_root, req.app_name, old_gpid, backup_id),
valid_chkpt_entry);
@@ -353,14 +353,14 @@ dsn::error_code replica::restore_checkpoint()
restore_req.__set_restore_path(iter->second);
}
- ddebug_f("{}: restore checkpoint(policy_name {}, backup_id {}), restore_path({}) from {} to "
- "local dir {}",
- name(),
- restore_req.policy_name,
- restore_req.time_stamp,
- restore_req.restore_path,
- restore_req.backup_provider_name,
- _dir);
+ LOG_INFO_F("{}: restore checkpoint(policy_name {}, backup_id {}), restore_path({}) from {} to "
+ "local dir {}",
+ name(),
+ restore_req.policy_name,
+ restore_req.time_stamp,
+ restore_req.restore_path,
+ restore_req.backup_provider_name,
+ _dir);
// then create a local restore dir if it doesn't exist
if (!utils::filesystem::directory_exists(_dir) && !utils::filesystem::create_directory(_dir)) {
@@ -373,7 +373,7 @@ dsn::error_code replica::restore_checkpoint()
std::string restore_dir = os.str();
if (!utils::filesystem::directory_exists(restore_dir) &&
!utils::filesystem::create_directory(restore_dir)) {
- derror_f("create restore dir {} failed", restore_dir);
+ LOG_ERROR_F("create restore dir {} failed", restore_dir);
return ERR_FILE_OPERATION_FAILED;
}
diff --git a/src/replica/replica_stub.cpp b/src/replica/replica_stub.cpp
index 8ba4f16ce..4345ddd05 100644
--- a/src/replica/replica_stub.cpp
+++ b/src/replica/replica_stub.cpp
@@ -550,7 +550,7 @@ void replica_stub::initialize(const replication_options &opts, bool clear /* = f
uint64_t start_time = dsn_now_ms();
for (auto &dir : dir_list) {
if (dsn::replication::is_data_dir_invalid(dir)) {
- ddebug_f("ignore dir {}", dir);
+ LOG_INFO_F("ignore dir {}", dir);
continue;
}
@@ -677,7 +677,7 @@ void replica_stub::initialize(const replication_options &opts, bool clear /* = f
pmax_commit = it->second->private_log()->max_commit_on_disk();
}
- ddebug_f(
+ LOG_INFO_F(
"{}: load replica done, err = {}, durable = {}, committed = {}, "
"prepared = {}, ballot = {}, "
"valid_offset_in_plog = {}, max_decree_in_plog = {}, max_commit_on_disk_in_plog = {}, "
@@ -771,13 +771,13 @@ void replica_stub::initialize_fs_manager(std::vector<std::string> &data_dirs,
if (dsn_unlikely(!utils::filesystem::create_directory(dir, cdir, err_msg) ||
!utils::filesystem::check_dir_rw(dir, err_msg))) {
if (FLAGS_ignore_broken_disk) {
- dwarn_f("data dir[{}] is broken, ignore it, error:{}", dir, err_msg);
+ LOG_WARNING_F("data dir[{}] is broken, ignore it, error:{}", dir, err_msg);
} else {
dassert_f(false, "{}", err_msg);
}
continue;
}
- ddebug_f("data_dirs[{}] = {}", count, cdir);
+ LOG_INFO_F("data_dirs[{}] = {}", count, cdir);
available_dirs.emplace_back(cdir);
available_dir_tags.emplace_back(data_dir_tags[i]);
count++;
@@ -1179,7 +1179,7 @@ void replica_stub::on_add_new_disk(add_new_disk_rpc rpc)
return;
}
- ddebug_f("Add a new disk in fs_manager, data_dir={}, tag={}", cdir, data_dir_tags[i]);
+ LOG_INFO_F("Add a new disk in fs_manager, data_dir={}, tag={}", cdir, data_dir_tags[i]);
_fs_manager.add_new_dir_node(cdir, data_dir_tags[i]);
}
}
@@ -1444,10 +1444,10 @@ void replica_stub::on_node_query_reply(error_code err,
return;
}
- ddebug_f("process query node partitions response for resp.err = ERR_OK, "
- "partitions_count({}), gc_replicas_count({})",
- resp.partitions.size(),
- resp.gc_replicas.size());
+ LOG_INFO_F("process query node partitions response for resp.err = ERR_OK, "
+ "partitions_count({}), gc_replicas_count({})",
+ resp.partitions.size(),
+ resp.gc_replicas.size());
replicas rs;
{
@@ -2034,10 +2034,11 @@ void replica_stub::open_replica(
id,
false);
if (!origin_tmp_dir.empty()) {
- ddebug_f("mark the dir {} is garbage, start revert and load disk migration origin "
- "replica data({})",
- dir,
- origin_tmp_dir);
+ LOG_INFO_F(
+ "mark the dir {} is garbage, start revert and load disk migration origin "
+ "replica data({})",
+ dir,
+ origin_tmp_dir);
dsn::utils::filesystem::rename_path(dir,
fmt::format("{}{}", dir, kFolderSuffixGar));
@@ -2763,7 +2764,7 @@ static int64_t get_tcmalloc_numeric_property(const char *prop)
{
size_t value;
if (!::MallocExtension::instance()->GetNumericProperty(prop, &value)) {
- derror_f("Failed to get tcmalloc property {}", prop);
+ LOG_ERROR_F("Failed to get tcmalloc property {}", prop);
return -1;
}
return value;
@@ -2779,7 +2780,7 @@ uint64_t replica_stub::gc_tcmalloc_memory(bool release_all)
}
if (_is_releasing_memory.load()) {
- dwarn_f("This node is releasing memory...");
+ LOG_WARNING_F("This node is releasing memory...");
return tcmalloc_released_bytes;
}
@@ -2797,7 +2798,7 @@ uint64_t replica_stub::gc_tcmalloc_memory(bool release_all)
if (reserved_bytes > max_reserved_bytes) {
int64_t release_bytes = reserved_bytes - max_reserved_bytes;
tcmalloc_released_bytes = release_bytes;
- ddebug_f("Memory release started, almost {} bytes will be released", release_bytes);
+ LOG_INFO_F("Memory release started, almost {} bytes will be released", release_bytes);
while (release_bytes > 0) {
// tcmalloc releasing memory will lock page heap, release 1MB at a time to avoid locking
// page heap for long time
@@ -2823,7 +2824,7 @@ void replica_stub::create_child_replica(rpc_address primary_address,
{
replica_ptr child_replica = create_child_replica_if_not_found(child_gpid, &app, parent_dir);
if (child_replica != nullptr) {
- ddebug_f("app({}), create child replica ({}) succeed", app.app_name, child_gpid);
+ LOG_INFO_F("app({}), create child replica ({}) succeed", app.app_name, child_gpid);
tasking::enqueue(LPC_PARTITION_SPLIT,
child_replica->tracker(),
std::bind(&replica_split_manager::child_init_replica,
@@ -2833,7 +2834,8 @@ void replica_stub::create_child_replica(rpc_address primary_address,
init_ballot),
child_gpid.thread_hash());
} else {
- dwarn_f("failed to create child replica ({}), ignore it and wait next run", child_gpid);
+ LOG_WARNING_F("failed to create child replica ({}), ignore it and wait next run",
+ child_gpid);
split_replica_error_handler(
parent_gpid,
std::bind(&replica_split_manager::parent_cleanup_split_context, std::placeholders::_1));
@@ -2849,7 +2851,7 @@ replica_ptr replica_stub::create_child_replica_if_not_found(gpid child_pid,
replica *rep = new replica(this, child_pid, *app, "./", false);
rep->_config.status = partition_status::PS_INACTIVE;
_replicas.insert(replicas::value_type(child_pid, rep));
- ddebug_f("mock create_child_replica_if_not_found succeed");
+ LOG_INFO_F("mock create_child_replica_if_not_found succeed");
return rep;
});
@@ -2859,10 +2861,10 @@ replica_ptr replica_stub::create_child_replica_if_not_found(gpid child_pid,
return it->second;
} else {
if (_opening_replicas.find(child_pid) != _opening_replicas.end()) {
- dwarn_f("failed create child replica({}) because it is under open", child_pid);
+ LOG_WARNING_F("failed create child replica({}) because it is under open", child_pid);
return nullptr;
} else if (_closing_replicas.find(child_pid) != _closing_replicas.end()) {
- dwarn_f("failed create child replica({}) because it is under close", child_pid);
+ LOG_WARNING_F("failed create child replica({}) because it is under close", child_pid);
return nullptr;
} else {
replica *rep = replica::newr(this, child_pid, *app, false, false, parent_dir);
@@ -2896,7 +2898,7 @@ replica_stub::split_replica_exec(dsn::task_code code, gpid pid, local_execution
pid.thread_hash());
return ERR_OK;
}
- dwarn_f("replica({}) is invalid", pid);
+ LOG_WARNING_F("replica({}) is invalid", pid);
return ERR_OBJECT_NOT_FOUND;
}
@@ -2955,12 +2957,12 @@ void replica_stub::on_bulk_load(bulk_load_rpc rpc)
const bulk_load_request &request = rpc.request();
bulk_load_response &response = rpc.response();
- ddebug_f("[{}@{}]: receive bulk load request", request.pid, _primary_address_str);
+ LOG_INFO_F("[{}@{}]: receive bulk load request", request.pid, _primary_address_str);
replica_ptr rep = get_replica(request.pid);
if (rep != nullptr) {
rep->get_bulk_loader()->on_bulk_load(request, response);
} else {
- derror_f("replica({}) is not existed", request.pid);
+ LOG_ERROR_F("replica({}) is not existed", request.pid);
response.err = ERR_OBJECT_NOT_FOUND;
}
}
@@ -2970,19 +2972,19 @@ void replica_stub::on_group_bulk_load(group_bulk_load_rpc rpc)
const group_bulk_load_request &request = rpc.request();
group_bulk_load_response &response = rpc.response();
- ddebug_f("[{}@{}]: received group bulk load request, primary = {}, ballot = {}, "
- "meta_bulk_load_status = {}",
- request.config.pid,
- _primary_address_str,
- request.config.primary.to_string(),
- request.config.ballot,
- enum_to_string(request.meta_bulk_load_status));
+ LOG_INFO_F("[{}@{}]: received group bulk load request, primary = {}, ballot = {}, "
+ "meta_bulk_load_status = {}",
+ request.config.pid,
+ _primary_address_str,
+ request.config.primary.to_string(),
+ request.config.ballot,
+ enum_to_string(request.meta_bulk_load_status));
replica_ptr rep = get_replica(request.config.pid);
if (rep != nullptr) {
rep->get_bulk_loader()->on_group_bulk_load(request, response);
} else {
- derror_f("replica({}) is not existed", request.config.pid);
+ LOG_ERROR_F("replica({}) is not existed", request.config.pid);
response.err = ERR_OBJECT_NOT_FOUND;
}
}
@@ -2992,11 +2994,11 @@ void replica_stub::on_detect_hotkey(detect_hotkey_rpc rpc)
const auto &request = rpc.request();
auto &response = rpc.response();
- ddebug_f("[{}@{}]: received detect hotkey request, hotkey_type = {}, detect_action = {}",
- request.pid,
- _primary_address_str,
- enum_to_string(request.type),
- enum_to_string(request.action));
+ LOG_INFO_F("[{}@{}]: received detect hotkey request, hotkey_type = {}, detect_action = {}",
+ request.pid,
+ _primary_address_str,
+ enum_to_string(request.type),
+ enum_to_string(request.action));
replica_ptr rep = get_replica(request.pid);
if (rep != nullptr) {
@@ -3044,9 +3046,9 @@ void replica_stub::update_disks_status()
continue;
}
replica->set_disk_status(dir_node->status);
- ddebug_f("{} update disk_status to {}",
- replica->name(),
- enum_to_string(replica->get_disk_status()));
+ LOG_INFO_F("{} update disk_status to {}",
+ replica->name(),
+ enum_to_string(replica->get_disk_status()));
}
}
}
diff --git a/src/replica/replication_app_base.cpp b/src/replica/replication_app_base.cpp
index 02dee9b74..8d9e6407d 100644
--- a/src/replica/replication_app_base.cpp
+++ b/src/replica/replication_app_base.cpp
@@ -98,7 +98,7 @@ error_code replica_init_info::load(const std::string &dir)
dassert_f(utils::filesystem::path_exists(info_path), "file({}) not exist", info_path);
ERR_LOG_AND_RETURN_NOT_OK(
load_json(info_path), "load replica_init_info from {} failed", info_path);
- ddebug_f("load replica_init_info from {} succeed: {}", info_path, to_string());
+ LOG_INFO_F("load replica_init_info from {} succeed: {}", info_path, to_string());
return ERR_OK;
}
@@ -110,10 +110,10 @@ error_code replica_init_info::store(const std::string &dir)
"store replica_init_info to {} failed, time_used_ns = {}",
info_path,
dsn_now_ns() - start);
- ddebug_f("store replica_init_info to {} succeed, time_used_ns = {}: {}",
- info_path,
- dsn_now_ns() - start,
- to_string());
+ LOG_INFO_F("store replica_init_info to {} succeed, time_used_ns = {}: {}",
+ info_path,
+ dsn_now_ns() - start,
+ to_string());
return ERR_OK;
}
diff --git a/src/replica/test/mutation_log_test.cpp b/src/replica/test/mutation_log_test.cpp
index 57dbea784..b302b3ac5 100644
--- a/src/replica/test/mutation_log_test.cpp
+++ b/src/replica/test/mutation_log_test.cpp
@@ -344,7 +344,7 @@ public:
if (err == ERR_OK) {
break;
}
- derror_f("mlog open failed, encountered error: {}", err);
+ LOG_ERROR_F("mlog open failed, encountered error: {}", err);
}
EXPECT_NE(mlog, nullptr);
return mlog;
diff --git a/src/replica/test/replica_disk_test.cpp b/src/replica/test/replica_disk_test.cpp
index e97160a52..e3a7c7723 100644
--- a/src/replica/test/replica_disk_test.cpp
+++ b/src/replica/test/replica_disk_test.cpp
@@ -51,7 +51,7 @@ public:
stub->on_add_new_disk(rpc);
error_code err = rpc.response().err;
if (err != ERR_OK) {
- ddebug_f("error msg: {}", rpc.response().err_hint);
+ LOG_INFO_F("error msg: {}", rpc.response().err_hint);
}
return err;
}
diff --git a/src/reporter/pegasus_counter_reporter.cpp b/src/reporter/pegasus_counter_reporter.cpp
index ad9992931..485b37a91 100644
--- a/src/reporter/pegasus_counter_reporter.cpp
+++ b/src/reporter/pegasus_counter_reporter.cpp
@@ -108,7 +108,7 @@ void pegasus_counter_reporter::prometheus_initialize()
dsn::make_unique<prometheus::Exposer>(fmt::format("0.0.0.0:{}", FLAGS_prometheus_port));
_exposer->RegisterCollectable(_registry);
- ddebug_f("prometheus exposer [0.0.0.0:{}] started", FLAGS_prometheus_port);
+ LOG_INFO_F("prometheus exposer [0.0.0.0:{}] started", FLAGS_prometheus_port);
}
void pegasus_counter_reporter::falcon_initialize()
diff --git a/src/runtime/security/client_negotiation.cpp b/src/runtime/security/client_negotiation.cpp
index 6a6b4e73b..30901b1e7 100644
--- a/src/runtime/security/client_negotiation.cpp
+++ b/src/runtime/security/client_negotiation.cpp
@@ -36,7 +36,7 @@ client_negotiation::client_negotiation(rpc_session_ptr session) : negotiation(se
void client_negotiation::start()
{
- ddebug_f("{}: start negotiation", _name);
+ LOG_INFO_F("{}: start negotiation", _name);
list_mechanisms();
}
@@ -51,9 +51,9 @@ void client_negotiation::handle_response(error_code err, const negotiation_respo
if (err != ERR_OK) {
// ERR_HANDLER_NOT_FOUND means server is old version, which doesn't support authentication
if (ERR_HANDLER_NOT_FOUND == err) {
- ddebug_f("{}: treat negotiation succeed because server is old version, which doesn't "
- "support authentication",
- _name);
+ LOG_INFO_F("{}: treat negotiation succeed because server is old version, which doesn't "
+ "support authentication",
+ _name);
succ_negotiation();
} else {
fail_negotiation();
@@ -63,7 +63,7 @@ void client_negotiation::handle_response(error_code err, const negotiation_respo
// make the negotiation succeed if server doesn't enable auth
if (negotiation_status::type::SASL_AUTH_DISABLE == response.status) {
- ddebug_f("{}: treat negotiation succeed as server doesn't enable it", _name);
+ LOG_INFO_F("{}: treat negotiation succeed as server doesn't enable it", _name);
succ_negotiation();
return;
}
@@ -104,9 +104,9 @@ void client_negotiation::on_recv_mechanisms(const negotiation_response &resp)
}
if (match_mechanism.empty()) {
- dwarn_f("server only support mechanisms of ({}), can't find expected ({})",
- boost::join(supported_mechanisms, ","),
- resp_string);
+ LOG_WARNING_F("server only support mechanisms of ({}), can't find expected ({})",
+ boost::join(supported_mechanisms, ","),
+ resp_string);
fail_negotiation();
return;
}
@@ -124,10 +124,10 @@ void client_negotiation::on_mechanism_selected(const negotiation_response &resp)
// init client sasl
auto err_s = _sasl->init();
if (!err_s.is_ok()) {
- dwarn_f("{}: initialize sasl client failed, error = {}, reason = {}",
- _name,
- err_s.code().to_string(),
- err_s.description());
+ LOG_WARNING_F("{}: initialize sasl client failed, error = {}, reason = {}",
+ _name,
+ err_s.code().to_string(),
+ err_s.description());
fail_negotiation();
return;
}
@@ -139,10 +139,10 @@ void client_negotiation::on_mechanism_selected(const negotiation_response &resp)
_status = negotiation_status::type::SASL_INITIATE;
send(_status, std::move(start_output));
} else {
- dwarn_f("{}: start sasl client failed, error = {}, reason = {}",
- _name,
- err_s.code().to_string(),
- err_s.description());
+ LOG_WARNING_F("{}: start sasl client failed, error = {}, reason = {}",
+ _name,
+ err_s.code().to_string(),
+ err_s.description());
fail_negotiation();
}
}
@@ -153,7 +153,7 @@ void client_negotiation::on_challenge(const negotiation_response &challenge)
blob response_msg;
auto err = _sasl->step(challenge.msg, response_msg);
if (!err.is_ok() && err.code() != ERR_SASL_INCOMPLETE) {
- dwarn_f("{}: negotiation failed, reason = {}", _name, err.description());
+ LOG_WARNING_F("{}: negotiation failed, reason = {}", _name, err.description());
fail_negotiation();
return;
}
@@ -168,7 +168,8 @@ void client_negotiation::on_challenge(const negotiation_response &challenge)
return;
}
- dwarn_f("{}: recv wrong negotiation msg type: {}", _name, enum_to_string(challenge.status));
+ LOG_WARNING_F(
+ "{}: recv wrong negotiation msg type: {}", _name, enum_to_string(challenge.status));
fail_negotiation();
}
@@ -196,7 +197,7 @@ void client_negotiation::succ_negotiation()
{
_status = negotiation_status::type::SASL_SUCC;
_session->set_negotiation_succeed();
- ddebug_f("{}: negotiation succeed", _name);
+ LOG_INFO_F("{}: negotiation succeed", _name);
}
} // namespace security
} // namespace dsn
diff --git a/src/runtime/security/init.cpp b/src/runtime/security/init.cpp
index eabbce480..982330cab 100644
--- a/src/runtime/security/init.cpp
+++ b/src/runtime/security/init.cpp
@@ -52,14 +52,14 @@ bool init(bool is_server)
{
error_s err = init_kerberos(is_server);
if (!err.is_ok()) {
- derror_f("initialize kerberos failed, with err = {}", err.description());
+ LOG_ERROR_F("initialize kerberos failed, with err = {}", err.description());
return false;
}
LOG_INFO("initialize kerberos succeed");
err = init_sasl(is_server);
if (!err.is_ok()) {
- derror_f("initialize sasl failed, with err = {}", err.description());
+ LOG_ERROR_F("initialize sasl failed, with err = {}", err.description());
return false;
}
LOG_INFO("initialize sasl succeed");
@@ -72,14 +72,14 @@ bool init_for_zookeeper_client()
{
error_s err = run_kinit();
if (!err.is_ok()) {
- derror_f("initialize kerberos failed, with err = {}", err.description());
+ LOG_ERROR_F("initialize kerberos failed, with err = {}", err.description());
return false;
}
LOG_INFO("initialize kerberos for zookeeper client succeed");
err = init_sasl(false);
if (!err.is_ok()) {
- derror_f("initialize sasl failed, with err = {}", err.description());
+ LOG_ERROR_F("initialize sasl failed, with err = {}", err.description());
return false;
}
LOG_INFO("initialize sasl for zookeeper client succeed");
diff --git a/src/runtime/security/kinit_context.cpp b/src/runtime/security/kinit_context.cpp
index 6d9b8c97d..df7ce3b87 100644
--- a/src/runtime/security/kinit_context.cpp
+++ b/src/runtime/security/kinit_context.cpp
@@ -226,9 +226,9 @@ error_s kinit_context::get_credentials()
_opt),
"get_init_cred");
if (!err.is_ok()) {
- dwarn_f("get credentials of {} from KDC failed, reason({})",
- FLAGS_krb5_principal,
- err.description());
+ LOG_WARNING_F("get credentials of {} from KDC failed, reason({})",
+ FLAGS_krb5_principal,
+ err.description());
return err;
}
auto cleanup = dsn::defer([&]() { krb5_free_cred_contents(_krb5_context, &creds); });
@@ -236,23 +236,23 @@ error_s kinit_context::get_credentials()
// store credentials into _ccache.
err = wrap_krb5_err(krb5_cc_store_cred(_krb5_context, _ccache, &creds), "store_cred");
if (!err.is_ok()) {
- dwarn_f("store credentials of {} to cache failed, err({})",
- FLAGS_krb5_principal,
- err.description());
+ LOG_WARNING_F("store credentials of {} to cache failed, err({})",
+ FLAGS_krb5_principal,
+ err.description());
return err;
}
_cred_expire_timestamp = creds.times.endtime;
- ddebug_f("get credentials of {} from KDC ok, expires at {}",
- FLAGS_krb5_principal,
- utils::time_s_to_date_time(_cred_expire_timestamp));
+ LOG_INFO_F("get credentials of {} from KDC ok, expires at {}",
+ FLAGS_krb5_principal,
+ utils::time_s_to_date_time(_cred_expire_timestamp));
return err;
}
void kinit_context::schedule_renew_credentials()
{
int64_t renew_gap = get_next_renew_interval();
- ddebug_f("schedule to renew credentials in {} seconds later", renew_gap);
+ LOG_INFO_F("schedule to renew credentials in {} seconds later", renew_gap);
// why don't we use timers in rDSN framework?
// 1. currently the rdsn framework may not started yet.
diff --git a/src/runtime/security/negotiation.cpp b/src/runtime/security/negotiation.cpp
index f5b7ee291..8f9e2ac8b 100644
--- a/src/runtime/security/negotiation.cpp
+++ b/src/runtime/security/negotiation.cpp
@@ -59,10 +59,10 @@ bool negotiation::check_status(negotiation_status::type status,
negotiation_status::type expected_status)
{
if (status != expected_status) {
- dwarn_f("{}: get message({}) while expect({})",
- _name,
- enum_to_string(status),
- enum_to_string(expected_status));
+ LOG_WARNING_F("{}: get message({}) while expect({})",
+ _name,
+ enum_to_string(status),
+ enum_to_string(expected_status));
return false;
}
diff --git a/src/runtime/security/negotiation_manager.cpp b/src/runtime/security/negotiation_manager.cpp
index 52589afc0..fcf1817c5 100644
--- a/src/runtime/security/negotiation_manager.cpp
+++ b/src/runtime/security/negotiation_manager.cpp
@@ -132,9 +132,9 @@ std::shared_ptr<negotiation> negotiation_manager::get_negotiation(negotiation_rp
utils::auto_read_lock l(_lock);
auto it = _negotiations.find(rpc.dsn_request()->io_session);
if (it == _negotiations.end()) {
- ddebug_f("negotiation was removed for msg: {}, {}",
- rpc.dsn_request()->rpc_code().to_string(),
- rpc.remote_address().to_string());
+ LOG_INFO_F("negotiation was removed for msg: {}, {}",
+ rpc.dsn_request()->rpc_code().to_string(),
+ rpc.remote_address().to_string());
return nullptr;
}
diff --git a/src/runtime/security/replica_access_controller.cpp b/src/runtime/security/replica_access_controller.cpp
index 30767a46f..29bf41358 100644
--- a/src/runtime/security/replica_access_controller.cpp
+++ b/src/runtime/security/replica_access_controller.cpp
@@ -39,7 +39,7 @@ bool replica_access_controller::allowed(message_ex *msg)
// they are finally ensured to be fully upgraded, they can specify some usernames to ACL and
// the table will be truly protected.
if (!_users.empty() && _users.find(user_name) == _users.end()) {
- ddebug_f("{}: user_name {} doesn't exist in acls map", _name, user_name);
+ LOG_INFO_F("{}: user_name {} doesn't exist in acls map", _name, user_name);
return false;
}
return true;
diff --git a/src/runtime/security/server_negotiation.cpp b/src/runtime/security/server_negotiation.cpp
index 467421f6c..af6ec36a1 100644
--- a/src/runtime/security/server_negotiation.cpp
+++ b/src/runtime/security/server_negotiation.cpp
@@ -37,7 +37,7 @@ server_negotiation::server_negotiation(rpc_session_ptr session) : negotiation(se
void server_negotiation::start()
{
_status = negotiation_status::type::SASL_LIST_MECHANISMS;
- ddebug_f("{}: start negotiation", _name);
+ LOG_INFO_F("{}: start negotiation", _name);
}
void server_negotiation::handle_request(negotiation_rpc rpc)
@@ -83,17 +83,17 @@ void server_negotiation::on_select_mechanism(negotiation_rpc rpc)
_selected_mechanism = request.msg.to_string();
if (supported_mechanisms.find(_selected_mechanism) == supported_mechanisms.end()) {
- dwarn_f("the mechanism of {} is not supported", _selected_mechanism);
+ LOG_WARNING_F("the mechanism of {} is not supported", _selected_mechanism);
fail_negotiation();
return;
}
error_s err_s = _sasl->init();
if (!err_s.is_ok()) {
- dwarn_f("{}: server initialize sasl failed, error = {}, msg = {}",
- _name,
- err_s.code().to_string(),
- err_s.description());
+ LOG_WARNING_F("{}: server initialize sasl failed, error = {}, msg = {}",
+ _name,
+ err_s.code().to_string(),
+ err_s.description());
fail_negotiation();
return;
}
@@ -131,10 +131,10 @@ void server_negotiation::on_challenge_resp(negotiation_rpc rpc)
void server_negotiation::do_challenge(negotiation_rpc rpc, error_s err_s, const blob &resp_msg)
{
if (!err_s.is_ok() && err_s.code() != ERR_SASL_INCOMPLETE) {
- dwarn_f("{}: negotiation failed, with err = {}, msg = {}",
- _name,
- err_s.code().to_string(),
- err_s.description());
+ LOG_WARNING_F("{}: negotiation failed, with err = {}, msg = {}",
+ _name,
+ err_s.code().to_string(),
+ err_s.description());
fail_negotiation();
return;
}
@@ -145,10 +145,10 @@ void server_negotiation::do_challenge(negotiation_rpc rpc, error_s err_s, const
if (retrive_err.is_ok()) {
succ_negotiation(rpc, user_name);
} else {
- dwarn_f("{}: retrive user name failed: with err = {}, msg = {}",
- _name,
- retrive_err.code().to_string(),
- retrive_err.description());
+ LOG_WARNING_F("{}: retrive user name failed: with err = {}, msg = {}",
+ _name,
+ retrive_err.code().to_string(),
+ retrive_err.description());
fail_negotiation();
}
} else {
@@ -164,7 +164,7 @@ void server_negotiation::succ_negotiation(negotiation_rpc rpc, const std::string
_status = response.status = negotiation_status::type::SASL_SUCC;
_session->set_client_username(user_name);
_session->set_negotiation_succeed();
- ddebug_f("{}: negotiation succeed", _name);
+ LOG_INFO_F("{}: negotiation succeed", _name);
}
} // namespace security
} // namespace dsn
diff --git a/src/runtime/task/task_engine.cpp b/src/runtime/task/task_engine.cpp
index 7236c668d..fbfb0fcbc 100644
--- a/src/runtime/task/task_engine.cpp
+++ b/src/runtime/task/task_engine.cpp
@@ -89,7 +89,7 @@ void task_worker_pool::start()
wk->start();
}
- ddebug_f(
+ LOG_INFO_F(
"[{}]: thread pool [{}] started, pool_code = {}, worker_count = {}, worker_share_core = "
"{}, partitioned = {}, ...",
_node->full_name(),
@@ -115,7 +115,7 @@ void task_worker_pool::stop()
wk->stop();
}
_is_running = false;
- ddebug_f("[{}]: thread pool {} stopped", _node->full_name(), _spec.name);
+ LOG_INFO_F("[{}]: thread pool {} stopped", _node->full_name(), _spec.name);
}
void task_worker_pool::add_timer(task *t)
@@ -239,7 +239,7 @@ void task_engine::start()
pl->start();
}
_is_running = true;
- ddebug_f("[{}]: task engine started", _node->full_name());
+ LOG_INFO_F("[{}]: task engine started", _node->full_name());
}
void task_engine::stop()
@@ -253,7 +253,7 @@ void task_engine::stop()
pl->stop();
}
_is_running = false;
- ddebug_f("[{}]: task engine stopped", _node->full_name());
+ LOG_INFO_F("[{}]: task engine stopped", _node->full_name());
}
volatile int *task_engine::get_task_queue_virtual_length_ptr(dsn::task_code code, int hash)
diff --git a/src/server/compaction_filter_rule.cpp b/src/server/compaction_filter_rule.cpp
index b81259c3f..8d6e93aa9 100644
--- a/src/server/compaction_filter_rule.cpp
+++ b/src/server/compaction_filter_rule.cpp
@@ -46,7 +46,7 @@ bool string_pattern_match(dsn::string_view value,
filter_pattern.data(),
filter_pattern.length()) == 0;
default:
- derror_f("invalid match type {}", type);
+ LOG_ERROR_F("invalid match type {}", type);
return false;
}
}
diff --git a/src/server/hotspot_partition_calculator.cpp b/src/server/hotspot_partition_calculator.cpp
index 2ded8c4e4..52c922e09 100644
--- a/src/server/hotspot_partition_calculator.cpp
+++ b/src/server/hotspot_partition_calculator.cpp
@@ -174,10 +174,10 @@ void hotspot_partition_calculator::detect_hotkey_in_hotpartition(int data_type)
for (int index = 0; index < _hot_points.size(); index++) {
if (_hot_points[index][data_type].get()->get_value() >= now_hot_partition_threshold) {
if (++_hotpartition_counter[index][data_type] >= now_occurrence_threshold) {
- derror_f("Find a {} hot partition {}.{}",
- (data_type == partition_qps_type::READ_HOTSPOT_DATA ? "read" : "write"),
- _app_name,
- index);
+ LOG_ERROR_F("Find a {} hot partition {}.{}",
+ (data_type == partition_qps_type::READ_HOTSPOT_DATA ? "read" : "write"),
+ _app_name,
+ index);
send_detect_hotkey_request(_app_name,
index,
(data_type == dsn::replication::hotkey_type::type::READ)
@@ -213,26 +213,26 @@ void hotspot_partition_calculator::send_detect_hotkey_request(
req.pid = dsn::gpid(app_id, partition_index);
auto error = _shell_context->ddl_client->detect_hotkey(target_address, req, resp);
- ddebug_f("{} {} hotkey detection in {}.{}, server address: {}",
- (action == dsn::replication::detect_action::STOP) ? "Stop" : "Start",
- (hotkey_type == dsn::replication::hotkey_type::WRITE) ? "write" : "read",
- app_name,
- partition_index,
- target_address.to_string());
+ LOG_INFO_F("{} {} hotkey detection in {}.{}, server address: {}",
+ (action == dsn::replication::detect_action::STOP) ? "Stop" : "Start",
+ (hotkey_type == dsn::replication::hotkey_type::WRITE) ? "write" : "read",
+ app_name,
+ partition_index,
+ target_address.to_string());
if (error != dsn::ERR_OK) {
- derror_f("Hotkey detect rpc sending failed, in {}.{}, error_hint:{}",
- app_name,
- partition_index,
- error.to_string());
+ LOG_ERROR_F("Hotkey detect rpc sending failed, in {}.{}, error_hint:{}",
+ app_name,
+ partition_index,
+ error.to_string());
}
if (resp.err != dsn::ERR_OK) {
- derror_f("Hotkey detect rpc executing failed, in {}.{}, error_hint:{} {}",
- app_name,
- partition_index,
- resp.err,
- resp.err_hint);
+ LOG_ERROR_F("Hotkey detect rpc executing failed, in {}.{}, error_hint:{} {}",
+ app_name,
+ partition_index,
+ resp.err,
+ resp.err_hint);
}
}
diff --git a/src/server/info_collector.cpp b/src/server/info_collector.cpp
index 162c4a7a7..1885a8092 100644
--- a/src/server/info_collector.cpp
+++ b/src/server/info_collector.cpp
@@ -170,10 +170,10 @@ void info_collector::on_app_stat()
}
get_app_counters(all_stats.row_name)->set(all_stats);
- ddebug_f("stat apps succeed, app_count = {}, total_read_qps = {}, total_write_qps = {}",
- all_rows.size(),
- all_stats.get_total_read_qps(),
- all_stats.get_total_write_qps());
+ LOG_INFO_F("stat apps succeed, app_count = {}, total_read_qps = {}, total_write_qps = {}",
+ all_rows.size(),
+ all_stats.get_total_read_qps(),
+ all_stats.get_total_write_qps());
}
info_collector::app_stat_counters *info_collector::get_app_counters(const std::string &app_name)
diff --git a/src/server/logging_utils.h b/src/server/logging_utils.h
index 77d5e5611..c9a379ece 100644
--- a/src/server/logging_utils.h
+++ b/src/server/logging_utils.h
@@ -24,14 +24,14 @@
/// Utilities for logging the operation on rocksdb.
#define derror_rocksdb(op, error, ...) \
- derror_f("{}: rocksdb {} failed: error = {} [{}]", \
- replica_name(), \
- op, \
- error, \
- fmt::format(__VA_ARGS__))
+ LOG_ERROR_F("{}: rocksdb {} failed: error = {} [{}]", \
+ replica_name(), \
+ op, \
+ error, \
+ fmt::format(__VA_ARGS__))
#define ddebug_rocksdb(op, ...) \
- ddebug_f("{}: rocksdb {}: [{}]", replica_name(), op, fmt::format(__VA_ARGS__))
+ LOG_INFO_F("{}: rocksdb {}: [{}]", replica_name(), op, fmt::format(__VA_ARGS__))
#define dwarn_rocksdb(op, ...) \
- dwarn_f("{}: rocksdb {}: [{}]", replica_name(), op, fmt::format(__VA_ARGS__))
+ LOG_WARNING_F("{}: rocksdb {}: [{}]", replica_name(), op, fmt::format(__VA_ARGS__))
diff --git a/src/server/pegasus_server_write.cpp b/src/server/pegasus_server_write.cpp
index eeaae1510..e156c3dcf 100644
--- a/src/server/pegasus_server_write.cpp
+++ b/src/server/pegasus_server_write.cpp
@@ -103,9 +103,9 @@ int pegasus_server_write::on_batched_writes(dsn::message_ex **requests, int coun
} else {
if (_non_batch_write_handlers.find(rpc_code) !=
_non_batch_write_handlers.end()) {
- dfatal_f("rpc code not allow batch: {}", rpc_code.to_string());
+ LOG_FATAL_F("rpc code not allow batch: {}", rpc_code.to_string());
} else {
- dfatal_f("rpc code not handled: {}", rpc_code.to_string());
+ LOG_FATAL_F("rpc code not handled: {}", rpc_code.to_string());
}
}
} catch (TTransportException &ex) {
diff --git a/src/test/function_test/base_api_test/test_copy.cpp b/src/test/function_test/base_api_test/test_copy.cpp
index d556512b7..5145528bb 100644
--- a/src/test/function_test/base_api_test/test_copy.cpp
+++ b/src/test/function_test/base_api_test/test_copy.cpp
@@ -166,14 +166,14 @@ const char copy_data_test::CCH[] =
TEST_F(copy_data_test, EMPTY_HASH_KEY_COPY)
{
- ddebug_f("TESTING_COPY_DATA, EMPTY HASH_KEY COPY ....");
+ LOG_INFO_F("TESTING_COPY_DATA, EMPTY HASH_KEY COPY ....");
pegasus_client::scan_options options;
options.return_expire_ts = true;
vector<pegasus::pegasus_client::pegasus_scanner *> raw_scanners;
ASSERT_EQ(PERR_OK, srouce_client_->get_unordered_scanners(INT_MAX, options, raw_scanners));
- ddebug_f("open source app scanner succeed, partition_count = {}", raw_scanners.size());
+ LOG_INFO_F("open source app scanner succeed, partition_count = {}", raw_scanners.size());
vector<pegasus::pegasus_client::pegasus_scanner_wrapper> scanners;
for (auto raw_scanner : raw_scanners) {
@@ -183,7 +183,7 @@ TEST_F(copy_data_test, EMPTY_HASH_KEY_COPY)
raw_scanners.clear();
int split_count = scanners.size();
- ddebug_f("prepare scanners succeed, split_count = {}", split_count);
+ LOG_INFO_F("prepare scanners succeed, split_count = {}", split_count);
std::atomic_bool error_occurred(false);
vector<std::unique_ptr<scan_data_context>> contexts;
diff --git a/src/test/function_test/base_api_test/test_scan.cpp b/src/test/function_test/base_api_test/test_scan.cpp
index 37d44613f..f5a3c9ddb 100644
--- a/src/test/function_test/base_api_test/test_scan.cpp
+++ b/src/test/function_test/base_api_test/test_scan.cpp
@@ -138,7 +138,7 @@ public:
}
// 1000 + 5000 + 5000 kvs
- ddebug_f("Database filled with kv count {}.", i);
+ LOG_INFO_F("Database filled with kv count {}.", i);
}
protected:
@@ -176,7 +176,7 @@ TEST_F(scan_test, OVERALL_COUNT_ONLY)
<< client_->get_error_string(ret);
delete scanner;
}
- ddebug_f("scan count {}", i);
+ LOG_INFO_F("scan count {}", i);
int base_data_count = 0;
for (auto &m : expect_kvs_) {
base_data_count += m.second.size();
diff --git a/src/utils/filesystem.cpp b/src/utils/filesystem.cpp
index 10024c928..52c88a42c 100644
--- a/src/utils/filesystem.cpp
+++ b/src/utils/filesystem.cpp
@@ -789,20 +789,20 @@ std::pair<error_code, bool> is_directory_empty(const std::string &dirname)
error_code read_file(const std::string &fname, std::string &buf)
{
if (!file_exists(fname)) {
- derror_f("file({}) doesn't exist", fname);
+ LOG_ERROR_F("file({}) doesn't exist", fname);
return ERR_FILE_OPERATION_FAILED;
}
int64_t file_sz = 0;
if (!file_size(fname, file_sz)) {
- derror_f("get file({}) size failed", fname);
+ LOG_ERROR_F("get file({}) size failed", fname);
return ERR_FILE_OPERATION_FAILED;
}
buf.resize(file_sz);
std::ifstream fin(fname, std::ifstream::in);
if (!fin.is_open()) {
- derror_f("open file({}) failed", fname);
+ LOG_ERROR_F("open file({}) failed", fname);
return ERR_FILE_OPERATION_FAILED;
}
fin.read(&buf[0], file_sz);
@@ -820,26 +820,26 @@ bool verify_file(const std::string &fname,
const int64_t &expected_fsize)
{
if (!file_exists(fname)) {
- derror_f("file({}) is not existed", fname);
+ LOG_ERROR_F("file({}) is not existed", fname);
return false;
}
int64_t f_size = 0;
if (!file_size(fname, f_size)) {
- derror_f("verify file({}) failed, becaused failed to get file size", fname);
+ LOG_ERROR_F("verify file({}) failed, becaused failed to get file size", fname);
return false;
}
std::string md5;
if (md5sum(fname, md5) != ERR_OK) {
- derror_f("verify file({}) failed, becaused failed to get file md5", fname);
+ LOG_ERROR_F("verify file({}) failed, becaused failed to get file md5", fname);
return false;
}
if (f_size != expected_fsize || md5 != expected_md5) {
- derror_f("verify file({}) failed, because file damaged, size: {} VS {}, md5: {} VS {}",
- fname,
- f_size,
- expected_fsize,
- md5,
- expected_md5);
+ LOG_ERROR_F("verify file({}) failed, because file damaged, size: {} VS {}, md5: {} VS {}",
+ fname,
+ f_size,
+ expected_fsize,
+ md5,
+ expected_md5);
return false;
}
return true;
@@ -848,19 +848,19 @@ bool verify_file(const std::string &fname,
bool verify_file_size(const std::string &fname, const int64_t &expected_fsize)
{
if (!file_exists(fname)) {
- derror_f("file({}) is not existed", fname);
+ LOG_ERROR_F("file({}) is not existed", fname);
return false;
}
int64_t f_size = 0;
if (!file_size(fname, f_size)) {
- derror_f("verify file({}) size failed, becaused failed to get file size", fname);
+ LOG_ERROR_F("verify file({}) size failed, becaused failed to get file size", fname);
return false;
}
if (f_size != expected_fsize) {
- derror_f("verify file({}) size failed, because file damaged, size: {} VS {}",
- fname,
- f_size,
- expected_fsize);
+ LOG_ERROR_F("verify file({}) size failed, because file damaged, size: {} VS {}",
+ fname,
+ f_size,
+ expected_fsize);
return false;
}
return true;
@@ -873,10 +873,10 @@ bool verify_data_md5(const std::string &fname,
{
std::string md5 = string_md5(data, data_size);
if (md5 != expected_md5) {
- derror_f("verify data({}) failed, because data damaged, size: md5: {} VS {}",
- fname,
- md5,
- expected_md5);
+ LOG_ERROR_F("verify data({}) failed, because data damaged, size: md5: {} VS {}",
+ fname,
+ md5,
+ expected_md5);
return false;
}
return true;
@@ -906,7 +906,7 @@ bool create_directory(const std::string &path, std::string &absolute_path, std::
bool write_file(const std::string &fname, std::string &buf)
{
if (!file_exists(fname)) {
- derror_f("file({}) doesn't exist", fname);
+ LOG_ERROR_F("file({}) doesn't exist", fname);
return false;
}
diff --git a/src/utils/fmt_logging.h b/src/utils/fmt_logging.h
index 5dd77d903..a9d6d26f1 100644
--- a/src/utils/fmt_logging.h
+++ b/src/utils/fmt_logging.h
@@ -31,13 +31,12 @@
dsn_log( \
__FILENAME__, __FUNCTION__, __LINE__, level, fmt::format(__VA_ARGS__).c_str()); \
} while (false)
-// TODO(yingchun): to reduce the changes in one patch, now log levels using 'fmt' are left and will
-// be refactored in a follow up patch.
-#define dinfo_f(...) dlog_f(LOG_LEVEL_DEBUG, __VA_ARGS__)
-#define ddebug_f(...) dlog_f(LOG_LEVEL_INFO, __VA_ARGS__)
-#define dwarn_f(...) dlog_f(LOG_LEVEL_WARNING, __VA_ARGS__)
-#define derror_f(...) dlog_f(LOG_LEVEL_ERROR, __VA_ARGS__)
-#define dfatal_f(...) dlog_f(LOG_LEVEL_FATAL, __VA_ARGS__)
+
+#define LOG_DEBUG_F(...) dlog_f(LOG_LEVEL_DEBUG, __VA_ARGS__)
+#define LOG_INFO_F(...) dlog_f(LOG_LEVEL_INFO, __VA_ARGS__)
+#define LOG_WARNING_F(...) dlog_f(LOG_LEVEL_WARNING, __VA_ARGS__)
+#define LOG_ERROR_F(...) dlog_f(LOG_LEVEL_ERROR, __VA_ARGS__)
+#define LOG_FATAL_F(...) dlog_f(LOG_LEVEL_FATAL, __VA_ARGS__)
#define dassert_f(x, ...) \
do { \
if (dsn_unlikely(!(x))) { \
@@ -48,11 +47,11 @@
} while (false)
// Macros for writing log message prefixed by gpid and address.
-#define dinfo_replica(...) dinfo_f("[{}] {}", replica_name(), fmt::format(__VA_ARGS__))
-#define ddebug_replica(...) ddebug_f("[{}] {}", replica_name(), fmt::format(__VA_ARGS__))
-#define dwarn_replica(...) dwarn_f("[{}] {}", replica_name(), fmt::format(__VA_ARGS__))
-#define derror_replica(...) derror_f("[{}] {}", replica_name(), fmt::format(__VA_ARGS__))
-#define dfatal_replica(...) dfatal_f("[{}] {}", replica_name(), fmt::format(__VA_ARGS__))
+#define dinfo_replica(...) LOG_DEBUG_F("[{}] {}", replica_name(), fmt::format(__VA_ARGS__))
+#define ddebug_replica(...) LOG_INFO_F("[{}] {}", replica_name(), fmt::format(__VA_ARGS__))
+#define dwarn_replica(...) LOG_WARNING_F("[{}] {}", replica_name(), fmt::format(__VA_ARGS__))
+#define derror_replica(...) LOG_ERROR_F("[{}] {}", replica_name(), fmt::format(__VA_ARGS__))
+#define dfatal_replica(...) LOG_FATAL_F("[{}] {}", replica_name(), fmt::format(__VA_ARGS__))
#define dassert_replica(x, ...) dassert_f(x, "[{}] {}", replica_name(), fmt::format(__VA_ARGS__))
// Macros to check expected condition. It will abort the application
@@ -73,7 +72,7 @@
#define ERR_LOG_AND_RETURN_NOT_TRUE(s, err, ...) \
do { \
if (dsn_unlikely(!(s))) { \
- derror_f("{}: {}", err, fmt::format(__VA_ARGS__)); \
+ LOG_ERROR_F("{}: {}", err, fmt::format(__VA_ARGS__)); \
return err; \
} \
} while (0)
diff --git a/src/utils/latency_tracer.cpp b/src/utils/latency_tracer.cpp
index b13908946..3266d1c88 100644
--- a/src/utils/latency_tracer.cpp
+++ b/src/utils/latency_tracer.cpp
@@ -204,7 +204,7 @@ std::shared_ptr<latency_tracer> latency_tracer::sub_tracer(const std::string &na
if (iter != _sub_tracers.end()) {
return iter->second;
}
- dwarn_f("can't find the [{}] sub tracer of {}", name, _name);
+ LOG_WARNING_F("can't find the [{}] sub tracer of {}", name, _name);
return nullptr;
}
@@ -276,7 +276,7 @@ void latency_tracer::dump_trace_points(/*out*/ std::string &traces)
}
if (!_is_sub && total_time_used >= _threshold) {
- dwarn_f("TRACE:the traces as fallow:\n{}", traces);
+ LOG_WARNING_F("TRACE:the traces as fallow:\n{}", traces);
return;
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pegasus.apache.org
For additional commands, e-mail: commits-help@pegasus.apache.org