Telegraf如何收集十六进制数据并转换后写入InfluxDB？

问题背景 #

需通过Telegraf的tail插件解析特定格式日志，当前配置已完成基础解析，但十六进制字段（如01hexa）以字符串形式存储，需转换为整数后执行公式01hexa_convert = ((IN_01hexa)*0.8125-4.25)计算浮点结果并写入InfluxDB。日志格式如下：

2022-09-15 00:04:39 AAA99 BBB99 24 43 B2 50
2022-09-15 00:04:40 BBB99 BBB99 2C 42 F1 4B

解决步骤 #

1. 调整Grok模式，自动转换十六进制为整数 #

在inputs.tail的grok_custom_patterns中，为十六进制字段添加:int类型转换标识，让Grok直接将十六进制字符串解析为十进制整数：

CUSTOM_LOG %{TIMESTAMP_ISO8601:timestamp:ts-"2006-01-02 15:04:05"} %{WORD:AAAinfo:tag} %{WORD:BBBinfo:tag} %{BASE16NUM:01hexa:int} %{BASE16NUM:02hexa:int} %{BASE16NUM:03hexa:int} %{BASE16NUM:04hexa:int}

2. 新增Starlark处理器执行数值计算 #

使用Telegraf的processors.starlark实现自定义计算逻辑，在配置中添加以下片段：

[[processors.starlark]]
  namepass = ["hexa_log"]
  source = '''
def apply(metric):
    in_01hexa = metric.fields.get("01hexa")
    if in_01hexa is not None:
        metric.fields["01hexa_convert"] = (in_01hexa * 0.8125) - 4.25
    return metric
'''

• namepass = ["hexa_log"]限定仅处理目标指标，避免干扰其他数据；
• 脚本读取01hexa整数值，按公式计算后将结果作为新字段01hexa_convert添加到指标中。

3. 完整配置示例 #

整合后的Telegraf配置如下：

[global_tags]
[agent]
  interval = "10s"
  round_interval = true
  metric_batch_size = 1000
  metric_buffer_limit = 10000
  collection_jitter = "0s"
  flush_interval = "10s"
  flush_jitter = "0s"
  precision = ""
  debug = true
  quiet = false
  logtarget = "file"
  logfile = "/var/log/telegraf.log"
  hostname = ""
  omit_hostname = true

[[outputs.influxdb]]
  urls = ["http://127.0.0.1:8086"]
  database = "data_hexa"
  skip_database_creation = false
  timeout = "5s"
  username = "telegraf"
  password = "***********"
  user_agent = "telegraf"

[[outputs.file]]
  files = ["stdout", "/tmp/metrics.out"]

[[inputs.tail]]
  files = ["/mnt/cle/hexa*.log"]
  name_override = "hexa_log"
  from_beginning = true
  max_undelivered_lines = 10000
  character_encoding = "utf-8"
  grok_patterns = ["%{CUSTOM_LOG}"]
  grok_custom_patterns = '''
CUSTOM_LOG %{TIMESTAMP_ISO8601:timestamp:ts-"2006-01-02 15:04:05"} %{WORD:AAAinfo:tag} %{WORD:BBBinfo:tag} %{BASE16NUM:01hexa:int} %{BASE16NUM:02hexa:int} %{BASE16NUM:03hexa:int} %{BASE16NUM:04hexa:int}
'''
  data_format = "grok"
  grok_timezone = "Local"

[[processors.starlark]]
  namepass = ["hexa_log"]
  source = '''
def apply(metric):
    in_01hexa = metric.fields.get("01hexa")
    if in_01hexa is not None:
        metric.fields["01hexa_convert"] = (in_01hexa * 0.8125) - 4.25
    return metric
'''

4. 验证效果 #

重启Telegraf服务后，可通过/tmp/metrics.out或InfluxDB查询验证结果：

• 第一条日志中24（十六进制=36十进制）计算后为36*0.8125 -4.25 =25.0；
• 新增字段01hexa_convert以浮点类型存储，符合需求。

内容的提问来源于stack exchange，提问作者oryble