Using custom ingest pipeline for Filebeat Netflow

Elastic Stack Beats
1

Hi,

I'm using Filebeat v8.8.0, and am trying to ingest Netflow data through a custom ingest pipeline to perform IP geo-location using a database that is not Maxmind. Currently, my Netflow data is enriched using the Maxmind database (I didn't specify any settings in Filebeat or ES, other than uploading the Maxmind databases to the appropriate folders for ES).

I have already created the pipeline, named new_filebeat_pipeline. However, I'm not sure where I should specify the pipeline in my Filebeat configuration files. I have tried adding pipeline: "new_filebeat_pipeline" at the following 3 places (separately), but my data is always enriched with Maxmind data, not my custom database.

Setting 1: In filebeat.yml

output.elasticsearch:
  ...
  pipeline: "new_filebeat_pipeline"

Setting 2: In filebeat.yml

filebeat.inputs:
- type: netflow
   enabled: true
   pipeline: "new_filebeat_pipeline"

Setting 3: In /etc/filebeat/modules.d/netflow.yml

- module: netflow
  log:
    enabled: true
    var: 
      ...
      input.pipeline: "new_filebeat_pipeline" # also tried 'pipeline: "new_filebeat_pipeline"'

I have always used netflow.yml to configure my Netflow data, not filebeat.inputs.

I have tested the pipeline using an existing document (already containing the Maxmind enrichment), and it was able to generate the fields expected from the new database, so the pipeline should be working correctly. But for some reason, my Netflow data is still going through the "default" pipeline, not my custom pipeline.

How should I configure Filebeat to do this?

Thank you.

2

ETA: I read here that "This module wraps the netflow input to enrich the flow records with geolocation information about the IP endpoints by using an Elasticsearch ingest pipeline." So I guessed that if I use the netflow module, it will always go to the default ingest pipeline. So I disabled it, and used filebeat.inputs instead, and I can see the fields added using the custom ingest pipeline now.

Although I'm facing another problem now, which is I get the error message Could not perform enrichment, enrich coordination queue at capacity [1024/1024].