[TOC]

Beats 收集、解析和发送组件

winlogbeat - 简述与使用

描述: 我们可以利用 Winlogbeat 来进行 Windows 日志监视，大致流程是在要监视的每个系统上安装Winlogbeat指定日志文件的位置将日志数据解析为字段并发送到Elasticsearch可视化Kibana中的日志数据。

下载地址: https://www.elastic.co/downloads/beats/winlogbeat
帮助文档: https://www.elastic.co/guide/en/beats/winlogbeat/current/winlogbeat-installation-configuration.html

Tips: 你需要进行提前准备Elasticsearch来存储和搜索你的数据和Kibana来可视化和管理它。

安装流程:

• Step 1.从下载页面下载Winlogbeat zip文件,并解压至任意目录此处我解压到d:\logs\目录之中然后再重命名winlogbeat-<version>目录为Winlogbeat.

• Step 2.以管理员身份打开PowerShell提示（右键单击PowerShell图标并选择以管理员身份运行）. 切换路径并运行以下命令安装服务。

  PS C:\Users\Administrator> cd 'd:\logs\Winlogbeat'
  PS d:\logs\Winlogbeat> .\install-service-winlogbeat.ps1
  
  # - 如果系统上禁用了脚本执行，则需要为当前会话设置执行策略以允许脚本运行。
  PowerShell.exe -ExecutionPolicy UnRestricted -File .\install-service-winlogbeat.ps1.

• Step 3.在winlogbeat.yml中设置elasticsearch、kibana连接信息

  # 需要连接到Elasticsearch和Kibana才能设置Winlogbeat
  # - elasticsearch
  output.elasticsearch:
    hosts: ["myEShost:9200"]
    username: "winlogbeat_internal"
    password: "YOUR_PASSWORD"
  
  # - kibana
  setup.kibana:
    host: "mykibanahost:5601" 
    username: "my_kibana_user"  
    password: "{pwd}"

Tips : 我们可以将敏感值存储在secrets密钥库中参考地址。

# Create a keystore
winlogbeat keystore create

# Add keys (When prompted enter a value for the key.)
winlogbeat keystore add ES_PWD   
winlogbeat keystore add ES_PWD --force  # To overwrite an existing key’s value
cat /file/containing/setting/value | winlogbeat keystore add ES_PWD --stdin --force
  # Successfully updated the keystore
  
# List keysedit
winlogbeat keystore list

# Remove keysedit
winlogbeat keystore remove ES_PWD

# 例如，假设keystore包含一个名为ES_PWD的密钥，其值为 yourelasticsearchpassword
In the configuration file, use output.elasticsearch.password: "${ES_PWD}"
On the command line, use: -E "output.elasticsearch.password=\${ES_PWD}"

• Step 4.配置Winlogbeatedit在winlogbeat.yml中配置要监视的事件日志。
  描述: 在 winlogbeat.event_log下指定要监视的事件日志列表, 默认情况下 Winlogbeat 监视应用程序、安全性和系统日志。

  # - 1.在winlogbeat.event_log下指定要监视的事件日志列表，默认监控 application, security, and system logs 我们可以根据需要进行添加相应事件
  winlogbeat.event_logs:
    - name: Application
    - name: Security
    - name: System
  
  # - 2.(可选)设置日志选项以将Winlogbeat日志写入文件：
  
  logging.to_files: true
  logging.files:
    path: C:\ProgramData\winlogbeat\Logs
  logging.level: info

Tips: 我们还可获取可用事件日志的列表 Get-EventLog * 命令

Get-EventLog * | Select-Object Log,LogDisplayName
  # Log                    LogDisplayName
  # ---                    --------------
  # Application            应用程序
  # HardwareEvents         硬件事件
  # Internet Explorer      Internet Explorer
  # Key Management Service 密钥管理服务
  # OAlerts                Microsoft Office Alerts
  # Security
  # System                 系统
  # Windows PowerShell     Windows PowerShell

