Logstash filter 插件之 date

使用 date 插件解析字段中的日期，然后使用該日期或時間戳作為事件的 logstash 時間戳。對於排序事件和導入舊數據，日期過濾器尤其重要。如果您在事件中沒有得到正確的日期，那么稍后搜索它們可能會出現問題。

如果時間戳尚未在事件中設置，logstash 將根據第一次看到事件(在輸入時)創建一個時間戳。例如，對於文件輸入，時間戳設置為每次讀取的時間。

本文演示如何把現有的日志數據導入到 elasticsearch 中，並用日志中的時間信息設置事件的時間戳。

拆分日志信息

比如我們的日志格式如下：

[Trace] [e1a618cf-186f-49c5-b486-111e1e5f0023] [2019/03/25] [23:47:20 618]

第一個字段為 loglevel，第二個字段標識 SessionID，第三個字段是產生日志的日志，第四個字段是產生日志的時間，第三、四字段記錄的是本地時間，即東八區區時(哈哈，居然不是記一個 UTC 時間戳！)。

我們先使用下面的 grok 規則切分出日志中的字段：

filter {
    grok {
        match => {
            "message" => "\[%{LOGLEVEL:loglevel}\]\s*\[(?<SessionID>.*)\]\s*\[(?<Date>%{YEAR}/%{MONTHNUM}/%{MONTHDAY})\]\s*\[(?<Time>%{HOUR}:%{MINUTE}:%{SECOND} %{INT})\]"
        }
    }
}

得到 json 格式的日志記錄如下(Grok Debugger 的輸出)：

{
  "loglevel": [
    [
      "Trace"
    ]
  ],
  "SessionID": [
    [
      "e1a618cf-186f-49c5-b486-111e1e5f0023"
    ]
  ],
  "Date": [
    [
      "2019/03/25"
    ]
  ],
  "Time": [
    [
      "23:47:20 618"
    ]
  ]
}

合並日期和時間字段

把分散的兩個字段合並為時間戳字段，並移除 Date 和 Time 字段：

filter {
    mutate {
        add_field => { "Datetime" => "%{Date} %{Time}" }
    }
    mutate {
        remove_field => ["Date"]
    }
    mutate {
        remove_field => ["Time"]
    }
}

拼出來的 DateTime 字段中的內容格式為：

"2019/03/25 23:47:20 618"

為事件設置時間戳

下面使用 date 插件解析字段中的日期，然后使用該日期或時間戳作為事件的 logstash 時間戳：

filter {
    mutate {
      add_field => { "logtime" => "%{Datetime}" }
    }
    date {
      timezone => "Asia/Chongqing"
      match => ["logtime", "yyyy/MM/dd HH:mm:ss SSS"]
      target => "@timestamp"
      remove_field => [ "logtime" ]
    }
}

這里使用了一個臨時字段 logtime 來保存時間戳，原因是 date 插件會把該字段的類型轉換為 date，使用一個臨時的字段就不會影響到 Datetime 字段的類型(這也是自定義 @timestamp 時的常用手法)。
注意上面的 timezone 配置，如果日志中的信息是以 UTC 格式保存的，就不需要指定時區。但是筆者處理的日志中，時間信息保存的是東八區的區時，因此需要指定時區信息，date 插件才能把它轉換為 UTC 時間。
下圖是該記錄導入后在 Kibana 中顯示的情況，可以看到日期信息和時間戳是一致的：

參考：
Date filter plugin