Elk集群安装+配置（Elasticsearch+Logstash+Filebeat+Kafka+zookeeper+Kibana）_logstash1218端口

Elk集群安装+配置（Elasticsearch+Logstash+Filebeat+Kafka+zookeeper+Kibana）

一、部署环境

1. 基础环境：

2.机器环境：elk服务器4台,业务服务器65台

二、安装jdk

解压

sudo tar -zxvf jdk-8u151-linux-x64.tar.gz -c /usr/local/etc/  

• 环境变量

sudo vi /etc/profile  
etc/profile文件末尾增加环境变量  
export JAVA\_HOME=/usr/local/etc/jdk1.8.0\_151  
export JRE\_HOME=${JAVA\_HOME}/jre  
export CLASSPATH=.:${JAVA\_HOME}/lib:${JRE\_HOME}/lib  
export PATH=$PATH:${JAVA_HOME}/bin  

• 使环境变量生效

source /etc/profile  

• 查看

java -version  

三、安装elasticsearch

1. 软件下载
   链接:elasticsearch-5.5.0.zip
2. 解压文件到指定目录
   /usr/local/etc
3. 修改配置文件

cluster.name: 66kuaiche  #集群名称，同一个集群的标识.  
node.name: es1 #节点名称  
node.master: true #允许一个节点可以成为一个master节  
node.data: true #允许该节点存储数据(默认开启)  
network.host: X.X.X.109 #绑定监听IP  
http.port: 9200 #设置对外服务的http端口  
transport.tcp.port: 9300 # 设置节点间交互的tcp端口  
discovery.zen.ping.unicast.hosts: \["X.X.X.109:9300","X.X.X.108:9300","X.X.X.105:9300"\] # 这是一个集群中的主节点的初始列表,当节点(主节点或者数据节点)启动时使用这个列表进行探测  
node.max\_local\_storage_nodes: 2 # 多个节点可以在同一个安装路径启动  
discovery.zen.minimum\_master\_nodes: 2 # 设置这个参数来保证集群中的节点可以知道其它N个有master资格的节点.默认为1,对于大的集群来说,可以设置大一点的值(2-4)  
http.cors.enabled: true       #允许跨域访问，head使用  
http.cors.allow-origin: "*"   #允许跨域访问，head使用  

1. 启动命令 前台运行：
   ./elasticserch
   后台台运行：
   ./elasticserch -d

2. 验证
   浏览器输入：X.X.X.109:9200

    返回结果如下：

    {  
      "name" : "es1",  
      "cluster_name" : "66kuaiche",  
      "cluster_uuid" : "jHr4yAzQQn2NnzAkakj-1A",  
      "version" : {  
        "number" : "5.5.0",  
        "build_hash" : "260387d",  
        "build_date" : "2017-06-30T23:16:05.735Z",  
        "build_snapshot" : false,  
        "lucene_version" : "6.6.0"  
      },  
      "tagline" : "You Know, for Search"  
    }  

1. 调整jvm内存
   #默认是1g官方建议对jvm进行一些修改，不然很容易出现OOM,参考官网改参数配置最好不要超过内存的50%
   cd /config/jvm.options
   -Xms4g
   -Xmx4g
2. es安装过程遇到的坑
   elasticsearch.yml文件不生效或读取不到
   -原因：该文件的执行权限不够， sudo chmod +x elasticsearch.yml

• 启动elasticsearch时报错,
  main ERROR Could not register mbeans java.security.AccessControlException: access denied (“javax.management.MBeanTrustPermission” “register”)
  -原因：elasticearch不能以root权限启动，改变elasticsearch文件夹所有者到非root用户，sudo chown -R centos:centos elasticsearch

• ERROR: bootstrap checks failed
  max virtual memory areas vm.max_map_count [65530] is too low, increase to at least [262144]
  -原因：堆内存过低，需设置内核参数
  -临时修改：sudo sysctl -w vm.max_map_count=262144
  -永久修改：：修改/etc/sysctl.conf 文件
  添加： vm.max_map_count=262144
  执行：sysctl -p

• max file descriptors [4096] for elasticsearch process is too low, increase to at least [65536]
  -原因：elasticsearch用户拥有的可创建文件描述的权限太低，需设置资源参数，至少需要65536；
  #切换到root用户修改：
  sudo -s
  vim /etc/security/limits.conf
  在最后面追加下面内容
  *** hard nofile 65536
  *** soft nofile 65536
  *** 是启动ES的用户，如：centos
  修改文件成功，仍报too low，是因为没有使用sudo -s切换到root权限，而是使用sudo vi limit,conf修改的配置文件

