Help with Custom Multiline Parser for Kafka Stack Trace

[kjq]

What am I doing wrong with this multiline parser? I am trying to capture a Strimzi Kafka Operator message that is not in JSON and dos not fall under the regular Java multiline parser.

This is what I have (truncated) for an input:

[INPUT]
        Alias                           kube_tail
        Name                         tail
        multiline.parser          docker, cri, java, python, go, kafka_stack_trace

Parser (these individually match in Rubular):

[MULTILINE_PARSER]
        Name                         kafka_stack_trace
        Type                            regex
        Flush_Timeout            1000
        Rule                            start_state   "^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2} [A-Z]+ [^:]+:\d+ -.*"       "cont"
        Rule                            cont             "^.*Exception:.*"                                                                     "cont"
        Rule                            cont             "^\s*at\s+.*\(.+\)"                                                                   "cont"  

Example (yest, there are 2 lines logged before the stack trace begins)

2025-04-04 20:34:31 WARN  AbstractOperator:566 - Reconciliation #143582(watch) Kafka(kafka/kafka-cluster): Failed to reconcile
io.fabric8.kubernetes.client.KubernetesClientException: Failure executing: PUT at: [https://172.20.0.1:443/apis/kafka.strimzi.io/v1beta2/namespaces/kafka/kafkas/kafka-cluster/status](https://172.20.0.1/apis/kafka.strimzi.io/v1beta2/namespaces/kafka/kafkas/kafka-cluster/status). Message: Operation cannot be fulfilled on kafkas.kafka.strimzi.io "kafka-cluster": the object has been modified; please apply your changes to the latest version and try again. Received status: Status(apiVersion=v1, code=409, details=StatusDetails(causes=[], group=kafka.strimzi.io, kind=kafkas, name=kafka-cluster, retryAfterSeconds=null, uid=null, additionalProperties={}), kind=Status, message=Operation cannot be fulfilled on kafkas.kafka.strimzi.io "kafka-cluster": the object has been modified; please apply your changes to the latest version and try again, metadata=ListMeta(_continue=null, remainingItemCount=null, resourceVersion=null, selfLink=null, additionalProperties={}), reason=Conflict, status=Failure, additionalProperties={}).
        at io.fabric8.kubernetes.client.KubernetesClientException.copyAsCause(KubernetesClientException.java:238) ~[io.fabric8.kubernetes-client-api-6.13.4.jar:?]
        at io.fabric8.kubernetes.client.dsl.internal.OperationSupport.waitForResult(OperationSupport.java:507) ~[io.fabric8.kubernetes-client-6.13.4.jar:?]
        at io.fabric8.kubernetes.client.dsl.internal.OperationSupport.handleResponse(OperationSupport.java:524) ~[io.fabric8.kubernetes-client-6.13.4.jar:?]
        at io.fabric8.kubernetes.client.dsl.internal.OperationSupport.handleUpdate(OperationSupport.java:358) ~[io.fabric8.kubernetes-client-6.13.4.jar:?]
        at io.fabric8.kubernetes.client.dsl.internal.BaseOperation.handleUpdate(BaseOperation.java:759) ~[io.fabric8.kubernetes-client-6.13.4.jar:?]
        at io.fabric8.kubernetes.client.dsl.internal.HasMetadataOperation.update(HasMetadataOperation.java:138) ~[io.fabric8.kubernetes-client-6.13.4.jar:?]
        at io.fabric8.kubernetes.client.dsl.internal.HasMetadataOperation.update(HasMetadataOperation.java:121) ~[io.fabric8.kubernetes-client-6.13.4.jar:?]
        at io.fabric8.kubernetes.client.dsl.internal.HasMetadataOperation.updateStatus(HasMetadataOperation.java:126) ~[io.fabric8.kubernetes-client-6.13.4.jar:?]
        at io.fabric8.kubernetes.client.dsl.internal.HasMetadataOperation.updateStatus(HasMetadataOperation.java:44) ~[io.fabric8.kubernetes-client-6.13.4.jar:?]
        at io.strimzi.operator.cluster.operator.resource.kubernetes.CrdOperator.lambda$updateStatusAsync$2(CrdOperator.java:122) ~[io.strimzi.cluster-operator-0.45.0.jar:0.45.0]
        at io.vertx.core.impl.ContextImpl.lambda$executeBlocking$4(ContextImpl.java:192) ~[io.vertx.vertx-core-4.5.11.jar:4.5.11]
        at io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:270) ~[io.vertx.vertx-core-4.5.11.jar:4.5.11]
        at io.vertx.core.impl.ContextImpl$1.execute(ContextImpl.java:221) ~[io.vertx.vertx-core-4.5.11.jar:4.5.11]
        at io.vertx.core.impl.WorkerTask.run(WorkerTask.java:56) ~[io.vertx.vertx-core-4.5.11.jar:4.5.11]
        at io.vertx.core.impl.TaskQueue.run(TaskQueue.java:81) ~[io.vertx.vertx-core-4.5.11.jar:4.5.11]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) ~[?:?]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) ~[?:?]
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[io.netty.netty-common-4.1.115.Final.jar:4.1.115.Final]
        at java.lang.Thread.run(Thread.java:840) ~[?:?]
