從0到1學會logstash的玩法（ELK）

時間  2021-02-28
標籤 css html java node mysql linux ios nginx git github 欄目 日誌分析 简体版
原文   原文鏈接

本篇文章採用的採用的是logstash-7.7.0版本，主要從以下幾個方面介紹css

一、logstash是什麼，能夠用來幹啥html

二、logstash的基本原理是什麼java

三、怎麼去玩這個elk的組件logstashnode

1、logstash是什麼，有哪些做用

1.一、概念

官方概念：Logstash是免費且開放的服務器端數據處理管道，可以從多個來源採集數據，轉換數據，而後將數據發送到您最喜歡的「存儲庫」中。mysql

1.二、功能

Logstash可以動態地採集、轉換和傳輸數據，不受格式或複雜度的影響。利用Grok從非結構化數據中派生出結構，從IP地址解碼出地理座標，匿名化或排除敏感字段，並簡化總體處理過程。linux

採集數據：

　　可以採集各類樣式、大小和來源的數據每每以各類各樣的形式，好比log日誌，收集redis、kafka等熱門分佈式技術的數據，而且還能夠收集實現了java的JMS規範的消息中心的數據，或分散或集中地存在於不少系統中。Logstash支持各類輸入選擇，能夠同時從衆多經常使用來源捕捉事件。可以以連續的流式傳輸方式，輕鬆地從日誌、指標、Web應用、數據存儲以及各類AWS服務採集數據ios

 

 

 

解析數據和轉換：

　　數據從源傳輸到存儲庫的過程當中，Logstash過濾器可以解析各個事件，識別已命名的字段以構建結構，並將它們轉換成通用格式，以便進行更強大的分析和實現商業價值。
Logstash可以動態地轉換和解析數據，不受格式或複雜度的影響：nginx

  • 利用Grok從非結構化數據中解析出結構數據
  • 從IP地址破譯出地理座標
  • 將PII數據匿名化，徹底排除敏感字段
  • 簡化總體處理，不受數據源、格式或架構的影響

　　數據輸入端從各類數據源收集到的數據可能會有不少不是咱們想要的，這時咱們能夠給Logstash定義過濾器，過濾器能夠定義多個，它們依次執行，最終把咱們想要的數據過濾出來，而後把這些數據解析成目標數據庫，如elasticsearch等能支持的數據格式存儲數據。git

 

數據轉存：

　　選擇好存儲庫，導出數據到存儲庫進行存儲，儘管Elasticsearch是咱們的首選輸出方向，可以爲咱們的搜索和分析帶來無限可能，但它並不是惟一選擇。Logstash提供衆多輸出選擇，能夠將數據發送到您要指定的地方，好比redis、kafka等github

 

 

 

 

2、logstash的基本原理

　　logstash分爲三個步驟：inputs（必須的）→ filters（可選的）→ outputs（必須的），inputs生成時間，filters對其事件進行過濾和處理，outputs輸出到輸出端或者決定其存儲在哪些組件裏。inputs和outputs支持編碼和解碼。

 

執行模型：

Logstash是協調inputs、filters和outputs執行事件處理的管道。

　　Logstash管道中的每一個input階段都在本身的線程中運行。將寫事件輸入到內存（默認）或磁盤上的中心隊列。每一個管道工做線程從該隊列中取出一批事件，經過配置的filter處理該批事件，而後經過output輸出到指定的組件存儲。管道處理數據量的大小和管道工做線程的數量是可配置的

　　默認狀況下，Logstash使用管道階段（input→filter和filter→output）之間的內存限制隊列來緩衝事件。若是Logstash不安全地終止，存儲在內存中的全部事件都將丟失。爲了幫助防止數據丟失，能夠啓用Logstash將飛行中的事件持久化到磁盤。有關詳細信息，請參閱持久隊列https://www.elastic.co/guide/en/logstash/current/persistent-queues.html

以下是目前logstash7.7.0支持的inputs、outputs、filters

inputs：

azure_event_hubs,beats,cloudwatch,couchdb_changes,dead_letter_queue,elasticsearch,exec,file,ganglia,gelf,generator,github,google_cloud_storage,google_pubsub,graphite,heartbeat,http,http_poller,imap,irc,java_generator,java_stdin,jdbc,jms,jmx,kafka,kinesis,log4j,lumberjack,meetup,pipe,puppet_facter,rabbitmq,redis,relp,rss,s3,s3-sns-sqs,salesforce,snmp,snmptrap,sqlite,sqs,stdin,stomp,syslog,tcp,twitter,udp,unix,varnishlog,websocket,wmi,xmpp

outputs：

boundary, circonus, cloudwatch, csv, datadog, datadog_metrics, elastic_app_search, elasticsearch, email, exec, file, ganglia, gelf, google_bigquery, google_cloud_storage, google_pubsub, graphite, graphtastic, http, influxdb, irc, sink, java_stdout, juggernaut, kafka, librato, loggly, lumberjack, metriccatcher, mongodb, nagios, nagios_nsca, opentsdb, pagerduty, pipe, rabbitmq, redis, redmine, riak, riemann, s3, sns, solr_http, sqs, statsd, stdout, stomp, syslog, tcp, timber, udp, webhdfs, websocket, xmpp, zabbix

filters：

aggregate, alter, bytes, cidr, cipher, clone, csv, date, de_dot, dissect, dns, drop, elapsed, elasticsearch, environment, extractnumbers, fingerprint, geoip, grok, http, i18n, java_uuid, jdbc_static, jdbc_streaming, json, json_encode, kv, memcached, metricize, metrics, mutate, prune, range, ruby, sleep, split, syslog_pri, threats_classifier, throttle, tld, translate, truncate, urldecode, useragent, uuid, xml

3、玩一玩logstash

3.一、壓縮包方式安裝

下載地址1：https://www.elastic.co/cn/downloads/logstash   

下載地址2：https://elasticsearch.cn/download/

這裏須要安裝jdk，我使用的是elasticsearch7.7.0自帶的jdk：

解壓即安裝：

tar -zxvf logstash-7.7.0.tar.gz

來個logstash版本的HelloWorld：

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

 

 

 

3.二、logstash配置文件

 logstash.yml：包含Logstash配置標誌。您能夠在此文件中設置標誌，而不是在命令行中傳遞標誌。在命令行上設置的任何標誌都會覆蓋logstash中的相應設置

 pipelines.yml：包含在單個Logstash實例中運行多個管道的框架和指令。

 jvm.options：包含JVM配置標誌。使用此文件設置總堆空間的初始值和最大值。您還可使用此文件爲Logsta設置語言環境

 log4j2.properties：包含log4j 2庫的默認設置

 start.options (Linux)：用於配置啓動服務腳本

logstash.yml文件詳解：

