29187058583

Committed 12 Jul 2026 09:11AM UTC coverage: 51.324% (-1.4%) from 52.754%

Build # 29187058583

Build Type

Pull #96

github

Committed by

web-flow

Commit Message

Merge 7e90b76dd into 056c79287

Pull Request Pull Request #96: fix: consolidate stale-index rebuilds across consumers and fix multi-run breaks

Coverage Stats

33807 of 84432 branches covered (40.04%)

Branch coverage included in aggregate %.

145 of 152 new or added lines in 16 files covered. (95.39%)

5186 existing lines in 197 files now uncovered.

34558 of 48770 relevant lines covered (70.86%)

10797.15 hits per line

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

53.72

/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/json_builder.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 <simdjson.h>

#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;

// 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::query::Query;

// --- GET /api/v1/files ---
static coro::CoroTask<HttpResponse> handle_files(const HttpRequest& /*req*/,
                                                 const QueryParams& /*params*/,
                                                 TraceIndex& index) {
    auto& b = scratch_json_builder();
    b.start_object();
    b.escape_and_append_with_quotes("files");
    b.append_colon();
    b.start_array();
    bool first = true;
    for (const auto& f : index.files()) {
        if (!first) b.append_comma();
        first = false;
        b.start_object();
        b.append_key_value("path", f.path);
        b.append_comma();
        b.append_key_value("has_bloom_data", f.has_bloom_data);
        b.append_comma();
        b.append_key_value("has_checkpoint_index", f.has_checkpoint_index);
        b.end_object();
    }
    b.end_array();
    b.append_comma();
    b.append_key_value("count", static_cast<std::int64_t>(index.file_count()));
    b.end_object();
    co_return HttpResponse::ok(std::string(b));
}

// --- 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();
    }

    auto& b = scratch_json_builder();
    b.start_object();
    b.append_key_value("path", info->path);
    b.append_comma();
    b.append_key_value("has_bloom_data", info->has_bloom_data);
    b.append_comma();
    b.append_key_value("has_checkpoint_index", info->has_checkpoint_index);
    b.append_comma();
    b.append_key_value("size_mb", info->size_mb);
    b.append_comma();
    b.append_key_value("compressed_size",
                       static_cast<std::int64_t>(info->compressed_size));
    b.append_comma();
    b.append_key_value("num_lines", static_cast<std::int64_t>(info->num_lines));
    b.append_comma();
    b.append_key_value("num_checkpoints",
                       static_cast<std::int64_t>(info->num_checkpoints));
    b.append_comma();
    b.append_key_value("uncompressed_size",
                       static_cast<std::int64_t>(info->uncompressed_size));
    b.end_object();
    co_return HttpResponse::ok(std::string(b));
}

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) {
    auto files = collect_candidate_files(index, params);

    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 || !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();
    }

    auto& sb = scratch_json_builder();
    sb.start_object();
    sb.append_key_value("file_count", static_cast<std::int64_t>(file_count));
    sb.append_comma();
    sb.append_key_value("total_events",
                        static_cast<std::int64_t>(total_events));
    sb.append_comma();
    sb.escape_and_append_with_quotes("files");
    sb.append_colon();
    sb.start_array();
    for (std::size_t i = 0; i < all_stats.size(); ++i) {
        if (i > 0) sb.append_comma();
        sb.append_raw(all_stats[i].to_json());  // already JSON
    }
    sb.end_array();
    sb.end_object();
    std::string body(sb);

    {
        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();

    auto& b = scratch_json_builder();
    b.start_object();
    b.append_key_value("file_count",
                       static_cast<std::int64_t>(index.file_count()));

    if (has_time_range) {
        b.append_comma();
        b.escape_and_append_with_quotes("time_range");
        b.append_colon();
        b.start_object();
        b.append_key_value("min_timestamp_us",
                           static_cast<std::int64_t>(global_min));
        b.append_comma();
        b.append_key_value("max_timestamp_us",
                           static_cast<std::int64_t>(global_max));
        b.end_object();
    }

    b.append_comma();
    b.escape_and_append_with_quotes("files");
    b.append_colon();
    b.start_array();
    bool first = true;
    for (const auto& f : index.files()) {
        if (!first) b.append_comma();
        first = false;
        b.start_object();
        b.append_key_value("path", f.path);
        b.append_comma();
        b.append_key_value("has_bloom_data", f.has_bloom_data);
        b.append_comma();
        b.append_key_value("has_checkpoint_index", f.has_checkpoint_index);
        if (f.min_timestamp_us > 0 || f.max_timestamp_us > 0) {
            b.append_comma();
            b.append_key_value("min_timestamp_us",
                               static_cast<std::int64_t>(f.min_timestamp_us));
            b.append_comma();
            b.append_key_value("max_timestamp_us",
                               static_cast<std::int64_t>(f.max_timestamp_us));
        }
        b.end_object();
    }
    b.end_array();
    b.end_object();
    co_return HttpResponse::ok(std::string(b));
}

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);
        },
        RouteDoc{
            "List the indexed trace files.",
            "Trace data",
            {},
            R"({"files":[{"path":"trace-0.pfw.gz","has_bloom_data":true}],)"
            R"("count":1})"});

    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);
        },
        RouteDoc{"Metadata for one trace file.",
                 "Trace data",
                 {{"file", "Trace file path", true, ""}},
                 R"({"path":"trace-0.pfw.gz","has_bloom_data":true})"});

    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);
        },
        RouteDoc{"Query raw events as NDJSON (filtered, limited).",
                 "Trace data",
                 {{"file", "Trace file path", false, ""},
                  {"name", "Filter by operation name", false, "read"},
                  {"dur_min", "Minimum duration (us)", false, ""},
                  {"limit", "Max events (0 = all)", false, "20"}},
                 R"({"id":1,"name":"read","cat":"POSIX","pid":100,"tid":100,)"
                 R"("ts":1000,"dur":150,"args":{"ret":4096}})"});

    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);
        },
        RouteDoc{"Stream all matching events as NDJSON (no limit).",
                 "Trace data",
                 {{"name", "Filter by operation name", false, ""}},
                 ""});

    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);
        },
        RouteDoc{"Aggregate statistics over the index.", "Trace data", {}, ""});

    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);
        },
        RouteDoc{"Global summary: file count and time bounds.",
                 "Trace data",
                 {},
                 R"({"file_count":2,"global_min_timestamp_us":1000000,)"
                 R"("global_max_timestamp_us":6999732})"});
}

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

llnl / dftracer-utils / 29187058583

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