如何用正则表达式在Logstash中从message字段生成新字段？

我来帮你搞定这个Logstash字段提取的需求！针对你给出的message格式，我们可以用Logstash的grok过滤器（或者更灵活的Ruby正则方式）来精准提取各个关键信息生成新字段。下面是具体的实现方案：

Grok内置了大量预定义的正则模式，上手快且易维护，非常适合处理这种结构化程度较高的日志。在你的Logstash配置的filter区块中添加以下内容：

filter {
  grok {
    match => { "message" => [
      # 匹配日志日期、时间、级别和异常触发方法
      "^%{DATE_EU:log_date} %{TIME:log_time} \[%{LOGLEVEL:log_level}\] at %{DATA:exception_origin}:",
      # 匹配上层异常提示消息
      "\s*Message: %{DATA:exception_message}",
      # 提取关联用户
      "\s*User: %{DATA:user}",
      # 提取主机名
      "\s*Host: %{DATA:host}",
      # 匹配异常类型和具体描述
      "\s*%{DATA:exception_type}: %{DATA:exception_details}",
      # 提取出错文件路径和行号
      "\s*at .* in %{PATH:error_file}:line %{NUMBER:error_line:int}"
    ] }
    # 允许匹配多个模式，不匹配的部分忽略（避免报错）
    break_on_match => false
    # 添加tag标记匹配成功的日志，方便后续筛选
    add_tag => ["exception_log_parsed"]
  }

  # 可选操作：把提取的日期时间合并成标准的@timestamp字段
  date {
    match => ["log_date %{log_time}", "dd.MM.yyyy HH:mm:ss"]
    target => "@timestamp"
  }
}

提取的字段说明： #

• log_date：日志日期（格式为dd.MM.yyyy）
• log_time：日志时间（格式为HH:mm:ss）
• log_level：日志级别（这里是ERROR）
• exception_origin：异常触发的方法（比如MyApp.Controllers.Controller.OnException）
• exception_message：上层异常提示（比如Controller exception!）
• user：关联的操作用户（比如TestUser）
• host：日志产生的主机名（比如MyLaptop）
• exception_type：异常类型（比如System.Exception）
• exception_details：异常的具体描述（比如Testing）
• error_file：出错的文件完整路径
• error_line：出错的行号（自动转为整数类型）

如果更倾向于用原生正则自定义匹配逻辑，可以用Ruby过滤器手动捕获字段，适合处理格式多变的日志：

filter {
  ruby {
    code => '
      message = event.get("message")
      # 匹配日期和时间
      date_time_match = message.match(/^(\d{2}\.\d{2}\.\d{4}) (\d{2}:\d{2}:\d{2})/)
      if date_time_match
        event.set("log_date", date_time_match[1])
        event.set("log_time", date_time_match[2])
      end
      # 匹配日志级别
      level_match = message.match(/\[(\w+)\]/)
      event.set("log_level", level_match[1]) if level_match
      # 匹配异常触发方法
      origin_match = message.match(/at (\S+):/)
      event.set("exception_origin", origin_match[1]) if origin_match
      # 提取用户信息
      user_match = message.match(/User: (\S+)/)
      event.set("user", user_match[1]) if user_match
      # 提取主机名
      host_match = message.match(/Host: (\S+)/)
      event.set("host", host_match[1]) if host_match
      # 匹配异常类型和描述
      exception_match = message.match(/(\w+\.\w+): (.+)/)
      if exception_match
        event.set("exception_type", exception_match[1])
        event.set("exception_details", exception_match[2])
      end
      # 提取出错文件和行号
      file_line_match = message.match(/in (.+):line (\d+)/)
      if file_line_match
        event.set("error_file", file_line_match[1])
        event.set("error_line", file_line_match[2].to_i)
      end
    '
  }

  # 可选：合并日期时间到标准@timestamp字段
  date {
    match => ["log_date %{log_time}", "dd.MM.yyyy HH:mm:ss"]
    target => "@timestamp"
  }
}

小提示： #

• 可以用logstash -f your_config.conf --config.test_and_exit命令验证配置语法是否正确
• 调试时可以加--config.reload.automatic参数，修改配置后自动重载，不用重启Logstash
• 如果日志格式有细微变动，只需对应调整Grok模式或正则表达式即可

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