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。

相关文章

MySQL逻辑删除与唯一索引冲突解决方案

MySQL逻辑删除与唯一索引冲突解决方案

《MySQL逻辑删除与唯一索引冲突解决方案》本文探讨MySQL逻辑删除与唯一索引冲突问题,提出四种解决方案：复合索引+时间戳、修改唯一字段、历史表、业务层校验,推荐方案1和方案3,适用于不同场景,感兴... 目录问题背景问题复现解决方案解决方案1.复合唯一索引 + 时间戳删除字段解决方案2：删除后修改唯一字
阅读更多...
IntelliJ IDEA2025创建SpringBoot项目的实现步骤

IntelliJ IDEA2025创建SpringBoot项目的实现步骤

《IntelliJIDEA2025创建SpringBoot项目的实现步骤》本文主要介绍了IntelliJIDEA2025创建SpringBoot项目的实现步骤,文中通过示例代码介绍的非常详细,对大家... 目录一、创建 Spring Boot 项目1. 新建项目2. 基础配置3. 选择依赖4. 生成项目5.
阅读更多...
Linux线程之线程的创建、属性、回收、退出、取消方式

Linux线程之线程的创建、属性、回收、退出、取消方式

《Linux线程之线程的创建、属性、回收、退出、取消方式》文章总结了线程管理核心知识：线程号唯一、创建方式、属性设置（如分离状态与栈大小）、回收机制（join/detach）、退出方法（返回/pthr... 目录1. 线程号2. 线程的创建3. 线程属性4. 线程的回收5. 线程的退出6. 线程的取消7.
阅读更多...
创建Java keystore文件的完整指南及详细步骤

创建Java keystore文件的完整指南及详细步骤

《创建Javakeystore文件的完整指南及详细步骤》本文详解Java中keystore的创建与配置,涵盖私钥管理、自签名与CA证书生成、SSL/TLS应用,强调安全存储及验证机制,确保通信加密和... 目录1. 秘密键（私钥）的理解与管理私钥的定义与重要性私钥的管理策略私钥的生成与存储2. 证书的创建与
阅读更多...
浅谈mysql的not exists走不走索引

浅谈mysql的not exists走不走索引

《浅谈mysql的notexists走不走索引》在MySQL中,​NOTEXISTS子句是否使用索引取决于子查询中关联字段是否建立了合适的索引,下面就来介绍一下mysql的notexists走不走索... 在mysql中，​NOT EXISTS子句是否使用索引取决于子查询中关联字段是否建立了合适的索引。以下
阅读更多...
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
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2