elk笔记22.2--通过api快速创建索引

2024-05-30 23:48
文章标签 快速 笔记 创建 索引 api elk 22.2

本文主要是介绍elk笔记22.2--通过api快速创建索引，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

elk笔记22.2--通过api快速创建索引

  • 1 简介
  • 2 功能实现
    • 2.1 源码
    • 2.2 测试
  • 3 注意事项
  • 说明

1 简介

本文基于 elk笔记22–通过api快速创建索引 继续通过 api 快速创建索引。本节将追加一个db模块存储索引，并添加通过flask程序提供对外的api，后续会在此基础之上新增一个简单的UI界面，同时会追加一个索引定期删除功能。

2 功能实现

本文前置条件需要安装一套elk实例和mysql，具体安装方法可以参考笔者博文：
elk笔记1–搭建elk集群
elk笔记2–使用docker启一套elk实例
docker笔记5–配置MySQL

本文设计的主要模块包括如下内容：

  1. config.py，记录常见es、kibana、mysql 等基础配置
  2. db_helper.py，封装常见的mysql 操作
  3. es_helper.py，封装常见的elk操作
  4. create_index.py，flask主程序，用于实现创建和查看索引的主逻辑

2.1 源码

config.py

#!/usr/bin/python3
# -*- coding:utf-8 -*-ES_URL = "http://127.0.0.1:9200"
KIBANA_URL = "http://127.0.0.1:5601"
ES_USER = "elastic"
ES_PWD = "elastic"
KIBANA_USER = "elastic"
KIBANA_PWD = "elastic"SQL_HOST = '127.0.0.1'
SQL_USER = 'root'
SQL_PWD = '111111'
SQL_DBNAME = 'bigdata_sre_log'

db_helper.py

#!/usr/bin/python3
# -*- coding:utf-8 -*-import pymysql
import tracebackclass MysqlHelper:def __init__(self, sql_host, sql_user, sql_pwd, db_name):self.host = sql_hostself.usr = sql_userself.pwd = sql_pwdself.dbname = db_nameself.port = 3306self.charset = 'utf8'self.db = pymysql.connect(host=self.host, user=self.usr, passwd=self.pwd, db=self.dbname, charset=self.charset)def close_db(self):self.db.close()def show_version(self):# 使用 cursor() 方法创建一个游标对象 cursorcursor = self.db.cursor()# 使用 execute()  方法执行 SQL 查询cursor.execute("SELECT VERSION()")# 使用 fetchone() 方法获取单条数据.data = cursor.fetchone()print(f"Database version : {data}")def get_indices_list(self):ret_list = []instance = Nonesql = f"select number, cluster, index_name,save_days, ilp_name, datetime from es_indices"try:cursor = self.db.cursor()cursor.execute(sql)ret = cursor.fetchall()if ret is None:return []for item in ret:instance = {'number': item[0], 'cluster': item[1], 'index_name': item[2], 'save_days': item[3],'ilp_name': item[4], 'datetime': item[5].strftime("%Y-%m-%d %H:%M:%S")}ret_list.append(instance)print(f"get_indices_list succeed!")return True, ret_listexcept Exception as e:self.db.rollback()print(f"get_indices_list error:\n{e}, {traceback.print_exc()}")return False, f"get_indices_list error:\n{e}, {traceback.print_exc()}"def get_index_instance(self, cluster, index_name):sql = f"select number, cluster, index_name, save_days, ilp_name, datetime from es_indices " \f"where cluster='{cluster}' and index_name='{index_name}'"try:cursor = self.db.cursor()cursor.execute(sql)item = cursor.fetchone()if item is None:return False, f"no {cluster}/{index_name}"instance = {'number': item[0], 'cluster': item[1], 'index_name': item[2], 'save_days': item[3],'ilp_name': item[4], 'datetime': item[5].strftime("%Y-%m-%d %H:%M:%S")}print(f"get_index_instance succeed!")return True, instanceexcept Exception as e:print(f"get_index_instance error:\n{e}, {traceback.print_exc()}")return False, f"get_index_instance error:\n{e}, {traceback.print_exc()}"def judge_index_instance_exist(self, cluster, index_name):tag, ret = self.get_index_instance(cluster, index_name)if tag:return Truereturn Falsedef add_index_instance(self, instance):if self.judge_index_instance_exist(instance['cluster'], instance['index_name']):return False, f"{instance['cluster']}/{instance['index_name']} exist, please do not repeat!"cursor = self.db.cursor()sql = "INSERT INTO es_indices(cluster, index_name, save_days, ilp_name) VALUES ( %s, %s, %s, %s)"try:cursor.execute(sql, [instance['cluster'], instance['index_name'], instance['save_days'],instance['ilp_name']])self.db.commit()return True, f"add_index_instance succeed!"except Exception as e:self.db.rollback()print(f"add_index_instance error:\n{e}, {traceback.print_exc()}")return False, f"add_index_instance error:\n{e}, {traceback.print_exc()}"if __name__ == "__main__":sql_host = '127.0.0.1'sql_user = 'root'sql_pwd = '111111'sql_dbname = 'bigdata_sre_log'sql = MysqlHelper(sql_host, sql_user, sql_pwd, sql_dbname)sql.show_version()instance = {"index_name": "test-elk", "cluster": "sre-elk","save_days": 10, "ilp_name": "ilp-default"}print(sql.add_index_instance(instance))print(sql.get_indices_list())sql.close_db()

