【ELK(Elasticsearch+Logstash+Kibana) 从零搭建实战记录：日志采集与可视化】

ELK(Elasticsearch+Logstash+Kibana) 从零搭建实战记录：日志采集与可视化

本文记录了我在搭建ELK(Elasticsearch, Logstash, Kibana)技术栈时的完整实战过程。使用Docker Compose快速搭建了ELK服务端（监控主机），并通过Filebeat实现了对自身系统日志以及另一台Web主机（host1）上Tomcat应用日志、MySQL数据库日志的集中采集。文中详细包含了：

1. 环境规划与Docker Compose部署： 基于elk_es, elk_logstash, elk_kibana容器的详细配置。
2. Filebeat配置详解： 服务端采集自身日志 + 客户端(host1)采集Tomcat/MySQL日志的关键步骤。
3. Logstash核心技巧： 输入协议选择(beats vs tcp)、日志解析过滤(grok, date, mutate)、日志级别提取与模块化处理。
4. Kibana实战： 创建数据视图(Data View)、利用Lens制作日志级别饼图与错误趋势图，并探索了Vega热力图高级可视化。

整体架构梳理

1. 日志采集（Beats）
   通常使用 Filebeat、Metricbeat、Packetbeat 等轻量级采集器（Beats Family）将宿主机或应用产生的日志采集并发送到 Logstash 或直接发送到 Elasticsearch。
2. 日志处理（Logstash）
   Logstash 负责对接收到的日志进行过滤、解析、转换（Grok、Date、Mutate 等 Filter 插件），并根据配置将最终格式化好的事件写入 Elasticsearch（或其他输出）。
3. 数据存储与检索（Elasticsearch）
   Elasticsearch 集群负责接收 Logstash 送来的日志文档，存储在对应的 Index 中，并为上层 Kibana 提供检索与聚合能力。
4. 可视化与分析（Kibana）
   在 Kibana 中，我们需要先创建索引模式（Index Pattern），才能看到“logs-*”之类的索引，并配置仪表盘（Dashboards）、可视化（Visualize）、警报（Alerting）等。

1 部署安装

当前环境:

• host4:192.168.0.224 (监控主机)
• host1:192.168.0.221(Web主机, 已部署tomcat, mysql)

1.2 部署流程

步骤1：初始化目录

mkdir -p /opt/monitor/elk/{elasticsearch,logstash/pipeline,kibana}
cd /opt/monitor/elk

步骤2：创建docker-compose.yml

nano docker-compose.yml

services:elasticsearch:image: elasticsearch:8.17.2dns: 8.8.8.8container_name: elk_esenvironment:- TZ=Asia/Shanghai- node.name=es-node-1- cluster.name=elk-docker- discovery.type=single-node- ELASTIC_PASSWORD=YourStrongPassword2025- ES_JAVA_OPTS=-Xms1g -Xmx1g- xpack.security.enabled=false- xpack.security.http.ssl.enabled=falsevolumes:- es-data:/usr/share/elasticsearch/dataports:- 9200:9200networks:- elk-netdeploy:resources:limits:memory: 2.5Gcpus: '1.5'kibana:image: kibana:8.17.2dns: 8.8.8.8container_name: elk_kibanaenvironment:- ELASTICSEARCH_HOSTS=http://elasticsearch:9200- TZ=Asia/Shanghai- ELASTICSEARCH_REQUESTTIMEOUT=120000  # 2分钟超时- SERVER_STARTUP_TIMEOUT=120000        # 2分钟启动超时ports:- 5601:5601volumes:- kibana-data:/usr/share/kibana/datadepends_on:- elasticsearchnetworks:- elk-netdeploy:resources:limits:memory: 1.5Gcpus: '0.5'logstash:image: logstash:8.17.3dns: 8.8.8.8container_name: elk_logstashenvironment:# 统一内存设置，移除冲突配置- LS_JAVA_OPTS=-Xms1g -Xmx1g- TZ=Asia/Shanghai- pipeline.workers=2- pipeline.batch.size=125volumes:- ./logstash/pipeline/logstash.conf:/usr/share/logstash/pipeline/logstash.conf- logstash-data:/usr/share/logstash/data- ./logstash/patterns:/usr/share/logstash/patterns  # 添加此行ports:- 5044:5044networks:- elk-netdeploy:resources:limits:memory: 1Gcpus: '0.5'volumes:es-data:kibana-data:logstash-data:networks:elk-net:driver: bridge

步骤3：创建Logstash配置文件

cat <<EOF | tee /opt/monitor/elk/logstash/pipeline/logstash.conf
input {tcp {port => 5044codec => json_lines}
}output {elasticsearch {hosts => ["http://elasticsearch:9200"]index => "logs-%{+YYYY.MM.dd}"}
}
EOF#检查文件开头（必须无空格）
head -c3 /opt/monitor/elk/logstash/pipeline/logstash.conf | hexdump -C
# 正常应显示: 00000000  69 6e 70                                         |inp|

这意味着，Logstash 会监听 TCP 5044 端口，使用 json_lines 解码器接收 JSON 格式日志，然后写入索引 logs-YYYY.MM.dd。

步骤4：启动服务

1.拉取镜像配置内核

