TCP logging on FluentBit from Python

[msSarriman]

Trying to export logs to a fluent-bit TCP input server, from a Python module.

fluent-bit is setup to listen to a localhost port with a TCP server, and in Python the logging creates a SocketHandler logging handler, to redirect the logs on that port.

Python:

import logging
from logging.handlers import SocketHandler
from pythonjsonlogger import jsonlogger


format = '%(asctime)s [%(levelname)s] %(funcName)s %(process)d %(message)s'
logger = logging.getLogger(module_name)

formatter = jsonlogger.JsonFormatter(self.format)
socket_handler = SocketHandler("localhost", 5170)
logger.addHandler(socket_handler)
logger.setFormatter(formatter)

logger.info("Hello Log!")

fluent-bit.conf

[SERVICE]
    Flush        1
    Grace        30
    Log_Level    debug
    Parsers_File parsers.conf

[INPUT]
    Name        tcp
    Tag         ApplicationLogs
    Listen      0.0.0.0
    Port        5170
    Chunk_Size  1024
    Buffer_Size 10024
    format      none

[OUTPUT]
    Name    file
    Match   ApplicationLogs
    Path    /tmp/fluent-test-output.logs

parsers.conf

[PARSER]
    Name   QueryLogSeparator
    Format regex
    Regex  (?<log>-{20,})

[PARSER]
    Name   QueryLog
    Format regex
    Regex  (?<start>-{20,})(?<content>[\S\s]+?EOE)

When the format is set to json in the fluent-bit.conf, fluent-bit return the error:

[2022/07/21 10:12:56] [ warn] [input:tcp:tcp.0] invalid JSON message, skipping,

and when it set to none, the error is:

[2022/07/21 10:07:46] [debug] [input chunk] skip ingesting data with 0 bytes.

Has anybody been able to intergrate fluent-bit TCP server, with Python, and solve the above problem?

[agup006]

@cosmo0920 this issue was showing up with other tcp agents as well, I'm wondering if there is something we can add to help parse / debug these types of issues

[cosmo0920]

I'm debugging for other tcp agents' payload ingestion. Currently, I have no idea where adding debug messages.
I'll consider about this issue.

[emilio-cea]

Hi, I wanted to hop in here as well. Trying @msSarriman code I experience the same, my td-agent-bit reports only this:

[2022/10/21 10:56:18] [ info] [input chunk] skip ingesting data with 0 bytes
[2022/10/21 10:56:18] [ info] [input chunk] skip ingesting data with 0 bytes
[2022/10/21 10:56:18] [ info] [input chunk] skip ingesting data with 0 bytes
[2022/10/21 10:56:18] [ info] [input chunk] skip ingesting data with 0 bytes
[2022/10/21 10:59:35] [ info] [input chunk] skip ingesting data with 0 bytes
[2022/10/21 10:59:35] [ info] [input chunk] skip ingesting data with 0 bytes
[2022/10/21 10:59:35] [ info] [input chunk] skip ingesting data with 0 bytes
[2022/10/21 10:59:35] [ info] [input chunk] skip ingesting data with 0 bytes
[2022/10/21 11:00:55] [ info] [input chunk] skip ingesting data with 0 bytes

I tried both with a formatter and without, still no use.

Then I tried using python's `socket' module and it does seem to catch the data, at least some of it. This is the test code:

from socket import *
s=socket(AF_INET, SOCK_STREAM)
s.connect(("localhost", 6969))
string = "This is my string\n\nAFTER_IT" # AFTER IT doesn't appear on fluent bit
s.sendall(string.encode())
s.close()

And this is the result on fluent bit:

Fluent Bit v1.7.8
* Copyright (C) 2019-2021 The Fluent Bit Authors
* Copyright (C) 2015-2018 Treasure Data
* Fluent Bit is a CNCF sub-project under the umbrella of Fluentd
* https://fluentbit.io

[2022/10/21 11:07:56] [ info] [engine] started (pid=13546)
[2022/10/21 11:07:56] [ info] [storage] version=1.1.1, initializing...
[2022/10/21 11:07:56] [ info] [storage] in-memory
[2022/10/21 11:07:56] [ info] [storage] normal synchronization mode, checksum disabled, max_chunks_up=128
[2022/10/21 11:07:56] [ info] [input:tcp:tcp.0] listening on 0.0.0.0:6969
[2022/10/21 11:07:56] [ info] [sp] stream processor started
[0] tcp.0: [1666343280.823772015, {"log"=>"This is my string"}]
^C[2022/10/21 11:12:10] [engine] caught signal (SIGINT)
[2022/10/21 11:12:10] [ warn] [engine] service will stop in 5 seconds
[2022/10/21 11:12:14] [ info] [engine] service stopped

I didn't try the HTTP plugin to do logging but I don't really feel comfortable having to change my logger to perform requests instead of logging as usual.

Hope this helps and hope to see some insight about this issue.

[goatwu1993]

Thanks for this thread. Turns out logging.handlers.SockeHandler pickles the message before sending in the emit method for some reason. Remove the pickles will work.

Here's a handler which works

from logging.handlers import SocketHandler

class FluentBitHandler(SocketHandler):
    def emit(self, record):
        try:
            self.send((self.format(record)).encode())
        except Exception:
            self.handleError(record)

Of source you need to make sure the format is JSON.