Skip to content

out_manticore: add Manticore Search output plugin - #3

Closed
tomatolog wants to merge 3 commits into
masterfrom
feat/out_plugin_manticore
Closed

tomatolog wants to merge 3 commits into
masterfrom
feat/out_plugin_manticore

Conversation

@tomatolog

Copy link
Copy Markdown

This adds out_manticore, an output that writes log records to a Manticore Search table through its HTTP /bulk endpoint, as NDJSON. A record goes as it is: the daemon takes its id field as the document id or generates one.

Regular mode (the default) sends every chunk as one /bulk request of insert or create operations, which Manticore runs as one transaction. A retryable item status, 408, 429 or a 5xx reply without item statuses retries the chunk; any other failure drops it.

Shared mode loads a finite source, for example a file read with Exit_On_Eof, into one table with Manticore's bulk_import. Data requests stage the records, Shared_Indexers sets how many indexers the daemon starts for the load, and one Finish request sent from the exit callback publishes it. The load is published only if every chunk was delivered; otherwise the log names the statement that cancels the staged data. Shared needs a Manticore build with shared bulk_import (dev packages since 29.15.0).

Options: table (required), mode (Regular or Shared), action (insert or create, Regular), shared_indexers (Shared), buffer_size, http_user, http_passwd, and the usual network, TLS and workers options.

Tests:

  • tests/runtime/out_manticore.c: the formatter output of Regular insert and create and of Shared, for records with and without an id, and settings that fail the start;
  • tests/integration/scenarios/out_manticore: a local receiver in place of /bulk checks the Regular requests, a retried and a dropped item; the requests of a Shared load and its Finish at exit, the Finish over TLS with workers 0, retried and malformed replies, a failed Finish, and loads that end with pending, exhausted or failed data and so are not published.

Both pass locally; the integration scenario also under VALGRIND=1 VALGRIND_STRICT=1.


Enter [N/A] in the box, if an item is not applicable to your change.

Testing
Before we can approve your change; please submit the following in a comment:

  • Example configuration file for the change
  • Debug log output from testing the change
  • Attached Valgrind output that shows no leaks or memory corruption was found

If this is a change to packaging of containers or native binaries then please confirm it works for all targets.

  • [N/A] Run local packaging test showing all targets (including any new ones) build.
  • [N/A] Set ok-package-test label to test for all targets (requires maintainer to do).

Documentation

  • Documentation required for this feature

Doc PR: to follow

Backporting

  • [N/A] Backport to latest stable release.

Example configuration

Manticore Search 29.15.0 dev (manticoresearch/manticore:dev-29.15.0-13685b0, the first build with shared bulk_import) runs as the container flb-pr-manticore on the Docker network of Fluent Bit, with the tables created first:

CREATE TABLE logs (message TEXT, status INTEGER);
CREATE TABLE shared_logs (message TEXT, status INTEGER);

/out/records.json holds 1000 lines like {"id": 1, "message": "record 1 from Fluent Bit", "status": 201}, and /out/parsers.conf defines a json parser.

Regular, fluent-bit-regular.yaml:

service:
  flush: 1
  log_level: debug
  parsers_file: /out/parsers.conf

pipeline:
  inputs:
    - name: tail
      path: /out/records.json
      read_from_head: true
      exit_on_eof: true
      parser: json

  outputs:
    - name: manticore
      match: "*"
      host: flb-pr-manticore
      port: 9308
      table: logs
      mode: Regular

Shared, fluent-bit-shared.yaml, differs in two lines:

      table: shared_logs
      mode: Shared

Both runs exit 0 and leave 1000 rows in their table (SELECT COUNT(*)).

Debug log output

Regular
Fluent Bit v5.1.4

