CentOS6.5 部署 ELK_随笔

CentOS6.5 部署 ELK

整体流程：

环境准备：

CentOS release 6.5 (Final)

Elasticsearch-6.4.0.rpm

下载：Past Releases of Elastic Stack Software | Elastic

Kibana-6.4.0-x86_64.rpm

下载：Past Releases of Elastic Stack Software | Elastic

logstash-6.4.0.rpm

下载：Past Releases of Elastic Stack Software | Elastic

Filebeat-6.4.0-x86_64.rpm

下载：Past Releases of Elastic Stack Software | Elastic

kafka_2.1102.0.0.taz

下载：Apache Kafka

Zookeeper-3.4.13.tar.gz

下载：Index of /zookeeper

1.ES安装

rpm -ivh Elasticsearch-6.4.0.rpm

ES不能用root 用户进行启动，需要进行相关设置

（1）创建elasticsearch 用户组

groupadd elasticsearch

(2) 创建用户esuser 并设置密码为 esuser

useradd esuser

passwd esuser

(3) 将用户esuser 添加到 elasticsearch 用户组中

usermod -G elasticsearch esuser

(4) 增加sudo 权限

vi /etc/sudoers

(5) 切换 esuser用户

su esuser

(6) 修改 elasticsearch 文件夹以及内部文件的所属用户为esuser,用户组为elasticsearch （rpm安装方式可以不用）

sudo chown -R esuser:elasticsearch elasticsearch-6.4.0

(7) 修改elasticsearch.yml 文件,rpm安装文件在/etc/elasticsearch中

bootstrap.memory_lock: false

bootstrap.system_call_filter: false

network.host: 0.0.0.0 修改本机地址

http.port:9200 监听端口

（8）修改sysctl.conf 文件，在最后增加 vm.max_map_count=262144,增加完之后，执行sysctl -p 刷新生效

（9）修改 /etc/security/limits.conf,最下面增加如下内容

esuser hard nofile 65536

esuser soft nofile 65536

（10）修改 /etc/security/limits.d/90-nproc.conf,修改如下

* soft nproc 4096

root soft nproc unlimited

(11) 启动 ES ,service elesticsearch start

(12) 访问 http://ip:9200

2.logstash 安装

(1) rpm -vih logstash-6.4.0.rpm

(2) 进入 /etc/logstash/conf.d，然后新建 kafka_logstash.conf 配置文件

input {
  kafka {
    ## app-log-服务名称
    topics_pattern => "app-log-.*"
    bootstrap_servers => "192.168.11.51:9092" ## kafka地址
 codec => json
 consumer_threads => 1 ## 增加consumer的并行消费线程数
 decorate_events => true
    #auto_offset_rest => "latest"
 group_id => "app-log-group"
   }
   
   kafka {
    ## error-log-服务名称
    topics_pattern => "error-log-.*"
    bootstrap_servers => "192.168.11.51:9092" ## kafka地址
 codec => json
 consumer_threads => 1
 decorate_events => true
    #auto_offset_rest => "latest"
 group_id => "error-log-group"
   }
   
}

filter {
  
  ## 时区转换
  ruby {
 code => "event.set('index_time',event.timestamp.time.localtime.strftime('%Y.%m.%d'))"
  }

  if "app-log" in [fields][logtopic]{
    grok {
        ## 表达式,这里对应的是Springboot输出的日志格式
        match => ["message", "[%{NOTSPACE:currentDateTime}] [%{NOTSPACE:level}] [%{NOTSPACE:thread-id}] [%{NOTSPACE:class}] [%{data:hostName}] [%{data:ip}] [%{data:applicationName}] [%{data:location}] [%{data:messageInfo}] ## (''|%{QUOTEDSTRING:throwable})"]
    }
  }

  if "error-log" in [fields][logtopic]{
    grok {
        ## 表达式
        match => ["message", "[%{NOTSPACE:currentDateTime}] [%{NOTSPACE:level}] [%{NOTSPACE:thread-id}] [%{NOTSPACE:class}] [%{data:hostName}] [%{data:ip}] [%{data:applicationName}] [%{data:location}] [%{data:messageInfo}] ## (''|%{QUOTEDSTRING:throwable})"]
    }
  }
  
}

## 测试输出到控制台：
output {
  stdout { codec => rubydebug }
}


