I think I figured it out index=original_index | addinfo | eval search_now=info_max_time | eval _raw=printf("_time=%d", info_min_time) | foreach "*" [| eval _raw = _raw.case(isnull('<<FIELD>>'),"", mvcount('<<FIELD>>')>1,", <<FIELD>>=\"".mvjoin('<<FIELD>>',"###")."\"", true(), ", <<FIELD>>=\"".'<<FIELD>>'."\"") | fields - "<<FIELD>>" ] | collect index=summary testmode=false file=summary_test_1.stash_new name=summary_test_1" marker="report=\"summary_test_1\""  

Can you explain the regex? it looks like (?::) looks for any field identifier... but what about {0}? Is that for zero of that field identifier? I am asking because I am trying the stanza below and it doesn't seem to be working: [source::(top)|(ps)|(bandwidth)] Context: sending ta_nix data from UF to HF, then using RFS to send to flat file. The stanza aboves simply routes the data to the destination using ingest actions.  Thanks!

An update: I've gotten pretty close by doing this:   index=index_1 (sourcetype=source_2 earliest=1 latest=now()) OR (sourcetype=source_1 field_D="DeviceType" field_E=*Down* OR field_E=*Up*) | eval field_AB=coalesce(field_A, field_B) | fields field_D field_E field_AB field_C | stats values(field_E) values(field_C) by field_AB    and then using the UI to sort ascending by field_E. However, there are still many cases where field_E is not populating with a matching field_AB, despite that same field_E matching a different field_AB. To illustrate, see below: field_AB field_E field_C 11122233 Up Down 00002 11111179 Down   11111178 Up 00001 11111178 and 11111179 share 00001 as field_c, but it's not populating. I checked to make sure it wasn't missing in source_2 and found multiple events where field_AB=11111178 and field_C=00001. With that in mind, I'm not sure how to ensure I'm getting all the events from the log besides using the earliest and latest filters, unless I'm misunderstanding how those filters work.

Thank you all, here are updated configurations adapted from the previous ones. NSG logs are written into an Azure Storage Account. Then a Splunk HF reads the logs from the Azure Storage account with "Splunk Add-on for Microsoft Cloud Services" and send back to the Indexers.   Configuration applied on the Splunk Heavy Forwarder (can be applied in an Indexer if you don't have an HF) hf_in_azure_nsg_app/default inputs.conf #Inputs is defined directly in the Splunk HF via WEB-UI with "Splunk Add-on for Microsoft Cloud Services" and can be found here /opt/splunk/etc/apps/Splunk_TA_microsoft-cloudservices/local/inputs.conf props.conf #NOTE: Following set-up allow to extract only the flowTuples from the payload and set _time based on flowTuples epoch #First LINE_BREAKER apply, then SEDCMD-remove_not_epoch that keeps only flowTuples, then TRANSFORMS with INGEST_EVAL that overwrite _time #flowTuples data parsing is done at search time in the Search Head with separate app #The "source" field already contains the resourceId informations (subscriptionID, resourceGroupName, nsgName, macAddress) that can be extracted on the Search Head at search time #NOTE 2: LINE_BREAKER has been enhanced to avoid extracting events with macAddress containing first 10 numeric digits #TO BE DONE: Understand if SEDCMD- has some limit on very huge payload #TO BE DONE 2: In the INGEST_EVAL with a case statement if length is lower than 10 digits valorize now() as _time #References: #https://docs.microsoft.com/en-us/azure/network-watcher/network-watcher-nsg-flow-logging-overview #https://community.splunk.com/t5/Splunk-Search/How-do-I-import-Azure-NSG-LOGs/td-p/396018 #https://community.splunk.com/t5/Getting-Data-In/How-to-extract-an-event-timestamp-where-seconds-and-milliseconds/m-p/428837 [mscs:nsg:flow] MAX_TIMESTAMP_LOOKAHEAD = 10 LINE_BREAKER = (\")\d{10}\,|(\"\,\")\d{10}\, SHOULD_LINEMERGE = false SEDCMD-remove_not_epoch = s/\"\D.*$|\{|\}|\[|\]//g TRUNCATE = 50000000 TRANSFORMS-evalingest = nsg_eval_substr_time transforms.conf [nsg_eval_substr_time] INGEST_EVAL = _time=substr(_raw,0,10)   Configuration applied on the Splunk Search Head sh_azure_nsg_app/default props.conf #References: #https://docs.microsoft.com/en-us/azure/network-watcher/network-watcher-nsg-flow-logging-overview #https://community.splunk.com/t5/Splunk-Search/How-do-I-import-Azure-NSG-LOGs/td-p/396018 #https://community.splunk.com/t5/Getting-Data-In/How-to-extract-an-event-timestamp-where-seconds-and-milliseconds/m-p/428837 [mscs:nsg:flow] REPORT-tuples = extract_tuple_v1, extract_tuple_v2 REPORT-nsg = sub_res_nsg FIELDALIAS-mscs_nsg_flow = dest_ip AS dest src_ip AS src host AS dvc EVAL-action = case(traffic_result == "A", "allowed", traffic_result == "D", "blocked") EVAL-protocol = if(match(src_ip, "^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$"), "ip", "unknown") EVAL-direction = case(traffic_flow == "I", "inbound", traffic_flow == "O", "outbound") EVAL-transport = case(transport == "T", "tcp", transport == "U", "udp") EVAL-bytes = (coalesce(bytes_in,0)) + (coalesce(bytes_out,0)) EVAL-packets = (coalesce(packets_in,0)) + (coalesce(packets_out,0)) EVAL-flow_state_desc = case(flow_state == "B", "begin", flow_state == "C", "continuing ", flow_state == "E", "end") transforms.conf [extract_tuple_v1] DELIMS = "," FIELDS = time,src_ip,dest_ip,src_port,dest_port,transport,traffic_flow,traffic_result [extract_tuple_v2] DELIMS = "," FIELDS = time,src_ip,dest_ip,src_port,dest_port,transport,traffic_flow,traffic_result,flow_state,packets_in,bytes_in,packets_out,bytes_out [sub_res_nsg] SOURCE_KEY = source REGEX = SUBSCRIPTIONS\/(\S+)\/RESOURCEGROUPS\/(\S+)\/PROVIDERS\/MICROSOFT.NETWORK\/NETWORKSECURITYGROUPS\/(\S+)\/y=\d+\/m=\d+\/d=\d+\/h=\d+\/m=\d+\/macAddress=(\S+)\/ FORMAT = subscriptionID::$1 resourceGroupName::$2 nsgName::$3 macAddress::$4 eventtypes.conf [mscs_nsg_flow] search = sourcetype=mscs:nsg:flow src_ip=* [mscs_nsg_flow_start] search = sourcetype=mscs:nsg:flow flow_state=B [mscs_nsg_flow_end] search = sourcetype=mscs:nsg:flow flow_state=E tags.conf [eventtype=mscs_nsg_flow] network = enabled communicate = enabled [eventtype=mscs_nsg_flow_start] network = enabled session = enabled start = enabled [eventtype=mscs_nsg_flow_end] network = enabled session = enabled end = enabled   Best Regards, Edoardo