一、       背景

二、       ELASTICSEARCH方案及架构

三、       部署 ELASTICSEARCH集群

四、       部署 KAFKA

五、       部署 FILEBEAT

六、       部署 LOGSTASH

七、       基于 LOGSTASH的日志文件解析

八、       部署 KIBANA

九、       使用 KIBANA进行日志的查询及展示

 

 

 

一、                    背景

      近年来互联网公司越来越流行使用ELK架构来实现对企业日志数据的集中管理及分析展现，一方面可以对日志进行集中汇总和分析，提供给运维部门进行日常监控告警；另一方面，由于在很多应用的日志里保留了大量重要的业务相关的信息(比如埋点日志)，通过分析这些信息可以提供供给运营部门，风控部门进行深入挖掘使用。

      本实验的目的是基于以上背景，搭建相对简化的环境，完成对几种常见日志的收集、汇总、分析及展现。

 

 

二、                    ElasticSearch方案及架构

 

1 、本方案中用到的组件包括

 

·        filebeat— 日志的收集

·        Kafka— 日志的汇聚

·        Logstash— 日志的分析处理

·        ElasticSearch— 日志的存储

·        Kibana— 日志的查询、分析及展示。

 

2 、架构图

 

 

 

 

 

三、                    部署 ElasticSearch集群

 

ElasticSearch 集群的节点角色及信息如下:

 

节点 IP	角色	说明	部署应用
10.0.0.149	master	ES 集群的master 节点	ES,filebeat
10.0.0.227	data1	ES 集群的数据节点1	ES,filebeat
10.0.0.65	data2	ES 集群的数据节点2	ES,filebeat
10.0.0.53	Kibana	Kibana 节点	Logstash,filebeat,kafka, zookeeper Nginx,Tomcat,MySQL

 

 

1 、创建 VCN 和 VM( 略 ) ，四台虚机配置如下

 

2 、将节点信息写到四个节点的 /etc/hosts 文件中

 

10.0.0.149    master

10.0.0.227    data1

10.0.0.65       data2

10.0.0.53        kibana

 

 

3 、部署 ElasticSearch 集群

 

a 、 登陆master 节点（windows 电脑请使用ssh 客户端操作）

 

% ssh -i huaiyuantan.key opc@master

b 、 创建实验目录

 

sudo mkdir /opt/es

sudo chown -R opc:opc /opt/es

cd /opt/es

 

c 、 下载 elasticSearch 安装包

 

wget  https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-8.4.2-linux-x86_64.tar.gz

 

d 、 解压缩包

 

tar -zxvf elasticsearch-8.4.2-linux-x86_64.tar.gz

 

同样的操作在其它两个节点 (data1 ， data2) 上执行一遍，完成 elasticsearch 软件包的安装。

 

e 、 修改三个节点 (master,data1,data2) 的系统限制参数

vi  /etc/security/limits.conf

*                soft    nofile          65536

*                hard    nofile          65536

退出重新登陆生效

 

 

f 、 修改系统内存参数

vim /etc/sysctl.conf

vm.max_map_count=262144

sysctl -p

 

 

g 、 修改三个节点 (master,data1,data2) 的 elasticsearch 配置文件 - config/elasticsearch.yml ，修改后内容如下 :

 

master 节点

 

# ======================== Elasticsearch Configuration =========================

#

# NOTE: Elasticsearch comes with reasonable defaults for most settings.

#       Before you set out to tweak and tune the configuration, make sure you

#       understand what are you trying to accomplish and the consequences.

#

# The primary way of configuring a node is via this file. This template lists

# the most important settings you may want to configure for a production cluster.

#

# Please consult the documentation for further information on configuration options:

# https://www.elastic.co/guide/en/elasticsearch/reference/index.html

#

# ---------------------------------- Cluster -----------------------------------

#

# Use a descriptive name for your cluster:

#

#cluster.name: es-cluster

#

# ------------------------------------ Node ------------------------------------

#

# Use a descriptive name for the node:

#

node.name: master

#

# Add custom attributes to the node:

#

#node.attr.rack: r1

#

# ----------------------------------- Paths ------------------------------------

#

# Path to directory where to store the data (separate multiple locations by comma):

#

path.data: /opt/es/elasticsearch-8.4.2/data

#

# Path to log files:

#

path.logs: /opt/es/elasticsearch-8.4.2/logs

#

# ----------------------------------- Memory -----------------------------------

#

# Lock the memory on startup:

#

#bootstrap.memory_lock: true

#

# Make sure that the heap size is set to about half the memory available

# on the system and that the owner of the process is allowed to use this

# limit.

#

# Elasticsearch performs poorly when the system is swapping the memory.

#

# ---------------------------------- Network -----------------------------------

#

# By default Elasticsearch is only accessible on localhost. Set a different

# address here to expose this node on the network:

#

network.host: 10.0.0.149

#

# By default Elasticsearch listens for HTTP traffic on the first free port it

# finds starting at 9200. Set a specific HTTP port here:

#

#http.port: 9200

#

# For more information, consult the network module documentation.

#

# --------------------------------- Discovery ----------------------------------

#

# Pass an initial list of hosts to perform discovery when this node is started:

# The default list of hosts is ["127.0.0.1", "[::1]"]

#

discovery.seed_hosts: ["master", "data1", "data2"]

#

# Bootstrap the cluster using an initial set of master-eligible nodes:

#