docker pull logstash:8.17.3
docker pull kibana:8.17.2
docker pull elasticsearch:8.17.2
# 应用内核参数
sudo sysctl -w vm.max_map_count=262144
echo "vm.max_map_count=262144" | sudo tee -a /etc/sysctl.conf#启动服务
docker compose up -d
docker compose logs -f --tail=50

无法拉取镜像配置镜像加速

sudo tee /etc/docker/daemon.json <<EOF
{"registry-mirrors": [
"https://docker.m.daocloud.io",
"https://docker.1ms.run",
"https://hub.rat.dev",
"https://lispy.org",
"https://docker.yomansunter.com",
"https://docker.1panel.live",
"https://a.ussh.net"]
}
EOF
sudo systemctl daemon-reload && sudo systemctl restart docker

1.2 验证与集成

1.2.1 检查服务状态

docker ps --format "table {{.Names}}\t{{.Status}}\t{{.Ports}}"

预期输出：

NAMES         STATUS              PORTS
elk_es        Up 2 minutes        0.0.0.0:9200->9200/tcp
elk_kibana    Up 1 minute         0.0.0.0:5601->5601/tcp 
elk_logstash  Up 1 minute         0.0.0.0:5044->5044/tcp

检查 Kibana 状态

docker compose logs kibana | grep -A 5 "Kibana is now available"
[INFO][root] Kibana is now available (green)

2 安装并配置 Filebeat

2.1 宿主机安装配置

既然 Logstash 已经就绪，我们接下来需要将业务主机上的日志（举例：系统日志 /var/log/*.log、Nginx 日志、应用自定义日志等）采集到 Logstash。官方推荐使用 Filebeat。以下示例以 Ubuntu/Debian 为例：

2.1.1 安装 Filebeat

通过apt/yum 安装

在想要采集日志的主机上（比如 host4 或其他待采集日志的服务器）执行：

bash复制编辑# 1. 下载并安装 Elastic GPG key（如果尚未添加）
wget -qO - https://artifacts.elastic.co/GPG-KEY-elasticsearch | sudo apt-key add -# 2. 添加 Elastic 源（以 8.17 版本为例）
sudo sh -c 'echo "deb https://artifacts.elastic.co/packages/8.x/apt stable main" > /etc/apt/sources.list.d/elastic-8.x.list'# 3. 更新并安装 Filebeat 8.17.2
sudo apt update
sudo apt install filebeat=8.17.2

说明：

• 如果你的服务器是 CentOS/RHEL，需要使用 rpm 方式安装：

  bash复制编辑sudo rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearch
  sudo yum install https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-8.17.2-x86_64.rpm
  

在现有 docker-compose.yml 中加入 Filebeat 服务

services:elasticsearch:image: elasticsearch:8.17.2dns: 8.8.8.8container_name: elk_esenvironment:- TZ=Asia/Shanghai- node.name=es-node-1- cluster.name=elk-docker- discovery.type=single-node- ELASTIC_PASSWORD=YourStrongPassword2025- ES_JAVA_OPTS=-Xms1g -Xmx1g- xpack.security.enabled=false- xpack.security.http.ssl.enabled=falsevolumes:- es-data:/usr/share/elasticsearch/dataports:- 9200:9200networks:- elk-netdeploy:resources:limits:memory: 2.5Gcpus: '1.5'kibana:image: kibana:8.17.2dns: 8.8.8.8container_name: elk_kibanaenvironment:- ELASTICSEARCH_HOSTS=http://elasticsearch:9200- TZ=Asia/Shanghai- ELASTICSEARCH_REQUESTTIMEOUT=120000  # 2分钟超时- SERVER_STARTUP_TIMEOUT=120000        # 2分钟启动超时ports:- 5601:5601volumes:- kibana-data:/usr/share/kibana/datadepends_on:- elasticsearchnetworks:- elk-netdeploy:resources:limits:memory: 1.5Gcpus: '0.5'logstash:image: logstash:8.17.3dns: 8.8.8.8container_name: elk_logstashenvironment:# 统一内存设置，移除冲突配置- LS_JAVA_OPTS=-Xms1g -Xmx1g- TZ=Asia/Shanghai- pipeline.workers=2- pipeline.batch.size=125volumes:- ./logstash/pipeline/logstash.conf:/usr/share/logstash/pipeline/logstash.conf- logstash-data:/usr/share/logstash/data- ./logstash/patterns:/usr/share/logstash/patterns  # 添加此行ports:- 5044:5044networks:- elk-netdeploy:resources:limits:memory: 1Gcpus: '0.5'filebeat:image: elastic/filebeat:8.17.2dns: 8.8.8.8hostname: host4  # 强制设置容器主机名container_name: elk_filebeatuser: rootvolumes:- ./filebeat/filebeat.yml:/usr/share/filebeat/filebeat.yml:ro- /var/log:/host/var/log:ro- ./filebeat/modules.d:/usr/share/filebeat/modules.d:ro- filebeat-data:/usr/share/filebeat/datadepends_on:- logstashnetworks:- elk-netenvironment:- HOSTNAME=host4 # 覆盖环境变量- TZ=Asia/Shanghaicommand: ["-e","-E", "filebeat.config.modules.path=/usr/share/filebeat/modules.d/*.yml","-E", "filebeat.config.modules.reload.enabled=false","-E", "output.logstash.hosts=[\"logstash:5044\"]"]deploy:resources:limits:memory: 500Mcpus: '0.25'volumes:es-data:kibana-data:logstash-data:filebeat-data:networks:elk-net:driver: bridge

2.1.2 准备 Filebeat 配置文件

在同级目录下创建 filebeat/filebeat.yml，内容示例如下（重点在于采集宿主机 /var/log 下的核心日志，并发送到 Logstash）：

mkdir filebeat

nano filebeat/filebeat.yml

############################ Filebeat 全局配置 ############################
filebeat.config.modules:# 注意：我们把 modules 文件挂载到 /usr/share/filebeat/modules.d# 如果将来需要启用 module 只要在宿主机上放入对应 yml 即可path: /usr/share/filebeat/modules.d/*.ymlreload.enabled: falsesetup.template.enabled: false
setup.dashboards.enabled: false############################ 输入（Inputs） ############################filebeat.inputs:- type: logenabled: truepaths:# 收集宿主机 /var/log 下所有 .log 文件- /host/var/log/*.log- /host/var/log/*/*.log- /host/var/log/syslog*fields:log_source: "host4-system"ignore_older: 72h# 如果日志行是 JSON，可加如下配置：# json.keys_under_root: true# json.add_error_key: trueprocessors:- add_host_metadata: {}- add_cloud_metadata: {}############################ 输出到 Logstash ############################
output.logstash:# 直接发送到 Compose 内的 logstash 容器hosts: ["logstash:5044"]# 增加重试设置retry: 3backoff: init: 1smax: 60s# 如果开启了 TLS/SSL，请在这里配置：# ssl.enabled: true# ssl.certificate_authorities: ["/usr/share/filebeat/certs/logstash-ca.crt"]# ssl.certificate: "/usr/share/filebeat/certs/filebeat.crt"# ssl.key: "/usr/share/filebeat/certs/filebeat.key"

