企業級實戰模組四：Logstash配置語法詳解

1 Logstash基本語法組成

logstash之所以功能強大和流行，還與其豐富的過濾器外掛是分不開的，過濾器提供的並不單單是過濾的功能，還可以對進入過濾器的原始資料進行復雜的邏輯處理，甚至新增獨特的事件到後續流程中。

Logstash配置檔案有如下三部分組成，其中input、output部分是必須配置，filter部分是可選配置，而filter就是過濾器外掛，可以在這部分實現各種日誌過濾功能。

input {
  #輸入外掛
}
filter {
  #過濾匹配外掛
}
output {
  #輸出外掛
}

2 Logstash輸入外掛(Input)

2.1 讀取檔案(File)

logstash使用一個名為filewatch的ruby gem庫來監聽檔案變化,並通過一個叫.sincedb的資料庫檔案來記錄被監聽的日誌檔案的讀取進度（時間戳），這個sincedb資料檔案的預設路徑在 <path.data>/plugins/inputs/file下面，檔名類似於.sincedb_452905a167cf4509fd08acb964fdb20c，而<path.data>表示logstash外掛儲存目錄，預設是LOGSTASH_HOME/data。

看下面一個事件配置檔案：

input {
   file {
     path => ["/var/log/messages"]
     type => "system"
     start_position => "beginning"
   }
}
output {
   stdout{
    codec=>rubydebug  
   }
}
 

這個配置是監聽並接收本機的/var/log/messages檔案內容，start_position表示按時間戳記錄的地方開始讀取，如果沒有時間戳則從頭開始讀取，有點類似cat命令，預設情況下，logstash會從檔案的結束位置開始讀取資料，也就是說logstash程序會以類似tail -f命令的形式逐行獲取資料。type用來標記事件型別,通常會在輸入區域通過type標記事件型別。

2.2 標準輸入(Stdin)

stdin是從標準輸入獲取資訊，關於stdin的使用，前面已經做過了一些簡單的介紹，這裡再看一個稍微複雜一點的例子，下面是一個關於stdin的事件配置檔案：

input{
   stdin{
    add_field=>{"key"=>"iivey"}
    tags=>["add1"]
    type=>"test1"
   }
}
output {
   stdout{
    codec=>rubydebug  
   }
}
 

2.3 讀取 Syslog日誌

如何將rsyslog收集到的日誌資訊傳送到logstash中，這裡以centos7.5為例，需要做如下兩個步驟的操作：

首先，在需要收集日誌的伺服器上找到rsyslog的配置檔案/etc/rsyslog.conf,新增如下內容：

*.*  @@192.168.5.7:5514

其中，192.168.5.7是logstash伺服器的地址。5514是logstash啟動的監聽埠。

接著，重啟rsyslog服務：

systemctl restart rsyslog

然後，在logstash伺服器上建立一個事件配置檔案，內容如下：

input {
  syslog {
   port => "5514"
  }
}
​
output {
   stdout{
    codec=>rubydebug  
   }
}

2.4 讀取TCP網路資料

下面的事件配置檔案就是通過"LogStash::Inputs::TCP"和"LogStash::Filters::Grok"配合實現syslog功能的例子，這裡使用了logstash的TCP/UDP外掛讀取網路資料：

input {
  tcp {
   port => "5514"
  }
}
​
filter {
  grok {
   match => { "message" => "%{SYSLOGLINE}" }
  }
}
​
output {
   stdout{
    codec=>rubydebug
   }
}

其中，5514埠是logstash啟動的tcp監聽埠。注意這裡用到了日誌過濾"LogStash::Filters::Grok"功能，下面馬上會介紹到。

3 Logstash編碼外掛(Codec)

其實我們就已經用過編碼外掛codec了，也就是這個rubydebug，它就是一種codec，雖然它一般只會用在stdout外掛中，作為配置測試或者除錯的工具。

編碼外掛(Codec)可以在logstash輸入或輸出時處理不同型別的資料，因此，Logstash不只是一個input-->filter-->output的資料流，而是一個input-->decode-->filter-->encode-->output的資料流。

    Codec支援的編碼格式常見的有plain、json、json_lines等。

3.1 codec外掛之plain

plain是一個空的解析器，它可以讓使用者自己指定格式，也就是說輸入是什麼格式，輸出就是什麼格式。下面是一個包含plain編碼的事件配置檔案：

input{
    stdin{
	}
}
output{
    stdout{
        codec => "plain"
	}
}

3.2 codec外掛之json、json_lines

