如何在Logstash中统计特定应用来源每分钟的事件字节总大小

开发者阿泽

2026-10-1

Logstash按应用维度统计每分钟事件总字节量实现方案

实现步骤

1. 计算单事件字节大小

在filter阶段通过ruby插件计算每个事件的字节长度，有两种可选方案：

方案一：直接统计事件序列化后的完整字节数，适配没有保留原始上报报文的场景

filter {
  ruby {
    code => "event.set('event_byte_size', event.to_json.bytesize)"
  }
}

方案二：统计原始上报报文的字节数，统计结果更精准，要求input阶段保留了完整原始报文字段（示例中原始报文字段为raw_message）

filter {
  ruby {
    code => "event.set('event_byte_size', event.get('raw_message').to_s.bytesize)"
  }
}

2. 按应用+时间维度聚合统计

首先要确保所有事件都携带唯一标识应用的字段（示例中为app_name），两种聚合方式可选：

方式一：使用aggregate过滤器实现精准的分钟级聚合

filter {
  aggregate {
    task_id => "%{app_name}_%{+YYYYMMddHHmm}"
    code => "
      map['total_byte_size'] ||= 0
      map['total_byte_size'] += event.get('event_byte_size')
      map['app_name'] = event.get('app_name')
      map['stat_minute'] = event.get('@timestamp').time.utc.strftime('%Y-%m-%d %H:%M:00')
      event.cancel()
    "
    push_previous_map_as_event => true
    timeout => 120
  }
}

该配置会每分钟生成一个聚合事件，包含app_name、stat_minute、total_byte_size三个核心统计字段，直接输出到对应存储即可。

方式二：使用metrics过滤器实现轻量统计，适配高吞吐量场景

filter {
  metrics {
    meter => "bytes.%{app_name}"
    add_field => { "type" => "byte_stat" }
    flush_interval => 60
    clear_interval => 60
  }
}