• Step 5.保存配置文件后，使用以下命令对其进行测试,并Winlogbeat加载附带用于解析、索引和可视化数据的预定义资产。

  # 解析验证配置文件
  PS d:\logs\Winlogbeat> .\winlogbeat.exe test config -c .\winlogbeat.yml -e
  
  # 加载推荐的索引模板以写入Elasticsearch，并部署示例仪表板以可视化Kibana中的数据。
  PS > .\winlogbeat.exe setup -e
    # Overwriting ILM policy is disabled. Set `setup.ilm.overwrite: true` for enabling.
    # Index setup finished.
    # Loading dashboards (Kibana must be running and reachable)
    # Loaded dashboards

• Step 6.启动服务或者停止服务，然后观察Kibana中主机数据的变化。

  Start-Service winlogbeat
  Stop-Service winlogbeat

winlogbeat 常用命令
描述: 除了上述的相关命令外，winlogbeat语法参数如下:

# - 索引模板加载
PS D:\Pro> Expand-Archive -Path .\winlogbeat-7.14.1-windows-x86_64.zip -DestinationPath .

winlogbeat.yml 语法
参考地址: https://www.elastic.co/guide/en/beats/winlogbeat/current/configuration-winlogbeat-options.html#configuration-winlogbeat-options-event_logs-name

# ==== Winlogbeat specific option ====
# - 存储重新启动后用于恢复监视的信息的文件名，作为windows服务运行时建议您将该值设置为c:/ProgramData/winlogbeat/.winlogbeat.yml
winlogbeat.registry_file: C:/ProgramData/winlogbeat/.winlogbeat.yml
# - 关闭时等待所有事件发布的时间量 (默认情况下没有关闭超时)
winlogbeat.shutdown_timeout: 30s
# - 条目列表（称为词典指定要监视的事件日志)。
winlogbeat.event_logs:
  - name: Application   # - 要监视的事件日志的名称
    ignore_older: 72h   # - 如果指定了此选项将筛选早于指定时间量的事件，当您开始监视包含要忽略的旧记录的事件日志时此选项非常有用。
    provider:           # - 要包含的提供程序（源名称）的列表（即日志来源）
      - Application Error
      - Application Hang
      - Windows Error Reporting
      - EMET
  - name: System
    batch_read_size: 1000  # - 在单个批处理中从Windows API读取的最大事件日志记录数, 默认100一定要小于1024否则保存
    event_id: 6005,6006,7001,7002
  - name: Security
    processors:                     # - 要应用于事件日志生成的数据的处理器列表（主要解决windows系统限制超过22个）。
      - drop_event.when.not.or:     # - 使用drop_event(不存在则丢弃)处理器进行筛选解决此Windows限制问题。
        - equals.winlog.event_id: 903
        - equals.winlog.event_id: 1024
        - equals.winlog.event_id: 4624
    level: critical, error, warning, info ,verbose
  - name: ForwardedEvents       # - 指示日志仅包含使用Windows事件收集器从远程主机收集的事件。
    api: wineventlog-experimental # - 将选择用于从Windows API读取事件的事件日志读取器实现。（只应在测试实验功能时设置此选项）
    tags: [forwarded]           # - 标记使得在Kibana中选择特定事件或Logstash中的applyconditional过滤变得容易标记将被附加到常规配置中指定的标签列表中。 
  - name: Windows PowerShell       # - 必须在配置文件中指定通道的全名。
    event_id: 400, 403, 600, -800, # - 包含和排除（阻止）事件ID的列表, 排除事件ID为800的事件
  - name: Microsoft-Windows-PowerShell/Operational
    event_id: 4103, 4104, 4105, 4106
  - name: Microsoft-Windows-Windows Defender/Operational
    include_xml: true              # - 用于控制事件的原始XML表示是否包含在由Winlogbeat发送的数据中。默认值为false
  - name: CustomLog
    tags: ["web"]
    fields:         # - 您可以指定用于向输出添加附加信息的可选字段
      customer_id: 51415432 
    fields_under_root: true # - 如果此选项设置为true，则自定义领域存储为输出文档中的顶级字段，而不是分组到fields小词典。
    index: "%{[agent.name]}-myindex-%{+yyyy.MM.dd}"  # 可能会扩展到 “winlogbeat-myindex-2019.12.13”
    # 如果存在，此格式化字符串将覆盖此事件日志中事件的索引（对于elasticsearch输出），或设置raw_index事件的元数据字段（用于其他输出）
    keep_null: true # 如果此选项设置为true，则字段null值将在输出文档中发布。默认情况下，保持空值设置为false no_more_events  # 当事件日志读取器从Windows接收到没有其他事件可读取的信号时应执行的操作。也可以wait为了编写更多的事件（默认行为），或者停止. 当所有单个事件日志读取器都停止时，overallWinlogbeat进程将停止。

