Desired fields did get create after modifying logstash config

Hi All,

I have created two different elastic search clusters.

In first cluster (created newly) ,(v 7.13.2), I have created below config in pipeline.conf under conf.d for logstash and it worked as I see expected fields getting created in index. I verified this from Stack Management --> Index Management --> there is only one index (as data is not getting generated regularly) ---> after clicking on it, in Mappings tab.

input {
    beats {
        port => "5044"
    }
}
filter {
    grok {
        match => { "message" => "%{YEAR}-%{MONTHNUM}-%{MONTHDAY}[T ]%{HOUR}:%{MINUTE}(?::?%{SECOND})\| %{USERNAME:exchangeId}\| %{DATA:trackingId}\| %{NUMBER:RoundTrip:int}%{SPACE}ms\| %{NUMBER:ProxyRoundTrip:int}%{SPACE}ms\| %{NUMBER:UserInfoRoundTrip:int}%{SPACE}ms\| %{DATA:Resource}\| %{DATA:subject}\| %{DATA:authmech}\| %{DATA:scopes}\| %{IPV4:Client}\| %{WORD:method}\| %{DATA:Request_URI}\| %{INT:response_code}\| %{DATA:failedRuleType}\| %{DATA:failedRuleName}\| %{DATA:APP_Name}\| %{DATA:Resource_Name}\| %{DATA:Path_Prefix}"} 
    }
}
output {
    elasticsearch {
        hosts => [ "localhost:9200" ]
    }
}

The second cluster (7.4.0) which was already running, It had below config in

if [log_type] == "access_server" and [app_id] == "as"
  {
    mutate { gsub => ["message","\|"," "] } grok { patterns_dir => ["/etc/logstash/patterns"] match => { "message" => "%{MY_DATE_PATTERN:timestamp}%{SPACE}%{LOGLEVEL:level}%{SPACE}%{UUID:ConsentID}%{SPACE}%{WORD:TransactionID}%{SPACE}%{WORD:TraceID}%{SPACE}%{GREEDYDATA:messagetext}" } }
    mutate {
             replace => {
               "[type]" => "access_server"
             }
           }
  }

output {
if [log_type] == "access_server" {
  elasticsearch {
    hosts => ['http://14.1.1.50:9200']
    index => "%{type}-%{+YYYY.MM.dd}"
        user => elastic
    password => NfBxxxxx
      }
 }

elasticsearch {
    hosts => ['http://14.1.1.50:9200']
    index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
    user => elastic
    password => NfBxxxxx
  }
}

I modified the config file yesterday (06-July-21) to below to have individual fields as per desired data types, (as per the above first cluster config)

if [log_type] == "access_server" and [app_id] == "as"
  {
    mutate { gsub => ["message","\|"," "] } grok { patterns_dir => ["/etc/logstash/patterns"] match => { "message" => "%{YEAR}-%{MONTHNUM}-%{MONTHDAY}[T ]%{HOUR}:%{MINUTE}(?::?%{SECOND})\| %{USERNAME:exchangeId}\| %{DATA:trackingId}\| %{NUMBER:RoundTrip:int}%{SPACE}ms\| %{NUMBER:ProxyRoundTrip:int}%{SPACE}ms\| %{NUMBER:UserInfoRoundTrip:int}%{SPACE}ms\| %{DATA:Resource}\| %{DATA:subject}\| %{DATA:authmech}\| %{DATA:scopes}\| %{IPV4:Client}\| %{WORD:method}\| %{DATA:Request_URI}\| %{INT:response_code}\| %{DATA:failedRuleType}\| %{DATA:failedRuleName}\| %{DATA:APP_Name}\| %{DATA:Resource_Name}\| %{DATA:Path_Prefix}" } }
    mutate {
             replace => {
               "[type]" => "access_server"
             }
           }
  }

