# -*- coding: utf-8 -*- """ Redis 主从复制模块 - 前端接口层 负责处理前端请求，调用具体的业务逻辑实现 """ import json import os import sys import time import socket # 添加路径 if '/www/server/panel/class' not in sys.path: sys.path.insert(0, '/www/server/panel/class') if '/www/server/panel' not in sys.path: sys.path.insert(0, '/www/server/panel') import public from mod.project.node.redis_slave_util.config_manager import ConfigManager from mod.project.node.redis_slave_util.redis_manager import RedisManager from mod.project.node.redis_slave_util.slave_manager import SlaveManager from mod.project.node.redis_slave_util.sync_service import SyncService from mod.project.node.redis_slave_util.monitor_service import MonitorService class main: """Redis 主从复制主模块""" def __init__(self): self.config_manager = ConfigManager() self.redis_manager = RedisManager() self.slave_manager = SlaveManager() self.sync_service = SyncService() self.monitor_service = MonitorService() # ==================== 版本和检查相关 ==================== def get_redis_version(self, get=None): """获取Redis版本""" version = self.redis_manager.get_redis_version() if not version: return public.returnMsg(False, "此功能仅支持Redis-3.2以上版本使用！") return public.returnMsg(True, version) def check_slave(self, get): """检查从库连接和配置""" required_params = ['panel_addr', 'panel_key', 'master_ip', 'slave_ip'] for param in required_params: if not hasattr(get, param): return public.returnMsg(False, "缺少参数！ {}".format(param)) return self.slave_manager.check_slave( get.panel_addr, get.panel_key, get.master_ip, get.slave_ip ) def check_redis_port(self, server_ip=None, port=6379): """检查Redis端口是否开放""" try: # 将端口转换为整数 if isinstance(port, str): port = int(port) # 创建socket连接 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(3) # 设置3秒超时 # 尝试连接 result = sock.connect_ex((server_ip, port)) sock.close() # connect_ex返回0表示连接成功 if result == 0: return True else: return False except Exception as e: return False # ==================== 主从复制组管理 ==================== def replication_group_list(self, get=None): """获取主从复制组列表（带分页和搜索功能）""" try: # 1. 检查Redis版本 version_check = self.redis_manager.get_redis_version() if not version_check: return public.returnMsg(False, "此功能仅支持Redis-3.2以上版本使用！") # 2. 获取并验证分页参数 search = getattr(get, 'search', '').strip() if get else '' p = int(getattr(get, 'p', 1)) if get and hasattr(get, 'p') else 1 limit = int(getattr(get, 'limit', 20)) if get and hasattr(get, 'limit') else 20 # 参数验证 if p < 1: p = 1 if limit < 1 or limit > 100: limit = 20 # 3. 获取所有复制组列表 all_data = self.config_manager.get_all_replication_groups() if not all_data: return { "status": True, "msg": "success", "data": [], "page": { "current_page": p, "total_pages": 0, "total_count": 0, "per_page": limit, "has_prev": False, "has_next": False, "start_item": 0, "end_item": 0, "showing": "显示第 0 到 0 项，共 0 项" }, "search": search, "total": 0 } # 4. 丰富数据 - 添加监控信息 enriched_data = [] for group in all_data: group_info = self.enrich_group_data(group) enriched_data.append(group_info) # 5. 搜索过滤 filtered_data = enriched_data if search: filtered_data = [] for item in enriched_data: # 支持按复制组名称、主库地址搜索 if (search.lower() in item.get('group_name', '').lower() or search.lower() in item.get('master_server_ip', '').lower()): filtered_data.append(item) # 6. 计算分页参数 total_count = len(filtered_data) total_pages = (total_count + limit - 1) // limit if total_count > 0 else 0 offset = (p - 1) * limit # 7. 获取当前页数据 current_page_data = filtered_data[offset:offset + limit] # 8. 构建分页信息 start_item = offset + 1 if total_count > 0 else 0 end_item = min(offset + limit, total_count) page_info = { "current_page": p, "total_pages": total_pages, "total_count": total_count, "per_page": limit, "has_prev": p > 1, "has_next": p < total_pages, "prev_page": p - 1 if p > 1 else None, "next_page": p + 1 if p < total_pages else None, "start_item": start_item, "end_item": end_item, "showing": "显示第 {} 到 {} 项，共 {} 项".format(start_item, end_item, total_count) } # 9. 返回结果 result = { "status": True, "msg": "success", "data": current_page_data, "page": page_info, "search": search, "total": total_count } return result except Exception as e: import traceback error_msg = "获取主从复制组列表失败: {}".format(str(e)) public.WriteLog("TYPE_REDIS", error_msg) return { "status": False, "msg": error_msg, "data": [], "page": {}, "search": "", "total": 0 } def enrich_group_data(self, group): """丰富复制组数据，添加监控信息""" try: # 获取主节点监控数据 master_info = self.monitor_service.get_master_status(group['master_node_id']) # 获取从节点状态 slave_count = len(group.get('slave_nodes', [])) normal_slaves = 0 total_lag = 0 lag_count = 0 for slave_node_id in group.get('slave_nodes', []): slave_status = self.monitor_service.get_slave_status(slave_node_id) if slave_status.get('is_online', False): normal_slaves += 1 if slave_status.get('lag') is not None: total_lag += slave_status['lag'] lag_count += 1 # 计算平均延迟 avg_lag = round(total_lag / lag_count, 2) if lag_count > 0 else 0 return { 'group_id': group['group_id'], 'group_name': group['group_name'], 'master_server_ip': "{}:{}".format(group['master_ip'], group['master_port']), 'memory_usage': master_info.get('memory_usage_percent', 0), 'qps': master_info.get('qps', 0), 'connections': master_info.get('connections', 0), 'slave_count': slave_count, 'normal_slaves': normal_slaves, 'avg_lag': avg_lag, 'created_time': group.get('created_time', ''), 'status': 'normal' if normal_slaves == slave_count else 'warning', 'task_id': group.get('task_id', '') } except Exception as e: public.WriteLog("TYPE_REDIS", "丰富复制组数据失败: {}".format(str(e))) return group # ==================== 创建主从复制组 ==================== def create_replication_group(self, get): """创建主从复制组""" required_params = ['group_name', 'master_node_id', 'slave_node_ids','master_ip','redis_password'] for param in required_params: if not hasattr(get, param): return public.returnMsg(False, "缺少参数！ {}".format(param)) # 验证参数 group_name = get.group_name.strip() if not group_name: return public.returnMsg(False, "主从名称不能为空！") # 检查名称是否已存在 if self.config_manager.group_name_exists(group_name): return public.returnMsg(False, "主从复制组名称已存在！") # 获取节点信息 master_node = public.M("node").where("id = ?", (get.master_node_id,)).find() if not master_node: return public.returnMsg(False, "未找到主节点信息") slave_node_ids = json.loads(get.slave_node_ids) if isinstance(get.slave_node_ids, str) else get.slave_node_ids slave_nodes = [] for slave_id in slave_node_ids: slave_node = public.M("node").where("id = ?", (slave_id,)).find() if not slave_node: return public.returnMsg(False, "未找到从节点信息: {}".format(slave_id)) slave_nodes.append(slave_node) # 验证主从IP不能相同 #master_ip = master_node.get("server_ip") master_ip = get.master_ip for slave_node in slave_nodes: slave_ip = slave_node.get("server_ip") if master_ip == slave_ip: return public.returnMsg(False, "主库和从库不能是同一个IP!") # if slave_ip == "127.0.0.1": # return public.returnMsg(False, "从库不能是127.0.0.1!") if master_ip == "127.0.0.1": return public.returnMsg(False, "主库不能是127.0.0.1!") # 检查密码强度 if len(get.redis_password) < 8: return public.returnMsg(False, "Redis密码长度不能少于8位!") ip_list = [master_ip] for slave_node in slave_nodes: ip_list.append(slave_node.get("server_ip")) for ip in ip_list: if not self.check_redis_port(ip): return public.returnMsg(False, "连接{} redis失败，请指定ip放行6379端口，并设置redis外网访问权限(注意设置密码)".format(ip)) # 创建配置目录 if not os.path.exists("/www/server/panel/config/redis_slave_info"): public.ExecShell("mkdir -p /www/server/panel/config/redis_slave_info") # 启动创建工作流 return self.sync_service.create_replication_group( group_name, master_node, slave_nodes, get.redis_password,master_ip ) def get_creation_status(self, get): """获取创建状态""" if not hasattr(get, 'task_id'): return public.returnMsg(False, "缺少参数！ task_id") return self.sync_service.get_creation_status(get.task_id) def get_group_creation_status(self, get): """通过复制组ID获取创建状态""" if not hasattr(get, 'group_id'): return public.returnMsg(False, "缺少参数！ group_id") try: # 通过group_id获取task_id task_id = self.config_manager.get_group_task_id(get.group_id) if not task_id: return public.returnMsg(False, "未找到该复制组的创建任务ID") # 获取任务状态 return self.sync_service.get_creation_status(task_id) except Exception as e: error_msg = "获取复制组创建状态失败: {}".format(str(e)) public.WriteLog("TYPE_REDIS", error_msg) return public.returnMsg(False, error_msg) # ==================== 复制组详细管理 ==================== def get_group_detail(self, get): """获取复制组详细信息""" if not hasattr(get, 'group_id'): return public.returnMsg(False, "缺少参数！ group_id") try: group_config = self.config_manager.get_group_config(get.group_id) if not group_config: return public.returnMsg(False, "未找到复制组配置") # 获取主节点详细信息 master_detail = self.monitor_service.get_master_detail_info(group_config['master_node_id']) # 获取从节点详细信息列表 slave_details = [] for slave_node_id in group_config.get('slave_nodes', []): slave_detail = self.monitor_service.get_slave_detail_info(slave_node_id) slave_details.append(slave_detail) result = { "status": True, "msg": "success", "data": { "group_info": group_config, "master_detail": master_detail, "slave_details": slave_details } } return result except Exception as e: error_msg = "获取复制组详细信息失败: {}".format(str(e)) public.WriteLog("TYPE_REDIS", error_msg) return public.returnMsg(False, error_msg) # ==================== 从库管理 ==================== def get_slave_status(self, get): """获取从库状态""" if not hasattr(get, 'slave_node_id'): return public.returnMsg(False, "缺少参数！ slave_node_id") return self.monitor_service.get_slave_status(get.slave_node_id) def reconnect_slave(self, get): """重连从库""" required_params = ['group_id', 'slave_node_id'] for param in required_params: if not hasattr(get, param): return public.returnMsg(False, "缺少参数！ {}".format(param)) return self.slave_manager.reconnect_slave(get.group_id, get.slave_node_id) def remove_slave(self, get): """移除从库""" required_params = ['group_id', 'slave_node_id'] for param in required_params: if not hasattr(get, param): return public.returnMsg(False, "缺少参数！ {}".format(param)) return self.slave_manager.remove_slave(get.group_id, get.slave_node_id) def add_slave_to_group(self, get): """向复制组添加新的从节点""" required_params = ['group_id', 'new_slave_node_id'] for param in required_params: if not hasattr(get, param): return public.returnMsg(False, "缺少参数！ {}".format(param)) return self.sync_service.add_slave_to_existing_group(get.group_id, get.new_slave_node_id) # ==================== 复制组操作 ==================== def delete_replication_group(self, get): """删除主从复制组""" if not hasattr(get, 'group_id'): return public.returnMsg(False, "缺少参数！ group_id") group_id = get.group_id group_config = self.config_manager.get_group_config(group_id) if not group_config: return public.returnMsg(False, "未找到复制组配置信息") try: cleanup_errors = [] # 记录开始删除 self.config_manager.write_group_log(group_id, "开始删除主从复制组", "INFO") # 1. 停止所有从库复制并清理配置 slave_nodes = group_config.get('slave_nodes', []) for slave_node_id in slave_nodes: try: # 停止从库复制 stop_result = self.slave_manager.stop_slave_replication(slave_node_id) if not stop_result.get("status"): cleanup_errors.append(f"停止从库 {slave_node_id} 复制失败: {stop_result.get('msg')}") # 清理从节点Redis配置（移除replicaof等配置） clean_result = self.redis_manager.stop_replication(slave_node_id) if not clean_result.get("status"): cleanup_errors.append(f"清理从库 {slave_node_id} 配置失败: {clean_result.get('msg')}") else: self.config_manager.write_group_log(group_id, f"从库 {slave_node_id} 配置已清理", "INFO") except Exception as e: cleanup_errors.append(f"处理从库 {slave_node_id} 异常: {str(e)}") # 2. 清理主库复制相关配置 master_node_id = group_config.get('master_node_id') if master_node_id: try: # 清理主库复制用户和配置 master_cleanup = self.redis_manager.cleanup_replication_users(master_node_id) if not master_cleanup: cleanup_errors.append(f"清理主库 {master_node_id} 配置失败") else: self.config_manager.write_group_log(group_id, f"主库 {master_node_id} 配置已清理", "INFO") except Exception as e: cleanup_errors.append(f"清理主库 {master_node_id} 异常: {str(e)}") # 3. 删除配置文件和日志文件 self.config_manager.write_group_log(group_id, "删除配置文件和日志", "INFO") # 先获取日志内容用于记录错误 if cleanup_errors: for error in cleanup_errors: self.config_manager.write_group_log(group_id, f"清理错误: {error}", "WARNING") # 删除配置和日志文件 remove_result = self.config_manager.remove_group_config(group_id) if not remove_result: return public.returnMsg(False, "删除配置文件失败") # 构建返回消息 if cleanup_errors: error_summary = "; ".join(cleanup_errors[:3]) # 只显示前3个错误 if len(cleanup_errors) > 3: error_summary += f" 等共{len(cleanup_errors)}个错误" return public.returnMsg(True, f"删除成功，但存在清理问题: {error_summary}") else: return public.returnMsg(True, "删除成功！") except Exception as e: error_msg = "删除复制组失败: {}".format(str(e)) public.WriteLog("TYPE_REDIS", error_msg) # 即使出现异常也尝试删除配置文件 try: self.config_manager.remove_group_config(group_id) except: pass return public.returnMsg(False, error_msg) # ==================== 监控相关 ==================== def get_monitor_data(self, get=None): """获取监控数据""" try: return self.monitor_service.get_all_monitor_data() except Exception as e: error_msg = "获取监控数据失败: {}".format(str(e)) public.WriteLog("TYPE_REDIS", error_msg) return public.returnMsg(False, error_msg) def get_alert_info(self, get=None): """获取告警信息""" try: alert_data = [] # 获取所有复制组 all_groups = self.config_manager.get_all_replication_groups() for group in all_groups: # 检查主节点告警 master_alerts = self.monitor_service.check_master_alerts(group['master_node_id']) if master_alerts: for alert in master_alerts: if alert.get("type") == "master_offline": alert_data.append(alert) # 检查从节点告警 for slave_node_id in group.get('slave_nodes', []): slave_alerts = self.monitor_service.check_slave_alerts(slave_node_id) if slave_alerts: for alert in slave_alerts: if alert.get("type") == "slave_offline": alert_data.append(alert) return { "status": True, "msg": "success", "data": alert_data, "count": len(alert_data) } except Exception as e: error_msg = "获取告警信息失败: {}".format(str(e)) public.WriteLog("TYPE_REDIS", error_msg) return public.returnMsg(False, error_msg) # ==================== 工具方法 ==================== def generate_redis_password(self, get=None): """生成Redis密码""" import random import string # 生成16位强密码，包含字母、数字和特殊字符 chars = string.ascii_letters + string.digits + "!@#$%^&*" password = ''.join(random.choice(chars) for _ in range(16)) return public.returnMsg(True, password) def validate_redis_connection(self, get): """验证Redis连接""" required_params = ['node_id', 'password'] for param in required_params: if not hasattr(get, param): return public.returnMsg(False, "缺少参数！ {}".format(param)) return self.redis_manager.test_connection(get.node_id, get.password) def get_available_nodes(self, get=None): """获取可用的Redis节点列表""" #_REPO_DIR="/www/server/panel/data/mod_node_status_cache/" try: nodes = public.M("node").select() available_nodes = [] # 获取所有复制组信息，用于检测节点是否已在复制组中 all_groups = self.config_manager.get_all_replication_groups() # 构建节点到复制组的映射关系 node_group_map = {} for group in all_groups: # 主节点 master_id = group.get('master_node_id') if master_id: node_group_map[master_id] = { 'group_id': group['group_id'], 'group_name': group['group_name'], 'role': 'master' } # 从节点 for slave_id in group.get('slave_nodes', []): node_group_map[slave_id] = { 'group_id': group['group_id'], 'group_name': group['group_name'], 'role': 'slave' } for node in nodes: try: # if node.get("api_key") != "local": # if not os.path.exists(_REPO_DIR+"server_{}.json".format(node.get("id"))): # continue # if not self.bt_api_client_test(node.get("id")): # continue # 检查节点是否安装了Redis并获取版本信息 redis_version = self.redis_manager.get_redis_version_by_node(node['id']) if redis_version: # 只有检测到Redis版本的节点才添加到列表 # 检查Redis服务状态 service_status = self.redis_manager.check_redis_service(node['id']) node_status = 'online' if service_status.get('status', False) else 'offline' # 检查节点是否在复制组中 is_in_replication_group = node['id'] in node_group_map group_info = node_group_map.get(node['id'], None) server_ip = node.get('server_ip', '') if server_ip == "127.0.0.1": server_ip = public.GetLocalIp() node_info = { 'id': node['id'], 'name': node.get('remarks', ''), 'server_ip': server_ip, 'status': node_status, 'redis_version': redis_version, 'type': 'standalone', # Redis类型，默认为单机 'is_in_replication_group': is_in_replication_group, 'group_info': group_info # 如果在复制组中，包含复制组信息；否则为None } available_nodes.append(node_info) except Exception as e: # 如果检测某个节点失败，记录日志但继续处理其他节点 public.WriteLog("TYPE_REDIS", f"检测节点 {node.get('id')} Redis状态失败: {str(e)}") continue return { "status": True, "msg": "success", "data": available_nodes } except Exception as e: error_msg = "获取可用节点失败: {}".format(str(e)) public.WriteLog("TYPE_REDIS", error_msg) return public.returnMsg(False, error_msg) # ==================== 后台任务方法 ==================== def auto_monitor_task(self): """自动监控任务（后台任务调用）""" return self.monitor_service.auto_monitor_task() def health_check_task(self): """健康检查任务（后台任务调用）""" return self.monitor_service.health_check_task()