ingest.asciidoc 25 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964
  1. [[ingest]]
  2. = Ingest pipelines
  3. Ingest pipelines let you perform common transformations on your data before
  4. indexing. For example, you can use pipelines to remove fields, extract values
  5. from text, and enrich your data.
  6. A pipeline consists of a series of configurable tasks called
  7. <<processors,processors>>. Each processor runs sequentially, making specific
  8. changes to incoming documents. After the processors have run, {es} adds the
  9. transformed documents to your data stream or index.
  10. image::images/ingest/ingest-process.svg[Ingest pipeline diagram,align="center"]
  11. You can create and manage ingest pipelines using {kib}'s **Ingest Node
  12. Pipelines** feature or the <<ingest-apis,ingest APIs>>. {es} stores pipelines in
  13. the <<cluster-state,cluster state>>.
  14. [discrete]
  15. [[ingest-prerequisites]]
  16. === Prerequisites
  17. * Nodes with the <<node-ingest-node,`ingest`>> node role handle pipeline
  18. processing. To use ingest pipelines, your cluster must have at least one node
  19. with the `ingest` role. For heavy ingest loads, we recommend creating
  20. <<node-ingest-node,dedicated ingest nodes>>.
  21. * If the {es} security features are enabled, you must have the `manage_pipeline`
  22. <<privileges-list-cluster,cluster privilege>> to manage ingest pipelines. To use
  23. {kib}'s **Ingest Node Pipelines** feature, you also need the
  24. `cluster:monitor/nodes/info` cluster privileges.
  25. * Pipelines including the `enrich` processor require additional setup. See
  26. <<ingest-enriching-data>>.
  27. [discrete]
  28. [[create-manage-ingest-pipelines]]
  29. === Create and manage pipelines
  30. In {kib}, open the main menu and click **Stack Management** > **Ingest Node
  31. Pipelines**. From the list view, you can:
  32. * View a list of your pipelines and drill down into details
  33. * Edit or clone existing pipelines
  34. * Delete pipelines
  35. To create a new pipeline, click **Create pipeline**. For an example tutorial,
  36. see <<common-log-format-example>>.
  37. [role="screenshot"]
  38. image::images/ingest/ingest-pipeline-list.png[Kibana's Ingest Node Pipelines list view,align="center"]
  39. You can also use the <<ingest-apis,ingest APIs>> to create and manage pipelines.
  40. The following <<put-pipeline-api,create pipeline API>> request creates
  41. a pipeline containing two <<set-processor,`set`>> processors followed by a
  42. <<lowercase-processor,`lowercase`>> processor. The processors run sequentially
  43. in the order specified.
  44. [source,console]
  45. ----
  46. PUT _ingest/pipeline/my-pipeline
  47. {
  48. "description": "My optional pipeline description",
  49. "processors": [
  50. {
  51. "set": {
  52. "description": "My optional processor description",
  53. "field": "my-long-field",
  54. "value": 10
  55. }
  56. },
  57. {
  58. "set": {
  59. "description": "Set 'my-boolean-field' to true",
  60. "field": "my-boolean-field",
  61. "value": true
  62. }
  63. },
  64. {
  65. "lowercase": {
  66. "field": "my-keyword-field"
  67. }
  68. }
  69. ]
  70. }
  71. ----
  72. // TESTSETUP
  73. [discrete]
  74. [[manage-pipeline-versions]]
  75. === Manage pipeline versions
  76. When you create or update a pipeline, you can specify an optional `version`
  77. integer. {es} doesn't use this `version` number internally, but you can use it
  78. to track changes to a pipeline.
  79. [source,console]
  80. ----
  81. PUT _ingest/pipeline/my-pipeline-id
  82. {
  83. "version": 1,
  84. "processors": [ ... ]
  85. }
  86. ----
  87. // TEST[s/\.\.\./{"lowercase": {"field":"my-keyword-field"}}/]
  88. To unset the `version` number using the API, replace or update the pipeline
  89. without specifying the `version` parameter.
  90. [discrete]
  91. [[test-pipeline]]
  92. === Test a pipeline
  93. Before using a pipeline in production, we recommend you test it using sample
  94. documents. When creating or editing a pipeline in {kib}, click **Add
  95. documents**. In the **Documents** tab, provide sample documents and click **Run
  96. the pipeline**.
  97. [role="screenshot"]
  98. image::images/ingest/test-a-pipeline.png[Test a pipeline in Kibana,align="center"]
  99. You can also test pipelines using the <<simulate-pipeline-api,simulate pipeline
  100. API>>. You can specify a configured pipeline in the request path. For example,
  101. the following request tests `my-pipeline`.
  102. [source,console]
  103. ----
  104. POST _ingest/pipeline/my-pipeline/_simulate
  105. {
  106. "docs": [
  107. {
  108. "_source": {
  109. "my-keyword-field": "FOO"
  110. }
  111. },
  112. {
  113. "_source": {
  114. "my-keyword-field": "BAR"
  115. }
  116. }
  117. ]
  118. }
  119. ----
  120. Alternatively, you can specify a pipeline and its processors in the request
  121. body.
  122. [source,console]
  123. ----
  124. POST _ingest/pipeline/_simulate
  125. {
  126. "pipeline": {
  127. "processors": [
  128. {
  129. "lowercase": {
  130. "field": "my-keyword-field"
  131. }
  132. }
  133. ]
  134. },
  135. "docs": [
  136. {
  137. "_source": {
  138. "my-keyword-field": "FOO"
  139. }
  140. },
  141. {
  142. "_source": {
  143. "my-keyword-field": "BAR"
  144. }
  145. }
  146. ]
  147. }
  148. ----
  149. The API returns transformed documents:
  150. [source,console-result]
  151. ----
  152. {
  153. "docs": [
  154. {
  155. "doc": {
  156. "_index": "_index",
  157. "_id": "_id",
  158. "_source": {
  159. "my-keyword-field": "foo"
  160. },
  161. "_ingest": {
  162. "timestamp": "2099-03-07T11:04:03.000Z"
  163. }
  164. }
  165. },
  166. {
  167. "doc": {
  168. "_index": "_index",
  169. "_id": "_id",
  170. "_source": {
  171. "my-keyword-field": "bar"
  172. },
  173. "_ingest": {
  174. "timestamp": "2099-03-07T11:04:04.000Z"
  175. }
  176. }
  177. }
  178. ]
  179. }
  180. ----
  181. // TESTRESPONSE[s/"2099-03-07T11:04:03.000Z"/$body.docs.0.doc._ingest.timestamp/]
  182. // TESTRESPONSE[s/"2099-03-07T11:04:04.000Z"/$body.docs.1.doc._ingest.timestamp/]
  183. [discrete]
  184. [[add-pipeline-to-indexing-request]]
  185. === Add a pipeline to an indexing request
  186. Use the `pipeline` query parameter to apply a pipeline to documents in
  187. <<docs-index_,individual>> or <<docs-bulk,bulk>> indexing requests.
  188. [source,console]
  189. ----
  190. POST my-data-stream/_doc?pipeline=my-pipeline
  191. {
  192. "@timestamp": "2099-03-07T11:04:05.000Z",
  193. "my-keyword-field": "foo"
  194. }
  195. PUT my-data-stream/_bulk?pipeline=my-pipeline
  196. { "create":{ } }
  197. { "@timestamp": "2099-03-07T11:04:06.000Z", "my-keyword-field": "foo" }
  198. { "create":{ } }
  199. { "@timestamp": "2099-03-07T11:04:07.000Z", "my-keyword-field": "bar" }
  200. ----
  201. // TEST[setup:my_data_stream]
  202. // TEST[teardown:data_stream_cleanup]
  203. You can also use the `pipeline` parameter with the <<docs-update-by-query,update
  204. by query>> or <<docs-reindex,reindex>> APIs.
  205. [source,console]
  206. ----
  207. POST my-data-stream/_update_by_query?pipeline=my-pipeline
  208. POST _reindex
  209. {
  210. "source": {
  211. "index": "my-data-stream"
  212. },
  213. "dest": {
  214. "index": "my-new-data-stream",
  215. "op_type": "create",
  216. "pipeline": "my-pipeline"
  217. }
  218. }
  219. ----
  220. // TEST[setup:my_data_stream]
  221. // TEST[teardown:data_stream_cleanup]
  222. [discrete]
  223. [[set-default-pipeline]]
  224. === Set a default pipeline
  225. Use the <<index-default-pipeline,`index.default_pipeline`>> index setting to set
  226. a default pipeline. {es} applies this pipeline if no `pipeline` parameter
  227. is specified.
  228. [discrete]
  229. [[set-final-pipeline]]
  230. === Set a final pipeline
  231. Use the <<index-final-pipeline,`index.final_pipeline`>> index setting to set a
  232. final pipeline. {es} applies this pipeline after the request or default
  233. pipeline, even if neither is specified.
  234. [discrete]
  235. [[pipelines-for-beats]]
  236. === Pipelines for {beats}
  237. To add an ingest pipeline to an Elastic Beat, specify the `pipeline`
  238. parameter under `output.elasticsearch` in `<BEAT_NAME>.yml`. For example,
  239. for {filebeat}, you'd specify `pipeline` in `filebeat.yml`.
  240. [source,yaml]
  241. ----
  242. output.elasticsearch:
  243. hosts: ["localhost:9200"]
  244. pipeline: my-pipeline
  245. ----
  246. [discrete]
  247. [[pipelines-for-fleet-elastic-agent]]
  248. === Pipelines for {fleet} and {agent}
  249. {fleet-guide}/index.html[{fleet}] automatically adds ingest pipelines for its
  250. integrations. {fleet} applies these pipelines using <<index-templates,index
  251. templates>> that include <<set-default-pipeline,pipeline index settings>>. {es}
  252. matches these templates to your {fleet} data streams based on the
  253. {fleet-guide}/data-streams.html#data-streams-naming-scheme[stream's naming
  254. scheme].
  255. WARNING: Do not change {fleet}'s ingest pipelines or use custom pipelines for
  256. your {fleet} integrations. Doing so can break your {fleet} data streams.
  257. {fleet} doesn't provide an ingest pipeline for the **Custom logs** integration.
  258. You can safely specify a pipeline for this integration in one of two ways: an
  259. <<pipeline-custom-logs-index-template,index template>> or a
  260. <<pipeline-custom-logs-configuration,custom configuration>>.
  261. [[pipeline-custom-logs-index-template]]
  262. **Option 1: Index template**
  263. // tag::create-name-custom-logs-pipeline[]
  264. . <<create-manage-ingest-pipelines,Create>> and <<test-pipeline,test>> your
  265. ingest pipeline. Name your pipeline `logs-<dataset-name>-default`. This makes
  266. tracking the pipeline for your integration easier.
  267. +
  268. --
  269. For example, the following request creates a pipeline for the `my-app` dataset.
  270. The pipeline's name is `logs-my_app-default`.
  271. [source,console]
  272. ----
  273. PUT _ingest/pipeline/logs-my_app-default
  274. {
  275. "description": "Pipeline for `my_app` dataset",
  276. "processors": [ ... ]
  277. }
  278. ----
  279. // TEST[s/\.\.\./{"lowercase": {"field":"my-keyword-field"}}/]
  280. --
  281. // end::create-name-custom-logs-pipeline[]
  282. . Create an <<index-templates,index template>> that includes your pipeline in
  283. the <<index-default-pipeline,`index.default_pipeline`>> or
  284. <<index-final-pipeline,`index.final_pipeline`>> index setting. Ensure the
  285. template is <<create-index-template,data stream enabled>>. The
  286. template's index pattern should match `logs-<dataset-name>-*`.
  287. +
  288. --
  289. You can create this template using {kib}'s <<manage-index-templates,**Index
  290. Management**>> feature or the <<indices-put-template,create index template
  291. API>>.
  292. For example, the following request creates a template matching `logs-my_app-*`.
  293. The template uses a component template that contains the
  294. `index.default_pipeline` index setting.
  295. [source,console]
  296. ----
  297. # Creates a component template for index settings
  298. PUT _component_template/logs-my_app-settings
  299. {
  300. "template": {
  301. "settings": {
  302. "index.default_pipeline": "logs-my_app-default",
  303. "index.lifecycle.name": "logs"
  304. }
  305. }
  306. }
  307. # Creates an index template matching `logs-my_app-*`
  308. PUT _index_template/logs-my_app-template
  309. {
  310. "index_patterns": ["logs-my_app-*"],
  311. "data_stream": { },
  312. "priority": 500,
  313. "composed_of": ["logs-my_app-settings", "logs-my_app-mappings"]
  314. }
  315. ----
  316. // TEST[continued]
  317. // TEST[s/, "logs-my_app-mappings"//]
  318. --
  319. // tag::name-custom-logs-dataset[]
  320. . When adding or editing your **Custom logs** integration in {fleet},
  321. click **Configure integration > Custom log file > Advanced options**.
  322. . In **Dataset name**, specify your dataset's name. {fleet} will add new data
  323. for the integration to the resulting `logs-<dataset-name>-default` data stream.
  324. +
  325. For example, if your dataset's name was `my_app`, {fleet} adds new data to the
  326. `logs-my_app-default` data stream.
  327. // end::name-custom-logs-dataset[]
  328. +
  329. [role="screenshot"]
  330. image::images/ingest/custom-logs.png[Set up custom log integration in Fleet,align="center"]
  331. . Use the <<indices-rollover-index,rollover API>> to roll over your data stream.
  332. This ensures {es} applies the index template and its pipeline settings to any
  333. new data for the integration.
  334. +
  335. --
  336. ////
  337. [source,console]
  338. ----
  339. PUT _data_stream/logs-my_app-default
  340. ----
  341. // TEST[continued]
  342. ////
  343. [source,console]
  344. ----
  345. POST logs-my_app-default/_rollover/
  346. ----
  347. // TEST[continued]
  348. ////
  349. [source,console]
  350. ----
  351. DELETE _data_stream/*
  352. DELETE _index_template/*
  353. ----
  354. // TEST[continued]
  355. ////
  356. --
  357. [[pipeline-custom-logs-configuration]]
  358. **Option 2: Custom configuration**
  359. include::ingest.asciidoc[tag=create-name-custom-logs-pipeline]
  360. include::ingest.asciidoc[tag=name-custom-logs-dataset]
  361. . In **Custom Configurations**, specify your pipeline in the `pipeline` policy
  362. setting.
  363. +
  364. [role="screenshot"]
  365. image::images/ingest/custom-logs-pipeline.png[Custom pipeline configuration for custom log integration,align="center"]
  366. **{agent} standalone**
  367. If you run {agent} standalone, you can apply pipelines using an
  368. <<index-templates,index template>> that includes the
  369. <<index-default-pipeline,`index.default_pipeline`>> or
  370. <<index-final-pipeline,`index.final_pipeline`>> index setting. Alternatively,
  371. you can specify the `pipeline` policy setting in your `elastic-agent.yml`
  372. configuration. See {fleet-guide}/run-elastic-agent-standalone.html[Run {agent}
  373. standalone].
  374. [discrete]
  375. [[access-source-fields]]
  376. === Access source fields in a processor
  377. Processors have read and write access to an incoming document's source fields.
  378. To access a field key in a processor, use its field name. The following `set`
  379. processor accesses `my-long-field`.
  380. [source,console]
  381. ----
  382. PUT _ingest/pipeline/my-pipeline
  383. {
  384. "processors": [
  385. {
  386. "set": {
  387. "field": "my-long-field",
  388. "value": 10
  389. }
  390. }
  391. ]
  392. }
  393. ----
  394. You can also prepend the `_source` prefix.
  395. [source,console]
  396. ----
  397. PUT _ingest/pipeline/my-pipeline
  398. {
  399. "processors": [
  400. {
  401. "set": {
  402. "field": "_source.my-long-field",
  403. "value": 10
  404. }
  405. }
  406. ]
  407. }
  408. ----
  409. Use dot notation to access object fields.
  410. IMPORTANT: If your document contains flattened objects, use the
  411. <<dot-expand-processor,`dot_expander`>> processor to expand them first. Other
  412. ingest processors cannot access flattened objects.
  413. [source,console]
  414. ----
  415. PUT _ingest/pipeline/my-pipeline
  416. {
  417. "processors": [
  418. {
  419. "dot_expander": {
  420. "description": "Expand 'my-object-field.my-property'",
  421. "field": "my-object-field.my-property"
  422. }
  423. },
  424. {
  425. "set": {
  426. "description": "Set 'my-object-field.my-property' to 10",
  427. "field": "my-object-field.my-property",
  428. "value": 10
  429. }
  430. }
  431. ]
  432. }
  433. ----
  434. [[template-snippets]]
  435. Several processor parameters support https://mustache.github.io[Mustache]
  436. template snippets. To access field values in a template snippet, enclose the
  437. field name in triple curly brackets:`{{{field-name}}}`. You can use template
  438. snippets to dynamically set field names.
  439. [source,console]
  440. ----
  441. PUT _ingest/pipeline/my-pipeline
  442. {
  443. "processors": [
  444. {
  445. "set": {
  446. "description": "Set dynamic '<service>' field to 'code' value",
  447. "field": "{{{service}}}",
  448. "value": "{{{code}}}"
  449. }
  450. }
  451. ]
  452. }
  453. ----
  454. [discrete]
  455. [[access-metadata-fields]]
  456. === Access metadata fields in a processor
  457. Processors can access the following metadata fields by name:
  458. * `_index`
  459. * `_id`
  460. * `_routing`
  461. * `_dynamic_templates`
  462. [source,console]
  463. ----
  464. PUT _ingest/pipeline/my-pipeline
  465. {
  466. "processors": [
  467. {
  468. "set": {
  469. "description": "Set '_routing' to 'geoip.country_iso_code' value",
  470. "field": "_routing",
  471. "value": "{{{geoip.country_iso_code}}}"
  472. }
  473. }
  474. ]
  475. }
  476. ----
  477. Use a Mustache template snippet to access metadata field values. For example,
  478. `{{{_routing}}}` retrieves a document's routing value.
  479. [source,console]
  480. ----
  481. PUT _ingest/pipeline/my-pipeline
  482. {
  483. "processors": [
  484. {
  485. "set": {
  486. "description": "Use geo_point dynamic template for address field",
  487. "field": "_dynamic_templates",
  488. "value": {
  489. "address": "geo_point"
  490. }
  491. }
  492. }
  493. ]
  494. }
  495. ----
  496. The set processor above tells ES to use the dynamic template named `geo_point`
  497. for the field `address` if this field is not defined in the mapping of the index
  498. yet. This processor overrides the dynamic template for the field `address` if
  499. already defined in the bulk request, but has no effect on other dynamic
  500. templates defined in the bulk request.
  501. WARNING: If you <<create-document-ids-automatically,automatically generate>>
  502. document IDs, you cannot use `{{{_id}}}` in a processor. {es} assigns
  503. auto-generated `_id` values after ingest.
  504. [discrete]
  505. [[access-ingest-metadata]]
  506. === Access ingest metadata in a processor
  507. Ingest processors can add and access ingest metadata using the `_ingest` key.
  508. Unlike source and metadata fields, {es} does not index ingest metadata fields by
  509. default. {es} also allows source fields that start with an `_ingest` key. If
  510. your data includes such source fields, use `_source._ingest` to access them.
  511. Pipelines only create the `_ingest.timestamp` ingest metadata field by default.
  512. This field contains a timestamp of when {es} received the document's indexing
  513. request. To index `_ingest.timestamp` or other ingest metadata fields, use the
  514. `set` processor.
  515. [source,console]
  516. ----
  517. PUT _ingest/pipeline/my-pipeline
  518. {
  519. "processors": [
  520. {
  521. "set": {
  522. "description": "Index the ingest timestamp as 'event.ingested'",
  523. "field": "event.ingested",
  524. "value": "{{{_ingest.timestamp}}}"
  525. }
  526. }
  527. ]
  528. }
  529. ----
  530. [discrete]
  531. [[handling-pipeline-failures]]
  532. === Handling pipeline failures
  533. A pipeline's processors run sequentially. By default, pipeline processing stops
  534. when one of these processors fails or encounters an error.
  535. To ignore a processor failure and run the pipeline's remaining processors, set
  536. `ignore_failure` to `true`.
  537. [source,console]
  538. ----
  539. PUT _ingest/pipeline/my-pipeline
  540. {
  541. "processors": [
  542. {
  543. "rename": {
  544. "description": "Rename 'provider' to 'cloud.provider'",
  545. "field": "provider",
  546. "target_field": "cloud.provider",
  547. "ignore_failure": true
  548. }
  549. }
  550. ]
  551. }
  552. ----
  553. Use the `on_failure` parameter to specify a list of processors to run
  554. immediately after a processor failure. If `on_failure` is specified, {es}
  555. afterward runs the pipeline's remaining processors, even if the `on_failure`
  556. configuration is empty.
  557. [source,console]
  558. ----
  559. PUT _ingest/pipeline/my-pipeline
  560. {
  561. "processors": [
  562. {
  563. "rename": {
  564. "description": "Rename 'provider' to 'cloud.provider'",
  565. "field": "provider",
  566. "target_field": "cloud.provider",
  567. "on_failure": [
  568. {
  569. "set": {
  570. "description": "Set 'error.message'",
  571. "field": "error.message",
  572. "value": "Field 'provider' does not exist. Cannot rename to 'cloud.provider'",
  573. "override": false
  574. }
  575. }
  576. ]
  577. }
  578. }
  579. ]
  580. }
  581. ----
  582. Nest a list of `on_failure` processors for nested error handling.
  583. [source,console]
  584. ----
  585. PUT _ingest/pipeline/my-pipeline
  586. {
  587. "processors": [
  588. {
  589. "rename": {
  590. "description": "Rename 'provider' to 'cloud.provider'",
  591. "field": "provider",
  592. "target_field": "cloud.provider",
  593. "on_failure": [
  594. {
  595. "set": {
  596. "description": "Set 'error.message'",
  597. "field": "error.message",
  598. "value": "Field 'provider' does not exist. Cannot rename to 'cloud.provider'",
  599. "override": false,
  600. "on_failure": [
  601. {
  602. "set": {
  603. "description": "Set 'error.message.multi'",
  604. "field": "error.message.multi",
  605. "value": "Document encountered multiple ingest errors",
  606. "override": true
  607. }
  608. }
  609. ]
  610. }
  611. }
  612. ]
  613. }
  614. }
  615. ]
  616. }
  617. ----
  618. You can also specify `on_failure` for a pipeline. If a processor without an
  619. `on_failure` value fails, {es} uses this pipeline-level parameter as a fallback.
  620. {es} will not attempt to run the pipeline's remaining processors.
  621. [source,console]
  622. ----
  623. PUT _ingest/pipeline/my-pipeline
  624. {
  625. "processors": [ ... ],
  626. "on_failure": [
  627. {
  628. "set": {
  629. "description": "Index document to 'failed-<index>'",
  630. "field": "_index",
  631. "value": "failed-{{{ _index }}}"
  632. }
  633. }
  634. ]
  635. }
  636. ----
  637. // TEST[s/\.\.\./{"lowercase": {"field":"my-keyword-field"}}/]
  638. Additional information about the pipeline failure may be available in the
  639. document metadata fields `on_failure_message`, `on_failure_processor_type`,
  640. `on_failure_processor_tag`, and `on_failure_pipeline`. These fields are
  641. accessible only from within an `on_failure` block.
  642. The following example uses the metadata fields to include information about
  643. pipeline failures in documents.
  644. [source,console]
  645. ----
  646. PUT _ingest/pipeline/my-pipeline
  647. {
  648. "processors": [ ... ],
  649. "on_failure": [
  650. {
  651. "set": {
  652. "description": "Record error information",
  653. "field": "error_information",
  654. "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 }}"
  655. }
  656. }
  657. ]
  658. }
  659. ----
  660. // TEST[s/\.\.\./{"lowercase": {"field":"my-keyword-field"}}/]
  661. [discrete]
  662. [[conditionally-run-processor]]
  663. === Conditionally run a processor
  664. Each processor supports an optional `if` condition, written as a
  665. {painless}/painless-guide.html[Painless script]. If provided, the processor only
  666. runs when the `if` condition is `true`.
  667. IMPORTANT: `if` condition scripts run in Painless's
  668. {painless}/painless-ingest-processor-context.html[ingest processor context]. In
  669. `if` conditions, `ctx` values are read-only.
  670. [source,console]
  671. ----
  672. PUT _ingest/pipeline/my-pipeline
  673. {
  674. "processors": [
  675. {
  676. "drop": {
  677. "description": "Drop documents with 'network.name' of 'Guest'",
  678. "if": "ctx?.network?.name == 'Guest'"
  679. }
  680. }
  681. ]
  682. }
  683. ----
  684. If the <<script-painless-regex-enabled,`script.painless.regex.enabled`>> cluster
  685. setting is enabled, you can use regular expressions in your `if` condition
  686. scripts. For supported syntax, see {painless}/painless-regexes.html[Painless
  687. regular expressions].
  688. TIP: If possible, avoid using regular expressions. Expensive regular expressions
  689. can slow indexing speeds.
  690. [source,console]
  691. ----
  692. PUT _ingest/pipeline/my-pipeline
  693. {
  694. "processors": [
  695. {
  696. "set": {
  697. "description": "If 'url.scheme' is 'http', set 'url.insecure' to true",
  698. "if": "ctx.url?.scheme =~ /^http[^s]/",
  699. "field": "url.insecure",
  700. "value": true
  701. }
  702. }
  703. ]
  704. }
  705. ----
  706. You must specify `if` conditions as valid JSON on a single line. However, you
  707. can use the {kibana-ref}/console-kibana.html#configuring-console[{kib}
  708. console]'s triple quote syntax to write and debug larger scripts.
  709. TIP: If possible, avoid using complex or expensive `if` condition scripts.
  710. Expensive condition scripts can slow indexing speeds.
  711. [source,console]
  712. ----
  713. PUT _ingest/pipeline/my-pipeline
  714. {
  715. "processors": [
  716. {
  717. "drop": {
  718. "description": "Drop documents that don't contain 'prod' tag",
  719. "if": """
  720. Collection tags = ctx.tags;
  721. if(tags != null){
  722. for (String tag : tags) {
  723. if (tag.toLowerCase().contains('prod')) {
  724. return false;
  725. }
  726. }
  727. }
  728. return true;
  729. """
  730. }
  731. }
  732. ]
  733. }
  734. ----
  735. You can also specify a <<modules-scripting-stored-scripts,stored script>> as the
  736. `if` condition.
  737. [source,console]
  738. ----
  739. PUT _scripts/my-stored-script
  740. {
  741. "script": {
  742. "lang": "painless",
  743. "source": """
  744. Collection tags = ctx.tags;
  745. if(tags != null){
  746. for (String tag : tags) {
  747. if (tag.toLowerCase().contains('prod')) {
  748. return false;
  749. }
  750. }
  751. }
  752. return true;
  753. """
  754. }
  755. }
  756. PUT _ingest/pipeline/my-pipeline
  757. {
  758. "processors": [
  759. {
  760. "drop": {
  761. "description": "If 'url.scheme' is 'http', set 'url.insecure' to true",
  762. "if": { "id": "my-stored-script" }
  763. }
  764. }
  765. ]
  766. }
  767. ----
  768. Incoming documents often contain object fields. If a processor script attempts
  769. to access a field whose parent object does not exist, {es} returns a
  770. `NullPointerException`. To avoid these exceptions, use
  771. {painless}/painless-operators-reference.html#null-safe-operator[null safe
  772. operators], such as `?.`, and write your scripts to be null safe.
  773. For example, `ctx.network?.name.equalsIgnoreCase('Guest')` is not null safe.
  774. `ctx.network?.name` can return null. Rewrite the script as
  775. `'Guest'.equalsIgnoreCase(ctx.network?.name)`, which is null safe because
  776. `Guest` is always non-null.
  777. If you can't rewrite a script to be null safe, include an explicit null check.
  778. [source,console]
  779. ----
  780. PUT _ingest/pipeline/my-pipeline
  781. {
  782. "processors": [
  783. {
  784. "drop": {
  785. "description": "Drop documents that contain 'network.name' of 'Guest'",
  786. "if": "ctx.network?.name != null && ctx.network.name.contains('Guest')"
  787. }
  788. }
  789. ]
  790. }
  791. ----
  792. [discrete]
  793. [[conditionally-apply-pipelines]]
  794. === Conditionally apply pipelines
  795. Combine an `if` condition with the <<pipeline-processor,`pipeline`>> processor
  796. to apply other pipelines to documents based on your criteria. You can use this
  797. pipeline as the <<set-default-pipeline,default pipeline>> in an
  798. <<index-templates,index template>> used to configure multiple data streams or
  799. indices.
  800. [source,console]
  801. ----
  802. PUT _ingest/pipeline/one-pipeline-to-rule-them-all
  803. {
  804. "processors": [
  805. {
  806. "pipeline": {
  807. "description": "If 'service.name' is 'apache_httpd', use 'httpd_pipeline'",
  808. "if": "ctx.service?.name == 'apache_httpd'",
  809. "name": "httpd_pipeline"
  810. }
  811. },
  812. {
  813. "pipeline": {
  814. "description": "If 'service.name' is 'syslog', use 'syslog_pipeline'",
  815. "if": "ctx.service?.name == 'syslog'",
  816. "name": "syslog_pipeline"
  817. }
  818. },
  819. {
  820. "fail": {
  821. "description": "If 'service.name' is not 'apache_httpd' or 'syslog', return a failure message",
  822. "if": "ctx.service?.name != 'apache_httpd' && ctx.service?.name != 'syslog'",
  823. "message": "This pipeline requires service.name to be either `syslog` or `apache_httpd`"
  824. }
  825. }
  826. ]
  827. }
  828. ----
  829. [discrete]
  830. [[get-pipeline-usage-stats]]
  831. === Get pipeline usage statistics
  832. Use the <<cluster-nodes-stats,node stats>> API to get global and per-pipeline
  833. ingest statistics. Use these stats to determine which pipelines run most
  834. frequently or spend the most time processing.
  835. [source,console]
  836. ----
  837. GET _nodes/stats/ingest?filter_path=nodes.*.ingest
  838. ----
  839. include::ingest/common-log-format-example.asciidoc[]
  840. include::ingest/enrich.asciidoc[]
  841. include::ingest/processors.asciidoc[]