1.前言

ELK开源实时日志分析平台。ELK是Elasticsearch,Logstash,Kibana 的缩写。
Elasticsearch：是个开源分布式搜索引擎,简称ES
Logstash：是一个完全开源的工具，可以对日志进行收集，过滤，存储到ES
Kibana: 也是一个开源和免费的工具,这里主要用作ES的可视化界面工具，用于查看日志。

2.搭建ES

调高jvm线程数限制
vim /etc/sysctl.conf
加上一行
vm.max_map_count=262144
执行生效
sysctl –p

拉取镜像

拉取镜像，指定版本号
 docker pull elasticsearch:7.13.2

新建es.yml配置文件并上传到主机目录

 http.host: 0.0.0.0
 #跨域
 http.cors.enabled:  true 
 http.cors.allow-origin:  "*"

启动es

docker run -d -p 9200:9200 -p 9300:9300 --name es -e ES_JAVA_OPTS="-Xms128m -Xmx256m" elasticsearch:7.13.2
#9200是对外端口，9300是es内部通信端口
#-e ES_JAVA_OPTS="-Xms128m -Xmx256m" 限制内存,最小128,最大256

在浏览器打开,ip:9200看到es信息，就成功了，启动可能需要一点时间，要等一会。 image.png

3.搭建Kibana

安装kibana

#如果没有镜像会自动到公共库拉取再启动，-e 环境配置指向es的地址
docker run -p 5601:5601 -d -e ELASTICSEARCH_URL=http://172.17.0.1:9200 -e ELASTICSEARCH_HOSTS=http://172.17.0.1:9200 kibana:7.13.2  

在浏览器输入ip:5601，能看到kibana的信息，说明成功了

image.png

4.kafka搭建

1)前置需要先安装zookeeper

#-v /etc/localtime:/etc/localtime把本机的时间挂载进docker，让docker同步主机的时间
docker run -d --name zookeeper -p 2181:2181 -v /etc/localtime:/etc/localtime zookeeper

2)安装kafka

docker run  -d --name kafka -p 9092:9092 -e KAFKA_BROKER_ID=0 -e KAFKA_ZOOKEEPER_CONNECT=主机ip:2181 -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://主机ip:9092 -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 -t wurstmeister/kafka
#KAFKA_BROKER_ID:kafka节点Id,集群时要指定
#KAFKA_ZOOKEEPER_CONNECT：配置zookeeper管理kafka的路径,内网ip
#KAFKA_ADVERTISED_LISTENERS：如果是外网的，则外网ip，把kafka的地址端口注册给zookeeper,将告诉 zookeeper 自己的地址为 XXXX，当消费者向 zookeeper 询问 kafka 的地址时，将会返回该地址
#KAFKA_LISTENERS： 配置kafka的监听端口

用kafka tool连接kafka看能否成功，下载地址：https://www.kafkatool.com/download.html

5.logstash搭建

新建文件logstash.yml

http.host: "0.0.0.0"
xpack.monitoring.elasticsearch.hosts: [ "http://172.17.0.1:9200" ]

新建文件logstash.conf

input {
     kafka {
      topics => "logkafka" 
      bootstrap_servers => "172.17.0.1:9092"  # 从kafka的leader主机上提取缓存
      codec => "json"  # 在提取kafka主机的日志时，需要写成json格式
            }
}
output {
  elasticsearch {
    hosts => ["172.17.0.1:9200"]
    index => "logkafka" #采集到es的索引名称
    #user => "elastic"
    #password => "changeme"
  }
}

把这两个文件放到主机上
安装logstash

docker run --rm -it --privileged=true -p 9600:9600  -d -v /myfile/logstash/logstash.conf:/usr/share/logstash/pipeline/logstash.conf -v /myfile/logstash/log/:/home/public/  -v /myfile/logstash/logstash.yml:/usr/share/logstash/config/logstash.yml logstash:7.13.2
#把上面的logstash.conf和logstash.yml挂载到docker下

安装完成在页面打开id:9600看到logstash信息 image.png

6.测试结果

往kafka的topics名为logkafka写消息，这里用的是.net程序写的,参考.NET5读写kafka:https://www.jianshu.com/p/e628eb6b911b

image.png
写了6条消息，然后打开kibana image.png
image.png
这里看到es已经产生了logkafka的索引，里面有6条消息,然后可以建一个索引匹配规则来查看索引里面的数据 image.png
打开面板界面 image.png
image.png
到这里ELK+kafka的搭建就完成了!