Tips : 如果指定的事件ID超过22个要包含或排除的事件ID超过22个，Windows将阻止Winlogbeat读取事件日志，因为它限制了事件日志查询中可以使用的条件数。
解决办法:

winlogbeat.event_logs:
  - name: Security
    processors:
      - drop_event.when.not.or:
        - equals.winlog.event_id: 903
        - equals.winlog.event_id: 1024
        - equals.winlog.event_id: 4624

Filebeat - 简述与使用

描述: Filebeat 是一个轻量级的传送器，用于转发和集中日志数据, 该模块收集并解析常见 Unix/Linux 分发的系统日志服务创建的日志。

参考地址: https://www.elastic.co/guide/en/beats/filebeat/current/index.html

Filebeat 工作原理:

• (1) 当您启动 Filebeat 时，它会查看您指定的日志数据位置。
• (2) Filebeat 定位到的每个日志，Filebeat 都会启动一个采集进程。
• (3) 每个 Harvester 读取新内容的单个日志并将新日志数据发送到Fliebeat Spooler(后台服务),并将将聚合数据发送到输出你为 Filebeat 配置的es中。

Filebeat-工作流程

Tips: Filebeat is an Elastic Beat. It’s based on the libbeat framework.

Filebeat 安装:
描述: 此处以Linux平台为例进行安装 Filebeat, 如您是其它平台或者采用其它方式(APT/YUM/Docker)则参考以下地址。
参考地址: https://www.elastic.co/guide/en/beats/filebeat/current/filebeat-installation-configuration.html

http://openjdk.java.net/

for i in {78..83};do
ssh -p 20211 10.4.4.${i} ‘wget http://10.4.4.21:8000/filebeatInstall.sh -O /tmp/filebeatInstall.sh && chmod +x /tmp/filebeatInstall.sh && cd /tmp/ && bash -c /tmp/filebeatInstall.sh’
done
for i in {16..20};do
ssh -p 20211 10.41.40.${i} ‘systemctl restart filebeat.service’
done

#!/bin/bash
# Filebeat 快速下载安装部署
SRCNAME="filebeat-7.14.1"
SRCDIR="/usr/local/${SRCNAME}"
export SRCDIR
# 第 1 步：filebeat 下载&安装
# curl -L -O https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.14.1-linux-x86_64.tar.gz
curl -L -O http://192.168.12.31/${SRCNAME}-linux-x86_64.tar.gz
tar xzf ${SRCNAME}-linux-x86_64.tar.gz -C /usr/local
mv /usr/local/${SRCNAME}-linux-x86_64 ${SRCDIR}
cd ${SRCDIR}


# 第 2 步：filebeat 安全配置
${SRCDIR}/filebeat keystore create
echo "weiyigeek-2021.1" | ${SRCDIR}/filebeat keystore add AUTH --stdin --force
${SRCDIR}/filebeat keystore list


# 第 3 步：filebeat 配置&验证
sed -i -e 's|#host: "localhost:5601"|host: "log.weiyigeek.top:5601"|g' \
-e 's|localhost:9200|log.weiyigeek.top:9200|g' \
-e 's|#username: "elastic"|username: "elastic"|g' \
-e 's|#password: "changeme"|password: "${AUTH}"|g' \
${SRCDIR}/filebeat.yml
${SRCDIR}/filebeat test config -e ${SRCDIR}/filebeat.yml