[2026/10/15 09:27:48.095] [ info] Configuration:
[2026/10/15 09:27:48.109] [ info]  flush time     | 1.000000 seconds
[2026/10/15 09:27:48.114] [ info]  grace          | 5 seconds
[2026/10/15 09:27:48.115] [ info]  daemon         | 0
[2026/10/15 09:27:48.115] [ info] ___________
[2026/10/15 09:27:48.116] [ info]  inputs:
[2026/10/15 09:27:48.116] [ info]      tail
[2026/10/15 09:27:48.117] [ info] ___________
[2026/10/15 09:27:48.117] [ info]  filters:
[2026/10/15 09:27:48.118] [ info] ___________
[2026/10/15 09:27:48.119] [ info]  outputs:
[2026/10/15 09:27:48.119] [ info]      manticore.0
[2026/10/15 09:27:48.120] [ info] ___________
[2026/10/15 09:27:48.121] [ info]  collectors:
[2026/10/15 09:27:48.176] [ info] [fluent bit] version=5.1.4, commit=a4efcec0f2, pid=17662
[2026/10/15 09:27:48.184] [debug] [engine] coroutine stack size: 65536 bytes (64.0K)
[2026/10/15 09:27:48.189] [ info] [storage] ver=1.5.4, type=memory, sync=normal, checksum=off, max_chunks_up=128
[2026/10/15 09:27:48.190] [ info] [simd    ] SSE2
[2026/10/15 09:27:48.190] [ info] [cmetrics] version=2.2.5
[2026/10/15 09:27:48.190] [ info] [ctraces ] version=0.7.3
[2026/10/15 09:27:48.209] [ info] [input:tail:tail.0] initializing
[2026/10/15 09:27:48.210] [ info] [input:tail:tail.0] storage_strategy='memory' (memory only)
[2026/10/15 09:27:48.211] [debug] [tail:tail.0] created event channels: read=22 write=23
[2026/10/15 09:27:48.221] [debug] [input:tail:tail.0] flb_tail_fs_inotify_init() initializing inotify tail input
[2026/10/15 09:27:48.221] [debug] [input:tail:tail.0] inotify watch fd=28
[2026/10/15 09:27:48.223] [debug] [input:tail:tail.0] scanning path /out/records.json
[2026/10/15 09:27:48.232] [debug] [input:tail:tail.0] file will be read in POSIX_FADV_DONTNEED mode /out/records.json
[2026/10/15 09:27:48.242] [debug] [input:tail:tail.0] inode=8162774326671879 with offset=0 appended as /out/records.json
[2026/10/15 09:27:48.243] [debug] [input:tail:tail.0] scan_glob add(): /out/records.json, inode 8162774326671879
[2026/10/15 09:27:48.244] [debug] [input:tail:tail.0] 1 new files found on path '/out/records.json'
[2026/10/15 09:27:48.249] [debug] [manticore:manticore.0] created event channels: read=30 write=31
[2026/10/15 09:27:48.296] [ info] [output:manticore:manticore.0] worker #1 started
[2026/10/15 09:27:48.297] [ info] [output:manticore:manticore.0] worker #0 started
[2026/10/15 09:27:48.304] [ info] [sp] stream processor started
[2026/10/15 09:27:48.308] [ info] [engine] Shutdown Grace Period=5, Shutdown Input Grace Period=2
[2026/10/15 09:27:48.403] [debug] [input:tail:tail.0] [static files] processed 32.0K
[2026/10/15 09:27:48.444] [debug] [input:tail:tail.0] [static files] processed 31.9K
[2026/10/15 09:27:48.448] [debug] [input:tail:tail.0] [static files] processed 2.3K
[2026/10/15 09:27:48.449] [ info] [input:tail:tail.0] inode=8162774326671879 file=/out/records.json ended, stop
[2026/10/15 09:27:48.450] [debug] [input:tail:tail.0] inode=8162774326671879 file=/out/records.json promote to TAIL_EVENT
[2026/10/15 09:27:48.453] [ info] [input:tail:tail.0] inotify_fs_add(): inode=8162774326671879 watch_fd=1 name=/out/records.json
[2026/10/15 09:27:48.456] [debug] [input:tail:tail.0] [static files] processed 0b, done
[2026/10/15 09:27:48.462] [debug] [task] created task=0x5f98940 id=0 OK
[2026/10/15 09:27:48.465] [debug] [output:manticore:manticore.0] task_id=0 assigned to thread #0
[2026/10/15 09:27:48.617] [ warn] [engine] service will shutdown in max 5 seconds
[2026/10/15 09:27:48.617] [debug] [engine] task 0 already scheduled to run, not re-scheduling it.
[2026/10/15 09:27:48.618] [ info] [engine] pausing all inputs..
[2026/10/15 09:27:48.619] [ info] [input] pausing tail.0
[2026/10/15 09:27:48.794] [debug] [upstream] KA connection #61 to flb-pr-manticore:9308 is connected
[2026/10/15 09:27:48.796] [debug] [http_client] not using http_proxy for header
[2026/10/15 09:27:48.826] [debug] [upstream] KA connection #61 to flb-pr-manticore:9308 is now available
[2026/10/15 09:27:48.830] [debug] [out flush] cb_destroy coro_id=0
[2026/10/15 09:27:48.838] [debug] [task] destroy task=0x5f98940 (task_id=0)
[2026/10/15 09:27:48.989] [ info] [engine] service has stopped (0 pending tasks)
[2026/10/15 09:27:48.989] [ info] [input] pausing tail.0
[2026/10/15 09:27:48.991] [ info] [output:manticore:manticore.0] thread worker #0 stopping...
[2026/10/15 09:27:48.998] [ info] [output:manticore:manticore.0] thread worker #0 stopped
[2026/10/15 09:27:49.015] [ info] [output:manticore:manticore.0] thread worker #1 stopping...
[2026/10/15 09:27:49.016] [ info] [output:manticore:manticore.0] thread worker #1 stopped
[2026/10/15 09:27:49.028] [debug] [input:tail:tail.0] inode=8162774326671879 removing file name /out/records.json
[2026/10/15 09:27:49.030] [ info] [input:tail:tail.0] inotify_fs_remove(): inode=8162774326671879 watch_fd=1
Shared: two data flushes, then the Finish after the workers stop
Fluent Bit v5.1.4

