Splunk Search

How do I "merge" events?

Communicator

Hello,

I have to parse this very custom LOG, and i'm having trouble figuring out how to do this: I have two different events, let's call them event A and event B, with no string in common, but related to accepted incoming connections and logging accounts. What I need is when I make a search of the event B, splunk automatically displays the first previous event to event B, which is an event of the server accepting an incoming connection from an IP.

Event samples:

EVENT A :
Apr 16 09:59:00 [info ] server_accept: accept connection from 192.168.1.2
(I extracted the field IP from this one)

EVENT B:
Apr 16 09:59:00 [info ] on_client_loginreq: got client (*Someuser) login request sessionnum=0x235
(From this one I extracted the field account (someuser on this case)

Everytime an account logs in, after the IP event it comes an account event, so, Im sure that if I look for account, the first event before is the correct IP event, indicating that THAT account has logged from THAT IP.
My idea and what I need to achieve, is to find which IP logged which account, for example if I run the search: Account=Someuser, the results should be a list of:

EVENT B
EVENT A

EVENT B
EVENT A

EVENT B
EVENT A

EVENT B
EVENT A

togheter. Any Ideas? Thanks in advance.

0 Karma

Splunk Employee
Splunk Employee

The following SPL generates some dummy data, and then uses the streamstats to combine ip_address and account onto the same line. Hopefully this should get you started.......

| makeresults count=10 
| streamstats count 
| eval _time = (_time - 10) + count
| eval ip_address = CASE(count==1, "192.168.100.100", count==4, "192.168.100.104", count==7, "192.168.100.107") 
| eval account = CASE(count==2, "sivill", count==6, "jgos", count==10, "gurav") 
| table _time,account,ip_address 
| search ip_address=* OR account=* 
| streamstats reset_after=isnull(ip_address) last(account) as logged_in_account, first(ip_address) as logged_in_ip_address 
| search logged_in_account=* AND logged_in_ip_address=*
| fields logged_in_account, logged_in_ip_address
| fields - _time
0 Karma

Splunk Employee
Splunk Employee

Given the information about irrelevant IPs I've tweaked the previous example..

| makeresults count=10 
| streamstats count 
| eval _time = (_time - 10) + count
| eval ip_address = CASE(count==1, "192.168.100.100", count==3, "192.168.100.103", count==4, "192.168.100.104", count==5, "192.168.100.105", count==7, "192.168.100.107", count==9, "192.168.100.109") 
| eval account = CASE(count==2, "sivill", count==6, "jgos", count==10, "gurav") 
| table _time,account,ip_address 
| search ip_address=* OR account=* 
| streamstats reset_after=isnull(ip_address) last(account) as logged_in_account, last(ip_address) as logged_in_ip_address 
| search logged_in_account=* AND logged_in_ip_address=*
| fields logged_in_account, logged_in_ip_address
| fields - _time

Optionally a

| search logged_in_account = "sivill"

can be added at the end to reduce to specific accounts

0 Karma

Communicator

Hello, I don't know if i'm doing it right, I've changed the account examples to make it more like the real log, but I'm getting the following results:

logged_in_account

logged_in_ip_address

sivill 192.168.100.100
jgos-23xed 192.168.100.105
gurav_2fsdff 192.168.100.109

😞

0 Karma

Communicator

i'm not sure if what I need is to merge events. I just need a way to tell splunk "hey, bring me the next event after the IP event". With that, i can make an account search, which brings me the ip events too.

0 Karma

Contributor

here is the sample search

index=* |stats values(_raw) as merged by _time |fields merged

not sure how reverse merging works .

0 Karma

Communicator

thanks for your response, it partially works. Doing this search:

index=myindex sourcetype=mysourcetype src_ip=* OR account=* |stats values(_raw) as merged by _time |fields merged

brings me right events. sometimes I get 4 events instead of two, because of the time being the same in the 4 events, but still it's partially solved because when I do this search (this is what I need to achieve):

index=d2cs sourcetype=d2cs account=matyc OR src_ip=* |stats values(_raw) as merged by _time |fields merged

here i'm trying to figure out the IPs for the account matyc, but when performing that search, it brings me the rest of the irrelevant IP events, here is an example:

Apr 16 09:00:27 [info ] server_accept: accept connection from ip address 2018-04-16 09:00:27
Apr 16 09:00:35 [info ] server_accept: accept connection from ip address 2018-04-16 09:00:35
Apr 16 09:00:53 [info ] on_client_loginreq: got client (*Matyc) login request sessionnum=0x37E
Apr 16 09:00:53 [info ] server_accept: accept connection from ip address
2018-04-16 09:00:53
Apr 16 09:01:08 [info ] server_accept: accept connection from ip address 2018-04-16 09:01:08
Apr 16 09:01:12 [info ] server_accept: accept connection from ip address 2018-04-16 09:01:12
Apr 16 09:01:56 [info ] server_accept: accept connection from ip address 2018-04-16 09:01:56
Apr 16 09:02:05 [info ] server_accept: accept connection from ip address 2018-04-16 09:02:05
Apr 16 09:02:29 [info ] server_accept: accept connection from ip address 2018-04-16 09:02:29

This might work, but if I want to search accounts which logged last time 2 months ago, I have to navegate backwards to 2 months ago in the results.

Thanks!

0 Karma

Champion

You can try something like: | transaction _time maxevents=2

Communicator

Hello, doing the search

index=myindex sourcetype=mysourcetype src_ip=* OR account=* | transaction _time maxevents=2

works, but what I need to achieve is to filter the irrelevant ip events which doesn't correspond to the account i'm looking for, for example if I run the search:

index=myindex sourcetype=mysourcetype src_ip=* OR account=myaccount | transaction _time maxevents=2

I get a lot of irrelevant ip events, example:

Apr 17 10:00:49 [info ] server_accept: accept connection from ip address
Apr 17 10:00:49 [info ] on_client_loginreq: got client (*myaccount) login request sessionnum=0xD9
Apr 17 09:58:16 [info ] server_accept: accept connection from ip address
Apr 17 09:58:16 [info ] server_accept: accept connection from ip address
Apr 17 09:58:16 [info ] server_accept: accept connection from ip address
Apr 17 09:58:16 [info ] server_accept: accept connection from ip address
Apr 17 09:58:15 [info ] server_accept: accept connection from ip address
Apr 17 09:58:15 [info ] server_accept: accept connection from ip address
Apr 17 09:58:14 [info ] server_accept: accept connection from ip address
Apr 17 09:58:14 [info ] server_accept: accept connection from ip address
Apr 17 09:55:29 [info ] server_accept: accept connection from ip address

thanks!

0 Karma

Communicator

I managed to solve it!

just added another pipe, like this:

index=myindex sourcetype=mysourcetype account="" OR src_ip=""
| transaction _time maxevents=2
| search account=myaccount

and it worked!! thanks!

0 Karma
Don’t Miss Global Splunk
User Groups Week!

Free LIVE events worldwide 2/8-2/12
Connect, learn, and collect rad prizes and swag!