ELK学习--将自定义nginx日志写入es中并通过kibana展示为例

今天只是记录总体思路，具体细节不展开，毕竟东西太多

学习目的

业务发展越来越庞大，服务器越来越多
各种访问日志、应用日志、错误日志量越来越多，导致运维人员无法很好的去管理日志
开发人员排查问题，需要到服务器上查日志，不方便
运营人员需要一些数据，需要我们运维到服务器上分析日志

为啥选用ELK

通常情况下我们通过linux的一些命令（grep，awk等）就可以查看一些命令，当随着业务量大，集群增多，日志管理比较麻烦，通过登陆服务器查看日志的做法已经跟不上日常工作，所以需要将日志统一管理，排查问题系统化，提高运维效率，更加方便的定位问题。

工具介绍

本次选用的工具主要为以下：

1、elasticsearch-6.8.10

2、filebeat-6.8.10-linux-x86_64

3、kibana-6.8.10-linux-x86_64

4、nginx-1.21.2

ELK选用的是6.8.10的版本，之所以选这个版本主要也是为了跟公司使用同步，具体下载地址可以通过官网：https://www.elastic.co/cn/downloads/，官网学习地址：https://www.elastic.co/guide/en/elasticsearch/reference/6.6/indices.html

安装步骤及配置文件部分讲解

ELK的安装使用是比较人性化的，开箱即用，这里对各个工具的配置文件进行简单演示，我用的用户统一为es用户，这里大家根据自己需要进行创建

1、elasticsearch安装，配置文件在/elasticsearch-6.8.10/config/elasticsearch.yml，因为我是单机配置，所以其他的不用配置，完成之前直接今日bin目录下通过命令：./elasticsearch启动即可

# ----------------------------------- Paths ------------------------------------
# Path to directory where to store the data (separate multiple locations by comma):
path.data: /home/es/elasticsearch/data
# Path to log files:
path.logs: /home/es/elasticsearch/logs
network.host: 0.0.0.0  # 这里配置0.0.0.0表示所有机器均可以访问，在集群中通常配置为内网ip
http.port: 9200   # 端口号不打开即默认9200

2、kibana安装，配置文件在/kibana-6.8.10/config/kibana.yml,主要更改以下地址，配置完成以后通过命令：./kibana即可

server.port: 5601# Specifies the address to which the Kibana server will bind. IP addresses and host names are both valid values.
# The default is 'localhost', which usually means remote machines will not be able to connect.
# To allow connections from remote users, set this parameter to a non-loopback address.
server.host: 172.29.16.150# Enables you to specify a path to mount Kibana at if you are running behind a proxy.
# Use the `server.rewriteBasePath` setting to tell Kibana if it should remove the basePath
# from requests it receives, and to prevent a deprecation warning at startup.
# This setting cannot end in a slash.
#server.basePath: ""# Specifies whether Kibana should rewrite requests that are prefixed with
# `server.basePath` or require that they are rewritten by your reverse proxy.
# This setting was effectively always `false` before Kibana 6.3 and will
# default to `true` starting in Kibana 7.0.
#server.rewriteBasePath: false# The maximum payload size in bytes for incoming server requests.
#server.maxPayloadBytes: 1048576# The Kibana server's name.  This is used for display purposes.
#server.name: "your-hostname"# The URLs of the Elasticsearch instances to use for all your queries.
elasticsearch.hosts: ["http://172.29.16.150:9200"]

3、自定义nginx日志就需要将其定义为json格式，方便filebeat识别，此处根据自己需要调节，仅仅做一个演示

1、先清空原来日志2、# 日志格式自定义
log_format json '{ "@timestamp": "$time_iso8601", ''"remote_addr": "$remote_addr", ''"request": "$uri", ''"request_uri": "$request_uri", ''"request_method":"$request_method",''"server_protocol":"$server_protocol",''"status": $status, ''"bytes": $body_bytes_sent, ''"up_addr": "$upstream_addr",''"request_time": $request_time'' }';access_log  logs/access.log  json;#为了后面根据索引取数据也需要将日志根据日期进行切割，具体如下if ($time_iso8601 ~ "^(\d{4})-(\d{2})-(\d{2})") {set $year $1;set $month $2;set $day $3;}access_log logs/access-${year}-${month}-${day}.log json;

4、filebeat-6.8.10安装，创建配置文件/filebeat-6.8.10/config/filebeat-es.yml文件，主要配置如下：配置完成后在bin目录下通过./filebeat -e -c filebeat-es.yml命令启动，其中-e表示标准启动，-c表示指定配置文件

filebeat.inputs:
- type: logenabled: trueharvester_buffer_size: 10240000filebeat.spool_size: 200000filebeat.idle_timeout: 5s# Paths that should be crawled and fetched. Glob based paths.# 配置日志路径paths:- /home/es/nginx/logs/access*.log# 这两行参数配置表示按json格式去解析json.keys_under_root: trueoverwrite_keys: true# 将结果输出到elasticsearch中
output.elasticsearch:# Array of hosts to connect to.es的安装服务器hosts: ["172.29.16.150:9200"]#  根据日期在es中创建索引，这个很重要,如果配置此配置，就需要配置setup.template.name， #  setup.template.patternindex: "zpl-aliyun-access-%{+yyyy.MM.dd}"bulk_max_size: 15000#flush_interval: 3s
# //设置一个新的模板，模板的名称
setup.template.name: "template-zpl-aliyun-access"
# 模板是否被覆盖，表示不用默认模板，而引用自定义
setup.template.overwrite: false
# //模板匹配那些索引，这里表示以zpl开头的所有的索引
setup.template.pattern: "zpl-aliyun-access*"

