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)
Style
two-column
Related Articles
