27171677342

Committed 08 Jun 2026 10:43PM UTC coverage: 51.99% (+0.05%) from 51.937%

Build # 27171677342

Build Type

Pull #77

github

Committed by

web-flow

Commit Message

Merge 3a1432eec into 8045f0be3

Pull Request Pull Request #77: chore: bump version to 0.0.10

Coverage Stats

36972 of 92663 branches covered (39.9%)

Branch coverage included in aggregate %.

33405 of 42703 relevant lines covered (78.23%)

20411.31 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

49.97

/src/dftracer/utils/server/trace_api.cpp

#include <dftracer/utils/core/common/filesystem.h>
#include <dftracer/utils/core/common/logging.h>
#include <dftracer/utils/core/common/transparent_string_hash.h>
#include <dftracer/utils/core/coro/channel.h>
#include <dftracer/utils/core/pipeline/executor.h>
#include <dftracer/utils/core/tasks/coro_scope.h>
#include <dftracer/utils/server/cursor.h>
#include <dftracer/utils/server/http_request.h>
#include <dftracer/utils/server/http_response.h>
#include <dftracer/utils/server/router.h>
#include <dftracer/utils/server/trace_api.h>
#include <dftracer/utils/server/trace_index.h>
#include <dftracer/utils/utilities/common/json/json_doc_guard.h>
#include <dftracer/utils/utilities/common/json/json_value.h>
#include <dftracer/utils/utilities/common/query/query.h>
#include <dftracer/utils/utilities/composites/dft/indexing/index_resolver_utility.h>
#include <dftracer/utils/utilities/composites/dft/internal/utils.h>
#include <dftracer/utils/utilities/composites/dft/statistics/shared_index_statistics_reader.h>
#include <dftracer/utils/utilities/composites/dft/statistics/statistics_aggregator_utility.h>
#include <dftracer/utils/utilities/composites/dft/statistics/statistics_query_utility.h>
#include <dftracer/utils/utilities/composites/dft/views/view_builder_utility.h>
#include <dftracer/utils/utilities/composites/dft/views/view_definition.h>
#include <dftracer/utils/utilities/composites/dft/views/view_reader_utility.h>
#include <dftracer/utils/utilities/fileio/lines/sources/async_streaming_gz_line_generator.h>

#include <atomic>
#include <cstddef>
#include <limits>
#include <mutex>
#include <string>
#include <unordered_map>
#include <unordered_set>
#include <vector>

