当前位置：首页 > news >正文

Python脚本阿里云服务监控脚本

news 来源：原创 2024/9/21 8:32:06

文章目录

简介
环境准备
脚本概述
脚本实现
- 依赖库
- 配置阿里云认证信息
- 定义相关参数
- - 示例ECS信息
其他资源
- ECS 实例信息获取
- Redis 实例信息获取
- RDS 实例信息获取
- 其他服务信息获取
注意事项
结语
完整代码

简介

在云计算时代，管理云资源变得尤为重要，特别是在使用阿里云ECS（Elastic Compute Service）时，监控和管理云服务器的性能参数，如CPU、内存、磁盘使用率等，是确保业务稳定运行的关键。

环境准备

Python环境：确保你的系统中已安装Python。
阿里云SDK：通过pip安装阿里云SDK

脚本概述

脚本主要实现了以下功能：

ECS信息
Redis信息
RDS信息
DTS信息
域名信息
证书信息

脚本实现

依赖库

公共库

import datetime  # 处理时间和日期
import json  # 处理 JSON 格式的数据
from aliyunsdkcore import client  # 阿里云 API 的客户端

脚本全部所需要的库

import datetime
from HTMLTable import HTMLTable
import json
from aliyunsdkcore import client
from aliyunsdkecs.request.v20140526 import DescribeInstancesRequest, DescribeDisksRequest
from aliyunsdkcms.request.v20190101 import DescribeMetricTopRequest
from aliyunsdkecs.request.v20140526.DescribeDisksRequest import DescribeDisksRequest
from aliyunsdkr_kvstore.request.v20150101.DescribeInstancesOverviewRequest import DescribeInstancesOverviewRequest
from aliyunsdkr_kvstore.request.v20150101.DescribeHistoryMonitorValuesRequest import DescribeHistoryMonitorValuesRequest
from aliyunsdkrds.request.v20140815.DescribeDBInstancesRequest import DescribeDBInstancesRequest
from aliyunsdkrds.request.v20140815.DescribeDBInstanceAttributeRequest import DescribeDBInstanceAttributeRequest
from aliyunsdkdts.request.v20200101.DescribeSynchronizationJobsRequest import DescribeSynchronizationJobsRequest
from aliyunsdkdomain.request.v20180129.QueryDomainListRequest import QueryDomainListRequest
from aliyunsdkcas.request.v20200407.ListUserCertificateOrderRequest import ListUserCertificateOrderRequest

配置阿里云认证信息

这个是阿里云后台上，申请api 的key

clt = client.AcsClient("你的AccessKeyID", "你的AccessKeySecret", "可用区")
##举例
clt = client.AcsClient("xxxxxxxxxxxxxxxxx", "xxxxxxxxxxxxxxxxxxxxxx", "cn-beijing")

定义相关参数

# 脚本所需参数
# 设置页数
page_number = "1"
# 设置每页返回多少，默认为10条
strip_number = "100"
# 磁盘使用率阀值(%)
Disk_use_rate = "50"

示例ECS信息

from aliyunsdkecs.request.v20140526 import DescribeInstancesRequest, DescribeDisksRequest
import jsonstrip_number = "100"clt = client.AcsClient("xxxxxxxxxxxxxxxxx", "xxxxxxxxxxxxxxxxxxxxxx", "cn-beijing")# 列表---ECS列表
def get_sys_info():request = DescribeInstancesRequest()request.set_PageSize(strip_number)  # 设置每页返回多少，默认为10条request.set_accept_format("json")response = json.loads(clt.do_action(request))return response# 执行
if __name__ == "__main__":print(get_sys_info())

其他资源

ECS 实例信息获取

通过调用 DescribeInstancesRequest 接口，可以获取 ECS 实例的相关信息。

from aliyunsdkecs.request.v20140526.DescribeInstancesRequest import DescribeInstancesRequestdef get_sys_info():"""获取 ECS 实例列表"""request = DescribeInstancesRequest()request.set_PageSize(strip_number)  # 设置每页返回条数request.set_accept_format("json")  # 设置返回格式为 JSONresponse = json.loads(clt.do_action(request))return response