cluster.initial_master_nodes: ["master"]

#

# For more information, consult the discovery and cluster formation module documentation.

#

# --------------------------------- Readiness ----------------------------------

#

# Enable an unauthenticated TCP readiness endpoint on localhost

#

#readiness.port: 9399

#

# ---------------------------------- Various -----------------------------------

#

# Allow wildcard deletion of indices:

#

#action.destructive_requires_name: false

#

xpack.security.enabled: false

node.roles: [master, remote_cluster_client]

 

data1 节点

 

# ======================== Elasticsearch Configuration =========================

#

# NOTE: Elasticsearch comes with reasonable defaults for most settings.

#       Before you set out to tweak and tune the configuration, make sure you

#       understand what are you trying to accomplish and the consequences.

#

# The primary way of configuring a node is via this file. This template lists

# the most important settings you may want to configure for a production cluster.

#

# Please consult the documentation for further information on configuration options:

# https://www.elastic.co/guide/en/elasticsearch/reference/index.html

#

# ---------------------------------- Cluster -----------------------------------

#

# Use a descriptive name for your cluster:

#

#cluster.name: es-cluster

#

# ------------------------------------ Node ------------------------------------

#

# Use a descriptive name for the node:

#

node.name: data2

#

# Add custom attributes to the node:

#

#node.attr.rack: r1

#

# ----------------------------------- Paths ------------------------------------

#

# Path to directory where to store the data (separate multiple locations by comma):

#

path.data: /opt/es/elasticsearch-8.4.2/data

#

# Path to log files:

#

path.logs: /opt/es/elasticsearch-8.4.2/logs

#

# ----------------------------------- Memory -----------------------------------

#

# Lock the memory on startup:

#

#bootstrap.memory_lock: true

#

# Make sure that the heap size is set to about half the memory available

# on the system and that the owner of the process is allowed to use this

# limit.

#

# Elasticsearch performs poorly when the system is swapping the memory.

#

# ---------------------------------- Network -----------------------------------

#

# By default Elasticsearch is only accessible on localhost. Set a different

# address here to expose this node on the network:

#

network.host: 10.0.0.227

#

# By default Elasticsearch listens for HTTP traffic on the first free port it

# finds starting at 9200. Set a specific HTTP port here:

#

#http.port: 9200

#

# For more information, consult the network module documentation.

#

# --------------------------------- Discovery ----------------------------------

#

# Pass an initial list of hosts to perform discovery when this node is started:

# The default list of hosts is ["127.0.0.1", "[::1]"]

#

discovery.seed_hosts: ["master", "data1", "data2"]

#

# Bootstrap the cluster using an initial set of master-eligible nodes:

#

cluster.initial_master_nodes: ["master"]

#

# For more information, consult the discovery and cluster formation module documentation.

#

# --------------------------------- Readiness ----------------------------------

#

# Enable an unauthenticated TCP readiness endpoint on localhost

#

#readiness.port: 9399

#

# ---------------------------------- Various -----------------------------------

#

# Allow wildcard deletion of indices:

#

#action.destructive_requires_name: false

#

xpack.security.enabled: false

node.roles: [data]

 

data2 节点

 

# ======================== Elasticsearch Configuration =========================

#

# NOTE: Elasticsearch comes with reasonable defaults for most settings.

#       Before you set out to tweak and tune the configuration, make sure you

#       understand what are you trying to accomplish and the consequences.

#

# The primary way of configuring a node is via this file. This template lists

# the most important settings you may want to configure for a production cluster.

#

# Please consult the documentation for further information on configuration options:

# https://www.elastic.co/guide/en/elasticsearch/reference/index.html

#

# ---------------------------------- Cluster -----------------------------------

#

# Use a descriptive name for your cluster:

#

#cluster.name: es-cluster

#

# ------------------------------------ Node ------------------------------------

#

# Use a descriptive name for the node:

#

node.name: data1

#

# Add custom attributes to the node:

#

#node.attr.rack: r1

#

# ----------------------------------- Paths ------------------------------------

#

# Path to directory where to store the data (separate multiple locations by comma):

#

path.data: /opt/es/elasticsearch-8.4.2/data

#

# Path to log files:

#

path.logs: /opt/es/elasticsearch-8.4.2/logs

#

# ----------------------------------- Memory -----------------------------------

#

# Lock the memory on startup:

#

#bootstrap.memory_lock: true

#

# Make sure that the heap size is set to about half the memory available

# on the system and that the owner of the process is allowed to use this

# limit.

#

# Elasticsearch performs poorly when the system is swapping the memory.

#

# ---------------------------------- Network -----------------------------------

#

# By default Elasticsearch is only accessible on localhost. Set a different

# address here to expose this node on the network:

#

network.host: 10.0.0.65

#

# By default Elasticsearch listens for HTTP traffic on the first free port it

# finds starting at 9200. Set a specific HTTP port here:

#

#http.port: 9200

#

# For more information, consult the network module documentation.

#

# --------------------------------- Discovery ----------------------------------

#

# Pass an initial list of hosts to perform discovery when this node is started:

# The default list of hosts is ["127.0.0.1", "[::1]"]

#

discovery.seed_hosts: ["master", "data1", "data2"]

#

# Bootstrap the cluster using an initial set of master-eligible nodes:

#

cluster.initial_master_nodes: ["master"]

#

# For more information, consult the discovery and cluster formation module documentation.

#