2.1.3 启用 Filebeat 自带 Module

如果想采集系统指标（比如 /var/log/syslog、/var/log/auth.log），可以在宿主机目录 filebeat/modules.d/ 下，把对应的 system.yml.disabled 改名为 system.yml 并修改路径，让它指向 /host/var/log/syslog 或 /host/var/log/auth.log，例如：

mkdir filebeat/modules.d

nano filebeat/modules.d/system.yml

#filebeat/modules.d/system.yml
- module: systemsyslog:enabled: truevar.paths: ["/host/var/log/syslog*"]auth:enabled: truevar.paths: ["/host/var/log/auth.log*"]

如果要采集 Nginx 访问日志，类似地创建或启用 nginx.yml：

- module: nginxaccess:enabled: truevar.paths: ["/host/var/log/nginx/access.log*"]error:enabled: truevar.paths: ["/host/var/log/nginx/error.log*"]

启用好对应 module 后，Filebeat 会自动加载这些 module 的 pipeline，把数据按照 ECS 模式打平，然后通过 Logstash 转发到 Elasticsearch。

---

2.1.4 Logstash 的输入改为 Beats 插件

• Filebeat 发送的是 Beats 格式，带有专用的协议头，如果用普通 TCP 解析，Logstash 拆出来的就是那些二进制头部当作文本——所以才会看到乱码警告。
• Beats 插件的好处还包括自动重试、流量控制、和可选的 TLS 加密。

编辑本地管道配置
在宿主机上，打开你的 pipeline 目录下的 logstash.conf：

nano ./logstash/pipeline/logstash.conf

将 input { tcp { … } } 一段替换为：beats

