在Logstash中格式化块数据并使用Grok模式按起止时间过滤的咨询

第一步：流式数据的多行聚合预处理 #

你的日志以固定分隔线拆分独立任务块，首先需要将分散的行聚合为单条完整事件，否则Grok无法匹配跨行字段。
以常用的Logstash流式处理为例，多行配置参考：

input {
  # 此处替换为你的实际流式输入源，比如kafka、实时文件等
  file {
    path => "/path/to/your/logfile.log"
    codec => multiline {
      pattern => "^\+{46}$"
      negate => true
      what => "previous"
      auto_flush_interval => 5
    }
  }
}

规则会自动把两个分隔线中间的所有行拼接为一个事件，超过5秒没收到下一个分隔线也会自动输出当前块，适配流式场景的延迟要求。

第二步：Grok匹配提取规则 #

针对你的日志结构，支持同时覆盖成功/失败场景的Grok模式如下，可直接提取所有需要的字段：

\+{46}\nName: %{DATA:task_name}\n~{46}\nBegin Date: %{DATA:begin_date}\nBegin Time: %{DATA:begin_time}\n\nActivity\n~{46}\n(?<activity_log>(?:.|\n)*?)\n~{46}\n\nResult\n~{46}\n%{DATA:result_msg} %{WORD:exec_status}(\n(?<error_info>(?:.|\n)*?))?\nEnd Date:\s*%{DATA:end_date}\nEnd Time:\s*%{DATA:end_time}\nExecuted by: %{DATA:executor}\n\+{46}

提取字段说明：

• 基础字段：task_name任务名、begin_date开始日期、begin_time开始时间、activity_log活动区全量日志、executor执行人
• 核心状态字段：exec_status执行状态，返回Completed/Failed、end_date结束日期、end_time结束时间
• 异常字段：error_info仅执行失败时返回，包含所有报错信息和堆栈内容

第三步：按时间过滤实现 #

把提取到的begin_date和begin_time拼接为完整时间戳，转换成标准时间格式后即可直接按范围过滤。
实时流式过滤可直接在处理阶段加判断，示例：

filter {
  grok {
    match => { "message" => "上文的Grok匹配规则" }
  }
  # 拼接开始时间为标准格式
  mutate {
    add_field => { "begin_timestamp" => "%{begin_date} %{begin_time}" }
  }
  date {
    match => [ "begin_timestamp", "MM/dd/yyyy HH:mm:ss" ]
    target => "begin_timestamp"
  }
  # 示例过滤：仅保留2024年1月1日之后执行的任务，可按需修改时间范围
  if [begin_timestamp] < "2024-01-01T00:00:00.000Z" {
    drop {}
  }
}

如果使用Fluentd、Elasticsearch摄入管道等其他组件，逻辑完全一致，先做多行聚合再套用上述Grok规则即可。

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