为了获取更详细的 ECS 实例信息，我们还需要获取磁盘信息：

from aliyunsdkecs.request.v20140526.DescribeDisksRequest import DescribeDisksRequestdef disk_info():"""获取 ECS 磁盘信息"""request = DescribeDisksRequest()request.set_PageSize(strip_number)request.set_accept_format("json")response = clt.do_action_with_exception(request)disks = json.loads(response).get("Disks").get("Disk")return disks

接下来，我们将综合 ECS 和磁盘信息，展示 ECS 的详细信息：

def ecs_information():"""获取并打印 ECS 详细信息"""ecs_info = []for file_info in get_sys_info():information = {"InstanceName": file_info["InstanceName"],"Status": file_info["Status"],"OSType": file_info["OSType"],"Cpu": file_info["Cpu"],"Memory": file_info["Memory"],"IpA": file_info["VpcAttributes"]["PrivateIpAddress"]["IpAddress"][0],"IpB": file_info["EipAddress"].get("IpAddress", "无"),"time": str(datetime.datetime.strptime(file_info["ExpiredTime"], "%Y-%m-%dT%H:%MZ")+ datetime.timedelta(hours=8)- datetime.timedelta(seconds=1)),}ecs_info.append(information)print(ecs_info)

Redis 实例信息获取

获取 Redis 实例信息的步骤类似于 ECS。首先通过 DescribeInstancesOverviewRequest 获取 Redis 实例列表：

from aliyunsdkr_kvstore.request.v20150101.DescribeInstancesOverviewRequest import DescribeInstancesOverviewRequestdef get_redis_info():"""获取 Redis 实例列表"""request = DescribeInstancesOverviewRequest()request.set_accept_format("json")response = json.loads(clt.do_action_with_exception(request)).get("Instances")return response

获取 Redis 实例详情：

def redis_info():"""获取并打印 Redis 详细信息"""redis_info = []for file_info in get_redis_info():information = {"InstanceName": file_info["InstanceName"],"Status": file_info["InstanceStatus"],"InstanceType": file_info["InstanceType"],"Capacity": file_info["Capacity"],"EngineVersion": file_info["EngineVersion"],"InstanceId": file_info["InstanceId"],"message": get_redis_memusage(file_info["InstanceId"]),"time": str(datetime.datetime.strptime(file_info["EndTime"], "%Y-%m-%dT%H:%M:%SZ")+ datetime.timedelta(hours=8)),}redis_info.append(information)print(redis_info)

获取 Redis 内存使用率：

from aliyunsdkr_kvstore.request.v20150101.DescribeHistoryMonitorValuesRequest import DescribeHistoryMonitorValuesRequestdef get_redis_memusage(instanceId):"""获取 Redis 内存使用情况"""end_time = datetime.datetime.utcnow()start_time = end_time - datetime.timedelta(minutes=1)start_time_str = start_time.strftime("%Y-%m-%dT%H:%M:%SZ")end_time_str = end_time.strftime("%Y-%m-%dT%H:%M:%SZ")request = DescribeHistoryMonitorValuesRequest()request.set_accept_format("json")request.set_InstanceId(instanceId)request.set_StartTime(start_time_str)request.set_EndTime(end_time_str)request.set_IntervalForHistory("60m")request.set_MonitorKeys("MemoryUsage,UsedMemory")response = json.loads(clt.do_action_with_exception(request))last_value = sorted(response["MonitorHistory"].items())[-1][1]used_memory_gb = float(last_value["UsedMemory"]) / (1024**3)return {"memoryUsage": f"{last_value['memoryUsage']}%","UsedMemoryGB": f"{used_memory_gb:.2f} GB",}

RDS 实例信息获取

类似于前面的 ECS 和 Redis，以下是获取 RDS 实例信息的代码：

