elk
|
EFLK
|
47

3540 字
|
1.1 小时

参考连接
ELK学习笔记 – 简书
【ELK学习笔记】ELK的简介_elk中beats包含哪些-CSDN博客
elk基础环境安装 | 二丫讲梵

1 elk生产环境方案

logstash 配置es 集群 logstash集群方案_jiecho的技术博客_51CTO博客

2 介绍

Elasticsearch：开源分布式搜索引擎，提供搜集、分析、存储数据功能。

Logstash：日志搜集、分析、过滤，支持大量数据获取。其自带输入（input）、过滤语法（grok）、输出（output）三部分。其输入有两种方式：
①由各beat采集器输入，经过滤后输出到ES
②本机数据输入，经过滤后输出到ES。

Kibana：提供日志分析友好的 Web 界面。数据存储到ES中后，可以在Kibana页面上增删改查，交互数据，并生成各种维度表格、图形。
新增的Filebeat是一个轻量级的日志收集处理工具(Agent)，占用资源少，官方也推荐此工具。还有其他beat等，可以在各服务器上搜集信息，传输给Logastash。

总的来说就是：beats+Logstash收集，ES存储，Kibana可视化及分析。下面分别是单机和集群的示意图（集群中也可以采用多beats）。


2. ELK 典型工作流程

  1. 数据采集
    • Beats/Logstash 从服务器、应用、网络设备等采集日志。

  2. 数据处理
    • Logstash 解析日志（如将 Nginx 日志拆分为字段）。

  3. 存储与索引
    • 数据存入 Elasticsearch 并建立索引。

  4. 可视化与分析
    • 通过 Kibana 查询日志、生成报表或设置告警。

3. ELK 常见应用场景
• 日志集中管理：聚合多台服务器的日志，快速定位问题。

• 安全分析（SIEM）：检测异常登录、攻击行为（如 ELK + Suricata）。

• 业务监控：实时展示订单量、API 响应时间等指标。

• 应用性能监控（APM）：跟踪微服务调用链路（需配合 Elastic APM）。

3 ELK常见术语


索引: index
索引，ES存储的逻辑概念，客户端基于索引进行数据的读写。

分片: shard
一个索引最少有1个分片，当分片数量大于1时，就能够实现数据的分布式存储。

副本: replica
一个分片可以有0个或多个副本，当分片的副本数量为0时，则相应节点宕机时，数据将无法访问。
当副本数量大于等于1时，相应的分片区分主分片(primary shard，rw)和副本分片(replica shard, ro)。

文档: document
写入ES集群的一条数据。

4 面试题:

	ES集群颜色有几种，分别代码什么含义？
		- Red:
			表示ES集群有部分主分片无法访问，集群处于不健康状态。
			一般这种情况，在集群启动时会出现，或者数据丢失的情况下。
			
		- Yellow:
			表示ES集群有部分副本分片无法访问，集群处于亚健康状态。
			
		- Green:
			表示ES集群的所有主分片和副本分片均可以正常访问，集群处于健康状态


5 filebeat

  • filebeat的log日志采集注意事项
    参考链接:
    https://www.elastic.co/guide/en/beats/filebeat/7.17/filebeat-input-log.html
	1.编写filebeat配置文件
[root@elk93 ~]# cat /etc/filebeat/config/02-log-to-console.yaml
filebeat.inputs:
- type: log
  paths:
    - /tmp/apps.log

output.console:
  pretty: true
[root@elk93 ~]# 


	2.启动filebeat配置文件 
[root@elk93 ~]# filebeat -e -c /etc/filebeat/config/02-log-to-console.yaml


	3.发送测试数据

得出结论:
<font color="#d83931"> – 1.filebeat默认按行读取数据，一条数据我们称之为一个事件"event";</font>
<font color="#d83931"> – 2.filebeat是有状态服务，会记录数据的采集偏移量offset，文件路径，inode等;</font>
<font color="#d83931"> – 3.如果需要从头采集数据，则使用’rm -rf /var/lib/filebeat/’就可以从头采集数据;</font>

  • EFK架构初体验
    参考链接:
    https://www.elastic.co/guide/en/beats/filebeat/7.17/configuration-template.html
    https://www.elastic.co/guide/en/beats/filebeat/7.17/ilm.html

5.1 采集系统日志

1.编写配置文件


[root@elk-93 ~]# cat /etc/filebeat/config/05-syslog-to-es.yml
filebeat.inputs:
- type: log
  # 给每条event事件添加一个tags字段。
  tags: bootstrap
  paths:
    - /var/log/bootstrap.log*
- type: log
  tags: kern
  paths:
    - /var/log/kern.log*
- type: log
  tags: auth
  paths:
    - /var/log/auth.log*
- type: log
  tags: syslog
  paths:
    - /var/log/syslog*

