Override pipeline on modules not working

Elastic Stack Beats
1

Hello

I've got the following flow

GCP PubSub => Filebeat (9.1.3) => Elasticsearch

In the filebeat source, i'm sending postgresql module to GCP pubsub.
The flow is working fine i'm ingesting logs in elasticsearch, but I wanted to parse better the logs. Currently the message contains all information and it do not populate other fields like postgresql.log.timestamp, or postgresql.log.database

I created a simple pipeline and added it in the filebeat configuration to ensure it's working first

// GET _ingest/pipeline/filebeat-postgresql-standard-log-pipeline
{
  "filebeat-postgresql-standard-log-pipeline": {
    "description": "Pipeline for parsing PostgreSQL logs using the following log_line_prefix parameter: %t [%p]: [%l-1] user=%u,db=%d,client=%h",
    "on_failure": [
      {
        "set": {
          "field": "error.message",
          "value": "{{ _ingest.on_failure_message }}"
        }
      }
    ],
    "processors": [
      {
        "set": {
          "if": "ctx?.service?.type == 'postgresql'",
          "field": "postgresql.parsing",
          "value": "working"
        }
      }
    ]
  }
}

Here is my filebeat 9.1.3 configuration

filebeat.inputs:
  - type: gcp-pubsub
    project_id: "<project_id>"
    topic: "<topic_name>"
    subscription.name: "<subscription_name>"
    subscription.create: false
    credentials_file: "/usr/share/filebeat/gcp-pubsub-credentials.json"
    processors:
      - decode_json_fields:
          fields: [ "message" ]
          target: ""
          overwrite_keys: true
          add_error_key: true
          expand_keys: true

setup.ilm.enabled: false
setup.template.enabled: false

output.elasticsearch:
  hosts: ["http://elasticsearch:9200"]
  pipelines:
    - pipeline: "filebeat-postgresql-standard-log-pipeline"
      when.contains:
        service.type: "postgresql"
  indices:
    - index: "filebeat-%{[agent.version]}"
      when.contains:
        agent.type: "filebeat"

I've simulated the ingestion in POST _ingest/pipeline/filebeat-postgresql-standard-log-pipeline/_simulate and the pipeline is working fine.
But when I ingest real logs, it does not work.

Here the log I'm ingesting in the pubsub topic

{
  "@timestamp": "2025-09-25T14:03:22.296Z",
  "event": {
    "module": "postgresql",
    "dataset": "postgresql.log"
  },
  "input": {
    "type": "log"
  },
  "host": {
    "name": "localhost"
  },
  "tags": [
    "beats_input_codec_plain_applied"
  ],
  "fileset": {
    "name": "log"
  },
  "message": "2025-09-25 14:03:17 UTC [2175]: [1-1] user=,db=,client= FATAL:  archive command failed with exit code 127",
  "@version": "1",
  "agent": {
    "ephemeral_id": "b4fccc1d-1bad-4298-840c-7af097b13195",
    "type": "filebeat",
    "hostname": "localhost",
    "id": "46eb61dc-b764-4bdd-92b9-6d8316da100e",
    "name": "localhost",
    "version": "7.10.2"
  },
  "service": {
    "type": "postgresql"
  },
  "log": {
    "offset": 209370,
    "file": {
      "path": "/var/log/postgresql/postgresql-15-main.log"
    }
  },
  "ecs": {
    "version": "1.5.0"
  }
}

I've read this post (Ingest pipeline not working for filebeat) saying I can add it my index but this is not the behavior I need as I'll have multiple pipeline based on the module (system, postgresql, nginx, etc.)

We found a working workaround by adding a processor in the gcp-pubsub input processors but we don't know if this is the right thing to do.

- add_fields:
    target: "@metadata"
    fields:
      pipeline: filebeat-postgresql-standard-log-pipeline
    when:
      equals:
        event.module: postgresql

Did I miss something in my configuration ?

Thanks in advance