# --------------------------------- Readiness ----------------------------------

#

# Enable an unauthenticated TCP readiness endpoint on localhost

#

#readiness.port: 9399

#

# ---------------------------------- Various -----------------------------------

#

# Allow wildcard deletion of indices:

#

#action.destructive_requires_name: false

#

xpack.security.enabled: false

node.roles: [data]

 

 

h 、启动ElasticSearch

 

分别登陆master,data1,data2, 启动 ElasticSearch ( 第一次启动可以注意输出日志是否有错，后面启动可以直接转后台运行)

 

cd /opt/es/elasticsearch-8.4.2
./bin/elasticsearch 

 

i 、 查看集群相关信息

 

$ curl -XGET "http://master:9200/_cluster/health?pretty=true"

 

green 表示集群状态正常。

 

查看集群节点信息：

 

$ curl -XGET "http://master:9200/_cluster/nodes?pretty=true"

 

输出信息显示 elasticsearch 集群包含三个节点: 一个master 节点，两个data 节点。

 

 

四、                    部署 Kafka

 

1 、 Kafka 以及自带的 zookeeper 依赖于 JDK ，所以在部署 kafka 之前，需要安装 JDK ，并配置环境变量

 

a、jdk下载网址

 

http://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html 

 

b 、下载到 /opt/elk 目录下，解压

 

$ tar -zxvf jdk-8u181-linux-x64.tar.gz

 

c 、配置 opc 用户的 JAVA 环境变量

vi /home/opc/.bash_profile

 

export JAVA_HOME=/opt/elk/jdk1.8.0_181

export JRE_HOME=${JAVA_HOME}/jre

export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib

export PATH=${JAVA_HOME}/bin:$PATH

 

2 、安装 kafka

 

a 、下载 kafka 安装包

 

wget  https://downloads.apache.org/kafka/3.2.3/kafka_2.12-3.2.3.tgz

 

b 、解压

 

$ tar -zxvf kafka_2.12-3.2.3.tgz

 

c 、修改 kakfa 配置文件   ./config/server.properties

 

############################# Server Basics #############################

 

# The id of the broker.  This must be set to a unique integer for each broker.

broker.id=1

 

############################# Socket Server Settings #############################

 

# The address the socket server listens on. If not configured, the host name will be equal to the value of

# java.net.InetAddress.getCanonicalHostName(), with PLAINTEXT listener name, and port 9092.

#   FORMAT:

#     listeners = listener_name://host_name:port

#   EXAMPLE:

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

listeners=PLAINTEXT://10.0.0.53:9092

 

# Listener name, hostname and port the broker will advertise to clients.

# If not set, it uses the value for "listeners".

advertised.listeners=PLAINTEXT://10.0.0.53:9092

 

# Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details

#listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

 

# The number of threads that the server uses for receiving requests from the network and sending responses to the network

num.network.threads=3

 

# The number of threads that the server uses for processing requests, which may include disk I/O

num.io.threads=8

 

# The send buffer (SO_SNDBUF) used by the socket server

socket.send.buffer.bytes=102400

 

# The receive buffer (SO_RCVBUF) used by the socket server

socket.receive.buffer.bytes=102400

 

# The maximum size of a request that the socket server will accept (protection against OOM)

socket.request.max.bytes=104857600

 

 

############################# Log Basics #############################

 

# A comma separated list of directories under which to store log files

log.dirs=/opt/elk/kafka_2.12-3.2.3/logs

 

# The default number of log partitions per topic. More partitions allow greater

# parallelism for consumption, but this will also result in more files across

# the brokers.

num.partitions=1

 

# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.

# This value is recommended to be increased for installations with data dirs located in RAID array.

num.recovery.threads.per.data.dir=1

 

############################# Internal Topic Settings  #############################

# The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state"

# For anything other than development testing, a value greater than 1 is recommended to ensure availability such as 3.

offsets.topic.replication.factor=1

transaction.state.log.replication.factor=1

transaction.state.log.min.isr=1

 

############################# Log Flush Policy #############################

 

# Messages are immediately written to the filesystem but by default we only fsync() to sync

# the OS cache lazily. The following configurations control the flush of data to disk.

# There are a few important trade-offs here:

#    1. Durability: Unflushed data may be lost if you are not using replication.

#    2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.

#    3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to excessive seeks.

# The settings below allow one to configure the flush policy to flush data after a period of time or

# every N messages (or both). This can be done globally and overridden on a per-topic basis.

 

# The number of messages to accept before forcing a flush of data to disk

#log.flush.interval.messages=10000

 

# The maximum amount of time a message can sit in a log before we force a flush

#log.flush.interval.ms=1000

 

############################# Log Retention Policy #############################

 

# The following configurations control the disposal of log segments. The policy can

# be set to delete segments after a period of time, or after a given size has accumulated.

# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens

# from the end of the log.

 

# The minimum age of a log file to be eligible for deletion due to age

log.retention.hours=168

 

# A size-based retention policy for logs. Segments are pruned from the log unless the remaining

# segments drop below log.retention.bytes. Functions independently of log.retention.hours.

#log.retention.bytes=1073741824

 

# The maximum size of a log segment file. When this size is reached a new log segment will be created.

log.segment.bytes=1073741824

 

# The interval at which log segments are checked to see if they can be deleted according

# to the retention policies

log.retention.check.interval.ms=300000

 

############################# Zookeeper #############################

 

