I've been wanting to build some integrity checking and other functionality based on knowing the fields in a sourcetype for a while now. At my company we've built a data dictionary of indexes and sourcetype of interest to the SOC. They can search the dictionary to help them remember the important data sources. I'd like to augment/use this info in a couple of new ways: 1) give them a field list for all of these sourcetypes so they could search for which sourcetypes have a relevant field (like src_ip) 2) I'd like to note the fields that appear in 100% of records for a sourcetype and then every day find out if is missing any of those fields. This would quickly clue me into data issues related to the event sent, parsing, or knowledge objects. I know how to get a list of fields for 1 sourcetype and store that info. And I know how to compare a sourcetype to a past set of fields to a current set. My challenge now is how do I get the list of fields for the 100 sourcetypes of interest so far my best idea is to create 100 jobs to handle each sourcetype. Something like ```1-get the sourcetypes of interest and pull back data for them``` [| inputlookup dataDictionary.csv where imf_critical=true | eval yesterday=relative_time(now(),"-1d@d") | where evalTS>yesterday | dedup sourcetype | sort sourcetype | head 5 | tail 1 | table sourcetype] earliest=-2d@d latest=-1@d ```2-get samples for all indexes in which the sourcetype appears``` | dedup 10 index sourcetype | fieldsummary ```3-determine field coverage so we can pick the hallmark fields``` | eventstats max(count) as maxCount | eval pctCov=round(count/maxCount,2)*100 | table field pctCov ```4-add back in the sourcetype name``` | append [| inputlookup dataDictionary.csv where imf_critical=true | eval yesterday=relative_time(now(),"-1d@d") | where evalTS>yesterday | dedup sourcetype | sort sourcetype | head 5 | tail 1 | table sourcetype] | eventstats first(sourcetype) as sourcetype | eval evalTS=now() | table sourcetype evalTS field pctCov ```5-collect the fields to a summary index daily``` | collect index=soc_summary marker="sumType=dataInfo, sumSubtype=stFields" If I ran 100 jobs like this, the number after head would increment to give me the next sourcetype. But I feel like there has to be a better way to do fieldsummary on a lot of sourcetypes. Any ideas? ... View more

We've been trying to set up the CIM datamodels in our environment.   One that seems particularly useful is Network_Resolution (DNS).  Network Resolution has just one dataset for DNS and the events look like they intend to capture request and response data answer dest message_type query query_type reply code   src of these some are clearly related to the request and some are clearly related to an answer.   Is the idea that the datamodel should be populated with just query data (no answer) for message type=query? and just response data (no query/query type) for message_type=response? Or is there some effort that should be made to correlate the request and responses? so that each record contains everything where possible?   We did set up Splunk Stream to handle DNS.  Splunk stream tries to pair up request and response but seems to miss some pairings: I find that about 30% of DNS events from Splunk Stream have both request and response, ~40% have only query, and ~30% have only response so it would be possible to add 30% of DNS events to Network Resolution with both query and response, but then the rest would have to be just query or just response events -- missing the other fields.  Or I could just split out the query and response from the joint records.  Does anyone who uses Network Resolution (especially in ES) have a recommendation on how to do this? ... View more

Lately my teammates have been searching for email logs and not finding them until several hours or even a day later.   I checked for latency issues by comparing log record times to _indextime but never found latency more than 20 minutes      index=proofpoint source=proofpoint_message_log | eval indextime=strftime(_indextime,"%Y-%m-%d %H:%M:%S") | eval delay=_indextime-_time | timechart span=1m max(delay)       so ingestion seems to have happened within 20 minutes of all records.  But my teammates cannot find some of the logs until hours later.   Has anyone here seen this sort of thing?  In the past I've only seen this as the result of delayed ingestion.  What else could cause it? We did recently move to SmartStore.  Is it possible that this is adding additional delay in Splunk determining which buckets to search?   ... View more