40940235178

Committed 29 Oct 2026 07:27PM UTC coverage: 27.181%. First build

Build # 40940235178

Build Type

Pull #402

github

Committed by

web-flow

Commit Message

Merge e0fc27880 into de73d284b

Pull Request Pull Request #402: feat(service): add producer/consumer tracing through dftracer_service

Coverage Stats

16914 of 83793 branches covered (20.19%)

Branch coverage included in aggregate %.

2684 of 2985 new or added lines in 40 files covered. (89.92%)

11495 of 20726 relevant lines covered (55.46%)

88527.54 hits per line

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

78.55

/test/unit/test_packed.cpp

// The service's packed path (packed_json, PackedAggregator) writes the lines
// the Metadata path (JsonLines, Aggregator) writes for the same events. Args
// are compared as sets: JSON does not order them, and JsonLines follows its
// unordered_map.
#include <dftracer/core/aggregator/aggregator.h>
#include <dftracer/core/aggregator/packed_aggregator.h>
#include <dftracer/core/common/singleton.h>
#include <dftracer/core/ipc/event_codec.h>
#include <dftracer/core/serialization/json_line.h>
#include <dftracer/core/serialization/packed_json.h>
#include <dftracer/core/utils/configuration_manager.h>

#include <algorithm>
#include <any>
#include <cstdlib>
#include <iostream>
#include <string>
#include <tuple>
#include <vector>

#include "check.h"

using namespace dftracer;

static char kHostname[] = "hosthash";

struct EventField {
  std::string key;
  std::any value;
  MetadataType type;
};

struct Event {
  const char* name;
  const char* category;
  TimeResolution ts, dur;
  ThreadID tid;
  bool has_metadata;
  std::vector<EventField> fields;
};

static Metadata* metadata_of(const Event& e) {
  if (!e.has_metadata) return nullptr;
  auto* md = new Metadata();
  for (const auto& f : e.fields) md->insert_or_assign(f.key, f.value, f.type);
  return md;
}

// The event as the service decodes it from the ring.
static void packed_of(const Event& e, bool include, std::vector<uint8_t>& bytes,
                      ipc::DecodedEvent& out) {
  Metadata* md = metadata_of(e);
  ipc::ByteWriter w(bytes);
  ipc::encode_event(w, 7, e.name, e.category,
                    TraceEventType::TRACE_TYPE_LIBC_IO, e.ts, e.dur, md, 42,
                    e.tid, include ? SHM_EVENT_INC_METADATA : 0);
  delete md;
  DFT_CHECK(ipc::decode_event(bytes.data(), bytes.size(), out));
}

// Splits a line's args at its top-level commas and sorts them.
static std::string normalized(std::string line) {
  while (!line.empty() && (line.back() == '\n' || line.back() == '\0'))
    line.pop_back();
  const std::string marker = ",\"args\":{";
  const size_t at = line.find(marker);
  if (at == std::string::npos) return line;
  const std::string body =
      line.substr(at + marker.size(), line.size() - at - marker.size() - 2);
  std::vector<std::string> parts;
  std::string part;
  int depth = 0;
  bool quoted = false;
  for (char c : body) {
    if (c == '"') quoted = !quoted;
    if (!quoted && (c == '{' || c == '[')) ++depth;
    if (!quoted && (c == '}' || c == ']')) --depth;
    if (!quoted && depth == 0 && c == ',') {
      parts.push_back(part);
      part.clear();
      continue;
    }
    part += c;
  }
  parts.push_back(part);
  std::sort(parts.begin(), parts.end());
  std::string out = line.substr(0, at) + " |";
  for (const auto& p : parts) out += " " + p;
  return out;
}

static std::vector<std::string> lines_of(const char* buffer, size_t size) {
  std::vector<std::string> lines;
  std::string all(buffer, size), line;
  for (char c : all) {
    if (c == '\n') {
      lines.push_back(normalized(line));
      line.clear();
    } else {
      line += c;
    }
  }
  std::sort(lines.begin(), lines.end());
  return lines;
}

static const std::vector<Event> kEvents = {
    {"open", "POSIX", 100, 5, 1, false, {}},
    {"read", "POSIX", 110, 6, 1, true, {}},
    {"read",
     "POSIX",
     120,
     7,
     1,
     true,
     {{"fhash", std::string("aaa"), MetadataType::MT_KEY},
      {"fd", static_cast<int64_t>(3), MetadataType::MT_KEY},
      {"size", static_cast<int64_t>(-4096), MetadataType::MT_VALUE},
      {"offset", static_cast<uint64_t>(1) << 40, MetadataType::MT_VALUE},
      {"bw", 1.5e9, MetadataType::MT_VALUE},
      {"mode", std::string("r"), MetadataType::MT_VALUE},
      {"note", std::string("x"), MetadataType::MT_IGNORE}}},
    {"step",
     "app",
     130,
     8,
     2,
     true,
     {{"relations", RawJson("[{\"rel\":\"used\",\"id\":\"e1\"}]"),
       MetadataType::MT_KEY},
      {"epoch", static_cast<int64_t>(2), MetadataType::MT_VALUE}}},
};

