ELK技术栈—Logstash—Input插件
生活随笔
收集整理的這篇文章主要介紹了
ELK技术栈—Logstash—Input插件
小編覺得挺不錯的,現在分享給大家,幫大家做個參考.
原文作者：歸來朝歌
原文地址：logstash之Input插件
?
1、stdin標準輸入和stdout標準輸出
Logsrtash含有兩個非常重要的基礎插件，input與output；首先執行命令：
bin/logstash -e 'input { stdin { } } output { stdout { codec ? => rubydebug } }' ?程序啟動之后輸入：hello logstash
2、監控日志文件變化
Logstash 使用一個名叫?FileWatch?的 Ruby?Gem 庫來監聽文件變化。這個庫支持 glob 展開文件路徑，而且會記錄一個叫?.sincedb的數據庫文件來跟蹤被監聽的日志文件的當前讀取位置。所以，不要擔心 logstash 會漏過你的數據（數據量并不是特別大的前提下）
創建logstash_conf文件(名字自定義)：monitor_file.conf
input{file{path => "/home/angel/servers/logstash-5.5.2/monitor_log/ "type => "log"start_position => "beginning"} } output{stdout{codec=>rubydebug} }啟動之后，logstash就會監控 /home/angel/servers/logstash-5.5.2/monitor_log/logs.txt文件，如果發現文件中有內容，則以標準輸出的形式輸出到控制臺；
檢測logstash 配置文件
bin/logstash -f /home/angel/servers/logstash-5.5.2/logstash_conf/monitor_file.conf -t啟動logstash：
bin/logstash -f /home/angel/servers/logstash-5.5.2/logstash_conf/monitor_file.conf測試：
echo "hello logstash" >> /home/angel/servers/logstash-5.5.2/monitor_log/logs.txt ?配置參數說明：
- Path=>表示監控的文件路徑
- Type=>給類型打標記，用來區分不同的文件類型。
- Start_postion=>從哪里開始記錄文件，默認是從結尾開始標記，要是你從頭導入一個文件就把改成”beginning”.
- discover_interval=>多久去監聽path下是否有文件，默認是15s
- exclude=>排除什么文件
- close_older=>一個已經監聽中的文件，如果超過這個值的時間內沒有更新內容，就關閉監聽它的文件句柄。默認是3600秒，即一個小時。
- sincedb_path=>監控庫存放位置(默認的讀取文件信息記錄在哪個文件中)。默認在：/data/plugins/inputs/file。
- sincedb_write_interval=> logstash 每隔多久寫一次 sincedb 文件，默認是 15 秒。
- stat_interval=>logstash 每隔多久檢查一次被監聽文件狀態（是否有更新），默認是 1 秒。
3、tcp插件
TCP是一種網絡傳輸控制協議，很多公司的數據不一定是在本地的，而是在傳輸網絡的；這個時候使用TCP建立連接后，通信雙方就可以進行數據傳輸了；Logstash提供了TCP插件，TCP插件可以監控某個端口，當數據打入logstash監聽的端口隊列的時候，logstash就可以進行數據的采集;
| codec=> | 可選項 （編碼解碼） |
| data_timeout=> | 可選項 （超時時間，秒為單位。如果設置-1,則永不超時，默認是5） |
| host=> | 可選項 （主機地址，字符串類型，如"localhost"或者"192.168.0.1?） |
| mode=> | 可選項 （值是["server","client"]其中之一,默認是server） |
| port=> | 必填項 （遠程監聽的端口） |
| ssl_cacert=> | 可選項，ssl認證相關 |
| ssl_cert=> | ? |
| ssk_key=> | ? |
| ssl_enable=> | 是否開啟ssl認證 |
| tags=> | 可選項 用于增加一些標簽，這個標簽可能在后續的處理中起到標志的作用 |
| type=> | 可選項 標記事件類型,通過type判斷 |
vim monitor_tcp.conf
input{tcp {port => 9876mode => "server" ?//值是["server","client"]其中之一,默認是server ?ssl_enable => false} } output{stdout{} }啟動并且監聽到9876端口;
檢測logstash 配置文件
bin/logstash -f /home/angel/servers/logstash-5.5.2/logstash_conf/monitor_tcp.conf -t啟動logstash：
bin/logstash -f /home/angel/servers/logstash-5.5.2/logstash_conf/monitor_tcp.conf寫一段java的socket代碼向9876發送數據：
public static void main(String[] args) throws Exception{// 向服務器端發送請求，服務器IP地址和服務器監聽的端口號Socket client = new Socket("hadoop01", 9876); ?// 通過printWriter 來向服務器發送消息PrintWriter printWriter = new PrintWriter(client.getOutputStream());System.out.println("連接已建立...");for(int i=0;i<10;i++){// 發送消息printWriter.println("hello logstash , 這是第"+i+" 條消息"); ?printWriter.flush();} ? }?
4、syslog插件
syslog**機制負責記錄內核和應用程序產生的日志信息，管理員可以通過查看日志記錄，來掌握系統狀況**。默認系統已經安裝了rsyslog.直接啟動即可。
vim /etc/rsyslog.conf，增加如下代碼，將日志信息發送至6789端口：
*.* @@hadoop01:6789重啟日志：
/etc/init.d/rsyslog restartvim monitor_syslog.conf
input{tcp{port=> 6789type=> syslog}udp{port=> 6789type=> 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{stdout{codec=> rubydebug} }啟動：
bin/logstash -f /home/angel/servers/logstash-5.5.2/logstash_conf/monitor_syslog.conf通過telnet hadoop01 6789傳輸數據，【注意，在logstash中的grok是正則表達式，用來解析當前數據】
Dec 23 12:11:43 louis postfix/smtpd[31499]: connect from unknown[95.75.93.154] Jun 05 08:00:00 louis named[16000]: client 199.48.164.7#64817: query (cache) 'amsterdamboothuren.com/MX/IN' denied Jun 05 08:10:00 louis CRON[620]: (www-data) CMD (php /usr/share/cacti/site/poller.php >/dev/null 2>/var/log/cacti/poller-error.log) Jun 05 08:05:06 louis rsyslogd: [origin software="rsyslogd" swVersion="4.2.0" x-pid="2253" x-info="http://www.rsyslog.com"] rsyslogd was HUPed, type 'lightweight'. ?日志截圖
5、logstash的codec監控nginx日志
Codec 來自 Coder/decoder兩個單詞的首字母縮寫；Logstash 不只是一個input | filter | output 的數據流，而是一個input | decode | filter | encode | output?的數據流，codec 就是用來decode、encode 事件的。簡單說，就是在logstash讀入的時候，通過codec編碼解析日志為相應格式，從logstash輸出的時候，通過codec解碼成相應格式。如果當前的日志是codec的，可以避免使用grok，從而提高開發效率和性能
編輯nginx文件，設置log格式：
log_format json '{"@timestamp":"$time_iso8601",''"@version":"1",''"host":"$server_addr",''"client":"$remote_addr",''"size":$body_bytes_sent,''"responsetime":$request_time,''"domain":"$host",''"url":"$uri",''"status":"$status"}'; access_log logs/access_codec.log json;截圖：
Nginx生成kibana瀏覽器的日志：
server {listen 80;server_name hadoop01;location / {proxy_pass http://hadoop01:5601;proxy_set_header Upgrade $http_upgrade;proxy_set_header Connection 'upgrade';proxy_set_header Host $host;proxy_cache_bypass $http_upgrade;}}?
?
vim nginx_codec.conf
啟動nginx：
/usr/local/nginx/sbin啟動logstash：
bin/logstash -f /home/angel/logstash-5.5.2/logstash_conf/nginx_codec.conf6、合并多行數據（Multiline）
在處理日志時，除了訪問日志外，還要處理運行時日志，該日志大都用程序寫的，比如 log4j。運行時日志跟訪問日志最大的不同是，運行時日志是多行，也就是說，連續的多行才能表達一個意思。multiline 插件來說，有三個設置比較重要：negate、pattern 和 what
| negate | ? | 類型是boolean | ? | 默認false | 如果沒有匹配，那么否定正則表達式 |
| Pattern | ?? | 類型是string | ?? | 沒有默認值 | 要匹配的正則表達式 |
| What | previous或者next | 沒有默認值 | 如果正則表達式匹配了，那么該事件是屬于（屬于下一次觸發的事件）還是上一個（觸發的事件） |
【注意】使用mutiline之前，需要安裝插件：
bin/logstash-plugin install logstash-filter-multiline //如果報錯，請使用： bin/logstash-plugin install --version 3.0.4 logstash-filter-multiline//如果還是報錯，請下載logstash-filter-multiline-3.0.4.gem，然后安裝 bin/logstash-plugin install /home/angel/logstash-5.5.2/vendor/bundle/jruby/1.9/cache/logstash-filter-multiline-3.0.4.gem- 百度搜索：logstash-filter-multiline-3.0.4.gem
- 將gem文件放入這個理：/home/angel/logstash-5.5.2/vendor/bundle/jruby/1.9/cache
- bin/logstash-plugin install /home/angel/logstash-5.5.2/vendor/bundle/jruby/1.9/cache/logstash-filter-multiline-3.0.4.gem
vim multiline.conf
input {file{path=>"/home/angel/logstash-5.5.2/logs/mutil/mu_log.out"type=>"runtimelog"codec=> multiline {pattern => "^\["negate => truewhat => "previous"}start_position=>"beginning"sincedb_path=>"/home/angel/logstash-5.5.2/logs/mutil/sincedb-access"ignore_older=>0} } filter{multiline{pattern => "^\["negate => truewhat => "previous"}date {match => [ "timestamp", "yyyy-MM-dd-HH:mm:ss" ]locale => "cn"}grok {match=>["message","\[%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level}\] %{GREEDYDATA:msg}"]} } output{stdout{codec => rubydebug} }啟動
bin/logstash -f /home/angel/logstash-5.5.2/logstash_conf/multiline.conf向/home/angel/logstash-5.5.2/logs/mutil/mu_log.out文件中輸入測試日志：
echo "[16-04-12 03:40:01 DEBUG] model.MappingNode:- ['/store/shopclass'] matched over. [16-04-12 03:40:02 DEBUG] impl.JdbcEntityInserter:- from product_category product_category where product_category.PARENT_ID is null and product_category.STATUS = ? and product_category.DEALER_ID is null order by product_category.ORDERS asc [16-04-12 03:40:03 DEBUG] model.MappingNode:- ['/store/shopclass'] matched over. [16-04-12 03:40:04 DEBUG] model.MappingNode:- ['/store/shopclass'] matched over. [16-04-12 03:40:05 DEBUG] impl.JdbcEntityInserter:- from product_category product_category where product_category.PARENT_ID is null and product_category.STATUS = ? and product_category.DEALER_ID is nullorder by product_category.ORDERS desc [16-04-12 03:40:06 DEBUG] impl.JdbcEntityInserter:- from product_category product_category where product_category.PARENT_ID is null and product_category.STATUS = ? and product_category.DEALER_ID is null order by product_category.ORDERS asc [16-04-12 03:40:07 DEBUG] model.MappingNode:- ['/store/shopclass'] matched over" >> /home/angel/logstash-5.5.2/logs/mutil/mu_log.out?
?
超強干貨來襲 云風專訪：近40年碼齡，通宵達旦的技術人生總結
以上是生活随笔為你收集整理的ELK技术栈—Logstash—Input插件的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： ELK技术栈—Kibana
- 下一篇： 并查集—岛屿数量