# Zookeeper connection string (see zookeeper docs for details).

# This is a comma separated host:port pairs, each corresponding to a zk

# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".

# You can also append an optional chroot string to the urls to specify the

# root directory for all kafka znodes.

zookeeper.connect=127.0.0.1:2181

 

# Timeout in ms for connecting to zookeeper

zookeeper.connection.timeout.ms=68000

 

 

############################# Group Coordinator Settings #############################

 

# The following configuration specifies the time, in milliseconds, that the GroupCoordinator will delay the initial consumer rebalance.

# The rebalance will be further delayed by the value of group.initial.rebalance.delay.ms as new members join the group, up to a maximum of max.poll.interval.ms.

# The default value for this is 3 seconds.

# We override this to 0 here as it makes for a better out-of-the-box experience for development and testing.

# However, in production environments the default value of 3 seconds is more suitable as this will help to avoid unnecessary, and potentially expensive, rebalances during application startup.

group.initial.rebalance.delay.ms=0

 

 

d 、 启动 zookeeper 和 kafka

 

$ ./bin/zookeeper-server-start.sh  -daemon config/zookeeper.properties

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

 

 

 

五、                    部署 Filebeat

 

1 、安装 Nginx,Tomcat 和 MySQL

 

         根据预设的方案需求，本次日志会收集互联网公司软件架构中最常见的几种系统组件:Nginx,Tomcat,MySQL 的日志，因此要先部署这三种软件。

我们使用极简的yum 方式进行安装，三个组件都部署在kibana 节点上。

 

# yum install -y nginx

# yum -y install mysql-server mysql mysql-devel

# yum install -y tomcat

 

2 、安装filebeat

 

Filebeat 是ELK 的一个日志收集组件，自身集成了很多常见软件日志的收集模板。

 

a 、创建目录

 

$ sudo mkdir /opt/elk/

 

b 、 下载filebeat 安装包

 

$ wget  https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-8.4.2-linux-x86_64.tar.gz

 

c 、 解压安装包

 

$ tar -zxvf filebeat-8.4.2-linux-x86_64.tar.gz

$ cd /opt/elk/filebeat-8.4.2-linux-x86_64/

 

d 、修改filebeat.yml 文件，配置要收集的日志信息:

 

filebeat.inputs:

- type: log

  enabled: true

  paths:

      - /var/log/nginx/access.log*

  tags: ["nginx_access"]                                          

  exclude_files: ['\.gz$']

- type: log

  enabled: true

  paths:

      - /var/log/nginx/error.log*

  exclude_files: ['\.gz$']

  tags: ["nginx_error"]

 

- type: log

  enabled: true

  paths:

      - /var/log/mysqld.log

  tags: ["mysql_log"]

- type: log

  enabled: true

  paths:

      - /opt/elk/tomcat8/logs/localhost_access_log.*

  tags: ["tomcat_log"]

- type: log

  enabled: true

  paths:

      - /opt/elk/kafka_2.12-3.2.3/logs/kafkaServer.out

  multiline.pattern: '^\['

  multiline.negate: true

  multiline.match: after

  tags: ["kafka_log"]

- type: log

  enabled: true

  paths:

      - /opt/elk/kafka_2.12-3.2.3/logs/zookeeper.out

  multiline.pattern: '^\['

  multiline.negate: true

  multiline.match: after

  tags: ["zookeeper_log"]

 

# ================================= Processors =================================

processors:

  - drop_fields:

        fields: ["log","host","input","agent","ecs"]

 

output.kafka:   

  hosts: ["10.0.0.53:9092"]  

  topic: "filebeat"   

  partition.round_robin:

    reachable_only: true

  required_acks: 1 

  compression: gzip 

  max_message_bytes: 1000000 

 

日志内容分为三部分，其中 :

 

filebeat.inputs:

 

这部分配置收集日志的路径、文件名等信息，其中 tags 用于给下游 logstash 进行日志处理时的标签分类使用。

 

- type: log

  enabled: true

  paths:

      - /opt/elk/kafka_2.12-3.2.3/logs/zookeeper.out

  multiline.pattern: '^\['

  multiline.negate: true

  multiline.match: after

  tags: ["zookeeper_log"]

 

这部分里， multiline 用于处理多行日志使用的，像 java error stack 信息，日志会打印出多行来，用这种方式可以把所有行当做一条记录处理，有利于信息查看。

比如这种信息 :

 

[2022-10-09 04:01:41,853] WARN [SocketServer listenerType=ZK_BROKER, nodeId=1] Unexpected error from /205.210.31.17 (channelId=10.0.0.53:9092-205.210.31.17:58204-368); closing connection (org.apache.kafka.common.network.Selector) org.apache.kafka.common.network.InvalidReceiveException: Invalid receive (size = 1195725856 larger than 104857600) at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:105) at org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:452) at org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:402) at org.apache.kafka.common.network.Selector.attemptRead(Selector.java:674) at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:576) at org.apache.kafka.common.network.Selector.poll(Selector.java:481) at kafka.network.Processor.poll(SocketServer.scala:1144) at kafka.network.Processor.run(SocketServer.scala:1047) at java.lang.Thread.run(Thread.java:748)

 

 

一条信息分成了很多行，需要进行多行处理 , 这样 ElasticSearch 存储的信息就是一条记录 :

 

 

 

processors:

  - drop_fields:

        fields: ["log","host","input","agent","ecs"]

 

