大佬log pipeline报丢日志，帮忙看下

[zillaer]

frontend报错信息：

pipline配置(custom_index_20260316)：

transform:
  - fields:
      - appname
      - domain
      - hostname
      - ip
      - source
      - service
      - context_id
    type: string
  - fields:
    - raw_message, message
    type: string
    index: fulltext
  - fields:
      - real_log_time
    type: epoch, ms
    index: timestamp

frontend是三副本，未设资源限制，服务器资源远未达瓶颈

日志每分钟10w条左右

两个vector实例写入日志，配置如下：

# ---------------------------------------------------------
# 1. Source: 从 Kafka 消费日志
# ---------------------------------------------------------
[sources.kafka_logs]
type = "kafka"
bootstrap_servers = "192.1.155.122:9092,192.1.155.117:9092,192.1.155.23:9092" # 替换为你的 Kafka 地址
topics = ["raw_message"]                                        # 替换为你的 Topic 名称
group_id = "tra-consumer"                        # 消费者组 ID
auto_offset_reset = "latest"                                 # earliest 或 latest
session_timeout_ms = 10000


# ---------------------------------------------------------
# 2. Transform: 数据清洗与格式化 (可选但推荐)
# ---------------------------------------------------------
# 假设 Kafka 中的消息是 JSON 格式，我们需要解析它并映射到 GreptimeDB 的列
[transforms.parse_json]
type = "remap"
inputs = ["kafka_logs"]
source = '''
  # 1. 安全解析 JSON，避免脏数据导致整个批次处理失败
  . = parse_json!(.message)
  m, err = parse_regex(.raw_message, r'^(?P<rlog_ts>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{3})')
  if err == null {
      ts, ts_err = parse_timestamp(m.rlog_ts, format: "%Y-%m-%d %H:%M:%S.%3f")
      if ts_err == null {
        .real_log_time = to_unix_timestamp(ts, unit: "milliseconds")
      }
  }
  .service = .appname
  .context_id = .hostname
  del(.log_id)
  del(.tag)
  del(.token)
'''

# ---------------------------------------------------------
# 3. Sink: 写入 GreptimeDB
# ---------------------------------------------------------
[sinks.greptime_output]
type = "greptimedb_logs"
inputs = ["parse_json"]

# GreptimeDB 的 HTTP 端点 (默认端口 4000)
endpoint = "http://192.1.213.34:32000/gtdb"

# 目标数据库名称 (默认为 public)
dbname = "public"

# 目标表名 (Vector 会自动创建表，如果不存在)
table = "kafka_logs_table"

pipeline_name = "custom_index_20260316"

# 写入模式：upsert (默认) 或 insert
#mode = "upsert"

# 批量写入配置 (优化性能)
batch.max_events = 1000
batch.timeout_secs = 10

greptimed版本：v1.0.0-rc.1

[shuiyisong]

你好，

1. 可以尝试在 vector 上加一个 console sink 然后输出几条 sample 数据吗，我们可以尝试在本地复现一下问题
2. 是所有的数据都无法通过这个配置写入吗，还是部分可以，但是部分会报错失败
3. 服务端的日志（frontend 或者 datanode）有其他更详细的报错内容吗

谢谢

[zillaer]

感谢提供排查思路，已排查出是日志打印不规范缺时间字段导致
vrl修改如下：

source = '''
  # 1. 安全解析 JSON，避免脏数据导致整个批次处理失败
  . = parse_json!(.message)
  m, err = parse_regex(.raw_message, r'^(?P<rlog_ts>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{3})')
  if err == null {
      ts, ts_err = parse_timestamp(m.rlog_ts, format: "%Y-%m-%d %H:%M:%S.%3f")
      if ts_err == null {
        .real_log_time = to_unix_timestamp(ts, unit: "milliseconds")
      }
  } else {
        .real_log_time = to_unix_timestamp(now(), unit: "milliseconds")
  }
  .service = .appname
  .context_id = .hostname
  del(.log_id)
  del(.tag)
  del(.token)
'''