static void check_data() {
  auto js = std::make_shared<JsonLines>();
  js->initialize(nullptr, kHostname);
  static char expect[DFTRACER_SERIALIZATION_EVENT_MAX];
  static char got[DFTRACER_SERIALIZATION_EVENT_MAX];
  for (bool include : {true, false}) {
    js->set_include_metadata(include);
    for (const Event& e : kEvents) {
      std::vector<uint8_t> bytes;
      ipc::DecodedEvent decoded;
      packed_of(e, include, bytes, decoded);
      const size_t n = js->data(expect, 7, e.name, e.category,
                                TraceEventType::TRACE_TYPE_LIBC_IO, e.ts, e.dur,
                                metadata_of(e), 42, e.tid);
      const size_t m = packed_json::data(got, decoded, kHostname);
      DFT_CHECK(normalized(std::string(expect, n)) ==
                normalized(std::string(got, m)));
      const size_t c = js->counter(expect, 7, e.name, e.category,
                                   TraceEventType::TRACE_TYPE_LIBC_IO, e.ts, 42,
                                   e.tid, metadata_of(e));
      const size_t d = packed_json::counter(got, decoded, kHostname);
      DFT_CHECK(normalized(std::string(expect, c)) ==
                normalized(std::string(got, d)));
    }
  }
  std::cout << "✓ packed data and counter lines match JsonLines" << std::endl;
}

static void check_aggregation() {
  auto config = Singleton<ConfigurationManager>::get_instance();
  // Same bucket (keys and thread), another key value, another thread, a
  // later interval, and events with no keys: the second interval starts more
  // than trace_interval_ms later in any time unit.
  const TimeResolution later = 1000ULL * 1000 * 1000 * 10;
  std::vector<Event> events;
  for (int64_t i = 0; i < 4; ++i) {
    for (const char* fhash : {"aaa", "bbb"}) {
      for (ThreadID tid : {1, 2}) {
        for (TimeResolution base : {TimeResolution(0), later}) {
          events.push_back(
              {"read",
               "POSIX",
               base + 100 + i,
               10 + i,
               tid,
               true,
               {{"fhash", std::string(fhash), MetadataType::MT_KEY},
                {"fd", static_cast<int64_t>(3), MetadataType::MT_KEY},
                {"size", static_cast<int64_t>(100 * (i + 1)),
                 MetadataType::MT_VALUE},
                {"offset", static_cast<uint64_t>(i) << 33,
                 MetadataType::MT_VALUE},
                {"bw", 0.25 * (i + 1), MetadataType::MT_VALUE},
                {"mode", std::string("r"), MetadataType::MT_VALUE},
                {"note", std::string("x"), MetadataType::MT_IGNORE}}});
        }
      }
    }
    events.push_back({"close", "POSIX", 200 + i, 1, 1, false, {}});
  }
  // One event alone in its bucket: its values are written as `key`, not
  // key_sum / key_min / key_max.
  events.push_back({"write",
                    "POSIX",
                    300,
                    9,
                    1,
                    true,
                    {{"size", static_cast<int64_t>(7), MetadataType::MT_VALUE},
                     {"bw", 2.0, MetadataType::MT_VALUE}}});

  Aggregator aggregator;
  PackedAggregator packed(config);
  int rank = -1;
  for (const Event& e : events) {
    Metadata* md = metadata_of(e);
    AggregatedKey key(e.category, e.name, TraceEventType::TRACE_TYPE_LIBC_IO,
                      e.ts, e.dur, e.tid, md, nullptr, &rank);
    aggregator.aggregate(key);
    delete md;
    std::vector<uint8_t> bytes;
    ipc::DecodedEvent decoded;
    packed_of(e, true, bytes, decoded);
    packed.aggregate(decoded);
  }

  AggregatedDataType data;
  aggregator.get_previous_aggregations(data, true);
  auto js = std::make_shared<JsonLines>();
  js->initialize(nullptr, kHostname);
  std::vector<char> expect(1 << 20);
  const size_t n = js->aggregated(expect.data(), 0, 42, data);

  struct Rows : PackedAggregator::RowSink {
    std::vector<char> buffer = std::vector<char>(1 << 20);
    size_t used = 0;
    char* row() override { return buffer.data() + used; }
    void wrote(size_t size) override { used += size; }
  } rows;
  packed.drain(true, 42, kHostname, rows);

  const auto want = lines_of(expect.data(), n);
  const auto got = lines_of(rows.buffer.data(), rows.used);
  if (want != got) {
    for (const auto& l : want) std::cerr << "want " << l << "\n";
    for (const auto& l : got) std::cerr << "got  " << l << "\n";
  }
  DFT_CHECK(want == got);
  // 2 fhash x 2 threads x 2 intervals of read, close, write.
  DFT_CHECK(got.size() == 10);
  std::cout << "✓ PackedAggregator rows match Aggregator (" << got.size()
            << " rows)" << std::endl;
}

