logstash高速入口

時間 2019-11-29

標籤 logstash 高速入口欄目日誌分析简体版

原文原文鏈接

原文地址：http://logstash.net/docs/1.4.2/tutorials/getting-started-with-logstash

英語水平有限，假設有錯誤請各位指正

簡單介紹

Logstash是一個接收，處理，轉發日誌的工具。支持系統日誌，webserver日誌。錯誤日誌。應用日誌，總之包含所有可以拋出來的日誌類型。

怎麼樣聽起來挺厲害的吧？
在一個典型的使用場景下(ELK)：用Elasticsearch做爲後臺數據的存儲，kibana用來前端的報表展現。php

Logstash在其過程當中擔任搬運工的角色，它爲數據存儲。報表查詢和日誌解析建立了一個功能強大的管道鏈。css

Logstash提供了多種多樣的 input,filters,codecs和output組件，讓使用者輕鬆實現強大的功能。好了讓咱們開始吧
html

依賴條件：JAVA

Logstash執行只依賴java執行環境(jre)。各位可以在命令行下執行java -version命令

顯示相似例如如下結果：

java -version
java version "1.7.0_45"
Java(TM) SE Runtime Environment (build 1.7.0_45-b18)
Java HotSpot(TM) 64-Bit Server VM (build 24.45-b08, mixed mode)

爲了確保成功執行Logstash建議你們使用較最近的jre版本號。

可以獲取開源版本號的jre在：http://openjdk.java.net

或者你可以在官網下載Oracle jdk版本號：http://www.oracle.com/technetwork/java/index.html

一旦jre已經成功在你的系統中安裝完畢，咱們就可以繼續了

啓動和執行Logstash的兩條命令演示樣例

第一步咱們先下載Logstash

curl -O https://download.elasticsearch.org/logstash/logstash/logstash-1.4.2.tar.gz

現在你應該有了一個叫logstash-1.4.2.tar.gz的文件了。咱們把它解壓一下

tar zxvf logstash-1.4.2.tar.gz
cd logstash-1.4.2

現在咱們來執行一下：

bin/logstash -e 'input { stdin { } } output { stdout {} }'

咱們現在可以在命令行下輸入一些字符。而後咱們將看到logstash的輸出內容：

hello world
2013-11-21T01:22:14.405+0000 0.0.0.0 hello world

Ok,還挺有意思的吧... 以上樣例咱們在執行logstash中，定義了一個叫"stdin"的input另外一個"stdout"的output，無論咱們輸入什麼字符。Logstash都會依照某種格式來返回咱們輸入的字符。

這裏注意咱們在命令行中使用了-e參數，該參數贊成Logstash直接經過命令行接受設置。這點尤爲高速的幫助咱們重複的測試配置是否正確而不用寫配置文件。前端

讓咱們再試個更有意思的樣例。首先咱們在命令行下使用CTRL-C命令退出以前執行的Logstash。現在咱們又一次執行Logstash使用如下的命令：

bin/logstash -e 'input { stdin { } } output { stdout { codec => rubydebug } }'

咱們再輸入一些字符，此次咱們輸入"goodnight moon"：

goodnight moon
{
  "message" => "goodnight moon",
  "@timestamp" => "2013-11-20T23:48:05.335Z",
  "@version" => "1",
  "host" => "my-laptop"
}

以上演示樣例經過又一次設置了叫"stdout"的output(加入了"codec"參數)，咱們就可以改變Logstash的輸出表現。相似的咱們可以經過在你的配置文件里加入或者改動inputs、outputs、filters，就可以使任意的格式化日誌數據成爲可能，從而訂製更合理的存儲格式爲查詢提供便利。

使用Elasticsearch存儲日誌

現在，你或許會說："它看起來還挺高大上的，只是手工輸入字符。並把字符從控制檯回顯出來。實際狀況並不有用"。

說的好，那麼接下來咱們將創建Elasticsearch來存儲輸入到Logstash的日誌數據。假設你尚未安裝Elasticsearch。你可以下載RPM/DEB包或者手動下載tar包。經過下面命令：java

curl -O https://download.elasticsearch.org/elasticsearch/elasticsearch/elasticsearch-1.1.1.tar.gz
tar zxvf elasticsearch-1.1.1.tar.gz
cd elasticsearch-1.1.1/
./bin/elasticsearch

注意

本篇文章實例使用Logstash 1.4.2和Elasticsearch 1.1.1。

不一樣的Logstash版本號都有相應的建議Elasticsearch版本號。請確認你使用的Logstash版本號！git

