# Filter by event.code and divide message field **URL:** \ **Category:** Logstash\ **Created:** [December 11, 2019, 1:43pm UTC](https://discuss.elastic.co/t/filter-by-event-code-and-divide-message-field/211473 "2019-12-11T13:43:18Z")\ **Posts on this page:** 13\ **Page:** 1
**Author:** ![KOTOXJle6](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/kotoxjle6/32/131136_2.png) [@KOTOXJle6](https://discuss.elastic.co/u/KOTOXJle6)\ **Post date:** [December 11, 2019, 1:43pm UTC](https://discuss.elastic.co/t/filter-by-event-code-and-divide-message-field/211473/1 "2019-12-11T13:43:19Z")
Hello! I have logstash 7.1 and winlogbeat 7.1. Im trying to get events 4624 from domain controllers, divide message filed into multiple field and then remove everything i don't need. I have 2 problems there: 1. I've started by trying to just remove message field from 4624 events, but couldn't even do that. If i use filter like that The message field is not deleted. Just nothing happens. I've tried many different variations of IF and AND like ``` [event.code] == 4624 "event.code" == 4624 "4624" in "[event][code]" [event.code] = 4624 "[event][code]" == 4624 ``` but no one worked for me. But if i try to remove that field with ``` if "windc" in [tags] { mutate { prune { remove_field => ["message","[event][code]"] } ``` everything working as it should. What i do wrong? 1. Is it possible to break this field into small fields one line at a time or something similar? ---
**Author:** ![Jenni](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/jenni/32/29684_2.png) [@Jenni](https://discuss.elastic.co/u/Jenni)\ **Post date:** [December 11, 2019, 1:49pm UTC](https://discuss.elastic.co/t/filter-by-event-code-and-divide-message-field/211473/2 "2019-12-11T13:49:22Z")
1. You tried everything but the correct syntax, I guess: ` if "windc" in [tags] and [event][code] == 4624 { ...` 2. I don't understand the question. What is the content of the field and what should the result look like? ---
**Author:** ![KOTOXJle6](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/kotoxjle6/32/131136_2.png) [@KOTOXJle6](https://discuss.elastic.co/u/KOTOXJle6)\ **Post date:** [December 11, 2019, 2:00pm UTC](https://discuss.elastic.co/t/filter-by-event-code-and-divide-message-field/211473/3 "2019-12-11T14:00:22Z")
Well, I tried probably everything, or 99% of the possible options, but really missed yours. Thanks, it works! There is many information that i dont need in that log, like ``` Subject: Security ID: S-1-0-0 Account Name: - Account Domain: - Logon ID: 0x0 Logon Information: Logon Type: 3 Restricted Admin Mode: - Virtual Account: No Elevated Token: Yes Impersonation Level: Impersonation Process Information: Process ID: 0x0 Process Name: - Detailed Authentication Information: Logon Process: Kerberos Authentication Package: Kerberos Transited Services: - Package Name (NTLM only): - Key Length: 0 ``` I need only Account Name, Source Network Address and maybe Logon Type. But i dont know how i can remove everything that i dont need. ---
**Author:** ![KOTOXJle6](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/kotoxjle6/32/131136_2.png) [@KOTOXJle6](https://discuss.elastic.co/u/KOTOXJle6)\ **Post date:** [December 11, 2019, 4:44pm UTC](https://discuss.elastic.co/t/filter-by-event-code-and-divide-message-field/211473/4 "2019-12-11T16:44:02Z")
I saw one solution for same question - use grok for message field. But im afraid that pattern will be too big. ---
**Author:** ![Jenni](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/jenni/32/29684_2.png) [@Jenni](https://discuss.elastic.co/u/Jenni)\ **Post date:** [December 11, 2019, 6:12pm UTC](https://discuss.elastic.co/t/filter-by-event-code-and-divide-message-field/211473/5 "2019-12-11T18:12:23Z")
Here's a ruby code suggestion for the data extraction that tries to parse your message (in a very simple way) and then decides which fields to save. Someone else might have a more elegant solution. ``` #parsing data message_content = Hash.new() str = event.get('message') fieldname = nil str.each_line do |line| next if line =~ /^$/ #empty line if !(line =~ /^\t/) then #line without indentation = root level fieldname = line.scan(/^(.*):/).first.first #the field name is the text before the ':' message_content[fieldname] = line.scan(/:\t(.*)$/).first.first if !(line =~ /:$/) #root level element with a value instead of children, so the value is behind the ':' elsif !fieldname.nil? #child element message_content[fieldname] = Hash.new if message_content[fieldname].nil? #create hash if it doesn't exist message_content[fieldname][line.scan(/^\t*(.*?):/).first.first] = line.scan(/:\t*(.*)$/).first.first #child element field name and value are before and after the ':' end end #collecting interesting fields wanted_fields = Hash.new(); wanted_fields['account_name'] = message_content['Subject']['Account Name'] if defined?message_content['Subject']['Account Name'] wanted_fields['network_address'] = message_content['Whatever']['Source Network Address'] if defined?message_content['Whatever']['Source Network Address'] wanted_fields['logon_type'] = message_content['Logon Information']['Logon Type'] if defined?message_content['Logon Information']['Logon Type'] event.set('wanted_fields', wanted_fields) ``` ---
**Author:** ![KOTOXJle6](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/kotoxjle6/32/131136_2.png) [@KOTOXJle6](https://discuss.elastic.co/u/KOTOXJle6)\ **Post date:** [December 11, 2019, 6:12pm UTC](https://discuss.elastic.co/t/filter-by-event-code-and-divide-message-field/211473/6 "2019-12-11T18:12:35Z")
I think i found solution - [https://www.elastic.co/guide/en/logstash/current/plugins-filters-split.html](https://www.elastic.co/guide/en/logstash/current/plugins-filters-split.html). I just need to join multiple documents to single, without strings that i dont need. ---
**Author:** ![KOTOXJle6](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/kotoxjle6/32/131136_2.png) [@KOTOXJle6](https://discuss.elastic.co/u/KOTOXJle6)\ **Post date:** [December 11, 2019, 6:16pm UTC](https://discuss.elastic.co/t/filter-by-event-code-and-divide-message-field/211473/7 "2019-12-11T18:16:59Z")
Thanks. Looks pretty good too. Ill try both options. If you know how i can rejoin documents after split plugin, but only with fields that i need, I would really appreciate it. ---
**Author:** ![yaauie](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/yaauie/32/23363_2.png) [@yaauie](https://discuss.elastic.co/u/yaauie)\ **Post date:** [December 11, 2019, 9:25pm UTC](https://discuss.elastic.co/t/filter-by-event-code-and-divide-message-field/211473/8 "2019-12-11T21:25:52Z")
> [@KOTOXJle6](#): > > I just need to join multiple documents to single, without strings that i dont need Re-joining events is not a trivial problem at all, so I would avoid splitting the event up if you can help it. @Jenni's code looks like a fantastic start. I would recommend using the parser bit to place the entire parsed result into a `@metadata` field (which is a part of the event but not typically included in outputs), and then using Logstash's mutate filter to extract the fields that you need. e.g., ```auto filter { ruby { code => " message_content = Hash.new() str = event.get('message') fieldname = nil str.each_line do |line| next if line =~ /^$/ #empty line if !(line =~ /^\t/) then #line without indentation = root level fieldname = line.scan(/^(.*):/).first.first #the field name is the text before the ':' message_content[fieldname] = line.scan(/:\t(.*)$/).first.first if !(line =~ /:$/) #root level element with a value instead of children, so the value is behind the ':' elsif !fieldname.nil? #child element message_content[fieldname] = Hash.new if message_content[fieldname].nil? #create hash if it doesn't exist message_content[fieldname][line.scan(/^\t*(.*?):/).first.first] = line.scan(/:\t*(.*)$/).first.first #child element field name and value are before and after the ':' end end event.set('[@metadata][parsed]', message_content) " } mutate { copy => { "[@metadata][parsed][Subject][Account Name]" => "[account_name]" "[@metadata][parsed][Logon Information][Logon Type]" => "[logon_type]" } } } ``` ---
**Author:** ![KOTOXJle6](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/kotoxjle6/32/131136_2.png) [@KOTOXJle6](https://discuss.elastic.co/u/KOTOXJle6)\ **Post date:** [December 13, 2019, 1:03pm UTC](https://discuss.elastic.co/t/filter-by-event-code-and-divide-message-field/211473/9 "2019-12-13T13:03:15Z")
Thank you very much for your reply. But when i use your filter i got error: [2019-12-13T15:59:42,589][ERROR][logstash.filters.ruby] Ruby exception occurred: undefined method `first' for nil:NilClass I've deleted everything from filter secion, except your code. ---
**Author:** ![Jenni](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/jenni/32/29684_2.png) [@Jenni](https://discuss.elastic.co/u/Jenni)\ **Post date:** [December 16, 2019, 2:50pm UTC](https://discuss.elastic.co/t/filter-by-event-code-and-divide-message-field/211473/10 "2019-12-16T14:50:39Z")
What did the file look like that should have been parsed? I kept my example very close to your example to save time instead of creating a recursive function for more levels of data or thinking of potential exceptions in the formatting. So maybe your full data contains something that cannot be matched with this simple code. If the regex doesn't match, but I just assumed that it would, this error might occur. ---
**Author:** ![KOTOXJle6](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/kotoxjle6/32/131136_2.png) [@KOTOXJle6](https://discuss.elastic.co/u/KOTOXJle6)\ **Post date:** [December 17, 2019, 10:23am UTC](https://discuss.elastic.co/t/filter-by-event-code-and-divide-message-field/211473/11 "2019-12-17T10:23:11Z")
Colleagues, I have to admit, it was an unfortunate mistake. By default, winlogbeat already splits the message field into the desired fields. I didn't notice this because we had previously specifically introduced a filter that removed these new fields and left only the message field. This is convenient for all other types of Windows logs, but not for 4626, 4776 . I'm ashamed. But I am sure that your code can be useful in other templates, it will definitely be useful to a large number of people. Thank you again for your help. ---
**Author:** ![KOTOXJle6](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/kotoxjle6/32/131136_2.png) [@KOTOXJle6](https://discuss.elastic.co/u/KOTOXJle6)\ **Post date:** [December 19, 2019, 9:21am UTC](https://discuss.elastic.co/t/filter-by-event-code-and-divide-message-field/211473/12 "2019-12-19T09:21:36Z")
BTW i want to share my solution to you: I've added this to winlogbeat: ``` processors: - add_host_metadata: ~ - add_cloud_metadata: ~ - drop_fields: fields: [winlog.event_data.ElevatedToken, winlog.event_data.ImpersonationLevel, winlog.event_data.KeyLength, winlog.event_data.LmPackageName, winlog.event_data.LogonGuid, winlog.event_data.RestrictedAdminMode, winlog.event_data.TargetLinkedLogonId, winlog.provider_guid, winlog.process.pid, winlog.event_data.TargetLogonId, winlog.event_data.TargetOutboundDomainName, winlog.event_data.TargetOutboundUserName, winlog.event_data.TargetUserSid, winlog.event_data.TransmittedServices, winlog.event_data.VirtualAccount, winlog.keywords, winlog.opcode, winlog.provider_name, winlog.record_id, winlog.process.thread.id, winlog.version] ``` and this to filter in logstash: ``` # Domain Controllers logs filtering if "windc" in [tags] and [event][code] == 4776 and [winlog][event_data][Status] == "0x0" { drop {} } if "windc" in [tags] and [event][code] == 4776 { prune { remove_field => ["[agent][id]","[agent][ephemeral_id]","[agent][hostname]","[agent][type]","[agent][version]","[ecs][version]","[event][kind]","[host][architecture]","[host][hostname]","[host][os][build]","[host][os][family]","[host][os][kernel]","[host][os][platform]","[host][os][version]"] blacklist_names => ["^.*winlog.*"] } } # Domain Controllers logon events if "windc" in [tags] and [event][code] == 4624 and [winlog][event_data][TargetUserName] == "admin1 or [winlog][event_data][TargetUserName] == "admin2" { drop {} } if "windc" in [tags] and [event][code] == 4624 { prune { remove_field => ["[agent][id]","[agent][ephemeral_id]","[agent][hostname]","[agent][type]","[agent][version]","[ecs][version]","[event][kind]","[host][architecture]","[host][hostname]","[host][os][build]","[host][os][family]","[host][os][kernel]","[host][os][platform]","[host][os][version]"] blacklist_names => ["message"] } } ``` Some field, which we never need, could be deleted via ingest pipeline: ``` "remove-fields" : { "description" : "remove a exchange of fields", "processors" : [ { "remove" : { "field" : [ "agent.ephemeral_id", "agent.hostname", "agent.id", "agent.type", "agent.version", "ecs.version", "input.type", "log.offset", "log.file.path", "version" ], "ignore_failure" : true } } ] } ``` It looks bulky, a bit, but works ok. Some filtering in logstash could be done via whitelist, but I couldn't make it work. ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [November 4, 2022, 7:37am UTC](https://discuss.elastic.co/t/filter-by-event-code-and-divide-message-field/211473/13 "2022-11-04T07:37:01Z")