[2026/10/15 09:27:50.721] [ info] Configuration:
[2026/10/15 09:27:50.734] [ info]  flush time     | 1.000000 seconds
[2026/10/15 09:27:50.739] [ info]  grace          | 5 seconds
[2026/10/15 09:27:50.740] [ info]  daemon         | 0
[2026/10/15 09:27:50.741] [ info] ___________
[2026/10/15 09:27:50.741] [ info]  inputs:
[2026/10/15 09:27:50.742] [ info]      tail
[2026/10/15 09:27:50.742] [ info] ___________
[2026/10/15 09:27:50.743] [ info]  filters:
[2026/10/15 09:27:50.744] [ info] ___________
[2026/10/15 09:27:50.744] [ info]  outputs:
[2026/10/15 09:27:50.745] [ info]      manticore.0
[2026/10/15 09:27:50.745] [ info] ___________
[2026/10/15 09:27:50.746] [ info]  collectors:
[2026/10/15 09:27:50.801] [ info] [fluent bit] version=5.1.4, commit=a4efcec0f2, pid=17680
[2026/10/15 09:27:50.809] [debug] [engine] coroutine stack size: 65536 bytes (64.0K)
[2026/10/15 09:27:50.814] [ info] [storage] ver=1.5.4, type=memory, sync=normal, checksum=off, max_chunks_up=128
[2026/10/15 09:27:50.814] [ info] [simd    ] SSE2
[2026/10/15 09:27:50.815] [ info] [cmetrics] version=2.2.5
[2026/10/15 09:27:50.815] [ info] [ctraces ] version=0.7.3
[2026/10/15 09:27:50.833] [ info] [input:tail:tail.0] initializing
[2026/10/15 09:27:50.834] [ info] [input:tail:tail.0] storage_strategy='memory' (memory only)
[2026/10/15 09:27:50.835] [debug] [tail:tail.0] created event channels: read=22 write=23
[2026/10/15 09:27:50.845] [debug] [input:tail:tail.0] flb_tail_fs_inotify_init() initializing inotify tail input
[2026/10/15 09:27:50.845] [debug] [input:tail:tail.0] inotify watch fd=28
[2026/10/15 09:27:50.847] [debug] [input:tail:tail.0] scanning path /out/records.json
[2026/10/15 09:27:50.856] [debug] [input:tail:tail.0] file will be read in POSIX_FADV_DONTNEED mode /out/records.json
[2026/10/15 09:27:50.869] [debug] [input:tail:tail.0] inode=8162774326671879 with offset=0 appended as /out/records.json
[2026/10/15 09:27:50.870] [debug] [input:tail:tail.0] scan_glob add(): /out/records.json, inode 8162774326671879
[2026/10/15 09:27:50.871] [debug] [input:tail:tail.0] 1 new files found on path '/out/records.json'
[2026/10/15 09:27:50.878] [debug] [manticore:manticore.0] created event channels: read=30 write=31
[2026/10/15 09:27:50.914] [ info] [output:manticore:manticore.0] worker #0 started
[2026/10/15 09:27:50.929] [ info] [output:manticore:manticore.0] worker #1 started
[2026/10/15 09:27:50.934] [ info] [sp] stream processor started
[2026/10/15 09:27:50.939] [ info] [engine] Shutdown Grace Period=5, Shutdown Input Grace Period=2
[2026/10/15 09:27:51.051] [debug] [input:tail:tail.0] [static files] processed 32.0K
[2026/10/15 09:27:51.058] [debug] [task] created task=0x8d2e060 id=0 OK
[2026/10/15 09:27:51.061] [debug] [output:manticore:manticore.0] task_id=0 assigned to thread #0
[2026/10/15 09:27:51.190] [debug] [input:tail:tail.0] [static files] processed 31.9K
[2026/10/15 09:27:51.230] [debug] [input:tail:tail.0] [static files] processed 2.3K
[2026/10/15 09:27:51.236] [ info] [input:tail:tail.0] inode=8162774326671879 file=/out/records.json ended, stop
[2026/10/15 09:27:51.236] [debug] [input:tail:tail.0] inode=8162774326671879 file=/out/records.json promote to TAIL_EVENT
[2026/10/15 09:27:51.239] [ info] [input:tail:tail.0] inotify_fs_add(): inode=8162774326671879 watch_fd=1 name=/out/records.json
[2026/10/15 09:27:51.242] [debug] [input:tail:tail.0] [static files] processed 0b, done
[2026/10/15 09:27:51.244] [debug] [task] created task=0x8ec5160 id=1 OK
[2026/10/15 09:27:51.244] [debug] [output:manticore:manticore.0] task_id=1 assigned to thread #1
[2026/10/15 09:27:51.341] [ warn] [engine] service will shutdown in max 5 seconds
[2026/10/15 09:27:51.342] [debug] [engine] task 0 already scheduled to run, not re-scheduling it.
[2026/10/15 09:27:51.342] [debug] [engine] task 1 already scheduled to run, not re-scheduling it.
[2026/10/15 09:27:51.342] [ info] [engine] pausing all inputs..
[2026/10/15 09:27:51.351] [ info] [input] pausing tail.0
[2026/10/15 09:27:51.402] [debug] [upstream] KA connection #61 to flb-pr-manticore:9308 is connected
[2026/10/15 09:27:51.404] [debug] [http_client] not using http_proxy for header
[2026/10/15 09:27:51.414] [debug] [upstream] KA connection #62 to flb-pr-manticore:9308 is connected
[2026/10/15 09:27:51.414] [debug] [http_client] not using http_proxy for header
[2026/10/15 09:27:51.431] [debug] [upstream] KA connection #61 to flb-pr-manticore:9308 is now available
[2026/10/15 09:27:51.434] [debug] [out flush] cb_destroy coro_id=0
[2026/10/15 09:27:51.443] [debug] [task] destroy task=0x8d2e060 (task_id=0)
[2026/10/15 09:27:51.447] [debug] [upstream] KA connection #62 to flb-pr-manticore:9308 is now available
[2026/10/15 09:27:51.447] [debug] [out flush] cb_destroy coro_id=0
[2026/10/15 09:27:51.448] [debug] [task] destroy task=0x8ec5160 (task_id=1)
[2026/10/15 09:27:51.982] [ info] [engine] service has stopped (0 pending tasks)
[2026/10/15 09:27:51.982] [ info] [input] pausing tail.0
[2026/10/15 09:27:51.985] [ info] [output:manticore:manticore.0] thread worker #0 stopping...
[2026/10/15 09:27:51.992] [ info] [output:manticore:manticore.0] thread worker #0 stopped
[2026/10/15 09:27:52.009] [ info] [output:manticore:manticore.0] thread worker #1 stopping...
[2026/10/15 09:27:52.009] [ info] [output:manticore:manticore.0] thread worker #1 stopped
[2026/10/15 09:27:52.056] [debug] [http_client] not using http_proxy for header
[2026/10/15 09:27:52.258] [ info] [output:manticore:manticore.0] shared bulk_import complete: table=shared_logs
[2026/10/15 09:27:52.270] [debug] [input:tail:tail.0] inode=8162774326671879 removing file name /out/records.json
[2026/10/15 09:27:52.272] [ info] [input:tail:tail.0] inotify_fs_remove(): inode=8162774326671879 watch_fd=1

