function-topology-streaming.c · cryptotaxi247/netdata

Code Commits Issues Pulls Sessions CI/CD

master

c 3,029 lines 136 KB

Raw

1	// SPDX-License-Identifier: GPL-3.0-or-later
2
3	#include "function-topology-streaming.h"
4
5	#define STREAMING_FUNCTION_UPDATE_EVERY 10
6
7
8	static bool streaming_topology_host_guid(RRDHOST host, char dst, size_t dst_size);
9	static bool streaming_topology_uuid_guid(ND_UUID host_id, char *dst, size_t dst_size);
10	static void streaming_topology_agent_id_for_host(RRDHOST host, char dst, size_t dst_size);
11	static void streaming_topology_actor_id_from_guid(const char guid, char dst, size_t dst_size);
12	static void streaming_topology_actor_id_for_uuid(ND_UUID host_id, char *dst, size_t dst_size);
13	static uint32_t streaming_topology_parent_child_count_get(DICTIONARY parent_child_count, RRDHOST *host);
14	static struct streaming_topology_descendant_list streaming_topology_descendants_get(DICTIONARY parent_descendants, RRDHOST *host);
15
16	enum streaming_topology_received_type {
17	STREAMING_TOPOLOGY_RECEIVED_STREAMING = 0,
18	STREAMING_TOPOLOGY_RECEIVED_VIRTUAL,
19	STREAMING_TOPOLOGY_RECEIVED_STALE,
20	};
21
22	struct streaming_topology_descendant {
23	RRDHOST *host;
24	ND_UUID source_uuid;
25	bool source_local;
26	uint8_t type;
27	};
28
29	struct streaming_topology_descendant_list {
30	struct streaming_topology_descendant *items;
31	size_t used;
32	size_t size;
33	};
34
35	struct streaming_topology_options {
36	bool info_only;
37	char *function_copy;
38	};
39
40	static bool streaming_topology_host_guid(RRDHOST host, char dst, size_t dst_size) {
41	if(!dst \|\| dst_size < UUID_STR_LEN)
42	return false;
43
44	dst[0] = '\0';
45	if(!host)
46	return false;
47
48	if(streaming_topology_uuid_guid(host->host_id, dst, dst_size))
49	return true;
50
51	if(host->machine_guid[0]) {
52	ND_UUID machine_guid = UUID_ZERO;
53	if(!uuid_parse(host->machine_guid, machine_guid.uuid))
54	return streaming_topology_uuid_guid(machine_guid, dst, dst_size);
55	}
56
57	return false;
58	}
59
60	static bool streaming_topology_uuid_guid(ND_UUID host_id, char *dst, size_t dst_size) {
61	if(!dst \|\| dst_size < UUID_STR_LEN)
62	return false;
63
64	dst[0] = '\0';
65	if(UUIDiszero(host_id))
66	return false;
67
68	uuid_unparse_lower(host_id.uuid, dst);
69	return true;
70	}
71
72	static void streaming_topology_actor_id_from_guid(const char guid, char dst, size_t dst_size) {
73	if(!dst \|\| !dst_size)
74	return;
75
76	if(guid && *guid)
77	snprintf(dst, dst_size, "netdata-machine-guid:%s", guid);
78	else
79	snprintf(dst, dst_size, "host:unknown");
80	}
81
82	static void streaming_topology_agent_id_for_host(RRDHOST host, char dst, size_t dst_size) {
83	if(!dst \|\| !dst_size)
84	return;
85
86	char host_guid[UUID_STR_LEN];
87	if(streaming_topology_host_guid(host, host_guid, sizeof(host_guid)))
88	snprintf(dst, dst_size, "%s", host_guid);
89	else if(host)
90	snprintf(dst, dst_size, "%s", rrdhost_hostname(host));
91	else
92	dst[0] = '\0';
93	}
94
95	static void streaming_topology_actor_id_for_uuid(ND_UUID host_id, char *dst, size_t dst_size) {
96	char guid[UUID_STR_LEN];
97	if(streaming_topology_uuid_guid(host_id, guid, sizeof(guid)))
98	streaming_topology_actor_id_from_guid(guid, dst, dst_size);
99	else
100	streaming_topology_actor_id_from_guid(NULL, dst, dst_size);
101	}
102
103	static uint32_t streaming_topology_parent_child_count_get(DICTIONARY parent_child_count, RRDHOST *host) {
104	char host_guid[UUID_STR_LEN];
105
106	if(!streaming_topology_host_guid(host, host_guid, sizeof(host_guid)))
107	return NULL;
108
109	return dictionary_get(parent_child_count, host_guid);
110	}
111
112	static struct streaming_topology_descendant_list streaming_topology_descendants_get(DICTIONARY parent_descendants, RRDHOST *host) {
113	char host_guid[UUID_STR_LEN];
114
115	if(!streaming_topology_host_guid(host, host_guid, sizeof(host_guid)))
116	return NULL;
117
118	return dictionary_get(parent_descendants, host_guid);
119	}
120
121	static struct streaming_topology_descendant_list streaming_topology_descendants_get_or_create(DICTIONARY parent_descendants, ND_UUID host_id) {
122	char host_guid[UUID_STR_LEN];
123	if(!streaming_topology_uuid_guid(host_id, host_guid, sizeof(host_guid)))
124	return NULL;
125
126	struct streaming_topology_descendant_list *list = dictionary_get(parent_descendants, host_guid);
127	if(list)
128	return list;
129
130	struct streaming_topology_descendant_list empty = {0};
131	return dictionary_set(parent_descendants, host_guid, &empty, sizeof(empty));
132	}
133
134	static void streaming_topology_descendants_append(
135	DICTIONARY *parent_descendants,
136	ND_UUID parent_id,
137	RRDHOST *host,
138	enum streaming_topology_received_type type,
139	bool source_local,
140	ND_UUID source_uuid
141	) {
142	struct streaming_topology_descendant_list *list = streaming_topology_descendants_get_or_create(parent_descendants, parent_id);
143	if(!list)
144	return;
145
146	if(list->used == list->size) {
147	size_t new_size = list->size ? list->size * 2 : 4;
148	list->items = reallocz(list->items, new_size * sizeof(*list->items));
149	list->size = new_size;
150	}
151
152	list->items[list->used++] = (struct streaming_topology_descendant) {
153	.host = host,
154	.source_uuid = source_uuid,
155	.source_local = source_local,
156	.type = (uint8_t)type,
157	};
158	}
159
160	static const char *streaming_topology_received_type_to_string(enum streaming_topology_received_type type) {
161	switch(type) {
162	case STREAMING_TOPOLOGY_RECEIVED_VIRTUAL:
163	return "virtual";
164
165	case STREAMING_TOPOLOGY_RECEIVED_STALE:
166	return "stale";
167
168	case STREAMING_TOPOLOGY_RECEIVED_STREAMING:
169	default:
170	return "streaming";
171	}
172	}
173
174	static void streaming_topology_actor_id_for_host(RRDHOST host, char dst, size_t dst_size) {
175	if(!dst \|\| !dst_size)
176	return;
177
178	char host_guid[UUID_STR_LEN];
179	if(streaming_topology_host_guid(host, host_guid, sizeof(host_guid)))
180	streaming_topology_actor_id_from_guid(host_guid, dst, dst_size);
181	else if(host)
182	snprintf(dst, dst_size, "hostname:%s", rrdhost_hostname(host));
183	else
184	snprintf(dst, dst_size, "host:unknown");
185	}
186
187	// get streaming_path host_ids, appending localhost only when the path already
188	// has upstream entries but does not yet include us; callers still use n == 0
189	// to detect hosts without an active path
190	static uint16_t streaming_topology_get_path_ids(RRDHOST host, uint16_t from, ND_UUID host_ids, uint16_t max) {
191	uint16_t n = rrdhost_stream_path_get_host_ids(host, from, host_ids, max);
192	uint16_t filtered_n = 0;
193
194	// check if localhost is already in the path
195	bool found_localhost = false;
196	for(uint16_t i = 0; i < n; i++) {
197	if(UUIDiszero(host_ids[i]))
198	continue;
199
200	host_ids[filtered_n++] = host_ids[i];
201	if(UUIDeq(host_ids[i], localhost->host_id)) {
202	found_localhost = true;
203	}
204	}
205	n = filtered_n;
206
207	// append localhost only when callers want the full path (from == 0). The
208	// self-append mirrors rrdhost_stream_path_to_json's emit-time semantics.
209	// For from > 0 (e.g. parent counting that asks for upstream-only entries)
210	// appending self would falsely count the host as its own parent.
211	if(from == 0 && !found_localhost && n < max && n > 0)
212	host_ids[n++] = localhost->host_id;
213
214	return n;
215	}
216
217	static void streaming_topology_parse_options(const char function, struct streaming_topology_options options) {
218	if(!options)
219	return;
220
221	*options = (struct streaming_topology_options){ 0 };
222	if(!function \|\| !*function)
223	return;
224
225	options->function_copy = strdupz(function);
226	char *words[1024];
227	size_t num_words = quoted_strings_splitter_whitespace(options->function_copy, words, 1024);
228	for(size_t i = 1; i < num_words; i++) {
229	char *param = get_word(words, num_words, i);
230	if(strcmp(param, "info") == 0)
231	options->info_only = true;
232	}
233	}
234
235	static int streaming_topology_return_error(BUFFER wb, char function_copy, int status, const char *error) {
236	buffer_flush(wb);
237	wb->content_type = CT_APPLICATION_JSON;
238	buffer_json_initialize(wb, "\"", "\"", 0, true, BUFFER_JSON_OPTIONS_DEFAULT);
239
240	buffer_json_member_add_uint64(wb, "status", status);
241	buffer_json_member_add_string(wb, "type", "topology");
242	buffer_json_member_add_time_t(wb, "update_every", STREAMING_FUNCTION_UPDATE_EVERY);
243	buffer_json_member_add_boolean(wb, "has_history", false);
244	buffer_json_member_add_string(wb, "help", RRDFUNCTIONS_STREAMING_TOPOLOGY_HELP);
245	buffer_json_member_add_string(wb, "error", error);
246	buffer_json_finalize(wb);
247
248	freez(function_copy);
249	return status;
250	}
251
252	static void streaming_topology_v1_emit_response_metadata(BUFFER *wb) {
253	buffer_json_member_add_uint64(wb, "status", HTTP_RESP_OK);
254	buffer_json_member_add_string(wb, "type", "topology");
255	buffer_json_member_add_time_t(wb, "update_every", STREAMING_FUNCTION_UPDATE_EVERY);
256	buffer_json_member_add_boolean(wb, "has_history", false);
257	buffer_json_member_add_string(wb, "help", RRDFUNCTIONS_STREAMING_TOPOLOGY_HELP);
258	buffer_json_member_add_array(wb, "accepted_params");
259	{
260	buffer_json_add_array_item_string(wb, "info");
261	}
262	buffer_json_array_close(wb);
263	buffer_json_member_add_array(wb, "required_params");
264	buffer_json_array_close(wb);
265	}
266
267	typedef struct streaming_topology_v1_actor {
268	char actor_id[256];
269	char type[32];
270	char machine_guid[UUID_STR_LEN];
271	char node_id[UUID_STR_LEN];
272	char hostname[256];
273	char display_name[256];
274	char severity[32];
275	char ephemerality[32];
276	char ingest_status[64];
277	char stream_status[64];
278	char ml_status[64];
279	char agent_name[128];
280	char agent_version[128];
281	char health_status[64];
282	char os_name[128];
283	char architecture[64];
284	char cpu_count[32];
285	uint64_t child_count;
286	uint64_t retained_node_count;
287	uint64_t health_critical;
288	uint64_t health_warning;
289	uint64_t health_clear;
290	RRDHOST *host;
291	bool synthetic;
292	} STREAMING_TOPOLOGY_V1_ACTOR;
293
294	typedef struct streaming_topology_v1_actor_label {
295	uint64_t actor;
296	char key[RRDLABELS_MAX_NAME_LENGTH + 1];
297	char value[RRDLABELS_MAX_VALUE_LENGTH + 1];
298	char source[32];
299	char kind[32];
300	bool has_value_index;
301	uint64_t value_index;
302	} STREAMING_TOPOLOGY_V1_ACTOR_LABEL;
303
304	typedef struct streaming_topology_v1_link {
305	uint64_t src_actor;
306	uint64_t dst_actor;
307	char type[32];
308	char state[64];
309	char port_name[256];
310	uint64_t discovered_at_ut;
311	uint64_t last_seen_ut;
312	int64_t hops;
313	uint64_t connections;
314	uint64_t replication_instances;
315	NETDATA_DOUBLE replication_completion;
316	uint64_t collected_metrics;
317	uint64_t collected_instances;
318	uint64_t collected_contexts;
319	} STREAMING_TOPOLOGY_V1_LINK;
320
321	typedef struct streaming_topology_v1_stream_path_row {
322	uint64_t actor;
323	uint64_t path_actor;
324	uint64_t path_index;
325	char hostname[256];
326	char host_id[UUID_STR_LEN];
327	char node_id[UUID_STR_LEN];
328	char claim_id[UUID_STR_LEN];
329	int64_t hops;
330	uint64_t since_ut;
331	uint64_t first_time_ut;
332	uint64_t start_time_ms;
333	uint64_t shutdown_time_ms;
334	uint64_t capabilities;
335	uint64_t flags;
336	} STREAMING_TOPOLOGY_V1_STREAM_PATH_ROW;
337
338	typedef struct streaming_topology_v1_retention_row {
339	uint64_t actor;
340	uint64_t observer_actor;
341	char db_status[64];
342	uint64_t db_from_ut;
343	uint64_t db_to_ut;
344	uint64_t db_duration;
345	uint64_t db_metrics;
346	uint64_t db_instances;
347	uint64_t db_contexts;
348	} STREAMING_TOPOLOGY_V1_RETENTION_ROW;
349
350	typedef struct streaming_topology_v1_inbound_row {
351	uint64_t parent_actor;
352	uint64_t child_actor;
353	bool has_source_actor;
354	uint64_t source_actor;
355	char received_type[32];
356	char ingest_status[64];
357	int64_t hops;
358	uint64_t collected_metrics;
359	uint64_t collected_instances;
360	uint64_t collected_contexts;
361	NETDATA_DOUBLE replication_completion;
362	uint64_t ingest_age;
363	char ssl[16];
364	uint64_t alerts_critical;
365	uint64_t alerts_warning;
366	} STREAMING_TOPOLOGY_V1_INBOUND_ROW;
367
368	typedef struct streaming_topology_v1_outbound_row {
369	uint64_t sender_actor;
370	uint64_t node_actor;
371	bool has_destination_actor;
372	uint64_t destination_actor;
373	char stream_status[64];
374	uint64_t stream_age;
375	int64_t hops;
376	char ssl[16];
377	char compression[24];
378	uint64_t collected_metrics;
379	uint64_t collected_instances;
380	uint64_t collected_contexts;
381	uint64_t replication_instances;
382	NETDATA_DOUBLE replication_completion;
383	} STREAMING_TOPOLOGY_V1_OUTBOUND_ROW;
384
385	typedef struct streaming_topology_v1_payload {
386	STREAMING_TOPOLOGY_V1_ACTOR *actors;
387	size_t actors_used;
388	size_t actors_size;
389
390	STREAMING_TOPOLOGY_V1_LINK *links;
391	size_t links_used;
392	size_t links_size;
393
394	STREAMING_TOPOLOGY_V1_ACTOR_LABEL *labels;
395	size_t labels_used;
396	size_t labels_size;
397
398	STREAMING_TOPOLOGY_V1_STREAM_PATH_ROW *stream_path_rows;
399	size_t stream_path_used;
400	size_t stream_path_size;
401
402	STREAMING_TOPOLOGY_V1_RETENTION_ROW *retention_rows;
403	size_t retention_used;
404	size_t retention_size;
405
406	STREAMING_TOPOLOGY_V1_INBOUND_ROW *inbound_rows;
407	size_t inbound_used;
408	size_t inbound_size;
409
410	STREAMING_TOPOLOGY_V1_OUTBOUND_ROW *outbound_rows;
411	size_t outbound_used;
412	size_t outbound_size;
413
414	DICTIONARY *actor_index;
415	DICTIONARY *emitted_links;
416	} STREAMING_TOPOLOGY_V1_PAYLOAD;
417
418	static void streaming_topology_v1_strncpy(char dst, size_t dst_size, const char src) {
419	if(!dst \|\| !dst_size)
420	return;
421
422	strncpyz(dst, src ? src : "", dst_size - 1);
423	}
424
425	static void streaming_topology_v1_uuid_str(ND_UUID uuid, char *dst, size_t dst_size) {
426	if(!dst \|\| !dst_size)
427	return;
428
429	if(!streaming_topology_uuid_guid(uuid, dst, dst_size))
430	dst[0] = '\0';
431	}
432
433	static void streaming_topology_v1_actor_index_set(STREAMING_TOPOLOGY_V1_PAYLOAD payload, const char actor_id, uint64_t index) {
434	dictionary_set(payload->actor_index, actor_id, &index, sizeof(index));
435	}
436
437	static bool streaming_topology_v1_actor_index_get(STREAMING_TOPOLOGY_V1_PAYLOAD payload, const char actor_id, uint64_t *index) {
438	uint64_t *stored = dictionary_get(payload->actor_index, actor_id);
439	if(!stored)
440	return false;
441
442	if(index)
443	index = stored;
444
445	return true;
446	}
447
448	static STREAMING_TOPOLOGY_V1_ACTOR streaming_topology_v1_add_actor(STREAMING_TOPOLOGY_V1_PAYLOAD payload, const char *actor_id) {
449	if(payload->actors_used == payload->actors_size) {
450	size_t new_size = payload->actors_size ? payload->actors_size * 2 : 16;
451	payload->actors = reallocz(payload->actors, new_size * sizeof(*payload->actors));
452	payload->actors_size = new_size;
453	}
454
455	STREAMING_TOPOLOGY_V1_ACTOR *actor = &payload->actors[payload->actors_used];
456	*actor = (STREAMING_TOPOLOGY_V1_ACTOR){ 0 };
457	streaming_topology_v1_strncpy(actor->actor_id, sizeof(actor->actor_id), actor_id);
458	streaming_topology_v1_actor_index_set(payload, actor_id, payload->actors_used);
459	payload->actors_used++;
460	return actor;
461	}
462
463	static STREAMING_TOPOLOGY_V1_LINK streaming_topology_v1_add_link(STREAMING_TOPOLOGY_V1_PAYLOAD payload) {
464	if(payload->links_used == payload->links_size) {
465	size_t new_size = payload->links_size ? payload->links_size * 2 : 16;
466	payload->links = reallocz(payload->links, new_size * sizeof(*payload->links));
467	payload->links_size = new_size;
468	}
469
470	STREAMING_TOPOLOGY_V1_LINK *link = &payload->links[payload->links_used++];
471	*link = (STREAMING_TOPOLOGY_V1_LINK){ 0 };
472	return link;
473	}
474
475	static const char *streaming_topology_v1_label_source(RRDLABEL_SRC source) {
476	if(source & RRDLABEL_SRC_K8S)
477	return "k8s";
478
479	if(source & RRDLABEL_SRC_ACLK)
480	return "aclk";
481
482	if(source & RRDLABEL_SRC_CONFIG)
483	return "config";
484
485	if(source & RRDLABEL_SRC_AUTO)
486	return "auto";
487
488	return "unknown";
489	}
490
491	static void streaming_topology_v1_add_actor_label_ex(
492	STREAMING_TOPOLOGY_V1_PAYLOAD *payload,
493	uint64_t actor,
494	const char *key,
495	const char *value,
496	const char *source,
497	const char *kind,
498	bool has_value_index,
499	uint64_t value_index) {
500	if(!payload \|\| !key \|\| !key \|\| !value \|\| !value)
501	return;
502
503	if(payload->labels_used == payload->labels_size) {
504	size_t new_size = payload->labels_size ? payload->labels_size * 2 : 64;
505	payload->labels = reallocz(payload->labels, new_size * sizeof(*payload->labels));
506	payload->labels_size = new_size;
507	}
508
509	STREAMING_TOPOLOGY_V1_ACTOR_LABEL *row = &payload->labels[payload->labels_used++];
510	*row = (STREAMING_TOPOLOGY_V1_ACTOR_LABEL){ 0 };
511	row->actor = actor;
512	streaming_topology_v1_strncpy(row->key, sizeof(row->key), key);
513	streaming_topology_v1_strncpy(row->value, sizeof(row->value), value);
514	streaming_topology_v1_strncpy(row->source, sizeof(row->source), source ? source : "producer");
515	streaming_topology_v1_strncpy(row->kind, sizeof(row->kind), kind ? kind : "metadata");
516	row->has_value_index = has_value_index;
517	row->value_index = value_index;
518	}
519
520	static void streaming_topology_v1_add_actor_label(
521	STREAMING_TOPOLOGY_V1_PAYLOAD *payload,
522	uint64_t actor,
523	const char *key,
524	const char *value,
525	const char *kind) {
526	streaming_topology_v1_add_actor_label_ex(payload, actor, key, value, "producer", kind, false, 0);
527	}
528
529	static void streaming_topology_v1_add_actor_label_uint(
530	STREAMING_TOPOLOGY_V1_PAYLOAD *payload,
531	uint64_t actor,
532	const char *key,
533	uint64_t value,
534	const char *kind) {
535	char text[32];
536	snprintfz(text, sizeof(text), "%"PRIu64, value);
537	streaming_topology_v1_add_actor_label(payload, actor, key, text, kind);
538	}
539
540	struct streaming_topology_v1_host_label_ctx {
541	STREAMING_TOPOLOGY_V1_PAYLOAD *payload;
542	uint64_t actor;
543	};
544
545	static int streaming_topology_v1_collect_host_label(
546	const char *name,
547	const char *value,
548	RRDLABEL_SRC source,
549	void *data) {
550	struct streaming_topology_v1_host_label_ctx *ctx = data;
551	streaming_topology_v1_add_actor_label_ex(
552	ctx->payload, ctx->actor, name, value,
553	streaming_topology_v1_label_source(source), "host_label", false, 0);
554	return 0;
555	}
556
557	static void streaming_topology_v1_collect_actor_labels(
558	STREAMING_TOPOLOGY_V1_PAYLOAD *payload,
559	uint64_t actor_index,
560	STREAMING_TOPOLOGY_V1_ACTOR *actor) {
561	if(!payload \|\| !actor)
562	return;
563
564	streaming_topology_v1_add_actor_label(payload, actor_index, "display_name", actor->display_name, "identity");
565	streaming_topology_v1_add_actor_label(payload, actor_index, "hostname", actor->hostname, "identity");
566	streaming_topology_v1_add_actor_label(payload, actor_index, "machine_guid", actor->machine_guid, "identity");
567	streaming_topology_v1_add_actor_label(payload, actor_index, "node_id", actor->node_id, "identity");
568	streaming_topology_v1_add_actor_label(payload, actor_index, "type", actor->type, "metadata");
569	streaming_topology_v1_add_actor_label(payload, actor_index, "severity", actor->severity, "status");
570	streaming_topology_v1_add_actor_label(payload, actor_index, "ephemerality", actor->ephemerality, "metadata");
571	streaming_topology_v1_add_actor_label(payload, actor_index, "ingest_status", actor->ingest_status, "status");
572	streaming_topology_v1_add_actor_label(payload, actor_index, "stream_status", actor->stream_status, "status");
573	streaming_topology_v1_add_actor_label(payload, actor_index, "ml_status", actor->ml_status, "status");
574	streaming_topology_v1_add_actor_label(payload, actor_index, "agent_name", actor->agent_name, "metadata");
575	streaming_topology_v1_add_actor_label(payload, actor_index, "agent_version", actor->agent_version, "metadata");
576	streaming_topology_v1_add_actor_label(payload, actor_index, "health_status", actor->health_status, "status");
577	streaming_topology_v1_add_actor_label(payload, actor_index, "os_name", actor->os_name, "system");
578	streaming_topology_v1_add_actor_label(payload, actor_index, "architecture", actor->architecture, "system");
579	streaming_topology_v1_add_actor_label(payload, actor_index, "cpu_count", actor->cpu_count, "system");
580	streaming_topology_v1_add_actor_label_uint(payload, actor_index, "child_count", actor->child_count, "metric");
581	streaming_topology_v1_add_actor_label_uint(
582	payload, actor_index, "retained_node_count", actor->retained_node_count, "metric");
583	streaming_topology_v1_add_actor_label_uint(payload, actor_index, "health_critical", actor->health_critical, "metric");
584	streaming_topology_v1_add_actor_label_uint(payload, actor_index, "health_warning", actor->health_warning, "metric");
585	streaming_topology_v1_add_actor_label_uint(payload, actor_index, "health_clear", actor->health_clear, "metric");
586
587	if(actor->host && actor->host->rrdlabels) {
588	struct streaming_topology_v1_host_label_ctx ctx = {
589	.payload = payload,
590	.actor = actor_index,
591	};
592	rrdlabels_walkthrough_read(actor->host->rrdlabels, streaming_topology_v1_collect_host_label, &ctx);
593	}
594	}
595
596	static STREAMING_TOPOLOGY_V1_STREAM_PATH_ROW *streaming_topology_v1_add_stream_path_row(
597	STREAMING_TOPOLOGY_V1_PAYLOAD *payload) {
598	if(payload->stream_path_used == payload->stream_path_size) {
599	size_t new_size = payload->stream_path_size ? payload->stream_path_size * 2 : 32;
600	payload->stream_path_rows = reallocz(payload->stream_path_rows, new_size * sizeof(*payload->stream_path_rows));
601	payload->stream_path_size = new_size;
602	}
603
604	STREAMING_TOPOLOGY_V1_STREAM_PATH_ROW *row = &payload->stream_path_rows[payload->stream_path_used++];
605	*row = (STREAMING_TOPOLOGY_V1_STREAM_PATH_ROW){ 0 };
606	return row;
607	}
608
609	static STREAMING_TOPOLOGY_V1_RETENTION_ROW *streaming_topology_v1_add_retention_row(
610	STREAMING_TOPOLOGY_V1_PAYLOAD *payload) {
611	if(payload->retention_used == payload->retention_size) {
612	size_t new_size = payload->retention_size ? payload->retention_size * 2 : 16;
613	payload->retention_rows = reallocz(payload->retention_rows, new_size * sizeof(*payload->retention_rows));
614	payload->retention_size = new_size;
615	}
616
617	STREAMING_TOPOLOGY_V1_RETENTION_ROW *row = &payload->retention_rows[payload->retention_used++];
618	*row = (STREAMING_TOPOLOGY_V1_RETENTION_ROW){ 0 };
619	return row;
620	}
621
622	static STREAMING_TOPOLOGY_V1_INBOUND_ROW *streaming_topology_v1_add_inbound_row(
623	STREAMING_TOPOLOGY_V1_PAYLOAD *payload) {
624	if(payload->inbound_used == payload->inbound_size) {
625	size_t new_size = payload->inbound_size ? payload->inbound_size * 2 : 32;
626	payload->inbound_rows = reallocz(payload->inbound_rows, new_size * sizeof(*payload->inbound_rows));
627	payload->inbound_size = new_size;
628	}
629
630	STREAMING_TOPOLOGY_V1_INBOUND_ROW *row = &payload->inbound_rows[payload->inbound_used++];
631	*row = (STREAMING_TOPOLOGY_V1_INBOUND_ROW){ 0 };
632	return row;
633	}
634
635	static STREAMING_TOPOLOGY_V1_OUTBOUND_ROW *streaming_topology_v1_add_outbound_row(
636	STREAMING_TOPOLOGY_V1_PAYLOAD *payload) {
637	if(payload->outbound_used == payload->outbound_size) {
638	size_t new_size = payload->outbound_size ? payload->outbound_size * 2 : 32;
639	payload->outbound_rows = reallocz(payload->outbound_rows, new_size * sizeof(*payload->outbound_rows));
640	payload->outbound_size = new_size;
641	}
642
643	STREAMING_TOPOLOGY_V1_OUTBOUND_ROW *row = &payload->outbound_rows[payload->outbound_used++];
644	*row = (STREAMING_TOPOLOGY_V1_OUTBOUND_ROW){ 0 };
645	return row;
646	}
647
648	static bool streaming_topology_v1_link_seen(STREAMING_TOPOLOGY_V1_PAYLOAD payload, uint64_t src, uint64_t dst, const char type) {
649	char link_key[128];
650	snprintfz(link_key, sizeof(link_key), "%"PRIu64"\|%"PRIu64"\|%s", src, dst, type ? type : "");
651	if(dictionary_get(payload->emitted_links, link_key))
652	return true;
653
654	uint8_t one = 1;
655	dictionary_set(payload->emitted_links, link_key, &one, sizeof(one));
656	return false;
657	}
658
659	static void streaming_topology_v1_free(STREAMING_TOPOLOGY_V1_PAYLOAD *payload) {
660	if(!payload)
661	return;
662
663	freez(payload->actors);
664	freez(payload->links);
665	freez(payload->labels);
666	freez(payload->stream_path_rows);
667	freez(payload->retention_rows);
668	freez(payload->inbound_rows);
669	freez(payload->outbound_rows);
670
671	if(payload->actor_index)
672	dictionary_destroy(payload->actor_index);
673	if(payload->emitted_links)
674	dictionary_destroy(payload->emitted_links);
675
676	*payload = (STREAMING_TOPOLOGY_V1_PAYLOAD){ 0 };
677	}
678
679	static void streaming_topology_v1_emit_column(
680	BUFFER *wb,
681	const char *id,
682	const char *type,
683	const char *role,
684	bool nullable,
685	const char *aggregation) {
686	buffer_json_add_array_item_object(wb);
687	buffer_json_member_add_string(wb, "id", id);
688	buffer_json_member_add_string(wb, "type", type);
689	if(nullable)
690	buffer_json_member_add_boolean(wb, "nullable", true);
691	if(role)
692	buffer_json_member_add_string(wb, "role", role);
693	if(aggregation)
694	buffer_json_member_add_string(wb, "aggregation", aggregation);
695	buffer_json_object_close(wb);
696	}
697
698	static void streaming_topology_v1_emit_values_start(BUFFER *wb) {
699	buffer_json_add_array_item_object(wb);
700	buffer_json_member_add_string(wb, "codec", "values");
701	buffer_json_member_add_array(wb, "values");
702	}
703
704	static void streaming_topology_v1_emit_values_end(BUFFER *wb) {
705	buffer_json_array_close(wb);
706	buffer_json_object_close(wb);
707	}
708
709	static void streaming_topology_v1_add_nullable_uint(BUFFER *wb, bool has_value, uint64_t value) {
710	if(has_value)
711	buffer_json_add_array_item_uint64(wb, value);
712	else
713	buffer_json_add_array_item_string(wb, NULL);
714	}
715
716	static uint64_t streaming_topology_v1_time_ut(time_t timestamp) {
717	return timestamp > 0 ? (uint64_t)timestamp * USEC_PER_SEC : 0;
718	}
719
720	static uint64_t streaming_topology_v1_best_since_ut(RRDHOST_STATUS *status) {
721	if(!status)
722	return 0;
723
724	if(status->ingest.since)
725	return streaming_topology_v1_time_ut(status->ingest.since);
726
727	if(status->stream.since)
728	return streaming_topology_v1_time_ut(status->stream.since);
729
730	if(status->db.first_time_s)
731	return streaming_topology_v1_time_ut(status->db.first_time_s);
732
733	return streaming_topology_v1_time_ut(netdata_start_time);
734	}
735
736	static uint64_t streaming_topology_v1_best_first_time_ut(RRDHOST_STATUS *status) {
737	if(!status)
738	return 0;
739
740	if(status->db.first_time_s)
741	return streaming_topology_v1_time_ut(status->db.first_time_s);
742
743	if(status->ingest.since)
744	return streaming_topology_v1_time_ut(status->ingest.since);
745
746	if(status->stream.since)
747	return streaming_topology_v1_time_ut(status->stream.since);
748
749	return streaming_topology_v1_time_ut(netdata_start_time);
750	}
751
752	static bool streaming_topology_v1_status_has_db_counts(RRDHOST_STATUS *status) {
753	return status && (status->db.metrics \|\| status->db.instances \|\| status->db.contexts);
754	}
755
756	static bool streaming_topology_v1_status_has_retention(RRDHOST_STATUS *status) {
757	return status &&
758	(status->db.first_time_s \|\| status->db.last_time_s \|\| streaming_topology_v1_status_has_db_counts(status));
759	}
760
761	static uint64_t streaming_topology_v1_count_local_retained_nodes(time_t now) {
762	uint64_t retained = 0;
763
764	RRDHOST *host;
765	dfe_start_read(rrdhost_root_index, host) {
766	RRDHOST_STATUS status;
767	rrdhost_status(host, now, &status, RRDHOST_STATUS_ALL);
768	if(streaming_topology_v1_status_has_retention(&status))
769	retained++;
770	}
771	dfe_done(host);
772
773	return retained;
774	}
775
776	static uint64_t streaming_topology_v1_retention_from_ut(RRDHOST_STATUS *status) {
777	if(!status)
778	return 0;
779
780	if(status->db.first_time_s)
781	return streaming_topology_v1_time_ut(status->db.first_time_s);
782
783	if(streaming_topology_v1_status_has_db_counts(status))
784	return streaming_topology_v1_best_first_time_ut(status);
785
786	return 0;
787	}
788
789	static uint64_t streaming_topology_v1_retention_to_ut(RRDHOST_STATUS *status) {
790	if(!status)
791	return 0;
792
793	if(status->db.last_time_s)
794	return streaming_topology_v1_time_ut(status->db.last_time_s);
795
796	if(streaming_topology_v1_status_has_db_counts(status))
797	return streaming_topology_v1_time_ut(status->now);
798
799	return 0;
800	}
801
802	static void streaming_topology_v1_add_timestamp(BUFFER *wb, uint64_t timestamp_ut) {
803	if(timestamp_ut)
804	buffer_json_add_array_item_datetime_rfc3339(wb, timestamp_ut, true);
805	else
806	buffer_json_add_array_item_string(wb, NULL);
807	}
808
809	static const char *streaming_topology_v1_node_type(
810	RRDHOST *host,
811	RRDHOST_STATUS *status,
812	DICTIONARY *parent_child_count) {
813	if(rrdhost_is_virtual(host))
814	return "vnode";
815
816	if(host != localhost && status->ingest.status == RRDHOST_INGEST_STATUS_ARCHIVED)
817	return "stale";
818
819	uint32_t *cc = streaming_topology_parent_child_count_get(parent_child_count, host);
820	return (cc && *cc > 0) ? "parent" : "child";
821	}
822
823	static const char streaming_topology_v1_severity(RRDHOST host, RRDHOST_STATUS *status) {
824	if(rrdhost_option_check(host, RRDHOST_OPTION_EPHEMERAL_HOST))
825	return "normal";
826
827	switch(status->ingest.status) {
828	case RRDHOST_INGEST_STATUS_OFFLINE:
829	case RRDHOST_INGEST_STATUS_ARCHIVED:
830	return "critical";
831	default:
832	break;
833	}
834
835	if(status->stream.status == RRDHOST_STREAM_STATUS_OFFLINE &&
836	status->stream.reason != STREAM_HANDSHAKE_SP_NO_DESTINATION)
837	return "warning";
838
839	return "normal";
840	}
841
842	struct streaming_topology_v1_synth_actor_ctx {
843	STREAMING_TOPOLOGY_V1_PAYLOAD *payload;
844	};
845
846	static bool streaming_topology_v1_collect_synth_actor(
847	void *userdata,
848	uint16_t index __maybe_unused,
849	STRING *hostname,
850	ND_UUID host_id,
851	ND_UUID node_id,
852	ND_UUID claim_id __maybe_unused,
853	int16_t hops __maybe_unused,
854	time_t since __maybe_unused,
855	time_t first_time_t __maybe_unused,
856	uint32_t start_time_ms __maybe_unused,
857	uint32_t shutdown_time_ms __maybe_unused,
858	STREAM_CAPABILITIES capabilities __maybe_unused,
859	uint32_t flags __maybe_unused) {
860	struct streaming_topology_v1_synth_actor_ctx *ctx = userdata;
861
862	if(UUIDeq(host_id, localhost->host_id))
863	return true;
864
865	char guid[UUID_STR_LEN];
866	if(!streaming_topology_uuid_guid(host_id, guid, sizeof(guid)))
867	return true;
868
869	char actor_id[256];
870	streaming_topology_actor_id_from_guid(guid, actor_id, sizeof(actor_id));
871	if(streaming_topology_v1_actor_index_get(ctx->payload, actor_id, NULL))
872	return true;
873
874	STREAMING_TOPOLOGY_V1_ACTOR *actor = streaming_topology_v1_add_actor(ctx->payload, actor_id);
875	actor->synthetic = true;
876	streaming_topology_v1_strncpy(actor->type, sizeof(actor->type), "parent");
877	streaming_topology_v1_strncpy(actor->machine_guid, sizeof(actor->machine_guid), guid);
878	streaming_topology_v1_uuid_str(node_id, actor->node_id, sizeof(actor->node_id));
879	streaming_topology_v1_strncpy(actor->hostname, sizeof(actor->hostname), string2str(hostname));
880	streaming_topology_v1_strncpy(actor->display_name, sizeof(actor->display_name), string2str(hostname));
881	streaming_topology_v1_strncpy(actor->severity, sizeof(actor->severity), "normal");
882	streaming_topology_v1_strncpy(actor->ephemerality, sizeof(actor->ephemerality), "permanent");
883	streaming_topology_v1_strncpy(actor->ingest_status, sizeof(actor->ingest_status), "unknown");
884	streaming_topology_v1_strncpy(actor->stream_status, sizeof(actor->stream_status), "unknown");
885	streaming_topology_v1_strncpy(actor->ml_status, sizeof(actor->ml_status), "unknown");
886	streaming_topology_v1_strncpy(actor->health_status, sizeof(actor->health_status), "unknown");
887	streaming_topology_v1_collect_actor_labels(ctx->payload, ctx->payload->actors_used - 1, actor);
888	return true;
889	}
890
891	static void streaming_topology_v1_collect_actors(
892	STREAMING_TOPOLOGY_V1_PAYLOAD *payload,
893	DICTIONARY *parent_child_count,
894	uint64_t local_retained_node_count,
895	time_t now) {
896	RRDHOST *host;
897	dfe_start_read(rrdhost_root_index, host) {
898	RRDHOST_STATUS status;
899	rrdhost_status(host, now, &status, RRDHOST_STATUS_ALL);
900
901	char actor_id[256];
902	streaming_topology_actor_id_for_host(host, actor_id, sizeof(actor_id));
903
904	STREAMING_TOPOLOGY_V1_ACTOR *actor = streaming_topology_v1_add_actor(payload, actor_id);
905	actor->host = host;
906	streaming_topology_v1_strncpy(actor->type, sizeof(actor->type),
907	streaming_topology_v1_node_type(host, &status, parent_child_count));
908	streaming_topology_host_guid(host, actor->machine_guid, sizeof(actor->machine_guid));
909	streaming_topology_v1_uuid_str(host->node_id, actor->node_id, sizeof(actor->node_id));
910	streaming_topology_v1_strncpy(actor->hostname, sizeof(actor->hostname), rrdhost_hostname(host));
911	streaming_topology_v1_strncpy(actor->display_name, sizeof(actor->display_name), rrdhost_hostname(host));
912	streaming_topology_v1_strncpy(actor->severity, sizeof(actor->severity),
913	streaming_topology_v1_severity(host, &status));
914	streaming_topology_v1_strncpy(actor->ephemerality, sizeof(actor->ephemerality),
915	rrdhost_option_check(host, RRDHOST_OPTION_EPHEMERAL_HOST) ? "ephemeral" : "permanent");
916	streaming_topology_v1_strncpy(actor->ingest_status, sizeof(actor->ingest_status),
917	rrdhost_ingest_status_to_string(status.ingest.status));
918	streaming_topology_v1_strncpy(actor->stream_status, sizeof(actor->stream_status),
919	rrdhost_streaming_status_to_string(status.stream.status));
920	streaming_topology_v1_strncpy(actor->ml_status, sizeof(actor->ml_status),
921	rrdhost_ml_status_to_string(status.ml.status));
922	streaming_topology_v1_strncpy(actor->agent_name, sizeof(actor->agent_name), rrdhost_program_name(host));
923	streaming_topology_v1_strncpy(actor->agent_version, sizeof(actor->agent_version), rrdhost_program_version(host));
924	streaming_topology_v1_strncpy(actor->health_status, sizeof(actor->health_status),
925	rrdhost_health_status_to_string(status.health.status));
926	rrdlabels_get_value_strcpyz(host->rrdlabels, actor->os_name, sizeof(actor->os_name), "_os_name");
927	rrdlabels_get_value_strcpyz(host->rrdlabels, actor->architecture, sizeof(actor->architecture), "_architecture");
928	rrdlabels_get_value_strcpyz(host->rrdlabels, actor->cpu_count, sizeof(actor->cpu_count), "_system_cores");
929
930	uint32_t *cc = streaming_topology_parent_child_count_get(parent_child_count, host);
931	actor->child_count = cc ? *cc : 0;
932	actor->retained_node_count = host == localhost ? local_retained_node_count : 0;
933	if(status.health.status == RRDHOST_HEALTH_STATUS_RUNNING) {
934	actor->health_critical = status.health.alerts.critical;
935	actor->health_warning = status.health.alerts.warning;
936	actor->health_clear = status.health.alerts.clear;
937	}
938
939	streaming_topology_v1_collect_actor_labels(payload, payload->actors_used - 1, actor);
940	}
941	dfe_done(host);
942
943	struct streaming_topology_v1_synth_actor_ctx ctx = { .payload = payload };
944	dfe_start_read(rrdhost_root_index, host) {
945	rrdhost_stream_path_visit(host, 1, streaming_topology_v1_collect_synth_actor, &ctx);
946	}
947	dfe_done(host);
948	}
949
950	static bool streaming_topology_v1_actor_index_for_host(
951	STREAMING_TOPOLOGY_V1_PAYLOAD *payload,
952	RRDHOST *host,
953	uint64_t *index) {
954	char actor_id[256];
955	streaming_topology_actor_id_for_host(host, actor_id, sizeof(actor_id));
956	return streaming_topology_v1_actor_index_get(payload, actor_id, index);
957	}
958
959	static bool streaming_topology_v1_actor_index_for_uuid(
960	STREAMING_TOPOLOGY_V1_PAYLOAD *payload,
961	ND_UUID host_id,
962	uint64_t *index) {
963	char actor_id[256];
964	streaming_topology_actor_id_for_uuid(host_id, actor_id, sizeof(actor_id));
965	return streaming_topology_v1_actor_index_get(payload, actor_id, index);
966	}
967
968	static void streaming_topology_v1_add_link_if_new(
969	STREAMING_TOPOLOGY_V1_PAYLOAD *payload,
970	uint64_t src_actor,
971	uint64_t dst_actor,
972	const char *type,
973	const char *state,
974	const char *port_name,
975	uint64_t discovered_at_ut,
976	uint64_t last_seen_ut,
977	int64_t hops,
978	uint64_t connections,
979	uint64_t replication_instances,
980	NETDATA_DOUBLE replication_completion,
981	uint64_t collected_metrics,
982	uint64_t collected_instances,
983	uint64_t collected_contexts) {
984	if(streaming_topology_v1_link_seen(payload, src_actor, dst_actor, type))
985	return;
986
987	STREAMING_TOPOLOGY_V1_LINK *link = streaming_topology_v1_add_link(payload);
988	link->src_actor = src_actor;
989	link->dst_actor = dst_actor;
990	streaming_topology_v1_strncpy(link->type, sizeof(link->type), type);
991	streaming_topology_v1_strncpy(link->state, sizeof(link->state), state);
992	streaming_topology_v1_strncpy(link->port_name, sizeof(link->port_name), port_name);
993	link->discovered_at_ut = discovered_at_ut;
994	link->last_seen_ut = last_seen_ut;
995	link->hops = hops;
996	link->connections = connections;
997	link->replication_instances = replication_instances;
998	link->replication_completion = replication_completion;
999	link->collected_metrics = collected_metrics;
1000	link->collected_instances = collected_instances;
1001	link->collected_contexts = collected_contexts;
1002	}
1003
1004	struct streaming_topology_v1_synth_link_ctx {
1005	STREAMING_TOPOLOGY_V1_PAYLOAD *payload;
1006	bool has_prev;
1007	uint64_t prev_actor;
1008	char prev_actor_id[256];
1009	char prev_hostname[256];
1010	time_t prev_since;
1011	time_t prev_first_time_t;
1012	};
1013
1014	static bool streaming_topology_v1_collect_synth_link(
1015	void *userdata,
1016	uint16_t index __maybe_unused,
1017	STRING *hostname,
1018	ND_UUID host_id,
1019	ND_UUID node_id __maybe_unused,
1020	ND_UUID claim_id __maybe_unused,
1021	int16_t hops __maybe_unused,
1022	time_t since,
1023	time_t first_time_t,
1024	uint32_t start_time_ms __maybe_unused,
1025	uint32_t shutdown_time_ms __maybe_unused,
1026	STREAM_CAPABILITIES capabilities __maybe_unused,
1027	uint32_t flags __maybe_unused) {
1028	struct streaming_topology_v1_synth_link_ctx *ctx = userdata;
1029
1030	char guid[UUID_STR_LEN];
1031	if(!streaming_topology_uuid_guid(host_id, guid, sizeof(guid))) {
1032	ctx->has_prev = false;
1033	return true;
1034	}
1035
1036	char cur_actor_id[256];
1037	streaming_topology_actor_id_from_guid(guid, cur_actor_id, sizeof(cur_actor_id));
1038
1039	uint64_t cur_actor;
1040	if(!streaming_topology_v1_actor_index_get(ctx->payload, cur_actor_id, &cur_actor)) {
1041	ctx->has_prev = false;
1042	return true;
1043	}
1044
1045	if(ctx->has_prev) {
1046	streaming_topology_v1_add_link_if_new(
1047	ctx->payload,
1048	ctx->prev_actor,
1049	cur_actor,
1050	"streaming",
1051	"online",
1052	ctx->prev_hostname,
1053	((uint64_t)(ctx->prev_first_time_t ? ctx->prev_first_time_t : ctx->prev_since)) * USEC_PER_SEC,
1054	((uint64_t)(since ? since : ctx->prev_since)) * USEC_PER_SEC,
1055	0, 0, 0, 0, 0, 0, 0);
1056	}
1057
1058	ctx->has_prev = true;
1059	ctx->prev_actor = cur_actor;
1060	streaming_topology_v1_strncpy(ctx->prev_actor_id, sizeof(ctx->prev_actor_id), cur_actor_id);
1061	streaming_topology_v1_strncpy(ctx->prev_hostname, sizeof(ctx->prev_hostname), string2str(hostname));
1062	ctx->prev_since = since;
1063	ctx->prev_first_time_t = first_time_t;
1064	return true;
1065	}
1066
1067	static void streaming_topology_v1_collect_links(
1068	STREAMING_TOPOLOGY_V1_PAYLOAD *payload,
1069	time_t now,
1070	usec_t now_ut) {
1071	char localhost_actor_id[256];
1072	streaming_topology_actor_id_for_host(localhost, localhost_actor_id, sizeof(localhost_actor_id));
1073	uint64_t localhost_actor = 0;
1074	streaming_topology_v1_actor_index_get(payload, localhost_actor_id, &localhost_actor);
1075
1076	RRDHOST *host;
1077	dfe_start_read(rrdhost_root_index, host) {
1078	if(host == localhost)
1079	continue;
1080
1081	RRDHOST_STATUS status;
1082	rrdhost_status(host, now, &status, RRDHOST_STATUS_ALL);
1083
1084	uint64_t src_actor;
1085	if(!streaming_topology_v1_actor_index_for_host(payload, host, &src_actor))
1086	continue;
1087
1088	char target_actor_id[256] = "";
1089	const char *link_type = NULL;
1090	bool is_vnode = rrdhost_is_virtual(host);
1091	ND_UUID link_ids[2];
1092	uint16_t link_n = streaming_topology_get_path_ids(host, 0, link_ids, 2);
1093
1094	if(is_vnode) {
1095	snprintfz(target_actor_id, sizeof(target_actor_id), "%s", localhost_actor_id);
1096	link_type = "virtual";
1097	}
1098	else if(link_n >= 2) {
1099	streaming_topology_actor_id_for_uuid(link_ids[1], target_actor_id, sizeof(target_actor_id));
1100	link_type = "streaming";
1101	}
1102	else {
1103	snprintfz(target_actor_id, sizeof(target_actor_id), "%s", localhost_actor_id);
1104	link_type = "stale";
1105	}
1106
1107	uint64_t dst_actor;
1108	if(!streaming_topology_v1_actor_index_get(payload, target_actor_id, &dst_actor))
1109	continue;
1110
1111	streaming_topology_v1_add_link_if_new(
1112	payload,
1113	src_actor,
1114	dst_actor,
1115	link_type,
1116	rrdhost_ingest_status_to_string(status.ingest.status),
1117	rrdhost_hostname(host),
1118	((uint64_t)(status.ingest.since ? status.ingest.since : now)) * USEC_PER_SEC,
1119	now_ut,
1120	status.ingest.hops,
1121	strcmp(link_type, "virtual") != 0 ? status.host->stream.rcv.status.connections : 0,
1122	strcmp(link_type, "virtual") != 0 ? status.ingest.replication.instances : 0,
1123	strcmp(link_type, "virtual") != 0 ? status.ingest.replication.completion : 0,
1124	strcmp(link_type, "virtual") != 0 ? status.ingest.collected.metrics : 0,
1125	strcmp(link_type, "virtual") != 0 ? status.ingest.collected.instances : 0,
1126	strcmp(link_type, "virtual") != 0 ? status.ingest.collected.contexts : 0);
1127	}
1128	dfe_done(host);
1129
1130	dfe_start_read(rrdhost_root_index, host) {
1131	struct streaming_topology_v1_synth_link_ctx ctx = {
1132	.payload = payload,
1133	.has_prev = false,
1134	};
1135	rrdhost_stream_path_visit(host, 0, streaming_topology_v1_collect_synth_link, &ctx);
1136	}
1137	dfe_done(host);
1138	}
1139
1140	struct streaming_topology_v1_stream_path_ctx {
1141	STREAMING_TOPOLOGY_V1_PAYLOAD *payload;
1142	uint64_t actor;
1143	bool seen_localhost;
1144	bool emitted;
1145	uint16_t next_index;
1146	};
1147
1148	static bool streaming_topology_v1_collect_stream_path_row(
1149	void *userdata,
1150	uint16_t index,
1151	STRING *hostname,
1152	ND_UUID host_id,
1153	ND_UUID node_id,
1154	ND_UUID claim_id,
1155	int16_t hops,
1156	time_t since,
1157	time_t first_time_t,
1158	uint32_t start_time_ms,
1159	uint32_t shutdown_time_ms,
1160	STREAM_CAPABILITIES capabilities,
1161	uint32_t flags) {
1162	struct streaming_topology_v1_stream_path_ctx *ctx = userdata;
1163
1164	STREAMING_TOPOLOGY_V1_STREAM_PATH_ROW *row = streaming_topology_v1_add_stream_path_row(ctx->payload);
1165	row->actor = ctx->actor;
1166	row->path_actor = ctx->actor;
1167	row->path_index = index;
1168	char path_actor_guid[UUID_STR_LEN];
1169	if(streaming_topology_uuid_guid(host_id, path_actor_guid, sizeof(path_actor_guid))) {
1170	char path_actor_id[256];
1171	streaming_topology_actor_id_from_guid(path_actor_guid, path_actor_id, sizeof(path_actor_id));
1172	streaming_topology_v1_actor_index_get(ctx->payload, path_actor_id, &row->path_actor);
1173	}
1174	streaming_topology_v1_strncpy(row->hostname, sizeof(row->hostname), string2str(hostname));
1175	streaming_topology_v1_uuid_str(host_id, row->host_id, sizeof(row->host_id));
1176	streaming_topology_v1_uuid_str(node_id, row->node_id, sizeof(row->node_id));
1177	streaming_topology_v1_uuid_str(claim_id, row->claim_id, sizeof(row->claim_id));
1178	row->hops = hops;
1179	row->since_ut = since > 0 ? (uint64_t)since * USEC_PER_SEC : 0;
1180	row->first_time_ut = first_time_t > 0 ? (uint64_t)first_time_t * USEC_PER_SEC : 0;
1181	row->start_time_ms = start_time_ms;
1182	row->shutdown_time_ms = shutdown_time_ms;
1183	row->capabilities = capabilities;
1184	row->flags = flags;
1185	if(UUIDeq(host_id, localhost->host_id))
1186	ctx->seen_localhost = true;
1187	ctx->emitted = true;
1188	ctx->next_index = index + 1;
1189	return true;
1190	}
1191
1192	static void streaming_topology_v1_collect_actor_detail_rows(
1193	STREAMING_TOPOLOGY_V1_PAYLOAD *payload,
1194	DICTIONARY *parent_descendants,
1195	time_t now) {
1196	uint64_t localhost_actor = 0;
1197	streaming_topology_v1_actor_index_for_host(payload, localhost, &localhost_actor);
1198
1199	for(size_t i = 0; i < payload->actors_used; i++) {
1200	STREAMING_TOPOLOGY_V1_ACTOR *actor = &payload->actors[i];
1201	if(!actor->host)
1202	continue;
1203
1204	RRDHOST_STATUS status;
1205	rrdhost_status(actor->host, now, &status, RRDHOST_STATUS_ALL);
1206
1207	size_t path_rows_start = payload->stream_path_used;
1208	struct streaming_topology_v1_stream_path_ctx sp_ctx = {
1209	.payload = payload,
1210	.actor = i,
1211	.emitted = false,
1212	};
1213	rrdhost_stream_path_visit(actor->host, 0, streaming_topology_v1_collect_stream_path_row, &sp_ctx);
1214	for(size_t pi = path_rows_start; pi < payload->stream_path_used; pi++) {
1215	STREAMING_TOPOLOGY_V1_STREAM_PATH_ROW *path_row = &payload->stream_path_rows[pi];
1216	if(!path_row->since_ut)
1217	path_row->since_ut = streaming_topology_v1_best_since_ut(&status);
1218	if(!path_row->first_time_ut)
1219	path_row->first_time_ut = streaming_topology_v1_best_first_time_ut(&status);
1220	}
1221	if(sp_ctx.emitted && !sp_ctx.seen_localhost) {
1222	// Match the legacy highlight path helper: stored paths do not
1223	// always carry the local agent, but rendered paths need it.
1224	STREAMING_TOPOLOGY_V1_STREAM_PATH_ROW *row = streaming_topology_v1_add_stream_path_row(payload);
1225	row->actor = i;
1226	row->path_actor = localhost_actor;
1227	row->path_index = sp_ctx.next_index;
1228	streaming_topology_v1_strncpy(row->hostname, sizeof(row->hostname), rrdhost_hostname(localhost));
1229	streaming_topology_host_guid(localhost, row->host_id, sizeof(row->host_id));
1230	streaming_topology_v1_uuid_str(localhost->node_id, row->node_id, sizeof(row->node_id));
1231	row->hops = status.ingest.hops;
1232	row->since_ut = streaming_topology_v1_best_since_ut(&status);
1233	row->first_time_ut = streaming_topology_v1_best_first_time_ut(&status);
1234	}
1235	else if(!sp_ctx.emitted) {
1236	STREAMING_TOPOLOGY_V1_STREAM_PATH_ROW *row = streaming_topology_v1_add_stream_path_row(payload);
1237	row->actor = i;
1238	row->path_actor = i;
1239	row->path_index = 0;
1240	streaming_topology_v1_strncpy(row->hostname, sizeof(row->hostname), actor->hostname);
1241	streaming_topology_v1_strncpy(row->host_id, sizeof(row->host_id), actor->machine_guid);
1242	streaming_topology_v1_strncpy(row->node_id, sizeof(row->node_id), actor->node_id);
1243	row->hops = status.ingest.hops;
1244	row->since_ut = streaming_topology_v1_best_since_ut(&status);
1245	row->first_time_ut = streaming_topology_v1_best_first_time_ut(&status);
1246	}
1247
1248	if(streaming_topology_v1_status_has_retention(&status)) {
1249	STREAMING_TOPOLOGY_V1_RETENTION_ROW *retention = streaming_topology_v1_add_retention_row(payload);
1250	retention->actor = i;
1251	retention->observer_actor = localhost_actor;
1252	streaming_topology_v1_strncpy(retention->db_status, sizeof(retention->db_status),
1253	rrdhost_db_status_to_string(status.db.status));
1254	retention->db_from_ut = streaming_topology_v1_retention_from_ut(&status);
1255	retention->db_to_ut = streaming_topology_v1_retention_to_ut(&status);
1256	retention->db_duration =
1257	retention->db_from_ut && retention->db_to_ut && retention->db_to_ut > retention->db_from_ut ?
1258	(retention->db_to_ut - retention->db_from_ut) / USEC_PER_SEC : 0;
1259	retention->db_metrics = status.db.metrics;
1260	retention->db_instances = status.db.instances;
1261	retention->db_contexts = status.db.contexts;
1262	}
1263	}
1264
1265	for(size_t i = 0; i < payload->actors_used; i++) {
1266	STREAMING_TOPOLOGY_V1_ACTOR *parent = &payload->actors[i];
1267	if(!parent->host \|\| parent->child_count == 0)
1268	continue;
1269
1270	struct streaming_topology_descendant_list *nodes =
1271	streaming_topology_descendants_get(parent_descendants, parent->host);
1272	if(!nodes)
1273	continue;
1274
1275	for(size_t j = 0; j < nodes->used; j++) {
1276	struct streaming_topology_descendant *descendant = &nodes->items[j];
1277	if(!descendant->host \|\| descendant->host == parent->host)
1278	continue;
1279
1280	uint64_t child_actor;
1281	if(!streaming_topology_v1_actor_index_for_host(payload, descendant->host, &child_actor))
1282	continue;
1283
1284	RRDHOST_STATUS status;
1285	rrdhost_status(descendant->host, now, &status, RRDHOST_STATUS_ALL);
1286
1287	STREAMING_TOPOLOGY_V1_INBOUND_ROW *row = streaming_topology_v1_add_inbound_row(payload);
1288	row->parent_actor = i;
1289	row->child_actor = child_actor;
1290	streaming_topology_v1_strncpy(row->received_type, sizeof(row->received_type),
1291	streaming_topology_received_type_to_string((enum streaming_topology_received_type)descendant->type));
1292	streaming_topology_v1_strncpy(row->ingest_status, sizeof(row->ingest_status),
1293	rrdhost_ingest_status_to_string(status.ingest.status));
1294	row->hops = status.ingest.hops;
1295	row->collected_metrics = status.ingest.collected.metrics;
1296	row->collected_instances = status.ingest.collected.instances;
1297	row->collected_contexts = status.ingest.collected.contexts;
1298	row->replication_completion = status.ingest.replication.completion;
1299	row->ingest_age = status.ingest.since ? (uint64_t)(status.now - status.ingest.since) : 0;
1300	streaming_topology_v1_strncpy(row->ssl, sizeof(row->ssl), status.ingest.ssl ? "SSL" : "PLAIN");
1301	row->alerts_critical =
1302	status.health.status == RRDHOST_HEALTH_STATUS_RUNNING ? status.health.alerts.critical : 0;
1303	row->alerts_warning =
1304	status.health.status == RRDHOST_HEALTH_STATUS_RUNNING ? status.health.alerts.warning : 0;
1305
1306	if(descendant->source_local) {
1307	row->has_source_actor = true;
1308	row->source_actor = child_actor;
1309	}
1310	else if(!UUIDiszero(descendant->source_uuid)) {
1311	char source_actor_id[256];
1312	streaming_topology_actor_id_for_uuid(descendant->source_uuid, source_actor_id, sizeof(source_actor_id));
1313	row->has_source_actor =
1314	streaming_topology_v1_actor_index_get(payload, source_actor_id, &row->source_actor);
1315	}
1316	}
1317	}
1318
1319	RRDHOST_STATUS sender_status;
1320	rrdhost_status(localhost, now, &sender_status, RRDHOST_STATUS_ALL);
1321
1322	for(size_t i = 0; i < payload->actors_used; i++) {
1323	STREAMING_TOPOLOGY_V1_ACTOR *actor = &payload->actors[i];
1324	if(!actor->host)
1325	continue;
1326
1327	RRDHOST *path_host = rrdhost_is_virtual(actor->host) ? localhost : actor->host;
1328	ND_UUID path[128];
1329	uint16_t path_n = streaming_topology_get_path_ids(path_host, 0, path, 128);
1330
1331	uint64_t destination_actor = 0;
1332	bool has_destination_actor = false;
1333	for(uint16_t pi = 0; pi + 1 < path_n; pi++) {
1334	if(UUIDeq(path[pi], localhost->host_id)) {
1335	has_destination_actor =
1336	streaming_topology_v1_actor_index_for_uuid(payload, path[pi + 1], &destination_actor);
1337	break;
1338	}
1339	}
1340
1341	if(!has_destination_actor)
1342	continue;
1343
1344	RRDHOST_STATUS node_status;
1345	rrdhost_status(actor->host, now, &node_status, RRDHOST_STATUS_ALL);
1346
1347	STREAMING_TOPOLOGY_V1_OUTBOUND_ROW *row = streaming_topology_v1_add_outbound_row(payload);
1348	row->sender_actor = localhost_actor;
1349	row->node_actor = i;
1350	row->has_destination_actor = true;
1351	row->destination_actor = destination_actor;
1352	streaming_topology_v1_strncpy(row->stream_status, sizeof(row->stream_status),
1353	rrdhost_streaming_status_to_string(sender_status.stream.status));
1354	row->stream_age =
1355	sender_status.stream.since && sender_status.now >= sender_status.stream.since ?
1356	(uint64_t)(sender_status.now - sender_status.stream.since) : 0;
1357	row->hops = sender_status.stream.hops;
1358	streaming_topology_v1_strncpy(row->ssl, sizeof(row->ssl), sender_status.stream.ssl ? "SSL" : "PLAIN");
1359	streaming_topology_v1_strncpy(row->compression, sizeof(row->compression),
1360	sender_status.stream.compression ? "COMPRESSED" : "UNCOMPRESSED");
1361	row->collected_metrics = node_status.ingest.collected.metrics ?
1362	node_status.ingest.collected.metrics : node_status.db.metrics;
1363	row->collected_instances = node_status.ingest.collected.instances ?
1364	node_status.ingest.collected.instances : node_status.db.instances;
1365	row->collected_contexts = node_status.ingest.collected.contexts ?
1366	node_status.ingest.collected.contexts : node_status.db.contexts;
1367	row->replication_instances = node_status.ingest.replication.instances;
1368	row->replication_completion = node_status.ingest.replication.completion;
1369	}
1370	}
1371
1372	static void streaming_topology_v1_emit_actor_columns(BUFFER *wb) {
1373	streaming_topology_v1_emit_column(wb, "id", "string", "identity", false, NULL);
1374	streaming_topology_v1_emit_column(wb, "type", "string", "group_key", false, NULL);
1375	streaming_topology_v1_emit_column(wb, "layer", "string", "group_key", false, NULL);
1376	streaming_topology_v1_emit_column(wb, "machine_guid", "string", "merge_identity", true, NULL);
1377	streaming_topology_v1_emit_column(wb, "node_id", "string", "merge_identity", true, NULL);
1378	streaming_topology_v1_emit_column(wb, "hostname", "string", "attribute", true, NULL);
1379	streaming_topology_v1_emit_column(wb, "display_name", "string", "attribute", true, NULL);
1380	streaming_topology_v1_emit_column(wb, "severity", "string", "attribute", true, NULL);
1381	streaming_topology_v1_emit_column(wb, "ephemerality", "string", "attribute", true, NULL);
1382	streaming_topology_v1_emit_column(wb, "ingest_status", "string", "attribute", true, NULL);
1383	streaming_topology_v1_emit_column(wb, "stream_status", "string", "attribute", true, NULL);
1384	streaming_topology_v1_emit_column(wb, "ml_status", "string", "attribute", true, NULL);
1385	streaming_topology_v1_emit_column(wb, "agent_name", "string", "attribute", true, NULL);
1386	streaming_topology_v1_emit_column(wb, "agent_version", "string", "attribute", true, NULL);
1387	streaming_topology_v1_emit_column(wb, "health_status", "string", "attribute", true, NULL);
1388	streaming_topology_v1_emit_column(wb, "os_name", "string", "attribute", true, NULL);
1389	streaming_topology_v1_emit_column(wb, "architecture", "string", "attribute", true, NULL);
1390	streaming_topology_v1_emit_column(wb, "cpu_count", "string", "attribute", true, NULL);
1391	streaming_topology_v1_emit_column(wb, "child_count", "uint", "metric", false, "sum");
1392	streaming_topology_v1_emit_column(wb, "retained_node_count", "uint", "metric", false, "max");
1393	streaming_topology_v1_emit_column(wb, "health_critical", "uint", "metric", false, "sum");
1394	streaming_topology_v1_emit_column(wb, "health_warning", "uint", "metric", false, "sum");
1395	streaming_topology_v1_emit_column(wb, "health_clear", "uint", "metric", false, "sum");
1396	}
1397
1398	static void streaming_topology_v1_emit_link_columns(BUFFER *wb) {
1399	streaming_topology_v1_emit_column(wb, "src_actor", "actor_ref", "reference", false, NULL);
1400	streaming_topology_v1_emit_column(wb, "dst_actor", "actor_ref", "reference", false, NULL);
1401	streaming_topology_v1_emit_column(wb, "type", "string", "group_key", false, NULL);
1402	streaming_topology_v1_emit_column(wb, "state", "string", "attribute", true, NULL);
1403	streaming_topology_v1_emit_column(wb, "port_name", "string", "attribute", true, NULL);
1404	streaming_topology_v1_emit_column(wb, "discovered_at", "timestamp", "timestamp", true, NULL);
1405	streaming_topology_v1_emit_column(wb, "last_seen", "timestamp", "timestamp", true, NULL);
1406	streaming_topology_v1_emit_column(wb, "hops", "int", "metric", false, "max");
1407	streaming_topology_v1_emit_column(wb, "evidence_count", "uint", "metric", false, "sum");
1408	streaming_topology_v1_emit_column(wb, "connections", "uint", "metric", false, "sum");
1409	streaming_topology_v1_emit_column(wb, "replication_instances", "uint", "metric", false, "sum");
1410	streaming_topology_v1_emit_column(wb, "replication_completion", "float", "metric", false, "avg");
1411	streaming_topology_v1_emit_column(wb, "collected_metrics", "uint", "metric", false, "sum");
1412	streaming_topology_v1_emit_column(wb, "collected_instances", "uint", "metric", false, "sum");
1413	streaming_topology_v1_emit_column(wb, "collected_contexts", "uint", "metric", false, "sum");
1414	}
1415
1416	static void streaming_topology_v1_emit_actor_label_columns(BUFFER *wb) {
1417	streaming_topology_v1_emit_column(wb, "actor", "actor_ref", "reference", false, NULL);
1418	streaming_topology_v1_emit_column(wb, "key", "string", "attribute", false, NULL);
1419	streaming_topology_v1_emit_column(wb, "value", "string", "attribute", false, NULL);
1420	streaming_topology_v1_emit_column(wb, "source", "string", "attribute", true, NULL);
1421	streaming_topology_v1_emit_column(wb, "kind", "string", "attribute", true, NULL);
1422	streaming_topology_v1_emit_column(wb, "value_index", "uint", "attribute", true, NULL);
1423	}
1424
1425	static void streaming_topology_v1_emit_evidence_columns(BUFFER *wb) {
1426	streaming_topology_v1_emit_column(wb, "link", "link_ref", "reference", false, NULL);
1427	streaming_topology_v1_emit_column(wb, "src_actor", "actor_ref", "reference", false, NULL);
1428	streaming_topology_v1_emit_column(wb, "dst_actor", "actor_ref", "reference", false, NULL);
1429	streaming_topology_v1_emit_column(wb, "type", "string", "group_key", false, NULL);
1430	streaming_topology_v1_emit_column(wb, "state", "string", "attribute", true, NULL);
1431	streaming_topology_v1_emit_column(wb, "port_name", "string", "attribute", true, NULL);
1432	streaming_topology_v1_emit_column(wb, "discovered_at", "timestamp", "timestamp", true, NULL);
1433	streaming_topology_v1_emit_column(wb, "last_seen", "timestamp", "timestamp", true, NULL);
1434	streaming_topology_v1_emit_column(wb, "hops", "int", "metric", false, "max");
1435	streaming_topology_v1_emit_column(wb, "connections", "uint", "metric", false, "sum");
1436	streaming_topology_v1_emit_column(wb, "replication_instances", "uint", "metric", false, "sum");
1437	streaming_topology_v1_emit_column(wb, "replication_completion", "float", "metric", false, "avg");
1438	streaming_topology_v1_emit_column(wb, "collected_metrics", "uint", "metric", false, "sum");
1439	streaming_topology_v1_emit_column(wb, "collected_instances", "uint", "metric", false, "sum");
1440	streaming_topology_v1_emit_column(wb, "collected_contexts", "uint", "metric", false, "sum");
1441	}
1442
1443	static void streaming_topology_v1_emit_stream_path_columns(BUFFER *wb) {
1444	streaming_topology_v1_emit_column(wb, "actor", "actor_ref", "reference", false, NULL);
1445	streaming_topology_v1_emit_column(wb, "path_actor", "actor_ref", "reference", false, NULL);
1446	streaming_topology_v1_emit_column(wb, "path_index", "uint", NULL, false, NULL);
1447	streaming_topology_v1_emit_column(wb, "hostname", "string", "attribute", true, NULL);
1448	streaming_topology_v1_emit_column(wb, "host_id", "string", "merge_identity", true, NULL);
1449	streaming_topology_v1_emit_column(wb, "node_id", "string", "merge_identity", true, NULL);
1450	streaming_topology_v1_emit_column(wb, "claim_id", "string", "attribute", true, NULL);
1451	streaming_topology_v1_emit_column(wb, "hops", "int", "metric", false, "max");
1452	streaming_topology_v1_emit_column(wb, "since", "timestamp", "timestamp", true, NULL);
1453	streaming_topology_v1_emit_column(wb, "first_time", "timestamp", "timestamp", true, NULL);
1454	streaming_topology_v1_emit_column(wb, "start_time_ms", "uint", "metric", false, "max");
1455	streaming_topology_v1_emit_column(wb, "shutdown_time_ms", "uint", "metric", false, "max");
1456	streaming_topology_v1_emit_column(wb, "capabilities", "uint", "attribute", false, NULL);
1457	streaming_topology_v1_emit_column(wb, "flags", "uint", "attribute", false, NULL);
1458	}
1459
1460	static void streaming_topology_v1_emit_retention_columns(BUFFER *wb) {
1461	streaming_topology_v1_emit_column(wb, "actor", "actor_ref", "reference", false, NULL);
1462	streaming_topology_v1_emit_column(wb, "observer_actor", "actor_ref", "reference", false, NULL);
1463	streaming_topology_v1_emit_column(wb, "db_status", "string", "attribute", true, NULL);
1464	streaming_topology_v1_emit_column(wb, "db_from", "timestamp", "timestamp", true, NULL);
1465	streaming_topology_v1_emit_column(wb, "db_to", "timestamp", "timestamp", true, NULL);
1466	streaming_topology_v1_emit_column(wb, "db_duration", "duration", "metric", false, "max");
1467	streaming_topology_v1_emit_column(wb, "db_metrics", "uint", "metric", false, "sum");
1468	streaming_topology_v1_emit_column(wb, "db_instances", "uint", "metric", false, "sum");
1469	streaming_topology_v1_emit_column(wb, "db_contexts", "uint", "metric", false, "sum");
1470	}
1471
1472	static void streaming_topology_v1_emit_inbound_columns(BUFFER *wb) {
1473	streaming_topology_v1_emit_column(wb, "parent_actor", "actor_ref", "reference", false, NULL);
1474	streaming_topology_v1_emit_column(wb, "child_actor", "actor_ref", "reference", false, NULL);
1475	streaming_topology_v1_emit_column(wb, "source_actor", "actor_ref", "reference", true, NULL);
1476	streaming_topology_v1_emit_column(wb, "received_type", "string", "attribute", true, NULL);
1477	streaming_topology_v1_emit_column(wb, "ingest_status", "string", "attribute", true, NULL);
1478	streaming_topology_v1_emit_column(wb, "hops", "int", "metric", false, "max");
1479	streaming_topology_v1_emit_column(wb, "collected_metrics", "uint", "metric", false, "sum");
1480	streaming_topology_v1_emit_column(wb, "collected_instances", "uint", "metric", false, "sum");
1481	streaming_topology_v1_emit_column(wb, "collected_contexts", "uint", "metric", false, "sum");
1482	streaming_topology_v1_emit_column(wb, "replication_completion", "float", "metric", false, "avg");
1483	streaming_topology_v1_emit_column(wb, "ingest_age", "duration", "metric", false, "max");
1484	streaming_topology_v1_emit_column(wb, "ssl", "string", "attribute", true, NULL);
1485	streaming_topology_v1_emit_column(wb, "alerts_critical", "uint", "metric", false, "sum");
1486	streaming_topology_v1_emit_column(wb, "alerts_warning", "uint", "metric", false, "sum");
1487	}
1488
1489	static void streaming_topology_v1_emit_outbound_columns(BUFFER *wb) {
1490	streaming_topology_v1_emit_column(wb, "sender_actor", "actor_ref", "reference", false, NULL);
1491	streaming_topology_v1_emit_column(wb, "node_actor", "actor_ref", "reference", false, NULL);
1492	streaming_topology_v1_emit_column(wb, "destination_actor", "actor_ref", "reference", true, NULL);
1493	streaming_topology_v1_emit_column(wb, "stream_status", "string", "attribute", true, NULL);
1494	streaming_topology_v1_emit_column(wb, "stream_age", "duration", "metric", false, "max");
1495	streaming_topology_v1_emit_column(wb, "hops", "int", "metric", false, "max");
1496	streaming_topology_v1_emit_column(wb, "ssl", "string", "attribute", true, NULL);
1497	streaming_topology_v1_emit_column(wb, "compression", "string", "attribute", true, NULL);
1498	streaming_topology_v1_emit_column(wb, "collected_metrics", "uint", "metric", false, "sum");
1499	streaming_topology_v1_emit_column(wb, "collected_instances", "uint", "metric", false, "sum");
1500	streaming_topology_v1_emit_column(wb, "collected_contexts", "uint", "metric", false, "sum");
1501	streaming_topology_v1_emit_column(wb, "replication_instances", "uint", "metric", false, "sum");
1502	streaming_topology_v1_emit_column(wb, "replication_completion", "float", "metric", false, "avg");
1503	}
1504
1505	static void streaming_topology_v1_emit_modal_direct_column(
1506	BUFFER *wb,
1507	const char *id,
1508	const char *label,
1509	const char *column,
1510	const char *cell) {
1511	buffer_json_add_array_item_object(wb);
1512	{
1513	buffer_json_member_add_string(wb, "id", id);
1514	buffer_json_member_add_string(wb, "label", label);
1515	buffer_json_member_add_object(wb, "projection");
1516	{
1517	buffer_json_member_add_string(wb, "kind", "direct");
1518	buffer_json_member_add_string(wb, "column", column);
1519	}
1520	buffer_json_object_close(wb);
1521	buffer_json_member_add_string(wb, "cell", cell);
1522	}
1523	buffer_json_object_close(wb);
1524	}
1525
1526	static void streaming_topology_v1_emit_modal_actor_ref_column(
1527	BUFFER *wb,
1528	const char *id,
1529	const char *label,
1530	const char *actor_column) {
1531	buffer_json_add_array_item_object(wb);
1532	{
1533	buffer_json_member_add_string(wb, "id", id);
1534	buffer_json_member_add_string(wb, "label", label);
1535	buffer_json_member_add_object(wb, "projection");
1536	{
1537	buffer_json_member_add_string(wb, "kind", "actor_ref_label");
1538	buffer_json_member_add_string(wb, "actor_column", actor_column);
1539	}
1540	buffer_json_object_close(wb);
1541	buffer_json_member_add_string(wb, "cell", "actor_link");
1542	}
1543	buffer_json_object_close(wb);
1544	}
1545
1546	static void streaming_topology_v1_emit_modal_label_lookup_column(
1547	BUFFER *wb,
1548	const char *id,
1549	const char *label,
1550	const char *actor_column,
1551	const char *label_key,
1552	const char *cell) {
1553	buffer_json_add_array_item_object(wb);
1554	{
1555	buffer_json_member_add_string(wb, "id", id);
1556	buffer_json_member_add_string(wb, "label", label);
1557	buffer_json_member_add_object(wb, "projection");
1558	{
1559	buffer_json_member_add_string(wb, "kind", "label_lookup");
1560	if(actor_column)
1561	buffer_json_member_add_string(wb, "actor_column", actor_column);
1562	buffer_json_member_add_string(wb, "label_key", label_key);
1563	}
1564	buffer_json_object_close(wb);
1565	buffer_json_member_add_string(wb, "cell", cell);
1566	}
1567	buffer_json_object_close(wb);
1568	}
1569
1570	static void streaming_topology_v1_emit_modal_actor_table_source(BUFFER wb, const char table) {
1571	buffer_json_member_add_object(wb, "source");
1572	{
1573	buffer_json_member_add_string(wb, "kind", "actor_table");
1574	buffer_json_member_add_string(wb, "table", table);
1575	}
1576	buffer_json_object_close(wb);
1577	}
1578
1579	static void streaming_topology_v1_emit_modal_actor_owner_filter(BUFFER wb, const char actor_column) {
1580	buffer_json_member_add_object(wb, "owner_filter");
1581	{
1582	buffer_json_member_add_string(wb, "mode", "actor_column");
1583	buffer_json_member_add_string(wb, "actor_column", actor_column);
1584	}
1585	buffer_json_object_close(wb);
1586	}
1587
1588	static void streaming_topology_v1_emit_modal_sort(BUFFER wb, const char column, const char *direction) {
1589	buffer_json_member_add_object(wb, "sort");
1590	{
1591	buffer_json_member_add_string(wb, "column", column);
1592	buffer_json_member_add_string(wb, "direction", direction);
1593	}
1594	buffer_json_object_close(wb);
1595	}
1596
1597	static void streaming_topology_v1_emit_modal_identification_field(BUFFER wb, const char key, const char *label) {
1598	buffer_json_add_array_item_object(wb);
1599	{
1600	buffer_json_member_add_string(wb, "key", key);
1601	buffer_json_member_add_string(wb, "label", label);
1602	buffer_json_member_add_uint64(wb, "max_values", 1);
1603	}
1604	buffer_json_object_close(wb);
1605	}
1606
1607	static void streaming_topology_v1_emit_host_identification_fields(BUFFER wb, const char actor_type) {
1608	streaming_topology_v1_emit_modal_identification_field(wb, "hostname", "Hostname");
1609	streaming_topology_v1_emit_modal_identification_field(wb, "type", "Node Type");
1610	streaming_topology_v1_emit_modal_identification_field(wb, "health_status", "Health");
1611	streaming_topology_v1_emit_modal_identification_field(wb, "stream_status", "Stream");
1612	streaming_topology_v1_emit_modal_identification_field(wb, "ingest_status", "Ingest");
1613	if(actor_type && strcmp(actor_type, "parent") == 0) {
1614	streaming_topology_v1_emit_modal_identification_field(wb, "retained_node_count", "Retained Nodes");
1615	streaming_topology_v1_emit_modal_identification_field(wb, "child_count", "Direct Children");
1616	}
1617	streaming_topology_v1_emit_modal_identification_field(wb, "os_name", "OS");
1618	streaming_topology_v1_emit_modal_identification_field(wb, "_os_version", "OS Version");
1619	streaming_topology_v1_emit_modal_identification_field(wb, "_kernel_version", "Kernel");
1620	streaming_topology_v1_emit_modal_identification_field(wb, "architecture", "Architecture");
1621	streaming_topology_v1_emit_modal_identification_field(wb, "_system_cpu_model", "CPU");
1622	streaming_topology_v1_emit_modal_identification_field(wb, "_system_cores", "Cores");
1623	streaming_topology_v1_emit_modal_identification_field(wb, "_system_ram_total", "RAM");
1624	streaming_topology_v1_emit_modal_identification_field(wb, "_virtualization", "Virtualization");
1625	streaming_topology_v1_emit_modal_identification_field(wb, "_container", "Container");
1626	streaming_topology_v1_emit_modal_identification_field(wb, "_cloud_provider_type", "Cloud");
1627	streaming_topology_v1_emit_modal_identification_field(wb, "_cloud_instance_type", "Instance");
1628	streaming_topology_v1_emit_modal_identification_field(wb, "_cloud_instance_region", "Region");
1629	streaming_topology_v1_emit_modal_identification_field(wb, "agent_version", "Agent");
1630	}
1631
1632	static void streaming_topology_v1_emit_vnode_identification_fields(BUFFER *wb) {
1633	streaming_topology_v1_emit_modal_identification_field(wb, "hostname", "Hostname");
1634	streaming_topology_v1_emit_modal_identification_field(wb, "type", "Node Type");
1635	streaming_topology_v1_emit_modal_identification_field(wb, "_vnode_type", "Vnode Type");
1636	streaming_topology_v1_emit_modal_identification_field(wb, "vendor", "Vendor");
1637	streaming_topology_v1_emit_modal_identification_field(wb, "model", "Model");
1638	streaming_topology_v1_emit_modal_identification_field(wb, "address", "Address");
1639	streaming_topology_v1_emit_modal_identification_field(wb, "location", "Location");
1640	streaming_topology_v1_emit_modal_identification_field(wb, "sys_object_id", "Sys Object ID");
1641	streaming_topology_v1_emit_modal_identification_field(wb, "lldp_loc_sys_name", "LLDP Name");
1642	streaming_topology_v1_emit_modal_identification_field(wb, "health_status", "Health");
1643	streaming_topology_v1_emit_modal_identification_field(wb, "stream_status", "Stream");
1644	streaming_topology_v1_emit_modal_identification_field(wb, "ingest_status", "Ingest");
1645	streaming_topology_v1_emit_modal_identification_field(wb, "agent_version", "Agent");
1646	}
1647
1648	static void streaming_topology_v1_emit_modal_label_identification(BUFFER wb, const char actor_type) {
1649	buffer_json_member_add_object(wb, "identification");
1650	{
1651	buffer_json_member_add_boolean(wb, "enabled", true);
1652	buffer_json_member_add_array(wb, "fields");
1653	{
1654	if(actor_type && strcmp(actor_type, "vnode") == 0)
1655	streaming_topology_v1_emit_vnode_identification_fields(wb);
1656	else
1657	streaming_topology_v1_emit_host_identification_fields(wb, actor_type);
1658	}
1659	buffer_json_array_close(wb);
1660	}
1661	buffer_json_object_close(wb);
1662	}
1663
1664	static void streaming_topology_v1_emit_actor_modal(BUFFER wb, const char actor_type) {
1665	buffer_json_member_add_object(wb, "modal");
1666	{
1667	buffer_json_member_add_boolean(wb, "enabled", true);
1668	buffer_json_member_add_object(wb, "labels");
1669	{
1670	buffer_json_member_add_boolean(wb, "enabled", true);
1671	buffer_json_member_add_string(wb, "table", "actor_labels");
1672	streaming_topology_v1_emit_modal_label_identification(wb, actor_type);
1673	}
1674	buffer_json_object_close(wb);
1675	buffer_json_member_add_object(wb, "mini_topology");
1676	{
1677	buffer_json_member_add_boolean(wb, "enabled", true);
1678	buffer_json_member_add_uint64(wb, "depth", 1);
1679	}
1680	buffer_json_object_close(wb);
1681	buffer_json_member_add_array(wb, "sections");
1682	{
1683	buffer_json_add_array_item_object(wb);
1684	{
1685	buffer_json_member_add_string(wb, "id", "stream_path");
1686	buffer_json_member_add_string(wb, "label", "Stream path");
1687	buffer_json_member_add_uint64(wb, "order", 1);
1688	streaming_topology_v1_emit_modal_actor_table_source(wb, "stream_path");
1689	streaming_topology_v1_emit_modal_actor_owner_filter(wb, "actor");
1690	buffer_json_member_add_array(wb, "columns");
1691	{
1692	streaming_topology_v1_emit_modal_direct_column(wb, "path_index", "Hop", "path_index", "number");
1693	streaming_topology_v1_emit_modal_actor_ref_column(wb, "path_actor", "Node", "path_actor");
1694	streaming_topology_v1_emit_modal_direct_column(wb, "hostname", "Hostname", "hostname", "text");
1695	streaming_topology_v1_emit_modal_direct_column(wb, "hops", "Hops", "hops", "number");
1696	streaming_topology_v1_emit_modal_direct_column(wb, "since", "Since", "since", "timestamp");
1697	streaming_topology_v1_emit_modal_direct_column(wb, "first_time", "First seen", "first_time", "timestamp");
1698	}
1699	buffer_json_array_close(wb);
1700	streaming_topology_v1_emit_modal_sort(wb, "path_index", "asc");
1701	buffer_json_member_add_string(wb, "empty_label", "No stream path rows");
1702	}
1703	buffer_json_object_close(wb);
1704
1705	buffer_json_add_array_item_object(wb);
1706	{
1707	buffer_json_member_add_string(wb, "id", "retained_nodes");
1708	buffer_json_member_add_string(wb, "label", "Retained nodes");
1709	buffer_json_member_add_uint64(wb, "order", 2);
1710	streaming_topology_v1_emit_modal_actor_table_source(wb, "retention");
1711	streaming_topology_v1_emit_modal_actor_owner_filter(wb, "observer_actor");
1712	buffer_json_member_add_array(wb, "columns");
1713	{
1714	streaming_topology_v1_emit_modal_actor_ref_column(wb, "actor", "Node", "actor");
1715	streaming_topology_v1_emit_modal_label_lookup_column(wb, "actor_type", "Node type", "actor", "type", "badge");
1716	streaming_topology_v1_emit_modal_direct_column(wb, "db_status", "Status", "db_status", "badge");
1717	streaming_topology_v1_emit_modal_direct_column(wb, "db_from", "From", "db_from", "timestamp");
1718	streaming_topology_v1_emit_modal_direct_column(wb, "db_to", "To", "db_to", "timestamp");
1719	streaming_topology_v1_emit_modal_direct_column(wb, "db_duration", "Duration", "db_duration", "duration");
1720	streaming_topology_v1_emit_modal_direct_column(wb, "db_metrics", "Metrics", "db_metrics", "number");
1721	streaming_topology_v1_emit_modal_direct_column(wb, "db_instances", "Instances", "db_instances", "number");
1722	streaming_topology_v1_emit_modal_direct_column(wb, "db_contexts", "Contexts", "db_contexts", "number");
1723	}
1724	buffer_json_array_close(wb);
1725	streaming_topology_v1_emit_modal_sort(wb, "db_duration", "desc");
1726	buffer_json_member_add_string(wb, "empty_label", "No retained nodes");
1727	}
1728	buffer_json_object_close(wb);
1729
1730	buffer_json_add_array_item_object(wb);
1731	{
1732	buffer_json_member_add_string(wb, "id", "inbound");
1733	buffer_json_member_add_string(wb, "label", "Received nodes");
1734	buffer_json_member_add_uint64(wb, "order", 3);
1735	streaming_topology_v1_emit_modal_actor_table_source(wb, "inbound");
1736	streaming_topology_v1_emit_modal_actor_owner_filter(wb, "parent_actor");
1737	buffer_json_member_add_array(wb, "columns");
1738	{
1739	streaming_topology_v1_emit_modal_actor_ref_column(wb, "child", "Node", "child_actor");
1740	streaming_topology_v1_emit_modal_actor_ref_column(wb, "source", "Received from", "source_actor");
1741	streaming_topology_v1_emit_modal_label_lookup_column(wb, "child_type", "Node type", "child_actor", "type", "badge");
1742	streaming_topology_v1_emit_modal_direct_column(wb, "received_type", "Received as", "received_type", "badge");
1743	streaming_topology_v1_emit_modal_direct_column(wb, "ingest_status", "Ingest", "ingest_status", "badge");
1744	streaming_topology_v1_emit_modal_direct_column(wb, "hops", "Hops", "hops", "number");
1745	streaming_topology_v1_emit_modal_direct_column(wb, "collected_metrics", "Metrics", "collected_metrics", "number");
1746	streaming_topology_v1_emit_modal_direct_column(wb, "collected_instances", "Instances", "collected_instances", "number");
1747	streaming_topology_v1_emit_modal_direct_column(wb, "collected_contexts", "Contexts", "collected_contexts", "number");
1748	streaming_topology_v1_emit_modal_direct_column(wb, "replication_completion", "Replication %", "replication_completion", "number");
1749	streaming_topology_v1_emit_modal_direct_column(wb, "ingest_age", "Age", "ingest_age", "duration");
1750	streaming_topology_v1_emit_modal_direct_column(wb, "ssl", "TLS", "ssl", "badge");
1751	streaming_topology_v1_emit_modal_direct_column(wb, "alerts_critical", "Critical", "alerts_critical", "number");
1752	streaming_topology_v1_emit_modal_direct_column(wb, "alerts_warning", "Warning", "alerts_warning", "number");
1753	}
1754	buffer_json_array_close(wb);
1755	buffer_json_member_add_string(wb, "empty_label", "No received nodes");
1756	}
1757	buffer_json_object_close(wb);
1758
1759	buffer_json_add_array_item_object(wb);
1760	{
1761	buffer_json_member_add_string(wb, "id", "outbound");
1762	buffer_json_member_add_string(wb, "label", "Outbound streams");
1763	buffer_json_member_add_uint64(wb, "order", 4);
1764	streaming_topology_v1_emit_modal_actor_table_source(wb, "outbound");
1765	streaming_topology_v1_emit_modal_actor_owner_filter(wb, "sender_actor");
1766	buffer_json_member_add_array(wb, "columns");
1767	{
1768	streaming_topology_v1_emit_modal_actor_ref_column(wb, "node", "Node", "node_actor");
1769	streaming_topology_v1_emit_modal_label_lookup_column(wb, "node_type", "Node type", "node_actor", "type", "badge");
1770	streaming_topology_v1_emit_modal_actor_ref_column(wb, "destination", "Destination", "destination_actor");
1771	streaming_topology_v1_emit_modal_direct_column(wb, "stream_status", "Status", "stream_status", "badge");
1772	streaming_topology_v1_emit_modal_direct_column(wb, "stream_age", "Age", "stream_age", "duration");
1773	streaming_topology_v1_emit_modal_direct_column(wb, "hops", "Hops", "hops", "number");
1774	streaming_topology_v1_emit_modal_direct_column(wb, "collected_metrics", "Metrics", "collected_metrics", "number");
1775	streaming_topology_v1_emit_modal_direct_column(wb, "collected_instances", "Instances", "collected_instances", "number");
1776	streaming_topology_v1_emit_modal_direct_column(wb, "collected_contexts", "Contexts", "collected_contexts", "number");
1777	streaming_topology_v1_emit_modal_direct_column(wb, "replication_completion", "Replication %", "replication_completion", "number");
1778	streaming_topology_v1_emit_modal_direct_column(wb, "ssl", "TLS", "ssl", "badge");
1779	streaming_topology_v1_emit_modal_direct_column(wb, "compression", "Compression", "compression", "badge");
1780	}
1781	buffer_json_array_close(wb);
1782	buffer_json_member_add_string(wb, "empty_label", "No outbound streams");
1783	}
1784	buffer_json_object_close(wb);
1785	}
1786	buffer_json_array_close(wb);
1787	}
1788	buffer_json_object_close(wb);
1789	}
1790
1791	static void streaming_topology_v1_emit_actor_type(
1792	BUFFER *wb,
1793	const char *id,
1794	const char *label,
1795	const char *color_slot,
1796	const char *icon,
1797	bool border,
1798	const char *size_mode,
1799	const char *size_metric_column,
1800	const char *size_scale,
1801	const char *layout_repulsion,
1802	bool show_port_bullets,
1803	const char *port_actor_column) {
1804	buffer_json_member_add_object(wb, id);
1805	{
1806	buffer_json_member_add_string(wb, "layer", "streaming");
1807	buffer_json_member_add_array(wb, "identity");
1808	buffer_json_add_array_item_string(wb, "id");
1809	buffer_json_array_close(wb);
1810	buffer_json_member_add_array(wb, "merge_identity");
1811	buffer_json_add_array_item_string(wb, "machine_guid");
1812	buffer_json_add_array_item_string(wb, "node_id");
1813	buffer_json_array_close(wb);
1814	buffer_json_member_add_array(wb, "aggregation_scopes");
1815	buffer_json_add_array_item_string(wb, "node");
1816	buffer_json_array_close(wb);
1817	buffer_json_member_add_object(wb, "search");
1818	{
1819	if(strcmp(id, "stale") == 0)
1820	buffer_json_member_add_boolean(wb, "enabled", true);
1821	buffer_json_member_add_array(wb, "columns");
1822	buffer_json_add_array_item_string(wb, "display_name");
1823	buffer_json_add_array_item_string(wb, "hostname");
1824	buffer_json_add_array_item_string(wb, "machine_guid");
1825	buffer_json_add_array_item_string(wb, "node_id");
1826	buffer_json_add_array_item_string(wb, "agent_version");
1827	buffer_json_array_close(wb);
1828	}
1829	buffer_json_object_close(wb);
1830	buffer_json_member_add_object(wb, "presentation");
1831	{
1832	buffer_json_member_add_string(wb, "label", label);
1833	buffer_json_member_add_string(wb, "role", "actor");
1834	buffer_json_member_add_string(wb, "icon", icon);
1835	buffer_json_member_add_string(wb, "color_slot", color_slot);
1836	buffer_json_member_add_string(wb, "opacity", strcmp(id, "stale") == 0 ? "faded" : "normal");
1837	buffer_json_member_add_object(wb, "border");
1838	{
1839	buffer_json_member_add_boolean(wb, "enabled", border);
1840	}
1841	buffer_json_object_close(wb);
1842	buffer_json_member_add_object(wb, "size");
1843	{
1844	buffer_json_member_add_string(wb, "mode", size_mode ? size_mode : "fixed");
1845	if(size_metric_column)
1846	buffer_json_member_add_string(wb, "metric_column", size_metric_column);
1847	if(size_scale)
1848	buffer_json_member_add_string(wb, "scale", size_scale);
1849	}
1850	buffer_json_object_close(wb);
1851	if(layout_repulsion) {
1852	buffer_json_member_add_object(wb, "layout");
1853	{
1854	buffer_json_member_add_string(wb, "repulsion", layout_repulsion);
1855	}
1856	buffer_json_object_close(wb);
1857	}
1858	buffer_json_member_add_object(wb, "label_policy");
1859	{
1860	buffer_json_member_add_array(wb, "columns");
1861	buffer_json_add_array_item_string(wb, "display_name");
1862	buffer_json_add_array_item_string(wb, "hostname");
1863	buffer_json_array_close(wb);
1864	buffer_json_member_add_string(wb, "fallback", "type_label");
1865	buffer_json_member_add_uint64(wb, "max_length", 80);
1866	buffer_json_member_add_string(wb, "array", "reject");
1867	}
1868	buffer_json_object_close(wb);
1869	buffer_json_member_add_object(wb, "ports");
1870	{
1871	buffer_json_member_add_boolean(wb, "show_bullets", show_port_bullets);
1872	if(show_port_bullets) {
1873	buffer_json_member_add_array(wb, "sources");
1874	{
1875	buffer_json_add_array_item_object(wb);
1876	buffer_json_member_add_string(wb, "source", "links");
1877	buffer_json_member_add_string(
1878	wb, "actor_column", port_actor_column ? port_actor_column : "src_actor");
1879	buffer_json_member_add_string(wb, "name_column", "port_name");
1880	buffer_json_member_add_string(wb, "type_column", "type");
1881	buffer_json_member_add_string(wb, "default_type", "streaming");
1882	buffer_json_object_close(wb);
1883	}
1884	buffer_json_array_close(wb);
1885	}
1886	}
1887	buffer_json_object_close(wb);
1888	streaming_topology_v1_emit_actor_modal(wb, id);
1889	}
1890	buffer_json_object_close(wb);
1891	}
1892	buffer_json_object_close(wb);
1893	}
1894
1895	static void streaming_topology_v1_emit_link_type(
1896	BUFFER *wb,
1897	const char *id,
1898	const char *direction_role,
1899	const char *semantic_role,
1900	const char *evidence_type,
1901	const char *label,
1902	const char *color_slot,
1903	const char *line_style,
1904	const char *width,
1905	const char *opacity) {
1906	buffer_json_member_add_object(wb, id);
1907	{
1908	buffer_json_member_add_string(wb, "orientation", "directed");
1909	buffer_json_member_add_string(wb, "direction_role", direction_role);
1910	if(semantic_role)
1911	buffer_json_member_add_string(wb, "semantic_role", semantic_role);
1912	buffer_json_member_add_object(wb, "aggregation");
1913	{
1914	buffer_json_member_add_string(wb, "direction", "preserve");
1915	buffer_json_member_add_string(wb, "evidence", "append");
1916	buffer_json_member_add_object(wb, "metrics");
1917	{
1918	buffer_json_member_add_string(wb, "hops", "max");
1919	buffer_json_member_add_string(wb, "evidence_count", "sum");
1920	buffer_json_member_add_string(wb, "connections", "sum");
1921	buffer_json_member_add_string(wb, "replication_instances", "sum");
1922	buffer_json_member_add_string(wb, "replication_completion", "avg");
1923	buffer_json_member_add_string(wb, "collected_metrics", "sum");
1924	buffer_json_member_add_string(wb, "collected_instances", "sum");
1925	buffer_json_member_add_string(wb, "collected_contexts", "sum");
1926	}
1927	buffer_json_object_close(wb);
1928	}
1929	buffer_json_object_close(wb);
1930	buffer_json_member_add_array(wb, "evidence_types");
1931	buffer_json_add_array_item_string(wb, evidence_type);
1932	buffer_json_array_close(wb);
1933	buffer_json_member_add_object(wb, "presentation");
1934	{
1935	buffer_json_member_add_string(wb, "label", label);
1936	buffer_json_member_add_string(wb, "color_slot", color_slot);
1937	buffer_json_member_add_string(wb, "line_style", line_style);
1938	buffer_json_member_add_string(wb, "width", width);
1939	buffer_json_member_add_string(wb, "opacity", opacity);
1940	buffer_json_member_add_string(wb, "curve", "auto");
1941	buffer_json_member_add_string(wb, "arrow", "forward");
1942	}
1943	buffer_json_object_close(wb);
1944	}
1945	buffer_json_object_close(wb);
1946	}
1947
1948	static void streaming_topology_v1_emit_evidence_type(BUFFER wb, const char id, const char *link_type) {
1949	buffer_json_member_add_object(wb, id);
1950	{
1951	buffer_json_member_add_string(wb, "link_type", link_type);
1952	buffer_json_member_add_string(wb, "role", "relationship_evidence");
1953	buffer_json_member_add_array(wb, "columns");
1954	streaming_topology_v1_emit_evidence_columns(wb);
1955	buffer_json_array_close(wb);
1956	buffer_json_member_add_array(wb, "match_columns");
1957	buffer_json_add_array_item_string(wb, "src_actor");
1958	buffer_json_add_array_item_string(wb, "dst_actor");
1959	buffer_json_add_array_item_string(wb, "type");
1960	buffer_json_array_close(wb);
1961	}
1962	buffer_json_object_close(wb);
1963	}
1964
1965	static void streaming_topology_v1_emit_table_type(
1966	BUFFER *wb,
1967	const char *id,
1968	const char *role,
1969	const char *owner,
1970	const char *aggregation,
1971	void (emit_columns)(BUFFER )) {
1972	buffer_json_member_add_object(wb, id);
1973	{
1974	buffer_json_member_add_string(wb, "role", role);
1975	buffer_json_member_add_string(wb, "owner", owner);
1976	buffer_json_member_add_string(wb, "aggregation", aggregation);
1977	buffer_json_member_add_array(wb, "columns");
1978	emit_columns(wb);
1979	buffer_json_array_close(wb);
1980	}
1981	buffer_json_object_close(wb);
1982	}
1983
1984	static void streaming_topology_v1_emit_type_registry(BUFFER *wb) {
1985	buffer_json_member_add_object(wb, "types");
1986	{
1987	buffer_json_member_add_object(wb, "actor_types");
1988	{
1989	streaming_topology_v1_emit_actor_type(
1990	wb, "parent", "Netdata Parent", "primary", "parent", true,
1991	"metric", "retained_node_count", "emphasized", "stronger", true, "dst_actor");
1992	streaming_topology_v1_emit_actor_type(
1993	wb, "child", "Netdata Child", "primary", "netdata-agent", false,
1994	"fixed", NULL, "normal", "normal", false, NULL);
1995	streaming_topology_v1_emit_actor_type(
1996	wb, "vnode", "Virtual Node", "warning", "netdata-agent", false,
1997	"fixed", NULL, "normal", "normal", false, NULL);
1998	streaming_topology_v1_emit_actor_type(
1999	wb, "stale", "Stale Node", "dim", "netdata-agent", false,
2000	"fixed", NULL, "compact", "weaker", false, NULL);
2001	}
2002	buffer_json_object_close(wb);
2003
2004	buffer_json_member_add_object(wb, "link_types");
2005	{
2006	streaming_topology_v1_emit_link_type(
2007	wb, "streaming", "dependency", "traffic", "streaming_link",
2008	"Streaming", "primary", "solid", "thick", "normal");
2009	streaming_topology_v1_emit_link_type(
2010	wb, "virtual", "dependency", "ownership", "virtual_link",
2011	"Virtual origin", "warning", "dashed", "thin", "muted");
2012	streaming_topology_v1_emit_link_type(
2013	wb, "stale", "observation", "normal", "stale_link",
2014	"Stale data", "dim", "dashed", "thin", "faded");
2015	}
2016	buffer_json_object_close(wb);
2017
2018	buffer_json_member_add_object(wb, "port_types");
2019	{
2020	buffer_json_member_add_object(wb, "streaming");
2021	{
2022	buffer_json_member_add_object(wb, "presentation");
2023	{
2024	buffer_json_member_add_string(wb, "label", "Streaming child");
2025	buffer_json_member_add_string(wb, "color_slot", "primary");
2026	buffer_json_member_add_string(wb, "opacity", "normal");
2027	}
2028	buffer_json_object_close(wb);
2029	}
2030	buffer_json_object_close(wb);
2031
2032	buffer_json_member_add_object(wb, "virtual");
2033	{
2034	buffer_json_member_add_object(wb, "presentation");
2035	{
2036	buffer_json_member_add_string(wb, "label", "Virtual node");
2037	buffer_json_member_add_string(wb, "color_slot", "warning");
2038	buffer_json_member_add_string(wb, "opacity", "normal");
2039	}
2040	buffer_json_object_close(wb);
2041	}
2042	buffer_json_object_close(wb);
2043
2044	buffer_json_member_add_object(wb, "stale");
2045	{
2046	buffer_json_member_add_object(wb, "presentation");
2047	{
2048	buffer_json_member_add_string(wb, "label", "Stale node");
2049	buffer_json_member_add_string(wb, "color_slot", "dim");
2050	buffer_json_member_add_string(wb, "opacity", "faded");
2051	}
2052	buffer_json_object_close(wb);
2053	}
2054	buffer_json_object_close(wb);
2055	}
2056	buffer_json_object_close(wb);
2057
2058	buffer_json_member_add_object(wb, "evidence_types");
2059	{
2060	streaming_topology_v1_emit_evidence_type(wb, "streaming_link", "streaming");
2061	streaming_topology_v1_emit_evidence_type(wb, "virtual_link", "virtual");
2062	streaming_topology_v1_emit_evidence_type(wb, "stale_link", "stale");
2063	}
2064	buffer_json_object_close(wb);
2065
2066	buffer_json_member_add_object(wb, "table_types");
2067	{
2068	streaming_topology_v1_emit_table_type(wb, "actor_labels", "actor_inventory", "actor", "set",
2069	streaming_topology_v1_emit_actor_label_columns);
2070	streaming_topology_v1_emit_table_type(wb, "stream_path", "actor_detail", "actor", "append",
2071	streaming_topology_v1_emit_stream_path_columns);
2072	streaming_topology_v1_emit_table_type(wb, "retention", "actor_detail", "actor", "append",
2073	streaming_topology_v1_emit_retention_columns);
2074	streaming_topology_v1_emit_table_type(wb, "inbound", "relationship_summary", "actor", "append",
2075	streaming_topology_v1_emit_inbound_columns);
2076	streaming_topology_v1_emit_table_type(wb, "outbound", "relationship_summary", "actor", "append",
2077	streaming_topology_v1_emit_outbound_columns);
2078	}
2079	buffer_json_object_close(wb);
2080
2081	buffer_json_member_add_object(wb, "aggregation_scopes");
2082	{
2083	buffer_json_member_add_object(wb, "node");
2084	{
2085	buffer_json_member_add_array(wb, "columns");
2086	buffer_json_add_array_item_string(wb, "machine_guid");
2087	buffer_json_add_array_item_string(wb, "node_id");
2088	buffer_json_array_close(wb);
2089	buffer_json_member_add_string(wb, "evidence_policy", "preserve");
2090	}
2091	buffer_json_object_close(wb);
2092	}
2093	buffer_json_object_close(wb);
2094	}
2095	buffer_json_object_close(wb);
2096	}
2097
2098	static void streaming_topology_v1_emit_presentation(BUFFER *wb) {
2099	buffer_json_member_add_object(wb, "presentation");
2100	{
2101	buffer_json_member_add_string(wb, "profile_version", "streaming.v1");
2102	buffer_json_member_add_object(wb, "selection");
2103	{
2104	buffer_json_member_add_object(wb, "actor_click");
2105	{
2106	buffer_json_member_add_string(wb, "mode", "highlight_path");
2107	buffer_json_member_add_string(wb, "path_table", "stream_path");
2108	buffer_json_member_add_string(wb, "path_owner_column", "actor");
2109	buffer_json_member_add_string(wb, "path_actor_column", "path_actor");
2110	buffer_json_member_add_string(wb, "path_order_column", "path_index");
2111	}
2112	buffer_json_object_close(wb);
2113	}
2114	buffer_json_object_close(wb);
2115
2116	buffer_json_member_add_object(wb, "legend");
2117	{
2118	buffer_json_member_add_array(wb, "actors");
2119	{
2120	buffer_json_add_array_item_object(wb);
2121	buffer_json_member_add_string(wb, "type", "parent");
2122	buffer_json_member_add_string(wb, "label", "Parent");
2123	buffer_json_object_close(wb);
2124
2125	buffer_json_add_array_item_object(wb);
2126	buffer_json_member_add_string(wb, "type", "child");
2127	buffer_json_member_add_string(wb, "label", "Child");
2128	buffer_json_object_close(wb);
2129
2130	buffer_json_add_array_item_object(wb);
2131	buffer_json_member_add_string(wb, "type", "vnode");
2132	buffer_json_member_add_string(wb, "label", "Virtual Node");
2133	buffer_json_object_close(wb);
2134
2135	buffer_json_add_array_item_object(wb);
2136	buffer_json_member_add_string(wb, "type", "stale");
2137	buffer_json_member_add_string(wb, "label", "Stale Node");
2138	buffer_json_object_close(wb);
2139	}
2140	buffer_json_array_close(wb);
2141
2142	buffer_json_member_add_array(wb, "links");
2143	{
2144	buffer_json_add_array_item_object(wb);
2145	buffer_json_member_add_string(wb, "type", "streaming");
2146	buffer_json_member_add_string(wb, "label", "Streaming");
2147	buffer_json_object_close(wb);
2148
2149	buffer_json_add_array_item_object(wb);
2150	buffer_json_member_add_string(wb, "type", "virtual");
2151	buffer_json_member_add_string(wb, "label", "Virtual origin");
2152	buffer_json_object_close(wb);
2153
2154	buffer_json_add_array_item_object(wb);
2155	buffer_json_member_add_string(wb, "type", "stale");
2156	buffer_json_member_add_string(wb, "label", "Stale data");
2157	buffer_json_object_close(wb);
2158	}
2159	buffer_json_array_close(wb);
2160
2161	buffer_json_member_add_array(wb, "ports");
2162	{
2163	buffer_json_add_array_item_object(wb);
2164	buffer_json_member_add_string(wb, "type", "streaming");
2165	buffer_json_member_add_string(wb, "label", "Streaming child");
2166	buffer_json_object_close(wb);
2167
2168	buffer_json_add_array_item_object(wb);
2169	buffer_json_member_add_string(wb, "type", "virtual");
2170	buffer_json_member_add_string(wb, "label", "Virtual node");
2171	buffer_json_object_close(wb);
2172
2173	buffer_json_add_array_item_object(wb);
2174	buffer_json_member_add_string(wb, "type", "stale");
2175	buffer_json_member_add_string(wb, "label", "Stale node");
2176	buffer_json_object_close(wb);
2177	}
2178	buffer_json_array_close(wb);
2179	}
2180	buffer_json_object_close(wb);
2181
2182	buffer_json_member_add_array(wb, "port_fields");
2183	{
2184	buffer_json_add_array_item_object(wb);
2185	buffer_json_member_add_string(wb, "key", "type");
2186	buffer_json_member_add_string(wb, "label", "Type");
2187	buffer_json_object_close(wb);
2188	}
2189	buffer_json_array_close(wb);
2190	}
2191	buffer_json_object_close(wb);
2192	}
2193
2194	static void streaming_topology_v1_emit_actor_table(BUFFER wb, STREAMING_TOPOLOGY_V1_PAYLOAD payload) {
2195	buffer_json_member_add_object(wb, "actors");
2196	{
2197	buffer_json_member_add_uint64(wb, "rows", payload->actors_used);
2198	buffer_json_member_add_array(wb, "columns");
2199	streaming_topology_v1_emit_actor_columns(wb);
2200	buffer_json_array_close(wb);
2201
2202	buffer_json_member_add_array(wb, "values");
2203	#define STREAMING_TOPOLOGY_ACTOR_STRING_VALUES(member) do { \
2204	streaming_topology_v1_emit_values_start(wb); \
2205	for(size_t i = 0; i < payload->actors_used; i++) \
2206	buffer_json_add_array_item_string(wb, payload->actors[i].member[0] ? payload->actors[i].member : NULL); \
2207	streaming_topology_v1_emit_values_end(wb); \
2208	} while(0)
2209
2210	STREAMING_TOPOLOGY_ACTOR_STRING_VALUES(actor_id);
2211	STREAMING_TOPOLOGY_ACTOR_STRING_VALUES(type);
2212	streaming_topology_v1_emit_values_start(wb);
2213	for(size_t i = 0; i < payload->actors_used; i++)
2214	buffer_json_add_array_item_string(wb, "streaming");
2215	streaming_topology_v1_emit_values_end(wb);
2216	STREAMING_TOPOLOGY_ACTOR_STRING_VALUES(machine_guid);
2217	STREAMING_TOPOLOGY_ACTOR_STRING_VALUES(node_id);
2218	STREAMING_TOPOLOGY_ACTOR_STRING_VALUES(hostname);
2219	STREAMING_TOPOLOGY_ACTOR_STRING_VALUES(display_name);
2220	STREAMING_TOPOLOGY_ACTOR_STRING_VALUES(severity);
2221	STREAMING_TOPOLOGY_ACTOR_STRING_VALUES(ephemerality);
2222	STREAMING_TOPOLOGY_ACTOR_STRING_VALUES(ingest_status);
2223	STREAMING_TOPOLOGY_ACTOR_STRING_VALUES(stream_status);
2224	STREAMING_TOPOLOGY_ACTOR_STRING_VALUES(ml_status);
2225	STREAMING_TOPOLOGY_ACTOR_STRING_VALUES(agent_name);
2226	STREAMING_TOPOLOGY_ACTOR_STRING_VALUES(agent_version);
2227	STREAMING_TOPOLOGY_ACTOR_STRING_VALUES(health_status);
2228	STREAMING_TOPOLOGY_ACTOR_STRING_VALUES(os_name);
2229	STREAMING_TOPOLOGY_ACTOR_STRING_VALUES(architecture);
2230	STREAMING_TOPOLOGY_ACTOR_STRING_VALUES(cpu_count);
2231
2232	#undef STREAMING_TOPOLOGY_ACTOR_STRING_VALUES
2233	#define STREAMING_TOPOLOGY_ACTOR_UINT_VALUES(member) do { \
2234	streaming_topology_v1_emit_values_start(wb); \
2235	for(size_t i = 0; i < payload->actors_used; i++) \
2236	buffer_json_add_array_item_uint64(wb, payload->actors[i].member); \
2237	streaming_topology_v1_emit_values_end(wb); \
2238	} while(0)
2239
2240	STREAMING_TOPOLOGY_ACTOR_UINT_VALUES(child_count);
2241	STREAMING_TOPOLOGY_ACTOR_UINT_VALUES(retained_node_count);
2242	STREAMING_TOPOLOGY_ACTOR_UINT_VALUES(health_critical);
2243	STREAMING_TOPOLOGY_ACTOR_UINT_VALUES(health_warning);
2244	STREAMING_TOPOLOGY_ACTOR_UINT_VALUES(health_clear);
2245	#undef STREAMING_TOPOLOGY_ACTOR_UINT_VALUES
2246
2247	buffer_json_array_close(wb);
2248	}
2249	buffer_json_object_close(wb);
2250	}
2251
2252	static void streaming_topology_v1_emit_link_table(BUFFER wb, STREAMING_TOPOLOGY_V1_PAYLOAD payload) {
2253	buffer_json_member_add_object(wb, "links");
2254	{
2255	buffer_json_member_add_uint64(wb, "rows", payload->links_used);
2256	buffer_json_member_add_array(wb, "columns");
2257	streaming_topology_v1_emit_link_columns(wb);
2258	buffer_json_array_close(wb);
2259	buffer_json_member_add_array(wb, "values");
2260
2261	streaming_topology_v1_emit_values_start(wb);
2262	for(size_t i = 0; i < payload->links_used; i++)
2263	buffer_json_add_array_item_uint64(wb, payload->links[i].src_actor);
2264	streaming_topology_v1_emit_values_end(wb);
2265
2266	streaming_topology_v1_emit_values_start(wb);
2267	for(size_t i = 0; i < payload->links_used; i++)
2268	buffer_json_add_array_item_uint64(wb, payload->links[i].dst_actor);
2269	streaming_topology_v1_emit_values_end(wb);
2270
2271	streaming_topology_v1_emit_values_start(wb);
2272	for(size_t i = 0; i < payload->links_used; i++)
2273	buffer_json_add_array_item_string(wb, payload->links[i].type);
2274	streaming_topology_v1_emit_values_end(wb);
2275
2276	streaming_topology_v1_emit_values_start(wb);
2277	for(size_t i = 0; i < payload->links_used; i++)
2278	buffer_json_add_array_item_string(wb, payload->links[i].state[0] ? payload->links[i].state : NULL);
2279	streaming_topology_v1_emit_values_end(wb);
2280
2281	streaming_topology_v1_emit_values_start(wb);
2282	for(size_t i = 0; i < payload->links_used; i++)
2283	buffer_json_add_array_item_string(wb, payload->links[i].port_name[0] ? payload->links[i].port_name : NULL);
2284	streaming_topology_v1_emit_values_end(wb);
2285
2286	streaming_topology_v1_emit_values_start(wb);
2287	for(size_t i = 0; i < payload->links_used; i++)
2288	streaming_topology_v1_add_timestamp(wb, payload->links[i].discovered_at_ut);
2289	streaming_topology_v1_emit_values_end(wb);
2290
2291	streaming_topology_v1_emit_values_start(wb);
2292	for(size_t i = 0; i < payload->links_used; i++)
2293	streaming_topology_v1_add_timestamp(wb, payload->links[i].last_seen_ut);
2294	streaming_topology_v1_emit_values_end(wb);
2295
2296	streaming_topology_v1_emit_values_start(wb);
2297	for(size_t i = 0; i < payload->links_used; i++)
2298	buffer_json_add_array_item_int64(wb, payload->links[i].hops);
2299	streaming_topology_v1_emit_values_end(wb);
2300
2301	streaming_topology_v1_emit_values_start(wb);
2302	for(size_t i = 0; i < payload->links_used; i++)
2303	buffer_json_add_array_item_uint64(wb, 1);
2304	streaming_topology_v1_emit_values_end(wb);
2305
2306	streaming_topology_v1_emit_values_start(wb);
2307	for(size_t i = 0; i < payload->links_used; i++)
2308	buffer_json_add_array_item_uint64(wb, payload->links[i].connections);
2309	streaming_topology_v1_emit_values_end(wb);
2310
2311	streaming_topology_v1_emit_values_start(wb);
2312	for(size_t i = 0; i < payload->links_used; i++)
2313	buffer_json_add_array_item_uint64(wb, payload->links[i].replication_instances);
2314	streaming_topology_v1_emit_values_end(wb);
2315
2316	streaming_topology_v1_emit_values_start(wb);
2317	for(size_t i = 0; i < payload->links_used; i++)
2318	buffer_json_add_array_item_double(wb, payload->links[i].replication_completion);
2319	streaming_topology_v1_emit_values_end(wb);
2320
2321	streaming_topology_v1_emit_values_start(wb);
2322	for(size_t i = 0; i < payload->links_used; i++)
2323	buffer_json_add_array_item_uint64(wb, payload->links[i].collected_metrics);
2324	streaming_topology_v1_emit_values_end(wb);
2325
2326	streaming_topology_v1_emit_values_start(wb);
2327	for(size_t i = 0; i < payload->links_used; i++)
2328	buffer_json_add_array_item_uint64(wb, payload->links[i].collected_instances);
2329	streaming_topology_v1_emit_values_end(wb);
2330
2331	streaming_topology_v1_emit_values_start(wb);
2332	for(size_t i = 0; i < payload->links_used; i++)
2333	buffer_json_add_array_item_uint64(wb, payload->links[i].collected_contexts);
2334	streaming_topology_v1_emit_values_end(wb);
2335
2336	buffer_json_array_close(wb);
2337	}
2338	buffer_json_object_close(wb);
2339	}
2340
2341	static bool streaming_topology_v1_link_is_type(STREAMING_TOPOLOGY_V1_LINK link, const char link_type) {
2342	return link && link_type && strcmp(link->type, link_type) == 0;
2343	}
2344
2345	static size_t streaming_topology_v1_count_links_by_type(STREAMING_TOPOLOGY_V1_PAYLOAD payload, const char link_type) {
2346	size_t count = 0;
2347	for(size_t i = 0; i < payload->links_used; i++) {
2348	if(streaming_topology_v1_link_is_type(&payload->links[i], link_type))
2349	count++;
2350	}
2351	return count;
2352	}
2353
2354	static void streaming_topology_v1_emit_evidence_section(
2355	BUFFER *wb,
2356	STREAMING_TOPOLOGY_V1_PAYLOAD *payload,
2357	const char *evidence_type,
2358	const char *link_type) {
2359	buffer_json_member_add_object(wb, evidence_type);
2360	{
2361	buffer_json_member_add_string(wb, "type", evidence_type);
2362	buffer_json_member_add_object(wb, "table");
2363	{
2364	buffer_json_member_add_uint64(wb, "rows", streaming_topology_v1_count_links_by_type(payload, link_type));
2365	buffer_json_member_add_array(wb, "columns");
2366	streaming_topology_v1_emit_evidence_columns(wb);
2367	buffer_json_array_close(wb);
2368	buffer_json_member_add_array(wb, "values");
2369
2370	streaming_topology_v1_emit_values_start(wb);
2371	for(size_t i = 0; i < payload->links_used; i++) {
2372	if(streaming_topology_v1_link_is_type(&payload->links[i], link_type))
2373	buffer_json_add_array_item_uint64(wb, i);
2374	}
2375	streaming_topology_v1_emit_values_end(wb);
2376
2377	streaming_topology_v1_emit_values_start(wb);
2378	for(size_t i = 0; i < payload->links_used; i++) {
2379	if(streaming_topology_v1_link_is_type(&payload->links[i], link_type))
2380	buffer_json_add_array_item_uint64(wb, payload->links[i].src_actor);
2381	}
2382	streaming_topology_v1_emit_values_end(wb);
2383
2384	streaming_topology_v1_emit_values_start(wb);
2385	for(size_t i = 0; i < payload->links_used; i++) {
2386	if(streaming_topology_v1_link_is_type(&payload->links[i], link_type))
2387	buffer_json_add_array_item_uint64(wb, payload->links[i].dst_actor);
2388	}
2389	streaming_topology_v1_emit_values_end(wb);
2390
2391	streaming_topology_v1_emit_values_start(wb);
2392	for(size_t i = 0; i < payload->links_used; i++) {
2393	if(streaming_topology_v1_link_is_type(&payload->links[i], link_type))
2394	buffer_json_add_array_item_string(wb, payload->links[i].type);
2395	}
2396	streaming_topology_v1_emit_values_end(wb);
2397
2398	streaming_topology_v1_emit_values_start(wb);
2399	for(size_t i = 0; i < payload->links_used; i++) {
2400	if(streaming_topology_v1_link_is_type(&payload->links[i], link_type))
2401	buffer_json_add_array_item_string(wb, payload->links[i].state[0] ? payload->links[i].state : NULL);
2402	}
2403	streaming_topology_v1_emit_values_end(wb);
2404
2405	streaming_topology_v1_emit_values_start(wb);
2406	for(size_t i = 0; i < payload->links_used; i++) {
2407	if(streaming_topology_v1_link_is_type(&payload->links[i], link_type))
2408	buffer_json_add_array_item_string(wb, payload->links[i].port_name[0] ? payload->links[i].port_name : NULL);
2409	}
2410	streaming_topology_v1_emit_values_end(wb);
2411
2412	streaming_topology_v1_emit_values_start(wb);
2413	for(size_t i = 0; i < payload->links_used; i++) {
2414	if(streaming_topology_v1_link_is_type(&payload->links[i], link_type))
2415	streaming_topology_v1_add_timestamp(wb, payload->links[i].discovered_at_ut);
2416	}
2417	streaming_topology_v1_emit_values_end(wb);
2418
2419	streaming_topology_v1_emit_values_start(wb);
2420	for(size_t i = 0; i < payload->links_used; i++) {
2421	if(streaming_topology_v1_link_is_type(&payload->links[i], link_type))
2422	streaming_topology_v1_add_timestamp(wb, payload->links[i].last_seen_ut);
2423	}
2424	streaming_topology_v1_emit_values_end(wb);
2425
2426	#define STREAMING_TOPOLOGY_LINK_INT_VALUES(member) do { \
2427	streaming_topology_v1_emit_values_start(wb); \
2428	for(size_t i = 0; i < payload->links_used; i++) { \
2429	if(streaming_topology_v1_link_is_type(&payload->links[i], link_type)) \
2430	buffer_json_add_array_item_int64(wb, payload->links[i].member); \
2431	} \
2432	streaming_topology_v1_emit_values_end(wb); \
2433	} while(0)
2434	#define STREAMING_TOPOLOGY_LINK_UINT_VALUES(member) do { \
2435	streaming_topology_v1_emit_values_start(wb); \
2436	for(size_t i = 0; i < payload->links_used; i++) { \
2437	if(streaming_topology_v1_link_is_type(&payload->links[i], link_type)) \
2438	buffer_json_add_array_item_uint64(wb, payload->links[i].member); \
2439	} \
2440	streaming_topology_v1_emit_values_end(wb); \
2441	} while(0)
2442
2443	STREAMING_TOPOLOGY_LINK_INT_VALUES(hops);
2444	STREAMING_TOPOLOGY_LINK_UINT_VALUES(connections);
2445	STREAMING_TOPOLOGY_LINK_UINT_VALUES(replication_instances);
2446	#undef STREAMING_TOPOLOGY_LINK_INT_VALUES
2447
2448	streaming_topology_v1_emit_values_start(wb);
2449	for(size_t i = 0; i < payload->links_used; i++) {
2450	if(streaming_topology_v1_link_is_type(&payload->links[i], link_type))
2451	buffer_json_add_array_item_double(wb, payload->links[i].replication_completion);
2452	}
2453	streaming_topology_v1_emit_values_end(wb);
2454
2455	STREAMING_TOPOLOGY_LINK_UINT_VALUES(collected_metrics);
2456	STREAMING_TOPOLOGY_LINK_UINT_VALUES(collected_instances);
2457	STREAMING_TOPOLOGY_LINK_UINT_VALUES(collected_contexts);
2458	#undef STREAMING_TOPOLOGY_LINK_UINT_VALUES
2459
2460	buffer_json_array_close(wb);
2461	}
2462	buffer_json_object_close(wb);
2463	}
2464	buffer_json_object_close(wb);
2465	}
2466
2467	static void streaming_topology_v1_emit_evidence_table(BUFFER wb, STREAMING_TOPOLOGY_V1_PAYLOAD payload) {
2468	buffer_json_member_add_object(wb, "evidence");
2469	{
2470	streaming_topology_v1_emit_evidence_section(wb, payload, "streaming_link", "streaming");
2471	streaming_topology_v1_emit_evidence_section(wb, payload, "virtual_link", "virtual");
2472	streaming_topology_v1_emit_evidence_section(wb, payload, "stale_link", "stale");
2473	}
2474	buffer_json_object_close(wb);
2475	}
2476
2477	static void streaming_topology_v1_emit_actor_labels_table(BUFFER wb, STREAMING_TOPOLOGY_V1_PAYLOAD payload) {
2478	buffer_json_member_add_object(wb, "actor_labels");
2479	{
2480	buffer_json_member_add_string(wb, "type", "actor_labels");
2481	buffer_json_member_add_object(wb, "table");
2482	{
2483	buffer_json_member_add_uint64(wb, "rows", payload->labels_used);
2484	buffer_json_member_add_array(wb, "columns");
2485	streaming_topology_v1_emit_actor_label_columns(wb);
2486	buffer_json_array_close(wb);
2487	buffer_json_member_add_array(wb, "values");
2488
2489	#define STREAMING_TOPOLOGY_LABEL_UINT_VALUES(member) do { \
2490	streaming_topology_v1_emit_values_start(wb); \
2491	for(size_t i = 0; i < payload->labels_used; i++) \
2492	buffer_json_add_array_item_uint64(wb, payload->labels[i].member); \
2493	streaming_topology_v1_emit_values_end(wb); \
2494	} while(0)
2495	#define STREAMING_TOPOLOGY_LABEL_STRING_VALUES(member) do { \
2496	streaming_topology_v1_emit_values_start(wb); \
2497	for(size_t i = 0; i < payload->labels_used; i++) \
2498	buffer_json_add_array_item_string(wb, payload->labels[i].member[0] ? payload->labels[i].member : NULL); \
2499	streaming_topology_v1_emit_values_end(wb); \
2500	} while(0)
2501
2502	STREAMING_TOPOLOGY_LABEL_UINT_VALUES(actor);
2503	STREAMING_TOPOLOGY_LABEL_STRING_VALUES(key);
2504	STREAMING_TOPOLOGY_LABEL_STRING_VALUES(value);
2505	STREAMING_TOPOLOGY_LABEL_STRING_VALUES(source);
2506	STREAMING_TOPOLOGY_LABEL_STRING_VALUES(kind);
2507
2508	streaming_topology_v1_emit_values_start(wb);
2509	for(size_t i = 0; i < payload->labels_used; i++)
2510	streaming_topology_v1_add_nullable_uint(wb, payload->labels[i].has_value_index, payload->labels[i].value_index);
2511	streaming_topology_v1_emit_values_end(wb);
2512
2513	#undef STREAMING_TOPOLOGY_LABEL_UINT_VALUES
2514	#undef STREAMING_TOPOLOGY_LABEL_STRING_VALUES
2515
2516	buffer_json_array_close(wb);
2517	}
2518	buffer_json_object_close(wb);
2519	}
2520	buffer_json_object_close(wb);
2521	}
2522
2523	static void streaming_topology_v1_emit_stream_path_table(BUFFER wb, STREAMING_TOPOLOGY_V1_PAYLOAD payload) {
2524	buffer_json_member_add_object(wb, "stream_path");
2525	{
2526	buffer_json_member_add_string(wb, "type", "stream_path");
2527	buffer_json_member_add_object(wb, "table");
2528	{
2529	buffer_json_member_add_uint64(wb, "rows", payload->stream_path_used);
2530	buffer_json_member_add_array(wb, "columns");
2531	streaming_topology_v1_emit_stream_path_columns(wb);
2532	buffer_json_array_close(wb);
2533	buffer_json_member_add_array(wb, "values");
2534
2535	#define STREAMING_TOPOLOGY_STREAM_PATH_UINT_VALUES(member) do { \
2536	streaming_topology_v1_emit_values_start(wb); \
2537	for(size_t i = 0; i < payload->stream_path_used; i++) \
2538	buffer_json_add_array_item_uint64(wb, payload->stream_path_rows[i].member); \
2539	streaming_topology_v1_emit_values_end(wb); \
2540	} while(0)
2541	#define STREAMING_TOPOLOGY_STREAM_PATH_STRING_VALUES(member) do { \
2542	streaming_topology_v1_emit_values_start(wb); \
2543	for(size_t i = 0; i < payload->stream_path_used; i++) \
2544	buffer_json_add_array_item_string(wb, payload->stream_path_rows[i].member[0] ? payload->stream_path_rows[i].member : NULL); \
2545	streaming_topology_v1_emit_values_end(wb); \
2546	} while(0)
2547
2548	STREAMING_TOPOLOGY_STREAM_PATH_UINT_VALUES(actor);
2549	STREAMING_TOPOLOGY_STREAM_PATH_UINT_VALUES(path_actor);
2550	STREAMING_TOPOLOGY_STREAM_PATH_UINT_VALUES(path_index);
2551	STREAMING_TOPOLOGY_STREAM_PATH_STRING_VALUES(hostname);
2552	STREAMING_TOPOLOGY_STREAM_PATH_STRING_VALUES(host_id);
2553	STREAMING_TOPOLOGY_STREAM_PATH_STRING_VALUES(node_id);
2554	STREAMING_TOPOLOGY_STREAM_PATH_STRING_VALUES(claim_id);
2555
2556	streaming_topology_v1_emit_values_start(wb);
2557	for(size_t i = 0; i < payload->stream_path_used; i++)
2558	buffer_json_add_array_item_int64(wb, payload->stream_path_rows[i].hops);
2559	streaming_topology_v1_emit_values_end(wb);
2560
2561	streaming_topology_v1_emit_values_start(wb);
2562	for(size_t i = 0; i < payload->stream_path_used; i++)
2563	streaming_topology_v1_add_timestamp(wb, payload->stream_path_rows[i].since_ut);
2564	streaming_topology_v1_emit_values_end(wb);
2565
2566	streaming_topology_v1_emit_values_start(wb);
2567	for(size_t i = 0; i < payload->stream_path_used; i++)
2568	streaming_topology_v1_add_timestamp(wb, payload->stream_path_rows[i].first_time_ut);
2569	streaming_topology_v1_emit_values_end(wb);
2570
2571	STREAMING_TOPOLOGY_STREAM_PATH_UINT_VALUES(start_time_ms);
2572	STREAMING_TOPOLOGY_STREAM_PATH_UINT_VALUES(shutdown_time_ms);
2573	STREAMING_TOPOLOGY_STREAM_PATH_UINT_VALUES(capabilities);
2574	STREAMING_TOPOLOGY_STREAM_PATH_UINT_VALUES(flags);
2575
2576	#undef STREAMING_TOPOLOGY_STREAM_PATH_UINT_VALUES
2577	#undef STREAMING_TOPOLOGY_STREAM_PATH_STRING_VALUES
2578	buffer_json_array_close(wb);
2579	}
2580	buffer_json_object_close(wb);
2581	}
2582	buffer_json_object_close(wb);
2583	}
2584
2585	static void streaming_topology_v1_emit_retention_table(BUFFER wb, STREAMING_TOPOLOGY_V1_PAYLOAD payload) {
2586	buffer_json_member_add_object(wb, "retention");
2587	{
2588	buffer_json_member_add_string(wb, "type", "retention");
2589	buffer_json_member_add_object(wb, "table");
2590	{
2591	buffer_json_member_add_uint64(wb, "rows", payload->retention_used);
2592	buffer_json_member_add_array(wb, "columns");
2593	streaming_topology_v1_emit_retention_columns(wb);
2594	buffer_json_array_close(wb);
2595	buffer_json_member_add_array(wb, "values");
2596
2597	#define STREAMING_TOPOLOGY_RETENTION_UINT_VALUES(member) do { \
2598	streaming_topology_v1_emit_values_start(wb); \
2599	for(size_t i = 0; i < payload->retention_used; i++) \
2600	buffer_json_add_array_item_uint64(wb, payload->retention_rows[i].member); \
2601	streaming_topology_v1_emit_values_end(wb); \
2602	} while(0)
2603
2604	STREAMING_TOPOLOGY_RETENTION_UINT_VALUES(actor);
2605	STREAMING_TOPOLOGY_RETENTION_UINT_VALUES(observer_actor);
2606
2607	streaming_topology_v1_emit_values_start(wb);
2608	for(size_t i = 0; i < payload->retention_used; i++)
2609	buffer_json_add_array_item_string(wb, payload->retention_rows[i].db_status[0] ? payload->retention_rows[i].db_status : NULL);
2610	streaming_topology_v1_emit_values_end(wb);
2611
2612	streaming_topology_v1_emit_values_start(wb);
2613	for(size_t i = 0; i < payload->retention_used; i++)
2614	streaming_topology_v1_add_timestamp(wb, payload->retention_rows[i].db_from_ut);
2615	streaming_topology_v1_emit_values_end(wb);
2616
2617	streaming_topology_v1_emit_values_start(wb);
2618	for(size_t i = 0; i < payload->retention_used; i++)
2619	streaming_topology_v1_add_timestamp(wb, payload->retention_rows[i].db_to_ut);
2620	streaming_topology_v1_emit_values_end(wb);
2621
2622	STREAMING_TOPOLOGY_RETENTION_UINT_VALUES(db_duration);
2623	STREAMING_TOPOLOGY_RETENTION_UINT_VALUES(db_metrics);
2624	STREAMING_TOPOLOGY_RETENTION_UINT_VALUES(db_instances);
2625	STREAMING_TOPOLOGY_RETENTION_UINT_VALUES(db_contexts);
2626	#undef STREAMING_TOPOLOGY_RETENTION_UINT_VALUES
2627	buffer_json_array_close(wb);
2628	}
2629	buffer_json_object_close(wb);
2630	}
2631	buffer_json_object_close(wb);
2632	}
2633
2634	static void streaming_topology_v1_emit_inbound_table(BUFFER wb, STREAMING_TOPOLOGY_V1_PAYLOAD payload) {
2635	buffer_json_member_add_object(wb, "inbound");
2636	{
2637	buffer_json_member_add_string(wb, "type", "inbound");
2638	buffer_json_member_add_object(wb, "table");
2639	{
2640	buffer_json_member_add_uint64(wb, "rows", payload->inbound_used);
2641	buffer_json_member_add_array(wb, "columns");
2642	streaming_topology_v1_emit_inbound_columns(wb);
2643	buffer_json_array_close(wb);
2644	buffer_json_member_add_array(wb, "values");
2645
2646	streaming_topology_v1_emit_values_start(wb);
2647	for(size_t i = 0; i < payload->inbound_used; i++)
2648	buffer_json_add_array_item_uint64(wb, payload->inbound_rows[i].parent_actor);
2649	streaming_topology_v1_emit_values_end(wb);
2650
2651	streaming_topology_v1_emit_values_start(wb);
2652	for(size_t i = 0; i < payload->inbound_used; i++)
2653	buffer_json_add_array_item_uint64(wb, payload->inbound_rows[i].child_actor);
2654	streaming_topology_v1_emit_values_end(wb);
2655
2656	streaming_topology_v1_emit_values_start(wb);
2657	for(size_t i = 0; i < payload->inbound_used; i++)
2658	streaming_topology_v1_add_nullable_uint(wb,
2659	payload->inbound_rows[i].has_source_actor, payload->inbound_rows[i].source_actor);
2660	streaming_topology_v1_emit_values_end(wb);
2661
2662	#define STREAMING_TOPOLOGY_INBOUND_STRING_VALUES(member) do { \
2663	streaming_topology_v1_emit_values_start(wb); \
2664	for(size_t i = 0; i < payload->inbound_used; i++) \
2665	buffer_json_add_array_item_string(wb, payload->inbound_rows[i].member[0] ? payload->inbound_rows[i].member : NULL); \
2666	streaming_topology_v1_emit_values_end(wb); \
2667	} while(0)
2668	#define STREAMING_TOPOLOGY_INBOUND_UINT_VALUES(member) do { \
2669	streaming_topology_v1_emit_values_start(wb); \
2670	for(size_t i = 0; i < payload->inbound_used; i++) \
2671	buffer_json_add_array_item_uint64(wb, payload->inbound_rows[i].member); \
2672	streaming_topology_v1_emit_values_end(wb); \
2673	} while(0)
2674
2675	STREAMING_TOPOLOGY_INBOUND_STRING_VALUES(received_type);
2676	STREAMING_TOPOLOGY_INBOUND_STRING_VALUES(ingest_status);
2677
2678	streaming_topology_v1_emit_values_start(wb);
2679	for(size_t i = 0; i < payload->inbound_used; i++)
2680	buffer_json_add_array_item_int64(wb, payload->inbound_rows[i].hops);
2681	streaming_topology_v1_emit_values_end(wb);
2682
2683	STREAMING_TOPOLOGY_INBOUND_UINT_VALUES(collected_metrics);
2684	STREAMING_TOPOLOGY_INBOUND_UINT_VALUES(collected_instances);
2685	STREAMING_TOPOLOGY_INBOUND_UINT_VALUES(collected_contexts);
2686
2687	streaming_topology_v1_emit_values_start(wb);
2688	for(size_t i = 0; i < payload->inbound_used; i++)
2689	buffer_json_add_array_item_double(wb, payload->inbound_rows[i].replication_completion);
2690	streaming_topology_v1_emit_values_end(wb);
2691
2692	STREAMING_TOPOLOGY_INBOUND_UINT_VALUES(ingest_age);
2693	STREAMING_TOPOLOGY_INBOUND_STRING_VALUES(ssl);
2694	STREAMING_TOPOLOGY_INBOUND_UINT_VALUES(alerts_critical);
2695	STREAMING_TOPOLOGY_INBOUND_UINT_VALUES(alerts_warning);
2696	#undef STREAMING_TOPOLOGY_INBOUND_STRING_VALUES
2697	#undef STREAMING_TOPOLOGY_INBOUND_UINT_VALUES
2698	buffer_json_array_close(wb);
2699	}
2700	buffer_json_object_close(wb);
2701	}
2702	buffer_json_object_close(wb);
2703	}
2704
2705	static void streaming_topology_v1_emit_outbound_table(BUFFER wb, STREAMING_TOPOLOGY_V1_PAYLOAD payload) {
2706	buffer_json_member_add_object(wb, "outbound");
2707	{
2708	buffer_json_member_add_string(wb, "type", "outbound");
2709	buffer_json_member_add_object(wb, "table");
2710	{
2711	buffer_json_member_add_uint64(wb, "rows", payload->outbound_used);
2712	buffer_json_member_add_array(wb, "columns");
2713	streaming_topology_v1_emit_outbound_columns(wb);
2714	buffer_json_array_close(wb);
2715	buffer_json_member_add_array(wb, "values");
2716
2717	streaming_topology_v1_emit_values_start(wb);
2718	for(size_t i = 0; i < payload->outbound_used; i++)
2719	buffer_json_add_array_item_uint64(wb, payload->outbound_rows[i].sender_actor);
2720	streaming_topology_v1_emit_values_end(wb);
2721
2722	streaming_topology_v1_emit_values_start(wb);
2723	for(size_t i = 0; i < payload->outbound_used; i++)
2724	buffer_json_add_array_item_uint64(wb, payload->outbound_rows[i].node_actor);
2725	streaming_topology_v1_emit_values_end(wb);
2726
2727	streaming_topology_v1_emit_values_start(wb);
2728	for(size_t i = 0; i < payload->outbound_used; i++)
2729	streaming_topology_v1_add_nullable_uint(wb,
2730	payload->outbound_rows[i].has_destination_actor, payload->outbound_rows[i].destination_actor);
2731	streaming_topology_v1_emit_values_end(wb);
2732
2733	streaming_topology_v1_emit_values_start(wb);
2734	for(size_t i = 0; i < payload->outbound_used; i++)
2735	buffer_json_add_array_item_string(wb, payload->outbound_rows[i].stream_status[0] ? payload->outbound_rows[i].stream_status : NULL);
2736	streaming_topology_v1_emit_values_end(wb);
2737
2738	streaming_topology_v1_emit_values_start(wb);
2739	for(size_t i = 0; i < payload->outbound_used; i++)
2740	buffer_json_add_array_item_uint64(wb, payload->outbound_rows[i].stream_age);
2741	streaming_topology_v1_emit_values_end(wb);
2742
2743	streaming_topology_v1_emit_values_start(wb);
2744	for(size_t i = 0; i < payload->outbound_used; i++)
2745	buffer_json_add_array_item_int64(wb, payload->outbound_rows[i].hops);
2746	streaming_topology_v1_emit_values_end(wb);
2747
2748	streaming_topology_v1_emit_values_start(wb);
2749	for(size_t i = 0; i < payload->outbound_used; i++)
2750	buffer_json_add_array_item_string(wb, payload->outbound_rows[i].ssl[0] ? payload->outbound_rows[i].ssl : NULL);
2751	streaming_topology_v1_emit_values_end(wb);
2752
2753	streaming_topology_v1_emit_values_start(wb);
2754	for(size_t i = 0; i < payload->outbound_used; i++)
2755	buffer_json_add_array_item_string(wb, payload->outbound_rows[i].compression[0] ? payload->outbound_rows[i].compression : NULL);
2756	streaming_topology_v1_emit_values_end(wb);
2757
2758	streaming_topology_v1_emit_values_start(wb);
2759	for(size_t i = 0; i < payload->outbound_used; i++)
2760	buffer_json_add_array_item_uint64(wb, payload->outbound_rows[i].collected_metrics);
2761	streaming_topology_v1_emit_values_end(wb);
2762
2763	streaming_topology_v1_emit_values_start(wb);
2764	for(size_t i = 0; i < payload->outbound_used; i++)
2765	buffer_json_add_array_item_uint64(wb, payload->outbound_rows[i].collected_instances);
2766	streaming_topology_v1_emit_values_end(wb);
2767
2768	streaming_topology_v1_emit_values_start(wb);
2769	for(size_t i = 0; i < payload->outbound_used; i++)
2770	buffer_json_add_array_item_uint64(wb, payload->outbound_rows[i].collected_contexts);
2771	streaming_topology_v1_emit_values_end(wb);
2772
2773	streaming_topology_v1_emit_values_start(wb);
2774	for(size_t i = 0; i < payload->outbound_used; i++)
2775	buffer_json_add_array_item_uint64(wb, payload->outbound_rows[i].replication_instances);
2776	streaming_topology_v1_emit_values_end(wb);
2777
2778	streaming_topology_v1_emit_values_start(wb);
2779	for(size_t i = 0; i < payload->outbound_used; i++)
2780	buffer_json_add_array_item_double(wb, payload->outbound_rows[i].replication_completion);
2781	streaming_topology_v1_emit_values_end(wb);
2782
2783	buffer_json_array_close(wb);
2784	}
2785	buffer_json_object_close(wb);
2786	}
2787	buffer_json_object_close(wb);
2788	}
2789
2790	static void streaming_topology_v1_emit_detail_tables(BUFFER wb, STREAMING_TOPOLOGY_V1_PAYLOAD payload) {
2791	buffer_json_member_add_object(wb, "tables");
2792	{
2793	buffer_json_member_add_object(wb, "actor");
2794	{
2795	streaming_topology_v1_emit_actor_labels_table(wb, payload);
2796	streaming_topology_v1_emit_stream_path_table(wb, payload);
2797	streaming_topology_v1_emit_retention_table(wb, payload);
2798	streaming_topology_v1_emit_inbound_table(wb, payload);
2799	streaming_topology_v1_emit_outbound_table(wb, payload);
2800	}
2801	buffer_json_object_close(wb);
2802	}
2803	buffer_json_object_close(wb);
2804	}
2805
2806	int function_streaming_topology(BUFFER wb, const char function, BUFFER payload __maybe_unused, const char source __maybe_unused) {
2807	time_t now = now_realtime_sec();
2808	usec_t now_ut = now_realtime_usec();
2809
2810	struct streaming_topology_options options = { 0 };
2811	streaming_topology_parse_options(function, &options);
2812	char *function_copy = options.function_copy;
2813
2814	if(options.info_only) {
2815	buffer_flush(wb);
2816	wb->content_type = CT_APPLICATION_JSON;
2817	buffer_json_initialize(wb, "\"", "\"", 0, true, BUFFER_JSON_OPTIONS_DEFAULT);
2818	streaming_topology_v1_emit_response_metadata(wb);
2819	buffer_json_member_add_time_t(wb, "expires", now_realtime_sec() + STREAMING_FUNCTION_UPDATE_EVERY);
2820	buffer_json_finalize(wb);
2821	freez(function_copy);
2822	return HTTP_RESP_OK;
2823	}
2824
2825	DICTIONARY *parent_child_count = dictionary_create_advanced(
2826	DICT_OPTION_SINGLE_THREADED \| DICT_OPTION_DONT_OVERWRITE_VALUE \| DICT_OPTION_FIXED_SIZE,
2827	NULL, sizeof(uint32_t));
2828	DICTIONARY *parent_descendants = dictionary_create_advanced(
2829	DICT_OPTION_SINGLE_THREADED \| DICT_OPTION_DONT_OVERWRITE_VALUE \| DICT_OPTION_FIXED_SIZE,
2830	NULL, sizeof(struct streaming_topology_descendant_list));
2831
2832	STREAMING_TOPOLOGY_V1_PAYLOAD topology = {
2833	.actor_index = dictionary_create_advanced(
2834	DICT_OPTION_SINGLE_THREADED \| DICT_OPTION_DONT_OVERWRITE_VALUE \| DICT_OPTION_FIXED_SIZE,
2835	NULL, sizeof(uint64_t)),
2836	.emitted_links = dictionary_create_advanced(
2837	DICT_OPTION_SINGLE_THREADED \| DICT_OPTION_DONT_OVERWRITE_VALUE \| DICT_OPTION_FIXED_SIZE,
2838	NULL, sizeof(uint8_t)),
2839	};
2840
2841	if(!parent_child_count \|\| !parent_descendants \|\| !topology.actor_index \|\| !topology.emitted_links) {
2842	streaming_topology_v1_free(&topology);
2843	if(parent_descendants)
2844	dictionary_destroy(parent_descendants);
2845	if(parent_child_count)
2846	dictionary_destroy(parent_child_count);
2847
2848	return streaming_topology_return_error(wb, function_copy,
2849	HTTP_RESP_INTERNAL_SERVER_ERROR,
2850	"failed to allocate streaming topology dictionaries");
2851	}
2852
2853	{
2854	RRDHOST *host;
2855	dfe_start_read(rrdhost_root_index, host) {
2856	ND_UUID path_ids[128];
2857	uint16_t n = streaming_topology_get_path_ids(host, 1, path_ids, 128);
2858	for(uint16_t i = 0; i < n; i++) {
2859	char guid[UUID_STR_LEN];
2860	if(!streaming_topology_uuid_guid(path_ids[i], guid, sizeof(guid)))
2861	continue;
2862
2863	uint32_t *count = dictionary_get(parent_child_count, guid);
2864	if(count)
2865	(*count)++;
2866	else {
2867	uint32_t one = 1;
2868	dictionary_set(parent_child_count, guid, &one, sizeof(one));
2869	}
2870	}
2871
2872	ND_UUID full_path_ids[128];
2873	uint16_t full_path_n = streaming_topology_get_path_ids(host, 0, full_path_ids, 128);
2874	ND_UUID empty_uuid = {};
2875
2876	if(rrdhost_is_virtual(host))
2877	continue;
2878
2879	for(uint16_t i = 0; i < full_path_n; i++) {
2880	if(UUIDeq(full_path_ids[i], localhost->host_id))
2881	continue;
2882
2883	bool source_local = (i == 0);
2884	ND_UUID source_uuid = source_local ? empty_uuid : full_path_ids[i - 1];
2885	streaming_topology_descendants_append(parent_descendants,
2886	full_path_ids[i], host, STREAMING_TOPOLOGY_RECEIVED_STREAMING, source_local, source_uuid);
2887	}
2888	}
2889	dfe_done(host);
2890	}
2891
2892	{
2893	char localhost_guid[UUID_STR_LEN];
2894	if(streaming_topology_uuid_guid(localhost->host_id, localhost_guid, sizeof(localhost_guid))) {
2895	uint32_t live_count = 0;
2896	ND_UUID empty_uuid_for_live = {};
2897	RRDHOST *host;
2898	dfe_start_read(rrdhost_root_index, host) {
2899	if(host == localhost)
2900	continue;
2901
2902	if(rrdhost_is_virtual(host)) {
2903	streaming_topology_descendants_append(parent_descendants,
2904	localhost->host_id, host,
2905	STREAMING_TOPOLOGY_RECEIVED_VIRTUAL, true, empty_uuid_for_live);
2906	continue;
2907	}
2908
2909	RRDHOST_STATUS status;
2910	rrdhost_status(host, now, &status, RRDHOST_STATUS_ALL);
2911
2912	if(status.ingest.type == RRDHOST_INGEST_TYPE_CHILD &&
2913	(status.ingest.status == RRDHOST_INGEST_STATUS_ONLINE \|\|
2914	status.ingest.status == RRDHOST_INGEST_STATUS_REPLICATING)) {
2915	live_count++;
2916	streaming_topology_descendants_append(parent_descendants,
2917	localhost->host_id, host,
2918	STREAMING_TOPOLOGY_RECEIVED_STREAMING, true, empty_uuid_for_live);
2919	}
2920	else {
2921	streaming_topology_descendants_append(parent_descendants,
2922	localhost->host_id, host,
2923	STREAMING_TOPOLOGY_RECEIVED_STALE, true, empty_uuid_for_live);
2924	}
2925	}
2926	dfe_done(host);
2927
2928	uint32_t *existing = dictionary_get(parent_child_count, localhost_guid);
2929	if(existing)
2930	*existing = live_count;
2931	else if(live_count > 0)
2932	dictionary_set(parent_child_count, localhost_guid, &live_count, sizeof(live_count));
2933	}
2934	}
2935
2936	uint64_t local_retained_node_count = streaming_topology_v1_count_local_retained_nodes(now);
2937	streaming_topology_v1_collect_actors(&topology, parent_child_count, local_retained_node_count, now);
2938	streaming_topology_v1_collect_links(&topology, now, now_ut);
2939	streaming_topology_v1_collect_actor_detail_rows(&topology, parent_descendants, now);
2940
2941	buffer_flush(wb);
2942	wb->content_type = CT_APPLICATION_JSON;
2943	buffer_json_initialize(wb, "\"", "\"", 0, true, BUFFER_JSON_OPTIONS_DEFAULT);
2944
2945	streaming_topology_v1_emit_response_metadata(wb);
2946
2947	buffer_json_member_add_object(wb, "data");
2948	{
2949	buffer_json_member_add_string(wb, "schema_version", "netdata.topology.v1");
2950
2951	buffer_json_member_add_object(wb, "producer");
2952	{
2953	char localhost_agent_id[256];
2954	char localhost_guid[UUID_STR_LEN];
2955	streaming_topology_agent_id_for_host(localhost, localhost_agent_id, sizeof(localhost_agent_id));
2956	streaming_topology_host_guid(localhost, localhost_guid, sizeof(localhost_guid));
2957
2958	buffer_json_member_add_string(wb, "source", "streaming");
2959	buffer_json_member_add_string(wb, "instance", localhost_agent_id);
2960	if(!UUIDiszero(localhost->node_id))
2961	buffer_json_member_add_uuid(wb, "node_id", localhost->node_id.uuid);
2962	if(localhost_guid[0])
2963	buffer_json_member_add_string(wb, "machine_guid", localhost_guid);
2964	buffer_json_member_add_string(wb, "agent_version", rrdhost_program_version(localhost));
2965	buffer_json_member_add_string(wb, "plugin", "netdata");
2966	buffer_json_member_add_array(wb, "capabilities");
2967	buffer_json_add_array_item_string(wb, "topology-v1");
2968	buffer_json_array_close(wb);
2969	}
2970	buffer_json_object_close(wb);
2971
2972	buffer_json_member_add_datetime_rfc3339(wb, "collected_at", now_ut, true);
2973	buffer_json_member_add_object(wb, "view");
2974	{
2975	buffer_json_member_add_string(wb, "id", "streaming");
2976	buffer_json_member_add_string(wb, "scope", "node");
2977	buffer_json_member_add_string(wb, "mode", "detailed");
2978	buffer_json_member_add_array(wb, "group_by");
2979	buffer_json_add_array_item_string(wb, "node");
2980	buffer_json_array_close(wb);
2981	}
2982	buffer_json_object_close(wb);
2983
2984	buffer_json_member_add_object(wb, "dictionaries");
2985	{
2986	buffer_json_member_add_array(wb, "strings");
2987	buffer_json_array_close(wb);
2988	}
2989	buffer_json_object_close(wb);
2990
2991	streaming_topology_v1_emit_type_registry(wb);
2992	streaming_topology_v1_emit_presentation(wb);
2993	streaming_topology_v1_emit_actor_table(wb, &topology);
2994	streaming_topology_v1_emit_link_table(wb, &topology);
2995	streaming_topology_v1_emit_evidence_table(wb, &topology);
2996	streaming_topology_v1_emit_detail_tables(wb, &topology);
2997
2998	buffer_json_member_add_object(wb, "stats");
2999	{
3000	buffer_json_member_add_uint64(wb, "actors", topology.actors_used);
3001	buffer_json_member_add_uint64(wb, "links", topology.links_used);
3002	buffer_json_member_add_uint64(wb, "evidence_rows", topology.links_used);
3003	buffer_json_member_add_uint64(wb, "stream_path_rows", topology.stream_path_used);
3004	buffer_json_member_add_uint64(wb, "retention_rows", topology.retention_used);
3005	buffer_json_member_add_uint64(wb, "inbound_rows", topology.inbound_used);
3006	buffer_json_member_add_uint64(wb, "outbound_rows", topology.outbound_used);
3007	}
3008	buffer_json_object_close(wb);
3009	}
3010	buffer_json_object_close(wb);
3011
3012	buffer_json_member_add_time_t(wb, "expires", now_realtime_sec() + STREAMING_FUNCTION_UPDATE_EVERY);
3013	buffer_json_finalize(wb);
3014
3015	struct streaming_topology_descendant_list *descendants;
3016	dfe_start_write(parent_descendants, descendants) {
3017	freez(descendants->items);
3018	descendants->items = NULL;
3019	descendants->used = 0;
3020	descendants->size = 0;
3021	}
3022	dfe_done(descendants);
3023
3024	dictionary_destroy(parent_descendants);
3025	dictionary_destroy(parent_child_count);
3026	streaming_topology_v1_free(&topology);
3027	freez(function_copy);
3028	return HTTP_RESP_OK;
3029	}