这部分是对日志内容进行处理， filebeat 在收集日志时会自动加上一些额外的元数据信息，如果不需要，可以在这里去掉。

 

output.kafka:   

  hosts: ["10.0.0.53:9092"]  

  topic: "filebeat"   

  partition.round_robin:

    reachable_only: true

  required_acks: 1 

  compression: gzip 

  max_message_bytes: 1000000 

 

日志的输出配置，在这里日志要写入 kafka 中，所以配置了 kafka 相关的信息。

 

3 、完成配置信息后，启动filebeat

 

$ nohup ./filebeat -e -c filebeat.yml &

 

4 、此时kafka 中应该有数据写入

 

$ ./bin/kafka-topics.sh --list --bootstrap-server 10.0.0.53:9092

 

__consumer_offsets

filebeat

 

filebeat 这个 topic 就是 filebeat 端写入数据使用的 topic ，表示 filebeat 已经将数据写入到 kafka 里。

 

 

六、                    部署 Logstash

 

1 、下载logstash 到/opt/elk 目录下

 

wget https://artifacts.elastic.co/downloads/kibana/logstash-8.4.2-linux-x86_64.tar.gz

 

2 、解压

tar -zxvf logstash-8.4.2-linux-x86_64.tar.gz

 

3 、配置logstash 参数

logstash 是 ELK 产品栈里对日志处理最核心的环节，所有日志的解析、拆分处理都在这个环节完成，由于每种日志的格式不尽相同，所以需要自己来写相应的日志处理语句 ( 通过正则表达式 ) 来完成日志文件的拆分。

以下是本次测试的 logstash 配置文件，我们对多种日志进行了拆分处理。

 

input{

   kafka{

    bootstrap_servers => ["10.0.0.53:9092"]

        client_id => "filebeat"

        group_id => "filebeat"

        auto_offset_reset => "latest" 

        consumer_threads => 1    

        topics => ["filebeat"]

        codec => json

  }

}

filter {

 if "nginx_access" in [tags] {

    grok {

        match => {

            "message" => '%{IP:remote_addr} - (%{WORD:remote_user}|-) \[%{HTTPDATE:time_local}\] "%{WORD:method} %{NOTSPACE:request} HTTP/%{NUMBER}" %{NUMBER:status} %{NUMBER:body_bytes_sent} %{QS} %{QS:http_user_agent}'

        }

    }

  }

 

 if "nginx_error" in [tags] {

    if "notice" in [message]{ 

        grok{

            match => {

              "message" =>  "(?<timestamp>%{YEAR}[./]%{MONTHNUM}[./]%{MONTHDAY} %{TIME}) \[%{LOGLEVEL:LEVEL}\] %{POSINT:pid}#%{NUMBER}: %{GREEDYDATA:errormessage}"

                }

            }

        }

    else{

        grok{

            match => {

            "message" =>  "(?<timestamp>%{YEAR}[./]%{MONTHNUM}[./]%{MONTHDAY} %{TIME}) \[%{LOGLEVEL:LEVEL}\] %{POSINT:pid}#%{NUMBER}: %{GREEDYDATA:errormessage} client: %{IP:client}, server: (?<server>\S+), request: \"%{GREEDYDATA:request}\", host: \"(?<host>\S+)\""

            }

             }

         }

}

 

 if "mysql_log" in [tags] {

    grok {

        match => {

            "message" => "(?<date>\d+-\d+-\d+)T(?<time>\d+:\d+:\d+).*\[%{WORD:level}\] \[(?<error>\S+)\] \[%{WORD:type}\](?<info>.*)"

        }

    }

  }

 if "kafka_log" in [tags] {

    grok {

        match => {

            "message" => "(?<time>.*?\]) %{WORD:level}%{GREEDYDATA:info}"

        }

    }

  }

 if "zookeeper_log" in [tags] {

    grok {

        match => {

            "message" => "(?<time>.*?\]) %{WORD:level}%{GREEDYDATA:info}"

        }

    }

  }

 

 if "es_master" in [tags] {

    grok {

        match => {

            "message" => "\[(?<time>\S{0,23})\]\[%{DATA:level}\s\]\[%{GREEDYDATA:info}"

        }

    }

  }

 

 if "es_data1" in [tags] {

    grok {

        match => {

            "message" => "\[(?<time>\S{0,23})\]\[%{DATA:level}\s\]\[%{GREEDYDATA:info}"

        }

    }

  }

 

 if "es_data2" in [tags] {

    grok {

        match => {

            "message" => "\[(?<time>\S{0,23})\]\[%{DATA:level}\s\]\[%{GREEDYDATA:info}"

        }

    }

  }

 

 if "tomcat_log" in [tags] {

    grok {

        match => {

            "message" => "%{IP:request_ip} - - \[(?<time>\S{0,26}).*\"%{GREEDYDATA:request}\" %{NUMBER:code}"

        }

#        remove_field => ["message"]

    }

  }

}

 

