Got it. The answer to both my questions. In Filebeats add a field ‘document_type’ and set to the type of log file you want parsed. In the below example I’m parsing the type ‘log4j’. In order to get the timestamp to be recognized as the “valid” one, you have to parse it as a date. In addition I’m saving the full original log line as “original_message”.
Create the rule and add it to the ‘all messages’ stream.
rule "parse log4j"
when
has_field(“document_type”) && to_string($message.document_type) == "log4j"
then
let message_field = to_string($message.message);
let parsed_fields = grok(pattern: “%{TIMESTAMP_ISO8601:tx_timestamp}\s+%{LOGLEVEL:loglevel}\s+%{NOTSPACE:classname}%{GREEDYDATA:message}”, value: message_field);
set_fields(parsed_fields);
let date = parse_date(to_string(parsed_fields.tx_timestamp), “YYYY-MM-dd HH:mm:ss,SSS”, “EST”);
set_field(“timestamp”, date);
set_field(“original_message”, message_field);
end