【ELK(Elasticsearch+Logstash+Kibana) 从零搭建实战记录：日志采集与可视化】_elk搭建
ELK(Elasticsearch+Logstash+Kibana) 从零搭建实战记录：日志采集与可视化
本文记录了我在搭建ELK(Elasticsearch, Logstash, Kibana)技术栈时的完整实战过程。使用Docker Compose快速搭建了ELK服务端（监控主机），并通过Filebeat实现了对自身系统日志以及另一台Web主机（host1）上Tomcat应用日志、MySQL数据库日志的集中采集。文中详细包含了：
- 环境规划与Docker Compose部署： 基于
elk_es,elk_logstash,elk_kibana容器的详细配置。 - Filebeat配置详解： 服务端采集自身日志 + 客户端(host1)采集Tomcat/MySQL日志的关键步骤。
- Logstash核心技巧： 输入协议选择(
beatsvstcp)、日志解析过滤(grok,date,mutate)、日志级别提取与模块化处理。 - Kibana实战： 创建数据视图(Data View)、利用Lens制作日志级别饼图与错误趋势图，并探索了Vega热力图高级可视化。
整体架构梳理
- 日志采集（Beats）
通常使用 Filebeat、Metricbeat、Packetbeat 等轻量级采集器（Beats Family）将宿主机或应用产生的日志采集并发送到 Logstash 或直接发送到 Elasticsearch。 - 日志处理（Logstash）
Logstash 负责对接收到的日志进行过滤、解析、转换（Grok、Date、Mutate 等 Filter 插件），并根据配置将最终格式化好的事件写入 Elasticsearch（或其他输出）。 - 数据存储与检索（Elasticsearch）
Elasticsearch 集群负责接收 Logstash 送来的日志文档，存储在对应的 Index 中，并为上层 Kibana 提供检索与聚合能力。 - 可视化与分析（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.2 dns: 8.8.8.8 container_name: elk_es environment: - 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=false volumes: - es-data:/usr/share/elasticsearch/data ports: - 9200:9200 networks: - elk-net deploy: resources: limits: memory: 2.5G cpus: \'1.5\' kibana: image: kibana:8.17.2 dns: 8.8.8.8 container_name: elk_kibana environment: - ELASTICSEARCH_HOSTS=http://elasticsearch:9200 - TZ=Asia/Shanghai - ELASTICSEARCH_REQUESTTIMEOUT=120000 # 2分钟超时 - SERVER_STARTUP_TIMEOUT=120000 # 2分钟启动超时 ports: - 5601:5601 volumes: - kibana-data:/usr/share/kibana/data depends_on: - elasticsearch networks: - elk-net deploy: resources: limits: memory: 1.5G cpus: \'0.5\' logstash: image: logstash:8.17.3 dns: 8.8.8.8 container_name: elk_logstash environment: # 统一内存设置，移除冲突配置 - LS_JAVA_OPTS=-Xms1g -Xmx1g - TZ=Asia/Shanghai - pipeline.workers=2 - pipeline.batch.size=125 volumes: - ./logstash/pipeline/logstash.conf:/usr/share/logstash/pipeline/logstash.conf - logstash-data:/usr/share/logstash/data - ./logstash/patterns:/usr/share/logstash/patterns # 添加此行 ports: - 5044:5044 networks: - elk-net deploy: resources: limits: memory: 1G cpus: \'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.confinput { tcp { port => 5044 codec => 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.3docker pull kibana:8.17.2docker pull elasticsearch:8.17.2# 应用内核参数sudo sysctl -w vm.max_map_count=262144echo \"vm.max_map_count=262144\" | sudo tee -a /etc/sysctl.conf#启动服务docker compose up -ddocker 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\" ]}EOFsudo systemctl daemon-reload && sudo systemctl restart docker1.2 验证与集成
1.2.1 检查服务状态
docker ps --format \"table {{.Names}}\\t{{.Status}}\\t{{.Ports}}\"预期输出：
NAMES STATUS PORTSelk_es Up 2 minutes 0.0.0.0:9200->9200/tcpelk_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.2sudo apt updatesudo apt install filebeat=8.17.2说明：
如果你的服务器是 CentOS/RHEL，需要使用
rpm方式安装：bash复制编辑sudo rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearchsudo 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.2 dns: 8.8.8.8 container_name: elk_es environment: - 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=false volumes: - es-data:/usr/share/elasticsearch/data ports: - 9200:9200 networks: - elk-net deploy: resources: limits: memory: 2.5G cpus: \'1.5\' kibana: image: kibana:8.17.2 dns: 8.8.8.8 container_name: elk_kibana environment: - ELASTICSEARCH_HOSTS=http://elasticsearch:9200 - TZ=Asia/Shanghai - ELASTICSEARCH_REQUESTTIMEOUT=120000 # 2分钟超时 - SERVER_STARTUP_TIMEOUT=120000 # 2分钟启动超时 ports: - 5601:5601 volumes: - kibana-data:/usr/share/kibana/data depends_on: - elasticsearch networks: - elk-net deploy: resources: limits: memory: 1.5G cpus: \'0.5\' logstash: image: logstash:8.17.3 dns: 8.8.8.8 container_name: elk_logstash environment: # 统一内存设置，移除冲突配置 - LS_JAVA_OPTS=-Xms1g -Xmx1g - TZ=Asia/Shanghai - pipeline.workers=2 - pipeline.batch.size=125 volumes: - ./logstash/pipeline/logstash.conf:/usr/share/logstash/pipeline/logstash.conf - logstash-data:/usr/share/logstash/data - ./logstash/patterns:/usr/share/logstash/patterns # 添加此行 ports: - 5044:5044 networks: - elk-net deploy: resources: limits: memory: 1G cpus: \'0.5\' filebeat: image: elastic/filebeat:8.17.2 dns: 8.8.8.8 hostname: host4 # 强制设置容器主机名 container_name: elk_filebeat user: root volumes: - ./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/data depends_on: - logstash networks: - elk-net environment: - HOSTNAME=host4 # 覆盖环境变量 - TZ=Asia/Shanghai command: [ \"-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: 500M cpus: \'0.25\'volumes: es-data: kibana-data: logstash-data: filebeat-data:networks: elk-net: driver: bridge2.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/*.yml reload.enabled: falsesetup.template.enabled: falsesetup.dashboards.enabled: false############################ 输入（Inputs） ############################filebeat.inputs: - type: log enabled: true paths: # 收集宿主机 /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: 3 backoff: init: 1s max: 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: system syslog: enabled: true var.paths: [\"/host/var/log/syslog*\"] auth: enabled: true var.paths: [\"/host/var/log/auth.log*\"]如果要采集 Nginx 访问日志，类似地创建或启用 nginx.yml：
- module: nginx access: enabled: true var.paths: [\"/host/var/log/nginx/access.log*\"] error: enabled: true var.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 { … } }nc host port 推送纯文本/JSON 行codec => json_lines），没有 ackbeats {}，不能当普通 TCP 用2.1.4 启动并验证
-
回到
docker-compose.yml所在目录，执行：docker compose pull # 可选，拉取最新镜像docker compose downdocker compose up -d这时会依次启动：
elasticsearch→kibana→logstash→filebeat。 -
查看 Filebeat 容器日志，确认其已成功连接 Logstash：
docker logs -f elk_filebeat你应该能看到类似：
Starting server on port: 5044 -
验证 Logstash 是否收到并转发事件：
docker logs -f elk_logstash | grep \"Received\"你会看到 Logstash 打印它接收到了来自 Filebeat 的事件。
-
确认 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 Current Dload Upload Total Spent Left Speed 100 2625 0 2625 0 0 102k 0 --:--:-- --:--:-- --:--:-- 102k yellow open logs-2025.06.09 veFnpfCrSd-eLY98mvrHGQ 1 1 150182 0 179.7mb 179.7mb 179.7mb yellow 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。 -
-
在 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” 页面就能实时看到采集到的日志条目
-
测试数据
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/*.yml reload.enabled: false # 禁用自动重载，避免状态冲突 reload.period: 30ssetup.template.enabled: falsesetup.dashboards.enabled: false############################ 主机标识配置 ############################name: \"host1\"fields: host_id: \"host1\" host_type: \"web\" environment: \"production\"############################ 输入配置 ############################filebeat.inputs: # 系统日志 - type: log paths: - /host/var/log/*.log - /host/var/log/*/*.log fields: log_source: \"host1-system\" ignore_older: 72h # Tomcat 日志 (原始采集，不使用模块) - type: log paths: - /host/tomcat_logs/catalina.out - /host/tomcat_logs/localhost_access_log*.txt fields: log_source: \"host1-tomcat\" log_type: \"raw\" # 添加类型标识 # MySQL 日志 (使用模块) - type: log paths: - /host/mysql_logs/error.log - /host/mysql_logs/slow.log fields: log_source: \"host1-mysql\" log_type: \"module\" # 标识使用模块处理############################ 输出配置 ############################output.logstash: hosts: [\"192.168.0.224:5044\"] # 增加重试设置 retry: 3 backoff: init: 1s max: 60s############################ 日志配置 ############################logging.level: infologging.to_files: truelogging.files: path: /usr/share/filebeat/logs name: filebeat keepfiles: 3 # 减少保留文件数 permissions: 06442.1.3 为 host1 配置 Filebeat 模块
- 未启用解析模块：Filebeat 需要加载模块才能正确解析和丰富日志数据
- 缺少字段映射：原始日志需要被解析为结构化字段才能在 Kibana 中搜索
- ECS 字段缺失：未使用模块会导致缺少
event.module等关键字段
1. 创建模块配置目录
mkdir -p /opt/host1_filebeat/config/modules.d2. 配置系统日志模块
sudo nano /opt/host1_filebeat/config/modules.d/system.yml- module: system syslog: enabled: true var.paths: [\"/host/var/log/syslog*\"] auth: enabled: true var.paths: [\"/host/var/log/auth.log*\"]mysql
sudo nano /opt/host1_filebeat/config/modules.d/mysql.yml- module: mysql error: enabled: true var.paths: [\"/host/mysql_logs/error.log*\"] slowlog: enabled: true var.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.2 hostname: host1 # 强制设置容器主机名 container_name: host1_filebeat restart: always user: root volumes: - ./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/data environment: - \"setup.dashboards.enabled=false\" - HOSTNAME=host1 # 覆盖环境变量 - TZ=Asia/Shanghai command: [\"-e\", \"--strict.perms=false\"] networks: - host1-netvolumes: app_tomcat_logs: # 声明相同的卷（如果已存在会自动复用） external: true # 使用外部已存在的卷 app_mysql_logs: # 声明相同的卷（如果已存在会自动复用） external: true # 使用外部已存在的卷networks: host1-net: driver: bridge2.2.4 启动 Filebeat 服务
# 启动服务sudo docker compose up -d# 查看状态sudo docker compose ps2.3 验证部署
2.3.1 检查容器状态
docker ps -f name=host1_filebeat --format \"table {{.Names}}\\t{{.Status}}\"预期输出：
NAMES STATUShost1_filebeat Up X minutes2.3.2 查看容器日志
docker logs -f host1_filebeat | grep -A 5 \"Connected to\"正常应包含：
Connected to 192.168.0.224:5044Publishing events...2.3.3 生成测试日志
# 系统日志echo \"$(date) HOST1_MODULE_TEST_SYSTEM\" | sudo tee -a /var/log/syslogecho \"$(date +\'%b %d %H:%M:%S\') [WARN] HOST1_MODULE_TEST_SYSTEM1\" | sudo tee -a /var/log/syslogecho \"$(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
- 打开浏览器访问
http://192.168.0.224:5601 - 进入 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 中索引的结构和行为，主要功能包括：
在您的场景中，模板专门用于确保：
event.module字段被正确映射为keyword类型- 所有字符串字段默认设为
keyword而非text - 为日志索引提供一致的字段结构
为什么您的 curl 命令失败
错误信息 unknown field [mappings] 表示：
- API 版本不匹配：Elasticsearch 8.x 的索引模板 API 格式有变化
- 格式错误：新版本要求不同的 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\"为什么需要模板？实际案例分析
假设没有模板时：
- 第一条日志是简单消息：
\"Test log\"- Elasticsearch 创建
message字段为text类型
- Elasticsearch 创建
- 第二条日志是 JSON：
{\"user\": \"Alice\"}- Elasticsearch 尝试将
user字段设为text
- Elasticsearch 尝试将
- 第三条日志：
{\"user\": {\"name\": \"Bob\"}}- 冲突！无法将
text类型改为object - 结果：数据丢失或解析失败
- 冲突！无法将
使用模板后：
- 预先定义字段类型
- 通过
dynamic_templates处理未知字段 - 确保数据结构一致性
生产环境建议
-
模板版本控制：
\"_meta\": { \"version\": \"1.0\", \"created_by\": \"elk-admin\"} -
生命周期策略：
\"settings\": { \"index.lifecycle.name\": \"logs_policy\"} -
组件集成：
# Logstash 配置output { elasticsearch { template => \"/path/to/logs-template.json\" template_name => \"logs-template\" }}
3.1 更新 Logstash 输入配置(服务端-host4)
nano /opt/monitor/elk/logstash/pipeline/logstash.confinput { beats { port => 5044 client_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: [pid]: ERROR message (syslog) \"%{SYSLOGHOST} %{PROG}(?:\\[%{POSINT}\\])?: %{LOGLEVEL:log.level} %{GREEDYDATA}\" ] } break_on_match => true tag_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\" => \"(?%{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 # 显示元数据 } }}-
system 模块为何自动工作：
- System 模块是 Filebeat 的内置官方模块
- 它包含预定义的处理器链，会自动添加
event.module和event.dataset字段 - 这些模块有专门的字段映射模板
-
mysql/tomcat 模块为何不工作：
- 虽然启用了模块，但可能：
- 模块处理器链未正确触发
- 日志格式不完全匹配模块的 grok 模式
- 缺少 Elasticsearch 索引模板支持
- 自定义模块（如您的 tomcat）没有内置处理器链
- 虽然启用了模块，但可能：
-
module 字段的重要性：
重要性 说明 ✅ 日志分类 区分不同来源的日志（系统/应用/数据库） ✅ 仪表盘集成 Kibana 官方仪表盘依赖此字段 ✅ 告警规则 基于模块创建特定告警 ✅ 权限控制 按模块设置访问权限 🔶 非必需 技术上可不使用，但强烈推荐
手动添加模块字段原因：
- Filebeat 模块系统局限：
- 只对完全匹配的日志格式自动处理
- 自定义日志路径可能导致处理器链不触发
- 字段映射问题：
- 第一个索引的映射决定了字段类型
- 如果第一条日志没有
event.module，后续可能无法添加
3.2 重启 Logstash
cd /opt/monitor/elkdocker compose restart logstash3.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 750kb2 检查索引映射
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 创建数据视图
- 左侧菜单 > Management > Stack Management
- 选择 Kibana > Data Views
- 点击 “Create data view”
- 输入：
- Name:
logs-* - Index pattern:
logs-* - Timestamp field:
@timestamp
- Name:
- 点击 “Save data view to Kibana”
5.3 验证数据视图
在 Discover 页面：
-
左上角选择
logs-*数据视图 -
在搜索栏输入：
event.module: \"system\" and host.name: \"host1\" -
应看到来自 host1 的系统日志
六、使用 Kibana 分析日志
6.1 基本搜索
# 查找特定错误log.level: \"ERROR\"# 查找特定主机host.name: \"host1\"# 查找特定模块event.module: \"mysql\"准备工作：创建索引模式
- 进入索引管理：
- 左侧菜单 → Management → Stack Management → Data Views
- 点击 Create data view
- 名称：
logs-* - 索引模式：
logs-* - 时间字段：
@timestamp - 点击 Create data view
6.2 使用 Lens 创建可视化
进入可视化创建界面
- 左侧导航栏 → Analytics → Dashboards
- 点击 Create dashboard
- 在仪表板编辑界面，点击 Add panel → Create visualization
- 选择 Lens（推荐）或 Custom visualization
注意：8.x 版本推荐使用 Lens，它替代了旧版 Visualize Builder
示例1：日志级别分布（饼图）
- 选择图表类型：
- 选择 Pie
- 配置数据层：
- save
- 保存到库 Save to library：
- 名称：
Log Level Distribution
- 名称：
示例2：错误日志趋势（折线图）
-
选择图表类型：
- 点击 Chart type → Line
-
配置坐标轴：
轴 字段 X轴 @timestamp Y轴 Count -
添加筛选器：
- 点击
log.level.keyword - 筛选：
ERROR
- 点击
-
保存到库 Save to library：
- 名称：
Error Log Trend
- 名称：
6.3 使用 Custom visualization (Vega) 创建高级图表
官网案例: Example Gallery | Vega
示例：主机日志量热力图
- Add panel：
- 选择 Custom visualization
- 输入 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
为什么需要热力图？
- 高效诊断：秒级定位问题主机+问题时段
- 预防性维护：发现潜在问题模式
- 资源优化：精准调整资源分配
- 安全防护：识别异常行为模式
- 业务洞察：理解用户行为时间模式