Caused by: io.fabric8.kubernetes.client.KubernetesClientException: Failure executing: PUT at: [https://172.20.0.1:443/apis/kafka.strimzi.io/v1beta2/namespaces/kafka/kafkas/kafka-cluster/status](https://172.20.0.1/apis/kafka.strimzi.io/v1beta2/namespaces/kafka/kafkas/kafka-cluster/status). Message: Operation cannot be fulfilled on kafkas.kafka.strimzi.io "kafka-cluster": the object has been modified; please apply your changes to the latest version and try again. Received status: Status(apiVersion=v1, code=409, details=StatusDetails(causes=[], group=kafka.strimzi.io, kind=kafkas, name=kafka-cluster, retryAfterSeconds=null, uid=null, additionalProperties={}), kind=Status, message=Operation cannot be fulfilled on kafkas.kafka.strimzi.io "kafka-cluster": the object has been modified; please apply your changes to the latest version and try again, metadata=ListMeta(_continue=null, remainingItemCount=null, resourceVersion=null, selfLink=null, additionalProperties={}), reason=Conflict, status=Failure, additionalProperties={}).
        at io.fabric8.kubernetes.client.dsl.internal.OperationSupport.requestFailure(OperationSupport.java:660) ~[io.fabric8.kubernetes-client-6.13.4.jar:?]
        at io.fabric8.kubernetes.client.dsl.internal.OperationSupport.requestFailure(OperationSupport.java:640) ~[io.fabric8.kubernetes-client-6.13.4.jar:?]
        at io.fabric8.kubernetes.client.dsl.internal.OperationSupport.assertResponseCode(OperationSupport.java:589) ~[io.fabric8.kubernetes-client-6.13.4.jar:?]
        at io.fabric8.kubernetes.client.dsl.internal.OperationSupport.lambda$handleResponse$0(OperationSupport.java:549) ~[io.fabric8.kubernetes-client-6.13.4.jar:?]
        at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:646) ~[?:?]
        at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510) ~[?:?]
        at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2147) ~[?:?]
        at io.fabric8.kubernetes.client.http.StandardHttpClient.lambda$completeOrCancel$10(StandardHttpClient.java:141) ~[io.fabric8.kubernetes-client-api-6.13.4.jar:?]
        at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863) ~[?:?]
        at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:841) ~[?:?]
        at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510) ~[?:?]
        at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2147) ~[?:?]
        at io.fabric8.kubernetes.client.utils.AsyncUtils.lambda$retryWithExponentialBackoff$3(AsyncUtils.java:91) ~[io.fabric8.kubernetes-client-api-6.13.4.jar:?]
        at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863) ~[?:?]
        at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:841) ~[?:?]
        at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510) ~[?:?]
        at java.util.concurrent.CompletableFuture.postFire(CompletableFuture.java:614) ~[?:?]
        at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:844) ~[?:?]
        at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:482) ~[?:?]

[kjq]

The only way I was able to get this to work was to add a multiline filter after the tail input. Is there any reason I cant use these in the tail input itself?

    [INPUT]
        Alias                           kube_tail
        Name                            tail
        Tag                             kube.<namespace_name>.<container_name>.<pod_name>.<docker_id>-
        Tag_Regex                       (?<pod_name>[a-z0-9](?:[-a-z0-9]*[a-z0-9])?(?:\\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*)_(?<namespace_name>[^_]+)_(?<container_name>.+)-(?<docker_id>[a-z0-9]{64})\.log$
        Path                            /var/log/containers/*.log
        Mem_Buf_Limit                   5MB
        Ignore_Older                    24h
        Skip_Long_Lines                 On
        Skip_Empty_Lines                On
        storage.type                    filesystem
        db                              kube.db
        db.locking                      On
        multiline.parser                docker, cri

    [FILTER]
        Alias                           is_multiline
        Name                            multiline
        Match                           kube.*
        buffer                          On
        emitter_name                    parse_multiline
        multiline.key_content           log
        multiline.parser                java, kafka_stack_trace, prom_stack_trace

[kjq]

So, oddly, if I dont use the separate multiline filter and instead add the filter to the tail plugin it does not seem to work. When called out separately the logs are parsed properly.

If I paste each of the lines into Rubular they match what they are supposed to plus the called out filter looks to work.

Referencing this discussion looks like they are calling it out in a separate parser as well.

I tried changed the "steps" but that did not help.

    [MULTILINE_PARSER]
        Name                            kafka_stack_trace
        Type                            regex
        Rule                            "start_state"   "^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2} (WARN|ERROR|FATAL) [^:]+:\d+ -.*"     "exception"
        Rule                            "exception"     "^\s*[a-zA-Z0-9_.\$]+: .+"                                                  "cause"
        Rule                            "cause"         "^(\s+at\s+.*|\s*Caused by:.*|\s*Suppressed:.*|\s*\.{3} \d+ more)$"         "cause"