in_amqp: Initial implementation for in_amqp plugin#11427
Open
matwey wants to merge 4 commits intofluent:masterfrom
Open
in_amqp: Initial implementation for in_amqp plugin#11427matwey wants to merge 4 commits intofluent:masterfrom
matwey wants to merge 4 commits intofluent:masterfrom
Conversation
matwey
requested review from
celalettin1286,
cosmo0920,
edsiper,
niedbalski and
patrick-stephens
as code owners
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughAdds a new AMQP input plugin: build option and package gating, plugin registration and CMake files, header and implementation for AMQP connection/consumption, parser/encoder integration, reconnect/retry collectors, packaging updates, and CI/build defaults adjustments. Changes
Sequence DiagramssequenceDiagram
participant FLB as Fluent Bit
participant Plugin as in_amqp Plugin
participant Broker as AMQP Broker
participant Parser as Parser
participant Encoder as Encoder
FLB->>Plugin: in_amqp_init()
Plugin->>Plugin: load config (uri, queue, parser, reconnect)
Plugin->>Broker: open socket / login / open channel / basic_consume
Plugin->>Parser: init parser (if configured)
Plugin->>Encoder: init encoder
Plugin->>FLB: register collect callback
sequenceDiagram
participant Collector as in_amqp collect loop
participant Broker as AMQP Broker
participant Parser as Parser
participant Encoder as Encoder
participant FLB as Fluent Bit
Collector->>Broker: amqp_consume_message()/poll
alt delivery received
Broker->>Collector: envelope (body, headers, props, routing_key)
Collector->>Parser: parse body (optional)
Parser-->>Collector: parsed record / error
Collector->>Collector: attach metadata (routing_key, headers, timestamp)
Collector->>Encoder: encode event
Encoder-->>Collector: msgpack buffer
Collector->>FLB: emit record
else error / connection loss
Broker-->>Collector: error / timeout
Collector->>Collector: schedule reconnect via retry collector
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: ead056ea6c
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 6
🤖 Fix all issues with AI agents
In `@plugins/in_amqp/in_amqp.c`:
- Around line 579-583: The cast in in_amqp_exit improperly dereferences the
unused parameter using "(void) *config;", which can cause undefined behavior if
config is NULL; change that to the standard unused-parameter idiom "(void)
config;" inside the in_amqp_exit function so you do not dereference config while
keeping the variable marked as intentionally unused.
- Around line 446-516: The ctx struct allocated in in_amqp_init is created with
flb_malloc so its members (notably ctx->encoder) are uninitialized and later
in_amqp_config_destroy calls flb_log_event_encoder_destroy which can free
invalid pointers; change the allocation in in_amqp_init to use flb_calloc (or
allocate with flb_malloc followed by memset to zero) so struct flb_amqp fields
(including ctx->encoder) are zero-initialized before any early teardown; keep
all other initialization logic the same.
- Around line 170-176: The flb_plg_trace call logs AMQP body using "%s" which
assumes NUL-termination and can read past the buffer; change the trace to use a
length-limited format (use "%.*s" with body->len) so it prints at most body->len
bytes from body->bytes. Update the flb_plg_trace invocation(s) in the parsing
block around flb_parser_do (referencing ctx->parser, flb_parser_do, body->bytes,
body->len, and ctx->ins) to pass (int)body->len and body->bytes accordingly and
keep the flb_plg_error call as-is.
- Around line 182-218: The code reads optional fields from
amqp_basic_properties_t (properties->timestamp, content_type, content_encoding,
correlation_id, reply_to, headers) without checking the properties->_flags
bitmask; update the logic around flb_time_set(&out_time, properties->timestamp,
0) and each call to in_amqp_append_metadata_bytes/in_amqp_append_metadata_entry
so you first test the appropriate flag (AMQP_BASIC_TIMESTAMP_FLAG,
AMQP_BASIC_CONTENT_TYPE_FLAG, AMQP_BASIC_CONTENT_ENCODING_FLAG,
AMQP_BASIC_CORRELATION_ID_FLAG, AMQP_BASIC_REPLY_TO_FLAG,
AMQP_BASIC_HEADERS_FLAG) before accessing the field; only call flb_time_set or
in_amqp_append_metadata_* when the corresponding flag is set, otherwise skip to
the fallback (flb_time_get) or omit that metadata entry to avoid reading
uninitialized memory (refer to symbols: properties, properties->_flags,
AMQP_BASIC_*_FLAG, out_time, flb_time_set, flb_time_get,
in_amqp_append_metadata_bytes, in_amqp_append_metadata_entry, headers,
ctx->encoder).
- Around line 350-386: The AMQP_RESPONSE_LIBRARY_EXCEPTION branch currently
handles only UNEXPECTED_STATE, CONNECTION_CLOSED and TIMEOUT; other
ret.library_error values from amqp_consume_message() should be handled to avoid
tight CPU loops—add an else clause after the existing checks that logs the
unexpected ret.library_error via flb_plg_warn (including the numeric error),
tears down the connection with in_amqp_connection_destroy(ctx) and either return
a non-zero error or call in_amqp_consumer_start(in, config, in_context) to
trigger a reconnect/backoff; reference the amqp_consume_message() return
handling, ret.library_error, AMQP_RESPONSE_LIBRARY_EXCEPTION, flb_plg_warn,
in_amqp_connection_destroy and in_amqp_consumer_start when applying the change.
- Around line 398-423: When reconnecting, the old socket collector
(ctx->coll_id) is left registered and must be removed before creating a new one;
modify the reconnect path (in the code that calls in_amqp_connection_init and
then sets ctx->coll_id) to check if ctx->coll_id >= 0 and call
flb_input_collector_destroy(ctx->coll_id, in) (or the appropriate
collector-remove API used elsewhere) to delete the stale collector, then clear
ctx->coll_id and proceed to create the new collector with
flb_input_set_collector_socket (refer to ctx->coll_id, in_amqp_collect,
amqp_socket_get_sockfd and in_amqp_connection_init).
3a5b411 to
d3d939e
Compare
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Fix all issues with AI agents
In `@plugins/in_amqp/in_amqp.c`:
- Around line 421-433: When collector creation or start fails, the code must
clean up the partially-created collector and the open AMQP connection to avoid
leaving the plugin stuck: if flb_input_set_collector_socket succeeded but
flb_input_collector_start failed, call flb_input_collector_remove(ctx->coll_id)
(or the equivalent collector-remove API) and set ctx->coll_id = -1; in both
failure branches close the underlying socket (c->sock) and tear down the AMQP
connection using the library cleanup functions (e.g., amqp_socket_close(c->sock)
and the appropriate amqp connection destroy/close function), clear any
connection pointers in ctx, and then return the error.
- Around line 346-386: in in_amqp_collect(), when amqp_consume_message() returns
AMQP_RESPONSE_SERVER_EXCEPTION or AMQP_RESPONSE_NONE, call
in_amqp_log_reply_error(&ret, in) to record the error, then call
in_amqp_connection_destroy(ctx) and restart the consumer by returning
in_amqp_consumer_start(in, config, in_context); this mirrors the existing
library-exception handling flow and prevents a busy spin; reference the
amqp_consume_message return handling, in_amqp_log_reply_error(),
in_amqp_connection_destroy(), and in_amqp_consumer_start().
- Around line 592-595: The default AMQP URI "amqp://" is missing a HOST and will
cause amqp_parse_url to fail; update the default for the config map entry that
sets struct flb_amqp->uri (the FLB_CONFIG_MAP_STR for "uri") to a parseable
value (either NULL to allow amqp_default_connection_info or a concrete
"amqp://localhost") and ensure any code that reads flb_amqp->uri and calls
amqp_parse_url handles NULL appropriately by falling back to
amqp_default_connection_info; locate the config map entry (FLB_CONFIG_MAP_STR,
"uri", "amqp://") and change the default to NULL or "amqp://localhost" and
adjust initialization paths that call amqp_parse_url to accommodate the new
default.
cosmo0920
requested changes
Feb 9, 2026
There was a problem hiding this comment.
Actionable comments posted: 5
🤖 Fix all issues with AI agents
In `@plugins/in_amqp/in_amqp.c`:
- Around line 139-141: Fix the typo in the comment inside the switch handling
AMQP field kinds: change the comment "fallthrought" to the correct "fallthrough"
where the case AMQP_FIELD_KIND_UTF8 falls into case AMQP_FIELD_KIND_BYTES so the
intent is clear; update the comment text near the AMQP_FIELD_KIND_UTF8 /
AMQP_FIELD_KIND_BYTES case labels to the correct spelling.
- Around line 413-451: in in_amqp_consumer_start: remove the direct call to
in_amqp_collect (the mutual recursion risk) and instead finish setup after
resetting ctx->retry and pausing the retry collector; rely on the socket/IO
collector you registered earlier to invoke in_amqp_collect naturally.
Specifically, stop calling in_amqp_collect(in, config, ctx) here, keep the
flb_input_collector_pause(ctx->retry_coll_id, in) and return 0 so the existing
socket collector will handle pending messages; ensure no other code path assumes
in_amqp_consumer_start will synchronously process the queue.
- Around line 276-327: The loop in in_amqp_collect can recurse unbounded by
calling in_amqp_consumer_start on AMQP_STATUS_CONNECTION_CLOSED; change the
behavior so that when ret.library_error == AMQP_STATUS_CONNECTION_CLOSED you
call in_amqp_connection_destroy(ctx) and break out of the for(;;) loop (or
return an error code) instead of calling in_amqp_consumer_start to avoid
recursion, letting the outer retry/timer logic handle reconnects; also add
catch-all branches: handle other ret.reply_type values
(AMQP_RESPONSE_SERVER_EXCEPTION, AMQP_RESPONSE_NONE) by logging and
breaking/returning appropriately, and add an else for ret.library_error to log
the unexpected error and break/return so the loop does not spin indefinitely.
- Around line 566-572: in in_amqp_pause ensure you only call
flb_input_collector_pause for the connection collector when a connection exists:
check ctx->conn.coll_id != -1 before invoking
flb_input_collector_pause(ctx->conn.coll_id, ctx->ins); still pause
ctx->retry_coll_id unconditionally as before. This mirrors the guard used in
in_amqp_resume and prevents attempting to pause a non-existent collector.
- Around line 585-593: Add a NULL guard in in_amqp_exit: check whether the
incoming data pointer is NULL before casting to struct flb_amqp *ctx and calling
in_amqp_config_destroy; if data is NULL simply return 0. This ensures
in_amqp_config_destroy (which dereferences ctx->encoder, ctx->retry_coll_id,
ctx->ins) is only called when ctx is non-NULL and avoids null pointer
dereference when the framework invokes cb_exit with a NULL context.
|
Caution Failed to replace (edit) comment. This is likely due to insufficient permissions or the comment being deleted. Error details |
1 similar comment
|
Caution Failed to replace (edit) comment. This is likely due to insufficient permissions or the comment being deleted. Error details |
✅ Actions performedReviews resumed. |
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Fix all issues with AI agents
In `@plugins/in_amqp/in_amqp.c`:
- Around line 451-459: in in_amqp_config_destroy, guard the call to
flb_input_collector_delete by checking ctx->retry_coll_id is valid (not -1)
before calling it; update the function handling for struct flb_amqp (used by
in_amqp_init and cleanup paths) to only call
flb_input_collector_delete(ctx->retry_coll_id, ctx->ins) when ctx->retry_coll_id
has been initialized to a non -1 value, leaving other cleanup
(flb_log_event_encoder_destroy, in_amqp_connection_destroy, flb_free) unchanged.
- Around line 327-339: in_amqp_connection_destroy currently calls
flb_input_collector_delete unconditionally which can be invoked with c->coll_id
== -1; update the function to check c->coll_id != -1 before calling
flb_input_collector_delete and only reset c->coll_id to -1 after successful
deletion; locate the logic in in_amqp_connection_destroy that references struct
flb_amqp_connection *c = &ctx->conn, the c->conn check, and the
flb_input_collector_delete(c->coll_id, ctx->ins) call to add the guard.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 83ae35cb81
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
Comment on lines
+333
to
+335
| in_amqp_log_reply_error(in, reply, "An error occurred during consuming message"); | ||
|
|
||
| in_amqp_connection_destroy(ctx); |
There was a problem hiding this comment.
After handling AMQP_STATUS_UNEXPECTED_STATE, this path always logs an error and destroys/recreates the connection. That means non-delivery frames that are not connection/channel-close (e.g., normal control/heartbeat traffic) still trigger a full reconnect cycle, which can cause repeated disconnect/reconnect churn and dropped throughput on otherwise healthy brokers. The branch should continue consuming for benign frames and only reconnect on definitive close/error conditions.
Useful? React with 👍 / 👎.
|
|
||
| queue_bytes.len = flb_sds_len(ctx->queue_name); | ||
| queue_bytes.bytes = ctx->queue_name; | ||
| if (amqp_basic_consume(c->conn, c->chan, queue_bytes, amqp_empty_bytes, 0, 1, 1, amqp_empty_table) == NULL) { |
There was a problem hiding this comment.
The consume call hardcodes exclusive=1, which asks RabbitMQ for an exclusive consumer every time. In queues that already have another consumer (including another Fluent Bit instance), the broker will reject the consume request and this plugin will loop through reconnect attempts instead of consuming. This should be non-exclusive by default (or configurable) to avoid breaking common shared-queue deployments.
Useful? React with 👍 / 👎.
Author
There was a problem hiding this comment.
This makes no sense. There can not be two consumers for this queue, because it will lead to data loss.
There was a problem hiding this comment.
🧹 Nitpick comments (3)
plugins/in_amqp/in_amqp.c (3)
538-543: Consider usingflb_callocfor consistency with other input plugins.The
flb_malloc+memsetapproach is functionally correct, butflb_calloc(1, sizeof(struct flb_amqp))is the prevailing idiom in Fluent Bit input plugins and achieves the same zero-initialization in one call. The explicit-1assignments forretry_coll_idandconn.coll_idon lines 547 and 552 are still needed regardless.
608-628: Pause/resume should handle both collectors to avoid missed state transitions.Currently,
in_amqp_pauseonly pauses one collector based on whether a connection exists. If the plugin is paused while connected, only the socket collector is paused. If the connection drops while paused (e.g., TCP keepalive detects a dead connection — the socket collector is paused soin_amqp_collectwon't run), and thenin_amqp_resumeis called, it would checkctx->conn.conn(still non-NULL since nothing cleaned it up), resume only the socket collector (now pointing at a dead FD), and the retry collector would never be resumed.This is an edge case, but unconditionally pausing/resuming both collectors (guarding against invalid
coll_id) would be more robust:Proposed fix
static void in_amqp_pause(void *data, struct flb_config *config) { struct flb_amqp *ctx = data; - if (ctx->conn.conn) { + if (ctx->conn.coll_id >= 0) { flb_input_collector_pause(ctx->conn.coll_id, ctx->ins); - } else { - flb_input_collector_pause(ctx->retry_coll_id, ctx->ins); } + flb_input_collector_pause(ctx->retry_coll_id, ctx->ins); } static void in_amqp_resume(void *data, struct flb_config *config) { struct flb_amqp *ctx = data; - if (ctx->conn.conn) { + if (ctx->conn.coll_id >= 0) { flb_input_collector_resume(ctx->conn.coll_id, ctx->ins); - } else { + } + if (!ctx->conn.conn) { flb_input_collector_resume(ctx->retry_coll_id, ctx->ins); } }
509-511:amqp_parse_urlmutates the URI string in-place, creating a fragile pattern.
amqp_parse_urlinserts NUL bytes into its input buffer and stores pointers into it withinconn_info. Sincectx->uriis a framework-managedflb_sds_tfrom the config map, this mutates the config string in-place. While this works correctly at runtime (the SDS length header preserves proper deallocation), it meansctx->uriis no longer a valid C-string after parsing. Although no logging ofctx->uricurrently occurs after the parse, this creates a fragile pattern where future debug/config logging could show truncated values. Consider making a mutable copy to avoid this brittleness:Proposed refactor
if (ctx->uri) { - ret = amqp_parse_url(ctx->uri, &ctx->conn_info); + ctx->uri_buf = flb_sds_create(ctx->uri); + if (!ctx->uri_buf) { + return -1; + } + ret = amqp_parse_url(ctx->uri_buf, &ctx->conn_info); if (ret != AMQP_STATUS_OK) {(with corresponding
flb_sds_destroy(ctx->uri_buf)inin_amqp_config_destroy)
There was a problem hiding this comment.
🧹 Nitpick comments (2)
plugins/in_amqp/in_amqp.c (2)
154-268: Solid envelope handling with proper_flagsguards and encoder error chaining.A few minor observations:
- Line 213: extra whitespace in
ret == FLB_EVENT_ENCODER_SUCCESS(double space beforeFLB_EVENT_ENCODER_SUCCESS).- If a header entry encoding fails mid-iteration (line 220-222), the map opened at line 217 is never explicitly rolled back via
flb_log_event_encoder_metadata_rollback_map. The overall record won't be committed (guarded at line 250), andencoder_resetat line 263 will clean up, so this doesn't cause a data corruption bug. However, an explicit rollback on error would be more robust.Neither is blocking — the encoder reset path handles cleanup correctly.
Minor: fix double space and add explicit map rollback
- if (ret == FLB_EVENT_ENCODER_SUCCESS && (properties->_flags & AMQP_BASIC_HEADERS_FLAG)) { + if (ret == FLB_EVENT_ENCODER_SUCCESS && (properties->_flags & AMQP_BASIC_HEADERS_FLAG)) {For robustness, consider adding an
elsebranch after the commit_map:if (ret == FLB_EVENT_ENCODER_SUCCESS) { ret = flb_log_event_encoder_metadata_commit_map((&ctx->encoder)); + } else { + flb_log_event_encoder_metadata_rollback_map((&ctx->encoder)); }
657-665: Consider validatingreconnect.retry_intervalto prevent a rapid-fire timer.A value of
0forreconnect.retry_intervalwould create a timer with zero-second interval, potentially causing high CPU usage during reconnect cycles. A minimum (e.g., 1 second) would be a reasonable safeguard.Suggested validation in in_amqp_configure
+ if (ctx->reconnect_retry_interval < 1) { + flb_plg_warn(in, "reconnect.retry_interval must be >= 1, defaulting to 1"); + ctx->reconnect_retry_interval = 1; + } + if (!ctx->queue_name) {
Author
|
@coderabbitai resume |
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@plugins/in_amqp/in_amqp.c`:
- Around line 586-603: The code sets the input context with
flb_input_set_context(in, ctx) but on the two late-init failure paths (after
flb_input_set_collector_time returning <0 and after in_amqp_consumer_start
returning <0) it calls in_amqp_config_destroy(ctx) without clearing in->context,
leaving a dangling pointer; update both failure branches to call
flb_input_set_context(in, NULL) (or otherwise clear the input context) before
calling in_amqp_config_destroy(ctx) and returning -1 so the freed ctx is not
referenced by the input instance.
ℹ️ Review info
Configuration used: defaults
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (6)
CMakeLists.txtcmake/plugins_options.cmakeplugins/CMakeLists.txtplugins/in_amqp/CMakeLists.txtplugins/in_amqp/in_amqp.cplugins/in_amqp/in_amqp.h
🚧 Files skipped from review as they are similar to previous changes (4)
- plugins/CMakeLists.txt
- plugins/in_amqp/in_amqp.h
- CMakeLists.txt
- plugins/in_amqp/CMakeLists.txt
Comment on lines
+586
to
+603
| flb_input_set_context(in, ctx); | ||
|
|
||
| ctx->retry_coll_id = flb_input_set_collector_time(in, &in_amqp_reconnect, ctx->reconnect_retry_interval, 0, config); | ||
| if (ctx->retry_coll_id < 0) { | ||
| flb_plg_error(in, "Cannot create reconnection collector"); | ||
| in_amqp_config_destroy(ctx); | ||
|
|
||
| return -1; | ||
| } | ||
|
|
||
| flb_input_collector_pause(ctx->retry_coll_id, in); | ||
|
|
||
| ret = in_amqp_consumer_start(ctx, config); | ||
| if (ret < 0) { | ||
| flb_plg_error(in, "Cannot start AMQP consumer"); | ||
| in_amqp_config_destroy(ctx); | ||
|
|
||
| return -1; |
There was a problem hiding this comment.
Clear the input context before freeing ctx on late init failures.
After flb_input_set_context(in, ctx) at Line 586, failure paths at Lines 589-593 and 599-603 free ctx but leave in->context pointing to freed memory. Clear the context before destroy in those branches.
🔧 Suggested fix
flb_input_set_context(in, ctx);
ctx->retry_coll_id = flb_input_set_collector_time(in, &in_amqp_reconnect, ctx->reconnect_retry_interval, 0, config);
if (ctx->retry_coll_id < 0) {
flb_plg_error(in, "Cannot create reconnection collector");
+ flb_input_set_context(in, NULL);
in_amqp_config_destroy(ctx);
return -1;
}
@@
ret = in_amqp_consumer_start(ctx, config);
if (ret < 0) {
flb_plg_error(in, "Cannot start AMQP consumer");
+ flb_input_set_context(in, NULL);
in_amqp_config_destroy(ctx);
return -1;
}🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@plugins/in_amqp/in_amqp.c` around lines 586 - 603, The code sets the input
context with flb_input_set_context(in, ctx) but on the two late-init failure
paths (after flb_input_set_collector_time returning <0 and after
in_amqp_consumer_start returning <0) it calls in_amqp_config_destroy(ctx)
without clearing in->context, leaving a dangling pointer; update both failure
branches to call flb_input_set_context(in, NULL) (or otherwise clear the input
context) before calling in_amqp_config_destroy(ctx) and returning -1 so the
freed ctx is not referenced by the input instance.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 8d485c33e5
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
Comment on lines
+315
to
+317
| } else if (frame.frame_type == AMQP_FRAME_METHOD) { | ||
| switch (frame.payload.method.id) { | ||
| case AMQP_CHANNEL_CLOSE_METHOD: { |
There was a problem hiding this comment.
When amqp_consume_message() returns AMQP_STATUS_UNEXPECTED_STATE, this code only treats method frames as recoverable and falls through to connection teardown for any non-method frame. In RabbitMQ deployments with negotiated heartbeats (commonly non-zero), heartbeat frames can hit this path and trigger needless disconnect/reconnect cycles, causing intermittent ingestion gaps and churn instead of continuing consumption.
Useful? React with 👍 / 👎.
|
|
||
| queue_bytes.len = flb_sds_len(ctx->queue_name); | ||
| queue_bytes.bytes = ctx->queue_name; | ||
| if (amqp_basic_consume(c->conn, c->chan, queue_bytes, amqp_empty_bytes, 0, 1, 1, amqp_empty_table) == NULL) { |
There was a problem hiding this comment.
The amqp_basic_consume call hard-codes exclusive=1, which makes the broker reject this consumer whenever the queue already has another active consumer. That turns normal shared-queue or failover topologies into repeated startup/reconnect failures; this should default to non-exclusive (or be configurable) so the plugin can attach to queues used by multiple consumers.
Useful? React with 👍 / 👎.
|
✅ Actions performedReviews resumed. |
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (1)
plugins/in_amqp/in_amqp.c (1)
471-479:⚠️ Potential issue | 🟠 MajorClear input context during teardown to avoid dangling pointer after late init failures.
After
flb_input_set_context(in, ctx)on Line 586, failures on Line 589 and Line 599 callin_amqp_config_destroy(ctx).ctxis freed, butin->contextis not cleared, leaving stale state and possible double cleanup paths.🔧 Proposed fix
static int in_amqp_config_destroy(struct flb_amqp *ctx) { + if (ctx->ins) { + flb_input_set_context(ctx->ins, NULL); + } + flb_log_event_encoder_destroy(&ctx->encoder); if (ctx->retry_coll_id >= 0) { flb_input_collector_delete(ctx->retry_coll_id, ctx->ins); }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@plugins/in_amqp/in_amqp.c` around lines 471 - 479, in_amqp_config_destroy currently frees ctx but does not clear the input instance's context pointer, leaving in->context dangling after failures that call this teardown; update the teardown so that after flb_input_set_context(in, ctx) failures and inside in_amqp_config_destroy you explicitly clear the input context (set in->context = NULL) before or immediately after freeing ctx, or change in_amqp_config_destroy signature to accept struct flb_input_instance *in and null out in->context inside the function (affecting functions: in_amqp_config_destroy, the code paths that call flb_input_set_context and call in_amqp_config_destroy).
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@plugins/in_amqp/in_amqp.c`:
- Around line 650-675: The new AMQP input options declared in config_map
(fields: uri, queue, parser, reconnect.retry_limits, reconnect.retry_interval
tied to struct flb_amqp) lack user-facing documentation; add a concise docs
section describing each option's purpose and type, state the defaults ("amqp://"
for uri, no default for queue and parser, reconnect.retry_limits=5,
reconnect.retry_interval=60), show a minimal working example config snippet
using these keys, and update the plugin's README/docs index and changelog so the
PR no longer requires a docs follow-up.
---
Duplicate comments:
In `@plugins/in_amqp/in_amqp.c`:
- Around line 471-479: in_amqp_config_destroy currently frees ctx but does not
clear the input instance's context pointer, leaving in->context dangling after
failures that call this teardown; update the teardown so that after
flb_input_set_context(in, ctx) failures and inside in_amqp_config_destroy you
explicitly clear the input context (set in->context = NULL) before or
immediately after freeing ctx, or change in_amqp_config_destroy signature to
accept struct flb_input_instance *in and null out in->context inside the
function (affecting functions: in_amqp_config_destroy, the code paths that call
flb_input_set_context and call in_amqp_config_destroy).
ℹ️ Review info
Configuration used: defaults
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (6)
CMakeLists.txtcmake/plugins_options.cmakeplugins/CMakeLists.txtplugins/in_amqp/CMakeLists.txtplugins/in_amqp/in_amqp.cplugins/in_amqp/in_amqp.h
🚧 Files skipped from review as they are similar to previous changes (2)
- CMakeLists.txt
- plugins/in_amqp/CMakeLists.txt
Comment on lines
+650
to
+675
| static struct flb_config_map config_map[] = { | ||
| { | ||
| FLB_CONFIG_MAP_STR, "uri", "amqp://", | ||
| 0, FLB_TRUE, offsetof(struct flb_amqp, uri), | ||
| "Specify an AMQP URI to connect the broker" | ||
| }, | ||
| { | ||
| FLB_CONFIG_MAP_STR, "queue", NULL, | ||
| 0, FLB_TRUE, offsetof(struct flb_amqp, queue_name), | ||
| "Specify an AMQP queue name to consume from" | ||
| }, | ||
| { | ||
| FLB_CONFIG_MAP_STR, "parser", NULL, | ||
| 0, FLB_TRUE, offsetof(struct flb_amqp, parser_name), | ||
| "Set a parser" | ||
| }, | ||
| { | ||
| FLB_CONFIG_MAP_INT, "reconnect.retry_limits", "5", | ||
| 0, FLB_TRUE, offsetof(struct flb_amqp, reconnect_retry_limits), | ||
| "Maximum number to retry to connect the broker" | ||
| }, | ||
| { | ||
| FLB_CONFIG_MAP_INT, "reconnect.retry_interval", "60", | ||
| 0, FLB_TRUE, offsetof(struct flb_amqp, reconnect_retry_interval), | ||
| "Retry interval to connect the broker" | ||
| }, |
There was a problem hiding this comment.
Add user-facing documentation for the new AMQP input options before merge.
Lines 652-675 introduce externally configurable AMQP settings (uri, queue, parser, reconnect.*). The PR is marked docs-required and this surface should be documented with option semantics/defaults and a minimal example.
I can draft a concise docs section for these options if you want me to open a follow-up issue template.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@plugins/in_amqp/in_amqp.c` around lines 650 - 675, The new AMQP input options
declared in config_map (fields: uri, queue, parser, reconnect.retry_limits,
reconnect.retry_interval tied to struct flb_amqp) lack user-facing
documentation; add a concise docs section describing each option's purpose and
type, state the defaults ("amqp://" for uri, no default for queue and parser,
reconnect.retry_limits=5, reconnect.retry_interval=60), show a minimal working
example config snippet using these keys, and update the plugin's README/docs
index and changelog so the PR no longer requires a docs follow-up.
|
|
||
| # Inputs (sources, data collectors) | ||
| # ================================= | ||
| DEFINE_OPTION(FLB_IN_AMQP "Enable AMQP input plugin" ON) |
Contributor
There was a problem hiding this comment.
Does this compile for all supported targets, including Windows and macOS? If not you need to provide the cmake overrides for those targets appropriately - there are macOS and Windows ones plus you can do the usual CMake config approach to enable/disable as required.
We provide a linked build script in the PR template to help test Linux targets.
There are no changes to any of the builds so no new dependencies are required?
Author
There was a problem hiding this comment.
I've tested this only on Linux since I don't have macOS and Windows. From what I see, this PR should work on all platforms.
There is single new dependency for this plugin:
https://github.com/matwey/fluent-bit/blob/8d485c33e5eeab4b42489f56caeafb58bce3b166/CMakeLists.txt#L1337
rabbitmq-c is cross-platform library. Existing vcpkg port proves that it can be built for all platforms.
Contributor
There was a problem hiding this comment.
We will need to include the dependency for all builds as part of this change then , it might be easier to disable as well for macos/windows for now - there are config files in cmake/ for this.
Linux builds are done using the containers under the packaging/ directory so we can add the dependencies there if required, or include a vendored version of it if we are building from source under lib/.
There was a problem hiding this comment.
🧹 Nitpick comments (1)
CMakeLists.txt (1)
1443-1447: Gate AMQP dependency probing to AMQP-enabled builds (and make it quiet).This works, but
find_package(rabbitmq-c)currently runs even whenFLB_IN_AMQP=OFF. Wrapping it and usingQUIETavoids unnecessary lookup noise while preserving your explicit status message when AMQP is requested but unavailable.Proposed patch
-find_package(rabbitmq-c) -if(FLB_IN_AMQP AND (NOT rabbitmq-c_FOUND)) - message(STATUS "rabbitmq-c is not found. Disabling AMQP support.") - FLB_OPTION(FLB_IN_AMQP OFF) +if(FLB_IN_AMQP) + find_package(rabbitmq-c QUIET) + if(NOT rabbitmq-c_FOUND) + message(STATUS "rabbitmq-c is not found. Disabling AMQP support.") + FLB_OPTION(FLB_IN_AMQP OFF) + endif() endif()You can verify this by configuring once with
-DFLB_IN_AMQP=OFFand confirming there is no rabbitmq-c lookup warning in configure output.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@CMakeLists.txt` around lines 1443 - 1447, The CMakeLists currently always calls find_package(rabbitmq-c) causing noisy lookups even when FLB_IN_AMQP is OFF; change the logic to only call find_package(rabbitmq-c QUIET) when FLB_IN_AMQP is ON and keep the existing fallback that checks rabbitmq-c_FOUND and calls message(STATUS "... Disabling AMQP support.") and FLB_OPTION(FLB_IN_AMQP OFF) when requested AMQP support cannot be satisfied; this gates the probe to AMQP-enabled builds and silences unnecessary warnings while preserving the explicit status message when AMQP is requested but unavailable.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Nitpick comments:
In `@CMakeLists.txt`:
- Around line 1443-1447: The CMakeLists currently always calls
find_package(rabbitmq-c) causing noisy lookups even when FLB_IN_AMQP is OFF;
change the logic to only call find_package(rabbitmq-c QUIET) when FLB_IN_AMQP is
ON and keep the existing fallback that checks rabbitmq-c_FOUND and calls
message(STATUS "... Disabling AMQP support.") and FLB_OPTION(FLB_IN_AMQP OFF)
when requested AMQP support cannot be satisfied; this gates the probe to
AMQP-enabled builds and silences unnecessary warnings while preserving the
explicit status message when AMQP is requested but unavailable.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: d7584256-a52b-45dd-9826-6da93b56e49e
📒 Files selected for processing (2)
CMakeLists.txtcmake/plugins_options.cmake
✅ Files skipped from review due to trivial changes (1)
- cmake/plugins_options.cmake
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (4)
packaging/distros/amazonlinux/Dockerfile (2)
146-161:⚠️ Potential issue | 🟠 MajorMissing
ARG FLB_IN_AMQPdeclaration in builder stage.The CMake command references
${FLB_IN_AMQP}but there's noARG FLB_IN_AMQPdeclared in the builder stage. While the value may be inherited from base images that define it viaENV, this pattern is inconsistent with how other flags likeFLB_KAFKAare handled (which have both ARG in base images AND in the builder stage).For
amazonlinux-2-base, which installslibrabbitmq-develbut has noFLB_IN_AMQPdeclaration, the CMake flag will receive an empty value.🐛 Proposed fix
ARG FLB_OUT_PGSQL=On ARG FLB_JEMALLOC=On ARG FLB_CHUNK_TRACE=On +ARG FLB_IN_AMQP=On RUN cmake -DCMAKE_INSTALL_PREFIX="$CMAKE_INSTALL_PREFIX" \🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packaging/distros/amazonlinux/Dockerfile` around lines 146 - 161, The builder stage is missing an ARG declaration for FLB_IN_AMQP while the RUN cmake invocation passes -DFLB_IN_AMQP="${FLB_IN_AMQP}", causing an empty/undefined value; add an ARG FLB_IN_AMQP line in the builder stage (matching the pattern used for ARG FLB_KAFKA, ARG FLB_JEMALLOC, etc.) so FLB_IN_AMQP is defined and available to the RUN cmake command that sets -DFLB_IN_AMQP.
103-120:⚠️ Potential issue | 🟡 MinorMissing
FLB_IN_AMQPdeclaration for amazonlinux-2023.arm64v8-base.The
amazonlinux-2023-base(x86_64) explicitly setsFLB_IN_AMQP=Offand omitslibrabbitmq-devel, butamazonlinux-2023.arm64v8-basehas neither the package nor an explicitFLB_IN_AMQPsetting. This creates an inconsistency between architectures.Since the builder stage references
FLB_IN_AMQPwithout declaring a default, the arm64 build will use an empty/unset value, which may cause unexpected behavior.For consistency with the x86_64 variant, either:
- Add
ARG FLB_IN_AMQP=OffandENV FLB_IN_AMQP=$FLB_IN_AMQPto explicitly disable AMQP, or- Install
librabbitmq-develand setFLB_IN_AMQP=Onto enable it🐛 Proposed fix (Option 1: Explicitly disable)
ENV PATH="${CMAKE_HOME}/bin:${PATH}" ARG FLB_KAFKA=On ENV FLB_KAFKA=$FLB_KAFKA +ARG FLB_IN_AMQP=Off +ENV FLB_IN_AMQP=$FLB_IN_AMQP🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packaging/distros/amazonlinux/Dockerfile` around lines 103 - 120, Add an explicit default for FLB_IN_AMQP to match the x86_64 variant: declare ARG FLB_IN_AMQP=Off and then ENV FLB_IN_AMQP=$FLB_IN_AMQP in the amazonlinux-2023.arm64v8-base Dockerfile (or alternatively install librabbitmq-devel and set FLB_IN_AMQP=On if you intend to enable AMQP); update the Dockerfile near the other ARG/ENV declarations so the builder stage that references FLB_IN_AMQP always sees a defined value.packaging/distros/ubuntu/Dockerfile (2)
173-186:⚠️ Potential issue | 🟠 MajorMissing
librabbitmq-devin ubuntu-22.04-base (x86_64).The
librabbitmq-devpackage is not installed in theubuntu-22.04-baseimage, but it's installed in all other Ubuntu variants (includingubuntu-22.04.arm64v8-baseat line 206). This will cause the AMQP plugin to be silently disabled for Ubuntu 22.04 x86_64 builds.🐛 Proposed fix
# hadolint ignore=DL3008,DL3015 RUN apt-get update && \ apt-get install -y curl ca-certificates build-essential libsystemd-dev \ make bash wget unzip nano vim valgrind dh-make flex bison \ libpq-dev postgresql-server-dev-all libpq5 \ libsasl2-2 libsasl2-dev openssl libssl-dev libssl3 libcurl4-openssl-dev \ libyaml-dev pkg-config zlib1g-dev \ + librabbitmq-dev \ tar gzip && \🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packaging/distros/ubuntu/Dockerfile` around lines 173 - 186, The ubuntu-22.04-base Dockerfile RUN apt-get install command is missing librabbitmq-dev causing AMQP plugin to be disabled; update the package list in the RUN block (the long apt-get install -y ... line that precedes the cmake download using environment vars CMAKE_HOME and CMAKE_VERSION) to include librabbitmq-dev alongside the other -dev packages so the AMQP development headers are installed for x86_64.
251-265:⚠️ Potential issue | 🟠 MajorMissing
librabbitmq-devin ubuntu-24.04.arm64v8-base.The
librabbitmq-devpackage is not installed in theubuntu-24.04.arm64v8-baseimage, but it's installed inubuntu-24.04-base(line 231). This inconsistency will cause the AMQP plugin to be silently disabled for Ubuntu 24.04 arm64 builds.🐛 Proposed fix
# hadolint ignore=DL3008,DL3015 RUN apt-get update && \ apt-get install -y curl ca-certificates build-essential libsystemd-dev \ make bash wget unzip nano vim valgrind dh-make flex bison \ libpq-dev postgresql-server-dev-all libpq5 \ libsasl2-2 libsasl2-dev openssl libssl-dev libssl3 libcurl4-openssl-dev \ libyaml-dev pkg-config zlib1g-dev \ + librabbitmq-dev \ tar gzip && \🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packaging/distros/ubuntu/Dockerfile` around lines 251 - 265, The ubuntu-24.04.arm64v8-base Dockerfile RUN layer that installs packages (the long RUN apt-get update && apt-get install -y ... && cmake download block which references CMAKE_HOME and CMAKE_VERSION and sets ENV PATH) is missing librabbitmq-dev; add librabbitmq-dev to the apt-get install -y package list alongside libpq-dev, libsasl2-dev, libssl-dev, etc., so the AMQP plugin is available on arm64 as it is in the other ubuntu-24.04-base image.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Outside diff comments:
In `@packaging/distros/amazonlinux/Dockerfile`:
- Around line 146-161: The builder stage is missing an ARG declaration for
FLB_IN_AMQP while the RUN cmake invocation passes
-DFLB_IN_AMQP="${FLB_IN_AMQP}", causing an empty/undefined value; add an ARG
FLB_IN_AMQP line in the builder stage (matching the pattern used for ARG
FLB_KAFKA, ARG FLB_JEMALLOC, etc.) so FLB_IN_AMQP is defined and available to
the RUN cmake command that sets -DFLB_IN_AMQP.
- Around line 103-120: Add an explicit default for FLB_IN_AMQP to match the
x86_64 variant: declare ARG FLB_IN_AMQP=Off and then ENV
FLB_IN_AMQP=$FLB_IN_AMQP in the amazonlinux-2023.arm64v8-base Dockerfile (or
alternatively install librabbitmq-devel and set FLB_IN_AMQP=On if you intend to
enable AMQP); update the Dockerfile near the other ARG/ENV declarations so the
builder stage that references FLB_IN_AMQP always sees a defined value.
In `@packaging/distros/ubuntu/Dockerfile`:
- Around line 173-186: The ubuntu-22.04-base Dockerfile RUN apt-get install
command is missing librabbitmq-dev causing AMQP plugin to be disabled; update
the package list in the RUN block (the long apt-get install -y ... line that
precedes the cmake download using environment vars CMAKE_HOME and CMAKE_VERSION)
to include librabbitmq-dev alongside the other -dev packages so the AMQP
development headers are installed for x86_64.
- Around line 251-265: The ubuntu-24.04.arm64v8-base Dockerfile RUN layer that
installs packages (the long RUN apt-get update && apt-get install -y ... &&
cmake download block which references CMAKE_HOME and CMAKE_VERSION and sets ENV
PATH) is missing librabbitmq-dev; add librabbitmq-dev to the apt-get install -y
package list alongside libpq-dev, libsasl2-dev, libssl-dev, etc., so the AMQP
plugin is available on arm64 as it is in the other ubuntu-24.04-base image.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: b1f1aeaf-b986-4e89-ac04-2cb4e57a38b2
📒 Files selected for processing (9)
.github/workflows/call-build-macos.yamlcmake/windows-setup.cmakepackaging/distros/almalinux/Dockerfilepackaging/distros/amazonlinux/Dockerfilepackaging/distros/centos/Dockerfilepackaging/distros/debian/Dockerfilepackaging/distros/raspbian/Dockerfilepackaging/distros/rockylinux/Dockerfilepackaging/distros/ubuntu/Dockerfile
✅ Files skipped from review due to trivial changes (2)
- .github/workflows/call-build-macos.yaml
- cmake/windows-setup.cmake
Signed-off-by: Matwey V. Kornilov <matwey.kornilov@gmail.com>
Signed-off-by: Matwey V. Kornilov <matwey.kornilov@gmail.com>
Signed-off-by: Matwey V. Kornilov <matwey.kornilov@gmail.com>
Signed-off-by: Matwey V. Kornilov <matwey.kornilov@gmail.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Here is initial implementation for input AMQP plugin. I guess it is stable enough to obtain some review. The plugin was tested with RabbitMQ 4.2.
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:
fluent-bit -i amqp -pqueue=flb.in_amqp -o stdoutDocumentation
Fluent Bit is licensed under Apache 2.0, by submitting this pull request I understand that this code will be released under the terms of that license.
Summary by CodeRabbit
New Features
Chores