2.4　輸出插件

2.4.1　輸出到Elasticsearch

Logstash早期有三個不同的Elasticsearch插件。到1.4.0版本的時候，開發者徹底重寫了LogStash：：Outputs：：Elasticsearch插件。從此，我們只需要用這一個插件，就能任意切換使用Elasticsearch集群支持的各種不同協議了。

1.配置示例

output {
  elasticsearch {
    host =>“192.168.0.2”
    protocol =>“http”
    index =>“logstash-%{type}-%{+YYYY.MM.dd}”
    index_type =>“%{type}”
    workers => 5
    template_overwrite => true
  }
}

2.解釋

□ 索引名　寫入的Elasticsearch索引的名稱，這里可以使用變量。為了更貼合日志場景，Logstash提供了%{+YYYY.MM.dd}這種寫法。在語法解析的時候，看到以+號開頭的，就會自動認為后面是時間格式，嘗試用時間格式來解析后續字符串。所以，之前處理過程中不要給自定義字段取個加號開頭的名字……

此外，注意索引名中不能有大寫字母，否則Elasticsearch在日志中會報InvalidIndex-NameException，但是Logstash不會報錯，這個錯誤比較隱晦，也容易掉進這個坑中。

□ 協議　現在，新插件支持三種協議：node、http和transport。

一個小集群里，使用node協議最方便了。Logstash以Elasticsearch的client節點身份（即不存數據不參加選舉）運行。如果你運行下面這行命令，你就可以看到自己的Logstash進程名，對應的node.role值是c：

# curl 127.0.0.1:9200/_cat/nodes?v
host       ip      heap.percent ram.percent load node.role master name
local 192.168.0.102  7      c         -      logstash-local-1036-2012
local 192.168.0.2    7      d         *      Sunstreak

特別的，作為一個快速運行示例的需要，你還可以在Logstash進程內部運行一個內嵌的Elasticsearch服務器。內嵌服務器默認會在$PWD/data目錄里存儲索引。如果你想變更這些配置，在$PWD/elasticsearch.yml文件里寫自定義配置即可，Logstash會嘗試自動加載這個文件。

對于擁有很多索引的大集群，你可以用transport協議。Logstash進程會轉發所有數據到你指定的某臺主機上。這種協議跟上面的node協議是不同的。node協議下的進程是可以接收到整個Elasticsearch集群狀態信息的，當進程收到一個事件時，它就知道這個事件應該存在集群內哪個機器的分片里，所以它就會直接連接該機器發送這條數據。而transport協議下的進程不會保存這個信息，在集群狀態更新（節點變化，索引變化都會發送全量更新）時，就不會對所有的Logstash進程也發送這種信息。更多Elasticsearch集群狀態的細節，參閱http://www.elasticsearch.org/guide。

如果你已經有現成的Elasticsearch集群，但是版本跟Logstash自帶的又不太一樣，建議你使用http協議。Logstash會使用POST方式發送數據。

3.protocol選擇建議

Logstash 1.4.2在transport和http協議的情況下是固定連接指定host發送數據。從1.5.0開始，host可以設置數組，它會從節點列表中選取不同的節點發送數據，達到Round-Robin負載均衡的效果。

Kibana 4強制要求Elasticsearch全集群所有node版本在1.4以上，所以采用node方式發送數據的Logstash 1.4版（攜帶的Elasticsearch.jar庫是1.1.1版本）會導致Kibana 4無法運行，采用Kibana 4的讀者務必改用http方式。

開發者在IRC freenode#logstash頻道里表示：“高于1.0版本的Elasticsearch應該都能跟最新版Logstash的node協議一起正常工作”。此信息僅供參考，請認真測試后再上線。

4.性能問題

Logstash 1.4.2在http協議下默認使用作者自己的ftw庫，隨同分發的是0.0.39版。該版本有內存泄露問題，長期運行下輸出性能越來越差！

解決辦法：對性能要求不高的，可以在啟動Logstash進程時，配置環境變量ENV["BULK"]，強制采用Elasticsearch官方Ruby庫。命令如下：

export BULK=“esruby”

