• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

llnl / dftracer-utils / 30074477855

24 Jul 2026 07:08AM UTC coverage: 52.727% (+0.07%) from 52.66%
30074477855

Pull #99

github

web-flow
Merge bfddb8de9 into 06bc84ec9
Pull Request #99: Support CM time_metric (NS/MS/SEC/US) across reader and viz

42091 of 103330 branches covered (40.73%)

Branch coverage included in aggregate %.

2472 of 3143 new or added lines in 65 files covered. (78.65%)

146 existing lines in 14 files now uncovered.

37235 of 47117 relevant lines covered (79.03%)

68905.24 hits per line

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

49.59
/src/dftracer/utils/server/trace_api.cpp
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",
286!
45
                                                                    "SH"};
286!
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*/,
126!
51
                                                 const QueryParams& /*params*/,
52
                                                 TraceIndex& index) {
21!
53
    auto& b = scratch_json_builder();
21!
54
    b.start_object();
21✔
55
    b.escape_and_append_with_quotes("files");
21✔
56
    b.append_colon();
21✔
57
    b.start_array();
21✔
58
    bool first = true;
21✔
59
    for (const auto& f : index.files()) {
44!
60
        if (!first) b.append_comma();
23✔
61
        first = false;
23✔
62
        b.start_object();
23✔
63
        b.append_key_value("path", f.path);
23!
64
        b.append_comma();
23✔
65
        b.append_key_value("has_bloom_data", f.has_bloom_data);
23✔
66
        b.append_comma();
23✔
67
        b.append_key_value("has_checkpoint_index", f.has_checkpoint_index);
23✔
68
        b.end_object();
23✔
69
    }
23✔
70
    b.end_array();
21✔
71
    b.append_comma();
21✔
72
    b.append_key_value("count", static_cast<std::int64_t>(index.file_count()));
21!
73
    b.end_object();
21✔
74
    co_return HttpResponse::ok(std::string(b));
42!
75
}
63!
76

77
// --- GET /api/v1/files/info ---
78
static coro::CoroTask<HttpResponse> handle_file_info(const HttpRequest& /*req*/,
16!
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!
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
}
6!
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();
×
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] + "\"";
×
138
    }
139
    s += "]";
×
140
    return s;
×
141
}
×
142

143
static std::optional<Query> build_query_from_params(const QueryParams& params) {
12✔
144
    std::string dsl;
12✔
145

146
    auto cat = params.get("cat");
12!
147
    if (!cat.empty()) {
12!
148
        auto vals = split_csv(cat);
×
149
        if (!vals.empty()) dsl += format_in_clause("cat", vals);
×
150
    }
×
151

152
    auto name = params.get("name");
12!
153
    if (!name.empty()) {
12!
154
        auto vals = split_csv(name);
×
155
        if (!vals.empty()) {
×
156
            if (!dsl.empty()) dsl += " and ";
×
157
            dsl += format_in_clause("name", vals);
×
158
        }
159
    }
×
160

161
    auto pid = params.get("pid");
12!
162
    if (!pid.empty()) {
12!
163
        if (!dsl.empty()) dsl += " and ";
×
164
        dsl += "pid == " + std::string(pid);
×
165
    }
166

167
    double ts_min = params.get_double("ts_min", 0);
12!
168
    double ts_max = params.get_double("ts_max", 0);
12!
169
    if (ts_min > 0) {
12!
170
        if (!dsl.empty()) dsl += " and ";
×
171
        dsl += "ts >= " + std::to_string(static_cast<uint64_t>(ts_min));
×
172
    }
173
    if (ts_max > 0) {
12!
174
        if (!dsl.empty()) dsl += " and ";
×
175
        dsl += "ts <= " + std::to_string(static_cast<uint64_t>(ts_max));
×
176
    }
177

178
    double dur_min = params.get_double("dur_min", 0);
12!
179
    double dur_max = params.get_double("dur_max", 0);
12!
180
    if (dur_min > 0) {
12!
181
        if (!dsl.empty()) dsl += " and ";
×
182
        dsl += "dur >= " + std::to_string(static_cast<uint64_t>(dur_min));
×
183
    }
184
    if (dur_max > 0) {
12!
185
        if (!dsl.empty()) dsl += " and ";
×
186
        dsl += "dur <= " + std::to_string(static_cast<uint64_t>(dur_max));
×
187
    }
188

189
    if (dsl.empty()) return std::nullopt;
12✔
190
    auto result = Query::from_string(dsl);
×
191
    if (!result) return std::nullopt;
×
192
    return std::move(*result);
×
193
}
12✔
194

