44 lines
3.8 KiB
C++
44 lines
3.8 KiB
C++
#include "Taskflow_Trace_Json.hpp"
|
|
#include <algorithm>
|
|
#include <string>
|
|
#include <unordered_map>
|
|
namespace aethera::web {
|
|
nlohmann::json taskflow_trace_json(const Taskflow_Execution_Trace& trace, std::uint64_t sequence, std::uint64_t correlation_id, const nlohmann::json& captured_components, const nlohmann::json& captured_backend) {
|
|
std::unordered_map<std::uint64_t, std::string> node_ids;
|
|
std::unordered_map<std::uint64_t, std::vector<std::uint64_t>> successors;
|
|
for (const auto& node : trace.nodes) {
|
|
node_ids.emplace(node.native_id, node.node_id);
|
|
for (const auto predecessor : node.predecessors) successors[predecessor].push_back(node.native_id);
|
|
}
|
|
nlohmann::json nodes = nlohmann::json::array();
|
|
for (const auto& node : trace.nodes) {
|
|
nlohmann::json predecessors = nlohmann::json::array();
|
|
for (const auto native_id : node.predecessors) predecessors.push_back(std::to_string(native_id));
|
|
nlohmann::json following = nlohmann::json::array();
|
|
for (const auto native_id : successors[node.native_id]) following.push_back(std::to_string(native_id));
|
|
nlohmann::json attributes = nlohmann::json::object();
|
|
for (const auto& [key, value] : node.attributes) attributes[key] = value;
|
|
nlohmann::json encoded{{"native_id", std::to_string(node.native_id)}, {"id", node.node_id}, {"parent_id", node.parent_node_id}, {"name", node.name}, {"type", node.type}, {"predecessors", std::move(predecessors)}, {"successors", std::move(following)}, {"attributes", std::move(attributes)}};
|
|
const auto owner = encoded["attributes"].value("owner_component", std::string{});
|
|
if (!owner.empty() && captured_components.contains(owner)) {
|
|
const auto& captured = captured_components.at(owner);
|
|
encoded["owner"] = {{"component", owner}, {"label", captured.value("label", owner)}, {"kind", captured.value("kind", std::string{})}};
|
|
encoded["prop"] = captured.value("prop", nlohmann::json::object());
|
|
encoded["state"] = captured.value("state", nlohmann::json::object());
|
|
}
|
|
nodes.push_back(std::move(encoded));
|
|
}
|
|
nlohmann::json executions = nlohmann::json::array();
|
|
for (std::size_t worker = 0; worker < trace.worker_tasks.size(); ++worker) {
|
|
for (const auto& task : trace.worker_tasks[worker]) {
|
|
const auto found = node_ids.find(task.native_id);
|
|
executions.push_back({{"native_id", std::to_string(task.native_id)}, {"node_id", found == node_ids.end() ? std::string{} : found->second}, {"worker_id", worker}, {"worker_queue_size", task.worker_queue_size}, {"worker_queue_capacity", task.worker_queue_capacity}, {"ready_ms", task.entered_ms}, {"entered_ms", task.entered_ms}, {"started_ms", task.started_ms}, {"finished_ms", task.finished_ms}, {"completed_ms", task.completed_ms}, {"duration_ms", std::max(0.0, task.finished_ms - task.started_ms)}, {"cooperative_wait_ms", 0.0}, {"cooperative_waits", nlohmann::json::array()}, {"observer_entry_ms", task.entered_ms}, {"observer_exit_ms", task.completed_ms}, {"queue_wait_ms", std::max(0.0, task.started_ms - task.entered_ms)}});
|
|
}
|
|
}
|
|
nlohmann::json graphs = nlohmann::json::array({{{"stage", trace.stage}, {"name", trace.taskflow_name}, {"submitted_ms", 0.0}, {"finished_ms", trace.executor_finished_ms}, {"completed", true}, {"nodes", std::move(nodes)}}});
|
|
nlohmann::json result{{"sequence", sequence}, {"correlation_id", correlation_id}, {"request_source", "diagnostic"}, {"created_time_unix_ns", trace.started_time_unix_ns}, {"worker_count", trace.worker_tasks.size()}, {"markers", nlohmann::json::object()}, {"measurements", nlohmann::json::object()}, {"graphs", std::move(graphs)}, {"executions", std::move(executions)}};
|
|
if (!captured_backend.empty()) result["datoviz"] = captured_backend;
|
|
return result;
|
|
}
|
|
} // namespace aethera::web
|