开源日志项目（java,zookeeper,kafka,fluentd,gohangout,clickhouse,grafana）

记一次日志项目从日志采集，分布式协调，中间件消费，到数据库存储管理，到图形化界面展示

1 安装java环境

官网下载想要的java版本，网址为：https://www.oracle.com/technetwork/java/javase/downloads/index.html

我下载的版本为jdk-12.0.1_linux-x64_bin.tar

创建目录

#mkdir -p /usr/local/java

解压到/usr/local/java目录下，

#tar -zvxf jdk-12.0.1_linux-x64_bin.tar

配置环境变量

#vi /etc/profile

export KAFKA_HOME=/app/kafka

export JAVA_HOME=/usr/local/java/jdk-12.0.1

export JRE_HOME=/usr/local/java/jdk-12.0.1

export CATALINA_BASE=/usr/local/tomcat

export CATALINA_HOME=/usr/local/tomcat

export CATALINA_TMPDIR=/usr/local/tomcat/temp

export CLASSPATH=/usr/local/tomcat/bin/bootstrap.jar:/usr/local/tomcat/bin/tomcat-juli.jar

export PATH=$PATH:$JAVA_HOME/bin:$CATALINA_HOME/bin:/app/mongodb/bin:$KAFKA_HOME/bin

# source /etc/profile

验证是否安装成功

#java --version

java 12.0.1 2019-04-16

Java(TM) SE Runtime Environment (build 12.0.1+12)

Java HotSpot(TM) 64-Bit Server VM (build 12.0.1+12, mixed mode, sharing)

安装jdk成功

2安装zookeeper

zookeeper集群的安装,准备三台服务器

server1:192.168.72.134

server2:192.168.72.135

server3:192.168.72.136

1)下载zookeeper

到http://zookeeper.apache.org/releases.html去下载最新版本，zookeeper-3.4.6.tar.gz

2)安装zookeeper

先在服务器server分别执行a-c步骤

a)解压

#tar -zxvf zookeeper-3.4.6.tar.gz

创建目录 mkdir /app

#cp -r zookeeper-3.4.6 /app/zookeeper

b）配置

将conf/zoo_sample.cfg拷贝一份命名为zoo.cfg，也放在conf目录下

dataDir,dataLogDir目录需要提前创建出来

#mkdir -p /app/zookeeper/zkdata

mkdir -p /app/zookeeper/logs

c ）配置myid 文件

vi myid

分别在server1，2，3上的配置文件里写1，2，3

3）依次启动server1，server2，server3

/app/zookeeper/bin/zkServer.sh start

4)检查集群状态

/app/zookeeper/bin/zkServer.sh status ,1台leader,2台follower，说明集群启动成功。

3安装kafka

1)下载kafka, http://kafka.apache.org/downloads

2)解压缩

tar -zvxf kafka_2.11-2.3.0.tgz

cp -r kafka_2.11-2.3.0 /app/kafka

3)修改配置文件

vi /app/kafka/config/server.properties ,server2,3, broker.id 分别改为2，3 ，其余配置项都一样

log.dirs 地址可以先不用创建，在首次启动kafka的时候，会自动进行创建。

zookeeper.connect 配置成zookeeper 集群的IP，端口

4）启动broker ，分别启动server1，2，3
这个是在启动broker之前需要保证zookeeper集群是运行着的,查看zookeeper集群状态

# /app/kafka/bin/kafka-server-start.sh /app/kafka/config/server.properties

如果没有出现日志错误，就使用下面的命令：

　# nohup /app/kafka/bin/kafka-server-start.sh /app/kafka/config/server.properties > logs/server-start.log 2>&1 &

　　其中，server-start.log是自己写的一个log文件，在原有的文件logs下面是没有的。

检验是否成功，查看进程ps -ef | grep kafka 或者jps

4安装fluentd

fluentd是一款优秀的开源日志收集工具，td-agent 为稳定的，分布式的fluentd ,且易于安装。

1）预安装

a)Set up NTP，详细https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/set-time.html

b)Increase Max # of File Descriptors

c) Optimize Network Kernel Parameters

2)安装td-agent , CentOS and RHEL 6, 7 64bit are currently supported.

# curl -L https://toolbelt.treasuredata.com/sh/install-redhat-td-agent3.sh | sh

3)修改td-agent的配置文件

@id kafka.access1

@type kafka_buffered

@log_level info

include_tag_key true