如果傳送給logstash的資料內容為json格式,可以在input欄位加入codec=>json來進行解析，這樣就可以根據具體內容生成欄位，方便分析和儲存。如果想讓logstash輸出為json格式，可以在output欄位加入codec=>json，下面是一個包含json編碼的事件配置檔案：

input {
    stdin {
        }
    }
output {
    stdout {
        codec => json
        }
}

這就是json格式的輸出，可以看出，json每個欄位是key:values格式，多個欄位之間通過逗號分隔。有時候，如果json檔案比較長，需要換行的話，那麼就要用json_lines編碼格式了。

4 Logstash過濾器外掛(Filter)

4.1 Grok 正則捕獲

grok是一個十分強大的logstash filter外掛，他可以通過正則解析任意文字，將非結構化日誌資料弄成結構化和方便查詢的結構。他是目前logstash 中解析非結構化日誌資料最好的方式。

Grok 的語法規則是：

	%{語法: 語義}

“語法”指的就是匹配的模式，例如使用NUMBER模式可以匹配出數字，IP模式則會匹配出127.0.0.1這樣的IP地址：

例如輸入的內容為：

192.168.50.21 [08/Oct/2021:23:24:19 +0800] "GET / HTTP/1.1" 403 5039

那麼，%{IP:clientip}匹配模式將獲得的結果為：

clientip: 192.168.50.21

%{HTTPDATE:timestamp}匹配模式將獲得的結果為：

timestamp: 08/Oct/2021:23:24:19 +0800

而%{QS:referrer}匹配模式將獲得的結果為：

referrer: "GET / HTTP/1.1"

下面是一個組合匹配模式，它可以獲取上面輸入的所有內容：

%{IP:clientip}\ \[%{HTTPDATE:timestamp}\]\ %{QS:referrer}\ %{NUMBER:response}\ %{NUMBER:bytes}	

通過上面這個組合匹配模式，我們將輸入的內容分成了五個部分，即五個欄位，將輸入內容分割為不同的資料欄位，這對於日後解析和查詢日誌資料非常有用，這正是使用grok的目的。

Logstash預設提供了近200個匹配模式（其實就是定義好的正則表示式）讓我們來使用，可以在logstash安裝目錄下，例如這裡是/usr/local/logstash/vendor/bundle/jruby/1.9/gems/logstash-patterns-core-4.1.2/patterns目錄裡面檢視，基本定義在grok-patterns檔案中。

從這些定義好的匹配模式中，可以查到上面使用的四個匹配模式對應的定義規則

示例：

input{
    stdin{}
}
filter{
    grok{
        match => ["message","%{IP:clientip}\ \[%{HTTPDATE:timestamp}\]\ %{QS:referrer}\ %{NUMBER:response}\ %{NUMBER:bytes}"]
    }
}
output{
    stdout{
        codec => "rubydebug"
    }
}

輸入內容：

192.168.50.21 [08/Oct/2021:23:24:19 +0800] "GET / HTTP/1.1" 403 5039

4.2 時間處理(Date)

date外掛是對於排序事件和回填舊資料尤其重要，它可以用來轉換日誌記錄中的時間欄位，變成LogStash::Timestamp物件，然後轉存到@timestamp欄位裡，這在之前已經做過簡單的介紹。

下面是date外掛的一個配置示例：

input{
    stdin{}
}
filter {
    grok {
        match => ["message", "%{HTTPDATE:timestamp}"]
    }
    date {
        match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
    }
}
output{
    stdout{
        codec => "rubydebug"
    }
}

4.3 資料修改(Mutate)

1）正則表示式替換匹配欄位

gsub可以通過正則表示式替換欄位中匹配到的值，只對字串欄位有效，下面是一個關於mutate外掛中gsub的示例（僅列出filter部分）：

filter {
    mutate {
        gsub => ["filed_name_1", "/" , "_"]
    }
}

這個示例表示將filed_name_1欄位中所有"/"字元替換為"_"。

2）分隔符分割字串為陣列

split可以通過指定的分隔符分割欄位中的字串為陣列，下面是一個關於mutate外掛中split的示例（僅列出filter部分）：

filter {
    mutate {
        split => ["filed_name_2", "|"]
    }
}

這個示例表示將filed_name_2欄位以"|"為區間分隔為陣列。

3）重新命名欄位

rename可以實現重新命名某個欄位的功能，下面是一個關於mutate外掛中rename的示例（僅列出filter部分）：

filter {
    mutate {
        rename => { "old_field" => "new_field" }
    }
}

這個示例表示將欄位old_field重新命名為new_field。

4）刪除欄位

remove_field可以實現刪除某個欄位的功能，下面是一個關於mutate外掛中remove_field的示例（僅列出filter部分）：

