A project I'm working on requires that I monitor who is logging into an application. As it is, the logs of this application only record the users as "root", not by their usernames. Another sourcetype does record user names on login, though. I'm trying to map the user names in this sourcetype to the events in the other sourcetype by making an eval statement to rename the acct "root" to the contents of the group field for events that fall within a certain time range. So far, I've been able to come up with this:
index=index_name source=source_name OR sourcetype=sourcetype_name OR acct=* OR username=* | eval acct2=if(match(acct,root),group,acct) | stats count by _time type src acct2 | sort - _time
However, this does not return anything in the stats. I don't see what I'm doing wrong in the eval, so can anyone help?
Also, is there any way to apply this only to events that occur directly after the login event in
sourcetype_name? I assume I can make a bucket for that, but I'm wondering if it can be done based on a field in source_name.
Edit: Removing the quotes around "root" at least gets results where acct2=root.
A more useful strategy is to put the events in order, then copy the correct username forward from the most recent
event of type 2 to the type 1 events.
Assumptions for the following code: There are two types of events
type 1 => index="foo" source="bar" acct="*" NOTE "*" is often "root". type 2 => index="foo" sourcetype="baz" username="*"
Here's some code that runs taht strategy, based on the above assumptions. We haven't actually done anything with the records that we passed through, we are just demonstrating the technique.
index="foo" (source="bar" AND acct="*" ) OR (sourcetype="baz" AND username="*" ) | sort 0 _time | eval realacct=if(sourcetype="baz",username,null()) | streamstats last(realacct) as realacct | where sourcetype!="baz" | rename COMMENT as "Now you have only the source=bar records, and they each know what the immediately prior baz login was."
This version clears out the other name from the record if it has been more than 10 seconds.
index="foo" (source="bar" AND acct="*" ) OR (sourcetype="baz" AND username="*" ) | sort 0 _time | eval realacct=if(sourcetype="baz",username,null()) | streamstats last(realacct) as realacct | streamstats current=f last(_time) as baztime | where sourcetype!="baz" | eval realacct=if(_time-baztime>10,null(),realacct)
No, acct is in the first record, while group and username are in the second record. But why would that stop the eval from mapping the field values if I have them both in the search query?