My Timestamp in audit log using SYSLOG input plugin not in correct format

Elastic Stack Logstash
1

Configuration is below :

input {
    syslog {
        port => 5514
		type => "scylladb"
    }
}
filter {
if [type] == "scylladb" {
	
	grok{
		   match => {"message" => "%{IP:server_ip}:%{DATA:server_port},\s%{DATA:category},\s%{DATA:consistency},\s%{DATA:table_name},\s%{DATA:keyspace_name},\s%{GREEDYDATA:operation},\s%{DATA:source}:%{DATA:source_port},\s%{DATA:username},\s%{GREEDYDATA:error}"
		   }
	}
	if "_grokparsefailure"  in [tags]
	{
		drop{}
	}
}
}
output {
    stdout { codec => rubydebug }
}

My syslog configuration:

# /etc/rsyslog.conf configuration file for rsyslog
#
# For more information install rsyslog-doc and see
# /usr/share/doc/rsyslog-doc/html/configuration/index.html
#
# Default logging rules can be found in /etc/rsyslog.d/50-default.conf


#################
#### MODULES ####
#################

module(load="imuxsock") # provides support for local system logging
#module(load="immark")  # provides --MARK-- message capability

# provides UDP syslog reception
#module(load="imudp")
#input(type="imudp" port="514")

# provides TCP syslog reception
#module(load="imtcp")
#input(type="imtcp" port="514")

# provides kernel logging support and enable non-kernel klog messages
module(load="imklog" permitnonkernelfacility="on")

###########################
#### GLOBAL DIRECTIVES ####
###########################

#
# Use traditional timestamp format.
# To enable high precision timestamps, comment out the following line.
#
$ActionFileDefaultTemplate RSYSLOG_TraditionalFileFormat

# Filter duplicated messages
$RepeatedMsgReduction on

#
# Set the default permissions for all log files.
#
$FileOwner syslog
$FileGroup adm
$FileCreateMode 0640
$DirCreateMode 0755
$Umask 0022
$PrivDropToUser syslog
$PrivDropToGroup syslog

#
# Where to place spool and state files
#
$WorkDirectory /var/spool/rsyslog

#
# Include all config files in /etc/rsyslog.d/
#
$IncludeConfig /etc/rsyslog.d/*.conf

#Send logs to client server
$PreserveFQDN on
*.* @<Ip of client where logstash is running>:5514
$ActionQueueFileName queue
$ActionQueueMaxDiskSpace 1g
$ActionQueueSaveOnShutdown on
$ActionQueueType LinkedList
$ActionResumeRetryCount -1

My time stamp in audit log is Jan 11 08:02:20 year is not present.

{
        "@timestamp" => 2024-01-11T08:02:20.000Z,
          "facility" => 1,
          "severity_label" => "Notice",
          "priority" => 13,
         "timestamp" => "Jan 11 08:02:20",
              "host" => "172.35.63.35",
    "facility_label" => "user-level",
         "logsource" => "ip-172-35-63-35.ec2.internal",
          "@version" => "1",
           "message" => "\"172.35.63.35:0\", \"DDL\", \"ONE\", \"newyeartest2\", \"mykeyspace\", \"CREATE TABLE newyeartest2 (a int,b int,c int,PRIMARY KEY (a, b, c))\", \"152.59.69.211:0\", \"cassandra\", \"false\"",
          "severity" => 5,
              "type" => "db"
}
2

HI,

you can use the date filter in Logstash to parse the timestamp and add the current year to it. Here is an example of how you can do this:

filter {
  if [type] == "scylladb" {
    grok {
      match => {"message" => "%{IP:server_ip}:%{DATA:server_port},\s%{DATA:category},\s%{DATA:consistency},\s%{DATA:table_name},\s%{DATA:keyspace_name},\s%{GREEDYDATA:operation},\s%{DATA:source}:%{DATA:source_port},\s%{DATA:username},\s%{GREEDYDATA:error}"}
    }
    if "_grokparsefailure"  in [tags] {
      drop{}
    }
    date {
      match => [ "timestamp", "MMM dd HH:mm:ss" ]
      target => "@timestamp"
    }
  }
}

In this configuration, the date filter is used to parse the timestamp field (which should be in the format "MMM dd HH:mm:ss") and store the result in the @timestamp field. The date filter will automatically add the current year to the timestamp.

Regards

3

Except when it doesn't. See here, here, and here. It's non-trivial.

4

Still the same

{
         "timestamp" => "Jan 17 07:55:43",
              "host" => "172.31.63.35",
          "category" => "\"AUTH\"",
          "severity" => 5,
       "consistency" => "\"\"",
           "message" => "\"172.31.63.35:0\", \"AUTH\", \"\", \"\", \"\", \"\", \"127.0.0.1:0\", \"cassandra\", \"false\"",
    "severity_label" => "Notice",
          "username" => "\"cassandra\"",
         "logsource" => "ip-172-31-63-35.ec2.internal",
        "table_name" => "\"\"",
         "operation" => "\"\"",
       "server_port" => "0\"",
           "program" => "scylla-audit",
          "facility" => 1,
    "facility_label" => "user-level",
             "error" => "\"false\"",
              "type" => "scylladb",
        "@timestamp" => 2024-01-17T07:55:43.000Z,
          "priority" => 13,
          "@version" => "1",
         "server_ip" => "172.31.63.35",
     "keyspace_name" => "\"\"",
            "source" => "\"127.0.0.1",
       "source_port" => "0\""
}
5

Looks good to me. The date filter has added the current year. What is your issue with this result?

6

It was already there no change please check my previous log

7

It is not clear what is your issue, what you are trying to do and what is not working.

The timestamp field comes from your log, it comes already without the year, Logstash will not change it.

8

I want to fetch Year from @timestamp field and want to add it to timestamp field. Is it possible?

Or any other way to add the current year to timestamp field

9 
ruby {
		code => "
		  event.set('current_year', Time.now.utc.year)
		"
	}
	mutate {
		convert => { "current_year" => "string" }
	}
	ruby {
		code => "
		  event.set('demotimestamp', event.get('current_year') + ' ' + event.get('timestamp'))
		"
	}
	mutate
	{
		remove_field => [ "timestamp","current_year"]
	}	
	mutate
	{
		rename => { "demotimestamp" => "timestamp" }
	}

I added this and now I am able to fetch current UTC year using Time.now.utc.year and add it to my timestamp field.

Result: "timestamp" => "2024 Jan 19 04:24:00"

10

I am not speaking to your use case, but in general adding the current year is not what you want. The date filter has heuristics to guess what year is applicable, but there will always be exceptions where that goes wrong.

11

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