通过 Python SDK 在 Splunk 上搜索答案

【问题标题】：Search on Splunk via Python SDK通过 Python SDK 在 Splunk 上搜索
【发布时间】：2020-10-09 07:27:46
【问题描述】：

我正在尝试通过 Python SDK（Python 3.8.5、splunk-sdk 1.6.14）运行简单搜索。 dev.splunk.com 上提供的示例很清楚，但是当我使用自己的参数运行搜索时出现问题

代码就这么简单

search_kwargs_params = {
    "exec_mode": "blocking",
    "earliest_time": "2020-09-04T06:57:00.000-00:00",
    "latest_time": "2020-11-08T07:00:00.000-00:00",        
}
search_query = 'search index=qwe1 trace=111-aaa-222 action=Event.OpenCase'
job = self.service.jobs.create(search_query, **search_kwargs_params)
for result in results.ResultsReader(job.results()):
    print(result)

但是搜索没有返回结果。当我在 Splunk Web GUI 中手动运行相同的查询时，它工作正常。

我还尝试将所有参数放入“search_kwargs_params”字典中，扩大搜索时间段并获得一些搜索结果，但它们似乎与我在 GUI 中得到的不合适。

有人可以建议吗？

【问题讨论】：

您希望看到什么？您是否已验证您的凭据以通过 API 连接到 Splunk？

标签： python splunk

【解决方案1】：

这对我有用。你也可以试试这个：

import requests
import time
import json

scheme = 'https'

host = '<your host>'

username = '<your username>'
password = '<your password>'

unique_id  = '2021-03-22T18-43-00' #You may give any unique identifier here

search_query = 'search <your splunk query>'

post_data = { 'id' : unique_id,
              'search' : search_query,
              'earliest_time' : '1',
              'latest_time' : 'now',
            }

#'earliest_time' : '1', 'latest_time' : 'now'
#This will run the search query for all time

splunk_search_base_url = scheme + '://' + host + 
'/servicesNS/{}/search/search/jobs'.format(username)
resp = requests.post(splunk_search_base_url, data = post_data, verify = False, auth = 
(username, password))

print(resp.text)

is_job_completed = ''

while(is_job_completed != 'DONE'):
    time.sleep(5)
    get_data = {'output_mode' : 'json'}
    job_status_base_url = scheme + '://' + host + 
    '/servicesNS/{}/search/search/jobs/{}'.format(username, unique_id)
    resp_job_status = requests.post(job_status_base_url, data = get_data, verify = 
    False, auth = (username, password))
    resp_job_status_data = resp_job_status.json()
    is_job_completed = resp_job_status_data['entry'][0]['content']['dispatchState']
    print("Current job status is {}".format(is_job_completed))

splunk_summary_base_url = scheme + '://' + host + 
'/servicesNS/{}/search/search/jobs/{}/results?count=0'.format(username, unique_id)
splunk_summary_results = requests.get(splunk_summary_base_url, data = get_data, verify 
= False, auth = (username, password))
splunk_summary_data = splunk_summary_results.json()

#Print the results in python format (strings will be in single quotes)
for data in splunk_summary_data['results']:
    print(data)


print('status code...')
print(splunk_summary_results.status_code)

print('raise for status...')
print(splunk_summary_results.raise_for_status())

print('Results as JSON : ')


#Print the results in valid JSON format (Strings will be in double quotes)

#To get complete json data:
print(json.dumps(splunk_summary_data))

#To get only the relevant json data:
print(json.dumps(splunk_summary_data['results']))

干杯！

您可能还想看看这个非常方便的教程。 https://www.youtube.com/watch?v=mmTzzp2ldgU

【讨论】：