不少其它有關安裝和設置Elasticsearch的信息可以參考Elasticsearch官網。因爲咱們主要介紹Logstash的入門使用，Elasticsearch默認的安裝和配置就已經知足咱們要求。

言歸正專。現在Elasticsearch已經執行並監聽9200port了(你們都搞定了，對嗎？)，經過簡單的設置Logstash就可以使用Elasticsearch做爲它的後端。

默認的配置對於Logstash和Elasticsearch已經足夠，咱們忽略一些額外的選項來設置elasticsearch做爲output：github

bin/logstash -e 'input { stdin { } } output { elasticsearch { host => localhost } }'

任意的輸入一些字符。Logstash會像以前同樣處理日誌(只是此次咱們將不會看到不論什麼的輸出，因爲咱們沒有設置stdout做爲output選項)

you know, for logs

咱們可以使用curl命令發送請求來查看ES是否接收到了數據：

curl 'http://localhost:9200/_search?pretty'

返回內容例如如下：

{
  "took" : 2,
  "timed_out" : false,
  "_shards" : {
    "total" : 5,
    "successful" : 5,
    "failed" : 0
  },
  "hits" : {
    "total" : 1,
    "max_score" : 1.0,
    "hits" : [ {
      "_index" : "logstash-2013.11.21",
      "_type" : "logs",
      "_id" : "2ijaoKqARqGvbMgP3BspJA",
      "_score" : 1.0, "_source" : {"message":"you know, for logs","@timestamp":"2013-11-21T18:45:09.862Z","@version":"1","host":"my-laptop"}
    } ]
  }
}

恭喜，至此你已經成功利用Elasticsearch和Logstash來收集日誌數據了。

Elasticsearch 插件(題外話)

這裏介紹另一個對於查詢你的Logstash數據(Elasticsearch中數據 )很實用的工具叫Elasticsearch-kopf插件。不少其它的信息請見Elasticsearch插件。

安裝elasticsearch-kopf，僅僅要在你安裝Elasticsearch的文件夾中運行下面命令就能夠：web

bin/plugin -install lmenezes/elasticsearch-kopf

接下來訪問 http://localhost:9200/_plugin/kopf 來瀏覽保存在Elasticsearch中的數據，設置及映射！

多重輸出

做爲一個簡單的樣例來設置多重輸出，讓咱們同一時候設置stdout和elasticsearch做爲output來又一次執行一下Logstash。例如如下：

bin/logstash -e 'input { stdin { } } output { elasticsearch { host => localhost } stdout { } }'

當咱們輸入了一些詞組以後。這些輸入的內容回回顯到咱們的終端，同一時候還會保存到Elasticsearch！

(可以使用curl和kopf插件來驗證)。redis

默認配置 - 依照每日日期創建索引

你將發現Logstash可以足夠機靈的在Elasticsearch上創建索引... 天天會依照默認格式是logstash-YYYY.MM.DD來創建索引。

在午夜(GMT)，Logstash本身主動依照時間戳更新索引。咱們可以根據追溯多長時間的數據做爲根據來制定保持多少數據。固然你也可以把比較老的數據遷移到其它的地方(又一次索引)來方便查詢，此外假設不過簡單的刪除一段時間數據咱們可以使用Elasticsearch Curator。shell

接下來

接下來咱們開始瞭解不少其它高級的配置項。在如下的章節，咱們着重討論logstash一些核心的特性，以及怎樣和logstash引擎交互的。

事件的生命週期

Inputs,Outputs,Codecs,Filters構成了Logstash的核心配置項。

Logstash經過創建一條事件處理的管道。從你的日誌提取出數據保存到Elasticsearch中。爲高效的查詢數據提供基礎。

爲了讓你高速的瞭解Logstash提供的多種選項，讓咱們先討論一下最常用的一些配置。

不少其它的信息，請參考Logstash事件管道。

Inputs

input 及輸入是指日誌傳輸數據到Logstash中。當中常見的配置例如如下：

file：從文件系統中讀取一個文件，很是像UNIX命令 "tail -0a"
syslog：監聽514port，依照RFC3164標準解析日誌數據
redis：從redisserver讀取數據。支持channel(公佈訂閱)和list模式。
redis通常在Logstash消費集羣中做爲"broker"角色，保存events隊列共Logstash消費。
lumberjack：使用lumberjack協議來接收數據，眼下已經改成 logstash-forwarder。

Filters

Fillters 在Logstash處理鏈中擔任中間處理組件。他們經常被組合起來實現一些特定的行爲來，處理匹配特定規則的事件流。常見的filters例如如下：

