# Making decisions on the basis of message keys

**URL:** <https://discuss.elastic.co/t/making-decisions-on-the-basis-of-message-keys/156784>\
**Category:** Logstash\
**Created:** [November 15, 2018, 5:29am UTC](https://discuss.elastic.co/t/making-decisions-on-the-basis-of-message-keys/156784 "2018-11-15T05:29:06Z")\
**Posts on this page:** 4\
**Page:** 1

<div class="post-metadata">

**Author:** ![Faraz\_Mateen](https://avatars.discourse-cdn.com/v4/letter/f/ce7236/32.png) [@Faraz\_Mateen](https://discuss.elastic.co/u/Faraz_Mateen)\
**Post date:** [November 15, 2018, 5:29am UTC](https://discuss.elastic.co/t/making-decisions-on-the-basis-of-message-keys/156784/1 "2018-11-15T05:29:06Z")

</div>

I am reading messages from a kafka topic and pushing them to logstash. It's pretty straight forward for a simple json but now I want to update my packet structure. Update messages have the following structure:

```auto

{

  "indexname": string,

  "clientname": string,

  "data": {key1: value1, key2: value2, ..... , key3: value3, "datetime":"2017-09-08 11:11:11"}

}

```

Now, I want to choose my index on the basis of indexname key but I only want push the json under "data" key to elasticsearch. Also, there is a datetime field inside data json that I am using in the index name.

Currently with **non-nested** json, my config looks like this:

```auto

input {

  kafka {

    bootstrap_servers => "localhost:9092"

    type => "redis-input"

    topics => ["test-topic"]

    codec => "json"

    consumer_threads => 5

    group_id => "group_id"

    auto_offset_reset => "latest"

  }

}

filter {

   date {

      match => [[data][datetime], "UNIX", "UNIX_MS", "dd/MM/yyyy:HH:mm:ss Z", "yyyy-MM-dd HH:mm:ss" ]

   }

}

output {

   elasticsearch {

      hosts => ["127.0.0.1"]

      index => "clientindex-%{+yyyy}-%{+MM}"

   }

}

```

Now I want to replace **"clientindex"** from the output block to whatever **indexname** I get in the message and then I want to forward the contents on "data" key only.

Can anyone advise the filter that can help me achieve this.

Any help will be appreciated.

--

Faraz Mateen

---

<div class="post-metadata">

**Author:** ![Christian\_Dahlqvist](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/christian_dahlqvist/32/4617_2.png) [@Christian\_Dahlqvist](https://discuss.elastic.co/u/Christian_Dahlqvist)\
**Post date:** [November 15, 2018, 8:01am UTC](https://discuss.elastic.co/t/making-decisions-on-the-basis-of-message-keys/156784/2 "2018-11-15T08:01:31Z")

</div>

If you know this field is always set and contains only valid field name prefixes (you may want to add logic to check this) you can create an output block like this:

```auto
output {
   elasticsearch {
      hosts => ["127.0.0.1"]
      index => "%{[indexname]}-%{+yyyy}-%{+MM}"
   }
}

```

---

<div class="post-metadata">

**Author:** ![Faraz\_Mateen](https://avatars.discourse-cdn.com/v4/letter/f/ce7236/32.png) [@Faraz\_Mateen](https://discuss.elastic.co/u/Faraz_Mateen)\
**Post date:** [November 15, 2018, 10:28am UTC](https://discuss.elastic.co/t/making-decisions-on-the-basis-of-message-keys/156784/3 "2018-11-15T10:28:59Z")

</div>

@Christian_Dahlqvist  
Thanks. That works for index naming

how can I extract only one key from the message and send it to the output? For example, from the sample packet above, I only want to send contents under **"data"** key to elasticsearch.

---

<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:** [December 13, 2018, 10:29am UTC](https://discuss.elastic.co/t/making-decisions-on-the-basis-of-message-keys/156784/4 "2018-12-13T10:29:02Z")

</div>

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