(hopefully) fix idaholab#151, last few seconds' Zeek logs prior to lo…

…g rotation may be lost
mmguero-dev · Mar 8, 2023 · 43e63a4 · 43e63a4
1 parent 3084dcb
commit 43e63a4
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 16 deletions.
diff --git a/logstash/pipelines/zeek/10_zeek_prep.conf b/logstash/pipelines/zeek/10_zeek_prep.conf
@@ -6,14 +6,18 @@ filter {
     drop { id => "drop_zeek_invalid_logs" }
   }
 
-  # tags may have been specified, like: conn(tagA,tagB,tagC).log, extract the log type (conn) and the tags (tagA,tagB,tagC)
-  # also normalize log types with - in their names to _ (e.g., opcua-binary -> opcua_binary)
+  # - Tags may have been specified, like: conn(tagA,tagB,tagC).log, extract the log type (conn) and the tags (tagA,tagB,tagC).
+  # - Normalize log types with - in their names to _ (e.g., opcua-binary -> opcua_binary).
+  # - Zeek log files might be caught by filebeat right in the middle of being renamed/moved as
+  #   part of log rotation (ie., renamed from conn.log to conn.2020-01-16-14-00-00.log or
+  #   conn.2020_01_16_14_00_00.log). We don't care about that, ignore the date part and just process
+  #   the log source as we normally would.
   ruby {
-    id => "ruby_zeek_source_extract"
-                                                                  #↓Type     ↓Tags
+    id => "ruby_zeek_log_source_extract"
+    #                                                                       ↓Type     ↓Tags   ↓Rotate Timestamp (discard)                                        ↓.log (discard)
     code => "
-      if fileParts = event.get('[log][file][path]').split('/').last.match(/^(.*?)(?:\((.*)\))?\.log/i) then
-        logType, tags = fileParts.captures
+      if fileParts = event.get('[log][file][path]').split('/').last.match(/^(.*?)(?:\((.*)\))?(?:\.\d{4}[_:-]?\d{2}[_:-]?\d{2}[_:-]?\d{2}[_:-]?\d{2}[_:-]?\d{2})?\.log/i) then
+        logType, tags  = fileParts.captures
         event.set('[log_source]', logType.gsub('-', '_')) unless logType.nil?
         event.set('[@metadata][zeek_log_tags]', tags) unless tags.nil?
       end"

diff --git a/logstash/pipelines/zeek/11_zeek_parse.conf b/logstash/pipelines/zeek/11_zeek_parse.conf
@@ -5077,10 +5077,6 @@ filter {
         add_tag => [ "ics" ]
       }
 
-    } else if ([log_source] =~ /\.\d{4}[_:-]?\d{2}[_:-]?\d{2}[_:-]?\d{2}[_:-]?\d{2}[_:-]?\d{2}$/) {
-      # filebeat caught a file right in the middle of being renamed/moved
-      drop { id => "drop_renamed_logfile_opcua" }
-
     } else {
       # some other unknown zeek opcua- log file. should start with ts at least!
       csv {
@@ -5105,12 +5101,6 @@ filter {
 
     } # if / else if for opcua log types
 
-  } else if ([log_source] =~ /\.\d{4}[_:-]?\d{2}[_:-]?\d{2}[_:-]?\d{2}[_:-]?\d{2}[_:-]?\d{2}$/) {
-      # filebeat caught a file right in the middle of being renamed/moved
-      # (ie., renamed from conn.log to conn.2020-01-16-14-00-00.log or conn.2020_01_16_14_00_00.log).
-      # this has actually already been processed, so ignore this event.
-      drop { id => "drop_renamed_logfile" }
-
   } else {
     # some other unknown zeek log file. should start with ts at least!
     csv {