一、需要

应用logstash收集零碎上的日志，并应用 grok解析日志，应用mutate批改解析进去的字段类型、删除字段、重命名字段，最初将解析好的日主输入到 elasticsearch中。

二、实现步骤

1、编写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"    }}

1、`elasticsearch`配置参数解析：

hosts: es的拜访地址，倡议应用非master节点。
user: 拜访es的用户名。
password：拜访es的明码。
index：在es中的索引名称。
template：设置本人的es模板门路。
template_name：应用es中的索引模板名称。
上方的es的明码是明文的，可能存在透露，能够应用 logstash keystore来解决。
1. 参考链接 https://www.elastic.co/guide/en/logstash/current/keystore.html

2、可能会报的一个异样

{    "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这个操作没有权限，解决办法如下（本人新建一个用户和角色）。

2、筹备测试数据

[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)

3、启动logstash

bin/logstash -f output-es.yml

4、在es上创立索引模式

5、进行日志搜寻

三、参考文档

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

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