本文作者：羅海鵬，叩丁狼高階講師。原創文章，轉載請註明出處。 

前言

在第九章節中，我們已經安裝好Logstash元件了，並且啟動例項測試它的資料輸入和輸出，但是用的是最簡單的控制檯標準輸入和標準輸出，那這節我們就來深入的學習Logstash的詳細使用。

常用啟動引數

我們在上一節中演示了啟動Logstash的例項，其中我們啟動的時候給Logstash指令碼傳入了-e的引數，但實際上，Logstash的啟動引數有很多，我們來看一下各個啟動引數的作用：

• -e #立即啟動例項，例如：./logstash -e "input {stdin {}} output {stdout {}}"
• -f #指定啟動例項的配置檔案，例如：./logstash -f config/test.conf
• -t #測試配置檔案的正確性，例如：./logstash -f config/test.conf -t
• -l #指定日誌檔名稱，例如：./logstash -f config/test.conf -l logs/test.log
• -w #指定filter執行緒數量，不指定預設是5，例如：./logstash-f config/test.conf -w 8

配置檔案語法

檔案結構

我們剛剛知道，啟動引數可以指定一個配置檔案，那麼接下來就有必要來了解一下配置檔案的結構：Logstash通過{}來定義區域，區域內可以定義外掛，一個區域內可以定義多個外掛，如下：

input {
    #標準輸入源外掛
    stdin {
    }
    #普通檔案源外掛
    file {
        path => ["/var/log/*.log", "/var/log/message"]
       ....
    }
   ......
}
filter {
      #grok過濾外掛
      grok {
            match => ["message", "%{HTTPDATE:logdate}"]
            .....
      }
     #date過濾外掛
     date {
            match => ["logdate", "dd/MMM/yyyy:HH:mm:ss Z"]
           .....
    }
   .....
}
output {
     stdout {
     }
     elasticsearch {
        hosts => ["127.0.0.1:9200"]
        ....
    }
    .....
}
 

我們先大概瞭解一下配置檔案的結構，接下來我們再詳細看這些外掛的配置。

資料型別

Logstash配置檔案支援的資料型別有： 1、Boolean，例如：ssl_enable => true 2、Number，例如：port => 33 3、String，例如：name => “Hello world” 4、hash，例如：options => {key1 => “value1”, key2 => “value2”} 5、array，例如：match => [“datetime”, “UNIX”, “ISO8601”]

欄位引用

Logstash資料流中的資料被稱之為Event物件，Event以JSON結構構成，Event的屬性被稱之為欄位，如果你想在配置檔案中引用這些欄位，只需要把欄位的名字寫在中括號[]

{
      "message" => "hello logstash",
      "@version" => "1",
      "@timestamp" => 2018-08-13T17:32:01.122Z,
      "host" => "localhost.localdomain"
}

條件判斷

Logstash支援下面的操作符： 1、==(等於), !=(不等於), <(小於), >(大於), <=(小於等於), >=(大於等於) 2、=~(匹配正則), !~（不匹配正則） 3、in(包含), not in(不包含) 4、and(與), or(或), nand(非與), xor(非或) 5、()(複合表示式), !()(對複合表示式結果取反)例如以下的條件判斷：

if "_grokparsefailure" not in [tags] {
} 
else if [status] !~ /^2\d\d/ or ( [url] == "/noc.gif" nand [geoip][city] != "beijing" ) {
} 
else {
}

環境變數引用

Logstash支援引用系統環境變數，環境變數不存在時可以設定預設值，例如：

export TCP_PORT=12345
input {
  tcp {
    port => "${TCP_PORT:54321}"
  }
}

常用輸入外掛

在第九章中，我們已經使用是標準輸入，以鍵盤的輸入資料作為Logstash資料來源，但實際上我們也知道，Logstash的資料來源有很多，每種資料來源都有相應的配置，在Logstash中，這些資料來源的相應配置稱為外掛，我們常用的輸入外掛有：file、jdbc、redis、tcp、syslog，這些輸入外掛會監聽資料來源的資料，如果新增資料，將資料封裝成Event程序處理或者傳遞，更多的輸入外掛大家可以看Logstash官網，接下來我們以file和jdbc兩個輸入外掛作為例子，來學習輸入外掛的使用，其他輸入外掛使用起來大同小異，大家自行擴充套件。

file輸入外掛

讀取檔案外掛主要用來抓取檔案的資料變化資訊，以此作為Logstash的資料來源。

• 配置示例：

  input{
  file {
    path => ["/var/log/*.log", "/var/log/message"]
    type => "system"
    start_position => "beginning"
  }
  }
  output{
  stdout{}
  }
  

