数据流

核心概念

默认数据流

安全数据湖内置名为'所有消息'的默认数据流，自动接收所有不符合其他数据流规则的日志数据。

该默认数据流不可删除或暂停，作为所有传入消息的兜底容器。

当创建带有特定规则的自定义数据流时，仅匹配规则的消息会被路由至对应数据流，其余消息仍保留在默认数据流中。

数据流规则

注意

数据流规则被视为遗留功能，将在未来安全数据湖版本中弃用。建议改用流水线，因其在消息过滤、富化和路由方面提供更高灵活性。

数据流规则用于定义消息路由至数据流的条件。每个数据流可包含多个实时评估字段、值或关键词的规则。

规则可通过AND、OR等逻辑运算符组合。例如：数据流规则可过滤特定IP地址的消息，或根据 错误 ”与 警告 .

流水线与数据流

流水线可附加至数据流进行深度处理。

流水线由包含流水线规则的连续阶段构成，这些规则对符合特定条件的消息生效。流水线规则可富化、转换或重路由消息——例如提取字段、重命名字段键或标准化值。

了解如何将流水线附加至数据流，请参阅数据路由 .

数据流与保存搜索对比

数据流与保存搜索在于时间性：

数据流实时处理消息，支持即时告警与转发。
保存的搜索是对存储数据的静态查询。

由于数据流在处理时会对每条消息打上标记，后续通过流ID筛选比每次搜索时重建复杂查询要快得多。

管理数据流

您可通过顶级菜单中的“数据流”页面管理现有数据流。

每个列出的数据流包含以下关键信息：

归档 – 是否已为归档启用与该数据流关联的索引集。
数据湖 – 显示是否为此数据流配置了数据湖目标。
规则 – 附加的流规则数量。
管道 – 应用的管道数量。
吞吐量 – 消息处理速率。
状态 – 显示数据流处于运行（活跃）或暂停（非活跃）状态。
数据路由 – 打开流规则、管道和目标的配置向导。更多信息请参阅此知识库文章 .
共享 – 将此数据流配置分享给其他用户或团队 .
更多操作 ——访问其他选项，如管理流规则、克隆流或设置告警 .

创建新流

要创建新流以路由日志数据，请按以下步骤操作：

转到页面顶部的流选项卡。
选择 创建流 .
输入唯一标题和描述，例如 所有错误消息 或 按来源分类的网络日志 .
选择此流消息将写入的索引集。

注意

之后可通过数据路由修改此目标，例如将消息路由至数据湖。
（可选）启用 从‘默认流’中移除匹配项 以将这些消息排除在默认流之外。
选择 创建流 .
该流现已保存但处于未激活状态。要开始处理数据，请选择“暂停”状态来激活它，激活后将变为“运行中”。

创建和修改流规则

流规则决定了哪些消息会流入您的流。您可以通过 “管理规则”菜单或 “快速添加规则” .

通过管理规则创建规则

要在管理规则页面创建规则，请按以下步骤操作：

在“流”页面，找到需要修改的流，选择“更多”，然后选择“管理规则”。
在 1. 加载消息以测试规则 下，选择最近的日志消息或通过消息ID加载一条消息。
在 2. 管理流规则 下，选择“添加流规则”并填写以下字段：
- 字段：要过滤的字段名称（例如：source_ip）。
- 类型：如何匹配值（例如：完全匹配、大于或包含）。
- 值：要匹配的值（例如：特定IP地址）。
- 反向：使规则具有排除性（例如：排除特定IP）。
- 描述（可选）：描述规则的简短说明。
选择 创建规则 .
根据需要重复添加更多条件。
指定消息必须匹配全部规则还是至少一条规则才能进入流。
选择 完成！ 以完成设置。

如需后续编辑或删除规则，请返回 管理规则 并使用编辑或删除选项。

使用快速添加规则创建规则

快速添加规则步骤如下：

在流页面，点击更多 > 快速添加规则 为目标流操作。
在 2. 管理流规则 中，选择添加流规则并填写字段：
- 字段 : 要筛选的字段名称（如source_ip）。
- 类型 : 匹配值的方式（如完全匹配、大于或包含）。
- 值 : 要匹配的值（如特定IP地址）。
- 反转：使规则具有排除性（例如排除特定IP）。
- 描述（可选）：描述规则的简短说明。
在此处可为当前数据流添加新的流式规则以过滤日志数据。请按照前一章节第5步的指引创建新规则。
点击 创建规则 进行保存。