output.elasticsearch:
  hosts:
  - "http://10.168.10.91:9200"
  - "http://10.168.10.92:9200"
  - "http://10.168.10.93:9200"
  #index: "cmy-nginx-%{+yyyy-MM-dd}"
  # 根据tags标签的信息，写入不同的索引
  indices:
    - index: "cmy-bootstrap-%{+yyyy.MM.dd}"
      when.contains:
        tags: bootstrap
    - index: "cmy-kern-%{+yyyy.MM.dd}"
      when.contains:
        tags: kern
    - index: "cmy-auth-%{+yyyy.MM.dd}"
      when.contains:
        tags: auth
    - index: "cmy-syslog-%{+yyyy.MM.dd}"
      when.contains:
        tags: syslog

setup.ilm.enabled: false
setup.template.name: "cmy"
setup.template.pattern: "cmy-*"
setup.template.overwrite: false
setup.template.settings:
  index.number_of_shards: 3
  index.number_of_replicas: 1

2.启动filelbeat
filebeat -e -c /etc/filebeat/config/05-syslog-to-es.yml
3.kibanna验证

5.2 filebeat多实例

	1.什么是filebeat多实例 
在同一个节点启动多个filebeat实例。


	2.实战案例
[root@elk93 ~]# filebeat -e -c /etc/filebeat/config/01-stdin-to-console.yaml 

[root@elk93 ~]# filebeat -e -c /etc/filebeat/config/02-log-to-console.yaml --path.data /tmp/xixi



	3.验证测试 
[root@elk93 ~]# ps -ef | grep filebeat | grep -v grep
root        6775    1948  0 17:46 pts/0    00:00:00 /usr/share/filebeat/bin/filebeat --path.home /usr/share/filebeat --path.config /etc/filebeat --path.data /var/lib/filebeat --path.logs /var/log/filebeat -e -c /etc/filebeat/config/01-stdin-to-console.yaml
root        6797    4912  4 17:47 pts/1    00:00:00 /usr/share/filebeat/bin/filebeat --path.home /usr/share/filebeat --path.config /etc/filebeat --path.data /var/lib/filebeat --path.logs /var/log/filebeat -e -c /etc/filebeat/config/02-log-to-console.yaml --path.data /tmp/xixi
[root@elk93 ~]#

温馨提示:
1.如果指定相同的数据目录则会报错。因为多个filebeat实例不能共享同一个数据目录。
2.如果指定的目录不存在，则会自动创建目录。

5.3 filebeat模块

1.查看filebeat支持的模块
[root@elk93 ~]# filebeat modules list
Enabled:

Disabled:
activemq
apache
auditd
aws
...

温馨提示:
	默认情况下是没有启动任何模块的
	
	
	2.启用模块
[root@elk93 ~]# filebeat modules enable tomcat
Enabled tomcat
[root@elk93 ~]# 
[root@elk93 ~]# filebeat modules enable nginx f5 redis
Enabled nginx
Enabled f5
Enabled redis
[root@elk93 ~]# 
[root@elk93 ~]# filebeat modules list
Enabled:
f5
nginx
redis
tomcat

Disabled:
activemq
apache
	3.禁用模块
[root@elk93 ~]# filebeat modules disable f5 
Disabled f5
[root@elk93 ~]# 
[root@elk93 ~]# filebeat modules disable redis tomcat
Disabled redis
Disabled tomcat
[root@elk93 ~]# 
[root@elk93 ~]# filebeat modules list
Enabled:
nginx

Disabled:
activemq
apache
auditd
aws
...


	4.验证filebeat模块底层的逻辑之启用模块
[root@elk93 ~]# ll /etc/filebeat/modules.d/*.yml
-rw-r--r-- 1 root root 784 Feb 14 00:58 /etc/filebeat/modules.d/nginx.yml
[root@elk93 ~]# 
[root@elk93 ~]# mv /etc/filebeat/modules.d/redis.yml{.disabled,}
[root@elk93 ~]# 
[root@elk93 ~]# ll /etc/filebeat/modules.d/*.yml
-rw-r--r-- 1 root root 784 Feb 14 00:58 /etc/filebeat/modules.d/nginx.yml
-rw-r--r-- 1 root root 567 Feb 14 00:58 /etc/filebeat/modules.d/redis.yml
[root@elk93 ~]# 
[root@elk93 ~]# 
[root@elk93 ~]# filebeat modules list
Enabled:
nginx
redis

5.3.1 nginx模块实战

1.查看默认启用的模块
[root@elk93 ~]# ll /etc/filebeat/modules.d/*.yml
-rw-r--r-- 1 root root 784 Feb 14 00:58 /etc/filebeat/modules.d/nginx.yml
[root@elk93 ~]# 
[root@elk93 ~]# 
[root@elk93 ~]# filebeat modules list
Enabled:
nginx

Disabled:
activemq
apache
auditd
aws
...



	2.准备Nginx访问日志
