Splunk command that runs a search for each event by passing the event's values as parameters

commands.conf

[mapsearch]
chunked = true
filename = map_search.py
python.version = python3

map_search.py

import os
import sys

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "lib"))
from splunklib.searchcommands import dispatch, StreamingCommand, Configuration, Option, validators

import splunklib.client
import splunklib.results


def build_param(key, value):
    if isinstance(value, str):
        return f'{key}="{value}"'
    raise Exception(f'Unsupported type for key {key}')


def build_params(record):
    """Returns a parameter string of the form `key1="value1" key2="value2"` for the `str` values of the dict `record`"""
    params = []
    for key, value in record.items():
        params.append(build_param(key, value))
    return ' '.join(params)


@Configuration()
class MapSearchCommand(StreamingCommand):

    saved_search = Option(name='search', require=True)
    maxsearches = Option(name='maxsearches', default=100, require=False,
                         validate=validators.Integer(1))

    def stream(self, records):
        # Connect to Splunk using credentials from calling command
        session_key = self.metadata.searchinfo.session_key
        app_name = self.metadata.searchinfo.app
        search_results = self.search_results_info
        service = splunklib.client.Service(
            token=session_key, app=app_name)
        self.logger.info(
            f'Connected to Splunk using app context `{app_name}` and using time range from calling command: {search_results.search_et} {search_results.search_lt}')

        # Run saved search for each input record
        for index, record in enumerate(records):
            if index == self.maxsearches:
                self.logger.warning(
                    f'Reached limit of max. {self.maxsearches} searches. Skipping remaining records. Try increasing the `maxsearches` argument.')
                break
            params = build_params(record)
            self.logger.info(
                f'Calling: | savedsearch {self.saved_search} {params}')
            response = service.jobs.oneshot(
                f'| savedsearch {self.saved_search} {params}', count=100000, earliest_time=search_results.search_et, latest_time=search_results.search_lt)
            reader = splunklib.results.ResultsReader(response)
            counter = 0
            for item in reader:
                if isinstance(item, dict):
                    counter += 1
                    # copy field values from input event to output event
                    for key in self.fieldnames:
                        item[key] = record[key]
                    yield item
            self.logger.info(
                f'Successfully called saved search `{self.saved_search}` - returned {counter} items')


dispatch(MapSearchCommand, sys.argv, sys.stdin, sys.stdout, __name__)

searchbnf.conf

[mapsearch-command]
syntax = mapsearch search=<string> maxsearches=<integer> <field-list> 
shortdesc = Runs a saved search for each input event by passing the events values as parameters and returns the events of each search.
description = Runs a saved search for each event by passing the events values as parameters. \
    Each event returned by the saved search is added to the result events.\
    Optionally, you can specify a list of fields that will be copied from the input events to the output events.
usage = public
comment1 = Runs the saved search `mysearch` three times, each time with a different parameter value. Each result of the saved search keeps the `Keep this value!` value.
example1 = | makeresults \
        | eval parameter="buttercup rarity tenderhoof"\
        | makemv delim=" " parameter \
        | mvexpand parameter\
        | eval value="Keep this value!"\
        | mapsearch search=mysearch value 

If you're running saved searches in Splunk as subsearches inside of the map command, they are bound by the subsearch limitation.

This is an alternative command that doesn't have this limitation as it starts a new job for each subsearch.

To use it, instead of calling:

| makeresults 
| map test

You're using:

| makeresults 
| mapsearch search=test

Missing the full flexibility of map, the command also passes each event's values as input parameters to each called saved search. Optionally, you can specify a list of fields that will be copied from the input events to the output events.