[rsyslog-notify] Forum Thread: omkafka segfaults - (Mode 'post')

[noreply at adiscon.com]

User: goodleaf 
Forumlink: http://kb.monitorware.com/viewtopic.php?p=25619#p25619

Message: 
----------
Hi,

While using rsyslog to receive arcsight events (tcp syslog) and adding the
raw message to kafka (without any cef parsing) using omkafka, I noticed
that once the throughput got high, omkafka (or actually librdkafka.so.1)
segfaults.

The configuration is quite simple:

[code:jhhr7dte]module(load="imtcp")
module(load="omkafka")
input(type="imtcp" port="60055" ruleset="arcsight")

template(
  name="cef"
  type="string"
  string="%msg%"
)

ruleset(name="arcsight") {
  action(
    type="omkafka"
    topic="arcsight_test"
    confParam="compression.codec=gzip"
    partitions.auto="on"
    partitions.number="6"
   
broker="10.0.0.1:6667,10.0.0.2:6667,10.0.0.3:6667"
    template="cef"
  )
}[/code:jhhr7dte]

Script to test (it triggers on real events, but I have tried to reproduce
with as simple data as possible. The following python script will trigger
the segfault for me):
[code:jhhr7dte]#!/usr/bin/python
from __future__ import with_statement
import socket

FACILITY = {
	'kern': 0, 'user': 1, 'mail': 2, 'daemon': 3,
	'auth': 4, 'syslog': 5, 'lpr': 6, 'news': 7,
	'uucp': 8, 'cron': 9, 'authpriv': 10, 'ftp': 11,
	'local0': 16, 'local1': 17, 'local2': 18, 'local3': 19,
	'local4': 20, 'local5': 21, 'local6': 22, 'local7': 23,
}

LEVEL = {
	'emerg': 0, 'alert':1, 'crit': 2, 'err': 3,
	'warning': 4, 'notice': 5, 'info': 6, 'debug': 7
}

class TCPSyslog:
    def __init__(self, host='localhost', port=10001):
        self.sock = None
        self.host = host
        self.port = port

    def open(self):
        self.sock = socket.socket(socket.AF_INET,
socket.SOCK_STREAM)
        self.sock.connect((self.host, self.port))
        
    def close(self):
        self.sock.close()

    def syslog(self, message, level=LEVEL['info'],
facility=FACILITY['user']):
        data = ('<%d>Sep 19 08:26:10 host %s\r\n' % (level +
facility*8, message)).encode('ascii')
        self.sock.send(data)

def main():
    syslog = TCPSyslog(host='10.0.0.100', port=60055)
    syslog.open()
    
    for x in range(1000000):
        msg = "A\r\n"*1000
        syslog.syslog(msg)

    syslog.close()


if __name__ == "__main__":
	main()[/code:jhhr7dte]

Errors (dmesg):
[quote:jhhr7dte]rs:main Q:Reg[43607]: segfault at 290 ip 00002ba5d11b3110
sp 00002ba63a91a9b8 error 4 in librdkafka.so.1[2ba5d11af000+26000]
rs:main Q:Reg[44120]: segfault at 290 ip 00002b1b81c5f110 sp
00002b1b941ff9b8 error 4 in librdkafka.so.1[2b1b81c5b000+26000]
rs:main Q:Reg[44576]: segfault at 290 ip 00002b41f7148110 sp
00002b41f7d759b8 error 4 in librdkafka.so.1[2b41f7144000+26000]
rs:main Q:Reg[45310]: segfault at 290 ip 00002baa9196f110 sp
00002baacc3339b8 error 4 in librdkafka.so.1[2baa9196b000+26000]
rs:main Q:Reg[45554]: segfault at 290 ip 00002b76563ad110 sp
00002b76c045c9b8 error 4 in librdkafka.so.1[2b76563a9000+26000]
rs:main Q:Reg[45761]: segfault at 290 ip 00002b2c9c2eb110 sp
00002b2ca86009b8 error 4 in librdkafka.so.1[2b2c9c2e7000+26000]
rs:main Q:Reg[31211]: segfault at 290 ip 00002ae96d311110 sp
00002ae9c81ff9b8 error 4 in librdkafka.so.1[2ae96d30d000+26000]
rs:main Q:Reg[42793]: segfault at 290 ip 00002b4296952110 sp
00002b42ec1ff9b8 error 4 in librdkafka.so.1[2b429694e000+26000]
rs:main Q:Reg[46188]: segfault at 290 ip 00002b20319b5110 sp
00002b20781ff9b8 error 4 in librdkafka.so.1[2b20319b1000+26000]
rs:main Q:Reg[46784]: segfault at 290 ip 00002b2f8fcef110 sp
00002b2fc04009b8 error 4 in librdkafka.so.1[2b2f8fceb000+26000]
rs:main Q:Reg[46858]: segfault at 290 ip 00002b7ed58bc110 sp
00002b7edc6009b8 error 4 in librdkafka.so.1[2b7ed58b8000+26000]
rs:main Q:Reg[47880]: segfault at 290 ip 00002ae753379110 sp
00002ae760e029b8 error 4 in librdkafka.so.1[2ae753375000+26000]
rs:main Q:Reg[48136]: segfault at 290 ip 00002aca6af22110 sp
00002aca7c6009b8 error 4 in librdkafka.so.1[2aca6af1e000+26000]
rs:main Q:Reg[48175]: segfault at 290 ip 00002b82ebc94110 sp
00002b82fd0049b8 error 4 in
librdkafka.so.1[2b82ebc90000+26000][/quote:jhhr7dte]

Running Redhat 6 with:

[quote:jhhr7dte]rsyslogd 8.9.0.ad1, compiled with:
        PLATFORM:                               x86_64-redhat-linux-gnu
        PLATFORM (lsb_release -d):
        FEATURE_REGEXP:                         Yes
        GSSAPI Kerberos 5 support:              No
        FEATURE_DEBUG (debug build, slow code): No
        32bit Atomic operations supported:      Yes
        64bit Atomic operations supported:      Yes
        memory allocator:                       system default
        Runtime Instrumentation (slow code):    No
        uuid support:                           Yes
        Number of Bits in RainerScript integers: 64
[/quote:jhhr7dte]