Logstash + Filebeat

在使用前，请先阅读数据模型和数据格式的介绍。

概述

神策分析支持使用 Logstash + Filebeat 的方式将 后端数据实时 导入神策分析。

Logstash 是由 Elastic 公司推出的一款开源的服务器端数据处理管道，能够同时从多个来源采集数据，转换数据，然后将数据发送指定的存储库中。Logstash 官方介绍。

Filebeat 是 Elastic 公司为解决 Logstash "太重" 的问题推出的一款轻量级日志采集器，在处理数量众多的服务器、虚拟机和容器生成的日志时可使用 Logstash + Filebeat 的日志采集方式。Filebeat 官方介绍。

基于 Logstash + Filbeat 的数据采集流程为：后端 SDK 生成数据文件 => Filebeat 读取文件 => Logstash Beat input => Logstash sensors_analytic output => 神策分析。

结构如下图所示：

本文将介绍以下三个场景中如何使用 Logstash + Filebeat 完成数据采集并发送至神策分析。

服务器场景中的数据采集。
使用 Docker 容器化场景下的数据采集。
使用 K8s（Kubernetes）自动编排容器场景中的数据采集。

在阅读详细方案前，请先阅读 Logstash 和 Filebeat 的使用说明与版本支持信息。

Logstash 使用说明

无论使用哪种方案 Logstash 都必须装有 sensors_analytics_output 插件。

Logstash 下载与安装

请参考 installing Logstash 官方说明文档，选择您喜欢的下载与安装方式。

安装 logstash-output-sensors_analytics 插件

该插件将检查数据是否为 Json 格式，并加入一些神策需要的字段值如 lib、data 等，打包数据，压缩并经过 base64 之后发送至神策的数据接收地址。
插件已经发布至 Ruby 官方公共库，Github repository : logstash-output-sensors_analytics 。直接在 Logstash 目录下执行安装即可，安装需要一段时间，请耐心等待。

bin/logstash-plugin install logstash-output-sensors_analytics

在安装完成后执行：

bin/logstash-plugin list

看见新安装的插件 logstash-output-sensors_analytics 证明安装成功。

插件在使用时直接配置在 output 里即可

output{
    sensors_analytics {
        url => "https://example.sensorsdata.cn/sa"
    }
}

sensors_analytics 参数说明：

参数名	类型	必须	说明
url	list	是	神策分析的数据接收地址，完整的 url 地址并以 sa 结尾，有端口号需要加上端口号。例如 :url => "https://example.sensorsdata.cn/sa" 。集群走内网 IP 上报的，可同时配置多个数据接收地址，例如：url => ["http://10.120.157.227:8106/sa","http://10.120.72.166:8106/sa"]。
project	string	否	项目名，不写默认为 default ，配置后会覆盖事件中和 url 中指定的 project。优先级为：project 参数配置 > 事件中指定 > url 中指定。
flush_interval_sec	number	否	触发 flush 间隔的时间（单位：秒），默认值为 2。
flush_batch_size	number	否	触发批量发送的最大 record 数量，默认值为 100 。
enable_filebeat_status_report	boolean	否	默认开启，在日志展示中一分钟内活动的 Filebeat 读取状态。

Logstash 配置

Logstash Pipeline 配置

Logstash 支持同时运行多个 Pipeline ，各个 Pipeline 之间互不影响，拥有各自独立的输入输出配置，Pipeline 的配置文件位于 config/pipelines.yml 。如果您目前正在使用 Logstash 完成一些其他的日志采集工作，可以在原有的 Logstash 上新增一条 Pipeline 专门负责收集神策的日志数据，并发送至神策分析。

pipelines.yml 参考示例：

# 原来使用的 Pipeline 配置
- pipeline.id: elastic-output
  pipeline.workers: 4
  path.config: "/home/app/logstash/elastic_output.config"

# 新增的 sensorsdata 的管道配置
- pipeline.id: sensorsdata-output
  # 使用不同的 Logstash 配置
  pipeline.workers: 1
  queue.type: persisted
  # 使用不同的输入输出配置
  path.config: "/home/app/logstash/beat_sa_output.config"

注意：神策分析日志的输入要不同于其他导入到 Logstash 的 Piplines 输入。比如之前使用的日志导入方式是通过 Filebeat 采集并发送到 Logstash 的 5044 端口，那么负责采集神策日志的 Filebeat 可以将数据发送至 5055 端口，从而能够应用 id = sensorsdata-output 的管道。

更多信息 Pipeline 信息请参考 : Multiple-Pipelines 官方说明文档。

Logstash 输入输出配置

配置中主要包含 input、filter 和 output 三部分，Logstash 处理神策的日志数据只需配置 input 和 output 即可

beat_sa_output.conf 参考示例：

# 使用 beats 作为输入
input {
    beats {
        port => "5044"
    }
}
# 使用 sensors_analytics 作为输出
output{
    sensors_analytics {
        url => "https://example.sensorsdata.cn/sa"
    }
}

提醒：在使用 logstash-file-input-plugin 时当 logstash 处于关闭状态下时重命名已读文件（重命名后依然符合 pattern），启动后会导致重复读取。

Logstash 运行配置

Logstash 默认使用 config/logstash.yml 作为运行配置。

这里需要注意的是：

1. 在需要保证数据导入顺序的情况下请更改配置 pipeline.workers 的值为 1。配置项 pipeline.workers 的值默认为 cpu 的核心数，当 workers 的值大于 1 时，会导致处理数据的顺序发生变化。
2. 为保证数据的传输不会因为程序的意外终止而丢失，请设置 queue.type: persisted，该配置为 Logstash 使用的缓冲队列类型，这样配置可在重启 Logstash 后继续发送缓冲队列中的数据。 queue.type 的默认值为 memory （基于内存的）。
3. 建议设置 queue.drain 的值为 true ，该配置项会使 Logstash 在正常退出之前将所有缓冲队列中的数据全部发送完毕。

更多 logstash.yml 信息请参考 : logstash.yml 官方说明文档。

Logstash 启动

直接启动，会使用 config/pipelines.yml 作为 Pipeline 配置和运行配置。

bin/logstash

指定 ~/logstash/beat_sa_output.conf 为输入输出配置文件启动，会使用 config/logstash.yml 作为运行配置。

bin/logstash -f ~/logstash/beat_sa_output.conf

通过命令参数指定输入输出启动，会使用 config/logstash.yml 作为运行配置。

bin/logstash -e 'output { sensors_analytics { url => "https://example.sensorsdata.cn/sa" }}'

更多启动相关信息请参考 : Getting Started with Logstash 官方说明文档。

Logstash 进度

Logstash 在使用 Filebeat 作为输入时文件的读取进度是由 Filebeat 进行控制的，当使用其他的输入方式时，例如 Logstash 读取文件，消费 Kafka 等，数据的读取进度存放在 Logstash 目录 data/plugins 下，基于硬盘的数据缓冲队列存放在 data/queue 中。可在 logstash.yml 中配置 path.data 来指定 Logstash 启动时使用的 data/ 目录的位置。

sensors-output-plugin 升级与回滚

已安装插件升级至最新版本

bin/logstash-plugin update logstash-output-sensors_analytics

安装指定版本的插件

# v0.1.0
bin/logstash-plugin install --version 0.1.0 logstash-output-sensors_analytics
# v0.1.2
bin/logstash-plugin install --version 0.1.2 logstash-output-sensors_analytics
# v0.1.4
bin/logstash-plugin install --version 0.1.4 logstash-output-sensors_analytics

卸载插件

bin/logstash-plugin remove logstash-output-sensors_analytics

Filebeat 使用说明

Filebeat 下载与安装

请参考：Install Filebeat 官方说明文档。选择您喜欢的下载与安装方式。

Filebeat 配置

使用 Filebeat 读取后端 SDK 产生的埋点日志文件。Filebeat 默认配置文件为：filebeat.yml 。修改配置文件请使用 log 类型作为 Filebeat 的输入，paths 指定数据文件所在的位置，使用通配符 `*` 匹配后端 SDK 输出的文件名路径。

Filebeat 的输入输出配置 `filebeat.yml` 参考示例：

# Filebeat 收集 /var/logs/ 目录下所有以 service_log. 开头的数据文件
filebeat.shutdown_timeout: 5s
filebeat.inputs:
- type: log
  paths:
    - /var/logs/service_log.*

# 将数据发送至地址为 10.42.32.70:5044 或 10.42.50.1:5044 的 logstash
output.logstash: 
  hosts: ["10.42.32.70:5044","10.42.50.1:5044"]

需要注意的是：

导入的数据必须是神策的数据格式。
在需要保证导入顺序的情况下不要额外设置 loadbalance : true ，当配置了多个 Logstash hosts 作为数据接收端时该设置会使用轮询的方式将数据发送至所有的 Logstash 这很可能导致数据的顺序被打乱。Filebeat 的默认配置为 loadbalance : false 。
Filebeat 的文件读取进度存放在 data/registry 目录下，在启动时用于恢复进度。
在 Filebeat 运行时间，避免使用 vim 之类的可能生成文件副本的编辑器编辑文件，Filebeat 会读取目录中临时生成的文件。
建议新增配置项 filebeat.shutdown_timeout: 5s ，filebeat 在退出时有可能造成少量数据重复。

更多配置相关信息请参考：Filebeat 官方文档。

启动 Filebeat

./filebeat -e -c filebeat.yml

-c 用于指定 filebeat.yml 配置文件的位置，-e 可在终端上显示 Filebeat 的日志信息。

Filebeat 进度

如果你的目录下有多个文件未被读取，filebeat 会同时读取多个文件，文件的读取进度存放在 Filebeat 目录下 data/registry 中，重启 Filebeat 时会根据进度继续执行发送。

服务器场景下的数据采集

如果您生产日志的后端应用直接部署在服务器上，本节内容将介绍如何使用 Filebeat + Logstash 采集产生的日志数据。该场景下也可使用 LogAgent 完成日志的收集工作。

部署 Logstash

如果您已经在使用 Logstash 做一些其他的日志收集工作请参考 Logstash 配置。

参考 Logstash 使用说明直接在您的一台或多台服务器上部署 Logstash 。

Logstash 输入输出配置 logstash.conf 示例：

# 使用 beats 作为输入
input {
    beats {
        port => "5044"
    }
}
# 使用 sensors_analytics 作为输出
output{
    sensors_analytics {
        url => "https://example.sensorsdata.cn/sa"
    }
}

Logstash 运行配置 logstash.yml 示例：

pipeline.workers: 1
queue.type: persisted
queue.drain: true

启动 Logstash 应用 logstash.conf ，更多启动方式请参考 Logstash 启动和 Logstash 官方文档。

bin/logstash -f logstash.conf

部署 Filebeat

在会产生埋点日志的服务器上部署 Filebeat 采集指定目录下的日志发送至神策分析。

神策分析各后端语言的 SDK 都支持将数据写入文件，例如使用 Java SDK 的 ConcurrentLoggingConsumer，PHP SDK 的 FileConsumer，Python SDK 的 LoggingConsumer 它们能将日志文件写入指定的目录下。

以 Java SDK 为例：

// 使用 ConcurrentLoggingConsumer 初始化 SensorsAnalytics
// 将数据输出到 /data/sa_log 下的 service_log 开头的文件中，每天一个文件
final SensorsAnalytics sa = new SensorsAnalytics(
        new SensorsAnalytics.ConcurrentLoggingConsumer("/data/sa_log/service_log"));

// 使用神策分析记录用户行为数据
sa.track(distinctId, true, "UserLogin");
sa.track(distinctId, true, "ViewProduct");

// 程序结束前，停止神策分析 SDK 所有服务
sa.shutdown();

以上配置将在 /data/sa_log 目录下生成数据文件，一天一个文件，文件列表如：

service_log.20170923
service_log.20170924
service_log.20170925

Filebeat 通过配置 filebeat.yml 读取目录 /data/sa_log 下的以 service_log. 开头的日志文件，发送至部署好的 Logstash 。

filebeat.yml 参考示例：

# Filebeat 收集 /data/sa_log 目录下所有以 service_log. 开头的数据文件
filebeat.inputs:
- type: log
  paths:
    - /data/sa_log/service_log.*

# 将数据发送至地址为 10.42.32.70:5044 或 10.42.50.1:5044 的 Logstash
output.logstash: 
  hosts: ["10.42.32.70:5044","10.42.50.1:5044"]

当在一台服务器上有多个产生日志的目录时可配置 Filebeat 同时读取多个目录。

读取多目录 filebeat.yml 参考示例：

filebeat.inputs:
- type: log
  paths:
    # 收集 /data/sa_log 目录下所有以 service_log. 开头的数据文件
    - /data/sa_log/service_log.*
    # 收集 /another/logs/ 目录下所有以 sdk_log. 开头的数据文件
    - /another/logs/sdk_log.*

# 将数据发送至地址为 10.42.32.70:5044 或 10.42.50.1:5044 的 Logstash
output.logstash: 
  hosts: ["10.42.32.70:5044","10.42.50.1:5044"]

后台启动 Filebeat

nohup ./filebeat -c filebeat.yml > /dev/null 2>&1 &

使用 Docker 容器化场景下的数据采集

部署 Logstash

为保证 Logstash 的稳定工作，建议直接部署 Logstash，下文为 docker 部署方式仅供参考。

如果您已经在使用 Logstash 做一些其他的日志收集工作请参考 Logstash 配置。为避免容器意外关闭导致丢失数据，请设法保存缓冲区内的数据。

首先，获取一个具有 sensors_analytics output 插件的 Logstash 镜像

方式一：直接下载我们已经安装好的插件的 Logstash 镜像。

docker pull sensorsdata/logstash:latest

方式二：自行制作带有 sensors_analytics output 插件的 Logstash 镜像。

Dockerfile 示例：

FROM docker.elastic.co/logstash/logstash:7.2.0

RUN /usr/share/logstash/bin/logstash-plugin install logstash-output-sensors_analytics

准备需要的配置文件。

Logstash 的输入输出配置 logstash.conf 示例：

input {
    beats {
        port => "5044"
    }
}
output {
    sensors_analytics{
        url => "http://10.42.34.189:8106/sa?project=default"
        project => "default"
    }
}

Logstash 的运行配置 logstash.yml 示例：

pipeline.workers: 1
queue.type: persisted
queue.drain: true

由于 Logstash 需要使用磁盘做缓冲队列，这里我们创建一个 Volume 专门用于保存 Logstash 的进度和缓冲队列，当重启该 Logstash 容器时请复用该 Volume 。

docker volume create logstash-data

在启动该容器时挂载配置文件和存放缓存队列的数据卷。

启动命令参考示例：

docker run -d -p 5044:5044 --name logstash \
--mount source=logstash-data,target=/usr/share/logstash/data \
-v ~/local/logstash/logstash.conf:/usr/share/logstash/pipeline/logstash.conf \ 
-v ~/local/logstash/logstash.yml:/usr/share/logstash/config/logstash.yml \ sensorsdata/logstash:latest

部署 Filebeat

方案一：在 SDK 容器中安装 Filebeat 采集日志并发送至 Logstash（推荐）

在您能够产生埋点日志的容器上安装一个 Filebeat 采集日志并发送至部署好的 Logstash，Filebeat 为一款轻量级的日志采集器，运行内存大概 10 MB 左右，并不会给您的工作容器带来太多的负担。

优点：部署方便，不用担心 Filebeat 的进度问题。
缺点：侵入了原有的 SDK 容器。

下面以 JavaSDK 作为工作容器举例：

Dockerfile 示例：

FROM centos

ADD jdk-8u211-linux-x64.tar.gz /usr/local/

ENV JAVA_HOME /usr/local/jdk1.8.0_211
ENV CLASSPATH $JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
ENV PATH $PATH:$JAVA_HOME/bin

COPY javasdk.jar /home