• main ERROR RollingFileManager (/usr/local/etc/elasticsearch-5.5.0-slave1/logs/66kuaiche.log)
  java.io.FileNotFoundException: /usr/local/etc/elasticsearch-5.5.0-slave1/logs/66kuaiche.log (权限不够)
  java.io.FileNotFoundException: /usr/local/etc/elasticsearch-5.5.0-slave1/logs/66kuaiche.log (权限不够)
  -原因：es在生成日志时需要写文件的权限，但es又不能以root权限启动，所以需要将log文件或整个es文件夹赋予当前用户的权限
  sudo chown -R elastic:elastic elasticsearch-5.5.0

• 搭建elsticsearch集群，报错with the same id but is a different node instance
  -复制elsticsearch将其文件夹下的data文件夹一并复制了，data文件夹下已经产生了data历史数据，删除即可

• 填坑问题（查了好几天）在一开始使用Tcp的方式传输日志到logstash ，发现有数据丢失的问题
  -原因：业务服务器的时间和logstash服务器的时间不同步，在开发环境服务器X.203.115.115上，时间比标准时间慢5分钟
  -查看本服务器时间：date
  -查看标准时间：rdate time-b.nist.gov (time-b.nist.gov为是美国标准技术院的时间自动同步服务器的域名）
  -同步标准时间到本服务器：rdate -s time-b.nist.gov

四、安装logstash

1. 软件下载
   链接:logstash-5.5.0.zip
2. 解压文件到指定目录
   /usr/local/etc
   3 启动

• 常用启动参数：

1. 配置文件及语法

• 区域
    Logstash通过{}来定义区域，区域内可以定义插件，一个区域内可以定义多个插件，如下：

  input {
    stdin {
    }
    beats {
        port => 5044
    }
}

• 数据类型

• 字段
  Logstash数据流中的数据被称之为Event对象，Event以JSON结构构成，Event的属性被称之为字段，如果你像在配置文件中引用这些字段，只需要把字段的名字写在中括号[]里就行了，如[type]，对于嵌套字段每层字段名称都写在[]里就可以了，比如：[tags][type]；除此之外，对于Logstash的arrag类型支持下标与倒序下表，如：[tags][type][0],[tags][type][-1]。

• 运算语法

input  
  file {  
    path => \["/var/log/*.log", "/var/log/message"\]  
    type => "system"  
    start_position => "beginning"  
  }  
}

• Beats监听插件
    Beats插件用于建立监听服务，接收Filebeat或者其他beat发送的Events

input {  
    beats {  
        port => 5044  
    }  
}  

• TCP监听插件
    TCP插件有两种工作模式，“Client”和“Server”，分别用于发送网络数据和监听网络数据。

tcp {  
    port => 41414  
}

• kafka 输入插件

kafka {  
        bootstrap_servers => \["X.X.X.109:9092,X.X.X.108:9092,X.X.13  
0.105:9092"\]  
        topics => \["ucenter-dev-kafka"\]  
        auto\_offset\_reset => "latest"  
    codec => "json"  
    }  

1. 过滤插件
   grok正则捕获
     grok 是Logstash中将非结构化数据解析成结构化数据以便于查询的最好工具，非常适合解析syslog logs，apache log， mysql log，以及一些其他的

 web log  
input  
  file {  
    path => \["/var/log/*.log", "/var/log/message"\]  
    type => "system"  
    start_position => "beginning"  
  }  
}  

• date时间处理插件
    该插件用于时间字段的格式转换，比如将“Apr 17 09:32:01”（MMM dd HH:mm:ss）转换为“MM-dd HH:mm:ss”。通常情况下，Logstash会为自动给Event打上时间戳，但是这个时间戳是Event的处理时间（主要是input接收数据的时间），和日志记录时间会存在偏差（主要原因是buffer），我们可以使用此插件用日志发生时间替换掉默认是时间戳的值。
  （天坑之一）#match匹配时间的格式必须和输入流格式严格匹配，特别要注意表达式前后的空格

date {  
      match => \['time', ' yyyy-MM-dd HH:mm:ss.SSS '\]  
      target => '@timestamp'  
    }  

• ruby语法
    ruby filter可以说是Logstash中最强大的一个filter了，通过自己编写ruby代码可以灵活地实现各种功能
    #ruby插件有两个属性，一个init 还有一个code
  #init属性是用来初始化字段的，你可以在这里初始化一个字段，无论是什么类型的都可以，这个字段只是在ruby{}作用域里面生效。