[root@elk93 ~]# cat /var/log/nginx/access.log 
123.112.19.177 - - [29/Apr/2025:16:37:00 +0800] "GET / HTTP/1.1" 200 396 "-" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/135.0.0.0 Safari/537.36"
44.55.66.77 - - [29/Apr/2025:16:37:01 +0800] "GET /favicon.ico HTTP/1.1" 404 197 "http://10.0.0.93/" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/135.0.0.0 Safari/537.36"
33.22.11.10 - - [29/Apr/2025:16:44:03 +0800] "GET / HTTP/1.1" 200 612 "-" "curl/7.81.0"
[root@elk93 ~]# 
[root@elk93 ~]# 


	3.修改模块的配置文件
[root@elk93 ~]# egrep -v "^.*#|^$" /etc/filebeat/modules.d/nginx.yml 
- module: nginx
  access:
    enabled: true
    var.paths: ["/var/log/nginx/access.log*"]
  error:
    enabled: false
  ingress_controller:
    enabled: false
[root@elk93 ~]# 

	
	4.编写filebeat的配置文件
[root@elk93 ~]# cat /etc/filebeat/config/07-modules-nginx-to-es.yaml 
# 启动filebeat模块
filebeat.config.modules:
  # 指定模块的加载路径，‘${path.config}’是filebeat默认配置文件的路径
  path: ${path.config}/modules.d/*.yml
  # 是否支持热加载
  reload.enabled: true


output.elasticsearch:
  hosts: 
  - "http://10.0.0.91:9200"
  - "http://10.0.0.92:9200"
  - "http://10.0.0.93:9200"
  index: "cmy-modules-nginx-%{+yyyy-MM-dd}" 

setup.ilm.enabled: false
setup.template.name: "cmy"
setup.template.pattern: "cmy-*"
setup.template.overwrite: false
setup.template.settings:
  index.number_of_shards: 3
  index.number_of_replicas: 1
[root@elk93 ~]# 


	5.启动filebeat实例 
[root@elk93 ~]# filebeat -e -c /etc/filebeat/config/07-modules-nginx-to-es.yaml 


	6.kibana查看验证

5.3.2 tomcat模块实战

- filebeat基于模块采集tomcat日志案例
	1.启动tomcat模块
[root@elk93 ~]# filebeat modules enable tomcat
Enabled tomcat
[root@elk93 ~]# 


	2.修改tomcat的配置文件
[root@elk-93 ~]# cat /etc/filebeat/modules.d/tomcat.yml
- module: tomcat
  log:
    enabled: true
    var.input: file
    var.paths:
      - "/app/tools/tomcat/logs/catalina.out"
      - "/app/tools/tomcat/logs/*.txt"
  # access:
  #   enabled: false  # 注释或删除 access 部分
	3.编写filebeat的配置文件
[root@elk-93 ~]# cat /etc/filebeat/config/07-modole-tomcat.yml
# 启动filebeat模块
filebeat.config.modules:
  # 指定模块的加载路径，‘${path.config}’是filebeat默认配置文件的路径
  path: ${path.config}/modules.d/tomcat.yml
  # 是否支持热加载
  reload.enabled: true


output.elasticsearch:
  hosts:
  - "http://10.168.10.91:9200"
  - "http://10.168.10.92:9200"
  - "http://10.168.10.93:9200"
  index: "cmy-modules-tomcat-%{+yyyy-MM-dd}"

setup.ilm.enabled: false
setup.template.name: "cmy"
setup.template.pattern: "cmy-*"
setup.template.overwrite: false
setup.template.settings:
  index.number_of_shards: 3
  index.number_of_replicas: 1


	4.启动filebeat实例
[root@elk93 ~]# filebeat -e -c /etc/filebeat/config/07-modole-tomcat.yml

	5.kibana出图展示
略，见视频。

5.4 filestream

[root@elk-93 ~]# cat /etc/filebeat/config/08-filestream-tomcat-to-es.yml
filebeat.inputs:
- type: filestream
  paths:
    - /app/tools/apache-tomcat-11.0.6/logs/*.txt

output.elasticsearch:
  hosts:
  - "http://10.168.10.91:9200"
  - "http://10.168.10.92:9200"
  - "http://10.168.10.93:9200"
  index: "cmy-filestream-tomcat-%{+yyyy-MM-dd}"

setup.ilm.enabled: false
setup.template.name: "cmy"
setup.template.pattern: "cmy-*"
setup.template.overwrite: false
setup.template.settings:
  index.number_of_shards: 3
  index.number_of_replicas: 1

5.5 基于行数多行匹配

1.数据主备
cat /tmp/test.log
{
  name: "孙悟空",
  hobby: ["蟠桃","仙丹","紫霞"]
}
{
  name: "猪八戒",
  hobby: ["猴哥","高老庄"]
}
{
  name: "沙和尚"
  hobby: ["唐僧","猴哥，师傅被妖怪抓走了"]
}
2.配置文件
filebeat.inputs:
- type: filestream
  paths:
    - /tmp/test.log
  parsers:
    - multiline:
        type: count
        count_lines: 4

#output.console:
#  pretty: true


output.elasticsearch:
  hosts:
  - "http://10.168.10.91:9200"
  - "http://10.168.10.92:9200"
  - "http://10.168.10.93:9200"
  index: "cmy-multiline-%{+yyyy-MM-dd}"

setup.ilm.enabled: false
setup.template.name: "cmy"
setup.template.pattern: "cmy-*"
setup.template.overwrite: false
setup.template.settings:
  index.number_of_shards: 3
  index.number_of_replicas: 1
3.启动

5.6 基于匹配模式多行匹配

filebeat.inputs:
- type: filestream
  paths:
    - /app/tools/tomcat/logs/catalina.out
  parsers:
    - multiline:
    
        type: pattern
        pattern: '^\d'  # 匹配日期开头的行
        negate: true     # 不匹配的行（如堆栈）合并到上一行
        match: after

#output.console:
#  pretty: true


output.elasticsearch:
  hosts:
  - "http://10.168.10.91:9200"
  - "http://10.168.10.92:9200"
  - "http://10.168.10.93:9200"
  index: "cmy-pattern-%{+yyyy-MM-dd}"

setup.ilm.enabled: false
setup.template.name: "cmy"
setup.template.pattern: "cmy-*"
setup.template.overwrite: false
setup.template.settings:
  index.number_of_shards: 3
  index.number_of_replicas: 1

5.7 json

vi /app/tools/tomcat/conf/server.xml
          <Host name="tomcat.oldboyedu.com"  appBase="webapps"
                unpackWARs="true" autoDeploy="true">

		<Valve className="org.apache.catalina.valves.AccessLogValve" directory="logs"
            prefix="tomcat.oldboyedu.com_access_log" suffix=".json"
pattern="{&quot;clientip&quot;:&quot;%h&quot;,&quot;ClientUser&quot;:&quot;%l&quot;,&quot;authenticated&quot;:&quot;%u&quot;,&quot;AccessTime&quot;:&quot;%t&quot;,&quot;request&quot;:&quot;%r&quot;,&quot;status&quot;:&quot;%s&quot;,&quot;SendBytes&quot;:&quot;%b&quot;,&quot;Query?string&quot;:&quot;%q&quot;,&quot;partner&quot;:&quot;%{Referer}i&quot;,&quot;http_user_agent&quot;:&quot;%{User-Agent}i&quot;}"/>

          </Host>
          

[root@elk-93 /app/tools/tomcat/logs]# startup.sh


[root@elk-93 /app/tools/tomcat/logs]# cat /etc/filebeat/config/17-tomcat-to-logstash.yaml
filebeat.inputs:
- type: filestream
  paths:
    - /app/tools/apache-tomcat-11.0.6/logs/tomcat.oldboyedu.com_access_log.*.json
  parsers:
    - ndjson:
        target: ""
        message_key: message
output.logstash:
  hosts: ["10.168.10.92:5044"]

6 logstach

6.1 工作原理

Logstash 事件处理管道有三个阶段：输入 → 过滤器 → 输出。

inputs 模块负责收集数据，filters 模块可以对收集到的数据进行格式化、过滤、简单的数据处理，outputs 模块负责将数据同步到目的地，Logstash的处理流程，就像管道一样，数据从管道的一端，流向另外一端。

inputs 和 outputs 支持编解码器，使您能够在数据进入或离开管道时对数据进行编码或解码，而无需使用单独的过滤器。

提示：inputs/filters/outputs是通过插件机制扩展各种能力。

1686562622214

6.1.1 inputs

inputs 可以收集多种数据源的数据，下面是常见的数据源：

  • file：扫描磁盘中的文件数据，例如： 扫描日志文件。

  • syslog：监听端口514以获取 syslog 消息，并根据 RFC3164 格式解析。

  • mysql ：扫描 Mysql 的表数据

  • redis：从 redis 服务器中读取数据，使用 redis 通道和 redis 列表。

    • Redis通常用作集中式 Logstash 安装中的“代理”，用于排队来自远程 Logstash “发件人”的 Logstash 事件。

  • Filebeat：轻量级的文件数据采集器，可以取代file的能力。

  • 消息队列 kafka、rabbitmq 等：支持从各种消息队列读取数据。

6.1.2 filters

filters 是一个可选模块，可以在数据同步到目的地之前，对数据进行一些格式化、过滤、简单的数据处理操作。常用的filters功能：

  • grok：logstash 中最常用的日志解释和结构化插件。

    • grok 是一种采用组合多个预定义的正则表达式，用来匹配分割文本并映射到关键字的工具。

  • mutate ：支持事件的变换，例如重命名、移除、替换、修改等

  • drop ：完全丢弃事件

  • clone ：克隆事件

  • geoip：添加关于 IP 地址的地理位置信息

6.1.3 outputs

Logstatsh的最后一个处理节点，outputs负责将数据同步到目的地。下面是常见的目的地：

  • elasticsearch：将事件数据发送到Elasticsearch。
  • file：将事件数据写入磁盘上的文件，也可以将数据同步到一个文件中。

6.2 采集数据

6.2.1 stdin

6.2.2 file

cat /etc/logstash/conf.d/02-file-to-stdout.conf

input {
  file {
    path => ["/tmp/xixi.log", "/tmp/haha.log"]
    start_position => "beginning"  # 从文件开头读取
  }
}

output {
  stdout {  }
}

rm -f  /usr/share/logstash/data/plugins/inputs/file/.sincedb_*

6.2.3 beats

- Logstash的input插件之beats案例
参考链接:
	https://www.elastic.co/guide/en/logstash/7.17/plugins-inputs-beats.html

	1.编写logstash配置文件
[root@elk92 ~]# cat /etc/logstash/conf.d/03-beats-to-stdout.conf
input { 
  beats {
    port => 5044
  }
}   

output { 
  stdout {} 
}
[root@elk92 ~]# 


	2.启动Logstash实例
[root@elk92 ~]# logstash -f /etc/logstash/conf.d/03-beats-to-stdout.conf


	3.编写filebeat配置文件 
[root@elk93 ~]# cat > /etc/filebeat/config/15-tcp-to-logstash.yaml <<EOF
filebeat.inputs:
- type: tcp
  host: "0.0.0.0:9001"

output.logstash:
  hosts: ["10.168.10.92:5044"]
EOF
[root@elk93 ~]# 


	4.启动filebeat实例
[root@elk93 ~]# filebeat -e -c /etc/filebeat/config/15-tcp-to-logstash.yaml

	
	5.发送测试数据到filebeat
[root@elk91 ~]# echo www.oldboyedu.com  | nc 10.0.0.93 9000


	6.观察Logstash的输出
{
       "message" => "www.oldboyedu.com",
         "agent" => {
                "name" => "elk93",
                "type" => "filebeat",
            "hostname" => "elk93",
        "ephemeral_id" => "6b7c1a6b-895d-464f-9928-170c6d29b8fa",
             "version" => "7.17.28",
                  "id" => "0355ae4d-628e-4d34-bdaa-378569977b99"
    },
           "log" => {
        "source" => {
            "address" => "10.0.0.91:36374"
        }
    },
      "@version" => "1",
          "tags" => [
        [0] "beats_input_codec_plain_applied"
    ],
           "ecs" => {
        "version" => "1.12.0"
    },
    "@timestamp" => 2025-05-06T01:15:49.945Z,
          "host" => {
        "name" => "elk93"
    },
         "input" => {
        "type" => "tcp"
    }
}

6.3 filter

丰富的过滤器插件的是 logstash威力如此强大的重要因素，过滤器插件主要处理流经当前Logstash的事件信息，可以<font color="#d83931">添加字段、移除字段、转换字段类型，通过正则表达式切分数据等，也可以根据条件判断来进行不同的数据处理方式。</font>

6.3.1 grok

将非结构化数据解析成结构化数据以便于查询，非常适合解析syslog logs，apache log， mysql log，nginx log以及一些其他的web log

Logstash 提供了多种预定义的复合 Grok 模式（combined patterns），可以快速解析常见日志格式。以下是几个典型示例和说明：

Nginx 访问日志**

192.168.1.1 - - [10/Oct/2023:14:30:02 +0800] "GET /api/data HTTP/1.1" 404 153 "http://example.com" "curl/7.68.0"

复合模式：

%{NGINXACCESSLOG}

解析字段：

  • remote_addr (192.168.1.1)
  • time_local ([10/Oct/2023:14:30:02 +0800])
  • request (GET /api/data HTTP/1.1)
  • status (404)
  • body_bytes_sent (153)
  • http_referer ("http://example.com")
  • http_user_agent ("curl/7.68.0")

Syslog 日志**

Oct 11 15:32:01 ubuntu systemd[1]: Started Daily apt upgrade and clean.

复合模式：

%{SYSLOGTIMESTAMP:timestamp} %{SYSLOGHOST:hostname} %{DATA:process}(?:\[%{POSINT:pid}\])?: %{GREEDYDATA:message}

解析字段：

  • timestamp (Oct 11 15:32:01)
  • hostname (ubuntu)
  • process (systemd)
  • pid (1)
  • message (Started Daily apt upgrade and clean.)

MySQL 慢查询日志

# Time: 2023-10-11T15:33:45.123456Z
# User@Host: root[root] @ localhost [] Id: 123
# Query_time: 5.002304 Lock_time: 0.000123 Rows_sent: 1 Rows_examined: 100000
SELECT * FROM large_table WHERE id = 1;

复合模式：

%{MYSQL_SLOW_QUERY_LOG}

解析字段：

  • timestamp (2023-10-11T15:33:45.123456Z)
  • user (root)
  • host (localhost)
  • query_time (5.002304)
  • lock_time (0.000123)
  • rows_sent (1)
  • sql (SELECT * FROM large_table WHERE id = 1)

查看所有内置模式

  1. 文件路径（Logstash安装目录）： 
    /usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-patterns-core-*/patterns/