• 常用引數

jdbc輸入外掛

該外掛可以使用jdbc把關係型資料庫的資料作為Logstash的資料來源

• 配置示例：

  input {
  jdbc {
    jdbc_driver_library => "/opt/logstash/mysql-connector-java-5.1.36-bin.jar"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://localhost:3306/mydb"
    jdbc_user => "mysql"
    jdbc_password => "123456"
    parameters => { "favorite_artist" => "Beethoven" }
    schedule => "* * * * *"
    statement => "SELECT * from songs where artist = :favorite_artist"
  }
  }
  output{
  stdout{}
  }
  

• 常用引數(空 = 同上)

常用過濾外掛

豐富的過濾器外掛的是 logstash威力如此強大的重要因素，過濾器外掛主要處理流經當前Logstash的事件資訊，可以新增欄位、移除欄位、轉換欄位型別，通過正則表示式切分資料等，也可以根據條件判斷來進行不同的資料處理方式。我們常用的過濾外掛有：grok、date、geoip、mutate、json、Split、ruby，更多的過濾外掛大家可以看Logstash官網，接下來我們以grok、date和geoip這3個過濾外掛作為例子，來學習過濾外掛的使用，其他過濾外掛使用起來大同小異，大家自行擴充套件。

grok正則外掛

grok正則捕獲是Logstash中將非結構化資料解析成結構化資料以便於查詢的最好工具，非常適合解析system log，web log， database log等任意的 log檔案。

• 內建正則表示式呼叫 grok提供100多個常用正則表示式可供使用，這100多個正則表示式定義在logstash/vendor/bundle/jruby/x.x/gems/logstash-patterns-core-xxx/patterns/grok-patterns檔案中，想要靈活的匹配各種資料，那麼必須檢視該檔案，大概瞭解grok提供了什麼內建的正則表示式。呼叫它們的語法如下：%{SYNTAX:SEMANTIC} SYNTAX：表示內建的正則表示式的名稱 SEMANTIC：表示在Event中建立該欄位名來儲存匹配到的值 例如：輸入的資料內容為“[debug] 127.0.0.1 - test log content”，我們想提取127.0.0.1這個IP地址，那麼可以使用以下語法來匹配：%{IP:client}，將獲得“client: 127.0.0.1”的結果，該結果將成為Event的一個新的欄位和欄位值；如果你在捕獲資料時想進行資料型別轉換可以使用%{NUMBER:num:int}這種語法，預設情況下，所有的返回結果都是string型別，當前Logstash所支援的轉換型別僅有“int”和“float”；

• 自定義表示式呼叫 與預定義表示式相同，你也可以將自定義的表示式配置到Logstash中，然後就可以像於定義的表示式一樣使用；以下是操作步驟說明： 1、在Logstash根目錄下建立資料夾“patterns”，在“patterns”資料夾中建立檔案“extra”（檔名稱無所謂，可自己選擇有意義的檔名稱）； 2、在檔案“extra”中新增表示式，格式：patternName regexp，名稱與表示式之間用空格隔開即可，例如：POSTFIX_QUEUEID [0-9A-F]{10,11} 3、使用自定義的表示式時需要在grok外掛配置“patterns_dir”屬性，屬性值為extra檔案所在的目錄。

• 配置示例 日誌檔案http.log每行內容為：55.3.244.1 GET /index.html 15824 0.043 message-id:BEF25A72965 grok表示式：表示式：%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration} 配置檔案內容：

  input {
  file {
    path => "/var/log/http.log"
  }
  }
  filter {
  grok {
    patterns_dir => ["./patterns"]
    match => {"message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration} message\-id:%{POSTFIX_QUEUEID:queueid}"}
  }
  }
  output{
    stdout{}
  }
  

  輸出結果：

  client: 55.3.244.1
  method: GET
  request: /index.html
  bytes: 15824
  duration: 0.043
  queueid: BEF25A72965
  

• 示例解析 1、/var/log/http.log檔案每一行的格式為55.3.244.1 GET /index.html 15824 0.043 message-id:BEF25A72965，到時候會把每一行資料封裝成一個Event。 2、使用grok過濾外掛處理該檔案，match為匹配Event中的message，message就是該檔案的一行資料，比如55.3.244.1 GET /index.html 15824 0.043 message-id:BEF25A72965。 3、匹配message的內容使用%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration} message\-id:%{POSTFIX_QUEUEID:queueid}表示式，把一條message拆成client、method、request、bytes、duration、queueid這6個欄位，並新增到Event中。 4、其中%{POSTFIX_QUEUEID:queueid}為自定義表示式的呼叫，其他5個是grok內建的表示式呼叫，如果需要使用自定義表示式，則需要在grok外掛配置patterns_dir屬性，屬性值資料型別為array，就是自定義表示式定義的檔案所在的目錄。 5、在表示式中，特殊字元需要使用\來轉義，比如-，""，[]這些特殊字元，所以我們上面例項中的message-id資料在用表示式匹配的時候是使用了message\-id去匹配了。
• 常用引數(空 = 同上)

