I used alert_logevent alert action to ingest events from custom alert action, you may try below : from future.moves.urllib.parse import urlencode
from future.moves.urllib.request import urlopen, Request
from future.moves.urllib.error import HTTPError, URLError
from splunk.util import unicode
def log_event(helper, event, source, sourcetype, host, index):
if event is None:
helper.log_error("ERROR No event provided\n")
return False
query = [('source', source), ('sourcetype', sourcetype), ('index', index)]
if host:
query.append(('host', host))
url = '%s/services/receivers/simple?%s' % (helper.settings['server_uri'], urlencode(query))
try:
encoded_body = unicode(event).encode('utf-8')
req = Request(url, encoded_body, {'Authorization': 'Splunk %s' % helper.settings['session_key']})
res = urlopen(req)
if 200 <= res.code < 300:
helper.log_debug("receiver endpoint responded with HTTP status=%d\n" % res.code)
return True
else:
helper.log_error("receiver endpoint responded with HTTP status=%d\n" % res.code)
return False
except HTTPError as e:
helper.log_error("Error sending receiver request: %s\n" % e)
except URLError as e:
helper.log_error("Error sending receiver request: %s\n" % e)
except Exception as e:
helper.log_error("Error %s\n" % e)
return False
def your_def_like_to_ingest_events():
success = log_event(
helper,
event=data_url_final,
source=source,
sourcetype=sourcetype,
host=host,
index=index
)
if not success:
sys.exit(2) Note: only one event per call to log_event will be ingested. if you want to ingest multiple events then you need to call log_event multiple times.
... View more