es_helper.py

#!/usr/bin/python
# -*- coding:utf-8 -*-import requests
from base64 import b64encodedef get_base64_str(str_text):str_user_pwd = b64encode(str_text.encode('ascii'))return str_user_pwd.decode('ascii')class ElasticHelper:def __init__(self, es_url, es_user, es_pwd, es_index_name, es_ilp_name):self.url = es_urlself.user = es_userself.pwd = es_pwdself.index_name = es_index_nameself.ilp_name = es_ilp_namedef get_index_template(self):print(f"get_index_template: {self.index_name}")url_full = self.url + f"/_template/{self.index_name}"re = requests.get(url_full, auth=(self.user, self.pwd))if re.status_code == 200:return re.json()else:print("request error, not 200")return Nonedef put_index_template(self):body = {'order': 0,'index_patterns': [f"{self.index_name}-*"],'settings': {'index': {'number_of_shards': '2','number_of_replicas': '0'}},'mappings': {"properties": {"@timestamp": {"type": "date"}}},'aliases': {}}url_full = self.url + f"/_template/{self.index_name}"re = requests.put(url_full, json=body, auth=(self.user, self.pwd))if re.status_code == 200:print(re.json())else:print("request error, not 200")def update_index_template_with_ilp(self):ret = self.get_index_template()if ret is None:print(f"index template {self.index_name}-* not exist, please create one!")body = ret[self.index_name]body['settings']['index']['lifecycle'] = {'name': f'{self.ilp_name}','rollover_alias': f'{self.index_name}_write'}re = requests.post(f"{self.url}/_template/{self.index_name}", json=body, auth=('elastic', 'elastic'))if re.status_code == 200:print(f"put_index_template(index={self.index_name},ilp={self.ilp_name}) succeed:\ninfo {re.status_code}, "f"{re.text}")else:print(f"put_index_template(index={self.index_name},ilp={self.ilp_name}) failed:\ninfo {re.status_code}, "f"{re.text}")def create_rollover_index(self):body = {"aliases": {f"{self.index_name}_write": {}}}re = requests.put(url=self.url+f"/%3C{self.index_name}-%7Bnow%2Fd%7D-000001%3E", json=body, auth=(self.user,self.pwd))if re.status_code == 200:print(re.json())else:print(f"create rollover index {self.index_name} failed!\n{re.status_code},\n{re.text}")def judge_alias_exist(self):re = requests.get(f"{self.url}/_cat/aliases", auth=('elastic', 'elastic'))if re.status_code == 200:ret_str = re.textif ret_str.find(f"{self.index_name}_write") != -1:return Trueelse:Falseelse:print(f"judge_alias_exist {self.index_name}_write failed!\n{re.status_code},\n{re.text}")return Falseclass KibanaHelper:def __init__(self, kibana_url, kibana_user, kibana_pwd, kibana_index_pattern_name):self.url = kibana_urlself.user = kibana_userself.pwd = kibana_pwdself.index_pattern_name = kibana_index_pattern_namedef get_index_pattern_id(self):url = f'{self.url}/api/saved_objects/_find?fields=title&fields=type&per_page=10000&type=index-pattern'headers = {'Authorization': 'Basic ' + get_base64_str(self.user + ':' + self.pwd)}ret = requests.get(url=url, headers=headers)for item in (ret.json())['saved_objects']:if item['attributes']['title'] == self.index_pattern_name:return item['id']return Nonedef get_kibana_index_pattern(self):print(f"get_kibana_index_pattern {index_pattern_name}-*")headers = {'Authorization': 'Basic ' + get_base64_str(self.user + ':' + self.pwd), 'kbn-xsrf': 'reporting'}id = self.get_index_pattern_id()url_full = self.url + f"/api/saved_objects/_bulk_get"body = [{"id": id, "type": "index-pattern"}]re = requests.post(url=url_full, json=body, headers=headers)if re.status_code == 200:print(re.text)else:print(f"error:\n {re.text}")def create_kibana_index_pattern(self):print(f"index_pattern={self.index_pattern_name}")ret = self.get_index_pattern_id()if ret is not None:print(f"{self.index_pattern_name} has been existed, please do not repeat")exit(0)headers = {'Authorization': 'Basic ' + get_base64_str(self.user + ':' + self.pwd), 'kbn-xsrf': 'reporting'}url_full = self.url + "/api/saved_objects/index-pattern"body = {"attributes": {"title": self.index_pattern_name,"timeFieldName": "@timestamp"}}re = requests.post(url_full, json=body, headers=headers)if re.status_code == 200:print(re.text)else:print(f"error:\n {re.text}")def delete_index_pattern_id(self, index_pattern_id):print(f"delete index_pattern_id={index_pattern_id}")headers = {'Authorization': 'Basic ' + get_base64_str(self.user + ':' + self.pwd), 'kbn-xsrf': 'reporting'}url_full = self.url + f"/api/saved_objects/index-pattern/{index_pattern_id}?force=false"re = requests.delete(url_full, headers=headers)if re.status_code == 200:print(f"delete_index_pattern_id({index_pattern_id}) succeed:\ninfo {re.status_code}, {re.text}")else:print(f"delete_index_pattern_id({index_pattern_id}) error:\ninfo {re.status_code}, {re.text}")def get_ilm(self):url_full = self.url+"/api/index_lifecycle_management/policies?withIndices=true"headers = {'Authorization': 'Basic ' + get_base64_str(self.user + ':' + self.pwd), 'kbn-xsrf': 'reporting'}re = requests.get(url_full, headers=headers)if re.status_code == 200:print(f"get_ilm succeed")return re.json()else:print(f"get_ilm failed\ninfo {re.status_code},{re.text}")return Nonedef judge_ilp_exist(self, ilp_name):ilp_list = self.get_ilm()if ilp_list is None:return Falseelse:for item in ilp_list:if ilp_name == item["name"]:return Truereturn Falseif __name__ == "__main__":es_url = "http://127.0.0.1:9200"es_user = "elastic"es_pwd = "elastic"kibana_url = "http://127.0.0.1:5601"kibana_user = "elastic"kibana_pwd = "elastic"index_name = "test-elk"index_pattern_name = index_name + "-*"ilp_name = "ilp-default"es = ElasticHelper(es_url, es_user, es_pwd, index_name, ilp_name)kibana = KibanaHelper(kibana_url, kibana_user, kibana_pwd, index_pattern_name)es.put_index_template()if not kibana.judge_ilp_exist(ilp_name):print(f"ilp_name={ilp_name} not exist, please change another one")exit(0)es.update_index_template_with_ilp()if es.judge_alias_exist():print(f"alias_name={index_name}_write exists, please change another one")exit(0)es.create_rollover_index()kibana.create_kibana_index_pattern()