总结表

日志类型 复合模式 典型字段
Apache访问日志 %{HTTPD_COMMONLOG} clientip, verb, response
Nginx访问日志 %{NGINXACCESSLOG} remote_addr, status, agent
Syslog %{SYSLOGLINE} timestamp, hostname, message
MySQL慢查询 %{MYSQL_SLOW_QUERY} query_time, sql
自定义格式 组合基础模式 灵活定义

通过预定义的复合模式，可以快速解析常见日志，减少正则编写工作量！

6.3.2 mutate插件

用于对事件字段进行修改，支持重命名、删除、类型转换等操作

filter {
  mutate {
    convert => { "status" => "integer" }      # 转换字段类型
    rename => { "old_field" => "new_field" } # 重命名字段
    remove_field => ["debug_info"]           # 删除字段
    replace => { "user_agent" => "Unknown" } # 替换字段值
    split => { "tags" => "," }                # 分割字符串为数组
    lowercase => ["log_level"]  # 将 "ERROR" → "error"
    uppercase => ["action"]     # 将 "login" → "LOGIN"
    merge => { "full_name" => "first_name" }  # 合并字段 将 first_name 合并到 full_name
  }
}

案例

自定义应用日志处理

INFO 2025-05-06 16:23:00 [com.oldboyedu.gen-log] - DAU|2831|浏览页面|0|24916.17
INFO 2025-05-06 16:23:02 [com.oldboyedu.gen-log] - DAU|7781|清空购物车|2|23793.37
INFO 2025-05-06 16:23:07 [com.oldboyedu.gen-log] - DAU|2179|使用优惠券|1|24697.02
INFO 2025-05-06 16:23:10 [com.oldboyedu.gen-log] - DAU|7817|提交订单|0|25007.48

