How to extract some part of event by flume?

[#1]

I used npm_syslog_client to produce syslog and then send it into Metron 0.4.1 by flume.

I set the configuration of flume as:

flume-config.png

according to { udp_client <target> <port> <facility> <severity> <syslog-hostname> <message> } I sent my syslog as:

node udp-client.js "localhost"  5424 1 2 localhost 07/14/18-16:49:25.210057 ,1,10000001,1,"icmp test",,fe80::34ac:7d5c:8b09:7c5c,,ff02::1:ff72:6111,,00:21:9B:D4:1C:69,33:33:FF:72:61:11,0x56,,,,,,255,0,0,32,32768,135,0,0,0

in which my message is { 07/14/18-16:49:25.210057 ,1,10000001,1,"icmp test",,fe80::34ac:7d5c:8b09:7c5c,,ff02::1:ff72:6111,,00:21:9B:D4:1C:69,33:33:FF:72:61:11,0x56,,,,,,255,0,0,32,32768,135,0,0,0 }

After I sent the syslog, I found this message in elasticsearch's topic error:

"_index": "error_index_2018.08.21.09","_type": "error_doc","_id": "AWVa18dBx1ZPgA468FnU","_version": 1,"_score": 1,"_timestamp": 1534827414380,"_source": {
"exception": "java.lang.IllegalStateException: Unable to parse message: <NaN>Aug 21 09:26:51 10.0.1.152 07/14/18-16:49:27.906780 ,1,10000001,1,"icmp test",,fe80::a4:45c8:46c2:9aeb,,ff02::16,,1C:5C:F2:96:BD:6D,33:33:00:00:00:16,0x5A,,,,,,1,0,84017152,36,36864,143,0,0,1",

In order to solve this issue, I want to omit the part <07/14/18-16:49:25.210057> from the syslog. but I don't know how to do it in flume

Thanks for answering my question.

[#2]

@mojgan ghasemi

Looks like the sink is Kafka topic and the record should be deleted in Kafka not in Flume.

further what is the kafka version you are using? If it is <0.11 we cannot delete a specific message from topic, We'd need to delete the topic and recreate one or set the retention period to 0 and revert it back. If kafka version is >=0.11 please refer this SO answer with commands.

[#3]

@Sandeep Nemuri

Thank you for your reply.

I want to extract some parts of message and send it to Elasticsearch.

I configured snort to produce CSV format and I sent it by NiFi.

If you are familiar with Apache Metron, there is Snort sensor in Kafka. I use it to send data to Storm and Elasticsearch.

I want to use Flume for transferring snort's syslog message because this link suggest it.

I create syslog client with npm_syslog_client and transfer it by Flume, I want to extract some parts of message(my message is csv format) not all of them like the example that I wrote above.

thanks.

[#4]

Flume has a mechanism called Interceptor, i.e. some optionally chained

____________________

Assignment Help UAE

[#5]

@Henry Raco

I know about it but I don't how can I use it to solve my problem.

thank you