linux环境搭建-Docker搭建Elk分布式日志收集系统 - 《Linux环境搭建》

#zookeeper
docker pull zookeeper:3.7.0;
#kafka
docker pull wurstmeister/kafka:2.13-2.7.1
#kafka-manager
docker pull kafkamanager/kafka-manager:3.0.0.4
#elk
docker pull sebp/elk:7.13.2

安装zk + kafka + kafka-manager

kafka目前版本，运行需要依赖zookeeper，因此我们需要先有一个zookeeper(后续kafka想要移除zookeeper依赖，自建服务注册中心)，接着我们将zookeeper运行起来

#创建一个zookeeper文件存储目录，用于存放elk相关zookeeper信息
mkdir -p /opt/data/zk/elk
#运行zookeeper
docker run --name zookeeper -v /opt/data/zk/elk:/data -p 12181:2181 -e ZOO_LOG4J_PROP="INFO,ROLLINGFILE" -d zookeeper:3.7.0

zookeeper运行完毕后，接着我们运行kafka，其中KAFKA_ADVERTISED_HOST_NAME参数指定的是外网地址(纯内网就是宿主机的ip),KAFKA_ADVERTISED_PORT参数指定的是外网映射端口(宿主机映射端口)

docker run -d --name kafka -p 19092:9092 --link zookeeper:zk --env KAFKA_BROKER_ID=999 --env HOST_IP=0.0.0.0 --env KAFKA_ZOOKEEPER_CONNECT=zk:2181 --env KAFKA_ADVERTISED_HOST_NAME=121.43.231.10 --env KAFKA_ADVERTISED_PORT=19092 --volume /etc/localtime:/etc/localtime wurstmeister/kafka:2.13-2.7.1
#进入kafka容器内，开启kafka的监听者 用于接受消息测试
docker exec -it kafka sh;
cd opt/kafka_2.13-2.7.1/bin/;
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic demo --from-beginning
#再去启动一个窗口，进入容器，开启kafka发送者发送消息
docker exec -it kafka sh;
cd opt/kafka_2.13-2.7.1/bin/;
./kafka-console-producer.sh --broker-list localhost:9092 --topic demo;
#用发送者发几个消息测试一下，看看监听者是否收到消息，确认无误后，退出容器 ctrl + c  exit

最后启动kafka-manager绑定zk和kafka，用于web页面查看

docker run -d --name kafka-manager --link zookeeper:zk --link kafka:kafka --env ZK_HOSTS=zk:2181 -p 29000:9000 kafkamanager/kafka-manager:3.0.0.4

我们去访问外网开放的地址，http://192.168.1.171:29000，初始化看到的界面是空白的，这个时候我们需要手动add一下:

添加完成后，再去list查看，如下:

说明当前我们从容器zk上找到了kafka的信息，接着点进去，查看一下topic，看看是否有刚刚发送消息的demo

至此，说明前面的zk + kafka + kafak-manager都查找到了

注:

kafka-manager的github下载地址: https://github.com/yahoo/CMAK

elk + ik分词器插件文件配置

我们在构建es的时候，会配置ik中文分词器搭配使用，因此，我们在确定了elk镜像每个中间件的版本以后，我们需要提前准备下载对应版本的ik分词器，在github的下载地址如下:

https://github.com/medcl/elasticsearch-analysis-ik/releases

将对应的

下载到本地，首先在服务器上创建目录，然后上传到目录下(/usr/local/bin/elk)

mkdir -p /usr/local/bin/elk;
mkdir -p /usr/local/bin/elk/logstash;

需要注意的是，我们需要将其解压，然后文件夹改名为ik目录

cd /usr/local/bin/elk;
unzip elasticsearch-analysis-ik-7.13.2.zip;
mv elasticsearch-analysis-ik-7.13.2 ik/;
rm -rf elasticsearch-analysis-ik-7.13.2.zip;

接着，我们需要编写一个kafka.conf文件，用于在启动elk的时候，将其加载映射到logstash的配置文件中去，让其可以从kafka中找到数据源

