@cryptotaxi247 / netdata-1 / commits / 22b9a36ec

Add offline Function test modes for NetFlow and systemd journal (#22638)

Costa Tsaousis committed Jun 6, 2026 at 23:13 UTC 22b9a36eceeb6389275e9f2a64478256ca63c62e

14 files changed +1353 -54

src/collectors/systemd-journal.plugin/README.md

+29

@@ -36,6 +36,35 @@ This plugin is a Netdata Function Plugin. A free Netdata Cloud account is requir

37 The plugin is designed for native package installations, source installations, and Docker installations (Debian-based). If using Docker, make sure you're using the Debian-based containers.

39 +## Offline Function test mode

40 +

41 +Fixture harnesses can execute the Function query path directly against an existing

42 +journal directory:

43 +

44 +```sh

45 +systemd-journal.plugin --test systemd-journal --dir <journal-dir> [--timeout <seconds>] < payload.json

46 +```

47 +

48 +Requirements:

49 +

50 +- `<journal-dir>` is scanned recursively for systemd journal files.

51 +- stdin is the JSON Function request body (non-empty, maximum 16 MiB).

52 +- `--request` is not supported and fails with usage output.

53 +- `--timeout <seconds>` controls the offline Function execution timeout. It

54 + defaults to `60`; use `--timeout 0` to map to a very large finite timeout for

55 + long-running fixture comparisons.

56 +- stdout contains only the raw JSON Function response.

57 +- errors are written to stderr and return non-zero.

58 +

59 +For example, an info request can use this payload:

60 +

61 +```json

62 +{"info":true}

63 +```

64 +

65 +Function output includes volatile fields such as versions and timing-derived values.

66 +Test harnesses should normalize those fields before comparing fixture outputs.

67 +

68 ## Journal sources

70 The plugin automatically detects available journal sources based on the journal files in `/var/log/journal` (persistent logs) and `/run/log/journal` (volatile logs).

src/collectors/systemd-journal.plugin/systemd-internals.h

+11

@@ -97,6 +97,8 @@ void available_journal_file_sources_to_json_array(BUFFER *wb);

97 bool nd_journal_files_completed_once(void);

98 void nd_journal_files_registry_update(void);

99 void nd_journal_directory_scan_recursively(DICTIONARY *files, DICTIONARY *dirs, const char *dirname, int depth);

100 +void nd_journal_set_scan_progress_enabled(bool enabled);

101 +void nd_journal_use_single_directory(const char *path);

102

103 FACET_ROW_SEVERITY syslog_priority_to_facet_severity(FACETS *facets, FACET_ROW *row, void *data);

104

@@ -124,6 +126,15 @@ struct journal_directory {

126 extern struct journal_directory journal_directories[MAX_JOURNAL_DIRECTORIES];

127

128 void nd_journal_init_files_and_directories(void);

129 +BUFFER *function_systemd_journal_result(

130 + const char *transaction,

131 + char *function,

132 + usec_t *stop_monotonic_ut,

133 + bool *cancelled,

134 + BUFFER *payload,

135 + HTTP_ACCESS access __maybe_unused,

136 + const char *source __maybe_unused,

137 + void *data __maybe_unused);

138 void function_systemd_journal(

139 const char *transaction,

140 char *function,

src/collectors/systemd-journal.plugin/systemd-journal-files.c

+26 -3

@@ -11,6 +11,29 @@ DICTIONARY *nd_journal_files_registry = NULL;

11 DICTIONARY *used_hashes_registry = NULL;

13 static usec_t systemd_journal_session = 0;

14 +static bool nd_journal_scan_progress_enabled = true;

15 +

16 +void nd_journal_set_scan_progress_enabled(bool enabled)

17 +{

18 + nd_journal_scan_progress_enabled = enabled;

19 +}

20 +

21 +static void nd_journal_scan_progress(void)

22 +{

23 + if (nd_journal_scan_progress_enabled)

24 + send_newline_and_flush(&stdout_mutex);

25 +}

26 +

27 +void nd_journal_use_single_directory(const char *path)

28 +{

29 + string_freez(journal_directories[0].path);

30 + journal_directories[0].path = string_strdupz(path);

31 +

32 + for (size_t i = 1; i < MAX_JOURNAL_DIRECTORIES; i++) {

33 + string_freez(journal_directories[i].path);

34 + journal_directories[i].path = NULL;

35 + }

36 +}

38 void buffer_json_journal_versions(BUFFER *wb)

39 {

@@ -659,14 +682,14 @@ void nd_journal_directory_scan_recursively(DICTIONARY *files, DICTIONARY *dirs,

682 if (files)

683 dictionary_set(files, full_path, NULL, 0);

684

662 - send_newline_and_flush(&stdout_mutex);

685 + nd_journal_scan_progress();

686 } else if (entry->d_type == DT_LNK) {

687 struct stat info;

688 if (stat(full_path, &info) == -1)

689 continue;

690

691 + // Journal discovery intentionally follows symlinked journal directories.

692 if (S_ISDIR(info.st_mode)) {

669 - // The symbolic link points to a directory

693 char resolved_path[FILENAME_MAX + 1];

694 if (realpath(full_path, resolved_path) != NULL) {

695 nd_journal_directory_scan_recursively(files, dirs, resolved_path, depth + 1);

@@ -675,7 +698,7 @@ void nd_journal_directory_scan_recursively(DICTIONARY *files, DICTIONARY *dirs,

698 if (files)

699 dictionary_set(files, full_path, NULL, 0);

700

678 - send_newline_and_flush(&stdout_mutex);

701 + nd_journal_scan_progress();

702 }

703 }

704 }

src/collectors/systemd-journal.plugin/systemd-journal.c

+27 -11

@@ -16,8 +16,8 @@

16 #define JOURNAL_KEY_ND_JOURNAL_PROCESS "ND_JOURNAL_PROCESS"

18 // functions needed by LQS

19 -static __always_inline

20 -SD_JOURNAL_FILE_SOURCE_TYPE get_internal_source_type(const char *value) {

19 +static __always_inline SD_JOURNAL_FILE_SOURCE_TYPE get_internal_source_type(const char *value)

20 +{

21 if (strcmp(value, ND_SD_JF_SOURCE_ALL_NAME) == 0)

22 return ND_SD_JF_ALL;

23 else if (strcmp(value, ND_SD_JF_SOURCE_LOCAL_NAME) == 0)

@@ -87,8 +87,7 @@ SD_JOURNAL_FILE_SOURCE_TYPE get_internal_source_type(const char *value) {

87 "|_UID" \

88 "|_GID" \

89 "|_COMM" \

90 - "|_EXE" /* "|_CMDLINE" */ \

91 - "|_CAP_EFFECTIVE" /* "|_AUDIT_SESSION" */ \

90 + "|_EXE" /* "|_CMDLINE" "|_CAP_EFFECTIVE" "|_AUDIT_SESSION" */ \

91 "|_AUDIT_LOGINUID" \

92 "|_SYSTEMD_CGROUP" \

93 "|_SYSTEMD_SLICE" \

@@ -101,8 +100,7 @@ SD_JOURNAL_FILE_SOURCE_TYPE get_internal_source_type(const char *value) {

100 "|_BOOT_ID" \

101 "|_MACHINE_ID" /* "|_SYSTEMD_INVOCATION_ID" */ \

102 "|_HOSTNAME" \

104 - "|_TRANSPORT" \

105 - "|_STREAM_ID" /* "|LINE_BREAK" */ \

103 + "|_TRANSPORT" /* "|_STREAM_ID" "|LINE_BREAK" */ \

104 "|_NAMESPACE" \

105 "|_RUNTIME_SCOPE" \

106 \

@@ -155,8 +153,8 @@ SD_JOURNAL_FILE_SOURCE_TYPE get_internal_source_type(const char *value) {

153

154 #include "systemd-journal-execute.h"

155

158 -static

159 -void systemd_journal_register_transformations(LOGS_QUERY_STATUS *lqs) {

156 +static void systemd_journal_register_transformations(LOGS_QUERY_STATUS *lqs)

157 +{

158 FACETS *facets = lqs->facets;

159 LOGS_QUERY_REQUEST *rq = &lqs->rq;

160

@@ -257,7 +255,7 @@ void systemd_journal_register_transformations(LOGS_QUERY_STATUS *lqs) {

255 NULL);

256 }

257

260 -void function_systemd_journal(

258 +BUFFER *function_systemd_journal_result(

259 const char *transaction,

260 char *function,

261 usec_t *stop_monotonic_ut,

@@ -292,7 +290,7 @@ void function_systemd_journal(

290 };

291 LOGS_QUERY_STATUS *lqs = &tmp_fqs;

292

295 - CLEAN_BUFFER *wb = lqs_create_output_buffer();

293 + BUFFER *wb = lqs_create_output_buffer();

294

295 // ------------------------------------------------------------------------

296 // parse the parameters

@@ -317,9 +315,27 @@ void function_systemd_journal(

315 }

316 }

317

318 + lqs_cleanup(lqs);

319 +

320 + return wb;

321 +}

322 +

323 +void function_systemd_journal(

324 + const char *transaction,

325 + char *function,

326 + usec_t *stop_monotonic_ut,

327 + bool *cancelled,

328 + BUFFER *payload,

329 + HTTP_ACCESS access,

330 + const char *source,

331 + void *data)

332 +{

333 + BUFFER *wb = function_systemd_journal_result(

334 + transaction, function, stop_monotonic_ut, cancelled, payload, access, source, data);

335 +

336 netdata_mutex_lock(&stdout_mutex);

337 pluginsd_function_result_to_stdout(transaction, wb);

338 netdata_mutex_unlock(&stdout_mutex);

339

324 - lqs_cleanup(lqs);

340 + buffer_free(wb);

341 }

src/collectors/systemd-journal.plugin/systemd-main.c

+375 -1

@@ -3,6 +3,8 @@

3 #include "systemd-internals.h"

5 #define ND_SD_JOURNAL_WORKER_THREADS 5

6 +#define ND_SD_JOURNAL_TEST_TIMEOUT_DISABLED_SECONDS (100ULL * 365ULL * 24ULL * 60ULL * 60ULL)

7 +#define ND_SD_JOURNAL_TEST_MAX_REQUEST_BYTES (16ULL * 1024ULL * 1024ULL)

9 netdata_mutex_t stdout_mutex;

@@ -16,6 +18,370 @@ static void __attribute__((destructor)) destroy_mutex(void) {

19 static bool plugin_should_exit = false;

21 +struct systemd_journal_test_command {

22 + bool enabled;

23 + const char *function_name;

24 + const char *backend_dir;

25 + uint64_t timeout_seconds;

26 + bool timeout_seconds_set;

27 +};

28 +

29 +static void systemd_journal_test_usage(FILE *stream)

30 +{

31 + fprintf(

32 + stream,

33 + "usage: systemd-journal.plugin --test systemd-journal --dir <journal-dir> [--timeout <seconds>] < payload.json\n");

34 +}

35 +

36 +static bool test_option_present(int argc, char **argv)

37 +{

38 + for (int i = 1; i < argc; i++) {

39 + if (strcmp(argv[i], "--test") == 0 || strncmp(argv[i], "--test=", strlen("--test=")) == 0)

40 + return true;

41 + }

42 +

43 + return false;

44 +}

45 +

46 +static int set_required_option_once(const char **slot, const char *value, const char *option)

47 +{

48 + if (*slot) {

49 + fprintf(stderr, "duplicate %s\n", option);

50 + systemd_journal_test_usage(stderr);

51 + return 2;

52 + }

53 +

54 + if (!value || !*value) {

55 + fprintf(stderr, "missing value for %s\n", option);

56 + systemd_journal_test_usage(stderr);

57 + return 2;

58 + }

59 +

60 + *slot = value;

61 + return 0;

62 +}

63 +

64 +static int set_timeout_option_once(uint64_t *slot, bool *slot_set, const char *value)

65 +{

66 + if (*slot_set) {

67 + fprintf(stderr, "duplicate --timeout\n");

68 + systemd_journal_test_usage(stderr);

69 + return 2;

70 + }

71 +

72 + if (!value || !*value) {

73 + fprintf(stderr, "missing value for --timeout\n");

74 + systemd_journal_test_usage(stderr);

75 + return 2;

76 + }

77 +

78 + for (const char *s = value; *s; s++) {

79 + if (*s < '0' || *s > '9') {

80 + fprintf(stderr, "invalid value for --timeout '%s'; expected seconds\n", value);

81 + systemd_journal_test_usage(stderr);

82 + return 2;

83 + }

84 + }

85 +

86 + errno = 0;

87 + unsigned long long parsed = strtoull(value, NULL, 10);

88 + if (errno == ERANGE) {

89 + fprintf(stderr, "invalid value for --timeout '%s'; expected seconds\n", value);

90 + systemd_journal_test_usage(stderr);

91 + return 2;

92 + }

93 +

94 +#if ULLONG_MAX > UINT64_MAX

95 + if (parsed > UINT64_MAX) {

96 + fprintf(stderr, "invalid value for --timeout '%s'; expected seconds\n", value);

97 + systemd_journal_test_usage(stderr);

98 + return 2;

99 + }

100 +#endif

101 +

102 + *slot = (uint64_t)parsed;

103 + *slot_set = true;

104 + return 0;

105 +}

106 +

107 +static int reject_request_option(void)

108 +{

109 + fprintf(stderr, "--request is no longer supported; pass the request payload on stdin\n");

110 + systemd_journal_test_usage(stderr);

111 + return 2;

112 +}

113 +

114 +static int parse_systemd_journal_test_command(int argc, char **argv, struct systemd_journal_test_command *cmd)

115 +{

116 + *cmd = (struct systemd_journal_test_command){0};

117 + if (!test_option_present(argc, argv))

118 + return 0;

119 +

120 + cmd->enabled = true;

121 +

122 + for (int i = 1; i < argc; i++) {

123 + const char *arg = argv[i];

124 +

125 + if (strcmp(arg, "--test") == 0) {

126 + if (++i >= argc)

127 + return set_required_option_once(&cmd->function_name, NULL, "--test");

128 +

129 + int rc = set_required_option_once(&cmd->function_name, argv[i], "--test");

130 + if (rc)

131 + return rc;

132 + }

133 + else if (strncmp(arg, "--test=", strlen("--test=")) == 0) {

134 + int rc = set_required_option_once(&cmd->function_name, arg + strlen("--test="), "--test");

135 + if (rc)

136 + return rc;

137 + }

138 + else if (strcmp(arg, "--dir") == 0) {

139 + if (++i >= argc)

140 + return set_required_option_once(&cmd->backend_dir, NULL, "--dir");

141 +

142 + int rc = set_required_option_once(&cmd->backend_dir, argv[i], "--dir");

143 + if (rc)

144 + return rc;

145 + }

146 + else if (strncmp(arg, "--dir=", strlen("--dir=")) == 0) {

147 + int rc = set_required_option_once(&cmd->backend_dir, arg + strlen("--dir="), "--dir");

148 + if (rc)

149 + return rc;

150 + }

151 + else if (strcmp(arg, "--request") == 0) {

152 + return reject_request_option();

153 + }

154 + else if (strncmp(arg, "--request=", strlen("--request=")) == 0) {

155 + return reject_request_option();

156 + }

157 + else if (strcmp(arg, "--timeout") == 0) {

158 + if (++i >= argc)

159 + return set_timeout_option_once(&cmd->timeout_seconds, &cmd->timeout_seconds_set, NULL);

160 +

161 + int rc = set_timeout_option_once(&cmd->timeout_seconds, &cmd->timeout_seconds_set, argv[i]);

162 + if (rc)

163 + return rc;

164 + }

165 + else if (strncmp(arg, "--timeout=", strlen("--timeout=")) == 0) {

166 + int rc = set_timeout_option_once(

167 + &cmd->timeout_seconds, &cmd->timeout_seconds_set, arg + strlen("--timeout="));

168 + if (rc)

169 + return rc;

170 + }

171 + else if (strcmp(arg, "-h") == 0 || strcmp(arg, "--help") == 0) {

172 + systemd_journal_test_usage(stderr);

173 + return 2;

174 + }

175 + else {

176 + fprintf(stderr, "unsupported systemd journal test option '%s'\n", arg);

177 + systemd_journal_test_usage(stderr);

178 + return 2;

179 + }

180 + }

181 +

182 + if (!cmd->function_name) {

183 + fprintf(stderr, "missing required --test\n");

184 + systemd_journal_test_usage(stderr);

185 + return 2;

186 + }

187 +

188 + if (!cmd->backend_dir) {

189 + fprintf(stderr, "missing required --dir\n");

190 + systemd_journal_test_usage(stderr);

191 + return 2;

192 + }

193 +

194 + if (!cmd->timeout_seconds_set)

195 + cmd->timeout_seconds = ND_SD_JOURNAL_DEFAULT_TIMEOUT;

196 +

197 + return 0;

198 +}

199 +

200 +static uint64_t systemd_journal_effective_timeout_seconds(uint64_t timeout_seconds)

201 +{

202 + return timeout_seconds ? timeout_seconds : ND_SD_JOURNAL_TEST_TIMEOUT_DISABLED_SECONDS;

203 +}

204 +

205 +static usec_t systemd_journal_test_stop_monotonic_usec(uint64_t timeout_seconds)

206 +{

207 + usec_t now_ut = now_monotonic_usec();

208 + uint64_t effective_timeout_seconds = systemd_journal_effective_timeout_seconds(timeout_seconds);

209 + uint64_t max_timeout_seconds = (UINT64_MAX - now_ut) / USEC_PER_SEC;

210 +

211 + if (effective_timeout_seconds > max_timeout_seconds)

212 + return UINT64_MAX;

213 +

214 + return now_ut + effective_timeout_seconds * USEC_PER_SEC;

215 +}

216 +

217 +static DIR *open_systemd_journal_test_backend_directory(const char *path, char *fd_path, size_t fd_path_size)

218 +{

219 + struct stat path_st, dir_st;

220 +

221 + // Pin the explicit --dir backend root; symlinked journal trees are handled by the shared scanner.

222 + if (!path || !*path) {

223 + errno = EINVAL;

224 + return NULL;

225 + }

226 +

227 + if (lstat(path, &path_st) == -1)

228 + return NULL;

229 +

230 + if (S_ISLNK(path_st.st_mode)) {

231 + errno = ELOOP;

232 + return NULL;

233 + }

234 +

235 + if (!S_ISDIR(path_st.st_mode)) {

236 + errno = ENOTDIR;

237 + return NULL;

238 + }

239 +

240 + DIR *dir = opendir(path);

241 + if (!dir)

242 + return NULL;

243 +

244 + int fd = dirfd(dir);

245 + if (fd == -1) {

246 + int saved_errno = errno;

247 + closedir(dir);

248 + errno = saved_errno;

249 + return NULL;

250 + }

251 +

252 + if (fstat(fd, &dir_st) == -1) {

253 + int saved_errno = errno;

254 + closedir(dir);

255 + errno = saved_errno;

256 + return NULL;

257 + }

258 +

259 + if (!S_ISDIR(dir_st.st_mode)) {

260 + closedir(dir);

261 + errno = ENOTDIR;

262 + return NULL;

263 + }

264 +

265 + if (path_st.st_dev != dir_st.st_dev || path_st.st_ino != dir_st.st_ino) {

266 + closedir(dir);

267 + errno = EAGAIN;

268 + return NULL;

269 + }

270 +

271 + int written = snprintfz(fd_path, fd_path_size, "/proc/self/fd/%d", fd);

272 + if (written < 0 || (size_t)written >= fd_path_size) {

273 + closedir(dir);

274 + errno = ENAMETOOLONG;

275 + return NULL;

276 + }

277 +

278 + return dir;

279 +}

280 +

281 +static BUFFER *read_request_payload_from_stdin(void)

282 +{

283 + BUFFER *payload = buffer_create(8192, NULL);

284 + size_t total = 0;

285 + while (true) {

286 + char buffer[8192];

287 + ssize_t bytes_read = read(STDIN_FILENO, buffer, sizeof(buffer));

288 + if (bytes_read == -1) {

289 + if (errno == EINTR)

290 + continue;

291 +

292 + fprintf(stderr, "failed to read request payload from stdin: %s\n", strerror(errno));

293 + buffer_free(payload);

294 + return NULL;

295 + }

296 +

297 + if (bytes_read == 0)

298 + break;

299 +

300 + if ((uint64_t)total + (uint64_t)bytes_read > ND_SD_JOURNAL_TEST_MAX_REQUEST_BYTES) {

301 + fprintf(

302 + stderr,

303 + "request payload from stdin is too large: max %llu bytes\n",

304 + (unsigned long long)ND_SD_JOURNAL_TEST_MAX_REQUEST_BYTES);

305 + buffer_free(payload);

306 + return NULL;

307 + }

308 +

309 + buffer_memcat(payload, buffer, (size_t)bytes_read);

310 + total += (size_t)bytes_read;

311 + }

312 +

313 + if (total == 0) {

314 + fprintf(stderr, "request payload from stdin is empty\n");

315 + buffer_free(payload);

316 + return NULL;

317 + }

318 +

319 + payload->content_type = CT_APPLICATION_JSON;

320 +

321 + return payload;

322 +}

323 +

324 +static int run_systemd_journal_test_command(const struct systemd_journal_test_command *cmd)

325 +{

326 + if (strcmp(cmd->function_name, ND_SD_JOURNAL_FUNCTION_NAME) != 0) {

327 + fprintf(

328 + stderr,

329 + "unsupported systemd journal test function '%s' (expected '%s')\n",

330 + cmd->function_name,

331 + ND_SD_JOURNAL_FUNCTION_NAME);

332 + return 2;

333 + }

334 +

335 + char backend_dir_path[FILENAME_MAX];

336 + DIR *backend_dir =

337 + open_systemd_journal_test_backend_directory(cmd->backend_dir, backend_dir_path, sizeof(backend_dir_path));

338 + if (!backend_dir) {

339 + fprintf(

340 + stderr,

341 + "systemd journal backend directory '%s' cannot be opened: %s\n",

342 + cmd->backend_dir,

343 + strerror(errno));

344 + return 1;

345 + }

346 +

347 + CLEAN_BUFFER *payload = read_request_payload_from_stdin();

348 + if (!payload) {

349 + closedir(backend_dir);

350 + return 1;

351 + }

352 +

353 + bool cancelled = false;

354 + usec_t stop_monotonic_ut = systemd_journal_test_stop_monotonic_usec(cmd->timeout_seconds);

355 +

356 + nd_journal_set_scan_progress_enabled(false);

357 + nd_journal_use_single_directory(backend_dir_path);

358 + nd_journal_files_registry_update();

359 +

360 + char *function = strdupz(cmd->function_name);

361 + BUFFER *result = function_systemd_journal_result(

362 + "test", function, &stop_monotonic_ut, &cancelled, payload, HTTP_ACCESS_ALL, "test-cli", NULL);

363 + freez(function);

364 +

365 + int rc = 1;

366 + if (result) {

367 + if (buffer_strlen(result))

368 + fwrite(buffer_tostring(result), buffer_strlen(result), 1, stdout);

369 + fprintf(stdout, "\n");

370 + fflush(stdout);

371 +

372 + if (result->response_code >= HTTP_RESP_OK && result->response_code < 300)

373 + rc = 0;

374 +

375 + buffer_free(result);

376 + }

377 + else {

378 + fprintf(stderr, "systemd journal test function returned no result\n");

379 + }

380 +

381 + closedir(backend_dir);

382 + return rc;

383 +}

384 +

385 static bool journal_data_directories_exist()

386 {

387 struct stat st;

@@ -26,8 +392,13 @@ static bool journal_data_directories_exist()

392 return false;

393 }

394

29 -int main(int argc __maybe_unused, char **argv __maybe_unused)

395 +int main(int argc, char **argv)

396 {

397 + struct systemd_journal_test_command test_command = {0};

398 + int test_parse_rc = parse_systemd_journal_test_command(argc, argv, &test_command);

399 + if (test_parse_rc)

400 + exit(test_parse_rc);

401 +

402 nd_thread_tag_set("sd-jrnl.plugin");

403 nd_log_initialize_for_external_plugins("systemd-journal.plugin");

404 netdata_threads_init_for_external_plugins(0);

@@ -42,6 +413,9 @@ int main(int argc __maybe_unused, char **argv __maybe_unused)

413 nd_sd_journal_annotations_init();

414 nd_journal_init_files_and_directories();

415

416 + if (test_command.enabled)

417 + exit(run_systemd_journal_test_command(&test_command));

418 +

419 if (!journal_data_directories_exist()) {

420 nd_log_collector(NDLP_INFO, "unable to locate journal data directories. Exiting...");

421 fprintf(stdout, "DISABLE\n");

src/crates/netflow-plugin/README.md

+29

@@ -7,6 +7,35 @@ Rust NetFlow/IPFIX/sFlow ingestion and query plugin.

7 It stores flow entries in journal tiers under the Netdata cache directory and exposes

8 `flows:netflow`.

10 +## Offline Function test mode

11 +

12 +Fixture harnesses can execute the Function query path directly against an existing

13 +NetFlow backend directory:

14 +

15 +```sh

16 +netflow-plugin --test flows:netflow --dir <flows-dir> [--timeout <seconds>] [--no-persist] < payload.json

17 +```

18 +

19 +Requirements:

20 +

21 +- `<flows-dir>` is the NetFlow backend root containing the `raw`, `1m`, `5m`, and `1h` tier directories.

22 +- stdin is the JSON Function request body (non-empty, maximum 16 MiB).

23 +- `--request` is not supported and fails with usage output.

24 +- `--timeout <seconds>` controls the offline Function execution timeout. It

25 + defaults to `30`; use `--timeout 0` to map to a very large finite timeout for

26 + long-running fixture comparisons.

27 +- stdout contains only the raw JSON Function response.

28 +- errors are written to stderr and return non-zero.

29 +

30 +Use `--no-persist` for shared fixture datasets. It prevents the test run from writing

31 +facet state or sidecar files under `<flows-dir>` while keeping facet data in memory for

32 +the Function response. Without `--no-persist`, the plugin may refresh facet state under

33 +the backend directory, matching normal runtime behavior.

34 +

35 +Function output includes volatile fields such as collection timestamps and runtime

36 +statistics. Test harnesses should normalize those fields before comparing fixture

37 +outputs.

38 +

39 ## Configuration

41 When running under Netdata, config is loaded from `netflow.yaml` in:

src/crates/netflow-plugin/src/api.rs

+2 -4

@@ -1,8 +1,6 @@

1 mod flows;

3 pub(crate) use flows::NetflowFlowsHandler;

4 +pub(crate) use flows::{FLOWS_FUNCTION_NAME, FlowsFunctionResponse};

5 #[cfg(test)]

5 -pub(crate) use flows::{

6 - FLOWS_FUNCTION_VERSION, FLOWS_UPDATE_EVERY_SECONDS, FlowsFunctionResponse,

7 - flows_required_params,

8 -};

6 +pub(crate) use flows::{FLOWS_FUNCTION_VERSION, FLOWS_UPDATE_EVERY_SECONDS, flows_required_params};

src/crates/netflow-plugin/src/api/flows.rs

+2 -1

@@ -3,7 +3,8 @@ mod model;

3 mod params;

5 pub(crate) use handler::NetflowFlowsHandler;

6 +pub(crate) use model::{FLOWS_FUNCTION_NAME, FlowsFunctionResponse};

7 #[cfg(test)]

7 -pub(crate) use model::{FLOWS_FUNCTION_VERSION, FLOWS_UPDATE_EVERY_SECONDS, FlowsFunctionResponse};

8 +pub(crate) use model::{FLOWS_FUNCTION_VERSION, FLOWS_UPDATE_EVERY_SECONDS};

9 #[cfg(test)]

10 pub(crate) use params::flows_required_params;

src/crates/netflow-plugin/src/api/flows/handler.rs

+7 -5

@@ -9,9 +9,9 @@ use std::sync::Arc;

9 use tokio::task;

11 use super::model::{

12 - FLOWS_FUNCTION_VERSION, FLOWS_SCHEMA_VERSION, FLOWS_UPDATE_EVERY_SECONDS, FlowAutocompleteData,

13 - FlowAutocompleteResponse, FlowMetricsData, FlowMetricsResponse, FlowsData,

14 - FlowsFunctionResponse, FlowsResponse,

12 + FLOWS_FUNCTION_NAME, FLOWS_FUNCTION_VERSION, FLOWS_SCHEMA_VERSION, FLOWS_UPDATE_EVERY_SECONDS,

13 + FlowAutocompleteData, FlowAutocompleteResponse, FlowMetricsData, FlowMetricsResponse,

14 + FlowsData, FlowsFunctionResponse, FlowsResponse,

15 };

16 use super::params::{accepted_params, flows_required_params};

@@ -264,8 +264,10 @@ impl FunctionHandler for NetflowFlowsHandler {

264 }

265

266 fn declaration(&self) -> FunctionDeclaration {

267 - let mut func_decl =

268 - FunctionDeclaration::new("flows:netflow", "NetFlow/IPFIX/sFlow flow analysis data");

267 + let mut func_decl = FunctionDeclaration::new(

268 + FLOWS_FUNCTION_NAME,

269 + "NetFlow/IPFIX/sFlow flow analysis data",

270 + );

271 func_decl.global = true;

272 func_decl.tags = Some("flows".to_string());

273 func_decl.access =

src/crates/netflow-plugin/src/api/flows/model.rs

+11

@@ -3,6 +3,7 @@ use serde_json::Value;

3 use std::collections::HashMap;

5 pub(crate) const FLOWS_SCHEMA_VERSION: &str = "2.0";

6 +pub(crate) const FLOWS_FUNCTION_NAME: &str = "flows:netflow";

7 pub(crate) const FLOWS_FUNCTION_VERSION: u32 = 4;

8 pub(crate) const FLOWS_UPDATE_EVERY_SECONDS: u32 = 60;

@@ -128,3 +129,13 @@ pub(crate) enum FlowsFunctionResponse {

129 Metrics(FlowMetricsResponse),

130 Autocomplete(FlowAutocompleteResponse),

131 }

132 +

133 +impl FlowsFunctionResponse {

134 + pub(crate) fn status(&self) -> u32 {

135 + match self {

136 + Self::Table(response) => response.status,

137 + Self::Metrics(response) => response.status,

138 + Self::Autocomplete(response) => response.status,

139 + }

140 + }

141 +}

src/crates/netflow-plugin/src/facet_runtime.rs

+161 -23

@@ -31,7 +31,7 @@ pub(crate) use contribution::{

31 FacetFileContribution, FacetValueSink, append_record_facet_values,

32 facet_contribution_from_flow_fields,

33 };

34 -use sidecar::{delete_sidecar_files, search_sidecar, write_sidecar_files};

34 +use sidecar::{delete_sidecar_files, search_sidecar, sidecar_path, write_sidecar_files};

35 use store::{FacetStore, FacetStoreValueRef, PersistedFacetStore};

37 const FACET_STATE_VERSION: u32 = 5;

@@ -97,10 +97,31 @@ pub(crate) struct FacetRuntime {

97 ready: AtomicBool,

98 ready_notify: Notify,

99 state_path: PathBuf,

100 + persistence: FacetPersistence,

101 +}

102 +

103 +#[derive(Debug, Clone, Copy, PartialEq, Eq)]

104 +enum FacetPersistence {

105 + Enabled,

106 + ReadOnly,

107 +}

108 +

109 +impl FacetPersistence {

110 + fn writes_enabled(self) -> bool {

111 + matches!(self, Self::Enabled)

112 + }

113 }

114

115 impl FacetRuntime {

116 pub(crate) fn new(base_dir: &Path) -> Self {

117 + Self::with_persistence(base_dir, FacetPersistence::Enabled)

118 + }

119 +

120 + pub(crate) fn new_read_only(base_dir: &Path) -> Self {

121 + Self::with_persistence(base_dir, FacetPersistence::ReadOnly)

122 + }

123 +

124 + fn with_persistence(base_dir: &Path, persistence: FacetPersistence) -> Self {

125 let state_path = base_dir.join(FACET_STATE_FILE_NAME);

126 let loaded = load_persisted_state(&state_path);

127 let ready = loaded.is_some();

@@ -115,6 +136,7 @@ impl FacetRuntime {

136 ready: AtomicBool::new(ready),

137 ready_notify: Notify::new(),

138 state_path,

139 + persistence,

140 }

141 }

142

@@ -221,7 +243,9 @@ impl FacetRuntime {

243 .collect::<Vec<_>>();

244 for path in removed_archived {

245 state.indexed_archived_paths.remove(&path);

224 - delete_sidecar_files(Path::new(&path));

246 + if self.persistence.writes_enabled() {

247 + delete_sidecar_files(Path::new(&path));

248 + }

249 state.rebuild_archived = true;

250 state.dirty = true;

251 }

@@ -233,7 +257,9 @@ impl FacetRuntime {

257

258 for (path, contribution) in archived_scans {

259 merge_global_contribution(&mut state.archived_fields, &contribution);

236 - write_sidecar_files(Path::new(&path), &contribution)?;

260 + if self.persistence.writes_enabled() {

261 + write_sidecar_files(Path::new(&path), &contribution)?;

262 + }

263 if state.indexed_archived_paths.insert(path) {

264 state.dirty = true;

265 }

@@ -248,7 +274,7 @@ impl FacetRuntime {

274 state.rebuild_archived = false;

275

276 publish_locked(&self.snapshot, &state);

251 - persist_state_locked(&self.state_path, &mut state)?;

277 + self.persist_state_locked(&mut state)?;

278 drop(state);

279 self.mark_ready();

280 Ok(())

@@ -314,11 +340,13 @@ impl FacetRuntime {

340 if let Some(contribution) = contribution {

341 merge_global_contribution(&mut state.archived_fields, &contribution);

342 rebuild_published_fields(&mut state);

317 - write_sidecar_files(archived_path, &contribution)?;

343 + if self.persistence.writes_enabled() {

344 + write_sidecar_files(archived_path, &contribution)?;

345 + }

346 state.indexed_archived_paths.insert(archived_path_str);

347 state.dirty = true;

348 publish_locked(&self.snapshot, &state);

321 - persist_state_locked(&self.state_path, &mut state)?;

349 + self.persist_state_locked(&mut state)?;

350 }

351

352 Ok(())

@@ -342,7 +370,9 @@ impl FacetRuntime {

370 state.rebuild_archived = true;

371 changed = true;

372 }

345 - delete_sidecar_files(path);

373 + if self.persistence.writes_enabled() {

374 + delete_sidecar_files(path);

375 + }

376 }

377

378 if changed {

@@ -351,7 +381,7 @@ impl FacetRuntime {

381 publish_locked(&self.snapshot, &state);

382 }

383 state.dirty = true;

354 - persist_state_locked(&self.state_path, &mut state)?;

384 + self.persist_state_locked(&mut state)?;

385 }

386

387 Ok(())

@@ -364,7 +394,7 @@ impl FacetRuntime {

394 };

395 let match_kind = spec.autocomplete_match;

396

367 - let (promoted, mut matches, archived_paths) = {

397 + let (use_sidecars, mut matches, archived_paths) = {

398 let state = self

399 .state

400 .lock()

@@ -378,7 +408,10 @@ impl FacetRuntime {

408 term,

409 match_kind,

410 );

381 - let archived_matches = if !spec.uses_sidecar || !published.autocomplete {

411 + let use_sidecars = spec.uses_sidecar && published.autocomplete;

412 + let archived_matches = if use_sidecars {

413 + Vec::new()

414 + } else {

415 state

416 .archived_fields

417 .get(normalized.as_str())

@@ -386,8 +419,6 @@ impl FacetRuntime {

419 store.autocomplete_matches(term, FACET_AUTOCOMPLETE_LIMIT, match_kind)

420 })

421 .unwrap_or_default()

389 - } else {

390 - Vec::new()

422 };

423 let archived_paths = state

424 .indexed_archived_paths

@@ -395,30 +426,49 @@ impl FacetRuntime {

426 .cloned()

427 .collect::<Vec<_>>();

428 (

398 - published.autocomplete,

429 + use_sidecars,

430 merge_autocomplete_values(active_matches, archived_matches),

431 archived_paths,

432 )

433 };

434

404 - if spec.uses_sidecar && promoted && matches.len() < FACET_AUTOCOMPLETE_LIMIT {

405 - for path in archived_paths {

435 + if use_sidecars && matches.len() < FACET_AUTOCOMPLETE_LIMIT {

436 + let mut missing_sidecar = false;

437 + for path in &archived_paths {

438 + let journal_path = Path::new(path);

439 + if !sidecar_path(journal_path, normalized.as_str()).exists() {

440 + missing_sidecar = true;

441 + continue;

442 + }

443 +

444 let needed = FACET_AUTOCOMPLETE_LIMIT.saturating_sub(matches.len());

445 if needed == 0 {

446 break;

447 }

410 - let sidecar_matches = search_sidecar(

411 - Path::new(&path),

412 - normalized.as_str(),

413 - term,

414 - needed,

415 - match_kind,

416 - )?;

448 + let sidecar_matches =

449 + search_sidecar(journal_path, normalized.as_str(), term, needed, match_kind)?;

450 matches = merge_autocomplete_values(matches, sidecar_matches);

451 if matches.len() >= FACET_AUTOCOMPLETE_LIMIT {

452 break;

453 }

454 }

455 +

456 + if missing_sidecar && matches.len() < FACET_AUTOCOMPLETE_LIMIT {

457 + let archived_matches = {

458 + let state = self

459 + .state

460 + .lock()

461 + .map_err(|_| anyhow::anyhow!("facet runtime lock poisoned"))?;

462 + state

463 + .archived_fields

464 + .get(normalized.as_str())

465 + .map(|store| {

466 + store.autocomplete_matches(term, FACET_AUTOCOMPLETE_LIMIT, match_kind)

467 + })

468 + .unwrap_or_default()

469 + };

470 + matches = merge_autocomplete_values(matches, archived_matches);

471 + }

472 }

473

474 Ok(matches)

@@ -429,7 +479,7 @@ impl FacetRuntime {

479 .state

480 .lock()

481 .map_err(|_| anyhow::anyhow!("facet runtime lock poisoned"))?;

432 - persist_state_locked(&self.state_path, &mut state)

482 + self.persist_state_locked(&mut state)

483 }

484

485 fn mark_ready(&self) {

@@ -438,6 +488,15 @@ impl FacetRuntime {

488 self.ready_notify.notify_waiters();

489 }

490 }

491 +

492 + fn persist_state_locked(&self, state: &mut FacetState) -> Result<()> {

493 + if self.persistence.writes_enabled() {

494 + persist_state_locked(&self.state_path, state)

495 + } else {

496 + state.dirty = false;

497 + Ok(())

498 + }

499 + }

500 }

501

502 impl FacetState {

@@ -1217,6 +1276,85 @@ mod tests {

1276 );

1277 }

1278

1279 + #[test]

1280 + fn read_only_runtime_does_not_write_state_or_sidecars() {

1281 + let tmp = tempfile::tempdir().expect("create temp dir");

1282 + let runtime = FacetRuntime::new_read_only(tmp.path());

1283 + let archived_path = tmp.path().join("flows-promoted.journal");

1284 +

1285 + for value in 0..120 {

1286 + let mut fields = FlowFields::new();

1287 + fields.insert("SRC_AS_NAME", format!("AS{value:03} EXAMPLE"));

1288 + let contribution = facet_contribution_from_flow_fields(&fields);

1289 + runtime

1290 + .observe_active_contribution(&archived_path, &contribution)

1291 + .expect("observe contribution");

1292 + }

1293 +

1294 + runtime

1295 + .observe_rotation(

1296 + &archived_path,

1297 + &tmp.path().join("flows-promoted-next.journal"),

1298 + )

1299 + .expect("rotate file");

1300 +

1301 + assert!(

1302 + !tmp.path().join(FACET_STATE_FILE_NAME).exists(),

1303 + "read-only facet runtime must not write facet state"

1304 + );

1305 + assert!(

1306 + !super::sidecar::sidecar_path(&archived_path, "SRC_AS_NAME").exists(),

1307 + "read-only facet runtime must not write sidecars"

1308 + );

1309 +

1310 + let results = runtime

1311 + .autocomplete("SRC_AS_NAME", "AS11")

1312 + .expect("autocomplete values");

1313 +

1314 + assert!(

1315 + results.iter().any(|value| value == "AS110 EXAMPLE"),

1316 + "read-only mode should search promoted archived values from memory"

1317 + );

1318 + }

1319 +

1320 + #[test]

1321 + fn read_only_runtime_reads_existing_sidecar_files() {

1322 + let tmp = tempfile::tempdir().expect("create temp dir");

1323 + let archived_path = tmp.path().join("flows-promoted.journal");

1324 + let runtime = FacetRuntime::new(tmp.path());

1325 +

1326 + for value in 0..120 {

1327 + let mut fields = FlowFields::new();

1328 + fields.insert("SRC_AS_NAME", format!("AS{value:03} EXAMPLE"));

1329 + let contribution = facet_contribution_from_flow_fields(&fields);

1330 + runtime

1331 + .observe_active_contribution(&archived_path, &contribution)

1332 + .expect("observe contribution");

1333 + }

1334 +

1335 + runtime

1336 + .observe_rotation(

1337 + &archived_path,

1338 + &tmp.path().join("flows-promoted-next.journal"),

1339 + )

1340 + .expect("rotate file");

1341 +

1342 + let runtime = FacetRuntime::new_read_only(tmp.path());

1343 + {

1344 + let mut state = runtime.state.lock().expect("lock facet state");

1345 + state.archived_fields.clear();

1346 + }

1347 +

1348 + let results = runtime

1349 + .autocomplete("SRC_AS_NAME", "AS11")

1350 + .expect("autocomplete values");

1351 +

1352 + assert!(

1353 + results.iter().any(|value| value == "AS110 EXAMPLE"),

1354 + "read-only mode should search existing sidecar files"

1355 + );

1356 + }

1357 +

1358 #[test]

1359 fn runtime_autocomplete_text_field_uses_substring_matching() {

1360 let tmp = tempfile::tempdir().expect("create temp dir");

src/crates/netflow-plugin/src/main.rs

+37 -6

@@ -22,6 +22,7 @@ mod rollup;

22 mod routing;

23 #[cfg(test)]

24 mod startup_memory_tests;

25 +mod test_cli;

26 mod tiering;

28 pub(crate) use api::NetflowFlowsHandler;

@@ -46,6 +47,30 @@ fn main() {

47 std::process::exit(1);

48 }

50 + match test_cli::TestCommand::parse_from_env_args() {

51 + Ok(Some(command)) => {

52 + let worker_threads = runtime_worker_threads();

53 + let max_blocking_threads = runtime_blocking_threads(worker_threads);

54 + let runtime = match build_tokio_runtime(worker_threads, max_blocking_threads) {

55 + Ok(runtime) => runtime,

56 + Err(err) => {

57 + eprintln!("failed to build tokio runtime: {}", err);

58 + std::process::exit(1);

59 + }

60 + };

61 + if let Err(err) = runtime.block_on(test_cli::run(command)) {

62 + eprintln!("{err:#}");

63 + std::process::exit(1);

64 + }

65 + return;

66 + }

67 + Ok(None) => {}

68 + Err(err) => {

69 + eprintln!("{err}");

70 + std::process::exit(2);

71 + }

72 + }

73 +

74 #[cfg(all(target_os = "linux", target_env = "gnu"))]

75 let glibc_arena_max = memory_allocator::limit_glibc_arenas_for_process();

@@ -79,12 +104,7 @@ fn main() {

104 "configured netflow tokio runtime"

105 );

106

82 - let runtime = match tokio::runtime::Builder::new_multi_thread()

83 - .enable_all()

84 - .worker_threads(worker_threads)

85 - .max_blocking_threads(max_blocking_threads)

86 - .build()

87 - {

107 + let runtime = match build_tokio_runtime(worker_threads, max_blocking_threads) {

108 Ok(runtime) => runtime,

109 Err(err) => {

110 eprintln!("failed to build tokio runtime: {}", err);

@@ -370,6 +390,17 @@ fn runtime_blocking_threads(worker_threads: usize) -> usize {

390 MIN_RUNTIME_BLOCKING_THREADS.max(worker_threads)

391 }

392

393 +fn build_tokio_runtime(

394 + worker_threads: usize,

395 + max_blocking_threads: usize,

396 +) -> std::io::Result<tokio::runtime::Runtime> {

397 + tokio::runtime::Builder::new_multi_thread()

398 + .enable_all()

399 + .worker_threads(worker_threads)

400 + .max_blocking_threads(max_blocking_threads)

401 + .build()

402 +}

403 +

404 #[cfg(test)]

405 #[path = "main_tests.rs"]

406 mod tests;

src/crates/netflow-plugin/src/plugin_config/runtime.rs

+15

@@ -20,6 +20,21 @@ impl PluginConfig {

20 Ok(cfg)

21 }

23 + pub(crate) fn for_test_backend_dir(backend_dir: &Path) -> Result<Self> {

24 + let mut cfg = Self::default();

25 + cfg.journal.journal_dir = backend_dir

26 + .to_str()

27 + .with_context(|| {

28 + format!(

29 + "netflow test backend directory {} is not valid UTF-8",

30 + backend_dir.display()

31 + )

32 + })?

33 + .to_string();

34 + cfg.validate()?;

35 + Ok(cfg)

36 + }

37 +

38 pub(super) fn auto_detect_geoip_databases(&mut self) {

39 let intel_dirs = [

40 self.inferred_cache_dir().join(TOPOLOGY_IP_INTEL_DIR),

src/crates/netflow-plugin/src/test_cli.rs new

+621

@@ -0,0 +1,621 @@

1 +use crate::api::{FLOWS_FUNCTION_NAME, FlowsFunctionResponse, NetflowFlowsHandler};

2 +use crate::{facet_runtime, ingest, plugin_config, query};

3 +use anyhow::{Context, Result, bail};

4 +use rt::ProgressState;

5 +use std::ffi::{OsStr, OsString};

6 +use std::io::{self, Read, Write};

7 +use std::path::PathBuf;

8 +use std::sync::Arc;

9 +use std::time::Duration;

10 +use tokio_util::sync::CancellationToken;

11 +

12 +const USAGE: &str = "usage: netflow-plugin --test flows:netflow --dir <flows-dir> [--timeout <seconds>] [--no-persist] < payload.json";

13 +const DEFAULT_TIMEOUT_SECONDS: u64 = 30;

14 +const DISABLED_TIMEOUT_SECONDS: u64 = 100 * 365 * 24 * 60 * 60;

15 +const MAX_REQUEST_BYTES: u64 = 16 * 1024 * 1024;

16 +

17 +#[derive(Debug, Clone, PartialEq, Eq)]

18 +pub(crate) struct TestCommand {

19 + pub(crate) function_name: String,

20 + pub(crate) backend_dir: PathBuf,

21 + pub(crate) timeout_seconds: u64,

22 + pub(crate) no_persist: bool,

23 +}

24 +

25 +impl TestCommand {

26 + pub(crate) fn parse_from_env_args() -> std::result::Result<Option<Self>, String> {

27 + parse_from_os(std::env::args_os().skip(1))

28 + }

29 +}

30 +

31 +pub(crate) async fn run(command: TestCommand) -> Result<()> {

32 + let request_bytes = read_request_payload_from_stdin(io::stdin().lock())?;

33 + let response = execute(command, &request_bytes).await?;

34 + let stdout = io::stdout();

35 + let mut handle = stdout.lock();

36 + write_json_response(&response, &mut handle)?;

37 + ensure_success_status(response.status())

38 +}

39 +

40 +pub(crate) async fn execute(

41 + command: TestCommand,

42 + request_bytes: &[u8],

43 +) -> Result<FlowsFunctionResponse> {

44 + if command.function_name != FLOWS_FUNCTION_NAME {

45 + bail!(

46 + "unsupported netflow test function `{}` (expected `{}`)",

47 + command.function_name,

48 + FLOWS_FUNCTION_NAME

49 + );

50 + }

51 +

52 + let effective_timeout = effective_timeout_seconds(command.timeout_seconds);

53 + let cancellation = CancellationToken::new();

54 + match tokio::time::timeout(

55 + Duration::from_secs(effective_timeout),

56 + execute_inner(command, request_bytes, cancellation.clone()),

57 + )

58 + .await

59 + {

60 + Ok(result) => result,

61 + Err(_) => {

62 + cancellation.cancel();

63 + bail!(

64 + "netflow test function timed out after {} seconds",

65 + effective_timeout

66 + );

67 + }

68 + }

69 +}

70 +

71 +async fn execute_inner(

72 + command: TestCommand,

73 + request_bytes: &[u8],

74 + cancellation: CancellationToken,

75 +) -> Result<FlowsFunctionResponse> {

76 + let request = serde_json::from_slice::<query::FlowsRequest>(request_bytes)

77 + .context("failed to parse request payload from stdin")?;

78 +

79 + let config = plugin_config::PluginConfig::for_test_backend_dir(&command.backend_dir)?;

80 + ensure_tier_directories_exist(&config)?;

81 +

82 + let facet_runtime = if command.no_persist {

83 + facet_runtime::FacetRuntime::new_read_only(&config.journal.base_dir())

84 + } else {

85 + facet_runtime::FacetRuntime::new(&config.journal.base_dir())

86 + };

87 + let facet_runtime = Arc::new(facet_runtime);

88 + let (query_service, _notify_rx) =

89 + query::FlowQueryService::new_with_facet_runtime(&config, Arc::clone(&facet_runtime))

90 + .await?;

91 + let query_service = Arc::new(query_service);

92 + query_service.initialize_facets().await?;

93 +

94 + let handler = NetflowFlowsHandler::new(

95 + Arc::new(ingest::IngestMetrics::default()),

96 + Arc::clone(&query_service),

97 + );

98 + let execution = if request.is_autocomplete_mode() {

99 + None

100 + } else {

101 + Some(query::QueryExecutionContext::new(

102 + ProgressState::default(),

103 + cancellation.clone(),

104 + ))

105 + };

106 +

107 + handler

108 + .handle_request_with_execution(execution, request)

109 + .await

110 + .map_err(Into::into)

111 +}

112 +

113 +pub(crate) fn write_json_response(

114 + response: &FlowsFunctionResponse,

115 + writer: &mut impl Write,

116 +) -> Result<()> {

117 + serde_json::to_writer(&mut *writer, response).context("failed to write JSON response")?;

118 + writer

119 + .write_all(b"\n")

120 + .context("failed to finish JSON response")?;

121 + writer.flush().context("failed to flush JSON response")

122 +}

123 +

124 +fn ensure_tier_directories_exist(config: &plugin_config::PluginConfig) -> Result<()> {

125 + for tier_dir in config.journal.all_tier_dirs() {

126 + if !tier_dir.is_dir() {

127 + bail!(

128 + "netflow backend tier directory {} does not exist",

129 + tier_dir.display()

130 + );

131 + }

132 + }

133 + Ok(())

134 +}

135 +

136 +fn effective_timeout_seconds(timeout_seconds: u64) -> u64 {

137 + if timeout_seconds == 0 {

138 + DISABLED_TIMEOUT_SECONDS

139 + } else {

140 + timeout_seconds

141 + }

142 +}

143 +

144 +fn ensure_success_status(status: u32) -> Result<()> {

145 + if !(200..300).contains(&status) {

146 + bail!("netflow test function returned status {status}");

147 + }

148 +

149 + Ok(())

150 +}

151 +

152 +fn read_request_payload_from_stdin(reader: impl Read) -> Result<Vec<u8>> {

153 + let mut request_bytes = Vec::new();

154 + reader

155 + .take(MAX_REQUEST_BYTES + 1)

156 + .read_to_end(&mut request_bytes)

157 + .context("failed to read request payload from stdin")?;

158 +

159 + if request_bytes.is_empty() {

160 + bail!("request payload from stdin is empty");

161 + }

162 +

163 + if request_bytes.len() as u64 > MAX_REQUEST_BYTES {

164 + bail!(

165 + "request payload from stdin is too large: more than {} bytes",

166 + MAX_REQUEST_BYTES

167 + );

168 + }

169 +

170 + Ok(request_bytes)

171 +}

172 +

173 +#[cfg(test)]

174 +fn parse_from(

175 + args: impl IntoIterator<Item = String>,

176 +) -> std::result::Result<Option<TestCommand>, String> {

177 + parse_from_os(args.into_iter().map(OsString::from))

178 +}

179 +

180 +fn parse_from_os(

181 + args: impl IntoIterator<Item = OsString>,

182 +) -> std::result::Result<Option<TestCommand>, String> {

183 + let args = args.into_iter().collect::<Vec<_>>();

184 + if !args

185 + .iter()

186 + .any(|arg| arg == OsStr::new("--test") || strip_os_prefix(arg, "--test=").is_some())

187 + {

188 + return Ok(None);

189 + }

190 +

191 + let mut function_name = None;

192 + let mut backend_dir = None;

193 + let mut timeout_seconds = None;

194 + let mut no_persist = false;

195 + let mut idx = 0;

196 +

197 + while idx < args.len() {

198 + let arg = &args[idx];

199 + if arg == OsStr::new("--test") {

200 + idx += 1;

201 + let value = args

202 + .get(idx)

203 + .ok_or_else(|| format!("missing value for --test\n{USAGE}"))?;

204 + set_once(

205 + &mut function_name,

206 + required_os_string_value(value, "--test")?,

207 + "--test",

208 + )?;

209 + } else if arg == OsStr::new("--dir") {

210 + idx += 1;

211 + let value = args

212 + .get(idx)

213 + .ok_or_else(|| format!("missing value for --dir\n{USAGE}"))?;

214 + set_once(

215 + &mut backend_dir,

216 + PathBuf::from(required_os_string_value(value, "--dir")?),

217 + "--dir",

218 + )?;

219 + } else if arg == OsStr::new("--request") {

220 + return Err(format!(

221 + "--request is no longer supported; pass the request payload on stdin\n{USAGE}"

222 + ));

223 + } else if arg == OsStr::new("--timeout") {

224 + idx += 1;

225 + let value = args

226 + .get(idx)

227 + .ok_or_else(|| format!("missing value for --timeout\n{USAGE}"))?;

228 + set_once(

229 + &mut timeout_seconds,

230 + parse_timeout_seconds(&required_os_string_value(value, "--timeout")?)?,

231 + "--timeout",

232 + )?;

233 + } else if arg == OsStr::new("--no-persist") {

234 + if no_persist {

235 + return Err(format!("duplicate --no-persist\n{USAGE}"));

236 + }

237 + no_persist = true;

238 + } else if let Some(value) = strip_os_prefix(arg, "--test=") {

239 + set_once(

240 + &mut function_name,

241 + required_os_string_value(&value, "--test")?,

242 + "--test",

243 + )?;

244 + } else if let Some(value) = strip_os_prefix(arg, "--dir=") {

245 + set_once(

246 + &mut backend_dir,

247 + PathBuf::from(required_os_string_value(&value, "--dir")?),

248 + "--dir",

249 + )?;

250 + } else if strip_os_prefix(arg, "--request=").is_some() {

251 + return Err(format!(

252 + "--request is no longer supported; pass the request payload on stdin\n{USAGE}"

253 + ));

254 + } else if let Some(value) = strip_os_prefix(arg, "--timeout=") {

255 + set_once(

256 + &mut timeout_seconds,

257 + parse_timeout_seconds(&required_os_string_value(&value, "--timeout")?)?,

258 + "--timeout",

259 + )?;

260 + } else if arg == OsStr::new("-h") || arg == OsStr::new("--help") {

261 + return Err(USAGE.to_string());

262 + } else {

263 + return Err(format!(

264 + "unsupported netflow test option `{}`\n{USAGE}",

265 + arg.to_string_lossy()

266 + ));

267 + }

268 + idx += 1;

269 + }

270 +

271 + Ok(Some(TestCommand {

272 + function_name: required(function_name, "--test")?,

273 + backend_dir: required(backend_dir, "--dir")?,

274 + timeout_seconds: timeout_seconds.unwrap_or(DEFAULT_TIMEOUT_SECONDS),

275 + no_persist,

276 + }))

277 +}

278 +

279 +fn set_once<T>(slot: &mut Option<T>, value: T, option: &str) -> std::result::Result<(), String> {

280 + if slot.is_some() {

281 + return Err(format!("duplicate {option}\n{USAGE}"));

282 + }

283 + *slot = Some(value);

284 + Ok(())

285 +}

286 +

287 +fn required_os_option_value<'a>(

288 + value: &'a OsStr,

289 + option: &str,

290 +) -> std::result::Result<&'a OsStr, String> {

291 + if value.is_empty() {

292 + return Err(format!("missing value for {option}\n{USAGE}"));

293 + }

294 +

295 + Ok(value)

296 +}

297 +

298 +fn required_os_string_value(value: &OsStr, option: &str) -> std::result::Result<String, String> {

299 + let value = required_os_option_value(value, option)?;

300 + value

301 + .to_str()

302 + .map(ToString::to_string)

303 + .ok_or_else(|| format!("invalid value for {option}; expected UTF-8\n{USAGE}"))

304 +}

305 +

306 +fn required<T>(slot: Option<T>, option: &str) -> std::result::Result<T, String> {

307 + slot.ok_or_else(|| format!("missing required {option}\n{USAGE}"))

308 +}

309 +

310 +fn parse_timeout_seconds(value: &str) -> std::result::Result<u64, String> {

311 + value

312 + .parse::<u64>()

313 + .map_err(|_| format!("invalid value for --timeout `{value}`; expected seconds\n{USAGE}"))

314 +}

315 +

316 +#[cfg(unix)]

317 +fn strip_os_prefix(value: &OsStr, prefix: &str) -> Option<OsString> {

318 + use std::os::unix::ffi::{OsStrExt, OsStringExt};

319 +

320 + value

321 + .as_bytes()

322 + .strip_prefix(prefix.as_bytes())

323 + .map(|suffix| OsString::from_vec(suffix.to_vec()))

324 +}

325 +

326 +#[cfg(not(unix))]

327 +fn strip_os_prefix(value: &OsStr, prefix: &str) -> Option<OsString> {

328 + value.to_str()?.strip_prefix(prefix).map(OsString::from)

329 +}

330 +

331 +#[cfg(test)]

332 +mod tests {

333 + use super::*;

334 + use std::fs;

335 +

336 + fn parse(args: &[&str]) -> std::result::Result<Option<TestCommand>, String> {

337 + parse_from(args.iter().map(|arg| (*arg).to_string()))

338 + }

339 +

340 + #[test]

341 + fn parser_ignores_normal_plugin_arguments_when_test_is_absent() {

342 + assert_eq!(parse(&["--netflow-enabled", "false"]).unwrap(), None);

343 + }

344 +

345 + #[cfg(unix)]

346 + #[test]

347 + fn parser_ignores_non_utf8_normal_arguments_when_test_is_absent() {

348 + use std::ffi::OsString;

349 + use std::os::unix::ffi::OsStringExt;

350 +

351 + let args = [

352 + OsString::from("--netflow-enabled"),

353 + OsString::from_vec(vec![0xff]),

354 + ];

355 +

356 + assert_eq!(parse_from_os(args).unwrap(), None);

357 + }

358 +

359 + #[cfg(unix)]

360 + #[test]

361 + fn parser_rejects_non_utf8_spaced_backend_dir() {

362 + use std::ffi::OsString;

363 + use std::os::unix::ffi::OsStringExt;

364 +

365 + let err = parse_from_os([

366 + OsString::from("--test"),

367 + OsString::from("flows:netflow"),

368 + OsString::from("--dir"),

369 + OsString::from_vec(b"flows-\xff".to_vec()),

370 + ])

371 + .expect_err("non-UTF-8 backend directory should fail");

372 +

373 + assert!(err.contains("invalid value for --dir; expected UTF-8"));

374 + }

375 +

376 + #[cfg(unix)]

377 + #[test]

378 + fn parser_rejects_non_utf8_equals_backend_dir() {

379 + use std::ffi::OsString;

380 + use std::os::unix::ffi::OsStringExt;

381 +

382 + let mut dir_arg = b"--dir=".to_vec();

383 + dir_arg.extend_from_slice(b"flows-\xff");

384 +

385 + let err = parse_from_os([

386 + OsString::from("--test=flows:netflow"),

387 + OsString::from_vec(dir_arg),

388 + ])

389 + .expect_err("non-UTF-8 backend directory should fail");

390 +

391 + assert!(err.contains("invalid value for --dir; expected UTF-8"));

392 + }

393 +

394 + #[test]

395 + fn parser_accepts_stdin_test_command() {

396 + let command = parse(&["--test", "flows:netflow", "--dir", "flows", "--no-persist"])

397 + .unwrap()

398 + .expect("test command");

399 +

400 + assert_eq!(command.function_name, "flows:netflow");

401 + assert_eq!(command.backend_dir, PathBuf::from("flows"));

402 + assert_eq!(command.timeout_seconds, DEFAULT_TIMEOUT_SECONDS);

403 + assert!(command.no_persist);

404 + }

405 +

406 + #[test]

407 + fn parser_accepts_equals_form() {

408 + let command = parse(&["--test=flows:netflow", "--dir=flows", "--timeout=0"])

409 + .unwrap()

410 + .expect("test command");

411 +

412 + assert_eq!(command.function_name, "flows:netflow");

413 + assert_eq!(command.backend_dir, PathBuf::from("flows"));

414 + assert_eq!(command.timeout_seconds, 0);

415 + assert_eq!(

416 + effective_timeout_seconds(command.timeout_seconds),

417 + DISABLED_TIMEOUT_SECONDS

418 + );

419 + assert!(!command.no_persist);

420 + }

421 +

422 + #[test]

423 + fn parser_accepts_spaced_timeout() {

424 + let command = parse(&[

425 + "--test",

426 + "flows:netflow",

427 + "--dir",

428 + "flows",

429 + "--timeout",

430 + "120",

431 + ])

432 + .unwrap()

433 + .expect("test command");

434 +

435 + assert_eq!(command.timeout_seconds, 120);

436 + assert_eq!(effective_timeout_seconds(command.timeout_seconds), 120);

437 + }

438 +

439 + #[test]

440 + fn parser_rejects_missing_required_options() {

441 + let err = parse(&["--test", "flows:netflow"])

442 + .expect_err("missing options should fail")

443 + .to_string();

444 + assert!(err.contains("missing required --dir"));

445 + }

446 +

447 + #[test]

448 + fn parser_rejects_empty_required_values() {

449 + for args in [

450 + ["--test=", "--dir=flows"].as_slice(),

451 + ["--test", "", "--dir=flows"].as_slice(),

452 + ["--test=flows:netflow", "--dir="].as_slice(),

453 + ["--test=flows:netflow", "--dir", ""].as_slice(),

454 + ] {

455 + let err = parse(args).expect_err("empty required option should fail");

456 + assert!(err.contains("missing value for --"));

457 + }

458 + }

459 +

460 + #[test]

461 + fn parser_rejects_request_option() {

462 + for args in [

463 + [

464 + "--test=flows:netflow",

465 + "--dir=flows",

466 + "--request=payload.json",

467 + ]

468 + .as_slice(),

469 + [

470 + "--test=flows:netflow",

471 + "--dir=flows",

472 + "--request",

473 + "payload.json",

474 + ]

475 + .as_slice(),

476 + ] {

477 + let err = parse(args).expect_err("--request should fail");

478 + assert!(err.contains("--request is no longer supported"));

479 + }

480 + }

481 +

482 + #[test]

483 + fn parser_rejects_unknown_test_options() {

484 + let err = parse(&["--test", "flows:netflow", "--dir", "flows", "--unknown"])

485 + .expect_err("unknown option should fail");

486 +

487 + assert!(err.contains("unsupported netflow test option"));

488 + }

489 +

490 + #[test]

491 + fn parser_rejects_invalid_timeout() {

492 + let err = parse(&[

493 + "--test",

494 + "flows:netflow",

495 + "--dir",

496 + "flows",

497 + "--timeout",

498 + "-1",

499 + ])

500 + .expect_err("invalid timeout should fail");

501 +

502 + assert!(err.contains("invalid value for --timeout"));

503 + }

504 +

505 + #[test]

506 + fn parser_rejects_duplicate_timeout() {

507 + let err = parse(&[

508 + "--test",

509 + "flows:netflow",

510 + "--dir",

511 + "flows",

512 + "--timeout",

513 + "30",

514 + "--timeout=60",

515 + ])

516 + .expect_err("duplicate timeout should fail");

517 +

518 + assert!(err.contains("duplicate --timeout"));

519 + }

520 +

521 + #[test]

522 + fn success_status_accepts_2xx() {

523 + ensure_success_status(200).expect("200 should pass");

524 + ensure_success_status(299).expect("299 should pass");

525 + }

526 +

527 + #[test]

528 + fn success_status_rejects_non_2xx() {

529 + let err = ensure_success_status(400).expect_err("400 should fail");

530 + assert!(err.to_string().contains("returned status 400"));

531 + }

532 +

533 + #[test]

534 + fn read_request_payload_from_stdin_rejects_empty_input() {

535 + let err = read_request_payload_from_stdin(io::Cursor::new(Vec::new()))

536 + .expect_err("empty request should fail");

537 + assert!(err.to_string().contains("is empty"));

538 + }

539 +

540 + #[test]

541 + fn read_request_payload_from_stdin_rejects_oversized_input() {

542 + let oversized = vec![b' '; (MAX_REQUEST_BYTES + 1) as usize];

543 + let err = read_request_payload_from_stdin(io::Cursor::new(oversized))

544 + .expect_err("oversized request should fail");

545 + assert!(err.to_string().contains("is too large"));

546 + }

547 +

548 + #[test]

549 + fn read_request_payload_from_stdin_reads_payload() {

550 + let request = read_request_payload_from_stdin(io::Cursor::new(br#"{"after":1}"#))

551 + .expect("read request payload");

552 + assert_eq!(request, br#"{"after":1}"#);

553 + }

554 +

555 + #[tokio::test(flavor = "multi_thread", worker_threads = 2)]

556 + async fn execute_empty_backend_writes_raw_json_without_persisting_facets() {

557 + let tmp = tempfile::tempdir().expect("create temp dir");

558 + let backend_dir = tmp.path().join("flows");

559 + for tier in ["raw", "1m", "5m", "1h"] {

560 + fs::create_dir_all(backend_dir.join(tier)).expect("create tier dir");

561 + }

562 +

563 + let response = execute(

564 + TestCommand {

565 + function_name: FLOWS_FUNCTION_NAME.to_string(),

566 + backend_dir: backend_dir.clone(),

567 + timeout_seconds: DEFAULT_TIMEOUT_SECONDS,

568 + no_persist: true,

569 + },

570 + br#"{"after":1,"before":2,"group_by":["PROTOCOL"],"top_n":"100"}"#,

571 + )

572 + .await

573 + .expect("execute test CLI request");

574 +

575 + let mut stdout = Vec::new();

576 + write_json_response(&response, &mut stdout).expect("write JSON response");

577 + let output = String::from_utf8(stdout).expect("stdout should be UTF-8 JSON");

578 +

579 + assert!(

580 + output.trim_start().starts_with('{'),

581 + "test CLI stdout should start with JSON object, got {output:?}"

582 + );

583 + assert!(

584 + !output.starts_with("TRUST_DURATIONS"),

585 + "test CLI stdout must not include PLUGINSD protocol lines"

586 + );

587 +

588 + let value = serde_json::from_str::<serde_json::Value>(&output).expect("parse JSON output");

589 + assert_eq!(value["status"], 200);

590 + assert_eq!(value["type"], "flows");

591 + assert!(

592 + value["data"]["flows"].as_array().is_some_and(Vec::is_empty),

593 + "empty backend should return an empty flows array"

594 + );

595 + assert!(

596 + !backend_dir.join("facet-state.bin").exists(),

597 + "--no-persist must not write facet state under the fixture backend"

598 + );

599 + }

600 +

601 + #[cfg(unix)]

602 + #[tokio::test(flavor = "multi_thread", worker_threads = 2)]

603 + async fn execute_rejects_non_utf8_backend_dir() {

604 + use std::ffi::OsString;

605 + use std::os::unix::ffi::OsStringExt;

606 +

607 + let err = execute(

608 + TestCommand {

609 + function_name: FLOWS_FUNCTION_NAME.to_string(),

610 + backend_dir: PathBuf::from(OsString::from_vec(b"flows-\xff".to_vec())),

611 + timeout_seconds: DEFAULT_TIMEOUT_SECONDS,

612 + no_persist: true,

613 + },

614 + br#"{"after":1,"before":2,"group_by":["PROTOCOL"],"top_n":"100"}"#,

615 + )

616 + .await

617 + .expect_err("non-UTF-8 backend directory should fail");

618 +

619 + assert!(err.to_string().contains("is not valid UTF-8"));

620 + }

621 +}