Blog

I often blog about Drupal, but recently I worked on a Splunk App, so thought about sharing my experience because it was an interesting one.  For that Splunk App, I was on the Function1 Slack channel posting an endless stream of Splunk related questions and my colleagues provided tremendous support, patiently responding to my non-stop inquiries. It is great working among such a great pool of talented individuals.

The gist of the Splunk App is to dynamically post a JSON request to a RESTful API (the client developed the API) that returns a JSON response, and then send the response to the Splunk engine for indexing. Although there were some nuances in the interim (authorization, pagination, throttling conditions), I just want to focus on Splunk’s SDK and specifically Modular Inputs. Modular Inputs is a feature that extends the Splunk Enterprise framework enabling us to use custom input definitions (the JSON RESTful API in this case) as if they were native Splunk Enterprise inputs. Simply put, it is a set of APIs that Splunk exposes allowing us to programmatically ingest events in Splunk by wrapping the code in a Splunk App that instantiates specific objects and implements specific methods. The alternative to a Modular Input is a Scripted Input. With the later technique, we lose on many core Splunk Enterprise functionalities; e.g. scheduling, multi-platform support, etc. 

There is a structure that Splunk requires when developing a Modular Input app. More information is provied here . This post focuses on:

• Define a scheme for the Modular Input
• Make the call to the RESTful API
• Send the JSON object to Splunk 

Define a Scheme

To define a scheme, you’ll need to implement the “get_scheme” method and then define all the required fields for that app. Here we are defining “Field1” which can then be configured when the Splunk App is installed in Splunk Enterprise

def get_scheme(self):
    # Setup App Scheme
    scheme = Scheme("My App Name")
    scheme.description = "My App name description"
    scheme.use_external_validation = True

    # Field 1
    field1_argument = Argument("field1")
    field1_argument.data_type = Argument.data_type_string
    field1_argument.description = "Field 1 description"
    field1_argument.required_on_create = True
    field1_argument.required_on_edit = True
    scheme.add_argument(field1_argument)
    .
    .
    .
    .
    return scheme

def stream_events(self, inputs, ew):
   for input_name, input_item in inputs.inputs.iteritems():
   token = input_item["token"]
   interval = int(input_item["interval"])
   endpoint = input_item["endpoint"]
   payload = input_item["json_request"]
   headers = {'authorization': token, 'content-type': "application/json"} data_response = requests.post(endpoint, data=json.dumps(json_request), headers=headers, timeout=15) json_response = data_response.json() some_json_elements = json_response['some_json_elements']

for some_json_element in some_json_elements:
    try:
       raw_event = Event()
       raw_event.stanza = "some input name"
       raw_event.host = "some_event host"
       raw_event.source = "some source"
       raw_event.data = json.dumps(some_json_element)
       raw_event.sourcetype = "some source type"
       ew.write_event(raw_event)   
    except Exception as error: 
       logging.error(error)