开源日志工具：vector，配置、转发及测试

开源日志工具vector配置说明，详细请访问官方网站：vector。

正则在线验证地址：Rustexp

实现功能：

message原文样例

Mar 19 01:14:46 10.10.4.4 Manager: {"Desc":"{"value":"|2001060009|内存上升幅度过大|critical|b87461b9e35148e1be1e51fin14|error|10.10.4.4|localhost-test-01|1|metric|2024-03-19 01:14:42 UTC|2024-03-19 01:14:42 UTC||监控|内存上升幅度为233.33%,阈值为200.0%。|","key":"2001060009"}"}

一、通过采集本地log文件，对文件内容进行切分、过滤后，将日志内容展示在终端/写入数据库。

1、配置文件内容：

# 配置vector路径
data_dir = "/usr/local/vector/"
acknowledgements.enabled = true

# 配置源方式，此处使用读取配置文件
[sources.source_log]
type = "file"
include = ["/var/log/messages"]
read_from = "end"

# 对“source_log”读取的内容做正则匹配，并对切分的字段进行重新赋值
[transforms.source_trans]
type = "remap"
inputs = ["source_log"]
source = '''
. |=parse_regex!(.message,r'^(?P<log_timestamp>\w+\s+\d+ \d+:\d+:\d+) (?P<source>\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}) (?P<message>.*)$')
#.message = parse_regex!(.message,r'^Manager:\{(?P<message>.*)\} \w+ .*$')
.Desc = split(.message,"{")[1]
.Desc,err = split(.Desc,"}")[0]
.id,err = split(.Desc,"|")[1]
.name,err = split(.Desc,"|")[2]
.serverity,err = split(.Desc,"|")[3]
.instanceid,err = split(.Desc,"|")[4]
.instancename,err = split(.Desc,"|")[5]
.hostip,err = split(.Desc,"|")[6]
.hostname,err = split(.Desc,"|")[7]
.count,err = split(.Desc,"|")[8]
.type,err = split(.Desc,"|")[9]
.firsttimeutc,err = split(.Desc,"|")[10]
.lasttimeutc,err = split(.Desc,"|")[11]
.cleartimeutc,err = split(.Desc,"|")[12]
.componentname,err = split(.Desc,"|")[13]
.description,err = split(.Desc,"|")[14]
del(.source_type)
del(.message)
del(.Desc)
del(.file)
del(.log_timestamp)
#del(.timestamp)
'''

# 对“source_trans”匹配后的内容进行过滤
[transforms.source_filter]
type = "filter"
inputs = ["source_trans"]
#condition = { type = "datadog_search", source = "*.info"}
condition = '.serverity== "critical"'

# 对“source_trans”过滤后的内容配置展示方式，此处配置为终端展示
[sinks.print]
type = "console"
inputs = ["source_filter"]
encoding.codec = "json"

2、简要说明

    通过配置文件自上而下执行，先读取文件 ——> 通过正则表达式截取字段并重新赋值 ——> 过滤告警告警级别为“critical”的事件保留，其他级别丢弃 ——> 在终端打印结果

二、通过监听本地514日志端口，对接收的日志原文进行切分、过滤后，将日志内容展示在终端/写入数据库。

1、配置文件内容：

# 配置vector路径
data_dir = "/usr/local/vector/"
acknowledgements.enabled = true

# 配置源方式，此处使用读取配置文件
[sources.source_socket]
type = "socket"
address = "0.0.0.0:514"
mode = "tcp"

# 对“source_log”读取的内容做正则匹配，并对切分的字段进行重新赋值
[transforms.source_trans]
type = "remap"
inputs = ["source_socket"]
source = '''
. |=parse_regex!(.message,r'^(?P<log_timestamp>\w+\s+\d+ \d+:\d+:\d+) (?P<source>\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}) (?P<message>.*)$')
#.message = parse_regex!(.message,r'^Manager:\{(?P<message>.*)\} \w+ .*$')
.Desc = split(.message,"{")[1]
.Desc,err = split(.Desc,"}")[0]
.id,err = split(.Desc,"|")[1]
.name,err = split(.Desc,"|")[2]
.serverity,err = split(.Desc,"|")[3]
.instanceid,err = split(.Desc,"|")[4]
.instancename,err = split(.Desc,"|")[5]
.hostip,err = split(.Desc,"|")[6]
.hostname,err = split(.Desc,"|")[7]
.count,err = split(.Desc,"|")[8]
.type,err = split(.Desc,"|")[9]
.firsttimeutc,err = split(.Desc,"|")[10]
.lasttimeutc,err = split(.Desc,"|")[11]
.cleartimeutc,err = split(.Desc,"|")[12]
.componentname,err = split(.Desc,"|")[13]
.description,err = split(.Desc,"|")[14]
del(.source_type)
del(.message)
del(.Desc)
del(.file)
del(.log_timestamp)
#del(.timestamp)
'''

# 对“source_trans”匹配后的内容进行过滤
[transforms.source_filter]
type = "filter"
inputs = ["source_trans"]
#condition = { type = "datadog_search", source = "*.info"}
condition = '.serverity== "critical"'

# 对“source_trans”过滤后的内容配置展示方式，此处配置为向ck数据库写入
[sinks.Out_source_filter]
type = "clickhouse"
inputs = ["source_filter"]
endpoint = "http://10.10.4.4:8123"
auth.strategy = "basic"
auth.user = "admin"
auth.password = "admin"
database = "source_log"
table = "source_log"
encoding.timestamp_format="unix"

2、简要说明

    通过配置文件自上而下执行，监听socket ——> 通过正则表达式截取字段并重新赋值 ——> 过滤告警告警级别为“critical”的事件保留，其他级别丢弃 ——> 将过滤的日志存储到ck中

三、其他说明

1、log接收和socket接收的内容会出现部分字段差异，需要打印到终端重新做规则。

2、一个配置文件中可以同时出现多个sources、transforms等模块，例如同时打印到终端和写到ck数据库，但是相同模块后跟的名称不能相同。

3、注意上下文“input”的都是上一个模块名字。