output {

 if "nginx_access" in [tags] {

        elasticsearch {

        hosts => ["10.0.0.149:9200"]

        index => "nginx-access-%{+YYYYMMdd}"

    }

       file{

 

              path => "/opt/elk/logstash-8.4.2/logs/nginx-access-%{+YYYY.MM.dd}.txt"

 

       }

  }

 

 if "nginx_error" in [tags] {

        elasticsearch {

        hosts => ["10.0.0.149:9200"]

        index => "nginx-error-%{+YYYYMMdd}"

    }

       file{

 

              path => "/opt/elk/logstash-8.4.2/logs/nginx-error-%{+YYYY.MM.dd}.txt"

 

       }

  }

 

 if "mysql_log" in [tags] {

        elasticsearch {

        hosts => ["10.0.0.149:9200"]

        index => "mysql-log-%{+YYYYMMdd}"

    }

       file{

 

              path => "/opt/elk/logstash-8.4.2/logs/mysql-log-%{+YYYY.MM.dd}.txt"

 

       }

  } 

 

   if "kafka_log" in [tags] {

        elasticsearch {

        hosts => ["10.0.0.149:9200"]

        index => "kafka_log-%{+YYYYMMdd}"

    }

       file{

 

              path => "/opt/elk/logstash-8.4.2/logs/kafka_log-%{+YYYY.MM.dd}.txt"

 

       }

  }

   if "zookeeper_log" in [tags] {

        elasticsearch {

        hosts => ["10.0.0.149:9200"]

        index => "zookeeper_log-%{+YYYYMMdd}"

    }

       file{

 

              path => "/opt/elk/logstash-8.4.2/logs/zookeeper_log-%{+YYYY.MM.dd}.txt"

 

       }

  }

 

 

 

   if "es_master" in [tags] {

        elasticsearch {

        hosts => ["10.0.0.149:9200"]

        index => "es-log-%{+YYYYMMdd}"

    }

       file{

 

              path => "/opt/elk/logstash-8.4.2/logs/es-master-log-%{+YYYY.MM.dd}.txt"

 

       }

  }

 

   if "es_data1" in [tags] {

        elasticsearch {

        hosts => ["10.0.0.149:9200"]

        index => "es-log-%{+YYYYMMdd}"

    }

       file{

 

              path => "/opt/elk/logstash-8.4.2/logs/es-data1-log-%{+YYYY.MM.dd}.txt"

 

       }

  }

     if "es_data2" in [tags] {

        elasticsearch {

        hosts => ["10.0.0.149:9200"]

        index => "es-log-%{+YYYYMMdd}"

    }

       file{

 

              path => "/opt/elk/logstash-8.4.2/logs/es-data2-log-%{+YYYY.MM.dd}.txt"

       }

  } 

 

 

if "tomcat_log" in [tags] {

        elasticsearch {

        hosts => ["10.0.0.149:9200"]

        index => "tomcat-log-%{+YYYYMMdd}"

    }

       file{

 

              path => "/opt/elk/logstash-8.4.2/logs/tomcat-log-%{+YYYY.MM.dd}.txt"

 

       }

  }

}

 

整个配置文件主要由三部分组成 :

 

input{

   kafka{

    bootstrap_servers => ["10.0.0.53:9092"]

        client_id => "filebeat"

        group_id => "filebeat"

        auto_offset_reset => "latest" 

        consumer_threads => 1    

        topics => ["filebeat"]

        codec => json

  }

}

 

input 部分 ,logstash 对写入 kafka 的 filebeat 收集的日志进行消费 (consume) ，这部分主要是配置 kafka 的相关信息。

 

