Please enhance this search ASAP!

[Gregg Woodcock]

You may not have heard but I recently quit my job at T-Mobile (was MetroPCS) and started my own company (Splunxter.com) to take Splunk on the road.  I have decided to boost my credentials by winning (that's right, not "entering" but WINNING) a Splunk contest and I could use a bit of help from YOU!  All I need you to do is play around with the 2 searches at the bottom of this post and give me some feedback.

The core idea is this: what useful intelligence is there to be had when we do a mashup between the following 2 sets of logs (from any search head)?

1: The phonehome events (splunkd_access.log on the DS) which tell us what forwarders are checking in with the DS.

2: The metrics log events (splunk_metrics.log on the Indexers) which tell us who is sending data into the indexers.

I have categorized 4 things that we can find so far as follows:

1: (Z)ombies who are properly authorized to be forwarding (and are doing so) but who are no longer being controlled by the DS.

2: (I)dle servers who are properly authorized to be forwarding but are not doing so.

3: (I)mpersonators who are using the same IP address as another forwarder but with a different hostname (sometimes this is a valid configuration).

4: (T)respassers who are not authorized to be forwarding but are (looks the same as a Zombie).

Please run the following searches on your biggest cluster and play around with them. Does it works as it should?  Can you find any bugs?  Does something not make sense to you?  Can you enhance the output?  Can you create a clever visualization (right now just tabular output)?

There are two variations of the same search: the first one shows hostnames by clientip and the second one is inverted showing clientips by hostname:

index="_internal" (sourcetype="splunkd_access" phonehome) OR ((source=*/metrics.log* OR source=*\\metrics.log*) group=tcpin_connections)

| rex field=uri ".+/connection_([^_]+_){3}(?<uri2hostname>[^_]+)_"

| rex field=uri_path ".+/connection_([^_]+_){3}(?<uripath2hostname>[^_]+)_"

| eval clientip=case(isnotnull(clientip),clientip, isnotnull(sourceIp),sourceIp, 1=1,"NO_CLIENTIP")

| eval hostname=case(isnotnull(hostname),hostname, isnotnull(uri2hostname),uri2hostname, isnotnull(uripath2hostname),uripath2hostname, 1=1,"NO_HOSTNAME")

| eval hostname=upper(hostname)

| eval host=if(sourcetype=="splunkd",null(),host)

| stats values(splunk_server) AS indexers values(host) AS DSHosts values(sourcetype) AS sourcetypes values(hostname) AS hostnames by clientip

| eval sourcetypeCount=mvcount(sourcetypes) | eval hostnameCount=mvcount(hostnames)

| eval ZIIT1=case(sourcetypeCount=2,"Normal", sourcetypes="splunkd_access","Idle", sourcetypes="splunkd","Zombie OR Trespasser", 1=1, "APP_ERROR")

| eval ZIIT2=case(hostnameCount>1,"Potential Impersonation(".hostnameCount.")")

| eval ZIIT=case(isnull(ZIIT2),ZIIT1, (ZIIT1="Normal"),ZIIT2, 1=1,ZIIT2.",".ZIIT1)

| fields clientip,indexers,DSHosts,hostnames,hostnameCount,ZIIT | sort 0 ZIIT

index="_internal" (sourcetype="splunkd_access" phonehome) OR ((source=*/metrics.log* OR source=*\\metrics.log*) group=tcpin_connections)

| rex field=uri ".+/connection_([^_]+_){3}(?<uri2hostname>[^_]+)_"

| rex field=uri_path ".+/connection_([^_]+_){3}(?<uripath2hostname>[^_]+)_"

| eval clientip=case(isnotnull(clientip),clientip, isnotnull(sourceIp),sourceIp, 1=1,"NO_CLIENTIP")

| eval hostname=case(isnotnull(hostname),hostname, isnotnull(uri2hostname),uri2hostname, isnotnull(uripath2hostname),uripath2hostname, 1=1,"NO_HOSTNAME")

| eval hostname=upper(hostname)

| eval host=if(sourcetype=="splunkd",null(),host)

| stats values(splunk_server) AS indexers values(host) AS DSHosts values(sourcetype) AS sourcetypes values(clientip) AS clientips by hostname

| eval sourcetypeCount=mvcount(sourcetypes) | eval clientipCount=mvcount(clientips)

| eval ZIIT1=case(sourcetypeCount=2,"Normal", sourcetypes="splunkd_access","Idle", sourcetypes="splunkd","Zombie OR Trespasser", 1=1, "APP_ERROR")

| eval ZIIT2=case(clientipCount>1,"Potential Impersonation(".clientipCount.")")

| eval ZIIT=case(isnull(ZIIT2),ZIIT1, (ZIIT1="Normal"),ZIIT2, 1=1,ZIIT2.",".ZIIT1)

| fields hostname,indexers,DSHosts,clientips,clientipCount,ZIIT | sort 0 ZIIT