output {

  if [log_type] == "access_server" {
  elasticsearch {
    hosts => ['http://14.1.1.50:9200']
    index => "%{type}-%{+YYYY.MM.dd}"
        user => elastic
    password => NfBxxxxxx
      }
 }

elasticsearch {
    hosts => ['http://14.1.1.50:9200']
    index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
    user => elastic
    password => NfBxxxxxx
  }

and restarted logstash but today's access_server index is not having above individual fields.

The difference between above two clusters is, first cluster which I created to test above scenario (v 7.13.2), is having only one index (it indexed old log files and there are no new log files getting created daily), and I have created Index pattern of it in kibana but in second cluster which was already running is having daily indices of access_server but I have not created index patterns of it so I looked into today's access_server index but it is not having desired individual fields but I think the main issue is because of configration in pipeline.conf file which I mentioned above.

Can you please let me know what is not correct in modified pipeline.conf file because of which I still can't see individual message fields in today's index.

Thanks,

Hi All,

Can someone please reply.

I read your post and I have no idea what your question is.

Hello @Badger,

I wanted to parse below message line, (specifically the ' ms' fields)

2021-07-07T10:48:07,899| wV30LWuaCidXyUw| | 2 ms| 2 ms| 0 ms| dev.dev.next.only.io [] / /*:443| | | | 152.48.92.180| GET| /component---src-pages-products-aisp-js-7808dca6d4f3.js| 304| | | Dev PA| Root Resource| /* 

so I am trying to say, below logstash pipeline config parsed the data correctly. (this is on different cluster)

filter {
    grok {
        match => { "message" => "%{YEAR}-%{MONTHNUM}-%{MONTHDAY}[T ]%{HOUR}:%{MINUTE}(?::?%{SECOND})\| %{USERNAME:exchangeId}\| %{DATA:trackingId}\| %{NUMBER:RoundTrip:int}%{SPACE}ms\| %{NUMBER:ProxyRoundTrip:int}%{SPACE}ms\| %{NUMBER:UserInfoRoundTrip:int}%{SPACE}ms\| %{DATA:Resource}\| %{DATA:subject}\| %{DATA:authmech}\| %{DATA:scopes}\| %{IPV4:Client}\| %{WORD:method}\| %{DATA:Request_URI}\| %{INT:response_code}\| %{DATA:failedRuleType}\| %{DATA:failedRuleName}\| %{DATA:APP_Name}\| %{DATA:Resource_Name}\| %{DATA:Path_Prefix}"} 
    }
}

Below config was already present on another cluster (where I acutally wanted to parse data) but as you can see its not written correctly to parse above log.

{
    mutate { gsub => ["message","\|"," "] } grok { patterns_dir => ["/etc/logstash/patterns"] match => { "message" => "%{MY_DATE_PATTERN:timestamp}%{SPACE}%{LOGLEVEL:level}%{SPACE}%{UUID:ConsentID}%{SPACE}%{WORD:TransactionID}%{SPACE}%{WORD:TraceID}%{SPACE}%{GREEDYDATA:messagetext}" } }
    mutate {
             replace => {
               "[type]" => "access_server"
             }
           }
  }

so I copied first config (which was correctly parsing) and replaced it into above config. (not completely replaced as you can see mutate, gsub, patterns are still present as it is, I just replaced from "message" => ... ) because I didn't wanted to change too many things and restarted logstash.

modified config -

{
    mutate { gsub => ["message","\|"," "] } grok { patterns_dir => ["/etc/logstash/patterns"] match => { "message" => "%{YEAR}-%{MONTHNUM}-%{MONTHDAY}[T ]%{HOUR}:%{MINUTE}(?::?%{SECOND})\| %{USERNAME:exchangeId}\| %{DATA:trackingId}\| %{NUMBER:RoundTrip:int}%{SPACE}ms\| %{NUMBER:ProxyRoundTrip:int}%{SPACE}ms\| %{NUMBER:UserInfoRoundTrip:int}%{SPACE}ms\| %{DATA:Resource}\| %{DATA:subject}\| %{DATA:authmech}\| %{DATA:scopes}\| %{IPV4:Client}\| %{WORD:method}\| %{DATA:Request_URI}\| %{INT:response_code}\| %{DATA:failedRuleType}\| %{DATA:failedRuleName}\| %{DATA:APP_Name}\| %{DATA:Resource_Name}\| %{DATA:Path_Prefix}" } }
    mutate {
             replace => {
               "[type]" => "access_server"
             }
           }
  }

but index created on next day is not having the fields as we are trying to parse. In fact its having same fields which it had in earlier index so my question was is there anything wrong in the modified config?

I can see log file created on the next day which has such data coming but index is not having such fields we are trying to create through modified config.

Is there anything I need to remove like mutate, gsub or add something to make it work? as you can see its not available in the first successful config but I kept those thing as it is so that I will have to make minimal changes.

Thanks,

You are using mutate+gsub to delete all the | in the [message] field and then asking grok to match a pattern that contains the | characters. That is not going to work.

1 Like

Thanks @Badger for your help and explanation.

Ok. I will remove the mutate+gsub and patterns_dir part i.e will use the above first successful parsing.

so it will be like below.

if [log_type] == "access_server" and [app_id] == "as"
  {
    grok {
        match => { "message" => "%{YEAR}-%{MONTHNUM}-%{MONTHDAY}[T ]%{HOUR}:%{MINUTE}(?::?%{SECOND})\| %{USERNAME:exchangeId}\| %{DATA:trackingId}\| %{NUMBER:RoundTrip:int}%{SPACE}ms\| %{NUMBER:ProxyRoundTrip:int}%{SPACE}ms\| %{NUMBER:UserInfoRoundTrip:int}%{SPACE}ms\| %{DATA:Resource}\| %{DATA:subject}\| %{DATA:authmech}\| %{DATA:scopes}\| %{IPV4:Client}\| %{WORD:method}\| %{DATA:Request_URI}\| %{INT:response_code}\| %{DATA:failedRuleType}\| %{DATA:failedRuleName}\| %{DATA:APP_Name}\| %{DATA:Resource_Name}\| %{DATA:Path_Prefix}" } }
    mutate {
             replace => {
               "[type]" => "access_server"
             }
           }
  }

@Badger,

Thanks. I made above changes and can see today's index has those fields.