cat >/etc/logstash/conf.d/08-file-to-es.conf <<EOF
input { 
  file {
    path => ["/tmp/apps.log"]
    start_position => "beginning"
  } 
}   


filter {
  mutate {
     split => { "message" => "|" }

     add_field => { 
      "uid" => "%{[message][1]}" 
	  "action" => "%{[message][2]}" 
      "svip" => "%{[message][3]}" 
      "price" => "%{[message][4]}" 
      "other" => "%{[message][0]}"
     }

  }

  mutate {
     convert => {
       "price" => "float"
     }

     
     split => { "other" => " " }

     add_field => { 
       "dt" => "%{[other][1]} %{[other][2]}" 
     }
  
     remove_field => [ "path","@version","host","message", "other"]
  }



  date {
    # "2025-05-06 15:53:12"
    match => [ "dt", "yyyy-MM-dd HH:mm:ss" ]
  }

}

output { 
  stdout {} 

  elasticsearch {
      hosts => ["http://10.168.10.91:9200","http://10.168.10.92:9200","http://10.168.10.93:9200"]
      index => "cmy-elk-apps-%{+yyyy-MM-dd}"
  }
}
EOF

6.3.3 geoip

根据 IP 地址解析地理位置信息（如国家、城市、经纬度），需依赖 MaxMind 的 GeoLite2 数据库