from aliyunsdkrds.request.v20140815.DescribeDBInstancesRequest import DescribeDBInstancesRequestdef get_rds_info():"""获取 RDS 实例列表"""request = DescribeDBInstancesRequest()request.set_PageSize(strip_number)request.set_accept_format("json")response = json.loads(clt.do_action_with_exception(request)).get("Items").get("DBInstance")return response

获取 RDS 实例详情：

def rds_info():"""获取并打印 RDS 详细信息"""rds_info = []for file_info in get_rds_info():db_instance_description = file_info.get("DBInstanceDescription") or file_info.get("DBInstanceId", "未知")information = {"DBInstanceDescription": db_instance_description,"Status": file_info["DBInstanceStatus"],"Engine": file_info["Engine"],"EngineVersion": file_info["EngineVersion"],"DBInstanceId": file_info["DBInstanceId"],"DBInstanceType": file_info["DBInstanceType"],"PayType": file_info["PayType"],"usage": rds_info_usage(file_info["DBInstanceId"]),"time": str(datetime.datetime.strptime(file_info["ExpireTime"], "%Y-%m-%dT%H:%M:%SZ")+ datetime.timedelta(hours=8)),}rds_info.append(information)print(rds_info)

获取 RDS 使用情况：

from aliyunsdkrds.request.v20140815.DescribeDBInstanceAttributeRequest import DescribeDBInstanceAttributeRequestdef rds_info_usage(instanceId):"""获取 RDS 使用情况"""request = DescribeDBInstanceAttributeRequest()request.set_accept_format("json")request.set_DBInstanceId(instanceId)response = json.loads(clt.do_action_with_exception(request)).get("Items").get("DBInstanceAttribute")information = {}for file_info in response:information.update({"DBInstanceMemory": file_info["DBInstanceMemory"],"DBInstanceCPU": file_info["DBInstanceCPU"],"DBInstanceStorage": file_info["DBInstanceStorage"],"DBInstanceDiskUsed": file_info["DBInstanceDiskUsed"],})return information

其他服务信息获取

我们还可以通过类似的方式获取 DTS、Domain、CAS 服务的信息。这里只列出部分代码，更多细节可以参阅完整脚本。

from aliyunsdkdts.request.v20200101.DescribeSynchronizationJobsRequest import DescribeSynchronizationJobsRequestdef get_dts_info():"""获取 DTS 实例列表"""request = DescribeSynchronizationJobsRequest()request.set_PageSize(strip_number)request.set_accept_format("json")response = json.loads(clt.do_action_with_exception(request)).get("SynchronizationInstances")return response

注意事项

API选择：确保使用正确的API来获取所需数据
权限问题：确保你的阿里云账号有足够的权限去调用这些API。
性能考虑：对于大量ECS实例，应考虑分页处理和API请求频率限制

结语

本文介绍了一个使用Python和阿里云SDK编写的脚本，用于监控和管理阿里云ECS实例的磁盘使用率。该脚本的灵活性使其可以应用于多种场景，如资源监控、自动化运维等。希望这篇博客对您有所帮助。如果有任何问题或改进建议，欢迎交流讨论。