ruby {  
        code =>"  
        arr=event.get('message').split('-|')  
        length=arr.length  
        if length ==7  
            event.set('logId',arr\[0\]\[-32..-1\])  
            event.set('module',arr\[1\])  
            event.set('time',arr\[2\])  
            event.set('level',arr\[3\])  
            event.set('thread',arr\[4\])  
            event.set('class',arr\[5\])  
            event.set('msg',arr\[6\])  
        else  
            event.set('msg',event.get('message'))  
        end  
        "  
       remove\_field => \['\_id','input_type','tags','message','beat','offset'\]  
    }  

输出插件
ElasticSearch输出插件
  用于将事件信息写入到Elasticsearch中

elasticsearch {  
        hosts => \["X.X.X.109:9200","X.X.X.108:9200","X.X.X.105:9200"\]  
        index => "truck-ka-prod-%{+YYYY.MM.dd}"  
        manage_template => true  
        template_name => "filebeat-tem"  
        template_overwrite => true  
        template => "/usr/local/etc/logstash-1/template/filebeat_template.json"  
    }  

五、安装kibana

1. 软件下载

• 链接:kibana-5.5.0-linux-x86_64.tar.gz
• 解压文件到指定目录
  /usr/local/etc
• 修改配置文件

server.port: 9000 # kibana 对外访问端口  
server.host: "X.X.X.100"  # kibana 对外访问ip  
elasticsearch.url: "http://X.X.X.100:9200" # kibana 连接es的ip地址，其中本项目中es 集群有4哥几点，3台数据/主节点，1台搜索节点，kibana连接搜索节点  

• 启动命令
  前台运行：
  ./kibana
  后台台运行：
  ./kibana &
• 停止：
  ps -ef|grep node kill 相应的进程
• 验证
  X.X.X.100:9000

六、安装kafka

1. 软件下载
   链接:kafka_2.11-0.10.2.1.tgz
2. 解压文件到指定目录
   /usr/local/etc
3. 修改配置文件

broker.id=1 #当前机器在集群中的唯一标识，和zookeeper的myid性质一样  
delete.topic.enable=true #如果kafka启动之前没有配置delete.topic.enable=true，删除topic只会标记为marked for deletion，加上配置，重启kafka，之前的topic就真正删除了  
auto.create.topics.enable=false #是否自动创建topic,如果kafka中没有topic，- 自动创建topic  
listeners=PLAINTEXT://X.X.X.109:9092 #kafka监听  
advertised.listeners=PLAINTEXT://X.X.X.109:9092  
num.network.threads=3 #这个是borker进行网络处理的线程数  
num.io.threads=8 #这个是borker进行I/O处理的线程数  
socket.send.buffer.bytes=102400  
socket.receive.buffer.bytes=102400  
socket.request.max.bytes=104857600  
log.dirs=/usr/local/etc/kafka/log  #消息存放的目录，这个目录可以配置为“，”逗号分割的表达式，上面的num.io.threads要大于这个目录的个数这个目录，如果配置多个目录，新创建的topic他把消息持久化的地方是，当前以逗号分割的目录中，那个分区数最少就放那一个  
num.partitions=3 #默认的分区数，一个topic默认1个分区数  
num.recovery.threads.per.data.dir=1  
log.retention.hours=168 #默认消息的最大持久化时间，168小时，7天  
log.segment.bytes=1073741824 #这个参数是：因为kafka的消息是以追加的形式落地到文件，当超过这个值的时候，kafka会新起一个文件  
log.retention.check.interval.ms=300000 #每隔300000毫秒去检查上面配置的log失效时间  
zookeeper.connect=X.X.X.109:12181,X.X.X.108:12181,X.X.X.105:12181  #设置zookeeper的连接端口  
zookeeper.connection.timeout.ms=6000  #设置zookeeper的连接超时时间  

1. 启动命令
   ./kafka-server-start.sh -daemon ../config/server.properties
2. 验证启动成功失败
   执行命令jps查看kafka进程

七、安装zookeeper

1. 软件下载
   链接:zookeeper-3.4.10.tar.gz
2. 解压文件到指定目录
   /usr/local/etc
3. 修改配置文件
   a. 配置zoo.cfg
   -　将/conf目录下的zoo-sample.cfg文件,复制并命名为zoo.cfg

tickTime=2000  
initLimit=10 #对于从节点最初连接到主节点时的超时时间，单位为tick值的倍数  
syncLimit=5 #对于主节点与从节点进行同步操作时的超时时间，单位为tick值的倍数  
dataDir=/usr/local/etc/zookeeper/zkdata #用于配置内存数据库保存的模糊快照的目录。即刚刚创建的data文件夹就是在此目录中,文件信息都存放在data目录下  
clientPort=12181 #表示客户端所连接的服务器所监听的端口号，默认是2181。即zookeeper对外提供访问的端口号  
server.1=X.X.X.109:12888:13888  
server.2=X.X.X.108:12888:13888  
server.3=X.X.X.105:12888:13888  
#server.1 这个1是服务器的标识也可以是其他的数字， 表示这个是第几号服务器，用来标识服务器，这个标识要写到快照目录下面myid文件里  
#第一个端口是master和slave之间的通信端口，默认是2888，第二个端口是leader选举的端口，集群刚启动的时候选举或者leader挂掉之后进行新的选举的端口默认是3888  