如果是将不同路径下日志写入es，参照下图所示，通过字段路径去识别

还可以通过打标签方式实现，如图所示

5、在elasticsearch中创建模板，这一步很重要

PUT _template/template-zpl-aliyun-access
{"template-zpl-aliyun-access" : {"order" : 0,"index_patterns" : ["zpl-aliyun-access*"],"settings" : {"index" : {"max_result_window" : "1000000","number_of_shards" : "3","number_of_replicas" : "0"}},"mappings" : {"doc" : {"properties" : {"remote_addr" : {"ignore_above" : 1024,"type" : "keyword"},"request" : {"ignore_above" : 1024,"type" : "keyword"},"offset" : {"type" : "long"},"prospector" : {"properties" : {"type" : {"ignore_above" : 1024,"type" : "keyword"}}},"read_timestamp" : {"ignore_above" : 1024,"type" : "keyword"},"request_method" : {"ignore_above" : 1024,"type" : "keyword"},"source" : {"ignore_above" : 1024,"type" : "keyword"},"message" : {"norms" : false,"type" : "text"},"request_uri" : {"ignore_above" : 1024,"type" : "keyword"},"@timestamp" : {"type" : "date"},"request_time" : {"type" : "float"},"up_addr" : {"ignore_above" : 1024,"type" : "keyword"},"bytes" : {"type" : "long"},"service" : {"properties" : {"name" : {"ignore_above" : 1024,"type" : "keyword"}}},"server_protocol" : {"ignore_above" : 1024,"type" : "keyword"}}}},"aliases" : {"zpl-aliyun-access" : { }}}
}

针对配置文件做几点说明：

1、index_patterns:此字段中配置的索引表示以zpl-aliyun-access*的索引都适合该mapping规则

2、order：多个索引模板可能与索引匹配，在这种情况下，设置和映射都合并到索引的最终配置中。可以使用order参数控制合并的顺序，首先应用较低的顺序，并且覆盖它们的较高顺序

3、aliases:别名，举例说明，zpl-aliyun-access-2021-12-12的index, 那么它同时生成了一个叫做logs-2019-04-01-alias的别名

其他字段属性可以参考es的学习，这里不说明，此外本文章适合具有一定基础的ELK学习者。

es具体学习可参考：ES基本查询语句教程 - @ 小浩 - 博客园

做聚合查询时候text跟keyword的取别

1、创建索引及其mapping，

1、创建索引及其mapping
PUT index-00000001
{"mappings" : {"doc" : {"properties" : {"age" : {"type" : "long"},"desc" : {"type" : "text","fields" : {"keyword" : {"type" : "keyword","ignore_above" : 256}}},"from" : {"type" : "text","fields" : {"keyword" : {"type" : "keyword","ignore_above" : 256}}},"name" : {"type" : "text","fields" : {"keyword" : {"type" : "keyword","ignore_above" : 256}}},"tags" : {"type" : "text","fields" : {"keyword" : {"type" : "keyword","ignore_above" : 256}}}}}}}2、查看已经创建好的索引
GET index-00000001/_mapping，结果如上述创建所示3、插入数据
PUT index-00000001/doc/1
{"name":"顾老二","age":30,"from": "gu","desc": "皮肤黑、武器长、性格直","tags": ["黑", "长", "直"]
}PUT index-00000001/doc/2
{"name":"大娘子","age":18,"from":"sheng","desc":"肤白貌美，娇憨可爱","tags":["白", "富","美"]
}PUT index-00000001/doc/3
{"name":"龙套偏房","age":22,"from":"gu","desc":"mmp，没怎么看，不知道怎么形容","tags":["造数据", "真","难"]
}PUT index-00000001/doc/4
{"name":"石头","age":29,"from":"gu","desc":"粗中有细，狐假虎威","tags":["粗", "大","猛"]
}PUT index-00000001/doc/5
{"name":"魏行首","age":25,"from":"广云台","desc":"仿佛兮若轻云之蔽月,飘飘兮若流风之回雪,mmp，最后竟然没有嫁给顾老二！","tags":["闭月","羞花"]
}4、此时我要对from字段进行聚合查询，并对每个from的人进行age平均值计算
错误示例：
GET zhifou/doc/_search
{"aggs": {"1": {"terms": {"field": "from"  //这里from采用的第一属性text，无法查询到结果},"aggs": {"2": {"avg": {"field": "age"}}}}},"size":0,"_source": ["name", "age"]
}
当采用如何方式查询时候有报错，所以将其调整为以下方式
正确示例：
GET zhifou/doc/_search
{"aggs": {"1": {"terms": {"field": "from.keyword"},"aggs": {"2": {"avg": {"field": "age"}}}}},"size":0,"_source": ["name", "age"]
}结果为：
{"took" : 3,"timed_out" : false,"_shards" : {"total" : 5,"successful" : 5,"skipped" : 0,"failed" : 0},"hits" : {"total" : 5,"max_score" : 0.0,"hits" : [ ]},"aggregations" : {"1" : {"doc_count_error_upper_bound" : 0,"sum_other_doc_count" : 0,"buckets" : [{"key" : "gu","doc_count" : 3,"2" : {"value" : 27.0}},{"key" : "sheng","doc_count" : 1,"2" : {"value" : 18.0}},{"key" : "广云台","doc_count" : 1,"2" : {"value" : 25.0}}]}}
}