from __future__ import annotations import json import time from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any from urllib.parse import parse_qs, urlparse from .auth import ( build_clear_session_cookie, build_session_cookie, create_session, delete_session, ensure_auth_store, get_session, parse_session_token, authenticate_user, ) from .config import APP_BIND, APP_PORT, STATIC_DIR, utc_now from .dashboard import build_monitor_payload from .nacos import ( get_nacos_config_list_payload, get_nacos_config_payload, save_nacos_config_payload, validate_nacos_config_payload, ) from .runtime_config import get_go_config_payload, save_go_config_payload, validate_go_config_payload from .service_logs import get_service_log_export_payload, get_service_log_preview_payload from .service_release import build_update_targets_payload from .service_update_jobs import get_update_job_log_text, get_update_job_payload, start_update_job from .service_ops import build_restart_targets_payload, rolling_restart_payload def response_json(payload: dict[str, Any] | list[Any]) -> bytes: # 统一把 JSON 编码成 UTF-8 文本。 return (json.dumps(payload, ensure_ascii=False, indent=2) + "\n").encode("utf-8") class MonitorHandler(BaseHTTPRequestHandler): # 自定义服务名，便于 curl 时识别来源。 server_version = "HyAppMonitor/2.0" def log_message(self, format: str, *args: Any) -> None: # noqa: A003 # 关闭默认访问日志，避免轮询刷屏。 return def current_session_token(self) -> str: # 同一个请求对象里只解析一次 Cookie。 if not hasattr(self, "_cached_session_token"): self._cached_session_token = parse_session_token(self.headers.get("Cookie")) return str(self._cached_session_token) def current_session(self) -> Any: # 同一个请求对象里也只查一次会话库。 if not hasattr(self, "_cached_session"): self._cached_session = get_session(self.current_session_token()) return self._cached_session def do_GET(self) -> None: # noqa: N802 # GET 请求需要响应头和响应体。 self.dispatch(send_body=True) def do_HEAD(self) -> None: # noqa: N802 # HEAD 请求只返回响应头。 self.dispatch(send_body=False) def do_POST(self) -> None: # noqa: N802 # POST 请求主要用于配置保存。 self.dispatch_post() def dispatch(self, send_body: bool) -> None: # 只解析 URL 的 path 段。 parsed = urlparse(self.path) # 顺手解析 query 参数，后面接口直接取。 query = parse_qs(parsed.query, keep_blank_values=True) # 登录页对未登录用户开放，已登录时直接回总览。 if parsed.path in {"/login", "/login.html"}: if self.current_session(): return self.redirect("/", send_body=send_body) return self.serve_static("login.html", "text/html; charset=utf-8", send_body=send_body) # 总览页需要先过登录校验，未登录统一跳到登录页。 if parsed.path in {"/", "/index.html"}: if not self.current_session(): return self.redirect("/login", send_body=send_body) return self.serve_static("index.html", "text/html; charset=utf-8", send_body=send_body) # CSS 文件。 if parsed.path == "/app.css": return self.serve_static("app.css", "text/css; charset=utf-8", send_body=send_body) # 登录页脚本。 if parsed.path == "/login.js": return self.serve_static("login.js", "application/javascript; charset=utf-8", send_body=send_body) # 前端业务 JS。 if parsed.path == "/app.js": return self.serve_static("app.js", "application/javascript; charset=utf-8", send_body=send_body) # 本地 Vue 运行时。 if parsed.path == "/vue.js": return self.serve_static("vue.js", "application/javascript; charset=utf-8", send_body=send_body) # 会话查询接口单独开放给登录页和主页面做状态判断。 if parsed.path == "/api/auth/session": return self.handle_auth_session(send_body=send_body) # 除鉴权接口外，其余 API 一律要求带有效会话。 if parsed.path.startswith("/api/") and not self.require_authentication(send_body=send_body): return # 轻量健康检查。 if parsed.path == "/api/health": return self.send_payload( HTTPStatus.OK, response_json({"status": "ok", "time": utc_now()}), "application/json; charset=utf-8", send_body=send_body, ) # 主监控接口。 if parsed.path == "/api/monitor/services": return self.send_payload( HTTPStatus.OK, response_json(build_monitor_payload()), "application/json; charset=utf-8", send_body=send_body, ) # 返回当前 namespace 下全部 Nacos 配置元数据。 if parsed.path == "/api/nacos/configs": try: payload = get_nacos_config_list_payload() except Exception as exc: # noqa: BLE001 return self.send_error_payload(HTTPStatus.BAD_GATEWAY, str(exc), send_body=send_body) return self.send_payload( HTTPStatus.OK, response_json(payload), "application/json; charset=utf-8", send_body=send_body, ) # 返回单个配置的原始文本内容。 if parsed.path == "/api/nacos/config": group_name = self.query_value(query, "group") data_id = self.query_value(query, "dataId") if not group_name or not data_id: return self.send_error_payload( HTTPStatus.BAD_REQUEST, "group and dataId are required", send_body=send_body, ) try: payload = get_nacos_config_payload(group_name, data_id) except Exception as exc: # noqa: BLE001 return self.send_error_payload(HTTPStatus.BAD_GATEWAY, str(exc), send_body=send_body) return self.send_payload( HTTPStatus.OK, response_json(payload), "application/json; charset=utf-8", send_body=send_body, ) # 返回当前线上 golang 的运行配置。 if parsed.path == "/api/runtime/golang-config": try: payload = get_go_config_payload() except Exception as exc: # noqa: BLE001 return self.send_error_payload(HTTPStatus.BAD_GATEWAY, str(exc), send_body=send_body) return self.send_payload( HTTPStatus.OK, response_json(payload), "application/json; charset=utf-8", send_body=send_body, ) # 返回可滚动重启的业务服务。 if parsed.path == "/api/ops/restart-targets": try: payload = build_restart_targets_payload() except Exception as exc: # noqa: BLE001 return self.send_error_payload(HTTPStatus.BAD_GATEWAY, str(exc), send_body=send_body) return self.send_payload( HTTPStatus.OK, response_json(payload), "application/json; charset=utf-8", send_body=send_body, ) # 返回可构建并发布的业务服务。 if parsed.path == "/api/ops/update-targets": try: payload = build_update_targets_payload() except Exception as exc: # noqa: BLE001 return self.send_error_payload(HTTPStatus.BAD_GATEWAY, str(exc), send_body=send_body) return self.send_payload( HTTPStatus.OK, response_json(payload), "application/json; charset=utf-8", send_body=send_body, ) # 返回当前服务更新任务的阶段和日志。 if parsed.path == "/api/ops/update-status": operation_id = self.query_value(query, "id") if not operation_id: return self.send_error_payload( HTTPStatus.BAD_REQUEST, "id is required", send_body=send_body, ) try: payload = get_update_job_payload(operation_id) except KeyError: return self.send_error_payload(HTTPStatus.NOT_FOUND, "update operation not found", send_body=send_body) except Exception as exc: # noqa: BLE001 return self.send_error_payload(HTTPStatus.BAD_GATEWAY, str(exc), send_body=send_body) return self.send_payload( HTTPStatus.OK, response_json(payload), "application/json; charset=utf-8", send_body=send_body, ) # 返回单个服务更新任务的完整文本日志。 if parsed.path == "/api/ops/update-log": operation_id = self.query_value(query, "id") if not operation_id: return self.send_error_payload( HTTPStatus.BAD_REQUEST, "id is required", send_body=send_body, ) try: content = get_update_job_log_text(operation_id) except KeyError: return self.send_error_payload(HTTPStatus.NOT_FOUND, "update log not found", send_body=send_body) except Exception as exc: # noqa: BLE001 return self.send_error_payload(HTTPStatus.BAD_GATEWAY, str(exc), send_body=send_body) return self.send_payload( HTTPStatus.OK, content.encode("utf-8"), "text/plain; charset=utf-8", send_body=send_body, ) # 返回单个服务实例最近 1000 条容器日志。 if parsed.path == "/api/ops/service-log": service_name = self.query_value(query, "service") host_name = self.query_value(query, "host") if not service_name or not host_name: return self.send_error_payload( HTTPStatus.BAD_REQUEST, "service and host are required", send_body=send_body, ) try: payload = get_service_log_preview_payload(service_name, host_name) except KeyError: return self.send_error_payload(HTTPStatus.NOT_FOUND, "service log target not found", send_body=send_body) except Exception as exc: # noqa: BLE001 return self.send_error_payload(HTTPStatus.BAD_GATEWAY, str(exc), send_body=send_body) return self.send_payload( HTTPStatus.OK, response_json(payload), "application/json; charset=utf-8", send_body=send_body, ) # 导出单个服务实例指定条数的容器日志。 if parsed.path == "/api/ops/service-log/export": service_name = self.query_value(query, "service") host_name = self.query_value(query, "host") lines = self.query_value(query, "lines") if not service_name or not host_name: return self.send_error_payload( HTTPStatus.BAD_REQUEST, "service and host are required", send_body=send_body, ) try: meta, content = get_service_log_export_payload(service_name, host_name, lines) except KeyError: return self.send_error_payload(HTTPStatus.NOT_FOUND, "service log target not found", send_body=send_body) except ValueError as exc: return self.send_error_payload(HTTPStatus.BAD_REQUEST, str(exc), send_body=send_body) except Exception as exc: # noqa: BLE001 return self.send_error_payload(HTTPStatus.BAD_GATEWAY, str(exc), send_body=send_body) filename = f"{meta['service']}-{meta['host']}-{meta['lines']}.log" return self.send_payload( HTTPStatus.OK, content.encode("utf-8"), "text/plain; charset=utf-8", send_body=send_body, extra_headers={"Content-Disposition": f'attachment; filename="{filename}"'}, ) # favicon 直接返回空。 if parsed.path == "/favicon.ico": return self.send_payload(HTTPStatus.NO_CONTENT, b"", "image/x-icon", send_body=send_body) # 其他路径统一 404。 return self.send_error_payload(HTTPStatus.NOT_FOUND, "not found", send_body=send_body) def dispatch_post(self) -> None: # 只解析 path 段，POST 不处理复杂 query。 parsed = urlparse(self.path) # 退出接口允许空请求体，先单独处理。 if parsed.path == "/api/auth/logout": return self.handle_logout() # 登录接口对未登录用户开放，但要求标准 JSON 请求体。 if parsed.path == "/api/auth/login": try: payload = self.read_json_body() except ValueError as exc: return self.send_error_payload(HTTPStatus.BAD_REQUEST, str(exc), send_body=True) return self.handle_login(payload) # 其余 POST 接口都要求先通过会话校验。 if not self.require_authentication(send_body=True): return try: payload = self.read_json_body() except ValueError as exc: return self.send_error_payload(HTTPStatus.BAD_REQUEST, str(exc), send_body=True) if parsed.path == "/api/nacos/config/validate": return self.handle_nacos_validate(payload) if parsed.path == "/api/nacos/config": return self.handle_nacos_save(payload) if parsed.path == "/api/runtime/golang-config/validate": return self.handle_go_validate(payload) if parsed.path == "/api/runtime/golang-config": return self.handle_go_save(payload) if parsed.path == "/api/ops/rolling-restart": return self.handle_rolling_restart(payload) if parsed.path == "/api/ops/update-services": return self.handle_update_services(payload) return self.send_error_payload(HTTPStatus.NOT_FOUND, "not found", send_body=True) def handle_auth_session(self, send_body: bool) -> None: session = self.current_session() # 未登录时返回 401，前端据此回到登录页。 if session is None: return self.send_error_payload( HTTPStatus.UNAUTHORIZED, "authentication required", send_body=send_body, extra_headers=self.auth_cookie_cleanup_headers(), ) return self.send_payload( HTTPStatus.OK, response_json(self.session_payload(session)), "application/json; charset=utf-8", send_body=send_body, ) def handle_login(self, payload: dict[str, Any]) -> None: username = str(payload.get("username") or "").strip() password = payload.get("password") # 登录接口要求账号密码都以字符串传入。 if not isinstance(password, str): return self.send_error_payload(HTTPStatus.BAD_REQUEST, "password must be a string", send_body=True) canonical_username = authenticate_user(username, password) if not canonical_username: return self.send_error_payload(HTTPStatus.UNAUTHORIZED, "invalid username or password", send_body=True) # 同一浏览器重复登录时，先清掉旧 token 对应会话。 delete_session(self.current_session_token()) session = create_session( canonical_username, remote_addr=self.client_address[0] if self.client_address else "", user_agent=self.headers.get("User-Agent", ""), ) cookie_header = build_session_cookie( session.token, max_age=max(1, session.expires_at_ts - int(time.time())), ) return self.send_payload( HTTPStatus.OK, response_json(self.session_payload(session)), "application/json; charset=utf-8", send_body=True, extra_headers={"Set-Cookie": cookie_header}, ) def handle_logout(self) -> None: # 退出时带不带当前 Cookie 都按成功处理，方便前端幂等调用。 delete_session(self.current_session_token()) return self.send_payload( HTTPStatus.OK, response_json({"ok": True}), "application/json; charset=utf-8", send_body=True, extra_headers={"Set-Cookie": build_clear_session_cookie()}, ) def session_payload(self, session: Any) -> dict[str, Any]: # 当前只给前端暴露最小必要字段。 return { "authenticated": True, "username": session.username, "expiresAt": session.expires_at, } def require_authentication(self, send_body: bool) -> bool: # 查到有效会话时，当前请求可以继续执行。 if self.current_session() is not None: return True # token 缺失或失效时统一打回 401。 self.send_error_payload( HTTPStatus.UNAUTHORIZED, "authentication required", send_body=send_body, extra_headers=self.auth_cookie_cleanup_headers(), ) return False def auth_cookie_cleanup_headers(self) -> dict[str, str] | None: # 只有请求里真的带了旧 token，才下发清 Cookie 头。 if not self.current_session_token(): return None return {"Set-Cookie": build_clear_session_cookie()} def redirect(self, location: str, *, send_body: bool) -> None: # 页面跳转只依赖 Location，响应体保持空即可。 self.send_payload( HTTPStatus.FOUND, b"", "text/plain; charset=utf-8", send_body=send_body, extra_headers={"Location": location}, ) def handle_nacos_validate(self, payload: dict[str, Any]) -> None: group_name = str(payload.get("group") or "").strip() data_id = str(payload.get("dataId") or "").strip() content = payload.get("content") config_type = str(payload.get("type") or "").strip() or None if not isinstance(content, str): return self.send_error_payload(HTTPStatus.BAD_REQUEST, "content must be a string", send_body=True) try: result = validate_nacos_config_payload( content, group_name=group_name, data_id=data_id, config_type=config_type, ) except Exception as exc: # noqa: BLE001 return self.send_error_payload(HTTPStatus.BAD_GATEWAY, str(exc), send_body=True) return self.send_payload( HTTPStatus.OK, response_json(result), "application/json; charset=utf-8", send_body=True, ) def handle_nacos_save(self, payload: dict[str, Any]) -> None: group_name = str(payload.get("group") or "").strip() data_id = str(payload.get("dataId") or "").strip() content = payload.get("content") config_type = str(payload.get("type") or "").strip() or None # group / dataId 是保存现有配置的唯一键。 if not group_name or not data_id: return self.send_error_payload(HTTPStatus.BAD_REQUEST, "group and dataId are required", send_body=True) if not isinstance(content, str): return self.send_error_payload(HTTPStatus.BAD_REQUEST, "content must be a string", send_body=True) try: result = save_nacos_config_payload(group_name, data_id, content, config_type=config_type) except Exception as exc: # noqa: BLE001 return self.send_error_payload(HTTPStatus.BAD_GATEWAY, str(exc), send_body=True) return self.send_payload( HTTPStatus.OK, response_json(result), "application/json; charset=utf-8", send_body=True, ) def handle_go_validate(self, payload: dict[str, Any]) -> None: content = payload.get("content") if not isinstance(content, str): return self.send_error_payload(HTTPStatus.BAD_REQUEST, "content must be a string", send_body=True) try: result = validate_go_config_payload(content) except Exception as exc: # noqa: BLE001 return self.send_error_payload(HTTPStatus.BAD_GATEWAY, str(exc), send_body=True) return self.send_payload( HTTPStatus.OK, response_json(result), "application/json; charset=utf-8", send_body=True, ) def handle_go_save(self, payload: dict[str, Any]) -> None: content = payload.get("content") if not isinstance(content, str): return self.send_error_payload(HTTPStatus.BAD_REQUEST, "content must be a string", send_body=True) try: result = save_go_config_payload(content) except Exception as exc: # noqa: BLE001 return self.send_error_payload(HTTPStatus.BAD_GATEWAY, str(exc), send_body=True) return self.send_payload( HTTPStatus.OK, response_json(result), "application/json; charset=utf-8", send_body=True, ) def handle_rolling_restart(self, payload: dict[str, Any]) -> None: services = payload.get("services") if not isinstance(services, list): return self.send_error_payload(HTTPStatus.BAD_REQUEST, "services must be a list", send_body=True) try: result = rolling_restart_payload([str(item or "").strip() for item in services]) except Exception as exc: # noqa: BLE001 return self.send_error_payload(HTTPStatus.BAD_GATEWAY, str(exc), send_body=True) return self.send_payload( HTTPStatus.OK, response_json(result), "application/json; charset=utf-8", send_body=True, ) def handle_update_services(self, payload: dict[str, Any]) -> None: services = payload.get("services") if not isinstance(services, list): return self.send_error_payload(HTTPStatus.BAD_REQUEST, "services must be a list", send_body=True) java_git_ref = str(payload.get("javaGitRef") or "").strip() go_git_ref = str(payload.get("goGitRef") or "").strip() image_tag = str(payload.get("imageTag") or "").strip() skip_maven_build = bool(payload.get("skipMavenBuild")) try: result = start_update_job( services, java_git_ref=java_git_ref, go_git_ref=go_git_ref, image_tag=image_tag, skip_maven_build=skip_maven_build, ) except Exception as exc: # noqa: BLE001 return self.send_error_payload(HTTPStatus.BAD_GATEWAY, str(exc), send_body=True) return self.send_payload( HTTPStatus.ACCEPTED, response_json(result), "application/json; charset=utf-8", send_body=True, ) def query_value(self, query: dict[str, list[str]], key: str) -> str: # 统一读取单值 query 参数。 return str((query.get(key) or [""])[0]).strip() def read_json_body(self) -> dict[str, Any]: # Content-Length 缺失时按空请求体处理。 raw_length = self.headers.get("Content-Length", "0").strip() or "0" try: content_length = int(raw_length) except ValueError as exc: raise ValueError("invalid Content-Length") from exc # 空请求体不符合当前 JSON 接口要求。 if content_length <= 0: raise ValueError("request body is required") # 读取原始请求体。 body = self.rfile.read(content_length).decode("utf-8", errors="replace") try: payload = json.loads(body) except json.JSONDecodeError as exc: raise ValueError("request body must be valid json") from exc # 顶层必须是对象。 if not isinstance(payload, dict): raise ValueError("request body must be a json object") return payload def serve_static(self, name: str, content_type: str, send_body: bool) -> None: # 计算静态文件路径。 path = STATIC_DIR / name # 文件不存在时返回 JSON 错误。 if not path.exists(): return self.send_error_payload(HTTPStatus.NOT_FOUND, "static file missing", send_body=send_body) # 文件存在时返回原始二进制内容。 self.send_payload(HTTPStatus.OK, path.read_bytes(), content_type, send_body=send_body) def send_error_payload( self, status: HTTPStatus, message: str, send_body: bool, *, extra_headers: dict[str, str] | None = None, ) -> None: # 错误响应也统一走 JSON。 self.send_payload( status, response_json({"status": int(status), "error": message}), "application/json; charset=utf-8", send_body=send_body, extra_headers=extra_headers, ) def send_payload( self, status: HTTPStatus, body: bytes, content_type: str, send_body: bool, *, extra_headers: dict[str, str] | None = None, ) -> None: # 写状态码。 self.send_response(status) # 写内容类型。 self.send_header("Content-Type", content_type) # 页面和接口都禁用缓存。 self.send_header("Cache-Control", "no-store") # 允许调用方补充下载文件名等额外头。 for key, value in (extra_headers or {}).items(): self.send_header(key, value) # 写入长度，便于 HEAD 请求工作正常。 self.send_header("Content-Length", str(len(body))) # 结束响应头。 self.end_headers() # 需要响应体时再写内容。 if send_body and body: self.wfile.write(body) def main() -> None: # 启动前先把账号库和默认账号准备好。 ensure_auth_store() # 创建线程型 HTTP 服务。 server = ThreadingHTTPServer((APP_BIND, APP_PORT), MonitorHandler) # 打印监听地址便于排查。 print(f"hy-app-monitor listening on http://{APP_BIND}:{APP_PORT}", flush=True) # 进入永久监听。 server.serve_forever()