You would utilize the stats command to find an average of the  diff_seconds field using a by-field of search_name. Something like this (following the search I shared before) index=notable | eval event_epoch=if( NOT isnum(event_time), strptime(event_time, "%m/%d/%Y %H:%M:%S"), 'event_time' ), orig_epoch=if( NOT isnum(orig_time), strptime(orig_time, "%m/%d/%Y %H:%M:%S"), 'orig_time' ) | eval event_epoch_standardized=coalesce(event_epoch, orig_epoch), diff_seconds='_time'-'event_epoch_standardized' | fields + _time, search_name, event_time, diff_seconds | stats count as sample_size, min(diff_seconds) as min_diff_seconds, max(diff_seconds) as max_diff_seconds, avg(diff_seconds) as avg_diff_seconds by search_name | eval avg_diff=tostring(avg_diff_seconds, "duration")   ... View more

SPL is a bit wonky but got results in the final format you were looking for, I'm curious how this SPL will perform against your live data.       | makeresults | eval _raw="{ \"a.com\": [ { \"yahoo.com\":\"10ms\",\"trans-id\": \"x1\"}, { \"google.com\":\"20ms\",\"trans-id\": \"x2\"} ], \"b.com\": [ { \"aspera.com\":\"30ms\",\"trans-id\": \"x3\"}, { \"arista.com\":\"40ms\",\"trans-id\": \"x4\"} ], \"trans-id\":\"m1\", \"duration\":\"33ms\" }" ``` start parsing json object ``` | fromjson _raw | foreach *.* [ | eval url_json=mvappend( url_json, case( mvcount('<<FIELD>>')==1, if(isnotnull(json('<<FIELD>>')), json_set('<<FIELD>>', "url", "<<FIELD>>"), null()), mvcount('<<FIELD>>')>1, mvmap('<<FIELD>>', if(isnotnull(json('<<FIELD>>')), json_set('<<FIELD>>', "url", "<<FIELD>>"), null())) ) ) ] | fields + _time, url_json, "trans-id", duration | rename "trans-id" as "top_trans-id" | fields - _raw | mvexpand url_json | fromjson url_json | fields - url_json | foreach *.* [ | eval sub_url=if( isnotnull('<<FIELD>>') AND isnull(sub_url), "<<FIELD>>", 'sub_url' ), sub_duration=if( isnotnull('<<FIELD>>') AND isnull(sub_duration), '<<FIELD>>', 'sub_duration' ) ] | rename "trans-id" as "sub_trans-id" | fields + _time, "top_trans-id", url, duration, sub_duration, sub_url, sub_trans-id | rename "top_trans-id" as "trans-id"      Final output:   There are some pretty big assumptions here, biggest being that the keys of the _raw json will have fields with the "*.*" format or a dot in the fieldname (domain names) ... View more

By going off what you pasted it is coming back as an invalid JSON, I would check that first. But assuming that it is just a copy/paste error and you do have a valid json object as _raw then I would probably do an spath like this to retain associations between url and durations. index=hello | spath input=_raw path=details.sub-trans{} output=sub_trans | fields - _raw | table sub_trans | mvexpand sub_trans | spath input=sub_trans | fields - sub_trans   You can see here all the field are extracted and they maintained their relationships to their individual url/duration according to the structure of detail.sub-trans{} array. Does require an mvexpand though, just keep an eye out for memory limits. To retain specific associations of the url to its respective duration by extracting both as individual multivalued fields is possible but can be problematic. If any of them have a null entry for whatever reason then all associations are thrown off from that point on. Thats why in these sort of situations I would much rather extract the entire nested json object out of the array, mvexpand that, then spath that internal json.  Also want to note that doing a mvexpand against two multivalue fields like in your original search will completely loose all association between which url should have which duration. you will actually end up with N^2 results when by the structure of the json I believe there should only be N results. ... View more

Notice that your requested output has more rows than the original input rows. To do this would require some sort of transformation, one way could to use an mvexpand method and would look something like this. <base_search> | eval field3=mvappend(field1, field2) | fields + field3 | mvexpand field3 | sort 0 +field3 You can see in the screenshot that field3 is in your requested format Full SPL to replicate | makeresults count=5 | streamstats count as field1 | eval field2=case( 'field1'==1, 10, 'field1'==2, 12, True(), null() ) | fields - _time ``` mvexpand method ``` | eval field3=mvappend(field1, field2) | mvexpand field3 | sort 0 +field3 Another method would be append (subsearches can be truncated if you hit any splunk limits) something like this <base_search> field1=* | eval field3='field1' | fields + field3 | append [ | search <base_search> field2=* | eval field3='field2' | fields + field3 ] Full SPL to replicate | makeresults count=5 | streamstats count as field1 | eval field2=case( 'field1'==1, 10, 'field1'==2, 12, True(), null() ) | fields - _time | search field1=* | eval field3='field1' ``` append method ``` | append [ | makeresults count=5 | streamstats count as field1 | eval field2=case( 'field1'==1, 10, 'field1'==2, 12, True(), null() ) | fields - _time | search field2=* | eval field3='field2' ] I bet there is also a slick way of using appendpipe command to achieve this as well. <base_search> | appendpipe [ | stats values(field2) as field2 ] | eval field3=coalesce(field1, field2) | mvexpand field3 output looks like this Full SPL to replicate | makeresults count=5 | streamstats count as field1 | eval field2=case( 'field1'==1, 10, 'field1'==2, 12, True(), null() ) | fields - _time ``` appendpipe method ``` | appendpipe [ | stats values(field2) as field2 ] | eval field3=coalesce(field1, field2) | mvexpand field3 ... View more

