一、环境准备
成都地区优秀IDC服务器托管提供商(创新互联).为客户提供专业的双线服务器托管,四川各地服务器托管,双线服务器托管、多线服务器托管.托管咨询专线：18980820575
1.安装java环境：
yum install java-1.8.0-openjdk* -y
2.添加elk执行用户：
groupadd -g 77 elk useradd -u 77 -g elk -d /home/elk -s /bin/bash elk
3.在 /etc/security/limits.conf 追加以下内容：
elk soft memlock unlimited elk hard memlock unlimited * soft nofile 65536 * hard nofile 131072
4.执行生效
sysctl -p
5.配置主机名
hostnamectl set-hostname monitor-elk echo "10.135.3.135 monitor-elk" >> /etc/hosts
二、服务部署
1.服务端：
1）下载ELK相关的源码包：
wget "https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-5.2.2.tar.gz" wget "https://artifacts.elastic.co/downloads/logstash/logstash-5.2.2.tar.gz" wget "https://artifacts.elastic.co/downloads/kibana/kibana-5.2.2-linux-x86_64.tar.gz" wget "http://mirror.bit.edu.cn/apache/kafka/0.10.2.0/kafka_2.12-0.10.2.0.tgz" wget "http://mirror.bit.edu.cn/apache/zookeeper/zookeeper-3.4.9/zookeeper-3.4.9.tar.gz"
2）创建elk目录，并将以上源码包解压至该目录：
mkdir /usr/local/elk
mkdir -p /data/elasticsearch/
chown -R elk.elk /data/elasticsearch/
mkdir -p /data/{kafka,zookeeper}
mv logstash-5.2.2 logstash && mv kibana-5.2.2-linux-x86_64 kibana && mv elasticsearch-5.2.2 elasticsearch && mv filebeat-5.2.2-linux-x86_64 filebeat && mv kafka_2.12-0.10.2.0 kafka && mv zookeeper-3.4.9 zookeeper
chown -R elk.elk /usr/local/elk/程序目录列表如下：
3）修改以下程序的相应配置文件
①kibana：
[root@monitor-elk ~]# cat /usr/local/elk/kibana/config/kibana.yml |grep -v "^#\|^$" server.host: "localhost" elasticsearch.url: "http://localhost:9200" elasticsearch.requestTimeout: 30000 logging.dest: /data/elk/logs/kibana.log [root@monitor-elk ~]#
②elasticsearch：
[root@monitor-elk ~]# cat /usr/local/elk/elasticsearch/config/elasticsearch.yml |grep -v "^#\|^$" node.name: node01 path.data: /data/elasticsearch/data path.logs: /data/elk/logs/elasticsearch bootstrap.memory_lock: true network.host: 127.0.0.1 http.port: 9200 [root@monitor-elk ~]# /usr/local/elk/elasticsearch/config/jvm.options #修改以下参数 -Xms1g -Xmx1g
③logstash：
[root@monitor-elk ~]# cat /usr/local/elk/logstash/config/logs.yml
input {
#使用kafka的数据作为日志数据源
kafka
{
bootstrap_servers => ["127.0.0.1:9092"]
topics => "beats"
codec => json
}
}
filter {
#过滤数据，如果日志数据里面包含有该IP地址，将会被丢弃
if [message] =~ "123.151.4.10" {
drop{}
}
# 转码，转成正常的url编码，如中文
# urldecode {
# all_fields => true
# }
# nginx access
#通过type来判断传入的日志类型
if [type] == "hongbao-nginx-access" or [type] == "pano-nginx-access" or [type] == "logstash-nginx-access" {
grok {
#指定自定义的grok表达式路径
patterns_dir => "./patterns"
#指定自定义的正则表达式名称解析日志内容，拆分成各个字段
match => { "message" => "%{NGINXACCESS}" }
#解析完毕后，移除默认的message字段
remove_field => ["message"]
}
#使用geoip库解析IP地址
geoip {
#指定解析后的字段作为数据源
source => "clientip"
fields => ["country_name", "ip", "region_name"]
}
date {
#匹配日志内容里面的时间，如 05/Jun/2017:03:54:01 +0800
match => ["timestamp","dd/MMM/yyyy:HH:mm:ss Z"]
#将匹配到的时间赋值给@timestamp字段
target => "@timestamp"
remove_field => ["timestamp"]
}
}
# tomcat access
if [type] == "hongbao-tomcat-access" or [type] == "ljq-tomcat-access" {
grok {
patterns_dir => "./patterns"
match => { "message" => "%{TOMCATACCESS}" }
remove_field => ["message"]
}
geoip {
source => "clientip"
fields => ["country_name", "ip", "region_name"]
}
date {
match => ["timestamp","dd/MMM/yyyy:HH:mm:ss Z"]
target => "@timestamp"
remove_field => ["timestamp"]
}
}
# tomcat catalina
if [type] == "hongbao-tomcat-catalina" {
grok {
match => {
"message" => "^(?\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d{3}) (?\w*) (?.+)"
}
remove_field => ["message"]
}
date {
match => ["log_time","yyyy-MM-dd HH:mm:ss,SSS"]
target => "@timestamp"
remove_field => ["log_time"]
}
}
}
output {
#将解析失败的记录写入到指定的文件中
if "_grokparsefailure" in [tags] {
file {
path => "/data/elk/logs/grokparsefailure-%{[type]}-%{+YYYY.MM}.log"
}
}
# nginx access
#根据type日志类型分别输出到elasticsearch不同的索引
if [type] == "hongbao-nginx-access" {
#将处理后的结果输出到elasticsearch
elasticsearch {
hosts => ["127.0.0.1:9200"]
#指定输出到当天的索引
index => "hongbao-nginx-access-%{+YYYY.MM.dd}"
}
}
if [type] == "pano-nginx-access" {
elasticsearch {
hosts => ["127.0.0.1:9200"]
index => "pano-nginx-access-%{+YYYY.MM.dd}"
}
}
if [type] == "logstash-nginx-access" {
elasticsearch {
hosts => ["127.0.0.1:9200"]
index => "logstash-nginx-access-%{+YYYY.MM.dd}"
}
}
# tomcat access
if [type] == "hongbao-tomcat-access" {
elasticsearch {
hosts => ["127.0.0.1:9200"]
index => "hongbao-tomcat-access-%{+YYYY.MM.dd}"
}
}
if [type] == "ljq-tomcat-access" {
elasticsearch {
hosts => ["127.0.0.1:9200"]
index => "ljq-tomcat-access-%{+YYYY.MM.dd}"
}
}
# tomcat catalina
if [type] == "hongbao-tomcat-catalina" {
elasticsearch {
hosts => ["127.0.0.1:9200"]
index => "hongbao-tomcat-catalina-%{+YYYY.MM.dd}"
}
}
} [root@monitor-elk ~]#
配置正则表达式
[root@monitor-elk ~]# cp /usr/local/elk/logstash/vendor/bundle/jruby/1.9/gems/logstash-patterns-core-4.0.2/patterns/grok-patterns /usr/local/elk/logstash/config/patterns
[root@monitor-elk ~]# tail -5 /usr/local/elk/logstash/config/patterns
# Nginx
NGINXACCESS %{COMBINEDAPACHELOG} %{QS:x_forwarded_for}
# Tomcat
TOMCATACCESS %{COMMONAPACHELOG}
[root@monitor-elk ~]# chown elk.elk /usr/local/elk/logstash/config/patterns4）配置zookeeper：
cp /usr/local/elk/zookeeper/conf/zoo_sample.cfg /usr/local/elk/zookeeper/conf/zoo.cfg
修改配置文件中的数据存储路径
vim /usr/local/elk/zookeeper/conf/zoo.cfg dataDir=/data/zookeeper
备份并修改脚本 /usr/local/elk/zookeeper/bin/zkEnv.sh
修改以下变量的参数
ZOO_LOG_DIR="/data/zookeeper-logs" ZOO_LOG4J_PROP="INFO,ROLLINGFILE"
备份并修改日志配置 /usr/local/elk/zookeeper/conf/log4j.properties
修改以下变量的参数
zookeeper.root.logger=INFO, ROLLINGFILE log4j.appender.ROLLINGFILE=org.apache.log4j.DailyRollingFileAppender# 每天轮转日志
启动zookeeper：
/usr/local/elk/zookeeper/bin/zkServer.sh start
5）配置kafka：
修改配置文件 /usr/local/elk/kafka/config/server.properties 的以下参数
log.dirs=/data/kafka zookeeper.connect=localhost:2181
备份并修改脚本 /usr/local/elk/kafka/bin/kafka-run-class.sh
在“base_dir=$(dirname $0)/.. ”的下一行追加LOG_DIR变量，并指定日志输出路径
LOG_DIR=/data/kafka-logs
创建日志存储目录：
mkdir -p /data/kafka-logs mkdir -p /data/elk/logs chown -R elk.elk /data/elk/logs
启动kafka：
nohup /usr/local/elk/kafka/bin/kafka-server-start.sh /usr/local/elk/kafka/config/server.properties &>> /data/elk/logs/kafka.log &
需要注意的是主机名一定要有配置在/etc/hosts文件中，否则kafka会无法启动
[root@monitor-elk ~]# cat /etc/hosts 127.0.0.1 localhost localhost.localdomain ::1 localhost localhost.localdomain localhost6 localhost6.localdomain6 10.135.3.135 monitor-elk
6）配置supervisor
①安装supervisor：
yum install supervisor -y
设置服务开机自启动(server程序也会一起启动)：
systemctl enable supervisord.service
②修改配置
a.创建日志存储路径：
mkdir -p /data/supervisor chown -R elk.elk /data/supervisor/
b.修改主配置文件 /etc/supervisord.conf
logfile=/data/supervisor/supervisord.log
c.创建elk程序对应的supervisor配置文件，并添加以下配置内容：
[root@monitor-elk ~]# cat /etc/supervisord.d/elk.ini [program:elasticsearch] directory=/usr/local/elk/elasticsearch command=su -c "/usr/local/elk/elasticsearch/bin/elasticsearch" elk autostart=true startsecs=5 autorestart=true startretries=3 priority=10 [program:logstash] directory=/usr/local/elk/logstash command=/usr/local/elk/logstash/bin/logstash -f /usr/local/elk/logstash/config/logs.yml user=elk autostart=true startsecs=5 autorestart=true startretries=3 redirect_stderr=true stdout_logfile=/data/elk/logs/logstash.log stdout_logfile_maxbytes=1024MB stdout_logfile_backups=10 priority=11 [program:kibana] directory=/usr/local/elk/kibana command=/usr/local/elk/kibana/bin/kibana user=elk autostart=true startsecs=5 autorestart=true startretries=3 priority=12 [root@monitor-elk ~]#
③启动supervisor：
systemctl start supervisord
查看程序进程和日志：
ps aux|grep -v grep|grep "elasticsearch\|logstash\|kibana"
tip：
重启配置的单个程序，如：
supervisorctl restart logstash
重启配置的所有程序：
supervisorctl restart all
重载配置（只重启配置变动的对应程序，其他配置未变动的程序不重启）：
supervisorctl update
7）配置nginx
①安装nginx
yum install nginx -y
②配置nginx代理:
[root@monitor-elk ~]# cat /etc/nginx/conf.d/kibana.conf
upstream kibana {
server 127.0.0.1:5601 max_fails=3 fail_timeout=30s;
}
server {
listen 8080;
server_name localhost;
location / {
proxy_pass http://kibana/;
index index.html index.htm;
#auth
auth_basic "kibana Private";
auth_basic_user_file /etc/nginx/.htpasswd;
}
}
[root@monitor-elk ~]# grep listen /etc/nginx/nginx.conf
listen 8000 default_server;
listen [::]:8000 default_server;
[root@monitor-elk ~]#③创建nginx认证：
[root@monitor-elk ~]# yum install httpd -y [root@monitor-elk ~]# htpasswd -cm /etc/nginx/.htpasswd elk New password: Re-type new password: Adding password for user elk [root@monitor-elk ~]# systemctl start nginx [root@monitor-elk ~]# systemctl enable nginx
8）配置ik中文分词：
①安装maven：
wget "http://mirror.bit.edu.cn/apache/maven/maven-3/3.3.9/binaries/apache-maven-3.3.9-bin.tar.gz" tar -zxf apache-maven-3.3.9-bin.tar.gz mv apache-maven-3.3.9 /usr/local/maven echo "export MAVEN_HOME=/usr/local/maven" >> /etc/bashrc echo "export PATH=$PATH:$MAVEN_HOME/bin" >> /etc/bashrc . /etc/bashrc
②编译安装ik（注意下载对应版本）：
wget "https://github.com/medcl/elasticsearch-analysis-ik/archive/v5.2.2.zip" unzip v5.2.2.zip cd elasticsearch-analysis-ik-5.2.2/ mvn package mkdir /usr/local/elk/elasticsearch/plugins/ik cp target/releases/elasticsearch-analysis-ik-5.2.2.zip /usr/local/elk/elasticsearch/plugins/ik/ cd /usr/local/elk/elasticsearch/plugins/ik/ unzip elasticsearch-analysis-ik-5.2.2.zip rm -f elasticsearch-analysis-ik-5.2.2.zip chown -R elk.elk ../ik supervisorctl restart elasticsearch
③创建索引模板：
要使用ik分词，需要在创建指定的索引前（不管是通过命令手动还是logstash配置来创建）先创建索引模板，否则使用默认的模板即可：
cd /usr/local/elk/logstash
创建并编辑文件 logstash.json ，添加以下内容：
{
"order" : 1,
"template" : "tomcatcat-*",
"settings" : {
"index" : {
"refresh_interval" : "5s"
}
},
"mappings" : {
"_default_" : {
"dynamic_templates" : [
{
"string_fields" : {
"mapping" : {
"norms" : false,
"type" : "text",
"analyzer": "ik_max_word",
"search_analyzer": "ik_max_word"
},
"match_mapping_type" : "text",
"match" : "*"
}
}
],
"_all" : {
"norms" : false,
"enabled" : true
},
"properties" : {
"@timestamp" : {
"include_in_all" : false,
"type" : "date"
},
"log_data": {
"include_in_all" : true,
"type" : "text",
"analyzer": "ik_max_word",
"search_analyzer": "ik_max_word",
"boost" : 8
},
"@version" : {
"include_in_all" : false,
"type" : "keyword"
}
}
}
},
"aliases" : { }
}'添加完毕后，执行curl命令创建索引模板
curl -XPUT 'http://localhost:9200/_template/tomcatcat' -d @logstash.json
执行成功后会返回结果 {"acknowledged":true}
④热更新配置：
有些词语ik无法识别分词，如公司名称、服务名称之类
curl -XGET 'http://localhost:9200/_analyze?pretty&analyzer=ik_smart' -d ' 腾讯云'
这时需要自己自定义词库，ik支持分词热更新的方式（不需要重启elasticsearch），每分钟自动检测一次
在nginx根路径下创建一个utf8格式的文本文件 ik.txt ，将自己需要分词的词语写入ik.txt，一行一词：
然后修改/usr/local/elk/elasticsearch/plugins/ik/config/IKAnalyzer.cfg.xml
生效。
