Logstash Aggregate插件timeout_code中map未定义问题求助

错误原因 #

Aggregate插件的timeout_code代码块与code块的作用域不同，map变量仅在code块内可直接访问，timeout_code中无法直接引用该变量。

解决方案 #

由于你已经设置了push_map_as_event_on_timeout => true，超时触发时生成的事件会自动包含map中的所有键值对（如event_count、timestamp），直接从event对象中获取这些字段即可。同时需要注意：timeout_code中无法重置原聚合map——超时触发后对应task_id的map会被销毁，下次相同task_id的事件进来时会重新初始化map。

修改后的完整过滤器代码 #

aggregate {
    task_id => "%{project_id}-%{enr_subscription_id}-%{asset_id}"
    code => "
      map['event_count'] ||= 0
      map['event_count'] += 1
      map['timestamp'] = event.get('@timestamp')
      # 将task_id关联的字段存入map，确保超时事件能获取到这些值
      map['project_id'] = event.get('project_id')
      map['enr_subscription_id'] = event.get('enr_subscription_id')
      map['asset_id'] = event.get('asset_id')
    "
    push_map_as_event_on_timeout => true
    timeout => 60  # 设置超时时间为1分钟
    timeout_tags => ['_aggregatetimeout']
    timeout_code => "
      event.set('kafka_map', {
        'event_count' => event.get('event_count'),
        'timestamp' => event.get('timestamp'),
        'project_id' => event.get('project_id'),
        'enr_subscription_id' => event.get('enr_subscription_id'),
        'asset_id' => event.get('asset_id')
      })
    "
}

关键调整点 #

1. 在code块中把project_id、enr_subscription_id、asset_id存入map，确保超时生成的事件能携带这些字段，避免event.get()取不到值的情况。
2. timeout_code中直接通过event.get()获取聚合后的字段值，不再引用map变量。

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