zhangwen/front_linux
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add function,1.limit log freq 2.realtime log control

Browse Source
This commit is contained in:
lnk
2026-01-08 15:55:46 +08:00
parent 02127869c0
commit df9760c1a7
5 changed files with 156 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
LFtid1056/cloudfront/code/cfg_parser.cpp
View File

@@ -1232,7 +1232,8 @@ int recall_json_handle_from_mq(const std::string& body)
} catch (const std::exception& e) {
std::cerr << "Error parsing JSON: " << e.what() << std::endl;
// ★与原逻辑等价：无法解析，不再进入 recall_json_handle
DIY_ERRORLOG_CODE("process",0,LOG_CODE_JSON,"主题:%s - tag:%s的补招触发消息失败",G_MQCONSUMER_TOPIC_RC.c_str(), FRONT_INST.c_str());
DIY_ERRORLOG_CODE("process",0,LOG_CODE_JSON,"主题:%s - tag:%s的补招触发消息失败",
G_MQCONSUMER_TOPIC_RC.c_str(), FRONT_INST.c_str());
return 10004;
}
@@ -1257,7 +1258,8 @@ int recall_json_handle_from_mq(const std::string& body)
mb = nlohmann::json::parse(messageBodyStr);
} catch (const std::exception& e) {
std::cerr << "Failed to parse 'messageBody' JSON: " << e.what() << std::endl;
DIY_ERRORLOG_CODE("process",0,LOG_CODE_JSON,"主题:%s - tag:%s的补招触发消息失败",G_MQCONSUMER_TOPIC_RC.c_str(), FRONT_INST.c_str());
DIY_ERRORLOG_CODE("process",0,LOG_CODE_JSON,"主题:%s - tag:%s的补招触发消息失败",
G_MQCONSUMER_TOPIC_RC.c_str(), FRONT_INST.c_str());
return 10004;
}

11
LFtid1056/cloudfront/code/interface.cpp
View File