b. 配置zoo.cfg
-　在三个linux机器上刚刚创建的data目录下新建一个myid文件并打开，vi myid命令，依次在3台机器上输入1、2、3，保存退出。这里的1,2,3是与server1,2,3相对应的

\[centos@supply-41b-6 zkdata\]$ more myid  

c. 配置环境变量
-　vi etc/profile 输入下面参数，保存退出，source profile命令使修改生效

export ZOOKEEPER_HOME=/usr/local/etc/zookeeper/zookeeper-3.4.10/  
export PATH=$PATH:$ZOOKEEPER\_HOME/bin:$ZOOKEEPER\_HOME/conf  

4.启动命令
./zkServer.sh start
5.验证
./zkServer.sh status

• 显示

ZooKeeper JMX enabled by default  
Using config: /usr/local/etc/zookeeper/zookeeper-3.4.10/bin/../conf/zoo.cfg #配置文件  
Mode: follower #是否为leader  

7.连接到zookeeper
./zkCli.sh -server X.X.X.108:12181

八、安装filebeat

1.软件下载
链接:filebeat-5.5.0-linux-x86_64.tar.gz 2.解压文件到指定目录
/usr/local/etc
3.修改配置文件

\# paths指定要监控的日志  
      paths:  
        \- /var/log/xxx.log  
        
\# 输出到logstash 配置  
output.logstash:  
  # The Logstash hosts  
  hosts: \["x.x.x.109:1091","x.x.x.108:1091","x.x.x.105:1091"\]  
  worker: 3  
  loadbalance: true  
    
\# 输出到kafka 配置  
output.kafka:  
  #hosts: \["x.x.x.109:9092","x.x.x.108:9092","x.x.x.105:9092"\]  
  topic: xxxx-kafka  
  required_acks: 1  

4.启动命令
sudo nohup ./filebeat -e -c beat-dev.yml >/dev/null 2>&1 &
5.验证启动成功失败
ps -ef|grep beat

九、安装ik分词插件

1.软件下载

• ik中文分词链接:elasticsearch-analysis-ik-5.5.0.zip
• 拼音分词链接:elasticsearch-analysis-pinyin-5.5.0.zip

2.解压文件到指定目录

• ik中文分词：/usr/local/etc/elasticsearch/plugins/ik
• ik拼音分词：/usr/local/etc/elasticsearch/plugins/pinyin

3.重启es

十、安装xpack

1.软件下载
5.5版本的xpack插件是收费的，需要安装破解版的，破解教程后续完善
安装命令
安装前，需要关闭elasticsearch和kibana，否则启动会报错
2.es安装xpack(离线)：./elasticsearch-plugin install file:usr/local/etc/package/x-pack-5.5.0.zip
[elastic@supply-7 bin]$ ./elasticsearch-plugin  install file:usr/local/etc/package/x-pack-5.5.0.zip

-\> Downloading file:usr/local/etc/package/x-pack-5.5.0.zip  
\[=================================================\] 100%
\* java.io.FilePermission \\\.\\pipe\\* read,write  
\* java.lang.RuntimePermission accessClassInPackage.com.sun.activation.registries  
\* java.lang.RuntimePermission getClassLoader  
\* java.lang.RuntimePermission setContextClassLoader  
\* java.lang.RuntimePermission setFactory  
\* java.security.SecurityPermission createPolicy.JavaPolicy  
\* java.security.SecurityPermission getPolicy  
\* java.security.SecurityPermission putProviderProperty.BC  
\* java.security.SecurityPermission setPolicy  
\* java.util.PropertyPermission * read,write  
\* java.util.PropertyPermission sun.nio.ch.bugLevel write  
\* javax.net.ssl.SSLPermission setHostnameVerifier  
See http://docs.oracle.com/javase/8/docs/technotes/guides/security/permissions.html  
for descriptions of what these permissions allow and the associated risks.
 
Continue with installation? \[y/N\]y  
 
This plugin launches a native controller that is not subject to the Java  
security manager nor to system call filters.
 
Continue with installation? \[y/N\]y  
-\> Installed x-pack  

3.kibana安装xpack(离线)：bin/kibana-plugin install

4.效果图：
1.es节点图，图中带星号的es2为master节点，es4为数据搜索节点

2.kibana效果图