如何用Python检查AWS上Elasticsearch的可用磁盘空间

前置准备 #

先安装所需依赖包：

pip install elasticsearch requests-aws4auth boto3

完整代码示例 #

import boto3
from elasticsearch import Elasticsearch, RequestsHttpConnection
from requests_aws4auth import AWS4Auth

# 配置参数，根据你的实际环境修改
AWS_REGION = 'us-east-1'  # 替换为你的ES集群所在区域
ES_ENDPOINT = 'https://your-es-domain-endpoint.us-east-1.es.amazonaws.com'  # 替换为你的ES域端点
DISK_THRESHOLD_PERCENT = 85  # 磁盘使用率告警阈值，超过此值触发告警

# 获取AWS认证凭证
credentials = boto3.Session().get_credentials()
aws_auth = AWS4Auth(credentials.access_key, credentials.secret_key, AWS_REGION, 'es', session_token=credentials.token)

# 创建Elasticsearch客户端实例
es = Elasticsearch(
    hosts=[{'host': ES_ENDPOINT.replace('https://', ''), 'port': 443}],
    http_auth=aws_auth,
    use_ssl=True,
    verify_certs=True,
    connection_class=RequestsHttpConnection
)

def check_disk_usage():
    # 查询集群磁盘使用统计数据
    cluster_stats = es.cluster.stats()
    nodes_fs = cluster_stats['nodes']['fs']
    
    # 转换磁盘空间单位为GB
    total_disk_gb = nodes_fs['total_in_bytes'] / (1024**3)
    used_disk_gb = nodes_fs['used_in_bytes'] / (1024**3)
    used_percent = (used_disk_gb / total_disk_gb) * 100
    
    # 打印磁盘使用详情
    print(f"集群总磁盘空间: {total_disk_gb:.2f} GB")
    print(f"已使用磁盘空间: {used_disk_gb:.2f} GB")
    print(f"当前磁盘使用率: {used_percent:.2f}%")
    
    # 判断是否触发告警
    if used_percent >= DISK_THRESHOLD_PERCENT:
        print("\n⚠️ 告警：磁盘使用率超过设定阈值！")
        # 此处可扩展告警方式，比如调用AWS SNS发通知、发送邮件等
    else:
        print("\n✅ 磁盘使用率处于正常范围")

if __name__ == '__main__':
    try:
        check_disk_usage()
    except Exception as e:
        print(f"脚本执行出错: {str(e)}")

关键说明 #

• AWS认证：脚本通过boto3自动获取本地AWS凭证（比如~/.aws/credentials文件或EC2实例角色权限），无需手动输入密钥。
• ES端点获取：在AWS控制台的Elasticsearch服务页面，找到你的域名，复制"域端点"地址即可。
• 权限要求：确保你的AWS账号/角色拥有es:ESHttpGet权限，否则会出现访问被拒绝的错误。
• 告警扩展：如果需要更正式的告警，可将打印告警的代码替换为调用AWS SNS发送消息、调用邮件API发送告警邮件等逻辑。

内容的提问来源于stack exchange，提问作者shadowskillz