對性能要求高的，可以嘗試升級到Logstash 1.5版。新版的logstash-output-elasticsearch放棄了ftw庫，改用了一個JRuby平臺專有的Manticore庫。根據測試，性能跟ftw比相當接近。

對性能要求極高的，可以手動更新ftw庫版本，目前最新版是0.0.42版，據稱內存問題在0.0.40版即解決。

5.模板

Elasticsearch支持給索引預定義設置和mapping（前提是你用的Elasticsearch版本支持這個API，不過估計應該都支持）。Logstash自帶有一個優化好的模板，內容如下：

{“template” : “logstash-*”,“settings” : {“index.refresh_interval” : “5s”
  },“mappings” : {“_default_” : {“_all” : {“enabled” : true},“dynamic_templates” : [ {“string_fields” : {“match” : “*”,“match_mapping_type” : “string”,“mapping” : {“type” : “string”, “index” : “analyzed”, “omit_norms” : true,“fields” : {“raw” : {“type”: “string”, “index” : “not_analyzed”, “ignore_above” : 256}
               }
           }
         }
       } ],“properties” : {“@version”: { “type”: “string”, “index”: “not_analyzed” },“geoip”  : {“type” : “object”,“dynamic”: true,“path”: “full”,“properties” : {“location” : { “type” : “geo_point” }
             }
         }
       }
    }
  }
}

這其中的關鍵設置包括：

□ template for index-pattern：只有匹配logstash-*的索引才會應用這個模板。有時候我們會變更Logstash的默認索引名稱，記住你也得通過PUT方法上傳可以匹配你自定義索引名的模板。當然，我更建議的做法是，把你自定義的名字放在“logstash-”后面，變成index=>"logstash-custom-%{+yyyy.MM.dd}"這樣。

□ refresh_interval for indexing：Elasticsearch是一個近實時搜索引擎。它實際上是每1秒鐘刷新一次數據。對于日志分析應用，我們用不著這么實時，所以Logstash自帶的模板修改成了5秒鐘。你還可以根據需要繼續放大這個刷新間隔以提高數據寫入性能。

□ multi-field with not_analyzed：Elasticsearch會自動使用自己的默認分詞器（空格，點，斜線等分割）來分析字段。分詞器對于搜索和評分是非常重要的，但是大大降低了索引寫入和聚合請求的性能。所以Logstash模板定義了一種叫“多字段”（multi-field）類型的字段。這種類型會自動添加一個“.raw”結尾的字段，并給這個字段設置為不啟用分詞器。簡單說，你想獲取url字段的聚合結果的時候，不要直接用“url”，而是用"url.raw"作為字段名。

□ geo_point：Elasticsearch支持geo_point類型，geo distance聚合等等。比如說，你可以請求某個geo_point點方圓10千米內數據點的總數。在Kibana的bettermap類型面板里，就會用到這個類型的數據。

6.其他模板配置建議

□ doc_values：doc_values是Elasticsearch 1.0版本引入的新特性。啟用該特性的字段，索引寫入的時候會在磁盤上構建fielddata。而過去，fielddata是固定只能使用內存的。在請求范圍加大的時候，很容易觸發OOM或者circuit breaker報錯：

ElasticsearchException[org.elasticsearch.common.breaker.CircuitBreakingException: Data too large, data for field [@timestamp] would be larger than limit of [639015321/609.4mb]]

doc_values只能給不分詞（對于字符串字段就是設置了"index"："not_analyzed"，數值和時間字段默認就沒有分詞）的字段配置生效。

doc_values雖然用的是磁盤，但是系統本身也有自帶VFS的cache效果并不會太差。據官方測試，經過1.4的優化后，只比使用內存的fielddata慢15%。所以，在數據量較大的情況下，強烈建議開啟該配置：

