Title

Easily Create Mod Inputs Using Splunk Add-on Builder 2.0 – Part IV

Description

Steps to leverage the Splunk Add-On Builder 2.0 to create custom codes and modular inputs.

Image

Template

Article

Author

Splunk

Author URL

/en_us/blog/author/admin

Tags

Category

Tips & Tricks

Published

December 07, 2016

Read Time

9 Minute Read

# encoding = utf-8

import os

import sys

import time

import datetime

IMPORTANT

Edit only the validate_input and collect_events functions.

Do not edit any other part in this file.

This file is generated only once when creating

the modular input.

def validate_input(helper, definition):

“””Implement your own validation logic to validate the input stanza configurations”””

# This example accesses the modular input variable

# query_max = definition.parameters.get(‘query_max’, None)

pass

def collect_events(helper, inputs, ew):

“””Implement your data collection logic here”””

# The following example accesses the configurations and arguments

# Get the arguments of this input

# opt_query_max = helper.get_arg(‘query_max’)

# Get options from setup page configuration

# Get the loglevel from the setup page

# loglevel = helper.get_log_level()

# Proxy setting configuration

# proxy_settings = helper.get_proxy()

# User credentials

# account = helper.get_user_credential(“username”)

# Global variable configuration

# global_api_uri_base = helper.get_global_setting(“api_uri_base”)

# global_api_version = helper.get_global_setting(“api_version”)

# Write to the log for this modular input

# helper.log_error(“log message”)

# helper.log_info(“log message”)

# helper.log_debug(“log message”)

# Set the log level for this modular input

# helper.set_log_level(‘debug’)

# helper.set_log_level(‘info’)

# helper.set_log_level(‘warning’)

# helper.set_log_level(‘error’)

# helper function to send http request

# response = helper.send_http_request(url, method, parameters=None, payload=None,

# headers=None, cookies=None, verify=True, cert=None, timeout=None, use_proxy=True)

# get the response headers

# r_headers = response.headers

# get the response body as text

# r_text = response.text

# get response body as json. If the body text is not a json string, raise a ValueError

# r_json = response.json()

# get response cookies

# r_cookies = response.cookies

# get redirect history

# historical_responses = response.history

# get response status code

# r_status = response.status_code

# check the response status, if the status is not sucessful, raise requests.HTTPError

# response.raise_for_status()

#

# checkpoint related helper functions

# save checkpoint

# helper.save_check_point(key, state)

# delete checkpoint

# helper.delete_check_point(key)

# get checkpoint

# state = helper.get_check_point(key)

#

”’

# The following example writes a random number as an event

import random

data = str(random.randint(0,100))

event = helper.new_event(source=helper.get_input_name(), index=helper.get_output_index(), sourcetype=helper.get_sourcetype(), data=data)

try:

ew.write_event(event)

except Exception as e:

raise e

”’

# encoding = utf-8import osimport sysimport timeimport datetime'''    IMPORTANT    Edit only the validate_input and collect_events functions.    Do not edit any other part in this file.    This file is generated only once when creating    the modular input.'''def validate_input(helper, definition):    """Implement your own validation logic to validate the input stanza configurations"""    # This example accesses the modular input variable    # query_max = definition.parameters.get('query_max', None)    passdef collect_events(helper, inputs, ew):  # We import json library for use in massaging data before writing the event  import json    # Return all the stanzas (per step #6)  stanzas = helper.input_stanzas    # Iterate through each defined Stanza (per step #6)  # NB: I only ident this with two spaces so I don't have to re-ident everything else  for stanza in stanzas:          # Another two-space identation keeps all the "give-me" code from step #7 in-play     # without more indenting exercises    helper.log_info('current stanza is: {}'.format(stanza))        """Implement your data collection logic here"""    # The following example accesses the args per defined input    opt_query_max = helper.get_arg('query_max')    # Test mode will yield single instance value, but once deployed,     # args are returned in dictionary so we take either one    if type(opt_query_max) == dict:        opt_query_max = int(opt_query_max[stanza])    else:        opt_query_max = int(opt_query_max)    # Fetch global variable configuration (add-on setup page vars)    # same as above regarding dictionary check    global_api_uri_base = helper.get_global_setting("api_uri_base")    if type(global_api_uri_base) == dict:        global_api_uri_base = global_api_uri_base[stanza]    global_api_version = helper.get_global_setting("api_version")    if type(global_api_version) == dict:        global_api_version = global_api_version[stanza]            # now we construct the actual URI from those global vars    api_uri = '/'.join([global_api_uri_base, 'v' + global_api_version])    helper.log_info('api uri: {}'.format(api_uri))    # set method & define url for initial API query    method = 'GET'    url = '/'.join([api_uri, 'maxitem.json?print=pretty'])    # submit query    response = helper.send_http_request(url, method, parameters=None, payload=None,                              headers=None, cookies=None, verify=True, cert=None, timeout=None, use_proxy=True)    # store total number of entries available from API    num_entries = int(response.text)    helper.log_info('number of entries available: {}'.format(num_entries))    # get checkpoint or make one up if it doesn't exist    state = helper.get_check_point('stanza' + '_max_id')    if not state:        # get some backlog if it doesn't exist by multiplying number of queries by 10        # and subtracting from total number of entries available        state = num_entries - (10 * opt_query_max)        if state < 0:            state = 0    helper.log_info('fetched checkpoint value for {}_max_id: {}'.format(stanza, state))        # Start a loop to grab up to number of queries per invocation without    # exceeding number of entries available    count = 0    while (count < opt_query_max) or (count + state > num_entries):        helper.log_info('while loop using count: {}, opt_query_max: {}, state: {}, and num_entries: {}'.format(count, opt_query_max, state, num_entries))        count += 1        # update url to examine actual record instead of getting number of entries        url = '/'.join([api_uri, 'item', str(state + count) + '.json?print=pretty'])        response = helper.send_http_request(url, method, parameters=None, payload=None,                              headers=None, cookies=None, verify=True, cert=None, timeout=None, use_proxy=True)        # store result as python dictionary        r_json = response.json()          # massage epoch to a human readable datetime and stash it in key named the same        if r_json['time']:            r_json['datetime'] = datetime.datetime.fromtimestamp(r_json['time']).strftime('%Y-%m-%d %H:%M:%S')           helper.log_info('item {} is: {}'.format(state + count, r_json))           # format python dict to json proper        data = json.dumps(r_json)        # similar to getting args for input instance, find sourcetype & index        # regardless of if we're in test mode (single value) or running as input (dict of values)        st = helper.get_sourcetype()        if type(st) == dict:            st = st[stanza]        idx = helper.get_output_index()        if type(idx) == dict:            idx = idx[stanza]              # write event to index if all goes well        # NB: source is modified to reflect input instance in addition to input type        event = helper.new_event(source=helper.get_input_name() + ':' + stanza, index=idx, sourcetype=st, data=data)        try:            ew.write_event(event)            # assuming everything went well, increment checkpoint value by 1            state += 1        except Exception as e:            raise e     # write new checkpoint value    helper.log_info('saving check point for stanza {} @ {}'.format(stanza + '_max_id', state))    helper.save_check_point('stanza' + '_max_id', state)

/en_us/blog/fragments/digital-resilience-pays-off

Style

two-column