elk部署

介绍

公司重新设定了日志的标准格式，日志打印格式均为json格式，之前是用filebeat进行收集的，没任何问题，但是这次的这个标准日志推的时候出现了，在日志显示和多行收集的时候出现了问题。
java应用在报错的时候一般会提示一个很长的错误信息，如果在日志标准里面加入堆内存的错误，那信息就更长，一条日志就更大了，所以这里就得考量这个堆内存错误要打印多少问题。在显示的时候也遇到一个问题就是内容没有被索引，google一下查询是not_analyzed字符串默认设置导致的。查询得知对超过ignore_above的字符串，analyzer 不会进行处理。所以就不会索引，导致的结果就是最终搜索引擎搜索不到。这个选项主要对 not_analyzed 字段有用，这些字段通常用来进行过滤、聚合和排序。而且这些字段都是结构化的，所以一般不会允许在这些字段中索引过长的项。

针对这些问题，和研发达成如协议，研发限制一条日志的message输出的长度，日志格式采用json，里面的字段不管。与似乎就要重置一下filebeat的参数，以及需要单独的建立自定义es模版，在logstash建立索引的时候需要用我们自定义的模版来建立索引。

filebeat

filebeat安装不阐述

filebeat配置文件

filebeat.yml

# cat filebeat.yml 
filebeat.inputs:
  - type: log
    enabled: true
    paths:
      - /opt/logs/*/*.log
  tags: ["system"]
  exclude_files: ['.gz$']
  fields:
    env: dev
    logtype: system-java
  queue_size: 3000
  multiline.pattern: '^[[:space:]]'
  multiline.negate: false
  multiline.match: after
  multiline.max_lines: 500
  multiline.timeout: 5s
  registry_file: /var/lib/filebeat/registry
  harvester_buffer_sizeedit: 16384
  max_bytes: 10485760
  max_message_bytes: 1000000
filebeat.config.modules:
  path: ${path.config}/modules.d/*.yml
  reload.enabled: false
output.logstash:
  hosts: ["logstash:5555","logstash:5556"]
  loadbalance: true
  worker: 1

参数详解

input.type: 默认log，从日志文件读取每一行。stdin，从标准输入读取
paths: 日志文件路径列表，可用通配符，不递归
tags: 标记tag，可用于分组
exclude_files: 排除文件，后接一个正则表达式的列表，默认无
fields: 添加附件字段，可以使values，arrays，dictionaries或者任何嵌套数据
- env: 设置环境环境
- logtype: 设置类别，在logstash里面判断，可用于建设不通的索引或者到不通的es集群里面
queue_size: 处理管道中单个事件内的队列大小，默认1000
multiline.pattern: 多行匹配模式，后接正则表达式，默认无
multiline.negate: 多行匹配模式后配置的模式是否取反，默认false
multiline.match: 定义多行内容被添加到模式匹配行之后还是之前，默认无，可以被设置为after或者before
multiline.max_lines: 单一多行匹配聚合的最大行数，超过定义行数后的行会被丢弃，默认500
multiline.timeout: 多行匹配超时时间，超过超时时间后的当前多行匹配事件将停止并发送，然后开始一个新的多行匹配事件，默认5秒
registry_file: 指定注册表文件，用于记录上次读取的位置，默认位于filebeat下的data目录
harvester_buffer_sizeedit: 每个采集器在获取文件时使用的缓冲区的大小（以字节为单位）。默认是16384
max_bytes: 单个日志消息可以拥有的最大字节数。之后的所有字节max_bytes被丢弃并且不被发送。此设置对于可能变大的多行日志消息特别有用。默认值是10MB（10485760）
max_message_bytes: 允许的最大json消息大小，默认为1000000，超出的会被丢弃，应该小于broker的 message.max.bytes（broker能接收消息的最大字节数）
output.logstash: 输出到logstash，从节点列表中选取不同的节点发送数据，达到 Round-Robin 负载均衡的效果。
- worker: 配置每个主机发布事件的worker数量。在负载均衡模式下最好启用。

logstash 部署

安装不阐述，这里使用的自定义elasticsearch模版，可参考

elasticsearch 模版

在Json日志文件中的KEY的位置不固定、或字段数不明确时使用动态映射模板。这里创建一个动态模版，模版是拷贝默认的 logstash-*模版来进行稍加修改的，index_patterns 该模版参数指定任意索引名称，并且指定允许添加新字段，匹配所有string类型的新字段会创建一个raw的嵌套字段，raw嵌套字段类型text，使用 ignore_above改变默认的长度。

$ cat dev-java-template.json
{
  "order": 5,
  "index_patterns": [
    "*"
  ],
  "settings": {
    "index": {
      "number_of_shards": "5",
      "number_of_replicas": "1",
      "refresh_interval": "10s"
    }
  },
  "mappings": {
    "_default_": {
      "dynamic_templates":[
        {
          "message_field": {
            "match": "message",
            "match_mapping_type": "string",
            "mapping": {
              "type": "text",
              "norms" : false,
              "fields": {
                "raw": {
                  "type": "keyword",
                  "ignore_above": 10922
                }
              }
            }
          }
        },
        {
          "strings_fields": {
            "match": "*",
            "match_mapping_type": "string",
            "mapping": {
              "type": "text",
              "norms": false,
              "fields": {
                "raw": {
                  "type": "keyword",
                  "ignore_above": 10922
                }
              }
            }
          }
        }
      ],
      "properties":{
        "@timestamp": {
          "type": "date"
        },
        "@version": {
          "type": "keyword"
        },
        "geoip": {
          "dynamic": true,
          "properties": {
            "ip": {
              "type": "ip"
            },
            "location": {
              "type": "geo_point"
            },
            "latitude": {
              "type": "half_float"
            },
            "longitude": {
              "type": "half_float"
            }
          }
        }
      }
    }
  },
  "aliases": {}
}

注:

dynamic_templates: 配置具体的动态模板匹配项
match_mapping_type: “string” 是匹配固定的类型
match: message 匹配字段名为message的数据
mapping: 将匹配的数据项映射为定义的数据类型

logstash 配置文件

$ cat java.conf
input  {
        beats {
                port => 5555
        }
}
filter {
        # application system logs
        if [fields][logtype] == "system-java" {
                json {
                    source => "message"
                }
        }
	# application business logs
        if [fileds][logtype] == "business-java" {
		json {
			source => "message"
		}
	}
}
output {
        # application system logs
        if [fields][logtype] == "system-java" {
                elasticsearch {
			codec => "json"
                        hosts => ["es1:9200","es2:9200","es3:9200"]
                        index => "%{[fields][kxlenv]}-%{[projectName]}-%{[logType]}-%{+YYYY-MM}"
                        template_overwrite => true
                        template_name => "system-java"
                        template => "/opt/templates/java.template"
                }
        }
	# applicatio business logs
	if [fields][logtype] == "business-java" {
		elasticsearch {
                        action => "index"
			hosts => ["es1:9200","es2:9200","es3:9200"]
			index => "%{[fields][kxlenv]}-%{[projectName]}-%{[logType]}-%{+YYYY-MM}"
			template_overwrite => true
                        template => "/opt/templates/dev-java-template.json"
		}
	}

}