195
static ViewDefinition build_view_from_params(const QueryParams& params) {
6✔
196
    ViewDefinition view;
6✔
197
    view.name = "api_query";
6!
198
    view.description = "HTTP API query";
6!
199

200
    auto q = build_query_from_params(params);
6!
201
    if (q) view.with_query(std::move(*q));
6!
202
    return view;
9✔
203
}
6!
204

205
// ============================================================================
206
// Shared helpers for event streaming endpoints
207
// ============================================================================
208

209
static std::vector<const TraceIndex::FileInfo*> resolve_target_files(
6✔
210
    TraceIndex& index, const QueryParams& params, double ts_min = 0,
211
    double ts_max = 0) {
212
    auto files = collect_candidate_files(index, params);
6✔
213

214
    if (ts_min > 0 || ts_max > 0) {
6!
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;
6✔
231
}
3!
232

233
using StreamChunk = HttpResponse::StreamChunk;
234

235
static coro::AsyncGenerator<StreamChunk> stream_events(
30!
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, std::size_t max_concurrent,
239
    CancelToken cancel) {
3!
240
    int emitted = 0;
3✔
241
    const std::size_t slots = std::max<std::size_t>(1, max_concurrent);
3!
242

243
    // Prune a batch of files in parallel (bloom/index I/O dominates and is
244
    // serial otherwise), then stream matches in file order so the `limit`
245
    // early-exit still stops after the first satisfying files.
246
    for (std::size_t base = 0; base < files.size(); base += slots) {
12✔
247
        if (cancel.cancelled()) co_return;
6!
248
        if (limit > 0 && emitted >= limit) break;
3!
249

250
        const std::size_t batch_end = std::min(base + slots, files.size());
3!
251
        const std::size_t batch_n = batch_end - base;
3✔
252
        std::vector<ViewBuilderOutput> outs(batch_n);
3!
253

254
        {
255
            CoroScope scope(Executor::current());
3!
256
            for (std::size_t i = 0; i < batch_n; ++i) {
6✔
257
                auto* fi = files[base + i];
3✔
258
                if (fi->uncompressed_size == 0 && fi->num_checkpoints == 0)
3!
259
                    continue;
260
                auto* slot = &outs[i];
3✔
261
                scope.spawn([fi, slot, &ev_view, bloom_cache, ts_min,
24!
262
                             ts_max](CoroScope&) -> coro::CoroTask<void> {
6!
263
                    ViewBuilderInput builder_input;
3✔
264
                    builder_input.with_view(ev_view)
6!
265
                        .with_file_path(fi->path)
3!
266
                        .with_index_path(fi->has_bloom_data ? fi->index_path
3!
267
                                                            : "")
×
268
                        .with_uncompressed_size(fi->uncompressed_size)
3!
269
                        .with_num_checkpoints(fi->num_checkpoints)
3!
270
                        .with_bloom_cache(bloom_cache)
3!
271
                        .with_time_range(ts_min, ts_max);
3!
272
                    ViewBuilderUtility builder;
3!
273
                    auto r = co_await builder.process(builder_input);
9!
274
                    if (r) *slot = std::move(*r);
3!
275
                });
15!
276
            }
3!
277
            co_await scope.join();
6!
278
        }
3!
279

280
        for (std::size_t i = 0; i < batch_n; ++i) {
12✔
281
            if (cancel.cancelled()) co_return;
3!
282
            if (limit > 0 && emitted >= limit) break;
3!
283
            if (!outs[i].file_may_match) continue;
3!
284
            auto* file_info = files[base + i];
3✔
285

286
            for (const auto& candidate : outs[i].candidates) {
12✔
287
                if (limit > 0 && emitted >= limit) break;
3!
288

289
                ViewReaderInput reader_input;
3✔
290
                reader_input.with_file_path(file_info->path)
3!
291
                    .with_index_path(file_info->index_path)
3!
292
                    .with_byte_range(candidate.start_byte, candidate.end_byte)
3!
293
                    .with_checkpoint_idx(candidate.checkpoint_idx)
3!
294
                    .with_view(ev_view);
3!
295

296
                ViewReaderUtility reader;
3!
297
                auto event_gen = reader.process(reader_input);
3!
298
                while (auto batch = co_await event_gen.next()) {
24!
299
                    if (cancel.cancelled()) co_return;
3!
300
                    int count = std::min(
6!
301
                        static_cast<int>(batch->events.size()),
3✔
302
                        limit > 0 ? limit - emitted
3✔
303
                                  : static_cast<int>(batch->events.size()));
2✔
304
                    if (count > 0) {
3✔
305
                        co_yield StreamChunk{std::span<const std::string_view>(
21!
306
                            batch->events.data(),
9✔
307
                            static_cast<std::size_t>(count))};
9✔
308
                        emitted += count;
3✔
309
                    }
3✔
310
                }
12✔
311
            }
9✔
312
        }
9✔
313
    }
9✔
314
}
51!
315

316
// ============================================================================
317
// Event endpoints
318
// ============================================================================
319

320
// --- GET /api/v1/events ---
321
static coro::CoroTask<HttpResponse> handle_events(const HttpRequest& req,
6!
322
                                                  const QueryParams& params,
323
                                                  TraceIndex& index) {
1!
324
    int limit = params.get_int("limit", 1000);
1!
325
    if (limit <= 0) limit = 1000;
1!
326
    if (limit > 100000) limit = 100000;
1!
327

328
    double ts_min = params.get_double("ts_min", 0);
1!
329
    double ts_max = params.get_double("ts_max", 0);
1!
330
    auto files = resolve_target_files(index, params, ts_min, ts_max);
1!
331
    auto view = build_view_from_params(params);
1!
332
    auto query = build_query_from_params(params);
1!
333

334
    auto gen = std::make_unique<HttpResponse::StreamGenerator>(stream_events(
2!
335
        std::move(files), std::move(view), std::move(query), ts_min, ts_max,
1✔
336
        &index.bloom_cache(), limit, index.max_concurrent(), req.cancel_token));
1!
337

338
    auto resp = HttpResponse::streaming(std::move(gen));
1!
339
    resp.headers.push_back({"X-Limit", std::to_string(limit)});
1!
340
    co_return resp;
2✔
341
}
3!
342

343
// --- GET /api/v1/events/stream ---
344
static coro::CoroTask<HttpResponse> handle_events_stream(
12!
345
    const HttpRequest& req, const QueryParams& params, TraceIndex& index) {
2!
346
    double ts_min = params.get_double("ts_min", 0);
2!
347
    double ts_max = params.get_double("ts_max", 0);
2!
348
    auto files = resolve_target_files(index, params, ts_min, ts_max);
2!
349
    auto view = build_view_from_params(params);
2!
350
    auto query = build_query_from_params(params);
2!
351
    int limit = params.get_int("limit", 0);
2!
352

353
    auto gen = std::make_unique<HttpResponse::StreamGenerator>(stream_events(
4!
354
        std::move(files), std::move(view), std::move(query), ts_min, ts_max,
2✔
355
        &index.bloom_cache(), limit, index.max_concurrent(), req.cancel_token));
2✔
356

357
    co_return HttpResponse::streaming(std::move(gen));
4!
358
}
6!
359

360
// --- GET /api/v1/stats ---
361
static coro::CoroTask<HttpResponse> handle_stats(const HttpRequest& req,
6!
362
                                                 const QueryParams& /*params*/,
363
                                                 TraceIndex& index) {
1!
364
    static std::mutex cache_mutex;
1!
365
    static std::unordered_map<std::string, std::string,
1!
366
                              dftracer::utils::TransparentStringHash,
367
                              dftracer::utils::TransparentStringEqual>
368
        stats_cache;
1✔
369

370
    {
371
        std::lock_guard<std::mutex> lock(cache_mutex);
1!
372
        auto it = stats_cache.find(req.path);
1!
373
        if (it != stats_cache.end()) {
1!
374
            co_return HttpResponse::ok(it->second);
1!
375
        }
376
    }
1!
377

378
    std::vector<TraceStatistics> all_stats;
1✔
379

380
    // Group files by index_path
381
    std::unordered_map<std::string,
1✔
382
                       std::vector<std::pair<std::size_t, std::string>>>
383
        files_by_index;
1✔
384
    std::size_t file_idx = 0;
1✔
385
    for (const auto& file_info : index.files()) {
2✔
386
        if (!file_info.has_bloom_data) continue;
1!
387
        files_by_index[file_info.index_path].emplace_back(file_idx++,
1!
388
                                                          file_info.path);
1✔
389
    }
1!
390

391
    // Resolve each group and read statistics
392
    for (auto& [idx_path, files] : files_by_index) {
4!
393
        if (req.cancel_token.cancelled()) co_return HttpResponse::ok("{}");
3!
394
        std::vector<std::string> file_paths;
3✔
395
        file_paths.reserve(files.size());
3!
396
        for (const auto& [_, path] : files) {
4✔
397
            file_paths.push_back(path);
1!
398
        }
1✔
399

400
        IndexResolverUtility resolver;
3!
401
        ResolverInput input;
3✔
402
        input.files = std::move(file_paths);
3✔
403
        input.require_checkpoints = false;
3✔
404

405
        auto result = co_await resolver.process(input);
4!
406

407
        if (result.cached.empty()) {
3!
408
            continue;
409
        }
410

411
        try {
412
            SharedIndexStatisticsReader reader;
3✔
413
            auto batch_rows = co_await reader.query(
4!
414
                result.index_path, std::move(result.cached),
3!
415
                StatisticsQueryType::SUMMARY);
416
            auto callback = [&all_stats](std::size_t /*file_index*/,
3✔
417
                                         TraceStatistics&& stats) {
1✔
418
                if (stats.success) {
2✔
419
                    all_stats.push_back(std::move(stats));
2✔
420
                }
1✔
421
            };
2✔
422
            SharedIndexStatisticsReader::process_batch_results(batch_rows,
1!
423
                                                               callback);
424
        } catch (const std::exception& e) {
1!
425
            DFTRACER_UTILS_LOG_WARN("Server stats batch read failed for %s: %s",
×
426
                                    idx_path.c_str(), e.what());
427
        }
×
428
    }
3!
429

430
    std::uint64_t total_events = 0;
1✔
431
    std::size_t file_count = all_stats.size();
1✔
432
    for (const auto& s : all_stats) {
2✔
433
        total_events += s.total_events();
1!
434
    }
1✔
435

436
    auto& sb = scratch_json_builder();
1!
437
    sb.start_object();
1✔
438
    sb.append_key_value("file_count", static_cast<std::int64_t>(file_count));
1✔
439
    sb.append_comma();
1✔
440
    sb.append_key_value("total_events",
2✔
441
                        static_cast<std::int64_t>(total_events));
1✔
442
    sb.append_comma();
1✔
443
    sb.escape_and_append_with_quotes("files");
1✔
444
    sb.append_colon();
1✔
445
    sb.start_array();
1✔
446
    for (std::size_t i = 0; i < all_stats.size(); ++i) {
2✔
447
        if (i > 0) sb.append_comma();
1!
448
        sb.append_raw(all_stats[i].to_json());  // already JSON
1!
449
    }
1✔
450
    sb.end_array();
1✔
451
    sb.end_object();
1✔
452
    std::string body(sb);
1!
453

454
    {
455
        std::lock_guard<std::mutex> lock(cache_mutex);
1!
456
        stats_cache.emplace(std::string(req.path), body);
1!
457
    }
1✔
458
    co_return HttpResponse::ok(body);
1!
459
}
11!
460

461
// --- GET /api/v1/info ---
462
static coro::CoroTask<HttpResponse> handle_info(const HttpRequest& /*req*/,
12!
463
                                                const QueryParams& /*params*/,
464
                                                TraceIndex& index) {
2!
465
    auto global_min_native = index.global_min_timestamp_us();
2!
466
    auto global_max_native = index.global_max_timestamp_us();
2!
467
    bool has_time_range =
4✔
468
        global_max_native > 0 &&
2!
469
        global_min_native != std::numeric_limits<std::uint64_t>::max();
2✔
470
    // Client works in microseconds; scale native (index-unit) bounds to us.
471
    auto global_min = index.native_to_us(global_min_native);
2!
472
    auto global_max = index.native_to_us(global_max_native);
2!
473

474
    auto& b = scratch_json_builder();
2!
475
    b.start_object();
2✔
476
    b.append_key_value("file_count",
4✔
477
                       static_cast<std::int64_t>(index.file_count()));
2✔
478

479
    if (has_time_range) {
2!
480
        b.append_comma();
2✔
481
        b.escape_and_append_with_quotes("time_range");
2✔
482
        b.append_colon();
2✔
483
        b.start_object();
2✔
484
        b.append_key_value("min_timestamp_us",
4✔
485
                           static_cast<std::int64_t>(global_min));
2✔
486
        b.append_comma();
2✔
487
        b.append_key_value("max_timestamp_us",
4✔
488
                           static_cast<std::int64_t>(global_max));
2✔
489
        b.end_object();
2✔
490
    }
2✔
491

492
    b.append_comma();
2✔
493
    b.escape_and_append_with_quotes("files");
2✔
494
    b.append_colon();
2✔
495
    b.start_array();
2✔
496
    bool first = true;
2✔
497
    for (const auto& f : index.files()) {
4✔
498
        if (!first) b.append_comma();
2!
499
        first = false;
2✔
500
        b.start_object();
2✔
501
        b.append_key_value("path", f.path);
2!
502
        b.append_comma();
2✔
503
        b.append_key_value("has_bloom_data", f.has_bloom_data);
2✔
504
        b.append_comma();
2✔
505
        b.append_key_value("has_checkpoint_index", f.has_checkpoint_index);
2✔
506
        if (f.min_timestamp_us > 0 || f.max_timestamp_us > 0) {
2!
507
            b.append_comma();
2✔
508
            b.append_key_value("min_timestamp_us",
2✔
509
                               static_cast<std::int64_t>(
510
                                   index.native_to_us(f.min_timestamp_us)));
2!
511
            b.append_comma();
2✔
512
            b.append_key_value("max_timestamp_us",
2✔
513
                               static_cast<std::int64_t>(
514
                                   index.native_to_us(f.max_timestamp_us)));
2!
515
        }
2✔
516
        b.end_object();
2✔
517
    }
2✔
518
    b.end_array();
2✔
519
    b.end_object();
2✔
520
    co_return HttpResponse::ok(std::string(b));
4!
521
}
6!
522

523
void register_trace_api(Router& router, TraceIndex& index) {
22✔
524
    auto* index_ptr = &index;
22✔
525

526
    router.get(
33!
527
        "/api/v1/files",
11✔
528
        [index_ptr](const HttpRequest& req,
179!
529
                    const QueryParams& params) -> coro::CoroTask<HttpResponse> {
21!
530
            co_return co_await handle_files(req, params, *index_ptr);
105!
531
        },
84!
532
        RouteDoc{
33!
533
            "List the indexed trace files.",
11!
534
            "Trace data",
11!
535
            {},
11✔
536
            R"({"files":[{"path":"trace-0.pfw.gz","has_bloom_data":true}],)"
11!
537
            R"("count":1})"});
538

539
    router.get(
33!
540
        "/api/v1/files/info",
11✔
541
        [index_ptr](const HttpRequest& req,
27!
542
                    const QueryParams& params) -> coro::CoroTask<HttpResponse> {
2!
543
            co_return co_await handle_file_info(req, params, *index_ptr);
10!
544
        },
8!
545
        RouteDoc{"Metadata for one trace file.",
66!
546
                 "Trace data",
11!
547
                 {{"file", "Trace file path", true, ""}},
11!
548
                 R"({"path":"trace-0.pfw.gz","has_bloom_data":true})"});
33!
549

550
    router.get(
33!
551
        "/api/v1/events",
11✔
552
        [index_ptr](const HttpRequest& req,
19!
553
                    const QueryParams& params) -> coro::CoroTask<HttpResponse> {
1!
554
            co_return co_await handle_events(req, params, *index_ptr);
5!
555
        },
4!
556
        RouteDoc{"Query raw events as NDJSON (filtered, limited).",
99!
557
                 "Trace data",
11!
558
                 {{"file", "Trace file path", false, ""},
44!
559
                  {"name", "Filter by operation name", false, "read"},
11!
560
                  {"dur_min", "Minimum duration (us)", false, ""},
11!
561
                  {"limit", "Max events (0 = all)", false, "20"}},
11!
562
                 R"({"id":1,"name":"read","cat":"POSIX","pid":100,"tid":100,)"
11!
563
                 R"("ts":1000,"dur":150,"args":{"ret":4096}})"});
55✔
564

565
    router.get(
33!
566
        "/api/v1/events/stream",
11✔
567
        [index_ptr](const HttpRequest& req,
27!
568
                    const QueryParams& params) -> coro::CoroTask<HttpResponse> {
2!
569
            co_return co_await handle_events_stream(req, params, *index_ptr);
10!
570
        },
8!
571
        RouteDoc{"Stream all matching events as NDJSON (no limit).",
66!
572
                 "Trace data",
11!
573
                 {{"name", "Filter by operation name", false, ""}},
11!
574
                 ""});
33!
575

576
    router.get(
33!
577
        "/api/v1/stats",
11✔
578
        [index_ptr](const HttpRequest& req,
19!
579
                    const QueryParams& params) -> coro::CoroTask<HttpResponse> {
1!
580
            co_return co_await handle_stats(req, params, *index_ptr);
5!
581
        },
4!
582
        RouteDoc{"Aggregate statistics over the index.", "Trace data", {}, ""});
33!
583

584
    router.get(
33!
585
        "/api/v1/info",
11✔
586
        [index_ptr](const HttpRequest& req,
27!
587
                    const QueryParams& params) -> coro::CoroTask<HttpResponse> {
2!
588
            co_return co_await handle_info(req, params, *index_ptr);
10!
589
        },
8!
590
        RouteDoc{"Global summary: file count and time bounds.",
33!
591
                 "Trace data",
11!
592
                 {},
11✔
593
                 R"({"file_count":2,"global_min_timestamp_us":1000000,)"
11!
594
                 R"("global_max_timestamp_us":6999732})"});
595

596
    router.get(
33!
597
        "/api/v1/resolve",
11✔
598
        [index_ptr](const HttpRequest& /*req*/,
11!
599
                    const QueryParams& params) -> coro::CoroTask<HttpResponse> {
×
600
            std::string hashes(params.get("hash"));
×
601
            if (hashes.empty())
×
602
                co_return HttpResponse::bad_request("Missing parameter: hash");
×
603
            auto kind = params.get("type");
×
604
            using HashType = TraceIndex::HashType;
605
            HashType type = HashType::FILE;
606
            if (kind == "host")
×
607
                type = HashType::HOST;
608
            else if (kind == "string")
×
609
                type = HashType::STRING;
610
            else if (kind == "proc")
×
611
                type = HashType::PROC;
612
            else if (!kind.empty() && kind != "file")
×
613
                co_return HttpResponse::bad_request("Invalid type: " +
×
614
                                                    std::string(kind));
×
615

616
            auto& b = scratch_json_builder();
×
617
            b.start_object();
618
            b.escape_and_append_with_quotes("names");
619
            b.append_colon();
620
            b.start_object();
621
            bool first = true;
622
            std::size_t start = 0;
623
            // Comma-separated so one click can resolve its file and host at
624
            // once; unknown hashes are simply absent from the reply.
625
            while (start <= hashes.size()) {
×
626
                auto end = hashes.find(',', start);
627
                if (end == std::string::npos) end = hashes.size();
×
628
                std::string one = hashes.substr(start, end - start);
×
629
                start = end + 1;
630
                if (one.empty()) continue;
×
631
                auto name = index_ptr->resolve_hash(type, one);
×
632
                if (name.empty()) continue;
×
633
                if (!first) b.append_comma();
×
634
                first = false;
635
                b.escape_and_append_with_quotes(one);
636
                b.append_colon();
637
                b.escape_and_append_with_quotes(name);
638
            }
×
639
            b.end_object();
640
            b.end_object();
641
            co_return HttpResponse::ok(std::string(b));
×
NEW
642
        },
×
643
        RouteDoc{
77!
644
            "Resolve content hashes (file/host/string) to their names.",
11!
645
            "Control",
11!
646
            {{"hash", "Hash, or several separated by commas", true, ""},
22!
647
             {"type", "file (default), host, string or proc", false, "file"}},
11!
648
            R"({"names":{"314c1a1cdb22a136":"/data/train/img_0.npz"}})"});
44!
649

650
    router.post(
33!
651
        "/api/v1/cancel",
11✔
652
        [](const HttpRequest& /*req*/,
17!
653
           const QueryParams& params) -> coro::CoroTask<HttpResponse> {
1!
654
            std::string id(params.get("id"));
1!
655
            bool found = CancelRegistry::instance().cancel(id);
1!
656
            co_return HttpResponse::ok(std::string("{\"cancelled\":") +
3!
657
                                       (found ? "true" : "false") + "}");
1!
658
        },
3!
659
        RouteDoc{"Cancel an in-flight request by its X-Request-Id.",
66!
660
                 "Control",
11!
661
                 {{"id", "Request id to cancel", true, ""}},
11!
662
                 R"({"cancelled":true})"});
33!
663
}
22✔
664

665
}  // namespace dftracer::utils::server
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE TRIAL · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc