413 Request Entity Too Large - how to debug this? [Elastic][Logstash][Filebeat]

[bhrt]

Hello,

I am facing issue with my logstash configuration. I am feeding logstash with log events using filebeat as a client.
There are multiple logfiles (with different file formats) parsed in single logstash pipeline (I know that I have to split this )
I am sending only text, no binary data.
On a client side I am using filebeat multiline config to group events based on starting string token.
On a logstash side I am using grok multiline match (?m) for those merged messages.
And in some different cases I am using aggregate filter to merge events based on task_id.

Once for a while I am getting error message like below:
[ERROR][logstash.outputs.elasticsearch] Encountered a retryable error. Will Retry with exponential backoff {:code=>413, :url=>"https://xxxxxxx:9200/_bulk", :content_length=>1348288}

I assume that this is due to my configuration error. Due to some mismatch in log file pattern that do not split events and makes message to big to process. I am grouping single transaction per event and I don't think those events should be that big.

My question is. What is the easiest way to debug issue like that.
I would like to flush this long event to ?stdout? if it's length is grater than acceptable configuration to check what the issue was.

At this moment as soon as Encountered a retryable error. Will Retry with exponential backoff will occur pipeline is stuck and I have to force stop logstash.

I was manage to reduce number of those error by reducing pipeline.batch.size but this is just workaround not a solution. Increasing max content length is also not a valid option.

Filebeat example config:

    - paths:
        - 'C:\some_path\*.log'
      fields_under_root: true
      fields:
        type: YYYYY
      multiline:
        pattern: "^[[:digit:]]{4}-[[:digit:]]{2}-[[:digit:]]{2} [[:digit:]]{2}:[[:digit:]]{2}:[[:digit:]]{2},[[:digit:]]{3}"
        negate: true
        match: after
        max_lines: 100
    - paths:
        - 'C:\other_paht\*.txt'
      fields_under_root: true
      fields:
        type: XXXXX
       multiline:
         pattern: "^[[:digit:]]{2}:[[:digit:]]{2}:[[:digit:]]{2},[[:digit:]]{3} *[=]* TOKEN [=]*"
         negate: true
         match: after
         max_lines: 200

Sample logstash config:

filter {
  if [type] == "XXXXX" {
    grok {
      match => [ "message", "%{TIME:timestamp} *(?m)%{GREEDYDATA:logline}" ]
    }
  } else if [type] == "YYYYY" {
    grok {
      match => [ "message", "%{TIME:timestamp} *\[%{DATA:transactionid} *\] *\[%{DATA:logger} *\] *\[%{DATA:threadid} *\] *%{WORD:loglevel} *%{GREEDYDATA:logline}" ]
    }
    #start event
    if "START:::TOKEN" in [logline] {
      aggregate {
        task_id => "%{transactionid}"
        code => "map['_msg'] = event.get('message') + 10.chr; event.cancel();"
        map_action => "create"
      }
    }
    #end event
    if "END:::TOKEN" in [logline] {
      aggregate {
        task_id => "%{transactionid}"
        code => "map['_msg'] += event.get('message') + 10.chr; event.set('message', map['_msg'])"
        map_action => "update"
        end_of_task => true
        timeout => 120
      }
    } else {
      aggregate {
        task_id => "%{transactionid}"
        code => "map['_msg'] += event.get('message') + 10.chr; event.cancel(); "
        map_action => "update"
      }
    }
  }
}

Thank you in advance for your help!
Ł.