Fluentbit UDP connection

[brunooliveiramac]

I'm trying to receive udp messages.
But I'm not getting anything being output. That's the only log I have:

[2024/07/16 23:19:56] [trace] [input:udp:udp.2 at /src/fluent-bit/plugins/in_udp/udp_conn.c:397] read()=149 pre_len=0 now_len=149
[2024/07/16 23:19:56] [debug] [input chunk] skip ingesting data with 0 bytes
[2024/07/16 23:19:56] [trace] [input:udp:udp.2 at /src/fluent-bit/plugins/in_udp/udp_conn.c:397] read()=148 pre_len=0 now_len=148
[2024/07/16 23:19:56] [debug] [input chunk] skip ingesting data with 0 bytes

and this is my config:

apiVersion: v1
kind: ConfigMap
metadata:
  name: fluent-bit-config
data:
  fluent-bit.conf: |
    [INPUT]
        Name         tcp
        Listen       0.0.0.0
        Port         24224
        Format       none

    [INPUT]
        Name         tcp
        Listen       0.0.0.0
        Port         515
        Format       none
        Buffer_Size  1062768KB   
        Chunk_Size   1036384KB   
        Threaded      on
    
    [INPUT]
        Name         udp
        Listen       0.0.0.0
        Port         515
        Format       none
    
    [SERVICE]
        HTTP_Server  On
        HTTP_Listen  0.0.0.0
        HTTP_PORT    2020
        Health_Check On 
        HC_Errors_Count 5 
        HC_Retry_Failure_Count 5 
        HC_Period 5
        Log_Level    trace
        storage.path              /storage
        storage.metrics           on
        storage.backlog.mem_limit 500M
    
    [OUTPUT]
        Name         stdout
        Match        *    

I'm using the latest version of FluentBit. Would any of you have any idea? Appreciate.

[brunooliveiramac]

[2024/07/17 02:47:45] [trace] [input:udp:udp.2 at /src/fluent-bit/plugins/in_udp/udp_conn.c:379] udp_conn_event triggered
[2024/07/17 02:47:45] [trace] [input:udp:udp.2 at /src/fluent-bit/plugins/in_udp/udp_conn.c:419] read()=196 pre_len=0 now_len=196
[2024/07/17 02:47:45] [trace] [input:udp:udp.2 at /src/fluent-bit/plugins/in_udp/udp_conn.c:289] Starting parse_payload_none
[2024/07/17 02:47:45] [trace] [input:udp:udp.2 at /src/fluent-bit/plugins/in_udp/udp_conn.c:296] Initial buffer content: <134>1 1721184465.371896644 XXXXXXXXXXXXXX ip_flow_end src=192.168.1.9 dst=XXXXXXXXXXXXXX protocol=udp sport=42392 dport=15600 translated_dst_ip=192.168.1.255 translated_port=15600
[2024/07/17 02:47:45] [trace] [input:udp:udp.2 at /src/fluent-bit/plugins/in_udp/udp_conn.c:297] Using separator:
 (length=1)
[2024/07/17 02:47:45] [debug] [input chunk] skip ingesting data with 0 bytes
[2024/07/17 02:47:45] [trace] [input:udp:udp.2 at /src/fluent-bit/plugins/in_udp/udp_conn.c:349] Log append success, total consumed=0
[2024/07/17 02:47:45] [trace] [input:udp:udp.2 at /src/fluent-bit/plugins/in_udp/udp_conn.c:356] Remaining buffer content: <134>1 1721184465.371896644 XXXXXXXXXXXXXX ip_flow_end src=XXXXXXXXXXXXXX dst=XXXXXXXXXXXXXXprotocol=udp sport=42392 dport=15600 translated_dst_ip=192.168.1.255 translated_port=15600
[2024/07/17 02:47:45] [trace] [input:udp:udp.2 at /src/fluent-bit/plugins/in_udp/udp_conn.c:450] Incomplete payload, waiting for more data
[2024/07/17 02:47:45] [trace] [input:udp:udp.2 at /src/fluent-bit/plugins/in_udp/udp_conn.c:379] udp_conn_event triggered
[2024/07/17 02:47:45] [trace] [input:udp:udp.2 at /src/fluent-bit/plugins/in_udp/udp_conn.c:419] read()=195 pre_len=0 now_len=195
[2024/07/17 02:47:45] [trace] [input:udp:udp.2 at /src/fluent-bit/plugins/in_udp/udp_conn.c:289] Starting parse_payload_none
[2024/07/17 02:47:45] [trace] [input:udp:udp.2 at /src/fluent-bit/plugins/in_udp/udp_conn.c:296] Initial buffer content: <134>1 1721184465.371910851 XXXXXXXXXXXXXX ip_flow_end src=XXXXXXXXXXXXXX dst=0.0.0.0. protocol=tcp sport=63471 dport=5223 translated_src_ip=0.0.0.0 translated_port=63471
[2024/07/17 02:47:45] [trace] [input:udp:udp.2 at /src/fluent-bit/plugins/in_udp/udp_conn.c:297] Using separator:
 (length=1)
