Elasticsearch 摄取管道
Elasticsearch ingest 管道允许您在索引数据之前对数据执行常见的转换。例如，您可以使用管道删除字段、从文本中提取值以及丰富您的数据。
一个管道由一系列可配置的任务组成，称为 processors（处理器）。每个处理器按顺序运行，对传入的文档进行特定的修改。在处理器运行完毕后，Elasticsearch 会将转换后的文档添加到您的数据流或索引中。
您可以使用 Kibana 的 Ingest Pipelines 功能或 ingest API 来创建和管理 ingest 管道。Elasticsearch 将管道存储在 集群状态 中。
- 具有
ingest节点角色的节点负责处理管道。要使用 ingest 管道，您的集群必须至少有一个带有ingest角色的节点。对于较重的 ingest 负载，我们建议创建 专用的 ingest 节点。 - 如果启用了 Elasticsearch 安全功能，您必须具有
manage_pipeline集群权限 才能管理 ingest 管道。要使用 Kibana 的 Ingest Pipelines 功能，您还需要cluster:monitor/nodes/info集群权限。 - 包含
enrich处理器的管道需要进行额外的设置。请参阅 丰富您的数据。
在 Kibana 中，使用导航菜单或全局搜索字段转到 Ingest Pipelines（采集管道）管理页面。
在列表视图中，您可以
- 查看您的管道列表并深入了解详细信息
- 编辑或克隆现有的管道
- 删除管道
要创建管道，请单击 Create pipeline > New pipeline。有关示例教程，请参阅 示例：解析日志。
New pipeline from CSV 选项允许您使用 CSV 创建一个 ingest 管道，该管道将自定义数据映射到 Elastic Common Schema (ECS)。将自定义数据映射到 ECS 可以使数据更易于搜索，并允许您重用其他数据集的可视化效果。要开始使用，请查看 将自定义数据映射到 ECS。
您还可以使用 ingest API 来创建和管理管道。以下 创建管道 API 请求创建一个管道，其中包含两个 set 处理器，后面跟着一个 lowercase 处理器。处理器按指定的顺序依次运行。
PUT _ingest/pipeline/my-pipeline
{
"description": "My optional pipeline description",
"processors": [
{
"set": {
"description": "My optional processor description",
"field": "my-long-field",
"value": 10
}
},
{
"set": {
"description": "Set 'my-boolean-field' to true",
"field": "my-boolean-field",
"value": true
}
},
{
"lowercase": {
"field": "my-keyword-field"
}
}
]
}
当您创建或更新管道时，可以指定一个可选的 version 整数。您可以将此版本号与 if_version 参数一起使用，以有条件地更新管道。指定 if_version 参数后，成功的更新将使管道的版本递增。
PUT _ingest/pipeline/my-pipeline-id
{
"version": 1,
"processors": [ ... ]
}
要使用 API 取消设置 version 号，请在不指定 version 参数的情况下替换或更新管道。
在生产环境中使用管道之前，我们建议您使用示例文档对其进行测试。在 Kibana 中创建或编辑管道时，单击 Add documents。在 Documents 选项卡中，提供示例文档并单击 Run the pipeline。
您还可以使用 simulate pipeline API 测试管道。您可以在请求路径中指定已配置的管道。例如，以下请求测试 my-pipeline。
POST _ingest/pipeline/my-pipeline/_simulate
{
"docs": [
{
"_source": {
"my-keyword-field": "FOO"
}
},
{
"_source": {
"my-keyword-field": "BAR"
}
}
]
}
或者，您可以在请求正文中指定管道及其处理器。
POST _ingest/pipeline/_simulate
{
"pipeline": {
"processors": [
{
"lowercase": {
"field": "my-keyword-field"
}
}
]
},
"docs": [
{
"_source": {
"my-keyword-field": "FOO"
}
},
{
"_source": {
"my-keyword-field": "BAR"
}
}
]
}
API 将返回转换后的文档
{
"docs": [
{
"doc": {
"_index": "_index",
"_id": "_id",
"_version": "-3",
"_source": {
"my-keyword-field": "foo"
},
"_ingest": {
"timestamp": "2099-03-07T11:04:03.000Z"
}
}
},
{
"doc": {
"_index": "_index",
"_id": "_id",
"_version": "-3",
"_source": {
"my-keyword-field": "bar"
},
"_ingest": {
"timestamp": "2099-03-07T11:04:04.000Z"
}
}
}
]
}
使用 pipeline 查询参数将管道应用于单个或批量索引请求中的文档。
POST my-data-stream/_doc?pipeline=my-pipeline
{
"@timestamp": "2099-03-07T11:04:05.000Z",
"my-keyword-field": "foo"
}
PUT my-data-stream/_bulk?pipeline=my-pipeline
{ "create":{ } }
{ "@timestamp": "2099-03-07T11:04:06.000Z", "my-keyword-field": "foo" }
{ "create":{ } }
{ "@timestamp": "2099-03-07T11:04:07.000Z", "my-keyword-field": "bar" }
您还可以在 按查询更新 或 重新索引 API 中使用 pipeline 参数。
POST my-data-stream/_update_by_query?pipeline=my-pipeline
POST _reindex
{
"source": {
"index": "my-data-stream"
},
"dest": {
"index": "my-new-data-stream",
"op_type": "create",
"pipeline": "my-pipeline"
}
}
使用 index.default_pipeline 索引设置来设置默认管道。如果未指定 pipeline 参数，Elasticsearch 会将此管道应用于索引请求。
使用 index.final_pipeline 索引设置来设置最终管道。Elasticsearch 会在请求管道或默认管道之后应用此管道，即使两者均未指定也是如此。
要将 ingest 管道添加到 Elastic Beat，请在 <BEAT_NAME>.yml 中的 output.elasticsearch 下指定 pipeline 参数。例如，对于 Filebeat，您可以在 filebeat.yml 中指定 pipeline。
output.elasticsearch:
hosts: ["localhost:9200"]
pipeline: my-pipeline
Elastic Agent 集成自带默认的 ingest 管道，这些管道在索引之前对数据进行预处理和丰富。Fleet 使用包含 管道索引设置 的索引模板来应用这些管道。Elasticsearch 根据流的命名方案将这些模板匹配到您的 Fleet 数据流。
每个默认的集成管道都会调用一个不存在的、无版本的 *@custom ingest 管道。如果不进行修改，此管道调用对您的数据没有影响。但是，您可以修改此调用，为集成创建在升级后仍然持久存在的自定义管道。有关详细信息，请参阅 教程：使用自定义 ingest 管道转换数据。
Fleet 没有为 Custom logs 集成提供默认的 ingest 管道，但您可以使用 索引模板 或 自定义配置 为此集成指定管道。
创建并测试您的 ingest 管道。将您的管道命名为
logs-<dataset-name>-default。这使跟踪集成的管道变得更加容易。例如，以下请求为
my-app数据集创建一个管道。该管道的名称为logs-my_app-default。PUT _ingest/pipeline/logs-my_app-default{ "description": "Pipeline for `my_app` dataset", "processors": [ ... ] }创建一个索引模板，在其
index.default_pipeline或index.final_pipeline索引设置中包含您的管道。确保该模板已启用数据流。该模板的索引模式应匹配logs-<dataset-name>-*。您可以使用 Kibana 的 Index Management 功能或 create index template API 来创建此模板。
例如，以下请求创建一个匹配
logs-my_app-*的模板。该模板使用了一个包含index.default_pipeline索引设置的组件模板。# Creates a component template for index settingsPUT _component_template/logs-my_app-settings{ "template": { "settings": { "index.default_pipeline": "logs-my_app-default", "index.lifecycle.name": "logs" } } } # Creates an index template matching `logs-my_app-*`PUT _index_template/logs-my_app-template{ "index_patterns": ["logs-my_app-*"], "data_stream": { }, "priority": 500, "composed_of": ["logs-my_app-settings", "logs-my_app-mappings"] }在 Fleet 中添加或编辑 Custom logs 集成时，单击 Configure integration > Custom log file > Advanced options。
在 Dataset name 中，指定您的数据集名称。Fleet 将把该集成的新数据添加到生成的
logs-<dataset-name>-default数据流中。例如，如果您的数据集名称为
my_app，Fleet 会将新数据添加到logs-my_app-default数据流中。
使用 rollover API 滚动更新您的数据流。这可确保 Elasticsearch 将索引模板及其管道设置应用于该集成的任何新数据。
POST logs-my_app-default/_rollover/
创建并测试您的 ingest 管道。将您的管道命名为
logs-<dataset-name>-default。这使跟踪集成的管道变得更加容易。例如，以下请求为
my-app数据集创建一个管道。该管道的名称为logs-my_app-default。PUT _ingest/pipeline/logs-my_app-default{ "description": "Pipeline for `my_app` dataset", "processors": [ ... ] }在 Fleet 中添加或编辑 Custom logs 集成时，单击 Configure integration > Custom log file > Advanced options。
在 Dataset name 中，指定您的数据集名称。Fleet 将把该集成的新数据添加到生成的
logs-<dataset-name>-default数据流中。例如，如果您的数据集名称为
my_app，Fleet 会将新数据添加到logs-my_app-default数据流中。在 Custom Configurations 中，在
pipeline策略设置中指定您的管道。
Elastic Agent 独立模式
如果您以独立模式运行 Elastic Agent，可以使用包含 index.default_pipeline 或 index.final_pipeline 索引设置的索引模板来应用管道。或者，您也可以在 elastic-agent.yml 配置中指定 pipeline 策略设置。请参阅 安装独立 Elastic Agent。
当您为搜索用例创建 Elasticsearch 索引时（例如，使用 网络爬虫^ 或 连接器），这些索引会自动配置特定的 ingest 管道。这些处理器有助于优化内容以进行搜索。有关更多信息，请参阅 搜索中的 ingest 管道。
处理器对传入文档的源字段具有读写访问权限。要在处理器中访问字段键，请使用其字段名。以下 set 处理器访问 my-long-field。
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"set": {
"field": "my-long-field",
"value": 10
}
}
]
}
您还可以加上 _source 前缀。
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"set": {
"field": "_source.my-long-field",
"value": 10
}
}
]
}
使用点号表示法访问对象字段。
如果您的文档包含展平的对象，请使用 dot_expander 处理器来展开它们。如果您希望保持文档结构，请在管道定义中使用 flexible 访问模式。否则，Ingest 处理器将无法访问包含点号的字段名。
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"dot_expander": {
"description": "Expand 'my-object-field.my-property'",
"field": "my-object-field.my-property"
}
},
{
"set": {
"description": "Set 'my-object-field.my-property' to 10",
"field": "my-object-field.my-property",
"value": 10
}
}
]
}
几个处理器参数支持 Mustache 模板片段。要在模板片段中访问字段值，请用三重大括号将字段名括起来：{{{field-name}}}。您可以使用模板片段来动态设置字段名。
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"set": {
"description": "Set dynamic '<service>' field to 'code' value",
"field": "{{{service}}}",
"value": "{{{code}}}"
}
}
]
}
默认的 ingest 管道访问模式无法识别文档中包含点号的字段名。从 ingest 文档中检索展平的和包含点号的字段名需要一种没有此限制的不同字段检索算法。我们知道，某些管道的逻辑已经依赖于这些带点号的字段名的限制。为了在继续支持原始行为的同时增加对带点号字段名的支持，ingest 管道现在支持为管道中的所有处理器配置要使用的访问模式。
ingest 管道上的 field_access_pattern 属性定义了如何为当前管道中的所有处理器读取和写入 ingest 文档字段。它接受两个值：classic（默认值）和 flexible。
PUT _ingest/pipeline/my-pipeline
{
"field_access_pattern": "classic",
"processors": [
{
"set": {
"description": "Set some searchable tags in our document's flattened field",
"field": "event.tags.ingest.processed_by",
"value": "my-pipeline"
}
}
]
}
- 此管道中的所有处理器都将使用
classic访问模式。 - 处理器用来向 ingest 文档读取和写入值的字段路径解析逻辑基于访问模式。
classic 访问模式是自 ingest 节点首次发布以来一直存在的默认访问模式。传递给处理器的字段路径（例如 event.tags.ingest.processed_by）会按点字符 (.) 进行分割。然后，处理器使用得到的字段名遍历文档，直到找到值。在向文档写入值时，如果源中不存在其父字段，处理器将为缺失的字段创建嵌套对象。
POST /_ingest/pipeline/_simulate
{
"pipeline" : {
"description": "example pipeline",
"field_access_pattern": "classic",
"processors": [
{
"set" : {
"description" : "Copy the foo.bar field into the a.b.c.d field if it exists",
"copy_from" : "foo.bar",
"field" : "a.b.c.d",
"ignore_empty_value": true
}
}
]
},
"docs": [
{
"_index": "index",
"_id": "id",
"_source": {
"foo": {
"bar": "baz"
}
}
},
{
"_index": "index",
"_id": "id",
"_source": {
"foo.bar": "baz"
}
}
]
}
- 在管道中显式声明使用
classic访问模式。这是默认值。 - 我们正在从字段
foo.bar读取一个值。 - 我们正在将其值写入字段
a.b.c.d。 - 此文档在其结构中使用了嵌套的 json 对象。
- 此文档在其结构中使用了带点号的字段名。
{
"docs": [
{
"doc": {
"_id": "id",
"_index": "index",
"_version": "-3",
"_source": {
"foo": {
"bar": "baz"
},
"a": {
"b": {
"c": {
"d": "baz"
}
}
}
},
"_ingest": {
"timestamp": "2017-05-04T22:30:03.187Z"
}
}
},
{
"doc": {
"_id": "id",
"_index": "index",
"_version": "-3",
"_source": {
"foo.bar": "baz"
},
"_ingest": {
"timestamp": "2017-05-04T22:30:03.188Z"
}
}
}
]
}
- 定位到了第一个文档的
foo.bar字段，因为它使用了嵌套的 json。处理器会查找foo字段，然后查找bar字段。 - 来自
foo.bar字段的值被写入字段a.b.c.d处的嵌套 json 结构中。处理器会为路径中的每个字段创建对象。 - 第二个文档为
foo.bar使用了带点号的字段名。classic访问模式无法识别带点号的字段名，因此不会复制任何内容。
如果您要提取的文档包含带点号的字段名，若要使用 classic 访问模式读取它们，您必须使用 dot_expander 处理器。不过，这种方法并不总是合理的。请考虑以下文档
{
"event": {
"tags": {
"http.host": "localhost:9200",
"http.host.name": "localhost",
"http.host.port": 9200
}
}
}
如果使用 dot_expander 处理器处理 event.tags 字段，字段值将会发生冲突。http.host 字段不能同时是文本值和对象值。
flexible 访问模式允许 ingest 管道在不使用 dot_expander 处理器的同时访问嵌套的和带点号的字段名。此外，当将值写入不存在的字段时，任何缺失的父字段都会被连接到新键的前面。如果您的文档具有带点号的字段名，或者您倾向于以带点号的名称将缺失的字段写入文档，请使用 flexible 访问模式。
POST /_ingest/pipeline/_simulate
{
"pipeline" : {
"description": "example pipeline",
"field_access_pattern": "flexible",
"processors": [
{
"set" : {
"description" : "Copy the foo.bar field into the a.b.c.d field if it exists",
"copy_from" : "foo.bar",
"field" : "a.b.c.d",
"ignore_empty_value": true
}
}
]
},
"docs": [
{
"_index": "index",
"_id": "id",
"_source": {
"foo": {
"bar": "baz"
},
"a": {}
}
},
{
"_index": "index",
"_id": "id",
"_source": {
"foo.bar": "baz"
}
}
]
}
- 在管道中使用
flexible访问模式。 - 我们正在从字段
foo.bar读取一个值。 - 我们正在将其值写入字段
a.b.c.d。 - 第一个文档在其结构中使用了嵌套的 json 对象。
- 第一个文档在根级别有一个现有的
a字段。 - 第二个文档使用了一个带点号的字段名。
{
"docs": [
{
"doc": {
"_id": "id",
"_index": "index",
"_version": "-3",
"_source": {
"foo": {
"bar": "baz"
},
"a": {
"b.c.d": "baz"
}
},
"_ingest": {
"timestamp": "2017-05-04T22:30:03.187Z"
}
}
},
{
"doc": {
"_id": "id",
"_index": "index",
"_version": "-3",
"_source": {
"foo.bar": "baz",
"a.b.c.d": "baz"
},
"_ingest": {
"timestamp": "2017-05-04T22:30:03.188Z"
}
}
}
]
}
flexible访问模式支持嵌套对象字段。处理器会查找foo字段，然后查找bar字段。- 来自
foo.bar字段的值被写入字段a下方的带点号字段名b.c.d中。处理器将缺失的字段名连接在一起作为键的前缀。 flexible访问模式还支持带点号的字段名。处理器会查找名为foo的字段，在未找到后，会查找名为foo.bar的字段。- 来自
foo.bar字段的值被写入带点号的字段名a.b.c.d。由于这些字段在文档中尚不存在，它们被连接成一个带点号的字段名。
处理器可以通过名称访问以下元数据字段
_index_id_routing_dynamic_templates
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"set": {
"description": "Set '_routing' to 'geoip.country_iso_code' value",
"field": "_routing",
"value": "{{{geoip.country_iso_code}}}"
}
}
]
}
使用 Mustache 模板片段访问元数据字段值。例如，{{{_routing}}} 会检索文档的 routing（路由）值。
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"set": {
"description": "Use geo_point dynamic template for address field",
"field": "_dynamic_templates",
"value": {
"address": "geo_point"
}
}
}
]
}
上面的 set 处理器告知 ES：如果字段 address 尚未在索引的映射中定义，则为其使用名为 geo_point 的动态模板。如果已在 bulk 请求中定义了字段 address 的动态模板，此处理器会将其覆盖，但对 bulk 请求中定义的其他动态模板没有影响。
如果您自动生成文档 ID，则不能在处理器中使用 {{{_id}}}。Elasticsearch 会在数据提取后分配自动生成的 _id 值。
Ingest 处理器可以使用 _ingest 键来添加和访问 ingest 元数据。
与源字段和元数据字段不同，Elasticsearch 默认不会对 ingest 元数据字段建立索引。Elasticsearch 还允许使用以 _ingest 键开头的源字段。如果您的数据包含此类源字段，请使用 _source._ingest 来访问它们。
管道默认只创建 _ingest.timestamp ingest 元数据字段。此字段包含 Elasticsearch 收到文档索引请求时的时间戳。若要对 _ingest.timestamp 或其他 ingest 元数据字段建立索引，请使用 set 处理器。
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"set": {
"description": "Index the ingest timestamp as 'event.ingested'",
"field": "event.ingested",
"value": "{{{_ingest.timestamp}}}"
}
}
]
}
管道的处理器按顺序运行。默认情况下，当其中一个处理器失败或遇到错误时，管道处理会停止。
若要忽略处理器故障并运行管道中其余的处理器，请将 ignore_failure 设置为 true。
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"rename": {
"description": "Rename 'provider' to 'cloud.provider'",
"field": "provider",
"target_field": "cloud.provider",
"ignore_failure": true
}
}
]
}
使用 on_failure 参数指定在处理器故障后立即运行的处理器列表。如果指定了 on_failure，Elasticsearch 随后将运行管道中剩余的处理器，即使 on_failure 配置为空也是如此。
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"rename": {
"description": "Rename 'provider' to 'cloud.provider'",
"field": "provider",
"target_field": "cloud.provider",
"on_failure": [
{
"set": {
"description": "Set 'error.message'",
"field": "error.message",
"value": "Field 'provider' does not exist. Cannot rename to 'cloud.provider'",
"override": false
}
}
]
}
}
]
}
嵌套一个 on_failure 处理器列表以进行嵌套错误处理。
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"rename": {
"description": "Rename 'provider' to 'cloud.provider'",
"field": "provider",
"target_field": "cloud.provider",
"on_failure": [
{
"set": {
"description": "Set 'error.message'",
"field": "error.message",
"value": "Field 'provider' does not exist. Cannot rename to 'cloud.provider'",
"override": false,
"on_failure": [
{
"set": {
"description": "Set 'error.message.multi'",
"field": "error.message.multi",
"value": "Document encountered multiple ingest errors",
"override": true
}
}
]
}
}
]
}
}
]
}
您还可以为管道指定 on_failure。如果缺少 on_failure 值的处理器失败，Elasticsearch 将使用此管道级参数作为回退。Elasticsearch 不会尝试运行管道中剩余的处理器。
PUT _ingest/pipeline/my-pipeline
{
"processors": [ ... ],
"on_failure": [
{
"set": {
"description": "Index document to 'failed-<index>'",
"field": "_index",
"value": "failed-{{{ _index }}}"
}
}
]
}
文档元数据字段 on_failure_message、on_failure_processor_type、on_failure_processor_tag 和 on_failure_pipeline 中可能会提供有关管道故障的更多信息。这些字段只能在 on_failure 块内部访问。
以下示例使用元数据字段将有关管道故障的信息包含在文档中。
PUT _ingest/pipeline/my-pipeline
{
"processors": [ ... ],
"on_failure": [
{
"set": {
"description": "Record error information",
"field": "error_information",
"value": "Processor {{ _ingest.on_failure_processor_type }} with tag {{ _ingest.on_failure_processor_tag }} in pipeline {{ _ingest.on_failure_pipeline }} failed with message {{ _ingest.on_failure_message }}"
}
}
]
}
每个处理器都支持一个可选的 if 条件，写成 Painless 脚本。如果提供了该条件，则处理器仅在 if 条件为 true 时运行。
if 条件脚本在 Painless 的 ingest 处理器上下文中运行。在 if 条件中，ctx 值是只读的。
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"drop": {
"description": "Drop documents with 'network.name' of 'Guest'",
"if": "ctx?.network?.name == 'Guest'"
}
}
]
}
如果启用了 script.painless.regex.enabled 集群设置，您便可以在 if 条件脚本中使用正则表达式。有关支持的语法，请参阅 Painless 正则表达式。
如果可能，请避免使用正则表达式。复杂的正则表达式会降低索引速度。
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"set": {
"description": "If 'url.scheme' is 'http', set 'url.insecure' to true",
"if": "ctx.url?.scheme =~ /^http[^s]/",
"field": "url.insecure",
"value": true
}
}
]
}
您必须将 if 条件指定为单行上的有效 JSON。不过，您可以使用 Kibana 控制台的三引号语法来编写和调试较大的脚本。
如果可能，请避免使用复杂或高开销的 if 条件脚本。高开销的条件脚本会减慢索引速度。
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"drop": {
"description": "Drop documents that don't contain 'prod' tag",
"if": """
Collection tags = ctx.tags;
if(tags != null){
for (String tag : tags) {
if (tag.toLowerCase().contains('prod')) {
return false;
}
}
}
return true;
"""
}
}
]
}
您还可以将存储的脚本指定为 if 条件。
PUT _scripts/my-prod-tag-script
{
"script": {
"lang": "painless",
"source": """
Collection tags = ctx.tags;
if(tags != null){
for (String tag : tags) {
if (tag.toLowerCase().contains('prod')) {
return false;
}
}
}
return true;
"""
}
}
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"drop": {
"description": "Drop documents that don't contain 'prod' tag",
"if": { "id": "my-prod-tag-script" }
}
}
]
}
传入的文档通常包含对象字段。如果处理器脚本尝试访问其父对象不存在的字段，Elasticsearch 将返回 NullPointerException。为避免这些异常，请使用诸如 ?. 等空安全运算符，并编写具备空安全性的脚本。
例如，ctx.network?.name.equalsIgnoreCase('Guest') 不是空安全的。ctx.network?.name 可能会返回 null。请将脚本重写为 'Guest'.equalsIgnoreCase(ctx.network?.name)，它是空安全的，因为 Guest 始终是非空的。
如果您无法将脚本重写为空安全的，请包含显式的 null 检查。
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"drop": {
"description": "Drop documents that contain 'network.name' of 'Guest'",
"if": "ctx.network?.name != null && ctx.network.name.contains('Guest')"
}
}
]
}
将 if 条件与 pipeline 处理器结合使用，以便根据您的准则将其他管道应用于文档。您可以将此管道用作配置多个数据流或索引的索引模板中的默认管道。
PUT _ingest/pipeline/one-pipeline-to-rule-them-all
{
"processors": [
{
"pipeline": {
"description": "If 'service.name' is 'apache_httpd', use 'httpd_pipeline'",
"if": "ctx.service?.name == 'apache_httpd'",
"name": "httpd_pipeline"
}
},
{
"pipeline": {
"description": "If 'service.name' is 'syslog', use 'syslog_pipeline'",
"if": "ctx.service?.name == 'syslog'",
"name": "syslog_pipeline"
}
},
{
"fail": {
"description": "If 'service.name' is not 'apache_httpd' or 'syslog', return a failure message",
"if": "ctx.service?.name != 'apache_httpd' && ctx.service?.name != 'syslog'",
"message": "This pipeline requires service.name to be either `syslog` or `apache_httpd`"
}
}
]
}
使用 node stats API 获取全局以及每个管道的 ingest 统计信息。使用这些统计信息来确定哪些管道运行最频繁或花费最多的处理时间。
GET _nodes/stats/ingest?filter_path=nodes.*.ingest