Problem with parsing json in syslog format

Hi, can you please help me with logstash config?

    input {
      #tcp {
      #  port => 5014
      #  type => syslog
      #}
      udp {
        port => 5014
        type => syslog
      }
    }

    filter {
      if [type] == "syslog" {
        json {
            source => "message"
            #remove_field => "message"
        }

        #date {
        #    match => [ "timestamp", "UNIX_MS" ]
        #}

        #mutate {
        #    remove_field => [ "host", "path", "_type" ]
        #}
      }
    }

    output {
      file {
        path => "/tmp/logstash_%{host}_%{+YYYY-MM-dd}.log"
        codec => rubydebug
      }
      elasticsearch {
        hosts => ["localhost:9200"]
        index => "logstash-syslog-%{+YYYY-MM-dd}"
        manage_template => false
        ilm_enabled => false
      }
    }

I got alert to file:

    {
    "@timestamp" => 2020-10-14T12:22:11.424Z,
          "host" => "192.168.1.1",
      "@version" => "1",
          "tags" => [
        [0] "_jsonparsefailure"
    ],
          "type" => "syslog",
       "message" => "<12>1 2020-10-14T12:22:11.370Z test.local TestServer 1079 - - {\"event_type\":\"Threat_Event\",\"ipv4\":\"192.168.2.2\",\"hostname\":\"Test\",\"source_uuid\":\"504e731a-9e71-48b5-b275-cfaee4f5dab1\",\"occured\":\"14-Oct-2020 12:21:55\",\"severity\":\"Warning\",\"threat_type\":\"potentially unwanted application\",\"threat_name\":\"Win32/WebCompanion.B\",\"threat_flags\":\"Variant\",\"scanner_id\":\"Real-time file system protection\",\"scan_id\":\"virlog.dat\",\"engine_version\":\"22151 (20201014)\",\"object_type\":\"file\",\"object_uri\":\"file:///C:/installer.exe\",\"action_taken\":\"cleaned by deleting\",\"threat_handled\":true,\"need_restart\":false,\"username\":\"DESKTOP-ABCD\\\\DELL\",\"processname\":\"C:\\\\utweb_installer (1).exe\",\"circumstances\":\"Event occurred on a newly created file.\",\"firstseen\":\"14-Oct-2020 12:12:52\",\"hash\":\"2179FD861CB63D4B627AEC617\"}\n"
}

Some logs also looks like:

       "message" => [
        [0] "<12>1 2020-10-14T12:22:11.370Z test.local TestServer 1079 - - {\"event_type\"[...]\"}\n",
        [1] "{\"event_type\"[...]\"}\n"
    ]

^ Have one syslog-like intro and then 2 (or more?) jsons included.

Best would be if also timestamp can be taken from "occured", however even if I will be able just parse this will be great, please help me because after reading documentation I still don't have idea how fix this :(...

This is syslog format however there is JSON included in this syslog event. As alternative I can select LEEF format, but I think JSON is better for elastic?

Here is an example Parse JSON string contained in a Syslog message

This uses grok to parse off the syslog headers before using the json filter

    if [type] == "syslog" {
            grok {
                    match   =>   {  "message" => "%{SYSLOGBASE} %{GREEDYDATA:syslog_message}"  }
            }

            json {
                    source  =>   {  source  => "syslog_message"   }
            }
    }

Thanks but unfortunately looks like this is not common syslog :frowning: i.a. because this <12> etc numbers at the begning.

That could be a syslog priority.

I would use a dissect filter on that.

But how? can you please elaborate more :)? I have no idea how deal with it...