filter {
    mutate {
        remove_field  =>  ["timestamp"]
    }
}

這個示例表示將欄位timestamp刪除。

5）綜合示例

input {
    stdin {}
}
filter {
    grok {
        match => { "message" => "%{IP:clientip}\ \[%{HTTPDATE:timestamp}\]\ %{QS:referrer}\ %{NUMBER:response}\ %{NUMBER:bytes}" }
        remove_field => [ "message" ]
   }
date {
        match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
    }
mutate {
           rename => { "response" => "response_new" }
           convert => [ "response","float" ]
           gsub => ["referrer","\"",""]
           remove_field => ["timestamp"]
           split => ["clientip", "."]
        }
}
output {
    stdout {
        codec => "rubydebug"
    }
}

4.4 GeoIP 地址查詢歸類

GeoIP是最常見的免費IP地址歸類查詢庫，當然也有收費版可以使用。GeoIP庫可以根據IP 地址提供對應的地域資訊，包括國別，省市，經緯度等，此外掛對於視覺化地圖和區域統計非常有用。

下面是一個關於GeoIP外掛的簡單示例（僅列出filter部分）：

filter {
    geoip {
        source => "ip_field"
    }
}

其中，ip_field欄位是輸出IP地址的一個欄位。

4.5 filter外掛綜合應用例項

下面給出一個業務系統輸出的日誌格式，由於業務系統輸出的日誌格式無法更改，因此就需要我們通過logstash的filter過濾功能以及grok外掛來獲取需要的資料格式，此業務系統輸出的日誌內容以及原始格式如下：

2021-10-09T0:57:42+08:00|~|123.87.240.97|~|Mozilla/5.0 (iPhone; CPU iPhone OS 11_2_2 like Mac OS X) AppleWebKit/604.4.7 Version/11.0 Mobile/15C202 Safari/604.1|~|http://m.sina.cn/cm/ads_ck_wap.html|~|1460709836200|~|DF0184266887D0E

可以看出，這段日誌都是以“|~|”為區間進行分隔的，那麼剛好我們就以“|~|”為區間分隔符，將這段日誌內容分割為6個欄位。這裡通過grok外掛進行正則匹配組合就能完成這個功能。

完整的grok正則匹配組合語句如下：

%{TIMESTAMP_ISO8601:localtime}\|\~\|%{IPORHOST:clientip}\|\~\|(%{GREEDYDATA:http_user_agent})\|\~\|(%{DATA:http_referer})\|\~\|%{GREEDYDATA:mediaid}\|\~\|%{GREEDYDATA:osid}

5 Logstash輸出外掛（output）

5.1 輸出到標準輸出(stdout)

stdout與之前介紹過的stdin外掛一樣，它是最基礎和簡單的輸出外掛，下面是一個配置例項：

output {
    stdout {
        codec => rubydebug
    }
}

stdout外掛，主要的功能和用途就是用於除錯，這個外掛，在前面已經多次使用過。這裡不再過多介紹。

5.2 儲存為檔案（file）

file外掛可以將輸出儲存到一個檔案中，配置例項如下：

output {
    file {
        path => "/data/log3/%{+yyyy-MM-dd}/%{host}_%{+HH}.log"
    }

上面這個配置中，使用了變數匹配，用於自動匹配時間和主機名，這在實際使用中很有幫助。

5.3 輸出到elasticsearch

Logstash將過濾、分析好的資料輸出到elasticsearch中進行儲存和查詢，是最經常使用的方法。下面是一個配置例項：

    output {
    elasticsearch {
        host => ["192.168.5.8:9200","192.168.5.9:9200","192.168.5.10:9200"]
        index => "logstash-%{+YYYY.MM.dd}"
        manage_template => false
        template_name => "template-web_access_log"
    }
}

上面配置中每個配置項含義如下：

• host：是一個數組型別的值，後面跟的值是elasticsearch節點的地址與埠，預設埠是9200。可新增多個地址。

• index：寫入elasticsearch的索引的名稱，這裡可以使用變數。Logstash提供了%{+YYYY.MM.dd}這種寫法。在語法解析的時候，看到以+ 號開頭的，就會自動認為後面是時間格式，嘗試用時間格式來解析後續字串。這種以天為單位分割的寫法，可以很容易的刪除老的資料或者搜尋指定時間範圍內的資料。此外，注意索引名中不能有大寫字母。

• manage_template:用來設定是否開啟logstash自動管理模板功能，如果設定為false將關閉自動管理模板功能。如果我們自定義了模板，那麼應該設定為false。

• template_name:這個配置項用來設定在Elasticsearch中模板的名稱。