date時間處理外掛

在講date外掛的使用前，我們需要先講解一下Logstash的時間記錄方式。在Logstash產生了一個Event物件的時候，會給該Event設定一個時間，欄位為“@timestamp”，同時，我們的日誌內容一般也會有時間，但是這兩個時間是不一樣的，因為日誌內容的時間是該日誌打印出來的時間，而“@timestamp”欄位的時間是input外掛接收到了一條資料並建立Event的時間，所有一般來說的話“@timestamp”的時間要比日誌內容的時間晚一點，因為Logstash監控資料變化，資料輸入，建立Event導致的時間延遲。這兩個時間都可以使用，具體要根據自己的需求來定。 但是不管“@timestamp”欄位的時間還是日誌內容中的時間，其時間格式一般都不是我們想要的，所以我們就需要使用date外掛，把時間格式轉成我們想要的格式，比如：比如將Apr 17 09:32:01（MMM dd HH:mm:ss）轉換為04-17 09:32:01 （MM-dd HH:mm:ss）。

• 配置示例：

  filter {
    date {
        match => ["timestamp","dd/MMM/YYYY:HH:mm:ss Z"]
    } 
  }
  

• 示例解析timestamp是自定義的Event欄位，用於存放通過grok解析到的日誌內容中的時間，dd/MMM/YYYY:HH:mm:ss Z是日誌內容中的時間格式，比如：16/Sep/2018:00:42:38 +0800，匹配到了之後，date外掛會預設把該時間轉成本地格式的時間，並且覆蓋掉Event為我們建立的欄位中的時間。
• 常用引數(空 = 同上)

geoip外掛

• 配置示例：

  filter {
     if [remote_ip] !~ "^127\.|^192\.168\.|^172\.1[6-9]\.|^172\.2[0-9]\.|^172\.3[01]\.|^10\." {
        geoip {
             source => "remote_ip"
             database => "/usr/share/GeoIP/GeoLite2-Country.mmdb"
        }
    }
  }
  

• 示例解析 由於geoip使用的ip資料庫不能匹配私網地址，所以在使用geoip外掛前，先判斷一下ip地址是否為私網ip，如果不是，這使用geoip外掛。source為指定一個ip地址，為其查詢歸屬地等資訊，remote_ip為Event中儲存ip地址的自定義欄位，database為IP地址資料庫所在的位置，不指定的話，使用geoip外掛內建的GeoLite2-City預設資料庫。
• 常用引數(空 = 同上)

常用輸出外掛

經過以上的學習，我們已經學習了Logstash三大主件中的其中兩個了，分別是input和filter，那現在我們就來學習最後一個元件：output。 每個資料流經過input和filter後，最終要由output輸出，以上的內容我們都是使用最簡單的標準輸出：stdout，把資料輸出到顯示器，但實際上stdout只是Logstash的其中一個輸出外掛而已，它的常用輸出外掛有：Elasticsearch，Redis，File，TCP等等，更多的輸出外掛大家可以看Logstash官網，接下來我們以Elasticsearch和Redis輸出外掛作為例子，來學習輸出外掛的使用，其他過濾外掛使用起來大同小異，大家自行擴充套件。

elasticsearch輸出外掛

用於將Event資訊寫入到Elasticsearch中，官方推薦外掛，ELK技術棧必備外掛。

• 配置示例

  output {
    elasticsearch {
        hosts => ["127.0.0.1:9200"]
        index => "logstash-%{type}-%{+YYYY.MM.dd}"
        document_type => "%{type}"
    }
  }
  

• 常用引數(空 = 同上)

redis輸出外掛

用於將Event寫入Redis中進行快取，由於Redis資料庫是先把資料存在記憶體的，所以效率會非常高，是一個常用的logstash輸出外掛

• 配置示例

  output {
    redis {
        host => ["127.0.0.1"]
        port => 6379
        data_type => "list"
        key => "logstash-list"
    }
  }
  

• 常用引數(空 = 同上)

後記

那到這裡，我們已經介紹了Logstash配置檔案的語法和常用外掛的配置方式了，這些常用的外掛都是使用頻率非常高的，所有我們後面需要來做一個Logstash的實戰案例，綜合運用我們這章所學的內容。