2020-06-27

ELK

ELK深入浅出

ELK是Elasticsearch、Logstash、Kinaba三个产品的简称。

Elasticsearch：数据存储与检索
Logstash：数据收集与分析
Kinaba：数据可视化

环境Centos7，Docker

Elasticsearch安装

首先用Docker拉取Elasticsearch镜像，版本用的是6.8.8

docker pull elasticsearch:6.8.8

创建容器

docker run -itd --name es -p 9200:9200 -p 9300:9300 -e "discovery.type=single-node" elasticsearch:6.8.8 -v ~/elasticsearch.yml:/usr/share/elasticsearch/elasticsearch.yml

安装IK中文分词器

如果不需要中文分词，可以跳过这一步
进入到elasticsearch容器中，使用elasticsearch-plugin命令安装插件

docker exec -it es /bin/bash
bin/elasticsearch-plugin install https://github.com/medcl/elasticsearch-analysis-ik/releases/download/v6.8.8/elasticsearch-analysis-ik-6.8.8.zip

ES配置

然后再到~/elasticsearch.yml下修改配置

# 集群名称
cluster.name: "docker-cluster"
network.host: 0.0.0.0

# elasticsearch-head访问需要允许跨域
http.cors.enabled: true
http.cors.allow-origin: "*"

然后重启服务即可生效

Logstash安装

docker pull logstash:6.8.8
docker run -itd --name logstash --link es:elasticsearch -v ~/pipeline/:/usr/share/logstash/pipeline logstash:6.8.8

es是ES容器的名称，用--link将其取个别名（elasticsearch）映射到logstash容器上，logstash容器可以通过elasticsearch访问ES容器，把elasticsearch作为ES容器的IP。

宿主机~/pipeline目录里面可以自定义conf文件，每一个conf文件就是一个logstash收集数据的任务。
配置文件格式：

input { #输入
	...
}

filter { #过滤，对数据进行分割、截取等处理 
	... 
}

output { #输出 
	...
}

input是数据来源，filter可以对数据进行过滤和分析，output确定数据的目的地。
举个简单例子：

input {
	 stdin { }
}

output  
	stdout { }
}

监听控制台的输入，然后输出到控制台

收集系统日志到ES

以监听本地文件为例，收集到的日志存储到ES上，index名称system-log-{yyyy.MM.dd}
在宿主机~/pipeline目录下新建个收集任务（创建system.conf文件）

input { 
  file {
    path => "/usr/share/logstash/data/a.log"
    start_position => "beginning"
  }
} 

filter {
  mutate {
    split => {"message" => "|"}
  }
}

output { 
  elasticsearch {
    hosts => "elasticsearch:9200"
    index => "system-log-%{+YYYY.MM.dd}"
  }
}

input中收集类型为file，path指向文件路径，注意这里是docker容器中的文件路径而不是宿主机的。如果需要监听宿主机请用-v挂载目录。start_position指定从第一行位置。
filter作为过滤器只做一件事，就是把每行的日志以|作为分隔符分割
output指定输出的目的地是ES，hosts也可以是个数组（如hosts => ["elasticsearch:9200"]），指向多个es，elasticsearch是es容器的别名，可以把它作为ip使用。index指定当前日志写入到system-log-开头当前日期的index上去。

添加或修改logstash收集任务需要重启logstash服务。

进入logstash容器中，用echo向a.log追加一条数据

echo "20080323|张三|男" >> a.log

需要注意的是，这里发送的类型默认是text类型，如果写成2008-03-23会被当做Date类型，导致上传失败，报400错误。

先确定当前机器时间正确，如果不正确可以用ntpdate工具同步下时间

# 安装ntpdate工具
yum -y install ntp ntpdate
# 设置系统时间与网络时间同步
ntpdate cn.pool.ntp.org
# 将系统时间写入硬件时间
hwclock --systohc

用date命令查看当前时间

上传成功之后打开Kibana会发现ES上多了一个system-log-2020.06.27的index

Kibana安装

docker pull kibana:6.8.8
docker run -itd --name kibana --link es:elasticsearch -p 5601:5601 kibana:6.8.8

Kibana中文界面显示

进入到Kibana容器，修改/usr/share/kibana/config/kibana.yml配置文件，添加这一行配置，就可以将kibana界面换成中文了。

i18n.locale: "zh-CN"

然后重启Kibana容器

可以用elasticsearch-head来查看ES的数据，现在用Kibana演示下。
依次点击管理 -> 索引模式 -> 创建索引模式，然后将system-log-2020.06.27这个索引名复制到输入框中，点击下一步

直接选“我不想使用时间筛选”然后创建就行了。

创建成功之后在 Discover 发现我们刚刚创建的索引模式，可以选中它来查看当前index中存储的数据。

在message字段这里可以看到我们发送的数据。

# Elasticsearch ELK Kibana Logstash