grok：解析無規則的文字並轉化爲有結構的格式。
Grok 是眼下最好的方式來將無結構的數據轉換爲有結構可查詢的數據。
有120多種匹配規則，會有一種知足你的需要。
mutate：mutate filter 贊成改變輸入的文檔，你可以從命名，刪除，移動或者改動字段在處理事件的過程當中。
drop：丟棄一部分events不進行處理，好比：debug events。
clone：拷貝 event。這個過程當中也可以加入或移除字段。
geoip：加入地理信息(爲前臺kibana圖形化展現使用)

Outputs

outputs是logstash處理管道的最末端組件。一個event可以在處理過程當中通過多重輸出，但是一旦所有的outputs都運行結束，這個event也就完畢生命週期。一些常用的outputs包含：

elasticsearch：假設你計劃將高效的保存數據，並且能夠方便和簡單的進行查詢...Elasticsearch是一個好的方式。是的，此處有作廣告的嫌疑,呵呵。
file：將event數據保存到文件裏。
graphite：將event數據發送到圖形化組件中，一個很是流行的開源存儲圖形化展現的組件。http://graphite.wikidot.com/。
statsd：statsd是一個統計服務，比方技術和時間統計。經過udp通信，聚合一個或者多個後臺服務，假設你已經開始使用statsd，該選項對你應該很是實用。

Codecs

codecs 是基於數據流的過濾器，它可以做爲input。output的一部分配置。

Codecs可以幫助你輕鬆的切割發送過來已經被序列化的數據。流行的codecs包含 json,msgpack,plain(text)。

json：使用json格式對數據進行編碼/解碼
multiline：將匯多個事件中數據彙總爲一個單一的行。
比方：java異常信息和堆棧信息

獲取完整的配置信息。請參考 Logstash文檔中 "plugin configuration"部分。

不少其它有趣Logstash內容

使用配置文件

使用-e參數在命令行中指定配置是非常常常使用的方式，只是假設需要配置不少其它設置則需要很是長的內容。這樣的狀況。咱們首先建立一個簡單的配置文件，並且指定logstash使用這個配置文件。如咱們建立一個文件名稱是"logstash-simple.conf"的配置文件並且保存在和Logstash一樣的文件夾中。

內容例如如下：

input { stdin { } }
output {
  elasticsearch { host => localhost }
  stdout { codec => rubydebug }
}

接下來，運行命令：

bin/logstash -f logstash-simple.conf

咱們看到logstash依照你剛剛建立的配置文件來執行樣例。這樣更加的方便。注意，咱們使用-f參數來從文件獲取而取代以前使用-e參數從命令行中獲取配置。以上演示很easy的樣例。固然解析來咱們繼續寫一些複雜一些的樣例。

過濾器

filters是一個行處理機制將提供的爲格式化的數據整理成你需要的數據，讓咱們看看如下的一個樣例，叫grok filter的過濾器。

input { stdin { } }

filter {
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}" }
  }
  date {
    match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]
  }
}

output {
  elasticsearch { host => localhost }
  stdout { codec => rubydebug }
}

運行Logstash依照例如如下參數：

bin/logstash -f logstash-filter.conf

現在粘貼如下一行信息到你的終端(固然Logstash就會處理這個標準的輸入內容)：

127.0.0.1 - - [11/Dec/2013:00:01:45 -0800] "GET /xampp/status.php HTTP/1.1" 200 3891 "http://cadenza/xampp/navi.php" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.9; rv:25.0) Gecko/20100101 Firefox/25.0"

你將看到相似例如如下內容的反饋信息：

{
        "message" => "127.0.0.1 - - [11/Dec/2013:00:01:45 -0800] \"GET /xampp/status.php HTTP/1.1\" 200 3891 \"http://cadenza/xampp/navi.php\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10.9; rv:25.0) Gecko/20100101 Firefox/25.0\"",
     "@timestamp" => "2013-12-11T08:01:45.000Z",
       "@version" => "1",
           "host" => "cadenza",
       "clientip" => "127.0.0.1",
          "ident" => "-",
           "auth" => "-",
      "timestamp" => "11/Dec/2013:00:01:45 -0800",
           "verb" => "GET",
        "request" => "/xampp/status.php",
    "httpversion" => "1.1",
       "response" => "200",
          "bytes" => "3891",
       "referrer" => "\"http://cadenza/xampp/navi.php\"",
          "agent" => "\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10.9; rv:25.0) Gecko/20100101 Firefox/25.0\""
}

