# Custom Log integration multiple pipelines **URL:** \ **Category:** Elastic Agent\ **Tags:** ingest-pipeline, integrations\ **Created:** [January 4, 2023, 4:16pm UTC](https://discuss.elastic.co/t/custom-log-integration-multiple-pipelines/322485 "2023-01-04T16:16:16Z")\ **Posts on this page:** 9\ **Page:** 1
**Author:** ![searchtastic](https://avatars.discourse-cdn.com/v4/letter/s/d6d6ee/32.png) [@searchtastic](https://discuss.elastic.co/u/searchtastic)\ **Post date:** [January 4, 2023, 4:16pm UTC](https://discuss.elastic.co/t/custom-log-integration-multiple-pipelines/322485/1 "2023-01-04T16:16:16Z")
I have a custom integration setup with Elastic agent. At the moment I have a directory with several log files containing quite different data. I can't use the same pipeline so I would like to reference another pipeline. I am happy for all of these to share the same data stream, even if some fields will be different. At the moment in the custom log integration I simply chose `pipeline: logs-ams-worker` This processes the worker logs and all is well. I also need to process a different log file with a different pipeline e.g logs-ams-node. Is it possible in the custom configuration where I have my pipeline declared to be able to have a condition with a distributor or an if statement or some other method. ```auto if doc['source'].value == 'logs-ams-worker.log' else if doc['source'].value == 'logs-ams-node.log' else if doc['source'].value == 'logs-ams-other.log' ``` ---
**Author:** ![stephenb](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/stephenb/32/40856_2.png) [@stephenb](https://discuss.elastic.co/u/stephenb)\ **Post date:** [January 4, 2023, 6:27pm UTC](https://discuss.elastic.co/t/custom-log-integration-multiple-pipelines/322485/2 "2023-01-04T18:27:36Z")
Hi @searchtastic Typically you would create a top-level pipeline that then calls the sub-pipelines... See [here](https://www.elastic.co/guide/en/elasticsearch/reference/current/ingest.html#conditionally-apply-pipelines) This is a powerful way to control pipeline flow and build component pipelines that can be reused. ---
**Author:** ![searchtastic](https://avatars.discourse-cdn.com/v4/letter/s/d6d6ee/32.png) [@searchtastic](https://discuss.elastic.co/u/searchtastic)\ **Post date:** [January 6, 2023, 12:51pm UTC](https://discuss.elastic.co/t/custom-log-integration-multiple-pipelines/322485/3 "2023-01-06T12:51:46Z")
@stephenb Thanks for the help. So I guess I could use something like the below code, would that work on file /opt/ams/logs/worker\_2.01.log? ```auto if":"ctx.log.file.path.contains('worker')", "name":"logs-ams-worker" if":"ctx.log.file.path.contains('node')", "name":"logs-ams-node" ``` I then need to figure out how to make just two of the four pipelines multiline, ideas welcome. Thanks again. ---
**Author:** ![stephenb](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/stephenb/32/40856_2.png) [@stephenb](https://discuss.elastic.co/u/stephenb)\ **Post date:** [January 6, 2023, 3:33pm UTC](https://discuss.elastic.co/t/custom-log-integration-multiple-pipelines/322485/4 "2023-01-06T15:33:23Z")
You know since the sorting is path based you could just create 2 or N custom Logs integrations with the paths / patterns and then just call the exact pipeline... For Multiline that happens **before** the pipeline you need to put it in the integration using the multiline syntax (actually the legacy syntax) see [here](https://www.elastic.co/guide/en/beats/filebeat/current/multiline-examples.html)... Make sure to use the Log input syntax Using log input: ```auto multiline.type: pattern multiline.pattern: '^[[:space:]]+(at|\.{3})[[:space:]]+\b|^Caused by:' multiline.negate: false multiline.match: after ``` Sample ![Screen Shot 2023-01-06 at 7.32.54 AM](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ad93b96a6f40b3fa31d12b128ba5e48460cc1aa.png) ProTip ... Multiline can be a be challenging, I just install a quick filebeat with log input to test...but that is just me. [Here](https://medium.zenika.com/how-to-integrate-custom-logs-with-elastic-agent-7f80aef5aec7) is a really nice article on all this ... we need to make our docs this good 🙂 ---
**Author:** ![searchtastic](https://avatars.discourse-cdn.com/v4/letter/s/d6d6ee/32.png) [@searchtastic](https://discuss.elastic.co/u/searchtastic)\ **Post date:** [January 6, 2023, 6:31pm UTC](https://discuss.elastic.co/t/custom-log-integration-multiple-pipelines/322485/5 "2023-01-06T18:31:12Z")
> [@stephenb](#): > > Typically you would create a top-level pipeline that then calls the sub-pipelines I really liked your idea of using a single pipeline in the integration that called a sub-pipeline. Although it doesn't seem to be working for me. Do you know if it's possible to post the log.file.path within the POST command to test the pipeline. ```auto POST _ingest/pipeline/worker/_simulate { "docs": [ { "_source": { "ctx.log.file.path": "/opt/logs/worker.debug.log", "message": "2022-01-10 15:47:54,757 INFO supervisord worker process" } } ] } ``` so that I have a way of testing if the pipeline is called. ```auto "script_stack" : [ "ctx.log.file.path.contains('worker')", " ^---- HERE" ], "script" : "ctx.log.file.path.contains('worker')", "lang" : "painless", "position" : { "offset" : 7, "start" : 0, "end" : 45 }, "caused_by" : { "type" : "null_pointer_exception", "reason" : "cannot access method/field [file] from a null def reference" ``` Thanks again. ---
**Author:** ![stephenb](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/stephenb/32/40856_2.png) [@stephenb](https://discuss.elastic.co/u/stephenb)\ **Post date:** [January 6, 2023, 6:43pm UTC](https://discuss.elastic.co/t/custom-log-integration-multiple-pipelines/322485/6 "2023-01-06T18:43:10Z")
Helps if you actually show the pipeline not just the error but I suspect you need null safety checks... Right above the doc I linked > Incoming documents often contain object fields. If a processor script attempts to access a field whose parent object does not exist, Elasticsearch returns a `NullPointerException`. To avoid these exceptions, use [null safe operators](https://www.elastic.co/guide/en/elasticsearch/painless/8.5/painless-operators-reference.html#null-safe-operator), such as `?.`, and write your scripts to be null safe. > > For example, `ctx.network?.name.equalsIgnoreCase('Guest')` is not null safe. `ctx.network?.name` can return null. Rewrite the script as `'Guest'.equalsIgnoreCase(ctx.network?.name)`, which is null safe because `Guest` is always non-null. > > If you can’t rewrite a script to be null safe, include an explicit null check. ```auto 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')" } } ] } ``` Ohh and this is not correct `ctx` means context of the document in the painless script in the processor.... you pre-pended it to the field in the conditional statements... don't prepend it in the simulate etc. > `if` condition scripts run in Painless’s [ingest processor context](https://www.elastic.co/guide/en/elasticsearch/painless/8.5/painless-ingest-processor-context.html). In `if` conditions, `ctx` values are read-only. ```auto "_source": { "ctx.log.file.path": "/opt/logs/worker.debug.log",