# Process two types of logs format
**URL:** https://discuss.elastic.co/t/process-two-types-of-logs-format/366021
**Category:** Logstash
**Created:** [September 4, 2024, 9:48am UTC](https://discuss.elastic.co/t/process-two-types-of-logs-format/366021 "2024-09-04T09:48:03Z")
**Posts on this page:** 7
**Page:** 1
### Author:  [@ELK\_enjoyer](https://discuss.elastic.co/u/ELK_enjoyer)
#### Post date: [September 4, 2024, 9:48am UTC](https://discuss.elastic.co/t/process-two-types-of-logs-format/366021/1 "2024-09-04T09:48:04Z")
Hi everyone!
I'm using the ELK stack with Kafka to collect and analyze logs from my K8s environment.
We have different log formats, mainly plaintext and JSON.
How can I process them in Logstash properly?
Right now, plain logs have the wrong breakdown.
For example, a plaintext message is breaking into two or more messages.
My simplified config is :
```auto
input {
kafka {
bootstrap_servers => "kafka_ip:port"
topics => ["sample_topic"]
group_id => "elk"
client_id => "logstash-sample"
auto_offset_reset => "earliest"
consumer_threads => 4
codec => json
decorate_events => true
add_field => { "kafka_topic" => "sample_topic"}
}
output {
if [kafka_topic] == "sample_topic" {
elasticsearch {
hosts => ["localhost:9200"]
index => "sample_topic-%{+YYYY.MM.dd}"
}
}
```
Would appreciate any help,
Thanks in advance.
---
### Author:  [@ashishtiwari1993](https://discuss.elastic.co/u/ashishtiwari1993)
#### Post date: [September 4, 2024, 12:22pm UTC](https://discuss.elastic.co/t/process-two-types-of-logs-format/366021/2 "2024-09-04T12:22:47Z")
Could you please give more details like how your plain text logs looks like and how you want to format them?
---
### Author:  [@ELK\_enjoyer](https://discuss.elastic.co/u/ELK_enjoyer)
#### Post date: [September 5, 2024, 9:22am UTC](https://discuss.elastic.co/t/process-two-types-of-logs-format/366021/3 "2024-09-05T09:22:43Z")
Imagine for example httpd log sample entry like this :
127.0.0.1 - frank [10/Oct/2023:13:55:36 -0700] "GET /apache\_pb.gif HTTP/1.0" 200 2326
I am getting aforementioned entry in two entries, like :
entry1: 127.0.0.1 - frank [10/Oct/2023:13:55:36 -0700]
entry2: "GET /apache\_pb.gif HTTP/1.0" 200 2326
It is obviously wrong, I want to receive it at least as it was originally
---
### Author:  [@ELK\_enjoyer](https://discuss.elastic.co/u/ELK_enjoyer)
#### Post date: [September 23, 2024, 6:29am UTC](https://discuss.elastic.co/t/process-two-types-of-logs-format/366021/4 "2024-09-23T06:29:51Z")
Anyone got any ideas ?
---
### Author:  [@leandrojmp](https://discuss.elastic.co/u/leandrojmp)
#### Post date: [September 23, 2024, 12:46pm UTC](https://discuss.elastic.co/t/process-two-types-of-logs-format/366021/5 "2024-09-23T12:46:10Z")
> [@ELK\_enjoyer](#):
>
> Imagine for example httpd log sample entry like this :
> 127.0.0.1 - frank [10/Oct/2023:13:55:36 -0700] "GET /apache\_pb.gif HTTP/1.0" 200 2326
> I am getting aforementioned entry in two entries, like :
>
> entry1: 127.0.0.1 - frank [10/Oct/2023:13:55:36 -0700]
> entry2: "GET /apache\_pb.gif HTTP/1.0" 200 2326
>
> It is obviously wrong, I want to receive it at least as it was originally
What is sending this log to your Kafka? This seems to be an issue in your producer, you need to fix it before sending to kafka.
Logstash will consume your messages as they are in Kafka.
---
### Author:  [@ELK\_enjoyer](https://discuss.elastic.co/u/ELK_enjoyer)
#### Post date: [September 23, 2024, 1:36pm UTC](https://discuss.elastic.co/t/process-two-types-of-logs-format/366021/6 "2024-09-23T13:36:28Z")
There is no issues with producer since I am getting all the logs.
Filebeat in k8s are sending this to kafka.
I experience no issues when a log has json format
---
### Author:  [@leandrojmp](https://discuss.elastic.co/u/leandrojmp)
#### Post date: [September 23, 2024, 1:46pm UTC](https://discuss.elastic.co/t/process-two-types-of-logs-format/366021/7 "2024-09-23T13:46:13Z")
> [@ELK\_enjoyer](#):
>
> Filebeat in k8s are sending this to kafka.
> I experience no issues when a log has json format
Yeah, but how it is sending it?
Logstash does not split anything into multiple events unless explicitly configured to do so.
If you have one event with this message `127.0.0.1 - frank [10/Oct/2023:13:55:36 -0700]` and another event with this message `"GET /apache_pb.gif HTTP/1.0" 200 2326`, this means that you have one kafka message for each.
Logstash just consumes the messages in the way they are stored in your Kafka.
If you have a message like this in Kafka:
```auto
127.0.0.1 - frank [10/Oct/2023:13:55:36 -0700] "GET /apache_pb.gif HTTP/1.0" 200 2326
```
Then you will have an event in logstash with this entire message.
Please provide some examples on how your filebeat is configured and how your output looks like in Elasticsearch, show an example of this message being split in two.