Valgrind output

Fluent Bit built with -DFLB_DEV=On -DFLB_VALGRIND=On -DFLB_JEMALLOC=Off, run as valgrind --leak-check=full --show-leak-kinds=definite,indirect,possible --error-exitcode=99 ./bin/fluent-bit -c <config>:

Regular
==N== Memcheck, a memory error detector
==N== Copyright (C) 2002-2022, and GNU GPL'd, by Julian Seward et al.
==N== Using Valgrind-3.22.0 and LibVEX; rerun with -h for copyright info
==N== Command: ./bin/fluent-bit -c /out/fluent-bit-regular.yaml
==N== Parent PID: 1
==N== 
==N== 
==N== HEAP SUMMARY:
==N==     in use at exit: 0 bytes in 0 blocks
==N==   total heap usage: 20,744 allocs, 20,744 frees, 89,699,560 bytes allocated
==N== 
==N== All heap blocks were freed -- no leaks are possible
==N== 
==N== For lists of detected and suppressed errors, rerun with: -s
==N== ERROR SUMMARY: 0 errors from 0 contexts (suppressed: 0 from 0)
Shared
==N== Memcheck, a memory error detector
==N== Copyright (C) 2002-2022, and GNU GPL'd, by Julian Seward et al.
==N== Using Valgrind-3.22.0 and LibVEX; rerun with -h for copyright info
==N== Command: ./bin/fluent-bit -c /out/fluent-bit-shared.yaml
==N== Parent PID: 1
==N== 
==N== 
==N== HEAP SUMMARY:
==N==     in use at exit: 0 bytes in 0 blocks
==N==   total heap usage: 20,907 allocs, 20,907 frees, 49,669,248 bytes allocated
==N== 
==N== All heap blocks were freed -- no leaks are possible
==N== 
==N== For lists of detected and suppressed errors, rerun with: -s
==N== ERROR SUMMARY: 0 errors from 0 contexts (suppressed: 0 from 0)
flb-rt-out_manticore
==N== Memcheck, a memory error detector
==N== Copyright (C) 2002-2022, and GNU GPL'd, by Julian Seward et al.
==N== Using Valgrind-3.22.0 and LibVEX; rerun with -h for copyright info
==N== Command: ./bin/flb-rt-out_manticore
==N== Parent PID: 1
==N== 
==N== 
==N== HEAP SUMMARY:
==N==     in use at exit: 0 bytes in 0 blocks
==N==   total heap usage: 18,723 allocs, 18,723 frees, 4,907,025 bytes allocated
==N== 
==N== All heap blocks were freed -- no leaks are possible
==N== 
==N== For lists of detected and suppressed errors, rerun with: -s
==N== ERROR SUMMARY: 0 errors from 0 contexts (suppressed: 0 from 0)

