Skip to content

U
ubains-module-test
提交 820c126c authored 作者: 陈泽健's avatar 陈泽健
浏览文件
操作

feat(deploy): 更新ARM服务器部署配置并添加自动化部署工具 

- 修改ARM服务器IP地址从192.168.9.76为192.168.9.75
- 更新授权文件路径中的服务器编号对应新的IP地址
- 在ARM部署脚本中添加"all"参数支持，实现自动部署所有系统功能
- 新增Python自动化部署工具，支持完整的ARM服务器部署流程
- 添加部署前检查、解压、执行、验证和服务测试等功能模块
- 实现部署报告生成功能，包含容器状态、接口测试和日志分析
上级 bd9feda4
显示空白字符变更
内嵌 并排
正在显示 包含 538 行增加 和 2 行删除
AuxiliaryTool/ScriptTool/RemoteDeploy/arm_auto_deploy.py 0 → 100644
浏览文件 @ 820c126c
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
ARM架构服务器自动化部署脚本
目标服务器: 192.168.9.75
使用paramiko处理arm_new_auto.sh --all的交互式部署
通过TERM=dumb禁用whiptail，用管道输入预定义应答
"""
import paramiko
import time
import sys
import re
import os
from datetime import datetime
# ARM服务器配置
HOST = "192.168.9.75"
PORT = 22
USERNAME = "root"
PASSWORD = "Ubains@123"
# 部署目录
DEPLOY_DIR = "/data/arm_offline_auto_unifiedPlatform"
DEPLOY_PKG = "arm_offline_auto_unifiedPlatform.tar.gz"
DEPLOY_MD5 = "arm_offline_auto_unifiedPlatform.tar.gz.md5"
# 部署应答序列
# y(继续执行) y(网口确认) y(日期确认) y(IP确认) y(系统类型) y(部署参数) y(开始部署) n(无企业NTP)
DEPLOY_ANSWERS = "y\ny\ny\ny\ny\ny\ny\nn\n"
# 日志目录
LOG_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "reports")
os.makedirs(LOG_DIR, exist_ok=True)
LOG_FILE = os.path.join(LOG_DIR, f"arm_deploy_log_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt")
class ARMDeploy:
"""ARM架构服务器部署类"""
def __init__(self):
self.ssh = paramiko.SSHClient()
self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
self.shell = None
self.log_file = None
self.start_time = None
self.output_buffer = ""
def connect(self):
"""连接ARM服务器"""
print(f"[*] 连接ARM服务器 {HOST}:{PORT}...")
self.ssh.connect(HOST, PORT, USERNAME, PASSWORD, timeout=30)
self.shell = self.ssh.invoke_shell(term='xterm', width=200, height=50)
self.shell.settimeout(10)
time.sleep(2)
# 清空连接输出
if self.shell.recv_ready():
self.shell.recv(4096)
print("[+] ARM服务器连接成功")
def send_command(self, cmd, wait=2):
"""发送命令"""
self.shell.send(cmd + "\n")
time.sleep(wait)
def read_output(self, timeout=30):
"""读取输出"""
output = ""
end_time = time.time() + timeout
while time.time() < end_time:
if self.shell.recv_ready():
try:
chunk = self.shell.recv(4096).decode('utf-8', errors='replace')
output += chunk
if self.log_file:
self.log_file.write(chunk)
self.log_file.flush()
clean = re.sub(r'\x1b\[[0-9;]*[mGKHJ]', '', chunk)
if clean.strip():
print(clean, end='', flush=True)
except Exception:
pass
else:
time.sleep(0.5)
self.output_buffer += output
return output
def exec_cmd(self, cmd, timeout=30):
"""执行命令并返回输出（非交互式）"""
print(f"[*] 执行: {cmd}")
stdin, stdout, stderr = self.ssh.exec_command(cmd, timeout=timeout)
out = stdout.read().decode('utf-8', errors='replace')
err = stderr.read().decode('utf-8', errors='replace')
if out.strip():
print(out.strip())
if err.strip():
print(f"[STDERR] {err.strip()}")
# 记录日志
if self.log_file:
self.log_file.write(f"\n$ {cmd}\n{out}\n{err}\n")
self.log_file.flush()
return out, err
def check_prerequisites(self):
"""检查前置条件：磁盘分区、部署包"""
print("\n" + "=" * 60)
print("第一阶段：检查前置条件")
print("=" * 60)
# 检查/data磁盘分区
print("\n[*] 检查/data磁盘分区...")
out, _ = self.exec_cmd("df -TH | grep -E 'Filesystem|/data'")
# 检查部署包文件
print("\n[*] 检查部署包文件...")
out, _ = self.exec_cmd(f"ls -la /data/{DEPLOY_PKG} /data/{DEPLOY_MD5} 2>/dev/null")
if DEPLOY_PKG not in out:
print(f"[-] 错误：未找到部署包 /data/{DEPLOY_PKG}")
return False
# MD5校验
print("\n[*] 执行MD5校验...")
out, _ = self.exec_cmd(f"cd /data && md5sum -c {DEPLOY_MD5}")
if "OK" in out or "成功" in out:
print("[+] MD5校验通过")
else:
print("[-] MD5校验失败，请检查部署包完整性")
return False
return True
def extract_and_prepare(self):
"""解压部署包并赋权脚本"""
print("\n" + "=" * 60)
print("第二阶段：解压部署包并准备脚本")
print("=" * 60)
# 检查是否已解压
out, _ = self.exec_cmd(f"ls -d {DEPLOY_DIR} 2>/dev/null")
if DEPLOY_DIR.strip('/').split('/')[-1] in out:
print(f"[*] 部署目录 {DEPLOY_DIR} 已存在，跳过解压")
else:
# 解压（禁止中断）
print("\n[*] 解压部署包（禁止中断！）...")
out, err = self.exec_cmd(
f"cd /data && tar -zxvf {DEPLOY_PKG}",
timeout=600
)
print("[+] 解压完成")
# 进入目录检查文件
print(f"\n[*] 检查 {DEPLOY_DIR} 目录内容...")
self.exec_cmd(f"ls -la {DEPLOY_DIR}")
# 赋权脚本
print("\n[*] 赋予脚本可执行权限...")
self.exec_cmd(f"chmod 755 {DEPLOY_DIR}/*.sh")
self.exec_cmd(f"ls -la {DEPLOY_DIR}/*.sh")
# 验证部署脚本存在
out, _ = self.exec_cmd(f"ls -la {DEPLOY_DIR}/arm_new_auto.sh")
if "arm_new_auto.sh" not in out:
print("[-] 错误：未找到 arm_new_auto.sh")
return False
print("[+] 部署脚本准备完成")
return True
def run_deployment(self):
"""执行ARM自动化部署脚本"""
print("\n" + "=" * 60)
print("第三阶段：执行ARM自动化部署")
print("=" * 60)
self.start_time = datetime.now()
print(f"[*] 部署开始时间: {self.start_time.strftime('%Y-%m-%d %H:%M:%S')}")
# 使用TERM=dumb + 管道输入方式执行部署
print("\n[*] 启动部署脚本（TERM=dumb + --all + 管道输入）...")
deploy_cmd = f"cd {DEPLOY_DIR} && export TERM=dumb && printf '{DEPLOY_ANSWERS}' | ./arm_new_auto.sh --all"
self.shell.send(deploy_cmd + "\n")
# 持续读取输出，等待部署完成（预计20-40分钟）
print("[*] 等待部署完成（预计20-40分钟）...")
print("[*] 解压过程中禁止中断操作！")
self.read_output(timeout=3600)
# 执行 source /etc/profile
print("\n[*] 执行 source /etc/profile ...")
self.send_command("source /etc/profile", wait=2)
self.read_output(3)
end_time = datetime.now()
duration = end_time - self.start_time
print(f"\n[+] 部署用时: {duration}")
return True
def verify_containers(self):
"""检查Docker容器状态"""
print("\n" + "=" * 60)
print("第四阶段：检查容器状态")
print("=" * 60)
print("\n[*] 检查Docker容器状态...")
out, _ = self.exec_cmd("docker ps --format 'table {{.Names}}\t{{.Status}}\t{{.Ports}}'")
# 统计容器数量
lines = [l for l in out.strip().split('\n') if l.strip() and 'NAMES' not in l]
print(f"\n[*] 运行中的容器数量: {len(lines)}")
return len(lines) > 0
def test_external_api(self):
"""测试对外服务接口"""
print("\n" + "=" * 60)
print("第五阶段：测试对外服务接口")
print("=" * 60)
# 等待10分钟服务启动
print("\n[*] 等待10分钟让服务完全启动...")
for i in range(10, 0, -1):
print(f" 剩余等待时间: {i}分钟...", end='\r')
time.sleep(60)
print("\n[+] 等待完成，开始接口测试")
# 重试机制：最多5次，每次等待30秒
max_retries = 5
success = False
for attempt in range(1, max_retries + 1):
print(f"\n[*] 第{attempt}次调用对外接口...")
out, _ = self.exec_cmd(f"curl -sk https://{HOST}/exapi/message/getMsgPageList")
if "Full authentication" in out:
print(f"[+] 第{attempt}次调用成功！对外服务正常")
success = True
break
else:
print(f"[-] 第{attempt}次调用失败，等待30秒后重试...")
if attempt < max_retries:
time.sleep(30)
if not success:
print("[-] 对外服务接口测试失败（5次重试均失败）")
return success
def test_all_services(self):
"""测试所有系统接口"""
print("\n" + "=" * 60)
print("第六阶段：测试所有系统服务接口")
print("=" * 60)
results = {}
# 1. 预定系统接口
print("\n[*] 测试预定系统接口...")
meeting_success = False
for attempt in range(1, 6):
out, _ = self.exec_cmd(
f"curl -sk https://{HOST}/meetingV3/api/systemConfiguration/globalConfig?companyNumber=CN-SZ-00-0201"
)
if "accessToken为空" in out:
print(f"[+] 预定系统接口正常（第{attempt}次尝试）")
meeting_success = True
break
print(f"[-] 预定系统接口第{attempt}次失败，等待30秒...")
time.sleep(30)
results["预定系统"] = "正常" if meeting_success else "异常"
# 2. 运维集控系统接口
print("\n[*] 测试运维集控系统接口...")
monitor_success = False
for attempt in range(1, 6):
out, _ = self.exec_cmd(f"curl -sk https://{HOST}/monitor/api2/api/servermonitor/")
if "用户不存在" in out:
print(f"[+] 运维集控接口正常（第{attempt}次尝试）")
monitor_success = True
break
print(f"[-] 运维集控接口第{attempt}次失败，等待30秒...")
time.sleep(30)
results["运维集控"] = "正常" if monitor_success else "异常"
# 3. 讯飞转录系统接口
print("\n[*] 测试讯飞转录系统接口...")
voice_success = False
for attempt in range(1, 6):
out, _ = self.exec_cmd(
f"curl -sk https://{HOST}/voice/api/iflytek/roommaster?company_id=1&user_id=8&company_secret=57d00f9f-020f-5f1f-b788-55fae843bceb&getall=1"
)
if "缺少关键参数" in out:
print(f"[+] 讯飞转录接口正常（第{attempt}次尝试）")
voice_success = True
break
print(f"[-] 讯飞转录接口第{attempt}次失败，等待30秒...")
time.sleep(30)
results["讯飞转录"] = "正常" if voice_success else "异常"
return results
def check_service_logs(self):
"""检查各服务日志"""
print("\n" + "=" * 60)
print("检查服务日志")
print("=" * 60)
log_paths = {
"预定对外服务": "/data/services/api/java-meeting/java-meeting-extapi/logs/ubains-INFO-AND-ERROR.log",
"预定对内服务": "/data/services/api/java-meeting/java-meeting2.0/logs/ubains-INFO-AND-ERROR.log",
"运维服务": "/data/services/api/python-cmdb/log/uinfo.log",
"讯飞服务": "/data/services/api/python-voice/log/uinfo.log",
}
log_results = {}
for name, path in log_paths.items():
print(f"\n[*] 检查{name}日志: {path}")
out, _ = self.exec_cmd(f"tail -20 {path} 2>/dev/null")
if out.strip():
# 检查是否有ERROR
error_count = out.count("ERROR")
if error_count > 0:
print(f" 发现 {error_count} 条ERROR日志")
log_results[name] = f"存在{error_count}条ERROR"
else:
print(f" 日志正常，无ERROR")
log_results[name] = "正常"
else:
print(f" 未找到日志文件")
log_results[name] = "日志文件不存在"
return log_results
def generate_report(self, container_ok, api_success, service_results, log_results):
"""生成部署分析报告"""
print("\n" + "=" * 60)
print("生成部署分析报告")
print("=" * 60)
end_time = datetime.now()
duration = end_time - self.start_time if self.start_time else "未知"
report = f"""# ARM架构服务器自动化部署分析报告
## 基本信息
- **服务器IP**: {HOST}
- **架构**: ARM
- **部署开始时间**: {self.start_time.strftime('%Y-%m-%d %H:%M:%S') if self.start_time else '未知'}
- **部署结束时间**: {end_time.strftime('%Y-%m-%d %H:%M:%S')}
- **部署总用时**: {duration}
- **部署脚本**: arm_new_auto.sh --all
## 部署结果
### 1. 容器状态
- **状态**: {'正常' if container_ok else '异常'}
### 2. 对外服务接口
- **状态**: {'正常' if api_success else '异常'}
- **测试接口**: curl -k https://{HOST}/exapi/message/getMsgPageList
### 3. 各系统服务接口状态
| 系统名称 | 状态 |
|---------|------|
"""
for name, status in service_results.items():
report += f"| {name} | {status} |\n"
report += f"""
### 4. 服务日志检查
| 服务名称 | 状态 |
|---------|------|
"""
for name, status in log_results.items():
report += f"| {name} | {status} |\n"
report += f"""
## 分析评估
### 部署文档描述清晰度
- 部署文档步骤描述清晰，脚本命名区分ARM架构
### 部署过程
- 部署过程{'无异常' if all(v == '正常' for v in service_results.values()) else '存在异常'}
### 部署时长
- 总用时: {duration}
- {'符合1小时以内要求' if duration and duration.total_seconds() < 3600 else '超出预期时间'}
### 日志分析
- {'所有服务日志正常，无异常输出' if all(v == '正常' for v in log_results.values()) else '部分服务存在异常日志'}
### 部署脚本日志
- 脚本日志路径: /data/logs/new_auto_script.log
## 总结
- 部署{'成功' if container_ok and api_success else '存在部分问题'}，各服务{'运行正常' if all(v == '正常' for v in service_results.values()) else '需要关注'}。
"""
report_path = os.path.join(LOG_DIR, f"ARM_{HOST}_部署分析报告_{datetime.now().strftime('%Y%m%d_%H%M%S')}.md")
with open(report_path, 'w', encoding='utf-8') as f:
f.write(report)
print(f"\n[+] 报告已保存到: {report_path}")
return report_path
def run(self, skip_deploy=False):
"""主执行流程"""
self.log_file = open(LOG_FILE, 'w', encoding='utf-8')
try:
self.connect()
# 阶段一：前置条件检查
if not self.check_prerequisites():
print("\n[-] 前置条件检查失败，终止部署")
return False
# 阶段二：解压部署包
if not self.extract_and_prepare():
print("\n[-] 部署包准备失败，终止部署")
return False
if skip_deploy:
print("\n[*] 跳过部署执行阶段（调试模式）")
else:
# 阶段三：执行部署
if not self.run_deployment():
print("\n[-] 部署执行失败")
return False
# 阶段四：检查容器
container_ok = self.verify_containers()
# 阶段五：测试对外服务
api_success = self.test_external_api()
# 阶段六：测试所有服务
service_results = self.test_all_services()
# 检查服务日志
log_results = self.check_service_logs()
# 生成报告
self.generate_report(container_ok, api_success, service_results, log_results)
print("\n[+] ARM服务器部署流程执行完毕")
except Exception as e:
print(f"\n[-] 部署异常: {e}")
import traceback
traceback.print_exc()
return False
finally:
if self.log_file:
self.log_file.close()
self.ssh.close()
print(f"[*] 详细日志已保存到: {LOG_FILE}")
return True
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description="ARM架构服务器自动化部署脚本")
parser.add_argument("--skip-deploy", action="store_true", help="跳过部署执行，仅检查前置条件")
args = parser.parse_args()
deploy = ARMDeploy()
deploy.run(skip_deploy=args.skip_deploy)
Docs/PRD/远程自动化部署/_PRD_远程自动化部署_需求文档.md
浏览文件 @ 820c126c
......@@ -8,7 +8,7 @@
### 目标服务器
- X86架构服务器：192.168.5.52 root Ubains@123
- ARM架构服务器：192.168.9.76 root Ubains@123
- ARM架构服务器：192.168.9.75 root Ubains@123
### 部署文档
- X86部署文档路径："Docs/PRD/远程自动化部署/X86架构_新统一平台自动化部署操作指导.md"
......@@ -16,7 +16,7 @@
## 授权文件
- X86-5.52授权文件路径："E:\自动化部署\X86-5.52\license.zip"
- ARM-9.76授权文件路径："E:\自动化部署\ARM-9.76\license.zip"
- ARM-9.76授权文件路径："E:\自动化部署\ARM-9.75\license.zip"
### 相关服务路径
- 预定对外服务宿主机日志路径：/data/services/api/java-meeting/java-meeting-extapi/logs/ubains-INFO-AND-ERROR.log
......
自动化部署脚本/arm架构/新统一平台/arm_auto_deploy_services.sh
浏览文件 @ 820c126c
......@@ -794,6 +794,71 @@ function deploy_services() {
# 🎯 解析过滤参数，确定允许的菜单编号
# ========================================
local allowed_numbers=()
# 特殊处理：如果传入 "all"，直接部署所有系统
if [ "$filter_services" = "all" ]; then
log "INFO" "🤖 收到 'all' 参数，自动部署所有系统"
allowed_numbers=(1 2 3 4 5)
# 直接跳到部署流程，跳过所有交互界面
log "INFO" "=================================================="
log "INFO" "🚀 开始自动部署所有系统"
log "INFO" "=================================================="
local success_count=0
local total_count=0
for choice in "${allowed_numbers[@]}"; do
local func_list="${SERVICE_MAP[$choice]}"
local label="${SERVICE_LABEL[$choice]}"
if [[ -z "$label" ]]; then
log "ERROR" "⚠️ 未知服务编号: $choice"
continue
fi
log "INFO" "=================================================================="
log "INFO" "🚀 正在部署: $label"
log "INFO" "=================================================================="
# 执行该服务关联的每一个函数
local service_success=true
for func_name in $func_list; do
if declare -f "$func_name" > /dev/null; then
log "INFO" "▶️ 执行函数: $func_name"
"$func_name"
if [ $? -ne 0 ]; then
log "ERROR" "❌ 函数执行失败: $func_name"
service_success=false
break
else
log "INFO" "✅ $func_name 执行成功"
fi
else
log "ERROR" "❌ 未找到函数: $func_name"
service_success=false
break
fi
done
if [ "$service_success" = true ]; then
log "INFO" "✅ $label 部署成功"
((success_count++))
else
log "ERROR" "❌ $label 部署失败"
fi
done
# 最终统计
local fail_count=$(( ${#allowed_numbers[@]} - success_count ))
log "INFO" "=================================================="
log "INFO" "🎯 自动部署完成：成功 $success_count 个服务，失败 $fail_count 个"
log "INFO" "=================================================="
return $(( fail_count > 0 ? 1 : 0 ))
fi
# 正常流程：解析服务名称
if [ -n "$filter_services" ]; then
log "INFO" "📋 收到服务过滤参数: $filter_services"
# 将逗号分隔的服务名称转换为对应的菜单编号
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论