如何高效管理和监控 Elasticsearch 别名及索引？

铭毅天下

发布于 2024-06-18 16:32:00

1150

发布于 2024-06-18 16:32:00

文章被收录于专栏：铭毅天下铭毅天下

0、引言

在 Elasticsearch 项目中，管理和监控索引是开发者的一项重要任务。

尤其是当我们需要在项目的管理部分展示索引和别名的统计信息时，了解如何有效地列出这些别名和索引显得尤为重要。

本篇博客将介绍几种在 Elasticsearch 中列出别名和索引的方法，并展示如何将这些信息集成到应用程序中，实现自动化监控。

1、为什么需要列出别名和索引？

在实际项目中，例如一个社交网络应用，我们可能会使用 Elasticsearch 进行全文搜索，并为不同的数据设置别名。

通过别名，我们可以方便地管理版本和数据流。

别名具体应用场景概括如下表所示：

应用场景	描述
索引生命周期管理	- 滚动更新：在更新索引时，创建新索引，将数据写入新索引，完成后将别名切换到新索引。 - 版本控制：为每个索引版本设置别名，使用别名指向当前版本。 - 数据归档：使用别名管理归档索引，方便查询历史数据。
数据分区	- 时间分区：按时间创建索引，使用别名如 logs。 - 地理分区：按地理位置分区，使用别名如 data。
数据隔离	- 多租户：为每个租户创建单独的索引，使用别名进行数据隔离和查询。
零停机部署	- 零停机时间部署：在新版本准备就绪后，将别名从旧索引切换到新索引，实现无缝切换。
实时索引与批量索引	- 实时数据索引：为实时数据创建索引，使用别名进行查询。 - 批量数据索引：为批量数据创建索引，使用别名进行查询。
动态索引切换	- 热数据与冷数据：将热数据和冷数据分别存储在不同索引中，并使用别名动态切换。
数据合并	- 数据迁移：将多个小索引合并为一个大索引，使用别名进行平滑过渡。
索引监控和告警	- 监控别名与索引数量：定期列出以 listing* 开头的别名和索引，监控业务逻辑。 - 告警通知：根据索引和别名数量，向企业微信、钉钉、邮件等发送消息进行提醒。
安全控制	- 访问控制：为不同用户或角色创建别名，限制访问权限。
数据同步与备份	- 跨集群复制：使用别名管理跨集群复制的源索引和目标索引。 - 数据备份：为备份数据创建索引和别名。
查询优化	- 查询分流：将不同类型的查询分流到不同的索引，使用别名进行统一查询。
多语言支持	- 多语言索引：为不同语言创建单独索引，使用别名进行多语言数据管理。

写入索引同时指定别名如下图所示。

在 Elasticsearch-head 插件中的别名显示如下图所示。

ElasticVue插件中别名显示如下图所示。

2、列出别名和索引的方法

2.1 方法一：使用 _cat/aliases 接口

这是最常用的方法。

通过 _cat/aliases 接口，我们可以列出所有别名及其对应的索引。

示例请求：

GET _cat/aliases/listing*?v=true&h=alias,index&s=alias:desc

此请求会返回以下结果：

alias           index
listings        listings-v1.0.3
listings        listings-v1.0.0
listings        listings-v1.0.4
listings        listings-v1.0.2
listings        listings-v1.0.1
listing-changes listing-changes-2024-05
listing-changes listing-changes-2024-04

关键参数解释：

?v=true：开启详细模式，显示响应的头部信息。
?s=alias:desc：按照 alias 字段降序排列。
?h=alias,index：过滤显示的列，只显示 alias 和 index 列。

获取 JSON 格式的结果：

为了方便解析，我们可以将结果格式化为 JSON：

GET _cat/aliases?v=true&s=alias:desc&h=alias,index&format=json

返回示例：

[
  {
    "alias": "listings",
    "index": "listings-v1.0.2"
  },
  {
    "alias": "listings",
    "index": "listings-v1.0.1"
  },
  {
    "alias": "listings",
    "index": "listings-v1.0.0"
  },
  {
    "alias": "listings",
    "index": "listings-v1.0.4"
  },
  {
    "alias": "listings",
    "index": "listings-v1.0.3"
  },
  {
    "alias": "listing-changes",
    "index": "listing-changes-2024-05"
  },
  {
    "alias": "listing-changes",
    "index": "listing-changes-2024-04"
  },
  {
    "alias": "fscrawler",
    "index": "test_docs"
  },
  {
    "alias": "fscrawler",
    "index": "fs_job_2024"
  }
]

2.2 方法二：使用 _aliases 接口

该接口返回的结果更加详细，包含所有索引及其别名信息。适合在应用中使用。

示例请求：

GET _aliases

返回示例：

{
  "listings-v1.0.4": {
    "aliases": {
      "listings": {}
    }
  },
  "listing-changes-2024-05": {
    "aliases": {
      "listing-changes": {}
    }
  },
  ...
}