By the look of your screenshot shared, it appears that as a result of the  | stats values(NumberOfAuthErrors) AS NumberOfAuthErrors, values(TotalRequest) AS TotalRequest is returning you two multivalued fields, so the eval is not working as intended. Try putting this stats with an additional by-field of _time again, that way each NumberOfAuthErrors and TotalRequest values should only have 1 value for each 15 minute interval and then the eval will probably work. If for whatever reason you are trying to sum up each row of two multivalued fields (Don't really know why you would want to do this), I would stay away from using stats values() as this is going to dedup values and then I believe sort them. using stats list() instead will retain the original order, but even then, if one of the datasets is missing events in one or more of the 15 minute intervals, then number will again be misaligned. You would be better off just using a stats by-field of _time again, something like this. [SEARCH] | bin _time span=15m | stats count as NumberOfAuthErrors by _time | append [ SEARCH | bin _time span=15m | stats count as TotalRequest by _time ] | stats values(NumberOfAuthErrors) AS NumberOfAuthErrors, values(TotalRequest) AS TotalRequest by _time | eval failureRate = round((NumberOfAuthErrors / TotalRequest) * 100,3) | table _time, TotalRequest NumberOfAuthErrors failureRate If you just want the overall failureReate through the entire timespan the using a stats sum() will probably be the way to go. [SEARCH] | bin _time span=15m | stats count as NumberOfAuthErrors by _time | append [ SEARCH | bin _time span=15m | stats count as TotalRequest by _time ] | stats sum(NumberOfAuthErrors) AS NumberOfAuthErrors, sum(TotalRequest) AS TotalRequest | eval failureRate = round((NumberOfAuthErrors / TotalRequest) * 100,3) | table TotalRequest NumberOfAuthErrors failureRate   ... View more

Probably a few ways of doing this, but if you have access to index=_internal you can try something like this. index=_internal component=Metrics group=per_index_thruput earliest=-30d@d latest=now | bucket span=1h _time | stats sum(kb) as hourly_kb, sum(ev) as hourly_events, by _time, series | stats earliest(_time) as earliest_event, latest(_time) as latest_event, count as sample_size, avg(hourly_kb) as avg_hourly_kb, sum(hourly_kb) as total_kb, avg(hourly_events) as avg_hourly_events, sum(hourly_events) as total_events by series | convert ctime(earliest_event), ctime(latest_event) | rename series as index ... View more

Maybe something like this?     <base_search> | eval employee_data=replace(replace(employee_data, "\'", "\""), "None", "\"None\"") | spath input=employee_data     Testing on my local instance looks like it worked out.         ... View more

There is a setting for roles in Splunk that configures what indexes are searched by default if an index is not specified in the search itself. This would be my guess is what is going on here if I understood your question correctly. The user's role that is utilizing the macro probably doesn't have the index set as a default searched index where the data resides. Here is a screenshot of the UI settings for roles default searched indexes.   ... View more

Splitting your lookup to include new fields "President", "VP", "Manager" would work but doesn't really scale if the role field has high cardinality. Here is another approach that is more scalable and can be generalized. You could make a net-new field in your lookup named role_json that would contain the mapping info of role<-->name.  Edit: Just realized your request was to not have results in the mvexpanded format I first showed. So here is an updated method.       <base_search> | lookup orgchart Org, Branch OUTPUT role_json | foreach mode=multivalue role_json [ | eval tmp_json=if( isnull(tmp_json), json_object(spath('<<ITEM>>', "Role"), spath('<<ITEM>>', "Name")), json_set(tmp_json, spath('<<ITEM>>', "Role"), spath('<<ITEM>>', "Name")) ) ] ``` capture any role_json values that are single values ``` | eval tmp_json=if( mvcount(role_json)==1, json_object(spath(role_json, "Role"), spath(role_json, "Name")), 'tmp_json' ) ``` remove role_json (no longer needed) ``` | fields - role_json ``` parse out tmp_json to table all the proper mappings ``` | spath input=tmp_json ``` remove tmp_json (no longer needed) ``` | fields - tmp_json          Directly after the lookup results would look something like this:   Then after the foreach loops. I added a few extra entries to lookup to demonstrate that this method is dynamic and doesn't need any hardcoded fieldnames to account for potential new values.   To get a new json object field into your existing lookup would look something like this (Provided its a CSV, if it is a kvstore then you would probably need to update the definition to include the new field)         | inputlookup orgchart | tojson str(Role) str(Name) output_field=role_json | outputlookup orgchart         ... View more

If a subsearch has more than 50,000 events or takes longer than 1 minute (i think) to run it will auto-finalize. Occurrence of either of these scenarios will cause the data returned from the subsearch to be truncated and incomplete. BTW I don't think the search you shared needs to use a join/subsearch, something like this will probably do the same thing. index="aaam_devops_elasticsearch_idx" ((project="Einstein360_TicketsCreated_ElasticSearch_20210419" AND "source.TransactionName"="ITGTicketCreated") OR (project="Einstein360_TruckRollCreated_ElasticSearch_20210420" AND "source.TransactionName"="Truck_Roll_Create_Result")) | timechart span=1d dc(eval(case('project'=="Einstein360_TicketsCreated_ElasticSearch_20210419" AND 'source.TransactionName'=="ITGTicketCreated", id))) as ITGTicketCreated, dc(eval(case('project'=="Einstein360_TruckRollCreated_ElasticSearch_20210420" AND 'source.TransactionName'=="Truck_Roll_Create_Result", id))) as TruckRollCreated ... View more

Since you are just wanting to display percentage of 200 and total count of all StatusCode in each minute. I think a search like this should work.   index=<index> sourcetype=<sourcetype> sc_status=* | bucket span=1m _time | stats count as Totalcount, count(eval('sc_status'==200)) as Count200 by _time | eval Percent200=round(('Count200'/'Totalcount')*100, 2) | fields + _time, Percent200, Totalcount   Example Output:   ... View more

Oh dang, good catch with the trailing comma! So just tried to limit the initially called out events as much as possible on the base search with the additional filter of  (disposition.disposition="TERMINATED" OR "connections{}.left.facets{}.number"=*) and then limiting the stats aggregations to just the fields that are required for your downstream analysis and display. Glad its running faster! ... View more

So I think on your streamstats example, the first usage of streamstats should be replaced with at "| stats" command instead. So something like this. | inputlookup direct_deposit_changes_v4_1_since_01012020.csv | eval _time=strptime(_time,"%Y-%m-%d") | stats count as daily_count by _time | sort 0 +_time | streamstats window=28 list(daily_count) as values_added_together, sum(daily_count) as sum_daily_count | table _time, daily_count, values_added_together, sum_daily_count   ... View more

Oh wow okay, you are using dashboard studio which is pretty different than legacy XML dashboards. I can try doing your use-case using dashboard studio and see if I can get it figured it out. ... View more

I agree with @gcusello here. I did notice the use of the | top limit=1 Source_Network_Address in the original subsearch which I think implies that you are trying to scope the search down to a single IP address that shows up the most often in the windows_logs index and not in the 192.168.0.0/16 range. Which I think can be done with a couple of additional lines like this. (index="iis_logs" sourcetype="iis" s_port="443" sc_status=401 cs_method!="HEAD") OR (index="windows_logs" LogName="Security" Account_Domain=EXCH OR Account_Domain="-" EventCode="4625" OR EventCode="4740" user="john@doe.com" OR user="johndoe") | eval c_ip=coalesce(Source_Network_Address,c_ip) | stats dc(index) AS index_count, count(eval('index'=="windows_logs")) as win_log_count, values(*) AS * BY c_ip | where index_count=2 AND NOT cidrmatch("192.168.0.0/16", c_ip) | sort 1 -win_log_count   ... View more

You can check if thats the case with this.       | makeresults count=15 | fields - _time | streamstats count as daily_count | streamstats window=5 list(daily_count) as values_added_together, sum(daily_count) as sum_daily_count         As you can see in the column "values_added_together" gives you a list of numbers that are being added together on each step.  This is basically the same thing the foreach loop is doing against the multivalue field just using the previous 4 entries to add instead of the next 4 entries. Ultimately, I think you would get the same results, just shifted a couple of rows depending on which method is used. The streamstats method also will include the first few entries summing up less than 5 values because there is not 5 preceeding values until it gets to row 5. Another screenshot as a POC: And to really drive it home here is a tabled visual representation of both methods together in comparison to their respective row of the original daily_count value. Values are all the same just shifted depending on if the moving window is looking back/forward   ... View more

And seeing how you will be utilizing the final results I think this query would probably give you the same results and would be much simpler.   | inputlookup direct_deposit_changes_v4_1_since_01012020.csv | eval _time=strptime(_time,"%Y-%m-%d") | stats count as daily_count by _time | sort 0 +_time | streamstats window=5 sum(daily_count) as sum_daily_count   ... View more