完整代码

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2021/9/8 16:52
# @Time : 2021/7/9 11:23
# @Time : 2022/10/36 11:00import datetime
from HTMLTable import HTMLTable
import json
from aliyunsdkcore import client# from aliyunsdkecs.request.v20140526 import DescribeInstancesRequest
from aliyunsdkecs.request.v20140526.DescribeInstancesRequest import (DescribeInstancesRequest,
)
from aliyunsdkcms.request.v20190101.DescribeMetricTopRequest import (DescribeMetricTopRequest,
)# 云服务的数据命名空间(磁盘使用率那进行调用)
ecs_namespace = "acs_ecs_dashboard"
# 云服务(ECS)的监控项名称
Disk_metricname = "vm.DiskUtilization"  # 磁盘# 阿里云认证
clt = client.AcsClient("xxxxxxxxxxxxxxxxx", "xxxxxxxxxxxxxxxxxxxxxx", "cn-beijing"
)# 脚本所需参数
# 设置页数
page_number = "1"
# 设置每页返回多少，默认为10条
strip_number = "100"
# 磁盘使用率阀值(%)
Disk_use_rate = "50"# 列表---ECS列表
def get_sys_info():request = DescribeInstancesRequest()request.set_PageSize(strip_number)  # 设置每页返回多少，默认为10条request.set_accept_format("json")response = json.loads(clt.do_action(request))return responsefrom aliyunsdkecs.request.v20140526.DescribeDisksRequest import DescribeDisksRequest# ECS磁盘
def disk_info():request = DescribeDisksRequest()request.set_PageSize(strip_number)request.set_accept_format("json")request = clt.do_action_with_exception(request)request = json.loads(request).get("Disks").get("Disk")return request# ecs详情
def ecs_information():ecs_info = []for file_info in get_sys_info():information = {}information["InstanceName"] = file_info["InstanceName"]information["Status"] = file_info["Status"]information["OSType"] = file_info["OSType"]information["Cpu"] = file_info["Cpu"]information["Memory"] = file_info["Memory"]information["IpA"] = file_info["VpcAttributes"]["PrivateIpAddress"]["IpAddress"][0]if file_info["EipAddress"]["IpAddress"]:information["IpB"] = file_info["EipAddress"]["IpAddress"]ecs_endtime_0 = file_info["ExpiredTime"]ecs_endtime_1 = "%Y-%m-%dT%H:%MZ"ecs_endtime_2 = datetime.datetime.strptime(ecs_endtime_0, ecs_endtime_1)ecs_endtime_3 = (ecs_endtime_2 + datetime.timedelta(hours=8) - datetime.timedelta(seconds=1))information["time"] = str(ecs_endtime_3)ecs_info.append(information)print(ecs_info)from aliyunsdkr_kvstore.request.v20150101.DescribeInstancesOverviewRequest import (DescribeInstancesOverviewRequest,
)# 列出redis实例列表
def get_redis_info():request = DescribeInstancesOverviewRequest()request.set_accept_format("json")response_0 = clt.do_action_with_exception(request)response = json.loads(response_0).get("Instances")return response# redis详情
def redis_info():redis_info = []for file_info in get_redis_info():information = {}information["InstanceName"] = file_info["InstanceName"]information["Status"] = file_info["InstanceStatus"]information["InstanceType"] = file_info["InstanceType"]information["Capacity"] = file_info["Capacity"]information["EngineVersion"] = file_info["EngineVersion"]information["InstanceId"] = file_info["InstanceId"]information["message"] = get_redis_memusage(file_info["InstanceId"])ecs_endtime_0 = file_info["EndTime"]ecs_endtime_1 = "%Y-%m-%dT%H:%M:%SZ"ecs_endtime_2 = datetime.datetime.strptime(ecs_endtime_0, ecs_endtime_1)ecs_endtime_3 = ecs_endtime_2 + datetime.timedelta(hours=8)information["time"] = str(ecs_endtime_3)redis_info.append(information)print(redis_info)# 列出redis 内存的使用率
from aliyunsdkr_kvstore.request.v20150101.DescribeHistoryMonitorValuesRequest import DescribeHistoryMonitorValuesRequest# 详情---Redis详情
def get_redis_memusage(instanceId):"""主要是redis使用情况以及使用率"""# 获取当前时间（UTC）end_time = datetime.datetime.utcnow()# 当前时间减去一分钟start_time = end_time - datetime.timedelta(minutes=1)# 将时间格式化为 `yyyy-MM-ddTHH:mm:ssZ`start_time_str = start_time.strftime("%Y-%m-%dT%H:%M:%SZ")end_time_str = end_time.strftime("%Y-%m-%dT%H:%M:%SZ")request = DescribeHistoryMonitorValuesRequest()request.set_accept_format("json")request.set_InstanceId(instanceId)request.set_StartTime(start_time_str)request.set_EndTime(end_time_str)request.set_IntervalForHistory("60m")request.set_MonitorKeys("MemoryUsage,UsedMemory")response_0 = clt.do_action_with_exception(request)response_1 = json.loads(json.loads(response_0).get("MonitorHistory"))# 将所有键按时间排序sorted_keys = sorted(response_1.keys())# 取最后一个键的值last_key = sorted_keys[-1]last_value = response_1[last_key]# 转换 UsedMemory 为 GBused_memory_bytes = float(last_value["UsedMemory"])used_memory_gb = used_memory_bytes / (1024**3)return {"memoryUsage": f"{last_value['memoryUsage']}%","UsedMemoryGB": f"{used_memory_gb:.2f} GB",}from aliyunsdkrds.request.v20140815.DescribeDBInstancesRequest import DescribeDBInstancesRequest# 列出RDS实例列表
def get_rds_info():request = DescribeDBInstancesRequest()request.set_accept_format("json")request.set_PageSize(strip_number)response_0 = clt.do_action_with_exception(request)response = json.loads(response_0).get("Items").get("DBInstance")return response# rds详情
def rds_info():rds_info = []for file_info in get_rds_info():information = {}db_instance_description = file_info.get("DBInstanceDescription")# 如果 DBInstanceDescription 为 None 或空字符串，则使用 DBInstanceIdif not db_instance_description:db_instance_description = file_info.get("DBInstanceId", "未知")information["DBInstanceDescription"] = db_instance_descriptioninformation["Status"] = file_info["DBInstanceStatus"]information["Engine"] = file_info["Engine"]information["EngineVersion"] = file_info["EngineVersion"]information["DBInstanceId"] = file_info["DBInstanceId"]information["DBInstanceType"] = file_info["DBInstanceType"]information["PayType"] = file_info["PayType"]information["usage"] = rds_info_usage(file_info["DBInstanceId"])ecs_endtime_0 = file_info["ExpireTime"]ecs_endtime_1 = "%Y-%m-%dT%H:%M:%SZ"ecs_endtime_2 = datetime.datetime.strptime(ecs_endtime_0, ecs_endtime_1)ecs_endtime_3 = ecs_endtime_2 + datetime.timedelta(hours=8)information["time"] = str(ecs_endtime_3)rds_info.append(information)print(rds_info)from aliyunsdkrds.request.v20140815.DescribeDBInstanceAttributeRequest import DescribeDBInstanceAttributeRequest# rds详情
def rds_info_usage(instanceId):"""主要是rds使用情况以及使用率"""request = DescribeDBInstanceAttributeRequest()request.set_accept_format("json")request.set_DBInstanceId(instanceId)response_0 = clt.do_action_with_exception(request)response_1 = json.loads(response_0).get("Items").get("DBInstanceAttribute")# print(response_1)information = {}for file_info in response_1:# information = {}information["DBInstanceMemory"] = file_info["DBInstanceMemory"]information["DBInstanceCPU"] = file_info["DBInstanceCPU"]information["DBInstanceStorage"] = file_info["DBInstanceStorage"]information["DBInstanceDiskUsed"] = file_info["DBInstanceDiskUsed"]# rds_info.append(information)return informationfrom aliyunsdkdts.request.v20200101.DescribeSynchronizationJobsRequest import DescribeSynchronizationJobsRequest# 列出DTS实例列表
def get_dts_info():request = DescribeSynchronizationJobsRequest()request.set_accept_format("json")request.set_PageSize(strip_number)response_0 = clt.do_action_with_exception(request)response = json.loads(response_0).get("SynchronizationInstances")# response = json.loads(response_0).get("TotalRecordCount")return responsedef dts_info():dts_info = []for file_info in get_dts_info():sync_obj = file_info["SynchronizationObjects"][0]schema_name = sync_obj["SchemaName"]new_schema_name = sync_obj["NewSchemaName"]tables = sync_obj["TableIncludes"]Syncobject = {}# Schema 名称处理if schema_name == new_schema_name:Syncobject["dataname"] = new_schema_nameelse:Syncobject["olddataname"] = schema_nameSyncobject["newdataname"] = new_schema_name# 表名和表数量处理Syncobject["count"] = len(tables)Syncobject["tablenames"] = [table["TableName"] for table in tables]# for table in tables:#     print(table["TableName"])information = {}information["SynchronizationJobName"] = file_info["SynchronizationJobName"]information["Status"] = file_info["Status"]information["SynchronizationJobId"] = file_info["SynchronizationJobId"]information["SynchronizationJobClass"] = file_info["SynchronizationJobClass"]information["Syncobject"] = Syncobjectif "IP" in file_info["SourceEndpoint"] and file_info["SourceEndpoint"]["IP"]:information["source"] = file_info["SourceEndpoint"]["IP"]else:information["source"] = file_info["SourceEndpoint"]["InstanceId"]if ("IP" in file_info["DestinationEndpoint"]and file_info["DestinationEndpoint"]["IP"]):information["objective"] = file_info["DestinationEndpoint"]["IP"]else:information["objective"] = file_info["DestinationEndpoint"]["InstanceId"]ecs_endtime_0 = file_info["ExpireTime"]ecs_endtime_1 = "%Y-%m-%dT%H:%M:%SZ"ecs_endtime_2 = datetime.datetime.strptime(ecs_endtime_0, ecs_endtime_1)ecs_endtime_3 = ecs_endtime_2 + datetime.timedelta(hours=8)information["time"] = str(ecs_endtime_3)dts_info.append(information)print(dts_info)from aliyunsdkdomain.request.v20180129.QueryDomainListRequest import QueryDomainListRequest# 列出 domain列表
def get_domain_info():request = QueryDomainListRequest()request.set_accept_format("json")request.set_PageNum(1)request.set_PageSize(strip_number)response_0 = clt.do_action_with_exception(request)response = json.loads(response_0).get("Data").get("Domain")# response = json.loads(response_0).get("TotalItemNum")return responsedef domain_info():domain_info = []for file_info in get_domain_info():information = {}# 域名information["DomainName"] = file_info["DomainName"]# 域名状态information["Status"] = file_info["DomainStatus"]# 域名持有者information["Ccompany"] = file_info["Ccompany"]# 注册时间information["RegistrationDate"] = file_info["RegistrationDate"]# 到期时间information["ExpirationDate"] = file_info["ExpirationDate"]# 域名到期日和当前的时间的天数差值information["ExpirationCurrDateDiff"] = file_info["ExpirationCurrDateDiff"]domain_info.append(information)print(domain_info)from aliyunsdkcas.request.v20200407.ListUserCertificateOrderRequest import ListUserCertificateOrderRequest# 列出 certificate 列表
def get_certificate_info():request = ListUserCertificateOrderRequest()request.set_accept_format("json")request.set_CurrentPage(1)request.set_ShowSize(strip_number)response_0 = clt.do_action_with_exception(request)response = json.loads(response_0).get("CertificateOrderList")# response = json.loads(response_0).get("TotalCount")return responsedef certtime(tttime):from datetime import datetime# 毫秒数# milliseconds = 1740844799000milliseconds = tttime# 将毫秒数转换为秒数seconds = milliseconds / 1000# 将秒数转换为日期时间expiration_date = datetime.fromtimestamp(seconds)# 打印结果return expiration_datedef certificate_info():certificate_info = []for file_info in get_certificate_info():information = {}# 域名information["Domain"] = file_info["Domain"]# 证书状态information["Status"] = file_info["Status"]# 证书类型information["CertType"] = file_info["CertType"]# 证书算法information["Algorithm"] = file_info["Algorithm"]# 到期时间information["CertEndTime"] = certtime(file_info["CertEndTime"]).strftime("%Y-%m-%d")certificate_info.append(information)print(certificate_info)# 执行
if __name__ == "__main__":# 直打印get_sys_infoprint(get_sys_info())# ecs_information()# print(disk_info())# print(get_redis_info())# redis_info()# print(get_redis_memusage("r-21235156432150n"))# print(get_rds_info())# rds_info()# print(get_dts_info())# dts_info()# rds_info_usage()# print(get_domain_info())# domain_info()# print(get_certificate_info())# certificate_info()