{“template” : “logstash-*”,“settings” : {“index.refresh_interval” : “5s”
  },“mappings” : {“_default_” : {“_all” : {“enabled” : true},“dynamic_templates” : [ {“string_fields” : {“match” : “*”,“match_mapping_type” : “string”,“mapping” : {“type” : “string”, “index” : “analyzed”, “omit_norms” : true,“fields” : {“raw” : { “type”: “string”, “index” : “not_analyzed”, “ignore_above” : 256,
“doc_values”: true }
               }
           }
         }
       } ],“properties” : {“@version”: { “type”: “string”, “index”: “not_analyzed” },“@timestamp”: { “type”: “date”, “index”: “not_analyzed”, “doc_values”: true,
“format”: “dateOptionalTime” },“geoip”  : {“type” : “object”,“dynamic”: true,“path”: “full”,“properties” : {“location” : { “type” : “geo_point” }
             }
         }
       }
    }
  }
}

□ order：如果你有自己單獨定制template的想法，很好。這時候有幾種選擇：

■ 在logstash-output-elasticsearch配置中開啟manage_template=>false選項，然后一切自己動手；

■ 在logstash-output-elasticsearch配置中開啟template=>"/path/to/your/tmpl.json"選項，讓logstash來發送你自己寫的template文件；

■ 避免變更Logstash里的配置，而是另外發送一個template，利用Elasticsearch的templates order功能。

這個order功能，就是Elasticsearch在創建一個索引的時候，如果發現這個索引同時匹配上了多個template，那么就會先應用order數值小的template設置，然后再應用一遍order數值高的作為覆蓋，最終達到一個merge的效果。

比如，對上面這個模板已經很滿意，只想修改一下refresh_interval，那么只需要新寫一個：

{“order” : 1,“template” : “logstash-*”,“settings” : {“index.refresh_interval” : “20s”
    }
}

然后運行curl-XPUT http://localhost：9200/_template/template_newid-d'@/path/to/your/tmpl.json'即可。

Logstash默認的模板，order是0，id是logstash，通過logstash-output-elasticsearch的配置選項template_name修改。你的新模板就不要跟這個名字沖突了。

2.4.2　發送email

配置示例如下：

output {
    email {
        to =>“admin@website.com,root@website.com”
        cc =>“other@website.com”
        via =>“smtp”
        subject =>“Warning: %{title}”
        options => {
            smtpIporHost       =>“localhost”,
            port               => 25,
            domain             => 'localhost.localdomain',
            userName           => nil,
            password           => nil,
            authenticationType => nil, # （plain, login and cram_md5）
            starttls           => true
        }
        htmlbody =>“”
        body =>“”
        attachments => [“/path/to/filename”]
    }
}

logstash-output-email插件支持SMTP協議和sendmail兩種方式，通過via參數設置。SMTP方式有較多的options參數可配置。sendmail只能利用本機上的sendmail服務來完成——文檔上描述了Mail庫支持的sendmail配置參數，但實際代碼中沒有相關處理，不要被迷惑了。

2.4.3　調用系統命令執行

logstash-output-exec插件的運用也非常簡單，如下所示，將Logstash切割成的內容作為參數傳遞給命令。這樣，在每個事件到達該插件的時候，都會觸發這個命令的執行。

output {
    exec {
        command =>“sendsms.pl \”%{message}\“ -t %{user}”
    }
}

需要注意的是。這種方式是每次都重新開始執行一次命令并退出。本身是比較慢速的處理方式（程序加載，網絡建聯等都有一定的時間消耗）。最好只用于少量的信息處理場景，比如不適用nagios的其他報警方式。示例就是通過短信發送消息。

2.4.4　保存成文件

通過日志收集系統將分散在數百臺服務器上的數據集中存儲在某中心服務器上，這是運維最原始的需求。早年的scribed，甚至直接就把輸出的語法命名為<store>。Logstash當然也能做到這點。

和LogStash：：Inputs：：File不同，LogStash：：Outputs：：File里可以使用sprintf format格式來自動定義輸出到帶日期命名的路徑。

配置示例如下：

output {
    file {
        path =>“/path/to/%{+yyyy/MM/dd/HH}/%{host}.log.gz”
        message_format =>“%{message}”
        gzip => true
    }
}

