#!/usr/bin/env python

# -*- coding: utf-8 -*-

from sfo_common.agent import Agent

from sfo_common.import_common import *

class ElkLog(object):

"""

處理ELK數(shù)據(jù)類(lèi)

"""

def __init__(self):

pass

def get_elk_log_json(self):

"""

通過(guò)調(diào)用elasticsearch接口查詢(xún)指定索引數(shù)據(jù)，計(jì)算集群的平均響應(yīng)時(shí)間

:return:

"""

try:

day = time.strftime("%Y.%m.%d",time.localtime(time.time()))

clusters = config.elk_index_name.split(',')

if clusters:

for cluster in clusters:

index_name="{}-swift-proxy-{}".format(cluster,day)

req_url = '{}{}/_search?pretty'.format(config.elk_server_url,index_name)

headers = {'content-type': "application/json"}

l_time = datetime.datetime.now() + datetime.timedelta(minutes=-5)

now_time = util.local2utc(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f'))

now_time_5m = util.local2utc(l_time.strftime('%Y-%m-%d %H:%M:%S.%f'))

body = {

"query": {

"bool":{

"must":{

"match_all":{}

},

"filter":{

"range":{

"@timestamp":{

"gte":now_time_5m,

"lte":now_time

}

}

}

}

},

"size": 10000,

"sort": {

"@timestamp": { "order": "asc" }

},

"_source": ["status", "method","client_ip","remote_ip","timestamp","request_time","@timestamp"]

}

#print req_url,body,headers

response = requests.post(req_url,data=json.dumps(body),headers=headers)

total_time=head_total_time=get_total_time=put_total_time=post_total_time=delete_total_time = 0.0

head_count=get_count=put_count=post_count=delete_count = 0

if response.status_code == 200:

tps = SfoClusterTps()

res_data = json.loads(response.text,encoding='UTF-8')

if res_data and res_data.has_key('hits'):

hits = res_data['hits']

total = hits['total']

list = hits['hits']

if list and total > 0:

for obj in list:

if isinstance(obj,dict) and obj.has_key('_source'):

source = obj['_source']

if source.has_key('request_time'):

total_time += float(source['request_time'])

if source.has_key('method') and str(source['method']).strip().upper()=='HEAD':

head_count += 1

if source.has_key('request_time'):

head_total_time += float(source['request_time'])

if source.has_key('method') and str(source['method']).strip().upper()=='GET':

get_count += 1

if source.has_key('request_time'):

get_total_time += float(source['request_time'])

if source.has_key('method') and str(source['method']).strip().upper()=='PUT':

put_count += 1

if source.has_key('request_time'):

put_total_time += float(source['request_time'])

if source.has_key('method') and str(source['method']).strip().upper()=='POST':

post_count += 1

if source.has_key('request_time'):

post_total_time += float(source['request_time'])

if source.has_key('method') and str(source['method']).strip().upper()=='DELETE':

delete_count += 1

if source.has_key('request_time'):

delete_total_time += float(source['request_time'])

tps.guid = str(uuid.uuid1())

tps.cluster_name = cluster

if total > 0:

tps.avg_time = '%.2f'%(total_time/total*1000)

else:

tps.avg_time = 0

if head_count > 0:

tps.head_time = '%.2f'%(head_total_time/head_count*1000)

else:

tps.head_time = 0

if get_count > 0:

tps.get_time = '%.2f'%(get_total_time/get_count*1000)

else:

tps.get_time = 0

if put_count > 0:

tps.put_time = '%.2f'%(put_total_time/put_count*1000)

else:

tps.put_time = 0

if post_count > 0:

tps.post_time = '%.2f'%(post_total_time/post_count*1000)

else:

tps.post_time = 0

if delete_count > 0:

tps.delete_time = '%.2f'%(delete_total_time/delete_count*1000)

else:

tps.delete_time = 0

tps.add_time = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))

db.session.add(tps)

db.session.commit()

else:

pass

else:

pass

else:

pass

except Exception as ex:

logger.exception("get_elk_log_json function execute exception:" + str(ex))

finally:

db.session.close()

db.session.remove()

#schedule tasks

def get_elklog_json_schl(executor):

"""

起線(xiàn)程執(zhí)行日志分析

:param executor:

:return:

"""

try:

el = ElkLog()

executor.submit(el.get_elk_log_json)

#threading.Thread(target=el.get_elk_log_json).start()

except Exception as ex:

logger.exception("get_elklog_json_schl function execute exception:" + str(ex))

class ElklogUnitAgnet(Agent):

def __init__(self, pidfile):

Agent.__init__(self, pidfile)

def run(self):

try:

sys.stdout.flush()

hostname = socket.getfqdn()

hostip = socket.gethostbyname(hostname)

logger.info("hostname is {}, ip is {}".format(hostname, hostip))

#use schedule

with ThreadPoolExecutor(config.thread_workers) as executor:

