Logstash configuration

puneet_makhija · January 26, 2022, 7:08am

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

# 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

[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

Tomo_M · January 26, 2022, 8:16am

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.

puneet_makhija · January 26, 2022, 8:36am

@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

Tomo_M · January 26, 2022, 8:46am

You can run logstash as a service.

puneet_makhija · January 27, 2022, 11:03am

@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

# 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

Tomo_M · January 27, 2022, 11:14am

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.

Tomo_M · January 27, 2022, 11:15am

You can get root hash by event.to_hash.

puneet_makhija · January 27, 2022, 12:08pm

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

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?

puneet_makhija · January 27, 2022, 12:32pm

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

Tomo_M · January 27, 2022, 12:40pm

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

puneet_makhija · January 27, 2022, 1:02pm

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

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

Tomo_M · January 27, 2022, 2:48pm

Really?

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.

puneet_makhija · January 27, 2022, 4:26pm

@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

system · February 24, 2022, 4:27pm

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