docker安装logstash及logstash配置
一、logstash跟es有版本對照關系
了解對照關系，決定要安裝的logstash版本
二、ELK出現的原因
三、Logstash工作原理
Logstash事件處理管道有三個階段：輸入→過濾器→輸出，輸入生成事件，過濾器修改它們，然后輸出將它們發送到其他地方。輸入和輸出支持編解碼器，使你能夠在數據進入或離開管道時對其進行編碼或解碼，而無需使用單獨的過濾器。
- 輸入
你使用輸入將數據獲取到Logstash中，一些比較常用的輸入是：
| file | 從文件系統上的文件進行讀取，非常類似于UNIX命令tail -0F。 |
| syslog | 在眾所周知的端口514上監聽syslog消息并根據RFC3164格式進行解析。 |
| redis | 從redis服務器讀取數據，同時使用Redis通道和Redis列表，Redis通常被用作集中式Logstash安裝中的“broker”，它將從遠程Logstash “shipper”中的Logstash事件排隊。 |
| beats | 處理Beats發送的事件。 |
有關可用輸入的更多信息，請參見輸入插件。
- 過濾器
過濾器是Logstash管道中的中間處理設備，如果事件符合一定的條件，你可以將過濾器與條件語句組合在一起，對其執行操作，一些有用的過濾器包括：
| grok | 解析和構造任意文本，Grok是目前Logstash中解析非結構化日志數據到結構化和可查詢數據的最佳方式，使用內置的120種模式，你很可能會找到一個滿足你的需要！ |
| mutate | 對事件字段執行一般的轉換，你可以重命名、刪除、替換和修改事件中的字段。 |
| drop | 完全刪除事件，例如debug事件。 |
| clone | 復制事件，可能添加或刪除字段。 |
| geoip | 添加關于IP地址地理位置的信息（在Kibana中還顯示了令人驚嘆的圖表！） |
有關可用過濾器的更多信息，請參見過濾器插件。
- 輸出
輸出是Logstash管道的最后階段，事件可以通過多個輸出，但是一旦所有的輸出處理完成，事件就完成了它的執行，一些常用的輸出包括：
| elasticsearch | 發送事件數據到Elasticsearch，如果你打算以一種高效、方便、易于查詢的格式保存數據，那么使用Elasticsearch是可行的。 |
| file | 將事件數據寫入磁盤上的文件。 |
| graphite | 將事件數據發送到graphite，這是一種流行的用于存儲和繪制指標的開源工具。http://graphite.readthedocs.io/en/latest/ |
| statsd | 發送事件到statsd，“監聽統計信息（如計數器和計時器）、通過UDP發送聚合并將聚合發送到一個或多個可插拔后端服務”的服務，如果你已經在使用statsd，這可能對你很有用！ |
有關可用輸出的更多信息，請參見輸出插件。
- 編解碼器
Codecs是基本的流過濾器，可以作為輸入或輸出的一部分進行操作，Codecs使你能夠輕松地將消息的傳輸與序列化過程分開，流行的codecs包括json、msgpack和plain（text）。
| json | 以JSON格式對數據進行編碼或解碼。 |
| multiline | 將多行文本事件（如java異常和stacktrace消息）合并到單個事件中。 |
有關可用編解碼器的更多信息，請參見編解碼器插件。
- 執行模型
Logstash事件處理管道協調輸入、過濾器和輸出的執行。Logstash管道中的每個輸入階段都在自己的線程中運行，輸入將事件寫入位于內存（默認）或磁盤上的中央隊列，每個管道工作線程從這個隊列中取出一批事件，通過配置的過濾器運行事件批處理，然后通過任何輸出運行過濾的事件，可以配置批處理的大小和管道工作線程的數量（參見調優和分析Logstash性能）。默認情況下，Logstash使用內存有限隊列在管道階段之間（輸入→過濾器和過濾器→輸出）來緩沖事件，如果Logstash不安全的終止，則存儲在內存中的任何事件都將丟失。為了防止數據丟失，你可以啟用Logstash將運行中的事件持久化到磁盤上，有關更多信息，請參見持久隊列。
四、logstash的配置文件
你可以在Logstash設置文件logstash.yml中設置選項來控制Logstash執行，例如，你可以指定管道設置、配置文件的位置、日志記錄選項和其他設置。當你運行Logstash時，logstash.yml文件中的大多數設置都可以作為命令行標志使用，在命令行中設置的任何標志都會覆蓋logstash.yml文件中的相應設置。logstash.yml文件是用YAML編寫的，它的位置因平臺而異（參見Logstash目錄布局），你可以以層次結構形式指定設置或使用平面鍵，例如，要使用分層表單設置管道批處理大小和批延遲，你需要指定：
要表示與平面鍵相同的值，需要指定：
pipeline.batch.size: 125 pipeline.batch.delay: 50logstash.yml文件還支持bash風格的環境變量插值設置值。
pipeline:batch:size: ${BATCH_SIZE}delay: ${BATCH_DELAY:50} node:name: "node_${LS_NODE_NAME}" path:queue: "/tmp/${QUEUE_DIR:queue}"注意，${VAR_NAME:default_value}表示法是受支持的，在上面的示例中，它設置了一個默認的批延遲50和一個默認的path.queue為/tmp/queue的。
模塊也可以在logstash.yml文件中指定，模塊定義將具有這種格式：
modules:- name: MODULE_NAME1var.PLUGIN_TYPE1.PLUGIN_NAME1.KEY1: VALUEvar.PLUGIN_TYPE1.PLUGIN_NAME1.KEY2: VALUEvar.PLUGIN_TYPE2.PLUGIN_NAME2.KEY1: VALUEvar.PLUGIN_TYPE3.PLUGIN_NAME3.KEY1: VALUE- name: MODULE_NAME2var.PLUGIN_TYPE1.PLUGIN_NAME1.KEY1: VALUEvar.PLUGIN_TYPE1.PLUGIN_NAME1.KEY2: VALUE如果使用命令行標志--modules，則忽略在logstash.yml文件中定義的任何模塊。
logstash.yml文件包括以下設置，如果你使用的是X-Pack，請參閱Logstash中的X-Pack設置。
| node.name | 節點的描述性名稱 | 機器的主機名 |
| path.data | Logstash及其插件用于任何持久需求的目錄 | LOGSTASH_HOME/data |
| pipeline.id | 管道的ID | main |
| pipeline.workers | 將并行執行管道的過濾和輸出階段的工人數量，如果你發現事件正在備份，或者CPU沒有飽和，請考慮增加這個數字，以更好地利用機器處理能力 | 主機CPU核心的數量 |
| pipeline.batch.size | 在嘗試執行過濾器和輸出之前，單個工作線程將從輸入中收集的最大事件數，更大的批處理大小通常更高效，但代價是增加內存開銷，你可能需要增加jvm.options配置文件中的JVM堆空間，有關更多信息，請參閱Logstash配置文件 | 125 |
| pipeline.batch.delay | 當創建管道事件批處理時，在向管道工作人員發送一個較小的批處理之前，等待每個事件的時間為多少毫秒 | 50 |
| pipeline.unsafe_shutdown | 當設置為true時，即使內存中仍然存在游離事件，也會在關閉期間強制Logstash退出，默認情況下，Logstash將拒絕退出，直到所有接收到的事件都被推送到輸出，啟用此選項可能導致關閉期間的數據丟失 | false |
| path.config | 主管道的Logstash配置路徑，如果指定目錄或通配符，配置文件將按字母順序從目錄中讀取 | None |
| config.test_and_exit | 當設置為true時，檢查配置是否有效，然后退出，注意，在此設置中沒有檢查grok模式的正確性，Logstash可以從一個目錄中讀取多個配置文件，如果你把這個設置和log.level: debug結合起來，Logstash將對合并后的配置文件進行日志記錄，并用它來自的源文件注解每個配置塊 | false |
| config.reload.automatic | 當設置為true時，定期檢查配置是否已更改，并在更改配置時重新加載配置，這也可以通過SIGHUP信號手動觸發 | false |
| config.reload.interval | Logstash多久檢查一次配置文件以查看更改 | 3s |
| config.debug | 當設置為true時，將完整編譯的配置顯示為debug日志消息，你還必須設置log.level: debug，警告：日志消息將包含傳遞給插件配置的任意密碼選項，可能會導致明文密碼出現在日志中！ | false |
| config.support_escapes | 當設置為true時，引號中的字符串將處理以下轉義序列：\n變成文字換行符（ASCII 10），\r變成文字回車（ASCII 13），\t變成文字制表符（ASCII 9），\變成字面反斜杠\，"變成一個文字雙引號，'變成文字引號 | false |
| modules | 當配置時，modules必須位于上表中描述的嵌套YAML結構中 | None |
| queue.type | 用于事件緩沖的內部隊列模型，為基于內存中的遺留隊列指定memory，或者persisted基于磁盤的ACKed隊列（持久隊列） | memory |
| path.queue | 啟用持久隊列時存儲數據文件的目錄路徑（queue.type: persisted） | path.data/queue |
| queue.page_capacity | 啟用持久隊列時使用的頁面數據文件的大小（queue.type: persisted），隊列數據由分隔成頁面的僅追加的數據文件組成 | 64mb |
| queue.max_events | 啟用持久隊列時隊列中未讀事件的最大數量（queue.type: persisted） | 0（無限） |
| queue.max_bytes | 隊列的總容量（字節數），確保磁盤驅動器的容量大于這里指定的值，如果queue.max_events和queue.max_bytes都指定，Logstash使用最先達到的任何標準 | 1024mb（1g） |
| queue.checkpoint.acks | 當啟用持久隊列時，在強制執行檢查點之前的最大ACKed事件數（queue.type: persisted），指定queue.checkpoint.acks: 0設置此值為無限制 | 1024 |
| queue.checkpoint.writes | 啟用持久隊列時強制執行檢查點之前的最大寫入事件數（queue.type: persisted），指定queue.checkpoint.writes: 0設置此值為無限制 | 1024 |
| queue.drain | 啟用時，Logstash會一直等到持久隊列耗盡后才關閉 | false |
| dead_letter_queue.enable | 標記指示Logstash以插件支持的DLQ特性 | false |
| dead_letter_queue.max_bytes | 每個dead letter隊列的最大大小，如果條目將增加dead letter隊列的大小，超過此設置，則刪除條目 | 1024mb |
| path.dead_letter_queue | 存儲dead letter隊列數據文件的目錄路徑 | path.data/dead_letter_queue |
| http.host | 指標REST端點的綁定地址 | “127.0.0.1” |
| http.port | 指標REST端點的綁定端口 | 9600 |
| log.level | 日志級別，有效的選項是：fatal、error、warn、info、debug、trace | info |
| log.format | 日志格式，設置為json日志以JSON格式，或plain使用Object#.inspect | plain |
| path.logs | Logstash將其日志寫到的目錄 | LOGSTASH_HOME/logs |
| path.plugins | 哪里可以找到自定義插件，你可以多次指定此設置以包含多個路徑，插件應該在特定的目錄層次結構中：PATH/logstash/TYPE/NAME.rb，TYPE是inputs、filters、outputs或codecs，NAME是插件的名稱 | 特定于平臺的 |
五、logstash配置
下面的示例演示如何配置Logstash來過濾事件，處理Apache日志和syslog消息，并使用條件來控制哪些事件由過濾器或輸出處理。如果你需要幫助構建grok模式，請嘗試grok調試器，Grok調試器是基本許可證下的X-Pack特性，因此可以免費使用。
- 配置過濾器
過濾器是一種在線處理機制，它提供了根據需要對數據進行切片和切割的靈活性，讓我們看一下活動中的一些過濾器，下面的配置文件設置了grok和date過濾器。
input { stdin { } }filter {grok {match => { "message" => "%{COMBINEDAPACHELOG}" }}date {match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]} }output {elasticsearch { hosts => ["localhost:9200"] }stdout { codec => rubydebug } }使用此配置運行Logstash：
bin/logstash -f logstash-filter.conf
現在，將下面的行粘貼到你的終端并按Enter鍵，這樣它就會被stdin輸入處理：
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”
你應該會看到返回到stdout的是這樣的：
如你所見，Logstash（在grok過濾器的幫助下）能夠解析日志行（碰巧是Apache的“組合日志”格式），并將其分解為許多不同的離散信息，一旦開始查詢和分析日志數據，這就非常有用。例如，你將能夠輕松地在HTTP響應碼、IP地址、referrers等上運行報表。Logstash有很多現成的grok模式，因此如果你需要解析通用的日志格式，很可能已經有人為你完成了這項工作，有關更多信息，請參閱GitHub上Logstash grok模式的列表。本例中使用的另一個過濾器是date過濾器，這個過濾器會解析一個時間戳，并將其用作事件的時間戳（不管你什么時候使用日志數據）。你將注意到，本例中的@timestamp字段設置為2013年12月11日，盡管Logstash在隨后的某個時間攝取了該事件，這在備份日志時非常方便，它使你能夠告訴Logstash“使用此值作為此事件的時間戳”。
- 處理Apache日志
讓我們做一些有用的事情：處理apache訪問日志文件！我們將從本地主機上的文件中讀取輸入，并根據需要使用條件處理事件。首先，創建一個名為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 {hosts => ["localhost:9200"]}stdout { codec => rubydebug } }然后，使用以下日志條目（或使用你自己的webserver中的一些日志條目）創建上面配置的輸入文件（在本例中為“/tmp/access_log”）：
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標志運行Logstash以將其傳遞到配置文件：
bin/logstash -f logstash-apache.conf
現在你應該在Elasticsearch中看到你的apache日志數據了，Logstash打開并讀取指定的輸入文件，處理遇到的每個事件。記錄到此文件的任何追加行也將被捕獲，由Logstash作為事件處理，并存儲在Elasticsearch中。還有一個額外的好處，他們儲藏的字段“type”設置為“apache_access”（這是由輸入配置中的type ? "apache_access"行）。
在這個配置中，Logstash只查看apache access_log，但是通過更改上面配置中的一行就可以同時查看access_log和error_log（實際上是任何文件匹配*log）：
input {file {path => "/tmp/*_log" ...當你重新啟動Logstash時，它將同時處理error和access日志，但是，如果你檢查數據（可能使用elasticsearch-kopf)，你會看到access_log被分解為離散字段，而error_log則不是。這是因為我們使用了grok過濾器來匹配標準的組合apache日志格式，并自動將數據分割為單獨的字段。如果我們能根據它的格式來控制行是如何被解析的，不是很好嗎？嗯，我們可以…
注意，Logstash不會重新處理access_log文件中已經查看的事件，從文件中讀取數據時，Logstash保存其位置，并只在添加新行時處理它們。
- 使用條件
你可以使用條件來控制哪些事件由過濾器或輸出處理，例如，你可以根據出現在哪個文件（access_log、error_log以及以“log”結尾的其他隨機文件）中來標記每個事件。
input {file {#標簽type => "systemlog-localhost"#采集點path => "/tmp/*_log"#開始收集點start_position => "beginning"#掃描間隔時間，默認是1s，建議5sstat_interval => "5"} } 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 { hosts => ["localhost:9200"] }stdout { codec => rubydebug } }這個示例使用type字段標記所有事件，但實際上不解析error或random文件，有很多類型的錯誤日志，它們應該如何標記取決于你使用的日志。
類似地，你可以使用條件將事件定向到特定的輸出，例如，你可以：
- 警告nagios任何狀態為5xx的apache事件
- 將任何4xx狀態記錄到Elasticsearch
- 通過statsd記錄所有的狀態代碼
要告訴nagios任何具有5xx狀態碼的http事件，首先需要檢查type字段的值，如果是apache，那么你可以檢查status字段是否包含5xx錯誤，如果是，發送到nagios。如果不是5xx錯誤，檢查status字段是否包含4xx錯誤，如果是，發送到Elasticsearch。最后，將所有apache狀態碼發送到statsd，無論狀態字段包含什么：
output {if [type] == "apache" {if [status] =~ /^5\d\d/ {nagios { ... }} else if [status] =~ /^4\d\d/ {elasticsearch { ... }}statsd { increment => "apache.%{status}" }} }- 處理Syslog消息
Syslog是Logstash最常見的用例之一，而且它處理得非常好（只要日志行大致符合RFC3164），Syslog實際上是UNIX網絡日志記錄標準，它將消息從客戶端發送到本地文件，或通過rsyslog發送到集中式日志服務器。對于本例，你不需要一個功能正常的syslog實例；我們將從命令行中偽造它，這樣你就可以了解發生了什么。
首先，讓我們為Logstash + syslog創建一個簡單的配置文件，名為logstash-syslog.conf。
input {tcp {port => 5000type => syslog}udp {port => 5000type => 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}" ]}date {match => [ "syslog_timestamp", "MMM d HH:mm:ss", "MMM dd HH:mm:ss" ]}} }output {elasticsearch { hosts => ["localhost:9200"] }stdout { codec => rubydebug } }使用這個新配置運行Logstash：
bin/logstash -f logstash-syslog.conf通常，客戶端將連接到端口5000上的Logstash實例并發送消息，對于本例，我們將telnet到Logstash并輸入一條日志行（類似于前面在STDIN中輸入日志行），打開另一個shell窗口與Logstash syslog輸入進行交互并輸入以下命令：
telnet localhost 5000復制粘貼以下行作為示例（你可以自己嘗試一些，但是要記住，如果grok過濾器對你的數據不正確，它們可能無法解析）。
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'.現在，當你的原始shell處理和解析消息時，你應該會看到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" }六、安裝及例子
docker pull logstash:7.14.0- 編輯/mydata/logstash/config/logstash.yml如下
控制臺的輸出，說明在es中已經生成索引
關于如何docker安裝es，docker安裝kibana
參考文章
總結
以上是生活随笔為你收集整理的docker安装logstash及logstash配置的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 互联网日报 | 3月13日 星期六 |
- 下一篇： SpringBoot应用日志通过logs