# Multiple ingest pipelines - docs going to the wrong pipeline

**URL:** <https://discuss.elastic.co/t/multiple-ingest-pipelines-docs-going-to-the-wrong-pipeline/80339>\
**Category:** Beats\
**Tags:** filebeat\
**Created:** [March 28, 2017, 5:43pm UTC](https://discuss.elastic.co/t/multiple-ingest-pipelines-docs-going-to-the-wrong-pipeline/80339 "2017-03-28T17:43:38Z")\
**Posts on this page:** 6\
**Page:** 1

<div class="post-metadata">

**Author:** ![jbarlow](https://avatars.discourse-cdn.com/v4/letter/j/94ad74/32.png) [@jbarlow](https://discuss.elastic.co/u/jbarlow)\
**Post date:** [March 28, 2017, 5:43pm UTC](https://discuss.elastic.co/t/multiple-ingest-pipelines-docs-going-to-the-wrong-pipeline/80339/1 "2017-03-28T17:43:38Z")

</div>

Hello,

I've got three ES master/data nodes, and one ingest node running kibana. All servers in the environment are running filebeat for log shipping. I'm seeing a lot of pipeline errors in the elasticsearch logs about documents that shouldn't have been tagged with the pipeline listed in the errors, so then the pattern matching fails.

I was seeing similar issues last week running 5.2.2 for ES/Kibana/Filebeat. Troubleshooting today and I ended up upgrading all components to 5.3.

It appears that all the documents are being tagged with a particular pipeline, ignoring the conditionals.

Am I configuring filebeat incorrectly? I'm tagging each prospector, sending docs with those tags to an index, and same for the pipeline.

My filebeat.yml:  
#=========================== Filebeat prospectors =============================  
filebeat.prospectors:  
- input\_type: log  
paths: ["/var/log/syslog"]  
tags: ["syslog"]  
exclude\_lines: ["salt-minion","salt-master"]

```
- input_type: log
  paths: ["/var/log/kibana/kibana.log"]
  tags: ["kibana"]
  json.message_key: message
  json.keys_under_root: false
  multiline.pattern: '^\s'
  multiline.match: after
  document_type: kibana-logs

- input_type: log
  tags: ["salt"]
  paths:
    - /var/log/salt/master
    - /var/log/salt/api
    - /var/log/salt/minion

- input_type: log
  tags: ["containers"]
  paths: ["/var/log/containers/*.log"]
  symlinks: true
  json.message_key: log
  json.keys_under_root: true
  multiline.pattern: '^\s'
  multiline.match: after
  document_type: kube-logs

- input_type: log
  tags: ["kube"]
  paths:
    - "/var/log/kube*.log"
    - "/var/log/etcd.log"
- input_type: log
  tags: ["haproxy"]
  paths: ["/var/log/haproxy.log"]

#================================ General =====================================
processors:
- drop_fields:
    fields: ["offset", "beat.name", "beat.version"]
#================================ Outputs =====================================

#-------------------------- Elasticsearch output ------------------------------
output.elasticsearch:
  # Array of hosts to connect to.
  hosts: ["logesnode01:9200", "logesnode02:9200", "logesnode03:9200"]
  index: "logs"
  indices:
    - index: "syslog"
      when.contains:
        tags: "syslog"
    - index: "kibana"
      when.contains:
        tags: "kibana"
    - index: "salt"
      when.contains:
        tags: "salt"
    - index: "kube"
      when.contains:
        tags: "kube"
    - index: "containers"
      when.contains:
        tags: "containers"
    - index: "haproxy"
      when.contains:
        tags: "haproxy"
  pipelines:
    - pipeline: "kube-pipeline"
      when.contains:
        tags: "containers"
    - pipeline: "haproxy-pipeline"
      when.contains:
        tags: "haproxy"
    - pipeline: "salt-pipeline"
      when.contains:
        tags: "salt"

```

Errors:  
[2017-03-28T17:30:49,194][DEBUG][o.e.a.b.TransportBulkAction] [logesclient01] failed to execute pipeline [salt-pipeline] for document [salt/log/null]  
org.elasticsearch.ElasticsearchException: java.lang.IllegalArgumentException: java.lang.IllegalArgumentException: Provided Grok expressions do not match field value: [linux-image-4.4.0-64-generic]  
at org.elasticsearch.ingest.CompoundProcessor.newCompoundProcessorException(CompoundProcessor.java:156) ~[elasticsearch-5.3.0.jar:5.3.0]  
[2017-03-28T17:30:49,195][DEBUG][o.e.a.b.TransportBulkAction] [logesclient01] failed to execute pipeline [salt-pipeline] for document [s alt/log/null]  
org.elasticsearch.ElasticsearchException: java.lang.IllegalArgumentException: java.lang.IllegalArgumentException: Provided Grok expressi  
ons do not match field value: ['filebeat' changed from 'absent' to '5.3.0']

```
salt-pipeline:
curl localhost:9200/_ingest/pipeline/salt-pipeline?pretty
{ "salt-pipeline" : {
    "description" : "Salt pipleine",
    "processors" : [ {
        "grok" : {
          "field" : "message",
          "patterns" : [
            "%{TIMESTAMP_ISO8601:timestamp} \\[%{GREEDYDATA:function}\\]\\[%{GREEDYDATA:level}\\]\\[%{GREEDYDATA:process}\\] %{GREEDYDATA:event}"
          ] } },
      { "remove" : { "field" : "message" } } ] } }

```

All the pipelines listed in the filebeat.yml exist.

---

<div class="post-metadata">

**Author:** ![steffens](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/steffens/32/79630_2.png) [@steffens](https://discuss.elastic.co/u/steffens)\
**Post date:** [March 29, 2017, 9:02am UTC](https://discuss.elastic.co/t/multiple-ingest-pipelines-docs-going-to-the-wrong-pipeline/80339/2 "2017-03-29T09:02:02Z")

</div>

please use the `</>` button to format config files and log output!

you're config looks somewhat overcomplicated to me. I'd use fields for setting pipeline and index name (btw. filebeat 5.3 adds ['pipeline'-setting](https://www.elastic.co/guide/en/beats/filebeat/current/configuration-filebeat-options.html#_pipeline) in the prospector configuration)

e.g.

```auto
filebeat.prospectors:
- input_type: log
  paths: ["/var/log/syslog"]
  fields.kind: "syslog" # <- used to configure the index (can be used for filtering in ES as well)
  exclude_lines: ["salt-minion","salt-master"]
  pipeline: "salt-pipeline" # <- configure ES pipeline in prospector
- ...

output.elasticsearch:
  hosts: ["logesnode01:9200", "logesnode02:9200", "logesnode03:9200"]
  index: "%{fields.kind:logs}" # <- use 'fields.kind' as index name, with default value 'logs'

```

btw. one can change the index name in ES ingest node as well.

The error seems to be when evaluating the grok expression. That is, the event does not match you're grok pattern. Using ingest node error handling (e.g.) `on_failure` at top-level you can mark/process failed events (e.g. send to another dead-letter-queue-index) for later introspection. e.g. see [mysql filebeat module](https://github.com/elastic/beats/blob/master/filebeat/module/mysql/slowlog/ingest/pipeline.json)

---

<div class="post-metadata">

**Author:** ![jbarlow](https://avatars.discourse-cdn.com/v4/letter/j/94ad74/32.png) [@jbarlow](https://discuss.elastic.co/u/jbarlow)\
**Post date:** [March 29, 2017, 4:46pm UTC](https://discuss.elastic.co/t/multiple-ingest-pipelines-docs-going-to-the-wrong-pipeline/80339/3 "2017-03-29T16:46:46Z")

</div>

Thanks for the suggestions! That does look much simpler, I like it.

I have implemented the suggestions, but now the filebeat service will not start.

Error:  
2017-03-29T16:21:57Z INFO Setup Beat: filebeat; Version: 5.3.0  
2017-03-29T16:21:57Z ERR failed to initialize elasticsearch plugin as output: unsupported format expression "fields.type" in output.elasticsearch.index  
2017-03-29T16:21:57Z CRIT Exiting: **error initializing publisher: unsupported format expression "fields.type" in output.elasticsearch.index**

Config:  
#=========================== Filebeat prospectors =============================  
filebeat.prospectors:  
- input\_type: log  
paths: ["/var/log/syslog"]  
fields.type: "syslog"  
exclude\_lines: ["salt-minion","salt-master"]  
pipeline: "salt-pipeline"

```
- input_type: log
  fields.type: "salt"
  paths:
    - /var/log/salt/master
    - /var/log/salt/api
    - /var/log/salt/minion
  pipeline: "salt-pipeline"
- . . .

#-------------------------- Elasticsearch output ------------------------------
output.elasticsearch:
  # Array of hosts to connect to.
  hosts: ["logesnode01:9200", "logesnode02:9200", "logesnode03:9200"]
  index: "%{fields.type:logs}"
#================================ Logging =====================================

```

I also tried the following, thinking it might have been complaining about the default:  
`index: "%{fields.type}"`

Any suggestions? Anything else I can add?

---

<div class="post-metadata">

**Author:** ![steffens](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/steffens/32/79630_2.png) [@steffens](https://discuss.elastic.co/u/steffens)\
**Post date:** [March 29, 2017, 5:05pm UTC](https://discuss.elastic.co/t/multiple-ingest-pipelines-docs-going-to-the-wrong-pipeline/80339/4 "2017-03-29T17:05:47Z")

</div>

ah, my fault. [Documentation](https://www.elastic.co/guide/en/beats/libbeat/current/config-file-format-type.html#_format_string_sprintf) says Format Strings requires `[]` in order to access event fields. That is, this should work: `%{[fields.type]:logs}`

---

<div class="post-metadata">

**Author:** ![jbarlow](https://avatars.discourse-cdn.com/v4/letter/j/94ad74/32.png) [@jbarlow](https://discuss.elastic.co/u/jbarlow)\
**Post date:** [March 29, 2017, 5:06pm UTC](https://discuss.elastic.co/t/multiple-ingest-pipelines-docs-going-to-the-wrong-pipeline/80339/5 "2017-03-29T17:06:37Z")

</div>

After looking in formatevents.go, it looks like it needed brackets around the fields.type:  
index: "%{[fields.kind]:logs}"  
After applying this change, filebeat is starting!

Thanks for the direction.

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [April 26, 2017, 5:06pm UTC](https://discuss.elastic.co/t/multiple-ingest-pipelines-docs-going-to-the-wrong-pipeline/80339/6 "2017-04-26T17:06:40Z")

</div>

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.