namespace dftracer::utils::server {

using namespace dftracer::utils::utilities::composites::dft;
using namespace dftracer::utils::utilities::composites::dft::indexing;
using namespace dftracer::utils::utilities::composites::dft::statistics;
using namespace dftracer::utils::utilities::composites::dft::views;

// JSON-escape a string value (minimal: quotes, backslash, control chars).
static std::string json_escape(const std::string& s) {
    std::string out;
    out.reserve(s.size() + 2);
    for (char c : s) {
        switch (c) {
            case '"':
                out += "\\\"";
                break;
            case '\\':
                out += "\\\\";
                break;
            case '\n':
                out += "\\n";
                break;
            case '\r':
                out += "\\r";
                break;
            case '\t':
                out += "\\t";
                break;
            default:
                out += c;
                break;
        }
    }
    return out;
}

using dftracer::utils::utilities::common::json::JsonValue;

// Hash metadata types that need smart filtering (FH, HH, SH).
static const std::unordered_set<std::string> HASH_METADATA_NAMES = {"FH", "HH",
                                                                    "SH"};

using dftracer::utils::utilities::common::json::JsonDocGuard;
using dftracer::utils::utilities::common::query::Query;

// --- GET /api/v1/files ---
static coro::CoroTask<HttpResponse> handle_files(const HttpRequest& /*req*/,
                                                 const QueryParams& /*params*/,
                                                 TraceIndex& index) {
    std::string body;
    body.reserve(128 * index.file_count() + 32);
    body += "{\"files\":[";
    bool first = true;
    for (const auto& f : index.files()) {
        if (!first) body += ',';
        first = false;
        body += "{\"path\":\"";
        body += json_escape(f.path);
        body += "\",\"has_bloom_data\":";
        body += f.has_bloom_data ? "true" : "false";
        body += ",\"has_checkpoint_index\":";
        body += f.has_checkpoint_index ? "true" : "false";
        body += '}';
    }
    body += "],\"count\":";
    body += std::to_string(index.file_count());
    body += '}';
    co_return HttpResponse::ok(body);
}

// --- GET /api/v1/files/info ---
static coro::CoroTask<HttpResponse> handle_file_info(const HttpRequest& /*req*/,
                                                     const QueryParams& params,
                                                     TraceIndex& index) {
    auto file_param = params.get("file");
    if (file_param.empty()) {
        co_return HttpResponse::bad_request("Missing required parameter: file");
    }

    std::string file_path(file_param);
    auto* info = index.find_file(file_path);
    if (!info) {
        co_return HttpResponse::not_found();
    }

    std::string body;
    body.reserve(512);
    body += "{\"path\":\"";
    body += json_escape(info->path);
    body += "\",\"has_bloom_data\":";
    body += info->has_bloom_data ? "true" : "false";
    body += ",\"has_checkpoint_index\":";
    body += info->has_checkpoint_index ? "true" : "false";
    body += ",\"size_mb\":";
    body += std::to_string(info->size_mb);
    body += ",\"compressed_size\":";
    body += std::to_string(info->compressed_size);
    body += ",\"num_lines\":";
    body += std::to_string(info->num_lines);
    body += ",\"num_checkpoints\":";
    body += std::to_string(info->num_checkpoints);
    body += ",\"uncompressed_size\":";
    body += std::to_string(info->uncompressed_size);

    body += '}';
    co_return HttpResponse::ok(body);
}

static std::vector<std::string> split_csv(std::string_view s) {
    std::vector<std::string> result;
    std::string token;
    for (char c : s) {
        if (c == ',') {
            if (!token.empty()) result.push_back(token);
            token.clear();
        } else {
            token += c;
        }
    }
    if (!token.empty()) result.push_back(token);
    return result;
}

static std::string format_in_clause(const std::string& field,
                                    const std::vector<std::string>& vals) {
    if (vals.size() == 1) return field + " == \"" + vals[0] + "\"";
    std::string s = field + " in [";
    for (std::size_t i = 0; i < vals.size(); ++i) {
        if (i > 0) s += ", ";
        s += "\"" + vals[i] + "\"";
    }
    s += "]";
    return s;
}

static std::optional<Query> build_query_from_params(const QueryParams& params) {
    std::string dsl;

    auto cat = params.get("cat");
    if (!cat.empty()) {
        auto vals = split_csv(cat);
        if (!vals.empty()) dsl += format_in_clause("cat", vals);
    }

    auto name = params.get("name");
    if (!name.empty()) {
        auto vals = split_csv(name);
        if (!vals.empty()) {
            if (!dsl.empty()) dsl += " and ";
            dsl += format_in_clause("name", vals);
        }
    }

    auto pid = params.get("pid");
    if (!pid.empty()) {
        if (!dsl.empty()) dsl += " and ";
        dsl += "pid == " + std::string(pid);
    }

    double ts_min = params.get_double("ts_min", 0);
    double ts_max = params.get_double("ts_max", 0);
    if (ts_min > 0) {
        if (!dsl.empty()) dsl += " and ";
        dsl += "ts >= " + std::to_string(static_cast<uint64_t>(ts_min));
    }
    if (ts_max > 0) {
        if (!dsl.empty()) dsl += " and ";
        dsl += "ts <= " + std::to_string(static_cast<uint64_t>(ts_max));
    }

    double dur_min = params.get_double("dur_min", 0);
    double dur_max = params.get_double("dur_max", 0);
    if (dur_min > 0) {
        if (!dsl.empty()) dsl += " and ";
        dsl += "dur >= " + std::to_string(static_cast<uint64_t>(dur_min));
    }
    if (dur_max > 0) {
        if (!dsl.empty()) dsl += " and ";
        dsl += "dur <= " + std::to_string(static_cast<uint64_t>(dur_max));
    }

    if (dsl.empty()) return std::nullopt;
    auto result = Query::from_string(dsl);
    if (!result) return std::nullopt;
    return std::move(*result);
}

static ViewDefinition build_view_from_params(const QueryParams& params) {
    ViewDefinition view;
    view.name = "api_query";
    view.description = "HTTP API query";

    auto q = build_query_from_params(params);
    if (q) view.with_query(std::move(*q));
    return view;
}

// ============================================================================
// Shared helpers for event streaming endpoints
// ============================================================================

static std::vector<const TraceIndex::FileInfo*> resolve_target_files(
    TraceIndex& index, const QueryParams& params, double ts_min = 0,
    double ts_max = 0) {
    std::vector<const TraceIndex::FileInfo*> files;
    auto file_param = params.get("file");
    if (!file_param.empty()) {
        auto* f = index.find_file(std::string(file_param));
        if (f) files.push_back(f);
    } else {
        for (const auto& f : index.files()) {
            files.push_back(&f);
        }
    }

    if (ts_min > 0 || ts_max > 0) {
        std::vector<const TraceIndex::FileInfo*> filtered;
        filtered.reserve(files.size());
        for (auto* fi : files) {
            if (fi->min_timestamp_us == 0 && fi->max_timestamp_us == 0) {
                filtered.push_back(fi);
                continue;
            }
            double fi_min = static_cast<double>(fi->min_timestamp_us);
            double fi_max = static_cast<double>(fi->max_timestamp_us);
            if (fi_max < ts_min || (ts_max > 0 && fi_min > ts_max)) continue;
            filtered.push_back(fi);
        }
        files = std::move(filtered);
    }

    return files;
}

using StreamChunk = HttpResponse::StreamChunk;

static coro::AsyncGenerator<StreamChunk> stream_events(
    std::vector<const TraceIndex::FileInfo*> files, ViewDefinition ev_view,
    std::optional<Query> /*query_opt*/, double ts_min, double ts_max,
    BloomFilterCache* bloom_cache, int limit) {
    int emitted = 0;

    for (auto* file_info : files) {
        if (limit > 0 && emitted >= limit) break;

        if (file_info->uncompressed_size == 0 &&
            file_info->num_checkpoints == 0)
            continue;

        ViewBuilderInput builder_input;
        builder_input.with_view(ev_view)
            .with_file_path(file_info->path)
            .with_index_path(file_info->has_bloom_data ? file_info->index_path
                                                       : "")
            .with_uncompressed_size(file_info->uncompressed_size)
            .with_num_checkpoints(file_info->num_checkpoints)
            .with_bloom_cache(bloom_cache)
            .with_time_range(ts_min, ts_max);

        ViewBuilderUtility builder;
        auto build_output = co_await builder.process(builder_input);
        if (!build_output.success || !build_output.file_may_match) continue;

        for (const auto& candidate : build_output.candidates) {
            if (limit > 0 && emitted >= limit) break;

            ViewReaderInput reader_input;
            reader_input.with_file_path(file_info->path)
                .with_index_path(file_info->index_path)
                .with_byte_range(candidate.start_byte, candidate.end_byte)
                .with_checkpoint_idx(candidate.checkpoint_idx)
                .with_view(ev_view);

            ViewReaderUtility reader;
            auto event_gen = reader.process(reader_input);
            while (auto batch = co_await event_gen.next()) {
                int count = std::min(
                    static_cast<int>(batch->events.size()),
                    limit > 0 ? limit - emitted
                              : static_cast<int>(batch->events.size()));
                if (count > 0) {
                    co_yield StreamChunk{std::span<const std::string_view>(
                        batch->events.data(), static_cast<std::size_t>(count))};
                    emitted += count;
                }
            }
        }
    }
}

// ============================================================================
// Event endpoints
// ============================================================================

// --- GET /api/v1/events ---
static coro::CoroTask<HttpResponse> handle_events(const HttpRequest& /*req*/,
                                                  const QueryParams& params,
                                                  TraceIndex& index) {
    int limit = params.get_int("limit", 1000);
    if (limit <= 0) limit = 1000;
    if (limit > 100000) limit = 100000;

    double ts_min = params.get_double("ts_min", 0);
    double ts_max = params.get_double("ts_max", 0);
    auto files = resolve_target_files(index, params, ts_min, ts_max);
    auto view = build_view_from_params(params);
    auto query = build_query_from_params(params);

    auto gen = std::make_unique<HttpResponse::StreamGenerator>(
        stream_events(std::move(files), std::move(view), std::move(query),
                      ts_min, ts_max, &index.bloom_cache(), limit));

    auto resp = HttpResponse::streaming(std::move(gen));
    resp.headers.push_back({"X-Limit", std::to_string(limit)});
    co_return resp;
}

// --- GET /api/v1/events/stream ---
static coro::CoroTask<HttpResponse> handle_events_stream(
    const HttpRequest& /*req*/, const QueryParams& params, TraceIndex& index) {
    double ts_min = params.get_double("ts_min", 0);
    double ts_max = params.get_double("ts_max", 0);
    auto files = resolve_target_files(index, params, ts_min, ts_max);
    auto view = build_view_from_params(params);
    auto query = build_query_from_params(params);
    int limit = params.get_int("limit", 0);

    auto gen = std::make_unique<HttpResponse::StreamGenerator>(
        stream_events(std::move(files), std::move(view), std::move(query),
                      ts_min, ts_max, &index.bloom_cache(), limit));

    co_return HttpResponse::streaming(std::move(gen));
}

// --- GET /api/v1/stats ---
static coro::CoroTask<HttpResponse> handle_stats(const HttpRequest& req,
                                                 const QueryParams& /*params*/,
                                                 TraceIndex& index) {
    static std::mutex cache_mutex;
    static std::unordered_map<std::string, std::string,
                              dftracer::utils::TransparentStringHash,
                              dftracer::utils::TransparentStringEqual>
        stats_cache;

    {
        std::lock_guard<std::mutex> lock(cache_mutex);
        auto it = stats_cache.find(req.path);
        if (it != stats_cache.end()) {
            co_return HttpResponse::ok(it->second);
        }
    }

    std::vector<TraceStatistics> all_stats;

    // Group files by index_path
    std::unordered_map<std::string,
                       std::vector<std::pair<std::size_t, std::string>>>
        files_by_index;
    std::size_t file_idx = 0;
    for (const auto& file_info : index.files()) {
        if (!file_info.has_bloom_data) continue;
        files_by_index[file_info.index_path].emplace_back(file_idx++,
                                                          file_info.path);
    }

    // Resolve each group and read statistics
    for (auto& [idx_path, files] : files_by_index) {
        std::vector<std::string> file_paths;
        file_paths.reserve(files.size());
        for (const auto& [_, path] : files) {
            file_paths.push_back(path);
        }

        IndexResolverUtility resolver;
        ResolverInput input;
        input.files = std::move(file_paths);
        input.require_checkpoints = false;

        auto result = co_await resolver.process(input);

        if (result.cached.empty()) {
            continue;
        }

        try {
            SharedIndexStatisticsReader reader;
            auto batch_rows = co_await reader.query(
                result.index_path, std::move(result.cached),
                StatisticsQueryType::SUMMARY);
            auto callback = [&all_stats](std::size_t /*file_index*/,
                                         TraceStatistics&& stats) {
                if (stats.success) {
                    all_stats.push_back(std::move(stats));
                }
            };
            SharedIndexStatisticsReader::process_batch_results(batch_rows,
                                                               callback);
        } catch (const std::exception& e) {
            DFTRACER_UTILS_LOG_WARN("Server stats batch read failed for %s: %s",
                                    idx_path.c_str(), e.what());
        }
    }

    std::uint64_t total_events = 0;
    std::size_t file_count = all_stats.size();
    for (const auto& s : all_stats) {
        total_events += s.total_events();
    }

    std::string body;
    body.reserve(256 * all_stats.size() + 64);
    body += "{\"file_count\":";
    body += std::to_string(file_count);
    body += ",\"total_events\":";
    body += std::to_string(total_events);
    body += ",\"files\":[";
    for (std::size_t i = 0; i < all_stats.size(); ++i) {
        if (i > 0) body += ',';
        body += all_stats[i].to_json();
    }
    body += "]}";

    {
        std::lock_guard<std::mutex> lock(cache_mutex);
        stats_cache.emplace(std::string(req.path), body);
    }
    co_return HttpResponse::ok(body);
}

// --- GET /api/v1/info ---
static coro::CoroTask<HttpResponse> handle_info(const HttpRequest& /*req*/,
                                                const QueryParams& /*params*/,
                                                TraceIndex& index) {
    auto global_min = index.global_min_timestamp_us();
    auto global_max = index.global_max_timestamp_us();
    bool has_time_range =
        global_max > 0 &&
        global_min != std::numeric_limits<std::uint64_t>::max();

    std::string body;
    body.reserve(256 * index.file_count() + 128);
    body += "{\"file_count\":";
    body += std::to_string(index.file_count());

    if (has_time_range) {
        body += ",\"time_range\":{\"min_timestamp_us\":";
        body += std::to_string(global_min);
        body += ",\"max_timestamp_us\":";
        body += std::to_string(global_max);
        body += "}";
    }

    body += ",\"files\":[";
    bool first = true;
    for (const auto& f : index.files()) {
        if (!first) body += ',';
        first = false;
        body += "{\"path\":\"";
        body += json_escape(f.path);
        body += "\",\"has_bloom_data\":";
        body += f.has_bloom_data ? "true" : "false";
        body += ",\"has_checkpoint_index\":";
        body += f.has_checkpoint_index ? "true" : "false";
        if (f.min_timestamp_us > 0 || f.max_timestamp_us > 0) {
            body += ",\"min_timestamp_us\":";
            body += std::to_string(f.min_timestamp_us);
            body += ",\"max_timestamp_us\":";
            body += std::to_string(f.max_timestamp_us);
        }
        body += '}';
    }
    body += "]}";
    co_return HttpResponse::ok(body);
}

void register_trace_api(Router& router, TraceIndex& index) {
    auto* index_ptr = &index;

    router.get(
        "/api/v1/files",
        [index_ptr](const HttpRequest& req,
                    const QueryParams& params) -> coro::CoroTask<HttpResponse> {
            co_return co_await handle_files(req, params, *index_ptr);
        });

    router.get(
        "/api/v1/files/info",
        [index_ptr](const HttpRequest& req,
                    const QueryParams& params) -> coro::CoroTask<HttpResponse> {
            co_return co_await handle_file_info(req, params, *index_ptr);
        });

    router.get(
        "/api/v1/events",
        [index_ptr](const HttpRequest& req,
                    const QueryParams& params) -> coro::CoroTask<HttpResponse> {
            co_return co_await handle_events(req, params, *index_ptr);
        });

    router.get(
        "/api/v1/events/stream",
        [index_ptr](const HttpRequest& req,
                    const QueryParams& params) -> coro::CoroTask<HttpResponse> {
            co_return co_await handle_events_stream(req, params, *index_ptr);
        });

    router.get(
        "/api/v1/stats",
        [index_ptr](const HttpRequest& req,
                    const QueryParams& params) -> coro::CoroTask<HttpResponse> {
            co_return co_await handle_stats(req, params, *index_ptr);
        });

    router.get(
        "/api/v1/info",
        [index_ptr](const HttpRequest& req,
                    const QueryParams& params) -> coro::CoroTask<HttpResponse> {
            co_return co_await handle_info(req, params, *index_ptr);
        });
}

}  // namespace dftracer::utils::server

