创建可读且可维护的摄取管道
在创建摄取管道时，有许多方法可以实现相似的结果，这可能会使维护和可读性变得困难。本指南概述了您可以遵循的模式，以简化摄取管道的维护和可读性，同时不牺牲功能。
本指南不提供关于优化摄取管道性能的指导。
在创建摄取管道时，在条件语句和脚本中访问字段有几种选项。所有格式都可以用来引用字段，因此请选择能让您的管道更易于阅读和维护的那一种。
| 表示法 | 示例 | 注意 |
|---|---|---|
| 点号表示法 | ctx.event.action |
在条件语句和 Painless 脚本中受支持。 |
| 方括号表示法 | ctx['event']['action'] |
在条件语句和 Painless 脚本中受支持。 |
| 混合点号和方括号表示法 | ctx.event['action'] |
在条件语句和 Painless 脚本中受支持。 |
| 字段 API (Field API) |
field('event.action', '') 或 $('event.action','') |
在条件语句和 Painless 脚本中受支持。 |
| 字段 API (Field API) | field('event.action', '') 或 $('event.action','') |
仅在 Painless 脚本中受支持。 |
以下是针对具体情况选择正确选项的一些通用指导原则。
除了可在脚本处理器本身中使用之外，字段 API 还可用于条件语句（处理器的 if 语句）。
这是访问字段的首选方式。
优势
- 干净且易于阅读。
- 自动处理空 (null) 值。
- 增加对诸如
isEmpty()等附加函数的支持，以简化比较。 - 将点号处理为字段名称的一部分。
- 将点号处理为对象表示法的点遍历 (dot walking)。
- 处理特殊字符。
限制
- 并非在所有条件语句版本中都可用。
优势
- 干净且易于阅读。
- 支持空安全操作符
?。在 使用空安全操作符 (?.) 中了解更多信息。
限制
- 不支持包含
.或任何特殊字符（例如@）的字段名称。请改用 方括号表示法。
优势
- 支持字段名称中包含诸如
@的特殊字符。例如，如果有一个名为has@!%&chars的字段，您将使用ctx['has@!%&chars']。 - 支持包含
.的字段名称。例如，如果有一个名为foo.bar的字段，如果您使用ctx.foo.bar，它将尝试访问对象ctx中对象foo中的bar字段。如果您使用ctx['foo.bar']，则可以直接访问该字段。
限制
- 比点号表示法稍微冗长一些。
- 不支持空安全操作
?。请改用 点号表示法。
优势
- 您还可以混合使用点号表示法和方括号表示法，以充分利用这两种格式的优点。例如，您可以使用
ctx.my.nested.object['has@!%&chars']。然后，您可以在使用点号表示法的字段上使用?操作符，同时仍然访问名称包含特殊字符的字段：ctx.my?.nested?.object['has@!%&chars']。
限制
- 稍微较难阅读。
使用条件语句 (if 语句) 来确保仅在满足特定条件时才应用摄取管道处理器。
预见数据的潜在问题，并使用 空安全操作符 (?.) 以防止数据被错误处理。
对于第一级对象，没有必要使用空安全操作符（例如，使用 ctx.openshift 而不是 ctx?.openshift）。只有在整个 _source 为空时，ctx 才可能为 null。
例如，如果您只想要 ctx.openshift.origin.threadId 字段中包含有效字符串的数据
不要：让条件容易发生故障并使用冗余检查ctx.openshift.origin != null
&& ctx.openshift.origin.threadId != null
- 同时检查
openshift.origin和openshift.origin.threadId是不必要的。 - 如果未正确设置
openshift，这将失败，因为它假设ctx.openshift和ctx.openshift.origin都存在。
应当：使用空安全操作符ctx.openshift?.origin?.threadId instanceof String
- 只有在存在
ctx.openshift和ctx.openshift.origin的情况下，它才会检查ctx.openshift.origin.threadId并确保它是一个字符串。
如果您使用的是空安全操作符，当值不为 null 时它会返回该值，因此在检查该值的类型之前，没有理由去检查该值是否不为 null。
例如，如果您只希望在 ctx.openshift.origin.eventPayload 字段的值为字符串时获取数据
不要：使用冗余检查ctx?.openshift?.eventPayload != null && ctx.openshift.eventPayload instanceof String
应当：将空安全操作符与类型检查结合使用ctx.openshift?.eventPayload instanceof String
当使用 布尔 OR 操作符 (||) 时，您需要为被检查的两个条件都使用空安全操作符。
例如，如果您希望在 ctx.event.type 字段的值为 null 或 '0' 时包含数据
不要：让条件容易发生故障ctx.event.type == null || ctx.event.type == '0'
- 如果未正确设置
ctx.event，这将失败，因为它假设ctx.event存在。如果它在第一个条件上失败，甚至不会尝试第二个条件。
应当：在两个条件中都使用空安全操作符ctx.event?.type == null || ctx.event?.type == '0'
- 两个条件都将被检查。
当您已经遍历了对象路径时，通常没有必要多次使用空安全操作符 (?.)。
例如，如果您正在检查 ctx.arbor.ddos 的两个不同子属性的值
不要：使用冗余的空安全操作符ctx.arbor?.ddos?.subsystem == 'CLI' && ctx.arbor?.ddos?.command_line != null
应当：仅在需要的地方使用空安全操作符ctx.arbor?.ddos?.subsystem == 'CLI' && ctx.arbor.ddos.command_line != null
- 由于
if条件是从左到右求值的，一旦ctx.arbor?.ddos?.subsystem == 'CLI'通过，您就知道ctx.arbor.ddos存在，因此您可以放心地省略第二个?。
在检查字段是否非空时，避免使用冗余的空安全操作符，并使用清晰、简洁的条件。
ctx?.user?.geo?.region != null && ctx?.user?.geo?.region != ''
一旦您检查了 ctx.user?.geo?.region != null，您就可以在下一个条件中安全地访问 ctx.user.geo.region。
ctx.user?.geo?.region != null && ctx.user.geo.region != ''
应当：对字符串使用 要检查字符串字段是否非空，请在条件中使用 isEmpty() 方法。例如
ctx.user?.geo?.region instanceof String && ctx.user.geo.region.isEmpty() == false
- 这可以确保字段存在、是字符串且不为空。
对于此类检查，您还可以省略 instanceof String 并使用 Elvis 操作符，例如 if: ctx.user?.geo?.region?.isEmpty() ?: false。这仅在 region 为 String 时才有效。如果它是 double、object 或任何其他没有 isEmpty() 函数的类型，它将失败并抛出 Java Function not found 错误。
完整示例
这是一个可复现的完整示例
POST _ingest/pipeline/_simulate
{
"docs": [
{
"_source": {
"user": {
"geo": {
"region": "123"
}
}
}
},
{
"_source": {
"user": {
"geo": {
"region": ""
}
}
}
},
{
"_source": {
"user": {
"geo": {
"region": null
}
}
}
},
{
"_source": {
"user": {
"geo": null
}
}
}
],
"pipeline": {
"processors": [
{
"set": {
"field": "demo",
"value": true,
"if": "if": "ctx.user?.geo?.region != null && ctx.user.geo.region != ''"
}
}
]
}
}
当使用 布尔 OR 操作符 (||) 时，if 条件可能会变得不必要地复杂且难以维护，尤其是当链式连接许多 OR 检查时。相反，考虑使用基于数组的检查（如 .contains()）来简化您的逻辑并提高可读性。
不要：运行许多 OR"if": "ctx?.kubernetes?.container?.name == 'admin' || ctx?.kubernetes?.container?.name == 'def'
|| ctx?.kubernetes?.container?.name == 'demo' || ctx?.kubernetes?.container?.name == 'acme'
|| ctx?.kubernetes?.container?.name == 'wonderful'
应当：使用 contains 进行比较["admin","def","demo","acme","wonderful"].contains(ctx.kubernetes?.container?.name)
此示例仅检查精确匹配。如果您需要检查部分匹配，请不要使用此方法。
处理数据大小时，请在 Elasticsearch 中将所有值存储为字节（使用 long 类型）。这确保了一致性，并允许您利用 Kibana 数据视图中的高级格式设置来显示人类可读的大小。
不要：使用多个 避免链式连接多个 gsub 处理器来剥离单位并手动转换值。这种方法容易出错、难以维护，并且很容易漏掉边界情况。
{
"gsub": {
"field": "document.size",
"pattern": "M",
"replacement": "",
"ignore_missing": true,
"if": "ctx?.document?.size != null && ctx.document.size.endsWith(\"M\")"
}
},
{
"gsub": {
"field": "document.size",
"pattern": "(\\d+)\\.(\\d+)G",
"replacement": "$1$200",
"ignore_missing": true,
"if": "ctx?.uws?.size != null && ctx.document.size.endsWith(\"G\")"
}
},
{
"gsub": {
"field": "document.size",
"pattern": "G",
"replacement": "000",
"ignore_missing": true,
"if": "ctx?.uws?.size != null && ctx.document.size.endsWith(\"G\")"
}
}
应当：使用 bytes 处理器可以自动解析并转换诸如 "100M" 或 "2.5GB" 的字符串为其字节值。这更可靠、更易于维护，并支持广泛的单位。
POST _ingest/pipeline/_simulate
{
"docs": [
{
"_source": {
"document": {
"size": "100M"
}
}
}
],
"pipeline": {
"processors": [
{
"bytes": {
"field": "document.size"
}
}
]
}
}
将值存储为字节后，您可以使用 Kibana 的字段格式设置以人类友好的格式（KB、MB、GB 等）显示它们，而无需更改底层数据。
重命名处理器 (rename processor) 用于重命名字段。有两个标志 (flags)：
ignore_missing：当您不确定要重命名的字段是否存在时非常有用。ignore_failure：有助于处理遇到的任何失败。例如，重命名处理器只能重命名为不存在的字段。如果您已经有了字段abc，并且想将def重命名为abc，则该操作将失败。
如果没有内置处理器能实现您的目标，您可能需要在摄取管道中使用 脚本处理器。请确保编写清晰、简洁且可维护的脚本。
上面讨论的所有 访问字段 及其检索值的方法都适用于脚本上下文。在访问字段时，空值处理 (Null handling) 仍然是一个重要的方面。
字段 API 是添加新字段的推荐方法。
例如，根据 cpu.usage 字段的值添加一个新字段 system.cpu.total.norm.pct。现有 cpu.usage 字段的值是一个 0-100 范围内的数字。新字段 system.cpu.total.norm.pct 的值范围将是 0-1.0，其中 1 相当于 cpu.usage 字段中的 100。
选项 1：字段 API（首选） 创建一个新字段 system.cpu.total.norm.pct 并将值设置为 cpu.usage 字段的值除以 100.0。
POST _ingest/pipeline/_simulate
{
"docs": [
{
"_source": {
"cpu": {
"usage": 90
}
}
}
],
"pipeline": {
"processors": [
{
"script": {
"source": """
field('system.cpu.total.norm.pct').set($('cpu.usage',0.0)/100.0)
"""
}
}
]
}
}
- 此字段期望值为 0-1，而不是 0-100。在重命名该字段时，将此值除以 100 以获得正确的值。
fieldAPI 的公开形式为field(<field name>)。set(<value>)负责设置值。我们在内部使用$(<field name>, fallback)从现有字段中读取值。最后我们除以100.0。.0非常重要，否则它将执行纯整数除法并仅返回 0 而不是 0.9。
选项 2：不使用字段 API 如果不使用字段 API，则需要编写更多代码来确保能够遍历 system.cpu.total.norm.pct 的完整路径。
{
"script": {
"source": "
if(ctx.system == null){
ctx.system = new HashMap();
}
if(ctx.system.cpu == null){
ctx.system.cpu = [:];
}
if(ctx.system.cpu.total == null){
ctx.system.cpu.total = [:];
}
if(ctx.system.cpu.total.norm == null){
ctx.system.cpu.total.norm = [:];
}
ctx.system.cpu.total.norm.pct = $('cpu.usage', 0.0)/100.0;
"
}
}
- 检查对象是否为 null，然后创建它们。
- 创建一个新的
HashMap以将所有对象存储在其中。 - 不要编写
new HashMap()，而是使用快捷方式[:]。 - 执行与上述相同的计算并设置值。
不要：使用冗长且容易出错的脚本编写模式{
"script": {
"source": """
String timeString = ctx['temp']['duration'];
ctx['event']['duration'] = Integer.parseInt(timeString.substring(0,2))*360000 + Integer.parseInt(timeString.substring(3,5))*60000 + Integer.parseInt(timeString.substring(6,8))*1000 + Integer.parseInt(timeString.substring(9,12)); <2> <3>
""",
"if": "ctx.temp != null && ctx.temp.duration != null"
}
}
- 避免使用方括号而不是点号表示法来访问字段。
ctx['event']['duration']：不要在未确保父属性存在的情况下尝试访问子属性。timeString.substring(0,2)：避免手动解析子字符串，而应利用日期/时间解析实用工具。- 按照 ECS 的期望，
event.duration应该是纳秒，而不是毫秒。 - 避免冗余的空值检查，而应使用空安全操作符 (
?.)。
这种方法难以阅读、容易出错，并且没有利用 Painless 中提供的强大日期/时间功能。
应当：使用空安全操作符和内置的日期/时间实用工具POST _ingest/pipeline/_simulate
{
"docs": [
{
"_source": {
"temp": {
"duration": "00:00:06.448"
}
}
}
],
"pipeline": {
"processors": [
{
"script": {
"source": """
if (ctx.event == null) {
ctx.event = [:];
}
DateTimeFormatter formatter = DateTimeFormatter.ofPattern("HH:mm:ss.SSS");
LocalTime time = LocalTime.parse(ctx.temp.duration, formatter);
ctx.event.duration = time.toNanoOfDay();
""",
"if": "ctx.temp?.duration != null"
}
}
]
}
}
- 在对
event对象进行赋值之前，确保该对象存在。 - 使用
DateTimeFormatter和LocalTime解析持续时间字符串。 - 以纳秒为单位存储持续时间，正如 ECS 所期望的那样。
- 使用空安全操作符检查字段是否存在。
在摄取管道中重建或规范化 IP 地址时，应避免不必要的复杂性和冗余操作。
{
"script": {
"source": """
String[] ipSplit = ctx['destination']['ip'].splitOnToken('.');
String ip = Integer.parseInt(ipSplit[0]) + '.' + Integer.parseInt(ipSplit[1]) + '.' + Integer.parseInt(ipSplit[2]) + '.' + Integer.parseInt(ipSplit[3]);
ctx['destination']['ip'] = ip;
""",
"if": "(ctx['destination'] != null) && (ctx['destination']['ip'] != null)"
}
}
- 使用方括号表示法而不是点号表示法来访问字段。
- 解析字符串片段时进行了不必要的
Integer类型转换。 - 为 IP 字符串分配了一个额外的变量，而不是直接设置字段。
- 没有检查
destination是否作为一个对象可用。
应当：使用简洁、可读且安全的脚本POST _ingest/pipeline/_simulate
{
"docs": [
{
"_source": {
"destination": {
"ip": "192.168.0.1.3.4.5.6.4"
}
}
}
],
"pipeline": {
"processors": [
{
"script": {
"source": """
def temp = ctx.destination.ip.splitOnToken('.');
ctx.destination.ip = temp[0] + "." + temp[1] + "." + temp[2] + "." + temp[3];
""",
"if": "ctx.destination?.ip != null"
}
}
]
}
}
- 使用点号表示法进行字段访问。
- 避免不必要的类型转换和额外变量。
- 使用空安全操作符 (
?.) 检查字段是否存在。
这种方法更具可维护性，避免了不必要的操作，并确保您的管道脚本稳健且易于理解。
不要：在使用日期处理器之前删除 运行日期处理器之前显式删除 @timestamp 字段是一个常见的错误，如下所示
{
"set": {
"field": "openshift.timestamp",
"value": "{{openshift.date}} {{openshift.time}}",
"if": "ctx?.openshift?.date != null && ctx?.openshift?.time != null && ctx?.openshift?.timestamp == null"
}
},
{
"remove": {
"field": "@timestamp",
"ignore_missing": true,
"if": "ctx?.openshift?.timestamp != null || ctx?.openshift?.timestamp1 != null"
}
},
{
"date": {
"field": "openshift.timestamp",
"formats": [
"yyyy-MM-dd HH:mm:ss",
"ISO8601"
],
"timezone": "Europe/Vienna",
"if": "ctx?.openshift?.timestamp != null"
}
}
此删除步骤是不必要的，甚至可能适得其反。date 处理器会自动用源字段中解析出的日期覆盖 @timestamp 中的值，除非您显式设置了不同的 target_field。无需事先删除 @timestamp——处理器会为您处理更新。
删除 @timestamp 还可能引入隐蔽的错误，特别是如果日期处理器被跳过或失败，导致您的文档缺少时间戳。
Mustache 是 Elasticsearch 摄取管道中使用的简单模板语言，用于将字段值动态插入到字符串中。您可以使用双大括号 ({{ }}) 来引用文档中的字段，从而在 set、rename 等处理器中实现灵活且动态的值分配。
例如，{{host.hostname}} 将在运行时替换为 host.hostname 字段的值。Mustache 支持访问嵌套字段、数组，甚至为条件渲染提供了一些基本逻辑。
当您需要使用 Mustache 模板引用数组中的特定元素时，可以使用带有从零开始的索引的点号表示法。例如，要访问 tags 数组中的第一个值，请在数组字段名称后面使用 .0。
应当：在 Mustache 模板中使用数组索引表示法POST _ingest/pipeline/_simulate
{
"docs": [
{
"_source": {
"host": {
"hostname": "abc"
},
"tags": [
"cool-host"
]
}
}
],
"pipeline": {
"processors": [
{
"set": {
"field": "host.alias",
"value": "{{tags.0}}"
}
}
]
}
}
在此示例中，{{tags.0}} 检索 tags 数组的第一个元素 ("cool-host") 并将其分配给 host.alias 字段。当您想要从数组中提取特定值以在文档的其他地方使用时，这种方法是必需的。使用正确的索引可确保您获得预期值，并且此模式适用于源数据中的任何数组字段。
每当您需要将原始 _source 存储在字段 event.original 中时，请使用 mustache 函数 {{#toJson}}<field>{{/toJson}}。
POST _ingest/pipeline/_simulate
{
"docs": [
{
"_source": {
"foo": "bar",
"key": 123
}
}
],
"pipeline": {
"processors": [
{
"set": {
"field": "event.original",
"value": "{{#toJson}}_source{{/toJson}}"
}
}
]
}
}