若要修改或删除通过快速添加创建的规则，请使用更多 > 管理规则 （位于同一数据流界面）。

从数据流路由日志

数据流是安全数据湖 .

数据路由始于数据流层级，您可在此应用规则、绑定管道并定义目标。有关配置数据路由的详细说明，请参阅数据路由 .

流式处理

安全数据湖中每条传入消息都会与所有已配置数据流的规则进行匹配评估。若消息满足某数据流的全部或任一规则（取决于数据流配置方式），该数据流的内部ID将被添加至消息的 streams 数组中。

此分类机制使得所有引用特定数据流的搜索、告警及处理操作能自动将其范围限定于属于该数据流的消息。该过滤过程自动完成，您无需在搜索中手动指定 streams:[STREAM_ID] 参数。

流式处理运行时限制

在消息处理过程中，流分类对系统整体性能起着关键作用。每条消息都会根据用户定义的流规则进行评估，若内容匹配规则条件，则该消息会被添加至相应数据流。

由于这种分类发生在消息索引过程中，规则匹配所需的时间直接影响系统吞吐量和响应速度。

如果某些流规则评估耗时过长（例如由于复杂的正则表达式），处理速度会变慢甚至停滞。严重情况下，消息可能在内存中积压，导致系统不稳定或数据丢失。

为防止这种情况发生，安全数据湖对流规则匹配设置了运行时限制：

若消息与流规则的匹配时间超过配置的超时阈值，该流的规则评估将中止。
消息处理会正常继续，但该事件会被记录为故障。
若某流超过允许的故障次数，安全数据湖将自动禁用该流以维护系统稳定性。

这种机制优先保障整体消息流和系统性能，而非小部分消息的流分类准确性。

但在高负载环境中，若处理时间持续超限可能导致流被频繁禁用，这通常表明系统过载无法处理当前消息量。

配置超时值

若默认超时设置不适合您的环境，可在Graylog服务器配置文件中调整以下参数：

参数	描述	默认值
stream_processing_timeout	定义流规则评估的最大允许时间（毫秒）。超时则中止匹配并记录故障。	`2000` (2秒)
stream_processing_max_faults	设置单流允许的最大超时次数。超过该阈值后流将自动禁用，需手动重新启用。	3

运行时过长的原因

若仅单个流被禁用而其他流正常，很可能该流中存在低效规则。

多数情况下这是由正则表达式引起——根据其复杂度和匹配输入，可能产生高昂计算成本。

特别是复杂或未优化的正则模式会导致灾难性回溯，使特定输入的评估时间呈指数级增长。

若流处理超时持续发生，请按以下步骤排查：

检查受影响流的规则，特别是使用正则表达式的规则。
简化或移除低效规则以减少评估时间。
修改完成后重新启用流，并确认处理性能有所提升。

流连接

流水线不会自行处理消息。要使流水线对传入数据执行操作，必须将其连接到一个或多个流。此连接决定了哪些消息由该流水线评估，从而精确控制不同类型日志数据的处理和丰富方式。

内置函数 route_to_stream 可在流水线规则中使用，将消息路由至另一个流。

消息被路由后，流水线处理器会自动识别并执行连接到目标流的所有流水线。虽然流水线可通过消息路由触发其他流水线，但所有传入消息必须首先通过至少一个连接了流水线的流。

默认流

安全数据湖接收的所有消息最初都会路由至默认流。您可将此流作为流水线处理的入口点，使传入消息能路由到更多流并接受后续处理。

但若您希望使用原始流匹配功能（即流规则），可配置 流水线处理器 使其在 消息过滤链 之后运行（位于 系统 > 配置 页面的 消息处理器配置 部分），并将流水线连接到现有流。这为您提供了对提取、转换和丰富过程的细粒度控制。

默认情况下，安全数据湖接收的所有消息最初都会路由至 默认流 ( 所有消息 ）。您可将此流作为流水线处理的起点，例如将消息路由到其他流或应用特定转换规则。

若您倾向使用传统的 基于流规则的匹配 功能，您可以将 流水线处理器 配置在 消息过滤链 之后运行，位于系统 → 配置 → 消息处理器配置 中。此方法允许您将流水线连接到现有流，并实现对提取、转换和丰富操作的精细控制。

消息处理器顺序

消息处理器的顺序会显著影响消息处理方式。设计流和流水线时，请确保处理器顺序符合预期逻辑，尤其是当后续处理器依赖于链中早期修改或丰富的数据时。

数据流