## elasticsearch：
output {

  if "app-log" in [fields][logtopic]{
 ## es插件
 elasticsearch {
       # es服务地址
        hosts => ["192.168.11.35:9200"] ##ES地址
        # 用户名密码      
        #user => "esuser"
        #password => "esuser"
        ## 索引名，+ 号开头的，就会自动认为后面是时间格式：
        ## javalog-app-service-2019.01.23 
        index => "app-log-%{[fields][logbiz]}-%{index_time}"
        # 是否嗅探集群ip：一般设置true；http://192.168.11.35:9200/_nodes/http?pretty
        # 通过嗅探机制进行es集群负载均衡发日志消息
        sniffing => true
        # logstash默认自带一个mapping模板，进行模板覆盖
        template_overwrite => true
    } 
  }
  
  if "error-log" in [fields][logtopic]{
 elasticsearch {
        hosts => ["192.168.11.35:9200"]##ES地址    
        #user => "esuser"
        #password => "esuser"
        index => "error-log-%{[fields][logbiz]}-%{index_time}"
        sniffing => true
        template_overwrite => true
    } 
  }
  

}

(3) 启动 logstash

initctl start logstash

3.kibana安装

(1) 安装

rpm -ivh Kibana-6.4.0-x86_64.rpm

（2）修改/etc/Kibana/kibana.yml

server.port:5601 # kibana端口号

server.host: "192.168.1.2" # kibana主机地址

elasticsearch.url: "http://192.168.1.3:9200" #es地址

kibana.index: ".kibana"

(3) 启动kibana,通过 http://ip:5601 进行访问

service kibana start

4.Zookeeper搭建

（1）解压

tar -zxvf Zookeeper-3.4.13.tar.gz

(2) 进入解压好的目录里面的conf目录中，zoo_sample.cfg 是官方的样板文件，复制一份并改名zoo.cfg

cp ./zoo_sample.cfg zoo.cfg

(3) 修改上述配置文件

tickTime=2000

initLimit=10

syncLimit=5

##可以在外部新建一个文件夹后单独指定

dataDir=/mnt/zookeeper/zkdata

##可以在外部新建一个文件夹后单独指定

dataLogDir=/mnt/zookeeper/zkdatalog

clientPort=12181

#默认是2181

# 如果是多个服务器，可以再写server.2 server.3

server.1=192.168.7.100:2888:3888

#server.1 这个1是服务器的标识也可以是其他的数字，表示这个是第几号服务器，用来标识服务器，这个标识要写到快照目录下面myid文件里

#192.168.7.100为集群里的IP地址，第一个端口是master和slave之间的通信端口，默认是2888，第二个端口是leader选举的端口，集群刚启动的时候选举或者leader挂掉之后进行新的选举的端口默认是3888.可以随便修改

配置文件解释：

#tickTime：

这个时间是作为 Zookeeper 服务器之间或客户端与服务器之间维持心跳的时间间隔，也就是每个 tickTime 时间就会发送一个心跳。

#initLimit：

这个配置项是用来配置 Zookeeper 接受客户端（这里所说的客户端不是用户连接 Zookeeper 服务器的客户端，而是 Zookeeper 服务器集群中连接到 Leader 的 Follower 服务器）初始化连接时最长能忍受多少个心跳时间间隔数。当已经超过 5个心跳的时间（也就是 tickTime）长度后 Zookeeper 服务器还没有收到客户端的返回信息，那么表明这个客户端连接失败。总的时间长度就是 5*2000=10 秒

#syncLimit：

这个配置项标识 Leader 与Follower 之间发送消息，请求和应答时间长度，最长不能超过多少个 tickTime 的时间长度，总的时间长度就是5*2000=10秒

#dataDir：

快照日志的存储路径

#dataLogDir：

事物日志的存储路径，如果不配置这个那么事物日志会默认存储到dataDir制定的目录，这样会严重影响zk的性能，当zk吞吐量较大的时候，产生的事物日志、快照日志太多

#clientPort：

这个端口就是客户端连接 Zookeeper 服务器的端口，Zookeeper 会监听这个端口，接受客户端的访问请求。修改他的端口改大点

创建myid文件

#server1

echo "1" > /opt/zookeeper/zkdata/myid

(3)启动服务

进入Zookeeper 的bin目录下

./zkServer.sh start

5.kafka 搭建

（1）解压

tar -zxvf kafka_2.1102.0.0.taz

(2) 进入对应的config 目录，找到 server.properties 并进行修改

broker.id=0

#当前机器在集群中的唯一标识

#开启删除topic的开关

delete.topic.enable=true

这个参数很重要，如果后续你使用命令去删除topic，那么只会把zk上标记的topic删除，但是对实际的kafka上主题保存的数据一点都没有影响，都不会删除。

port=9092

#当前kafka对外提供服务的端口默认是9092 可以自己修改

listeners=PLAINTEXT://:9092

#advertised.listeners=PLAINTEXT://your.host.name:9092

advertised.listeners=PLAINTEXT://192.168.10.14:9092

这个参数非常重要，一定要添加，这个主要是broker（kafka服务器）向生产者和消费者连接使用的，如果你的生产者和消费者是跨主机的存在，一定要把这条配置打开，否则，就算是防火墙关闭了，producer 和 consumer也无法正常通信。

host.name=192.168.7.100