[2024/07/17 02:47:45] [debug] [input chunk] skip ingesting data with 0 bytes
[2024/07/17 02:47:45] [trace] [input:udp:udp.2 at /src/fluent-bit/plugins/in_udp/udp_conn.c:349] Log append success, total consumed=0
[2024/07/17 02:47:45] [trace] [input:udp:udp.2 at /src/fluent-bit/plugins/in_udp/udp_conn.c:356] Remaining buffer content: <134>1 1721184465.371910851 XXXXXXXXXXXXXX ip_flow_end src=0.0.0.0 dst=0.0.0.0. protocol=tcp sport=63471 dport=5223 translated_src_ip=0.0.0.0. translated_port=63471
[2024/07/17 02:47:45] [trace] [input:udp:udp.2 at /src/fluent-bit/plugins/in_udp/udp_conn.c:450] Incomplete payload, waiting for more data
[

seems not reaching the end of the line, to enter the while loop:

static ssize_t parse_payload_none(struct udp_conn *conn)
{
    int ret;
    int len;
    int sep_len;
    size_t consumed = 0;
    char *buf;
    char *s;
    char *separator;
    struct flb_in_udp_config *ctx;

    ctx = conn->ctx;

    separator = conn->ctx->separator;
    sep_len = flb_sds_len(conn->ctx->separator);

    buf = conn->buf_data;
    ret = FLB_EVENT_ENCODER_SUCCESS;

    flb_log_event_encoder_reset(ctx->log_encoder);
    flb_plg_trace(ctx->ins, "Starting parse_payload_none");

    if (ctx->separator == NULL || flb_sds_len(ctx->separator) == 0) {
        ctx->separator = flb_sds_create("\n");
    }

    // Log the initial buffer content and separator
    flb_plg_trace(ctx->ins, "Initial buffer content: %.*s", (int)conn->buf_len, buf);
    flb_plg_trace(ctx->ins, "Using separator: %s (length=%d)", separator, sep_len);

    while ((s = strstr(buf, separator))) {
        len = (s - buf);
        flb_plg_trace(ctx->ins, "Found separator at len=%d", len);

        if (len == 0) {
            flb_plg_trace(ctx->ins, "Zero length log entry found, breaking loop");
            break;
        }
        else if (len > 0) {
            flb_plg_trace(ctx->ins, "Processing log entry of len=%d", len);
            ret = flb_log_event_encoder_begin_record(ctx->log_encoder);
            flb_plg_trace(ctx->ins, "Begin record: ret=%d", ret);

            if (ret == FLB_EVENT_ENCODER_SUCCESS) {
                ret = flb_log_event_encoder_set_current_timestamp(ctx->log_encoder);
                flb_plg_trace(ctx->ins, "Set current timestamp: ret=%d", ret);
            }

            if (ret == FLB_EVENT_ENCODER_SUCCESS) {
                ret = flb_log_event_encoder_append_body_values(
                        ctx->log_encoder,
                        FLB_LOG_EVENT_CSTRING_VALUE("log"),
                        FLB_LOG_EVENT_STRING_VALUE(buf, len));
                flb_plg_trace(ctx->ins, "Append body values: ret=%d", ret);
            }

            if (ret == FLB_EVENT_ENCODER_SUCCESS) {
                ret = flb_log_event_encoder_commit_record(ctx->log_encoder);
                flb_plg_trace(ctx->ins, "Commit record: ret=%d", ret);
            }

            if (ret != FLB_EVENT_ENCODER_SUCCESS) {
                flb_plg_error(ctx->ins, "Error committing record: ret=%d", ret);
                break;
            }

            consumed += len + sep_len;
            buf += len + sep_len;
            flb_plg_trace(ctx->ins, "Consumed %zu bytes, buffer moved forward", consumed);
        }
        else {
            flb_plg_trace(ctx->ins, "Unexpected len=%d, breaking loop", len);
            break;
        }
    }

    // Copy remaining incomplete message to the beginning of the buffer
    if (conn->buf_len > consumed) {
        memmove(conn->buf_data, buf, conn->buf_len - consumed);
    }
    conn->buf_len -= consumed;
    conn->buf_data[conn->buf_len] = '\0';

    if (ret == FLB_EVENT_ENCODER_SUCCESS) {
        flb_input_log_append(conn->ins, NULL, 0,
                             ctx->log_encoder->output_buffer,
                             ctx->log_encoder->output_length);
        flb_plg_trace(ctx->ins, "Log append success, total consumed=%zu", consumed);
    }
    else {
        flb_plg_error(ctx->ins, "Log event encoding error: %d", ret);
    }

    // Log the remaining buffer content
    flb_plg_trace(ctx->ins, "Remaining buffer content: %.*s", (int)(conn->buf_len - consumed), buf);

    return consumed;
}