geoip {
          database => "/oldboyedu/data/geoip/GeoLite2-City_20250311/GeoLite2-City.mmdb"
          default_database_type => "City"
          source => "clientip"
        }

6.3.4 json

json 是 Logstash 的核心编解码器（Codec）和过滤器（Filter），用于​​解析 JSON 格式的日志​​或​​将字段转换为 JSON 字符串​​。它支持两种主要用途：

  1. ​​输入阶段​​：解码 JSON 格式的原始日志（如 API 响应、应用程序日志）。
  2. ​​过滤阶段​​：解析字符串字段中的 JSON 数据，或生成 JSON 字符串。
​**​输入阶段
input {
  file {
    path => "/var/log/app.json"
    codec => json  # 自动解析 JSON 日志
  }
}


过滤阶段

filter {
  json {
    source => "message"  # 解析 message 字段中的 JSON
    target => "payload"  # 结果存储到 payload 字段（不指定则平铺到顶层）
    }}
  1. ​​优先使用 Codec​​：<font color="#d83931">在输入阶段解析 JSON 比 Filter 阶段更高效</font>。
  2. ​​裁剪字段​​：解析后通过 remove_field 删除冗余数据。

6.3.5 date

filter {
  date {
  #[09/May/2025:17:50:08 +0800]
    match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]  # 匹配原始时间字段和格式
    target => "@timestamp"                            # 存储到 @timestamp（默认）
  }
}



date {
#2023-10-15T14:32:18.123Z
match => ["timestamp", "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"]
}

6.3.6 useragent

用于​​解析 HTTP 请求中的 User-Agent 字符串​​，提取浏览器、设备、操作系统等详细信息

filter {
  user_agent {
    source => "user_agent"   # 原始 UA 字符串字段
    target => "ua"           # 解析结果存储字段（默认直接平铺到事件顶层）
  }
}




user_agent {
  source => "user_agent"
  fields => ["name", "os", "device"]  # 只提取这些字段

}

6.4 logstash多实例



课堂练习：
	将nginx，tomcat，apps的日志使用elk架构分别写入ES的三个不同的索引中。
	
	
解决方案:
	- logstash多实例
	- logstash多分支语法
	- Logstash的pipeline
	
	
	
	