input {beats {port => 5044# ssl => true# ssl_certificate => "/usr/share/logstash/certs/logstash.crt"# ssl_key => "/usr/share/logstash/certs/logstash.key"}
}filter {# （根据需要添加 grok、date、mutate 等）
}output {elasticsearch {hosts => ["http://elasticsearch:9200"]index => "logs-%{+YYYY.MM.dd}"# user/password if xpack.security.enabled=true}
}

两者的核心区别

	input { tcp { … } }	input { beats { … } }
协议	原生 TCP，默认按行（line）拆分	Beats 协议（带 header、心跳、ack、SSL 支持）
典型用途	自己用 nc host port 推送纯文本/JSON 行	Filebeat、Metricbeat、Winlogbeat 等 Beats 家族
优点	简单纯粹、调试方便	专用协议可靠不丢包、自动重连、可选 TLS 加密
缺点	需要自己编写 codec（如 codec => json_lines），没有 ack	必须用专用插件 beats {}，不能当普通 TCP 用

2.1.4 启动并验证

1. 回到 docker-compose.yml 所在目录，执行：

   docker compose pull   # 可选，拉取最新镜像
   docker compose down
   docker compose up -d
   

   这时会依次启动：elasticsearch → kibana → logstash → filebeat。

2. 查看 Filebeat 容器日志，确认其已成功连接 Logstash：

   docker logs -f elk_filebeat
   

   你应该能看到类似：

   Starting server on port: 5044
   

3. 验证 Logstash 是否收到并转发事件：

   docker logs -f elk_logstash | grep "Received"
   

   你会看到 Logstash 打印它接收到了来自 Filebeat 的事件。

4. 确认 Elasticsearch 中已有索引：
   在宿主机或任意能访问 ES 的终端执行：

   curl -XGET 'http://localhost:9200/_cat/indices?v' | grep filebeat-
   

   或如果你没有加载 Filebeat 默认模板，就用自定义 logs-*，用：

   curl -XGET 'http://localhost:9200/_cat/indices?v' | grep logs-% Total    % Received % Xferd  Average Speed   Time    Time     Time  CurrentDload  Upload   Total   Spent    Left  Speed100  2625    0  2625    0     0   102k      0 --:--:-- --:--:-- --:--:--  102kyellow open   logs-2025.06.09                                                    veFnpfCrSd-eLY98mvrHGQ   1   1     150182            0    179.7mb        179.7mb      179.7mbyellow open   logs-2025.06.04                                                    kdei2cXXQxG9FfVQXp48qA   1   1      27237            0     31.3mb         31.3mb       31.3mb
   

   如果出现类似 green open logs-2025.06.04 … 的索引，说明数据写入成功。

   • 在单节点（Single-Node）Elasticsearch 集群里，索引的副本（Replica）无法被分配到其他节点上，所以默认状态会是 yellow（主分片已分配、但副本未分配）。

   • 你在 curl -XGET 'http://localhost:9200/_cat/indices?v' | grep logs- 看到的：

     yellow open   logs-2025.06.04  …  1 1  …
     

     表示：

     • 1 / 1：这个索引有 1 个主分片（Primary）和 1 个副本分片（Replica）。
     • 因为是单节点，Elasticsearch 只能把主分片放到本节点、副本找不到“其他节点”可放，故索引为黄色状态。

   小结：在单节点环境下看到 yellow 是正常现象，不会影响索引读写。但如果后续要做高可用（多节点）集群，就需要把至少两个物理/虚拟机都加入集群，这样副本才能分配到其他节点，索引才会变成 green。

5. 在 Kibana 中创建对应索引模式

   1.打开 Kibana → Discover

   • 在 Kibana 左侧主菜单中，点击 “Discover”（如果菜单中没看到，先点最下面的 “View all apps”）。
   • Discover 第一次打开时，会提示你 “Select a data view” 或 “Create data view”。

   2.创建 Data view（索引模式）

   • 在 Discover 里，点击 “Create data view”（如果已经有其它 data view，则在页面右上角的下拉里选择 “Create data view”）。

   • 这时会弹出一个只有 一个输入框 的对话框：Data view name。

     • 这里就可以输入通配符了：

       logs-*
       

     • 它会自动帮你匹配所有 ES 中现有的、以 logs- 开头的索引（例如 logs-2025.06.04、logs-2025.06.09 等）。

   • 点 “Create data view”，创建完成后页面会自动跳到 Discover，并开始加载数据。

   3.选择时间字段

   • 如果你的索引文档里包含时间字段（如 @timestamp），Kibana 会自动检测并让你选择。新版 UI 会在创建后直接提示 “Select a time field”；如果你的索引里没时间字段，则可选 “I don’t want to use the Time Filter” 但通常我们用 @timestamp。

   4.开始浏览日志

   • 创建好 Data view 后，Discover 页面左上角会默认选中它，然后你就能看到所有 logs-* 索引里的事件了，并且可以用时间范围和搜索栏来筛选。

   • 打开浏览器访问 http://192.168.0.224:5601

   • 登录后进入 “Stack Management → Index Management”

   • 点击 “Create index pattern”，输入 logs-* 或 filebeat-*（取决于输出的索引前缀），选择 @timestamp 作为时间字段

   • 完成后到 “Discover” 页面就能实时看到采集到的日志条目

   

6. 测试数据

   • echo "ELK-TAIL-TEST $(date '+%Y-%m-%d %H:%M:%S')" | sudo tee -a /var/log/syslog
   • http://192.168.0.224:5601/app/discover
     • 搜索ELK-TAIL-TEST, 看到对应的数据
     • 

---

2.2 应用端安装配置（host1）

2.2.1 创建配置目录

# 在 host1 创建配置目录
sudo mkdir -p /opt/host1_filebeat/{config,logs,data}

2.2.2 编写 Filebeat 配置文件

创建配置文件 /opt/host1_filebeat/config/filebeat.yml：

sudo nano /opt/host1_filebeat/config/filebeat.yml

内容如下（根据实际路径调整）：

############################ Filebeat Modules ############################
filebeat.config.modules:path: ${path.config}/modules.d/*.ymlreload.enabled: false  # 禁用自动重载，避免状态冲突reload.period: 30ssetup.template.enabled: false
setup.dashboards.enabled: false############################ 主机标识配置 ############################
name: "host1"
fields:host_id: "host1"host_type: "web"environment: "production"############################ 输入配置 ############################
filebeat.inputs:# 系统日志- type: logpaths:- /host/var/log/*.log- /host/var/log/*/*.logfields:log_source: "host1-system"ignore_older: 72h# Tomcat 日志 (原始采集，不使用模块)- type: logpaths:- /host/tomcat_logs/catalina.out- /host/tomcat_logs/localhost_access_log*.txtfields:log_source: "host1-tomcat"log_type: "raw"  # 添加类型标识# MySQL 日志 (使用模块)- type: logpaths:- /host/mysql_logs/error.log- /host/mysql_logs/slow.logfields:log_source: "host1-mysql"log_type: "module"  # 标识使用模块处理############################ 输出配置 ############################
output.logstash:hosts: ["192.168.0.224:5044"]# 增加重试设置retry: 3backoff: init: 1smax: 60s############################ 日志配置 ############################
logging.level: info
logging.to_files: true
logging.files:path: /usr/share/filebeat/logsname: filebeatkeepfiles: 3  # 减少保留文件数permissions: 0644

2.1.3 为 host1 配置 Filebeat 模块

1. 未启用解析模块：Filebeat 需要加载模块才能正确解析和丰富日志数据
2. 缺少字段映射：原始日志需要被解析为结构化字段才能在 Kibana 中搜索
3. ECS 字段缺失：未使用模块会导致缺少 event.module 等关键字段

1. 创建模块配置目录

mkdir -p /opt/host1_filebeat/config/modules.d

2. 配置系统日志模块

sudo nano /opt/host1_filebeat/config/modules.d/system.yml

- module: systemsyslog:enabled: truevar.paths: ["/host/var/log/syslog*"]auth:enabled: truevar.paths: ["/host/var/log/auth.log*"]

mysql

sudo nano /opt/host1_filebeat/config/modules.d/mysql.yml

- module: mysqlerror:enabled: truevar.paths: ["/host/mysql_logs/error.log*"]slowlog:enabled: truevar.paths: ["/host/mysql_logs/slow.log*"]

2.2.3 创建 Docker Compose

创建启动脚本 /opt/host1_filebeat/run_filebeat.sh：

sudo nano /opt/host1_filebeat/docker-compose.yml

内容如下：

services:filebeat:image: elastic/filebeat:8.17.2hostname: host1  # 强制设置容器主机名container_name: host1_filebeatrestart: alwaysuser: rootvolumes:- ./config/filebeat.yml:/usr/share/filebeat/filebeat.yml:ro- ./config/modules.d:/usr/share/filebeat/modules.d:ro  # 新增此行- /var/log:/host/var/log:ro- app_tomcat_logs:/host/tomcat_logs:ro  # 使用相同的卷名称- app_mysql_logs:/host/mysql_logs:ro- ./logs:/usr/share/filebeat/logs- ./data:/usr/share/filebeat/dataenvironment:- "setup.dashboards.enabled=false"- HOSTNAME=host1 # 覆盖环境变量- TZ=Asia/Shanghaicommand: ["-e", "--strict.perms=false"]networks:- host1-netvolumes:app_tomcat_logs:  # 声明相同的卷（如果已存在会自动复用）external: true  # 使用外部已存在的卷app_mysql_logs:  # 声明相同的卷（如果已存在会自动复用）external: true  # 使用外部已存在的卷networks:host1-net:driver: bridge

2.2.4 启动 Filebeat 服务

# 启动服务
sudo docker compose up -d# 查看状态
sudo docker compose ps

2.3 验证部署

2.3.1 检查容器状态

docker ps -f name=host1_filebeat --format "table {{.Names}}\t{{.Status}}"

预期输出：

NAMES            STATUS
host1_filebeat   Up X minutes

2.3.2 查看容器日志

docker logs -f host1_filebeat | grep -A 5 "Connected to"

正常应包含：

Connected to 192.168.0.224:5044
Publishing events...

2.3.3 生成测试日志

# 系统日志
echo "$(date) HOST1_MODULE_TEST_SYSTEM" | sudo tee -a /var/log/syslog
echo "$(date +'%b %d %H:%M:%S') [WARN] HOST1_MODULE_TEST_SYSTEM1" | sudo tee -a /var/log/syslog
echo "$(date +'%b %d %H:%M:%S') [ERROR] HOST1_MODULE_TEST_SYSTEM2" | sudo tee -a /var/log/syslog# Tomcat 日志
docker exec ry-tomcat sh -c 'echo "[$(date)] INFO HOST1_MODULE_TEST_TOMCAT" >> /opt/tomcat/logs/catalina.out'# MySQL 日志
docker exec ry-mysql sh -c 'echo "[$(date)] [ERROR] HOST1_MODULE_TEST_MYSQL" >> /var/log/mysql/error.log'

2.4 Kibana 验证

2.4.1 访问 Kibana

1. 打开浏览器访问 http://192.168.0.224:5601
2. 进入 Discover 页面

2.4.2 筛选 host1 日志

在查询栏输入：

HOST1_MODULE_TEST_*

应能看到测试日志条目

2.4.3 验证日志路径

fields.log_source : "host1-tomcat"  # 检查Tomcat日志
fields.log_source : "host1-mysql"   # 检查MySQL日志

2.5 排错指南

问题1：日志未采集

# 检查容器内文件是否存在
docker exec host1_filebeat ls /host/var/log# 检查文件权限
docker exec host1_filebeat ls -l /host/var/log/syslog

问题2：Logstash 连接失败

# 在容器内测试连接
docker exec host1_filebeat nc -zv 192.168.0.224 5044# 检查防火墙
sudo ufw status | grep 5044

问题3：字段未显示

# 检查字段映射
docker exec host1_filebeat filebeat test config -c /usr/share/filebeat/filebeat.yml

---

此配置方案保持了与 ELK 服务器相同的 Docker 化部署方式，通过卷映射实现日志采集，并通过自定义字段 log_source 区分不同主机的日志源。后续添加新主机时，只需复制此结构并修改 IP 和日志路径即可。

三、在 Logstash 中配置过滤器（Filter）解析日志

3.1 配置索引模版

索引模板的作用

索引模板用于定义 Elasticsearch 中索引的结构和行为，主要功能包括：

功能	说明	重要性
字段映射	定义字段类型（keyword/text/date等）	✅ 防止字段类型冲突
动态模板	自动处理新字段的映射规则	✅ 避免 “mapping explosion”
索引设置	配置分片数、副本数、刷新间隔等	⚙️ 优化性能
别名管理	为索引组创建统一别名	🔗 简化查询
生命周期	自动管理索引的生命周期	📅 自动化维护

在您的场景中，模板专门用于确保：

1. event.module 字段被正确映射为 keyword 类型
2. 所有字符串字段默认设为 keyword 而非 text
3. 为日志索引提供一致的字段结构

为什么您的 curl 命令失败

错误信息 unknown field [mappings] 表示：

1. API 版本不匹配：Elasticsearch 8.x 的索引模板 API 格式有变化
2. 格式错误：新版本要求不同的 JSON 结构

正确配置索引模板

步骤 1：创建正确的模板文件

nano /opt/monitor/elk/elasticsearch/logs-template.json

{"index_patterns": ["logs-*"],"template": {"mappings": {"dynamic_templates": [{"strings_as_keywords": {"match_mapping_type": "string","mapping": {"type": "keyword","ignore_above": 1024}}}],"properties": {"@timestamp": {"type": "date"},"message": {"type": "text","fields": {"keyword": {"type": "keyword"}}}}}},"priority": 200,"_meta": {"description": "Custom template for logs indices"}
}

步骤 2：应用模板到 Elasticsearch

# 使用正确的 API 端点
curl -XPUT "http://localhost:9200/_index_template/logs-template" \-H "Content-Type: application/json" \-d @/opt/monitor/elk/elasticsearch/logs-template.json

步骤 3：验证模板是否应用成功

curl -XGET "http://localhost:9200/_index_template/logs-template?pretty"

预期输出应包含您定义的映射规则。

替代方案：使用 Filebeat 自带模板

如果您不需要高度定制，可以使用 Filebeat 内置模板：

# 在 Filebeat 容器中执行
docker exec elk_filebeat filebeat setup --index-management# 输出示例
Loaded index template

模板应用后的效果验证

1. 创建新索引

curl -XPUT "http://localhost:9200/logs-test-2025.06.12"

2. 检查映射

curl -XGET "http://localhost:9200/logs-test-2025.06.12/_mapping?pretty"

为什么需要模板？实际案例分析

假设没有模板时：

1. 第一条日志是简单消息："Test log"
   • Elasticsearch 创建 message 字段为 text 类型
2. 第二条日志是 JSON：{"user": "Alice"}
   • Elasticsearch 尝试将 user 字段设为 text
3. 第三条日志：{"user": {"name": "Bob"}}
   • 冲突！无法将 text 类型改为 object
   • 结果：数据丢失或解析失败

使用模板后：

1. 预先定义字段类型
2. 通过 dynamic_templates 处理未知字段
3. 确保数据结构一致性

生产环境建议

1. 模板版本控制：

   "_meta": {"version": "1.0","created_by": "elk-admin"
   }
   

2. 生命周期策略：

   "settings": {"index.lifecycle.name": "logs_policy"
   }
   

3. 组件集成：

   # Logstash 配置
   output {elasticsearch {template => "/path/to/logs-template.json"template_name => "logs-template"}
   }
   

3.1 更新 Logstash 输入配置(服务端-host4)

nano /opt/monitor/elk/logstash/pipeline/logstash.conf

input {beats {port => 5044client_inactivity_timeout => 600}
}filter {# ====== 基础字段提取 ======# 1. 提取系统日志时间戳和消息主体grok {match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{GREEDYDATA:log_message}"}tag_on_failure => []  # 不记录失败}# 2. 时间戳处理date {match => [ "syslog_timestamp", "MMM dd HH:mm:ss", "MMM  d HH:mm:ss" ]timezone => "Asia/Shanghai"target => "@timestamp"}# 3. 清理临时字段mutate {remove_field => [ "syslog_timestamp", "agent", "ecs", "input", "log" ]}# ====== 日志级别解析 ====== (核心改进)# 1. 尝试从已有字段提取日志级别if [log.level] {mutate {uppercase => [ "log.level" ]}} # 2. 尝试从消息中提取日志级别else {grok {patterns_dir => [ "/usr/share/logstash/patterns" ]match => { "log_message" => [# 常见格式1: [ERROR] message"\[%{LOGLEVEL:log.level}\]",# 常见格式2: ERROR: message"^%{LOGLEVEL:log.level}:",# 常见格式3: level=ERROR"level=%{LOGLEVEL:log.level}\b",# 常见格式4: <timestamp> <host> <program>[pid]: ERROR message (syslog)"%{SYSLOGHOST} %{PROG}(?:\[%{POSINT}\])?: %{LOGLEVEL:log.level} %{GREEDYDATA}"]}break_on_match => truetag_on_failure => []}}# 3. 标准化日志级别if [log.level] {mutate {gsub => ["log.level", "^(?i)tr(ace)?$", "TRACE","log.level", "^(?i)dbg$|^debug?$", "DEBUG","log.level", "^(?i)inf(o)?$", "INFO","log.level", "^(?i)warn(ing)?$", "WARN","log.level", "^(?i)err(or)?$", "ERROR","log.level", "^(?i)crit(ical)?$", "CRITICAL","log.level", "^(?i)fatal$|^emerg(ency)?$|^severe$", "FATAL"]}} # 4. 标记无法识别的日志级别else {mutate {add_field => { "log.level" => "UNKNOWN" }}}# ====== 日志来源处理 ======# 1. 设置基础模块if [fields][log_source] {mutate {add_field => { "[event][module]" => "%{[fields][log_source]}" }}}# 2. 特定模块处理## 系统日志处理if [fields][log_source] == "host1-system" {grok {patterns_dir => [ "/usr/share/logstash/patterns" ]match => { "log_message" => "%{SYSLOG5424PRI}?%{SYSLOGTIMESTAMP} %{SYSLOGHOST:syslog_host} %{PROG:syslog_program}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}" }overwrite => [ "syslog_message" ]}}## MySQL 日志处理else if [fields][log_source] == "host1-mysql" {mutate {replace => { "[event][module]" => "mysql" }add_field => { "[event][dataset]" => "mysql.error" }}grok {patterns_dir => [ "/usr/share/logstash/patterns" ]match => { "log_message" => "(?<mysql_time>%{YEAR}-%{MONTHNUM}-%{MONTHDAY} %{TIME}) %{LOGLEVEL} \[%{DATA:mysql_component}\] %{GREEDYDATA:mysql_message}" }}date {match => [ "mysql_time", "ISO8601" ]target => "@timestamp"remove_field => [ "mysql_time" ]}}## Tomcat 日志处理else if [fields][log_source] == "host1-tomcat" {mutate {replace => { "[event][module]" => "tomcat" }add_field => { "[event][dataset]" => "tomcat.access" }}grok {patterns_dir => [ "/usr/share/logstash/patterns" ]match => { "log_message" => "\[%{DATA:thread}\] %{LOGLEVEL} %{JAVACLASS:class} - %{GREEDYDATA:tomcat_message}" }}}# ====== 最终清理和优化 ======# 1. 保留原始消息mutate {rename => { "message" => "original_message""log_message" => "message"}}# 2. 添加主机标识if [fields][host_id] {mutate {add_field => { "[host][id]" => "%{[fields][host_id]}" }}}# 3. 调试信息 (生产环境可注释掉)mutate {add_field => {"[debug][processing_time]" => "%{@timestamp}""[debug][config_version]" => "2025-06-15-v2"}}
}output {elasticsearch {hosts => ["http://elasticsearch:9200"]index => "logs-%{+YYYY.MM.dd}"data_stream => false}# 调试输出 (生产环境建议关闭)stdout {codec => rubydebug {metadata => true  # 显示元数据}}
}

1. system 模块为何自动工作：

   • System 模块是 Filebeat 的内置官方模块
   • 它包含预定义的处理器链，会自动添加 event.module 和 event.dataset 字段
   • 这些模块有专门的字段映射模板

2. mysql/tomcat 模块为何不工作：

   • 虽然启用了模块，但可能：
     • 模块处理器链未正确触发
     • 日志格式不完全匹配模块的 grok 模式
     • 缺少 Elasticsearch 索引模板支持
   • 自定义模块（如您的 tomcat）没有内置处理器链

3. module 字段的重要性：

   重要性	说明
   ✅ 日志分类	区分不同来源的日志（系统/应用/数据库）
   ✅ 仪表盘集成	Kibana 官方仪表盘依赖此字段
   ✅ 告警规则	基于模块创建特定告警
   ✅ 权限控制	按模块设置访问权限
   🔶 非必需	技术上可不使用，但强烈推荐

手动添加模块字段原因：

1. Filebeat 模块系统局限：
   • 只对完全匹配的日志格式自动处理
   • 自定义日志路径可能导致处理器链不触发
2. 字段映射问题：
   • 第一个索引的映射决定了字段类型
   • 如果第一条日志没有 event.module，后续可能无法添加

3.2 重启 Logstash

cd /opt/monitor/elk
docker compose restart logstash

3.3 验证 Logstash 配置

docker compose logs -f logstash | grep "Pipeline started"

预期输出：

[INFO]  Pipeline started successfully {:pipeline_id=>"main"}

---

3.4 在 Elasticsearch 中确认索引已创建

1 检查 ES 索引

curl -XGET 'http://localhost:9200/_cat/indices?v' | grep logs-

预期输出：

green  open logs-2025.06.10  1 1 1000 0 1.5mb 750kb

2 检查索引映射

curl -XGET 'http://localhost:9200/logs-2025.06.10/_mapping?pretty'

确认包含 event.module, log.level 等字段

---

五、在 Kibana 中创建数据视图（Data View）

5.1 访问 Kibana

打开浏览器访问：http://192.168.0.224:5601

5.2 创建数据视图

1. 左侧菜单 > Management > Stack Management
2. 选择 Kibana > Data Views
3. 点击 “Create data view”
4. 输入：
   • Name: logs-*
   • Index pattern: logs-*
   • Timestamp field: @timestamp
5. 点击 “Save data view to Kibana”

5.3 验证数据视图

在 Discover 页面：

1. 左上角选择 logs-* 数据视图

2. 在搜索栏输入：

   event.module: "system" and host.name: "host1"
   

3. 应看到来自 host1 的系统日志

---

六、使用 Kibana 分析日志

6.1 基本搜索

# 查找特定错误
log.level: "ERROR"# 查找特定主机
host.name: "host1"# 查找特定模块
event.module: "mysql"

准备工作：创建索引模式

1. 进入索引管理：
   • 左侧菜单 → Management → Stack Management → Data Views
   • 点击 Create data view
   • 名称：logs-*
   • 索引模式：logs-*
   • 时间字段：@timestamp
   • 点击 Create data view

6.2 使用 Lens 创建可视化

进入可视化创建界面

1. 左侧导航栏 → Analytics → Dashboards
2. 点击 Create dashboard
3. 在仪表板编辑界面，点击 Add panel → Create visualization
4. 选择 Lens（推荐）或 Custom visualization

注意：8.x 版本推荐使用 Lens，它替代了旧版 Visualize Builder

示例1：日志级别分布（饼图）

1. 选择图表类型：
   • 选择 Pie
2. 配置数据层：
3. save
4. 保存到库 Save to library：
   • 名称：Log Level Distribution
     

---

示例2：错误日志趋势（折线图）

1. 选择图表类型：

   • 点击 Chart type → Line

2. 配置坐标轴：

   轴	字段
   X轴	@timestamp
   Y轴	Count

3. 添加筛选器：

   • 点击 log.level.keyword
   • 筛选：ERROR

4. 保存到库 Save to library：

   • 名称：Error Log Trend

---

6.3 使用 Custom visualization (Vega) 创建高级图表

官网案例: Example Gallery | Vega

示例：主机日志量热力图

1. Add panel：
   • 选择 Custom visualization
2. 输入 Vega 代码：

{"$schema": "https://vega.github.io/schema/vega/v5.json","description": "日志量热力图 - 按小时和主机","autosize": "none","width": 800,"height": 500,"padding": 40,  // 增加内边距确保标签显示"signals": [{"name": "palette","value": "Viridis","bind": {"input": "select","options": ["Viridis","Plasma","Inferno","Magma","Cividis"]}}],"data": [{"name": "logs","url": {"index": "logs-*","body": {"size": 0,"aggs": {"hosts": {"terms": {"field": "host.id.keyword","size": 5,"order": {"_key": "asc"}  // 确保主机有序},"aggs": {"hours": {"date_histogram": {"field": "@timestamp","calendar_interval": "hour","min_doc_count": 0}}}}}}},"format": {"property": "aggregations.hosts.buckets"},"transform": [{"type": "flatten","fields": ["hours.buckets"],"as": ["hour"]},{"type": "formula","as": "host","expr": "datum.key"},{"type": "formula","as": "hour_of_day","expr": "hours(datum.hour.key)"},{"type": "formula","as": "count","expr": "datum.hour.doc_count"}]}],"scales": [{"name": "x","type": "band","domain": {"data": "logs", "field": "host"},"range": "width","padding": 0.1},{"name": "y","type": "band","domain": [0,1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23],"range": "height","padding": 0.1},{"name": "color","type": "linear","range": {"scheme": {"signal": "palette"}},"domain": {"data": "logs", "field": "count"},"nice": true,"zero": true}],"axes": [{"orient": "bottom","scale": "x","title": "主机","labelAngle": -45,"labelLimit": 100,"zindex": 1  // 确保轴标签显示在最上层},{"orient": "left","scale": "y","title": "小时","labelFontSize": 12,"encode": {"labels": {"update": {"text": {"signal": "datum.value === 0 ? '午夜' : datum.value === 12 ? '中午' : datum.value < 12 ? datum.value + '时' : (datum.value - 12) + '时'"}}}}}],"legends": [{"fill": "color","type": "gradient","title": "日志数量","titleFontSize": 12,"titlePadding": 10,"gradientLength": 300,"orient": "right","offset": 10}],"marks": [{"type": "rect","from": {"data": "logs"},"encode": {"enter": {"x": {"scale": "x", "field": "host"},"y": {"scale": "y", "field": "hour_of_day"},"width": {"scale": "x", "band": 1},"height": {"scale": "y", "band": 1},"stroke": {"value": "white"},"strokeWidth": {"value": 0.5}},"update": {"fill": {"scale": "color", "field": "count"},"tooltip": {"signal": "{'主机': datum.host, '小时': datum.hour_of_day, '日志数量': datum.count}"}}}},{"type": "text","from": {"data": "logs"},"encode": {"enter": {"x": {"scale": "x", "field": "host", "band": 0.5},"y": {"scale": "y", "field": "hour_of_day", "band": 0.5},"fill": {"value": "#000"},"align": {"value": "center"},"baseline": {"value": "middle"},"fontSize": {"value": 10}},"update": {"text": {"signal": "datum.count > 0 ? datum.count : ''"}}}}]
}

保存到库 Save to library：

• 名称：Host Log Heatmap

为什么需要热力图？

1. 高效诊断：秒级定位问题主机+问题时段
2. 预防性维护：发现潜在问题模式
3. 资源优化：精准调整资源分配
4. 安全防护：识别异常行为模式
5. 业务洞察：理解用户行为时间模式