提交 1e7ae2e8 authored 作者: 陈泽健's avatar 陈泽健

chore: 移除大Excel文件和测试报告历史,加入.gitignore

上级 584cfe65
......@@ -46,6 +46,19 @@ description: ARM集群夜间监控 - 监测9网段四台ARM架构服务器集群
- **必须等待用户确认计划后方可开始执行**,不得跳过确认步骤直接操作;
- 计划执行文档保存到 `.claude/skills/ARM-CLUSTER-MONITOR/计划执行_{YYYYMMDD_HHMMSS}.md`
### 代码实现规范
- **代码存放位置**:如需代码实现(如Python脚本、工具类等),统一存放在 `.claude/skills/ARM-CLUSTER-MONITOR/code/` 目录下;
- **代码文件命名**:使用有意义的文件名,例如:
- `ssh_manager.py` - SSH连接管理工具
- `monitor_agent.py` - 监控Agent实现
- `report_generator.py` - 报告生成工具
- **代码规范**
- 所有代码必须包含中文注释
- 代码风格符合项目规范(参考 `Docs/PRD/01规范文档/_PRD_规范文档_代码规范.md`
- 使用UTF-8编码
- 函数和类需包含文档字符串说明
- **代码版本管理**:代码文件应纳入Git版本控制,提交信息使用中文描述。
## Usage
```
......
@echo off
cd /d E:\GithubData\ubains-module-test
cd /d E:\GithubData\ubains-module-test\develop
E:\nodejs\claude.cmd --permission-mode bypassPermissions
......@@ -70,7 +70,7 @@ schedule_interval: "2h"
### 0.4 需求文档处理规范
- **每次理解需求文档都需要输出《计划执行》文档**
- **每次理解需求文档,都需要输出《计划执行》文档**
- 计划执行文档需包含:执行目标、执行步骤、预期结果、风险评估
- 用户确认后方可执行
......@@ -96,7 +96,95 @@ schedule_interval: "2h"
| [风险1] | [高/中/低] | [应对方法] |
## 确认信息
请确认是否执行?(是/否)
请确认是否执行?(是/否)
```
### 0.5 代码实现约束
> ⚠️ 如果本 Skill 需要通过代码实现功能,必须遵守以下约束:
#### 代码存放位置规范
**强制要求**: 所有代码文件必须存放在本 Skill 的 `code/` 目录下:
```
.claude/skills/X86-CLUSTER-MONITOR/
├── SKILL.md # Skill说明文档
├── config.json # Skill配置文件
├── 集群信息配置.md # 集群配置信息
└── code/ # 代码实现目录(强制)
├── __init__.py # Python包初始化文件
├── ssh_manager.py # SSH连接管理模块
├── server_checker.py # 服务器检查模块
├── report_generator.py # 报告生成模块
├── utils.py # 工具函数模块
└── main.py # 主执行入口
```
#### 代码规范要点
1. **目录结构**:
- 代码文件必须位于 `code/` 目录内
- 每个模块职责单一,便于维护
- 必须包含 `__init__.py` 文件标识为Python包
2. **代码风格**:
- 所有注释使用中文
- 函数命名遵循 `snake_case` 规范
- 类命名遵循 `PascalCase` 规范
- 每个模块必须有详细的模块级注释
3. **模块划分建议**:
- `ssh_manager.py`: SSH连接管理、免密配置
- `server_checker.py`: 各项服务器检查逻辑
- `report_generator.py`: 报告生成与保存
- `utils.py`: 通用工具函数(时间处理、格式化等)
- `main.py`: 主入口,协调各模块执行
4. **依赖管理**:
-`code/` 目录下创建 `requirements.txt`
- 列出所需Python包及版本
5. **配置文件引用**:
- 读取 `../集群信息配置.md` 获取服务器信息
- 读取 `../config.json` 获取Skill配置
- 配置文件路径使用相对路径
#### 示例代码结构
```python
# ssh_manager.py
"""
SSH连接管理模块
负责SSH连接建立、免密配置、连接池管理
"""
import paramiko
import os
from pathlib import Path
class SSHManager:
"""SSH连接管理器"""
def __init__(self, server_ip):
"""
初始化SSH管理器
Args:
server_ip (str): 服务器IP地址
"""
self.server_ip = server_ip
self.key_path = Path.home() / '.ssh' / 'keys' / server_ip
def connect(self):
"""
建立SSH连接
Returns:
paramiko.SSHClient: SSH客户端对象
"""
# 实现连接逻辑...
pass
```
---
......
"""
X86集群监控代码实现包
提供SSH连接管理、服务器状态检查、报告生成等功能
"""
__version__ = '1.0.0'
__author__ = 'czj'
\ No newline at end of file
"""
X86集群监控主执行模块
协调SSH连接、服务器检查、报告生成
功能:
1. 读取集群配置
2. 并行检查四台服务器
3. 生成监控报告
4. 输出执行摘要
使用方法:
python main.py
"""
import sys
import logging
from pathlib import Path
from datetime import datetime
import concurrent.futures
# 导入其他模块
from ssh_manager import SSHManager, connect_server
from server_checker import ServerChecker
from report_generator import ReportGenerator
from utils import setup_logging, parse_server_config_md, format_timestamp
logger = logging.getLogger(__name__)
class ClusterMonitor:
"""集群监控主控制器"""
def __init__(self, config_path=None):
"""
初始化集群监控器
Args:
config_path (str): 集群配置文件路径
"""
# 配置日志
setup_logging()
# 读取配置
if config_path:
self.config_path = Path(config_path)
else:
# 默认配置文件
self.config_path = Path(__file__).parent.parent / '集群信息配置.md'
# 读取服务器列表
self.servers = parse_server_config_md(self.config_path)
logger.info(f'已加载 {len(self.servers)} 台服务器配置')
def check_server(self, server_info):
"""
检查单台服务器
Args:
server_info (dict): 服务器信息
Returns:
dict: 检查结果
"""
server_ip = server_info['ip']
logger.info(f'开始检查服务器: {server_ip}')
try:
# 连接服务器
ssh_manager = connect_server(
server_ip=server_ip,
username=server_info.get('username', 'root')
)
# 执行检查
checker = ServerChecker(ssh_manager)
result = checker.run_full_check()
# 关闭连接
ssh_manager.close()
logger.info(f'服务器检查完成: {server_ip}')
return result
except Exception as e:
logger.error(f'服务器检查失败: {server_ip}, 错误: {e}')
return {
'server_ip': server_ip,
'error': str(e)
}
def run_cluster_check(self):
"""
执行集群检查(并行)
Returns:
list: 所有服务器检查结果
"""
logger.info('开始执行集群检查...')
results = []
# 并行检查四台服务器
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
future_to_server = {
executor.submit(self.check_server, server): server
for server in self.servers
}
for future in concurrent.futures.as_completed(future_to_server):
server = future_to_server[future]
try:
result = future.result()
results.append(result)
except Exception as e:
logger.error(f'服务器 {server["ip"]} 检查异常: {e}')
results.append({
'server_ip': server['ip'],
'error': str(e)
})
logger.info(f'集群检查完成,共检查 {len(results)} 台服务器')
return results
def generate_report(self, check_results):
"""
生成监控报告
Args:
check_results (list): 检查结果列表
Returns:
str: 报告文件路径
"""
logger.info('开始生成监控报告...')
generator = ReportGenerator()
report_path = generator.generate_report(check_results)
logger.info(f'报告已生成: {report_path}')
return report_path
def print_summary(self, check_results):
"""
输出执行摘要到控制台
Args:
check_results (list): 检查结果列表
"""
print("\n" + "="*60)
print("X86集群监控执行摘要")
print("="*60)
print(f"\n执行时间: {format_timestamp()}")
print(f"监控服务器数: {len(check_results)}")
# 统计状态
normal_count = 0
warning_count = 0
error_count = 0
for result in check_results:
if 'error' in result:
error_count += 1
else:
# 检查资源使用
cpu = float(result['system_resources']['cpu'].replace('%', '').strip())
memory = float(result['system_resources']['memory'].replace('%', '').strip())
if cpu > 90 or memory > 90:
error_count += 1
elif cpu > 80 or memory > 80:
warning_count += 1
else:
normal_count += 1
print(f"\n服务器状态:")
print(f" 🟢 正常: {normal_count}")
print(f" 🟡 警告: {warning_count}")
print(f" 🔴 异常: {error_count}")
# 输出异常服务器
if error_count > 0:
print("\n异常服务器:")
for result in check_results:
if 'error' in result:
print(f" - {result['server_ip']}: {result['error']}")
print("\n" + "="*60)
def run(self):
"""
执行完整监控流程
"""
logger.info('开始执行X86集群监控...')
# 执行检查
check_results = self.run_cluster_check()
# 生成报告
report_path = self.generate_report(check_results)
# 输出摘要
self.print_summary(check_results)
logger.info(f'监控完成,报告路径: {report_path}')
return report_path
def main():
"""主函数"""
try:
# 创建监控器
monitor = ClusterMonitor()
# 执行监控
report_path = monitor.run()
print(f"\n✅ 监控完成")
print(f"📄 报告路径: {report_path}")
return 0
except Exception as e:
logger.error(f'监控执行失败: {e}')
print(f"\n❌ 监控失败: {e}")
return 1
if __name__ == '__main__':
sys.exit(main())
\ No newline at end of file
# X86集群监控依赖包
paramiko>=2.7.2
requests>=2.28.0
colorama>=0.4.6
python-dateutil>=2.8.2
\ No newline at end of file
"""
SSH连接管理模块
负责SSH连接建立、免密配置、连接池管理
功能:
1. 检查本地免密配置是否存在
2. 自动生成SSH密钥对
3. 上传公钥到目标服务器实现免密登录
4. 管理SSH连接池
"""
import paramiko
import os
from pathlib import Path
import logging
logger = logging.getLogger(__name__)
class SSHManager:
"""SSH连接管理器"""
def __init__(self, server_ip, username='root', port=22):
"""
初始化SSH管理器
Args:
server_ip (str): 服务器IP地址
username (str): SSH用户名,默认为root
port (int): SSH端口,默认为22
"""
self.server_ip = server_ip
self.username = username
self.port = port
self.key_dir = Path.home() / '.ssh' / 'keys' / server_ip
self.client = None
def check_key_exists(self):
"""
检查免密密钥是否已存在
Returns:
bool: 密钥存在返回True,否则返回False
"""
private_key = self.key_dir / 'id_rsa'
public_key = self.key_dir / 'id_rsa.pub'
return private_key.exists() and public_key.exists()
def generate_keys(self):
"""
生成SSH密钥对
Returns:
tuple: (私钥路径, 公钥路径)
"""
# 创建密钥目录
self.key_dir.mkdir(parents=True, exist_ok=True)
private_key = self.key_dir / 'id_rsa'
public_key = self.key_dir / 'id_rsa.pub'
# 使用paramiko生成RSA密钥
key = paramiko.RSAKey.generate(2048)
# 保存私钥
key.write_private_key_file(str(private_key))
# 保存公钥
with open(public_key, 'w') as f:
f.write(f'{key.get_name()} {key.get_base64()} {self.username}@{self.server_ip}')
logger.info(f'已生成SSH密钥对: {self.key_dir}')
return private_key, public_key
def connect_with_password(self, password):
"""
使用密码建立SSH连接
Args:
password (str): SSH密码
Returns:
paramiko.SSHClient: SSH客户端对象
"""
try:
self.client = paramiko.SSHClient()
self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
self.client.connect(
hostname=self.server_ip,
port=self.port,
username=self.username,
password=password,
timeout=10
)
logger.info(f'SSH连接成功: {self.server_ip}')
return self.client
except Exception as e:
logger.error(f'SSH连接失败: {self.server_ip}, 错误: {e}')
raise
def connect_with_key(self):
"""
使用密钥建立SSH连接(免密登录)
Returns:
paramiko.SSHClient: SSH客户端对象
"""
private_key = self.key_dir / 'id_rsa'
try:
self.client = paramiko.SSHClient()
self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
self.client.connect(
hostname=self.server_ip,
port=self.port,
username=self.username,
key_filename=str(private_key),
timeout=10
)
logger.info(f'免密SSH连接成功: {self.server_ip}')
return self.client
except Exception as e:
logger.error(f'免密SSH连接失败: {self.server_ip}, 错误: {e}')
raise
def setup_passwordless_auth(self, password):
"""
配置SSH免密登录
1. 生成密钥对
2. 连接服务器
3. 上传公钥
Args:
password (str): SSH密码
Returns:
bool: 配置成功返回True
"""
# 生成密钥对
private_key, public_key = self.generate_keys()
# 读取公钥内容
with open(public_key, 'r') as f:
pubkey_content = f.read().strip()
# 连接服务器
client = self.connect_with_password(password)
# 上传公钥到目标服务器
command = f'mkdir -p ~/.ssh && echo "{pubkey_content}" >> ~/.ssh/authorized_keys && chmod 600 ~/.ssh/authorized_keys'
stdin, stdout, stderr = client.exec_command(command)
if stdout.channel.recv_exit_status() == 0:
logger.info(f'公钥上传成功,已配置免密登录: {self.server_ip}')
client.close()
return True
else:
error = stderr.read().decode()
logger.error(f'公钥上传失败: {error}')
client.close()
raise Exception(f'公钥上传失败: {error}')
def execute_command(self, command):
"""
在远程服务器执行命令
Args:
command (str): 要执行的命令
Returns:
tuple: (stdout输出, stderr输出, 返回码)
"""
if not self.client:
raise Exception('SSH未连接')
stdin, stdout, stderr = self.client.exec_command(command)
exit_code = stdout.channel.recv_exit_status()
return stdout.read().decode(), stderr.read().decode(), exit_code
def close(self):
"""关闭SSH连接"""
if self.client:
self.client.close()
logger.info(f'SSH连接已关闭: {self.server_ip}')
def __enter__(self):
"""支持with语句"""
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""支持with语句"""
self.close()
def connect_server(server_ip, username='root', password=None):
"""
连接服务器的高层接口
Args:
server_ip (str): 服务器IP
username (str): 用户名
password (str): 密码(可选,如果已配置免密)
Returns:
SSHManager: SSH管理器实例
"""
manager = SSHManager(server_ip, username)
# 检查是否存在免密配置
if manager.check_key_exists():
# 使用免密连接
manager.connect_with_key()
else:
# 需要密码
if not password:
raise Exception(f'未配置免密登录,请提供密码: {server_ip}')
manager.connect_with_password(password)
return manager
\ No newline at end of file
"""
工具函数模块
提供通用工具函数(时间处理、格式化、配置读取等)
功能:
1. 时间格式化工具
2. 配置文件读取
3. 数据格式化工具
4. 日志配置
"""
import json
import logging
from datetime import datetime
from pathlib import Path
import yaml
def setup_logging(log_level=logging.INFO):
"""
配置日志系统
Args:
log_level: 日志级别,默认INFO
"""
logging.basicConfig(
level=log_level,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.StreamHandler(),
logging.FileHandler('cluster_monitor.log', encoding='utf-8')
]
)
def format_timestamp(timestamp=None, format_str='%Y-%m-%d %H:%M:%S'):
"""
格式化时间戳
Args:
timestamp (datetime): 时间对象,默认为当前时间
format_str (str): 格式化字符串
Returns:
str: 格式化后的时间字符串
"""
if timestamp is None:
timestamp = datetime.now()
return timestamp.strftime(format_str)
def read_config(config_path):
"""
读取配置文件(支持JSON和YAML)
Args:
config_path (str): 配置文件路径
Returns:
dict: 配置字典
"""
path = Path(config_path)
if not path.exists():
raise FileNotFoundError(f'配置文件不存在: {config_path}')
# 根据扩展名判断格式
if path.suffix == '.json':
with open(path, 'r', encoding='utf-8') as f:
return json.load(f)
elif path.suffix in ['.yaml', '.yml']:
with open(path, 'r', encoding='utf-8') as f:
return yaml.safe_load(f)
else:
raise ValueError(f'不支持的配置文件格式: {path.suffix}')
def parse_server_config_md(md_path):
"""
解析Markdown格式的服务器配置文件
Args:
md_path (str): Markdown配置文件路径
Returns:
list: 服务器列表
"""
path = Path(md_path)
if not path.exists():
raise FileNotFoundError(f'配置文件不存在: {md_path}')
servers = []
with open(path, 'r', encoding='utf-8') as f:
content = f.read()
# 简单的Markdown表格解析
lines = content.split('\n')
for line in lines:
# 查找包含IP地址的行
if '192.168.5.' in line:
# 提取IP地址
parts = line.split('|')
for part in parts:
if '192.168.5.' in part.strip():
servers.append({
'ip': part.strip(),
'username': 'root',
'port': 22
})
return servers
def format_percentage(value):
"""
格式化百分比数值
Args:
value (str or float): 百分比值
Returns:
str: 格式化后的百分比字符串
"""
try:
if isinstance(value, str):
# 移除百分号
num = float(value.replace('%', '').strip())
else:
num = float(value)
return f'{num:.2f}%'
except:
return value
def format_size_mb(mb_value):
"""
格式化大小(MB为单位)
Args:
mb_value (float): MB值
Returns:
str: 格式化后的大小字符串
"""
try:
mb = float(mb_value)
if mb > 1024:
gb = mb / 1024
return f'{gb:.2f} GB'
else:
return f'{mb:.2f} MB'
except:
return str(mb_value)
def get_status_emoji(status):
"""
根据状态返回表情符号
Args:
status (str): 状态字符串
Returns:
str: 表情符号
"""
if 'running' in status.lower() or 'normal' in status.lower() or 'ok' in status.lower():
return '✅'
elif 'warning' in status.lower():
return '🟡'
else:
return '❌'
def truncate_text(text, max_length=100):
"""
截断文本
Args:
text (str): 文本内容
max_length (int): 最大长度
Returns:
str: 截断后的文本
"""
if len(text) <= max_length:
return text
else:
return text[:max_length] + '...'
\ No newline at end of file
......@@ -19,6 +19,16 @@
/AuxiliaryTool/ScriptTool/ApiSecurityTest/logs/
/AuxiliaryTool/ScriptTool/ApiSecurityTest/*.log
# --- 大文件过滤规则(Excel测试数据和测试报告)---
# 超大Excel测试数据文件(不提交)
预定系统/测试数据/**/会议预定测试用例.xlsx
预定系统/测试数据/兰州中石化项目测试用例.xlsx
Docs/PRD/问题知识库/问题反馈跟踪表.xlsx
# 测试报告历史记录
/AuxiliaryTool/DocumentAutoOptimizationCalibration/reports/
# --- Python 核心缓存与编译文件 (必须配置) ---
# 忽略所有 __pycache__ 目录 (包含任意层级的子目录)
......
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论