# Input from kafka to logstash
**URL:** https://discuss.elastic.co/t/input-from-kafka-to-logstash/184738
**Category:** Logstash
**Created:** [June 7, 2019, 10:39am UTC](https://discuss.elastic.co/t/input-from-kafka-to-logstash/184738 "2019-06-07T10:39:26Z")
**Posts on this page:** 6
**Page:** 1
### Author:  [@Harsha606](https://discuss.elastic.co/u/Harsha606)
#### Post date: [June 7, 2019, 10:39am UTC](https://discuss.elastic.co/t/input-from-kafka-to-logstash/184738/1 "2019-06-07T10:39:26Z")
Hi All,
my architecture is as below
filebeat -\> kafka -\> logstash -\> elastic
from, filebeat i am successfully pushing logs to kafka, what i observed in kafka topic message is **log":{"file":{"path":**"/mnt/volume0/XXXX-v2-free-service/services/free-v-2-0-13/namesrv/logs/name-srv-express-info1.log"},"offset":4501449},"level":"info"}
now my question is, while considering kafka input to logstash, can i do a filter based on **log":{"file":{"path":** , why i need that is because, as you observe in the folder structure there is folder named **free-v-2-0-13** , based on that name i need to add field called version and that folder name changes dynamically!!!
if this can be done, plz explain how....
any help would be really appreciated.
---
### Author:  [@Harsha606](https://discuss.elastic.co/u/Harsha606)
#### Post date: [June 12, 2019, 9:57am UTC](https://discuss.elastic.co/t/input-from-kafka-to-logstash/184738/2 "2019-06-12T09:57:16Z")
can any one help me on this plz?
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [June 12, 2019, 1:56pm UTC](https://discuss.elastic.co/t/input-from-kafka-to-logstash/184738/3 "2019-06-12T13:56:19Z")
Assuming you have valid JSON coming from kafka, you can use a json filter to parse it. Then you can use grok or dissect to extract the folder name from [log][file][path]
---
### Author:  [@Harsha606](https://discuss.elastic.co/u/Harsha606)
#### Post date: [June 13, 2019, 9:18am UTC](https://discuss.elastic.co/t/input-from-kafka-to-logstash/184738/4 "2019-06-13T09:18:38Z")
thanks Badger.. it helped me lot!!!
---
### Author:  [@Harsha606](https://discuss.elastic.co/u/Harsha606)
#### Post date: [June 13, 2019, 9:20am UTC](https://discuss.elastic.co/t/input-from-kafka-to-logstash/184738/5 "2019-06-13T09:20:07Z")
for someone, who is facing same issue, here is the solution for it..
filter {
grok {
match =\> { "[log][file][path]" =\> "/mnt/volume0/somename-v[0-9]-free-service/services/(?[^/]+)/" }
}
}
---
### Author:  [@system](https://discuss.elastic.co/u/system)
#### Post date: [July 11, 2019, 9:20am UTC](https://discuss.elastic.co/t/input-from-kafka-to-logstash/184738/6 "2019-07-11T09:20:12Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.