# Logstash configuration

**URL:** <https://discuss.elastic.co/t/logstash-configuration/295423>\
**Category:** Logstash\
**Created:** [January 26, 2022, 7:08am UTC](https://discuss.elastic.co/t/logstash-configuration/295423 "2022-01-26T07:08:13Z")\
**Posts on this page:** 14\
**Page:** 1

<div class="post-metadata">

**Author:** ![puneet\_makhija](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/puneet_makhija/32/100881_2.png) [@puneet\_makhija](https://discuss.elastic.co/u/puneet_makhija)\
**Post date:** [January 26, 2022, 7:08am UTC](https://discuss.elastic.co/t/logstash-configuration/295423/1 "2022-01-26T07:08:13Z")

</div>

Hello everyone,  
I am stuck in log stash configuration  
here below is my configuration please let me know what is wrong in this configuration

```auto
# Sample Logstash configuration for creating a simple
# Beats -> Logstash -> Elasticsearch pipeline.

input {
  kafka {
        bootstrap_servers => "http://localhost:9092"
        topics => ["tracking"]
        codec => avro {
            schema_uri => "/home/abc/python/working/divolte-collector-with-apache-kafka/divolte-collector-0.9.0/conf/MyEventRecord3.avsc"
        }
        value_deserializer_class => "org.apache.kafka.common.serialization.ByteArrayDeserializer"
    }
}

filter {
    if [eventType] == 'product_detail' {
        prune {
            blacklist_names => ["^user___.*"]
        }
    }

    if [eventType] == 'user_information' {
        prune {
            blacklist_names => ["^product___.*"]
        }
    }
}
output {
    stdout {
        codec => rubydebug
    }

    if [eventType] == 'product_detail' {
        kafka {
            bootstrap_servers => "http://localhost:9092"
            topic_id => 'product_detail'
        }
    }
    else if [eventType] == 'user_information' {
        kafka {
            bootstrap_servers => "http://localhost:9092"
            topic_id => 'user_information'
        }
    }
    else {
        kafka {
            bootstrap_servers => "http://localhost:9092"
            topic_id => 'unrecognized'
        }

    }
}

```

Below error is coming

```auto
[2022-01-26T12:26:56,585][INFO][org.reflections.Reflections] Reflections took 201 ms to scan 1 urls, producing 119 keys and 417 values
[2022-01-26T12:26:59,973][FATAL][logstash.runner] The given configuration is invalid. Reason: Unable to configure plugins: (SchemaParseError) Error validating default for eventType: at . expected type string, got null
[2022-01-26T12:26:59,991][FATAL][org.logstash.Logstash] Logstash stopped processing because of an error: (SystemExit) exit
org.jruby.exceptions.SystemExit: (SystemExit) exit
at org.jruby.RubyKernel.exit(org/jruby/RubyKernel.java:747) ~[jruby-complete-9.2.20.1.jar:?]
at org.jruby.RubyKernel.exit(org/jruby/RubyKernel.java:710) ~[jruby-complete-9.2.20.1.jar:?]
at usr.share.logstash.lib.bootstrap.environment.<main>(/usr/share/logstash/lib/bootstrap/environment.rb:94) ~[?:?]

```

Please provide some guidance

---

<div class="post-metadata">

**Author:** ![Tomo\_M](https://avatars.discourse-cdn.com/v4/letter/t/848f3c/32.png) [@Tomo\_M](https://discuss.elastic.co/u/Tomo_M)\
**Post date:** [January 26, 2022, 8:16am UTC](https://discuss.elastic.co/t/logstash-configuration/295423/2 "2022-01-26T08:16:50Z")

</div>

> [@puneet\_makhija](#):
>
> `SchemaParseError`

Isn't it an error on schema defined by `"/home/abc/python/working/divolte-collector-with-apache-kafka/divolte-collector-0.9.0/conf/MyEventRecord3.avsc"`?

You script around 'eventType' looks fine. I'm not familiar with kafka and avro, sorry.

---

<div class="post-metadata">

**Author:** ![puneet\_makhija](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/puneet_makhija/32/100881_2.png) [@puneet\_makhija](https://discuss.elastic.co/u/puneet_makhija)\
**Post date:** [January 26, 2022, 8:36am UTC](https://discuss.elastic.co/t/logstash-configuration/295423/3 "2022-01-26T08:36:05Z")

</div>

@Tom thanks as i remove defualt null from avro and its working perfectly now

as i have one question

`sudo bin/logstash --path.settings /etc/logstash/ -f /etc/logstash/logstash-genome.conf --config.reload.automatic`

I want to run this in background permanently what steps should i need to be taken on this, right now after executing this above command terminal process never shutdown until press Ctrl+c

---

<div class="post-metadata">

**Author:** ![Tomo\_M](https://avatars.discourse-cdn.com/v4/letter/t/848f3c/32.png) [@Tomo\_M](https://discuss.elastic.co/u/Tomo_M)\
**Post date:** [January 26, 2022, 8:46am UTC](https://discuss.elastic.co/t/logstash-configuration/295423/4 "2022-01-26T08:46:26Z")

</div>

You can run logstash as a service.

> **[Running Logstash as a Service on Debian or RPM | Logstash Reference \[7.16\] |...](https://www.elastic.co/guide/en/logstash/current/running-logstash.html)**

---

<div class="post-metadata">

**Author:** ![puneet\_makhija](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/puneet_makhija/32/100881_2.png) [@puneet\_makhija](https://discuss.elastic.co/u/puneet_makhija)\
**Post date:** [January 27, 2022, 11:03am UTC](https://discuss.elastic.co/t/logstash-configuration/295423/5 "2022-01-27T11:03:00Z")

</div>

@Tomo_M Thanks for your help , I am done with this above configuration  
and I want one more help actually I want to set some logic to change my key name  
here below is my logstash configuration

```auto
# Sample Logstash configuration for creating a simple
# Beats -> Logstash -> Elasticsearch pipeline.

input {
  kafka {
        bootstrap_servers => "http://localhost:9092"
        topics => ["tracking"]
        codec => avro {
            schema_uri => "/home/abc/python/working/divolte-collector-with-apache-kafka/divolte-collector-0.9.0/conf/MyEventRecord3.avsc"
        }
        value_deserializer_class => "org.apache.kafka.common.serialization.ByteArrayDeserializer"
    }
}

filter {
    if [eventType] == 'product_detail' {
        prune {
            blacklist_names => ["^user___.*"]
             ********************Question (mutate change in all fields because schema contains 62 fields)**************************
               # Then here I want remove "product___" from front in all the keys
               # product___bar = "10+"
            ******************************************************
        }
    }

    if [eventType] == 'user_information' {
        prune {
            blacklist_names => ["^product___.*"]
            *************Question (mutate change in all fields because schema contains 62 fields)**************************
             # Then here I want remove "user___" from front in all the keys 
             # user___id = "1"
            **********************************************
        }
    }
}
output {
    stdout {
        codec => rubydebug
    }

    if [eventType] == 'product_detail' {
        kafka {
            bootstrap_servers => "http://localhost:9092"
            topic_id => 'product_detail'
        }
    }
    else if [eventType] == 'user_information' {
        kafka {
            bootstrap_servers => "http://localhost:9092"
            topic_id => 'user_information'
        }
    }
    else {
        kafka {
            bootstrap_servers => "http://localhost:9092"
            topic_id => 'unrecognized'
        }

    }
}

```

Query is in this above configuration

---

<div class="post-metadata">

**Author:** ![Tomo\_M](https://avatars.discourse-cdn.com/v4/letter/t/848f3c/32.png) [@Tomo\_M](https://discuss.elastic.co/u/Tomo_M)\
**Post date:** [January 27, 2022, 11:14am UTC](https://discuss.elastic.co/t/logstash-configuration/295423/6 "2022-01-27T11:14:23Z")

</div>

If you can specify all field names, mutate rename filter should be a simple solution.

If not, it seems you have to use ruby filter of your own script.

---

<div class="post-metadata">

**Author:** ![Tomo\_M](https://avatars.discourse-cdn.com/v4/letter/t/848f3c/32.png) [@Tomo\_M](https://discuss.elastic.co/u/Tomo_M)\
**Post date:** [January 27, 2022, 11:15am UTC](https://discuss.elastic.co/t/logstash-configuration/295423/7 "2022-01-27T11:15:55Z")

</div>

You can get root hash by `event.to_hash`.

---

<div class="post-metadata">

**Author:** ![puneet\_makhija](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/puneet_makhija/32/100881_2.png) [@puneet\_makhija](https://discuss.elastic.co/u/puneet_makhija)\
**Post date:** [January 27, 2022, 12:08pm UTC](https://discuss.elastic.co/t/logstash-configuration/295423/8 "2022-01-27T12:08:35Z")

</div>

Hi @Tomo_M  
I don't know that much about ruby filter  
But I made it can you please verify

```auto
ruby {

        code => "

            event.to_hash.each {|k, v|

                event[k.split('product___').last] = event[v]

                event.remove(k)

            }

        "

    }

```

So I can paste this code just after prune object, after prune complete then this filter works?

---

<div class="post-metadata">

**Author:** ![puneet\_makhija](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/puneet_makhija/32/100881_2.png) [@puneet\_makhija](https://discuss.elastic.co/u/puneet_makhija)\
**Post date:** [January 27, 2022, 12:32pm UTC](https://discuss.elastic.co/t/logstash-configuration/295423/9 "2022-01-27T12:32:48Z")

</div>

getting tags =\> [[0] "\_rubyexception"]

---

<div class="post-metadata">

**Author:** ![Tomo\_M](https://avatars.discourse-cdn.com/v4/letter/t/848f3c/32.png) [@Tomo\_M](https://discuss.elastic.co/u/Tomo_M)\
**Post date:** [January 27, 2022, 12:40pm UTC](https://discuss.elastic.co/t/logstash-configuration/295423/10 "2022-01-27T12:40:50Z")

</div>

The 4th line should be  
`event.set(k.split('product___').last, v)`

---

<div class="post-metadata">

**Author:** ![puneet\_makhija](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/puneet_makhija/32/100881_2.png) [@puneet\_makhija](https://discuss.elastic.co/u/puneet_makhija)\
**Post date:** [January 27, 2022, 1:02pm UTC](https://discuss.elastic.co/t/logstash-configuration/295423/11 "2022-01-27T13:02:47Z")

</div>

😶  
@Tomo_M through above line its working perfectly but all 61 fields gone due to event.remove(k) inside the for loop.

```auto
z = {"purchase ___product_id": "22", "purchase___ tax": "1", "normal": 1}
for i in list(z):
    try:
        z[i.split('___')[1]] = z[i]
    except IndexError:
        continue
    del z[i]
print(z)
#{'normal': 1, 'product_id': '22', 'tax': '1'}

```

want this exception handler in ruby filter

---

<div class="post-metadata">

**Author:** ![Tomo\_M](https://avatars.discourse-cdn.com/v4/letter/t/848f3c/32.png) [@Tomo\_M](https://discuss.elastic.co/u/Tomo_M)\
**Post date:** [January 27, 2022, 2:48pm UTC](https://discuss.elastic.co/t/logstash-configuration/295423/12 "2022-01-27T14:48:01Z")

</div>

Really?

```auto
filter {
  ruby {
    code => "prefix = 'product___'
    event.to_hash.each{|k,v|
      if (!k.start_with?('@')) && (k.include?(prefix))
        event.set(k.split(prefix).last, v)
        event.remove(k)
      end
    }"
  }
}

```

worked for me.

---

<div class="post-metadata">

**Author:** ![puneet\_makhija](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/puneet_makhija/32/100881_2.png) [@puneet\_makhija](https://discuss.elastic.co/u/puneet_makhija)\
**Post date:** [January 27, 2022, 4:26pm UTC](https://discuss.elastic.co/t/logstash-configuration/295423/13 "2022-01-27T16:26:22Z")

</div>

@Tomo_M Thank you so much its working perfect now  
Actually I am not aware about how to use include in this "(k.include?(prefix))" and the if syntax.  
For practice sake is there any online playground available for this. to see and compile the out  
Can you please provide some links

---

<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:** [February 24, 2022, 4:27pm UTC](https://discuss.elastic.co/t/logstash-configuration/295423/14 "2022-02-24T16:27:18Z")

</div>

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