Need to configure message timestamp as timestamp in elastic stack

[Mobin]

We have created an elastic stack to monitor logs from the Linux clients. On Linux clients, filebeat is configured to ship the logs to the elastic stack.
When we are searching for the logs using elastic discover option @timestamp and message timestamp is not matching. How can we configure timestamp as same as message timestamp?

eg: fields ( time, @timestamp, beat.hostname, ,message, source)

November 1st 2021, 08:41:20.986 November 1st 2021, 08:41:20.986 scnmgmt3 "Nov 1 06:00:03 scnmgmt3 sshd[84934]: Accepted password for root from 172.22.76.114 port 43778 ssh2"
/var/log/secure

logstash input:-

input {
  beats {
	port => 5044
	ssl => false
  }
}

logstash filter:-

filter 
{
if [type] == "syslog" {
	grok {
  	match => { "message" => "%{SYSLOGLINE}" }
	}

	date {
match => [ "timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ]
}
  }
}

logstash output:-

output {
  elasticsearch {
	hosts => ["localhost:9200"]
	sniffing => true
	manage_template => false
	index => "%{[@metadata][beat]}-%{+YYYY.MM.dd}"
	document_type => "%{[@metadata][type]}"
  }
}

filebeat config:-

filebeat.prospectors:
- input_type: log

   paths:
    - /var/log/messages
    - /var/log/secure
output.logstash:
  hosts: ["172.x.x.x:5044"]

[Mustafa_NAJIB (Mustafa)]

Hello :

this is your input :

November 1st 2021, 08:41:20.986 scnmgmt3 "Nov 1 06:00:03 scnmgmt3 sshd[84934]: Accepted password for root from 172.22.76.114 port 43778 ssh2"/var/log/secure

this is the new filter :

grok {     
    match          => {
        "message"    => [
          "^%{GREEDYDATA:time} %{DATA:beat_hostname} \"%{GREEDYDATA:TIMESTAMP} %{DATA:beat_hostname} %{GREEDYDATA}: %{GREEDYDATA:log_message}\"%{GREEDYDATA:source}",
          "%{GREEDYDATA:FAILPARSE}"
        ]
    }
    overwrite => [ "beat_hostname" ]
  }

OUTPUT :

{
  "TIMESTAMP": "Nov 1 06:00:03",
  "beat_hostname": "scnmgmt3",
  "log_message": "Accepted password for root from 172.22.76.114 port 43778 ssh2",
  "message": "November 1st 2021, 08:41:20.986 scnmgmt3 \"Nov 1 06:00:03 scnmgmt3 sshd[84934]: Accepted password for root from 172.22.76.114 port 43778 ssh2\"/var/log/secure",
  "source": "/var/log/secure",
  "time": "November 1st 2021, 08:41:20.986"
}

[Mobin]

Thanks for the reply.

I have changed the filter

filter 
{
if [type] == "syslog" {
grok {     
    match          => {
        "message"    => [
          "^%{GREEDYDATA:time} %{DATA:beat_hostname} \"%{GREEDYDATA:TIMESTAMP} %{DATA:beat_hostname} %{GREEDYDATA}: %{GREEDYDATA:log_message}\"%{GREEDYDATA:source}",
 }
}
'''
          "%{GREEDYDATA:FAILPARSE}"
        ]
    }
    overwrite => [ "beat_hostname" ]
  }

Still I am not able to search time from message filed. In the kibana there is no available fileds for @time.

[Rios]

Avoid to use GREEDYDATA,it's expensive.

Make a custom pattern:
%{MONTH} %{MONTHDAY}.{0,2} %{YEAR}, %{TIME}\.%{NONNEGINT}

Then you have to convert to date "November 1st 2021 08:41:20.986" with

date {
match => [ "timestamp", "MMM d yyyy HH:mm:ss.SSS", "MMM d yyyy, HH:mm:ss.SSS"]

At the end, only correct format the time field must be like: "time": "2021-11-01T08:41:20.986Z"

[Mobin]

My requirement is to grep the events based on time. I am already able to do that but the problem is the timestamp in the message filed is not matching with filebeat timestamp. So I need to add a new filed as same as message timestamp and the same needs to be added to filebeat fields.

[Rios]

@timestamp - time from Logstash, default datetime format
timestamp - time from Filebeat, default string format

If you need to @timestamp became as timestamp, just copy. But the "timestamp" field must be in date format converted not string.

date {
match => [ "timestamp", "MMM d yyyy HH:mm:ss.SSS", "MMM d yyyy, HH:mm:ss.SSS"]
}

Then copy after conversion

      mutate {
       copy => { "timestamp" => "@timestamp"}
      }

[Mobin]

Dear Rios,

Sorry. Since I am new to elk, I didn't understand your solution.

We are using file beat to ship the logs from the client and elk server logstash input created to receive the logs from filebeat. I didn't understand why we need to change the timestamp here.

In our elk server there is a timestamp ( I think that is based on filebeat sync time ) and one more is there on the filebeat message field ( It is a single field wiith time stamp and message ). Now we need to seperate the timestamp from the message field So we can query based on the exact time of the event/message.