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

发布于：2025-06-26 ⋅ 阅读:(19) ⋅ 点赞:(0)

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.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.conf
input {
  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.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.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: 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/*.yml
  reload.enabled: false

setup.template.enabled: false
setup.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: true

processors:
  - 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 { … } }
协议 原生 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  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。

  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/*.yml
  reload.enabled: false  # 禁用自动重载，避免状态冲突
  reload.period: 30s

setup.template.enabled: false
setup.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: info
logging.to_files: true
logging.files:
  path: /usr/share/filebeat/logs
  name: filebeat
  keepfiles: 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: 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-net

volumes:
  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 => 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: <timestamp> <host> <program>[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" => "(?<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. 业务洞察：理解用户行为时间模式

网站公告

今日签到

点亮在社区的每一天
去签到

热门文章

最新发布