diff --git a/LFtid1056/cloudfront/code/cfg_parser.cpp b/LFtid1056/cloudfront/code/cfg_parser.cpp index 534e85d..ecb986c 100644 --- a/LFtid1056/cloudfront/code/cfg_parser.cpp +++ b/LFtid1056/cloudfront/code/cfg_parser.cpp @@ -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; } diff --git a/LFtid1056/cloudfront/code/interface.cpp b/LFtid1056/cloudfront/code/interface.cpp index 1e3d2e1..638d62a 100644 --- a/LFtid1056/cloudfront/code/interface.cpp +++ b/LFtid1056/cloudfront/code/interface.cpp @@ -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& 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& 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& 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 terminal_dev_map; @@ -749,7 +750,7 @@ int parse_device_cfg_web() // 5. 台账数量与配置比对 int count_cfg = static_cast(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; diff --git a/LFtid1056/cloudfront/code/log4.cpp b/LFtid1056/cloudfront/code/log4.cpp index 4027a36..76f6e47 100644 --- a/LFtid1056/cloudfront/code/log4.cpp +++ b/LFtid1056/cloudfront/code/log4.cpp @@ -17,6 +17,8 @@ #include #include #include +#include +#include ////////////////////////////////////////////////////////////////////////////////////////////////////////// #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 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 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 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 s_rate_map; + static std::unordered_map s_rate_map; //频率map static std::mutex s_rate_mutex; // 定义“同一条日志”的规则：logger + level + code + msg //原来只区分了日志登记名和等级，现在具体到每一条日志 @@ -218,9 +245,19 @@ private: std::lock_guard 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(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::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. / monitor. + 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 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 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(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(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); } diff --git a/LFtid1056/cloudfront/code/log4.h b/LFtid1056/cloudfront/code/log4.h index a74db7e..666a409 100644 --- a/LFtid1056/cloudfront/code/log4.h +++ b/LFtid1056/cloudfront/code/log4.h @@ -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 /* 设备告警 */ diff --git a/LFtid1056/cloudfront/code/rocketmq.cpp b/LFtid1056/cloudfront/code/rocketmq.cpp index 131e13b..04ccec4 100644 --- a/LFtid1056/cloudfront/code/rocketmq.cpp +++ b/LFtid1056/cloudfront/code/rocketmq.cpp @@ -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(); @@ -639,7 +640,7 @@ bool parseJsonMessageLOG(const std::string& json_str) { id = messageBody["id"].get(); level = messageBody["level"].get(); grade = messageBody["grade"].get(); - logtype = messageBody["logtype"].get(); + logtype = messageBody["logtype"].get(); frontType = messageBody["frontType"].get(); } 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(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()) {