如何在Elasticsearch中实现指定时间范围的事件共现查询与告警？

我正尝试在指定时间范围内查询事件的共现情况，但始终无法找到可行的查询方法。

例如，Events表的片段如下：

**目标是创建告警：当同一agent.name在60分钟内同时触发file-created和created-process事件时，生成告警。**事件顺序无关紧要。基于上述数据，应返回两条告警：

在SQL中可通过自连接Events表实现此需求，但Elasticsearch不支持该操作。Elasticsearch提供sequence查询选项，但它要求事件有预定义顺序，而非无序的事件组。虽然查询两个事件的共现可通过两个sequence实现，但当需要查询三个及以上event.action的共现时，操作会变得极为复杂。

我曾尝试使用阈值规则，但似乎无法在聚合层面检查多个不同的阈值条件；也尝试过聚合规则，但它仅支持数值型聚合。

我想到一种位填充的方法：通过二进制列的按位或运算，再利用位掩码操作判断索引中是否存在目标event.action，但这种方法可读性极差且难以调试。

以下是该方法的简短示例：

位填充示例开始

首先，定义二进制值映射：

event.action	agent.name	binary
file-created	PC 1	0000 0000 0001
permissions-changed	PC 1	0000 0000 0010
created-process	PC 1	0000 0000 0100
logged-in-explicit	PC 1	0000 0000 1000
privileged-service-called	PC 1	0000 0001 0000
logged-in	PC 1	0000 0010 0000
file-created	PC 2	0000 0100 0000
permissions-changed	PC 2	0000 1000 0000
created-process	PC 2	0001 0000 0000
logged-in-explicit	PC 2	0010 0000 0000
privileged-service-called	PC 2	0100 0000 0000
logged-in	PC 2	1000 0000 0000

接着，转换Events表中的部分行：

ID	@timestamp	event.action	agent.name	binary
1	14:12:00	file-created	PC 1	0000 0000 0001
2	14:15:00	permissions-changed	PC 2	0000 1000 0000
3	14:28:34	created-process	PC 2	0001 0000 0000
7	14:58:27	file-created	PC 2	0000 0100 0000

随后对二进制列执行按位或运算：

0000 0000 0001
   0000 1000 0000  
   0001 0000 0000 
   0000 0100 0000 
 -----------------
   0001 1100 0001

通过0001 1100 0001对照映射表可知，由于从右数第7位和第8位均为1，应生成告警。

位填充示例结束

除位填充外，我还考虑创建新索引（如file-process-created alert），并在数据管道中查询由agent.name关联的两个唯一event.action。但此方法会为每个告警生成单独的索引，同时还需额外规则来查询唯一事件动作。

请问还有其他方法或解决方案可实现这种嵌套查询吗？

可行解决方案 #

1. 滑动窗口聚合 + 脚本化指标聚合 #

利用Elasticsearch的滑动窗口聚合按时间分片，结合scripted_metric聚合检查每个agent的时间窗口内是否包含目标事件类型，最后用bucket_selector筛选出符合条件的结果。

{
  "size": 0,
  "aggs": {
    "time_windows": {
      "sliding_window": {
        "interval": "60m",
        "shift": "1m"
      },
      "aggs": {
        "by_agent": {
          "terms": {
            "field": "agent.name.keyword"
          },
          "aggs": {
            "check_event_cooccurrence": {
              "scripted_metric": {
                "init_script": "state.action_set = new HashSet();",
                "map_script": "if (doc['event.action.keyword'].size() > 0) { state.action_set.add(doc['event.action.keyword'].value); }",
                "combine_script": "return state.action_set;",
                "reduce_script": "def required_actions = ['file-created', 'created-process']; def has_all = true; for (action in required_actions) { def found = false; for (set in states) { if (set.contains(action)) { found = true; break; } } if (!found) { has_all = false; break; } } return has_all;"
              }
            },
            "filter_alerts": {
              "bucket_selector": {
                "buckets_path": {
                  "is_alert": "check_event_cooccurrence"
                },
                "script": "params.is_alert == true"
              }
            }
          }
        }
      }
    }
  }
}

2. Elasticsearch Transform 物化视图 #

创建Transform按agent和60分钟时间窗口聚合事件类型，生成物化视图后直接在视图上配置告警规则，无需重复计算。

{
  "source": {
    "index": ["events"]
  },
  "dest": {
    "index": "agent-event-cooccurrence"
  },
  "pivot": {
    "group_by": {
      "agent_name": {
        "terms": {
          "field": "agent.name.keyword"
        }
      },
      "time_window": {
        "date_histogram": {
          "field": "@timestamp",
          "calendar_interval": "60m"
        }
      }
    },
    "aggregations": {
      "distinct_actions": {
        "terms": {
          "field": "event.action.keyword",
          "size": 20
        }
      }
    }
  },
  "sync": {
    "time": {
      "field": "@timestamp",
      "delay": "1m"
    }
  }
}

创建完成后，可在agent-event-cooccurrence索引上设置告警规则，检查distinct_actions.buckets是否同时包含file-created和created-process两个key。

3. Watcher 定时告警脚本 #

通过Watcher定时查询指定时间范围的事件，用脚本判断agent是否满足共现条件，触发告警通知。

{
  "trigger": {
    "schedule": {
      "interval": "5m"
    }
  },
  "input": {
    "search": {
      "request": {
        "indices": ["events"],
        "body": {
          "size": 0,
          "query": {
            "range": {
              "@timestamp": {
                "gte": "now-60m"
              }
            }
          },
          "aggs": {
            "agents": {
              "terms": {
                "field": "agent.name.keyword"
              },
              "aggs": {
                "actions": {
                  "terms": {
                    "field": "event.action.keyword"
                  }
                }
              }
            }
          }
        }
      }
    }
  },
  "condition": {
    "script": {
      "source": "def required = ['file-created', 'created-process']; def alertAgents = []; for (bucket in ctx.payload.aggregations.agents.buckets) { def actionKeys = bucket.actions.buckets.stream().map(b -> b.key).collect(Collectors.toSet()); if (actionKeys.containsAll(required)) { alertAgents.add(bucket.key); } } return alertAgents.size() > 0;"
    }
  },
  "actions": {
    "send_email_alert": {
      "email": {
        "to": "your-alert-recipient@example.com",
        "subject": "Event Co-occurrence Alert",
        "body": "Agents with file-created and created-process events in the last 60 minutes: {{ctx.payload.aggregations.agents.buckets.stream().filter(b -> b.actions.buckets.stream().map(a -> a.key).collect(Collectors.toSet()).containsAll(['file-created', 'created-process'])).map(b -> b.key).collect(Collectors.joining(', '))}}"
      }
    }
  }
}

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