brokers 192.168.72.134:9092,192.168.72.135:9092,192.168.72.136:9092 #kafka brokers的IP，端口号

default_topic demsg_sys #把日志采集到kafka中，topic 名称为demsg_sys

output_data_type "json"

output_include_tag true

output_include_time true

# get_kafka_client_log true

# max_send_retries 3

@type file

path /var/log/td-agent/buffers/dmesg_2019.buffer #自己配置路径

flush_mode interval

retry_type exponential_backoff

flush_thread_count 2

flush_interval 5s

retry_forever

retry_max_interval 30

chunk_limit_size 2M

queue_limit_length 8

overflow_action block

</buffer>

</match>

@id dmesg_2019.log

@type tail #从日志文件的最后追加的部分会采集到

time_key time_local

path /var/log/dmesg #需要采集的日志来源文件路径

pos_file /var/log/td-agent/dmesg_2019.log.pos #自己配置路径

tag dmesg_2019 #与开头的<match *** >同步

format /^(?<message>[^.*].*)/ #将日志格式化成一条消息

read_from_head true

</source>

4）启动td-agent

systemctl start td-agent.service

5）验证日志是否采集到kafka中

打一条日志到/var/log/dmesg文件中

echo "Today is last work day before National Day" >> /var/log/dmesg

在kafka上查看消费情况

./kafka-console-consumer.sh --bootstrap-server 192.168.72.134:9092,192.168.72.135:9092,192.168.72.136:9092 --topic demsg_sys

5安装clickhouse

1)首先，您需要添加官方存储库：

sudo yum install yum-utils

sudo rpm --import https://repo.yandex.ru/clickhouse/CLICKHOUSE-KEY.GPG

sudo yum-config-manager --add-repo https://repo.yandex.ru/clickhouse/rpm/stable/x86_64

如果您想使用最新版本，请将stable替换为testing（建议您在测试环境中使用）。

2)然后运行这些命令以实际安装包：

sudo yum install clickhouse-server clickhouse-client

3)启动

可以运行如下命令在后台启动服务：

sudo service clickhouse-server start

可以在/var/log/clickhouse-server/目录中查看日志。

如果服务没有启动，请检查配置文件 /etc/clickhouse-server/config.xml。

你也可以在控制台中直接启动服务：

clickhouse-server --config-file=/etc/clickhouse-server/config.xml

在这种情况下，日志将被打印到控制台中，这在开发过程中很方便。如果配置文件在当前目录中，你可以不指定‘--config-file’参数。它默认使用‘./config.xml’。

你可以使用命令行客户端连接到服务：

clickhouse-client

默认情况下它使用‘default’用户无密码的与localhost:9000服务建立连接。

4)建表，创建一张表存储想要从kafka中存储的字段，加一列date字段。

CREATE TABLE cms.dmesg_sys2 (`date` Date, `message` String, `time` DateTime, `tag` String) ENGINE = MergeTree(date, (message, date), 8192)；

6安装gohangout

利用工具gohangout将kafka内的数据打到clickhouse，详细信息请参阅 https://github.com/childe/gohangout

1）下载编译后二进制文件

https://github.com/childe/gohangout/releases 直接下载

我下载的是最新的 gohangout-linux-x64-dc56981，加上执行权限

chmod +x gohangout-linux-x64-dc56981

2)编辑配置文件

vi config.yml

inputs:

- Kafka:

topic:

demsg_sys: 1 #kafka里的topic名称

codec: json

consumer_settings:

bootstrap.servers: "192.168.72.134:9092,192.168.72.135:9092,192.168.72.136:9092" #kafka brokers的IP，端口

group.id: gohangout.dmesg

filters:

- Date:

location: 'Asia/Shanghai'

src: time

target: 'time'

formats:

- 'UNIX'

outputs:

- Clickhouse:

table: 'cms.dmesg_sys2' #clickhouse里创建的库名，表名

hosts:

- 'tcp://localhost:9000' #clickhouse地址

fields: ['message','time','tag'] #要从kafka订阅的列

username: 'default' #clickhouse 用户名

password: "" #clickhouse default 默认无密码

bulk_actions: 1000

flush_interval: 30

concurrent: 1

3)启动gohangout

./gohangout-linux-x64-dc56981 --config config.yml

4）验证

向日志文件打印一条消息

echo "I am so happy today" >> /var/log/dmesg

在kafka看到消费端已经消费此条消息，