@@ -195,7 +195,7 @@ void handleUploadResponse(const std::string& response, std::string& wavepath) {
wavepath = nameWithoutExt;
std::cout << "wavepath: " << wavepath << std::endl;
DIY_INFOLOG("process", "前置上传暂态录波文件成功,远端文件名:%s", wavepath.c_str());
DIY_INFOLOG_CODE("process",0,LOG_CODE_FILE, "前置上传文件成功,远端文件名:%s", wavepath.c_str());
}
//上传文件
@@ -569,7 +569,7 @@ int terminal_ledger_web(std::map<std::string, terminal_dev>& terminal_dev_map,
break;
}
std::cerr << "data 无效或为空数组，重试\n";
DIY_ERRORLOG_CODE("process",0,LOG_CODE_JSON,"前置从web接口中获取的台账信息为空或者无效信息无法解析,请核对前置使用的入参信息:%s",inputparm.c_str());
DIY_ERRORLOG_CODE("process",0,LOG_CODE_JSON,"前置从web接口中获取的台账信息为空或者无效信息无法解析,请核对接口参数和前置使用的入参信息:%s",inputparm.c_str());
} catch (const nlohmann::json::parse_error& e) {
std::cerr << "parse error: " << e.what() << ", retrying...\n";
}
@@ -579,6 +579,7 @@ int terminal_ledger_web(std::map<std::string, terminal_dev>& terminal_dev_map,
if (++retry > 3) {
std::cerr << "web error after 3 retry, fallback to local file\n";
DIY_WARNLOG_CODE("process",0,LOG_CODE_CONFIG, "前置无法从接口获取台账，从本地读取上一次缓存的台账,请核对接口参数和前置使用的入参信息:%s", inputparm.c_str());
std::string ledger = read_latest_ledger_file();
if (!ledger.empty()) {
try {
@@ -586,7 +587,7 @@ int terminal_ledger_web(std::map<std::string, terminal_dev>& terminal_dev_map,
if (json_data.contains("data") && json_data["data"].is_array() && !json_data["data"].empty()) {
break;
}
DIY_ERRORLOG_CODE("process",0,LOG_CODE_JSON, "前置从本地台账中获取的台账信息为空或者无效信息无法解析,请核对前置使用的入参信息:%s", inputparm.c_str());
DIY_ERRORLOG_CODE("process",0,LOG_CODE_JSON, "前置从本地台账中获取的台账信息为空或者无效信息无法解析,请核对接口参数和前置使用的入参信息:%s", inputparm.c_str());
} catch (const nlohmann::json::parse_error& e) {
std::cerr << "local parse error: " << e.what() << "\n";
}
@@ -709,7 +710,7 @@ int parse_device_cfg_web()
input_jstr += "}";
std::cout << "input_jstr: " << input_jstr << std::endl;
DIY_DEBUGLOG("process","前置获取台账使用的请求输入为:%s", input_jstr.c_str());
DIY_DEBUGLOG_CODE("process",0,LOG_CODE_LEDGER,"前置获取台账使用的请求输入为:%s", input_jstr.c_str());
// 2. 调用接口
std::map<std::string, terminal_dev> terminal_dev_map;
@@ -749,7 +750,7 @@ int parse_device_cfg_web()
// 5. 台账数量与配置比对
int count_cfg = static_cast<int>(terminal_dev_map.size());
std::cout << "terminal_ledger_num: " << count_cfg << std::endl;
DIY_DEBUGLOG("process", "前置获取到的台账的数量为:%d",count_cfg);
DIY_DEBUGLOG_CODE("process",0,LOG_CODE_LEDGER,"前置获取到的台账的数量为:%d",count_cfg);
if (IED_COUNT < count_cfg) {
std::cout << "!!!!!!!!!!single process has ledger count more than config!!!!!!!" << std::endl;

156
LFtid1056/cloudfront/code/log4.cpp
View File

@@ -17,6 +17,8 @@
#include <vector>
#include <array>
#include <fnmatch.h>
#include <unordered_map>
#include <chrono>
//////////////////////////////////////////////////////////////////////////////////////////////////////////
#include "log4cplus/logger.h"
@@ -48,9 +50,35 @@ extern std::string subdir;
//日志主题
extern std::string G_LOG_TOPIC;
////////////////////////////////////////////////////////////////////////////////////////////////////
const int LOGTYPE_DEFAULT = LOG_CODE_OTHER;
static const int LOGTYPE_WILDCARD = 999; // logtype 通配号码
static const char* ID_WILDCARD = "all"; // id 通配字段
std::map<std::string, TypedLogger> logger_map;
DebugSwitch g_debug_switch;
//用来控制日志上送的结构
struct LOGEntry {
std::string id; //测点和装置需要的id
std::string level; // terminal / measurepoint /process
int logtype; // 日志类型
int min_grade; // DEBUG / INFO / WARN / ERROR
int countdown; // 倒计时，单位秒
};
//日志上送map管理
std::map<std::string, LOGEntry> g_log_entries;
pthread_mutex_t g_log_mutex = PTHREAD_MUTEX_INITIALIZER;
/////////////////////////////////////////////////////////////////////
std::string build_debug_key(const std::string& id, const std::string& level, int logtype);
//////////////////////////////////////////////////////////
/* log4.cpp 顶部 */
#if __cplusplus >= 201103L
thread_local int g_log_code_tls = 0;
#else
@@ -122,8 +150,7 @@ bool DebugSwitch::match(const std::string& logger_name, int level, int logtype)
return false;
}
std::map<std::string, TypedLogger> logger_map;
DebugSwitch g_debug_switch;
/*class SendAppender : public Appender {
protected:
@@ -201,7 +228,7 @@ private:
std::chrono::steady_clock::time_point::min();
};
static std::unordered_map<std::string, RateState> s_rate_map;
static std::unordered_map<std::string, RateState> s_rate_map; //频率map
static std::mutex s_rate_mutex;
// 定义“同一条日志”的规则：logger + level + code + msg //原来只区分了日志登记名和等级，现在具体到每一条日志
@@ -218,9 +245,19 @@ private:
std::lock_guard<std::mutex> lk(s_rate_mutex);
RateState& st = s_rate_map[key];
// 超过恢复时间，重置计数
const int RESET_SEC = 3600; // 一小时
if (st.last_emit != steady_clock::time_point::min()) {
auto idle = duration_cast<seconds>(now - st.last_emit).count();
if (idle >= RESET_SEC) {
st.hit_count = 0; // 恢复为“新日志”
}
}
st.hit_count++;
const int period_sec = (st.hit_count > 5) ? 300 : 1;
const int period_sec = (st.hit_count > 3) ? 300 : 1;
if (st.last_emit == steady_clock::time_point::min()) {
st.last_emit = now;
@@ -236,6 +273,48 @@ private:
return false;
}
static bool find_entry_allow(const std::string& key, int level_val) { //通配方式查找
std::map<std::string, LOGEntry>::iterator it = g_log_entries.find(key);
if (it == g_log_entries.end() || it->second.countdown <= 0) return false;
return level_val >= it->second.min_grade;
}
static bool allow_low_level_send(const std::string& id,
const std::string& level_str,//层级
int logtype,
int level_val) {//告警等级
pthread_mutex_lock(&g_log_mutex);
// 1) 精确匹配：id + level + logtype
if (find_entry_allow(build_debug_key(id, level_str, logtype), level_val)) {
pthread_mutex_unlock(&g_log_mutex);
return true;
}
// 2) logtype 通配：id + level + -1
if (find_entry_allow(build_debug_key(id, level_str, LOGTYPE_WILDCARD), level_val)) {
pthread_mutex_unlock(&g_log_mutex);
return true;
}
// 3) id 通配：* + level + logtype
if (find_entry_allow(build_debug_key(ID_WILDCARD, level_str, logtype), level_val)) {
pthread_mutex_unlock(&g_log_mutex);
return true;
}
// 4) 双通配：* + level + -1
if (find_entry_allow(build_debug_key(ID_WILDCARD, level_str, LOGTYPE_WILDCARD), level_val)) {
pthread_mutex_unlock(&g_log_mutex);
return true;
}
pthread_mutex_unlock(&g_log_mutex);
return false;
}
protected:
void append(const spi::InternalLoggingEvent& event) override {
std::string logger_name = event.getLoggerName();
@@ -253,11 +332,33 @@ protected:
// TLS code
int code = g_log_code_tls;
const int safe_logtype = 0;
const int safe_logtype = code; // 使用 code 作为 logtype
if (!(level == ERROR_LOG_LEVEL ||
level == WARN_LOG_LEVEL ||
g_debug_switch.match(logger_name, level, safe_logtype))) {
bool allow_send = false;
if (level >= WARN_LOG_LEVEL) {
allow_send = true;
} else {
// NORMAL/DEBUG 默认不上送，必须命令打开
std::string ctrl_level = level_str; // "process" / "terminal" / "measurepoint"
std::string ctrl_id;
if (ctrl_level == "process") {
ctrl_id = "process"; // process 用固定 id
} else {
ctrl_id = extract_logger_id(logger_name); // terminal.<id> / monitor.<id>
if (ctrl_id.empty()) {
// 没解析出 id，就不给低等级上送（避免误发）
allow_send = false;
}
}
if (!ctrl_id.empty()) {
allow_send = allow_low_level_send(ctrl_id, ctrl_level, safe_logtype, level);
}
}
if (!allow_send) {
return;
}
@@ -318,22 +419,11 @@ public:
std::unordered_map<std::string, SendAppender::RateState> SendAppender::s_rate_map;
std::mutex SendAppender::s_rate_mutex;
//用来控制日志上送的结构
struct LOGEntry {
std::string id; //测点和装置需要的id
std::string level; // terminal / measurepoint /process
int min_grade; // DEBUG / INFO / WARN / ERROR
int countdown; // 倒计时，单位秒
};
//日志上送map管理
std::map<std::string, LOGEntry> g_log_entries;
pthread_mutex_t g_log_mutex = PTHREAD_MUTEX_INITIALIZER;
// 生成唯一 key
std::string build_debug_key(const std::string& id, const std::string& level) {
return id + "|" + level + "|";
std::string build_debug_key(const std::string& id, const std::string& level, int logtype) {
std::ostringstream oss;
oss << id << "|" << level << "|" << logtype;
return oss.str();
}
// 外部线程中调用：每秒更新所有倒计时，0 则删除
@@ -354,19 +444,19 @@ void update_log_entries_countdown() {
pthread_mutex_unlock(&g_log_mutex);
}
void process_log_command(const std::string& id, const std::string& level, const std::string& grade, const std::string& logtype_str) {
if (level != "terminal" && level != "measurepoint") return;
void process_log_command(const std::string& id, const std::string& level, const std::string& grade, int logtype) {
if (level != "terminal" && level != "measurepoint" && level != "process") return;
int grade_level = (grade == "DEBUG") ? DEBUG_LOG_LEVEL : INFO_LOG_LEVEL;
std::string key = build_debug_key(id, level);
std::string key = build_debug_key(id, level, logtype);
pthread_mutex_lock(&g_log_mutex);
LOGEntry& entry = g_log_entries[key]; // 会自动 insert 或取已有
entry.id = id;
entry.level = level;
entry.logtype = logtype;
entry.min_grade = grade_level;
entry.countdown = 60; // 重置倒计时
@@ -405,7 +495,7 @@ log4cplus::Logger init_logger(const std::string& full_name,
//进程的日志
void init_logger_process() {
std::string base_dir = FRONT_PATH + "/" + subdir + "/processNo" + std::to_string(g_front_seg_index) + "/log";
logger_map["process"] = TypedLogger(init_logger(std::string("process"), base_dir, std::string("process")), LOGTYPE_DATA);
logger_map["process"] = TypedLogger(init_logger(std::string("process"), base_dir, std::string("process")), LOGTYPE_DEFAULT);
std::cout << "process log init ok" << std::endl;
}
@@ -444,7 +534,7 @@ void init_loggers_bydevid(const std::string& dev_id)
device_appender->setLayout(std::unique_ptr<Layout>(new PatternLayout("%D{%Y-%m-%d %H:%M:%S} [%p] [%c] %m%n")));
Logger device_logger = init_logger(device_key, device_dir, dev_id, device_appender);
logger_map[device_key] = TypedLogger(device_logger, LOGTYPE_DATA);
logger_map[device_key] = TypedLogger(device_logger, LOGTYPE_DEFAULT);
}
@@ -469,7 +559,7 @@ void init_loggers_bydevid(const std::string& dev_id)
monitor_appender->setLayout(std::unique_ptr<Layout>(new PatternLayout("%D{%Y-%m-%d %H:%M:%S} [%p] [%c] %m%n")));
Logger mon_logger = init_logger(mon_key.str(), mon_path.str(), mon_name.str(), monitor_appender);
logger_map[mon_key.str()] = TypedLogger(mon_logger, LOGTYPE_DATA);
logger_map[mon_key.str()] = TypedLogger(mon_logger, LOGTYPE_DEFAULT);
}
@@ -509,7 +599,7 @@ void init_loggers()
Logger device_logger = init_logger(device_key, device_dir, term.terminal_id, device_appender);
logger_map[device_key] = TypedLogger(device_logger, LOGTYPE_DATA);
logger_map[device_key] = TypedLogger(device_logger, LOGTYPE_DEFAULT);
@@ -532,7 +622,7 @@ void init_loggers()
Logger mon_logger = init_logger(mon_key.str(), mon_path.str(), mon_name.str(), monitor_appender);
logger_map[mon_key.str()] = TypedLogger(mon_logger, LOGTYPE_DATA);
logger_map[mon_key.str()] = TypedLogger(mon_logger, LOGTYPE_DEFAULT);
}

9
LFtid1056/cloudfront/code/log4.h
View File

@@ -82,7 +82,7 @@ log4cplus::Logger init_logger(const std::string& full_name,
const std::string& base_file,
log4cplus::SharedAppenderPtr fileAppender);
void process_log_command(const std::string& id, const std::string& level, const std::string& grade, const std::string& logtype_str);
void process_log_command(const std::string& id, const std::string& level, const std::string& grade, int logtype);
void update_log_entries_countdown();
@@ -115,18 +115,23 @@ typedef enum LogCode {
LOG_CODE_JSON = 98, /* JSON结构 */
LOG_CODE_OTHER = 99, /* 其他类型 */
LOG_CODE_LEDGER = 100, /* 台账类型 */
LOG_CODE_RPTINIT = 101, /* 报告初始化 */
LOG_CODE_ICD_AND_DOWNLOAD = 200, /* ICD 类型 */
LOG_CODE_TRANSIENT = 300, /* 暂态发生 */
LOG_CODE_TRANSIENT_COMM = 301, /* 暂态接口 */
LOG_CODE_COMTRADE_FILE = 302, /* 录波文件（Comtrade） */
LOG_CODE_FILE = 302, /* 文件上传 */
LOG_CODE_MQ = 400, /* MQ发送 */
LOG_CODE_RT_DATA = 401, /* 实时数据 */
LOG_CODE_LEDGER_UPDATE = 402, /* 台账更新 */
LOG_CODE_PROCESS_CONTROL = 403, /* 进程控制 */
LOG_CODE_RECALL = 404, /* 补招相关 */
LOG_CODE_LOG_REQUEST = 405, /* 日志请求 */
LOG_CODE_CLOUD = 406, /* 云前置控制 */
LOG_CODE_REPORT = 500, /* 报告处理 */
LOG_CODE_COMM = 600, /* 通讯状态 */
LOG_CODE_SPACE_ALARM = 700, /* 空间告警 */
LOG_CODE_DEV_ALARM = 800 /* 设备告警 */

26
LFtid1056/cloudfront/code/rocketmq.cpp
View File

@@ -520,7 +520,7 @@ bool parseJsonMessageSET(const std::string& json_str) {
std::cout << "msg index: " << index_value << " self index: " << g_front_seg_index << std::endl;
DIY_INFOLOG("process", "收到主题:%s - tag:%s的进程控制消息",
DIY_INFOLOG_CODE("process",0,LOG_CODE_PROCESS_CONTROL, "收到主题:%s - tag:%s的进程控制消息",
G_MQCONSUMER_TOPIC_SET.c_str(),FRONT_INST.c_str());
if (code_str == "set_process") {
@@ -629,8 +629,9 @@ bool parseJsonMessageLOG(const std::string& json_str) {
}
// 提取字段
std::string guid, code_str, id, level, grade, logtype, frontType;
std::string guid, code_str, id, level, grade, frontType;
int processNo = 0;
int logtype = 0;
try {
guid = messageBody["guid"].get<std::string>();
@@ -639,7 +640,7 @@ bool parseJsonMessageLOG(const std::string& json_str) {
id = messageBody["id"].get<std::string>();
level = messageBody["level"].get<std::string>();
grade = messageBody["grade"].get<std::string>();
logtype = messageBody["logtype"].get<std::string>();
logtype = messageBody["logtype"].get<int>();
frontType = messageBody["frontType"].get<std::string>();
} catch (const std::exception& e) {
std::cerr << "Error extracting fields: " << e.what() << std::endl;
@@ -658,7 +659,7 @@ bool parseJsonMessageLOG(const std::string& json_str) {
return true;
}*/
DIY_INFOLOG("process", "收到主题:%s - tag:%s的日志控制消息",
DIY_INFOLOG_CODE("process",0,LOG_CODE_LOG_REQUEST,"收到主题:%s - tag:%s的日志控制消息",
G_MQCONSUMER_TOPIC_LOG.c_str(),FRONT_INST.c_str());
std::cout << "msg index: " << processNo << " self index: " << g_front_seg_index << std::endl;
@@ -670,16 +671,16 @@ bool parseJsonMessageLOG(const std::string& json_str) {
if (code_str == "set_log") {
// 校验数据合法性
bool valid =
(level == "terminal" || level == "measurepoint") &&
(level == "terminal" || level == "measurepoint" || level == "process") && //all是通配
(grade == "NORMAL" || grade == "DEBUG") &&
(logtype == "com" || logtype == "data") &&
(logtype > 0 && logtype < 1000) &&//999是通配
(!id.empty() && !is_blank(id));
if (valid) {
process_log_command(id, level, grade, logtype);
} else {
std::cout << "type doesn't match" << std::endl;
DIY_WARNLOG_CODE("process",0,LOG_CODE_JSON, "日志控制指令失败");
DIY_ERRORLOG_CODE("process",0,LOG_CODE_JSON, "日志控制指令失败");
}
std::cout << "this msg should only execute once" << std::endl;
@@ -733,7 +734,7 @@ bool parseJsonMessageUD(const std::string& json_str, const std::string& output_d
std::cout << "msg index: " << process_No << " self index: " << g_front_seg_index << std::endl;
DIY_INFOLOG("process", "收到主题:%s - tag:%s的台账更新消息",
DIY_INFOLOG_CODE("process",0,LOG_CODE_LEDGER_UPDATE,"收到主题:%s - tag:%s的台账更新消息",
G_MQCONSUMER_TOPIC_UD.c_str(), FRONT_INST.c_str());
//send_reply_to_queue(guid, static_cast<int>(ResponseCode::ACCEPTED), "收到台账更新指令");
@@ -1025,7 +1026,7 @@ rocketmq::ConsumeStatus myMessageCallbackrtdata(const rocketmq::MQMessageExt& ms
}
// 日志记录
DIY_INFOLOG("process", "收到主题:%s - tag:%s的实时触发消息",G_MQCONSUMER_TOPIC_RT.c_str(),FRONT_INST.c_str());
DIY_INFOLOG_CODE("process",0,LOG_CODE_RT_DATA,"收到主题:%s - tag:%s的实时触发消息",G_MQCONSUMER_TOPIC_RT.c_str(),FRONT_INST.c_str());
std::cout << "rtdata Callback received message: " << body << std::endl;
if (!key.empty()) {
@@ -1235,6 +1236,8 @@ rocketmq::ConsumeStatus myMessageCallbackrecall(const rocketmq::MQMessageExt& ms
// 调试输出
std::cout << "myMessageCallbackrecall" << std::endl;
std::string body = msg.getBody();
std::string key = msg.getKeys();
@@ -1251,6 +1254,9 @@ rocketmq::ConsumeStatus myMessageCallbackrecall(const rocketmq::MQMessageExt& ms
std::cout << "Message Key: N/A" << std::endl;
}
DIY_INFOLOG_CODE("process",0,LOG_CODE_RECALL,"收到主题:%s - tag:%s的补招消息",
G_MQCONSUMER_TOPIC_RC.c_str(), FRONT_INST.c_str());
// 解析 JSON 字符串
recall_json_handle_from_mq(body);//不再使用文件补招方式
@@ -2264,7 +2270,7 @@ rocketmq::ConsumeStatus cloudMessageCallback(const rocketmq::MQMessageExt& msg)
}
// 日志记录
DIY_INFOLOG("process", "收到主题:%s - tag:%s的云前置控制消息",G_MQCONSUMER_TOPIC_CLOUD.c_str(),FRONT_INST.c_str());
DIY_INFOLOG_CODE("process",0,LOG_CODE_CLOUD, "收到主题:%s - tag:%s的云前置控制消息",G_MQCONSUMER_TOPIC_CLOUD.c_str(),FRONT_INST.c_str());
std::cout << "cloud Callback received message: " << body << std::endl;
if (!key.empty()) {