# Create multiple indexes with single logstash configuration file depending upon the kafka channel

**URL:** <https://discuss.elastic.co/t/create-multiple-indexes-with-single-logstash-configuration-file-depending-upon-the-kafka-channel/177003>\
**Category:** Logstash\
**Created:** [April 16, 2019, 6:17am UTC](https://discuss.elastic.co/t/create-multiple-indexes-with-single-logstash-configuration-file-depending-upon-the-kafka-channel/177003 "2019-04-16T06:17:01Z")\
**Posts on this page:** 6\
**Page:** 1

<div class="post-metadata">

**Author:** ![Nandita\_Rane](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/nandita_rane/32/44214_2.png) [@Nandita\_Rane](https://discuss.elastic.co/u/Nandita_Rane)\
**Post date:** [April 16, 2019, 6:17am UTC](https://discuss.elastic.co/t/create-multiple-indexes-with-single-logstash-configuration-file-depending-upon-the-kafka-channel/177003/1 "2019-04-16T06:17:01Z")

</div>

Please help me to resolve this issue.

Issue Description: I want to create a logstash configuration which will put logs in two different index.

I tried with this configuration but not able to create index in logstash.

please find the logstash configuration here,

input {  
kafka {  
codec =\> 'json'  
bootstrap\_servers =\>['']  
topics =\> 'application-logs'  
auto\_offset\_reset =\> 'earliest'  
group\_id =\> 'logstash-test'  
}  
}  
filter {  
date {  
match =\> ["[Kafka][EventTimestamp]", "yyyy-MM-dd-HH:mm:ss.SSSSSS"]  
timezone =\> "Asia/Kolkata"  
}  
if [org\_name] == "\*\*\*gg-apic-uat" {  
if [status\_code] == "200 OK" {  
mutate {  
add\_field =\> {"[Kafka][EventType]" =\> "LOG"}  
}  
} else {  
mutate {  
add\_field =\> {"[Kafka][EventType]" =\> "ERROR"}  
}  
}  
ruby {  
code =\> "event.set('datetime2', DateTime.parse(event.get('[datetime]')).strftime('%Y-%m-%d-%H:%M:%S.%L'))"  
}  
mutate {  
add\_field =\> {"[Kafka][Channel]" =\> "APIConnect"}  
add\_field =\> {"[Kafka][TransactionTimeStamp]" =\> "%{[datetime2]}"}  
add\_field =\> {"[Kafka][TransactionID]" =\> "%{[transaction\_id]}"}  
add\_field =\> {"[Kafka][EventTimestamp]" =\> "%{[datetime2]}"}  
add\_field =\> {"[Kafka][Code]" =\> "%{[status\_code]}"}  
add\_field =\> {"[Kafka][Comoponent]" =\> "%{[api\_name]}"}  
add\_field =\> {"[Kafka][Application]" =\> "%{[org\_name]}"}  
add\_field =\> {"[Kafka][Node]" =\> "%{[uri\_path]}"}  
add\_field =\> {"[Kafka][SourceIP]" =\> "%{[client\_ip]}"}  
add\_field =\> {"[Kafka][Server]" =\> "%{[host]}"}  
add\_field =\> {"[Kafka][Description]" =\> "%{status\_code}"}  
}  
# prune {  
# whitelist\_names =\> ["Kafka","@version","@timestamp"]  
# }  
}  
if ![Kafka] {  
drop {}  
}  
if ![Kafka][EventType] {  
drop{}  
}  
if [Kafka][Channel] == "" {  
drop {}  
}  
mutate {  
add\_field =\> ["[@metadata][DateSuffix]", ""]  
}  
ruby {  
code =\> "event.set('[@metadata][DateSuffix]', DateTime.now.strftime('%Y-%m-%d')) "  
}  
ruby {  
#code =\> "event.set('message\_size', event.get('[Kafka][OriginalPayload]').bytesize)"  
code =\> "event.set('message\_size', event.get('[Kafka]').to\_s.bytesize)"  
}  
}  
output {  
if ([Kafka][Channel] == "bulk-emal" or [Kafka][Channel] == "bulk-push") {  
elasticsearch {  
codec =\> json  
hosts=\> "http://"  
index=\>"test-bulk-logs-%{[@metadata][DateSuffix]}"  
document\_type =\> "logs"  
}  
}  
else  
{  
elasticsearch {  
codec =\> json  
hosts=\> "http:"  
index=\>"test-logs-%{[@metadata][DateSuffix]}"  
document\_type =\> "logs"

}  
}  
}

---

<div class="post-metadata">

**Author:** ![staodd](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/staodd/32/24509_2.png) [@staodd](https://discuss.elastic.co/u/staodd)\
**Post date:** [April 16, 2019, 6:41am UTC](https://discuss.elastic.co/t/create-multiple-indexes-with-single-logstash-configuration-file-depending-upon-the-kafka-channel/177003/2 "2019-04-16T06:41:12Z")

</div>

Did you try to replace both outputs with just a file to verify that your documents look like you expect? Always good to confirm that the docs are correct before you implement the if-then clauses.

---

<div class="post-metadata">

**Author:** ![Nandita\_Rane](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/nandita_rane/32/44214_2.png) [@Nandita\_Rane](https://discuss.elastic.co/u/Nandita_Rane)\
**Post date:** [April 16, 2019, 6:54am UTC](https://discuss.elastic.co/t/create-multiple-indexes-with-single-logstash-configuration-file-depending-upon-the-kafka-channel/177003/3 "2019-04-16T06:54:05Z")

</div>

Hi sir,  
Thanks for quick reply,

the else loop in oouput is getting executed and test-log-\* index is getting created  
but the if loop is simply ignored by logstash and not throwing any error or exception

Your guidance will be of great help as I am stuck here since 4 days.

Thanks and regards,  
Nandita Rasam.

---

<div class="post-metadata">

**Author:** ![staodd](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/staodd/32/24509_2.png) [@staodd](https://discuss.elastic.co/u/staodd)\
**Post date:** [April 16, 2019, 10:56am UTC](https://discuss.elastic.co/t/create-multiple-indexes-with-single-logstash-configuration-file-depending-upon-the-kafka-channel/177003/4 "2019-04-16T10:56:29Z")

</div>

I dont think the IF loop is ignored, but maybe the conditions are not met.. Could you paste an anonymized version of a document so we get to see a real example of one.

---

<div class="post-metadata">

**Author:** ![Nandita\_Rane](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/nandita_rane/32/44214_2.png) [@Nandita\_Rane](https://discuss.elastic.co/u/Nandita_Rane)\
**Post date:** [April 17, 2019, 5:59am UTC](https://discuss.elastic.co/t/create-multiple-indexes-with-single-logstash-configuration-file-depending-upon-the-kafka-channel/177003/5 "2019-04-17T05:59:13Z")

</div>

Hello Team,

Thanks for your help.

Issue is resolved.

Solution:  
I checked kibana json and gave proper json fields in if condition.

Please find the solution here for future reference.

* * *

input {  
kafka {  
codec =\> 'json'  
bootstrap\_servers =\>['xx.xxx.xxx.xx:9092']  
topics =\> 'test-logs'  
auto\_offset\_reset =\> 'earliest'  
group\_id =\> 'logstash-XXXX'  
}  
}  
filter {

```
    date {
            match => ["[Kafka][EventTimestamp]", "yyyy-MM-dd-HH:mm:ss.SSSSSS"]
            timezone => "Asia/Kolkata"
    }

    if [org_name] == "XXX-XXXXX-XX" {
            if [status_code] == "200 OK" {
                    mutate {
                            add_field => {"[Kafka][EventType]" => "LOG"}
                    }
            } else {
                    mutate {
                            add_field => {"[Kafka][EventType]" => "ERROR"}
                    }
            }
            ruby {
                    code => "event.set('datetime2', DateTime.parse(event.get('[datetime]')).strftime('%Y-%m-%d-%H:%M:%S.%L'))"
            }
            mutate {
                    add_field => {"[Kafka][Channel]" => "APIConnect"}
                    add_field => {"[Kafka][TransactionTimeStamp]" => "%{[datetime2]}"}
                    add_field => {"[Kafka][TransactionID]" => "%{[transaction_id]}"}
                    add_field => {"[Kafka][EventTimestamp]" => "%{[datetime2]}"}
                    add_field => {"[Kafka][Code]" => "%{[status_code]}"}
                    add_field => {"[Kafka][Comoponent]" => "%{[api_name]}"}
                    add_field => {"[Kafka][Application]" => "%{[org_name]}"}
                    add_field => {"[Kafka][Node]" => "%{[uri_path]}"}
                    add_field => {"[Kafka][SourceIP]" => "%{[client_ip]}"}
                    add_field => {"[Kafka][Server]" => "%{[host]}"}
                    add_field => {"[Kafka][Description]" => "%{status_code}"}
            }               
    }

    if ![Kafka] {
            drop {}
    }
    if ![Kafka][EventType] {
            drop{}
    }
    if [Kafka][Channel] == "" {
            drop {}
    }
    mutate {
            add_field => ["[@metadata][DateSuffix]", ""]
    }
    ruby {
            code => "event.set('[@metadata][DateSuffix]', DateTime.now.strftime('%Y-%m-%d')) "
    }
    ruby {                
            code => "event.set('message_size', event.get('[Kafka]').to_s.bytesize)"
    }

```

}

output {  
if ([Kafka][SubComponent]=="PPPPP" or [Kafka][SubComponent]=="QQQQ" or [Kafka][SubComponent]=="RRRR") {  
elasticsearch {  
codec =\> json  
hosts=\> "[http://xx.xxx.xxx.xx:9200](http://xx.xxx.xxx.xx:9200)"  
index=\>"test-logs-%{[@metadata][DateSuffix]}"  
document\_type =\> "logs"  
}  
}  
else {  
elasticsearch {  
codec =\> json  
hosts=\> "[http://xx.xx.xxx.xx:9200](http://xx.xx.xxx.xx:9200)"  
index=\>"rest-logs-%{[@metadata][DateSuffix]}"  
document\_type =\> "logs"  
}  
}  
stdout {  
codec =\> rubydebug  
}  
}input {  
kafka {  
codec =\> 'json'  
bootstrap\_servers =\>['xx.xxx.xxx.xx:9092']  
topics =\> 'test-logs'  
auto\_offset\_reset =\> 'earliest'  
group\_id =\> 'logstash-XXXX'  
}  
}  
filter {

```
    date {
            match => ["[Kafka][EventTimestamp]", "yyyy-MM-dd-HH:mm:ss.SSSSSS"]
            timezone => "Asia/Kolkata"
    }

    if [org_name] == "XXX-XXXXX-XX" {
            if [status_code] == "200 OK" {
                    mutate {
                            add_field => {"[Kafka][EventType]" => "LOG"}
                    }
            } else {
                    mutate {
                            add_field => {"[Kafka][EventType]" => "ERROR"}
                    }
            }
            ruby {
                    code => "event.set('datetime2', DateTime.parse(event.get('[datetime]')).strftime('%Y-%m-%d-%H:%M:%S.%L'))"
            }
            mutate {
                    add_field => {"[Kafka][Channel]" => "APIConnect"}
                    add_field => {"[Kafka][TransactionTimeStamp]" => "%{[datetime2]}"}
                    add_field => {"[Kafka][TransactionID]" => "%{[transaction_id]}"}
                    add_field => {"[Kafka][EventTimestamp]" => "%{[datetime2]}"}
                    add_field => {"[Kafka][Code]" => "%{[status_code]}"}
                    add_field => {"[Kafka][Comoponent]" => "%{[api_name]}"}
                    add_field => {"[Kafka][Application]" => "%{[org_name]}"}
                    add_field => {"[Kafka][Node]" => "%{[uri_path]}"}
                    add_field => {"[Kafka][SourceIP]" => "%{[client_ip]}"}
                    add_field => {"[Kafka][Server]" => "%{[host]}"}
                    add_field => {"[Kafka][Description]" => "%{status_code}"}
            }               
    }

    if ![Kafka] {
            drop {}
    }
    if ![Kafka][EventType] {
            drop{}
    }
    if [Kafka][Channel] == "" {
            drop {}
    }
    mutate {
            add_field => ["[@metadata][DateSuffix]", ""]
    }
    ruby {
            code => "event.set('[@metadata][DateSuffix]', DateTime.now.strftime('%Y-%m-%d')) "
    }
    ruby {                
            code => "event.set('message_size', event.get('[Kafka]').to_s.bytesize)"
    }

```

}

output {  
if ([Kafka][SubComponent]=="PPPPP" or [Kafka][SubComponent]=="QQQQ" or [Kafka][SubComponent]=="RRRR") {  
elasticsearch {  
codec =\> json  
hosts=\> "[http://xx.xxx.xxx.xx:9200](http://xx.xxx.xxx.xx:9200)"  
index=\>"test-logs-%{[@metadata][DateSuffix]}"  
document\_type =\> "logs"  
}  
}  
else {  
elasticsearch {  
codec =\> json  
hosts=\> "[http://xx.xx.xxx.xx:9200](http://xx.xx.xxx.xx:9200)"  
index=\>"rest-logs-%{[@metadata][DateSuffix]}"  
document\_type =\> "logs"  
}  
}  
stdout {  
codec =\> rubydebug  
}  
}

* * *

Thanks and Regards,

Nandita Rasam.

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [May 15, 2019, 5:59am UTC](https://discuss.elastic.co/t/create-multiple-indexes-with-single-logstash-configuration-file-depending-upon-the-kafka-channel/177003/6 "2019-05-15T05:59:14Z")

</div>

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.