INS-26353 [Offset fix for MariaDB OnPrem]

Signed-off-by: Apurva-Birajdar <[email protected]>

filter-plugin/logstash-filter-mariadb-guardium/MariaDB.conf

@@ -1,6 +1,6 @@
-input 
+input
 {
-  beats 
+  beats
   {
   port => 8541
   type => "mariadb"
@@ -11,17 +11,15 @@ if [type] == "mariadb" and "MariaDB_On_Premise" in [tags][0]{
 mutate {
 add_field => {"server_Ip" => "%{[host][ip][0]}"}
 }
-mutate {add_field => { "TZ" => "%{[event][timezone]}" }}
+mutate {add_field => { "Offset" => "%{[event][timezone]}" }}
 
 # In case of duplicate records enable add_id feature in Filebeat configuration and uncomment below mentioned lines replacing event_id and _id.
 
 #mutate {
 # replace => { "event_id" => "%{[@metadata][_id]}"}
 # replace => { "_id" => "%{[@metadata][_id]}"}
 #}
-grok { match => { "TZ" => "(?<minutes>[^:]*):(?<seconds>[^,]*)" } }
-grok { match => { "minutes" => "(?<offset_diff>[^[0-9]]*)%{GREEDYDATA:actual_minutes}" } }ruby { code => "event.set('minutes1', event.get('actual_minutes').to_i * 60)" }
-ruby { code => "event.set('offset1', event.get('minutes1') +  event.get('seconds').to_i)" }mutate { add_field => { "totalOffset" => "%{offset_diff}%{offset1}" } }
+
 grok
 {
 match => {"message" => "(?<timestamp>[^[A-Z][a-z]]*),(?<serverhost>[^\s]*),(?<username>[^\s]*),(?<hostname>[^\s]*),(?<connectionid>[^\s]*),(?<queryid>[^\s]*),(?<operation>[^\s]*),(?<database>[^\s]*),(?:%{GREEDYDATA:object}),(?<retcode>[^\s]*)"}

filter-plugin/logstash-filter-mariadb-guardium/MariaDBOverFilebeatPackage/MariaDB/filter.conf

@@ -1,30 +1,31 @@
 filter{
-if [type] == "filebeat" and "guc_filter_param_datasource_tag" in [tags] {
+if [type] == "mariadb" and "guc_filter_param_datasource_tag" in [tags]{
 mutate {
 add_field => {"server_Ip" => "%{[host][ip][0]}"}
 }
-mutate {add_field => { "TZ" => "%{[event][timezone]}" }}
- 
+mutate {add_field => { "Offset" => "%{[event][timezone]}" }}
+
 # In case of duplicate records enable add_id feature in Filebeat configuration and uncomment below mentioned lines replacing event_id and _id.
- 
+
 #mutate {
 # replace => { "event_id" => "%{[@metadata][_id]}"}
 # replace => { "_id" => "%{[@metadata][_id]}"}
 #}
 
-grok { match => { "TZ" => "(?<minutes>[^:]*):(?<seconds>[^,]*)" } }
-grok { match => { "minutes" => "(?<offset_diff>[^[0-9]]*)%{GREEDYDATA:actual_minutes}" } }ruby { code => "event.set('minutes1', event.get('actual_minutes').to_i * 60)" }
-ruby { code => "event.set('offset1', event.get('minutes1') +  event.get('seconds').to_i)" }mutate { add_field => { "totalOffset" => "%{offset_diff}%{offset1}" } }
 grok
 {
 match => {"message" => "(?<timestamp>[^[A-Z][a-z]]*),(?<serverhost>[^\s]*),(?<username>[^\s]*),(?<hostname>[^\s]*),(?<connectionid>[^\s]*),(?<queryid>[^\s]*),(?<operation>[^\s]*),(?<database>[^\s]*),(?:%{GREEDYDATA:object}),(?<retcode>[^\s]*)"}
 }
+# This is to ignore system logs which are having correct format but database name is missing.
+# create database command is exception thats why excluded from this condition.
 if "QUERY" == [operation] and "0" == [retcode] and ![database] and "create database" not in [message]{
 drop{}
 }
+# The "FAILED_CONNECT" system generated logs is appearing multiple time, which doesn't have any relation with authentication log.
+# so ignoring it.
 if "FAILED_CONNECT" == [operation] and "0" == [queryid] and ![username]{
 drop{}
 }
 mariadb_guardium_filter{}
 }
-}
+}

filter-plugin/logstash-filter-mariadb-guardium/mariadb-test.conf

@@ -1,10 +1,12 @@
 input {
   generator {
-  message =>"20220203 08:13:03,ip-172-31-44-208.ap-south-1.compute.internal,root,localhost,60,477,QUERY,testDB,'select * from Orders',0" 
-  count => 1 
+  message =>"20220203 08:13:03,ip-172-31-44-208.ap-south-1.compute.internal,root,localhost,60,477,QUERY,testDB,'select * from Orders',0"
+  count => 1
+  add_field => {"timezone" => "-07:00"}
+  }
   }
-  }  
 filter{
+mutate {add_field => { "Offset" => "%{[timezone]}" }}
 grok
 {
 match => {"message" => "(?<timestamp>[^[A-Z][a-z]]*),(?<serverhost>[^\s]*),(?<username>[^\s]*),(?<hostname>[^\s]*),(?<connectionid>[^\s]*),(?<queryid>[^\s]*),(?<operation>[^\s]*),(?<database>[^\s]*),(?:%{GREEDYDATA:object}),(?<retcode>[^\s]*)"}

filter-plugin/logstash-filter-mariadb-guardium/src/main/java/com/ibm/guardium/mariadb/ParserHelper.java

@@ -6,6 +6,7 @@
 package com.ibm.guardium.mariadb;
 
 import java.time.LocalDateTime;
+import java.time.ZoneOffset;
 import java.time.ZonedDateTime;
 import java.time.format.DateTimeFormatter;
 import java.time.format.DateTimeFormatterBuilder;
@@ -225,20 +226,23 @@ public static SessionLocator parseSessionLocator(final Event event) {
 	 * @return Time
 	 */
 	public static Time getTime(final Event event) {
-		String dateString = null;
-		dateString = event.getField(ApplicationConstant.TIMESTAMP_KEY).toString();
-		LocalDateTime localDateTime = LocalDateTime.parse(dateString, DATE_TIME_FORMATTER);
-		localDateTime.format(DateTimeFormatter.ISO_DATE_TIME);
+		String dateString = event.getField(ApplicationConstant.TIMESTAMP_KEY).toString();
+		String timeZone = event.getField(ApplicationConstant.MIN_OFFSET).toString();
+		if(dateString!=null) {
+			LocalDateTime dt = LocalDateTime.parse(dateString, DATE_TIME_FORMATTER);
+			ZoneOffset offset = ZoneOffset.of(ZoneOffset.UTC.getId());
+			if(timeZone != null ){
+				offset = ZoneOffset.of(timeZone);
+			}
+			ZonedDateTime zdt = dt.atOffset(offset).toZonedDateTime();
+			long millis = zdt.toInstant().toEpochMilli();
+			int minOffset = zdt.getOffset().getTotalSeconds() / 60;
+			return new Time(millis, minOffset, 0);
+		}
+		return null;
+	}
 
-		ZonedDateTime date = ZonedDateTime.parse(localDateTime.toString().concat("Z"), DateTimeFormatter.ISO_DATE_TIME);
-		long millis = date.toInstant().toEpochMilli();
-		int minOffset = date.getOffset().getTotalSeconds() / 60;
-		Time time = new Time(millis, minOffset, 0);
-		long t = (time.getTimstamp()) - (Integer.valueOf(event.getField("totalOffset").toString()) * 60000);
-		time.setTimstamp(t);
 
-		return time;
-	}
 	/**
 	 * parseSQL() method will perform operation on String inputs, set the expected
 	 * value into respective dataset name and then return the value as response

filter-plugin/logstash-filter-mariadb-guardium/src/main/java/com/ibm/guardium/mariadb/constant/ApplicationConstant.java

@@ -12,6 +12,7 @@ public class ApplicationConstant {
 	public static final String USERNAME_KEY = "username";
 	public static final String HOSTNAME_KEY = "hostname";
 	public static final String CONNECTIONID_KEY = "connectionid";
+
 	public static final String QUERYID_KEY = "queryid";
 	public static final String OPERATION_KEY = "operation";
 	public static final String DATABASE_KEY = "database";
@@ -24,7 +25,8 @@ public class ApplicationConstant {
 	public static final String SOURCEPROGRAM_KEY = "source_program";
 	public static final String SOURCEPROGRAM_VALUE = "mariadb";
 	public static final String DBPROTOCAL_STRING = "MariaDB";
-	public static final String SERVICE_NAME_STRING = "mariadb";
+
+	public static final String MIN_OFFSET = "Offset";
 	public static final String SERVER_IP_STRING = "0.0.0.0";
 	public static final String CLIENT_IP_STRING = "0.0.0.0";
 	public static final String UNKNOWN_STRING = "";