如何在Splunk中合并主备系统消息线程并按到达时间排序

需求说明 #

• 主、备系统各有10条消息组成独立线程，需合并为包含主系统5条、备系统5条关联消息的新线程
• 主系统同线程内所有消息共用唯一srcMsgId；备系统消息通过Mainsys_srcMsgId字段关联对应主系统的srcMsgId，且备系统消息晚于主系统消息到达Splunk
• 需从服务器接收的混杂消息中筛选出目标主备系统消息，最终按消息到达服务器的时间顺序（主系统消息在前，备系统消息在后）展示全部10条消息

消息示例 #

主系统消息 #

{"event": "Sourcetype test please","sourcetype": "testsystem-2","host": "some-host-123","fields":{"messageId": "ED280816-E404-444A-A2D9-FFD2D171F32","srcMsgId": "rwfsdfsfqwe121432gsgsfgd71","Mainsys_srcMsgId": "","baseSystemId": "abc1","routeInstanceId": "abc2","routepointID": "abc3","eventTime": "1985-04-12T23:20:50Z","messageType": "abc4"}}

备系统消息 #

{"event": "Sourcetype test please","sourcetype": "testsystem-2","host": "some-host-123","fields":{"messageId": "ED280816-E404-444A-A2D9-FFD2D171F23","srcMsgId": "rwfsdfsfqwe121432gsgsfgd72","Mainsys_srcMsgId": "rwfsdfsfqwe121432gsgsfgd71","baseSystemId": "abc1","routeInstanceId": "abc2","routepointID": "abc3","eventTime": "1985-04-12T23:20:50Z","messageType": "abc4","GISGMPRequestID": "PS000BA780816-E404-444A-A2D9-FFD2D1712345","GISGMPResponseID": "PS000BA780816-E404-444B-A2D9-FFD2D1712345","resultcode": "abc7","resultdesc": "abc8"}}

合并查询方案 #

以下是满足需求的Splunk查询语句，关键步骤已标注注释：

index="bl_logging" sourcetype="testsystem-2"
# 筛选主系统（Mainsys_srcMsgId为空）和备系统（Mainsys_srcMsgId非空）的有效消息
| where isnotnull(srcMsgId) AND (isnull(Mainsys_srcMsgId) OR isnotnull(Mainsys_srcMsgId))
# 统一线程标识：主系统用自身srcMsgId，备系统用关联的主系统srcMsgId
| eval thread_id=if(isnull(Mainsys_srcMsgId), srcMsgId, Mainsys_srcMsgId)
# 按线程分组，统计主备消息数量，筛选出各5条的目标线程
| stats 
    count(eval(isnull(Mainsys_srcMsgId))) as main_msg_count 
    count(eval(isnotnull(Mainsys_srcMsgId))) as backup_msg_count 
    values(_time) as event_times 
    values(*) as msg_details 
    by thread_id
| where main_msg_count=5 AND backup_msg_count=5
# 拆分聚合结果，还原单条消息
| mvexpand msg_details
# 提取原始字段并恢复对应时间戳
| eval 
    _time=mvindex(event_times, mvfind(msg_details, "messageId=\"".mvindex(msg_details, "messageId")."\"")),
    srcMsgId=mvindex(msg_details, "srcMsgId"),
    Mainsys_srcMsgId=mvindex(msg_details, "Mainsys_srcMsgId"),
    messageId=mvindex(msg_details, "messageId"),
    event=mvindex(msg_details, "event")
# 先按主备类型排序（主在前），再按到达时间排序
| sort 0 -isnull(Mainsys_srcMsgId) _time
# 展示指定字段，可按需调整
| table _time thread_id srcMsgId Mainsys_srcMsgId messageId event

语句说明 #

1. 筛选目标消息：过滤出带有有效srcMsgId的主、备系统消息
2. 标记线程ID：用主系统的srcMsgId统一标识关联的主备消息线程
3. 筛选符合数量要求的线程：统计每个线程内主备消息数量，仅保留各5条的线程
4. 还原单条消息：拆分聚合结果，恢复每条原始消息的字段和时间戳
5. 排序展示：优先按主备类型排序（主系统在前），再按消息到达时间排序，最终输出指定字段

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