#这个参数默认是关闭的，在0.8.1有个bug，DNS解析问题，失败率的问题。

#log.dirs=/tmp/kafka-logs

log.dirs=/usr/local/src/zookeeper_kafka/kafka/kafkalogs

#消息存放的目录，这个目录可以配置为“，”逗号分割的表达式，上面的num.io.threads要大于这个目录的个数这个目录，如果配置多个目录，新创建的topic他把消息持久化的地方是，当前以逗号分割的目录中，那个分区数最少就放那一个

这个文件夹建议你新建一个单独文件夹来指定

num.partitions=1

#默认的分区数，一个topic默认1个分区数

zookeeper.connect=192.168.7.100:12181

#设置zookeeper的连接端口
（3）启动 kafka

./kafka-server-start.sh -daemon ../config/server.properties

(4) 创建Topic

kafka-topics.sh --zookeeper 192.168.11.111:12181 --create --topic app-log-collector --partitions 1  --replication-factor 1
kafka-topics.sh --zookeeper 192.168.11.111:12181 --create --topic error-log-collector --partitions 1  --replication-factor 1

查看Topic情况

./kafka-topics.sh --zookeeper 192.168.11.111:12181 --list

查看是否已经生成app-log-collector 和 error-log-collector

（5）相关测试方式

手动插入数据

./kafka-console-producer.sh --broker-list 192.168.7.100:9092 --topic capture_test

消费消息

./kafka-console-consumer.sh --bootstrap-server 192.168.7.100:9092 --topic topicName
./kafka-console-consumer.sh --zookeeper 192.168.11.111:12181 --topic topicName
#表示从 latest 位移位置开始消费该主题的所有分区消息，即仅消费正在写入的消息。

从开始位置消费

./kafka-console-consumer.sh --bootstrap-server 192.168.7.100:9092 --from-beginning --topic topicName
./kafka-console-consumer.sh --zookeeper 192.168.11.111:12181 --from-beginning --topic topicName

显示消费key

./kafka-console-consumer.sh --bootstrap-server 192.168.7.100:9092 --property print.key=true --topic topicName
./kafka-console-consumer.sh --zookeeper 192.168.11.111:12181 --property print.key=true --topic topicName
#消费出的消息结果将打印出消息体的 key 和 value。

6.filebeat安装

（1）在需要采集log日志的机器上，安装filebeat

rpm -ivh Filebeat-6.4.0-x86_64.rpm

(2) 进入 /etc/filebeat 修改 filebeat.yml

filebeat.prospectors:

- input_type: log

  paths:
    ## app-服务名称.log, 为什么写死，防止发生轮转抓取历史数据
    - /usr/local/logs/app-collector.log
  #定义写入 ES 时的 _type 值
  document_type: "app-log"
  multiline:
    #pattern: '^s*(d{4}|d{2})-(d{2}|[a-zA-Z]{3})-(d{2}|d{4})'   # 指定匹配的表达式（匹配以 2017-11-15 08:04:23:889 时间格式开头的字符串）
    pattern: '^['                              # 指定匹配的表达式（匹配以 "{ 开头的字符串）
    negate: true                                # 是否匹配到
    match: after                                # 合并到上一行的末尾
    max_lines: 2000                             # 最大的行数
    timeout: 2s                                 # 如果在规定时间没有新的日志事件就不等待后面的日志
  fields:
    logbiz: collector
    logtopic: app-log-collector   ## 按服务划分用作kafka topic
    evn: dev

- input_type: log

  paths:
    - /usr/local/logs/error-collector.log
  document_type: "error-log"
  multiline:
    #pattern: '^s*(d{4}|d{2})-(d{2}|[a-zA-Z]{3})-(d{2}|d{4})'   # 指定匹配的表达式（匹配以 2017-11-15 08:04:23:889 时间格式开头的字符串）
    pattern: '^['                              # 指定匹配的表达式（匹配以 "{ 开头的字符串）
    negate: true                                # 是否匹配到
    match: after                                # 合并到上一行的末尾
    max_lines: 2000                             # 最大的行数
    timeout: 2s                                 # 如果在规定时间没有新的日志事件就不等待后面的日志
  fields:
    logbiz: collector
    logtopic: error-log-collector   ## 按服务划分用作kafka topic
    evn: dev
    
output.kafka:
  enabled: true
  hosts: ["192.168.11.51:9092"]
  topic: '%{[fields.logtopic]}'
  partition.hash:
    reachable_only: true
  compression: gzip
  max_message_bytes: 1000000
  required_acks: 1
logging.to_files: true

(3) 启动filebeat

service start filebeat

ELK的log的java的程序：ELK的Springboot程序-Java文档类资源-CSDN下载

欢迎分享，转载请注明来源：内存溢出

原文地址: http://outofmemory.cn/zaji/5652672.html

CentOS6.5 部署 ELK

发表评论

评论列表（0条）