2.2.1 过滤特定别名的索引

使用特定别名过滤：

GET /_alias/listings

这将返回所有与别名 listings 相关联的索引。

2.2.2 过滤符合特定模式的索引

使用通配符过滤：

GET /_alias/listing*

这将返回所有别名以 listing 开头的索引。

2.2.3 过滤包含特定别名的索引

使用 filter_path 参数过滤只显示特定别名的索引：

GET /_aliases?filter_path=**.listings

这将返回只包含 listings 别名的索引。

GET /_aliases?filter_path=**.listings 是一个用于过滤 Elasticsearch 响应的查询语句。

具体含义如下：

GET /_aliases: 这个端点用于获取集群中所有索引的别名信息。
filter_path 参数: 这个参数用于过滤 Elasticsearch 响应的数据路径，只返回匹配指定路径的数据。它允许我们精确控制返回的 JSON 结构，减少不必要的数据传输和解析。
**.listings 的含义是：
**: 双星号（**）表示递归地匹配所有层级的路径。它可以匹配零个或多个层级的路径。
.listings: 表示包含 listings 关键字的路径。

结合在一起，filter_path=**.listings 的含义是：

过滤响应，只返回包含 listings 别名的索引信息。由于 ** 可以匹配所有层级，所以会在返回的 JSON 结构中递归地查找并返回所有路径中包含 listings 的部分。

如果我们要获取别名为：listing-changes 的索引-别名列表信息。

命令行为：


GET /_aliases?filter_path=**.listing-changes

2.3 方法三：使用 _cluster/state 接口

该方法可以获取集群状态的内部表示，适用于调试或诊断。

示例请求：

GET _cluster/state?filter_path=metadata.indices.*.aliases

返回示例：

{
  "metadata": {
    "indices": {
      "listings-v1.0.4": {
        "aliases": {
          "listings": {}
        }
      },
      ...
    }
  }
}

3、别名定时邮件发送获取

在管理页面展示索引和别名：通过上述接口获取数据，在管理页面上展示当前的索引和别名情况，方便开发者进行监控和管理。

自动化消息通知：定期检查别名和索引的数量，根据情况发送消息到企业微信、钉钉、邮件等通知平台，及时提醒相关人员进行处理。

示例：通过定时任务发送邮件

以下是一个简单的 Python 脚本示例，展示如何通过定时任务定期向指定邮箱发送 Elasticsearch 别名信息。

3.1 安装依赖

首先，安装所需的 Python 库：

pip install requests schedule smtplib

3.2 Python 脚本实现

import requests
import json
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import schedule
import time

# Elasticsearch 配置
ES_HOST = 'https://172.21.0.14:9200'
ALIAS_ENDPOINT = '/_cat/aliases?v=true&h=alias,index&format=json'
ES_USERNAME = 'elastic'
ES_PASSWORD = 'changeme'

# 163 邮箱配置
SMTP_SERVER = 'smtp.163.com'
SMTP_PORT = 465
EMAIL_USERNAME = 'changeme@163.com'
EMAIL_PASSWORD = 'changeme'
EMAIL_TO = 'changeme@163.com'

def get_aliases():
    url = f"{ES_HOST}{ALIAS_ENDPOINT}"
    response = requests.get(url, auth=(ES_USERNAME, ES_PASSWORD), verify=False)
    if response.status_code == 200:
        return response.json()
    else:
        print(f"Failed to fetch aliases: {response.status_code}, {response.text}")
        return []

def send_email(subject, body):
    msg = MIMEMultipart()
    msg['From'] = EMAIL_USERNAME
    msg['To'] = EMAIL_TO
    msg['Subject'] = subject

    msg.attach(MIMEText(body, 'plain'))
    text = msg.as_string()

    try:
        with smtplib.SMTP_SSL(SMTP_SERVER, SMTP_PORT) as server:
            server.login(EMAIL_USERNAME, EMAIL_PASSWORD)
            server.sendmail(EMAIL_USERNAME, EMAIL_TO, text)
        print("Email sent successfully")
    except Exception as e:
        print(f"Failed to send email: {e}")

def main():
    aliases = get_aliases()
    if aliases:
        subject = "Elasticsearch 别名和索引统计信息"
        body = json.dumps(aliases, indent=4, ensure_ascii=False)
        send_email(subject, body)

if __name__ == "__main__":
    main()

# def job():
#     aliases = get_aliases()
#     if aliases:
#         subject = "Elasticsearch 别名和索引统计信息"
#         body = json.dumps(aliases, indent=4, ensure_ascii=False)
#         send_email(subject, body)

# # 定时任务
# schedule.every().day.at("08:00").do(job)
#
# while True:
#     schedule.run_pending()
#     time.sleep(1)

3.3 结果展示