使用output/file插件首先需要注意的就是message_format參數。插件默認是輸出整個event的JSON形式數據的。這可能跟大多數情況下使用者的期望不符。大家可能只是希望按照日志的原始格式保存就好了。所以需要定義為%{message}，當然，前提是在之前的filter插件中，你沒有使用remove_field或者update等參數刪除或修改%{message}字段的內容。

另一個非常有用的參數是gzip。gzip格式是一個非常奇特而友好的格式。其格式包括有：

□ 10字節的頭，包含幻數、版本號以及時間戳。

□ 可選的擴展頭，如原文件名。

□ 文件體，包括DEFLATE壓縮的數據。

□ 8字節的尾注，包括CRC-32校驗和以及未壓縮的原始數據長度。

這樣gzip就可以一段一段的識別出來數據——反過來說，也就是可以一段一段壓縮了添加在后面！

這對于我們流式添加數據簡直太棒了！

2.4.5　報警發送到Nagios

Logstash中有兩個output插件是與Nagios有關的。logstash-output-nagios插件發送數據給本機的nagios.cmd管道命令文件，logstash-output-nagios_nsca插件則是調用send_nsca命令以NSCA協議格式把數據發送給Nagios服務器（遠端或者本地皆可）。

1.nagios.cmd

nagios.cmd是Nagios服務器的核心組件。Nagios事件處理和內外交互都是通過這個管道文件來完成的。

使用CMD方式，需要自己保證發送的Logstash事件符合Nagios事件的格式。即必須在filter階段預先準備好nagios_host和nagios_service字段；此外，如果在filter階段也準備好nagios_annotation和nagios_level字段，這里也會自動轉換成nagios事件信息。

filter {
    if [message] =~ /err/ {
        mutate {
            add_tag =>“nagios”
            rename => [“host”, “nagios_host”]
            replace => [“nagios_service”, “logstash_check_%{type}”]
        }
    }
}
output {
    if “nagios” in [tags] {
        nagios { }
    }
}

如果不打算在filter階段提供nagios_level，那么也可以在該插件中通過參數配置。

所謂nagios_level，即我們通過nagiosplugin檢查數據時的返回值。其取值范圍和含義如下：

□ “0”，代表“OK”，服務正常。

□ “1”，代表“WARNNING”，服務警告，一般nagios plugin命令中使用-w參數設置該閾值。

□ “2”，代表“CRITICAL”，服務危急，一般nagios plugin命令中使用-c參數設置該閾值。

□ “3”，代表“UNKNOWN”，未知狀態，一般會在timeout等情況下出現。

默認情況下，該插件會以“CRITICAL”等級發送報警給Nagios服務器。

nagios.cmd文件的具體位置，可以使用command_file參數設置。默認位置是“/var/lib/nagios3/rw/nagios.cmd”。

關于和nagios.cmd交互的具體協議說明，有興趣的讀者請閱讀Using external commands in Nagios一文（http://archive09.linux.com/feature/153285），這是《Learning Nagios 3.0》書中內容節選。

2.NSCA

NSCA是一種標準的Nagios分布式擴展協議。分布在各機器上的send_nsca進程主動將監控數據推送給遠端Nagios服務器的NSCA進程。

當Logstash跟Nagios服務器沒有在同一個主機上運行的時候，就只能通過NSCA方式來發送報警了——當然也必須在Logstash服務器上安裝send_nsca命令。

Nagios事件所需要的幾個屬性在上一段中已經有過描述。不過在使用這個插件的時候，不要求提前準備好，而是可以在該插件內部定義參數：

output {
    nagios_nsca {
        nagios_host =>“%{host}”
        nagios_service =>“logstash_check_%{type}”
        nagios_status =>“2”
        message_format =>“%{@timestamp}: %{message}”
        host =>“nagiosserver.domain.com”
    }
}

這里請注意，host和nagios_host兩個參數，分別是用來設置Nagios服務器的地址，和報警信息中有問題的服務器地址。

關于NSCA原理，架構和配置說明，還不了解的讀者請閱讀官方網站Using NSClient++from nagios with NSCA一節（http://nsclient.org/nscp/wiki/doc/usage/nagios/nsca）。

3.其他類似插件

