一、需要

应用 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配置参数解析：

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

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