正像你看到的那樣，Logstash(使用了grok過濾器)能夠將一行的日誌數據(Apache的"combined log"格式)切割設置爲不一樣的數據字段。這一點對於往後解析和查詢咱們本身的日誌數據很實用。比方：HTTP的返回狀態碼。IP地址相關等等。很的easy。不多有匹配規則沒有被grok包括，因此假設你正嘗試的解析一些常見的日誌格式。也許已經有人爲了作了這種工做。假設查看具體匹配規則。參考logstash grok patterns。

另一個過濾器是date filter。這個過濾器來負責解析出來日誌中的時間戳並將值賦給timestame字段(不管這個數據是何時收集到logstash的)。

你或許注意到在這個樣例中@timestamp字段是設置成December 11, 2013, 說明logstash在日誌產生以後一段時間進行處理的。這個字段在處理日誌中回添到數據中的，舉例來講... 這個值就是logstash處理event的時間戳。

有用的樣例

Apache 日誌(從文件獲取)

現在，讓咱們使用一些頗有用的配置... apache2訪問日誌！咱們將從本地讀取日誌文件，並且經過條件設置處理知足咱們需要的event。

首先。咱們建立一個文件名稱是logstash-apache.conf的配置文件。內容例如如下(你可以依據實際狀況改動你的文件名稱和路徑)：

input {
  file {
    path => "/tmp/access_log"
    start_position => beginning
  }
}

filter {
  if [path] =~ "access" {
    mutate { replace => { "type" => "apache_access" } }
    grok {
      match => { "message" => "%{COMBINEDAPACHELOG}" }
    }
  }
  date {
    match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]
  }
}

output {
  elasticsearch {
    host => localhost
  }
  stdout { codec => rubydebug }
}

接下來。咱們依照上面的配置建立一個文件(在樣例中是"/tmp/access.log"),可以將如下日誌信息做爲文件內容(也可以用你本身的webserver產生的日誌)：

71.141.244.242 - kurt [18/May/2011:01:48:10 -0700] "GET /admin HTTP/1.1" 301 566 "-" "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US; rv:1.9.2.3) Gecko/20100401 Firefox/3.6.3"
134.39.72.245 - - [18/May/2011:12:40:18 -0700] "GET /favicon.ico HTTP/1.1" 200 1189 "-" "Mozilla/4.0 (compatible; MSIE 8.0; Windows NT 5.1; Trident/4.0; .NET CLR 2.0.50727; .NET CLR 3.0.4506.2152; .NET CLR 3.5.30729; InfoPath.2; .NET4.0C; .NET4.0E)"
98.83.179.51 - - [18/May/2011:19:35:08 -0700] "GET /css/main.css HTTP/1.1" 200 1837 "http://www.safesand.com/information.htm" "Mozilla/5.0 (Windows NT 6.0; WOW64; rv:2.0.1) Gecko/20100101 Firefox/4.0.1"

現在使用-f參數來運行一下上面的樣例：

bin/logstash -f logstash-apache.conf

你可以看到apache的日誌數據已經導入到ES中了。這裏logstash會依照你的配置讀取，處理指定的文件。不論什麼後加入到文件的內容也會被捕獲處理最後保存到ES中。

此外，數據中type的字段值會被替換成"apache_access"(這個功能在配置中已經指定)。

這個配置僅僅是讓Logstash監控了apache access_log，但是在實際中每每並不夠用可能還需要監控error_log。僅僅要在上面的配置中改變一行既可以實現。例如如下：

