Ingestion Rate decreases over time

Elastic Stack Logstash
1

Hi everyone, I've created a Logstash ingestion pipeline that takes the logs from OpenSearch (about 30M logs per day) and forwards them into elasticSearch. My goal is to be able to load into ELK the last day of logs (as of now) being able to ingest all the logs in (sort of) real-time.
The pipeline configuration follows:

input {
  opensearch {
    hosts =>  ["https://<IP>:9200"]
    user  =>  "<USR>"
    password  =>  "${open_PW}"
    index =>  "wazuh-a*"
    schedule => "* * * * *" 
    query =>  '{"query": {
            "bool": {
             "must": [{
              "range": {
                "timestamp": {
                  "time_zone": "+01:00",
                  "gte": "now-1d/d",
                  "lte": "now/d"
                }
              }
            }
          ]
        }
      }
    }'
  }
}

filter {
    mutate {
        remove_field => [
        "@version",
        "[agent][id]",
        "[agent][name]",
        "[cluster][name]",
        "[cluster][node]",
        "[data][Dettaglio][timestamp]",
        "[data][Dettaglio][assertion_id]",
        "[data][Dettaglio][attributes]",
        "[data][Dettaglio][authentication_context]",
        "[data][Dettaglio][inbound_binding]",
        "[data][Dettaglio][inbound_message_id]",
        "[data][Dettaglio][nameid_value]",
        "[decoder][name]",
        "[input][type]",
        "[rule][id]"]
    }
}

output {
    elasticsearch {
         hosts => ["<IP>:9200"]
         index => "opensearch-logs"
         user => "${logstash_USR}"
         password => "${logstash_PW}"
         ssl_enabled => true
         ssl_verification_mode => none
         ssl_certificate => "/usr/share/logstash/certs/ca/ca.crt"
         ssl_key => "/usr/share/logstash/certs/ca/ca.pkcs8.key"
         template_name => "opensearch"
         template_overwrite => false
         data_stream => false
    }
}

The issue is that the log ingestion rate decreases over time: when I deploy the pipeline the speed is high but it starts dropping. Below a screen taken from Metricbeats:

image
image1789×593 79.4 KB

ELK is deployed in a dockerized environment with 3 elastic nodes and 1 logstash node. The machine has the following HW infos:
CPU: Intel(R) Xeon(R) Gold 6238R CPU @ 2.20GHz
Core: 16
RAM: 64G

And the pipeline has the following parameters:
Pipeline Workers: 6
Pipeline batch size: 125
batch delay: 50
queue type: memory
queue max bytes 1 gigabytes
queue checkpoint writes: 0

Have you any idea of how that happens?