一、需求

使用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