Using in-stream aggregation to manage event storms and reduce data volume

Why use stats in the stream?

You're probably familiar with running stats at search time; one of the most common Splunk platform operations is to pipe search results to stats to get aggregated results. This process takes a large set of events and extracts specific fields, usually grouped by other fields.

For example:

index=fw sourcetype="cisco:asa" | stats sum(rcvd_bytes) BY src_ip

You can do this easily at search time, so why would you want to do this in the stream before the data is indexed, especially when the aggregations are all you're typically concerned with? Does it make sense to ingest thousands, or millions of lines of discrete events if your day-to-day only consists of sum/avg/count?

By leveraging stats in the stream, you can build aggregations on large sets of data before they reach the index. This reduces the number of events arriving into the Splunk platform without losing significant data fidelity, while also reducing workload when retrieving those stats for your searches, reports, and dashboards.

Data sources

Firewall logs are typically among the most verbose data sources in both security-centric and observability-centric environments. In the case of Cisco ASA, there are some especially noisy events like ASA-4-733104, which indicates SYN flood detection during DDoS attacks or other brute force attempts.

In most cases, these log events occur fairly infrequently. However, during attacks, large spikes in these events can bloat indexes and reduce search efficiency due to repeated duplicate events. The screenshot below shows that at search time, a SYN flood attack results in the same event code occurring very frequently during the flood or attack event. These events are largely the same, with the difference being the flood rate (in this example they happen to be static). What you likely care about is the total rate in aggregate and the destination IP.

At significant scale, searching raw text logs for this data and running stats at search time wastes computing power, storage budget, and analyst time. All raw events have to be retrieved, search time fields have to be extracted, and statistics have to be calculated. When these searches run routinely, the inefficiency compounds.

Using stats in the stream, you can focus on this ASA event code and always aggregate those events. This will reduce the event flow generally, but will also automatically deduplicate, count, and summarize that data for quick and efficient analysis during attacks.

Beginner method

This beginner method provides a simple way to reduce the number of raw events coming to the index while retaining important field information. However, there are some caveats to this method:

• Your searches and detections have to change. Because stats used in the stream results in data being shipped to new indexes with new sourcetypes, your detections and dashboards have to be updated to refer to this new data.
• The more cardinality, the less effective. In-stream aggregations have relatively limited windows of data captured. When your data has a lot of variety in the groupings, the aggregations contain fewer groups and result in less dense aggregations.
• Make sure you are comfortable with the loss of the original events. Sometimes keeping the original, unaggregated data is important. This beginner method results in the loss of the original events. If you need to store the original events, you can use the advanced method to route the original events to another location like Amazon S3 for long term storage. Coupled with federated search, these original events can be searched in-place, in real time if needed.
• Consider using metrics, not events. The beginner method shows how to use a normal Splunk platform event index to store these aggregate results. But these results are now just metrics, and you can take advantage of Splunk platform metrics indexes or Splunk Observability Cloud to store and search these metrics, allowing you to search at even faster speeds while costing less to store. The advanced method shows you how turn stats into metrics that you can search in this way.

1. Start by creating a new pipeline. Both Splunk Edge Processor and Splunk Ingest Processor are able to run this logic. Choose whichever fits best in your architecture.
   
2. Set up your pipeline to with cisco:asa sourcetype data and use the default splunk_indexer destination.
   
3. Select splunk_indexer as the default destination.
   
4. In the pipeline authoring canvas, remove all of the boilerplate code and start by importing the route command to send the SYN flood events to their own index.
   

   import route from /splunk.ingest.commands

5. Start a new pipeline and use rex to extract the ASA code from all events. This gives you an initial field to work with.
   

   $pipeline = | from $source
   | rex field=_raw /%(?PASA-[^:]+)/