input {
  file {
    path => "/tmp/*_log"
...

現在你可以看到logstash處理了error日誌和access日誌。

然而，假設你檢查了你的數據(或許用elasticsearch-kopf)，你將發現access_log日誌被分紅不一樣的字段，但是error_log確沒有這樣。這是因爲咱們使用了「grok」filter並只配置匹配combinedapachelog日誌格式，這樣知足條件的日誌就會本身主動的被切割成不一樣的字段。咱們可以經過控制日誌依照它本身的某種格式來解析日誌，不是很是好的嗎？對吧。

此外，你或許還會發現Logstash不會反覆處理文件裏已經處理過得events。因爲Logstash已經記錄了文件處理的位置。這樣就僅僅處理文件裏新增長的行數。美麗！

條件推斷

咱們利用上一個樣例來介紹一下條件推斷的概念。

這個概念普通狀況下應該被大多數的Logstash用戶熟悉掌握。

你可以像其它普通的編程語言同樣來使用if,else if和else語句。讓咱們把每個event依賴的日誌文件類型都標記出來(access_log,error_log其它以log結尾的日誌文件)。

input {
  file {
    path => "/tmp/*_log"
  }
}

filter {
  if [path] =~ "access" {
    mutate { replace => { type => "apache_access" } }
    grok {
      match => { "message" => "%{COMBINEDAPACHELOG}" }
    }
    date {
      match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]
    }
  } else if [path] =~ "error" {
    mutate { replace => { type => "apache_error" } }
  } else {
    mutate { replace => { type => "random_logs" } }
  }
}

output {
  elasticsearch { host => localhost }
  stdout { codec => rubydebug }
}

我想你已經注意到了，咱們使用"type"字段來標記每個event,但是咱們實際上沒有解析"error"和」random"類型的日誌... 而實際狀況下可能會有很是多很是多類型的錯誤日誌，怎樣解析就做爲練習留給各位讀者吧。你可以依賴已經存在的日誌。

Syslog

Ok,現在咱們繼續瞭解一個很是有用的樣例：syslog。Syslog對於Logstash是一個很是長用的配置，並且它有很是好的表現(協議格式符合RFC3164)。Syslog其實是UNIX的一個網絡日誌標準，由client發送日誌數據到本地文件或者日誌server。在這個樣例中，你根本不用創建syslog實例;咱們經過命令行就可以實現一個syslog服務，經過這個樣例你將會看到發生什麼。

首先，讓咱們建立一個簡單的配置文件來實現logstash+syslog。文件名稱是 logstash-syslog.conf

input {
  tcp {
    port => 5000
    type => syslog
  }
  udp {
    port => 5000
    type => syslog
  }
}

filter {
  if [type] == "syslog" {
    grok {
      match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}" }
      add_field => [ "received_at", "%{@timestamp}" ]
      add_field => [ "received_from", "%{host}" ]
    }
    syslog_pri { }
    date {
      match => [ "syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ]
    }
  }
}

output {
  elasticsearch { host => localhost }
  stdout { codec => rubydebug }
}

運行logstash：

bin/logstash -f logstash-syslog.conf

一般。需要一個client連接到Logstashserver上的5000port而後發送日誌數據。在這個簡單的演示中咱們簡單的使用telnet連接到logstashserver發送日誌數據(與以前樣例中咱們在命令行標準輸入狀態下發送日誌數據相似)。

首先咱們打開一個新的shell窗體，而後輸入如下的命令：

telnet localhost 5000

你可以複製粘貼如下的例子信息(固然也可以使用其它字符，只是這樣可能會被grok filter不能正確的解析)：

Dec 23 12:11:43 louis postfix/smtpd[31499]: connect from unknown[95.75.93.154]
Dec 23 14:42:56 louis named[16000]: client 199.48.164.7#64817: query (cache) 'amsterdamboothuren.com/MX/IN' denied
Dec 23 14:30:01 louis CRON[619]: (www-data) CMD (php /usr/share/cacti/site/poller.php >/dev/null 2>/var/log/cacti/poller-error.log)
Dec 22 18:28:06 louis rsyslogd: [origin software="rsyslogd" swVersion="4.2.0" x-pid="2253" x-info="http://www.rsyslog.com"] rsyslogd was HUPed, type 'lightweight'.

以後你可以在你以前執行Logstash的窗體中看到輸出結果。信息被處理和解析！

{
                 "message" => "Dec 23 14:30:01 louis CRON[619]: (www-data) CMD (php /usr/share/cacti/site/poller.php >/dev/null 2>/var/log/cacti/poller-error.log)",
              "@timestamp" => "2013-12-23T22:30:01.000Z",
                "@version" => "1",
                    "type" => "syslog",
                    "host" => "0:0:0:0:0:0:0:1:52617",
        "syslog_timestamp" => "Dec 23 14:30:01",
         "syslog_hostname" => "louis",
          "syslog_program" => "CRON",
              "syslog_pid" => "619",
          "syslog_message" => "(www-data) CMD (php /usr/share/cacti/site/poller.php >/dev/null 2>/var/log/cacti/poller-error.log)",
             "received_at" => "2013-12-23 22:49:22 UTC",
           "received_from" => "0:0:0:0:0:0:0:1:52617",
    "syslog_severity_code" => 5,
    "syslog_facility_code" => 1,
         "syslog_facility" => "user-level",
         "syslog_severity" => "notice"
}

恭喜全部。看到你已經成爲一個貼近格的Logstash的用戶。您將可以輕鬆配置。執行Logstash，還可以發送event給Logstash，但這個過程將有很是多的使用的值這個地方被挖。

相關標籤/搜索

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。