- logstash多实例实现数据写入不同索引
	1.安装Logstash 
		1.1 拷贝Logstash安装包
[root@elk92 ~]# scp logstash-7.17.28-amd64.deb 10.168.10.93:~
[root@elk92 ~]# scp -r /oldboyedu/ 10.168.10.93:/


		1.2 安装Logstash 
[root@elk93 ~]# dpkg -i logstash-7.17.28-amd64.deb


		1.3 创建软连接
[root@elk93 ~]# ln -svf /usr/share/logstash/bin/logstash /usr/local/bin/
'/usr/local/bin/logstash' -> '/usr/share/logstash/bin/logstash'
[root@elk93 ~]# 
[root@elk93 ~]# mkdir /usr/share/logstash/config
[root@elk93 ~]# 
[root@elk93 ~]# ln -svf /etc/logstash/pipelines.yml /usr/share/logstash/config/
'/usr/share/logstash/config/pipelines.yml' -> '/etc/logstash/pipelines.yml'
[root@elk93 ~]# 


	2.准备配置文件采集nginx日志
		2.1 编写配置文件 
[root@elk93 ~]# cat /etc/logstash/conf.d/09-nginx-to-es.conf
input { 
  file {
    path => ["/var/log/nginx/access.log*"]
    start_position => "beginning"
  } 
}   


filter {
  grok {
    match => {
      "message" => "%{HTTPD_COMMONLOG}"
    }
  }

  useragent {
    source => "message"
    target => "cmy"
  }

  mutate {
    convert => {
        "bytes" => "integer"
    }
  }

  date {
    match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
  }

  geoip {
    database => "/oldboyedu/data/geoip/GeoLite2-City_20250311/GeoLite2-City.mmdb"
    default_database_type => "City"
    source => "clientip"
  }

}

output { 
 # stdout {} 

  elasticsearch {
      hosts => ["http://10.168.10.91:9200","http://10.168.10.92:9200","http://10.168.10.93:9200"]
      index => "cmy-elk-multiple-logstash-nginx-%{+yyyy-MM-dd}"
  }
}
[root@elk93 ~]# 

		2.2 启动Logstash实例
[root@elk93 ~]# rm -f /usr/share/logstash/data/plugins/inputs/file/.sincedb_*
[root@elk93 ~]# 
[root@elk93 ~]# logstash -rf /etc/logstash/conf.d/09-nginx-to-es.conf


	3.准备配置文件采集tomcat日志
		3.1 编写配置文件 
[root@elk93 ~]# cat /etc/logstash/conf.d/10-tomcat-to-es.conf 
input { 
  file {
    path => ["/usr/local/apache-tomcat-11.0.6/logs/tomcat.oldboyedu.com_access_log*"]
    start_position => "beginning"
  } 
}   


filter {
  json {
    source => "message"
  
    remove_field => [ "message"]
  }


  geoip {
    database => "/oldboyedu/data/geoip/GeoLite2-City_20250311/GeoLite2-City.mmdb"
    default_database_type => "City"
    source => "clientip"
  }


  mutate {
    convert => {
        "SendBytes" => "integer"
    }
  }


  date {
    match => [ "AccessTime", "[dd/MMM/yyyy:HH:mm:ss Z]" ]
  }

  useragent {
    source => "http_user_agent"
    target => "cmy"
  }

}


output { 
 #  stdout {} 

  elasticsearch {
      hosts => ["http://10.168.10.91:9200","http://10.168.10.92:9200","http://10.168.10.93:9200"]
      index => "cmy-elk-multiple-logstash-tomcat-%{+yyyy-MM-dd}"
  }
}
[root@elk93 ~]# 


		3.2 启动Logstash实例 
[root@elk93 ~]# logstash -rf /etc/logstash/conf.d/10-tomcat-to-es.conf --path.data /tmp/logstash-tomcat-mutiple



	3 采集apps日志
		3.1 编写Logstash的配置文件
[root@elk92 ~]# cat /etc/logstash/conf.d/08-file-to-es.conf 
input { 
  file {
    path => ["/tmp/apps.log"]
    start_position => "beginning"
  } 
}   


filter {
  mutate {
     split => { "message" => "|" }

     add_field => { 
      "uid" => "%{[message][1]}" 
	  "action" => "%{[message][2]}" 
      "svip" => "%{[message][3]}" 
      "price" => "%{[message][4]}" 
      "other" => "%{[message][0]}"
     }

  }

  mutate {
     convert => {
       "price" => "float"
     }

     
     split => { "other" => " " }

     add_field => { 
       "dt" => "%{[other][1]} %{[other][2]}" 
     }
  
     remove_field => [ "path","@version","host","message", "other"]
  }



  date {
    # "2025-05-06 15:53:12"
    match => [ "dt", "yyyy-MM-dd HH:mm:ss" ]
  }

}