cd /usr/local/bin/elk/logstash;
vim kafka.conf;
input {
  kafka {
    bootstrap_servers => ["52.82.98.209:9103"]
    group_id => "logstash"
    topics => ["mylog"] 
    add_field => {"channel" => "kafka"}
    codec => json {
    charset => "UTF-8"
    }
  }
}
filter {
}
output {
  elasticsearch {
    hosts => "localhost:9200"
    index => "mylog"
  }
  stdout {
  }
}

将其配置修改为对应kafka的外网ip端口，以及指定的topic，一切做好后，我们就可以正式配置elk了，不过需要注意的是，我们之前下载的镜像—sebp/elk:7.13.2里面默认不是自带ik插件的，因此，我们需要基于这个镜像，以及我们的ik目录，重新构建一个新的elk镜像，使其内部拥有对中文分词过滤的能力，因此我们需要在 /usr/local/bin/elk目录下编写Dockerfile文件

cd /usr/local/bin/elk;
vim Dockerfile;
FROM sebp/elk:7.13.2
ADD ik/ /opt/elasticsearch/plugins/ik/
RUN chown -R elasticsearch:elasticsearch /opt/elasticsearch/plugins/ik/

接着我们进行编译构建新的镜像，语法docker build -t 镜像名:tag 目录，这里我们是在当前目录的Dockerfile文件，目录直接为.

docker build -t elk-ik:7.13.2 .

等待镜像构建完成即可，需要注意的是，在运行elk之前，我们需要在宿主机配置一下vm.max_map_count参数，否则，运行elk以后，会出现max virtual memory areas vm.max_map_count [65530] is too low, increase to错误，主要是因为elasticsearch需要需要的jvm内存和文件map映射有大小规定，因此我们需要在宿主机上先执行 sudo sysctl -w vm.max_map_count=262144

接着我们就可以使用当前镜像，而不是之前的elk镜像了，然后我们开始启动elk

docker run -d -v /usr/local/bin/elk/logstash:/etc/logstash/conf.d -p 15044:5044 -p 15601:5601 -p 19200:9200 -p 19300:9300 -p 15044:5044 -p 19600:9600 --link kafka:kafka -e ES_MIN_MEM=128m -e ES_MAX_MEM=1024m -e ES_JAVA_OPTS="-Xms1g -Xmx1g" -it --name elk-dev elk-ik:7.13.2

启动完成后，我们可以输入 http://192.168.1.72:15601查看kibana控制台，当然，为了验证ik分词器效果，我们需要用curl或者postman，先发个测试的数据，查看分词效果:

curl http://localhost:9200/_analyze -X POST -H 'Content-
Type:application/json' -d '{"text":"test elasticsearch 测试分词效
果","analyzer": "ik_smart"}'

如果看到结果:

{
"tokens": [
 {
"token": "test",
"start_offset": 0,
"end_offset": 4,
"type": "ENGLISH",
"position": 0
},
{
"token": "elasticsearch",
"start_offset": 5,
"end_offset": 18,
"type": "ENGLISH",
"position": 1
},
{
"token": "测试",
"start_offset": 19,
"end_offset": 21,
"type": "CN_WORD",
"position": 2
},
{
"token": "分词",
"start_offset": 21,
"end_offset": 23,
"type": "CN_WORD",
"position": 3
},
{
"token": "效果",
"start_offset": 23,
"end_offset": 25,
"type": "CN_WORD",
"position": 4
}
]
}

说明ik分词已经生效，接着，我们可以再次进入kafka容器内，发送几条消息到elk监听的topic上，测试一下，是否能监测到数据

docker exec -it kafka sh;
cd opt/kafka_2.13-2.7.1/bin/;
./kafka-console-producer.sh --broker-list localhost:9092 --topic mylog;

最终效果大概如下，即说明elk搭建完成:

扩展

只保留10天的elk数据日志

vim delelk.sh;
#!/bin/bash
#定时清除elk索引，10天
DATE=date -d "10 days ago" +%Y.%m.%d
curl -XDELETE "http://127.0.0.1:9400/*-${DATE}"

接着需要用的时候 sh delelk.sh即可