Logstash Ingest Pipeline - Failure with 'system' Elastic Integration

[roaksoax]

I have a Logstash configured to run Ingest Pipelines, leveraging an Agent/Fleet server with a policy that only includes the 'System' integration failing. The whole error is attached here.

Logstash config is:

input {
    beats {
      id => "elastic-agent-input"
      port => 5044
      ssl => true
      ssl_certificate_authorities => ["/etc/logstash/certs/ca.crt"]
      ssl_certificate => "/etc/logstash/certs/logstash.crt"
      ssl_key => "/etc/logstash/certs/logstash.pkcs8.key"
      ssl_verify_mode => "force_peer"
    }
}
filter {
    elastic_integration {
        cloud_id => "..."
        auth_basic_username => "logstash"
        auth_basic_password => "..."
    }
}
output {
    elasticsearch {
        id => "elasticsearch_output"
        ssl => true
        data_stream => true
        cloud_id => "..."
        api_key => "..."
    }
}

• The logstash user has the following roles: logstash_admin logstash_system and a custom logstash_ingest_pipeline role with the monitor, read_pipeline, and manage_index_templates permissions.
• Fleet Agent Policy Integration is the default 'system' one:

[roaksoax]

logstash.log

[yaauie]

To summarize two separate issues from the logs:

1. The loaded pipeline has a pipeline processor configured to run events through metrics-elastic_agent.metricbeat@custom if it exists, but when when ES gives us a 404 Not Found for that pipeline's definition, we log a warning and the exception which floods the logs with useless noise.
2. When reconstructing an event from the ingest document, we begin with IngestDocument#getSourceAndMetadata, which includes Elasticsearch-reserved "metadata" fields like _version, which are a poison pill that must not be included in the Elasticsearch source. We should instead use IngestDocument#getSource for the base of our Event, and decide what to do with the separately-available reserved metadata fields available with IngestDocument#getMetadata.

[roaksoax]

Workaround per @jsvd on slack is to add remove_field, like below:

filter {
  elastic_integration {
    cloud_id => "..."
    auth_basic_username => logstash 
    auth_basic_password => ''
    remove_field => ['_version']
  }
}