Initialize counter in logstash and replace the '%{[message][0]}'?

Elastic Stack Logstash
1

Hi,
after doing the split filter, so my message was an array ok, so %{[message][0]} %{[message][1]} ... that gives me their value fine. now I have an array of length 30 for example and I start the condition if {}
example logstash.conf:

filter {
mutate { split => {"message" => "|"} }
mutate {
      add_field => { "DateTime" => "%{[message][0]}" }
      add_field => { "version" => "%{[message][1]}" } ............
}
if [a]  == "Offline" {
              mutate { add_field => {"example1" => "%{[message][29]}"} }
            }
}

I mean, can I create a counter for example to replace %{[message][29]} with x+1 and I declare int x=0 or something like that because it's a bad idea to keep doing it manually %{[message][0]} %{[message][1]} %{[message][29]} %{[message][40]} %{[message][50]} ...... :confused:

Thanks!

2

i want to do a loop from %{[message][0]} to number_of_elements i did this

filter {
 mutate { split => {"message" => "|"} }
ruby {
      code => "event.set('number_of_elements', event.get('message').length)"
      }
ruby {
        code => '
            for i in number_of_elements
                k = "message#{i}"
                s = event.get(" ... ??????????")
                if s
                    event.set(k, s)
                end
            end
        '
    }
}

I know that my configuration was so bad but i just want to add field just make it somthing like this

mutate {
      add_field => { "DateTime" => "%{[message][x+1]}" }
      add_field => { "version" => "%{[message][x+1]}" }
}

and make x= x+1 in the loop i mean when a field added x++ hope that is possible :confused:

3

Can you give an example of your message? It is not clear how your message look and what you want to do.

From where are you going to get the field names?

4

Thank you for the reply.
yes here an example of my message

2312-15:45:43:904|V2.5.4|PAI   |Gojf40555saa|Comp|Ofline|M|076|561-30687|34aa|580796|Total: 123|
2312-15:45:44:904|V2.5.4|DOW   |Gsanqajkd0555saa|Commmp|Offffline|S|044|561-30687|34aa|580796|Comp|Ofline|M|076|561-30687||iapp|kkslllssTotal: 123|
...................

I just want to add fied for each | so i did the split filter so

mutate {
      add_field => { "DateTime" => "%{[message][0]}" }
      add_field => { "version" => "%{[message][1]}" }
      add_field => { "Operation" => "%{[message][2]}" }
......
    }

I have files that contain these logs so admin knows every field name so he will give me all possible cases so in this situation I have so many if{} statement for example in one log sometimes contains 40 pipeline "|" sometimes 50 ... so it's a bad idea to keep doing it manually %{[message][0]} %{[message][1]} %{[message][29]} %{[message][40]} %{[message][50]} ......
So i'm asking if there is any solution for this for example create a counter x can replace %{[message][0]} to %{[message][x]} %{[message][x+1]} %{[message][x+1]} and it should increment x i mean x=x+1.
I mean if it can be like this

mutate {
      add_field => { "DateTime" => "%{[message][x]}" }
      add_field => { "version" => "%{[message][x+1]}" }
      add_field => { "Operation" => "%{[message][x+1]}" }
}
      if ([version] == V1) {
           mutate {
                add_field => { "example" => "%{[message][x+1]}" }
                           }
      }
mutate {
      add_field => { "IdMerchant" => "%{[message][x+1]}" }
    }

Note: I'm using filebeat to read logs and then i send the data to kafka and logstash consume this data --> Elasticsearch --> kibana.
Thanks

5

You probably can do that with a ruby code and the ruby filter, but it is still not clear from where are you going to get the field names.

For example:

add_field => { "DateTime" => "%{[message][x]}" }

The field named DateTime is hardcoded, the same for the others fields.

The example you gave you are dynamically setting the value, but not the field name, how will you get the correct field name?

Also, does your log messages have any order? For example, if you have a message with 5 fields, and another one with 10 fields, are the fields always in the same order, like this:

field1|field2|field3|field4|field5
field1|field2|field3|field4|field5|field6|field7|field8|field9|field10

If you have something like that, or can change your source file to be something like that, you can easily parse your messages using the csv filter, you would just need to set all the possible columns in the order they appear and the filter will populate them when they exist.

6

These are transaction logs so the developer who made the application "Gateway" writes the logs in file with the variables that made so here are the fields names comes from. So the developer who made the application who gave me the fields names.
my log messages are not in the same order :confused:
I already tried the ruby filter but didn't know how to do it, I get the message size with

ruby {
      code => "event.set('number_of_elements', event.get('message').length)"
      }

but i didn't know how to initialize a variable and make it ++ i mean for example x=x+1 in a loop or ... something like that maybe can you please correct me or if there to something else

ruby {
        code => '
            for i in number_of_elements
                k = "message#{i}"
                s = event.get(" ... ??????????")
                if s
                    event.set(k, s)
                end
            end
        '
    }

Thanks.

7

I do not know much of ruby, but to get a field from the event to the ruby code you use something like this:

fieldName_in_ruby_code = event.get("[fieldName_in_logstash]")

Then I think you could iterate in that array using each_with_index.

fieldName_in_ruby_code.each_with_index do |val, index|
    ruby code
end

But again, where will the field names come from? How will you know that the value of message[2] in one message is different from the value of message[2] and should be in another field?

Your examples only shows iterations over the value, not the field name.

8

If there are a limited number of formats you may be able to count the number of columns and use that to determine which csv filter to use.

9

Thank you @leandrojmp @Badger
yes I think I'll do that @Badger last thing take a look at that maybe it works?

filter {
mutate { split => {"message" => "|"} }
      ruby {
      code => "event.set('number_of_elements', event.get('message').length)"
      }
      ruby {
      code => "event.set('x', 0)"
      }

      ruby {
        code => '
            for i in number_of_elements
                x = x + 1
                i = i + 1
            end
        '
    }
 mutate {
      add_field => { "test_x" => "%{x}" }
      add_field => { "DateTime" => "%{[message][%{x}]}" }
      add_field => { "version" => "%{[message][x+1]}" }
      add_field => { "GateWay Operation" => "%{[message][2]}" }
    }
}

It gives me

"number_of_elements" => 51,
"version" => "%{[message][x+1]}",
"GateWay Operation" => "PAI   ",
"test_x" => "0",
"Idtransaction" => "Gojfaaaajkl,lksq=",
"x" => 0,

but didn't get the field DateTime Exception caught while applying mutate filter {:exception=>"Invalid FieldReference: [message][%{x"}
how can i enter the value of x under %{[message][%{x}]} if you have any idea please ?
Thanks!!

10

It won't work this way, you need to do everything inside one ruby filter and you can't use the variables from your ruby code outside it, in the mutate filter.

The loop will only exist in inside the ruby code, so you will need to use event.set if you want to add fields.

But as I said, your field names are all static, it makes no sense to have a loop on the values if you are you going to statically set the field names, you need to provide more context about the name of the fields.

If you splitted your message into an array, you can access it using the index in any mutate filter.

1 Like
11

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