create_index.py

#!/usr/bin/python3
# -*- coding:utf-8 -*-'''
add_index
list/<cluster>/<index_name>
'''import json
from flask import Flask
from flask import requestfrom db_helper import MysqlHelper
from es_helper import ElasticHelper, KibanaHelper
from config import *app = Flask(__name__)@app.route('/')
def voice_root():return 'Hello, this web service is for create es index!\n'@app.route('/add_index/', methods=['GET', 'POST'])
def add_index():if request.method == 'POST':post_data = request.datapost_dict = json.loads(post_data)# print(f"add_index post_data:\n{post_dict}")# post_json_str = json.dumps(post_dict)# print(f"add_index post_json_str:\n{post_json_str}")instance = {'index_name': post_dict['index_name'], "cluster": post_dict['cluster'],"save_days": post_dict['save_days'], "ilp_name": post_dict['ilp_name']}sql = MysqlHelper(SQL_HOST, SQL_USER, SQL_PWD, SQL_DBNAME)ret = sql.judge_index_instance_exist(post_dict['cluster'], post_dict['index_name'])if ret:return f"index {post_dict['cluster']}/{post_dict['index_name']} exists, please do not repeat"index_name = post_dict['index_name']index_pattern_name = index_name + "-*"ilp_name = post_dict['ilp_name']es = ElasticHelper(ES_URL, ES_USER, ES_PWD, index_name, ilp_name)kibana = KibanaHelper(KIBANA_URL, KIBANA_USER, KIBANA_PWD, index_pattern_name)if not kibana.judge_ilp_exist(ilp_name):print(f"ilp_name={ilp_name} not exist, please change another one")return f"ilp_name={ilp_name} not exist, please change another one"es.put_index_template()es.update_index_template_with_ilp()if es.judge_alias_exist():print(f"alias_name={index_name}_write exists, please change another one")return f"alias_name={index_name}_write exists, please change another one"es.create_rollover_index()kibana.create_kibana_index_pattern()tag, ret = sql.add_index_instance(instance)return f"{tag}\n{ret}"else:return 'add_index, please post data!\n'@app.route('/list/')
@app.route('/list/<cluster>/<index_name>/')
def list_index(cluster=None, index_name=None):print(f"1,{cluster},2{index_name}")if (cluster is None) and (index_name is None):sql = MysqlHelper(SQL_HOST, SQL_USER, SQL_PWD, SQL_DBNAME)tag, ret = sql.get_indices_list()sql.close_db()if tag:print(ret)return json.dumps(ret)else:return retelse:sql = MysqlHelper(SQL_HOST, SQL_USER, SQL_PWD, SQL_DBNAME)tag, ret = sql.get_index_instance(cluster, index_name)sql.close_db()if tag:print(ret)return json.dumps(ret)else:return retif __name__ == '__main__':app.run(host="0.0.0.0", port=8080)

2.2 测试

新建数据库:

数据库名称：bigdata_sre_log
表名称：es_indices
字段：
number INT，自增
cluster VARCHAR
index_name VARCHAR
save_days INT
ilp_name VARCHAR
datetime DATETIME, 默认值 CURRENT_TIMESTAMP

post 如下内容：

POST http://127.0.0.1:8080/add_index
{"index_name": "test-elk", "cluster": "sre-elk", "save_days": 10, "ilp_name": "ilp-default"}

在这里插入图片描述
kibana 查看数据(先写入1条数据):

POST test-elk_write/_doc
{"@timestamp": "2021-08-30T23:00:00+08:00","name": "test001"
}

在这里插入图片描述
调试结果(正常输入相关日志)：
在这里插入图片描述
查看索引：
GET http://127.0.0.1:8080/list/
在这里插入图片描述

3 注意事项

  1. 本文只提供了简单的索引增加和查找功能，实际中最好增加配置修改和索引删除功能。

说明

软件环境：
ubuntu 20.04 desktop
elk 7.10.0(docker)
python 3.8
mysql 8.0(docker)
参考文档：
1 elasticsearch/reference/current/index.html
2 kibana/current/index.html
3 docker笔记5–配置MySQL

这篇关于elk笔记22.2--通过api快速创建索引的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1017093。

相关文章

Knife4j+Axios+Redis前后端分离架构下的 API 管理与会话方案(最新推荐)

Knife4j+Axios+Redis前后端分离架构下的 API 管理与会话方案(最新推荐)

《Knife4j+Axios+Redis前后端分离架构下的API管理与会话方案(最新推荐)》本文主要介绍了Swagger与Knife4j的配置要点、前后端对接方法以及分布式Session实现原理,... 目录一、Swagger 与 Knife4j 的深度理解及配置要点Knife4j 配置关键要点1.Spri
阅读更多...
MySQL之InnoDB存储引擎中的索引用法及说明

MySQL之InnoDB存储引擎中的索引用法及说明

《MySQL之InnoDB存储引擎中的索引用法及说明》：本文主要介绍MySQL之InnoDB存储引擎中的索引用法及说明,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐... 目录1、背景2、准备3、正篇【1】存储用户记录的数据页【2】存储目录项记录的数据页【3】聚簇索引【4】二
阅读更多...
全面解析MySQL索引长度限制问题与解决方案

全面解析MySQL索引长度限制问题与解决方案

《全面解析MySQL索引长度限制问题与解决方案》MySQL对索引长度设限是为了保持高效的数据检索性能,这个限制不是MySQL的缺陷,而是数据库设计中的权衡结果,下面我们就来看看如何解决这一问题吧... 目录引言：为什么会有索引键长度问题？一、问题根源深度解析mysql索引长度限制原理实际场景示例二、五大解决
阅读更多...
python如何创建等差数列

python如何创建等差数列

《python如何创建等差数列》：本文主要介绍python如何创建等差数列的问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录python创建等差数列例题运行代码回车输出结果总结python创建等差数列import numpy as np x=int(in
阅读更多...
MySQL中的索引结构和分类实战案例详解

MySQL中的索引结构和分类实战案例详解

《MySQL中的索引结构和分类实战案例详解》本文详解MySQL索引结构与分类,涵盖B树、B+树、哈希及全文索引,分析其原理与优劣势,并结合实战案例探讨创建、管理及优化技巧,助力提升查询性能,感兴趣的朋... 目录一、索引概述1.1 索引的定义与作用1.2 索引的基本原理二、索引结构详解2.1 B树索引2.2
阅读更多...
Linux如何快速检查服务器的硬件配置和性能指标

Linux如何快速检查服务器的硬件配置和性能指标

《Linux如何快速检查服务器的硬件配置和性能指标》在运维和开发工作中,我们经常需要快速检查Linux服务器的硬件配置和性能指标,本文将以CentOS为例,介绍如何通过命令行快速获取这些关键信息,... 目录引言一、查询CPU核心数编程（几C？）1. 使用 nproc（最简单）2. 使用 lscpu（详细信
阅读更多...
python3如何找到字典的下标index、获取list中指定元素的位置索引

python3如何找到字典的下标index、获取list中指定元素的位置索引

《python3如何找到字典的下标index、获取list中指定元素的位置索引》：本文主要介绍python3如何找到字典的下标index、获取list中指定元素的位置索引问题,具有很好的参考价值,... 目录enumerate()找到字典的下标 index获取list中指定元素的位置索引总结enumerat
阅读更多...
怎么用idea创建一个SpringBoot项目

怎么用idea创建一个SpringBoot项目

《怎么用idea创建一个SpringBoot项目》本文介绍了在IDEA中创建SpringBoot项目的步骤,包括环境准备（JDK1.8+、Maven3.2.5+）、使用SpringInitializr... 目录如何在idea中创建一个SpringBoot项目环境准备1.1打开IDEA，点击New新建一个项
阅读更多...
从入门到精通MySQL 数据库索引(实战案例)

从入门到精通MySQL 数据库索引(实战案例)

《从入门到精通MySQL数据库索引(实战案例)》索引是数据库的目录,提升查询速度,主要类型包括BTree、Hash、全文、空间索引,需根据场景选择,建议用于高频查询、关联字段、排序等,避免重复率高或... 目录一、索引是什么？能干嘛？核心作用：二、索引的 4 种主要类型（附通俗例子）1. BTree 索引（
阅读更多...
如何使用Maven创建web目录结构

如何使用Maven创建web目录结构

《如何使用Maven创建web目录结构》：本文主要介绍如何使用Maven创建web目录结构的问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录创建web工程第一步第二步第三步第四步第五步第六步第七步总结创建web工程第一步js通过Maven骨架创pytho
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2