elasticsear+kibana+logstash 优化

Posted 2021-01-18 ipyanthony

##关于s2安装cerebro来可视化集群管理#####
https://blog.csdn.net/RWSLR6/article/details/79648767
https://github.com/lmenezes/cerebro/releases

######关于head安装#################
https://blog.csdn.net/s740556472/article/details/78995843

##关于logstash日志规则制定###############
https://discuss.elastic.co/t/grok-for-parsing-java-log/43718/3

https://www.cnblogs.com/xing901022/p/4805586.html

https://blog.csdn.net/gebitan505/article/details/70255727

https://blog.csdn.net/zhaoyangjian724/article/details/54343178


#####logstash grok过滤器条件#################
https://github.com/elastic/logstash/blob/v1.4.2/patterns/grok-patterns

####logstash grok表达式在线匹配#############
http://grokdebug.herokuapp.com/


##关键笔记
https://blog.csdn.net/gebitan505/article/details/70336282


##关于Elk安全##################################
http://www.cnblogs.com/cjsblog/p/9501858.html
https://www.cnblogs.com/dmjx/p/9120474.html




##Elk相关地址##########################
cerebro地址：http://192.168.2.161:1234/#/overview?host=elk
kibana地址：http://192.168.2.161:5601
elasticsearch：http://192.168.2.161:9200/


####关于ELK版本#############################
elasticsearch     6.4.0
logstash          6.4.2
kibana            6.4.0
cerebro           0.8.1



###关于ELK配置###################################
###关于logstash配置
创建/data/service/logstash/etc目录,启动配置文件为logstash.conf
创建自定义正则表达式匹配文件,在/data/service/logstash/patterns目录下,创建postfix文件
###关于elasticsearch配置#########################
network.host: 192.168.2.161
cluster.name: yutang_elk
node.name: node-1               ###多节点设置不同node名称
http.cors.enabled: true
http.cors.allow-origin: "*"


###关于kibana配置####################

elasticsearch.url: "http://192.168.2.161:9200"

##########关于需要优化内容#################
x-pack使用




####示例版本  Filter日志格式规则##########################################
##日志格式介绍
以|符号隔开，第一个是访问时间，timestamp，作为logstash的时间戳，接下来的依次为：服务端IP，客户端的IP，
机器类型(WEB/APP/ADMIN)，用户的ID(没有用0表示)，请求的完整网址，请求的控制器路径，reference，设备的信息，duringTime，请求所花的时间。

##########################################################################

##打印格式
2015-05-07-16:03:04|10.4.29.158|120.131.74.116|WEB|11299073|
http://quxue.renren.com/shareApp?isappinstalled=0&userId=11299073&from=groupmessage
|/shareApp|null|Mozilla/5.0 (iPhone; CPU iPhone OS 8_2 like Mac OS X) AppleWebKit/600.1.4 (KHTML, like Gecko) Mobile/12D508 MicroMessenger/6.1.5 NetType/WIFI
|duringTime|98||
################################################################

filter {

      #定义数据的格式
      grok {
        match => { "message" => "%{DATA:timestamp}|%{IP:serverIp}|%{IP:clientIp}|%{DATA:logSource}|%{DATA:userId}|%{DATA:reqUrl}|%{DATA:reqUri}|%{DATA:refer}|%{DATA:device}|%{DATA:textDuring}|%{DATA:duringTime:int}||"}
      }

}


##########################################
###编码解码###########
# codec类：定于处理数据格式，如plain,json，json_lines等格式。这个可以定义在input、output区域中。
json_lines          # 用于json过长使用
plain               # 用于文件字节使用
json_lines          # 用于json使用

#####关于Logback and logstash 使用方式###################

1、可以提供哪些字段
    1.1 logback可以提供哪些字段供logstash做切割筛选





2、扫描方式
    2.1 读取指定目录下文件/**/**/*error.log 文件
    2.2 logstash通过接受logback指定的tcp端口发送的数据
        https://blog.csdn.net/gebitan505/article/details/70336343



3、logbak日志通过tcp传输，logbak需要添加配置(主机地址/端口)
    3.1 示例：
        logback.xml增加
        <appender name="LOGSTASH" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
        <destination>169.254.247.253:9250</destination>
        <!-- encoder必须配置,有多种可选 -->
        <encoder charset="UTF-8" class="net.logstash.logback.encoder.LogstashEncoder" />
        </appender>

        <root level="${log.level}">
        <appender-ref ref="LOGSTASH" />
        </root>

    3.2 示例2：
        <appender name="stash1" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
            <destination>10.8.120.37:4560</destination>
            <encoder charset="UTF-8" class="net.logstash.logback.encoder.LoggingEventCompositeJsonEncoder">
                <providers>
                        <pattern>
                         <pattern>
                            {
                                "timestamp": "%d{yyyy-MM-dd HH:mm:ss.SSS}", //时间戳格式
                                "thread": "%thread", //进程名
                                "level": "%level", //日志等级
                                "logger_name": "%logger", //日志记录器名
                                "class_name": "%class", //类名
                                "method_name": "%method", //方法名
                                "line_number": "%line", //行号
                                "message": "%message" //日志主题信息
                            }
                         </pattern>
                    </pattern>
                </providers>
            </encoder>
        </appender>


####成品############################################
###注意事项##########################
#设置多长时间扫描目录，发现新文件，按秒为单位
discover_interval => 15
#设置多长时间检测文件是否修改
stat_interval => 1
#####################################
创建postfix文件
################################
CUSTOMER_ERROR_LOG customererrorlog
###############################
vim logstash.conf
input {
        file {
                path => "/data/service/logs/*study-center-api*.log"
                type => "elk-java-logs"
                start_position=> "beginning"
                discover_interval => 15
                stat_interval => 1
        }
}


filter {
        grok {
                patterns_dir => ["/data/service/logstash/patterns/postfix"]
                match => { "message" => "%{CUSTOMER_ERROR_LOG:customererrorlog}|%{NOTSPACE:applicationName}|%{NOTSPACE:applicationNode}|%{NOTSPACE:header}|%{NOTSPACE:param}|%{NOTSPACE:msg}|%{NOTSPACE:stac
ktrace}" }
                match => {"customererrorlog" => "%{CUSTOMER_ERROR_LOG:customererrorlog}"}
                match => {"applicationName" => "%{NOTSPACE:applicationName}"}
                match => {"applicationNode" => "%{NOTSPACE:applicationNode}"}
                match => {"header" => "%{NOTSPACE:header}"}
                match => {"param" => "%{NOTSPACE:param}"}
                match => {"msg" => "%{NOTSPACE:msg}"}
                match => {"stacktrace" => "%{NOTSPACE:stacktrace}"}
                break_on_match => true
        }

}


output {
        if !("_grokparsefailure" in [tags]) {
                elasticsearch {
                        hosts => ["192.168.2.161:9200"]
                        index => "elk-java-log-%{+YYYY.MM.dd}"
                }
        }
}