除了Nagios以外，Logstash同樣可以發送信息給其他常見監控系統，方式和Nagios大同小異：

□ logstash-output-ganglia插件通過UDP協議，發送gmetric型數據給本機/遠端的gmond或者gmetad。

□ logstash-output-zabbix插件調用本機的zabbix_sender命令發送。

2.4.6　statsd

statsd最早是2008年Flickr公司用Perl寫的針對graphite、datadog等監控數據后端存儲開發的前端網絡應用，2011年Etsy公司用Nodejs重構。用于接收、寫入、讀取和聚合時間序列數據，包括即時值和累積值等。

配置示例如下：

output {
    statsd {
        host =>“statsdserver.domain.com”
        namespace =>“logstash”
        sender =>“%{host}”
        increment => [“httpd.response.%{status}”]
    }
}

Graphite以樹狀結構存儲監控數據，所以statsd也是如此。所以發送給statsd的數據的key也一定得是“first.second.tree.four”這樣的形式。而在logstash-output-statsd插件中，就會以三個配置參數來拼接成這種形式：

namespace.sender.metric

其中namespace和sender都是直接設置的，而metric又分為好幾個不同的參數可以分別設置。statsd支持的metric類型如下：

□ increment

示例語法：increment=>["nginx.status.%{status}"]。該配置即可在statsd中生成對應的nginx.status.200，nginx.status.206，nginx.status.304，nginx.status.404，nginx.status.502，nginx.status.503，nginx.status.504等一系列監控項。同時，在statsd內配置的一個時間周期內，各狀態碼的次數，會自動累加到各自監控項里。

□ decrement

語法同increment。不過是遞減而不是遞增。

□ count

示例語法：count=>{"nginx.bytes"=>"%{bytes}"}。該配置可以在statsd中生成一個nginx.bytes監控項。而每條Nginx訪問記錄的響應字節數，累加成帶寬。

□ gauge

語法同count。gauge和count（在rrdtool中叫counter）兩種計數器的區別從早年的rrdtool時代就被反復強調，即gauge直接存儲原始數值，不累加。

□ set

示例語法：set=>{"online"=>"%{user_id}"}。是新版statsd才支持的功能，可以用來做去重計算，比如在線人數統計。

□ timing

示例語法：timing=>["nginx.requesttime"=>"%{request_time}"]。該配置可以在statsd中生成一個nginx.requesttime監控項，記錄時間周期內，Nginx訪問記錄的響應時間的數值統計情況，包括平均值，最大值，最小值，標準差，百分比值等。

關于這些metric類型的詳細說明，請閱讀statsd文檔：https://github.com/etsy/statsd/blob/master/docs/metric_types.md。

推薦閱讀

Etsy發布nodejs版本statsd的博客：Measure Anything，Measure Everything（http://codeascraft.etsy.com/2011/02/15/measure-anything-measure-everything/）

Flickr發布statsd的博客：Counting&Timing（http://code.flickr.net/2008/10/27/counting-timing/）

Librato有關statsd協作的博客：Using StatsD with Librato（http://support.metrics.librato.com/knowledgebase/articles/77199-using-statsd-with-librato）

2.4.7　標準輸出stdout

和之前logstash-input-stdin插件一樣，logstash-output-stdout插件也是最基礎和簡單的輸出插件。同樣在這里簡單介紹一下，作為輸出插件的一個共性了解。

配置示例如下：

output {
    stdout {
        codec => rubydebug
        workers => 2
    }
}

輸出插件統一具有一個參數是workers。Logstash為輸出做了多線程的準備。

其次是codec設置。codec的作用在之前已經講過。可能除了logstash-codec-multiline，其他codec插件本身并沒有太多的設置項。所以一般省略掉后面的配置區段。換句話說。上面配置示例的完全寫法應該是：

output {
    stdout {
        codec => rubydebug {
        }
        workers => 2
    }
}

單就logstash-output-stdout插件來說，其最重要和常見的用途就是調試。所以在不太有效的時候，加上命令行參數-vv運行，查看更多詳細調試信息。

2.4.8　TCP發送數據