6. Use route to create a new path for the target events.
   

   | route asa_code == "ASA-4-733104", [

7. Use rex to extract all of the event fields.
   

   | rex field=_raw /see_full_spl2_for_regex/

   Click here to see the full SPL2 regex.

   import route from /splunk.ingest.commands 
   
   $pipeline = | from $source  
   //extract the asa code from the asa events 
   | rex field=_raw /%(?P<asa_code>ASA-[^:]+)/ 
   
   //remove the SYN flood codes from the main stream and process them separately 
   | route asa_code == "ASA-4-733104", [ 
       //extract the fields specific to this ASA event 
       | rex field=_raw /^(?P<timestamp>.*?)\s+(?P<device>\S+)\s+:\s+%ASA-4-733104:\s+TCP\s+Intercept\s+SYN\s+flood\s+attack\s+detected\s+to\s+(?P<target_ip>[\da-fA-F:.]+)\/(?P<target_port>\d+)\s+\((?P<real_ip>[\da-fA-F:.]+)\/(?P<real_port>\d+)\)\.\s+Average\s+rate\s+of\s+(?P<avg_rate>\d+)\s+SYNs\/sec\s+exceeded\s+the\s+threshold\s+of\s+(?P<threshold>\d+)/ 
   
       //create a valid timestamp for grouping 
       | eval _time = strptime(timestamp, "%Y-%m-%d %H:%M:%S %Z") 
   
       //count the number of actual events from the firewall 
       //and aggregate the average rate to get a total for this grouping 
       | stats count() AS syn_flood_event_count, sum(avg_rate) AS syn_rate BY target_ip, target_port, real_ip, real_port, span(_time, 10s) 
   
       //set an appropriate index and sourcetype for delivery to splunk 
       | eval index="fw_stats", sourcetype="cisco:asa:dedup:4-733104" 

8.  Convert the timestamp into unix time.
   

   | eval _time = strptime(timestamp, "%Y-%m-%d %H:%M:%S %Z")

9. Create the aggregations using stats. count() gives you the number of actual firewall events. sum(avg_rate) gives you the aggregated SYN rate.
   

   | stats count() AS syn_flood_event_count, sum(avg_rate) AS syn_rate BY target_ip, target_port, real_ip, real_port, span(_time, 10s)

   This assumes relatively low cardinality of the grouped fields. High cardinality events will not benefit from stats as much as low cardinality events.

10. (Optional) add CIM-related fields by adding them directly to the _raw event. You should perform this step if you want these events to be CIM-compliant so that Splunk Enterprise Security and other CIM-related detections see the aggregations in the same way we would see the raw events.

    You should follow typical "new sourcetype" steps to create search-time knowledge objects that map the aggregation output to something like the Intrusion Detection model. 

    You should generally rely on search-time artifacts and knowledge objects to do these field mappings and extractions, but this example shows how you might add context to the _raw event to assist in those search-time extractions. By adding these fields to the _raw event, you lock in that data and making it part of the permanent record. Using search-time extractions is more flexible and adaptable if you need to make changes later.

     | eval _raw = json_set(_raw, "dest_ip", real_ip,  
                                     "dest_port", real_port,  
                                     "protocol", "TCP",  
                                     "signature", "SYN Flood Attack", 
                                     "action", "blocked", 
                                     "vendor", "Cisco", 
                                     "product", "ASA", 
                                     "category", "intrusion_detection", 
                                     "dest", real_ip 
                                     ) 

11. The stats command creates a new _raw as well as top level fields. You are only interested in _raw and the normal metadata fields.
    

    | fields _raw, host, source

    If you do not remove top level fields, you get indexed fields alongside raw fields, which produces duplicate data. You want either indexed fields or raw, not both.

12. Use eval to set a proper index and sourcetype for the new stats-based output. Note that storing stats using a new index and sourcetype is a best practice to avoid co-mingling raw and stats data.
    

    | eval index="fw_stats", sourcetype="cisco:asa:dedup:4-733104"

13. Use into to send the aggregated results into the Splunk platform destination. The closing brace finalizes the route you started before.
    

    | into $destination]

14.  A final into sends all other events to their default destination.
    

    | into $destination;

Advanced method

You might want to use this method if:

• You're concerned about aggregations of events resulting in the loss of the original data. While the aggregations shown in the beginner method can address tactical use cases and help you get answers quickly and efficiently, the aggregations might not solve for both technical as well as business requirements.
• Your use case is observability-focused. The beginner method sends aggregations to the Splunk platform as normal raw events. However, because the data in the events consists of just numbers and dimensions, they are good candidates to store as metrics instead of logs. If your use case is observability-focused, you can store these metrics in Splunk Observability Cloud. The native metrics indexes in the Splunk platform can also function as an optimized destination for this data.

You can make two changes to the pipeline shown in the beginner method to satisfy both of these needs. In the following copy of the SPL2 pipeline (previously shown in the beginner method), two placeholder sections have been added where this additional behavior can be inserted:

• <CLONE ORIGINAL EVENTS TO ANOTHER INDEX OR OBJECT STORE>
• <CREATE METRICS FROM STATS> 

import route from /splunk.ingest.commands 

$pipeline = | from $source  
//extract the asa code from the asa events 
| rex field=_raw /%(?P<asa_code>ASA-[^:]+)/ 

//remove the SYN flood codes from the main stream and process them separately 
| route asa_code == "ASA-4-733104", [ 

    <CLONE ORIGINAL EVENTS TO ANOTHER INDEX OR OBJECT STORE> 
    //extract the fields specific to this ASA event 
    | rex field=_raw /^(?P<timestamp>.*?)\s+(?P<device>\S+)\s+:\s+%ASA-4-733104:\s+TCP\s+Intercept\s+SYN\s+flood\s+attack\s+detected\s+to\s+(?P<target_ip>[\da-fA-F:.]+)\/(?P<target_port>\d+)\s+\((?P<real_ip>[\da-fA-F:.]+)\/(?P<real_port>\d+)\)\.\s+Average\s+rate\s+of\s+(?P<avg_rate>\d+)\s+SYNs\/sec\s+exceeded\s+the\s+threshold\s+of\s+(?P<threshold>\d+)/ 

    //create a valid timestamp for grouping 
    | eval _time = strptime(timestamp, "%Y-%m-%d %H:%M:%S %Z") 
    //count the number of actual events from the firewall 
    //and aggregate the average rate to get a total for this grouping 
    | stats count() AS syn_flood_event_count, sum(avg_rate) AS syn_rate BY target_ip, target_port, real_ip, real_port, span(_time, 10s) 
 
   <CREATE METRICS FROM STATS> 
] 

//all other codes go to the default splunk destination without any modifications 
| into $destination; 

Next steps

Aggregations are a powerful tool for reducing noise and optimizing storage and search, but careful consideration must be given to the integration of these changes to your existing production workflows, knowledge objects, and search artifacts. Additional benefit and further optimizations can be found where the scale of your data is very high or specific performance is needed.

Data management pipelines open up new opportunities with a wide variety of options and implementation patterns. Make sure you review Splunk Validated Architectures, have proper training, and work with data stakeholders in your journey to implement these techniques.

In addition, these resources might help you understand and implement this guidance:

• Splunk OnDemand Services: Use these credit-based services for direct access to Splunk technical consultants with a variety of technical services from a pre-defined catalog. Most customers have OnDemand Services per their Success Plan. Engage the ODS team at ondemand@cisco.com if you would like assistance.