node.name  #默認主機名，該節點的描述名字
path.data  #LOGSTASH_HOME/data ，Logstash及其插件用於任何持久需求的目錄
pipeline.id #默認main，pipeline的id
pipeline.java_execution #默認true，使用java執行引擎
pipeline.workers #默認爲主機cpu的個數，表示並行執行管道的過濾和輸出階段的worker的數量
pipeline.batch.size #默認125 表示單個工做線程在嘗試執行過濾器和輸出以前從輸入中收集的最大事件數
pipeline.batch.delay #默認50 在建立管道事件時，在將一個小批分派給管道工做者以前，每一個事件須要等待多長時間(毫秒)
pipeline.unsafe_shutdown  #默認false，當設置爲true時，即便內存中仍有運行的事件，強制Logstash在關閉期間將會退出。默認狀況下，Logstash將拒絕退出，直到全部接收到的事件都被推入輸出。啓用此選項可能致使關機期間數據丟失
pipeline.ordered #默認auto，設置管道事件順序。true將強制對管道進行排序，若是有多個worker，則阻止logstash啓動。若是爲false，將禁用維持秩序所需的處理。訂單順序不會獲得保證，但能夠節省維護訂單的處理成本
path.config #默認LOGSTASH_HOME/config  管道的Logstash配置的路徑
config.test_and_exit #默認false，設置爲true時，檢查配置是否有效，而後退出。請注意，使用此設置不會檢查grok模式的正確性
config.reload.automatic #默認false，當設置爲true時，按期檢查配置是否已更改，並在更改時從新加載配置。這也能夠經過SIGHUP信號手動觸發
config.reload.interval  #默認3s ，檢查配置文件頻率
config.debug #默認false 當設置爲true時，將徹底編譯的配置顯示爲調試日誌消息
queue.type #默認memory ，用於事件緩衝的內部排隊模型。爲基於內存中的遺留隊列指定內存，或爲基於磁盤的脫機隊列(持久隊列)指定持久內存
path.queue #默認path.data/queue  ,在啓用持久隊列時存儲數據文件的目錄路徑
queue.page_capacity #默認64mb ，啓用持久隊列時(隊列)，使用的頁面數據文件的大小。隊列數據由分隔爲頁面的僅追加數據文件組成
queue.max_events #默認0，表示無限。啓用持久隊列時，隊列中未讀事件的最大數量
queue.max_bytes  #默認1024mb，隊列的總容量，以字節爲單位。確保磁盤驅動器的容量大於這裏指定的值
queue.checkpoint.acks #默認1024，當啓用持久隊列(隊列)時，在強制執行檢查點以前被隔離的事件的最大數量
queue.checkpoint.writes #默認1024，當啓用持久隊列(隊列)時，強制執行檢查點以前的最大寫入事件數
queue.checkpoint.retry #默認false，啓用後，對於任何失敗的檢查點寫，Logstash將對每一個嘗試的檢查點寫重試一次。任何後續錯誤都不會重試。而且不推薦使用，除非是在那些特定的環境中
queue.drain #默認false，啓用後，Logstash將等待，直到持久隊列耗盡，而後關閉
path.dead_letter_queue#默認path.data/dead_letter_queue，存儲dead-letter隊列的目錄
http.host #默認"127.0.0.1" 表示endpoint REST端點的綁定地址。
http.port #默認9600 表示endpoint REST端點的綁定端口。
log.level #默認info，日誌級別fatal，error，warn，info，debug，trace，
log.format #默認plain 日誌格式
path.logs  #默認LOGSTASH_HOME/logs 日誌目錄

3.三、keystore

keystore能夠保護一些敏感的信息，使用變量的方式替代，好比使用ES_PWD代替elasticsearch的密碼，能夠經過${ES_PWD}來獲取elasticsearch的密碼，這樣就是的密碼再也不是明文密碼。

./bin/logstash-keystore create   #建立一個keyword
 ./bin/logstash-keystore add ES_PWD #建立一個elastic的passwd，而後經過${ES_PWD}使用該密碼
 ./bin/logstash-keystore list  #查看已經設置好的鍵值對
 ./bin/logstash-keystore remove ES_PWD #刪除在keyword中的key

例如：

 

 

 3.四、logstash命令

注意：參數和logstash.yml配置文件對應（這裏不詳解，請查看3.2節）

-n, --node.name NAME          
-f, --path.config CONFIG_PATH 
-e, --config.string CONFIG_STRING
--field-reference-parser MODE 
--modules MODULES             
-M, --modules.variable MODULES_VARIABLE
--setup                      
--cloud.id CLOUD_ID                                            
--cloud.auth CLOUD_AUTH       
--pipeline.id ID              
-w, --pipeline.workers COUNT 
--pipeline.ordered ORDERED   
--java-execution                                               
--plugin-classloaders         
-b, --pipeline.batch.size SIZE 
-u, --pipeline.batch.delay DELAY_IN_MS 
--pipeline.unsafe_shutdown    
--path.data PATH              
-p, --path.plugins PATH       
-l, --path.logs PATH         
--log.level LEVEL             
--config.debug                
-i, --interactive SHELL      
-V, --version                 
-t, --config.test_and_exit    
-r, --config.reload.automatic 
--config.reload.interval 
--http.host HTTP_HOST         
--http.port HTTP_PORT         
--log.format FORMAT           
--path.settings SETTINGS_DIR

3.五、logstash配置文件的格式和value type

一、logstash配置文件的格式以下：

輸入，解析過濾，輸出，其中filter不是必須的，其餘兩個是必須的。

input {
  ...
}

filter {
  ...
}

output {
  ...
}

二、value types（logstash支持的數據類型）

array：數組能夠是單個或者多個字符串值。

users => [ {id => 1, name => bob}, {id => 2, name => jane} ]

Lists：集合

path => [ "/var/log/messages", "/var/log/*.log" ]
uris => [ "http://elastic.co", "http://example.net" ]

Boolean：true 或者false

ssl_enable => true

Bytes：字節類型

my_bytes => "1113"   # 1113 bytes
my_bytes => "10MiB"  # 10485760 bytes
my_bytes => "100kib" # 102400 bytes
my_bytes => "180 mb" # 180000000 bytes

Codec：編碼類型

codec => "json"

Hash：哈希（散列）

match => {
  "field1" => "value1"
  "field2" => "value2"
  ...
}
# or as a single line. No commas between entries:
match => { "field1" => "value1" "field2" => "value2" }

Number：數字類型

port => 33

Password：密碼類型

my_password => "password"

URI：uri類型

my_uri => "http://foo:bar@example.net"

Path： 路徑類型

my_path => "/tmp/logstash"

String：字符串類型，字符串必須是單個字符序列。注意，字符串值被括在雙引號或單引號中

3.六、logstash的權限配置

配置帳號，一個種是role，一種是user，配置方式有兩種，一種是經過elasticsearch的API配置，一種是經過kibana配置：

第一種方式：經過elasticsearch的API配置：

#添加一個logstash_writer的角色
POST _xpack/security/role/logstash_writer
{
  "cluster": ["manage_index_templates", "monitor", "manage_ilm"], 
  "indices": [
    {
      "names": [ "logstash-*" ],  #索引的模式匹配
      "privileges": ["write","create","delete","create_index","manage","manage_ilm"]   #權限內容
    }
  ]
}

#添加一個有logstash_writer角色權限的用戶：logstash_internal
POST _xpack/security/user/logstash_internal
{
  "password" : "x-pack-test-password",
  "roles" : [ "logstash_writer"],  #分配角色
  "full_name" : "Internal Logstash User"
}

#添加一個logstash_reader角色，只有read權限
POST _xpack/security/role/logstash_reader
{
  "indices": [
    {
      "names": [ "logstash-*" ], 
      "privileges": ["read","view_index_metadata"]
    }
  ]
}

#添加一個有logstash_reader角色權限的用戶：logstash_user
POST _xpack/security/user/logstash_user
{
  "password" : "x-pack-test-password",
  "roles" : [ "logstash_reader", "logstash_admin"], 
  "full_name" : "Kibana User for Logstash"
}

第二種：經過kibana的界面配置

Management > Roles
Management > Users

 

 

 權限選擇見elasticsearch官網：

https://www.elastic.co/guide/en/elasticsearch/reference/current/authorization.html

https://www.elastic.co/guide/en/elasticsearch/reference/current/security-privileges.html

 3.七、多管道配置

　　若是須要在同一個進程中運行多個管道，Logstash提供了一種經過名爲pipelines.yml的配置文件來實現此目的的方法。

例如：

- pipeline.id: my-pipeline_1
  path.config: "/etc/path/to/p1.config"
  pipeline.workers: 3
- pipeline.id: my-other-pipeline
  path.config: "/etc/different/path/p2.cfg"
  queue.type: persisted

　　該文件在YAML文件格式中，幷包含一個字典列表，其中每一個字典描述一個管道，每一個鍵/值對指定該管道的設置。該示例展現了經過id和配置路徑描述的兩個不一樣管道。對於第一個管道，爲pipeline.workers的值設置爲3，而在另外一箇中，持久隊列特性被啓用。未在pipelines.yml顯式設置的值。yml文件將使用到logstash中指定的默認值。
　　當啓動Logstash不帶參數時，它將讀取管道pipelines.yml。yml文件並實例化文件中指定的全部管道。另外一方面，當使用-e或-f時，Logstash會忽略管道。

注意：

  • 若是當前配置的事件流不共享相同的輸入/過濾器和輸出，而且使用標記和條件將它們彼此分離，這顯得多個管道尤爲重要
  • 在單個實例中擁有多個管道還容許這些事件流具備不一樣的性能和持久性參數(例如，pipeline.workers和persistent queues的不一樣設置)。這種分離意味着一條管道中的阻塞輸出不會對另外一條管道產生反壓力。
  • 考慮管道之間的資源競爭是很重要的，由於默認值是針對單個管道調優的。所以，例如，考慮減小每一個管道使用pipeline.worker的數量，由於每一個管道在默認狀況下每一個CPU核使用一個worker。
  • 每一個管道都隔離持久隊列和死信隊列，它們的位置命名空間爲pipeline.id的值

各管道之間的通訊原理：https://www.elastic.co/guide/en/logstash/current/pipeline-to-pipeline.html，有興趣的能夠了解下。

3.八、配置的從新加載

在咱們運行logstash的過程，不想停掉logstash進程，可是又想修改配置，就可使用到配置的從新加載了，有兩種方式。

第一種是在啓動的時候指定參數：

bin/logstash -f apache.config --config.reload.automatic

Logstash每3秒檢查一次配置更改。要更改此間隔，請使用--config.reload.interval <interval>選項，其中interval指定Logstash檢查配置文件更改的頻率（以秒爲單位）,請注意，必須使用單位限定符（s）

第二種就是強制加載配置文件

kill -SIGHUP pid  #pid爲logstash的pid

自動配置從新加載配置注意點：

  • 當Logstash檢測到配置文件中的更改時，它將經過中止全部輸入來中止當前管道，並嘗試建立使用更新後的配置的新管道。驗證新配置的語法後，Logstash驗證全部輸入和輸出均可以初始化（例如，全部必需的端口都已打開）。若是檢查成功，則Logstash會將現有管道與新管道交換。若是檢查失敗，舊管道將繼續運行，而且錯誤將傳播到控制檯。
  • 在自動從新加載配置期間，不會從新啓動JVM。管道的建立和交換都在同一過程當中進行。
  • 對grok模式文件的更改也將從新加載，但僅在配置文件中的更改觸發從新加載（或從新啓動管道）時。

3.九、經常使用filter介紹

一、grok

注：grok插件是一個十分耗費資源的插件

官網：https://www.elastic.co/guide/en/logstash/current/plugins-filters-grok.html

Grok是將非結構化日誌數據解析爲結構化和可查詢內容的好方法，很是適合syslog日誌，apache和其餘Web服務器日誌，mysql日誌等等

首先官方提供了120中的匹配模式（可是我一直都沒打開這個網址）：https://github.com/logstash-plugins/logstash-patterns-core/tree/master/patterns

還有一個用來驗證自定義的解析是否正確的一個網址：http://grokdebug.herokuapp.com/

既然我沒能打開官方提供的120個的模式匹配，從一片博客中找到了一部分的匹配模式（以下）：https://blog.csdn.net/cui929434/article/details/94390617

USERNAME [a-zA-Z0-9._-]+
USER %{USERNAME}
INT (?:[+-]?(?:[0-9]+))
BASE10NUM (?<![0-9.+-])(?>[+-]?(?:(?:[0-9]+(?:\.[0-9]+)?)|(?:\.[0-9]+)))
NUMBER (?:%{BASE10NUM})
BASE16NUM (?<![0-9A-Fa-f])(?:[+-]?(?:0x)?(?:[0-9A-Fa-f]+))
BASE16FLOAT \b(?<![0-9A-Fa-f.])(?:[+-]?(?:0x)?(?:(?:[0-9A-Fa-f]+(?:\.[0-9A-Fa-f]*)?)|(?:\.[0-9A-Fa-f]+)))\b

POSINT \b(?:[1-9][0-9]*)\b
NONNEGINT \b(?:[0-9]+)\b
WORD \b\w+\b
NOTSPACE \S+
SPACE \s*
DATA .*?
GREEDYDATA .*
QUOTEDSTRING (?>(?<!\\)(?>"(?>\\.|[^\\"]+)+"|""|(?>'(?>\\.|[^\\']+)+')|''|(?>`(?>\\.|[^\\`]+)+`)|``))
UUID [A-Fa-f0-9]{8}-(?:[A-Fa-f0-9]{4}-){3}[A-Fa-f0-9]{12}

# Networking
MAC (?:%{CISCOMAC}|%{WINDOWSMAC}|%{COMMONMAC})
CISCOMAC (?:(?:[A-Fa-f0-9]{4}\.){2}[A-Fa-f0-9]{4})
WINDOWSMAC (?:(?:[A-Fa-f0-9]{2}-){5}[A-Fa-f0-9]{2})
COMMONMAC (?:(?:[A-Fa-f0-9]{2}:){5}[A-Fa-f0-9]{2})
IPV6 ((([0-9A-Fa-f]{1,4}:){7}([0-9A-Fa-f]{1,4}|:))|(([0-9A-Fa-f]{1,4}:){6}(:[0-9A-Fa-f]{1,4}|((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3})|:))|(([0-9A-Fa-f]{1,4}:){5}(((:[0-9A-Fa-f]{1,4}){1,2})|:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3})|:))|(([0-9A-Fa-f]{1,4}:){4}(((:[0-9A-Fa-f]{1,4}){1,3})|((:[0-9A-Fa-f]{1,4})?:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:))|(([0-9A-Fa-f]{1,4}:){3}(((:[0-9A-Fa-f]{1,4}){1,4})|((:[0-9A-Fa-f]{1,4}){0,2}:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:))|(([0-9A-Fa-f]{1,4}:){2}(((:[0-9A-Fa-f]{1,4}){1,5})|((:[0-9A-Fa-f]{1,4}){0,3}:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:))|(([0-9A-Fa-f]{1,4}:){1}(((:[0-9A-Fa-f]{1,4}){1,6})|((:[0-9A-Fa-f]{1,4}){0,4}:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:))|(:(((:[0-9A-Fa-f]{1,4}){1,7})|((:[0-9A-Fa-f]{1,4}){0,5}:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:)))(%.+)?
IPV4 (?<![0-9])(?:(?:25[0-5]|2[0-4][0-9]|[0-1]?[0-9]{1,2})[.](?:25[0-5]|2[0-4][0-9]|[0-1]?[0-9]{1,2})[.](?:25[0-5]|2[0-4][0-9]|[0-1]?[0-9]{1,2})[.](?:25[0-5]|2[0-4][0-9]|[0-1]?[0-9]{1,2}))(?![0-9])
IP (?:%{IPV6}|%{IPV4})
HOSTNAME \b(?:[0-9A-Za-z][0-9A-Za-z-]{0,62})(?:\.(?:[0-9A-Za-z][0-9A-Za-z-]{0,62}))*(\.?|\b)
HOST %{HOSTNAME}
IPORHOST (?:%{HOSTNAME}|%{IP})
HOSTPORT %{IPORHOST}:%{POSINT}

# paths
PATH (?:%{UNIXPATH}|%{WINPATH})
UNIXPATH (?>/(?>[\w_%!$@:.,-]+|\\.)*)+
TTY (?:/dev/(pts|tty([pq])?)(\w+)?/?(?:[0-9]+))
WINPATH (?>[A-Za-z]+:|\\)(?:\\[^\\?*]*)+
URIPROTO [A-Za-z]+(\+[A-Za-z+]+)?
URIHOST %{IPORHOST}(?::%{POSINT:port})?
# uripath comes loosely from RFC1738, but mostly from what Firefox
# doesn't turn into %XX
URIPATH (?:/[A-Za-z0-9$.+!*'(){},~:;=@#%_\-]*)+
#URIPARAM \?(?:[A-Za-z0-9]+(?:=(?:[^&]*))?(?:&(?:[A-Za-z0-9]+(?:=(?:[^&]*))?)?)*)?
URIPARAM \?[A-Za-z0-9$.+!*'|(){},~@#%&/=:;_?\-\[\]]*
URIPATHPARAM %{URIPATH}(?:%{URIPARAM})?
URI %{URIPROTO}://(?:%{USER}(?::[^@]*)?@)?(?:%{URIHOST})?(?:%{URIPATHPARAM})?

# Months: January, Feb, 3, 03, 12, December
MONTH \b(?:Jan(?:uary)?|Feb(?:ruary)?|Mar(?:ch)?|Apr(?:il)?|May|Jun(?:e)?|Jul(?:y)?|Aug(?:ust)?|Sep(?:tember)?|Oct(?:ober)?|Nov(?:ember)?|Dec(?:ember)?)\b
MONTHNUM (?:0?[1-9]|1[0-2])
MONTHNUM2 (?:0[1-9]|1[0-2])
MONTHDAY (?:(?:0[1-9])|(?:[12][0-9])|(?:3[01])|[1-9])

# Days: Monday, Tue, Thu, etc...
DAY (?:Mon(?:day)?|Tue(?:sday)?|Wed(?:nesday)?|Thu(?:rsday)?|Fri(?:day)?|Sat(?:urday)?|Sun(?:day)?)

# Years?
YEAR (?>\d\d){1,2}
HOUR (?:2[0123]|[01]?[0-9])
MINUTE (?:[0-5][0-9])
# '60' is a leap second in most time standards and thus is valid.
SECOND (?:(?:[0-5]?[0-9]|60)(?:[:.,][0-9]+)?)
TIME (?!<[0-9])%{HOUR}:%{MINUTE}(?::%{SECOND})(?![0-9])
# datestamp is YYYY/MM/DD-HH:MM:SS.UUUU (or something like it)
DATE_US %{MONTHNUM}[/-]%{MONTHDAY}[/-]%{YEAR}
DATE_EU %{MONTHDAY}[./-]%{MONTHNUM}[./-]%{YEAR}
ISO8601_TIMEZONE (?:Z|[+-]%{HOUR}(?::?%{MINUTE}))
ISO8601_SECOND (?:%{SECOND}|60)
TIMESTAMP_ISO8601 %{YEAR}-%{MONTHNUM}-%{MONTHDAY}[T ]%{HOUR}:?%{MINUTE}(?::?%{SECOND})?%{ISO8601_TIMEZONE}?
DATE %{DATE_US}|%{DATE_EU}
DATESTAMP %{DATE}[- ]%{TIME}
TZ (?:[PMCE][SD]T|UTC)
DATESTAMP_RFC822 %{DAY} %{MONTH} %{MONTHDAY} %{YEAR} %{TIME} %{TZ}
DATESTAMP_RFC2822 %{DAY}, %{MONTHDAY} %{MONTH} %{YEAR} %{TIME} %{ISO8601_TIMEZONE}
DATESTAMP_OTHER %{DAY} %{MONTH} %{MONTHDAY} %{TIME} %{TZ} %{YEAR}
DATESTAMP_EVENTLOG %{YEAR}%{MONTHNUM2}%{MONTHDAY}%{HOUR}%{MINUTE}%{SECOND}

# Syslog Dates: Month Day HH:MM:SS
SYSLOGTIMESTAMP %{MONTH} +%{MONTHDAY} %{TIME}
PROG (?:[\w._/%-]+)
SYSLOGPROG %{PROG:program}(?:\[%{POSINT:pid}\])?
SYSLOGHOST %{IPORHOST}
SYSLOGFACILITY <%{NONNEGINT:facility}.%{NONNEGINT:priority}>
HTTPDATE %{MONTHDAY}/%{MONTH}/%{YEAR}:%{TIME} %{INT}

# Shortcuts
QS %{QUOTEDSTRING}

# Log formats
SYSLOGBASE %{SYSLOGTIMESTAMP:timestamp} (?:%{SYSLOGFACILITY} )?%{SYSLOGHOST:logsource} %{SYSLOGPROG}:
COMMONAPACHELOG %{IPORHOST:clientip} %{USER:ident} %{USER:auth} \[%{HTTPDATE:timestamp}\] "(?:%{WORD:verb} %{NOTSPACE:request}(?: HTTP/%{NUMBER:httpversion})?|%{DATA:rawrequest})" %{NUMBER:response} (?:%{NUMBER:bytes}|-)
COMBINEDAPACHELOG %{COMMONAPACHELOG} %{QS:referrer} %{QS:agent}

# Log Levels
LOGLEVEL ([Aa]lert|ALERT|[Tt]race|TRACE|[Dd]ebug|DEBUG|[Nn]otice|NOTICE|[Ii]nfo|INFO|[Ww]arn?(?:ing)?|WARN?(?:ING)?|[Ee]rr?(?:or)?|ERR?(?:OR)?|[Cc]rit?(?:ical)?|CRIT?(?:ICAL)?|[Ff]atal|FATAL|[Ss]evere|SEVERE|EMERG(?:ENCY)?|[Ee]merg(?:ency)?)
grok內置的一些匹配模式

基礎語法，一種是自帶的模式，一種是自定義的模式：

自帶的模式語法:  %{SYNTAX:SEMANTIC}

SYNTAX是將匹配文本模式的名稱，grok自帶的那些匹配模式名
SEMANTIC是你給一段文字的標識相匹配該匹配模式匹配到的內容，至關於一個字段名

例如：

%{NUMBER:duration} %{IP:client}

好比要解析以下的日誌：

55.3.244.1 GET /index.html 15824 0.043

就可使用該匹配模式去匹配：

%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}

得出的結果以下：

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

 

自定義的模式語法：(?<field_name>the pattern here)

例如：

(?<queue_id>[0-9A-F]{10,11})  #表示10-11個字符的16進制

咱們能夠建立一個目錄pattters，把我自定義的模式添加進去，在使用的時候就可使用grok自帶的匹配模式的語法，例如：

咱們在./patterns/postfix文件中添加以下內容
POSTFIX_QUEUEID [0-9A-F]{10,11}

如上咱們就定義了一個匹配模式了，可使用以下的方式使用。假如咱們有以下的日誌格式：

Jan  1 06:25:43 mailserver14 postfix/cleanup[21403]: BEF25A72965: message-id=<20130101142543.5828399CCAF@mailserver14.example.com>

而後咱們對其就行解析：

filter {
      grok {
        patterns_dir => ["./patterns"]  #指定自定義的匹配模式路徑
        match => { "message" => "%{SYSLOGBASE} %{POSTFIX_QUEUEID:queue_id}: %{GREEDYDATA:syslog_message}" }
      }
    }

解析出的結果以下：

timestamp: Jan  1 06:25:43
logsource: mailserver14
program: postfix/cleanup
pid: 21403
queue_id: BEF25A72965
syslog_message: message-id=<20130101142543.5828399CCAF@mailserver14.example.com>

 grok的配置選項：

break_on_match
值類型是布爾值
默認是true
描述：match能夠一次設定多組,預設會依照順序設定處理，若是日誌知足設定條件，則會終止向下處理。但有的時候咱們會但願讓Logstash跑完全部的設定，這時能夠將break_on_match設爲false。

keep_empty_captures
值類型是布爾值
默認值是 false
描述：若是爲true，捕獲失敗的字段將設置爲空值

match
值類型是數組
默認值是 {}
描述：字段值的模式匹配
例如：
filter {
  grok { match => { "message" => "Duration: %{NUMBER:duration}" } }
}
#若是你須要針對單個字段匹配多個模式，則該值能夠是一組,例如：
filter {
  grok { match => { "message" => [ "Duration: %{NUMBER:duration}", "Speed: %{NUMBER:speed}" ] } }
}

named_captures_only
值類型是布爾值
默認值是 true
描述：若是設置爲true，則僅存儲來自grok的命名捕獲

overwrite
值類型是 array
默認是[]
描述：覆蓋已經存在的字段內容
例如：

filter {
  grok {
    match => { "message" => "%{SYSLOGBASE} %{DATA:message}" }
    overwrite => [ "message" ]
  }
}
若是日誌是May 29 16:37:11 sadness logger: hello world通過match屬性match => { 「message」 => 「%{SYSLOGBASE} %{DATA:message}」 }處理後，message的值變成了hello world。這時若是使用了overwrite => [ 「message」 ]屬性，那麼原來的message的值將被覆蓋成新值。

pattern_definitions
值類型是 數組
默認值是 {}
描述：模式名稱和模式正則表達式，也是用於定義當前過濾器要使用的自定義模式。匹配現有名稱的模式將覆蓋預先存在的定義。能夠將此視爲僅適用於grok定義的內聯模式，patterns_dir是將模式寫在外部。
例如：
filter {
    grok {
        patterns_dir => "/usr/local/elk/logstash/patterns"
        pattern_definitions => {"MYSELFTIMESTAMP" => "20%{YEAR}-%{MONTHNUM}-%{MONTHDAY} %{HOUR}:?%{MINUTE}(?::?%{SECOND})"}
        match => {"message" => ["%{MYSELFTIMESTAMP:timestamp} %{JAVACLASS:message}","%{MYSELF:content}"]}
    }
}

patterns_dir
值類型是數組
默認值是 []
描述：一些複雜的正則表達式，不適合直接寫到filter中，能夠指定一個文件夾，用來專門保存正則表達式的文件，須要注意的是該文件夾中的全部文件中的正則表達式都會被依次加載，包括備份文件。

patterns_dir => ["/opt/logstash/patterns", "/opt/logstash/extra_patterns"]
正則文件以文本格式描述：

patterns_file_glob
屬性值的類型：string
默認值：「*」
描述：針對patterns_dir屬性中指定的文件夾裏哪些正則文件，能夠在這個filter中生效，須要本屬性來指定。默認值「*」是指全部正則文件都生效。

tag_on_failure
值類型是數組
默認值是 [「_grokparsefailure」]
描述：沒有成功匹配時，將值附加到字段到tags

tag_on_timeout
值類型是字符串
默認值是 「_groktimeout」
描述：若是Grok正則表達式超時，則應用標記。

timeout_millis
值類型是數字
默認值是 30000
描述： 嘗試在這段時間後終止正則表達式。若是應用了多個模式，則這適用於每一個模式。這將永遠不會提早超時，但超時可能須要一些時間。實際的超時時間是基於250ms量化的近似值。設置爲0以禁用超時。

各組件的公共配置選項：

break_on_match
值類型是布爾值
默認是true
描述：match能夠一次設定多組,預設會依照順序設定處理，若是日誌知足設定條件，則會終止向下處理。但有的時候咱們會但願讓Logstash跑完全部的設定，這時能夠將break_on_match設爲false。

keep_empty_captures
值類型是布爾值
默認值是 false
描述：若是爲true，捕獲失敗的字段將設置爲空值

match
值類型是數組
默認值是 {}
描述：字段值的模式匹配
例如：

filter {
  grok { match => { "message" => "Duration: %{NUMBER:duration}" } }
}

#若是你須要針對單個字段匹配多個模式，則該值能夠是一組,例如：
filter {
  grok { match => { "message" => [ "Duration: %{NUMBER:duration}", "Speed: %{NUMBER:speed}" ] } }
}

named_captures_only
值類型是布爾值
默認值是 true
描述：若是設置爲true，則僅存儲來自grok的命名捕獲

overwrite
值類型是 array
默認是[]
描述：覆蓋已經存在的字段內容
例如：

filter {
  grok {
    match => { "message" => "%{SYSLOGBASE} %{DATA:message}" }
    overwrite => [ "message" ]
  }
}

若是日誌是May 29 16:37:11 sadness logger: hello world通過match屬性match => { 「message」 => 「%{SYSLOGBASE} %{DATA:message}」 }處理後，message的值變成了hello world。這時若是使用了overwrite => [ 「message」 ]屬性，那麼原來的message的值將被覆蓋成新值。

pattern_definitions
值類型是 數組
默認值是 {}
描述：模式名稱和模式正則表達式，也是用於定義當前過濾器要使用的自定義模式。匹配現有名稱的模式將覆蓋預先存在的定義。能夠將此視爲僅適用於grok定義的內聯模式，patterns_dir是將模式寫在外部。
例如：

filter {
    grok {
        patterns_dir => "/usr/local/elk/logstash/patterns"
        pattern_definitions => {"MYSELFTIMESTAMP" => "20%{YEAR}-%{MONTHNUM}-%{MONTHDAY} %{HOUR}:?%{MINUTE}(?::?%{SECOND})"}
        match => {"message" => ["%{MYSELFTIMESTAMP:timestamp} %{JAVACLASS:message}","%{MYSELF:content}"]}
    }
}

patterns_dir
值類型是數組
默認值是 []
描述：一些複雜的正則表達式，不適合直接寫到filter中，能夠指定一個文件夾，用來專門保存正則表達式的文件，須要注意的是該文件夾中的全部文件中的正則表達式都會被依次加載，包括備份文件。

patterns_dir => ["/opt/logstash/patterns", "/opt/logstash/extra_patterns"]
正則文件以文本格式描述：


patterns_file_glob
屬性值的類型：string
默認值：「*」
描述：針對patterns_dir屬性中指定的文件夾裏哪些正則文件，能夠在這個filter中生效，須要本屬性來指定。默認值「*」是指全部正則文件都生效。

tag_on_failure
值類型是數組
默認值是 [「_grokparsefailure」]
描述：沒有成功匹配時，將值附加到字段到tags

tag_on_timeout
值類型是字符串
默認值是 「_groktimeout」
描述：若是Grok正則表達式超時，則應用標記。

timeout_millis
值類型是數字
默認值是 30000
描述： 嘗試在這段時間後終止正則表達式。若是應用了多個模式，則這適用於每一個模式。這將永遠不會提早超時，但超時可能須要一些時間。實際的超時時間是基於250ms量化的近似值。設置爲0以禁用超時。

經常使用選項
全部過濾器插件都支持如下配置選項：


dd_field
值類型是散列
默認值是 {}
描述：若是匹配成功，向此事件添加任意字段。字段名能夠是動態的，並使用%{Field}包含事件的一部分
filter {
      grok {
        add_field => { "foo_%{somefield}" => "Hello world, from %{host}" }
      }
    }

# 你也能夠一次添加多個字段
filter {
  grok {
    add_field => {
      "foo_%{somefield}" => "Hello world, from %{host}"
      "new_field" => "new_static_value"
    }
  }
}

add_tag
值類型是數組
默認值是 []
描述：若是此過濾器成功，請向該事件添加任意標籤。標籤能夠是動態的，並使用%{field} 語法包含事件的一部分。
例如：

filter {
  grok {
    add_tag => [ "foo_%{somefield}" ]
  }
}

# 你也能夠一次添加多個標籤
filter {
  grok {
    add_tag => [ "foo_%{somefield}", "taggedy_tag"]
  }
}

enable_metric
值類型是布爾值
默認值是 true
描述：禁用或啓用度量標準

id
值類型是字符串
此值沒有默認值。
描述：向插件實例添加惟一ID，此ID用於跟蹤插件特定配置的信息。
例如：

filter {
  grok {
    id => "ABC"
  }
}

periodic_flush
值類型是布爾值
默認值是 false
描述：若是設置爲ture，會定時的調用filter的更新函數（flush method）

remove_field
值的類型：array
默認值：[]
描述：刪除當前文檔中的指定filted

filter {
  grok {
    remove_field => [ "foo_%{somefield}" ]
  }
}
# 你也能夠一次移除多個字段:
filter {
  grok {
    remove_field => [ "foo_%{somefield}", "my_extraneous_field" ]
  }
}

remove_tag
值類型是數組
默認值是 []
描述：若是此過濾器成功，請從該事件中移除任意標籤。標籤能夠是動態的，並使用%{field} 語法包括事件的一部分。
例如：

filter {
  grok {
    remove_tag => [ "foo_%{somefield}" ]
  }
}
# 你也能夠一次刪除多個標籤
filter {
  grok {
    remove_tag => [ "foo_%{somefield}", "sad_unwanted_tag"]
  }
}

二、mutate

官網網址：https://www.elastic.co/guide/en/logstash/current/plugins-filters-mutate.html

mutate容許對字段執行常規改變。能夠重命名、刪除、替換和修改事件中的字段。

二話不說，來個例子先：

filter {
    mutate {
        split => ["hostname", "."]  #切分
        add_field => { "shortHostname" => "%{hostname[0]}" } #獲取切分後的第一個字段做爲添加字段
    }
    mutate {
        rename => ["shortHostname", "hostname" ] #重命名
    }
}

mutate的配置選項：

經常使用的一些操做：
convert：轉換數據類型，數據類型hash
能夠裝換的數據類型有：integer，string，integer_eu（和integer相同，顯示格式爲1.000），float，float_eu，boolean
實例：
filter {
      mutate {
        convert => {
          "fieldname" => "integer"
          "booleanfield" => "boolean"
        }
      }
    }

copy：類型hash，將現有字段複製到另外一個字段。現有的目標域將被覆蓋
實例：
 filter {
      mutate {
         copy => { "source_field" => "dest_field" }
      }
    }
    
gsub：類型array，根據字段值匹配正則表達式，並用替換字符串替換全部匹配項。只支持字符串或字符串數組的字段。對於其餘類型的字段，將不採起任何操做。
實例：
filter {
      mutate {
        gsub => [
          # replace all forward slashes with underscore
          "fieldname", "/", "_",
          # replace backslashes, question marks, hashes, and minuses
          # with a dot "."
          "fieldname2", "[\\?#-]", "."
        ]
      }
    }
    
join：類型hash，用分隔符鏈接數組。對非數組字段不執行任何操做
實例：
filter {
     mutate {
       join => { "fieldname" => "," }
     }
   }
  
lowercase：類型array，轉爲小寫
實例：
filter {
      mutate {
        lowercase => [ "fieldname" ]
      }
    }
    
merge：類型hash，合併數組或散列的兩個字段。字符串字段將自動轉換爲數組
實例：
filter {
     mutate {
        merge => { "dest_field" => "added_field" }
     }
   }
   
coerce：類型hash，設置存在但爲空的字段的默認值
實例：
filter {
      mutate {
        # Sets the default value of the 'field1' field to 'default_value'
        coerce => { "field1" => "default_value" }
      }
    }
    
rename：類型hash，重命名
實例：
filter {
      mutate {
        # Renames the 'HOSTORIP' field to 'client_ip'
        rename => { "HOSTORIP" => "client_ip" }
      }
    }
    
replace：類型hash，用新值替換字段的值
實例：
filter {
      mutate {
        replace => { "message" => "%{source_host}: My new message" }
      }
    }
    
split：類型hash，使用分隔符將字段分割爲數組。只對字符串字段有效
實例：
filter {
      mutate {
         split => { "fieldname" => "," }
      }
    }
    
strip：類型array，字段中刪除空白。注意:這隻對前導和後導空格有效。
實例：
filter {
      mutate {
         strip => ["field1", "field2"]
      }
    }
    
update：類型hash，使用新值更新現有字段。若是該字段不存在，則不採起任何操做。
實例：
filter {
      mutate {
        update => { "sample" => "My new message" }
      }
    }
    
uppercase：類型array，轉爲大寫
實例：
filter {
      mutate {
        uppercase => [ "fieldname" ]
      }
    }
    
capitalize：類型array，將字符串轉換爲等效的大寫字母。
實例：
filter {
      mutate {
        capitalize => [ "fieldname" ]
      }
    }

tag_on_failure：類型string，若是在應用此變異篩選器期間發生故障，則終止其他操做，默認值：_mutate_error

公共配置見：grok公共配置

三、date

date filter用於從字段解析日期，而後使用該日期或時間戳做爲事件的logstash時間戳

date經常使用的配置選項：

match：類型array，字段名在前，格式模式在後的數組[ field,formats... ]，表示該字段可以匹配到的時間模式，時間模式能夠有多種
實例：
filter {
      date {
        match => [ "logdate", "MMM dd yyyy HH:mm:ss" ]
      }
    }
    
tag_on_failure：類型array，默認值["_dateparsefailure"]，當沒有成功匹配時，將值追加到tags字段

target：類型String，默認值"@timestamp"，將匹配的時間戳存儲到給定的目標字段中。若是沒有提供，默認更新事件到@timestamp字段。

timezone：類型String，表示時區，能夠在該網址查看：http://joda-time.sourceforge.net/timezones.html

公共配置見：grok公共配置

這裏只對如上三種filter說明，具體其餘的filter請見官網：https://www.elastic.co/guide/en/logstash/current/filter-plugins.html

3.十、案例1、apache日誌解析

這個例子屬官網的一個例子：https://www.elastic.co/guide/en/logstash/current/advanced-pipeline.html

可是我這裏不弄這麼負責，咱們不使用filebeat，直接使用logstash，apache日誌的數據集下載：https://download.elastic.co/demos/logstash/gettingstarted/logstash-tutorial.log.gz

我這裏不打算安裝apach，因此直接使用官方提供的數據集。

下載數據集，而後解壓文件，就能夠獲得咱們的一個日誌文件：logstash-tutorial.log

首先咱們看一下apache日誌的格式：

[elk@lgh ~]$ tail -3 logstash-tutorial.log 
86.1.76.62 - - [04/Jan/2015:05:30:37 +0000] "GET /projects/xdotool/ HTTP/1.1" 200 12292 "http://www.haskell.org/haskellwiki/Xmonad/Frequently_asked_questions" "Mozilla/5.0 (X11; Linux x86_64; rv:24.0) Gecko/20140205 Firefox/24.0 Iceweasel/24.3.0"
86.1.76.62 - - [04/Jan/2015:05:30:37 +0000] "GET /reset.css HTTP/1.1" 200 1015 "http://www.semicomplete.com/projects/xdotool/" "Mozilla/5.0 (X11; Linux x86_64; rv:24.0) Gecko/20140205 Firefox/24.0 Iceweasel/24.3.0"
86.1.76.62 - - [04/Jan/2015:05:30:37 +0000] "GET /style2.css HTTP/1.1" 200 4877 "http://www.semicomplete.com/projects/xdotool/" "Mozilla/5.0 (X11; Linux x86_64; rv:24.0) Gecko/20140205 Firefox/24.0 Iceweasel/24.3.0"

而後開始配置：

cd logstash-7.7.0/ && mkdir conf.d
cd conf.d/ 
vim  apache.conf
#############apache.conf的內容以下###################
input {
        file {
            path => "/home/elk/logstash-tutorial.log"
            type => "log"
            start_position => "beginning"
            }
    }
filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}"}
    }
}

output {
    stdout { codec => rubydebug }
}

而後啓動命令（能夠選擇用nohup後臺啓動）：

 cd logstash-7.7.0/ && ./bin/logstash -f conf.d/apache.conf

執行結果以下（部分結果）：

{
           "verb" => "GET",
          "bytes" => "8948",
           "type" => "log",
           "host" => "gxt_126_233",
    "httpversion" => "1.0",
        "message" => "67.214.178.190 - - [04/Jan/2015:05:20:59 +0000] \"GET /blog/geekery/installing-windows-8-consumer-preview.html HTTP/1.0\" 200 8948 \"http://www.semicomplete.com/\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10.7; rv:21.0) Gecko/20100101 Firefox/21.0\"",
      "timestamp" => "04/Jan/2015:05:20:59 +0000",
       "referrer" => "\"http://www.semicomplete.com/\"",
     "@timestamp" => 2020-06-17T01:23:47.817Z,
           "path" => "/data/hd05/elk/logstash-tutorial.log",
          "ident" => "-",
       "response" => "200",
       "@version" => "1",
        "request" => "/blog/geekery/installing-windows-8-consumer-preview.html",
       "clientip" => "67.214.178.190",
          "agent" => "\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10.7; rv:21.0) Gecko/20100101 Firefox/21.0\"",
           "auth" => "-"
}
{
           "verb" => "GET",
          "bytes" => "1015",
           "type" => "log",
           "host" => "gxt_126_233",
    "httpversion" => "1.1",
        "message" => "66.249.73.185 - - [04/Jan/2015:05:18:48 +0000] \"GET /reset.css HTTP/1.1\" 200 1015 \"-\" \"Mozilla/5.0 (compatible; Googlebot/2.1; +http://www.google.com/bot.html)\"",
      "timestamp" => "04/Jan/2015:05:18:48 +0000",
       "referrer" => "\"-\"",
     "@timestamp" => 2020-06-17T01:23:47.815Z,
           "path" => "/data/hd05/elk/logstash-tutorial.log",
          "ident" => "-",
       "response" => "200",
       "@version" => "1",
        "request" => "/reset.css",
       "clientip" => "66.249.73.185",
          "agent" => "\"Mozilla/5.0 (compatible; Googlebot/2.1; +http://www.google.com/bot.html)\"",
           "auth" => "-"
}
{
           "verb" => "GET",
          "bytes" => "28370",
           "type" => "log",
           "host" => "gxt_126_233",
    "httpversion" => "1.0",
        "message" => "207.241.237.220 - - [04/Jan/2015:05:21:16 +0000] \"GET /blog/tags/projects HTTP/1.0\" 200 28370 \"http://www.semicomplete.com/blog/tags/C\" \"Mozilla/5.0 (compatible; archive.org_bot +http://www.archive.org/details/archive.org_bot)\"",
      "timestamp" => "04/Jan/2015:05:21:16 +0000",
       "referrer" => "\"http://www.semicomplete.com/blog/tags/C\"",
     "@timestamp" => 2020-06-17T01:23:47.817Z,
           "path" => "/data/hd05/elk/logstash-tutorial.log",
          "ident" => "-",
       "response" => "200",
       "@version" => "1",
        "request" => "/blog/tags/projects",
       "clientip" => "207.241.237.220",
          "agent" => "\"Mozilla/5.0 (compatible; archive.org_bot +http://www.archive.org/details/archive.org_bot)\"",
           "auth" => "-"
}
執行結果

2.十一、案例2、nginx日誌解析

首先安裝nginx：nginx功能介紹和基本安裝

這裏咱們也不使用filebeat，由於這篇文章只是介紹logstash

配置：

cd conf.d/ 
vim  nginx.conf
############nginx.conf配置以下#####################
input {
        file {
            path => "/usr/local/nginx/logs/access.log"
            type => "log"
            start_position => "beginning"
            }
    }
   
filter {
grok {
        match => { "message" => ["(?<RemoteIP>(\d*.\d*.\d*.\d*)) - %{DATA:[nginx][access][user_name]} \[%{HTTPDATE:[nginx][access][time]}\] \"%{WORD:[nginx][access][method]} %{DATA:[nginx][access][url]} HTTP/%{NUMBER:[nginx][access][http_version]}\" %{NUMBER:[nginx][access][response_code]} %{NUMBER:[nginx][access][body_sent][bytes]} \"%{DATA:[nginx][access][referrer]}\" \"%{DATA:[nginx][access][agent]}\""] }
        add_field => {
                "Device" => "Charles Desktop"
        }
        remove_field => "message"
        remove_field => "beat.version"
        remove_field => "beat.name"
    }
}
    output {    
             elasticsearch {
                    hosts => ["192.168.110.130:9200"]
                    index => "nginx-log-%{+YYYY.MM.dd}"
                }
    }

如上的配置中輸出到elasticsearch中，這裏沒有設置密碼，因此不須要用戶和密碼，還有就是這裏使用的默認模板，若是想要修改的話可使用，能夠添加以下配置：

user => "elastic"  #用戶
password => "${ES_PWD}" #經過keystore存儲的密碼
manage_template => false #關閉默認的模板
template_name => "elastic-slowquery" #指定自定義的模板

執行命令啓動logstash

cd logstash-7.7.0/ && ./bin/logstash -f conf.d/nginx.conf

執行的結果（查看elasticsearch集羣）：

 

 

 從上面的兩個圖看，一個建立了咱們在nginx.conf中指定的一個索引，而後索引的內容都是解析出來的一些字段內容。

3.十二、案例3、elasticsearch慢日誌解析

這是實例咱們採用filebeat+logstash+elasticsearch，還有權限驗證進行試驗：

這裏主要是對elasticsearch的慢日誌查詢作解析，雖然我在一篇文章搞懂filebeat（ELK）中篇文章中直接經過filebeat的elasticsearch（beat版本）的模塊對其作過解析，可是解析的仍是不夠特別完善，這裏引入logstash對其解析，過濾。

首先配置filebeat文件（這裏只配置了一個輸入和一個輸出，沒有作多餘的處理，只是用來收集日誌）：

#=========================== Filebeat inputs =============================
filebeat.inputs:

# Each - is an input. Most options can be set at the input level, so
# you can use different inputs for various configurations.
# Below are the input specific configurations.

- type: log

  # Change to true to enable this input configuration.
  enabled: true

  # Paths that should be crawled and fetched. Glob based paths.
  paths:
    - /var/logs/es_aaa_index_search_slowlog.log
    - /var/logs/es_bbb_index_search_slowlog.log
    #- c:\programdata\elasticsearch\logs\*

#================================ Outputs =====================================

# Configure what output to use when sending the data collected by the beat.

#----------------------------- Logstash output --------------------------------
output.logstash:
  # The Logstash hosts
  hosts: ["192.168.110.130:5044","192.168.110.131:5044","192.168.110.132:5044"]
  loadbalance: true   #這裏採用負載均衡機制，

  # Optional SSL. By default is off.
  # List of root certificates for HTTPS server verifications
  #ssl.certificate_authorities: ["/etc/pki/root/ca.pem"]

  # Certificate for SSL client authentication
  #ssl.certificate: "/etc/pki/client/cert.pem"

  # Client Certificate Key
  #ssl.key: "/etc/pki/client/cert.key"

而後啓動filebeat：

cd filebeat-7.7.0-linux-x86_64 && ./filebeat -e

而後配置logstash的配置文件：

cd conf.d/ 
vim  es.conf
############es.conf配置以下##############
input {
        beats{
                port => 5044
        }
    }
   
filter {
grok {
        match => {"message" => "\[%{TIMESTAMP_ISO8601:query_time},%{NUMBER:number1}\]\s*\[%{DATA:log_type}\]\s*\[%{DATA:index_query_type}\]\s*\[%{DATA:es_node}\]\s*\[%{DATA:index_name}\]\s*\[%{NUMBER:share_id}\]\s*took\[%{DATA:times_s}\],\s*took_millis\[%{NUMBER:query_times_ms}\],\s*types\[%{DATA:types}\],\s*stats\[%{DATA:status}\],\s*search_type\[%{DATA:search_type}\],\s*total_shards\[%{NUMBER:total_shards}\],\s*source\[%{DATA:json_query}\],\s*extra_source"}
        remove_field => ["message","@version","status","times_s","@timestamp","number1"]
    }
  mutate{
        convert => {
        "query_times_ms" => "integer"
}
}
}
    output {    
             elasticsearch {
                    hosts => ["192.168.110.130:9200","192.168.110.131:9200","192.168.110.132:9200"]
                    index => "elastic-slowquery-222"
                    user => "elastic"
                    password => "${ES_PWD}"
                    manage_template => false
                    template_name => "elastic-slowquery"
                }
    }

如上配置使用了用戶的權限驗證，以及elasticsearch的自定義模板

啓動logstash

cd logstash-7.7.0/ && ./bin/logstash -f conf.d/es.conf

登陸es查看結果：

 

logstash就介紹到這裏了，若是有疑問多看官網比較好

 

參考：

logstash官網：https://www.elastic.co/guide/en/logstash/current/index.html

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程