如何解决Logstash aggregation超时后message、newfield为空的问题

根因说明 #

开启push_map_as_event_on_timeout参数后，超时触发推送的聚合事件是基于aggregate插件内部维护的map字典生成的全新事件，你当前的code逻辑仅在map中存储了message_count计数字段，没有把需要保留的message、newfield存入map，因此生成的聚合事件默认没有这两个字段。

修复后的配置 #

只需修改aggregate段的code逻辑，首次匹配相同task_id的事件时，将需要保留的公共字段存入map即可，修改后的aggregate配置如下：

aggregate {
    task_id => "%{message}"
    code => "
        map['message_count'] ||= 0; 
        map['message_count'] += 1;
        // 首次初始化map时存入需要保留的公共字段，同task_id的事件message一致，仅需存储一次
        map['message'] ||= event.get('message');
        map['newfield'] ||= event.get('newfield');
    "
    push_map_as_event_on_timeout => true
    timeout_timestamp_field => "@timestamp"
    timeout => 60
    inactivity_timeout => 50
    timeout_tags => ['_aggregatetimeout']
}

生产环境优化建议 #

1. 当前prune插件的白名单配置的是timestamp，但syslog输入默认携带的时间字段为@timestamp，如果需要保留时间字段可同步修正白名单配置，避免时间字段被过滤丢失。
2. 高并发场景下建议将task_id调整为%{[@metadata][event_type]}，而非全量message。如果同类型事件的message存在微小差异（如携带不同的用户ID、末尾时间戳），用全量message作为task_id会生成大量独立聚合任务，占用过多内存，也不符合按事件类型聚合统计的需求。

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