# 第 4 步：设置采集的资产
# Filebeat 带有预定义的资产，用于解析、索引和 可视化您的数据。 
# 查看可用模块：${SRCDIR}/filebeat modules list
sudo ${SRCDIR}/filebeat modules enable system
sudo chown root ${SRCDIR}/filebeat.yml 
sudo chown root ${SRCDIR}/modules.d/system.yml 
sudo ${SRCDIR}/filebeat setup


# 第 5 步：设置自启与启动 Filebeat
sudo tee /usr/lib/systemd/system/filebeat.service <<'EOF'
[Unit]
Description=Elastic kibana Filebeat Service
Documentation=https://elastic.co
After=network.target

[Service]
Type=simple
StandardError=journal
ExecStart=SRCDIR/filebeat -c SRCDIR/filebeat.yml
Restart=on-failure
RestartSec=3s

[Install]
WantedBy=multi-user.target
EOF
sed -i "s#SRCDIR#${SRCDIR}#g" /usr/lib/systemd/system/filebeat.service 
systemctl daemon-reload
systemctl start filebeat.service

# 第 6 步：在 Kibana 查看您的数据
# Filebeat 带有预构建的 Kibana 仪表板和用于可视化日志的 UI 数据。
systemctl status filebeat.service
echo -e "\e[32m* 请启动 Kibana 并打开仪表板 -> 发现 -> 要查看 Filebeat 数据，请执行 确保预定义 filebeat-* 索引模式被选中\e[0m"

入坑出坑

问题1.Kibana无法登陆且在登陆时显示糟糕！错误请重试,后台日志报index ... blocked by: [TOO_MANY_REQUESTS/12/disk usage exceeded flood-stage watermark, index has read-only-allow-delete block]

• 错误日志: TOO_MANY_REQUESTS/12/disk usage exceeded flood-stage watermark, index has read-only-allow-delete block

  {"type":"log","@timestamp":"2021-10-09T08:01:04+00:00","tags":["error","plugins","security","session","index"],"pid":1213,"message":"Failed to create s
  ession value: cluster_block_exception: [cluster_block_exception] Reason: index [.kibana_security_session_1] blocked by: [TOO_MANY_REQUESTS/12/disk usag
  e exceeded flood-stage watermark, index has read-only-allow-delete block];"}

• 错误原因: 由于ES存储的data磁盘空间使用率已达到95%则将索引设置成为只读模式。
• 解决办法:

  # 1.增加磁盘空间大小(操作前注意数据迁移备份)或者删除无用的日志文件
  root@elk:~# ls /app/
   elastic   kibana  '#recycle'
  root@elk:~# df -Th /app/
  Filesystem                  Type  Size  Used Avail Use% Mounted on
  192.168.2.1:/volume1/share nfs    19T  3.1T   16T  16% /app
  
  # 2.临时在在elasticsearch.yml文件末尾添加以下内容,等等开了进去Kibana后删除无用索引。
  # 控制洪水阶段水印最大百分比
  cluster.routing.allocation.disk.watermark.flood_stage: 98%
  # 根据磁盘使用情况来决定是否继续分配shard
  cluster.routing.allocation.disk.threshold_enabled: false
  
  curl -H "Content-Type:application/json" \
      -XPUT -u elastic:weiyigeek 'http://log.weiyigeek.top:9200/_cluster/_settings' \ 
      -d '{"transient": {"cluster.routing.allocation.disk.watermark.flood_stage": "99%"}}'
  
  # 3.临时解锁指定Kibana的索引的读写权限(其它索引被锁类似)
  curl -H "Content-Type:application/json" -XPUT -u elastic:weiyigeek 'http://log.weiyigeek.top:9200/.kibana_7.14.1_001/_settings' -d '{ "index.blocks.read_only_allow_delete" : null }'
  curl -H "Content-Type:application/json" -XPUT -u elastic:weiyigeek 'http://log.weiyigeek.top:9200/.kibana_security_session_1/_settings' -d '{ "index.blocks.read_only_allow_delete" : null }'