Filter out events by JSON content

[smoking81 (Federico)]

Hello there,
I am running Filebeat on K8S and want to index just application files encoded in JSON which contain a field "classtype" with value "application". It seems a quite easy requirement, after trying all the possible combinations of json.message_key, include_lines and decode_json_fields I still didn't succeed in achieving what I wanted.. Can you please help me?

My extract from https://github.com/elastic/beats/blob/master/deploy/kubernetes/filebeat-kubernetes.yaml looks now like this

  kubernetes.yml: |-
- type: docker
  containers.ids:
  - "*"
  #      json.message_key: message
  #      json.keys_under_root: true
  #      json.overwrite_keys: true
  tail_files: true
  include_lines: ['"classtype":"application"']
  processors:
    - add_kubernetes_metadata:
        in_cluster: true
    - decode_json_fields:
        fields: ["message"]
        target: ""
  #            overwrite_keys: true

Thanks in advance!
Regards

[shaunak (Shaunak Kashyap)]

What about using the decode_json_fields processor like you are, and then after that, using the drop_event processor with an appropriate when condition: https://www.elastic.co/guide/en/beats/filebeat/current/drop-event.html?

[smoking81 (Federico)]

Hi @shaunak, thank you for replying!
I think my problem is there are a lot of log entries from different sources and I want to do following:

• Discard non JSON
• Discard JSON not containing the field "classtype" with value of "application"

Example:

{"log":"2019-04-24 14:52:35.571 I must be discarded\n","stream":"stdout","time":"2019-04-24T14:52:35.571865144Z"}
{"log":"{"severity":"INFO","classtype":"application","service":"my-service","trace":"","span":"","parent":"","exportab
le":"","pid":"11","thread":"scheduling-1","class":"MyJavaClass","message":"I must be fully indexed","stacktrace":""}\n","stream":"stdout","time":"2019-04-23T15:57:36.409648401Z"}
{"log":"{"severity":"INFO","service":"my-service","trace":"","span":"","parent":"","exportab
le":"","pid":"11","thread":"scheduling-1","class":"MyJavaClass","message":"I must NOT be indexed (no classtype)","stacktrace":""}\n","stream":"stdout","time":"2019-04-23T15:57:36.409648401Z"}

What happens using for example json.message_key: log is that logs which are not JSON at all cause following exception:

ERROR json/json.go:51 Error decoding JSON: json: cannot unmarshal number into Go value of type map[string]interface {}

Do you have any idea ho to achieve my goal? Thanks

[smoking81 (Federico)]

I am really stuck with this problem.. It seems a quite easy requirement but I don't understand how to solve it.
Is the "\n" after the JSON which leads to the "Error decoding JSON" error for this log with both decode_json_fields fields ["log"] and json.message_key: log?

"log":"{"severity":"INFO","classtype":"application","service":"my-service","trace":"","span":"","parent":"","exportab
le":"","pid":"11","thread":"scheduling-1","class":"MyJavaClass","message":"I must be fully indexed","stacktrace":""}\n"

Is there an easy solution for this? Thanks in advance!