1	#include <dftracer/utils/core/common/filesystem.h>
2	#include <dftracer/utils/core/common/logging.h>
3	#include <dftracer/utils/core/common/transparent_string_hash.h>
4	#include <dftracer/utils/core/coro/channel.h>
5	#include <dftracer/utils/core/pipeline/executor.h>
6	#include <dftracer/utils/core/tasks/coro_scope.h>
7	#include <dftracer/utils/server/cursor.h>
8	#include <dftracer/utils/server/http_request.h>
9	#include <dftracer/utils/server/http_response.h>
10	#include <dftracer/utils/server/router.h>
11	#include <dftracer/utils/server/trace_api.h>
12	#include <dftracer/utils/server/trace_index.h>
13	#include <dftracer/utils/utilities/common/json/json_doc_guard.h>
14	#include <dftracer/utils/utilities/common/json/json_value.h>
15	#include <dftracer/utils/utilities/common/query/query.h>
16	#include <dftracer/utils/utilities/composites/dft/indexing/index_resolver_utility.h>
17	#include <dftracer/utils/utilities/composites/dft/internal/utils.h>
18	#include <dftracer/utils/utilities/composites/dft/statistics/shared_index_statistics_reader.h>
19	#include <dftracer/utils/utilities/composites/dft/statistics/statistics_aggregator_utility.h>
20	#include <dftracer/utils/utilities/composites/dft/statistics/statistics_query_utility.h>
21	#include <dftracer/utils/utilities/composites/dft/views/view_builder_utility.h>
22	#include <dftracer/utils/utilities/composites/dft/views/view_definition.h>
23	#include <dftracer/utils/utilities/composites/dft/views/view_reader_utility.h>
24	#include <dftracer/utils/utilities/fileio/lines/sources/async_streaming_gz_line_generator.h>
25
26	#include <atomic>
27	#include <cstddef>
28	#include <limits>
29	#include <mutex>
30	#include <string>
31	#include <unordered_map>
32	#include <unordered_set>
33	#include <vector>
34
35	namespace dftracer::utils::server {
36
37	using namespace dftracer::utils::utilities::composites::dft;
38	using namespace dftracer::utils::utilities::composites::dft::indexing;
39	using namespace dftracer::utils::utilities::composites::dft::statistics;
40	using namespace dftracer::utils::utilities::composites::dft::views;
41
42	// JSON-escape a string value (minimal: quotes, backslash, control chars).
43	static std::string json_escape(const std::string& s) {	8✔
44	std::string out;	8✔
45	out.reserve(s.size() + 2);	8!
46	for (char c : s) {	740✔
47	switch (c) {	732!
48	case '"':
49	out += "\\\"";	×
50	break;	×
51	case '\\':
52	out += "\\\\";	×
53	break;	×
54	case '\n':
55	out += "\\n";	×
56	break;	×
57	case '\r':
58	out += "\\r";	×
59	break;	×
60	case '\t':
61	out += "\\t";	×
62	break;	×
63	default:	280✔
64	out += c;	732!
65	break;	732✔
66	}
67	}
68	return out;	8✔
69	}	4!
70
71	using dftracer::utils::utilities::common::json::JsonValue;
72
73	// Hash metadata types that need smart filtering (FH, HH, SH).
74	static const std::unordered_set<std::string> HASH_METADATA_NAMES = {"FH", "HH",	256!
75	"SH"};	256!
76
77	using dftracer::utils::utilities::common::json::JsonDocGuard;
78	using dftracer::utils::utilities::common::query::Query;
79
80	// --- GET /api/v1/files ---
81	static coro::CoroTask<HttpResponse> handle_files(const HttpRequest& /req/,	12!
82	const QueryParams& /params/,
83	TraceIndex& index) {	2!
84	std::string body;	2✔
85	body.reserve(128 * index.file_count() + 32);	2!
86	body += "{\"files\":[";	2!
87	bool first = true;	2✔
88	for (const auto& f : index.files()) {	4!
89	if (!first) body += ',';	2!
90	first = false;	2✔
91	body += "{\"path\":\"";	2!
92	body += json_escape(f.path);	2!
93	body += "\",\"has_bloom_data\":";	2!
94	body += f.has_bloom_data ? "true" : "false";	2!
95	body += ",\"has_checkpoint_index\":";	2!
96	body += f.has_checkpoint_index ? "true" : "false";	2!
97	body += '}';	2!
98	}	2✔
99	body += "],\"count\":";	2!
100	body += std::to_string(index.file_count());	2!
101	body += '}';	2!
102	co_return HttpResponse::ok(body);	4!
103	}	6!
104
105	// --- GET /api/v1/files/info ---
106	static coro::CoroTask<HttpResponse> handle_file_info(const HttpRequest& /req/,	16!
107	const QueryParams& params,
108	TraceIndex& index) {	2!
109	auto file_param = params.get("file");	2!
110	if (file_param.empty()) {	2✔
111	co_return HttpResponse::bad_request("Missing required parameter: file");	3!
112	}
113
114	std::string file_path(file_param);	1!
115	auto* info = index.find_file(file_path);	1!
116	if (!info) {	1!
117	co_return HttpResponse::not_found();	×
118	}
119
120	std::string body;	1✔
121	body.reserve(512);	1!
122	body += "{\"path\":\"";	1!
123	body += json_escape(info->path);	1!
124	body += "\",\"has_bloom_data\":";	1!
125	body += info->has_bloom_data ? "true" : "false";	1!
126	body += ",\"has_checkpoint_index\":";	1!
127	body += info->has_checkpoint_index ? "true" : "false";	1!
128	body += ",\"size_mb\":";	1!
129	body += std::to_string(info->size_mb);	1!
130	body += ",\"compressed_size\":";	1!
131	body += std::to_string(info->compressed_size);	1!
132	body += ",\"num_lines\":";	1!
133	body += std::to_string(info->num_lines);	1!
134	body += ",\"num_checkpoints\":";	1!
135	body += std::to_string(info->num_checkpoints);	1!
136	body += ",\"uncompressed_size\":";	1!
137	body += std::to_string(info->uncompressed_size);	1!
138
139	body += '}';	1!
140	co_return HttpResponse::ok(body);	1!
141	}	6!
142
143	static std::vector<std::string> split_csv(std::string_view s) {	×
144	std::vector<std::string> result;	×
145	std::string token;	×
146	for (char c : s) {	×
147	if (c == ',') {	×
148	if (!token.empty()) result.push_back(token);	×
149	token.clear();	×
150	} else {
151	token += c;	×
152	}
153	}
154	if (!token.empty()) result.push_back(token);	×
155	return result;	×
156	}	×
157
158	static std::string format_in_clause(const std::string& field,	×
159	const std::vector<std::string>& vals) {
160	if (vals.size() == 1) return field + " == \"" + vals[0] + "\"";	×
161	std::string s = field + " in [";	×
162	for (std::size_t i = 0; i < vals.size(); ++i) {	×
163	if (i > 0) s += ", ";	×
164	s += "\"" + vals[i] + "\"";	×
165	}
166	s += "]";	×
167	return s;	×
168	}	×
169
170	static std::optional<Query> build_query_from_params(const QueryParams& params) {	8✔
171	std::string dsl;	8✔
172
173	auto cat = params.get("cat");	8!
174	if (!cat.empty()) {	8!
175	auto vals = split_csv(cat);	×
176	if (!vals.empty()) dsl += format_in_clause("cat", vals);	×
177	}	×
178
179	auto name = params.get("name");	8!
180	if (!name.empty()) {	8!
181	auto vals = split_csv(name);	×
182	if (!vals.empty()) {	×
183	if (!dsl.empty()) dsl += " and ";	×
184	dsl += format_in_clause("name", vals);	×
185	}
186	}	×
187
188	auto pid = params.get("pid");	8!
189	if (!pid.empty()) {	8!
190	if (!dsl.empty()) dsl += " and ";	×
191	dsl += "pid == " + std::string(pid);	×
192	}
193
194	double ts_min = params.get_double("ts_min", 0);	8!
195	double ts_max = params.get_double("ts_max", 0);	8!
196	if (ts_min > 0) {	8!
197	if (!dsl.empty()) dsl += " and ";	×
198	dsl += "ts >= " + std::to_string(static_cast<uint64_t>(ts_min));	×
199	}
200	if (ts_max > 0) {	8!
201	if (!dsl.empty()) dsl += " and ";	×
202	dsl += "ts <= " + std::to_string(static_cast<uint64_t>(ts_max));	×
203	}
204
205	double dur_min = params.get_double("dur_min", 0);	8!
206	double dur_max = params.get_double("dur_max", 0);	8!
207	if (dur_min > 0) {	8!
208	if (!dsl.empty()) dsl += " and ";	×
209	dsl += "dur >= " + std::to_string(static_cast<uint64_t>(dur_min));	×
210	}
211	if (dur_max > 0) {	8!
212	if (!dsl.empty()) dsl += " and ";	×
213	dsl += "dur <= " + std::to_string(static_cast<uint64_t>(dur_max));	×
214	}
215
216	if (dsl.empty()) return std::nullopt;	8✔
217	auto result = Query::from_string(dsl);	×
218	if (!result) return std::nullopt;	×
219	return std::move(*result);	×
220	}	8✔
221
222	static ViewDefinition build_view_from_params(const QueryParams& params) {	4✔
223	ViewDefinition view;	4✔
224	view.name = "api_query";	4!
225	view.description = "HTTP API query";	4!
226
227	auto q = build_query_from_params(params);	4!
228	if (q) view.with_query(std::move(*q));	4!
229	return view;	6✔
230	}	4!
231
232	// ============================================================================
233	// Shared helpers for event streaming endpoints
234	// ============================================================================
235
236	static std::vector<const TraceIndex::FileInfo*> resolve_target_files(	4✔
237	TraceIndex& index, const QueryParams& params, double ts_min = 0,
238	double ts_max = 0) {
239	std::vector<const TraceIndex::FileInfo*> files;	4✔
240	auto file_param = params.get("file");	4!
241	if (!file_param.empty()) {	4!
242	auto* f = index.find_file(std::string(file_param));	×
243	if (f) files.push_back(f);	×
244	} else {
245	for (const auto& f : index.files()) {	8✔
246	files.push_back(&f);	4!
247	}
248	}
249
250	if (ts_min > 0 \|\| ts_max > 0) {	4!
251	std::vector<const TraceIndex::FileInfo*> filtered;	×
252	filtered.reserve(files.size());	×
253	for (auto* fi : files) {	×
254	if (fi->min_timestamp_us == 0 && fi->max_timestamp_us == 0) {	×
255	filtered.push_back(fi);	×
256	continue;	×
257	}
258	double fi_min = static_cast<double>(fi->min_timestamp_us);	×
259	double fi_max = static_cast<double>(fi->max_timestamp_us);	×
260	if (fi_max < ts_min \|\| (ts_max > 0 && fi_min > ts_max)) continue;	×
261	filtered.push_back(fi);	×
262	}
263	files = std::move(filtered);	×
264	}	×
265
266	return files;	6✔
267	}	2!
268
269	using StreamChunk = HttpResponse::StreamChunk;
270
271	static coro::AsyncGenerator<StreamChunk> stream_events(	20!
272	std::vector<const TraceIndex::FileInfo*> files, ViewDefinition ev_view,
273	std::optional<Query> /query_opt/, double ts_min, double ts_max,
274	BloomFilterCache* bloom_cache, int limit) {	2!
275	int emitted = 0;	2✔
276
277	for (auto* file_info : files) {	8✔
278	if (limit > 0 && emitted >= limit) break;	2!
279
280	if (file_info->uncompressed_size == 0 &&	2!
281	file_info->num_checkpoints == 0)
282	continue;
283
284	ViewBuilderInput builder_input;	2✔
285	builder_input.with_view(ev_view)	4!
286	.with_file_path(file_info->path)	2!
287	.with_index_path(file_info->has_bloom_data ? file_info->index_path	2!
288	: "")	×
289	.with_uncompressed_size(file_info->uncompressed_size)	2!
290	.with_num_checkpoints(file_info->num_checkpoints)	2!
291	.with_bloom_cache(bloom_cache)	2!
292	.with_time_range(ts_min, ts_max);	2!
293
294	ViewBuilderUtility builder;	2!
295	auto build_output = co_await builder.process(builder_input);	6!
296	if (!build_output.success \|\| !build_output.file_may_match) continue;	2!
297
298	for (const auto& candidate : build_output.candidates) {	8✔
299	if (limit > 0 && emitted >= limit) break;	2!
300
301	ViewReaderInput reader_input;	2✔
302	reader_input.with_file_path(file_info->path)	2!
303	.with_index_path(file_info->index_path)	2!
304	.with_byte_range(candidate.start_byte, candidate.end_byte)	2!
305	.with_checkpoint_idx(candidate.checkpoint_idx)	2!
306	.with_view(ev_view);	2!
307
308	ViewReaderUtility reader;	2!
309	auto event_gen = reader.process(reader_input);	2!
310	while (auto batch = co_await event_gen.next()) {	16!
311	int count = std::min(	4!
312	static_cast<int>(batch->events.size()),	2✔
313	limit > 0 ? limit - emitted	2✔
314	: static_cast<int>(batch->events.size()));	1✔
315	if (count > 0) {	2✔
316	co_yield StreamChunk{std::span<const std::string_view>(	14!
317	batch->events.data(), static_cast<std::size_t>(count))};	6✔
318	emitted += count;	2✔
319	}	2✔
320	}	8✔
321	}	6✔
322	}	6!
323	}	34!
324
325	// ============================================================================
326	// Event endpoints
327	// ============================================================================
328
329	// --- GET /api/v1/events ---
330	static coro::CoroTask<HttpResponse> handle_events(const HttpRequest& /req/,	6!
331	const QueryParams& params,
332	TraceIndex& index) {	1!
333	int limit = params.get_int("limit", 1000);	1!
334	if (limit <= 0) limit = 1000;	1!
335	if (limit > 100000) limit = 100000;	1!
336
337	double ts_min = params.get_double("ts_min", 0);	1!
338	double ts_max = params.get_double("ts_max", 0);	1!
339	auto files = resolve_target_files(index, params, ts_min, ts_max);	1!
340	auto view = build_view_from_params(params);	1!
341	auto query = build_query_from_params(params);	1!
342
343	auto gen = std::make_unique<HttpResponse::StreamGenerator>(	2!
344	stream_events(std::move(files), std::move(view), std::move(query),	1!
345	ts_min, ts_max, &index.bloom_cache(), limit));	1!
346
347	auto resp = HttpResponse::streaming(std::move(gen));	1!
348	resp.headers.push_back({"X-Limit", std::to_string(limit)});	1!
349	co_return resp;	2✔
350	}	3!
351
352	// --- GET /api/v1/events/stream ---
353	static coro::CoroTask<HttpResponse> handle_events_stream(	6!
354	const HttpRequest& /req/, const QueryParams& params, TraceIndex& index) {	1!
355	double ts_min = params.get_double("ts_min", 0);	1!
356	double ts_max = params.get_double("ts_max", 0);	1!
357	auto files = resolve_target_files(index, params, ts_min, ts_max);	1!
358	auto view = build_view_from_params(params);	1!
359	auto query = build_query_from_params(params);	1!
360	int limit = params.get_int("limit", 0);	1!
361
362	auto gen = std::make_unique<HttpResponse::StreamGenerator>(	1!
363	stream_events(std::move(files), std::move(view), std::move(query),	2!
364	ts_min, ts_max, &index.bloom_cache(), limit));	1✔
365
366	co_return HttpResponse::streaming(std::move(gen));	2!
367	}	3!
368
369	// --- GET /api/v1/stats ---
370	static coro::CoroTask<HttpResponse> handle_stats(const HttpRequest& req,	6!
371	const QueryParams& /params/,
372	TraceIndex& index) {	1!
373	static std::mutex cache_mutex;	1!
374	static std::unordered_map<std::string, std::string,	1!
375	dftracer::utils::TransparentStringHash,
376	dftracer::utils::TransparentStringEqual>
377	stats_cache;	1✔
378
379	{
380	std::lock_guard<std::mutex> lock(cache_mutex);	1!
381	auto it = stats_cache.find(req.path);	1!
382	if (it != stats_cache.end()) {	1!
383	co_return HttpResponse::ok(it->second);	1!
384	}
385	}	1!
386
387	std::vector<TraceStatistics> all_stats;	1✔
388
389	// Group files by index_path
390	std::unordered_map<std::string,	1✔
391	std::vector<std::pair<std::size_t, std::string>>>
392	files_by_index;	1✔
393	std::size_t file_idx = 0;	1✔
394	for (const auto& file_info : index.files()) {	2✔
395	if (!file_info.has_bloom_data) continue;	1!
396	files_by_index[file_info.index_path].emplace_back(file_idx++,	1!
397	file_info.path);	1✔
398	}	1!
399
400	// Resolve each group and read statistics
401	for (auto& [idx_path, files] : files_by_index) {	4!
402	std::vector<std::string> file_paths;	3✔
403	file_paths.reserve(files.size());	3!
404	for (const auto& [_, path] : files) {	7✔
405	file_paths.push_back(path);	1!
406	}	1✔
407
408	IndexResolverUtility resolver;	3!
409	ResolverInput input;	3✔
410	input.files = std::move(file_paths);	3✔
411	input.require_checkpoints = false;	3✔
412
413	auto result = co_await resolver.process(input);	4!
414
415	if (result.cached.empty()) {	3!
416	continue;
417	}
418
419	try {
420	SharedIndexStatisticsReader reader;	3✔
421	auto batch_rows = co_await reader.query(	4!
422	result.index_path, std::move(result.cached),	3!
423	StatisticsQueryType::SUMMARY);
424	auto callback = [&all_stats](std::size_t /file_index/,	3✔
425	TraceStatistics&& stats) {	1✔
426	if (stats.success) {	2✔
427	all_stats.push_back(std::move(stats));	2✔
428	}	1✔
429	};	2✔
430	SharedIndexStatisticsReader::process_batch_results(batch_rows,	1!
431	callback);
432	} catch (const std::exception& e) {	1!
433	DFTRACER_UTILS_LOG_WARN("Server stats batch read failed for %s: %s",	×
434	idx_path.c_str(), e.what());
435	}	×
436	}	3!
437
438	std::uint64_t total_events = 0;	1✔
439	std::size_t file_count = all_stats.size();	1✔
440	for (const auto& s : all_stats) {	2✔
441	total_events += s.total_events();	1!
442	}	1✔
443
444	std::string body;	1✔
445	body.reserve(256 * all_stats.size() + 64);	1!
446	body += "{\"file_count\":";	1!
447	body += std::to_string(file_count);	1!
448	body += ",\"total_events\":";	1!
449	body += std::to_string(total_events);	1!
450	body += ",\"files\":[";	1!
451	for (std::size_t i = 0; i < all_stats.size(); ++i) {	2✔
452	if (i > 0) body += ',';	1!
453	body += all_stats[i].to_json();	1!
454	}	1✔
455	body += "]}";	1!
456
457	{
458	std::lock_guard<std::mutex> lock(cache_mutex);	1!
459	stats_cache.emplace(std::string(req.path), body);	1!
460	}	1✔
461	co_return HttpResponse::ok(body);	1!
462	}	11!
463
464	// --- GET /api/v1/info ---
465	static coro::CoroTask<HttpResponse> handle_info(const HttpRequest& /req/,	6!
466	const QueryParams& /params/,
467	TraceIndex& index) {	1!
468	auto global_min = index.global_min_timestamp_us();	1!
469	auto global_max = index.global_max_timestamp_us();	1!
470	bool has_time_range =	2✔
471	global_max > 0 &&	1!
472	global_min != std::numeric_limits<std::uint64_t>::max();	1✔
473
474	std::string body;	1✔
475	body.reserve(256 * index.file_count() + 128);	1!
476	body += "{\"file_count\":";	1!
477	body += std::to_string(index.file_count());	1!
478
479	if (has_time_range) {	1!
480	body += ",\"time_range\":{\"min_timestamp_us\":";	1!
481	body += std::to_string(global_min);	1!
482	body += ",\"max_timestamp_us\":";	1!
483	body += std::to_string(global_max);	1!
484	body += "}";	1!
485	}	1✔
486
487	body += ",\"files\":[";	1!
488	bool first = true;	1✔
489	for (const auto& f : index.files()) {	2✔
490	if (!first) body += ',';	1!
491	first = false;	1✔
492	body += "{\"path\":\"";	1!
493	body += json_escape(f.path);	1!
494	body += "\",\"has_bloom_data\":";	1!
495	body += f.has_bloom_data ? "true" : "false";	1!
496	body += ",\"has_checkpoint_index\":";	1!
497	body += f.has_checkpoint_index ? "true" : "false";	1!
498	if (f.min_timestamp_us > 0 \|\| f.max_timestamp_us > 0) {	1!
499	body += ",\"min_timestamp_us\":";	1!
500	body += std::to_string(f.min_timestamp_us);	1!
501	body += ",\"max_timestamp_us\":";	1!
502	body += std::to_string(f.max_timestamp_us);	1!
503	}	1✔
504	body += '}';	1!
505	}	1✔
506	body += "]}";	1!
507	co_return HttpResponse::ok(body);	2!
508	}	3!
509
510	void register_trace_api(Router& router, TraceIndex& index) {	4✔
511	auto* index_ptr = &index;	4✔
512
513	router.get(	4!
514	"/api/v1/files",	2✔
515	[index_ptr](const HttpRequest& req,	18!
516	const QueryParams& params) -> coro::CoroTask<HttpResponse> {	2!
517	co_return co_await handle_files(req, params, *index_ptr);	10!
518	});	8!
519
520	router.get(	4!
521	"/api/v1/files/info",	2✔
522	[index_ptr](const HttpRequest& req,	18!
523	const QueryParams& params) -> coro::CoroTask<HttpResponse> {	2!
524	co_return co_await handle_file_info(req, params, *index_ptr);	10!
525	});	8!
526
527	router.get(	4!
528	"/api/v1/events",	2✔
529	[index_ptr](const HttpRequest& req,	10!
530	const QueryParams& params) -> coro::CoroTask<HttpResponse> {	1!
531	co_return co_await handle_events(req, params, *index_ptr);	5!
532	});	4!
533
534	router.get(	4!
535	"/api/v1/events/stream",	2✔
536	[index_ptr](const HttpRequest& req,	10!
537	const QueryParams& params) -> coro::CoroTask<HttpResponse> {	1!
538	co_return co_await handle_events_stream(req, params, *index_ptr);	5!
539	});	4!
540
541	router.get(	4!
542	"/api/v1/stats",	2✔
543	[index_ptr](const HttpRequest& req,	10!
544	const QueryParams& params) -> coro::CoroTask<HttpResponse> {	1!
545	co_return co_await handle_stats(req, params, *index_ptr);	5!
546	});	4!
547
548	router.get(	4!
549	"/api/v1/info",	2✔
550	[index_ptr](const HttpRequest& req,	10!
551	const QueryParams& params) -> coro::CoroTask<HttpResponse> {	1!
552	co_return co_await handle_info(req, params, *index_ptr);	5!
553	});	4!
554	}	4✔
555
556	} // namespace dftracer::utils::server

llnl / dftracer-utils / 27171677342

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous