日志处理工具--向量练习

vector是什么？

以下描述翻译自vector官网： Vector 是一种高性能的可观察性数据管道 可以收集、转换所有日志、指标和跟踪信息（ logs, metrics, and traces），并将其写到想要的存储当中； Vector 可以实现显着的成本降低、丰富的数据处理和数据安全； 开源，比所有替代方案快 10 倍。

类似logstash，但比logstash强悍太多的性能、简单明了的配置文件、强大的数据处理函数、智能均衡kafka分区消费等；在这些特性中，性能直接关乎成本

vector官网 vector.dev

GitHub地址 github.com/vectordotde…

基准性能测试

Test	Vector	Filebeat	FluentBit	FluentD	Logstash	SplunkUF	SplunkHF
TCP to Blackhole	86mib/s	n/a	64.4mib/s	27.7mib/s	40.6mib/s	n/a	n/a
File to TCP	76.7mib/s	7.8mib/s	35mib/s	26.1mib/s	3.1mib/s	40.1mib/s	39mib/s
Regex Parsing	13.2mib/s	n/a	20.5mib/s	2.6mib/s	4.6mib/s	n/a	7.8mib/s
TCP to HTTP	26.7mib/s	n/a	19.6mib/s	<1mib/s	2.7mib/s	n/a	n/a
TCP to TCP	69.9mib/s	5mib/s	67.1mib/s	3.9mib/s	10mib/s	70.4mib/s	7.6mib/s

Vector 的主要优势：

• 高效的内存/CPU消耗和高数据吞吐量。
• 良好的可靠性，正确性和交付方面有保证。
• 包括自定义 DSL，以一种安全、高性能的方式动态转换数据。
• 支持基于指标和日志的有效负载。
• 大量的输入和输出集成。
• 可以部署为代理或聚合器。

支持广泛的部署选项、同时支持指标和日志以及可用集成的数量。 Vector 用 Rust 编写，提供内存安全和效率保证。Vector 引入了单元测试框架，可以更轻松地维护复杂的日志收集工具拓扑。

概念

Sources

来源 input

Transforms

变换 filter

Sinks

接收器 output

自适应请求并发 (ARC)

自适应并发，会根据请求的往返时间 (RTT) 和 HTTP 响应代码（失败与成功），从而决策出一个最佳的速率！ 最终结果是提高整个可观察性基础架构的性能和可靠性。 请求的往返时间 (RTT) 和 HTTP 响应码（失败与成功）。

• 如果 RTT 下降/恒定和/或响应代码为 (200-299)， ???? 并线性增加吞吐量。 AIMD “和性增长”。
• 如果 RTT 正在增加和/或响应代码始终指示失败 - 代码429 Too Many Requests和503 Service Unavailable- ???? 并以指数方式降低并发性。AIMD “乘性降低”。

和性增长/乘性降低（additive-increase/multiplicative-decrease、AIMD）算法是一个反馈控制算法，最多的用途是在TCP拥塞控制。AIMD将拥塞窗口的线性增长与监测到拥塞时的指数降低相结合。使用AIMD拥塞控制的多个流将最终收敛到使用等量的共享链路。

vector的自动均衡kafka消费

部署角色

守护进程

守护程序角色旨在收集单个主机上的所有数据。

Sidecar

Sidecar 角色将 Vector 与每个服务结合起来，只专注于该单个服务的数据收集

聚合器

聚合器用于处理从多个上游源收集数据并执行跨主机聚合和分析。 Vector 既可以用作代理，也可以用作聚合器。

拓扑

安装部署

vector.dev/docs/setup/… 官方提供了安装包、docker等多种安装方式,这里使用二进制的方式安装

### Linux (x86_64)
# 下载并解压存档：

# Latest (version 0.24.2)
mkdir -p vector && \
  curl -sSfL --proto '=https' --tlsv1.2 https://packages.timber.io/vector/0.24.2/vector-0.24.2-x86_64-unknown-linux-musl.tar.gz  | \
  tar xzf - -C vector --strip-components=2

# 切换到 vector 目录：
cd vector

# 添加到 $PATH 中
echo "export PATH=\"$(pwd)/vector/bin:\$PATH\"" >> $HOME/.profile
source $HOME/.profile

# Vector 配置文件位于：
config/vector.toml
> 示例配置位于config/vector/examples/*


# 将 Vector 安装到 Systemd 中
cat /usr/lib/systemd/system/vector.service

[Unit]
Description="Vector - An observability pipelines tool"
Documentation=https://vector.dev/
Wants=network-online.target
After=network-online.target

[Service]
LimitNOFILE=1000000
#LimitCORE=infinity
LimitSTACK=10485760
User=root
ExecStart=/data/ops/vector/bin/vector -t 128 --config-dir=/data/ops/vector/config/
Restart=always
AmbientCapabilities=CAP_NET_BIND_SERVICE

[Install]
WantedBy=multi-user.target

# 启动
systemctl status vector.service 
systemctl enable vector.service
systemctl start vector.service

vector配置文件

来源（sources）

即vector的数据来源，支持文件、kafka、http、各类metrics等等数据源 各类数据源均可在文档中找到配置方式 vector.dev/docs/refere… 使用kafka数据来源

[sources.kafka-nginx-error]     # "数据源"名称
  type = "kafka"      # 类型
  bootstrap_servers = "10.xxx.xxx.xxx:9092,10.xxx.xxx.xxx:9092,10.xxx.xxx.xxx:9092"       # kafka链接地址
  group_id = "consumer-group-name"        # 消费组id
  topics = [ "^(topic1|topic2)-.+" ]    # topic，支持正则

变换[可选]（transforms）

如果原始日志不需要处理，可以忽略 vector.dev/docs/refere…

remap

remap在vector中使用VRL 用于解析和转换 Vector 中的数据。Vector Remap Language，一种面向表达式的语言，旨在以安全和高性能的方式处理可观察性数据（日志和指标） vector处理nginx错误日志的配置：

[transforms.remap-nginx-error]      # "变换"名称
  type = "remap"                    # 类型
  inputs = ["kafka-nginx-error"]    # 输入，这里的输入自然是上一层的"来源"
source = '''        # 正式开始处理
  . = parse_json!(.message)     # 将每一条错误日志解析成json，message的值就是从kafka中读取到的原始值
  del(.@metadata)               # 删除自动携带的一些信息
  .parse = parse_nginx_log!(.message, "error")      # 解析nginx错误日志
'''

VRL 功能参考

vector.dev/docs/refere… VRL 程序作用于单个可观察性事件，可用于：

• 转换可观察性事件
• 指定路由和过滤事件的条件

parse_nginx_log

解析 Nginx 访问和错误日志行。行可以是combined, 或error格式。

# combined
parse_nginx_log!(
    s'172.17.0.1 - alice [01/Apr/2021:12:02:31 +0000] "POST /not-found HTTP/1.1" 404 153 "http://localhost/somewhere" "Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/72.0.3626.119 Safari/537.36" "2.75"',
    "combined",
)

{
    "agent": "Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/72.0.3626.119 Safari/537.36",
    "client": "172.17.0.1",
    "compression": "2.75",
    "method": "POST",
    "path": "/not-found",
    "protocol": "HTTP/1.1",
    "referer": "http://localhost/somewhere",
    "request": "POST /not-found HTTP/1.1",
    "size": 153,
    "status": 404,
    "timestamp": "2021-04-01T12:02:31Z",
    "user": "alice"
}

# error
parse_nginx_log!(
    s'2021/04/01 13:02:31 [error] 31#31: *1 open() "/usr/share/nginx/html/not-found" failed (2: No such file or directory), client: 172.17.0.1, server: localhost, request: "POST /not-found HTTP/1.1", host: "localhost:8081"',
    "error"
)

{
    "cid": 1,
    "client": "172.17.0.1",
    "host": "localhost:8081",
    "message": "open() \"/usr/share/nginx/html/not-found\" failed (2: No such file or directory)",
    "pid": 31,
    "request": "POST /not-found HTTP/1.1",
    "server": "localhost",
    "severity": "error",
    "tid": 31,
    "timestamp": "2021-04-01T13:02:31Z"
}

parse_grok

value使用Grok格式解析. 支持此处列出的所有模式 grokdebug.herokuapp.com/

parse_grok!(
  "2020-10-02T23:22:12.223222Z info Hello world",
  "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} %{GREEDYDATA:message}"
)

{
    "level": "info",
    "message": "Hello world",
    "timestamp": "2020-10-02T23:22:12.223222Z"
}

filter

筛选，根据一组条件过滤事件

[transforms.my_transform_id]
  type = "filter"
  inputs = [ "my-source-or-transform-id" ]
  condition = '.level != "debug"'

接收器 (sinks)

支持console、elasticsearch、kafka、vector、http 等 vector.dev/docs/refere…

elasticsearch

[sinks.my_sink_id]             # 接收器名称
  type = "elasticsearch"       # 类型
  inputs = [ "my-source-or-transform-id" ]   # 输入，这里的输入是上一层的"变换"名称
  endpoint = "http://10.xxx.xxx.xxx:9200"    # 输出的链接地址
  bulk.index = "{{ project_name }}-%Y-%m-%d" # 索引名称，可以使用日志当中的字段作为变量、以及日期来作为索引名称

clickhouse

[sinks.my_sink_id]             # 接收器名称
  type = "clickhouse"          # 类型
  inputs = [ "my-source-or-transform-id" ]   # 输入，这里的输入是上一层的"变换"名称
  endpoint = "http://localhost:8123"         # 输出的链接地址
  database = "mydatabase"
  table = "mytable"
  auth.strategy = "basic"
  auth.user = "ck_admin"
  auth.password = "password"
  compression = "gzip"

全局配置参数

data_dir

保存 Vector 状态的目录，例如磁盘缓冲区、文件检查点等。

"/var/local/lib/vector/"

扩充表 (enrichment_tables )

• CSV files
• MaxMind databases

加密密码 (secret)

实践

将结果输出到console

调试时使用

[sinks.my_sink_id]            # 接收器名称
  type = "console"            # 类型
  inputs = [ "my-source-or-transform-id" ]       # 输入，这里的输入是上一层的"变换"名称
  encoding.codec = "json"     # 可选json 或者 text

多配置文件启动

vector -c /etc/vector/*.toml -w /etc/vector/*.toml

-c, --config <配置> 
            从一个或多个文件中读取配置。支持通配符路径
-C, --config-dir <配置目录>
-t, --threads <线程>
            用于处理的线程数（默认为可用内核数）
-w, --watch 配置
            监视配置文件的变化

多个配置文件，在同一个vector实例中各阶段的命名也不能重名

多topic使用正则匹配

统一过日志格式，可以在“来源”中指定消费同一类topic

topics = [ "^(prefix1|prefix2)-.+" ]

索引使用日志中的字段值作为索引名称

可以使用日志中的字段值作为变量名称，还可以使用%Y、%m、%d分别表示年、月、日，这是一种很好的日志索引管理方式

bulk.index = "{{ project_name }}-{{ env }}-%Y-%m-%d"

查看vector各任务的处理情况

以命令行的方式实时查看各任务的处理情况

[api]
  enabled = true
  address = "127.0.0.1:8686"

执行以下命令即可： vector top

vector更加详细的metrics指标

使用prometheus_exporter格式的输出

[sources.vector_metrics]
  type = "internal_metrics"
  namespace = "vector"
  scrape_interval_secs = 30
 
[sinks.prometheus]
  type = "prometheus_exporter"
  inputs = [ "vector_metrics" ]
  address = "0.0.0.0:9598"
  default_namespace = "service"

Vector 命令行

# 生成包含组件列表的 Vector 配置
vector generate [OPTIONS] <EXPRESSION>

# 生成可视化的拓扑
vector graph [OPTIONS]

vector graph --config /etc/vector/vector.toml | dot -Tsvg > graph.svg

# 列出可用组件
vector list [FLAGS] [OPTIONS]

# 观察流入组件（转换、接收器）和流出组件（源、转换）的事件。以指定的时间间隔对事件进行采样。
vector tap [FLAGS] [OPTIONS] [ARGUMENTS]

# 配置单元测试
vector test [OPTIONS] [ARGUMENTS]

# 在控制台显示本地或远程 Vector 实例的拓扑和指标
vector top [FLAGS] [OPTIONS]

# 验证目标配置
vector validate [FLAGS] [OPTIONS] [ARGUMENTS]

# vrl CLI
vector vrl [FLAGS] [OPTIONS] [ARGUMENTS]