The output writes log records to a Manticore Search table through the
HTTP /bulk endpoint, as NDJSON lines. A record goes as it is: the daemon
takes its "id" field as the document id or generates one.

Regular mode sends every chunk as one /bulk request of insert or create
operations, which Manticore runs as one transaction. A retryable item
status, 408, 429 or a 5xx reply without item statuses retries the
chunk; any other failure drops it.

Shared mode loads a finite source into one table with Manticore's
bulk_import. Data requests stage the records, Shared_Indexers sets how
many indexers the daemon starts for the load, and one Finish request
sent from the exit callback publishes it. The load is published only
if every chunk was delivered; otherwise the log names the statement
that cancels the staged data.

Signed-off-by: Stas Klinov <stas@manticoresearch.com>
The formatter test checks the NDJSON lines of Regular insert and create
and of Shared, for records with and without their own id. A bad mode,
action, shared_indexers or an empty table fails the start.

Signed-off-by: Stas Klinov <stas@manticoresearch.com>
A local receiver answers as the Manticore /bulk endpoint. The Regular
case checks the request lines, a retried item and a dropped one. The
Shared cases check the requests of one load and the Finish sent at
exit, the Finish over TLS with workers 0, an empty source, retried and
malformed replies, a failed Finish, and loads that end with pending,
exhausted or failed data and so are not published.

Signed-off-by: Stas Klinov <stas@manticoresearch.com>

This branch was successfully deployed

1 active deployment
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant