logstash收集的日誌輸出到elasticsearch中

1、需求

使用logstash收集系統上的日誌，並使用 grok解析日誌，使用mutate修改解析出來的字段類型、刪除字段、重命名字段，最後將解析好的日主輸出到 elasticsearch中。

2、實現步驟

一、編寫pipeline文件

vim output-es.yml

input {
    file {
        id => "mutate-id"
        path => ["/Users/huan/soft/elastic-stack/logstash/logstash/pipeline.conf/output-es/*.log"]
        start_position => "beginning"
        sincedb_path => "/Users/huan/soft/elastic-stack/logstash/logstash/pipeline.conf/output-es/sincedb.db"
        codec => multiline {
             pattern => "^\[+"
             negate => "true"
             what => "previous"
             charset => "UTF-8"
             auto_flush_interval => 2
        }
    }
}

filter {
    grok {
        match => {
            "message" => "(?m)^\[%{INT:pid}\]%{SPACE}%{TIMESTAMP_ISO8601:createTime}%{SPACE}\[%{DATA:threadName}\]%{SPACE}%{LOGLEVEL:LEVEL}%{SPACE}%{JAVACLASS:javaClass}#(?<methodName>[a-zA-Z_]+):%{INT:linenumber}%{SPACE}-%{GREEDYDATA:msg}"
            remove_field => ["message"]
        }
    }  

    mutate {
        convert => {
            "pid" => "integer"
        }

        rename => {
            "msg" => "message"
        }
    }

    # 格式化 createTime 將 源格式 轉換成 目標格式
    date {
        match => ["createTime","yyyy-MM-dd HH:mm:ss.SSS","yyyy-MM-dd HH:mm:ss.SSS"]
        target => "@timestamp"
        remove_field => ["createTime"]
    }
}

output {
    # 能夠經過 template 或 template_name 指定es模板的名字
    elasticsearch {
        hosts => ["http://localhost:9200","http://localhost:9201","http://localhost:9202"]
        user => "springboot_logstash"
        password => "123456"
        index => "springboot-%{+YYYY.MM.dd}"
        template_overwrite => "false"
    }
}

一、elasticsearch配置參數解析：

1. hosts: es的訪問地址，建議使用非master節點。
2. user: 訪問es的用戶名。
3. password：訪問es的密碼。
4. index：在es中的索引名稱。
5. template：設置本身的es模板路徑。
6. template_name：使用es中的索引模板名稱。

7. 上方的es的密碼是明文的，可能存在泄漏，能夠使用 logstash keystore來解決。

   1. 參考連接 https://www.elastic.co/guide/en/logstash/current/keystore.html

二、可能會報的一個異常

{
    "error": {
        "root_cause": [
            {
                "type": "security_exception",
                "reason": "action [indices:data/      write/bulk] is unauthorized for user [logstash_system] on indices [], this action is granted by the index privileges [create_doc,create,delete,index,write,all]"
            }
        ],
        "type": "secu      rity_exception",
        "reason": "action [indices:data/write/bulk] is unauthorized for user [logstash_system] on indices [], this action is granted by the index privileges [create_doc      ,create,delete,index,write,all]"
    },
    "status": 403
}

當咱們使用系統自帶的logstash_system用戶時，可能會報indices:data/write/bulk這個操做沒有權限，解決方法以下（本身新建一個用戶和角色）。

二、準備測試數據

[9708] 2021-05-13 11:14:51.873 [http-nio-8080-exec-1] INFO  org.springframework.web.servlet.DispatcherServlet#initServletBean:547 -Completed initialization in 1 ms
[9708] 2021-05-13 11:14:51.910 [http-nio-8080-exec-1] ERROR com.huan.study.LogController#showLog:32 -請求:[/showLog]發生了異常
java.lang.ArithmeticException: / by zero
    at com.huan.study.LogController.showLog(LogController.java:30)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)

三、啓動logstash

bin/logstash -f output-es.yml

四、在es上建立索引模式

五、進行日誌搜索

3、參考文檔

一、https://www.elastic.co/guide/en/logstash/current/keystore.html

二、https://www.elastic.co/guide/en/logstash/current/plugins-outputs-elasticsearch.html