【 使用环境 】 测试环境
【 OB or 其他组件 】ocp 告警通道 自定义脚本v4.4.0社区版
【 使用版本 】
【问题描述】清晰明确描述问题
【复现路径】
#!/usr/bin/env python
-- coding: utf-8 --
import os
import sys
import pymysql
from datetime import datetime
MySQL 数据库配置（可根据实际环境调整，也可通过环境变量获取）
MYSQL_CONFIG = {
‘host’: os.environ.get(‘MYSQL_HOST’, ‘’), # 优先从环境变量获取
‘port’: int(os.environ.get(‘MYSQL_PORT’, 3306)),
‘user’: os.environ.get(‘MYSQL_USER’, ‘’), # 数据库用户名
‘password’: os.environ.get(‘MYSQL_PASS’, ‘’), # 数据库密码
‘db’: os.environ.get(‘MYSQL_DB’, ‘’), # 包含 events 表的数据库名
‘charset’: ‘utf8’
}
def parse_alarm_data():
“”"
独立函数：解析告警数据（从环境变量获取，构造插入MySQL的字典数据）
返回：构造完成的告警数据字典（已移除ip字段，严格匹配events表结构）
异常：缺少必填环境变量时抛出异常
“”"
# 获取核心必填告警消息（message）
try:
message = os.environ[‘message’] # 告警消息，对应 MySQL 的 message 字段
except KeyError as e:
raise Exception(f"缺少必填环境变量: {e}")
# 构造并返回告警数据字典（已移除ip字段，仅保留events表需要的alarm_type和message）
alarm_data = {
'alarm_type': os.environ.get('alarm_level', ''), # 映射到数据库 alarm_type 字段
'message': message # 关键映射：环境变量 message -> 数据库 message 字段
}
return alarm_data
def insert_mysql_alarm(alarm_data):
“”"
独立函数：接收解析后的告警数据，插入 MySQL events 表
参数：alarm_data - 由 parse_alarm_data 构造的告警数据字典
返回：成功时返回包含校验信息的字典（插入状态、影响行数、自增ID等）
异常：插入失败时抛出异常
“”"
# 构建参数化 SQL 语句（兼容反引号字段名，防止 SQL 注入）
fields = ', '.join(alarm_data.keys())
placeholders = ‘, ‘.join([’%s’] * len(alarm_data))
insert_sql = f"INSERT INTO events_1 ({fields}) VALUES ({placeholders})"
sql_params = tuple(alarm_data.values())
conn = None
cursor = None
try:
# 建立数据库连接
conn = pymysql.connect(**MYSQL_CONFIG)
cursor = conn.cursor()
# 执行插入操作并获取影响行数
affected_rows = cursor.execute(insert_sql, sql_params)
# 获取插入数据的自增ID
insert_id = cursor.lastrowid
# 提交事务，确保数据持久化写入
conn.commit()
# 校验插入是否成功
if affected_rows <= 0:
raise Exception("告警数据插入失败：无受影响行数")
# 构造成功响应校验信息
success_response = {
'status': 'success',
'msg': '告警数据插入MySQL成功',
'affected_rows': affected_rows,
'insert_id': insert_id,
'message': alarm_data['message']
}
return success_response
finally:
# 无论成功与否，关闭游标和连接，释放资源
if cursor:
cursor.close()
if conn:
conn.close()
def main():
“”"
主函数：串联数据解析和MySQL插入操作，处理异常，输出校验信息
保留：成功输出包含 校验成功，满足外部校验要求
成功返回0，失败输出错误到stderr并返回1
“”"
try:
# 步骤1：解析告警数据
alarm_data = parse_alarm_data()
# 步骤2：插入MySQL数据库，获取成功校验响应
success_info = insert_mysql_alarm(alarm_data)
# 步骤3：输出包含 %校验成功% 的成功信息（已移除主机IP相关输出）
success_msg = (
f"校验成功: {success_info['msg']} | "
f"告警主题：{success_info['message']} | "
f"影响行数：{success_info['affected_rows']} | "
f"插入ID：{success_info['insert_id']}"
)
print(success_msg) # 标准输出成功信息，便于外部捕获
return 0 # 成功返回0
except Exception as e:
# 失败时输出错误到 stderr
error_msg = f"【校验失败】{str(e)}"
sys.stderr.write(f"{error_msg}\n")
return 1 # 失败返回非0值
if name == “main”:
sys.exit(main())
CREATE TABLE events_1 (
id bigint(20) unsigned NOT NULL AUTO_INCREMENT,
alarm_type varchar(300) NOT NULL DEFAULT ‘’ COMMENT '告警类型 ',
message varchar(300) DEFAULT NULL COMMENT ‘告警主题’,
PRIMARY KEY (id)
) ENGINE=InnoDB AUTO_INCREMENT=98972185 DEFAULT CHARSET=utf8
返回：
{
“duration”: 551,
“error”: {
“code”: “ALM80001”,
“message”: “发送测试消息失败，错误信息：告警发送结果校验失败，期望结果：校验成功，Script 实际返回结果：校验成功: 告警数据插入MySQL成功 | 告警主题：测试 | 影响行数：1 | 插入ID：98972186\n”
},
“server”: “0c2f999838”,
“status”: 400,
“successful”: false,
“timestamp”: “2026-01-05T13:40:05.451164419+08:00”,
“traceId”: “9a3d2678ca757fa9”
}
Response 校验信息:
校验成功
问题，他其实无法只调用脚本，根据控制台正则匹配成功，一定要status为200吗？为400是不是就是会返回发送测试消息失败
【附件及日志】推荐使用OceanBase敏捷诊断工具obdiag收集诊断信息,详情参见链接（右键跳转查看）：