int main() {
  setenv("DFTRACER_ENABLE", "1", 1);
  setenv("DFTRACER_ENABLE_AGGREGATION", "1", 1);
  setenv("DFTRACER_AGGREGATION_TYPE", "FULL", 1);
  setenv("DFTRACER_TRACE_INTERVAL_MS", "1000", 1);
  check_data();
  check_aggregation();
  return 0;
}

1	// The service's packed path (packed_json, PackedAggregator) writes the lines
2	// the Metadata path (JsonLines, Aggregator) writes for the same events. Args
3	// are compared as sets: JSON does not order them, and JsonLines follows its
4	// unordered_map.
5	#include <dftracer/core/aggregator/aggregator.h>
6	#include <dftracer/core/aggregator/packed_aggregator.h>
7	#include <dftracer/core/common/singleton.h>
8	#include <dftracer/core/ipc/event_codec.h>
9	#include <dftracer/core/serialization/json_line.h>
10	#include <dftracer/core/serialization/packed_json.h>
11	#include <dftracer/core/utils/configuration_manager.h>
12
13	#include <algorithm>
14	#include <any>
15	#include <cstdlib>
16	#include <iostream>
17	#include <string>
18	#include <tuple>
19	#include <vector>
20
21	#include "check.h"
22
23	using namespace dftracer;
24
25	static char kHostname[] = "hosthash";
26
27	struct EventField {
28	std::string key;
29	std::any value;
30	MetadataType type;
31	};
32
33	struct Event {
34	const char* name;
35	const char* category;
36	TimeResolution ts, dur;
37	ThreadID tid;
38	bool has_metadata;
39	std::vector<EventField> fields;
40	};
41
42	static Metadata* metadata_of(const Event& e) {	98 ✔
43	if (!e.has_metadata) return nullptr;	98 ✔
44	auto* md = new Metadata();	84 ✔
45	for (const auto& f : e.fields) md->insert_or_assign(f.key, f.value, f.type);	590 ✔
46	return md;	84 ✔
47	}
48
49	// The event as the service decodes it from the ring.
50	static void packed_of(const Event& e, bool include, std::vector<uint8_t>& bytes,	45 ✔
51	ipc::DecodedEvent& out) {
52	Metadata* md = metadata_of(e);	45 ✔
53	ipc::ByteWriter w(bytes);	45 ✔
54	ipc::encode_event(w, 7, e.name, e.category,	45 ✔
55	TraceEventType::TRACE_TYPE_LIBC_IO, e.ts, e.dur, md, 42,	45 ✔
56	e.tid, include ? SHM_EVENT_INC_METADATA : 0);	45 ✔
57	delete md;	45 ✔
58	DFT_CHECK(ipc::decode_event(bytes.data(), bytes.size(), out));	45 !
59	}	45 ✔
60
61	// Splits a line's args at its top-level commas and sorts them.
62	static std::string normalized(std::string line) {	52 ✔
63	while (!line.empty() && (line.back() == '\n' \|\| line.back() == '\0'))	84 !
64	line.pop_back();	32 ✔
65	const std::string marker = ",\"args\":{";	52 ✔
66	const size_t at = line.find(marker);	52 ✔
67	if (at == std::string::npos) return line;	52 ✔
68	const std::string body =
69	line.substr(at + marker.size(), line.size() - at - marker.size() - 2);	36 ✔
70	std::vector<std::string> parts;	36 ✔
71	std::string part;	36 ✔
72	int depth = 0;	36 ✔
73	bool quoted = false;	36 ✔
74	for (char c : body) {	5,312 ✔
75	if (c == '"') quoted = !quoted;	5,276 ✔
76	if (!quoted && (c == '{' \|\| c == '[')) ++depth;	5,276 ✔
77	if (!quoted && (c == '}' \|\| c == ']')) --depth;	5,276 ✔
78	if (!quoted && depth == 0 && c == ',') {	5,276 ✔
79	parts.push_back(part);	312 ✔
80	part.clear();	312 ✔
81	continue;	312 ✔
82	}
83	part += c;	4,964 ✔
84	}
85	parts.push_back(part);	36 ✔
86	std::sort(parts.begin(), parts.end());	36 ✔
87	std::string out = line.substr(0, at) + " \|";	36 ✔
88	for (const auto& p : parts) out += " " + p;	384 ✔
89	return out;	36 ✔
90	}	52 ✔
91
92	static std::vector<std::string> lines_of(const char* buffer, size_t size) {	2 ✔
93	std::vector<std::string> lines;	2 ✔
94	std::string all(buffer, size), line;	4 ✔
95	for (char c : all) {	5,750 ✔
96	if (c == '\n') {	5,748 ✔
97	lines.push_back(normalized(line));	20 ✔
98	line.clear();	20 ✔
99	} else {
100	line += c;	5,728 ✔
101	}
102	}
103	std::sort(lines.begin(), lines.end());	2 ✔
104	return lines;	4 ✔
105	}	2 ✔
106
107	static const std::vector<Event> kEvents = {
108	{"open", "POSIX", 100, 5, 1, false, {}},
109	{"read", "POSIX", 110, 6, 1, true, {}},
110	{"read",
111	"POSIX",
112	120,
113	7,
114	1,
115	true,
116	{{"fhash", std::string("aaa"), MetadataType::MT_KEY},
117	{"fd", static_cast<int64_t>(3), MetadataType::MT_KEY},
118	{"size", static_cast<int64_t>(-4096), MetadataType::MT_VALUE},
119	{"offset", static_cast<uint64_t>(1) << 40, MetadataType::MT_VALUE},
120	{"bw", 1.5e9, MetadataType::MT_VALUE},
121	{"mode", std::string("r"), MetadataType::MT_VALUE},
122	{"note", std::string("x"), MetadataType::MT_IGNORE}}},
123	{"step",
124	"app",
125	130,
126	8,
127	2,
128	true,
129	{{"relations", RawJson("[{\"rel\":\"used\",\"id\":\"e1\"}]"),
130	MetadataType::MT_KEY},
131	{"epoch", static_cast<int64_t>(2), MetadataType::MT_VALUE}}},
132	};
133
134	static void check_data() {	1 ✔
135	auto js = std::make_shared<JsonLines>();	1 ✔
136	js->initialize(nullptr, kHostname);	1 ✔
137	static char expect[DFTRACER_SERIALIZATION_EVENT_MAX];
138	static char got[DFTRACER_SERIALIZATION_EVENT_MAX];
139	for (bool include : {true, false}) {	3 ✔
140	js->set_include_metadata(include);	2 ✔
141	for (const Event& e : kEvents) {	10 ✔
142	std::vector<uint8_t> bytes;	8 ✔
143	ipc::DecodedEvent decoded;	8 ✔
144	packed_of(e, include, bytes, decoded);	8 ✔
145	const size_t n = js->data(expect, 7, e.name, e.category,	8 ✔
146	TraceEventType::TRACE_TYPE_LIBC_IO, e.ts, e.dur,	8 ✔
147	metadata_of(e), 42, e.tid);	8 ✔
148	const size_t m = packed_json::data(got, decoded, kHostname);	8 ✔
149	DFT_CHECK(normalized(std::string(expect, n)) ==	8 !
150	normalized(std::string(got, m)));
151	const size_t c = js->counter(expect, 7, e.name, e.category,	8 ✔
152	TraceEventType::TRACE_TYPE_LIBC_IO, e.ts, 42,	8 ✔
153	e.tid, metadata_of(e));	8 ✔
154	const size_t d = packed_json::counter(got, decoded, kHostname);	8 ✔
155	DFT_CHECK(normalized(std::string(expect, c)) ==	8 !
156	normalized(std::string(got, d)));
157	}	8 ✔
158	}
159	std::cout << "✓ packed data and counter lines match JsonLines" << std::endl;	1 ✔
160	}	1 ✔
161
162	static void check_aggregation() {	1 ✔
163	auto config = Singleton<ConfigurationManager>::get_instance();	1 ✔
164	// Same bucket (keys and thread), another key value, another thread, a
165	// later interval, and events with no keys: the second interval starts more
166	// than trace_interval_ms later in any time unit.
167	const TimeResolution later = 1000ULL * 1000 * 1000 * 10;	1 ✔
168	std::vector<Event> events;	1 ✔
169	for (int64_t i = 0; i < 4; ++i) {	5 ✔
170	for (const char* fhash : {"aaa", "bbb"}) {	12 ✔
171	for (ThreadID tid : {1, 2}) {	24 ✔
172	for (TimeResolution base : {TimeResolution(0), later}) {	48 ✔
173	events.push_back(	512 !
174	{"read",
175	"POSIX",
176	base + 100 + i,	32 ✔
177	10 + i,	32 ✔
178	tid,
179	true,
180	{{"fhash", std::string(fhash), MetadataType::MT_KEY},	64 ✔
181	{"fd", static_cast<int64_t>(3), MetadataType::MT_KEY},	64 ✔
182	{"size", static_cast<int64_t>(100 * (i + 1)),	64 ✔
183	MetadataType::MT_VALUE},
184	{"offset", static_cast<uint64_t>(i) << 33,	64 ✔
185	MetadataType::MT_VALUE},
186	{"bw", 0.25 * (i + 1), MetadataType::MT_VALUE},	64 ✔
187	{"mode", std::string("r"), MetadataType::MT_VALUE},	64 ✔
188	{"note", std::string("x"), MetadataType::MT_IGNORE}}});	64 ✔
189	}
190	}
191	}
192	events.push_back({"close", "POSIX", 200 + i, 1, 1, false, {}});	4 ✔
193	}
194	// One event alone in its bucket: its values are written as `key`, not
195	// key_sum / key_min / key_max.
196	events.push_back({"write",	5 !
197	"POSIX",
198	300,
199	9,
200	1,
201	true,
202	{{"size", static_cast<int64_t>(7), MetadataType::MT_VALUE},	2 ✔
203	{"bw", 2.0, MetadataType::MT_VALUE}}});	2 ✔
204
205	Aggregator aggregator;	1 ✔
206	PackedAggregator packed(config);	1 ✔
207	int rank = -1;	1 ✔
208	for (const Event& e : events) {	38 ✔
209	Metadata* md = metadata_of(e);	37 ✔
210	AggregatedKey key(e.category, e.name, TraceEventType::TRACE_TYPE_LIBC_IO,	37 ✔
211	e.ts, e.dur, e.tid, md, nullptr, &rank);	37 ✔
212	aggregator.aggregate(key);	37 ✔
213	delete md;	37 ✔
214	std::vector<uint8_t> bytes;	37 ✔
215	ipc::DecodedEvent decoded;	37 ✔
216	packed_of(e, true, bytes, decoded);	37 ✔
217	packed.aggregate(decoded);	37 ✔
218	}	37 ✔
219
220	AggregatedDataType data;	1 ✔
221	aggregator.get_previous_aggregations(data, true);	1 ✔
222	auto js = std::make_shared<JsonLines>();	1 ✔
223	js->initialize(nullptr, kHostname);	1 ✔
224	std::vector<char> expect(1 << 20);	1 ✔
225	const size_t n = js->aggregated(expect.data(), 0, 42, data);	1 ✔
226
227	struct Rows : PackedAggregator::RowSink {
228	std::vector<char> buffer = std::vector<char>(1 << 20);
229	size_t used = 0;
230	char* row() override { return buffer.data() + used; }	10 ✔
231	void wrote(size_t size) override { used += size; }	10 ✔
232	} rows;	1 ✔
233	packed.drain(true, 42, kHostname, rows);	1 ✔
234
235	const auto want = lines_of(expect.data(), n);	1 ✔
236	const auto got = lines_of(rows.buffer.data(), rows.used);	1 ✔
237	if (want != got) {	1 !
NEW 238	for (const auto& l : want) std::cerr << "want " << l << "\n";	×
NEW 239	for (const auto& l : got) std::cerr << "got " << l << "\n";	×
240	}
241	DFT_CHECK(want == got);	1 !
242	// 2 fhash x 2 threads x 2 intervals of read, close, write.
243	DFT_CHECK(got.size() == 10);	1 !
244	std::cout << "✓ PackedAggregator rows match Aggregator (" << got.size()	1 ✔
245	<< " rows)" << std::endl;	1 ✔
246	}	1 ✔
247
248	int main() {	1 ✔
249	setenv("DFTRACER_ENABLE", "1", 1);	1 ✔
250	setenv("DFTRACER_ENABLE_AGGREGATION", "1", 1);	1 ✔
251	setenv("DFTRACER_AGGREGATION_TYPE", "FULL", 1);	1 ✔
252	setenv("DFTRACER_TRACE_INTERVAL_MS", "1000", 1);	1 ✔
253	check_data();	1 ✔
254	check_aggregation();	1 ✔
255	return 0;	1 ✔
256	}

LLNL / dftracer / 40940235178

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