雖然之前我們已經提到過不建議直接使用LogStash：：Inputs：：TCP和LogStash：：Outputs：：TCP做轉發工作，不過在實際交流中，發現確實有不少朋友覺得這種簡單配置足夠使用，因而不愿意多加一層消息隊列的。所以，還是把Logstash如何直接發送TCP數據也稍微提點一下。

配置示例如下：

output {
    tcp {
        host  =>“192.168.0.2”
        port  => 8888
        codec => json_lines
    }
}

在收集端采用TCP方式發送給遠端的TCP端口。這里需要注意的是，默認的Codec選項是json。而遠端的LogStash：：Inputs：：TCP的默認Codec選項卻是plain！所以不指定各自的Codec，對接肯定是失敗的。

另外，由于IO BUFFER的原因，即使是兩端共同約定為json依然無法正常運行，接收端會認為一行數據沒結束，一直等待直至自己OutOfMemory！

所以，正確的做法是，發送端指定Codec為json_lines，這樣每條數據后面會加上一個回車，接收端指定Codec為json_lines或者json均可，這樣才能正常處理。包括在收集端已經切割好的字段，也可以直接帶入收集端使用了。

2.4.9　輸出到HDFS

數據寫入HDFS是很多日志收集系統的最終目的。不過Logstash偏巧不是其中之一。到目前為止，Logstash還沒有官方支持的直接寫入HDFS的插件。而在社區，則有兩種不同的解決方案可供選擇。下面分別介紹。

1.通過HTTP接口

插件源碼地址見：https://github.com/dstore-dbap/logstash-webhdfs

該插件使用Hadoop的WebHDFS接口，其本質就是發送POST數據，可以說實現起來比較簡單。未來logstash-plugins官方可能也會收這個插件。

配置示例如下：

output {
    hadoop_webhdfs {
        workers => 2
        server =>“your.nameno.de:14000”
        user =>“flume”
        path =>“/user/flume/logstash/dt=%{+Y}-%{+M}-%{+d}/logstash-%{+H}.log”
        f?lush_size => 500
        compress =>“snappy”
        idle_f?lush_time => 10
        retry_interval => 0.5
    }
}

插件使用方式，和其他自定義插件一樣，通過—pluginpath或者打包gem均可。

2.通過Java接口

插件源碼地址見：https://github.com/avishai-ish-shalom/logstash-hdfs

該插件使用Hadoop的HDFS接口，利用JRuby可以直接導入Java類的特性，直接使用了org.apache.hadoop.fs.FileSystem等類來實現。

配置示例如下：

output {
    hdfs {
        path =>“/path/to/output_file.log”
        enable_append => true
    }
}

因為需要導入各種Hadoop的jar包，所以這個運行比較麻煩。Logstash-1.4.2上的運行命令示例如下：

# LD_LIBRARY_PATH=“/usr/lib/hadoop/lib/native” GEM_HOME=./logstash-1.4.2/vendor/bundle/jruby/1.9 CLASSPATH=$（find ./logstash-1.4.2/vendor/jar -type f -name '*.jar'|tr '\n' ':‘）：$（find /usr/lib/hadoop-hdfs -type f -name '*.jar' | tr '\n' ':’）：$（find /usr/lib/hadoop -type f -name '*.jar' | tr '\n' ':‘）：/etc/hadoop/conf java org.jruby.Main -I./logstash-1.4.2/lib ./logstash-1.4.2/lib/logstash/runner.rb agent -f logstash.conf -p ./logstash

如果使用Logstash-1.5版本，可以通過rubygems.org直接安裝打包好的插件：

# bin/plugin install logstash-output-hdfs

然后這樣運行：

# LD_LIBRARY_PATH=“/usr/lib/hadoop/lib/native” CLASSPATH=$（find /usr/lib/hadoop-hdfs -type f -name '*.jar' | tr '\n' ':‘）：$（find /usr/lib/hadoop -type f -name '*.jar' | grep -v sources | tr '\n' ':’）：/etc/hadoop/conf $LOGSTASH_DIR/bin/logstash agent -f logstash.conf