# 在容器中安装一个 Filebaet
ADD filebeat-7.2.0-linux-x86_64.tar.gz /home
# 一份默认的配置文件 
COPY filebeat.yml /etc

COPY run.sh /home
WORKDIR /home
# 在 run.sh 中启动 SDK 进程和 Filebeat 进程
CMD ["/bin/bash","-e","run.sh"]

run.sh 中的内容：

#!/bin/bash
nohup java -jar javasdk.jar > /dev/null 2>&1 &
nohup filebeat-7.2.0-linux-x86_64/filebeat -c /etc/filebeat.yml > /dev/null 2>&1 &

while [[ true ]]; do
    sleep 10000
done

在容器中要保证 SDK 的日志写入路径与 Filebeat 的日志读取路径相同。

以 Java SDK 为例：

// 使用 ConcurrentLoggingConsumer 初始化 SensorsAnalytics
// 将数据输出到 /data/sa_log 下的 service_log 开头的文件中，每天一个文件
final SensorsAnalytics sa = new SensorsAnalytics(
        new SensorsAnalytics.ConcurrentLoggingConsumer("/data/sa_log/service_log"));

// 使用神策分析记录用户行为数据
sa.track(distinctId, true, "UserLogin");
sa.track(distinctId, true, "ViewProduct");

// 程序结束前，停止神策分析 SDK 所有服务
sa.shutdown();

Filebeat 配置示例：

# Filebeat 收集 /data/sa_log 目录下所有以 service_log. 开头的数据文件
filebeat.inputs:
- type: log
  paths:
    - /data/sa_log/service_log.*

# 将数据发送至地址为 10.42.32.70:5044 或 10.42.50.1:5044 的 Logstash
output.logstash: 
  hosts: ["10.42.32.70:5044","10.42.50.1:5044"]

启动命令参考示例：

docker run -d --name sdk-beat \
-v ~/local/filebeat/filebeat.yml:/filebeat.yml \
sdk-beat

方案二：SDK 使用共享数据保存日志 Filebeat 进行读取

后端 SDK 和 Filebeat 分别运行在不同的容器上，SDK 将生产的日志存放在数据卷上，Filebeat 从数据卷内读取数据发送至部署好的 Logstash 。

优点：不会侵入原有的 SDK 容器。
缺点：使用起来比较麻烦。

首先，创建一个数据卷选择您喜欢的存储方式，下面以本地磁盘为例，要保证你的容器对该数据卷有写权限：

docker volume create sa-log

启动后端 SDK 容器，将产生日志目录挂载到数据卷上。

docker run -d --name sdk \
--mount source=sa-logs,target=/your/logs/path \
your-sdk-image

启动 Filebeat 容器，将日志读取目录挂载到数据卷上。同时将存放文件读取进度的目录也挂载到数据卷上，以每一个数据卷为一个读取进度，当重启 Filebeat 时复用该进度即可继续执行发送。

配置文件 filebeat.yml 示例：

filebeat.inputs:
- type: log
  paths:
    - /usr/share/filebeat/input/service_log.*

# 将数据发送至地址为 10.42.32.70:5044 或 10.42.50.1:5044 的 Logstash
output.logstash: 
  hosts: ["10.42.32.70:5044","10.42.50.1:5044"]

docker run -d --name filebeat \
--mount source=sa-logs,target=/usr/share/filebeat/input \
--mount source=sa-logs,target=/usr/share/filebeat/data/ \
-v ~/docker_workspace/filebeat/filebeat.yml:/usr/share/filebeat/filebeat.yml \
docker.elastic.co/beats/filebeat:7.2.0

如果想用多个 SDK 容器挂载同一个数据卷的，建议容器以环境变量 HOSTNAME 为路径名存放日志文件，再将上级目录挂载到数据卷上。

举个例子：

容器内日志的输出路径：/mount/${HOSTNAEM}-logs/service_log.20190708

将容器中的 /mount 目录挂载至 Volume 。

因此 Volume 中日志的目录的存放格式为：

|-- Volume
| |-- c1369239e7ba-logs
| |-- fcdfdb3bdb2b-logs
| | |-- service_logs.20190702
| | |-- service_logs.20190703
| |-- da86e6ba6ca1-logs
| | |-- service_logs.20190701
| | |-- service_logs.20190702
| | |-- service_logs.20190703

更改 Filebeat 的文件读取路径为:

filebeat.inputs:
- type: log
  paths:
    - /usr/share/filebeat/input/*/service_log.*

以 Java SDK 为例生成带有 HOSTNAME 的路径存放日志：

// 获取 HOSTNAME
String hostname = System.getenv("HOSTNAME");
File logPath = new File("/mount/" + hostname + "-logs/");
if (!logPath.exists()) {
    logPath.mkdirs();
}
// 使用 ConcurrentLoggingConsumer 初始化 SensorsAnalytics
// 将数据输出到 /mount/${HOSTNAME}-logs/ 目录下以 service_log 为开头保存，运行容器时将 /mount 目录挂载到宿主机上
final SensorsAnalytics sa = new SensorsAnalytics(
        new SensorsAnalytics.ConcurrentLoggingConsumer(logPath.getAbsolutePath() + "/service_log"));

// 使用神策分析记录用户行为数据
sa.track(distinctId, true, "ViewProduct");

// 程序结束前，停止神策分析 SDK 所有服务
sa.shutdown();

如果您不希望更改原容器的日志路径存放方式，可以在容器启动时建立一条软链指向日志目录，将软链挂载到 Volume 上。

rm -rf /your/logs/path \
&& mkdir -p /mount/${HOSTNAME}_logs \
&& ln -s /mount/${HOSTNAME}_logs /your/logs/path \
&& bin/sdk start

在启动容器时将 /mount 挂载到 Volume 中。

docker run -d --name sdk \
--mount source=sa-logs,target=/mount \
your-sdk-image

使用 K8s（Kubernetes）自动编排容器场景中的数据采集

Logstash 部署

为保证 Logstash 的稳定工作，建议直接部署 Logstash 在服务器上，下文为 K8s 的部署方式供参考。
如果您已经在使用 Logstash 做一些其他的日志收集工作请参考 Logstash 配置。为避免容器意外关闭导致丢失数据，请设法保存缓冲区内的数据。

注册一份 Logstash 配置文件，使用 Filebeat 作为输入，sensors_analytics 作为输出，并指定运行配置。

配置文件 logstash-conf.yaml 参考示例：

apiVersion: v1
kind: ConfigMap
metadata:
  name: logstash-set
  labels:
    sa-app: logstash
data:
  logstash.yml: |-
    http.host: 0.0.0.0
    pipeline.workers: 1
    queue.type: persisted
    # 用于限制缓冲队列的大小默认值为 1024MB ，该数值在设置时应该小于 Pod 使用的存储卷大小。
    queue.max_bytes: 900mb
    queue.drain: true
---
apiVersion: v1
kind: ConfigMap
metadata:
  name: logstash-pipe-conf
  labels:
    sa-app: logstash
data:
  logstash.conf: |-
    input {
      beats {
        port => "5044"
      }
    }

    output {
      sensors_analytics {
          url => "http://10.42.34.189:8106/sa?project=default"
      }
    }

为了不丢失数据，使用了基于硬盘的数据缓冲队列 (queue.type: persisted)，所以需要在容器外保存 Logstash 的进度信息，这样在重启 Logstash 的时候可以继续完成发送。

建议通过 StatefulSet 的方式进行部署从而保存 Logstash 的状态。

首先，创建一个 StorageClass 用于生成保存进度的 PV ，设置手动回收，下面以 NFS 为例。

logstash-sc.yaml 的参考示例如下：

apiVersion: storage.k8s.io/v1
kind: StorageClass
metadata:
  name: logstash-nfs-storage
provisioner: nfs-provisioner
reclaimPolicy: Retain

然后，创建一个 StatefulSet 应用 logstash-nfs-storage ，通过 Headless Service 来为每个 Logstash Pod 提供网络访问方式。

logstash-sts.yaml 参考示例如下：

apiVersion: v1
kind: Service
metadata:
  name: logstash
  labels:
    app: logstash
spec:
  ports:
  - port: 5044
    name: beat-in
  clusterIP: None
  selector:
    app: logstash
---
apiVersion: apps/v1beta1
kind: StatefulSet
metadata:
  name: logstash
spec:
  # 使用上面的 Headless Service
  serviceName: "logstash"
  selector:
    matchLabels:
      app: logstash
  replicas: 3
  template:
    metadata:
      labels:
        app: logstash
    spec:
      containers:
      - name: logstash
        image: sensorsdata/logstash:latest
        ports:
        - containerPort: 5044
          name: beat-in
        volumeMounts:
        - name: logstash-pipe-conf
          mountPath: /usr/share/logstash/pipeline/logstash.conf
          subPath: logstash.conf
        - name: logstash-set
          mountPath: /usr/share/logstash/config/logstash.yml
          subPath: logstash.yml
        # 容器中 /usr/share/logstash/data 目录下保存着缓冲队列 ，与进度信息。
        - name: ldata
          mountPath: /usr/share/logstash/data
      volumes:
      - name: logstash-pipe-conf
        configMap:
          name: logstash-pipe-conf
      - name: logstash-set
        configMap:
          name: logstash-set
  volumeClaimTemplates: # Logstash 进度数据使用的 PVC 模板
  - metadata:
      name: ldata
    spec:
      accessModes: [ "ReadWriteOnce" ]
      # 使用的存储类名称，需要提前创建。
      storageClassName: "logstash-nfs-storage"
      resources:
        requests:
          # 大小要高于缓冲队列的最大长度限制
          storage: 1Gi

StatefulSet 创建完成后 Pod name 的生成规则为 StatefulSetName - Pod - 序号。

上面的配置文件会生成 logstash-0、logstash-1，logstash-2 这样命名的 Pod。 Pod 副本也是按照序号 0 到 N-1 的顺序依次进行创建的，在删除时是按照序号 N-1 到 0 依次删除。

Headless Service 为控制的每个 Pod 副本创建了一个 DNS 域名，完整的域名规则为：(pod name).(headless server name).namespace.svc.cluster.local，因此 Filebeat 是通过域名来寻找 Logstash 的，而不是 IP 。当使用默认的 namespace 时可省略 namespace.svc.cluster.local 。

StatefulSet 根据 volumeClaimTemplates，为每个 Pod 创建一个 PVC，PVC 的命名前缀为：namespace-volumeMounts.name - volumeClaimTemplates.name - pod_name，删除一个 Pod 副本不会删除 PVC ，在重启后新的 Pod 会复用之前 PVC 中的进度继续完成发送。

创建完成后检查一下运行的情况：

kubectl get pods -l app=logstash
NAME READY STATUS RESTARTS AGE
logstash-0 1/1 Running 0 3h56m
logstash-1 1/1 Running 0 3h56m
logstash-2 1/1 Running 0 3h56m

查看一下数据卷的创建情况：

kubectl get pvc -l app=logstash
NAME STATUS VOLUME CAPACITY ACCESS MODES STORAGECLASS AGE
ldata-logstash-0 Bound pvc-c1833d35-d2ee-49a5-ae16-3a9d3227ebe5 1Gi RWO logstash-nfs-storage 3h56m
ldata-logstash-1 Bound pvc-9aa4b50c-45f7-4b64-9e4d-056838906675 1Gi RWO logstash-nfs-storage 3h56m
ldata-logstash-2 Bound pvc-95bcdbf0-e84d-4068-9967-3c69c731311b 1Gi RWO logstash-nfs-storage 3h56m

检查一下集群内部的 DNS 创建情况：

for i in 0 1; do kubectl exec logstash-$i -- sh -c 'hostname'; done
logstash-0
logstash-1
logstash-2

kubectl run -i --tty --image busybox:1.28.3 dns-test --restart=Never --rm /bin/sh
nslookup logstash-0.logstash
Server: 10.96.0.10
Address 1: 10.96.0.10 kube-dns.kube-system.svc.cluster.local

Name: logstash-0.logstash
Address 1: 10.244.7.54 logstash-0.logstash.default.svc.cluster.local

nslookup logstash-1.logstash
Server: 10.96.0.10
Address 1: 10.96.0.10 kube-dns.kube-system.svc.cluster.local

Name: logstash-1.logstash
Address 1: 10.244.5.150 logstash-1.logstash.default.svc.cluster.local


nslookup logstash-2.logstash
Server: 10.96.0.10
Address 1: 10.96.0.10 kube-dns.kube-system.svc.cluster.local

Name: logstash-2.logstash
Address 1: 10.244.34.177 logstash-2.logstash.default.svc.cluster.local

Logstash 的扩容/缩容

StatefulSet 的更新策略也是顺序的。

将之前设置的 StatefulSet 容量从 3 变成 5 。

kubectl scale sts web --replicas=5
kubectl get pods -l app=logstash
NAME READY STATUS RESTARTS AGE
logstash-0 1/1 Running 0 6h1m
logstash-1 1/1 Running 0 6h1m
logstash-2 1/1 Running 0 6h1m
logstash-3 1/1 Running 0 1h3m
logstash-4 1/1 Running 0 1h3m

新增的 Pod 在原有的基础上序号递增。

将 StatefulSet 容量从 5 变回 3 。

kubectl scale sts web --replicas=3

之前新增的 PVC 不会被删除，当下次达到该容量时会继续复用。不用担心有 Filebeat 会向被删除的 Logstash 发送数据， Filebeat 会自行寻找另一个运行正常的 Logstash。
由于设置了 queue.drain: true 所以撤除的 Logstash 在关闭前会将缓冲区内的数据发送完毕。

部署 Filebeat

方案一：将 Filebeat 与后端 SDK 封装在同一个 Pod 里采集日志文件（推荐）

将 Filebeat 容器与能够产生日志的后端 SDK 容器配置在同一个 Pod 里，后端 SDK 将日志写入 emptyDir 中，由 Filebeat 进行读取并发送至 Logstash。

优点：部署方便，哪个 Pod 里有日志就在哪个 Pod 里新增一个 Filebeat。
缺点：与 SDK Pod 有耦合，Filebeat 的数量可能较多，稍显冗余。

神策分析各后端语言的 SDK 都支持将数据写入文件，例如： Java SDK 的 ConcurrentLoggingConsumer，PHP SDK 的 FileConsumer，Python SDK 的 LoggingConsumer。

以 Java SDK 为例：

// 使用 ConcurrentLoggingConsumer 初始化 SensorsAnalytics
// 将数据输出到 /data/sa_log 下的 service_log 开头的文件中，每天一个文件
final SensorsAnalytics sa = new SensorsAnalytics(
        new SensorsAnalytics.ConcurrentLoggingConsumer("/data/sa_log/service_log"));

// 使用神策分析记录用户行为数据
sa.track(distinctId, true, "UserLogin");
sa.track(distinctId, true, "ViewProduct");

// 程序结束前，停止神策分析 SDK 所有服务
sa.shutdown();

以上配置将在 /data/sa_log 目录下生成数据文件，一天一个文件，文件列表如下：

service_log.20170923
service_log.20170924
service_log.20170925

在部署 Pod 时首先将 SDK 容器中的 /data/sa_log 目录下的内容挂载到 emptyDir: {} 上。然后设置 Filebeat 的读取的文件目录为：/var/log/containers/service_log.* 。Filebeat 将会读取该目录下所有以 service_log. 开头的文件。最后把 Filebeat 容器的 /var/log/containers/ 目录也挂载到 emptyDir: {} 上，运行时即可读取 SDK 容器产生的日志文件。

部署文件 pod.yaml 参考示例：

apiVersion: v1
kind: ConfigMap
metadata:
  name: filebeat-config-in
  labels:
    sa-app: filebeat
data:
  filebeat.yml: |-
    filebeat.inputs:
    - type: log
      # 读取 /var/log/containers 目录下以 service_log 开头的文件。
      paths:
        - /var/log/containers/service_log.*

    output.logstash:
      # 集群内网 Logstash
      hosts: ["logstash-0.logstash:5044","logstash-1.logstash:5044"]
---
apiVersion: apps/v1beta1
kind: Deployment
metadata:
  name: javasdk-beat
  labels:
    sa-app: javasdk-beat
spec:
  replicas: 3
  template:
    metadata:
      name: javasdk-beat
      labels:
        sa-app: javasdk-beat
    spec:
      containers:
      - name: javasdk
        image: javasdk:20190705
        command: ["/bin/bash", "-c"]
        args:
        - "bin/javasdk start"
        volumeMounts:
        - name: log-path
          # /data/sa_log 为后端 SDK 存放日志的目录，挂载到 emptyDir 上
          mountPath: /data/sa_log
      - name: filebeat
        image: docker.elastic.co/beats/filebeat:7.2.0
        args: [
          "-c", "/etc/filebeat.yml",
          "-e",
        ]
        volumeMounts:
        - name: config
          mountPath: /etc/filebeat.yml
          readOnly: true
          subPath: filebeat.yml
        - name: log-path
          # 文件读取目录也挂载到 emptyDir 上
          mountPath: /var/log/containers
          readOnly: true
      volumes:
      - name: log-path
        emptyDir: {}
      - name: config
        configMap:
          name: filebeat-config-in

方案二：Filebeat 部署在 K8s 节点上采集日志文件

Filebeat 以 DaemonSet 的方式部署在 K8s 节点上收集日志数据。节点上运行的后端 SDK 统一将日志存放在宿主机的指定目录内由 Filebeat 进行读取并发送至 Logstash。

优点：Filebeat 部署方便，与 SDK Pod 无耦合。
缺点：需要解决目录问题，在宿主机上会存在额外的日志文件。

考虑到在同一宿主机上可能存在多个相同的后端 SDK 容器，因此需要使每个容器在向宿主机目录写入日志的时候使用不同的目录。建议在启动容器时使用系统环境变量 HOSTNAME 作为路径名存放日志文件，然后将上一级目录挂载到宿主机目录上。

举个例子：

容器内日志的输出路径：/mount/${HOSTNAEM}-logs/service_log.20190708

宿主机存放日志的路径：/home/data/javasdk_logs/

将 /mount 挂载至 /home/data/javasdk_logs/ 下。

因此宿主机的 /home/data/javasdk_logs/ 目录下存放的内容大致如下：

[root@node-1 javasdk_logs]$ pwd
/home/data/javasdk_logs/
[root@node-1 javasdk_logs]$ ls -l
drwxr-xr-x 2 root root 22 Jul 8 12:06 javasdk-7d878c784d-5fpjz_logs
drwxr-xr-x 2 root root 22 Jul 6 18:33 javasdk-7d878c784d-7xmbb_logs
drwxr-xr-x 2 root root 22 Jul 6 18:52 javasdk-7d878c784d-vv9fz_logs
drwxr-xr-x 2 root root 22 Jul 8 12:08 javasdk-7d878c784d-w7q65_logs
drwxr-xr-x 2 root root 22 Jul 8 11:19 javasdk-7d878c784d-wkvxd_logs
[root@node-1 javasdk_logs]$ cd javasdk-7d878c784d-5fpjz_logs
[root@node-1 javasdk-7d878c784d-5fpjz_logs]$ ls -l
-rw-r--r-- 1 root root 6592991 Jul 8 23:59 service_log.20190706
-rw-r--r-- 1 root root 4777188 Jul 8 23:58 service_log.20190707
-rw-r--r-- 1 root root 137778 Jul 8 12:03 service_log.20190708

以 Java SDK 为例在保存日志时以 HOSTNAME 作为路径，参考如下：

// 获取 HOSTNAME
String hostname = System.getenv("HOSTNAME");
File logPath = new File("/mount/" + hostname + "-logs/");
if (!logPath.exists()) {
    logPath.mkdirs();
}
// 使用 ConcurrentLoggingConsumer 初始化 SensorsAnalytics
// 将数据输出到 /mount/${HOSTNAME}-logs/ 目录下以 service_log 为开头保存，运行容器时将 /mount 目录挂载到宿主机上
final SensorsAnalytics sa = new SensorsAnalytics(
        new SensorsAnalytics.ConcurrentLoggingConsumer(logPath.getAbsolutePath() + "/service_log"));

// 使用神策分析记录用户行为数据
sa.track(distinctId, true, "ViewProduct");

// 程序结束前，停止神策分析 SDK 所有服务
sa.shutdown();

参考的 javasdk.yaml 文件配置：

apiVersion: apps/v1beta1
kind: Deployment
metadata:
  name: javasdk
  labels:
    k8s-app: javasdk
spec:
  replicas: 3
  template:
    metadata:
      name: javasdk
      labels:
        k8s-app: javasdk
    spec:
      containers:
      - name: javasdk
        image: java-sdk-host:0715
        command: ["/bin/bash", "-c"]
        args:
        - "bin/javasdk start"
        volumeMounts:
        - name: logfile
          mountPath: /mount
      volumes:
      - name: logfile
        hostPath:
          path: /home/data/javasdk_logs/
          type: DirectoryOrCreate

如果您不希望更改原容器的日志路径存放方式，可以在容器启动时建立一条软链指向日志目录，将软链挂载在宿主机即可。

参考的 javasdk.yaml 文件配置：

apiVersion: apps/v1beta1
kind: Deployment
metadata:
  name: javasdk
  labels:
    k8s-app: javasdk
spec:
  replicas: 3
  template:
    metadata:
      name: javasdk
      labels:
        k8s-app: javasdk
    spec:
      containers:
      - name: javasdk
        image: java-sdk:0712
        command: ["/bin/bash", "-c"]
        args:
        - "rm -rf /your/logs/path
        && mkdir -p /mount/${HOSTNAME}_logs
        && ln -s /mount/${HOSTNAME}_logs /your/logs/path
        && bin/javasdk start"
        volumeMounts:
        - name: logfile
          mountPath: /mount
      volumes:
      - name: logfile
        hostPath:
          path: /home/data/javasdk_logs/
          type: DirectoryOrCreate

将 Filebeat 匹配的路径设置为 /home/data/javasdk_logs/*/service_log.*，并且把 Filebeat 存放进度的目录也挂载在宿主机上，这样在重启 DaemonSet 的时候节点上的 Filebeat 会继续之前的发送进度。

DaemonSet 配置文件 filebeat-ds.yaml 参考如下：

apiVersion: v1
kind: ConfigMap
metadata:
  name: filebeat-config
  labels:
    sa-app: filebeat
data:
  filebeat.yml: |-
    filebeat.inputs:
    - type: log
      paths:
        # 采集 service_log 开头的日志文件
        - /var/log/containers/*/service_log.*

    output.logstash:
      # 部署好的 Logstash
      hosts: ["logstash-0.logstash:5044","logstash-1.logstash:5044"]
---
apiVersion: extensions/v1beta1
kind: DaemonSet
metadata:
  name: filebeat
  labels:
    sa-app: filebeat
spec:
  template:
    metadata:
      labels:
        sa-app: filebeat
    spec:
      serviceAccountName: filebeat
      terminationGracePeriodSeconds: 30
      containers:
      - name: filebeat
        image: docker.elastic.co/beats/filebeat:7.2.0
        imagePullPolicy: IfNotPresent
        args: [
          "-c", "/etc/filebeat.yml",
          "-e",
        ]
        volumeMounts:
        - name: config
          mountPath: /etc/filebeat.yml
          readOnly: true
          subPath: filebeat.yml
        - name: inputs
          mountPath: /var/log/containers
          readOnly: true
        - name: data
          mountPath: /usr/share/filebeat/data
      volumes:
      - name: config
        configMap:
          name: filebeat-config
      - name: inputs
        hostPath:
          path: /home/data/javasdk_logs/
      - name: data
        hostPath:
          path: /home/data/filebeat_data/
          type: DirectoryOrCreate
---
apiVersion: rbac.authorization.k8s.io/v1beta1
kind: ClusterRoleBinding
metadata:
  name: filebeat
subjects:
- kind: ServiceAccount
  name: filebeat
  namespace: default
roleRef:
  kind: ClusterRole
  name: filebeat
  apiGroup: rbac.authorization.k8s.io
---
apiVersion: rbac.authorization.k8s.io/v1beta1
kind: ClusterRole
metadata:
  name: filebeat
  labels:
    sa-app: filebeat
rules:
- apiGroups: [""]
  resources:
  - namespaces
  - pods
  verbs:
  - get
  - watch
  - list
---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: filebeat
  namespace: default
  labels:
    sa-app: filebeat

Logstash 数据格式说明

神策插件是根据 logstash 提供的标准数据格式来解析数据，完成数据上报的；在配置来源和目标插件的时候，需要保证数据格式是 logstash 标准的格式，具体格式信息如下：

{
	"host" => "localhost",
	"@version" => 1,
	"@timestamp" => 2023-01-01T00:00:00,
	"message" => 具体需要上报的 json 数据
}

常见问题：

:exceptionMessage=>"no implicit conversion of nil into String"

这个问题一般就是由于上报的数据没有放在 message 中导致插件解析 json 报错，需要检查一下配置文件，比如提前将信息 json 化，这样就会导致 message 为空。无需提前解析，神策插件会完成数据解析上报

Logstash + Filebeat

概述

Logstash 使用说明

Logstash 下载与安装

安装 logstash-output-sensors_analytics 插件

Logstash 配置

Logstash Pipeline 配置

Logstash 输入输出配置

Logstash 运行配置

Logstash 启动

Logstash 进度

sensors-output-plugin 升级与回滚

Filebeat 使用说明

Filebeat 下载与安装

Filebeat 配置

启动 Filebeat

Filebeat 进度

服务器场景下的数据采集

部署 Logstash

部署 Filebeat

使用 Docker 容器化场景下的数据采集

部署 Logstash

部署 Filebeat

方案一：在 SDK 容器中安装 Filebeat 采集日志并发送至 Logstash（推荐）

方案二：SDK 使用共享数据保存日志 Filebeat 进行读取

使用 K8s（Kubernetes）自动编排容器场景中的数据采集

Logstash 部署

部署 Filebeat

方案一：将 Filebeat 与 后端 SDK 封装在同一个 Pod 里采集日志文件（推荐）

方案二：Filebeat 部署在 K8s 节点上采集日志文件

Logstash 数据格式说明

方案一：将 Filebeat 与后端 SDK 封装在同一个 Pod 里采集日志文件（推荐）