schedule.every(config.upload_refresh).seconds.do(get_elklog_json_schl,executor)

schedule.run_all(0)

while True:

schedule.run_pending()

time.sleep(0.1)

except Exception as ex:

logger.exception("elk log agent run exception:" + str(ex))

def main():

agent = ElklogUnitAgnet(config.elklog_agnet_pfile)

try:

if len(sys.argv) == 3:

if 'elklog' == sys.argv[1]:

if 'start' == sys.argv[2]:

agent.start()

if 'stop' == sys.argv[2]:

agent.stop()

else:

print("Unknown command")

sys.exit(2)

else:

print("usage: %s" % (sys.argv[0],))

sys.exit(2)

except Exception as ex:

logger.exception("elk log process run exception:" + str(ex))

if __name__ == '__main__':

main()

###########################################################################################

更多查詢(xún)方式接口：

查詢(xún)一條記錄

curl -H "Content-Type: application/json" -X POST 'http://192.168.1.1:9200/swift-nginx-2018.08.31/_search?pretty' -d '{"query": { "match_all": {} },"size": 1}'

查詢(xún)offset為20的記錄

curl -H "Content-Type: application/json" -X POST 'http://192.168.1.1:9200/swift-nginx-2018.08.31/_search?pretty' -d '{"query": { "match": { "offset": 20 } }}'

查詢(xún)結(jié)果只返回指定的字段

curl -H "Content-Type: application/json" -X POST 'http://192.168.1.1:9200/swift-nginx-2018.08.31/_search?pretty' -d '{"query": { "match_all": {} },"_source": ["host", "message"]}'

查詢(xún)結(jié)果排序

curl -H "Content-Type: application/json" -X POST 'http://192.168.1.1:9200/swift-nginx-2018.08.31/_search?pretty' -d '{"query": { "match_all": {} },"_source": ["offset","host", "message"]}，"sort": { "offset": { "order": "desc" } }'

返回從10開(kāi)始的10條記錄

curl -H "Content-Type: application/json" -X POST 'http://192.168.1.1:9200/swift-nginx-2018.08.31/_search?pretty' -d '{"query": { "match_all": {} },"from": 10,"size": 10,"_source": ["offset","host", "message"]}，"sort": { "offset": { "order": "desc" } }'

集群健康狀態(tài)查詢(xún)：

curl '192.168.1.1:9200/_cat/health?v'

查詢(xún)索引列表：

curl '192.168.1.1:9200/_cat/indices?v'

查詢(xún)集群節(jié)點(diǎn)列表：

curl '192.168.1.1:9200/_cat/nodes?v'

創(chuàng)建索引：

curl -XPUT '192.168.1.1:9200/test-index?pretty'

注意:索引名中不能使用大寫(xiě),否則會(huì)報(bào)錯(cuò):

Could not index event to Elasticsearch. ? ??"reason"=>"Invalid index name [iTech-swift-proxy-2018.11.08], must be lowercase",

刪除索引：

curl -XDELETE '192.168.1.1:9200/test-index?pretty'

向索引中插入數(shù)據(jù)：

curl -XPUT '192.168.1.1:9200/test-index/<_type>/<_id>?pretty' -d '{"name": "test name"}'

獲取插入的數(shù)據(jù)：

curl -XGET '192.168.1.1:9200/test-index/<_type>/<_id>?pretty'

(<_type>和<_id>用實(shí)際需要插入的屬性名和id值替換)

更新數(shù)據(jù)：

curl -XPOST '192.168.1.1:9200/test-index/<_type>/<_id>/_update?pretty' -d '{"doc": { "name": "test2 Name" }}'

刪除數(shù)據(jù)：

curl -XDELETE '192.168.1.1:9200/test-index/<_type>/<_id>?pretty'

查詢(xún)數(shù)據(jù)：

curl -XPOST '192.168.1.1:9200/test-index/_search?pretty' -d '{"query": { "match_all": {} }}'

如果有一臺(tái)elasticsearch磁盤(pán)空間不足，將會(huì)導(dǎo)致index變成readonly狀態(tài)，此時(shí)擴(kuò)容后需要用以下命令修改其狀態(tài)，恢復(fù)正常：

curl -XPUT -H "Content-Type: application/json" http://10.202.233.78:9200/_all/_settings -d '{"index.blocks.read_only_allow_delete": null}'

全部查詢(xún)接口實(shí)例請(qǐng)參考：

https://www.cnblogs.com/pilihaotian/p/5830754.html

創(chuàng)作挑戰(zhàn)賽新人創(chuàng)作獎(jiǎng)勵(lì)來(lái)咯，堅(jiān)持創(chuàng)作打卡瓜分現(xiàn)金大獎(jiǎng)

總結(jié)

以上是生活随笔為你收集整理的pythonsearch结果_python 查询Elasticsearch的小例子的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。