output { 
  stdout {} 

  elasticsearch {
      hosts => ["http://10.168.10.91:9200","http://10.168.10.92:9200","http://10.168.10.93:9200"]
      index => "cmy-elk-apps-%{+yyyy-MM-dd}"
  }
}
[root@elk92 ~]# 


		3.2 启动Logstash实例
[root@elk92 ~]# logstash -rf /etc/logstash/conf.d/08-file-to-es.conf

6.5 logstash多分支

cat >> /etc/logstash/conf.d/12-if-to-es.conf <<EOF
input { 
  file {
    path => ["/var/log/nginx/access.log*"]
    start_position => "beginning"
    type => nginx
  } 


  file {
    path => ["/usr/local/apache-tomcat-11.0.6/logs/tomcat.oldboyedu.com_access_log*"]
    start_position => "beginning"
    type => tomcat
  } 


  file {
    path => ["/tmp/apps.log"]
    start_position => "beginning"
    type => apps
  } 
}   



filter {
  if [type] == "nginx" {
      grok {
         match => {
           "message" => "%{HTTPD_COMMONLOG}"
         }
       }
     
       useragent {
         source => "message"
         target => "cmy"
       }
     
       mutate {
         convert => {
             "bytes" => "integer"
         }
       }
     
       date {
         match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
       }
     
       geoip {
         database => "/oldboyedu/data/geoip/GeoLite2-City_20250311/GeoLite2-City.mmdb"
         default_database_type => "City"
         source => "clientip"
       }


  } else if [type] == "tomcat" {
        json {
          source => "message"
        
          remove_field => [ "message"]
        }
      
      
        geoip {
          database => "/oldboyedu/data/geoip/GeoLite2-City_20250311/GeoLite2-City.mmdb"
          default_database_type => "City"
          source => "clientip"
        }
      
      
        mutate {
          convert => {
              "SendBytes" => "integer"
          }
        }
      
      
        date {
          match => [ "AccessTime", "[dd/MMM/yyyy:HH:mm:ss Z]" ]
        }
      
        useragent {
          source => "http_user_agent"
          target => "cmy"
        }


  } else {
        mutate {
           split => { "message" => "|" }
      
           add_field => { 
            "uid" => "%{[message][1]}" 
      	  "action" => "%{[message][2]}" 
            "svip" => "%{[message][3]}" 
            "price" => "%{[message][4]}" 
            "other" => "%{[message][0]}"
           }
      
        }
      
        mutate {
           convert => {
             "price" => "float"
           }
      
           
           split => { "other" => " " }
      
           add_field => { 
             "dt" => "%{[other][1]} %{[other][2]}" 
           }
        
           remove_field => [ "path","@version","host","message", "other"]
        }
      
      
      
        date {
          match => [ "dt", "yyyy-MM-dd HH:mm:ss" ]
        }

  }

}


output {
   if [type] == "nginx" {
       elasticsearch {
           hosts => ["http://10.168.10.91:9200","http://10.168.10.92:9200","http://10.168.10.93:9200"]
           index => "cmy-elk-if-nginx-%{+yyyy-MM-dd}"
       }
   } else if [type] == "tomcat" {
       elasticsearch {
           hosts => ["http://10.168.10.91:9200","http://10.168.10.92:9200","http://10.168.10.93:9200"]
           index => "cmy-elk-if-tomcat-%{+yyyy-MM-dd}"
       }
   } else {
       elasticsearch {
           hosts => ["http://10.168.10.91:9200","http://10.168.10.92:9200","http://10.168.10.93:9200"]
           index => "cmy-elk-if-apps-%{+yyyy-MM-dd}"
       }
   }

}
EOF
	
	2.启动Logstash实例
[root@elk93 ~]# rm -f /usr/share/logstash/data/plugins/inputs/file/.sincedb_*
[root@elk93 ~]#   
[root@elk93 ~]# logstash -rf /etc/logstash/conf.d/12-if-to-es.conf 


	3.kibana出图展示

6.6 pipeline

 cat > /etc/logstash/pipelines.yml  <<EOF
# This file is where you define your pipelines. You can define multiple.
# For more information on multiple pipelines, see the documentation:
#   https://www.elastic.co/guide/en/logstash/current/multiple-pipelines.html

#- pipeline.id: main
#  path.config: "/etc/logstash/conf.d/*.conf"


- pipeline.id: nginx
  path.config: "/etc/logstash/conf.d/09-nginx-to-es.conf"

- pipeline.id: tomcat
  path.config: "/etc/logstash/conf.d/10-tomcat-to-es.conf"

- pipeline.id: apps
  path.config: "/etc/logstash/conf.d/11-apps-to-es.conf"
EOF


	2.kibana删除ES的索引
略，见视频。

tips：  "multiple"



	3.尝试运行Logstash，观察 数据是否写入成功
[root@elk93 ~]# rm -f /usr/share/logstash/data/plugins/inputs/file/.sincedb_*
[root@elk93 ~]# 
[root@elk93 ~]# logstash -r



温馨提示:
	如果使用'logstash -f'选项，将忽略'/etc/logstash/pipelines.yml '文件的配置。

#elk

上一篇
下一篇