filter {

 if "nginx_error" in [tags] {

……

 

filter 部分主要是对消费的日志数据进行处理，这里会根据 filebeat 打上的 tags 标签来判断日志属于哪种数据并应用相应的日志处理模板。

 

output {

 if "nginx_access" in [tags] {

……

 

output 部分是处理完的日志的输出，同样的， logstash 根据日志打的 tags 来判断属于哪种类型的日志，然后将他们写入到 elasticsearch 里，同时根据日志的名字创建相应的索引 ( 也可以提前自己创建好 ) ，在我们的实验里，同时还向本地写入一份日志文件，可以直观的查看解析后的日志格式是否有错误。

 

       file{

 

              path => "/opt/elk/logstash-8.4.2/logs/es-data2-log-%{+YYYY.MM.dd}.txt"

       }

 

 

 

七、                    基于 Logstash的日志文件解析

 

     鉴于 logstash 在整个日志处理过程中的核心角色，我们在这里简单介绍一下一个日志文件是如何进行拆分的。

比如这里有一段 Nginx 日志 :

 

114.254.2.82 - - [09/Oct/2022:06:46:26 +0000] "POST /kibana/api/ui_counters/_report HTTP/1.1" 200 15 "http://146.56.172.44/kibana/app/dashboards" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/15.4 Safari/605.1.15" "-"

 

如果需求只是使用 ElasticSearch 的搜索功能进行关键字查询的话，我们并不需要对日志数据进行拆分

 

这种情况下将原始日志直接写入 ElasticSearch 中就可以了。

 

但是对于某些分析场景来说，还是需要进行数据的拆分，比如运维人员想知道 Nginx 的请求里面都有哪些返回代码， 200,403,404 这样的情况有多少，以确定业务的运行情况，比如这样的展示效果 :

我们需要对文档进行拆分，拆分出 method 字段 (GET,POST,PUT…) ，还有每个请求的返回代码 (200,403,404….) ，也需要进行汇总统计，这个过程实际上就是把非结构化数据进行结构化处理的过程，经过结构化处理之后，数据就可以很容易的进行多维度分析了。

 

因此我们需要对 Nginx 的日志进行拆分，由于分析的需要，我们至少要拆分出 Method 字段， status 这两个字段。

 

114.254.2.82 - - [09/Oct/2022:06:46:26 +0000] "POST /kibana/api/ui_counters/_report HTTP/1.1" 200 15 "http://146.56.172.44/kibana/app/dashboards" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/15.4 Safari/605.1.15" "-"

 

这部分在 logstash 的配置文件中是这样的 :

 

if "nginx_access" in [tags] {

    grok {

        match => {

            "message" => '%{IP:remote_addr} - (%{WORD:remote_user}|-) \[%{HTTPDATE:time_local}\] "%{WORD:method} %{NOTSPACE:request} HTTP/%{NUMBER}" %{NUMBER:status} %{NUMBER:body_bytes_sent} %{QS} %{QS:http_user_agent}'

        }

#        remove_field => ["message"]

    }

  }

 

其中 match 部分是是对日志的拆分，从这里我们可以看到这个日志被拆分成

remote_addr/ time_local/method/request/status/body_bytes_sent/http_user_agent

等字段，后续在进行分析时，我们可以对这些字段进行多种组合分析和展现。

 

说明一下， grok 是一种数据处理的语法，实际上它是对正则表达式的二次封装，做成了一些现成的模板，大多数时候可以可以直接使用，省掉了反复写一些正则表达式的繁琐工作。

比如对于字串 10.0.0.53 这个 IP 地址， logstash 如果要用正则表达式来进行处理，我们需要写成这样：

 

(<ip_addr>\d+-\d+-\d+-\d+)

 

如果使用 grok 语法的话，可以直接写成 :

 

%{IP: ip_addr }

 

直接使用 IP 这个模板来替换上面的冗长的正则表达式 , 非常方便。

Logstash 既支持 grok 语法，也支持直接用正则表达式的写法。

 

 

 

八、                    部署 Kibana

 

Kibana 在 ELK 产品栈里提供数据的查询，分析和展示。

1 、下载kibana

 

$ cd /opt/elk

$ wget wget  https://artifacts.elastic.co/downloads/kibana/kibana-8.4.2-linux-x86_64.tar.gz

 

2 、解压

 

$ tar -zxvf kibana-8.4.2-linux-x86_64.tar.gz

 

3 、配置kibana

 

vi config/kibana.yml

 

# For more configuration options see the configuration guide for Kibana in

# https://www.elastic.co/guide/index.html

 

# =================== System: Kibana Server ===================

# Kibana is served by a back end server. This setting specifies the port to use.

#server.port: 5601

 

# Specifies the address to which the Kibana server will bind. IP addresses and host names are both valid values.

# The default is 'localhost', which usually means remote machines will not be able to connect.

# To allow connections from remote users, set this parameter to a non-loopback address.

server.host: "10.0.0.53"

 

# Enables you to specify a path to mount Kibana at if you are running behind a proxy.

# Use the `server.rewriteBasePath` setting to tell Kibana if it should remove the basePath

# from requests it receives, and to prevent a deprecation warning at startup.

# This setting cannot end in a slash.

server.basePath: "/kibana"

 

# Specifies whether Kibana should rewrite requests that are prefixed with

# `server.basePath` or require that they are rewritten by your reverse proxy.

# Defaults to `false`.

server.rewriteBasePath: true

 

# Specifies the public URL at which Kibana is available for end users. If

# `server.basePath` is configured this URL should end with the same basePath.

#server.publicBaseUrl: ""

 

# The maximum payload size in bytes for incoming server requests.

#server.maxPayload: 1048576

 

# The Kibana server's name. This is used for display purposes.

server.name: "kibana"

 

# =================== System: Kibana Server (Optional) ===================

# Enables SSL and paths to the PEM-format SSL certificate and SSL key files, respectively.

# These settings enable SSL for outgoing requests from the Kibana server to the browser.

#server.ssl.enabled: false

#server.ssl.certificate: /path/to/your/server.crt

#server.ssl.key: /path/to/your/server.key

 

# =================== System: Elasticsearch ===================

# The URLs of the Elasticsearch instances to use for all your queries.

elasticsearch.hosts: ["http://master:9200"]

 

# If your Elasticsearch is protected with basic authentication, these settings provide

# the username and password that the Kibana server uses to perform maintenance on the Kibana

# index at startup. Your Kibana users still need to authenticate with Elasticsearch, which

# is proxied through the Kibana server.

#elasticsearch.username: "kibana_system"

#elasticsearch.password: "pass"

 

# Kibana can also authenticate to Elasticsearch via "service account tokens".

# Service account tokens are Bearer style tokens that replace the traditional username/password based configuration.

# Use this token instead of a username/password.

# elasticsearch.serviceAccountToken: "my_token"

 

# Time in milliseconds to wait for Elasticsearch to respond to pings. Defaults to the value of

# the elasticsearch.requestTimeout setting.

#elasticsearch.pingTimeout: 1500

 

# Time in milliseconds to wait for responses from the back end or Elasticsearch. This value

# must be a positive integer.

#elasticsearch.requestTimeout: 30000

 

# The maximum number of sockets that can be used for communications with elasticsearch.

# Defaults to `Infinity`.

#elasticsearch.maxSockets: 1024

 

# Specifies whether Kibana should use compression for communications with elasticsearch

# Defaults to `false`.

#elasticsearch.compression: false

 

# List of Kibana client-side headers to send to Elasticsearch. To send *no* client-side

# headers, set this value to [] (an empty list).

#elasticsearch.requestHeadersWhitelist: [ authorization ]

 

# Header names and values that are sent to Elasticsearch. Any custom headers cannot be overwritten

# by client-side headers, regardless of the elasticsearch.requestHeadersWhitelist configuration.

#elasticsearch.customHeaders: {}

 

# Time in milliseconds for Elasticsearch to wait for responses from shards. Set to 0 to disable.

#elasticsearch.shardTimeout: 30000

 

# =================== System: Elasticsearch (Optional) ===================

# These files are used to verify the identity of Kibana to Elasticsearch and are required when

# xpack.security.http.ssl.client_authentication in Elasticsearch is set to required.

#elasticsearch.ssl.certificate: /path/to/your/client.crt

#elasticsearch.ssl.key: /path/to/your/client.key

 

# Enables you to specify a path to the PEM file for the certificate

# authority for your Elasticsearch instance.

#elasticsearch.ssl.certificateAuthorities: [ "/path/to/your/CA.pem" ]

 

# To disregard the validity of SSL certificates, change this setting's value to 'none'.

#elasticsearch.ssl.verificationMode: full

 

# =================== System: Logging ===================

# Set the value of this setting to off to suppress all logging output, or to debug to log everything. Defaults to 'info'

#logging.root.level: debug

 

# Enables you to specify a file where Kibana stores log output.

#logging.appenders.default:

#  type: file

#  fileName: /var/logs/kibana.log

#  layout:

#    type: json

 

# Logs queries sent to Elasticsearch.

#logging.loggers:

#  - name: elasticsearch.query

#    level: debug

 

# Logs http responses.

#logging.loggers:

#  - name: http.server.response

#    level: debug

 

# Logs system usage information.

#logging.loggers:

#  - name: metrics.ops

#    level: debug

 

# =================== System: Other ===================

# The path where Kibana stores persistent data not saved in Elasticsearch. Defaults to data

#path.data: data

 

# Specifies the path where Kibana creates the process ID file.

#pid.file: /run/kibana/kibana.pid

 

# Set the interval in milliseconds to sample system and process performance

# metrics. Minimum is 100ms. Defaults to 5000ms.

#ops.interval: 5000

 

# Specifies locale to be used for all localizable strings, dates and number formats.

# Supported languages are the following: English (default) "en", Chinese "zh-CN", Japanese "ja-JP", French "fr-FR".

#i18n.locale: "en"

 

# =================== Frequently used (Optional)===================

 

# =================== Saved Objects: Migrations ===================

# Saved object migrations run at startup. If you run into migration-related issues, you might need to adjust these settings.

 

# The number of documents migrated at a time.

# If Kibana can't start up or upgrade due to an Elasticsearch `circuit_breaking_exception`,

# use a smaller batchSize value to reduce the memory pressure. Defaults to 1000 objects per batch.

#migrations.batchSize: 1000

 

# The maximum payload size for indexing batches of upgraded saved objects.

# To avoid migrations failing due to a 413 Request Entity Too Large response from Elasticsearch.

# This value should be lower than or equal to your Elasticsearch cluster’s `http.max_content_length`

# configuration option. Default: 100mb

#migrations.maxBatchSizeBytes: 100mb

 

# The number of times to retry temporary migration failures. Increase the setting

# if migrations fail frequently with a message such as `Unable to complete the [...] step after

# 15 attempts, terminating`. Defaults to 15

#migrations.retryAttempts: 15

 

# =================== Search Autocomplete ===================

# Time in milliseconds to wait for autocomplete suggestions from Elasticsearch.

# This value must be a whole number greater than zero. Defaults to 1000ms

#unifiedSearch.autocomplete.valueSuggestions.timeout: 1000

 

# Maximum number of documents loaded by each shard to generate autocomplete suggestions.

# This value must be a whole number greater than zero. Defaults to 100_000

#unifiedSearch.autocomplete.valueSuggestions.terminateAfter: 100000

这里主要设置好 elasticsearch 的链接信息

 

elasticsearch.hosts: ["http://master:9200"]

 

为了能够从外网访问 kibana ，我们在 nginx 上做了代理 :

 

    location /kibana {

        proxy_pass http://10.0.0.53:5601;

    }

 

这样我们可以进行直接通过 publicIP/kibana 的方式来访问 kibana.

 

4 、启动kibana

 

nohup bin/kibana &

 

 

九、                    使用 Kibana进行日志的查询及展示

 

我们简单的展示一下 Kibana 在这次实验中用到的几个功能 :

 

1 、搜索功能

 

数据在 ElasticSearch 中是以索引的方式存储， kibana 的搜索功能，就是直接对这些数据按照关键字进行搜索，使用的算法还是基于底层 Lucene 的搜索算法。

 

在 Discover 部分可以进行数据内容的查询 :

 

2 、数据聚合分析及可视化展示

 

基于在前部分 logstash 中进行的数据字段拆分，我们可以在 kibana 中对这些字段进行聚合分析，类似 SQL 中的 group by 操作。

 

 

通过对 Method 和 status 进行聚合，我们可以知道在 15 分钟之内， POST,GET 方法的请求比例，以及在各自方法中 200,304,404 状态码占的比例。

 

3 、Dashbaord 集中展示

 

通过 kibana 的 dashboard 功能，可以将多种图表放到一起来集中展示，可以更直观的获取多方面的数据信息。