lnk/datatrace
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

完成工具

Browse Source
This commit is contained in:
lnk
2026-07-13 15:12:55 +08:00
parent b61388443f
commit 9181e2d4e9
77 changed files with 1847801 additions and 20269 deletions
Download Patch File Download Diff File Expand all files Collapse all files

254
source/mq.cpp
View File

@@ -477,6 +477,114 @@ std::string reportNameFromJsonText(const std::string& text) {
return "";
}
std::string lowerAscii(std::string value);
bool isMonitorIdKey(const std::string& key) {
std::string k = lowerAscii(trimCopy(key));
return k == "monitor" ||
k == "monitorid" ||
k == "monitor_id" ||
k == "mp_id" ||
k == "mpid" ||
k == "measurepointid" ||
k == "measure_point_id" ||
k == "measureid" ||
k == "measure_id";
}
bool looksLikeJsonText(const std::string& text) {
std::string s = trimCopy(text);
return !s.empty() && (s.front() == '{' || s.front() == '[');
}
bool findMonitorIdValue(const json& value, std::string& out) {
if (value.is_object()) {
for (auto it = value.begin(); it != value.end(); ++it) {
if (isMonitorIdKey(it.key())) {
out = trimCopy(jsonScalarToString(it.value()));
if (!out.empty()) return true;
}
}
for (auto it = value.begin(); it != value.end(); ++it) {
if ((it.key() == "messageBody" || it.key() == "body") && it.value().is_string()) {
std::string embedded = it.value().get<std::string>();
if (looksLikeJsonText(embedded)) {
try {
if (findMonitorIdValue(json::parse(embedded), out)) return true;
} catch (...) {
}
}
}
if (findMonitorIdValue(it.value(), out)) return true;
}
} else if (value.is_array()) {
for (const auto& item : value) {
if (findMonitorIdValue(item, out)) return true;
}
}
return false;
}
std::string monitorIdFromJsonText(const std::string& text) {
try {
std::string monitorId;
if (findMonitorIdValue(json::parse(text), monitorId)) return monitorId;
} catch (...) {
}
return "";
}
bool findGuidValue(const json& value, std::string& out) {
if (value.is_object()) {
for (auto it = value.begin(); it != value.end(); ++it) {
std::string key = lowerAscii(trimCopy(it.key()));
if (key == "guid" || key == "traceguid" || key == "trace_guid") {
out = trimCopy(jsonScalarToString(it.value()));
if (!out.empty()) return true;
}
}
for (auto it = value.begin(); it != value.end(); ++it) {
if ((it.key() == "messageBody" || it.key() == "body") && it.value().is_string()) {
std::string embedded = it.value().get<std::string>();
if (looksLikeJsonText(embedded)) {
try {
if (findGuidValue(json::parse(embedded), out)) return true;
} catch (...) {
}
}
}
if (findGuidValue(it.value(), out)) return true;
}
} else if (value.is_array()) {
for (const auto& item : value) {
if (findGuidValue(item, out)) return true;
}
}
return false;
}
std::string guidFromJsonText(const std::string& text) {
try {
std::string guid;
if (findGuidValue(json::parse(text), guid)) return guid;
json root = json::parse(text);
if (root.is_object()) {
auto it = root.find("key");
if (it != root.end()) return trimCopy(jsonScalarToString(*it));
}
} catch (...) {
}
return "";
}
std::string normalizeMonitorId(std::string value) {
value = trimCopy(value);
value.erase(std::remove_if(value.begin(), value.end(), [](unsigned char ch) {
return std::isspace(ch) != 0 || ch == '"' || ch == '\'' || ch == '{' || ch == '}';
}), value.end());
return lowerAscii(value);
}
std::string safePathPart(std::string value, const std::string& fallback) {
if (value.empty()) value = fallback;
for (char& ch : value) {
@@ -566,6 +674,18 @@ class RocketMqRuntime {
int (*DestroyPushConsumer)(CPushConsumer*) = nullptr;
};
struct TraceSession {
std::string monitorId;
std::string monitorName;
std::string traceDir;
std::string xmlPath;
std::string frontType;
std::string guid;
std::unordered_set<std::string> receivedDataTypes;
long long dataTimestamp = 0;
int originCounter = 0;
};
public:
~RocketMqRuntime() {
DetachedHandles old;
@@ -624,6 +744,10 @@ public:
dataTimestamp_ = 0;
originCounter_ = 0;
currentTraceDir_.clear();
currentMonitorId_.clear();
currentMonitorName_.clear();
sessions_.clear();
sessionsByGuid_.clear();
lastError_.clear();
if (!loadLibraryLocked()) {
@@ -703,16 +827,36 @@ public:
void setTraceTarget(const PqToolConfig& cfg,
const PqLedgerDevice& device,
const PqLedgerMonitor& monitor) {
const PqLedgerMonitor& monitor,
const std::string& frontType,
const std::string& guid) {
std::lock_guard<std::mutex> lock(mu_);
cfg_ = cfg;
currentTraceDir_ = pqTraceDirectoryFor(cfg, device, monitor);
currentXmlPath_ = pqFindDeviceXmlPath(cfg, device);
currentMonitorId_ = monitor.id;
currentMonitorName_ = monitor.name;
fs::path dir = fs::u8path(currentTraceDir_);
fs::create_directories(dir);
clearTraceTextRecords(dir);
dataTimestamp_ = 0;
originCounter_ = 0;
std::string monitorKey = normalizeMonitorId(monitor.id);
auto old = sessions_.find(monitorKey);
if (old != sessions_.end() && !old->second.guid.empty()) {
sessionsByGuid_.erase(normalizeMonitorId(old->second.guid));
}
TraceSession& session = sessions_[monitorKey];
session.monitorId = monitor.id;
session.monitorName = monitor.name;
session.traceDir = currentTraceDir_;
session.xmlPath = currentXmlPath_;
session.frontType = frontType;
session.guid = guid;
session.receivedDataTypes.clear();
session.dataTimestamp = 0;
session.originCounter = 0;
if (!guid.empty()) sessionsByGuid_[normalizeMonitorId(guid)] = monitorKey;
json context = {
{"deviceId", device.id},
@@ -721,6 +865,8 @@ public:
{"processNo", device.processNo},
{"monitorId", monitor.id},
{"monitorName", monitor.name},
{"guid", guid},
{"frontType", frontType},
{"xmlPath", currentXmlPath_},
{"jsonDataPath", (dir / "jsondata_01.txt").u8string()},
{"jsonDataPaths", json::array({
@@ -776,6 +922,37 @@ private:
return instance().onMessage(msg);
}
TraceSession* findSessionForMessageLocked(const std::string& body,
const std::string& key,
std::string& actualMonitor,
std::string& reason) {
actualMonitor.clear();
reason.clear();
actualMonitor = monitorIdFromJsonText(body);
std::string normalized = normalizeMonitorId(actualMonitor);
if (!normalized.empty()) {
auto it = sessions_.find(normalized);
if (it != sessions_.end()) return &it->second;
reason = "monitor not being traced, actual=" + actualMonitor;
}
std::string guid = guidFromJsonText(body);
if (guid.empty()) guid = trimCopy(key);
std::string normalizedGuid = normalizeMonitorId(guid);
if (!normalizedGuid.empty()) {
auto git = sessionsByGuid_.find(normalizedGuid);
if (git != sessionsByGuid_.end()) {
auto sit = sessions_.find(git->second);
if (sit != sessions_.end()) return &sit->second;
}
if (reason.empty()) reason = "guid not being traced, actual=" + guid;
}
if (reason.empty()) reason = "missing monitor id/guid";
return nullptr;
}
int onMessage(CMessageExt* msg) {
std::string topic = GetMessageTopic && GetMessageTopic(msg) ? GetMessageTopic(msg) : "";
std::string tag = GetMessageTags && GetMessageTags(msg) ? GetMessageTags(msg) : "";
@@ -790,25 +967,58 @@ private:
+ " tag=" + tag
+ " key=" + key
+ " bodyLen=" + std::to_string(body.size()) + "\n");
bool isData = topic == cfg_.dataTopic && tagMatches(cfg_.dataTag, tag);
std::string dataFrontType = dataFrontTypeForMessageLocked(topic, tag);
bool isData = !dataFrontType.empty();
bool isTrace = topic == cfg_.traceTopic && tagMatches(cfg_.traceTag, tag);
TraceSession* session = nullptr;
if (isData || isTrace) {
std::string actualMonitor;
std::string reason;
session = findSessionForMessageLocked(body, key, actualMonitor, reason);
if (!session) {
appendTextFile(runtimeLogPath("mq_messages.log"), nowIsoMillis()
+ " skipped " + (isData ? "DATATopic" : "TraceTopic")
+ " by monitor router: " + reason
+ " topic=" + topic
+ " tag=" + tag + "\n");
return 0;
}
if (isData && !session->frontType.empty() && session->frontType != dataFrontType) {
appendTextFile(runtimeLogPath("mq_messages.log"), nowIsoMillis()
+ " skipped DATATopic by frontType router: session="
+ session->frontType + ", message=" + dataFrontType
+ " topic=" + topic + " tag=" + tag + "\n");
return 0;
}
}
if (isData) {
std::string dataType = dataTypeFromJsonText(body);
fs::path out = jsonDataOutputPathLocked(dataType);
std::string normalizedType = normalizeDataType(dataType);
if (!session->receivedDataTypes.insert(normalizedType).second) {
appendTextFile(runtimeLogPath("mq_messages.log"), nowIsoMillis()
+ " skipped DATATopic duplicate DATA_TYPE=" + normalizedType
+ " monitor=" + session->monitorId
+ " topic=" + topic + " tag=" + tag + "\n");
return 0;
}
fs::path out = jsonDataOutputPathLocked(*session, dataType);
writeTextFile(out, body);
long long bodyTs = timestampFromJsonText(body);
dataTimestamp_ = bodyTs > 0 ? bodyTs : storeTs;
session->dataTimestamp = bodyTs > 0 ? bodyTs : storeTs;
if (normalizeMonitorId(session->monitorId) == normalizeMonitorId(currentMonitorId_)) {
dataTimestamp_ = session->dataTimestamp;
}
appendTextFile(runtimeLogPath("mq_messages.log"), nowIsoMillis() + " DATATopic DATA_TYPE=" + dataType + " tag=" + tag + " -> " + out.u8string() + "\n");
if (event_) event_({"data", eventTitle("DATATopic", out.u8string()), body, out.u8string()});
if (event_) event_({"data", eventTitle("DATATopic", out.u8string()), body, out.u8string(), session->monitorId, session->traceDir});
} else if (isTrace) {
long long bodyTs = timestampFromJsonText(body);
if (dataTimestamp_ > 0 && bodyTs > 0 && bodyTs < dataTimestamp_) {
if (session->dataTimestamp > 0 && bodyTs > 0 && bodyTs < session->dataTimestamp) {
appendTextFile(runtimeLogPath("mq_messages.log"), nowIsoMillis() + " TraceTopic tag=" + tag + " skipped by timestamp\n");
} else {
fs::path out = nextOriginOutputPathLocked(reportNameFromJsonText(body));
fs::path out = nextOriginOutputPathLocked(*session, reportNameFromJsonText(body));
writeTextFile(out, body);
appendTextFile(runtimeLogPath("mq_messages.log"), nowIsoMillis() + " TraceTopic tag=" + tag + " -> " + out.u8string() + "\n");
if (event_) event_({"trace", eventTitle("TraceTopic", out.u8string()), body, out.u8string()});
if (event_) event_({"trace", eventTitle("TraceTopic", out.u8string()), body, out.u8string(), session->monitorId, session->traceDir});
}
} else {
appendTextFile(runtimeLogPath("mq_messages.log"), nowIsoMillis() + " skipped topic=" + topic + " tag=" + tag + "\n");
@@ -833,9 +1043,16 @@ private:
return fs::u8path("origin.txt");
}
fs::path jsonDataOutputPathLocked(const std::string& dataType) const {
std::string dataFrontTypeForMessageLocked(const std::string& topic, const std::string& tag) const {
if (topic == cfg_.dataStatTopic && tagMatches(cfg_.dataStatTag, tag)) return "cfg_stat_data";
if (topic == cfg_.dataRealtimeTopic && tagMatches(cfg_.dataRealtimeTag, tag)) return "cfg_3s_data";
return "";
}
fs::path jsonDataOutputPathLocked(const TraceSession& session, const std::string& dataType) const {
std::string suffix = normalizeDataType(dataType);
std::string filename = "jsondata_" + safePathPart(suffix, "unknown") + ".txt";
if (!session.traceDir.empty()) return fs::u8path(session.traceDir) / fs::u8path(filename);
if (!currentTraceDir_.empty()) return fs::u8path(currentTraceDir_) / fs::u8path(filename);
if (!cfg_.jsonDataPath.empty()) {
std::vector<std::string> paths;
@@ -852,18 +1069,18 @@ private:
return runtimeJsonPath(filename);
}
fs::path nextOriginOutputPathLocked(const std::string& reportName) {
if (currentTraceDir_.empty()) return originOutputPath();
fs::path nextOriginOutputPathLocked(TraceSession& session, const std::string& reportName) {
if (session.traceDir.empty() && currentTraceDir_.empty()) return originOutputPath();
std::string safeReport = reportName.empty() ? "" : safePathPart(reportName, "report");
std::string baseName;
if (safeReport.empty()) {
++originCounter_;
baseName = "origin" + std::to_string(originCounter_);
++session.originCounter;
baseName = "origin" + std::to_string(session.originCounter);
} else {
baseName = "origin_" + safeReport;
}
fs::path dir = fs::u8path(currentTraceDir_);
fs::path dir = session.traceDir.empty() ? fs::u8path(currentTraceDir_) : fs::u8path(session.traceDir);
fs::path out = dir / fs::u8path(baseName + ".txt");
int suffix = 2;
while (fs::exists(out)) {
@@ -1063,6 +1280,10 @@ private:
std::string lastError_;
std::string currentTraceDir_;
std::string currentXmlPath_;
std::string currentMonitorId_;
std::string currentMonitorName_;
std::unordered_map<std::string, TraceSession> sessions_;
std::unordered_map<std::string, std::string> sessionsByGuid_;
bool active_ = false;
long long dataTimestamp_ = 0;
int originCounter_ = 0;
@@ -1383,14 +1604,15 @@ bool pqSendTraceRequest(const PqToolConfig& cfg,
const std::string& frontType,
PqLogFn log,
PqMqEventFn event) {
rocketRuntime().setTraceTarget(cfg, device, monitor);
std::string payload = pqBuildTracePayload(cfg, device, monitor, frontType);
std::string guid = guidFromJsonText(payload);
rocketRuntime().setTraceTarget(cfg, device, monitor, frontType, guid);
fs::path traceDir = fs::u8path(pqTraceDirectoryFor(cfg, device, monitor));
fs::create_directories(traceDir);
fs::path sentPath = traceDir / "sent_trace_request.json";
writeTextFile(sentPath, prettyJsonOrRaw(payload));
writeTextFile(runtimeJsonPath("last_trace_request.json"), json::parse(payload).dump(2));
if (event) event({"sent", eventTitle("Send", sentPath.u8string()), payload, sentPath.u8string()});
if (event) event({"sent", eventTitle("Send", sentPath.u8string()), payload, sentPath.u8string(), monitor.id, traceDir.u8string()});
std::string mqError;
if (rocketRuntime().sendTrace(cfg, payload, mqError, log)) {
if (log) log("Trace target monitor=" + monitor.name + "(" + monitor.id + "), frontType=" + frontType);